rune/tests/test_surface_completeness.py

"""test_surface_completeness.py — SCAFFOLD-TRANCHE-005 test plans
§24 (affected surfaces), §25 (requirement completeness), §26 (iterative
trace), §27 (expected runtime effects), §28 (backward compat)."""

import json
import os
import subprocess
import sys

import pytest

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

from rune import construction_decision as cd  # noqa: E402
from rune import layout_contract as lc  # noqa: E402
from rune import scaffold as sc  # noqa: E402
from rune import structural_store as ss  # noqa: E402


def _authorities():
    """Spine-like authority: deploy module with a promotion contract the
    requirement does NOT represent (the deploy-incident shape), plus a
    path-mapped module, plus a UI component with a Layout Contract."""
    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="contract:fail_closed_deployment", kind="contract", identities={}, 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_relationship(
        spine,
        type="implements",
        from_id="module:deploy_qualify",
        to_id="contract:fail_closed_deployment",
        evidence_refs=["f"],
    )
    scaffold = ss.new_doc()
    lc.define_region(scaffold, "footer-bar", provenance="scaffold")
    lc.bind_component(
        scaffold, "version_display", "footer-bar", provenance="scaffold", order=0, implementation="src/rune/rune_tui.py"
    )
    return {"rune_spine": spine, "project_scaffold": scaffold}


def _need(**kw):
    d = dict(
        need_id="n5",
        objective="prevent deploy regression",
        origin="user",
        constraints={"contracts": ["contract:fail_closed_deployment"]},
        affected_surface=["scripts/deploy_qualify.py"],
        evidence_refs=["x"],
        approval_context="covered",
    )
    d.update(kw)
    return cd.Need(**d)


# ── §24 affected surfaces ───────────────────────────────────────────────────


def test_24a_source_path_propagates_to_file_scope(tmp_path):
    s, doc = _gen(tmp_path)
    c = s.contract(doc, "G5")
    assert "scripts/deploy_qualify.py" in c["ext"]["affected_paths"]
    assert "scripts/deploy_qualify.py" in c["identities"]["scope"]


def test_24b_mapped_path_resolves_to_module():
    res = cd.resolve_affected_surface(_need(), _authorities())
    assert "module:deploy_qualify" in res["affected_existing_entities"]
    r = next(r for r in res["resolutions"] if r["surface"] == "scripts/deploy_qualify.py")
    assert r["resolved"][0]["via"] == "path_attribute"


def test_24c_unresolved_path_retained_not_dropped():
    need = _need(affected_surface=["scripts/brand_new_thing.py"])
    res = cd.resolve_affected_surface(need, _authorities())
    assert any(u["surface"] == "scripts/brand_new_thing.py" for u in res["unresolved_surfaces"])
    assert "scripts/brand_new_thing.py" in res["affected_paths"]  # still scoped


def test_24d_ui_component_pulls_layout_contract():
    need = _need(affected_surface=["ui:version_display"])
    res = cd.resolve_affected_surface(need, _authorities())
    assert "ui:version_display" in res["affected_ui_surfaces"]
    r = res["resolutions"][0]
    assert r["resolved"][0]["via"] == "layout_contract"
    trace = cd.make_decision(need, authorities=_authorities())
    assert any(o["component"] == "ui:version_display" for o in trace["layout_obligations"])


def test_24e_region_surface_pulls_layout_obligations():
    need = _need(affected_surface=["region:footer-bar"])
    trace = cd.make_decision(need, authorities=_authorities())
    assert any(o["region"] == "footer-bar" for o in trace["layout_obligations"])


def test_24f_workflow_config_mechanically_scoped(tmp_path):
    s, doc = _gen(tmp_path, affected_surface=["scripts/deploy_qualify.py", ".gitea/workflows/deploy.yml"])
    c = s.contract(doc, "G5")
    assert ".gitea/workflows/deploy.yml" in c["ext"]["workflow_config_surfaces"]
    assert ".gitea/workflows/deploy.yml" in c["identities"]["scope"]


def test_24g_affected_surface_not_auto_expected_new(tmp_path):
    s, doc = _gen(tmp_path)
    c = s.contract(doc, "G5")
    # affected existing entities never leak into expected-new
    assert c["identities"]["expected_new_entities"] == []
    assert c["identities"]["affected_entities"]  # but affected is populated


