verification.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/core/verification.py
 
"""
Verification backpressure — the deterministic inner-loop feedback sensor (plan 137).
 
Purpose
-------
Run the *existing* computational controls (``oct lint`` + ``oct typecheck``, and
an opt-in narrowest test selection) automatically over the files an iteration
mutated, diff the result against a session-local baseline so only **new** findings
surface, and return a compacted, typed :class:`VerificationResult` the caller
injects back into the loop. The sensor runs **no model call** — it is deterministic
subprocess work. It never blocks the turn: any infrastructure problem fails open.
 
Responsibilities
----------------
- :func:`run_verification` — changed-files-only, time-boxed, classified, baseline-
  diffed lint/typecheck/tests over **explicit targets** (never ``--changed``).
- :func:`capture_baseline` — a session-local lint+typecheck snapshot (a finding
  multiset) over a bounded scope, fail-open; never OCT's global baseline file.
- :func:`render_observation` — compact the result into the ``[verification]``
  message body (12-Factor #9 — never raw tracebacks).
- :class:`Finding` / :class:`VerificationResult` — the typed shapes.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 — lifecycle events (verification start/result)
    L3 — semantic details (per-tool classification, counts)
    L4 — deep tracing (argv, parse details)
 
Contracts
---------
- Pure of model calls. Subprocesses run with an argv list (Windows-safe), a hard
  ``timeout``, and ``cwd`` = workspace root. A missing binary / rejected flag /
  timeout / unparseable output / too-many-files classifies **degraded** (fail-open).
- A non-zero exit **with parseable findings** is ``findings`` (a real failure to
  surface), not fail-open.
- Empty ``changed_files`` short-circuits to ``clean`` — ``oct lint`` is never
  invoked with no targets (which would lint the whole workspace).
- The baseline diff keys on the full ``(path, code, normalized-message)`` tuple
  with count-based subtraction, so a line-shifted pre-existing finding stays
  suppressed while a genuinely new message surfaces.
- ``capture_baseline`` stores findings in-process only; it never runs
  ``oct lint --baseline`` and never touches ``.option_c/cache/lint-baseline.json``.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.core._paths (lazy; workspace-root resolution)
- oc_chat.tools._sandbox (lazy; scrubbed environment for the verification subprocesses)
- oct (not imported; python -m oct lint and typecheck over the changed files and the session baseline scope)
- pytest (not imported; python -m pytest over the companion test files)
"""
from __future__ import annotations
 
import json
import re
import subprocess
import sys
from collections import Counter
from dataclasses import dataclass, field
from pathlib import Path
from typing import Callable
 
from oc_diagnostics import _dbg
 
_dbg("OC_CHAT.MAIN", "verification", "module", 4)
 
# Per-command classifications.
CLEAN = "clean"
FINDINGS = "findings"
INFRA_ERROR = "infra_error"
TIMEOUT = "timeout"
UNPARSEABLE = "unparseable"
SKIPPED = "skipped"
 
#: pyright text line: ``  <path>:<line>:<col> - error: <message> (reportRule)``.
_PYRIGHT_RE = re.compile(
    # OCT-LINT: disable=no_hardcoded_secrets reason="Regex for one pyright text output line; no credential material (operator-ruled 2026-09-28)"
    r"^\s*(?P<path>.+?):(?P<line>\d+):(?P<col>\d+)\s*-\s*"
    r"(?P<sev>error|warning|information)\s*:\s*(?P<msg>.*?)\s*$"
)
#: trailing ``(reportSomething)`` rule tag on a pyright message.
_RULE_RE = re.compile(r"\‍(([A-Za-z][A-Za-z0-9]+)\‍)\s*$")
#: digit runs and address-like tokens stripped for message normalisation.
_POS_RE = re.compile(r"\b\d+\b")
 
 
@dataclass(frozen=True)
class Finding:
    """A single verification finding (lint check failure or type error)."""
 
    path: str
    line: int | None
    code: str
    message: str
    severity: str = "blocking"
    tool: str = "lint"  # "lint" | "typecheck" | "tests"
 
    def key(self) -> tuple[str, str, str]:
        """Baseline identity: ``(path, code, normalized-message)`` — NOT the line,
        so a pre-existing finding shifted by inserted lines keeps its identity."""
        return (self.path, self.code, _normalize_message(self.message))
 
 
