From a0d1c14bfc801ca02737a78ff3ec81055ec0964d Mon Sep 17 00:00:00 2001 From: fayekelmith Date: Tue, 28 Jul 2026 01:46:56 +0000 Subject: [PATCH 1/2] Generated with Hive: Fix canvas agent observability gaps with per-message timestamps, cache-token accounting, timing stats, and AgentLog writes --- .../migration.sql | 5 + prisma/schema.prisma | 10 +- .../unit/lib/ai/publicChatBudget.test.ts | 84 +++++++++++++++- .../unit/lib/ai/runCanvasAgent-timing.test.ts | 72 ++++++++++++++ .../services/canvas-turn-persistence.test.ts | 50 ++++++++++ src/app/api/ask/quick/route.ts | 96 ++++++++++++++++--- src/lib/ai/publicChatBudget.ts | 61 ++++++++++-- src/lib/ai/runCanvasAgent.ts | 40 ++++++-- src/services/canvas-turn-persistence.ts | 5 +- 9 files changed, 393 insertions(+), 30 deletions(-) create mode 100644 prisma/migrations/20260728000000_add_cache_token_columns_to_shared_conversations/migration.sql diff --git a/prisma/migrations/20260728000000_add_cache_token_columns_to_shared_conversations/migration.sql b/prisma/migrations/20260728000000_add_cache_token_columns_to_shared_conversations/migration.sql new file mode 100644 index 0000000000..691c61fb1b --- /dev/null +++ b/prisma/migrations/20260728000000_add_cache_token_columns_to_shared_conversations/migration.sql @@ -0,0 +1,5 @@ +-- AddColumn cache_read_input_tokens to shared_conversations +ALTER TABLE "shared_conversations" ADD COLUMN "cache_read_input_tokens" INTEGER NOT NULL DEFAULT 0; + +-- AddColumn cache_write_input_tokens to shared_conversations +ALTER TABLE "shared_conversations" ADD COLUMN "cache_write_input_tokens" INTEGER NOT NULL DEFAULT 0; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 669c72039f..ca1f811bf5 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -638,8 +638,14 @@ model SharedConversation { // `src/app/api/ask/quick/route.ts`). Output is the dominant cost // (~5x input on Anthropic Sonnet); the public rate-limit gate // weights it accordingly when summing recent spend. - inputTokens Int @default(0) @map("input_tokens") - outputTokens Int @default(0) @map("output_tokens") + inputTokens Int @default(0) @map("input_tokens") + outputTokens Int @default(0) @map("output_tokens") + // Cache token counters for Anthropic prompt-caching cost tracking. + // Cache reads are priced far cheaper than fresh input; cache writes + // are slightly more expensive. Tracked separately so the rate-limit + // gate can weight them accurately rather than ignoring them. + cacheReadInputTokens Int @default(0) @map("cache_read_input_tokens") + cacheWriteInputTokens Int @default(0) @map("cache_write_input_tokens") createdAt DateTime @default(now()) @map("created_at") updatedAt DateTime @updatedAt @map("updated_at") isShared Boolean @default(false) @map("is_shared") diff --git a/src/__tests__/unit/lib/ai/publicChatBudget.test.ts b/src/__tests__/unit/lib/ai/publicChatBudget.test.ts index d84df666a4..d1157aed55 100644 --- a/src/__tests__/unit/lib/ai/publicChatBudget.test.ts +++ b/src/__tests__/unit/lib/ai/publicChatBudget.test.ts @@ -1,12 +1,26 @@ -import { describe, it, expect } from "vitest"; +import { describe, it, expect, vi, beforeEach } from "vitest"; import { NextRequest } from "next/server"; import { ANON_DAILY_TOKEN_CAP, OUTPUT_TOKEN_WEIGHT, + CACHE_READ_TOKEN_WEIGHT, + CACHE_WRITE_TOKEN_WEIGHT, WORKSPACE_PUBLIC_DAILY_TOKEN_CAP, deriveAnonymousId, + recordTurnTokens, } from "@/lib/ai/publicChatBudget"; +vi.mock("@/lib/db", () => ({ + db: { + sharedConversation: { + update: vi.fn().mockResolvedValue({}), + aggregate: vi.fn().mockResolvedValue({ _sum: {} }), + }, + }, +})); + +import { db } from "@/lib/db"; + function makeReq(headers: Record): NextRequest { return new NextRequest("http://localhost/api/ask/quick", { method: "POST", @@ -64,6 +78,17 @@ describe("token budget constants", () => { expect(OUTPUT_TOKEN_WEIGHT).toBe(5); }); + it("cache reads are cheaper than fresh input (CACHE_READ_TOKEN_WEIGHT < 1)", () => { + // Anthropic prices cache reads at ~10% of fresh input cost. + expect(CACHE_READ_TOKEN_WEIGHT).toBeGreaterThan(0); + expect(CACHE_READ_TOKEN_WEIGHT).toBeLessThan(1); + }); + + it("cache writes cost more than fresh input (CACHE_WRITE_TOKEN_WEIGHT > 1)", () => { + // Anthropic prices cache writes at ~125% of fresh input cost. + expect(CACHE_WRITE_TOKEN_WEIGHT).toBeGreaterThan(1); + }); + it("per-workspace cap dwarfs per-anon cap", () => { // Sanity: a single visitor cannot exhaust the workspace bucket. expect(WORKSPACE_PUBLIC_DAILY_TOKEN_CAP).toBeGreaterThan( @@ -71,3 +96,60 @@ describe("token budget constants", () => { ); }); }); + +describe("recordTurnTokens", () => { + const update = db.sharedConversation.update as ReturnType; + + beforeEach(() => { + vi.clearAllMocks(); + }); + + it("persists all four token fields including cache read/write", async () => { + await recordTurnTokens({ + conversationId: "conv-1", + inputTokens: 100, + outputTokens: 50, + cacheReadInputTokens: 200, + cacheWriteInputTokens: 30, + }); + + expect(update).toHaveBeenCalledWith({ + where: { id: "conv-1" }, + data: { + inputTokens: { increment: 100 }, + outputTokens: { increment: 50 }, + cacheReadInputTokens: { increment: 200 }, + cacheWriteInputTokens: { increment: 30 }, + }, + }); + }); + + it("defaults cache fields to 0 when omitted", async () => { + await recordTurnTokens({ + conversationId: "conv-2", + inputTokens: 10, + outputTokens: 5, + }); + + expect(update).toHaveBeenCalledWith( + expect.objectContaining({ + data: expect.objectContaining({ + cacheReadInputTokens: { increment: 0 }, + cacheWriteInputTokens: { increment: 0 }, + }), + }), + ); + }); + + it("skips the DB call when all token counts are zero", async () => { + await recordTurnTokens({ + conversationId: "conv-3", + inputTokens: 0, + outputTokens: 0, + cacheReadInputTokens: 0, + cacheWriteInputTokens: 0, + }); + + expect(update).not.toHaveBeenCalled(); + }); +}); diff --git a/src/__tests__/unit/lib/ai/runCanvasAgent-timing.test.ts b/src/__tests__/unit/lib/ai/runCanvasAgent-timing.test.ts index 834a98061b..972cb00c9f 100644 --- a/src/__tests__/unit/lib/ai/runCanvasAgent-timing.test.ts +++ b/src/__tests__/unit/lib/ai/runCanvasAgent-timing.test.ts @@ -413,6 +413,78 @@ describe("runCanvasAgent — stage-timing logs", () => { expect(stages).toContain("time-to-first-token"); }); + // ------------------------------------------------------------------------- + // 9. timingStats is passed to hooks.onFinish with the correct shape + // ------------------------------------------------------------------------- + it("passes timingStats to hooks.onFinish with ttfMs, toolRoundTrips, durationMs", async () => { + const chunks = [ + { type: "tool-call", toolCallId: "tc1", toolName: "web_search" }, + { type: "tool-result", toolCallId: "tc1", toolName: "web_search" }, + { type: "text-delta", text: "answer" }, + ]; + + mockStreamText.mockImplementation(makeStreamResult(chunks)); + + const onFinishSpy = vi.fn(); + await runCanvasAgent( + baseOpts({ hooks: { onFinish: onFinishSpy } }), + ); + + expect(onFinishSpy).toHaveBeenCalledTimes(1); + const callArg = onFinishSpy.mock.calls[0][0] as { + usage: unknown; + timingStats: { + ttfMs: number | null; + totalToolCalls: number; + toolRoundTrips: Array<{ tool: string; ms: number }>; + durationMs: number; + }; + }; + + // usage is forwarded + expect(callArg).toHaveProperty("usage"); + + // timingStats is present and well-formed + expect(callArg).toHaveProperty("timingStats"); + const ts = callArg.timingStats; + + // ttfMs: a text-delta was emitted so it should be set (non-null, ≥ 0) + expect(ts.ttfMs).not.toBeNull(); + expect(ts.ttfMs).toBeGreaterThanOrEqual(0); + + // toolRoundTrips: one tool call/result pair was emitted + expect(ts.toolRoundTrips).toHaveLength(1); + expect(ts.toolRoundTrips[0]).toMatchObject({ tool: "web_search" }); + expect(ts.toolRoundTrips[0]!.ms).toBeGreaterThanOrEqual(0); + + // totalToolCalls matches toolRoundTrips count + expect(ts.totalToolCalls).toBe(1); + + // durationMs is non-negative + expect(ts.durationMs).toBeGreaterThanOrEqual(0); + }); + + it("passes timingStats.ttfMs as null when no text-delta chunks were emitted", async () => { + // Tool-only run — no text-delta → ttfMs stays null + const chunks = [ + { type: "tool-call", toolCallId: "tc1", toolName: "list_concepts" }, + { type: "tool-result", toolCallId: "tc1", toolName: "list_concepts" }, + ]; + + mockStreamText.mockImplementation(makeStreamResult(chunks)); + + const onFinishSpy = vi.fn(); + await runCanvasAgent( + baseOpts({ hooks: { onFinish: onFinishSpy } }), + ); + + const callArg = onFinishSpy.mock.calls[0][0] as { + timingStats: { ttfMs: number | null; totalToolCalls: number }; + }; + expect(callArg.timingStats.ttfMs).toBeNull(); + expect(callArg.timingStats.totalToolCalls).toBe(1); + }); + // ------------------------------------------------------------------------- // 8. Cache-hit path logs `skipped: "cache hit"` instead of a fetch time // ------------------------------------------------------------------------- diff --git a/src/__tests__/unit/services/canvas-turn-persistence.test.ts b/src/__tests__/unit/services/canvas-turn-persistence.test.ts index 99bd7adc00..6c128ebf81 100644 --- a/src/__tests__/unit/services/canvas-turn-persistence.test.ts +++ b/src/__tests__/unit/services/canvas-turn-persistence.test.ts @@ -121,6 +121,56 @@ describe("messagesFromSteps", () => { errorText: "Tool call failed", }); }); + + test("rows from different steps get distinct timestamps", () => { + // Each step in messagesFromSteps() calls `new Date().toISOString()` inside + // the for-loop body. The fix moved that call into the loop so different + // steps produce different timestamps. This test verifies the fix. + // + // Strategy: mock the global Date constructor so each instantiation advances + // a counter, producing a different ISO string. We don't mock + // Date.prototype.toISOString (would cause infinite recursion) — instead we + // replace the Date constructor entirely to return controlled dates. + const base = new Date("2024-01-01T00:00:00.000Z").getTime(); + let callCount = 0; + + const OrigDate = Date; + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const MockDate = function (this: any, ...args: unknown[]) { + if (args.length === 0) { + // `new Date()` path — the one messagesFromSteps uses. + return new OrigDate(base + callCount++ * 60_000); + } + // Delegate all other usages (e.g. new Date("2024-…")) to the real ctor. + return new (OrigDate as unknown as new (...a: unknown[]) => Date)(...args); + } as unknown as typeof Date; + Object.setPrototypeOf(MockDate, OrigDate); + MockDate.prototype = OrigDate.prototype; + MockDate.now = OrigDate.now; + MockDate.parse = OrigDate.parse; + MockDate.UTC = OrigDate.UTC; + + const spy = vi.spyOn(globalThis, "Date").mockImplementation(MockDate); + + try { + const steps = [ + { text: "Step one." }, + { text: "Step two." }, + { text: "Step three." }, + ]; + const rows = messagesFromSteps(steps, "t-"); + expect(rows).toHaveLength(3); + + rows.forEach((r) => expect(r.timestamp).toBeDefined()); + + // Each step gets a different minute offset → different ISO strings. + const timestamps = rows.map((r) => r.timestamp!); + const unique = new Set(timestamps); + expect(unique.size).toBeGreaterThan(1); + } finally { + spy.mockRestore(); + } + }); }); describe("appendTurnMessages", () => { diff --git a/src/app/api/ask/quick/route.ts b/src/app/api/ask/quick/route.ts index 71d4c3c361..1ea8d8ecfe 100644 --- a/src/app/api/ask/quick/route.ts +++ b/src/app/api/ask/quick/route.ts @@ -1,4 +1,5 @@ import { NextRequest, NextResponse, after } from "next/server"; +import { put } from "@vercel/blob"; import { validationError, serverError, forbiddenError, isApiError } from "@/types/errors"; import { validateUserBelongsToOrg, validateWorkspaceAccess } from "@/services/workspace"; import { ModelMessage } from "ai"; @@ -11,6 +12,7 @@ import { recordTurnTokens, } from "@/lib/ai/publicChatBudget"; import { db } from "@/lib/db"; +import { Prisma } from "@prisma/client"; import { resolveMessageImageUrls } from "@/lib/ai/resolveMessageImages"; import type { MessageLike } from "@/lib/proposals/handleApproval"; import { runProposalIntent } from "@/lib/proposals/runProposalIntent"; @@ -43,6 +45,12 @@ import { emitFollowUpQuestions, emitProvenance, } from "@/services/canvas-turn-enrichments"; +import { + pusherServer, + getFeatureChannelName, + getTaskChannelName, + PUSHER_EVENTS, +} from "@/lib/pusher"; // Tier-1 backend-driven canvas turns (docs/plans/backend-driven-canvas-turns.md): // the org-canvas turn is persisted server-side in `after()` so it survives the @@ -575,22 +583,86 @@ export async function POST(request: NextRequest) { const conceptIds = extractConceptIdsFromStep(sf.content); conceptIds.forEach((id) => learnedConceptIds.add(id)); }, - onFinish: async ({ usage }) => { + onFinish: async ({ usage, timingStats }) => { // Persist the turn's token usage to the conversation row so // the public-chat rate-limit gate can sum recent spend on // the next request. Best-effort; failures are logged but // do not surface to the user — the stream already finished. - if (!tokenAttributionRowId) return; - const u = usage as - | { inputTokens?: number; outputTokens?: number } - | undefined; - const inputTokens = Number(u?.inputTokens ?? 0); - const outputTokens = Number(u?.outputTokens ?? 0); - await recordTurnTokens({ - conversationId: tokenAttributionRowId, - inputTokens, - outputTokens, - }); + if (tokenAttributionRowId) { + const u = usage as + | { + inputTokens?: number; + outputTokens?: number; + cacheReadInputTokens?: number; + cacheWriteInputTokens?: number; + } + | undefined; + await recordTurnTokens({ + conversationId: tokenAttributionRowId, + inputTokens: Number(u?.inputTokens ?? 0), + outputTokens: Number(u?.outputTokens ?? 0), + cacheReadInputTokens: Number(u?.cacheReadInputTokens ?? 0), + cacheWriteInputTokens: Number(u?.cacheWriteInputTokens ?? 0), + }); + } + + // Gap 4: Write an AgentLog row for this canvas turn so it + // appears in the Agent Logs UI (Canvas tab, source="canvas_chat"). + // Best-effort — never throws or breaks the stream. + // Uses only server-validated context: primaryWorkspaceId comes + // from runCanvasAgent's return (validated internally); the + // conversation id comes from canvasConversationRowId or + // tokenAttributionRowId — both are IDOR-guarded earlier in + // this route. + const agentLogWorkspaceId = primaryWorkspaceId; + const agentLogConversationId = + canvasConversationRowId ?? tokenAttributionRowId; + if (agentLogWorkspaceId && agentLogConversationId) { + try { + const turnId = turnIdStr ?? agentLogConversationId; + const blobPath = `agent-logs/${agentLogWorkspaceId}/${agentLogConversationId}/canvas-agent-${turnId}.json`; + const blobPayload = JSON.stringify({ + sessionId: agentLogConversationId, + messages: [], + }); + const agentBlob = await put(blobPath, blobPayload, { + access: "private", + contentType: "application/json", + addRandomSuffix: false, + allowOverwrite: true, + }); + + const agentLog = await db.agentLog.create({ + data: { + workspaceId: agentLogWorkspaceId, + source: "canvas_chat", + agent: "canvas-agent", + blobUrl: agentBlob.url, + sessionId: agentLogConversationId, + provider: "anthropic", + config: { + model: chatAgentModel ?? "default", + } as Prisma.InputJsonValue, + stats: { + ttfMs: timingStats.ttfMs, + totalToolCalls: timingStats.totalToolCalls, + toolRoundTrips: timingStats.toolRoundTrips, + durationMs: timingStats.durationMs, + } as Prisma.InputJsonValue, + }, + }); + + console.info("[quick-ask] AgentLog created", { + id: agentLog.id, + workspaceId: agentLogWorkspaceId, + conversationId: agentLogConversationId, + }); + } catch (agentLogErr) { + console.error("[quick-ask] AgentLog write failed (non-fatal)", { + error: agentLogErr instanceof Error ? agentLogErr.message : String(agentLogErr), + }); + } + } }, }, }); diff --git a/src/lib/ai/publicChatBudget.ts b/src/lib/ai/publicChatBudget.ts index d0f8fe68ec..63321c6dc8 100644 --- a/src/lib/ai/publicChatBudget.ts +++ b/src/lib/ai/publicChatBudget.ts @@ -20,13 +20,24 @@ import { getClientIp } from "@/lib/rate-limit"; * * Output tokens are weighted 5x because Anthropic Sonnet output costs * ~5x input. The cap is therefore a real cost-cap rather than a - * volume-cap. + * volume-cap. Cache reads and writes are priced relative to fresh input: + * Anthropic charges cache reads at ~10% of fresh input cost and cache + * writes at ~125% of fresh input cost. We use rounded weights to keep + * the arithmetic simple while staying close to the real dollar impact. */ // Anthropic Sonnet pricing as of writing: ~$3 / Mtok input, ~$15 / Mtok // output. 5:1 weighting keeps the cap close to actual dollar cost. export const OUTPUT_TOKEN_WEIGHT = 5; +// Cache reads cost ~10% of a fresh input token (Anthropic prompt-caching +// pricing: $0.30/Mtok reads vs $3/Mtok fresh input on Sonnet). +export const CACHE_READ_TOKEN_WEIGHT = 0.1; + +// Cache writes cost ~125% of a fresh input token (Anthropic: +// $3.75/Mtok writes vs $3/Mtok fresh input on Sonnet). +export const CACHE_WRITE_TOKEN_WEIGHT = 1.25; + // Per-visitor daily cap (~$3-5 of Anthropic spend at Sonnet rates). export const ANON_DAILY_TOKEN_CAP = 100_000; @@ -49,8 +60,18 @@ export function deriveAnonymousId(req: NextRequest): string { return createHash("sha256").update(`${ip}|${ua}`).digest("hex").slice(0, 16); } -function weightedCost(input: number, output: number): number { - return input + output * OUTPUT_TOKEN_WEIGHT; +function weightedCost( + input: number, + output: number, + cacheRead = 0, + cacheWrite = 0, +): number { + return ( + input + + output * OUTPUT_TOKEN_WEIGHT + + cacheRead * CACHE_READ_TOKEN_WEIGHT + + cacheWrite * CACHE_WRITE_TOKEN_WEIGHT + ); } interface BudgetResult { @@ -79,7 +100,12 @@ export async function checkPublicChatBudget(args: { anonymousId: args.anonymousId, createdAt: { gte: since }, }, - _sum: { inputTokens: true, outputTokens: true }, + _sum: { + inputTokens: true, + outputTokens: true, + cacheReadInputTokens: true, + cacheWriteInputTokens: true, + }, }), db.sharedConversation.aggregate({ where: { @@ -87,13 +113,20 @@ export async function checkPublicChatBudget(args: { userId: null, createdAt: { gte: since }, }, - _sum: { inputTokens: true, outputTokens: true }, + _sum: { + inputTokens: true, + outputTokens: true, + cacheReadInputTokens: true, + cacheWriteInputTokens: true, + }, }), ]); const anonCost = weightedCost( anonAgg._sum.inputTokens ?? 0, anonAgg._sum.outputTokens ?? 0, + anonAgg._sum.cacheReadInputTokens ?? 0, + anonAgg._sum.cacheWriteInputTokens ?? 0, ); if (anonCost >= ANON_DAILY_TOKEN_CAP) { return { allowed: false, reason: "anon", retryAfterSecs: 24 * 60 * 60 }; @@ -102,6 +135,8 @@ export async function checkPublicChatBudget(args: { const wsCost = weightedCost( wsAgg._sum.inputTokens ?? 0, wsAgg._sum.outputTokens ?? 0, + wsAgg._sum.cacheReadInputTokens ?? 0, + wsAgg._sum.cacheWriteInputTokens ?? 0, ); if (wsCost >= WORKSPACE_PUBLIC_DAILY_TOKEN_CAP) { return { allowed: false, reason: "workspace", retryAfterSecs: 24 * 60 * 60 }; @@ -120,14 +155,24 @@ export async function recordTurnTokens(args: { conversationId: string; inputTokens: number; outputTokens: number; + cacheReadInputTokens?: number; + cacheWriteInputTokens?: number; }): Promise { - if (!args.inputTokens && !args.outputTokens) return; + const { + inputTokens, + outputTokens, + cacheReadInputTokens = 0, + cacheWriteInputTokens = 0, + } = args; + if (!inputTokens && !outputTokens && !cacheReadInputTokens && !cacheWriteInputTokens) return; try { await db.sharedConversation.update({ where: { id: args.conversationId }, data: { - inputTokens: { increment: Math.max(0, args.inputTokens) }, - outputTokens: { increment: Math.max(0, args.outputTokens) }, + inputTokens: { increment: Math.max(0, inputTokens) }, + outputTokens: { increment: Math.max(0, outputTokens) }, + cacheReadInputTokens: { increment: Math.max(0, cacheReadInputTokens) }, + cacheWriteInputTokens: { increment: Math.max(0, cacheWriteInputTokens) }, }, }); } catch (error) { diff --git a/src/lib/ai/runCanvasAgent.ts b/src/lib/ai/runCanvasAgent.ts index 5ca0861342..ad1b76b169 100644 --- a/src/lib/ai/runCanvasAgent.ts +++ b/src/lib/ai/runCanvasAgent.ts @@ -150,9 +150,19 @@ export interface CanvasAgentHooks { onStepFinish?: (sf: { content: unknown }) => void | Promise; /** * Called once when the stream finishes successfully. Receives the - * final `usage` so the caller can record token spend. + * final `usage` so the caller can record token spend, plus timing + * stats collected during the stream (TTFT, per-tool round-trips, + * total duration). */ - onFinish?: (args: { usage: unknown }) => void | Promise; + onFinish?: (args: { + usage: unknown; + timingStats: { + ttfMs: number | null; + totalToolCalls: number; + toolRoundTrips: Array<{ tool: string; ms: number }>; + durationMs: number; + }; + }) => void | Promise; } /** @@ -679,6 +689,13 @@ export async function runCanvasAgent( // so concurrent calls never share flags or maps. let firstTokenLogged = false; const toolCallStartTimes = new Map(); + // Structured timing stats accumulated during the stream and forwarded + // to hooks.onFinish so callers can write them to AgentLog.stats. + const timingStats = { + ttfMs: null as number | null, + toolRoundTrips: [] as Array<{ tool: string; ms: number }>, + totalToolCalls: 0, + }; if (isMultiWorkspace) { // Multi-workspace mode is auth-only — public-viewer requests are @@ -1099,19 +1116,30 @@ export async function runCanvasAgent( const callTs = toolCallStartTimes.get(chunk.toolCallId); if (callTs !== undefined) { toolCallStartTimes.delete(chunk.toolCallId); - console.log("[runCanvasAgent] timing", { stage: "tool-round-trip", tool: chunk.toolName, ms: Date.now() - callTs, workspaces: workspaceSlugs, orgId: orgId ?? null }); + const ms = Date.now() - callTs; + console.log("[runCanvasAgent] timing", { stage: "tool-round-trip", tool: chunk.toolName, ms, workspaces: workspaceSlugs, orgId: orgId ?? null }); + // Accumulate into structured stats for hooks.onFinish. + timingStats.toolRoundTrips.push({ tool: chunk.toolName, ms }); + timingStats.totalToolCalls++; } } // TTFT: fire once on first text-delta only (not tool-call/reasoning chunks). if (!firstTokenLogged && chunk.type === "text-delta") { firstTokenLogged = true; - console.log("[runCanvasAgent] timing", { stage: "time-to-first-token", ms: Date.now() - streamStart, model: resolvedModelId, workspaces: workspaceSlugs, orgId: orgId ?? null }); + const ttfMs = Date.now() - streamStart; + console.log("[runCanvasAgent] timing", { stage: "time-to-first-token", ms: ttfMs, model: resolvedModelId, workspaces: workspaceSlugs, orgId: orgId ?? null }); + // Capture TTFT in structured stats. + timingStats.ttfMs = ttfMs; } }, onFinish: async ({ usage }) => { - console.log("[runCanvasAgent] timing", { stage: "streaming-duration-total", ms: Date.now() - streamStart, model: resolvedModelId, workspaces: workspaceSlugs, orgId: orgId ?? null }); + const durationMs = Date.now() - streamStart; + console.log("[runCanvasAgent] timing", { stage: "streaming-duration-total", ms: durationMs, model: resolvedModelId, workspaces: workspaceSlugs, orgId: orgId ?? null }); if (hooks?.onFinish) { - await hooks.onFinish({ usage }); + await hooks.onFinish({ + usage, + timingStats: { ...timingStats, durationMs }, + }); } }, // Surface errors that occur DURING streaming (after the 200 response diff --git a/src/services/canvas-turn-persistence.ts b/src/services/canvas-turn-persistence.ts index 3042822368..92d13f342a 100644 --- a/src/services/canvas-turn-persistence.ts +++ b/src/services/canvas-turn-persistence.ts @@ -146,9 +146,12 @@ export function messagesFromSteps( const rows: StoredMessage[] = []; let idx = 0; const nextId = () => `${idPrefix}${idx++}`; - const now = new Date().toISOString(); for (const step of steps) { + // Capture a fresh timestamp per step so rows from different steps + // get distinct wall-clock times rather than sharing a single value + // captured once before the loop. + const now = new Date().toISOString(); // Extract any schedule_check result from this step so it can be // attached to the text row as `deferredCheck` metadata. const deferredCheck = extractDeferredCheckFromStep(step); From 675110a967d9296fc83fd7e911f5efaa2f2f9035 Mon Sep 17 00:00:00 2001 From: tomsmith8 Date: Tue, 28 Jul 2026 06:25:09 +0000 Subject: [PATCH 2/2] Generated with Hive: Fix AgentLog blob to serialize actual messages and correct AI SDK cache token field usage --- .../unit/lib/utils/agent-log-stats.test.ts | 81 +++++++++++++++++++ src/app/api/ask/quick/route.ts | 60 +++++++++++++- 2 files changed, 137 insertions(+), 4 deletions(-) diff --git a/src/__tests__/unit/lib/utils/agent-log-stats.test.ts b/src/__tests__/unit/lib/utils/agent-log-stats.test.ts index 6d4202e76d..ff59d2617a 100644 --- a/src/__tests__/unit/lib/utils/agent-log-stats.test.ts +++ b/src/__tests__/unit/lib/utils/agent-log-stats.test.ts @@ -600,4 +600,85 @@ describe("parseAgentLogStats", () => { expect(stats.actualOutputTokens).toBe(40); }); }); + + describe("canvas AgentLog blob round-trip", () => { + // Tests the serialization shape written by the canvas AgentLog writer + // in src/app/api/ask/quick/route.ts → hooks.onFinish (Bug 2 fix). + // StoredMessage rows with toolCalls are serialized as AI SDK-format + // content[] arrays so parseAgentLogStats can count tool calls, etc. + it("round-trips StoredMessage[] with tool-call rows through parseAgentLogStats", () => { + // Simulate what the canvas AgentLog writer produces: + // one text row + one tool-call row (each StoredMessage with toolCalls + // is serialized as content[] with type:"tool-call" + type:"tool-result"). + const messages = [ + // Text message + { + role: "assistant", + content: "Looking into it.", + timestamp: "2024-01-01T00:00:00.000Z", + }, + // Tool-call message (the flat-map format from the canvas writer) + { + role: "assistant", + timestamp: "2024-01-01T00:00:01.000Z", + content: [ + { + type: "tool-call", + toolCallId: "tc1", + toolName: "web_search", + input: { query: "example" }, + }, + { + type: "tool-result", + toolCallId: "tc1", + toolName: "web_search", + output: { results: [] }, + }, + ], + }, + // Second tool in a different row + { + role: "assistant", + timestamp: "2024-01-01T00:00:02.000Z", + content: [ + { + type: "tool-call", + toolCallId: "tc2", + toolName: "list_concepts", + input: {}, + }, + { + type: "tool-result", + toolCallId: "tc2", + toolName: "list_concepts", + output: { concepts: [] }, + }, + ], + }, + ]; + + // Serialize in the { sessionId, messages } blob shape the canvas writer uses. + const blobJson = JSON.stringify({ sessionId: "sess-canvas-1", messages }); + + const { stats, conversation } = parseAgentLogStats(blobJson); + + // Two distinct tool calls should be counted. + expect(stats.totalToolCalls).toBe(2); + expect(stats.toolFrequency).toEqual({ web_search: 1, list_concepts: 1 }); + + // The conversation should include all three rows (text + two tool rows). + expect(conversation.length).toBe(3); + + // Token estimate is non-zero (text + JSON tool content). + expect(stats.estimatedTokens).toBeGreaterThan(0); + }); + + it("returns zero toolCalls for an empty messages array (the old broken behaviour)", () => { + // Empty blob — was produced before the Bug 2 fix. + const blobJson = JSON.stringify({ sessionId: "sess-empty", messages: [] }); + const { stats } = parseAgentLogStats(blobJson); + expect(stats.totalToolCalls).toBe(0); + expect(stats.totalMessages).toBe(0); + }); + }); }); diff --git a/src/app/api/ask/quick/route.ts b/src/app/api/ask/quick/route.ts index 1ea8d8ecfe..0ec1a11934 100644 --- a/src/app/api/ask/quick/route.ts +++ b/src/app/api/ask/quick/route.ts @@ -589,20 +589,34 @@ export async function POST(request: NextRequest) { // the next request. Best-effort; failures are logged but // do not surface to the user — the stream already finished. if (tokenAttributionRowId) { + // Bug fix: @ai-sdk/anthropic uses `cacheCreationInputTokens` + // (not `cacheWriteInputTokens`) on the usage object. Cache + // fields may also appear under providerMetadata?.anthropic + // depending on SDK version — mirror the fallback chain in + // src/lib/streaming/useStreamProcessor.ts. const u = usage as | { inputTokens?: number; outputTokens?: number; cacheReadInputTokens?: number; - cacheWriteInputTokens?: number; + cacheCreationInputTokens?: number; } | undefined; + const providerMeta = (usage as Record | undefined) + ?.providerMetadata as + | { anthropic?: { cacheReadInputTokens?: number; cacheCreationInputTokens?: number } } + | undefined; + const anthropicMeta = providerMeta?.anthropic; await recordTurnTokens({ conversationId: tokenAttributionRowId, inputTokens: Number(u?.inputTokens ?? 0), outputTokens: Number(u?.outputTokens ?? 0), - cacheReadInputTokens: Number(u?.cacheReadInputTokens ?? 0), - cacheWriteInputTokens: Number(u?.cacheWriteInputTokens ?? 0), + cacheReadInputTokens: Number( + u?.cacheReadInputTokens ?? anthropicMeta?.cacheReadInputTokens ?? 0, + ), + cacheWriteInputTokens: Number( + u?.cacheCreationInputTokens ?? anthropicMeta?.cacheCreationInputTokens ?? 0, + ), }); } @@ -621,9 +635,47 @@ export async function POST(request: NextRequest) { try { const turnId = turnIdStr ?? agentLogConversationId; const blobPath = `agent-logs/${agentLogWorkspaceId}/${agentLogConversationId}/canvas-agent-${turnId}.json`; + + // Convert StoredMessage[] → ParsedMessage[] so parseAgentLogStats + // can count tool calls, compute token estimates, etc. + // Tool-call rows use content[] with type:"tool-call" entries; + // text rows use a plain string content. + const agentLogSteps = await result.steps; + const agentLogStoredRows = messagesFromSteps( + agentLogSteps as Parameters[0], + `${turnId}-a`, + ); + const agentLogMessages = agentLogStoredRows.map((row) => { + if (row.toolCalls && row.toolCalls.length > 0) { + return { + role: row.role, + timestamp: row.timestamp ?? null, + content: row.toolCalls.flatMap((tc) => [ + { + type: "tool-call" as const, + toolCallId: tc.id, + toolName: tc.toolName, + input: tc.input, + }, + { + type: "tool-result" as const, + toolCallId: tc.id, + toolName: tc.toolName, + output: tc.output, + }, + ]), + }; + } + return { + role: row.role, + content: row.content, + timestamp: row.timestamp ?? null, + }; + }); + const blobPayload = JSON.stringify({ sessionId: agentLogConversationId, - messages: [], + messages: agentLogMessages, }); const agentBlob = await put(blobPath, blobPayload, { access: "private",