def test_24h_expected_new_path_entity_distinct(tmp_path):
    s = sc.Scaffold(tmp_path / ".rune" / "scaffold", "p5")
    doc = s.init(purpose="t")
    need = _need(
        constraints={
            "contracts": ["contract:fail_closed_deployment"],
            "expected_new_entities": ["module:fresh_thing"],
            "expected_new_source_paths": ["scripts/fresh_thing.py"],
        },
        affected_surface=["scripts/deploy_qualify.py"],
        evidence_refs=["x"],
    )
    trace = cd.make_decision(need, authorities=_authorities())
    cd.generate_contract(s, doc, trace, contract_id="G5H", scope=None)
    c = s.contract(doc, "G5H")
    assert c["identities"]["expected_new_entities"] == ["module:fresh_thing"]
    assert "scripts/fresh_thing.py" in c["identities"]["scope"]  # expected-new path scoped
    assert "scripts/fresh_thing.py" not in c["ext"]["affected_paths"]


def test_24i_contract_roundtrip_preserves_surfaces(tmp_path):
    s, doc = _gen(tmp_path, affected_surface=["scripts/deploy_qualify.py", "ui:version_display", "scripts/unmapped.py"])
    s.save(doc)
    doc2 = sc.Scaffold(tmp_path / ".rune" / "scaffold", "p5").load()
    c = doc2 and sc.Scaffold(tmp_path / ".rune" / "scaffold", "p5").contract(doc2, "G5")
    assert c["ext"]["affected_paths"] == ["scripts/deploy_qualify.py", "scripts/unmapped.py"]
    assert c["ext"]["affected_ui_surfaces"] == ["ui:version_display"]
    assert any(u["surface"] == "scripts/unmapped.py" for u in c["ext"]["unresolved_surfaces"])


# ── §25 requirement completeness ────────────────────────────────────────────


def _gen(tmp_path, **need_kw):
    s = sc.Scaffold(tmp_path / ".rune" / "scaffold", "p5")
    doc = s.init(purpose="t5")
    need = _need(**need_kw)
    trace = cd.make_decision(need, authorities=_authorities(), project_toolchain={"existing_language": "python"})
    cd.generate_contract(s, doc, trace, contract_id="G5", scope=None)
    return s, doc


def test_25a_complete_requirement_complete_enough():
    need = _need(
        constraints={"contracts": ["contract:fail_closed_deployment"]}, affected_surface=["scripts/deploy_qualify.py"]
    )
    trace = cd.make_decision(need, authorities=_authorities())
    assert trace["completeness"]["verdict"] == cd.COMPLETE_ENOUGH
    assert trace["resolution"] == "REUSE"  # fit proceeds


def test_25b_affected_stage_missing_material_gap():
    """THE deploy-incident shape: requirement names qualification only, but
    the affected surface includes the promotion implementation carrying an
    unrepresented ordering contract."""
    need = _need(affected_surface=["scripts/deploy_qualify.py", "scripts/deploy_stage.py"])
    trace = cd.make_decision(need, authorities=_authorities())
    assert trace["completeness"]["verdict"] == cd.MATERIAL_GAP
    gaps = {g["missing_contract"] for g in trace["completeness"]["gaps"]}
    assert "contract:monotonic_promotion" in gaps
    # PREMATURE-FIT GUARD: without the guard this resolves REUSE/SATISFIED
    assert trace["resolution"] == "INSUFFICIENT_EVIDENCE"
    assert trace["_merged"].get("premature_fit_blocked", {}).get("gaps")
    # gap provenance: derived_from the entity, with rule + evidence
    g = next(g for g in trace["completeness"]["gaps"] if g["missing_contract"] == "contract:monotonic_promotion")
    assert g["derived_from"] == "module:deploy_stage"
    assert g["rule"] and g["evidence"]


def test_25c_no_resolution_unknown():
    need = _need(affected_surface=["totally/unknown.xyz"])
    trace = cd.make_decision(need, authorities=_authorities())
    assert trace["completeness"]["verdict"] == cd.COMPLETENESS_UNKNOWN


