diff --git a/.env.example b/.env.example
index 6578363..5f715e5 100644
--- a/.env.example
+++ b/.env.example
@@ -264,3 +264,13 @@ KC_SMTP_STARTTLS=true
 KC_SMTP_SSL=false
 KC_SMTP_USERNAME=
 KC_SMTP_PASSWORD=
+
+
+# --- Logs to ClickHouse over OTLP (ADR 0007). Empty endpoint = off.
+# Local: http://otel-collector:4318 with `docker compose --profile observability up -d`
+# Shared env: http://clickstack-otel-collector.observability.svc.cluster.local:4318
+OTEL_EXPORTER_OTLP_ENDPOINT=
+# Shared env only: authorization=<hyperdx-ingestion-api-key>
+OTEL_EXPORTER_OTLP_HEADERS=
+# Seconds before an export is dropped
+OTEL_EXPORTER_OTLP_TIMEOUT=5
diff --git a/AGENTS.md b/AGENTS.md
index 66099bb..406cb6e 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -19,6 +19,7 @@ Full narrative docs (read these before deep architectural changes, don't re-deri
 # Run the full stack (api, worker, temporal, minio, qdrant, ui)
 docker compose up -d --build
 docker compose down
+docker compose --profile observability up -d   # + otel-collector & ClickHouse (needs OTEL_EXPORTER_OTLP_ENDPOINT=http://otel-collector:4318)
 
 # Tests (pytest.ini: testpaths=tests, markers unit/integration/slow/api/db/workflow)
 uv run pytest
@@ -77,6 +78,7 @@ Core modules (all under `pipeline/`):
 - `discovery_publish_service.py` — `DiscoveryPublishService`: owns the Publish to Network env vars and the HTTP call. Deliberately Temporal-free
 - `models.py` — Pydantic models, including `DocumentStage` enum and `PIPELINE_STAGES` (the stepper-UI stage list)
 - `config.py` — `Config` dataclass reading env vars, with defaults
+- `observability.py` — bridges root-logger `INFO`+ lines to OTLP when `OTEL_EXPORTER_OTLP_ENDPOINT` is set (ClickHouse/HyperDX); no-op otherwise. See ADR 0007
 - `instances.py` — instance/tenant (state) code ↔ human name mapping; `bv` = Bharat Vistaar platform-wide, others are state codes matching Keycloak group paths (`/states/MH/...` → `mh`)
 - `master_catalog_pg.py`, `scheme_catalog.py` — Postgres-backed master scheme catalog, synced to Redis for the AI layer (bharat-oan-api) to consume
 - `ocr/`, `translation/`, `chunking/`, `vector_store/` — pluggable provider packages, each with a `base.py` protocol/ABC and a `service.py` (or equivalent) that dispatches to the configured provider
diff --git a/ENV.md b/ENV.md
index 00a4e61..beb1e05 100644
--- a/ENV.md
+++ b/ENV.md
@@ -66,6 +66,17 @@ Root `.env` (see `.env.example`). Required by FastAPI (`pipeline/api/`), Tempora
 | `MINIO_BUCKET` | `documents` | Object storage bucket |
 | `DOCUMENT_DB_PATH` | `/data/documents.db` | SQLite path (use `./data/documents.db` for local non-Docker) |
 
+### Logs to ClickHouse (OTLP)
+
+Off unless `OTEL_EXPORTER_OTLP_ENDPOINT` is set. When it is set, API and worker log lines at `INFO` and above (root logger, including `/search` query text) also go over OTLP/HTTP to a collector, which writes them to ClickHouse for HyperDX. See `docs/ADR/0007-ship-logs-over-otlp.md`.
+
+| Variable | Default | Purpose |
+|----------|---------|---------|
+| `OTEL_EXPORTER_OTLP_ENDPOINT` | *(unset — off)* | Collector base URL. Local: `http://otel-collector:4318` with `docker compose --profile observability`. Shared env: the ClickStack collector, e.g. `http://clickstack-otel-collector.observability.svc.cluster.local:4318` |
+| `OTEL_EXPORTER_OTLP_HEADERS` | *(unset)* | Headers sent with each export, e.g. `authorization=<HyperDX ingestion API key>`; ClickStack rejects unauthenticated telemetry |
+| `OTEL_SERVICE_NAME` | `knowledge-provider` | Labels our lines in the shared store. Compose sets `knowledge-provider-api` / `knowledge-provider-worker`; set the same per process in other deployments |
+| `OTEL_EXPORTER_OTLP_TIMEOUT` | `5` (seconds) | How long one export may take before it is dropped; kept under Docker's 10s stop grace so shutdown with the collector down stays clean |
+
 ### API HTTP surface
 
 | Variable | Default | Purpose |
diff --git a/README.md b/README.md
index abe03e6..3534caa 100644
--- a/README.md
+++ b/README.md
@@ -333,6 +333,18 @@ docker compose up -d --build
 docker compose down
 ```
 
+### Logs in ClickHouse (optional)
+
+Set `OTEL_EXPORTER_OTLP_ENDPOINT=http://otel-collector:4318` in `.env`, then:
+
+```bash
+docker compose --profile observability up -d --build
+curl 'http://127.0.0.1:18123/?user=knowledge&password=knowledge' \
+  --data-binary 'SELECT ServiceName, Body FROM otel.otel_logs ORDER BY Timestamp DESC LIMIT 20'
+```
+
+API and worker lines land in `otel.otel_logs`. Shared environments send to the ClickStack collector instead; see `ENV.md` and `docs/ADR/0007-ship-logs-over-otlp.md`.
+
 ### Health Checks
 
 Useful endpoints after startup:
diff --git a/docker-compose.yml b/docker-compose.yml
index 1f6f11c..b649b0e 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -14,6 +14,8 @@
 # Profiles:
 #   (default) — all core services
 #   ops       — also starts temporal-ui and keycloak-ui (already included in keycloak)
+#   observability — also starts otel-collector and clickhouse; set
+#                   OTEL_EXPORTER_OTLP_ENDPOINT=http://otel-collector:4318 in .env
 #
 # Host ports (all bound to 127.0.0.1 for nginx reverse-proxy; override in .env):
 #   API          127.0.0.1:${API_HOST_PORT:-8011}:8001
@@ -23,6 +25,7 @@
 #   MinIO API    127.0.0.1:9000:9000
 #   MinIO UI     127.0.0.1:9001:9001
 #   Temporal UI  127.0.0.1:${TEMPORAL_UI_PORT:-8090}:8080  (profile: ops)
+#   ClickHouse   127.0.0.1:${CLICKHOUSE_HTTP_PORT:-18123}:8123  (profile: observability)
 
 services:
 
@@ -254,6 +257,7 @@ services:
       # HF cache path propagated into the container
       HF_HOME: ${HF_HOME:-/opt/hf-cache}
       HF_HUB_CACHE: ${HF_HOME:-/opt/hf-cache}
+      OTEL_SERVICE_NAME: knowledge-provider-api
     ports:
       - "127.0.0.1:${API_HOST_PORT:-8011}:8001"
     depends_on:
@@ -295,6 +299,7 @@ services:
       ALLOWED_FILE_PATHS: /app/books,/data/documents
       HF_HOME: ${HF_HOME:-/opt/hf-cache}
       HF_HUB_CACHE: ${HF_HOME:-/opt/hf-cache}
+      OTEL_SERVICE_NAME: knowledge-provider-worker
     depends_on:
       temporal:
         condition: service_healthy
@@ -343,6 +348,39 @@ services:
       - api
     restart: unless-stopped
 
+  otel-collector:
+    image: otel/opentelemetry-collector-contrib:${OTEL_COLLECTOR_TAG:-0.161.0}
+    profiles: [observability]
+    command:
+      - --config=/etc/otel/collector.yaml
+    volumes:
+      - ./otel/collector.local.yaml:/etc/otel/collector.yaml:ro
+    environment:
+      CLICKHOUSE_USER: knowledge
+      CLICKHOUSE_PASSWORD: knowledge
+    depends_on:
+      clickhouse:
+        condition: service_healthy
+    restart: unless-stopped
+
+  clickhouse:
+    image: clickhouse/clickhouse-server:${CLICKHOUSE_TAG:-25.12}
+    profiles: [observability]
+    environment:
+      CLICKHOUSE_DB: otel
+      CLICKHOUSE_USER: knowledge
+      CLICKHOUSE_PASSWORD: knowledge
+    volumes:
+      - clickhouse-data:/var/lib/clickhouse
+    ports:
+      - "127.0.0.1:${CLICKHOUSE_HTTP_PORT:-18123}:8123"
+    healthcheck:
+      test: wget --no-verbose --tries=1 --spider http://localhost:8123/ping || exit 1
+      interval: 5s
+      timeout: 5s
+      retries: 10
+    restart: unless-stopped
+
 volumes:
   temporal-db-data:
   qdrant-data:
@@ -351,3 +389,4 @@ volumes:
   minio-data:
   sqlite-data:
   documents-data:
+  clickhouse-data:
diff --git a/docs/ADR/0007-ship-logs-over-otlp.md b/docs/ADR/0007-ship-logs-over-otlp.md
new file mode 100644
index 0000000..2e131c3
--- /dev/null
+++ b/docs/ADR/0007-ship-logs-over-otlp.md
@@ -0,0 +1,26 @@
+---
+status: accepted
+---
+
+# Ship API and worker logs to ClickHouse over OTLP
+
+Context: The API and worker only log to stdout, so a support engineer has to reach into a server to see why a farmer's search went wrong. The decision support system already sends its telemetry to ClickHouse, where HyperDX reads it (engineering-tracker#241).
+
+Decision:
+- The API and worker attach an OpenTelemetry log handler to the root logger when `OTEL_EXPORTER_OTLP_ENDPOINT` is set. With it unset, nothing changes. The handler goes on root because `/search` and `api.py` log through `logging.info(...)` directly, not through module loggers.
+- Only `INFO` and above is exported. The level sits on the handler, so `LOG_LEVEL=DEBUG` still prints locally without shipping debug output.
+- `opentelemetry.*` records are not exported, so the exporter's own failures can't feed back into it.
+- Lines are batched and dropped when the collector is unreachable. Search and uploads never wait on the log store.
+- Endpoint and auth headers come from the standard `OTEL_*` variables. Service name falls back to `knowledge-provider` when `OTEL_SERVICE_NAME` is unset, so lines are never labelled `unknown_service`.
+- An export gives up after 5 seconds unless `OTEL_EXPORTER_OTLP_TIMEOUT` says otherwise, so shutdown with the collector down fits inside Docker's 10-second stop grace.
+- Query text, the raw `/search` payload and the caller's username or email stay in the exported lines, as do email-OTP lines naming an operator's email. Operators are staff, so shipping their identity is accepted. Farmer questions carry no personal data, and a support engineer needs the question to judge a poor answer. Unlike the DSS collector, there is no allowlist.
+- This repo's compose ships a collector and ClickHouse under the `observability` profile, for laptops only. Shared environments point at the existing ClickStack collector instead of running their own.
+
+Consequences:
+- Logs only. Traces and metrics are not emitted.
+- Every API request logs one line on `pipeline.api.requests`: method, path with query string, status and duration, never a body. A pure ASGI middleware times the whole response, so streamed downloads are timed to their last byte and a failure mid-stream is logged.
+- Status below 400 logs at INFO (successful `/health` at DEBUG, since the UI polls it). 4xx logs at WARNING and 5xx at ERROR, with the error detail (validation errors keep only type, field and message, never the submitted value), or the HTTP reason when there is none. A 5xx raised from a caught exception carries that exception's traceback. Unhandled exceptions log at ERROR with their traceback. uvicorn's own access log doesn't reach the root logger and is not shipped.
+- A Temporal worker interceptor (`pipeline/activity_logging.py`) logs every workflow's and activity's start and its completion, failure or cancellation, so new workflows and activities are covered without extra code. Activity failures log at WARNING without a traceback, because each attempt may be retried and Temporal's own `Completing activity as failed` line already carries it. Workflow failures log at ERROR with the traceback.
+- Startup messages from the API and worker are log lines, not `print`s. A failed API startup or a worker crash logs at ERROR with its traceback, and the log buffer is flushed before the process exits.
+- `/search` lines don't carry transaction or message ids yet. That comes with the next story, once callers send them.
+- The shared-environment Helm values (endpoint, ingestion key) are set in the helmcharts repo, not here.
diff --git a/otel/collector.local.yaml b/otel/collector.local.yaml
new file mode 100644
index 0000000..3692218
--- /dev/null
+++ b/otel/collector.local.yaml
@@ -0,0 +1,24 @@
+receivers:
+  otlp:
+    protocols:
+      http:
+        endpoint: 0.0.0.0:4318
+
+processors:
+  batch:
+
+exporters:
+  clickhouse:
+    endpoint: tcp://clickhouse:9000
+    username: ${env:CLICKHOUSE_USER}
+    password: ${env:CLICKHOUSE_PASSWORD}
+    database: otel
+    create_schema: true
+    ttl: 720h
+
+service:
+  pipelines:
+    logs:
+      receivers: [otlp]
+      processors: [batch]
+      exporters: [clickhouse]
diff --git a/pipeline/activity_logging.py b/pipeline/activity_logging.py
new file mode 100644
index 0000000..67ffd9e
--- /dev/null
+++ b/pipeline/activity_logging.py
@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+import time
+from typing import Any
+
+from temporalio import activity, workflow
+from temporalio.worker import (
+    ActivityInboundInterceptor,
+    ExecuteActivityInput,
+    ExecuteWorkflowInput,
+    Interceptor,
+    WorkflowInboundInterceptor,
+    WorkflowInterceptorClassInput,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class ActivityLogging(Interceptor):
+    def intercept_activity(self, next: ActivityInboundInterceptor) -> ActivityInboundInterceptor:
+        return _LogActivity(next)
+
+    def workflow_interceptor_class(
+        self, input: WorkflowInterceptorClassInput
+    ) -> type[WorkflowInboundInterceptor]:
+        return _LogWorkflow
+
+
+class _LogActivity(ActivityInboundInterceptor):
+    async def execute_activity(self, input: ExecuteActivityInput) -> Any:
+        info = activity.info()
+        fields = (info.activity_type, info.workflow_id, info.attempt)
+        logger.info("activity started name=%s workflow_id=%s attempt=%s", *fields)
+        started = time.perf_counter()
+        try:
+            result = await self.next.execute_activity(input)
+        except asyncio.CancelledError:
+            logger.warning(
+                "activity cancelled name=%s workflow_id=%s attempt=%s duration_ms=%d",
+                *fields,
+                _elapsed_ms(started),
+            )
+            raise
+        except Exception as exc:
+            logger.warning(
+                "activity failed name=%s workflow_id=%s attempt=%s duration_ms=%d error=%r",
+                *fields,
+                _elapsed_ms(started),
+                exc,
+            )
+            raise
+        logger.info(
+            "activity completed name=%s workflow_id=%s attempt=%s duration_ms=%d",
+            *fields,
+            _elapsed_ms(started),
+        )
+        return result
+
+
+class _LogWorkflow(WorkflowInboundInterceptor):
+    async def execute_workflow(self, input: ExecuteWorkflowInput) -> Any:
+        info = workflow.info()
+        fields = (info.workflow_type, info.workflow_id, info.run_id)
+        workflow.logger.info("workflow started name=%s workflow_id=%s run_id=%s", *fields)
+        try:
+            result = await self.next.execute_workflow(input)
+        except asyncio.CancelledError:
+            workflow.logger.warning(
+                "workflow cancelled name=%s workflow_id=%s run_id=%s", *fields
+            )
+            raise
+        except Exception as exc:
+            workflow.logger.error(
+                "workflow failed name=%s workflow_id=%s run_id=%s error=%r",
+                *fields,
+                exc,
+                exc_info=True,
+            )
+            raise
+        workflow.logger.info("workflow completed name=%s workflow_id=%s run_id=%s", *fields)
+        return result
+
+
+def _elapsed_ms(started: float) -> int:
+    return int((time.perf_counter() - started) * 1000)
diff --git a/pipeline/api/api.py b/pipeline/api/api.py
index 593f2eb..6b5c5f0 100644
--- a/pipeline/api/api.py
+++ b/pipeline/api/api.py
@@ -101,6 +101,7 @@
 )
 from ..network_constants import RESULT_REJECTED
 from ..network_group import validate_network_provider_config
+from ..observability import configure_logs, reset_logs
 from ..workflows import (
     ChunkingOnlyWorkflow,
     DocumentPipelineWorkflow,
@@ -109,6 +110,7 @@
     ReingestionWorkflow,
     TranslationOnlyWorkflow,
 )
+from . import request_logging
 
 LOG_LEVEL = getattr(logging, os.environ.get("LOG_LEVEL", "INFO").strip().upper(), logging.INFO)
 # uvicorn only configures its own "uvicorn"/"uvicorn.access" loggers, so plain
@@ -132,6 +134,19 @@
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     """Initialize Temporal and MinIO clients on startup."""
+    configure_logs()
+    try:
+        await _start_services()
+    except Exception:
+        logging.exception("api startup failed")
+        reset_logs()
+        raise
+    yield
+    logging.info("api stopping")
+    reset_logs()
+
+
+async def _start_services():
     global temporal_client, minio_client, MINIO_BUCKET
 
     validate_network_provider_config()
@@ -154,12 +169,12 @@ async def lifespan(app: FastAPI):
         )
 
     # Initialize SQLite database
