rune/tests/test_construction_decision.py

"""test_construction_decision.py — SCAFFOLD-TRANCHE-004 §29 test plan (A-M)."""

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 capability_fit as cf  # noqa: E402
from rune import construction_decision as cd  # noqa: E402
from rune import scaffold as sc  # noqa: E402
from rune import structural_store as ss  # noqa: E402
from rune import toolchain as tc  # noqa: E402


# ── fixtures ────────────────────────────────────────────────────────────────


def _spine_doc():
    """Rune-Spine-like authority with an existing export capability."""
    doc = ss.new_doc()
    ss.add_entity(
        doc, id="capability:exporter", kind="capability", identities={}, maturity="WIRED", evidence_refs=["fixture"]
    )
    ss.add_entity(doc, id="contract:csv_export", kind="contract", identities={}, evidence_refs=["fixture"])
    ss.add_relationship(
        doc, type="implements", from_id="capability:exporter", to_id="contract:csv_export", evidence_refs=["fixture"]
    )
    ss.add_entity(doc, id="module:cli", kind="module", identities={}, evidence_refs=["fixture"])
    ss.add_relationship(
        doc, type="consumes", from_id="module:cli", to_id="capability:exporter", evidence_refs=["fixture"]
    )
    return doc


def _need(**kw):
    defaults = dict(
        need_id="need-001",
        objective="persistent project export",
        origin="user",
        project="rune",
        constraints={"contracts": ["contract:csv_export"]},
        affected_surface=[],
        evidence_refs=["user-request"],
        approval_context="covered by SCAFFOLD-TRANCHE-004",
    )
    defaults.update(kw)
    return cd.Need(**defaults)


# ── A. Need → CapabilityRequirement normalization ───────────────────────────


def test_a_normalization():
    req, missing = cd.normalize_need(_need())
    assert req.objective == "persistent project export"
    assert "contract:csv_export" in req.contracts
    assert req.props["source_need"] == "need-001"
    assert req.props["task_ref"] == ""


def test_a_normalization_preserves_provenance_and_flags_gaps():
    need = _need(evidence_refs=[], constraints={})
    req, missing = cd.normalize_need(need)
    assert req.props["source_need"] == "need-001"
    assert "evidence_refs" in missing  # recorded, not invented


# ── B. Need references Task/Intent without duplicating authority ───────────


def test_b_need_references_task_authority():
    need = _need(task_ref="intent:EXPR-42")
    d = need.to_dict()
    assert d["task_ref"] == "intent:EXPR-42"
    # the Need carries a REFERENCE — no task status/state fields of its own
    assert "status" not in d and "state" not in d and "tasks" not in d


# ── C. structural discovery invoked automatically ───────────────────────────


def test_c_live_discovery():
    trace = cd.make_decision(_need(), authorities={"rune_spine": _spine_doc()})
    assert any(c["id"] == "capability:exporter" for c in trace["candidates"])
    assert trace["candidates"][0]["authority"] == "rune_spine"


# ── D. exact REUSE path ─────────────────────────────────────────────────────


def test_d_reuse_path():
    trace = cd.make_decision(_need(), authorities={"rune_spine": _spine_doc()})
    assert trace["resolution"] == "REUSE"
    assert trace["research"]["route"] == "NONE"


# ── E. ADAPT_EXISTING path (with auto regression obligations) ───────────────


def test_e_adapt_path_auto_regression():
    doc = ss.new_doc()
    ss.add_entity(
        doc, id="capability:exporter", kind="capability", identities={}, maturity="WIRED", evidence_refs=["f"]
    )
    ss.add_entity(doc, id="contract:csv_export", kind="contract", identities={}, evidence_refs=["f"])
    ss.add_relationship(
        doc, type="implements", from_id="capability:exporter", to_id="contract:csv_export", evidence_refs=["f"]
    )
    ss.add_entity(doc, id="module:cli", kind="module", identities={}, evidence_refs=["f"])
    ss.add_relationship(doc, type="consumes", from_id="module:cli", to_id="capability:exporter", evidence_refs=["f"])
    need = _need(evidence_refs=["x"])
    need.constraints["contracts"] = ["contract:csv_export", "contract:encrypted_export"]
    trace = cd.make_decision(need, authorities={"rune_spine": doc})
    assert trace["resolution"] == "ADAPT_EXISTING"
    # BUILD-BUT-DON'T-BREAK obligations generated AUTOMATICALLY (§9):
    reg = " ".join(trace["obligations"]["regression"])
    assert "preserve existing behavior of capability:exporter" in reg
    assert "module:cli" in reg


