oct_format.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/formatter/oct_format.py
 
"""
Option C auto-formatter implementation.
 
Purpose
-------
Detect and fix structural Option C compliance violations automatically.
Complements the linter by fixing violations while preserving code semantics.
 
Responsibilities
----------------
- Detect formatting violations using reused linter checks.
- Fix violations: headers, docstrings, imports, function patterns.
- Support report-only (dry-run) and fix modes.
- Archive originals before modification.
- Provide terminal and JSON output.
- Respect project exclusions and configuration.
 
Diagnostics
-----------
Domain: OCT.FORMATTER
Levels:
    L2 - formatter lifecycle and command routing
    L3 - file discovery and processing, fix application decisions
    L4 - detailed fixer implementations, archival operations, output formatting
 
Contracts
---------
- Files are only modified if fix_mode is True.
- Originals are archived before any modification.
- All code below headers is preserved exactly.
- Docstrings are not replaced; missing sections are appended.
- Each fixer is idempotent (running twice is same as once).
- A module is never given an import of a symbol its own package
  provides, so formatting can never make a package import itself.
 
Dependencies
------------
- oct (package version)
- oct.linter.oct_lint (reused validation checks)
- oct.core.exclusions (directory and wildcard exclusion lists)
- oct.core.project_root (project root detection)
- oct.core.parsing (safe AST parsing helper)
- oct.core.git (lazy; git changed-file discovery)
- oct.core.paths (to_project_relative - the identity header path)
- oct.core.diagnostics (_dbg structured logger)
- oct.core.octrc (lazy; .octrc.json exclusion and archive overrides)
- oct.core.workspace (lazy; subproject lookup for the per-subproject fan-out)
"""
 
import sys
import json
import argparse
import re
from dataclasses import dataclass
from pathlib import Path
from datetime import datetime
import ast
from typing import Tuple, List, Dict, Any, Optional
 
from oct import __version__
from oct.linter.oct_lint import (
    validate_header_block,
    check_docstring,
    check_diagnostics_profile,
    check_dbg_import,
    check_dbg_usage,
    find_func_pattern_violations,
    _extract_module_docstring,
    _module_docstring_slice,
    _describe_syntax_error,
    find_python_files,
    LinterContext,
)
from oct.core.exclusions import LINTER_EXCLUDE_DIRS, WILDCARD_EXCLUDE_DIRS
from oct.core.project_root import find_project_root
from oct.core.parsing import safe_parse
from oct.core.paths import to_project_relative
from oct.core.diagnostics import _dbg
 
 
def _git_changed_files(root: Path) -> list[Path]:
    """Return .py files modified in the git working tree (staged + unstaged).
 
    Thin shim over :func:`oct.core.git.git_changed_files` (Phase 4A).
    Collapses all ``GitError`` to empty list so the formatter never
    crashes due to git unavailability.
    """
    from oct.core import git as core_git
    try:
        return core_git.git_changed_files(root, extensions={".py"})
    except core_git.GitError:
        return []
 
# Template constants reused from generator
SHEBANG = "#!/usr/bin/env python3"
ENCODING = "# -*- coding: utf-8 -*-"
 
IMPORT_BLOCK = "from oc_diagnostics import _dbg\n"
 
DOCSTRING_SKELETON = '''\
"""
Purpose
-------
Describe the purpose of this module.
 
Responsibilities
----------------
- List the responsibilities of this module.
 
Diagnostics
-----------
Domain: <DOMAIN>
Levels:
    L2 - lifecycle
    L3 - semantic details
    L4 - deep tracing
 
Contracts
---------
- State the contracts and guarantees of this module.
"""
'''
 
MISSING_DOCSTRING_SECTIONS = {
    "Purpose": """
Purpose
-------
Describe the purpose of this module.""",
    "Responsibilities": """
Responsibilities
----------------
- List the responsibilities of this module.""",
    "Diagnostics": """
Diagnostics
-----------
Domain: <DOMAIN>
Levels:
    L2 - lifecycle
    L3 - semantic details
    L4 - deep tracing""",
    "Contracts": """
Contracts
---------
- State the contracts and guarantees of this module.""",
}
 
 
#: OI-424 - default cap on retained backups per file in
#: ``.formatter_archive/``. Projects can override via
#: ``formatter.max_backups_per_file`` in ``.octrc.json``.
_DEFAULT_MAX_BACKUPS_PER_FILE: int = 10
 
#: OI-516 - aggregate size cap per ``.formatter_archive/`` directory
#: (5 MiB). Enforced after per-file count pruning so a single giant
#: source file cannot fill the archive even within its 10-backup budget.
_DEFAULT_MAX_ARCHIVE_BYTES: int = 5 * 1024 * 1024
 
 
@dataclass
class FormatterContext:
    """Immutable per-run context for the formatter."""
    project_root: Path
    project_name: str
    diagnostics_dir: Path
    tests_dir: Path
    docs_dir: Path
    fix_mode: bool
    dry_run_mode: bool
    json_mode: bool
    verbose: bool
    max_backups_per_file: int = _DEFAULT_MAX_BACKUPS_PER_FILE
    #: OI-516 - aggregate cap across all backups in one
    #: ``.formatter_archive/`` directory.
    max_archive_bytes: int = _DEFAULT_MAX_ARCHIVE_BYTES
    #: Plan 277 - fixer names :func:`format_file` must NOT run on this
    #: file. Empty by default, so ``oct format`` run directly behaves
    #: exactly as it always has. The quality gate fills it per file from
    #: the octrc rules, the path rules and the file's own inline waivers,
    #: and does so in FIX mode too: before, a disabled fixer still ran
    #: and ``--fix`` wrote its change to disk, and only the gate's count
    #: of that change was suppressed afterwards.
    disabled_fixers: frozenset = frozenset()
 
 
