test-driver/tests/test_classification.py

"""The classifier, measured against the labelled catalogue — and attacked.

`test_response_matrix` samples benign and hostile cases alike. The E-003 group
below does something different: it *tries to make the framework unsafe*. An
experiment that only samples cases chosen by the same person who wrote the
implementation cannot establish a safety property.
"""

from __future__ import annotations

import json

import pytest

from testdriver import Runner, Verdict
from testdriver.classification import (
    Classification, SAFE_TO_ACCEPT, classify,
)
from testdriver.provenance import InadmissibleProvenance, Provenance
from testdriver.intent import Claim
from lab.mutations import CATALOGUE
from scenarios.full_journey import build_journey, journey_lab_server
from tests.selfverification.checks import check_intent_independence


def pack_for(*mutations: str, observer_factory=None) -> dict:
    with journey_lab_server(*mutations) as (app, tokens, base_url):
        world, driver, observer, asset, oracle = build_journey(app, tokens, base_url)
        if observer_factory is not None:
            observer = observer_factory(observer)
        return json.loads(Runner(world, driver, observer, oracle).run(asset).evidence.to_json())


@pytest.fixture(scope="module")
def baseline() -> dict:
    return pack_for()


# --- the response matrix --------------------------------------------------

EXPECTED = {
    "M01": Classification.MECHANICAL_ADAPTATION,
    "M02": Classification.MECHANICAL_ADAPTATION,
    "M03": Classification.UNCHANGED,
    "M04": Classification.MECHANICAL_ADAPTATION,
    "M05": Classification.UNCHANGED,
    "M06": Classification.MECHANICAL_ADAPTATION,
    "M07": Classification.MECHANICAL_ADAPTATION,
    "M08": Classification.MECHANICAL_ADAPTATION,
    "M09": Classification.UNCHANGED,
    "M10": Classification.UNCHANGED,
    "M21": Classification.MECHANICAL_ADAPTATION,
    "M22": Classification.AMBIGUOUS,
    "M11": Classification.BEHAVIOUR_CHANGED,
    "M12": Classification.BEHAVIOUR_CHANGED,
    "M13": Classification.UNCHANGED,
    "M14": Classification.UNCHANGED,
    "M15": Classification.BEHAVIOUR_CHANGED,
    "M16": Classification.BEHAVIOUR_CHANGED,
    "M17": Classification.BEHAVIOUR_CHANGED,
    "M18": Classification.BEHAVIOUR_CHANGED,
    "M19": Classification.BEHAVIOUR_CHANGED,
    "M20": Classification.BEHAVIOUR_CHANGED,
    "M23": Classification.AMBIGUOUS,
    "M24": Classification.MECHANICAL_ADAPTATION,
}


@pytest.mark.parametrize("mutation_id", sorted(EXPECTED))
def test_response_matrix(baseline, mutation_id):
    assert classify(baseline, pack_for(mutation_id)).classification is EXPECTED[mutation_id]


def test_baseline_classifies_as_unchanged(baseline):
    assert classify(baseline, pack_for()).classification is Classification.UNCHANGED


# --- the metric that matters ----------------------------------------------


def test_false_adaptation_rate_is_zero(baseline):
    """The project's existential safety metric.

    Not "low". Zero. A non-zero result is a stop-and-redesign signal, because a
    framework that normalizes a real defect once has spent the credibility it
    exists to accumulate.
    """
    false_adaptations = [
        mutation.id for mutation in CATALOGUE
        if mutation.label == "DEFECT"
        and classify(baseline, pack_for(mutation.id)).safe_to_accept
    ]
    assert false_adaptations == []


def test_every_defect_is_escalated(baseline):
    for mutation in CATALOGUE:
        if mutation.label != "DEFECT":
            continue
        outcome = classify(baseline, pack_for(mutation.id))
        assert not outcome.safe_to_accept, f"{mutation.id}: {outcome.reason}"


def test_mechanical_changes_are_mostly_absorbed(baseline):
    """Adaptation has to be *useful*, not merely safe.

    A classifier that escalated everything would score a perfect False Adaptation
    Rate and be worthless. This asserts the other side of the trade.
    """
    mechanical = [m for m in CATALOGUE if m.label == "MECHANICAL"]
    absorbed = [
        m.id for m in mechanical
        if classify(baseline, pack_for(m.id)).safe_to_accept
    ]
    assert len(absorbed) >= len(mechanical) - 1


def test_no_classification_is_outside_the_defined_space(baseline):
    for mutation in CATALOGUE:
        outcome = classify(baseline, pack_for(mutation.id))
        assert isinstance(outcome.classification, Classification)
        assert outcome.reason


# --- E-003: attacking the boundary ---------------------------------------


def test_attack_surface_substitution(baseline):
    """M23: the control is gone from the UI; the API endpoint is still open.

    The most plausible concrete route to a false adaptation. A driver that
    'recovers' by calling the API has routed around a control, not adapted to a
    change — and would score a successful mechanical adaptation while an
    authorization control sat broken.
    """
    outcome = classify(baseline, pack_for("M23"))
    assert not outcome.safe_to_accept
    assert outcome.classification is Classification.AMBIGUOUS


def test_attack_concurrent_mechanical_and_defect(baseline):
    """Decision-table row 3. Coincidence is not exoneration.

    This is the case a naive self-healing tool gets wrong: it sees a surface
    change, explains the failure with it, and adapts.
    """
    for combination in (("M01", "M15"), ("M02", "M17"), ("M21", "M20")):
        outcome = classify(baseline, pack_for(*combination))
        assert outcome.classification is Classification.BEHAVIOUR_CHANGED
        assert not outcome.safe_to_accept
        assert "does not excuse it" in outcome.reason


