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>

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