From 3384a33b29c3d7ae92feb52fad9fe96f11bbb58a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?An=C4=B1l=20G=C3=BClero=C4=9Flu?= Date: Sun, 16 Aug 2026 09:10:40 +0300 Subject: [PATCH] fix(models): stop recording a client hang-up as a provider error MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A streamed call whose client disconnects lands in the same catch as a provider failure, so every user who pressed stop was written to the usage log as `status: 'error'` with `usage: {}`. Two things followed from that. The model error rate — and the alerting on top of it — tracked user behaviour rather than provider health, and the output the provider had already generated, and bills us for, was recorded as zero tokens. Only `cancel()` aborts the stream's signal, and only a closed response socket reaches `cancel()`, so an aborted signal is an exact test for "the client went away". Split that case out: - Log it as `cancelled`, a third usage status. Both the SQLite and Mongo aggregations match `success` and `error` by exact value, so the new state is additive: it stays in `totalCalls` and in the token sums without landing in either bucket. The status column has no CHECK constraint and the Model Hub badge already knows the value, so nothing needs migrating. - Record the output that was produced. Provider usage rides on the terminal chunk, which is exactly the chunk a cancelling client does not wait for, so when it never arrived the output is estimated from the streamed text with the same chars/4 rule the quota pre-flight uses. The entry carries `output_tokens_estimated` so a measured count and a derived one stay distinguishable in cost reporting — an estimate is defensible, quietly passing one off as measured is not. - Stop writing to the controller. It is already cancelled; the error frame and `[DONE]` the old path emitted threw a second exception into a stream with no reader left. Also run the streaming output guardrail on a cancelled stream. Those tokens reached the caller, and auditing only the completions that ran to the end made hanging up a way to skip the audit. Both paths now go through one `auditStreamedOutput`, tagged `chat.completions:stream:cancelled` when the caller left early so the two are separable in the evaluation log. Co-Authored-By: Claude Opus 5 (1M context) --- docs/guide/model-hub.md | 2 +- src/__tests__/unit/inference-service.test.ts | 86 ++++++++++++++ src/app/dashboard/models/[id]/page.tsx | 8 +- src/app/dashboard/overview/page.tsx | 4 +- src/lib/database/index.ts | 1 + src/lib/database/provider/types.base.ts | 13 ++- src/lib/database/sqlite/model.mixin.ts | 2 +- src/lib/services/models/inferenceService.ts | 111 +++++++++++++++---- src/lib/services/models/usageLogger.ts | 10 +- src/lib/services/usage/usageEvents.ts | 3 +- 10 files changed, 206 insertions(+), 34 deletions(-) diff --git a/docs/guide/model-hub.md b/docs/guide/model-hub.md index 6d74dec5..32cee0eb 100644 --- a/docs/guide/model-hub.md +++ b/docs/guide/model-hub.md @@ -61,7 +61,7 @@ The page is organised in tabs: - **Overview** (shown above) — performance over the chosen window, a ready-to-paste `curl` against the local runtime, recent requests, and the static metadata panel on the right (pricing, settings, status, timestamps). - **Playground** — a chat sandbox bound to this model. The system prompt and runtime settings come from the model definition, so the playground exactly mirrors what production traffic experiences. - **Configure** — read-only access to the underlying definition with edit-in-place affordances for the fields that don't require a redeploy (description, key, status, pricing). -- **Logs** — request-level entries with prompts, completions, token usage, and tool calls. Streams from the same store as [Agent Tracing](/guide/tracing). +- **Logs** — request-level entries with prompts, completions, token usage, and tool calls. Streams from the same store as [Agent Tracing](/guide/tracing). Each entry is `success`, `error`, or `cancelled`. **Cancelled** is a streamed call whose client disconnected before the answer finished — a user pressing stop, a closed tab, a request timing out on the caller's side. It is deliberately neither of the other two: it does not count towards the error rate, because nothing failed on our side or the provider's, and it does not count as a success, because the answer was never delivered. The output tokens generated up to that point are still recorded and billed — the provider charges for them — but when the provider never reported a count (it rides on the terminal chunk the client did not wait for) the figure is estimated from the streamed text and the log entry is marked `output_tokens_estimated`. - **Usage** — aggregated cost and call counts grouped by token, API token, and time bucket. The `Endpoint` panel exposes the canonical `curl` snippet: diff --git a/src/__tests__/unit/inference-service.test.ts b/src/__tests__/unit/inference-service.test.ts index 10cf08ed..3ca3adc0 100644 --- a/src/__tests__/unit/inference-service.test.ts +++ b/src/__tests__/unit/inference-service.test.ts @@ -26,6 +26,10 @@ vi.mock('@/lib/services/models/usageLogger', () => ({ logModelUsage: vi.fn().mockResolvedValue(undefined), })); +vi.mock('@/lib/services/guardrail', () => ({ + evaluateGuardrail: vi.fn().mockResolvedValue({ action: 'allow', findings: [] }), +})); + vi.mock('@/lib/services/models/openaiAdapter', async (importOriginal) => { const original = await importOriginal(); return { @@ -40,6 +44,7 @@ import { getModelByKey } from '@/lib/services/models/modelService'; import { buildModelRuntime } from '@/lib/services/models/runtimeService'; import { isSemanticCacheEnabled, lookupCache, storeInCache } from '@/lib/services/models/semanticCacheService'; import { logModelUsage } from '@/lib/services/models/usageLogger'; +import { evaluateGuardrail } from '@/lib/services/guardrail'; import { toOpenAIChatResponse, toOpenAIStreamChunk, @@ -501,6 +506,87 @@ describe('handleChatCompletion', () => { expect(invoke).not.toHaveBeenCalled(); }); + describe('client disconnects mid-stream', () => { + const startCancellableStream = async (modelOverrides = {}) => { + (getModelByKey as ReturnType).mockResolvedValue(makeLlmModel(modelOverrides)); + let aborted = false; + (buildModelRuntime as ReturnType).mockResolvedValue({ + runtime: { + createChatModel: vi.fn().mockResolvedValue({ + invoke: vi.fn(), + stream: vi.fn().mockImplementation((_messages, options) => { + options?.signal?.addEventListener('abort', () => { aborted = true; }); + return Promise.resolve((async function* () { + yield new AIMessageChunk({ content: 'partial answer' }); + // Stay open so the consumer's cancel lands mid-stream, then + // surface the abort the way a provider SDK does. + await new Promise((resolve) => { setTimeout(resolve, 30); }); + if (aborted) throw Object.assign(new Error('Aborted'), { name: 'AbortError' }); + yield new AIMessageChunk({ content: ' never delivered' }); + })()); + }), + }), + }, + }); + (toOpenAIStreamChunk as ReturnType).mockImplementation((chunk: { + content: string; + }) => ({ + id: 'chatcmpl-1', + object: 'chat.completion.chunk', + choices: [{ index: 0, delta: { content: chunk.content }, finish_reason: null }], + })); + + const result = await handleChatCompletion({ + ...BASE_PARAMS, + body: { messages: [] }, + stream: true, + }); + + const reader = result.stream!.getReader(); + await reader.read(); + await reader.read(); + await reader.cancel('client gone'); + // Let the aborted iterator unwind and the fire-and-forget log settle. + await new Promise((resolve) => { setTimeout(resolve, 80); }); + return (logModelUsage as ReturnType).mock.calls.at(-1); + }; + + it('records the call as cancelled, not as a provider error', async () => { + // Pressing stop is not an outage. Logging it as one drove the error rate + // and its alerting off user behaviour rather than provider health. + const call = await startCancellableStream(); + + expect(call?.[2]).toMatchObject({ status: 'cancelled', route: 'chat.completions' }); + expect(call?.[2].errorMessage).toBeUndefined(); + }); + + it('still bills the output the provider generated before the client left', async () => { + // The provider charges for what it produced; recording zero wrote it off. + const call = await startCancellableStream(); + + expect(call?.[2].usage.outputTokens).toBeGreaterThan(0); + expect(call?.[2].providerResponse).toMatchObject({ + cancelled: 'client_disconnected', + output_tokens_estimated: true, + }); + }); + + it('still audits the output guardrail over the text that was delivered', async () => { + // Text the caller already received has to be audited whether or not they + // stayed for the rest, otherwise hanging up is a way to skip the audit. + await startCancellableStream({ outputGuardrailKey: 'no-pii' }); + + expect(evaluateGuardrail).toHaveBeenCalledWith( + expect.objectContaining({ + guardrailKey: 'no-pii', + phase: 'output', + text: expect.stringContaining('partial answer'), + source: 'chat.completions:stream:cancelled', + }), + ); + }); + }); + it('forwards each frame as the provider produces it, without collecting the answer first', async () => { // The distinction that matters to a caller: do we relay the provider's // chunks as they land, or wait for the completion and replay it? Assert the diff --git a/src/app/dashboard/models/[id]/page.tsx b/src/app/dashboard/models/[id]/page.tsx index 77bf300b..5e8a484c 100644 --- a/src/app/dashboard/models/[id]/page.tsx +++ b/src/app/dashboard/models/[id]/page.tsx @@ -165,7 +165,7 @@ interface UsageLogDto { _id?: string; requestId?: string; route: string; - status: 'success' | 'error'; + status: 'success' | 'error' | 'cancelled'; latencyMs?: number; inputTokens: number; outputTokens: number; @@ -834,7 +834,7 @@ export default function ModelDetailPage() {
- + {selectedLog.latencyMs ? ( {Math.round(selectedLog.latencyMs)} ms @@ -1201,7 +1201,7 @@ function OverviewTab({ {l.latencyMs ? `${Math.round(l.latencyMs)}ms` : '—'} - + ))} @@ -2071,7 +2071,7 @@ function LogsTab({
- + {l.cacheHit === true ? ( cache ) : null} diff --git a/src/app/dashboard/overview/page.tsx b/src/app/dashboard/overview/page.tsx index 3bbb10fc..6a0aeabb 100644 --- a/src/app/dashboard/overview/page.tsx +++ b/src/app/dashboard/overview/page.tsx @@ -49,7 +49,7 @@ interface RecentActivity { type: string; service: string; endpoint: string; - status: 'success' | 'error'; + status: 'success' | 'error' | 'cancelled'; timestamp: string; } @@ -412,7 +412,7 @@ export default function DashboardOverviewPage() { {a.endpoint}
- + {a.relTime} diff --git a/src/lib/database/index.ts b/src/lib/database/index.ts index 94aeaf58..09230e04 100644 --- a/src/lib/database/index.ts +++ b/src/lib/database/index.ts @@ -172,6 +172,7 @@ export type { IAgentTracingEvent, IModel, IModelUsageLog, + ModelUsageStatus, IModelUsageAggregate, IUsageAttributionFields, UsageActorType, diff --git a/src/lib/database/provider/types.base.ts b/src/lib/database/provider/types.base.ts index 213f28f7..82c0363a 100644 --- a/src/lib/database/provider/types.base.ts +++ b/src/lib/database/provider/types.base.ts @@ -734,6 +734,17 @@ export interface IModelUsageCostSnapshot { totalCost?: number; } +/** + * `cancelled` is a call the client walked away from mid-stream — it is neither + * a completion nor a provider fault, and counting it as either is wrong: the + * error rate would blame us for a user pressing stop, and the success rate + * would claim an answer that was never delivered. Both aggregations match on + * the exact value, so a third state is additive: it stays in `totalCalls` and + * in the token sums (we are billed for what the provider generated) without + * landing in either bucket. + */ +export type ModelUsageStatus = 'success' | 'error' | 'cancelled'; + export interface IModelUsageLog extends IUsageAttributionFields { _id?: ObjectId | string; tenantId: string; @@ -742,7 +753,7 @@ export interface IModelUsageLog extends IUsageAttributionFields { modelId?: string; requestId: string; route: string; - status: 'success' | 'error'; + status: ModelUsageStatus; providerRequest: Record; providerResponse: Record; errorMessage?: string; diff --git a/src/lib/database/sqlite/model.mixin.ts b/src/lib/database/sqlite/model.mixin.ts index 2fbb3a9e..4f7d5f90 100644 --- a/src/lib/database/sqlite/model.mixin.ts +++ b/src/lib/database/sqlite/model.mixin.ts @@ -358,7 +358,7 @@ export function ModelMixin>(Base: modelId: r.modelId as string | undefined, requestId: r.requestId as string, route: r.route as string, - status: r.status as 'success' | 'error', + status: r.status as IModelUsageLog['status'], providerRequest: this.parseJson(r.providerRequest, {}), providerResponse: this.parseJson(r.providerResponse, {}), errorMessage: r.errorMessage as string | undefined, diff --git a/src/lib/services/models/inferenceService.ts b/src/lib/services/models/inferenceService.ts index 12e56e95..52ca77ae 100644 --- a/src/lib/services/models/inferenceService.ts +++ b/src/lib/services/models/inferenceService.ts @@ -1138,6 +1138,59 @@ export async function handleChatCompletion(params: { let outputLimitError: OutputTokenLimitError | null = null; const toolCalls: ToolCallPayload[] = []; + // A cancelled stream almost never carries provider usage: the counts + // ride on the terminal chunk, which is precisely the chunk the client + // did not stay for. Reporting zero there writes off output the + // provider generated and bills us for, so fall back to an estimate + // over the text we did stream — the same chars/4 rule the quota + // pre-flight uses. The log marks it `output_tokens_estimated` so a + // measured count and a guessed one stay distinguishable downstream. + // Input tokens are never guessed; they are reported or absent. + const partialUsageOnCancel = (): TokenUsage => { + const toolCallCount = toolCalls.length || undefined; + if (lastUsage) { + return { ...lastUsage, toolCalls: toolCallCount }; + } + + const streamed = guardrailContentToText( + (aggregatedChunk as { content?: unknown } | null)?.content, + ); + const outputTokens = streamed ? Math.ceil(streamed.length / 4) : 0; + return { + outputTokens, + totalTokens: outputTokens, + toolCalls: toolCallCount, + }; + }; + + // Output guardrail (streaming): the text has already reached the + // client, so this is a post-hoc audit — violations land in the + // evaluation log and alert metrics rather than blocking the stream. + // It runs on a cancelled stream too: those tokens were delivered, and + // auditing only completed answers would let a caller skip the audit by + // hanging up. + const auditStreamedOutput = (source: string) => { + if (!model.outputGuardrailKey) return; + + const streamedText = guardrailContentToText( + (aggregatedChunk as { content?: unknown } | null)?.content, + ); + if (!streamedText.trim()) return; + + fireAndForget('guardrail-stream-output-audit', async () => { + await evaluateGuardrail({ + tenantDbName, + tenantId: model.tenantId, + projectId, + guardrailKey: model.outputGuardrailKey!, + text: streamedText, + phase: 'output', + requestId, + source, + }); + }); + }; + try { // OpenAI opens every stream with the assistant role before any content. controller.enqueue( @@ -1273,30 +1326,44 @@ export async function handleChatCompletion(params: { }), ); - // Output guardrail (streaming): the response is already delivered, - // so this is a post-hoc audit — violations land in the evaluation - // log and alert metrics rather than blocking the stream. - if (model.outputGuardrailKey) { - const streamedText = guardrailContentToText( - (aggregatedChunk as { content?: unknown } | null)?.content, - ); - if (streamedText.trim()) { - fireAndForget('guardrail-stream-output-audit', async () => { - await evaluateGuardrail({ - tenantDbName, - tenantId: model.tenantId, - projectId, - guardrailKey: model.outputGuardrailKey!, - text: streamedText, - phase: 'output', - requestId, - source: 'chat.completions:stream', - }); - }); - } - } + auditStreamedOutput('chat.completions:stream'); } catch (error: unknown) { const latencyMs = Date.now() - startedAt; + + // Only `cancel()` aborts this signal, and only a closed response + // socket reaches `cancel()`. So an aborted signal means the client + // walked away — the upstream call we abort in response then throws + // out of the loop above. That is not a provider failure, and + // recording it as one blamed us for every user who pressed stop: + // the error rate rose, alerting fired, and the tokens the provider + // had already generated (and bills us for) were written as zero. + if (abortController.signal.aborted) { + fireAndForget('log-stream-cancelled', () => + logModelUsage(tenantDbName, model, { + requestId, + route: 'chat.completions', + status: 'cancelled', + providerRequest: sanitizeForLogging({ + model: modelKey, + messages: body.messages, + overrides, + stream: true, + }), + providerResponse: sanitizeForLogging({ + cancelled: 'client_disconnected', + partial: aggregatedChunk ?? { tool_calls: toolCalls }, + ...(lastUsage ? {} : { output_tokens_estimated: true }), + }), + latencyMs, + usage: partialUsageOnCancel(), + }), + ); + auditStreamedOutput('chat.completions:stream:cancelled'); + // Nothing is left to write to: the controller is already cancelled, + // and enqueueing on it throws. + return; + } + const normalizedError = normalizeInferenceError(error); const errorMessage = normalizedError.error.message; fireAndForget('log-stream-error', () => diff --git a/src/lib/services/models/usageLogger.ts b/src/lib/services/models/usageLogger.ts index a8df6d76..c621eb17 100644 --- a/src/lib/services/models/usageLogger.ts +++ b/src/lib/services/models/usageLogger.ts @@ -1,4 +1,10 @@ -import { getDatabase, IModel, IModelPricing, IModelUsageRouting } from '@/lib/database'; +import { + getDatabase, + IModel, + IModelPricing, + IModelUsageRouting, + ModelUsageStatus, +} from '@/lib/database'; import { recordUsageEvent, type UsageAttribution, @@ -104,7 +110,7 @@ export async function logModelUsage( payload: { requestId: string; route: string; - status: 'success' | 'error'; + status: ModelUsageStatus; providerRequest: unknown; providerResponse: unknown; errorMessage?: string; diff --git a/src/lib/services/usage/usageEvents.ts b/src/lib/services/usage/usageEvents.ts index 956f9f6e..3864c775 100644 --- a/src/lib/services/usage/usageEvents.ts +++ b/src/lib/services/usage/usageEvents.ts @@ -17,6 +17,7 @@ import { getRequestContext } from '@/lib/core/requestContext'; import type { IUsageAttributionFields, + ModelUsageStatus, UsageActorType, UsageSource, } from '@/lib/database'; @@ -74,7 +75,7 @@ export interface UsageEventInput { /** Overrides the actor-derived origin — e.g. 'tracing' for usage derived * from observability ingests rather than gateway-served calls. */ source?: UsageSource; - status?: 'success' | 'error'; + status?: ModelUsageStatus; latencyMs?: number; /** When `latencyMs` aggregates multiple calls (trace-derived usage), the * number of calls it covers — weights the rollup's latency average. */