def test_25d_unknown_never_becomes_complete():
    need = _need(affected_surface=["totally/unknown.xyz"])
    trace = cd.make_decision(need, authorities=_authorities())
    assert trace["completeness"]["verdict"] != cd.COMPLETE_ENOUGH
    assert "undeterminable" in trace["completeness"]["note"]


def test_25e_derived_requirement_provenance():
    need = _need(affected_surface=["scripts/deploy_qualify.py", "scripts/deploy_stage.py"])
    trace = cd.make_decision(need, authorities=_authorities())
    g = trace["completeness"]["gaps"][0]
    assert g["derived_from"] and g["evidence"] and g["rule"]


def test_25f_model_proposal_not_structural_authority():
    """A model-proposed contract is recorded with provider/confidence and
    verification state PENDING — it does not enter the requirement's
    contract set as authority."""
    proposed = cd.record_proposed_requirement(
        "contract:model_claimed", provider="local-model", model="m", evidence=["hunch"], confidence=0.4
    )
    assert proposed["verification_state"] == "PENDING"
    assert proposed["provider"] == "local-model"
    # normalization does not accept it as a contract
    need = _need(constraints={"contracts": ["contract:fail_closed_deployment"], "model_proposed": [proposed]})
    req, _ = cd.normalize_need(need)
    assert "contract:model_claimed" not in req.contracts


def test_25g_consumer_contracts_derive_regression():
    spine = _authorities()["rune_spine"]
    ss.add_entity(spine, id="module:consumer", kind="module", identities={}, evidence_refs=["f"])
    ss.add_relationship(
        spine, type="consumes", from_id="module:consumer", to_id="module:deploy_qualify", evidence_refs=["f"]
    )
    need = _need(affected_surface=["scripts/deploy_qualify.py"])
    trace = cd.make_decision(need, authorities={"rune_spine": spine})
    assert trace["resolution"] == "REUSE"
    reg = " ".join(trace["obligations"]["regression"])
    assert "module:consumer" in reg


def test_25h_adapt_existing_includes_old_plus_new():
    spine = _authorities()["rune_spine"]
    ss.add_entity(spine, id="contract:extra_obligation", kind="contract", identities={}, evidence_refs=["f"])
    ss.add_relationship(
        spine,
        type="implements",
        from_id="module:deploy_qualify",
        to_id="contract:extra_obligation",
        evidence_refs=["f"],
    )
    need = _need(affected_surface=["scripts/deploy_qualify.py"])
    need.constraints["contracts"] = ["contract:fail_closed_deployment", "contract:extra_obligation"]
    trace = cd.make_decision(need, authorities={"rune_spine": spine})
    assert trace["completeness"]["verdict"] == cd.COMPLETE_ENOUGH
    assert trace["resolution"] == "REUSE"  # exact fit on both contracts
    assert "new behavior proven by tests" in " ".join(trace["obligations"]["verification"])


def test_25i_compose_derives_boundary_verification():
    spine = _authorities()["rune_spine"]
    need = _need(affected_surface=["scripts/deploy_qualify.py", "scripts/deploy_stage.py"])
    need.constraints["contracts"] = ["contract:fail_closed_deployment", "contract:monotonic_promotion"]
    trace = cd.make_decision(need, authorities={"rune_spine": spine})
    assert trace["resolution"] == "COMPOSE"
    assert "composition boundary behavior verified end-to-end" in " ".join(trace["obligations"]["verification"])
    comp = trace["obligations"]["composition"][0]
    assert set(comp["components"]) == {"module:deploy_qualify", "module:deploy_stage"}


def test_25j_layout_lock_derives_preservation():
    scaffold = _authorities()["project_scaffold"]
    lc.bind_component(
        scaffold, "status_bar", "footer-bar", lock=True, provenance="user", order=1, evidence_ref="user-2026"
    )
    need = _need(affected_surface=["ui:status_bar"])
    trace = cd.make_decision(need, authorities=_authorities() | {"project_scaffold": scaffold})
    lo = [o for o in trace["layout_obligations"] if o["component"] == "ui:status_bar"]
    assert lo and lo[0]["lock"] is True
    assert "explicit user authorization" in lo[0]["obligation"]
    # and the approval gate fires for the locked surface
    assert any("USER-LOCKED" in r for r in trace["approval"]["reasons"]) or trace["approval"]["covered_by"]


