oct_git.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/git/oct_git.py
 
"""
Purpose
-------
Click command definitions for the ``oct git`` command group (Phase 4B).
Contains only CLI glue - argument parsing, JSON rendering, exit-code
translation. All computational work is delegated to
:mod:`oct.git.quality_gate`, :mod:`oct.git.policy`, and
:mod:`oct.core.git`.
 
Responsibilities
----------------
- Define the ``git`` Click group mounted under the top-level ``cli``.
- Implement ``oct git status`` with ``--json`` and ``--no-check`` flags.
- Implement ``oct git check`` with ``--staged-only``, ``--all``,
  ``--include-tests``, ``--fix``, ``--profile``, ``--json`` flags.
- Render human-readable terminal output and machine-readable JSON.
- Populate ``ctx.obj["_audit_record"]`` for the ``@audited`` decorator.
 
Diagnostics
-----------
Domain: GIT
Levels:
    L2 - lifecycle: command start/end
    L3 - details: scope/profile resolution
    L4 - deep trace: per-file rendering decisions
 
Contracts
---------
- Each command is wrapped with ``@audited`` from :mod:`oct.git.audit`.
- Exit codes for ``oct git check`` follow blueprint §5.3:
  0=pass, 1=lint, 2=format, 3=lint+format, 4=tests, 5=secrets.
- ``oct git status`` always exits 0 (informational only).
- Plan 199: every octrc read in this module goes through
  :func:`oct.core.octrc.load_octrc_safe`, the same loader the lint,
  secrets and formatter passes use. It resolves ``.option_c/octrc.json``
  first and falls back to the legacy ``<root>/.octrc.json`` (FS-539), so
  a migrated project is never invisible to the git layer. Loading a
  second way here is what made a subproject's config unreadable on the
  commit path; do not reintroduce a local reader.
- The loader returns ``{}`` (never ``None``) when no config exists. Every
  consumer in this module guards with ``isinstance(octrc, dict)`` or
  ``(octrc or {})``, both of which treat ``{}`` exactly as they treated
  the old ``None``.
 
Dependencies
------------
- oct.core.diagnostics (_dbg structured logger)
- oct.core.octrc (the ONE octrc loader - see the contract above)
- oct.core.paths (project-relative path helper)
- oct.core.project_root (project root detection)
- oct.git.audit (AuditRecord, audited decorator)
- click (CLI framework)
- oct.git.policy (the .gitignore marker fences, the profile exit-code policy and the branch and ignore-pattern rules)
- oct.core.exclusions (lazy; GIT_IGNORE_DEFAULTS - the ignore patterns oct git init writes)
- oct.core.git (lazy; git_run, current_branch, get_repo_root and the other git primitives the commands use)
- oct.core.workspace (lazy; manifest discovery and subproject bucketing for the workspace fan-out)
- oct.git.changelog (lazy; collect_changelog_entries and format_changelog for oct git changelog)
- oct.git.commit_observer (lazy; unstaged-mutation snapshots taken around the git commit call)
- oct.git.conventional (lazy; commit message validation and diagnostics-impact analysis)
- oct.git.quality_gate (lazy; run_quality_gate and the secrets scan - the lint, format and secrets gate)
- oct.hooks.oct_hooks (lazy; install_hooks for oct git init)
- oct.linter.oct_lint (lazy; _lint_single_file - the quick header and docstring check for status)
"""
 
from __future__ import annotations
 
import json
import sys
from dataclasses import dataclass
from pathlib import Path
 
import click
 
from oct.core.diagnostics import _dbg
from oct.core.octrc import load_octrc_safe
from oct.core.paths import to_project_relative
from oct.core.project_root import get_project_root, find_project_root
from oct.git.audit import AuditRecord, audited, _utc_now_iso
 
 
# =====================================================================
# Constants
# =====================================================================
 
_LARGE_FILE_THRESHOLD_BYTES: int = 10 * 1024 * 1024  # 10 MB
 
 
# =====================================================================
# Git command group
# =====================================================================
 
 
@click.group()
@click.pass_context
def git(ctx: click.Context) -> None:
    """Git integration with Option C compliance checks."""
    pass
 
 
# =====================================================================
# oct git status
# =====================================================================
 
 
def _quick_check_file(path: Path, project_root: Path) -> tuple[bool, list[str]]:
    """Run a minimal lint check (header + docstring only) on a Python file.
 
    Returns ``(compliant, issues_list)``. This is the "quick check" from
    blueprint §5.2 - not the full gate, just enough for status badges.
    """
    from oct.linter.oct_lint import LinterContext, _lint_single_file
 
    ctx = LinterContext(
        project_root=project_root,
        project_name=project_root.name,
        diagnostics_dir=project_root / "oc_diagnostics",
        tests_dir=project_root / "tests",
        docs_dir=project_root / "docs",
    )
    # Quick-check rules: only header and docstring.
    quick_rules = {
        "header": True,
        "docstring": True,
        "diagnostics_profile": False,
        "dbg_usage": False,
        "dbg_import": True,
        "func_pattern": False,
        "tests_exist": False,
        "type_hints": False,
        "no_hardcoded_secrets": False,
        "dbg_assert_safety": False,
        # Advisory and never read here; omitted rules default to on.
        "dependencies_match": False,
    }
    try:
        result = _lint_single_file(path, ctx, quick_rules, False)
        issues = []
        detail = result.get("json_detail", {}).get("checks", {})
        for check_name in ("header", "docstring", "dbg_import"):
            info = detail.get(check_name, {})
            if not info.get("pass", True):
                msg = info.get("message", check_name)
                issues.append(msg)
        return result.get("compliant", True), issues
    except Exception:
        return True, []  # If we can't check, don't penalise
 
 
def _badge(compliant: bool | None) -> str:
    """Return ``[OK]``, ``[!!]``, or ``---`` for status rendering."""
    if compliant is None:
        return "---"
    return "[OK]" if compliant else "[!!]"
 
 
def _is_within_project(path: Path, project_root_resolved: Path) -> bool:
    """Return ``True`` if *path* is inside the OCT project root.
 
    Used by ``git_status_cmd`` to filter git results down to the
    project scope in monorepo layouts where the git repo root is an
    ancestor of the OCT project root.
    """
    try:
        path.resolve().relative_to(project_root_resolved)
        return True
    except ValueError:
        return False
 
 
def _maybe_load_workspace(root: Path):
    """Probe for workspace mode and load the manifest if applicable.
 
    F1.2 helper. Returns ``(workspace, in_workspace_mode)``:
 
    - ``in_workspace_mode = True`` when *root* is itself the workspace
      root (the directory containing ``.option_c_workspace.json``).
      This activates per-subproject fan-out.
    - ``in_workspace_mode = False`` when invocation came from inside a
      subproject (project-root discovery resolved to that subproject
      first), even if a workspace manifest exists higher up. This
      preserves backwards-compat byte-for-byte for single-project
      callers.
 
    Returns ``(None, False)`` on any load failure; the caller falls
    back to the legacy single-project rendering path.
    """
    func = "_maybe_load_workspace"
    try:
        from oct.core.workspace import (
            find_workspace_root,
            load_workspace,
        )
    except ImportError:
        return None, False
 
    ws_path = find_workspace_root(root)
    if ws_path is None:
        return None, False
    in_workspace_mode = ws_path.resolve() == root.resolve()
    if not in_workspace_mode:
        return None, False
    try:
        ws = load_workspace(ws_path)
    except Exception as exc:
        _dbg(
            "GIT", func,
            f"workspace manifest at {ws_path} failed to load: {exc!r}; "
            f"falling back to single-project mode",
            1, override=True,
        )
        return None, False
    return ws, True
 
 
def _build_workspace_view(
    *,
    workspace,
    staged: list[Path],
    modified: list[Path],
    untracked: list[Path],
    no_check: bool,
) -> dict:
    """Group files into per-subproject + workspace buckets for F1.2.
 
    Returns a dict keyed by subproject name, plus ``"workspace"`` for
    files matching ``workspace.workspace_files`` and
    :data:`UNSCOPED_BUCKET` for everything else; each value is
    ``{"staged": [...], "modified": [...], "untracked": [...]}`` of
    annotated entries.
 
    Plan 213 / D-E: this view used to DROP files matching no subproject
    and no ``workspace_files`` glob. That was the pre-208 silent-drop
    shape surviving in the advisory view - incoherent with the commit
    gate's partition invariant, and misleading in exactly the case that
    matters, since a file the user cannot see in any section is the one
    they are most likely to be surprised by at commit time. Unmatched
    files now render in an ``__unscoped__`` section, under the same
    workspace-root lens the gate uses for that bucket.
    """
    from oct.core.workspace import (
        assign_to_subproject,
        matches_workspace_files,
    )
 
    workspace_root_resolved = workspace.root.resolve()
 
    buckets: dict[str, dict[str, list[dict]]] = {
        sp.name: {"staged": [], "modified": [], "untracked": []}
        for sp in workspace.subprojects
    }
    buckets["workspace"] = {
        "staged": [], "modified": [], "untracked": [],
    }
    buckets[UNSCOPED_BUCKET] = {
        "staged": [], "modified": [], "untracked": [],
    }
 
    def _annotate_for_bucket(
        path: Path, bucket_root: Path,
    ) -> dict:
        """Annotate *path* using *bucket_root* for relative-path display."""
        try:
            rel = path.resolve().relative_to(bucket_root.resolve()).as_posix()
        except ValueError:
            rel = path.as_posix()
        if path.suffix != ".py":
            return {"path": rel, "status": "non_python", "issues": []}
        if no_check:
            return {"path": rel, "status": "unchecked", "issues": []}
        compliant, issues = _quick_check_file(path, bucket_root)
        return {
            "path": rel,
            "status": "ok" if compliant else "violations",
            "issues": issues,
        }
 
    def _annotate_workspace(path: Path) -> dict:
        """Workspace-bucket annotation: path rendered repo-relative,
        compliance not quick-checked (workspace files are usually
        non-Python - README, .gitignore, etc.)."""
        try:
            rel = path.resolve().relative_to(
                workspace_root_resolved,
            ).as_posix()
        except ValueError:
            rel = path.as_posix()
        if path.suffix != ".py":
            return {"path": rel, "status": "non_python", "issues": []}
        if no_check:
            return {"path": rel, "status": "unchecked", "issues": []}
        compliant, issues = _quick_check_file(path, workspace.root)
        return {
            "path": rel,
            "status": "ok" if compliant else "violations",
            "issues": issues,
        }
 
    def _route(path: Path, kind: str) -> None:
        sp = assign_to_subproject(path, workspace)
        if sp is not None:
            buckets[sp.name][kind].append(
                _annotate_for_bucket(path, sp.path),
            )
            return
        # Not in any subproject; consult workspace_files.
        try:
            rel = path.resolve().relative_to(
                workspace_root_resolved,
            ).as_posix()
        except ValueError:
            rel = path.as_posix()
        if matches_workspace_files(rel, workspace.workspace_files):
            buckets["workspace"][kind].append(_annotate_workspace(path))
        else:
            # Plan 213 / D-E: no longer dropped. Same workspace-root lens
            # the commit gate applies to its own __unscoped__ bucket.
            buckets[UNSCOPED_BUCKET][kind].append(_annotate_workspace(path))
 
    for p in staged:
        _route(p, "staged")
    for p in modified:
        _route(p, "modified")
    # Untracked files should also bucket via the same logic so the user
    # sees per-subproject context for new files.
    for p in untracked:
        sp = assign_to_subproject(p, workspace)
        if sp is not None:
            try:
                rel = p.resolve().relative_to(sp.path).as_posix()
            except ValueError:
                rel = p.as_posix()
            buckets[sp.name]["untracked"].append({
                "path": rel, "status": "untracked", "issues": [],
            })
        else:
            try:
                rel = p.resolve().relative_to(
                    workspace_root_resolved,
                ).as_posix()
            except ValueError:
                rel = p.as_posix()
            target = (
                "workspace"
                if matches_workspace_files(rel, workspace.workspace_files)
                else UNSCOPED_BUCKET
            )
            buckets[target]["untracked"].append({
                "path": rel, "status": "untracked", "issues": [],
            })
 
    # Plan 213 / D-E: the unscoped bucket is emitted ONLY when it holds
    # something. That keeps the change strictly additive - a workspace
    # where every file is owned or declared produces byte-identical JSON
    # to before - while the files that used to vanish now have a home.
    if not any(buckets[UNSCOPED_BUCKET].values()):
        del buckets[UNSCOPED_BUCKET]
 
    return buckets
 
 
def _get_untracked_files(root: Path) -> list[Path]:
    """Return untracked files (not ignored) via ``git ls-files``."""
    func = "_get_untracked_files"
    _dbg("GIT", func, f"root={root}", 3)
    from oct.core.git import git_run
    proc = git_run(
        ["ls-files", "--others", "--exclude-standard"],
        cwd=root,
    )
    result = []
    for line in proc.stdout.splitlines():
        line = line.strip()
        if line:
            result.append((root / line).resolve())
    return result
 
 
def _get_modified_unstaged(root: Path) -> list[Path]:
    """Return tracked files with unstaged changes."""
    func = "_get_modified_unstaged"
    _dbg("GIT", func, f"root={root}", 3)
    from oct.core.git import git_run
    proc = git_run(
        ["diff", "--name-only", "--diff-filter=ACMR"],
        cwd=root,
    )
    result = []
    for line in proc.stdout.splitlines():
        line = line.strip()
        if line:
            result.append((root / line).resolve())
    return result
 
 
def _get_ahead_behind(root: Path) -> tuple[int, int]:
    """Return ``(ahead, behind)`` counts relative to upstream.
 
    Returns ``(0, 0)`` if there is no upstream or if git fails.
    """
    from oct.core.git import git_run
    proc = git_run(
        ["rev-list", "--count", "--left-right", "@{upstream}...HEAD"],
        cwd=root,
        check=False,
    )
    if proc.returncode != 0:
        return (0, 0)
    parts = proc.stdout.strip().split()
    if len(parts) == 2:
        try:
            return int(parts[1]), int(parts[0])
        except ValueError:
            pass
    return (0, 0)
 
 
