info-tech-canon/src/info_tech_canon/maintenance.py

"""Reproducible projections, portable contract exports, and read measurements."""

from collections import Counter
from dataclasses import replace
from datetime import datetime, timezone
import hashlib
import io
import json
from pathlib import Path
import re
import shutil
import statistics
import subprocess
import tarfile
import tempfile
import time

from . import generation


def scope_inventory(context) -> dict:
    return {"artifact_count": len(context.infospace.artifacts),
            "kinds": dict(sorted(Counter(a.kind for a in context.infospace.artifacts).items())),
            "standards": sorted(a.id for a in context.infospace.artifacts if a.kind == "standard"),
            "models": sorted(a.id for a in context.infospace.artifacts if a.kind == "model")}


def check_generated(context) -> dict:
    """Render only into a disposable copy; never repair the source as a check."""
    with tempfile.TemporaryDirectory(prefix="canon-freshness-") as temporary:
        target = Path(temporary) / "infospace"
        shutil.copytree(context.infospace_root, target)
        copied = replace(context, infospace_root=target)
        files = {}
        for render in (generation.generate_indexes, generation.generate_tree,
                       generation.generate_agent_briefs):
            for item in render(copied)["files"]:
                path = Path(item["path"])
                files[str(path.relative_to(target))] = path.read_bytes()
        stale = []
        for relative, expected in files.items():
            original = context.infospace_root / relative
            if not original.is_file() or original.read_bytes() != expected:
                stale.append(relative)
        return {"ok": not stale, "checked": len(files), "stale": sorted(stale)}


def source_evidence(root: Path) -> dict:
    digest = hashlib.sha256()
    # Include source and generated corpus; exclude reports to avoid self-hashing.
    for path in sorted(root.rglob("*")):
        if path.is_file() and "validation" not in path.relative_to(root).parts:
            digest.update(str(path.relative_to(root)).encode() + b"\0" + path.read_bytes() + b"\0")
    def git(*args):
        try:
            result = subprocess.run(["git", "-C", str(root.parent), *args],
                                    capture_output=True, text=True, check=False)
        except FileNotFoundError:
            return None
        return result.stdout.strip() if result.returncode == 0 else None
    status = git("status", "--porcelain")
    return {"generated_at": datetime.now(timezone.utc).isoformat(),
            "revision": git("rev-parse", "HEAD"),
            "dirty": bool(status) if status is not None else None,
            "corpus_sha256": digest.hexdigest(),
            "digest_scope": "all infospace files except validation reports"}


def export_emission_bundle(root: Path, destination: Path) -> dict:
    """Content-addressed deterministic tar; a changed contract gets a new name."""
    paths = ["schemas/emission-cadence.schema.yaml",
             "standards/emission-cadence/InfoTechCanonEmissionCadenceStandard.md",
             "standards/emission-cadence/examples/qonto-assistant.yaml"]
    files = {path: (root / path).read_bytes() for path in paths}
    manifest = {"contract": "emission-cadence/0.1", "status": "draft",
                "files": {name: hashlib.sha256(data).hexdigest() for name, data in files.items()},
                "semantic_checks": ["unique source_id in sources"],
                "adoption_evidence": "Two independent source-owned implementations required; example does not count."}
    files["manifest.json"] = (json.dumps(manifest, sort_keys=True, indent=2) + "\n").encode()
    output = io.BytesIO()
    with tarfile.open(fileobj=output, mode="w", format=tarfile.USTAR_FORMAT) as archive:
        for name, data in sorted(files.items()):
            info = tarfile.TarInfo(name)
            info.size = len(data)
            info.mode = 0o644
            archive.addfile(info, io.BytesIO(data))
    data = output.getvalue()
    digest = hashlib.sha256(data).hexdigest()
    destination.mkdir(parents=True, exist_ok=True)
    path = destination / f"emission-cadence-0.1-{digest}.tar"
    try:
        with path.open("xb") as stream:
            stream.write(data)
    except FileExistsError:
        if path.read_bytes() != data:
            raise ValueError(f"Refusing to overwrite different bundle contents: {path}")
    return {"ok": True, "path": str(path), "sha256": digest, "manifest": manifest}


def benchmark_reads(root: Path, runs: int = 10) -> dict:
    from .service import inspect_canon
    samples = []
    for _ in range(runs):
        started = time.perf_counter()
        inspect_canon(root)
        samples.append((time.perf_counter() - started) * 1000)
    return {"ok": True, "operation": "inspect", "runs": runs,
            "median_ms": statistics.median(samples), "max_ms": max(samples),
            "cache": "none; benchmark does not establish a production latency requirement"}


