Nick
6348386869
Failed stages now land in phase=failed (legal building→failed transition, evidence checkpoint on contract) instead of a swallowed close attempt. S110/BLE001 eliminated — no exception swallowing in the failure path. Ratchet-clean.
161 lines
6.4 KiB
Python
161 lines
6.4 KiB
Python
"""test_programme_wiring.py — T17 §17D/§17E: canonical programme
|
|
continuation wiring held-outs.
|
|
|
|
The 15M canary proved the RULE; these prove the WIRING: stages bound to
|
|
real Scaffolding contracts, driven through the Construction Interface
|
|
lifecycle, with restart/duplicate/idempotency/adversarial cases.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
REPO = Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(REPO / "src"))
|
|
|
|
from rune import construction_interface as ci # noqa: E402
|
|
from rune import programme_continuation as pc # noqa: E402
|
|
|
|
|
|
@pytest.fixture()
|
|
def iface(tmp_path):
|
|
return ci.ConstructionInterface(
|
|
builder="t17-programme-test",
|
|
project_store=str(tmp_path / "scaffold"),
|
|
project_id="prog-test")
|
|
|
|
|
|
def _make_contract(iface, contract_id, need="stage need"):
|
|
doc = iface.scaffold.load()
|
|
iface.scaffold.new_contract(doc, contract_id=contract_id, need=need,
|
|
origin="programme", affected_entities=[])
|
|
iface.scaffold.save(doc)
|
|
|
|
|
|
def _prog(prog_id, stages):
|
|
p = pc.Programme(programme_id=prog_id, objective="test programme")
|
|
for sid, work in stages:
|
|
st = pc.ProgrammeStage(stage_id=sid, objective=sid,
|
|
priority=len(p.stages))
|
|
if work is not None:
|
|
st.result["work"] = work
|
|
p.stages.append(st)
|
|
return p
|
|
|
|
|
|
def _phase(iface, contract_id):
|
|
doc = iface.scaffold.load()
|
|
c = iface.scaffold.contract(doc, contract_id)
|
|
return (c or {}).get("phase", "")
|
|
|
|
|
|
def test_17e_all_four_stages_close_with_evidence(iface, tmp_path):
|
|
stages = [("st-1", None),
|
|
("st-2", {"ok": True, "evidence": "did the thing"}),
|
|
("st-3", {"ok": True, "evidence": "more work"})]
|
|
prog = _prog("p-four", stages)
|
|
for sid, _ in stages:
|
|
_make_contract(iface, sid)
|
|
log = pc.run_construction_programme(prog, iface,
|
|
scaffold_store=tmp_path / "scaffold")
|
|
assert prog.continuation_canary()["verdict"] == "PROGRAMME_COMPLETE"
|
|
for sid, _ in stages:
|
|
assert _phase(iface, sid) == "closed"
|
|
doc = iface.scaffold.load()
|
|
c = next(e for e in doc["entities"] if e["id"] == "contract:st-2")
|
|
assert any("did the thing" in (cp.get("note", "") + cp.get("evidence", ""))
|
|
for cp in c.get("checkpoints", []))
|
|
|
|
|
|
def test_17e_failure_does_not_stop_programme(iface, tmp_path):
|
|
stages = [("f-1", {"ok": False, "evidence": "verification failed"}),
|
|
("f-2", {"ok": True, "evidence": "independent work ok"})]
|
|
prog = _prog("p-fail", stages)
|
|
for sid, _ in stages:
|
|
_make_contract(iface, sid)
|
|
pc.run_construction_programme(prog, iface,
|
|
scaffold_store=tmp_path / "scaffold")
|
|
verdict = prog.continuation_canary()["verdict"]
|
|
assert verdict == "PROGRAMME_COMPLETE_WITH_FAILURES"
|
|
assert _phase(iface, "f-1") == "failed" # truthful failure, not hung
|
|
assert _phase(iface, "f-2") == "closed"
|
|
|
|
|
|
def test_17e_restart_resumes_from_durable_record(iface, tmp_path):
|
|
stages = [("r-1", {"ok": True, "evidence": "done"}), ("r-2", None)]
|
|
prog = _prog("p-restart", stages)
|
|
for sid, _ in stages:
|
|
_make_contract(iface, sid)
|
|
base = tmp_path / "scaffold" / "programmes"
|
|
# stage 1 genuinely completed before the "restart": close its contract
|
|
iface.transition("r-1", "approved", evidence_ref="pre")
|
|
iface.transition("r-1", "building", evidence_ref="pre")
|
|
iface.transition("r-1", "verifying", evidence_ref="pre")
|
|
iface.close("r-1", observed_evidence=[{"source": "pre", "detail": "x"}],
|
|
scope_files=[])
|
|
prog.stages[0].state = pc.COMPLETE
|
|
prog.save(base)
|
|
# "restart": reload from the durable record only
|
|
prog2 = pc.Programme.load(base, "p-restart")
|
|
assert prog2.stages[0].state == pc.COMPLETE
|
|
assert prog2.stages[1].state == pc.PENDING
|
|
pc.run_construction_programme(prog2, iface,
|
|
scaffold_store=tmp_path / "scaffold")
|
|
assert _phase(iface, "r-1") == "closed"
|
|
assert _phase(iface, "r-2") == "closed"
|
|
|
|
|
|
def test_17e_duplicate_continuation_signal_idempotent(iface, tmp_path):
|
|
_make_contract(iface, "d-1")
|
|
prog = _prog("p-dup", [("d-1", {"ok": True, "evidence": "once"})])
|
|
base = tmp_path / "scaffold"
|
|
log1 = pc.run_construction_programme(prog, iface, scaffold_store=base)
|
|
# re-run the whole programme against the SAME contracts
|
|
prog2 = pc.Programme.load(base / "programmes", "p-dup")
|
|
log2 = pc.run_construction_programme(prog2, iface, scaffold_store=base)
|
|
assert _phase(iface, "d-1") == "closed" # not re-opened, not corrupted
|
|
assert any(e["canary"]["verdict"] == "PROGRAMME_COMPLETE"
|
|
for e in log2 if "canary" in e)
|
|
|
|
|
|
def test_17e_already_completed_stage_skipped(iface, tmp_path):
|
|
_make_contract(iface, "s-1")
|
|
iface.transition("s-1", "approved", evidence_ref="pre")
|
|
iface.transition("s-1", "building", evidence_ref="pre")
|
|
iface.transition("s-1", "verifying", evidence_ref="pre")
|
|
iface.close("s-1", observed_evidence=[{"source": "pre", "detail": "x"}],
|
|
scope_files=[])
|
|
prog = _prog("p-skip", [("s-1", {"ok": True, "evidence": "rework?"})])
|
|
log = pc.run_construction_programme(prog, iface,
|
|
scaffold_store=tmp_path / "scaffold")
|
|
# already-closed → COMPLETE without re-running work
|
|
assert prog.continuation_canary()["verdict"] == "PROGRAMME_COMPLETE"
|
|
assert not any("rework" in str(e) for e in log)
|
|
|
|
|
|
def test_17e_work_callable_receives_interface_and_stage(iface, tmp_path):
|
|
seen = {}
|
|
|
|
def work(i, stage):
|
|
seen["iface"] = type(i).__name__
|
|
seen["stage"] = stage.stage_id
|
|
return {"ok": True, "evidence": "callable ran"}
|
|
|
|
_make_contract(iface, "w-1")
|
|
prog = _prog("p-call", [("w-1", work)])
|
|
pc.run_construction_programme(prog, iface,
|
|
scaffold_store=tmp_path / "scaffold")
|
|
assert seen == {"iface": "ConstructionInterface", "stage": "w-1"}
|
|
assert _phase(iface, "w-1") == "closed"
|
|
|
|
|
|
def test_17e_missing_contract_fails_honestly(iface, tmp_path):
|
|
prog = _prog("p-missing", [("ghost-1", {"ok": True, "evidence": "x"})])
|
|
log = pc.run_construction_programme(prog, iface,
|
|
scaffold_store=tmp_path / "scaffold")
|
|
assert prog.stages[0].state == pc.FAILED
|
|
assert "no contract" in (prog.stages[0].result.get("error") or "")
|