From e49bb118781a2f9c4e5540f1203fc3292156dfb1 Mon Sep 17 00:00:00 2001 From: teamchong <25894545+teamchong@users.noreply.github.com> Date: Thu, 2 Jul 2026 23:07:37 -0400 Subject: [PATCH] feat(telemetry): log stop_reason + safety-flag per request --- src/core/proxy.ts | 82 +++++++++++++++++--- src/core/tracker.ts | 17 +++++ tests/proxy-usage.test.ts | 156 ++++++++++++++++++++++++++++++++++++++ tests/tracker.test.ts | 38 ++++++++++ 4 files changed, 282 insertions(+), 11 deletions(-) diff --git a/src/core/proxy.ts b/src/core/proxy.ts index 9788f4f..05525e1 100644 --- a/src/core/proxy.ts +++ b/src/core/proxy.ts @@ -48,6 +48,12 @@ export interface ProxyEvent { info?: TransformInfo; /** Usage block from Anthropic's response — input/output/cache tokens. */ usage?: Usage; + /** Model stop reason from the response ("end_turn", "tool_use", "max_tokens", + * "refusal", …). "refusal" = safety classifier fired on the transformed request — + * scorers must fail cost comparisons on refusal rows (refusals emit almost no + * output and would otherwise look "cheaper"). OpenAI finish_reason ("stop", + * "length", "content_filter", …) is normalized into the same field. */ + stopReason?: string; error?: string; /** First ~2 KiB of the upstream 4xx body (not captured on 2xx or 5xx). */ errorBody?: string; @@ -112,7 +118,7 @@ export interface OutputMeasurement { function processSseEvent( block: string, m: OutputMeasurement, - state: { usage: Usage | undefined }, + state: { usage: Usage | undefined; stopReason: string | undefined }, ): void { // Parse `event:` + `data:` lines; continuation data: lines concatenate per SSE spec. let event = ''; @@ -136,11 +142,25 @@ function processSseEvent( // OpenAI Responses API streams usage nested under `response` on the terminal // `response.completed` (or `.incomplete`) event — not at the top level. if (event === 'response.completed' || event === 'response.incomplete') { - const resp = obj.response as { usage?: unknown } | undefined; + const resp = obj.response as + | { usage?: unknown; incomplete_details?: { reason?: unknown } } + | undefined; const respUsage = normalizeUsage(resp?.usage); if (respUsage) state.usage = respUsage; + // Responses API has no stop_reason; normalize the terminal status/reason instead. + const reason = resp?.incomplete_details?.reason; + state.stopReason = typeof reason === 'string' ? reason + : event === 'response.incomplete' ? 'incomplete' : 'stop'; } measureOpenAIChoices(obj, m); + // OpenAI chat chunks: the final chunk carries choices[].finish_reason (earlier chunks ship null). + const choices = obj.choices; + if (Array.isArray(choices)) { + for (const c of choices) { + const fr = (c as { finish_reason?: unknown } | undefined)?.finish_reason; + if (typeof fr === 'string') state.stopReason = fr; + } + } if (event === 'message_start') { const msg = obj.message as { usage?: Usage } | undefined; @@ -161,6 +181,9 @@ function processSseEvent( m.toolUseChars += d.partial_json.length; } } else if (event === 'message_delta') { + // Anthropic ships the final stop_reason here ("end_turn", "refusal", …). + const d = obj.delta as { stop_reason?: unknown } | undefined; + if (typeof d?.stop_reason === 'string') state.stopReason = d.stop_reason; // Authoritative final output_tokens; merge over message_start (which ships output_tokens: 1). const u = obj.usage as Partial | undefined; if (u) { @@ -260,6 +283,30 @@ function measureFromMessageJson(j: unknown): OutputMeasurement { return m; } +/** Stop reason from a non-streaming response JSON: Anthropic `stop_reason`, + * OpenAI chat `choices[].finish_reason`, Responses `incomplete_details.reason`. */ +function readStopReasonFromJson(j: unknown): string | undefined { + if (!j || typeof j !== 'object') return undefined; + const obj = j as { + stop_reason?: unknown; + choices?: unknown; + status?: unknown; + incomplete_details?: { reason?: unknown }; + }; + if (typeof obj.stop_reason === 'string') return obj.stop_reason; + if (Array.isArray(obj.choices)) { + for (const c of obj.choices) { + const fr = (c as { finish_reason?: unknown } | undefined)?.finish_reason; + if (typeof fr === 'string') return fr; + } + } + if (obj.status === 'incomplete') { + const reason = obj.incomplete_details?.reason; + return typeof reason === 'string' ? reason : 'incomplete'; + } + return undefined; +} + /** * Tee the response body to extract usage + output measurement without blocking the client. * Streams are scanned to EOF (final output_tokens is in message_delta; redacted_thinking @@ -270,6 +317,7 @@ function teeForUsage(res: Response): { usagePromise: Promise; errorBodyPromise: Promise; measurementPromise: Promise; + stopReasonPromise: Promise; } { // No body at all: nothing to extract on either path. if (!res.body) { @@ -278,6 +326,7 @@ function teeForUsage(res: Response): { usagePromise: Promise.resolve(undefined), errorBodyPromise: Promise.resolve(undefined), measurementPromise: Promise.resolve(undefined), + stopReasonPromise: Promise.resolve(undefined), }; } // 4xx: tee for the error body but skip usage scanning entirely. @@ -313,6 +362,7 @@ function teeForUsage(res: Response): { usagePromise: Promise.resolve(undefined), errorBodyPromise, measurementPromise: Promise.resolve(undefined), + stopReasonPromise: Promise.resolve(undefined), }; } // 5xx: skip both (the host already synthesizes an error message). @@ -322,15 +372,17 @@ function teeForUsage(res: Response): { usagePromise: Promise.resolve(undefined), errorBodyPromise: Promise.resolve(undefined), measurementPromise: Promise.resolve(undefined), + stopReasonPromise: Promise.resolve(undefined), }; } const ct = (res.headers.get('content-type') ?? '').toLowerCase(); const [forClient, forUs] = res.body.tee(); - // Single read loop resolves both; exposed as separate promises for call-site readability. + // Single read loop resolves all three; exposed as separate promises for call-site readability. const scanResult = (async (): Promise<{ usage: Usage | undefined; measurement: OutputMeasurement | undefined; + stopReason: string | undefined; }> => { const reader = forUs.getReader(); const decoder = new TextDecoder(); @@ -345,7 +397,10 @@ function teeForUsage(res: Response): { toolUseChars: 0, redactedBlockCount: 0, }; - const state: { usage: Usage | undefined } = { usage: undefined }; + const state: { usage: Usage | undefined; stopReason: string | undefined } = { + usage: undefined, + stopReason: undefined, + }; while (true) { const { done, value } = await reader.read(); if (done) break; @@ -360,7 +415,7 @@ function teeForUsage(res: Response): { } buf += decoder.decode(); if (buf.trim().length > 0) processSseEvent(buf, m, state); // trailing partial event - return { usage: state.usage, measurement: m }; + return { usage: state.usage, measurement: m, stopReason: state.stopReason }; } if (ct.includes('application/json')) { @@ -376,9 +431,10 @@ function teeForUsage(res: Response): { return { usage: normalizeUsage(j?.usage), measurement: measureFromMessageJson(j), + stopReason: readStopReasonFromJson(j), }; } catch { - return { usage: undefined, measurement: undefined }; + return { usage: undefined, measurement: undefined, stopReason: undefined }; } } } catch { @@ -393,7 +449,7 @@ function teeForUsage(res: Response): { } catch { /* ignore */ } - return { usage: undefined, measurement: undefined }; + return { usage: undefined, measurement: undefined, stopReason: undefined }; })(); return { @@ -405,6 +461,7 @@ function teeForUsage(res: Response): { usagePromise: scanResult.then((s) => s.usage), errorBodyPromise: Promise.resolve(undefined), measurementPromise: scanResult.then((s) => s.measurement), + stopReasonPromise: scanResult.then((s) => s.stopReason), }; } @@ -573,6 +630,7 @@ export function createProxy(config: ProxyConfig = {}) { usage?: Usage, errorBody?: string, measurement?: OutputMeasurement, + stopReason?: string, ): void => { const is4xx = status >= 400 && status < 500; // Gzip body lazily (only on 4xx). Async IIFE keeps fire() synchronous. @@ -632,6 +690,7 @@ export function createProxy(config: ProxyConfig = {}) { reqBodySha8, reqBodyGz, measurement, + stopReason, }); }; void finalize(); @@ -761,16 +820,17 @@ export function createProxy(config: ProxyConfig = {}) { const firstByteMs = Date.now() - t0; // Tee: client gets one side; scanner reads the other for usage/measurement/error body. - const { response: teed, usagePromise, errorBodyPromise, measurementPromise } = + const { response: teed, usagePromise, errorBodyPromise, measurementPromise, stopReasonPromise } = teeForUsage(upstreamRes); - // Fire event in background once all three resolve (measurementPromise shares the same stream read). + // Fire event in background once all four resolve (all share the same stream read). void Promise.all([ usagePromise.catch(() => undefined), errorBodyPromise.catch(() => undefined), measurementPromise.catch(() => undefined), - ]).then(([usage, errorBody, measurement]) => - fire(upstreamRes.status, info, undefined, firstByteMs, usage, errorBody, measurement), + stopReasonPromise.catch(() => undefined), + ]).then(([usage, errorBody, measurement, stopReason]) => + fire(upstreamRes.status, info, undefined, firstByteMs, usage, errorBody, measurement, stopReason), ); return new Response(teed.body, { diff --git a/src/core/tracker.ts b/src/core/tracker.ts index d48c84c..745349e 100644 --- a/src/core/tracker.ts +++ b/src/core/tracker.ts @@ -112,6 +112,16 @@ export interface TrackEvent { /** Server-side web search calls billed per-request (not per-token). */ web_search_requests?: number; + /** Model stop reason ("end_turn", "tool_use", "max_tokens", "refusal", …). + * OpenAI finish_reason ("stop", "length", "content_filter", …) lands in the same field. */ + stop_reason?: string; + /** True when the stop reason indicates a safety classifier fired ("refusal" / + * "content_filter"). Refusal rows emit almost no output and would otherwise + * read as "cheap" — scorers MUST fail cost comparisons on these rows, and a + * cluster of them after a transform change means the imaged prompt itself is + * tripping the classifier (see transform.ts reasoning_extraction notes). */ + safety_flagged?: boolean; + /** Ground-truth output chars measured by streaming the response body ourselves — independent of * usage.output_tokens. redacted_block_count_measured counts opaque server-encrypted blocks; * dashboard applies a low/mid/high estimate for those. Absent on non-scannable responses. */ @@ -299,9 +309,16 @@ export function toTrackEvent(ev: ProxyEvent): TrackEvent { if (m.redactedBlockCount > 0) out.redacted_block_count_measured = m.redactedBlockCount; } + if (ev.stopReason) { + out.stop_reason = ev.stopReason; + if (SAFETY_STOP_REASONS.has(ev.stopReason)) out.safety_flagged = true; + } return out; } +/** Stop reasons that mean a safety classifier fired (Anthropic / OpenAI spellings). */ +const SAFETY_STOP_REASONS = new Set(['refusal', 'content_filter']); + /** Writes one JSON line per event. Worker host uses console.log; Node host uses a file-backed variant. */ export class JsonLogTracker implements Tracker { constructor(private readonly sink: (line: string) => void = (s) => console.log(s)) {} diff --git a/tests/proxy-usage.test.ts b/tests/proxy-usage.test.ts index 8514d03..ad29a8d 100644 --- a/tests/proxy-usage.test.ts +++ b/tests/proxy-usage.test.ts @@ -1230,4 +1230,160 @@ describe('proxy usage extraction', () => { expect(captured).toBeDefined(); expect(captured!.measurement?.textChars).toBe(8); }); + + it('extracts stop_reason from the SSE message_delta event', async () => { + const sseBody = + 'event: message_start\n' + + 'data: {"type":"message_start","message":{"id":"x","type":"message","role":"assistant","content":[],"usage":{"input_tokens":10,"output_tokens":1}}}\n\n' + + 'event: content_block_delta\n' + + 'data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"ok"}}\n\n' + + 'event: message_delta\n' + + 'data: {"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null},"usage":{"output_tokens":9}}\n\n' + + 'event: message_stop\n' + + 'data: {"type":"message_stop"}\n\n'; + + const restore = mockUpstream( + () => + new Response(sseBody, { + status: 200, + headers: { 'content-type': 'text/event-stream' }, + }), + ); + + let captured: ProxyEvent | undefined; + const proxy = createProxy({ + transform: {}, + onRequest: (e) => { captured = e; }, + }); + + const res = await proxy( + new Request('http://localhost/v1/messages', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: SAMPLE_REQ_BODY, + }), + ); + await res.text(); + await new Promise((r) => setTimeout(r, 20)); + restore(); + + expect(captured).toBeDefined(); + expect(captured!.stopReason).toBe('end_turn'); + // message_delta output_tokens must win over message_start's placeholder 1. + expect(captured!.usage?.output_tokens).toBe(9); + }); + + it('extracts stop_reason "refusal" from a non-stream JSON response', async () => { + const restore = mockUpstream( + () => + new Response( + JSON.stringify({ + id: 'msg_r', + type: 'message', + role: 'assistant', + content: [], + stop_reason: 'refusal', + usage: { input_tokens: 5, output_tokens: 2 }, + }), + { status: 200, headers: { 'content-type': 'application/json' } }, + ), + ); + + let captured: ProxyEvent | undefined; + const proxy = createProxy({ + transform: {}, + onRequest: (e) => { captured = e; }, + }); + + const res = await proxy( + new Request('http://localhost/v1/messages', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: SAMPLE_REQ_BODY, + }), + ); + await res.text(); + await new Promise((r) => setTimeout(r, 20)); + restore(); + + expect(captured).toBeDefined(); + expect(captured!.stopReason).toBe('refusal'); + }); + + it('extracts OpenAI choices[].finish_reason from a JSON body', async () => { + const restore = mockUpstream( + () => + new Response( + JSON.stringify({ + id: 'chatcmpl-1', + object: 'chat.completion', + choices: [ + { + index: 0, + message: { role: 'assistant', content: '' }, + finish_reason: 'content_filter', + }, + ], + usage: { prompt_tokens: 11, completion_tokens: 3 }, + }), + { status: 200, headers: { 'content-type': 'application/json' } }, + ), + ); + + let captured: ProxyEvent | undefined; + const proxy = createProxy({ + transform: {}, + onRequest: (e) => { captured = e; }, + }); + + const res = await proxy( + new Request('http://localhost/v1/messages', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: SAMPLE_REQ_BODY, + }), + ); + await res.text(); + await new Promise((r) => setTimeout(r, 20)); + restore(); + + expect(captured).toBeDefined(); + expect(captured!.stopReason).toBe('content_filter'); + }); + + it('leaves stopReason undefined when the stream never ships one', async () => { + const sseBody = + 'event: message_start\n' + + 'data: {"type":"message_start","message":{"id":"x","type":"message","role":"assistant","content":[],"usage":{"input_tokens":10,"output_tokens":1}}}\n\n' + + 'event: message_stop\n' + + 'data: {"type":"message_stop"}\n\n'; + + const restore = mockUpstream( + () => + new Response(sseBody, { + status: 200, + headers: { 'content-type': 'text/event-stream' }, + }), + ); + + let captured: ProxyEvent | undefined; + const proxy = createProxy({ + transform: {}, + onRequest: (e) => { captured = e; }, + }); + + const res = await proxy( + new Request('http://localhost/v1/messages', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: SAMPLE_REQ_BODY, + }), + ); + await res.text(); + await new Promise((r) => setTimeout(r, 20)); + restore(); + + expect(captured).toBeDefined(); + expect(captured!.stopReason).toBeUndefined(); + }); }); diff --git a/tests/tracker.test.ts b/tests/tracker.test.ts index b8bbdec..07842b2 100644 --- a/tests/tracker.test.ts +++ b/tests/tracker.test.ts @@ -195,6 +195,44 @@ describe('toTrackEvent', () => { expect(out.cwd).toBeUndefined(); expect(out.input_tokens).toBeUndefined(); }); + + it('maps stopReason → stop_reason without flagging benign reasons', () => { + const out = toTrackEvent({ + method: 'POST', + path: '/v1/messages', + status: 200, + durationMs: 10, + stopReason: 'end_turn', + }); + expect(out.stop_reason).toBe('end_turn'); + expect(out.safety_flagged).toBeUndefined(); + }); + + it.each(['refusal', 'content_filter'])( + 'sets safety_flagged for stop reason %s', + (reason) => { + const out = toTrackEvent({ + method: 'POST', + path: '/v1/messages', + status: 200, + durationMs: 10, + stopReason: reason, + }); + expect(out.stop_reason).toBe(reason); + expect(out.safety_flagged).toBe(true); + }, + ); + + it('omits stop_reason entirely when the event has none', () => { + const out = toTrackEvent({ + method: 'POST', + path: '/v1/messages', + status: 200, + durationMs: 10, + }); + expect('stop_reason' in out).toBe(false); + expect('safety_flagged' in out).toBe(false); + }); }); describe('JsonLogTracker', () => {