diff --git a/packages/cloud-agents/src/server/__tests__/judgment-usage-accounting.test.ts b/packages/cloud-agents/src/server/__tests__/judgment-usage-accounting.test.ts new file mode 100644 index 0000000000..182fa17590 --- /dev/null +++ b/packages/cloud-agents/src/server/__tests__/judgment-usage-accounting.test.ts @@ -0,0 +1,300 @@ +const { mockRecordLlmUsage } = vi.hoisted(() => ({ + mockRecordLlmUsage: vi.fn(), +})); + +vi.mock('@roomote/db/server', () => ({ + recordLlmUsage: mockRecordLlmUsage, +})); + +vi.mock('../non-task-provider-usage', () => ({ + NON_TASK_INFERENCE_SURFACES: { judgmentModel: 'judgment_model' }, +})); + +import { + JudgmentHttpError, + JudgmentResponseParseError, + normalizeJudgmentUsage, + trackJudgmentRequest, + type JudgmentRequestTracking, +} from '../judgment-usage-accounting'; + +function response( + body: Record, + init: { status?: number; headers?: HeadersInit } = {}, +) { + return { + body, + status: init.status ?? 200, + headers: new Headers(init.headers), + }; +} + +function tracking( + overrides: Partial = {}, +): JudgmentRequestTracking { + return { + requestId: 'request-123', + provider: 'typesafe' as const, + model: 'jev-latest', + role: 'primary' as const, + ...overrides, + }; +} + +async function waitForUsageCall() { + await vi.waitFor(() => expect(mockRecordLlmUsage).toHaveBeenCalledTimes(1)); + return mockRecordLlmUsage.mock.calls[0]?.[0] as Record; +} + +describe('judgment usage accounting', () => { + beforeEach(() => { + vi.clearAllMocks(); + mockRecordLlmUsage.mockResolvedValue({ recorded: true }); + }); + + afterEach(() => { + vi.restoreAllMocks(); + }); + + it('normalizes provider response usage without recording judgment content', async () => { + const tracked = await trackJudgmentRequest(tracking(), async () => + response({ + model: 'jev-1.13.0', + answers: { decision: 'private answer' }, + usage: { + input_tokens: 120, + output_tokens: 30, + total_tokens: 150, + cost: 0.001234, + }, + }), + ); + + tracked.finish('success'); + const usage = await waitForUsageCall(); + + expect(usage).toMatchObject({ + source: 'judgment_model', + usageType: 'inference', + eventKey: 'judgment-model:typesafe:primary:request-123', + providerId: 'typesafe', + modelId: 'jev-1.13.0', + inputTokens: 120, + outputTokens: 30, + totalTokens: 150, + costMicroUsd: 1234, + costSource: 'provider_response', + details: { + surface: 'judgment_model', + requestRole: 'primary', + status: 200, + outcome: 'success', + latencyMs: expect.any(Number), + usageMetadataAvailable: true, + usageMetadataSource: 'response', + metadataReadFailed: false, + missingUsageFields: [], + }, + }); + expect(JSON.stringify(usage)).not.toContain('private answer'); + }); + + it('keeps missing metadata explicit and does not derive reported totals', async () => { + const tracked = await trackJudgmentRequest(tracking(), async () => + response({ usage: { input_tokens: 12, output_tokens: 5 } }), + ); + + tracked.finish('success'); + + await expect(waitForUsageCall()).resolves.toMatchObject({ + inputTokens: 12, + outputTokens: 5, + totalTokens: 0, + costMicroUsd: null, + costSource: 'missing', + details: { + usageMetadataAvailable: true, + usageMetadataSource: 'response', + missingUsageFields: ['total_tokens', 'cost'], + }, + }); + }); + + it('exposes normalized token aliases through the typed metadata interface', () => { + expect( + normalizeJudgmentUsage( + response({ usage: { prompt_tokens: '17', completion_tokens: 6 } }), + ), + ).toMatchObject({ + inputTokens: 17, + outputTokens: 6, + usageMetadataAvailable: true, + usageMetadataSource: 'response', + }); + }); + + it('normalizes provider metadata from response headers', async () => { + const tracked = await trackJudgmentRequest( + tracking({ provider: 'openrouter', model: 'typesafe/jev-1.13' }), + async () => + response( + {}, + { + headers: { + 'x-input-tokens': '11', + 'x-output-tokens': '4', + 'x-total-tokens': '15', + 'x-openrouter-cost': '0.0005', + 'x-provider-model': 'jev-provider-revision', + }, + }, + ), + ); + + tracked.finish('success'); + + await expect(waitForUsageCall()).resolves.toMatchObject({ + providerId: 'openrouter', + modelId: 'jev-provider-revision', + inputTokens: 11, + outputTokens: 4, + totalTokens: 15, + costMicroUsd: 500, + costSource: 'provider_response', + details: { + usageMetadataAvailable: true, + usageMetadataSource: 'header', + }, + }); + }); + + it('keeps primary and shadow requests distinct while sharing their request id', async () => { + const primary = await trackJudgmentRequest(tracking(), async () => + response({}), + ); + const shadow = await trackJudgmentRequest( + tracking({ + provider: 'roomote', + model: 'roomote-judgment', + role: 'shadow', + primaryProvider: 'typesafe', + }), + async () => response({}), + ); + + primary.finish('success'); + shadow.finish('validation_error'); + await vi.waitFor(() => expect(mockRecordLlmUsage).toHaveBeenCalledTimes(2)); + + const entries = mockRecordLlmUsage.mock.calls.map( + ([entry]) => entry as Record, + ); + expect(entries.map((entry) => entry.eventKey)).toEqual([ + 'judgment-model:typesafe:primary:request-123', + 'judgment-model:roomote:shadow:request-123', + ]); + expect(entries[1]).toMatchObject({ + providerId: 'roomote', + details: { + requestRole: 'shadow', + primaryProvider: 'typesafe', + outcome: 'validation_error', + }, + }); + }); + + it('records HTTP status and safe headers without persisting the response body', async () => { + const error = new JudgmentHttpError( + 529, + 'private upstream response body', + new Headers({ 'x-input-tokens': '8' }), + ); + + await expect( + trackJudgmentRequest(tracking(), async () => { + throw error; + }), + ).rejects.toBe(error); + + const usage = await waitForUsageCall(); + expect(usage).toMatchObject({ + inputTokens: 8, + totalTokens: 0, + details: { + status: 529, + outcome: 'http_error', + usageMetadataSource: 'header', + metadataReadFailed: false, + }, + }); + expect(JSON.stringify(usage)).not.toContain( + 'private upstream response body', + ); + }); + + it.each([ + [ + 'response_parse_error', + () => new JudgmentResponseParseError(200, new Headers()), + 200, + true, + ], + [ + 'timeout', + () => Object.assign(new Error('timed out'), { name: 'TimeoutError' }), + null, + false, + ], + ['transport_error', () => new Error('connection reset'), null, false], + ] as const)( + 'classifies %s request failures', + async (outcome, createError, status, metadataReadFailed) => { + const error = createError(); + + await expect( + trackJudgmentRequest(tracking(), async () => { + throw error; + }), + ).rejects.toBe(error); + + await expect(waitForUsageCall()).resolves.toMatchObject({ + details: { + status, + outcome, + metadataReadFailed, + }, + }); + }, + ); + + it('records a later validation outcome once', async () => { + const tracked = await trackJudgmentRequest(tracking(), async () => + response({ answers: { secret: 'not recorded' } }), + ); + + tracked.finish('validation_error'); + tracked.finish('success'); + + const usage = await waitForUsageCall(); + expect(usage).toMatchObject({ + details: { outcome: 'validation_error' }, + }); + expect(JSON.stringify(usage)).not.toContain('not recorded'); + }); + + it('keeps ledger persistence failures non-blocking', async () => { + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}); + mockRecordLlmUsage.mockRejectedValue(new Error('database unavailable')); + const tracked = await trackJudgmentRequest(tracking(), async () => + response({}), + ); + + expect(() => tracked.finish('success')).not.toThrow(); + + await vi.waitFor(() => + expect(warn).toHaveBeenCalledWith( + '[JudgmentUsage] Failed to record typesafe primary usage', + ), + ); + }); +}); diff --git a/packages/cloud-agents/src/server/__tests__/typesafe-judgment.test.ts b/packages/cloud-agents/src/server/__tests__/typesafe-judgment.test.ts index 9d53e6be58..aa0011fb80 100644 --- a/packages/cloud-agents/src/server/__tests__/typesafe-judgment.test.ts +++ b/packages/cloud-agents/src/server/__tests__/typesafe-judgment.test.ts @@ -405,211 +405,6 @@ describe('evaluateTypeSafeJudgments', () => { }); }); - it('records provider usage and cost without persisting judgment data', async () => { - mockFetchResponse({ - model: 'jev-1.13.0', - answers: directAnswers, - usage: { - input_tokens: 120, - output_tokens: 30, - total_tokens: 150, - cost: 0.001234, - }, - }); - - await evaluateTypeSafeJudgments({ - state: { secret: 'do not record' }, - questions, - }); - - await vi.waitFor(() => { - expect(mockRecordLlmUsage).toHaveBeenCalledWith( - expect.objectContaining({ - source: 'judgment_model', - usageType: 'inference', - eventKey: expect.stringMatching( - /^judgment-model:typesafe:primary:[0-9a-f-]+$/u, - ), - providerId: 'typesafe', - modelId: 'jev-1.13.0', - inputTokens: 120, - outputTokens: 30, - totalTokens: 150, - costMicroUsd: 1234, - costSource: 'provider_response', - details: { - surface: 'judgment_model', - requestRole: 'primary', - status: 200, - outcome: 'success', - latencyMs: expect.any(Number), - usageMetadataAvailable: true, - usageMetadataSource: 'response', - metadataReadFailed: false, - missingUsageFields: [], - }, - }), - ); - }); - - const [usage] = mockRecordLlmUsage.mock.calls[0] as [ - Record, - ]; - expect(JSON.stringify(usage)).not.toContain('do not record'); - }); - - it('records missing provider metadata with zero normalized totals', async () => { - mockFetchResponse({ answers: directAnswers }); - - await evaluateTypeSafeJudgments({ state: 'metadata-free', questions }); - - await vi.waitFor(() => { - expect(mockRecordLlmUsage).toHaveBeenCalledWith( - expect.objectContaining({ - inputTokens: null, - outputTokens: null, - totalTokens: 0, - costMicroUsd: null, - costSource: 'missing', - details: expect.objectContaining({ - outcome: 'success', - usageMetadataAvailable: false, - usageMetadataSource: 'none', - metadataReadFailed: false, - missingUsageFields: [ - 'input_tokens', - 'output_tokens', - 'total_tokens', - 'cost', - ], - }), - }), - ); - }); - }); - - it('records usage metadata exposed through response headers', async () => { - mockGetJudgmentSelection.mockResolvedValue('openrouter'); - mockKeys({ OPENROUTER_API_KEY: 'or-key' }); - mockFetchResponse( - { answers: directAnswers }, - { - headers: { - 'x-input-tokens': '11', - 'x-output-tokens': '4', - 'x-total-tokens': '15', - 'x-openrouter-cost': '0.0005', - }, - }, - ); - - await evaluateTypeSafeJudgments({ state: 'header metadata', questions }); - - await vi.waitFor(() => { - expect(mockRecordLlmUsage).toHaveBeenCalledWith( - expect.objectContaining({ - providerId: 'openrouter', - modelId: 'typesafe/jev-1.13', - inputTokens: 11, - outputTokens: 4, - totalTokens: 15, - costMicroUsd: 500, - costSource: 'provider_response', - details: expect.objectContaining({ - usageMetadataAvailable: true, - usageMetadataSource: 'header', - }), - }), - ); - }); - }); - - it('records an HTTP failure without recording the response body', async () => { - mockFetchResponse({ error: 'secret response body' }, { status: 529 }); - - await expect( - evaluateTypeSafeJudgments({ state: 'hi', questions }), - ).rejects.toThrow('HTTP 529'); - - await vi.waitFor(() => { - expect(mockRecordLlmUsage).toHaveBeenCalledWith( - expect.objectContaining({ - details: expect.objectContaining({ - status: 529, - outcome: 'http_error', - }), - inputTokens: null, - outputTokens: null, - totalTokens: 0, - costSource: 'missing', - }), - ); - }); - - expect(JSON.stringify(mockRecordLlmUsage.mock.calls[0])).not.toContain( - 'secret response body', - ); - }); - - it('records transport failures without affecting the thrown fallback error', async () => { - const fetchMock = vi - .fn() - .mockRejectedValue(new Error('connect ECONNREFUSED')); - vi.stubGlobal('fetch', fetchMock); - - await expect( - evaluateTypeSafeJudgments({ state: 'hi', questions }), - ).rejects.toThrow('connect ECONNREFUSED'); - - await vi.waitFor(() => { - expect(mockRecordLlmUsage).toHaveBeenCalledWith( - expect.objectContaining({ - details: expect.objectContaining({ - status: null, - outcome: 'transport_error', - }), - }), - ); - }); - }); - - it('records response validation failures after a successful HTTP request', async () => { - mockFetchResponse({ - answers: { urgent: { type: 'noul', noul: 0.5 } }, - }); - - await expect( - evaluateTypeSafeJudgments({ state: 'hi', questions }), - ).rejects.toThrow('missing a valid answer'); - - await vi.waitFor(() => { - expect(mockRecordLlmUsage).toHaveBeenCalledWith( - expect.objectContaining({ - details: expect.objectContaining({ - status: 200, - outcome: 'validation_error', - }), - }), - ); - }); - }); - - it('does not let ledger persistence failures change a valid judgment', async () => { - const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}); - mockRecordLlmUsage.mockRejectedValue(new Error('database unavailable')); - mockFetchResponse({ answers: directAnswers }); - - await expect( - evaluateTypeSafeJudgments({ state: 'hi', questions }), - ).resolves.toEqual(directAnswers); - - await vi.waitFor(() => - expect(warn).toHaveBeenCalledWith( - '[JudgmentUsage] Failed to record typesafe primary usage', - ), - ); - }); - it('resolves the hosted judgment model before the helper fallback', async () => { await expect(resolveDecisionModel()).resolves.toEqual({ kind: 'judgment', diff --git a/packages/cloud-agents/src/server/judgment-usage-accounting.ts b/packages/cloud-agents/src/server/judgment-usage-accounting.ts new file mode 100644 index 0000000000..bf81f2aea2 --- /dev/null +++ b/packages/cloud-agents/src/server/judgment-usage-accounting.ts @@ -0,0 +1,407 @@ +import { recordLlmUsage } from '@roomote/db/server'; + +import { NON_TASK_INFERENCE_SURFACES } from './non-task-provider-usage'; + +export type JudgmentUsageProvider = + | 'roomote' + | 'typesafe' + | 'openrouter' + | 'vercel'; + +export type JudgmentRequestRole = 'primary' | 'shadow'; + +export type JudgmentRequestOutcome = + | 'success' + | 'http_error' + | 'timeout' + | 'transport_error' + | 'response_parse_error' + | 'response_error' + | 'validation_error'; + +export type JudgmentRequestTracking = { + requestId: string; + provider: JudgmentUsageProvider; + model: string; + role: JudgmentRequestRole; + primaryProvider?: JudgmentUsageProvider; +}; + +export type JudgmentUsageResponse = { + body: Record; + status: number; + headers: Headers; +}; + +export type TrackedJudgmentResponse = JudgmentUsageResponse & { + finish: (outcome: JudgmentRequestOutcome) => void; +}; + +type NormalizedJudgmentUsage = { + modelId?: string; + inputTokens?: number; + outputTokens?: number; + totalTokens?: number; + costMicroUsd?: number; + pricingMetadata?: Record; + usageMetadataAvailable: boolean; + usageMetadataSource: 'response' | 'header' | 'none'; + metadataReadFailed: boolean; + missingUsageFields: string[]; +}; + +/** Carries an HTTP status and headers without persisting an error response body. */ +export class JudgmentHttpError extends Error { + constructor( + readonly status: number, + detail: string, + readonly headers?: Headers, + ) { + super( + `Judgment model request failed with HTTP ${status}${ + detail ? `: ${detail.slice(0, 200)}` : '' + }`, + ); + this.name = 'JudgmentHttpError'; + } +} + +export class JudgmentResponseParseError extends Error { + constructor( + readonly status: number, + readonly headers: Headers, + ) { + super('Judgment model response was not valid JSON'); + this.name = 'JudgmentResponseParseError'; + } +} + +function asRecord(value: unknown): Record | undefined { + return typeof value === 'object' && value !== null + ? (value as Record) + : undefined; +} + +function nonNegativeNumber(value: unknown): number | undefined { + const number = + typeof value === 'number' + ? value + : typeof value === 'string' && value.trim() + ? Number(value) + : undefined; + + return number !== undefined && Number.isFinite(number) && number >= 0 + ? number + : undefined; +} + +function firstNonNegativeNumber(...values: unknown[]): number | undefined { + for (const value of values) { + const number = nonNegativeNumber(value); + if (number !== undefined) return number; + } + + return undefined; +} + +function readHeaderNumber( + headers: Headers | undefined, + names: readonly string[], +): number | undefined { + if (!headers) return undefined; + + for (const name of names) { + const number = nonNegativeNumber(headers.get(name)); + if (number !== undefined) return number; + } + + return undefined; +} + +function readHeaderString( + headers: Headers | undefined, + names: readonly string[], +): string | undefined { + if (!headers) return undefined; + + for (const name of names) { + const value = headers.get(name)?.trim(); + if (value) return value; + } + + return undefined; +} + +export function normalizeJudgmentUsage( + response: JudgmentUsageResponse | undefined, + metadataReadFailed = false, +): NormalizedJudgmentUsage { + const body = response?.body ?? {}; + const usage = + asRecord(body.usage) ?? + asRecord(body.usageMetadata) ?? + asRecord(body.usage_metadata); + const headers = response?.headers; + + const responseInputTokens = firstNonNegativeNumber( + usage?.input_tokens, + usage?.prompt_tokens, + usage?.inputTokens, + usage?.promptTokens, + ); + const responseOutputTokens = firstNonNegativeNumber( + usage?.output_tokens, + usage?.completion_tokens, + usage?.outputTokens, + usage?.completionTokens, + ); + const responseTotalTokens = firstNonNegativeNumber( + usage?.total_tokens, + usage?.totalTokens, + ); + const responseCostMicroUsd = firstNonNegativeNumber( + usage?.cost_micro_usd, + usage?.costMicroUsd, + ); + const responseCostUsd = firstNonNegativeNumber( + usage?.cost, + usage?.cost_usd, + usage?.costUsd, + ); + + const headerInputTokens = readHeaderNumber(headers, [ + 'x-input-tokens', + 'x-prompt-tokens', + 'x-usage-input-tokens', + 'x-openrouter-input-tokens', + ]); + const headerOutputTokens = readHeaderNumber(headers, [ + 'x-output-tokens', + 'x-completion-tokens', + 'x-usage-output-tokens', + 'x-openrouter-output-tokens', + ]); + const headerTotalTokens = readHeaderNumber(headers, [ + 'x-total-tokens', + 'x-usage-total-tokens', + 'x-openrouter-total-tokens', + ]); + const headerCostMicroUsd = readHeaderNumber(headers, [ + 'x-cost-micro-usd', + 'x-usage-cost-micro-usd', + ]); + const headerCostUsd = readHeaderNumber(headers, [ + 'x-cost-usd', + 'x-cost', + 'x-provider-cost', + 'x-openrouter-cost', + ]); + + const inputTokens = responseInputTokens ?? headerInputTokens; + const outputTokens = responseOutputTokens ?? headerOutputTokens; + const totalTokens = responseTotalTokens ?? headerTotalTokens; + const costMicroUsd = + responseCostMicroUsd ?? + (responseCostUsd === undefined + ? (headerCostMicroUsd ?? + (headerCostUsd === undefined + ? undefined + : Math.round(headerCostUsd * 1_000_000))) + : Math.round(responseCostUsd * 1_000_000)); + const responseMetadataAvailable = + responseInputTokens !== undefined || + responseOutputTokens !== undefined || + responseTotalTokens !== undefined || + responseCostMicroUsd !== undefined || + responseCostUsd !== undefined; + const headerMetadataAvailable = + headerInputTokens !== undefined || + headerOutputTokens !== undefined || + headerTotalTokens !== undefined || + headerCostMicroUsd !== undefined || + headerCostUsd !== undefined; + const missingUsageFields: string[] = []; + + if (inputTokens === undefined) missingUsageFields.push('input_tokens'); + if (outputTokens === undefined) missingUsageFields.push('output_tokens'); + if (totalTokens === undefined) missingUsageFields.push('total_tokens'); + if (costMicroUsd === undefined) missingUsageFields.push('cost'); + + const costDetails = asRecord(usage?.cost_details ?? usage?.costDetails); + const responseModel = + typeof body.model === 'string' && body.model.trim() + ? body.model.trim() + : undefined; + const headerModel = readHeaderString(headers, [ + 'x-model', + 'x-provider-model', + ]); + + return { + ...(responseModel || headerModel + ? { modelId: responseModel ?? headerModel } + : {}), + ...(inputTokens === undefined ? {} : { inputTokens }), + ...(outputTokens === undefined ? {} : { outputTokens }), + ...(totalTokens === undefined ? {} : { totalTokens }), + ...(costMicroUsd === undefined ? {} : { costMicroUsd }), + ...(costDetails ? { pricingMetadata: { costDetails } } : {}), + usageMetadataAvailable: + responseMetadataAvailable || headerMetadataAvailable, + usageMetadataSource: responseMetadataAvailable + ? 'response' + : headerMetadataAvailable + ? 'header' + : 'none', + metadataReadFailed, + missingUsageFields, + }; +} + +function classifyJudgmentRequestError(error: unknown): { + outcome: JudgmentRequestOutcome; + status: number | null; + headers?: Headers; + metadataReadFailed: boolean; +} { + if (error instanceof JudgmentHttpError) { + return { + outcome: 'http_error', + status: error.status, + headers: error.headers, + metadataReadFailed: false, + }; + } + + if (error instanceof JudgmentResponseParseError) { + return { + outcome: 'response_parse_error', + status: error.status, + headers: error.headers, + metadataReadFailed: true, + }; + } + + if ( + error instanceof Error && + (error.name === 'TimeoutError' || error.name === 'AbortError') + ) { + return { + outcome: 'timeout', + status: null, + metadataReadFailed: false, + }; + } + + return { + outcome: 'transport_error', + status: null, + metadataReadFailed: false, + }; +} + +function recordJudgmentUsage(input: { + tracking: JudgmentRequestTracking; + startedAt: number; + response?: JudgmentUsageResponse; + status: number | null; + outcome: JudgmentRequestOutcome; + headers?: Headers; + metadataReadFailed?: boolean; +}): void { + const response = + input.response ?? + (input.headers + ? { body: {}, status: input.status ?? 0, headers: input.headers } + : undefined); + const usage = normalizeJudgmentUsage( + response, + input.metadataReadFailed ?? false, + ); + const eventKey = `judgment-model:${input.tracking.provider}:${input.tracking.role}:${input.tracking.requestId}`; + const details: Record = { + surface: NON_TASK_INFERENCE_SURFACES.judgmentModel, + requestRole: input.tracking.role, + status: input.status, + outcome: input.outcome, + latencyMs: Math.max(0, Date.now() - input.startedAt), + usageMetadataAvailable: usage.usageMetadataAvailable, + usageMetadataSource: usage.usageMetadataSource, + metadataReadFailed: usage.metadataReadFailed, + missingUsageFields: usage.missingUsageFields, + ...(input.tracking.primaryProvider + ? { primaryProvider: input.tracking.primaryProvider } + : {}), + }; + + const persist = async () => { + await recordLlmUsage({ + source: NON_TASK_INFERENCE_SURFACES.judgmentModel, + usageType: 'inference', + eventKey, + providerId: input.tracking.provider, + modelId: usage.modelId ?? input.tracking.model, + inputTokens: usage.inputTokens ?? null, + outputTokens: usage.outputTokens ?? null, + totalTokens: usage.totalTokens ?? 0, + contextTokens: usage.inputTokens ?? 0, + costMicroUsd: usage.costMicroUsd ?? null, + costSource: + usage.costMicroUsd === undefined ? 'missing' : 'provider_response', + ...(usage.pricingMetadata + ? { pricingMetadata: usage.pricingMetadata } + : {}), + details, + }); + }; + + void persist().catch(() => { + // Usage telemetry must never change judgment behavior. + console.warn( + `[JudgmentUsage] Failed to record ${input.tracking.provider} ${input.tracking.role} usage`, + ); + }); +} + +/** + * Run one provider request and expose an idempotent completion hook for answer + * translation/validation. Failed requests are classified and recorded before + * the original error is rethrown; ledger persistence remains detached. + */ +export async function trackJudgmentRequest( + tracking: JudgmentRequestTracking, + request: () => Promise, +): Promise { + const startedAt = Date.now(); + + try { + const response = await request(); + let finished = false; + + return { + ...response, + finish: (outcome) => { + if (finished) return; + finished = true; + recordJudgmentUsage({ + tracking, + startedAt, + response, + status: response.status, + outcome, + }); + }, + }; + } catch (error) { + const failure = classifyJudgmentRequestError(error); + recordJudgmentUsage({ + tracking, + startedAt, + status: failure.status, + outcome: failure.outcome, + headers: failure.headers, + metadataReadFailed: failure.metadataReadFailed, + }); + throw error; + } +} diff --git a/packages/cloud-agents/src/server/typesafe-judgment.ts b/packages/cloud-agents/src/server/typesafe-judgment.ts index 1a0185c68e..71ab6ffdf6 100644 --- a/packages/cloud-agents/src/server/typesafe-judgment.ts +++ b/packages/cloud-agents/src/server/typesafe-judgment.ts @@ -3,7 +3,6 @@ import { randomUUID } from 'node:crypto'; import { Env } from '@roomote/env'; import { getDeploymentJudgmentModelSelection, - recordLlmUsage, resolveModelProviderEnvValue, } from '@roomote/db/server'; import { @@ -20,6 +19,15 @@ import { NON_TASK_INFERENCE_SURFACES, resolveNonTaskHelperModel, } from './non-task-provider-usage'; +import { + JudgmentHttpError, + JudgmentResponseParseError, + trackJudgmentRequest, + type JudgmentRequestTracking, + type JudgmentUsageProvider, + type JudgmentUsageResponse, + type TrackedJudgmentResponse, +} from './judgment-usage-accounting'; /** * Optional judgment-model backend. A judgment model answers typed questions @@ -136,9 +144,7 @@ export type DecisionModelResolution = */ export type JudgmentBackend = | { provider: 'roomote'; url: string; apiKey: string | undefined } - | { provider: 'typesafe'; apiKey: string } - | { provider: 'openrouter'; apiKey: string } - | { provider: 'vercel'; apiKey: string }; + | { provider: Exclude; apiKey: string }; export type TypeSafeJudgmentTiming = { onRequestStarted?: () => void; @@ -330,256 +336,11 @@ function isValidAnswer(question: TypeSafeQuestion, answer: unknown): boolean { ); } -/** Carries the status separately so a caller can report it without the body. */ -class JudgmentHttpError extends Error { - constructor( - readonly status: number, - detail: string, - readonly headers?: Headers, - ) { - super( - `Judgment model request failed with HTTP ${status}${ - detail ? `: ${detail.slice(0, 200)}` : '' - }`, - ); - this.name = 'JudgmentHttpError'; - } -} - -class JudgmentResponseParseError extends Error { - constructor( - readonly status: number, - readonly headers: Headers, - ) { - super('Judgment model response was not valid JSON'); - this.name = 'JudgmentResponseParseError'; - } -} - -type JudgmentRequestRole = 'primary' | 'shadow'; - -type JudgmentUsageOutcome = - | 'success' - | 'http_error' - | 'timeout' - | 'transport_error' - | 'response_parse_error' - | 'response_error' - | 'validation_error'; - -type JudgmentResponse = { - body: Record; - status: number; - headers: Headers; -}; - -type JudgmentTracking = { - requestId: string; - provider: JudgmentBackend['provider']; - model: string; - role: JudgmentRequestRole; - primaryProvider?: JudgmentBackend['provider']; -}; - -type ParsedJudgmentUsage = { - modelId?: string; - inputTokens?: number; - outputTokens?: number; - totalTokens?: number; - costMicroUsd?: number; - pricingMetadata?: Record; - usageMetadataAvailable: boolean; - usageMetadataSource: 'response' | 'header' | 'none'; - metadataReadFailed: boolean; - missingUsageFields: string[]; -}; - -type TrackedJudgmentResponse = JudgmentResponse & { - finish: (outcome: JudgmentUsageOutcome) => void; -}; - type JudgmentRequestResult = { answers: Record | undefined; response: TrackedJudgmentResponse; }; -function nonNegativeNumber(value: unknown): number | undefined { - const number = - typeof value === 'number' - ? value - : typeof value === 'string' && value.trim() - ? Number(value) - : undefined; - - return number !== undefined && Number.isFinite(number) && number >= 0 - ? number - : undefined; -} - -function firstNonNegativeNumber(...values: unknown[]): number | undefined { - for (const value of values) { - const number = nonNegativeNumber(value); - if (number !== undefined) return number; - } - - return undefined; -} - -function readHeaderNumber( - headers: Headers | undefined, - names: readonly string[], -): number | undefined { - if (!headers) return undefined; - - for (const name of names) { - const number = nonNegativeNumber(headers.get(name)); - if (number !== undefined) return number; - } - - return undefined; -} - -function readHeaderString( - headers: Headers | undefined, - names: readonly string[], -): string | undefined { - if (!headers) return undefined; - - for (const name of names) { - const value = headers.get(name)?.trim(); - if (value) return value; - } - - return undefined; -} - -function parseJudgmentUsage( - response: JudgmentResponse | undefined, - metadataReadFailed = false, -): ParsedJudgmentUsage { - const body = response?.body ?? {}; - const usage = - asRecord(body.usage) ?? - asRecord(body.usageMetadata) ?? - asRecord(body.usage_metadata); - const headers = response?.headers; - - const responseInputTokens = firstNonNegativeNumber( - usage?.input_tokens, - usage?.prompt_tokens, - usage?.inputTokens, - usage?.promptTokens, - ); - const responseOutputTokens = firstNonNegativeNumber( - usage?.output_tokens, - usage?.completion_tokens, - usage?.outputTokens, - usage?.completionTokens, - ); - const responseTotalTokens = firstNonNegativeNumber( - usage?.total_tokens, - usage?.totalTokens, - ); - const responseCostMicroUsd = firstNonNegativeNumber( - usage?.cost_micro_usd, - usage?.costMicroUsd, - ); - const responseCostUsd = firstNonNegativeNumber( - usage?.cost, - usage?.cost_usd, - usage?.costUsd, - ); - - const headerInputTokens = readHeaderNumber(headers, [ - 'x-input-tokens', - 'x-prompt-tokens', - 'x-usage-input-tokens', - 'x-openrouter-input-tokens', - ]); - const headerOutputTokens = readHeaderNumber(headers, [ - 'x-output-tokens', - 'x-completion-tokens', - 'x-usage-output-tokens', - 'x-openrouter-output-tokens', - ]); - const headerTotalTokens = readHeaderNumber(headers, [ - 'x-total-tokens', - 'x-usage-total-tokens', - 'x-openrouter-total-tokens', - ]); - const headerCostMicroUsd = readHeaderNumber(headers, [ - 'x-cost-micro-usd', - 'x-usage-cost-micro-usd', - ]); - const headerCostUsd = readHeaderNumber(headers, [ - 'x-cost-usd', - 'x-cost', - 'x-provider-cost', - 'x-openrouter-cost', - ]); - - const inputTokens = responseInputTokens ?? headerInputTokens; - const outputTokens = responseOutputTokens ?? headerOutputTokens; - const totalTokens = responseTotalTokens ?? headerTotalTokens; - const costMicroUsd = - responseCostMicroUsd ?? - (responseCostUsd === undefined - ? (headerCostMicroUsd ?? - (headerCostUsd === undefined - ? undefined - : Math.round(headerCostUsd * 1_000_000))) - : Math.round(responseCostUsd * 1_000_000)); - const responseMetadataAvailable = - responseInputTokens !== undefined || - responseOutputTokens !== undefined || - responseTotalTokens !== undefined || - responseCostMicroUsd !== undefined || - responseCostUsd !== undefined; - const headerMetadataAvailable = - headerInputTokens !== undefined || - headerOutputTokens !== undefined || - headerTotalTokens !== undefined || - headerCostMicroUsd !== undefined || - headerCostUsd !== undefined; - const missingUsageFields: string[] = []; - - if (inputTokens === undefined) missingUsageFields.push('input_tokens'); - if (outputTokens === undefined) missingUsageFields.push('output_tokens'); - if (totalTokens === undefined) missingUsageFields.push('total_tokens'); - if (costMicroUsd === undefined) missingUsageFields.push('cost'); - - const costDetails = asRecord(usage?.cost_details ?? usage?.costDetails); - const responseModel = - typeof body.model === 'string' && body.model.trim() - ? body.model.trim() - : undefined; - - return { - ...(responseModel || - readHeaderString(headers, ['x-model', 'x-provider-model']) - ? { - modelId: - responseModel ?? - readHeaderString(headers, ['x-model', 'x-provider-model']), - } - : {}), - ...(inputTokens === undefined ? {} : { inputTokens }), - ...(outputTokens === undefined ? {} : { outputTokens }), - ...(totalTokens === undefined ? {} : { totalTokens }), - ...(costMicroUsd === undefined ? {} : { costMicroUsd }), - ...(costDetails ? { pricingMetadata: { costDetails } } : {}), - usageMetadataAvailable: - responseMetadataAvailable || headerMetadataAvailable, - usageMetadataSource: responseMetadataAvailable - ? 'response' - : headerMetadataAvailable - ? 'header' - : 'none', - metadataReadFailed, - missingUsageFields, - }; -} - function judgmentModelForBackend(backend: JudgmentBackend): string { switch (backend.provider) { case 'roomote': @@ -593,112 +354,10 @@ function judgmentModelForBackend(backend: JudgmentBackend): string { } } -function classifyJudgmentRequestError(error: unknown): { - outcome: JudgmentUsageOutcome; - status: number | null; - headers?: Headers; - metadataReadFailed: boolean; -} { - if (error instanceof JudgmentHttpError) { - return { - outcome: 'http_error', - status: error.status, - headers: error.headers, - metadataReadFailed: false, - }; - } - - if (error instanceof JudgmentResponseParseError) { - return { - outcome: 'response_parse_error', - status: error.status, - headers: error.headers, - metadataReadFailed: true, - }; - } - - if ( - error instanceof Error && - (error.name === 'TimeoutError' || error.name === 'AbortError') - ) { - return { - outcome: 'timeout', - status: null, - metadataReadFailed: false, - }; - } - - return { - outcome: 'transport_error', - status: null, - metadataReadFailed: false, - }; -} - -function recordJudgmentUsage(input: { - tracking: JudgmentTracking; - startedAt: number; - response?: JudgmentResponse; - status: number | null; - outcome: JudgmentUsageOutcome; - headers?: Headers; - metadataReadFailed?: boolean; -}): void { - const response = - input.response ?? - (input.headers - ? { body: {}, status: input.status ?? 0, headers: input.headers } - : undefined); - const usage = parseJudgmentUsage(response, input.metadataReadFailed ?? false); - const eventKey = `judgment-model:${input.tracking.provider}:${input.tracking.role}:${input.tracking.requestId}`; - const details: Record = { - surface: NON_TASK_INFERENCE_SURFACES.judgmentModel, - requestRole: input.tracking.role, - status: input.status, - outcome: input.outcome, - latencyMs: Math.max(0, Date.now() - input.startedAt), - usageMetadataAvailable: usage.usageMetadataAvailable, - usageMetadataSource: usage.usageMetadataSource, - metadataReadFailed: usage.metadataReadFailed, - missingUsageFields: usage.missingUsageFields, - ...(input.tracking.primaryProvider - ? { primaryProvider: input.tracking.primaryProvider } - : {}), - }; - - const persist = async () => { - await recordLlmUsage({ - source: NON_TASK_INFERENCE_SURFACES.judgmentModel, - usageType: 'inference', - eventKey, - providerId: input.tracking.provider, - modelId: usage.modelId ?? input.tracking.model, - inputTokens: usage.inputTokens ?? null, - outputTokens: usage.outputTokens ?? null, - totalTokens: usage.totalTokens ?? 0, - contextTokens: usage.inputTokens ?? 0, - costMicroUsd: usage.costMicroUsd ?? null, - costSource: - usage.costMicroUsd === undefined ? 'missing' : 'provider_response', - ...(usage.pricingMetadata - ? { pricingMetadata: usage.pricingMetadata } - : {}), - details, - }); - }; - - void persist().catch(() => { - // Usage telemetry must never change judgment behavior. - console.warn( - `[JudgmentUsage] Failed to record ${input.tracking.provider} ${input.tracking.role} usage`, - ); - }); -} - async function postJson( url: string, init: { headers: Record; body: unknown; timeoutMs: number }, -): Promise { +): Promise { const response = await fetch(url, { method: 'POST', headers: { ...init.headers, 'Content-Type': 'application/json' }, @@ -728,40 +387,9 @@ async function postJson( async function postTrackedJson( url: string, init: { headers: Record; body: unknown; timeoutMs: number }, - tracking: JudgmentTracking, + tracking: JudgmentRequestTracking, ): Promise { - const startedAt = Date.now(); - - try { - const response = await postJson(url, init); - let finished = false; - - return { - ...response, - finish: (outcome) => { - if (finished) return; - finished = true; - recordJudgmentUsage({ - tracking, - startedAt, - response, - status: response.status, - outcome, - }); - }, - }; - } catch (error) { - const failure = classifyJudgmentRequestError(error); - recordJudgmentUsage({ - tracking, - startedAt, - status: failure.status, - outcome: failure.outcome, - headers: failure.headers, - metadataReadFailed: failure.metadataReadFailed, - }); - throw error; - } + return trackJudgmentRequest(tracking, () => postJson(url, init)); } async function requestNativeDecisions( @@ -770,7 +398,7 @@ async function requestNativeDecisions( questions: Record, timeoutMs: number, options: { url: string; model: string }, - tracking: JudgmentTracking, + tracking: JudgmentRequestTracking, ): Promise { const response = await postTrackedJson( options.url, @@ -795,7 +423,7 @@ async function requestRoomoteDecisions( state: unknown, questions: Record, timeoutMs: number, - tracking: JudgmentTracking, + tracking: JudgmentRequestTracking, ): Promise { // The upstream reports probabilities; confidence is derived here the same // way it is for OpenRouter so every backend yields one answer shape. @@ -874,7 +502,7 @@ async function requestVercelGateway( state: unknown, questions: Record, timeoutMs: number, - tracking: JudgmentTracking, + tracking: JudgmentRequestTracking, ): Promise { const response = await postTrackedJson( VERCEL_AI_GATEWAY_EVALUATION_URL, @@ -985,7 +613,7 @@ export async function evaluateTypeSafeJudgments< ? DEFAULT_ROOMOTE_TIMEOUT_MS : DEFAULT_TYPESAFE_TIMEOUT_MS); const requestId = randomUUID(); - const tracking: JudgmentTracking = { + const tracking: JudgmentRequestTracking = { requestId, provider: backend.provider, model: judgmentModelForBackend(backend),