oct_diag.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/diag/oct_diag.py
 
"""
Purpose
-------
Provide oc_diagnostics configuration management tools for the OCT CLI.
 
Responsibilities
----------------
- Load, validate, and modify ``oc_diagnostics/debug_config.json``.
- List configured domains with their levels and colors.
- Update domain debug levels in-place.
- Report validation warnings without importing oc_diagnostics at runtime
  (preserving the dev-time/runtime architectural boundary).
 
Diagnostics
-----------
Domain: OCT-DIAG
Levels:
    L2 - lifecycle
    L3 - semantic details
    L4 - deep tracing
 
Contracts
---------
- Must not import ``oc_diagnostics`` - operates purely on the JSON config file.
- Must preserve JSON formatting and key order when writing back.
- Level values must be integers 0-9, in either form oc_diagnostics accepts:
  compact (``"GIT": 3``) or rich (``"GIT": {"level": 3, "color": "cyan"}``).
  Until 2026-09-28 this module allowed only 0-4 and the rich form, so a
  valid config such as ``{"*": 9}`` (open every domain for a trace) failed
  ``oct diag validate-config`` and crashed ``list-domains``.
 
Dependencies
------------
- oct.core.option_c_dir (debug_config.json path resolution)
- oct.core.diagnostics (_dbg structured logger)
"""
 
import json
from pathlib import Path
 
from oct.core.option_c_dir import resolve_debug_config
from oct.core.diagnostics import _dbg
 
 
#: oc_diagnostics' range (``_DOMAIN_SCHEMA`` and ``set_debug_level``: 0-9).
VALID_LEVELS = set(range(10))
 
#: The file formats ``global_settings.log_format`` accepts (oc_diagnostics 2.2.0).
VALID_LOG_FORMATS = ("text", "jsonl")
 
# OI-508 / FS-515: schema version written by migrations.
CURRENT_SCHEMA_VERSION = "2.0"
 
EXPECTED_GLOBAL_KEYS = {
    "silent_mode", "enable_timestamps", "enable_colors", "include_filename",
    "output_mode", "instrumentation_enabled", "max_log_files", "max_log_size_mb",
    "instrumentation_allow_all", "mode", "http_redact_headers",
    "ui_event_properties", "structural_id_prefixes", "structural_id_exact",
}
 
 
def _find_config(project_root: Path) -> Path:
    """Locate debug_config.json (FS-539: `.option_c/`-first with fallbacks).
 
    Delegates to :func:`oct.core.option_c_dir.resolve_debug_config`, which
    tries ``.option_c/debug_config.json`` first, then
    ``oc_diagnostics/debug_config.json`` (v0.17+), then
    ``diagnostics/debug_config.json`` (pre-v0.17).
    """
    return resolve_debug_config(project_root)
 
 
def load_config(project_root: Path) -> dict:
    """Load and parse debug_config.json. Raises on missing/invalid file."""
    func = "load_config"
    _dbg("OCT-DIAG", func, "loading the debug config", 3)
    config_path = _find_config(project_root)
    if not config_path.is_file():
        raise FileNotFoundError(f"Config not found: {config_path}")
    text = config_path.read_text(encoding="utf-8")
    return json.loads(text)
 
 
def _get_debug_domain_map(config: dict) -> tuple[dict, bool]:
    """OI-508 / FS-515: return ``(domain_map, is_legacy)`` from a config dict.
 
    Accepts both the legacy v1 ``domains:`` layout and the current v2
    ``DEBUG_LEVELS:`` layout. The returned map is always in the v2 shape
    (``{name: {"level": int, "color": str}}``) regardless of what the file
    contains - call sites can read the result uniformly.
 
    ``is_legacy`` is True whenever the file used the ``domains:`` key
    **and** the ``DEBUG_LEVELS:`` key was absent - that is the trigger
    for the migration hint.
    """
    if "DEBUG_LEVELS" in config and isinstance(config["DEBUG_LEVELS"], dict):
        # The compact form ("GIT": 3) is valid v2 too; normalise it so every
        # reader sees {"level": ..., "color": ...}.
        return {
            name: dict(entry) if isinstance(entry, dict)
            else {"level": entry, "color": "default"}
            for name, entry in config["DEBUG_LEVELS"].items()
        }, False
    legacy = config.get("domains")
    if not isinstance(legacy, dict):
        return {}, False
    normalised: dict = {}
    for name, entry in legacy.items():
        if isinstance(entry, dict):
            normalised[name] = dict(entry)
        elif isinstance(entry, int):
            # Short form: domain → level. Fabricate default color.
            normalised[name] = {"level": entry, "color": "default"}
        else:
            normalised[name] = {"level": entry, "color": "default"}
    return normalised, True
 
 
