fix: add reader-safe OTel server diagnostics

AGENTS.md

@@ -129,6 +129,18 @@ contextual-orchestrator owns model discovery and selection.
   region-level evidence; never show an internal LLM instruction such as
   `This post is an image` to a buyer.
 
+## Observability boundary
+
+- Follow governance-risk-compliance ADR 0009 and LineageWeave ADR 0122 for
+  OpenTelemetry. Use `OTEL_SERVICE_NAME` and
+  `OTEL_EXPORTER_OTLP_ENDPOINT`; exporting is opt-in and provider-neutral.
+- Correlate one post's HTTP, contextual-orchestrator, and Valkey work with the
+  existing post-scoped session metadata. Do not create an ad hoc session table.
+- Telemetry may contain bounded operation, route-template, service-peer, and
+  correlation attributes, but never post body, prompt, answer, source content,
+  actor or tenant identifiers, credentials, raw stream keys, or provider
+  responses. GRC remains the control/evidence owner.
+
 ## Source parsing and semantic units
 
 - Preserve the source representation and provenance, then derive semantic

CHANGELOG.d/2.13.2-otel-server-diagnostics.md

@@ -0,0 +1,6 @@
+## 2.13.2
+
+- Add reader-safe Global Ask and post-chat failures with bounded OpenTelemetry
+  metrics, traces, and structured server diagnostics for GRC consumption.
+  Failure logs carry the active TraceId and SpanId so another agent can join
+  the Error span to the audit record.

backend/app/activity_stream.py

@@ -14,6 +14,8 @@
 import redis.asyncio as redis
 from fastapi import Request
 
+from lineageweave.observability import traced
+
 
 def create_valkey_client(url: str) -> redis.Redis:
     """One shared async client for the process, mirroring db.create_pool."""
@@ -66,12 +68,16 @@ async def publish_activity_event(
     ``approximate=True``) so one very active post's stream can't grow
     without bound -- the panel only ever shows the most recent 50 anyway.
     """
-    return await client.xadd(
-        _stream_key(post_id),
-        _activity_fields(event_type, actor_account_id, summary),
-        maxlen=1000,
-        approximate=True,
-    )
+    with traced(
+        "lineageweave.valkey.activity_xadd",
+        {"db.system": "redis", "db.operation.name": "xadd", "lineageweave.stream.kind": "activity"},
+    ):
+        return await client.xadd(
+            _stream_key(post_id),
+            _activity_fields(event_type, actor_account_id, summary),
+            maxlen=1000,
+            approximate=True,
+        )
 
 
 def publish_activity_event_sync(
@@ -83,20 +89,32 @@ def publish_activity_event_sync(
 ) -> str | None:
     """Sync ``XADD`` for ``make seed``. Returns None if ``summary`` is already on the stream."""
     key = _stream_key(post_id)
-    existing = client.xrevrange(key, count=50)
+    with traced(
+        "lineageweave.valkey.activity_xrevrange",
+        {"db.system": "redis", "db.operation.name": "xrevrange", "lineageweave.stream.kind": "activity"},
+    ):
+        existing = client.xrevrange(key, count=50)
     if any(fields.get("summary") == summary for _entry_id, fields in existing):
         return None
-    return client.xadd(
-        key,
-        _activity_fields(event_type, str(actor_account_id), summary),
-        maxlen=1000,
-        approximate=True,
-    )
+    with traced(
+        "lineageweave.valkey.activity_xadd",
+        {"db.system": "redis", "db.operation.name": "xadd", "lineageweave.stream.kind": "activity"},
+    ):
+        return client.xadd(
+            key,
+            _activity_fields(event_type, str(actor_account_id), summary),
+            maxlen=1000,
+            approximate=True,
+        )
 
 
 async def read_activity_events(client: redis.Redis, post_id: str, count: int = 50) -> list[dict[str, Any]]:
     """The post's most recent events, newest first."""
