From f7853f11ac040869a471d10634edab4893db0e71 Mon Sep 17 00:00:00 2001 From: Mux Date: Thu, 10 Sep 2026 16:32:43 -0500 Subject: [PATCH 1/2] =?UTF-8?q?[telemetry]=20=F0=9F=A4=96=20feat:=20measur?= =?UTF-8?q?e=20advisor=20cache=20usage=20and=20savings?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary Add an advisor completion event to PostHog. Measure cache usage, estimated savings, timing, and call outcomes without changing cache behavior. ## Background Advisor cache writes can cost more than they save. Existing telemetry cannot measure this tradeoff. ## Implementation - Record successful, failed, and cancelled calls. Exclude calls rejected by the usage limit. - Round token counts, timing, and estimated costs. Preserve unknown values as null. - Compare read savings with the write premium before rounding. - Limit cost estimates to known Anthropic pricing with one known TTL. - Exclude prompt content, cache keys, endpoint URLs, and unresolved custom model names. ## Risks Cache cost estimates use catalog rates, not invoice charges. Mixed TTLs and unsupported pricing remain unknown. Telemetry errors do not change the advisor result. No personal deployment validation is available. --- _Generated with `xum` • Model: `openai:gpt-6-astra` • Thinking: `medium` • Cost: `$13.41`_ --- docs/reference/telemetry.mdx | 24 ++- src/common/orpc/schemas/telemetry.ts | 32 ++++ src/common/telemetry/payload.ts | 9 +- src/common/utils/tools/tools.ts | 3 + .../builtInSkillContent.generated.ts | 24 ++- src/node/services/tools/advisor.test.ts | 172 +++++++++++++++++- src/node/services/tools/advisor.ts | 79 +++++++- .../services/tools/advisorTelemetry.test.ts | 135 ++++++++++++++ src/node/services/tools/advisorTelemetry.ts | 115 ++++++++++++ src/node/services/turnRequestBuilder.ts | 6 + 10 files changed, 582 insertions(+), 17 deletions(-) create mode 100644 src/node/services/tools/advisorTelemetry.test.ts create mode 100644 src/node/services/tools/advisorTelemetry.ts diff --git a/docs/reference/telemetry.mdx b/docs/reference/telemetry.mdx index 03fc8dfc910..c7d65d64484 100644 --- a/docs/reference/telemetry.mdx +++ b/docs/reference/telemetry.mdx @@ -8,7 +8,7 @@ Xum collects anonymous usage telemetry to help improve the product. ## Privacy policy - **No personal information**: Xum does not collect usernames, project names, file paths, or code content. -- **Random IDs only**: Only randomly generated workspace IDs are sent. +- **Random IDs only**: Workspace, parent-turn, and advisor-call IDs contain no user content. - **No hashing**: Hashing is vulnerable to rainbow table attacks. - **Transparent payload**: See exactly what is sent in [`src/common/telemetry/payload.ts`](https://github.com/coder/mux/blob/main/src/common/telemetry/payload.ts). @@ -28,6 +28,28 @@ All telemetry events include basic system information: - **Message sending**: When messages are sent (model, mode, message length rounded to base-2) - **Errors**: Error types and context (no sensitive data) +### Advisor completion + +The backend sends advisor_call_completed after each admitted advisor call, including errors and cancellations. +Calls rejected by the usage limit do not send this event. + +The event includes: + +- Random workspace, parent-turn, and advisor-call IDs, plus the built-in provider route. +- A catalog model name. Unresolved custom model names become unknown. +- The outcome, call index, and time since the previous call within the current tool instance. +- The duration and time to the first text or reasoning token. +- Input, uncached input, cache-read, cache-write, and output token counts. +- Explicit Anthropic cache marker counts and their requested TTL. +- Estimated input cost, cache-write premium, cache-read savings, and net cache savings. + +Token counts, times, and costs use base-2 rounding. Unknown measurements use null, not zero. +Cost estimates use catalog rates for known Anthropic models with one known TTL and complete token counts. +They are not invoice charges. Mixed TTLs and unsupported pricing produce unknown costs. +Net savings subtract the write premium from read savings before rounding. Negative values indicate an estimated loss. +Request markers do not prove that the provider accepts or reuses a cache. +The event contains no prompt content, questions, transcript hashes, cache keys, or endpoint URLs. + ### What Xum does _not_ track - Your messages or code diff --git a/src/common/orpc/schemas/telemetry.ts b/src/common/orpc/schemas/telemetry.ts index 2571ad8fbdc..65694d6e56f 100644 --- a/src/common/orpc/schemas/telemetry.ts +++ b/src/common/orpc/schemas/telemetry.ts @@ -187,6 +187,34 @@ const MCPOAuthFlowFailedPropertiesSchema = z.object({ error_category: TelemetryMCPOAuthFlowErrorCategorySchema, }); +export const AdvisorCallCompletedPropertiesSchema = z.object({ + call_id: z.string(), + parent_turn_id: z.string().optional(), + provider_route: z.string().nullable(), + workspaceId: z.string().optional(), + model: z.string(), + outcome: z.enum(["success", "error", "cancelled"]), + call_index: z.number(), + // This gap covers the current tool instance, not the whole workspace history. + previous_call_gap_ms_b2: z.number().nullable(), + duration_ms_b2: z.number(), + time_to_first_token_ms_b2: z.number().nullable(), + usage_available: z.boolean(), + input_tokens_b2: z.number().nullable(), + uncached_input_tokens_b2: z.number().nullable(), + cache_read_tokens_b2: z.number().nullable(), + cache_write_tokens_b2: z.number().nullable(), + output_tokens_b2: z.number().nullable(), + // Request markers do not prove that the provider accepts or reuses the cache. + cache_marker_count: z.number(), + cache_ttl: z.enum(["5m", "1h", "mixed", "unknown"]), + // Estimates use catalog rates, not invoice charges. Null means unknown. + input_cost_usd_b2: z.number().nullable(), + cache_write_premium_usd_b2: z.number().nullable(), + cache_read_savings_usd_b2: z.number().nullable(), + cache_net_savings_usd_b2: z.number().nullable(), +}); + const StreamCompletedPropertiesSchema = z.object({ model: z.string(), wasInterrupted: z.boolean(), @@ -228,6 +256,10 @@ const ExperimentOverriddenPropertiesSchema = z.object({ // Union of all telemetry events export const TelemetryEventSchema = z.discriminatedUnion("event", [ + z.object({ + event: z.literal("advisor_call_completed"), + properties: AdvisorCallCompletedPropertiesSchema, + }), z.object({ event: z.literal("app_started"), properties: AppStartedPropertiesSchema, diff --git a/src/common/telemetry/payload.ts b/src/common/telemetry/payload.ts index 3c08b89e32d..e23b6853e08 100644 --- a/src/common/telemetry/payload.ts +++ b/src/common/telemetry/payload.ts @@ -20,6 +20,8 @@ * code only needs to provide event-specific properties. */ +import type { z } from "zod"; +import type { AdvisorCallCompletedPropertiesSchema } from "@/common/orpc/schemas/telemetry"; import type { RuntimeMode } from "@/common/types/runtime"; /** @@ -243,9 +245,9 @@ export interface StreamTimingInvalidPayload { reason: string; } -/** - * Stream completion event - tracks when AI responses finish - */ +/** Advisor cache measurements use rounded values, not prompt content. */ +export type AdvisorCallCompletedPayload = z.infer; + export interface StreamCompletedPayload { /** Model used for generation */ model: string; @@ -359,6 +361,7 @@ export interface ExperimentOverriddenPayload { * Frontend sends these; backend adds BaseTelemetryProperties before forwarding to PostHog */ export type TelemetryEventPayload = + | { event: "advisor_call_completed"; properties: AdvisorCallCompletedPayload } | { event: "app_started"; properties: AppStartedPayload } | { event: "workspace_created"; properties: WorkspaceCreatedPayload } | { event: "workspace_switched"; properties: WorkspaceSwitchedPayload } diff --git a/src/common/utils/tools/tools.ts b/src/common/utils/tools/tools.ts index 0e2cbca084b..c2cba74b887 100644 --- a/src/common/utils/tools/tools.ts +++ b/src/common/utils/tools/tools.ts @@ -1,3 +1,4 @@ +import type { AdvisorCallCompletedPayload } from "@/common/telemetry/payload"; import type { HistoryService } from "@/node/services/historyService"; import { createSessionHistoryTool } from "@/node/services/tools/session_history"; import { createNewContextTool } from "@/node/services/tools/new_context"; @@ -355,6 +356,8 @@ export interface ToolConfiguration { }; /** Runtime bundle for the advisor tool (present only when advisor is eligible for this stream). */ advisorRuntime?: { + /** Report cache economics without sending transcript content. */ + reportTelemetry?: (event: AdvisorCallCompletedPayload) => void; /** The advisor model string (e.g. "anthropic:claude-sonnet-4-20250514") */ advisorModelString: string; /** Optional reasoning/thinking level metadata for the advisor request. */ diff --git a/src/node/services/agentSkills/builtInSkillContent.generated.ts b/src/node/services/agentSkills/builtInSkillContent.generated.ts index 585ec6ad33d..e762db643c5 100644 --- a/src/node/services/agentSkills/builtInSkillContent.generated.ts +++ b/src/node/services/agentSkills/builtInSkillContent.generated.ts @@ -7821,7 +7821,7 @@ export const BUILTIN_SKILL_FILES: Record> = { "## Privacy policy", "", "- **No personal information**: Xum does not collect usernames, project names, file paths, or code content.", - "- **Random IDs only**: Only randomly generated workspace IDs are sent.", + "- **Random IDs only**: Workspace, parent-turn, and advisor-call IDs contain no user content.", "- **No hashing**: Hashing is vulnerable to rainbow table attacks.", "- **Transparent payload**: See exactly what is sent in [`src/common/telemetry/payload.ts`](https://github.com/coder/mux/blob/main/src/common/telemetry/payload.ts).", "", @@ -7841,6 +7841,28 @@ export const BUILTIN_SKILL_FILES: Record> = { "- **Message sending**: When messages are sent (model, mode, message length rounded to base-2)", "- **Errors**: Error types and context (no sensitive data)", "", + "### Advisor completion", + "", + "The backend sends advisor_call_completed after each admitted advisor call, including errors and cancellations.", + "Calls rejected by the usage limit do not send this event.", + "", + "The event includes:", + "", + "- Random workspace, parent-turn, and advisor-call IDs, plus the built-in provider route.", + "- A catalog model name. Unresolved custom model names become unknown.", + "- The outcome, call index, and time since the previous call within the current tool instance.", + "- The duration and time to the first text or reasoning token.", + "- Input, uncached input, cache-read, cache-write, and output token counts.", + "- Explicit Anthropic cache marker counts and their requested TTL.", + "- Estimated input cost, cache-write premium, cache-read savings, and net cache savings.", + "", + "Token counts, times, and costs use base-2 rounding. Unknown measurements use null, not zero.", + "Cost estimates use catalog rates for known Anthropic models with one known TTL and complete token counts.", + "They are not invoice charges. Mixed TTLs and unsupported pricing produce unknown costs.", + "Net savings subtract the write premium from read savings before rounding. Negative values indicate an estimated loss.", + "Request markers do not prove that the provider accepts or reuses a cache.", + "The event contains no prompt content, questions, transcript hashes, cache keys, or endpoint URLs.", + "", "### What Xum does _not_ track", "", "- Your messages or code", diff --git a/src/node/services/tools/advisor.test.ts b/src/node/services/tools/advisor.test.ts index dca7ea07479..cc8774028c8 100644 --- a/src/node/services/tools/advisor.test.ts +++ b/src/node/services/tools/advisor.test.ts @@ -7,6 +7,8 @@ import { ADVISOR_HANDOFF_MAX_REASONING_CHARS, ADVISOR_HANDOFF_MAX_TEXT_CHARS, } from "@/common/constants/advisor"; +import { TelemetryEventSchema } from "@/common/orpc/schemas/telemetry"; +import type { AdvisorCallCompletedPayload } from "@/common/telemetry/payload"; import type { ModelMessage } from "@/common/types/message"; import type { WorkspaceChatMessage } from "@/common/orpc/types"; import type { AdvisorToolCallSnapshot, ToolModelUsageEvent } from "@/common/utils/tools/tools"; @@ -40,6 +42,9 @@ function createToolConfig( tempDir: string, options?: { reportModelUsage?: Parameters[0]["reportModelUsage"]; + reportTelemetry?: NonNullable< + Parameters[0]["advisorRuntime"] + >["reportTelemetry"]; metadataModel?: string; transcript?: ModelMessage[]; snapshot?: AdvisorToolCallSnapshot | undefined; @@ -65,6 +70,7 @@ function createToolConfig( emitChatEvent: options?.emitChatEvent, reportModelUsage: options?.reportModelUsage, advisorRuntime: { + reportTelemetry: options?.reportTelemetry, advisorModelString: ADVISOR_MODEL, reasoningLevel: "medium", maxUsesPerTurn: 3, @@ -85,7 +91,13 @@ type StreamTextFinishReason = Awaited; function mockStreamTextSuccess(result: { text: string; - usage: LanguageModelV2Usage; + usage: LanguageModelV2Usage & { + inputTokenDetails?: { + noCacheTokens?: number; + cacheReadTokens?: number; + cacheWriteTokens?: number; + }; + }; providerMetadata?: Record; chunks?: Array<{ type: string; text?: string; delta?: string; textDelta?: string }>; finishReason?: StreamTextFinishReason; @@ -111,13 +123,13 @@ function mockStreamTextSuccess(result: { }) as unknown as typeof ai.streamText); } -function mockStreamTextFailure(error: Error) { +function mockStreamTextFailure(error: Error, usage?: LanguageModelV2Usage) { return spyOn(ai, "streamText").mockImplementation( (() => ({ text: Promise.reject(error), finishReason: Promise.resolve("error"), - usage: Promise.resolve(undefined), + usage: Promise.resolve(usage), providerMetadata: Promise.resolve(undefined), }) as unknown as StreamTextResult) as unknown as typeof ai.streamText ); @@ -640,6 +652,160 @@ describe("advisor tool", () => { ); }); + describe("completion telemetry", () => { + it("reports SDK7 cache usage and first-token timing once", async () => { + using tempDir = new TestTempDir("advisor-telemetry-sdk7"); + const reportTelemetry = mock((_event: AdvisorCallCompletedPayload) => undefined); + const { config } = createToolConfig(tempDir.path, { reportTelemetry }); + mockStreamTextSuccess({ + text: "Use a queue.", + chunks: [ + { type: "reasoning-delta", text: "Check ordering." }, + { type: "text-delta", text: "Use a queue." }, + ], + usage: { + inputTokens: 128, + outputTokens: 16, + totalTokens: 144, + inputTokenDetails: { noCacheTokens: 32, cacheReadTokens: 64, cacheWriteTokens: 32 }, + }, + }); + + const tool = createAdvisorTool(config); + await tool.execute!({}, mockToolCallOptions); + + expect(reportTelemetry).toHaveBeenCalledTimes(1); + expect(reportTelemetry.mock.calls[0]?.[0]).toMatchObject({ + outcome: "success", + call_index: 1, + usage_available: true, + input_tokens_b2: 128, + cache_read_tokens_b2: 64, + cache_write_tokens_b2: 32, + uncached_input_tokens_b2: 32, + output_tokens_b2: 16, + }); + expect(typeof reportTelemetry.mock.calls[0]?.[0].duration_ms_b2).toBe("number"); + expect(typeof reportTelemetry.mock.calls[0]?.[0].time_to_first_token_ms_b2).toBe("number"); + const event = { + event: "advisor_call_completed" as const, + properties: reportTelemetry.mock.calls[0]?.[0], + }; + expect(TelemetryEventSchema.parse(event)).toEqual(event); + }); + + it("excludes rejected calls from telemetry and increments admitted call indexes", async () => { + using tempDir = new TestTempDir("advisor-telemetry-limit"); + const reportTelemetry = mock((_event: AdvisorCallCompletedPayload) => undefined); + const { config } = createToolConfig(tempDir.path, { reportTelemetry }); + mockStreamTextSuccess({ + text: "Proceed.", + usage: { inputTokens: 8, outputTokens: 4, totalTokens: 12 }, + }); + const tool = createAdvisorTool(config); + for (let index = 0; index < 3; index++) { + await tool.execute!({}, mockToolCallOptions); + } + const rejected: unknown = await tool.execute!({}, mockToolCallOptions); + + expect(rejected).toEqual(expect.objectContaining({ type: "limit_reached" })); + expect(reportTelemetry).toHaveBeenCalledTimes(3); + expect(reportTelemetry.mock.calls.map(([event]) => event.call_index)).toEqual([1, 2, 3]); + for (const [event] of reportTelemetry.mock.calls) { + expect(event.outcome).toBe("success"); + expect(event.time_to_first_token_ms_b2).toBeNull(); + } + }); + + it.each([ + ["error", new Error("request failed")], + ["cancelled", new DOMException("request cancelled", "AbortError")], + ] as const)("reports %s without usage when the stream rejects", async (outcome, error) => { + using tempDir = new TestTempDir("advisor-telemetry-no-usage"); + const reportTelemetry = mock((_event: AdvisorCallCompletedPayload) => undefined); + const { config } = createToolConfig(tempDir.path, { reportTelemetry }); + mockStreamTextFailure(error); + const tool = createAdvisorTool(config); + await tool.execute!({}, mockToolCallOptions); + + expect(reportTelemetry).toHaveBeenCalledTimes(1); + expect(reportTelemetry.mock.calls[0]?.[0]).toMatchObject({ + outcome, + call_index: 1, + usage_available: false, + input_tokens_b2: null, + cache_read_tokens_b2: null, + cache_write_tokens_b2: null, + uncached_input_tokens_b2: null, + output_tokens_b2: null, + time_to_first_token_ms_b2: null, + }); + expect(typeof reportTelemetry.mock.calls[0]?.[0].duration_ms_b2).toBe("number"); + }); + + it("retains resolved usage when the text promise rejects", async () => { + using tempDir = new TestTempDir("advisor-telemetry-failed-usage"); + const reportTelemetry = mock((_event: AdvisorCallCompletedPayload) => undefined); + const { config } = createToolConfig(tempDir.path, { reportTelemetry }); + mockStreamTextFailure(new Error("stream failed"), { + inputTokens: 64, + outputTokens: 8, + totalTokens: 72, + }); + const tool = createAdvisorTool(config); + await tool.execute!({}, mockToolCallOptions); + + expect(reportTelemetry).toHaveBeenCalledTimes(1); + expect(reportTelemetry.mock.calls[0]?.[0]).toMatchObject({ + outcome: "error", + usage_available: true, + input_tokens_b2: 64, + output_tokens_b2: 8, + }); + }); + + it("retains usage when an error chunk ends the stream", async () => { + using tempDir = new TestTempDir("advisor-telemetry-error-chunk"); + const reportTelemetry = mock((_event: AdvisorCallCompletedPayload) => undefined); + const { config } = createToolConfig(tempDir.path, { reportTelemetry }); + mockStreamTextSuccess({ + text: "Partial advice.", + finishReason: "error", + streamError: new Error("stream failed"), + chunks: [{ type: "text-delta", text: "Partial advice." }], + usage: { inputTokens: 64, outputTokens: 8, totalTokens: 72 }, + }); + const tool = createAdvisorTool(config); + await tool.execute!({}, mockToolCallOptions); + + expect(reportTelemetry).toHaveBeenCalledTimes(1); + expect(reportTelemetry.mock.calls[0]?.[0]).toMatchObject({ + outcome: "error", + usage_available: true, + input_tokens_b2: 64, + output_tokens_b2: 8, + }); + expect(typeof reportTelemetry.mock.calls[0]?.[0].time_to_first_token_ms_b2).toBe("number"); + }); + + it("returns advice when telemetry reporting throws", async () => { + using tempDir = new TestTempDir("advisor-telemetry-callback-error"); + const reportTelemetry = mock((_event: AdvisorCallCompletedPayload) => { + throw new Error("telemetry failed"); + }); + const { config } = createToolConfig(tempDir.path, { reportTelemetry }); + mockStreamTextSuccess({ + text: "Use a queue.", + usage: { inputTokens: 8, outputTokens: 4, totalTokens: 12 }, + }); + const tool = createAdvisorTool(config); + const result: unknown = await tool.execute!({}, mockToolCallOptions); + + expect(result).toEqual(expect.objectContaining({ type: "advice", advice: "Use a queue." })); + expect(reportTelemetry).toHaveBeenCalledTimes(1); + }); + }); + it("does not report usage when the advisor model call fails", async () => { using tempDir = new TestTempDir("advisor-tool-no-usage-on-error"); const reportModelUsage = mock((_event: ToolModelUsageEvent) => undefined); diff --git a/src/node/services/tools/advisor.ts b/src/node/services/tools/advisor.ts index 5ab419641c1..726efaf72c4 100644 --- a/src/node/services/tools/advisor.ts +++ b/src/node/services/tools/advisor.ts @@ -1,4 +1,8 @@ import assert from "node:assert/strict"; +import { randomUUID } from "node:crypto"; +import type { AdvisorCallCompletedPayload } from "@/common/telemetry/payload"; +import { roundToBase2 } from "@/common/telemetry/utils"; +import { advisorCachePolicy, advisorUsageTelemetry } from "./advisorTelemetry"; import { streamText, tool, type Tool } from "ai"; @@ -154,6 +158,7 @@ export function createAdvisorTool(config: ToolConfiguration): Tool { assert(typeof runtime.createModel === "function", "advisor createModel must be a function"); let usesThisTurn = 0; + let previousCallStartedAt: number | undefined; return tool({ description: TOOL_DEFINITIONS.advisor.description, @@ -233,6 +238,17 @@ export function createAdvisorTool(config: ToolConfiguration): Tool { } // Reserve the slot before any await so concurrent advisor calls cannot bypass the per-turn cap. usesThisTurn++; + const callIndex = usesThisTurn; + const startedAt = performance.now(); + const previousCallGap = + previousCallStartedAt == null ? null : startedAt - previousCallStartedAt; + previousCallStartedAt = startedAt; + let outcome: AdvisorCallCompletedPayload["outcome"] = "error"; + let firstTokenAt: number | undefined; + let telemetryModel = advisorModelString; + let providerRoute: string | null = null; + let telemetryResult: ReturnType | undefined; + let cachePolicy = advisorCachePolicy([]); const remainingUses = runtime.maxUsesPerTurn !== null ? runtime.maxUsesPerTurn - usesThisTurn : null; @@ -241,17 +257,17 @@ export function createAdvisorTool(config: ToolConfiguration): Tool { // which cannot be replayed against a different provider (e.g. OpenAI // rejects foreign `srvtoolu_...` ids as unknown item_references). See // flattenProviderExecutedToolParts for details. - const transcript = flattenProviderExecutedToolParts(runtime.getTranscriptSnapshot()); - assert(Array.isArray(transcript), "advisor transcript snapshot must be an array"); - assert(transcript.length > 0, "advisor transcript snapshot must not be empty"); - assert(toolCallId, "advisor requires toolCallId"); + try { + const transcript = flattenProviderExecutedToolParts(runtime.getTranscriptSnapshot()); + assert(Array.isArray(transcript), "advisor transcript snapshot must be an array"); + assert(transcript.length > 0, "advisor transcript snapshot must not be empty"); + assert(toolCallId, "advisor requires toolCallId"); - const snapshot = runtime.takeToolCallSnapshot(toolCallId); - const handoffMessage = buildAdvisorHandoffMessage(question, snapshot); - const messages: ModelMessage[] = - handoffMessage != null ? [...transcript, handoffMessage] : transcript; + const snapshot = runtime.takeToolCallSnapshot(toolCallId); + const handoffMessage = buildAdvisorHandoffMessage(question, snapshot); + const messages: ModelMessage[] = + handoffMessage != null ? [...transcript, handoffMessage] : transcript; - try { const { model, metadataModel, @@ -280,6 +296,9 @@ export function createAdvisorTool(config: ToolConfiguration): Tool { runtime.reasoningMode ) as unknown as StreamTextProviderOptions; + telemetryModel = metadataModel ?? optionsModelString; + providerRoute = optionsRouteProvider ?? null; + cachePolicy = advisorCachePolicy(messages, providerOptions); emitAdvisorPhase("waiting_for_response"); let advisorStreamError: unknown; @@ -302,6 +321,12 @@ export function createAdvisorTool(config: ToolConfiguration): Tool { advisorStreamError = error; }, onChunk: ({ chunk }) => { + if ( + firstTokenAt == null && + (getAdvisorReasoningDelta(chunk) != null || getAdvisorTextDelta(chunk) != null) + ) { + firstTokenAt = performance.now(); + } const reasoningText = getAdvisorReasoningDelta(chunk); if (reasoningText != null) { emitAdvisorReasoningOutput(reasoningText); @@ -317,6 +342,7 @@ export function createAdvisorTool(config: ToolConfiguration): Tool { emitAdvisorOutput(text); }, }); + telemetryResult = result; const finalAdvice = await result.text; const finishReason = await result.finishReason; if (advisorStreamError != null || finishReason === "error") { @@ -366,6 +392,7 @@ export function createAdvisorTool(config: ToolConfiguration): Tool { } } + outcome = "success"; return { type: "advice" as const, advice, @@ -375,6 +402,7 @@ export function createAdvisorTool(config: ToolConfiguration): Tool { }; } catch (error) { if (error instanceof Error && error.name === "AbortError") { + outcome = "cancelled"; return { type: "error" as const, isError: true, @@ -387,6 +415,39 @@ export function createAdvisorTool(config: ToolConfiguration): Tool { isError: true, message: `Advisor request failed: ${sanitizeErrorMessageForDisplay(getErrorMessage(error))}`, }; + } finally { + if (runtime.reportTelemetry) { + try { + const finishedAt = performance.now(); + const [usageResult, metadataResult] = await Promise.allSettled([ + telemetryResult?.usage, + telemetryResult?.providerMetadata, + ]); + const usage = usageResult.status === "fulfilled" ? usageResult.value : undefined; + const metadata = + metadataResult.status === "fulfilled" ? metadataResult.value : undefined; + runtime.reportTelemetry({ + call_id: randomUUID(), + provider_route: providerRoute, + workspaceId: config.workspaceId, + outcome: + (abortSignal ?? runtime.abortSignal).aborted && outcome !== "success" + ? "cancelled" + : outcome, + call_index: callIndex, + previous_call_gap_ms_b2: + previousCallGap == null ? null : roundToBase2(previousCallGap), + duration_ms_b2: roundToBase2(finishedAt - startedAt), + time_to_first_token_ms_b2: + firstTokenAt == null ? null : roundToBase2(firstTokenAt - startedAt), + ...cachePolicy, + ...advisorUsageTelemetry(telemetryModel, usage, metadata, cachePolicy.cache_ttl), + }); + } catch (error) { + // Telemetry must not change the advisor result. + log.debug("advisor: failed to report telemetry", { error: getErrorMessage(error) }); + } + } } }, }); diff --git a/src/node/services/tools/advisorTelemetry.test.ts b/src/node/services/tools/advisorTelemetry.test.ts new file mode 100644 index 00000000000..a248293f37f --- /dev/null +++ b/src/node/services/tools/advisorTelemetry.test.ts @@ -0,0 +1,135 @@ +import { describe, expect, it } from "bun:test"; +import type { ModelMessage } from "@/common/types/message"; +import { advisorCachePolicy, advisorUsageTelemetry } from "./advisorTelemetry"; + +const MODEL = "anthropic:claude-sonnet-4-20250514"; + +function usage(input: number, read: number, write: number) { + return { + inputTokens: input, + outputTokens: 0, + inputTokenDetails: { cacheReadTokens: read, cacheWriteTokens: write }, + }; +} + +function marker(ttl?: string) { + return { anthropic: { cacheControl: { type: "ephemeral", ...(ttl ? { ttl } : {}) } } }; +} + +describe("advisorUsageTelemetry", () => { + it("excludes private model names and unsupported cost estimates", () => { + const event = advisorUsageTelemetry( + "private-provider:customer-secret-model", + usage(100, 50, 25), + undefined, + "5m" + ); + expect(event.model).toBe("unknown"); + expect(JSON.stringify(event)).not.toContain("customer-secret"); + expect(event.input_tokens_b2).toBe(128); + expect(event.input_cost_usd_b2).toBeNull(); + expect(event.cache_net_savings_usd_b2).toBeNull(); + }); + + it("distinguishes missing usage from measured zero usage", () => { + const missing = advisorUsageTelemetry(MODEL, undefined, undefined, "5m"); + expect(missing.usage_available).toBe(false); + expect(missing.input_tokens_b2).toBeNull(); + expect(missing.output_tokens_b2).toBeNull(); + expect(missing.cache_read_tokens_b2).toBeNull(); + expect(missing.cache_write_tokens_b2).toBeNull(); + expect(missing.uncached_input_tokens_b2).toBeNull(); + expect(missing.input_cost_usd_b2).toBeNull(); + + const zero = advisorUsageTelemetry(MODEL, usage(0, 0, 0), undefined, "5m"); + expect(zero.usage_available).toBe(true); + expect(zero.input_tokens_b2).toBe(0); + expect(zero.output_tokens_b2).toBe(0); + expect(zero.cache_read_tokens_b2).toBe(0); + expect(zero.cache_write_tokens_b2).toBe(0); + expect(zero.uncached_input_tokens_b2).toBe(0); + expect(zero.input_cost_usd_b2).toBe(0); + expect(zero.cache_net_savings_usd_b2).toBe(0); + }); + + it("keeps absent cache details unknown when input usage exists", () => { + const event = advisorUsageTelemetry( + MODEL, + { inputTokens: 100, outputTokens: 0 }, + undefined, + "5m" + ); + expect(event.usage_available).toBe(true); + expect(event.cache_read_tokens_b2).toBeNull(); + expect(event.cache_write_tokens_b2).toBeNull(); + expect(event.uncached_input_tokens_b2).toBeNull(); + expect(event.input_cost_usd_b2).toBeNull(); + }); + + it("preserves the sign of cache savings and write premiums", () => { + const reads = advisorUsageTelemetry(MODEL, usage(1000, 1000, 0), undefined, "5m"); + const writes = advisorUsageTelemetry(MODEL, usage(1000, 0, 1000), undefined, "5m"); + expect(reads.model).toBe("claude-sonnet-4-20250514"); + expect(reads.cache_net_savings_usd_b2).toBeGreaterThan(0); + expect(writes.cache_net_savings_usd_b2).toBeLessThan(0); + expect(writes.cache_write_premium_usd_b2).toBeGreaterThan(0); + expect(reads.cache_write_premium_usd_b2).toBe(0); + expect(writes.cache_read_savings_usd_b2).toBe(0); + }); + + it("charges the long TTL write rate instead of the short TTL rate", () => { + const short = advisorUsageTelemetry(MODEL, usage(1000, 0, 1000), undefined, "5m"); + const long = advisorUsageTelemetry(MODEL, usage(1000, 0, 1000), undefined, "1h"); + // These buckets distinguish the catalog write rate from the long TTL rate. + expect(short.input_cost_usd_b2).toBe(2 ** -8); + expect(long.input_cost_usd_b2).toBe(2 ** -7); + expect(long.cache_write_premium_usd_b2).toBe(2 ** -8); + expect(long.cache_net_savings_usd_b2).toBe(-(2 ** -8)); + }); + + it.each(["unknown", "mixed"] as const)("omits cost estimates for %s TTL", (ttl) => { + const event = advisorUsageTelemetry(MODEL, usage(1000, 500, 500), undefined, ttl); + expect(event.input_cost_usd_b2).toBeNull(); + expect(event.cache_write_premium_usd_b2).toBeNull(); + expect(event.cache_read_savings_usd_b2).toBeNull(); + expect(event.cache_net_savings_usd_b2).toBeNull(); + }); +}); + +describe("advisorCachePolicy", () => { + it.each([ + [undefined, "5m"], + ["5m", "5m"], + ["1h", "1h"], + ["unsupported", "unknown"], + ] as const)("detects marker TTL %s", (ttl, expected) => { + expect(advisorCachePolicy([], marker(ttl))).toEqual({ + cache_marker_count: 1, + cache_ttl: expected, + }); + }); + + it("counts request, message, and content markers without inspecting text", () => { + const messages: ModelMessage[] = [ + { + role: "user", + providerOptions: marker(), + content: [ + { type: "text", text: "Private content", providerOptions: marker("1h") }, + { type: "text", text: "cacheControl ttl 1h" }, + ], + }, + ]; + expect(advisorCachePolicy(messages, marker("5m"))).toEqual({ + cache_marker_count: 3, + cache_ttl: "mixed", + }); + }); + + it("does not infer cache markers from message text", () => { + expect(advisorCachePolicy([{ role: "user", content: "cacheControl ttl 1h" }])).toEqual({ + cache_marker_count: 0, + cache_ttl: "unknown", + }); + }); +}); diff --git a/src/node/services/tools/advisorTelemetry.ts b/src/node/services/tools/advisorTelemetry.ts new file mode 100644 index 00000000000..dcf7d754eee --- /dev/null +++ b/src/node/services/tools/advisorTelemetry.ts @@ -0,0 +1,115 @@ +import type { AdvisorCallCompletedPayload } from "@/common/telemetry/payload"; +import { roundToBase2 } from "@/common/telemetry/utils"; +import type { ModelMessage } from "@/common/types/message"; +import type { AiSdkUsageLike } from "@/common/utils/tokens/usageHelpers"; +import { getModelStats, resolveRawModelEntry } from "@/common/utils/tokens/modelStats"; + +function record(value: unknown): Record | undefined { + return value != null && typeof value === "object" && !Array.isArray(value) + ? (value as Record) + : undefined; +} + +function count(value: unknown): number | null { + return typeof value === "number" && Number.isFinite(value) && value >= 0 ? value : null; +} + +function rounded(value: number | null): number | null { + return value == null ? null : Math.sign(value) * roundToBase2(Math.abs(value)); +} + +/** Inspect explicit request markers without sending their content or cache keys. */ +export function advisorCachePolicy(messages: ModelMessage[], providerOptions?: unknown) { + let markerCount = 0; + const ttls = new Set<"5m" | "1h" | "unknown">(); + const inspect = (options: unknown) => { + const marker = record(record(record(options)?.anthropic)?.cacheControl); + if (!marker) return; + markerCount++; + ttls.add( + marker.ttl == null || marker.ttl === "5m" ? "5m" : marker.ttl === "1h" ? "1h" : "unknown" + ); + }; + inspect(providerOptions); + for (const message of messages) { + inspect(message.providerOptions); + if (Array.isArray(message.content)) { + for (const part of message.content) { + if ("providerOptions" in part) inspect(part.providerOptions); + } + } + } + const ttl: AdvisorCallCompletedPayload["cache_ttl"] = + ttls.size > 1 ? "mixed" : (ttls.values().next().value ?? "unknown"); + return { cache_marker_count: markerCount, cache_ttl: ttl }; +} + +/** Compare cache premiums with read savings before privacy rounding. */ +export function advisorUsageTelemetry( + model: string, + usage: AiSdkUsageLike | undefined, + metadata: Record | undefined, + ttl: AdvisorCallCompletedPayload["cache_ttl"] +) { + // Catalog membership prevents custom aliases and endpoint names from entering PostHog. + const safeModel = resolveRawModelEntry(model)?.key ?? "unknown"; + const input = count(usage?.inputTokens); + const read = count(usage?.inputTokenDetails?.cacheReadTokens ?? usage?.cachedInputTokens); + const write = count( + usage?.inputTokenDetails?.cacheWriteTokens ?? + record(metadata?.anthropic)?.cacheCreationInputTokens + ); + const uncached = + input != null && read != null && write != null ? Math.max(0, input - read - write) : null; + let cost: number | null = null; + let premium: number | null = null; + let savings: number | null = null; + const stats = safeModel === "unknown" ? undefined : getModelStats(safeModel); + // Unknown TTLs, routes, or token details cannot support reliable cache cost estimates. + if ( + (model.startsWith("anthropic:") || model.startsWith("anthropic/")) && + stats && + input != null && + uncached != null && + read != null && + write != null && + (ttl === "5m" || ttl === "1h") + ) { + const high = + stats.tiered_pricing_threshold_tokens != null && + input > stats.tiered_pricing_threshold_tokens; + const inputRate = high + ? (stats.input_cost_per_token_above_200k_tokens ?? stats.input_cost_per_token) + : stats.input_cost_per_token; + const readRate = high + ? (stats.cache_read_input_token_cost_above_200k_tokens ?? stats.cache_read_input_token_cost) + : stats.cache_read_input_token_cost; + const writeRate = + ttl === "1h" + ? 2 * inputRate + : high + ? (stats.cache_creation_input_token_cost_above_200k_tokens ?? + stats.cache_creation_input_token_cost) + : stats.cache_creation_input_token_cost; + if (readRate != null && writeRate != null) { + cost = uncached * inputRate + read * readRate + write * writeRate; + premium = write * (writeRate - inputRate); + savings = read * (inputRate - readRate); + } + } + return { + model: safeModel, + usage_available: usage != null, + input_tokens_b2: rounded(input), + uncached_input_tokens_b2: rounded(uncached), + cache_read_tokens_b2: rounded(read), + cache_write_tokens_b2: rounded(write), + output_tokens_b2: rounded(count(usage?.outputTokens)), + input_cost_usd_b2: rounded(cost), + cache_write_premium_usd_b2: rounded(premium), + cache_read_savings_usd_b2: rounded(savings), + cache_net_savings_usd_b2: rounded( + premium != null && savings != null ? savings - premium : null + ), + }; +} diff --git a/src/node/services/turnRequestBuilder.ts b/src/node/services/turnRequestBuilder.ts index be3ff56dcbc..ca37ba53609 100644 --- a/src/node/services/turnRequestBuilder.ts +++ b/src/node/services/turnRequestBuilder.ts @@ -2193,6 +2193,12 @@ export class TurnRequestBuilder { ...(advisorToolEligible ? { advisorRuntime: { + reportTelemetry: (properties) => { + this.dependencies.telemetryService?.capture({ + event: "advisor_call_completed", + properties: { ...properties, parent_turn_id: assistantMessageId }, + }); + }, advisorModelString, reasoningLevel: advisorReasoningLevel, reasoningMode: cfg.advisorReasoningMode, From f5d59fa595a37b16b769960c40a22c390e49b7a0 Mon Sep 17 00:00:00 2001 From: Mux Date: Thu, 10 Sep 2026 16:54:23 -0500 Subject: [PATCH 2/2] =?UTF-8?q?[telemetry]=20=F0=9F=A4=96=20fix:=20observe?= =?UTF-8?q?=20final=20advisor=20cache=20markers?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Measure cache policy after transport injection and TTL overrides. --- _Generated with `xum` • Model: `openai:gpt-6-astra` • Thinking: `medium` • Cost: `$26.20`_ --- src/common/utils/tools/tools.ts | 5 +- .../services/providerModelFactory.test.ts | 87 +++++++++++++++++++ src/node/services/providerModelFactory.ts | 37 +++++--- src/node/services/tools/advisor.test.ts | 70 ++++++++++++++- src/node/services/tools/advisor.ts | 10 ++- src/node/services/tools/advisorTelemetry.ts | 30 +++++-- src/node/services/turnRequestBuilder.ts | 11 ++- 7 files changed, 223 insertions(+), 27 deletions(-) diff --git a/src/common/utils/tools/tools.ts b/src/common/utils/tools/tools.ts index c2cba74b887..f165524947d 100644 --- a/src/common/utils/tools/tools.ts +++ b/src/common/utils/tools/tools.ts @@ -377,7 +377,10 @@ export interface ToolConfiguration { * Coder identities retain their actual instance and scoped aliases; option * construction resolves their wire from this snapshot, never live config. */ - createModel: (modelString: string) => Promise<{ + createModel: ( + modelString: string, + onAnthropicRequest?: (requestBody: unknown) => void + ) => Promise<{ model: LanguageModel; metadataModel?: string; optionsModelString: string; diff --git a/src/node/services/providerModelFactory.test.ts b/src/node/services/providerModelFactory.test.ts index be8224bbc92..6ea95c448f4 100644 --- a/src/node/services/providerModelFactory.test.ts +++ b/src/node/services/providerModelFactory.test.ts @@ -39,6 +39,7 @@ import { CodexOauthService } from "./codexOauthService"; import type { CoderOauthService } from "./coderOauthService"; import { PolicyService } from "./policyService"; import { ProviderService } from "./providerService"; +import { advisorWireCachePolicy } from "./tools/advisorTelemetry"; const LOCAL_VLLM_BASE_URL = "http://localhost:8000/v1"; const LOCAL_VLLM_MODEL = "qwen3-coder"; @@ -2494,11 +2495,93 @@ describe("wrapFetchWithXAIServiceTier", () => { // Effort "xhigh" and thinking.display flow through the SDK directly as of // @ai-sdk/anthropic 4.0.11 (see buildProviderOptions), so the wrapper must NOT // rewrite reasoning fields — it only normalizes cache_control. +describe("Anthropic cache request observation", () => { + it.each([ + { ttl: undefined, marked: false, count: 1, expectedTtl: "5m" }, + { ttl: "1h", marked: false, count: 1, expectedTtl: "1h" }, + { ttl: "1h", marked: true, count: 2, expectedTtl: "1h" }, + ] as const)( + "observes pinned model wire markers: %j", + async ({ ttl, marked, count, expectedTtl }) => { + await withTempConfig(async (config, factory, _oauth, store) => { + store.saveProvidersConfig({ anthropic: { apiKey: "test-key", cacheTtl: ttl } }); + await saveRoutePriority(config, ["direct"]); + const observed: unknown[] = []; + const { calls, fakeFetch } = createCapturingFetch(); + const fetchSpy = spyOn(globalThis, "fetch").mockImplementation(fakeFetch); + try { + const result = await factory.createModelWithPinnedOptions( + "anthropic:claude-sonnet-4-20250514", + { + onAnthropicRequest: (body) => observed.push(body), + } + ); + if (!result.success) throw new Error(result.error.type); + // The capture fetch has no model response. Only request serialization matters here. + await generateText({ + model: result.data.model, + maxRetries: 0, + messages: [ + ...(marked + ? [ + { + role: "assistant" as const, + content: "Earlier advice", + providerOptions: { + anthropic: { cacheControl: { type: "ephemeral", ttl: "1h" } }, + }, + }, + ] + : []), + { role: "user", content: "hello" }, + ], + }).catch((error: unknown) => { + if (calls.length === 0) throw error; + }); + expect(observed).toHaveLength(1); + expect(calls).toHaveLength(1); + expect(observed[0]).toEqual(parseSentBody(calls[0])); + expect(advisorWireCachePolicy(observed[0])).toEqual({ + cache_marker_count: count, + cache_ttl: expectedTtl, + }); + } finally { + fetchSpy.mockRestore(); + } + }); + } + ); + + it("keeps wire injection when the observer throws", async () => { + const { calls, fakeFetch } = createCapturingFetch(); + const wrapped = wrapFetchWithAnthropicCacheControl(fakeFetch, "1h", { + onRequest: () => { + throw new Error("observer failed"); + }, + }); + await wrapped("https://api.anthropic.com/v1/messages", { + method: "POST", + body: JSON.stringify({ + messages: [{ role: "user", content: [{ type: "text", text: "hello" }] }], + }), + }); + expect(calls).toHaveLength(1); + expect(advisorWireCachePolicy(parseSentBody(calls[0]))).toEqual({ + cache_marker_count: 1, + cache_ttl: "1h", + }); + }); +}); + describe("wrapFetchWithAnthropicCacheControl — ZDR stripping", () => { it("strips existing cache markers when injection is disabled", async () => { const { calls, fakeFetch } = createCapturingFetch(); + let observed: unknown; const wrapped = wrapFetchWithAnthropicCacheControl(fakeFetch, undefined, { injectCacheControl: false, + onRequest: (body) => { + observed = body; + }, }); // Markers the request pipeline can serialize before the wrapper runs: @@ -2523,6 +2606,10 @@ describe("wrapFetchWithAnthropicCacheControl — ZDR stripping", () => { const sent = JSON.stringify(parseSentBody(calls[0])); expect(sent).not.toContain("cache_control"); expect(sent).not.toContain("cacheControl"); + expect(advisorWireCachePolicy(observed)).toEqual({ + cache_marker_count: 0, + cache_ttl: "unknown", + }); }); it("keeps markers when injection is enabled", async () => { diff --git a/src/node/services/providerModelFactory.ts b/src/node/services/providerModelFactory.ts index b432960101c..b3b02ef41ae 100644 --- a/src/node/services/providerModelFactory.ts +++ b/src/node/services/providerModelFactory.ts @@ -517,7 +517,10 @@ function stripAnthropicCacheControlMarkers(json: Record): void export function wrapFetchWithAnthropicCacheControl( baseFetch: typeof fetch, cacheTtl?: AnthropicCacheTtl | null, - options?: { injectCacheControl?: boolean } + options?: { + injectCacheControl?: boolean; + onRequest?: (requestBody: unknown) => void; + } ): typeof fetch { const injectCacheControl = options?.injectCacheControl ?? true; const cachingFetch = async ( @@ -582,6 +585,12 @@ export function wrapFetchWithAnthropicCacheControl( // Update body with modified JSON const newBody = JSON.stringify(json); + // Observe the final markers, including injected markers and TTL overrides. + try { + options?.onRequest?.(json); + } catch (error) { + log.debug("Anthropic request observer failed", { error }); + } const outHeaders = new Headers(init.headers); outHeaders.delete("content-length"); // Body size changed return baseFetch(input, { ...init, headers: outHeaders, body: newBody }); @@ -1177,6 +1186,7 @@ export interface PinnedModelOptions extends Pick< } interface CreateModelOptions { + onAnthropicRequest?: (requestBody: unknown) => void; agentInitiated?: boolean; workspaceId?: string; routeContext?: RouteContext; @@ -1355,11 +1365,7 @@ export class ProviderModelFactory { private createModelCoreEffect( modelString: string, muxProviderOptions?: MuxProviderOptions, - opts?: { - agentInitiated?: boolean; - routeContext?: RouteContext; - providersConfig?: ProvidersConfig; - } + opts?: CreateModelOptions ): Effect.Effect> { // eslint-disable-next-line @typescript-eslint/no-this-alias -- Effect.gen generator bodies do not inherit `this` const self = this; @@ -1614,7 +1620,7 @@ export class ProviderModelFactory { fetch: wrapFetchWithAnthropicCacheControl( customAdapterFetch, effectiveAnthropicCacheTtl, - { injectCacheControl: !disableBeta } + { injectCacheControl: !disableBeta, onRequest: opts?.onAnthropicRequest } ), }); return Ok(provider(modelId)); @@ -1657,7 +1663,7 @@ export class ProviderModelFactory { const fetchWithCacheControl = wrapFetchWithAnthropicCacheControl( baseFetch, effectiveAnthropicCacheTtl, - { injectCacheControl: !disableBeta } + { injectCacheControl: !disableBeta, onRequest: opts?.onAnthropicRequest } ); const providerFetch = fetchWithCacheControl; const provider = createAnthropic({ @@ -2214,6 +2220,7 @@ export class ProviderModelFactory { const fetchWithCacheControl = isAnthropicModel ? wrapFetchWithAnthropicCacheControl(baseFetch, effectiveAnthropicCacheTtl, { injectCacheControl: !disableBeta, + onRequest: opts?.onAnthropicRequest, }) : baseFetch; const fetchWithAutoLogout = wrapFetchWithMuxGatewayAutoLogout( @@ -2557,7 +2564,7 @@ export class ProviderModelFactory { const providerFetch = wrapFetchWithAnthropicCacheControl( coderFetch, effectiveAnthropicCacheTtl, - { injectCacheControl: !disableBeta } + { injectCacheControl: !disableBeta, onRequest: opts?.onAnthropicRequest } ); const { createAnthropic } = yield* Effect.promise(async () => PROVIDER_REGISTRY.anthropic() @@ -2668,6 +2675,7 @@ export class ProviderModelFactory { modelString: string, opts?: { thinkingLevel?: ThinkingLevel; + onAnthropicRequest?: (requestBody: unknown) => void; providerOptions?: MuxProviderOptions; agentInitiated?: boolean; workspaceId?: string; @@ -2683,6 +2691,7 @@ export class ProviderModelFactory { { agentInitiated: opts?.agentInitiated, workspaceId: opts?.workspaceId, + onAnthropicRequest: opts?.onAnthropicRequest, providersConfig, } ); @@ -2733,7 +2742,10 @@ export class ProviderModelFactory { modelString: string, thinkingLevel: ThinkingLevel | undefined, muxProviderOptions?: MuxProviderOptions, - opts?: Pick + opts?: Pick< + CreateModelOptions, + "agentInitiated" | "workspaceId" | "providersConfig" | "onAnthropicRequest" + > ): Promise> { return Effect.runPromise( this.resolveAndCreateModelEffect(modelString, thinkingLevel, muxProviderOptions, opts) @@ -2744,7 +2756,10 @@ export class ProviderModelFactory { modelString: string, thinkingLevel: ThinkingLevel | undefined, muxProviderOptions?: MuxProviderOptions, - opts?: Pick + opts?: Pick< + CreateModelOptions, + "agentInitiated" | "workspaceId" | "providersConfig" | "onAnthropicRequest" + > ): Effect.Effect> { // eslint-disable-next-line @typescript-eslint/no-this-alias -- Effect.gen generator bodies do not inherit `this` const self = this; diff --git a/src/node/services/tools/advisor.test.ts b/src/node/services/tools/advisor.test.ts index cc8774028c8..62b7e4bc579 100644 --- a/src/node/services/tools/advisor.test.ts +++ b/src/node/services/tools/advisor.test.ts @@ -13,6 +13,7 @@ import type { ModelMessage } from "@/common/types/message"; import type { WorkspaceChatMessage } from "@/common/orpc/types"; import type { AdvisorToolCallSnapshot, ToolModelUsageEvent } from "@/common/utils/tools/tools"; import { log } from "@/node/services/log"; +import { wrapFetchWithAnthropicCacheControl } from "@/node/services/providerModelFactory"; import { createAdvisorTool } from "./advisor"; import { TestTempDir, createTestToolConfig } from "./testHelpers"; @@ -102,9 +103,11 @@ function mockStreamTextSuccess(result: { chunks?: Array<{ type: string; text?: string; delta?: string; textDelta?: string }>; finishReason?: StreamTextFinishReason; streamError?: Error; + beforeText?: () => Promise; }) { return spyOn(ai, "streamText").mockImplementation(((args: StreamTextArgs) => { const text = (async () => { + await result.beforeText?.(); for (const chunk of result.chunks ?? []) { await args.onChunk?.({ chunk } as Parameters>[0]); } @@ -289,7 +292,7 @@ describe("advisor tool", () => { const tool = createAdvisorTool(config); const rawResult: unknown = await Promise.resolve(tool.execute!({}, mockToolCallOptions)); - expect(createModel).toHaveBeenCalledWith(ADVISOR_MODEL); + expect(createModel).toHaveBeenCalledWith(ADVISOR_MODEL, expect.any(Function)); expect(streamTextSpy).toHaveBeenCalledTimes(1); expect(rawResult).toEqual({ type: "advice", @@ -653,6 +656,71 @@ describe("advisor tool", () => { }); describe("completion telemetry", () => { + it.each([ + { ttl: undefined, marked: false, expectedTtl: "5m", expectedCount: 1 }, + { ttl: "1h", marked: false, expectedTtl: "1h", expectedCount: 1 }, + { ttl: "1h", marked: true, expectedTtl: "1h", expectedCount: 2 }, + ] as const)( + "uses final wire cache markers: %j", + async ({ ttl, marked, expectedTtl, expectedCount }) => { + using tempDir = new TestTempDir("advisor-wire-cache-telemetry"); + const reportTelemetry = mock((_event: AdvisorCallCompletedPayload) => undefined); + const { config, createModel } = createToolConfig(tempDir.path, { reportTelemetry }); + let observeRequest: ((body: unknown) => void) | undefined; + const fakeFetch = Object.assign(() => Promise.resolve(new Response("{}")), fetch); + const wrapped = wrapFetchWithAnthropicCacheControl(fakeFetch, ttl, { + onRequest: (body) => observeRequest?.(body), + }); + mockStreamTextSuccess({ + text: "Use a queue.", + usage: { + inputTokens: 1000, + outputTokens: 8, + totalTokens: 1008, + inputTokenDetails: { noCacheTokens: 0, cacheReadTokens: 0, cacheWriteTokens: 1000 }, + }, + beforeText: async () => { + await wrapped("https://api.anthropic.com/v1/messages", { + method: "POST", + body: JSON.stringify({ + ...(marked + ? { + system: [ + { + type: "text", + text: "Rules", + cache_control: { type: "ephemeral", ttl: "1h" }, + }, + ], + } + : {}), + messages: [{ role: "user", content: [{ type: "text", text: "hello" }] }], + }), + }); + }, + }); + const tool = createAdvisorTool({ + ...config, + advisorRuntime: { + ...config.advisorRuntime, + createModel: (_model, onAnthropicRequest) => { + observeRequest = onAnthropicRequest; + return createModel(); + }, + }, + }); + + await tool.execute!({}, mockToolCallOptions); + + expect(reportTelemetry).toHaveBeenCalledTimes(1); + expect(reportTelemetry.mock.calls[0]?.[0]).toMatchObject({ + cache_ttl: expectedTtl, + cache_marker_count: expectedCount, + input_cost_usd_b2: ttl === "1h" ? 2 ** -7 : 2 ** -8, + }); + } + ); + it("reports SDK7 cache usage and first-token timing once", async () => { using tempDir = new TestTempDir("advisor-telemetry-sdk7"); const reportTelemetry = mock((_event: AdvisorCallCompletedPayload) => undefined); diff --git a/src/node/services/tools/advisor.ts b/src/node/services/tools/advisor.ts index 726efaf72c4..21907fd64cf 100644 --- a/src/node/services/tools/advisor.ts +++ b/src/node/services/tools/advisor.ts @@ -2,7 +2,11 @@ import assert from "node:assert/strict"; import { randomUUID } from "node:crypto"; import type { AdvisorCallCompletedPayload } from "@/common/telemetry/payload"; import { roundToBase2 } from "@/common/telemetry/utils"; -import { advisorCachePolicy, advisorUsageTelemetry } from "./advisorTelemetry"; +import { + advisorCachePolicy, + advisorWireCachePolicy, + advisorUsageTelemetry, +} from "./advisorTelemetry"; import { streamText, tool, type Tool } from "ai"; @@ -275,7 +279,9 @@ export function createAdvisorTool(config: ToolConfiguration): Tool { optionsProvidersConfig, optionsMuxProviderOptions, optionsRouteProvider, - } = await runtime.createModel(advisorModelString); + } = await runtime.createModel(advisorModelString, (requestBody) => { + cachePolicy = advisorWireCachePolicy(requestBody); + }); // Keep the creation-time identity, including the actual Coder instance // and scoped aliases. buildProviderOptions resolves its wire namespace // from the same captured config and returns provider SDK option types; diff --git a/src/node/services/tools/advisorTelemetry.ts b/src/node/services/tools/advisorTelemetry.ts index dcf7d754eee..17a8d0e4eaf 100644 --- a/src/node/services/tools/advisorTelemetry.ts +++ b/src/node/services/tools/advisorTelemetry.ts @@ -20,23 +20,35 @@ function rounded(value: number | null): number | null { /** Inspect explicit request markers without sending their content or cache keys. */ export function advisorCachePolicy(messages: ModelMessage[], providerOptions?: unknown) { + return advisorWireCachePolicy({ messages, providerOptions }); +} + +/** Inspect effective wire markers after provider cache injection and TTL overrides. */ +export function advisorWireCachePolicy(requestBody: unknown) { let markerCount = 0; const ttls = new Set<"5m" | "1h" | "unknown">(); - const inspect = (options: unknown) => { - const marker = record(record(record(options)?.anthropic)?.cacheControl); + const inspect = (value: unknown) => { + const marker = record(value); if (!marker) return; markerCount++; ttls.add( marker.ttl == null || marker.ttl === "5m" ? "5m" : marker.ttl === "1h" ? "1h" : "unknown" ); }; - inspect(providerOptions); - for (const message of messages) { - inspect(message.providerOptions); - if (Array.isArray(message.content)) { - for (const part of message.content) { - if ("providerOptions" in part) inspect(part.providerOptions); - } + const pending: unknown[] = [requestBody]; + while (pending.length > 0) { + const value = pending.pop(); + if (Array.isArray(value)) { + for (const item of value) pending.push(item); + continue; + } + const entry = record(value); + if (!entry) continue; + inspect(entry.cache_control); + inspect(record(record(entry.providerOptions)?.anthropic)?.cacheControl); + // Do not treat tool input schemas or tool results as request cache markers. + for (const key of ["system", "messages", "prompt", "tools", "content"]) { + if (Array.isArray(entry[key])) pending.push(entry[key]); } } const ttl: AdvisorCallCompletedPayload["cache_ttl"] = diff --git a/src/node/services/turnRequestBuilder.ts b/src/node/services/turnRequestBuilder.ts index ca37ba53609..5a72be15673 100644 --- a/src/node/services/turnRequestBuilder.ts +++ b/src/node/services/turnRequestBuilder.ts @@ -2167,7 +2167,11 @@ export class TurnRequestBuilder { const allowLegacyInvalidWorkflowAgentOutputSchema = await this.dependencies.shouldAllowLegacyInvalidWorkflowAgentOutputSchema(metadata); // Share creation-time provider/pricing snapshots for both headless tools. - const createToolModel = async (ms: string, toolThinkingLevel?: ThinkingLevel) => { + const createToolModel = async ( + ms: string, + toolThinkingLevel?: ThinkingLevel, + onAnthropicRequest?: (requestBody: unknown) => void + ) => { const toolModelString = ms.trim(); assert( toolModelString.length > 0, @@ -2175,7 +2179,7 @@ export class TurnRequestBuilder { ); const created = await this.dependencies.providerModelFactory.createModelWithPinnedOptions( toolModelString, - { thinkingLevel: toolThinkingLevel, workspaceId, agentInitiated: true } + { thinkingLevel: toolThinkingLevel, workspaceId, agentInitiated: true, onAnthropicRequest } ); if (!created.success) { throw new Error(`Failed to create tool model: ${getErrorMessage(created.error)}`); @@ -2226,7 +2230,8 @@ export class TurnRequestBuilder { assert(snapshot.toolName === "advisor", "advisor snapshot must belong to advisor"); return snapshot; }, - createModel: createToolModel, + createModel: (ms, onAnthropicRequest) => + createToolModel(ms, undefined, onAnthropicRequest), abortSignal: combinedAbortSignal, }, }