openvibers/openvibe.ai Commits

Codes C3: a person's own key per provider, and an Anthropic-compatible /v1/messages (#202)

OP OpenVibers committed 14 days ago · 1fa18bd62b · 23 files, +1068 −98

* Codes C3: a person's own key per provider, and an Anthropic-compatible /v1/messages

A person may hold one credential per provider (anthropic, openai, deepseek, minimax, openrouter or a custom https endpoint), each with its own daily budget. PUT/GET/DELETE /api/v1/credentials/:subject/:provider joins the provider-less routes, which keep reading and writing the default one (the row a pre-C3 caller stored).

The key is sealed with AES-256-GCM bound to its owner, subject and provider, so a row moved to another provider does not decrypt. A row written before the provider dimension (v1) is read with the old AAD and rewritten with the new one on first use.

POST /v1/messages speaks Anthropic's Messages API (messages, system, tools, tool_choice, max_tokens, stop_sequences, temperature, stream with Anthropic's own SSE events) behind the same Network-token auth, quotas, metering and audit as /v1/chat/completions. x-openvibe-credential: <provider> uses that person's own key; without it the router picks an upstream that speaks Anthropic's format. Errors are Anthropic's shape.

* Credentials: a per-provider update keeps the default flag

Updating the default credential through its provider route (a budget or
model change) wrote is_default = false, so the provider-less routes and
runs that name no provider lost it. Moving the default and writing the
row now run in one transaction.

Changes