# ── F. COMPOSE path ──────────────────────────────────────────────────────────


def test_f_compose_path():
    doc = ss.new_doc()
    ss.add_entity(doc, id="capability:a", kind="capability", identities={}, maturity="WIRED", evidence_refs=["f"])
    ss.add_entity(doc, id="capability:b", kind="capability", identities={}, maturity="WIRED", evidence_refs=["f"])
    for c in ("contract:streaming", "contract:bounded_memory"):
        ss.add_entity(doc, id=c, kind="contract", identities={}, evidence_refs=["f"])
    ss.add_relationship(doc, type="implements", from_id="capability:a", to_id="contract:streaming", evidence_refs=["f"])
    ss.add_relationship(
        doc, type="implements", from_id="capability:b", to_id="contract:bounded_memory", evidence_refs=["f"]
    )
    need = _need(
        objective="stream with bounded memory",
        constraints={"contracts": ["contract:streaming", "contract:bounded_memory"]},
        evidence_refs=["x"],
    )
    trace = cd.make_decision(need, authorities={"rune_spine": doc})
    assert trace["resolution"] == "COMPOSE"
    assert set(trace["obligations"]["composition"][0]["components"]) == {"capability:a", "capability:b"}
    assert "temporary" in trace["obligations"]["composition"][0]["permanence"]


# ── G. ACQUIRE_EXTERNAL routes to research seam ─────────────────────────────


def test_g_acquire_routes_to_research():
    trace = cd.make_decision(
        _need(objective="quantum error correction", constraints={"contracts": ["contract:qec"]}, evidence_refs=["x"]),
        authorities={"rune_spine": ss.new_doc()},
    )
    assert trace["resolution"] == "INSUFFICIENT_EVIDENCE"  # no candidates at all
    trace2 = cd.make_decision(
        _need(objective="quantum error correction", constraints={"contracts": ["contract:qec"]}, evidence_refs=["x"]),
        authorities={"rune_spine": _spine_doc()},
    )
    assert trace2["resolution"] == "INSUFFICIENT_EVIDENCE"
    # external evidence → ACQUIRE + full research route recorded (seam)
    doc = _spine_doc()
    ss.add_entity(doc, id="capability:x", kind="capability", identities={}, maturity="WIRED", evidence_refs=["f"])
    trace3 = cd.make_decision(
        _need(objective="qec", constraints={"contracts": ["contract:qec"]}, evidence_refs=["x"]),
        authorities={"rune_spine": doc},
    )
    assert trace3["resolution"] == "INSUFFICIENT_EVIDENCE"
    assert trace3["research"]["status"] == "RECORDED_NOT_EXECUTED" or trace3["research"]["route"] in (
        "NONE",
        "TARGETED_EVIDENCE_PROBE",
    )


def test_g_acquire_with_external_evidence():
    doc = ss.new_doc()
    ss.add_entity(doc, id="capability:p", kind="capability", identities={}, maturity="WIRED", evidence_refs=["f"])
    ss.add_entity(doc, id="contract:csv_export", kind="contract", identities={}, evidence_refs=["f"])
    # candidate provides nothing required → all incompatible
    res = cf.decide(
        doc,
        cd.normalize_need(_need())[0],
        [
            cf.evaluate_fit(
                cd.normalize_need(_need())[0],
                {
                    "id": "capability:p",
                    "kind": "capability",
                    "status": "CURRENT",
                    "superseded_by": None,
                    "provides": ["contract:other"],
                    "consumers": [],
                    "maturity": "WIRED",
                    "evidence_refs": [],
                    "identities": {},
                },
            )
        ],
        external_evidence=[{"tool": "libcsv", "license": "MIT"}],
    )
    assert res["resolution"] == "ACQUIRE_EXTERNAL"