-    entries = await client.xrevrange(_stream_key(post_id), count=count)
+    with traced(
+        "lineageweave.valkey.activity_xrevrange",
+        {"db.system": "redis", "db.operation.name": "xrevrange", "lineageweave.stream.kind": "activity"},
+    ):
+        entries = await client.xrevrange(_stream_key(post_id), count=count)
     return [
         {
             "event_id": entry_id,

backend/app/analysis_run_outbox.py

@@ -13,6 +13,8 @@
 
 import redis.asyncio as redis
 
+from lineageweave.observability import traced
+
 OUTBOX_STREAM_KEY = "analysis-run-outbox"
 _CLAIMED = "analysis_outbox_claimed"
 _DELIVERED = "analysis_outbox_delivered"
@@ -69,16 +71,20 @@ async def publish_outbox_event(
     if client is None:
         return None
     try:
-        entry_id = await client.xadd(
-            OUTBOX_STREAM_KEY,
-            outbox_stream_fields(
-                analysis_run_id=analysis_run_id,
-                work_kind_code=work_kind_code,
-                request_sha256=request_sha256,
-            ),
-            maxlen=1000,
-            approximate=True,
-        )
+        with traced(
+            "lineageweave.valkey.analysis_outbox_xadd",
+            {"db.system": "redis", "db.operation.name": "xadd", "lineageweave.stream.kind": "analysis_outbox"},
+        ):
+            entry_id = await client.xadd(
+                OUTBOX_STREAM_KEY,
+                outbox_stream_fields(
+                    analysis_run_id=analysis_run_id,
+                    work_kind_code=work_kind_code,
+                    request_sha256=request_sha256,
+                ),
+                maxlen=1000,
+                approximate=True,
+            )
     except redis.RedisError:
         return None
     return str(entry_id)

backend/app/analysis_run_start.py

@@ -102,7 +102,13 @@ def configured_tepp_client(transport_url: str = "", api_key: str = "") -> TeppCl
     def transport(payload: dict[str, Any]) -> dict[str, Any]:
         try:
             headers = {"authorization": f"Bearer {api_key}"} if api_key.strip() else {}
-            return post_json(url, payload, headers=headers, timeout=30.0)
+            return post_json(
+                url,
+                payload,
+                headers=headers,
+                timeout=30.0,
+                service_peer_name="tepp",
+            )
         except (HttpClientError, OSError, ValueError, TypeError) as exc:
             raise TeppNotAvailable(str(exc)) from exc
 

backend/app/analysis_run_worker.py

@@ -7,16 +7,23 @@
 
 from __future__ import annotations
 
+import asyncio
+import logging
+
 import asyncpg
 import redis.asyncio as redis
 from uuid import UUID
 
 from lineageweave.adjudication_client import AdjudicationClient
+from lineageweave.observability import traced
 from lineageweave.tepp_client import TeppClient
 
 from backend.app.analysis_run_outbox import OUTBOX_STREAM_KEY
 from backend.app.analysis_run_start import deliver_queued_analysis_run
 
+_BROKER_RECOVERY_DELAY_SECONDS = 1.0
+_worker_logger = logging.getLogger(__name__)
+
 
 async def consume_analysis_run_stream_once(
     client: redis.Redis,
@@ -31,36 +38,58 @@ async def consume_analysis_run_stream_once(
     Invalid or stale entries are acknowledged by advancing the cursor; the
     durable PostgreSQL outbox remains available for a later explicit retry.
     """
-    batches = await client.xread({OUTBOX_STREAM_KEY: last_id}, count=10, block=1000)
-    for _stream_name, entries in batches:
-        for entry_id, fields in entries:
-            analysis_run_id = str(fields.get("analysis_run_id", "")).strip()
-            try:
-                UUID(analysis_run_id)
-            except ValueError:
-                analysis_run_id = ""
-            if analysis_run_id:
-                async with pool.acquire() as conn:
-                    async with conn.transaction():
-                        owner = await conn.fetchrow(
-                            """
-                            select requested_by_account_id
-                            from analysis_run
-                            where analysis_run_id = $1::uuid
-                            """,
-                            analysis_run_id,
-                        )
-                        if owner is not None:
-                            await deliver_queued_analysis_run(
-                                conn,
-                                analysis_run_id=analysis_run_id,
-                                account_id=str(owner["requested_by_account_id"]),
-                                affiliated_entity_ids=[],
-                                tepp_client=tepp_client,
-                                adjudication_client=adjudication_client,
-                                valkey_stream_entry_id=str(entry_id),
+    try:
+        batches = await client.xread({OUTBOX_STREAM_KEY: last_id}, count=10, block=1000)
+    except Exception:
+        # Keep idle polls silent, but retain a diagnostic span for broker failures.
+        with traced(
+            "lineageweave.valkey.analysis_outbox_xread",
+            {
+                "db.system": "redis",
+                "db.operation.name": "xread",
+                "lineageweave.stream.kind": "analysis_outbox",
+            },
+        ):
+            raise
+    if not batches:
+        return last_id
+    with traced(
+        "lineageweave.valkey.analysis_outbox_batch",
+        {
+            "db.system": "redis",
+            "db.operation.name": "xread",
+            "lineageweave.stream.kind": "analysis_outbox",
+        },
+    ):
+        for _stream_name, entries in batches:
+            for entry_id, fields in entries:
+                analysis_run_id = str(fields.get("analysis_run_id", "")).strip()
+                try:
+                    UUID(analysis_run_id)
+                except ValueError:
+                    analysis_run_id = ""
+                if analysis_run_id:
+                    async with pool.acquire() as conn:
+                        async with conn.transaction():
+                            owner = await conn.fetchrow(
+                                """
+                                select requested_by_account_id
+                                from analysis_run
+                                where analysis_run_id = $1::uuid
+                                """,
+                                analysis_run_id,
                             )
-            last_id = str(entry_id)
+                            if owner is not None:
+                                await deliver_queued_analysis_run(
+                                    conn,
+                                    analysis_run_id=analysis_run_id,
+                                    account_id=str(owner["requested_by_account_id"]),
+                                    affiliated_entity_ids=[],
+                                    tepp_client=tepp_client,
+                                    adjudication_client=adjudication_client,
+                                    valkey_stream_entry_id=str(entry_id),
+                                )
+                last_id = str(entry_id)
     return last_id
 
 
@@ -74,10 +103,16 @@ async def run_analysis_run_worker(
     """Run the single-process wake-up consumer until task cancellation."""
     last_id = "0-0"
     while True:
-        last_id = await consume_analysis_run_stream_once(
-            client,
-            pool,
-            last_id=last_id,
-            tepp_client=tepp_client,
-            adjudication_client=adjudication_client,
-        )
+        try:
+            last_id = await consume_analysis_run_stream_once(
+                client,
+                pool,
+                last_id=last_id,
+                tepp_client=tepp_client,
+                adjudication_client=adjudication_client,
+            )
+        except (redis.RedisError, OSError) as exc:
+            _worker_logger.warning(
+                "analysis-run Valkey poll failed; retrying (error_type=%s)", type(exc).__name__
+            )
+            await asyncio.sleep(_BROKER_RECOVERY_DELAY_SECONDS)