refactor(corpus-seeded-mechanisms): apply review round 2 suggestions

- variant_search.search_tree skips a malformed store row with a
  variants_store_row_invalid degradation instead of aborting the scan;
  a hit on a demoted or flowless overlay record is never merged (test).
- seed.pair_lessons is the one shape derivation shared by the exporter
  and leave-one-out; labels pass line and sink together; pairs without a
  labeled function are reported as no_labeled_function.
- config parses [variants].max_mechanisms with the shared int helper.
- regress.candidate.hotspot_from_finding is the one zero-score hotspot
  helper for forced inventory and variant findings.
- design synced (SearchResult, hits_to_findings summaries, candidate.py).

Co-Authored-By: Claude Fable 5.1 <[email protected]>
Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG

.kiro/specs/corpus-seeded-mechanisms/design.md

@@ -108,7 +108,7 @@ tests/
 ### Modified Files
 
 - `semantic/overlay.py` — `OverlayRecord.mechanism_id: str | None = None` (additive).
-- `regress/candidate.py` — variant findings eligible after overlay promotions (reuses forced-hotspot path).
+- `regress/candidate.py` — exposes `hotspot_from_finding`; variant findings enter the ranked candidate slice as zero-score hotspots from `cli.py` (after every scored promotion, inside the cap, through the safety check).
 - `config.py` — `[variants] enabled=true, max_mechanisms=500`.
 
 ## System Flows
@@ -199,8 +199,8 @@ class VariantHit:
     sink_name: str
     source_kind: str
 