@git.command(name="status")
@click.option("--json", "json_mode", is_flag=True, help="Emit JSON output.")
@click.option(
    "--no-check", is_flag=True,
    help="Skip compliance annotations (plain file list).",
)
@click.option(
    "--scope", "scope",
    type=click.Choice(["project", "repo", "auto"]),
    default=None,
    help=(
        "Scope of files to show: project (default), repo (whole git "
        "repo), auto (project unless project_root != repo_root)."
    ),
)
@click.pass_context
@audited("oct git status")
def git_status_cmd(
    ctx: click.Context,
    json_mode: bool,
    no_check: bool,
    scope: str | None,
) -> None:
    """Show git status with Option C compliance annotations."""
    func = "git_status_cmd"
    root = get_project_root(ctx)
    _dbg(
        "GIT", func,
        f"root={root} json={json_mode} no_check={no_check} scope={scope!r}",
        2,
    )
 
    from oct.core.git import (
        current_branch,
        get_repo_root,
        git_staged_files,
        is_detached_head,
        is_git_repo,
    )
 
    if not is_git_repo(root):
        click.echo("Not a git repository.", err=True)
        ctx.exit(1)
        return
 
    # Git returns paths relative to the repo root, which may differ from
    # the OCT project root in monorepo layouts.  Use the actual git repo
    # root for file-listing calls so paths resolve correctly, and keep
    # ``root`` (project root) for compliance checks and display.
    repo_root = get_repo_root(root)
    _dbg("GIT", func, f"repo_root={repo_root}", 3)
 
    # F1.2: detect workspace mode. Mode activates only when the project
    # root resolved to the workspace root itself - invocation from
    # inside a subproject keeps the legacy single-project rendering.
    workspace, in_workspace_mode = _maybe_load_workspace(root)
    _dbg(
        "GIT", func,
        f"workspace_mode={in_workspace_mode} "
        f"subprojects={(len(workspace.subprojects) if workspace else 0)}",
        3,
    )
 
    # Resolve effective scope: CLI > octrc > "project".
    if scope is None:
        octrc = load_octrc_safe(root)
        if isinstance(octrc, dict):
            git_cfg = octrc.get("git")
            if isinstance(git_cfg, dict):
                raw = git_cfg.get("status_default_scope")
                if isinstance(raw, str) and raw in ("project", "repo", "auto"):
                    scope = raw
    if scope is None:
        scope = "project"
 
    root_resolved = root.resolve()
    repo_root_resolved = repo_root.resolve()
 
    # ``auto`` collapses to ``project`` when project root == repo root.
    effective_scope = scope
    if scope == "auto":
        effective_scope = (
            "project" if root_resolved == repo_root_resolved else "repo"
        )
    _dbg("GIT", func, f"effective_scope={effective_scope}", 3)
 
    branch = current_branch(repo_root)
    detached = is_detached_head(repo_root)
    ahead, behind = _get_ahead_behind(repo_root)
    staged = git_staged_files(repo_root)
    modified = _get_modified_unstaged(repo_root)
    untracked = _get_untracked_files(repo_root)
 
    # Partition into in-project and out-of-project so the renderer can
    # show them separately when scope=repo or scope=auto-monorepo.
    def _partition(files: list[Path]) -> tuple[list[Path], list[Path]]:
        in_proj: list[Path] = []
        out_proj: list[Path] = []
        for p in files:
            if _is_within_project(p, root_resolved):
                in_proj.append(p)
            else:
                out_proj.append(p)
        return in_proj, out_proj
 
    staged_in, staged_out = _partition(staged)
    modified_in, modified_out = _partition(modified)
    untracked_in, untracked_out = _partition(untracked)
 
    # Deduplicate: same-file appearing in both staged and modified is
    # an artefact of path-resolution noise; mirror real git semantics
    # by keeping it in staged only (real "modified" entries have
    # additional unstaged hunks which the user will see on diff).
    staged_in_set = {p.resolve() for p in staged_in}
    modified_in = [p for p in modified_in if p.resolve() not in staged_in_set]
    staged_out_set = {p.resolve() for p in staged_out}
    modified_out = [p for p in modified_out if p.resolve() not in staged_out_set]
 
    # Apply scope to determine what to render.
    show_out_of_project = effective_scope == "repo"
 
    # -- Build annotated file lists --
    def annotate(files: list[Path]) -> list[dict]:
        annotated = []
        for path in files:
            rel = to_project_relative(path, root)
            if path.suffix != ".py":
                annotated.append({
                    "path": rel, "status": "non_python", "issues": [],
                })
            elif no_check:
                annotated.append({
                    "path": rel, "status": "unchecked", "issues": [],
                })
            else:
                compliant, issues = _quick_check_file(path, root)
                annotated.append({
                    "path": rel,
                    "status": "ok" if compliant else "violations",
                    "issues": issues,
                })
        return annotated
 
    def annotate_out(files: list[Path]) -> list[dict]:
        """Out-of-project files: render with repo-relative path, unchecked."""
        out: list[dict] = []
        for path in files:
            try:
                rel = path.resolve().relative_to(repo_root_resolved).as_posix()
            except ValueError:
                rel = path.as_posix()
            out.append({"path": rel, "status": "unchecked", "issues": []})
        return out
 
    staged_ann = annotate(staged_in)
    modified_ann = annotate(modified_in)
    untracked_ann = [
        {"path": to_project_relative(p, root),
         "status": "untracked", "issues": []}
        for p in untracked_in
    ]
 
    out_of_project_ann: list[dict] = []
    if show_out_of_project:
        for entry in annotate_out(staged_out):
            entry["section"] = "staged"
            out_of_project_ann.append(entry)
        for entry in annotate_out(modified_out):
            entry["section"] = "modified"
            out_of_project_ann.append(entry)
        for path in untracked_out:
            try:
                rel = path.resolve().relative_to(repo_root_resolved).as_posix()
            except ValueError:
                rel = path.as_posix()
            out_of_project_ann.append({
                "path": rel, "status": "untracked",
                "issues": [], "section": "untracked",
            })
 
    # -- F1.2 workspace_view: per-subproject + workspace buckets --
    workspace_view: dict | None = None
    if in_workspace_mode and workspace is not None and workspace.subprojects:
        workspace_view = _build_workspace_view(
            workspace=workspace,
            staged=staged_in,
            modified=modified_in,
            untracked=untracked_in,
            no_check=no_check,
        )
 
    # -- Populate audit record --
    record = ctx.obj.get("_audit_record", AuditRecord())
    record.timestamp = record.timestamp or _utc_now_iso()
    record.command = "oct git status"
    record.branch = branch if not detached else None
    record.staged_files = len(staged_in)
    record.checks_passed = True
    record.exit_code = 0
    ctx.obj["_audit_record"] = record
 
    # -- Render output --
    if json_mode:
        output = {
            "branch": branch if not detached else None,
            "detached": detached,
            "ahead": ahead,
            "behind": behind,
            "scope": effective_scope,
            "staged": staged_ann,
            "modified": modified_ann,
            "untracked": untracked_ann,
            "out_of_project": out_of_project_ann,
            "workspace_view": workspace_view,
        }
        click.echo(json.dumps(output, indent=2, ensure_ascii=False))
        return
 
    # Terminal output (blueprint §5.2 format).
    if detached:
        click.echo("HEAD detached")
    else:
        click.echo(f"On branch {branch}")
    if ahead > 0 or behind > 0:
        parts = []
        if ahead > 0:
            parts.append(f"ahead by {ahead}")
        if behind > 0:
            parts.append(f"behind by {behind}")
        click.echo(f"  {', '.join(parts)}")
    click.echo()
 
    def _render_section(label: str, items: list[dict]):
        if not items:
            return
        click.echo(f"{label} ({len(items)} file{'s' if len(items) != 1 else ''}):")
        for item in items:
            status = item["status"]
            path = item["path"]
            if status == "ok":
                badge = "  [OK]  "
            elif status == "violations":
                issues_str = "  (" + ", ".join(item["issues"]) + ")" if item["issues"] else ""
                click.echo(f"  [!!]  {path}{issues_str}")
                continue
            elif status == "non_python":
                badge = "  ---   "
            elif status == "unchecked":
                badge = "        "
            else:
                badge = "        "
            click.echo(f"{badge}{path}")
        click.echo()
 
    if workspace_view is not None:
        # F1.2: workspace mode - render one block per subproject + a
        # workspace-level block for files matching workspace_files[].
        for bucket_name in (
            *(sp.name for sp in workspace.subprojects),
            "workspace",
            UNSCOPED_BUCKET,
        ):
            bucket = workspace_view.get(bucket_name, {})
            staged_b = bucket.get("staged", [])
            modified_b = bucket.get("modified", [])
            untracked_b = bucket.get("untracked", [])
            if not (staged_b or modified_b or untracked_b):
                continue
            click.echo(f"=== {bucket_name} ===")
            _render_section("Staged", staged_b)
            _render_section("Modified", modified_b)
            _render_section("Untracked", untracked_b)
    else:
        _render_section("Staged", staged_ann)
        _render_section("Modified", modified_ann)
        _render_section("Untracked", untracked_ann)
    if show_out_of_project and out_of_project_ann:
        click.echo(
            f"Other repo files ({len(out_of_project_ann)} "
            f"file{'s' if len(out_of_project_ann) != 1 else ''}, "
            f"outside OCT project):"
        )
        for item in out_of_project_ann:
            section = item.get("section", "?")
            click.echo(f"  [{section[:3]}]  {item['path']}")
        click.echo()
 
 
# =====================================================================
# oct git check
# =====================================================================
 
 
#: D-C refusal text for ``oct git check --fix`` at a workspace root.
#: Fix-mode through ``run_quality_gate_on_files`` is unexercised in
#: fan-out, and the formatter has a known multi-line-``def`` fixer
#: landmine. A loud refusal is coherent; a silently different - or
#: destructive - verdict is not.
_FANOUT_FIX_REFUSAL: str = (
    "Refusing to run --fix across the workspace.\n"
    "Fix mode is supported per subproject only. Run it from the "
    "subproject you want to fix, for example:\n"
    "    cd {example}\n"
    "    oct git check --fix\n"
    "Without --fix, this command checks the whole workspace."
)
 
 
def _git_check_fanout(
    ctx: click.Context,
    *,
    workspace,
    scope: str,
    cli_profile: str | None,
    fallback_profile: str,
    branch: str | None,
    include_tests: bool,
    fix: bool,
    json_mode: bool,
) -> int:
    """Render ``oct git check`` in workspace fan-out mode (plan 213 / D-B).
 
    Resolves the scope from ``workspace.root`` with the SAME helper the
    commit gate uses (:func:`oct.git.quality_gate._resolve_scope`), gates
    it through :func:`_run_fanout_gate`, renders one section per non-empty
    bucket (mirroring ``oct git status``), and returns the worst-wins exit
    code.
 
    ``--fix`` refuses loudly here (D-C). JSON output keeps the legacy
    top-level shape, computed from the merged result, and ADDS a
    ``per_subproject`` array - the same precedent ``oct git status`` set
    with ``workspace_view``.
    """
    func = "_git_check_fanout"
    from oct.git.quality_gate import _resolve_scope
 
    if fix:
        example = (
            workspace.subprojects[0].name
            if workspace.subprojects else "<subproject>"
        )
        message = _FANOUT_FIX_REFUSAL.format(example=example)
        _dbg("GIT", func, "refusing --fix in fan-out mode", 2)
        if json_mode:
            click.echo(json.dumps(
                {
                    "scope": scope,
                    "profile": fallback_profile,
                    "workspace": workspace.name,
                    "exit_code": 2,
                    "errors": [message],
                },
                indent=2, ensure_ascii=False,
            ))
        else:
            click.echo(message, err=True)
        return 2
 
    files = _resolve_scope(workspace.root, scope)
    _dbg("GIT", func, f"scope={scope} files={len(files)}", 3)
 
    outcome = _run_fanout_gate(
        workspace, files,
        cli_profile=cli_profile, branch=branch,
        fallback_profile=fallback_profile,
        include_tests=include_tests,
    )
 
    # Fail-closed: the buckets did not partition the scope, so no verdict
    # this command could print would be trustworthy.
    if outcome.unbucketed:
        names = ", ".join(
            to_project_relative(p, workspace.root)
            for p in outcome.unbucketed
        )
        message = (
            f"{len(outcome.unbucketed)} file(s) could not be assigned to "
            f"any quality-gate scope and were NOT checked: {names}."
        )
        _dbg("GIT", func, f"partition invariant violated: {names}", 1,
             override=True)
        if json_mode:
            click.echo(json.dumps(
                {
                    "scope": scope,
                    "profile": fallback_profile,
                    "workspace": workspace.name,
                    "exit_code": 2,
                    "errors": [message],
                },
                indent=2, ensure_ascii=False,
            ))
        else:
            click.echo(f"Quality gate aborted: {message}", err=True)
        return 2
 
    result = outcome.result
    exit_code = outcome.gate_exit
    result.exit_code = exit_code
    per_bucket = _build_per_subproject_audit(
        outcome.per_bucket_results, outcome.per_bucket_meta,
    )
 
    # -- Populate audit record --
    record = ctx.obj.get("_audit_record", AuditRecord())
    record.timestamp = record.timestamp or _utc_now_iso()
    record.command = "oct git check"
    record.branch = branch
    record.staged_files = result.files_checked
    record.checks_passed = (exit_code == 0)
    record.secrets_detected = bool(result.secrets_findings)
    record.lint_violations = result.lint_violations
    record.format_violations = result.format_violations
    record.exit_code = exit_code
    ctx.obj["_audit_record"] = record
    ctx.obj["_per_subproject_results"] = per_bucket
 
    if json_mode:
        click.echo(json.dumps(
            {
                "scope": scope,
                "profile": fallback_profile,
                "workspace": workspace.name,
                "files_checked": result.files_checked,
                "lint_violations": result.lint_violations,
                "format_violations": result.format_violations,
                "secrets_findings": [
                    {"path": s[0], "line": s[1], "reason": s[2]}
                    for s in result.secrets_findings
                ],
                "test_failures": result.test_failures,
                "exit_code": exit_code,
                "duration_ms": result.duration_ms,
                "per_subproject": per_bucket,
            },
            indent=2, ensure_ascii=False,
        ))
        return exit_code
 
    # Terminal summary - one section per non-empty bucket, then the total.
    click.echo(
        f"Quality gate: scope={scope}, workspace={workspace.name}, "
        f"buckets={len(per_bucket)}"
    )
    click.echo()
    for entry in per_bucket:
        click.echo(f"=== {entry['name']} ===")
        click.echo(f"  Profile:          {entry['profile']}")
        click.echo(f"  Files checked:    {entry['files']}")
        click.echo(f"  Lint violations:  {entry['lint_violations']}")
        click.echo(f"  Format issues:    {entry['format_violations']}")
        click.echo(f"  Secrets findings: {entry['secrets']}")
        click.echo()
 
    click.echo("=== total ===")
    click.echo(f"  Files checked:    {result.files_checked}")
    click.echo(f"  Lint violations:  {result.lint_violations}")
    click.echo(f"  Format issues:    {result.format_violations}")
    click.echo(f"  Secrets findings: {len(result.secrets_findings)}")
    if include_tests:
        click.echo(f"  Test failures:    {result.test_failures}")
    click.echo(f"  Duration:         {result.duration_ms} ms")
    click.echo()
 
    if result.secrets_findings:
        click.echo("SECRETS DETECTED (always blocking):", err=True)
        for path_str, line, reason in result.secrets_findings:
            loc = f"{path_str}:{line}" if line else path_str
            click.echo(f"  {loc} - {reason}", err=True)
        click.echo()
 
    if exit_code == 0:
        click.echo("All checks passed.")
    else:
        labels = {
            1: "Lint violations",
            2: "Format violations",
            3: "Lint + format violations",
            4: "Test failures",
            5: "Secrets detected",
        }
        click.echo(
            f"FAILED (exit {exit_code}): {labels.get(exit_code, 'unknown')}",
            err=True,
        )
    return exit_code
 
 
@git.command(name="check")
@click.option(
    "--staged-only", is_flag=True,
    help="Check only staged (index) files.",
)
@click.option(
    "--all", "check_all", is_flag=True,
    help="Check every Python file in the project.",
)
@click.option(
    "--include-tests", is_flag=True,
    help="Also run pytest (exit code 4 on failure).",
)
@click.option(
    "--fix", is_flag=True,
    help="Attempt auto-fix of lint/format violations, then re-check.",
)
@click.option(
    "--profile",
    type=click.Choice(["proto", "compact", "strict"], case_sensitive=False),
    default=None,
    help="Override the lint profile.",
)
@click.option("--json", "json_mode", is_flag=True, help="Emit JSON output.")
@click.option(
    "--sandbox", is_flag=True, default=False,
    help="OI-517: run pytest (when --include-tests) under the MCP sandbox "
         "environment so PYTHONPATH and secret-named env vars are stripped.",
)
@click.pass_context
@audited("oct git check")
def git_check_cmd(
    ctx: click.Context,
    staged_only: bool,
    check_all: bool,
    include_tests: bool,
    fix: bool,
    profile: str | None,
    json_mode: bool,
    sandbox: bool,
) -> None:
    """Run the unified quality gate: lint + format + secrets."""
    func = "git_check_cmd"
    root = get_project_root(ctx)
    _dbg("GIT", func, f"root={root}", 2)
 
    from oct.core.git import current_branch, is_git_repo
    from oct.git.policy import (
        apply_profile_policy,
        resolve_profile_for_project,
    )
    from oct.git.quality_gate import run_quality_gate
 
    if not is_git_repo(root):
        click.echo("Not a git repository.", err=True)
        ctx.exit(1)
        return
 
    # -- Resolve scope --
    if staged_only:
        scope = "staged"
    elif check_all:
        scope = "all"
    else:
        scope = "changed"
 
    # -- Resolve profile --
    branch = current_branch(root)
    octrc = load_octrc_safe(root)
    # Plan 197: manifest-aware so a subproject is gated at the profile
    # the workspace declares, not just whatever its octrc says.
    effective_profile = resolve_profile_for_project(
        root, branch, octrc, profile,
    )
    _dbg(
        "GIT", func,
        f"scope={scope} profile={effective_profile} fix={fix}",
        3,
    )
 
    # -- Plan 213 / D-B: workspace fan-out --
    # Invoked at a workspace root, check must partition the scope by owning
    # subproject exactly as the commit gate does. Before 213 it ran ONE
    # run_quality_gate pass rooted here - and the workspace root has no
    # octrc and matches no subproject, so the profile collapsed to compact
    # (or strict on a protected branch) and workspace_profile was never
    # consulted. check --staged-only and the commit gate could therefore
    # disagree on an identical staged set. Invocation from inside a
    # subproject, or in a non-workspace repo, is byte-identical to before:
    # _maybe_load_workspace returns (None, False) in both cases.
    check_workspace, check_in_workspace_mode = _maybe_load_workspace(root)
    use_fanout = (
        check_in_workspace_mode
        and check_workspace is not None
        and bool(check_workspace.subprojects)
    )
 
    if use_fanout:
        exit_code = _git_check_fanout(
            ctx,
            workspace=check_workspace,
            scope=scope,
            cli_profile=profile,
            fallback_profile=effective_profile,
            branch=branch,
            include_tests=include_tests,
            fix=fix,
            json_mode=json_mode,
        )
        ctx.exit(exit_code)
        return
 
    # -- Run quality gate --
    result = run_quality_gate(
        project_root=root,
        scope=scope,
        profile=effective_profile,
        include_tests=include_tests,
        fix=fix,
        sandbox=sandbox,
    )
 
    # -- Derive exit code via policy --
    exit_code = apply_profile_policy(effective_profile, result)
    result.exit_code = exit_code
 
    # -- Populate audit record --
    record = ctx.obj.get("_audit_record", AuditRecord())
    record.timestamp = record.timestamp or _utc_now_iso()
    record.command = "oct git check"
    record.branch = branch
    record.staged_files = result.files_checked
    record.checks_passed = (exit_code == 0)
    record.secrets_detected = bool(result.secrets_findings)
    record.lint_violations = result.lint_violations
    record.format_violations = result.format_violations
    record.exit_code = exit_code
    ctx.obj["_audit_record"] = record
 
    # -- Render output --
    if json_mode:
        output = {
            "scope": scope,
            "profile": effective_profile,
            "files_checked": result.files_checked,
            "lint_violations": result.lint_violations,
            "format_violations": result.format_violations,
            "secrets_findings": [
                {"path": s[0], "line": s[1], "reason": s[2]}
                for s in result.secrets_findings
            ],
            "test_failures": result.test_failures,
            "exit_code": exit_code,
            "duration_ms": result.duration_ms,
        }
        click.echo(json.dumps(output, indent=2, ensure_ascii=False))
        ctx.exit(exit_code)
        return
 
    # Terminal summary.
    click.echo(f"Quality gate: scope={scope}, profile={effective_profile}")
    click.echo(f"  Files checked:    {result.files_checked}")
    click.echo(f"  Lint violations:  {result.lint_violations}")
    click.echo(f"  Format issues:    {result.format_violations}")
    click.echo(f"  Secrets findings: {len(result.secrets_findings)}")
    if include_tests:
        click.echo(f"  Test failures:    {result.test_failures}")
    click.echo(f"  Duration:         {result.duration_ms} ms")
    click.echo()
 
    if result.secrets_findings:
        click.echo("SECRETS DETECTED (always blocking):", err=True)
        for path_str, line, reason in result.secrets_findings:
            loc = f"{path_str}:{line}" if line else path_str
            click.echo(f"  {loc} - {reason}", err=True)
        click.echo()
 
    if exit_code == 0:
        click.echo("All checks passed.")
    else:
        labels = {
            1: "Lint violations",
            2: "Format violations",
            3: "Lint + format violations",
            4: "Test failures",
            5: "Secrets detected",
        }
        click.echo(
            f"FAILED (exit {exit_code}): {labels.get(exit_code, 'unknown')}",
            err=True,
        )
 
    ctx.exit(exit_code)
 
 
