unified_debug.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_diagnostics/unified_debug.py
 
"""
Unified Debugging System
========================
 
Purpose
-------
Provide a single, centralized ``_dbg()`` debug logger for Option C projects,
together with per-domain level filtering, ANSI color output, timestamped log
files, stdout/stderr interception, and an uncaught-exception hook.
 
Responsibilities
----------------
- Load debug configuration from ``diagnostics/debug_config.json`` (project
  root) or the bundled fallback, recording ``CONFIG_SOURCE`` for the log header.
- Provide ``_dbg()`` with domain/level filtering, caller-site reporting, and
  optional override mode.
- Write formatted output to terminal (ANSI colors retained) and/or a log file
  (all ANSI escape sequences stripped) based on ``OUTPUT_MODE``.
- Provide ``_StreamInterceptor`` wrapping ``sys.stdout``/``sys.stderr`` with a
  full ``io.TextIOBase``-compatible interface and a thread-local re-entrant
  recursion guard.
- Provide ``init()`` to activate stream interception and the exception hook
  explicitly; importing this module has zero side effects.
- Provide ``instrumentation_active()``, ``instrumentation_verbose()``, and
  ``instrumentation_allow_custom_components()`` helpers consumed by other modules.
 
Diagnostics
-----------
Domain: GENERAL
Levels:
    L2 — lifecycle
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- Importing this module must have zero side effects on ``sys.stdout``,
  ``sys.stderr``, or ``sys.excepthook``.
- ``init()`` must be idempotent; calling it more than once is a no-op.
- ``_dbg()`` must never raise; all internal errors are silently swallowed.
- ``CONFIG_SOURCE`` must be a non-empty string after module load.
 
Configuration
-------------
Debug domains, levels, colors, and global settings are loaded from
`diagnostics/debug_config.json` in the project root if present, or from
`debug_config.json` in the same directory as this module as a fallback.
 
Each project supplies its own `diagnostics/debug_config.json` with
project-specific domains. This keeps the library project-independent.
 
Environment Variables
---------------------
``OC_DIAGNOSTICS_CONFIG``
    Absolute path to a ``debug_config.json`` file.  When set, this path is
    used instead of ``<cwd>/diagnostics/debug_config.json``.  Useful for
    embedded usage, subprocess invocation, or CI environments where the
    working directory is not the project root.
 
``OC_DIAGNOSTICS_LOG_DIR``
    Absolute path to the directory where log files are written.  When set,
    this replaces the default ``<cwd>/logs`` directory.
 
``UNIFIED_DEBUG_DISABLE_EXCEPTION_HOOK``
    Set to ``1`` to prevent ``init()`` from installing the custom
    ``sys.excepthook``, even when ``intercept_exceptions=True`` is passed.
    Useful in environments (pytest, debuggers, ASGI frameworks) that install
    their own exception handlers and must not be displaced.
 
Log Format
----------
When filename inclusion is enabled:
 
    [DOMAIN][L2][file.py:123::caller=foo|logical context=bar] 14:03:12.345 Message...
 
When disabled:
 
    [DOMAIN][L2][logical context=bar] 14:03:12.345 Message...
"""
 
from __future__ import annotations
 
import asyncio
import datetime
import functools
import io
import json
import os
import re
import sys
import threading
import time
import traceback
import warnings
from contextlib import contextmanager
 
 
# ============================================================
# COLOR NAME → ANSI CODE MAPPING
# ============================================================
 
_COLOR_NAME_MAP = {
    "red":     "\033[91m",
    "green":   "\033[92m",
    "yellow":  "\033[93m",
    "blue":    "\033[94m",
    "magenta": "\033[95m",
    "cyan":    "\033[96m",
    "white":   "\033[97m",
    "gray":    "\033[90m",
    "default": "\033[0m",
}
 
 
# ============================================================
# CONFIG SCHEMA (FS-18: optional jsonschema validation)
# ============================================================
 
_DOMAIN_SCHEMA = {
    "type": "object",
    "additionalProperties": {
        "oneOf": [
            {"type": "integer", "minimum": 0, "maximum": 9},
            {
                "type": "object",
                "properties": {
                    "level": {"type": "integer", "minimum": 0, "maximum": 9},
                    "color": {"type": "string"},
                },
                "required": ["level"],
            },
        ]
    },
}
 
_CONFIG_SCHEMA = {
    "type": "object",
    "properties": {
        # V2.0 spec key
        "DEBUG_LEVELS": _DOMAIN_SCHEMA,
        # Legacy key (backwards compat)
        "domains": _DOMAIN_SCHEMA,
        "global_settings": {"type": "object"},
    },
}
 
# FS-09: config path and mtime tracked so the hot-reload watcher can detect changes.
_CONFIG_PATH: str = ""
_CONFIG_MTIME: float = 0.0
 
# OI-515: hardening for ``OC_DIAGNOSTICS_*`` env vars. The config loader and
# log-directory resolver accept operator-supplied paths, so we apply three
# containment rules before trusting them:
#   * length cap (``_MAX_ENV_PATH_LEN``) — refuses absurd strings that
#     likely originate from shell-splice accidents.
#   * NUL byte rejection — POSIX paths cannot embed ``\0``.
#   * size cap (``_MAX_CONFIG_BYTES``) — a 10 MiB ceiling on the config
#     file so a compromised or runaway env var cannot blow memory.
_MAX_ENV_PATH_LEN: int = 4096
_MAX_CONFIG_BYTES: int = 10 * 1024 * 1024
 
 
def _safe_env_path(name: str) -> str:
    """Return ``os.environ[name]`` if it passes OI-515 sanity checks.
 
    Returns an empty string when the env var is unset, empty, contains a
    NUL byte, or exceeds :data:`_MAX_ENV_PATH_LEN`. A warning is emitted
    in the length/NUL-byte case so operators notice the rejection.
    """
    raw = os.environ.get(name, "")
    value = raw.strip()
    if not value:
        return ""
    if "\x00" in value:
        warnings.warn(
            f"oc_diagnostics: {name} contains NUL byte — ignored (OI-515)",
            UserWarning,
            stacklevel=2,
        )
        return ""
    if len(value) > _MAX_ENV_PATH_LEN:
        warnings.warn(
            f"oc_diagnostics: {name} exceeds {_MAX_ENV_PATH_LEN} chars — "
            "ignored (OI-515)",
            UserWarning,
            stacklevel=2,
        )
        return ""
    return value
 
 
# ============================================================
# CONFIG LOADER
# ============================================================
 
