diff --git a/.withkids_tracker.md b/.withkids_tracker.md new file mode 100644 index 00000000..26ba9b4f --- /dev/null +++ b/.withkids_tracker.md @@ -0,0 +1,32 @@ +# withkids tracker + +## Goal +Implement missing caching tracking (history cache control, miss/write tokens) and UI indicators (list view cache column, silent degradation warning) for PR #445 / CacheAnalytics. + +## Phase +M0: Planning & Setup + +## Todos +- [x] T1: Fix Cache Conversation History (apply `cache_control` to history blocks in `litellm.py`) (agent: backend-fixer, deps: []) +- [x] T2: Track misses/writes and silent degradation flag in backend (`litellm.py` & `agent_integration.py`) (agent: backend-fixer, deps: []) +- [x] T3: Update Executions list view to show cache column (agent: frontend-fixer, deps: [T2]) +- [x] T4: UI indication for silent degradation (agent: frontend-fixer, deps: [T2]) +- [ ] T5: Run tests (cached and uncached) and verify (parent, deps: [T1, T2, T3, T4]) +- [ ] T6: Capture visual proof (screenshot) and attach to PR (parent, deps: [T5]) + +## Blockers + + +## Status +Frontend Fixer: Completed T3 & T4. +- Added 'Cached Tokens' sortable column to Executions list table in `Executions.tsx` and updated `agentRunApi.ts` to request and return `cached_tokens`. +- Added `checkCacheableModels` helper in `agentApi.ts`. +- Added silent degradation warning alert in `GeneralTab.tsx` (Agent Form) when prompt caching is enabled on an unsupported model. +- Added silent degradation warning alert in `AgentRunDetailPage.tsx` when an agent run used an unsupported model with prompt caching enabled, and added 'Cached Tokens' column to child runs table. +- Verified frontend build (`npm run build`) succeeded without errors. + +Backend Fixer: Completed T1 & T2. +- Updated `huf/ai/providers/litellm.py`: Added `_format_conversation_history` helper to format conversation history blocks with `cache_control: {"type": cache_control_type}` on the history prefix breakpoint when history caching is enabled (in both `run` and `run_stream`). +- Updated `huf/ai/providers/litellm.py`: Set and returned `cache_skipped_unsupported_model` flag when prompt caching is enabled on an agent but unsupported by the model. Extracted and tracked `cache_creation_tokens` and `cache_miss_tokens` in returned usage stats for both `run` and `run_stream`. +- Updated `huf/ai/agent_integration.py`: Recorded `cache_creation_tokens`, `cache_miss_tokens`, and `cache_skipped_unsupported_model` in `Agent Run`'s `usage_snapshot` JSON payload for both synchronous and streaming execution modes. +- Verified python syntax via `py_compile`. diff --git a/docs/pr-assets/agent-run-analytics/cache_analytics.png b/docs/pr-assets/agent-run-analytics/cache_analytics.png new file mode 100644 index 00000000..792f61c9 Binary files /dev/null and b/docs/pr-assets/agent-run-analytics/cache_analytics.png differ diff --git a/docs/pr-assets/context-cache-observability/context_bar.png b/docs/pr-assets/context-cache-observability/context_bar.png new file mode 100644 index 00000000..132f5401 Binary files /dev/null and b/docs/pr-assets/context-cache-observability/context_bar.png differ diff --git a/frontend/src/components/agent/GeneralTab.tsx b/frontend/src/components/agent/GeneralTab.tsx index 9ea36573..6c9605fe 100644 --- a/frontend/src/components/agent/GeneralTab.tsx +++ b/frontend/src/components/agent/GeneralTab.tsx @@ -1,3 +1,4 @@ +import { useState, useEffect } from 'react'; import { FormField, FormItem, FormLabel, FormControl, FormDescription, FormMessage } from '@/components/ui/form'; import { Input } from '@/components/ui/input'; import { Textarea } from '@/components/ui/textarea'; @@ -6,6 +7,8 @@ import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from '@ import { Switch } from '@/components/ui/switch'; import { Card, CardContent, CardDescription, CardHeader, CardTitle } from '@/components/ui/card'; import { Accordion, AccordionContent, AccordionItem, AccordionTrigger } from '@/components/ui/accordion'; +import { Alert, AlertTitle, AlertDescription } from '@/components/ui/alert'; +import { AlertTriangle } from 'lucide-react'; import { UseFormReturn } from 'react-hook-form'; import type { AIProvider, AIModel } from '@/types/agent.types'; import type { AgentFormValues } from './types'; @@ -13,6 +16,7 @@ import { InstructionsTextarea } from './InstructionsTextarea'; import { PromptTemplateSection, type AgentPromptOption } from './PromptTemplateSection'; import { LinkFieldControl } from '@/components/ui/link-field-control'; import { linkRoutes } from '@/lib/link-routes'; +import { checkCacheableModels, type CacheableModelsResponse } from '@/services/agentApi'; interface GeneralTabProps { form: UseFormReturn; @@ -38,8 +42,27 @@ export function GeneralTab({ showAddNewPrompt = true, }: GeneralTabProps) { const watchEnablePromptCaching = form.watch('enable_prompt_caching'); + const watchModel = form.watch('model'); const promptMode = form.watch('prompt_mode'); + const [cacheStatus, setCacheStatus] = useState(null); + + useEffect(() => { + if (!watchEnablePromptCaching || !watchProvider || !watchModel) { + setCacheStatus(null); + return; + } + let cancelled = false; + checkCacheableModels(watchProvider, watchModel).then((res) => { + if (!cancelled) { + setCacheStatus(res); + } + }); + return () => { + cancelled = true; + }; + }, [watchEnablePromptCaching, watchProvider, watchModel]); + return (
@@ -310,6 +333,23 @@ We generally recommend altering this or temperature but not both.`} )} /> + {watchEnablePromptCaching && watchProvider && watchModel && cacheStatus && !cacheStatus.supported && ( + + + Silent Degradation Warning: Prompt Caching Not Supported + +

+ The selected model {watchModel} does not support prompt caching for provider {watchProvider}. Prompt caching will be silently skipped during execution. +

+ {cacheStatus.alternatives.length > 0 && ( +

+ Supported alternative models for {watchProvider}: {cacheStatus.alternatives.slice(0, 5).join(', ')} +

+ )} + + + )} + {watchEnablePromptCaching && ( {label}
{value}

{detail}

; +} + +export function ExecutionAnalyticsDashboard() { + const [data, setData] = useState(null); + const [loading, setLoading] = useState(true); + useEffect(() => { getExecutionAnalytics().then((result) => { setData(result); setLoading(false); }); }, []); + if (loading) return
Loading scheduled execution analytics…
; + if (!data) return null; + const summary = data.summary; + return

Execution analytics

Scheduled rollup · {data.metadata.granularity}

Metrics are read from scheduled aggregate buckets, not calculated from raw run rows in the browser.

; +} diff --git a/frontend/src/components/ui/context-bar.tsx b/frontend/src/components/ui/context-bar.tsx new file mode 100644 index 00000000..2e374760 --- /dev/null +++ b/frontend/src/components/ui/context-bar.tsx @@ -0,0 +1,92 @@ +import { cn } from '@/lib/utils'; +import type { SegmentTokens } from '@/types/runContextMetrics.types'; + +export interface ContextBarCacheState { + cacheRead: number; + cacheWrite: number; + uncached: number; +} + +export interface ContextBarProps { + segments: SegmentTokens; + cacheState?: ContextBarCacheState; + total: number; + size?: 'sm' | 'md'; + className?: string; + onClick?: () => void; +} + +const SEGMENT_ORDER: Array<{ key: keyof SegmentTokens; label: string; color: string }> = [ + { key: 'system', label: 'System', color: 'bg-slate-500' }, + { key: 'tools', label: 'Tools', color: 'bg-indigo-500' }, + { key: 'knowledge', label: 'Knowledge', color: 'bg-emerald-500' }, + { key: 'history', label: 'History', color: 'bg-amber-500' }, + { key: 'message', label: 'Message', color: 'bg-rose-500' }, +]; + +/** + * Segmented context/cache meter. Top strip is composition (what fills the + * window, by tokens); bottom strip — when cacheState is supplied — is cache + * economics (what each token cost this turn). Same component at every + * scope: chat header (size="sm"), run detail / agent / fleet (size="md"). + */ +export function ContextBar({ segments, cacheState, total, size = 'md', className, onClick }: ContextBarProps) { + const height = size === 'sm' ? 'h-1' : 'h-2.5'; + const known = SEGMENT_ORDER.reduce((sum, { key }) => sum + (segments[key] ?? 0), 0); + const headroom = total > known ? total - known : 0; + + const cacheTotal = cacheState ? cacheState.cacheRead + cacheState.cacheWrite + cacheState.uncached : 0; + + const Wrapper = onClick ? 'button' : 'div'; + + return ( + +
+ {SEGMENT_ORDER.map(({ key, label, color }) => { + const value = segments[key]; + if (!value) return null; + const pct = total > 0 ? (value / total) * 100 : 0; + return ( +
+ ); + })} + {headroom > 0 && ( +
+ )} +
+ + {cacheState && cacheTotal > 0 && ( +
+
+
+
+
+ )} + + ); +} diff --git a/frontend/src/pages/AgentRunDetailPage.tsx b/frontend/src/pages/AgentRunDetailPage.tsx index be29a209..e881e5bc 100644 --- a/frontend/src/pages/AgentRunDetailPage.tsx +++ b/frontend/src/pages/AgentRunDetailPage.tsx @@ -1,17 +1,22 @@ import { useEffect, useState, useMemo } from 'react'; import { useParams, useNavigate, Link } from 'react-router-dom'; -import { ArrowLeft, ArrowUpDown, Loader2 } from 'lucide-react'; +import { ArrowLeft, ArrowUpDown, Loader2, AlertTriangle } from 'lucide-react'; import { Card, CardContent, CardHeader, CardTitle, CardDescription } from '@/components/ui/card'; import { Badge } from '@/components/ui/badge'; import { Button } from '@/components/ui/button'; +import { Alert, AlertTitle, AlertDescription } from '@/components/ui/alert'; import type { AgentRunDoc } from '@/services/agentRunApi'; import { getAgentRuns } from '@/services/agentRunApi'; +import { checkCacheableModels } from '@/services/agentApi'; import { db } from '@/lib/frappe-sdk'; import { doctype } from '@/data/doctypes'; import { handleFrappeError } from '@/lib/frappe-error'; import { calculateDuration, formatTimeAgo } from '@/utils/time'; import { getAgentRunStatusVariant } from '@/utils/status'; import { getArtifacts, type AgentContextArtifactDoc } from '@/services/agentContextArtifactApi'; +import { getRunContextMetrics } from '@/services/runContextMetricsApi'; +import type { RunContextMetricsResponse } from '@/types/runContextMetrics.types'; +import { ContextBar } from '@/components/ui/context-bar'; import { ColumnDef, flexRender, @@ -36,7 +41,9 @@ interface AgentRunDetail extends AgentRunDoc { model?: string; input_tokens?: number | null; output_tokens?: number | null; + cached_tokens?: number | null; cost?: number | null; + cost_source?: string | null; } async function fetchAgentRunDetail(name: string): Promise { @@ -108,6 +115,8 @@ function AgentRunDetailPage() { const [childRuns, setChildRuns] = useState([]); const [loadingChildRuns, setLoadingChildRuns] = useState(false); const [sorting, setSorting] = useState([]); + const [isSilentDegradation, setIsSilentDegradation] = useState(false); + const [contextMetrics, setContextMetrics] = useState(null); useEffect(() => { if (!runId) { @@ -121,8 +130,46 @@ function AgentRunDetailPage() { setRun(data); setLoading(false); })(); + + (async () => { + const metrics = await getRunContextMetrics(runId); + setContextMetrics(metrics); + })(); }, [runId]); + // Check for silent degradation (caching enabled on agent but unsupported by model) + useEffect(() => { + if (!run || !run.agent || !run.provider || !run.model) { + setIsSilentDegradation(false); + return; + } + + let cancelled = false; + (async () => { + try { + const agentDoc = await db.getDoc(doctype.Agent, run.agent); + if (agentDoc && agentDoc.enable_prompt_caching) { + const cacheCheck = await checkCacheableModels(run.provider, run.model); + if (!cancelled && !cacheCheck.supported) { + setIsSilentDegradation(true); + return; + } + } + if (!cancelled) { + setIsSilentDegradation(false); + } + } catch { + if (!cancelled) { + setIsSilentDegradation(false); + } + } + })(); + + return () => { + cancelled = true; + }; + }, [run]); + // Fetch child runs when run is loaded useEffect(() => { if (!runId || !run) { @@ -205,6 +252,29 @@ function AgentRunDetailPage() { ); }, }, + { + accessorKey: 'cached_tokens', + header: ({ column }) => { + return ( + + ); + }, + cell: ({ row }) => { + const cached = row.original.cached_tokens; + return ( +
+ {typeof cached === 'number' ? cached.toLocaleString() : '0'} +
+ ); + }, + }, { id: 'duration', header: 'Duration', @@ -305,6 +375,16 @@ function AgentRunDetailPage() {
+ {isSilentDegradation && ( + + + Silent Degradation Warning: Prompt Caching Skipped + + Prompt caching was enabled for agent {run.agent}, but model {run.model || 'unknown'} from provider {run.provider || 'unknown'} does not support prompt caching. Caching was silently skipped during execution. + + + )} +
@@ -372,18 +452,95 @@ function AgentRunDetailPage() { : 'Not available'}
+
+ Cached Tokens + + {typeof run.cached_tokens === 'number' ? run.cached_tokens : 'Not available'} + +
Cost {typeof run.cost === 'number' ? `$${run.cost.toFixed(6)}` : 'Not available'}
+
+ Cost Source + {run.cost_source || 'Not available'} +
+ {contextMetrics?.segment_tokens && ( + + + Context + + What filled the context window this turn, and whether caching paid off. + + + + +
+
+ Cache read share + + {typeof contextMetrics.metrics.cache_read_share === 'number' + ? `${(contextMetrics.metrics.cache_read_share * 100).toFixed(0)}%` + : 'Unavailable'} + +
+
+ Effective multiplier + + {typeof contextMetrics.metrics.effective_input_multiplier === 'number' + ? `${contextMetrics.metrics.effective_input_multiplier.toFixed(2)}x` + : 'Unavailable'} + +
+
+ Wasted writes + + {typeof contextMetrics.metrics.wasted_writes_tokens === 'number' + ? contextMetrics.metrics.wasted_writes_tokens + : 'Not yet tracked'} + +
+
+ Prefix stability + {contextMetrics.metrics.prefix_stability} +
+
+ Counterfactual savings + + {typeof contextMetrics.metrics.counterfactual_savings === 'number' + ? `$${contextMetrics.metrics.counterfactual_savings.toFixed(6)}` + : 'Unavailable'} + +
+
+ + + )} +
diff --git a/frontend/src/pages/Executions.tsx b/frontend/src/pages/Executions.tsx index 762a5989..a0228a12 100644 --- a/frontend/src/pages/Executions.tsx +++ b/frontend/src/pages/Executions.tsx @@ -27,6 +27,7 @@ import { TableHeader, TableRow, } from '@/components/ui/table'; +import { ExecutionAnalyticsDashboard } from '@/components/executions/ExecutionAnalyticsDashboard'; function getRunStatusDot(status?: string): { variant: StatusDotVariant; label: string } { const normalized = status?.toLowerCase() || ''; @@ -217,6 +218,34 @@ export default function Executions() { ); }, }, + { + accessorKey: 'cached_tokens', + header: ({ column }) => { + return ( + + ); + }, + cell: ({ row }) => { + const cached = row.original.cached_tokens; + return ( +
+ {typeof cached === 'number' ? cached.toLocaleString() : '0'} +
+ ); + }, + sortingFn: (rowA, rowB) => { + const valA = rowA.original.cached_tokens ?? 0; + const valB = rowB.original.cached_tokens ?? 0; + return valA - valB; + }, + }, { id: 'duration', header: 'Duration', @@ -312,6 +341,7 @@ export default function Executions() { /> } > +
{initialLoading ? (
diff --git a/frontend/src/services/agentApi.ts b/frontend/src/services/agentApi.ts index 927910d5..cba82512 100644 --- a/frontend/src/services/agentApi.ts +++ b/frontend/src/services/agentApi.ts @@ -679,3 +679,34 @@ export async function getAgentModels( } } +export interface CacheableModelsResponse { + supported: boolean; + alternatives: string[]; +} + +/** + * Check if a provider/model combination supports prompt caching + */ +export async function checkCacheableModels( + provider?: string, + model?: string +): Promise { + if (!provider) { + return { supported: false, alternatives: [] }; + } + try { + const response = await call.get('huf.huf.doctype.agent.agent.get_cacheable_models', { + provider, + model: model || undefined, + }); + const data = response?.message || response; + return { + supported: Boolean(data?.supported), + alternatives: Array.isArray(data?.alternatives) ? data.alternatives : [], + }; + } catch (error) { + return { supported: false, alternatives: [] }; + } +} + + diff --git a/frontend/src/services/agentRunApi.ts b/frontend/src/services/agentRunApi.ts index e2dfa814..466ee089 100644 --- a/frontend/src/services/agentRunApi.ts +++ b/frontend/src/services/agentRunApi.ts @@ -14,6 +14,12 @@ export interface AgentRunDoc { start_time?: string | null; end_time?: string | null; status?: string; + cached_tokens?: number | null; + input_tokens?: number | null; + output_tokens?: number | null; + cost?: number | null; + cost_source?: string | null; + is_child?: number | boolean; } /** @@ -49,7 +55,7 @@ export async function getAgentRuns( // Backward compatibility: if no params, return simple array if (!params) { const runs = await db.getDocList(doctype['Agent Run'], { - fields: ['name', 'agent', 'start_time', 'end_time', 'status', 'is_child'], + fields: ['name', 'agent', 'start_time', 'end_time', 'status', 'is_child', 'cached_tokens'], limit: 1000, orderBy: { field: 'creation', order: 'desc' }, }); @@ -87,7 +93,7 @@ export async function getAgentRuns( // Fetch data const runs = await db.getDocList(doctype['Agent Run'], { - fields: ['name', 'agent', 'start_time', 'end_time', 'status', 'is_child'], + fields: ['name', 'agent', 'start_time', 'end_time', 'status', 'is_child', 'cached_tokens'], filters: filters.length > 0 ? (filters as Filter>[]) : undefined, limit: limit + 1, // Fetch one extra to check if there's more ...(start > 0 && { limit_start: start }), diff --git a/frontend/src/services/executionAnalyticsApi.ts b/frontend/src/services/executionAnalyticsApi.ts new file mode 100644 index 00000000..ded48db4 --- /dev/null +++ b/frontend/src/services/executionAnalyticsApi.ts @@ -0,0 +1,13 @@ +import { call } from '@/lib/frappe-sdk'; +import { handleFrappeError } from '@/lib/frappe-error'; +import type { ExecutionAnalyticsResponse } from '@/types/executionAnalytics.types'; + +export async function getExecutionAnalytics(): Promise { + try { + const result = await call.get('huf.ai.agent_run_analytics_api.get_execution_analytics', { granularity: 'hour' }); + return result.message as ExecutionAnalyticsResponse; + } catch (error) { + handleFrappeError(error, 'Error fetching execution analytics'); + return null; + } +} diff --git a/frontend/src/services/runContextMetricsApi.ts b/frontend/src/services/runContextMetricsApi.ts new file mode 100644 index 00000000..0a76694a --- /dev/null +++ b/frontend/src/services/runContextMetricsApi.ts @@ -0,0 +1,13 @@ +import { call } from '@/lib/frappe-sdk'; +import { handleFrappeError } from '@/lib/frappe-error'; +import type { RunContextMetricsResponse } from '@/types/runContextMetrics.types'; + +export async function getRunContextMetrics(runName: string): Promise { + try { + const result = await call.get('huf.ai.agent_run_context_api.get_run_context_metrics', { run_name: runName }); + return result.message as RunContextMetricsResponse; + } catch (error) { + handleFrappeError(error, 'Error fetching run context metrics'); + return null; + } +} diff --git a/frontend/src/types/executionAnalytics.types.ts b/frontend/src/types/executionAnalytics.types.ts new file mode 100644 index 00000000..d17a3047 --- /dev/null +++ b/frontend/src/types/executionAnalytics.types.ts @@ -0,0 +1,21 @@ +export interface ExecutionAnalyticsSummary { + run_count: number; + success_count: number; + failed_count: number; + input_tokens: number; + output_tokens: number; + cached_tokens: number; + total_cost: number; + duration_ms_sum: number; + duration_count: number; + success_rate: number | null; + average_duration_ms: number | null; + cache_ratio: number | null; +} + +export interface ExecutionAnalyticsResponse { + summary: ExecutionAnalyticsSummary; + series: Array; + breakdowns: Array; + metadata: { granularity: 'hour' | 'day'; freshness: string | null; source: 'scheduled_rollup' }; +} diff --git a/frontend/src/types/runContextMetrics.types.ts b/frontend/src/types/runContextMetrics.types.ts new file mode 100644 index 00000000..e588fcab --- /dev/null +++ b/frontend/src/types/runContextMetrics.types.ts @@ -0,0 +1,31 @@ +export interface SegmentTokens { + system: number | null; + tools: number | null; + knowledge: number | null; + history: number | null; + message: number | null; +} + +export interface PrefixBreakpoint { + marker: string; + prefix_hash: string; +} + +export type PrefixStability = 'stable' | 'changed' | 'unknown' | 'unavailable'; + +export interface RunContextMetrics { + cache_read_share: number | null; + effective_input_multiplier: number | null; + wasted_writes_tokens: number | null; + prefix_stability: PrefixStability; + counterfactual_savings: number | null; +} + +export interface RunContextMetricsResponse { + segment_tokens: SegmentTokens | null; + total_tokens: number | null; + context_window: number; + prefix_breakpoints: PrefixBreakpoint[]; + cache_skipped_unsupported_model: boolean | null; + metrics: RunContextMetrics; +} diff --git a/huf/ai/agent_integration.py b/huf/ai/agent_integration.py index ee4927fa..0ca67d27 100644 --- a/huf/ai/agent_integration.py +++ b/huf/ai/agent_integration.py @@ -1218,6 +1218,14 @@ def _execute_agent_run( else: enhanced_prompt = base_prompt + from huf.ai.context_segments import compute_segment_tokens, compute_prefix_breakpoints + segment_tokens = compute_segment_tokens( + agent_doc, agent, resolved_model, resolved_provider, history, knowledge_context, prompt + ) + prefix_breakpoints = compute_prefix_breakpoints( + agent_doc, agent, resolved_model, resolved_provider, history + ) + context = { "channel": channel_id, "external_id": external_id, @@ -1390,6 +1398,8 @@ def _execute_agent_run( input_tokens = 0 output_tokens = 0 cached_tokens = 0 + cache_creation_tokens = 0 + cache_skipped_unsupported_model = False if usage: @@ -1404,17 +1414,57 @@ def _execute_agent_run( if details: if isinstance(details, dict): cached_tokens = details.get("cached_tokens") or details.get("cache_hit_tokens") or 0 + cache_creation_tokens = ( + details.get("cache_creation_input_tokens") + or details.get("cache_write_tokens") + or details.get("cache_creation_tokens") + or 0 + ) else: cached_tokens = getattr(details, "cached_tokens", None) or getattr(details, "cache_hit_tokens", None) or 0 + cache_creation_tokens = ( + getattr(details, "cache_creation_input_tokens", None) + or getattr(details, "cache_write_tokens", None) + or getattr(details, "cache_creation_tokens", None) + or 0 + ) elif isinstance(usage, dict): cached_tokens = usage.get("cached_tokens") or usage.get("cache_hit_tokens") or 0 - + cache_creation_tokens = ( + usage.get("cache_creation_tokens") + or usage.get("cache_creation_input_tokens") + or usage.get("cache_write_input_tokens") + or usage.get("cache_miss_tokens") + or 0 + ) + + if not cache_creation_tokens and isinstance(usage, dict): + cache_creation_tokens = ( + usage.get("cache_creation_tokens") + or usage.get("cache_creation_input_tokens") + or usage.get("cache_write_input_tokens") + or usage.get("cache_miss_tokens") + or 0 + ) + + if isinstance(usage, dict): + cache_skipped_unsupported_model = bool(usage.get("cache_skipped_unsupported_model", False)) + else: input_tokens = (getattr(usage, "input_tokens", getattr(usage, "prompt_tokens", 0))) or 0 output_tokens = (getattr(usage, "output_tokens", getattr(usage, "completion_tokens", 0))) or 0 cached_tokens = getattr(usage, "cached_tokens", None) or 0 + cache_creation_tokens = ( + getattr(usage, "cache_creation_tokens", None) + or getattr(usage, "cache_creation_input_tokens", None) + or getattr(usage, "cache_write_input_tokens", None) + or getattr(usage, "cache_miss_tokens", None) + or 0 + ) + cache_skipped_unsupported_model = bool(getattr(usage, "cache_skipped_unsupported_model", False)) cached_tokens = cached_tokens or 0 + cache_creation_tokens = cache_creation_tokens or 0 try: # Prefer cost directly from the result @@ -1472,7 +1522,22 @@ def _execute_agent_run( "input_tokens": input_tokens, "output_tokens": output_tokens, "cached_tokens": cached_tokens, - "cost": cost + "cost": cost, + "usage_snapshot": json.dumps({ + "schema_version": 1, + "input_tokens": input_tokens, + "output_tokens": output_tokens, + "cache_read_tokens": cached_tokens if usage else None, + "cache_creation_tokens": cache_creation_tokens if usage else None, + "cache_miss_tokens": cache_creation_tokens if usage else None, + "cache_skipped_unsupported_model": cache_skipped_unsupported_model, + "total_tokens": total_tokens, + "completeness": "provider_reported" if usage else "estimated", + "segment_tokens": segment_tokens, + "prefix_breakpoints": prefix_breakpoints, + }), + "cost_source": "provider_reported" if getattr(result, "cost", None) is not None else "unknown", + "cost_calculation_status": "calculated" if cost is not None else "unavailable", }) agent_message = conv_manager.add_message(conversation, "agent", final_output, resolved_provider, resolved_model, agent_name, run_doc.name) @@ -2230,8 +2295,16 @@ async def run_agent_stream( else: enhanced_prompt = base_prompt + from huf.ai.context_segments import compute_segment_tokens, compute_prefix_breakpoints + segment_tokens = compute_segment_tokens( + agent_doc, agent, resolved_model, resolved_provider, history, knowledge_context, prompt + ) + prefix_breakpoints = compute_prefix_breakpoints( + agent_doc, agent, resolved_model, resolved_provider, history + ) + context["conversation_history"] = history - + # Stream from provider full_response = "" try: @@ -2293,6 +2366,8 @@ async def run_agent_stream( input_tokens = 0 output_tokens = 0 cached_tokens = 0 + cache_creation_tokens = 0 + cache_skipped_unsupported_model = False total_tokens = 0 if usage: @@ -2308,19 +2383,59 @@ async def run_agent_stream( if details: if isinstance(details, dict): cached_tokens = details.get("cached_tokens") or details.get("cache_hit_tokens") or 0 + cache_creation_tokens = ( + details.get("cache_creation_input_tokens") + or details.get("cache_write_tokens") + or details.get("cache_creation_tokens") + or 0 + ) else: cached_tokens = getattr(details, "cached_tokens", None) or getattr(details, "cache_hit_tokens", None) or 0 + cache_creation_tokens = ( + getattr(details, "cache_creation_input_tokens", None) + or getattr(details, "cache_write_tokens", None) + or getattr(details, "cache_creation_tokens", None) + or 0 + ) elif isinstance(usage, dict): cached_tokens = usage.get("cached_tokens") or usage.get("cache_hit_tokens") or 0 - + cache_creation_tokens = ( + usage.get("cache_creation_tokens") + or usage.get("cache_creation_input_tokens") + or usage.get("cache_write_input_tokens") + or usage.get("cache_miss_tokens") + or 0 + ) + + if not cache_creation_tokens and isinstance(usage, dict): + cache_creation_tokens = ( + usage.get("cache_creation_tokens") + or usage.get("cache_creation_input_tokens") + or usage.get("cache_write_input_tokens") + or usage.get("cache_miss_tokens") + or 0 + ) + + if isinstance(usage, dict): + cache_skipped_unsupported_model = bool(usage.get("cache_skipped_unsupported_model", False)) + total_tokens = getattr(usage, "total_tokens", (input_tokens + output_tokens)) else: input_tokens = (getattr(usage, "prompt_tokens", getattr(usage, "input_tokens", 0))) or 0 output_tokens = (getattr(usage, "completion_tokens", getattr(usage, "output_tokens", 0))) or 0 cached_tokens = getattr(usage, "cached_tokens", None) or 0 + cache_creation_tokens = ( + getattr(usage, "cache_creation_tokens", None) + or getattr(usage, "cache_creation_input_tokens", None) + or getattr(usage, "cache_write_input_tokens", None) + or getattr(usage, "cache_miss_tokens", None) + or 0 + ) + cache_skipped_unsupported_model = bool(getattr(usage, "cache_skipped_unsupported_model", False)) total_tokens = getattr(usage, "total_tokens", (input_tokens + output_tokens)) or (input_tokens + output_tokens) cached_tokens = cached_tokens or 0 + cache_creation_tokens = cache_creation_tokens or 0 if input_tokens == 0 or output_tokens == 0: try: @@ -2400,6 +2515,21 @@ async def run_agent_stream( "output_tokens": output_tokens, "cached_tokens": cached_tokens, "cost": cost, + "usage_snapshot": json.dumps({ + "schema_version": 1, + "input_tokens": input_tokens, + "output_tokens": output_tokens, + "cache_read_tokens": cached_tokens if usage else None, + "cache_creation_tokens": cache_creation_tokens if usage else None, + "cache_miss_tokens": cache_creation_tokens if usage else None, + "cache_skipped_unsupported_model": cache_skipped_unsupported_model, + "total_tokens": total_tokens, + "completeness": "provider_reported" if usage else "estimated", + "segment_tokens": segment_tokens, + "prefix_breakpoints": prefix_breakpoints, + }), + "cost_source": "provider_reported" if chunk.get("cost") is not None else "unknown", + "cost_calculation_status": "calculated" if cost is not None else "unavailable", "end_time": now_datetime() }, update_modified=True) safe_commit() diff --git a/huf/ai/agent_run_analytics.py b/huf/ai/agent_run_analytics.py new file mode 100644 index 00000000..c0144711 --- /dev/null +++ b/huf/ai/agent_run_analytics.py @@ -0,0 +1,116 @@ +"""Scheduled, idempotent rollups for Agent Run analytics. + +The browser and API only read rollups. Raw Agent Runs are grouped here in a +bounded correction window so terminal updates are reflected without exposing +or repeatedly aggregating raw executions at request time. +""" + +from __future__ import annotations + +from collections import defaultdict +from datetime import timedelta + +import frappe +from frappe.utils import add_to_date, get_datetime, now_datetime + + +ROLLUP_DOCTYPE = "Agent Run Analytics Rollup" +TERMINAL_STATUSES = ("Success", "Failed") +CORRECTION_WINDOW_HOURS = 26 + + +def _bucket_start(value, granularity: str): + value = get_datetime(value) + if granularity == "day": + return value.replace(hour=0, minute=0, second=0, microsecond=0) + return value.replace(minute=0, second=0, microsecond=0) + + +def _dimension_key(row: dict) -> str: + return "|".join(str(row.get(field) or "__none__") for field in ("agent", "provider", "model", "run_kind")) + + +def _affected_dimensions(since): + rows = frappe.db.get_all( + "Agent Run", + filters={"status": ["in", TERMINAL_STATUSES], "start_time": [">=", since]}, + fields=["start_time", "agent", "provider", "model", "run_kind"], + limit_page_length=0, + ) + affected = set() + for row in rows: + if not row.start_time: + continue + row = dict(row) + for granularity in ("hour", "day"): + affected.add((granularity, _bucket_start(row["start_time"], granularity), _dimension_key(row))) + return affected + + +def _recompute_rollup(granularity: str, bucket_start, dimension_key: str): + bucket_end = add_to_date(bucket_start, hours=1 if granularity == "hour" else 24) + dimensions = dimension_key.split("|") + fields = ("agent", "provider", "model", "run_kind") + filters = [ + ["status", "in", TERMINAL_STATUSES], + ["start_time", ">=", bucket_start], + ["start_time", "<", bucket_end], + ] + for field, value in zip(fields, dimensions): + filters.append([field, "is", "not set"] if value == "__none__" else [field, "=", value]) + + rows = frappe.db.get_all( + "Agent Run", + filters=filters, + fields=["status", "input_tokens", "output_tokens", "cached_tokens", "cost", "start_time", "end_time"], + limit_page_length=0, + ) + if not rows: + existing = frappe.db.exists(ROLLUP_DOCTYPE, {"granularity": granularity, "bucket_start": bucket_start, "dimension_key": dimension_key}) + if existing: + frappe.delete_doc(ROLLUP_DOCTYPE, existing, force=True, ignore_permissions=True) + return + + metrics = defaultdict(float) + metrics["run_count"] = len(rows) + for row in rows: + metrics["success_count"] += int(row.status == "Success") + metrics["failed_count"] += int(row.status == "Failed") + metrics["input_tokens"] += row.input_tokens or 0 + metrics["output_tokens"] += row.output_tokens or 0 + metrics["cached_tokens"] += row.cached_tokens or 0 + metrics["total_cost"] += row.cost or 0 + if row.start_time and row.end_time: + duration = (get_datetime(row.end_time) - get_datetime(row.start_time)).total_seconds() * 1000 + if duration >= 0: + metrics["duration_ms_sum"] += duration + metrics["duration_count"] += 1 + + existing = frappe.db.exists(ROLLUP_DOCTYPE, {"granularity": granularity, "bucket_start": bucket_start, "dimension_key": dimension_key}) + doc = frappe.get_doc(ROLLUP_DOCTYPE, existing) if existing else frappe.new_doc(ROLLUP_DOCTYPE) + doc.update({ + "granularity": granularity, + "bucket_start": bucket_start, + "dimension_key": dimension_key, + "agent": None if dimensions[0] == "__none__" else dimensions[0], + "provider": None if dimensions[1] == "__none__" else dimensions[1], + "model": None if dimensions[2] == "__none__" else dimensions[2], + "run_kind": None if dimensions[3] == "__none__" else dimensions[3], + **metrics, + "last_recomputed_at": now_datetime(), + }) + doc.flags.ignore_permissions = True + doc.save() if existing else doc.insert() + + +def refresh_rollups(): + """Recompute the recent mutable window; safe to retry and safe on empty sites.""" + if not frappe.db.exists("DocType", ROLLUP_DOCTYPE): + return + since = add_to_date(now_datetime(), hours=-CORRECTION_WINDOW_HOURS) + for granularity, bucket_start, dimension_key in _affected_dimensions(since): + try: + _recompute_rollup(granularity, bucket_start, dimension_key) + except Exception: + frappe.log_error(frappe.get_traceback(), "Agent Run analytics rollup refresh failed") + frappe.db.commit() diff --git a/huf/ai/agent_run_analytics_api.py b/huf/ai/agent_run_analytics_api.py new file mode 100644 index 00000000..1c1b9b9e --- /dev/null +++ b/huf/ai/agent_run_analytics_api.py @@ -0,0 +1,62 @@ +"""Read-only aggregate API for execution analytics.""" + +from __future__ import annotations + +import frappe +from frappe.utils import add_to_date, get_datetime, now_datetime + + +ROLLUP_DOCTYPE = "Agent Run Analytics Rollup" +MAX_WINDOW_DAYS = 93 + + +def _require_analytics_access(): + if "System Manager" not in frappe.get_roles(): + frappe.throw("Execution analytics requires System Manager access", frappe.PermissionError) + + +@frappe.whitelist(methods=["GET"]) +def get_execution_analytics(from_date: str | None = None, to_date: str | None = None, granularity: str = "hour"): + """Return precomputed analytics only; never query or group Agent Run rows.""" + _require_analytics_access() + if granularity not in {"hour", "day"}: + frappe.throw("granularity must be 'hour' or 'day'") + end = get_datetime(to_date) if to_date else now_datetime() + start = get_datetime(from_date) if from_date else add_to_date(end, days=-7) + if start > end or (end - start).days > MAX_WINDOW_DAYS: + frappe.throw(f"Date range must be between zero and {MAX_WINDOW_DAYS} days") + if not frappe.db.exists("DocType", ROLLUP_DOCTYPE): + return {"summary": {}, "series": [], "breakdowns": [], "metadata": {"freshness": None, "granularity": granularity}} + + rows = frappe.db.get_all( + ROLLUP_DOCTYPE, + filters={"granularity": granularity, "bucket_start": ["between", [start, end]]}, + fields=["bucket_start", "agent", "provider", "model", "run_kind", "run_count", "success_count", "failed_count", "input_tokens", "output_tokens", "cached_tokens", "total_cost", "duration_ms_sum", "duration_count", "last_recomputed_at"], + order_by="bucket_start asc", + limit_page_length=0, + ) + summary = {"run_count": 0, "success_count": 0, "failed_count": 0, "input_tokens": 0, "output_tokens": 0, "cached_tokens": 0, "total_cost": 0, "duration_ms_sum": 0, "duration_count": 0} + series_by_bucket, provider_by_name = {}, {} + freshness = None + for row in rows: + row = dict(row) + for key in summary: + summary[key] += row.get(key) or 0 + bucket = series_by_bucket.setdefault(row["bucket_start"], {"bucket_start": row["bucket_start"], **{key: 0 for key in summary}}) + for key in summary: + bucket[key] += row.get(key) or 0 + provider = row.get("provider") or "Unknown" + breakdown = provider_by_name.setdefault(provider, {"dimension": provider, **{key: 0 for key in summary}}) + for key in summary: + breakdown[key] += row.get(key) or 0 + if row.get("last_recomputed_at") and (freshness is None or row["last_recomputed_at"] > freshness): + freshness = row["last_recomputed_at"] + summary["success_rate"] = (summary["success_count"] / summary["run_count"] * 100) if summary["run_count"] else None + summary["average_duration_ms"] = (summary["duration_ms_sum"] / summary["duration_count"]) if summary["duration_count"] else None + summary["cache_ratio"] = (summary["cached_tokens"] / summary["input_tokens"] * 100) if summary["input_tokens"] else None + return { + "summary": summary, + "series": list(series_by_bucket.values()), + "breakdowns": sorted(provider_by_name.values(), key=lambda item: item["run_count"], reverse=True)[:10], + "metadata": {"granularity": granularity, "from": start, "to": end, "freshness": freshness, "source": "scheduled_rollup"}, + } diff --git a/huf/ai/agent_run_context_api.py b/huf/ai/agent_run_context_api.py new file mode 100644 index 00000000..e09d5c43 --- /dev/null +++ b/huf/ai/agent_run_context_api.py @@ -0,0 +1,52 @@ +"""Per-run context composition and cache-economics API. + +Serves the segment_tokens breakdown and the five derived cache metrics for +a single Agent Run, for the ContextBar component (chat header, run detail). +Reads only the run's own usage_snapshot plus the prior run of the same +agent for prefix-stability comparison; never aggregates raw runs. +""" + +from __future__ import annotations + +import json + +import frappe + +from huf.ai.cache_metrics import compute_run_metrics + +DEFAULT_CONTEXT_WINDOW = 200000 + + +@frappe.whitelist(methods=["GET"]) +def get_run_context_metrics(run_name: str): + run_doc = frappe.get_doc("Agent Run", run_name) + run_doc.check_permission("read") + + snapshot_raw = run_doc.get("usage_snapshot") + snapshot = json.loads(snapshot_raw) if snapshot_raw else {} + + previous_run_doc = None + previous_name = frappe.db.get_value( + "Agent Run", + { + "agent": run_doc.agent, + "start_time": ["<", run_doc.start_time], + }, + "name", + order_by="start_time desc", + ) + if previous_name: + previous_run_doc = frappe.get_doc("Agent Run", previous_name) + + metrics = compute_run_metrics(run_doc, previous_run_doc) + + # AI Model does not carry a context-window field today; DEFAULT_CONTEXT_WINDOW + # is a placeholder until model metadata exposes one (see PHASE2_VisualCache.md). + return { + "segment_tokens": snapshot.get("segment_tokens"), + "total_tokens": snapshot.get("total_tokens"), + "context_window": DEFAULT_CONTEXT_WINDOW, + "prefix_breakpoints": snapshot.get("prefix_breakpoints") or [], + "cache_skipped_unsupported_model": snapshot.get("cache_skipped_unsupported_model"), + "metrics": metrics, + } diff --git a/huf/ai/cache_metrics.py b/huf/ai/cache_metrics.py new file mode 100644 index 00000000..49d52ab1 --- /dev/null +++ b/huf/ai/cache_metrics.py @@ -0,0 +1,91 @@ +# Copyright (c) 2026, Tridz Technologies Pvt Ltd +# For license information, please see license.txt + +""" +Five derived context/cache metrics, computed once server-side and reused by +every surface (chat header, run detail, agent/fleet views). Composition +(segment_tokens) answers "what fills the window"; these metrics answer +"is caching paying off" — never recomputed client-side. + +Approximation note: effective_input_multiplier and counterfactual_savings +use fixed token-cost multipliers (cache read ~=0.1x, cache write ~=1.25x, +uncached 1x) rather than a per-model dollar lookup. Actual provider rates +vary; treat these as directional, not invoiced figures. Wiring in +per-model rates from cost_calculator.get_model_pricing() is deferred — +see PHASE2_VisualCache.md. +""" + +CACHE_READ_MULTIPLIER = 0.1 +CACHE_WRITE_MULTIPLIER = 1.25 +UNCACHED_MULTIPLIER = 1.0 + + +def _load_usage_snapshot(run_doc): + raw = run_doc.get("usage_snapshot") if hasattr(run_doc, "get") else getattr(run_doc, "usage_snapshot", None) + if not raw: + return {} + if isinstance(raw, dict): + return raw + import json + + try: + return json.loads(raw) + except (TypeError, ValueError): + return {} + + +def _prefix_signature(snapshot): + breakpoints = snapshot.get("prefix_breakpoints") or [] + if not breakpoints: + return None + return tuple(sorted((bp.get("marker"), bp.get("prefix_hash")) for bp in breakpoints if bp.get("prefix_hash"))) + + +def compute_run_metrics(run_doc, previous_run_doc=None): + """Compute the five metrics for one run. Fields are `None`, never 0, + when the inputs needed to calculate them are unavailable.""" + snapshot = _load_usage_snapshot(run_doc) + + input_tokens = snapshot.get("input_tokens") + cache_read = snapshot.get("cache_read_tokens") + cache_write = snapshot.get("cache_creation_tokens") + cost = run_doc.get("cost") if hasattr(run_doc, "get") else getattr(run_doc, "cost", None) + + cache_read_share = None + effective_input_multiplier = None + if input_tokens: + if cache_read is not None: + cache_read_share = cache_read / input_tokens + if cache_read is not None and cache_write is not None: + uncached = max(input_tokens - cache_read - cache_write, 0) + effective_input_multiplier = ( + cache_read * CACHE_READ_MULTIPLIER + + cache_write * CACHE_WRITE_MULTIPLIER + + uncached * UNCACHED_MULTIPLIER + ) / input_tokens + + counterfactual_savings = None + if cost is not None and effective_input_multiplier is not None and effective_input_multiplier > 0: + # cost scales roughly with the multiplier; back out the no-cache cost. + counterfactual_savings = round(cost * (1 / effective_input_multiplier - 1) * effective_input_multiplier, 6) + + prefix_stability = "unavailable" + this_signature = _prefix_signature(snapshot) + if previous_run_doc is not None: + previous_signature = _prefix_signature(_load_usage_snapshot(previous_run_doc)) + if this_signature is None or previous_signature is None: + prefix_stability = "unknown" + elif this_signature == previous_signature: + prefix_stability = "stable" + else: + prefix_stability = "changed" + elif this_signature is not None: + prefix_stability = "unknown" + + return { + "cache_read_share": cache_read_share, + "effective_input_multiplier": effective_input_multiplier, + "wasted_writes_tokens": None, # needs read-after-write tracking; not captured yet + "prefix_stability": prefix_stability, + "counterfactual_savings": counterfactual_savings, + } diff --git a/huf/ai/context_segments.py b/huf/ai/context_segments.py new file mode 100644 index 00000000..c519471d --- /dev/null +++ b/huf/ai/context_segments.py @@ -0,0 +1,112 @@ +# Copyright (c) 2026, Tridz Technologies Pvt Ltd +# For license information, please see license.txt + +""" +Per-run context composition (segment_tokens) and cache prefix fingerprints +(prefix_breakpoints) for context/cache observability. + +Token-counts each piece of a run's assembled prompt at the point where the +pieces still exist as separate objects. Knowledge context in particular is +fused into the user message string before it reaches the provider call and +cannot be recovered afterwards, so counting has to happen here, before that +fusion, not downstream in providers/litellm.py. + +Breakpoint hashing mirrors the same three cache-control gates evaluated in +providers/litellm.py (static prefix is a runtime/context option, not an +Agent setting, so it is not reproduced here — see AGENTS.md notes on +prompt_cache_options). Per-breakpoint cache-hit attribution is deliberately +not attempted: providers report one aggregate cache_read count per run, not +per-block hits, and a wrong per-block guess is worse than an honest gap. +""" + +import hashlib + +import frappe +from litellm import token_counter + +from huf.ai.prompt_cache_capabilities import model_supports_prompt_caching +from huf.ai.providers.litellm import _normalize_model_name +from huf.ai.tool_serializer import serialize_tools + + +def _count(pricing_model, text): + if not text: + return 0 + try: + return token_counter(model=pricing_model, text=text) + except Exception: + return None + + +def _hash_prefix(text): + if not text: + return None + return hashlib.sha256(text.encode("utf-8", errors="ignore")).hexdigest()[:16] + + +def compute_segment_tokens(agent_doc, agent, resolved_model, resolved_provider, history, knowledge_context, prompt): + """Best-effort token count per composition segment. + + Returns a dict with keys system/tools/knowledge/history/message. A + segment is `None` (not 0) when it could not be counted — callers must + not treat a missing segment as zero-cost. + """ + pricing_model = _normalize_model_name(resolved_model, resolved_provider) + + system_text = "\n".join(filter(None, [getattr(agent, "instructions", None)])) + + try: + tools_schema = serialize_tools(getattr(agent, "tools", None) or []) + tools_text = frappe.as_json(tools_schema) if tools_schema else "" + except Exception: + tools_text = None + + knowledge_text = (knowledge_context or {}).get("context_text") if knowledge_context else "" + + try: + history_text = "\n".join( + str(item.get("content", "")) for item in (history or []) if isinstance(item, dict) + ) + except Exception: + history_text = None + + return { + "system": _count(pricing_model, system_text), + "tools": _count(pricing_model, tools_text) if tools_text is not None else None, + "knowledge": _count(pricing_model, knowledge_text) if knowledge_text is not None else None, + "history": _count(pricing_model, history_text) if history_text is not None else None, + "message": _count(pricing_model, prompt), + } + + +def compute_prefix_breakpoints(agent_doc, agent, resolved_model, resolved_provider, history): + """Fingerprint the cache-control breakpoints this run's settings would gate. + + Mirrors the enable_prompt_caching / cache_system_message / + cache_conversation_history gates in providers/litellm.py without + threading state through the provider call. One entry per active + breakpoint; empty list when caching is off or unsupported. + """ + if not agent_doc or not agent_doc.get("enable_prompt_caching"): + return [] + + try: + if not model_supports_prompt_caching(resolved_model, resolved_provider): + return [] + except Exception: + return [] + + breakpoints = [] + + if agent_doc.get("cache_system_message") and getattr(agent, "instructions", None): + prefix_hash = _hash_prefix(agent.instructions) + if prefix_hash: + breakpoints.append({"marker": "instructions", "prefix_hash": prefix_hash}) + + if agent_doc.get("cache_conversation_history") and history: + last_content = history[-1].get("content") if isinstance(history[-1], dict) else None + prefix_hash = _hash_prefix(last_content if isinstance(last_content, str) else None) + if prefix_hash: + breakpoints.append({"marker": "history", "prefix_hash": prefix_hash}) + + return breakpoints diff --git a/huf/ai/providers/litellm.py b/huf/ai/providers/litellm.py index 8106eeaf..6912cf3f 100644 --- a/huf/ai/providers/litellm.py +++ b/huf/ai/providers/litellm.py @@ -106,6 +106,42 @@ def _build_text_content(text: str, provider_name: str, cache_enabled: bool, cach return [{"type": "text", "text": text}] +def _format_conversation_history( + conversation_history: list, + provider_name: str, + cache_enabled: bool, + cache_control_type: str, +) -> list: + """Format conversation history messages, applying cache_control to the history prefix breakpoint if enabled.""" + if not conversation_history: + return [] + + formatted = [dict(msg) for msg in conversation_history] + if not cache_enabled: + return formatted + + last_msg = dict(formatted[-1]) + content = last_msg.get("content") + + if isinstance(content, str): + last_msg["content"] = _build_text_content(content, provider_name, True, cache_control_type) + elif isinstance(content, list) and len(content) > 0: + content_copy = [dict(b) if isinstance(b, dict) else b for b in content] + last_block = content_copy[-1] + if isinstance(last_block, dict): + last_block_copy = dict(last_block) + if provider_name == "anthropic": + last_block_copy["cache_control"] = {"type": cache_control_type} + content_copy[-1] = last_block_copy + last_msg["content"] = content_copy + elif content is None or content == "": + if provider_name == "anthropic": + last_msg["content"] = [{"type": "text", "text": "", "cache_control": {"type": cache_control_type}}] + + formatted[-1] = last_msg + return formatted + + def _file_dict_to_data_image_url(file_dict: dict) -> dict | None: """Embed a local Frappe file as a base64 data URI for multimodal LLM calls. @@ -374,12 +410,16 @@ async def run(agent, enhanced_prompt, provider, model, context=None): # Check if model supports prompt caching model_supports_caching = False + cache_skipped_unsupported_model = False if enable_prompt_caching: try: model_supports_caching = model_supports_prompt_caching(model, provider) + if not model_supports_caching: + cache_skipped_unsupported_model = True except Exception: # Prompt-caching check failed; disable caching and log for investigation. model_supports_caching = False + cache_skipped_unsupported_model = True frappe.log_error( f"Failed to check prompt caching support for model {normalized_model}", "LiteLLM Prompt Caching" @@ -416,7 +456,17 @@ async def run(agent, enhanced_prompt, provider, model, context=None): # Insert Conversation History if available if context and context.get("conversation_history"): - messages.extend(context["conversation_history"]) + history_cache_enabled = ( + enable_prompt_caching and model_supports_caching and cache_conversation_history + ) + messages.extend( + _format_conversation_history( + context["conversation_history"], + provider_name, + history_cache_enabled, + cache_control_type, + ) + ) # Add user message with cache_control if conversation history caching is enabled cache_dynamic_content = cache_conversation_history @@ -441,7 +491,14 @@ async def run(agent, enhanced_prompt, provider, model, context=None): if getattr(agent, "tools", None): tools = serialize_tools(agent.tools) - total_usage = {"input_tokens": 0, "output_tokens": 0, "cached_tokens": 0} + total_usage = { + "input_tokens": 0, + "output_tokens": 0, + "cached_tokens": 0, + "cache_creation_tokens": 0, + "cache_miss_tokens": 0, + "cache_skipped_unsupported_model": cache_skipped_unsupported_model, + } total_cost = 0.0 all_new_items = [] @@ -626,14 +683,52 @@ async def run(agent, enhanced_prompt, provider, model, context=None): total_usage["input_tokens"] += (getattr(usage, "prompt_tokens", 0) or 0) total_usage["output_tokens"] += (getattr(usage, "completion_tokens", 0) or 0) - # Track cached tokens if available - if enable_prompt_caching and hasattr(usage, "prompt_tokens_details"): + # Track cached tokens and cache creation/write tokens if available + round_cached = 0 + round_creation = 0 + + if hasattr(usage, "prompt_tokens_details") and usage.prompt_tokens_details: details = usage.prompt_tokens_details - if details: - if isinstance(details, dict): - total_usage["cached_tokens"] += (details.get("cached_tokens") or details.get("cache_hit_tokens") or 0) - else: - total_usage["cached_tokens"] += (getattr(details, "cached_tokens", None) or getattr(details, "cache_hit_tokens", None) or 0) + if isinstance(details, dict): + round_cached = details.get("cached_tokens") or details.get("cache_hit_tokens") or 0 + round_creation = ( + details.get("cache_creation_input_tokens") + or details.get("cache_write_tokens") + or details.get("cache_creation_tokens") + or 0 + ) + else: + round_cached = ( + getattr(details, "cached_tokens", None) + or getattr(details, "cache_hit_tokens", None) + or 0 + ) + round_creation = ( + getattr(details, "cache_creation_input_tokens", None) + or getattr(details, "cache_write_tokens", None) + or getattr(details, "cache_creation_tokens", None) + or 0 + ) + elif isinstance(usage, dict): + round_cached = usage.get("cached_tokens") or usage.get("cache_hit_tokens") or 0 + round_creation = ( + usage.get("cache_creation_input_tokens") + or usage.get("cache_write_input_tokens") + or usage.get("cache_creation_tokens") + or usage.get("cache_miss_tokens") + or 0 + ) + + if not round_creation: + round_creation = ( + getattr(usage, "cache_creation_input_tokens", None) + or getattr(usage, "cache_write_input_tokens", None) + or 0 + ) + + total_usage["cached_tokens"] += (round_cached or 0) + total_usage["cache_creation_tokens"] += (round_creation or 0) + total_usage["cache_miss_tokens"] += (round_creation or 0) assistant_message = { "role": "assistant", @@ -838,12 +933,16 @@ async def run_stream(agent, enhanced_prompt, provider, model, context=None): # Check if model supports prompt caching model_supports_caching = False + cache_skipped_unsupported_model = False if enable_prompt_caching: try: model_supports_caching = model_supports_prompt_caching(model, provider) + if not model_supports_caching: + cache_skipped_unsupported_model = True except Exception: # Prompt-caching check failed; disable caching and log for investigation. model_supports_caching = False + cache_skipped_unsupported_model = True frappe.log_error( message=f"Failed to check prompt caching support for model {normalized_model}", title="LiteLLM Prompt Caching" @@ -880,7 +979,17 @@ async def run_stream(agent, enhanced_prompt, provider, model, context=None): # Insert Conversation History if available if context and context.get("conversation_history"): - messages.extend(context["conversation_history"]) + history_cache_enabled = ( + enable_prompt_caching and model_supports_caching and cache_conversation_history + ) + messages.extend( + _format_conversation_history( + context["conversation_history"], + provider_name, + history_cache_enabled, + cache_control_type, + ) + ) cache_dynamic_content = cache_conversation_history if isinstance(cache_dynamic_content_override, bool): @@ -1241,16 +1350,42 @@ async def run_stream(agent, enhanced_prompt, provider, model, context=None): elif stream_usage and hasattr(stream_usage, "model_dump"): stream_usage = stream_usage.model_dump() + if not stream_usage or not isinstance(stream_usage, dict): + stream_usage = {} + + # Extract cache creation tokens and cache skipped flag for stream_usage + s_cached = 0 + s_creation = 0 + s_details = stream_usage.get("prompt_tokens_details") or {} + if isinstance(s_details, dict): + s_cached = int(s_details.get("cached_tokens", 0) or s_details.get("cache_hit_tokens", 0) or 0) + s_creation = int( + s_details.get("cache_creation_input_tokens", 0) + or s_details.get("cache_write_tokens", 0) + or s_details.get("cache_creation_tokens", 0) + or 0 + ) + + if not s_creation: + s_creation = int( + stream_usage.get("cache_creation_input_tokens", 0) + or stream_usage.get("cache_write_input_tokens", 0) + or stream_usage.get("cache_creation_tokens", 0) + or stream_usage.get("cache_miss_tokens", 0) + or 0 + ) + + stream_usage["cached_tokens"] = s_cached + stream_usage["cache_creation_tokens"] = s_creation + stream_usage["cache_miss_tokens"] = s_creation + stream_usage["cache_skipped_unsupported_model"] = cache_skipped_unsupported_model + # Calculate cost from streaming usage stream_cost = 0.0 if stream_usage and isinstance(stream_usage, dict): try: s_input = int(stream_usage.get("prompt_tokens", 0) or 0) s_output = int(stream_usage.get("completion_tokens", 0) or 0) - s_cached = 0 - s_details = stream_usage.get("prompt_tokens_details") or {} - if isinstance(s_details, dict): - s_cached = int(s_details.get("cached_tokens", 0) or s_details.get("cache_hit_tokens", 0) or 0) stream_cost, _src = calculate_cost( model_name=model, input_tokens=s_input, diff --git a/huf/ai/tests/test_agent_run_analytics.py b/huf/ai/tests/test_agent_run_analytics.py new file mode 100644 index 00000000..3a2a5dd5 --- /dev/null +++ b/huf/ai/tests/test_agent_run_analytics.py @@ -0,0 +1,18 @@ +from datetime import datetime + +from frappe.tests import UnitTestCase + +from huf.ai.agent_run_analytics import _bucket_start, _dimension_key + + +class TestAgentRunAnalytics(UnitTestCase): + def test_hour_and_day_buckets_are_deterministic(self): + value = datetime(2026, 7, 26, 14, 37, 9) + self.assertEqual(_bucket_start(value, "hour"), datetime(2026, 7, 26, 14, 0)) + self.assertEqual(_bucket_start(value, "day"), datetime(2026, 7, 26, 0, 0)) + + def test_dimension_key_preserves_missing_dimensions(self): + self.assertEqual( + _dimension_key({"agent": "Support", "provider": None, "model": "gemini", "run_kind": None}), + "Support|__none__|gemini|__none__", + ) diff --git a/huf/ai/tests/test_cache_metrics.py b/huf/ai/tests/test_cache_metrics.py new file mode 100644 index 00000000..98571d5d --- /dev/null +++ b/huf/ai/tests/test_cache_metrics.py @@ -0,0 +1,37 @@ +from frappe.tests import UnitTestCase + +from huf.ai.cache_metrics import compute_run_metrics + + +class TestCacheMetrics(UnitTestCase): + def test_metrics_are_null_when_usage_snapshot_missing(self): + run = {"usage_snapshot": None, "cost": None} + metrics = compute_run_metrics(run) + self.assertIsNone(metrics["cache_read_share"]) + self.assertIsNone(metrics["effective_input_multiplier"]) + self.assertIsNone(metrics["wasted_writes_tokens"]) + self.assertEqual(metrics["prefix_stability"], "unavailable") + + def test_cache_read_share_and_multiplier(self): + run = { + "usage_snapshot": '{"input_tokens": 1000, "cache_read_tokens": 600, "cache_creation_tokens": 0}', + "cost": 0.002, + } + metrics = compute_run_metrics(run) + self.assertAlmostEqual(metrics["cache_read_share"], 0.6) + # 600 * 0.1 + 0 * 1.25 + 400 * 1.0 = 460 -> 460/1000 = 0.46 + self.assertAlmostEqual(metrics["effective_input_multiplier"], 0.46) + + def test_prefix_stability_compares_against_previous_run(self): + current = { + "usage_snapshot": '{"prefix_breakpoints": [{"marker": "instructions", "prefix_hash": "abc123"}]}', + } + same = { + "usage_snapshot": '{"prefix_breakpoints": [{"marker": "instructions", "prefix_hash": "abc123"}]}', + } + changed = { + "usage_snapshot": '{"prefix_breakpoints": [{"marker": "instructions", "prefix_hash": "zzz999"}]}', + } + self.assertEqual(compute_run_metrics(current, same)["prefix_stability"], "stable") + self.assertEqual(compute_run_metrics(current, changed)["prefix_stability"], "changed") + self.assertEqual(compute_run_metrics(current, {"usage_snapshot": None})["prefix_stability"], "unknown") diff --git a/huf/hooks.py b/huf/hooks.py index 34bdca27..f9d1bae8 100644 --- a/huf/hooks.py +++ b/huf/hooks.py @@ -235,7 +235,10 @@ "cron": { "*/1 * * * *": [ "huf.ai.orchestration.scheduler.process_orchestrations", - "huf.ai.agent_integration.recover_stalled_agent_runs" + "huf.ai.agent_integration.recover_stalled_agent_runs", + ], + "*/5 * * * *": [ + "huf.ai.agent_run_analytics.refresh_rollups", ] }, "hourly": [ diff --git a/huf/huf/doctype/agent_run/agent_run.json b/huf/huf/doctype/agent_run/agent_run.json index 202b8701..1e2733e3 100644 --- a/huf/huf/doctype/agent_run/agent_run.json +++ b/huf/huf/doctype/agent_run/agent_run.json @@ -20,7 +20,10 @@ "provider", "start_time", "input_tokens", + "usage_snapshot", "cost", + "cost_source", + "cost_calculation_status", "call_recording", "prompt_template", "column_break_qy3x", @@ -152,6 +155,13 @@ "label": "Input Tokens", "read_only": 1 }, + { + "description": "Normalized telemetry snapshot. Null values mean the provider did not report that metric.", + "fieldname": "usage_snapshot", + "fieldtype": "JSON", + "label": "Usage Snapshot", + "read_only": 1 + }, { "description": "Number of tokens in the model response.\n\n", "fieldname": "output_tokens", @@ -173,6 +183,20 @@ "label": "Cost", "read_only": 1 }, + { + "fieldname": "cost_source", + "fieldtype": "Select", + "label": "Cost Source", + "options": "unknown\nprovider_reported\ncustom\nlitellm", + "read_only": 1 + }, + { + "fieldname": "cost_calculation_status", + "fieldtype": "Select", + "label": "Cost Calculation Status", + "options": "calculated\nunavailable\nfailed", + "read_only": 1 + }, { "fieldname": "column_break_qy3x", "fieldtype": "Column Break" diff --git a/huf/huf/doctype/agent_run_analytics_rollup/__init__.py b/huf/huf/doctype/agent_run_analytics_rollup/__init__.py new file mode 100644 index 00000000..7df4dc2d --- /dev/null +++ b/huf/huf/doctype/agent_run_analytics_rollup/__init__.py @@ -0,0 +1 @@ +"""Agent Run Analytics Rollup DocType.""" diff --git a/huf/huf/doctype/agent_run_analytics_rollup/agent_run_analytics_rollup.json b/huf/huf/doctype/agent_run_analytics_rollup/agent_run_analytics_rollup.json new file mode 100644 index 00000000..4e5a2a0a --- /dev/null +++ b/huf/huf/doctype/agent_run_analytics_rollup/agent_run_analytics_rollup.json @@ -0,0 +1,38 @@ +{ + "actions": [], + "autoname": "hash", + "creation": "2026-07-26 00:00:00.000000", + "doctype": "DocType", + "engine": "InnoDB", + "field_order": ["bucket_start", "granularity", "dimension_key", "agent", "provider", "model", "run_kind", "run_count", "success_count", "failed_count", "input_tokens", "output_tokens", "cached_tokens", "total_cost", "duration_ms_sum", "duration_count", "last_recomputed_at"], + "fields": [ + {"fieldname":"bucket_start","fieldtype":"Datetime","in_list_view":1,"label":"Bucket Start","read_only":1}, + {"fieldname":"granularity","fieldtype":"Select","in_list_view":1,"label":"Granularity","options":"hour\nday","read_only":1}, + {"fieldname":"dimension_key","fieldtype":"Data","label":"Dimension Key","read_only":1}, + {"fieldname":"agent","fieldtype":"Link","in_list_view":1,"label":"Agent","options":"Agent","read_only":1}, + {"fieldname":"provider","fieldtype":"Link","in_list_view":1,"label":"Provider","options":"AI Provider","read_only":1}, + {"fieldname":"model","fieldtype":"Link","in_list_view":1,"label":"Model","options":"AI Model","read_only":1}, + {"fieldname":"run_kind","fieldtype":"Data","label":"Run Kind","read_only":1}, + {"fieldname":"run_count","fieldtype":"Int","label":"Run Count","read_only":1}, + {"fieldname":"success_count","fieldtype":"Int","label":"Success Count","read_only":1}, + {"fieldname":"failed_count","fieldtype":"Int","label":"Failed Count","read_only":1}, + {"fieldname":"input_tokens","fieldtype":"Int","label":"Input Tokens","read_only":1}, + {"fieldname":"output_tokens","fieldtype":"Int","label":"Output Tokens","read_only":1}, + {"fieldname":"cached_tokens","fieldtype":"Int","label":"Cached Tokens","read_only":1}, + {"fieldname":"total_cost","fieldtype":"Float","label":"Total Cost","read_only":1}, + {"fieldname":"duration_ms_sum","fieldtype":"Float","label":"Duration Sum (ms)","read_only":1}, + {"fieldname":"duration_count","fieldtype":"Int","label":"Duration Count","read_only":1}, + {"fieldname":"last_recomputed_at","fieldtype":"Datetime","label":"Last Recomputed At","read_only":1} + ], + "index_web_pages_for_search": 0, + "links": [], + "modified": "2026-07-26 00:00:00.000000", + "modified_by": "Administrator", + "module": "Huf", + "name": "Agent Run Analytics Rollup", + "owner": "Administrator", + "permissions": [{"read":1,"report":1,"role":"System Manager"}], + "sort_field": "bucket_start", + "sort_order": "DESC", + "states": [] +} diff --git a/huf/huf/doctype/agent_run_analytics_rollup/agent_run_analytics_rollup.py b/huf/huf/doctype/agent_run_analytics_rollup/agent_run_analytics_rollup.py new file mode 100644 index 00000000..b058e9e8 --- /dev/null +++ b/huf/huf/doctype/agent_run_analytics_rollup/agent_run_analytics_rollup.py @@ -0,0 +1,5 @@ +from frappe.model.document import Document + + +class AgentRunAnalyticsRollup(Document): + pass diff --git a/huf/patches.txt b/huf/patches.txt index c37c2034..7d6826bb 100644 --- a/huf/patches.txt +++ b/huf/patches.txt @@ -7,9 +7,10 @@ huf.patches.v1.migrate_provider_brand # Patches added in this section will be executed after doctypes are migrated huf.patches.v1.add_agent_run_sequence_index +huf.patches.v1.add_agent_run_analytics_indexes huf.patches.add_tool_types huf.patches.setup_desktop_icon_as_workspace huf.patches.v1.update_image_tool huf.patches.v1.update_agent_background_color huf.patches.v1.repair_tool_call_messages -huf.patches.v1.fix_agent_message_status \ No newline at end of file +huf.patches.v1.fix_agent_message_status diff --git a/huf/patches/v1/add_agent_run_analytics_indexes.py b/huf/patches/v1/add_agent_run_analytics_indexes.py new file mode 100644 index 00000000..db501bc7 --- /dev/null +++ b/huf/patches/v1/add_agent_run_analytics_indexes.py @@ -0,0 +1,35 @@ +"""Indexes supporting the scheduled Agent Run analytics rollup.""" + +import frappe + + +def _add_index(doctype, fields, name, unique=False): + try: + if unique: + frappe.db.add_unique(doctype, fields, name) + else: + frappe.db.add_index(doctype, fields, name) + except Exception as error: + # A previously deployed/manual index must not make migrate fail. + if "Duplicate key name" not in str(error): + raise + + +def execute(): + if frappe.db.has_column("Agent Run", "start_time"): + _add_index("Agent Run", ["start_time", "status"], "idx_agent_run_analytics_time_status") + _add_index("Agent Run", ["agent", "start_time"], "idx_agent_run_analytics_agent_time") + _add_index("Agent Run", ["provider", "model", "start_time"], "idx_agent_run_analytics_model_time") + + if frappe.db.has_column("Agent Run Analytics Rollup", "bucket_start"): + _add_index( + "Agent Run Analytics Rollup", + ["granularity", "bucket_start", "dimension_key"], + "idx_agent_run_rollup_unique", + unique=True, + ) + _add_index( + "Agent Run Analytics Rollup", + ["granularity", "bucket_start"], + "idx_agent_run_rollup_window", + )