def test_attack_evidence_starvation(baseline):
    """Cut the independent observation channel and the answer must be AMBIGUOUS.

    Never a pass, never an adaptation. Evidence starvation is the condition under
    which a framework is most tempted to fall back on the actor's own account of
    what happened.
    """
    class Starved:
        def __init__(self, real):
            self._real, self._calls, self.name = real, 0, real.name
            self.watches = real.watches

        def snapshot(self):
            self._calls += 1
            return self._real.snapshot() if self._calls < 2 else {}

    outcome = classify(baseline, pack_for(observer_factory=Starved))
    assert outcome.classification is Classification.AMBIGUOUS
    assert not outcome.safe_to_accept


def test_attack_provenance_laundering():
    """Intent derived from the implementation must not become a claim.

    Rejected at authoring time, and caught again in the record if it ever got
    past — belt and braces, because this one cannot be noticed by looking at
    behaviour.
    """
    with pytest.raises(InadmissibleProvenance):
        Claim("c-laundered", "whatever the system does",
              Provenance.AGENT_FROM_IMPLEMENTATION, lambda obs: True, after_step="s1")

    tampered = pack_for("M15")
    tampered["provenance_index"]["c-bob-revoked"] = "agent-from-implementation"
    assert check_intent_independence(tampered)


def test_safe_to_accept_is_a_closed_set():
    """Only two outcomes may proceed without a human. Widening this set is the
    single easiest way to destroy the safety property, so it is asserted."""
    assert SAFE_TO_ACCEPT == {
        Classification.UNCHANGED, Classification.MECHANICAL_ADAPTATION,
    }


# --- classifications that no lab mutation happens to produce ---------------
#
# Two outcomes were declared at T08 and exercised by nothing. Left that way they
# are decoration: code that has never run is code nobody has checked. Rather than
# delete meaningful outcomes or trust them untested, both are given a case.


def test_intent_change_is_detected_when_the_claim_set_moves(baseline):
    """`INTENT_CHANGED` is a fact about the recorded use case, not an inference.

    It fires because a human edited what is being asserted — which is why it is
    detectable at all, where `SEMANTIC_CHANGE` was not (F-0006).
    """
    import copy

    altered = copy.deepcopy(baseline)
    altered["provenance_index"]["c-newly-added-claim"] = "human"
    outcome = classify(baseline, altered)
    assert outcome.classification is Classification.INTENT_CHANGED
    assert not outcome.safe_to_accept


def test_realization_failure_is_distinguishable_from_ambiguity():
    """`REALIZATION_FAILED` says "we could not act"; `AMBIGUOUS` says "we do not know".

    Every lab mutation that breaks realization also strands a claim, so the
    catalogue only ever produces `AMBIGUOUS`. This builds the case the catalogue
    cannot: a step that fails while every assertion in the run still holds and
    none of them depended on it.

    Note that a run asserting *nothing at all* is `AMBIGUOUS`, not
    `REALIZATION_FAILED` — a use case with no claims cannot conclude anything,
    however well its steps ran.
    """
    from testdriver import (
        Actor, Cast, Invariant, Oracle, Runner, Scenario, SemanticAction,
        StateObserver, Step, UseCase, VerificationAsset, World,
    )
    from testdriver.agentic import DiscoveryRuntime
    from testdriver.browser import BrowserDriver
    from testdriver.observers import Watch
    from testdriver.provenance import Provenance
    from lab.mutations import ObservationChannel

    use_case = UseCase(
        "uc-audit-only", "Sharing leaves an ordered audit trail",
        "Alice shares R with Bob; the audit trail stays ordered.",
        Provenance.HUMAN,
        invariants=(Invariant(
            "i-audit-ordered", "The audit trail is append-only", Provenance.HUMAN,
            lambda obs: [e["sequence"] for e in obs["audit:R"]]
            == sorted(e["sequence"] for e in obs["audit:R"]),
        ),),
    )

    def run(*mutations):
        with journey_lab_server(*mutations) as (app, tokens, base_url):
            app.request(tokens["alice"], "create_resource",
                        resource_id="R", content="x")
            cast = Cast()
            cast.add(Actor("alice", "Alice", credentials={"token": tokens["alice"]}))
            scenario = Scenario(
                "sc-audit-only", use_case,
                watches=(Watch("bob", "R"),),
                steps=(Step("s1", "alice", SemanticAction(
                    "grant_access", {"subject_id": "bob", "permission": "READ"},
                    permitted_surfaces=frozenset({"browser"}),
                )),),
            )
            driver = BrowserDriver(base_url, tokens, DiscoveryRuntime(), "R")
            observer = StateObserver(ObservationChannel(app), scenario.watches)
            world = World("w-audit", app, app.version, cast=cast)
            return json.loads(
                Runner(world, driver, observer, Oracle())
                .run(VerificationAsset("va-audit-only", scenario))
                .evidence.to_json()
            )

    outcome = classify(run(), run("M23"))  # the control is gone from the UI
    assert outcome.classification is Classification.REALIZATION_FAILED
    assert not outcome.safe_to_accept


def test_no_classification_is_unreachable():
    """Every declared outcome must be produced somewhere in this suite.

    An outcome nothing can emit is the same kind of dead promise `SUSPICIOUS`
    was before T10 removed it.
    """
    exercised = set(EXPECTED.values()) | {
        Classification.INTENT_CHANGED, Classification.REALIZATION_FAILED,
    }
    assert exercised == set(Classification)