@dataclass
class VerificationResult:
    """Typed outcome of one :func:`run_verification` call (plan 137 §3.3)."""
 
    clean: bool
    findings: list[Finding] = field(default_factory=list)
    degraded: bool = False
    skip_reason: str | None = None
    lint_status: str = SKIPPED
    typecheck_status: str = SKIPPED
    tests_status: str = SKIPPED
 
 
@dataclass
class _CmdOutcome:
    """Raw result of one subprocess invocation (or a non-completion sentinel)."""
 
    returncode: int | None
    stdout: str = ""
    stderr: str = ""
    error: str | None = None  # "timeout" | "missing_binary" | None
 
 
# A runner is injectable so tests stub the subprocess hermetically.
Runner = Callable[[list[str], int], _CmdOutcome]
 
 
def _workspace_root(explicit: str | Path | None) -> Path:
    if explicit is not None:
        return Path(explicit)
    try:
        from oc_chat.core._paths import find_workspace_root
        return find_workspace_root()
    except Exception:  # noqa: BLE001 — total by contract; never raise into the loop
        return Path.cwd()
 
 
def _oct_argv(args: list[str]) -> list[str]:
    """Full argv for an ``oct`` subcommand (``python -m oct`` — PATH-independent)."""
    return [sys.executable, "-m", "oct", *args]
 
 
def _default_runner(argv: list[str], timeout: int) -> _CmdOutcome:
    """Run *argv* with a hard timeout; classify non-completion (Windows-safe)."""
    func = "_default_runner"
    # M8.0 WU-3c: run verification subprocesses with a minimal scrubbed env
    # (drop poisonable GIT_*/PIP_*/proxy/cert/PYTHONPATH/... families;
    # F25/F72/F78/F79) rather than inheriting the full process env.
    from oc_chat.tools._sandbox import scrubbed_env
    try:
        proc = subprocess.run(
            argv, capture_output=True, text=True, timeout=timeout, check=False,
            env=scrubbed_env(),
        )
    except subprocess.TimeoutExpired:
        _dbg("OC_CHAT.MAIN", func, f"timeout after {timeout}s: {argv[:4]}", 3)
        return _CmdOutcome(returncode=None, error="timeout")
    except (FileNotFoundError, OSError) as exc:
        _dbg("OC_CHAT.MAIN", func, f"missing binary / OS error: {type(exc).__name__}", 3)
        return _CmdOutcome(returncode=None, error="missing_binary")
    return _CmdOutcome(returncode=proc.returncode, stdout=proc.stdout, stderr=proc.stderr)
 
 
def _norm_path(path: str, root: Path) -> str:
    """Workspace-root-relative POSIX path (pyright emits absolute paths)."""
    try:
        p = Path(path)
        if p.is_absolute():
            p = p.relative_to(root)
        return p.as_posix()
    except (ValueError, OSError):
        return str(path).replace("\\", "/")
 
 
def _normalize_message(msg: str) -> str:
    """Strip embedded positions/values so the message is not a second drift source."""
    text = " ".join(str(msg).split())
    text = _POS_RE.sub("N", text)  # collapse line/col/size numbers to a placeholder
    return text
 
 
def _extract_json(stdout: str) -> dict | None:
    """Parse the first JSON object in *stdout* (warnings may precede it)."""
    i = stdout.find("{")
    if i < 0:
        return None
    try:
        data = json.loads(stdout[i:])
    except (json.JSONDecodeError, ValueError):
        return None
    return data if isinstance(data, dict) else None
 
 