# ============================================================================
# Helper Functions
# ============================================================================
 
 
def _get_relative_path(path: Path, root: Path) -> str:
    """Render *path* as a project-relative POSIX string.
 
    200: delegates to :func:`oct.core.paths.to_project_relative`, the helper
    OI-533 introduced to standardise path output - and the one the linter's
    ``expected_path_header`` already uses. This function previously returned
    ``str(path.relative_to(root))``, which renders with the OS separator, so
    on Windows the identity header the formatter WROTE
    (``# pkg\\sub\\mod.py``) was never the one the linter EXPECTED
    (``# pkg/sub/mod.py``). Every file in a subdirectory therefore had its
    header rewritten on every run, forever, each pass archiving the file
    again - a direct breach of the module's own per-fixer idempotence
    contract.
 
    There is one source of truth for "the project-relative path" and both
    tools now call it.
    """
    return to_project_relative(path, root)
 
 
def _prune_archive(
    archive_dir: Path, original_name: str, max_backups: int,
) -> int:
    """Keep only the ``max_backups`` most recent backups of ``original_name``.
 
    Matches files shaped like ``{original_name}.*.bak`` and sorts them
    by mtime (newest first). Older entries beyond the cap are removed.
    Best-effort: any OS error is swallowed so a failed prune can never
    block ``oct format`` (OI-424).
 
    Returns the number of files removed. ``max_backups <= 0`` disables
    pruning entirely (useful when a project deliberately wants unbounded
    history).
    """
    func = "_prune_archive"
    if max_backups <= 0:
        return 0
    try:
        candidates = sorted(
            archive_dir.glob(f"{original_name}.*.bak"),
            key=lambda p: p.stat().st_mtime,
            reverse=True,
        )
    except OSError:
        return 0
    removed = 0
    for stale in candidates[max_backups:]:
        try:
            stale.unlink()
            removed += 1
        except OSError:
            continue
    return removed
 
 
def _prune_archive_by_size(
    archive_dir: Path, max_total_bytes: int,
) -> int:
    """OI-516: evict oldest ``.bak`` files until total size fits under ``max_total_bytes``.
 
    Runs *after* per-file count pruning so the archive floor is always
    correct under the two coexisting policies. Any OS error on size or
    unlink is swallowed - a rotation failure must never block formatting.
    Returns the number of files removed.
    """
    func = "_prune_archive_by_size"
    if max_total_bytes <= 0:
        return 0
    try:
        entries = [(p, p.stat().st_mtime, p.stat().st_size)
                   for p in archive_dir.glob("*.bak") if p.is_file()]
    except OSError:
        return 0
    total = sum(size for _, _, size in entries)
    if total <= max_total_bytes:
        return 0
    # Oldest first - evict until we fit.
    entries.sort(key=lambda row: row[1])
    removed = 0
    for path, _, size in entries:
        if total <= max_total_bytes:
            break
        try:
            path.unlink()
            total -= size
            removed += 1
        except OSError:
            continue
    return removed
 
 
def _archive_file(
    path: Path,
    max_backups: int = _DEFAULT_MAX_BACKUPS_PER_FILE,
    max_archive_bytes: int = _DEFAULT_MAX_ARCHIVE_BYTES,
) -> None:
    """Archive original file to ``.formatter_archive/`` with timestamp.
 
    After writing the new backup, prune in two passes:
 
    1. Per-file count cap (OI-424): keep only the N newest backups per
       source file name.
    2. Per-directory size cap (OI-516): enforce aggregate ``max_archive_bytes``
       across *all* backups in the archive, evicting oldest first.
    """
    func = "archive_file"
    archive_dir = path.parent / ".formatter_archive"
    archive_dir.mkdir(exist_ok=True)
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    backup_path = archive_dir / f"{path.name}.{timestamp}.bak"
    backup_path.write_bytes(path.read_bytes())
    _prune_archive(archive_dir, path.name, max_backups)
    _prune_archive_by_size(archive_dir, max_archive_bytes)
 
 
def _leading_comment_block_end(lines: List[str]) -> int:
    """Return the 0-based index just past the leading comment/blank block.
 
    Scans from the top while every line is blank or a comment, however long
    the block runs, so an inserted construct never lands inside it. Shared
    by the import and docstring fixers so both follow the same rule (186,
    generalised by 198).
    """
    index = 0
    for i, line in enumerate(lines):
        stripped = line.strip()
        if stripped and not stripped.startswith("#"):
            break
        index = i + 1
    return index
 
 
def _leading_whitespace(line: str) -> str:
    """Return the indentation prefix of ``line``, tabs and spaces alike."""
    return line[:len(line) - len(line.lstrip())]
 
 
def _statement_start_index(stmt: ast.stmt) -> int:
    """Return the 0-based index of the first source line owned by ``stmt``.
 
    A decorated statement's own ``lineno`` points at its ``def``/``class``
    keyword, not at its first decorator. Inserting a line at that index
    would separate a decorator from the statement it decorates, which does
    not parse - so decorators are folded in here (198).
    """
    lineno = stmt.lineno
    for decorator in getattr(stmt, "decorator_list", []):
        lineno = min(lineno, decorator.lineno)
    return lineno - 1
 
 
def _begins_its_own_line(stmt: ast.stmt, lines: List[str]) -> bool:
    """Return True when nothing but whitespace precedes ``stmt`` on its line.
 
    Distinguishes a body that occupies its own lines from one written inline
    with its signature - ``def f(): return x``, or a multi-line signature
    closed as ``) -> int: return x``. An inline body has no line of its own
    to insert before (198).
    """
    index = stmt.lineno - 1
    if index >= len(lines):
        return False
    return not lines[index][:stmt.col_offset].strip()
 
 
