oct_health.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/health/oct_health.py
 
"""
Option C project health dashboard implementation.
 
Purpose
-------
Provide a single-command compliance dashboard that aggregates lint results,
fixable violations, documentation status, test status, and oc_diagnostics
compatibility into one report.
 
Responsibilities
----------------
- Run linter checks on all Python files and aggregate statistics.
- Run formatter checks in dry-run mode to count fixable violations.
- Validate presence of required documentation files.
- Run project tests and capture pass/fail status.
- Check oc_diagnostics version compatibility.
- Output results as terminal report or JSON.
 
Diagnostics
-----------
Domain: OCT.HEALTH
Levels:
    L2 - health check lifecycle and command routing
    L3 - individual check categories (lint, docs, tests, compat)
    L4 - per-file analysis and detailed aggregation
 
Contracts
---------
- Health checks are read-only; no files are modified.
- All checks are independent; failure of one does not prevent others.
- JSON output is always valid JSON with a stable schema.
- Terminal output uses ASCII-safe characters for Windows compatibility.
 
Dependencies
------------
- oct (package version)
- oct.core.progress (terminal dashboard renderer)
- oct.core.parsing (safe AST parsing helper)
- oct.core.paths (project-relative path helper)
- oct.linter.oct_lint (reused validation checks and profile defaults)
- oct.core.exclusions (directory and wildcard exclusion lists)
- oct.core.project_root (project root detection)
- oct.core.compat (oc_diagnostics version compatibility check)
- oct.core.octrc (.octrc.json loader)
- oct.core.terminal (ANSI colour helpers)
- oct.core.waivers (inline waiver semantics, shared with the linter and
  the quality gate)
- oct.core.option_c_dir (lint cache, debug_config.json and octrc path resolution)
- oct.linter.lint_cache (LintCache, content hashes and current_linter_identity - reuse results for unchanged files, in the store oct lint shares)
- oct.timing.ledger (timed_stage - per-check timing records)
- oct.core.diagnostics (_dbg structured logger)
- oct.core.git (lazy; repository, branch, worktree and last-commit state for the git checks)
- oct.core.log_writer (lazy; write_command_log for the health log file)
- oct.formatter.oct_format (lazy; fixers and FormatterContext to count fixable violations)
- oct.git.conventional (lazy; Conventional Commits check of the last commit)
- oct.git.policy (lazy; branch naming validation)
- oct.timing.report (lazy; timing report, baseline and session views)
- click (lazy; ClickException for timing report errors)
- pytest (not imported; found by name and run as python -m pytest for the test check)
- pytest_cov (not imported; pytest plugin found by name, for the --cov coverage measurement)
- xdist (not imported; pytest plugin behind -n auto for the parallel test run)
"""
 
import json
import re
import subprocess
import sys
from contextlib import nullcontext
from dataclasses import dataclass, field, asdict
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Any, Optional
 
from oct import __version__
from oct.core.progress import Dashboard
from oct.core.parsing import safe_parse
from oct.core.paths import to_project_relative
from oct.linter.oct_lint import (
    validate_header_block,
    check_docstring,
    check_dependencies_section,
    check_diagnostics_profile,
    check_dbg_usage,
    check_dbg_import,
    check_func_pattern,
    check_syntax_warnings,
    find_python_files,
    LinterContext,
    _DEFAULT_RULES,
    _PROFILES,
    _describe_syntax_error,
    effective_rules,
    parse_path_rules,
)
from oct.core.exclusions import LINTER_EXCLUDE_DIRS, WILDCARD_EXCLUDE_DIRS
from oct.core.project_root import find_project_root
from oct.core.option_c_dir import option_c_dir, resolve_debug_config
from oct.core.compat import check_oc_diagnostics_compat
from oct.core.octrc import load_octrc_safe
from oct.core.terminal import CYAN, GREEN, RED, YELLOW, RESET
from oct.core.waivers import inline_disabled_rules
from oct.linter.lint_cache import (
    LintCache,
    LintCacheEntry,
    content_hash,
    current_linter_identity,
    rules_hash,
)
from oct.timing.ledger import timed_stage
from oct.core.diagnostics import _dbg
 
 
# ============================================================================
# Health Check Functions
# ============================================================================
 
 
# F21 / Plan 175 (item 1): health checks whose verdict is derived from the
# parsed AST. On an unparseable file (``tree is None``) they cannot run, so
# they must be marked NOT-RUN rather than credited a vacuous pass - the exact
# fail-open shape Plan 170 closed in the linter. ``syntax_error`` itself is
# the gate and is evaluated directly from ``tree is None``, so it is not in
# this set.
_AST_DEPENDENT_HEALTH_CHECKS = frozenset(
    {"dbg_usage", "dbg_import", "func_pattern", "syntax_warnings"}
)
 
