probe.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/complexity/probe.py
 
"""
Purpose
-------
Run the Plan 214 worst-first CRAP probe over the Option C monorepo.
 
Responsibilities
----------------
- Execute each required MAIN-tree subproject suite under coverage.py.
- Invoke ``python -m oct.complexity`` against MAIN sources through worktree PYTHONPATH.
- Preserve failed suites as UNMEASURED with their captured output.
- Generate the full JSON artifact and top-25-per-project Markdown report.
 
Diagnostics
-----------
Domain: OCT-COMPLEXITY
Levels:
    L2 - probe lifecycle
    L3 - subproject measurement details
    L4 - deep tracing
 
Contracts
---------
- The probe measures and never gates.
- Failed suites, parse failures, and never-imported functions are never omitted.
- Analyzer commands run against MAIN sources with this worktree's instrument.
 
Dependencies
------------
- oct.complexity.coverage (coverage suite runner)
- oct.complexity (not imported; run as python -m oct.complexity for the JSON analysis subprocess)
"""
 
from __future__ import annotations
 
import argparse
import datetime as dt
import json
import os
import subprocess
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Sequence
 
from oct.complexity.coverage import CoverageRun, run_pytest_with_coverage
 
# OCT-LINT: disable-file=dbg_usage reason="Probe output is a deterministic committed artifact and must not contain diagnostics noise"
# OCT-LINT: disable-file=dbg_import reason="Diagnostics would contaminate analyzer JSON and probe console output"
 
 
BASE_SHA = "e4d56a63"
REPORT_BASENAME = "crap-probe-2026-08"
 
 
@dataclass(frozen=True, slots=True)
class ProbeProject:
    """One monorepo subproject and its suite/source boundaries."""
 
    name: str
    project_root: str
    pytest_args: tuple[str, ...]
    source_paths: tuple[str, ...]
 
 
PROJECTS = (
    ProbeProject(
        name="oct",
        project_root="oct",
        pytest_args=("tests", "-q"),
        source_paths=("oct/oct",),
    ),
    ProbeProject(
        name="oc_diagnostics",
        project_root="oc_diagnostics",
        pytest_args=("tests", "-q"),
        source_paths=("oc_diagnostics/oc_diagnostics",),
    ),
    ProbeProject(
        name="oc_chat",
        project_root="oc_chat",
        pytest_args=("oc_chat/tests", "-q"),
        source_paths=("oc_chat/oc_chat",),
    ),
    ProbeProject(
        name="oc_pricing",
        project_root="oc_pricing",
        pytest_args=("tests", "-q"),
        source_paths=("oc_pricing/oc_pricing",),
    ),
    ProbeProject(
        name="option_c_agent_contracts",
        project_root="contracts/option_c_agent_contracts",
        pytest_args=("tests", "-q"),
        source_paths=("contracts/option_c_agent_contracts/option_c_agent_contracts",),
    ),
    ProbeProject(
        name="option_c_registry_contracts",
        project_root="contracts/option_c_registry_contracts",
        pytest_args=("tests", "-q"),
        source_paths=(
            "contracts/option_c_registry_contracts/option_c_registry_contracts",
        ),
    ),
)
 
 
def _parser() -> argparse.ArgumentParser:
    """Build the probe driver's argument parser."""
    parser = argparse.ArgumentParser(
        description="Run the Plan 214 monorepo CRAP probe.",
    )
    parser.add_argument("--repo-root", required=True, type=Path)
    parser.add_argument("--output-dir", required=True, type=Path)
    parser.add_argument("--data-dir", required=True, type=Path)
    parser.add_argument(
        "--python-executable",
        type=Path,
        default=Path(sys.executable),
    )
    return parser
 
 
def main(argv: Sequence[str] | None = None) -> int:
    """Run every probe project and write both committed artifacts."""
    args = _parser().parse_args(argv)
    repo_root = args.repo_root.resolve()
    output_dir = args.output_dir.resolve()
    data_dir = args.data_dir.resolve()
    python_executable = args.python_executable.resolve()
    output_dir.mkdir(parents=True, exist_ok=True)
    data_dir.mkdir(parents=True, exist_ok=True)
    projects = [
        _run_project(
            project,
            repo_root=repo_root,
            data_dir=data_dir,
            python_executable=python_executable,
        )
        for project in PROJECTS
    ]
    generated_at = dt.datetime.now(dt.timezone.utc).isoformat()
    payload = {
        "report_version": 1,
        "generated_at": generated_at,
        "base_sha": BASE_SHA,
        "instrument_root": str(_instrument_root()),
        "repo_root": str(repo_root),
        "projects": projects,
    }
    json_path = output_dir / f"{REPORT_BASENAME}.json"
    markdown_path = output_dir / f"{REPORT_BASENAME}.md"
    json_path.write_text(
        json.dumps(payload, indent=2, sort_keys=True) + "\n",
        encoding="utf-8",
    )
    markdown_path.write_text(
        _render_markdown(payload, data_dir=data_dir),
        encoding="utf-8",
    )
    print(f"Wrote {markdown_path}")
    print(f"Wrote {json_path}")
    return 0
 
 