def _load_debug_config():
    """Load debug_config.json from the project diagnostics/ directory if present,
    otherwise fall back to debug_config.json next to this module.
 
    Returns (domains_dict, color_codes_dict, global_settings_dict, config_source_str).
    Falls back to minimal defaults if the file is missing or invalid.
    config_source_str describes which file was loaded, or why defaults were used.
    """
    global _CONFIG_PATH, _CONFIG_MTIME
 
    # Resolution order:
    #   1. OC_DIAGNOSTICS_CONFIG env var (absolute path)
    #   2. <cwd>/diagnostics/debug_config.json  (project root convention)
    #   3. debug_config.json next to this module  (bundled fallback / tests)
    env_config_path = _safe_env_path("OC_DIAGNOSTICS_CONFIG")
    project_config_path = os.path.join(
        os.getcwd(), "diagnostics", "debug_config.json"
    )
    # Fallback: next to this module (useful for tests)
    module_config_path = os.path.join(os.path.dirname(__file__), "debug_config.json")
 
    if env_config_path:
        config_path = env_config_path
        config_source = f"ENV:OC_DIAGNOSTICS_CONFIG={env_config_path}"
    elif os.path.exists(project_config_path):
        config_path = project_config_path
        config_source = project_config_path
    else:
        config_path = module_config_path
        config_source = f"DEFAULTS (project config not found at: {project_config_path})"
 
    # FS-09: track resolved config path for hot-reload watcher
    _CONFIG_PATH = config_path
    try:
        _CONFIG_MTIME = os.path.getmtime(config_path)
    except OSError:
        _CONFIG_MTIME = 0.0
 
    # Fallback defaults
    default_levels = {"GENERAL": 3}
    default_colors = {"DEFAULT": "\033[0m", "ERROR": "\033[91m"}
    default_settings = {
        "silent_mode": False,
        "enable_timestamps": True,
        "enable_colors": True,
        "include_filename": True,
        "output_mode": "both",
        "instrumentation_enabled": False,
        "structural_id_prefixes": [],
        "structural_id_exact": [],
        "max_log_files": 0,
        "max_log_size_mb": 0.0,
        "instrumentation_allow_all": False,
        "ui_event_properties": [],
        "http_redact_headers": ["authorization", "cookie", "set-cookie", "x-api-key"],
        "mode": "",
        "prod_max_level": 2,
    }
 
    # OI-515: refuse oversized config files (10 MiB cap) so a runaway
    # or malicious env var cannot blow memory at import time.
    try:
        size = os.path.getsize(config_path)
        if size > _MAX_CONFIG_BYTES:
            warnings.warn(
                f"oc_diagnostics: config {config_path} exceeds "
                f"{_MAX_CONFIG_BYTES} bytes ({size}) — using defaults (OI-515)",
                UserWarning,
                stacklevel=2,
            )
            config_source = f"DEFAULTS (config > {_MAX_CONFIG_BYTES} bytes)"
            return default_levels, default_colors, default_settings, config_source
    except OSError:
        # Missing/unreadable falls through to json.load which raises
        # FileNotFoundError below; keep the original control flow.
        pass
 
    try:
        with open(config_path, "r", encoding="utf-8") as f:
            config = json.load(f)
    except (FileNotFoundError, json.JSONDecodeError) as exc:
        config_source = f"DEFAULTS (load error: {exc})"
        return default_levels, default_colors, default_settings, config_source
 
    # FS-18: optional jsonschema validation (silently skipped if not installed)
    try:
        import jsonschema as _jsonschema
        try:
            _jsonschema.validate(config, _CONFIG_SCHEMA)
        except _jsonschema.ValidationError as _ve:
            warnings.warn(
                f"oc_diagnostics: debug_config.json validation error: {_ve.message}",
                UserWarning,
                stacklevel=2,
            )
    except ImportError:
        pass
 
    # Parse domains → DEBUG_LEVELS and COLOR_CODES
    # V2.0 schema: accept both "DEBUG_LEVELS" (spec) and "domains" (legacy).
    # Prefer "DEBUG_LEVELS" if both are present.
    domains = config.get("DEBUG_LEVELS") or config.get("domains", {})
    levels = {}
    colors = {"DEFAULT": "\033[0m", "ERROR": "\033[91m"}
 
    for domain_name, domain_def in domains.items():
        if isinstance(domain_def, dict):
            levels[domain_name] = domain_def.get("level", 3)
            color_name = domain_def.get("color", "default")
            colors[domain_name] = _COLOR_NAME_MAP.get(color_name, "\033[0m")
        elif isinstance(domain_def, int):
            # Simple format: "DOMAIN": 3
            levels[domain_name] = domain_def
            colors[domain_name] = "\033[0m"
 
    # Parse global settings
    raw_settings = config.get("global_settings", {})
 
    # V2.0 schema: accept nested log_rotation object alongside flat keys.
    log_rotation = raw_settings.get("log_rotation", {})
    max_log_files = int(
        log_rotation.get("max_files", 0)
        or raw_settings.get("max_log_files", 0)
    )
    max_log_size_mb = float(
        log_rotation.get("max_size_mb", 0)
        or raw_settings.get("max_log_size_mb", 0)
    )
 
    # V2.0 schema: log_to_file / log_file_path / log_format
    log_to_file = raw_settings.get("log_to_file", None)
    log_file_path = raw_settings.get("log_file_path", None)
    log_format = raw_settings.get("log_format", None)
 
    # Resolve output_mode from V2.0 keys if present, otherwise use legacy key
    output_mode = raw_settings.get("output_mode", "both")
    if log_format == "jsonl":
        output_mode = "json"
    elif log_to_file is False:
        output_mode = "terminal"
    elif log_to_file is True and output_mode == "both":
        pass  # keep "both"
 
    settings = {
        "silent_mode": raw_settings.get("silent_mode", False),
        "enable_timestamps": raw_settings.get("enable_timestamps", True),
        "enable_colors": raw_settings.get("enable_colors", True),
        "include_filename": raw_settings.get("include_filename", True),
        "output_mode": output_mode,
        "instrumentation_enabled": raw_settings.get("instrumentation_enabled", False),
        # FS-02 / OI-17: project-defined extensions to the structural-ID lists.
        "structural_id_prefixes": list(raw_settings.get("structural_id_prefixes", [])),
        "structural_id_exact": list(raw_settings.get("structural_id_exact", [])),
        # FS-06: log rotation (supports nested V2.0 and flat legacy keys)
        "max_log_files": max_log_files,
        "max_log_size_mb": max_log_size_mb,
        # FS-13: allow instrumenting all components regardless of type
        "instrumentation_allow_all": bool(raw_settings.get("instrumentation_allow_all", False)),
        # FS-10: configurable UI event properties
        "ui_event_properties": list(raw_settings.get("ui_event_properties", [])),
        # FS-17: headers to redact in HTTP L4 logging
        "http_redact_headers": list(raw_settings.get("http_redact_headers", [
            "authorization", "cookie", "set-cookie", "x-api-key"
        ])),
        # FS-15: deployment mode ("prod" / "production" / "safe" suppresses interception)
        "mode": raw_settings.get("mode", "").strip().lower(),
        # OI-407: fail-safe ceiling for _dbg() levels in production mode.
        "prod_max_level": raw_settings.get("prod_max_level", 2),
        # V2.0: global debug level override (None = disabled)
        "global_debug_level": raw_settings.get("global_debug_level", None),
        # V2.0: hot-reload from config (runtime-only watch_config also accepted)
        "watch_config": bool(raw_settings.get("watch_config", False)),
        # V2.0: AI-Trace shadow logging
        "ai_trace_level": bool(raw_settings.get("ai_trace_level", False)),
        # V2.0: production-mode redaction patterns
        "redaction_patterns": list(raw_settings.get("redaction_patterns", [])),
        # OI-506: master toggle for built-in secret/URL/token redaction
        "redact_secrets": bool(raw_settings.get("redact_secrets", True)),
        # V2.0: log file path override from config
        "log_file_path": log_file_path,
    }
 
    # V2.0: global_debug_level overrides all domain levels
    gdl = settings["global_debug_level"]
    if gdl is not None and isinstance(gdl, int) and 0 <= gdl <= 9:
        for domain_name in levels:
            levels[domain_name] = gdl
 
    return levels, colors, settings, config_source
 
 
