Spaces:
Paused
Paused
| /** Metered, provider-neutral completion transport. No environment loading or funding fallback. */ | |
| import fs from 'node:fs'; | |
| import path from 'node:path'; | |
| import {createHash, randomUUID} from 'node:crypto'; | |
| import type {ForecastCompletion} from './coherence.ts'; | |
| export type Rates = {input: number; output: number}; // USD per million tokens | |
| export type CompletionTrace = { | |
| id: string; model: string; stage: string; started_at: string; finished_at?: string; | |
| status: 'reserved' | 'complete' | 'error'; input: {system: string; user: string}; | |
| max_tokens: number; reservation_usd: number; accounted_usd: number; | |
| response?: string; error?: string; http_status?: number; finish_reason?: string; | |
| usage?: {prompt_tokens: number; completion_tokens: number; cost?: number}; | |
| }; | |
| type Ledger = {schema_version: 1; spent_usd: number; calls: CompletionTrace[]}; | |
| export class RuntimeError extends Error { | |
| readonly kind: 'budget' | 'transport' | 'provider' | 'output' | 'configuration'; | |
| constructor(message: string, kind: RuntimeError['kind']) { | |
| super(message); this.name = 'RuntimeError'; this.kind = kind; | |
| } | |
| } | |
| export const sha256 = (value: string | Buffer) => createHash('sha256').update(value).digest('hex'); | |
| const finite = (n: unknown): n is number => typeof n === 'number' && Number.isFinite(n) && n >= 0; | |
| export function atomicJson(filename: string, value: unknown): void { | |
| fs.mkdirSync(path.dirname(path.resolve(filename)), {recursive: true}); | |
| const temp = `${filename}.${randomUUID()}.tmp`; | |
| try { fs.writeFileSync(temp, JSON.stringify(value, null, 2) + '\n', {mode: 0o600, flag: 'wx'}); fs.renameSync(temp, filename); } | |
| finally { if (fs.existsSync(temp)) fs.unlinkSync(temp); } | |
| } | |
| export type RuntimeOptions = { | |
| model: string; apiKey: string; endpoint?: string; rates: Rates; | |
| maxCost: number; ledger: string; timeoutMs?: number; idleTimeoutMs?: number; | |
| reasoning?: boolean; maxOutputTokens?: number; | |
| /** Dependency injection for offline tests or an explicitly configured local provider. */ | |
| fetch?: typeof fetch; | |
| }; | |
| /** Read-only discovery. Price caps remain explicit; discovery is not spending authorization. */ | |
| export async function openRouterModel(model: string, request = fetch, options: {textOnly?: boolean} = {}): Promise<{rates: Rates; reasoning: boolean; context: number}> { | |
| const response = await request('https://openrouter.ai/api/v1/models', {signal: AbortSignal.timeout(20000)}); | |
| if (!response.ok) throw new RuntimeError(`Model catalog HTTP ${response.status}`, 'configuration'); | |
| const data = await response.json() as {data?: Array<Record<string, any>>}; | |
| const item = data.data?.find(x => x.id === model); | |
| if (!item) throw new RuntimeError(`Model is absent from the provider catalog: ${model}`, 'configuration'); | |
| const prices = item.pricing ?? {}; | |
| let input = Number(prices.prompt) * 1e6, output = Number(prices.completion) * 1e6; | |
| if (!finite(input) || !finite(output)) throw new RuntimeError('Model prices are unavailable', 'configuration'); | |
| // Reserve the highest time/context tier, even when catalog discovery happens off-peak. | |
| for (const tier of prices.overrides ?? []) { | |
| const tierInput = Number(tier.prompt ?? prices.prompt) * 1e6; | |
| const tierOutput = Number(tier.completion ?? prices.completion) * 1e6; | |
| if (!finite(tierInput) || !finite(tierOutput)) throw new RuntimeError('Invalid tier prices', 'configuration'); | |
| input = Math.max(input, tierInput); output = Math.max(output, tierOutput); | |
| } | |
| const tokenFields = ['prompt', 'completion', 'input_cache_read', 'input_cache_write', 'input_cache_write_1h', 'discount', 'overrides']; | |
| // These features are never sent by the text-only Space (no tools, images, audio or plugins). | |
| const optionalFeatures = ['web_search', 'image', 'audio', 'input_audio_cache', 'image_output', 'audio_output']; | |
| for (const [name, amount] of Object.entries(prices)) { | |
| if (tokenFields.includes(name) || (options.textOnly && optionalFeatures.includes(name))) continue; | |
| if (options.textOnly && name === 'internal_reasoning') { | |
| const rate = Number(amount) * 1e6; | |
| if (!finite(rate)) throw new RuntimeError('Invalid reasoning price', 'configuration'); | |
| output = Math.max(output, rate); continue; | |
| } | |
| if (Number(amount) > 0) throw new RuntimeError(`Unsupported additional model charge: ${name}`, 'configuration'); | |
| } | |
| return {rates: {input, output}, reasoning: item.supported_parameters?.includes('reasoning') === true, | |
| context: Number(item.context_length) || 0}; | |
| } | |
| export class MeteredRuntime { | |
| private readonly options: RuntimeOptions; | |
| private ledger: Ledger; | |
| private lock: number; | |
| private closed = false; | |
| private busy = false; | |
| private stopped = false; | |
| private readonly request: typeof fetch; | |
| private readonly url: URL; | |
| readonly traceStart: number; | |
| constructor(options: RuntimeOptions) { | |
| this.options = options; | |
| if (!options.model.trim() || !options.apiKey.trim()) throw new RuntimeError('A model and personal API key are required', 'configuration'); | |
| if (!finite(options.maxCost) || options.maxCost <= 0 || !finite(options.rates.input) || !finite(options.rates.output)) | |
| throw new RuntimeError('Set a positive cumulative spending ceiling and nonnegative token rates', 'configuration'); | |
| this.url = new URL(options.endpoint ?? 'https://openrouter.ai/api/v1/chat/completions'); | |
| if (this.url.username || this.url.password || (this.url.protocol !== 'https:' && !(this.url.protocol === 'http:' && ['localhost', '127.0.0.1', '[::1]'].includes(this.url.hostname)))) | |
| throw new RuntimeError('Provider URL must use HTTPS (HTTP allowed only for an explicit local provider)', 'configuration'); | |
| this.request = options.fetch ?? fetch; | |
| fs.mkdirSync(path.dirname(path.resolve(options.ledger)), {recursive: true}); | |
| this.lock = fs.openSync(options.ledger + '.lock', 'wx', 0o600); | |
| try { | |
| this.ledger = fs.existsSync(options.ledger) ? JSON.parse(fs.readFileSync(options.ledger, 'utf8')) : {schema_version: 1, spent_usd: 0, calls: []}; | |
| const l = this.ledger; | |
| if (l.schema_version !== 1 || !finite(l.spent_usd) || !Array.isArray(l.calls) || l.calls.some(c => !finite(c.accounted_usd)) || | |
| Math.abs(l.calls.reduce((sum, c) => sum + c.accounted_usd, 0) - l.spent_usd) > 1e-8) | |
| throw new RuntimeError('Invalid budget ledger; do not reset it to recover allowance', 'configuration'); | |
| this.traceStart = l.calls.length; | |
| this.save(); | |
| } catch (error) { fs.closeSync(this.lock); fs.unlinkSync(options.ledger + '.lock'); throw error; } | |
| } | |
| private save() { atomicJson(this.options.ledger, this.ledger); } | |
| get spent(): number { return this.ledger.spent_usd; } | |
| get traces(): CompletionTrace[] { return structuredClone(this.ledger.calls.slice(this.traceStart)); } | |
| get terminal(): boolean { return this.stopped; } | |
| close(): void { | |
| if (this.busy) throw new RuntimeError('Cannot close during an active completion', 'configuration'); | |
| if (!this.closed) { this.closed = true; fs.closeSync(this.lock); fs.unlinkSync(this.options.ledger + '.lock'); } | |
| } | |
| readonly complete: ForecastCompletion = async (system, user, stage, requested) => { | |
| if (this.closed || this.busy || this.stopped) throw new RuntimeError('Runtime closed, busy, or stopped after an uncertain charge', 'configuration'); | |
| // Reasoning consumes the same output allowance; a tiny review cap can starve the answer. | |
| const max = Math.min(this.options.reasoning ? Math.max(requested,8400) : requested, this.options.maxOutputTokens ?? 8400); | |
| if (!Number.isInteger(max) || max <= 0) throw new RuntimeError('Invalid output-token ceiling', 'configuration'); | |
| const messages = [{role: 'system', content: system}, {role: 'user', content: user}]; | |
| const bytes = Buffer.byteLength(JSON.stringify(messages), 'utf8'); | |
| if (bytes > 300000) throw new RuntimeError('Input exceeds the executor message limit', 'configuration'); | |
| // A byte bound plus framing allowance deliberately over-reserves ordinary text tokenization. | |
| const reserve = ((bytes + 4096) * this.options.rates.input + max * this.options.rates.output) / 1e6; | |
| if (this.spent + reserve > this.options.maxCost + 1e-12) throw new RuntimeError('Cumulative spending ceiling reached before request', 'budget'); | |
| const trace: CompletionTrace = {id: randomUUID(), model: this.options.model, stage, started_at: new Date().toISOString(), | |
| input: {system, user}, max_tokens: max, reservation_usd: reserve, accounted_usd: reserve, status: 'reserved'}; | |
| this.ledger.calls.push(trace); this.ledger.spent_usd += reserve; this.save(); | |
| this.busy = true; | |
| const controller = new AbortController(); | |
| const totalTimer = setTimeout(() => controller.abort(new Error('Total completion timeout')), this.options.timeoutMs ?? 240000); | |
| let idleTimer: ReturnType<typeof setTimeout>; | |
| const touch = () => { clearTimeout(idleTimer); idleTimer = setTimeout(() => controller.abort(new Error('Completion stream idle timeout')), this.options.idleTimeoutMs ?? 60000); }; | |
| touch(); | |
| let bodyComplete = false; | |
| let reportedUsage: any; | |
| let output = ''; | |
| const consume = (data: any) => { | |
| if (data.error) throw new RuntimeError(`Provider error: ${JSON.stringify(data.error).slice(0, 1600)}`, 'provider'); | |
| if (data.usage) reportedUsage = data.usage; | |
| const choice = data.choices?.[0]; | |
| if (typeof choice?.delta?.content === 'string') output += choice.delta.content; | |
| if (typeof choice?.message?.content === 'string') output = choice.message.content; | |
| if (choice?.finish_reason) trace.finish_reason = choice.finish_reason; | |
| if (output.length > 2_000_000) throw new RuntimeError('Completion exceeded response size limit', 'output'); | |
| }; | |
| try { | |
| const body: Record<string, unknown> = {model: this.options.model, messages, max_tokens: max, stream: true, stream_options: {include_usage: true}}; | |
| if (this.url.hostname === 'openrouter.ai') { | |
| body.provider = {allow_fallbacks: false, require_parameters: true, | |
| max_price: {prompt: this.options.rates.input, completion: this.options.rates.output}}; | |
| // Never disable a model's required reasoning on contract/review stages. | |
| if (this.options.reasoning) body.reasoning = {enabled: true}; | |
| } | |
| const response = await this.request(this.url, {method: 'POST', signal: controller.signal, | |
| headers: {Authorization: `Bearer ${this.options.apiKey}`, 'Content-Type': 'application/json'}, body: JSON.stringify(body)}); | |
| trace.http_status = response.status; | |
| touch(); | |
| const readBounded=async()=>{ | |
| if(!response.body)throw new RuntimeError('Provider returned no body','transport'); | |
| const reader=response.body.getReader(),chunks:Uint8Array[]=[];let size=0; | |
| try{for(;;){const {done,value}=await reader.read();if(done)break;touch();size+=value.length;if(size>2_000_000)throw new RuntimeError('Provider body too large','transport');chunks.push(value);}} | |
| finally{await reader.cancel().catch(()=>{});reader.releaseLock();} | |
| return Buffer.concat(chunks).toString('utf8'); | |
| }; | |
| if (!response.ok) { | |
| const text = (await readBounded()).slice(0, 3000); bodyComplete = true; | |
| throw new RuntimeError(`Provider HTTP ${response.status}: ${text}`, 'provider'); | |
| } | |
| if (response.headers.get('content-type')?.includes('text/event-stream')) { | |
| if (!response.body) throw new RuntimeError('Provider returned no response stream', 'transport'); | |
| const reader = response.body.getReader(), decoder = new TextDecoder(); | |
| let buffer = '', doneMarker = false; | |
| const processEvent = (event: string) => { | |
| const payload = event.split('\n').filter(line => line.startsWith('data:')).map(line => line.slice(5).trim()).join('\n'); | |
| if (!payload) return; | |
| if (payload === '[DONE]') { doneMarker = true; return; } | |
| consume(JSON.parse(payload)); | |
| }; | |
| try { | |
| for (;;) { | |
| const {done, value} = await reader.read(); | |
| if (done) break; | |
| touch(); buffer = (buffer + decoder.decode(value, {stream: true})).replace(/\r\n/g, '\n'); | |
| if (buffer.length > 2_000_000) throw new RuntimeError('Provider stream event too large', 'transport'); | |
| let boundary: number; | |
| while ((boundary = buffer.indexOf('\n\n')) >= 0) { processEvent(buffer.slice(0, boundary)); buffer = buffer.slice(boundary + 2); } | |
| } | |
| buffer += decoder.decode(); if (buffer.trim()) processEvent(buffer.trim()); | |
| bodyComplete = doneMarker || Boolean(trace.finish_reason); | |
| if (!bodyComplete) throw new RuntimeError('Provider stream ended before a completion marker', 'transport'); | |
| } finally { await reader.cancel().catch(() => {}); reader.releaseLock(); } | |
| } else { consume(JSON.parse(await readBounded())); bodyComplete = true; } | |
| trace.response = output; | |
| if (trace.finish_reason === 'length' || trace.finish_reason === 'content_filter') | |
| throw new RuntimeError(`Unusable completion: ${trace.finish_reason}`, 'output'); | |
| if (!output.trim()) throw new RuntimeError('Provider produced no answer text', 'output'); | |
| trace.status = 'complete'; | |
| return output.trim(); | |
| } catch (error) { | |
| trace.status = 'error'; trace.response = output; | |
| trace.error = String(error).split(this.options.apiKey).join('[REDACTED]'); | |
| // Unknown/partial charges must remain reserved; stop rather than multiply them. | |
| if (!bodyComplete || [401, 402, 403, 429].includes(trace.http_status ?? 0)) this.stopped = true; | |
| throw new RuntimeError(trace.error, error instanceof RuntimeError ? error.kind : 'transport'); | |
| } finally { | |
| clearTimeout(totalTimer); clearTimeout(idleTimer!); | |
| if (reportedUsage && finite(reportedUsage.prompt_tokens) && finite(reportedUsage.completion_tokens)) { | |
| trace.usage = {prompt_tokens: reportedUsage.prompt_tokens, completion_tokens: reportedUsage.completion_tokens, | |
| ...(finite(reportedUsage.cost) ? {cost: reportedUsage.cost} : {})}; | |
| const accounted = Math.max((reportedUsage.prompt_tokens * this.options.rates.input + reportedUsage.completion_tokens * this.options.rates.output) / 1e6, | |
| finite(reportedUsage.cost) ? reportedUsage.cost : 0); | |
| // Only release a reservation after a complete response with actual usage. | |
| if (bodyComplete) { this.ledger.spent_usd += accounted - trace.accounted_usd; trace.accounted_usd = accounted; } | |
| else if (accounted > trace.accounted_usd) { this.ledger.spent_usd += accounted - trace.accounted_usd; trace.accounted_usd = accounted; } | |
| if (accounted > reserve + 1e-9) this.stopped = true; | |
| } | |
| trace.finished_at = new Date().toISOString(); this.busy = false; this.save(); | |
| } | |
| }; | |
| } | |