rune/tests/test_construction_pattern.py

"""test_construction_pattern.py — SCAFFOLD-TRANCHE-006 test plans §28-§32."""

import json
import os
import sys

import pytest

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

from rune import construction_decision as cd  # noqa: E402
from rune import construction_pattern as cp  # noqa: E402
from rune import structural_store as ss  # noqa: E402
from rune.construction_patterns_builtin import (  # noqa: E402
    BUILTIN_PATTERNS,
    RELEASE_DEPLOYMENT,
    get_pattern,
)

import construction_guards as cg  # noqa: E402


def _spine():
    spine = ss.new_doc()
    ss.add_entity(
        spine,
        id="module:deploy_stage",
        kind="module",
        identities={"path": "scripts/deploy_stage.py"},
        maturity="WIRED",
        evidence_refs=["f"],
    )
    ss.add_entity(spine, id="contract:monotonic_promotion", kind="contract", identities={}, evidence_refs=["f"])
    ss.add_relationship(
        spine,
        type="implements",
        from_id="module:deploy_stage",
        to_id="contract:monotonic_promotion",
        evidence_refs=["f"],
    )
    ss.add_entity(
        spine,
        id="module:deploy_qualify",
        kind="module",
        identities={"path": "scripts/deploy_qualify.py"},
        maturity="WIRED",
        evidence_refs=["f"],
    )
    ss.add_entity(spine, id="contract:fail_closed_deployment", kind="contract", identities={}, evidence_refs=["f"])
    ss.add_relationship(
        spine,
        type="implements",
        from_id="module:deploy_qualify",
        to_id="contract:fail_closed_deployment",
        evidence_refs=["f"],
    )
    return spine


def _need(**kw):
    d = dict(
        need_id="n6",
        objective="repair the deploy promotion ordering",
        origin="user",
        constraints={"contracts": ["contract:monotonic_promotion"]},
        affected_surface=["scripts/deploy_stage.py", "scripts/deploy_promotion.py"],
        evidence_refs=["x"],
        approval_context="covered",
    )
    d.update(kw)
    return cd.Need(**d)


def _surface(need):
    authorities = {"rune_spine": _spine()}
    return cd.resolve_affected_surface(need, authorities), authorities


# ── §28 pattern object ──────────────────────────────────────────────────────


def test_28a_roundtrip():
    d = RELEASE_DEPLOYMENT.to_dict()
    p2 = cp.ConstructionPattern.from_dict(d)
    assert p2.to_dict() == d


def test_28b_open_world_ext():
    p = cp.ConstructionPattern("p", "purpose", wibble="future-data")
    assert p.to_dict()["wibble"] == "future-data"
    assert cp.ConstructionPattern.from_dict(p.to_dict()).ext["wibble"] == "future-data"


def test_28c_revision_preserved_and_advances():
    p2 = RELEASE_DEPLOYMENT.revise(
        reason="new incident evidence", evidence=["new-incident"], changes={"maturity": "EXERCISED"}
    )
    assert p2.revision == RELEASE_DEPLOYMENT.revision + 1
    hist = p2.ext["revision_history"]
    assert hist[-1]["prior_revision"] == RELEASE_DEPLOYMENT.revision
    assert hist[-1]["evidence"] == ["new-incident"]
    # revision history survives a roundtrip (§17: old-rule traceability)
    assert cp.ConstructionPattern.from_dict(p2.to_dict()).ext["revision_history"] == hist
    # original untouched (no invisible mutation)
    assert RELEASE_DEPLOYMENT.maturity == "UNIT-PROVEN"


def test_28d_provenance_preserved():
    d = RELEASE_DEPLOYMENT.to_dict()
    # provenance entries are enriched "ref (why)" strings — check by ref
    assert any(e.startswith("maintenance:deploy-order-race") for e in d["provenance"])
    assert any(e.startswith("DEPLOY-MONOTONIC-001") for e in d["provenance"])
    assert cp.ConstructionPattern.from_dict(d).provenance == d["provenance"]