-    print("Initializing SQLite database...")
+    logging.info("api startup: initializing SQLite database")
     db.init_db()
 
     # Temporal
     temporal_host = os.environ.get("TEMPORAL_HOST", "localhost:7233")
-    print(f"Connecting to Temporal at {temporal_host}")
+    logging.info("api startup: connecting to Temporal at %s", temporal_host)
     temporal_client = await Client.connect(temporal_host)
 
     # MinIO - credentials required via environment variables
@@ -171,7 +186,7 @@ async def lifespan(app: FastAPI):
     if not minio_access_key or not minio_secret_key:
         raise RuntimeError("MINIO_ACCESS_KEY and MINIO_SECRET_KEY environment variables are required")
 
-    print(f"Connecting to MinIO at {minio_endpoint}")
+    logging.info("api startup: connecting to MinIO at %s", minio_endpoint)
     minio_client = Minio(
         minio_endpoint,
         access_key=minio_access_key,
@@ -182,10 +197,15 @@ async def lifespan(app: FastAPI):
     # Ensure bucket exists
     if not minio_client.bucket_exists(MINIO_BUCKET):
         minio_client.make_bucket(MINIO_BUCKET)
-        print(f"Created MinIO bucket: {MINIO_BUCKET}")
+        logging.info("api startup: created MinIO bucket %s", MINIO_BUCKET)
 
-    yield
-    # Cleanup if needed
+    logging.info(
+        "api started auth=%s temporal=%s minio=%s bucket=%s",
+        "disabled" if auth_cfg.disabled else "enabled",
+        temporal_host,
+        minio_endpoint,
+        MINIO_BUCKET,
+    )
 
 
 app = FastAPI(
@@ -240,6 +260,7 @@ async def lifespan(app: FastAPI):
     allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"],
     allow_headers=["*"],
 )
+request_logging.install(app)
 
 # Rate limiting configuration
 # Default: 100 requests/minute for general endpoints, 10/minute for uploads
@@ -251,7 +272,7 @@ async def lifespan(app: FastAPI):
 
 limiter = Limiter(key_func=get_remote_address)
 app.state.limiter = limiter
-app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
+app.add_exception_handler(RateLimitExceeded, request_logging.recording(_rate_limit_exceeded_handler))
 
 # Routers split out of this module, one per resource. Register them here.
 from .documents import router as documents_router  # noqa: E402
@@ -557,6 +578,7 @@ def delete_single_chunk_from_index(document_id: str, chunk_num: int, index_name:
         resolved_index = index_name or get_default_index_name()
         return store.delete_chunk(resolved_index, document_id, chunk_num)
     except Exception as e:
+        logging.exception("vector chunk delete failed doc_id=%s chunk=%s", document_id, chunk_num)
         return {"deleted": False, "error": str(e)}
 
 
@@ -572,6 +594,7 @@ def delete_chunks_from_index(document_id: str, index_name: str = "documents-inde
             result["doc_id"] = document_id
         return result
     except Exception as e:
+        logging.exception("vector document delete failed doc_id=%s", document_id)
         return {"deleted": 0, "doc_id": document_id, "error": str(e)}
 
 
@@ -1007,6 +1030,7 @@ def _delete_document_points_from_prod(doc: dict) -> dict:
         store = QdrantVectorStore(client=client)
         return store.delete_by_doc_id(collection, doc_id)
     except Exception as exc:
+        logging.exception("prod vector delete failed doc_id=%s collection=%s", doc_id, collection)
         return {"deleted": False, "error": str(exc), "collection": collection}
 
 
@@ -1036,6 +1060,7 @@ def _delete_scheme_points_from_prod(doc: dict) -> dict:
         store = QdrantVectorStore(client=client)
         return store.delete_by_doc_id(collection, doc_id)
     except Exception as exc:
+        logging.exception("prod vector delete failed doc_id=%s collection=%s", doc_id, collection)
         return {"deleted": False, "error": str(exc), "collection": collection}
 
 
diff --git a/pipeline/api/request_logging.py b/pipeline/api/request_logging.py
new file mode 100644
index 0000000..d5bc3ba
--- /dev/null
+++ b/pipeline/api/request_logging.py
@@ -0,0 +1,114 @@
+from __future__ import annotations
+
+import inspect
+import logging
+import time
+from http import HTTPStatus
+from typing import Any, Callable
+
+from fastapi import FastAPI, Request
+from fastapi.exception_handlers import (
+    http_exception_handler,
+    request_validation_exception_handler,
+)
+from fastapi.exceptions import RequestValidationError
+from starlette.exceptions import HTTPException as StarletteHTTPException
+from starlette.types import ASGIApp, Message, Receive, Scope, Send
+
+logger = logging.getLogger("pipeline.api.requests")
+
+QUIET_PATHS = frozenset({"/health"})
+
+
+def install(app: FastAPI) -> None:
+    app.add_exception_handler(StarletteHTTPException, recording(http_exception_handler))
+    app.add_exception_handler(RequestValidationError, recording(request_validation_exception_handler))
+    app.add_middleware(RequestLogging)
+
+
+def recording(handler: Callable) -> Callable:
+    async def record_then_handle(request: Request, exc: Exception):
+        request.state.error = _detail(exc)
+        request.state.error_cause = exc.__cause__
+        result = handler(request, exc)
+        return await result if inspect.isawaitable(result) else result
+
+    return record_then_handle
+
+
+def _detail(exc: Exception) -> Any:
+    if isinstance(exc, RequestValidationError):
+        return [{"type": e.get("type"), "loc": e.get("loc"), "msg": e.get("msg")} for e in exc.errors()]
+    return getattr(exc, "detail", None)
+
+
+class RequestLogging:
+    def __init__(self, app: ASGIApp) -> None:
+        self.app = app
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        if scope["type"] != "http":
+            await self.app(scope, receive, send)
+            return
+
+        started = time.perf_counter()
+        status: dict[str, int] = {}
+
+        async def send_and_watch(message: Message) -> None:
+            if message["type"] == "http.response.start":
+                status["code"] = message["status"]
+            await send(message)
+
+        try:
+            await self.app(scope, receive, send_and_watch)
+        except Exception as exc:
+            logger.error(
+                "request method=%s uri=%s status=%s duration_ms=%d error=%r",
+                scope["method"],
+                _uri(scope),
+                status.get("code", 500),
+                _elapsed_ms(started),
+                exc,
+                exc_info=True,
+            )
+            raise
+
+        _log_response(scope, status.get("code", 500), _elapsed_ms(started))
+
+
+def _log_response(scope: Scope, code: int, duration_ms: int) -> None:
+    method, uri = scope["method"], _uri(scope)
+    if code < 400:
+        level = logging.DEBUG if scope["path"] in QUIET_PATHS else logging.INFO
+        logger.log(level, "request method=%s uri=%s status=%s duration_ms=%d", method, uri, code, duration_ms)
+        return
+
+    state = scope.get("state") or {}
+    error = state.get("error") or _reason(code)
+    cause = state.get("error_cause")
+    logger.log(
+        logging.ERROR if code >= 500 else logging.WARNING,
+        "request method=%s uri=%s status=%s duration_ms=%d error=%r",
+        method,
+        uri,
+        code,
+        duration_ms,
+        error,
+        exc_info=(type(cause), cause, cause.__traceback__) if code >= 500 and cause else None,
+    )
+
+
+def _uri(scope: Scope) -> str:
+    query = scope.get("query_string", b"").decode("latin-1")
+    return f"{scope['path']}?{query}" if query else scope["path"]
+
+
+def _reason(code: int) -> str:
+    try:
+        return HTTPStatus(code).phrase
+    except ValueError:
+        return str(code)
+
+
+def _elapsed_ms(started: float) -> int:
+    return int((time.perf_counter() - started) * 1000)
diff --git a/pipeline/observability.py b/pipeline/observability.py
new file mode 100644
index 0000000..6a58f67
--- /dev/null
+++ b/pipeline/observability.py
@@ -0,0 +1,89 @@
+from __future__ import annotations
+
+import logging
+import os
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from opentelemetry.sdk._logs import LoggerProvider
+
+_MARKER = "_knowledge_provider_otel_handler"
+_TIMEOUT_ENV = ("OTEL_EXPORTER_OTLP_LOGS_TIMEOUT", "OTEL_EXPORTER_OTLP_TIMEOUT")
+
+DEFAULT_SERVICE_NAME = "knowledge-provider"
+DEFAULT_EXPORT_TIMEOUT_SECONDS = 5.0
+
+_provider: LoggerProvider | None = None
+_provider_is_ours = False
+
+
+_NOT_EXPORTED = ("opentelemetry",)
+
+
+class _SkipUnexported(logging.Filter):
+    def filter(self, record: logging.LogRecord) -> bool:
+        return not record.name.startswith(_NOT_EXPORTED)
+
+
+def is_otel_handler(handler: logging.Handler) -> bool:
+    return getattr(handler, _MARKER, False)
+
+
+def _resource():
+    from opentelemetry.sdk.resources import SERVICE_NAME, Resource
+
+    name = os.environ.get("OTEL_SERVICE_NAME", "").strip() or DEFAULT_SERVICE_NAME
+    return Resource.create({SERVICE_NAME: name})
+
+
+def _export_timeout() -> float | None:
+    if any(os.environ.get(name, "").strip() for name in _TIMEOUT_ENV):
+        return None
+    return DEFAULT_EXPORT_TIMEOUT_SECONDS
+
+
+def configure_logs(*, logger_provider: LoggerProvider | None = None) -> None:
+    global _provider, _provider_is_ours
+
+    _detach()
+
+    ours = logger_provider is None
+    if ours:
+        if not os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT", "").strip():
+            return
+
+        from opentelemetry.exporter.otlp.proto.http._log_exporter import OTLPLogExporter
+        from opentelemetry.sdk._logs import LoggerProvider as SdkLoggerProvider
+        from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
+
+        logger_provider = SdkLoggerProvider(resource=_resource())
+        logger_provider.add_log_record_processor(
+            BatchLogRecordProcessor(OTLPLogExporter(timeout=_export_timeout()))
+        )
+
+    from opentelemetry.instrumentation.logging.handler import LoggingHandler
+
+    handler = LoggingHandler(level=logging.INFO, logger_provider=logger_provider)
+    handler.addFilter(_SkipUnexported())
+    setattr(handler, _MARKER, True)
+    logging.getLogger().addHandler(handler)
+
+    _provider = logger_provider
+    _provider_is_ours = ours
+
+
+def reset_logs() -> None:
+    global _provider, _provider_is_ours
+
+    _detach()
+    _provider = None
+    _provider_is_ours = False
+
+
+def _detach() -> None:
+    root = logging.getLogger()
+    for handler in [h for h in root.handlers if is_otel_handler(h)]:
+        root.removeHandler(handler)
+
+    if _provider is not None and _provider_is_ours:
+        _provider.shutdown()
diff --git a/pipeline/worker.py b/pipeline/worker.py
index 6cd996b..438e36f 100644
--- a/pipeline/worker.py
+++ b/pipeline/worker.py
@@ -26,7 +26,9 @@
     run_ocr_and_store,
     update_document_state,
 )
+from .activity_logging import ActivityLogging
 from .network_group import validate_network_provider_config
+from .observability import configure_logs, reset_logs
 from .workflows import (
     ChunkingOnlyWorkflow,
     DocumentPipelineWorkflow,
@@ -50,26 +52,38 @@
 logging.getLogger("temporalio").setLevel(logging.INFO)
 
 TASK_QUEUE = "ocr-pipeline"
+INTERCEPTORS = [ActivityLogging()]
 
 
 async def main():
     """Start the worker."""
+    configure_logs()
+    try:
+        await _run()
+    except Exception:
+        logging.exception("worker crashed")
+        raise
+    finally:
+        logging.info("worker stopping")
+        reset_logs()
+
+
+async def _run():
     validate_network_provider_config()
 
     translation_provider = os.environ.get("TRANSLATION_PROVIDER", "gemma_vllm").strip().lower()
     if translation_provider in {"gemma_vllm", "gemma4", "gemma"}:
         if not os.environ.get("TRANSLATION_VLLM_BASE_URL", "").strip():
-            print("Error: TRANSLATION_VLLM_BASE_URL not set")
+            logging.error("worker startup: TRANSLATION_VLLM_BASE_URL not set, exiting")
             return
 
     # Initialize SQLite database
-    print("Initializing SQLite database...")
+    logging.info("worker startup: initializing SQLite database")
     db.init_db()
 
     temporal_host = os.environ.get("TEMPORAL_HOST", "localhost:7233")
     max_concurrent_activities = int(os.environ.get("TEMPORAL_MAX_CONCURRENT_ACTIVITIES", "4"))
-    print(f"Connecting to Temporal at {temporal_host}")
-    print(f"Worker activity concurrency: {max_concurrent_activities}")
+    logging.info("worker startup: connecting to Temporal at %s", temporal_host)
 
     client = await Client.connect(temporal_host)
 
@@ -77,6 +91,7 @@ async def main():
         client,
         task_queue=TASK_QUEUE,
         max_concurrent_activities=max_concurrent_activities,
+        interceptors=INTERCEPTORS,
         workflows=[
             DocumentPipelineWorkflow,
             ReingestionWorkflow,
@@ -103,8 +118,12 @@ async def main():
         ],
     )
 
-    print(f"Worker started on queue: {TASK_QUEUE}")
-    print("Waiting for workflows...")
+    logging.info(
+        "worker started queue=%s temporal=%s activity_concurrency=%s",
+        TASK_QUEUE,
+        temporal_host,
+        max_concurrent_activities,
+    )
     await worker.run()
 
 
diff --git a/pyproject.toml b/pyproject.toml
index ca47fb0..833e675 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -59,6 +59,10 @@ dependencies = [
 
     # Auth (Keycloak JWT validation)
     "PyJWT[crypto]>=2.8.0",
+
+    "opentelemetry-sdk>=1.44.0",
+    "opentelemetry-exporter-otlp-proto-http>=1.44.0",
+    "opentelemetry-instrumentation-logging>=0.65b0",
 ]
 
 [dependency-groups]
diff --git a/tests/test_activity_logging.py b/tests/test_activity_logging.py
new file mode 100644
index 0000000..ac93276
--- /dev/null
+++ b/tests/test_activity_logging.py
@@ -0,0 +1,129 @@
+import asyncio
+import logging
+import uuid
+from datetime import timedelta
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from temporalio import activity, workflow
+from temporalio.exceptions import ApplicationError
+from temporalio.testing import ActivityEnvironment, WorkflowEnvironment
+from temporalio.worker import Worker
+
+from pipeline import activity_logging
+
+LOGGER = "pipeline.activity_logging"
+
+
+def _interceptor(next_result=None, error=None):
+    following = MagicMock()
+    following.execute_activity = AsyncMock(return_value=next_result, side_effect=error)
+    return activity_logging.ActivityLogging().intercept_activity(following)
+
+
+def _messages(caplog, name=LOGGER):
+    return [(r.levelno, r.getMessage()) for r in caplog.records if r.name == name]
+
+
+@pytest.mark.asyncio
+async def test_activity_logs_start_and_completion(caplog):
+    interceptor = _interceptor(next_result="done")
+
+    with caplog.at_level(logging.INFO, logger=LOGGER):
+        result = await ActivityEnvironment().run(interceptor.execute_activity, MagicMock())
+
+    assert result == "done"
+    [(start_level, start), (end_level, end)] = _messages(caplog)
+    assert start_level == end_level == logging.INFO
+    assert start.startswith("activity started name=unknown")
+    assert "workflow_id=test" in start and "attempt=1" in start
+    assert end.startswith("activity completed name=unknown") and "duration_ms=" in end
+
+
+@pytest.mark.asyncio
+async def test_activity_failure_is_logged_and_reraised(caplog):
+    interceptor = _interceptor(error=ValueError("bad page"))
+
+    with caplog.at_level(logging.INFO, logger=LOGGER):
+        with pytest.raises(ValueError):
+            await ActivityEnvironment().run(interceptor.execute_activity, MagicMock())
+
+    [_, (level, failed)] = _messages(caplog)
+    assert level == logging.WARNING
+    assert failed.startswith("activity failed name=unknown")
+    assert "error=ValueError('bad page')" in failed
+
+
+@pytest.mark.asyncio
+async def test_activity_cancellation_logs_an_end(caplog):
+    interceptor = _interceptor(error=asyncio.CancelledError())
+
+    with caplog.at_level(logging.INFO, logger=LOGGER):
+        with pytest.raises(asyncio.CancelledError):
+            await ActivityEnvironment().run(interceptor.execute_activity, MagicMock())
+
+    [_, (level, cancelled)] = _messages(caplog)
+    assert level == logging.WARNING
+    assert cancelled.startswith("activity cancelled name=unknown")
+
+
+@activity.defn
+async def _echo(value: str) -> str:
+    return value
+
+
+@workflow.defn(sandboxed=False)
+class _Succeeds:
+    @workflow.run
+    async def run(self) -> str:
+        return await workflow.execute_activity(_echo, "ok", start_to_close_timeout=timedelta(seconds=10))
+
+
+@workflow.defn
+class _Fails:
+    @workflow.run
+    async def run(self) -> None:
+        raise ApplicationError("chunking rejected", non_retryable=True)
+
+
+async def _run_workflow(workflow_cls):
+    async with await WorkflowEnvironment.start_time_skipping() as env:
+        async with Worker(
+            env.client,
+            task_queue="logging-test",
+            workflows=[workflow_cls],
+            activities=[_echo],
+            interceptors=[activity_logging.ActivityLogging()],
+        ):
+            return await env.client.execute_workflow(
+                workflow_cls.run, id=f"wf-{uuid.uuid4()}", task_queue="logging-test"
+            )
+
+
+@pytest.mark.asyncio
+@pytest.mark.workflow
+async def test_workflow_and_its_activities_log_start_and_completion(caplog):
+    with caplog.at_level(logging.INFO):
+        assert await _run_workflow(_Succeeds) == "ok"
+
+    lines = [r.getMessage() for r in caplog.records]
+    for prefix in (
+        "workflow started name=_Succeeds",
+        "activity started name=_echo",
+        "activity completed name=_echo",
+        "workflow completed name=_Succeeds",
+    ):
+        assert any(line.startswith(prefix) for line in lines), prefix
+
+
+@pytest.mark.asyncio
+@pytest.mark.workflow
+async def test_workflow_failure_is_logged_with_its_error(caplog):
+    with caplog.at_level(logging.INFO):
+        with pytest.raises(Exception):
+            await _run_workflow(_Fails)
+
+    [failed] = [r for r in caplog.records if r.getMessage().startswith("workflow failed name=_Fails")]
+    assert failed.levelno == logging.ERROR
+    assert "chunking rejected" in failed.getMessage()
+    assert failed.exc_info is not None
diff --git a/tests/test_observability.py b/tests/test_observability.py
new file mode 100644
index 0000000..d63d951
--- /dev/null
+++ b/tests/test_observability.py
@@ -0,0 +1,146 @@
+import logging
+
+import pytest
+from opentelemetry.sdk._logs import LoggerProvider
+from opentelemetry.sdk._logs.export import InMemoryLogRecordExporter, SimpleLogRecordProcessor
+
+from pipeline import observability
+
+
+@pytest.fixture
+def exported(monkeypatch):
+    monkeypatch.setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://collector:4318")
+    exporter = InMemoryLogRecordExporter()
+    provider = LoggerProvider()
+    provider.add_log_record_processor(SimpleLogRecordProcessor(exporter))
+    root = logging.getLogger()
+    previous_level = root.level
+    root.setLevel(logging.DEBUG)
+    observability.configure_logs(logger_provider=provider)
+    yield exporter
+    observability.reset_logs()
+    root.setLevel(previous_level)
+
+
+def _bodies(exporter):
+    return [item.log_record.body for item in exporter.get_finished_logs()]
+
+
+def _our_handlers():
+    return [h for h in logging.getLogger().handlers if observability.is_otel_handler(h)]
+
+
+def test_nothing_is_attached_without_an_endpoint(monkeypatch):
+    monkeypatch.delenv("OTEL_EXPORTER_OTLP_ENDPOINT", raising=False)
+
+    observability.configure_logs()
+
+    assert _our_handlers() == []
+
+
+def test_an_endpoint_attaches_one_handler(monkeypatch):
+    monkeypatch.setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://collector:4318")
+
+    observability.configure_logs()
+    try:
+        assert len(_our_handlers()) == 1
+    finally:
+        observability.reset_logs()
+
+
+def test_info_lines_from_root_and_module_loggers_are_exported(exported):
+    logging.info("search request query=%r", "cotton pest")
+    logging.getLogger("pipeline.activities").warning("ocr retry")
+
+    assert _bodies(exported) == ["search request query='cotton pest'", "ocr retry"]
+
+
+def test_exported_line_keeps_its_level(exported):
+    logging.getLogger("pipeline.activities").error("ingest failed")
+
+    [item] = exported.get_finished_logs()
+    assert item.log_record.severity_text == "ERROR"
+
+
+def test_debug_lines_are_not_exported(exported):
+    logging.getLogger("pipeline.activities").debug("page text")
+
+    assert _bodies(exported) == []
+
+
+def test_opentelemetry_own_lines_are_not_exported(exported):
+    logging.getLogger("opentelemetry.exporter.otlp").warning("export failed")
+
+    assert _bodies(exported) == []
+
+
+def test_configuring_twice_keeps_one_handler(exported):
+    observability.configure_logs(logger_provider=LoggerProvider())
+
+    assert len(_our_handlers()) == 1
+
+
+def test_reset_removes_the_handler(exported):
+    observability.reset_logs()
+
+    assert _our_handlers() == []
+
+
+@pytest.fixture
+def built_provider(monkeypatch):
+    monkeypatch.setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://collector:4318")
+    yield lambda: observability._provider
+    observability.reset_logs()
+
+
+def test_lines_are_labelled_knowledge_provider_without_a_service_name(monkeypatch, built_provider):
+    monkeypatch.delenv("OTEL_SERVICE_NAME", raising=False)
+
+    observability.configure_logs()
+
+    assert built_provider().resource.attributes["service.name"] == "knowledge-provider"
+
+
+def test_service_name_from_env_wins(monkeypatch, built_provider):
+    monkeypatch.setenv("OTEL_SERVICE_NAME", "knowledge-provider-worker")
+
+    observability.configure_logs()
+
+    assert built_provider().resource.attributes["service.name"] == "knowledge-provider-worker"
+
+
+def test_export_gives_up_within_five_seconds_by_default(monkeypatch):
+    monkeypatch.delenv("OTEL_EXPORTER_OTLP_TIMEOUT", raising=False)
+    monkeypatch.delenv("OTEL_EXPORTER_OTLP_LOGS_TIMEOUT", raising=False)
+
+    assert observability._export_timeout() == 5.0
+
+
+def test_export_timeout_from_env_is_left_to_the_sdk(monkeypatch):
+    monkeypatch.setenv("OTEL_EXPORTER_OTLP_TIMEOUT", "2")
+
+    assert observability._export_timeout() is None
+
+
+def test_reset_leaves_an_injected_provider_running(monkeypatch):
+    monkeypatch.setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://collector:4318")
+    exporter = InMemoryLogRecordExporter()
+    provider = LoggerProvider()
+    provider.add_log_record_processor(SimpleLogRecordProcessor(exporter))
+    observability.configure_logs(logger_provider=provider)
+
+    observability.reset_logs()
+    provider.get_logger("caller").emit(body="still mine")
+
+    assert [item.log_record.body for item in exporter.get_finished_logs()] == ["still mine"]
+
+
+def test_reset_shuts_down_a_provider_it_built(built_provider):
+    observability.configure_logs()
+    provider = built_provider()
+    shutdowns = []
+    provider.shutdown = lambda: shutdowns.append(True)
+
+    observability.reset_logs()
+
+    assert shutdowns == [True]
diff --git a/tests/test_observability_wiring.py b/tests/test_observability_wiring.py
new file mode 100644
index 0000000..199815d
--- /dev/null
+++ b/tests/test_observability_wiring.py
@@ -0,0 +1,111 @@
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from pipeline import worker
+from pipeline.api import api
+
+
+@pytest.fixture
+def calls():
+    return []
+
+
+def test_worker_configures_logs_and_resets_them_on_exit(monkeypatch, calls):
+    monkeypatch.setattr(worker, "configure_logs", lambda: calls.append("configure"))
+    monkeypatch.setattr(worker, "reset_logs", lambda: calls.append("reset"))
+    monkeypatch.setattr(worker, "_run", AsyncMock(side_effect=lambda: calls.append("run")))
+
+    asyncio.run(worker.main())
+
+    assert calls == ["configure", "run", "reset"]
+
+
+def test_worker_resets_logs_when_it_fails(monkeypatch, calls):
+    monkeypatch.setattr(worker, "configure_logs", lambda: calls.append("configure"))
+    monkeypatch.setattr(worker, "reset_logs", lambda: calls.append("reset"))
+    monkeypatch.setattr(worker, "_run", AsyncMock(side_effect=RuntimeError("temporal down")))
+
+    with pytest.raises(RuntimeError):
+        asyncio.run(worker.main())
+
+    assert calls == ["configure", "reset"]
+
+
+def test_api_lifespan_configures_logs_first_and_resets_them_on_shutdown(monkeypatch, calls, caplog):
+    monkeypatch.setattr(api, "configure_logs", lambda: calls.append("configure"))
+    monkeypatch.setattr(api, "reset_logs", lambda: calls.append("reset"))
+    monkeypatch.setattr(api, "validate_network_provider_config", lambda: calls.append("validate"))
+    monkeypatch.setattr(api, "load_auth_config", lambda: SimpleNamespace(disabled=True))
+    monkeypatch.setattr(api, "validate_auth_config", lambda _cfg: None)
+    monkeypatch.setattr(api.db, "init_db", lambda: None)
+    monkeypatch.setattr(api.Client, "connect", AsyncMock(return_value=MagicMock()))
+    monkeypatch.setattr(api, "Minio", lambda *a, **k: MagicMock(bucket_exists=lambda _b: True))
+
+    async def run():
+        async with api.lifespan(api.app):
+            calls.append("serving")
+
+    monkeypatch.setenv("MINIO_ACCESS_KEY", "key")
+    monkeypatch.setenv("MINIO_SECRET_KEY", "secret")
+
+    with caplog.at_level("INFO"):
+        asyncio.run(run())
+
+    assert calls == ["configure", "validate", "serving", "reset"]
+    assert any(r.getMessage().startswith("api started auth=disabled") for r in caplog.records)
+
+
+def test_worker_crash_is_logged_with_its_traceback(monkeypatch, calls, caplog):
+    monkeypatch.setattr(worker, "configure_logs", lambda: None)
+    monkeypatch.setattr(worker, "reset_logs", lambda: calls.append("reset"))
+    monkeypatch.setattr(worker, "_run", AsyncMock(side_effect=ConnectionError("temporal down")))
+
+    with caplog.at_level("INFO"), pytest.raises(ConnectionError):
+        asyncio.run(worker.main())
+
+    [crash] = [r for r in caplog.records if r.getMessage() == "worker crashed"]
+    assert crash.levelname == "ERROR" and isinstance(crash.exc_info[1], ConnectionError)
+    assert calls == ["reset"]
+
+
+def test_worker_logs_startup_and_hands_its_interceptors_to_temporal(monkeypatch, caplog):
+    built = {}
+
+    class FakeWorker:
+        def __init__(self, *args, **kwargs):
+            built.update(kwargs)
+
+        async def run(self):
+            return None
+
+    monkeypatch.setenv("TRANSLATION_PROVIDER", "none")
+    monkeypatch.setattr(worker, "validate_network_provider_config", lambda: None)
+    monkeypatch.setattr(worker.db, "init_db", lambda: None)
+    monkeypatch.setattr(worker.Client, "connect", AsyncMock(return_value=MagicMock()))
+    monkeypatch.setattr(worker, "Worker", FakeWorker)
+
+    with caplog.at_level("INFO"):
+        asyncio.run(worker._run())
+
+    assert built["interceptors"] is worker.INTERCEPTORS
+    assert any(r.getMessage().startswith("worker started queue=ocr-pipeline") for r in caplog.records)
+
+
+def test_api_startup_failure_is_logged_and_flushed(monkeypatch, calls, caplog):
+    monkeypatch.setattr(api, "configure_logs", lambda: None)
+    monkeypatch.setattr(api, "reset_logs", lambda: calls.append("reset"))
+    monkeypatch.setattr(api, "_start_services", AsyncMock(side_effect=RuntimeError("MINIO keys missing")))
+
+    async def run():
+        async with api.lifespan(api.app):
+            calls.append("serving")
+
+    with caplog.at_level("INFO"), pytest.raises(RuntimeError):
+        asyncio.run(run())
+
+    [failed] = [r for r in caplog.records if r.getMessage() == "api startup failed"]
+    assert failed.levelname == "ERROR" and isinstance(failed.exc_info[1], RuntimeError)
+    assert calls == ["reset"]
diff --git a/tests/test_request_logging.py b/tests/test_request_logging.py
new file mode 100644
index 0000000..d594a07
--- /dev/null
+++ b/tests/test_request_logging.py
@@ -0,0 +1,178 @@
+import logging
+
+import pytest
+from fastapi import FastAPI, HTTPException, Request
+from fastapi.responses import StreamingResponse
+from fastapi.testclient import TestClient
+from pydantic import BaseModel
+from slowapi import Limiter, _rate_limit_exceeded_handler
+from slowapi.errors import RateLimitExceeded
+from slowapi.util import get_remote_address
+
+from pipeline.api import request_logging
+
+LOGGER = "pipeline.api.requests"
+
+
+class Upload(BaseModel):
+    pages: int
+
+
+@pytest.fixture
+def client():
+    app = FastAPI()
+    limiter = Limiter(key_func=get_remote_address)
+    app.state.limiter = limiter
+    app.add_exception_handler(RateLimitExceeded, request_logging.recording(_rate_limit_exceeded_handler))
+    request_logging.install(app)
+
+    @app.get("/ok")
+    def ok():
+        return {"secret": "body"}
+
+    @app.get("/health")
+    def health():
+        return {"status": "ok"}
+
+    @app.get("/missing")
+    def missing():
+        raise HTTPException(404, "Document not found")
+
+    @app.get("/boom")
+    def boom():
+        raise RuntimeError("database locked")
+
+    @app.get("/wrapped")
+    def wrapped():
+        try:
+            raise ConnectionError("qdrant unreachable")
+        except ConnectionError as exc:
+            raise HTTPException(503, "Search backend unavailable") from exc
+
+    @app.post("/upload")
+    def upload(body: Upload):
+        return body
+
+    @app.get("/limited")
+    @limiter.limit("1/minute")
+    def limited(request: Request):
+        return {"ok": True}
+
+    @app.get("/stream")
+    def stream():
+        def chunks():
+            yield b"first"
+            raise RuntimeError("minio dropped")
+
+        return StreamingResponse(chunks())
+
+    return TestClient(app, raise_server_exceptions=False)
+
+
+def _lines(caplog):
+    return [r for r in caplog.records if r.name == LOGGER]
+
+
+def test_success_logs_method_uri_and_status_without_body(client, caplog):
+    with caplog.at_level(logging.INFO, logger=LOGGER):
+        client.get("/ok?index=documents")
+
+    [record] = _lines(caplog)
+    message = record.getMessage()
+    assert record.levelno == logging.INFO
+    assert "method=GET uri=/ok?index=documents status=200" in message
+    assert "duration_ms=" in message
+    assert "secret" not in message
+
+
+def test_successful_health_checks_stay_at_debug(client, caplog):
+    with caplog.at_level(logging.DEBUG, logger=LOGGER):
+        client.get("/health")
+
+    [record] = _lines(caplog)
+    assert record.levelno == logging.DEBUG
+
+
+def test_client_error_logs_warning_with_detail(client, caplog):
+    with caplog.at_level(logging.INFO, logger=LOGGER):
+        response = client.get("/missing")
+
+    assert response.json() == {"detail": "Document not found"}
+    [record] = _lines(caplog)
+    assert record.levelno == logging.WARNING
+    assert "uri=/missing status=404" in record.getMessage()
+    assert "error='Document not found'" in record.getMessage()
+
+
+def test_validation_error_logs_where_and_why_but_not_the_submitted_value(client, caplog):
+    with caplog.at_level(logging.INFO, logger=LOGGER):
+        response = client.post("/upload", json={"pages": "farmer-typed-value"})
+
+    assert response.status_code == 422
+    [record] = _lines(caplog)
+    assert "status=422" in record.getMessage()
+    assert "int_parsing" in record.getMessage() and "pages" in record.getMessage()
+    assert "farmer-typed-value" not in record.getMessage()
+
+
+def test_unknown_route_logs_404(client, caplog):
+    with caplog.at_level(logging.INFO, logger=LOGGER):
+        client.get("/nowhere")
+
+    [record] = _lines(caplog)
+    assert "uri=/nowhere status=404" in record.getMessage()
+    assert "error='Not Found'" in record.getMessage()
+
+
+def test_rate_limited_request_logs_429_with_its_limit(client, caplog):
+    client.get("/limited")
+    with caplog.at_level(logging.INFO, logger=LOGGER):
+        response = client.get("/limited")
+
+    assert response.status_code == 429
+    [record] = _lines(caplog)
+    assert "status=429" in record.getMessage()
+    assert "1 per 1 minute" in record.getMessage()
+
+
+def test_server_error_from_a_caught_exception_keeps_its_cause(client, caplog):
+    with caplog.at_level(logging.INFO, logger=LOGGER):
+        response = client.get("/wrapped")
+
+    assert response.status_code == 503
+    [record] = _lines(caplog)
+    assert record.levelno == logging.ERROR
+    assert "error='Search backend unavailable'" in record.getMessage()
+    assert isinstance(record.exc_info[1], ConnectionError)
+
+
+def test_unhandled_exception_logs_500_with_traceback(client, caplog):
+    with caplog.at_level(logging.INFO, logger=LOGGER):
+        response = client.get("/boom")
+
+    assert response.status_code == 500
+    [record] = _lines(caplog)
+    assert record.levelno == logging.ERROR
+    assert "uri=/boom status=500" in record.getMessage()
+    assert "error=RuntimeError('database locked')" in record.getMessage()
+    assert record.exc_info is not None
+
+
+def test_failure_mid_stream_is_logged(client, caplog):
+    with caplog.at_level(logging.INFO, logger=LOGGER):
+        try:
+            client.get("/stream")
+        except Exception:
+            pass
+
+    [record] = _lines(caplog)
+    assert record.levelno == logging.ERROR
+    assert "uri=/stream status=200" in record.getMessage()
+    assert "error=RuntimeError('minio dropped')" in record.getMessage()
+
+
+def test_the_app_logs_its_requests(test_client, caplog):
+    with caplog.at_level(logging.INFO, logger=LOGGER):
+        test_client.get("/documents/summary")
+
+    assert any("uri=/documents/summary status=" in r.getMessage() for r in _lines(caplog))
diff --git a/tests/test_search_provider_integration.py b/tests/test_search_provider_integration.py
index bd28f3f..244aa9f 100644
--- a/tests/test_search_provider_integration.py
+++ b/tests/test_search_provider_integration.py
@@ -64,3 +64,4 @@ def test_provider_composes_with_group(local_store, test_client):
     body = search(test_client, provider_id="bv.provider.oan.dev", group="mh")
 
     assert body["hits"] == []
+
diff --git a/uv.lock b/uv.lock
index 9292f9a..5e9aedd 100644
--- a/uv.lock
+++ b/uv.lock
@@ -740,6 +740,18 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/fd/3c/6a2bf344106328fd04963664a60b9bb6496fc25df8e962fcdc1367285fb9/fsspec-2026.7.0-py3-none-any.whl", hash = "sha256:b57ddbafedfaef7018c1ecab32aa200a9d7ca26b77965f64e48b70061249d279", size = 206583, upload-time = "2026-07-28T16:34:49.538Z" },
 ]
 
+[[package]]
+name = "googleapis-common-protos"
+version = "1.76.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "protobuf" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/62/b6/bc095b23a3e0d978e75c8c18cabe4c11f791837410670e4486f0d539bfba/googleapis_common_protos-1.76.0.tar.gz", hash = "sha256:71db0e44a1d4c2dd024ebad44ca67129cc1e2892e4861df466764749d5ef8642", size = 156250, upload-time = "2026-10-22T23:43:36.263Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/05/6d/78dc67a9e779e594b49bf62604034d3a6aa428a55afc729cafda4b7333e1/googleapis_common_protos-1.76.0-py3-none-any.whl", hash = "sha256:59e13344a9f3651aa201a03e2bee815a3bf43a4baf1f9ff8a9b62a840ce25fac", size = 307707, upload-time = "2026-10-22T23:43:10.445Z" },
+]
+
 [[package]]
 name = "grpcio"
 version = "1.83.0"
