diff --git a/cordis.patch.yml b/cordis.patch.yml new file mode 100644 index 0000000..72af8c6 --- /dev/null +++ b/cordis.patch.yml @@ -0,0 +1,6 @@ +# dsh-session-cost bundle patch: mounts the plugin into a profile layer stack. +# The row id is the key users override in their profile's cordis.patch.yml; the row +# name is the module this package exports, so it must stay the package name. +- insert: + - id: session-cost + name: 'dsh-session-cost' diff --git a/lib/client.js b/lib/client.js new file mode 100644 index 0000000..f3da622 --- /dev/null +++ b/lib/client.js @@ -0,0 +1,482 @@ +window.__ModuleLoader__.load({ + id: "dsh-session-cost", + factory: (require) => { +var module = { exports: {} }; var exports = module.exports; +// dsh-session-cost — client half (static bundle). +// +// One pill in the composer dock, right after the harness's own stats, showing +// what the current session has really cost. The number comes from the host +// half, which reads it out of the provider response; nothing is computed here. +// +// Layout contract: +// - the dock is a LIST slot, so registering under our own id adds a neighbour +// instead of taking the native `stats` seat the way dsh-bottom-info-bar does; +// - host data arrives over POST /_dsh/dsh-session-cost/ as JSON. +'use strict'; + +const React = require('react'); + +/** Route prefix the host half registers. */ +const RPC_BASE = '/_dsh/dsh-session-cost'; + +/** + * The plugin context, captured in `apply`. + * + * Slot components receive only framework props, so the context is held here — + * the same shape the reference bottom-info-bar plugin uses. + */ +let runtimeCtx = null; + +/** Poll gap while a session is active, and once idle. */ +const POLL_ACTIVE_MS = 1000; +const POLL_IDLE_MS = 5000; + +/** Give up on a request rather than leaving the pill loading forever. */ +const RPC_TIMEOUT_MS = 15000; + +/** Fall back to the browser language when the harness serves no locale. */ +function detectLanguage() { + try { + const tag = typeof navigator !== 'undefined' && navigator.language ? navigator.language : 'en'; + return String(tag).toLowerCase().startsWith('ru') ? 'ru' : 'en'; + } catch (err) { + return 'en'; + } +} + +const LOCALES = { + ru: { + 'pill.aria': 'Стоимость текущей сессии', + 'pill.title': 'Стоимость сессии', + 'pill.none': 'Пока ничего не потрачено', + 'status.title': 'Стоимость сессии', + 'status.description': 'Показывает реально списанную стоимость текущей сессии в нижнем баре.', + 'status.session': 'Текущая сессия', + 'status.total': 'Итого', + 'status.calls': 'Оплаченных вызовов', + 'status.byPurpose': 'По назначению', + 'status.purpose.agent': 'Ответы агента', + 'status.purpose.compaction': 'Компактификация', + 'status.purpose.session-title': 'Заголовки сессий', + 'status.health': 'Состояние учёта', + 'status.observed': 'Прочитано ответов с ценой', + 'status.attributed': 'Отнесено к сессиям', + 'status.cancelled': 'Прервано (не тарифицируется)', + 'status.failed': 'Завершилось ошибкой', + 'status.unpriced': 'Без цены', + 'status.unattributed': 'Не удалось отнести к сессии', + 'status.waiting': 'Ожидают цену', + 'status.parked': 'Цены без вызова', + 'status.sessions': 'Сессий в реестре', + 'status.readonly': 'Настроек у плагина нет: цена приходит от провайдера.', + 'status.unavailable': 'Данные недоступны: хостовая половина не ответила.', + }, + en: { + 'pill.aria': 'Cost of the current session', + 'pill.title': 'Session cost', + 'pill.none': 'Nothing spent yet', + 'status.title': 'Session cost', + 'status.description': 'Shows what the current session has really cost, in the bottom bar.', + 'status.session': 'Current session', + 'status.total': 'Total', + 'status.calls': 'Paid calls', + 'status.byPurpose': 'By purpose', + 'status.purpose.agent': 'Agent replies', + 'status.purpose.compaction': 'Compaction', + 'status.purpose.session-title': 'Session titles', + 'status.health': 'Accounting health', + 'status.observed': 'Responses read with a price', + 'status.attributed': 'Attributed to sessions', + 'status.cancelled': 'Cancelled (not billed)', + 'status.failed': 'Finished with an error', + 'status.unpriced': 'Without a price', + 'status.unattributed': 'Could not place in a session', + 'status.waiting': 'Waiting for a price', + 'status.parked': 'Prices without a call', + 'status.sessions': 'Sessions in the ledger', + 'status.readonly': 'This plugin has no settings: the price comes from the provider.', + 'status.unavailable': 'No data: the host half did not answer.', + }, +}; + +/** Dictionary namespace owned by this plugin. */ +const LOCALE_NAMESPACE = 'sessionCost'; + +/** Build the reader for one language, falling back to English. */ +function makeTranslator() { + let language = detectLanguage(); + const listeners = new Set(); + const read = (key, fallback) => { + const table = LOCALES[language] ?? LOCALES.en; + return table[key] ?? LOCALES.en[key] ?? fallback ?? key; + }; + return { + t: read, + get language() { return language; }, + setLanguage(next) { + if (next !== 'ru' && next !== 'en') return; + if (next === language) return; + language = next; + listeners.forEach((fn) => fn()); + }, + subscribe(fn) { + listeners.add(fn); + return () => listeners.delete(fn); + }, + }; +} + +const translator = makeTranslator(); +const t = (key, fallback) => translator.t(key, fallback); + +/** Milliseconds, and never longer than the timeout. */ +function scheduleTimeout(fn, ms) { + try { + return setTimeout(fn, ms); + } catch (err) { + return null; + } +} + +function cancelTimeout(handle) { + if (handle === null || handle === undefined) return; + try { + clearTimeout(handle); + } catch (err) { + /* nothing to clear */ + } +} + +/** One JSON request to the host half. */ +function rpc(method, args) { + let timedOut = false; + const controller = new AbortController(); + const timer = scheduleTimeout(() => { + timedOut = true; + controller.abort(); + }, RPC_TIMEOUT_MS); + const done = () => cancelTimeout(timer); + return fetch(RPC_BASE + '/' + method, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify(args ?? {}), + signal: controller.signal, + }) + .then((res) => { + if (!res.ok) throw new Error('HTTP ' + res.status); + return res.json(); + }) + .catch((err) => { + if (timedOut) throw new Error('timeout'); + throw err; + }) + .finally(done); +} + +/** Two decimals would show 0,00 almost always; the owner chose four. */ +function formatAmount(value) { + if (typeof value !== 'number' || !Number.isFinite(value)) return null; + try { + return new Intl.NumberFormat(translator.language === 'ru' ? 'ru-RU' : 'en-US', { + minimumFractionDigits: 4, + maximumFractionDigits: 4, + }).format(value); + } catch (err) { + return value.toFixed(4).replace('.', translator.language === 'ru' ? ',' : '.'); + } +} + +/** Insert this plugin's stylesheet once. */ +function installStyles() { + const tagId = 'dsh-session-cost/styles'; + if (typeof document === 'undefined') return () => {}; + if (document.querySelector('style[data-plugin-css="' + tagId + '"]') !== null) return () => {}; + const style = document.createElement('style'); + style.dataset.plugin = 'dsh-session-cost'; + style.dataset.pluginCss = tagId; + style.textContent = [ + '.dsh-session-cost{display:inline-flex;align-items:center;gap:6px;padding:4px 11px;', + 'border-radius:999px;background:var(--dsw-alias-bg-l2, #1b2b21);', + 'border:1px solid var(--dsw-alias-border-l1, #2c4a37);', + 'color:var(--dsw-alias-label-secondary, #8fd6a8);font-size:12px;line-height:1.4;', + 'font-variant-numeric:tabular-nums;white-space:nowrap;}', + '.dsh-session-cost__amount{font-weight:600;}', + '.dsh-sc-status{display:flex;flex-direction:column;gap:10px;padding:4px 0;font-size:13px;line-height:1.6;}', + '.dsh-sc-status h4{margin:6px 0 0;font-size:13px;font-weight:600;}', + '.dsh-sc-status table{border-collapse:collapse;width:100%;max-width:520px;}', + '.dsh-sc-status th{text-align:left;font-weight:400;color:var(--dsw-alias-label-tertiary, #8b929e);padding:2px 12px 2px 0;}', + '.dsh-sc-status td{text-align:right;font-variant-numeric:tabular-nums;padding:2px 0;}', + '.dsh-sc-status__note{color:var(--dsw-alias-label-tertiary, #8b929e);}', + '.dsh-sc-status__error{color:var(--dsw-alias-label-error, #e06c75);}', + ].join(''); + document.head.appendChild(style); + return () => { + if (style.parentNode) style.parentNode.removeChild(style); + }; +} + +/** + * Ask the host who this session is. + * + * The standard slot kit answers first; the session snapshot and the runtime + * sessions service are fallbacks, because host versions differ. + */ +function useSessionId(props) { + return React.useMemo(() => { + try { + if (typeof props.sessionId === 'string' && props.sessionId.length > 0) return props.sessionId; + if (props.session && typeof props.session.sessionId === 'string') return props.session.sessionId; + const sessions = runtimeCtx && runtimeCtx.get ? runtimeCtx.get('sessions') : null; + const current = sessions && sessions.list ? sessions.list.getSnapshot().current : null; + if (typeof current === 'string' && current.length > 0) return current; + } catch (err) { + /* fall through to the empty id, which the host answers with null */ + } + return ''; + }, [props.sessionId, props.session]); +} + +/** Poll the host for one session's running total. */ +function useSessionCost(sessionId) { + const [state, setState] = React.useState({ total: null, loading: true, error: null }); + React.useEffect(() => { + let cancelled = false; + let timer = null; + + const load = () => { + rpc('session', { sessionId }) + .then((data) => { + if (cancelled) return; + setState({ total: data && typeof data.total === 'number' ? data.total : null, loading: false, error: null }); + }) + .catch(() => { + if (cancelled) return; + setState((prev) => ({ total: prev.total, loading: false, error: 'unavailable' })); + }) + .finally(() => { + if (cancelled) return; + const gap = typeof document !== 'undefined' && document.visibilityState === 'visible' + ? POLL_ACTIVE_MS + : POLL_IDLE_MS; + timer = scheduleTimeout(load, gap); + }); + }; + + load(); + return () => { + cancelled = true; + cancelTimeout(timer); + }; + }, [sessionId]); + return state; +} + +/** The pill itself: one figure, right after the harness's own stats. */ +function SessionCostPill(props) { + const sessionId = useSessionId(props); + const { total } = useSessionCost(sessionId); + const [, bump] = React.useState(0); + + React.useEffect(() => translator.subscribe(() => bump((v) => v + 1)), []); + + if (total === null) return null; // no paid call yet: say nothing + const amount = formatAmount(total); + if (amount === null) return null; + + return React.createElement( + 'span', + { className: 'dsh-session-cost', title: t('pill.title'), 'aria-label': t('pill.aria') }, + React.createElement('span', { className: 'dsh-session-cost__amount' }, '\u20BD ' + amount), + ); +} + +/** One label/value row in the status page. */ +function Row(props) { + return React.createElement( + 'tr', + null, + React.createElement('th', null, props.label), + React.createElement('td', null, props.value), + ); +} + +/** The plugin page: read-only diagnostics, because there is nothing to set. */ +function SessionCostStatus(props) { + const [data, setData] = React.useState(null); + const [failed, setFailed] = React.useState(false); + const [, bump] = React.useState(0); + + React.useEffect(() => translator.subscribe(() => bump((v) => v + 1)), []); + + React.useEffect(() => { + let cancelled = false; + let timer = null; + const load = () => { + rpc('status', {}) + .then((payload) => { + if (cancelled) return; + setData(payload); + setFailed(false); + }) + .catch(() => { + if (cancelled) return; + setFailed(true); + }) + .finally(() => { + if (!cancelled) timer = scheduleTimeout(load, 3000); + }); + }; + load(); + return () => { + cancelled = true; + cancelTimeout(timer); + }; + }, []); + + if (props.view === 'summary') return t('status.description'); + + const stats = data && data.stats ? data.stats : null; + const purposes = data && data.byPurpose ? Object.entries(data.byPurpose) : []; + + return React.createElement( + 'div', + { className: 'dsh-sc-status' }, + React.createElement('div', { className: 'dsh-sc-status__note' }, t('status.readonly')), + failed ? React.createElement('div', { className: 'dsh-sc-status__error' }, t('status.unavailable')) : null, + React.createElement('h4', null, t('status.session')), + React.createElement( + 'table', + null, + React.createElement( + 'tbody', + null, + React.createElement(Row, { label: t('status.session'), value: data?.sessionId || '\u2014' }), + React.createElement(Row, { label: t('status.total'), value: '\u20BD ' + (formatAmount(data?.total ?? 0) ?? '0,0000') }), + React.createElement(Row, { label: t('status.calls'), value: String(data?.calls ?? 0) }), + ), + ), + purposes.length > 0 + ? React.createElement( + React.Fragment, + null, + React.createElement('h4', null, t('status.byPurpose')), + React.createElement( + 'table', + null, + React.createElement( + 'tbody', + null, + purposes.map(([name, amount]) => + React.createElement(Row, { + key: name, + label: t('status.purpose.' + name, name), + value: '\u20BD ' + (formatAmount(amount) ?? '0,0000'), + }), + ), + ), + ), + ) + : null, + stats !== null + ? React.createElement( + React.Fragment, + null, + React.createElement('h4', null, t('status.health')), + React.createElement( + 'table', + null, + React.createElement( + 'tbody', + null, + React.createElement(Row, { label: t('status.observed'), value: String(stats.observed) }), + React.createElement(Row, { label: t('status.attributed'), value: String(stats.attributed) }), + React.createElement(Row, { label: t('status.cancelled'), value: String(stats.cancelled) }), + React.createElement(Row, { label: t('status.failed'), value: String(stats.failed) }), + React.createElement(Row, { label: t('status.unpriced'), value: String(stats.unpriced) }), + React.createElement(Row, { label: t('status.unattributed'), value: String(stats.unattributed) }), + React.createElement(Row, { label: t('status.waiting'), value: String(stats.waiting) }), + React.createElement(Row, { label: t('status.parked'), value: String(stats.parked) }), + React.createElement(Row, { label: t('status.sessions'), value: String(stats.sessions) }), + ), + ), + ) + : null, + ); +} + +exports.inject = ['slots', 'locale']; +exports.apply = async function apply(ctx) { + // The locale service may be absent or late; either way the plugin keeps working. + let locale = null; + try { + locale = ctx.locale; + } catch (err) { + locale = null; + } + if (!locale && typeof ctx.get === 'function') { + try { + locale = ctx.get('locale'); + } catch (err) { + locale = null; + } + } + if (locale && typeof locale.register === 'function') { + try { + const dispose = locale.register(LOCALE_NAMESPACE, LOCALES); + ctx.effect(() => dispose, 'dsh-session-cost: dictionaries'); + } catch (err) { + /* fall back to the browser language */ + } + } + + ctx.effect(() => installStyles(), 'dsh-session-cost: styles'); + + runtimeCtx = ctx; + ctx.effect(() => () => { runtimeCtx = null; }, 'dsh-session-cost: context'); + + let slots = ctx.slots || (ctx.get ? ctx.get('slots') : undefined); + for (let i = 0; slots === undefined && i < 60; i++) { + await new Promise((resolve) => { + if (scheduleTimeout(resolve, Math.min(300 + i * 20, 1000)) === null) resolve(); + }); + slots = ctx.slots || (ctx.get ? ctx.get('slots') : undefined); + } + if (slots === undefined) { + console.warn('[dsh-session-cost] the slots service never became available'); + return; + } + + // The dock is a list slot: our own id adds a neighbour and leaves the + // harness's `stats` pill exactly where it is. + ctx.effect( + () => + slots.inject('conversation.composer.dock', () => + slots.register( + { name: 'conversation.composer.dock', id: 'session-cost', order: 10, locale: LOCALE_NAMESPACE }, + SessionCostPill, + ), + ), + 'dsh-session-cost: pill', + ); + + // The plugin's own page on the Plugins screen. + ctx.effect( + () => + slots.inject('plugins.bundle.config', () => + slots.register( + { + name: 'plugins.bundle.config', + key: 'dsh-session-cost', + locale: LOCALE_NAMESPACE, + label: () => t('status.title'), + }, + SessionCostStatus, + ), + ), + 'dsh-session-cost: page', + ); +}; + +return module.exports; + }, +}); diff --git a/lib/index.js b/lib/index.js new file mode 100644 index 0000000..e1005a8 --- /dev/null +++ b/lib/index.js @@ -0,0 +1,454 @@ +/** + * dsh-session-cost — host half. + * + * The harness throws away the cost its provider reports: `dsh-llm-pi-ai`'s + * `mapUsage` copies only token counts, and `pi-ai`'s `parseChunkUsage` fills a + * `cost` object from the model catalog (which does not describe a hand-declared + * route, so it is all zeros). The raw provider response still carries the real + * amount, so this plugin reads it there instead. + * + * Two observers, joined by generation id: + * 1. a `globalThis.fetch` wrapper reads `usage.cost` out of provider responses; + * 2. the `llm/stream` waterfall supplies the session the call belongs to. + * + * Neither is enough alone: the wrapper sees money but not the session, the seam + * sees the session but not the money. + * + * @module dsh-session-cost + */ + +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; + +/** Marker so a reloaded plugin does not wrap `fetch` twice. */ +const WRAP_MARK = Symbol.for('dsh-session-cost.fetch-wrapper'); + +/** How long to wait for a cost before calling a completed call unpriced. */ +const PENDING_TIMEOUT_MS = 60_000; + +/** Upper bounds, so a long-lived host cannot grow without limit. */ +const MAX_REGISTRY = 5000; +const MAX_SETTLED = 5000; + +/** Where the ledgers live. */ +function dataDir() { + const home = process.env.DSH_HOME || path.join(os.homedir(), '.dsh'); + return path.join(home, 'dsh-session-cost'); +} + +/** True for the response of a model call we should read. */ +function looksLikeModelCall(url, response) { + if (response.headers.get('x-generation-id')) return true; + const type = response.headers.get('content-type') || ''; + if (type.includes('text/event-stream')) return true; + if (!type.includes('json')) return false; + try { + const pathname = new URL(url).pathname; + return /(chat\/completions|completions|messages|responses)$/.test(pathname); + } catch { + return false; + } +} + +/** Read a request's URL whichever shape `fetch` was called with. */ +function urlOf(input) { + if (typeof input === 'string') return input; + if (input instanceof URL) return input.href; + if (input && typeof input.url === 'string') return input.url; + return ''; +} + +/** A finite, non-negative number, or null. */ +function finiteAmount(value) { + if (typeof value !== 'number' || !Number.isFinite(value) || value < 0) return null; + return value; +} + +/** Locate `usage.cost` and the generation id in one parsed provider payload. */ +function readPayload(payload) { + if (!payload || typeof payload !== 'object') return null; + const usage = payload.usage; + if (!usage || typeof usage !== 'object') return null; + const cost = finiteAmount(usage.cost); + if (cost === null) return null; + const id = typeof payload.id === 'string' && payload.id.length > 0 ? payload.id : null; + return { id, cost }; +} + +/** + * Split an SSE byte stream into `data:` payloads. + * Chunk boundaries fall anywhere, so the tail is carried between pieces. + */ +async function eachSsePayload(stream, onPayload) { + const decoder = new TextDecoder(); + let buffer = ''; + for await (const piece of stream) { + buffer += decoder.decode(piece, { stream: true }); + const lines = buffer.split('\n'); + buffer = lines.pop() ?? ''; + for (const line of lines) { + if (!line.startsWith('data:')) continue; + const body = line.slice(5).trim(); + if (body.length === 0 || body === '[DONE]') continue; + try { + onPayload(JSON.parse(body)); + } catch { + /* a non-JSON keepalive line is not an error */ + } + } + } + buffer += decoder.decode(); + const tail = buffer.trim(); + if (tail.startsWith('data:')) { + const body = tail.slice(5).trim(); + if (body.length > 0 && body !== '[DONE]') { + try { + onPayload(JSON.parse(body)); + } catch { + /* ignore */ + } + } + } +} + +export default { + inject: ['timer'], + + apply(ctx) { + // ---------------------------------------------------------------- state + /** generation id -> cost in the provider's own currency. */ + const costByGeneration = new Map(); + /** generation ids already added to a session total. */ + const settled = new Set(); + /** session id -> { total, calls, byPurpose }. */ + const sessionTotals = new Map(); + /** generation id -> { sessionId, purpose }, waiting for its cost. */ + const waiting = new Map(); + /** Diagnostics surfaced on the status page. */ + const stats = { + observed: 0, // provider responses whose cost we read + attributed: 0, // calls joined to a session + cancelled: 0, // aborted by the user; the provider bills nothing + failed: 0, // finished with an error; no cost expected + unpriced: 0, // completed, but no cost ever arrived + unattributed: 0, // cost read, but the call carried no session id + wrappers: 0, + }; + + let flushTimer = null; + let dirty = false; + + // ---------------------------------------------------------- persistence + const stateFile = path.join(dataDir(), 'state.json'); + + function load() { + try { + const raw = fs.readFileSync(stateFile, 'utf8'); + const parsed = JSON.parse(raw); + for (const [id, cost] of Object.entries(parsed.costs ?? {})) { + costByGeneration.set(id, cost); + } + for (const [id, total] of Object.entries(parsed.sessions ?? {})) { + sessionTotals.set(id, { + total: finiteAmount(total) ?? 0, + calls: 0, + byPurpose: {}, + }); + } + for (const id of parsed.settled ?? []) settled.add(id); + } catch { + /* first run, or an unreadable ledger: start clean rather than fail */ + } + } + + function flush() { + flushTimer = null; + if (!dirty) return; + dirty = false; + try { + fs.mkdirSync(path.dirname(stateFile), { recursive: true }); + const payload = { + version: 1, + costs: Object.fromEntries(costByGeneration), + sessions: Object.fromEntries( + [...sessionTotals].map(([id, value]) => [id, value.total]), + ), + settled: [...settled], + }; + fs.writeFileSync(stateFile, JSON.stringify(payload)); + } catch (error) { + console.warn('[dsh-session-cost] could not write the ledger:', error?.message ?? error); + } + } + + function scheduleFlush() { + dirty = true; + if (flushTimer !== null) return; + flushTimer = ctx.setTimeout(flush, 2000); + } + + /** Keep the in-memory maps bounded without losing recent history. */ + function trim(map, cap) { + while (map.size > cap) { + const oldest = map.keys().next().value; + map.delete(oldest); + } + } + + load(); + + // ------------------------------------------------------- cost recording + /** Add one cost to a session, at most once per generation. */ + function settle(sessionId, generationId, cost, purpose) { + if (generationId !== null) { + if (settled.has(generationId)) return; + settled.add(generationId); + trim(settled, MAX_SETTLED); + } + let entry = sessionTotals.get(sessionId); + if (entry === undefined) { + entry = { total: 0, calls: 0, byPurpose: {} }; + sessionTotals.set(sessionId, entry); + } + const bucket = purpose === undefined || purpose === '' ? 'agent' : purpose; + entry.total += cost; + entry.calls += 1; + entry.byPurpose[bucket] = (entry.byPurpose[bucket] ?? 0) + cost; + stats.attributed += 1; + scheduleFlush(); + } + + /** A cost arrived: join it to a waiting call, or park it for a later one. */ + function recordCost(generationId, cost) { + stats.observed += 1; + const parked = waiting.get(generationId); + if (parked !== undefined) { + waiting.delete(generationId); + settle(parked.sessionId, generationId, cost, parked.purpose); + return; + } + if (generationId === null) { + // No id to join on: the money is real but cannot be placed. + stats.unattributed += 1; + console.warn('[dsh-session-cost] a provider response reported cost without an id'); + return; + } + costByGeneration.set(generationId, cost); + trim(costByGeneration, MAX_REGISTRY); + scheduleFlush(); + // If nothing ever comes back for it, it stays here and is reported as + // parked rather than silently dropped. + } + + // ---------------------------------------------------- observer 1: fetch + function installFetchWrapper() { + const current = globalThis.fetch; + if (typeof current !== 'function') return; + if (current[WRAP_MARK] === true) { + stats.wrappers = 1; + return; // a previous instance already wrapped it + } + + const wrapped = async function (input, init) { + const response = await current.call(this, input, init); + try { + const url = urlOf(input); + if (!looksLikeModelCall(url, response)) return response; + 'use strict'; + const body = response.body; + if (body === null || typeof body.tee !== 'function') return response; + + const generationId = + response.headers.get('x-generation-id') ?? + (typeof response.headers.get('x-session-id') === 'string' + ? response.headers.get('x-session-id') + : null); + + const [mine, theirs] = body.tee(); + const type = response.headers.get('content-type') || ''; + + // Read our branch to completion; never leave it unread, or the + // consumer's branch would be back-pressured behind it. + const observe = (async () => { + try { + if (type.includes('text/event-stream')) { + let id = generationId; + await eachSsePayload(mine, (payload) => { + const found = readPayload(payload); + if (found === null) return; + if (found.id !== null) id = found.id; + if (id !== null) recordCost(id, found.cost); + }); + } else { + const text = await new Response(mine).text(); + let parsed = null; + try { + parsed = JSON.parse(text); + } catch { + return; + } + const found = readPayload(parsed); + if (found !== null) { + const id = found.id ?? generationId; + if (id !== null) recordCost(id, found.cost); + } + } + } catch { + /* an aborted or truncated body carries no cost; nothing to do */ + } + })(); + // Detach: the consumer must not wait on our bookkeeping. + void observe; + + const rebuilt = new Response(theirs, { + status: response.status, + statusText: response.statusText, + headers: response.headers, + }); + // A rebuilt Response loses `url`; the OpenAI SDK logs it on retries. + try { + Object.defineProperty(rebuilt, 'url', { value: response.url, configurable: true }); + } catch { + /* not fatal: url is used for diagnostics only */ + } + stats.wrappers = 1; + return rebuilt; + } catch (error) { + // Any failure here must leave the request working. + console.warn('[dsh-session-cost] read failed, passing the response through:', error?.message ?? error); + return response; + } + }; + + Object.defineProperty(wrapped, WRAP_MARK, { value: true }); + globalThis.fetch = wrapped; + ctx.effect(() => () => { + if (globalThis.fetch === wrapped) globalThis.fetch = current; + }, 'dsh-session-cost: fetch wrapper'); + } + + installFetchWrapper(); + + // ----------------------------------------------------- observer 2: seam + ctx.on('llm/stream', (options, next) => { + const sessionId = typeof options?.sessionId === 'string' ? options.sessionId : undefined; + const purpose = typeof options?.purpose === 'string' ? options.purpose : undefined; + return (async function* passThrough() { + let generationId = null; + let finished = null; + try { + for await (const chunk of await next()) { + if (chunk && chunk.type === 'usage') { + // Nothing to take: the harness's usage has no cost field. + } else if (chunk && chunk.type === 'finish') { + finished = chunk.reason ?? null; + const id = chunk.replayState?.response?.responseId; + if (typeof id === 'string' && id.length > 0) generationId = id; + } + yield chunk; + } + } finally { + const kind = finished?.kind ?? 'stop'; + if (kind === 'aborted') { + // The provider bills nothing for a cancelled generation (verified: + // a late abort still reports total_cost 0), so this is not a loss. + stats.cancelled += 1; + } else if (kind === 'error') { + stats.failed += 1; + } else if (generationId === null) { + stats.unpriced += 1; + } else if (costByGeneration.has(generationId)) { + const cost = costByGeneration.get(generationId); + costByGeneration.delete(generationId); + if (sessionId === undefined) { + stats.unattributed += 1; + } else { + settle(sessionId, generationId, cost, purpose); + } + } else if (sessionId === undefined) { + // Future plugins (e.g. an experimental auto-review) call the model + // with no sessionId at all; those calls cannot be placed. + stats.unattributed += 1; + } else { + // Normal path is cost-first, so this is rare: park the call and let + // the cost find it. + waiting.set(generationId, { sessionId, purpose, at: Date.now() }); + } + } + })(); + }); + + // Drop waiters whose cost never arrived, and report them. + ctx.setInterval(() => { + const now = Date.now(); + for (const [id, entry] of waiting) { + if (now - entry.at < PENDING_TIMEOUT_MS) continue; + waiting.delete(id); + stats.unpriced += 1; + console.warn('[dsh-session-cost] a completed call never reported a cost:', id); + } + }, 30_000); + + // --------------------------------------------------------------- status + function snapshot(sessionId) { + const entry = sessionTotals.get(sessionId); + return { + sessionId, + total: entry?.total ?? null, + calls: entry?.calls ?? 0, + byPurpose: entry?.byPurpose ?? {}, + currency: 'RUB', + stats: { ...stats, waiting: waiting.size, parked: costByGeneration.size, sessions: sessionTotals.size }, + }; + } + + // ---------------------------------------------------------------- route + const ROUTE_PREFIX = '/_dsh/dsh-session-cost'; + + /** Answer with JSON, or with a plain error the client can show. */ + function respond(res, status, payload) { + const body = JSON.stringify(payload); + res.writeHead(status, { + 'content-type': 'application/json; charset=utf-8', + 'cache-control': 'no-store', + 'content-length': Buffer.byteLength(body), + }); + res.end(body); + } + + ctx.inject(['connection', 'webServer'], (webCtx) => { + webCtx.effect(() => { + const dispose = webCtx.webServer.register({ + kind: 'prefix', + path: ROUTE_PREFIX, + handler: async (req, res) => { + try { + const rejection = webCtx.connection.requestRejection(req); + if (rejection !== undefined) { + respond(res, rejection, { error: rejection === 401 ? 'authentication required' : 'request rejected' }); + return; + } + const url = new URL(req.url || '/', 'http://localhost'); + const tail = url.pathname.slice(ROUTE_PREFIX.length); + if (tail !== '/session' && tail !== '/status') { + respond(res, 404, { error: 'unknown method' }); + return; + } + const sessionId = url.searchParams.get('sessionId') ?? ''; + respond(res, 200, tail === '/session' ? snapshot(sessionId) : snapshot('')); + } catch (error) { + respond(res, 500, { error: String(error?.message ?? error) }); + } + }, + }); + return dispose; + }, 'dsh-session-cost: http route'); + }); + + ctx.effect(() => () => { + if (flushTimer !== null) ctx.clearTimeout(flushTimer); + flush(); + }, 'dsh-session-cost: ledger flush'); + }, +}; diff --git a/package.json b/package.json new file mode 100644 index 0000000..5634cf8 --- /dev/null +++ b/package.json @@ -0,0 +1,33 @@ +{ + "name": "dsh-session-cost", + "version": "0.1.0", + "private": true, + "description": "Shows the real cost of the current session in the composer dock, next to the harness's own stats.", + "type": "module", + "main": "lib/index.js", + "exports": { + ".": "./lib/index.js", + "./client": "./lib/client.js", + "./cordis.patch.yml": "./cordis.patch.yml", + "./package.json": "./package.json" + }, + "files": [ + "lib", + "cordis.patch.yml" + ], + "dsh": { + "bundle": { + "patch": "./cordis.patch.yml" + }, + "client": { + "platform": "web", + "inject": [ + "@deepseek-ai/dsh-client-ui-conversation", + "@deepseek-ai/dsh-client-locale" + ] + } + }, + "peerDependencies": { + "@deepseek-ai/dsh": "^0.2.0-rc.2" + } +}