diff --git a/.changeset/post-model-call-telemetry.md b/.changeset/post-model-call-telemetry.md new file mode 100644 index 0000000..0650204 --- /dev/null +++ b/.changeset/post-model-call-telemetry.md @@ -0,0 +1,11 @@ +--- +'@openrouter/agent': minor +--- + +Add a `PostModelCall` lifecycle hook and aggregate usage totals on `SessionEnd` — the telemetry primitives for tracing and benchmark consumers. + +`PostModelCall` fires once per completed model response, on **every** request the agent loop makes: the initial request, each tool-round follow-up, the empty-final retry, the `allowFinalResponse` final turn, and approval-resume requests. The payload carries `responseId` (the OpenRouter generation id, deep-linkable), `model`, `durationMs` (dispatch to fully materialized response, including stream consumption), `turnType` (`'initial' | 'resume' | 'tool_round' | 'final' | 'retry'`), `turnNumber`, and a normalized `usage` block (`inputTokens`, `outputTokens`, `totalTokens`, `cachedTokens`, `reasoningTokens`, `cost?`) when the server reported usage accounting. Purely observational: handlers cannot mutate or block. + +`SessionEnd` now carries an optional `totalUsage` aggregate (`modelCalls` plus the summed usage fields, with `cost` present when any call reported one) whenever at least one model call completed during the run. + +New exported types: `PostModelCallPayload`, `ModelCallUsage`, `SessionUsageTotals`. diff --git a/packages/agent/README.md b/packages/agent/README.md index 4fd7f6b..09c9bf7 100644 --- a/packages/agent/README.md +++ b/packages/agent/README.md @@ -313,7 +313,8 @@ const result = callModel(client, { model, input, tools, hooks }); | `PermissionRequest` | When a tool requires approval, before pausing for the human gate | `decision: 'allow'` skips the gate (the tool runs once via the normal round), `'deny'` synthesizes a rejected result without executing, `'ask_user'` (default) falls through to the approval flow. Last handler wins. Payload includes a `riskLevel` derived from the approval gate's shape (`'high'` for tool- or call-level functions, `'medium'` for blanket `true`) | | `Stop` | When a `stopWhen` condition halts the loop (`reason: 'max_turns'`) | `forceResume: true` continues the loop (capped at 3 consecutive overrides without tool progress — a bare `forceResume` that changes no state will typically re-trigger the stop condition immediately and burn through the cap, so pair it with `appendPrompt` or external state the stop condition observes); `appendPrompt` injects a user message for the next turn (honored independently of `forceResume`). Blocked/rejected tool outputs count as progress for the cap: the model receives that feedback, and each round costs a full request, so the loop cannot spin hot | | `SessionStart` | Once per run, before the initial request. `config` summarizes the session (`hasTools`, `hasApproval`, `hasState`) | none (void) | -| `SessionEnd` | Once per run, on every exit path — completion, approval pause, interruption, error, and the no-tools streaming paths. `reason` is `'complete' \| 'error' \| 'max_turns' \| 'user'` | none (void) | +| `SessionEnd` | Once per run, on every exit path — completion, approval pause, interruption, error, and the no-tools streaming paths. `reason` is `'complete' \| 'error' \| 'max_turns' \| 'user'`. When at least one model call completed, `totalUsage` aggregates tokens/cost across all of them (`modelCalls`, `inputTokens`, `outputTokens`, `totalTokens`, `cachedTokens`, `reasoningTokens`, and `cost` when the server reported it) | none (void) | +| `PostModelCall` | Once per completed model response, on **every** request the loop makes — initial, each tool-round follow-up, the empty-final retry, the `allowFinalResponse` final turn, and approval-resume requests. Payload: `responseId` (the OpenRouter generation id), `model`, `durationMs` (dispatch → fully materialized response, including stream consumption), `turnType` (`'initial' \| 'resume' \| 'tool_round' \| 'final' \| 'retry'`), `turnNumber`, and `usage` (`inputTokens`, `outputTokens`, `totalTokens`, `cachedTokens`, `reasoningTokens`, `cost?`) when the server reported usage accounting. Purely observational — the telemetry primitive for tracing/benchmark consumers: one span per model call | none (void) | Notes on lifecycle pairing: `SessionEnd` only fires when a matching `SessionStart` succeeded, and at most once per run. Pending async hook work is @@ -321,6 +322,13 @@ always drained on teardown — including on paths that skip `SessionStart`, such as resuming from a tool approval. A throwing `SessionEnd` handler never masks the run's original error (teardown failures are logged as warnings). +On no-tools streaming paths the initial response is only materialized when the +stream is consumed, so `PostModelCall` for that response fires during session +teardown (before `SessionEnd`). A stream that fails or errors before producing +a materialized response emits no `PostModelCall`; a `response.incomplete` +response (e.g. truncated at `max_output_tokens`) **does** emit — it carries a +real generation id and consumed tokens. Note `usage.cost` is only present when +the request had usage accounting enabled server-side. Every handler receives `(payload, context)` — `context` carries the `sessionId` (the single source of session identity; payloads do not repeat it), the `hookName`, and an `AbortSignal` for cooperative cancellation. The diff --git a/packages/agent/src/index.ts b/packages/agent/src/index.ts index 5269739..9bc2350 100644 --- a/packages/agent/src/index.ts +++ b/packages/agent/src/index.ts @@ -127,14 +127,17 @@ export type { HooksManagerOptions, InlineHookConfig, LifecycleHookContext, + ModelCallUsage, PermissionRequestPayload, PermissionRequestResult, + PostModelCallPayload, PostToolUseFailurePayload, PostToolUsePayload, PreToolUsePayload, PreToolUseResult, SessionEndPayload, SessionStartPayload, + SessionUsageTotals, StopPayload, StopResult, ToolMatcher, diff --git a/packages/agent/src/lib/hooks-schemas.ts b/packages/agent/src/lib/hooks-schemas.ts index 6546e78..461b8ae 100644 --- a/packages/agent/src/lib/hooks-schemas.ts +++ b/packages/agent/src/lib/hooks-schemas.ts @@ -18,6 +18,7 @@ export const HookName = { PermissionRequest: 'PermissionRequest', SessionStart: 'SessionStart', SessionEnd: 'SessionEnd', + PostModelCall: 'PostModelCall', } as const; export type HookName = (typeof HookName)[keyof typeof HookName]; @@ -113,8 +114,22 @@ export const SessionStartPayloadSchema = z4.object({ config: z4.record(z4.string(), z4.unknown()).optional(), }); +const ModelCallUsageSchema = z4.object({ + inputTokens: z4.number(), + outputTokens: z4.number(), + totalTokens: z4.number(), + cachedTokens: z4.number(), + reasoningTokens: z4.number(), + cost: z4.number().optional(), +}); + +export type ModelCallUsage = Readonly>; + export type SessionStartPayload = Readonly>; +const SessionUsageTotalsSchema = ModelCallUsageSchema.extend({ + modelCalls: z4.number(), +}); export const SessionEndPayloadSchema = z4.object({ reason: z4.enum([ 'user', @@ -122,9 +137,28 @@ export const SessionEndPayloadSchema = z4.object({ 'max_turns', 'complete', ]), + totalUsage: SessionUsageTotalsSchema.optional(), +}); + +export const PostModelCallPayloadSchema = z4.object({ + sessionId: z4.string(), + responseId: z4.string(), + model: z4.string(), + durationMs: z4.number(), + turnType: z4.enum([ + 'initial', + 'resume', + 'tool_round', + 'final', + 'retry', + ]), + turnNumber: z4.number(), + usage: ModelCallUsageSchema.optional(), }); export type SessionEndPayload = Readonly>; +export type PostModelCallPayload = Readonly>; +export type SessionUsageTotals = Readonly>; //#endregion @@ -238,6 +272,10 @@ export const BUILT_IN_HOOKS: Record = { payload: SessionEndPayloadSchema, result: VoidResultSchema, }, + PostModelCall: { + payload: PostModelCallPayloadSchema, + result: VoidResultSchema, + }, }; export const BUILT_IN_HOOK_NAMES = new Set(Object.keys(BUILT_IN_HOOKS)); diff --git a/packages/agent/src/lib/hooks-types.ts b/packages/agent/src/lib/hooks-types.ts index a79d622..3383b37 100644 --- a/packages/agent/src/lib/hooks-types.ts +++ b/packages/agent/src/lib/hooks-types.ts @@ -2,6 +2,7 @@ import * as z4 from 'zod/v4'; import type { PermissionRequestPayload, PermissionRequestResult, + PostModelCallPayload, PostToolUseFailurePayload, PostToolUsePayload, PreToolUsePayload, @@ -151,14 +152,17 @@ export interface HooksManagerOptions { // and the runtime validation is structurally impossible). Re-exported here // so the public import surface is unchanged. export type { + ModelCallUsage, PermissionRequestPayload, PermissionRequestResult, + PostModelCallPayload, PostToolUseFailurePayload, PostToolUsePayload, PreToolUsePayload, PreToolUseResult, SessionEndPayload, SessionStartPayload, + SessionUsageTotals, StopPayload, StopResult, UserPromptSubmitPayload, @@ -206,6 +210,11 @@ export interface BuiltInHookDefinitions { /** Observation-only hook: handlers have no meaningful result. */ result: undefined; }; + PostModelCall: { + payload: PostModelCallPayload; + /** Observation-only hook: handlers have no meaningful result. */ + result: undefined; + }; } //#endregion diff --git a/packages/agent/src/lib/model-result.ts b/packages/agent/src/lib/model-result.ts index 4d3818e..7a0402e 100644 --- a/packages/agent/src/lib/model-result.ts +++ b/packages/agent/src/lib/model-result.ts @@ -17,6 +17,7 @@ import { updateState, } from './conversation-state.js'; import type { HooksManager } from './hooks-manager.js'; +import type { ModelCallUsage, PostModelCallPayload } from './hooks-types.js'; import { applyNextTurnParamsToRequest, executeNextTurnParamsFunctions, @@ -128,6 +129,27 @@ function isEventStream(value: unknown): value is EventStream(); + // Telemetry for the PostModelCall hook: the initial/resume request is + // dispatched in initStream but its response is materialized later (stream + // consumption), so the dispatch time and turn labeling are parked here + // until a completed response is available. Cleared on emit. + private pendingModelCall: + | { + startedAt: number; + turnType: 'initial' | 'resume'; + turnNumber: number; + } + | undefined; + // Running aggregate across every PostModelCall emit; surfaced as + // SessionEnd.totalUsage. + private readonly sessionUsage = { + modelCalls: 0, + inputTokens: 0, + outputTokens: 0, + totalTokens: 0, + cachedTokens: 0, + reasoningTokens: 0, + cost: 0, + hasCost: false, + }; constructor(options: GetResponseOptions) { this.options = options; @@ -609,6 +654,28 @@ export class ModelResult< ); } + /** + * Materialize a betaResponsesSend result value into a completed response: + * streaming values are consumed to completion (through the turn + * broadcaster when one is attached), non-streaming values pass through. + */ + private async materializeResponse( + value: unknown, + turnNumber: number, + ): Promise { + if (isEventStream(value)) { + const stream = new ReusableReadableStream(value); + if (this.turnBroadcaster) { + return this.pipeAndConsumeStream(stream, turnNumber); + } + return consumeStreamForCompletion(stream); + } + if (this.isNonStreamingResponse(value)) { + return value; + } + throw new Error('Unexpected response type from API'); + } + // ========================================================================= // Extracted Helper Methods for executeToolsIfNeeded // ========================================================================= @@ -625,7 +692,9 @@ export class ModelResult< return this.finalResponse; } if (this.reusableStream) { - return consumeStreamForCompletion(this.reusableStream); + const response = await consumeStreamForCompletion(this.reusableStream); + await this.emitPendingModelCallOnce(response); + return response; } throw new Error('Neither stream nor response initialized'); } @@ -1087,11 +1156,82 @@ export class ModelResult< 'SessionEnd', { reason, + ...(this.sessionUsage.modelCalls > 0 && { + totalUsage: { + modelCalls: this.sessionUsage.modelCalls, + inputTokens: this.sessionUsage.inputTokens, + outputTokens: this.sessionUsage.outputTokens, + totalTokens: this.sessionUsage.totalTokens, + cachedTokens: this.sessionUsage.cachedTokens, + reasoningTokens: this.sessionUsage.reasoningTokens, + ...(this.sessionUsage.hasCost && { + cost: this.sessionUsage.cost, + }), + }, + }), + }, + this.hookEmitContext(), + ); + } + + /** + * Emit PostModelCall for a completed model response and fold its usage + * into the session aggregate. One emit per materialized response. + */ + private async emitPostModelCall( + response: models.OpenResponsesResult, + startedAt: number, + turnType: PostModelCallPayload['turnType'], + turnNumber: number, + ): Promise { + if (!this.hooksManager) { + return; + } + const usage = extractModelCallUsage(response.usage); + this.sessionUsage.modelCalls++; + if (usage) { + this.sessionUsage.inputTokens += usage.inputTokens; + this.sessionUsage.outputTokens += usage.outputTokens; + this.sessionUsage.totalTokens += usage.totalTokens; + this.sessionUsage.cachedTokens += usage.cachedTokens; + this.sessionUsage.reasoningTokens += usage.reasoningTokens; + if (usage.cost !== undefined) { + this.sessionUsage.cost += usage.cost; + this.sessionUsage.hasCost = true; + } + } + await this.hooksManager.emit( + 'PostModelCall', + { + sessionId: this.currentState?.id ?? '', + responseId: response.id, + model: response.model ?? '', + durationMs: performance.now() - startedAt, + turnType, + turnNumber, + ...(usage && { + usage, + }), }, this.hookEmitContext(), ); } + /** + * Emit the parked initial/resume PostModelCall once its response has been + * materialized. No-ops when nothing is parked (e.g. the non-streaming + * branch already emitted). Safe to call from multiple materialization + * sites; the first wins. + */ + private async emitPendingModelCallOnce(response: models.OpenResponsesResult): Promise { + const pending = this.pendingModelCall; + if (!pending) { + return; + } + this.pendingModelCall = undefined; + await this.emitPostModelCall(response, pending.startedAt, pending.turnType, pending.turnNumber); + } + /** * Emit the Stop hook when a stopWhen condition halts the loop, and decide * whether the loop should resume. @@ -1159,6 +1299,27 @@ export class ModelResult< return; } try { + // Materialize the parked initial-call telemetry when the stream fully + // completed (the retained buffer replays without touching the source). + // A failed/errored stream skips it: no materialized response exists. + // (`response.incomplete` responses DO emit — they are materialized, have + // a generation id, and consumed tokens.) + // Isolated try: a telemetry failure (e.g. a buffer without a completion + // event, or a throwing strict-mode handler) must not skip SessionEnd or + // the drain below — those are contractual on every exit path. + try { + if (this.pendingModelCall) { + if (this.finalResponse) { + await this.emitPendingModelCallOnce(this.finalResponse); + } else if (this.reusableStream?.isComplete) { + await this.emitPendingModelCallOnce( + await consumeStreamForCompletion(this.reusableStream), + ); + } + } + } catch (telemetryError) { + console.warn('[PostModelCall] error during stream teardown:', telemetryError); + } await this.emitSessionEndOnce(reason); await this.hooksManager.drain(); } catch (teardownError) { @@ -1958,6 +2119,7 @@ export class ModelResult< stream: true, }; + const startedAt = performance.now(); const newResult = await betaResponsesSend( this.options.client, { @@ -1970,21 +2132,9 @@ export class ModelResult< throw newResult.error; } - // Handle streaming or non-streaming response - const value = newResult.value; - if (isEventStream(value)) { - const followUpStream = new ReusableReadableStream(value); - - if (this.turnBroadcaster) { - return this.pipeAndConsumeStream(followUpStream, turnNumber); - } - - return consumeStreamForCompletion(followUpStream); - } - if (this.isNonStreamingResponse(value)) { - return value; - } - throw new Error('Unexpected response type from API'); + const response = await this.materializeResponse(newResult.value, turnNumber); + await this.emitPostModelCall(response, startedAt, 'tool_round', turnNumber); + return response; } /** @@ -2050,6 +2200,7 @@ export class ModelResult< }; this.resolvedRequest = finalRequest; + const startedAt = performance.now(); const result = await betaResponsesSend( this.options.client, { @@ -2062,18 +2213,9 @@ export class ModelResult< throw result.error; } - const value = result.value; - if (isEventStream(value)) { - const stream = new ReusableReadableStream(value); - if (this.turnBroadcaster) { - return this.pipeAndConsumeStream(stream, turnNumber); - } - return consumeStreamForCompletion(stream); - } - if (this.isNonStreamingResponse(value)) { - return value; - } - throw new Error('Unexpected response type from API'); + const response = await this.materializeResponse(result.value, turnNumber); + await this.emitPostModelCall(response, startedAt, 'final', turnNumber); + return response; } /** @@ -2126,6 +2268,7 @@ export class ModelResult< stream: true, }; + const startedAt = performance.now(); const newResult = await betaResponsesSend( this.options.client, { @@ -2138,20 +2281,9 @@ export class ModelResult< throw newResult.error; } - const value = newResult.value; - if (isEventStream(value)) { - const followUpStream = new ReusableReadableStream(value); - - if (this.turnBroadcaster) { - return this.pipeAndConsumeStream(followUpStream, turnNumber); - } - - return consumeStreamForCompletion(followUpStream); - } - if (this.isNonStreamingResponse(value)) { - return value; - } - throw new Error('Unexpected response type from API'); + const response = await this.materializeResponse(newResult.value, turnNumber); + await this.emitPostModelCall(response, startedAt, 'retry', turnNumber); + return response; } /** @@ -2534,6 +2666,15 @@ export class ModelResult< // Force stream mode for initial request const request = this.resolvedRequest; + // Park PostModelCall telemetry for this dispatch: the response is + // materialized later (getInitialResponse or the no-tools stream + // teardown), which completes the emit with the true duration. + this.pendingModelCall = { + startedAt: performance.now(), + turnType: 'initial', + turnNumber: 0, + }; + // Make the API request const apiResult = await betaResponsesSend( this.options.client, @@ -2562,6 +2703,7 @@ export class ModelResult< } else if (this.isNonStreamingResponse(apiResult.value)) { // API returned a complete response directly - use it as the final response this.finalResponse = apiResult.value; + await this.emitPendingModelCallOnce(this.finalResponse); } else { throw new Error('Unexpected response type from API'); } @@ -2775,6 +2917,13 @@ export class ModelResult< this.resolvedRequest = request; + // Park PostModelCall telemetry for the resume dispatch (see initStream). + this.pendingModelCall = { + startedAt: performance.now(), + turnType: 'resume', + turnNumber: turnContext.numberOfTurns, + }; + // Make the API request const apiResult = await betaResponsesSend( this.options.client, @@ -2793,6 +2942,7 @@ export class ModelResult< this.reusableStream = new ReusableReadableStream(apiResult.value); } else if (this.isNonStreamingResponse(apiResult.value)) { this.finalResponse = apiResult.value; + await this.emitPendingModelCallOnce(this.finalResponse); } else { throw new Error('Unexpected response type from API'); } @@ -3688,6 +3838,7 @@ export class ModelResult< } const completedResponse = await consumeStreamForCompletion(this.reusableStream); + await this.emitPendingModelCallOnce(completedResponse); return extractToolCallsFromResponse(completedResponse) as ParsedToolCall[]; } diff --git a/packages/agent/src/lib/reusable-stream.ts b/packages/agent/src/lib/reusable-stream.ts index 7b5e2c1..fe8e92c 100644 --- a/packages/agent/src/lib/reusable-stream.ts +++ b/packages/agent/src/lib/reusable-stream.ts @@ -19,6 +19,15 @@ export class ReusableReadableStream { constructor(private sourceStream: ReadableStream) {} + /** + * True once the source stream has been fully read into the buffer. + * A fresh consumer created after this point replays the retained buffer + * without waiting on the source. + */ + get isComplete(): boolean { + return this.sourceComplete; + } + /** * Create a new consumer that can independently iterate over the stream. * Multiple consumers can be created and will all receive the same data. diff --git a/packages/agent/tests/unit/hooks-post-model-call.test.ts b/packages/agent/tests/unit/hooks-post-model-call.test.ts new file mode 100644 index 0000000..242fe8a --- /dev/null +++ b/packages/agent/tests/unit/hooks-post-model-call.test.ts @@ -0,0 +1,395 @@ +/** + * Tests for the PostModelCall telemetry hook and SessionEnd usage totals, + * driven through the public callModel surface with betaResponsesSend mocked + * at the module level. + * + * Covers: + * - one emit per model call across a tool loop (initial + tool_round) + * - payload shape: responseId, model, turnType, turnNumber, usage mapping + * - no-tools getText() and getTextStream() paths emit exactly once + * - usage-less responses emit without `usage`, still counted in totals + * - SessionEnd carries aggregated totalUsage (tokens summed, cost summed) + * - failed initial request emits no PostModelCall and no totalUsage + */ +import type * as models from '@openrouter/sdk/models'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { z } from 'zod/v4'; + +const mockBetaResponsesSend = vi.hoisted(() => vi.fn()); + +vi.mock('@openrouter/sdk/funcs/betaResponsesSend', () => ({ + betaResponsesSend: mockBetaResponsesSend, +})); + +import type { OpenRouterCore } from '@openrouter/sdk/core'; +import { callModel } from '../../src/inner-loop/call-model.js'; +import { HooksManager } from '../../src/lib/hooks-manager.js'; +import type { PostModelCallPayload, SessionEndPayload } from '../../src/lib/hooks-types.js'; +import { ToolType } from '../../src/lib/tool-types.js'; + +afterEach(() => { + mockBetaResponsesSend.mockReset(); + vi.restoreAllMocks(); +}); + +function usageBlock(overrides?: Partial): models.Usage { + return { + inputTokens: 100, + inputTokensDetails: { + cachedTokens: 25, + }, + outputTokens: 50, + outputTokensDetails: { + reasoningTokens: 10, + }, + totalTokens: 150, + cost: 0.002, + ...overrides, + } as models.Usage; +} + +function textResponse(id = 'resp_text', usage?: models.Usage | null): models.OpenResponsesResult { + return { + id, + model: 'test-model-v1', + output: [ + { + type: 'message', + id: `msg_${id}`, + role: 'assistant', + content: [ + { + type: 'output_text', + text: 'hello back', + }, + ], + status: 'completed', + }, + ], + ...(usage !== null && { + usage: usage ?? usageBlock(), + }), + } as unknown as models.OpenResponsesResult; +} + +function toolCallResponse(id = 'resp_tool'): models.OpenResponsesResult { + return { + id, + model: 'test-model-v1', + output: [ + { + type: 'function_call', + id: `out_${id}`, + callId: `call_${id}`, + name: 'echo', + arguments: '{}', + status: 'completed', + }, + ], + usage: usageBlock(), + } as unknown as models.OpenResponsesResult; +} + +function makeEchoTool() { + return { + type: ToolType.Function, + function: { + name: 'echo', + description: 'echo', + inputSchema: z.object({}).loose(), + outputSchema: z.unknown(), + execute: async () => ({ + ok: true, + }), + }, + }; +} + +function collectHooks() { + const hooks = new HooksManager(); + const calls: PostModelCallPayload[] = []; + const ends: SessionEndPayload[] = []; + hooks.on('PostModelCall', { + handler: (payload) => { + calls.push(payload); + }, + }); + hooks.on('SessionEnd', { + handler: (payload) => { + ends.push(payload); + }, + }); + return { + hooks, + calls, + ends, + }; +} + +const client = {} as unknown as OpenRouterCore; + +describe('PostModelCall hook', () => { + it('emits once per model call across a tool loop with correct turn labels', async () => { + mockBetaResponsesSend + .mockResolvedValueOnce({ + ok: true, + value: toolCallResponse('r1'), + }) + .mockResolvedValueOnce({ + ok: true, + value: textResponse('r2'), + }); + + const { hooks, calls } = collectHooks(); + const result = callModel(client, { + model: 'test-model', + input: 'hi', + tools: [ + makeEchoTool(), + ], + hooks, + }); + await result.getText(); + + expect(calls).toHaveLength(2); + expect(calls[0]).toMatchObject({ + responseId: 'r1', + model: 'test-model-v1', + turnType: 'initial', + turnNumber: 0, + }); + expect(calls[1]).toMatchObject({ + responseId: 'r2', + turnType: 'tool_round', + turnNumber: 1, + }); + expect(calls[0]?.durationMs).toBeGreaterThanOrEqual(0); + }); + + it('maps the usage block onto the payload (cached/reasoning/cost)', async () => { + mockBetaResponsesSend.mockResolvedValue({ + ok: true, + value: textResponse('r1'), + }); + + const { hooks, calls } = collectHooks(); + const result = callModel(client, { + model: 'test-model', + input: 'hi', + hooks, + }); + await result.getText(); + + expect(calls).toHaveLength(1); + expect(calls[0]?.usage).toEqual({ + inputTokens: 100, + outputTokens: 50, + totalTokens: 150, + cachedTokens: 25, + reasoningTokens: 10, + cost: 0.002, + }); + }); + + it('emits without usage when the response carries none, still counted in totals', async () => { + mockBetaResponsesSend.mockResolvedValue({ + ok: true, + value: textResponse('r1', null), + }); + + const { hooks, calls, ends } = collectHooks(); + const result = callModel(client, { + model: 'test-model', + input: 'hi', + hooks, + }); + await result.getText(); + + expect(calls).toHaveLength(1); + expect(calls[0]?.usage).toBeUndefined(); + expect(ends[0]?.totalUsage).toMatchObject({ + modelCalls: 1, + totalTokens: 0, + }); + expect(ends[0]?.totalUsage?.cost).toBeUndefined(); + }); + + it('emits exactly once on the no-tools getTextStream() path', async () => { + mockBetaResponsesSend.mockResolvedValue({ + ok: true, + value: textResponse('r1'), + }); + + const { hooks, calls } = collectHooks(); + const result = callModel(client, { + model: 'test-model', + input: 'hi', + hooks, + }); + for await (const _chunk of result.getTextStream()) { + // drain + } + + expect(calls).toHaveLength(1); + expect(calls[0]).toMatchObject({ + responseId: 'r1', + turnType: 'initial', + }); + }); + + it('aggregates totalUsage on SessionEnd across all calls', async () => { + mockBetaResponsesSend + .mockResolvedValueOnce({ + ok: true, + value: toolCallResponse('r1'), + }) + .mockResolvedValueOnce({ + ok: true, + value: textResponse( + 'r2', + usageBlock({ + inputTokens: 200, + outputTokens: 100, + totalTokens: 300, + cost: 0.003, + }), + ), + }); + + const { hooks, ends } = collectHooks(); + const result = callModel(client, { + model: 'test-model', + input: 'hi', + tools: [ + makeEchoTool(), + ], + hooks, + }); + await result.getText(); + + expect(ends).toHaveLength(1); + expect(ends[0]?.totalUsage).toEqual({ + modelCalls: 2, + inputTokens: 300, + outputTokens: 150, + totalTokens: 450, + cachedTokens: 50, + reasoningTokens: 20, + cost: 0.005, + }); + }); + + it('emits no PostModelCall and no totalUsage when the initial request fails', async () => { + mockBetaResponsesSend.mockResolvedValue({ + ok: false, + error: new Error('api down'), + }); + + const { hooks, calls, ends } = collectHooks(); + const result = callModel(client, { + model: 'test-model', + input: 'hi', + hooks, + }); + await expect(result.getText()).rejects.toThrow('api down'); + + expect(calls).toHaveLength(0); + expect(ends).toHaveLength(1); + expect(ends[0]).toMatchObject({ + reason: 'error', + }); + expect(ends[0]?.totalUsage).toBeUndefined(); + }); + + it('emits on a genuinely streaming initial response (response.completed event)', async () => { + const streamed = textResponse('r_streamed'); + const readable = new ReadableStream({ + start(controller) { + controller.enqueue({ + type: 'response.completed', + response: streamed, + sequenceNumber: 0, + }); + controller.close(); + }, + }); + mockBetaResponsesSend.mockResolvedValue({ + ok: true, + value: readable, + }); + + const { hooks, calls } = collectHooks(); + const result = callModel(client, { + model: 'test-model', + input: 'hi', + hooks, + }); + const text = await result.getText(); + + expect(text).toBe('hello back'); + expect(calls).toHaveLength(1); + expect(calls[0]).toMatchObject({ + responseId: 'r_streamed', + turnType: 'initial', + usage: { + totalTokens: 150, + }, + }); + }); + + it('still emits SessionEnd and drains when teardown telemetry cannot materialize', async () => { + // A stream that ends without a completion event (deltas only, then close) + // makes the parked-telemetry materialization in teardown throw + // ("Stream ended without completion event"). That failure must be + // contained: SessionEnd still fires and no PostModelCall is emitted. + const readable = new ReadableStream({ + start(controller) { + controller.enqueue({ + type: 'response.output_item.added', + item: { + type: 'message', + id: 'msg_partial', + role: 'assistant', + content: [], + status: 'in_progress', + }, + outputIndex: 0, + sequenceNumber: 0, + }); + controller.enqueue({ + type: 'response.output_text.delta', + itemId: 'msg_partial', + outputIndex: 0, + contentIndex: 0, + delta: 'partial', + sequenceNumber: 1, + }); + controller.close(); + }, + }); + mockBetaResponsesSend.mockResolvedValue({ + ok: true, + value: readable, + }); + const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {}); + + const { hooks, calls, ends } = collectHooks(); + const result = callModel(client, { + model: 'test-model', + input: 'hi', + hooks, + }); + for await (const _chunk of result.getTextStream()) { + // drain + } + + expect(calls).toHaveLength(0); + expect(ends).toHaveLength(1); + expect(warnSpy).toHaveBeenCalledWith( + '[PostModelCall] error during stream teardown:', + expect.objectContaining({ + message: expect.stringContaining('without completion event'), + }), + ); + }); +});