feat(coverage): pattern-library coverage measurement (#2692, milestone 288 step 7)

Server-side shape enumeration per bound repo — one archive download via the
forge adapter, definitions extracted with a Python mirror of the write-path
hook's awk rules (shared test vectors pin the two together) — compared
against recorded snippet locations by path+symbol. Summary is cached in the
settings KV with a freshness stamp; recomputed on webhook push (spawned off
the delivery path) or explicit refresh, never in a request path.

Surfaces: GET/POST /api/projects/<id>/coverage[/refresh], a project-page
card (estimate-labeled, largest-gaps chips), and a one-line evidence-carrying
entry in enter_project read from cache only.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

frontend/src/views/ProjectView.vue

@@ -418,11 +418,66 @@ async function loadNotes() {
   }
 }
 
+/* ── Pattern-library coverage (#2692) ─────────────────────────── */
+
+interface CoverageGap {
+  dir: string;
+  uncovered: number;
+  total: number;
+}
+interface Coverage {
+  total: number;
+  recorded: number;
+  estimate: boolean;
+  computed_at: string;
+  repos: { repo: string; ref: string; total: number; recorded: number }[];
+  largest_gaps: CoverageGap[];
+}
+
+const coverage = ref<Coverage | null>(null);
+const coverageConfigured = ref(false);
+const coverageRefreshing = ref(false);
+const coverageError = ref<string | null>(null);
+
+/** Swallows failure like loadDesignSystems: no forge is the ordinary state
+ *  for most installs, and this card must never break the project page. */
+async function loadCoverage() {
+  try {
+    const res = await apiGet<{ configured: boolean; coverage: Coverage | null }>(
+      `/api/projects/${projectId.value}/coverage`
+    );
+    coverageConfigured.value = res.configured;
+    coverage.value = res.coverage;
+  } catch {
+    coverageConfigured.value = false;
+    coverage.value = null;
+  }
+}
+
+async function refreshCoverage() {
+  if (coverageRefreshing.value) return;
+  coverageRefreshing.value = true;
+  coverageError.value = null;
+  try {
+    const res = await apiPost<{ coverage: Coverage }>(
+      `/api/projects/${projectId.value}/coverage/refresh`,
+      {}
+    );
+    coverage.value = res.coverage;
+  } catch (e) {
+    coverageError.value =
+      e instanceof Error ? e.message : "Coverage refresh failed";
+  } finally {
+    coverageRefreshing.value = false;
+  }
+}
+
 onMounted(async () => {
   await loadProject();
   loadTasks();
   loadNotes();
   loadDesignSystems();
+  loadCoverage();
 });
 
 /** Populate the design-system picker. Swallows failure on purpose: with no
@@ -440,6 +495,7 @@ watch(projectId, async () => {
   await loadProject();
   loadTasks();
   loadNotes();
+  loadCoverage();
 });
 
 watch(
@@ -592,6 +648,53 @@ async function confirmDelete() {
         </div>
       </div>
 
+      <!-- Pattern-library coverage (#2692) — only when a forge is configured;
+           forge-less installs never see this card at all. -->
+      <div v-if="coverageConfigured" class="coverage-card">
+        <div class="coverage-head">
+          <span class="coverage-title">Pattern coverage</span>
+          <span class="coverage-estimate" title="Keyword-based extraction over- and under-counts; watch the trend, not the digit">estimate</span>
+          <span v-if="coverage" class="coverage-when">computed {{ relativeTime(coverage.computed_at) }}</span>
+          <button
+            class="btn-ghost btn-compact coverage-refresh"
+            :disabled="coverageRefreshing"
+            @click="refreshCoverage"
+          >
+            {{ coverageRefreshing ? "Measuring…" : "Refresh" }}
+          </button>
+        </div>
+        <template v-if="coverage">
+          <div class="coverage-numbers">
+            <span class="coverage-count">{{ coverage.recorded }}/{{ coverage.total }}</span>
+            <span class="coverage-label">shapes recorded</span>
+          </div>
+          <div
+            class="coverage-bar"
+            role="progressbar"
+            :aria-valuenow="coverage.recorded"
+            :aria-valuemin="0"
+            :aria-valuemax="coverage.total"
+            aria-label="Shapes with a recorded snippet"
+          >
+            <div
+              class="coverage-bar-fill"
+              :style="{ width: (coverage.total ? (coverage.recorded / coverage.total) * 100 : 0) + '%' }"
+            ></div>
+          </div>
+          <div v-if="coverage.largest_gaps?.length" class="coverage-gaps">
+            <span class="coverage-gaps-label">Largest gaps:</span>
+            <span v-for="gap in coverage.largest_gaps" :key="gap.dir" class="coverage-gap-chip">
+              {{ gap.dir }} <span class="coverage-gap-count">{{ gap.uncovered }}</span>
+            </span>
+          </div>
+        </template>
+        <p v-else class="coverage-empty">
+          Not measured yet — Refresh compares the bound repo's definitions
+          against recorded snippets.
+        </p>
+        <p v-if="coverageError" class="coverage-error">{{ coverageError }}</p>
+      </div>
+
       <div class="project-body">
 
         <!-- Edit panel -->
@@ -1037,6 +1140,89 @@ async function confirmDelete() {
 .stat-done      { background: color-mix(in srgb, var(--fs-success) 10%, transparent); color: var(--fs-success); border-color: color-mix(in srgb, var(--fs-success) 28%, transparent); }
 .stat-notes     { background: color-mix(in srgb, var(--fs-accent) 8%, transparent); color: var(--fs-accent); border-color: color-mix(in srgb, var(--fs-accent) 22%, transparent); }
 
+/* ── Pattern-library coverage card ───────────────────────────── */
+.coverage-card {
+  background: var(--fs-surface-raised);
+  border: 1px solid var(--fs-border-color);
+  border-radius: var(--fs-radius-lg);
+  padding: 0.75rem 1rem;
+  margin-bottom: 1.25rem;
+  display: flex;
+  flex-direction: column;
+  gap: 0.5rem;
+}
+.coverage-head {
+  display: flex;
+  align-items: center;
+  gap: 0.5rem;
+}
+.coverage-title {
+  font-size: 0.72rem;
+  font-weight: 500;
+  text-transform: uppercase;
+  letter-spacing: 0.06em;
+  color: var(--fs-text-tertiary);
+}
+.coverage-estimate {
+  font-size: 0.68rem;
+  padding: 0.05rem 0.4rem;
+  border-radius: var(--fs-radius-lg);
+  border: 1px solid var(--fs-border-color);
+  color: var(--fs-text-tertiary);
+  cursor: help;
+}
+.coverage-when {
+  font-size: 0.72rem;
+  color: var(--fs-text-tertiary);
+}
+.coverage-refresh { margin-left: auto; }
+.coverage-numbers {
+  display: flex;
+  align-items: baseline;
+  gap: 0.4rem;
+}
+.coverage-count { font-size: 1.15rem; font-weight: 500; }
+.coverage-label { font-size: 0.82rem; color: var(--fs-text-secondary); }
+.coverage-bar {
+  height: 6px;
+  border-radius: 3px;
+  background: color-mix(in srgb, var(--fs-text-tertiary) 14%, transparent);
+  overflow: hidden;
+}
+.coverage-bar-fill {
+  height: 100%;
+  border-radius: 3px;
+  background: var(--fs-accent);
+  transition: width 0.3s ease;
+}
+.coverage-gaps {
+  display: flex;
+  align-items: center;
+  gap: 0.4rem;
+  flex-wrap: wrap;
+  font-size: 0.78rem;
+}
+.coverage-gaps-label { color: var(--fs-text-tertiary); }
+.coverage-gap-chip {
+  padding: 0.1rem 0.5rem;
+  border-radius: var(--fs-radius-lg);
+  border: 1px solid var(--fs-border-color);
+  color: var(--fs-text-secondary);
+  font-family: var(--fs-font-mono);
+  font-size: 0.74rem;
+}
+.coverage-gap-count { opacity: 0.65; }
+.coverage-empty {
+  margin: 0;
+  font-size: 0.8rem;
+  color: var(--fs-text-tertiary);
+}
+.coverage-error {
+  margin: 0;
+  font-size: 0.8rem;
+  color: var(--fs-error);
+}
+
 /* ── Two-column body ─────────────────────────────────────────── */
 /* `minmax(0, 1fr)`, not `1fr`. A bare `1fr` track has an AUTO minimum, so it
    cannot shrink below its content — one wide descendant anywhere in the

src/scribe/mcp/tools/projects.py

@@ -17,6 +17,7 @@ keeps working.
 from __future__ import annotations
 
 from scribe.mcp._context import current_user_id
+from scribe.services import coverage as coverage_svc
 from scribe.services import design_systems as design_systems_svc
 from scribe.services import milestones as milestones_svc
 from scribe.services import notes as notes_svc
@@ -56,7 +57,13 @@ async def enter_project(project_id: int) -> dict:
 
     Returns a dict with keys: project, milestone_summary, applicable_rules,
     project_rules, subscribed_rulebooks, applicable_rules_truncated,
-    open_tasks, recent_notes, design_system, systems.
+    open_tasks, recent_notes, design_system, systems, pattern_coverage.
+
+    `pattern_coverage` (usually null) is a one-line estimate of how much of
+    the bound repo's code has recorded snippets — e.g. "pattern-library
+    coverage: 34/210 shapes recorded (estimate); largest gaps: internal/api".
+    When present, treat the gaps as a standing invitation: as you touch code
+    in those areas, record the shapes you find with create_snippet.
 
     `systems` is the project's vocabulary of named subsystems/areas. It is
     returned here so you can TAG as you write: when creating or meaningfully
@@ -128,8 +135,17 @@ async def enter_project(project_id: int) -> dict:
             uid, project.design_system_id,
         )
 
+    # Cache read ONLY — computing coverage moves a repo tarball and never
+    # belongs in this request path. Null is the ordinary state (no forge, or
+    # never computed); the line appears exactly when there is evidence. Read
+    # on the OWNER's id: bindings and the cache live with the project owner.
+    coverage = await coverage_svc.cached_coverage(
+        project.user_id or uid, project_id
+    )
+
     return {
         "project": project.to_dict(),
+        "pattern_coverage": coverage_svc.coverage_line(coverage) if coverage else None,
         # Trimmed to what tagging needs. The full charter is get_system's job —
         # this list rides along on every session start, so it stays lean.
         "systems": [

src/scribe/routes/projects.py

@@ -120,6 +120,61 @@ async def delete_project_route(project_id: int):
     return "", 204
 
 
+@projects_bp.route("/<int:project_id>/coverage", methods=["GET"])
+@login_required
+async def get_coverage_route(project_id: int):
+    """The cached pattern-library coverage summary — never computes.
+
+    `configured` tells the card whether offering a Refresh button makes
+    sense; `coverage` is null until something has computed it (a webhook
+    push or an explicit refresh).
+    """
+    from scribe.services.coverage import cached_coverage
+    from scribe.services.forge import get_forge
+
+    uid = get_current_user_id()
+    result = await get_project_for_user(uid, project_id)
+    if result is None:
+        return not_found("Project")
+    project, _ = result
+    owner_uid = project.user_id or uid
+    return jsonify({
+        "configured": await get_forge() is not None,
+        "coverage": await cached_coverage(owner_uid, project_id),
+    })
+
+
+@projects_bp.route("/<int:project_id>/coverage/refresh", methods=["POST"])
+@login_required
+async def refresh_coverage_route(project_id: int):
+    """Recompute coverage now (archive fetch — seconds, not milliseconds).
+
+    Synchronous on purpose: the caller is a person who just clicked Refresh
+    and wants the new number, and the forge timeout bounds the wait.
+    """
+    from scribe.services.coverage import refresh_coverage
+    from scribe.services.forge import ForgeError, get_forge
+
+    uid = get_current_user_id()
+    result = await get_project_for_user(uid, project_id)
+    if result is None:
+        return not_found("Project")
+    project, _ = result
+    owner_uid = project.user_id or uid
+    if await get_forge() is None:
+        return jsonify({"error": "No git forge is configured (Settings → Config → Git Forge)"}), 400
+    try:
+        coverage = await refresh_coverage(owner_uid, project_id)
+    except ForgeError as exc:
+        return jsonify({"error": str(exc)}), 502
+    if coverage is None:
+        return jsonify({
+            "error": "No bound repo is served by the configured forge — "
+            "bind the project's repo (bind_repo) on a remote the forge hosts"
+        }), 400
+    return jsonify({"coverage": coverage})
+
+
 @projects_bp.route("/<int:project_id>/notes", methods=["GET"])
 @login_required
 async def get_project_notes_route(project_id: int):

src/scribe/routes/webhooks.py

@@ -29,7 +29,9 @@ import traceback
 from quart import Blueprint, jsonify, request
 
 from scribe.config import Config
-from scribe.services.repo_bindings import normalize_repo_key
+from scribe.services.background import spawn
+from scribe.services.coverage import refresh_coverage
+from scribe.services.repo_bindings import bindings_for_key, normalize_repo_key
 from scribe.services.settings import get_admin_setting
 from scribe.services.snippets import invalidate_for_push
 
@@ -88,6 +90,16 @@ async def forge_push():
             logger.info(
                 "forge push %s flagged %d snippet(s) for recheck", head[:12], flagged
             )
+        # A push is exactly when the coverage number goes stale — recompute it
+        # off the delivery path (#2692). Fire-and-forget: the forge's delivery
+        # loop must not wait on an archive download, and a failure is a
+        # WARNING from spawn(), never a failed delivery. This also SEEDS the
+        # cache on a webhook-configured instance — no manual first refresh.
+        for binding in await bindings_for_key(repo_key):
+            spawn(
+                refresh_coverage(binding.user_id, binding.project_id),
+                site="webhooks.coverage_refresh",
+            )
         return jsonify({"ok": True, "flagged": flagged})
     except Exception:
         logger.warning("forge webhook processing failed", exc_info=True)

src/scribe/services/coverage.py

@@ -0,0 +1,336 @@
+"""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 GiteaForge, get_forge
+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, *, forge: GiteaForge | None = None
+) -> dict | None:
+    """Measure a project's pattern-library coverage against its bound repos.
+
+    None means "nothing to measure" — no forge configured, or none of the
+    project's bound repos is served by it. That is the ordinary state for a
+    forge-less install 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.
+    """
+    forge = forge if forge is not None else await get_forge()
+    if forge is None:
+        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):
+        api_repo = forge.resolve_repo(key)
+        if api_repo is None:
+            continue  # bound to a host this forge doesn't serve
+        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, *, forge: GiteaForge | None = None
+) -> dict | None:
+    """Compute and cache. The only writer of the cache key."""
+    coverage = await compute_coverage(user_id, project_id, forge=forge)
+    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

src/scribe/services/forge.py

@@ -58,6 +58,12 @@ FORGE_KINDS = ("gitea",)
 # hung socket, and there is no retry: the fallback IS the retry policy.
 _TIMEOUT = httpx.Timeout(5.0)
 
+# Archive downloads move a whole-repo tarball and only ever run off the
+# request path (coverage recompute, step 7), so they get a bigger budget than
+# the per-file reads — but still a bound, because a hung background task
+# holds a connection slot as surely as a foreground one.
+_ARCHIVE_TIMEOUT = httpx.Timeout(60.0)
+
 
 class ForgeError(RuntimeError):
     """A forge call failed (network, auth, unexpected payload). Token-free."""
@@ -179,6 +185,22 @@ class GiteaForge:
             path=payload.get("path") or path,
         )
 
+    async def archive(self, repo: str, ref: str) -> bytes:
+        """The repo's content at ``ref`` as a gzipped tarball, in one request.
+
+        Coverage measurement (step 7) needs every source file's text; per-file
+        reads would mean one API call per file, so the archive endpoint is the
+        only shape that scales past toy repos. Callers must never run this in
+        a request path — it moves the whole repo.
+        """
+        async with self._client() as client:
+            resp = await self._get(
+                client,
+                f"/repos/{repo}/archive/{quote(ref, safe='')}.tar.gz",
+                timeout=_ARCHIVE_TIMEOUT,
+            )
+        return resp.content
+
     async def default_branch(self, repo: str) -> str:
         async with self._client() as client:
             resp = await self._get(client, f"/repos/{repo}")

tests/test_mcp_tool_projects.py

@@ -29,6 +29,17 @@ def _no_systems():
         yield
 
 
+@pytest.fixture(autouse=True)
+def _no_coverage():
+    """enter_project also reads the pattern-coverage cache (#2692) — same
+    deal: no database here, stub the common case (nothing computed). The
+    populated line is asserted in tests/test_pattern_coverage.py.
+    """
+    with patch("scribe.mcp.tools.projects.coverage_svc.cached_coverage",
+               AsyncMock(return_value=None)):
+        yield
+
+
 def _fake_project(design_system_id=None, **overrides) -> MagicMock:
     p = MagicMock()
     base = {"id": 1, "title": "P", "description": "", "goal": "",

tests/test_pattern_coverage.py

@@ -0,0 +1,317 @@
+"""Pattern-library coverage (#2692) — the extractor that mirrors the hook,
+the shape/record matcher, and the end-to-end measurement on real Postgres.
+
+The extractor here and the hook's awk program (scribe_prior_art.sh ARM 1)
+must agree on what counts as "a definition" — the metric and the write-path
+backstop are two views of the same doctrine. The EXTRACTION_VECTORS below
+deliberately reuse the definitions test_write_path_trigger stages for the
+hook; extending one detector means extending both, and this comment is the
+tripwire.
+"""
+import io
+import json
+import tarfile
+
+import pytest
+import pytest_asyncio
+
+from scribe.services.coverage import (
+    coverage_line,
+    extract_shapes,
+    largest_gaps,
+    match_shapes,
+    scannable,
+    shapes_from_archive,
+)
+
+# --- unit: the definition extractor (shared vectors with the hook) -----------
+
+EXTRACTION_VECTORS = [
+    # (id, source text, expected (kind, name) list)
+    ("python", "def make_app():\n    pass\nclass Config:\n    pass\n",
+     [("sym", "make_app"), ("sym", "Config")]),
+    ("python-dunder-skip", "class C:\n    def __init__(self):\n        pass\n",
+     [("sym", "C")]),
+    ("go-func", "func Resolve(x int) error {\n\treturn nil\n}\n",
+     [("sym", "Resolve")]),
+    ("go-method", "func (s *Scanner) Resolve(x int) error {\n\treturn nil\n}\n",
+     [("sym", "Resolve")]),
+    ("kotlin-fun", "suspend fun refreshQueue(id: Long) {\n}\n",
+     [("sym", "refreshQueue")]),
+    ("rust-fn", "pub async fn fetch_all() -> u32 {\n    0\n}\n",
+     [("sym", "fetch_all")]),
+    ("go-type", "type ForgeAdapter struct {\n\tname string\n}\n",
+     [("sym", "ForgeAdapter")]),
+    ("rust-pub-crate", "pub(crate) struct Widget {}\n",
+     [("sym", "Widget")]),
+    ("js-export-default", "export default function App() {}\n",
+     [("sym", "App")]),
+    ("js-arrow", "const useThing = (id) => id;\nlet fetcher = async () => 0;\n",
+     [("sym", "useThing"), ("sym", "fetcher")]),
+    # Every selector line in a group counts — .btn-ghost, and .btn-text { }
+    # both announce a class, exactly as the hook's awk sees them.
+    ("css", ".btn-primary {\n  color: red;\n}\n.btn-ghost,\n.btn-text { }\n",
+     [("css", "btn-primary"), ("css", "btn-ghost"), ("css", "btn-text")]),
+    # Call sites, imports, and impl blocks are NOT definitions — matching
+    # them would drown the metric exactly as it would drown the hook.
+    ("non-definitions",
+     "make_app()\nimpl Widget {\nreturn fetch_all\nimport os\nx = 1\n",
+     []),
+    ("dedup-within-file", "def f():\n    pass\ndef f():\n    pass\n",
+     [("sym", "f")]),
+]
+
+
+@pytest.mark.parametrize(
+    ("text", "expected"),
+    [(t, e) for _i, t, e in EXTRACTION_VECTORS],
+    ids=[i for i, _t, _e in EXTRACTION_VECTORS],
+)
+def test_extractor_agrees_with_the_hook_on_what_defines(text, expected):
+    assert extract_shapes(text) == expected
+
+
+def test_scannable_gates_prose_vendored_and_sourcemaps():
+    assert scannable("src/app.py")
+    assert scannable("web/button.css")
+    assert scannable(".gitea/workflows/ci.yml")  # config IS worth recording
+    assert not scannable("README.md")
+    assert not scannable("dist/bundle.js.map")
+    assert not scannable("node_modules/x/index.js")
+    assert not scannable("web/node_modules/y/util.ts")
+    # A FILE named like a skip-dir is not a directory hit.
+    assert scannable("src/vendor.py")
+
+
+# --- unit: reading shapes out of a forge tarball -----------------------------
+
+
+def _tarball(files: dict[str, bytes], top: str = "widget") -> bytes:
+    buf = io.BytesIO()
+    with tarfile.open(fileobj=buf, mode="w:gz") as tar:
+        for path, data in files.items():
+            info = tarfile.TarInfo(f"{top}/{path}")
+            info.size = len(data)
+            tar.addfile(info, io.BytesIO(data))
+    return buf.getvalue()
+
+
+TREE = {
+    "src/app.py": b"def make_app():\n    pass\nclass Config:\n    def __init__(self):\n        pass\n",
+    "src/util.py": b"def helper():\n    pass\n",
+    "web/button.css": b".btn {\n  color: red;\n}\n",
+    "README.md": b"def not_code(): pass\n",
+    "node_modules/x/index.js": b"function vendored() {}\n",
+    "data.bin": b"\xff\xfe\x00\x01",
+}
+# What TREE holds once the gates run: 4 shapes, none from the skipped files.
+TREE_SHAPES = [
+    ("src/app.py", "sym", "make_app"),
+    ("src/app.py", "sym", "Config"),
+    ("src/util.py", "sym", "helper"),
+    ("web/button.css", "css", "btn"),
+]
+
+
+def test_shapes_from_archive_strips_the_wrapper_and_gates_files():
+    assert shapes_from_archive(_tarball(TREE)) == TREE_SHAPES
+
+
+# --- unit: matching shapes against recorded locations ------------------------
+
+
+def test_match_covers_by_exact_path_dir_prefix_and_css_dot():
+    recorded = [
+        ("src/app.py", "make_app"),   # exact file
+        ("web", ".btn"),              # dir prefix + css dot normalization
+    ]
+    matched = match_shapes(TREE_SHAPES, recorded)
+    covered = {name for _p, _k, name, ok in matched if ok}
+    assert covered == {"make_app", "btn"}
+
+
+def test_a_symbol_less_record_covers_nothing():
+    """A whole-file snippet makes no claim about any particular definition
+    inside it — crediting all of them would inflate the number for free."""
+    matched = match_shapes(TREE_SHAPES, [("src/app.py", "")])
+    assert not any(ok for *_x, ok in matched)
+
+
+def test_no_prefix_bleed_between_sibling_directories():
+    matched = match_shapes(
+        [("src/library/x.py", "sym", "helper")], [("src/lib", "helper")]
+    )
+    assert not matched[0][3]
+
+
+def test_largest_gaps_ranks_by_uncovered_and_drops_clean_dirs():
+    matched = match_shapes(TREE_SHAPES, [("src/app.py", "make_app"), ("web", ".btn")])
+    gaps = largest_gaps(matched)
+    assert gaps == [{"dir": "src", "uncovered": 2, "total": 3}]
+
+
+def test_coverage_line_is_evidence_carrying_and_labeled_estimate():
+    line = coverage_line({
+        "total": 210, "recorded": 34, "estimate": True,
+        "computed_at": "2026-08-16T12:00:00+00:00",
+        "largest_gaps": [
+            {"dir": "internal/api", "uncovered": 40, "total": 60},
+            {"dir": "web/src/components", "uncovered": 25, "total": 30},
+        ],
+    })
+    assert "34/210 shapes recorded" in line
+    assert "estimate" in line
+    assert "2026-08-16" in line
+    assert "internal/api, web/src/components" in line
+
+
+def test_coverage_routes_are_registered():
+    from scribe.app import create_app
+
+    rules = {r.rule for r in create_app().url_map.iter_rules()}
+    assert "/api/projects/<int:project_id>/coverage" in rules
+    assert "/api/projects/<int:project_id>/coverage/refresh" in rules
+
+
+# --- integration: the measurement end to end on real Postgres ----------------
+
+
+def _forge(tar_bytes: bytes):
+    import httpx
+
+    from scribe.services.forge import GiteaForge
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        path = request.url.path
+        if path == "/api/v1/repos/alice/widget":
+            return httpx.Response(200, json={"default_branch": "main"})
+        if path == "/api/v1/repos/alice/widget/archive/main.tar.gz":
+            return httpx.Response(200, content=tar_bytes)
+        return httpx.Response(404, json={"message": "not found"})
+
+    return GiteaForge(
+        "https://git.example.com", "tok", transport=httpx.MockTransport(handler)
+    )
+
+
+@pytest_asyncio.fixture
+async def _dispose_engine():
+    from scribe.models import engine
+    yield
+    await engine.dispose()
+
+
+@pytest_asyncio.fixture
+async def seeded(_dispose_engine):
+    """User + project + binding + two snippets that cover 2 of TREE's 4 shapes."""
+    from sqlalchemy import select
+
+    from scribe.models import async_session
+    from scribe.models.project import Project
+    from scribe.models.user import User
+    from scribe.services import snippets as svc
+    from scribe.services.repo_bindings import set_binding
+
+    async with async_session() as s:
+        user = (
+            await s.execute(select(User).where(User.username == "coverage_itest"))
+        ).scalar_one_or_none()
+        if user is None:
+            user = User(username="coverage_itest")
+            s.add(user)
+            await s.flush()
+        project = Project(user_id=user.id, title="Widget")
+        s.add(project)
+        await s.flush()
+        uid, pid = user.id, project.id
+        await s.commit()
+
+    await set_binding(uid, "https://git.example.com/alice/widget.git", pid)
+
+    await svc.create_snippet(
+        uid, name="cov_make_app", code="def make_app():\n    pass\n",
+        language="python", repo="Widget", path="src/app.py",
+        symbol="make_app", project_id=pid,
+    )
+    await svc.create_snippet(
+        uid, name="cov_btn", code=".btn {\n  color: red;\n}\n",
+        language="css", repo="Widget", path="web", symbol=".btn",
+        project_id=pid,
+    )
+    return {"uid": uid, "pid": pid}
+
+
+@pytest.mark.integration
+async def test_coverage_measures_the_tree_exactly_and_caches(seeded):
+    from scribe.services.coverage import (
+        cached_coverage,
+        compute_coverage,
+        refresh_coverage,
+    )
+
+    uid, pid = seeded["uid"], seeded["pid"]
+    forge = _forge(_tarball(TREE))
+
+    coverage = await compute_coverage(uid, pid, forge=forge)
+    assert coverage is not None
+    assert coverage["total"] == 4
+    assert coverage["recorded"] == 2
+    assert coverage["estimate"] is True
+    assert coverage["repos"] == [{
+        "repo": "git.example.com/alice/widget", "ref": "main",
+        "total": 4, "recorded": 2,
+    }]
+    assert coverage["largest_gaps"] == [{"dir": "src", "uncovered": 2, "total": 3}]
+
+    # Nothing computed → nothing cached; refresh writes; the cache reads back
+    # byte-equal, because enter_project will serve exactly this.
+    assert await cached_coverage(uid, pid) is None
+    stored = await refresh_coverage(uid, pid, forge=forge)
+    assert (await cached_coverage(uid, pid)) == json.loads(json.dumps(stored))
+
+
+@pytest.mark.integration
+async def test_enter_project_surfaces_the_line_only_once_computed(seeded):
+    from scribe.mcp._context import _user_id_ctx
+    from scribe.mcp.tools.projects import enter_project
+    from scribe.services.coverage import refresh_coverage
+
+    uid, pid = seeded["uid"], seeded["pid"]
+    token = _user_id_ctx.set(uid)
+    try:
+        # Forge-less / never-computed instance: the key is present, null, and
+        # nothing else about the response changes.
+        before = await enter_project(project_id=pid)
+        assert before["pattern_coverage"] is None
+
+        await refresh_coverage(uid, pid, forge=_forge(_tarball(TREE)))
+        after = await enter_project(project_id=pid)
+        line = after["pattern_coverage"]
+        assert line.startswith(
+            "pattern-library coverage: 2/4 shapes recorded (estimate, computed "
+        )
+        assert line.endswith("; largest gaps: src")
+    finally:
+        _user_id_ctx.reset(token)
+
+
+@pytest.mark.integration
+async def test_unservable_binding_measures_nothing(seeded):
+    """A project bound only to a host the forge doesn't serve returns None —
+    the same silence as no forge at all, never an error."""
+    from scribe.services.coverage import compute_coverage
+    from scribe.services.repo_bindings import set_binding
+
+    from scribe.models import async_session
+    from scribe.models.project import Project
+
+    uid = seeded["uid"]
+    async with async_session() as s:
+        other = Project(user_id=uid, title="Elsewhere")
+        s.add(other)
+        await s.flush()
+        other_pid = other.id
+        await s.commit()
+    await set_binding(uid, "https://github.com/somebody/else.git", other_pid)
+
+    assert await compute_coverage(uid, other_pid, forge=_forge(_tarball(TREE))) is None