GEI-13: Conflict report as the ship artifact

src/semianalyst/analyze/__init__.py

@@ -1,31 +1,38 @@
-"""analyze — cross-source corroboration and divergence detection (FUTURE WORK).
+"""analyze — cross-source corroboration and the per-CVE conflict ship artifact.
 
-This module is intentionally empty this phase. It will read claims from the store
-(never touching SQLite directly — it goes through store.db) and answer the
-questions that make this an analyst tool rather than a scraper:
-
-  Corroboration
-    Do independent sources agree on the same metric for the same entity? Group
-    claims by (entity, normalized metric), compare values within tolerance, and
-    weight agreement by source_tier (conference > foundry > vendor). Populate
-    claim.corroboration.status and related_claim_ids.
-
-  Divergence
-    Where do sources contradict each other, and does the disagreement track with
-    tier or with disclosed conditions? A vendor's "2.5x" that only holds under
-    sparsity, against a foundry's iso-condition number, is a divergence the tool
-    must surface rather than average away. Relative claims are compared as
-    ratios against their stated baselines — never as reconstructed absolutes.
-
-Design constraint: analysis is read-only over the store and must preserve the
-schema's normalization rules (no relative->absolute conversion).
+Library entry points (CLI is a thin wrapper):
+  run_analysis / analyze_claims — derive-on-read corroboration over the store
+  conflict_report / build_conflict_report — GEI-13 per-CVE conflict report
 """
 
+from .conflict_report import (
+    QUOTE_SPAN_MAX,
+    ClaimSide,
+    ConflictGroup,
+    ConflictReport,
+    CveConflictReport,
+    bound_quote_span,
+    build_conflict_report,
+    conflict_report,
+)
 from .corroborate import (
     Assessment,
     AnalysisReport,
     analyze_claims,
     run_analysis,
 )
 
-__all__ = ["Assessment", "AnalysisReport", "analyze_claims", "run_analysis"]
+__all__ = [
+    "Assessment",
+    "AnalysisReport",
+    "analyze_claims",
+    "run_analysis",
+    "QUOTE_SPAN_MAX",
+    "ClaimSide",
+    "ConflictGroup",
+    "ConflictReport",
+    "CveConflictReport",
+    "bound_quote_span",
+    "build_conflict_report",
+    "conflict_report",
+]

src/semianalyst/analyze/conflict_report.py

@@ -0,0 +1,210 @@
+"""GEI-13 — per-CVE conflict report (the ship artifact).
+
+Conflict is the product: group advisory assessments by CVE and surface
+contradicted (and related weakly_corroborated) groups for affected_range /
+patched_in / exploit_status, with each side's bounded quote_span and
+source_record_kind/doc. Never picks a winning range or kind. Never
+auto-resolves. Does not interpolate workaround_text.
+"""
+
+from __future__ import annotations
+
+from collections import defaultdict
+from dataclasses import dataclass, field
+
+from ..config import Config, load_config
+from ..store import ClaimView, connect, get_claims_for_analysis
+from .corroborate import AnalysisReport, Assessment, run_analysis
+
+# Schema bounds.cite_quote_span.max / Span2000 — flood guard for display.
+QUOTE_SPAN_MAX = 2000
+
+_FOCUS_CLASSES = frozenset({"affected_range", "patched_in", "exploit_status"})
+# Conflict is the product; related context (weakly_corroborated) rides along.
+_REPORT_STATUSES = frozenset({"contradicted", "weakly_corroborated"})
+
+
+def bound_quote_span(text: str | None, *, max_len: int = QUOTE_SPAN_MAX) -> str:
+    """Truncate to schema cite_quote_span max. Empty/None → \"\"."""
+    if not text:
+        return ""
+    if len(text) <= max_len:
+        return text
+    return text[:max_len]
+
+
+@dataclass
+class ClaimSide:
+    """One persisting side of a conflict — never dropped for a 'winner'."""
+
+    claim_id: str
+    doc_id: str
+    source_record_kind: str | None
+    source_tier: int
+    publisher: str
+    quote_span: str  # bounded; CLI display-sanitizes via _ansi_safe
+    version_range: dict | None = None
+    exploit_status: str | None = None
+    value: float | None = None
+    unit: str = ""
+
+    def to_dict(self) -> dict:
+        d: dict = {
+            "claim_id": self.claim_id,
+            "doc_id": self.doc_id,
+            "source_record_kind": self.source_record_kind,
+            "source_tier": self.source_tier,
+            "publisher": self.publisher,
+            "quote_span": self.quote_span,
+        }
+        if self.version_range is not None:
+            d["version_range"] = self.version_range
+        if self.exploit_status is not None:
+            d["exploit_status"] = self.exploit_status
+        if self.value is not None:
+            d["value"] = self.value
+            d["unit"] = self.unit
+        # workaround_text intentionally omitted — injection surface.
+        return d
+
+
+@dataclass
+class ConflictGroup:
+    claim_class: str
+    package_or_product: str
+    status: str
+    set_relation: str | None
+    confidence: str
+    flags: list[str]
+    favored_tier: int | None  # display-only ordering hint; not a resolver
+    sides: list[ClaimSide]
+
+    def to_dict(self) -> dict:
+        d = {
+            "claim_class": self.claim_class,
+            "package_or_product": self.package_or_product,
+            "status": self.status,
+            "confidence": self.confidence,
+            "flags": list(self.flags),
+            "sides": [s.to_dict() for s in self.sides],
+        }
+        if self.set_relation is not None:
+            d["set_relation"] = self.set_relation
+        if self.favored_tier is not None:
+            d["favored_tier"] = self.favored_tier
+        return d
+
+
+@dataclass
+class CveConflictReport:
+    cve_id: str
+    groups: list[ConflictGroup] = field(default_factory=list)
+
+    def to_dict(self) -> dict:
+        return {"cve_id": self.cve_id, "groups": [g.to_dict() for g in self.groups]}
+
+
+@dataclass
+class ConflictReport:
+    """Per-CVE conflict ship artifact — not a CVE feed."""
+
+    cves: list[CveConflictReport] = field(default_factory=list)
+
+    def to_dict(self) -> dict:
+        return {"cves": [c.to_dict() for c in self.cves]}
+
+
+def _claims_by_id(claims: list[ClaimView]) -> dict[str, ClaimView]:
+    return {c.claim_id: c for c in claims}
+
+
+def _side_from_claim(claim: ClaimView) -> ClaimSide:
+    return ClaimSide(
+        claim_id=claim.claim_id,
+        doc_id=claim.doc_id,
+        source_record_kind=claim.source_record_kind,
+        source_tier=claim.source_tier,
+        publisher=claim.publisher or "",
+        quote_span=bound_quote_span(claim.quote_span),
+        version_range=claim.version_range,
+        exploit_status=claim.exploit_status,
+        value=claim.value,
+        unit=claim.unit or "",
+    )
+
+
+def _include_assessment(a: Assessment) -> bool:
+    if not a.claim_class or a.claim_class not in _FOCUS_CLASSES:
+        return False
+    if not (a.cve_id or "").strip():
+        return False
+    return a.status in _REPORT_STATUSES
+
+
+def build_conflict_report(
+    analysis: AnalysisReport,
+    claims: list[ClaimView],
+    *,
+    cve_id: str | None = None,
+) -> ConflictReport:
+    """Group advisory assessments into a per-CVE conflict report.
+
+    Never auto-resolves: every member claim becomes a ClaimSide. Filter to
+    affected_range / patched_in / exploit_status with contradicted or
+    weakly_corroborated status. Optional cve_id narrows to one CVE.
+    """
+    by_id = _claims_by_id(claims)
+    wanted = (cve_id or "").strip().upper() or None
+
+    by_cve: dict[str, list[ConflictGroup]] = defaultdict(list)
+    for a in analysis.assessments:
+        if not _include_assessment(a):
+            continue
+        cid = (a.cve_id or "").strip()
+        if wanted and cid.upper() != wanted:
+            continue
+        sides: list[ClaimSide] = []
+        for mid in a.members:
+            claim = by_id.get(mid)
+            if claim is None:
+                # Member id without a joinable claim — still surface the id.
+                sides.append(ClaimSide(
+                    claim_id=mid, doc_id="", source_record_kind=None,
+                    source_tier=0, publisher="", quote_span="",
+                ))
+            else:
+                sides.append(_side_from_claim(claim))
+        sides.sort(key=lambda s: (s.doc_id, s.claim_id))
+        by_cve[cid].append(ConflictGroup(
+            claim_class=a.claim_class or a.metric,
+            package_or_product=a.entity_id,
+            status=a.status,
+            set_relation=a.set_relation,
+            confidence=a.confidence,
+            flags=list(a.flags),
+            favored_tier=a.favored_tier,
+            sides=sides,
+        ))
+
+    cves: list[CveConflictReport] = []
+    for cid in sorted(by_cve):
+        groups = by_cve[cid]
+        groups.sort(key=lambda g: (g.package_or_product, g.claim_class, g.status))
+        cves.append(CveConflictReport(cve_id=cid, groups=groups))
+    return ConflictReport(cves=cves)
+
+
+def conflict_report(
+    config: Config | None = None,
+    *,
+    cve_id: str | None = None,
+) -> ConflictReport:
+    """Library entry: run analysis + build the per-CVE conflict ship artifact."""
+    config = config or load_config()
+    analysis = run_analysis(config)
+    conn = connect(config.paths.db_path)
+    try:
+        claims = get_claims_for_analysis(conn)
+    finally:
+        conn.close()
+    return build_conflict_report(analysis, claims, cve_id=cve_id)