# =====================================================================
# oct git init
# =====================================================================
 
 
_GITATTRIBUTES_DEFAULTS: tuple[str, ...] = (
    "*.py text eol=lf",
    "*.pem binary",
    "*.key binary",
)
 
 
_GITHUB_WORKFLOW_TEMPLATE = """\
name: Option C CI
on: [push, pull_request]
jobs:
  check:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install -e ./oct
      - run: oct lint --json
      - run: oct format --dry-run
      - run: oct test
"""
 
# Marker comments used to delimit the Option C block in .gitignore / .gitattributes.
# Shared with ``oct/git/policy.py`` so ``oct git init`` (which writes the initial
# block with section comments) and ``oct git ignore`` (which adds bare patterns)
# agree on the literal fence strings.
from oct.git.policy import (
    OC_GITIGNORE_MARKER_END as _OC_MARKER_END,
    OC_GITIGNORE_MARKER_START as _OC_MARKER_START,
)
 
 
def _merge_ignore_file(
    path: Path, entries: tuple[str, ...], dry_run: bool,
) -> list[str]:
    """Merge *entries* into an ignore/attributes file, appending only missing lines.
 
    Returns the list of entries that were (or would be) added.
    """
    existing_lines: set[str] = set()
    if path.is_file():
        for line in path.read_text(encoding="utf-8").splitlines():
            existing_lines.add(line.strip())
 
    # Filter to entries not already present (skip comments/blanks for dedup).
    to_add: list[str] = []
    for entry in entries:
        stripped = entry.strip()
        if stripped == "" or stripped.startswith("#"):
            # Always include structural lines in the block, but they don't
            # count as "added entries" for reporting.
            continue
        if stripped not in existing_lines:
            to_add.append(stripped)
 
    if not to_add:
        return []
 
    # Build the block to append, including all entries (with comments/blanks).
    block_lines: list[str] = [_OC_MARKER_START]
    for entry in entries:
        stripped = entry.strip()
        if stripped == "" or stripped.startswith("#"):
            block_lines.append(entry)
        elif stripped in {a for a in to_add}:
            block_lines.append(entry)
    block_lines.append(_OC_MARKER_END)
 
    if not dry_run:
        # Ensure existing content ends with a newline before appending.
        existing = ""
        if path.is_file():
            existing = path.read_text(encoding="utf-8")
        if existing and not existing.endswith("\n"):
            existing += "\n"
        with path.open("w", encoding="utf-8") as fh:
            fh.write(existing + "\n".join(block_lines) + "\n")
 
    return to_add
 
 
def _create_gitattributes(root: Path, dry_run: bool) -> list[str]:
    """Create or merge ``.gitattributes`` with Option C defaults."""
    path = root / ".gitattributes"
    return _merge_ignore_file(path, _GITATTRIBUTES_DEFAULTS, dry_run)
 
 
def _generate_github_workflow(root: Path, dry_run: bool) -> bool:
    """Write ``.github/workflows/option-c.yml``. Returns True if created."""
    workflow_dir = root / ".github" / "workflows"
    workflow_path = workflow_dir / "option-c.yml"
    if workflow_path.is_file():
        return False
    if not dry_run:
        workflow_dir.mkdir(parents=True, exist_ok=True)
        workflow_path.write_text(_GITHUB_WORKFLOW_TEMPLATE, encoding="utf-8")
    return True
 
 
@git.command(name="init")
@click.option("--dry-run", is_flag=True, help="Show what would be created/modified.")
@click.option("--github", is_flag=True, help="Also generate GitHub Actions workflow.")
@click.option("--no-hooks", is_flag=True, help="Skip hook installation.")
@click.pass_context
@audited("oct git init")
def git_init_cmd(
    ctx: click.Context,
    dry_run: bool,
    github: bool,
    no_hooks: bool,
) -> None:
    """Initialise or enhance a git repo with Option C conventions."""
    func = "git_init_cmd"
    # Accept either a --root-dir context or cwd.
    try:
        root = get_project_root(ctx)
    except click.ClickException:
        root = Path.cwd()
    _dbg("GIT", func, f"root={root} dry_run={dry_run}", 2)
 
    from oct.core.git import is_git_repo, git_run
    from oct.core.exclusions import GIT_IGNORE_DEFAULTS
    from oct.hooks.oct_hooks import install_hooks
 
    actions: list[str] = []
 
    # 1. git init if needed
    if not is_git_repo(root):
        if dry_run:
            actions.append("Would run: git init")
        else:
            git_run(["init"], cwd=root)
            actions.append("Initialised git repository")
 
    # F1.4: when running at a workspace root, augment the standard
    # ignore-defaults with the manifest's ignore_at_workspace_scope[]
    # patterns. Subproject roots get the standard defaults only.
    init_ws, init_in_ws = _maybe_load_workspace(root)
    if (
        init_in_ws and init_ws is not None
        and init_ws.ignore_at_workspace_scope
    ):
        ignore_entries: tuple[str, ...] = (
            *GIT_IGNORE_DEFAULTS,
            "",
            "# --- workspace-scope additions ---",
            *init_ws.ignore_at_workspace_scope,
        )
        _dbg(
            "GIT", func,
            f"workspace-mode init: appending "
            f"{len(init_ws.ignore_at_workspace_scope)} workspace patterns",
            3,
        )
    else:
        ignore_entries = GIT_IGNORE_DEFAULTS
 
    # 2. Merge .gitignore
    added_ignores = _merge_ignore_file(
        root / ".gitignore", ignore_entries, dry_run,
    )
    if added_ignores:
        verb = "Would add" if dry_run else "Added"
        actions.append(f"{verb} {len(added_ignores)} entries to .gitignore")
    else:
        actions.append(".gitignore already up-to-date")
 
    # 3. Create/merge .gitattributes
    added_attrs = _create_gitattributes(root, dry_run)
    if added_attrs:
        verb = "Would add" if dry_run else "Added"
        actions.append(f"{verb} {len(added_attrs)} entries to .gitattributes")
    else:
        actions.append(".gitattributes already up-to-date")
 
    # 4. Install hooks
    if no_hooks:
        actions.append("Skipped hook installation (--no-hooks)")
    else:
        hook_path = root / ".pre-commit-config.yaml"
        if hook_path.is_file():
            actions.append(".pre-commit-config.yaml already exists")
        elif dry_run:
            actions.append("Would create .pre-commit-config.yaml")
        else:
            install_hooks(root, force=False)
            actions.append("Created .pre-commit-config.yaml")
 
    # 5. GitHub Actions workflow
    if github:
        created = _generate_github_workflow(root, dry_run)
        if created:
            verb = "Would create" if dry_run else "Created"
            actions.append(f"{verb} .github/workflows/option-c.yml")
        else:
            actions.append(".github/workflows/option-c.yml already exists")
 
    # -- Populate audit record --
    record = ctx.obj.get("_audit_record", AuditRecord())
    record.timestamp = record.timestamp or _utc_now_iso()
    record.command = "oct git init"
    record.checks_passed = True
    record.exit_code = 0
    ctx.obj["_audit_record"] = record
 
    # -- Render output --
    prefix = "[DRY RUN] " if dry_run else ""
    for action in actions:
        click.echo(f"{prefix}{action}")
    if not dry_run:
        click.echo("Option C git initialisation complete.")
 
 
# =====================================================================
# oct git commit (Phase 4D - G-D3, G-D4)
# =====================================================================
 
 
@git.command(
    name="commit",
    context_settings={"ignore_unknown_options": True},
)
@click.option(
    "-m", "--message", "messages", required=True, multiple=True,
    help=(
        "Commit message (Conventional Commits format). May be passed "
        "multiple times - each value becomes a paragraph separated by "
        "a blank line, matching ``git commit -m`` semantics."
    ),
)
@click.option("--force", "force_commit", is_flag=True,
              help="Skip lint/format checks (secrets still checked).")
@click.option("--no-verify", is_flag=True,
              help="Forward --no-verify to git (hooks skipped).")
