openvibers/openvibe.ai Commits

Accept Media file attachments on AI turns (#422)

OP OpenVibers committed 3 days ago · 69182fe161 · 15 files, +824 −22

* Accept Media file attachments on AI turns

* Run AI file input test suite

* Trigger AI attachment verification

* Retry AI attachment verification

* Read Media files on a turn and count them against the quota

* Sequence the quota files migration after agent effort

* Merge remote-tracking branch 'origin/main' into ai-files-cont

* Sequence the quota files migration after provider circuits

* Merge remote-tracking branch 'origin/main' into ai-files-cont

# Conflicts:
#	package-lock.json
#	package.json

Changes

ddiff --git a/migrations/0066_quota_files.sql b/migrations/0066_quota_files.sql
nnew file mode 100644
iindex 0000000..fcccd10
---- /dev/null
++++ b/migrations/0066_quota_files.sql
@@@ -0,0 +1,9 @@
++-- phase: expand
++-- Files on an AI turn (ai.file@1): per-person attachment limits live in the existing quota system.
++-- A quota may cap the number of files (max_files) and their total bytes (max_file_bytes) in its
++-- window; usage_counters gains the running totals the same way it counts requests, tokens and cost.
++-- Both are nullable, so every existing quota keeps meaning exactly what it meant before.
++ALTER TABLE quotas ADD COLUMN IF NOT EXISTS max_files bigint;
++ALTER TABLE quotas ADD COLUMN IF NOT EXISTS max_file_bytes bigint;
++ALTER TABLE usage_counters ADD COLUMN IF NOT EXISTS files bigint NOT NULL DEFAULT 0;
++ALTER TABLE usage_counters ADD COLUMN IF NOT EXISTS file_bytes bigint NOT NULL DEFAULT 0;
ddiff --git a/server/api/messages.js b/server/api/messages.js
iindex 87f3b27..aa17de1 100644
---- a/server/api/messages.js
++++ b/server/api/messages.js
@@@ -38,7 +38,7 @@ const { EXACT_ID } = modelCatalog;
const { resolveRequest } = require('../exchange/presets');
const { compact, receiptId } = require('../exchange/receipts');
--const FIELDS = ['system', 'tools', 'tool_choice', 'stop_sequences', 'temperature', 'top_p', 'top_k', 'metadata'];
++const FIELDS = ['system', 'tools', 'tool_choice', 'stop_sequences', 'temperature', 'top_p', 'top_k', 'metadata', 'files'];
const MAX_TOKENS = 64000;
// Anthropic's error types by HTTP status (docs.anthropic.com/en/api/errors).
ddiff --git a/server/api/openai.js b/server/api/openai.js
iindex fa08fda..2374443 100644
---- a/server/api/openai.js
++++ b/server/api/openai.js
@@@ -52,7 +52,7 @@ const { EXACT_ID } = modelCatalog;
const { resolveRequest } = require('../exchange/presets');
const { compact, receiptId } = require('../exchange/receipts');
--const FIELDS = ['messages', 'tools', 'tool_choice', 'parallel_tool_calls', 'response_format', 'stop', 'seed', 'top_p', 'presence_penalty', 'frequency_penalty', 'temperature'];
++const FIELDS = ['messages', 'tools', 'tool_choice', 'parallel_tool_calls', 'response_format', 'stop', 'seed', 'top_p', 'presence_penalty', 'frequency_penalty', 'temperature', 'files'];
// Cloudflare answers a 502 or 504 from the origin with its own HTML page, so an OpenAI client would get no JSON error:
// an upstream that failed or could not be reached is 503 (OpenAI's own "unavailable, retry"), the body kept. An invalid
ddiff --git a/server/artifacts.js b/server/artifacts.js
nnew file mode 100644
iindex 0000000..e782b1e
---- /dev/null
++++ b/server/artifacts.js
@@@ -0,0 +1,60 @@
++'use strict';
++/**
++ * Files a model produced on a turn, returned as stored files (ai.file@1, role 'artifact').
++ *
++ * A provider can answer with files beside its text: each entry names a Media object it already stored
++ * (the reference shape), which is returned as it is. A provider that hands back the bytes themselves
++ * (`data`/`b64`) needs a Media store in this service: `store.put({ name, media_type, bytes, subject })`
++ * returns the stored ai.file@1. When no store is configured such a file is REFUSED with
++ * files.store_unavailable, never dropped and never put in the run output as inline bytes. The store is
++ * configured by the operator (a Media tenant AI may write to); nothing here pretends to have one.
++ */
++const { AiError } = require('./util');
++
++const MAX_ARTIFACTS = 10;
++const MEDIA_ID = /^med_[0-9A-HJKMNP-TV-Z]{26}$/;
++
++function storedRef(file) {
++ const id = String((file && file.id) || (file && file.storage && file.storage.key) || '');
++ if (!MEDIA_ID.test(id)) return null;
++ const name = String((file && file.name) || '').replace(/[\u0000-\u001f\u007f]/g, ' ').trim().slice(0, 255) || 'file';
++ const mediaType = String((file && file.media_type) || 'application/octet-stream').split(';')[0].trim().slice(0, 255) || 'application/octet-stream';
++ const size = Math.max(0, Math.floor(Number(file && file.size) || 0));
++ return {
++ id, name, media_type: mediaType, size,
++ storage: { provider: 'media', key: id },
++ access: { readers: Array.isArray(file && file.access && file.access.readers) && file.access.readers.length ? file.access.readers.map(String).slice(0, 50) : [] },
++ read: { method: 'not_readable', reason: 'Stored by Media; read it by reference.' },
++ role: 'artifact',
++ };
++}
++
++/** The store a run may write to (config.media.artifactStore with put(), or null). */
++function storeOf(config) {
++ const s = config && config.media && config.media.artifactStore;
++ return s && typeof s.put === 'function' ? s : null;
++}
++
++/**
++ * The files a provider's message carries, as stored ai.file@1 artifacts.
++ * message.artifacts (or message.files) is the list; each is either a Media reference or inline bytes.
++ */
++async function collectArtifacts(message, { store = null, subject = null } = {}) {
++ const raw = message && (Array.isArray(message.artifacts) ? message.artifacts : Array.isArray(message.files) ? message.files : []);
++ if (!raw.length) return [];
++ if (raw.length > MAX_ARTIFACTS) throw new AiError(413, 'files.too_many', `a model may return at most ${MAX_ARTIFACTS} files`);
++ const out = [];
++ for (const file of raw) {
++ const ref = storedRef(file);
++ if (ref) { out.push(ref); continue; }
++ const bytes = Buffer.isBuffer(file && file.bytes) ? file.bytes
++ : typeof (file && (file.data || file.b64)) === 'string' ? Buffer.from(file.data || file.b64, file.encoding === 'utf8' ? 'utf8' : 'base64')
++ : null;
++ if (!bytes) throw new AiError(422, 'files.invalid', `a model returned a file without bytes or a Media reference: ${String((file && file.name) || 'unnamed').slice(0, 80)}`);
++ if (!store) throw new AiError(503, 'files.store_unavailable', 'a model produced a file but AI has no Media store configured to keep it');
++ out.push(await store.put({ name: file.name, media_type: file.media_type || file.type, bytes, subject }));
++ }
++ return out;
++}
++
++module.exports = { collectArtifacts, storeOf, MAX_ARTIFACTS };
ddiff --git a/server/fetcher.js b/server/fetcher.js
iindex 9cccca8..495061f 100644
---- a/server/fetcher.js
++++ b/server/fetcher.js
@@@ -134,7 +134,9 @@ function createFetcher(config, { transport = null } = {}) {
}
if (res.statusCode !== 200) {
res.resume();
-- return done(() => reject(new AiError(502, 'source.unavailable', `source answered HTTP ${res.statusCode}`)));
++ const err = new AiError(502, 'source.unavailable', `source answered HTTP ${res.statusCode}`);
++ err.upstreamStatus = res.statusCode; // a caller can tell "gone" from "down" (file-inputs does)
++ return done(() => reject(err));
}
const declared = Number(res.headers['content-length']);
if (Number.isFinite(declared) && declared > maxBytes) { res.resume(); return done(() => reject(new AiError(413, 'source.too_large', `source is larger than ${maxBytes} bytes`))); }
ddiff --git a/server/file-inputs.js b/server/file-inputs.js
nnew file mode 100644
iindex 0000000..d59edfb
---- /dev/null
++++ b/server/file-inputs.js
@@@ -0,0 +1,352 @@
++'use strict';
++/**
++ * Files on a turn: Media-backed attachments read for the model, never stored in the run.
++ *
++ * A conversational request names its files by Media reference (`ai.file@1`: an `id`/`storage.key` that
++ * is a `med_…` object id). The bytes are fetched from Media ONLY for this provider call and are never
++ * written to the database: the run keeps the references (and a passthrough workflow does not keep its
++ * prompt at all), the provider gets text or image content, and the buffers die with the request.
++ *
++ * What a file becomes depends on its real type (the Media content type, else the name's extension):
++ * text, markdown, code, CSV and JSON a text block, cut at MAX_TEXT_CHARS with an honest "(truncated)" line
++ * PDF its text, page by page, through the small extractor below (no dependency)
++ * PNG, JPEG, WebP, GIF an image part, when the model can see (vision); otherwise a named note
++ * anything else a named note saying it could not be read, never silently dropped
++ *
++ * Limits: MAX_FILES per turn, MAX_FILE_BYTES each and MAX_TOTAL_BYTES together. Per-person, per-window
++ * limits come from the existing quota system (server/quota.js `chargeFiles`), which counts the files a
++ * person attaches before the provider is called.
++ *
++ * The virus scanner is a named extension point: `setScanner(fn)` (or the `scan` option) is awaited on
++ * every file and may throw to refuse the turn. There is NO built-in scan and none is pretended: leaving
++ * it null means no scan ran, and the turn says nothing to the model about scanning.
++ */
++const zlib = require('node:zlib');
++const { AiError } = require('./util');
++
++const MAX_FILES = 10;
++const MAX_FILE_BYTES = 10 * 1024 * 1024;
++const MAX_TOTAL_BYTES = 25 * 1024 * 1024;
++const MAX_TEXT_CHARS = 200000;
++const MAX_PDF_STREAMS = 500;
++const MEDIA_ID = /^med_[0-9A-HJKMNP-TV-Z]{26}$/;
++const TEXT_EXT = /\.(txt|text|md|markdown|csv|tsv|json|jsonl|log|xml|yaml|yml|toml|ini|cfg|conf|html?|css|scss|less|m?js|cjs|jsx|ts|tsx|py|rb|go|rs|java|kt|kts|c|h|cc|cpp|hpp|cs|php|sh|bash|zsh|sql|graphql|gql|proto|swift|lua|pl|r|m|vue|svelte|astro|tex|env)$/i;
++const CODE_TYPE = /^(text\/[a-z0-9.+-]+|application\/(json|jsonl|xml|javascript|ecmascript|x-javascript|x-httpd-php|x-sh|x-shellscript|x-python|x-ruby|x-yaml|x-yml|x-toml|x-tex|x-latex|x-c|x-c\+\+|x-java-source|x-php|x-perl|x-rust|x-go|x-typescript|x-www-form-urlencoded))$/i;
++const IMAGE_TYPE = /^image\/(png|jpe?g|webp|gif)$/i;
++const PDF_TYPE = /^application\/pdf$/i;
++
++/** The named virus-scan extension point; null means no scan is configured (and none is pretended). */
++let scanner = null;
++
++/** Install the virus scan (server start-up or a test). fn({ file, name, media_type, bytes }) → may throw. */
++function setScanner(fn) {
++ if (fn != null && typeof fn !== 'function') throw new TypeError('setScanner: pass a function or null');
++ scanner = fn;
++}
++
++const typeOf = (ct) => String(ct || '').split(';')[0].trim().toLowerCase();
++
++function extOf(name) {
++ const m = /\.([a-z0-9]+)$/i.exec(String(name || ''));
++ return m ? m[1].toLowerCase() : '';
++}
++
++/** The type a file is treated as: the bytes' own content type when Media knows it, else the name. */
++function effectiveType(name, contentType, declared) {
++ let type = typeOf(contentType);
++ if (!type || type === 'application/octet-stream' || type === 'binary/octet-stream') type = typeOf(declared);
++ if ((!type || type === 'application/octet-stream' || type === 'binary/octet-stream') && TEXT_EXT.test(String(name || ''))) type = 'text/plain';
++ if ((!type || type === 'application/octet-stream') && extOf(name) === 'pdf') type = 'application/pdf';
++ if ((!type || type === 'application/octet-stream') && /^(png|jpe?g|webp|gif)$/.test(extOf(name))) type = `image/${extOf(name).replace('jpg', 'jpeg')}`;
++ return type || 'application/octet-stream';
++}
++
++/** The Media object id a file reference names: `storage.key` (the contract) or `id`. */
++function mediaIdOf(file) {
++ const id = String((file && ((file.storage && file.storage.key) || file.id)) || '');
++ if (!MEDIA_ID.test(id)) throw new AiError(422, 'files.invalid', `file ${String((file && file.name) || '(unnamed)').slice(0, 80)} must reference a stored Media object (med_…)`);
++ return id;
++}
++
++function nameOf(file) {
++ const name = String((file && file.name) || '').replace(/[\u0000-\u001f\u007f]/g, ' ').trim().slice(0, 255);
++ return name || 'unnamed file';
++}
++
++/** The text of a PDF, from its content streams, without a dependency. */
++function pdfText(buffer) {
++ const raw = buffer.toString('latin1');
++ if (/\/Encrypt\s/.test(raw)) return { pages: [], reason: 'the PDF is encrypted' };
++ const pages = [];
++ let total = 0;
++ let seen = 0;
++ const re = /stream\r?\n?([\s\S]*?)\r?\n?endstream/g;
++ let m;
++ while ((m = re.exec(raw)) && seen < MAX_PDF_STREAMS) {
++ seen++;
++ const before = raw.slice(Math.max(0, m.index - 400), m.index);
++ let data = Buffer.from(m[1], 'latin1');
++ if (/\/FlateDecode\b/.test(before)) {
++ try { data = zlib.inflateSync(data, { maxOutputLength: 8 * 1024 * 1024 }); } catch { continue; }
++ } else if (/\/Filter\b/.test(before) && !/\/FlateDecode\b/.test(before)) {
++ continue; // an image (DCT) or another filter, not text
++ }
++ const text = contentText(data.toString('latin1'));
++ if (text) { pages.push(text); total += text.length; }
++ if (total >= MAX_TEXT_CHARS) break;
++ }
++ if (!pages.length) return { pages: [], reason: 'the PDF has no readable text (it may be scanned images, or use a custom encoding)' };
++ return { pages };
++}
++
++/** The text Tj/TJ/`'` operators show in one PDF content stream. */
++function contentText(stream) {
++ const parts = [];
++ let i = 0;
++ const n = stream.length;
++ const readString = () => {
++ let out = '';
++ let depth = 1;
++ i++;
++ while (i < n && depth > 0) {
++ const c = stream[i++];
++ if (c === '\\') {
++ const d = stream[i++];
++ if (d === 'n') out += '\n';
++ else if (d === 'r') out += '\r';
++ else if (d === 't') out += '\t';
++ else if (d === 'b') out += '\b';
++ else if (d === 'f') out += '\f';
++ else if (d === '\n' || d === '\r') { if (d === '\r' && stream[i] === '\n') i++; }
++ else if (/[0-7]/.test(d)) {
++ let oct = d;
++ while (oct.length < 3 && /[0-7]/.test(stream[i])) oct += stream[i++];
++ out += String.fromCharCode(parseInt(oct, 8));
++ } else out += d;
++ } else if (c === '(') { depth++; out += c; }
++ else if (c === ')') { depth--; if (depth) out += c; }
++ else out += c;
++ }
++ return out;
++ };
++ while (i < n) {
++ const c = stream[i];
++ if (c === '(') {
++ const s = readString();
++ let j = i;
++ while (j < n && /\s/.test(stream[j])) j++;
++ if (stream.startsWith('Tj', j)) { parts.push(s); i = j + 2; continue; }
++ if (stream.startsWith("'", j)) { parts.push(`\n${s}`); i = j + 1; continue; }
++ if (stream.startsWith('"', j)) { parts.push(`\n${s}`); i = j + 1; continue; }
++ continue;
++ }
++ if (c === '[') {
++ i++;
++ const arr = [];
++ while (i < n && stream[i] !== ']') {
++ if (stream[i] === '(') arr.push(readString());
++ else i++;
++ }
++ i++;
++ const end = stream.slice(i, i + 8);
++ const tm = /^\s*TJ/.exec(end);
++ if (tm) { parts.push(arr.join('')); i += tm[0].length; }
++ continue;
++ }
++ if (stream.startsWith('T*', i) || stream.startsWith('Td', i) || stream.startsWith('TD', i) || stream.startsWith('ET', i)) {
++ parts.push('\n'); i += 2; continue;
++ }
++ i++;
++ }
++ return parts.join('')
++ .replace(/[ \t]+\n/g, '\n')
++ .replace(/\n{3,}/g, '\n\n')
++ .trim();
++}
++
++/** The ai.file@1 a turn keeps for its reference: metadata only, never the bytes. */
++function referenceOf(file, mediaId, name, mediaType, size, read) {
++ return {
++ id: mediaId,
++ name,
++ media_type: mediaType,
++ size,
++ storage: { provider: 'media', key: mediaId },
++ access: {
++ readers: Array.isArray(file && file.access && file.access.readers) && file.access.readers.length
++ ? file.access.readers.map(String).slice(0, 50)
++ : ['ai'],
++ },
++ read,
++ role: 'input',
++ };
++}
++
++/** Read one Media-backed file for the model. → { reference, part, note } */
++async function readFile(file, fetcher, { signal, format = 'openai', vision = true, scan = null } = {}) {
++ const mediaId = mediaIdOf(file);
++ const name = nameOf(file);
++ const declaredSize = Number(file && file.size);
++ // A caller that already read the file (the SDK's ov.ai.files.read) may send the ai.file@1 read result;
++ // that is used without a Media fetch. Only an actual reading is trusted: a `not_readable` marker is not.
++ const supplied = file && file.read && typeof file.read === 'object' ? file.read : null;
++ if (supplied && supplied.method === 'text' && typeof supplied.content === 'string') {
++ const type = effectiveType(name, null, file && file.media_type);
++ const size = Number.isFinite(declaredSize) && declaredSize >= 0 ? declaredSize : supplied.content.length;
++ const hook = scan || scanner;
++ if (hook) await hook({ file, media_id: mediaId, name, media_type: type, bytes: Buffer.from(supplied.content) });
++ const text = capText(supplied.content);
++ return { reference: referenceOf(file, mediaId, name, type, size, { method: 'text' }), part: textPart(format, `\n\n[Attached file: ${name} (${type})]\n${text}\n`), note: null };
++ }
++ if (supplied && supplied.method === 'pages' && Array.isArray(supplied.pages) && supplied.pages.length) {
++ const type = effectiveType(name, null, file && file.media_type);
++ const size = Number.isFinite(declaredSize) && declaredSize >= 0 ? declaredSize : supplied.pages.join('').length;
++ const text = capText(supplied.pages.map((p, i) => `--- page ${i + 1} ---\n${String(p || '')}`).join('\n\n'));
++ return { reference: referenceOf(file, mediaId, name, type, size, { method: 'pages' }), part: textPart(format, `\n\n[Attached PDF: ${name} (${type})]\n${text}\n`), note: null };
++ }
++ if (supplied && supplied.method === 'image' && typeof supplied.content === 'string' && /^data:image\//i.test(supplied.content)) {
++ const type = effectiveType(name, null, file && file.media_type);
++ const size = Number.isFinite(declaredSize) && declaredSize >= 0 ? declaredSize : 0;
++ if (vision) {
++ const mm = /^data:(image\/[a-z0-9.+-]+);base64,(.+)$/i.exec(supplied.content);
++ if (mm) {
++ const part = format === 'anthropic'
++ ? { type: 'image', source: { type: 'base64', media_type: mm[1], data: mm[2] } }
++ : { type: 'image_url', image_url: { url: supplied.content } };
++ return { reference: referenceOf(file, mediaId, name, type, size, { method: 'image' }), part, note: null };
++ }
++ }
++ }
++ let fetched;
++ try {
++ fetched = await fetcher.fetchUrl(fetcher.mediaRefUrl({ media_id: mediaId }), { signal, maxBytes: MAX_FILE_BYTES, timeoutMs: 30000 });
++ } catch (err) {
++ if (err && (err.status === 404 || err.upstreamStatus === 404)) throw new AiError(422, 'files.unreadable', `file ${name} is not a readable Media object`);
++ throw err;
++ }
++ const { buffer } = fetched;
++ const type = effectiveType(name, fetched.contentType, file && file.media_type);
++ const size = Number.isFinite(declaredSize) && declaredSize >= 0 ? Math.max(declaredSize, buffer.length) : buffer.length;
++ const hook = scan || scanner;
++ if (hook) await hook({ file, media_id: mediaId, name, media_type: type, bytes: buffer });
++ if (PDF_TYPE.test(type) || extOf(name) === 'pdf') {
++ const pdf = pdfText(buffer);
++ if (pdf.pages.length) {
++ const text = capText(pdf.pages.map((p, i) => `--- page ${i + 1} ---\n${p}`).join('\n\n'));
++ return { reference: referenceOf(file, mediaId, name, type, size, { method: 'pages' }), part: textPart(format, `\n\n[Attached PDF: ${name} (${type})]\n${text}\n`), note: null };
++ }
++ const reason = `could not read this PDF: ${pdf.reason}`;
++ return { reference: referenceOf(file, mediaId, name, type, size, { method: 'not_readable', reason }), part: null, note: `[Attached file: ${name} (${type}); ${reason}.]` };
++ }
++ if (IMAGE_TYPE.test(type)) {
++ const data = buffer.toString('base64');
++ if (vision) {
++ const part = format === 'anthropic'
++ ? { type: 'image', source: { type: 'base64', media_type: type, data } }
++ : { type: 'image_url', image_url: { url: `data:${type};base64,${data}` } };
++ return { reference: referenceOf(file, mediaId, name, type, size, { method: 'image' }), part, note: null };
++ }
++ const reason = 'this model cannot see images, so the picture itself was not passed';
++ return { reference: referenceOf(file, mediaId, name, type, size, { method: 'not_readable', reason }), part: null, note: `[Attached image: ${name} (${type}); ${reason}.]` };
++ }
++ if (CODE_TYPE.test(type) || TEXT_EXT.test(name) || /^text\//.test(type)) {
++ const text = capText(buffer.toString('utf8').replace(/\u0000/g, ''));
++ if (!text.trim()) {
++ const reason = 'the file is empty';
++ return { reference: referenceOf(file, mediaId, name, type, size, { method: 'not_readable', reason }), part: null, note: `[Attached file: ${name} (${type}); ${reason}.]` };
++ }
++ return { reference: referenceOf(file, mediaId, name, type, size, { method: 'text' }), part: textPart(format, `\n\n[Attached file: ${name} (${type})]\n${text}\n`), note: null };
++ }
++ const reason = 'this file type cannot be read as text or seen as an image';
++ return {
++ reference: referenceOf(file, mediaId, name, type, size, { method: 'not_readable', reason }),
++ part: null,
++ note: `[Attached file: ${name} (${type}); ${reason}. Use its name and type when you answer, and say what you could not read.]`,
++ };
++}
++
++function textPart(format, text) {
++ return { type: 'text', text };
++}
++
++function capText(text) {
++ const s = String(text || '');
++ if (s.length <= MAX_TEXT_CHARS) return s;
++ return `${s.slice(0, MAX_TEXT_CHARS)}\n\n(truncated: ${s.length - MAX_TEXT_CHARS} more characters were not passed)`;
++}
++
++function capPages(pages) {
++ let left = MAX_TEXT_CHARS;
++ const out = [];
++ for (const p of pages) {
++ if (left <= 0) break;
++ const s = String(p || '');
++ out.push(s.length <= left ? s : `${s.slice(0, left)}\n(truncated)`);
++ left -= Math.min(s.length, left);
++ }
++ return out;
++}
++
++/** The declared size of a file reference, for the count/byte limit before any fetch. */
++function declaredSize(file) {
++ const n = Number(file && file.size);
++ return Number.isFinite(n) && n >= 0 ? n : 0;
++}
++
++function checkLimits(files) {
++ if (files.length > MAX_FILES) throw new AiError(413, 'files.too_many', `at most ${MAX_FILES} files may be attached to a turn`);
++ let total = 0;
++ for (const f of files) {
++ const size = declaredSize(f);
++ if (size > MAX_FILE_BYTES) throw new AiError(413, 'files.too_large', `each attached file may be up to ${MAX_FILE_BYTES / 1024 / 1024} MiB`);
++ total += size;
++ }
++ if (total > MAX_TOTAL_BYTES) throw new AiError(413, 'files.too_large', `attached files may be up to ${MAX_TOTAL_BYTES / 1024 / 1024} MiB together`);
++ return total;
++}
++
++/** The last user message of a protocol message list (or a new one). */
++function lastUserIndex(messages) {
++ for (let i = messages.length - 1; i >= 0; i--) if (messages[i] && messages[i].role === 'user') return i;
++ return -1;
++}
++
++function userContent(format, content) {
++ if (content == null || content === '') return [];
++ if (typeof content === 'string') return [{ type: 'text', text: content }];
++ return Array.isArray(content) ? content.map(x => (format === 'anthropic' && x && x.type === 'text' ? { type: 'text', text: String(x.text || '') } : x)) : [{ type: 'text', text: String(content) }];
++}
++
++/**
++ * Resolve a turn's `files` (Media references) and add what the model can read to its last user message.
++ * Returns a NEW input: the messages gain parts, the file references stay metadata-only (`read` says what
++ * happened to each, which is the ai.file@1 shape). No bytes or extracted text are put into `input.files`.
++ */
++async function attachFiles(input, fetcher, { signal, format = 'openai', vision = true, scan = null } = {}) {
++ const files = Array.isArray(input && input.files) ? input.files : [];
++ if (!files.length) return input;
++ checkLimits(files);
++ const references = [];
++ const additions = [];
++ let total = 0;
++ let hasImage = false;
++ for (const file of files) {
++ const read = await readFile(file, fetcher, { signal, format, vision, scan });
++ total += read.reference.size;
++ if (total > MAX_TOTAL_BYTES) throw new AiError(413, 'files.too_large', `attached files may be up to ${MAX_TOTAL_BYTES / 1024 / 1024} MiB together`);
++ references.push(read.reference);
++ if (read.part && read.part.type === 'image') hasImage = true;
++ if (read.part && read.part.type === 'image_url') hasImage = true;
++ if (read.part) additions.push(read.part);
++ if (read.note) additions.push({ type: 'text', text: read.note });
++ }
++ const messages = Array.isArray(input.messages) ? input.messages.map(x => ({ ...x })) : [];
++ let index = lastUserIndex(messages);
++ if (index < 0) { messages.push({ role: 'user', content: [] }); index = messages.length - 1; }
++ messages[index] = { ...messages[index], content: [...userContent(format, messages[index].content), ...additions] };
++ return { ...input, messages, files: references, hasImage };
++}
++
++module.exports = {
++ attachFiles, readFile, pdfText, effectiveType, mediaIdOf, checkLimits, setScanner, MAX_FILES, MAX_FILE_BYTES, MAX_TOTAL_BYTES, MAX_TEXT_CHARS,
++};
ddiff --git a/server/providers/task.js b/server/providers/task.js
iindex 47c9af5..a361832 100644
---- a/server/providers/task.js
++++ b/server/providers/task.js
@@@ -28,7 +28,7 @@ const DEFAULT_TIER = 'balanced';
/** The kind of work a run does: the person's ai.me-preferences.defaults has one default per kind. */
function workKind(operation, req = {}, ctx = {}) {
if (operation === 'image') return 'image';
-- if (req && req.image) return 'vision';
++ if (req && (req.image || req.vision)) return 'vision';
const hint = String(ctx.workflowKey || '') + ' ' + String(ctx.routeKey || '');
return /\b(code|coding|codes|build|design|program|developer)\b/i.test(hint) ? 'code' : 'text';
}
@@@ -99,7 +99,7 @@ function weightOf({ operation, kind, reasoning, tools, min_context, maxTokens })
*/
function detectTask({ operation = 'chat', req = {}, ctx = {}, route = null } = {}) {
const kind = workKind(operation, req, ctx);
-- const vision = Boolean(req.image) || operation === 'vision';
++ const vision = Boolean(req.image) || Boolean(req.vision) || operation === 'vision';
const tools = toolCount(req) > 0;
const json = wantsJson(req, route);
const min_context = Math.max(1024, tokensOf(leadingText(req)) + (vision ? 1024 : 0) + 512);
ddiff --git a/server/quota.js b/server/quota.js
iindex e77d9fc..a5f794b 100644
---- a/server/quota.js
++++ b/server/quota.js
@@@ -107,6 +107,10 @@ function createQuotas(db, { clock = { now: () => Date.now() }, registry, freeAll
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(scope_type, scope_id, "window", window_start, workflow_prefix)
DO UPDATE SET requests = usage_counters.requests + excluded.requests, tokens = usage_counters.tokens + excluded.tokens, cost_usd = usage_counters.cost_usd + excluded.cost_usd`);
++ const bumpFiles = db.prepare(`INSERT INTO usage_counters (scope_type, scope_id, "window", window_start, workflow_prefix, files, file_bytes)
++ VALUES (?, ?, ?, ?, ?, ?, ?)
++ ON CONFLICT(scope_type, scope_id, "window", window_start, workflow_prefix)
++ DO UPDATE SET files = usage_counters.files + excluded.files, file_bytes = usage_counters.file_bytes + excluded.file_bytes`);
const reserveTx = async (ctx) => await db.tx(async () => {
const qs = await applicable(ctx);
@@@ -151,6 +155,49 @@ function createQuotas(db, { clock = { now: () => Date.now() }, registry, freeAll
return r.quotas;
}
++ /**
++ * Check + count the files a turn attaches, before any provider is called. A quota may cap the number
++ * of files and/or their total bytes in its window (max_files, max_file_bytes); the counts go into the
++ * same per-scope counters requests/tokens/cost use, so a per-person (actor) quota limits that person's
++ * attachments and a global one limits everyone. A quota that leaves both fields null is unaffected.
++ */
++ async function chargeFiles(ctx, files) {
++ const list = Array.isArray(files) ? files : [];
++ if (!list.length) return null;
++ const count = list.length;
++ const bytes = list.reduce((n, f) => n + Math.max(0, Number(f && f.size) || 0), 0);
++ return await db.tx(async () => {
++ const qs = await applicable(ctx);
++ const t = nowSec();
++ let worst = null;
++ for (const { q, cid } of qs) {
++ const ws = windowStart(q.window, t);
++ const c = await getCounter.get(q.scope_type, cid, q.window, ws, q.workflow_prefix || '') || { requests: 0, tokens: 0, cost_usd: 0, files: 0, file_bytes: 0 };
++ let reason = null;
++ if (q.max_files != null && (Number(c.files) || 0) + count > q.max_files) reason = `${q.max_files} files per ${q.window}`;
++ else if (q.max_file_bytes != null && (Number(c.file_bytes) || 0) + bytes > q.max_file_bytes) reason = `${q.max_file_bytes} file bytes per ${q.window}`;
++ if (reason) {
++ const retryAfter = ws + WINDOWS[q.window] - t;
++ if (!worst || retryAfter > worst.retryAfter) worst = { q, cid, reason, retryAfter };
++ }
++ }
++ if (worst) throw new AiError(429, 'quota.exceeded', `quota exceeded for ${worst.q.scope_type} ${worst.cid}: ${worst.reason}`, {
++ retry_after_seconds: Math.max(1, worst.retryAfter), quota: { id: worst.q.id, scope_type: worst.q.scope_type, scope_id: worst.cid, window: worst.q.window },
++ });
++ // The file counts follow the same scope x window rule as reserve(): every scope gets the
++ // running total, and a workflow-prefix quota gets its own counter.
++ const keys = new Map();
++ for (const [st, sid] of scopesFor(ctx)) {
++ for (const w of Object.keys(WINDOWS)) keys.set(`${st}|${sid}|${w}|`, { scope_type: st, scope_id: sid, window: w, workflow_prefix: '', ws: windowStart(w, t) });
++ }
++ for (const { q, cid } of qs) {
++ if (q.workflow_prefix) keys.set(`${q.scope_type}|${cid}|${q.window}|${q.workflow_prefix}`, { scope_type: q.scope_type, scope_id: cid, window: q.window, workflow_prefix: q.workflow_prefix, ws: windowStart(q.window, t) });
++ }
++ for (const k of keys.values()) await bumpFiles.run(k.scope_type, k.scope_id, k.window, k.ws, k.workflow_prefix, count, bytes);
++ return { files: count, file_bytes: bytes };
++ });
++ }
++
const insDaily = db.prepare(`INSERT INTO usage_daily (day, requester, attribution, workflow_key, provider_key, model_key, requests, tokens_in, tokens_out, tokens_cached, cost_usd)
VALUES (?, ?, ?, ?, ?, ?, 1, ?, ?, ?, ?)
ON CONFLICT(day, requester, attribution, workflow_key, provider_key, model_key) DO UPDATE SET requests = usage_daily.requests + 1,
@@@ -265,21 +312,21 @@ function createQuotas(db, { clock = { now: () => Date.now() }, registry, freeAll
const scopeId = q.scope_type === 'global' ? '*' : String(q.scope_id || ''); // '*' on other scopes = each one separately
if (!scopeId || scopeId.length > 200) throw new AiError(422, 'ai.invalid', 'scope_id required');
const num = (v) => (v == null || v === '' ? null : Number(v));
-- for (const k of ['max_requests', 'max_tokens', 'max_cost_usd']) if (q[k] != null && !(Number(q[k]) >= 0)) throw new AiError(422, 'ai.invalid', `${k} must be >= 0`);
++ for (const k of ['max_requests', 'max_tokens', 'max_cost_usd', 'max_files', 'max_file_bytes']) if (q[k] != null && !(Number(q[k]) >= 0)) throw new AiError(422, 'ai.invalid', `${k} must be >= 0`);
const status = q.status || 'active';
if (!['active', 'disabled'].includes(status)) throw new AiError(422, 'ai.invalid', 'status must be active or disabled');
const t = iso(clock.now());
const prefix = q.workflow_prefix || null;
const prev = await db.prepare("SELECT * FROM quotas WHERE scope_type = ? AND scope_id = ? AND \"window\" = ? AND COALESCE(workflow_prefix, '') = ?").get(q.scope_type, scopeId, q.window, prefix || '');
if (prev) {
-- await db.prepare('UPDATE quotas SET max_requests = ?, max_tokens = ?, max_cost_usd = ?, status = ?, updated_at = ? WHERE id = ?')
-- .run(num(q.max_requests), num(q.max_tokens), num(q.max_cost_usd), status, t, prev.id);
++ await db.prepare('UPDATE quotas SET max_requests = ?, max_tokens = ?, max_cost_usd = ?, max_files = ?, max_file_bytes = ?, status = ?, updated_at = ? WHERE id = ?')
++ .run(num(q.max_requests), num(q.max_tokens), num(q.max_cost_usd), num(q.max_files), num(q.max_file_bytes), status, t, prev.id);
} else {
-- await db.prepare('INSERT INTO quotas (scope_type, scope_id, "window", max_requests, max_tokens, max_cost_usd, workflow_prefix, status, origin, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)')
-- .run(q.scope_type, scopeId, q.window, num(q.max_requests), num(q.max_tokens), num(q.max_cost_usd), prefix, status, origin, t, t);
++ await db.prepare('INSERT INTO quotas (scope_type, scope_id, "window", max_requests, max_tokens, max_cost_usd, max_files, max_file_bytes, workflow_prefix, status, origin, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)')
++ .run(q.scope_type, scopeId, q.window, num(q.max_requests), num(q.max_tokens), num(q.max_cost_usd), num(q.max_files), num(q.max_file_bytes), prefix, status, origin, t, t);
}
const row = await db.prepare("SELECT * FROM quotas WHERE scope_type = ? AND scope_id = ? AND \"window\" = ? AND COALESCE(workflow_prefix, '') = ?").get(q.scope_type, scopeId, q.window, prefix || '');
-- if (registry) await registry.audit(actor, prev ? 'quota.update' : 'quota.create', 'quota', row.id, { trace, metadata: { scope_type: row.scope_type, scope_id: row.scope_id, window: row.window, max_requests: row.max_requests, max_tokens: row.max_tokens, max_cost_usd: row.max_cost_usd, status: row.status, origin } });
++ if (registry) await registry.audit(actor, prev ? 'quota.update' : 'quota.create', 'quota', row.id, { trace, metadata: { scope_type: row.scope_type, scope_id: row.scope_id, window: row.window, max_requests: row.max_requests, max_tokens: row.max_tokens, max_cost_usd: row.max_cost_usd, max_files: row.max_files, max_file_bytes: row.max_file_bytes, status: row.status, origin } });
return row;
}
@@@ -362,8 +409,8 @@ function createQuotas(db, { clock = { now: () => Date.now() }, registry, freeAll
const t = nowSec();
return (await Promise.all(qs.map(async ({ q, cid }) => {
const ws = windowStart(q.window, t);
-- const c = await getCounter.get(q.scope_type, cid, q.window, ws, q.workflow_prefix || '') || { requests: 0, tokens: 0, cost_usd: 0 };
-- return { quota: q, scope_id: cid, window_start: iso(ws * 1000), resets_in_seconds: ws + WINDOWS[q.window] - t, used: { requests: c.requests, tokens: c.tokens, cost_usd: c.cost_usd } };
++ const c = await getCounter.get(q.scope_type, cid, q.window, ws, q.workflow_prefix || '') || { requests: 0, tokens: 0, cost_usd: 0, files: 0, file_bytes: 0 };
++ return { quota: q, scope_id: cid, window_start: iso(ws * 1000), resets_in_seconds: ws + WINDOWS[q.window] - t, used: { requests: c.requests, tokens: c.tokens, cost_usd: c.cost_usd, files: Number(c.files) || 0, file_bytes: Number(c.file_bytes) || 0 } };
})));
}
@@@ -381,7 +428,7 @@ function createQuotas(db, { clock = { now: () => Date.now() }, registry, freeAll
return { max_cost_usd: Math.min(...rows.map((r) => r.max_cost_usd)), spent, window: 'day' };
}
-- return { reserve, account, statsFor, list, upsert, usage, usageSeries, counters, paidBudget, freeAllowance, WINDOWS };
++ return { reserve, chargeFiles, account, statsFor, list, upsert, usage, usageSeries, counters, paidBudget, freeAllowance, WINDOWS };
}
module.exports = { createQuotas, WINDOWS };
ddiff --git a/server/runs.js b/server/runs.js
iindex e6aaa2f..d48646c 100644
---- a/server/runs.js
++++ b/server/runs.js
@@@ -565,6 +565,16 @@ function createRuns({ choices = null, db, registry, engine, cache, quotas, confi
// BA2/BA3: a batch run reserves its quota when the answer is here, not while it waits.
reserved = credentialSubject !== null || selfKeys || (principal && principal.deferBatch) || autoBatch ? null : await quotas.reserve(ctx);
} catch (err) { if (hold) await tiers.release(hold); throw err; }
++ // Files on the turn (ai.file@1): the per-turn caps first, then the per-person quota's file limits
++ // (server/quota.js chargeFiles). Both are checked before the run is stored or a provider is called;
++ // the references only are counted here, never the bytes.
++ const filesOnTurn = Array.isArray(input && input.files) ? input.files : [];
++ if (filesOnTurn.length) {
++ try {
++ require('./file-inputs').checkLimits(filesOnTurn);
++ await quotas.chargeFiles(ctx, filesOnTurn);
++ } catch (err) { if (hold) await tiers.release(hold); throw err; }
++ }
try {
await db.tx(async () => {
ddiff --git a/server/workflows/core.js b/server/workflows/core.js
iindex 55e2fbe..12168b7 100644
---- a/server/workflows/core.js
++++ b/server/workflows/core.js
@@@ -8,6 +8,15 @@ const { PASSTHROUGH_OUTPUT } = require('./live');
const STR = (n) => ({ type: 'string', maxLength: n });
const OBJ = (props, required = []) => ({ type: 'object', additionalProperties: false, required, properties: props });
++// A file on a turn (ai.file@1): a Media reference, never inline bytes. The full contract shape is
++// accepted (the SDK returns it); a caller that has only the object id may send just { id } / { media_id }.
++const FILE_REF = { type: 'object', additionalProperties: false, properties: {
++ id: STR(200), media_id: STR(80), name: STR(255), media_type: STR(255), size: { type: 'integer', minimum: 0 },
++ storage: { type: 'object', additionalProperties: false, required: ['provider', 'key'], properties: { provider: { const: 'media' }, key: STR(500) } },
++ access: { type: 'object', additionalProperties: false, properties: { readers: { type: 'array', minItems: 1, uniqueItems: true, items: STR(200) } } },
++ read: { type: 'object' }, role: { enum: ['input', 'artifact'] },
++} };
++const TURN_FILES = { type: 'array', maxItems: 10, items: FILE_REF };
const COMMON = { max_tokens: { type: 'integer', minimum: 1, maximum: 16000 }, temperature: { type: ['number', 'null'], minimum: 0, maximum: 2 },
// AI Exchange F: a liability cap on the Balance hold for this request (optional; nothing when unset).
max_cost_usd: { type: ['number', 'string', 'null'] } };
@@@ -46,7 +55,7 @@ const templates = [
const workflows = [
{
key: 'ai.chat', name: 'Chat completion', namespace: 'ai',
-- input_schema: OBJ({ system: STR(200000), messages: { type: 'array', minItems: 1, maxItems: 200, items: { type: 'object', required: ['role', 'content'], properties: { role: { enum: ['user', 'assistant'] }, content: STR(200000) } } }, ...COMMON }, ['messages']),
++ input_schema: OBJ({ system: STR(200000), messages: { type: 'array', minItems: 1, maxItems: 200, items: { type: 'object', required: ['role', 'content'], properties: { role: { enum: ['user', 'assistant'] }, content: STR(200000) } } }, files: TURN_FILES, ...COMMON }, ['messages']),
output_schema: PASSTHROUGH_OUTPUT,
steps: [{ kind: 'passthrough', route: 'default.chat' }], cache_mode: 'none',
},
@@@ -109,9 +118,10 @@ const workflows = [
top_p: { type: ['number', 'null'], minimum: 0, maximum: 1 },
presence_penalty: { type: ['number', 'null'], minimum: -2, maximum: 2 },
frequency_penalty: { type: ['number', 'null'], minimum: -2, maximum: 2 },
++ files: TURN_FILES,
...COMMON,
}, ['messages']),
-- output_schema: OBJ({ choices: { type: 'array', minItems: 1, maxItems: 8, items: { type: 'object' } }, system_fingerprint: { type: ['string', 'null'] }, model: { type: ['string', 'null'] } }, ['choices']),
++ output_schema: OBJ({ choices: { type: 'array', minItems: 1, maxItems: 8, items: { type: 'object' } }, system_fingerprint: { type: ['string', 'null'] }, model: { type: ['string', 'null'] }, artifacts: { type: 'array', maxItems: 10, items: { type: 'object' } } }, ['choices']),
metadata: { passthrough: true },
steps: [{ kind: 'complete', route: 'default.chat' }], cache_mode: 'none',
},
@@@ -138,11 +148,13 @@ const workflows = [
top_p: { type: ['number', 'null'], minimum: 0, maximum: 1 },
top_k: { type: ['integer', 'null'], minimum: 0 },
metadata: { type: 'object' },
++ files: TURN_FILES,
}, ['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' },
++ artifacts: { type: 'array', maxItems: 10, items: { type: 'object' } },
}, ['content']),
metadata: { passthrough: true },
steps: [{ kind: 'messages', route: 'default.messages' }], cache_mode: 'none',
ddiff --git a/server/workflows/engine.js b/server/workflows/engine.js
iindex 9a0dd34..3d0c2fe 100644
---- a/server/workflows/engine.js
++++ b/server/workflows/engine.js
@@@ -36,6 +36,8 @@ const { PREPARE, POSTPROCESS } = require('./hooks');
const modelCatalog = require('../providers/model-catalog');
const { buildBody, buildCompleteBody } = require('../providers/openai');
const { PROVIDERS: ENGINES } = require('../research');
++const fileInputs = require('../file-inputs');
++const filesOut = require('../artifacts');
/** The last user message's text, for the queue's own record of what one item asks (never sent as-is). */
function lastUserText(messages) {
@@@ -260,10 +262,16 @@ function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spa
const route = await routeFor(routeKey);
const sys = typeof input.system === 'string' ? (input.system.trim() ? [{ text: input.system, cache: false }] : [])
: Array.isArray(input.system) ? input.system.filter(x => x && String(x.text || '').trim()).map(x => ({ text: String(x.text), cache: Boolean(x.cache) })) : [];
-- const messages = normMessages(input.messages, input.user);
++ const base = { ...input, messages: normMessages(input.messages, input.user) };
++ // Files on the turn (ai.file@1) become text or image content for this call only.
++ const turn = Array.isArray(input.files) && input.files.length
++ ? await fileInputs.attachFiles(base, fetcher, { signal: ctx.signal, format: 'openai', vision: true })
++ : base;
++ const messages = turn.messages;
const image = input.image ? await fetcher.loadImage({ ...input.image, max_width: input.image.max_width || 1024 }, { signal: ctx.signal }) : null;
const req = {
system: sys, messages, image,
++ ...(turn.hasImage ? { vision: true } : {}),
json: input.json ? { name: input.json.name || 'result', schema: input.json.schema, strict: input.json.strict !== false, description: input.json.description } : null,
maxTokens: Math.max(1, Math.round(input.max_tokens || 400)),
temperature: input.temperature == null ? null : input.temperature,
@@@ -336,6 +344,8 @@ function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spa
*/
async function prepareBatch(wf, run, input, ctx = {}) {
if (!wf || !Array.isArray(wf.steps) || wf.steps.length !== 1) return null;
++ // Attachments are read from Media at execution time; a provider batch request cannot carry them.
++ if (Array.isArray(input && input.files) && input.files.length) return null;
const step = wf.steps[0];
if (step.kind === 'llm') {
// An image cannot travel in a provider batch request here: the answer would need the media file.
@@@ -383,9 +393,13 @@ function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spa
/** complete: the OpenAI Chat Completions request as the caller wrote it (ai.complete), to an upstream that speaks it. */
async function completeStep(step, input, wf, run, ctx) {
const { route, selection } = await selectedRoute(await routeFor(step.route || 'default.chat'), input);
++ // Media-backed files on the turn (ai.file@1) become text or image content for this call only.
++ const turn = Array.isArray(input.files) && input.files.length
++ ? await fileInputs.attachFiles(input, fetcher, { signal: ctx.signal, format: 'openai', vision: true })
++ : input;
const openai = {};
for (const k of ['messages', 'tools', 'tool_choice', 'parallel_tool_calls', 'response_format', 'stop', 'seed', 'top_p', 'presence_penalty', 'frequency_penalty']) {
-- if (input[k] !== undefined) openai[k] = input[k];
++ if (turn[k] !== undefined) openai[k] = turn[k];
}
const req = {
openai,
@@@ -393,6 +407,7 @@ function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spa
temperature: input.temperature == null ? null : input.temperature,
timeoutMs: Math.max(route.timeout_ms || 0, 120000),
prefer: ctx.prefer || null,
++ ...(turn.hasImage ? { vision: true } : {}),
...(typeof ctx.onDelta === 'function' ? { onDelta: ctx.onDelta } : {}),
};
const batch = batchAnswerFor(ctx, 'complete');
@@@ -401,7 +416,11 @@ function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spa
: await pool.execute(route, 'complete', req, { ...ctx, ...selectionCtx(selection), routeKey: route.key, routeVersion: route.version, promptHash: sha256({ m: openai.messages, t: openai.tools || null }) });
const c = exec.result.completion;
if (!c || !Array.isArray(c.choices) || !c.choices.length) throw new AiError(502, 'output.empty', 'the provider answered, but with no choices');
-- return { output: { choices: c.choices, system_fingerprint: c.system_fingerprint || null, model: exec.model }, exec, route };
++ // Files the model produced beside its text: Media references pass through; inline bytes need the
++ // configured Media store (server/artifacts.js), never the run's output.
++ const subject = ctx.onBehalfOf && ctx.onBehalfOf.id ? ctx.onBehalfOf.id : null;
++ const artifacts = await filesOut.collectArtifacts(c.choices[0] && c.choices[0].message, { store: filesOut.storeOf(config), subject });
++ return { output: { choices: c.choices, system_fingerprint: c.system_fingerprint || null, model: exec.model, ...(artifacts.length ? { artifacts } : {}) }, exec, route };
}
/**
@@@ -410,20 +429,27 @@ function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spa
*/
async function messagesStep(step, input, wf, run, ctx) {
const { route, selection } = await selectedRoute(await routeFor(step.route || 'default.messages'), input);
++ // Media-backed files on the turn (ai.file@1): text parts, or base64 image sources for a vision model.
++ const turn = Array.isArray(input.files) && input.files.length
++ ? await fileInputs.attachFiles(input, fetcher, { signal: ctx.signal, format: 'anthropic', vision: true })
++ : input;
const req = {
-- messages: input.messages,
++ messages: turn.messages,
maxTokens: Math.max(1, Math.round(input.max_tokens || route.max_output_tokens || 1024)),
timeoutMs: Math.max(route.timeout_ms || 0, 120000),
prefer: ctx.prefer || null,
++ ...(turn.hasImage ? { vision: true } : {}),
...(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, ...selectionCtx(selection), routeKey: route.key, routeVersion: route.version, promptHash: sha256({ m: input.messages, t: input.tools || null }) });
++ const exec = await pool.execute(route, 'messages', req, { ...ctx, ...selectionCtx(selection), routeKey: route.key, routeVersion: route.version, promptHash: sha256({ m: turn.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 };
++ const subject = ctx.onBehalfOf && ctx.onBehalfOf.id ? ctx.onBehalfOf.id : null;
++ const artifacts = await filesOut.collectArtifacts(m, { store: filesOut.storeOf(config), subject });
++ return { output: { ...m, model: m.model || exec.model, ...(artifacts.length ? { artifacts } : {}) }, exec, route };
}
async function transcribeStep(step, input, wf, run, ctx) {
ddiff --git a/test/file-artifacts.test.js b/test/file-artifacts.test.js
nnew file mode 100644
iindex 0000000..ce25b0f
---- /dev/null
++++ b/test/file-artifacts.test.js
@@@ -0,0 +1,45 @@
++'use strict';
++const assert = require('node:assert/strict');
++const { collectArtifacts, storeOf } = require('../server/artifacts');
++const { suite } = require('./helpers');
++
++const t = suite('file-artifacts');
++const ID = 'med_01ARZ3NDEKTSV4RRFFQ69G5FAV';
++
++t.test('a Media reference the model returned becomes a stored artifact', async () => {
++ const out = await collectArtifacts({ content: 'Here it is.', artifacts: [{ id: ID, name: 'answer.csv', media_type: 'text/csv', size: 8, storage: { provider: 'media', key: ID }, access: { readers: ['usr_01ARZ3NDEKTSV4RRFFQ69G5FAV'] } }] });
++ assert.equal(out.length, 1);
++ assert.equal(out[0].role, 'artifact');
++ assert.equal(out[0].storage.provider, 'media');
++ assert.equal(out[0].read.method, 'not_readable');
++ assert.equal('content' in out[0].read, false, 'the run output does not inline the bytes');
++});
++
++t.test('inline bytes need a Media store; without one they are refused, not dropped', async () => {
++ await assert.rejects(() => collectArtifacts({ artifacts: [{ name: 'a.txt', media_type: 'text/plain', data: Buffer.from('hi').toString('base64') }] }),
++ (err) => err.status === 503 && err.code === 'files.store_unavailable');
++});
++
++t.test('inline bytes are stored through the configured store and returned by reference', async () => {
++ const puts = [];
++ const store = { put: async (what) => { puts.push(what); return { id: ID, name: what.name, media_type: what.media_type, size: what.bytes.length, storage: { provider: 'media', key: ID }, access: { readers: [what.subject] }, read: { method: 'not_readable', reason: 'Stored by Media; read it by reference.' }, role: 'artifact' }; } };
++ const out = await collectArtifacts({ artifacts: [{ name: 'a.txt', media_type: 'text/plain', data: Buffer.from('hi').toString('base64') }] }, { store, subject: 'usr_01ARZ3NDEKTSV4RRFFQ69G5FAV' });
++ assert.equal(puts.length, 1);
++ assert.equal(puts[0].bytes.toString(), 'hi');
++ assert.equal(puts[0].subject, 'usr_01ARZ3NDEKTSV4RRFFQ69G5FAV');
++ assert.equal(out[0].id, ID);
++});
++
++t.test('storeOf finds a configured store and nothing else', () => {
++ assert.equal(storeOf({}), null);
++ assert.equal(storeOf({ media: {} }), null);
++ const store = { put() {} };
++ assert.equal(storeOf({ media: { artifactStore: store } }), store);
++});
++
++t.test('more than ten produced files is a refusal', async () => {
++ const artifacts = Array.from({ length: 11 }, () => ({ id: ID, name: 'a.txt', media_type: 'text/plain', size: 1 }));
++ await assert.rejects(() => collectArtifacts({ artifacts }), (err) => err.status === 413 && err.code === 'files.too_many');
++});
++
++t.run();
ddiff --git a/test/file-inputs.test.js b/test/file-inputs.test.js
nnew file mode 100644
iindex 0000000..734d68a
---- /dev/null
++++ b/test/file-inputs.test.js
@@@ -0,0 +1,123 @@
++'use strict';
++const assert = require('node:assert/strict');
++const { attachFiles, readFile, pdfText, setScanner, MAX_FILES, MAX_FILE_BYTES } = require('../server/file-inputs');
++const { suite } = require('./helpers');
++
++const t = suite('file-inputs');
++const ID = (n = 1) => `med_01ARZ3NDEKTSV4RRFFQ69G5F${String(n).padStart(2, '0').slice(-2)}`;
++
++/** A PDF whose one content stream shows `text` (uncompressed, so the extractor is what is tested). */
++function pdf(text) {
++ const stream = `BT /F1 12 Tf 72 720 Td (${String(text).replace(/[()\\]/g, '\\$&')}) Tj ET`;
++ return Buffer.from(`%PDF-1.4\n1 0 obj<</Type/Catalog/Pages 2 0 R>>endobj\n2 0 obj<</Type/Pages/Kids[3 0 R]/Count 1>>endobj\n3 0 obj<</Type/Page/Parent 2 0 R/Contents 4 0 R>>endobj\n4 0 obj<</Length ${stream.length}>>stream\n${stream}\nendstream\nendobj\n%%EOF`, 'latin1');
++}
++
++function fetcherOf(files) {
++ return {
++ mediaRefUrl: ({ media_id }) => `https://openvibe.media/o/${media_id}`,
++ fetchUrl: async (url) => {
++ const id = String(url).split('/').pop();
++ const f = files[id];
++ if (!f) { const e = new Error('not found'); e.status = 404; throw e; }
++ return { buffer: Buffer.from(f.body), contentType: f.type };
++ },
++ };
++}
++
++const csv = { body: 'name,score\nAda,10\nGrace,9\n', type: 'text/csv' };
++const pdfFile = { body: pdf('Quarterly report: revenue up 12%'), type: 'application/pdf' };
++const png = { body: Buffer.from([0x89, 0x50, 0x4e, 0x47, 1, 2, 3]), type: 'image/png' };
++const zip = { body: Buffer.from([0x50, 0x4b, 3, 4, 5]), type: 'application/zip' };
++
++t.test('CSV, PDF, image and unknown types become model content by type; references never carry bytes', async () => {
++ const fetcher = fetcherOf({ [ID(1)]: csv, [ID(2)]: pdfFile, [ID(3)]: png, [ID(4)]: zip });
++ const out = await attachFiles({
++ messages: [{ role: 'user', content: 'Read these' }],
++ files: [
++ { id: ID(1), name: 'scores.csv', media_type: 'text/csv', size: 28 },
++ { id: ID(2), name: 'report.pdf', media_type: 'application/pdf', size: 300 },
++ { id: ID(3), name: 'shot.png', media_type: 'image/png', size: 8 },
++ { id: ID(4), name: 'bundle.zip', media_type: 'application/zip', size: 5 },
++ ],
++ }, fetcher);
++ const parts = out.messages[0].content;
++ assert.ok(Array.isArray(parts));
++ const texts = parts.filter((p) => p.type === 'text').map((p) => p.text).join('\n');
++ assert.match(texts, /name,score/);
++ assert.match(texts, /Quarterly report: revenue up 12%/);
++ assert.match(texts, /bundle\.zip \(application\/zip\); this file type cannot be read/);
++ assert.ok(parts.some((p) => p.type === 'image_url' && /^data:image\/png;base64,/.test(p.image_url.url)), 'the picture goes to a vision model');
++ assert.equal(out.files.length, 4);
++ assert.deepEqual(out.files.map((f) => f.read.method), ['text', 'pages', 'image', 'not_readable']);
++ for (const f of out.files) {
++ assert.equal(f.storage.provider, 'media');
++ assert.equal('content' in f.read, false, 'extracted text is not kept in the reference');
++ assert.equal('pages' in f.read, false, 'PDF pages are not kept in the reference');
++ }
++ assert.equal(out.hasImage, true);
++});
++
++t.test('a text-only model gets an honest note instead of a dropped picture', async () => {
++ const fetcher = fetcherOf({ [ID(3)]: png });
++ const out = await attachFiles({ messages: [{ role: 'user', content: 'What is this?' }], files: [{ id: ID(3), name: 'shot.png', size: 8 }] }, fetcher, { vision: false });
++ const texts = out.messages[0].content.filter((p) => p.type === 'text').map((p) => p.text).join('');
++ assert.match(texts, /cannot see images/);
++ assert.ok(!out.messages[0].content.some((p) => p.type === 'image_url'));
++ assert.equal(out.hasImage, false);
++});
++
++t.test('the Anthropic shape gets its own image source and text parts', async () => {
++ const fetcher = fetcherOf({ [ID(1)]: csv, [ID(3)]: png });
++ const out = await attachFiles({
++ messages: [{ role: 'user', content: [{ type: 'text', text: 'Look' }] }],
++ files: [{ id: ID(1), name: 'scores.csv', size: 28 }, { id: ID(3), name: 'shot.png', size: 8 }],
++ }, fetcher, { format: 'anthropic' });
++ const parts = out.messages[0].content;
++ assert.equal(parts[0].type, 'text');
++ assert.ok(parts.some((p) => p.type === 'image' && p.source.type === 'base64' && p.source.media_type === 'image/png'));
++ assert.ok(!parts.some((p) => p.type === 'image_url'));
++});
++
++t.test('a caller-supplied read is used without fetching Media', async () => {
++ const fetcher = fetcherOf({});
++ const out = await attachFiles({
++ messages: [{ role: 'user', content: 'Summarize' }],
++ files: [{ id: ID(1), name: 'notes.txt', size: 11, read: { method: 'text', content: 'hello world' } }],
++ }, fetcher);
++ assert.match(out.messages[0].content[1].text, /hello world/);
++ assert.equal(out.files[0].read.method, 'text');
++});
++
++t.test('the PDF extractor reads Tj/TJ text and is honest about a PDF with no text', async () => {
++ assert.deepEqual(pdfText(pdf('Hello PDF')).pages, ['Hello PDF']);
++ assert.equal(pdfText(Buffer.from('%PDF-1.4\n1 0 obj<</Encrypt 2 0 R>>>>\n%%EOF')).reason.includes('encrypted'), true);
++});
++
++t.test('per-turn count and size caps refuse before any fetch', async () => {
++ const fetcher = fetcherOf({});
++ const many = Array.from({ length: MAX_FILES + 1 }, (_, i) => ({ id: ID(i + 1), name: `f${i}.txt`, size: 1 }));
++ await assert.rejects(() => attachFiles({ messages: [{ role: 'user', content: 'x' }], files: many }, fetcher), /at most 10 files/);
++ await assert.rejects(() => attachFiles({
++ messages: [{ role: 'user', content: 'x' }],
++ files: [{ id: ID(1), name: 'big.bin', size: MAX_FILE_BYTES + 1 }],
++ }, fetcher), /up to 10 MiB/);
++});
++
++t.test('the virus scan is a named extension point, not a pretend check', async () => {
++ const fetcher = fetcherOf({ [ID(1)]: csv });
++ let seen = null;
++ setScanner(async (info) => { seen = info; throw new Error('infected'); });
++ try {
++ await assert.rejects(() => attachFiles({ messages: [{ role: 'user', content: 'x' }], files: [{ id: ID(1), name: 'scores.csv', size: 28 }] }, fetcher), /infected/);
++ assert.equal(seen.name, 'scores.csv');
++ assert.equal(seen.media_id, ID(1));
++ } finally { setScanner(null); }
++ const out = await attachFiles({ messages: [{ role: 'user', content: 'x' }], files: [{ id: ID(1), name: 'scores.csv', size: 28 }] }, fetcher);
++ assert.equal(out.files[0].read.method, 'text');
++});
++
++t.test('a file that Media does not serve is a clear 422, not an empty turn', async () => {
++ await assert.rejects(() => readFile({ id: ID(9), name: 'gone.txt', size: 1 }, fetcherOf({})), (err) => err.status === 422 && /not a readable Media object/.test(err.detail));
++});
++
++t.run();
ddiff --git a/test/file-turn.test.js b/test/file-turn.test.js
nnew file mode 100644
iindex 0000000..c679ac3
---- /dev/null
++++ b/test/file-turn.test.js
@@@ -0,0 +1,64 @@
++'use strict';
++// A turn carries files by Media reference: the provider gets their text (or image), the run does not.
++const assert = require('node:assert/strict');
++const nodeHttp = require('node:http');
++const { boot, request, token, suite, seamServer, ALL } = require('./helpers');
++
++const t = suite('file-turn');
++const ID = 'med_01ARZ3NDEKTSV4RRFFQ69G5FAV';
++let h; let seam; let media; let service;
++
++t.test('a Media server and a counting OpenAI-shaped provider', async () => {
++ media = nodeHttp.createServer((req, res) => {
++ if (req.url === `/o/${ID}`) { res.writeHead(200, { 'Content-Type': 'text/csv' }); return res.end('name,score\nAda,10\n'); }
++ res.writeHead(404, { 'Content-Type': 'application/json' }); res.end('{"error":"Not found"}');
++ });
++ await new Promise((r) => media.listen(0, '127.0.0.1', r));
++ const mediaUrl = `http://127.0.0.1:${media.address().port}`;
++ seam = await seamServer(() => ({ choices: [{ index: 0, message: { role: 'assistant', content: 'It has two rows.' } }], usage: { prompt_tokens: 10, completion_tokens: 5 } }));
++ h = await boot({ env: { AI_STUB_FALLBACK: 'false', OV_MEDIA_URL: mediaUrl, OV_MEDIA_INTERNAL_URL: mediaUrl } });
++ service = token('chat', ALL);
++ let r = await request(h.base, 'POST', '/api/v1/providers', { tok: service, body: { key: 'seam', kind: 'openai', base_url: `${seam.url}/v1`, auth_mode: 'none', capabilities: ['chat', 'complete'], default_model: 'm1' } });
++ assert.equal(r.status, 201, r.text);
++ r = await request(h.base, 'POST', '/api/v1/models', { tok: service, body: { provider_key: 'seam', model_key: 'm1', cost: { in_per_mtok: 1, out_per_mtok: 1 } } });
++ assert.equal(r.status, 201, r.text);
++ r = await request(h.base, 'POST', '/api/v1/routes/default.chat/versions', { tok: service, body: { primary: { provider: 'seam', model: 'm1' }, fallbacks: [] } });
++ assert.equal(r.status, 201, r.text);
++});
++
++t.test('POST /v1/chat/completions reads the file into the provider request, not the run', async () => {
++ const r = await request(h.base, 'POST', '/v1/chat/completions', {
++ tok: service,
++ body: {
++ model: 'openvibe', messages: [{ role: 'user', content: 'Count the rows' }],
++ files: [{ id: ID, name: 'scores.csv', media_type: 'text/csv', size: 20, storage: { provider: 'media', key: ID }, access: { readers: ['usr_01ARZ3NDEKTSV4RRFFQ69G5FAV'] } }],
++ },
++ });
++ assert.equal(r.status, 200, r.text);
++ const sent = JSON.stringify(seam.last.messages);
++ assert.match(sent, /Attached file: scores\.csv/);
++ assert.match(sent, /name,score/);
++ assert.equal(seam.calls, 1);
++ const stored = await h.db.prepare('SELECT input FROM runs ORDER BY created_at DESC LIMIT 1').get();
++ const raw = typeof stored.input === 'string' ? stored.input : JSON.stringify(stored.input);
++ assert.ok(!raw.includes('Ada,10'), 'the file bytes are not in the run input');
++});
++
++t.test('an unreadable Media reference is refused before the provider is called', async () => {
++ const before = seam.calls;
++ const r = await request(h.base, 'POST', '/v1/chat/completions', {
++ tok: service,
++ body: { model: 'openvibe', messages: [{ role: 'user', content: 'x' }], files: [{ id: 'med_01ARZ3NDEKTSV4RRFFQ69G5FZZ', name: 'gone.txt', size: 1 }] },
++ });
++ assert.equal(r.status, 503, r.text); // the run failed at the file; the provider was never called
++ assert.equal(r.body.error.code, 'files.unreadable');
++ assert.equal(seam.calls, before);
++});
++
++t.test('close', async () => {
++ await h.stop();
++ await seam.close();
++ await new Promise((r) => media.close(r));
++});
++
++t.run();
ddiff --git a/test/quota-files.test.js b/test/quota-files.test.js
nnew file mode 100644
iindex 0000000..a96f393
---- /dev/null
++++ b/test/quota-files.test.js
@@@ -0,0 +1,52 @@
++'use strict';
++// The per-person file limits live in the existing quota system: a quota row's max_files / max_file_bytes
++// caps what that scope may attach in its window, and the usage counters keep the running totals.
++const assert = require('node:assert/strict');
++const { testDb } = require('./db');
++const { createQuotas } = require('../server/quota');
++const { suite } = require('./helpers');
++
++const t = suite('quota-files');
++let h;
++const person = { requesterType: 'service', requesterId: 'chat', actorKey: 'user:usr_01ARZ3NDEKTSV4RRFFQ69G5FAV', workflowKey: 'ai.complete' };
++
++t.test('a migrated database', async () => { h = await testDb(); });
++t.test('an actor quota caps one person\'s files and file bytes in its window', async () => {
++ const quotas = createQuotas(h.db, {});
++ await quotas.upsert({ scope_type: 'actor', scope_id: person.actorKey, window: 'day', max_files: 3, max_file_bytes: 100 });
++ await quotas.chargeFiles(person, [{ size: 10 }, { size: 20 }]);
++ await quotas.chargeFiles(person, [{ size: 70 }]);
++ const used = (await quotas.counters(person))[0].used;
++ assert.equal(used.files, 3);
++ assert.equal(used.file_bytes, 100);
++ await assert.rejects(() => quotas.chargeFiles(person, [{ size: 1 }]), (err) => err.status === 429 && /3 files per day/.test(err.detail));
++ await assert.rejects(() => quotas.chargeFiles(person, [{ size: 1 }]), (err) => err.status === 429 && err.extra && err.extra.retry_after_seconds >= 1);
++});
++
++t.test('another person is not limited by the first person\'s quota', async () => {
++ const quotas = createQuotas(h.db, {});
++ const other = { ...person, actorKey: 'user:usr_01ARZ3NDEKTSV4RRFFQ69G5FBV' };
++ const out = await quotas.chargeFiles(other, [{ size: 5 }, { size: 5 }, { size: 5 }, { size: 5 }]);
++ assert.deepEqual(out, { files: 4, file_bytes: 20 });
++});
++
++t.test('a quota with no file limits still counts files for the window', async () => {
++ const quotas = createQuotas(h.db, {});
++ const ctx = { ...person, actorKey: 'user:usr_01ARZ3NDEKTSV4RRFFQ69G5FCV' };
++ await quotas.chargeFiles(ctx, [{ size: 7 }]);
++ const row = await h.db.prepare("SELECT files, file_bytes FROM usage_counters WHERE scope_type = 'actor' AND scope_id = ? AND \"window\" = 'day'").get(ctx.actorKey);
++ assert.equal(Number(row.files), 1);
++ assert.equal(Number(row.file_bytes), 7);
++});
++
++t.test('the admin quota accepts the new fields and validates them', async () => {
++ const quotas = createQuotas(h.db, {});
++ const row = await quotas.upsert({ scope_type: 'global', scope_id: '*', window: 'hour', max_files: 2, max_file_bytes: 500 });
++ assert.equal(Number(row.max_files), 2);
++ assert.equal(Number(row.max_file_bytes), 500);
++ await assert.rejects(() => quotas.upsert({ scope_type: 'global', scope_id: '*', window: 'hour', max_files: -1 }), /max_files must be >= 0/);
++});
++
++t.test('close', async () => { await h.close(); });
++
++t.run();