feat(observability): add reranker scores, streaming model fix, eval mode tracing

.s2s/BACKLOG.md

@@ -1,6 +1,6 @@
 # Vektra Backlog
 
-**Updated**: 2026-03-28
+**Updated**: 2026-04-09
 **Format**: Single markdown file for tracking work items
 
 ---
@@ -24,7 +24,7 @@
 
 ### BUG-020: System prompt "use only this material" conflicts with multi-turn history
 
-**Status**: planned | **Priority**: high | **Created**: 2026-03-28
+**Status**: completed | **Priority**: high | **Created**: 2026-03-28 | **Completed**: 2026-04-04 | **PR**: #54
 
 **Context**: the system prompt instructs the LLM to "Use only this material to answer", where "material" refers to the `<context>` tags in the current user message. In multi-turn conversations, the conversation history is injected as separate user/assistant message pairs *before* the current message. The LLM correctly interprets the rule as applying only to the current `<context>` and ignores information from its own previous answers.
 
@@ -93,7 +93,7 @@ Design considerations:
 
 ### FEAT-019: Full prompt observability in eval mode
 
-**Status**: planned | **Priority**: low | **Created**: 2026-03-28
+**Status**: completed | **Priority**: low | **Created**: 2026-03-28 | **Completed**: 2026-04-07 | **PR**: #55
 **Related**: useful for diagnosing BUG-020 but not a fix for it.
 
 **Context**: when diagnosing RAG behavior, the assembled prompt (system + history + context + question) is the most important artifact, but it is never persisted. The `build_prompt` trace step records chunk count and history turn count, but not the actual text. Without seeing the full prompt, it is impossible to understand why the LLM produced a specific answer (e.g., was Art. 33 in the context? how was the history formatted? did the token budget truncate anything?).
@@ -115,7 +115,7 @@ Related to DEBT-015 (rewritten query in eval mode) but broader scope: this captu
 
 ### FEAT-020: Configurable prompt grounding mode (strict/hybrid)
 
-**Status**: in_progress | **Priority**: high | **Created**: 2026-03-28 | **Branch**: feat/prompt-grounding-mode
+**Status**: completed | **Priority**: high | **Created**: 2026-03-28 | **Completed**: 2026-04-09 | **PR**: #54
 **Blocks**: BUG-020 (this implements the fix)
 **Research**: `vektra-internal/stack/20260328-rag-prompt-research-multi-turn.md`
 
