src/rune/construction_interface.py

@@ -88,6 +88,39 @@ def _envelope(
     return env
 
 
+def _to_stage_check(spec: dict):
+    """One guard spec → a stage_guards.StageCheck (T5.15 wire)."""
+    from rune import stage_guards as sg
+
+    return sg.StageCheck(
+        name=str(spec.get("name", "guard")),
+        kind=str(spec.get("kind", "assert")),
+        argv=spec.get("argv"),
+        output_pattern=str(spec.get("output_pattern", "")),
+        path=str(spec.get("path", "")),
+        check=spec.get("check"),
+        description=str(spec.get("description", "")),
+    )
+
+
+def _guard_specs_from_obligations(c: dict) -> list[dict]:
+    """The contract's guard declarations: verification_obligations rows
+    shaped `guard:<json-spec>` (the decision path records them; anything
+    else stays a free-text obligation for the projection). JSONDecodeError
+    → skipped with no invented facts."""
+    specs: list[dict] = []
+    for item in (c.get("identities") or {}).get("verification_obligations", []) or []:
+        if not isinstance(item, str) or not item.startswith("guard:"):
+            continue
+        try:
+            spec = json.loads(item[6:])
+            if isinstance(spec, dict):
+                specs.append(spec)
+        except json.JSONDecodeError:
+            continue
+    return specs
+
+
 class ConstructionInterface:
     """The builder-facing construction lifecycle for ONE project target.
 
@@ -291,6 +324,13 @@ class ConstructionInterface:
                     "transition": {
                         "phase": {"type": "str", "required": True,
                                   "allowed": list(sc.PHASES)},
+                        "evidence_ref": {"type": "str", "required": False},
+                        "guard_checks": {
+                            "type": "list[dict] | None", "required": False,
+                            "notes": ("stage-guard specs run on the "
+                                      "completion edges (verifying/closed); "
+                                      "a failure BLOCKS with "
+                                      "VERIFICATION_FAILED — T5.15 wire")},
                     },
                     "close": {
                         "observed_evidence": {
@@ -607,8 +647,57 @@ class ConstructionInterface:
             warnings=warnings,
         )
 
-    def transition(self, contract_id: str, phase: str, *, evidence_ref: str = "") -> dict:
+    def transition(self, contract_id: str, phase: str, *, evidence_ref: str = "",
+                   guard_checks: list[dict] | None = None) -> dict:
+        """Move a contract phase. Completion-wave slice (T5.15, the
+        stage_guards wire): entering `verifying` and `closed` runs the
+        §15F/§15G guards — readiness/exit StageChecks and test-truth
+        verification — when the contract declares them
+        (identities.verification_obligations carries `guard:<spec>` rows,
+        or guard_checks passes inline specs).
+
+        guard_checks spec: {"name", "kind": "command"|"path_exists"|"assert",
+        "argv"/"path"/"output_pattern", "check": callable (in-process only),
+        "description"}. A guard failure is a VERIFICATION_FAILED envelope:
+        the phase does NOT move.
+
+        The legality map stays the scaffold's authority; guards ADD evidence
+        obligations on the completion-facing edges (building→verifying,
+        verifying→closed) — a tiny bounded change can pass guards=[] freely.
+        """
         doc = self.scaffold.load()
+        c = self.scaffold.contract(doc, contract_id)
+        if c is None:
+            return _envelope(
+                "transition",
+                success=False,
+                ids={"contract_id": contract_id},
+                errors=[{"error": Errors.UNKNOWN_CONTRACT, "detail": f"no contract {contract_id}"}],
+            )
+        if phase in ("verifying", "closed"):
+            from rune import stage_guards as sg
+
+            specs = guard_checks
+            if specs is None:
+                specs = _guard_specs_from_obligations(c)
+            if specs:
+                stage = sg.StageContract(
+                    stage_id=f"{contract_id}:{phase}",
+                    contract_id=contract_id,
+                    ready_when=[_to_stage_check(s) for s in specs],
+                    done_when=[_to_stage_check(s) for s in specs],
+                )
+                canary = sg.readiness_canary(stage)
+                if canary["verdict"] == "BLOCKED":
+                    return _envelope(
+                        "transition",
+                        success=False,
+                        ids={"contract_id": contract_id},
+                        state={"phase": c.get("phase"), "guard": canary},
+                        errors=[{"error": Errors.VERIFICATION_FAILED,
+                                 "detail": "stage guards blocked the transition",
+                                 "failures": canary["failures"]}],
+                    )
         try:
             self.scaffold.transition_contract(
                 doc, contract_id, phase, evidence_ref=evidence_ref or f"builder:{self.builder}"
@@ -1019,7 +1108,9 @@ def main(argv=None) -> int:
     if cmd == "verify":
         return _out(iface.verify(args[0], changed_files=(json.loads(args[1]) if len(args) > 1 else None)))
     if cmd == "transition":
-        return _out(iface.transition(args[0], args[1], evidence_ref=args[2] if len(args) > 2 else ""))
+        guard_checks = json.loads(args[3]) if len(args) > 3 else None
+        return _out(iface.transition(args[0], args[1], evidence_ref=args[2] if len(args) > 2 else "",
+                                     guard_checks=guard_checks))
     if cmd == "close":
         return _out(
             iface.close(

tests/test_construction_interface.py

@@ -585,3 +585,89 @@ def test_15u_discover_active_contract_migrated_to_interface(tmp_path):
     # 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"]