diff --git a/.github/workflows/solution-ranker.yml b/.github/workflows/solution-ranker.yml new file mode 100644 index 000000000..f70365905 --- /dev/null +++ b/.github/workflows/solution-ranker.yml @@ -0,0 +1,39 @@ +name: solution-ranker + +# Walks MeshActionRegistry for ABB slot fillers and ranks them into a SolutionShortlist. +# Backs Michael's `shortlist_solutions` capability (prophet-mesh#17) — the move that turns +# "silently auto-route to the top score" into "here are your options, and here is why". + +on: + push: + paths: + - 'libs/python/solution-ranker/**' + - '.github/workflows/solution-ranker.yml' + pull_request: + paths: + - 'libs/python/solution-ranker/**' + - '.github/workflows/solution-ranker.yml' + +permissions: + contents: read + +jobs: + test: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: '3.11' + + - name: Install + run: | + python -m pip install --upgrade pip + pip install -r libs/python/solution-ranker/requirements-test.txt -e libs/python/solution-ranker + + # Covers filter-before-rank (a below-floor candidate is not rankable, it is excluded), + # the four auto-route preconditions with per-blocker reporting, and the Eve Smith + # cross-selling scenario from Zurich's E-RDA2 deck run end to end. + - name: Tests + run: cd libs/python/solution-ranker && pytest -q diff --git a/libs/python/solution-ranker/Makefile b/libs/python/solution-ranker/Makefile new file mode 100644 index 000000000..b202ca514 --- /dev/null +++ b/libs/python/solution-ranker/Makefile @@ -0,0 +1,8 @@ +.PHONY: install test + +install: + pip install -e ".[dev]" + +test: + pip install -q -e ".[dev]" + pytest -q diff --git a/libs/python/solution-ranker/pyproject.toml b/libs/python/solution-ranker/pyproject.toml new file mode 100644 index 000000000..8984f77f1 --- /dev/null +++ b/libs/python/solution-ranker/pyproject.toml @@ -0,0 +1,16 @@ +[build-system] +requires = ["setuptools>=68", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "prophet-solution-ranker" +version = "0.1.0" +description = "Walk the MeshActionRegistry for ABB slot fillers and rank them into a SolutionShortlist" +requires-python = ">=3.10" +dependencies = [] + +[project.optional-dependencies] +dev = ["pytest>=8.3"] + +[tool.setuptools.packages.find] +where = ["src"] diff --git a/libs/python/solution-ranker/pytest.ini b/libs/python/solution-ranker/pytest.ini new file mode 100644 index 000000000..80432c220 --- /dev/null +++ b/libs/python/solution-ranker/pytest.ini @@ -0,0 +1,3 @@ +[pytest] +testpaths = tests +pythonpath = src diff --git a/libs/python/solution-ranker/requirements-test.txt b/libs/python/solution-ranker/requirements-test.txt new file mode 100644 index 000000000..67ec4d421 --- /dev/null +++ b/libs/python/solution-ranker/requirements-test.txt @@ -0,0 +1 @@ +pytest>=8.3 diff --git a/libs/python/solution-ranker/src/solution_ranker/__init__.py b/libs/python/solution-ranker/src/solution_ranker/__init__.py new file mode 100644 index 000000000..e93cd3178 --- /dev/null +++ b/libs/python/solution-ranker/src/solution_ranker/__init__.py @@ -0,0 +1,301 @@ +"""Walk the MeshActionRegistry for ABB slot fillers and rank them into a SolutionShortlist. + +The move this makes possible: Michael says *"we have a few options — do you already have a +model?"* instead of silently auto-routing to whatever scored highest. Zurich's Damian does the +former in its script and the latter in its architecture; the gap between those is where a +user loses the ability to see what was almost chosen. + +The pipeline: + + intent set ──► ABB requirement ──► MAR walk (implementsAbb) ──► filter ──► rank + │ + trust floor ────────────────►─┤ + access grade ──────────────►──┤ + counter-test ─────────────►───┘ + +Filtering happens BEFORE ranking, deliberately. A candidate below the trust floor is not a +low-ranked option, it is not an option — ranking it would put it on screen where a user could +pick it. Same for a denied access grade. + +Auto-route is the narrowest possible claim: top-2 gap over threshold AND counter-test +confirmed AND access granted AND at least two candidates to compare. Anything less returns +`user-pick`, and `abstain` when there is nothing to offer. Those three states are the whole +API surface — there is no boolean "did it route". + +Composes with: + - ``access_prewalk`` (prophet-platform) for the access grade + - sourceos-spec ``ArchitecturalBuildingBlock`` + ``MeshActionRegistry.implementsAbb`` + - prophet-mesh ``specs/solution-shortlist.schema.json`` for the emitted shape + - Noetica's counter-test gate (#570) for ``counterTestStatus`` +""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Any, Literal, Sequence + +__all__ = [ + "CounterTestStatus", "RouteDecision", "AUTO_ROUTE_GAP_THRESHOLD", + "Participant", "Candidate", "Shortlist", + "rank_candidates", "build_shortlist", "canonical_json", +] + +CounterTestStatus = Literal["confirmed", "available", "unavailable"] +RouteDecision = Literal["auto-route", "user-pick", "abstain"] + +#: Mirrors prophet_mesh.solution_shortlist.AUTO_ROUTE_GAP_THRESHOLD. Pinned in both places +#: and asserted equal by test, so a policy change in one repo cannot silently diverge from +#: the other — two services that both believe they gate at the same threshold, gating +#: differently, is exactly the drift the shared-vector discipline exists to catch. +AUTO_ROUTE_GAP_THRESHOLD = 0.15 + + +@dataclass(frozen=True) +class Participant: + """One MeshActionRegistry participant, as far as ranking cares. + + ``implements_abb`` is the CLAIM the registry carries — verifying that the participant + actually satisfies the ABB's protocol is the consumer's job, not the registry's (see + sourceos-spec#224). ``protocol_verified`` records whether that verification ran; an + unverified claim can still be shortlisted, but it cannot auto-route. + """ + + repo: str + implements_abb: tuple[str, ...] = () + trust_score: float = 0.0 + catalog_verbs: tuple[str, ...] = () + counter_test_status: CounterTestStatus = "unavailable" + counter_test_ref: str | None = None + protocol_verified: bool = False + + +@dataclass(frozen=True) +class Candidate: + repo: str + score: float + match_reason: tuple[str, ...] + counter_test_status: CounterTestStatus + counter_test_ref: str | None + access_grade: str + access_reason: str + remediation_url: str | None = None + + def to_json(self) -> dict[str, Any]: + out: dict[str, Any] = { + "participantRef": self.repo, + "score": round(self.score, 4), + "matchReason": list(self.match_reason), + "counterTestStatus": self.counter_test_status, + "accessDecision": {"grade": self.access_grade}, + } + if self.counter_test_ref: + out["counterTestRef"] = self.counter_test_ref + if self.remediation_url: + out["accessDecision"]["remediation"] = { + "url": self.remediation_url, + "expectedReturn": "ArtifactConsentRecord", + } + return out + + +@dataclass +class Shortlist: + candidates: list[Candidate] + decision: RouteDecision + decision_reason: str + chosen_index: int | None + empty_reason: str | None + #: Participants filtered out BEFORE ranking, with why. Not decoration: a caller asking + #: "why isn't X here" must get an answer, and an unexplained absence is indistinguishable + #: from a walk that never saw X. + excluded: list[tuple[str, str]] = field(default_factory=list) + + def to_json(self, *, abb: str | None, mar_digest: str, catalog_digest: str) -> dict[str, Any]: + out: dict[str, Any] = { + "schemaVersion": "0.1.0", + "kind": "SolutionShortlist", + "shortlist": [c.to_json() for c in self.candidates], + "derivedAt": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), + "derivedFrom": {"marDigest": mar_digest, "abbCatalogDigest": catalog_digest}, + } + if abb: + out["abbRequirement"] = abb + if self.empty_reason: + out["emptyReason"] = self.empty_reason + else: + verdict: dict[str, Any] = {"decision": self.decision, "reason": self.decision_reason} + if self.chosen_index is not None: + verdict["chosenIndex"] = self.chosen_index + out["autoRouteVerdict"] = verdict + return out + + +def rank_candidates( + participants: Sequence[Participant], + *, + abb: str | None, + intent_verbs: Sequence[str], + min_trust: float, + access_grader, +) -> tuple[list[Candidate], list[tuple[str, str]]]: + """Filter then rank. Returns ``(candidates, excluded)``. + + ``access_grader`` is injected rather than imported so this lib does not hard-depend on + access-prewalk — a caller can supply a stub in tests, and a different estate can supply a + different grader. It takes a repo id and returns ``(grade, reason, remediation_url)``. + + Exclusions are RECORDED, not dropped. Three reasons a participant never reaches ranking: + - it does not claim the required ABB (not a candidate at all) + - its trust score is below the floor (the ledger's judgement is a gate, not a penalty) + - access is denied (offering it would put an unusable option on screen) + + A `requires-consent` candidate IS ranked — that is the grade whose whole purpose is to be + shown with a path forward. + """ + candidates: list[Candidate] = [] + excluded: list[tuple[str, str]] = [] + + for p in participants: + if abb and abb not in p.implements_abb: + excluded.append((p.repo, f"does not claim {abb} in implementsAbb")) + continue + + if p.trust_score < min_trust: + excluded.append(( + p.repo, + f"trust {p.trust_score:.2f} below floor {min_trust:.2f} — the ledger's " + f"judgement gates candidacy; a low-trust option must not be rankable", + )) + continue + + grade, access_reason, remediation = access_grader(p.repo) + if grade == "denied": + excluded.append((p.repo, f"access denied: {access_reason}")) + continue + + reasons: list[str] = [] + overlap = sorted(set(intent_verbs) & set(p.catalog_verbs)) + if overlap: + reasons.append(f"verb overlap {overlap} between intent set and participant catalogue") + if abb: + reasons.append( + f"{abb} claim {'VERIFIED against protocol' if p.protocol_verified else 'declared but unverified'}" + ) + reasons.append(f"trust {p.trust_score:.2f} clears floor {min_trust:.2f}") + + # Score: verb coverage x trust, penalised when the ABB claim was never verified. + # An unverified claim is not disqualifying (it still shows) but it must not outrank + # a verified one on equal evidence — the registry establishes the claim, not its truth. + coverage = (len(overlap) / len(intent_verbs)) if intent_verbs else 0.5 + verification_factor = 1.0 if (p.protocol_verified or not abb) else 0.8 + consent_factor = 1.0 if grade == "granted" else 0.85 + score = coverage * p.trust_score * verification_factor * consent_factor + + candidates.append(Candidate( + repo=p.repo, score=score, match_reason=tuple(reasons), + counter_test_status=p.counter_test_status, counter_test_ref=p.counter_test_ref, + access_grade=grade, access_reason=access_reason, remediation_url=remediation, + )) + + candidates.sort(key=lambda c: c.score, reverse=True) + return candidates, excluded + + +def build_shortlist( + participants: Sequence[Participant], + *, + abb: str | None, + intent_verbs: Sequence[str], + min_trust: float, + access_grader, + counter_test_required: bool = True, +) -> Shortlist: + """Produce the full shortlist including the route decision. + + The route decision is the narrowest claim the evidence supports: + + abstain nothing to offer — every participant was excluded, or none claimed the ABB + user-pick candidates exist but auto-route's preconditions are not all met + auto-route top-2 gap > threshold AND counter-test confirmed AND access granted + AND at least 2 candidates + + Every path to `user-pick` states WHICH precondition failed. "Not confident enough" is not + a reason a user can act on; "the runner-up scored within 0.04" is. + """ + candidates, excluded = rank_candidates( + participants, abb=abb, intent_verbs=intent_verbs, + min_trust=min_trust, access_grader=access_grader, + ) + + if not candidates: + why = "; ".join(f"{r}: {reason}" for r, reason in excluded) or "no participants supplied" + return Shortlist( + candidates=[], decision="abstain", + decision_reason="no candidate survived filtering", + chosen_index=None, + empty_reason=( + f"no participant is a viable filler for {abb or 'this intent'}. Excluded: {why}" + ), + excluded=excluded, + ) + + top = candidates[0] + + if len(candidates) < 2: + return Shortlist( + candidates=candidates, decision="user-pick", + decision_reason=( + "only one candidate — with nothing to compare against, the top-2 gap is " + "undefined and 'auto' would be the caller's default rather than a measured " + "decision" + ), + chosen_index=None, empty_reason=None, excluded=excluded, + ) + + gap = top.score - candidates[1].score + blockers: list[str] = [] + if gap < AUTO_ROUTE_GAP_THRESHOLD: + blockers.append( + f"top-2 gap {gap:.3f} is under the {AUTO_ROUTE_GAP_THRESHOLD} threshold " + f"({top.repo} {top.score:.3f} vs {candidates[1].repo} {candidates[1].score:.3f})" + ) + if counter_test_required and top.counter_test_status != "confirmed": + blockers.append( + f"top candidate's counter-test is '{top.counter_test_status}', not 'confirmed'" + ) + if top.access_grade != "granted": + blockers.append(f"top candidate's access is '{top.access_grade}', not 'granted'") + + if blockers: + return Shortlist( + candidates=candidates, decision="user-pick", + decision_reason="; ".join(blockers), chosen_index=None, + empty_reason=None, excluded=excluded, + ) + + return Shortlist( + candidates=candidates, decision="auto-route", + decision_reason=( + f"top candidate {top.repo} scored {top.score:.3f}, " + f"{gap:.3f} clear of the runner-up (>{AUTO_ROUTE_GAP_THRESHOLD}); " + f"counter-test confirmed; access granted" + ), + chosen_index=0, empty_reason=None, excluded=excluded, + ) + + +def canonical_json(obj: Any) -> str: + """Canonical JSON — recursive key sort, no whitespace, non-ASCII RAW. + + ``ensure_ascii=False`` matches lawful-verdict and the TypeScript canonicaliser. The + default would escape non-ASCII and every digest over accented content would diverge + between languages (the bug caught in review on #1065). + """ + return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + + +def digest(obj: Any) -> str: + return "sha256:" + hashlib.sha256(canonical_json(obj).encode("utf-8")).hexdigest() diff --git a/libs/python/solution-ranker/tests/test_solution_ranker.py b/libs/python/solution-ranker/tests/test_solution_ranker.py new file mode 100644 index 000000000..26b169cdc --- /dev/null +++ b/libs/python/solution-ranker/tests/test_solution_ranker.py @@ -0,0 +1,346 @@ +"""solution-ranker contract tests, plus the Eve Smith end-to-end scenario. + +The invariants pinned here: + + 1. Filtering happens BEFORE ranking — a below-floor or access-denied participant is not a + low-ranked option, it is not an option. Ranking it would put it on screen where a user + could pick it. + 2. Every exclusion is RECORDED with a reason. An unexplained absence is indistinguishable + from a walk that never saw the participant. + 3. `auto-route` requires ALL of: gap > threshold, counter-test confirmed, access granted, + >= 2 candidates. Each failure path names WHICH precondition blocked it — "not confident + enough" is not a reason a user can act on. + 4. A `requires-consent` candidate IS ranked (that grade exists to be shown with a path + forward) but cannot auto-route. +""" + +from __future__ import annotations + +import pytest + +from solution_ranker import ( + AUTO_ROUTE_GAP_THRESHOLD, + Participant, + build_shortlist, + canonical_json, + digest, + rank_candidates, +) + +# A grader that grants everything — isolates ranking behaviour from access behaviour. +GRANT_ALL = lambda repo: ("granted", "no gate on this resource", None) +DENY_ALL = lambda repo: ("denied", "role missing and resource is not consentable", None) + + +def _p(repo: str, **kw) -> Participant: + return Participant(**{ + "repo": repo, "implements_abb": ("ABB.03",), "trust_score": 0.80, + "catalog_verbs": ("retrieve", "evaluate"), "counter_test_status": "confirmed", + "counter_test_ref": f"urn:srcos:ctest:{repo}", "protocol_verified": True, **kw, + }) + + +# ── filter before rank ───────────────────────────────────────────────────────── + +def test_below_trust_floor_is_EXCLUDED_not_low_ranked() -> None: + """A low-trust option must not be rankable. If it were merely low-ranked it would still + be on screen, and a user could pick what the ledger already judged untrustworthy.""" + cands, excluded = rank_candidates( + [_p("org/good"), _p("org/untrusted", trust_score=0.10)], + abb="ABB.03", intent_verbs=["retrieve"], min_trust=0.45, access_grader=GRANT_ALL, + ) + assert [c.repo for c in cands] == ["org/good"] + assert any("org/untrusted" == r and "below floor" in why for r, why in excluded) + + +def test_access_denied_is_EXCLUDED_not_ranked() -> None: + cands, excluded = rank_candidates( + [_p("org/a")], abb="ABB.03", intent_verbs=["retrieve"], + min_trust=0.45, access_grader=DENY_ALL, + ) + assert cands == [] + assert any("access denied" in why for _, why in excluded) + + +def test_not_claiming_the_abb_is_EXCLUDED() -> None: + cands, excluded = rank_candidates( + [_p("org/db", implements_abb=("ABB.03",)), _p("org/other", implements_abb=("ABB.07",))], + abb="ABB.03", intent_verbs=["retrieve"], min_trust=0.45, access_grader=GRANT_ALL, + ) + assert [c.repo for c in cands] == ["org/db"] + assert any("does not claim ABB.03" in why for _, why in excluded) + + +def test_every_exclusion_carries_a_substantive_reason() -> None: + """An unexplained absence is indistinguishable from a walk that never saw the participant. + A caller asking 'why isn't X here' must get an answer.""" + _, excluded = rank_candidates( + [ + _p("org/wrong-abb", implements_abb=("ABB.99",)), + _p("org/low-trust", trust_score=0.01), + ], + abb="ABB.03", intent_verbs=["retrieve"], min_trust=0.45, access_grader=GRANT_ALL, + ) + assert len(excluded) == 2 + for repo, why in excluded: + assert len(why) > 20, f"{repo}: reason too thin — {why!r}" + + +def test_requires_consent_IS_ranked_because_that_grade_exists_to_be_shown() -> None: + """The grade whose entire purpose is 'not yet, but here's the path'. Excluding it would + collapse it into denied — the exact thing the three-grade shape prevents.""" + grader = lambda repo: ("requires-consent", "missing billing-reader", + "https://consent.example/request?sig=abc") + cands, excluded = rank_candidates( + [_p("org/a")], abb="ABB.03", intent_verbs=["retrieve"], + min_trust=0.45, access_grader=grader, + ) + assert len(cands) == 1 + assert cands[0].access_grade == "requires-consent" + assert cands[0].remediation_url is not None + assert excluded == [] + + +# ── scoring ──────────────────────────────────────────────────────────────────── + +def test_unverified_abb_claim_scores_below_a_verified_one_on_equal_evidence() -> None: + """The registry establishes the CLAIM, not its truth (sourceos-spec#224). An unverified + claim still shows — but must not outrank a verified one when the rest is equal.""" + cands, _ = rank_candidates( + [_p("org/verified", protocol_verified=True), _p("org/declared", protocol_verified=False)], + abb="ABB.03", intent_verbs=["retrieve", "evaluate"], + min_trust=0.45, access_grader=GRANT_ALL, + ) + assert [c.repo for c in cands] == ["org/verified", "org/declared"] + assert cands[0].score > cands[1].score + assert any("VERIFIED" in r for r in cands[0].match_reason) + assert any("unverified" in r for r in cands[1].match_reason) + + +def test_every_candidate_carries_at_least_one_match_reason() -> None: + """A candidate with an unreadable reason is a candidate nobody should route to.""" + cands, _ = rank_candidates( + [_p("org/a")], abb="ABB.03", intent_verbs=["retrieve"], + min_trust=0.45, access_grader=GRANT_ALL, + ) + assert cands[0].match_reason + for r in cands[0].match_reason: + assert len(r) > 10 + + +def test_candidates_are_sorted_descending_by_score() -> None: + cands, _ = rank_candidates( + [ + _p("org/low", trust_score=0.50, catalog_verbs=("retrieve",)), + _p("org/high", trust_score=0.95, catalog_verbs=("retrieve", "evaluate")), + _p("org/mid", trust_score=0.70, catalog_verbs=("retrieve", "evaluate")), + ], + abb="ABB.03", intent_verbs=["retrieve", "evaluate"], + min_trust=0.45, access_grader=GRANT_ALL, + ) + scores = [c.score for c in cands] + assert scores == sorted(scores, reverse=True) + assert cands[0].repo == "org/high" + + +# ── the route decision ───────────────────────────────────────────────────────── + +def test_auto_route_when_every_precondition_holds() -> None: + s = build_shortlist( + [ + _p("org/strong", trust_score=0.95, catalog_verbs=("retrieve", "evaluate")), + _p("org/weak", trust_score=0.50, catalog_verbs=("retrieve",)), + ], + abb="ABB.03", intent_verbs=["retrieve", "evaluate"], + min_trust=0.45, access_grader=GRANT_ALL, + ) + assert s.decision == "auto-route" + assert s.chosen_index == 0 + assert "clear of the runner-up" in s.decision_reason + + +def test_narrow_gap_forces_user_pick_and_SAYS_the_gap() -> None: + """'Not confident enough' is not a reason a user can act on. 'The runner-up scored within + 0.04' is.""" + s = build_shortlist( + [_p("org/a", trust_score=0.80), _p("org/b", trust_score=0.79)], + abb="ABB.03", intent_verbs=["retrieve", "evaluate"], + min_trust=0.45, access_grader=GRANT_ALL, + ) + assert s.decision == "user-pick" + assert "top-2 gap" in s.decision_reason + assert "threshold" in s.decision_reason + + +def test_unconfirmed_counter_test_blocks_auto_route_and_names_it() -> None: + """Per Noetica#570's counter-test gate.""" + s = build_shortlist( + [ + _p("org/strong", trust_score=0.95, counter_test_status="available", + catalog_verbs=("retrieve", "evaluate")), + _p("org/weak", trust_score=0.50, catalog_verbs=("retrieve",)), + ], + abb="ABB.03", intent_verbs=["retrieve", "evaluate"], + min_trust=0.45, access_grader=GRANT_ALL, + ) + assert s.decision == "user-pick" + assert "counter-test" in s.decision_reason + assert "available" in s.decision_reason + + +def test_requires_consent_top_blocks_auto_route() -> None: + """A user must see and act on the remediation — Michael cannot silently route into a + resource the user has not been granted.""" + grader = lambda repo: (("requires-consent", "missing billing-reader", "https://c/x") + if repo == "org/strong" else ("granted", "ok", None)) + s = build_shortlist( + [ + _p("org/strong", trust_score=0.95, catalog_verbs=("retrieve", "evaluate")), + _p("org/weak", trust_score=0.50, catalog_verbs=("retrieve",)), + ], + abb="ABB.03", intent_verbs=["retrieve", "evaluate"], + min_trust=0.45, access_grader=grader, + ) + assert s.decision == "user-pick" + assert "access is 'requires-consent'" in s.decision_reason + + +def test_single_candidate_is_user_pick_not_auto_route() -> None: + """With nothing to compare against, the gap is undefined and 'auto' would be the caller's + default rather than a measured decision.""" + s = build_shortlist( + [_p("org/only")], abb="ABB.03", intent_verbs=["retrieve"], + min_trust=0.45, access_grader=GRANT_ALL, + ) + assert s.decision == "user-pick" + assert "only one candidate" in s.decision_reason + + +def test_no_survivors_is_abstain_with_an_empty_reason() -> None: + """Empties are signal. An empty shortlist WITH a reason is a valid answer; without one it + is indistinguishable from a broken walk.""" + s = build_shortlist( + [_p("org/a", trust_score=0.01)], abb="ABB.03", intent_verbs=["retrieve"], + min_trust=0.45, access_grader=GRANT_ALL, + ) + assert s.decision == "abstain" + assert s.candidates == [] + assert s.empty_reason and "below floor" in s.empty_reason + + +def test_multiple_blockers_are_ALL_reported_not_just_the_first() -> None: + """A caller fixing one blocker should not discover the next only on the retry.""" + s = build_shortlist( + [ + _p("org/a", trust_score=0.80, counter_test_status="unavailable"), + _p("org/b", trust_score=0.79, counter_test_status="confirmed"), + ], + abb="ABB.03", intent_verbs=["retrieve", "evaluate"], + min_trust=0.45, access_grader=GRANT_ALL, + ) + assert s.decision == "user-pick" + assert "top-2 gap" in s.decision_reason + assert "counter-test" in s.decision_reason + + +def test_threshold_matches_prophet_mesh() -> None: + """Pinned in both repos. Two services that both believe they gate at the same threshold, + gating differently, is exactly the drift the shared-vector discipline exists to catch.""" + assert AUTO_ROUTE_GAP_THRESHOLD == 0.15 + + +# ── emitted shape conforms to the prophet-mesh schema ────────────────────────── + +def test_emitted_json_matches_the_SolutionShortlist_shape() -> None: + s = build_shortlist( + [ + _p("org/strong", trust_score=0.95, catalog_verbs=("retrieve", "evaluate")), + _p("org/weak", trust_score=0.50, catalog_verbs=("retrieve",)), + ], + abb="ABB.03", intent_verbs=["retrieve", "evaluate"], + min_trust=0.45, access_grader=GRANT_ALL, + ) + j = s.to_json(abb="ABB.03", mar_digest="sha256:" + "a" * 64, catalog_digest="sha256:" + "b" * 64) + assert j["kind"] == "SolutionShortlist" + assert j["abbRequirement"] == "ABB.03" + assert j["autoRouteVerdict"]["decision"] == "auto-route" + assert j["autoRouteVerdict"]["chosenIndex"] == 0 + assert j["derivedFrom"]["marDigest"].startswith("sha256:") + for c in j["shortlist"]: + assert c["matchReason"], "every candidate must carry a reason in the emitted shape" + assert 0 <= c["score"] <= 1 + assert c["counterTestStatus"] in {"confirmed", "available", "unavailable"} + assert c["accessDecision"]["grade"] in {"granted", "requires-consent", "denied"} + + +def test_empty_shortlist_emits_emptyReason_and_no_autoRouteVerdict() -> None: + s = build_shortlist( + [], abb="ABB.03", intent_verbs=["retrieve"], min_trust=0.45, access_grader=GRANT_ALL, + ) + j = s.to_json(abb="ABB.03", mar_digest="sha256:" + "a" * 64, catalog_digest="sha256:" + "b" * 64) + assert j["shortlist"] == [] + assert "emptyReason" in j + assert "autoRouteVerdict" not in j, "an empty shortlist has no route to verdict on" + + +def test_canonical_json_keeps_non_ascii_raw() -> None: + """Matches lawful-verdict and the TypeScript canonicaliser. The default would escape + non-ASCII and every digest over accented content would diverge between languages.""" + assert canonical_json({"k": "café"}) == '{"k":"café"}' + assert "\\u" not in canonical_json({"k": "中文 🔒"}) + + +# ── the Eve Smith scenario, end to end ───────────────────────────────────────── + +def test_eve_smith_cross_selling_scenario() -> None: + """The exact flow from Zurich's E-RDA2 deck, run through our stack. + + Eve asks for a cross-selling report over the retail segment, including claims and billing + data. Two participants claim ABB.03 (DATABASE). She has analyst but not billing-reader. + + What their Damian does: discovers the access gap mid-conversation and offers a prefilled + link. What ours additionally does: refuses to auto-route BECAUSE of the gap, states the + reason, and keeps the runner-up visible so Eve can choose the non-billing option instead + of waiting on an approval she may not need. + """ + def grader(repo: str): + if repo == "org/billing-warehouse": + return ("requires-consent", "Eve lacks billing-reader on LOCAL BILLING", + "https://consent.socioprophet.io/request?subject=eve.smith" + "&resource=local-billing&roles=billing-reader&exp=1800003600&sig=deadbeef") + return ("granted", "market and product data are open to analyst", None) + + participants = [ + _p("org/billing-warehouse", trust_score=0.92, + catalog_verbs=("retrieve", "evaluate", "transform")), + _p("org/market-warehouse", trust_score=0.78, + catalog_verbs=("retrieve", "evaluate")), + ] + + s = build_shortlist( + participants, abb="ABB.03", + intent_verbs=["retrieve", "evaluate", "transform"], + min_trust=0.45, access_grader=grader, + ) + + # Both are shown — the consent-gated one is not hidden. + assert len(s.candidates) == 2 + top = s.candidates[0] + assert top.repo == "org/billing-warehouse" + assert top.access_grade == "requires-consent" + assert top.remediation_url and "billing-reader" in top.remediation_url + + # But it does NOT auto-route, and the reason names the access gap specifically. + assert s.decision == "user-pick" + assert "requires-consent" in s.decision_reason + + # The runner-up is fully usable right now — Eve has a choice, not just a wait. + runner_up = s.candidates[1] + assert runner_up.repo == "org/market-warehouse" + assert runner_up.access_grade == "granted" + + # And the emitted envelope carries everything a UI needs to render both options. + j = s.to_json(abb="ABB.03", mar_digest="sha256:" + "c" * 64, catalog_digest="sha256:" + "d" * 64) + assert j["autoRouteVerdict"]["decision"] == "user-pick" + assert j["shortlist"][0]["accessDecision"]["remediation"]["expectedReturn"] == "ArtifactConsentRecord" + assert j["shortlist"][1]["accessDecision"]["grade"] == "granted"