feat(sp18): structured JSON logging + PII scrubber

Cyclone previously emitted stdlib default-formatted log lines that
operators couldn't parse with anything beyond grep. SP18 replaces
that with:

- JsonFormatter: newline-delimited JSON, ISO-8601 ms timestamps,
  structured "extra" dict, exception tracebacks serialized as a
  single string.
- CycloneDevFormatter: tabular format for tail -f in dev.
- PiiScrubber: logging.Filter that redacts NPIs, SSNs, DOBs,
  patient names — both inline in the message ("npi 1881068062")
  and via PHI-keyed extras ({"dob": "1980-04-12"}).
- setup_logging(): idempotent entry point used by the API lifespan
  and CLI main; respects CYCLONE_LOG_LEVEL/FILE/JSON/NO_PII_SCRUB.
- CLI --log-format=json|dev + --log-file=… flags.
- Migrate 9 highest-value log sites in scheduler/backup_scheduler/
  backup_service to use extra={...} (input_filename, claims, parser,
  backup_id, db_fingerprint, etc.).

34 new tests (test_logging_formatter + test_logging_scrubber +
test_logging_setup). All 867 backend tests pass.

backend/src/cyclone/api.py

@@ -99,8 +99,19 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
     from cyclone import backup_service as backup_svc_mod
     from cyclone import backup_scheduler as backup_sched_mod
     from cyclone import secrets as secrets_mod
+    from cyclone.logging_config import setup_logging
     import os as _os
 
+    # SP18: structured JSON logging. Idempotent — safe even if the
+    # CLI pre-configured the root logger.
+    setup_logging(
+        level=_os.environ.get("CYCLONE_LOG_LEVEL", "INFO"),
+        log_file=_os.environ.get("CYCLONE_LOG_FILE") or None,
+        json_format=_os.environ.get("CYCLONE_LOG_JSON", "true").lower() not in (
+            "false", "0", "no",
+        ),
+    )
+
     db.init_db()
     app.state.event_bus = EventBus()
     # SP9: load payer config and seed singleton + 3 providers + CO_TXIX.

backend/src/cyclone/backup_scheduler.py

@@ -139,8 +139,11 @@ class BackupScheduler:
         self._stop_event.clear()
         self._task = asyncio.create_task(self._run(), name="backup-scheduler")
         log.info(
-            "BackupScheduler started: every %.1fh, dir=%s",
-            self._interval_hours, self._service.backup_dir,
+            "BackupScheduler started",
+            extra={
+                "interval_hours": self._interval_hours,
+                "backup_dir": str(self._service.backup_dir),
+            },
         )
 
     async def stop(self) -> None:
@@ -209,7 +212,7 @@ class BackupScheduler:
                 # tick() catches its own exceptions and returns them
                 # in the result. This is the safety net for
                 # programmer errors in the loop body.