def _parse_lint(data: dict, root: Path) -> list[Finding]:
    """Blocking lint check failures from the ``files[].checks`` JSON shape."""
    findings: list[Finding] = []
    for f in data.get("files") or []:
        if not isinstance(f, dict):
            continue
        path = _norm_path(str(f.get("path", "")), root)
        for code, chk in (f.get("checks") or {}).items():
            if not isinstance(chk, dict) or chk.get("pass", True):
                continue
            if str(chk.get("severity", "blocking")) != "blocking":
                continue  # advisory (e.g. tests_exist) is informational, not surfaced
            findings.append(Finding(
                path=path, line=None, code=str(code),
                message=str(chk.get("message", "")),
                severity="blocking", tool="lint",
            ))
    return findings
 
 
def _parse_typecheck(output: str, root: Path) -> list[Finding]:
    """pyright ``error`` lines from the wrapper's ``output`` text."""
    findings: list[Finding] = []
    for line in output.splitlines():
        m = _PYRIGHT_RE.match(line)
        if not m or m.group("sev") != "error":
            continue
        msg = m.group("msg")
        rule = _RULE_RE.search(msg)
        if rule:
            code = rule.group(1)
            message = _RULE_RE.sub("", msg).strip()
        else:
            code, message = "type-error", msg.strip()
        findings.append(Finding(
            path=_norm_path(m.group("path"), root), line=int(m.group("line")),
            code=code, message=message, severity="blocking", tool="typecheck",
        ))
    return findings
 
 
def _classify_lint(outcome: _CmdOutcome, root: Path) -> tuple[str, list[Finding]]:
    if outcome.error == "timeout":
        return TIMEOUT, []
    if outcome.error == "missing_binary":
        return INFRA_ERROR, []
    data = _extract_json(outcome.stdout)
    if data is None:
        # No JSON: a rejected flag / unknown profile / usage error → infra; else unparseable.
        return (INFRA_ERROR if (outcome.returncode or 0) != 0 else UNPARSEABLE), []
    findings = _parse_lint(data, root)
    return (FINDINGS if findings else CLEAN), findings
 
 
def _classify_typecheck(outcome: _CmdOutcome, root: Path) -> tuple[str, list[Finding]]:
    if outcome.error == "timeout":
        return TIMEOUT, []
    if outcome.error == "missing_binary":
        return INFRA_ERROR, []
    data = _extract_json(outcome.stdout)
    if data is None:
        return (INFRA_ERROR if (outcome.returncode or 0) not in (0, 1) else UNPARSEABLE), []
    exit_code = int(data.get("exit_code", outcome.returncode or 0) or 0)
    if exit_code == 0:
        return CLEAN, []
    if exit_code != 1:
        return INFRA_ERROR, []  # pyright fatal/config error
    findings = _parse_typecheck(str(data.get("output", "")), root)
    return (FINDINGS if findings else UNPARSEABLE), findings
 
 
def _map_test_targets(changed_files: list[str], root: Path) -> list[str]:
    """Conservative changed-file→test mapping (OQ-4): ``test_<stem>.py`` under any
    ``tests/`` dir. Never the full suite; missing mappings contribute nothing."""
    targets: list[str] = []
    for rel in changed_files:
        stem = Path(rel).stem
        if stem.startswith("test_"):
            if (root / rel).exists():
                targets.append(rel)
            continue
        for cand in root.glob(f"**/tests/**/test_{stem}.py"):
            targets.append(cand.relative_to(root).as_posix())
    # de-dup, preserve order
    seen: set[str] = set()
    return [t for t in targets if not (t in seen or seen.add(t))]
 
 
def _classify_tests(outcome: _CmdOutcome) -> str:
    if outcome.error == "timeout":
        return TIMEOUT
    if outcome.error == "missing_binary":
        return INFRA_ERROR
    rc = outcome.returncode or 0
    if rc == 0:
        return CLEAN
    if rc == 5:  # pytest: no tests collected — nothing to verify, not a failure
        return CLEAN
    return FINDINGS
 
 