# ============================================================
# GLOBAL SETTINGS (loaded from config)
# ============================================================
 
_levels, _colors, _settings, _config_source = _load_debug_config()
 
DEBUG_LEVELS = _levels
COLOR_CODES = _colors
 
CONFIG_SOURCE = _config_source
INSTRUMENTATION_ENABLED = _settings["instrumentation_enabled"]
SILENT_MODE = _settings["silent_mode"]
 
# FS-02 / OI-17: project-defined structural ID extensions loaded from config.
# id_instrumentation.is_structural_id() merges these with the built-in lists.
ADDITIONAL_STRUCTURAL_PREFIXES: tuple[str, ...] = tuple(
    _settings.get("structural_id_prefixes", [])
)
ADDITIONAL_STRUCTURAL_EXACT: frozenset[str] = frozenset(
    _settings.get("structural_id_exact", [])
)
ENABLE_TIMESTAMPS = _settings["enable_timestamps"]
ENABLE_COLORS = _settings["enable_colors"]
INCLUDE_FILENAME = _settings["include_filename"]
OUTPUT_MODE = _settings["output_mode"]
 
# FS-06: log rotation limits (0 = disabled)
MAX_LOG_FILES: int = _settings["max_log_files"]
MAX_LOG_SIZE_BYTES: int = (
    int(_settings["max_log_size_mb"] * 1024 * 1024)
    if _settings["max_log_size_mb"] > 0 else 0
)
 
# FS-17: header names to redact in HTTP L4 logging (always lowercase for comparison)
HTTP_REDACT_HEADERS: frozenset[str] = frozenset(
    h.lower() for h in _settings.get("http_redact_headers",
    ["authorization", "cookie", "set-cookie", "x-api-key"])
)
 
# FS-13: when True, instrument_ids() bypasses the _is_dash_component() guard
INSTRUMENTATION_ALL: bool = _settings.get("instrumentation_allow_all", False)
 
# FS-10: UI event properties to watch; falls back to built-in defaults if not configured
_DEFAULT_UI_EVENT_PROPERTIES: tuple[str, ...] = (
    "n_clicks", "value", "data", "figure", "selectedData"
)
UI_EVENT_PROPERTIES: tuple[str, ...] = tuple(
    _settings.get("ui_event_properties") or _DEFAULT_UI_EVENT_PROPERTIES
)
 
# FS-15: production mode — env var takes priority, config "mode" key is the fallback.
# In prod mode, init() skips stream interception and the exception hook.
PROD_MODE: bool = (
    os.environ.get("OC_DIAGNOSTICS_MODE", "").strip().lower() in ("prod", "production", "safe")
    or _settings.get("mode", "") in ("prod", "production", "safe")
)
 
# OI-407: Production-mode fail-safe ceiling for _dbg() levels.
# In PROD_MODE, any _dbg() call with a level above PROD_MAX_LEVEL is
# suppressed regardless of per-domain settings. A one-shot warning is
# emitted per (domain, level) key to make the downgrade visible without
# log spam. Protects against accidental L3/L4 verbosity in production.
try:
    PROD_MAX_LEVEL: int = int(_settings.get("prod_max_level", 2))
except (TypeError, ValueError):
    PROD_MAX_LEVEL = 2
_prod_downgrade_warned: set[str] = set()
 
# AI-Trace: shadow logging level for AI agent monitoring.
# When enabled, _dbg() calls with level="ai_trace" write to file/JSONL but
# skip console output.  Loaded from config global_settings.ai_trace_level.
AI_TRACE_ENABLED: bool = bool(_settings.get("ai_trace_level", False))
 
# V2.0: OC_DIAGNOSTICS_LEVEL env var overrides all domain levels at load time.
_env_level = os.environ.get("OC_DIAGNOSTICS_LEVEL", "").strip()
if _env_level.isdigit() and 0 <= int(_env_level) <= 9:
    _env_level_int = int(_env_level)
    for _k in DEBUG_LEVELS:
        DEBUG_LEVELS[_k] = _env_level_int
 
# V2.0: redaction patterns for production-mode output scrubbing.
# Each pattern is matched case-insensitively against key=value or key: value
# pairs in log output; matched values are replaced with [REDACTED].
REDACTION_PATTERNS: list[re.Pattern] = []
for _pat in _settings.get("redaction_patterns", []):
    try:
        REDACTION_PATTERNS.append(
            re.compile(
                rf'(?i)({re.escape(_pat)})\s*[=:]\s*(?:"[^"]*"|\'[^\']*\'|\S+)',
            )
        )
    except re.error:
        pass
 
# OI-506 / FS-506: master toggle for built-in secret/URL/token redaction.
# Applies to the three always-on regexes below (git creds, token prefixes,
# key/value secrets). User-defined ``REDACTION_PATTERNS`` run regardless.
REDACT_SECRETS: bool = bool(_settings.get("redact_secrets", True))
 
# OI-506: http(s) URLs carrying ``user[:token]@host`` credentials.
# Mirrors oct.core.git._CRED_URL_RE so the same invariant applies inside
# oc_diagnostics without a cross-package import.
_GIT_URL_RE: re.Pattern[str] = re.compile(
    r"(?P<scheme>https?://)[^/@\s]+@",
)
 
# OI-506: common opaque-secret prefixes (GitHub PATs, Slack bots, OpenAI
# keys, JWTs). Matches the prefix plus the token body up to whitespace.
_TOKEN_PREFIXES_RE: re.Pattern[str] = re.compile(
    r"\b(?P<prefix>ghp_|gho_|ghu_|ghs_|ghr_|sk-|xoxb-|xoxp-|eyJ)[A-Za-z0-9_\-\.]{10,}",
)
 
# OI-506: ``secret=value`` / ``secret: value`` style key-value pairs.
# Case-insensitive; the value may be quoted or a bare non-whitespace run.
_KV_SECRET_RE: re.Pattern[str] = re.compile(
    r"(?i)\b(?P<key>token|secret|api[_-]?key|password|passwd|"
    r"auth[_-]?token|access[_-]?token|bearer)\s*[=:]\s*"
    r"(?:\"[^\"]*\"|'[^']*'|\S+)",
)
 
# Logs go into ./logs relative to the current working directory,
# or into OC_DIAGNOSTICS_LOG_DIR if that env var is set.
# OI-515: env-provided paths run through _safe_env_path so oversized or
# NUL-bearing values fall back to the default instead of propagating.
LOG_DIR = _safe_env_path("OC_DIAGNOSTICS_LOG_DIR") or os.path.join(os.getcwd(), "logs")
 
# V2.0: OC_DIAGNOSTICS_LOG_FILE env var overrides log file path entirely.
# When set, LOG_FILE is fixed to this value (not lazily computed).
# Also check config log_file_path as fallback.
_env_log_file = _safe_env_path("OC_DIAGNOSTICS_LOG_FILE")
_config_log_file_path = _settings.get("log_file_path")
 
# LOG_FILE is computed lazily on the first write so that pre-fork server models
# (e.g. Gunicorn with multiple workers) each resolve their own unique path after
# fork(), preventing workers from sharing a file and interleaving output.
LOG_FILE: str | None = _env_log_file or _config_log_file_path or None
 
ORIGINAL_STDOUT = sys.stdout
ORIGINAL_STDERR = sys.stderr
 
_header_written = False
_log_lock = threading.Lock()
 
 
def _get_log_file() -> str:
    """Return the LOG_FILE path, computing it on the first call.
 
    Always invoked from within a ``_log_lock``-protected block so the lazy
    one-time initialization is thread-safe.  After the first call ``LOG_FILE``
    is a non-None string and subsequent calls are a single None-check.
    """
    global LOG_FILE
    if LOG_FILE is None:
        LOG_FILE = os.path.join(
            LOG_DIR,
            f"unified_debug_logs_{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.txt",
        )
    return LOG_FILE
 
# Precompiled regex that strips all ANSI/VT100 escape sequences from log output.
# Covers:
#   \033[...letter  — CSI sequences (SGR colors, cursor movement, clear-screen, etc.)
#   \033]...BEL     — OSC sequences (terminal hyperlinks, window titles)
#   \033]...ESC\    — OSC sequences with ST (String Terminator) instead of BEL
#   \033.           — Two-character escapes (ESC c, ESC M, ESC = / > etc.)
_ANSI_ESCAPE_RE = re.compile(
    r'\033(?:\‍[[^A-Za-z]*[A-Za-z]|\‍][^\007\033]*(?:\007|\033\\)|.)'
)
 
 
# ============================================================
# INTERNAL HELPERS
# ============================================================
 
def _get_timestamp() -> str:
    if not ENABLE_TIMESTAMPS:
        return ""
    ts = datetime.datetime.now().strftime("%H:%M:%S.%f")[:-3]
    return f"{ts} "
 
 
def _get_color(domain: str, level: int) -> str:
    if not ENABLE_COLORS:
        return ""
    if level == 1:
        return COLOR_CODES.get("ERROR", "\033[91m")
    return COLOR_CODES.get(domain, COLOR_CODES.get("DEFAULT", "\033[0m"))
 
 
def _reset_color() -> str:
    return COLOR_CODES.get("DEFAULT", "\033[0m") if ENABLE_COLORS else ""
 
 
def _prune_old_log_files(log_dir: str, max_count: int) -> None:
    """Delete the oldest log files when the count exceeds max_count. (FS-06)"""
    try:
        import glob as _glob
        pattern = os.path.join(log_dir, "unified_debug_logs_*.txt")
        files = sorted(_glob.glob(pattern), key=os.path.getmtime)
        for path in files[:-max_count]:
            try:
                os.remove(path)
            except OSError:
                pass
    except Exception:
        pass
 
 
def _ensure_log_dir_and_header():
    global _header_written
    with _log_lock:
        if _header_written:
            return
 
        try:
            log_file = _get_log_file()   # resolve path lazily, under the lock (OI-22)
            os.makedirs(LOG_DIR, exist_ok=True)
            # FS-06: prune excess log files on each new file creation
            if MAX_LOG_FILES > 0:
                _prune_old_log_files(LOG_DIR, MAX_LOG_FILES)
            is_new_file = not os.path.exists(log_file)
 
            with open(log_file, "a", encoding="utf-8") as f:
                if is_new_file:
                    now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                    f.write("============================================================\n")
                    f.write(" Unified Debug Log\n")
                    f.write("============================================================\n")
                    f.write(f"Created at      : {now}\n")
                    f.write(f"Log file        : {log_file}\n")
                    f.write(f"Config source   : {CONFIG_SOURCE}\n")
                    f.write(f"SILENT_MODE     : {SILENT_MODE}\n")
                    f.write(f"INSTRUMENTATION_ENABLED: {INSTRUMENTATION_ENABLED}\n")
                    f.write(f"ENABLE_TIMESTAMPS: {ENABLE_TIMESTAMPS}\n")
                    f.write(f"INCLUDE_FILENAME: {INCLUDE_FILENAME}\n")
                    f.write(f"ENABLE_COLORS   : {ENABLE_COLORS}\n")
                    f.write(f"OUTPUT_MODE     : {OUTPUT_MODE}\n")
                    f.write("DEBUG_LEVELS    :\n")
                    for dom, lvl in sorted(DEBUG_LEVELS.items()):
                        f.write(f"  - {dom}: {lvl}\n")
                    f.write("============================================================\n\n")
 
            _header_written = True
        except Exception:
            pass
 
 
def _write_to_file(line: str):
    global LOG_FILE, _header_written, OUTPUT_MODE, _io_fallback_warned
    try:
        # FS-06: size-based rotation — reset lazily before header guard so the
        # next _ensure_log_dir_and_header() call creates a fresh timestamped file.
        if MAX_LOG_SIZE_BYTES > 0:
            with _log_lock:
                lf = _get_log_file()
                try:
                    if os.path.exists(lf) and os.path.getsize(lf) > MAX_LOG_SIZE_BYTES:
                        LOG_FILE = None
                        _header_written = False
                except OSError:
                    pass
        _ensure_log_dir_and_header()
        with _log_lock:                                      # OI-21: serialize concurrent writes
            with open(_get_log_file(), "a", encoding="utf-8") as f:
                f.write(line + "\n")
    except OSError as _exc:
        # FS-25: one-time warning; fall back to terminal-only mode so console
        # output is preserved.  Use ORIGINAL_STDOUT directly to avoid recursion.
        if not _io_fallback_warned:
            _io_fallback_warned = True
            try:
                ORIGINAL_STDOUT.write(
                    f"\n[oc_diagnostics] WARNING: file log write failed "
                    f"({_exc}); switching OUTPUT_MODE to 'terminal'.\n"
                )
                ORIGINAL_STDOUT.flush()
            except Exception:
                pass
            OUTPUT_MODE = "terminal"
    except Exception:
        pass
 
 
def _apply_redaction(text: str) -> str:
    """Apply configured redaction patterns to log output (production mode only).
 
    OI-506 / FS-506 layer order (production mode):
 
    1. Built-in secret/URL/token scrubbers (gated by ``REDACT_SECRETS``):
       git ``user:token@`` credentials, common token prefixes (``ghp_``,
       ``sk-``, ``xoxb-``, ``eyJ``...), and ``secret=value`` KV pairs.
    2. User-defined ``REDACTION_PATTERNS`` from config.
 
    The built-ins run unconditionally when ``REDACT_SECRETS`` is True so
    production logs cannot leak credentials even if the operator forgot
    to configure a pattern list. The toggle exists purely as an escape
    hatch for environments that redact at the shipping layer instead.
    """
    if not PROD_MODE:
        return text
    if REDACT_SECRETS:
        text = _GIT_URL_RE.sub(r"\g<scheme>***@", text)
        text = _TOKEN_PREFIXES_RE.sub(r"\g<prefix>[REDACTED]", text)
        text = _KV_SECRET_RE.sub(lambda m: f"{m.group('key')}=[REDACTED]", text)
    for pattern in REDACTION_PATTERNS:
        text = pattern.sub(lambda m: f"{m.group(1)}=[REDACTED]", text)
    return text
 
 
def _emit(line: str, level: int, record: dict | None = None):
    if SILENT_MODE:
        return
 
    line = _apply_redaction(line)
 
    # Terminal output: KEEP ANSI COLORS (shown for all modes including "json")
    if OUTPUT_MODE in ("terminal", "both", "json"):
        try:
            ORIGINAL_STDOUT.write(line + "\n")
            ORIGINAL_STDOUT.flush()
        except Exception:
            pass
 
    # File output: JSONL when json mode, plain text otherwise
    if OUTPUT_MODE == "json":
        if record is not None:
            _write_to_file(json.dumps(record, ensure_ascii=False))
    elif OUTPUT_MODE in ("file", "both"):
        plain = _ANSI_ESCAPE_RE.sub("", line)
        _write_to_file(plain)
 
 
# ============================================================
# PUBLIC DEBUG FUNCTION
# ============================================================
 
def _dbg(domain: str, func: str, msg: str, level: int | str | None = None, override: bool = False, _depth: int = 1):
    """
    Unified debug logger.
 
    Parameters
    ----------
    domain : str
        Debug domain (e.g., "CONTROLLER", "AHK", "MODE").
    func : str
        Logical context name provided by the caller.
    msg : str
        Message text.
    level : int | str | None
        Debug level (0-9), or ``"ai_trace"`` for AI-Trace shadow logging.
        Defaults to 2.
    override : bool
        If True, bypasses domain-level filtering.
 
    Behavior
    --------
    - Automatically captures filename, line number, and caller function.
    - Formats output according to INCLUDE_FILENAME and ENABLE_TIMESTAMPS.
    - Writes to terminal, file, or both.
    - When ``level="ai_trace"``: writes to file/JSONL only (no console output).
      Requires ``ai_trace_level: true`` in config; otherwise silently discarded.
    """
    if SILENT_MODE:
        return
 
    # AI-Trace: shadow logging — file/JSONL only, never console
    if level == "ai_trace":
        if not AI_TRACE_ENABLED:
            return
        _dbg_ai_trace(domain, func, msg, _depth=_depth + 1)
        return
 
    # OI-13 / FS-05: emit a one-shot RuntimeWarning when _dbg() is called
    # before init() so that misconfiguration (forgotten init() call) surfaces
    # immediately instead of silently producing no log-file output.
    global _init_warned
    if not _initialized and not _init_warned:
        _init_warned = True
        warnings.warn(
            "_dbg() called before init() — log file output and stream interception "
            "are not yet active. Call oc_diagnostics.init() at application startup.",
            RuntimeWarning,
            stacklevel=_depth + 1,
        )
 
    effective_level = level if level is not None else 2
 
    # OI-407: production-mode fail-safe. Suppress any level above PROD_MAX_LEVEL
    # regardless of per-domain filtering or override, and emit a one-shot warning
    # per (domain, level) key on first occurrence so operators notice the downgrade.
    if PROD_MODE and isinstance(effective_level, int) and effective_level > PROD_MAX_LEVEL:
        _key = f"{domain}:{effective_level}"
        if _key not in _prod_downgrade_warned:
            _prod_downgrade_warned.add(_key)
            _emit(
                f"[WARN] _dbg() L{effective_level} suppressed in production mode "
                f"(domain={domain}, max=L{PROD_MAX_LEVEL}). This warning is one-shot.",
                1,
            )
        return
 
    # Capture caller info via sys._getframe — O(1), no full-stack allocation
    frame = sys._getframe(_depth)
    filename = os.path.basename(frame.f_code.co_filename)
    lineno = frame.f_lineno
    caller_func = frame.f_code.co_name
 
    ts = _get_timestamp()
    color = _get_color(domain, effective_level)
    reset = _reset_color()
 
    # Build location string
    if INCLUDE_FILENAME:
        location = (
            f"{filename}:{lineno}::caller={caller_func}|logical context={func}"
        )
    else:
        location = f"logical context={func}"
 
    # FS-07: build JSON record when json output mode is active
    record: dict | None = None
    if OUTPUT_MODE == "json":
        record = {
            "ts": datetime.datetime.now().isoformat(timespec="milliseconds"),
            "domain": domain,
            "level": effective_level,
            "func": func,
            "file": filename,
            "line": lineno,
            "caller": caller_func,
            "msg": msg,
            "override": override,
        }
 
    if override:
        line = f"{color}[{domain}][OVERRIDE][{location}] {ts}{msg}{reset}"
        _emit(line, effective_level, record=record)
        return
 
    allowed = DEBUG_LEVELS.get(domain, 0)
    if effective_level > allowed:
        return
 
    line = f"{color}[{domain}][L{effective_level}][{location}] {ts}{msg}{reset}"
    _emit(line, effective_level, record=record)
 
 
def _dbg_ai_trace(domain: str, func: str, msg: str, _depth: int = 1):
    """Write an AI-Trace message to file/JSONL only (never console).
 
    Called internally by ``_dbg()`` when ``level="ai_trace"``.
    """
    frame = sys._getframe(_depth)
    filename = os.path.basename(frame.f_code.co_filename)
    lineno = frame.f_lineno
    caller_func = frame.f_code.co_name
 
    ts = _get_timestamp()
 
    if INCLUDE_FILENAME:
        location = f"{filename}:{lineno}::caller={caller_func}|logical context={func}"
    else:
        location = f"logical context={func}"
 
    if OUTPUT_MODE == "json":
        record = {
            "ts": datetime.datetime.now().isoformat(timespec="milliseconds"),
            "domain": domain,
            "level": "ai_trace",
            "func": func,
            "file": filename,
            "line": lineno,
            "caller": caller_func,
            "msg": msg,
            "override": False,
        }
        _write_to_file(json.dumps(record, ensure_ascii=False))
    elif OUTPUT_MODE in ("file", "both"):
        plain = f"[{domain}][AI_TRACE][{location}] {ts}{msg}"
        _write_to_file(plain)
 
 
# ============================================================
# ASYNC DEBUG FUNCTION (FS-16)
# ============================================================
 