-                log.exception("BackupScheduler tick raised: %s", exc)
+                log.exception("BackupScheduler tick raised", extra={"error": str(exc)})
             try:
                 await asyncio.wait_for(
                     self._stop_event.wait(),

backend/src/cyclone/backup_service.py

@@ -284,8 +284,13 @@ class BackupService:
 
             record = self._row_to_record(row)
             log.info(
-                "Backup created: id=%d filename=%s size=%d db_fp=%s",
-                record.id, record.filename, record.size_bytes, record.db_fingerprint,
+                "Backup created",
+                extra={
+                    "backup_id": record.id,
+                    "backup_filename": record.filename,
+                    "size_bytes": record.size_bytes,
+                    "db_fingerprint": record.db_fingerprint,
+                },
             )
             return CreateResult(backup=record, sidecar=sidecar)
 
@@ -435,8 +440,12 @@ class BackupService:
             self._pending_restores[token] = (record.id, expires_at)
 
         log.info(
-            "Restore initiated: backup_id=%d token=%s... expires=%s",
-            record.id, token[:8], expires_at.isoformat(),
+            "Restore initiated",
+            extra={
+                "backup_id": record.id,
+                "token_prefix": token[:8],
+                "expires_at": expires_at.isoformat(),
+            },
         )
         return RestoreInitiateResult(
             backup_id=record.id,
@@ -566,7 +575,13 @@ class BackupService:
                 s.add(row)
             s.commit()
 
-        log.info("Pruned %d backup files older than %s", len(deleted), cutoff.isoformat())
+        log.info(
+            "Pruned old backups",
+            extra={
+                "deleted_count": len(deleted),
+                "cutoff": cutoff.isoformat(),
+            },
+        )
         return deleted
 
     def status(self) -> dict:
@@ -677,6 +692,7 @@ class BackupService:
             "Backup using fallback key derived from SQLCipher DB key "
             "(no separate backup passphrase set); set one via "
             "`cyclone backup init-passphrase` for stronger isolation",
+            extra={"key_source": "sqlcipher_fallback"},
         )
         self._used_fallback = True
         self._key = backup_mod.derive_key(db_key, backup_mod.FALLBACK_SALT)

backend/src/cyclone/cli.py

@@ -3,11 +3,13 @@
 from __future__ import annotations
 
 import logging
+import os
 import sys
 from pathlib import Path
 
 import click
 
+from cyclone.logging_config import setup_logging
 from cyclone.parsers.exceptions import CycloneParseError, CycloneValidationError
 from cyclone.parsers.payer import PayerConfig, PayerConfig835
 from cyclone.parsers.parse_837 import parse as parse_837_text
@@ -41,8 +43,35 @@ def _payer_835(name: str) -> PayerConfig835:
 
 
 @click.group()
-def main() -> None:
+@click.option(
+    "--log-format",
+    default=None,
+    type=click.Choice(["json", "dev"]),
+    help="Log format (default: json; honors CYCLONE_LOG_JSON).",
+)
+@click.option(
+    "--log-file",
+    default=None,
+    type=click.Path(dir_okay=False, path_type=Path),
+    help="Optional rotating log file (honors CYCLONE_LOG_FILE).",
+)
+@click.pass_context
+def main(ctx: click.Context, log_format: str | None, log_file: Path | None) -> None:
     """Cyclone EDI suite — X12 parser."""
+    # SP18: structured JSON logging. Run once per CLI invocation; each
+    # subcommand still gets its own --log-level to override.
+    json_format = True
+    if log_format == "dev":
+        json_format = False
+    elif os.environ.get("CYCLONE_LOG_JSON", "").lower() in ("false", "0", "no"):
+        json_format = False
+    setup_logging(
+        level=os.environ.get("CYCLONE_LOG_LEVEL", "INFO"),
+        log_file=str(log_file) if log_file else None,
+        json_format=json_format,
+    )
+    # Stash on context so subcommands can read it.
+    ctx.ensure_object(dict)
 
 
 @main.command("parse-837")
@@ -63,7 +92,9 @@ def parse_837(
     log_level: str,
 ) -> None:
     """Parse an X12 837P file into one JSON per claim."""
-    logging.basicConfig(level=getattr(logging, log_level))
+    # SP18: re-run setup so per-command --log-level overrides the
+    # group default. ``setup_logging`` is idempotent.
+    setup_logging(level=log_level)
 
     text = input_file.read_text()
     config = _payer(payer)
@@ -133,7 +164,9 @@ def parse_835(
     log_level: str,
 ) -> None:
     """Parse an X12 835 ERA file into one JSON per claim payment."""
-    logging.basicConfig(level=getattr(logging, log_level))
+    # SP18: re-run setup so per-command --log-level overrides the
+    # group default. ``setup_logging`` is idempotent.
+    setup_logging(level=log_level)
 
     text = input_file.read_text()
     config = _payer_835(payer)

backend/src/cyclone/logging_config.py

@@ -0,0 +1,379 @@
+"""SP18 — Structured JSON logging.
+
+Wraps Python's stdlib ``logging`` to emit newline-delimited JSON
+(or a dev-friendly tabular format) and to scrub obvious PHI
+patterns (NPIs, SSNs, DOBs, patient names) from the message +
+extra fields.
+
+Design choices
+--------------
+
+* **No third-party deps.** stdlib ``logging`` + ``json`` + ``re``
+  is enough. ``loguru`` / ``structlog`` were considered; both add
+  a dependency for marginal gain.
+
+* **JSON by default.** Operators running Cyclone in production
+  almost certainly want logs in a format their aggregator
+  (Loki/ELK/Vector) can parse. The dev format (``CycloneDevFormatter``)
+  is the opt-out for ``tail -f`` in dev.
+
+* **Conservative PII scrubber.** Redacts unambiguous PHI patterns
+  only. False positives are not free — an operator's diagnostic
+  dump that says ``<redacted:npi>`` instead of the actual NPI
+  makes root-causing a parse failure harder. The scrubber can be
+  disabled with ``CYCLONE_LOG_NO_PII_SCRUB=1`` for tests /
+  forensic mode.
+
+* **Idempotent setup.** :func:`setup_logging` can be called
+  multiple times (CLI re-invocation, FastAPI lifespan re-entry
+  under TestClient). Each call clears existing handlers on the
+  root logger before attaching fresh ones — so the format toggle
+  actually takes effect on the second call.
+"""
+from __future__ import annotations
+
+import json
+import logging
+import os
+import re
+import sys
+from datetime import datetime, timezone
+from logging.handlers import RotatingFileHandler
+from typing import Any, Optional
+
+
+# ---------------------------------------------------------------------------
+# Formatters
+# ---------------------------------------------------------------------------
+
+
+# Stdlib LogRecord attributes we don't want to dump into the
+# structured payload (they're noise for log consumers).
+_RESERVED_LOGRECORD_ATTRS = frozenset({
+    "args", "asctime", "created", "exc_info", "exc_text", "filename",
+    "funcName", "levelname", "levelno", "lineno", "module", "msecs",
+    "message", "msg", "name", "pathname", "process", "processName",
+    "relativeCreated", "stack_info", "thread", "threadName",
+    "taskName",
+})
+
+
+class JsonFormatter(logging.Formatter):
+    """Format a LogRecord as a single JSON line.
+
+    Fields:
+        ts      — ISO 8601 UTC timestamp with milliseconds.
+        level   — uppercase level name (INFO, WARNING, etc.).
+        logger  — the logger name (e.g. "cyclone.scheduler").
+        msg     — the formatted log message (after %-substitution).
+        extra   — dict of any non-reserved LogRecord attributes.
+
+    If ``exc_info`` is set, the formatter appends a ``traceback``
+    field with the formatted exception text (NOT a serialized
+    object — just the stdlib-rendered string).
+    """
+
+    def format(self, record: logging.LogRecord) -> str:
+        ts = datetime.fromtimestamp(record.created, tz=timezone.utc).isoformat(
+            timespec="milliseconds",
+        )
+        payload: dict[str, Any] = {
+            "ts": ts,
+            "level": record.levelname,
+            "logger": record.name,
+            "msg": record.getMessage(),
+        }
+        # Collect user-provided extras.
+        extras = {
+            k: v
+            for k, v in record.__dict__.items()
+            if k not in _RESERVED_LOGRECORD_ATTRS and not k.startswith("_")
+        }
+        if extras:
+            payload["extra"] = extras
+        if record.exc_info:
+            payload["traceback"] = self.formatException(record.exc_info)
+        if record.stack_info:
+            payload["stack"] = self.formatStack(record.stack_info)
+        return json.dumps(payload, default=str, sort_keys=True)
+
+
+class CycloneDevFormatter(logging.Formatter):
+    """Dev-friendly tabular format.
+
+    Example:
+        2026-06-21T15:30:00.123Z INFO  cyclone.scheduler Processed inbound foo.x12 parser=parse_999 claims=3
+
+    Same fields as ``JsonFormatter`` but human-readable. Useful for
+    ``tail -f cyclone.log`` in dev.
+    """
+
+    def format(self, record: logging.LogRecord) -> str:
+        ts = datetime.fromtimestamp(record.created, tz=timezone.utc).isoformat(
+            timespec="milliseconds",
+        )
+        extras = {
+            k: v
+            for k, v in record.__dict__.items()
+            if k not in _RESERVED_LOGRECORD_ATTRS and not k.startswith("_")
+        }
+        extra_str = ""
+        if extras:
+            pairs = " ".join(f"{k}={v!r}" for k, v in extras.items())
+            extra_str = " " + pairs
+        base = f"{ts} {record.levelname:<7s} {record.name} {record.getMessage()}{extra_str}"
+        if record.exc_info:
+            base += "\n" + self.formatException(record.exc_info)
+        return base
+
+
+# ---------------------------------------------------------------------------
+# PII scrubber
+# ---------------------------------------------------------------------------
+
+
+# Conservative PHI patterns. Each pattern is (label, compiled regex,
+# replacement). Some patterns use a backreference so the field name
+# (e.g. "dob=") is preserved and only the value is redacted — that
+# keeps the surrounding context readable in the log line.
+_PII_PATTERNS: tuple[tuple[str, "re.Pattern[str]", str], ...] = (
+    # 10-digit NPI. Word-boundary anchored so we don't redact, e.g.,
+    # the "10" in "10 claims processed".
+    ("npi", re.compile(r"\b\d{10}\b"), "<redacted:npi>"),
+    # SSN: NNN-NN-NNNN or NNNNNNNNN.
+    (
+        "ssn",
+        re.compile(r"\b\d{3}-\d{2}-\d{4}\b|\b\d{9}\b(?=[\s,;)}])"),
+        "<redacted:ssn>",
+    ),
+    # DOB: "dob=YYYY-MM-DD" / "date_of_birth=YYYY-MM-DD". Capture the
+    # field name + separator, redact only the date — keeps the
+    # surrounding sentence readable.
+    (
+        "dob",
+        re.compile(
+            r"(?i)(\b(?:dob|date[ _]?of[ _]?birth)[:=]\s*)\d{4}-\d{2}-\d{2}"
+        ),
+        r"\1<redacted:dob>",
+    ),
+    # Patient name: explicit field marker, redact the whole
+    # "patient_name=..." chunk so the value can't leak in a quoted form.
+    (
+        "patient_name",
+        re.compile(
+            r'(?i)\bpatient[_ ]?name[:=]\s*"?[^\",\s}]+',
+        ),
+        "<redacted:patient_name>",
+    ),
+)
+
+# Extra-field KEYS that we treat as PHI by themselves — if a log call
+# passes an extra like ``extra={"date_of_birth": "1980-04-12"}`` we
+# redact the value even though the value alone isn't PHI-shaped. The
+# key is the signal. Matched case-insensitively against the full key
+# (with underscores normalized to spaces for "date of birth").
+_PHI_EXTRA_KEYS: dict[str, str] = {
+    "npi": "npi",
+    "provider_npi": "npi",
+    "rendering_npi": "npi",
+    "billing_npi": "npi",
+    "ssn": "ssn",
+    "dob": "dob",
+    "date_of_birth": "dob",
+    "patient_name": "patient_name",
+    "patient first name": "patient_name",
+    "patient last name": "patient_name",
+}
+
+# When an extra key matches one of these, redact any string value
+# wholesale (don't try to parse it — just replace).
+_PHI_EXTRA_WHOLE_VALUE = {"npi", "ssn", "dob", "patient_name"}
+
+
+class PiiScrubber(logging.Filter):
+    """Filter that redacts obvious PHI from log records.
+
+    Walks the formatted message + every ``extra`` field value (if
+    it's a string) and rewrites matches to ``<redacted:<name>``.
+    Non-string extras are left alone (we don't try to serialize and
+    re-scrub dicts — too risky for false positives).
+    """
+
+    def __init__(self, name: str = "pii_scrubber") -> None:
+        super().__init__(name)
+        self._enabled = True
+
+    def disable(self) -> None:
+        """Disable scrubbing (for tests / forensic mode)."""
+        self._enabled = False
+
+    def enable(self) -> None:
+        self._enabled = True
+
+    def _scrub(self, text: str) -> str:
+        for label, pat, repl in _PII_PATTERNS:
+            text = pat.sub(repl, text)
+        return text
+
+    @staticmethod
+    def _normalize_extra_key(key: str) -> set[str]:
+        """Return all candidate normalizations of a key.
+
+        ``date_of_birth`` should match a lookup table that uses either
+        ``date_of_birth`` or ``date of birth`` — so return both. Same
+        for ``patient_name`` vs ``patient name``.
+        """
+        norm = key.strip().lower()
+        spaced = norm.replace("_", " ")
+        return {norm, spaced}
+
+    def _redact_extra_value(self, key: str, value: Any) -> Any:
+        """Redact a single extra field value if its key signals PHI."""
+        for norm in self._normalize_extra_key(key):
+            label = _PHI_EXTRA_KEYS.get(norm)
+            if label:
+                if not isinstance(value, str):
+                    return value
+                return f"<redacted:{label}>"
+        return value
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        if not self._enabled:
+            return True
+        # Scrub the formatted message.
+        try:
+            msg = record.getMessage()
+            scrubbed_msg = self._scrub(msg)
+            if scrubbed_msg != msg:
+                record.msg = scrubbed_msg
+                record.args = ()
+        except Exception:  # noqa: BLE001
+            pass  # never let the scrubber crash a log call
+        # Scrub string extras in place. We mutate the record's
+        # __dict__ directly so the formatter sees the scrubbed value.
+        for k, v in list(record.__dict__.items()):
+            if k in _RESERVED_LOGRECORD_ATTRS or k.startswith("_"):
+                continue
+            # First, key-based redaction (covers `extra={"dob": "..."}`).
+            redacted = self._redact_extra_value(k, v)
+            if redacted is not v:
+                record.__dict__[k] = redacted
+                continue
+            # Second, value-pattern redaction (covers `extra={"note":
+            # "patient_name=John Doe"}`).
+            if isinstance(v, str):
+                scrubbed = self._scrub(v)
+                if scrubbed != v:
+                    record.__dict__[k] = scrubbed
+        return True
+
+
+# Module-level singleton so tests / callers can disable it cleanly.
+_scrubber = PiiScrubber()
+
+
+def get_scrubber() -> PiiScrubber:
+    """Return the module-level PII scrubber singleton."""
+    return _scrubber
+
+
+# ---------------------------------------------------------------------------
+# setup_logging entry point
+# ---------------------------------------------------------------------------
+
+
+def _resolve_level(level: str | int | None) -> int:
+    """Resolve a level string/int, falling back to INFO."""
+    if level is None:
+        return logging.INFO
+    if isinstance(level, int):
+        return level
+    name = str(level).strip().upper()
+    return logging.getLevelNamesMapping().get(name, logging.INFO)
+
+
+def setup_logging(
+    *,
+    level: str | int | None = None,
+    log_file: str | None = None,
+    json_format: bool = True,
+    scrub_pii: bool = True,
+    propagate_from: str | None = None,
+) -> logging.Logger:
+    """Configure the root logger + attach handlers.
+
+    Idempotent: re-calling clears existing handlers on the root
+    logger before attaching fresh ones. Safe to call from
+    ``click.command`` invocations and the FastAPI lifespan.
+
+    Args:
+        level: ``"DEBUG"`` / ``"INFO"`` / etc. or an int. ``None``
+            means honor ``CYCLONE_LOG_LEVEL`` env var, then INFO.
+        log_file: Path to a rotating log file. ``None`` means
+            honor ``CYCLONE_LOG_FILE`` env var, then stderr.
+        json_format: Emit JSON lines (default). ``False`` uses
+            :class:`CycloneDevFormatter`.
+        scrub_pii: Apply the PII scrubber (default). Honored via
+            ``CYCLONE_LOG_NO_PII_SCRUB=1`` to disable.
+        propagate_from: Optional logger name to attach the scrubber
+            to (defaults to root).
+
+    Returns:
+        The configured root logger.
+    """
+    # Resolve env-var defaults.
+    if level is None:
+        level = os.environ.get("CYCLONE_LOG_LEVEL", "INFO")
+    if log_file is None:
+        log_file = os.environ.get("CYCLONE_LOG_FILE") or None
+    if not json_format and os.environ.get("CYCLONE_LOG_JSON", "").lower() in (
+        "false", "0", "no",
+    ):
+        json_format = True
+    if os.environ.get("CYCLONE_LOG_NO_PII_SCRUB", "").lower() in ("1", "true", "yes"):
+        scrub_pii = False
+
+    root = logging.getLogger()
+    root.setLevel(_resolve_level(level))
+
+    # Clear existing handlers (idempotent re-setup).
+    for h in list(root.handlers):
+        root.removeHandler(h)
+    # Also clear our scrubber so we don't add duplicates.
+    target = logging.getLogger(propagate_from) if propagate_from else root
+    for flt in list(target.filters):
+        if isinstance(flt, PiiScrubber):
+            target.removeFilter(flt)
+
+    # Build the formatter.
+    fmt: logging.Formatter
+    if json_format:
+        fmt = JsonFormatter()
+    else:
+        fmt = CycloneDevFormatter()
+
+    # Build the handler.
+    if log_file:
+        handler: logging.Handler = RotatingFileHandler(
+            log_file,
+            maxBytes=10 * 1024 * 1024,
+            backupCount=5,
+            encoding="utf-8",
+        )
+    else:
+        handler = logging.StreamHandler(stream=sys.stderr)
+    handler.setFormatter(fmt)
+    root.addHandler(handler)
+
+    # Attach the scrubber.
+    if scrub_pii:
+        _scrubber.enable()
+    else:
+        _scrubber.disable()
+    target.addFilter(_scrubber)
+
+    # Quiet down noisy third-party libs.
+    for noisy in ("urllib3", "paramiko", "sqlalchemy.engine"):
+        logging.getLogger(noisy).setLevel(max(root.level, logging.WARNING))
+
+    return root

backend/src/cyclone/scheduler.py

@@ -424,8 +424,11 @@ class Scheduler:
         self._stop_event.clear()
         self._task = asyncio.create_task(self._run(), name="mft-scheduler")
         log.info(
-            "Scheduler started: poll every %ds, sftp_block=%r",
-            self._poll_interval, self._sftp_block_name,
+            "Scheduler started",
+            extra={
+                "poll_interval_s": self._poll_interval,
+                "sftp_block": self._sftp_block_name,
+            },
         )
 
     async def stop(self) -> None:
@@ -500,7 +503,7 @@ class Scheduler:
                 # tick() should never raise — it catches per-file
                 # exceptions. This is the safety net for SFTP outages
                 # or DB connectivity issues.
-                log.exception("Scheduler tick raised: %s", exc)
+                log.exception("Scheduler tick raised", extra={"error": str(exc)})
             try:
                 await asyncio.wait_for(
                     self._stop_event.wait(),
@@ -565,7 +568,7 @@ class Scheduler:
                 self._download_and_parse, f, file_type,
             )
         except Exception as exc:  # noqa: BLE001
-            log.exception("Failed to process %s", f.name)
+            log.exception("Failed to process inbound file", extra={"input_filename": f.name})
             await self._record(
                 name=f.name, size=f.size, modified_at=f.modified_at,
                 file_type=file_type, parser_used=None, claim_count=0,
@@ -586,8 +589,12 @@ class Scheduler:
         )
         result.files_processed += 1
         log.info(
-            "Processed inbound %s: parser=%s claims=%d",
-            f.name, parser_used, claim_count,
+            "Processed inbound file",
+            extra={
+                "input_filename": f.name,
+                "parser": parser_used,
+                "claims": claim_count,
+            },
         )
 
     async def _already_processed(self, name: str) -> bool: