FabledScribe/src/scribe/services/coverage.py

"""Pattern-library coverage — what fraction of a bound repo's shapes have a
recorded snippet (#2692, forge job 3 of decision #2686).

The all-shapes doctrine says every shape gets recorded at first build. This
module is the hoping→knowing move: it enumerates the definitions that exist in
a project's bound repos (via the forge, one archive download per repo) and
compares them against recorded snippet locations, so "record everything"
becomes a watched number instead of an aspiration.

The definition extractor MIRRORS the write-path hook's awk rules
(plugin/hooks/scribe_prior_art.sh, ARM 1) — one shared notion of "a
definition" between the hook and the server, so the metric and the backstop
agree on what counts. The two are pinned together by shared test vectors in
tests/test_pattern_coverage.py; change one, change both.

The number is an ESTIMATE and every surface must say so: keyword extraction
over-counts (private one-offs, generated code that slips the dir filter) and
under-counts (keyword-less declaration syntax — C/Java/Dart — needs a real
parser and is out of scope, exactly as it is for the hook). The trend carries
the meaning, like the usage counters; the raw number is not a grade.

Compute is on demand + cached with a freshness stamp — recomputed on webhook
push and explicit refresh, NEVER in the request path of enter_project, which
only ever reads the cache.
"""
from __future__ import annotations

import io
import json
import logging
import posixpath
import re
import tarfile
from datetime import datetime, timezone

from scribe.services.forge import ForgeSelector, get_forges
from scribe.services.repo_bindings import keys_for_project
from scribe.services.settings import get_setting, set_setting

logger = logging.getLogger(__name__)

# Cache key in the settings KV, on the project OWNER's user_id — the same
# channel the scheduler's last-run summary uses for machine-written state.
_CACHE_KEY_PREFIX = "pattern_coverage_"

# Files whose content can't hold definitions — the hook's skip list, verbatim,
# plus sourcemaps (which are JSON in a trenchcoat).
_SKIP_SUFFIXES = (
    ".md", ".mdx", ".txt", ".rst", ".json", ".lock", ".log", ".csv", ".tsv",
    ".svg", ".png", ".jpg", ".jpeg", ".gif", ".ico", ".pdf", ".map",
)

# Vendored/generated trees would swamp the metric with shapes nobody should
# record — the dunder-skip lesson at directory scale: guaranteed noise teaches
# people to ignore the number.
_SKIP_DIRS = frozenset({
    "node_modules", "vendor", "dist", "build", "target",
    "__pycache__", ".git", ".venv", "venv",
})

# A single source file bigger than this is almost certainly generated or
# vendored (bundles, lockstep protos) — skipped, and part of why the number
# is labeled an estimate.
_MAX_FILE_BYTES = 1_000_000


# --- the definition extractor (mirror of scribe_prior_art.sh ARM 1) ----------

_CSS_RE = re.compile(r"^\s*\.([A-Za-z][A-Za-z0-9_-]*)\s*[,{]")
# Leading declaration modifiers, so the definition keyword is the first word
# regardless of language (export/pub/private/suspend/...).
_MODIFIERS_RE = re.compile(
    r"^(?:(?:pub(?:\([a-z]+\))?|export|default|private|internal|protected"
    r"|public|static|suspend|async|open|sealed|data|abstract|final|inline"
    r"|unsafe|extern|override)\s+)*"
)
# Go method with receiver: func (r *T) Name(
_GO_METHOD_RE = re.compile(r"^func\s*\([^)]*\)\s*([A-Za-z_][A-Za-z0-9_]*)")
# Keyword-announced definitions, functions and named types alike. `impl` is
# excluded on purpose — several per type is normal Rust, not duplication.
_KEYWORD_RE = re.compile(
    r"^(?:function|def|class|func|fun|fn|sub|struct|trait|interface|enum"
    r"|object|protocol|type)\s+([A-Za-z_$][A-Za-z0-9_$]*)"
)
# Arrow/expression assignment: const name = (…) / let name = async (
_ARROW_RE = re.compile(
    r"^(?:const|let)\s+([A-Za-z_$][A-Za-z0-9_$]*)\s*=\s*(?:async\s*)?[(<]"
)


