feat: trace post processing and Valkey sessions

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

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", "lineageweave.work_kind": work_kind_code},
+        ):
+            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_worker.py

@@ -12,6 +12,7 @@
 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
@@ -31,7 +32,15 @@ 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)
+    with traced(
+        "lineageweave.valkey.analysis_outbox_xread",
+        {
+            "db.system": "redis",
+            "db.operation.name": "xread",
+            "lineageweave.stream.kind": "analysis_outbox",
+        },
+    ):
+        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()

backend/app/main.py

@@ -185,6 +185,7 @@
     has_real_source_context,
 )
 from lineageweave.http_client import HttpClientError
+from lineageweave.observability import configure_telemetry
 
 _POST_READ = "post_read"
 _POST_ADMIN = "post_admin"
@@ -194,6 +195,7 @@
 async def lifespan(app: FastAPI):
     """Open one asyncpg pool and one Valkey client for the process, and
     close both on shutdown."""
+    configure_telemetry("lineageweave")
     settings = load_settings()
     app.state.pool = await create_pool(settings.database_url)
     app.state.valkey = create_valkey_client(settings.valkey_url)

backend/app/post_content_queue.py

@@ -10,6 +10,8 @@
 import asyncpg
 import redis.asyncio as redis
 
+from lineageweave.observability import traced
+
 POST_CONTENT_STREAM_KEY = "post-content-ingestion"
 QUEUED = "post_content_ingestion_queued"
 RUNNING = "post_content_ingestion_running"