@click.option(
    "--profile",
    type=click.Choice(["proto", "compact", "strict"], case_sensitive=False),
    default=None,
    help="Override lint profile.",
)
@click.option("--json", "json_mode", is_flag=True, help="Emit JSON output.")
@click.option(
    "--ignore-unstaged-mutation", "ignore_unstaged_mutation", is_flag=True,
    help=(
        "Downgrade B1.0 unstaged-tracked-file mutation detection from "
        "block to warn for this invocation."
    ),
)
@click.argument("extra_args", nargs=-1, type=click.UNPROCESSED)
@click.pass_context
@audited("oct git commit")
def git_commit_cmd(
    ctx: click.Context,
    messages: tuple[str, ...],
    force_commit: bool,
    no_verify: bool,
    profile: str | None,
    json_mode: bool,
    ignore_unstaged_mutation: bool,
    extra_args: tuple[str, ...],
) -> None:
    """Commit staged changes with quality gate enforcement."""
    func = "git_commit_cmd"
    root = get_project_root(ctx)
    # L4: join multiple -m values with a blank line between paragraphs
    # to mirror plain ``git commit -m subject -m body`` semantics.
    message = "\n\n".join(m for m in messages if m is not None)
    _dbg("GIT", func, f"root={root} force={force_commit} -m count={len(messages)}", 2)
 
    from oct.core.git import current_branch, git_run, is_git_repo
    from oct.git.conventional import (
        analyse_diagnostics_impact,
        format_validation_errors,
    )
    from oct.git.policy import apply_profile_policy
    from oct.git.quality_gate import pre_commit_checks, run_quality_gate
 
    if not is_git_repo(root):
        click.echo("Not a git repository.", err=True)
        ctx.exit(1)
        return
 
    branch = current_branch(root)
    octrc = load_octrc_safe(root)
 
    # -- 1-3. Pre-gate checks (OI-527): commit message, staged files, protected branch --
    pre = pre_commit_checks(
        project_root=root,
        branch=branch,
        message=message,
        force_commit=force_commit,
        profile=profile,
        octrc=octrc,
    )
    effective_profile = pre.effective_profile
    if pre.pb_advisory:
        click.echo(pre.pb_advisory, err=True)
 
    if not pre.ok:
        # -- Auto-branch on protected-branch block (Phase 4F) --
        if pre.exit_code == 1 and _should_auto_branch(branch, octrc):
            auto_branch_name = _auto_branch_for_commit(
                root, branch, message, ctx, json_mode,
            )
            if auto_branch_name is None:
                # _auto_branch_for_commit already emitted errors and
                # populated the audit record.
                return
            branch = auto_branch_name
            # Re-run pre-checks on the new (non-protected) branch.
            pre = pre_commit_checks(
                project_root=root, branch=branch, message=message,
                force_commit=force_commit, profile=profile, octrc=octrc,
            )
            effective_profile = pre.effective_profile
 
        if not pre.ok:
            if json_mode:
                _emit_commit_json(
                    ctx, exit_code=pre.exit_code, message=message,
                    branch=branch, profile=effective_profile,
                    errors=pre.errors,
                )
            elif pre.exit_code == 3:
                click.echo(format_validation_errors(pre.errors), err=True)
            elif pre.exit_code == 4:
                click.echo("Nothing to commit (no staged files).", err=True)
            # exit_code == 1 already surfaced via pb_advisory echo above.
            _populate_commit_audit(
                ctx, branch=branch, exit_code=pre.exit_code,
                checks_passed=False, staged_count=len(pre.staged_files),
            )
            ctx.exit(pre.exit_code)
            return
 
    staged = pre.staged_files
 
    # -- B1.0: snapshot tracked-but-unstaged files for mutation detection --
    observe_mode = _resolve_observe_mode(octrc, ignore_unstaged_mutation)
    # Plan 242: the method is resolved fail-closed and unconditionally,
    # before any snapshot and regardless of mode, so a typo in the key
    # is loud rather than silently picking a detection method.
    try:
        observe_method = _resolve_observe_method(octrc)
    except ValueError as exc:
        click.echo(str(exc), err=True)
        if json_mode:
            _emit_commit_json(
                ctx, exit_code=2, message=message, branch=branch,
                profile=effective_profile, errors=[str(exc)],
            )
        _populate_commit_audit(
            ctx, branch=branch, exit_code=2,
            checks_passed=False, staged_count=len(staged),
        )
        ctx.exit(2)
        return
    pre_snapshot: dict[str, str] = {}
    if observe_mode != "off":
        try:
            from oct.core.git import get_repo_root
            from oct.git.commit_observer import snapshot_unstaged_tracked
            repo_root_for_obs = get_repo_root(root)
            staged_rel = {
                str(p.resolve().relative_to(repo_root_for_obs.resolve()))
                .replace("\\", "/")
                for p in staged
                if _safe_relative(p, repo_root_for_obs) is not None
            }
            pre_snapshot = snapshot_unstaged_tracked(
                repo_root_for_obs, staged_set=staged_rel,
                method=observe_method,
            )
        except Exception as exc:
            _dbg(
                "GIT", func,
                f"observer snapshot failed: {exc!r}; continuing without",
                1, override=True,
            )
            pre_snapshot = {}
            observe_mode = "off"
 
    # -- 4. Run quality gate on staged files --
    # F1.3: in workspace mode with subprojects, fan the staged set out
    # by subproject so each subproject's slice runs under its own
    # profile. The aggregated result preserves the secrets-always-block
    # invariant (G-D4) - a finding in any subproject still trips
    # secret enforcement below.
    commit_workspace, commit_in_workspace_mode = _maybe_load_workspace(root)
    per_sp_results: dict[str, "QualityGateResult"] = {}
    per_sp_meta: dict[str, dict] = {}
    use_fanout = (
        commit_in_workspace_mode
        and commit_workspace is not None
        and bool(commit_workspace.subprojects)
    )
 
    if use_fanout:
        # Plan 213 / D-A: one shared fan-out implementation, also used by
        # ``oct git check``, so the two commands cannot disagree on the
        # same file set. Behaviour here is unchanged from the inline block
        # it replaces.
        outcome = _run_fanout_gate(
            commit_workspace, staged,
            cli_profile=profile, branch=branch,
            fallback_profile=effective_profile,
            apply_policy=not force_commit,
        )
        unbucketed = outcome.unbucketed
        if unbucketed:
            names = ", ".join(
                to_project_relative(p, root) for p in unbucketed
            )
            _dbg(
                "GIT", func,
                f"partition invariant violated; {len(unbucketed)} staged "
                f"file(s) reached no bucket: {names}",
                1, override=True,
            )
            if json_mode:
                _emit_commit_json(
                    ctx, exit_code=2, message=message, branch=branch,
                    profile=effective_profile,
                    errors=[
                        "Refusing to commit: "
                        f"{len(unbucketed)} staged file(s) could not be "
                        "assigned to any quality-gate scope and would be "
                        f"committed unscanned: {names}."
                    ],
                )
            else:
                click.echo(
                    "Refusing to commit: staged file(s) could not be "
                    "assigned to any quality-gate scope and would be "
                    f"committed unscanned:\n  {names}",
                    err=True,
                )
            ctx.exit(2)
            return
        per_sp_results = outcome.per_bucket_results
        per_sp_meta = outcome.per_bucket_meta
        result = outcome.result
        gate_exit = outcome.gate_exit
    elif not force_commit:
        result = run_quality_gate(
            project_root=root,
            scope="staged",
            profile=effective_profile,
            include_tests=False,
            fix=False,
        )
        gate_exit = apply_profile_policy(effective_profile, result)
    else:
        # --force: still run secrets check only.
        result = run_quality_gate(
            project_root=root,
            scope="staged",
            profile=effective_profile,
            include_tests=False,
            fix=False,
        )
        gate_exit = 0  # Force ignores lint/format.
 
    # G-D4 D3 (plan 208): `git commit` writes the WHOLE repo index regardless of
    # the invocation cwd, but the profile gate above scanned only the
    # invocation-scoped staged set - `git_staged_files(root)` drops every staged
    # file resolving outside `root` (core/git.py `_resolve_and_filter`). So a
    # secret staged OUTSIDE the invoking subproject (a sibling subproject, or a
    # workspace-root file) was committed unscanned. The workspace fan-out path is
    # exempt: there root == workspace root and the buckets already partition the
    # full index. For every other path - single-project, subproject-cwd, and
    # --force - adopt the secrets verdict of the FULL committed index, so the set
    # scanned for secrets always equals the set `git commit` will write.
    committed_secrets = result.secrets_findings
    if not use_fanout:
        try:
            from oct.core.git import get_repo_root, git_staged_files
            from oct.git.quality_gate import run_secrets_scan_on_files
            commit_repo_root = get_repo_root(root)
            full_index = git_staged_files(commit_repo_root)
            # The secrets pass alone, on ALL file types, configured exactly as
            # the gate's own secrets pass (one shared function). Lint and format
            # are not re-run: they used to be, over the full index, and their
            # result was discarded (found by the chart C runtime trace,
            # 2026-09-27). Scanning at the repo root merges each subproject's
            # entropy_exclusions, so operator-waivered files stay waivered.
            committed_secrets = run_secrets_scan_on_files(commit_repo_root, full_index)
        except Exception as exc:
            # Fail closed: if the full committed index cannot be scanned, refuse
            # rather than fall back to the narrower, bypassable scan.
            _dbg(
                "GIT", func,
                f"full-index secrets scan failed ({exc!r}); refusing commit",
                1, override=True,
            )
            if json_mode:
                _emit_commit_json(
                    ctx, exit_code=2, message=message, branch=branch,
                    profile=effective_profile,
                    errors=[
                        "Refusing to commit: could not scan the full committed "
                        f"index for secrets ({type(exc).__name__})."
                    ],
                )
            else:
                click.echo(
                    "Refusing to commit: could not scan the full committed "
                    f"index for secrets ({type(exc).__name__}).",
                    err=True,
                )
            ctx.exit(2)
            return
 
    # G-D4 invariant: secrets are NEVER bypassable.
    if committed_secrets:
        if json_mode:
            _emit_commit_json(
                ctx, exit_code=2, message=message, branch=branch,
                profile=effective_profile,
                errors=[f"Secrets detected in {len(committed_secrets)} location(s)."],
                secrets=committed_secrets,
            )
        else:
            click.echo("SECRETS DETECTED (non-bypassable):", err=True)
            for path_str, line, reason in committed_secrets:
                loc = f"{path_str}:{line}" if line else path_str
                click.echo(f"  {loc} - {reason}", err=True)
        _populate_commit_audit(
            ctx, branch=branch, exit_code=2,
            checks_passed=False, staged_count=len(staged),
            secrets=True, lint_v=result.lint_violations,
            format_v=result.format_violations,
        )
        ctx.exit(2)
        return
 
    # Non-secret quality gate failures (only when not --force).
    if gate_exit != 0 and not force_commit:
        if json_mode:
            _emit_commit_json(
                ctx, exit_code=1, message=message, branch=branch,
                profile=effective_profile,
                errors=[f"Quality gate failed (exit {gate_exit})."],
            )
        else:
            click.echo(
                f"Quality gate failed (exit {gate_exit}). "
                f"Use --force to skip lint/format checks.",
                err=True,
            )
        _populate_commit_audit(
            ctx, branch=branch, exit_code=1,
            checks_passed=False, staged_count=len(staged),
            lint_v=result.lint_violations,
            format_v=result.format_violations,
        )
        ctx.exit(1)
        return
 
    if force_commit and gate_exit != 0:
        click.echo(
            f"Warning: quality gate would have failed (exit {gate_exit}), "
            f"bypassed with --force.",
            err=True,
        )
 
    # -- 5. Large-file advisory (G-D6) --
    oversized = _check_large_staged_files(root, staged, octrc)
    for rel, size in oversized:
        mb = size / (1024 * 1024)
        click.echo(
            f"Warning: {rel} is {mb:.1f} MB. Consider .gitignore or Git LFS.",
            err=True,
        )
 
    # -- 6. Advisory checks --
    _advisory_checks(root, staged, message, err=True)
 
    # -- 7. --no-verify warning --
    if no_verify:
        click.echo(
            "Skipping git hooks (--no-verify). "
            "Note: oct git commit still runs its own quality checks.",
            err=True,
        )
 
    # -- 7.5 B1.0: pre-commit-call mutation check (catches gate writers) --
    if observe_mode != "off" and pre_snapshot:
        gate_mutations = _check_unstaged_mutation(
            root, pre_snapshot, staged,
            stage_label="quality gate",
            method=observe_method,
        )
        if gate_mutations and observe_mode == "block":
            _emit_observer_block(
                ctx, mutations=gate_mutations,
                message=message, branch=branch,
                profile=effective_profile, json_mode=json_mode,
                staged_count=len(staged),
            )
            return
 
    # -- 8. Execute git commit --
    git_args = ["commit", "-m", message]
    if no_verify:
        git_args.append("--no-verify")
    git_args.extend(extra_args)
 
    try:
        git_run(git_args, cwd=root)
    except Exception as exc:
        err_msg = str(exc)
        if json_mode:
            _emit_commit_json(
                ctx, exit_code=4, message=message, branch=branch,
                profile=effective_profile,
                errors=[f"git commit failed: {err_msg}"],
            )
        else:
            click.echo(f"git commit failed: {err_msg}", err=True)
        _populate_commit_audit(
            ctx, branch=branch, exit_code=4,
            checks_passed=False, staged_count=len(staged),
        )
        ctx.exit(4)
        return
 
    # -- 8.5 B1.0: post-commit mutation check (catches git-hook writers) --
    if observe_mode != "off" and pre_snapshot:
        hook_mutations = _check_unstaged_mutation(
            root, pre_snapshot, staged,
            stage_label="git-hook",
            method=observe_method,
        )
        if hook_mutations and observe_mode == "block":
            _emit_observer_block(
                ctx, mutations=hook_mutations,
                message=message, branch=branch,
                profile=effective_profile, json_mode=json_mode,
                staged_count=len(staged),
            )
            return
 
    # -- 9. Diagnostics impact (G-D5) --
    try:
        from oct.core.git import git_run as _gr
        diff_proc = _gr(
            ["diff", "HEAD~1..HEAD"], cwd=root, check=False,
        )
        impact = analyse_diagnostics_impact(diff_proc.stdout)
        if impact:
            click.echo("Diagnostics Impact:", err=True)
            for line in impact:
                click.echo(f"  {line}", err=True)
    except Exception:
        pass  # Best-effort; never block.
 
    # -- 10. Success output --
    if json_mode:
        _emit_commit_json(
            ctx, exit_code=0, message=message, branch=branch,
            profile=effective_profile, errors=[],
        )
    else:
        click.echo("Commit successful.")
 
    psr_audit = (
        _build_per_subproject_audit(per_sp_results, per_sp_meta)
        if use_fanout else None
    )
    _populate_commit_audit(
        ctx, branch=branch, exit_code=0,
        checks_passed=True, staged_count=len(staged),
        per_subproject_results=psr_audit,
    )
 
 
def _emit_commit_json(
    ctx, *, exit_code: int, message: str, branch: str | None,
    profile: str, errors: list[str] | None = None,
    secrets: list | None = None,
) -> None:
    """Emit JSON output for the commit command."""
    output: dict = {
        "exit_code": exit_code,
        "message": message,
        "branch": branch,
        "profile": profile,
        "errors": errors or [],
    }
    if secrets:
        output["secrets_findings"] = [
            {"path": s[0], "line": s[1], "reason": s[2]} for s in secrets
        ]
    click.echo(json.dumps(output, indent=2, ensure_ascii=False))
 
 
def _populate_commit_audit(
    ctx, *, branch: str | None, exit_code: int,
    checks_passed: bool, staged_count: int,
    secrets: bool = False, lint_v: int = 0, format_v: int = 0,
    tracked_mutations: list[str] | None = None,
    per_subproject_results: list[dict] | None = None,
) -> None:
    """Populate the audit record for oct git commit."""
    record = ctx.obj.get("_audit_record", AuditRecord())
    record.timestamp = record.timestamp or _utc_now_iso()
    record.command = "oct git commit"
    record.branch = branch
    record.staged_files = staged_count
    record.checks_passed = checks_passed
    record.secrets_detected = secrets
    record.lint_violations = lint_v
    record.format_violations = format_v
    record.exit_code = exit_code
    if tracked_mutations:
        # AuditRecord schema has no field for this yet; stash on the
        # ctx.obj for downstream serialisation. Test code reads it from
        # the JSONL line directly.
        ctx.obj["_tracked_mutations"] = list(tracked_mutations)
    if per_subproject_results is not None:
        # F1.3: per-subproject fan-out results. Stashed on ctx.obj for
        # downstream JSONL serialisation; AuditRecord doesn't yet
        # carry the field but the audit writer picks it up from
        # ctx.obj if present.
        ctx.obj["_per_subproject_results"] = list(per_subproject_results)
    ctx.obj["_audit_record"] = record
 
 
# =====================================================================
# F1.3 fan-out helpers
# =====================================================================
 
 
#: Bucket name for staged files that match no subproject and no
#: ``workspace_files`` glob. Plan 208: such files were previously DROPPED and
#: committed without any lint/format/secrets scan - a G-D4 bypass. They now
#: land here and are gated under ``workspace_profile`` like the workspace
#: bucket, so the three buckets PARTITION the staged set: nothing is unscanned.
UNSCOPED_BUCKET = "__unscoped__"
 
 
def _group_files_by_subproject(
    staged: list[Path], workspace,
) -> dict[str, list[Path]]:
    """Group *staged* paths into per-subproject + workspace + unscoped buckets.
 
    Files matched by a subproject's path go to that subproject's bucket; files
    not under any subproject AND matching ``workspace.workspace_files`` go to
    the ``"workspace"`` bucket; **every remaining file goes to the
    :data:`UNSCOPED_BUCKET`** (plan 208).
 
    The returned buckets PARTITION *staged* - the union of all bucket lists
    equals the input set, with no duplicates. This is the invariant the commit
    gate relies on so that no staged file is ever committed unscanned; the
    caller asserts it and fails closed. Previously a file matching neither a
    subproject nor ``workspace_files`` - or one that failed ``relative_to`` -
    was silently discarded, and the commit fan-out then never scanned it.
    """
    from oct.core.workspace import (
        assign_to_subproject,
        matches_workspace_files,
    )
    workspace_root_resolved = workspace.root.resolve()
    buckets: dict[str, list[Path]] = {
        sp.name: [] for sp in workspace.subprojects
    }
    buckets["workspace"] = []
    buckets[UNSCOPED_BUCKET] = []
    for path in staged:
        sp = assign_to_subproject(path, workspace)
        if sp is not None:
            buckets[sp.name].append(path)
            continue
        try:
            rel = path.resolve().relative_to(
                workspace_root_resolved,
            ).as_posix()
        except ValueError:
            # Outside the workspace root entirely - cannot be a declared
            # workspace file, but it IS staged, so it must still be scanned.
            # Fail-closed: route to the unscoped bucket rather than drop.
            buckets[UNSCOPED_BUCKET].append(path)
            continue
        if matches_workspace_files(rel, workspace.workspace_files):
            buckets["workspace"].append(path)
        else:
            buckets[UNSCOPED_BUCKET].append(path)
    return buckets
 
 
@dataclass
class FanoutGateOutcome:
    """Result of one workspace fan-out quality-gate pass (plan 213 / D-A).
 
    ``result`` is the merged :class:`~oct.git.quality_gate.QualityGateResult`
    across every non-empty bucket; ``gate_exit`` is the worst per-bucket
    policy verdict. ``per_bucket_results`` / ``per_bucket_meta`` are keyed by
    bucket name - a subproject name, ``"workspace"``, or
    :data:`UNSCOPED_BUCKET`.
 
    ``unbucketed`` is the fail-closed signal. When it is non-empty NO gate
    ran: the buckets did not partition the input, so at least one file would
    have been passed through unscanned. Callers MUST refuse the operation
    and surface the listed paths. Rendering is left to the caller because
    the two callers differ: ``oct git commit`` refuses the commit, while
    ``oct git check`` reports a non-zero verdict.
    """
 
    result: object
    gate_exit: int
    per_bucket_results: dict
    per_bucket_meta: dict
    unbucketed: list
 
 
def _run_fanout_gate(
    workspace,
    files: list[Path],
    *,
    cli_profile: str | None,
    branch: str | None,
    fallback_profile: str,
    include_tests: bool = False,
    apply_policy: bool = True,
) -> FanoutGateOutcome:
    """Gate *files* per owning subproject under a workspace manifest.
 
    Plan 213 / D-A: the ONE fan-out implementation. ``oct git commit``
    (F1.3) and ``oct git check`` (plan 213 / D-B) both route through it, so
    the two commands cannot report different verdicts for the same file set
    - which is exactly what they did before, because ``check`` ran a single
    ``run_quality_gate`` pass rooted at the cwd-discovered project root. At
    the workspace root that root has no octrc and matches no subproject, so
    the profile collapsed to ``compact`` (or ``strict`` on a protected
    branch) and ``workspace.workspace_profile`` was never consulted.
 
    Each subproject's slice runs under its own root and its own resolved
    profile (:func:`_resolve_subproject_profile`, the plan-197/209 shared
    resolver). The ``workspace`` bucket (declared ``workspace_files``) and
    the :data:`UNSCOPED_BUCKET` (plan 208 - everything else) both run at
    *cli_profile* → ``workspace.workspace_profile`` → *fallback_profile*.
 
    The G-D4 secrets-always-block invariant is preserved by construction:
    the buckets partition *files*, every bucket is scanned, and the merged
    result concatenates every bucket's findings. The partition is verified
    rather than trusted - see :attr:`FanoutGateOutcome.unbucketed`.
 
    Parameters
    ----------
    workspace
        Loaded :class:`~oct.core.workspace.Workspace` manifest.
    files
        The full file set to gate (staged, changed, or all).
    cli_profile
        Explicit ``--profile`` override, or ``None``.
    branch
        Current branch, for the protected-branch override.
    fallback_profile
        Profile used for the workspace/unscoped buckets when neither
        *cli_profile* nor the manifest supplies one. Callers pass their
        own single-project resolution so behaviour is unchanged when a
        manifest omits ``workspace_profile``.
    include_tests
        Forwarded per bucket. ``oct git commit`` never sets it.
    apply_policy
        When False every bucket's policy verdict is skipped and
        ``gate_exit`` is 0 (``oct git commit --force``). Secrets findings
        still flow through ``result`` - force never suppresses them.
    """
    func = "_run_fanout_gate"
    from oct.git.policy import apply_profile_policy
    from oct.git.quality_gate import run_quality_gate_on_files
 
    groups = _group_files_by_subproject(files, workspace)
    # G-D4 fail-closed backstop (plan 208): the buckets MUST partition the
    # input set, or a file would slip past the secrets scan. Verify it here
    # rather than trust it - a future bucketing bug must abort the caller,
    # not silently drop a file.
    bucketed = {p for bucket in groups.values() for p in bucket}
    unbucketed = [p for p in files if p not in bucketed]
    if unbucketed:
        _dbg(
            "GIT", func,
            f"partition invariant violated; {len(unbucketed)} file(s) "
            f"reached no bucket",
            1, override=True,
        )
        return FanoutGateOutcome(
            result=_merge_quality_gate_results({}),
            gate_exit=2,
            per_bucket_results={},
            per_bucket_meta={},
            unbucketed=unbucketed,
        )
 
    per_bucket_results: dict[str, "QualityGateResult"] = {}
    per_bucket_meta: dict[str, dict] = {}
 
    for sp in workspace.subprojects:
        sp_files = groups.get(sp.name, [])
        if not sp_files:
            continue
        sp_octrc = load_octrc_safe(sp.path)
        sp_profile = _resolve_subproject_profile(
            sp, sp_octrc, cli_profile, branch,
        )
        per_bucket_results[sp.name] = run_quality_gate_on_files(
            project_root=sp.path, files=sp_files,
            profile=sp_profile, include_tests=include_tests, fix=False,
        )
        per_bucket_meta[sp.name] = {
            "profile": sp_profile, "files": len(sp_files),
        }
 
    ws_profile = (
        cli_profile or workspace.workspace_profile or fallback_profile
    )
    for bucket_name in ("workspace", UNSCOPED_BUCKET):
        bucket_files = groups.get(bucket_name, [])
        if not bucket_files:
            continue
        per_bucket_results[bucket_name] = run_quality_gate_on_files(
            project_root=workspace.root, files=bucket_files,
            profile=ws_profile, include_tests=include_tests, fix=False,
        )
        per_bucket_meta[bucket_name] = {
            "profile": ws_profile, "files": len(bucket_files),
        }
 
    result = _merge_quality_gate_results(per_bucket_results)
    if apply_policy:
        gate_exit = max(
            (
                apply_profile_policy(per_bucket_meta[name]["profile"], r)
                for name, r in per_bucket_results.items()
            ),
            default=0,
        )
    else:
        gate_exit = 0
 
    _dbg(
        "GIT", func,
        f"buckets={list(per_bucket_results)} gate_exit={gate_exit}",
        3,
    )
    return FanoutGateOutcome(
        result=result,
        gate_exit=gate_exit,
        per_bucket_results=per_bucket_results,
        per_bucket_meta=per_bucket_meta,
        unbucketed=[],
    )
 
 