-def search_tree(root: Path, targets: Sequence[FileTarget], store: MechanismStore, facts: SemanticFacts, *, max_mechanisms: int) -> list[VariantHit]: ...
-def hits_to_findings(hits, records: Sequence[OverlayRecord]) -> tuple[list[StaticFinding], list[OverlayRecord]]: ...  # merge with overlay flows
+def search_tree(root: Path, targets: Sequence[FileTarget], store: MechanismStore, facts: SemanticFacts, *, max_mechanisms: int) -> SearchResult: ...  # hits, mechanisms_searched, files_searched, degradations
+def hits_to_findings(hits, records: Sequence[OverlayRecord], summaries: Mapping[str, tuple[str, tuple[str, ...], str]]) -> tuple[list[StaticFinding], list[OverlayRecord]]: ...  # merge with overlay flows
 ```
 
 Finding: id `variant:<mechanism_id>:<path>:<line>`, evidence `suspicion`, tags `["variant", "mechanism:<id>"]`, rationale names the mechanism summary and pair provenance.

src/openultrasast/config.py

@@ -343,5 +343,5 @@ def _load_variants(value: object) -> VariantsConfig:
     data = _section(value)
     return VariantsConfig(
         enabled=bool(data.get("enabled", True)),
-        max_mechanisms=int(str(data.get("max_mechanisms", 500))),
+        max_mechanisms=_int_value(data.get("max_mechanisms"), 500),
     )

src/openultrasast/regress/candidate.py

@@ -99,19 +99,24 @@ def _hotspot_for_finding(finding: StaticFinding, hotspots: Sequence[Hotspot]) ->
     return None
 
 
-def _hotspot_from_forced_finding(finding: StaticFinding) -> Hotspot:
+def hotspot_from_finding(finding: StaticFinding, *, rationale: str) -> Hotspot:
+    """A zero-score hotspot wrapping one finding: forced sev-5 inventory or a variant of a known mechanism."""
     return Hotspot(
         path=finding.path,
         function_name=finding.function_name,
         score=0.0,
         band="low",
         signals={},
-        rationale="forced reachable severity-5 inventory",
+        rationale=rationale,
         test_hint=None,
         inventory_finding_ids=(finding.finding_id,),
     )
 
 
+def _hotspot_from_forced_finding(finding: StaticFinding) -> Hotspot:
+    return hotspot_from_finding(finding, rationale="forced reachable severity-5 inventory")
+
+
 class RegressionRunner:
     """Run a snippet in the sandbox only after the structural safety check passes."""
 

src/openultrasast/semantic/seed.py

@@ -14,9 +14,9 @@ from pathlib import Path
 from ..pairs import GATING_TIERS, PairCase
 from ..preprocess import LANGUAGE_BY_EXTENSION
 from .facts import FactLoadError, SemanticFacts, load_facts
-from .ir import parse_file
+from .ir import FileIR, parse_file
 from .mechanisms import MechanismStore, append_from_pair
-from .variants import derive_shape
+from .variants import Shape, derive_shape
 
 LANGUAGE_UNSUPPORTED = "variants_language_unsupported"
 
@@ -67,36 +67,22 @@ def export_mechanisms(cases: Sequence[PairCase], store: MechanismStore, *, facts
             else:
                 skipped.append((case.name, "parse_failed"))
             continue
-        produced = 0
-        for row in case.expected:
-            if not row.function or not row.mechanism:
-                continue
-            shape = derive_shape(
-                vuln_ir,
-                fixed_ir,
-                function=row.function,
-                sink=row.sink,
-                line=None if row.sink else row.line,
-                mechanism=row.mechanism,
-                facts=loaded,
-                vuln_text=vuln_text,
-                fixed_text=fixed_text,
-                cwe=row.cwe or None,
-            )
-            if shape is None:
-                continue
+        if not any(row.function and row.mechanism for row in case.expected):
+            skipped.append((case.name, "no_labeled_function"))
+            continue
+        lessons = pair_lessons(case, loaded, vuln_ir=vuln_ir, fixed_ir=fixed_ir, vuln_text=vuln_text, fixed_text=fixed_text)
+        for lesson in lessons:
             record = append_from_pair(
                 store,
-                shape,
-                summary=f"{shape.mechanism}: {'/'.join(shape.source_kinds)} into {shape.sink_name} ({row.cwe}); fix adds {shape.guard}",
-                cwe=row.cwe,
+                lesson.shape,
+                summary=lesson.summary,
+                cwe=lesson.cwe,
                 pair=case.name,
                 provenance=case.provenance,
                 tier=case.review_tier,
             )
             record_ids.add(record.id)
-            produced += 1
-        if produced:
+        if lessons:
             seeded.append(case.name)
         else:
             skipped.append((case.name, "no_labeled_sink_call_site"))
@@ -109,6 +95,54 @@ def export_mechanisms(cases: Sequence[PairCase], store: MechanismStore, *, facts
     )
 
 
+@dataclass(frozen=True)
+class Lesson:
+    """One shape a trusted pair teaches, with the CWE and a text-free summary."""
+
+    shape: Shape
+    cwe: str
+    summary: str
+
+
+def pair_lessons(
+    case: PairCase,
+    facts: SemanticFacts,
+    *,
+    vuln_ir: FileIR | None = None,
+    fixed_ir: FileIR | None = None,
+    vuln_text: str | None = None,
+    fixed_text: str | None = None,
+) -> list[Lesson]:
+    """Shapes derived from every labeled row of a pair (shared by the exporter and leave-one-out). Empty when a side fails to parse."""
+    vuln_text = vuln_text if vuln_text is not None else _read(case.vuln_file)
+    fixed_text = fixed_text if fixed_text is not None else _read(case.fixed_file)
+    language = _parse_language(case)
+    vuln_ir = vuln_ir if vuln_ir is not None else parse_file(case.relpath, vuln_text, language)
+    fixed_ir = fixed_ir if fixed_ir is not None else parse_file(case.relpath, fixed_text, language)
+    if not vuln_ir.parse_ok or not fixed_ir.parse_ok:
+        return []
+    lessons: list[Lesson] = []
+    for row in case.expected:
+        if not row.function or not row.mechanism:
+            continue
+        shape = derive_shape(
+            vuln_ir,
+            fixed_ir,
+            function=row.function,
+            sink=row.sink,
+            line=row.line,
+            mechanism=row.mechanism,
+            facts=facts,
+            vuln_text=vuln_text,
+            fixed_text=fixed_text,
+            cwe=row.cwe or None,
+        )
+        if shape is not None:
+            summary = f"{shape.mechanism}: {'/'.join(shape.source_kinds)} into {shape.sink_name} ({row.cwe}); fix adds {shape.guard}"
+            lessons.append(Lesson(shape=shape, cwe=row.cwe, summary=summary))
+    return lessons
+
+
 def _skip_reason(case: PairCase) -> str | None:
     if case.review_tier not in GATING_TIERS:
         return f"tier:{case.review_tier}"

src/openultrasast/semantic/variant_search.py

@@ -41,16 +41,22 @@ def search_tree(
     records = corpus_mechanisms(store.load())[: max(max_mechanisms, 0)]
     shapes: list[Shape] = []
     ids: dict[str, str] = {}
+    degradations: list[dict[str, object]] = []
     for record in records:
         assert record.shape is not None
-        shape = Shape.from_dict(record.shape)
+        try:
+            shape = Shape.from_dict(record.shape)
+        except (ValueError, TypeError) as exc:  # one malformed operator-visible row must not abort the scan
+            degradations.append(
+                {"stage": "variants", "reason": "variants_store_row_invalid", "mechanism_id": record.id, "detail": str(exc)[:120]}
+            )
+            continue
         shapes.append(shape)
         ids[shape.key()] = record.id
     if not shapes:
-        return SearchResult(hits=(), mechanisms_searched=0, files_searched=0)
+        return SearchResult(hits=(), mechanisms_searched=0, files_searched=0, degradations=tuple(degradations))
     hits: list[VariantHit] = []
     files = 0
-    degradations: list[dict[str, object]] = []
     unsupported: set[str] = set()
     for target in targets:
         try:

tests/test_variant_search.py

@@ -96,3 +96,29 @@ def test_search_tree_uses_the_store_and_respects_the_mechanism_cap(tmp_path: Pat
     assert result.mechanisms_searched == 1 and result.files_searched == 2
     capped = search_tree(root, targets, store, PY_FACTS, max_mechanisms=0)
     assert capped.hits == () and capped.mechanisms_searched == 0
+
+
+def test_hits_never_merge_into_demoted_or_flowless_records() -> None:
+    from openultrasast.semantic.variant_search import hits_to_findings
+    from openultrasast.semantic.variants import VariantHit
+
+    hit = VariantHit(path="a.py", line=7, mechanism_id="corpus:abc", sink_name="system", source_kind="parameter")
+    demoted = OverlayRecord(
+        proposal_id="python-os-command:a.py:7",
+        path="a.py",
+        line=7,
+        disposition="demote",
+        reason="constant argument",
+        cwe="CWE-78",
+        sources=(),
+        sinks=("os.system",),
+        sanitizers=(),
+        evidence_level="suspicion",
+        dominating_fact="constant",
+        origin="inventory",
+        engine="python-ast",
+        language="python",
+    )
+    findings, records = hits_to_findings([hit], [demoted], {})
+    assert records[0].mechanism_id is None and records[0].disposition == "demote"  # Req 3.6: never demote, never re-label a demotion
+    assert [f.finding_id for f in findings] == ["variant:corpus:abc:a.py:7"]