def _subtract_baseline(post: list[Finding], baseline: "Counter[tuple[str, str, str]] | None") -> list[Finding]:
    """Return only findings whose ``(path, code, normalized-message)`` key is not
    covered (count-wise) by *baseline* (plan 137 §3.8)."""
    if not baseline:
        return post
    remaining: Counter = Counter(baseline)
    new: list[Finding] = []
    for f in post:
        k = f.key()
        if remaining.get(k, 0) > 0:
            remaining[k] -= 1
        else:
            new.append(f)
    return new
 
 
def run_verification(
    changed_files: list[str],
    baseline: "Counter[tuple[str, str, str]] | None",
    cfg: "VerificationConfig",  # noqa: F821 — typed in settings; avoid an import cycle
    *,
    workspace_root: str | Path | None = None,
    presenter: object | None = None,
    runner: Runner | None = None,
) -> VerificationResult:
    """Verify *changed_files* and return only baseline-new findings (plan 137 §3.3).
 
    Deterministic, time-boxed, classified, fail-open. Brackets the subprocess work
    with a presenter spinner (no-op on plain presenters). Never streams raw output,
    never blocks the turn, never calls a model.
    """
    func = "run_verification"
    run = runner or _default_runner
    root = _workspace_root(workspace_root)
    targets = sorted({str(p).replace("\\", "/") for p in changed_files if str(p).strip()})
 
    if not targets:  # e.g. a turn whose every write_file failed (§3.3 / DS-3.1)
        return VerificationResult(clean=True, skip_reason="no files changed",
                                  lint_status=SKIPPED, typecheck_status=SKIPPED)
    if len(targets) > cfg.max_files:
        _dbg("OC_CHAT.MAIN", func, f"degraded: {len(targets)} > max_files {cfg.max_files}", 2)
        return VerificationResult(clean=True, degraded=True,
                                  skip_reason=f"too many changed files ({len(targets)} > {cfg.max_files})")
 
    _dbg("OC_CHAT.MAIN", func, f"verifying {len(targets)} file(s) profile={cfg.profile}", 2)
    spinner = _spinner_start(presenter, "verifying changes…")
    findings: list[Finding] = []
    degraded = False
    lint_status = typecheck_status = tests_status = SKIPPED
    try:
        if cfg.lint:
            # --no-cache: oct lint's cache is content-keyed and would return a STALE
            # path for an identical-content file (verified) — the sensor needs the
            # real changed-file path for the model and the baseline key (§3.3).
            outcome = run(
                _oct_argv(["lint", *targets, "--profile", cfg.profile,
                           "--json", "--jobs", str(cfg.jobs), "--no-cache"]),
                cfg.timeout_s,
            )
            lint_status, lint_findings = _classify_lint(outcome, root)
            findings.extend(lint_findings)
            degraded = degraded or lint_status in (INFRA_ERROR, TIMEOUT, UNPARSEABLE)
        if cfg.typecheck:
            outcome = run(_oct_argv(["typecheck", *targets, "--json"]), cfg.timeout_s)
            typecheck_status, tc_findings = _classify_typecheck(outcome, root)
            findings.extend(tc_findings)
            degraded = degraded or typecheck_status in (INFRA_ERROR, TIMEOUT, UNPARSEABLE)
        if cfg.tests:
            test_targets = _map_test_targets(targets, root)
            if test_targets:
                outcome = run([sys.executable, "-m", "pytest", *test_targets, "-q"], cfg.timeout_s)
                tests_status = _classify_tests(outcome)
                degraded = degraded or tests_status in (INFRA_ERROR, TIMEOUT)
            else:
                tests_status = CLEAN
    finally:
        _spinner_stop(presenter, spinner)
 
    new_findings = _subtract_baseline(findings, baseline)
    # ``clean`` means "nothing actionable for the model": no NEW findings (baseline-
    # subtracted). A degraded run with no new findings is clean-but-degraded (a
    # one-line skip note, never a block).
    result = VerificationResult(
        clean=not new_findings,
        findings=new_findings,
        degraded=degraded,
        skip_reason=("verification degraded" if degraded and not new_findings else None),
        lint_status=lint_status,
        typecheck_status=typecheck_status,
        tests_status=tests_status,
    )
    _dbg("OC_CHAT.MAIN", func,
         f"result clean={result.clean} new={len(new_findings)} "
         f"lint={lint_status} typecheck={typecheck_status} tests={tests_status} degraded={degraded}", 2)
    return result
 
 