def test_25k_completeness_prevents_premature_reuse():
    """§23 core test, fixture form: the deploy-like Need must NOT resolve
    REUSE while the affected promotion surface carries an unrepresented
    contract."""
    need = _need(affected_surface=["scripts/deploy_qualify.py", "scripts/deploy_stage.py"])
    trace = cd.make_decision(need, authorities=_authorities())
    assert trace["resolution"] != "REUSE"
    assert trace["completeness"]["verdict"] == cd.MATERIAL_GAP
    blocked = trace["_merged"]["premature_fit_blocked"]
    assert blocked["would_have_been"] in ("REUSE", "SATISFIED")


# ── §26 iterative trace ──────────────────────────────────────────────────────


def test_26_iterative_trace_durable(tmp_path):
    """initial normalization → gap → refinement → revised → final decision;
    all retained; fresh process reconstructs without transcript."""
    s = sc.Scaffold(tmp_path / ".rune" / "scaffold", "p26")
    doc = s.init(purpose="t26")
    # iteration 1: narrow requirement → premature fit blocked
    need1 = _need(affected_surface=["scripts/deploy_qualify.py", "scripts/deploy_stage.py"])
    t1 = cd.make_decision(need1, authorities=_authorities())
    assert t1["resolution"] == "INSUFFICIENT_EVIDENCE"
    assert t1["_merged"]["premature_fit_blocked"]["would_have_been"] == "REUSE"
    # evidence/refinement: the promotion contract is added to the requirement
    need2 = _need(
        constraints={"contracts": ["contract:fail_closed_deployment", "contract:monotonic_promotion"]},
        affected_surface=["scripts/deploy_qualify.py", "scripts/deploy_stage.py"],
        evidence_refs=["x", "refinement:completeness-gap-iter1"],
    )
    t2 = cd.make_decision(
        need2,
        authorities=_authorities(),
        prior_iterations=[
            {
                "iteration": 1,
                "normalized_contracts": need1.constraints["contracts"],
                "completeness": t1["completeness"]["verdict"],
                "gaps": [g["missing_contract"] for g in t1["completeness"]["gaps"]],
                "resolution": t1["resolution"],
                "refined_by": "completeness gap evidence",
            }
        ],
    )
    assert t2["resolution"] == "COMPOSE"
    assert t2["iterations"][0]["gaps"] == ["contract:monotonic_promotion"]
    cd.generate_contract(s, doc, t2, contract_id="IT1", scope=None)
    s.transition_contract(doc, "IT1", "approved")
    s.save(doc)
    # fresh process
    script = (
        "import sys, json;"
        "sys.path.insert(0, %r);"
        "from rune import scaffold;"
        "s = scaffold.Scaffold(%r, 'p26');"
        "doc = s.load();"
        "c = s.contract(doc, 'IT1');"
        "t = c['ext']['decision_trace'];"
        "print(json.dumps({'iterations': len(t['iterations']),"
        "'iter1_gaps': t['iterations'][0]['gaps'],"
        "'iter1_resolution': t['iterations'][0]['resolution'],"
        "'refined_by': t['iterations'][0]['refined_by'],"
        "'final_resolution': t['resolution'],"
        "'final_contracts': t['requirement']['contracts']}))"
    ) % (os.path.join(os.path.dirname(__file__), "..", "src"), str(tmp_path / ".rune" / "scaffold"))
    r = subprocess.run([sys.executable, "-c", script], capture_output=True, text=True, timeout=60)
    assert r.returncode == 0, r.stderr
    rec = json.loads(r.stdout)
    assert rec["iterations"] == 1
    assert rec["iter1_gaps"] == ["contract:monotonic_promotion"]
    assert rec["iter1_resolution"] == "INSUFFICIENT_EVIDENCE"
    assert rec["refined_by"] == "completeness gap evidence"
    assert rec["final_resolution"] == "COMPOSE"
    assert "contract:monotonic_promotion" in rec["final_contracts"]


# ── §27 expected runtime effects ─────────────────────────────────────────────