def migrate_config(config_path: Path, *, dry_run: bool = False) -> dict:
    """OI-508 / FS-515: rewrite ``config_path`` in-place as a v2 config.
 
    Behaviour:
    * Idempotent - if the file already declares ``_schema_version == "2.0"``
      and carries ``DEBUG_LEVELS``, nothing is written.
    * Writes ``<config_path>.bk`` before overwriting the original, so a
      botched migration is recoverable.
    * When ``dry_run`` is True, the target file is never touched and no
      backup is produced; the returned dict is the post-migration shape.
 
    Returns the (would-be) post-migration config dict so callers can
    inspect / pretty-print the result.
    """
    func = "migrate_config"
    _dbg("OCT-DIAG", func, "migrating the debug config", 3)
    if not config_path.is_file():
        raise FileNotFoundError(f"Config not found: {config_path}")
    original_text = config_path.read_text(encoding="utf-8")
    config = json.loads(original_text)
    if not isinstance(config, dict):
        raise ValueError("Config root must be a JSON object")
 
    domain_map, is_legacy = _get_debug_domain_map(config)
    already_v2 = (
        config.get("_schema_version") == CURRENT_SCHEMA_VERSION
        and "DEBUG_LEVELS" in config
        and not is_legacy
    )
    if already_v2:
        return config
 
    # Build the v2 shape: drop ``domains`` if present; insert ordered keys.
    new_config: dict = {"_schema_version": CURRENT_SCHEMA_VERSION, "DEBUG_LEVELS": domain_map}
    for key, value in config.items():
        if key in ("_schema_version", "DEBUG_LEVELS", "domains"):
            continue
        new_config[key] = value
 
    if dry_run:
        return new_config
 
    backup_path = config_path.with_suffix(config_path.suffix + ".bk")
    backup_path.write_text(original_text, encoding="utf-8")
    config_path.write_text(json.dumps(new_config, indent=2) + "\n", encoding="utf-8")
    return new_config
 
 
def validate_config(project_root: Path) -> list[str]:
    """Validate debug_config.json schema. Returns list of warnings/errors."""
    func = "validate_config"
    _dbg("OCT-DIAG", func, "validating the debug config", 3)
    issues: list[str] = []
    config_path = _find_config(project_root)
 
    if not config_path.is_file():
        issues.append(f"ERROR: Config file not found: {config_path}")
        return issues
 
    try:
        text = config_path.read_text(encoding="utf-8")
    except OSError as e:
        issues.append(f"ERROR: Cannot read {config_path}: {e}")
        return issues
 
    try:
        config = json.loads(text)
    except json.JSONDecodeError as e:
        issues.append(f"ERROR: Invalid JSON in {config_path}: {e}")
        return issues
 
    if not isinstance(config, dict):
        issues.append("ERROR: Config root must be a JSON object")
        return issues
 
    # Schema version
    version = config.get("_schema_version")
    if version is None:
        issues.append("WARNING: Missing '_schema_version' key")
 
    # OI-508 / FS-515: accept both schemas. Legacy ``domains:`` is a
    # WARNING with a clear migration hint rather than a hard ERROR so a
    # valid v1 file still passes non-strict validation.
    domain_map, is_legacy = _get_debug_domain_map(config)
    if is_legacy:
        issues.append(
            "WARNING: 'domains' key is legacy v1 schema; run 'oct diag migrate-config' "
            "to upgrade to v2 ('DEBUG_LEVELS')."
        )
    elif not domain_map and "DEBUG_LEVELS" not in config and "domains" not in config:
        issues.append("WARNING: Missing 'DEBUG_LEVELS' key")
    elif "DEBUG_LEVELS" in config and not isinstance(config.get("DEBUG_LEVELS"), dict):
        issues.append("ERROR: 'DEBUG_LEVELS' must be a JSON object")
    for name, entry in domain_map.items():
        if not isinstance(entry, dict):
            issues.append(f"ERROR: Domain '{name}' must be a JSON object")
            continue
        level = entry.get("level")
        if level is None:
            issues.append(f"WARNING: Domain '{name}' missing 'level' key")
        elif not isinstance(level, int) or isinstance(level, bool) or level not in VALID_LEVELS:
            issues.append(f"ERROR: Domain '{name}' level must be 0-9, got: {level}")
 
    # Global settings
    settings = config.get("global_settings")
    if settings is None:
        issues.append("WARNING: Missing 'global_settings' key")
    elif not isinstance(settings, dict):
        issues.append("ERROR: 'global_settings' must be a JSON object")
    else:
        output_mode = settings.get("output_mode")
        if output_mode is not None and output_mode not in ("terminal", "file", "both", "json"):
            issues.append(f"WARNING: 'output_mode' should be terminal|file|both|json, got: {output_mode!r}")
        log_format = settings.get("log_format")
        if log_format is not None and log_format not in VALID_LOG_FORMATS:
            issues.append(f"WARNING: 'log_format' should be text|jsonl, got: {log_format!r}")
        mode = settings.get("mode")
        if mode is not None and mode not in ("dev", "prod"):
            issues.append(f"WARNING: 'mode' should be dev|prod, got: {mode!r}")
 
    return issues
 
 
