fix: derive overall token totals from per-agent events instead of broken summary event

The LLM_Token_Usage_Summary event only captured MapHandler tokens because
save_handler doesn't have visibility into Workflow service steps (Summarize,
RAI, GapAnalysis). Fixed all queries that need total token counts to aggregate
from LLM_Agent_Token_Usage events grouped by process_id.

Queries fixed:
- Overall Token Usage Summary
- Token Usage Over Time
- Estimated Daily Cost
- Top 20 Token Consumers (join with Summary for file_name)
- Token Usage by File Type (join with Summary for mime_type)
- Token Usage Percentiles
- Daily Processing Volume

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: Prachig-Microsoft

infra/dashboards/token-usage-queries.kql

@@ -6,11 +6,13 @@
 
 // 1. Overall token usage summary (last 7 days)
 customEvents
-| where name == 'LLM_Token_Usage_Summary'
+| where name == 'LLM_Agent_Token_Usage'
 | where timestamp > ago(7d)
-| extend input_tokens = toint(customDimensions['total_input_tokens'])
-| extend output_tokens = toint(customDimensions['total_output_tokens'])
+| extend process_id = tostring(customDimensions['process_id'])
+| extend input_tokens = toint(customDimensions['input_tokens'])
+| extend output_tokens = toint(customDimensions['output_tokens'])
 | extend total_tokens = toint(customDimensions['total_tokens'])
+| summarize input_tokens=sum(input_tokens), output_tokens=sum(output_tokens), total_tokens=sum(total_tokens) by process_id
 | summarize
     TotalDocuments = count(),
     TotalInputTokens = sum(input_tokens),
@@ -36,10 +38,12 @@ customEvents
 
 // 3. Token usage over time (hourly)
 customEvents
-| where name == 'LLM_Token_Usage_Summary'
+| where name == 'LLM_Agent_Token_Usage'
 | where timestamp > ago(7d)
-| extend input_tokens = toint(customDimensions['total_input_tokens'])
-| extend output_tokens = toint(customDimensions['total_output_tokens'])
+| extend process_id = tostring(customDimensions['process_id'])
+| extend input_tokens = toint(customDimensions['input_tokens'])
+| extend output_tokens = toint(customDimensions['output_tokens'])
+| summarize input_tokens=sum(input_tokens), output_tokens=sum(output_tokens), timestamp=min(timestamp) by process_id
 | summarize InputTokens = sum(input_tokens), OutputTokens = sum(output_tokens) by bin(timestamp, 1h)
 | order by timestamp asc
 | render areachart
@@ -48,10 +52,12 @@ customEvents
 let input_price_per_million = 2.50;
 let output_price_per_million = 10.00;
 customEvents
-| where name == 'LLM_Token_Usage_Summary'
+| where name == 'LLM_Agent_Token_Usage'
 | where timestamp > ago(30d)
-| extend input_tokens = toint(customDimensions['total_input_tokens'])
-| extend output_tokens = toint(customDimensions['total_output_tokens'])
+| extend process_id = tostring(customDimensions['process_id'])
+| extend input_tokens = toint(customDimensions['input_tokens'])
+| extend output_tokens = toint(customDimensions['output_tokens'])
+| summarize input_tokens=sum(input_tokens), output_tokens=sum(output_tokens), timestamp=min(timestamp) by process_id
 | summarize TotalInput = sum(input_tokens), TotalOutput = sum(output_tokens) by bin(timestamp, 1d)
 | extend InputCost = round(TotalInput * input_price_per_million / 1000000.0, 4)
 | extend OutputCost = round(TotalOutput * output_price_per_million / 1000000.0, 4)
@@ -61,12 +67,20 @@ customEvents
 
 // 5. Top token consumers by document
 customEvents
-| where name == 'LLM_Token_Usage_Summary'
+| where name == 'LLM_Agent_Token_Usage'
 | where timestamp > ago(7d)
-| extend total_tokens = toint(customDimensions['total_tokens'])
 | extend process_id = tostring(customDimensions['process_id'])
-| extend file_name = tostring(customDimensions['file_name'])
-| summarize TotalTokens = sum(total_tokens) by process_id, file_name
+| extend total_tokens = toint(customDimensions['total_tokens'])
+| summarize TotalTokens = sum(total_tokens) by process_id
+| join kind=leftouter (
+    customEvents
+    | where name == 'LLM_Token_Usage_Summary'
+    | where timestamp > ago(7d)
+    | extend process_id = tostring(customDimensions['process_id'])
+    | extend file_name = tostring(customDimensions['file_name'])
+    | summarize file_name=take_any(file_name) by process_id
+) on process_id
+| project process_id, file_name, TotalTokens
 | order by TotalTokens desc
 | take 20
 