@@ -126,15 +128,19 @@ async def publish_post_content_event(
     if client is None:
         return None
     try:
-        entry_id = await client.xadd(
-            POST_CONTENT_STREAM_KEY,
-            post_content_stream_fields(
-                post_id=post_id,
-                source_body_digest=source_body_digest,
-            ),
-            maxlen=1000,
-            approximate=True,
-        )
+        with traced(
+            "lineageweave.valkey.post_content_xadd",
+            {"db.system": "redis", "db.operation.name": "xadd", "lineageweave.stream.kind": "post_content"},
+        ):
+            entry_id = await client.xadd(
+                POST_CONTENT_STREAM_KEY,
+                post_content_stream_fields(
+                    post_id=post_id,
+                    source_body_digest=source_body_digest,
+                ),
+                maxlen=1000,
+                approximate=True,
+            )
     except redis.RedisError:
         return None
     return str(entry_id)

backend/app/post_content_worker.py

@@ -16,6 +16,7 @@
 from lineageweave.llm_context import build_post_llm_metadata, use_llm_metadata
 from lineageweave.post_content_normalization import normalize_post_body
 from lineageweave.post_content_persistence import persist_post_content
+from lineageweave.observability import traced
 from lineageweave.post_structure import PostStructureClient
 
 from backend.app.config import load_settings
@@ -42,7 +43,15 @@
 
 async def _stream_tail(client: redis.Redis) -> str:
     """Start after historical wake-ups; the normalized ledger drives recovery."""
-    rows = await client.xrevrange(POST_CONTENT_STREAM_KEY, count=1)
+    with traced(
+        "lineageweave.valkey.post_content_xrevrange",
+        {
+            "db.system": "redis",
+            "db.operation.name": "xrevrange",
+            "lineageweave.stream.kind": "post_content",
+        },
+    ):
+        rows = await client.xrevrange(POST_CONTENT_STREAM_KEY, count=1)
     return str(rows[0][0]) if rows else "0-0"
 
 
@@ -284,7 +293,15 @@ async def consume_post_content_stream_once(
     embedding_factory: Callable[[], EmbeddingClient],
     structure_factory: Callable[[], PostStructureClient],
 ) -> str:
-    batches = await client.xread({POST_CONTENT_STREAM_KEY: last_id}, count=10, block=1000)
+    with traced(
+        "lineageweave.valkey.post_content_xread",
+        {
+            "db.system": "redis",
+            "db.operation.name": "xread",
+            "lineageweave.stream.kind": "post_content",
+        },
+    ):
+        batches = await client.xread({POST_CONTENT_STREAM_KEY: last_id}, count=10, block=1000)
     for _stream_name, entries in batches:
         for entry_id, fields in entries:
             post_id = str(fields.get("post_id", "")).strip()

docker-compose.yml

@@ -117,6 +117,8 @@ services:
       # explicit bounded 8 MiB limit rather than an unbounded request size.
       CONTEXTUAL_ORCHESTRATOR_MAX_BODY_BYTES: ${CONTEXTUAL_ORCHESTRATOR_MAX_BODY_BYTES:-8388608}
       CONTEXTUAL_ORCHESTRATOR_ALLOWED_PROVIDER_HOSTS: ${CONTEXTUAL_ORCHESTRATOR_ALLOWED_PROVIDER_HOSTS:-host.docker.internal}
+      OTEL_SERVICE_NAME: ${OTEL_ORCHESTRATOR_SERVICE_NAME:-contextual-orchestrator}
+      OTEL_EXPORTER_OTLP_ENDPOINT: ${OTEL_EXPORTER_OTLP_ENDPOINT:-}
     command: ["python", "/app/start.py"]
     ports:
       - "${ORCHESTRATOR_PORT:-18000}:8000"
@@ -155,6 +157,8 @@ services:
       OIDC_CLOCK_SKEW_SECONDS: ${OIDC_CLOCK_SKEW_SECONDS:-5}
       FRONTEND_ORIGINS: http://localhost:${FRONTEND_PORT:-15173}
       VALKEY_URL: redis://valkey:6379/0
+      OTEL_SERVICE_NAME: ${OTEL_SERVICE_NAME:-lineageweave}
+      OTEL_EXPORTER_OTLP_ENDPOINT: ${OTEL_EXPORTER_OTLP_ENDPOINT:-}
       # Empty by default: every LLM/vision channel stays the Null client
       # (dropped, not faked). Set these to a running contextual-orchestrator
       # to turn the channels on. Provider credentials use LLM_GATEWAY_API_URL /

docker/contextual-orchestrator/Dockerfile

@@ -5,12 +5,16 @@ WORKDIR /app
 # Reuse the upstream implementation without copying it into LineageWeave.
 # Pin the runtime to a reviewed immutable upstream commit; model selection,
 # structured synthesis, and reasoning policy stay in contextual-orchestrator.
-ADD https://github.com/ContextualWisdomLab/contextual-orchestrator/archive/7df051ac2b929e5910071ac1848d0447c5d6744e.tar.gz /tmp/contextual-orchestrator.tar.gz
+ADD https://github.com/ContextualWisdomLab/contextual-orchestrator/archive/1a40e0f7ad10d1a24137d69d20e44fc9a5dcdd89.tar.gz /tmp/contextual-orchestrator.tar.gz
 RUN mkdir /tmp/contextual-orchestrator \
     && tar -xzf /tmp/contextual-orchestrator.tar.gz --strip-components=1 -C /tmp/contextual-orchestrator \
     && cp -R /tmp/contextual-orchestrator/contextual_orchestrator /app/contextual_orchestrator \
     && cp -R /tmp/contextual-orchestrator/examples /app/examples \
     && rm -rf /tmp/contextual-orchestrator /tmp/contextual-orchestrator.tar.gz \
+    && python -m pip install --no-cache-dir \
+        'opentelemetry-api>=1.30.0' \
+        'opentelemetry-sdk>=1.30.0' \
+        'opentelemetry-exporter-otlp-proto-http>=1.30.0' \
     && useradd --uid 10001 --no-create-home orchestrator
 
 COPY agents.json /app/agents.json

docs/adr/0083-orchestrator-runtime-commit-pin.md

@@ -15,7 +15,7 @@ multi-agent.
 ## Decision
 
 `docker/contextual-orchestrator/Dockerfile` pins the downloaded archive to
-commit `7df051ac2b929e5910071ac1848d0447c5d6744e`. The pin remains explicit
+commit `1a40e0f7ad10d1a24137d69d20e44fc9a5dcdd89`. The pin remains explicit
 and immutable until the reviewed upstream change is superseded; it is not a
 moving `main` reference and it is not a LineageWeave monkey patch.
 

docs/adr/0122-otel-session-observability.md

@@ -0,0 +1,54 @@
+# ADR 0122: Correlate product, orchestrator, and Valkey telemetry by post session
+
+## Status
+
+Accepted.
+
+## Context
+
+Post structure, VISION, OCR, embeddings, summaries, and queue work can run in
+different processes. Existing ADR 0071 already defines a deterministic,
+post-scoped session metadata value, but transport and queue failures were not
+visible as one trace. The organization GRC repository owns the telemetry
+control contract in [ADR 0009](https://github.com/ContextualWisdomLab/governance-risk-compliance/blob/develop/docs/adr/0009-opentelemetry-request-telemetry.md).
+
+## Decision
+
+1. LineageWeave uses the OpenTelemetry Python API and SDK and exports OTLP only
+   when OTEL_EXPORTER_OTLP_ENDPOINT is explicitly configured. The exporter
+   treats that value as a base URL and sends traces to its normalized
+   /v1/traces signal endpoint. The service resource name is lineageweave
+   unless the operator overrides it with the standard OTEL_SERVICE_NAME
+   variable.
+2. Every contextual-orchestrator POST carries the existing
+   `lineageweave_post_session_id` as `X-LineageWeave-Session-Id`. The
+   orchestrator binds it to the request context and adds it to provider spans,
+   so chat, Responses, structured output, VISION, and embedding work for one
+   post can be investigated together.
+3. LineageWeave emits bounded HTTP and Valkey operation spans. Valkey spans
+   identify the operation and logical stream kind, not the stream key, post
+   body, summary, actor, source identifiers, token, or provider response.
+4. Failure logs contain operation, error type, status, and the bounded session
+   correlation only. They do not become a second evidence database. GRC may
+   consume aggregate control evidence and OTLP-derived SLO signals through its
+   existing contracts; LineageWeave does not copy GRC tables or credentials.
+5. No ad hoc session table is introduced. The existing normalized post-scoped
+   session metadata remains the source of correlation.
+
+## Consequences
+
+Operators can follow a slow or failed post-content job from the LineageWeave
+HTTP client through contextual-orchestrator and Valkey without exposing source
+content. An OTLP collector is a deployment concern, not a local default, so a
+developer stack remains usable without a telemetry backend. Raw session IDs
+remain correlation data and must not be used as tenant, actor, or evidence
+labels in GRC dashboards.
+
+## References
+
+OpenTelemetry Authors. (n.d.). *Manual instrumentation with OpenTelemetry
+Python*. Retrieved August 21, 2026, from
+https://opentelemetry.io/docs/languages/python/instrumentation/
+
+OpenTelemetry Authors. (n.d.). *Service semantic conventions*. Retrieved
+August 21, 2026, from https://opentelemetry.io/docs/specs/semconv/registry/attributes/service/

docs/doctoring/OPENTELEMETRY_REFERENCES.md

@@ -0,0 +1,27 @@
+# OpenTelemetry references and implementation traceability
+
+## Normative references
+
+- OpenTelemetry Authors. (n.d.). *Manual instrumentation with OpenTelemetry
+  Python*. Retrieved August 21, 2026, from
+  https://opentelemetry.io/docs/languages/python/instrumentation/
+- OpenTelemetry Authors. (n.d.). *Service semantic conventions*. Retrieved
+  August 21, 2026, from
+  https://opentelemetry.io/docs/specs/semconv/registry/attributes/service/
+- ContextualWisdomLab governance-risk-compliance. (2026). *ADR 0009:
+  Emit bounded OpenTelemetry request telemetry*. Retrieved August 21, 2026,
+  from https://github.com/ContextualWisdomLab/governance-risk-compliance/blob/develop/docs/adr/0009-opentelemetry-request-telemetry.md
+
+## Implementation mapping
+
+| Concern | Implementation | Evidence boundary |
+| --- | --- | --- |
+| Service resource | `OTEL_SERVICE_NAME`, default `lineageweave` | One logical service name per deployment |
+| Post correlation | Existing ADR 0071 session metadata plus `X-LineageWeave-Session-Id` | Correlation only; not identity or authorization |
+| LLM/VISION/embedding transport | `lineageweave.http_client.post_json` | Method, peer, bounded path, status; no body or credential |
+| Valkey queue | `backend/app/*worker.py` and stream producers | Operation and logical stream kind; no stream key or event content |
+| Export | OTEL_EXPORTER_OTLP_ENDPOINT | Disabled by default; base URL normalized to /v1/traces |
+
+The GRC repository remains the organization control and evidence owner. This
+repository emits operational signals and does not copy GRC tables or persist
+provider credentials.