def capture_baseline(
    cfg: "VerificationConfig",  # noqa: F821
    *,
    workspace_root: str | Path | None = None,
    runner: Runner | None = None,
) -> "Counter[tuple[str, str, str]]":
    """Capture a session-local lint+typecheck finding multiset over the bounded
    ``baseline_scope`` (plan 137 §3.8). In-process only; never ``oct lint
    --baseline``; fail-open to an empty baseline (degraded ⇒ first findings simply
    unsuppressed). Returns a ``Counter`` of finding keys.
    """
    func = "capture_baseline"
    run = runner or _default_runner
    root = _workspace_root(workspace_root)
    scope = _resolve_baseline_scope(cfg.baseline_scope, root)
    keys: Counter = Counter()
    if not scope:
        _dbg("OC_CHAT.MAIN", func, "no baseline scope resolved; empty baseline", 2)
        return keys
    if cfg.lint:
        outcome = run(_oct_argv(["lint", *scope, "--profile", cfg.profile,
                                 "--json", "--jobs", str(cfg.jobs), "--no-cache"]),
                      cfg.baseline_timeout_s)
        status, findings = _classify_lint(outcome, root)
        for f in findings:
            keys[f.key()] += 1
    if cfg.typecheck:
        outcome = run(_oct_argv(["typecheck", *scope, "--json"]), cfg.baseline_timeout_s)
        status, findings = _classify_typecheck(outcome, root)
        for f in findings:
            keys[f.key()] += 1
    _dbg("OC_CHAT.MAIN", func, f"baseline captured: {len(keys)} distinct finding key(s)", 2)
    return keys
 
 
def _resolve_baseline_scope(scope: tuple[str, ...] | None, root: Path) -> list[str]:
    """Bounded baseline targets. Explicit scope wins; else the top-level source
    packages under the workspace root (dirs with a ``.py`` file, minus common
    non-source dirs) — never an unbounded full-workspace scan."""
    if scope:
        return [s for s in scope if (root / s).exists()]
    skip = {"__pycache__", "logs", "memory_store", "_backup", "nulldir",
            "venv", ".venv", "_work", "sessions", ".git", "node_modules"}
    out: list[str] = []
    try:
        for child in sorted(root.iterdir()):
            if not child.is_dir() or child.name in skip or child.name.startswith("."):
                continue
            if any(child.rglob("*.py")):
                out.append(child.name)
    except OSError:
        return []
    return out
 
 
def render_observation(result: VerificationResult, max_findings: int = 10) -> str:
    """Compact the result into the ``[verification]`` message body (plan 137 §3.5).
 
    A header line + the top ``max_findings`` items as ``path:line — code: message``
    (truncated) + a one-line instruction. Never raw tracebacks or full JSON.
    """
    n_lint = sum(1 for f in result.findings if f.tool == "lint")
    n_type = sum(1 for f in result.findings if f.tool == "typecheck")
    n_test = sum(1 for f in result.findings if f.tool == "tests")
    n_files = len({f.path for f in result.findings})
    parts = [
        f"[verification] {n_lint} new lint + {n_type} new type"
        + (f" + {n_test} test" if n_test else "")
        + f" finding(s) in {n_files} changed file(s):"
    ]
    for f in result.findings[:max_findings]:
        loc = f"{f.path}:{f.line}" if f.line else f.path
        msg = (f.message[:160] + "…") if len(f.message) > 160 else f.message
        parts.append(f"  - {loc} — {f.code}: {msg}")
    extra = len(result.findings) - max_findings
    if extra > 0:
        parts.append(f"  …and {extra} more.")
    parts.append("Fix these in the changed files and re-verify, or explain why they are acceptable.")
    return "\n".join(parts)
 
 