# ── H/I. INSUFFICIENT_EVIDENCE never becomes BUILD_NEW ──────────────────────


def test_h_insufficient_never_build_new():
    trace = cd.make_decision(
        _need(objective="impossible thing", constraints={"contracts": ["contract:nope"]}, evidence_refs=["x"]),
        authorities={"rune_spine": _spine_doc()},
    )
    assert trace["resolution"] == "INSUFFICIENT_EVIDENCE"


def test_i_build_new_requires_positive_evidence_everywhere():
    doc1 = _spine_doc()  # has partials → ADAPT/COMPOSE territory
    doc2 = ss.new_doc()  # nothing at all → INSUFFICIENT
    need = _need(
        objective="odd requirement",
        constraints={"contracts": ["contract:csv_export", "contract:never_seen"]},
        evidence_refs=["x"],
    )
    t1 = cd.make_decision(need, authorities={"rune_spine": doc1})
    assert t1["resolution"] != "BUILD_NEW"  # csv_export partial exists → adapt
    t2 = cd.make_decision(need, authorities={"rune_spine": doc1, "project_scaffold": doc2})
    assert t2["resolution"] == "ADAPT_EXISTING"  # merge: concrete beats insufficient
    # BUILD_NEW only via brittle-coupling positive evidence (unit-covered in
    # test_capability_fit) — merge never upgrades insufficient to build


def test_i_normalization_gap_blocks_build_new():
    doc = ss.new_doc()
    # a partial fit exists; normalization is incomplete
    ss.add_entity(doc, id="capability:p", kind="capability", identities={}, maturity="WIRED", evidence_refs=["f"])
    ss.add_entity(doc, id="contract:csv_export", kind="contract", identities={}, evidence_refs=["f"])
    ss.add_relationship(
        doc, type="implements", from_id="capability:p", to_id="contract:csv_export", evidence_refs=["f"]
    )
    ss.add_entity(doc, id="contract:missing_half", kind="contract", identities={}, evidence_refs=["f"])
    ss.add_relationship(
        doc, type="implements", from_id="capability:p", to_id="contract:missing_half", evidence_refs=["f"]
    )
    ss.add_entity(doc, id="contract:unverifiable", kind="contract", identities={}, evidence_refs=["f"])
    need = _need(
        objective="x", constraints={"contracts": ["contract:unverifiable"]}, evidence_refs=[]
    )  # no evidence refs → normalization gap
    trace = cd.make_decision(need, authorities={"rune_spine": doc})
    assert trace["resolution"] == "INSUFFICIENT_EVIDENCE"
    assert "normalization gaps recorded, not invented" in trace["rationale"]


# ── J/K. toolchain decision enters the trace ────────────────────────────────


def test_j_toolchain_in_trace():
    trace = cd.make_decision(
        _need(), authorities={"rune_spine": _spine_doc()}, project_toolchain={"existing_language": "python"}
    )
    assert trace["toolchain"]["selected_language"] == "python"
    assert trace["toolchain"]["resolution"] == "KEEP_EXISTING"


def test_k_material_toolchain_change_marks_approval():
    trace = cd.make_decision(
        _need(),
        authorities={"rune_spine": _spine_doc()},
        toolchain_constraints=tc.ToolchainConstraints(existing_language="python"),
    )
    # simulate a requested change via direct resolver check (the live path
    # records the toolchain decision; a change request surfaces approval)
    d = tc.resolve_toolchain(tc.ToolchainConstraints(existing_language="python"), requested_language="rust")
    assert d.user_approval_required is True
    # and the approval gate honors it:
    gate = cd.needs_user_approval("REUSE", d, [], _need(approval_context=""))
    assert gate["required"] is True
    assert "toolchain change" in gate["reasons"][0]


# ── L. approved bounded maintenance does not repeatedly ask ─────────────────