def _resolve_subproject_profile(
    subproject, sp_octrc: dict | None,
    cli_override: str | None, branch: str | None,
) -> str:
    """Resolve the lint profile for *subproject* under workspace fan-out.
 
    Plan 197: this is now a thin adapter over the ONE resolver,
    :func:`oct.git.policy.resolve_effective_profile`. Fan-out and
    single-project resolution therefore cannot drift apart - the
    property is pinned by
    ``tests/tests_git/test_profile_resolution_workspace.py``.
 
    Precedence (highest first), unchanged in shape but now shared:
 
    1. **Protected-branch override** - force ``"strict"``. Pre-197 this
       path only honoured an EXPLICIT ``git.protected_branches`` list,
       so a subproject without one was gated at its manifest profile
       even on ``master``; the shared resolver applies
       :data:`~oct.git.policy.DEFAULT_PROTECTED_BRANCHES`
       (``main``/``master``) as the fallback, matching what
       single-project mode has always done. Plan 209: an explicit
       ``protected_branches`` list only OPTS OUT of this override for a
       ``nested_git`` subproject (a foreign nested repo, e.g.
       ``oc_explorer``); every other subproject can only add to the
       default protected set, never remove ``main``/``master`` from it.
    2. **CLI ``--profile``** - explicit user override.
    3. **Manifest ``subproject.profile``** - workspace-declared
       per-subproject default.
    4. **``sp_octrc.git.pre_commit_profile``** - newly honoured here;
       single-project mode has always read it.
    5. **``sp_octrc.linter.profile``** - the subproject's own
       ``.octrc.json`` setting.
    6. **``"compact"``** - final fallback
       (:data:`~oct.git.policy.DEFAULT_PROFILE`).
    """
    func = "_resolve_subproject_profile"
    from oct.git.policy import resolve_effective_profile
 
    profile = resolve_effective_profile(
        branch, sp_octrc, cli_override,
        manifest_profile=getattr(subproject, "profile", None),
        allow_optout=getattr(subproject, "nested_git", False),
    )
    _dbg(
        "GIT", func,
        f"subproject={getattr(subproject, 'name', '?')!r} "
        f"branch={branch!r} -> {profile}",
        3,
    )
    return profile
 
 
def _merge_quality_gate_results(
    per_sp: dict[str, "QualityGateResult"],
):
    """Aggregate per-subproject results into a single QualityGateResult.
 
    Counters (lint, format, files_checked, test_failures) are summed.
    Secrets findings are concatenated (preserves the
    ``(path, line, reason)`` tuples the G-D4 invariant uses to surface
    every leak). per_file is concatenated. duration_ms is the max
    (forward-compatible with parallel execution; v1 runs serially so
    it's effectively the longest single bucket).
    """
    from oct.git.quality_gate import QualityGateResult
    if not per_sp:
        return QualityGateResult()
    files_checked = 0
    lint_violations = 0
    format_violations = 0
    test_failures = 0
    duration_ms = 0
    secrets_findings: list = []
    per_file: list = []
    for r in per_sp.values():
        files_checked += r.files_checked
        lint_violations += r.lint_violations
        format_violations += r.format_violations
        test_failures += r.test_failures
        duration_ms = max(duration_ms, r.duration_ms)
        secrets_findings.extend(r.secrets_findings)
        per_file.extend(r.per_file)
    return QualityGateResult(
        files_checked=files_checked,
        lint_violations=lint_violations,
        format_violations=format_violations,
        secrets_findings=secrets_findings,
        test_failures=test_failures,
        duration_ms=duration_ms,
        per_file=per_file,
    )
 
 
def _build_per_subproject_audit(
    per_sp_results: dict, per_sp_meta: dict,
) -> list[dict]:
    """Shape the per-subproject results dict for the audit record."""
    out: list[dict] = []
    for name, r in per_sp_results.items():
        meta = per_sp_meta.get(name, {})
        out.append({
            "name": name,
            "profile": meta.get("profile", "compact"),
            "files": meta.get("files", r.files_checked),
            "lint_violations": r.lint_violations,
            "format_violations": r.format_violations,
            "secrets": len(r.secrets_findings),
        })
    return out
 
 
def _resolve_observe_method(octrc: dict | None) -> str:
    """Return ``"git"`` / ``"hash"`` for the B1.0 observer (plan 242).
 
    Read from ``.octrc.json -> git.observe_unstaged_mutation_method``.
    An absent key resolves to ``"git"``, the git-native detection that
    costs about 0.2 s per pass instead of the content walk's 6 to
    188 s. ``"hash"`` selects the walk for a paranoid run.
 
    Fails closed: any other value raises :class:`ValueError` naming the
    key, and the caller refuses the commit with exit code 2 before a
    snapshot is taken. A typo must never silently pick a method - in
    either direction. Independent of the mode resolved by
    :func:`_resolve_observe_mode`; the two keys do not interact.
    """
    from oct.git.commit_observer import METHOD_GIT, OBSERVE_METHODS
 
    raw = None
    if isinstance(octrc, dict):
        git_cfg = octrc.get("git")
        if isinstance(git_cfg, dict):
            raw = git_cfg.get("observe_unstaged_mutation_method")
    if raw is None:
        return METHOD_GIT
    if isinstance(raw, str) and raw in OBSERVE_METHODS:
        return raw
    raise ValueError(
        f"Invalid octrc value for git.observe_unstaged_mutation_method: "
        f"{raw!r}. Expected one of {', '.join(OBSERVE_METHODS)}."
    )
 
 
def _resolve_observe_mode(
    octrc: dict | None, ignore_unstaged_mutation: bool,
) -> str:
    """Return ``"off"`` / ``"warn"`` / ``"block"`` for the B1.0 observer.
 
    Precedence (highest first): ``--ignore-unstaged-mutation`` flag
    forces ``"warn"``; ``.octrc.json -> git.observe_unstaged_mutation``
    sets the persistent default; final fallback is ``"warn"``.
    """
    if ignore_unstaged_mutation:
        return "warn"
    if isinstance(octrc, dict):
        git_cfg = octrc.get("git")
        if isinstance(git_cfg, dict):
            raw = git_cfg.get("observe_unstaged_mutation")
            if raw is False or raw == "off":
                return "off"
            if raw in ("warn", "block"):
                return raw
    return "warn"
 
 
def _safe_relative(path: Path, root: Path) -> str | None:
    """Return *path* relative to *root* as a forward-slash string, or
    ``None`` when *path* is outside *root*.
    """
    try:
        rel = path.resolve().relative_to(root.resolve())
    except ValueError:
        return None
    return str(rel).replace("\\", "/")
 
 
def _check_unstaged_mutation(
    root: Path, pre_snapshot: dict[str, str],
    staged: list[Path], stage_label: str,
    method: str = "hash",
) -> list[str]:
    """Re-snapshot and diff against *pre_snapshot*. Emit a stderr
    warning for each mutated path. Returns the list of changed paths.
 
    *method* is the detection method resolved once per commit (plan
    242) and must match the one that produced *pre_snapshot*. Under
    ``"git"`` the post snapshot additionally hashes every path that
    carried something other than the CLEAN sentinel in *pre_snapshot*,
    so a file that was already dirty when the commit started is still
    compared hash to hash - including when it was reverted to its
    index content, which would otherwise leave git's dirty set and
    read as unchanged.
    """
    func = "_check_unstaged_mutation"
    try:
        from oct.core.git import get_repo_root
        from oct.git.commit_observer import (
            _CLEAN_SENTINEL, diff_snapshots, snapshot_unstaged_tracked,
        )
        repo_root = get_repo_root(root)
        staged_rel: set[str] = set()
        for p in staged:
            r = _safe_relative(p, repo_root)
            if r is not None:
                staged_rel.add(r)
        also_hash = {
            k for k, v in pre_snapshot.items() if v != _CLEAN_SENTINEL
        }
        post = snapshot_unstaged_tracked(
            repo_root, staged_set=staged_rel,
            method=method, also_hash=also_hash,
        )
    except Exception as exc:
        _dbg(
            "GIT", func,
            f"observer post-snapshot failed: {exc!r}; skipping",
            1, override=True,
        )
        return []
 
    mutations = diff_snapshots(pre_snapshot, post)
    if mutations:
        click.echo(
            f"WARNING: {len(mutations)} unstaged tracked file(s) "
            f"mutated during {stage_label}:",
            err=True,
        )
        for rel in mutations:
            click.echo(f"  {rel}", err=True)
    return mutations
 
 
def _emit_observer_block(
    ctx, *, mutations: list[str], message: str, branch: str | None,
    profile: str, json_mode: bool, staged_count: int,
) -> None:
    """Abort the commit with exit code 6 (B1.0 block mode)."""
    click.echo(
        "ABORTED: tracked-file mutation detected during commit pipeline. "
        "Inspect the listed files and re-run with "
        "--ignore-unstaged-mutation to override.",
        err=True,
    )
    if json_mode:
        _emit_commit_json(
            ctx, exit_code=6, message=message, branch=branch,
            profile=profile,
            errors=[
                "Unstaged tracked file mutation detected: "
                + ", ".join(mutations),
            ],
        )
    _populate_commit_audit(
        ctx, branch=branch, exit_code=6,
        checks_passed=False, staged_count=staged_count,
        tracked_mutations=mutations,
    )
    ctx.exit(6)
 
 
def _advisory_checks(
    root: Path, staged: list[Path], message: str, err: bool = True,
) -> None:
    """Print advisory warnings for missing docs files."""
    staged_names = {p.name for p in staged}
    staged_rel = {to_project_relative(p, root) for p in staged}
 
    # Warn if new/deleted .py files without ARCHITECTURE.md staged.
    py_staged = [p for p in staged if p.suffix == ".py"]
    if len(py_staged) > 0 and "docs/ARCHITECTURE.md" not in staged_rel:
        has_arch = (root / "docs" / "ARCHITECTURE.md").is_file()
        if has_arch:
            click.echo(
                "Advisory: ARCHITECTURE.md is not staged. "
                "Consider updating it if you changed module structure.",
                err=err,
            )
 
    # Warn if multi-file commit without CHANGELOG.md staged.
    if len(staged) > 3 and "CHANGELOG.md" not in staged_names:
        if (root / "CHANGELOG.md").is_file():
            click.echo(
                "Advisory: CHANGELOG.md is not staged for this multi-file commit.",
                err=err,
            )
 
 
# =====================================================================
# oct git branch (Phase 4F - G-F1)
# =====================================================================
 
 
@git.command(name="branch")
@click.option("--create", "create_name", default=None,
              help="Create a new branch from HEAD.")
@click.option("--switch", "switch_name", default=None,
              help="Switch to an existing branch.")
@click.option("--json", "json_mode", is_flag=True, help="Emit JSON output.")
@click.pass_context
@audited("oct git branch")
def git_branch_cmd(
    ctx: click.Context,
    create_name: str | None,
    switch_name: str | None,
    json_mode: bool,
) -> None:
    """Manage branches with Option C naming conventions."""
    func = "git_branch_cmd"
    root = get_project_root(ctx)
    _dbg("GIT", func, f"root={root} create={create_name!r} switch={switch_name!r}", 2)
 
    from oct.core.git import (
        branch_exists,
        create_branch,
        has_uncommitted_changes,
        is_git_repo,
        switch_branch,
    )
    from oct.git.policy import validate_branch_name
 
    if not is_git_repo(root):
        click.echo("Not a git repository.", err=True)
        _branch_audit(ctx, exit_code=1)
        ctx.exit(1)
        return
 
    octrc = load_octrc_safe(root)
 
    # -- Mutual exclusivity ------------------------------------------------
    if create_name and switch_name:
        msg = "Error: --create and --switch are mutually exclusive."
        click.echo(msg, err=True)
        if json_mode:
            _emit_branch_json(exit_code=1, errors=[msg])
        _branch_audit(ctx, exit_code=1)
        ctx.exit(1)
        return
 
    if not create_name and not switch_name:
        msg = "Use --create NAME or --switch NAME. Branch listing coming soon."
        click.echo(msg, err=True)
        if json_mode:
            _emit_branch_json(exit_code=1, errors=[msg])
        _branch_audit(ctx, exit_code=1)
        ctx.exit(1)
        return
 
    # -- --create -----------------------------------------------------------
    if create_name:
        valid, val_msg = validate_branch_name(create_name, octrc)
        if not valid:
            click.echo(f"Error: {val_msg}", err=True)
            if json_mode:
                _emit_branch_json(
                    operation="create", branch=create_name,
                    exit_code=1, errors=[val_msg],
                )
            _branch_audit(ctx, exit_code=1, branch=create_name)
            ctx.exit(1)
            return
 
        if branch_exists(root, create_name):
            msg = f"Error: branch '{create_name}' already exists."
            click.echo(msg, err=True)
            if json_mode:
                _emit_branch_json(
                    operation="create", branch=create_name,
                    exit_code=1, errors=[msg],
                )
            _branch_audit(ctx, exit_code=1, branch=create_name)
            ctx.exit(1)
            return
 
        try:
            create_branch(root, create_name)
        except Exception as exc:
            msg = f"Git error: {exc}"
            click.echo(msg, err=True)
            if json_mode:
                _emit_branch_json(
                    operation="create", branch=create_name,
                    exit_code=2, errors=[msg],
                )
            _branch_audit(ctx, exit_code=2, branch=create_name)
            ctx.exit(2)
            return
 
        if json_mode:
            _emit_branch_json(
                operation="create", branch=create_name, exit_code=0,
            )
        else:
            click.echo(f"Created branch '{create_name}' (not switched to it).")
        _branch_audit(ctx, exit_code=0, branch=create_name)
        return
 
    # -- --switch -----------------------------------------------------------
    if switch_name:
        if not branch_exists(root, switch_name):
            msg = f"Error: branch '{switch_name}' does not exist."
            click.echo(msg, err=True)
            if json_mode:
                _emit_branch_json(
                    operation="switch", branch=switch_name,
                    exit_code=1, errors=[msg],
                )
            _branch_audit(ctx, exit_code=1, branch=switch_name)
            ctx.exit(1)
            return
 
        if has_uncommitted_changes(root):
            msg = (
                "Error: uncommitted changes would be lost. "
                "Commit or stash them first."
            )
            click.echo(msg, err=True)
            if json_mode:
                _emit_branch_json(
                    operation="switch", branch=switch_name,
                    exit_code=1, errors=[msg],
                )
            _branch_audit(ctx, exit_code=1, branch=switch_name)
            ctx.exit(1)
            return
 
        try:
            switch_branch(root, switch_name)
        except Exception as exc:
            msg = f"Git error: {exc}"
            click.echo(msg, err=True)
            if json_mode:
                _emit_branch_json(
                    operation="switch", branch=switch_name,
                    exit_code=2, errors=[msg],
                )
            _branch_audit(ctx, exit_code=2, branch=switch_name)
            ctx.exit(2)
            return
 
        if json_mode:
            _emit_branch_json(
                operation="switch", branch=switch_name, exit_code=0,
            )
        else:
            click.echo(f"Switched to branch '{switch_name}'.")
        _branch_audit(ctx, exit_code=0, branch=switch_name)
        return
 
 
def _emit_branch_json(
    *,
    operation: str = "",
    branch: str = "",
    exit_code: int = 0,
    errors: list[str] | None = None,
) -> None:
    """Emit JSON for ``oct git branch`` to stdout."""
    click.echo(json.dumps({
        "operation": operation,
        "branch": branch,
        "exit_code": exit_code,
        "errors": errors or [],
    }, indent=2))
 
 
