Add per-article BCQuality coverage for code-review dataset (AB#646153)

src/bcbench/commands/dataset.py

@@ -182,6 +182,92 @@ def version(
         write_step_outputs({github_output: entry.environment_setup_version})
 
 
+@dataset_app.command("coverage")
+def coverage(
+    category: EvaluationCategoryOption = EvaluationCategory.CODE_REVIEW,
+    bcquality_root: Annotated[
+        str | None,
+        typer.Option(help="Path to a BCQuality checkout; enables zero-coverage detection. Falls back to $BCQUALITY_ROOT."),
+    ] = None,
+    show_zero: Annotated[bool, typer.Option(help="List every zero-coverage article (can be long)")] = False,
+    github_output: Annotated[str | None, typer.Option(help="Write the coverage report as JSON to GITHUB_OUTPUT with this key name")] = None,
+) -> None:
+    """Report per-article BCQuality coverage of the code-review dataset."""
+    from rich.console import Console
+    from rich.table import Table
+
+    from bcbench.dataset import (
+        CodeReviewEntry,
+        build_coverage_report,
+        enumerate_inventory,
+        resolve_bcquality_root,
+    )
+
+    if category is not EvaluationCategory.CODE_REVIEW:
+        raise typer.BadParameter("coverage is only supported for the code-review category")
+
+    entries: list[CodeReviewEntry] = CodeReviewEntry.load(category.dataset_path)
+
+    resolved_root = resolve_bcquality_root(bcquality_root)
+    inventory: set[str] | None = None
+    if resolved_root is not None:
+        inventory = enumerate_inventory(resolved_root)
+
+    report = build_coverage_report(entries, inventory)
+    console = Console()
+
+    if not report.inventory_available:
+        console.print("[yellow]No BCQuality checkout provided (--bcquality-root / $BCQUALITY_ROOT); reporting declared articles only, zero-coverage undetermined.[/yellow]")
+
+    domains = sorted({c.domain for c in report.covered} | {_domain_of(a) for a in report.zero_coverage})
+    covered_by_domain: dict[str, int] = {}
+    entries_by_domain: dict[str, int] = {}
+    for cov in report.covered:
+        covered_by_domain[cov.domain] = covered_by_domain.get(cov.domain, 0) + 1
+        entries_by_domain[cov.domain] = entries_by_domain.get(cov.domain, 0) + cov.count
+    zero_by_domain: dict[str, int] = {}
+    for article in report.zero_coverage:
+        domain = _domain_of(article)
+        zero_by_domain[domain] = zero_by_domain.get(domain, 0) + 1
+
+    table = Table(title="Per-article coverage (code-review)", title_justify="left", title_style="bold cyan")
+    table.add_column("Domain", style="cyan")
+    table.add_column("Covered", justify="right")
+    if report.inventory_available:
+        table.add_column("Inventory", justify="right")
+        table.add_column("Zero-cov", justify="right", style="red")
+    table.add_column("Gold entries", justify="right")
+    for domain in domains:
+        covered_count = covered_by_domain.get(domain, 0)
+        row = [domain, str(covered_count)]
+        if report.inventory_available:
+            inv = covered_count + zero_by_domain.get(domain, 0)
+            row += [str(inv), str(zero_by_domain.get(domain, 0))]
+        row.append(str(entries_by_domain.get(domain, 0)))
+        table.add_row(*row)
+    console.print(table)
+
+    summary = f"[bold]{len(report.covered)}[/bold] articles covered across [bold]{report.annotated_entries}[/bold]/{report.total_entries} annotated entries"
+    if report.inventory_available:
+        summary += f"; [red]{len(report.zero_coverage)}[/red] of {report.inventory_size} articles have zero coverage"
+    console.print(summary)
+
+    if report.unknown_articles:
+        console.print(f"[yellow]Unknown article slugs (not in inventory): {', '.join(report.unknown_articles)}[/yellow]")
+
+    if show_zero and report.zero_coverage:
+        console.print("\n[bold red]Zero-coverage articles:[/bold red]")
+        for article in report.zero_coverage:
+            console.print(f"  - {article}")
+
+    if github_output:
+        write_step_outputs({github_output: report.model_dump_json()})
+
+
+def _domain_of(article: str) -> str:
+    return article.split("/", 1)[0]
+
+
 def _modified_instance_ids_from_diff(diff_output: str) -> list[str]:
     instance_ids = []
 

src/bcbench/dataset/__init__.py

@@ -1,16 +1,30 @@
 """Dataset module for querying, validating and analyze dataset entries."""
 
 from bcbench.dataset.codereview import CodeReviewEntry, ReviewComment, Severity
+from bcbench.dataset.coverage import (
+    ArticleCoverage,
+    CoverageReport,
+    build_coverage_report,
+    collect_declared_articles,
+    enumerate_inventory,
+    resolve_bcquality_root,
+)
 from bcbench.dataset.dataset_entry import BaseDatasetEntry, BugFixEntry, NL2ALEntry, RepoGroundedEntry, TestEntry, TestGenEntry
 
 __all__ = [
+    "ArticleCoverage",
     "BaseDatasetEntry",
     "BugFixEntry",
     "CodeReviewEntry",
+    "CoverageReport",
     "NL2ALEntry",
     "RepoGroundedEntry",
     "ReviewComment",
     "Severity",
     "TestEntry",
     "TestGenEntry",
+    "build_coverage_report",
+    "collect_declared_articles",
+    "enumerate_inventory",
+    "resolve_bcquality_root",
 ]

src/bcbench/dataset/codereview.py

@@ -58,6 +58,10 @@ class ReviewComment(BaseModel):
     domain: str | None = None
     body: Annotated[str, Field(min_length=1)]
     severity: Severity | None = None
+    # BCQuality knowledge article this finding derives from, as `<domain>/<slug>`
+    # (e.g. `security/hardcoded-secret`). Optional and backward-compatible; drives
+    # per-article coverage tracking. Older entries leave it unset (counted as unannotated).
+    article: Annotated[str, Field(pattern=r"^[a-z0-9][a-z0-9-]*/[a-z0-9][a-z0-9-]*$")] | None = None
 
     @field_validator("severity", mode="before")
     @classmethod
@@ -87,3 +91,14 @@ def get_task(self) -> str:
 
     def get_expected_output(self) -> str:
         return "\n".join(str(c) for c in self.expected_comments)
+
+    def declared_articles(self) -> set[str]:
+        """BCQuality articles this entry is annotated against.
+
+        Union of every expected comment's `article` and the entry-level
+        `metadata.articles` (which carries the association for false-positive-guard
+        entries whose `expected_comments` is empty).
+        """
+        articles = {c.article for c in self.expected_comments if c.article}
+        articles.update(self.metadata.articles)
+        return articles

src/bcbench/dataset/coverage.py

@@ -0,0 +1,138 @@
+"""Per-article coverage tracking for the code-review dataset.
+
+Gold entries are annotated with the BCQuality knowledge article(s) they exercise
+(`<domain>/<slug>`). This module aggregates those annotations and, when a BCQuality
+checkout is available, compares them against the full article inventory to surface
+which articles have zero gold coverage.
+"""
+
+from __future__ import annotations
+
+import os
+from collections.abc import Iterable, Sequence
+from pathlib import Path
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from bcbench.dataset.codereview import CodeReviewEntry
+
+# Relative to a BCQuality checkout root; knowledge articles live here as
+# `<domain>/<slug>.md`, so the article id is the relative path minus the suffix.
+_KNOWLEDGE_SUBDIR = Path("microsoft") / "knowledge"
+_BCQUALITY_ROOT_ENV = "BCQUALITY_ROOT"
+
+
+class ArticleCoverage(BaseModel):
+    """One article and the gold entries that exercise it."""
+
+    model_config = ConfigDict(frozen=True)
+
+    article: str
+    domain: str
+    entry_ids: list[str] = Field(default_factory=list)
+
+    @property
+    def count(self) -> int:
+        return len(self.entry_ids)
+
+
+class CoverageReport(BaseModel):
+    """Per-article coverage of the code-review dataset."""
+
+    model_config = ConfigDict(frozen=True)
+
+    covered: list[ArticleCoverage] = Field(default_factory=list)
+    zero_coverage: list[str] = Field(default_factory=list)
+    unknown_articles: list[str] = Field(default_factory=list)
+    unannotated_entry_ids: list[str] = Field(default_factory=list)
+    total_entries: int = 0
+    inventory_available: bool = False
+
+    @property
+    def inventory_size(self) -> int:
+        if not self.inventory_available:
+            return 0
+        return len(self.covered) + len(self.zero_coverage)
+
+    @property
+    def annotated_entries(self) -> int:
+        return self.total_entries - len(self.unannotated_entry_ids)
+
+
+def _domain_of(article: str) -> str:
+    return article.split("/", 1)[0]
+
+
+def collect_declared_articles(entries: Sequence[CodeReviewEntry]) -> dict[str, list[str]]:
+    """Map each declared article to the sorted, de-duplicated entry ids that declare it."""
+    article_to_entries: dict[str, set[str]] = {}
+    for entry in entries:
+        for article in entry.declared_articles():
+            article_to_entries.setdefault(article, set()).add(entry.instance_id)
+    return {article: sorted(ids) for article, ids in article_to_entries.items()}
+
+
+def enumerate_inventory(bcquality_root: Path) -> set[str]:
+    """Enumerate `<domain>/<slug>` article ids from a BCQuality checkout.
+
+    Only `.md` files under `microsoft/knowledge/` count; sibling `.good.al` / `.bad.al`
+    sample files and the generated `knowledge-index.json` are ignored.
+    """
+    knowledge_dir = bcquality_root / _KNOWLEDGE_SUBDIR
+    if not knowledge_dir.is_dir():
+        raise FileNotFoundError(f"BCQuality knowledge directory not found: {knowledge_dir}")
+
+    inventory: set[str] = set()
+    for path in knowledge_dir.rglob("*.md"):
+        relative = path.relative_to(knowledge_dir).with_suffix("")
+        inventory.add(relative.as_posix())
+    return inventory
+
+
+def resolve_bcquality_root(explicit: Path | str | None = None) -> Path | None:
+    """Resolve a BCQuality checkout root from an explicit value or `BCQUALITY_ROOT`.
+
+    Returns None when neither is set, so callers can degrade to declared-only coverage.
+    """
+    candidate = explicit if explicit is not None else os.environ.get(_BCQUALITY_ROOT_ENV)
+    if not candidate:
+        return None
+    return Path(candidate).expanduser()
+
+
+def build_coverage_report(
+    entries: Sequence[CodeReviewEntry],
+    inventory: Iterable[str] | None = None,
+) -> CoverageReport:
+    """Compute per-article coverage.
+
+    When `inventory` is provided, articles in the inventory with no declaring entry are
+    reported as `zero_coverage`, and declared articles absent from the inventory (typos /
+    stale slugs) are reported as `unknown_articles`. Without an inventory, only declared
+    articles are reported (`covered`), and zero-coverage cannot be determined.
+    """
+    declared = collect_declared_articles(entries)
+    inventory_set = set(inventory) if inventory is not None else None
+
+    covered: list[ArticleCoverage] = []
+    unknown: list[str] = []
+    for article in sorted(declared):
+        if inventory_set is not None and article not in inventory_set:
+            unknown.append(article)
+            continue
+        covered.append(ArticleCoverage(article=article, domain=_domain_of(article), entry_ids=declared[article]))
+
+    zero_coverage: list[str] = []
+    if inventory_set is not None:
+        zero_coverage = sorted(inventory_set - set(declared))
+
+    unannotated = sorted(e.instance_id for e in entries if not e.declared_articles())
+
+    return CoverageReport(
+        covered=covered,
+        zero_coverage=zero_coverage,
+        unknown_articles=sorted(unknown),
+        unannotated_entry_ids=unannotated,
+        total_entries=len(entries),
+        inventory_available=inventory_set is not None,
+    )

src/bcbench/dataset/dataset_entry.py

@@ -30,6 +30,10 @@ class EntryMetadata(BaseModel):
     area: str | None = None
     image_count: Annotated[int, Field(ge=0)] | None = None
     persona: str | None = None
+    # BCQuality knowledge articles this entry exercises as `<domain>/<slug>`. Primarily
+    # for false-positive-guard entries (expected_comments=[]) that test an article by
+    # omission and thus have no per-comment `article` to carry the association.
+    articles: list[Annotated[str, Field(pattern=r"^[a-z0-9][a-z0-9-]*/[a-z0-9][a-z0-9-]*$")]] = Field(default_factory=list)
 
 
 class BaseDatasetEntry(BaseModel):