# Plan 277: checks no inline waiver may excuse, however it is written.
# ``syntax_error`` is F21 / Plan 170's unconditional gate - the linter
# evaluates it BEFORE its own ``_rule_active`` gate and says so in a
# comment, so unparseable Python is never compliant at any profile and
# under any waiver. The dashboard honours waivers on exactly the rules
# the linter honours them on, and this is the one it does not.
_UNWAIVABLE_HEALTH_CHECKS = frozenset({"syntax_error"})
 
 
def check_lint_health(
    project_root: Path,
    ctx: LinterContext,
    verbose: bool = False,
    exclude_dirs: Optional[set] = None,
    wildcard_exclude: Optional[list] = None,
    active_rules: Optional[dict] = None,
    analysis_cache: Optional[Dict[Path, Dict[str, Any]]] = None,
) -> Dict[str, Any]:
    """
    Run linter checks on all Python files and aggregate pass/fail statistics.
 
    Returns a dict with total_files, fully_compliant count, per-check pass
    counts, and compliance percentage.
    """
    func = "check_lint_health"
 
    if exclude_dirs is None:
        exclude_dirs = set(LINTER_EXCLUDE_DIRS)
    if wildcard_exclude is None:
        wildcard_exclude = list(WILDCARD_EXCLUDE_DIRS)
    if active_rules is None:
        active_rules = dict(_DEFAULT_RULES)
 
    python_files = list(find_python_files(project_root, exclude_dirs, wildcard_exclude))
 
    # Build counters only for active rules (excluding tests_exist which health doesn't check).
    # F21 / Plan 175: "syntax_error" leads the list (mirroring the linter, Plan 170) so an
    # unparseable file fails a real check instead of being reported healthy.
    health_checks = [
        "syntax_error",
        "header", "docstring", "dependencies_section", "diagnostics_profile",
        "dbg_usage", "dbg_import", "func_pattern", "syntax_warnings",
    ]
    active_checks = [c for c in health_checks if active_rules.get(c, True)]
    # The same identity ``oct lint`` writes: the store is shared, and a
    # different header discards every entry on load (2026-09-28).
    cache = LintCache(
        option_c_dir(project_root) / "cache" / "lint-cache.json",
        __version__,
        linter_identity=current_linter_identity(),
    )
 
    total = len(python_files)
    counters = {c: 0 for c in active_checks}
    # NOT-RUN tally: AST-dependent checks that were skipped because the file
    # did not parse. Kept separate so their pass percentage is computed over
    # the files where the check actually ran, not inflated by vacuous passes.
    not_run = {c: 0 for c in active_checks}
    # Plan 268: APPLICABLE tally - the files a check was active on at all.
    # A check a path rule switched off for one file is neither a pass nor a
    # failure there, so it must not drag that file into the denominator. A
    # project with no path rules has every check applying to every file, so
    # this is ``total`` and the reported percentages are unchanged.
    applicable = {c: 0 for c in active_checks}
    fully_compliant = 0
    file_details = []
    cache_hits = 0
    cache_misses = 0
 
    check_funcs = {
        "header": lambda text, lines, py_file, tree, sw: validate_header_block(
            lines[:4] if len(lines) >= 4 else lines, py_file, ctx),
        "docstring": lambda text, lines, py_file, tree, sw: check_docstring(text, tree=tree),
        "dependencies_section": lambda text, lines, py_file, tree, sw: check_dependencies_section(text, tree=tree),
        "diagnostics_profile": lambda text, lines, py_file, tree, sw: check_diagnostics_profile(text, tree=tree),
        "dbg_usage": lambda text, lines, py_file, tree, sw: check_dbg_usage(text, tree=tree),
        "dbg_import": lambda text, lines, py_file, tree, sw: check_dbg_import(text, tree=tree),
        "func_pattern": lambda text, lines, py_file, tree, sw: check_func_pattern(text, tree=tree),
        "syntax_warnings": lambda text, lines, py_file, tree, sw: check_syntax_warnings(text, warnings_list=sw),
    }
 
    for py_file in python_files:
        try:
            text = py_file.read_text(encoding="utf-8")
        except Exception:
            continue
 
        lines = text.split('\n')
        file_hash = content_hash(text)
        # Plan 268: resolve this file's rule set before anything keys on
        # it. The cache key carries the per-file set, so two byte-identical
        # files under different path rules never share one entry, and the
        # dashboard decides each file on exactly the rules ``oct lint``
        # decides it on.
        file_rules, _file_rule_changes = effective_rules(
            active_rules, to_project_relative(py_file, project_root),
            ctx.path_rules,
        )
        file_checks = [c for c in health_checks if file_rules.get(c, True)]
        cache_rules = dict(file_rules)
        cache_rules["_health_dashboard"] = True
        file_rules_hash = rules_hash(cache_rules)
        cached = cache.get(file_hash, file_rules_hash)
        syntax_error_message: str | None = None
        checks: Dict[str, Any]
 
        if cached is not None and all(
            check_name in cached.rule_results for check_name in file_checks
        ):
            checks = {
                check_name: cached.rule_results[check_name]
                for check_name in file_checks
            }
            cache_hits += 1
            syntax_error_message = cached.json_detail.get("syntax_error_message")
            # Header validity depends on the current path as well as content.
            # Match the linter cache's stale-path correction by refreshing only
            # this cheap path-sensitive verdict on an otherwise complete hit.
            if "header" in checks:
                checks["header"], _ = validate_header_block(
                    lines[:4] if len(lines) >= 4 else lines,
                    py_file,
                    ctx,
                )
        else:
            cache_misses += 1
            tree, syntax_warnings_list = safe_parse(text)
            syntax_error_message = (
                _describe_syntax_error(text) if tree is None else None
            )
            checks = {}
            for check_name in file_checks:
                if check_name == "syntax_error":
                    # The gate itself: an unparseable file (tree is None) fails,
                    # unconditionally, at every profile (Plan 170 semantics).
                    ok = tree is not None
                elif tree is None and check_name in _AST_DEPENDENT_HEALTH_CHECKS:
                    # NOT-RUN: the file did not parse, so every AST-derived verdict
                    # is meaningless. Record it as not-run (None) - never a pass.
                    checks[check_name] = None
                    continue
                else:
                    ok, _ = check_funcs[check_name](
                        text, lines, py_file, tree, syntax_warnings_list,
                    )
                checks[check_name] = ok
            cache.put(
                file_hash,
                file_rules_hash,
                LintCacheEntry(
                    content_hash=file_hash,
                    rule_results=checks,
                    json_detail={
                        "syntax_error_message": syntax_error_message,
                    },
                    oct_version=__version__,
                    timestamp=datetime.now().isoformat(),
                ),
            )
 
        # Plan 277: the file's own inline waivers, applied as an overlay
        # on top of whatever the checks (or the cache) said. The linter
        # never RUNS a waived rule, so it reports neither a pass nor a
        # failure for it; the dashboard counts per check, so the nearest
        # true statement here is that the file does not fail it - which
        # is also what makes the file fully_compliant, exactly as
        # ``oct lint`` calls it compliant.
        #
        # An overlay rather than a gate on purpose. The LintCache is
        # keyed by content hash, so a waiver IS in the key, but an entry
        # written before this plan carries the old verdict; recomputing
        # the overlay on every read makes the cache-hit and cache-miss
        # paths agree the way ``_lint_single_file`` makes them agree.
        # Only a genuine ``False`` is flipped: a NOT-RUN (``None``) stays
        # NOT-RUN, so a waiver can never buy an unparseable file a
        # vacuous pass.
        waived_checks = inline_disabled_rules(text) - _UNWAIVABLE_HEALTH_CHECKS
        if waived_checks:
            for check_name, ok in checks.items():
                if ok is False and check_name in waived_checks:
                    checks[check_name] = True
 
        if analysis_cache is not None:
            analysis_cache[py_file] = {
                "text": text,
                "checks": checks,
            }
 
        for check_name, ok in checks.items():
            # Plan 268: a path rule can also switch a check ON for one
            # path that the project disabled everywhere else, so the
            # tallies are reached by name rather than pre-seeded.
            applicable[check_name] = applicable.get(check_name, 0) + 1
            if ok is None:
                not_run[check_name] = not_run.get(check_name, 0) + 1
                continue
            if ok:
                counters[check_name] = counters.get(check_name, 0) + 1
            else:
                counters.setdefault(check_name, 0)
 
        # A file is fully compliant only if every check that RAN passed;
        # syntax_error (False on an unparseable file) alone sinks it.
        file_compliant = all(v for v in checks.values() if v is not None)
        if file_compliant:
            fully_compliant += 1
 
        if verbose:
            rel_path = to_project_relative(py_file, project_root)
            detail: Dict[str, Any] = {
                "path": rel_path,
                "checks": checks,
                "compliant": file_compliant,
            }
            if syntax_error_message is not None:
                # Reuse Plan 170's helper (do not reimplement) to surface the
                # concrete SyntaxError text for the broken file.
                detail["syntax_error_message"] = syntax_error_message
            file_details.append(detail)
 
    if cache_misses:
        try:
            cache.save()
        except OSError:
            pass
 
    compliance_pct = round((fully_compliant / total * 100), 1) if total > 0 else 0.0
 
    result = {
        "total_files": total,
        "fully_compliant": fully_compliant,
        "compliance_pct": compliance_pct,
        "checks": {},
        "cache_hits": cache_hits,
    }
 
    for check_name, count in counters.items():
        # Percentage is over the files where the check actually ran: an
        # AST-dependent check skipped on an unparseable file is NOT-RUN, not
        # a failure, so it must not drag the denominator (F21 / Plan 175),
        # and neither does a file a path rule excused from the check
        # entirely (plan 268).
        skipped = not_run.get(check_name, 0)
        ran = applicable.get(check_name, total) - skipped
        pct = round((count / ran * 100), 1) if ran > 0 else 0.0
        result["checks"][check_name] = {
            "pass": count,
            "total": ran,
            "not_run": skipped,
            "pct": pct,
        }
 
    if verbose:
        result["files"] = file_details
 
    return result
 
 
