rune/tests/test_capability_fit.py

"""test_capability_fit.py — SCAFFOLD-TRANCHE-003 §27 test plan (A-K).

Structural discovery, requirement-fit, gap/resolution surface, composition,
acquire-external seam, semantic-fit evidence recording.
"""

import os
import sys

import pytest

sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src"))

from rune import capability_fit as cf  # noqa: E402
from rune import structural_store as ss  # noqa: E402


def _doc_with(*caps):
    """Build a store doc with capability fixtures.
    Each cap: (id, provides, consumers, maturity, environment, status_extra)"""
    doc = ss.new_doc()
    for c in caps:
        cid, provides, consumers, maturity = c[0], c[1], c[2], c[3]
        env = c[4] if len(c) > 4 else {}
        extra = c[5] if len(c) > 5 else {}
        ss.add_entity(
            doc,
            id=cid,
            kind="capability",
            identities={"environment": env},
            maturity=maturity,
            evidence_refs=["fixture"],
        )
        for p in provides:
            ss.add_entity(doc, id=p, kind="contract", identities={}, evidence_refs=["fixture"])
            ss.add_relationship(doc, type="implements", from_id=cid, to_id=p, evidence_refs=["fixture"])
        for cons in consumers:
            ss.add_entity(doc, id=cons, kind="module", identities={}, evidence_refs=["fixture"])
            ss.add_relationship(doc, type="consumes", from_id=cons, to_id=cid, evidence_refs=["fixture"])
        for k, v in extra.items():
            e = ss.entity(doc, cid)
            if k == "superseded_by":
                e["superseded_by"] = v
            elif k == "status":
                e["status"] = v
    return doc


REQ_STREAM = dict(contracts=["contract:streaming", "contract:bounded_memory"], environment={"os": "linux"})


# ── A. structural candidate discovery ──────────────────────────────────────


class TestDiscovery:
    def test_finds_capability_kinds_and_providers(self):
        doc = _doc_with(("capability:parser", ["contract:streaming"], [], "WIRED"))
        req = cf.CapabilityRequirement("parse streams", **REQ_STREAM)
        ids = [c["id"] for c in cf.discover_candidates(doc, req)]
        assert "capability:parser" in ids

    def test_open_world_kind_with_provides_is_discoverable(self):
        doc = _doc_with()
        ss.add_entity(doc, id="weird_thing", kind="quantum_gizmo", identities={}, evidence_refs=["f"])
        ss.add_entity(doc, id="contract:streaming", kind="contract", identities={}, evidence_refs=["f"])
        ss.add_relationship(
            doc, type="provides", from_id="weird_thing", to_id="contract:streaming", evidence_refs=["f"]
        )
        req = cf.CapabilityRequirement("x", **REQ_STREAM)
        assert "weird_thing" in [c["id"] for c in cf.discover_candidates(doc, req)]

    def test_environment_contradiction_excludes_candidate(self):
        doc = _doc_with(("capability:winonly", ["contract:streaming"], [], "WIRED", {"os": "windows"}))
        req = cf.CapabilityRequirement("x", **REQ_STREAM)
        assert "capability:winonly" not in [c["id"] for c in cf.discover_candidates(doc, req)]

    def test_unknown_environment_is_not_contradiction(self):
        doc = _doc_with(("capability:anyos", ["contract:streaming"], [], "WIRED", {}))
        req = cf.CapabilityRequirement("x", **REQ_STREAM)
        assert "capability:anyos" in [c["id"] for c in cf.discover_candidates(doc, req)]


# ── B/C/D/E. fit levels ────────────────────────────────────────────────────


