diff --git a/corpus/README.md b/corpus/README.md index 202aa37..bde8c22 100644 --- a/corpus/README.md +++ b/corpus/README.md @@ -106,3 +106,21 @@ else to increment. a match. The prior scoring series for these seven rows stays in git history, never overwritten; every future run against them scores under this new expect. + +## Persisted family-confirmation link (CHAOS-5722) + +An `any_of` SERVE branch's family-confirmation link (previously always +`unscored`/`family_confirmation_unavailable` — see CHAOS-5620 above) can now +reach `agree` from the served result's PERSISTED semantic state (acr's M2 +`semantic_state` column), never the wire, which carries only the +`semantic_reading` unavailable/absent/unreadable disclosure (D49). The +runner supplies a `persisted_semantic_state(result_id) -> dict | None` +adapter to `score`/`score_branch`/`build_verdict`; `semantic_verdict.py` +never opens a connection of its own, and the two absent/unreadable reasons +it reports reuse D49's own closed tokens +(`expect_schema.SEMANTIC_STATE_ABSENT`/`SEMANTIC_STATE_UNREADABLE`), read +from the synced contract schema, never hand-copied. Bumped +`POLICY_VERSION` (`any_of-fail-closed-v2-persisted-family-link`) — a +rescore under it is not comparable to a `v1` run without saying so. +Omitting the adapter (every call site that predates this ticket) scores +exactly as if it found no persisted row. diff --git a/corpus/expect_schema.py b/corpus/expect_schema.py index b436db1..e67cbcc 100644 --- a/corpus/expect_schema.py +++ b/corpus/expect_schema.py @@ -48,6 +48,31 @@ def _read_question_family_vocabulary(): FAMILIES = _read_question_family_vocabulary() + +def _read_semantic_reading_reasons(): + """The two closed reasons a stored result's semantic-reading disclosure + can carry (D49, CHAOS-5672, acr + internal/contracts/v1/context_fabric_semantic_reading.go's + `ContextFabricSemanticReadingReason` vocabulary), read directly from the + pinned, synced contract schema's own `semantic_reading.reason` enum -- + not hand-copied, so CHAOS-5722's persisted-state absent/unreadable + verdict reasons can never spell these two tokens differently from the + engine's own wire disclosure. Fails loudly at import time (never a + silent guess) if the synced schema's vocabulary ever stops being exactly + these two members.""" + doc = json.loads((_SCHEMAS_DIR / "context_fabric_investigation_result.v1.schema.json").read_text()) + enum = frozenset(doc["properties"]["semantic_reading"]["properties"]["reason"]["enum"]) + absent, unreadable = "semantic_state_absent", "semantic_state_unreadable" + if enum != {absent, unreadable}: + raise RuntimeError( + "context_fabric_investigation_result.v1.schema.json's semantic_reading.reason " + f"vocabulary changed to {sorted(enum)!r} -- update expect_schema.py's mirror" + ) + return absent, unreadable + + +SEMANTIC_STATE_ABSENT, SEMANTIC_STATE_UNREADABLE = _read_semantic_reading_reasons() + # Bump on any change to branch shape, vocabulary, or validation rule below -- # a published semantic verdict record carries this so a rescore under a # changed schema is never silently compared to one scored under the old diff --git a/corpus/semantic_verdict.py b/corpus/semantic_verdict.py index 891da28..3894ef0 100644 --- a/corpus/semantic_verdict.py +++ b/corpus/semantic_verdict.py @@ -52,6 +52,18 @@ unscored/unverifiable counts on every published verdict record, so a rescore under a changed policy is never mistaken for one under the old policy (see corpus/README.md). + * CHAOS-5722: an `any_of` SERVE branch's family-confirmation link, read + from the PERSISTED semantic state of the served result (acr's M2 + `semantic_state` column -- internal/contextfabric/semantic_state.go), + not the wire. The wire carries only the `semantic_reading` disclosure + (unavailable/absent/unreadable, D49) -- never the accepted reading + itself -- so this module never opens a connection of its own: the + caller (the corpus runner) injects a `persisted_semantic_state(result_id) + -> dict | None` adapter, the same injection shape `legacy_score` already + uses, and every reason this function returns for it reuses D49's own two + closed tokens (`expect_schema.SEMANTIC_STATE_ABSENT`/ + `SEMANTIC_STATE_UNREADABLE`) so a harness report can never spell "no + reading" differently from the engine's own disclosure. """ import expect_schema as S @@ -80,7 +92,14 @@ # oracle agreement suffices without a caller-confirmation receipt") is a # product/data decision (see the design of record) and gets its own version # string, never a silent redefinition of this one. -POLICY_VERSION = "any_of-fail-closed-v1" +# +# Bumped for CHAOS-5722: a serve branch's family-confirmation link is now +# read from the served result's PERSISTED semantic state (when the caller +# supplies the `persisted_semantic_state` adapter) instead of being +# uniformly `unscored`/`family_confirmation_unavailable` -- a rescore under +# this version is not comparable to one under `any_of-fail-closed-v1` +# without saying so (see corpus/README.md). +POLICY_VERSION = "any_of-fail-closed-v2-persisted-family-link" # Mirrors acr scripts/corpus/contract.py:63 (`is_success_status`) and its # `SERVED_STATUSES = frozenset({200})`. The producer's contract, not @@ -102,6 +121,39 @@ class ValidatorUnavailable(Exception): (which means the DATA was bad, not the tooling).""" +class PersistedSemanticStateUnreadable(Exception): + """CHAOS-5722: raised by a caller-supplied `persisted_semantic_state` + adapter when the trial store carries a row for the given result id but + it cannot be trusted (fails to decode, exceeds a bound, or otherwise + cannot be read) -- distinct from the adapter returning `None`, which + means no row exists at all. Mirrors acr's own absent-vs-unreadable + distinction for a stored semantic reading (D49, + internal/contracts/v1/context_fabric_semantic_reading.go). An adapter + that cannot tell the two apart should prefer raising this over + returning `None` -- "unreadable" is strictly more informative and this + module reports it under its own D49 reason + (`expect_schema.SEMANTIC_STATE_UNREADABLE`), never conflated with + `SEMANTIC_STATE_ABSENT`. + """ + + +# acr internal/contextfabric/semantic_state.go: SemanticStateFormatVersion. +# Never on the wire (see this module's docstring), so unlike FAMILIES/ +# SEMANTIC_STATE_ABSENT/SEMANTIC_STATE_UNREADABLE above there is no synced +# schema file to read this from -- mirrored here by hand, the same +# discipline _SERVED_HTTP_STATUSES below already uses for an acr constant +# this repo cannot import. A stored snapshot naming any other format is +# unavailable to this scorer, exactly as it is to acr's own reader. +_SEMANTIC_STATE_FORMAT_VERSION = "semantic-state.v1" + +# acr internal/contextfabric/frame_gate.go: FrameGateOutcome's accepted +# member, FrameGatePassed. Hand-mirrored for the same reason as +# _SEMANTIC_STATE_FORMAT_VERSION above -- FrameGateOutcome never reaches the +# wire, so there is no synced contract file this module could read it from +# instead. +_GATE_OUTCOME_PASSED = "passed" + + def _validate_shape(schema_def, payload): """Validate `payload` against `_COMMON_SCHEMA#/$defs/` and return its `ordered_ascending` fact (see schema_shim.validate_via_schema @@ -328,7 +380,59 @@ def audit_window_exchange(attempts): return {**out, "reason": "unreadable_exchange"} -def score_branch(row, branch, bucket, status, final, audit, legacy_score, **identity): +def _score_persisted_family_confirmation(observed_family, final, persisted_semantic_state): + """CHAOS-5722: the terminal step of a serve branch's family-confirmation + link, once the observed family matches the declared one and the + window-binding audit is clean (score_branch's caller already checked + both). Never called for a scalar row, a refuse branch, or a branch that + already failed an earlier check -- see score_branch. + + `persisted_semantic_state` is `None` when the caller wired in no + adapter at all (every legacy `build_verdict`/`score` call site that + predates this ticket) -- treated exactly like an adapter that found no + row: `unscored`/`SEMANTIC_STATE_ABSENT`, never a crash and never a + promotion to `agree`. + """ + detail = {} + if persisted_semantic_state is None: + return "unscored", S.SEMANTIC_STATE_ABSENT, detail + result_id = final.get("result_id") if isinstance(final, dict) else None + if not isinstance(result_id, str) or not result_id: + return "unscored", S.SEMANTIC_STATE_ABSENT, detail + detail["result_id"] = result_id + try: + persisted = persisted_semantic_state(result_id) + except PersistedSemanticStateUnreadable: + return "unscored", S.SEMANTIC_STATE_UNREADABLE, detail + if persisted is None: + return "unscored", S.SEMANTIC_STATE_ABSENT, detail + if not isinstance(persisted, dict): + return "unscored", S.SEMANTIC_STATE_UNREADABLE, detail + format_version = persisted.get("format_version") + detail["persisted_format_version"] = format_version + if format_version != _SEMANTIC_STATE_FORMAT_VERSION: + return "unscored", "semantic_state_version_unknown", detail + persisted_family = persisted.get("family") + validation = persisted.get("validation") + gate_outcome = validation.get("gate_outcome") if isinstance(validation, dict) else None + if not isinstance(persisted_family, str) or not isinstance(validation, dict) or not isinstance(gate_outcome, str): + # A row at a format_version this scorer knows must carry these + # fields (they are required, non-nullable, in that format) -- one + # missing or mistyped is the row failing to be what its own declared + # format promises, i.e. unreadable, never a mismatch verdict against + # values that were never really there. + return "unscored", S.SEMANTIC_STATE_UNREADABLE, detail + detail["persisted_family"] = persisted_family + if persisted_family != observed_family: + return "disagree", "persisted_family_mismatch", detail + detail["persisted_gate_outcome"] = gate_outcome + if gate_outcome != _GATE_OUTCOME_PASSED: + return "unscored", "gate_outcome_not_accepted", detail + return "agree", "family_confirmed", detail + + +def score_branch(row, branch, bucket, status, final, audit, legacy_score, persisted_semantic_state=None, + **identity): """Score one `any_of` alternative. `legacy_score(scalar_row, bucket, status, **identity) -> (verdict, reason)` @@ -336,43 +440,82 @@ def score_branch(row, branch, bucket, status, final, audit, legacy_score, **iden production; a test double in ask-dev's own tests -- see test_semantic_verdict.py). This function never grants `agree` for a serve branch on outcome alone: it additionally requires the observed - family to match the declared one, and then requires a verified, - unmistaken family-confirmation link before calling it more than - `unscored` -- which the current wire cannot supply (see - audit_window_exchange), so a serve branch's ceiling today is - `unscored`/`family_confirmation_unavailable`, never `agree`. + family to match the declared one, the window-binding audit to have found + NO POSITIVE MISTAKE (`!= "mismatch"` -- see the next paragraph for why + this is not the same as "verified"), and (CHAOS-5722) the + family-confirmation link from the served result's PERSISTED semantic + state (`persisted_semantic_state`, injected the same way `legacy_score` + is -- this function never opens a connection of its own) -- see + `_score_persisted_family_confirmation`. Any step short of a complete, + matching link stays `unscored` under one of D49's two closed reasons; a + positive family mismatch (declared, or persisted) or a window-binding + mismatch is `disagree`, never `unscored`. + + WHY `window_binding != "mismatch"`, NOT `== "verified"`. The window + audit (`audit_window_exchange`) and the persisted-family link below are + two INDEPENDENT sources of evidence about two DIFFERENT questions: the + audit asks "was a window-clarification receipt correctly offered and + applied", the persisted link asks "does the engine's own record of what + it served and validated match what was declared". A row that never went + through a window-clarification exchange at all (an ordinary single-turn + serve -- the audit's own default, `window_binding="unknown"`, + `reason="unsupported_exchange"`) has NOTHING for the first question to + confirm or deny; that says nothing about the second question, which the + persisted link answers on its own, from the engine's own stored state, + not from the corpus harness's capture of a conversation. Only a + POSITIVE, PROVEN mistake in the window mechanics (`"mismatch"` -- a + receipt applied to the wrong offer, a conflicting confirmation, ...) + contradicts what a served answer's family/gate says; an inconclusive or + absent audit does not, and must not silently downgrade a real, + independently-confirmed persisted link to `unscored`. See + `test_persisted_link_promotes_a_single_turn_serve_with_no_window_exchange_at_all` + for the executed proof, and the ticket's own rescore evidence (a real + single-turn `any_of` row, `window_binding="unknown"`, correctly reaching + `agree` from its persisted state). + + Returns `(verdict, reason, detail)` -- `detail` is `{}` unless the + persisted-state check actually ran and read something, in which case it + carries exactly the fields this function consulted (`result_id`, + `persisted_format_version`, `persisted_family`, + `persisted_gate_outcome`), so a published record can show why a serve + branch won or did not without a reader needing to re-run the query. """ scalar = {**row, "expect": branch["outcome"], "basis": branch.get("basis")} verdict, why = legacy_score(scalar, bucket, status, **identity) if branch["outcome"] != S.SERVE or verdict not in {"agree", "agree_weak"}: - return verdict, why + return verdict, why, {} # `final` is evidence, not a guaranteed shape: a row with no successful # result at all (error/no_match/turns-exhausted) legitimately has no # `final` payload to read a family from. Missing or malformed evidence # is `family_unavailable`, never an exception. observed = (final.get("answer_plan") or {}).get("family") if isinstance(final, dict) else None if observed is None: - return "unscored", "family_unavailable" + return "unscored", "family_unavailable", {} if observed != branch["answer"]["family"]: - return "disagree", "declared_family_mismatch" + return "disagree", "declared_family_mismatch", {} if audit["window_binding"] == "mismatch": - return "disagree", audit["reason"] + return "disagree", audit["reason"], {} if audit["family_relation"] == "changed": - return "unscored", "unratified_family_change" - return "unscored", "family_confirmation_unavailable" + return "unscored", "unratified_family_change", {} + return _score_persisted_family_confirmation(observed, final, persisted_semantic_state) _RANK = {"agree": 3, "agree_weak": 2, "unscored": 1, "disagree": 0} -def score(row, bucket, status, final, audit, legacy_score, **identity): +def score(row, bucket, status, final, audit, legacy_score, persisted_semantic_state=None, **identity): """(verdict, reason, branch_results) for one row/rep. `branch_results` is `[]` for a scalar/absent declaration (legacy_score's own verdict is authoritative and unchanged) and the full per-branch list - for an `any_of` declaration, in declaration order -- every branch result - is retained, never only the winner, so a report can show why an - undecidable alternative did not manufacture a pass. + of `(verdict, reason, detail)` triples for an `any_of` declaration, in + declaration order -- every branch result is retained, never only the + winner, so a report can show why an undecidable or losing alternative + did not manufacture a pass. `persisted_semantic_state` (CHAOS-5722) is + forwarded to `score_branch` unchanged; it is never consulted for a + scalar/absent declaration (this function never calls score_branch at + all in that case) or for a refuse/decline/clarify branch (score_branch + returns before reaching it -- see score_branch). """ ok, reason, choices = S.parse_expect(row.get("expect") if isinstance(row, dict) else None) if not ok: @@ -386,13 +529,16 @@ def score(row, bucket, status, final, audit, legacy_score, **identity): if choices is None: verdict, why = legacy_score(row, bucket, status, **identity) return verdict, why, [] - results = [score_branch(row, b, bucket, status, final, audit, legacy_score, **identity) for b in choices] + results = [ + score_branch(row, b, bucket, status, final, audit, legacy_score, persisted_semantic_state, **identity) + for b in choices + ] best = max(results, key=lambda r: _RANK[r[0]]) return best[0], best[1], results def build_verdict(row, bucket, status, final, audit, legacy_score, corpus_version, - legacy_scorer_version, **identity): + legacy_scorer_version, persisted_semantic_state=None, **identity): """The published, versioned semantic-verdict record for one row/rep. Carries SCORER_VERSION/POLICY_VERSION/`corpus_version` explicitly (a @@ -411,8 +557,15 @@ def build_verdict(row, bucket, status, final, audit, legacy_score, corpus_versio name its own scorer's identity (e.g. acr's git sha, or a test double's own label) so a rescore under a changed `legacy_score` is never mistaken for a rescore under the same one. + + `persisted_semantic_state` (CHAOS-5722, optional) is the same adapter + `score`/`score_branch` take; omitting it (every call site that predates + this ticket) scores exactly as if the adapter found no persisted row for + every serve branch it would otherwise have consulted -- see + `_score_persisted_family_confirmation`. """ - verdict, reason, branch_results = score(row, bucket, status, final, audit, legacy_score, **identity) + verdict, reason, branch_results = score(row, bucket, status, final, audit, legacy_score, + persisted_semantic_state, **identity) return { "scorer_version": SCORER_VERSION, "policy_version": POLICY_VERSION, @@ -425,7 +578,7 @@ def build_verdict(row, bucket, status, final, audit, legacy_score, corpus_versio "reason": reason, "unscored": verdict == "unscored", "branch_results": [ - {"verdict": v, "reason": r} for v, r in branch_results + {"verdict": v, "reason": r, **detail} for v, r, detail in branch_results ], "family_relation": audit.get("family_relation", "unknown"), "window_binding": audit.get("window_binding", "unknown"), diff --git a/corpus/test_semantic_verdict.py b/corpus/test_semantic_verdict.py index 9ba544b..38cc0f5 100644 --- a/corpus/test_semantic_verdict.py +++ b/corpus/test_semantic_verdict.py @@ -117,7 +117,12 @@ def base_exchange(final_family="discovered_cohort_ranking", first_family="discov return [t1, t2] -def test_family_match_is_not_confirmation(): +def test_family_match_is_not_confirmation_without_a_persisted_link(): + # CHAOS-5722: matching family + a verified window receipt still does + # NOT manufacture confirmation on their own -- only a caller-supplied + # `persisted_semantic_state` adapter can promote past `unscored`. No + # adapter is passed here (every call site that predates this ticket), + # so this must read exactly as if the adapter found no persisted row. attempts = base_exchange() final = attempts[-1]["response"]["result"] audit = SV.audit_window_exchange(attempts) @@ -125,8 +130,9 @@ def test_family_match_is_not_confirmation(): _require(audit["family_relation"] == "same", audit) row = declaration(serve()) verdict, reason, _ = SV.score(row, "served_with_data", "complete", final, audit, fake_legacy_score) - _require((verdict, reason) == ("unscored", "family_confirmation_unavailable"), - f"a verified window receipt + matching family must NOT manufacture confirmation, got {(verdict, reason)}") + _require((verdict, reason) == ("unscored", ES.SEMANTIC_STATE_ABSENT), + f"a verified window receipt + matching family must NOT manufacture confirmation " + f"without a persisted link, got {(verdict, reason)}") def test_family_mismatch_is_a_failure(): @@ -165,6 +171,298 @@ def test_retry_is_not_a_third_turn(): _require(audit["window_binding"] == "verified", f"a retry must collapse to two semantic turns: {audit}") +# --- CHAOS-5722: persisted semantic-state family-confirmation link ---------- + +def _verified_exchange_with_result_id(result_id="result-final"): + """A clean base_exchange() (verified window binding, same family) whose + `final` carries a `result_id` -- base_exchange()'s own `final` has none + (only the first turn's clarification does), so every test below that + needs the persisted-state adapter to actually run adds one.""" + attempts = base_exchange() + final = {**attempts[-1]["response"]["result"], "result_id": result_id} + attempts[-1] = {**attempts[-1], "response": {"result": final}} + audit = SV.audit_window_exchange(attempts) + _require(audit["window_binding"] == "verified", audit) + _require(audit["family_relation"] == "same", audit) + return final, audit + + +def _persisted_state(family="discovered_cohort_ranking", gate_outcome="passed", + format_version="semantic-state.v1"): + return {"format_version": format_version, "family": family, + "validation": {"gate_outcome": gate_outcome}} + + +class _CountingAdapter: + """A `persisted_semantic_state` test double that records every + `result_id` it was called with, so a test can assert it was NEVER + called for a scalar row or a refuse/decline/clarify branch (score_branch + must short-circuit before reaching it in both cases).""" + + def __init__(self, response): + self.response = response + self.calls = [] + + def __call__(self, result_id): + self.calls.append(result_id) + if isinstance(self.response, Exception): + raise self.response + return self.response + + +def test_persisted_link_promotes_serve_to_agree_and_outranks_a_wrongly_certified_refuse(): + # The exact defect this ticket fixes: a refuse branch on the same + # any_of row can score agree_weak (acr's real table matches it on a + # disclosed basis, independent of the actual served bucket) while a + # correct, persisted-confirmed serve is capped below it by _RANK. A + # complete persisted link must let the serve branch win. + def legacy_score_refuse_agrees_on_a_served_bucket(row, bucket, status, **identity): + if row.get("expect") == ES.SERVE: + return ("agree", "served") if bucket == "served_with_data" else ("disagree", "not served") + return "agree_weak", "matched_basis_on_served_bucket" + + final, audit = _verified_exchange_with_result_id() + row = declaration(serve(), {"outcome": "refuse"}) + adapter = _CountingAdapter(_persisted_state()) + verdict, reason, results = SV.score(row, "served_with_data", "complete", final, audit, + legacy_score_refuse_agrees_on_a_served_bucket, adapter) + _require(verdict == "agree", (verdict, reason, results)) + _require(reason == "family_confirmed", (verdict, reason)) + _require(results[0][0] == "agree", f"serve branch must reach agree: {results}") + _require(results[1][0] == "agree_weak", f"refuse branch must be untouched by this ticket: {results}") + _require(SV._RANK[results[0][0]] > SV._RANK[results[1][0]], + f"the persisted-confirmed serve must outrank the wrongly-certified refuse: {results}") + _require(adapter.calls == [final["result_id"]], adapter.calls) + + +def test_persisted_link_promotes_a_single_turn_serve_with_no_window_exchange_at_all(): + # The window audit and the persisted-family link are independent + # evidence about independent questions (see score_branch's own + # docstring) -- an ORDINARY single-turn serve never goes through a + # window-clarification exchange at all, so the audit's own default is + # `window_binding="unknown"`/`reason="unsupported_exchange"`, never + # "verified". That must not block a real, independently-confirmed + # persisted link: this is exactly the real-world shape a served, + # non-window-clarification any_of row has (confirmed against real + # trial-store data in the ticket's own rescore evidence). + audit = SV.audit_window_exchange([]) # no attempts at all -> no exchange, ever + _require(audit["window_binding"] == "unknown", audit) + final = {"answer_plan": {"family": "explicit_comparison"}, "result_id": "result-single-turn"} + row = declaration(serve("explicit_comparison"), {"outcome": "refuse"}) + adapter = _CountingAdapter(_persisted_state(family="explicit_comparison")) + verdict, reason, _ = SV.score(row, "served_with_data", "complete", final, audit, fake_legacy_score, adapter) + _require((verdict, reason) == ("agree", "family_confirmed"), (verdict, reason)) + _require(adapter.calls == [final["result_id"]], adapter.calls) + + +def test_persisted_link_promotes_a_serve_whose_window_exchange_audit_is_merely_inconclusive(): + # A DIFFERENT shape from "no exchange at all" above: a two-turn exchange + # WAS attempted but the audit could not positively verify it (a + # corpus-harness capture gap, not a proven mistake) -- window_binding + # still reads "unknown", never "mismatch". The persisted link is a + # separate, independently-grounded signal (the engine's own stored + # state for the SERVED result) and must not be downgraded just because + # the corpus's OWN capture of the surrounding conversation was + # unreadable. + attempts = base_exchange() + final = attempts[-1]["response"]["result"] + del final["confirmed_structure"] # -> audit_window_exchange: "missing_window_ack", still "unknown" + final = {**final, "result_id": "result-inconclusive-audit"} + audit = SV.audit_window_exchange([attempts[0], {**attempts[1], "response": {"result": final}}]) + _require(audit["window_binding"] == "unknown", audit) + _require(audit["reason"] == "missing_window_ack", audit) + row = declaration(serve()) + adapter = _CountingAdapter(_persisted_state()) + verdict, reason, _ = SV.score(row, "served_with_data", "complete", final, audit, fake_legacy_score, adapter) + _require((verdict, reason) == ("agree", "family_confirmed"), (verdict, reason)) + + +def test_persisted_state_absent_stays_unscored(): + final, audit = _verified_exchange_with_result_id() + row = declaration(serve()) + adapter = _CountingAdapter(None) # adapter ran, found no row + verdict, reason, _ = SV.score(row, "served_with_data", "complete", final, audit, fake_legacy_score, adapter) + _require((verdict, reason) == ("unscored", ES.SEMANTIC_STATE_ABSENT), (verdict, reason)) + _require(adapter.calls == [final["result_id"]], adapter.calls) + + +def test_persisted_state_unreadable_stays_unscored(): + final, audit = _verified_exchange_with_result_id() + row = declaration(serve()) + adapter = _CountingAdapter(SV.PersistedSemanticStateUnreadable("corrupt row")) + verdict, reason, _ = SV.score(row, "served_with_data", "complete", final, audit, fake_legacy_score, adapter) + _require((verdict, reason) == ("unscored", ES.SEMANTIC_STATE_UNREADABLE), (verdict, reason)) + + +def test_persisted_state_non_dict_return_is_unreadable(): + final, audit = _verified_exchange_with_result_id() + row = declaration(serve()) + for bad in ["not-a-dict", 12345, ["a", "list"]]: + adapter = _CountingAdapter(bad) + verdict, reason, _ = SV.score(row, "served_with_data", "complete", final, audit, fake_legacy_score, adapter) + _require((verdict, reason) == ("unscored", ES.SEMANTIC_STATE_UNREADABLE), (bad, verdict, reason)) + + +def test_persisted_state_malformed_shape_is_unreadable(): + # Domain table over the two shape-bearing fields (`family`, + # `validation.gate_outcome`) this scorer reads once format_version is + # known: {absent, null, wrong scalar/container type} each -- every cell + # must fail closed to `SEMANTIC_STATE_UNREADABLE`, never a crash and + # never a manufactured mismatch/agree against a value that was never + # really there. + final, audit = _verified_exchange_with_result_id() + row = declaration(serve()) + cells = [ + ("family_absent", {"format_version": "semantic-state.v1", "validation": {"gate_outcome": "passed"}}), + ("family_null", {"format_version": "semantic-state.v1", "family": None, + "validation": {"gate_outcome": "passed"}}), + ("family_wrong_type", {"format_version": "semantic-state.v1", "family": 42, + "validation": {"gate_outcome": "passed"}}), + ("validation_absent", {"format_version": "semantic-state.v1", "family": "discovered_cohort_ranking"}), + ("validation_null", {"format_version": "semantic-state.v1", "family": "discovered_cohort_ranking", + "validation": None}), + ("validation_wrong_type_string", {"format_version": "semantic-state.v1", + "family": "discovered_cohort_ranking", "validation": "not-a-dict"}), + ("validation_wrong_container_type_list", {"format_version": "semantic-state.v1", + "family": "discovered_cohort_ranking", "validation": []}), + ("gate_outcome_absent", {"format_version": "semantic-state.v1", "family": "discovered_cohort_ranking", + "validation": {}}), + ("gate_outcome_null", {"format_version": "semantic-state.v1", "family": "discovered_cohort_ranking", + "validation": {"gate_outcome": None}}), + ("gate_outcome_wrong_type", {"format_version": "semantic-state.v1", "family": "discovered_cohort_ranking", + "validation": {"gate_outcome": 200}}), + ] + for label, bad in cells: + adapter = _CountingAdapter(bad) + verdict, reason, _ = SV.score(row, "served_with_data", "complete", final, audit, fake_legacy_score, adapter) + _require((verdict, reason) == ("unscored", ES.SEMANTIC_STATE_UNREADABLE), (label, verdict, reason)) + + +def test_persisted_family_mismatch_is_disagree(): + final, audit = _verified_exchange_with_result_id() + row = declaration(serve()) + adapter = _CountingAdapter(_persisted_state(family="grouped_cohort_status")) + verdict, reason, _ = SV.score(row, "served_with_data", "complete", final, audit, fake_legacy_score, adapter) + _require((verdict, reason) == ("disagree", "persisted_family_mismatch"), (verdict, reason)) + + +def test_persisted_gate_outcome_not_accepted_is_not_agree(): + final, audit = _verified_exchange_with_result_id() + row = declaration(serve()) + for gate_outcome in ["refused_basis", "rejected_invalid", "not_proposed", ""]: + adapter = _CountingAdapter(_persisted_state(gate_outcome=gate_outcome)) + verdict, reason, _ = SV.score(row, "served_with_data", "complete", final, audit, fake_legacy_score, adapter) + _require(verdict != "agree", f"gate_outcome={gate_outcome!r} must never reach agree: {(verdict, reason)}") + _require((verdict, reason) == ("unscored", "gate_outcome_not_accepted"), + f"gate_outcome={gate_outcome!r}: {(verdict, reason)}") + + +def test_persisted_state_unknown_format_version_is_unscored(): + final, audit = _verified_exchange_with_result_id() + row = declaration(serve()) + adapter = _CountingAdapter(_persisted_state(format_version="semantic-state.v0-legacy")) + verdict, reason, _ = SV.score(row, "served_with_data", "complete", final, audit, fake_legacy_score, adapter) + _require((verdict, reason) == ("unscored", "semantic_state_version_unknown"), (verdict, reason)) + + +def test_malformed_result_id_is_absent_and_adapter_is_never_called(): + # Domain table over `final.result_id`: {absent, null, empty string, + # wrong scalar type} -- none of these may reach the adapter at all + # (there is nothing to look up), and every one must read as + # SEMANTIC_STATE_ABSENT, never a crash. + row = declaration(serve()) + for label, mutate in [ + ("absent", lambda f: f.pop("result_id")), + ("null", lambda f: f.__setitem__("result_id", None)), + ("empty_string", lambda f: f.__setitem__("result_id", "")), + ("wrong_type_int", lambda f: f.__setitem__("result_id", 12345)), + ]: + final, audit = _verified_exchange_with_result_id() + mutate(final) + adapter = _CountingAdapter(_persisted_state()) + verdict, reason, _ = SV.score(row, "served_with_data", "complete", final, audit, fake_legacy_score, adapter) + _require((verdict, reason) == ("unscored", ES.SEMANTIC_STATE_ABSENT), (label, verdict, reason)) + _require(adapter.calls == [], f"{label}: an adapter must never be called with no result_id: {adapter.calls}") + + +def test_adapter_is_never_called_for_a_scalar_row(): + adapter = _CountingAdapter(_persisted_state()) + row = {"id": "synthetic", "text": "Synthetic", "expect": "serve"} + SV.score(row, "served_with_data", "complete", {}, {}, fake_legacy_score, adapter) + _require(adapter.calls == [], f"a scalar row must never consult the persisted-state adapter: {adapter.calls}") + + +def test_adapter_is_never_called_for_a_refuse_branch(): + final, audit = _verified_exchange_with_result_id() + adapter = _CountingAdapter(_persisted_state()) + row = declaration({"outcome": "refuse"}, {"outcome": "decline"}, {"outcome": "clarify"}) + SV.score(row, "unserved", "complete", final, audit, fake_legacy_score, adapter) + _require(adapter.calls == [], + f"a refuse/decline/clarify branch must never consult the persisted-state adapter: {adapter.calls}") + + +def test_adapter_is_never_called_when_window_binding_mismatches_or_family_changed(): + # (window_binding mismatch, family_relation changed) both short-circuit + # score_branch BEFORE the persisted-state check -- see score_branch. + mismatched = base_exchange() + mismatched[1]["request"]["priorWindowReceipts"][0]["receipt_id"] = "winr_unoffered" + final_mismatch = {**mismatched[1]["response"]["result"], "result_id": "result-mismatch"} + mismatched[1] = {**mismatched[1], "response": {"result": final_mismatch}} + audit_mismatch = SV.audit_window_exchange(mismatched) + _require(audit_mismatch["window_binding"] == "mismatch", audit_mismatch) + adapter = _CountingAdapter(_persisted_state()) + row = declaration(serve()) + verdict, reason, _ = SV.score(row, "served_with_data", "complete", final_mismatch, audit_mismatch, + fake_legacy_score, adapter) + _require(verdict == "disagree", (verdict, reason)) + _require(adapter.calls == [], f"a window-binding mismatch must never reach the adapter: {adapter.calls}") + + changed = base_exchange(final_family="grouped_cohort_status", first_family="discovered_cohort_ranking") + final_changed = {**changed[1]["response"]["result"], "result_id": "result-changed"} + changed[1] = {**changed[1], "response": {"result": final_changed}} + audit_changed = SV.audit_window_exchange(changed) + _require(audit_changed["family_relation"] == "changed", audit_changed) + adapter2 = _CountingAdapter(_persisted_state(family="grouped_cohort_status")) + row2 = declaration(serve("grouped_cohort_status")) + verdict2, reason2, _ = SV.score(row2, "served_with_data", "complete", final_changed, audit_changed, + fake_legacy_score, adapter2) + _require((verdict2, reason2) == ("unscored", "unratified_family_change"), (verdict2, reason2)) + _require(adapter2.calls == [], f"an unratified family change must never reach the adapter: {adapter2.calls}") + + +def test_persisted_link_fields_are_published_in_branch_results(): + final, audit = _verified_exchange_with_result_id() + row = declaration(serve()) + record = SV.build_verdict(row, "served_with_data", "complete", final, audit, fake_legacy_score, + corpus_version="test-corpus-v0", legacy_scorer_version="fake-legacy-v1", + persisted_semantic_state=_CountingAdapter(_persisted_state())) + branch = record["branch_results"][0] + _require(branch["verdict"] == "agree", branch) + _require(branch["result_id"] == final["result_id"], branch) + _require(branch["persisted_format_version"] == "semantic-state.v1", branch) + _require(branch["persisted_family"] == "discovered_cohort_ranking", branch) + _require(branch["persisted_gate_outcome"] == "passed", branch) + + +def test_build_verdict_with_no_persisted_state_argument_is_backward_compatible(): + # Every call site that predates CHAOS-5722 (acr's semantic_verdict_bridge.py + # included) calls build_verdict()/score() with no `persisted_semantic_state` + # argument at all -- this must keep working exactly as documented (absent + # adapter == adapter that found no row), never a TypeError. + final, audit = _verified_exchange_with_result_id() + row = declaration(serve()) + record = SV.build_verdict(row, "served_with_data", "complete", final, audit, fake_legacy_score, + corpus_version="test-corpus-v0", legacy_scorer_version="fake-legacy-v1") + _require(record["verdict"] == "unscored", record) + _require(record["reason"] == ES.SEMANTIC_STATE_ABSENT, record) + + +def test_rank_ordering_is_unchanged(): + # Regression pin: CHAOS-5722 adds a new terminal outcome (`agree` from a + # persisted link) but must never touch the ordinal ranking itself. + _require(SV._RANK == {"agree": 3, "agree_weak": 2, "unscored": 1, "disagree": 0}, SV._RANK) + + _MUTATIONS = [ ("receipt", "mismatch"), ("parent", "mismatch"),