def check_fixable_health(
    project_root: Path,
    ctx: LinterContext,
    exclude_dirs: Optional[set] = None,
    wildcard_exclude: Optional[list] = None,
    active_rules: Optional[dict] = None,
    verbose: bool = False,
    analysis_cache: Optional[Dict[Path, Dict[str, Any]]] = None,
) -> Dict[str, Any]:
    """
    Count violations that oct format could fix automatically.
 
    Runs formatter checks in dry-run mode (no modifications).
    Only counts fixes for rules that are active.
 
    Plan 269 - "active" is a PER-FILE question wherever the project uses
    ``linter.path_rules``. *active_rules* is the run's set; each file's
    own set is resolved from it with
    :func:`~oct.linter.oct_lint.effective_rules` over ``ctx.path_rules``,
    which ``run_health`` has already parsed for :func:`check_lint_health`,
    so the panel counts a fix only where the fixer's rule applies to that
    file. Without it a module a ``tests/`` entry excused from
    ``dbg_import`` was still offered the fix - the panel disagreeing with
    ``oct lint`` on exactly the files the operator asked to treat
    differently. The counter set is widened the same way: a rule a path
    entry switches ON somewhere gets a counter even where the run
    disables it, or the files it applies to could never be counted at
    all. A project with no path rules produces byte-identical output.
 
    Plan 277 - the file's own inline ``# OCT-LINT: disable[-file]=<rule>``
    waivers are the third thing that can excuse a fixer, and the panel
    now reads them too, through
    :func:`~oct.core.waivers.inline_disabled_rules`: the linter's set,
    written once and shared with the gate's format pass. Before, the
    panel offered a ``dbg_import`` fix on every one of the nineteen
    modules the operator had ruled exempt - a fix the gate would have
    refused, and which on ``oct/core/diagnostics.py`` writes the
    module-level self-import that file's own waiver reason says raises
    ``ImportError``.
 
    When *verbose* is True the result includes a ``"files"`` key mapping
    to a list of ``{"path": <rel_path>, "fixes": [<fix_name>, ...]}``
    entries so the caller can show which files have fixable violations.
    """
    func = "check_fixable_health"
 
    from oct.formatter.oct_format import (
        fix_header_block,
        fix_docstring,
        fix_dbg_import,
        fix_func_pattern,
        FormatterContext,
    )
 
    if active_rules is None:
        active_rules = dict(_DEFAULT_RULES)
 
    fmt_ctx = FormatterContext(
        project_root=project_root,
        project_name=project_root.name,
        diagnostics_dir=ctx.diagnostics_dir,
        tests_dir=ctx.tests_dir,
        docs_dir=ctx.docs_dir,
        fix_mode=False,
        dry_run_mode=True,
        json_mode=False,
        verbose=False,
    )
 
    if analysis_cache is None:
        if exclude_dirs is None:
            exclude_dirs = set(LINTER_EXCLUDE_DIRS)
        if wildcard_exclude is None:
            wildcard_exclude = list(WILDCARD_EXCLUDE_DIRS)
        python_files = list(
            find_python_files(project_root, exclude_dirs, wildcard_exclude),
        )
    else:
        python_files = list(analysis_cache)
 
    # Map fixable check names to their corresponding lint rule names
    fix_rule_map = {
        "header": "header",
        "docstring": "docstring",
        "dbg_import": "dbg_import",
        "func_pattern": "func_pattern",
    }
    # Plan 268 populated ``ctx.path_rules``; plan 269 makes this panel read
    # it. ``getattr`` rather than attribute access because external callers
    # construct their own context and a missing attribute must mean "no
    # path scoping", not a crash in the dashboard.
    path_rules = list(getattr(ctx, "path_rules", None) or ())
    path_enabled = {
        rule
        for _prefix, rules in path_rules
        for rule, value in rules.items()
        if value
    }
    fixable = {
        k: 0 for k in fix_rule_map
        if active_rules.get(fix_rule_map[k], True)
        or fix_rule_map[k] in path_enabled
    }
 
    fix_funcs = {
        "header": fix_header_block,
        "docstring": fix_docstring,
        "dbg_import": fix_dbg_import,
        "func_pattern": fix_func_pattern,
    }
 
    file_details: list[dict] = []
 
    for py_file in python_files:
        lint_checks: Optional[Dict[str, Any]] = None
        if analysis_cache is not None:
            analysis = analysis_cache[py_file]
            text = analysis["text"]
            lint_checks = analysis["checks"]
        else:
            try:
                text = py_file.read_text(encoding="utf-8")
            except Exception:
                continue
 
        rel_path = to_project_relative(py_file, project_root)
        # Plan 269: the run's set is what applies wherever no entry matches,
        # so a project with no path rules never enters effective_rules and
        # every count below is what it was.
        if path_rules:
            file_rules, _changed = effective_rules(
                active_rules, rel_path, path_rules,
            )
        else:
            file_rules = active_rules
 
        # Plan 277: the file's own inline waivers are the third thing
        # that can excuse a fixer, beside the run's rules and the path
        # rules, and they are read here on the linter's terms. Without
        # this the panel offered a ``dbg_import`` fix on each of the
        # nineteen ruled modules - a fix the gate's format pass would
        # then refuse to accept, and which on ``oct/core/diagnostics.py``
        # writes the self-import its waiver reason says raises.
        waived_rules = inline_disabled_rules(text)
 
        file_fixes: list[str] = []
        for fix_name in fixable:
            rule_name = fix_rule_map[fix_name]
            if not file_rules.get(rule_name, True):
                continue
            if rule_name in waived_rules:
                continue
            if lint_checks is not None and lint_checks.get(rule_name) is True:
                continue
            if fix_name == "header" and lint_checks is not None:
                changed = lint_checks.get(rule_name) is False
            else:
                changed, _, _ = fix_funcs[fix_name](py_file, text, fmt_ctx)
            if changed:
                fixable[fix_name] += 1
                file_fixes.append(fix_name)
 
        if verbose and file_fixes:
            file_details.append({"path": rel_path, "fixes": file_fixes})
 
    fixable["total"] = sum(fixable.values())
    if verbose:
        fixable["files"] = file_details
    return fixable
 
 
def check_docs_health(
    project_root: Path,
    *,
    production_mode: bool = False,
) -> Dict[str, Any]:
    """
    Validate presence of required documentation files.
 
    Standard required set: ARCHITECTURE.md, CHANGELOG.md, Tests_README.md,
    run_tests.py, debug_config.json. When ``production_mode=True`` (Appendix H
    Level 2), SECURITY.md is also required; otherwise it is reported as
    optional alongside the required count.
    """
    func = "check_docs_health"
 
    base_required = [
        ("ARCHITECTURE.md", project_root / "docs" / "ARCHITECTURE.md"),
        ("CHANGELOG.md", project_root / "CHANGELOG.md"),
        ("Tests_README.md", project_root / "tests" / "Tests_README.md"),
        ("run_tests.py", project_root / "tests" / "run_tests.py"),
        # Plan 213 / T0: route through the ONE resolver instead of
        # hard-coding the legacy path, so a project migrated to
        # ``.option_c/`` is not reported as missing its diagnostics config.
        ("debug_config.json", resolve_debug_config(project_root)),
    ]
    production_only = [
        ("SECURITY.md", project_root / "SECURITY.md"),
    ]
 
    required_files = base_required + (production_only if production_mode else [])
    optional_files = [] if production_mode else production_only
 
    files_status = []
    present_count = 0
 
    for name, path in required_files:
        exists = path.is_file()
        files_status.append({"name": name, "present": exists, "required": True})
        if exists:
            present_count += 1
 
    for name, path in optional_files:
        exists = path.is_file()
        files_status.append({"name": name, "present": exists, "required": False})
 
    return {
        "total_required": len(required_files),
        "total_present": present_count,
        "files": files_status,
        "production_mode": production_mode,
    }
 
 