def list_domains(project_root: Path) -> list[dict]:
    """Return list of domain info dicts: [{name, level, color}, ...].
 
    OI-508 / FS-515: reads via :func:`_get_debug_domain_map` so both v1
    and v2 configs surface identical output.
    """
    config = load_config(project_root)
    domain_map, _ = _get_debug_domain_map(config)
    result = []
    for name, entry in domain_map.items():
        result.append({
            "name": name,
            "level": entry.get("level", "?"),
            "color": entry.get("color", "default"),
        })
    return result
 
 
def set_level(
    project_root: Path,
    domain: str,
    level: int,
    *,
    dry_run: bool = False,
) -> dict:
    """Update a domain's debug level in debug_config.json.
 
    OI-508 / FS-515: writes to whichever key the file actually uses
    (``DEBUG_LEVELS`` on v2, ``domains`` on legacy v1) so an operator
    can tweak a level without a full migration first.
 
    OI-528: safe-edit semantics. A ``<config_path>.bk`` backup is written
    before overwriting the original so a bad edit is recoverable via
    :func:`restore_config`. When ``dry_run`` is True, neither the target
    nor the backup is touched; the returned dict is the *would-be*
    post-edit config shape so callers can preview the result.
    """
    func = "set_level"
    _dbg("OCT-DIAG", func, f"setting level for domain {domain}", 3)
    if level not in VALID_LEVELS:
        raise ValueError(f"Level must be 0-9, got: {level}")
 
    config_path = _find_config(project_root)
    if not config_path.is_file():
        raise FileNotFoundError(f"Config not found: {config_path}")
 
    original_text = config_path.read_text(encoding="utf-8")
    config = json.loads(original_text)
    if "DEBUG_LEVELS" in config and isinstance(config["DEBUG_LEVELS"], dict):
        target_key = "DEBUG_LEVELS"
    elif "domains" in config and isinstance(config["domains"], dict):
        target_key = "domains"
    else:
        raise KeyError("Config has neither 'DEBUG_LEVELS' nor 'domains' keys")
 
    domains = config[target_key]
    if domain not in domains:
        raise KeyError(f"Domain '{domain}' not found in config. Available: {', '.join(domains)}")
 
    entry = domains[domain]
    if isinstance(entry, dict):
        entry["level"] = level
    else:
        domains[domain] = {"level": level, "color": "default"}
 
    if dry_run:
        return config
 
    backup_path = config_path.with_suffix(config_path.suffix + ".bk")
    backup_path.write_text(original_text, encoding="utf-8")
    config_path.write_text(json.dumps(config, indent=2) + "\n", encoding="utf-8")
    return config
 
 
def restore_config(project_root: Path) -> Path:
    """OI-528: restore ``debug_config.json`` from its ``.bk`` backup.
 
    Returns the path that was restored. Raises :class:`FileNotFoundError`
    when no backup exists so the CLI can surface a clean error instead of
    silently no-op-ing.
    """
    config_path = _find_config(project_root)
    backup_path = config_path.with_suffix(config_path.suffix + ".bk")
    if not backup_path.is_file():
        raise FileNotFoundError(f"No backup found at {backup_path}")
    config_path.write_text(backup_path.read_text(encoding="utf-8"), encoding="utf-8")
    return config_path