def extract_shapes(text: str) -> list[tuple[str, str]]:
    """Every (kind, name) this text DEFINES — kind is "css" or "sym".

    Rule-for-rule mirror of the hook's awk program: first match wins per
    line, dunders are skipped (every class defines __init__ — guaranteed
    noise), duplicates within one text count once.
    """
    seen: set[tuple[str, str]] = set()
    out: list[tuple[str, str]] = []
    for raw in text.splitlines():
        m = _CSS_RE.match(raw)
        if m:
            shape = ("css", m.group(1))
        else:
            line = _MODIFIERS_RE.sub("", raw.lstrip())
            if m := _GO_METHOD_RE.match(line):
                shape = ("sym", m.group(1))
            elif m := _KEYWORD_RE.match(line):
                name = m.group(1)
                if name.startswith("__") and name.endswith("__"):
                    continue
                shape = ("sym", name)
            elif m := _ARROW_RE.match(line):
                shape = ("sym", m.group(1))
            else:
                continue
        if shape not in seen:
            seen.add(shape)
            out.append(shape)
    return out


def scannable(path: str) -> bool:
    """Should this repo file be scanned for shapes at all?"""
    parts = path.split("/")
    if any(p in _SKIP_DIRS for p in parts[:-1]):
        return False
    return not path.lower().endswith(_SKIP_SUFFIXES)


def shapes_from_archive(blob: bytes) -> list[tuple[str, str, str]]:
    """(path, kind, name) for every definition in a repo tarball.

    Forge archives wrap content in a single top-level directory (repo-ref/);
    that component is stripped so paths match recorded snippet locations,
    which are repo-relative. Non-UTF-8 files are binaries and skipped.
    """
    shapes: list[tuple[str, str, str]] = []
    with tarfile.open(fileobj=io.BytesIO(blob), mode="r:gz") as tar:
        for member in tar:
            if not member.isfile() or "/" not in member.name:
                continue
            path = member.name.split("/", 1)[1]
            if not path or not scannable(path) or member.size > _MAX_FILE_BYTES:
                continue
            handle = tar.extractfile(member)
            if handle is None:
                continue
            try:
                text = handle.read().decode("utf-8")
            except UnicodeDecodeError:
                continue
            shapes.extend((path, kind, name) for kind, name in extract_shapes(text))
    return shapes


# --- matching shapes against recorded locations ------------------------------


def _norm_symbol(kind_or_symbol: str) -> str:
    # CSS shapes and recorded CSS symbols may or may not carry the leading
    # dot; compare without it so ".btn-primary" and "btn-primary" agree.
    return kind_or_symbol.lstrip(".").strip()


def _location_covers(loc_path: str, loc_symbol: str, path: str, name: str) -> bool:
    if _norm_symbol(loc_symbol) != _norm_symbol(name):
        return False
    if not loc_path:
        # Symbol-only record: the symbol match is all the claim there is.
        return True
    # The drift check's location semantics, not a second copy of them: exact
    # file, or the recorded path is a directory the file lives under.
    from scribe.services.snippets import _path_touches

    return _path_touches(loc_path, path)


def match_shapes(
    shapes: list[tuple[str, str, str]],
    recorded: list[tuple[str, str]],
) -> list[tuple[str, str, str, bool]]:
    """Each shape with whether some recorded (path, symbol) location covers it.

    Symbol-less recorded locations never cover a shape — a whole-file record
    makes no claim about any particular definition inside it. The recorded
    repo NAME is deliberately not consulted: it is free-form ("Scribe") and
    the project binding already did the scoping; on a project binding several
    repos this can over-credit a same-named symbol, which the estimate label
    owns.
    """
    usable = [(p, s) for p, s in recorded if (s or "").strip()]
    return [
        (
            path,
            kind,
            name,
            any(_location_covers(lp, ls, path, name) for lp, ls in usable),
        )
        for path, kind, name in shapes
    ]


def largest_gaps(
    matched: list[tuple[str, str, str, bool]], *, top: int = 3
) -> list[dict]:
    """The directories with the most uncovered shapes — where a backlog
    session should start, named the way the repo names them."""
    by_dir: dict[str, dict[str, int]] = {}
    for path, _kind, _name, covered in matched:
        d = posixpath.dirname(path) or "(root)"
        row = by_dir.setdefault(d, {"total": 0, "uncovered": 0})
        row["total"] += 1
        if not covered:
            row["uncovered"] += 1
    ranked = sorted(
        by_dir.items(), key=lambda kv: (-kv[1]["uncovered"], kv[0])
    )
    return [
        {"dir": d, "uncovered": row["uncovered"], "total": row["total"]}
        for d, row in ranked[:top]
        if row["uncovered"]
    ]


