feat(replay-vision): capture gemini reasoning and group scan traces in llma

products/ai_observability/frontend/messageNormalization.test.ts

@@ -48,6 +48,44 @@ describe('messageNormalization', () => {
         })
     })
 
+    describe('thinking blocks alongside typed function blocks', () => {
+        // The Gemini SDK emits a thought summary block next to a tool call. The thinking
+        // block knocks the message off compat_array's envelope rule, so the function block
+        // must survive per-block delegation as a typed tool call, not stringified JSON.
+        it('renders the thought as thinking and keeps the sibling tool call typed', () => {
+            const { messages, recognized } = normalizeMessages(
+                [
+                    {
+                        role: 'assistant',
+                        content: [
+                            { type: 'thinking', thinking: 'Two rapid clicks, check the events.' },
+                            {
+                                type: 'function',
+                                function: { name: 'get_events_around', arguments: { rec_t: 320 } },
+                            },
+                        ],
+                    },
+                ],
+                'assistant'
+            )
+            expect(recognized).toBe(true)
+            expect(messages).toEqual([
+                { role: 'assistant (thinking)', content: 'Two rapid clicks, check the events.' },
+                {
+                    role: 'assistant',
+                    content: '',
+                    tool_calls: [
+                        {
+                            type: 'function',
+                            id: undefined,
+                            function: { name: 'get_events_around', arguments: { rec_t: 320 } },
+                        },
+                    ],
+                },
+            ])
+        })
+    })
+
     describe('offloaded media', () => {
         const HASH = 'a'.repeat(64)
         const POINTER = `phaiblob://v1/sha256/${HASH}?mime=image%2Fpng&size=332378`

products/ai_observability/frontend/normalizer/recipe/default_recipes/compat_array.yaml

@@ -36,3 +36,19 @@ rules:
                       id: $.id
                       name: $.function.name
                       args: $.function.arguments
+
+    # A bare `{type: function}` block — reached when a sibling block outside the
+    # allowlist above (e.g. a `thinking` block from a Gemini thought summary)
+    # knocks the whole message off the envelope rule and per-block delegation
+    # dispatches each block on its own. Without this no rule claims the block and
+    # salvage stringifies the tool call into raw JSON.
+    - on:
+          type: function
+          function:
+              name: { exists: true }
+      emit:
+          content: ''
+          toolCall:
+              id: $.id
+              name: $.function.name
+              args: $.function.arguments

products/replay_vision/backend/temporal/activities/call_scanner_provider.py

@@ -14,7 +14,7 @@
 from dataclasses import dataclass
 from datetime import timedelta
 from typing import Any, TypeVar
-from uuid import UUID
+from uuid import UUID, uuid4
 
 from django.utils import timezone
 
@@ -144,12 +144,22 @@ async def _call_scanner_provider(inputs: CallScannerProviderInputs) -> ScannerCa
         preamble_text=preamble_text,
         team_id=inputs.team_id,
         llm_inputs=llm_inputs,
+        trace_id=_scan_trace_id(inputs),
     )
     duration_ms = int(llm_inputs.metadata.duration_seconds * 1000)
     finalized = _resolve_citations(finalized, scanner, duration_ms)
     return ScannerCallOutput(model_output=finalized, signals=signals)
 
 
+def _scan_trace_id(inputs: CallScannerProviderInputs) -> str:
+    """LLM analytics trace id for one scan: the observation id, so every step, tool round-trip, and retry of
+    a scan reads as a single conversation and the observation id doubles as the trace search key. Evaluation
+    re-runs (snapshot_override) get a fresh id so they don't interleave with the real scan's trace."""
+    if inputs.snapshot_override is not None:
+        return str(uuid4())
+    return str(inputs.observation_id)
+
+
 def _resolve_citations(
     finalized: _OutputT,
     scanner: BaseScanner,
@@ -297,6 +307,7 @@ async def _run_mission(
     preamble_text: str,
     team_id: int,
     llm_inputs: ScannerLlmInputs,
+    trace_id: str,
 ) -> tuple[BaseScannerOutput, list[SignalFinding]]:
     """Cache the video, run every mission step as a tool-using turn, then assemble the output + side-mission findings.
 
@@ -337,6 +348,7 @@ def dispatch(call: Any) -> dict[str, Any]:
         dispatch=dispatch,
         team_id=team_id,
         metric_labels=metric_labels,
+        trace_id=trace_id,
     )
     try:
         step_outputs = await _run_mission_attempts(run=run, cache=cache, model=snapshot.model)
@@ -405,6 +417,7 @@ async def _run_steps(
     dispatch: Any,
     team_id: int,
     metric_labels: dict[str, str],
+    trace_id: str,
 ) -> dict[str, BaseModel]:
     """Run the ordered steps over one growing conversation; return the validated output keyed by step name."""
     # The video + preamble lead the conversation inline unless they're already cached as the prefix.
@@ -422,6 +435,7 @@ async def _run_steps(
             dispatch=dispatch,
             team_id=team_id,
             metric_labels=metric_labels,
+            trace_id=trace_id,
         )
         if result.output is None:
             # Roll the failed step's half-finished exchange back so the next instruction follows the last good
@@ -459,6 +473,7 @@ async def _run_step(
     dispatch: Any,
     team_id: int,
     metric_labels: dict[str, str],
+    trace_id: str,
 ) -> "_StepResult":
     """Run one step's tool loop with one re-prompt on failure. Returns the validated output, or why it was exhausted.
 
@@ -474,6 +489,8 @@ async def _generate(c: list[Any], cfg: types.GenerateContentConfig = config) ->
             contents=c,
             config=cfg,
             posthog_distinct_id=replay_vision_distinct_id(team_id),
+            posthog_trace_id=trace_id,
+            posthog_properties={"$ai_span_name": step.name},
             posthog_groups={"project": str(team_id)},
         )
 
@@ -589,6 +606,9 @@ def _step_config(step: MissionStep, cache_name: str | None, *, allow_tools: bool
     kwargs: dict[str, Any] = {
         "response_mime_type": "application/json",
         "response_json_schema": step.response_model.model_json_schema(),
+        # Return thought summaries so the model's reasoning is visible in LLM analytics. Answer parsing is
+        # unaffected (`response.text` skips thought parts); models with thinking off just return none.
+        "thinking_config": types.ThinkingConfig(include_thoughts=True),
     }
     if cache_name:
         kwargs["cached_content"] = cache_name  # video, preamble, and the tool all live in the cache

products/replay_vision/backend/tests/test_call_scanner_provider.py

@@ -75,6 +75,7 @@ async def _run(client: _FakeClient, steps: list[MissionStep], dispatch: Any = la
         dispatch=dispatch,
         team_id=1,
         metric_labels=_LABELS,
+        trace_id="trace-1",
     )
 
 
@@ -328,6 +329,7 @@ def test_inline_path_carries_tools_and_no_cache(self) -> None:
         assert config.tools is not None
         assert config.cached_content is None
         assert config.response_json_schema is not None
+        assert config.thinking_config is not None and config.thinking_config.include_thoughts is True
 
     def test_cached_path_references_the_cache_and_omits_tools(self) -> None:
         config = _step_config(MissionStep(name="core", instruction="c", response_model=_Core), cache_name="caches/abc")