patch: Use Loguru's built-in JSON logging (#1941)

## Why

FCC's custom JSON formatter dropped exception details and bound fields
outside its fixed allowlist. An intercepted exception could be written
as only "request failed", losing the information needed to diagnose it.

## How

Use Loguru's built-in JSON serialization (`serialize=True`) and delete
the custom formatter and its redaction functions. New JSON lines contain
Loguru's `text` and `record` fields, including rendered tracebacks,
exception metadata and bound/contextual fields. The sink preserves
supplied message text without an additional redaction hook, as agreed
for these local diagnostic logs.

Update trace readers and smoke assertions to read
`record.extra.trace_payload` and log readers to handle Loguru's UTF-8
output.




<!-- greptile_comment -->

<!-- greptile_summary -->

<h2><a
href="https://app.greptile.com/api/retrigger?id=71172606"><picture><source
media="(prefers-color-scheme: dark)"
srcset="https://greptile-static-assets.s3.amazonaws.com/badges/RetriggerDark.svg?v=2"><source
media="(prefers-color-scheme: light)"
srcset="https://greptile-static-assets.s3.amazonaws.com/badges/Retrigger.svg?v=2"><img
alt="Retrigger"
src="https://greptile-static-assets.s3.amazonaws.com/badges/Retrigger.svg?v=2"
align="right"></picture></a>Confidence Score: 5/5</h2>

<!-- greptile-risk -->

No outstanding findings block merging.

<details open><summary>Summary</summary>

The PR updates serialized trace logging and its readers. The smoke trace
parser test now uses fixed output instead of reconfiguring process-wide
logging.
</details>

<!-- greptile_confidence_score:5 -->