src/semianalyst/cli.py

@@ -21,7 +21,7 @@
 import typer
 
 from . import store
-from .analyze import run_analysis
+from .analyze import conflict_report, run_analysis
 from .config import load_config
 from .extract.pipeline import run_extract, run_rebuild
 from .ingest import SidecarCollision
@@ -243,5 +243,53 @@ def report() -> None:
             )
 
 
+
+@app.command("conflict-report")
+def conflict_report_cmd(
+    cve_id: str = typer.Option(
+        None,
+        help="Optional CVE id to narrow the report (e.g. CVE-2024-21626).",
+    ),
+) -> None:
+    """Print the per-CVE conflict ship artifact (not a CVE feed).
+
+    Surfaces contradicted (and related weakly_corroborated) groups for
+    affected_range / patched_in / exploit_status with each side's bounded,
+    display-sanitized quote_span. Never picks a winning range or kind.
+    """
+    try:
+        report = conflict_report(load_config(), cve_id=cve_id)
+    except store.StoreNotInitialized as exc:
+        typer.echo(str(exc), err=True)
+        raise typer.Exit(1) from exc
+    if not report.cves:
+        typer.echo("no advisory conflicts in the store — nothing to report yet.")
+        return
+    for cve in report.cves:
+        typer.echo(f"CVE {_ansi_safe(cve.cve_id)}")
+        for g in cve.groups:
+            line = (
+                f"  [{_ansi_safe(g.status)}] {_ansi_safe(g.claim_class)}"
+                f" {_ansi_safe(g.package_or_product)}"
+            )
+            if g.set_relation:
+                line += f" set_relation={_ansi_safe(g.set_relation)}"
+            line += f" confidence={_ansi_safe(g.confidence)}"
+            if g.flags:
+                line += f" flags={g.flags}"
+            if g.favored_tier is not None:
+                line += f" favored_tier={g.favored_tier}"
+            typer.echo(line)
+            for side in g.sides:
+                kind = side.source_record_kind or "-"
+                span = _ansi_safe(side.quote_span)
+                typer.echo(
+                    f"    side doc={_ansi_safe(side.doc_id)}"
+                    f" kind={_ansi_safe(kind)}"
+                    f" tier={side.source_tier}"
+                    f" quote={span!r}"
+                )
+
+
 if __name__ == "__main__":
     app()

