From 41eae0de12c9ceddc37ff902d8fedae607822172 Mon Sep 17 00:00:00 2001 From: zuiho <2324465096@qq.com> Date: Sat, 26 Sep 2026 11:13:12 +0800 Subject: [PATCH 1/2] feat: add channel cache and cost statistics with model pricing --- apps/desktop/src-tauri/src/lib.rs | 11 + apps/desktop/src-tauri/src/sidecar.rs | 18 + apps/desktop/src/ServiceManager.tsx | 5 + .../src/ServiceStatisticsDialog.test.tsx | 127 ++++ apps/desktop/src/ServiceStatisticsDialog.tsx | 548 ++++++++++++++++++ .../desktop/src/components/ServiceListRow.tsx | 6 +- apps/desktop/src/pricing-model.test.ts | 84 ++- apps/desktop/src/pricing-model.ts | 99 ++++ contracts/control-api.openapi.yaml | 170 ++++++ .../internal/controlapi/service_statistics.go | 38 ++ core/internal/controlapi/services.go | 2 + core/internal/pricing/model_rates.go | 16 + core/internal/pricing/types.go | 27 +- core/internal/storage/service_statistics.go | 34 ++ core/internal/storage/sqlite/pricing.go | 8 + .../storage/sqlite/service_statistics.go | 128 ++++ .../storage/sqlite/service_statistics_test.go | 102 ++++ 17 files changed, 1403 insertions(+), 20 deletions(-) create mode 100644 apps/desktop/src/ServiceStatisticsDialog.test.tsx create mode 100644 apps/desktop/src/ServiceStatisticsDialog.tsx create mode 100644 core/internal/controlapi/service_statistics.go create mode 100644 core/internal/pricing/model_rates.go create mode 100644 core/internal/storage/service_statistics.go create mode 100644 core/internal/storage/sqlite/service_statistics.go create mode 100644 core/internal/storage/sqlite/service_statistics_test.go diff --git a/apps/desktop/src-tauri/src/lib.rs b/apps/desktop/src-tauri/src/lib.rs index 9890604..c1548aa 100644 --- a/apps/desktop/src-tauri/src/lib.rs +++ b/apps/desktop/src-tauri/src/lib.rs @@ -879,6 +879,16 @@ async fn delete_service( manager.delete_service(&service_id, &etag).await } +#[tauri::command] +async fn service_statistics( + service_id: String, + from: String, + to: String, + manager: State<'_, Arc>, +) -> Result { + manager.service_statistics(&service_id, &from, &to).await +} + #[tauri::command] async fn pricing( operation: String, @@ -1442,6 +1452,7 @@ pub fn run() { delete_service, get_service_usage, pricing, + service_statistics, reset_service_usage, test_service, probe_service_models, diff --git a/apps/desktop/src-tauri/src/sidecar.rs b/apps/desktop/src-tauri/src/sidecar.rs index ab5bb7a..4a7a7e2 100644 --- a/apps/desktop/src-tauri/src/sidecar.rs +++ b/apps/desktop/src-tauri/src/sidecar.rs @@ -1659,6 +1659,24 @@ impl CoreManager { Ok(()) } + pub async fn service_statistics( + &self, + service_id: &str, + from: &str, + to: &str, + ) -> Result { + validate_resource_id(service_id)?; + let path = format!( + "/control/v1/services/{service_id}/statistics?from={}&to={}", + percent_encode_query(from), + percent_encode_query(to) + ); + let (_, body) = self + .authenticated_control(Method::GET, &path, None, None) + .await?; + serde_json::from_slice(&body).map_err(|_| "statistics returned invalid JSON".into()) + } + pub async fn pricing( &self, operation: &str, diff --git a/apps/desktop/src/ServiceManager.tsx b/apps/desktop/src/ServiceManager.tsx index c8475e3..0958e72 100644 --- a/apps/desktop/src/ServiceManager.tsx +++ b/apps/desktop/src/ServiceManager.tsx @@ -6,6 +6,7 @@ import { validProxyDraft, type ProxyDraft, } from "./service-proxy-model"; +import { ServiceStatisticsEntry } from "./ServiceStatisticsDialog"; import { ServiceTestDialog } from "./ServiceTestDialog"; import { PricingWorkspace, ServiceBillingMeter } from "./PricingWorkspace"; import { useServiceOrder } from "./use-service-order"; @@ -1869,6 +1870,10 @@ export function ServiceManager({ } actions={ <> + ({ invoke: vi.fn(), billing: vi.fn() })); +vi.mock("@tauri-apps/api/core", () => ({ invoke: mocks.invoke })); +vi.mock("./pricing-bridge", () => ({ getServiceBilling: mocks.billing })); +import { ServiceStatisticsEntry } from "./ServiceStatisticsDialog"; +let root: Root; +let host: HTMLDivElement; +const service = { + id: "service_price", + name: "Price test", + models: ["gpt-6-astra"], +} as Service; +beforeEach(() => { + ( + globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean } + ).IS_REACT_ACT_ENVIRONMENT = true; + host = document.createElement("div"); + document.body.append(host); + root = createRoot(host); + mocks.billing.mockResolvedValue({ + config: { + provider: "", + bindings: {}, + billing_day: 1, + time_zone: "UTC", + monthly_budget_usd: "", + }, + }); + mocks.invoke.mockImplementation( + async (command: string, args: { operation?: string }) => { + if (command === "service_statistics") return undefined; + if (args.operation === "catalog") + return { + prices: [ + { + provider: "openai", + model: "gpt-6-astra", + name: "GPT-6 Astra", + expression: + 'len <= 272000 ? tier("small", p * 10 + cr * 1 + cc * 12.5 + c * 50) : tier("large", p * 20 + cr * 2 + cc * 25 + c * 75)', + }, + ], + }; + return { ok: true }; + }, + ); +}); +afterEach(async () => { + await act(async () => root.unmount()); + host.remove(); + vi.clearAllMocks(); +}); +const button = (text: string) => + [...document.querySelectorAll("button")].find( + (b) => b.textContent === text, + )!; +async function openPrices(models: string[]) { + await act(async () => + root.render( + , + ), + ); + await act(async () => + document + .querySelector( + '[aria-label="缓存与费用 · Price test"]', + )! + .click(), + ); + await act(async () => + button("模型价格").dispatchEvent( + new MouseEvent("mousedown", { bubbles: true, button: 0 }), + ), + ); +} +it("shows effective catalog tiers before offering explicit channel overrides", async () => { + await openPrices(["gpt-6-astra"]); + expect(document.body.textContent).toContain( + "来源：价格目录 · OpenAI · GPT-6 Astra", + ); + expect(document.body.textContent).toContain("$12.5"); + expect(document.body.textContent).toContain("$75"); + expect(document.querySelector('[aria-label="输入价格"]')).toBeNull(); + await act(async () => button("设置渠道自定义价格").click()); + for (const label of ["输入", "输出", "缓存读取", "缓存写入"]) { + const input = document.querySelector( + `[aria-label="${label}价格"]`, + )!; + await act(async () => { + Object.getOwnPropertyDescriptor( + HTMLInputElement.prototype, + "value", + )!.set!.call(input, "2"); + input.dispatchEvent(new Event("input", { bubbles: true })); + }); + } + await act(async () => button("保存模型价格").click()); + expect(mocks.invoke).toHaveBeenCalledWith( + "pricing", + expect.objectContaining({ + operation: "configure", + input: expect.objectContaining({ + overrides: { + "gpt-6-astra": { + input: "2", + output: "2", + cache_read: "2", + cache_write: "2", + }, + }, + }), + }), + ); + expect(document.body.textContent).toContain("来源：渠道自定义价格"); +}); +it("marks an unmatched model unpriced instead of showing zero or unrelated catalog rates", async () => { + await openPrices(["deepseek-v4.1-flash"]); + expect(document.body.textContent).toContain("未配置价格"); + expect(document.body.textContent).not.toContain("$10"); +}); diff --git a/apps/desktop/src/ServiceStatisticsDialog.tsx b/apps/desktop/src/ServiceStatisticsDialog.tsx new file mode 100644 index 0000000..adce559 --- /dev/null +++ b/apps/desktop/src/ServiceStatisticsDialog.tsx @@ -0,0 +1,548 @@ +import { useEffect, useState } from "react"; +import { invoke } from "@tauri-apps/api/core"; +import type { Service } from "./service-model"; +import { getServiceBilling } from "./pricing-bridge"; +import { + formatUSD, + resolveCatalogPrice, + catalogRateRows, + OFFICIAL_PROVIDERS, + type CatalogPrice, + type ModelRates, + type PricingConfig, +} from "./pricing-model"; +import { Button } from "./components/ui/button"; +import { IconButton } from "./components/IconButton"; +import { CircleDollarSign } from "./components/icons"; +import { Input } from "./components/ui/input"; +import { + Dialog, + DialogContent, + DialogTitle, + DialogDescription, +} from "./components/ui/dialog"; +import { Tabs, TabsContent, TabsList, TabsTrigger } from "./components/ui/tabs"; +import { Field } from "./components/Field"; +import { FilterSelect } from "./components/FilterSelect"; +import { ModelSelect } from "./components/ModelSelect"; +import { Panel } from "./components/Panel"; +import { + Table, + TableBody, + TableCell, + TableHead, + TableHeader, + TableRow, +} from "./components/ui/table"; + +interface Totals { + records: number; + priced: number; + unpriced: number; + amount_usd: string; + per_million_usd: string | null; + cost_samples: number; + input_tokens: number; + output_tokens: number; + cache_samples: number; + cache_hits: number; + cache_read_tokens: number; + cache_input_tokens: number; + first_token_ms: number | null; + first_token_samples: number; + duration_ms: number | null; + duration_samples: number; + tps: number | null; + tps_samples: number; +} +interface Statistics extends Totals { + from: string; + to: string; + models: (Totals & { model: string })[]; +} +const percent = (n: number, d: number) => + d > 0 ? `${((n / d) * 100).toFixed(1)}%` : "—"; +const amount = (s: Totals) => (s.priced ? formatUSD(s.amount_usd) : "—"); +const number = (n: number) => n.toLocaleString(); +const ms = (n: number | null) => + n == null ? "—" : `${(n / 1000).toFixed(2)}s`; +const localDate = (date: Date) => + new Date(date.getTime() - date.getTimezoneOffset() * 60000) + .toISOString() + .slice(0, 16); + +export function ServiceStatisticsEntry({ + service, + disabled, +}: { + service: Service; + disabled: boolean; +}) { + const [open, setOpen] = useState(false); + return ( + <> + setOpen(true)} + > + + {open && ( + setOpen(false)} + /> + )} + + ); +} +function ServiceStatisticsDialog({ + service, + onClose, +}: { + service: Service; + onClose: () => void; +}) { + const [range, setRange] = useState("1"); + const [from, setFrom] = useState(() => + localDate(new Date(Date.now() - 86400000)), + ); + const [to, setTo] = useState(() => localDate(new Date())); + const [stats, setStats] = useState(); + const [config, setConfig] = useState(); + const [catalog, setCatalog] = useState([]); + const [catalogError, setCatalogError] = useState(""); + const [customize, setCustomize] = useState(false); + const [model, setModel] = useState(service.models[0] ?? ""); + const [rate, setRate] = useState({ + input: "", + output: "", + cache_read: "", + cache_write: "", + }); + const [error, setError] = useState(""); + const [message, setMessage] = useState(""); + const [busy, setBusy] = useState(false); + const refresh = async () => { + setBusy(true); + setError(""); + try { + const end = range === "custom" ? new Date(to) : new Date(); + const begin = + range === "custom" + ? new Date(from) + : new Date(end.getTime() - Number(range) * 86400000); + if ( + !Number.isFinite(begin.getTime()) || + !Number.isFinite(end.getTime()) || + end <= begin || + end.getTime() - begin.getTime() > 31 * 86400000 + ) + throw new Error("请选择不超过 31 天的时间范围"); + begin.setMilliseconds(0); + end.setMilliseconds(0); + setStats( + await invoke("service_statistics", { + serviceId: service.id, + from: begin.toISOString(), + to: end.toISOString(), + }), + ); + } catch (e) { + setError(String(e)); + } finally { + setBusy(false); + } + }; + useEffect(() => { + void refresh(); + void getServiceBilling(service.id) + .then((r) => setConfig(r.config)) + .catch((e) => setError(String(e))); + void invoke<{ prices: CatalogPrice[] }>("pricing", { operation: "catalog" }) + .then((value) => setCatalog(value.prices ?? [])) + .catch(() => setCatalogError("价格目录读取失败，请重新打开重试")); + }, [service.id]); + useEffect(() => { + setCustomize(!!config?.overrides?.[model]); + setRate( + config?.overrides?.[model] ?? { + input: "", + output: "", + cache_read: "", + cache_write: "", + }, + ); + }, [model, config]); + const catalogPrice = config + ? resolveCatalogPrice(config, model, catalog) + : undefined; + const override = config?.overrides?.[model]; + const effectiveRows = override + ? [{ label: "渠道自定义价格", rates: override }] + : catalogRateRows(catalogPrice?.expression ?? ""); + const save = async (remove = false) => { + if (!config || !model.trim()) return; + setBusy(true); + setError(""); + setMessage(""); + try { + const overrides = { ...config.overrides }; + if (remove) delete overrides[model]; + else overrides[model] = rate; + const next = { ...config, overrides }; + await invoke("pricing", { + operation: "configure", + serviceId: service.id, + input: next, + }); + setConfig(next); + setMessage("模型价格已保存，将用于新请求。已计价的历史费用保持原金额。"); + } catch (e) { + setError(String(e)); + } finally { + setBusy(false); + } + }; + return ( + { + if (!v) onClose(); + }} + > + +
+ 缓存与费用 · {service.name} + + 按渠道查看缓存命中、费用和模型明细 + +
+ +
+ + 统计 + 模型价格 + +
+ {error && ( +

+ {error} +

+ )} + +
+ + {range === "custom" && ( + <> + setFrom(e.target.value)} + /> + setTo(e.target.value)} + /> + + )} + +
+
+ {stats && ( + <> +

+ {new Date(stats.from).toLocaleString()} —{" "} + {new Date(stats.to).toLocaleString()} +

+
+ {[ + ["总费用", amount(stats)], + [ + "每 1M Token 费用", + formatUSD(stats.per_million_usd ?? undefined), + ], + [ + "Token 缓存命中率", + percent( + stats.cache_read_tokens, + stats.cache_input_tokens, + ), + ], + [ + "缓存命中请求占比", + percent(stats.cache_hits, stats.cache_samples), + ], + [ + "缓存读取 / 总输入 Token", + `${number(stats.cache_read_tokens)} / ${number(stats.cache_input_tokens)}`, + ], + [ + "缓存命中 / 有效请求", + `${number(stats.cache_hits)} / ${number(stats.cache_samples)}`, + ], + [ + "输入 / 输出 Token", + `${number(stats.input_tokens)} / ${number(stats.output_tokens)}`, + ], + [ + "平均首字 / 耗时 / TPS", + `${ms(stats.first_token_ms)} / ${ms(stats.duration_ms)} / ${stats.tps?.toFixed(1) ?? "—"}`, + ], + ].map(([label, value]) => ( + +
+ {label} +
+
+ {value} +
+ + ))} +
+
+ + + + {[ + "模型", + "请求", + "费用", + "每 1M Token", + "输入 / 输出", + "Token 命中率", + "请求命中率", + "首字", + "耗时", + "TPS", + ].map((h) => ( + + {h} + + ))} + + + + {stats.models.map((row) => ( + + + {row.model || "未知模型"} + + {number(row.records)} + {amount(row)} + + {formatUSD(row.per_million_usd ?? undefined)} + + + {number(row.input_tokens)} /{" "} + {number(row.output_tokens)} + + + {percent( + row.cache_read_tokens, + row.cache_input_tokens, + )} + + + {percent(row.cache_hits, row.cache_samples)} + + {ms(row.first_token_ms)} + {ms(row.duration_ms)} + {row.tps?.toFixed(1) ?? "—"} + + ))} + +
+
+ {!stats.records && ( +

+ 此时间范围暂无请求记录 +

+ )} +

+ 总记录 {stats.records} · 缓存统计排除{" "} + {stats.records - stats.cache_samples} · 未计价{" "} + {stats.unpriced} · 每百万 Token 统计样本{" "} + {stats.cost_samples} +

+
+ 统计口径与样本 +

+ 输入与输出 Token、每百万 Token + 费用取同一批可计价记录。缓存字段缺失的记录不计入命中率。每次实际上游调用单独统计；重试按实际渠道归属。TPS + 为各请求输出 Token / + 完整请求秒数的平均值。历史请求明细清理后，费用和 Token + 仍保留，耗时仅取现存明细。 +

+

+ 首字样本 {stats.first_token_samples} · 耗时样本{" "} + {stats.duration_samples} · TPS 样本 {stats.tps_samples} +

+
+ + )} +
+ + +
+ + + +

