diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..20282d7 --- /dev/null +++ b/.env.example @@ -0,0 +1,9 @@ +# Required Postgres connection string +DATABASE_URL=postgres://user:password@localhost:5432/opencode_proxy + +# Optional: use this instead of keys.txt. Must be a JSON array of strings. +# OPENCODE_API_KEYS=["go-first-key","go-second-key"] + +# Optional server settings +PORT=4005 +MAX_CONTEXT_TOKENS=262144 diff --git a/.gitignore b/.gitignore index af9e112..e1d3f85 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,7 @@ node_modules/ +.env +.env.* +!.env.example keys.txt npm-debug.log* .DS_Store diff --git a/README.md b/README.md index 7509293..5ae2f1f 100644 --- a/README.md +++ b/README.md @@ -8,8 +8,11 @@ Requires Node.js 18 or newer. ```sh npm install +cp .env.example .env ``` +Edit `.env` with your Postgres connection string and optional server settings. Values already present in the process environment take precedence over `.env`, which makes the same configuration work locally and in hosted deployments. `.env` is ignored by Git. + Create `keys.txt` in the project root. Put one OpenCode Go API key on each line. Blank lines and comments are ignored; inline comments are supported. ```text @@ -34,7 +37,7 @@ The global context limit defaults to 256k tokens and can be changed with `MAX_CO Inference requests are limited per client IP to 2 requests per second, 100 requests per five hours, and $10 of reported upstream cost per five hours. Railway's `X-Real-IP` header is used to identify clients. -Start the proxy: +Set `DATABASE_URL` in `.env` (or the process environment) to a Postgres connection string. The proxy creates its `requests` table and index automatically on startup. Every incoming request and its complete response—including streaming responses and errors—is stored with headers, status, model, client IP, and timestamp. Successful inference requests also store a normalized training conversation containing the full chat history and generated assistant output, including reasoning, tool calls, and tool results. ```sh npm start @@ -62,6 +65,31 @@ The Anthropic-compatible adapter exposes `POST /ant/v1/messages` and `GET /ant/v For Claude Code, either set `ANTHROPIC_BASE_URL=http://localhost:4005/ant` or use the displayed `http://localhost:4005/ant/v1` value; both forms are supported. +## Exporting training data + +Exports use JSONL: each line is one inference request containing its complete input history followed by the generated assistant message. The normalized OpenAI-style `messages` preserve system/user/assistant roles, `reasoning_content`, assistant `tool_calls`, and `tool` results. Non-chat routes and failed inference requests are excluded. + +Each line also has `metadata` with the exact tool definitions supplied on that request, tool choice, model, API format, endpoint, streaming mode, generation parameters, caller-supplied request metadata, response ID/model, finish reason, usage, HTTP status, duration, request ID, and timestamp. Tool metadata remains in its original OpenAI, Responses, or Anthropic format so no provider-specific schema information is lost. + +```json +{"messages":[{"role":"user","content":"What is 2+2?"},{"role":"assistant","content":"4","reasoning_content":"Adding the values gives four."}],"metadata":{"schema_version":1,"api":"openai_chat_completions","model":"model-a","stream":false,"tools":[],"response_status":200}} +``` + +Omit the limit to export every training request, newest first: + +```sh +npm run --silent export > requests.jsonl +``` + +Pass a positive limit to export that many of the most recent training requests: + +```sh +npm run --silent export -- 100 > recent-requests.jsonl +# Equivalent: npm run --silent export -- --limit 100 +``` + +The export command uses the same `DATABASE_URL` as the server. `--silent` suppresses npm's banner, and the script's progress message is written to standard error, so redirected JSONL remains valid. Rows collected before normalized training storage was added are normalized from their saved raw request and response during export. + ## Tests ```sh diff --git a/package-lock.json b/package-lock.json index f92c6c0..7f381e9 100644 --- a/package-lock.json +++ b/package-lock.json @@ -8,7 +8,9 @@ "name": "opencode-go-proxy", "version": "1.0.0", "dependencies": { - "express": "^5.1.0" + "dotenv": "^16.6.1", + "express": "^5.1.0", + "pg": "^8.22.0" }, "devDependencies": { "supertest": "^7.1.1", @@ -1256,6 +1258,18 @@ "wrappy": "1" } }, + "node_modules/dotenv": { + "version": "16.6.1", + "resolved": "https://registry.npmjs.org/dotenv/-/dotenv-16.6.1.tgz", + "integrity": "sha512-uBq4egWHTcTt33a72vpSG0z3HnPuIl6NqYcTrKEg2azoEyl2hpW0zqlxysq2pK9HlDIHyHyakeYaYnSAwd8bow==", + "license": "BSD-2-Clause", + "engines": { + "node": ">=12" + }, + "funding": { + "url": "https://dotenvx.com" + } + }, "node_modules/dunder-proto": { "version": "1.0.1", "resolved": "https://registry.npmjs.org/dunder-proto/-/dunder-proto-1.0.1.tgz", @@ -1955,6 +1969,95 @@ "node": ">= 14.16" } }, + "node_modules/pg": { + "version": "8.22.0", + "resolved": "https://registry.npmjs.org/pg/-/pg-8.22.0.tgz", + "integrity": "sha512-8wih1vVIBMxoUM2oB4soJsD9tDnDpLv4OXBJ+EJzFsvycD+lfyIreC2gGHq78f8jbLLt+bvlPTFdFZfJkOuzAA==", + "license": "MIT", + "dependencies": { + "pg-connection-string": "^2.14.0", + "pg-pool": "^3.14.0", + "pg-protocol": "^1.15.0", + "pg-types": "2.2.0", + "pgpass": "1.0.5" + }, + "engines": { + "node": ">= 16.0.0" + }, + "optionalDependencies": { + "pg-cloudflare": "^1.4.0" + }, + "peerDependencies": { + "pg-native": ">=3.0.1" + }, + "peerDependenciesMeta": { + "pg-native": { + "optional": true + } + } + }, + "node_modules/pg-cloudflare": { + "version": "1.4.0", + "resolved": "https://registry.npmjs.org/pg-cloudflare/-/pg-cloudflare-1.4.0.tgz", + "integrity": "sha512-Vo7z/6rrQYxpNRylp4Tlob2elzbh+N/MOQbxFVWCxS7oEx6jF53GTJFxK2WWpKuBRkmiin4Mt+xofFDjx09R0A==", + "license": "MIT", + "optional": true + }, + "node_modules/pg-connection-string": { + "version": "2.14.0", + "resolved": "https://registry.npmjs.org/pg-connection-string/-/pg-connection-string-2.14.0.tgz", + "integrity": "sha512-XwWDGcLRGCXAR8F/AM5bG7Q+A3Wm2s6QeEjlOKZLlH3UYcguiqCWKyWXVag5TLTIjR7oOJUY8kcADaZgWPyLeg==", + "license": "MIT" + }, + "node_modules/pg-int8": { + "version": "1.0.1", + "resolved": "https://registry.npmjs.org/pg-int8/-/pg-int8-1.0.1.tgz", + "integrity": "sha512-WCtabS6t3c8SkpDBUlb1kjOs7l66xsGdKpIPZsg4wR+B3+u9UAum2odSsF9tnvxg80h4ZxLWMy4pRjOsFIqQpw==", + "license": "ISC", + "engines": { + "node": ">=4.0.0" + } + }, + "node_modules/pg-pool": { + "version": "3.14.0", + "resolved": "https://registry.npmjs.org/pg-pool/-/pg-pool-3.14.0.tgz", + "integrity": "sha512-gKtPkFdQPU3DksooVLi9LsjZxrsBUZIpa+7aVx+LV5pNh0KzP4Zleud2po+ConrxbuXGBJ6Hfer6hdgpIBpBaw==", + "license": "MIT", + "peerDependencies": { + "pg": ">=8.0" + } + }, + "node_modules/pg-protocol": { + "version": "1.15.0", + "resolved": "https://registry.npmjs.org/pg-protocol/-/pg-protocol-1.15.0.tgz", + "integrity": "sha512-cq9sECI5s0+uPUXjbz8ioyPJni6RzsRib0US67i5IoTZKw8fNeYlVE7u8F4dG7vEJJtc5wdD1K189lCCUwqWTQ==", + "license": "MIT" + }, + "node_modules/pg-types": { + "version": "2.2.0", + "resolved": "https://registry.npmjs.org/pg-types/-/pg-types-2.2.0.tgz", + "integrity": "sha512-qTAAlrEsl8s4OiEQY69wDvcMIdQN6wdz5ojQiOy6YRMuynxenON0O5oCpJI6lshc6scgAY8qvJ2On/p+CXY0GA==", + "license": "MIT", + "dependencies": { + "pg-int8": "1.0.1", + "postgres-array": "~2.0.0", + "postgres-bytea": "~1.0.0", + "postgres-date": "~1.0.4", + "postgres-interval": "^1.1.0" + }, + "engines": { + "node": ">=4" + } + }, + "node_modules/pgpass": { + "version": "1.0.5", + "resolved": "https://registry.npmjs.org/pgpass/-/pgpass-1.0.5.tgz", + "integrity": "sha512-FdW9r/jQZhSeohs1Z3sI1yxFQNFvMcnmfuj4WBMUTxOrAyLMaTcE1aAMBiTlbMNaXvBCQuVi0R7hd8udDSP7ug==", + "license": "MIT", + "dependencies": { + "split2": "^4.1.0" + } + }, "node_modules/picocolors": { "version": "1.1.1", "resolved": "https://registry.npmjs.org/picocolors/-/picocolors-1.1.1.tgz", @@ -2004,6 +2107,45 @@ "node": "^10 || ^12 || >=14" } }, + "node_modules/postgres-array": { + "version": "2.0.0", + "resolved": "https://registry.npmjs.org/postgres-array/-/postgres-array-2.0.0.tgz", + "integrity": "sha512-VpZrUqU5A69eQyW2c5CA1jtLecCsN2U/bD6VilrFDWq5+5UIEVO7nazS3TEcHf1zuPYO/sqGvUvW62g86RXZuA==", + "license": "MIT", + "engines": { + "node": ">=4" + } + }, + "node_modules/postgres-bytea": { + "version": "1.0.1", + "resolved": "https://registry.npmjs.org/postgres-bytea/-/postgres-bytea-1.0.1.tgz", + "integrity": "sha512-5+5HqXnsZPE65IJZSMkZtURARZelel2oXUEO8rH83VS/hxH5vv1uHquPg5wZs8yMAfdv971IU+kcPUczi7NVBQ==", + "license": "MIT", + "engines": { + "node": ">=0.10.0" + } + }, + "node_modules/postgres-date": { + "version": "1.0.7", + "resolved": "https://registry.npmjs.org/postgres-date/-/postgres-date-1.0.7.tgz", + "integrity": "sha512-suDmjLVQg78nMK2UZ454hAG+OAW+HQPZ6n++TNDUX+L0+uUlLywnoxJKDou51Zm+zTCjrCl0Nq6J9C5hP9vK/Q==", + "license": "MIT", + "engines": { + "node": ">=0.10.0" + } + }, + "node_modules/postgres-interval": { + "version": "1.2.0", + "resolved": "https://registry.npmjs.org/postgres-interval/-/postgres-interval-1.2.0.tgz", + "integrity": "sha512-9ZhXKM/rw350N1ovuWHbGxnGh/SNJ4cnxHiM0rxE4VN41wsg8P8zWn9hv/buK00RP4WvlOyr/RBDiptyxVbkZQ==", + "license": "MIT", + "dependencies": { + "xtend": "^4.0.0" + }, + "engines": { + "node": ">=0.10.0" + } + }, "node_modules/proxy-addr": { "version": "2.0.7", "resolved": "https://registry.npmjs.org/proxy-addr/-/proxy-addr-2.0.7.tgz", @@ -2268,6 +2410,15 @@ "node": ">=0.10.0" } }, + "node_modules/split2": { + "version": "4.2.0", + "resolved": "https://registry.npmjs.org/split2/-/split2-4.2.0.tgz", + "integrity": "sha512-UcjcJOWknrNkF6PLX83qcHM6KHgVKNkV62Y8a5uYDVv9ydGQVwAHMKqHdJje1VTWpljG0WYpCDhrCdAOYH4TWg==", + "license": "ISC", + "engines": { + "node": ">= 10.x" + } + }, "node_modules/stackback": { "version": "0.0.2", "resolved": "https://registry.npmjs.org/stackback/-/stackback-0.0.2.tgz", @@ -2652,6 +2803,15 @@ "resolved": "https://registry.npmjs.org/wrappy/-/wrappy-1.0.2.tgz", "integrity": "sha512-l4Sp/DRseor9wL6EvV2+TuQn63dMkPjZ/sp9XkghTEbV9KlPS1xUsZ3u7/IQO4wxtcFB4bgpQPRcR3QCvezPcQ==", "license": "ISC" + }, + "node_modules/xtend": { + "version": "4.0.2", + "resolved": "https://registry.npmjs.org/xtend/-/xtend-4.0.2.tgz", + "integrity": "sha512-LKYU1iAXJXUgAXn9URjiu+MWhyUXHsvfp7mcuYm9dSUKK0/CjtrUwFAxD82/mCWbtLsGjFIad0wIsod4zrTAEQ==", + "license": "MIT", + "engines": { + "node": ">=0.4" + } } } } diff --git a/package.json b/package.json index 6503775..e30aefe 100644 --- a/package.json +++ b/package.json @@ -7,6 +7,7 @@ "main": "src/server.js", "scripts": { "start": "node src/server.js", + "export": "node src/export.js", "keys:env": "node src/keys-env.js", "test": "vitest run" }, @@ -14,7 +15,9 @@ "node": ">=18" }, "dependencies": { - "express": "^5.1.0" + "dotenv": "^16.6.1", + "express": "^5.1.0", + "pg": "^8.22.0" }, "devDependencies": { "supertest": "^7.1.1", diff --git a/src/db.js b/src/db.js new file mode 100644 index 0000000..c3560f7 --- /dev/null +++ b/src/db.js @@ -0,0 +1,74 @@ +import pg from 'pg'; + +const { Pool } = pg; + +export function createPool(connectionString = process.env.DATABASE_URL) { + if (!connectionString) throw new Error('DATABASE_URL is required'); + return new Pool({ connectionString }); +} + +export async function ensureSchema(pool) { + await pool.query(` + CREATE TABLE IF NOT EXISTS requests ( + id BIGSERIAL PRIMARY KEY, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + method TEXT NOT NULL, + endpoint TEXT NOT NULL, + request_headers JSONB NOT NULL DEFAULT '{}'::jsonb, + request_body JSONB, + response_headers JSONB NOT NULL DEFAULT '{}'::jsonb, + response_body TEXT, + response_status INTEGER NOT NULL, + duration_ms DOUBLE PRECISION, + model TEXT, + client_ip TEXT, + stream BOOLEAN NOT NULL DEFAULT FALSE, + training_data JSONB + ) + `); + + // Keep databases created by an earlier version compatible. + await pool.query(` + ALTER TABLE requests ADD COLUMN IF NOT EXISTS method TEXT; + ALTER TABLE requests ADD COLUMN IF NOT EXISTS request_headers JSONB NOT NULL DEFAULT '{}'::jsonb; + ALTER TABLE requests ADD COLUMN IF NOT EXISTS response_headers JSONB NOT NULL DEFAULT '{}'::jsonb; + ALTER TABLE requests ADD COLUMN IF NOT EXISTS duration_ms DOUBLE PRECISION; + ALTER TABLE requests ADD COLUMN IF NOT EXISTS training_data JSONB; + `); + await pool.query('CREATE INDEX IF NOT EXISTS requests_created_at_idx ON requests (created_at DESC, id DESC)'); +} + +export async function insertRequest(pool, entry) { + await pool.query( + `INSERT INTO requests ( + method, endpoint, request_headers, request_body, response_headers, + response_body, response_status, duration_ms, model, client_ip, stream, training_data + ) VALUES ($1, $2, $3::jsonb, $4::jsonb, $5::jsonb, $6, $7, $8, $9, $10, $11, $12::jsonb)`, + [ + entry.method, + entry.endpoint, + JSON.stringify(entry.requestHeaders ?? {}), + entry.requestBody == null ? null : JSON.stringify(entry.requestBody), + JSON.stringify(entry.responseHeaders ?? {}), + entry.responseBody ?? null, + entry.responseStatus, + entry.durationMs ?? null, + entry.model ?? null, + entry.clientIp ?? null, + Boolean(entry.stream), + entry.trainingData == null ? null : JSON.stringify(entry.trainingData), + ], + ); +} + +export function createRequestLogger(pool, { onError = console.error } = {}) { + return { + async log(entry) { + try { + await insertRequest(pool, entry); + } catch (error) { + onError(`Failed to store request: ${error.message}`); + } + }, + }; +} diff --git a/src/env.js b/src/env.js new file mode 100644 index 0000000..274a9e4 --- /dev/null +++ b/src/env.js @@ -0,0 +1,11 @@ +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; +import dotenv from 'dotenv'; + +export const projectRoot = path.dirname(path.dirname(fileURLToPath(import.meta.url))); + +export function loadEnv(envPath = path.join(projectRoot, '.env')) { + return dotenv.config({ path: envPath }); +} + +loadEnv(); diff --git a/src/export.js b/src/export.js new file mode 100644 index 0000000..8d10432 --- /dev/null +++ b/src/export.js @@ -0,0 +1,90 @@ +import './env.js'; +import { once } from 'node:events'; +import { createPool, ensureSchema } from './db.js'; +import { createTrainingExample } from './training-data.js'; + +function usage() { + return 'Usage: npm run export -- [limit]\n npm run export -- --limit '; +} + +function parseLimit(args) { + if (args.length === 0) return null; + + let value; + if (args.length === 1 && !args[0].startsWith('-')) value = args[0]; + else if (args.length === 2 && ['--limit', '-n'].includes(args[0])) value = args[1]; + else if (args.length === 1 && args[0].startsWith('--limit=')) value = args[0].slice('--limit='.length); + else throw new Error(usage()); + + const limit = Number(value); + if (!Number.isSafeInteger(limit) || limit <= 0) { + throw new Error(`Limit must be a positive integer, got "${value}"\n${usage()}`); + } + return limit; +} + +let pool; +try { + const limit = parseLimit(process.argv.slice(2)); + pool = createPool(); + await ensureSchema(pool); + let exported = 0; + let cursor = null; + while (limit === null || exported < limit) { + const remaining = limit === null ? 1_000 : limit - exported; + const batchSize = Math.min(1_000, Math.max(100, remaining * 2)); + const params = cursor ? [cursor.cursor_created_at, cursor.id, batchSize] : [batchSize]; + const cursorClause = cursor ? 'AND (created_at, id) < ($1::timestamptz, $2::bigint)' : ''; + const batchParameter = cursor ? '$3' : '$1'; + const result = await pool.query( + `SELECT id, created_at, created_at::text AS cursor_created_at, endpoint, request_body, response_body, + response_status, duration_ms, stream, training_data + FROM requests + WHERE (method = 'POST' OR method IS NULL) + AND response_status BETWEEN 200 AND 299 + AND ( + endpoint LIKE '/oai/v1/chat/completions%' + OR endpoint LIKE '/oai/v1/responses%' + OR endpoint LIKE '/ant/v1/messages%' + OR endpoint LIKE '/ant/v1/v1/messages%' + ) + ${cursorClause} + ORDER BY created_at DESC, id DESC + LIMIT ${batchParameter}`, + params, + ); + + for (const row of result.rows) { + const example = row.training_data || createTrainingExample({ + endpoint: row.endpoint, + requestBody: row.request_body, + responseBody: row.response_body, + responseStatus: row.response_status, + durationMs: row.duration_ms, + stream: row.stream, + }); + if (!example) continue; + const exportedExample = { + ...example, + metadata: { + ...(example.metadata || {}), + request_id: String(row.id), + created_at: row.created_at, + response_status: row.response_status, + ...(row.duration_ms == null ? {} : { duration_ms: row.duration_ms }), + }, + }; + if (!process.stdout.write(`${JSON.stringify(exportedExample)}\n`)) await once(process.stdout, 'drain'); + exported += 1; + if (limit !== null && exported >= limit) break; + } + if (result.rows.length < batchSize || (limit !== null && exported >= limit)) break; + cursor = result.rows.at(-1); + } + console.error(`Exported ${exported} training request(s) as JSONL, newest first`); +} catch (error) { + console.error(`Error: ${error.message}`); + process.exitCode = 1; +} finally { + if (pool) await pool.end(); +} diff --git a/src/proxy.js b/src/proxy.js index b1a94cf..d6d21fd 100644 --- a/src/proxy.js +++ b/src/proxy.js @@ -14,9 +14,71 @@ import { translateOpenAIChunk, } from './anthropic.js'; import { translateResponsesRequest, translateResponsesResponse, translateResponsesChunk } from './responses.js'; +import { createTrainingExample } from './training-data.js'; export const DEFAULT_UPSTREAM_BASE_URL = 'https://opencode.ai/zen/go/v1'; +export function collectRequestData(requestLogger) { + return (request, response, next) => { + const startedAt = process.hrtime.bigint(); + const chunks = []; + const originalWrite = response.write.bind(response); + const originalEnd = response.end.bind(response); + let stored = false; + + const capture = (chunk, encoding) => { + if (chunk === undefined || chunk === null || typeof chunk === 'function') return; + chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk, typeof encoding === 'string' ? encoding : undefined)); + }; + + response.write = (...args) => { + capture(args[0], args[1]); + return originalWrite(...args); + }; + response.end = (...args) => { + capture(args[0], args[1]); + return originalEnd(...args); + }; + + const store = () => { + if (stored) return; + stored = true; + const responseBody = Buffer.concat(chunks).toString('utf8'); + const entry = { + method: request.method, + endpoint: request.originalUrl, + requestHeaders: request.headers, + requestBody: request.body ?? null, + responseHeaders: response.getHeaders(), + responseBody, + responseStatus: response.statusCode, + durationMs: Number(process.hrtime.bigint() - startedAt) / 1_000_000, + model: request.body?.model ?? null, + clientIp: clientIp(request), + stream: Boolean(request.body?.stream), + }; + try { + entry.trainingData = createTrainingExample(entry); + } catch (error) { + console.error(`Failed to normalize request for training: ${error.message}`); + entry.trainingData = null; + } + + try { + Promise.resolve(requestLogger.log(entry)).catch((error) => { + console.error(`Failed to store request: ${error.message}`); + }); + } catch (error) { + console.error(`Failed to store request: ${error.message}`); + } + }; + + response.once('finish', store); + response.once('close', store); + next(); + }; +} + function copyResponseHeaders(upstream, response) { const contentType = upstream.headers.get('content-type'); const cacheControl = upstream.headers.get('cache-control'); @@ -60,9 +122,10 @@ function withContextLimit(payload, maxContextTokens) { }; } -export function createProxyApp({ keys, fetchImpl = globalThis.fetch, upstreamBaseUrl = DEFAULT_UPSTREAM_BASE_URL, maxContextTokens = maxContextFromEnv(), rateLimiter = createRateLimiter(), pagesDirectory = defaultPagesDirectory } = {}) { +export function createProxyApp({ keys, fetchImpl = globalThis.fetch, upstreamBaseUrl = DEFAULT_UPSTREAM_BASE_URL, maxContextTokens = maxContextFromEnv(), rateLimiter = createRateLimiter(), pagesDirectory = defaultPagesDirectory, requestLogger } = {}) { const rotator = createKeyRotator(keys); const app = express(); + if (requestLogger) app.use(collectRequestData(requestLogger)); app.use(express.json({ limit: '10mb' })); app.get('/', (_request, response) => { diff --git a/src/server.js b/src/server.js index dca8036..05fe25e 100644 --- a/src/server.js +++ b/src/server.js @@ -1,13 +1,25 @@ import path from 'node:path'; -import { fileURLToPath } from 'node:url'; import { createProxyApp } from './proxy.js'; +import { projectRoot } from './env.js'; +import { createPool, createRequestLogger, ensureSchema } from './db.js'; import { loadConfiguredKeys } from './key-store.js'; -const projectRoot = path.dirname(path.dirname(fileURLToPath(import.meta.url))); const keys = await loadConfiguredKeys({ path: path.join(projectRoot, 'keys.txt') }); -const app = createProxyApp({ keys }); +const pool = createPool(); +await ensureSchema(pool); +const app = createProxyApp({ keys, requestLogger: createRequestLogger(pool) }); const port = Number(process.env.PORT || 4005); -app.listen(port, '0.0.0.0', () => { +const server = app.listen(port, '0.0.0.0', () => { console.log(`OpenCode Go proxy listening at http://localhost:${port}/oai/v1`); }); + +async function shutdown() { + server.close(async () => { + await pool.end(); + process.exit(0); + }); +} + +process.once('SIGINT', shutdown); +process.once('SIGTERM', shutdown); diff --git a/src/training-data.js b/src/training-data.js new file mode 100644 index 0000000..2a6f137 --- /dev/null +++ b/src/training-data.js @@ -0,0 +1,359 @@ +function parseJson(value) { + if (value && typeof value === 'object') return value; + if (typeof value !== 'string') return null; + try { return JSON.parse(value); } catch { return null; } +} + +function parseSse(body = '') { + const events = []; + for (const block of body.split(/\r?\n\r?\n/)) { + const data = block.split(/\r?\n/) + .filter((line) => line.startsWith('data:')) + .map((line) => line.slice(5).trimStart()) + .join('\n'); + if (!data || data === '[DONE]') continue; + const parsed = parseJson(data); + if (parsed) events.push(parsed); + } + return events; +} + +function normalizedContent(content) { + if (content === undefined || content === null || typeof content === 'string') return content ?? null; + if (!Array.isArray(content)) return String(content); + return content.map((part) => { + if (!part || typeof part !== 'object') return { type: 'text', text: String(part ?? '') }; + if (['input_text', 'output_text'].includes(part.type)) return { type: 'text', text: part.text || '' }; + if (part.type === 'image' && part.source?.type === 'base64') { + return { type: 'image_url', image_url: { url: `data:${part.source.media_type};base64,${part.source.data}` } }; + } + return structuredClone(part); + }); +} + +function toolCall(call = {}) { + const fn = call.function || call; + const args = fn.arguments ?? call.arguments ?? '{}'; + return { + id: call.call_id || call.id || 'tool_call', + type: 'function', + function: { + name: fn.name || call.name || '', + arguments: typeof args === 'string' ? args : JSON.stringify(args), + }, + }; +} + +function normalizedMessage(message = {}) { + const role = message.role === 'developer' ? 'system' : message.role; + if (!['system', 'user', 'assistant', 'tool'].includes(role)) return null; + const result = { role, content: normalizedContent(message.content) }; + if (message.name !== undefined) result.name = message.name; + const reasoning = message.reasoning_content ?? (typeof message.reasoning === 'string' ? message.reasoning : undefined); + if (reasoning) result.reasoning_content = reasoning; + if (message.tool_calls?.length) result.tool_calls = message.tool_calls.map(toolCall); + if (role === 'tool') result.tool_call_id = message.tool_call_id || message.tool_use_id || 'tool_call'; + return result; +} + +function textFromParts(content = []) { + if (typeof content === 'string') return content; + if (!Array.isArray(content)) return ''; + return content + .filter((part) => ['text', 'input_text', 'output_text'].includes(part?.type)) + .map((part) => part.text || '') + .join(''); +} + +function reasoningFromItem(item = {}) { + if (typeof item.reasoning_content === 'string') return item.reasoning_content; + if (typeof item.text === 'string') return item.text; + if (typeof item.content === 'string') return item.content; + return (item.summary || []).map((part) => part?.text || '').join(''); +} + +function assistantFromResponsesOutput(output = []) { + const assistant = { role: 'assistant', content: null }; + let text = ''; + let reasoning = ''; + const calls = []; + + for (const item of output || []) { + if (item?.type === 'message') text += textFromParts(item.content); + else if (item?.type === 'reasoning') reasoning += reasoningFromItem(item); + else if (item?.type === 'function_call') calls.push(toolCall(item)); + } + + if (text) assistant.content = text; + if (reasoning) assistant.reasoning_content = reasoning; + if (calls.length) assistant.tool_calls = calls; + return text || reasoning || calls.length ? assistant : null; +} + +function inputFromResponses(body = {}) { + const messages = []; + if (body.instructions !== undefined) messages.push({ role: 'system', content: normalizedContent(body.instructions) }); + const input = Array.isArray(body.input) ? body.input : [{ type: 'message', role: 'user', content: body.input ?? '' }]; + + const ensureAssistant = () => { + if (messages.at(-1)?.role !== 'assistant') messages.push({ role: 'assistant', content: null }); + return messages.at(-1); + }; + + for (const item of input) { + if (item?.type === 'function_call_output') { + messages.push({ role: 'tool', tool_call_id: item.call_id || 'tool_call', content: normalizedContent(item.output ?? '') }); + } else if (item?.type === 'function_call') { + const assistant = ensureAssistant(); + assistant.tool_calls = assistant.tool_calls || []; + assistant.tool_calls.push(toolCall(item)); + } else if (item?.type === 'reasoning') { + const reasoning = reasoningFromItem(item); + if (reasoning) ensureAssistant().reasoning_content = (ensureAssistant().reasoning_content || '') + reasoning; + } else if (item?.type === 'message' || item?.role) { + const message = normalizedMessage({ ...item, role: item.role || 'user' }); + if (!message) continue; + if (message.role === 'assistant' && messages.at(-1)?.role === 'assistant') { + const pending = messages.at(-1); + pending.content = message.content; + if (message.reasoning_content) pending.reasoning_content = (pending.reasoning_content || '') + message.reasoning_content; + if (message.tool_calls) pending.tool_calls = [...(pending.tool_calls || []), ...message.tool_calls]; + } else messages.push(message); + } + } + return messages; +} + +function assistantFromChatResponse(responseBody, stream) { + if (!stream) { + const payload = parseJson(responseBody); + return normalizedMessage(payload?.choices?.[0]?.message); + } + + let content = ''; + let reasoning = ''; + const calls = new Map(); + for (const payload of parseSse(responseBody)) { + const delta = payload.choices?.[0]?.delta; + if (!delta) continue; + if (typeof delta.content === 'string') content += delta.content; + const reasoningDelta = delta.reasoning_content ?? delta.reasoning; + if (typeof reasoningDelta === 'string') reasoning += reasoningDelta; + for (const call of delta.tool_calls || []) { + const index = call.index ?? calls.size; + const current = calls.get(index) || { id: '', name: '', arguments: '' }; + if (call.id) current.id += call.id; + if (call.function?.name) current.name += call.function.name; + if (call.function?.arguments) current.arguments += call.function.arguments; + calls.set(index, current); + } + } + + if (!content && !reasoning && !calls.size) return null; + const assistant = { role: 'assistant', content: content || null }; + if (reasoning) assistant.reasoning_content = reasoning; + if (calls.size) assistant.tool_calls = [...calls.values()].map((call, index) => toolCall({ + id: call.id || `tool_call_${index}`, function: { name: call.name, arguments: call.arguments || '{}' }, + })); + return assistant; +} + +function assistantFromResponsesResponse(responseBody, stream) { + if (!stream) return assistantFromResponsesOutput(parseJson(responseBody)?.output); + const events = parseSse(responseBody); + const completed = [...events].reverse().find((event) => event.type === 'response.completed'); + if (completed?.response?.output) return assistantFromResponsesOutput(completed.response.output); + + let content = ''; + let reasoning = ''; + const calls = new Map(); + for (const event of events) { + if (event.type === 'response.output_text.delta') content += event.delta || ''; + else if (event.type === 'response.reasoning_summary_text.delta') reasoning += event.delta || ''; + else if (event.type === 'response.output_item.added' && event.item?.type === 'function_call') { + calls.set(event.item.id, { id: event.item.call_id || event.item.id, name: event.item.name || '', arguments: event.item.arguments || '' }); + } else if (event.type === 'response.function_call_arguments.delta') { + const call = calls.get(event.item_id) || { id: event.item_id, name: '', arguments: '' }; + call.arguments += event.delta || ''; + calls.set(event.item_id, call); + } + } + const assistant = { role: 'assistant', content: content || null }; + if (reasoning) assistant.reasoning_content = reasoning; + if (calls.size) assistant.tool_calls = [...calls.values()].map(toolCall); + return content || reasoning || calls.size ? assistant : null; +} + +function inputFromAnthropic(body = {}) { + const messages = []; + if (body.system !== undefined) messages.push({ role: 'system', content: normalizedContent(body.system) }); + for (const source of body.messages || []) { + if (!Array.isArray(source.content)) { + const message = normalizedMessage(source); + if (message) messages.push(message); + continue; + } + + const ordinary = source.content.filter((part) => !['thinking', 'tool_use', 'tool_result'].includes(part?.type)); + const thinking = source.content.filter((part) => part?.type === 'thinking').map((part) => part.thinking || '').join(''); + const calls = source.content.filter((part) => part?.type === 'tool_use').map((part) => toolCall({ id: part.id, name: part.name, arguments: part.input })); + if (ordinary.length || thinking || calls.length) { + const message = { role: source.role === 'developer' ? 'system' : source.role, content: ordinary.length ? normalizedContent(ordinary) : null }; + if (thinking) message.reasoning_content = thinking; + if (calls.length) message.tool_calls = calls; + messages.push(message); + } + for (const result of source.content.filter((part) => part?.type === 'tool_result')) { + messages.push({ role: 'tool', tool_call_id: result.tool_use_id || 'tool_call', content: normalizedContent(result.content ?? '') }); + } + } + return messages; +} + +function assistantFromAnthropicResponse(responseBody, stream) { + if (!stream) { + const payload = parseJson(responseBody); + const assistant = { role: 'assistant', content: null }; + let text = ''; + let reasoning = ''; + const calls = []; + for (const block of payload?.content || []) { + if (block.type === 'text') text += block.text || ''; + else if (block.type === 'thinking') reasoning += block.thinking || ''; + else if (block.type === 'tool_use') calls.push(toolCall({ id: block.id, name: block.name, arguments: block.input })); + } + if (text) assistant.content = text; + if (reasoning) assistant.reasoning_content = reasoning; + if (calls.length) assistant.tool_calls = calls; + return text || reasoning || calls.length ? assistant : null; + } + + const blocks = new Map(); + for (const event of parseSse(responseBody)) { + if (event.type === 'content_block_start') { + const block = event.content_block || {}; + blocks.set(event.index, { type: block.type, text: block.text || '', thinking: block.thinking || '', id: block.id, name: block.name, arguments: '' }); + } else if (event.type === 'content_block_delta') { + const block = blocks.get(event.index) || {}; + if (event.delta?.type === 'text_delta') block.text = (block.text || '') + (event.delta.text || ''); + else if (event.delta?.type === 'thinking_delta') block.thinking = (block.thinking || '') + (event.delta.thinking || ''); + else if (event.delta?.type === 'input_json_delta') block.arguments = (block.arguments || '') + (event.delta.partial_json || ''); + blocks.set(event.index, block); + } + } + return assistantFromAnthropicResponse(JSON.stringify({ content: [...blocks.values()].map((block) => { + if (block.type === 'tool_use') return { type: block.type, id: block.id, name: block.name, input: parseJson(block.arguments) || block.arguments || {} }; + return block; + }) }), false); +} + +function definedEntries(object) { + return Object.fromEntries(Object.entries(object).filter(([, value]) => value !== undefined && value !== null)); +} + +function responseMetadata(pathname, responseBody, stream) { + if (pathname === '/oai/v1/chat/completions') { + const payloads = stream ? parseSse(responseBody) : [parseJson(responseBody)].filter(Boolean); + const idPayload = payloads.find((payload) => payload.id) || {}; + const modelPayload = payloads.find((payload) => payload.model) || {}; + const finishPayload = [...payloads].reverse().find((payload) => payload.choices?.[0]?.finish_reason) || {}; + const usagePayload = [...payloads].reverse().find((payload) => payload.usage) || {}; + return definedEntries({ + id: idPayload.id, + model: modelPayload.model, + finish_reason: finishPayload.choices?.[0]?.finish_reason, + usage: usagePayload.usage, + cost: usagePayload.cost ?? usagePayload.usage?.cost, + }); + } + + if (pathname === '/oai/v1/responses') { + const payload = stream + ? [...parseSse(responseBody)].reverse().find((event) => event.type === 'response.completed')?.response + : parseJson(responseBody); + return definedEntries({ + id: payload?.id, + model: payload?.model, + status: payload?.status, + incomplete_details: payload?.incomplete_details, + usage: payload?.usage, + }); + } + + const payloads = stream ? parseSse(responseBody) : [parseJson(responseBody)].filter(Boolean); + if (!stream) { + const payload = payloads[0] || {}; + return definedEntries({ + id: payload.id, + model: payload.model, + finish_reason: payload.stop_reason, + stop_sequence: payload.stop_sequence, + usage: payload.usage, + }); + } + const start = payloads.find((payload) => payload.type === 'message_start')?.message || {}; + const delta = [...payloads].reverse().find((payload) => payload.type === 'message_delta') || {}; + const usage = definedEntries({ ...(start.usage || {}), ...(delta.usage || {}) }); + return definedEntries({ + id: start.id, + model: start.model, + finish_reason: delta.delta?.stop_reason, + stop_sequence: delta.delta?.stop_sequence, + usage: Object.keys(usage).length ? usage : undefined, + }); +} + +function trainingMetadata(pathname, requestBody, responseBody, stream, responseStatus, durationMs) { + const api = pathname === '/oai/v1/chat/completions' + ? 'openai_chat_completions' + : pathname === '/oai/v1/responses' + ? 'openai_responses' + : 'anthropic_messages'; + const parameterNames = [ + 'temperature', 'top_p', 'top_k', 'min_p', 'max_tokens', 'max_output_tokens', + 'stop', 'stop_sequences', 'frequency_penalty', 'presence_penalty', 'n', 'seed', + 'logprobs', 'top_logprobs', 'response_format', 'parallel_tool_calls', 'reasoning', + 'thinking', 'verbosity', 'text', 'include', 'truncation', 'modalities', 'audio', + 'service_tier', + ]; + const parameters = definedEntries(Object.fromEntries(parameterNames.map((name) => [name, requestBody[name]]))); + const metadata = { + schema_version: 1, + api, + endpoint: pathname, + model: requestBody.model ?? null, + stream: Boolean(stream), + tools: structuredClone(requestBody.tools || []), + response_status: responseStatus, + response: responseMetadata(pathname, responseBody, stream), + }; + if (Number.isFinite(durationMs)) metadata.duration_ms = durationMs; + if (requestBody.tool_choice !== undefined) metadata.tool_choice = structuredClone(requestBody.tool_choice); + if (requestBody.metadata !== undefined) metadata.request_metadata = structuredClone(requestBody.metadata); + if (Object.keys(parameters).length) metadata.parameters = parameters; + return metadata; +} + +export function createTrainingExample({ endpoint = '', requestBody, responseBody, responseStatus, durationMs, stream = false } = {}) { + if (responseStatus < 200 || responseStatus >= 300 || !requestBody) return null; + const pathname = endpoint.split('?')[0]; + let messages; + let assistant; + + if (pathname === '/oai/v1/chat/completions') { + messages = (requestBody.messages || []).map(normalizedMessage).filter(Boolean); + assistant = assistantFromChatResponse(responseBody, stream); + } else if (pathname === '/oai/v1/responses') { + messages = inputFromResponses(requestBody); + assistant = assistantFromResponsesResponse(responseBody, stream); + } else if (['/ant/v1/messages', '/ant/v1/v1/messages'].includes(pathname)) { + messages = inputFromAnthropic(requestBody); + assistant = assistantFromAnthropicResponse(responseBody, stream); + } else return null; + + if (!assistant || !messages.length) return null; + return { + messages: [...messages, assistant], + metadata: trainingMetadata(pathname, requestBody, responseBody, stream, responseStatus, durationMs), + }; +} diff --git a/test/db.test.js b/test/db.test.js new file mode 100644 index 0000000..4f9ee5f --- /dev/null +++ b/test/db.test.js @@ -0,0 +1,47 @@ +import { describe, expect, it, vi } from 'vitest'; +import { createRequestLogger, ensureSchema, insertRequest } from '../src/db.js'; + +describe('request database', () => { + it('creates the requests table, compatible columns, and recent-request index', async () => { + const pool = { query: vi.fn().mockResolvedValue({}) }; + + await ensureSchema(pool); + + expect(pool.query).toHaveBeenCalledTimes(3); + expect(pool.query.mock.calls[0][0]).toContain('CREATE TABLE IF NOT EXISTS requests'); + expect(pool.query.mock.calls[1][0]).toContain('ADD COLUMN IF NOT EXISTS request_headers'); + expect(pool.query.mock.calls[1][0]).toContain('ADD COLUMN IF NOT EXISTS duration_ms'); + expect(pool.query.mock.calls[1][0]).toContain('ADD COLUMN IF NOT EXISTS training_data'); + expect(pool.query.mock.calls[2][0]).toContain('created_at DESC, id DESC'); + }); + + it('stores request and response data as one row', async () => { + const pool = { query: vi.fn().mockResolvedValue({}) }; + const entry = { + method: 'POST', endpoint: '/oai/v1/responses', + requestHeaders: { host: 'proxy.test' }, requestBody: { model: 'm', input: 'Hi' }, + responseHeaders: { 'content-type': 'application/json' }, responseBody: '{"id":"r1"}', + responseStatus: 200, durationMs: 123.5, model: 'm', clientIp: '198.51.100.2', stream: false, + trainingData: { messages: [{ role: 'user', content: 'Hi' }, { role: 'assistant', content: 'Hello' }] }, + }; + + await insertRequest(pool, entry); + + expect(pool.query).toHaveBeenCalledOnce(); + expect(pool.query.mock.calls[0][0]).toContain('INSERT INTO requests'); + expect(pool.query.mock.calls[0][1]).toEqual([ + 'POST', '/oai/v1/responses', '{"host":"proxy.test"}', + '{"model":"m","input":"Hi"}', '{"content-type":"application/json"}', + '{"id":"r1"}', 200, 123.5, 'm', '198.51.100.2', false, + '{"messages":[{"role":"user","content":"Hi"},{"role":"assistant","content":"Hello"}]}', + ]); + }); + + it('reports storage failures without rejecting request handling', async () => { + const onError = vi.fn(); + const logger = createRequestLogger({ query: vi.fn().mockRejectedValue(new Error('database offline')) }, { onError }); + + await expect(logger.log({ method: 'GET', endpoint: '/', responseStatus: 200 })).resolves.toBeUndefined(); + expect(onError).toHaveBeenCalledWith('Failed to store request: database offline'); + }); +}); diff --git a/test/env.test.js b/test/env.test.js new file mode 100644 index 0000000..80f1162 --- /dev/null +++ b/test/env.test.js @@ -0,0 +1,30 @@ +import fs from 'node:fs/promises'; +import os from 'node:os'; +import path from 'node:path'; +import { afterEach, describe, expect, it } from 'vitest'; +import { loadEnv } from '../src/env.js'; + +const loadedKey = 'OPENCODE_PROXY_TEST_ENV_LOADED'; +const existingKey = 'OPENCODE_PROXY_TEST_ENV_EXISTING'; + +afterEach(() => { + delete process.env[loadedKey]; + delete process.env[existingKey]; +}); + +describe('.env loading', () => { + it('loads values without overriding the process environment', async () => { + const directory = await fs.mkdtemp(path.join(os.tmpdir(), 'opencode-proxy-env-')); + const envPath = path.join(directory, '.env'); + await fs.writeFile(envPath, `${loadedKey}=from-file\n${existingKey}=from-file\n`); + process.env[existingKey] = 'from-process'; + + try { + loadEnv(envPath); + expect(process.env[loadedKey]).toBe('from-file'); + expect(process.env[existingKey]).toBe('from-process'); + } finally { + await fs.rm(directory, { recursive: true, force: true }); + } + }); +}); diff --git a/test/proxy.test.js b/test/proxy.test.js index 8de905f..262d218 100644 --- a/test/proxy.test.js +++ b/test/proxy.test.js @@ -87,6 +87,58 @@ describe('proxy', () => { expect(fetchImpl.mock.calls[0][1].headers.accept).toBe('text/event-stream'); }); + it('collects the request, chat history, and complete streamed response', async () => { + const stream = 'data: {"choices":[{"delta":{"reasoning_content":"Think. ","content":"Hi"}}]}\n\ndata: {"choices":[{"delta":{},"finish_reason":"stop"}]}\n\ndata: [DONE]\n\n'; + const fetchImpl = vi.fn().mockResolvedValue(response(stream, { + headers: { 'content-type': 'text/event-stream' }, + })); + const requestLogger = { log: vi.fn() }; + const app = createProxyApp({ keys: ['key'], fetchImpl, requestLogger }); + const body = { model: 'm', messages: [{ role: 'user', content: 'Hi' }], stream: true }; + + await request(app) + .post('/oai/v1/chat/completions?source=test') + .set('X-Real-IP', '198.51.100.10') + .send(body) + .expect(200); + + expect(requestLogger.log).toHaveBeenCalledTimes(1); + expect(requestLogger.log.mock.calls[0][0]).toMatchObject({ + method: 'POST', + endpoint: '/oai/v1/chat/completions?source=test', + requestBody: body, + responseBody: stream, + responseStatus: 200, + model: 'm', + clientIp: '198.51.100.10', + stream: true, + trainingData: { messages: [ + { role: 'user', content: 'Hi' }, + { role: 'assistant', content: 'Hi', reasoning_content: 'Think. ' }, + ] }, + }); + expect(requestLogger.log.mock.calls[0][0].requestHeaders['content-type']).toMatch('application/json'); + expect(requestLogger.log.mock.calls[0][0].responseHeaders['content-type']).toMatch('text/event-stream'); + expect(requestLogger.log.mock.calls[0][0].durationMs).toBeGreaterThanOrEqual(0); + expect(requestLogger.log.mock.calls[0][0].trainingData.metadata).toMatchObject({ + schema_version: 1, api: 'openai_chat_completions', model: 'm', stream: true, + tools: [], response_status: 200, response: { finish_reason: 'stop' }, + }); + }); + + it('collects error and not-found responses too', async () => { + const requestLogger = { log: vi.fn() }; + const app = createProxyApp({ keys: ['key'], requestLogger }); + + await request(app).get('/missing').expect(404); + + expect(requestLogger.log).toHaveBeenCalledOnce(); + expect(requestLogger.log.mock.calls[0][0]).toMatchObject({ + method: 'GET', endpoint: '/missing', requestBody: null, responseStatus: 404, + }); + expect(requestLogger.log.mock.calls[0][0].responseBody).toContain('Cannot GET /missing'); + }); + it('translates Responses requests and returns a Responses object', async () => { const fetchImpl = vi.fn().mockResolvedValue(response(JSON.stringify({ id: 'chatcmpl-response', model: 'm', diff --git a/test/training-data.test.js b/test/training-data.test.js new file mode 100644 index 0000000..0a181cc --- /dev/null +++ b/test/training-data.test.js @@ -0,0 +1,168 @@ +import { describe, expect, it } from 'vitest'; +import { createTrainingExample } from '../src/training-data.js'; + +describe('training data normalization', () => { + it('exports Chat Completions histories, reasoning, tool calls, and tool results', () => { + const example = createTrainingExample({ + endpoint: '/oai/v1/chat/completions', responseStatus: 200, + requestBody: { + model: 'model-a', temperature: 0.2, max_tokens: 200, tool_choice: 'auto', metadata: { dataset: 'test' }, + tools: [{ type: 'function', function: { name: 'weather', description: 'Get weather', parameters: { type: 'object' } } }], + messages: [ + { role: 'system', content: 'Be useful' }, + { role: 'user', content: 'Find the weather' }, + { role: 'assistant', content: null, tool_calls: [{ id: 'call_1', type: 'function', function: { name: 'weather', arguments: '{"city":"Oslo"}' } }] }, + { role: 'tool', tool_call_id: 'call_1', content: '{"temp":12}' }, + { role: 'user', content: 'Summarize it' }, + ], + }, + responseBody: JSON.stringify({ id: 'chatcmpl_1', model: 'model-a', usage: { prompt_tokens: 20, completion_tokens: 8 }, choices: [{ finish_reason: 'tool_calls', message: { + role: 'assistant', reasoning_content: 'The result says 12 degrees.', content: 'It is 12°C in Oslo.', + tool_calls: [{ id: 'call_2', type: 'function', function: { name: 'log', arguments: '{"temp":12}' } }], + } }] }), + durationMs: 123.5, + }); + + expect(example.messages).toHaveLength(6); + expect(example.messages[2].tool_calls[0]).toMatchObject({ id: 'call_1', function: { name: 'weather' } }); + expect(example.messages[3]).toEqual({ role: 'tool', tool_call_id: 'call_1', content: '{"temp":12}' }); + expect(example.messages.at(-1)).toEqual({ + role: 'assistant', content: 'It is 12°C in Oslo.', reasoning_content: 'The result says 12 degrees.', + tool_calls: [{ id: 'call_2', type: 'function', function: { name: 'log', arguments: '{"temp":12}' } }], + }); + expect(example.metadata).toEqual({ + schema_version: 1, + api: 'openai_chat_completions', endpoint: '/oai/v1/chat/completions', model: 'model-a', stream: false, + tools: [{ type: 'function', function: { name: 'weather', description: 'Get weather', parameters: { type: 'object' } } }], + tool_choice: 'auto', request_metadata: { dataset: 'test' }, parameters: { temperature: 0.2, max_tokens: 200 }, response_status: 200, duration_ms: 123.5, + response: { id: 'chatcmpl_1', model: 'model-a', finish_reason: 'tool_calls', usage: { prompt_tokens: 20, completion_tokens: 8 } }, + }); + }); + + it('reconstructs streamed Chat Completions output', () => { + const responseBody = [ + 'data: {"choices":[{"delta":{"reasoning_content":"check "}}]}', + 'data: {"choices":[{"delta":{"content":"Done","tool_calls":[{"index":0,"id":"call_1","function":{"name":"save","arguments":"{\\"ok\\":"}}]}}]}', + 'data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"true}"}}]},"finish_reason":"tool_calls"}]}', + 'data: [DONE]', '', + ].join('\n\n'); + const example = createTrainingExample({ + endpoint: '/oai/v1/chat/completions', responseStatus: 200, stream: true, + requestBody: { messages: [{ role: 'user', content: 'Do it' }] }, responseBody, + }); + + expect(example.messages.at(-1)).toEqual({ + role: 'assistant', content: 'Done', reasoning_content: 'check ', + tool_calls: [{ id: 'call_1', type: 'function', function: { name: 'save', arguments: '{"ok":true}' } }], + }); + }); + + it('normalizes Responses input and output items', () => { + const example = createTrainingExample({ + endpoint: '/oai/v1/responses', responseStatus: 200, + requestBody: { + model: 'model-r', instructions: 'Use tools', + tools: [{ type: 'function', name: 'read', parameters: { type: 'object' } }, { type: 'web_search_preview' }], + input: [ + { type: 'message', role: 'user', content: [{ type: 'input_text', text: 'Read file' }] }, + { type: 'reasoning', summary: [{ type: 'summary_text', text: 'Need the file.' }] }, + { type: 'function_call', call_id: 'call_read', name: 'read', arguments: '{"path":"a.txt"}' }, + { type: 'function_call_output', call_id: 'call_read', output: 'hello' }, + ], + }, + responseBody: JSON.stringify({ output: [ + { type: 'reasoning', summary: [{ type: 'summary_text', text: 'The file says hello.' }] }, + { type: 'message', role: 'assistant', content: [{ type: 'output_text', text: 'It says hello.' }] }, + { type: 'function_call', call_id: 'call_log', name: 'log', arguments: '{"value":"hello"}' }, + ] }), + }); + + expect(example.messages).toEqual([ + { role: 'system', content: 'Use tools' }, + { role: 'user', content: [{ type: 'text', text: 'Read file' }] }, + { role: 'assistant', content: null, reasoning_content: 'Need the file.', tool_calls: [{ id: 'call_read', type: 'function', function: { name: 'read', arguments: '{"path":"a.txt"}' } }] }, + { role: 'tool', tool_call_id: 'call_read', content: 'hello' }, + { role: 'assistant', content: 'It says hello.', reasoning_content: 'The file says hello.', tool_calls: [{ id: 'call_log', type: 'function', function: { name: 'log', arguments: '{"value":"hello"}' } }] }, + ]); + expect(example.metadata.tools).toEqual([ + { type: 'function', name: 'read', parameters: { type: 'object' } }, + { type: 'web_search_preview' }, + ]); + expect(example.metadata).toMatchObject({ api: 'openai_responses', model: 'model-r', stream: false, response_status: 200 }); + }); + + it('uses the completed Responses streaming event as the assistant output', () => { + const responseBody = [ + 'event: response.completed', + 'data: {"type":"response.completed","response":{"output":[{"type":"reasoning","summary":[{"type":"summary_text","text":"Need a lookup."}]},{"type":"message","role":"assistant","content":[{"type":"output_text","text":"Found it."}]},{"type":"function_call","call_id":"call_1","name":"lookup","arguments":"{\\"q\\":\\"x\\"}"}]}}', + '', '', + ].join('\n'); + const example = createTrainingExample({ + endpoint: '/oai/v1/responses', responseStatus: 200, stream: true, + requestBody: { input: 'Find x' }, responseBody, + }); + + expect(example.messages.at(-1)).toEqual({ + role: 'assistant', content: 'Found it.', reasoning_content: 'Need a lookup.', + tool_calls: [{ id: 'call_1', type: 'function', function: { name: 'lookup', arguments: '{"q":"x"}' } }], + }); + }); + + it('normalizes Anthropic thinking, tool use, and tool results', () => { + const example = createTrainingExample({ + endpoint: '/ant/v1/messages', responseStatus: 200, + requestBody: { + model: 'model-ant', system: 'Be useful', + tools: [{ name: 'calculator', description: 'Calculate', input_schema: { type: 'object' } }], + messages: [ + { role: 'user', content: 'Calculate' }, + { role: 'assistant', content: [{ type: 'thinking', thinking: 'Need calculator.' }, { type: 'tool_use', id: 'tool_1', name: 'calculator', input: { expression: '2+2' } }] }, + { role: 'user', content: [{ type: 'tool_result', tool_use_id: 'tool_1', content: '4' }] }, + ], + }, + responseBody: JSON.stringify({ content: [ + { type: 'thinking', thinking: 'The result is four.' }, + { type: 'text', text: 'The answer is 4.' }, + { type: 'tool_use', id: 'tool_2', name: 'record', input: { answer: 4 } }, + ] }), + }); + + expect(example.messages).toEqual([ + { role: 'system', content: 'Be useful' }, + { role: 'user', content: 'Calculate' }, + { role: 'assistant', content: null, reasoning_content: 'Need calculator.', tool_calls: [{ id: 'tool_1', type: 'function', function: { name: 'calculator', arguments: '{"expression":"2+2"}' } }] }, + { role: 'tool', tool_call_id: 'tool_1', content: '4' }, + { role: 'assistant', content: 'The answer is 4.', reasoning_content: 'The result is four.', tool_calls: [{ id: 'tool_2', type: 'function', function: { name: 'record', arguments: '{"answer":4}' } }] }, + ]); + expect(example.metadata.tools).toEqual([ + { name: 'calculator', description: 'Calculate', input_schema: { type: 'object' } }, + ]); + expect(example.metadata).toMatchObject({ api: 'anthropic_messages', model: 'model-ant', stream: false, response_status: 200 }); + }); + + it('reconstructs streamed Anthropic content blocks', () => { + const responseBody = [ + 'event: content_block_start\ndata: {"type":"content_block_start","index":0,"content_block":{"type":"thinking","thinking":""}}', + 'event: content_block_delta\ndata: {"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"Need tool."}}', + 'event: content_block_start\ndata: {"type":"content_block_start","index":1,"content_block":{"type":"text","text":""}}', + 'event: content_block_delta\ndata: {"type":"content_block_delta","index":1,"delta":{"type":"text_delta","text":"Working"}}', + 'event: content_block_start\ndata: {"type":"content_block_start","index":2,"content_block":{"type":"tool_use","id":"tool_1","name":"run","input":{}}}', + 'event: content_block_delta\ndata: {"type":"content_block_delta","index":2,"delta":{"type":"input_json_delta","partial_json":"{\\"x\\":1}"}}', + '', + ].join('\n\n'); + const example = createTrainingExample({ + endpoint: '/ant/v1/messages', responseStatus: 200, stream: true, + requestBody: { messages: [{ role: 'user', content: 'Work' }] }, responseBody, + }); + + expect(example.messages.at(-1)).toEqual({ + role: 'assistant', content: 'Working', reasoning_content: 'Need tool.', + tool_calls: [{ id: 'tool_1', type: 'function', function: { name: 'run', arguments: '{"x":1}' } }], + }); + }); + + it('excludes failed and non-chat requests', () => { + expect(createTrainingExample({ endpoint: '/oai/v1/models', responseStatus: 200, requestBody: {} })).toBeNull(); + expect(createTrainingExample({ endpoint: '/oai/v1/chat/completions', responseStatus: 500, requestBody: { messages: [] } })).toBeNull(); + }); +});