_COVERAGE_TOTAL_RE = re.compile(r"^TOTAL\s+\d+\s+\d+\s+(\d+(?:\.\d+)?)%", re.MULTILINE)
 
 
def check_test_health(
    project_root: Path,
    timeout: int = 300,
    *,
    coverage_threshold: Optional[float] = None,
    production_mode: bool = False,
    coverage_target: Optional[str] = None,
) -> Dict[str, Any]:
    """
    Run project tests and capture pass/fail status.
 
    Uses pytest subprocess to run tests and captures the exit code
    and summary line.
 
    Parameters
    ----------
    project_root:
        Path to the project root containing a ``tests/`` directory.
    timeout:
        Maximum seconds to wait for pytest to complete. Default 300 (OI-413).
    coverage_threshold:
        FS-C3 / Drift item C3 - when set, run pytest with ``--cov`` and
        compare the resulting TOTAL coverage percentage against this
        threshold. ``None`` (default) skips the coverage measurement
        entirely so existing CI keeps its prior behaviour.
    production_mode:
        When ``True`` together with a ``coverage_threshold``, a coverage
        result below the threshold flips ``status`` to ``"fail"``.
        Outside production mode, below-threshold coverage is reported
        but never blocks the dashboard.
    coverage_target:
        Module / package name to measure (passed as ``--cov=<target>``).
        Defaults to the project root directory name when ``None``.
    """
    func = "check_test_health"
 
    import importlib.util
 
    tests_dir = project_root / "tests"
    if not tests_dir.is_dir():
        return {
            "status": "not_run",
            "exit_code": -1,
            "summary": "No tests/ directory found",
            "details": [],
            "coverage_pct": None,
            "coverage_threshold": coverage_threshold,
            "coverage_blocking": production_mode,
        }
 
    # Detect missing pytest before attempting subprocess - avoids silent FAIL
    if importlib.util.find_spec("pytest") is None:
        return {
            "status": "fail",
            "exit_code": -1,
            "summary": "pytest not installed - run: pip install pytest",
            "details": [],
            "coverage_pct": None,
            "coverage_threshold": coverage_threshold,
            "coverage_blocking": production_mode,
        }
 
    # FS-C3: when coverage measurement is requested, ensure pytest-cov is
    # available; degrade gracefully (run without --cov) when missing.
    coverage_requested = coverage_threshold is not None
    coverage_available = (
        coverage_requested
        and importlib.util.find_spec("pytest_cov") is not None
    )
 
    pytest_args = [
        sys.executable, "-m", "pytest", str(tests_dir),
        "-n", "auto", "--dist=worksteal", "--tb=line", "--no-header",
    ]
    if coverage_available:
        target = coverage_target or project_root.name
        pytest_args.extend([f"--cov={target}", "--cov-report=term"])
 
    try:
        result = subprocess.run(
            pytest_args,
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
            timeout=timeout,
            cwd=str(project_root),
        )
 
        # Extract summary line - scan from end since pytest always puts it last.
        # Match on keywords OR the timing pattern ("N passed in 0.31s") as a fallback
        # since encoding issues can occasionally mangle keyword characters on Windows.
        _SUMMARY_RE = re.compile(r"\b\d+ \w+ in [\d.]+s")
        _DECORATION_RE = re.compile(r"^=+\s*|\s*=+$")
        summary = ""
        for line in reversed(result.stdout.strip().split('\n')):
            line = line.strip()
            if "passed" in line or "failed" in line or "error" in line or _SUMMARY_RE.search(line):
                summary = _DECORATION_RE.sub("", line).strip()
                break
 
        # Extract actionable error detail lines (max 5) from --tb=line output.
        # pytest --tb=line produces lines like:
        #   path/to/test.py - ErrorType: message
        #   FAILED tests/test_foo.py::test_bar - AssertionError
        detail_lines = []
        if result.returncode != 0:
            for line in result.stdout.strip().split('\n'):
                line = line.strip()
                if not line or line.startswith("=") or line.startswith("."):
                    continue
                if ("Error" in line or "Exception" in line
                        or line.startswith("FAILED ")
                        or line.startswith("ERROR ")):
                    detail_lines.append(line)
                    if len(detail_lines) >= 5:
                        break
 
        # Fall back to stderr when stdout has no useful summary (e.g. import errors)
        if not summary and result.stderr.strip():
            err_lines = result.stderr.strip().split('\n')
            summary = err_lines[-1].strip()
 
        # Tertiary fallback - always provide a summary even if regex + stderr both fail
        if not summary:
            summary = f"pytest exited with code {result.returncode}"
 
        status = "pass" if result.returncode == 0 else "fail"
 
        # FS-C3: parse pytest-cov "TOTAL ... NN%" line when coverage was requested.
        coverage_pct: Optional[float] = None
        if coverage_available:
            cov_match = _COVERAGE_TOTAL_RE.search(result.stdout)
            if cov_match:
                try:
                    coverage_pct = float(cov_match.group(1))
                except ValueError:
                    coverage_pct = None
 
        # Production-mode gating: a coverage result below threshold flips
        # status to "fail" only when production_mode is set. Outside
        # production, below-threshold coverage is informational.
        if (
            production_mode
            and coverage_pct is not None
            and coverage_threshold is not None
            and coverage_pct < coverage_threshold
            and status == "pass"
        ):
            status = "fail"
            summary = (
                f"{summary} | coverage {coverage_pct:.1f}% below "
                f"production threshold {coverage_threshold:.1f}%"
            ).strip()
 
        # Surface a clear diagnostic when the user asked for coverage but
        # pytest-cov isn't installed.
        coverage_msg: Optional[str] = None
        if coverage_requested and not coverage_available:
            coverage_msg = (
                "pytest-cov not installed - install with: "
                "pip install -e .[test]"
            )
 
        return {
            "status": status,
            "exit_code": result.returncode,
            "summary": summary,
            "details": detail_lines,
            "coverage_pct": coverage_pct,
            "coverage_threshold": coverage_threshold,
            "coverage_blocking": production_mode,
            "coverage_message": coverage_msg,
        }
 
    except subprocess.TimeoutExpired:
        return {
            "status": "timeout",
            "exit_code": -1,
            "summary": f"Tests timed out after {timeout} seconds",
            "details": [],
            "coverage_pct": None,
            "coverage_threshold": coverage_threshold,
            "coverage_blocking": production_mode,
        }
    except Exception as e:
        return {
            "status": "error",
            "exit_code": -1,
            "summary": str(e),
            "details": [],
            "coverage_pct": None,
            "coverage_threshold": coverage_threshold,
            "coverage_blocking": production_mode,
        }
 
 
def check_compat_health() -> Dict[str, Any]:
    """
    Check oc_diagnostics version compatibility.
    """
    func = "check_compat_health"
    _dbg("OCT.HEALTH", func, "checking oc_diagnostics compatibility", 3)
 
    msg = check_oc_diagnostics_compat()
 
    try:
        from importlib.metadata import version as _pkg_version
        version = _pkg_version("oc_diagnostics")
        installed = True
    except Exception:
        version = "not installed"
        installed = False
 
    return {
        "installed": installed,
        "version": version,
        "compatible": msg is None or msg == "",
        "message": msg or "",
    }
 
 
def check_venv_health(project_root: Path) -> Dict[str, Any]:
    """
    Check virtual environment status for the project.
 
    Performs three layered checks:
    1. Whether the current process is running inside *any* venv.
    2. Whether a canonical ``.venv`` directory exists at the project root or its
       immediate parent (monorepo pattern: shared venv one level above subproject).
    3. Whether the active venv is the project's own ``.venv`` (not a foreign env).
 
    Returns a dict with ``status`` (OK / WARN / FAIL), ``active`` (bool),
    ``venv_found`` (bool), ``active_path`` (str | None), and ``message`` (str).
    """
    func = "check_venv_health"  # noqa: F841 - Option C func pattern
    _dbg("OCT.HEALTH", func, "checking the virtualenv", 3)
 
    in_venv: bool = sys.prefix != sys.base_prefix
    active_path: Optional[str] = sys.prefix if in_venv else None
 
    # Probe subproject root first, then repo root (monorepo: oct/ lives inside
    # Option_C/ which owns the shared .venv).
    canonical_venv: Optional[Path] = None
    for _candidate in (project_root / ".venv", project_root.parent / ".venv"):
        if _candidate.is_dir():
            canonical_venv = _candidate
            break
    venv_found: bool = canonical_venv is not None
 
    # Is the active venv the project's own .venv?
    own_venv = False
    if in_venv and venv_found and canonical_venv is not None:
        try:
            own_venv = Path(sys.prefix).resolve() == canonical_venv.resolve()
        except Exception:
            own_venv = False
 
    if not in_venv:
        status = "WARN"
        message = (
            "Not running inside a virtual environment. "
            "Run: .venv\\Scripts\\activate (Windows) or source .venv/bin/activate (Unix)"
        )
    elif not venv_found:
        status = "WARN"
        message = (
            f"Active venv: {active_path}. "
            "No .venv/ found at project root or repo root -- "
            "create one with: python -m venv .venv"
        )
    elif not own_venv:
        status = "WARN"
        message = (
            f"Active venv ({active_path}) is not the project .venv "
            f"({canonical_venv}). Activate the correct venv for dependency isolation."
        )
    else:
        status = "OK"
        message = f"Project .venv active ({active_path})"
 
    return {
        "status": status,
        "active": in_venv,
        "venv_found": venv_found,
        "own_venv": own_venv,
        "active_path": active_path,
        "venv_path": str(canonical_venv) if canonical_venv else None,
        "message": message,
    }
 
 