CONCEPT_BOLD = re.compile(
    r"^(?:A |An |The )?\*\*(?P<name>[A-Z][A-Za-z0-9 /\-]{2,60}?)\*\*"
    r"\s*(?:—|-|is|are|defines|denotes|identifies|represents)\b"
)
CONCEPT_HEADING = re.compile(r"^#{2,4}\s+\d+(?:\.\d+)*\.?\s+(?P<name>[A-Z][A-Za-z0-9 /\-]{2,60})\s*$")
CONCEPT_TABLE_HEADER = re.compile(r"^\|\s*(?:Concept|Term|Name|Entity)\s*\|", re.IGNORECASE)
CONCEPT_TABLE_ROW = re.compile(r"^\|\s*`?(?P<name>[A-Z][A-Za-z0-9 /\-]{2,60}?)`?\s*\|")
SOURCE_ONLY_PREFIXES = ("assimilation/", "seeds/", "incoming/")


def _heading_is_defined(lines: list[str], start: int, name: str) -> bool:
    """A numbered heading names a concept only if a definition follows it."""
    pattern = re.compile(r"^(?:A |An |The )?\*\*" + re.escape(name) + r"\*\*")
    return any(pattern.match(lines[offset].strip()) for offset in range(start + 1, min(start + 6, len(lines))))


def _extract_concepts(text: str) -> dict[str, list[str]]:
    lines = text.splitlines()
    found: dict[str, set[str]] = {"bold": set(), "heading": set(), "table": set()}
    in_concept_table = False
    for number, raw in enumerate(lines):
        line = raw.strip()
        if CONCEPT_TABLE_HEADER.match(line):
            in_concept_table = True
            continue
        if in_concept_table:
            if not line.startswith("|"):
                in_concept_table = False
            elif not set(line) <= set("|- :"):
                match = CONCEPT_TABLE_ROW.match(line)
                if match:
                    found["table"].add(match.group("name").strip())
                continue
        match = CONCEPT_BOLD.match(line)
        if match:
            found["bold"].add(match.group("name").strip())
            continue
        match = CONCEPT_HEADING.match(line)
        if match and _heading_is_defined(lines, number, match.group("name").strip()):
            found["heading"].add(match.group("name").strip())
    return {form: sorted(names) for form, names in found.items()}


def concept_candidates(context) -> dict:
    """Measure declared concepts against candidates defined in artifact prose.

    Candidates are review input, never ownership. A concept becomes owned by
    being declared, not by being matched here.
    """
    ownership = generation.concept_ownership(context)
    declared_by_owner: dict[str, set[str]] = {}
    for item in ownership["concepts"]:
        declared_by_owner.setdefault(item["owner"], set()).add(
            generation._normalize_concept(item["concept"])
        )

    artifacts = []
    for artifact in sorted(context.infospace.artifacts, key=lambda item: item.id):
        if artifact.path.startswith(SOURCE_ONLY_PREFIXES) or not artifact.path.endswith(".md"):
            continue
        path = context.infospace_root / artifact.path
        if not path.exists():
            continue
        forms = _extract_concepts(path.read_text(encoding="utf-8"))
        declared = declared_by_owner.get(artifact.id, set())
        candidates = sorted({name for names in forms.values() for name in names})
        undeclared = [name for name in candidates
                      if generation._normalize_concept(name) not in declared]
        artifacts.append({
            "artifact": artifact.id,
            "path": artifact.path,
            "declared_count": len(declared),
            "declares_frontmatter": bool(
                generation._frontmatter(path).get("owned_concepts")
            ),
            "candidate_count": len(candidates),
            "undeclared_count": len(undeclared),
            "by_form": {form: len(names) for form, names in forms.items()},
            "undeclared": undeclared,
        })

    silent = [item["artifact"] for item in artifacts
              if not item["declares_frontmatter"] and item["candidate_count"]]
    return {
        "artifact_count": len(artifacts),
        "declared_total": sum(item["declared_count"] for item in artifacts),
        "candidate_total": sum(item["candidate_count"] for item in artifacts),
        "undeclared_total": sum(item["undeclared_count"] for item in artifacts),
        "silent_artifacts": sorted(silent),
        "artifacts": artifacts,
    }