def _run_project(
    project: ProbeProject,
    *,
    repo_root: Path,
    data_dir: Path,
    python_executable: Path,
) -> dict[str, object]:
    """Run one suite, then map its data through the module command."""
    print(f"Measuring {project.name}...")
    project_root = repo_root / project.project_root
    source_paths = tuple(repo_root / item for item in project.source_paths)
    data_file = data_dir / f"{project.name}.coverage"
    suite_env = os.environ.copy()
    suite_env.pop("PYTHONPATH", None)
    suite = run_pytest_with_coverage(
        project_root=project_root,
        pytest_args=project.pytest_args,
        data_file=data_file,
        source=source_paths,
        python_executable=python_executable,
        env=suite_env,
    )
    suite_record = _suite_record(suite)
    base_record: dict[str, object] = {
        "name": project.name,
        "project_root": project.project_root,
        "source_paths": list(project.source_paths),
        "suite": suite_record,
    }
    if suite.returncode != 0:
        reason = f"pytest exited {suite.returncode}"
        if not suite_record["output"]:
            suite_record["output"] = "(no pytest output captured)"
        return {
            **base_record,
            "status": "UNMEASURED",
            "reason": reason,
            "analysis_command": None,
            "functions": [],
            "parse_errors": [],
        }
    analysis_command = (
        str(python_executable),
        "-m",
        "oct.complexity",
        *(str(item) for item in source_paths),
        "--coverage-data",
        str(data_file),
        "--json",
    )
    analysis_env = os.environ.copy()
    analysis_env["PYTHONPATH"] = str(_instrument_root())
    analysis = subprocess.run(
        analysis_command,
        cwd=repo_root,
        env=analysis_env,
        capture_output=True,
        text=True,
        errors="replace",
    )
    if analysis.returncode != 0:
        reason = f"complexity analysis exited {analysis.returncode}"
        failure_output = _joined_output(analysis.stdout, analysis.stderr)
        suite_record["output"] = _joined_output(
            str(suite_record["output"]),
            failure_output or "(no complexity output captured)",
        )
        return {
            **base_record,
            "status": "UNMEASURED",
            "reason": reason,
            "analysis_command": _command_text(analysis_command),
            "functions": [],
            "parse_errors": [],
        }
    try:
        analysis_payload = json.loads(analysis.stdout)
    except json.JSONDecodeError as exc:
        reason = f"complexity JSON decode failed: {exc}"
        suite_record["output"] = _joined_output(
            str(suite_record["output"]),
            analysis.stdout,
            analysis.stderr,
        )
        return {
            **base_record,
            "status": "UNMEASURED",
            "reason": reason,
            "analysis_command": _command_text(analysis_command),
            "functions": [],
            "parse_errors": [],
        }
    functions = [
        _relativize_row(row, repo_root)
        for row in analysis_payload["functions"]
    ]
    parse_errors = [
        _relativize_error(row, repo_root)
        for row in analysis_payload["parse_errors"]
    ]
    return {
        **base_record,
        "status": "MEASURED",
        "reason": None,
        "analysis_command": _command_text(analysis_command),
        "functions": functions,
        "parse_errors": parse_errors,
    }
 
 
def _suite_record(suite: CoverageRun) -> dict[str, object]:
    """Convert a coverage suite result to stable probe JSON."""
    return {
        "command": _command_text(suite.command),
        "returncode": suite.returncode,
        "output": suite.output,
        "coverage_data": suite.data_file,
    }
 
 
def _relativize_row(row: dict[str, object], repo_root: Path) -> dict[str, object]:
    """Replace an absolute source filename with a repository-relative path."""
    result = dict(row)
    result["file"] = _relative_path(str(row["file"]), repo_root)
    return result
 
 
def _relativize_error(
    row: dict[str, object],
    repo_root: Path,
) -> dict[str, object]:
    """Replace an absolute parse-error filename with a relative path."""
    result = dict(row)
    result["file"] = _relative_path(str(row["file"]), repo_root)
    return result
 
 
