fix(observability): persist QueryTrace and add admin conversation turns endpoint

.env.example

@@ -112,7 +112,13 @@
 # Fallback model when primary times out.
 # VEKTRA_LLM_FALLBACK_MODEL=
 # VEKTRA_LLM_FALLBACK_TIMEOUT_MS=60000
-# VEKTRA_LLM_CONTEXT_WINDOW=           # Required for models not in litellm registry (e.g. local vLLM). Example: 32768
+# VEKTRA_LLM_CONTEXT_WINDOW=           # Required for models not in litellm registry (e.g. local vLLM).
+#   Set to the model's actual context length. Check your provider:
+#     vLLM: curl /v1/models → max_model_len
+#     Ollama: ollama show <model> → context_length
+#   If unset, litellm tries to look it up; for unknown models it falls back
+#   to 4096, which causes most retrieved chunks to be discarded (BUG-017).
+#   Examples: Qwen 3.5 27B → 262144, Llama 3 8B → 8192, GPT-4o → 128000
 # VEKTRA_LLM_CONTEXT_ONLY_ENABLED=true
 
 # --------------------------------------------------------------------------

.s2s/plans/20260328-core-trace-observability.md

@@ -0,0 +1,117 @@
+# Implementation Plan: QueryTrace persistence and conversation observability
+
+**ID**: 20260328-core-trace-observability
+**Status**: in-progress
+**Branch**: fix/query-trace-observability
+**Created**: 2026-03-28T12:00:00Z
+**Updated**: 2026-03-28T12:00:00Z
+
+## Traceability
+
+**Source**: BUG-013, DEBT-011
+**Source Type**: backlog
+
+## References
+
+### Requirements
+- REQ-060: QueryTrace for RAG observability @.s2s/requirements.md
+- REQ-051: Operator privacy (QueryTrace has no query/response text) @.s2s/requirements.md
+
+### Architecture
+- ARCH-041: Audit/analytics separation via QueryTrace @.s2s/architecture.md
+- ARCH-017: Audit log vs analytics separation @.s2s/architecture.md
+- ARCH-031: Conversation encryption at rest @.s2s/architecture.md
+
+### Decisions
+- ADR-0005: Module boundary enforcement @.s2s/decisions/ADR-0005-module-boundary-enforcement.md
+- ADR-0017: Audit/analytics separation via QueryTrace @.s2s/decisions/ADR-0017-audit-analytics-separation.md
+- ADR-0022: SQLAlchemy 2.0 async with asyncpg @.s2s/decisions/ADR-0022-orm-sqlalchemy-async.md
+
+### Dependencies
+- 20260301-component-analytics.md: AnalyticsService and QueryTraceOrm already implemented
+- 20260301-core-conversations.md: PersistentConversationStore and conversation_turns table
+
+## Overview
+
+`AnalyticsService.store_trace()` exists and is tested but is never called. Both pipelines (Simple + Advanced) build QueryTrace objects in execute() and execute_stream(), but the traces are logged to structlog and discarded. The `query_traces` table is always empty, making post-hoc diagnosis of query behavior impossible.
+
+Additionally, the `response_id` column in `conversation_turns` is never populated (always NULL), and there is no admin API to read decrypted conversation turns. This plan wires existing components together and adds a minimal admin endpoint to close these observability gaps.
+
+## Design notes
+
+### Trace persistence location
+
+Pipelines are framework-agnostic and must not import `vektra_analytics` (ADR-0005). Trace persistence is done in the API layer (vektra-core/api.py and vektra-learn/api.py), which already has access to `app.state.analytics_service` and `app.state.db_session_factory`.
+
+BUG-013 acceptance criteria says "pipeline calls store_trace" but the intent is "traces get persisted after pipeline execution". The API layer is the correct location per module boundaries.
+
+### Streaming path
+
+In streaming mode, the trace is emitted as a QueryChunk with `type="trace"` near the end of the stream. The SSE generators (`_sse_generator` in core, `_learn_sse_generator` in learn) intercept this chunk and persist it. This happens after all tokens have been streamed, so there is no impact on token latency.
+
+### trace_from_dict helper
+
+Both core and learn API layers need to reconstruct a `QueryTrace` from the dict emitted by `_trace_to_dict()`. Since `vektra-learn` cannot import from `vektra_core` (ADR-0005), the reconstruction function lives in `vektra_shared/types.py` next to the `QueryTrace` dataclass.
+
+### Admin endpoint duck-typing
+
+The admin conversation turns endpoint (`vektra-admin`) cannot import from `vektra_core` (ADR-0005). It accesses the conversation store via the registry and uses `hasattr` duck-typing, following the same pattern `vektra-learn` uses for `ensure_conversation`.
+
+### Config flag
+
+`VEKTRA_ANALYTICS_STORE_TRACES` controls trace persistence. `None` (default) means auto-detect: on when `VEKTRA_ENV=development`, off otherwise. Explicit `true`/`false` overrides auto-detection.
+
+### Best-effort semantics
+
+All trace persistence is wrapped in try/except. A database failure must never turn a successful query into a 500 error. Failures are logged as warnings.
+
+## Tasks
+
+### Config and wiring
+- [ ] Add `analytics_store_traces: bool | None` to `ObservabilityConfig` and `VektraSettings` in `vektra-shared/src/vektra_shared/config.py`
+- [ ] Resolve flag at startup in `vektra-app/src/vektra_app/main.py` and set `app.state.store_traces_enabled`
+- [ ] Add `trace_from_dict()` function in `vektra-shared/src/vektra_shared/types.py`
+
+### Trace persistence - core endpoint
+- [ ] Add best-effort `store_trace()` call after `pipeline.execute()` in `vektra-core/src/vektra_core/api.py` (non-streaming path)
+- [ ] Extend `_sse_generator` to accept trace persistence params and persist on `chunk.type == "trace"` (streaming path)
+
+### Trace persistence - learn endpoint
+- [ ] Add best-effort `store_trace()` call after `pipeline.execute()` in `vektra-learn/src/vektra_learn/api.py` (non-streaming path)
+- [ ] Extend `_learn_sse_generator` to accept trace persistence params and persist on `chunk.type == "trace"` (streaming path)
+
+### response_id in conversation turns
+- [ ] Add `response_id: UUID | None = None` keyword-only arg to `ConversationStore` Protocol, `InMemoryConversationStore`, and `PersistentConversationStore` `add_turn()` methods in `vektra-core/src/vektra_core/conversation.py`
+- [ ] Pass `response_id=response_id` in all 4 `add_turn()` call sites: `pipeline.py:561`, `pipeline.py:867`, `advanced_pipeline.py:530`, `advanced_pipeline.py:695`
+
+### Admin conversation turns endpoint
+- [ ] Add `get_turns_detail()` method to `PersistentConversationStore` in `vektra-core/src/vektra_core/conversation.py` (decrypts question/answer, returns full metadata)
+- [ ] Add `GET /api/v1/admin/conversations/{conversation_id}/turns` endpoint in `vektra-admin/src/vektra_admin/api.py` (admin scope, duck-typed access to conversation store)
+
+### Tests
+- [ ] New `vektra-core/tests/test_trace_persistence.py`: store_trace called when enabled, not called when disabled, failure doesn't propagate
+- [ ] Update `vektra-core/tests/test_conversation.py`: add_turn with response_id, get_turns_detail
+- [ ] New `vektra-admin/tests/test_admin_turns.py`: GET endpoint returns decrypted turns, 403 for non-admin
+
+## Acceptance criteria
+
+- [ ] Query traces persisted to DB for all pipelines (simple + advanced, sync + stream) via both `/api/v1/query` and `/api/v1/learn/query`
+- [ ] `response_id` populated in `conversation_turns` on turn save
+- [ ] Admin endpoint to read decrypted conversation turns with full metadata
+- [ ] Trace persistence configurable via `VEKTRA_ANALYTICS_STORE_TRACES` (auto: on in dev, off in prod)
+- [ ] Trace persistence is best-effort (DB failure does not turn a successful query into a 500)
+- [ ] Existing `GET /api/v1/traces/{response_id}` returns persisted traces (no new endpoint needed)
+- [ ] `make lint` and `make test` pass
+
+## Testing approach
+
+1. Unit tests: mock AnalyticsService, verify store_trace called/not-called based on config flag, verify failure isolation
+2. Unit tests: verify response_id written to conversation_turns ORM
+3. Integration tests: admin endpoint returns decrypted conversation content
+4. Manual verification: run stack, execute queries (streaming + non-streaming), verify rows in `query_traces` and `conversation_turns.response_id` via psql
+
+## Integration notes
+
+- No new database migrations needed. `query_traces` table and `conversation_turns.response_id` column already exist (20260301-database-phase2).
+- No new Protocol interfaces. Uses existing `AnalyticsService.store_trace()` and `ConversationStore.add_turn()`.
+- Existing analytics endpoints (`GET /api/v1/traces/{response_id}`, `GET /api/v1/traces`, `GET /api/v1/metrics`) become functional once traces are persisted.