def changed_paths_from_result(
    tool_name: str,
    tool_args: dict | None,
    result: object,
    *,
    workspace_root: str | Path | None = None,
) -> list[str]:
    """Workspace-root-relative paths a **successful** mutating tool wrote (§3.4).
 
    From the tool's *result* (not its args): ``write_file`` adds ``actual_path``
    when not chunked, else every ``chunks[*].chunk_path`` (the logical target is
    never materialised in chunked mode); ``oct_new`` returns the generic
    ``_run_oct`` shape, so on ``returncode == 0`` and not ``dry_run`` it attributes
    the validated ``path`` arg, the ``tests/`` companion (when ``test=True``), and
    any ancestor ``__init__.py`` package markers that exist post-call (newly-created
    header-less markers are a real blocking lint failure; the baseline suppresses
    pre-existing ones). Failed / dry-run / error results add nothing.
    """
    if not isinstance(result, dict) or result.get("error"):
        return []
    root = _workspace_root(workspace_root)
    if tool_name == "write_file":
        if result.get("chunked"):
            return [c["chunk_path"] for c in (result.get("chunks") or [])
                    if isinstance(c, dict) and c.get("chunk_path")]
        ap = result.get("actual_path")
        return [str(ap)] if ap else []
    if tool_name == "oct_new":
        rc = result.get("returncode", 1)
        if rc is None or int(rc) != 0:
            return []
        args = tool_args if isinstance(tool_args, dict) else {}
        if args.get("dry_run"):
            return []
        path = args.get("path")
        if not path:
            return []
        out: list[str] = [str(path)]
        if args.get("test"):
            out.extend(_companion_test_paths(str(path), root))
        out.extend(_ancestor_package_markers(str(path), root))
        # de-dup, preserve order
        seen: set[str] = set()
        return [p for p in out if p and not (p in seen or seen.add(p))]
    return []
 
 
def _companion_test_paths(path: str, root: Path) -> list[str]:
    """Existence-checked ``test_<stem>.py`` companions for an ``oct new --test``
    target (no path guessing — only paths that actually exist are attributed)."""
    stem = Path(path).stem
    parent = Path(path).parent
    candidates = [
        parent / "tests" / f"test_{stem}.py",
        Path("tests") / f"test_{stem}.py",
        Path("tests") / "unit" / f"test_{stem}.py",
    ]
    return [c.as_posix() for c in candidates if (root / c).exists()]
 
 
def _ancestor_package_markers(path: str, root: Path) -> list[str]:
    """Existing ``__init__.py`` markers along *path*'s ancestor package dirs, under
    the workspace root (§3.4 — newly-created header-less ones surface; the baseline
    suppresses pre-existing)."""
    out: list[str] = []
    try:
        rel = Path(path)
        if rel.is_absolute():
            rel = rel.relative_to(root)
    except (ValueError, OSError):
        return []
    for parent in rel.parents:
        if parent == Path("."):
            break
        marker = parent / "__init__.py"
        if (root / marker).exists():
            out.append(marker.as_posix())
    return out
 
 
def _spinner_start(presenter: object | None, label: str) -> object | None:
    fn = getattr(presenter, "spinner_start", None)
    if callable(fn):
        try:
            return fn(label)
        except Exception:  # noqa: BLE001 — a presenter spinner must never break the turn
            return None
    return None
 
 
def _spinner_stop(presenter: object | None, handle: object | None) -> None:
    fn = getattr(presenter, "spinner_stop", None)
    if callable(fn):
        try:
            fn(handle)
        except Exception:  # noqa: BLE001 — non-fatal
            pass
 
 
__all__ = [
    "Finding",
    "VerificationResult",
    "run_verification",
    "capture_baseline",
    "render_observation",
    "changed_paths_from_result",
]