def _branch_audit(
    ctx: click.Context,
    *,
    exit_code: int,
    branch: str = "",
) -> None:
    """Populate the audit record for ``oct git branch``."""
    record = ctx.obj.get("_audit_record", AuditRecord())
    record.timestamp = record.timestamp or _utc_now_iso()
    record.command = "oct git branch"
    record.branch = branch or None
    record.checks_passed = exit_code == 0
    record.exit_code = exit_code
    ctx.obj["_audit_record"] = record
 
 
# =====================================================================
# oct git add (Phase 4G - staging command)
# =====================================================================
 
 
def _emit_add_json(
    *,
    operation: str,
    paths_staged: list[str],
    paths_skipped: list[dict],
    post_check: dict | None,
    exit_code: int,
    errors: list[str] | None = None,
) -> None:
    """Emit JSON for ``oct git add`` to stdout."""
    click.echo(json.dumps({
        "operation": operation,
        "paths_staged": paths_staged,
        "paths_skipped": paths_skipped,
        "post_check": post_check,
        "exit_code": exit_code,
        "errors": errors or [],
    }, indent=2))
 
 
def _add_audit(
    ctx: click.Context,
    *,
    exit_code: int,
    staged_count: int = 0,
) -> None:
    """Populate the audit record for ``oct git add``."""
    record = ctx.obj.get("_audit_record", AuditRecord())
    record.timestamp = record.timestamp or _utc_now_iso()
    record.command = "oct git add"
    record.staged_files = staged_count
    record.checks_passed = exit_code == 0
    record.exit_code = exit_code
    ctx.obj["_audit_record"] = record
 
 
def _post_check_summary(
    root: Path,
    octrc: dict | None,
) -> dict | None:
    """Run the staged-only quality gate and return a summary dict.
 
    Honours ``git.add_post_check`` (default ``True``). Returns ``None``
    when disabled.
 
    Plan 285 (OCTGIT-06): at a workspace root the staged set is gated
    through :func:`_run_fanout_gate` - the ONE fan-out implementation
    ``oct git check`` (plan 213 / D-B) and ``oct git commit`` (F1.3)
    already share - so the advisory and the gate cannot report different
    verdicts for the same staged set. They did before: this ran a single
    :func:`~oct.git.quality_gate.run_quality_gate` rooted at the
    invocation root, and at a workspace root that root has no octrc and
    matches no subproject, so the profile collapsed to ``compact``, every
    subproject-relative identity header failed the header rule against
    the workspace-relative path, and the header fixer counted as the
    format issue. A clean file therefore drew ``FAIL (lint=1 format=1)``
    from ``add`` while ``check --staged-only`` reported ``0/0``.
 
    In workspace mode the returned ``profile`` is the ``(bucket,
    profile)`` list the fan-out actually used, so the JSON says which
    profile gated which file, and ``unbucketed`` carries the fail-closed
    signal: when it is non-empty NO gate ran and the verdict is 2, as it
    is for the commit and check paths. Invocation from inside a
    subproject, or in a non-workspace repo, is byte-identical to pre-285
    - :func:`_maybe_load_workspace` returns ``(None, False)`` in both
    cases - and so is the degenerate empty-staged-set case, which has no
    buckets to name.
 
    The advisory remains advisory: ``oct git add`` still exits 0 over a
    FAIL verdict (the plan-208 contract). What changes is only whether
    the verdict is true.
    """
    func = "_post_check_summary"
    if isinstance(octrc, dict):
        git_cfg = octrc.get("git")
        if isinstance(git_cfg, dict):
            raw = git_cfg.get("add_post_check")
            if isinstance(raw, bool) and not raw:
                _dbg("GIT", func, "post_check disabled by octrc", 3)
                return None
    try:
        from oct.core.git import current_branch
        from oct.git.policy import resolve_profile_for_project
        from oct.git.quality_gate import run_quality_gate
    except ImportError:
        _dbg("GIT", func, "quality_gate import failed; skipping", 1, override=True)
        return None
    try:
        branch = current_branch(root)
    except Exception:
        branch = None
    # Plan 197: manifest-aware, so the post-add summary reports the same
    # profile the commit gate will actually apply.
    profile = resolve_profile_for_project(root, branch, octrc, None)
 
    # Plan 285: one decision, made where the other paths make it. The
    # fan-out owns the per-subproject roots and profiles; nothing about
    # bucketing, profile resolution or the secrets pass is re-derived
    # here. The G-D4 invariant crosses buckets unchanged, because the
    # merged result concatenates every bucket's findings and the printer
    # still gives SECRETS DETECTED precedence over a lint/format FAIL.
    workspace, in_workspace_mode = _maybe_load_workspace(root)
    if in_workspace_mode and workspace is not None and workspace.subprojects:
        try:
            from oct.git.quality_gate import _resolve_scope
            staged = _resolve_scope(root, "staged")
        except Exception as exc:
            _dbg(
                "GIT", func,
                f"staged scope resolution raised {exc!r}; falling back to "
                f"the single-project summary",
                1, override=True,
            )
            staged = []
        if staged:
            try:
                outcome = _run_fanout_gate(
                    workspace, staged,
                    cli_profile=None, branch=branch,
                    fallback_profile=profile, include_tests=False,
                )
            except Exception as exc:
                _dbg(
                    "GIT", func,
                    f"fan-out gate raised {exc!r}; skipping", 1,
                    override=True,
                )
                return None
            merged = outcome.result
            _dbg(
                "GIT", func,
                f"fan-out summary buckets={list(outcome.per_bucket_meta)} "
                f"verdict={outcome.gate_exit}",
                3,
            )
            return {
                "profile": [
                    [name, meta.get("profile", profile)]
                    for name, meta in outcome.per_bucket_meta.items()
                ],
                "lint_violations": merged.lint_violations,
                "format_violations": merged.format_violations,
                "secrets_findings": len(merged.secrets_findings),
                "exit_code": outcome.gate_exit,
                "unbucketed": [
                    to_project_relative(p, root) for p in outcome.unbucketed
                ],
            }
 
    try:
        result = run_quality_gate(
            project_root=root, scope="staged",
            profile=profile, fix=False, include_tests=False,
        )
    except Exception as exc:
        _dbg("GIT", func, f"quality_gate raised {exc!r}; skipping", 1, override=True)
        return None
    # Plan 208 (D2): the real gate verdict is what `apply_profile_policy`
    # computes, NOT `result.exit_code` - `run_quality_gate` leaves that field at
    # its dataclass default 0, so reading it made the advisory print PASS over a
    # staged secret. Compute the verdict the commit gate would actually apply.
    from oct.git.policy import apply_profile_policy
    verdict = apply_profile_policy(profile, result)
    return {
        "profile": profile,
        "lint_violations": result.lint_violations,
        "format_violations": result.format_violations,
        "secrets_findings": len(result.secrets_findings),
        "exit_code": verdict,
    }
 
 
@git.command(name="add")
@click.argument("paths", nargs=-1, type=click.Path())
@click.option(
    "--patch", "-p", "patch_mode", is_flag=True,
    help="Interactive hunk-level staging (passthrough to git add -p).",
)
@click.option(
    "--update", "-u", "update_mode", is_flag=True,
    help="Stage all already-tracked modified files (-u).",
)
@click.option(
    "--all", "-A", "all_mode", is_flag=True,
    help="Stage all changes within project scope (-A).",
)
@click.option(
    "--dry-run", is_flag=True,
    help="Show what would be staged without staging.",
)
@click.option(
    "--json", "json_mode", is_flag=True, help="Emit JSON output.",
)
@click.pass_context
@audited("oct git add")
def git_add_cmd(
    ctx: click.Context,
    paths: tuple[str, ...],
    patch_mode: bool,
    update_mode: bool,
    all_mode: bool,
    dry_run: bool,
    json_mode: bool,
) -> None:
    """Stage files for commit (whole-file or interactive hunk)."""
    func = "git_add_cmd"
    root = get_project_root(ctx)
    _dbg(
        "GIT", func,
        f"root={root} paths={paths} patch={patch_mode} "
        f"update={update_mode} all={all_mode} dry_run={dry_run}",
        2,
    )
 
    from oct.core.git import (
        get_repo_root,
        is_git_repo,
        stage_paths,
        stage_paths_interactive,
    )
 
    if not is_git_repo(root):
        click.echo("Not a git repository.", err=True)
        _add_audit(ctx, exit_code=1)
        ctx.exit(1)
        return
 
    repo_root = get_repo_root(root)
    octrc = load_octrc_safe(root)
 
    # -- Mutual exclusivity --------------------------------------------------
    if patch_mode and (update_mode or all_mode):
        msg = "Error: --patch cannot be combined with --update or --all."
        click.echo(msg, err=True)
        if json_mode:
            _emit_add_json(
                operation="patch", paths_staged=[], paths_skipped=[],
                post_check=None, exit_code=1, errors=[msg],
            )
        _add_audit(ctx, exit_code=1)
        ctx.exit(1)
        return
 
    if update_mode and all_mode:
        msg = "Error: --update and --all are mutually exclusive."
        click.echo(msg, err=True)
        if json_mode:
            _emit_add_json(
                operation="add", paths_staged=[], paths_skipped=[],
                post_check=None, exit_code=1, errors=[msg],
            )
        _add_audit(ctx, exit_code=1)
        ctx.exit(1)
        return
 
    # -- Resolve and validate paths ------------------------------------------
    root_resolved = root.resolve()
    skipped: list[dict] = []
    resolved_paths: list[Path] = []
    for raw in paths:
        candidate = (Path.cwd() / raw).resolve() \
            if not Path(raw).is_absolute() else Path(raw).resolve()
        if not _is_within_project(candidate, root_resolved):
            skipped.append({
                "path": str(raw), "reason": "outside project scope",
            })
            continue
        resolved_paths.append(candidate)
 
    if skipped and not json_mode:
        for entry in skipped:
            click.echo(
                f"Skipping {entry['path']} ({entry['reason']}).",
                err=True,
            )
 
    # -- Empty selection guard -----------------------------------------------
    if not resolved_paths and not (update_mode or all_mode or patch_mode):
        msg = (
            "Error: no paths to stage. Provide PATHS, "
            "or use --update / --all / --patch."
        )
        click.echo(msg, err=True)
        if json_mode:
            _emit_add_json(
                operation="add", paths_staged=[], paths_skipped=skipped,
                post_check=None, exit_code=1, errors=[msg],
            )
        _add_audit(ctx, exit_code=1)
        ctx.exit(1)
        return
 
    # -- Dry-run -------------------------------------------------------------
    if dry_run:
        operation = "dry-run"
        rel_strings = [
            to_project_relative(p, root) for p in resolved_paths
        ]
        if update_mode or all_mode:
            # Compute what git would stage for -u / -A by inspecting status.
            from oct.core.git import git_run as _git_run
            argv = ["status", "--porcelain"]
            proc = _git_run(argv, cwd=repo_root)
            for line in proc.stdout.splitlines():
                # Porcelain format: "XY path"; XY are status flags.
                if len(line) < 4:
                    continue
                xy = line[:2]
                rel = line[3:].strip()
                if update_mode and xy[1] in (" ", ""):
                    # Tracked + unmodified - git add -u skips it.
                    continue
                if update_mode and xy[1] == "?":
                    continue
                full = (repo_root / rel).resolve()
                if not _is_within_project(full, root_resolved):
                    continue
                rel_strings.append(to_project_relative(full, root))
        if json_mode:
            _emit_add_json(
                operation=operation, paths_staged=rel_strings,
                paths_skipped=skipped, post_check=None, exit_code=0,
            )
        else:
            click.echo("Would stage:")
            for s in rel_strings:
                click.echo(f"  {s}")
        _add_audit(ctx, exit_code=0, staged_count=len(rel_strings))
        return
 
    # -- Patch mode ----------------------------------------------------------
    if patch_mode:
        if json_mode:
            click.echo(
                "Note: --patch is interactive; --json output is limited.",
                err=True,
            )
        try:
            rc = stage_paths_interactive(repo_root, resolved_paths)
        except Exception as exc:
            msg = f"Git error: {exc}"
            click.echo(msg, err=True)
            if json_mode:
                _emit_add_json(
                    operation="patch", paths_staged=[],
                    paths_skipped=skipped, post_check=None,
                    exit_code=2, errors=[msg],
                )
            _add_audit(ctx, exit_code=2)
            ctx.exit(2)
            return
        if rc != 0:
            _add_audit(ctx, exit_code=rc)
            ctx.exit(rc)
            return
        post = _post_check_summary(root, octrc)
        if json_mode:
            _emit_add_json(
                operation="patch", paths_staged=[],
                paths_skipped=skipped, post_check=post, exit_code=0,
            )
        else:
            _print_post_check_advisory(post)
        _add_audit(ctx, exit_code=0)
        return
 
    # -- Whole-file staging --------------------------------------------------
    # L2: capture pre-stage index so we can decide, on a non-zero git
    # exit, whether anything actually got staged. Git emits exit 1
    # alongside the stderr "ignored path" warning when staging tracked
    # files inside an ignored directory; the operation succeeds but the
    # exit code lies. We mirror plain-git semantics by treating
    # "warning + something staged" as success.
    from oct.core.git import git_staged_files
    pre_index = {p.resolve() for p in git_staged_files(repo_root)}
 
    git_warning: str | None = None
    try:
        stage_paths(
            repo_root, resolved_paths,
            update=update_mode, all_=all_mode,
        )
    except Exception as exc:
        # Re-query the index. If it grew, this was a "warning"-class
        # git failure (exit 1 + something staged). Treat as success
        # with the stderr message preserved.
        post_index = {p.resolve() for p in git_staged_files(repo_root)}
        newly_staged = post_index - pre_index
        if newly_staged:
            git_warning = str(exc)
            click.echo(
                f"Note: git emitted a warning during staging but "
                f"{len(newly_staged)} file(s) were staged successfully:\n"
                f"  {git_warning}",
                err=True,
            )
        else:
            msg = f"Git error: {exc}"
            click.echo(msg, err=True)
            if json_mode:
                _emit_add_json(
                    operation="add", paths_staged=[],
                    paths_skipped=skipped,
                    post_check=None, exit_code=2, errors=[msg],
                )
            _add_audit(ctx, exit_code=2)
            ctx.exit(2)
            return
 
    # Compute what was actually staged for reporting.
    staged_now = git_staged_files(repo_root)
    staged_within = [
        p for p in staged_now if _is_within_project(p, root_resolved)
    ]
    staged_strings = [to_project_relative(p, root) for p in staged_within]
 
    operation = "all" if all_mode else "update" if update_mode else "add"
    post = _post_check_summary(root, octrc)
 
    if json_mode:
        _emit_add_json(
            operation=operation, paths_staged=staged_strings,
            paths_skipped=skipped, post_check=post, exit_code=0,
        )
    else:
        if resolved_paths:
            for p in resolved_paths:
                click.echo(f"Staged {to_project_relative(p, root)}")
        elif update_mode:
            click.echo(
                f"Staged {len(staged_strings)} tracked modified file(s).",
            )
        elif all_mode:
            click.echo(
                f"Staged {len(staged_strings)} file(s) (including untracked).",
            )
        _print_post_check_advisory(post)
 
    _add_audit(ctx, exit_code=0, staged_count=len(staged_strings))
 
 
def _print_post_check_advisory(post: dict | None) -> None:
    """Print a single-line summary of the post-stage quality gate result.
 
    Plan 285: the three shapes and their precedence are unchanged, and in
    single-project mode the bytes are unchanged too. In workspace mode
    ``post["profile"]`` is the ``(bucket, profile)`` list the fan-out used
    rather than one profile name, and the line appends it in one short
    parenthesis - ``Quality gate: PASS (oc_chat strict; workspace
    compact)`` - so a verdict says where it looked and an operator can
    see at a glance that their file was gated at its own profile.
 
    A non-empty ``unbucketed`` replaces that parenthesis with the files
    no bucket claimed: no gate ran over them, so the counts beside the
    verdict describe nothing and the names are the only useful detail.
    """
    if post is None:
        return
    buckets = post.get("profile")
    where = ""
    if isinstance(buckets, list):
        unbucketed = post.get("unbucketed") or []
        if unbucketed:
            where = (
                f" (assigned to no gate scope, so NOT checked: "
                f"{', '.join(unbucketed)})"
            )
        elif buckets:
            named = "; ".join(f"{name} {prof}" for name, prof in buckets)
            where = f" ({named})"
    if post["exit_code"] == 0:
        click.echo(f"Quality gate: PASS{where}", err=True)
    elif post["secrets_findings"]:
        # Plan 208 (D2): secrets are the non-bypassable class - call them out
        # first and unambiguously so a staged credential can never read as a
        # generic lint failure, let alone as PASS. Plan 285: a finding in ANY
        # bucket reaches the merged result, so this precedence holds across
        # the fan-out exactly as it held over the single pass.
        click.echo(
            f"Quality gate: SECRETS DETECTED "
            f"({post['secrets_findings']} finding(s); "
            f"lint={post['lint_violations']} "
            f"format={post['format_violations']}){where}. "
            f"Run `oct git check --staged-only` for details.",
            err=True,
        )
    else:
        click.echo(
            f"Quality gate: FAIL "
            f"(lint={post['lint_violations']} "
            f"format={post['format_violations']} "
            f"secrets={post['secrets_findings']}){where}. "
            f"Run `oct git check --staged-only` for details.",
            err=True,
        )
 
 