+ 价格单位：USD / 1M Token。渠道自定义价格优先于价格目录。 +

+ +
当前生效价格
+

+ {override + ? "来源：渠道自定义价格" + : catalogPrice + ? `来源：价格目录 · ${OFFICIAL_PROVIDERS[catalogPrice.provider] ?? catalogPrice.provider} · ${catalogPrice.name}` + : catalogError || + "未配置价格：该模型没有自定义价格，也未匹配到目录价格；新请求将记录为未计价。"} +

+ {effectiveRows.map((row) => ( +
+ {row.label} +
+ {( + [ + ["input", "输入"], + ["output", "输出"], + ["cache_read", "缓存读取"], + ["cache_write", "缓存写入"], + ] as const + ).map(([key, label]) => ( +
+ {label} +
+ ${row.rates[key]} +
+
+ ))} +
+
+ ))} + {!override && catalogPrice && ( +
+ 查看完整目录计价规则 +

+ {catalogPrice.expression} +

+

+ p 输入 · c 输出 · cr 缓存读取 · cc 缓存写入；按每百万 + Token 计价。 +

+
+ )} + + {!customize && ( + + )} + {customize && ( + <> +

+ 渠道自定义价格（保存后覆盖目录价格） +

+
+ {( + [ + ["input", "输入"], + ["output", "输出"], + ["cache_read", "缓存读取"], + ["cache_write", "缓存写入"], + ] as const + ).map(([key, label]) => ( + + + setRate({ ...rate, [key]: e.target.value }) + } + /> + + ))} +
+
+ + +
+ + )} + {message && ( +

+ {message} +

+ )} +
+ + + + + ); +} diff --git a/apps/desktop/src/components/ServiceListRow.tsx b/apps/desktop/src/components/ServiceListRow.tsx index 0942104..ad6b491 100644 --- a/apps/desktop/src/components/ServiceListRow.tsx +++ b/apps/desktop/src/components/ServiceListRow.tsx @@ -3,10 +3,10 @@ import type { ReactNode } from "react"; import { DataRow } from "@/components/DataRow"; import { cn } from "@/lib/utils"; -// Share spare width across the content columns and reserve all three actions. +// Share spare width across the content columns and reserve all four actions. // Query the scroller itself, including the space taken by its scrollbar. const columns = - "@[860px]/service-list:grid-cols-[3.25rem_minmax(0,1.25fr)_minmax(6rem,0.5fr)_minmax(9.5rem,0.95fr)_minmax(6.5rem,0.6fr)_3.75rem_6.25rem]"; + "@[860px]/service-list:grid-cols-[3.25rem_minmax(0,1.25fr)_minmax(6rem,0.5fr)_minmax(9.5rem,0.95fr)_minmax(6.5rem,0.6fr)_3.75rem_8.75rem]"; export function ServiceListHeader({ labels }: { labels: readonly string[] }) { return ( @@ -77,7 +77,7 @@ export function ServiceListRow({ diff --git a/apps/desktop/src/pricing-model.test.ts b/apps/desktop/src/pricing-model.test.ts index bbc22e7..720d445 100644 --- a/apps/desktop/src/pricing-model.test.ts +++ b/apps/desktop/src/pricing-model.test.ts @@ -6,6 +6,8 @@ import { billingAmount, currentBillingPeriod, formatUSD, + resolveCatalogPrice, + catalogRateRows, type BillingPeriod, } from "./pricing-model"; const config = { @@ -17,6 +19,52 @@ const config = { billing_day: 31, time_zone: "Asia/Shanghai", }; +it("shows the same unique catalog binding used for billing and preserves tier prices", () => { + const price = { + provider: "openai", + model: "gpt-6-astra", + name: "GPT-6 Astra", + expression: + 'len <= 272000 ? tier("0_272k", p * 10 + cr * 1 + cc * 12.5 + c * 50) : tier("272k_plus", p * 20 + cr * 2 + cc * 25 + c * 75)', + }; + expect( + resolveCatalogPrice( + { ...config, provider: "", bindings: {} }, + price.model, + [price], + ), + ).toBe(price); + expect( + resolveCatalogPrice( + { ...config, provider: "", bindings: {} }, + "deepseek-v4.1-flash", + [price], + ), + ).toBeUndefined(); + expect( + resolveCatalogPrice( + { ...config, provider: "", bindings: {} }, + price.model, + [price, { ...price, provider: "other" }], + ), + ).toBeUndefined(); + expect(catalogRateRows(price.expression).map((row) => row.rates)).toEqual([ + { input: "10", output: "50", cache_read: "1", cache_write: "12.5" }, + { input: "20", output: "75", cache_read: "2", cache_write: "25" }, + ]); + expect(catalogRateRows('hour("UTC") < 8 ? p * 1 : p * 2')).toEqual([]); + expect(catalogRateRows("p * 2 + c * 8")[0].rates).toEqual({ + input: "2", + output: "8", + cache_read: "2", + cache_write: "2", + }); + expect( + catalogRateRows( + 'len <= 100 ? tier("low", p * 2 + c * 8) : tier("high", p * 4 + cr * 1 + c * 16)', + )[0].rates.cache_read, + ).toBe("0"); +}); describe("official pricing boundary", () => { it("preserves money as decimal strings and accepts only canonical suppliers", () => { expect(parsePricingConfig(config)).toEqual(config); @@ -100,11 +148,21 @@ describe("official pricing boundary", () => { coverage: "complete", summary, }; - expect(parseServiceBilling({ config, periods: [period] }).periods[0].summary.by_token).toEqual([]); + expect( + parseServiceBilling({ config, periods: [period] }).periods[0].summary + .by_token, + ).toEqual([]); for (const by_token of [undefined, null, {}]) { const invalid = { ...summary, by_token }; - expect(() => parseBillingSummary(invalid)).toThrow("Invalid pricing list"); - expect(() => parseServiceBilling({ config, periods: [{ ...period, summary: invalid }] })).toThrow("Invalid pricing list"); + expect(() => parseBillingSummary(invalid)).toThrow( + "Invalid pricing list", + ); + expect(() => + parseServiceBilling({ + config, + periods: [{ ...period, summary: invalid }], + }), + ).toThrow("Invalid pricing list"); } }); it("parses token billing groups with the stable token_id field", () => { @@ -118,15 +176,17 @@ describe("official pricing boundary", () => { from: "2026-09-19T00:00:00Z", to: "2026-09-20T00:00:00Z", by_model: [], - by_token: [{ - token_id: "token_a", - amount_usd: "1.25", - priced: 1, - unpriced: 0, - pending: 0, - revalued: 0, - requests: 1, - }], + by_token: [ + { + token_id: "token_a", + amount_usd: "1.25", + priced: 1, + unpriced: 0, + pending: 0, + revalued: 0, + requests: 1, + }, + ], }; expect(parseBillingSummary(value).by_token).toEqual(value.by_token); }); diff --git a/apps/desktop/src/pricing-model.ts b/apps/desktop/src/pricing-model.ts index 8b37fc5..db22e8d 100644 --- a/apps/desktop/src/pricing-model.ts +++ b/apps/desktop/src/pricing-model.ts @@ -3,12 +3,94 @@ export interface PriceBinding { model: string; } export interface PricingConfig { + overrides?: Record; provider: string; bindings: Record; monthly_budget_usd: string; billing_day: number; time_zone: string; } +export interface ModelRates { + input: string; + output: string; + cache_read: string; + cache_write: string; +} +export interface CatalogPrice { + provider: string; + model: string; + name: string; + expression: string; +} +export function resolveCatalogPrice( + config: PricingConfig, + model: string, + prices: CatalogPrice[], +) { + const binding = config.bindings[model] ?? { + provider: config.provider, + model, + }; + const matches = prices.filter( + (price) => + price.model === binding.model && + (!binding.provider || price.provider === binding.provider), + ); + return matches.length === 1 ? matches[0] : undefined; +} +// Display only recognized token-rate formulas; other catalog rules remain +// visible as their exact expression, never guessed as a flat token price. +export function catalogRateRows( + expression: string, +): { label: string; rates: ModelRates }[] { + const linear = (source: string): ModelRates | undefined => { + const rates: ModelRates = { + input: "0", + output: "0", + cache_read: "0", + cache_write: "0", + }; + const keys = { + p: "input", + c: "output", + cr: "cache_read", + cc: "cache_write", + } as const; + const seen = new Set(); + for (const term of source.split("+")) { + const match = term.trim().match(/^(p|c|cr|cc)\s*\*\s*(\d+(?:\.\d+)?)$/); + if (!match || seen.has(match[1])) return undefined; + seen.add(match[1]); + rates[keys[match[1] as keyof typeof keys]] = match[2]; + } + // Core keeps cache tokens in p unless the formula prices them separately. + if (!/\bcr\b/.test(expression)) rates.cache_read = rates.input; + if (!/\bcc\b/.test(expression)) rates.cache_write = rates.input; + return rates; + }; + const tiered = expression.match( + /^len\s*<=\s*(\d+)\s*\?\s*tier\("[^"]+",\s*([^()]+)\)\s*:\s*tier\("[^"]+",\s*([^()]+)\)$/, + ); + if (tiered) { + const low = linear(tiered[2]), + high = linear(tiered[3]); + if (low && high) + return [ + { + label: `输入 ≤ ${Number(tiered[1]).toLocaleString()} Token`, + rates: low, + }, + { + label: `输入 > ${Number(tiered[1]).toLocaleString()} Token`, + rates: high, + }, + ]; + return []; + } + const simple = expression.match(/^tier\("[^"]+",\s*([^()]+)\)$/); + const rates = linear(simple?.[1] ?? expression); + return rates ? [{ label: "标准价格", rates }] : []; +} export interface BillingAmounts { amount_usd: string; priced: number; @@ -183,7 +265,24 @@ export function parsePricingConfig(value: unknown): PricingConfig { throw new Error("Unsupported official provider"); const day = count(o.billing_day); if (day < 1 || day > 31) throw new Error("Invalid billing day"); + const overrides = Object.fromEntries( + Object.entries(o.overrides == null ? {} : object(o.overrides)).map( + ([model, value]) => { + const rate = object(value); + return [ + model, + { + input: decimal(rate.input), + output: decimal(rate.output), + cache_read: decimal(rate.cache_read), + cache_write: decimal(rate.cache_write), + }, + ]; + }, + ), + ); return { + ...(o.overrides == null ? {} : { overrides }), provider, bindings, monthly_budget_usd: decimal(o.monthly_budget_usd, true), diff --git a/contracts/control-api.openapi.yaml b/contracts/control-api.openapi.yaml index 898588c..c69833e 100644 --- a/contracts/control-api.openapi.yaml +++ b/contracts/control-api.openapi.yaml @@ -44,6 +44,7 @@ security: x-astrlink-implementation: milestone: M3-unified-service-slice implemented_operations: + - "GET /control/v1/services/{service_id}/statistics" - GET /control/v1/pricing/status - GET /control/v1/pricing/catalog - POST /control/v1/pricing/sync @@ -126,6 +127,49 @@ tags: - name: Requests description: Request metadata and explicitly enabled encrypted audit content. paths: + "/control/v1/services/{service_id}/statistics": + { + "parameters": + [ + { "$ref": "#/components/parameters/ServiceId" }, + { + "name": "from", + "in": "query", + "required": true, + "schema": { "type": "string", "format": "date-time" }, + }, + { + "name": "to", + "in": "query", + "required": true, + "schema": { "type": "string", "format": "date-time" }, + }, + ], + "get": + { + "operationId": "getServiceStatistics", + "x-astrlink-implementation-status": "implemented", + "responses": + { + "200": + { + "description": "Success", + "content": + { + "application/json": + { + "schema": + { + "$ref": "#/components/schemas/ServiceStatistics", + }, + }, + }, + }, + "400": { "$ref": "#/components/responses/BadRequest" }, + "401": { "$ref": "#/components/responses/Unauthorized" }, + }, + }, + } /control/v1/pricing/status: get: x-astrlink-implementation-status: implemented @@ -2679,6 +2723,127 @@ components: schema: $ref: "#/components/schemas/ErrorResponse" schemas: + "ServiceStatistics": + { + "type": "object", + "required": + [ + "records", + "priced", + "unpriced", + "cost_samples", + "input_tokens", + "output_tokens", + "cache_samples", + "cache_hits", + "cache_read_tokens", + "cache_input_tokens", + "first_token_samples", + "duration_samples", + "tps_samples", + "amount_usd", + "per_million_usd", + "first_token_ms", + "duration_ms", + "tps", + "from", + "to", + "models", + ], + "properties": + { + "records": { "type": "integer", "minimum": 0 }, + "priced": { "type": "integer", "minimum": 0 }, + "unpriced": { "type": "integer", "minimum": 0 }, + "cost_samples": { "type": "integer", "minimum": 0 }, + "input_tokens": { "type": "integer", "minimum": 0 }, + "output_tokens": { "type": "integer", "minimum": 0 }, + "cache_samples": { "type": "integer", "minimum": 0 }, + "cache_hits": { "type": "integer", "minimum": 0 }, + "cache_read_tokens": { "type": "integer", "minimum": 0 }, + "cache_input_tokens": { "type": "integer", "minimum": 0 }, + "first_token_samples": { "type": "integer", "minimum": 0 }, + "duration_samples": { "type": "integer", "minimum": 0 }, + "tps_samples": { "type": "integer", "minimum": 0 }, + "amount_usd": { "type": "string" }, + "per_million_usd": { "type": ["string", "null"] }, + "first_token_ms": { "type": ["number", "null"] }, + "duration_ms": { "type": ["number", "null"] }, + "tps": { "type": ["number", "null"] }, + "from": { "type": "string" }, + "to": { "type": "string" }, + "models": + { + "type": "array", + "items": + { + "type": "object", + "required": + [ + "records", + "priced", + "unpriced", + "cost_samples", + "input_tokens", + "output_tokens", + "cache_samples", + "cache_hits", + "cache_read_tokens", + "cache_input_tokens", + "first_token_samples", + "duration_samples", + "tps_samples", + "amount_usd", + "per_million_usd", + "first_token_ms", + "duration_ms", + "tps", + "model", + ], + "properties": + { + "records": { "type": "integer", "minimum": 0 }, + "priced": { "type": "integer", "minimum": 0 }, + "unpriced": { "type": "integer", "minimum": 0 }, + "cost_samples": { "type": "integer", "minimum": 0 }, + "input_tokens": { "type": "integer", "minimum": 0 }, + "output_tokens": { "type": "integer", "minimum": 0 }, + "cache_samples": { "type": "integer", "minimum": 0 }, + "cache_hits": { "type": "integer", "minimum": 0 }, + "cache_read_tokens": + { "type": "integer", "minimum": 0 }, + "cache_input_tokens": + { "type": "integer", "minimum": 0 }, + "first_token_samples": + { "type": "integer", "minimum": 0 }, + "duration_samples": { "type": "integer", "minimum": 0 }, + "tps_samples": { "type": "integer", "minimum": 0 }, + "amount_usd": { "type": "string" }, + "per_million_usd": { "type": ["string", "null"] }, + "first_token_ms": { "type": ["number", "null"] }, + "duration_ms": { "type": ["number", "null"] }, + "tps": { "type": ["number", "null"] }, + "model": { "type": "string" }, + }, + "additionalProperties": false, + }, + }, + }, + "additionalProperties": false, + } + "ModelRates": + { + "type": "object", + "required": ["input", "output", "cache_read", "cache_write"], + "properties": + { + "input": { "type": "string" }, + "output": { "type": "string" }, + "cache_read": { "type": "string" }, + "cache_write": { "type": "string" }, + }, + "additionalProperties": false, + } PricingStatus: type: object required: @@ -2725,6 +2890,11 @@ components: type: object required: [provider, bindings, monthly_budget_usd, billing_day, time_zone] properties: + overrides: + { + type: object, + additionalProperties: { $ref: "#/components/schemas/ModelRates" }, + } provider: { type: string, diff --git a/core/internal/controlapi/service_statistics.go b/core/internal/controlapi/service_statistics.go new file mode 100644 index 0000000..56f5e0f --- /dev/null +++ b/core/internal/controlapi/service_statistics.go @@ -0,0 +1,38 @@ +package controlapi + +import ( + "context" + "github.com/QuantumNous/astrlink/core/contract" + "github.com/QuantumNous/astrlink/core/internal/storage" + "net/http" + "time" +) + +type serviceStatisticsStore interface { + ServiceStatistics(context.Context, contract.ServiceID, time.Time, time.Time) (storage.ServiceStatistics, error) +} + +func (h *Handler) serviceStatistics(w http.ResponseWriter, r *http.Request, id contract.ServiceID) { + if r.Method != "GET" { + writeMethodNotAllowed(w, "GET") + return + } + store, ok := h.serviceStore.(serviceStatisticsStore) + if !ok { + writeError(w, 503, "statistics_unavailable", "统计不可用") + return + } + q := r.URL.Query() + from, e1 := time.Parse(time.RFC3339, q.Get("from")) + to, e2 := time.Parse(time.RFC3339, q.Get("to")) + if len(q) != 2 || len(q["from"]) != 1 || len(q["to"]) != 1 || e1 != nil || e2 != nil { + writeError(w, 400, "invalid_range", "时间范围无效") + return + } + value, err := store.ServiceStatistics(r.Context(), id, from, to) + if err != nil { + writeError(w, 400, "statistics_error", err.Error()) + return + } + writeJSON(w, 200, value) +} diff --git a/core/internal/controlapi/services.go b/core/internal/controlapi/services.go index ca12aab..a2709e4 100644 --- a/core/internal/controlapi/services.go +++ b/core/internal/controlapi/services.go @@ -113,6 +113,8 @@ func (handler *Handler) serviceItem(writer http.ResponseWriter, request *http.Re } if len(parts) == 2 { switch parts[1] { + case "statistics": + handler.serviceStatistics(writer, request, id) case "authorization": handler.serviceAuthorization(writer, request, id) case "logout": diff --git a/core/internal/pricing/model_rates.go b/core/internal/pricing/model_rates.go new file mode 100644 index 0000000..1fe08d4 --- /dev/null +++ b/core/internal/pricing/model_rates.go @@ -0,0 +1,16 @@ +package pricing + +import "fmt" + +// ModelRates are USD per million tokens. The existing rational evaluator +// snapshots these expressions for each attempt, just like catalog prices. +type ModelRates struct { + Input string `json:"input"` + Output string `json:"output"` + CacheRead string `json:"cache_read"` + CacheWrite string `json:"cache_write"` +} + +func (r ModelRates) Price(model string) Price { + return Price{Provider: "custom", Model: model, Name: model, Expression: fmt.Sprintf("p * %s + c * %s + cr * %s + cc * %s", r.Input, r.Output, r.CacheRead, r.CacheWrite)} +} diff --git a/core/internal/pricing/types.go b/core/internal/pricing/types.go index 45acd5b..91f5f51 100644 --- a/core/internal/pricing/types.go +++ b/core/internal/pricing/types.go @@ -54,11 +54,12 @@ type Binding struct { Model string `json:"model"` } type Config struct { - Provider string `json:"provider"` - Bindings map[string]Binding `json:"bindings"` - MonthlyBudgetUSD string `json:"monthly_budget_usd"` - BillingDay int `json:"billing_day"` - TimeZone string `json:"time_zone"` + Overrides map[string]ModelRates `json:"overrides,omitempty"` + Provider string `json:"provider"` + Bindings map[string]Binding `json:"bindings"` + MonthlyBudgetUSD string `json:"monthly_budget_usd"` + BillingDay int `json:"billing_day"` + TimeZone string `json:"time_zone"` } func DefaultConfig(kind contract.ServiceKind) Config { @@ -81,6 +82,19 @@ func ParseUSD(s string) (*big.Rat, error) { return v, nil } func (c Config) Validate() error { + if len(c.Overrides) > 1000 { + return fmt.Errorf("too many model prices") + } + for model, rates := range c.Overrides { + if model == "" || len(model) > 256 { + return fmt.Errorf("invalid price model") + } + for _, value := range []string{rates.Input, rates.Output, rates.CacheRead, rates.CacheWrite} { + if _, err := ParseUSD(value); err != nil { + return fmt.Errorf("invalid model price: %s", model) + } + } + } if c.Provider != "" && Providers[c.Provider] == "" { return fmt.Errorf("select a canonical official provider") } @@ -109,6 +123,9 @@ func (c Config) Validate() error { return nil } func (c Config) Resolve(model string, prices []Price) (Price, bool) { + if rates, ok := c.Overrides[model]; ok { + return rates.Price(model), true + } b, ok := c.Bindings[model] if !ok { b = Binding{Provider: c.Provider, Model: model} diff --git a/core/internal/storage/service_statistics.go b/core/internal/storage/service_statistics.go new file mode 100644 index 0000000..7ed5894 --- /dev/null +++ b/core/internal/storage/service_statistics.go @@ -0,0 +1,34 @@ +package storage + +import "time" + +type StatisticsTotals struct { + Records int64 `json:"records"` + Priced int64 `json:"priced"` + Unpriced int64 `json:"unpriced"` + AmountUSD string `json:"amount_usd"` + PerMillionUSD *string `json:"per_million_usd"` + CostSamples int64 `json:"cost_samples"` + InputTokens int64 `json:"input_tokens"` + OutputTokens int64 `json:"output_tokens"` + CacheSamples int64 `json:"cache_samples"` + CacheHits int64 `json:"cache_hits"` + CacheReadTokens int64 `json:"cache_read_tokens"` + CacheInputTokens int64 `json:"cache_input_tokens"` + FirstTokenMS *float64 `json:"first_token_ms"` + FirstTokenSamples int64 `json:"first_token_samples"` + DurationMS *float64 `json:"duration_ms"` + DurationSamples int64 `json:"duration_samples"` + TPS *float64 `json:"tps"` + TPSSamples int64 `json:"tps_samples"` +} +type StatisticsModel struct { + Model string `json:"model"` + StatisticsTotals +} +type ServiceStatistics struct { + From time.Time `json:"from"` + To time.Time `json:"to"` + StatisticsTotals + Models []StatisticsModel `json:"models"` +} diff --git a/core/internal/storage/sqlite/pricing.go b/core/internal/storage/sqlite/pricing.go index d29ad26..34e3b4e 100644 --- a/core/internal/storage/sqlite/pricing.go +++ b/core/internal/storage/sqlite/pricing.go @@ -208,6 +208,14 @@ WHERE root_id = ? AND attempt = ? AND local_access_token_id IS NULL`, if len(matches) == 1 { priceJSON = matches[0] } + if rates, ok := c.Overrides[model]; ok { + raw, e := json.Marshal(rates.Price(model)) + if e != nil { + return e + } + priceJSON = string(raw) + version = "channel" + } } terminal = 0 amount = "0.000000000" diff --git a/core/internal/storage/sqlite/service_statistics.go b/core/internal/storage/sqlite/service_statistics.go new file mode 100644 index 0000000..56d14ca --- /dev/null +++ b/core/internal/storage/sqlite/service_statistics.go @@ -0,0 +1,128 @@ +package sqlite + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "math/big" + "sort" + "time" + + "github.com/QuantumNous/astrlink/core/contract" + "github.com/QuantumNous/astrlink/core/internal/storage" +) + +type statisticsAccumulator struct { + storage.StatisticsTotals + amount, eligibleAmount big.Rat + first, duration, tps float64 +} + +func (a *statisticsAccumulator) add(raw, reason, amount string, first, duration sql.NullInt64) { + a.Records++ + var u *contract.Usage + valid := json.Unmarshal([]byte(raw), &u) == nil && u != nil && u.Validate() == nil && !u.BillingIncomplete + cost, ok := new(big.Rat).SetString(amount) + if reason == "priced" && ok { + a.Priced++ + a.amount.Add(&a.amount, cost) + if valid && u.InputTokens+u.OutputTokens > 0 { + a.CostSamples++ + a.InputTokens += int64(u.InputTokens) + a.OutputTokens += int64(u.OutputTokens) + a.eligibleAmount.Add(&a.eligibleAmount, cost) + } + } else { + a.Unpriced++ + } + if valid && u.CacheReadTokens != nil && *u.CacheReadTokens <= u.InputTokens { + a.CacheSamples++ + a.CacheReadTokens += int64(*u.CacheReadTokens) + a.CacheInputTokens += int64(u.InputTokens) + if *u.CacheReadTokens > 0 { + a.CacheHits++ + } + } + if first.Valid && first.Int64 >= 0 { + a.FirstTokenSamples++ + a.first += float64(first.Int64) + } + if duration.Valid && duration.Int64 > 0 { + a.DurationSamples++ + a.duration += float64(duration.Int64) + if valid { + a.TPSSamples++ + a.tps += float64(u.OutputTokens) * 1000 / float64(duration.Int64) + } + } +} +func (a *statisticsAccumulator) finish() storage.StatisticsTotals { + a.AmountUSD = a.amount.FloatString(9) + if tokens := a.InputTokens + a.OutputTokens; tokens > 0 { + v := new(big.Rat).Mul(&a.eligibleAmount, big.NewRat(1000000, tokens)).FloatString(9) + a.PerMillionUSD = &v + } + mean := func(total float64, n int64) *float64 { + if n == 0 { + return nil + } + v := total / float64(n) + return &v + } + a.FirstTokenMS = mean(a.first, a.FirstTokenSamples) + a.DurationMS = mean(a.duration, a.DurationSamples) + a.TPS = mean(a.tps, a.TPSSamples) + return a.StatisticsTotals +} +func (s *Store) ServiceStatistics(ctx context.Context, id contract.ServiceID, from, to time.Time) (storage.ServiceStatistics, error) { + result := storage.ServiceStatistics{From: from, To: to, Models: []storage.StatisticsModel{}} + if from.IsZero() || !to.After(from) || to.Sub(from) > 31*24*time.Hour || from.Nanosecond() != 0 || to.Nanosecond() != 0 { + return result, fmt.Errorf("请选择不超过 31 天的时间范围") + } + if _, err := s.GetService(ctx, id); err != nil { + return result, err + } + // The ledger uniquely owns each actual upstream attempt. Root records are + // updated on retry, while children retain earlier attempts; never sum both. + rows, err := s.db.QueryContext(ctx, `WITH metadata AS ( + SELECT COALESCE(parent_request_id,id) root_id,attempt_index,requested_model,usage_json,first_token_ms,latency_ms, + ROW_NUMBER() OVER(PARTITION BY COALESCE(parent_request_id,id),attempt_index ORDER BY parent_request_id IS NULL,id) n + FROM request_records WHERE service_id=? AND started_at>=? AND started_at=1 + AND status IN ('succeeded','failed','cancelled') AND input_protocol NOT IN ('openai.models','google.models') + ) + SELECT b.model,COALESCE(b.usage_json,'null'),b.reason,b.amount_usd,m.first_token_ms,m.latency_ms + FROM billing_ledger b LEFT JOIN metadata m ON m.root_id=b.root_id AND m.attempt_index=b.attempt AND m.n=1 + WHERE b.service_id=? AND b.started_at>=? AND b.started_at31 day range") + } +} + +func TestServiceStatisticsRetryUsesActualChannelOnce(t *testing.T) { + ctx := context.Background() + s := openTestStore(t, filepath.Join(t.TempDir(), "retry.db")) + defer s.Close() + a, b := pathTestService("service_stats_a"), pathTestService("service_stats_b") + for _, service := range []contract.Service{a, b} { + if _, err := s.CreateService(ctx, service, storage.CredentialMutation{}); err != nil { + t.Fatal(err) + } + } + start := time.Now().UTC().Truncate(time.Second) + model := "x" + r := contract.RequestRecord{ID: "request_stats_root", AttemptIndex: 1, ServiceID: &a.ID, RequestedModel: &model, StartedAt: start, Status: contract.RequestStatusFailed, InputProtocol: contract.ProtocolOpenAIChat, Audit: contract.NotCapturedAuditSummary()} + if err := s.UpsertRequestRecord(ctx, r); err != nil { + t.Fatal(err) + } + child := r + child.ID = "request_stats_child" + child.ParentRequestID = &r.ID + if err := s.InsertRequestRecord(ctx, child); err != nil { + t.Fatal(err) + } + r.AttemptIndex = 2 + r.ServiceID = &b.ID + r.Status = contract.RequestStatusSucceeded + if err := s.UpsertRequestRecord(ctx, r); err != nil { + t.Fatal(err) + } + for _, id := range []contract.ServiceID{a.ID, b.ID} { + v, err := s.ServiceStatistics(ctx, id, start, start.Add(time.Minute)) + if err != nil || v.Records != 1 { + t.Fatalf("channel=%s stats=%+v err=%v", id, v, err) + } + } +} From a13a18b310eeb8e88169caa678be6ffd3ac19b5b Mon Sep 17 00:00:00 2001 From: zuiho <2324465096@qq.com> Date: Sat, 26 Sep 2026 11:52:13 +0800 Subject: [PATCH 2/2] fix: address channel statistics and pricing review feedback --- apps/desktop/src-tauri/src/sidecar.rs | 22 ++++++- .../internal/controlapi/service_statistics.go | 12 +++- .../controlapi/service_statistics_test.go | 38 ++++++++++++ core/internal/storage/sqlite/pricing.go | 8 ++- core/internal/storage/sqlite/pricing_test.go | 58 +++++++++++++++++++ .../storage/sqlite/service_statistics.go | 2 +- .../storage/sqlite/service_statistics_test.go | 5 +- 7 files changed, 135 insertions(+), 10 deletions(-) create mode 100644 core/internal/controlapi/service_statistics_test.go diff --git a/apps/desktop/src-tauri/src/sidecar.rs b/apps/desktop/src-tauri/src/sidecar.rs index 4a7a7e2..a5a8aed 100644 --- a/apps/desktop/src-tauri/src/sidecar.rs +++ b/apps/desktop/src-tauri/src/sidecar.rs @@ -2540,7 +2540,10 @@ fn control_request_timeout(method: &Method, path: &str) -> Duration { if (*method == Method::GET || *method == Method::POST) && is_subscription_usage_path(path) { return SUBSCRIPTION_USAGE_TIMEOUT; } - if *method == Method::GET && is_request_record_list_path(path) { + if *method == Method::GET + && (is_request_record_list_path(path) + || (path.starts_with(&format!("{SERVICES_PATH}/")) && path.ends_with("/statistics"))) + { return REQUEST_LIST_TIMEOUT; } REQUEST_TIMEOUT @@ -5452,6 +5455,23 @@ mod tests { assert!(parse_authorization_session_value(&device).is_err()); } + #[test] + fn service_statistics_allow_aggregation_time_with_query_parameters() { + let path = "/control/v1/services/service_test/statistics?from=2026-09-01T00%3A00%3A00Z&to=2026-09-25T00%3A00%3A00Z"; + assert_eq!( + control_request_timeout(&Method::GET, path), + REQUEST_LIST_TIMEOUT + ); + assert_eq!( + control_request_timeout(&Method::POST, path), + REQUEST_TIMEOUT + ); + assert_eq!( + control_request_timeout(&Method::GET, "/control/v1/services/service_test"), + REQUEST_TIMEOUT + ); + } + #[test] fn provider_tests_allow_the_core_deadline_to_finish() { assert_eq!( diff --git a/core/internal/controlapi/service_statistics.go b/core/internal/controlapi/service_statistics.go index 56f5e0f..79411d2 100644 --- a/core/internal/controlapi/service_statistics.go +++ b/core/internal/controlapi/service_statistics.go @@ -2,10 +2,12 @@ package controlapi import ( "context" - "github.com/QuantumNous/astrlink/core/contract" - "github.com/QuantumNous/astrlink/core/internal/storage" + "errors" "net/http" "time" + + "github.com/QuantumNous/astrlink/core/contract" + "github.com/QuantumNous/astrlink/core/internal/storage" ) type serviceStatisticsStore interface { @@ -31,7 +33,11 @@ func (h *Handler) serviceStatistics(w http.ResponseWriter, r *http.Request, id c } value, err := store.ServiceStatistics(r.Context(), id, from, to) if err != nil { - writeError(w, 400, "statistics_error", err.Error()) + if errors.Is(err, storage.ErrInvalidArgument) { + writeError(w, http.StatusBadRequest, "invalid_range", "请选择不超过 31 天的时间范围") + } else { + h.writeStoreError(w, err) + } return } writeJSON(w, 200, value) diff --git a/core/internal/controlapi/service_statistics_test.go b/core/internal/controlapi/service_statistics_test.go new file mode 100644 index 0000000..8a797b2 --- /dev/null +++ b/core/internal/controlapi/service_statistics_test.go @@ -0,0 +1,38 @@ +package controlapi + +import ( + "net/http" + "strings" + "testing" +) + +func TestServiceStatisticsErrorResponses(t *testing.T) { + store, handler := newServiceHandler(t, "service_statistics") + service := createServiceForTest(t, handler, `{"name":"statistics","kind":"newapi","http":{"base_url":"https://gateway.example/v1","auth":{"scheme":"none"}},"capabilities":[{"protocol":"openai.chat","mode":"native","streaming":true}]}`) + path := ServicesPath + "/" + string(service.ID) + "/statistics" + validRange := "?from=2026-09-01T00:00:00Z&to=2026-09-25T00:00:00Z" + for _, tc := range []struct { + name, path, code string + status int + }{ + {"valid", path + validRange, "", http.StatusOK}, + {"missing service", ServicesPath + "/service_missing/statistics" + validRange, "not_found", http.StatusNotFound}, + {"malformed", path + "?from=bad&to=bad", "invalid_range", http.StatusBadRequest}, + {"too long", path + "?from=2026-08-01T00:00:00Z&to=2026-09-25T00:00:00Z", "invalid_range", http.StatusBadRequest}, + {"reversed", path + "?from=2026-09-25T00:00:00Z&to=2026-09-01T00:00:00Z", "invalid_range", http.StatusBadRequest}, + } { + t.Run(tc.name, func(t *testing.T) { + response := serviceRequestForTest(t, handler, http.MethodGet, tc.path, "", "", "") + if response.Code != tc.status || (tc.code != "" && !strings.Contains(response.Body.String(), `"code":"`+tc.code+`"`)) { + t.Fatalf("status=%d body=%s", response.Code, response.Body.String()) + } + }) + } + if err := store.Close(); err != nil { + t.Fatal(err) + } + response := serviceRequestForTest(t, handler, http.MethodGet, path+validRange, "", "", "") + if response.Code != http.StatusInternalServerError || !strings.Contains(response.Body.String(), `"code":"storage_unavailable"`) || strings.Contains(response.Body.String(), "database is closed") { + t.Fatalf("storage failure status=%d body=%s", response.Code, response.Body.String()) + } +} diff --git a/core/internal/storage/sqlite/pricing.go b/core/internal/storage/sqlite/pricing.go index 34e3b4e..e53f252 100644 --- a/core/internal/storage/sqlite/pricing.go +++ b/core/internal/storage/sqlite/pricing.go @@ -252,12 +252,12 @@ WHERE root_id = ? AND attempt = ? AND local_access_token_id IS NULL`, return err } -// PriceUnpriced fills missing prices from new catalogs and retries audio +// PriceUnpriced fills missing prices from new catalogs or channel overrides and retries audio // breakdown failures with the original price snapshot after evaluator fixes. // It does not import legacy logs or rewrite amounts that were already priced. func (s *Store) PriceUnpriced(ctx context.Context) (int, error) { catalog, err := s.PricingCatalog(ctx) - if err != nil || catalog.Version == "" { + if err != nil { return 0, err } processed := 0 @@ -266,7 +266,9 @@ func (s *Store) PriceUnpriced(ctx context.Context) (int, error) { for { rows, err := s.db.QueryContext(ctx, `SELECT b.root_id,b.attempt,b.service_id,b.started_at,b.model,b.usage_json,COALESCE(b.price_json,''),b.reason FROM billing_ledger b JOIN services s ON s.id=b.service_id -WHERE b.terminal=1 AND ((b.reason='missing_price' AND b.price_version<>?) +WHERE b.terminal=1 AND ((b.reason='missing_price' AND (b.price_version<>? +OR EXISTS (SELECT 1 FROM pricing_configs pc, json_each(pc.document_json, '$.overrides') o +WHERE pc.service_id=b.service_id AND o.key=b.model))) OR b.reason IN ('missing_audio_usage','missing_audio_cache_partition')) AND (b.root_id>? OR (b.root_id=? AND b.attempt>?)) ORDER BY b.root_id,b.attempt LIMIT 100`, catalog.Version, lastRoot, lastRoot, lastAttempt) diff --git a/core/internal/storage/sqlite/pricing_test.go b/core/internal/storage/sqlite/pricing_test.go index 0c80fca..84f6cba 100644 --- a/core/internal/storage/sqlite/pricing_test.go +++ b/core/internal/storage/sqlite/pricing_test.go @@ -218,6 +218,64 @@ func TestMissingPricesFillAutomaticallyOncePerCatalog(t *testing.T) { } } +func TestChannelOverridesFillMissingPricesWithoutCatalogChanges(t *testing.T) { + for _, catalogVersion := range []string{"", "existing_v1"} { + t.Run("catalog="+catalogVersion, func(t *testing.T) { + ctx := context.Background() + s := openTestStore(t, filepath.Join(t.TempDir(), "billing.db")) + defer s.Close() + start := time.Now().UTC().Truncate(time.Second).Add(-time.Minute) + if catalogVersion != "" { + catalog := pricing.Catalog{Version: catalogVersion, ActivatedAt: start.Add(-time.Hour), Prices: []pricing.Price{{Provider: "openai", Model: "known", Expression: `tier("standard",p * 1)`}}} + if err := s.SavePricingCatalog(ctx, catalog); err != nil { + t.Fatal(err) + } + } + service, other := pathTestService("service_override"), pathTestService("service_other") + model := `custom.model/"v1"` + for _, channel := range []contract.Service{service, other} { + if _, err := s.CreateService(ctx, channel, storage.CredentialMutation{}); err != nil { + t.Fatal(err) + } + r := contract.RequestRecord{ID: contract.RequestID("request_" + channel.ID), AttemptIndex: 1, ServiceID: &channel.ID, RequestedModel: &model, StartedAt: start, Status: contract.RequestStatusSucceeded, InputProtocol: contract.ProtocolOpenAIResponses, Audit: contract.NotCapturedAuditSummary(), Usage: &contract.Usage{InputTokens: 1000000, TotalTokens: 1000000}} + if err := s.InsertRequestRecord(ctx, r); err != nil { + t.Fatal(err) + } + // Automatic valuation must work even after request-detail retention expires. + if err := s.DeleteRequestRecord(ctx, r.ID); err != nil { + t.Fatal(err) + } + } + if n, err := s.PriceUnpriced(ctx); err != nil || n != 0 { + t.Fatalf("before override fill=%d err=%v", n, err) + } + config := pricing.DefaultConfig(service.Kind) + config.Overrides = map[string]pricing.ModelRates{model: {Input: "2", Output: "4", CacheRead: "2", CacheWrite: "2"}} + if err := s.SavePricingConfig(ctx, service.ID, config); err != nil { + t.Fatal(err) + } + if n, err := s.PriceUnpriced(ctx); err != nil || n != 1 { + t.Fatalf("override fill=%d err=%v", n, err) + } + config.Overrides[model] = pricing.ModelRates{Input: "9", Output: "9", CacheRead: "9", CacheWrite: "9"} + if err := s.SavePricingConfig(ctx, service.ID, config); err != nil { + t.Fatal(err) + } + if n, err := s.PriceUnpriced(ctx); err != nil || n != 0 { + t.Fatalf("repeated fill=%d err=%v", n, err) + } + summary, err := s.BillingSummary(ctx, service.ID, "", start, start.Add(time.Hour), pricing.BillingSummaryOptions{}) + if err != nil || summary.Priced != 1 || summary.Unpriced != 0 || summary.Revalued != 1 || summary.AmountUSD != "2.000000000" { + t.Fatalf("override summary=%+v err=%v", summary, err) + } + summary, err = s.BillingSummary(ctx, other.ID, "", start, start.Add(time.Hour), pricing.BillingSummaryOptions{}) + if err != nil || summary.Priced != 0 || summary.Unpriced != 1 { + t.Fatalf("other channel summary=%+v err=%v", summary, err) + } + }) + } +} + func TestInterruptedBillingBecomesUnpricedAndCannotBeBackfilledAsComplete(t *testing.T) { ctx := context.Background() s := openTestStore(t, filepath.Join(t.TempDir(), "billing.db")) diff --git a/core/internal/storage/sqlite/service_statistics.go b/core/internal/storage/sqlite/service_statistics.go index 56d14ca..40d9750 100644 --- a/core/internal/storage/sqlite/service_statistics.go +++ b/core/internal/storage/sqlite/service_statistics.go @@ -78,7 +78,7 @@ func (a *statisticsAccumulator) finish() storage.StatisticsTotals { func (s *Store) ServiceStatistics(ctx context.Context, id contract.ServiceID, from, to time.Time) (storage.ServiceStatistics, error) { result := storage.ServiceStatistics{From: from, To: to, Models: []storage.StatisticsModel{}} if from.IsZero() || !to.After(from) || to.Sub(from) > 31*24*time.Hour || from.Nanosecond() != 0 || to.Nanosecond() != 0 { - return result, fmt.Errorf("请选择不超过 31 天的时间范围") + return result, fmt.Errorf("%w: 请选择不超过 31 天的时间范围", storage.ErrInvalidArgument) } if _, err := s.GetService(ctx, id); err != nil { return result, err diff --git a/core/internal/storage/sqlite/service_statistics_test.go b/core/internal/storage/sqlite/service_statistics_test.go index 34dc00e..297e56a 100644 --- a/core/internal/storage/sqlite/service_statistics_test.go +++ b/core/internal/storage/sqlite/service_statistics_test.go @@ -2,6 +2,7 @@ package sqlite import ( "context" + "errors" "github.com/QuantumNous/astrlink/core/contract" "github.com/QuantumNous/astrlink/core/internal/pricing" "github.com/QuantumNous/astrlink/core/internal/storage" @@ -60,8 +61,8 @@ func TestServiceStatisticsRatesCacheMissingAndPinnedHistory(t *testing.T) { t.Fatal(err) } assert() - if _, err := s.ServiceStatistics(ctx, service.ID, start, start.Add(32*24*time.Hour)); err == nil { - t.Fatal("accepted >31 day range") + if _, err := s.ServiceStatistics(ctx, service.ID, start, start.Add(32*24*time.Hour)); !errors.Is(err, storage.ErrInvalidArgument) { + t.Fatalf("invalid range error=%v", err) } }