def test_l_bounded_work_inside_approved_contract_does_not_reask():
    gate = cd.needs_user_approval(
        "REUSE",
        tc.resolve_toolchain(tc.ToolchainConstraints(existing_language="python")),
        [],
        _need(approval_context="covered by SCAFFOLD-TRANCHE-004"),
    )
    assert gate["required"] is False
    assert gate["covered_by"] == "covered by SCAFFOLD-TRANCHE-004"


# ── M. fresh-runtime decision trace recovery ────────────────────────────────


def test_m_fresh_runtime_trace_recovery(tmp_path):
    s = sc.Scaffold(tmp_path / ".rune" / "scaffold", "proj")
    doc = s.init(purpose="test")
    trace = cd.make_decision(
        _need(), authorities={"rune_spine": _spine_doc()}, project_toolchain={"existing_language": "python"}
    )
    cd.generate_contract(s, doc, trace, contract_id="LIVE-1", scope=["src/export"])
    s.transition_contract(doc, "LIVE-1", "approved")
    s.save(doc)
    # fresh interpreter, no chat transcript
    script = (
        "import sys, json;"
        "sys.path.insert(0, %r);"
        "from rune import scaffold;"
        "s = scaffold.Scaffold(%r, 'proj');"
        "doc = s.load();"
        "c = s.contract(doc, 'LIVE-1');"
        "t = c['ext']['decision_trace'];"
        "print(json.dumps({'need': t['need']['objective'],"
        "'resolution': t['resolution'],"
        "'rationale': t['rationale'][:40],"
        "'toolchain': t['toolchain']['selected_language'],"
        "'approval_required': t['approval']['required'],"
        "'contract_phase': c['phase'],"
        "'candidates': [x['id'] for x in t['candidates']]}))"
    ) % (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.strip())
    assert rec["need"] == "persistent project export"
    assert rec["resolution"] == "REUSE"
    assert rec["toolchain"] == "python"
    assert rec["approval_required"] is False
    assert rec["contract_phase"] == "approved"
    assert "capability:exporter" in rec["candidates"]


# ── contract generation populates fields from the trace ─────────────────────


def test_generated_contract_fields_from_trace(tmp_path):
    s = sc.Scaffold(tmp_path / ".rune" / "scaffold", "proj")
    doc = s.init(purpose="test")
    trace = cd.make_decision(
        _need(
            objective="add encrypted export",
            constraints={"contracts": ["contract:csv_export", "contract:encrypted_export"]},
            evidence_refs=["x"],
            approval_context="covered by SCAFFOLD-TRANCHE-004",
        ),
        authorities={"rune_spine": _spine_doc()},
        project_toolchain={"existing_language": "python"},
    )
    cd.generate_contract(s, doc, trace, contract_id="GEN-1", scope=["src/export"])
    c = s.contract(doc, "GEN-1")
    assert c["identities"]["resolution"] == "ADAPT_EXISTING"
    assert c["identities"]["toolchain"]["selected_language"] == "python"
    assert "capability:exporter" in c["identities"]["affected_entities"]
    assert c["identities"]["regression_obligations"]  # auto BUILD-BUT-DON'T-BREAK
    assert c["ext"]["decision_trace"]["need"]["need_id"] == "need-001"


# ── failure behavior (§28) ───────────────────────────────────────────────────


def test_stale_candidate_rejected_in_live_path():
    doc = _spine_doc()
    e = ss.entity(doc, "capability:exporter")
    e["status"] = "STALE"
    trace = cd.make_decision(_need(), authorities={"rune_spine": doc})
    assert trace["resolution"] == "INSUFFICIENT_EVIDENCE"


def test_merge_build_new_never_upgrades_from_insufficient():
    build = {"resolution": "BUILD_NEW", "rationale": "brittle"}
    insuff = {"resolution": "INSUFFICIENT_EVIDENCE", "rationale": "weak"}
    merged = cd.merge_resolutions([build, insuff])
    assert merged["resolution"] == "INSUFFICIENT_EVIDENCE"