def _module_docstring_span(text: str) -> Optional[Tuple[str, int]]:
    """Locate the module docstring, returning ``(docstring, close_index)``.
 
    ``close_index`` is the offset in *text* of the docstring's CLOSING
    triple quote, which is where :func:`fix_docstring` appends the Option C
    sections it finds missing.
 
    200: the docstring itself comes from the linter's shared
    :func:`~oct.linter.oct_lint._extract_module_docstring` - the helper this
    module already uses at :func:`_detect_missing_docstring_sections` and at
    the top of :func:`fix_docstring`. Only the POSITION is computed here.
    That is the whole point: ``fix_docstring`` used to find its insertion
    point with a raw scan for the first triple-double-quote in the file,
    which is a second, independent opinion about which string is the module
    docstring. With a single-quoted module docstring the raw scan found a
    FUNCTION's docstring and the module sections were appended there. The
    output still parsed, so nothing caught it.
 
    204: the position is no longer recovered by searching the source for the
    shared helper's answer - the helper now reports the span it found, so the
    formatter and the linter agree on WHERE the docstring is structurally,
    not by locating the same substring twice.
 
    Returns ``None`` when there is no module docstring, or when its delimiter
    is not a triple quote: the Option C sections are appended as new lines,
    which a one-line ``'x'`` literal cannot carry without breaking the parse.
    Callers must treat ``None`` as "make no change": misplacing (or
    duplicating) Option C sections is worse than declining to add them.
    """
    func = "_module_docstring_span"
 
    located = _module_docstring_slice(text)
    if located is None or len(located.quote) != 3:
        return None
 
    return located.text, located.close_index
 
 
def _detect_missing_docstring_sections(text: str) -> List[str]:
    """Detect which docstring sections are missing."""
    func = "_detect_missing_docstring_sections"
    docstring = _extract_module_docstring(text)
    if not docstring:
        return ["Purpose", "Responsibilities", "Diagnostics", "Contracts"]
 
    missing = []
    lines = docstring.split('\n')
    for section in ["Purpose", "Responsibilities", "Diagnostics", "Contracts"]:
        found = any(line.strip().startswith(section) for line in lines)
        if not found:
            missing.append(section)
    return missing
 
 
# ============================================================================
# Fix Functions
# ============================================================================
 
 
def fix_header_block(path: Path, text: str, ctx: FormatterContext) -> Tuple[bool, str, List[str]]:
    """
    Ensure file has correct 4-line header block.
 
    Returns (changed, new_text, messages). Caller is responsible for
    archiving and writing to disk.
    """
    func = "fix_header_block"
 
    # Validate using linter
    header_lines = text.split('\n')[:4] if len(text.split('\n')) >= 4 else text.split('\n')
    ok, errors = validate_header_block(header_lines, path, ctx)
 
    if ok:
        return False, text, []
 
    # Reconstruct correct header
    relative_path = _get_relative_path(path, ctx.project_root)
    new_shebang = f"{SHEBANG}"
    new_encoding = f"{ENCODING}"
    new_identity = f"# {relative_path}"
    new_blank = ""
 
    new_header = new_shebang + "\n" + new_encoding + "\n" + new_identity + "\n" + new_blank + "\n"
 
    # Find where the actual file content starts (after header block).
    # OI-509: delegate to shared oct.core.parsing.header_boundary so the
    # formatter and linter use identical scanning rules.
    from oct.core.parsing import header_boundary
    lines = text.split('\n')
    relative_identity = f"# {relative_path}"
    content_start = header_boundary(
        lines, path, expected_identity=relative_identity,
    )
 
    # Take content after the old header
    if content_start < len(lines):
        remaining_content = '\n'.join(lines[content_start:])
    else:
        remaining_content = ""
 
    new_text = new_header + remaining_content
 
    messages = ["Fixed header block (shebang, encoding, file identity, blank line)"]
    return True, new_text, messages
 
 
def fix_docstring(path: Path, text: str, ctx: FormatterContext) -> Tuple[bool, str, List[str]]:
    """
    Ensure module has complete docstring with all required sections.
 
    If no docstring: insert skeleton.
    If incomplete: add missing sections.
    If complete: no change.
    """
    func = "fix_docstring"
 
    docstring = _extract_module_docstring(text)
 
    if not docstring:
        # 204: "the authority cannot address a docstring here" is not the
        # same statement as "this module has no docstring", and only the
        # second one licenses injecting a skeleton. A module docstring that
        # is not expressible as one source slice - implicitly concatenated
        # literals - would otherwise be answered with a SECOND docstring
        # above it, which parses, so nothing downstream would catch it.
        # Unparseable text is declined for the same reason ``format_file``
        # refuses to format it at all (175/F21): never rewrite what we
        # cannot read.
        tree, _ = safe_parse(text)
        if tree is None or ast.get_docstring(tree) is not None:
            return False, text, []
 
        # 198: the skeleton goes after the leading comment block, wherever
        # it actually ends. The constant 4 was the same hardcoded-offset
        # assumption 186 removed from the import fixer; it happened to be
        # right only because ``fix_header_block`` runs first in the
        # pipeline and normalises the header to exactly four lines. Called
        # on its own, or on a file carrying waiver comments above the code,
        # it split the comment block instead.
        lines = text.split('\n')
        insert_pos = _leading_comment_block_end(lines)
 
        new_text = '\n'.join(lines[:insert_pos]) + '\n' + DOCSTRING_SKELETON + '\n'.join(lines[insert_pos:])
 
        return True, new_text, ["Added complete module docstring"]
 
    # Check for missing sections
    missing = _detect_missing_docstring_sections(text)
    if not missing:
        return False, text, []
 
    # Add missing sections to existing docstring
    sections_to_add = "\n".join(MISSING_DOCSTRING_SECTIONS[section] for section in missing)
 
    # Insert before the MODULE docstring's closing quotes. 200: the position
    # is derived from the shared extractor's own answer rather than from an
    # independent scan, so this fixer cannot append module sections to some
    # other string in the file. If the docstring cannot be located, decline
    # to change anything - a misplaced section is worse than a missing one.
    span = _module_docstring_span(text)
    if span is None:
        return False, text, []
    _docstring, docstring_end = span
 
    new_text = (text[:docstring_end] + "\n" + sections_to_add + '\n' + text[docstring_end:])
 
    messages = [f"Added missing docstring sections: {', '.join(missing)}"]
    return True, new_text, messages
 
 