def _relative_path(filename: str, repo_root: Path) -> str:
    """Return a forward-slash repository-relative path when possible."""
    try:
        return Path(filename).resolve().relative_to(repo_root).as_posix()
    except ValueError:
        return Path(filename).as_posix()
 
 
def _render_markdown(
    payload: dict[str, object],
    *,
    data_dir: Path,
) -> str:
    """Render the human probe report from the same payload as JSON."""
    lines = [
        "# CRAP probe - August 2026",
        "",
        "This report is a measurement snapshot, not a gate. It presents data only and makes no recommendations.",
        "",
        f"- BASE_SHA: `{payload['base_sha']}`",
        f"- Generated (UTC): `{payload['generated_at']}`",
        f"- MAIN source tree: `{payload['repo_root']}`",
        f"- Instrument worktree path: `{payload['instrument_root']}`",
        "",
        "## Reproduction",
        "",
        "The probe was launched from the MAIN tree with the worktree instrument on `PYTHONPATH`:",
        "",
        "```powershell",
        f"$env:PYTHONPATH='{payload['instrument_root']}'",
        _probe_command(payload, data_dir),
        "```",
        "",
        "For each measured project, the exact coverage and analysis commands actually executed are recorded below.",
        "",
    ]
    for project in payload["projects"]:
        lines.extend(_render_project(project))
    return "\n".join(lines).rstrip() + "\n"
 
 
def _probe_command(payload: dict[str, object], data_dir: Path) -> str:
    """Return the exact top-level probe command for the report."""
    output_dir = Path(__file__).resolve().parents[3] / "oc_chat" / "oc_chat" / "docs" / "reports"
    command = (
        sys.executable,
        "-m",
        "oct.complexity.probe",
        "--repo-root",
        str(payload["repo_root"]),
        "--output-dir",
        str(output_dir),
        "--data-dir",
        str(data_dir),
        "--python-executable",
        sys.executable,
    )
    return f"& {_command_text(command)}"
 
 
def _render_project(project: dict[str, object]) -> list[str]:
    """Render one measured or honestly unmeasured project section."""
    suite = project["suite"]
    lines = [
        f"## {project['name']}",
        "",
        f"Status: **{project['status']}**",
        "",
        f"- Coverage command: `{suite['command']}`",
        f"- Suite return code: `{suite['returncode']}`",
    ]
    if project["analysis_command"]:
        lines.append(f"- Analysis command: `{project['analysis_command']}`")
    lines.append("")
    if project["status"] == "UNMEASURED":
        lines.extend(
            [
                f"Reason: {project['reason']}",
                "",
                "```text",
                str(suite["output"]).rstrip(),
                "```",
                "",
            ]
        )
        return lines
    functions = project["functions"]
    lines.extend(
        [
            f"Measured functions: {len(functions)}. Worst 25 shown.",
            "",
            "| Function | CC | Coverage | CRAP | Status | Span |",
            "|---|---:|---:|---:|---|---|",
        ]
    )
    for row in functions[:25]:
        path = str(row["file"])
        span = f"{path}:{row['lineno']}-{row['end_lineno']}"
        link = f"../../../../{path}#L{row['lineno']}-L{row['end_lineno']}"
        lines.append(
            f"| {_escape_cell(row['qualname'])} | {row['complexity']} | "
            f"{row['coverage'] * 100:.1f}% | {row['crap']:.2f} | "
            f"{row['coverage_status']} | [{span}]({link}) |"
        )
    lines.append("")
    parse_errors = project["parse_errors"]
    if parse_errors:
        lines.append("Parse errors:")
        lines.append("")
        for error in parse_errors:
            lines.append(f"- `{error['file']}`: {error['error']}")
        lines.append("")
    else:
        lines.extend(["Parse errors: none.", ""])
    return lines
 
 
def _escape_cell(value: object) -> str:
    """Escape Markdown table separators in one cell value."""
    return str(value).replace("|", "\\|")
 
 
def _instrument_root() -> Path:
    """Return the worktree oct project directory used for PYTHONPATH."""
    return Path(__file__).resolve().parents[2]
 
 
def _command_text(command: Sequence[str]) -> str:
    """Return a Windows-rerunnable rendering of an argument sequence."""
    return subprocess.list2cmdline([str(item) for item in command])
 
 
def _joined_output(*parts: str) -> str:
    """Join captured output fragments without silently dropping any."""
    return "\n".join(part.rstrip() for part in parts if part).strip()
 
 
if __name__ == "__main__":
    raise SystemExit(main())