test-driver/tests/test_crystallization.py

"""H-003: a stable agentic realization becomes deterministic code.

What must be preserved is the judgment, not the path. These tests check that the
descendant asserts everything its ancestor did, agrees with it wherever both can
run, and costs less — and that it fails loudly rather than searching when the
surface moves out from under it.
"""

from __future__ import annotations

import json
import time

import pytest

from testdriver import Runner, Verdict
from testdriver.crystallization import (
    CrystallizedDriver, Trajectory, assess_stability, capture,
    generate_test_module,
)
from scenarios.alice_bob_carol import USE_CASE
from scenarios.browser_grant import build_agentic, build_crystallized, lab_server

RUNS = 4


def agentic_pack(*mutations: str) -> dict:
    with lab_server(*mutations) as (app, tokens, base_url):
        world, driver, observer, asset, oracle = build_agentic(app, tokens, base_url)
        return json.loads(Runner(world, driver, observer, oracle).run(asset).evidence.to_json())


@pytest.fixture(scope="module")
def stability():
    return assess_stability([agentic_pack() for _ in range(RUNS)])


@pytest.fixture(scope="module")
def frozen(stability):
    assert stability.stable
    return stability.trajectories


# --- stability ------------------------------------------------------------


def test_one_successful_run_is_not_enough_to_freeze():
    """One run proves the agent can find a way, not that the surface has settled."""
    report = assess_stability([agentic_pack()])
    assert not report.stable
    assert "at least" in report.reason


def test_a_moving_surface_is_not_frozen():
    """Different paths across runs means the surface is still moving."""
    packs = [agentic_pack(), agentic_pack("M24"), agentic_pack(), agentic_pack()]
    report = assess_stability(packs)
    assert not report.stable
    assert report.distinct_paths > 1


def test_a_settled_surface_is_frozen(stability):
    assert stability.stable
    assert stability.observations == RUNS
    assert stability.distinct_paths == 1


# --- the descendant -------------------------------------------------------


def run_descendant(frozen, *mutations: str):
    with lab_server(*mutations) as (app, tokens, base_url):
        world, driver, observer, asset, oracle = build_crystallized(
            app, tokens, base_url, frozen
        )
        return Runner(world, driver, observer, oracle).run(asset)


def test_the_descendant_runs_with_no_model_and_no_discovery(frozen):
    result = run_descendant(frozen)
    assert result.verdict is Verdict.PASS
    mechanics = [
        obs.data["mechanics"] for obs in result.evidence.observations
        if obs.kind == "realization"
    ][0]
    assert mechanics["runtime"] == "crystallized-driver"
    assert mechanics["metrics"]["candidates_considered"] == 0
    assert mechanics["metrics"]["tokens_in"] == 0


def test_the_descendant_preserves_the_ancestors_oracle_set(frozen):
    ancestor = agentic_pack()
    descendant = run_descendant(frozen)
    before = {v["assertion_id"] for v in ancestor["verdicts"]}
    after = {j.assertion_id for j in descendant.judgments}
    assert before <= after, f"crystallization dropped {before - after}"


@pytest.mark.parametrize("mutation", ["", "M15", "M17", "M16", "M13"])
def test_ancestor_and_descendant_agree(frozen, mutation):
    """Same verdicts on the same lab version. Disagreement falsifies H-003."""
    mutations = (mutation,) if mutation else ()
    ancestor = agentic_pack(*mutations)
    descendant = run_descendant(frozen, *mutations)
    before = {(v["assertion_id"], v["step_id"]): v["verdict"] for v in ancestor["verdicts"]}
    after = {(j.assertion_id, j.step_id): j.verdict.value for j in descendant.judgments}
    assert before == after


def test_the_descendant_still_catches_a_defect(frozen):
    """Crystallization must not trade rigour for speed."""
    result = run_descendant(frozen, "M17")
    assert result.verdict is Verdict.FAIL
    assert result.judgment("c-carol-denied").verdict is Verdict.FAIL