ddiff --git a/.gitleaks.toml b/.gitleaks.toml
iindex 327e92e..9edc1d2 100644
---- a/.gitleaks.toml
++++ b/.gitleaks.toml
@@@ -21,3 +21,15 @@ description = "Made-up provider key in test/provider-responses.test.js (asserts
condition = "AND"
paths = ['''^test/provider-responses\.test\.js$''']
regexes = ['''^sk-resp-secret-0123456789$''']
++
++[[allowlists]]
++description = "Made-up provider keys in test/credentials.test.js (Codes C3: one key per provider, the AAD, the upgrade of an old row and the 'a key is never in a log line' test; placeholders, not secrets)"
++condition = "AND"
++paths = ['''^test/credentials\.test\.js$''']
++regexes = ['''^sk-test-deepseek-0123456789$''', '''^sk-ant-test-0123456789FGHI$''', '''^sk-test-gus-0123456789ABCD$''', '''^sk-test-helen-0123456789JKLM$''', '''^sk-test-ivan-LEAKME-0123456789$''']
++
++[[allowlists]]
++description = "Made-up provider key in test/messages-api.test.js (the person's own Anthropic key on POST /v1/messages; a placeholder, not a secret)"
++condition = "AND"
++paths = ['''^test/messages-api\.test\.js$''']
++regexes = ['''^sk-ant-test-person-0123456789$''']
ddiff --git a/README.md b/README.md
iindex b3f0817..4c78de9 100644
---- a/README.md
++++ b/README.md
@@@ -23,7 +23,8 @@ owns publication truth: Wiki, Blog, News, Live, … decide what to publish.
prompt templates, workflow definitions, runs, the request log (hashes, not prompts), citations, the
cache, quotas and usage, and the audit log, in AI's own PostgreSQL database (`ov_ai`, ADR-035; schema in
[migrations/](migrations/))
--- people's own provider keys (`ai.credential.manage`), encrypted at rest, and per-streamer
++- people's own provider keys (`ai.credential.manage`), encrypted at rest, one per provider (Codes C3: `anthropic`,
++ `openai`, `deepseek`, `minimax`, `openrouter` or a `custom` endpoint), and per-streamer
attribution quotas (`ai.quota.attribution.manage`)
- a person's own computer as their provider (credential provider `node`, contracts 0.220.0): a model on their own
machine (Ollama, LM Studio, llama.cpp through the OpenVibe Node), reached through OpenVibe.Bot with `bot.machine.ai`
@@@ -55,7 +56,7 @@ owns publication truth: Wiki, Blog, News, Live, … decide what to publish.
Implemented here (the service manifest's `capabilities`, audience `openvibe.ai`; routes under
[API](#api)): `ai.run.create`, `ai.run.read`, `ai.workflow.manage`, `ai.provider.manage`,
--`ai.usage.read`, `ai.credential.manage` (`/api/v1/credentials/:subject`),
++`ai.usage.read`, `ai.credential.manage` (`/api/v1/credentials/:subject[/:provider]`),
`ai.quota.attribution.manage` (per-streamer quotas) and `ai.app.run` (developer apps, ADR-014; public,
registered since openvibe-contracts 0.109.0). A service token's `ns`
claim limits which workflow namespaces it may run; an app token is fixed to its project's.
@@@ -202,6 +203,29 @@ coding agents, Codes' harnesses) points its base URL at `https://ai.openvibe.ser
- Production app runs that use the router are limited to free and local capacity (see Developer apps). Today that is
a small local model without tools, so a coding agent's loop needs its project's own provider.
++## Anthropic-compatible endpoint
++
++`POST /v1/messages` and `GET /v1/models` (`server/api/messages.js`), so a client built for Anthropic's Messages API
++(Claude Code, an SDK) points `ANTHROPIC_BASE_URL` at `https://ai.openvibe.services` and works (Codes C3).
++
++- **Tokens:** the same Network tokens as `POST /v1/chat/completions`: an app's (`ai.app.run`; its project pays and owns the
++ run) or a first-party service's (`ai.run.create`).
++- **Request:** Anthropic's Messages API as written: `messages`, `system`, `tools`, `tool_choice`, `stop_sequences`,
++ `temperature`, `top_p`, `top_k`, `max_tokens`, `metadata`, `stream`.
++- **The person's own key:** `x-openvibe-credential: <provider>` uses that person's credential for that provider
++ (`anthropic`, `deepseek`, `minimax`, `openrouter` or a `custom` endpoint) instead of the router. The person is
++ `<provider>:<subject>`, or the token's `on_behalf_of` (an agent acting for them). Without the header the router picks an
++ upstream that speaks Anthropic's format (Anthropic, DeepSeek's and MiniMax's Anthropic-compatible endpoints); an
++ OpenAI-format provider is never converted.
++- **Answer:** a `message` (`id`, `role`, `model`, `content`, `stop_reason`, `usage`), and the run id in
++ `X-OpenVibe-Run-Id`.
++- **Streaming:** `stream: true` answers Anthropic's own server-sent events as they arrive (`message_start`,
++ `content_block_start`, `content_block_delta`, `content_block_stop`, `message_delta`, `message_stop`, `ping`); the run
++ still ends with the whole message.
++- **Each request is a run** of workflow `ai.messages` (step kind `messages`): the run's quota, free allowance, cost,
++ audit and explain apply, and the prompt is not kept. Errors are Anthropic's shape
++ (`{ "type": "error", "error": { "type", "message" } }`).
++
## Developer apps
Roadmap Wave 20, ADR-014. A developer app gets a token from OpenVibe.Network (`POST /oauth/token`,
@@@ -697,7 +721,7 @@ ai.openvibe.network notice from OpenVibe.Sites.
Part of the [OpenVibe network](https://openvibe.network). Built in the open by [OpenVibers](https://github.com/OpenVibers).
<!-- versions:start -->
--- openvibe-contracts: v0.229.0
++- openvibe-contracts: v0.241.0
- openvibe-sdk: v0.67.0
- openvibe-shared: v4.48.0
<!-- versions:end -->
ddiff --git a/STATUS.json b/STATUS.json
iindex 68bc248..74c1144 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -7,7 +7,7 @@
"runtime": "node 22, express, postgresql",
"port": 4700,
"domain": "ai.openvibe.services (the old ai.openvibe.network address was retired on 2026-10-10); the service runs on openvibe-ovh (/opt/openvibe.ai, unit openvibe-ai) on 127.0.0.1:4700 behind nginx; OpenVibe.Live and OpenVibe.Network call it on loopback",
-- "contracts": "openvibe-contracts v0.229.0",
++ "contracts": "openvibe-contracts v0.241.0",
"sdk": "openvibe-sdk v0.67.0",
"shared": "openvibe-shared v4.48.0",
"capabilities": [
ddiff --git a/migrations/0014_credential_provider.sql b/migrations/0014_credential_provider.sql
nnew file mode 100644
iindex 0000000..b5b2c9b
---- /dev/null
++++ b/migrations/0014_credential_provider.sql
@@@ -0,0 +1,19 @@
++-- phase: expand
++-- Codes C3 (contracts 0.241.0 ai.credential-put@2): one credential per provider. A person's key now has a provider
++-- dimension, so the same owner and subject may hold an anthropic, openai, deepseek, minimax, openrouter or custom key
++-- side by side. The row a pre-C3 caller stored stays the default one (the route without a provider reads it), and its
++-- spend keeps the provider_key its requests were already logged under (byo:<owner>:<subject>); a row added per provider
++-- logs under byo:<owner>:<subject>:<provider>. Only an expand step: the old key_enc format (v1, bound to owner and
++-- subject) is read and rewritten as v2 (bound to owner, subject and provider) on first use, in server/credentials.js.
++ALTER TABLE subject_credentials ADD COLUMN IF NOT EXISTS is_default boolean NOT NULL DEFAULT false;
++ALTER TABLE subject_credentials ADD COLUMN IF NOT EXISTS spend_key text COLLATE "C" NOT NULL DEFAULT '';
++UPDATE subject_credentials SET spend_key = 'byo:' || owner || ':' || subject WHERE spend_key = '';
++-- Before this migration there is one row per (owner, subject): it becomes the default.
++UPDATE subject_credentials c SET is_default = true
++ WHERE NOT EXISTS (SELECT 1 FROM subject_credentials d WHERE d.owner = c.owner AND d.subject = c.subject AND d.is_default);
++ALTER TABLE subject_credentials DROP CONSTRAINT IF EXISTS subject_credentials_pkey;
++ALTER TABLE subject_credentials ADD CONSTRAINT subject_credentials_pkey PRIMARY KEY (owner, subject, provider);
++ALTER TABLE subject_credentials DROP CONSTRAINT IF EXISTS subject_credentials_provider_check;
++ALTER TABLE subject_credentials ADD CONSTRAINT subject_credentials_provider_check
++ CHECK (provider IN ('openai', 'anthropic', 'node', 'deepseek', 'minimax', 'openrouter', 'custom'));
++CREATE UNIQUE INDEX IF NOT EXISTS subject_credentials_default_idx ON subject_credentials (owner, subject) WHERE is_default;
ddiff --git a/package-lock.json b/package-lock.json
iindex 16a98cd..ae24c64 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -13,7 +13,7 @@
"dotenv": "^16.4.7",
"express": "^4.21.2",
"iovalkey": "^0.4.0",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.229.0",
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.241.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.67.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.48.0",
"pg": "^8.23.0"
@@@ -1540,9 +1540,9 @@
}
},
"node_modules/openvibe-contracts": {
-- "version": "0.229.0",
-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.229.0",
-- "integrity": "sha512-BYazi/dna8bgdM6+Go8t+D7lTyI9UaWYnJIx0hYEdD0v2ps87iTPONkjnCBadk5RUrBDtvcNoSONS3R+bwFG0w==",
++ "version": "0.241.0",
++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.241.0",
++ "integrity": "sha512-FRTfO31tBk2znR/uvDM1PuLQ1mXkBSuJq/bTbN4wcYCMWXdOrbRprGMFJfuAIIp/7dfSnjxo0A0yiYk7g/9nEQ==",
"license": "AGPL-3.0-only",
"dependencies": {
"ajv": "^8.17.1",
ddiff --git a/package.json b/package.json
iindex 52c5601..0adadf1 100644
---- a/package.json
++++ b/package.json
@@@ -25,7 +25,7 @@
"dotenv": "^16.4.7",
"express": "^4.21.2",
"iovalkey": "^0.4.0",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.229.0",
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.241.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.67.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.48.0",
"pg": "^8.23.0"
ddiff --git a/server/api/messages.js b/server/api/messages.js
nnew file mode 100644
iindex 0000000..32705c6
---- /dev/null
++++ b/server/api/messages.js
@@@ -0,0 +1,215 @@
++'use strict';
++/**
++ * The Anthropic-compatible endpoint: POST /v1/messages on ai.openvibe.services, so a client built for Anthropic's
++ * Messages API (Claude Code, an SDK) points ANTHROPIC_BASE_URL at https://ai.openvibe.services and works (Codes C3).
++ *
++ * Authorization: Bearer <a Network token> an app's token (ai.app.run, its project pays) or a first-party service's
++ * (ai.run.create), exactly as POST /v1/chat/completions takes them
++ * x-openvibe-credential: <provider> the person's own key for that provider instead of the router: anthropic,
++ * deepseek or minimax (their Anthropic-compatible endpoints), or a custom
++ * endpoint. The person is `<provider>:<subject>`, or the token's
++ * on_behalf_of (an agent acting for them)
++ * body Anthropic's Messages API: messages, system, tools, tool_choice,
++ * stop_sequences, temperature, top_p, top_k, max_tokens, metadata, stream
++ *
++ * Each request is a run of workflow ai.messages, so it has the run's quota, free allowance, cost, audit and explain
++ * (the run id rides X-OpenVibe-Run-Id) and the prompt is not kept. Without the header, AI's router picks an upstream that
++ * speaks Anthropic's format (Anthropic's API, or DeepSeek's and MiniMax's Anthropic-compatible endpoints); an
++ * OpenAI-format provider is never converted. stream: true answers Anthropic's own server-sent events as they arrive
++ * (message_start, content_block_start, content_block_delta, content_block_stop, message_delta, message_stop, ping), and
++ * the run still ends with the whole message. Errors are Anthropic's shape ({ type: 'error', error: { type, message } })
++ * with the HTTP status the run's problem had.
++ */
++const express = require('express');
++const { CAPS } = require('../auth');
++const { appAttribution } = require('../apps');
++
++const FIELDS = ['system', 'tools', 'tool_choice', 'stop_sequences', 'temperature', 'top_p', 'top_k', 'metadata'];
++const MAX_TOKENS = 64000;
++
++// Anthropic's error types by HTTP status (docs.anthropic.com/en/api/errors).
++function errorType(status) {
++ if (status === 400 || status === 422) return 'invalid_request_error';
++ if (status === 401) return 'authentication_error';
++ if (status === 403) return 'permission_error';
++ if (status === 404) return 'not_found_error';
++ if (status === 413) return 'request_too_large';
++ if (status === 429) return 'rate_limit_error';
++ if (status === 503 || status === 529) return 'overloaded_error';
++ return 'api_error';
++}
++
++function anthropicError(res, status, message, type) {
++ res.status(status).json({ type: 'error', error: { type: type || errorType(status), message: String(message || 'error').slice(0, 1000) } });
++}
++
++/**
++ * The event stream of one streamed message: each upstream event goes out as `event: <name>` + its JSON as it arrives (or
++ * waits until open(run) has sent the headers). fail() sends Anthropic's error event; done() ends the stream.
++ */
++function createEventStream(res) {
++ let opened = false;
++ let any = false;
++ let stopped = false;
++ const waiting = [];
++ const alive = () => !res.writableEnded && !res.destroyed;
++ const write = (event, data) => { if (alive()) res.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`); };
++ const emit = (event, data) => {
++ if (!data) return;
++ if (event === 'message_stop') stopped = true;
++ write(event, data);
++ };
++ const out = {
++ get opened() { return opened; },
++ event(e) {
++ if (!e || !e.event || !e.data) return;
++ any = true;
++ if (!opened) { waiting.push(e); return; }
++ emit(e.event, e.data);
++ },
++ open(r) {
++ res.status(200);
++ res.setHeader('Content-Type', 'text/event-stream; charset=utf-8');
++ res.setHeader('Cache-Control', 'no-store');
++ res.setHeader('X-Accel-Buffering', 'no');
++ res.setHeader('X-OpenVibe-Run-Id', r.id);
++ if (typeof res.flushHeaders === 'function') res.flushHeaders();
++ opened = true;
++ for (const e of waiting.splice(0)) emit(e.event, e.data);
++ },
++ emit,
++ sentAny: () => any,
++ get stopped() { return stopped; },
++ fail(message, code) {
++ const status = /^(input|validation)\./.test(code || '') ? 400 : /quota|allowance|budget/.test(code || '') ? 429 : 529;
++ emit('error', { type: 'error', error: { type: errorType(status), message: String(message || 'error').slice(0, 1000) } });
++ return out.done();
++ },
++ done() { if (alive()) res.end(); },
++ };
++ return out;
++}
++
++/** The message as a sequence of Messages API events, for an answer that came back in one piece. */
++function replay(res, message, run) {
++ const u = message.usage || {};
++ const usage = { input_tokens: u.input_tokens != null ? u.input_tokens : (run.usage && run.usage.tokens_in) || 0, output_tokens: 0,
++ ...(u.cache_read_input_tokens != null ? { cache_read_input_tokens: u.cache_read_input_tokens } : {}),
++ ...(u.cache_creation_input_tokens != null ? { cache_creation_input_tokens: u.cache_creation_input_tokens } : {}) };
++ res.emit('message_start', { type: 'message_start', message: { ...message, content: [], stop_reason: null, stop_sequence: null, usage } });
++ (message.content || []).forEach((block, index) => {
++ res.emit('content_block_start', { type: 'content_block_start', index, content_block: block.type === 'tool_use' ? { ...block, input: {} } : { ...block, text: '' } });
++ if (block.type === 'text' && block.text) res.emit('content_block_delta', { type: 'content_block_delta', index, delta: { type: 'text_delta', text: block.text } });
++ else if (block.type === 'tool_use') res.emit('content_block_delta', { type: 'content_block_delta', index, delta: { type: 'input_json_delta', partial_json: JSON.stringify(block.input || {}) } });
++ res.emit('content_block_stop', { type: 'content_block_stop', index });
++ });
++ res.emit('message_delta', { type: 'message_delta', delta: { stop_reason: message.stop_reason || 'end_turn', stop_sequence: message.stop_sequence || null }, usage: { output_tokens: u.output_tokens != null ? u.output_tokens : (run.usage && run.usage.tokens_out) || 0 } });
++ res.emit('message_stop', { type: 'message_stop' });
++}
++
++/** x-openvibe-credential: <provider> or <provider>:<subject>, or nothing (the router picks). */
++function credentialFor(req) {
++ const raw = String(req.get('x-openvibe-credential') || '').trim();
++ if (!raw) return {};
++ const m = /^([a-z]+)(?::(.+))?$/.exec(raw);
++ if (!m) return { status: 400, error: 'x-openvibe-credential must be <provider> or <provider>:<subject>' };
++ const provider = m[1];
++ const subject = m[2] || (req.principal && req.principal.claims && req.principal.claims.on_behalf_of) || null;
++ if (!subject) return { status: 400, error: `x-openvibe-credential ${provider} needs the person whose key it is: send <provider>:<subject>, or use a token that acts for them (on_behalf_of)` };
++ return { credential: { subject, provider } };
++}
++
++function messagesRouter({ runs, auth, config, log = console }) {
++ const r = express.Router();
++ const allow = auth.appOrCap(CAPS.appRun, CAPS.runCreate);
++
++ // The auth layer writes problem+json; a client built for Anthropic reads { type: 'error', error: { … } }. Convert it
++ // as it is written, whichever way the layer writes it.
++ const asAnthropicErrors = (req, res, next) => {
++ const end = res.end.bind(res);
++ res.end = (chunk, ...rest) => {
++ const ct = String(res.getHeader('Content-Type') || '');
++ if (res.statusCode >= 400 && /problem\+json/.test(ct) && chunk) {
++ let p = null;
++ try { p = JSON.parse(Buffer.isBuffer(chunk) ? chunk.toString('utf8') : String(chunk)); } catch { p = null; }
++ if (p && (p.type !== 'error')) {
++ const body = JSON.stringify({ type: 'error', error: { type: errorType(res.statusCode), message: String(p.detail || p.title || p.code || 'error') } });
++ res.setHeader('Content-Type', 'application/json; charset=utf-8');
++ res.setHeader('Content-Length', Buffer.byteLength(body));
++ return end(body);
++ }
++ }
++ return end(chunk, ...rest);
++ };
++ next();
++ };
++
++ r.post('/v1/messages', asAnthropicErrors, allow, async (req, res) => {
++ const b = req.body && typeof req.body === 'object' ? req.body : {};
++ if (!Array.isArray(b.messages) || !b.messages.length) return anthropicError(res, 400, 'messages: must be a non-empty array');
++ if (!Number.isInteger(b.max_tokens) || b.max_tokens < 1) return anthropicError(res, 400, 'max_tokens: must be a positive integer');
++ const input = { messages: b.messages, max_tokens: Math.min(b.max_tokens, MAX_TOKENS) };
++ for (const k of FIELDS) if (b[k] !== undefined && b[k] !== null) input[k] = b[k];
++ let body = { workflow: 'ai.messages', input };
++ if (req.principal.kind === 'app') body = { ...body, attribution: appAttribution(req.principal) };
++ const own = credentialFor(req);
++ if (own.error) return anthropicError(res, own.status, own.error);
++ if (own.credential) body = { ...body, credential: own.credential };
++ // stream: true. The run is made first (a refused one keeps its status code); then the event stream opens and every
++ // event the upstream sends goes out as it arrives. One that came before the stream opened waits for it.
++ const live = b.stream === true ? createEventStream(res) : null;
++ let run;
++ try {
++ const created = await runs.create(body, req.principal, { trace: req.ov && req.ov.traceId, requestId: req.ov && req.ov.requestId, ...(live ? { onDelta: live.event } : {}) });
++ if (live) live.open(created.run);
++ run = await runs.explained(await runs.wait(created, Math.max(config.runs.maxWaitMs, 120000)));
++ } catch (err) {
++ if (live && live.opened) return live.fail(err && (err.detail || err.message), err && err.code);
++ if ((err && err.status) >= 500) log.warn(`[ai] /v1/messages: ${err && err.message}`);
++ return anthropicError(res, (err && err.status) || 500, err && (err.detail || err.message), err && err.code === 'credential.unsupported' ? 'invalid_request_error' : null);
++ }
++ if (live && (run.status === 'failed' || run.status === 'cancelled' || !run.output || !Array.isArray(run.output.content))) {
++ const code = (run.error && run.error.code) || (['failed', 'cancelled'].includes(run.status) ? 'run.failed' : 'run.timeout');
++ return live.fail((run.error && run.error.detail) || (code === 'run.timeout' ? `the message is still running; GET /api/v1/runs/${run.id}` : code), code);
++ }
++ if (run.status === 'failed' || run.status === 'cancelled') {
++ const code = (run.error && run.error.code) || 'run.failed';
++ const status = /^(input|validation)\./.test(code) ? 400 : /quota|allowance|budget/.test(code) ? 429
++ : code === 'credential.unsupported' ? 422 : code === 'credential.not_found' ? 404 : 529;
++ return anthropicError(res, status, (run.error && run.error.detail) || code);
++ }
++ if (!['succeeded', 'cached'].includes(run.status) || !run.output || !Array.isArray(run.output.content)) {
++ return anthropicError(res, 529, `the message is still running; GET /api/v1/runs/${run.id}`, 'overloaded_error');
++ }
++ const out = run.output;
++ const message = {
++ id: out.id || `msg_${run.id}`, type: 'message', role: 'assistant',
++ model: out.model || (run.provenance && run.provenance.model) || 'openvibe',
++ content: out.content, stop_reason: out.stop_reason || 'end_turn', stop_sequence: out.stop_sequence || null,
++ usage: {
++ input_tokens: out.usage && out.usage.input_tokens != null ? out.usage.input_tokens : (run.usage.tokens_in || 0),
++ output_tokens: out.usage && out.usage.output_tokens != null ? out.usage.output_tokens : (run.usage.tokens_out || 0),
++ ...(out.usage && out.usage.cache_read_input_tokens != null ? { cache_read_input_tokens: out.usage.cache_read_input_tokens } : {}),
++ ...(out.usage && out.usage.cache_creation_input_tokens != null ? { cache_creation_input_tokens: out.usage.cache_creation_input_tokens } : {}),
++ },
++ };
++ if (live) {
++ // The events already went out as they arrived; an answer that came in one piece (an upstream that cannot
++ // stream, a replayed run) is sent as the events it is, and a stream that never closed is closed.
++ if (!live.sentAny()) replay(live, message, run);
++ else if (!live.stopped) live.emit('message_stop', { type: 'message_stop' });
++ return live.done();
++ }
++ res.setHeader('X-OpenVibe-Run-Id', run.id);
++ return res.status(200).json(message);
++ });
++
++ // The model list a client asks for: one router model, the upstreams are AI's choice.
++ r.get('/v1/models', asAnthropicErrors, allow, (req, res) => {
++ res.json({ data: [{ type: 'model', id: 'openvibe', display_name: 'OpenVibe AI', created_at: '2026-10-30T00:00:00Z' }], has_more: false, first_id: 'openvibe', last_id: 'openvibe' });
++ });
++
++ return r;
++}
++
++module.exports = { messagesRouter, errorType };
ddiff --git a/server/app.js b/server/app.js
iindex 974b7f9..4af406a 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -106,6 +106,8 @@ function createApp({ config, db, registry, pool, quotas, cache, runs, auth, keys
app.use(actionsRouter({ runs, registry, auth, log }));
// The OpenAI-compatible endpoint (POST /v1/chat/completions, GET /v1/models): runs of ai.complete.
app.use(require('./api/openai').openaiRouter({ runs, auth, config, log }));
++ // The Anthropic-compatible endpoint (POST /v1/messages, GET /v1/models): runs of ai.messages (Codes C3).
++ app.use(require('./api/messages').messagesRouter({ runs, auth, config, log }));
app.use(require('./api/attribution-quotas').attributionQuotasRouter({ db, quotas, auth, clock, sendError: (res, err, ctx) => sendError(res, err, ctx, log) }));
if (credentials) app.use(require('./credentials').credentialsRouter({ credentials, auth, registry, sendError: (res, err, ctx) => sendError(res, err, ctx, log) }));
app.use(adminRouter({ db, registry, pool, quotas, cache, runs, auth, ops, log }));
ddiff --git a/server/console/presets.js b/server/console/presets.js
iindex 01f5f36..c37d2a4 100644
---- a/server/console/presets.js
++++ b/server/console/presets.js
@@@ -16,6 +16,10 @@ const PRESETS = [
{ id: 'xai', name: 'xAI', kind: 'openai', base_url: 'https://api.x.ai/v1', model: '', keys: 'https://console.x.ai/' },
{ id: 'openai', name: 'OpenAI', kind: 'openai', base_url: 'https://api.openai.com/v1', model: 'gpt-5-mini', keys: 'https://platform.openai.com/api-keys' },
{ id: 'anthropic', name: 'Anthropic', kind: 'anthropic', base_url: 'https://api.anthropic.com/v1', model: '', keys: 'https://console.anthropic.com/settings/keys' },
++ // Anthropic-format endpoints of providers that also speak the OpenAI format (Codes C3): POST /v1/messages reaches
++ // these as they are, an OpenAI-shaped record never is converted.
++ { id: 'deepseek-anthropic', name: 'DeepSeek (Anthropic format)', kind: 'anthropic', base_url: 'https://api.deepseek.com/anthropic', model: 'deepseek-chat', keys: 'https://platform.deepseek.com/api_keys' },
++ { id: 'minimax-anthropic', name: 'MiniMax (Anthropic format)', kind: 'anthropic', base_url: 'https://api.minimax.io/anthropic', model: 'MiniMax-M2', keys: 'https://platform.minimax.io/user-center/basic-information/interface-key' },
{ id: 'custom', name: 'Another OpenAI-compatible API', kind: 'openai', base_url: '', model: '', keys: null },
];
ddiff --git a/server/credentials.js b/server/credentials.js
iindex 4c049bb..8644a53 100644
---- a/server/credentials.js
++++ b/server/credentials.js
@@@ -1,18 +1,27 @@
'use strict';
/**
-- * A person's own provider key (roadmap WS-O task 2; Contracts 0.73.0 ai.credential.manage, ai.credential-put@1,
-- * ai.credential@1). The service that holds the person's consent (Live, for a streamer's AI viewers) stores the
-- * key here; it never leaves OpenVibe.AI again. Runs that name the credential (ai.run-request credential { subject })
-- * call that provider with that key only (server/providers/index.js executeWithCredential), never a shared key.
++ * A person's own provider key (roadmap WS-O task 2; Contracts ai.credential.manage, ai.credential-put@2 and
++ * ai.credential@2 since Codes C3). The service that holds the person's consent (Live, for a streamer's AI viewers;
++ * Codes, for the people it builds for) stores the key here; it never leaves OpenVibe.AI again. Runs that name the
++ * credential (ai.run-request credential { subject, provider? }) call that provider with that key only
++ * (server/providers/index.js executeWithCredential), never a shared key.
*
-- * PUT /api/v1/credentials/:subject store or replace (the storing service is the owner)
-- * GET /api/v1/credentials/:subject read it back without the key: a four-character hint, today's spend
-- * DELETE /api/v1/credentials/:subject
++ * PUT /api/v1/credentials/:subject store or replace the DEFAULT provider's credential (the v1 route)
++ * PUT /api/v1/credentials/:subject/:provider store or replace that provider's credential
++ * GET /api/v1/credentials/:subject[/:provider] read it back without the key: a four-character hint, today's spend
++ * DELETE /api/v1/credentials/:subject[/:provider]
*
-- * At rest the key is AES-256-GCM encrypted with AI_CREDENTIALS_KEY (64 hex characters), bound to its owner and
-- * subject (the additional data), so a row copied to another owner or subject does not decrypt. Without the key
-- * configured the routes answer 503 and credentialed runs are refused. Spend is what the credential's requests
-- * cost today (provider_key byo:<owner>:<subject>), checked against the optional budget_usd_per_day.
++ * A person may keep one credential per provider (anthropic, openai, deepseek, minimax, openrouter, custom, node), each
++ * with its own daily budget; the provider-less routes read and write the default one, which is the row a pre-C3 caller
++ * stored (migrations/0014).
++ *
++ * At rest the key is AES-256-GCM encrypted with AI_CREDENTIALS_KEY (64 hex characters) and bound to its owner, subject
++ * AND provider (the additional data), so a row copied to another owner, subject or provider does not decrypt. Rows
++ * written before the provider dimension were sealed with owner|subject only (`v1:`); they are read with that AAD and
++ * rewritten as `v2:` (owner|subject|provider) on first use. Without the key configured the routes answer 503 and
++ * credentialed runs are refused. Spend is what the credential's requests cost today (provider_key byo:<owner>:<subject>
++ * for a row stored before C3 or through the provider-less route, byo:<owner>:<subject>:<provider> for the rest),
++ * checked against the optional budget_usd_per_day.
*
* Provider node (contracts 0.220.0) is the person's own machine instead: `machine` names it, there is no key (nothing is
* sealed, no hint), and a run reaches it only through OpenVibe.Bot (server/machines.js), for that person only.
@@@ -23,11 +32,22 @@ const { validate } = require('openvibe-contracts');
const { AiError } = require('./util');
const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
++const PROVIDERS = ['openai', 'anthropic', 'node', 'deepseek', 'minimax', 'openrouter', 'custom'];
++/** Where each provider answers when the person stored no base_url. DeepSeek and MiniMax speak the Anthropic format at
++ * their own /anthropic path; a custom provider always names its endpoint. */
++const PROVIDER_BASE_URL = {
++ anthropic: 'https://api.anthropic.com/v1', openai: 'https://api.openai.com/v1', deepseek: 'https://api.deepseek.com/anthropic',
++ minimax: 'https://api.minimax.io/anthropic', openrouter: 'https://openrouter.ai/api/v1',
++};
--function providerKeyFor(owner, subject) { return `byo:${owner}:${subject}`; }
++function providerKeyFor(owner, subject, provider = null) { return provider ? `byo:${owner}:${subject}:${provider}` : `byo:${owner}:${subject}`; }
++/** The AAD a key is sealed with: moving the row to another owner, subject or provider must not decrypt it. */
++function aadFor(owner, subject, provider) { return `${owner}|${subject}|${provider}`; }
++function legacyAadFor(owner, subject) { return `${owner}|${subject}`; }
function createCredentials({ db, config, clock = { now: () => Date.now() } }) {
-- // subject_credentials is in migrations/0001_initial.sql (ADR-035).
++ // subject_credentials is in migrations/0001_initial.sql (ADR-035); 0011 added the node machine, 0014 the provider
++ // dimension (is_default, spend_key, one row per provider).
const iso = (ms) => new Date(ms).toISOString();
function secretKey() {
@@@ -35,25 +55,39 @@ function createCredentials({ db, config, clock = { now: () => Date.now() } }) {
if (!/^[0-9a-f]{64}$/i.test(hex)) throw new AiError(503, 'credentials.unavailable', 'person credentials are not configured on this service (AI_CREDENTIALS_KEY)');
return Buffer.from(hex, 'hex');
}
++ /** v2: bound to owner, subject and provider (Codes C3). */
function seal(plain, aad) {
const iv = crypto.randomBytes(12);
const c = crypto.createCipheriv('aes-256-gcm', secretKey(), iv);
c.setAAD(Buffer.from(aad));
const body = Buffer.concat([c.update(String(plain), 'utf8'), c.final()]);
-- return `v1:${iv.toString('base64')}:${c.getAuthTag().toString('base64')}:${body.toString('base64')}`;
++ return `v2:${iv.toString('base64')}:${c.getAuthTag().toString('base64')}:${body.toString('base64')}`;
}
-- function open(sealed, aad) {
++ /**
++ * A sealed key. `v2:` opens with the provider-bound AAD; `v1:` is a row written before the provider dimension
++ * and opens with owner|subject only (the caller then rewrites it). Any other prefix is refused.
++ */
++ function open(sealed, aad, legacyAad) {
const [v, iv, tag, body] = String(sealed).split(':');
-- if (v !== 'v1') throw new Error('unknown credential format');
++ if (v !== 'v1' && v !== 'v2') throw new Error('unknown credential format');
const d = crypto.createDecipheriv('aes-256-gcm', secretKey(), Buffer.from(iv, 'base64'));
-- d.setAAD(Buffer.from(aad));
++ d.setAAD(Buffer.from(v === 'v1' ? legacyAad : aad));
d.setAuthTag(Buffer.from(tag, 'base64'));
return Buffer.concat([d.update(Buffer.from(body, 'base64')), d.final()]).toString('utf8');
}
++ /** The row's key; a pre-C3 row is upgraded to the provider-bound AAD as it is read. */
++ async function openRow(r) {
++ const key = open(r.key_enc, aadFor(r.owner, r.subject, r.provider), legacyAadFor(r.owner, r.subject));
++ if (String(r.key_enc).startsWith('v1:')) {
++ await db.prepare("UPDATE subject_credentials SET key_enc = ? WHERE owner = ? AND subject = ? AND provider = ? AND key_enc LIKE 'v1:%'")
++ .run(seal(key, aadFor(r.owner, r.subject, r.provider)), r.owner, r.subject, r.provider);
++ }
++ return key;
++ }
-- async function spentToday(owner, subject) {
++ async function spentToday(row) {
const day = iso(clock.now()).slice(0, 10);
-- const r = await db.prepare("SELECT COALESCE(SUM(cost_usd), 0) AS c FROM requests WHERE provider_key = ? AND status = 'ok' AND created_at >= ?").get(providerKeyFor(owner, subject), `${day}T00:00:00.000Z`);
++ const r = await db.prepare("SELECT COALESCE(SUM(cost_usd), 0) AS c FROM requests WHERE provider_key = ? AND status = 'ok' AND created_at >= ?").get(row.spend_key || providerKeyFor(row.owner, row.subject), `${day}T00:00:00.000Z`);
return r ? Number(r.c) || 0 : 0;
}
@@@ -62,77 +96,119 @@ function createCredentials({ db, config, clock = { now: () => Date.now() } }) {
subject: row.subject, owner: row.owner, provider: row.provider, base_url: row.base_url || null,
...(row.provider === 'node' ? { machine: row.machine } : { key_hint: row.key_hint }),
models: JSON.parse(row.models || '{}'), budget_usd_per_day: row.budget_usd_day == null ? null : row.budget_usd_day,
-- spent_usd_today: Math.round(await spentToday(row.owner, row.subject) * 1e6) / 1e6,
++ spent_usd_today: Math.round(await spentToday(row) * 1e6) / 1e6,
created_at: row.created_at, updated_at: row.updated_at, last_used_at: row.last_used_at || null,
};
}
-- const row = async (owner, subject) => await db.prepare('SELECT * FROM subject_credentials WHERE owner = ? AND subject = ?').get(owner, subject);
++ const getRow = async (owner, subject, provider) => provider
++ ? await db.prepare('SELECT * FROM subject_credentials WHERE owner = ? AND subject = ? AND provider = ?').get(owner, subject, provider)
++ : await db.prepare('SELECT * FROM subject_credentials WHERE owner = ? AND subject = ? AND is_default').get(owner, subject);
-- async function put(owner, subject, body) {
++ /**
++ * put(owner, subject, body, provider?) — the provider is the route's when it names one (the body may then leave it
++ * out). The provider-less route stores the DEFAULT credential.
++ */
++ async function put(owner, subject, body, provider = null) {
if (!SUBJECT_RE.test(String(subject))) throw new AiError(400, 'input.invalid', 'subject must be a usr_ subject');
-- const v = validate('ai.credential-put@1', body);
-- if (!v.valid) throw new AiError(422, 'input.invalid', 'body does not match ai.credential-put@1', { errors: v.errors });
++ const v = validate('ai.credential-put@2', body);
++ if (!v.valid) throw new AiError(422, 'input.invalid', 'body does not match ai.credential-put@2', { errors: v.errors });
++ if (provider !== null && body.provider !== provider) throw new AiError(422, 'input.invalid', `body.provider is ${body.provider}, the route names ${provider}`);
const at = iso(clock.now());
-- const prev = await row(owner, subject);
++ const makeDefault = provider === null;
++ const prev = await getRow(owner, subject, body.provider);
if (body.provider === 'node') {
// The person's own machine: nothing to seal. Bot checks it is theirs and that they turned it on, at each run.
-- await db.prepare(`INSERT INTO subject_credentials (owner, subject, provider, base_url, machine, key_enc, key_hint, models, budget_usd_day, created_at, updated_at)
-- VALUES (?, ?, 'node', NULL, ?, '', '', ?, NULL, ?, ?)
-- ON CONFLICT(owner, subject) DO UPDATE SET provider = excluded.provider, base_url = NULL, machine = excluded.machine, key_enc = '',
-- key_hint = '', models = excluded.models, budget_usd_day = NULL, updated_at = excluded.updated_at`)
-- .run(owner, subject, body.machine, JSON.stringify(body.models || {}), prev ? prev.created_at : at, at);
-- return await view(await row(owner, subject));
++ await db.tx(async () => {
++ if (makeDefault) await db.prepare('UPDATE subject_credentials SET is_default = false WHERE owner = ? AND subject = ? AND provider <> ?').run(owner, subject, 'node');
++ await db.prepare(`INSERT INTO subject_credentials (owner, subject, provider, base_url, machine, key_enc, key_hint, models, budget_usd_day, is_default, spend_key, created_at, updated_at)
++ VALUES (?, ?, 'node', NULL, ?, '', '', ?, NULL, ?, ?, ?, ?)
++ ON CONFLICT(owner, subject, provider) DO UPDATE SET base_url = NULL, machine = excluded.machine, key_enc = '',
++ key_hint = '', models = excluded.models, budget_usd_day = NULL, is_default = subject_credentials.is_default OR excluded.is_default, updated_at = excluded.updated_at`)
++ .run(owner, subject, body.machine, JSON.stringify(body.models || {}), makeDefault,
++ (prev && prev.spend_key) || (makeDefault ? providerKeyFor(owner, subject) : providerKeyFor(owner, subject, 'node')), prev ? prev.created_at : at, at);
++ });
++ return await view(await getRow(owner, subject, 'node'));
}
// The stored key stays for a change of models or budget; a new provider or endpoint needs the key again.
-- const sameEndpoint = prev && prev.provider !== 'node' && prev.provider === body.provider && (prev.base_url || null) === (body.base_url || null);
-- if (body.api_key == null && !sameEndpoint) {
-- throw new AiError(400, 'credential.key_required', prev ? 'enter the key again to change the provider or its address' : 'api_key is required the first time');
++ const newKey = body.key != null ? String(body.key) : (body.api_key != null ? String(body.api_key) : null);
++ const sameEndpoint = prev && prev.provider !== 'node' && (prev.base_url || null) === (body.base_url || null);
++ if (newKey == null && !sameEndpoint) {
++ throw new AiError(400, 'credential.key_required', prev ? 'enter the key again to change the provider or its address' : 'key is required the first time');
}
-- const key = body.api_key != null ? String(body.api_key) : open(prev.key_enc, `${owner}|${subject}`);
-- await db.prepare(`INSERT INTO subject_credentials (owner, subject, provider, base_url, key_enc, key_hint, models, budget_usd_day, created_at, updated_at)
-- VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
-- ON CONFLICT(owner, subject) DO UPDATE SET provider = excluded.provider, base_url = excluded.base_url, key_enc = excluded.key_enc,
-- key_hint = excluded.key_hint, models = excluded.models, budget_usd_day = excluded.budget_usd_day, updated_at = excluded.updated_at`)
-- .run(owner, subject, body.provider, body.base_url || null, seal(key, `${owner}|${subject}`), `…${key.slice(-4)}`, JSON.stringify(body.models || {}),
-- body.budget_usd_per_day ? Number(body.budget_usd_per_day) : null, prev ? prev.created_at : at, at);
-- return await view(await row(owner, subject));
++ const key = newKey != null ? newKey : await openRow(prev);
++ await db.tx(async () => {
++ if (makeDefault) await db.prepare('UPDATE subject_credentials SET is_default = false WHERE owner = ? AND subject = ? AND provider <> ?').run(owner, subject, body.provider);
++ await db.prepare(`INSERT INTO subject_credentials (owner, subject, provider, base_url, key_enc, key_hint, models, budget_usd_day, is_default, spend_key, created_at, updated_at)
++ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
++ ON CONFLICT(owner, subject, provider) DO UPDATE SET base_url = excluded.base_url, key_enc = excluded.key_enc,
++ key_hint = excluded.key_hint, models = excluded.models, budget_usd_day = excluded.budget_usd_day, is_default = subject_credentials.is_default OR excluded.is_default, updated_at = excluded.updated_at`)
++ .run(owner, subject, body.provider, body.base_url || null, seal(key, aadFor(owner, subject, body.provider)), `…${key.slice(-4)}`, JSON.stringify(body.models || {}),
++ body.budget_usd_per_day ? Number(body.budget_usd_per_day) : null, makeDefault,
++ (prev && prev.spend_key) || (makeDefault ? providerKeyFor(owner, subject) : providerKeyFor(owner, subject, body.provider)), prev ? prev.created_at : at, at);
++ });
++ return await view(await getRow(owner, subject, body.provider));
}
-- async function get(owner, subject) {
-- const r = await row(owner, subject);
-- if (!r) throw new AiError(404, 'credential.not_found', 'no credential stored for that subject');
++ async function get(owner, subject, provider = null) {
++ if (provider !== null && !PROVIDERS.includes(provider)) throw new AiError(404, 'credential.not_found', 'no credential stored for that provider');
++ const r = await getRow(owner, subject, provider);
++ if (!r) throw new AiError(404, 'credential.not_found', provider ? `no ${provider} credential stored for that subject` : 'no credential stored for that subject');
return await view(r);
}
-- async function remove(owner, subject) {
-- return (await db.prepare('DELETE FROM subject_credentials WHERE owner = ? AND subject = ?').run(owner, subject)).changes > 0;
++ /** The newest of what is left becomes the default, so the provider-less routes keep answering. */
++ async function ensureDefault(owner, subject) {
++ await db.prepare(`UPDATE subject_credentials SET is_default = true WHERE owner = ? AND subject = ?
++ AND NOT EXISTS (SELECT 1 FROM subject_credentials d WHERE d.owner = ? AND d.subject = ? AND d.is_default)
++ AND provider = (SELECT provider FROM subject_credentials WHERE owner = ? AND subject = ? ORDER BY updated_at DESC, provider LIMIT 1)`)
++ .run(owner, subject, owner, subject, owner, subject);
++ }
++
++ async function remove(owner, subject, provider = null) {
++ if (provider !== null) {
++ let gone = false;
++ await db.tx(async () => {
++ gone = (await db.prepare('DELETE FROM subject_credentials WHERE owner = ? AND subject = ? AND provider = ?').run(owner, subject, provider)).changes > 0;
++ if (gone) await ensureDefault(owner, subject);
++ });
++ return gone;
++ }
++ let gone = false;
++ await db.tx(async () => {
++ gone = (await db.prepare('DELETE FROM subject_credentials WHERE owner = ? AND subject = ? AND is_default').run(owner, subject)).changes > 0;
++ if (gone) await ensureDefault(owner, subject);
++ });
++ return gone;
}
/** Checks a run may use the credential (exists, within budget); throws AiError otherwise. */
-- async function admit(owner, subject) {
-- const r = await row(owner, subject);
-- if (!r) throw new AiError(404, 'credential.not_found', 'no credential stored for that subject');
++ async function admit(owner, subject, provider = null) {
++ if (provider !== null && !PROVIDERS.includes(provider)) throw new AiError(404, 'credential.not_found', 'no credential stored for that provider');
++ const r = await getRow(owner, subject, provider);
++ if (!r) throw new AiError(404, 'credential.not_found', provider ? `no ${provider} credential stored for that subject` : 'no credential stored for that subject');
if (r.provider === 'node') return; // the person's own machine: no key, no spend
secretKey();
-- if (r.budget_usd_day != null && r.budget_usd_day > 0 && await spentToday(owner, subject) >= r.budget_usd_day) {
++ if (r.budget_usd_day != null && r.budget_usd_day > 0 && await spentToday(r) >= r.budget_usd_day) {
const tomorrow = new Date(iso(clock.now()).slice(0, 10) + 'T00:00:00.000Z').getTime() + 86400000;
throw new AiError(429, 'quota.exceeded', "this person's own daily budget for their key is spent", { retry_after_seconds: Math.max(60, Math.ceil((tomorrow - clock.now()) / 1000)), quota: { scope: 'credential' } });
}
}
-- /** The decrypted credential for one execution: { providerKey, provider, base_url, apiKey, models }. */
-- async function forRun(owner, subject) {
-- const r = await row(owner, subject);
-- if (!r) throw new AiError(404, 'credential.not_found', 'the credential was deleted');
-- await db.prepare('UPDATE subject_credentials SET last_used_at = ? WHERE owner = ? AND subject = ?').run(iso(clock.now()), owner, subject);
-- if (r.provider === 'node') return { providerKey: providerKeyFor(owner, subject), provider: 'node', subject, machine: r.machine, models: JSON.parse(r.models || '{}') };
-- return { providerKey: providerKeyFor(owner, subject), provider: r.provider, base_url: r.base_url || null, apiKey: open(r.key_enc, `${owner}|${subject}`), models: JSON.parse(r.models || '{}') };
++ /** The decrypted credential for one execution: { providerKey, provider, base_url, apiKey|machine, models }. */
++ async function forRun(owner, subject, provider = null) {
++ if (provider !== null && !PROVIDERS.includes(provider)) throw new AiError(404, 'credential.not_found', 'no credential stored for that provider');
++ const r = await getRow(owner, subject, provider);
++ if (!r) throw new AiError(404, 'credential.not_found', provider ? `the ${provider} credential was deleted` : 'the credential was deleted');
++ await db.prepare('UPDATE subject_credentials SET last_used_at = ? WHERE owner = ? AND subject = ? AND provider = ?').run(iso(clock.now()), owner, subject, r.provider);
++ const providerKey = r.spend_key || providerKeyFor(owner, subject);
++ if (r.provider === 'node') return { providerKey, provider: 'node', subject, machine: r.machine, models: JSON.parse(r.models || '{}') };
++ return { providerKey, provider: r.provider, subject, base_url: r.base_url || PROVIDER_BASE_URL[r.provider] || null, apiKey: await openRow(r), models: JSON.parse(r.models || '{}') };
}
return { put, get, remove, admit, forRun, spentToday, providerKeyFor };
}
--/** PUT/GET/DELETE /api/v1/credentials/:subject (ai.credential.manage; a service token speaks for itself). */
++/** PUT/GET/DELETE /api/v1/credentials/:subject[/:provider] (ai.credential.manage; a service token speaks for itself). */
function credentialsRouter({ credentials, auth, registry, sendError }) {
const r = express.Router();
const guard = auth.requireCap('ai.credential.manage');
@@@ -141,16 +217,28 @@ function credentialsRouter({ credentials, auth, registry, sendError }) {
if (!s || s.type !== 'service') throw new AiError(403, 'capability.denied', 'only a service stores credentials for the people it serves');
return s.id;
};
++ const audit = async (req, action, out) => registry.audit(req.principal.sub, action, 'credential', `${out.owner}:${out.subject}:${out.provider}`, { trace: req.ov && req.ov.traceId, metadata: { provider: out.provider, base_url: out.base_url } });
r.put('/api/v1/credentials/:subject', guard, async (req, res) => {
try {
const out = await credentials.put(owner(req), req.params.subject, req.body || {});
-- await registry.audit(req.principal.sub, 'credential.put', 'credential', `${out.owner}:${out.subject}`, { trace: req.ov && req.ov.traceId, metadata: { provider: out.provider, base_url: out.base_url } });
++ await audit(req, 'credential.put', out);
++ res.set('Cache-Control', 'no-store').json(out);
++ } catch (err) { sendError(res, err, req.ov); }
++ });
++ r.put('/api/v1/credentials/:subject/:provider', guard, async (req, res) => {
++ try {
++ const body = { ...(req.body || {}), ...(req.body && req.body.provider == null ? { provider: req.params.provider } : {}) };
++ const out = await credentials.put(owner(req), req.params.subject, body, req.params.provider);
++ await audit(req, 'credential.put', out);
res.set('Cache-Control', 'no-store').json(out);
} catch (err) { sendError(res, err, req.ov); }
});
r.get('/api/v1/credentials/:subject', guard, async (req, res) => {
try { res.set('Cache-Control', 'no-store').json(await credentials.get(owner(req), req.params.subject)); } catch (err) { sendError(res, err, req.ov); }
});
++ r.get('/api/v1/credentials/:subject/:provider', guard, async (req, res) => {
++ try { res.set('Cache-Control', 'no-store').json(await credentials.get(owner(req), req.params.subject, req.params.provider)); } catch (err) { sendError(res, err, req.ov); }
++ });
r.delete('/api/v1/credentials/:subject', guard, async (req, res) => {
try {
const o = owner(req);
@@@ -159,7 +247,15 @@ function credentialsRouter({ credentials, auth, registry, sendError }) {
res.status(gone ? 204 : 404).end();
} catch (err) { sendError(res, err, req.ov); }
});
++ r.delete('/api/v1/credentials/:subject/:provider', guard, async (req, res) => {
++ try {
++ const o = owner(req);
++ const gone = await credentials.remove(o, req.params.subject, req.params.provider);
++ if (gone) await registry.audit(req.principal.sub, 'credential.delete', 'credential', `${o}:${req.params.subject}:${req.params.provider}`, { trace: req.ov && req.ov.traceId });
++ res.status(gone ? 204 : 404).end();
++ } catch (err) { sendError(res, err, req.ov); }
++ });
return r;
}
--module.exports = { createCredentials, credentialsRouter, providerKeyFor };
++module.exports = { createCredentials, credentialsRouter, providerKeyFor, aadFor, PROVIDERS, PROVIDER_BASE_URL };
ddiff --git a/server/providers/anthropic.js b/server/providers/anthropic.js
iindex d0df553..ab81634 100644
---- a/server/providers/anthropic.js
++++ b/server/providers/anthropic.js
@@@ -4,11 +4,19 @@
* (_anthropicBody/_callAnthropic): real system blocks with cache_control on the stable ones
* (prompt caching), images as base64 blocks, structured output through a forced tool.
* No embeddings or transcription on this API — supports() says so and the router skips it.
++ *
++ * `messages` (Codes C3) is the caller's own Messages API request passed through as written (POST /v1/messages on AI):
++ * messages, system, tools, tool_choice, stop_sequences, temperature, max_tokens, and stream with Anthropic's own SSE
++ * event names (message_start, content_block_start, content_block_delta, content_block_stop, message_delta, message_stop,
++ * ping). DeepSeek and MiniMax publish Anthropic-compatible endpoints, so the same adapter serves them at another
++ * base_url.
*/
--const { postJson } = require('./common');
++const { postJson, ProviderError, deadline, describeAbort } = require('./common');
++const { sseFrames } = require('./stream');
const DEFAULT_BASE_URL = 'https://api.anthropic.com/v1';
const TEXT_OPS = ['chat', 'generate', 'summarize', 'classify', 'extract', 'enrich'];
++const MESSAGES_FIELDS = ['system', 'tools', 'tool_choice', 'stop_sequences', 'temperature', 'top_p', 'top_k', 'metadata'];
function buildBody({ model, system, messages, image, json, maxTokens, temperature }) {
const body = { model, max_tokens: Math.max(1, maxTokens) };
@@@ -31,12 +39,81 @@ function buildBody({ model, system, messages, image, json, maxTokens, temperatur
return body;
}
++/** The caller's Messages API body, with the routed model and only the fields they set. */
++function messagesBody(req) {
++ const body = { model: req.model, max_tokens: Math.max(1, Number(req.maxTokens) || 1024) };
++ if (Array.isArray(req.messages)) body.messages = req.messages;
++ for (const k of MESSAGES_FIELDS) if (req[k] !== undefined && req[k] !== null) body[k] = req[k];
++ return body;
++}
++
++const usageOf = (message) => {
++ const u = (message && message.usage) || {};
++ const cached = u.cache_read_input_tokens || 0;
++ return { input: (u.input_tokens || 0) + cached + (u.cache_creation_input_tokens || 0), output: u.output_tokens || 0, cached };
++};
++const textOf = (message) => (message && Array.isArray(message.content) ? message.content : []).filter(c => c && c.type === 'text').map(c => c.text).join('').trim();
++
++/**
++ * The message a streamed answer adds up to: content blocks by index (a tool call's input joined from its
++ * input_json_delta pieces and parsed), the stop reason, and the usage from message_start and message_delta.
++ */
++function createMessageAccumulator(model) {
++ let message = null;
++ let usage = {};
++ let stopReason = null;
++ let stopSequence = null;
++ const blocks = new Map();
++ return {
++ add(event, data) {
++ if (!data || typeof data !== 'object') return;
++ if (event === 'message_start') {
++ message = data.message && typeof data.message === 'object' ? data.message : null;
++ usage = { ...((message && message.usage) || {}) };
++ } else if (event === 'content_block_start') {
++ const block = { ...(data.content_block || {}) };
++ if (block.type === 'tool_use' && (block.input == null || typeof block.input !== 'object')) block.input = {};
++ blocks.set(Number(data.index) || 0, { block, json: '' });
++ } else if (event === 'content_block_delta') {
++ const slot = blocks.get(Number(data.index) || 0);
++ const d = data.delta || {};
++ if (!slot) return;
++ if (d.type === 'text_delta') slot.block.text = String(slot.block.text || '') + String(d.text || '');
++ else if (d.type === 'input_json_delta') slot.json += String(d.partial_json || '');
++ else if (d.type === 'thinking_delta') slot.block.thinking = String(slot.block.thinking || '') + String(d.thinking || '');
++ else if (d.type === 'signature_delta') slot.block.signature = String(slot.block.signature || '') + String(d.signature || '');
++ } else if (event === 'message_delta') {
++ const d = data.delta || {};
++ if (d.stop_reason !== undefined) stopReason = d.stop_reason;
++ if (d.stop_sequence !== undefined) stopSequence = d.stop_sequence;
++ if (data.usage) usage = { ...usage, ...data.usage };
++ }
++ },
++ result() {
++ const content = [...blocks.entries()].sort((a, b) => a[0] - b[0]).map(([, slot]) => {
++ const block = { ...slot.block };
++ if (block.type === 'tool_use') {
++ if (slot.json) { try { block.input = JSON.parse(slot.json); } catch { /* an incomplete tool call keeps what it has */ } }
++ if (block.input == null) block.input = {};
++ }
++ return block;
++ });
++ return { ...(message || {}), type: 'message', role: 'assistant', model: (message && message.model) || model, content, stop_reason: stopReason, stop_sequence: stopSequence, usage };
++ },
++ };
++}
++
function createAnthropicProvider(record, { apiKey = '', fetchImpl = globalThis.fetch } = {}) {
const baseUrl = String(record.base_url || DEFAULT_BASE_URL).replace(/\/+$/, '');
const caps = new Set(record.capabilities && record.capabilities.length ? record.capabilities.filter(f => f !== 'embed' && f !== 'transcribe') : [...TEXT_OPS, 'json', 'vision']);
++ // Anthropic authenticates with x-api-key. The Anthropic-compatible endpoints (DeepSeek, MiniMax) take a bearer token,
++ // so one is sent there too; api.anthropic.com gets only the header it documents.
++ const headers = () => (apiKey
++ ? { 'x-api-key': apiKey, 'anthropic-version': '2023-06-01', ...(/anthropic\.com/i.test(baseUrl) ? {} : { Authorization: `Bearer ${apiKey}` }) }
++ : { 'anthropic-version': '2023-06-01' });
async function chat(req) {
-- const j = await postJson(`${baseUrl}/messages`, { 'x-api-key': apiKey, 'anthropic-version': '2023-06-01' }, buildBody(req), { signal: req.signal, timeoutMs: req.timeoutMs, fetchImpl });
++ const j = await postJson(`${baseUrl}/messages`, headers(), buildBody(req), { signal: req.signal, timeoutMs: req.timeoutMs, fetchImpl });
const content = j.content || [];
let text = content.filter(c => c.type === 'text').map(c => c.text).join('').trim();
let jsonOut = null;
@@@ -47,9 +124,56 @@ function createAnthropicProvider(record, { apiKey = '', fetchImpl = globalThis.f
return { text, json: jsonOut, model: j.model || req.model, usage: { input: (u.input_tokens || 0) + cached + (u.cache_creation_input_tokens || 0), output: u.output_tokens || 0, cached } };
}
-- const adapter = { key: record.key, kind: 'anthropic', supports: (f) => caps.has(f), chat };
++ /** POST /v1/messages as the caller wrote it → { message, text, model, usage }. */
++ async function messages(req) {
++ const body = messagesBody(req);
++ if (typeof req.onDelta === 'function') return await messagesStreaming(body, req);
++ const j = await postJson(`${baseUrl}/messages`, headers(), body, { signal: req.signal, timeoutMs: req.timeoutMs, fetchImpl });
++ if (!j || j.type !== 'message' || !Array.isArray(j.content)) throw new ProviderError('message response has no content', { code: 'provider.output' });
++ return { message: j, text: textOf(j), model: j.model || req.model, usage: usageOf(j) };
++ }
++
++ /** The same request with stream: true: each event goes to req.onDelta as it arrives, then the answer it adds up to. */
++ async function messagesStreaming(body, req) {
++ let res;
++ try {
++ res = await fetchImpl(`${baseUrl}/messages`, {
++ method: 'POST', headers: { 'Content-Type': 'application/json', Accept: 'text/event-stream', ...headers() },
++ body: JSON.stringify({ ...body, stream: true }), signal: deadline(req.signal, req.timeoutMs), stream: true,
++ });
++ } catch (err) {
++ throw describeAbort(err, req.signal, req.timeoutMs) || new ProviderError(`connection failed: ${err.message}`, { code: 'provider.unreachable' });
++ }
++ if (!res.ok) {
++ const text = await res.text().catch(() => '');
++ let j = {};
++ try { j = JSON.parse(text); } catch { /* not JSON */ }
++ const msg = (j.error && (j.error.message || j.error)) || `HTTP ${res.status}: ${String(text).slice(0, 200)}`;
++ throw new ProviderError(typeof msg === 'string' ? msg : JSON.stringify(msg), { status: res.status, code: res.status === 401 || res.status === 403 ? 'provider.auth' : 'provider.http', body: j });
++ }
++ const acc = createMessageAccumulator(req.model);
++ try {
++ for await (const frame of sseFrames(res.body)) {
++ const event = frame.event || (frame.data && frame.data.type) || 'message';
++ if (event === 'error' || (frame.data && frame.data.type === 'error')) {
++ const e = (frame.data && frame.data.error) || {};
++ throw new ProviderError(String(e.message || 'the stream reported an error').slice(0, 300), { code: 'provider.http' });
++ }
++ acc.add(event, frame.data);
++ if (frame.data) req.onDelta({ event, data: frame.data });
++ }
++ } catch (err) {
++ if (err instanceof ProviderError) throw err;
++ throw describeAbort(err, req.signal, req.timeoutMs) || new ProviderError(`the stream broke: ${err.message}`, { code: 'provider.unreachable' });
++ }
++ const message = acc.result();
++ if (!Array.isArray(message.content) || !message.content.length) throw new ProviderError('message response has no content', { code: 'provider.output' });
++ return { message, text: textOf(message), model: message.model || req.model, usage: usageOf(message) };
++ }
++
++ const adapter = { key: record.key, kind: 'anthropic', supports: (f) => caps.has(f) || (f === 'messages' && caps.has('chat')), chat, messages };
for (const op of TEXT_OPS) if (op !== 'chat') adapter[op] = chat;
return adapter;
}
--module.exports = { createAnthropicProvider, buildBody, DEFAULT_BASE_URL };
++module.exports = { createAnthropicProvider, buildBody, messagesBody, DEFAULT_BASE_URL };
ddiff --git a/server/providers/index.js b/server/providers/index.js
iindex 0fc3e5c..c1feff2 100644
---- a/server/providers/index.js
++++ b/server/providers/index.js
@@@ -43,6 +43,9 @@ function rateLimited(err) {
return (err && err.status === 429) || (err && err.code === 'provider.rate_limited');
}
++/** The model a person's credential uses when they stored none, per provider (executeWithCredential). */
++const CREDENTIAL_DEFAULT_MODEL = { anthropic: 'claude-haiku-4-5-20251001', openai: 'gpt-4o-mini', deepseek: 'deepseek-chat', minimax: 'MiniMax-M2', openrouter: 'openrouter/auto' };
++
function createProviderPool({ db, registry, config, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, env = process.env, log = console, credentialFetch = null, machineModels = null, bench = null }) {
// A person's own machine as their model (credential provider node, server/machines.js).
const machines = machineModels || require('../machines').createMachineModels({ config, env, fetchImpl });
@@@ -333,9 +336,12 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
* declares `complete`, so without this the pool had no offer for /v1/chat/completions (2026-10-15).
*/
async function offeredCapabilities(p) {
-- if (!p.capabilities.includes('chat') || p.capabilities.includes('complete')) return p.capabilities;
++ if (!p.capabilities.includes('chat')) return p.capabilities;
++ const extra = ['complete', 'messages'].filter(f => !p.capabilities.includes(f));
++ if (!extra.length) return p.capabilities;
const entry = await adapter(p.key);
-- return entry && entry.adapter && entry.adapter.supports('complete') ? [...p.capabilities, 'complete'] : p.capabilities;
++ const serves = extra.filter(f => entry && entry.adapter && entry.adapter.supports(f));
++ return serves.length ? [...p.capabilities, ...serves] : p.capabilities;
}
async function candidates(route, features, operation, ctx = {}) {
@@@ -419,19 +425,34 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
// Text operations are all chat completions on the person's provider (templated workflows use generate).
// A project's own provider (ADR-056 secret, ai.complete) passes a Chat Completions request through as written.
const completing = operation === 'complete' && cred.provider === 'openai';
-- if (!completing && !['chat', 'generate', 'summarize', 'classify', 'extract', 'enrich'].includes(operation)) throw new AiError(422, 'credential.unsupported', `a person's own key is used for text only, not ${operation}`);
++ // Codes C3: POST /v1/messages on the person's own key. Only a provider that speaks Anthropic's Messages API serves
++ // it — Anthropic itself, DeepSeek and MiniMax (their own Anthropic-compatible endpoints) and a custom endpoint.
++ const messaging = operation === 'messages';
++ const anthropicFormat = ['anthropic', 'deepseek', 'minimax', 'custom'].includes(cred.provider);
++ if (messaging && !anthropicFormat) throw new AiError(422, 'credential.unsupported', `the ${cred.provider} credential does not speak the Messages API`);
++ if (!messaging && !completing && !['chat', 'generate', 'summarize', 'classify', 'extract', 'enrich'].includes(operation)) throw new AiError(422, 'credential.unsupported', `a person's own key is used for text only, not ${operation}`);
const guardedFetch = credentialFetch || require('./guarded-fetch').guardedFetch;
// The person's model for this role: the run's input.role, else the Live role its route names (live.director).
const wanted = ctx.role || ((/^live\.(chat|vision|director|summary|legacy)$/.exec(ctx.routeKey || '') || [])[1]) || null;
const role = wanted && cred.models[wanted] ? wanted : null;
-- const model = (req.image && cred.models.vision) || (role && cred.models[role]) || cred.models.chat || (cred.provider === 'anthropic' ? 'claude-haiku-4-5-20251001' : 'gpt-4o-mini');
++ // On the person's Anthropic key the model the caller asked for is theirs to choose; elsewhere the model they stored
++ // is the one their provider serves (a Claude Code model name does not exist at DeepSeek).
++ const model = messaging
++ ? ((cred.provider === 'anthropic' || cred.provider === 'custom') && req.model) || cred.models.chat || CREDENTIAL_DEFAULT_MODEL[cred.provider] || 'claude-haiku-4-5-20251001'
++ : (req.image && cred.models.vision) || (role && cred.models[role]) || cred.models.chat || CREDENTIAL_DEFAULT_MODEL[cred.provider] || 'gpt-4o-mini';
const record = { key: cred.providerKey, kind: cred.provider, base_url: cred.base_url, capabilities: cred.capabilities || ['chat', 'json', 'vision'], timeout_ms: 60000 };
-- const a = cred.provider === 'anthropic' ? createAnthropicProvider(record, { apiKey: cred.apiKey, fetchImpl: guardedFetch }) : createOpenAiProvider(record, { apiKey: cred.apiKey, fetchImpl: guardedFetch });
++ const a = anthropicFormat ? createAnthropicProvider(record, { apiKey: cred.apiKey, fetchImpl: guardedFetch }) : createOpenAiProvider(record, { apiKey: cred.apiKey, fetchImpl: guardedFetch });
const base = { operation, provider_key: cred.providerKey, model_key: model, route_key: ctx.routeKey || null, route_version: ctx.routeVersion || null, fallback: 0, prompt_hash: ctx.promptHash || null };
++ // An upstream's error text is kept in the request log and the run's error; a provider that echoes the key (in a
++ // 401 body, a URL) must not put it there, or in a log line, or in an answer.
++ const scrub = (text) => {
++ const t = String(text == null ? '' : text);
++ return cred.apiKey ? t.split(String(cred.apiKey)).join('[redacted]') : t;
++ };
const timeoutMs = Math.max(1000, Math.min(req.timeoutMs || 60000, 120000));
-- const call = completing ? a.complete : a.chat;
++ const call = messaging ? a.messages : completing ? a.complete : a.chat;
let streamed = false;
-- const onDelta = completing && typeof req.onDelta === 'function' ? (d) => { streamed = true; req.onDelta(d); } : null;
++ const onDelta = (completing || messaging) && typeof req.onDelta === 'function' ? (d) => { streamed = true; req.onDelta(d); } : null;
let lastErr = null;
for (let attempt = 0; attempt <= config.providerRetries; attempt++) {
if (ctx.signal && ctx.signal.aborted) throw new AiError(409, 'run.cancelled', 'run was cancelled');
@@@ -450,14 +471,14 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
} catch (err) {
lastErr = err;
if (ctx.signal && ctx.signal.aborted) throw new AiError(409, 'run.cancelled', 'run was cancelled');
-- if (ctx.logRequest) ctx.logRequest({ ...base, status: err && err.code === 'provider.timeout' ? 'timeout' : 'error', latency_ms: Date.now() - t0, error: String(err && err.message || err).slice(0, 500) });
-- if (err && err.code === 'EADDRNOTPUBLIC') throw new AiError(422, 'credential.endpoint_refused', err.message);
-- if (streamed) throw new AiError(502, 'provider.stream_broken', `the answer broke off after it began streaming: ${String(err && err.message || err).slice(0, 200)}`);
++ if (ctx.logRequest) ctx.logRequest({ ...base, status: err && err.code === 'provider.timeout' ? 'timeout' : 'error', latency_ms: Date.now() - t0, error: scrub(err && err.message || err).slice(0, 500) });
++ if (err && err.code === 'EADDRNOTPUBLIC') throw new AiError(422, 'credential.endpoint_refused', scrub(err.message));
++ if (streamed) throw new AiError(502, 'provider.stream_broken', `the answer broke off after it began streaming: ${scrub(err && err.message || err).slice(0, 200)}`);
if (attempt < config.providerRetries && retryable(err)) { await sleep(config.retryDelayMs, ctx.signal).catch(() => {}); continue; }
break;
}
}
-- 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: ${scrub(lastErr && lastErr.message || lastErr).slice(0, 200)}`);
}
/**
ddiff --git a/server/providers/stream.js b/server/providers/stream.js
iindex 2661ec2..819e2da 100644
---- a/server/providers/stream.js
++++ b/server/providers/stream.js
@@@ -5,6 +5,8 @@
*
* sseEvents(body) async iterator of the JSON objects in `data:` lines, until `data: [DONE]` (a web
* ReadableStream or any async iterable of bytes)
++ * sseFrames(body) async iterator of { event, data } for a stream that names its events (`event: message_start`),
++ * as Anthropic's Messages API does, until the stream ends
* createAccumulator() add(choiceDelta) for each chunk's choices; result() → the choices as a non-streamed answer
* (content concatenated, tool calls assembled by index with their arguments joined)
*/
@@@ -29,6 +31,48 @@ async function* sseEvents(body) {
}
}
++/**
++ * Server-sent events that carry their own name, as Anthropic's Messages API does:
++ *
++ * event: content_block_delta
++ * data: {"type":"content_block_delta",...}
++ *
++ * Yields { event, data } per frame (data parsed, null when it is not JSON). The frame's `event:` line wins; a stream that
++ * sends only `data:` (some Anthropic-compatible endpoints) is named by the data object's own `type`.
++ */
++async function* sseFrames(body) {
++ const dec = new TextDecoder();
++ let buf = '';
++ let event = null;
++ let data = [];
++ const flush = () => {
++ const out = { event, data: null };
++ const raw = data.join('\n').trim();
++ if (raw && raw !== '[DONE]') { try { out.data = JSON.parse(raw); } catch { out.data = null; } }
++ event = null;
++ data = [];
++ return raw === '[DONE]' ? null : out;
++ };
++ for await (const part of body) {
++ buf += dec.decode(part, { stream: true });
++ let i;
++ while ((i = buf.indexOf('\n')) >= 0) {
++ const line = buf.slice(0, i).replace(/\r$/, '');
++ buf = buf.slice(i + 1);
++ if (line === '') {
++ const frame = flush();
++ if (frame === null) return;
++ if (frame.event || frame.data) yield frame;
++ continue;
++ }
++ if (line.startsWith('event:')) event = line.slice(6).trim();
++ else if (line.startsWith('data:')) data.push(line.slice(5).trimStart());
++ }
++ }
++ const last = flush();
++ if (last && (last.event || last.data)) yield last;
++}
++
function createAccumulator() {
const choices = new Map();
return {
@@@ -62,4 +106,4 @@ function createAccumulator() {
};
}
--module.exports = { sseEvents, createAccumulator };
++module.exports = { sseEvents, sseFrames, createAccumulator };
ddiff --git a/server/providers/stub.js b/server/providers/stub.js
iindex 2b0fe47..7cfacf5 100644
---- a/server/providers/stub.js
++++ b/server/providers/stub.js
@@@ -179,12 +179,55 @@ function createStubProvider(record = { key: 'stub', metadata: {} }, { stats = nu
return { completion: { choices: [{ index: 0, message: { role: 'assistant', content: text }, finish_reason: 'stop' }] }, text, model: req.model || 'stub-1', synthetic: true, usage: { input: inTok, output: approxTokens(text), cached: 0, estimated: true } };
}
++ /** messages: an Anthropic-shaped answer, streamed as Anthropic's own events when asked (POST /v1/messages). */
++ async function messages(req) {
++ const msgs = Array.isArray(req.messages) ? req.messages : [];
++ const last = msgs[msgs.length - 1] || {};
++ const rnd = seeded(`messages:${JSON.stringify(msgs).slice(-2000)}`);
++ const inTok = msgs.reduce((n, m) => n + approxTokens(typeof m.content === 'string' ? m.content : JSON.stringify(m.content || '')), 0)
++ + approxTokens((Array.isArray(req.system) ? req.system : []).map(x => x && x.text).join(''));
++ const hasResult = Array.isArray(last.content) && last.content.some(c => c && c.type === 'tool_result');
++ const tool = Array.isArray(req.tools) && req.tools.length && !hasResult && !(req.tool_choice && req.tool_choice.type === 'none') ? req.tools[0] : null;
++ const model = req.model || 'stub-1';
++ const digest = crypto.createHash('sha256').update(JSON.stringify(msgs)).digest('hex').slice(0, 24);
++ let content;
++ let stop;
++ if (tool && tool.name) {
++ content = [{ type: 'tool_use', id: `toolu_${digest}`, name: tool.name, input: tool.input_schema ? synthesize(tool.input_schema, rnd, 0, 'input') : {} }];
++ stop = 'tool_use';
++ } else {
++ const text = hasResult ? `(synthetic) the tool answered: ${String(last.content.find(c => c && c.type === 'tool_result').content || '').slice(0, 80)}` : `(synthetic) ${phrase(rnd, 8)}`;
++ content = [{ type: 'text', text }];
++ stop = 'end_turn';
++ }
++ const outTok = approxTokens(JSON.stringify(content));
++ const message = { id: `msg_${digest}`, type: 'message', role: 'assistant', model, content, stop_reason: stop, stop_sequence: null,
++ usage: { input_tokens: inTok, output_tokens: outTok, cache_read_input_tokens: 0, cache_creation_input_tokens: 0 } };
++ if (typeof req.onDelta === 'function') {
++ req.onDelta({ event: 'message_start', data: { type: 'message_start', message: { ...message, content: [], stop_reason: null, usage: { ...message.usage, output_tokens: 0 } } } });
++ content.forEach((block, index) => {
++ if (block.type === 'text') {
++ req.onDelta({ event: 'content_block_start', data: { type: 'content_block_start', index, content_block: { type: 'text', text: '' } } });
++ for (const piece of block.text.split(/(?<= )/)) req.onDelta({ event: 'content_block_delta', data: { type: 'content_block_delta', index, delta: { type: 'text_delta', text: piece } } });
++ } else {
++ req.onDelta({ event: 'content_block_start', data: { type: 'content_block_start', index, content_block: { ...block, input: {} } } });
++ req.onDelta({ event: 'content_block_delta', data: { type: 'content_block_delta', index, delta: { type: 'input_json_delta', partial_json: JSON.stringify(block.input || {}) } } });
++ }
++ req.onDelta({ event: 'content_block_stop', data: { type: 'content_block_stop', index } });
++ });
++ req.onDelta({ event: 'message_delta', data: { type: 'message_delta', delta: { stop_reason: stop, stop_sequence: null }, usage: { output_tokens: outTok } } });
++ req.onDelta({ event: 'message_stop', data: { type: 'message_stop' } });
++ }
++ return { message, text: content.filter(c => c.type === 'text').map(c => c.text).join(''), model, synthetic: true, usage: { input: inTok, output: outTok, cached: 0, estimated: true } };
++ }
++
const adapter = {
key: record.key,
kind: 'stub',
synthetic: true,
supports: () => true,
complete,
++ messages,
transcribe: async (req) => await run('transcribe', req),
embed: async (req) => await run('embed', req),
image: async (req) => await run('image', req),
ddiff --git a/server/registry.js b/server/registry.js
iindex 6c942ec..d2519ad 100644
---- a/server/registry.js
++++ b/server/registry.js
@@@ -15,7 +15,7 @@ const { provenance } = require('./providers/rate-cards');
const PROVIDER_KINDS = ['stub', 'openai', 'responses', 'anthropic', 'http', 'whisper'];
const BILLING_PROFILES = ['metered', 'subscription', 'payg', 'free', 'byok'];
--const FEATURES = ['chat', 'complete', 'generate', 'summarize', 'classify', 'extract', 'enrich', 'embed', 'vision', 'json', 'tools', 'transcribe', 'image'];
++const FEATURES = ['chat', 'complete', 'messages', 'generate', 'summarize', 'classify', 'extract', 'enrich', 'embed', 'vision', 'json', 'tools', 'transcribe', 'image'];
const LIFECYCLE = ['draft', 'active', 'deprecated', 'archived'];
const KEY_RE = /^[a-z][a-z0-9_-]*(\.[a-z0-9_-]+)*$/;
const flag = (v) => v === true || v === 1 || /^(1|true|yes|on)$/i.test(String(v == null ? '' : v));
@@@ -350,7 +350,7 @@ function createRegistry(db, { clock = { now: () => Date.now() }, env = process.e
schemas.assertSchema(w.output_schema, 'output_schema');
if (!Array.isArray(w.steps) || !w.steps.length) throw new AiError(422, 'ai.invalid', 'workflow needs at least one step');
for (const s of w.steps) {
-- if (!s || !['llm', 'passthrough', 'complete', 'transcribe', 'embed', 'image', 'media_analysis', 'research'].includes(s.kind)) throw new AiError(422, 'ai.invalid', 'step kind must be llm, passthrough, complete, transcribe, embed, image, media_analysis or research');
++ if (!s || !['llm', 'passthrough', 'complete', 'messages', 'transcribe', 'embed', 'image', 'media_analysis', 'research'].includes(s.kind)) throw new AiError(422, 'ai.invalid', 'step kind must be llm, passthrough, complete, messages, transcribe, embed, image, media_analysis or research');
if ((s.kind === 'llm' || (s.kind === 'media_analysis' && s.template)) && !(s.template && await latestAny('templates', s.template))) throw new AiError(422, 'ai.invalid', `step references unknown template ${s.template}`);
}
if (!['none', 'private', 'service'].includes(w.cache_mode)) throw new AiError(422, 'ai.invalid', 'cache_mode must be none, private or service');
ddiff --git a/server/runs.js b/server/runs.js
iindex 7c4a05f..18f0628 100644
---- a/server/runs.js
++++ b/server/runs.js
@@@ -231,11 +231,13 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
}
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: the default one, or the provider the run names (Codes C3: one credential per provider).
const credentialSubject = body.credential && typeof body.credential === 'object' ? String(body.credential.subject || '') : null;
++ const credentialProvider = body.credential && typeof body.credential === 'object' && body.credential.provider != null ? String(body.credential.provider) : null;
if (credentialSubject !== null) {
if (!credentials) throw new AiError(503, 'credentials.unavailable', 'person credentials are not configured on this service');
if (requester.type !== 'service') throw new AiError(403, 'capability.denied', 'only the service that stored a credential runs with it');
-- await credentials.admit(requester.id, credentialSubject);
++ await credentials.admit(requester.id, credentialSubject, credentialProvider);
}
// An app's own provider (ADR-056 project secret): only on ai.complete, only for an app; its key spends the
// project's money at its provider, so no tier budget holds it (the app's quotas still count it).
@@@ -266,7 +268,8 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
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 && !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,
++ ...(credentialSubject !== null ? { credential_subject: credentialSubject } : {}), ...(credentialProvider !== null ? { credential_provider: credentialProvider } : {}),
++ ...(projectProvider ? { project_provider: projectProvider } : {}), ...(pinned ? { pinned_offer: pinned } : {}) }), created_at: now,
};
base.input = stableStringify(kept.value);
@@@ -397,7 +400,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
if (preferences.history === false) {
await db.prepare("UPDATE runs SET input = 'null', cache_key = NULL, options = (COALESCE(options, '{}')::jsonb || jsonb_build_object('input_retained', false, 'history', false))::text WHERE id = ?").run(id);
}
-- const credential = opts.credential_subject ? await credentials.forRun(row.requester_id, opts.credential_subject)
++ const credential = opts.credential_subject ? await credentials.forRun(row.requester_id, opts.credential_subject, opts.credential_provider || null)
: opts.project_provider ? await projectCredential(opts.project_provider, ctx) : null;
const r = await engine.execute(wf, row, input, { signal: controller.signal, logRequest, debugRaw: Boolean(config.debugRawLog && opts.debug), inputHash: row.input_hash, preferences,
// A developer app's run (ADR-014): the provider pool excludes a person's byok key, and the run is
ddiff --git a/server/workflows/core.js b/server/workflows/core.js
iindex 49f960f..26579b8 100644
---- a/server/workflows/core.js
++++ b/server/workflows/core.js
@@@ -110,6 +110,34 @@ const workflows = [
metadata: { passthrough: true },
steps: [{ kind: 'complete', route: 'default.chat' }], cache_mode: 'none',
},
++ {
++ // The Anthropic-compatible endpoint (POST /v1/messages, server/api/messages.js): the caller's Messages API request,
++ // tools and streaming included, passed to an upstream that speaks Anthropic's format; the run keeps its usage,
++ // cost and explain but not the prompt (passthrough). The output is the message it answered with.
++ key: 'ai.messages', name: 'Messages API (Anthropic-compatible)', namespace: 'ai',
++ input_schema: OBJ({
++ messages: { type: 'array', minItems: 1, maxItems: 500, items: { type: 'object', required: ['role'], properties: {
++ role: { enum: ['user', 'assistant'] },
++ content: { anyOf: [STR(400000), { type: 'array', maxItems: 500 }, { type: 'null' }] },
++ } } },
++ system: { anyOf: [STR(400000), { type: 'array', maxItems: 500, items: { type: 'object' } }] },
++ tools: { type: 'array', maxItems: 128, items: { type: 'object' } },
++ tool_choice: { type: 'object' },
++ max_tokens: { type: 'integer', minimum: 1, maximum: 64000 },
++ stop_sequences: { type: 'array', maxItems: 8, items: STR(200) },
++ temperature: { type: ['number', 'null'], minimum: 0, maximum: 1 },
++ top_p: { type: ['number', 'null'], minimum: 0, maximum: 1 },
++ top_k: { type: ['integer', 'null'], minimum: 0 },
++ metadata: { type: 'object' },
++ }, ['messages', 'max_tokens']),
++ output_schema: OBJ({
++ id: STR(200), type: STR(20), role: STR(20), model: { type: ['string', 'null'] },
++ content: { type: 'array', minItems: 1, maxItems: 500, items: { type: 'object' } },
++ stop_reason: { type: ['string', 'null'] }, stop_sequence: { type: ['string', 'null'] }, usage: { type: 'object' },
++ }, ['content']),
++ metadata: { passthrough: true },
++ steps: [{ kind: 'messages', route: 'default.messages' }], cache_mode: 'none',
++ },
{
// Off until AI_IMAGE_MODEL seeds route default.image: without it a run answers 503 route.unavailable.
key: 'ai.image', name: 'Generate an image', namespace: 'ai',
ddiff --git a/server/workflows/engine.js b/server/workflows/engine.js
iindex f8049a5..36acd21 100644
---- a/server/workflows/engine.js
++++ b/server/workflows/engine.js
@@@ -213,6 +213,27 @@ function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spa
return { output: { choices: c.choices, system_fingerprint: c.system_fingerprint || null, model: exec.model }, exec, route };
}
++ /**
++ * messages: the caller's Anthropic Messages API request as they wrote it (ai.messages, POST /v1/messages), to an
++ * upstream that speaks that format. The run's output is the message (consumed by server/api/messages.js).
++ */
++ async function messagesStep(step, input, wf, run, ctx) {
++ const route = await routeFor(step.route || 'default.messages');
++ const req = {
++ messages: input.messages,
++ maxTokens: Math.max(1, Math.round(input.max_tokens || route.max_output_tokens || 1024)),
++ timeoutMs: Math.max(route.timeout_ms || 0, 120000),
++ ...(typeof ctx.onDelta === 'function' ? { onDelta: ctx.onDelta } : {}),
++ };
++ for (const k of ['system', 'tools', 'tool_choice', 'stop_sequences', 'temperature', 'top_p', 'top_k', 'metadata']) {
++ if (input[k] !== undefined) req[k] = input[k];
++ }
++ const exec = await pool.execute(route, 'messages', req, { ...ctx, routeKey: route.key, routeVersion: route.version, promptHash: sha256({ m: input.messages, t: input.tools || null }) });
++ const m = exec.result.message;
++ if (!m || !Array.isArray(m.content)) throw new AiError(502, 'output.empty', 'the provider answered, but not with a message');
++ return { output: { ...m, model: m.model || exec.model }, exec, route };
++ }
++
async function transcribeStep(step, input, wf, run, ctx) {
const route = await routeFor(step.route || 'live.stt');
// Read where it lies (the loopback reader), one window at a time: start_sec seeks with a range request, and
@@@ -277,7 +298,7 @@ function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spa
return { cited: [], gaps: gaps.slice(0, 20), read: (output && output.sources ? output.sources.length : 0) };
}
-- const STEP = { llm: llmStep, passthrough: passthroughStep, complete: completeStep, transcribe: transcribeStep, embed: embedStep, image: imageStep, media_analysis: mediaAnalysis.step, research: researchStep };
++ const STEP = { llm: llmStep, passthrough: passthroughStep, complete: completeStep, messages: messagesStep, transcribe: transcribeStep, embed: embedStep, image: imageStep, media_analysis: mediaAnalysis.step, research: researchStep };
/**
* Run a workflow. ctx: { signal, logRequest, debugRaw, inputHash }
ddiff --git a/server/workflows/seed.js b/server/workflows/seed.js
iindex 4622d5a..c61119e 100644
---- a/server/workflows/seed.js
++++ b/server/workflows/seed.js
@@@ -6,7 +6,7 @@
* AI_MODEL), http-seam (AI_HTTP_SEAM_URL), whisper (WHISPER_*), local (AI_LOCAL_LLM_URL)
* — records whose origin is 'seed' follow the environment; admin-edited ones are left alone
* models the configured default model
-- * routes live.<role> for Live's roles, live.stt, live.creative (AI_CREATIVE_MODEL), default.chat|json|embedding and media.local|paid
++ * routes live.<role> for Live's roles, live.stt, live.creative (AI_CREATIVE_MODEL), default.chat|json|messages|embedding and media.local|paid
* as capability pools ordered by openvibe-sdk/placement; and the historical
* product route keys as explicit aliases of default.json
* templates, workflows from server/workflows/{live,core,products,media,chat}.js — a code change becomes a
@@@ -134,6 +134,10 @@ async function seed({ registry, quotas, config, env = process.env, db }) {
// embeddings without a local model), an unavailable sole provider still answers the explicit 503.
await registry.seedVersioned('route', 'live.stt', pool(600000, { 'capability': 'transcribe', constraints: { objective: 'balanced' } }));
await registry.seedVersioned('route', 'default.chat', pool(30000, { max_output_tokens: 800 }));
++ // POST /v1/messages (Codes C3): only providers that speak Anthropic's Messages API join this pool — the shared
++ // provider when it is Anthropic's own API, DeepSeek's and MiniMax's Anthropic-compatible endpoints, and any record an
++ // admin adds as kind anthropic. The others are skipped as unsupported, never converted.
++ await registry.seedVersioned('route', 'default.messages', pool(120000, { constraints: { objective: 'correctness', latency_class: 'interactive' }, max_output_tokens: 8192 }));
await registry.seedVersioned('route', 'default.json', pool(60000, { options: { temperature: 0.3 }, max_output_tokens: 2400, response_format: 'json' }));
await registry.seedVersioned('route', 'default.embedding', pool(30000, { 'capability': 'embed', constraints: { objective: 'balanced' } }));
// media.analyze (WS-O task 5): the local model when there is one; the paid route is used only under a media.paid budget.
ddiff --git a/test/credentials.test.js b/test/credentials.test.js
iindex 4edb4ca..4fdd56a 100644
---- a/test/credentials.test.js
++++ b/test/credentials.test.js
@@@ -6,8 +6,11 @@
// logged under byo:<owner>:<subject>, and its own daily budget stops further runs (429);
// - the guarded fetch reaches only public https endpoints.
const assert = require('assert');
++const crypto = require('crypto');
++const { validate } = require('openvibe-contracts');
const { boot, request, token, suite, ALL } = require('./helpers');
const { guardedFetch } = require('../server/providers/guarded-fetch');
++const { createCredentials, aadFor } = require('../server/credentials');
const t = suite('credentials');
const KEY = 'ab'.repeat(32);
@@@ -17,15 +20,20 @@ const live = token('live', [...ALL, 'ai.credential.manage']);
const tools = token('tools', [...ALL, 'ai.credential.manage']);
const noCap = token('live', ALL);
const calls = [];
++const logs = []; // every line the service logs, for "the key is never in one"
++const capture = { log: (...a) => logs.push(a.join(' ')), warn: (...a) => logs.push(a.join(' ')), error: (...a) => logs.push(a.join(' ')) };
++let upstream = 'ok'; // 'echo401': the provider refuses and echoes the key it was sent
// The person's provider: an OpenAI-compatible endpoint, answered in-process (the guarded fetch is tested below).
const credentialFetch = async (url, init) => {
-- calls.push({ url, auth: init.headers.Authorization || init.headers.authorization, body: JSON.parse(init.body) });
++ const sent = init.headers['x-api-key'] || init.headers.Authorization || init.headers.authorization;
++ calls.push({ url, auth: sent, body: JSON.parse(init.body) });
++ if (upstream === 'echo401') return new Response(JSON.stringify({ error: { message: `invalid key ${sent}` } }), { status: 401, headers: { 'content-type': 'application/json' } });
return new Response(JSON.stringify({ model: 'gpt-4o-mini', choices: [{ message: { role: 'assistant', content: 'hello from dana\'s key' } }], usage: { prompt_tokens: 1000, completion_tokens: 1000 } }), { status: 200, headers: { 'content-type': 'application/json' } });
};
let h;
t.test('boot with a credentials key', async () => {
-- h = await boot({ env: { AI_CREDENTIALS_KEY: KEY, AI_PRICING_JSON: JSON.stringify({ 'gpt-4o-mini': { in: 1, out: 1 } }) }, credentialFetch });
++ h = await boot({ env: { AI_CREDENTIALS_KEY: KEY, AI_PRICING_JSON: JSON.stringify({ 'gpt-4o-mini': { in: 1, out: 1 } }) }, credentialFetch, log: capture });
});
t.test('store, read without the key, refuse other services', async () => {
@@@ -41,7 +49,7 @@ t.test('store, read without the key, refuse other services', async () => {
const { validate } = require('openvibe-contracts');
assert.ok(validate('ai.credential@1', r.body).valid, JSON.stringify(validate('ai.credential@1', r.body).errors));
const row = await h.db.prepare('SELECT key_enc FROM subject_credentials WHERE owner = ? AND subject = ?').get('live', DANA);
-- assert.ok(row.key_enc.startsWith('v1:') && !row.key_enc.includes(API_KEY), 'encrypted at rest');
++ assert.ok(row.key_enc.startsWith('v2:') && !row.key_enc.includes(API_KEY), 'encrypted at rest, bound to its provider');
// A models change keeps the stored key; a new endpoint needs it again.
r = await request(h.base, 'PUT', `/api/v1/credentials/${DANA}`, { tok: live, body: { provider: 'openai', base_url: 'https://llm.example.com/v1', models: { chat: 'gpt-4o-mini', vision: 'gpt-4o' }, budget_usd_per_day: 0.003 } });
assert.deepStrictEqual([r.status, r.body.key_hint, r.body.models.vision], [200, '…WXYZ', 'gpt-4o']);
@@@ -103,6 +111,108 @@ t.test('the guarded fetch reaches only public https endpoints', async () => {
}
});
++
++// Codes C3: one credential per provider. The provider-less route is the default one (what a v1 caller stored).
++t.test('one credential per provider: the legacy route is the default, each provider keeps its own', async () => {
++ const FRAN = 'usr_01JAB2C3D4E5F6G7H8J9K0MNP3';
++ const DEEPSEEK_KEY = 'sk-test-deepseek-0123456789';
++ const ANTHROPIC_KEY = 'sk-ant-test-0123456789FGHI';
++ let r = await request(h.base, 'PUT', `/api/v1/credentials/${FRAN}/deepseek`, { tok: live, body: { key: DEEPSEEK_KEY, models: { chat: 'deepseek-chat' } } });
++ assert.strictEqual(r.status, 200, r.text);
++ assert.deepStrictEqual([r.body.provider, r.body.key_hint], ['deepseek', '…6789']);
++ assert.ok(!r.text.includes(DEEPSEEK_KEY), 'the key never comes back');
++ assert.ok(validate('ai.credential@2', r.body).valid, JSON.stringify(validate('ai.credential@2', r.body).errors));
++ // A provider added on its own is not the default: the v1 route still has none.
++ assert.strictEqual((await request(h.base, 'GET', `/api/v1/credentials/${FRAN}`, { tok: live })).status, 404);
++ assert.strictEqual((await request(h.base, 'GET', `/api/v1/credentials/${FRAN}/deepseek`, { tok: live })).body.provider, 'deepseek');
++ // The legacy route stores the default provider.
++ r = await request(h.base, 'PUT', `/api/v1/credentials/${FRAN}`, { tok: live, body: { provider: 'anthropic', key: ANTHROPIC_KEY, models: { chat: 'claude-haiku-4-5-20251001' } } });
++ assert.strictEqual(r.status, 200, r.text);
++ assert.strictEqual(r.body.provider, 'anthropic');
++ assert.strictEqual((await request(h.base, 'GET', `/api/v1/credentials/${FRAN}`, { tok: live })).body.provider, 'anthropic', 'the default is the v1 route');
++ assert.strictEqual((await request(h.base, 'GET', `/api/v1/credentials/${FRAN}/anthropic`, { tok: live })).body.provider, 'anthropic');
++ assert.strictEqual((await request(h.base, 'GET', `/api/v1/credentials/${FRAN}/deepseek`, { tok: live })).body.provider, 'deepseek', 'the other provider is still there');
++ // Changing the default's budget through its own provider route keeps it the default (the flag is not cleared).
++ r = await request(h.base, 'PUT', `/api/v1/credentials/${FRAN}/anthropic`, { tok: live, body: { budget_usd_per_day: 2, models: { chat: 'claude-haiku-4-5-20251001' } } });
++ assert.strictEqual(r.status, 200, r.text);
++ assert.strictEqual((await request(h.base, 'GET', `/api/v1/credentials/${FRAN}`, { tok: live })).body.provider, 'anthropic', 'still the default after a per-provider update');
++ // A body that names another provider than the route is refused; an unknown provider is a miss.
++ r = await request(h.base, 'PUT', `/api/v1/credentials/${FRAN}/deepseek`, { tok: live, body: { provider: 'anthropic', key: ANTHROPIC_KEY } });
++ assert.deepStrictEqual([r.status, r.body.code], [422, 'input.invalid']);
++ assert.strictEqual((await request(h.base, 'GET', `/api/v1/credentials/${FRAN}/groq`, { tok: live })).status, 404);
++ // custom needs its https endpoint (ai.credential-put@2).
++ r = await request(h.base, 'PUT', `/api/v1/credentials/${FRAN}/custom`, { tok: live, body: { key: 'sk-test-custom-0123456789' } });
++ assert.deepStrictEqual([r.status, r.body.code], [422, 'input.invalid']);
++ r = await request(h.base, 'PUT', `/api/v1/credentials/${FRAN}/custom`, { tok: live, body: { key: 'sk-test-custom-0123456789', base_url: 'https://llm.example.com/anthropic' } });
++ assert.strictEqual(r.status, 200, r.text);
++ assert.strictEqual(r.body.base_url, 'https://llm.example.com/anthropic');
++ // Deleting one provider leaves the others, and the default falls back to what is left.
++ assert.strictEqual((await request(h.base, 'DELETE', `/api/v1/credentials/${FRAN}/anthropic`, { tok: live })).status, 204);
++ r = await request(h.base, 'GET', `/api/v1/credentials/${FRAN}`, { tok: live });
++ assert.strictEqual(r.status, 200);
++ assert.ok(['deepseek', 'custom'].includes(r.body.provider), r.body.provider);
++});
++
++// The additional data binds the provider: a row moved to another provider does not decrypt.
++t.test('the AAD binds the provider: a row moved to another provider does not decrypt', async () => {
++ const GUS = 'usr_01JAB2C3D4E5F6G7H8J9K0MNP4';
++ const key = 'sk-test-gus-0123456789ABCD';
++ let r = await request(h.base, 'PUT', `/api/v1/credentials/${GUS}/anthropic`, { tok: live, body: { key, models: { chat: 'claude-haiku-4-5-20251001' } } });
++ assert.strictEqual(r.status, 200, r.text);
++ const row = await h.db.prepare('SELECT key_enc FROM subject_credentials WHERE owner = ? AND subject = ? AND provider = ?').get('live', GUS, 'anthropic');
++ assert.ok(row.key_enc.startsWith('v2:') && !row.key_enc.includes(key));
++ // Exactly what a person with the database (or a bad migration) could do: move the sealed key to another provider.
++ await h.db.prepare('UPDATE subject_credentials SET provider = ? WHERE owner = ? AND subject = ? AND provider = ?').run('deepseek', 'live', GUS, 'anthropic');
++ const creds = createCredentials({ db: h.db, config: h.config });
++ await assert.rejects(creds.forRun('live', GUS, 'deepseek'), /unable to authenticate|bad decrypt|Unsupported state/i, 'the provider is part of the AAD');
++ // The AAD string itself names all three.
++ assert.strictEqual(aadFor('live', GUS, 'deepseek'), `live|${GUS}|deepseek`);
++});
++
++// Rows written before the provider dimension (v1: bound to owner|subject) still read, and are rewritten as v2.
++t.test('an old v1 row still decrypts, and is upgraded to the provider-bound AAD on first use', async () => {
++ const HELEN = 'usr_01JAB2C3D4E5F6G7H8J9K0MNP5';
++ const key = 'sk-test-helen-0123456789JKLM';
++ const iv = crypto.randomBytes(12);
++ const c = crypto.createCipheriv('aes-256-gcm', Buffer.from(KEY, 'hex'), iv);
++ c.setAAD(Buffer.from(`live|${HELEN}`)); // the pre-C3 AAD
++ const body = Buffer.concat([c.update(key, 'utf8'), c.final()]);
++ const sealed = `v1:${iv.toString('base64')}:${c.getAuthTag().toString('base64')}:${body.toString('base64')}`;
++ const at = new Date().toISOString();
++ await h.db.prepare(`INSERT INTO subject_credentials (owner, subject, provider, base_url, key_enc, key_hint, models, budget_usd_day, is_default, spend_key, created_at, updated_at)
++ VALUES (?, ?, 'anthropic', NULL, ?, ?, '{}', NULL, false, ?, ?, ?)`).run('live', HELEN, sealed, `…${key.slice(-4)}`, `byo:live:${HELEN}`, at, at);
++ const creds = createCredentials({ db: h.db, config: h.config });
++ assert.strictEqual((await creds.forRun('live', HELEN, 'anthropic')).apiKey, key, 'an old row still decrypts with the old AAD');
++ const after = await h.db.prepare('SELECT key_enc FROM subject_credentials WHERE owner = ? AND subject = ? AND provider = ?').get('live', HELEN, 'anthropic');
++ assert.ok(after.key_enc.startsWith('v2:'), 'rewritten with the provider-bound AAD');
++ assert.strictEqual((await creds.forRun('live', HELEN, 'anthropic')).apiKey, key, 'and it reads back');
++ const got = await request(h.base, 'GET', `/api/v1/credentials/${HELEN}/anthropic`, { tok: live });
++ assert.deepStrictEqual([got.status, got.body.provider], [200, 'anthropic']);
++ assert.ok(!got.text.includes(key));
++ assert.ok(validate('ai.credential@2', got.body).valid, JSON.stringify(validate('ai.credential@2', got.body).errors));
++});
++
++// A key is never in a response, a log line or an error, even when the provider echoes it back.
++t.test('a key never appears in a response, a log line or an error', async () => {
++ const IVAN = 'usr_01JAB2C3D4E5F6G7H8J9K0MNP6';
++ const key = 'sk-test-ivan-LEAKME-0123456789';
++ let r = await request(h.base, 'PUT', `/api/v1/credentials/${IVAN}/openai`, { tok: live, body: { key, models: { chat: 'gpt-4o-mini' } } });
++ assert.strictEqual(r.status, 200, r.text);
++ upstream = 'echo401';
++ r = await request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok: live, body: { workflow: 'live.viewers.line', input: { role: 'chat', user: 'say hi' }, credential: { subject: IVAN, provider: 'openai' } } });
++ upstream = 'ok';
++ assert.strictEqual(r.status, 201, r.text);
++ assert.strictEqual(r.body.run.status, 'failed');
++ assert.ok(!r.text.includes(key), 'not in the run response');
++ const run = await request(h.base, 'GET', `/api/v1/runs/${r.body.run.id}`, { tok: live });
++ assert.ok(!run.text.includes(key), 'not in the run read back');
++ const reqs = await h.db.prepare('SELECT error FROM requests WHERE run_id = ?').all(r.body.run.id);
++ assert.ok(reqs.length && reqs.every((x) => !String(x.error || '').includes(key)), 'not in the request log');
++ assert.ok(!logs.join('\n').includes(key), 'not in a log line');
++ assert.ok(!calls[calls.length - 1].body.messages.some((m) => String(m.content).includes(key)));
++ await request(h.base, 'DELETE', `/api/v1/credentials/${IVAN}/openai`, { tok: live });
++});
++
t.test('without AI_CREDENTIALS_KEY the routes answer 503', async () => {
const off = await boot({});
const r = await request(off.base, 'PUT', `/api/v1/credentials/${DANA}`, { tok: live, body: { provider: 'openai', api_key: API_KEY } });
ddiff --git a/test/helpers.js b/test/helpers.js
iindex 79e96ae..c038637 100644
---- a/test/helpers.js
++++ b/test/helpers.js
@@@ -45,7 +45,7 @@ function tmpDir() {
// A boot on a dir used before is a restart on the same database: its database stays open.
const restartable = new Map();
--async function boot({ env = {}, dir, clock, fetchImpl, credentialFetch, research, machineModels } = {}) {
++async function boot({ env = {}, dir, clock, fetchImpl, credentialFetch, research, machineModels, log } = {}) {
const reuse = dir !== undefined;
if (dir === undefined) dir = tmpDir();
const fullEnv = {
@@@ -59,7 +59,7 @@ async function boot({ env = {}, dir, clock, fetchImpl, credentialFetch, research
if (reuse) restartable.set(dir, testdb);
// Research reads a canned web (test/fixtures/research-fetch.js) unless a test brings its own: no test reaches the internet.
const researcher = research || require('../server/research').createResearch({ config: config.research, env: {}, fetchImpl: require('./fixtures/research-fetch').createResearchFetch().fetchImpl, log: silent });
-- const h = await start({ config, db: testdb.db, env: fullEnv, log: silent, clock, fetchImpl, credentialFetch, research: researcher, machineModels });
++ const h = await start({ config, db: testdb.db, env: fullEnv, log: log || silent, clock, fetchImpl, credentialFetch, research: researcher, machineModels });
const base = `http://127.0.0.1:${h.server.address().port}`;
return { ...h, base, dir, env: fullEnv, testdb, stop: async () => { await h.close(); if (!reuse) await testdb.close(); } };
}
ddiff --git a/test/messages-api.test.js b/test/messages-api.test.js
nnew file mode 100644
iindex 0000000..7503983
---- /dev/null
++++ b/test/messages-api.test.js
@@@ -0,0 +1,199 @@
++'use strict';
++// POST /v1/messages, Anthropic's Messages API on AI (Codes C3): a run of ai.messages behind the same Network-token auth,
++// quotas, metering and audit as /v1/chat/completions.
++// - text, tools and a tool result pass through as written to an upstream that speaks Anthropic's format;
++// - stream: true answers Anthropic's own SSE events, in order, and the run ends with the whole message;
++// - x-openvibe-credential: <provider> uses the person's own key instead of the router, and that key never appears in a
++// response, a log line or an error;
++// - auth is refused without ai.run.create, and errors are Anthropic's shape.
++const assert = require('assert');
++const nodeHttp = require('http');
++const { boot, request, token, suite, ALL } = require('./helpers');
++
++const t = suite('messages-api');
++const KEY = 'cd'.repeat(32);
++const PERSON = 'usr_01JAB2C3D4E5F6G7H8J9K0MNP7';
++const PERSON_KEY = 'sk-ant-test-person-0123456789';
++const service = token('live', ALL);
++const noCap = token('live', ALL.filter((c) => c !== 'ai.run.create'));
++const acting = token('codes', ALL, { extra: { on_behalf_of: PERSON } });
++const manager = token('codes', [...ALL, 'ai.credential.manage'], { extra: { on_behalf_of: PERSON } });
++const WEATHER = { name: 'get_weather', description: 'Weather for a city', input_schema: { type: 'object', required: ['city'], properties: { city: { type: 'string' } } } };
++
++// A fake Anthropic upstream: a message in one piece, or Anthropic's SSE frames in order.
++const upstream = { calls: [], mode: 'text', delayMs: 5 };
++function messageFor(body) {
++ if (upstream.mode === 'tool') {
++ return { id: 'msg_tool_1', type: 'message', role: 'assistant', model: body.model, content: [
++ { type: 'text', text: 'Let me check.' },
++ { type: 'tool_use', id: 'toolu_01', name: 'get_weather', input: { city: 'Lisbon' } },
++ ], stop_reason: 'tool_use', stop_sequence: null, usage: { input_tokens: 21, output_tokens: 9 } };
++ }
++ return { id: 'msg_test_1', type: 'message', role: 'assistant', model: body.model, content: [{ type: 'text', text: 'Hello there' }],
++ stop_reason: 'end_turn', stop_sequence: null, usage: { input_tokens: 11, output_tokens: 2 } };
++}
++const FRAMES = [
++ ['message_start', { type: 'message_start', message: { id: 'msg_stream_1', type: 'message', role: 'assistant', model: 'claude-test-1', content: [], stop_reason: null, stop_sequence: null, usage: { input_tokens: 11, output_tokens: 0 } } }],
++ ['ping', { type: 'ping' }],
++ ['content_block_start', { type: 'content_block_start', index: 0, content_block: { type: 'text', text: '' } }],
++ ['content_block_delta', { type: 'content_block_delta', index: 0, delta: { type: 'text_delta', text: 'Hello' } }],
++ ['content_block_delta', { type: 'content_block_delta', index: 0, delta: { type: 'text_delta', text: ' there' } }],
++ ['content_block_stop', { type: 'content_block_stop', index: 0 }],
++ ['message_delta', { type: 'message_delta', delta: { stop_reason: 'end_turn', stop_sequence: null }, usage: { output_tokens: 2 } }],
++ ['message_stop', { type: 'message_stop' }],
++];
++async function startUpstream() {
++ const server = nodeHttp.createServer((req, res) => {
++ let raw = '';
++ req.on('data', (c) => { raw += c; });
++ req.on('end', async () => {
++ const body = JSON.parse(raw || '{}');
++ upstream.calls.push({ url: req.url, headers: req.headers, body });
++ if (body.stream) {
++ res.writeHead(200, { 'Content-Type': 'text/event-stream' });
++ for (const [event, data] of FRAMES) {
++ res.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`);
++ await new Promise((r) => setTimeout(r, upstream.delayMs));
++ }
++ return res.end();
++ }
++ res.writeHead(200, { 'Content-Type': 'application/json' });
++ return res.end(JSON.stringify(messageFor(body)));
++ });
++ });
++ await new Promise((r) => server.listen(0, '127.0.0.1', r));
++ upstream.url = `http://127.0.0.1:${server.address().port}`;
++ upstream.close = () => new Promise((r) => { server.closeAllConnections?.(); server.close(() => r()); });
++ return upstream;
++}
++
++// The person's own Anthropic key, answered in-process (the guarded fetch is replaced, as in credentials.test.js).
++const ownCalls = [];
++const credentialFetch = async (url, init) => {
++ const body = JSON.parse(init.body);
++ ownCalls.push({ url, key: init.headers['x-api-key'], body });
++ if (body.stream) {
++ const frames = FRAMES.map(([event, data]) => `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`).join('');
++ return new Response(frames, { status: 200, headers: { 'content-type': 'text/event-stream' } });
++ }
++ return new Response(JSON.stringify(messageFor(body)), { status: 200, headers: { 'content-type': 'application/json' } });
++};
++
++const logs = [];
++const capture = { log: (...a) => logs.push(a.join(' ')), warn: (...a) => logs.push(a.join(' ')), error: (...a) => logs.push(a.join(' ')) };
++let h;
++
++const sse = async (res) => {
++ const text = await res.text();
++ return text.split('\n\n').filter(Boolean).map((block) => {
++ const e = /^event: (.+)$/m.exec(block);
++ const d = /^data: (.*)$/m.exec(block);
++ return { event: e && e[1].trim(), data: d ? JSON.parse(d[1]) : null };
++ });
++};
++const post = (body, { tok = service, headers = {} } = {}) => fetch(`${h.base}/v1/messages`, { method: 'POST', headers: { Authorization: `Bearer ${tok}`, 'Content-Type': 'application/json', ...headers }, body: JSON.stringify(body) });
++
++t.test('boot with a fake Anthropic upstream as the router\'s Messages provider', async () => {
++ await startUpstream();
++ h = await boot({ env: { AI_CREDENTIALS_KEY: KEY, AI_STUB_FALLBACK: 'false' }, credentialFetch, log: capture });
++ const r = await request(h.base, 'POST', '/api/v1/providers', { tok: service, body: { key: 'anthropic-fake', kind: 'anthropic', base_url: upstream.url, auth_mode: 'none', default_model: 'claude-test-1', capabilities: ['chat', 'json'], timeout_ms: 5000 } });
++ assert.strictEqual(r.status, 201, r.text);
++});
++
++t.test('a message answers Anthropic\'s shape, from a run of ai.messages that keeps no prompt', async () => {
++ upstream.mode = 'text';
++ const res = await post({ model: 'claude-sonnet-4-5', max_tokens: 64, system: 'Be brief.', messages: [{ role: 'user', content: 'Say hello' }] });
++ assert.strictEqual(res.status, 200);
++ const body = await res.json();
++ assert.deepStrictEqual([body.type, body.role, body.id], ['message', 'assistant', 'msg_test_1']);
++ assert.deepStrictEqual(body.content, [{ type: 'text', text: 'Hello there' }]);
++ assert.deepStrictEqual([body.stop_reason, body.usage.input_tokens, body.usage.output_tokens], ['end_turn', 11, 2]);
++ const runId = res.headers.get('x-openvibe-run-id');
++ assert.match(String(runId), /^run_/);
++ const run = await request(h.base, 'GET', `/api/v1/runs/${runId}`, { tok: service });
++ assert.strictEqual(run.body.run.workflow.key, 'ai.messages');
++ assert.ok(!JSON.stringify(run.body.run.input || {}).includes('Say hello'), 'the prompt is not kept');
++ // The caller's fields reached the upstream as written, with the routed model.
++ const sent = upstream.calls[upstream.calls.length - 1].body;
++ assert.deepStrictEqual([sent.model, sent.system, sent.max_tokens], ['claude-test-1', 'Be brief.', 64]);
++ assert.deepStrictEqual(sent.messages, [{ role: 'user', content: 'Say hello' }]);
++});
++
++t.test('tools and tool_choice pass through; a tool_use comes back as Anthropic sends it', async () => {
++ upstream.mode = 'tool';
++ const res = await post({ max_tokens: 64, tools: [WEATHER], tool_choice: { type: 'auto' }, messages: [{ role: 'user', content: 'Weather in Lisbon?' }] });
++ assert.strictEqual(res.status, 200);
++ const body = await res.json();
++ assert.strictEqual(body.stop_reason, 'tool_use');
++ assert.deepStrictEqual(body.content[1], { type: 'tool_use', id: 'toolu_01', name: 'get_weather', input: { city: 'Lisbon' } });
++ const sent = upstream.calls[upstream.calls.length - 1].body;
++ assert.deepStrictEqual([sent.tools, sent.tool_choice], [[WEATHER], { type: 'auto' }]);
++ // The tool result goes back as a user content block, as Anthropic's clients send it.
++ const second = await post({ max_tokens: 64, messages: [{ role: 'user', content: 'Weather in Lisbon?' }, { role: 'assistant', content: body.content }, { role: 'user', content: [{ type: 'tool_result', tool_use_id: 'toolu_01', content: '19 C and sunny' }] }] });
++ assert.strictEqual(second.status, 200);
++ assert.deepStrictEqual(upstream.calls[upstream.calls.length - 1].body.messages[2].content[0].type, 'tool_result');
++ upstream.mode = 'text';
++});
++
++t.test('stream: Anthropic\'s events in order, and the run ends with the whole message', async () => {
++ const res = await post({ max_tokens: 64, stream: true, messages: [{ role: 'user', content: 'Stream please' }] });
++ assert.strictEqual(res.status, 200);
++ assert.match(res.headers.get('content-type'), /^text\/event-stream/);
++ const runId = res.headers.get('x-openvibe-run-id');
++ const events = await sse(res);
++ assert.deepStrictEqual(events.map((e) => e.event), FRAMES.map(([name]) => name));
++ assert.deepStrictEqual(events[0].data.message.id, 'msg_stream_1');
++ const text = events.filter((e) => e.data.type === 'content_block_delta').map((e) => e.data.delta.text).join('');
++ assert.strictEqual(text, 'Hello there');
++ assert.strictEqual(events[events.length - 1].data.type, 'message_stop');
++ const run = await request(h.base, 'GET', `/api/v1/runs/${runId}`, { tok: service });
++ assert.strictEqual(run.body.run.output.content[0].text, text, 'the run kept the whole answer the stream sent');
++ assert.strictEqual(run.body.run.output.stop_reason, 'end_turn');
++});
++
++t.test('x-openvibe-credential: the person\'s own key, not the router\'s provider', async () => {
++ let r = await request(h.base, 'PUT', `/api/v1/credentials/${PERSON}/anthropic`, { tok: manager, body: { key: PERSON_KEY, models: { chat: 'claude-test-1' } } });
++ assert.strictEqual(r.status, 200, r.text);
++ assert.ok(!r.text.includes(PERSON_KEY));
++ const before = upstream.calls.length;
++ // The person's subject comes from the token that acts for them (on_behalf_of).
++ const res = await post({ max_tokens: 64, messages: [{ role: 'user', content: 'Say hello' }] }, { tok: acting, headers: { 'x-openvibe-credential': 'anthropic' } });
++ assert.strictEqual(res.status, 200);
++ const body = await res.json();
++ assert.strictEqual(body.content[0].text, 'Hello there');
++ assert.strictEqual(upstream.calls.length, before, 'the router\'s provider was not called');
++ const call = ownCalls[ownCalls.length - 1];
++ assert.strictEqual(call.key, PERSON_KEY, 'the person\'s key was sent to their provider');
++ assert.strictEqual(call.body.model, 'claude-test-1');
++ assert.ok(!JSON.stringify(body).includes(PERSON_KEY), 'the key is not in the answer');
++ const run = await request(h.base, 'GET', `/api/v1/runs/${res.headers.get('x-openvibe-run-id')}`, { tok: acting });
++ assert.strictEqual(run.body.run.provenance.provider, `byo:codes:${PERSON}:anthropic`);
++ // <provider>:<subject> names the person explicitly, without an on_behalf_of claim.
++ const explicit = await post({ max_tokens: 64, messages: [{ role: 'user', content: 'hi' }] }, { tok: manager, headers: { 'x-openvibe-credential': `anthropic:${PERSON}` } });
++ assert.strictEqual(explicit.status, 200);
++ // A provider the credential does not speak is refused in Anthropic's shape.
++ const wrong = await post({ max_tokens: 64, messages: [{ role: 'user', content: 'hi' }] }, { tok: acting, headers: { 'x-openvibe-credential': 'openai' } });
++ assert.strictEqual(wrong.status, 404);
++ assert.strictEqual((await wrong.json()).type, 'error');
++ assert.ok(!logs.join('\n').includes(PERSON_KEY), 'the key is never in a log line');
++ await request(h.base, 'DELETE', `/api/v1/credentials/${PERSON}/anthropic`, { tok: manager });
++});
++
++t.test('auth: a token without ai.run.create is refused, in Anthropic\'s error shape', async () => {
++ let res = await fetch(`${h.base}/v1/messages`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ max_tokens: 8, messages: [{ role: 'user', content: 'hi' }] }) });
++ assert.strictEqual(res.status, 401);
++ let body = await res.json();
++ assert.deepStrictEqual([body.type, body.error.type], ['error', 'authentication_error']);
++ res = await post({ max_tokens: 8, messages: [{ role: 'user', content: 'hi' }] }, { tok: noCap });
++ assert.strictEqual(res.status, 403);
++ body = await res.json();
++ assert.deepStrictEqual([body.type, body.error.type], ['error', 'permission_error']);
++ res = await post({ messages: [{ role: 'user', content: 'hi' }] }, { tok: service });
++ assert.strictEqual(res.status, 400);
++ body = await res.json();
++ assert.deepStrictEqual([body.error.type, /max_tokens/.test(body.error.message)], ['invalid_request_error', true]);
++});
++
++t.test('stop', async () => { await upstream.close(); await h.stop(); });
++
++t.run();
ddiff --git a/test/workflows.test.js b/test/workflows.test.js
iindex 3faed4b..5ab29d0 100644
---- a/test/workflows.test.js
++++ b/test/workflows.test.js
@@@ -14,6 +14,7 @@ const SRC = (n) => Array.from({ length: n }, (_, i) => ({ source_type: 'news.art
const PASS = { role: 'chat', kind: 'chat_global', system: [{ text: 'You summarize chat.', cache: true }], user: 'Summarize: hi all', max_tokens: 80 };
const EXAMPLES = {
++ 'ai.messages': { max_tokens: 64, messages: [{ role: 'user', content: 'Say hello' }] },
'live.translate': { text: 'こんにちは、みなさん', from: 'ja', to: 'en', context: 'chat' },
'live.paste.describe_image': { title: 'my screenshot', image: { data_url: PNG } },
'live.paste.summarize_text': { title: 'notes', content: 'A list of things to buy: eggs, milk.' },