class TestFit:
    def _cand(self, doc, cid):
        return next(
            c for c in cf.discover_candidates(doc, cf.CapabilityRequirement("x", **REQ_STREAM)) if c["id"] == cid
        )

    def test_exact_fit(self):
        doc = _doc_with(("capability:p", ["contract:streaming", "contract:bounded_memory"], [], "WIRED"))
        req = cf.CapabilityRequirement("x", **REQ_STREAM)
        f = cf.evaluate_fit(req, self._cand(doc, "capability:p"))
        assert f["fit"] == "EXACT"

    def test_partial_fit(self):
        doc = _doc_with(("capability:p", ["contract:streaming"], [], "WIRED"))
        req = cf.CapabilityRequirement("x", **REQ_STREAM)
        f = cf.evaluate_fit(req, self._cand(doc, "capability:p"))
        assert f["fit"] == "PARTIAL"
        assert f["missing_contracts"] == ["contract:bounded_memory"]

    def test_incompatible_candidate(self):
        doc = _doc_with(("capability:p", ["contract:other"], [], "WIRED"))
        req = cf.CapabilityRequirement("x", **REQ_STREAM)
        f = cf.evaluate_fit(req, self._cand(doc, "capability:p"))
        assert f["fit"] == "INCOMPATIBLE"

    def test_stale_candidate_rejected_as_incompatible(self):
        doc = _doc_with(
            ("capability:p", ["contract:streaming", "contract:bounded_memory"], [], "WIRED", {}, {"status": "STALE"})
        )
        req = cf.CapabilityRequirement("x", **REQ_STREAM)
        f = cf.evaluate_fit(req, self._cand(doc, "capability:p"))
        assert f["fit"] == "INCOMPATIBLE"
        assert "STALE" in f["reason"]

    def test_superseded_candidate_rejected(self):
        doc = _doc_with(
            (
                "capability:old",
                ["contract:streaming", "contract:bounded_memory"],
                [],
                "WIRED",
                {},
                {"superseded_by": "capability:new"},
            )
        )
        req = cf.CapabilityRequirement("x", **REQ_STREAM)
        f = cf.evaluate_fit(req, self._cand(doc, "capability:old"))
        assert f["fit"] == "INCOMPATIBLE"

    def test_unknown_fit_when_no_contracts(self):
        doc = _doc_with()
        ss.add_entity(
            doc, id="capability:opaque", kind="capability", identities={}, maturity="WIRED", evidence_refs=["f"]
        )
        req = cf.CapabilityRequirement("x", **REQ_STREAM)
        cand = next(c for c in cf.discover_candidates(doc, req) if c["id"] == "capability:opaque")
        f = cf.evaluate_fit(req, cand)
        assert f["fit"] == "UNKNOWN"


# ── F/G/H/I/J. resolution surface ──────────────────────────────────────────