# --- compute, cache, surface -------------------------------------------------


async def _recorded_locations(user_id: int, project_id: int) -> list[tuple[str, str]]:
    """(path, symbol) for every location of every live snippet in a project."""
    from sqlalchemy import select

    from scribe.models import async_session
    from scribe.models.note import Note
    from scribe.services.snippets import SNIPPET_NOTE_TYPE, snippet_fields

    async with async_session() as session:
        rows = await session.execute(
            select(Note).where(
                Note.user_id == user_id,
                Note.project_id == project_id,
                Note.note_type == SNIPPET_NOTE_TYPE,
                Note.deleted_at.is_(None),
            )
        )
        notes = list(rows.scalars().all())
    out: list[tuple[str, str]] = []
    for note in notes:
        for loc in snippet_fields(note).get("locations") or []:
            out.append((loc.get("path") or "", loc.get("symbol") or ""))
    return out


async def compute_coverage(
    user_id: int, project_id: int, *, selector: ForgeSelector | None = None
) -> dict | None:
    """Measure a project's pattern-library coverage against its bound repos.

    None means "nothing to measure" — the owner's keyring serves none of the
    project's bound repos (#2778). That is the ordinary state for a
    forge-less user and every caller treats it as silence, not failure.
    Forge errors (unreachable, bad token) RAISE — the two callers are a
    refresh button and a background task, and both want to know.

    ``user_id`` is the project OWNER's id: the cache lives there, and the
    keyring resolved here must be the same one every other read uses.
    """
    if selector is None:
        selector = await get_forges(user_id, project_id)
    if not selector.configured:
        return None

    repos: list[dict] = []
    matched_all: list[tuple[str, str, str, bool]] = []
    recorded = await _recorded_locations(user_id, project_id)
    for key in await keys_for_project(user_id, project_id):
        hit = selector.resolve(key)
        if hit is None:
            continue  # bound to a host no connection serves
        forge, api_repo = hit
        ref = await forge.default_branch(api_repo)
        shapes = shapes_from_archive(await forge.archive(api_repo, ref))
        matched = match_shapes(shapes, recorded)
        matched_all.extend(matched)
        repos.append({
            "repo": key,
            "ref": ref,
            "total": len(matched),
            "recorded": sum(1 for *_x, covered in matched if covered),
        })
    if not repos:
        return None

    return {
        "total": len(matched_all),
        "recorded": sum(1 for *_x, covered in matched_all if covered),
        # Honesty flag, not decoration: every surface that shows the number
        # is expected to carry it through.
        "estimate": True,
        "computed_at": datetime.now(timezone.utc).isoformat(),
        "repos": repos,
        "largest_gaps": largest_gaps(matched_all),
    }


async def refresh_coverage(
    user_id: int, project_id: int, *, selector: ForgeSelector | None = None
) -> dict | None:
    """Compute and cache. The only writer of the cache key."""
    coverage = await compute_coverage(user_id, project_id, selector=selector)
    if coverage is not None:
        await set_setting(
            user_id, f"{_CACHE_KEY_PREFIX}{project_id}", json.dumps(coverage)
        )
    return coverage


async def cached_coverage(user_id: int, project_id: int) -> dict | None:
    """The last computed summary, or None — never computes."""
    raw = await get_setting(user_id, f"{_CACHE_KEY_PREFIX}{project_id}", "")
    if not raw:
        return None
    try:
        parsed = json.loads(raw)
    except ValueError:
        return None
    return parsed if isinstance(parsed, dict) else None


def coverage_line(coverage: dict) -> str:
    """The one-line evidence-carrying summary enter_project surfaces."""
    day = (coverage.get("computed_at") or "")[:10]
    line = (
        f"pattern-library coverage: {coverage.get('recorded', 0)}"
        f"/{coverage.get('total', 0)} shapes recorded"
        f" (estimate{', computed ' + day if day else ''})"
    )
    gaps = [g["dir"] for g in coverage.get("largest_gaps") or []]
    if gaps:
        line += "; largest gaps: " + ", ".join(gaps)
    return line