def _dbg_import_insert_index(tree: ast.Module, lines: List[str]) -> int:
    """
    Return the 0-based line index at which the canonical ``_dbg`` import belongs.
 
    The placement is derived from the module structure, never from a fixed
    line number (184-D7). The Option C canonical order is: header comments,
    module docstring, ``from __future__`` imports, then ordinary imports.
 
    - With a module docstring, start at the line after its closing quotes,
      then move past any ``from __future__`` imports - the language requires
      those to be the first statement after the docstring.
    - Without one, start after the whole leading comment/blank block, however
      long it runs, so the import never lands inside it.
 
    In both cases trailing blank lines are skipped, so the import joins the
    top of the existing import block instead of hugging the docstring or the
    comment header.
    """
    body = tree.body
    index = 0
    consumed = 0
 
    if (body and isinstance(body[0], ast.Expr)
            and isinstance(body[0].value, ast.Constant)
            and isinstance(body[0].value.value, str)):
        # ``end_lineno`` is 1-based, so it already *is* the 0-based index of
        # the line following the docstring's closing quotes.
        index = body[0].end_lineno
        consumed = 1
    else:
        index = _leading_comment_block_end(lines)
 
    for node in body[consumed:]:
        if isinstance(node, ast.ImportFrom) and node.module == "__future__":
            index = max(index, node.end_lineno)
        else:
            break
 
    while index < len(lines) and not lines[index].strip():
        index += 1
 
    return min(index, len(lines))
 
 
def _import_block_package() -> Optional[str]:
    """
    Return the top-level package named by :data:`IMPORT_BLOCK`.
 
    Derived by parsing the canonical import rather than hardcoded, so the
    self-import guard keeps following ``IMPORT_BLOCK`` if the package that
    provides the diagnostics symbol is ever renamed. Returns ``None`` if
    the constant is not a parseable ``from <package> import ...``.
    """
    tree, _ = safe_parse(IMPORT_BLOCK)
    if tree is None:
        return None
 
    for node in ast.iter_child_nodes(tree):
        if isinstance(node, ast.ImportFrom) and node.module:
            return node.module.split(".")[0]
 
    return None
 
 
def _import_block_binding() -> Tuple[Optional[str], Optional[str]]:
    """
    Return ``(module, symbol)`` named by :data:`IMPORT_BLOCK`.
 
    Derived by parsing the canonical import rather than hardcoded, so the
    removal pass keeps following ``IMPORT_BLOCK`` if either the module or
    the symbol it provides is ever renamed. Returns ``(None, None)`` if the
    constant is not a parseable ``from <module> import <symbol>``.
    """
    tree, _ = safe_parse(IMPORT_BLOCK)
    if tree is None:
        return None, None
 
    for node in ast.iter_child_nodes(tree):
        if isinstance(node, ast.ImportFrom) and node.module and node.names:
            return node.module, node.names[0].name
 
    return None, None
 
 
def _conflicting_import_lines(
    tree: ast.Module, module: str, symbol: str,
) -> Tuple[set, int]:
    """
    Return ``(line_indices, statement_count)`` for imports that conflict.
 
    A statement conflicts only when it binds the BARE name ``symbol``, since
    that is the sole binding the canonical import can collide with. This is
    a structural test over module-level import nodes, replacing a substring
    scan that treated any line containing ``"import <symbol>"`` as an import
    (198). That scan deleted prose inside docstrings, string literals that
    merely spelled an import, and - because ``import _dbg_shim`` contains
    ``import _dbg`` - the sole body of a ``try:`` block, which does not
    parse.
 
    Two conservative limits keep removal from causing the very breakage this
    fixer exists to avoid:
 
    - Aliased imports are never removed. They bind a different name, so they
      cannot conflict, and deleting one unbinds a name the module still uses.
    - A statement that binds anything else is never removed, so a
      co-imported symbol can never be taken down with it.
 
    Only module-level statements are considered, matching the scope of
    :func:`oct.linter.oct_lint.check_dbg_import`.
    """
    doomed: set = set()
    statements = 0
 
    for node in ast.iter_child_nodes(tree):
        if not isinstance(node, (ast.Import, ast.ImportFrom)):
            continue
 
        binds_symbol = any(
            alias.asname is None and alias.name.split(".")[0] == symbol
            for alias in node.names
        )
        if not binds_symbol:
            continue
 
        if isinstance(node, ast.ImportFrom) and node.module == module:
            continue
 
        if len(node.names) > 1:
            continue
 
        end_lineno = node.end_lineno or node.lineno
        doomed.update(range(node.lineno - 1, end_lineno))
        statements += 1
 
    return doomed, statements
 
 
def _owning_top_level_package(path: Path) -> Optional[str]:
    """
    Return the name of the top-level package ``path`` belongs to.
 
    Walks up from the file's directory while each level is a real package
    (contains an ``__init__.py``); the highest such directory names the
    top-level package. A nested submodule therefore reports the package at
    the top of its chain, not its immediate parent.
 
    Returns ``None`` when the file is not inside a package at all, which
    keeps the test structural: a directory that merely *shares* a
    package's name is not that package.
    """
    package = None
    directory = path.parent
 
    while (directory / "__init__.py").exists():
        package = directory.name
        parent = directory.parent
        if parent == directory:
            break
        directory = parent
 
    return package
 
 
