From 1bdd1be8c6a3cfdc026cb7ab1acc1aa51df97d57 Mon Sep 17 00:00:00 2001 From: diegosouzapw Date: Thu, 6 Aug 2026 21:40:40 -0300 Subject: [PATCH] feat(plugins): add onStreamComplete built-in event exposing streaming usage and timing (#9571) --- .../9571-plugin-streaming-usage-timing.md | 20 +++ open-sse/handlers/chatCore.ts | 16 +- .../handlers/chatCore/pluginOnResponse.ts | 54 +++++++ src/lib/plugins/hooks.ts | 30 ++++ tests/unit/chatcore-plugin-onresponse.test.ts | 143 +++++++++++++++++- 5 files changed, 259 insertions(+), 4 deletions(-) create mode 100644 changelog.d/features/9571-plugin-streaming-usage-timing.md diff --git a/changelog.d/features/9571-plugin-streaming-usage-timing.md b/changelog.d/features/9571-plugin-streaming-usage-timing.md new file mode 100644 index 0000000000..d14ea459d8 --- /dev/null +++ b/changelog.d/features/9571-plugin-streaming-usage-timing.md @@ -0,0 +1,20 @@ +--- +kind: feature +ref: "#9571" +--- + +feat(plugins): add onStreamComplete built-in event exposing streaming usage and timing (#9571) + +Adds a new `onStreamComplete` plugin event that fires after an SSE stream is fully +consumed, carrying usage token counts and timing metrics (latency, TTFT). Built-in +events now include "onStreamComplete" as a fire-and-forget lifecycle hook. + +The payload shape: + +- `status` — HTTP status of the stream +- `usage.prompt_tokens`, `.completion_tokens`, `.reasoning_tokens`, + `.cache_read_input_tokens`, `.cache_creation_input_tokens` +- `timing.latencyMs`, `.ttft` +- `model`, `provider`, `errorCode` + +Non-breaking — existing `onResponse` hooks with `{ streamed: true }` remain unchanged. diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index 26120abce3..1ba610c0e9 100644 --- a/open-sse/handlers/chatCore.ts +++ b/open-sse/handlers/chatCore.ts @@ -245,7 +245,10 @@ import { recordCompressionCacheStats } from "./chatCore/compressionCacheStats.ts import { writeCavemanOutputAnalytics } from "./chatCore/cavemanOutputAnalytics.ts"; import { scheduleQuotaShareConsumption } from "./chatCore/quotaShareConsumption.ts"; import { emitRequestGamificationEvent } from "./chatCore/gamificationEvent.ts"; -import { runPluginOnResponseHook } from "./chatCore/pluginOnResponse.ts"; +import { + runPluginOnResponseHook, + runPluginOnStreamCompleteHook, +} from "./chatCore/pluginOnResponse.ts"; import { scheduleStreamingQuotaShareConsumption } from "./chatCore/streamingQuotaShare.ts"; import { recordStreamingUsageStats } from "./chatCore/streamingUsageStats.ts"; import { recordStreamingCost } from "./chatCore/streamingCost.ts"; @@ -4881,6 +4884,17 @@ export async function handleChatCore({ streamUsage, log, }); + + // Plugin onStreamComplete hook — fire-and-forget, fail-open (#9571) + runPluginOnStreamCompleteHook({ + status: normalizedStreamStatus, + usage: streamUsage as Record | undefined, + ttft, + model, + provider, + errorCode: streamErrorCode, + startTime, + }); }; const streamFailureFinalizers = streamFailure.createStreamFailureFinalizers({ diff --git a/open-sse/handlers/chatCore/pluginOnResponse.ts b/open-sse/handlers/chatCore/pluginOnResponse.ts index 1d74ca2989..92e9c266a1 100644 --- a/open-sse/handlers/chatCore/pluginOnResponse.ts +++ b/open-sse/handlers/chatCore/pluginOnResponse.ts @@ -43,3 +43,57 @@ export async function runPluginOnResponseHook(args: { /* plugin onResponse optional */ } } + +/** + * Payload passed to plugin onStreamComplete hooks after a streaming response is consumed. + * Carries usage token counts, timing metrics (latency, TTFT), model, provider, and error code. + */ +export type PluginOnStreamCompletePayload = { + status: number; + usage?: { + prompt_tokens?: number; + completion_tokens?: number; + reasoning_tokens?: number; + cache_read_input_tokens?: number; + cache_creation_input_tokens?: number; + }; + timing?: { + latencyMs: number; + ttft?: number; + }; + model?: string; + provider?: string; + errorCode?: string; +}; + +/** + * Run plugin onStreamComplete hooks — fire-and-forget and fail-open. + * Called inside the onStreamComplete callback (chatCore.ts) where usage and timing data + * converge after an SSE stream is fully consumed. + */ +export async function runPluginOnStreamCompleteHook(args: { + status: number; + usage?: Record; + ttft?: number; + model: string | null | undefined; + provider: string | null | undefined; + errorCode?: string | null | undefined; + startTime: number; +}): Promise { + try { + const { runOnStreamComplete } = await import("@/lib/plugins/hooks"); + runOnStreamComplete({ + status: args.status, + usage: args.usage as PluginOnStreamCompletePayload["usage"], + timing: { + latencyMs: Date.now() - args.startTime, + ttft: args.ttft, + }, + model: args.model ?? undefined, + provider: args.provider ?? undefined, + errorCode: args.errorCode ?? undefined, + }).catch(() => {}); + } catch (_) { + /* plugin onStreamComplete optional */ + } +} diff --git a/src/lib/plugins/hooks.ts b/src/lib/plugins/hooks.ts index 2ad5345308..1c46523db4 100644 --- a/src/lib/plugins/hooks.ts +++ b/src/lib/plugins/hooks.ts @@ -40,6 +40,7 @@ export const BUILTIN_EVENTS = [ "onActivate", "onDeactivate", "onUninstall", + "onStreamComplete", ] as const; export type BuiltinEvent = (typeof BUILTIN_EVENTS)[number]; @@ -251,6 +252,35 @@ export interface Plugin { onActivate?: (payload: unknown) => Promise | void; onDeactivate?: (payload: unknown) => Promise | void; onUninstall?: (payload: unknown) => Promise | void; + onStreamComplete?: (payload: PluginOnStreamCompletePayload) => Promise | void; +} + +// ── onStreamComplete event types ── + +export type PluginOnStreamCompletePayload = { + status: number; + usage?: { + prompt_tokens?: number; + completion_tokens?: number; + reasoning_tokens?: number; + cache_read_input_tokens?: number; + cache_creation_input_tokens?: number; + }; + timing?: { + latencyMs: number; + ttft?: number; + }; + model?: string; + provider?: string; + errorCode?: string; +}; + +/** + * Run onStreamComplete hooks — fire-and-forget notification with usage/timing data. + * Called when an SSE stream is fully consumed and usage/timing data is available. + */ +export async function runOnStreamComplete(payload: PluginOnStreamCompletePayload): Promise { + await emitHook("onStreamComplete", payload); } /** diff --git a/tests/unit/chatcore-plugin-onresponse.test.ts b/tests/unit/chatcore-plugin-onresponse.test.ts index 4d27eb44ad..82172447e5 100644 --- a/tests/unit/chatcore-plugin-onresponse.test.ts +++ b/tests/unit/chatcore-plugin-onresponse.test.ts @@ -7,9 +7,8 @@ import { test, after } from "node:test"; import assert from "node:assert/strict"; const { registerHook, unregisterHook } = await import("../../src/lib/plugins/hooks.ts"); -const { runPluginOnResponseHook } = await import( - "../../open-sse/handlers/chatCore/pluginOnResponse.ts" -); +const { runPluginOnResponseHook, runPluginOnStreamCompleteHook } = + await import("../../open-sse/handlers/chatCore/pluginOnResponse.ts"); async function waitFor(pred: () => boolean, timeoutMs = 2000): Promise { const deadline = Date.now() + timeoutMs; @@ -20,6 +19,7 @@ async function waitFor(pred: () => boolean, timeoutMs = 2000): Promise { after(() => { unregisterHook("onResponse", "test-onresponse-plugin"); + unregisterHook("onStreamComplete", "test-onstreamcomplete-plugin"); }); test("no registered hooks → resolves without throwing (no-op)", async () => { @@ -101,3 +101,140 @@ test("a throwing hook never rejects the caller (fail-open)", async () => { ); await new Promise((r) => setTimeout(r, 30)); }); + +// ── onStreamComplete hook tests (#9571) ── + +test("onStreamComplete: no registered hooks resolves without throwing (no-op)", async () => { + const start = Date.now(); + await assert.doesNotReject( + runPluginOnStreamCompleteHook({ + status: 200, + usage: { prompt_tokens: 10, completion_tokens: 20 }, + ttft: 150, + model: "gpt-4", + provider: "openai", + errorCode: undefined, + startTime: start - 500, + }) + ); +}); + +test("onStreamComplete: registered hook receives usage + timing payload", async () => { + let captured: Record | undefined; + registerHook( + "onStreamComplete", + "test-onstreamcomplete-plugin", + async (payload: Record) => { + captured = payload; + } + ); + + const startTime = Date.now() - 500; + await runPluginOnStreamCompleteHook({ + status: 200, + usage: { prompt_tokens: 42, completion_tokens: 100, reasoning_tokens: 5 }, + ttft: 200, + model: "claude-3-opus", + provider: "anthropic", + errorCode: undefined, + startTime, + }); + + await waitFor(() => captured !== undefined); + assert.ok(captured, "expected onStreamComplete hook to be invoked"); + + // payload shape: status, usage, timing, model, provider + assert.equal(captured!.status, 200); + assert.ok(captured!.usage, "usage should be present"); + assert.equal((captured!.usage as Record).prompt_tokens, 42); + assert.equal((captured!.usage as Record).completion_tokens, 100); + assert.equal((captured!.usage as Record).reasoning_tokens, 5); + + assert.ok(captured!.timing, "timing should be present"); + const timing = captured!.timing as Record; + assert.equal(timing.ttft, 200); + assert.ok(timing.latencyMs > 450, "latencyMs should be near 500"); + + assert.equal(captured!.model, "claude-3-opus"); + assert.equal(captured!.provider, "anthropic"); + assert.equal(captured!.errorCode, undefined); +}); + +test("onStreamComplete: payload includes cache token fields when present", async () => { + let captured: Record | undefined; + registerHook( + "onStreamComplete", + "test-onstreamcomplete-plugin", + async (payload: Record) => { + captured = payload; + } + ); + + await runPluginOnStreamCompleteHook({ + status: 200, + usage: { + prompt_tokens: 50, + completion_tokens: 30, + cache_read_input_tokens: 20, + cache_creation_input_tokens: 10, + }, + ttft: 100, + model: "gpt-4", + provider: "openai", + errorCode: undefined, + startTime: Date.now(), + }); + + await waitFor(() => captured !== undefined); + assert.ok(captured); + const usage = captured!.usage as Record; + assert.equal(usage.cache_read_input_tokens, 20); + assert.equal(usage.cache_creation_input_tokens, 10); +}); + +test("onStreamComplete: throwing hook never rejects the caller (fail-open)", async () => { + registerHook("onStreamComplete", "test-onstreamcomplete-plugin", async () => { + throw new Error("stream-complete-boom"); + }); + + await assert.doesNotReject( + runPluginOnStreamCompleteHook({ + status: 500, + usage: undefined, + ttft: undefined, + model: "gpt-4", + provider: "openai", + errorCode: "upstream_error", + startTime: Date.now(), + }) + ); + await new Promise((r) => setTimeout(r, 30)); +}); + +test("onStreamComplete: errorCode is passed through when provided", async () => { + let captured: Record | undefined; + registerHook( + "onStreamComplete", + "test-onstreamcomplete-plugin", + async (payload: Record) => { + captured = payload; + } + ); + + await runPluginOnStreamCompleteHook({ + status: 502, + usage: undefined, + ttft: undefined, + model: "grok-3", + provider: "xai", + errorCode: "upstream_timeout", + startTime: Date.now(), + }); + + await waitFor(() => captured !== undefined); + assert.ok(captured); + assert.equal(captured!.status, 502); + assert.equal(captured!.errorCode, "upstream_timeout"); + assert.equal(captured!.model, "grok-3"); + assert.equal(captured!.provider, "xai"); +});