def test_28e_maturity_preserved():
    for p in BUILTIN_PATTERNS:
        assert p.maturity in ("CANDIDATE", "UNIT-PROVEN", "EXERCISED", "REPEATEDLY-PROVEN", "DEGRADED")
    # initial patterns stay appropriately immature (§15)
    assert all(p.maturity in ("CANDIDATE", "UNIT-PROVEN") for p in BUILTIN_PATTERNS)


def test_28f_obligation_validation():
    with pytest.raises(ValueError):
        cp.ConstructionPattern("p", "purpose", obligations=[{"id": "x"}])  # missing text/when


def test_28g_registry_revision_guard():
    assert get_pattern("release-deployment-change") is not None
    with pytest.raises(ValueError):
        get_pattern("release-deployment-change", revision=99)


# ── §29 applicability ───────────────────────────────────────────────────────


def test_29a_matching_evidence_applies():
    need = _need()
    surface, auth = _surface(need)
    app = cp.evaluate_applicability(RELEASE_DEPLOYMENT, need, surface, auth)
    assert app["verdict"] == cp.APPLIES
    assert app["evidence"]  # evidence-backed, not name-based


def test_29b_partial_evidence_partial():
    need = _need(objective="release notes wording", affected_surface=["docs/RELEASE.md"])
    surface, auth = _surface(need)
    app = cp.evaluate_applicability(RELEASE_DEPLOYMENT, need, surface, auth)
    assert app["verdict"] in (cp.PARTIAL, cp.DOES_NOT_APPLY)


def test_29c_contradictory_project_evidence_does_not_apply():
    """§3/§29C: project evidence overrides the pattern assumption."""
    need = _need(
        constraints={"contracts": [], "pattern_exclusions": ["no deployment system in this project"]},
        objective="pure algorithm work, no deployment system here",
    )
    surface, auth = _surface(need)
    app = cp.evaluate_applicability(RELEASE_DEPLOYMENT, need, surface, auth)
    assert app["verdict"] == cp.DOES_NOT_APPLY
    assert "declared exclusion" in app["reason"]


def test_29d_insufficient_evidence_unknown():
    p = cp.ConstructionPattern("empty", "no signals declared")
    need = _need()
    surface, auth = _surface(need)
    app = cp.evaluate_applicability(p, need, surface, auth)
    assert app["verdict"] == cp.UNKNOWN


def test_29e_name_similarity_alone_never_applies():
    """A pattern whose NAME matches but whose signals find no evidence stays
    UNKNOWN/DOES_NOT_APPLY — names establish nothing (§4)."""
    p = cp.ConstructionPattern(
        "deployment-thing",
        "sounds relevant",
        signals={"path_contains": ["/nonexistent/xyz/"], "objective_any": ["zzznever"]},
    )
    need = _need()
    surface, auth = _surface(need)
    app = cp.evaluate_applicability(p, need, surface, auth)
    assert app["verdict"] in (cp.DOES_NOT_APPLY, cp.UNKNOWN)


def test_29f_model_proposal_pending():
    prop = cp.record_pattern_proposal("some-pattern", provider="local", model="m", evidence=["hunch"], confidence=0.5)
    assert prop["verification_state"] == "PENDING"
    with pytest.raises(ValueError):
        cp.record_pattern_proposal("x", provider="", model="m", evidence=[], confidence=0.5)


def test_29g_project_evidence_overrides_pattern():
    """The strongest form of §3: pattern signals say APPLIES, but the project
    records an explicit exception for the derived obligation — the obligation
    is suppressed WITH provenance, not deleted (§18)."""
    need = _need()
    surface, auth = _surface(need)
    deriv = cp.derive_pattern_obligations(
        RELEASE_DEPLOYMENT,
        need,
        surface,
        auth,
        project_exceptions=[
            {
                "obligation": "rel-monotonic-promotion",
                "reason": "single-runner serialized deploys documented in project policy",
                "evidence": "project policy doc",
                "authority": "user",
            }
        ],
    )
    suppressed = [e for e in deriv["rejected"] if e["id"] == "rel-monotonic-promotion"]
    assert suppressed and suppressed[0]["verdict"] == "EXCEPTED"
    assert suppressed[0]["exception_evidence"] == "project policy doc"
    # the pattern's other obligations still derive
    assert any(e["id"] == "rel-fail-closed-qualification" for e in deriv["accepted"])


# ── §30 obligation derivation ───────────────────────────────────────────────


def test_30a_applicable_pattern_derives():
    need = _need()
    surface, auth = _surface(need)
    deriv = cp.derive_pattern_obligations(RELEASE_DEPLOYMENT, need, surface, auth)
    assert deriv["accepted"]
    e = deriv["accepted"][0]
    assert e["provenance"] == "PATTERN"
    assert e["pattern_provenance"]  # evidence chain carried


def test_30b_non_applicable_pattern_derives_nothing():
    need = _need(objective="write a poem", affected_surface=["docs/poem.md"])
    surface, auth = _surface(need)
    deriv = cp.derive_pattern_obligations(RELEASE_DEPLOYMENT, need, surface, auth)
    # no obligations accepted as REQUIRED (pattern-level always-obligations
    # only fire when the PATTERN applies — a DOES_NOT_APPLY pattern's
    # derivation is not consulted at all by make_decision)
    assert deriv["accepted"] == [] or all(e["verdict"] == "REQUIRED" for e in deriv["accepted"])


def test_30d_unknown_obligation_not_silently_required():
    p = cp.ConstructionPattern(
        "weird",
        "purpose",
        obligations=[{"id": "mystery", "obligation": "do the unknown thing", "when": {"unsupported_shape": True}}],
    )
    need = _need()
    surface, auth = _surface(need)
    deriv = cp.derive_pattern_obligations(p, need, surface, auth)
    assert deriv["unresolved"] and deriv["unresolved"][0]["verdict"] == cp.OBL_UNKNOWN
    assert deriv["accepted"] == []


def test_30e_provenance_distinct():
    """§8: structural derivations and pattern derivations carry different
    provenance kinds in the trace."""
    need = _need()
    trace = cd.make_decision(
        need, authorities={"rune_spine": _spine()}, project_toolchain={"existing_language": "python"}
    )
    obl = trace["obligations"]
    assert "pattern" in obl and obl["pattern"]
    assert all(o["provenance"] == "PATTERN" for o in obl["pattern"])
    # structural obligations (regression etc.) remain their own kind
    assert "regression" in obl


def test_30f_duplicate_obligations_collapse():
    p1 = cp.ConstructionPattern(
        "a", "pa", obligations=[{"id": "shared", "obligation": "regression proof", "when": {"always": True}}]
    )
    p2 = cp.ConstructionPattern(
        "b", "pb", obligations=[{"id": "shared", "obligation": "regression proof", "when": {"always": True}}]
    )
    d1 = cp.derive_pattern_obligations(p1, _need(), *(_surface(_need())[::-1]))
    d2 = cp.derive_pattern_obligations(p2, _need(), *(_surface(_need())[::-1]))
    combined = cp.combine_pattern_obligations([d1, d2])
    shared = [o for o in combined["obligations"] if o["id"] == "shared"]
    assert len(shared) == 1
    # first pattern's provenance kept, later duplicates recorded in also_from
    assert shared[0]["pattern"] == "a"
    assert "b" in shared[0]["also_from"]


def test_30g_conflicting_obligations_surface():
    p1 = cp.ConstructionPattern(
        "x", "px", obligations=[{"id": "shared", "obligation": "version A of the rule", "when": {"always": True}}]
    )
    p2 = cp.ConstructionPattern(
        "y", "py", obligations=[{"id": "shared", "obligation": "version B of the rule", "when": {"always": True}}]
    )
    d1 = cp.derive_pattern_obligations(p1, _need(), *(_surface(_need())[::-1]))
    d2 = cp.derive_pattern_obligations(p2, _need(), *(_surface(_need())[::-1]))
    combined = cp.combine_pattern_obligations([d1, d2])
    assert combined["conflicts"]
    assert combined["conflicts"][0]["conflict"] == "same id, different text"


