chore(observability): align AI service log format with web/worker logs (#51)

CHANGELOG.md

@@ -70,6 +70,9 @@ This file records what changes **in the product** – process and session state
 - Verify commands `pnpm verify:changed`, `pnpm verify`, `pnpm verify:full`; CI runs integration
   tests against real PostgreSQL + SeaweedFS.
 
+### Changed
+- AI service logs use the web/worker format: `time` (ISO 8601, `Z`) instead of `ts`, lower-case pino level labels (`warn`, not `WARNING`), `logger` only on library records.
+
 ### Fixed
 - AI verifier: a unit quoted together with the neighbouring table cell (e.g. `60    | Stk.`) is now
   confirmed as `found` when one cell of the quote is exactly the unit; quotes that differ from the

docs/technical/operations.md

@@ -79,13 +79,22 @@
 
 ## Logs and correlation
 
-- Web and worker write one JSON line per event (pino, stdout): `level`, `time`, `event` plus IDs and codes
-  only (`requestId`, `jobId`, `companyId`, `documentId`, `attempt`, `code`, `status`, `durationMs`,
-  `count`). The key set is fixed in code; a test runs a full synthetic request (upload → export) and
-  fails if a log line contains document content or personal data.
-- Correlation: the request id. The worker sends it to the AI service as `X-Request-Id`; the AI service
-  logs it as `requestId` (JSON logging, `services/ai/src/requestflow_ai/jsonlog.py`). Follow one request with
-  `docker compose logs web worker ai | grep <requestId>`.
+### Shared log format (web, worker, AI service)
+
+All three write one JSON line per event to stdout with the same keys: `level` (lower-case pino label:
+`debug`, `info`, `warn`, `error`, `fatal`), `time` (UTC ISO 8601 with milliseconds, `Z`), `event` (a
+constant name, never free text) plus allow-listed IDs and codes only. Correlation keys: `requestId`
+everywhere, `documentId`; `jobId`, `companyId` and `attempt` only in web/worker lines (the AI service
+receives just `X-Request-Id`). Web/worker (pino, `src/features/observability/log.ts`) add `code`,
+`status`, `durationMs`, `count`; the AI service (`services/ai/src/requestflow_ai/jsonlog.py`) keeps its
+own whitelist (`status`, `latencyMs`, `errorCode`, token counts, …), plus `excType` (exception class only,
+never its message) and `logger` for records of a library (docling, httpx, uvicorn). Compact JSON like pino. Example:
+`{"level":"info","time":"2026-09-24T10:00:00.123Z","event":"request_completed","requestId":"req-0001","status":200}`
+
+- Both key sets are fixed in code; a test runs a full synthetic request (upload → export) and fails if a
+  log line contains document content or personal data.
+- Correlation: the request id. The worker sends it to the AI service as `X-Request-Id`. Follow one request
+  with `docker compose logs web worker ai | grep <requestId>`.
 - `/api/health` also reports `dependencies.aiService` (reachable or not) and `backlog` (waiting jobs per
   queue). Both are informational and never turn the status into 503 – the AI service is an optional
   compose profile.

services/ai/src/requestflow_ai/jsonlog.py

@@ -3,6 +3,9 @@
 Each record carries a constant event name (the log message template, never formatted with its
 arguments), the request and document IDs from context variables, and only allow-listed extras.
 Exceptions are reduced to their type name: tracebacks and messages can contain document data.
+The line shape matches the web/worker pino logs (`level`, `time`, `event`, `requestId`, ...), see
+docs/technical/operations.md "Logs and correlation". The service never learns `jobId` or `companyId`
+(it only gets `X-Request-Id`), so those keys appear in web/worker lines only.
 """
 
 from __future__ import annotations
@@ -52,12 +55,25 @@
 )
 
 
+# Same labels as pino in the web/worker logs (src/features/observability/log.ts).
+_LEVEL_LABELS = {
+    "DEBUG": "debug",
+    "INFO": "info",
+    "WARNING": "warn",
+    "ERROR": "error",
+    "CRITICAL": "fatal",
+}
+
+
 class JsonFormatter(logging.Formatter):
+    """One line per record with the web/worker key set: `level`, `time`, `event` plus IDs."""
+
     def format(self, record: logging.LogRecord) -> str:
+        created = datetime.fromtimestamp(record.created, UTC)
         payload: dict[str, Any] = {
-            "ts": datetime.fromtimestamp(record.created, UTC).isoformat(timespec="milliseconds"),
-            "level": record.levelname,
-            "logger": record.name,
+            "level": _LEVEL_LABELS.get(record.levelname, record.levelname.lower()),
+            # pino's isoTime (`Date.toISOString()`): UTC, milliseconds, `Z` suffix.
+            "time": created.isoformat(timespec="milliseconds").replace("+00:00", "Z"),
             "event": record.msg if isinstance(record.msg, str) else type(record.msg).__name__,
         }
         request_id = request_id_var.get()
@@ -69,9 +85,14 @@ def format(self, record: logging.LogRecord) -> str:
         for key in _ALLOWED_EXTRAS:
             if key in record.__dict__:
                 payload[key] = record.__dict__[key]
+        # A library's own record (docling, httpx, uvicorn) names its source; our records do not
+        # need it, like the pino lines (#51 review).
+        if not record.name.startswith("requestflow_ai"):
+            payload["logger"] = record.name
         if record.exc_info and record.exc_info[0] is not None:
             payload["excType"] = record.exc_info[0].__name__
-        return json.dumps(payload, ensure_ascii=False, default=str)
+        # Compact separators like pino, so a line is byte-for-byte the same shape as web/worker.
+        return json.dumps(payload, ensure_ascii=False, default=str, separators=(",", ":"))
 
 
 def configure_logging(level: str = "INFO") -> None:

services/ai/tests/test_jsonlog.py

@@ -0,0 +1,96 @@
+"""The AI service log line has the same shape as the web/worker pino lines (#51)."""
+
+from __future__ import annotations
+
+import io
+import json
+import logging
+import re
+
+from requestflow_ai.jsonlog import JsonFormatter, document_id_var, request_id_var
+
+ISO_UTC_MILLIS = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z$")
+
+
+def capture_line(
+    level: int,
+    event: str,
+    extra: dict[str, object],
+    name: str = "requestflow_ai.test_jsonlog",
+    exc_info: bool = False,
+) -> str:
+    stream = io.StringIO()
+    handler = logging.StreamHandler(stream)
+    handler.setFormatter(JsonFormatter())
+    logger = logging.getLogger(name)
+    logger.addHandler(handler)
+    propagate = logger.propagate
+    logger.propagate = False
+    request_token = request_id_var.set("req-synthetic-51")
+    document_token = document_id_var.set("doc-synthetic-51")
+    try:
+        logger.log(level, event, extra=extra, exc_info=exc_info)
+    finally:
+        request_id_var.reset(request_token)
+        document_id_var.reset(document_token)
+        logger.removeHandler(handler)
+        logger.propagate = propagate
+    lines = stream.getvalue().splitlines()
+    assert len(lines) == 1
+    return lines[0]
+
+
+def capture(level: int, event: str, extra: dict[str, object]) -> dict[str, object]:
+    return json.loads(capture_line(level, event, extra))
+
+
+def test_log_line_uses_the_shared_key_set_of_the_ts_logs() -> None:
+    record = capture(
+        logging.WARNING,
+        "extraction_failed",
+        {"errorCode": "parse_failed", "quote": "Musterbau GmbH, max@example.com"},
+    )
+
+    assert set(record) == {"time", "level", "event", "requestId", "documentId", "errorCode"}
+    assert record["level"] == "warn"
+    assert isinstance(record["time"], str)
+    assert ISO_UTC_MILLIS.match(record["time"])
+    assert record["event"] == "extraction_failed"
+    assert record["requestId"] == "req-synthetic-51"
+    assert record["documentId"] == "doc-synthetic-51"
+
+
+def test_level_labels_match_pino() -> None:
+    expected = {
+        logging.DEBUG: "debug",
+        logging.INFO: "info",
+        logging.WARNING: "warn",
+        logging.ERROR: "error",
+        logging.CRITICAL: "fatal",
+    }
+    logging.getLogger("requestflow_ai.test_jsonlog").setLevel(logging.DEBUG)
+    for level, label in expected.items():
+        assert capture(level, "probe", {})["level"] == label
+
+
+def test_line_is_compact_like_pino() -> None:
+    line = capture_line(logging.INFO, "request_completed", {"status": 200})
+    assert ", " not in line
+    assert '":' in line and '": ' not in line
+
+
+def test_exception_names_only_its_type_never_its_message() -> None:
+    try:
+        raise ValueError("Musterbau GmbH, max@example.com")
+    except ValueError:
+        line = capture_line(logging.ERROR, "extraction_failed", {}, exc_info=True)
+    record = json.loads(line)
+    assert record["excType"] == "ValueError"
+    assert "Musterbau" not in line
+    assert "example.com" not in line
+
+
+def test_library_records_name_their_logger_ours_do_not() -> None:
+    assert "logger" not in json.loads(capture_line(logging.WARNING, "x", {}))
+    library = json.loads(capture_line(logging.WARNING, "x", {}, name="docling.synthetic"))
+    assert library["logger"] == "docling.synthetic"