@@ -301,7 +301,7 @@ The `title` field would contain `filename + page` (e.g., "Costituzione italiana.
 
 ### BUG-013: QueryTrace not persisted to database
 
-**Status**: in_progress | **Priority**: high | **Created**: 2026-03-23 | **Branch**: fix/query-trace-observability
+**Status**: completed | **Priority**: high | **Created**: 2026-03-23 | **Completed**: 2026-04-04 | **PR**: #53
 **Analysis**: `vektra-internal/stack/20260323-rag-prompt-chunk-confusion-analysis.md`
 
 **Context**: `AnalyticsService.store_trace()` exists and is tested but is never called by any pipeline or endpoint. The `query_traces` table is always empty. Traces are generated by all pipelines (SimpleQueryPipeline, AdvancedQueryPipeline) and emitted via SSE to the client, but discarded server-side. This makes post-hoc diagnosis of query failures impossible - as demonstrated when a multi-turn failure ("si, entrambi") could not be investigated because all diagnostic data was lost.
@@ -474,7 +474,7 @@ A client using SSE without generating its own `conversation_id` cannot discover
 
 ### DEBT-011: Conversation and query trace observability gaps
 
-**Status**: in_progress | **Priority**: medium | **Created**: 2026-03-25 | **Branch**: fix/query-trace-observability
+**Status**: completed | **Priority**: medium | **Created**: 2026-03-25 | **Completed**: 2026-04-04 | **PR**: #53
 **Related**: BUG-018 (SSE conversation_id)
 
 **Context**: Diagnosing a conversation (`5bf50682`, namespace `ita-100`) revealed multiple observability gaps that make post-hoc analysis of query behavior difficult:
@@ -506,7 +506,7 @@ A client using SSE without generating its own `conversation_id` cannot discover
 
 ### BUG-019: llm_model field inconsistent between streaming and non-streaming traces
 
-**Status**: planned | **Priority**: low | **Created**: 2026-03-28
+**Status**: completed | **Priority**: low | **Created**: 2026-03-28 | **Completed**: 2026-04-07 | **PR**: #55
 
 **Context**: QueryTrace `llm_model` field has different values depending on the execution path. Non-streaming `execute()` sets it from the return value of `_call_llm_with_fallback()`, which returns the litellm-resolved model name (e.g. `qwen35-27b`). Streaming `_stream()` sets it from `self._llm_config.provider` (raw config value, e.g. `openai/qwen35-27b`). This inconsistency affects trace queries and metrics aggregation by model.
 
@@ -583,7 +583,7 @@ Also consider making `_REWRITE_TOP_K=20` configurable or deriving it from the re
 
 ### DEBT-014: Include all reranker scores in QueryTrace (not just post-threshold)
 
-**Status**: planned | **Priority**: medium | **Created**: 2026-03-28
+**Status**: completed | **Priority**: medium | **Created**: 2026-03-28 | **Completed**: 2026-04-07 | **PR**: #55
 
 **Context**: `chunks_retrieved` in QueryTrace contains only the chunks that pass the relevance threshold filter. Chunks scored by the reranker but filtered out are lost - there is no record of their chunk_id or score. This makes it impossible to evaluate whether the threshold is too aggressive (cutting good chunks) or too permissive without re-running the query.
 
@@ -602,7 +602,7 @@ The `StepTrace` metadata for the `rerank` step only contains `after_rerank: N` (
 
 ### DEBT-015: Persist rewritten query text in QueryTrace (dev/eval mode only)
 
-**Status**: planned | **Priority**: medium | **Created**: 2026-03-28
+**Status**: completed | **Priority**: medium | **Created**: 2026-03-28 | **Completed**: 2026-04-07 | **PR**: #55
 
 **Context**: when query rewriting is active, `_rewrite_query()` produces a rewritten query that replaces the original for embedding and retrieval. The rewritten text is not stored anywhere - the `query_rewrite` step metadata contains only `rewritten: true/false`, `history_turns_used`, and `original_query_hash`. Without the rewritten text, it is impossible to understand why the retrieval returned certain chunks in a multi-turn conversation.
 
@@ -623,7 +623,7 @@ DEBT-009 addresses debug logging of the rewritten query to structlog. This entry
 
 ### DEBT-009: Debug logging for rewritten queries
 
-**Status**: planned | **Priority**: medium | **Created**: 2026-03-23
+**Status**: completed | **Priority**: medium | **Created**: 2026-03-23 | **Completed**: 2026-04-07 | **PR**: #55
 **Analysis**: `vektra-internal/stack/20260323-rag-prompt-chunk-confusion-analysis.md`
 
 **Context**: the `_rewrite_query()` method in AdvancedQueryPipeline does not log the rewritten query text. Only a SHA-256 hash of the original query is stored in StepTrace metadata. This is by design for GDPR (ARCH-041: "QueryTrace does not contain query text or response content"), but makes it impossible to diagnose rewrite failures in development.
@@ -642,7 +642,7 @@ DEBT-009 addresses debug logging of the rewritten query to structlog. This entry
 
 ### DEBT-016: Remove unused conversation.j2 template and render_conversation()
 
-**Status**: in_progress | **Priority**: low | **Created**: 2026-03-28 | **Branch**: feat/prompt-grounding-mode
+**Status**: completed | **Priority**: low | **Created**: 2026-03-28 | **Completed**: 2026-04-04 | **PR**: #54
 
 **Context**: ARCH-054 designed three composable Jinja2 templates: `system.j2`, `context.j2`, `conversation.j2`. During Phase 1 implementation (Wave 3, commit 7939b22), the pipeline chose to pass history as native chat messages via `_history_to_messages()` (user/assistant role pairs) instead of rendering it as text via `conversation.j2`. This is the correct approach for modern chat models.
 

vektra-core/src/vektra_core/advanced_pipeline.py

@@ -96,6 +96,8 @@ def __init__(
         self._sparse_embedding = sparse_embedding
         self._reranker = reranker
         self._rewrite_enabled = pipeline_config.rewrite.enabled
+        self._eval_mode = pipeline_config.eval_mode
+        self._debug_log_queries = pipeline_config.debug_log_queries
 
     # -- Helpers (delegating to shared module-level functions) --
 
@@ -124,10 +126,17 @@ async def _rewrite_query(
         t0 = time.monotonic()
 
         if not self._rewrite_enabled or not history:
+            skip_meta: dict[str, object] = {
+                "rewritten": False,
+                "history_turns_used": 0,
+            }
+            if self._eval_mode:
+                skip_meta["original_query"] = question
+                skip_meta["rewritten_query"] = question
             return question, StepTrace(
                 name="query_rewrite",
                 duration_ms=_elapsed_ms(t0),
-                metadata={"rewritten": False, "history_turns_used": 0},
+                metadata=skip_meta,
             )
 
         try:
@@ -150,22 +159,42 @@ async def _rewrite_query(
             if not rewritten:
                 rewritten = question
 
+            if self._debug_log_queries:
+                log.debug(
+                    "query_rewritten",
+                    original_query=question,
+                    rewritten_query=rewritten,
+                    history_turns=len(history),
+                )
+
             original_hash = hashlib.sha256(question.encode()).hexdigest()[:8]
+            metadata: dict[str, object] = {
+                "original_query_hash": original_hash,
+                "rewritten": True,
+                "history_turns_used": len(history),
+            }
+            if self._eval_mode:
+                metadata["original_query"] = question
+                metadata["rewritten_query"] = rewritten
+
             return rewritten, StepTrace(
                 name="query_rewrite",
                 duration_ms=_elapsed_ms(t0),
-                metadata={
-                    "original_query_hash": original_hash,
-                    "rewritten": True,
-                    "history_turns_used": len(history),
-                },
+                metadata=metadata,
             )
         except Exception as exc:
             log.warning("query_rewrite_failed", error=str(exc))
+            err_meta: dict[str, object] = {
+                "rewritten": False,
+                "error": str(exc),
+            }
+            if self._eval_mode:
+                err_meta["original_query"] = question
+                err_meta["rewritten_query"] = question
             return question, StepTrace(
                 name="query_rewrite",
                 duration_ms=_elapsed_ms(t0),
-                metadata={"rewritten": False, "error": str(exc)},
+                metadata=err_meta,
             )
 
     # -- Pre-LLM steps shared between execute() and execute_stream() --
@@ -276,14 +305,28 @@ async def _run_pre_llm_steps(
         if self._reranker and results:
             t0 = time.monotonic()
             try:
-                results = await self._reranker.rerank(
+                rerank_result = await self._reranker.rerank(
                     effective_query, results, top_k=query.top_k
                 )
+                results = rerank_result.top_k
+                rerank_meta: dict[str, object] = {
+                    "after_rerank": len(results),
+                    "candidates_evaluated": len(rerank_result.all_scores),
+                }
+                if self._eval_mode:
+                    rerank_meta["scores"] = [
+                        {
+                            "chunk_id": cid,
+                            "reranker_score": rs,
+                            "original_score": orig,
+                        }
+                        for cid, rs, orig in rerank_result.all_scores
+                    ]
                 steps.append(
                     StepTrace(
                         name="rerank",
                         duration_ms=_elapsed_ms(t0),
-                        metadata={"after_rerank": len(results)},
+                        metadata=rerank_meta,
                     )
                 )
             except Exception as exc:
@@ -413,14 +456,19 @@ def _build_prompt(
         messages.extend(_history_to_messages(selected_history))
         messages.append(Message(role="user", content=user_content))
 
+        build_meta: dict[str, object] = {
+            "prompt_version": self._renderer.prompt_version,
+            "chunks_in_prompt": len(selected_chunks),
+            "history_turns_in_prompt": len(selected_history),
+        }
+        if self._eval_mode:
+            build_meta["messages"] = [
+                {"role": m.role, "content": m.content} for m in messages
+            ]
         step = StepTrace(
             name="build_prompt",
             duration_ms=_elapsed_ms(t0),
-            metadata={
-                "prompt_version": self._renderer.prompt_version,
-                "chunks_in_prompt": len(selected_chunks),
-                "history_turns_in_prompt": len(selected_history),
-            },
+            metadata=build_meta,
         )
         return messages, selected_chunks, selected_history, step
 
@@ -622,15 +670,18 @@ async def _stream(
         # Step 8: Stream LLM tokens
         t0 = time.monotonic()
         full_answer_parts: list[str] = []
-        llm_model = self._llm_config.provider
+        llm_model: str | None = None
         try:
             token_stream = await self._llm.stream(
                 messages, model=self._llm_config.provider
             )
             async for chunk in token_stream:
+                if chunk.model and llm_model is None:
+                    llm_model = chunk.model
                 if chunk.content:
                     full_answer_parts.append(chunk.content)
                     yield QueryChunk(type="token", data=chunk.content)
+            llm_model = llm_model or self._llm_config.provider
             steps.append(
                 StepTrace(
                     name="llm_stream",
@@ -640,11 +691,12 @@ async def _stream(
             )
         except Exception as exc:
             log.warning("stream_llm_failed", error=str(exc))
+            llm_model = llm_model or self._llm_config.provider
             steps.append(
                 StepTrace(
                     name="llm_stream",
                     duration_ms=_elapsed_ms(t0),
-                    metadata={"error": str(exc)},
+                    metadata={"model": llm_model, "error": str(exc)},
                 )
             )
             yield QueryChunk(type="error", data="LLM unavailable")

vektra-core/src/vektra_core/pipeline.py

@@ -262,6 +262,8 @@ def __init__(
         self._conversation_store = conversation_store
         self._renderer = renderer
         self._config = pipeline_config
+        self._eval_mode = pipeline_config.eval_mode
+        self._debug_log_queries = pipeline_config.debug_log_queries
 
     def _count_tokens(self, text: str) -> int:
         return _count_tokens_impl(self._llm, self._llm_config.provider, text)
@@ -498,15 +500,20 @@ async def execute(
         messages.extend(_history_to_messages(selected_history))
         messages.append(Message(role="user", content=user_content))
 
+        build_meta: dict[str, object] = {
+            "prompt_version": self._renderer.prompt_version,
+            "chunks_in_prompt": len(selected_chunks),
+            "history_turns_in_prompt": len(selected_history),
+        }
+        if self._eval_mode:
+            build_meta["messages"] = [
+                {"role": m.role, "content": m.content} for m in messages
+            ]
         steps.append(
             StepTrace(
                 name="build_prompt",
                 duration_ms=_elapsed_ms(t0),
-                metadata={
-                    "prompt_version": self._renderer.prompt_version,
-                    "chunks_in_prompt": len(selected_chunks),
-                    "history_turns_in_prompt": len(selected_history),
-                },
+                metadata=build_meta,
             )
         )
 
@@ -798,43 +805,53 @@ async def _stream(self, query: QueryRequest) -> AsyncGenerator[QueryChunk, None]
         messages: list[Message] = [Message(role="system", content=system_text)]
         messages.extend(_history_to_messages(selected_history))
         messages.append(Message(role="user", content=user_content))
+        stream_build_meta: dict[str, object] = {
+            "prompt_version": self._renderer.prompt_version,
+            "chunks_in_prompt": len(selected_chunks),
+            "history_turns_in_prompt": len(selected_history),
+        }
+        if self._eval_mode:
+            stream_build_meta["messages"] = [
+                {"role": m.role, "content": m.content} for m in messages
+            ]
         steps.append(
             StepTrace(
                 name="build_prompt",
                 duration_ms=_elapsed_ms(t0),
-                metadata={
-                    "prompt_version": self._renderer.prompt_version,
-                    "chunks_in_prompt": len(selected_chunks),
-                    "history_turns_in_prompt": len(selected_history),
-                },
+                metadata=stream_build_meta,
             )
         )
 
         # Step 5: Stream LLM tokens
         t0 = time.monotonic()
         full_answer_parts: list[str] = []
+        llm_model_resolved: str | None = None
         try:
             token_stream = await self._llm.stream(
                 messages, model=self._llm_config.provider
             )
             async for chunk in token_stream:
+                if chunk.model and llm_model_resolved is None:
+                    llm_model_resolved = chunk.model
                 if chunk.content:
                     full_answer_parts.append(chunk.content)
                     yield QueryChunk(type="token", data=chunk.content)
+            llm_model_resolved = llm_model_resolved or self._llm_config.provider
             steps.append(
                 StepTrace(
                     name="llm_stream",
                     duration_ms=_elapsed_ms(t0),
-                    metadata={"model": self._llm_config.provider},
+                    metadata={"model": llm_model_resolved},
                 )
             )
         except Exception as exc:
             log.warning("stream_llm_failed", error=str(exc))
+            llm_model_resolved = llm_model_resolved or self._llm_config.provider
             steps.append(
                 StepTrace(
                     name="llm_stream",
                     duration_ms=_elapsed_ms(t0),
-                    metadata={"error": str(exc)},
+                    metadata={"model": llm_model_resolved, "error": str(exc)},
                 )
             )
             yield QueryChunk(type="error", data="LLM unavailable")
@@ -845,7 +862,7 @@ async def _stream(self, query: QueryRequest) -> AsyncGenerator[QueryChunk, None]
                 chunks_retrieved=[
                     ChunkRef(chunk_id=r.chunk_id, score=r.score) for r in filtered
                 ],
-                llm_model=self._llm_config.provider,
+                llm_model=llm_model_resolved,
                 prompt_version=self._renderer.prompt_version,
                 created_at=datetime.now(UTC),
             )
@@ -915,7 +932,7 @@ async def _stream(self, query: QueryRequest) -> AsyncGenerator[QueryChunk, None]
             chunks_retrieved=[
                 ChunkRef(chunk_id=r.chunk_id, score=r.score) for r in filtered
             ],
-            llm_model=self._llm_config.provider,
+            llm_model=llm_model_resolved,
             prompt_version=self._renderer.prompt_version,
             created_at=datetime.now(UTC),
         )

vektra-core/src/vektra_core/providers/litellm_provider.py

@@ -104,11 +104,16 @@ async def _stream_impl(
             **self._base_kwargs,
             **kwargs,
         )
+        resolved_model: str | None = None
         async for chunk in response:
+            if resolved_model is None:
+                resolved_model = getattr(chunk, "model", None) or model
             delta = chunk.choices[0].delta
             content = (delta.content or "") if delta else ""
             finish = chunk.choices[0].finish_reason
-            yield CompletionChunk(content=content, done=finish is not None)
+            yield CompletionChunk(
+                content=content, done=finish is not None, model=resolved_model
+            )
 
     async def health_check(self) -> HealthStatus:
         """Probe the primary model with a short completion (5-second timeout)."""