# ── §31 pattern-specific ────────────────────────────────────────────────────


def test_31_release_pattern_negative_and_forward():
    need = _need()
    surface, auth = _surface(need)
    deriv = cp.derive_pattern_obligations(RELEASE_DEPLOYMENT, need, surface, auth)
    ids = {e["id"] for e in deriv["accepted"]}
    assert "rel-negative-candidate-proof" in ids
    assert "rel-forward-candidate-proof" in ids
    assert "rel-deployed-identity-verification" in ids


def test_31_ui_pattern_layout_inheritance():
    scaffold = ss.new_doc()
    from rune import layout_contract as lc

    lc.define_region(scaffold, "footer-bar", provenance="scaffold")
    lc.bind_component(scaffold, "version_display", "footer-bar", provenance="user", lock=True, order=0)
    need = cd.Need(
        "u",
        "adjust the tui version widget",
        affected_surface=["ui:version_display"],
        constraints={},
        evidence_refs=["x"],
        approval_context="covered",
    )
    authorities = {"rune_spine": ss.new_doc(), "project_scaffold": scaffold}
    surface = cd.resolve_affected_surface(need, authorities)
    from rune.construction_patterns_builtin import UI_TUI_STRUCTURAL

    app = cp.evaluate_applicability(UI_TUI_STRUCTURAL, need, surface, authorities)
    # ui-surface + entity signals hit; path signal misses (no source paths
    # in this fixture) — honest PARTIAL, obligations still derive
    assert app["verdict"] in (cp.APPLIES, cp.PARTIAL)
    deriv = cp.derive_pattern_obligations(UI_TUI_STRUCTURAL, need, surface, authorities)
    ids = {e["id"] for e in deriv["accepted"]}
    assert "ui-layout-contract-inherited" in ids
    assert "ui-user-locked-preserved" in ids
    # and the actual layout obligation confirms the LOCK
    lo = [o for o in cd.layout_obligations_for(scaffold, ["ui:version_display"])]
    assert lo and lo[0]["lock"] is True


def test_31_store_pattern_backward_readability():
    from rune.construction_patterns_builtin import STORE_SCHEMA_EVOLUTION

    need = cd.Need(
        "s",
        "add expected effects to the scaffold store",
        affected_surface=["src/rune/scaffold.py"],
        constraints={},
        evidence_refs=["x"],
        approval_context="c",
    )
    surface, auth = _surface(need)
    app = cp.evaluate_applicability(STORE_SCHEMA_EVOLUTION, need, surface, auth)
    assert app["verdict"] in (cp.APPLIES, cp.PARTIAL)
    deriv = cp.derive_pattern_obligations(STORE_SCHEMA_EVOLUTION, need, surface, auth)
    ids = {e["id"] for e in deriv["accepted"]}
    assert "store-backward-readability" in ids


# ── §32 process guards ──────────────────────────────────────────────────────


def test_32a_durable_mutation_reload_disagrees(tmp_path):
    p = tmp_path / "store.json"
    p.write_text(json.dumps({"flag": False}))
    doc = json.loads(p.read_text())
    doc["flag"] = True  # in-memory mutation...
    # ...never saved: the guard must catch the lie
    assert not cg.verify_persisted(p, lambda d: d["flag"], True)
    with pytest.raises(AssertionError):
        cg.assert_persisted(p, lambda d: d["flag"], True, what="flag")


def test_32b_durable_mutation_saved_reload_matches(tmp_path):
    p = tmp_path / "store.json"
    p.write_text(json.dumps({"flag": False}))
    doc = json.loads(p.read_text())
    doc["flag"] = True
    p.write_text(json.dumps(doc))  # actually saved
    assert cg.verify_persisted(p, lambda d: d["flag"], True)
    cg.assert_persisted(p, lambda d: d["flag"], True, what="flag")


def test_32c_zero_match_edit_is_noop_not_success(tmp_path):
    with pytest.raises(cg.EditNoOp):
        cg.apply_source_edit("nothing matches here", "ABSENT TEXT", "new")
    # intentional no-op is representable truthfully (§32E)
    result = cg.apply_source_edit("nothing matches here", "ABSENT TEXT", "new", allow_noop=True)
    assert result == "nothing matches here"