class TestResolution:
    def test_exact_fit_reuses(self):
        doc = _doc_with(("capability:p", ["contract:streaming", "contract:bounded_memory"], [], "WIRED"))
        out = cf.resolve(doc, cf.CapabilityRequirement("x", **REQ_STREAM))
        assert out["resolution"] == "REUSE"
        assert out["selected"] == "capability:p"

    def test_satisfied_when_context_already_consumes(self):
        doc = _doc_with(("capability:p", ["contract:streaming", "contract:bounded_memory"], ["module:mytask"], "WIRED"))
        req = cf.CapabilityRequirement("x", already_consumed_by="module:mytask", **REQ_STREAM)
        out = cf.resolve(doc, req)
        assert out["resolution"] == "SATISFIED"

    def test_clean_partial_adaptation_selected(self):
        doc = _doc_with(("capability:p", ["contract:streaming"], [], "WIRED"))
        req = cf.CapabilityRequirement("x", **REQ_STREAM)
        out = cf.resolve(doc, req)
        assert out["resolution"] == "ADAPT_EXISTING"
        assert out["selected"] == "capability:p"
        assert out["obligations"]["regression_proof"] is True

    def test_brittle_coupling_forces_build_new_with_rationale(self):
        # bounded_memory is load-bearing: another module consumes it
        doc = _doc_with(
            ("capability:p", ["contract:streaming"], [], "WIRED"),
            ("contract:bounded_memory", [], [], "WIRED"),
            ("module:hot_consumer", [], [], "WIRED"),
        )
        ss.add_relationship(
            doc, type="consumes", from_id="module:hot_consumer", to_id="contract:bounded_memory", evidence_refs=["f"]
        )
        req = cf.CapabilityRequirement("x", **REQ_STREAM)
        out = cf.resolve(doc, req)
        assert out["resolution"] == "BUILD_NEW"
        assert "brittle" in out["rationale"].lower() or "load-bearing" in out.get("brittle", "")
        assert out["obligations"]["new_behavior_proof"] is True

    def test_two_capability_composition(self):
        doc = _doc_with(
            ("capability:a", ["contract:streaming"], [], "WIRED"),
            ("capability:b", ["contract:bounded_memory"], [], "WIRED"),
        )
        req = cf.CapabilityRequirement("x", **REQ_STREAM)
        out = cf.resolve(doc, req)
        assert out["resolution"] == "COMPOSE"
        assert set(out["components"]) == {"capability:a", "capability:b"}
        assert out["permanence"] == "temporary (user-controlled)"

    def test_no_candidates_no_external_evidence_insufficient(self):
        doc = _doc_with()
        req = cf.CapabilityRequirement("x", **REQ_STREAM)
        out = cf.resolve(doc, req)
        assert out["resolution"] == "INSUFFICIENT_EVIDENCE"
        assert "BUILD_NEW not silently chosen" in out["rationale"]

    def test_no_candidates_with_external_evidence_acquires(self):
        doc = _doc_with()
        req = cf.CapabilityRequirement("x", **REQ_STREAM)
        out = cf.resolve(
            doc,
            req,
            external_evidence=[{"tool": "ripgrep", "evidence": "mature streaming regex engine", "license": "MIT"}],
        )
        assert out["resolution"] == "ACQUIRE_EXTERNAL"
        assert out["external"][0]["tool"] == "ripgrep"

    def test_all_incompatible_no_evidence_insufficient_not_build(self):
        doc = _doc_with(("capability:p", ["contract:other"], [], "WIRED"))
        req = cf.CapabilityRequirement("x", **REQ_STREAM)
        out = cf.resolve(doc, req)
        assert out["resolution"] == "INSUFFICIENT_EVIDENCE"

    def test_adapt_brittle_with_external_prefers_acquire(self):
        doc = _doc_with(
            ("capability:p", ["contract:streaming"], [], "WIRED"),
            ("contract:bounded_memory", [], [], "WIRED"),
            ("module:hot_consumer", [], [], "WIRED"),
        )
        ss.add_relationship(
            doc, type="consumes", from_id="module:hot_consumer", to_id="contract:bounded_memory", evidence_refs=["f"]
        )
        req = cf.CapabilityRequirement("x", **REQ_STREAM)
        out = cf.resolve(doc, req, external_evidence=[{"tool": "libstream", "license": "Apache-2.0"}])
        assert out["resolution"] == "ACQUIRE_EXTERNAL"

    def test_unknown_only_candidates_insufficient(self):
        doc = _doc_with()
        ss.add_entity(
            doc, id="capability:opaque", kind="capability", identities={}, maturity="WIRED", evidence_refs=["f"]
        )
        req = cf.CapabilityRequirement("x", **REQ_STREAM)
        out = cf.resolve(doc, req)
        assert out["resolution"] == "INSUFFICIENT_EVIDENCE"


# ── K. semantic-fit evidence recording ─────────────────────────────────────


class TestSemanticSeam:
    def test_record_requires_provider_and_model(self):
        with pytest.raises(ValueError):
            cf.record_semantic_fit("", "", "why", [], 0.5, "prose")

    def test_record_validates_confidence(self):
        with pytest.raises(ValueError):
            cf.record_semantic_fit("p", "m", "why", [], 1.5, "prose")

    def test_record_carries_full_provenance(self):
        r = cf.record_semantic_fit("local-ollama", "llama3", "name ambiguity", ["evidence-1"], 0.6, "probably fits")
        assert r["provider"] == "local-ollama" and r["model"] == "llama3"
        assert r["confidence"] == 0.6
        assert r["kind"] == "semantic_fit"

    def test_decision_carries_semantic_records_not_authority(self):
        doc = _doc_with(("capability:opaque", [], [], "WIRED"))
        req = cf.CapabilityRequirement("x", **REQ_STREAM)
        sem = [cf.record_semantic_fit("p", "m", "no contracts to check", ["evidence"], 0.4, "may fit")]
        out = cf.resolve(doc, req, semantic_fits=sem)
        assert out["resolution"] == "INSUFFICIENT_EVIDENCE"  # prose ≠ authority
        assert out["semantic_fits"][0]["provider"] == "p"