# =====================================================================
# oct git ignore (Phase 4G - .gitignore manager)
# =====================================================================
 
 
def _emit_ignore_json(
    *,
    operation: str,
    patterns_added: list[str] | None = None,
    patterns_removed: list[str] | None = None,
    oct_managed: list[str] | None = None,
    other: list[str] | None = None,
    exit_code: int = 0,
    errors: list[str] | None = None,
) -> None:
    """Emit JSON for ``oct git ignore`` to stdout."""
    payload: dict = {
        "operation": operation,
        "exit_code": exit_code,
        "errors": errors or [],
    }
    if patterns_added is not None:
        payload["patterns_added"] = patterns_added
    if patterns_removed is not None:
        payload["patterns_removed"] = patterns_removed
    if oct_managed is not None:
        payload["oct_managed"] = oct_managed
    if other is not None:
        payload["other"] = other
    click.echo(json.dumps(payload, indent=2))
 
 
def _ignore_audit(
    ctx: click.Context,
    *,
    exit_code: int,
) -> None:
    """Populate the audit record for ``oct git ignore``."""
    record = ctx.obj.get("_audit_record", AuditRecord())
    record.timestamp = record.timestamp or _utc_now_iso()
    record.command = "oct git ignore"
    record.checks_passed = exit_code == 0
    record.exit_code = exit_code
    ctx.obj["_audit_record"] = record
 
 
# =====================================================================
# F1.4 ignore --scope helpers
# =====================================================================
 
 
def _resolve_ignore_scope(
    cli_scope: str | None,
    octrc: dict | None,
) -> str:
    """Apply scope precedence: CLI > octrc > 'nearest'.
 
    Mirrors the precedence used by ``oct git status --scope``.
    """
    if cli_scope in ("workspace", "project", "nearest"):
        return cli_scope
    if isinstance(octrc, dict):
        git_cfg = octrc.get("git")
        if isinstance(git_cfg, dict):
            raw = git_cfg.get("ignore_default_scope")
            if raw in ("workspace", "project", "nearest"):
                return raw
    return "nearest"
 
 
def _resolve_ignore_target_dir(
    *,
    project_root: Path,
    repo_root: Path,
    scope: str,
) -> tuple[Path | None, str | None]:
    """Resolve the directory whose ``.gitignore`` should be edited.
 
    Returns ``(target_dir, error_message)`` - exactly one is ``None``.
 
    Scope semantics:
 
    - ``workspace``: walk up from *project_root* to find a workspace
      manifest; the workspace root is the target. Errors when no
      manifest exists above CWD.
    - ``project``: the *project_root* itself.
    - ``nearest``: workspace root if *project_root* IS a workspace
      root (i.e. discovery resolved through the manifest fallback);
      otherwise *project_root* (a subproject or a single project).
    """
    from oct.core.workspace import (
        WORKSPACE_MANIFEST,
        find_workspace_root,
    )
    if scope == "workspace":
        ws_path = find_workspace_root(project_root)
        if ws_path is None:
            return None, (
                "scope=workspace requires a workspace manifest "
                f"({WORKSPACE_MANIFEST}) above CWD; none found"
            )
        return ws_path, None
    if scope == "project":
        return project_root, None
    # scope == "nearest"
    project_root_resolved = project_root.resolve()
    is_at_workspace_root = (
        project_root_resolved
        / WORKSPACE_MANIFEST
    ).is_file()
    if is_at_workspace_root:
        return project_root_resolved, None
    return project_root, None
 
 
# =====================================================================
# oct git restore (Phase 4G-followup L3)
# =====================================================================
 
 
def _restore_audit(ctx: click.Context, *, exit_code: int) -> None:
    """Populate the audit record for ``oct git restore``."""
    record = ctx.obj.get("_audit_record", AuditRecord())
    record.timestamp = record.timestamp or _utc_now_iso()
    record.command = "oct git restore"
    record.checks_passed = exit_code == 0
    record.exit_code = exit_code
    ctx.obj["_audit_record"] = record
 
 
@git.command(name="restore")
@click.argument("paths", nargs=-1, type=click.Path())
@click.option(
    "--staged", "staged_mode", is_flag=True,
    help="Unstage paths from the index (leaves working tree intact).",
)
@click.option(
    "--all", "-A", "all_mode", is_flag=True,
    help="Restore all in-scope paths (requires explicit confirmation when used with --staged).",
)
@click.option(
    "--yes", "skip_prompt", is_flag=True,
    help="Skip the confirmation prompt for --staged --all.",
)
@click.option(
    "--json", "json_mode", is_flag=True, help="Emit JSON output.",
)
@click.pass_context
@audited("oct git restore")
def git_restore_cmd(
    ctx: click.Context,
    paths: tuple[str, ...],
    staged_mode: bool,
    all_mode: bool,
    skip_prompt: bool,
    json_mode: bool,
) -> None:
    """Restore working-tree edits or unstage paths.
 
    Without ``--staged``: discards unstaged working-tree edits in
    *paths*, reverting them to the index (i.e. ``git restore <paths>``).
 
    With ``--staged``: removes *paths* from the index without touching
    the working tree (i.e. ``git restore --staged <paths>``).
 
    With ``--all``: scope is every modified file inside the OCT
    project root. ``--staged --all`` requires confirmation (irreversible
    destruction of all staged content); pass ``--yes`` to skip.
    """
    func = "git_restore_cmd"
    root = get_project_root(ctx)
    _dbg(
        "GIT", func,
        f"root={root} paths={paths} staged={staged_mode} "
        f"all={all_mode} json={json_mode}",
        2,
    )
 
    from oct.core.git import (
        get_repo_root, is_git_repo, restore_paths,
    )
 
    if not is_git_repo(root):
        click.echo("Not a git repository.", err=True)
        _restore_audit(ctx, exit_code=1)
        ctx.exit(1)
        return
 
    repo_root = get_repo_root(root)
    root_resolved = root.resolve()
 
    if not paths and not all_mode:
        msg = "Error: provide PATHS or use --all."
        click.echo(msg, err=True)
        if json_mode:
            click.echo(json.dumps({
                "operation": "restore",
                "paths_restored": [], "exit_code": 1, "errors": [msg],
            }, indent=2))
        _restore_audit(ctx, exit_code=1)
        ctx.exit(1)
        return
 
    # Resolve and scope-filter explicit paths.
    skipped: list[dict] = []
    resolved_paths: list[Path] = []
    for raw in paths:
        candidate = Path(raw)
        if not candidate.is_absolute():
            candidate = (Path.cwd() / candidate).resolve()
        else:
            candidate = candidate.resolve()
        if not _is_within_project(candidate, root_resolved):
            skipped.append(
                {"path": str(raw), "reason": "outside project scope"},
            )
            continue
        resolved_paths.append(candidate)
 
    if skipped and not json_mode:
        for entry in skipped:
            click.echo(
                f"Skipping {entry['path']} ({entry['reason']}).",
                err=True,
            )
 
    # --all mode: enumerate the in-scope candidates from git.
    if all_mode:
        from oct.core.git import git_run as _gr
        if staged_mode:
            argv = ["diff", "--cached", "--name-only"]
        else:
            argv = ["diff", "--name-only"]
        proc = _gr(argv, cwd=repo_root)
        for line in proc.stdout.splitlines():
            rel = line.strip()
            if not rel:
                continue
            full = (repo_root / rel).resolve()
            if _is_within_project(full, root_resolved):
                resolved_paths.append(full)
 
    if not resolved_paths:
        msg = (
            "Nothing to restore (no in-scope paths matched)."
            if all_mode else
            "Nothing to restore."
        )
        if json_mode:
            click.echo(json.dumps({
                "operation": "restore",
                "paths_restored": [], "exit_code": 0,
                "skipped": skipped, "errors": [],
            }, indent=2))
        else:
            click.echo(msg)
        _restore_audit(ctx, exit_code=0)
        return
 
    # Confirmation prompt for --staged --all (irreversibly drops the
    # staged changes for every file in scope).
    if staged_mode and all_mode and not skip_prompt and not json_mode:
        click.echo(
            f"About to UNSTAGE {len(resolved_paths)} file(s). This "
            f"discards every staged change in the project scope and "
            f"cannot be undone. Continue? [y/N] ",
            nl=False, err=True,
        )
        try:
            answer = click.get_text_stream("stdin").readline().strip().lower()
        except (KeyboardInterrupt, OSError):
            answer = ""
        if answer not in ("y", "yes"):
            click.echo("Aborted.", err=True)
            _restore_audit(ctx, exit_code=1)
            ctx.exit(1)
            return
 
    try:
        restore_paths(repo_root, resolved_paths, staged=staged_mode)
    except Exception as exc:
        msg = f"Git error: {exc}"
        click.echo(msg, err=True)
        if json_mode:
            click.echo(json.dumps({
                "operation": "restore",
                "paths_restored": [], "exit_code": 2,
                "errors": [msg],
            }, indent=2))
        _restore_audit(ctx, exit_code=2)
        ctx.exit(2)
        return
 
    rel_strings = [to_project_relative(p, root) for p in resolved_paths]
 
    if json_mode:
        click.echo(json.dumps({
            "operation": "restore",
            "staged": staged_mode,
            "paths_restored": rel_strings,
            "skipped": skipped,
            "exit_code": 0,
            "errors": [],
        }, indent=2))
    else:
        verb = "Unstaged" if staged_mode else "Restored"
        click.echo(f"{verb} {len(rel_strings)} file(s):")
        for rel in rel_strings:
            click.echo(f"  {rel}")
    _restore_audit(ctx, exit_code=0)
 
 
@git.command(name="ignore")
@click.argument("patterns", nargs=-1)
@click.option(
    "--remove", "remove_pattern", default=None,
    help="Remove a pattern from the OCT-managed block.",
)
@click.option(
    "--list", "list_mode", is_flag=True,
    help="List currently active ignore patterns.",
)
@click.option(
    "--scope", "scope",
    type=click.Choice(["workspace", "project", "nearest"]),
    default=None,
    help=(
        "Which .gitignore to target: 'workspace' (workspace-root file), "
        "'project' (subproject-root file), 'nearest' (workspace if at "
        "workspace root, subproject otherwise; default)."
    ),
)
@click.option(
    "--json", "json_mode", is_flag=True, help="Emit JSON output.",
)
@click.pass_context
@audited("oct git ignore")
def git_ignore_cmd(
    ctx: click.Context,
    patterns: tuple[str, ...],
    remove_pattern: str | None,
    list_mode: bool,
    scope: str | None,
    json_mode: bool,
) -> None:
    """Manage .gitignore patterns within the OCT-managed block."""
    func = "git_ignore_cmd"
    root = get_project_root(ctx)
    _dbg(
        "GIT", func,
        f"root={root} patterns={patterns} "
        f"remove={remove_pattern!r} list={list_mode} scope={scope!r}",
        2,
    )
 
    from oct.core.git import get_repo_root, is_git_repo
    from oct.git.policy import (
        add_ignore_patterns,
        list_ignore_patterns,
        remove_ignore_pattern,
        validate_ignore_pattern,
    )
 
    if not is_git_repo(root):
        click.echo("Not a git repository.", err=True)
        _ignore_audit(ctx, exit_code=1)
        ctx.exit(1)
        return
 
    repo_root = get_repo_root(root)
 
    # F1.4: resolve which .gitignore to target.
    octrc = load_octrc_safe(root)
    effective_scope = _resolve_ignore_scope(scope, octrc)
    target_dir, scope_err = _resolve_ignore_target_dir(
        project_root=root,
        repo_root=repo_root,
        scope=effective_scope,
    )
    if scope_err is not None:
        click.echo(f"Error: {scope_err}", err=True)
        if json_mode:
            _emit_ignore_json(
                operation="ignore", exit_code=1, errors=[scope_err],
            )
        _ignore_audit(ctx, exit_code=1)
        ctx.exit(1)
        return
    _dbg(
        "GIT", func,
        f"effective_scope={effective_scope} target_dir={target_dir}",
        3,
    )
 
    # -- Mutual exclusivity --------------------------------------------------
    op_count = sum([bool(patterns), bool(remove_pattern), bool(list_mode)])
    if op_count > 1:
        msg = "Error: PATTERNS, --remove, and --list are mutually exclusive."
        click.echo(msg, err=True)
        if json_mode:
            _emit_ignore_json(
                operation="ignore", exit_code=1, errors=[msg],
            )
        _ignore_audit(ctx, exit_code=1)
        ctx.exit(1)
        return
    if op_count == 0:
        msg = (
            "Use PATTERNS to add, --remove PATTERN to remove, "
            "or --list to show current patterns."
        )
        click.echo(msg, err=True)
        if json_mode:
            _emit_ignore_json(
                operation="ignore", exit_code=1, errors=[msg],
            )
        _ignore_audit(ctx, exit_code=1)
        ctx.exit(1)
        return
 
    # -- --list --------------------------------------------------------------
    if list_mode:
        oct_managed, other = list_ignore_patterns(target_dir)
        if json_mode:
            _emit_ignore_json(
                operation="list", oct_managed=oct_managed,
                other=other, exit_code=0,
            )
        else:
            click.echo(
                f"OCT-managed patterns ({len(oct_managed)}):"
            )
            for p in oct_managed:
                click.echo(f"  {p}")
            click.echo()
            click.echo(
                f"Other patterns ({len(other)}, read-only via oct):"
            )
            for p in other:
                click.echo(f"  {p}")
        _ignore_audit(ctx, exit_code=0)
        return
 
    # -- --remove ------------------------------------------------------------
    if remove_pattern:
        valid, val_msg = validate_ignore_pattern(remove_pattern)
        if not valid:
            click.echo(f"Error: {val_msg}", err=True)
            if json_mode:
                _emit_ignore_json(
                    operation="remove", exit_code=1, errors=[val_msg],
                )
            _ignore_audit(ctx, exit_code=1)
            ctx.exit(1)
            return
        try:
            removed, msg = remove_ignore_pattern(target_dir, remove_pattern)
        except OSError as exc:
            err = f"IO error: {exc}"
            click.echo(err, err=True)
            if json_mode:
                _emit_ignore_json(
                    operation="remove", exit_code=2, errors=[err],
                )
            _ignore_audit(ctx, exit_code=2)
            ctx.exit(2)
            return
        if not removed:
            click.echo(f"Error: {msg}", err=True)
            if json_mode:
                _emit_ignore_json(
                    operation="remove", patterns_removed=[],
                    exit_code=1, errors=[msg],
                )
            _ignore_audit(ctx, exit_code=1)
            ctx.exit(1)
            return
        if json_mode:
            _emit_ignore_json(
                operation="remove",
                patterns_removed=[remove_pattern], exit_code=0,
            )
        else:
            click.echo(f"Removed pattern '{remove_pattern}' from .gitignore.")
        _ignore_audit(ctx, exit_code=0)
        return
 
    # -- Add patterns --------------------------------------------------------
    validated: list[str] = []
    errors: list[str] = []
    for pat in patterns:
        valid, val_msg = validate_ignore_pattern(pat)
        if not valid:
            errors.append(f"{pat!r}: {val_msg}")
        else:
            validated.append(pat)
 
    if errors:
        for e in errors:
            click.echo(f"Error: {e}", err=True)
        if json_mode:
            _emit_ignore_json(
                operation="add", patterns_added=[],
                exit_code=1, errors=errors,
            )
        _ignore_audit(ctx, exit_code=1)
        ctx.exit(1)
        return
 
    try:
        added = add_ignore_patterns(target_dir, validated)
    except OSError as exc:
        err = f"IO error: {exc}"
        click.echo(err, err=True)
        if json_mode:
            _emit_ignore_json(
                operation="add", exit_code=2, errors=[err],
            )
        _ignore_audit(ctx, exit_code=2)
        ctx.exit(2)
        return
 
    if json_mode:
        _emit_ignore_json(
            operation="add", patterns_added=added, exit_code=0,
        )
    else:
        if added:
            click.echo(
                f"Added {len(added)} pattern(s) to .gitignore "
                f"(OCT-managed block):"
            )
            for p in added:
                click.echo(f"  {p}")
        else:
            click.echo("All patterns already present in .gitignore.")
    _ignore_audit(ctx, exit_code=0)
 
 