src/semianalyst/store/db.py

@@ -242,6 +242,8 @@ class ClaimView:
     version_range: dict | None = None
     exploit_status: str | None = None
     workaround_text: str | None = None
+    # GEI-13: citation quote for conflict-report ship artifact (schema Span2000).
+    quote_span: str = ""
 
     def advisory_group_key(self) -> tuple[str, str, str]:
         """GEI-9 grouping key (cve_id, package_or_product, claim_class).
@@ -453,7 +455,8 @@ def get_claims_for_analysis(conn: sqlite3.Connection) -> list[ClaimView]:
             "       c.cmp_is_relative, c.cmp_baseline_entity, c.cond_sparsity, "
             "       c.completeness, d.source_tier, d.publisher, "
             "       c.claim_class, c.source_record_kind, c.cve_id, "
-            "       c.version_range, c.exploit_status, c.workaround_text "
+            "       c.version_range, c.exploit_status, c.workaround_text, "
+            "       c.cite_quote_span "
             "FROM claim c JOIN document d ON c.doc_id = d.doc_id"
         ).fetchall()
     except sqlite3.OperationalError as exc:
@@ -477,6 +480,7 @@ def get_claims_for_analysis(conn: sqlite3.Connection) -> list[ClaimView]:
             version_range=vr,
             exploit_status=r["exploit_status"],
             workaround_text=r["workaround_text"],
+            quote_span=r["cite_quote_span"] or "",
         ))
     return out