From 606102550618749a70a4e5a724d326ed09447da2 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Mon, 6 Jul 2026 18:31:06 +0100 Subject: [PATCH] feat: group Fusion usage by request --- CHANGELOG.md | 1 + admin/src/demo-data.ts | 27 ++++++ admin/src/screens/users-usage.tsx | 59 +++++++++---- admin/src/styles/usage.css | 89 ++++++++++++++++++++ admin/src/usage-analytics.ts | 45 +++++++++- admin/test/usage-analytics.test.mjs | 30 ++++++- docs/fusion-router.md | 8 +- scripts/smoke-local-worker.mjs | 60 +++++++++++-- shared/contracts.ts | 5 ++ worker/proxy.ts | 126 +++++++++++++++++++++------- worker/types.ts | 5 ++ 11 files changed, 400 insertions(+), 55 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 07d8d18..bd83d0c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ## Unreleased +- Group Fusion adviser and synthesizer usage into one expandable request with shared lineage, aggregate cost, end-to-end latency, and preserved billable-call detail. - Add an on-demand `clawrouter/fusion` chat model with parallel local or hosted advisers, a policy-native final synthesizer, OpenAI-compatible Ollama/LM Studio routing, bounded fail-open orchestration, and web-console configuration. - Add accessible 30-day request and provider analytics to Dashboard and Usage while preserving rolling usage totals across mixed-version deployments. - Reject non-object OpenAI, manifest, and native proxy request bodies before routing or budget reservation. diff --git a/admin/src/demo-data.ts b/admin/src/demo-data.ts index 7eb117f..4793409 100644 --- a/admin/src/demo-data.ts +++ b/admin/src/demo-data.ts @@ -13,6 +13,33 @@ export function demoUsageSnapshot(): UsageSnapshot { const now = Date.now(); const summary = { requestCount: 1284, successCount: 1247, errorCount: 37, inputTokens: 1_482_402, outputTokens: 382_151, totalTokens: 1_864_553, actualCostMicros: 8_432_100 }; const events: UsageAuditEvent[] = [ + demoUsageEvent("usage_fusion_final", now - 18_000, "admin@example.com", "maintainer_models", "openai", "llm.chat", "openai/gpt-4.1-mini", 200, 690, 1900, "success", 1512, { + request_id: "req_fusion_7", + compound_request_id: "req_fusion_7", + compound_request_stage: "fusion_synthesizer", + compound_request_index: null, + compound_request_size: 3, + compound_request_started_at_ms: now - 19_100, + cost_basis: "manifest_pricing", + }), + demoUsageEvent("usage_fusion_adviser_2", now - 18_420, "admin@example.com", "maintainer_models", "openai", "llm.chat", "openai/gpt-4.1-mini", 503, 244, 0, "provider_error", 0, { + request_id: "fusion-adviser-2-demo", + compound_request_id: "req_fusion_7", + compound_request_stage: "fusion_adviser", + compound_request_index: 2, + compound_request_size: 3, + compound_request_started_at_ms: now - 19_100, + cost_basis: "manifest_pricing", + }), + demoUsageEvent("usage_fusion_adviser_1", now - 18_510, "admin@example.com", "maintainer_models", "local-openai", "llm.chat", "local/qwen3:8b", 200, 410, 0, "success", 846, { + request_id: "fusion-adviser-1-demo", + compound_request_id: "req_fusion_7", + compound_request_stage: "fusion_adviser", + compound_request_index: 1, + compound_request_size: 3, + compound_request_started_at_ms: now - 19_100, + cost_basis: "manifest_pricing", + }), demoUsageEvent("usage_6", now - 26_000, "admin@example.com", "maintainer_models", "openai", "llm.responses", "gpt-5.4", 200, 842, 1000, "success", 1814, { agent_id: "codex/reviewer", parent_agent_id: "codex/orchestrator", diff --git a/admin/src/screens/users-usage.tsx b/admin/src/screens/users-usage.tsx index ba54d04..811664b 100644 --- a/admin/src/screens/users-usage.tsx +++ b/admin/src/screens/users-usage.tsx @@ -65,6 +65,7 @@ export function UsageScreen({ keys, credentials, services, overview, tenants, us const [retainedContent, setRetainedContent] = useState(null); const [contentError, setContentError] = useState(""); const [contentLoading, setContentLoading] = useState(false); + const [expandedRequestId, setExpandedRequestId] = useState(null); const contentFeedbackRef = useRef(null); useEffect(() => { if (contentLoading || contentError || retainedContent) contentFeedbackRef.current?.scrollIntoView({ block: "nearest" }); @@ -92,6 +93,7 @@ export function UsageScreen({ keys, credentials, services, overview, tenants, us const untrackedRows = rows.filter((row) => row.enabled && row.budget.ledger === "untracked"); const exhaustedRows = rows.filter((row) => row.enabled && row.budget.configured && row.budget.remainingMicros !== undefined && row.budget.remainingMicros !== null && row.budget.remainingMicros <= 0); const ledgerFailureRows = rows.filter((row) => row.enabled && (row.budget.ledger === "unavailable" || row.budget.ledger === "invalid_policy")); + const requestGroups = usageEventGroups(usage.events); return (
@@ -140,26 +142,32 @@ export function UsageScreen({ keys, credentials, services, overview, tenants, us
: null}
-
Recent requests{usage.events.length} most recent audit events
{usageLoaded ? usage.ledger : "unavailable"}
+
Recent requests{requestGroups.length} requests · {usage.events.length} billable calls
{usageLoaded ? usage.ledger : "unavailable"}
{ - const service = serviceByProvider.get(event.provider); + rows={requestGroups.map((group) => { + const event = group.primary; + const service = group.compound ? undefined : serviceByProvider.get(event.provider); const verifiedIdentity = event.principal_id ?? event.credential_id ?? event.policy_id ?? event.tenant_id; const agentIdentity = event.agent_id ? `agent ${event.agent_id}` : event.auth_type ?? "authenticated"; const agentContext = [event.parent_agent_id && `parent ${event.parent_agent_id}`, event.client && `client ${event.client}`, event.project_id && `project ${event.project_id}`, event.session_id && `session ${event.session_id}`].filter(Boolean).join(" · "); return { - id: event.id, + id: group.id, cells: [ - {formatTimestamp(event.occurred_at_ms)}, + {formatTimestamp(group.occurredAtMs)}, {verifiedIdentity}{agentIdentity}{agentContext ? {agentContext} : null}, - , - {event.capability ?? event.type}{[event.model, event.cost_basis].filter(Boolean).join(" · ") || event.request_id || "request"}, - , - event.content_retained ? : not stored, - formatMicros(event.actual_cost_micros), + , + group.compound + ? + : {event.capability ?? event.type}{[event.model, event.cost_basis].filter(Boolean).join(" · ") || event.request_id || "request"}, + , + group.compound + ? {group.events.filter((item) => item.content_retained).length} stored + : event.content_retained ? : not stored, + `${group.complete ? "" : "≥"}${formatMicros(group.actualCostMicros)}`, ], + detail: group.compound && expandedRequestId === group.id ? : undefined, }; })} /> @@ -174,6 +182,25 @@ export function UsageScreen({ keys, credentials, services, overview, tenants, us ); } +function CompoundRequestCalls({ group, onInspect }: { group: UsageEventGroup; onInspect: (event: UsageAuditEvent) => Promise }) { + return ( +
+
{group.complete ? "Billable call detail" : "Partial billable call detail"}{group.durationMs != null ? `${group.complete ? "End-to-end" : "Visible span"} ${formatDuration(group.durationMs)}` : "Latency unavailable"} · {group.complete ? formatMicros(group.actualCostMicros) : `at least ${formatMicros(group.actualCostMicros)}`}
+
+ {group.events.map((event) =>
+ {compoundStage(event)}{event.provider} · {event.model ?? event.capability ?? "request"} + {event.duration_ms != null ? formatDuration(event.duration_ms) : "—"} · {formatMicros(event.actual_cost_micros)}{event.content_retained ? : null} +
)} +
+ {!group.complete ?

This recent-event window contains {group.events.length} of {group.expectedCallCount} calls. Totals exclude older calls outside the window.

: null} +
+ ); +} + +function compoundStage(event: UsageAuditEvent) { + return event.compound_request_stage === "fusion_synthesizer" ? "Synthesizer" : `Adviser ${event.compound_request_index ?? ""}`.trim(); +} + export function Metric({ label, value, meta }: { label: string; value: string; meta: string }) { return
{label}{value}{meta}
; } @@ -211,16 +238,19 @@ export function UsageHealth({ row }: { row: AdminUsageRow }) { return ; } -export function EntityTable({ columns, columnTemplate, rows }: { columns: string[]; columnTemplate?: string; rows: Array<{ id: string; active?: boolean; onClick?: () => void; cells: React.ReactNode[] }> }) { +export function EntityTable({ columns, columnTemplate, rows }: { columns: string[]; columnTemplate?: string; rows: Array<{ id: string; active?: boolean; onClick?: () => void; cells: React.ReactNode[]; detail?: React.ReactNode }> }) { return (
{columns.map((column) => {column})}
{rows.map((row) => { const content = row.cells.map((cell, index) => {cell}); const className = `tableRow${row.active ? " selected" : ""}${row.onClick ? " interactive" : ""}`; - return row.onClick - ? - :
{content}
; + return + {row.onClick + ? + :
{content}
} + {row.detail ?
{row.detail}
: null} + ; })}
); @@ -230,5 +260,6 @@ import { Activity, CalendarDays, KeyRound, Plus, Search, ServerCog, ShieldCheck, import { bindingKey, effectiveAccess, errorMessage, policyUsageFallback, tenantSummaryFallback } from "../domain"; import { EntityName, InlineError, InlineNote, InspectorHeader, PanelTitle, Status, kindLabel } from "../components"; import { ProviderUsageChart, TrafficAreaChart } from "../analytics-charts"; +import { usageEventGroups, type UsageEventGroup } from "../usage-analytics"; import { budgetPercent, effectiveProviderCount, formatBudget, formatCount, formatDuration, formatMicros, formatTimestamp, providerBrandIcon, readyCount, request, usageEventTone, usagePolicyId } from "../ui-helpers"; import type { AccessForm,AccessPolicy,AccessRole,AccessTab,AccessUser,AdminOverview,AdminTenantSummary,AdminUsageRow,AssignmentRule,AssignmentRuleForm,BindingForm,BrandIcon,BudgetStatus,ContentRetention,CredentialForm,EntitlementsResponse,IconComponent,OutcomeTone,PlaygroundForm,PlaygroundHttpResponse,PlaygroundTurn,PolicyBinding,PolicyForm,ProviderAccess,ProviderConnection,ProviderReadiness,ProviderResponse,ProviderRow,ProviderUsageSummary,ProxyCredential,RefreshOptions,RetainedRequestContent,RouteCatalog,ServiceItem,ServiceOutcome,SessionResponse,UpstreamGrant,UpstreamGrantForm,UsageAuditEvent,UsageSnapshot,UsageSummary,View } from "../ui-types"; diff --git a/admin/src/styles/usage.css b/admin/src/styles/usage.css index ad54f15..4fc76ec 100644 --- a/admin/src/styles/usage.css +++ b/admin/src/styles/usage.css @@ -60,6 +60,95 @@ white-space: nowrap; } +.compoundRequestToggle { + display: grid; + width: 100%; + gap: 2px; + padding: 0; + border: 0; + background: transparent; + text-align: left; + cursor: pointer; +} + +.compoundRequestToggle strong { + color: var(--ink); + font-size: 10.5px; + font-weight: 600; +} + +.compoundRequestToggle small, +.compoundRequestCalls small { + color: var(--muted); + font-family: var(--font-mono); + font-size: 9.5px; +} + +.usageTablePanel .tableRowDetail { + border-bottom: 1px solid var(--line); + background: var(--surface-raised); +} + +.compoundRequest { + padding: 12px 16px 14px; +} + +.compoundRequestHeader { + display: flex; + align-items: baseline; + justify-content: space-between; + gap: 16px; + margin-bottom: 8px; +} + +.compoundRequestHeader strong { + color: var(--ink); + font-size: 11px; +} + +.compoundRequestHeader span, +.compoundRequestWarning { + color: var(--muted); + font-family: var(--font-mono); + font-size: 9.5px; +} + +.compoundRequestCalls { + display: grid; + width: 100%; + border: 1px solid var(--line); + background: var(--surface-raised); + box-shadow: var(--shadow-raised); +} + +.compoundRequestWarning { + margin: 8px 0 0; +} + +.compoundRequestCalls > div { + display: flex; + align-items: center; + justify-content: space-between; + gap: 12px; + padding: 8px 10px; + border-bottom: 1px solid var(--line); +} + +.compoundRequestCalls > div:last-child { + border-bottom: 0; +} + +.compoundRequestCalls > div > span { + display: grid; + gap: 2px; + min-width: 0; +} + +.compoundRequestCalls > div > span:last-child { + justify-items: end; + white-space: nowrap; +} + .budgetUsage { display: grid; gap: 7px; diff --git a/admin/src/usage-analytics.ts b/admin/src/usage-analytics.ts index 737c70b..be82658 100644 --- a/admin/src/usage-analytics.ts +++ b/admin/src/usage-analytics.ts @@ -1,7 +1,50 @@ -import type { UsageDailySummary, UsageSnapshot, UsageSummary } from "./ui-types"; +import type { UsageAuditEvent, UsageDailySummary, UsageSnapshot, UsageSummary } from "./ui-types"; export const usageDayMs = 86_400_000; +export interface UsageEventGroup { + id: string; + compound: boolean; + events: UsageAuditEvent[]; + primary: UsageAuditEvent; + occurredAtMs: number; + durationMs: number | null; + actualCostMicros: number; + successCount: number; + expectedCallCount: number; + complete: boolean; +} + +export function usageEventGroups(events: UsageAuditEvent[]): UsageEventGroup[] { + const grouped = new Map(); + const order: string[] = []; + for (const event of events) { + const key = event.compound_request_id ? `compound:${event.compound_request_id}` : `event:${event.id}`; + if (!grouped.has(key)) order.push(key); + grouped.set(key, [...(grouped.get(key) ?? []), event]); + } + return order.map((key) => summarizeUsageEvents(key, grouped.get(key) ?? [])); +} + +function summarizeUsageEvents(key: string, events: UsageAuditEvent[]): UsageEventGroup { + const primary = events.find((event) => event.compound_request_stage === "fusion_synthesizer") ?? events[0]; + if (!primary) throw new Error("usage event group cannot be empty"); + const startedAt = Math.min(...events.map((event) => event.compound_request_started_at_ms ?? event.occurred_at_ms - Math.max(0, event.duration_ms ?? 0))); + const occurredAtMs = Math.max(...events.map((event) => event.occurred_at_ms)); + return { + id: key, + compound: Boolean(primary.compound_request_id), + events: [...events].sort((left, right) => (left.compound_request_index ?? Number.MAX_SAFE_INTEGER) - (right.compound_request_index ?? Number.MAX_SAFE_INTEGER)), + primary, + occurredAtMs, + durationMs: events.some((event) => event.duration_ms != null) ? occurredAtMs - startedAt : null, + actualCostMicros: events.reduce((total, event) => total + event.actual_cost_micros, 0), + successCount: events.filter((event) => event.status === "success").length, + expectedCallCount: Math.max(...events.map((event) => event.compound_request_size ?? events.length)), + complete: !primary.compound_request_id || events.length >= Math.max(...events.map((event) => event.compound_request_size ?? events.length)), + }; +} + export function usageTimeline(snapshot: UsageSnapshot, days = 30, now = Date.now()): UsageDailySummary[] { const today = Math.floor(now / usageDayMs) * usageDayMs; const firstDay = today - (days - 1) * usageDayMs; diff --git a/admin/test/usage-analytics.test.mjs b/admin/test/usage-analytics.test.mjs index 622249f..d4f2988 100644 --- a/admin/test/usage-analytics.test.mjs +++ b/admin/test/usage-analytics.test.mjs @@ -1,6 +1,6 @@ import assert from "node:assert/strict"; import test from "node:test"; -import { niceChartMaximum, syntheticUsageTimeline, usageDayMs, usageTimeline } from "../src/usage-analytics.ts"; +import { niceChartMaximum, syntheticUsageTimeline, usageDayMs, usageEventGroups, usageTimeline } from "../src/usage-analytics.ts"; const summary = { requestCount: 10, successCount: 9, errorCount: 1, inputTokens: 80, outputTokens: 20, totalTokens: 100, actualCostMicros: 500 }; @@ -48,3 +48,31 @@ test("synthetic usage timelines preserve their scoped summary totals", () => { assert.equal(timeline.reduce((total, point) => total + point.totalTokens, 0), summary.totalTokens); assert.equal(timeline.reduce((total, point) => total + point.actualCostMicros, 0), summary.actualCostMicros); }); + +test("compound Fusion calls collapse into one request with aggregate cost and wall time", () => { + const base = { type: "clawrouter.usage.v1", tenant_id: "tenant", provider: "openai", reserved_cost_micros: 0, status: "success", compound_request_id: "req_fusion", compound_request_size: 3, compound_request_started_at_ms: 900 }; + const groups = usageEventGroups([ + { ...base, id: "synth", occurred_at_ms: 2_000, duration_ms: 400, actual_cost_micros: 30, compound_request_stage: "fusion_synthesizer", compound_request_index: null }, + { ...base, id: "adviser-2", occurred_at_ms: 1_500, duration_ms: 300, actual_cost_micros: 20, compound_request_stage: "fusion_adviser", compound_request_index: 2 }, + { ...base, id: "adviser-1", occurred_at_ms: 1_450, duration_ms: 350, actual_cost_micros: 10, compound_request_stage: "fusion_adviser", compound_request_index: 1 }, + { ...base, id: "ordinary", occurred_at_ms: 900, duration_ms: 50, actual_cost_micros: 5, compound_request_id: null }, + ]); + assert.equal(groups.length, 2); + assert.equal(groups[0].primary.id, "synth"); + assert.deepEqual(groups[0].events.map((event) => event.id), ["adviser-1", "adviser-2", "synth"]); + assert.equal(groups[0].actualCostMicros, 60); + assert.equal(groups[0].durationMs, 1100); + assert.equal(groups[0].complete, true); + assert.equal(groups[1].compound, false); +}); + +test("truncated Fusion groups are explicitly partial", () => { + const base = { type: "clawrouter.usage.v1", tenant_id: "tenant", provider: "openai", reserved_cost_micros: 10, actual_cost_micros: 10, status: "success", compound_request_id: "req_partial", compound_request_size: 4 }; + const [group] = usageEventGroups([ + { ...base, id: "synth", occurred_at_ms: 2_000, compound_request_stage: "fusion_synthesizer" }, + { ...base, id: "adviser", occurred_at_ms: 1_000, compound_request_stage: "fusion_adviser", compound_request_index: 1 }, + ]); + assert.equal(group.complete, false); + assert.equal(group.expectedCallCount, 4); + assert.equal(group.actualCostMicros, 20); +}); diff --git a/docs/fusion-router.md b/docs/fusion-router.md index 20f2a45..d1c5559 100644 --- a/docs/fusion-router.md +++ b/docs/fusion-router.md @@ -71,7 +71,13 @@ Each adviser and the synthesizer is a separate normal ClawRouter request. The caller's policy must grant all providers that should participate. Budgets are reserved and usage is recorded per subrequest, so a two-adviser fusion request can account for three model calls. A failed adviser can still consume upstream -tokens before its failure becomes visible. +tokens before its failure becomes visible. Usage events retain their individual +request ids and also carry a shared `compound_request_id`, stage, and adviser +index. The Usage console groups those calls into one expandable Fusion request +with aggregate spend and end-to-end latency measured from Fusion entry; opening +it preserves the billable subrequest detail. A request rejected during final +route preflight or budget reservation appears as a one-call failed Fusion group +because adviser fan-out never began. ## Ollama and LM Studio diff --git a/scripts/smoke-local-worker.mjs b/scripts/smoke-local-worker.mjs index 0505215..c7494dd 100644 --- a/scripts/smoke-local-worker.mjs +++ b/scripts/smoke-local-worker.mjs @@ -108,7 +108,7 @@ try { const deniedModels = await fetch(`${base}/v1/models`, { headers: { authorization: `Bearer ${proxyKey}` } }); assert.equal(deniedModels.status, 200); assert.equal((await deniedModels.json()).data.some((model) => model.id === "clawrouter/fusion"), false, "fusion stays hidden until its synthesizer route is executable for the caller"); - const budgetBlockedFusion = await fetch(`${base}/v1/chat/completions`, { method: "POST", headers: { authorization: `Bearer ${fusionKey}`, "content-type": "application/json" }, body: JSON.stringify({ model: "clawrouter/fusion", messages: [{ role: "user", content: "solve" }] }) }); + const budgetBlockedFusion = await fetch(`${base}/v1/chat/completions`, { method: "POST", headers: { authorization: `Bearer ${fusionKey}`, "content-type": "application/json", "x-request-id": "fusion-e2e-budget-blocked" }, body: JSON.stringify({ model: "clawrouter/fusion", messages: [{ role: "user", content: "solve" }] }) }); assert.equal(budgetBlockedFusion.status, 402); assert.equal((await budgetBlockedFusion.json()).error.code, "budget_exhausted"); assert.equal(upstreamCalls.length, 0, "synthesizer budget reservation blocks adviser spend before the final answer can be funded"); @@ -122,24 +122,68 @@ try { const fusionModels = await fetch(`${base}/v1/models`, { headers: { authorization: `Bearer ${fusionKey}` } }); assert.equal(fusionModels.status, 200); assert.equal((await fusionModels.json()).data.some((model) => model.id === "clawrouter/fusion"), true); - const fusionResponse = await fetch(`${base}/v1/chat/completions`, { method: "POST", headers: { authorization: `Bearer ${fusionKey}`, "content-type": "application/json" }, body: JSON.stringify({ model: "clawrouter/fusion", messages: [{ role: "user", content: "solve" }] }) }); + const fusionResponse = await fetch(`${base}/v1/chat/completions`, { method: "POST", headers: { authorization: `Bearer ${fusionKey}`, "content-type": "application/json", "x-request-id": "fusion-e2e-retry" }, body: JSON.stringify({ model: "clawrouter/fusion", messages: [{ role: "user", content: "solve" }] }) }); assert.equal(fusionResponse.status, 200, JSON.stringify(await fusionResponse.clone().json())); assert.equal((await fusionResponse.json()).choices[0].message.content, "fused answer"); assert.equal(fusionResponse.headers.get("x-clawrouter-fusion-adviser-count"), "1"); assert.equal(fusionResponse.headers.get("x-clawrouter-fusion-failed-count"), "0"); assert.deepEqual(upstreamCalls.map((call) => call.model), ["adviser", "final"]); + const selectionFailureCallStart = upstreamCalls.length; + const selectionFailureConfig = { ...localFusionConfig, adviserModels: ["azure-openai/deployment"] }; + assert.equal((await fetch(`${base}/v1/admin/fusion`, { method: "PUT", headers: adminHeaders, body: JSON.stringify(selectionFailureConfig) })).status, 200); + const selectionFailureResponse = await fetch(`${base}/v1/chat/completions`, { method: "POST", headers: { authorization: `Bearer ${fusionKey}`, "content-type": "application/json", "x-request-id": "fusion-e2e-selection-failure" }, body: JSON.stringify({ model: "clawrouter/fusion", messages: [{ role: "user", content: "solve without configured adviser" }] }) }); + assert.equal(selectionFailureResponse.status, 200, JSON.stringify(await selectionFailureResponse.clone().json())); + assert.equal(selectionFailureResponse.headers.get("x-clawrouter-fusion-adviser-count"), "0"); + assert.equal(selectionFailureResponse.headers.get("x-clawrouter-fusion-failed-count"), "1"); + assert.deepEqual(upstreamCalls.slice(selectionFailureCallStart).map((call) => call.model), ["final"], "selection failures are audited without attempting the invalid upstream route"); const stalledCallStart = upstreamCalls.length; const stalledFusionConfig = { ...localFusionConfig, adviserModels: ["local/stall"], adviserTimeoutMs: 1_000 }; assert.equal((await fetch(`${base}/v1/admin/fusion`, { method: "PUT", headers: adminHeaders, body: JSON.stringify(stalledFusionConfig) })).status, 200); - const stalledFusionResponse = await fetch(`${base}/v1/chat/completions`, { method: "POST", headers: { authorization: `Bearer ${fusionKey}`, "content-type": "application/json" }, body: JSON.stringify({ model: "clawrouter/fusion", messages: [{ role: "user", content: "solve after timeout" }] }) }); + const stalledFusionResponse = await fetch(`${base}/v1/chat/completions`, { method: "POST", headers: { authorization: `Bearer ${fusionKey}`, "content-type": "application/json", "x-request-id": "fusion-e2e-retry" }, body: JSON.stringify({ model: "clawrouter/fusion", messages: [{ role: "user", content: "solve after timeout" }] }) }); assert.equal(stalledFusionResponse.status, 200, JSON.stringify(await stalledFusionResponse.clone().json())); assert.equal(stalledFusionResponse.headers.get("x-clawrouter-fusion-adviser-count"), "0"); assert.equal(stalledFusionResponse.headers.get("x-clawrouter-fusion-failed-count"), "1"); await waitUntil(() => stalledUpstreamClosed, "timed-out adviser upstream connection did not close"); assert.deepEqual(upstreamCalls.slice(stalledCallStart).map((call) => call.model), ["stall", "final"]); - const fusionUsage = await fetch(`${base}/v1/usage`, { headers: { authorization: `Bearer ${fusionKey}` } }); - assert.equal(fusionUsage.status, 200); - assert.equal((await fusionUsage.json()).budget.spentMicros, 0, "dynamic local models retain zero-price accounting under a budgeted policy"); + let fusionUsageBody; + await waitUntil(async () => { + const fusionUsage = await fetch(`${base}/v1/usage`, { headers: { authorization: `Bearer ${fusionKey}` } }); + assert.equal(fusionUsage.status, 200); + fusionUsageBody = await fusionUsage.json(); + const synthesizers = fusionUsageBody.usage.events.filter((event) => event.request_id === "fusion-e2e-retry" && event.compound_request_stage === "fusion_synthesizer"); + const visibleCompoundIds = new Set(synthesizers.map((event) => event.compound_request_id)); + const blocked = fusionUsageBody.usage.events.find((event) => event.request_id === "fusion-e2e-budget-blocked"); + const selectionSynthesizer = fusionUsageBody.usage.events.find((event) => event.request_id === "fusion-e2e-selection-failure" && event.compound_request_stage === "fusion_synthesizer"); + const selectionEvents = selectionSynthesizer ? fusionUsageBody.usage.events.filter((event) => event.compound_request_id === selectionSynthesizer.compound_request_id) : []; + return visibleCompoundIds.size >= 2 && fusionUsageBody.usage.events.filter((event) => visibleCompoundIds.has(event.compound_request_id)).length >= 4 && blocked?.compound_request_stage === "fusion_synthesizer" && selectionEvents.length >= 2; + }, "fusion usage lineage was not delivered"); + const compoundIds = [...new Set(fusionUsageBody.usage.events.filter((event) => event.request_id === "fusion-e2e-retry").map((event) => event.compound_request_id))]; + assert.equal(compoundIds.length, 2, "reused caller request ids must not merge separate Fusion invocations"); + const lineage = fusionUsageBody.usage.events + .filter((event) => compoundIds.includes(event.compound_request_id)) + .map((event) => ({ request_id: event.request_id, compound_request_id: event.compound_request_id, stage: event.compound_request_stage, index: event.compound_request_index, model: event.model })); + assert.equal(lineage.length, 4, JSON.stringify(lineage)); + assert.equal(fusionUsageBody.budget.spentMicros, 0, "dynamic local models retain zero-price accounting under a budgeted policy"); + const blockedLineage = fusionUsageBody.usage.events.find((event) => event.request_id === "fusion-e2e-budget-blocked"); + assert.equal(blockedLineage.status_code, 402); + assert.equal(blockedLineage.compound_request_stage, "fusion_synthesizer"); + assert.equal(blockedLineage.compound_request_size, 1, "pre-fan-out rejection is a complete one-call Fusion group"); + assert.ok(blockedLineage.compound_request_started_at_ms <= blockedLineage.occurred_at_ms); + const selectionSynthesizer = fusionUsageBody.usage.events.find((event) => event.request_id === "fusion-e2e-selection-failure" && event.compound_request_stage === "fusion_synthesizer"); + const selectionLineage = fusionUsageBody.usage.events.filter((event) => event.compound_request_id === selectionSynthesizer.compound_request_id); + assert.equal(selectionLineage.length, 2); + const selectionAdviser = selectionLineage.find((event) => event.compound_request_stage === "fusion_adviser"); + assert.equal(selectionAdviser.provider, "azure-openai"); + assert.equal(selectionAdviser.status_code, 503); + for (const compoundId of compoundIds) { + const events = fusionUsageBody.usage.events.filter((event) => event.compound_request_id === compoundId); + assert.deepEqual(events.map((event) => event.compound_request_stage).sort(), ["fusion_adviser", "fusion_synthesizer"]); + assert.equal(events.find((event) => event.compound_request_stage === "fusion_synthesizer").request_id, "fusion-e2e-retry"); + assert.notEqual(events.find((event) => event.compound_request_stage === "fusion_adviser").request_id, "fusion-e2e-retry"); + assert.equal(events.find((event) => event.compound_request_stage === "fusion_adviser").compound_request_index, 1); + assert.ok(events.every((event) => event.compound_request_size === 2)); + assert.equal(new Set(events.map((event) => event.compound_request_started_at_ms)).size, 1); + } const legacyInvalidGrant = bootstrapBody.grants.find((entry) => entry.tokenRef === "legacy_invalid"); assert.equal(legacyInvalidGrant.hasCredential, false, "stored empty credential bundles are not reported as configured"); assert.deepEqual(legacyInvalidGrant.credentialFields, []); @@ -379,9 +423,9 @@ try { async function json(url) { const response = await fetch(url); assert.equal(response.status, 200, `${url} returned ${response.status}`); return response.json(); } async function waitUntil(predicate, message) { - const deadline = Date.now() + 2_000; + const deadline = Date.now() + 5_000; while (Date.now() < deadline) { - if (predicate()) return; + if (await predicate()) return; await new Promise((resolve) => setTimeout(resolve, 25)); } throw new Error(message); diff --git a/shared/contracts.ts b/shared/contracts.ts index 5619567..f7dcd02 100644 --- a/shared/contracts.ts +++ b/shared/contracts.ts @@ -184,6 +184,11 @@ export interface UsageAuditEvent { client?: string | null; key_id?: string | null; request_id?: string | null; + compound_request_id?: string | null; + compound_request_stage?: "fusion_adviser" | "fusion_synthesizer" | null; + compound_request_index?: number | null; + compound_request_size?: number | null; + compound_request_started_at_ms?: number | null; provider: string; capability?: string | null; model?: string | null; diff --git a/worker/proxy.ts b/worker/proxy.ts index cdb5ed2..4f8621f 100644 --- a/worker/proxy.ts +++ b/worker/proxy.ts @@ -30,6 +30,11 @@ interface ProxySelection { timeoutMs?: number; } +interface ProxySelectionFailure { + response: Response; + auditSelection: ProxySelection | null; +} + interface PreparedUpstream { headers: Headers; url: URL; @@ -45,6 +50,14 @@ interface ReservedProxyBudget { capability: string; } +interface CompoundRequestContext { + id: string; + stage: "fusion_adviser" | "fusion_synthesizer"; + index: number | null; + size: number; + startedAtMs: number; +} + export async function proxyOpenAi(request: Request, env: Env, context: ExecutionContext, path: string, mode: AuthMode): Promise { const preauthenticated = await preauthenticate(request, env, mode); if (preauthenticated instanceof Response) return preauthenticated; @@ -68,28 +81,44 @@ async function proxyConcreteOpenAi( preauthenticated: AuthorizedIdentity | null, timeoutMs?: number, reservedBudget?: ReservedProxyBudget, + compound?: CompoundRequestContext, + auditAuth?: AuthorizedIdentity, ): Promise { - const selection = concreteOpenAiSelection(path, body, env, timeoutMs); - return selection instanceof Response ? selection : proxySelected(request, env, context, mode, selection, {}, preauthenticated, reservedBudget); + const result = concreteOpenAiSelection(path, body, env, timeoutMs); + if (isSelectionFailure(result)) { + if (compound && result.auditSelection) await auditSelectionFailure(request, env, context, mode, result.auditSelection, preauthenticated, result.response.status, compound, auditAuth); + return result.response; + } + return proxySelected(request, env, context, mode, result, {}, preauthenticated, reservedBudget, compound, auditAuth); } -function concreteOpenAiSelection(path: string, body: Record, env: Env, timeoutMs?: number): ProxySelection | Response { +function concreteOpenAiSelection(path: string, body: Record, env: Env, timeoutMs?: number): ProxySelection | ProxySelectionFailure { const modelId = typeof body.model === "string" ? body.model : ""; const route = modelRoute(modelId); - if (!route) return errorResponse("model_not_found", `model ${modelId} is not registered`, 404); + if (!route) return selectionFailure(errorResponse("model_not_found", `model ${modelId} is not registered`, 404)); const capability = capabilityForPath(path); const endpoint = endpointForPath(route.provider, path); - if (!capability || !endpoint || !route.model.capabilities.includes(capability)) return errorResponse("model_capability_unsupported", `model ${modelId} does not support ${path}`, 400); + if (!capability || !endpoint || !route.model.capabilities.includes(capability)) return selectionFailure(errorResponse("model_capability_unsupported", `model ${modelId} does not support ${path}`, 400)); try { const upstreamModel = resolvedUpstreamModel(route.provider, route.model, env); const transformed = transformRequestBody(route.provider, path, upstreamModel, { ...body, model: upstreamModel }, env); return { provider: route.provider, endpoint, model: route.model, capability, body: transformed, pathParams: { model: upstreamModel, deployment: upstreamModel }, method: "POST", timeoutMs }; } catch (error) { const failure = error instanceof HttpError ? error : new HttpError(503, "provider_request_invalid", "provider request configuration is invalid"); - return errorResponse(failure.code, failure.message, failure.status); + return selectionFailure(errorResponse(failure.code, failure.message, failure.status), { + provider: route.provider, endpoint, model: route.model, capability, body, pathParams: { model: modelId, deployment: modelId }, method: "POST", timeoutMs, + }); } } +function selectionFailure(response: Response, auditSelection: ProxySelection | null = null): ProxySelectionFailure { + return { response, auditSelection }; +} + +function isSelectionFailure(value: ProxySelection | ProxySelectionFailure): value is ProxySelectionFailure { + return "response" in value; +} + async function proxyFusion( request: Request, env: Env, @@ -98,21 +127,40 @@ async function proxyFusion( body: Record, preauthenticated: AuthorizedIdentity | null, ): Promise { + const compoundStartedAtMs = Date.now(); const config = await loadFusionConfig(env); if (!config.enabled) return errorResponse("fusion_disabled", `${FUSION_MODEL_ID} is not enabled`, 404); if (!fusionMessagesValid(body.messages)) return errorResponse("fusion_messages_invalid", "fusion messages must be an array of objects with string roles", 400); + const requestId = request.headers.get("x-request-id") ?? randomId("req"); + const compoundRequestId = randomId("fusion"); + const compoundRequestSize = config.adviserModels.length + 1; + const fusionHeaders = new Headers(request.headers); + fusionHeaders.set("x-request-id", requestId); + // The original body was already parsed. Concrete routes serialize their transformed body separately. + const fusionRequest = new Request(request.url, { method: request.method, headers: fusionHeaders, signal: request.signal }); const aggregatorSelection = concreteOpenAiSelection("/v1/chat/completions", buildAggregatorBody(body, config, buildFusionReservationProposals(config)), env); - if (aggregatorSelection instanceof Response) return aggregatorSelection; - const aggregatorBudget = await reserveSelected(request, env, context, mode, aggregatorSelection, preauthenticated); + if (isSelectionFailure(aggregatorSelection)) { + if (aggregatorSelection.auditSelection) await auditSelectionFailure(fusionRequest, env, context, mode, aggregatorSelection.auditSelection, preauthenticated, aggregatorSelection.response.status, { + id: compoundRequestId, stage: "fusion_synthesizer", index: null, size: 1, startedAtMs: compoundStartedAtMs, + }); + return aggregatorSelection.response; + } + const aggregatorBudget = await reserveSelected(fusionRequest, env, context, mode, aggregatorSelection, preauthenticated, { + id: compoundRequestId, stage: "fusion_synthesizer", index: null, size: 1, startedAtMs: compoundStartedAtMs, + }); if (aggregatorBudget instanceof Response) return aggregatorBudget; const result = await collectFusionProposals(config, body, async (model, adviserBody, timeoutMs, index, signal) => { - const headers = new Headers(request.headers); + const headers = new Headers(fusionRequest.headers); headers.set("x-request-id", randomId(`fusion-adviser-${index + 1}`)); - const adviserRequest = new Request(request.url, { method: "POST", headers, signal: AbortSignal.any([request.signal, signal]) }); - return proxyConcreteOpenAi(adviserRequest, env, context, "/v1/chat/completions", mode, adviserBody, preauthenticated, timeoutMs); + const adviserRequest = new Request(fusionRequest.url, { method: "POST", headers, signal: AbortSignal.any([fusionRequest.signal, signal]) }); + return proxyConcreteOpenAi(adviserRequest, env, context, "/v1/chat/completions", mode, adviserBody, preauthenticated, timeoutMs, undefined, { + id: compoundRequestId, stage: "fusion_adviser", index: index + 1, size: compoundRequestSize, startedAtMs: compoundStartedAtMs, + }, aggregatorBudget.auth); }); const aggregatorBody = buildAggregatorBody(body, config, result.proposals); - const response = await proxyConcreteOpenAi(request, env, context, "/v1/chat/completions", mode, aggregatorBody, preauthenticated, undefined, aggregatorBudget); + const response = await proxyConcreteOpenAi(fusionRequest, env, context, "/v1/chat/completions", mode, aggregatorBody, preauthenticated, undefined, aggregatorBudget, { + id: compoundRequestId, stage: "fusion_synthesizer", index: null, size: compoundRequestSize, startedAtMs: compoundStartedAtMs, + }); const headers = new Headers(response.headers); headers.set("x-clawrouter-fusion", result.proposals.length ? "advisers" : "aggregator-only"); headers.set("x-clawrouter-fusion-aggregator", config.aggregatorModel); @@ -244,15 +292,21 @@ export async function inspectKey(headers: Headers, env: Env): Promise }); } -async function proxySelected(request: Request, env: Env, context: ExecutionContext, mode: AuthMode, selection: ProxySelection, queryInput: Record = {}, preauthenticated: AuthorizedIdentity | null = null, reservedBudget?: ReservedProxyBudget): Promise { +async function proxySelected(request: Request, env: Env, context: ExecutionContext, mode: AuthMode, selection: ProxySelection, queryInput: Record = {}, preauthenticated: AuthorizedIdentity | null = null, reservedBudget?: ReservedProxyBudget, compound?: CompoundRequestContext, auditAuth?: AuthorizedIdentity): Promise { const auth = reservedBudget?.auth ?? await selectedAuth(request, env, mode, selection, preauthenticated); - if (auth instanceof Response) return auth; + if (auth instanceof Response) { + if (compound && auditAuth) { + const requestId = request.headers.get("x-request-id") ?? randomId("req"); + auditFailure(context, env, auditAuth, selection, request, requestId, Date.now(), estimateCost(selection.model, selection.body, auditAuth.policy.requestCostMicros, selection.capability), auth.status, auth.status === 403 ? "denied" : auth.status < 500 ? "client_error" : "provider_error", compound); + } + return auth; + } const requestId = request.headers.get("x-request-id") ?? randomId("req"); const started = Date.now(); const estimatedCost = estimateCost(selection.model, selection.body, auth.policy.requestCostMicros, selection.capability); const cost = reservedBudget?.cost ?? estimatedCost; if (reservedBudget && (reservedBudget.providerId !== selection.provider.id || reservedBudget.modelId !== selection.model?.id || reservedBudget.capability !== selection.capability || estimatedCost.reserveMicros > reservedBudget.cost.reserveMicros)) { - context.waitUntil(finalizeAccounting(env, auth, reservedBudget.reservation, 0, usageEvent(auth, selection, request, requestId, started, 500, null, cost, reservedBudget.reservation, null, "provider_error"))); + context.waitUntil(finalizeAccounting(env, auth, reservedBudget.reservation, 0, usageEvent(auth, selection, request, requestId, started, 500, null, cost, reservedBudget.reservation, null, "provider_error", 0, compound))); return errorResponse("fusion_reservation_invalid", "fusion synthesizer reservation does not cover the final request", 500); } let prepared: PreparedUpstream; @@ -260,8 +314,8 @@ async function proxySelected(request: Request, env: Env, context: ExecutionConte catch (error) { const failure = selectedFailure(error); const status = failure.status === 403 ? "denied" : failure.status < 500 ? "client_error" : "provider_error"; - if (reservedBudget) context.waitUntil(finalizeAccounting(env, auth, reservedBudget.reservation, 0, usageEvent(auth, selection, request, requestId, started, failure.status, null, cost, reservedBudget.reservation, null, status))); - else auditFailure(context, env, auth, selection, request, requestId, started, cost, failure.status, status); + if (reservedBudget) context.waitUntil(finalizeAccounting(env, auth, reservedBudget.reservation, 0, usageEvent(auth, selection, request, requestId, started, failure.status, null, cost, reservedBudget.reservation, null, status, 0, compound))); + else auditFailure(context, env, auth, selection, request, requestId, started, cost, failure.status, status, compound); return errorResponse(failure.code, failure.message, failure.status); } let reservation = reservedBudget?.reservation; @@ -269,14 +323,14 @@ async function proxySelected(request: Request, env: Env, context: ExecutionConte try { reservation = await reserveBudget(env, auth, selection.capability, cost); } catch (error) { const failure = error instanceof HttpError ? error : new HttpError(503, "budget_store_unavailable", "budget ledger is unavailable"); - auditFailure(context, env, auth, selection, request, requestId, started, cost, failure.status, failure.status === 402 ? "denied" : failure.status < 500 ? "client_error" : "provider_error"); + auditFailure(context, env, auth, selection, request, requestId, started, cost, failure.status, failure.status === 402 ? "denied" : failure.status < 500 ? "client_error" : "provider_error", compound); return errorResponse(failure.code, failure.message, failure.status); } } let content: string | null; try { content = await retainRequestContent(env, auth, selection, requestId); } catch { - context.waitUntil(finalizeAccounting(env, auth, reservation, 0, usageEvent(auth, selection, request, requestId, started, 503, null, cost, reservation, null, "provider_error"))); + context.waitUntil(finalizeAccounting(env, auth, reservation, 0, usageEvent(auth, selection, request, requestId, started, 503, null, cost, reservation, null, "provider_error", 0, compound))); return errorResponse("content_retention_unavailable", "required request-content retention is temporarily unavailable", 503); } const controller = new AbortController(); @@ -287,12 +341,12 @@ async function proxySelected(request: Request, env: Env, context: ExecutionConte response = await fetch(prepared.url, { method: selection.method, headers: prepared.headers, body: prepared.requestBody, signal: AbortSignal.any([request.signal, controller.signal]) }); } catch (error) { clearTimeout(timeout); - context.waitUntil(finalizeAccounting(env, auth, reservation, 0, usageEvent(auth, selection, request, requestId, started, 502, null, cost, reservation, content, error instanceof DOMException && error.name === "AbortError" ? "timeout" : "provider_error"))); + context.waitUntil(finalizeAccounting(env, auth, reservation, 0, usageEvent(auth, selection, request, requestId, started, 502, null, cost, reservation, content, error instanceof DOMException && error.name === "AbortError" ? "timeout" : "provider_error", 0, compound))); return errorResponse("provider_unavailable", `upstream request to provider ${selection.provider.id} failed`, 502, undefined); } clearTimeout(timeout); const clone = response.clone(); - context.waitUntil(finalizeResponse(clone, env, auth, selection, request, requestId, started, cost, reservation, content)); + context.waitUntil(finalizeResponse(clone, env, auth, selection, request, requestId, started, cost, reservation, content, compound)); const outputHeaders = new Headers(response.headers); for (const name of ["connection", "keep-alive", "proxy-authenticate", "proxy-authorization", "set-cookie", "trailer", "transfer-encoding", "upgrade"]) outputHeaders.delete(name); outputHeaders.set("x-clawrouter-upstream-provider", selection.provider.id); @@ -300,7 +354,7 @@ async function proxySelected(request: Request, env: Env, context: ExecutionConte return new Response(response.body, { status: response.status, statusText: response.statusText, headers: outputHeaders }); } -async function reserveSelected(request: Request, env: Env, context: ExecutionContext, mode: AuthMode, selection: ProxySelection, preauthenticated: AuthorizedIdentity | null): Promise { +async function reserveSelected(request: Request, env: Env, context: ExecutionContext, mode: AuthMode, selection: ProxySelection, preauthenticated: AuthorizedIdentity | null, compound?: CompoundRequestContext): Promise { const auth = await selectedAuth(request, env, mode, selection, preauthenticated); if (auth instanceof Response) return auth; const requestId = request.headers.get("x-request-id") ?? randomId("req"); @@ -311,7 +365,7 @@ async function reserveSelected(request: Request, env: Env, context: ExecutionCon } catch (error) { const failure = selectedFailure(error); const status = failure.status === 403 ? "denied" : failure.status < 500 ? "client_error" : "provider_error"; - auditFailure(context, env, auth, selection, request, requestId, started, cost, failure.status, status); + auditFailure(context, env, auth, selection, request, requestId, started, cost, failure.status, status, compound); return errorResponse(failure.code, failure.message, failure.status); } try { @@ -319,7 +373,7 @@ async function reserveSelected(request: Request, env: Env, context: ExecutionCon return { auth, reservation, cost, providerId: selection.provider.id, modelId: selection.model?.id ?? null, capability: selection.capability }; } catch (error) { const failure = error instanceof HttpError ? error : new HttpError(503, "budget_store_unavailable", "budget ledger is unavailable"); - auditFailure(context, env, auth, selection, request, requestId, started, cost, failure.status, failure.status === 402 ? "denied" : failure.status < 500 ? "client_error" : "provider_error"); + auditFailure(context, env, auth, selection, request, requestId, started, cost, failure.status, failure.status === 402 ? "denied" : failure.status < 500 ? "client_error" : "provider_error", compound); return errorResponse(failure.code, failure.message, failure.status); } } @@ -354,9 +408,18 @@ function selectedFailure(error: unknown): HttpError { return error instanceof HttpError ? error : new HttpError(503, "provider_unavailable", "provider request preflight failed"); } -function auditFailure(context: ExecutionContext, env: Env, auth: AuthorizedIdentity, selection: ProxySelection, request: Request, requestId: string, started: number, cost: Cost, statusCode: number, status: UsageEvent["status"]): void { +function auditFailure(context: ExecutionContext, env: Env, auth: AuthorizedIdentity, selection: ProxySelection, request: Request, requestId: string, started: number, cost: Cost, statusCode: number, status: UsageEvent["status"], compound?: CompoundRequestContext): void { const reservation = { reservationId: null, reservedMicros: 0 }; - context.waitUntil(finalizeAccounting(env, auth, reservation, 0, usageEvent(auth, selection, request, requestId, started, statusCode, null, cost, reservation, null, status))); + context.waitUntil(finalizeAccounting(env, auth, reservation, 0, usageEvent(auth, selection, request, requestId, started, statusCode, null, cost, reservation, null, status, 0, compound))); +} + +async function auditSelectionFailure(request: Request, env: Env, context: ExecutionContext, mode: AuthMode, selection: ProxySelection, preauthenticated: AuthorizedIdentity | null, statusCode: number, compound: CompoundRequestContext, auditAuth?: AuthorizedIdentity): Promise { + const selected = await selectedAuth(request, env, mode, selection, preauthenticated); + const auth = selected instanceof Response ? auditAuth : selected; + if (!auth) return; + const requestId = request.headers.get("x-request-id") ?? randomId("req"); + const status = statusCode === 403 ? "denied" : statusCode < 500 ? "client_error" : "provider_error"; + auditFailure(context, env, auth, selection, request, requestId, Date.now(), estimateCost(selection.model, selection.body, auth.policy.requestCostMicros, selection.capability), statusCode, status, compound); } async function preauthenticate(request: Request, env: Env, mode: AuthMode, providerId?: string): Promise { @@ -364,7 +427,7 @@ async function preauthenticate(request: Request, env: Env, mode: AuthMode, provi return providerId ? accessIdentity(request, env, providerId) : null; } -async function finalizeResponse(response: Response, env: Env, auth: AuthorizedIdentity, selection: ProxySelection, request: Request, requestId: string, started: number, estimated: Cost, reservation: BudgetReservation, content: string | null): Promise { +async function finalizeResponse(response: Response, env: Env, auth: AuthorizedIdentity, selection: ProxySelection, request: Request, requestId: string, started: number, estimated: Cost, reservation: BudgetReservation, content: string | null, compound?: CompoundRequestContext): Promise { let tokens: Tokens | null = null; try { const contentType = response.headers.get("content-type") ?? ""; @@ -373,7 +436,7 @@ async function finalizeResponse(response: Response, env: Env, auth: AuthorizedId } catch { /* usage is best-effort; reservation stays conservative */ } const measured = tokens ? actualCost(selection.model, tokens, auth.policy.requestCostMicros) : null; const actual = response.ok && measured != null ? measured : response.ok ? estimated.reserveMicros : 0; - await finalizeAccounting(env, auth, reservation, actual, usageEvent(auth, selection, request, requestId, started, response.status, tokens, estimated, reservation, content, response.ok ? "success" : response.status < 500 ? "client_error" : "provider_error", actual)); + await finalizeAccounting(env, auth, reservation, actual, usageEvent(auth, selection, request, requestId, started, response.status, tokens, estimated, reservation, content, response.ok ? "success" : response.status < 500 ? "client_error" : "provider_error", actual, compound)); } type Cost = EstimatedCost; @@ -395,14 +458,17 @@ function actualCost(model: CompiledModel | null, tokens: Tokens, fixed: number | return actualModelCost(pricing, tokens); } -function usageEvent(auth: AuthorizedIdentity, selection: ProxySelection, request: Request, requestId: string, started: number, statusCode: number, tokens: Tokens | null, cost: Cost, reservation: BudgetReservation, contentRef: string | null, status: UsageEvent["status"], actual = 0): UsageEvent { +function usageEvent(auth: AuthorizedIdentity, selection: ProxySelection, request: Request, requestId: string, started: number, statusCode: number, tokens: Tokens | null, cost: Cost, reservation: BudgetReservation, contentRef: string | null, status: UsageEvent["status"], actual = 0, compound?: CompoundRequestContext): UsageEvent { return { id: randomId("usage"), type: "clawrouter.usage.v1", occurred_at_ms: Date.now(), tenant_id: auth.policy.tenantId ?? "default", policy_id: auth.policyId, credential_id: auth.credentialId, principal_id: auth.principalId, auth_type: auth.authType, session_id: clampAudit(request.headers.get("x-clawrouter-session-id") ?? request.headers.get("session-id")), agent_id: clampAudit(request.headers.get("x-clawrouter-agent-id")), parent_agent_id: clampAudit(request.headers.get("x-clawrouter-parent-agent-id")), project_id: clampAudit(request.headers.get("x-clawrouter-project-id")), client: clampAudit(request.headers.get("x-clawrouter-client")), - key_id: auth.credentialId ?? auth.policyId, request_id: requestId, provider: selection.provider.id, capability: selection.capability, + key_id: auth.credentialId ?? auth.policyId, request_id: requestId, + compound_request_id: compound?.id ?? null, compound_request_stage: compound?.stage ?? null, compound_request_index: compound?.index ?? null, + compound_request_size: compound?.size ?? null, compound_request_started_at_ms: compound?.startedAtMs ?? null, + provider: selection.provider.id, capability: selection.capability, model: selection.model?.id ?? null, input_tokens: tokens?.input ?? null, output_tokens: tokens?.output ?? null, total_tokens: tokens?.total ?? null, cached_input_tokens: tokens?.cached ?? null, cache_write_input_tokens: tokens?.cacheWrite ?? null, reserved_cost_micros: reservation.reservedMicros, actual_cost_micros: actual, reserved_input_tokens: cost.inputTokens, diff --git a/worker/types.ts b/worker/types.ts index 4546d6d..5665669 100644 --- a/worker/types.ts +++ b/worker/types.ts @@ -193,6 +193,11 @@ export interface UsageEvent { client: string | null; key_id: string; request_id: string; + compound_request_id: string | null; + compound_request_stage: "fusion_adviser" | "fusion_synthesizer" | null; + compound_request_index: number | null; + compound_request_size: number | null; + compound_request_started_at_ms: number | null; provider: string; capability: string; model: string | null;