Traces topic

[rakshak-akto]

Original Flow (before our work)

Cursor/AI Agent
      │  HTTP request via Akto proxy
      ▼
mini-runtime: HttpCallParser
  ├─ checks AGENT_TRAFFIC_LOGS feature flag (allowAnalysis)
  ├─ checks isAgenticTraffic() on tags
  ├─ builds AgentQueryRecord from headers/payload
  └─ POST to cyborg /api/storeAgentQueryData
         │
         ▼
cyborg: database-abstractor
  └─ ElasticSearchClient.bulkIndexAgentQueryRecords()
       → writes doc to ES index agent_query
         (was missing: traceId, spanId, topicProcessed)
         │
         ▼
ES: agent_query index
  [accountId, sessionIdentifier, serviceId, userName,
   timestamp, queryPayload, responsePayload,
   inputTokens, outputTokens, isAtlasTraffic]
     │
     ▼ (every 10 min)

master: UserAnalysisCron

├─ scrolls ES for recent records

├─ calls UserQueryTopicClassifier (LLM) per record

└─ stored aggregate topic counts in MongoDB ONLY

(topics were NOT written back to ES docs)
     │
     ▼

master: LLMObservabilityAction (dashboard API)

├─ fetchSessions()  → agg by sessionIdentifier

├─ fetchMessages()  → agg by traceId

└─ searchPrompts()  → flat span search

(no topic field in response, no topic filter)
     │
     ▼

UI: /observe/llm-observability

Sessions | Messages | Prompts tabs

(no Topic column, no Topic filter)

---

Changes We Made

1. mini-runtime — AgentQueryRecord.java

File: libs/utils/.../AgentQueryRecord.java

Added fallback header for sessionIdentifier:

// was: only read x-akto-installer-akto_session_id

String sessionIdentifier = getFirstHeader(headers, HEADER_PREFIX + HEADER_SESSION_ID);

// added fallback:

if (sessionIdentifier == null || sessionIdentifier.isEmpty()) {

sessionIdentifier = getFirstHeader(headers, HEADER_PREFIX + "akto_conversation_id");

}

Cursor sends x-akto-installer-akto_conversation_id, not akto_session_id, so sessions were always null. This fixes the Sessions tab.

2. mini-runtime — HttpCallParser.java

File: apps/mini-runtime/.../HttpCallParser.java

• Added agentQueryDebug log line exposing all gate conditions (allowAnalysis, isAgentic, deviceUserMapSize, tags, URL) — critical for diagnosing why records weren't flowing
• Bypassed AGENT_TRAFFIC_LOGS FF: boolean allowAnalysis = true (TODO: revert before prod merge)
• Refactored to cache deviceUserMap in a variable and reuse it (avoided double DB fetch)

3. cyborg — ElasticSearchClient.java

File: apps/database-abstractor/.../ElasticSearchClient.java

Added 3 missing fields to the NDJSON written to ES on every prompt ingestion:

.put("traceId",        r.getTraceId())   // was missing — Messages tab needs it

.put("spanId",         r.getSpanId())    // was missing

.put("topicProcessed", false)            // new — cron uses this as a processed flag

4. master — ElasticSearchClient.java (dashboard lib)

File: libs/utils/.../ElasticSearchClient.java

Added topic write-back infrastructure:

public static final class TopicUpdate { String docId; String topic; }

public void bulkUpdateTopics(List<TopicUpdate> updates) {

// POSTs _bulk NDJSON: {"update":{"_id":"<docId>"}}\n{"doc":{"topic":"...","topicProcessed":true}}\n

}

5. master — UserAnalysisCron.java

File: apps/dashboard/.../crons/UserAnalysisCron.java

Wired topic evaluation → ES write-back:

• processAccount() accumulates a topicUpdates list during the scroll
• processRecord() extracts the primary topic from classifier result and buffers a TopicUpdate
• After scroll completes, flushes all updates via bulkUpdateTopics()

Before: topics computed by LLM but discarded (only MongoDB aggregate counts stored)

After: primary topic written back to each ES document with topicProcessed: true

6. master — LLMObservabilityAction.java

File: apps/dashboard/.../action/monitoring/LLMObservabilityAction.java

What	Detail
topicFilters field	@Setter List topicFilters — accepts topic filter from UI
topicsAgg in fetchSessions	terms agg on topic.keyword size 10, nested under session groups
topicsAgg in fetchMessages	same, nested under trace groups
parseBuckets()	extracts topics from topicsAgg.buckets[].key → puts topics: [...] in each row
fetchPromptFilters()	adds topic terms agg → filterChoices.put("topic", [...])
buildMultiFilters()	when topicFilters non-empty → adds topic.keyword terms clause to ES query
debug logs	temporary logger.info in fetchMessages() to diagnose ES config / result count

7. master — LLMCellRenderers.jsx

File: apps/dashboard/web/.../observe/llm/LLMCellRenderers.jsx

New TopicChipCell component — renders topics as colored pill badges with deterministic color per topic (hash-based, 6 pastel colors). Shows up to 2 chips per row.

8. master — columns.jsx

File: apps/dashboard/web/.../observe/llm/columns.jsx

Added shared topicCol (width 180) using TopicChipCell, inserted into:

• SESSION_COLUMN_DEFS (after trace count)
• getTraceColumnDefs() (after span count)
• MESSAGE_FLAT_COLUMN_DEFS (after model column)

9. master — docker-compose.yml

Changed akto-api-security-runtime image from the ECR published tag to akto-mini-runtime:local so local code changes are picked up without re-publishing.

---

Updated Flow

Cursor → mini-runtime → cyborg → ES
                                  ▼
                    ES doc now has: traceId, spanId,
                    sessionIdentifier (from conversation_id),
                    topicProcessed: false
Every 10 min — UserAnalysisCron:

scroll ES → LLM classify → bulkUpdateTopics()

ES doc updated: topic: "Code Generation", topicProcessed: true
Dashboard API:

fetchSessions/Messages  → includes topicsAgg sub-agg

parseBuckets()          → topic chips in response

fetchPromptFilters()    → topic choices for filter dropdown

buildMultiFilters()     → topic filter applied to ES query
UI:

Sessions | Messages | Prompts  — all now show Topic column

Filter panel includes Topic dropdown