def _close_with(tmp_path, expected_runtime, observed):
    s = sc.Scaffold(tmp_path / ".rune" / "scaffold", "p27")
    doc = s.init(purpose="t27")
    s.new_contract(
        doc, contract_id="R1", need="n", scope=["src/x"], affected_entities=["module:a"], expected_new_entities=[]
    )
    c = s.contract(doc, "R1")
    c.setdefault("ext", {})["expected_runtime_effects"] = expected_runtime
    s.transition_contract(doc, "R1", "approved")
    s.transition_contract(doc, "R1", "building")
    s.transition_contract(doc, "R1", "verifying")
    s.close_contract(doc, "R1", observed_evidence=observed)
    return s.contract(doc, "R1")["ext"]["contract_quality"]


def test_27a_expected_runtime_observed_not_unexpected(tmp_path):
    q = _close_with(
        tmp_path, ["deployment:current advances"], [{"entity": "deployment:current", "source": "deployed-identity"}]
    )
    assert q["expected_runtime_observed"] == ["deployment:current advances"]
    assert "deployment:current" not in q["genuinely_unexpected_new"]


def test_27b_expected_runtime_missing(tmp_path):
    q = _close_with(tmp_path, ["cache invalidated"], [])
    assert q["expected_runtime_missing"] == ["cache invalidated"]
    assert q["expected_runtime_observed"] == []


def test_27c_unpredicted_runtime_genuinely_unexpected(tmp_path):
    q = _close_with(tmp_path, [], [{"entity": "deployment:current", "source": "deployed-identity"}])
    assert "deployment:current" in q["genuinely_unexpected_new"]


def test_27d_no_self_certification(tmp_path):
    """Predicting the runtime effect does not observe it: no evidence item →
    missing, and correctly_predicted unaffected."""
    q = _close_with(tmp_path, ["deployment:current advances"], [])
    assert q["expected_runtime_missing"] == ["deployment:current advances"]
    assert q["expected_runtime_observed"] == []


def test_27e_deployment_current_fixture(tmp_path):
    """The DEPLOY-MONOTONIC-001 defect shape, now correct: a release-class
    contract that predicts the deployment advance closes clean."""
    q = _close_with(
        tmp_path,
        ["deployment:current advances"],
        [{"entity": "module:a", "source": "git-diff"}, {"entity": "deployment:current", "source": "deployed-identity"}],
    )
    assert q["correctly_predicted"] == ["module:a"]
    assert q["expected_runtime_observed"] == ["deployment:current advances"]
    assert q["genuinely_unexpected_new"] == []


# ── §28 backward compatibility ───────────────────────────────────────────────


def test_28_old_contract_closes_without_new_fields(tmp_path):
    s = sc.Scaffold(tmp_path / ".rune" / "scaffold", "p28")
    doc = s.init(purpose="t28")
    # old-shape: no expected_runtime_effects, no expected_new_entities
    s.new_contract(doc, contract_id="OLD", need="old", scope=["src/x"], affected_entities=["module:a"])
    c = s.contract(doc, "OLD")
    assert not c.get("ext", {}).get("expected_runtime_effects")  # empty, not crash
    s.transition_contract(doc, "OLD", "approved")
    s.transition_contract(doc, "OLD", "building")
    s.transition_contract(doc, "OLD", "verifying")
    s.close_contract(doc, "OLD", observed_evidence=[{"entity": "module:a", "source": "git-diff"}])
    q = s.contract(doc, "OLD")["ext"]["contract_quality"]
    assert q["correctly_predicted"] == ["module:a"]
    assert q["expected_runtime_effects"] == []  # empty, not crash


def test_28_tranche4_contract_data_still_readable(tmp_path):
    s = sc.Scaffold(tmp_path / ".rune" / "scaffold", "p28b")
    doc = s.init(purpose="t28b")
    s.new_contract(
        doc,
        contract_id="T4SHAPE",
        need="n",
        scope=["src/x"],
        affected_entities=["module:a"],
        expected_new_entities=["module:b"],
    )
    s.transition_contract(doc, "T4SHAPE", "approved")
    s.transition_contract(doc, "T4SHAPE", "building")
    s.transition_contract(doc, "T4SHAPE", "verifying")
    s.close_contract(doc, "T4SHAPE", observed_evidence=[{"entity": "module:b", "source": "file-added"}])
    q = s.contract(doc, "T4SHAPE")["ext"]["contract_quality"]
    assert q["expected_new_observed"] == ["module:b"]