async def _adbg(
    domain: str,
    func: str,
    msg: str,
    level: int | None = None,
    override: bool = False,
) -> None:
    """Async variant of _dbg().
 
    Terminal output is written synchronously (fast, low-latency).
    File I/O is offloaded to a thread via asyncio.to_thread() so that
    async callers are not blocked by disk writes.
 
    The caller frame is captured *synchronously* before any await because
    frame references become invalid once the coroutine is suspended.
    """
    if SILENT_MODE:
        return
 
    effective_level = level if level is not None else 2
 
    # Capture caller frame synchronously — must happen before any await
    frame = sys._getframe(1)
    filename = os.path.basename(frame.f_code.co_filename)
    lineno = frame.f_lineno
    caller_func = frame.f_code.co_name
 
    if not override:
        allowed = DEBUG_LEVELS.get(domain, 0)
        if effective_level > allowed:
            return
 
    ts = _get_timestamp()
    color = _get_color(domain, effective_level)
    reset = _reset_color()
 
    if INCLUDE_FILENAME:
        location = f"{filename}:{lineno}::caller={caller_func}|logical context={func}"
    else:
        location = f"logical context={func}"
 
    if override:
        line = f"{color}[{domain}][OVERRIDE][{location}] {ts}{msg}{reset}"
    else:
        line = f"{color}[{domain}][L{effective_level}][{location}] {ts}{msg}{reset}"
 
    # Terminal: synchronous (fast path)
    if OUTPUT_MODE in ("terminal", "both", "json"):
        try:
            ORIGINAL_STDOUT.write(line + "\n")
            ORIGINAL_STDOUT.flush()
        except Exception:
            pass
 
    # File I/O: offload to thread (non-blocking for async callers)
    if OUTPUT_MODE in ("file", "both"):
        plain = _ANSI_ESCAPE_RE.sub("", line)
        await asyncio.to_thread(_write_to_file, plain)
    elif OUTPUT_MODE == "json":
        record = {
            "ts": datetime.datetime.now().isoformat(timespec="milliseconds"),
            "domain": domain,
            "level": effective_level,
            "func": func,
            "file": filename,
            "line": lineno,
            "caller": caller_func,
            "msg": msg,
            "override": override,
        }
        await asyncio.to_thread(_write_to_file, json.dumps(record, ensure_ascii=False))
 
 
# ============================================================
# INSTRUMENTATION HELPERS
# ============================================================
 
def instrumentation_active() -> bool:
    if not INSTRUMENTATION_ENABLED:
        return False
    return DEBUG_LEVELS.get("INSTRUMENTATION", 0) >= 1
 
 
def instrumentation_verbose() -> bool:
    return DEBUG_LEVELS.get("INSTRUMENTATION", 0) >= 4
 
 
def instrumentation_allow_custom_components() -> bool:
    return DEBUG_LEVELS.get("INSTRUMENTATION", 0) >= 5
 
 
def instrumentation_allow_all() -> bool:
    """Return True when all components should be instrumented regardless of type. (FS-13)"""
    return INSTRUMENTATION_ALL
 
 
# ============================================================
# RUNTIME API (FS-03)
# ============================================================
 
def set_debug_level(domain: str, level: int) -> None:
    """Set the debug level for a domain at runtime (0–9).
 
    Updates ``DEBUG_LEVELS`` immediately; takes effect on the next ``_dbg()``
    call.  Emits a ``UserWarning`` and returns without modifying anything if
    ``level`` is not an integer in 0–9.
    """
    if not isinstance(level, int) or level < 0 or level > 9:
        warnings.warn(
            f"set_debug_level: invalid level {level!r} for domain {domain!r}. "
            "Must be an integer 0-9.",
            UserWarning,
            stacklevel=2,
        )
        return
    DEBUG_LEVELS[domain] = level
 
 
def set_global_setting(key: str, value) -> None:
    """Update a global diagnostic setting at runtime.
 
    Valid keys: ``silent_mode``, ``enable_timestamps``, ``enable_colors``,
    ``include_filename``, ``output_mode``.  Emits a ``UserWarning`` for
    unknown keys and returns without modifying anything.
    """
    global SILENT_MODE, ENABLE_TIMESTAMPS, ENABLE_COLORS, INCLUDE_FILENAME, OUTPUT_MODE
    _SETTABLE = {
        "silent_mode":        "SILENT_MODE",
        "enable_timestamps":  "ENABLE_TIMESTAMPS",
        "enable_colors":      "ENABLE_COLORS",
        "include_filename":   "INCLUDE_FILENAME",
        "output_mode":        "OUTPUT_MODE",
    }
    if key not in _SETTABLE:
        warnings.warn(
            f"set_global_setting: unknown key {key!r}. "
            f"Valid keys: {sorted(_SETTABLE)}",
            UserWarning,
            stacklevel=2,
        )
        return
    globals()[_SETTABLE[key]] = value
 
 
# ============================================================
# PROFILING / TIMING HELPERS (FS-12)
# ============================================================
 
@contextmanager
def dbg_timed(domain: str, func: str, label: str = "", level: int = 3):
    """Context manager that logs ENTER and EXIT with elapsed time in milliseconds.
 
    Example::
 
        with dbg_timed("CONTROLLER", "my_func", label="heavy_op"):
            do_work()
    """
    tag = f"[{label}] " if label else ""
    _dbg(domain, func, f"{tag}ENTER", level=level, _depth=2)
    t0 = time.perf_counter()
    try:
        yield
    finally:
        elapsed_ms = (time.perf_counter() - t0) * 1000
        _dbg(domain, func, f"{tag}EXIT ({elapsed_ms:.1f} ms)", level=level, _depth=2)
 
 
@contextmanager
def dbg_scope(domain: str, func: str, label: str = "", level: int = 2):
    """Context manager that logs ENTER and EXIT at the specified level without timing.
 
    Use for pure lifecycle tracing where elapsed time is not needed.
    For timed tracing, use :func:`dbg_timed` instead.
 
    Example::
 
        with dbg_scope("DATA", "load_records"):
            result = expensive_operation()
    """
    tag = f"[{label}] " if label else ""
    _dbg(domain, func, f"{tag}ENTER", level=level, _depth=2)
    try:
        yield
    finally:
        _dbg(domain, func, f"{tag}EXIT", level=level, _depth=2)
 
 
def dbg_profile(domain: str = "GENERAL", func: str = "", label: str = "", level: int = 3):
    """Decorator that wraps a function with :func:`dbg_timed`.
 
    Example::
 
        @dbg_profile("CONTROLLER", label="fetch_data")
        def fetch_data():
            ...
    """
    def decorator(fn):
        import functools
        tag = label or fn.__name__
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            with dbg_timed(domain, func or fn.__name__, label=tag, level=level):
                return fn(*args, **kwargs)
        return wrapper
    return decorator
 
 
# ============================================================
# RUNTIME CONTRACT ASSERTIONS
# ============================================================
 
def _dbg_assert(
    domain: str,
    condition,
    message: str,
    func: str | None = None,
) -> None:
    """Domain-aware runtime assertion, suppressed in production mode.
 
    Parameters
    ----------
    domain : str
        Debug domain for the assertion message.
    condition
        Expression to evaluate.  If falsy (and not in production mode),
        emits an L1 override ``_dbg`` message and raises ``AssertionError``.
    message : str
        Descriptive failure message.
    func : str | None
        Logical context name.  If ``None``, uses ``"<unknown>"``.
 
    Notes
    -----
    In ``PROD_MODE`` this function returns immediately on entry; no
    assertion check is performed, no ``_dbg`` is emitted, and no
    ``AssertionError`` is raised.  Argument expressions (including
    ``condition`` and ``message``) are still evaluated by Python before
    the call enters this function — do not pass expensive computations
    as arguments.  **Never** use for safety-critical invariants — see §6a
    and AP-16 in the specification.
    """
    if PROD_MODE:
        return
    if not condition:
        ctx = func or "<unknown>"
        _dbg(domain, ctx, f"ASSERT FAILED: {message}", 1, override=True, _depth=2)
        raise AssertionError(message)
 
 