<sub>Reviews (2) · Last reviewed commit: ["Isolate smoke trace parser
test from
glo..."](3fb4fd6484)</sub>

<!-- /greptile_comment -->

smoke/product/test_client_product_live.py

@@ -47,6 +47,21 @@ def _json_object_lines(text: str) -> list[JsonObject]:
     return objects
 
 
+def _trace_log_events(text: str) -> list[JsonObject]:
+    events: list[JsonObject] = []
+    for row in _json_object_lines(text):
+        record = row.get("record")
+        if not isinstance(record, dict):
+            continue
+        extra = record.get("extra")
+        if not isinstance(extra, dict):
+            continue
+        payload = extra.get("trace_payload")
+        if isinstance(payload, dict):
+            events.append(payload)
+    return events
+
+
 @pytest.mark.smoke_target("clients")
 def test_vscode_protocol_e2e(smoke_config: SmokeConfig) -> None:
     provider_model = ProviderMatrixDriver(smoke_config).first_model()
@@ -985,7 +1000,7 @@ def test_claude_cli_web_search_e2e(smoke_config: SmokeConfig, tmp_path: Path) ->
     assert '"type": "server_tool_use"' in automatic_payload
     assert '"type": "web_search_tool_result"' in automatic_payload
     assert "github.com" in automatic_payload
-    log_rows = _json_object_lines(server_log)
+    log_rows = _trace_log_events(server_log)
     assert (
         sum(
             row.get("event") == "free_claude_code.api.web_search.automatic_recognized"
@@ -1087,7 +1102,7 @@ def test_claude_auto_mode_openai_connected_e2e(
     ):
         assert unexpected not in combined_lower
 
-    log_rows = _json_object_lines(server_log)
+    log_rows = _trace_log_events(server_log)
     policy_rows = [
         row
         for row in log_rows

src/free_claude_code/config/logging_config.py

@@ -2,13 +2,11 @@
 
 Structured logs are written as JSON lines to a configurable path (default
 ``~/.fcc/logs/server.log``). Stdlib logging is intercepted and funneled to loguru.
-Context vars (request_id, node_id, chat_id) from contextualize() are
-included at top level for easy grep/filter.
+Loguru's JSON stores exception metadata in ``record.exception`` and context in
+``record.extra``. Structured traces live in ``record.extra.trace_payload``.
 """
 
-import json
 import logging
-import re
 import threading
 from pathlib import Path
 
@@ -30,63 +28,6 @@ _THIRD_PARTY_LOGGERS = (
     "telegram.ext",
 )
 
-# Loguru ``logger.bind()`` key used by structured TRACE payloads; ``core/trace.py``
-# uses the identical string constant ``TRACE_PAYLOAD_BINDING``.
-_TRACE_PAYLOAD_BINDING = "trace_payload"
-
-# Context keys we promote to top-level JSON for traceability / grep
-_CONTEXT_KEYS = (
-    "request_id",
-    "node_id",
-    "chat_id",
-    "claude_session_id",
-    "http_method",
-    "http_path",
-)
-
-_TELEGRAM_BOT_RE = re.compile(
-    r"(https?://api\.telegram\.org/)bot([0-9]+:[A-Za-z0-9_-]+)(/?)",
-    re.IGNORECASE,
-)
-# Authorization: Bearer <token> (HTTP client / proxy debug lines)
-_AUTH_BEARER_RE = re.compile(
-    r"(\bAuthorization\s*:\s*Bearer\s+)([^\s'\"]+)",
-    re.IGNORECASE,
-)
-
-
-def _redact_sensitive_substrings(message: str) -> str:
-    """Remove obvious API tokens and secrets before JSON log line emission."""
-    text = _TELEGRAM_BOT_RE.sub(r"\1bot<redacted>\3", message)
-    return _AUTH_BEARER_RE.sub(r"\1<redacted>", text)
-
-
-def _serialize_with_context(record) -> str:
-    """Format record as JSON with context vars at top level.
-    Returns a format template; we inject _json into record for output.
-    """
-    extra = record.get("extra", {})
-    out = {
-        "time": str(record["time"]),
-        "level": record["level"].name,
-        "message": _redact_sensitive_substrings(str(record["message"])),
-        "module": record["name"],
-        "function": record["function"],
-        "line": record["line"],
-    }
-    trace_payload = extra.get(_TRACE_PAYLOAD_BINDING)
-    for key in _CONTEXT_KEYS:
-        if key in extra and extra[key] is not None:
-            out[key] = extra[key]
-    if isinstance(trace_payload, dict):
-        for tk, tv in trace_payload.items():
-            if tk in out:
-                continue
-            out[tk] = tv
-        out["trace"] = True
-    record["_json"] = json.dumps(out, default=str)
-    return "{_json}\n"
-
 
 class InterceptHandler(logging.Handler):
     """Redirect stdlib logging to loguru."""
@@ -129,7 +70,7 @@ def _add_file_sink(log_file: str | Path, level: str) -> int:
     return logger.add(
         log_path,
         level=level,
-        format=_serialize_with_context,
+        serialize=True,
         encoding="utf-8",
         mode="a",
         rotation="50 MB",

src/free_claude_code/core/trace.py

@@ -1,6 +1,6 @@
 """Structured DEBUG traces for end-to-end request / CLI / provider logging.
 
-Emitted lines are merged into JSON log rows by ``config.logging_config``.
+Trace payloads are preserved in ``record.extra.trace_payload`` by Loguru.
 Conversation and Claude Code prompts are logged verbatim unless values live under
 sanitized credential keys (e.g. ``api_key``, ``authorization``). The default
 INFO log level excludes these detailed request traces.
@@ -50,7 +50,7 @@ def sanitize_trace_value(obj: Any) -> Any:
 
 
 def trace_event(*, stage: str, event: str, source: str, **fields: Any) -> None:
-    """Emit one structured DEBUG trace row merged into JSON by the log sink."""
+    """Emit a DEBUG record with a structured payload bound to its extra fields."""
     payload = sanitize_trace_value(
         {
             "stage": stage,

tests/config/test_logging_config.py

@@ -7,6 +7,7 @@ import sys
 from pathlib import Path
 from unittest.mock import patch
 
+import pytest
 from loguru import logger
 
 from free_claude_code.config import logging_config
@@ -30,6 +31,51 @@ def test_configure_logging_creates_parent_directories(tmp_path) -> None:
     assert log_file.is_file()
 
 
+def test_native_json_preserves_bound_and_request_context(tmp_path):
+    log_file = tmp_path / "context.log"
+    configure_logging(log_file, force=True)
+    with logger.contextualize(request_id="req-test", session_id="session-test"):
+        logger.bind(provider_id="provider-test", run_id="run-test").info("completed")
+    logger.complete()
+    row = json.loads(log_file.read_text(encoding="utf-8"))
+    assert row["record"]["extra"] == {
+        "request_id": "req-test",
+        "session_id": "session-test",
+        "provider_id": "provider-test",
+        "run_id": "run-test",
+    }
+    assert row["record"]["level"]["name"] == "INFO"
+    assert row["record"]["message"] == "completed"
+
+
+@pytest.mark.parametrize("stdlib", [False, True])
+def test_native_json_preserves_exception_chain_and_stack(tmp_path, stdlib):
+    log_file = tmp_path / "exception.log"
+    configure_logging(log_file, force=True)
+    try:
+        try:
+            raise ValueError("upstream HTTP 429: rate limited")
+        except ValueError as exc:
+            raise RuntimeError("provider request failed") from exc
+    except RuntimeError:
+        if stdlib:
+            logging.getLogger("test.exception").exception("request failed")
+        else:
+            logger.exception("request failed")
+    logger.complete()
+    rows = log_file.read_text(encoding="utf-8").splitlines()
+    assert len(rows) == 1
+    row = json.loads(rows[0])
+    assert row["record"]["exception"] == {
+        "type": "RuntimeError",
+        "value": "provider request failed",
+        "traceback": True,
+    }
+    assert "ValueError: upstream HTTP 429: rate limited" in row["text"]
+    assert "RuntimeError: provider request failed" in row["text"]
+    assert "test_native_json_preserves_exception_chain_and_stack" in row["text"]
+
+
 def test_logging_preserves_records_across_process_restarts(tmp_path):
     log_file = tmp_path / "server.log"
     script = """
@@ -49,8 +95,13 @@ logger.remove()
             text=True,
             timeout=10,
         )
-    records = [json.loads(line) for line in log_file.read_text().splitlines()]
-    assert [record["message"] for record in records] == ["first run", "second run"]
+    records = [
+        json.loads(line) for line in log_file.read_text(encoding="utf-8").splitlines()
+    ]
+    assert [record["record"]["message"] for record in records] == [
+        "first run",
+        "second run",
+    ]
 
 
 def test_forced_logging_reconfiguration_preserves_records(tmp_path):
@@ -60,8 +111,10 @@ def test_forced_logging_reconfiguration_preserves_records(tmp_path):
     configure_logging(log_file, force=True)
     logger.info("after reconfiguration")
     logger.complete()
-    records = [json.loads(line) for line in log_file.read_text().splitlines()]
-    assert [record["message"] for record in records] == [
+    records = [
+        json.loads(line) for line in log_file.read_text(encoding="utf-8").splitlines()
+    ]
+    assert [record["record"]["message"] for record in records] == [
         "before reconfiguration",
         "after reconfiguration",
     ]
@@ -122,26 +175,24 @@ def test_configure_logging_replaces_sink_when_path_changes(tmp_path):
     assert "second destination" in second_text
 
 
-def test_telegram_bot_token_redacted_in_message_field(tmp_path) -> None:
-    log_file = str(tmp_path / "redact.log")
-    configure_logging(log_file, force=True, verbose_third_party=False)
-    token = "123456:ABCDEF-ghij-klm"
-    logger.info("Calling {}", f"https://api.telegram.org/bot{token}/getMe")
+@pytest.mark.parametrize(
+    "message",
+    [
+        "Calling https://api.telegram.org/bot123456:synthetic-token/getMe",
+        "Request headers: Authorization: Bearer synthetic-token",
+        "first line\nsecond line \u2603",
+    ],
+)
+def test_native_json_preserves_message_text(tmp_path, message):
+    log_file = tmp_path / "message.log"
+    configure_logging(log_file, force=True)
+    logger.info(message)
     logger.complete()
-    text = Path(log_file).read_text(encoding="utf-8")
-    assert token not in text
-    assert "bot<redacted>/" in text or "redacted" in text
-
-
-def test_bearer_substring_redacted_in_log_file(tmp_path) -> None:
-    log_file = str(tmp_path / "bearer.log")
-    configure_logging(log_file, force=True, verbose_third_party=False)
-    secret = "ya29.secret-token-abc"
-    logger.info("Request headers: Authorization: Bearer {}", secret)
-    logger.complete()
-    text = Path(log_file).read_text(encoding="utf-8")
-    assert secret not in text
-    assert "Bearer" in text
+    lines = log_file.read_text(encoding="utf-8").splitlines()
+    assert len(lines) == 1
+    row = json.loads(lines[0])
+    assert row["record"]["message"] == message
+    assert message in row["text"]
 
 
 def test_noisy_third_party_loggers_quieted_when_not_verbose(tmp_path) -> None:

tests/contracts/test_smoke_child_process.py

@@ -20,9 +20,28 @@ from smoke.lib.config import ProviderModel, SmokeConfig
 from smoke.lib.e2e import ConversationDriver
 from smoke.lib.http import collect_message_stream
 from smoke.lib.server import RunningServer
+from smoke.product import test_client_product_live as client_smoke
 from smoke.product import test_provider_product_live as provider_smoke
 
 
+def test_smoke_reads_native_trace_events_from_mixed_subprocess_output():
+    output = (
+        'INFO: server started\n{"type":"result"}\n'
+        '{"text":"ordinary log record\\n","record":{"extra":{}}}\n'
+        '{"text":"trace event\\n","record":{"extra":{"trace_payload":'
+        '{"stage":"execution","event":"test.completed","source":"test","status":200}}}}\n'
+    )
+    assert client_smoke._trace_log_events(output) == [
+        {
+            "stage": "execution",
+            "event": "test.completed",
+            "source": "test",
+            "status": 200,
+        }
+    ]
+    assert {"type": "result"} in client_smoke._json_object_lines(output)
+
+
 def test_fcc_server_command_uses_cli_entrypoint() -> None:
     assert cmd_fcc_server() == [
         child_process.python_exe(),

tests/core/test_trace.py

@@ -54,13 +54,21 @@ def _json_log_rows(log_file: str) -> list[dict]:
     return [json.loads(line) for line in text.split("\n")]
 
 
-def test_trace_payload_merged_into_json_line(tmp_path) -> None:
+def _trace_payloads(log_file: str) -> list[dict]:
+    return [
+        row["record"]["extra"][TRACE_PAYLOAD_BINDING]
+        for row in _json_log_rows(log_file)
+        if TRACE_PAYLOAD_BINDING in row["record"]["extra"]
+    ]
+
+
+def test_trace_payload_preserved_in_native_json(tmp_path) -> None:
     log_file = str(tmp_path / "t.log")
     configure_logging(log_file, force=True, level="DEBUG")
     trace_event(stage="s", event="e.v1", source="unit", hello="world", n=42)
     row = _json_log_rows(log_file)[-1]
-    assert row["level"] == "DEBUG"
-    assert row["trace"] is True
+    assert row["record"]["level"]["name"] == "DEBUG"
+    row = row["record"]["extra"][TRACE_PAYLOAD_BINDING]
     assert row["stage"] == "s"
     assert row["event"] == "e.v1"
     assert row["source"] == "unit"
@@ -77,7 +85,7 @@ def test_trace_payload_excluded_from_default_info_logs(tmp_path) -> None:
     logger.info("visible lifecycle event")
 
     rows = _json_log_rows(log_file)
-    assert [row["message"] for row in rows] == ["visible lifecycle event"]
+    assert [row["record"]["message"] for row in rows] == ["visible lifecycle event"]
 
 
 def test_sanitize_masks_nested_api_key_strings() -> None:
@@ -112,7 +120,7 @@ async def test_traced_async_stream_logs_completion(tmp_path) -> None:
 
     assert chunks == ["hello", " world"]
     assert source.close_calls == 1
-    rows = _json_log_rows(log_file)
+    rows = _trace_payloads(log_file)
     completed = [row for row in rows if row.get("event") == "stream.completed"]
     assert len(completed) == 1
     assert completed[0]["request_id"] == "req_complete"
@@ -144,7 +152,7 @@ async def test_traced_async_stream_logs_real_exception(tmp_path) -> None:
 
     assert source.close_calls == 1
 
-    rows = _json_log_rows(log_file)
+    rows = _trace_payloads(log_file)
     interrupted = [row for row in rows if row.get("event") == "stream.interrupted"]
     assert len(interrupted) == 1
     assert interrupted[0]["request_id"] == "req_error"
@@ -180,7 +188,7 @@ async def test_traced_async_stream_closes_quietly_on_generator_exit(tmp_path) ->
     await stream.aclose()
 
     assert source.close_calls == 1
-    rows = _json_log_rows(log_file)
+    rows = _trace_payloads(log_file)
     events = {row.get("event") for row in rows}
     assert "stream.completed" not in events
     assert "stream.interrupted" not in events