vektra-admin/src/vektra_admin/api.py

@@ -441,6 +441,61 @@ async def revoke_api_key(
 # ---------------------------------------------------------------------------
 
 
+# ---------------------------------------------------------------------------
+# Admin conversation turns (DEBT-011)
+# ---------------------------------------------------------------------------
+
+
+class ConversationTurnDetail(BaseModel):
+    """Decrypted conversation turn with full metadata."""
+
+    turn_number: int
+    question: str
+    answer: str | None
+    response_id: UUID | None
+    model: str | None
+    prompt_tokens: int | None
+    completion_tokens: int | None
+    created_at: datetime
+
+
+@router.get(
+    "/api/v1/admin/conversations/{conversation_id}/turns",
+    response_model=list[ConversationTurnDetail],
+)
+async def get_conversation_turns(
+    conversation_id: UUID,
+    request: Request,
+    _key: ApiKeyInfo = Depends(require_scope("admin")),
+) -> Any:
+    """Return decrypted conversation turns with full metadata (admin only)."""
+    registry = getattr(request.app.state, "registry", None)
+    if registry is None:
+        raise HTTPException(status_code=500, detail="ProviderRegistry not initialized")
+
+    try:
+        conv_store = registry.get("conversation_store", "default")
+    except ValueError:
+        raise HTTPException(status_code=503, detail="Conversation store not available")
+
+    if not hasattr(conv_store, "get_turns_detail"):
+        raise HTTPException(
+            status_code=501,
+            detail="Conversation decryption not available (in-memory store)",
+        )
+
+    turns = await conv_store.get_turns_detail(conversation_id)
+    if turns is None:
+        raise HTTPException(status_code=404, detail="Conversation not found")
+
+    return turns
+
+
+# ---------------------------------------------------------------------------
+# Admin HTML dashboard (REQ-006)
+# ---------------------------------------------------------------------------
+
+
 @router.get("/admin")
 async def admin_dashboard_redirect() -> Response:
     """Redirect legacy /admin to the new HTMX dashboard (ADR-0024)."""

vektra-admin/tests/test_admin_turns.py

@@ -0,0 +1,90 @@
+"""Unit tests for admin conversation turns endpoint (DEBT-011)."""
+
+from unittest.mock import AsyncMock, MagicMock
+from uuid import uuid4
+
+import pytest
+from fastapi import HTTPException
+
+from vektra_admin.api import get_conversation_turns
+
+
+@pytest.mark.asyncio
+async def test_returns_turns_from_persistent_store():
+    """Admin endpoint should return decrypted turns from persistent store."""
+    cid = uuid4()
+    rid = uuid4()
+    turns = [
+        {
+            "turn_number": 1,
+            "question": "What is RAG?",
+            "answer": "Retrieval-augmented generation.",
+            "response_id": rid,
+            "model": "gpt-4o",
+            "prompt_tokens": 150,
+            "completion_tokens": 30,
+            "created_at": "2026-03-28T10:00:00+00:00",
+        }
+    ]
+
+    conv_store = MagicMock()
+    conv_store.get_turns_detail = AsyncMock(return_value=turns)
+
+    registry = MagicMock()
+    registry.get = MagicMock(return_value=conv_store)
+
+    app_state = MagicMock()
+    app_state.registry = registry
+
+    request = MagicMock()
+    request.app.state = app_state
+
+    key_info = MagicMock()
+    key_info.scopes = ["admin"]
+
+    result = await get_conversation_turns(cid, request, key_info)
+    assert result == turns
+    conv_store.get_turns_detail.assert_called_once_with(cid)
+
+
+@pytest.mark.asyncio
+async def test_returns_404_when_conversation_not_found():
+    """Should raise 404 when conversation doesn't exist."""
+    conv_store = MagicMock()
+    conv_store.get_turns_detail = AsyncMock(return_value=None)
+
+    registry = MagicMock()
+    registry.get = MagicMock(return_value=conv_store)
+
+    app_state = MagicMock()
+    app_state.registry = registry
+
+    request = MagicMock()
+    request.app.state = app_state
+
+    key_info = MagicMock()
+
+    with pytest.raises(HTTPException) as exc_info:
+        await get_conversation_turns(uuid4(), request, key_info)
+    assert exc_info.value.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_returns_501_for_inmemory_store():
+    """Should raise 501 when conversation store has no get_turns_detail method."""
+    conv_store = MagicMock(spec=[])  # no methods
+
+    registry = MagicMock()
+    registry.get = MagicMock(return_value=conv_store)
+
+    app_state = MagicMock()
+    app_state.registry = registry
+
+    request = MagicMock()
+    request.app.state = app_state
+
+    key_info = MagicMock()
+
+    with pytest.raises(HTTPException) as exc_info:
+        await get_conversation_turns(uuid4(), request, key_info)
+    assert exc_info.value.status_code == 501

vektra-app/src/vektra_app/main.py

@@ -517,6 +517,13 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
 
     app.state.db_session_factory = _get_sf()
     app.state.analytics_service = registry.get("analytics", "default")
+
+    # Resolve trace persistence flag (DEBT-011)
+    _store = settings.analytics_store_traces
+    if _store is None:
+        _store = settings.env == "development"
+    app.state.store_traces_enabled = _store
+
     if registry.has("learn", "default"):
         app.state.learn_service = registry.get("learn", "default")
         app.state.learn_require_enrollment = settings.learn_require_enrollment

vektra-core/src/vektra_core/advanced_pipeline.py

@@ -528,7 +528,10 @@ async def execute(
         if query.conversation_id is not None:
             try:
                 await self._conversation_store.add_turn(
-                    query.conversation_id, query.question, answer
+                    query.conversation_id,
+                    query.question,
+                    answer,
+                    response_id=response_id,
                 )
             except Exception as exc:
                 log.warning("conversation_turn_store_failed", error=str(exc))
@@ -693,7 +696,10 @@ async def _stream(
         if query.conversation_id is not None and full_answer:
             try:
                 await self._conversation_store.add_turn(
-                    query.conversation_id, query.question, full_answer
+                    query.conversation_id,
+                    query.question,
+                    full_answer,
+                    response_id=response_id,
                 )
             except Exception as exc:
                 log.warning("conversation_turn_store_failed", error=str(exc))