# ============================================================
# dbg_trace DECORATOR
# ============================================================
 
def dbg_trace(domain: str, level: int = 2):
    """Decorator that emits START/END lifecycle messages and catches exceptions.
 
    Injects a ``func`` local variable set to the decorated function's
    ``__name__`` so that inner ``_dbg`` calls can reference it without
    hardcoding a string.
 
    Parameters
    ----------
    domain : str
        Debug domain for lifecycle messages.
    level : int
        Debug level for START/END messages (default 2).
 
    Example
    -------
    ::
 
        @dbg_trace("DATA")
        def process_data(data):
            _dbg("DATA", func, "working...", 3)
            return transform(data)
    """
    def decorator(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            func = fn.__name__
            _dbg(domain, func, "START", level=level, _depth=2)
            try:
                result = fn(*args, **kwargs)
            except Exception as exc:
                _dbg(domain, func, f"EXCEPTION: {type(exc).__name__}: {exc}", level=1, override=True, _depth=2)
                raise
            _dbg(domain, func, "END", level=level, _depth=2)
            return result
        return wrapper
    return decorator
 
 
# ============================================================
# dbg_metrics DECORATOR
# ============================================================
 
def dbg_metrics(domain: str, level: int = 3):
    """Decorator that logs structured performance metrics (time + memory).
 
    Emits a JSON-style string with ``duration_ms`` and ``memory_mb`` fields
    suitable for observability tooling.
 
    Parameters
    ----------
    domain : str
        Debug domain for the metrics message.
    level : int
        Debug level (default 3).
    """
    def decorator(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            func = fn.__name__
            # Memory tracking via tracemalloc if available; falls back gracefully
            mem_before = None
            try:
                import tracemalloc as _tm
                if not _tm.is_tracing():
                    _tm.start()
                mem_before = _tm.get_traced_memory()[0]
            except Exception:
                pass
 
            t0 = time.perf_counter()
            try:
                result = fn(*args, **kwargs)
            finally:
                elapsed_ms = (time.perf_counter() - t0) * 1000
                mem_mb = 0.0
                if mem_before is not None:
                    try:
                        import tracemalloc as _tm
                        mem_after = _tm.get_traced_memory()[0]
                        mem_mb = (mem_after - mem_before) / (1024 * 1024)
                    except Exception:
                        pass
                metrics = json.dumps({
                    "duration_ms": round(elapsed_ms, 2),
                    "memory_mb": round(mem_mb, 2),
                })
                _dbg(domain, func, f"METRICS: {metrics}", level=level, _depth=2)
            return result
        return wrapper
    return decorator
 
 
# ============================================================
# EXCEPTION HOOK
# ============================================================
 
# Saved at init() time so _exception_hook can delegate to it.
_original_excepthook = None
 
 
def _exception_hook(exc_type, exc_value, exc_traceback):
    tb_lines = traceback.format_exception(exc_type, exc_value, exc_traceback)
    tb_text = "".join(tb_lines).rstrip()
 
    ts = _get_timestamp()
    color = _get_color("PYERR", 1)
    reset = _reset_color()
 
    header_line = f"{color}[PYERR][L1][UNCAUGHT] {ts}Uncaught exception:{reset}"
    _emit(header_line, 1)
 
    for line in tb_text.splitlines():
        formatted = f"{color}[PYERR][L1][TRACE] {ts}{line}{reset}"
        _emit(formatted, 1)
 
    # Delegate to the hook that was active before init() installed ours so
    # that debuggers, pytest, rich.traceback, and ASGI frameworks are not
    # silently bypassed.
    #
    # OI-24 (intentional behaviour): if sys.stderr is currently intercepted by
    # _StreamInterceptor, the original hook's write to stderr will appear a
    # second time in the log — once from the _emit calls above, and once routed
    # through the interceptor.  The re-entrant guard prevents infinite recursion
    # but not this cosmetic duplication.  This is documented and accepted.
    if _original_excepthook is not None and _original_excepthook is not _exception_hook:
        try:
            _original_excepthook(exc_type, exc_value, exc_traceback)
        except Exception:
            pass
 
 
# ============================================================
# STDOUT / STDERR REDIRECTION
# ============================================================
 
# Thread-local flag that prevents re-entrant calls from spiralling into
# infinite recursion.  If _StreamInterceptor.write() is called while it is
# already running on the same thread (e.g. because _dbg's own formatting
# code triggers a print(), or because an exception handler tries to report
# an error via sys.stdout after a BaseException propagates out of _dbg),
# the second call bypasses processing and writes directly to the original
# stream instead of looping back through _dbg again.
_interceptor_active = threading.local()
 
 
class _StreamInterceptor:
    """
    Wraps sys.stdout or sys.stderr, routing all output through _dbg().
 
    Implements the full io.TextIOBase-compatible interface so that
    third-party libraries (tqdm, click, rich, logging, etc.) that
    call isatty(), fileno(), or access .encoding on sys.stdout do not
    receive AttributeError after oc_diagnostics.init() is called.
    """
 
    def __init__(self, stream_name: str):
        self.stream_name = stream_name
        self._original = ORIGINAL_STDOUT if stream_name == "stdout" else ORIGINAL_STDERR
 
    def write(self, text: str):
        # Re-entrant-call guard: if this thread is already inside write()
        # (e.g. a BaseException propagated out of _dbg and an exception
        # handler is printing via sys.stdout), fall back to the original
        # stream to avoid infinite recursion.
        if getattr(_interceptor_active, "value", False):
            try:
                self._original.write(text)
                self._original.flush()
            except Exception:
                pass
            return
 
        if not text:
            return
 
        _interceptor_active.value = True
        try:
            domain = "STDOUT" if self.stream_name == "stdout" else "STDERR"
            for line in text.splitlines():
                stripped = line.rstrip()
                if not stripped:
                    # Pass blank lines through to the original stream so that
                    # formatted output (tables, separators, blank print() calls)
                    # is not silently discarded.
                    try:
                        self._original.write("\n")
                        self._original.flush()
                    except Exception:
                        pass
                    continue
 
                # _depth=2 skips this write() frame so _dbg reports the code
                # that called print() as the call site, not unified_debug.py.
                _dbg(domain, "stream", stripped, level=2, override=True, _depth=2)
        finally:
            _interceptor_active.value = False
 
    def flush(self):
        try:
            self._original.flush()
        except Exception:
            pass
 
    def isatty(self) -> bool:
        try:
            return self._original.isatty()
        except Exception:
            return False
 
    def fileno(self) -> int:
        return self._original.fileno()
 
    @property
    def encoding(self) -> str:
        return getattr(self._original, "encoding", "utf-8")
 
    @property
    def errors(self) -> str:
        return getattr(self._original, "errors", "strict")
 
    def readable(self) -> bool:
        return False
 
    def writable(self) -> bool:
        return True
 
    def seekable(self) -> bool:
        return False
 
    # OI-16: stubs for advanced TextIOBase methods accessed by some libraries
    # (colorama on Windows, some logging handlers, etc.).
 
    @property
    def write_through(self) -> bool:
        """Always True — _StreamInterceptor does not buffer writes."""
        return True
 
    @property
    def buffer(self):
        """Not available — _StreamInterceptor wraps a text stream, not bytes."""
        raise io.UnsupportedOperation("buffer")
 
    def detach(self):
        """Not supported — raises UnsupportedOperation."""
        raise io.UnsupportedOperation("detach")
 
    def reconfigure(self, **_kwargs):
        """Not supported — raises UnsupportedOperation."""
        raise io.UnsupportedOperation("reconfigure")
 
 
# ============================================================
# HOT-RELOAD (FS-09)
# ============================================================
 
def _reload_config() -> None:
    """Reload debug_config.json in-place, updating all module-level globals.
 
    Preserves the existing ``DEBUG_LEVELS`` dict *object* so that downstream
    modules holding a reference to the dict automatically see the updated levels.
 
    Uses an update-first, prune-stale-after pattern (OI-29) to avoid the brief
    empty-dict window that ``clear()`` + ``update()`` would create.
    """
    global COLOR_CODES, CONFIG_SOURCE, INSTRUMENTATION_ENABLED, SILENT_MODE
    global ADDITIONAL_STRUCTURAL_PREFIXES, ADDITIONAL_STRUCTURAL_EXACT
    global ENABLE_TIMESTAMPS, ENABLE_COLORS, INCLUDE_FILENAME, OUTPUT_MODE
    global MAX_LOG_FILES, MAX_LOG_SIZE_BYTES, HTTP_REDACT_HEADERS
    global INSTRUMENTATION_ALL, UI_EVENT_PROPERTIES, _CONFIG_MTIME
    global AI_TRACE_ENABLED, REDACTION_PATTERNS, REDACT_SECRETS
 
    levels, colors, settings, source = _load_debug_config()
    # OI-29: update first so the dict is never empty, then remove stale keys.
    DEBUG_LEVELS.update(levels)
    for k in list(DEBUG_LEVELS):
        if k not in levels:
            del DEBUG_LEVELS[k]
    COLOR_CODES = colors
    CONFIG_SOURCE = source
    INSTRUMENTATION_ENABLED = settings["instrumentation_enabled"]
    SILENT_MODE = settings["silent_mode"]
    ADDITIONAL_STRUCTURAL_PREFIXES = tuple(settings.get("structural_id_prefixes", []))
    ADDITIONAL_STRUCTURAL_EXACT = frozenset(settings.get("structural_id_exact", []))
    ENABLE_TIMESTAMPS = settings["enable_timestamps"]
    ENABLE_COLORS = settings["enable_colors"]
    INCLUDE_FILENAME = settings["include_filename"]
    OUTPUT_MODE = settings["output_mode"]
    MAX_LOG_FILES = settings["max_log_files"]
    MAX_LOG_SIZE_BYTES = (
        int(settings["max_log_size_mb"] * 1024 * 1024)
        if settings["max_log_size_mb"] > 0 else 0
    )
    HTTP_REDACT_HEADERS = frozenset(
        h.lower() for h in settings.get("http_redact_headers", [])
    )
    INSTRUMENTATION_ALL = bool(settings.get("instrumentation_allow_all", False))
    UI_EVENT_PROPERTIES = tuple(
        settings.get("ui_event_properties") or _DEFAULT_UI_EVENT_PROPERTIES
    )
    AI_TRACE_ENABLED = bool(settings.get("ai_trace_level", False))
    new_patterns = []
    for pat in settings.get("redaction_patterns", []):
        try:
            new_patterns.append(re.compile(rf'(?i)({re.escape(pat)})\s*[=:]\s*\S+'))
        except re.error:
            pass
    REDACTION_PATTERNS = new_patterns
    REDACT_SECRETS = bool(settings.get("redact_secrets", True))
 
 
def _watch_config_loop() -> None:
    """Daemon thread: poll the config file mtime every 2 s; reload on change. (FS-09)"""
    global _CONFIG_MTIME
    while True:
        time.sleep(2)
        if not _CONFIG_PATH:
            continue
        try:
            mtime = os.path.getmtime(_CONFIG_PATH)
            if mtime != _CONFIG_MTIME:
                _CONFIG_MTIME = mtime
                _reload_config()
        except Exception:
            pass
 
 
# ============================================================
# PUBLIC INITIALIZATION
# ============================================================
 
_initialized = False
_init_warned = False       # OI-13/FS-05: one-shot flag for the pre-init RuntimeWarning
_io_fallback_warned = False  # FS-25: one-shot flag for the disk-I/O fallback warning
 
 
def init(
    intercept_streams: bool = True,
    intercept_exceptions: bool = True,
    watch_config: bool = False,
) -> None:
    """
    Initialize oc_diagnostics side effects.
 
    Must be called explicitly after importing oc_diagnostics.
    Importing the package does NOT activate stream interception or
    the custom exception hook automatically.
 
    Parameters
    ----------
    intercept_streams : bool
        If True, replace sys.stdout and sys.stderr with _StreamInterceptor
        instances that route all print() calls through _dbg().
        Can also be disabled by setting the environment variable
        UNIFIED_DEBUG_DISABLE_STDIO_INTERCEPT=1 before calling init().
        Forced to False in production mode (PROD_MODE=True).  Default True.
    intercept_exceptions : bool
        If True, install a custom sys.excepthook that logs uncaught
        exceptions through _dbg() and then delegates to the previously
        active hook. Can also be disabled by setting the environment
        variable UNIFIED_DEBUG_DISABLE_EXCEPTION_HOOK=1 before calling
        init(). Forced to False in production mode (PROD_MODE=True).
        Default True.
    watch_config : bool
        If True, start a daemon thread that polls the loaded config file
        every 2 seconds and reloads it when the mtime changes.  Useful
        during development to adjust log levels without restarting the
        application.  Default False. (FS-09)
 
    Notes
    -----
    Calling init() more than once is a no-op.
    """
    global _initialized, _original_excepthook
 
    # FS-15: production mode disables all side effects that could interfere
    # with prod environments (stream interception, exception hook).
    if PROD_MODE:
        intercept_streams = False
        intercept_exceptions = False
 
    if _initialized:
        return
 
    disable_hook = os.environ.get("UNIFIED_DEBUG_DISABLE_EXCEPTION_HOOK") == "1"
    if intercept_exceptions and not disable_hook:
        _original_excepthook = sys.excepthook
        sys.excepthook = _exception_hook
 
    disable_intercept = os.environ.get("UNIFIED_DEBUG_DISABLE_STDIO_INTERCEPT") == "1"
    if intercept_streams and not disable_intercept:
        sys.stdout = _StreamInterceptor("stdout")
        sys.stderr = _StreamInterceptor("stderr")
 
    # V2.0: honour watch_config from config file as well as the parameter
    effective_watch = watch_config or _settings.get("watch_config", False)
 
    # FS-09: start hot-reload watcher thread if requested
    if effective_watch and _CONFIG_PATH:
        t = threading.Thread(target=_watch_config_loop, daemon=True)
        t.start()
 
    _initialized = True