"""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 hashlib
import io
import json
import logging
import posixpath
import re
import tarfile
from typing import NamedTuple
from datetime import datetime, timedelta, timezone

from scribe.services.forge import ForgeSelector, get_forges
from scribe.services.repo_bindings import bindings_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.
# v2 suffix with #2788: the payload shape inverted (accounted/unclassified);
# pre-ledger blobs under the old key simply stop being found, so the card
# honestly reads "not measured yet" until the first ledger-era refresh.
_CACHE_KEY_PREFIX = "pattern_coverage_v2_"

# 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*)?[(<]"
)


class Definition(NamedTuple):
    """One extracted definition with its content fingerprint (#2792).

    `signature` is the definition line itself; `body_sha` hashes the block
    whitespace- and comment-insensitively; `body` is the block's text, held
    only for the duration of a refresh (the proposer matches on it) and
    never stored.
    """

    kind: str
    name: str
    signature: str
    body_sha: str
    body: str
    line: int = -1   # 0-based line the definition starts on (#2869)


def _definition_on(raw: str) -> tuple[str, str] | None:
    """The (kind, name) this one line defines, or None. First match wins —
    the same order the hook's awk program tries."""
    m = _CSS_RE.match(raw)
    if m:
        return ("css", m.group(1))
    line = _MODIFIERS_RE.sub("", raw.lstrip())
    if m := _GO_METHOD_RE.match(line):
        return ("sym", m.group(1))
    if m := _KEYWORD_RE.match(line):
        name = m.group(1)
        if name.startswith("__") and name.endswith("__"):
            return None
        # `type` announces a definition only when something is declared after
        # the name (`type Foo = …`, `type Foo struct {`); an import specifier
        # (`import { type Foo, bar }`) is the same two words and defines
        # nothing — it showed up as a two-file "identical body" family (#2904).
        if line.startswith("type") and not re.search(r"[={]", line[m.end():]):
            return None
        return ("sym", name)
    if m := _ARROW_RE.match(line):
        return ("sym", m.group(1))
    return None


# A definition's block runs from its line until the next non-blank line at
# its own indentation or shallower that is not a closer — so a Python def ends
# at the next top-level statement, a braces block keeps its `}`, a CSS rule
# keeps its `}`. Capped so a generated monolith can't make one shape's
# fingerprint cover the file.
_BLOCK_CAP = 120
_CLOSERS = ("}", ")", "]", "end", "};", "});", ");", "})", "]);")
# Lines that don't change what a shape IS: comments and decorators. Dropped
# from the fingerprint so touching a comment above the next function doesn't
# read as this one's body changing.
_NOISE_PREFIXES = ("#", "//", "/*", "*", "*/", "@", "<!--", "-->")
_SIGNATURE_CAP = 300


def _indent(line: str) -> int:
    return len(line) - len(line.lstrip())


def _block_sha(lines: list[str]) -> str:
    kept = [
        " ".join(ln.split())
        for ln in lines
        if ln.strip() and not ln.lstrip().startswith(_NOISE_PREFIXES)
    ]
    return hashlib.sha1("\n".join(kept).encode("utf-8")).hexdigest()[:16]


def _declaration_count(lines: list[str]) -> int:
    """How many `prop: value` declarations a CSS block body carries."""
    body = " ".join(lines)
    return sum(1 for part in body.replace("}", "").split(";") if ":" in part)


def extract_definitions(text: str) -> list[Definition]:
    """Every definition this text makes, with signature + fingerprint.

    Duplicate (kind, name) within one text collapse to the first — the
    ledger's identity is per file, so a second definition of the same name
    (an overload, a re-declaration) is the same shape to it.
    """
    lines = text.splitlines()
    starts: list[tuple[int, str, str]] = []
    for i, raw in enumerate(lines):
        hit = _definition_on(raw)
        if hit:
            starts.append((i, hit[0], hit[1]))
    seen: set[tuple[str, str]] = set()
    out: list[Definition] = []
    for i, kind, name in starts:
        if (kind, name) in seen:
            continue
        seen.add((kind, name))
        base = _indent(lines[i])
        end = min(len(lines), i + _BLOCK_CAP)
        for j in range(i + 1, min(len(lines), i + _BLOCK_CAP)):
            ln = lines[j]
            if not ln.strip():
                continue
            if _indent(ln) <= base and ln.strip() not in _CLOSERS:
                end = j
                break
        block = lines[i:end]
        # A CSS rule's fingerprint is its DECLARATIONS, not its selector
        # (#2872): the row's identity already carries the selector, and the
        # question the fingerprint answers for derive grouping is "is this the
        # same rule under another name?" — .closed-msg / .error-block /
        # .success-msg with identical bodies are one dup group, not three
        # lonely rows. Sym blocks keep their signature line in the hash.
        if kind == "css":
            # One-line rules (`.x { color: red; }`) carry their declarations on
            # the selector line itself; a block that is only the selector plus
            # trailing blanks must not hash to the empty string (which grouped
            # 68 unrelated one-liners as one "copy" on first deploy, #2872).
            first = lines[i]
            brace = first.find("{")
            head = [first[brace + 1:]] if brace >= 0 and first[brace + 1:].strip() else []
            hashed = head + block[1:]
            if not any(x.strip() for x in hashed):
                hashed = block
            # A SINGLE declaration is not a shape (#2903): `color: var(--fs-
            # text-tertiary)` under .text-muted, .task-mark and .pin-badge-auto
            # is three meanings sharing one line, not three copies of one
            # rule — the first pay-down found 5-file "families" of exactly
            # this and nobody would consolidate them. Keep the selector in the
            # hash for one-liners, so they group only with same-name copies;
            # two declarations and up stay selector-agnostic.
            elif _declaration_count(hashed) < 2:
                hashed = block
        else:
            hashed = block
        out.append(Definition(
            kind, name, lines[i].strip()[:_SIGNATURE_CAP], _block_sha(hashed),
            "\n".join(block), i,
        ))
    return out


# --- by-construction scope (#2869) -------------------------------------------
#
# A Vue single-file component's `<style scoped>` rules and its `<script setup>`
# functions cannot be reached from any other file: they are one-offs by
# construction, not by judgment. The sync stamps them `scoped` (mechanical) so
# the human todo holds only shapes a person should look at, while the bodies
# stay in play for the proposer, derive grouping and divergence — the five
# auth views' identical rules were found exactly there. Unscoped `<style>` in
# a .vue and every non-.vue file stay ordinary.
_STYLE_OPEN_RE = re.compile(r"^\s*<style\b[^>]*\bscoped\b", re.IGNORECASE)
_STYLE_CLOSE_RE = re.compile(r"^\s*</style\s*>", re.IGNORECASE)


def scoped_definitions(path: str, text: str, defs: list[Definition]) -> set[tuple[str, str]]:
    """The (kind, name) pairs among ``defs`` that are one-offs by
    construction in this file: every sym in a .vue, and every css rule
    that starts inside a `<style scoped>` block. Empty for other files."""
    if not (path or "").lower().endswith(".vue"):
        return set()
    ranges: list[tuple[int, int]] = []
    open_at: int | None = None
    for i, ln in enumerate(text.splitlines()):
        if open_at is None and _STYLE_OPEN_RE.match(ln):
            open_at = i
        elif open_at is not None and _STYLE_CLOSE_RE.match(ln):
            ranges.append((open_at, i))
            open_at = None
    out: set[tuple[str, str]] = set()
    for d in defs:
        if d.kind == "sym":
            out.add((d.kind, d.name))
        elif any(a <= d.line <= b for a, b in ranges):
            out.add((d.kind, d.name))
    return out


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.
    """
    return [(d.kind, d.name) for d in extract_definitions(text)]


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)


class ArchiveShape(NamedTuple):
    """A definition located in a repo archive — what the sync upserts and
    the proposer matches. The leading (path, kind, name) triple is the
    ledger identity; the rest is the fingerprint and the transient body."""

    path: str
    kind: str
    name: str
    signature: str
    body_sha: str
    body: str
    scoped: bool = False   # one-off by construction (#2869)


def shapes_from_archive(blob: bytes) -> list[tuple[str, str, str]]:
    """(path, kind, name) for every definition in a repo tarball — the
    identity view of definitions_from_archive."""
    return [(d.path, d.kind, d.name) for d in definitions_from_archive(blob)]


def definitions_from_archive(blob: bytes) -> list[ArchiveShape]:
    """Every definition in a repo tarball, with its fingerprint and body.

    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[ArchiveShape] = []
    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
            defs = extract_definitions(text)
            scoped = scoped_definitions(path, text, defs)
            shapes.extend(
                ArchiveShape(
                    path, d.kind, d.name, d.signature, d.body_sha, d.body,
                    (d.kind, d.name) in scoped,
                )
                for d in defs
            )
    return shapes


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


# The covering predicate (location_covers) lives in services/shape_ledger.py
# since #2788 — the ledger's canonical marking and this module's readout are
# two consumers of ONE doctrine, and the ledger is its home.


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


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


async def _recorded_locations(
    user_id: int, project_id: int
) -> list[tuple[int, str, str]]:
    """(snippet_note_id, path, symbol) for every location of every live
    snippet in a project — the canonical-marking input (#2788): the id is what
    lets a ledger row point back at the snippet it references."""
    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[int, str, str]] = []
    for note in notes:
        for loc in snippet_fields(note).get("locations") or []:
            out.append(
                (int(note.id), 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:
    """Sync the shape ledger from the bound repos and read the accounting.

    Since #2788 this is the ledger's sync point, not just a measurement:
    every walk upserts the extracted shapes (new → unclassified, vanished →
    stamped), re-stamps snippet reference locations as canonical, and then
    reports the ACCOUNTING — how many shapes carry a classification at all —
    rather than the old "has a snippet" fraction. Unclassified is the todo
    (note 2786).

    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; the
    ledger is untouched in that case. 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.
    """
    from scribe.services import shape_ledger
    from scribe.services.forge import ForgeError

    if selector is None:
        selector = await get_forges(user_id, project_id)
    if not selector.configured:
        return None

    served: list[tuple[str, str]] = []
    recorded = await _recorded_locations(user_id, project_id)
    # The proposer's canon catalog, read once per refresh and shared across
    # the project's repos (#2792).
    canons = None
    proposer_stats = {"examined": 0, "proposed": 0, "semantic_checked": 0}
    for binding in await bindings_for_project(user_id, project_id):
        key = binding.repo_key
        hit = selector.resolve(key)
        if hit is None:
            continue  # bound to a host no connection serves
        forge, api_repo = hit
        # The binding's own ref when it names one (#2873: a dev-first project
        # has its ledger follow dev), else the forge's default branch.
        ref = binding.ref or await forge.default_branch(api_repo)
        definitions = definitions_from_archive(await forge.archive(api_repo, ref))
        # The head commit is provenance sugar on the ledger rows; failing to
        # learn it must not fail the sync — the ref names the point well
        # enough and the row timestamps carry the when.
        try:
            marker = await forge.latest_commit(api_repo, "", ref) or ref
        except ForgeError:
            marker = ref
        await shape_ledger.sync_repo_shapes(
            project_id, key, definitions, seen_marker=marker
        )
        served.append((key, ref))
        # Propose while the bodies are in hand — the one moment they exist.
        # Canonical marking below only touches rows the proposer leaves
        # alone (a canon's own location never gets a proposal), so the order
        # is immaterial; the proposer must not be able to fail the refresh.
        try:
            if canons is None:
                canons = await shape_ledger.canon_catalog(user_id)
            stats = await shape_ledger.propose_for_repo(
                user_id, project_id, key, definitions, canons=canons
            )
            for k in proposer_stats:
                proposer_stats[k] += stats.get(k, 0)
        except Exception:
            logger.warning("shape proposer failed for %s", key, exc_info=True)
    if not served:
        return None

    await shape_ledger.mark_canonicals(project_id, recorded)
    try:
        await shape_ledger.apply_derive_groups(project_id)
    except Exception:
        logger.warning("derive-first grouping failed", exc_info=True)
    # "Since the previous computation" — the cache's stamp. A first seed has
    # none, so nothing is new then. Read once; two passes use it: the
    # button-B flag (#2793) and the derive-new drift count (#2899).
    since = None
    try:
        previous = await get_setting(user_id, f"{_CACHE_KEY_PREFIX}{project_id}")
        if previous:
            stamp = (json.loads(previous) or {}).get("computed_at")
            since = datetime.fromisoformat(stamp) if stamp else None
    except Exception:
        logger.warning("previous coverage stamp unreadable", exc_info=True)
    # The button-B pass (#2793): shapes new since the PREVIOUS computation,
    # where a canon dominates.
    try:
        await shape_ledger.flag_divergence(project_id, since=since)
    except Exception:
        logger.warning("divergence pass failed", exc_info=True)

    # Project-wide readout, deliberately wider than this walk: a second bound
    # repo that was unreachable today still has live rows, and they count.
    rows = await shape_ledger.live_rows(project_id)
    counts = {"canonical": 0, "instance": 0, "variant": 0, "exempt": 0,
              "scoped": 0, "unclassified": 0}
    for row in rows:
        counts[row.status] = counts.get(row.status, 0) + 1
    by_repo: dict[str, dict[str, int]] = {}
    for row in rows:
        agg = by_repo.setdefault(row.repo_key, {"total": 0, "accounted": 0})
        agg["total"] += 1
        agg["accounted"] += row.status != "unclassified"

    unclassified = counts.pop("unclassified")
    proposals = shape_ledger.proposal_summary(rows)
    divergence = shape_ledger.divergence_summary(rows)
    derive_new = shape_ledger.derive_new_summary(rows, since=since)
    return {
        "total": len(rows),
        "accounted": len(rows) - unclassified,
        "unclassified": unclassified,
        "counts": counts,
        # The proposer's standing (#2792): canon proposals awaiting a
        # confirm, the largest derive-first groups, and what this refresh did.
        "proposed": proposals["proposed"],
        "derive_groups": proposals["derive_groups"],
        "top_canon": proposals.get("top_canon"),
        # Drift since the previous refresh (#2899): copies that joined a
        # duplicate family — what the arrival line names so drift is noticed
        # on entering, not found by an audit.
        "derive_new": derive_new,
        "proposer": proposer_stats,
        # The divergence readout (#2793): button B where button A is canon,
        # and judged shapes whose bodies moved since they were judged.
        "divergent": divergence["divergent"],
        "divergence": divergence["divergence"],
        "recheck": divergence["recheck"],
        # 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": [
            {"repo": key, "ref": ref,
             **by_repo.get(key, {"total": 0, "accounted": 0})}
            for key, ref in served
        ],
        "largest_gaps": largest_gaps([
            (r.path, r.kind, r.symbol, r.status != "unclassified")
            for r in rows
        ]),
    }


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


# The arrival-moment self-seed (#2802). A ledger that has never been computed
# used to wait for someone to find the UI Refresh button; entering the project
# is the moment the number is wanted, so entering is the moment it seeds.
_SEED_MAX_AGE = timedelta(hours=24)

# Projects with a refresh already running — concurrent enters must not fetch
# the same tarball N times. In-process on purpose: the cost being bounded is
# per-process forge traffic, and a rare double-fetch across workers is
# harmless (the upsert is idempotent).
_inflight_seed: set[int] = set()


async def refresh_if_stale(
    user_id: int, project_id: int, cached: dict | None = None
) -> None:
    """Background-refresh a project's ledger when its readout is absent or
    older than a day. Fire-and-forget material (background.spawn): every exit
    is quiet, every failure a WARNING — a seed must never surface as an
    enter_project error. ``user_id`` is the project OWNER (whose keyring and
    cache this is); pass ``cached`` when the caller already read it.
    """
    try:
        if cached is None:
            cached = await cached_coverage(user_id, project_id)
        if cached:
            try:
                computed = datetime.fromisoformat(cached.get("computed_at") or "")
                if datetime.now(timezone.utc) - computed < _SEED_MAX_AGE:
                    return
            except ValueError:
                pass  # an unreadable stamp reads as stale
        if project_id in _inflight_seed:
            return
        selector = await get_forges(user_id, project_id)
        if not selector.configured:
            return  # rule #115: forge-less stays exactly as it was
        _inflight_seed.add(project_id)
        try:
            await refresh_coverage(user_id, project_id, selector=selector)
        finally:
            _inflight_seed.discard(project_id)
    except Exception:
        logger.warning(
            "background coverage seed failed for project %s", project_id,
            exc_info=True,
        )


async def refresh_for_caller(caller_id: int, project_id: int) -> dict:
    """The explicit agent-facing refresh (#2802) — synchronous, named errors.

    Write-gated: recomputing spends the owner's forge API budget and rewrites
    ledger rows' seen-markers, so a read share doesn't grant it. Resolution
    runs on the OWNER's keyring like every other forge read. Raises
    ValueError with a fixable message instead of silently measuring nothing —
    the caller is an agent mid-task, and "None" would strand it exactly the
    way the button-only path did.
    """
    from scribe.models import async_session
    from scribe.models.project import Project
    from scribe.services import access

    if not await access.can_write_project(caller_id, project_id):
        raise ValueError(f"project {project_id} not found or no write access")
    async with async_session() as session:
        project = await session.get(Project, project_id)
    if project is None:
        raise ValueError(f"project {project_id} not found")
    owner_id = project.user_id or caller_id
    selector = await get_forges(owner_id, project_id)
    if not selector.configured:
        raise ValueError(
            "No forge connection serves this project — the owner adds one "
            "under Settings → Integrations → Git Forges"
        )
    coverage = await refresh_coverage(owner_id, project_id, selector=selector)
    if coverage is None:
        raise ValueError(
            "No bound repo is served by the owner's forge connections — "
            "bind_repo the project's repo on a host a connection serves"
        )
    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]
    counts = coverage.get("counts") or {}
    breakdown = " · ".join(
        f"{counts[k]} {k}"
        for k in ("canonical", "instance", "variant", "exempt", "scoped")
        if counts.get(k)
    )
    line = (
        f"shape accounting: {coverage.get('accounted', 0)}"
        f"/{coverage.get('total', 0)} shapes accounted for"
    )
    if breakdown:
        line += f" — {breakdown}"
    line += f" (estimate{', computed ' + day if day else ''})"
    unclassified = coverage.get("unclassified", 0)
    # The standing work, built whatever the todo count (#2899). Since the
    # scoped bucket (#2869) a ledger can read 100% accounted and still carry
    # derive groups, proposals and divergence; gating this block on
    # `unclassified > 0` is how 439 derive rows went unmentioned.
    standing = []
    if coverage.get("proposed"):
        standing.append(f"{coverage['proposed']} proposed")
    n_groups = len(coverage.get("derive_groups") or [])
    if n_groups:
        standing.append(f"{n_groups} derive group{'s' if n_groups != 1 else ''}")
    # Drift since the previous refresh: copies that joined a family, the
    # first one named — the sentence the arrival moment exists to say.
    new = coverage.get("derive_new") or {}
    if new.get("count"):
        n = new["count"]
        first_new = (new.get("examples") or [{}])[0]
        where = (
            f": {first_new['label']} in {first_new['path']}"
            if first_new.get("label") and first_new.get("path") else ""
        )
        standing.append(f"+{n} new cop{'y' if n == 1 else 'ies'} since last refresh{where}")
    if coverage.get("divergent"):
        standing.append(f"{coverage['divergent']} DIVERGENT")
    # The next action, on the line (#2874): the canon with the biggest
    # queue to confirm, and the widest body-identical copy to consolidate.
    top = coverage.get("top_canon") or {}
    if top.get("snippet_id"):
        standing.append(f"top canon #{top['snippet_id']} ×{top.get('count', 0)}")
    first = (coverage.get("derive_groups") or [{}])[0]
    if first.get("label") and first.get("files"):
        standing.append(f"top copy {first['label']} ×{first['files']} files")
    if unclassified:
        line += f"; {unclassified} unclassified"
        if standing:
            line += f" ({', '.join(standing)})"
        gaps = [g["dir"] for g in coverage.get("largest_gaps") or []]
        if gaps:
            line += ", largest: " + ", ".join(gaps)
    elif standing:
        line += f"; standing: {', '.join(standing)}"
    if coverage.get("recheck"):
        line += f"; {coverage['recheck']} judged shape{'s' if coverage['recheck'] != 1 else ''} changed since judged — recheck"
    return line