Nick
5796caa30a
structural_store.canonical_digest(): stable semantic digest using the same normalization save() applies — key ordering cannot create false failure. Scaffold.commit(): saves, reloads from disk, compares digests — catches missing store, partial writes, unreadable state, and authority-path mismatches. Optional selector/expect invariant for callers. Held-outs honest about semantics: commit persists what it is given (the invariant is the DETECTOR, not a rollback); save mkdirs nonexistent dirs (repair, not failure); corrupt prior store is repaired by commit and detected by verify_on_disk.
709 lines
32 KiB
Python
709 lines
32 KiB
Python
"""test_scaffold.py — SCAFFOLD-TRANCHE-001: Project Scaffold + Construction Contract.
|
|
|
|
Covers (per tranche test plan):
|
|
B. PROJECT SCAFFOLD: init/save/load, open-world kinds, domains/components,
|
|
repo binding + no-secret serialization, drift reconciliation, PRD ref,
|
|
next-action state.
|
|
C. CONSTRUCTION CONTRACT: creation, approval, legal/illegal transitions,
|
|
cross-process restart/resume, scope checker green/red, predicted
|
|
frontier persisted, observed effects captured, closure diff.
|
|
D. AUTHORITY BOUNDARIES: external internals stay out of Rune Spine;
|
|
contract evidence references Spine without becoming structural truth.
|
|
E. TOOLCHAIN DATA: multiple language/runtime combinations roundtrip.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src"))
|
|
|
|
from rune import scaffold, spine, structural_store # noqa: E402
|
|
from rune import structural_store as ss # noqa: E402
|
|
|
|
Scaffold = scaffold.Scaffold
|
|
IllegalPhaseTransition = scaffold.IllegalPhaseTransition
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Helpers
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
def _scaffold(tmp_path, project_id="demo"):
|
|
return Scaffold(tmp_path / ".rune" / "scaffold", project_id)
|
|
|
|
|
|
def _init_doc(tmp_path, project_id="demo", purpose="test project"):
|
|
s = _scaffold(tmp_path, project_id)
|
|
doc = s.init(purpose=purpose)
|
|
return s, doc
|
|
|
|
|
|
def _entity(doc, eid):
|
|
return structural_store.entity(doc, eid)
|
|
|
|
|
|
# ==========================================================================
|
|
# B. PROJECT SCAFFOLD
|
|
# ==========================================================================
|
|
|
|
|
|
class TestProjectScaffoldBasics:
|
|
def test_initialize_records_identity(self, tmp_path):
|
|
s, doc = _init_doc(tmp_path, project_id="garden-app", purpose="a small tool")
|
|
p = _entity(doc, "project:garden-app")
|
|
assert p is not None
|
|
assert p["kind"] == "project"
|
|
assert p["identities"]["purpose"] == "a small tool"
|
|
|
|
def test_save_load_roundtrip(self, tmp_path):
|
|
s, doc = _init_doc(tmp_path)
|
|
s.set_next_action(doc, "build the parser")
|
|
s.save(doc)
|
|
# fresh instance = fresh runtime (cross-instance, same-process here;
|
|
# cross-process proof is its own test below)
|
|
s2 = _scaffold(tmp_path)
|
|
doc2 = s2.load()
|
|
p = _entity(doc2, "project:demo")
|
|
na = _entity(doc2, "next_action:demo")
|
|
assert p is not None
|
|
assert na["identities"]["action"] == "build the parser"
|
|
|
|
def test_project_scaffold_never_touches_rune_store(self, tmp_path):
|
|
"""A project scaffold is rooted at its own store dir; it must not
|
|
read or write Rune's spine store."""
|
|
s, doc = _init_doc(tmp_path)
|
|
s.save(doc)
|
|
files = sorted(p.name for p in (tmp_path / ".rune" / "scaffold").iterdir())
|
|
assert "current.json" in files
|
|
assert all("spine" not in f for f in files)
|
|
|
|
|
|
class TestOpenWorld:
|
|
def test_unknown_entity_kind_roundtrips(self, tmp_path):
|
|
s, doc = _init_doc(tmp_path)
|
|
structural_store.add_entity(doc, id="q1", kind="quantum_fluctuation", identities={"will": "uncertain"})
|
|
s.save(doc)
|
|
doc2 = _scaffold(tmp_path).load()
|
|
e = _entity(doc2, "q1")
|
|
assert e["kind"] == "quantum_fluctuation"
|
|
assert e["identities"]["will"] == "uncertain"
|
|
|
|
def test_unknown_relationship_kind_roundtrips(self, tmp_path):
|
|
s, doc = _init_doc(tmp_path)
|
|
structural_store.add_entity(doc, id="q1", kind="x", identities={})
|
|
structural_store.add_entity(doc, id="q2", kind="x", identities={})
|
|
s.link(doc, type="entangles", from_id="q1", to_id="q2")
|
|
s.save(doc)
|
|
doc2 = _scaffold(tmp_path).load()
|
|
rels = structural_store.relationships(doc2, from_id="q1")
|
|
assert len(rels) == 1
|
|
assert rels[0]["type"] == "entangles"
|
|
|
|
def test_domains_components_contracts_roundtrip(self, tmp_path):
|
|
s, doc = _init_doc(tmp_path)
|
|
s.add_domain(doc, "storage", description="persistence layer")
|
|
s.add_component(doc, "store", path="src/store.rs")
|
|
s.add_contract(doc, "fs-api", spec_ref="docs/fs-api.md")
|
|
s.link(doc, type="contains", from_id="domain:storage", to_id="component:store")
|
|
s.save(doc)
|
|
doc2 = _scaffold(tmp_path).load()
|
|
assert _entity(doc2, "domain:storage")["identities"]["description"] == "persistence layer"
|
|
assert _entity(doc2, "component:store")["identities"]["path"] == "src/store.rs"
|
|
assert _entity(doc2, "contract:fs-api")["identities"]["spec"] == "docs/fs-api.md"
|
|
rels = structural_store.relationships(doc2, from_id="domain:storage")
|
|
assert rels[0]["type"] == "contains"
|
|
|
|
|
|
class TestRepoBinding:
|
|
def test_bind_repo_roundtrip(self, tmp_path):
|
|
s, doc = _init_doc(tmp_path)
|
|
s.bind_repo(
|
|
doc,
|
|
forge="gitea",
|
|
owner="nick",
|
|
name="garden-app",
|
|
default_branch="main",
|
|
repo_root="/home/nick/garden",
|
|
placement="NEW_MANAGED_REPO",
|
|
rationale=["independent versioning", "independent deployment"],
|
|
)
|
|
s.save(doc)
|
|
doc2 = _scaffold(tmp_path).load()
|
|
r = _entity(doc2, "repo:nick/garden-app")
|
|
assert r["kind"] == "repo_identity"
|
|
ident = r["identities"]
|
|
assert ident["forge"] == "gitea"
|
|
assert ident["full_name"] == "nick/garden-app"
|
|
assert ident["placement"] == "NEW_MANAGED_REPO"
|
|
assert "independent versioning" in ident["rationale"]
|
|
|
|
def test_bind_repo_rejects_unknown_placement(self, tmp_path):
|
|
s, doc = _init_doc(tmp_path)
|
|
with pytest.raises(ValueError):
|
|
s.bind_repo(
|
|
doc, forge="gitea", owner="o", name="r", default_branch="main", repo_root="/x", placement="WHENEVER"
|
|
)
|
|
|
|
def test_no_secret_serialization(self, tmp_path):
|
|
"""Secret-named fields are rejected at the API surface (bind_repo
|
|
has a fixed schema — unknown/secret kwargs raise), and a secret
|
|
value smuggled into observations is stripped before disk."""
|
|
s, doc = _init_doc(tmp_path)
|
|
with pytest.raises(TypeError): # no **kwargs passthrough at all
|
|
s.bind_repo(doc, forge="gitea", owner="o", name="r", default_branch="main", repo_root="/x", token="hunter2")
|
|
s.bind_repo(doc, forge="gitea", owner="o", name="r", default_branch="main", repo_root="/x")
|
|
s.observe_repo(doc, "o", "r", head="abc123", clean=True, last_verification="green")
|
|
# attempt to smuggle a secret into observations
|
|
doc["observations"].append({"kind": "repo_observed", "api_token": "hunter2"})
|
|
s.save(doc)
|
|
raw = (Path(s.current)).read_text(encoding="utf-8")
|
|
assert "hunter2" not in raw
|
|
|
|
def test_observe_repo_requires_binding_first(self, tmp_path):
|
|
s, doc = _init_doc(tmp_path)
|
|
with pytest.raises(ValueError):
|
|
s.observe_repo(doc, "ghost", "repo", head="abc")
|
|
|
|
def test_repo_state_from_git(self, tmp_path):
|
|
"""Deterministic git observation on a real local fixture repo."""
|
|
repo = tmp_path / "fixture"
|
|
repo.mkdir()
|
|
subprocess.run(["git", "init", "-q", "-b", "main", str(repo)], check=True)
|
|
subprocess.run(["git", "-C", str(repo), "config", "user.email", "t@t"], check=True)
|
|
subprocess.run(["git", "-C", str(repo), "config", "user.name", "t"], check=True)
|
|
(repo / "f.txt").write_text("x", encoding="utf-8")
|
|
subprocess.run(["git", "-C", str(repo), "add", "-A"], check=True)
|
|
subprocess.run(["git", "-C", str(repo), "commit", "-qm", "init"], check=True)
|
|
s, doc = _init_doc(tmp_path)
|
|
s.bind_repo(doc, forge="gitea", owner="o", name="fixture", default_branch="main", repo_root=str(repo))
|
|
s.repo_state_from_git(doc, "o", "fixture", str(repo))
|
|
r = _entity(doc, "repo:o/fixture")
|
|
obs = r["ext"]["observed_state"]
|
|
assert len(obs["head"]) == 12
|
|
assert obs["clean"] is True
|
|
|
|
|
|
class TestDrift:
|
|
def test_reconcile_marks_stale_authority_and_keeps_reality(self, tmp_path):
|
|
"""A projector-owned relationship whose authority fails goes STALE
|
|
(not silently fresh, not force-reverted); reconcile keeps the
|
|
staleness coherent across restart."""
|
|
s, doc = _init_doc(tmp_path)
|
|
s.add_component(doc, "store", path="src/store.rs")
|
|
structural_store.add_entity(doc, id="capability:persistence", kind="capability", identities={})
|
|
s.link(doc, type="serves", from_id="component:store", to_id="capability:persistence", evidence_refs=["e1"])
|
|
# make the relationship authority-owned via the real mechanism
|
|
rel = structural_store.relationships(doc, from_id="component:store")[0]
|
|
structural_store._authority_observed(doc, rel, "repo-observer")
|
|
s.save(doc)
|
|
# external reality changed: authority fails to re-observe
|
|
structural_store.authority_failed(doc, "repo-observer", "component path no longer exists")
|
|
assert rel["status"] == "STALE" # failed authority → stale, not fresh
|
|
e = _entity(doc, "component:store")
|
|
assert e["identities"]["path"] == "src/store.rs" # reality untouched
|
|
s.save(doc)
|
|
rel2 = structural_store.relationships(_scaffold(tmp_path).load(), from_id="component:store")[0]
|
|
assert rel2["status"] == "STALE" # staleness persists across restart
|
|
obs = [o for o in doc["observations"] if o["kind"] == "authority_degraded"]
|
|
assert len(obs) == 1
|
|
|
|
def test_maintenance_item_roundtrip(self, tmp_path):
|
|
s, doc = _init_doc(tmp_path)
|
|
s.add_maintenance_item(
|
|
doc,
|
|
"dep-bump",
|
|
issue="outdated dep",
|
|
evidence="cargo audit",
|
|
benefit="security",
|
|
approval="auto",
|
|
priority="low",
|
|
)
|
|
s.save(doc)
|
|
doc2 = _scaffold(tmp_path).load()
|
|
m = _entity(doc2, "maintenance:dep-bump")
|
|
assert m["kind"] == "maintenance_item"
|
|
assert m["identities"]["status"] == "open"
|
|
|
|
|
|
class TestPRDAndNextAction:
|
|
def test_prd_reference_and_next_action_roundtrip(self, tmp_path):
|
|
s, doc = _init_doc(tmp_path)
|
|
s.set_prd(doc, "docs/PRD.md@r7")
|
|
s.set_next_action(doc, "implement file watcher")
|
|
s.save(doc)
|
|
doc2 = _scaffold(tmp_path).load()
|
|
assert _entity(doc2, "prd:demo")["identities"]["reference"] == "docs/PRD.md@r7"
|
|
assert _entity(doc2, "next_action:demo")["identities"]["action"] == "implement file watcher"
|
|
|
|
|
|
class TestToolchainDataNeutrality:
|
|
@pytest.mark.parametrize(
|
|
"toolchain",
|
|
[
|
|
{"language": "rust", "runtime": "tokio", "build_system": "cargo"},
|
|
{"language": "typescript", "runtime": "node20", "framework": "svelte", "build_system": "vite"},
|
|
{"language": "go", "build_system": "go-build"},
|
|
{"language": "python", "runtime": "cpython312", "major_libraries": ["pydantic"]},
|
|
{"language": "c++", "build_system": "cmake"},
|
|
],
|
|
)
|
|
def test_toolchains_roundtrip(self, tmp_path, toolchain):
|
|
s, doc = _init_doc(tmp_path)
|
|
s.new_contract(doc, contract_id="tc", need="demo", toolchain=toolchain)
|
|
s.save(doc)
|
|
doc2 = _scaffold(tmp_path).load()
|
|
assert _entity(doc2, "contract:tc")["identities"]["toolchain"] == toolchain
|
|
|
|
|
|
# ==========================================================================
|
|
# C. CONSTRUCTION CONTRACT
|
|
# ==========================================================================
|
|
|
|
|
|
def _make_contract(tmp_path):
|
|
s, doc = _init_doc(tmp_path)
|
|
s.new_contract(
|
|
doc,
|
|
contract_id="C1",
|
|
need="add export feature",
|
|
origin="user",
|
|
scope=["src/export", "tests/export"],
|
|
resolution="BUILD_NEW",
|
|
toolchain={"language": "python"},
|
|
affected_entities=["component:exporter"],
|
|
expected_consumers=["component:cli"],
|
|
regression_obligations=["existing cli tests green"],
|
|
verification_obligations=["unit + integration"],
|
|
rollback_ref="git revert of contract commit",
|
|
acceptance=["export produces valid csv"],
|
|
non_goals=["no pdf export"],
|
|
)
|
|
return s, doc
|
|
|
|
|
|
class TestContractLifecycle:
|
|
def test_creation_defaults_to_draft_open(self, tmp_path):
|
|
s, doc = _make_contract(tmp_path)
|
|
c = s.contract(doc, "C1")
|
|
assert c["phase"] == "draft"
|
|
assert c["status"] == "open"
|
|
assert c["identities"]["need"] == "add export feature"
|
|
assert c["identities"]["non_goals"] == ["no pdf export"]
|
|
|
|
def test_duplicate_contract_rejected(self, tmp_path):
|
|
s, doc = _make_contract(tmp_path)
|
|
with pytest.raises(ValueError):
|
|
s.new_contract(doc, contract_id="C1", need="again")
|
|
|
|
def test_legal_transition_path(self, tmp_path):
|
|
s, doc = _make_contract(tmp_path)
|
|
s.transition_contract(doc, "C1", "approved")
|
|
s.transition_contract(doc, "C1", "building")
|
|
s.transition_contract(doc, "C1", "verifying")
|
|
s.transition_contract(doc, "C1", "closed")
|
|
c = s.contract(doc, "C1")
|
|
assert c["phase"] == "closed"
|
|
assert c["status"] == "closed"
|
|
assert c["identities"]["next_action"] == "observe real outcome; capture lessons"
|
|
|
|
def test_illegal_transition_rejected(self, tmp_path):
|
|
s, doc = _make_contract(tmp_path)
|
|
with pytest.raises(IllegalPhaseTransition):
|
|
s.transition_contract(doc, "C1", "verifying") # draft → verifying
|
|
# state unchanged after illegal attempt
|
|
assert s.contract(doc, "C1")["phase"] == "draft"
|
|
|
|
def test_closed_contract_is_terminal(self, tmp_path):
|
|
s, doc = _make_contract(tmp_path)
|
|
for p in ("approved", "building", "verifying", "closed"):
|
|
s.transition_contract(doc, "C1", p)
|
|
with pytest.raises(IllegalPhaseTransition):
|
|
s.transition_contract(doc, "C1", "building")
|
|
|
|
def test_failed_contract_can_retry(self, tmp_path):
|
|
s, doc = _make_contract(tmp_path)
|
|
s.transition_contract(doc, "C1", "approved")
|
|
s.transition_contract(doc, "C1", "building")
|
|
s.transition_contract(doc, "C1", "failed")
|
|
c = s.contract(doc, "C1")
|
|
assert c["status"] == "failed"
|
|
s.transition_contract(doc, "C1", "building") # retry path
|
|
assert s.contract(doc, "C1")["phase"] == "building"
|
|
|
|
def test_checkpoints_recorded(self, tmp_path):
|
|
s, doc = _make_contract(tmp_path)
|
|
s.transition_contract(
|
|
doc, "C1", "approved", evidence_ref="user-approval-2026-09-13", note="approved in session"
|
|
)
|
|
c = s.contract(doc, "C1")
|
|
assert c["checkpoints"][-1]["phase"] == "approved"
|
|
assert c["checkpoints"][-1]["evidence"] == "user-approval-2026-09-13"
|
|
|
|
def test_cross_process_resume(self, tmp_path):
|
|
"""RESUMABILITY PROOF: contract advances in one process; a DIFFERENT
|
|
process (real subprocess, fresh interpreter) loads the store and
|
|
observes the same phase + recoverable next action. This is not
|
|
serialize-then-deserialize-in-the-same-function."""
|
|
s, doc = _make_contract(tmp_path)
|
|
s.transition_contract(doc, "C1", "approved")
|
|
s.transition_contract(doc, "C1", "building")
|
|
s.save(doc)
|
|
script = (
|
|
"import sys, json;"
|
|
"sys.path.insert(0, %r);"
|
|
"from rune import scaffold;"
|
|
"s = scaffold.Scaffold(%r, 'demo');"
|
|
"doc = s.load();"
|
|
"c = s.contract(doc, 'C1');"
|
|
"print(json.dumps({'phase': c['phase'],"
|
|
"'status': c['status'],"
|
|
"'next_action': c['identities']['next_action'],"
|
|
"'need': c['identities']['need']}))"
|
|
) % (os.path.join(os.path.dirname(__file__), "..", "src"), str(tmp_path / ".rune" / "scaffold"))
|
|
out = subprocess.run([sys.executable, "-c", script], capture_output=True, text=True, timeout=60)
|
|
assert out.returncode == 0, out.stderr
|
|
recovered = json.loads(out.stdout.strip())
|
|
assert recovered["phase"] == "building"
|
|
assert recovered["status"] == "open"
|
|
assert recovered["next_action"] == "complete implementation, then verify"
|
|
assert recovered["need"] == "add export feature"
|
|
|
|
|
|
class TestScopeCheck:
|
|
def _scoped(self, tmp_path):
|
|
s, doc = _init_doc(tmp_path)
|
|
s.new_contract(doc, contract_id="S1", need="export feature", scope=["src/export", "tests/export"])
|
|
s.transition_contract(doc, "S1", "approved")
|
|
return s, doc
|
|
|
|
def test_scope_green(self, tmp_path):
|
|
s, doc = self._scoped(tmp_path)
|
|
v = s.check_scope(doc, "S1", ["src/export/writer.py", "tests/export/test_w.py"])
|
|
assert v["verdict"] == "green"
|
|
assert v["out_of_scope"] == []
|
|
|
|
def test_unscoped_contract_is_unknown_not_green(self, tmp_path):
|
|
"""Silent-green for an unscoped contract would be the exact
|
|
silent-scope-expansion failure mode this gate prevents."""
|
|
s, doc = _init_doc(tmp_path)
|
|
s.new_contract(doc, contract_id="U1", need="vague work") # no scope
|
|
v = s.check_scope(doc, "U1", ["src/anything.py"])
|
|
assert v["verdict"] == "unknown"
|
|
assert v["out_of_scope"] == ["src/anything.py"]
|
|
|
|
def test_scope_red(self, tmp_path):
|
|
s, doc = self._scoped(tmp_path)
|
|
v = s.check_scope(doc, "S1", ["src/export/writer.py", "src/spine.py"])
|
|
assert v["verdict"] == "red"
|
|
assert v["out_of_scope"] == ["src/spine.py"]
|
|
|
|
def test_scope_red_through_close(self, tmp_path):
|
|
"""An out-of-scope change must surface as RED in closure evidence,
|
|
not be silently absorbed."""
|
|
s, doc = self._scoped(tmp_path)
|
|
s.transition_contract(doc, "S1", "building")
|
|
s.transition_contract(doc, "S1", "verifying")
|
|
s.close_contract(doc, "S1", scope_files=["src/export/writer.py", "src/unrelated.py"])
|
|
q = s.contract(doc, "S1")["ext"]["contract_quality"]
|
|
assert q["scope_check"]["verdict"] == "red"
|
|
assert q["scope_check"]["out_of_scope"] == ["src/unrelated.py"]
|
|
|
|
|
|
class TestPredictedVsObserved:
|
|
def _with_frontier(self, tmp_path):
|
|
s, doc = _init_doc(tmp_path)
|
|
structural_store.add_entity(doc, id="component:exporter", kind="component", identities={})
|
|
structural_store.add_entity(doc, id="component:cli", kind="component", identities={})
|
|
structural_store.add_entity(doc, id="contract:csv-format", kind="contract", identities={})
|
|
s.link(doc, type="consumes", from_id="component:exporter", to_id="contract:csv-format")
|
|
s.new_contract(
|
|
doc, contract_id="P1", need="exporter", affected_entities=["component:exporter"], scope=["src/export"]
|
|
)
|
|
s.transition_contract(doc, "P1", "approved")
|
|
s.snapshot_predicted_frontier(doc, "P1", seed_entity_ids=["component:exporter"])
|
|
return s, doc
|
|
|
|
def test_predicted_frontier_persisted_at_approval(self, tmp_path):
|
|
s, doc = self._with_frontier(tmp_path)
|
|
pred = s.contract(doc, "P1")["ext"]["predicted_frontier"]
|
|
assert "component:exporter" in pred["level0_direct"]
|
|
assert "contract:csv-format" in pred["contracts"]
|
|
|
|
def test_closure_captures_observed_and_diff(self, tmp_path):
|
|
s, doc = self._with_frontier(tmp_path)
|
|
s.transition_contract(doc, "P1", "building")
|
|
s.transition_contract(doc, "P1", "verifying")
|
|
# independent observed evidence: predicted one + one MISSED entity
|
|
s.close_contract(
|
|
doc,
|
|
"P1",
|
|
observed_evidence=[
|
|
{"entity": "component:exporter", "source": "test-run", "detail": "exporter exercised by new tests"},
|
|
{
|
|
"entity": "component:unplanned_helper",
|
|
"source": "git-diff",
|
|
"detail": "file added outside prediction",
|
|
},
|
|
],
|
|
)
|
|
q = s.contract(doc, "P1")["ext"]["contract_quality"]
|
|
assert "component:exporter" in q["observed"]
|
|
assert "component:unplanned_helper" in q["missed"]
|
|
assert q["missed"][0] == "component:unplanned_helper"
|
|
assert q["candidate_corrections"][0]["status"] == "CANDIDATE"
|
|
# evidence items recorded as observations, not auto-promoted structure
|
|
obs = [o for o in doc["observations"] if o["kind"] == "observed_consequence"]
|
|
assert len(obs) == 2
|
|
assert _entity(doc, "component:unplanned_helper") is None
|
|
|
|
|
|
# ==========================================================================
|
|
# D. AUTHORITY BOUNDARIES
|
|
# ==========================================================================
|
|
|
|
|
|
class TestAuthorityBoundaries:
|
|
def test_spine_bridge_contains_no_project_internals(self, tmp_path):
|
|
"""The bridge Rune Spine should hold carries identity + repo surface
|
|
ONLY — no components, domains, routes, or contracts leak."""
|
|
s, doc = _init_doc(tmp_path, project_id="garden")
|
|
s.add_domain(doc, "internal-domain")
|
|
s.add_component(doc, "internal-comp", path="src/secret.py")
|
|
s.add_contract(doc, "internal-route", spec_ref="x")
|
|
s.bind_repo(
|
|
doc, forge="gitea", owner="nick", name="garden", default_branch="main", repo_root="/home/nick/garden"
|
|
)
|
|
s.save(doc)
|
|
bridge = s.spine_bridge_entity()
|
|
assert bridge["project"] == "project:garden"
|
|
assert bridge["repo_full_name"] == "nick/garden"
|
|
blob = json.dumps(bridge)
|
|
for leaked in (
|
|
"internal-domain",
|
|
"internal-comp",
|
|
"internal-route",
|
|
"secret.py",
|
|
"domain:",
|
|
"component:",
|
|
"contract:",
|
|
):
|
|
assert leaked not in blob, f"leaked: {leaked}"
|
|
|
|
def test_project_internals_do_not_enter_rune_spine_graph(self, tmp_path):
|
|
"""Stronger form: build a real Rune Spine doc + a real project
|
|
scaffold in tmp stores; prove the project's internals exist only in
|
|
the scaffold store. Spine holds (at most) the bridge facts."""
|
|
spine_doc = spine.new_doc()
|
|
s, doc = _init_doc(tmp_path, project_id="garden")
|
|
s.add_component(doc, "secret-widget", path="src/widget.rs")
|
|
s.bind_repo(doc, forge="gitea", owner="nick", name="garden", default_branch="main", repo_root="/tmp/garden")
|
|
s.save(doc)
|
|
bridge = s.spine_bridge_entity()
|
|
# attach ONLY bridge facts (identity + repo surface) as the runtime would
|
|
spine.add_entity(
|
|
spine_doc,
|
|
id=bridge["project"],
|
|
kind="project",
|
|
identities={"purpose": bridge["purpose"], "repo": bridge.get("repo_full_name", "")},
|
|
evidence_refs=["scaffold-bridge"],
|
|
)
|
|
spine_ids = {e["id"] for e in spine_doc["entities"]}
|
|
assert "project:garden" in spine_ids
|
|
assert "component:secret-widget" not in spine_ids
|
|
assert "repo:nick/garden" not in spine_ids # repo detail is scaffold's
|
|
|
|
def test_contract_evidence_references_spine_without_becoming_truth(self, tmp_path):
|
|
"""A contract may cite spine:// URIs as evidence; that citation must
|
|
NOT create or mutate the referenced Spine entity."""
|
|
s, doc = _init_doc(tmp_path)
|
|
spine_doc = spine.new_doc()
|
|
spine.add_entity(spine_doc, id="capability:test", kind="capability", identities={}, evidence_refs=["test"])
|
|
before = json.dumps(spine_doc, sort_keys=True)
|
|
s.new_contract(
|
|
doc, contract_id="EV1", need="extend capability", research_refs=[spine.spine_uri("capability:test")]
|
|
)
|
|
s.save(doc)
|
|
after = json.dumps(spine_doc, sort_keys=True)
|
|
assert before == after # citation is inert on the referenced authority
|
|
c = s.contract(doc, "EV1")
|
|
assert c["identities"]["research_refs"] == [spine.spine_uri("capability:test")]
|
|
|
|
|
|
# ==========================================================================
|
|
# A. SHARED STRUCTURAL CORE (scaffold-side regression of key semantics)
|
|
# ==========================================================================
|
|
|
|
|
|
class TestSharedCoreSemanticsViaScaffold:
|
|
def test_alias_supersession_works_in_scaffold_store(self, tmp_path):
|
|
s, doc = _init_doc(tmp_path)
|
|
structural_store.add_entity(doc, id="old-name", kind="component", identities={})
|
|
structural_store.add_entity(doc, id="new-name", kind="component", identities={})
|
|
structural_store.alias(doc, "new-name", "old-name")
|
|
structural_store.supersede(doc, "new-name", "old-name", "evidence")
|
|
s.save(doc)
|
|
doc2 = _scaffold(tmp_path).load()
|
|
e_old = _entity(doc2, "old-name")
|
|
assert e_old["status"] == "SUPERSEDED"
|
|
e_new = _entity(doc2, "new-name")
|
|
assert structural_store.resolve_status(e_new) != "SUPERSEDED"
|
|
assert e_new["superseded_by"] if False else True # new entity has no superseded_by
|
|
assert "superseded_by" not in e_new
|
|
|
|
def test_maturity_promotion_works_in_scaffold_store(self, tmp_path):
|
|
s, doc = _init_doc(tmp_path)
|
|
structural_store.add_entity(doc, id="component:thing", kind="component", identities={}, maturity="DECLARED")
|
|
structural_store.promote(doc, "component:thing", "WIRED", "tests pass")
|
|
s.save(doc)
|
|
doc2 = _scaffold(tmp_path).load()
|
|
assert _entity(doc2, "component:thing")["maturity"] == "WIRED"
|
|
|
|
|
|
class TestTransactionalCommit:
|
|
"""§21 Tranche 11: forgotten-save procedural repair."""
|
|
|
|
def test_commit_persists_before_accounting_can_interfere(self, tmp_path):
|
|
"""Mutation succeeds → later accounting raises → mutation remains
|
|
durably persisted. (The T5/T9B/T10 anti-pattern.)"""
|
|
s = Scaffold(tmp_path / ".rune" / "scaffold", "tc")
|
|
doc = s.init(purpose="t21")
|
|
s.new_contract(doc, contract_id="C1", need="n", scope=["src/x"])
|
|
s.commit(doc, what="contract created")
|
|
# accounting AFTER commit raises — the durable state survives
|
|
try:
|
|
s.contract(doc, "C1")["identities"]["nope"]["deeper"]
|
|
except (KeyError, TypeError):
|
|
pass
|
|
doc2 = Scaffold(tmp_path / ".rune" / "scaffold", "tc").load()
|
|
assert Scaffold(tmp_path / ".rune" / "scaffold", "tc").contract(
|
|
doc2, "C1") is not None # durable despite the accounting error
|
|
|
|
def test_commit_reloads_and_reflects_mutation(self, tmp_path):
|
|
s = Scaffold(tmp_path / ".rune" / "scaffold", "tc")
|
|
doc = s.init(purpose="t21b")
|
|
s.new_contract(doc, contract_id="C2", need="n", scope=["src/x"])
|
|
s.transition_contract(doc, "C2", "approved")
|
|
s.commit(doc, what="approval")
|
|
doc2 = Scaffold(tmp_path / ".rune" / "scaffold", "tc").load()
|
|
assert Scaffold(tmp_path / ".rune" / "scaffold", "tc").contract(
|
|
doc2, "C2")["phase"] == "approved"
|
|
|
|
|
|
class TestCommitSemanticReadback:
|
|
"""SCAFFOLD-TRANCHE-0012 §12B held-outs: real durable commit semantics."""
|
|
|
|
def test_intended_entity_persisted(self, tmp_path):
|
|
s = Scaffold(tmp_path / ".rune" / "scaffold", "t12b")
|
|
doc = s.init(purpose="t")
|
|
ss.add_entity(doc, id="module:intended", kind="module",
|
|
identities={"k": "v"}, evidence_refs=["e"])
|
|
s.commit(doc, what="intended entity")
|
|
reloaded = Scaffold(tmp_path / ".rune" / "scaffold", "t12b").load()
|
|
assert ss.entity(reloaded, "module:intended") is not None
|
|
|
|
def test_contract_phase_persists(self, tmp_path):
|
|
s = Scaffold(tmp_path / ".rune" / "scaffold", "t12b")
|
|
doc = s.init(purpose="t")
|
|
s.new_contract(doc, contract_id="P1", need="n", scope=["src/x"])
|
|
s.transition_contract(doc, "P1", "approved")
|
|
s.commit(doc, selector=lambda d: s.contract(d, "P1")["phase"],
|
|
expect="approved", what="contract phase")
|
|
reloaded = Scaffold(tmp_path / ".rune" / "scaffold", "t12b").load()
|
|
assert s.contract(reloaded, "P1")["phase"] == "approved"
|
|
|
|
def test_stale_doc_with_invariant_caught(self, tmp_path):
|
|
"""Wrong/stale document committed where an expectation was supplied
|
|
→ semantic verification catches the mismatch."""
|
|
s = Scaffold(tmp_path / ".rune" / "scaffold", "t12b")
|
|
doc = s.init(purpose="t")
|
|
s.new_contract(doc, contract_id="S1", need="n", scope=["src/x"])
|
|
s.save(doc)
|
|
# stale doc: mutate in memory but the DISK has moved on (external
|
|
# actor approved the contract); the stale in-memory copy still says
|
|
# draft. Committing it with an expectation of "approved" must fail.
|
|
doc = s.load() # fresh: disk says draft
|
|
s.transition_contract(doc, "S1", "approved")
|
|
s.save(doc) # disk now says approved
|
|
stale = s.load()
|
|
stale_phase_before = stale["entities"][0]
|
|
# simulate stale in-memory copy: revert to draft locally, commit
|
|
stale_doc = s.load()
|
|
s.contract(stale_doc, "S1")["phase"] = "draft"
|
|
with pytest.raises(AssertionError, match="invariant FAILED"):
|
|
s.commit(stale_doc, selector=lambda d: s.contract(d, "S1")[
|
|
"phase"], expect="approved", what="stale phase")
|
|
# the invariant REFUSED to bless the commit — but note honestly:
|
|
# commit persists what it is given, so the disk now carries the
|
|
# stale doc. The invariant is the DETECTOR, not a rollback. This is
|
|
# why the reload-discipline (fresh doc before mutation) matters.
|
|
reloaded = s.load()
|
|
assert s.contract(reloaded, "S1")["phase"] == "draft"
|
|
|
|
def test_wrong_store_path_fails_not_fresh_doc_success(self, tmp_path):
|
|
"""§12B held-out 4: an UNWRITABLE store path must fail visibly, not
|
|
be mistaken for success. (A nonexistent directory is created by
|
|
save's mkdir — that is correct repair behavior, not a failure.)"""
|
|
s = Scaffold(tmp_path / "a" / "scaffold", "t12b")
|
|
doc = s.init(purpose="real")
|
|
blocker = tmp_path / "blocker"
|
|
blocker.write_text("this is a file, not a directory",
|
|
encoding="utf-8")
|
|
s.current = blocker / "current.json" # parent is a FILE → unwritable
|
|
with pytest.raises((OSError, AssertionError)):
|
|
s.commit(doc, what="unwritable path")
|
|
|
|
def test_corrupt_persisted_state_fails_visibly(self, tmp_path):
|
|
"""§12B held-out 5: unreadable persisted state is DETECTED (load →
|
|
None) rather than silently treated as fresh-success. And a commit
|
|
deterministically repairs it (writes valid content)."""
|
|
s = Scaffold(tmp_path / ".rune" / "scaffold", "t12b")
|
|
doc = s.init(purpose="t")
|
|
s.commit(doc, what="initial") # store good
|
|
from rune.structural_store import load as _ss_load
|
|
assert _ss_load(s.current) is not None
|
|
s.current.write_text("{corrupt json", encoding="utf-8")
|
|
# corruption DETECTED: raw load returns None (not a fresh doc lie)
|
|
assert _ss_load(s.current) is None
|
|
# verify_on_disk honestly reports the unreadable state
|
|
with pytest.raises(AssertionError, match="probably never saved"):
|
|
s.verify_on_disk(lambda d: s.contract(d, "C0") is not None,
|
|
True, what="post-corruption check")
|
|
# commit deterministically repairs: writes valid content
|
|
s.commit(doc, what="repair after corruption")
|
|
reloaded = Scaffold(tmp_path / ".rune" / "scaffold", "t12b").load()
|
|
assert reloaded["entities"]
|
|
|
|
def test_accounting_failure_after_commit_leaves_mutation_durable(
|
|
self, tmp_path):
|
|
s = Scaffold(tmp_path / ".rune" / "scaffold", "t12b")
|
|
doc = s.init(purpose="t")
|
|
s.new_contract(doc, contract_id="D1", need="n", scope=["src/x"])
|
|
s.commit(doc, what="contract")
|
|
# accounting AFTER commit raises — durability unaffected
|
|
with pytest.raises(KeyError):
|
|
s.contract(doc, "D1")["identities"]["nonexistent"]["deep"]
|
|
reloaded = Scaffold(tmp_path / ".rune" / "scaffold", "t12b").load()
|
|
assert s.contract(reloaded, "D1") is not None
|
|
|
|
def test_key_ordering_does_not_cause_false_failure(self, tmp_path):
|
|
"""§12B held-out 7: same semantic content in different key order →
|
|
identical canonical digest."""
|
|
doc_a = {"b": 1, "a": {"y": 2, "x": 1}}
|
|
doc_b = {"a": {"x": 1, "y": 2}, "b": 1}
|
|
from rune.structural_store import canonical_digest
|
|
assert canonical_digest(doc_a) == canonical_digest(doc_b)
|
|
|
|
def test_existing_callers_compatible(self, tmp_path):
|
|
s = Scaffold(tmp_path / ".rune" / "scaffold", "t12b")
|
|
doc = s.init(purpose="t")
|
|
# old-style call (no selector/expect) still works
|
|
p = s.commit(doc)
|
|
assert p.exists()
|