# ============================================================================
# Git Health (Phase 4E - G-E5)
# ============================================================================
 
 
def check_git_health(project_root: Path) -> Dict[str, Any]:
    """Run git-related health checks.
 
    Returns a dict of checks, each with ``status`` (OK / WARN / SKIP)
    and ``detail``.  If git is not installed or the project is not a
    repo, all items return SKIP with an explanatory message.
    """
    func = "check_git_health"
    _dbg("OCT.HEALTH", func, "checking git state", 3)
    import shutil
 
    checks: Dict[str, Dict[str, str]] = {}
 
    # Guard: git installed?
    if shutil.which("git") is None:
        skip = {"status": "SKIP", "detail": "Git is not installed"}
        return {
            "is_repo": skip,
            "branch": skip,
            "naming_conformant": skip,
            "clean_worktree": skip,
            "ahead_behind": skip,
            "hooks_installed": skip,
            "last_commit_conventional": skip,
        }
 
    try:
        from oct.core.git import (
            is_git_repo,
            current_branch,
            has_uncommitted_changes,
            last_commit_message,
            ahead_behind as _ahead_behind,
        )
    except ImportError:
        skip = {"status": "SKIP", "detail": "oct.core.git not available"}
        return {
            "is_repo": skip,
            "branch": skip,
            "naming_conformant": skip,
            "clean_worktree": skip,
            "ahead_behind": skip,
            "hooks_installed": skip,
            "last_commit_conventional": skip,
        }
 
    # is_repo
    if not is_git_repo(project_root):
        skip = {"status": "SKIP", "detail": "Not a git repository"}
        return {
            "is_repo": {"status": "FAIL", "detail": "Not a git repository. Run 'oct git init' to initialise."},
            "branch": skip,
            "naming_conformant": skip,
            "clean_worktree": skip,
            "ahead_behind": skip,
            "hooks_installed": skip,
            "last_commit_conventional": skip,
        }
 
    checks["is_repo"] = {"status": "OK", "detail": "Git repository detected"}
 
    # branch
    try:
        branch = current_branch(project_root)
        if branch == "HEAD":
            checks["branch"] = {"status": "WARN", "detail": "Detached HEAD"}
        else:
            checks["branch"] = {"status": "OK", "detail": branch}
    except Exception:
        branch = None
        checks["branch"] = {"status": "WARN", "detail": "Could not determine branch"}
 
    # naming_conformant
    try:
        from oct.git.policy import validate_branch_name
        valid, msg = validate_branch_name(branch, _load_octrc_for_health(project_root))
        if valid:
            checks["naming_conformant"] = {"status": "OK", "detail": "Branch name conforms to naming convention"}
        else:
            checks["naming_conformant"] = {"status": "WARN", "detail": msg}
    except Exception:
        checks["naming_conformant"] = {"status": "SKIP", "detail": "Could not validate branch name"}
 
    # clean_worktree
    try:
        dirty = has_uncommitted_changes(project_root)
        if dirty:
            checks["clean_worktree"] = {"status": "WARN", "detail": "Uncommitted changes present"}
        else:
            checks["clean_worktree"] = {"status": "OK", "detail": "Working tree is clean"}
    except Exception:
        checks["clean_worktree"] = {"status": "SKIP", "detail": "Could not check worktree"}
 
    # ahead_behind
    try:
        ahead, behind = _ahead_behind(project_root)
        if ahead == 0 and behind == 0:
            checks["ahead_behind"] = {"status": "OK", "detail": "Up to date with remote"}
        else:
            parts = []
            if ahead:
                parts.append(f"{ahead} ahead")
            if behind:
                parts.append(f"{behind} behind")
            checks["ahead_behind"] = {"status": "WARN", "detail": ", ".join(parts)}
    except Exception:
        checks["ahead_behind"] = {"status": "OK", "detail": "No upstream configured"}
 
    # hooks_installed
    config_path = project_root / ".pre-commit-config.yaml"
    if config_path.is_file():
        try:
            content = config_path.read_text(encoding="utf-8")
            oct_hooks = ["oct-quality-gate", "oct-secrets-check", "oct-commit-msg", "oct-pre-push"]
            found = [h for h in oct_hooks if f"id: {h}" in content]
            if len(found) == len(oct_hooks):
                checks["hooks_installed"] = {"status": "OK", "detail": f"All {len(oct_hooks)} hooks installed"}
            else:
                missing = [h for h in oct_hooks if h not in found]
                checks["hooks_installed"] = {"status": "WARN", "detail": f"Missing: {', '.join(missing)}"}
        except Exception:
            checks["hooks_installed"] = {"status": "WARN", "detail": "Could not read hooks config"}
    else:
        checks["hooks_installed"] = {"status": "WARN", "detail": "No .pre-commit-config.yaml found"}
 
    # last_commit_conventional
    try:
        from oct.git.conventional import parse_commit_message
        msg = last_commit_message(project_root)
        if not msg:
            checks["last_commit_conventional"] = {"status": "SKIP", "detail": "No commits yet"}
        elif parse_commit_message(msg) is not None:
            checks["last_commit_conventional"] = {"status": "OK", "detail": "Last commit follows Conventional Commits"}
        else:
            checks["last_commit_conventional"] = {"status": "WARN", "detail": "Last commit is not Conventional Commits format"}
    except Exception:
        checks["last_commit_conventional"] = {"status": "SKIP", "detail": "Could not check last commit"}
 
    return checks
 
 
def _load_octrc_for_health(project_root: Path) -> Optional[dict]:
    """Load ``.octrc.json`` for health checks (best-effort).
 
    OI-519: thin wrapper around :func:`oct.core.octrc.load_octrc_safe`;
    retains the historical ``None`` return on missing/invalid so existing
    callers keep their truthy-check semantics.
    """
    from oct.core.octrc import load_octrc_safe
    from oct.core.option_c_dir import resolve_octrc
    # FS-539: the octrc may live at .option_c/octrc.json (modern) or
    # .octrc.json (legacy) - resolve_octrc picks whichever exists and
    # returns the legacy path when neither does, giving us a truthy
    # existence check either way.
    if not resolve_octrc(project_root).is_file():
        return None
    data = load_octrc_safe(project_root)
    return data or None
 
 
# ============================================================================
# Output Formatting
# ============================================================================
 
 
# Lint checks shown in the health dashboard (subset of all lint rules).
# F21 / Plan 175: "syntax_error" leads so a broken file is visibly RED in the
# dashboard instead of silently healthy.
_HEALTH_CHECKS = [
    "syntax_error",
    "header", "docstring", "diagnostics_profile",
    "dbg_usage", "dbg_import", "func_pattern", "syntax_warnings",
]
 
# Fixable categories and their display labels.
_FIX_LABELS: Dict[str, str] = {
    "header": "Header blocks",
    "docstring": "Docstrings",
    "dbg_import": "_dbg imports",
    "func_pattern": "func patterns",
}
 
# Documentation files surfaced in the dashboard. CHANGELOG.md is required at
# Standard maturity; SECURITY.md is required only at Production maturity but is
# always shown in the dashboard so its absence is visible.
_DOC_FILES = [
    "ARCHITECTURE.md",
    "CHANGELOG.md",
    "Tests_README.md",
    "run_tests.py",
    "debug_config.json",
    "SECURITY.md",
]
 
# Git check keys and display labels.
_GIT_LABELS: Dict[str, str] = {
    "is_repo": "Repository",
    "branch": "Branch",
    "naming_conformant": "Branch naming",
    "clean_worktree": "Working tree",
    "ahead_behind": "Remote sync",
    "hooks_installed": "Hooks",
    "last_commit_conventional": "Last commit",
}
 
 
@dataclass
class _HealthLineMap:
    """Maps dashboard sections to line indices for in-place updates."""
    lint: int = 0
    lint_checks: Dict[str, int] = field(default_factory=dict)
    fixable: int = 0
    fixable_items: Dict[str, int] = field(default_factory=dict)
    docs: int = 0
    docs_files: Dict[str, int] = field(default_factory=dict)
    tests: int = 0
    compat: int = 0
    venv: int = 0
    git: int = 0
    git_checks: Dict[str, int] = field(default_factory=dict)
    footer: int = 0
 
 
