diff --git a/products/ai_observability/frontend/messageNormalization.test.ts b/products/ai_observability/frontend/messageNormalization.test.ts index 1f75ceea2ce8..d40eb41f2425 100644 --- a/products/ai_observability/frontend/messageNormalization.test.ts +++ b/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` diff --git a/products/ai_observability/frontend/normalizer/recipe/default_recipes/compat_array.yaml b/products/ai_observability/frontend/normalizer/recipe/default_recipes/compat_array.yaml index 719a445fbb40..bce2deb5a483 100644 --- a/products/ai_observability/frontend/normalizer/recipe/default_recipes/compat_array.yaml +++ b/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 diff --git a/products/replay_vision/backend/temporal/activities/call_scanner_provider.py b/products/replay_vision/backend/temporal/activities/call_scanner_provider.py index d556c818e5f6..e43a88b4e2b9 100644 --- a/products/replay_vision/backend/temporal/activities/call_scanner_provider.py +++ b/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 diff --git a/products/replay_vision/backend/tests/test_call_scanner_provider.py b/products/replay_vision/backend/tests/test_call_scanner_provider.py index 33c845a15819..bcc4c27c8e43 100644 --- a/products/replay_vision/backend/tests/test_call_scanner_provider.py +++ b/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")