From 716101eed024c174b37c5d98c524113663a6fa72 Mon Sep 17 00:00:00 2001 From: Hong Jiarong Date: Thu, 9 Jul 2026 20:29:56 +0800 Subject: [PATCH] feat: add agent cost reporting --- .../migration.sql | 6 + hub/prisma/schema.prisma | 5 + hub/src/agent/cost.ts | 23 +++ hub/src/agent/runner.ts | 6 + hub/src/feishu/card/streaming-card.ts | 17 ++- hub/src/feishu/slashCommands.ts | 137 ++++++++++++++++++ hub/src/feishu/trigger.ts | 4 +- hub/test/integration/trigger.test.ts | 65 ++++++++- hub/test/unit/runner.test.ts | 19 ++- 9 files changed, 275 insertions(+), 7 deletions(-) create mode 100644 hub/prisma/migrations/20260709143000_agent_run_cost_tracking/migration.sql create mode 100644 hub/src/agent/cost.ts diff --git a/hub/prisma/migrations/20260709143000_agent_run_cost_tracking/migration.sql b/hub/prisma/migrations/20260709143000_agent_run_cost_tracking/migration.sql new file mode 100644 index 0000000..2e51a51 --- /dev/null +++ b/hub/prisma/migrations/20260709143000_agent_run_cost_tracking/migration.sql @@ -0,0 +1,6 @@ +-- Record real provider-reported run cost for /cost reporting. +-- No backfill: pre-existing test runs remain cost-unrecorded by design. +ALTER TABLE "AgentRun" ADD COLUMN "costUsd" DECIMAL(18, 8); +ALTER TABLE "AgentRun" ADD COLUMN "costSource" TEXT; + +CREATE INDEX "AgentRun_projectId_finishedAt_idx" ON "AgentRun"("projectId", "finishedAt"); diff --git a/hub/prisma/schema.prisma b/hub/prisma/schema.prisma index fa34433..651e501 100644 --- a/hub/prisma/schema.prisma +++ b/hub/prisma/schema.prisma @@ -247,6 +247,10 @@ model AgentRun { summary String? inputTokens Int? outputTokens Int? + /// Provider/gateway-reported cost in USD. Null means this run has no trusted cost fact. + costUsd Decimal? @db.Decimal(18, 8) + /// Semantic source of costUsd, e.g. provider_reported. Not a pricing-estimate fallback. + costSource String? metadata Json error String? startedAt DateTime @default(now()) @@ -261,6 +265,7 @@ model AgentRun { fileChanges AgentFileChange[] @relation("runFileChanges") @@index([projectId, status]) + @@index([projectId, finishedAt]) @@index([sessionId]) @@index([requestedByUserId]) @@index([updatedAt]) diff --git a/hub/src/agent/cost.ts b/hub/src/agent/cost.ts new file mode 100644 index 0000000..1dc0d86 --- /dev/null +++ b/hub/src/agent/cost.ts @@ -0,0 +1,23 @@ +export function formatRunCostLine(costUsd: number | undefined): string { + return costUsd === undefined + ? "本次成本: 未记录" + : `本次成本: ${formatUsd(costUsd)}`; +} + +export function formatInteger(value: number): string { + return new Intl.NumberFormat("en-US").format(value); +} + +export function formatUsd(value: number): string { + const digits = value > 0 && value < 0.01 ? 4 : 2; + return `$${value.toFixed(digits)}`; +} + +export function decimalToNumberOrNull(value: unknown): number | null { + if (value === null || value === undefined) return null; + const numberValue = typeof value === "number" ? value : Number(value); + if (!Number.isFinite(numberValue)) { + throw new Error("invalid cost value"); + } + return numberValue; +} diff --git a/hub/src/agent/runner.ts b/hub/src/agent/runner.ts index aa1a90c..a14f66a 100644 --- a/hub/src/agent/runner.ts +++ b/hub/src/agent/runner.ts @@ -73,6 +73,8 @@ export interface RunResult { readonly status: RunStatus; readonly text: string; readonly usage: { inputTokens: number; outputTokens: number }; + /** Provider- or gateway-reported cost for this run, in USD. */ + readonly costUsd?: number | undefined; readonly numTurns: number; readonly sdkSessionId?: string | undefined; readonly error?: string; @@ -123,6 +125,7 @@ export async function runAgent(req: RunRequest): Promise { let fullText = ""; let usage = { inputTokens: 0, outputTokens: 0 }; + let costUsd: number | undefined; let numTurns = 0; let sdkSessionId: string | undefined; let error: string | undefined; @@ -248,6 +251,7 @@ export async function runAgent(req: RunRequest): Promise { case "result": { const result = message as SDKResultMessage; sdkSessionId = result.session_id; + costUsd = Number.isFinite(result.total_cost_usd) ? result.total_cost_usd : undefined; if (result.subtype !== "success") { error = `result_${result.subtype}`; } @@ -261,6 +265,7 @@ export async function runAgent(req: RunRequest): Promise { status: error !== undefined ? "failed" : "completed", text: fullText, usage, + ...(costUsd !== undefined ? { costUsd } : {}), numTurns, sdkSessionId, ...(error !== undefined ? { error } : {}), @@ -271,6 +276,7 @@ export async function runAgent(req: RunRequest): Promise { status: aborted ? "interrupted" : "failed", text: fullText, usage, + ...(costUsd !== undefined ? { costUsd } : {}), numTurns, sdkSessionId, ...(aborted ? {} : { error: e instanceof Error ? e.message : String(e) }), diff --git a/hub/src/feishu/card/streaming-card.ts b/hub/src/feishu/card/streaming-card.ts index 907b375..59bbd2a 100644 --- a/hub/src/feishu/card/streaming-card.ts +++ b/hub/src/feishu/card/streaming-card.ts @@ -106,20 +106,23 @@ export class StreamingAgentCard { recordToolUseEnd({ runId: this.runId, ...params }); this.scheduleFlush(); } - async finish(fallbackText: string, options: { readonly interrupted?: boolean } = {}): Promise { + async finish(fallbackText: string, options: { readonly interrupted?: boolean; readonly footerText?: string | undefined } = {}): Promise { await this.flushChain; this.interrupted = options.interrupted === true; + const footerText = options.footerText ?? ""; + const fallbackWithFooter = appendFooter(fallbackText, footerText); try { let updated = true; if (this.text.length > 0) { + this.text = appendFooter(this.text, footerText); updated = await this.flushCard("complete", this.text); - } else if (this.currentMessageId === null && fallbackText.length > 0) { + } else if (this.currentMessageId === null && fallbackWithFooter.length > 0) { // No streaming text was sent. If we never created a card, send one now. - this.text = fallbackText; + this.text = fallbackWithFooter; updated = await this.flushCard("complete", this.text); } else if (this.currentMessageId !== null) { // Patch the existing card with the final text. - updated = await this.flushCard("complete", fallbackText); + updated = await this.flushCard("complete", fallbackWithFooter); } if (!updated && this.interrupted) { await sendText(this.rt, this.chatId, "\u5DF2\u4E2D\u65AD\u5F53\u524D\u8FD0\u884C\u3002", this.sendOptions); @@ -228,3 +231,9 @@ export class StreamingAgentCard { return "thinking"; } } + +function appendFooter(text: string, footerText: string): string { + if (footerText === "") return text; + if (text === "") return footerText; + return `${text.trimEnd()}\n\n${footerText}`; +} diff --git a/hub/src/feishu/slashCommands.ts b/hub/src/feishu/slashCommands.ts index f8b5c8b..264f852 100644 --- a/hub/src/feishu/slashCommands.ts +++ b/hub/src/feishu/slashCommands.ts @@ -1,5 +1,6 @@ import type { PrismaClient } from "@prisma/client"; import type { FastifyBaseLogger } from "fastify"; +import { decimalToNumberOrNull, formatInteger, formatUsd } from "../agent/cost.js"; import type { ModelRegistry, RoleEntry } from "../agent/models.js"; import type { RuntimeSettings } from "../settings/runtime.js"; import { sendText, type FeishuRuntime, type SendMessageOptions } from "./client.js"; @@ -33,6 +34,8 @@ export interface SlashCommandRegistryDeps { readonly triggerQueue: TriggerQueue; } +const TERMINAL_RUN_STATUSES = ["COMPLETED", "FAILED", "TIMED_OUT", "CANCELED"] as const; + export function parseSlashInvocation(prompt: string): SlashInvocation | null { const trimmed = prompt.trim(); if (!trimmed.startsWith("/")) return null; @@ -118,6 +121,51 @@ export function createSlashCommandRegistry(deps: SlashCommandRegistryDeps): Read }, }); + add({ + name: "cost", + usage: "/cost", + summary: "查看当前会话已记录的 agent 成本。", + details: [ + "只读取当前项目未归档 agent session 下已经结束的 run,不创建 agent run。", + "只统计运行时真实记录到 AgentRun.costUsd 的成本;未记录成本的 run 会单独列出。", + ], + run: async ({ invocation, projectId, chatId, rt, sendOptions }) => { + if (invocation.args.length > 0) { + await sendText(rt, chatId, ["用法错误: /cost 暂不接受参数。", "", formatBuiltinSlashCommandHelp(commands.get("cost")!)].join("\n"), sendOptions); + return; + } + + const sessions = await deps.prisma.agentSession.findMany({ + where: { projectId, archivedAt: null }, + orderBy: { updatedAt: "asc" }, + select: { id: true }, + }); + if (sessions.length === 0) { + await sendText(rt, chatId, "当前会话还没有 agent session。", sendOptions); + return; + } + + const runs = await deps.prisma.agentRun.findMany({ + where: { + projectId, + sessionId: { in: sessions.map((session) => session.id) }, + status: { in: [...TERMINAL_RUN_STATUSES] }, + finishedAt: { not: null }, + }, + orderBy: { finishedAt: "asc" }, + select: { + model: true, + provider: true, + inputTokens: true, + outputTokens: true, + costUsd: true, + }, + }); + + await sendText(rt, chatId, formatCostReport(runs), sendOptions); + }, + }); + add({ name: "reset", usage: "/reset", @@ -142,6 +190,95 @@ export function createSlashCommandRegistry(deps: SlashCommandRegistryDeps): Read return commands; } +interface CostReportRun { + readonly model: string; + readonly provider: string; + readonly inputTokens: number | null; + readonly outputTokens: number | null; + readonly costUsd: unknown; +} + +interface CostReportBucket { + readonly provider: string; + readonly model: string; + runs: number; + inputTokens: number; + outputTokens: number; + costUsd: number; +} + +function formatCostReport(runs: readonly CostReportRun[]): string { + if (runs.length === 0) { + return "当前会话还没有已结束的 agent run。"; + } + + const buckets = new Map(); + let recordedRuns = 0; + let unrecordedRuns = 0; + let totalInputTokens = 0; + let totalOutputTokens = 0; + let totalCostUsd = 0; + + for (const run of runs) { + const costUsd = decimalToNumberOrNull(run.costUsd); + if (costUsd === null) { + unrecordedRuns++; + continue; + } + + recordedRuns++; + const inputTokens = run.inputTokens ?? 0; + const outputTokens = run.outputTokens ?? 0; + totalInputTokens += inputTokens; + totalOutputTokens += outputTokens; + totalCostUsd += costUsd; + + const key = `${run.provider}\u0000${run.model}`; + let bucket = buckets.get(key); + if (bucket === undefined) { + bucket = { + provider: run.provider, + model: run.model, + runs: 0, + inputTokens: 0, + outputTokens: 0, + costUsd: 0, + }; + buckets.set(key, bucket); + } + bucket.runs++; + bucket.inputTokens += inputTokens; + bucket.outputTokens += outputTokens; + bucket.costUsd += costUsd; + } + + if (recordedRuns === 0) { + return [ + "当前会话已有已结束 agent run,但还没有任何 run 记录到真实成本。", + `未记录成本: ${formatInteger(unrecordedRuns)} runs。`, + "后续 run 需要 SDK 返回 total_cost_usd 才会进入 /cost 合计。", + ].join("\n"); + } + + const lines = [ + "当前会话已记录 agent 成本:", + `总计: ${formatUsd(totalCostUsd)}`, + `Runs: ${formatInteger(recordedRuns)} 已记录${unrecordedRuns > 0 ? ` / ${formatInteger(unrecordedRuns)} 未记录` : ""}`, + `Tokens: input ${formatInteger(totalInputTokens)} / output ${formatInteger(totalOutputTokens)}`, + "", + "按模型:", + ]; + + const sortedBuckets = [...buckets.values()].sort((a, b) => b.costUsd - a.costUsd); + for (const bucket of sortedBuckets) { + lines.push( + `- ${bucket.provider} / ${bucket.model}: ${formatInteger(bucket.runs)} runs, ${formatUsd(bucket.costUsd)}, input ${formatInteger(bucket.inputTokens)} / output ${formatInteger(bucket.outputTokens)}`, + ); + } + + return lines.join("\n"); +} + function formatHelpCommandInvocation( invocation: SlashInvocation, registry: ModelRegistry, diff --git a/hub/src/feishu/trigger.ts b/hub/src/feishu/trigger.ts index 44c8f26..dfccd44 100644 --- a/hub/src/feishu/trigger.ts +++ b/hub/src/feishu/trigger.ts @@ -32,6 +32,7 @@ import type { RuntimeSettings } from "../settings/runtime.js"; import { acquireLock, currentLockRunId, releaseLock } from "../lock.js"; import { createPermissionAuthorizer, type AuthorizationDecision, type PermissionAuthorizer } from "../permission.js"; import { writeAudit } from "../audit.js"; +import { formatRunCostLine } from "../agent/cost.js"; import { StreamingAgentCard } from "./card/streaming-card.js"; import { createFileDeliveryMcpServer } from "./fileDeliveryTool.js"; import { readFeishuContext } from "./read.js"; @@ -364,7 +365,7 @@ export function makeTriggerHandler(deps: TriggerDeps): TriggerHandler { : result.status === "failed" && result.error !== undefined ? `\u5904\u7406\u5931\u8D25: ${result.error}` : result.text; - await card.finish(finalText, { interrupted }); + await card.finish(finalText, { interrupted, footerText: formatRunCostLine(result.costUsd) }); const metadataPatch = sessionMetadataPatch(result.sdkSessionId); if (metadataPatch !== null) { await deps.prisma.agentSession.update({ @@ -378,6 +379,7 @@ export function makeTriggerHandler(deps: TriggerDeps): TriggerHandler { status: interrupted ? "CANCELED" : result.status === "completed" ? "COMPLETED" : result.status === "length" ? "TIMED_OUT" : "FAILED", inputTokens: result.usage.inputTokens, outputTokens: result.usage.outputTokens, + ...(result.costUsd !== undefined ? { costUsd: result.costUsd, costSource: "provider_reported" } : {}), error: result.error ?? null, finishedAt: new Date(), }, diff --git a/hub/test/integration/trigger.test.ts b/hub/test/integration/trigger.test.ts index 3757f34..124b6bb 100644 --- a/hub/test/integration/trigger.test.ts +++ b/hub/test/integration/trigger.test.ts @@ -73,6 +73,7 @@ function createMockRunAgent(calls: RunRequest[] = []): TestRunner { status: "completed", text: "mock response", usage: { inputTokens: 10, outputTokens: 5 }, + costUsd: 0.0023, numTurns: 1, sdkSessionId: "sdk-session-1", }; @@ -128,7 +129,7 @@ describe("trigger full lifecycle (integration)", () => { msgType: "interactive", replyInThread: undefined, }); - expect(rt.sentTexts).toContain("mock response"); + expect(rt.sentTexts.some((text) => text.includes("mock response") && text.includes("本次成本: $0.0023"))).toBe(true); expect(runAgentCalls).toHaveLength(1); expect(runAgentCalls[0]?.providerEnv).toMatchObject({ ANTHROPIC_BASE_URL: "https://openrouter.ai/api", @@ -136,6 +137,9 @@ describe("trigger full lifecycle (integration)", () => { ANTHROPIC_API_KEY: "", }); expect(runAgentCalls[0]?.maxTurns).toBe(7); + const run = await prisma.agentRun.findFirstOrThrow(); + expect(Number(run.costUsd)).toBeCloseTo(0.0023); + expect(run.costSource).toBe("provider_reported"); }); it("batches quick text messages from the same chat and sender into one run", async () => { @@ -228,6 +232,65 @@ describe("trigger full lifecycle (integration)", () => { expect(await prisma.agentRun.findMany()).toHaveLength(0); }); + it("/cost reports recorded current-session cost without creating a run", async () => { + await seedProject("proj-cost", "chat-cost"); + const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent, messageBatcherOptions: { maxMessages: 1 } }); + + await trigger(makeEvent("chat-cost", "@_user_1 写教案"), rt); + await vi.waitFor(async () => { + const runs = await prisma.agentRun.findMany(); + expect(runs).toHaveLength(1); + expect(runs[0]?.status).toBe("COMPLETED"); + }); + + await trigger(makeEvent("chat-cost", "@_user_1 /cost"), rt); + + const costText = rt.sentTexts.at(-1) ?? ""; + expect(costText).toContain("当前会话已记录 agent 成本"); + expect(costText).toContain("总计: $0.0023"); + expect(costText).toContain("Runs: 1 已记录"); + expect(costText).toContain("openrouter / mock-model"); + expect(runAgentCalls).toHaveLength(1); + expect(await prisma.agentRun.findMany()).toHaveLength(1); + }); + + it("/cost surfaces finished runs without recorded cost", async () => { + await seedProject("proj-cost-missing", "chat-cost-missing"); + const session = await prisma.agentSession.create({ + data: { + projectId: "proj-cost-missing", + provider: "openrouter", + roleId: "draft", + model: "mock-model", + metadata: {}, + }, + select: { id: true }, + }); + await prisma.agentRun.create({ + data: { + projectId: "proj-cost-missing", + sessionId: session.id, + entrypoint: "FEISHU", + status: "COMPLETED", + prompt: "old test run", + model: "mock-model", + provider: "openrouter", + inputTokens: 10, + outputTokens: 5, + metadata: {}, + finishedAt: new Date(), + }, + }); + const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent, messageBatcherOptions: { maxMessages: 1 } }); + + await trigger(makeEvent("chat-cost-missing", "@_user_1 /cost"), rt); + + const costText = rt.sentTexts.at(-1) ?? ""; + expect(costText).toContain("还没有任何 run 记录到真实成本"); + expect(costText).toContain("未记录成本: 1 runs"); + expect(runAgentCalls).toHaveLength(0); + }); + it("/help is built from the current registry on each request", async () => { await seedProject("proj-help-live", "chat-help-live"); let currentModels = new InMemoryModelRegistry( diff --git a/hub/test/unit/runner.test.ts b/hub/test/unit/runner.test.ts index 902fcb7..af94ee6 100644 --- a/hub/test/unit/runner.test.ts +++ b/hub/test/unit/runner.test.ts @@ -21,11 +21,12 @@ function assistantMessage(text: string) { }; } -function resultMessage(sessionId: string) { +function resultMessage(sessionId: string, costUsd?: number) { return { type: "result", subtype: "success", session_id: sessionId, + ...(costUsd !== undefined ? { total_cost_usd: costUsd } : {}), }; } @@ -103,6 +104,22 @@ describe("runAgent", () => { expect(call?.options).not.toHaveProperty("resume"); }); + it("returns SDK-reported cost when present", async () => { + queryMock.mockReturnValue(messages(assistantMessage("ok"), resultMessage("sdk-session-1", 0.0042))); + + const result = await runAgent({ + prompt: "算一下成本", + model: undefined, + project: { projectId: "p", boundChatId: "c", workspaceDir: "/tmp/ws" }, + systemPrompt: undefined, + runId: "run-1", + sessionId: "hub-session-1", + prisma: stubPrisma, + }); + + expect(result.costUsd).toBe(0.0042); + }); + it("passes provider env per run without mutating process env", async () => { delete process.env["CPH_TEST_PROVIDER_ENV_MUTATION"]; queryMock.mockReturnValue(messages(assistantMessage("ok"), resultMessage("sdk-session-1")));