def _build_health_template(
    project_name: str,
    excluded_dirs: List[str],
    disabled_rules: List[str],
    active_checks: List[str],
    fixable_names: List[str],
    use_color: bool,
) -> tuple:
    """Build the dashboard template lines and line-index map.
 
    Returns ``(lines, line_map)`` where *lines* is a list of strings and
    *line_map* is a :class:`_HealthLineMap` with indices into that list.
    """
    C = "\033[96m" if use_color else ""
    X = "\033[0m" if use_color else ""
    lm = _HealthLineMap()
    lines: List[str] = []
 
    def add(text: str = "") -> int:
        idx = len(lines)
        lines.append(text)
        return idx
 
    # Header
    add(f"{C}{'=' * 70}{X}")
    add(f"{C}OCT HEALTH REPORT -- {project_name}{X}")
    add(f"{C}{'=' * 70}{X}")
    add()
 
    # Config overrides
    if excluded_dirs:
        add(f"Excluded directories: {', '.join(excluded_dirs)} (via .octrc.json)")
    if disabled_rules:
        labels = [r.replace("_", " ") for r in disabled_rules]
        add(f"Disabled rules:       {', '.join(labels)} (via .octrc.json)")
    if excluded_dirs or disabled_rules:
        add()
 
    # Lint compliance
    lm.lint = add("[ ] Lint Compliance:")
    for ck in active_checks:
        label = ck.replace("_", " ").title()
        lm.lint_checks[ck] = add(f"  [ ] {label + ':':<20}")
    add()
 
    # Fixable violations
    lm.fixable = add("[ ] Fixable Violations:")
    for fn in fixable_names:
        lm.fixable_items[fn] = add(f"  [ ] {_FIX_LABELS.get(fn, fn) + ':':<20}")
    add()
 
    # Documentation
    lm.docs = add("[ ] Documentation:")
    for df in _DOC_FILES:
        lm.docs_files[df] = add(f"  [ ] {df}")
    add()
 
    # Tests
    lm.tests = add("[ ] Tests:")
    add()
 
    # Compatibility
    lm.compat = add("[ ] oc_diagnostics:")
    add()
 
    # Virtual environment
    lm.venv = add("[ ] Virtual Env:")
    add()
 
    # Git integration
    lm.git = add("[ ] Git Integration:")
    for gk, gl in _GIT_LABELS.items():
        lm.git_checks[gk] = add(f"  [ ] {gl + ':':<20}")
    add()
 
    # Footer
    lm.footer = add("  Waiting to start...")
    add()
    add(f"{C}{'=' * 70}{X}")
 
    return lines, lm
 
 
def _pct_str(count: int, total: int) -> str:
    """Format count/total with percentage."""
    if total == 0:
        return "0/0 (0.0%)"
    pct = round(count / total * 100, 1)
    return f"{count}/{total} ({pct}%)"
 
 
def _print_terminal_report(report: Dict[str, Any], verbose: bool) -> None:
    """Print human-readable terminal health report."""
    func = "_print_terminal_report"
 
    lint = report["lint"]
    fixable = report["fixable"]
    docs = report["docs"]
    tests = report["tests"]
    compat = report["compat"]
 
    print(f"\n{CYAN}{'=' * 70}{RESET}")
    print(f"{CYAN}OCT HEALTH REPORT -- {report['project']}{RESET}")
    print(f"{CYAN}{'=' * 70}{RESET}\n")
 
    # FS-504: prominent top-of-report banner when running outside a venv.
    # Kept as WARN (not FAIL) so CI consumers can still exit 0; the banner
    # is what turns "venv missing" from an easily-missed per-section note
    # into something the operator cannot overlook.
    venv_data = report.get("venv") or {}
    if venv_data.get("active") is False:
        print(f"{YELLOW}{'!' * 70}{RESET}")
        print(f"{YELLOW}WARNING: not running inside a virtual environment.{RESET}")
        print(f"{YELLOW}Create one with:  python -m venv .venv && activate{RESET}")
        print(f"{YELLOW}{'!' * 70}{RESET}\n")
 
    # Config overrides (from .octrc.json)
    excluded = report.get("excluded_dirs", [])
    disabled = report.get("disabled_rules", [])
    if excluded:
        print(f"Excluded directories: {YELLOW}{', '.join(excluded)}{RESET} (via .octrc.json)")
    if disabled:
        labels = [r.replace("_", " ") for r in disabled]
        print(f"Disabled rules:       {YELLOW}{', '.join(labels)}{RESET} (via .octrc.json)")
    if excluded or disabled:
        print()
 
    # Lint compliance
    total = lint["total_files"]
    compliant = lint["fully_compliant"]
    pct = lint["compliance_pct"]
    pct_color = GREEN if pct == 100.0 else RED
    cache_hits = lint.get("cache_hits", 0)
    print(
        f"Lint Compliance:      {pct_color}{pct}%{RESET} "
        f"({compliant}/{total} files fully compliant; {cache_hits} cache hits)"
    )
 
    for check_name, check_data in lint["checks"].items():
        label = check_name.replace("_", " ").title()
        c = GREEN if check_data["pass"] == check_data["total"] else RED
        status = _pct_str(check_data["pass"], check_data["total"])
        print(f"  {label + ':':<22}{c}{status}{RESET}")
 
    # Fixable violations
    fix_total = fixable["total"]
    fix_color = GREEN if fix_total == 0 else YELLOW
    print(f"\nFixable Violations:   {fix_color}{fix_total} total{RESET}")
    if fix_total > 0:
        for fix_name in ("header", "docstring", "dbg_import", "func_pattern"):
            if fix_name in fixable:
                fix_labels = {
                    "header": "Header blocks",
                    "docstring": "Docstrings",
                    "dbg_import": "_dbg imports",
                    "func_pattern": "func patterns",
                }
                print(f"  {fix_labels[fix_name] + ':':<22}{YELLOW}{fixable[fix_name]}{RESET}")
 
    # Documentation
    doc_color = GREEN if docs["total_present"] == docs["total_required"] else RED
    doc_status = f"{docs['total_present']}/{docs['total_required']} present"
    print(f"\nDocumentation:        {doc_color}{doc_status}{RESET}")
    for file_info in docs["files"]:
        if file_info["present"]:
            print(f"  {file_info['name']:<22}{GREEN}[OK]{RESET}")
        else:
            print(f"  {file_info['name']:<22}{RED}[MISSING]{RESET}")
 
    # Tests
    test_label = tests["status"].upper().replace("_", " ")
    test_color = GREEN if tests["status"] == "pass" else (
        YELLOW if tests["status"] in ("skipped", "not_run") else RED
    )
    summary = tests.get("summary") or tests.get("reason", "")
    test_summary = f" ({summary})" if summary else ""
    print(f"\nTests:                {test_color}{test_label}{RESET}{test_summary}")
    for detail in tests.get("details", []):
        print(f"  {detail}")
 
    # Compatibility
    if compat["installed"]:
        if compat["compatible"]:
            print(f"\noc_diagnostics:       v{compat['version']} -- {GREEN}compatible{RESET}")
        else:
            print(f"\noc_diagnostics:       v{compat['version']} -- {RED}INCOMPATIBLE{RESET}")
    else:
        print(f"\noc_diagnostics:       {RED}NOT INSTALLED{RESET}")
    if compat["message"]:
        print(f"  {compat['message']}")
 
    # Virtual environment
    venv_data = report.get("venv")
    if venv_data:
        venv_color = GREEN if venv_data["status"] == "OK" else YELLOW
        venv_label = "[OK]" if venv_data["status"] == "OK" else "[WARN]"
        print(f"\nVirtual Env:          {venv_color}{venv_label}{RESET} {venv_data['message']}")
 
    # Git
    git_data = report.get("git")
    if git_data:
        print(f"\nGit Integration:")
        _STATUS_COLORS = {"OK": GREEN, "WARN": YELLOW, "FAIL": RED, "SKIP": YELLOW}
        _STATUS_LABELS = {
            "is_repo": "Repository",
            "branch": "Branch",
            "naming_conformant": "Branch naming",
            "clean_worktree": "Working tree",
            "ahead_behind": "Remote sync",
            "hooks_installed": "Hooks",
            "last_commit_conventional": "Last commit",
        }
        for key in _STATUS_LABELS:
            check = git_data.get(key)
            if check is None:
                continue
            label = _STATUS_LABELS[key]
            status = check["status"]
            detail = check["detail"]
            color = _STATUS_COLORS.get(status, RESET)
            print(f"  {label + ':':<22}{color}[{status}]{RESET} {detail}")
 
    # Verbose: per-file breakdown
    if verbose and "files" in lint:
        print(f"\n{CYAN}{'-' * 70}{RESET}")
        print(f"{CYAN}Per-file breakdown:{RESET}")
        print(f"{CYAN}{'-' * 70}{RESET}\n")
 
        for file_info in lint["files"]:
            if not file_info["compliant"]:
                failed = [k for k, v in file_info["checks"].items() if not v]
                print(f"  {file_info['path']}")
                print(f"    Failed: {RED}{', '.join(failed)}{RESET}")
 
    print(f"\n{CYAN}{'=' * 70}{RESET}\n")
 
 
def _print_json_report(report: Dict[str, Any]) -> None:
    """Print JSON health report."""
    print(json.dumps(report, indent=2))
 
 