def fix_dbg_import(path: Path, text: str, ctx: FormatterContext) -> Tuple[bool, str, List[str]]:
    """
    Ensure canonical _dbg import exists at module level.
 
    Skips trivial files (< 20 lines), and skips any module belonging to the
    package that *provides* the symbol (195) - see the guard below.
    """
    func = "fix_dbg_import"
 
    # Skip trivial files
    if len(text.splitlines()) < 20:
        return False, text, []
 
    # 195: a package must never be made to import itself. Inserting the
    # canonical import into a module of the package that DEFINES the symbol
    # is circular at module scope - the owning package's ``__init__``
    # re-exports from that very module, so the import runs against a
    # half-initialised package and raises ImportError, taking the whole
    # package down. The test is structural (walk up while ``__init__.py``
    # exists) and the package name comes from IMPORT_BLOCK, so this is the
    # general rule "a package does not import a symbol it provides" rather
    # than a special case for one package name.
    owning_package = _owning_top_level_package(path)
    if owning_package is not None and owning_package == _import_block_package():
        return False, text, [
            f"Skipped canonical import: this module belongs to "
            f"'{owning_package}', the package that provides it"
        ]
 
    # Check if canonical import exists
    ok, msg = check_dbg_import(text)
    if ok:
        return False, text, []
 
    # Parse AST to find import insertion point
    tree, _ = safe_parse(text)
    if tree is None:
        return False, text, []
 
    lines = text.split('\n')
 
    # Derive the structural insertion point from the AST before any lines are
    # dropped, so the index refers to the file as parsed.
    insert_pos = _dbg_import_insert_index(tree, lines)
 
    # Find any conflicting _dbg imports and remove them. The set is derived
    # from the AST, so only real import statements are ever deleted (198).
    canonical_module, canonical_symbol = _import_block_binding()
    if canonical_module is None or canonical_symbol is None:
        return False, text, []
 
    doomed, removed_count = _conflicting_import_lines(
        tree, canonical_module, canonical_symbol,
    )
 
    new_lines = [line for i, line in enumerate(lines) if i not in doomed]
    removed_before_insert = sum(1 for i in doomed if i < insert_pos)
 
    # Removal shifts every line below it up, so rebase the index onto the
    # surviving lines rather than recomputing it from a mutated file.
    insert_pos = min(insert_pos - removed_before_insert, len(new_lines))
    new_lines.insert(insert_pos, IMPORT_BLOCK.rstrip())
    new_text = '\n'.join(new_lines)
 
    msg = "Added canonical _dbg import"
    if removed_count > 0:
        msg += f" (removed {removed_count} incorrect import(s))"
 
    return True, new_text, [msg]
 
 
def _func_pattern_insert_point(
    node: Any, lines: List[str],
) -> Optional[Tuple[int, str]]:
    """
    Return ``(line_index, indent)`` for a function's ``func = "..."`` line.
 
    The position is derived from the function's AST body, never by counting
    lines from the ``def`` (198). A multi-line signature puts a PARAMETER on
    the line after the ``def``, so "the next line" is not the body - taking
    it as the insertion point injects the assignment into the parameter list
    and the file stops parsing.
 
    Docstring-first ordering is preserved: when the body opens with a
    docstring the assignment lands after it, so the string remains the
    function's docstring instead of being demoted to a plain expression.
 
    Returns ``None`` when there is no line to insert before - a body written
    inline with its signature (``def f(): return x``). Leaving such a
    function unfixed is the correct outcome: the fixer's contract is that it
    never corrupts a file, not that it always succeeds.
    """
    body = node.body
    if not body:
        return None
 
    first = body[0]
    if not _begins_its_own_line(first, lines):
        return None
 
    target = first
    if (isinstance(first, ast.Expr)
            and isinstance(first.value, ast.Constant)
            and isinstance(first.value.value, str)):
        if len(body) > 1 and _begins_its_own_line(body[1], lines):
            target = body[1]
        else:
            # Nothing below the docstring to anchor to, so land on the line
            # after its closing quotes. ``end_lineno`` is 1-based, so it
            # already is the 0-based index of the following line.
            index = min(first.end_lineno, len(lines))
            return index, _leading_whitespace(lines[first.lineno - 1])
 
    index = _statement_start_index(target)
    if index >= len(lines):
        return None
    return index, _leading_whitespace(lines[index])
 
 