# The positional arguments are gitignore PATTERNS, not paths. On Windows,
# Click expands every argv token that happens to match something on disk
# (cmd and PowerShell do no expansion themselves), which turned
# "**/docs/html/" into the fourteen directories it matched and got each
# one rejected by validate_ignore_pattern. The attribute tells OctGroup.main
# to hand this command its argv verbatim; no other command is affected.
# See oct.cli._expand_windows_args (2026-09-27).
git_ignore_cmd.no_glob_expansion = True  # type: ignore[attr-defined]
 
 
# =====================================================================
# oct git commit-msg (OI-503 - entry for pre-commit-framework commit-msg stage)
# =====================================================================
 
 
@git.command(name="commit-msg")
@click.argument(
    "message_file",
    type=click.Path(exists=True, dir_okay=False, path_type=Path),
)
@click.option("--json", "json_mode", is_flag=True, help="Emit JSON result.")
@click.pass_context
def git_commit_msg_cmd(ctx: click.Context, message_file: Path, json_mode: bool) -> None:
    """Validate a Conventional Commit message file.
 
    Intended as the entry for the ``commit-msg`` pre-commit-framework stage:
    ``pre-commit`` invokes ``oct git commit-msg <file>`` with the path to
    ``.git/COMMIT_EDITMSG``. Exits 0 on valid, 1 on invalid, 2 on I/O errors.
    """
    func = "git_commit_msg_cmd"
    _dbg("GIT", func, f"validating {message_file}", 2)
 
    from oct.git.conventional import (
        format_validation_errors,
        validate_ai_model_trailer,
        validate_commit_message,
    )
 
    try:
        text = message_file.read_text(encoding="utf-8")
    except OSError as exc:
        if json_mode:
            click.echo(
                json.dumps(
                    {"valid": False, "errors": [f"cannot read {message_file}: {exc}"]},
                    indent=2,
                )
            )
        else:
            click.echo(
                f"[oct git commit-msg] cannot read {message_file}: {exc}",
                err=True,
            )
        ctx.exit(2)
        return
 
    errors = list(validate_commit_message(text))
    # Option C commit-forensics convention (Plan 151 T6, inventory item 4):
    # enforce the AI-Model trailer only when ``git.require_ai_model_trailer``
    # is enabled in the project's ``.octrc.json`` (default off).
    try:
        _root = get_project_root(ctx)
        _octrc = load_octrc_safe(_root)
    except Exception:  # pragma: no cover - config discovery is best-effort
        _octrc = {}
    if (_octrc or {}).get("git", {}).get("require_ai_model_trailer", False):
        errors.extend(validate_ai_model_trailer(text))
    valid = not errors
 
    if json_mode:
        click.echo(
            json.dumps({"valid": valid, "errors": errors}, indent=2)
        )
    else:
        if valid:
            click.echo(f"[oct git commit-msg] OK: {message_file.name}")
        else:
            click.echo(format_validation_errors(errors), err=True)
 
    ctx.exit(0 if valid else 1)
 
 
# =====================================================================
# oct git hooks (Phase 4D - G-D8, G-D9, G-D10)
# =====================================================================
 
 
_ENHANCED_HOOKS_TEMPLATE = """\
# Option C pre-commit configuration
# Generated by: oct git hooks install
# See https://pre-commit.com for more information
 
repos:
  - repo: local
    hooks:
      - id: oct-quality-gate
        name: Option C Quality Gate
        entry: oct git check --staged-only
        language: system
        types: [python]
        pass_filenames: false
 
      - id: oct-secrets-check
        name: Option C Secrets Check
        entry: oct git check --staged-only --profile strict
        language: system
        always_run: true
        pass_filenames: false
 
      - id: oct-commit-msg
        name: Option C Commit Message
        entry: oct git commit-msg
        language: system
        stages: [commit-msg]
 
  - repo: local
    hooks:
      - id: oct-pre-push
        name: Option C Pre-Push Gate
        entry: oct git check --include-tests
        language: system
        stages: [pre-push]
        pass_filenames: false
"""
 
# Hook IDs used to identify OCT hooks in a pre-commit config.
_OCT_HOOK_IDS = (
    "oct-quality-gate",
    "oct-secrets-check",
    "oct-commit-msg",
    "oct-pre-push",
    # Legacy hook IDs from oct install-hooks (Phase 4C and earlier).
    "oct-lint",
    "oct-format-check",
)
 
 
@git.group(name="hooks")
@click.pass_context
def hooks_group(ctx: click.Context) -> None:
    """Manage Option C git hooks."""
    pass
 
 
@hooks_group.command(name="install")
@click.option("--force", is_flag=True, help="Overwrite existing configuration.")
@click.pass_context
@audited("oct git hooks install")
def hooks_install_cmd(ctx: click.Context, force: bool) -> None:
    """Generate .pre-commit-config.yaml with all Option C hooks."""
    func = "hooks_install_cmd"
    root = get_project_root(ctx)
    _dbg("GIT", func, f"root={root} force={force}", 2)
 
    config_path = root / ".pre-commit-config.yaml"
    if config_path.exists() and not force:
        click.echo(
            f"Error: {config_path} already exists. Use --force to overwrite.",
            err=True,
        )
        record = ctx.obj.get("_audit_record", AuditRecord())
        record.timestamp = record.timestamp or _utc_now_iso()
        record.command = "oct git hooks install"
        record.checks_passed = False
        record.exit_code = 1
        ctx.obj["_audit_record"] = record
        ctx.exit(1)
        return
 
    config_path.write_text(_ENHANCED_HOOKS_TEMPLATE, encoding="utf-8")
    click.echo(f"Created {config_path}")
    click.echo("Run 'pre-commit install' to activate the hooks.")
 
    record = ctx.obj.get("_audit_record", AuditRecord())
    record.timestamp = record.timestamp or _utc_now_iso()
    record.command = "oct git hooks install"
    record.checks_passed = True
    record.exit_code = 0
    ctx.obj["_audit_record"] = record
 
 
@hooks_group.command(name="status")
@click.pass_context
@audited("oct git hooks status")
def hooks_status_cmd(ctx: click.Context) -> None:
    """Show installed hook state."""
    func = "hooks_status_cmd"
    root = get_project_root(ctx)
    _dbg("GIT", func, f"root={root}", 2)
 
    config_path = root / ".pre-commit-config.yaml"
    expected = ("oct-quality-gate", "oct-secrets-check", "oct-commit-msg", "oct-pre-push")
 
    found: set[str] = set()
    if config_path.is_file():
        content = config_path.read_text(encoding="utf-8")
        for hook_id in expected:
            if f"id: {hook_id}" in content:
                found.add(hook_id)
 
    for hook_id in expected:
        status = "[INSTALLED]" if hook_id in found else "[MISSING]"
        click.echo(f"  {status}  {hook_id}")
 
    record = ctx.obj.get("_audit_record", AuditRecord())
    record.timestamp = record.timestamp or _utc_now_iso()
    record.command = "oct git hooks status"
    record.checks_passed = True
    record.exit_code = 0
    ctx.obj["_audit_record"] = record
 
 
@hooks_group.command(name="remove")
@click.pass_context
@audited("oct git hooks remove")
def hooks_remove_cmd(ctx: click.Context) -> None:
    """Remove Option C hooks from .pre-commit-config.yaml."""
    func = "hooks_remove_cmd"
    root = get_project_root(ctx)
    _dbg("GIT", func, f"root={root}", 2)
 
    config_path = root / ".pre-commit-config.yaml"
    if not config_path.is_file():
        click.echo("No .pre-commit-config.yaml found.", err=True)
        record = ctx.obj.get("_audit_record", AuditRecord())
        record.timestamp = record.timestamp or _utc_now_iso()
        record.command = "oct git hooks remove"
        record.checks_passed = True
        record.exit_code = 0
        ctx.obj["_audit_record"] = record
        return
 
    content = config_path.read_text(encoding="utf-8")
 
    # Check if any non-OCT hooks exist.
    has_non_oct = False
    in_oct_hook = False
    filtered_lines: list[str] = []
    for line in content.splitlines():
        stripped = line.strip()
        # Detect OCT hook blocks by their id.
        if stripped.startswith("- id:"):
            hook_id = stripped.split(":", 1)[1].strip()
            in_oct_hook = hook_id in _OCT_HOOK_IDS
            if not in_oct_hook:
                has_non_oct = True
 
        if in_oct_hook:
            # Skip lines belonging to OCT hooks.
            # The hook block ends at the next "- id:" or "- repo:" or end of file.
            if stripped.startswith("- id:") or stripped.startswith("- repo:"):
                # Re-evaluate: is this the start of a new hook or repo?
                if stripped.startswith("- id:"):
                    hook_id = stripped.split(":", 1)[1].strip()
                    in_oct_hook = hook_id in _OCT_HOOK_IDS
                    if in_oct_hook:
                        continue
                    else:
                        has_non_oct = True
                        filtered_lines.append(line)
                        in_oct_hook = False
                        continue
                elif stripped.startswith("- repo:"):
                    in_oct_hook = False
                    filtered_lines.append(line)
                    continue
            else:
                continue
        else:
            filtered_lines.append(line)
 
    if not has_non_oct:
        # All hooks were OCT hooks - delete the file.
        config_path.unlink()
        click.echo("Removed .pre-commit-config.yaml (all hooks were Option C hooks).")
    else:
        # Rewrite with non-OCT hooks preserved.
        config_path.write_text("\n".join(filtered_lines) + "\n", encoding="utf-8")
        click.echo("Removed Option C hooks (non-OCT hooks preserved).")
 
    record = ctx.obj.get("_audit_record", AuditRecord())
    record.timestamp = record.timestamp or _utc_now_iso()
    record.command = "oct git hooks remove"
    record.checks_passed = True
    record.exit_code = 0
    ctx.obj["_audit_record"] = record
 
 
# =====================================================================
# oct git changelog (Phase 4E - G-E2)
# =====================================================================
 
 
@git.command(name="changelog")
@click.option("--release", "version", default=None,
              help="Release version tag (e.g. v0.13.0).")
@click.option("--since", default=None,
              help="Git ref to start from (default: last tag).")
@click.option("--dry-run", is_flag=True,
              help="Print to stdout without writing CHANGELOG.md.")
@click.option("--json", "json_mode", is_flag=True,
              help="Emit structured JSON.")
@click.pass_context
@audited("oct git changelog")
def git_changelog_cmd(
    ctx: click.Context,
    version: str | None,
    since: str | None,
    dry_run: bool,
    json_mode: bool,
) -> None:
    """Generate changelog from Conventional Commits history."""
    func = "git_changelog_cmd"
    root = get_project_root(ctx)
    _dbg("GIT", func, f"root={root} version={version!r} since={since!r}", 2)
 
    from oct.git.changelog import (
        collect_changelog_entries,
        format_changelog,
        changelog_to_json,
    )
 
    root = _require_git_repo(root)
 
    entries = collect_changelog_entries(root, since_ref=since)
 
    if json_mode:
        data = changelog_to_json(entries, version=version)
        click.echo(json.dumps(data, indent=2))
        record = ctx.obj.get("_audit_record", AuditRecord())
        record.timestamp = record.timestamp or _utc_now_iso()
        record.command = "oct git changelog"
        record.checks_passed = True
        record.exit_code = 0
        ctx.obj["_audit_record"] = record
        return
 
    text = format_changelog(entries, version=version)
 
    if dry_run or version is None:
        click.echo(text)
    else:
        # Prepend to CHANGELOG.md.
        changelog_path = root / "CHANGELOG.md"
        existing = ""
        if changelog_path.is_file():
            existing = changelog_path.read_text(encoding="utf-8")
        # Insert after the top-level heading if present, else prepend.
        if existing.startswith("# "):
            # Keep the top heading, insert after first blank line.
            first_newline = existing.find("\n\n")
            if first_newline >= 0:
                new_content = (
                    existing[:first_newline + 2]
                    + text + "\n"
                    + existing[first_newline + 2:]
                )
            else:
                new_content = existing + "\n\n" + text + "\n"
        elif existing:
            new_content = text + "\n" + existing
        else:
            new_content = f"# Changelog\n\n{text}\n"
        changelog_path.write_text(new_content, encoding="utf-8")
        click.echo(f"Updated {to_project_relative(changelog_path, root)}")
 
    record = ctx.obj.get("_audit_record", AuditRecord())
    record.timestamp = record.timestamp or _utc_now_iso()
    record.command = "oct git changelog"
    record.checks_passed = True
    record.exit_code = 0
    ctx.obj["_audit_record"] = record
 
 
# =====================================================================
# Helpers
# =====================================================================
 
 
def _require_git_repo(root: Path) -> Path:
    """Return the git repo root, or exit with a clear UX message.
 
    - Git not installed: *"Git is not installed. Git-related features
      are disabled."*
    - Not a repo: *"Not a git repository. Run 'oct git init' to
      initialise."*
    """
    from oct.core.git import (
        is_git_repo, get_repo_root as _get_repo_root,
        GitNotFoundError, GitNotARepoError,
    )
    import shutil
    if shutil.which("git") is None:
        raise click.ClickException(
            "Git is not installed. Git-related features are disabled."
        )
    if not is_git_repo(root):
        raise click.ClickException(
            "Not a git repository. Run 'oct git init' to initialise."
        )
    try:
        return _get_repo_root(root)
    except (GitNotFoundError, GitNotARepoError) as exc:
        raise click.ClickException(str(exc))
 
 
def _check_large_staged_files(
    root: Path, staged: list[Path], octrc: dict | None,
) -> list[tuple[str, int]]:
    """Check staged files for size exceeding the threshold.
 
    Returns a list of ``(rel_path, size_bytes)`` for oversized files.
    Threshold is configurable via ``octrc["git"]["large_file_threshold_mb"]``
    (default: 10 MB).
    """
    threshold = _LARGE_FILE_THRESHOLD_BYTES
    if isinstance(octrc, dict):
        git_cfg = octrc.get("git")
        if isinstance(git_cfg, dict):
            raw = git_cfg.get("large_file_threshold_mb")
            if isinstance(raw, (int, float)) and raw > 0:
                threshold = int(raw * 1024 * 1024)
 
    oversized: list[tuple[str, int]] = []
    for path in staged:
        try:
            size = path.stat().st_size
            if size > threshold:
                rel = to_project_relative(path, root)
                oversized.append((rel, size))
        except OSError:
            continue
    return oversized
 
 
# Plan 199: the module-local ``_load_octrc`` that used to live here read
# a bare ``root / ".octrc.json"``, which made every project on the
# canonical ``.option_c/octrc.json`` layout invisible to the git layer.
# It is gone deliberately: :func:`oct.core.octrc.load_octrc_safe`,
# imported at module scope, is the ONE loader for the whole toolchain.
 
 
# ---------------------------------------------------------------------
# Auto-branch helpers (Phase 4F - G-F2)
# ---------------------------------------------------------------------
 
 
def _should_auto_branch(branch: str | None, octrc: dict | None) -> bool:
    """Return True when auto-branching is enabled for *branch*.
 
    Auto-branching activates when the branch is protected and the
    ``git.auto_branch_on_protected`` setting is not explicitly ``False``
    (defaults to ``True``).
    """
    from oct.git.policy import is_protected_branch
 
    if not is_protected_branch(branch, octrc):
        return False
    if isinstance(octrc, dict):
        git_cfg = octrc.get("git")
        if isinstance(git_cfg, dict):
            raw = git_cfg.get("auto_branch_on_protected")
            if isinstance(raw, bool):
                return raw
    return True  # default: enabled
 
 
def _deduplicate_branch_name(root: Path, name: str) -> str:
    """Return *name* if it does not exist, otherwise append ``-2``, ``-3``, etc."""
    from oct.core.git import branch_exists
 
    if not branch_exists(root, name):
        return name
    counter = 2
    while counter <= 100:
        candidate = f"{name}-{counter}"
        if not branch_exists(root, candidate):
            return candidate
        counter += 1
    # Extremely unlikely; surface as-is and let git error naturally.
    return f"{name}-{counter}"
 
 
def _auto_branch_for_commit(
    root: Path,
    original_branch: str,
    message: str,
    ctx: click.Context,
    json_mode: bool,
) -> str | None:
    """Create a feature branch from a commit message and switch to it.
 
    Returns the new branch name on success, or ``None`` on failure
    (after emitting errors and populating the audit record).
    """
    func = "_auto_branch_for_commit"
    from oct.core.git import create_and_switch_branch
    from oct.git.policy import generate_branch_name_from_message
 
    auto_name = generate_branch_name_from_message(message)
    if auto_name is None:
        msg = "Cannot auto-branch: failed to derive branch name from commit message."
        click.echo(msg, err=True)
        if json_mode:
            _emit_commit_json(
                ctx, exit_code=1, message=message, branch=original_branch,
                profile="strict", errors=[msg],
            )
        _populate_commit_audit(ctx, branch=original_branch, exit_code=1,
                               checks_passed=False, staged_count=0)
        ctx.exit(1)
        return None
 
    final_name = _deduplicate_branch_name(root, auto_name)
 
    try:
        create_and_switch_branch(root, final_name)
    except Exception as exc:
        msg = f"Auto-branch failed: {exc}"
        click.echo(msg, err=True)
        _dbg("GIT", func, msg, 1)
        if json_mode:
            _emit_commit_json(
                ctx, exit_code=1, message=message, branch=original_branch,
                profile="strict", errors=[msg],
            )
        _populate_commit_audit(ctx, branch=original_branch, exit_code=1,
                               checks_passed=False, staged_count=0)
        ctx.exit(1)
        return None
 
    click.echo(
        f"Auto-created branch '{final_name}' from '{original_branch}'.",
        err=True,
    )
    _dbg("GIT", func, f"auto-branched to {final_name!r}", 2)
    return final_name