def test_32d_edit_postcondition_absent_fails():
    with pytest.raises(cg.EditPostconditionFailed):
        cg.apply_source_edit("alpha beta", "alpha", "gamma", postcondition="delta")


def test_32e_file_edit_with_readback(tmp_path):
    p = tmp_path / "mod.py"
    p.write_text("def old_fn():\n    pass\n")
    n = cg.apply_source_edit_file(p, "old_fn", "new_fn", expected_count=1, postcondition="def new_fn():")
    assert n == 1
    assert "new_fn" in p.read_text()
    # and the readback guard ran inside apply_source_edit_file


# ── §25 HELD-OUT: pattern must not overfit its origin incident ──────────────


def test_25heldout_release_config_change_not_demanding_promotion_rework():
    """Held-out release case: a deployment CONFIGURATION change that does not
    touch promotion ordering. The release pattern must NOT demand
    monotonic-promotion implementation change, but SHOULD still require
    release regression verification (release-level obligation)."""
    need = cd.Need(
        "heldout",
        "change the deploy notification webhook URL",
        affected_surface=[".gitea/workflows/deploy.yml"],
        constraints={},
        evidence_refs=["x"],
        approval_context="c",
    )
    authorities = {"rune_spine": _spine()}
    surface = cd.resolve_affected_surface(need, authorities)
    app = cp.evaluate_applicability(RELEASE_DEPLOYMENT, need, surface, authorities)
    # workflow-config surface hit → the pattern applies to the release class
    assert app["verdict"] in (cp.APPLIES, cp.PARTIAL)
    deriv = cp.derive_pattern_obligations(RELEASE_DEPLOYMENT, need, surface, authorities)
    accepted = {e["id"] for e in deriv["accepted"]}
    rejected = {e["id"] for e in deriv["rejected"]}
    # promotion-ordering work is NOT applicable to a webhook URL change
    assert "rel-monotonic-promotion" in rejected
    # but release-level invariants still apply
    assert "rel-release-regression-verification" in accepted
    assert "rel-deployed-identity-verification" in accepted


# ── §24 LIVE CHANGE: save_verified at the durable boundary ──────────────────

def test_24live_save_verified(tmp_path):
    """The pattern-governed live change: Scaffold.save_verified composes
    save + reload-readback. The UNSAVED-mutation lie fails; the saved truth
    passes. (Pattern contribution proven separately: store-roundtrip
    obligation derived by store-schema-evolution.)"""
    from rune import scaffold as sc  # local alias for the live test
    s = sc.Scaffold(tmp_path / ".rune" / "scaffold", "lv")
    doc = s.init(purpose="live")
    s.new_contract(doc, contract_id="LV1", need="n", scope=["src/x"])
    s.save(doc)  # contract IS persisted at this point
    # the T5 defect shape: mutate in memory, NEVER save, claim persisted —
    # verify_on_disk (pure readback, no write) catches the lie
    c = s.contract(doc, "LV1")
    c["identities"]["next_action"] = "tampered claim"
    with pytest.raises(AssertionError, match="probably never saved"):
        s.verify_on_disk(lambda d: s.contract(d, "LV1")["identities"][
            "next_action"], "tampered claim", what="LV1 next_action")
    # save-and-confirm catches disk-write failures (different semantic)
    s.transition_contract(doc, "LV1", "approved")
    s.save_verified(doc, lambda d: s.contract(d, "LV1")["phase"], "approved",
                    what="LV1 phase")
    # and the forgotten-save check now passes — the state is really durable
    s.verify_on_disk(lambda d: s.contract(d, "LV1")["phase"], "approved",
                     what="LV1 phase")
    # and a fresh instance sees it (the actual continuity requirement)
    doc2 = sc.Scaffold(tmp_path / ".rune" / "scaffold", "lv").load()
    assert sc.Scaffold(tmp_path / ".rune" / "scaffold", "lv").contract(
        doc2, "LV1")["phase"] == "approved"