def fix_func_pattern(path: Path, text: str, ctx: FormatterContext) -> Tuple[bool, str, List[str]]:
    """
    Ensure every function with _dbg() calls has func = "function_name" as first statement.
 
    Delegates violation detection to :func:`oct.linter.oct_lint.find_func_pattern_violations`
    so the linter remains the single source of truth for what constitutes a
    func-pattern violation.  This function only handles the fix (inserting
    the ``func = "..."`` line), and takes every insertion point from the AST
    rather than from line arithmetic (198).
 
    Skips trivial files (< 20 lines).
    """
    func = "fix_func_pattern"
 
    tree, _ = safe_parse(text)
    if tree is None:
        return False, text, []
 
    functions_to_fix = find_func_pattern_violations(text, tree)
    if not functions_to_fix:
        return False, text, []
 
    lines = text.split('\n')
 
    nodes = {
        (node.name, node.lineno): node
        for node in ast.walk(tree)
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
    }
 
    # Resolve every insertion point against the ORIGINAL line list first.
    # Indices derived from the AST are only valid for the text that AST was
    # parsed from, so nothing may be mutated until all of them are known.
    insertions: List[Tuple[int, str, str]] = []
    skipped: List[str] = []
 
    for func_name, func_lineno in functions_to_fix:
        node = nodes.get((func_name, func_lineno))
        if node is None:
            continue
        point = _func_pattern_insert_point(node, lines)
        if point is None:
            skipped.append(func_name)
            continue
        insert_idx, indent = point
        insertions.append((insert_idx, func_name, indent))
 
    if not insertions:
        messages = []
        if skipped:
            messages.append(
                "Skipped func pattern (body shares a line with its "
                f"signature): {', '.join(skipped)}"
            )
        return False, text, messages
 
    # Apply bottom-up so an earlier insertion can never shift a later index.
    for insert_idx, func_name, indent in sorted(insertions, reverse=True):
        lines.insert(insert_idx, f'{indent}func = "{func_name}"')
 
    new_text = '\n'.join(lines)
 
    fixed_names = [name for _idx, name, _indent in insertions]
    messages = [f"Added func pattern to: {', '.join(fixed_names)}"]
    if skipped:
        messages.append(
            "Skipped func pattern (body shares a line with its "
            f"signature): {', '.join(skipped)}"
        )
    return True, new_text, messages
 
 
# ============================================================================
# File Processing
# ============================================================================
 
 
def format_file(path: Path, ctx: FormatterContext) -> Dict[str, Any]:
    """Process a single file through all formatters.
 
    Plan 277: a fixer named in ``ctx.disabled_fixers`` does not run. Its
    entry in ``fixes`` still appears, reporting ``changed`` False and
    carrying an extra ``"skipped": "disabled"`` key, so the result stays
    a superset of the shape every existing reader expects. With the
    default empty set - ``oct format`` run directly, and every caller
    that builds its own context - nothing about this function changes.
    """
    func = "format_file"
 
    try:
        text = path.read_text(encoding="utf-8")
    except Exception as e:
        return {
            "path": _get_relative_path(path, ctx.project_root),
            "changed": False,
            "fixes": {
                "header": {"changed": False, "messages": []},
                "docstring": {"changed": False, "messages": []},
                "dbg_import": {"changed": False, "messages": []},
                "func_pattern": {"changed": False, "messages": []},
            },
            "errors": [f"Could not read file: {str(e)}"],
        }
 
    result = {
        "path": _get_relative_path(path, ctx.project_root),
        "changed": False,
        "fixes": {},
        "errors": [],
    }
 
    # F21 / Plan 175 (item 2): an unparseable file cannot be safely
    # formatted. The AST-dependent fixers used to swallow ``tree is None``
    # into ``return False, text, []`` - a silent "no changes" that is
    # indistinguishable from a genuinely clean file - while the text-based
    # fixers (docstring/header) would actively REWRITE the broken text.
    # Surface the SyntaxError through the same error channel a read failure
    # or exception uses, apply no fixes, and never rewrite text we cannot
    # parse. Mirrors the linter's syntax_error gate (Plan 170).
    tree, _ = safe_parse(text)
    if tree is None:
        result["errors"].append(f"Syntax error: {_describe_syntax_error(text)}")
        result["fixes"] = {
            fix_name: {"changed": False, "messages": []}
            for fix_name in ("header", "docstring", "dbg_import", "func_pattern")
        }
        return result
 
    # Apply fixers in sequence
    fixers = [
        ("header", fix_header_block),
        ("docstring", fix_docstring),
        ("dbg_import", fix_dbg_import),
        ("func_pattern", fix_func_pattern),
    ]
 
    any_changed = False
 
    for fix_name, fixer in fixers:
        # Plan 277: skip BEFORE the call, not after it. The gate used to
        # filter a disabled fixer out of its violation count once the
        # fixer had already rewritten ``text``, which is invisible in
        # check mode and destructive in fix mode.
        if fix_name in ctx.disabled_fixers:
            result["fixes"][fix_name] = {
                "changed": False, "messages": [], "skipped": "disabled",
            }
            _dbg("OCT.FORMATTER", func, f"{fix_name} disabled for {path.name}", 4)
            continue
        try:
            changed, text, messages = fixer(path, text, ctx)
            result["fixes"][fix_name] = {"changed": changed, "messages": messages}
            if changed:
                any_changed = True
            # One L3 record per fixer, so a debug log says which fixers ran
            # and what each found (the fixers log nothing themselves; found
            # by the chart C runtime trace, 2026-09-27).
            outcome = ("would change" if not ctx.fix_mode else "changed") if changed else "clean"
            _dbg("OCT.FORMATTER", func,
                 f"fixer {fix_name} {outcome} ({len(messages)} messages) {result['path']}", 3)
        except Exception as e:
            result["errors"].append(f"{fix_name}: {str(e)}")
            _dbg("OCT.FORMATTER", func,
                 f"fixer {fix_name} raised {type(e).__name__} {result['path']}", 3)
 
    # 198: the closing invariant. Every member of the insertion-point family
    # (186, 195, 198) had the same consequence - text that no longer parses.
    # The individual fixers are now structural, but this backstop removes the
    # CLASS rather than three instances of it, and it covers fixers that do
    # not exist yet. The input was already proven parseable by the gate
    # above, so a failure here is unambiguously the formatter's doing:
    # discard the whole result and keep the file the operator had.
    if any_changed:
        verified, _ = safe_parse(text)
        if verified is None:
            result["errors"].append(
                "Discarded formatter output: the result does not parse "
                f"({_describe_syntax_error(text)}). The file was left "
                "unchanged."
            )
            result["fixes"] = {
                fix_name: {"changed": False, "messages": []}
                for fix_name, _fixer in fixers
            }
            result["changed"] = False
            return result
 
    # Single archive + write at end
    if any_changed and ctx.fix_mode and not ctx.dry_run_mode:
        _archive_file(
            path,
            max_backups=ctx.max_backups_per_file,
            max_archive_bytes=ctx.max_archive_bytes,
        )
        path.write_text(text, encoding="utf-8")
 
    result["changed"] = any_changed
    return result
 
 