@@ -81,9 +95,11 @@ customEvents
 
 // 7. Token usage percentiles per document
 customEvents
-| where name == 'LLM_Token_Usage_Summary'
+| where name == 'LLM_Agent_Token_Usage'
 | where timestamp > ago(7d)
+| extend process_id = tostring(customDimensions['process_id'])
 | extend total_tokens = toint(customDimensions['total_tokens'])
+| summarize total_tokens=sum(total_tokens) by process_id
 | summarize
     p50 = percentile(total_tokens, 50),
     p90 = percentile(total_tokens, 90),
@@ -230,12 +246,21 @@ dependencies
 
 // 16. Token usage by file type (PDF, DOCX, image, etc.)
 customEvents
-| where name == 'LLM_Token_Usage_Summary'
+| where name == 'LLM_Agent_Token_Usage'
 | where timestamp > ago(7d)
+| extend process_id = tostring(customDimensions['process_id'])
+| extend input_tokens = toint(customDimensions['input_tokens'])
+| extend output_tokens = toint(customDimensions['output_tokens'])
 | extend total_tokens = toint(customDimensions['total_tokens'])
-| extend input_tokens = toint(customDimensions['total_input_tokens'])
-| extend output_tokens = toint(customDimensions['total_output_tokens'])
-| extend mime_type = tostring(customDimensions['file_mime_type'])
+| summarize input_tokens=sum(input_tokens), output_tokens=sum(output_tokens), total_tokens=sum(total_tokens) by process_id
+| join kind=leftouter (
+    customEvents
+    | where name == 'LLM_Token_Usage_Summary'
+    | where timestamp > ago(7d)
+    | extend process_id = tostring(customDimensions['process_id'])
+    | extend mime_type = tostring(customDimensions['file_mime_type'])
+    | summarize mime_type=take_any(mime_type) by process_id
+) on process_id
 | extend file_type = case(
     mime_type has "pdf", "PDF",
     mime_type has "image", "Image",
@@ -270,10 +295,11 @@ customEvents
 
 // 18. Daily processing volume with token costs
 customEvents
-| where name == 'LLM_Token_Usage_Summary'
+| where name == 'LLM_Agent_Token_Usage'
 | where timestamp > ago(30d)
+| extend process_id = tostring(customDimensions['process_id'])
 | extend total_tokens = toint(customDimensions['total_tokens'])
-| extend file_name = tostring(customDimensions['file_name'])
+| summarize total_tokens=sum(total_tokens), timestamp=min(timestamp) by process_id
 | summarize
     DocumentsProcessed = count(),
     TotalTokens = sum(total_tokens),

infra/dashboards/token-usage-workbook.json

@@ -59,7 +59,7 @@
       "type": 3,
       "content": {
         "version": "KqlItem/1.0",
-        "query": "customEvents\r\n| where name == 'LLM_Token_Usage_Summary'\r\n| where timestamp > ago(7d)\r\n| extend process_id = tostring(customDimensions['process_id'])\r\n| extend input_tokens = toint(customDimensions['total_input_tokens'])\r\n| extend output_tokens = toint(customDimensions['total_output_tokens'])\r\n| extend total_tokens = toint(customDimensions['total_tokens'])\r\n| summarize input_tokens=max(input_tokens), output_tokens=max(output_tokens), total_tokens=max(total_tokens) by process_id\r\n| summarize\r\n    TotalDocuments = count(),\r\n    TotalInputTokens = sum(input_tokens),\r\n    TotalOutputTokens = sum(output_tokens),\r\n    TotalTokens = sum(total_tokens),\r\n    AvgTokensPerDocument = round(avg(total_tokens), 0)",
+        "query": "customEvents\r\n| where name == 'LLM_Agent_Token_Usage'\r\n| where timestamp > ago(7d)\r\n| extend process_id = tostring(customDimensions['process_id'])\r\n| extend input_tokens = toint(customDimensions['input_tokens'])\r\n| extend output_tokens = toint(customDimensions['output_tokens'])\r\n| extend total_tokens = toint(customDimensions['total_tokens'])\r\n| summarize input_tokens=sum(input_tokens), output_tokens=sum(output_tokens), total_tokens=sum(total_tokens) by process_id\r\n| summarize\r\n    TotalDocuments = count(),\r\n    TotalInputTokens = sum(input_tokens),\r\n    TotalOutputTokens = sum(output_tokens),\r\n    TotalTokens = sum(total_tokens),\r\n    AvgTokensPerDocument = round(avg(total_tokens), 0)",
         "size": 3,
         "title": "Overall Token Usage Summary",
         "timeContextFromParameter": "TimeRange",
@@ -95,7 +95,7 @@
       "type": 3,
       "content": {
         "version": "KqlItem/1.0",
-        "query": "customEvents\r\n| where name == 'LLM_Token_Usage_Summary'\r\n| where timestamp > ago(7d)\r\n| extend process_id = tostring(customDimensions['process_id'])\r\n| extend input_tokens = toint(customDimensions['total_input_tokens'])\r\n| extend output_tokens = toint(customDimensions['total_output_tokens'])\r\n| summarize input_tokens=max(input_tokens), output_tokens=max(output_tokens), timestamp=min(timestamp) by process_id\r\n| summarize InputTokens = sum(input_tokens), OutputTokens = sum(output_tokens) by bin(timestamp, 1h)\r\n| order by timestamp asc",
+        "query": "customEvents\r\n| where name == 'LLM_Agent_Token_Usage'\r\n| where timestamp > ago(7d)\r\n| extend process_id = tostring(customDimensions['process_id'])\r\n| extend input_tokens = toint(customDimensions['input_tokens'])\r\n| extend output_tokens = toint(customDimensions['output_tokens'])\r\n| summarize input_tokens=sum(input_tokens), output_tokens=sum(output_tokens), timestamp=min(timestamp) by process_id\r\n| summarize InputTokens = sum(input_tokens), OutputTokens = sum(output_tokens) by bin(timestamp, 1h)\r\n| order by timestamp asc",
         "size": 0,
         "title": "Token Usage Over Time (Hourly)",
         "timeContextFromParameter": "TimeRange",
@@ -130,7 +130,7 @@
       "type": 3,
       "content": {
         "version": "KqlItem/1.0",
-        "query": "let input_price_per_million = 2.50;\r\nlet output_price_per_million = 10.00;\r\ncustomEvents\r\n| where name == 'LLM_Token_Usage_Summary'\r\n| where timestamp > ago(7d)\r\n| extend process_id = tostring(customDimensions['process_id'])\r\n| extend input_tokens = toint(customDimensions['total_input_tokens'])\r\n| extend output_tokens = toint(customDimensions['total_output_tokens'])\r\n| summarize input_tokens=max(input_tokens), output_tokens=max(output_tokens), timestamp=min(timestamp) by process_id\r\n| summarize TotalInput = sum(input_tokens), TotalOutput = sum(output_tokens) by bin(timestamp, 1d)\r\n| extend InputCost = round(TotalInput * input_price_per_million / 1000000.0, 4)\r\n| extend OutputCost = round(TotalOutput * output_price_per_million / 1000000.0, 4)\r\n| extend TotalCost = InputCost + OutputCost\r\n| project Day = timestamp, TotalInput, TotalOutput, InputCost, OutputCost, TotalCost\r\n| order by Day desc",
+        "query": "let input_price_per_million = 2.50;\r\nlet output_price_per_million = 10.00;\r\ncustomEvents\r\n| where name == 'LLM_Agent_Token_Usage'\r\n| where timestamp > ago(7d)\r\n| extend process_id = tostring(customDimensions['process_id'])\r\n| extend input_tokens = toint(customDimensions['input_tokens'])\r\n| extend output_tokens = toint(customDimensions['output_tokens'])\r\n| summarize input_tokens=sum(input_tokens), output_tokens=sum(output_tokens), timestamp=min(timestamp) by process_id\r\n| summarize TotalInput = sum(input_tokens), TotalOutput = sum(output_tokens) by bin(timestamp, 1d)\r\n| extend InputCost = round(TotalInput * input_price_per_million / 1000000.0, 4)\r\n| extend OutputCost = round(TotalOutput * output_price_per_million / 1000000.0, 4)\r\n| extend TotalCost = InputCost + OutputCost\r\n| project Day = timestamp, TotalInput, TotalOutput, InputCost, OutputCost, TotalCost\r\n| order by Day desc",
         "size": 0,
         "title": "Estimated Daily Cost (GPT-4o Pricing: $2.50/1M input, $10.00/1M output)",
         "timeContextFromParameter": "TimeRange",
@@ -193,7 +193,7 @@
       "type": 3,
       "content": {
         "version": "KqlItem/1.0",
-        "query": "customEvents\r\n| where name == 'LLM_Token_Usage_Summary'\r\n| where timestamp > ago(7d)\r\n| extend total_tokens = toint(customDimensions['total_tokens'])\r\n| extend process_id = tostring(customDimensions['process_id'])\r\n| extend file_name = tostring(customDimensions['file_name'])\r\n| summarize TotalTokens = max(total_tokens) by process_id, file_name\r\n| order by TotalTokens desc\r\n| take 20",
+        "query": "customEvents\r\n| where name == 'LLM_Agent_Token_Usage'\r\n| where timestamp > ago(7d)\r\n| extend process_id = tostring(customDimensions['process_id'])\r\n| extend total_tokens = toint(customDimensions['total_tokens'])\r\n| summarize TotalTokens = sum(total_tokens) by process_id\r\n| join kind=leftouter (\r\n    customEvents\r\n    | where name == 'LLM_Token_Usage_Summary'\r\n    | where timestamp > ago(7d)\r\n    | extend process_id = tostring(customDimensions['process_id'])\r\n    | extend file_name = tostring(customDimensions['file_name'])\r\n    | summarize file_name=take_any(file_name) by process_id\r\n) on process_id\r\n| project process_id, file_name, TotalTokens\r\n| order by TotalTokens desc\r\n| take 20",
         "size": 0,
         "title": "Top 20 Token Consumers by Document",
         "timeContextFromParameter": "TimeRange",
@@ -207,7 +207,7 @@
       "type": 3,
       "content": {
         "version": "KqlItem/1.0",
-        "query": "customEvents\r\n| where name == 'LLM_Token_Usage_Summary'\r\n| where timestamp > ago(7d)\r\n| extend process_id = tostring(customDimensions['process_id'])\r\n| extend total_tokens = toint(customDimensions['total_tokens'])\r\n| extend input_tokens = toint(customDimensions['total_input_tokens'])\r\n| extend output_tokens = toint(customDimensions['total_output_tokens'])\r\n| extend mime_type = tostring(customDimensions['file_mime_type'])\r\n| extend file_type = case(\r\n    mime_type has \"pdf\", \"PDF\",\r\n    mime_type has \"image\", \"Image\",\r\n    mime_type has \"word\" or mime_type has \"docx\", \"Word\",\r\n    mime_type has \"excel\" or mime_type has \"xlsx\", \"Excel\",\r\n    mime_type has \"text\", \"Text\",\r\n    \"Other\")\r\n| summarize input_tokens=max(input_tokens), output_tokens=max(output_tokens), total_tokens=max(total_tokens), file_type=take_any(file_type) by process_id\r\n| summarize\r\n    Documents = count(),\r\n    TotalInputTokens = sum(input_tokens),\r\n    TotalOutputTokens = sum(output_tokens),\r\n    TotalTokens = sum(total_tokens),\r\n    AvgTokensPerDoc = round(avg(total_tokens), 0)\r\n    by FileType = file_type\r\n| order by TotalTokens desc",
+        "query": "customEvents\r\n| where name == 'LLM_Agent_Token_Usage'\r\n| where timestamp > ago(7d)\r\n| extend process_id = tostring(customDimensions['process_id'])\r\n| extend input_tokens = toint(customDimensions['input_tokens'])\r\n| extend output_tokens = toint(customDimensions['output_tokens'])\r\n| extend total_tokens = toint(customDimensions['total_tokens'])\r\n| summarize input_tokens=sum(input_tokens), output_tokens=sum(output_tokens), total_tokens=sum(total_tokens) by process_id\r\n| join kind=leftouter (\r\n    customEvents\r\n    | where name == 'LLM_Token_Usage_Summary'\r\n    | where timestamp > ago(7d)\r\n    | extend process_id = tostring(customDimensions['process_id'])\r\n    | extend mime_type = tostring(customDimensions['file_mime_type'])\r\n    | summarize mime_type=take_any(mime_type) by process_id\r\n) on process_id\r\n| extend file_type = case(\r\n    mime_type has \"pdf\", \"PDF\",\r\n    mime_type has \"image\", \"Image\",\r\n    mime_type has \"word\" or mime_type has \"docx\", \"Word\",\r\n    mime_type has \"excel\" or mime_type has \"xlsx\", \"Excel\",\r\n    mime_type has \"text\", \"Text\",\r\n    \"Other\")\r\n| summarize\r\n    Documents = count(),\r\n    TotalInputTokens = sum(input_tokens),\r\n    TotalOutputTokens = sum(output_tokens),\r\n    TotalTokens = sum(total_tokens),\r\n    AvgTokensPerDoc = round(avg(total_tokens), 0)\r\n    by FileType = file_type\r\n| order by TotalTokens desc",
         "size": 0,
         "title": "Token Usage by File Type",
         "timeContextFromParameter": "TimeRange",
@@ -228,7 +228,7 @@
       "type": 3,
       "content": {
         "version": "KqlItem/1.0",
-        "query": "customEvents\r\n| where name == 'LLM_Token_Usage_Summary'\r\n| where timestamp > ago(7d)\r\n| extend process_id = tostring(customDimensions['process_id'])\r\n| extend total_tokens = toint(customDimensions['total_tokens'])\r\n| summarize total_tokens=max(total_tokens) by process_id\r\n| summarize\r\n    p50 = percentile(total_tokens, 50),\r\n    p90 = percentile(total_tokens, 90),\r\n    p95 = percentile(total_tokens, 95),\r\n    p99 = percentile(total_tokens, 99),\r\n    Max = max(total_tokens)",
+        "query": "customEvents\r\n| where name == 'LLM_Agent_Token_Usage'\r\n| where timestamp > ago(7d)\r\n| extend process_id = tostring(customDimensions['process_id'])\r\n| extend total_tokens = toint(customDimensions['total_tokens'])\r\n| summarize total_tokens=sum(total_tokens) by process_id\r\n| summarize\r\n    p50 = percentile(total_tokens, 50),\r\n    p90 = percentile(total_tokens, 90),\r\n    p95 = percentile(total_tokens, 95),\r\n    p99 = percentile(total_tokens, 99),\r\n    Max = max(total_tokens)",
         "size": 3,
         "title": "Token Usage Percentiles Per Document",
         "timeContextFromParameter": "TimeRange",
@@ -250,7 +250,7 @@
       "type": 3,
       "content": {
         "version": "KqlItem/1.0",
-        "query": "customEvents\r\n| where name == 'LLM_Token_Usage_Summary'\r\n| where timestamp > ago(7d)\r\n| extend process_id = tostring(customDimensions['process_id'])\r\n| extend total_tokens = toint(customDimensions['total_tokens'])\r\n| summarize total_tokens=max(total_tokens), timestamp=min(timestamp) by process_id\r\n| summarize\r\n    DocumentsProcessed = count(),\r\n    TotalTokens = sum(total_tokens),\r\n    AvgTokensPerDoc = round(avg(total_tokens), 0),\r\n    MaxTokensPerDoc = max(total_tokens)\r\n    by Day = bin(timestamp, 1d)\r\n| order by Day desc",
+        "query": "customEvents\r\n| where name == 'LLM_Agent_Token_Usage'\r\n| where timestamp > ago(7d)\r\n| extend process_id = tostring(customDimensions['process_id'])\r\n| extend total_tokens = toint(customDimensions['total_tokens'])\r\n| summarize total_tokens=sum(total_tokens), timestamp=min(timestamp) by process_id\r\n| summarize\r\n    DocumentsProcessed = count(),\r\n    TotalTokens = sum(total_tokens),\r\n    AvgTokensPerDoc = round(avg(total_tokens), 0),\r\n    MaxTokensPerDoc = max(total_tokens)\r\n    by Day = bin(timestamp, 1d)\r\n| order by Day desc",
         "size": 0,
         "title": "Daily Processing Volume with Token Usage",
         "timeContextFromParameter": "TimeRange",

src/ContentProcessor/src/libs/pipeline/handlers/save_handler.py

@@ -170,6 +170,9 @@ def find_process_result(step_name: str):
         )
 
         # Emit token usage summary and per-model events to Application Insights
+        # NOTE: This summary only contains tokens from the evaluate/map step.
+        # For true totals across all pipeline steps (Summarize, RAI, GapAnalysis),
+        # aggregate from LLM_Agent_Token_Usage events grouped by process_id.
         emit_summary_token_event(
             total_input_tokens=evaluated_result.prompt_tokens,
             total_output_tokens=evaluated_result.completion_tokens,