Nick
0873b521e3
Slice 3 of the completion wave: closing a contract that is a stage of a persisted Programme marks that stage COMPLETE, persists the ledger, and rides the continuation canary on the close envelope — the next eligible stage is SURFACED, never guessed (task done ≠ programme done). No programme bound → no continuation block; ledger failure degrades without failing the closure. Three contracts. Also fixed en route: contract ids were second-granularity timestamps that collided when two needs filed in the same second (red test proves distinct ids now, uuid suffix added).
758 lines
29 KiB
Python
758 lines
29 KiB
Python
"""test_construction_interface.py — SCAFFOLD-TRANCHE-007 test plans
|
|
§29 (interface core), §30 (authority), §31 (pattern precision — integration),
|
|
§17 (provider independence)."""
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src"))
|
|
|
|
from rune import construction_interface as ci # noqa: E402
|
|
from rune import structural_store as ss # noqa: E402
|
|
from rune import scaffold as sc # noqa: E402
|
|
|
|
ROOT = os.path.join(os.path.dirname(__file__), "..")
|
|
|
|
|
|
def _fixture_authorities():
|
|
"""A self-contained project-scaffold authority (external-project mode
|
|
without any real external repo — §16)."""
|
|
spine = ss.new_doc()
|
|
ss.add_entity(
|
|
spine, id="capability:exporter", kind="capability", identities={}, maturity="WIRED", evidence_refs=["f"]
|
|
)
|
|
ss.add_entity(spine, id="contract:csv_export", kind="contract", identities={}, evidence_refs=["f"])
|
|
ss.add_relationship(
|
|
spine, type="implements", from_id="capability:exporter", to_id="contract:csv_export", evidence_refs=["f"]
|
|
)
|
|
return spine
|
|
|
|
|
|
def _iface(tmp_path, builder="test-builder", with_authority=True):
|
|
store = str(tmp_path / ".rune" / "scaffold")
|
|
iface = ci.ConstructionInterface(builder=builder, project_store=store, project_id="fixture-proj")
|
|
if with_authority:
|
|
# inject the fixture authority INTO the fixture scaffold store so
|
|
# external-project mode has structural truth to consult
|
|
iface.scaffold.init(purpose="fixture")
|
|
doc = iface.scaffold.load()
|
|
for e in _fixture_authorities()["entities"]:
|
|
doc["entities"].append(json.loads(json.dumps(e)))
|
|
for r in _fixture_authorities()["relationships"]:
|
|
doc["relationships"].append(json.loads(json.dumps(r)))
|
|
iface.scaffold.save(doc)
|
|
return iface
|
|
|
|
|
|
def _submit(iface, **kw):
|
|
payload = dict(
|
|
objective="persistent project export",
|
|
constraints={"contracts": ["contract:csv_export"]},
|
|
affected_surface=[],
|
|
evidence_refs=["x"],
|
|
approval_context="covered by fixture authority",
|
|
)
|
|
payload.update(kw)
|
|
return iface.submit_need(**payload)
|
|
|
|
|
|
# ── §29 interface core ──────────────────────────────────────────────────────
|
|
|
|
|
|
def test_29a_submit_need(tmp_path):
|
|
iface = _iface(tmp_path)
|
|
env = _submit(iface)
|
|
assert env["schema"] == ci.INTERFACE_SCHEMA_VERSION
|
|
assert env["operation"] == "submit_need"
|
|
assert env["success"]
|
|
assert env["ids"]["need_id"]
|
|
assert env["state"]["resolution"] == "REUSE"
|
|
assert env["state"]["builder"] == "test-builder"
|
|
|
|
|
|
def test_29b_retrieve_decision(tmp_path):
|
|
iface = _iface(tmp_path)
|
|
env = _submit(iface)
|
|
cid = env["ids"]["contract_id"]
|
|
d = iface.inspect_decision(cid)
|
|
assert d["success"]
|
|
assert d["state"]["resolution"] == "REUSE"
|
|
assert d["state"]["toolchain"]["selected_language"] == "python"
|
|
assert any(f["candidate"] == "capability:exporter" for f in d["state"]["fits"])
|
|
|
|
|
|
def test_29c_retrieve_contract(tmp_path):
|
|
iface = _iface(tmp_path)
|
|
cid = _submit(iface)["ids"]["contract_id"]
|
|
r = iface.resume(cid)
|
|
assert r["success"]
|
|
assert r["state"]["phase"] == "draft"
|
|
assert r["state"]["need"]["objective"] == "persistent project export"
|
|
|
|
|
|
def test_29d_allowed_scope(tmp_path):
|
|
iface = _iface(tmp_path)
|
|
cid = _submit(iface)["ids"]["contract_id"]
|
|
s = iface.get_scope(cid)
|
|
assert s["success"]
|
|
assert "allowed_paths" in s["state"]
|
|
assert "affected_structural_entities" in s["state"]
|
|
assert "unresolved_surfaces" in s["state"]
|
|
assert "layout_obligations" in s["state"]
|
|
|
|
|
|
def test_29e_obligations(tmp_path):
|
|
iface = _iface(tmp_path)
|
|
cid = _submit(iface)["ids"]["contract_id"]
|
|
o = iface.get_obligations(cid)
|
|
assert o["success"]
|
|
assert o["state"]["obligations"]
|
|
provs = {x["provenance"] for x in o["state"]["obligations"]}
|
|
assert provs <= {"STRUCTURAL", "PATTERN", "LAYOUT_CONTRACT", "CONTRACT"}
|
|
|
|
|
|
def test_29f_report_observation(tmp_path):
|
|
iface = _iface(tmp_path)
|
|
cid = _submit(iface)["ids"]["contract_id"]
|
|
r = iface.report_observation(cid, [{"entity": "capability:exporter", "detail": "exercised by new test"}])
|
|
assert r["success"]
|
|
assert r["state"]["recorded"] == 1
|
|
assert "cannot self-certify" not in r["state"]["note"]
|
|
assert "determined by Scaffolding gates" in r["state"]["note"]
|
|
|
|
|
|
def test_29g_verification_state(tmp_path):
|
|
iface = _iface(tmp_path)
|
|
cid = _submit(iface)["ids"]["contract_id"]
|
|
v = iface.verify(cid, changed_files=["src/new_thing.py"])
|
|
assert v["success"] or v["errors"]
|
|
assert "scope_check" in v["state"] or v["errors"]
|
|
|
|
|
|
def test_29h_resume_fresh_process(tmp_path):
|
|
"""§11 THE MAJOR ACCEPTANCE CONDITION: a fresh builder process resumes
|
|
from contract id alone."""
|
|
iface = _iface(tmp_path, builder="first-builder")
|
|
env = iface.submit_need(
|
|
"persistent project export",
|
|
constraints={"contracts": ["contract:csv_export"]},
|
|
affected_surface=["src/export_module.py"],
|
|
evidence_refs=["x"],
|
|
approval_context="covered by fixture authority",
|
|
)
|
|
cid = env["ids"]["contract_id"]
|
|
iface.transition(cid, "approved")
|
|
script = f"""
|
|
import sys, json
|
|
sys.path.insert(0, {os.path.join(ROOT, "src")!r})
|
|
from rune import construction_interface as ci
|
|
iface = ci.ConstructionInterface(builder="fresh-builder",
|
|
project_store={str(tmp_path / ".rune" / "scaffold")!r},
|
|
project_id="fixture-proj")
|
|
env = iface.resume({cid!r})
|
|
print(json.dumps(env))
|
|
"""
|
|
r = subprocess.run([sys.executable, "-c", script], capture_output=True, text=True, timeout=60)
|
|
assert r.returncode == 0, r.stderr
|
|
env = json.loads(r.stdout)
|
|
assert env["success"]
|
|
st = env["state"]
|
|
assert st["need"]["objective"] == "persistent project export"
|
|
assert st["decision"]["resolution"] == "REUSE"
|
|
assert st["phase"] == "approved"
|
|
assert st["scope"]
|
|
assert st["obligations"]
|
|
assert st["next_action"]
|
|
assert st["builder"] == "fresh-builder" # this session's identity
|
|
|
|
|
|
def test_29i_json_roundtrip(tmp_path):
|
|
iface = _iface(tmp_path)
|
|
env = _submit(iface)
|
|
blob = json.dumps(env)
|
|
back = json.loads(blob)
|
|
assert back["operation"] == env["operation"]
|
|
assert back["state"]["resolution"] == env["state"]["resolution"]
|
|
|
|
|
|
def test_29j_unknown_contract(tmp_path):
|
|
iface = _iface(tmp_path)
|
|
for op in (
|
|
lambda: iface.inspect_decision("nope"),
|
|
lambda: iface.get_scope("nope"),
|
|
lambda: iface.get_obligations("nope"),
|
|
lambda: iface.resume("nope"),
|
|
):
|
|
env = op()
|
|
assert not env["success"]
|
|
assert env["errors"][0]["error"] == ci.Errors.UNKNOWN_CONTRACT
|
|
|
|
|
|
def test_29k_approval_required(tmp_path):
|
|
iface = _iface(tmp_path)
|
|
env = iface.submit_need(
|
|
"build a whole new application from scratch",
|
|
constraints={},
|
|
affected_surface=[],
|
|
evidence_refs=["x"],
|
|
approval_context="",
|
|
) # no approval context
|
|
# new-capability construction without approval context → gate fires
|
|
if env["state"]["approval"]["required"]:
|
|
assert env["errors"]
|
|
assert any(e["error"] == ci.Errors.APPROVAL_REQUIRED for e in env["errors"])
|
|
else:
|
|
# covered path must say so
|
|
assert env["state"]["approval"]["covered_by"] or not env["state"]["approval"]["reasons"]
|
|
|
|
|
|
def test_29l_scope_violation_refused(tmp_path):
|
|
iface = _iface(tmp_path)
|
|
env = iface.submit_need(
|
|
"persistent project export",
|
|
constraints={"contracts": ["contract:csv_export"]},
|
|
affected_surface=["src/export_module.py"], # scoped surface
|
|
evidence_refs=["x"],
|
|
approval_context="covered by fixture authority",
|
|
)
|
|
cid = env["ids"]["contract_id"]
|
|
iface.transition(cid, "approved")
|
|
iface.transition(cid, "building")
|
|
iface.transition(cid, "verifying")
|
|
v = iface.verify(cid, changed_files=["src/completely/unrelated.py"])
|
|
# out-of-scope change must surface as a violation, never green
|
|
assert v["state"]["scope_check"]["verdict"] == "red" or any(
|
|
e["error"] == ci.Errors.SCOPE_VIOLATION for e in v["errors"]
|
|
)
|
|
|
|
|
|
def test_29m_failed_verification_cannot_close(tmp_path):
|
|
iface = _iface(tmp_path)
|
|
cid = _submit(iface)["ids"]["contract_id"]
|
|
# contract is in draft — closure must be refused by the phase map
|
|
env = iface.close(cid, observed_evidence=[{"entity": "capability:exporter", "detail": "claimed done"}])
|
|
assert not env["success"]
|
|
assert env["errors"][0]["error"] == ci.Errors.VERIFICATION_FAILED
|
|
|
|
|
|
def test_29n_stale_state_store_write(tmp_path):
|
|
"""STORE_WRITE_FAILED: an unwritable store surfaces as an explicit
|
|
error, not a silent lie."""
|
|
iface = _iface(tmp_path)
|
|
# break the store path
|
|
iface.scaffold.current = "/proc/nonexistent/current.json"
|
|
env = iface.report_observation("any", [{"entity": "x"}])
|
|
assert not env["success"] or env["errors"]
|
|
|
|
|
|
# ── §30 authority ────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_30a_rune_mode_reads_rune_authority():
|
|
"""Rune-internal mode consults the real Spine store (read-only)."""
|
|
iface = ci.ConstructionInterface(builder="t")
|
|
auth = iface._authorities()
|
|
assert "rune_spine" in auth
|
|
assert iface.mode == "rune_internal"
|
|
|
|
|
|
def test_30b_external_mode_reads_project_scaffold(tmp_path):
|
|
iface = _iface(tmp_path)
|
|
auth = iface._authorities()
|
|
assert "rune_spine" not in auth # external project has no Rune Spine
|
|
assert iface.mode == "external_project"
|
|
assert auth["project_scaffold"]["entities"]
|
|
|
|
|
|
def test_30c_builder_cannot_rewrite_structural_authority():
|
|
"""The interface exposes NO operation that writes structural truth.
|
|
(The only writes are contract/observation/session records in the
|
|
scaffold store — never Spine entities.)"""
|
|
public = [m for m in dir(ci.ConstructionInterface) if not m.startswith("_")]
|
|
for forbidden in ("rewrite_spine", "set_entity", "promote", "supersede", "build_spine", "add_relationship"):
|
|
assert forbidden not in public
|
|
|
|
|
|
def test_30d_builder_cannot_override_user_locked(tmp_path):
|
|
iface = _iface(tmp_path)
|
|
from rune import layout_contract as lc
|
|
|
|
doc = iface.scaffold.load()
|
|
lc.define_region(doc, "footer-bar", provenance="scaffold")
|
|
lc.bind_component(doc, "version_display", "footer-bar", lock=True, provenance="user", order=0)
|
|
iface.scaffold.save(doc)
|
|
# no interface operation amends layout at all
|
|
public = [m for m in dir(ci.ConstructionInterface) if not m.startswith("_")]
|
|
assert not any("layout" in m and "amend" in m for m in public)
|
|
# and the direct amend path still refuses non-user authority
|
|
doc2 = iface.scaffold.load()
|
|
with pytest.raises(ValueError):
|
|
lc.amend_layout_contract(
|
|
doc2, "version_display", new_region="header", authorized_by="builder", reason="prefer it"
|
|
)
|
|
|
|
|
|
def test_30e_builder_cannot_self_certify(tmp_path):
|
|
iface = _iface(tmp_path)
|
|
cid = _submit(iface)["ids"]["contract_id"]
|
|
# no interface operation takes "done=True"
|
|
v = iface.verify(cid)
|
|
assert "cannot self-certify" in v["state"]["remaining"]
|
|
# closure from draft fails
|
|
env = iface.close(cid, observed_evidence=[{"entity": "x"}])
|
|
assert not env["success"]
|
|
|
|
|
|
def test_30f_builder_identity_neutral(tmp_path):
|
|
"""§17: two builders with identical input get identical construction
|
|
behavior (identity recorded, semantics unchanged)."""
|
|
results = []
|
|
for b in ("mimo", "kimi", "deterministic-script"):
|
|
iface = _iface(tmp_path / b, builder=b)
|
|
env = _submit(iface)
|
|
results.append(
|
|
(
|
|
env["state"]["resolution"],
|
|
env["state"]["completeness"],
|
|
env["state"]["toolchain"] if "toolchain" in env["state"] else None,
|
|
[o["id"] for o in iface.get_obligations(env["ids"]["contract_id"])]["obligations"] if False else None,
|
|
)
|
|
)
|
|
resolutions = {r[0] for r in results}
|
|
completions = {r[1] for r in results}
|
|
assert len(resolutions) == 1 and len(completions) == 1
|
|
|
|
|
|
# ── §31 pattern precision (integration through the interface) ───────────────
|
|
|
|
|
|
def test_31a_ui_task_derives_ui_obligations(tmp_path):
|
|
iface = _iface(tmp_path)
|
|
from rune import layout_contract as lc
|
|
|
|
doc = iface.scaffold.load()
|
|
lc.define_region(doc, "footer-bar", provenance="scaffold")
|
|
lc.bind_component(doc, "version_display", "footer-bar", provenance="scaffold")
|
|
iface.scaffold.save(doc)
|
|
env = iface.submit_need(
|
|
"adjust the tui version widget",
|
|
constraints={},
|
|
affected_surface=["ui:version_display"],
|
|
evidence_refs=["x"],
|
|
approval_context="covered",
|
|
)
|
|
cid = env["ids"]["contract_id"]
|
|
o = iface.get_obligations(cid)
|
|
ids = {x["id"] for x in o["state"]["obligations"]}
|
|
assert any("ui-" in i or "layout:" in i for i in ids)
|
|
|
|
|
|
def test_31b_non_ui_task_no_active_ui_obligations(tmp_path):
|
|
"""§0 regression through the interface: non-UI work gets NO active
|
|
ui-* obligations."""
|
|
iface = _iface(tmp_path)
|
|
env = _submit(iface)
|
|
cid = env["ids"]["contract_id"]
|
|
o = iface.get_obligations(cid)
|
|
ui_active = [x for x in o["state"]["obligations"] if x["id"].startswith("ui-")]
|
|
assert ui_active == []
|
|
|
|
|
|
def test_31c_release_task_derives_release_obligations(tmp_path):
|
|
iface = _iface(tmp_path)
|
|
# the fixture authority needs the deploy structural truth for fit
|
|
doc = iface.scaffold.load()
|
|
ss.add_entity(
|
|
doc,
|
|
id="module:deploy_stage",
|
|
kind="module",
|
|
identities={"path": "scripts/deploy_stage.py"},
|
|
maturity="WIRED",
|
|
evidence_refs=["f"],
|
|
)
|
|
ss.add_entity(doc, id="contract:monotonic_promotion", kind="contract", identities={}, evidence_refs=["f"])
|
|
ss.add_relationship(
|
|
doc, type="implements", from_id="module:deploy_stage", to_id="contract:monotonic_promotion", evidence_refs=["f"]
|
|
)
|
|
iface.scaffold.save(doc)
|
|
env = iface.submit_need(
|
|
"repair the deploy promotion ordering",
|
|
constraints={"contracts": ["contract:monotonic_promotion"]},
|
|
affected_surface=["scripts/deploy_stage.py", "scripts/deploy_promotion.py"],
|
|
evidence_refs=["x"],
|
|
approval_context="covered",
|
|
)
|
|
cid = env["ids"]["contract_id"]
|
|
o = iface.get_obligations(cid)
|
|
ids = {x["id"] for x in o["state"]["obligations"]}
|
|
assert "rel-monotonic-promotion" in ids or any("rel-" in i for i in ids)
|
|
|
|
|
|
def test_31d_non_release_task_no_promotion_obligations(tmp_path):
|
|
iface = _iface(tmp_path)
|
|
env = _submit(iface)
|
|
cid = env["ids"]["contract_id"]
|
|
o = iface.get_obligations(cid)
|
|
rel_active = [x for x in o["state"]["obligations"] if x["id"].startswith("rel-")]
|
|
assert rel_active == []
|
|
|
|
|
|
def test_31g_partial_pattern_inspectable_without_activation(tmp_path):
|
|
iface = _iface(tmp_path)
|
|
env = iface.submit_need(
|
|
"adjust the tui backlog plumbing",
|
|
constraints={"contracts": ["contract:csv_export"]},
|
|
affected_surface=["src/rune/tui_backlog.py", "src/export_module.py"],
|
|
evidence_refs=["x"],
|
|
approval_context="covered",
|
|
)
|
|
cid = env["ids"]["contract_id"]
|
|
d = iface.inspect_decision(cid)
|
|
considered = {a["pattern"]: a["verdict"] for a in d["state"]["patterns_considered"]}
|
|
# PARTIAL patterns remain VISIBLE as candidates (inspectable)
|
|
assert considered
|
|
# but their UI obligations did not enter the ACTIVE obligation set
|
|
o = iface.get_obligations(cid)
|
|
ui_active = [x for x in o["state"]["obligations"] if x["id"].startswith("ui-")]
|
|
assert ui_active == []
|
|
|
|
|
|
# ── CLI transport ────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_cli_roundtrip(tmp_path):
|
|
store = tmp_path / ".rune" / "scaffold"
|
|
iface = ci.ConstructionInterface(builder="seed", project_store=str(store), project_id="cli-fixture")
|
|
iface.scaffold.init(purpose="cli")
|
|
doc = iface.scaffold.load()
|
|
for e in _fixture_authorities()["entities"]:
|
|
doc["entities"].append(json.loads(json.dumps(e)))
|
|
for r in _fixture_authorities()["relationships"]:
|
|
doc["relationships"].append(json.loads(json.dumps(r)))
|
|
iface.scaffold.save(doc)
|
|
env = {
|
|
"objective": "persistent project export",
|
|
"constraints": {"contracts": ["contract:csv_export"]},
|
|
"affected_surface": [],
|
|
"evidence_refs": ["x"],
|
|
"approval_context": "covered",
|
|
}
|
|
r = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
"-m",
|
|
"rune.construction_interface",
|
|
"submit-need",
|
|
json.dumps(env),
|
|
"--builder",
|
|
"cli-builder",
|
|
"--project-store",
|
|
str(store),
|
|
"--project-id",
|
|
"cli-fixture",
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=120,
|
|
env={**os.environ, "PYTHONPATH": os.path.join(ROOT, "src")},
|
|
)
|
|
assert r.returncode == 0, r.stderr[-400:]
|
|
out = json.loads(r.stdout[r.stdout.index("{") :])
|
|
assert out["success"]
|
|
assert out["state"]["resolution"] == "REUSE"
|
|
# resume through the CLI from the returned contract id
|
|
r2 = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
"-m",
|
|
"rune.construction_interface",
|
|
"resume",
|
|
out["ids"]["contract_id"],
|
|
"--builder",
|
|
"cli-2",
|
|
"--project-store",
|
|
str(store),
|
|
"--project-id",
|
|
"cli-fixture",
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=120,
|
|
env={**os.environ, "PYTHONPATH": os.path.join(ROOT, "src")},
|
|
)
|
|
assert r2.returncode == 0, r2.stderr[-400:]
|
|
out2 = json.loads(r2.stdout[r2.stdout.index("{") :])
|
|
assert out2["state"]["need"]["objective"] == "persistent project export"
|
|
|
|
|
|
# ── §28 capabilities / schema negotiation (SCAFFOLD-TRANCHE-008) ────────────
|
|
|
|
|
|
def test_capabilities_operation(tmp_path):
|
|
"""Builders can detect the supported interface schema + operations
|
|
BEFORE calling anything (T7 §28 requirement, closed by this tranche)."""
|
|
iface = _iface(tmp_path)
|
|
env = iface.capabilities()
|
|
assert env["success"]
|
|
st = env["state"]
|
|
assert st["schema_version"] == ci.INTERFACE_SCHEMA_VERSION
|
|
for op in (
|
|
"submit_need",
|
|
"inspect_decision",
|
|
"get_scope",
|
|
"get_obligations",
|
|
"report_observation",
|
|
"verify",
|
|
"transition",
|
|
"close",
|
|
"resume",
|
|
"capabilities",
|
|
):
|
|
assert op in st["operations"]
|
|
assert ci.Errors.SCOPE_VIOLATION in st["error_states"]
|
|
assert ci.Errors.UNKNOWN_CONTRACT in st["error_states"]
|
|
# mode + project identity exposed for builder routing
|
|
assert st["mode"] == "external_project"
|
|
assert st["project"] == "fixture-proj"
|
|
|
|
|
|
def test_capabilities_cli_roundtrip(tmp_path):
|
|
"""The CLI transport exposes the same capability set (§3 transport
|
|
parity: the builder must not need the Python API for schema detection)."""
|
|
import subprocess
|
|
|
|
env = {"objective": "x", "constraints": {}, "affected_surface": [], "evidence_refs": [], "approval_context": ""}
|
|
r = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
"-m",
|
|
"rune.construction_interface",
|
|
"capabilities",
|
|
"--builder",
|
|
"probe",
|
|
"--project-store",
|
|
str(tmp_path / ".rune" / "scaffold"),
|
|
"--project-id",
|
|
"fixture-proj",
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=120,
|
|
env={**os.environ, "PYTHONPATH": os.path.join(ROOT, "src")},
|
|
)
|
|
assert r.returncode == 0, r.stderr[-300:]
|
|
out = json.loads(r.stdout[r.stdout.index("{") :])
|
|
assert out["success"]
|
|
assert out["state"]["schema_version"] == ci.INTERFACE_SCHEMA_VERSION
|
|
assert "capabilities" in out["state"]["operations"]
|
|
|
|
|
|
def test_15u_discover_active_contract_migrated_to_interface(tmp_path):
|
|
"""§15U simplification: resume discovery lives in the INTERFACE (one
|
|
canonical home), not in session-driver scripts. The drivers' helpers
|
|
are shims over this."""
|
|
from rune import construction_interface as ci
|
|
store = tmp_path / "scaffold"
|
|
iface = ci.ConstructionInterface(builder="probe",
|
|
project_store=str(store),
|
|
project_id="t15u")
|
|
iface.scaffold.init(purpose="15u probe")
|
|
empty = ci.ConstructionInterface.discover_active_contract(str(store))
|
|
assert not empty["success"], "empty store must not invent a contract"
|
|
doc = iface.scaffold.load()
|
|
from rune import structural_store as ss
|
|
ss.add_entity(doc, id="contract:spec-15u", kind="contract",
|
|
identities={"objective": "spec"}, evidence_refs=["e"])
|
|
ss.add_entity(doc, id="capability:planned", kind="capability",
|
|
maturity="DECLARED", evidence_refs=["e"])
|
|
ss.add_relationship(doc, type="implements",
|
|
from_id="capability:planned",
|
|
to_id="contract:spec-15u", evidence_refs=["e"])
|
|
iface.scaffold.commit(doc, what="seed 15u spec")
|
|
sub = iface.submit_need(
|
|
"build the 15u module",
|
|
constraints={"environment": {"os": "linux"},
|
|
"contracts": ["contract:spec-15u"]},
|
|
evidence_refs=["e"], approval_context="15u")
|
|
assert sub["success"]
|
|
found = ci.ConstructionInterface.discover_active_contract(str(store))
|
|
assert found["success"]
|
|
assert found["ids"]["contract_id"] == sub["ids"]["contract_id"]
|
|
# capabilities() advertises the new op
|
|
assert "discover_active_contract" in \
|
|
iface.capabilities()["state"]["operations"]
|
|
|
|
|
|
# ── T5.15 completion wave: stage_guards wired into transition ───────────────
|
|
|
|
|
|
def _guard_contract(iface, tmp_path):
|
|
"""A contract whose obligations declare a PATH guard and a failing
|
|
command guard — proving both BLOCK shapes."""
|
|
doc = iface.scaffold.load()
|
|
ss.add_entity(
|
|
doc, id="contract:guarded-1", kind="contract",
|
|
identities={
|
|
"objective": "guarded stage",
|
|
"verification_obligations": [
|
|
"guard:" + json.dumps({
|
|
"name": "artifact present", "kind": "path_exists",
|
|
"path": str(tmp_path / "artifact.txt"),
|
|
}),
|
|
],
|
|
},
|
|
evidence_refs=["e"],
|
|
)
|
|
iface.scaffold.commit(doc, what="seed guarded contract")
|
|
doc = iface.scaffold.load()
|
|
# move through the legal path so guards face verifying from building
|
|
sc.Scaffold.transition_contract(iface.scaffold, doc, "guarded-1", "approved",
|
|
evidence_ref="t")
|
|
sc.Scaffold.transition_contract(iface.scaffold, doc, "guarded-1", "building",
|
|
evidence_ref="t")
|
|
iface.scaffold.commit(doc, what="advance to building")
|
|
return "guarded-1"
|
|
|
|
|
|
def test_15f_transition_to_verifying_blocked_when_guard_fails(tmp_path):
|
|
iface = _iface(tmp_path)
|
|
cid = _guard_contract(iface, tmp_path)
|
|
env = iface.transition(cid, "verifying")
|
|
assert not env["success"]
|
|
assert env["errors"][0]["error"] == ci.Errors.VERIFICATION_FAILED
|
|
assert env["state"]["guard"]["verdict"] == "BLOCKED"
|
|
# phase UNCHANGED — the blocked stage does not move
|
|
doc = iface.scaffold.load()
|
|
c = iface.scaffold.contract(doc, cid)
|
|
assert c["phase"] == "building"
|
|
|
|
|
|
def test_15f_transition_passes_when_guard_satisfied(tmp_path):
|
|
iface = _iface(tmp_path)
|
|
cid = _guard_contract(iface, tmp_path)
|
|
(tmp_path / "artifact.txt").write_text("real artifact")
|
|
env = iface.transition(cid, "verifying")
|
|
assert env["success"], env["errors"]
|
|
doc = iface.scaffold.load()
|
|
assert iface.scaffold.contract(doc, cid)["phase"] == "verifying"
|
|
|
|
|
|
def test_15f_inline_guard_checks_override_obligations(tmp_path):
|
|
iface = _iface(tmp_path)
|
|
cid = _guard_contract(iface, tmp_path) # has the path guard, NOT satisfied
|
|
env = iface.transition(
|
|
cid, "verifying",
|
|
guard_checks=[{"name": "inline truth", "kind": "command",
|
|
"argv": [sys.executable, "-c", "print('ok')"]}],
|
|
)
|
|
assert env["success"], env["errors"]
|
|
assert iface.scaffold.contract(iface.scaffold.load(), cid)["phase"] == "verifying"
|
|
|
|
|
|
def test_15f_transition_to_building_runs_no_guards(tmp_path):
|
|
"""Completion-facing edges ONLY (verifying/closed): building→verifying
|
|
is guarded, but draft→approved passes freely even with obligations."""
|
|
iface = _iface(tmp_path)
|
|
doc = iface.scaffold.load()
|
|
ss.add_entity(doc, id="contract:guarded-2", kind="contract",
|
|
identities={"objective": "2"}, evidence_refs=["e"])
|
|
iface.scaffold.commit(doc, what="seed guarded-2")
|
|
cid = "guarded-2"
|
|
doc = iface.scaffold.load()
|
|
c = iface.scaffold.contract(doc, cid)
|
|
c["identities"]["verification_obligations"] = [
|
|
"guard:" + json.dumps({"name": "never", "kind": "path_exists",
|
|
"path": "/nonexistent/for-sure"})
|
|
]
|
|
iface.scaffold.commit(doc, what="attach guard")
|
|
env = iface.transition(cid, "approved") # NOT a completion edge
|
|
assert env["success"]
|
|
|
|
|
|
# ── T5.15 slice 3: programme_continuation wired into close() ────────────────
|
|
|
|
|
|
def _seed_programme(iface, programme_id, stage_contract_ids, tmp_path):
|
|
"""Persist a real Programme bound to the interface's scaffold store."""
|
|
from rune import programme_continuation as pc
|
|
|
|
base = Path(iface.scaffold.current).parent / pc.DEFAULT_PROGRAMME_AREA
|
|
prog = pc.Programme(programme_id=programme_id, objective="wave test")
|
|
for i, cid in enumerate(stage_contract_ids):
|
|
prog.stages.append(pc.ProgrammeStage(
|
|
stage_id=cid, objective=f"stage {i}", depends_on=[]))
|
|
prog.save(base)
|
|
return prog, base
|
|
|
|
|
|
def test_close_marks_programme_stage_complete_and_surfaces_continuation(tmp_path):
|
|
"""Closing a contract that is a stage of a persisted Programme marks
|
|
that stage COMPLETE and the envelope carries the continuation canary —
|
|
the programme's next eligible stage is surfaced, not guessed."""
|
|
from rune import programme_continuation as pc
|
|
|
|
iface = _iface(tmp_path)
|
|
env1 = _submit(iface, objective="wave stage one")
|
|
cid1 = env1["ids"]["contract_id"]
|
|
env2 = _submit(iface, objective="wave stage two")
|
|
cid2 = env2["ids"]["contract_id"]
|
|
# drive both through to verifying so close is legal
|
|
for cid in (cid1, cid2):
|
|
iface.transition(cid, "approved")
|
|
iface.transition(cid, "building")
|
|
iface.transition(cid, "verifying")
|
|
prog, base = _seed_programme(iface, "wave-x", [cid1, cid2], tmp_path)
|
|
env = iface.close(cid1, observed_evidence=[{"entity": "capability:exporter", "detail": "done"}])
|
|
assert env["success"], env["errors"]
|
|
cont = env["state"].get("programme_continuation")
|
|
assert cont, "close envelope must carry the programme continuation"
|
|
assert cont["programme_id"] == "wave-x"
|
|
assert cont["stage_marked_complete"] == cid1
|
|
reloaded = pc.Programme.load(base, "wave-x")
|
|
stage1 = next(s for s in reloaded.stages if s.stage_id == cid1)
|
|
assert stage1.state == pc.COMPLETE
|
|
assert cont["canary"]["verdict"] == "PROGRAMME_OPEN"
|
|
assert cont["canary"]["next_action"] == ("start", cid2)
|
|
|
|
|
|
def test_close_without_programme_has_no_continuation_block(tmp_path):
|
|
iface = _iface(tmp_path)
|
|
env = _submit(iface)
|
|
cid = env["ids"]["contract_id"]
|
|
iface.transition(cid, "approved")
|
|
iface.transition(cid, "building")
|
|
iface.transition(cid, "verifying")
|
|
env = iface.close(cid, observed_evidence=[{"entity": "capability:exporter", "detail": "done"}])
|
|
assert env["success"]
|
|
assert "programme_continuation" not in env["state"]
|
|
|
|
|
|
def test_close_programme_completion_verdict_surfaces(tmp_path):
|
|
"""Closing the LAST stage yields PROGRAMME_COMPLETE — the canary says
|
|
stop, the runner never guesses."""
|
|
iface = _iface(tmp_path)
|
|
env = _submit(iface)
|
|
cid = env["ids"]["contract_id"]
|
|
iface.transition(cid, "approved")
|
|
iface.transition(cid, "building")
|
|
iface.transition(cid, "verifying")
|
|
prog, base = _seed_programme(iface, "solo", [cid], tmp_path)
|
|
env = iface.close(cid, observed_evidence=[{"entity": "capability:exporter", "detail": "done"}])
|
|
cont = env["state"]["programme_continuation"]
|
|
assert cont["canary"]["verdict"] == "PROGRAMME_COMPLETE"
|
|
|
|
|
|
def test_two_submits_same_second_get_distinct_contract_ids(tmp_path):
|
|
"""Second-granularity timestamps collide when two needs are filed in
|
|
the same second (real builder-pair scenario) — ids must be distinct."""
|
|
iface = _iface(tmp_path)
|
|
e1 = _submit(iface, objective="wave stage one")
|
|
e2 = _submit(iface, objective="wave stage two")
|
|
assert e1["success"] and e2["success"], (e1["errors"], e2["errors"])
|
|
assert e1["ids"]["contract_id"] != e2["ids"]["contract_id"]
|