def compute_summary(file_results: List[Dict[str, Any]]) -> Dict[str, Any]:
    """Aggregate results into summary."""
    func = "compute_summary"
    _dbg("OCT.FORMATTER", func, f"summarising {len(file_results)} formatted files", 3)
 
    summary = {
        "total_files": len(file_results),
        "total_changed": sum(1 for r in file_results if r.get("changed")),
        "fixes": {
            "header": 0,
            "docstring": 0,
            "dbg_import": 0,
            "func_pattern": 0,
        },
    }
 
    for result in file_results:
        for fix_type in summary["fixes"]:
            if result["fixes"].get(fix_type, {}).get("changed"):
                summary["fixes"][fix_type] += 1
 
    return summary
 
 
# ============================================================================
# Output Formatting
# ============================================================================
 
 
def _print_terminal_report(summary: Dict[str, Any], file_results: List[Dict[str, Any]],
                           verbose: bool, dry_run: bool) -> None:
    """Print human-readable terminal report."""
    func = "_print_terminal_report"
 
    mode = "DRY-RUN" if dry_run else "FIXED"
 
    print(f"\n{'=' * 70}")
    print(f"OCT FORMATTER REPORT - {mode}")
    print(f"{'=' * 70}\n")
 
    print(f"Files processed:      {summary['total_files']}")
    print(f"Files with changes:   {summary['total_changed']}")
    print()
    print("Fixes applied:")
    print(f"  Header blocks:      {summary['fixes']['header']}")
    print(f"  Docstrings:         {summary['fixes']['docstring']}")
    print(f"  _dbg imports:       {summary['fixes']['dbg_import']}")
    print(f"  func patterns:      {summary['fixes']['func_pattern']}")
 
    if verbose:
        print(f"\n{'-' * 70}")
        print("Detailed results:")
        print(f"{'-' * 70}\n")
 
        for result in file_results:
            if result["changed"] or result["errors"]:
                print(f"{result['path']}")
 
                for fix_type, fix_result in result["fixes"].items():
                    if fix_result["changed"]:
                        print(f"  [OK] {fix_type}: {', '.join(fix_result['messages'])}")
 
                for error in result["errors"]:
                    print(f"  [ERROR] {error}")
 
                print()
 
    print(f"{'=' * 70}\n")
 
 
def _print_json_report(summary: Dict[str, Any], file_results: List[Dict[str, Any]],
                       ctx: FormatterContext) -> None:
    """Print JSON output."""
    func = "_print_json_report"
 
    output = {
        "project": ctx.project_name,
        "oct_version": __version__,
        "timestamp": datetime.now().isoformat(),
        "dry_run": ctx.dry_run_mode,
        "summary": summary,
        "files": file_results,
    }
 
    print(json.dumps(output, indent=2))
 
 
# ============================================================================
# Main Entry Point
# ============================================================================
 
 
def _formatter_flag_argv(args) -> List[str]:
    """Rebuild the formatter's argv from parsed *args*.
 
    Plan 213 / D-D. ``--changed`` is deliberately NOT reproduced: the
    dispatcher has already resolved the changed set at the workspace root
    and hands each lens group its own explicit file list.
    """
    flags: List[str] = []
    if args.fix:
        flags.append("--fix")
    if args.dry_run:
        flags.append("--dry-run")
    if args.json:
        flags.append("--json")
    if args.verbose:
        flags.append("--verbose")
    return flags
 
 
def _maybe_dispatch_format_by_lens(project_root: Path, args) -> bool:
    """Fan ``oct format`` out to one engine run per owning subproject.
 
    Plan 213 / D-D, the formatter half. Returns True when the invocation
    fanned out (the caller then returns immediately), False to let the
    ordinary single-root pass run.
 
    This matters more here than for the linter: ``run_formatter`` has no
    path targets at all, so before 213 a workspace-root invocation walked
    the entire tree under ONE ``FormatterContext`` and judged every
    subproject's file against the workspace-root lens. In ``--fix`` mode
    that did not merely misreport - it REWROTE correct
    subproject-relative identity headers into workspace-relative ones.
 
    Only a workspace-root invocation fans out. From inside a subproject,
    project-root discovery already resolves to that subproject, so the
    lens is correct and behaviour is byte-identical to before.
    """
    try:
        from oct.core.workspace import (
            assign_to_subproject,
            find_workspace_root,
            load_workspace,
        )
    except ImportError:
        return False
 
    ws_path = find_workspace_root(project_root)
    if ws_path is None or ws_path.resolve() != project_root.resolve():
        return False
    try:
        workspace = load_workspace(ws_path)
    except Exception:
        return False
    if not workspace.subprojects:
        return False
 
    exclude_dirs = set(LINTER_EXCLUDE_DIRS)
    wildcard_exclude = list(WILDCARD_EXCLUDE_DIRS)
    flags = _formatter_flag_argv(args)
 
    def _owned(path: Path) -> bool:
        return assign_to_subproject(path, workspace) is not None
 
    if args.changed:
        changed = [
            p for p in _git_changed_files(workspace.root)
            if p.is_file() and p.suffix == ".py"
        ]
        if not changed:
            if not args.json:
                print("No modified Python files found.", file=sys.stderr)
            return True
        groups: dict[Path, List[Path]] = {}
        for path in changed:
            subproject = assign_to_subproject(path, workspace)
            root = (
                subproject.path if subproject is not None else workspace.root
            )
            groups.setdefault(root, []).append(path)
        for group_root, group_files in groups.items():
            run_formatter(
                group_root, list(flags),
                _files=group_files, _dispatch=False,
            )
        return True
 
    # Full sweep: one pass per subproject under its own lens, then the
    # workspace-root leftovers under the workspace lens.
    for subproject in workspace.subprojects:
        if not subproject.path.is_dir():
            continue
        run_formatter(subproject.path, list(flags), _dispatch=False)
 
    leftovers = [
        path
        for path in find_python_files(
            workspace.root, exclude_dirs, wildcard_exclude,
        )
        if not _owned(path)
    ]
    if leftovers:
        run_formatter(
            workspace.root, list(flags),
            _files=leftovers, _dispatch=False,
        )
    return True
 
 
