From 3fa8b7bbcef6a8e8ddc1476142f7e53f00407c6e Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 10:59:58 +0000 Subject: [PATCH 1/2] chore(observability): align AI service log format with web/worker logs (#51) The AI service JSON formatter now emits the pino key set used by web and worker: `level` (lower-case pino label, WARNING -> warn, CRITICAL -> fatal), `time` (UTC ISO 8601 with milliseconds and `Z`, like Date.toISOString) and `event`, plus the allow-listed IDs/extras as before. `ts` and `logger` are gone. jobId/companyId are not known to the AI service (it only receives X-Request-Id), so they stay web/worker-only. Documented once in operations.md; test pins the exact key set and level labels. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01DJ5vaKvTYiMvdngT4d3xo1 --- CHANGELOG.md | 3 ++ docs/technical/operations.md | 22 +++++--- services/ai/src/requestflow_ai/jsonlog.py | 22 ++++++-- services/ai/tests/test_jsonlog.py | 61 +++++++++++++++++++++++ 4 files changed, 98 insertions(+), 10 deletions(-) create mode 100644 services/ai/tests/test_jsonlog.py diff --git a/CHANGELOG.md b/CHANGELOG.md index d461102..4d6c381 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -59,3 +59,6 @@ This file records what changes **in the product** – process and session state - Database roles `app_owner` (migrations) and `app_rw` (runtime, no RLS bypass); schema `app`. - 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`), no `logger` key. diff --git a/docs/technical/operations.md b/docs/technical/operations.md index 89a7267..15b14f2 100644 --- a/docs/technical/operations.md +++ b/docs/technical/operations.md @@ -79,13 +79,21 @@ ## 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 `. +### 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, …). 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 `. - `/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. diff --git a/services/ai/src/requestflow_ai/jsonlog.py b/services/ai/src/requestflow_ai/jsonlog.py index 49575e8..f0352f4 100644 --- a/services/ai/src/requestflow_ai/jsonlog.py +++ b/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() diff --git a/services/ai/tests/test_jsonlog.py b/services/ai/tests/test_jsonlog.py new file mode 100644 index 0000000..a08b1a0 --- /dev/null +++ b/services/ai/tests/test_jsonlog.py @@ -0,0 +1,61 @@ +"""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(level: int, event: str, extra: dict[str, object]) -> dict[str, object]: + stream = io.StringIO() + handler = logging.StreamHandler(stream) + handler.setFormatter(JsonFormatter()) + logger = logging.getLogger("requestflow_ai.test_jsonlog") + logger.addHandler(handler) + 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) + finally: + request_id_var.reset(request_token) + document_id_var.reset(document_token) + logger.removeHandler(handler) + lines = stream.getvalue().splitlines() + assert len(lines) == 1 + return json.loads(lines[0]) + + +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 From 0f223c4788816da8ca2db296f0b87251f60a7711 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 11:16:23 +0000 Subject: [PATCH 2/2] chore(observability): compact log lines, library source and exception tests (#51 review) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01DJ5vaKvTYiMvdngT4d3xo1 --- CHANGELOG.md | 2 +- docs/technical/operations.md | 3 +- services/ai/src/requestflow_ai/jsonlog.py | 7 +++- services/ai/tests/test_jsonlog.py | 43 ++++++++++++++++++++--- 4 files changed, 48 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4d6c381..80ebc3a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -61,4 +61,4 @@ This file records what changes **in the product** – process and session state 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`), no `logger` key. +- 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. diff --git a/docs/technical/operations.md b/docs/technical/operations.md index 15b14f2..3e51dee 100644 --- a/docs/technical/operations.md +++ b/docs/technical/operations.md @@ -87,7 +87,8 @@ constant name, never free text) plus allow-listed IDs and codes only. Correlatio 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, …). Example: +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 diff --git a/services/ai/src/requestflow_ai/jsonlog.py b/services/ai/src/requestflow_ai/jsonlog.py index f0352f4..87d675e 100644 --- a/services/ai/src/requestflow_ai/jsonlog.py +++ b/services/ai/src/requestflow_ai/jsonlog.py @@ -85,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: diff --git a/services/ai/tests/test_jsonlog.py b/services/ai/tests/test_jsonlog.py index a08b1a0..c4b5004 100644 --- a/services/ai/tests/test_jsonlog.py +++ b/services/ai/tests/test_jsonlog.py @@ -12,24 +12,36 @@ ISO_UTC_MILLIS = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z$") -def capture(level: int, event: str, extra: dict[str, object]) -> dict[str, object]: +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("requestflow_ai.test_jsonlog") + 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) + 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 json.loads(lines[0]) + 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: @@ -59,3 +71,26 @@ def test_level_labels_match_pino() -> None: 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"