@@ -1127,6 +1139,9 @@ dependencies = [
     { name = "langchain-text-splitters" },
     { name = "minio" },
     { name = "openpyxl" },
+    { name = "opentelemetry-exporter-otlp-proto-http" },
+    { name = "opentelemetry-instrumentation-logging" },
+    { name = "opentelemetry-sdk" },
     { name = "pdf-inspector" },
     { name = "pillow" },
     { name = "psycopg2-binary" },
@@ -1165,6 +1180,9 @@ requires-dist = [
     { name = "langchain-text-splitters", specifier = ">=0.0.1" },
     { name = "minio", specifier = ">=7.0.0" },
     { name = "openpyxl", specifier = ">=3.1.0" },
+    { name = "opentelemetry-exporter-otlp-proto-http", specifier = ">=1.44.0" },
+    { name = "opentelemetry-instrumentation-logging", specifier = ">=0.65b0" },
+    { name = "opentelemetry-sdk", specifier = ">=1.44.0" },
     { name = "pdf-inspector", specifier = ">=1.14.0" },
     { name = "pillow", specifier = ">=10.0.0" },
     { name = "psycopg2-binary", specifier = ">=2.9.0" },
@@ -1713,6 +1731,147 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/c0/da/977ded879c29cbd04de313843e76868e6e13408a94ed6b987245dc7c8506/openpyxl-3.1.5-py2.py3-none-any.whl", hash = "sha256:5282c12b107bffeef825f4617dc029afaf41d0ea60823bbb665ef3079dc79de2", size = 250910, upload-time = "2024-06-28T14:03:41.161Z" },
 ]
 
+[[package]]
+name = "opentelemetry-api"
+version = "1.45.2"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "typing-extensions" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/19/14/cf3a6b6448849656e1e5f4e2101f3988dc6789724050096aba1bc5ccf592/opentelemetry_api-1.45.2.tar.gz", hash = "sha256:d7b0008e21e5fd61cbe7a7f34ca3b6905a78ab677d9bc395e299359c75cda6b4", size = 73277, upload-time = "2026-10-29T17:29:37.888Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/a9/2d/610db864d7513158cebb8f36cc88e67e73bd6c7aa1905ef18d89e1281c81/opentelemetry_api-1.45.2-py3-none-any.whl", hash = "sha256:2cbd737377e710ee740d261c94e93fa0f08e6d037c18f17513cc0a2d94479a34", size = 60335, upload-time = "2026-10-29T17:29:06.955Z" },
+]
+
+[[package]]
+name = "opentelemetry-exporter-http-transport"
+version = "0.66b2"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "opentelemetry-api" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/a6/e6/dbeb3e217418071656232a4028c791060d219f218cef1249b664049d12c0/opentelemetry_exporter_http_transport-0.66b2.tar.gz", hash = "sha256:66d6ed810a515432c8891da9a834464332fd70badeca45bdeeae81b4280c903b", size = 11693, upload-time = "2026-10-29T17:29:39.444Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/a1/b2/c622a101f31c0e6670b82f4264c4e77c161bce415bc6fc158fda0b82e218/opentelemetry_exporter_http_transport-0.66b2-py3-none-any.whl", hash = "sha256:fd5f5b490c099acda1cd7b909d5e8981436bf18cc8c6098d302d114c271720e9", size = 12154, upload-time = "2026-10-29T17:29:10.244Z" },
+]
+
+[package.optional-dependencies]
+requests = [
+    { name = "requests" },
+]
+
+[[package]]
+name = "opentelemetry-exporter-otlp-common"
+version = "0.66b2"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "opentelemetry-sdk" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/5a/3c/3f2cc8c42bde2620e65024d71e2eb025f7bd0821bb363b52bcc5a3260dbd/opentelemetry_exporter_otlp_common-0.66b2.tar.gz", hash = "sha256:06f5a9b88638e310efa1f1d206cdf6b75393e112aba4e174a963416192bd1028", size = 14325, upload-time = "2026-10-29T17:29:41.701Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/a8/47/b3049e30b339c82e3c7ec1fc692185a5d5a3ba65a94c570bb39b2c0b8424/opentelemetry_exporter_otlp_common-0.66b2-py3-none-any.whl", hash = "sha256:ace86efb75489f925f1ef809856ae5f0c7db534e4afdc624530222228b1236a5", size = 12385, upload-time = "2026-10-29T17:29:13.667Z" },
+]
+
+[[package]]
+name = "opentelemetry-exporter-otlp-proto-common"
+version = "1.45.2"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "opentelemetry-proto" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/69/58/239c98b36d9ea317098769957043c251703636a26189c884fba46c1dad84/opentelemetry_exporter_otlp_proto_common-1.45.2.tar.gz", hash = "sha256:f1f2806f3f28ce7938033823e50934fcd91595a39de38e69a28a51e19f40fc6d", size = 18873, upload-time = "2026-10-29T17:29:44.496Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/e6/d6/0f30d38120a949aef70528f845bf539f80df62abb4b8e62fdf69c2b6df5a/opentelemetry_exporter_otlp_proto_common-1.45.2-py3-none-any.whl", hash = "sha256:af483b24217028892f017096731b794b196b785025d2e6808874720fd0b05a80", size = 15392, upload-time = "2026-10-29T17:29:17.913Z" },
+]
+
+[[package]]
+name = "opentelemetry-exporter-otlp-proto-http"
+version = "1.45.2"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "googleapis-common-protos" },
+    { name = "opentelemetry-api" },
+    { name = "opentelemetry-exporter-http-transport", extra = ["requests"] },
+    { name = "opentelemetry-exporter-otlp-common" },
+    { name = "opentelemetry-exporter-otlp-proto-common" },
+    { name = "opentelemetry-proto" },
+    { name = "opentelemetry-sdk" },
+    { name = "requests" },
+    { name = "typing-extensions" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/db/07/b7b01ecb69d02f540e679a9e36f7413d081f225bbbed4469d212ada10bb7/opentelemetry_exporter_otlp_proto_http-1.45.2.tar.gz", hash = "sha256:5984a3fb46bc4628e4e2b85d4a6e72637e03962aae6cf372791137768a4ad939", size = 28843, upload-time = "2026-10-29T17:29:46.15Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/32/65/f64d7d2450217e2989714e1dee15e19bf79a77a313387abac0cac77bc10e/opentelemetry_exporter_otlp_proto_http-1.45.2-py3-none-any.whl", hash = "sha256:6e4d025138829d0ee57946930c2ccad6491ad4d92a958e9c7eea69278fe60cfc", size = 22180, upload-time = "2026-10-29T17:29:20.154Z" },
+]
+
+[[package]]
+name = "opentelemetry-instrumentation"
+version = "0.66b2"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "opentelemetry-api" },
+    { name = "opentelemetry-semantic-conventions" },
+    { name = "packaging" },
+    { name = "wrapt" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/76/b9/46a136199fa6928eecefec8418867a460d94f01c608a85d22bd2ed4e8d48/opentelemetry_instrumentation-0.66b2.tar.gz", hash = "sha256:02b094150e53b76f5906d75e628dbb3833bbc6e62fa109b5c7ba0c081865e923", size = 43305, upload-time = "2026-10-29T17:38:11.618Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/3b/41/014d40e2b08a5d9d9058ed4967ad992657c203ce0c292ded97364ee7b82a/opentelemetry_instrumentation-0.66b2-py3-none-any.whl", hash = "sha256:fa9ccfdab147a14af8107d19ef0a7b5015675273c5adced7a1fc38f28b42ec4c", size = 36902, upload-time = "2026-10-29T17:37:06.994Z" },
+]
+
+[[package]]
+name = "opentelemetry-instrumentation-logging"
+version = "0.66b2"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "opentelemetry-api" },
+    { name = "opentelemetry-instrumentation" },
+    { name = "opentelemetry-semantic-conventions" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/23/2f/900c10a10c780f06c0e90e885e5fdcb95f93a91efbd73d3b87d11c6ac72a/opentelemetry_instrumentation_logging-0.66b2.tar.gz", hash = "sha256:a94210291ae10395ad617c2b69bd7c2e10f77818b8beac11368460f47c370eb1", size = 20350, upload-time = "2026-10-29T17:38:34.161Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/6d/52/1f18b14487fa599f96537b2550c7096f824216fedeeb03fdb7f0f67db847/opentelemetry_instrumentation_logging-0.66b2-py3-none-any.whl", hash = "sha256:6141bc09a7868235172748e8a2cd9806549d742b4f5019ce2a571e8c157e5e7d", size = 15926, upload-time = "2026-10-29T17:37:36.938Z" },
+]
+
+[[package]]
+name = "opentelemetry-proto"
+version = "1.45.2"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "protobuf" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/24/9c/1aa4ce60e9bc72af20dc49c92cc0566825092d92b3d6a277439fb5fb7c01/opentelemetry_proto-1.45.2.tar.gz", hash = "sha256:bd8d23cf770605b0af9b7be2071d26a464fea289d35429abfd2a66226114ad64", size = 46486, upload-time = "2026-10-29T17:29:53.007Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/7e/0d/a0b8f2b9efe2ec1abdf639d3696db9b1563ebeb11d91aac08ba7341e9122/opentelemetry_proto-1.45.2-py3-none-any.whl", hash = "sha256:03a7ca1b43f89fd5b965d5975902bb2515c0c86e2428b456836e9a6149c6235a", size = 72484, upload-time = "2026-10-29T17:29:31.403Z" },
+]
+
+[[package]]
+name = "opentelemetry-sdk"
+version = "1.45.2"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "opentelemetry-api" },
+    { name = "opentelemetry-semantic-conventions" },
+    { name = "typing-extensions" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/7b/1a/74a21d4d6969fb8b432ac397e66112a8167a6deb267ea4559dbaf2fe91ef/opentelemetry_sdk-1.45.2.tar.gz", hash = "sha256:7eec4de13dd21ff3a1670dcb0e02079e692736f94d733dfdce5af461d884248d", size = 221705, upload-time = "2026-10-29T17:29:54.685Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/9f/a5/9a3b793ef48245f2336a16beb2324c2397dbb3f987cd68942f77cca06156/opentelemetry_sdk-1.45.2-py3-none-any.whl", hash = "sha256:ff974ea70ee4f1d51e2ec152db37c2b626c7bf82bb3f7d86114fd8704e7199c6", size = 141756, upload-time = "2026-10-29T17:29:33.722Z" },
+]
+
+[[package]]
+name = "opentelemetry-semantic-conventions"
+version = "0.66b2"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "opentelemetry-api" },
+    { name = "typing-extensions" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/37/3f/50f95972b50e8e353f5b9342034c9cd17730f1956f009ce6a3fbbd89f9ce/opentelemetry_semantic_conventions-0.66b2.tar.gz", hash = "sha256:7da5f1a5dc4383eb66711d4d3d3e71824d9403ac1559fea021cc103c1a4c4976", size = 150246, upload-time = "2026-10-29T17:29:55.682Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/67/bd/6bf33ed31521c565931c037e9e81bac47c1acb000e850c6e24220e2a0c00/opentelemetry_semantic_conventions-0.66b2-py3-none-any.whl", hash = "sha256:4cbe61e5f9275d57ab31708446e7938ef4291ebbe6a7fca6f7a629c5acae87d8", size = 206280, upload-time = "2026-10-29T17:29:35.062Z" },
+]
+
 [[package]]
 name = "orjson"
 version = "3.12.0"