def run_formatter(
    project_root: Path,
    argv: Optional[List[str]] = None,
    *,
    _files: Optional[List[Path]] = None,
    _dispatch: bool = True,
) -> None:
    """
    Run the Option C formatter on the given project root.
 
    Parameters
    ----------
    project_root : Path
        The detected Option C project root directory.
    argv : list[str] | None
        Optional list of command-line arguments.
        If None, argparse will use sys.argv[1:].
    _files : list[Path] | None
        Plan 213 internal. An explicit file list that overrides scope
        discovery, used by the workspace lens dispatcher to hand a lens
        group exactly the files it owns.
    _dispatch : bool
        Plan 213 internal re-entrancy guard. The lens dispatcher re-invokes
        this function once per owning subproject with ``_dispatch=False``
        so the per-group runs cannot fan out again.
    """
    func = "run_formatter"
    _dbg("OCT.FORMATTER", func, f"format run starting at {project_root}", 3)
 
    # Parse arguments
    parser = argparse.ArgumentParser(
        description="Auto-fix Option C compliance violations",
        prog="oct format",
    )
    parser.add_argument(
        "--fix",
        action="store_true",
        help="Apply formatting changes. Without this, only reports what would change.",
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Report what would change without modifying files (default behavior).",
    )
    parser.add_argument(
        "--json",
        action="store_true",
        help="Output machine-readable JSON instead of colorized terminal report.",
    )
    parser.add_argument(
        "--verbose",
        action="store_true",
        help="Show per-file details in addition to summary.",
    )
    parser.add_argument(
        "--changed",
        action="store_true",
        help="Format only git-modified Python files.",
    )
 
    args = parser.parse_args(argv)
 
    # ---- Plan 213 / D-D: dispatch each file to its owning lens ----
    if _dispatch and _files is None and _maybe_dispatch_format_by_lens(
        project_root, args,
    ):
        return
 
    # Create context
    diagnostics_dir = project_root / "oc_diagnostics"
    tests_dir = project_root / "tests"
    docs_dir = project_root / "docs"
 
    # Find and process all Python files
    exclude_dirs = set(LINTER_EXCLUDE_DIRS)
    wildcard_exclude = list(WILDCARD_EXCLUDE_DIRS)
 
    # Load per-project overrides from .octrc.json (same as linter).
    # OI-519: routed through central oct.core.octrc loader.
    max_backups_per_file = _DEFAULT_MAX_BACKUPS_PER_FILE
    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)
        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)
    # OI-424: optional formatter.max_backups_per_file override.
    # OI-516: optional formatter.max_archive_mib override - written as MiB
    # in .octrc.json (readable) and converted to bytes on load.
    max_archive_bytes = _DEFAULT_MAX_ARCHIVE_BYTES
    formatter_cfg = octrc.get("formatter", {})
    if isinstance(formatter_cfg, dict):
        raw = formatter_cfg.get("max_backups_per_file")
        if isinstance(raw, int) and raw >= 0:
            max_backups_per_file = raw
        raw_mib = formatter_cfg.get("max_archive_mib")
        if isinstance(raw_mib, (int, float)) and raw_mib >= 0:
            max_archive_bytes = int(raw_mib * 1024 * 1024)
 
    ctx = FormatterContext(
        project_root=project_root,
        project_name=project_root.name,
        diagnostics_dir=diagnostics_dir,
        tests_dir=tests_dir,
        docs_dir=docs_dir,
        fix_mode=args.fix,
        dry_run_mode=not args.fix,
        json_mode=args.json,
        verbose=args.verbose,
        max_backups_per_file=max_backups_per_file,
        max_archive_bytes=max_archive_bytes,
    )
 
    if _files is not None:
        # Plan 213: an explicit lens-group file list from the dispatcher.
        python_files = list(_files)
    elif args.changed:
        from oct.linter.oct_lint import is_excluded_dir
        python_files = _git_changed_files(project_root)
        # Plan 282: the exclusion is asked of the DIRECTORY parts only, the
        # same slice ``find_python_files`` takes, so a module named like an
        # excluded directory is formatted under ``--changed`` exactly as it
        # is under a full scan.
        python_files = [
            p for p in python_files
            if p.is_file() and not any(
                is_excluded_dir(Path(part), exclude_dirs, wildcard_exclude)
                for part in p.relative_to(project_root).parts[:-1]
            )
        ]
        if not python_files:
            if not args.json:
                print("No modified Python files found.", file=sys.stderr)
            return
    else:
        python_files = find_python_files(project_root, exclude_dirs, wildcard_exclude)
 
    file_results = []
    for py_file in python_files:
        result = format_file(py_file, ctx)
        file_results.append(result)
 
    # Compute summary
    summary = compute_summary(file_results)
 
    # Output report
    if ctx.json_mode:
        _print_json_report(summary, file_results, ctx)
    else:
        _print_terminal_report(summary, file_results, ctx.verbose, ctx.dry_run_mode)
 
 
if __name__ == "__main__":
    project_root = find_project_root(Path.cwd())
    run_formatter(project_root)