# ============================================================================
# Main Entry Point
# ============================================================================
 
 
def run_health(
    project_root: Path,
    json_mode: bool = False,
    verbose: bool = False,
    test_timeout: int | None = None,
    run_tests: bool | None = None,
    log_output: bool = False,
    timing: bool = False,
    timing_last: int = 20,
    write_baseline: bool = False,
    force_host: bool = False,
    session: Path | None = None,
) -> None:
    """
    Run the project health dashboard.
 
    Parameters
    ----------
    project_root : Path
        The detected Option C project root directory.
    json_mode : bool
        Output machine-readable JSON instead of terminal report.
    verbose : bool
        Show per-file details in addition to summary.
    test_timeout : int | None
        Maximum seconds to wait for pytest. Overrides ``health.test_timeout``
        in ``.octrc.json``. Defaults to 300 when neither is set (OI-413).
    run_tests : bool | None
        Run the project test suite. The CLI always passes a boolean and skips
        tests by default; ``None`` preserves legacy direct-call behaviour.
    """
    func = "run_health"
    _dbg("OCT.HEALTH", func, f"health run starting at {project_root}", 3)
    if timing or write_baseline or session is not None or force_host:
        import click
        from oct.timing.report import (
            timing_report, write_baseline as save_baseline, session_report, render_report,
        )
        try:
            data = timing_report(project_root, timing_last)
            if write_baseline:
                save_baseline(project_root, data, force_host=force_host)
                data = timing_report(project_root, timing_last)
            if session is not None:
                data["session"] = session_report(session)
        except (OSError, ValueError) as exc:
            raise click.ClickException(str(exc)) from exc
        if json_mode:
            _print_json_report({"timing": data})
        else:
            print(render_report(data))
        return
 
    # Plan 245's command-seam witness invokes health against a completely
    # empty, explicitly allowed outside-project directory and requires only
    # the outer ``health`` record. Real project trees and health fixtures
    # receive the seven Plan 250 child stages.
    record_check_timings = any(project_root.iterdir())
 
    def check_timing(stage: str) -> Any:
        return (
            timed_stage(project_root, stage)
            if record_check_timings
            else nullcontext(None)
        )
 
    # Build linter context for check functions
    diagnostics_dir = project_root / "oc_diagnostics"
    if not diagnostics_dir.is_dir():
        diagnostics_dir = project_root / "diagnostics"
        if diagnostics_dir.is_dir():
            print(
                "Warning: 'diagnostics/' is deprecated and will be removed in v1.0.0. "
                "Rename to 'oc_diagnostics/'.",
                file=sys.stderr,
            )
 
    ctx = LinterContext(
        project_root=project_root,
        project_name=project_root.name,
        diagnostics_dir=diagnostics_dir,
        tests_dir=project_root / "tests",
        docs_dir=project_root / "docs",
    )
 
    # Load per-project overrides from .octrc.json
    exclude_dirs = set(LINTER_EXCLUDE_DIRS)
    wildcard_exclude = list(WILDCARD_EXCLUDE_DIRS)
    excluded_by_config: List[str] = []
    active_rules = dict(_DEFAULT_RULES)
    disabled_rules: List[str] = []
 
    # Default timeout for tests; overridable via .octrc.json health.test_timeout
    # or explicit test_timeout parameter (OI-413).
    # OI-519: routed through central oct.core.octrc loader.
    effective_test_timeout: int = 300
    from oct.core.octrc import load_octrc_safe
    octrc = load_octrc_safe(project_root)
    linter_cfg = octrc.get("linter", {})
    if isinstance(linter_cfg, dict):
        for d in linter_cfg.get("exclude_dirs_add", []):
            exclude_dirs.add(d)
            excluded_by_config.append(d)
        for d in linter_cfg.get("exclude_dirs_remove", []):
            exclude_dirs.discard(d)
        for d in linter_cfg.get("wildcard_exclude_add", []):
            if d not in wildcard_exclude:
                wildcard_exclude.append(d)
        # Resolve rule profile and per-rule overrides
        profile_name = linter_cfg.get("profile", "default")
        active_rules = dict(_PROFILES.get(profile_name, _DEFAULT_RULES))
        rule_overrides = linter_cfg.get("rules", {})
        if isinstance(rule_overrides, dict):
            active_rules.update(rule_overrides)
        disabled_rules = [r for r, v in active_rules.items() if not v]
        # Plan 268: the per-path overrides ride the context, so
        # check_lint_health resolves the same per-file rule set oct lint
        # does and the dashboard cannot disagree with it on a file.
        ctx.path_rules = parse_path_rules(linter_cfg)
    health_cfg = octrc.get("health", {})
    if isinstance(health_cfg, dict):
        cfg_timeout = health_cfg.get("test_timeout")
        if isinstance(cfg_timeout, int) and cfg_timeout > 0:
            effective_test_timeout = cfg_timeout
 
    # Explicit parameter takes precedence over config
    if test_timeout is not None:
        effective_test_timeout = test_timeout
 
    # Compute dashboard parameters from loaded config
    active_checks = [c for c in _HEALTH_CHECKS if active_rules.get(c, True)]
    _fix_rule_map = {
        "header": "header", "docstring": "docstring",
        "dbg_import": "dbg_import", "func_pattern": "func_pattern",
    }
    fixable_names = [k for k in _fix_rule_map if active_rules.get(_fix_rule_map[k], True)]
 
    from oct.core.terminal import supports_ansi_on
    use_color = supports_ansi_on(sys.stdout) and not json_mode
    G = "\033[92m" if use_color else ""
    R = "\033[91m" if use_color else ""
    Y = "\033[93m" if use_color else ""
    X = "\033[0m" if use_color else ""
 
    template_lines, lm = _build_health_template(
        project_name=project_root.name,
        excluded_dirs=excluded_by_config,
        disabled_rules=disabled_rules,
        active_checks=active_checks,
        fixable_names=fixable_names,
        use_color=use_color,
    )
 
    dashboard = Dashboard(quiet=json_mode)
    dashboard.render(template_lines, footer_line=lm.footer, tasks_total=7)
    lint_analysis: Dict[Path, Dict[str, Any]] = {}
 
    # --- Check 1: Lint compliance ---
    dashboard.set_task("Lint compliance")
    dashboard.update_line(lm.lint, f"{Y}[>]{X} Lint Compliance:  running...")
    with check_timing("health.lint") as timing_sample:
        lint_result = check_lint_health(
            project_root, ctx, verbose=verbose,
            exclude_dirs=exclude_dirs, wildcard_exclude=wildcard_exclude,
            active_rules=active_rules,
            analysis_cache=lint_analysis,
        )
        if timing_sample is not None:
            timing_sample.files = lint_result["total_files"]
    pct = lint_result["compliance_pct"]
    pc = G if pct == 100.0 else R
    dashboard.update_line(
        lm.lint,
        f"{G}[x]{X} Lint Compliance:      "
        f"{pc}{pct}%{X} ({lint_result['fully_compliant']}/{lint_result['total_files']}"
        f" files fully compliant; {lint_result.get('cache_hits', 0)} cache hits)",
    )
    for ck, idx in lm.lint_checks.items():
        cd = lint_result["checks"].get(ck)
        label = ck.replace("_", " ").title()
        if cd:
            c = G if cd["pass"] == cd["total"] else R
            dashboard.update_line(
                idx, f"  {G}[x]{X} {label + ':':<20}{c}{_pct_str(cd['pass'], cd['total'])}{X}",
            )
        else:
            dashboard.update_line(idx, f"  {Y}[-]{X} {label + ':':<20}disabled")
    dashboard.complete_task()
 
    # --- Check 2: Fixable violations ---
    dashboard.set_task("Auto-fixable checks")
    dashboard.update_line(lm.fixable, f"{Y}[>]{X} Fixable Violations:  running...")
    with check_timing("health.fixable") as timing_sample:
        fixable_result = check_fixable_health(
            project_root, ctx,
            exclude_dirs=exclude_dirs, wildcard_exclude=wildcard_exclude,
            active_rules=active_rules,
            verbose=verbose,
            analysis_cache=lint_analysis,
        )
        if timing_sample is not None:
            timing_sample.files = lint_result["total_files"]
    fix_total = fixable_result.get("total", 0)
    fc = G if fix_total == 0 else Y
    dashboard.update_line(lm.fixable, f"{G}[x]{X} Fixable Violations:   {fc}{fix_total} total{X}")
    for fn, idx in lm.fixable_items.items():
        val = fixable_result.get(fn, 0)
        vc = G if val == 0 else Y
        dashboard.update_line(
            idx, f"  {G}[x]{X} {_FIX_LABELS.get(fn, fn) + ':':<20}{vc}{val}{X}",
        )
    dashboard.complete_task()
 
    # --- Check 3: Documentation ---
    dashboard.set_task("Documentation")
    dashboard.update_line(lm.docs, f"{Y}[>]{X} Documentation:  running...")
    _octrc_cfg = load_octrc_safe(project_root)
    _production_mode = bool(
        (_octrc_cfg.get("health") or {}).get("production_mode", False)
    )
    with check_timing("health.docs") as timing_sample:
        docs_result = check_docs_health(project_root, production_mode=_production_mode)
        if timing_sample is not None:
            timing_sample.files = len(docs_result["files"])
    dc = G if docs_result["total_present"] == docs_result["total_required"] else R
    dashboard.update_line(
        lm.docs,
        f"{G}[x]{X} Documentation:        "
        f"{dc}{docs_result['total_present']}/{docs_result['total_required']} present{X}",
    )
    for fi in docs_result["files"]:
        idx = lm.docs_files.get(fi["name"])
        if idx is None:
            continue
        if fi["present"]:
            dashboard.update_line(idx, f"  {G}[x]{X} {fi['name']:<20}{G}[OK]{X}")
        elif fi.get("required", True):
            dashboard.update_line(idx, f"  {R}[!]{X} {fi['name']:<20}{R}[MISSING]{X}")
        else:
            dashboard.update_line(idx, f"  {Y}[-]{X} {fi['name']:<20}{Y}[OPTIONAL]{X}")
    dashboard.complete_task()
 
    # --- Check 4: Test suite ---
    dashboard.set_task("Test suite")
    dashboard.update_line(lm.tests, f"{Y}[>]{X} Tests:  running...")
    # FS-C3: read coverage threshold from .octrc.json health section.
    _health_cfg = _octrc_cfg.get("health") or {}
    _coverage_threshold_raw = _health_cfg.get("coverage_threshold")
    _coverage_threshold: Optional[float] = None
    if isinstance(_coverage_threshold_raw, (int, float)):
        _coverage_threshold = float(_coverage_threshold_raw)
    with check_timing("health.tests"):
        if run_tests is not False:
            test_result = check_test_health(
                project_root,
                timeout=effective_test_timeout,
                coverage_threshold=_coverage_threshold,
                production_mode=_production_mode,
            )
        else:
            test_result = {
                "status": "skipped",
                "reason": "run with --tests",
            }
    ts = test_result["status"]
    cov_pct = test_result.get("coverage_pct")
    cov_thr = test_result.get("coverage_threshold")
    cov_msg = test_result.get("coverage_message")
    cov_suffix = ""
    if cov_pct is not None and cov_thr is not None:
        cov_state = "OK" if cov_pct >= cov_thr else "below"
        cov_suffix = f" | cov {cov_pct:.1f}%/{cov_thr:.0f}% [{cov_state}]"
    elif cov_msg:
        cov_suffix = f" | {cov_msg}"
    test_summary = test_result.get("summary") or test_result.get("reason", "")
    tsm = f" ({test_summary}{cov_suffix})" if test_summary else cov_suffix
    if ts == "pass":
        dashboard.update_line(lm.tests, f"{G}[x]{X} Tests:                {G}PASS{X}{tsm}")
    elif ts == "skipped":
        dashboard.update_line(lm.tests, f"{Y}[-]{X} Tests:                {Y}SKIPPED{X}{tsm}")
    elif ts == "not_run":
        dashboard.update_line(lm.tests, f"{Y}[-]{X} Tests:                {Y}NOT RUN{X}{tsm}")
    else:
        dashboard.update_line(lm.tests, f"{R}[!]{X} Tests:                {R}{ts.upper()}{X}{tsm}")
    dashboard.complete_task()
 
    # --- Check 5: Compatibility ---
    dashboard.set_task("Compatibility check")
    dashboard.update_line(lm.compat, f"{Y}[>]{X} oc_diagnostics:  checking...")
    with check_timing("health.compat"):
        compat_result = check_compat_health()
    if compat_result["installed"]:
        if compat_result["compatible"]:
            dashboard.update_line(
                lm.compat,
                f"{G}[x]{X} oc_diagnostics:       v{compat_result['version']} -- {G}compatible{X}",
            )
        else:
            dashboard.update_line(
                lm.compat,
                f"{R}[!]{X} oc_diagnostics:       v{compat_result['version']} -- {R}INCOMPATIBLE{X}",
            )
    else:
        dashboard.update_line(lm.compat, f"{R}[!]{X} oc_diagnostics:       {R}NOT INSTALLED{X}")
    dashboard.complete_task()
 
    # --- Check 6: Virtual environment ---
    dashboard.set_task("Virtual environment")
    dashboard.update_line(lm.venv, f"{Y}[>]{X} Virtual Env:  checking...")
    with check_timing("health.venv"):
        venv_result = check_venv_health(project_root)
    if venv_result["status"] == "OK":
        dashboard.update_line(
            lm.venv, f"{G}[x]{X} Virtual Env:          {G}[OK]{X} {venv_result['message']}",
        )
    else:
        dashboard.update_line(
            lm.venv, f"{R}[!]{X} Virtual Env:          {Y}[WARN]{X} {venv_result['message']}",
        )
    dashboard.complete_task()
 
    # --- Check 7: Git integration ---
    dashboard.set_task("Git integration")
    dashboard.update_line(lm.git, f"{Y}[>]{X} Git Integration:  checking...")
    with check_timing("health.git"):
        git_result = check_git_health(project_root)
    _GC = {"OK": G, "WARN": Y, "FAIL": R, "SKIP": Y}
    git_ok = 0
    git_total = 0
    for gk, gl in _GIT_LABELS.items():
        idx = lm.git_checks.get(gk)
        check = git_result.get(gk)
        if idx is None or check is None:
            continue
        git_total += 1
        status = check["status"]
        color = _GC.get(status, X)
        if status == "OK":
            git_ok += 1
            mk = f"{G}[x]{X}"
        elif status in ("WARN", "SKIP"):
            mk = f"{Y}[-]{X}"
        else:
            mk = f"{R}[!]{X}"
        dashboard.update_line(idx, f"  {mk} {gl + ':':<20}{color}[{status}]{X} {check['detail']}")
    gc = G if git_ok == git_total else Y
    dashboard.update_line(lm.git, f"{G}[x]{X} Git Integration:      {gc}{git_ok}/{git_total} checks OK{X}")
    dashboard.complete_task()
 
    dashboard.finish()
 
    # Build report dict for JSON output
    report = {
        "project": project_root.name,
        "oct_version": __version__,
        "timestamp": datetime.now().isoformat(),
        "excluded_dirs": excluded_by_config,
        "disabled_rules": disabled_rules,
        "lint": lint_result,
        "fixable": fixable_result,
        "docs": docs_result,
        "tests": test_result,
        "compat": compat_result,
        "venv": venv_result,
        "git": git_result,
    }
 
    if json_mode:
        _print_json_report(report)
 
    if log_output:
        from oct.core.log_writer import write_command_log
        log_path = write_command_log(project_root, "health", report)
        print(f"Log written to: {log_path}", file=sys.stderr)
 
    if verbose and not json_mode and "files" in lint_result:
        print(f"\n{Y}{'-' * 70}{X}")
        print(f"{Y}Per-file breakdown:{X}")
        print(f"{Y}{'-' * 70}{X}\n")
        for file_info in lint_result["files"]:
            if not file_info["compliant"]:
                failed = [k for k, v in file_info["checks"].items() if not v]
                print(f"  {file_info['path']}")
                print(f"    Failed: {R}{', '.join(failed)}{X}")
 
    if verbose and not json_mode and fixable_result.get("files"):
        print(f"\n{Y}{'-' * 70}{X}")
        print(f"{Y}Fixable violations per file:{X}")
        print(f"{Y}{'-' * 70}{X}\n")
        for file_info in fixable_result["files"]:
            print(f"  {file_info['path']}")
            print(f"    Fixable: {Y}{', '.join(file_info['fixes'])}{X}")