def test_the_descendant_fails_loudly_when_its_path_breaks(frozen):
    """A frozen driver must not fall back to searching.

    Falling back would quietly turn a T5 asset into a T1 one and hide the very
    signal crystallization exists to produce: the surface moved.

    M24 is the reversibility case from `INTENT.md`: the agentic ancestor recovers
    by reading the form, the frozen descendant cannot, and the right response is
    to thaw the asset rather than patch the generated file.
    """
    result = run_descendant(frozen, "M24")  # the grant endpoint is renamed
    raised = [
        obs.data["raised"] for obs in result.evidence.observations
        if obs.kind == "realization"
    ]
    assert any(raised), "the frozen path should have stopped working"


def test_lineage_is_retained(frozen):
    with lab_server() as (app, tokens, base_url):
        _, _, _, descendant, _ = build_crystallized(app, tokens, base_url, frozen)
    assert descendant.parent_id == "va-grant-via-browser"
    assert descendant.maturity == "T5"


# --- cost -----------------------------------------------------------------


def test_the_descendant_costs_less_to_execute(frozen):
    """H-003's economic criterion.

    With a scripted runtime the saving is only the page fetch, parse and scoring
    — real but small. See F-0007: the meaningful comparison needs a live model,
    and this number is a floor, not the answer.
    """
    def timed(fn, *args):
        started = time.perf_counter()
        fn(*args)
        return time.perf_counter() - started

    with lab_server() as (app, tokens, base_url):
        agentic = build_agentic(app, tokens, base_url)
        crystal = build_crystallized(app, tokens, base_url, frozen)
        agentic_ms = min(timed(lambda: Runner(*agentic[:3], agentic[4]).run(agentic[3]))
                         for _ in range(3))
        crystal_ms = min(timed(lambda: Runner(*crystal[:3], crystal[4]).run(crystal[3]))
                         for _ in range(3))
    assert crystal_ms < agentic_ms, (agentic_ms, crystal_ms)


# --- the generated artefact ----------------------------------------------


def test_the_generated_module_carries_its_lineage(frozen, stability):
    source = generate_test_module(
        trajectory=frozen[0],
        action_args={"subject_id": "bob", "permission": "READ"},
        ancestor_id="va-grant-via-browser", ancestor_maturity="T1",
        descendant_id="va-grant-crystallized", runs=stability.observations,
        sut_version="lab-0.2.0-baseline",
        claims=[c for c in USE_CASE.claims if c.id == "c-bob-reads"],
        claims_module="scenarios.alice_bob_carol",
    )
    assert "ancestor asset : va-grant-via-browser" in source
    assert f"frozen from    : {RUNS} identical realizations" in source
    assert "No model is involved" in source


def test_the_generated_module_imports_no_agentic_machinery():
    source = open("crystallized/test_grant_access.py").read()
    imports = [
        line for line in source.splitlines()
        if line.startswith(("import ", "from ")) and "urllib" not in line
    ]
    assert imports == [
        "from __future__ import annotations",
        "from scenarios.alice_bob_carol import _bob_can_read, "
        "_bob_cannot_write, _carol_cannot_read",
    ], imports


def test_the_generated_module_restates_no_assertions():
    """Assertions are imported, never paraphrased.

    A generated test that restates its claims creates a second, unverified
    statement of intent, and any drift between the two is silent.
    """
    source = open("crystallized/test_grant_access.py").read()
    assert "from scenarios.alice_bob_carol import" in source


def test_the_ancestor_recovers_where_the_descendant_cannot(frozen):
    """The reversibility case, both halves.

    This is what distinguishes crystallization from ordinary code generation: the
    agentic form is not discarded when the deterministic one is created, because
    it is the thing that can recover when the surface moves again.
    """
    ancestor = agentic_pack("M24")
    ancestor_raised = [
        obs["data"].get("raised") for obs in ancestor["observations"]
        if obs["kind"] == "realization"
    ]
    assert not any(ancestor_raised), "the agentic ancestor should have recovered"

    descendant = run_descendant(frozen, "M24")
    descendant_raised = [
        obs.data.get("raised") for obs in descendant.evidence.observations
        if obs.kind == "realization"
    ]
    assert any(descendant_raised), "the frozen descendant should have failed loudly"