feat(programme_continuation): port to FULL Rust — dichotomy wave

FULL parity port of src/rune/programme_continuation.py (349 lines, 11 fns,
dichotomy wave #2479 tranche). The kernel owns ALL of it: the 15M
continuation canary (all 7 verdicts), eligible_stages, stage to_dict,
save-doc build, load, run_programme, _programme_base, the
scaffolding stage runner (draft→approved→building→verifying→closed with
truthful failed-path checkpoints), and run_construction_programme.

The seams the oracle leaves to the caller cross the FFI as JSON ops
served by pc_loader.OpsProxy against the live Python objects: the
Scaffolding interface (load/save/contract/transition/close), the event
bridge, the work callable, the stage runner, and the clock. Doc-envelope
FFI; no free-string export (cs_free_string via capability_scoring).
Facade delegation with RUNE_PC_CORE=python pin (strangler law).

Frozen truth: 46 cases vs the LIVE oracle (RUNE_PC_CORE=python pinned)
covering every function and branch, embedded verbatim
(frozen_pc_truth.json) and replayed by the cfg(test) branch battery —
interface-driven cases replay the oracle's RECORDED OPS verbatim
(name+args+order). Battery caught the T5.15 programme-continuation wire
inside the real close() (host-side effects tagged host_internal).

Verified: battery 8/8 green, clippy 0, live parity 546/546 (46 truth
cases re-executed oracle-vs-kernel + 500 fuzz, seed 2479), Python tests
65/65 green (programme continuation/wiring/gates/interface/event_wait).

src/rune/pc_loader.py

@@ -0,0 +1,337 @@
+#!/usr/bin/env python3
+"""pc_loader.py — ctypes loader for the Rust programme_continuation core
+(dichotomy wave, #2479 tranche). Mirrors ci_loader.py's contract:
+
+FFI contract (doc-envelope law):
+  * pc_canary({prog}) -> {"verdict", "next_action", "reason"}
+  * pc_eligible({prog}) -> ["stage_id", ...] priority-first
+  * pc_stage_to_dict({stage}) -> durable stage dict
+  * pc_save_doc({prog, saved_at}) -> durable programme record doc
+  * pc_load({doc | null}) -> programme dict or null
+  * pc_run_programme({prog, max_transitions}, op_host)
+        -> {"log", "states"}
+  * pc_stage_run({cfg, stage}, op_host) -> {"state", "result"}
+  * pc_run_construction({prog, cfg, max_transitions}, op_host)
+        -> {"log", "states", "results"}
+
+The drivers run FULL Rust; the seams the oracle leaves to the caller —
+the Scaffolding interface, the event bridge, the work callable, the
+stage runner, and the clock — cross back as JSON ops served by
+OpsProxy against the live Python objects. No free-string export here —
+callers free through capability_scoring's cs_free_string.
+
+Override the .so with PC_RUST_LIB (a FILE path).
+"""
+from __future__ import annotations
+
+import copy
+import ctypes
+import json
+import os
+from pathlib import Path
+
+_REPO = Path(__file__).resolve().parents[2]
+_CRATE_DIR = _REPO / "src" / "rust" / "programme_continuation"
+_LIB_PATH = Path(os.environ.get("PC_RUST_LIB",
+                                _CRATE_DIR / "target" / "release" /
+                                "libprogramme_continuation.so"))
+
+_lib = None
+
+# Rust OpFn: extern "C" fn(op: *const c_char, user_data) -> *mut c_char
+OP_CB = ctypes.CFUNCTYPE(ctypes.c_void_p, ctypes.c_char_p, ctypes.c_void_p)
+
+
+class ProgrammeContinuationCoreUnavailable(FileNotFoundError):
+    """The Rust programme-continuation core is not built. Callers fall
+    back to the Python oracle."""
+
+
+def available() -> bool:
+    return _LIB_PATH.is_file()
+
+
+def _load() -> ctypes.CDLL:
+    global _lib
+    if _lib is not None:
+        return _lib
+    if not _LIB_PATH.is_file():
+        raise ProgrammeContinuationCoreUnavailable(str(_LIB_PATH))
+    lib = ctypes.CDLL(str(_LIB_PATH))
+    for name in ("pc_run_programme", "pc_stage_run", "pc_run_construction"):
+        fn = getattr(lib, name)
+        fn.argtypes = [ctypes.c_void_p, OP_CB, ctypes.c_void_p]
+        fn.restype = ctypes.c_void_p
+    for name in ("pc_canary", "pc_eligible", "pc_stage_to_dict",
+                 "pc_save_doc", "pc_load"):
+        fn = getattr(lib, name)
+        fn.argtypes = [ctypes.c_void_p]
+        fn.restype = ctypes.c_void_p
+    _lib = lib
+    return lib
+
+
+def _free(ptr) -> None:
+    # free via any sibling free symbol already in-process (capability_scoring)
+    try:
+        scoring = ctypes.CDLL(str(
+            _REPO / "src" / "rust" / "capability_scoring" / "target" / "release" /
+            "libcapability_scoring.so"))
+        scoring.cs_free_string.argtypes = [ctypes.c_void_p]
+        scoring.cs_free_string(ptr)
+    except OSError:
+        pass  # leak-on-fallback only if capability_scoring absent; bounded
+
+
+def _unwrap(ptr) -> dict:
+    if not ptr:
+        raise RuntimeError("pc core returned null")
+    out = ctypes.string_at(ptr).decode()
+    _free(ptr)
+    doc = json.loads(out)
+    if not doc.get("ok"):
+        raise RuntimeError(doc.get("err", "pc core failed"))
+    return doc["value"]
+
+
+def _call_simple(name: str, payload: dict):
+    lib = _load()
+    buf = ctypes.create_string_buffer(json.dumps(payload).encode())
+    ptr = getattr(lib, name)(ctypes.cast(buf, ctypes.c_void_p))
+    return _unwrap(ptr)
+
+
+def _call_host(name: str, payload: dict, host):
+    """Drive a host-op FFI export: the kernel issues JSON ops, `host`
+    serves each and the result returns through the envelope."""
+    lib = _load()
+    buf = ctypes.create_string_buffer(json.dumps(payload).encode())
+    keep = []
+
+    def trampoline(op_ptr, _user_data):
+        try:
+            op = json.loads(ctypes.string_at(op_ptr).decode())
+            result = host(op)
+        except Exception as exc:  # noqa: BLE001 — surfaced by the kernel
+            result = {"__host_error__": str(exc)[:500]}
+        data = json.dumps(_jsonable(result)).encode()
+        b = ctypes.create_string_buffer(data)
+        keep[:] = [b]  # previous buffer already consumed; keep current alive
+        return ctypes.cast(b, ctypes.c_void_p).value
+
+    cb = OP_CB(trampoline)
+    ptr = getattr(lib, name)(ctypes.cast(buf, ctypes.c_void_p), cb, None)
+    return _unwrap(ptr)
+
+
+def _jsonable(v):
+    return json.loads(json.dumps(v, default=str))
+
+
+def _entity_by_id(doc: dict, entity_id: str) -> dict | None:
+    for e in (doc or {}).get("entities", []):
+        if e.get("id") == entity_id:
+            return e
+    return None
+
+
+class OpsProxy:
+    """Serves the kernel's JSON ops against the live Python objects —
+    the mirror image of the truth generator's recording proxy."""
+
+    def __init__(self, iface=None, stages: dict | None = None, runner=None):
+        self.iface = iface
+        self.stages = stages or {}
+        self.runner = runner
+        self._doc = None
+
+    def __call__(self, op: dict):
+        name = op.get("op")
+        a = op.get("args") or {}
+        if name == "scaffold_load":
+            self._doc = self.iface.scaffold.load()
+            return copy.deepcopy(self._doc)
+        if name == "contract":
+            c = _entity_by_id(self._doc, f"contract:{a['contract_id']}") \
+                if self._doc else None
+            return {"contract": copy.deepcopy(c) if c else None,
+                    "entity_id": (c or {}).get("id")}
+        if name == "scaffold_save":
+            return _jsonable(self.iface.scaffold.save(a["doc"]))
+        if name == "transition":
+            return _jsonable(self.iface.transition(
+                a["contract_id"], a["phase"],
+                evidence_ref=a.get("evidence_ref", "")))
+        if name == "close":
+            return _jsonable(self.iface.close(
+                a["contract_id"], observed_evidence=a["observed_evidence"],
+                scope_files=a.get("scope_files")))
+        if name == "work":
+            stage = self.stages[a["stage_id"]]
+            w = stage.result.get("work")
+            outcome = dict(w(self.iface, stage)) if callable(w) else dict(w)
+            return {"outcome": _jsonable(outcome)}
+        if name == "now":
+            from rune import programme_continuation as pc_mod
+            return {"now": pc_mod.time.strftime("%Y-%m-%dT%H:%M:%SZ",
+                                                pc_mod.time.gmtime()),
+                    "epoch": pc_mod.time.time()}
+        if name == "register_process_wait":
+            from rune import event_wait as ew
+            ew.register_process_wait(
+                a["op"], contract_id=a["contract_id"], project=a["project"],
+                objective=a["objective"], watched_for=a["watched_for"],
+                watchdog_secs=a["watchdog_secs"])
+            return None
+        if name == "run_builder_under_events":
+            from rune import event_wait as ew
+            notifier = ew.run_builder_under_events(
+                a["async_cmd"], a["op"], daemon_url=a["daemon_url"])
+            return {"pid": notifier.pid}
+        if name == "await_resume_event":
+            from rune import event_wait as ew
+            return _jsonable(ew.await_resume_event(
+                a["op"], watchdog_secs=a["watchdog_secs"], pgid=a["pgid"]))
+        if name == "run_stage":
+            return {"state": self.runner(self.stages[a["stage_id"]])}
+        if name == "save_programme":
+            base = Path(a["base"])
+            base.mkdir(parents=True, exist_ok=True)
+            # oracle byte format: json.dumps(..., indent=1) — reordered to
+            # the oracle's key order first (serde maps are alphabetical)
+            (base / f"{a['doc']['programme_id']}.json").write_text(
+                json.dumps(_normalize_programme_doc(a["doc"]), indent=1))
+            return None
+        raise RuntimeError(f"pc kernel issued unknown op: {name}")
+
+
+def _normalize_programme_doc(doc: dict) -> dict:
+    """Reorder a kernel-built doc into the oracle's key order so the
+    written bytes match json.dumps of the oracle's dict (indent=1)."""
+    out = {
+        "programme_id": doc.get("programme_id", ""),
+        "objective": doc.get("objective", ""),
+        "stages": [
+            {"stage_id": s.get("stage_id", ""),
+             "objective": s.get("objective", ""),
+             "state": s.get("state", "PENDING"),
+             "depends_on": list(s.get("depends_on") or []),
+             "priority": s.get("priority", 0)}
+            for s in doc.get("stages", [])
+        ],
+        "fallbacks": list(doc.get("fallbacks") or []),
+        "saved_at": doc.get("saved_at", ""),
+    }
+    return out
+
+
+def _normalize_contract_doc(doc: dict) -> dict:
+    """Checkpoint dicts the kernel appends are alphabetical; reorder to
+    the oracle's {at, phase, evidence, note} before a real store save."""
+    for e in (doc or {}).get("entities", []):
+        cps = e.get("checkpoints")
+        if isinstance(cps, list):
+            e["checkpoints"] = [
+                {"at": cp.get("at", ""), "phase": cp.get("phase", ""),
+                 "evidence": cp.get("evidence", ""), "note": cp.get("note", "")}
+                if isinstance(cp, dict) else cp
+                for cp in cps
+            ]
+    return doc
+
+
+# ── payload builders ────────────────────────────────────────────────────
+
+def prog_payload(prog, *, work_marker: bool = False) -> dict:
+    return {
+        "programme_id": prog.programme_id,
+        "objective": prog.objective,
+        "stages": [stage_payload(s, work_marker=work_marker) for s in prog.stages],
+        "fallbacks": list(prog.fallbacks),
+    }
+
+
+def stage_payload(s, *, work_marker: bool = False) -> dict:
+    d = {
+        "stage_id": s.stage_id,
+        "objective": s.objective,
+        "state": s.state,
+        "depends_on": list(s.depends_on),
+        "priority": s.priority,
+        "result": {k: v for k, v in s.result.items() if k != "work"},
+    }
+    if work_marker and "work" in s.result:
+        w = s.result["work"]
+        # the kernel never reads work contents — presence is the signal;
+        # the host resolves callables vs literal specs behind the work op
+        d["result"]["work"] = ({"__callable__": True} if callable(w)
+                               else _jsonable(w))
+    return d
+
+
+# ── pure exports ────────────────────────────────────────────────────────
+
+def canary(prog) -> dict:
+    return _call_simple("pc_canary", {"prog": prog_payload(prog)})
+
+
+def eligible(prog) -> list:
+    return _call_simple("pc_eligible", {"prog": prog_payload(prog)})
+
+
+def stage_to_dict(s) -> dict:
+    return _call_simple("pc_stage_to_dict", {"stage": stage_payload(s)})
+
+
+def save_doc(prog) -> dict:
+    from rune import programme_continuation as pc_mod
+    saved_at = pc_mod.time.strftime("%Y-%m-%dT%H:%M:%SZ", pc_mod.time.gmtime())
+    doc = _call_simple("pc_save_doc",
+                       {"prog": prog_payload(prog), "saved_at": saved_at})
+    return _normalize_programme_doc(doc)
+
+
+def load_doc(doc: dict | None):
+    out = _call_simple("pc_load", {"doc": doc})
+    return None if out is None else out
+
+
+# ── host-driven exports ─────────────────────────────────────────────────
+
+def run_programme(prog, stage_runner, *, max_transitions: int = 100) -> dict:
+    proxy = OpsProxy(stages={s.stage_id: s for s in prog.stages},
+                     runner=stage_runner)
+    out = _call_host("pc_run_programme",
+                     {"prog": prog_payload(prog),
+                      "max_transitions": max_transitions}, proxy)
+    out["log"] = [_normalize_log_entry(e) for e in out["log"]]
+    return out
+
+
+def stage_run(stage, cfg: dict, iface) -> dict:
+    proxy = OpsProxy(iface=iface, stages={stage.stage_id: stage})
+    out = _call_host("pc_stage_run",
+                     {"cfg": cfg, "stage": stage_payload(stage,
+                                                         work_marker=True)},
+                     proxy)
+    return out
+
+
+def run_construction(prog, cfg: dict, iface, *,
+                     max_transitions: int = 100) -> dict:
+    proxy = OpsProxy(iface=iface,
+                     stages={s.stage_id: s for s in prog.stages})
+    out = _call_host("pc_run_construction",
+                     {"prog": prog_payload(prog, work_marker=True), "cfg": cfg,
+                      "max_transitions": max_transitions}, proxy)
+    out["log"] = [_normalize_log_entry(e) for e in out["log"]]
+    return out
+
+
+def _normalize_log_entry(e: dict) -> dict:
+    """The oracle's next_action is a tuple; the kernel's is an array."""
+    if "canary" in e:
+        na = e["canary"].get("next_action")
+        if isinstance(na, list):
+            e["canary"]["next_action"] = tuple(na)
+    return e

src/rune/programme_continuation.py

@@ -19,6 +19,7 @@ list.
 from __future__ import annotations
 
 import json
+import os
 import time
 from dataclasses import dataclass, field
 from pathlib import Path
@@ -47,6 +48,13 @@ class ProgrammeStage:
     result: dict = field(default_factory=dict)
 
     def to_dict(self) -> dict:
+        if os.environ.get("RUNE_PC_CORE", "auto") != "python":
+            try:
+                from rune import pc_loader
+                if pc_loader.available():
+                    return pc_loader.stage_to_dict(self)
+            except Exception:
+                pass  # Python oracle below; never crash the caller
         return {"stage_id": self.stage_id, "objective": self.objective,
                 "state": self.state, "depends_on": list(self.depends_on),
                 "priority": self.priority}
@@ -66,6 +74,17 @@ class Programme:
     def continuation_canary(self) -> dict:
         """'task done' ≠ 'programme done'. Returns the verdict AND the
         next action — the runner never has to guess whether to stop."""
+        if os.environ.get("RUNE_PC_CORE", "auto") != "python":
+            try:
+                from rune import pc_loader
+                if pc_loader.available():
+                    out = pc_loader.canary(self)
+                    na = out.get("next_action")
+                    return {"verdict": out["verdict"],
+                            "next_action": tuple(na) if na is not None else None,
+                            "reason": out["reason"]}
+            except Exception:
+                pass  # Python oracle below; never crash the caller
         complete = [s for s in self.stages if s.state == COMPLETE]
         if len(complete) == len(self.stages):
             return {"verdict": "PROGRAMME_COMPLETE",
@@ -118,6 +137,15 @@ class Programme:
         """PENDING stages whose dependencies are COMPLETE, priority-first.
         ELIGIBLE counts as runnable too — an eligible stage nobody started
         is exactly the case the continuation canary exists to catch."""
+        if os.environ.get("RUNE_PC_CORE", "auto") != "python":
+            try:
+                from rune import pc_loader
+                if pc_loader.available():
+                    ids = pc_loader.eligible(self)
+                    by_id = {s.stage_id: s for s in self.stages}
+                    return [by_id[i] for i in ids if i in by_id]
+            except Exception:
+                pass  # Python oracle below; never crash the caller
         done = {s.stage_id for s in self.stages if s.state == COMPLETE}
         out = [s for s in self.stages
                if s.state in (PENDING, ELIGIBLE)
@@ -127,6 +155,18 @@ class Programme:
 
     # ── durable persistence (canonical store, not transcript) ────────
     def save(self, base: Path) -> Path:
+        if os.environ.get("RUNE_PC_CORE", "auto") != "python":
+            try:
+                from rune import pc_loader
+                if pc_loader.available():
+                    doc = pc_loader.save_doc(self)  # clock read via pc.time
+                    base = Path(base)
+                    base.mkdir(parents=True, exist_ok=True)
+                    p = base / f"{self.programme_id}.json"
+                    p.write_text(json.dumps(doc, indent=1))
+                    return p
+            except Exception:
+                pass  # Python oracle below; never crash the caller
         base.mkdir(parents=True, exist_ok=True)
         p = base / f"{self.programme_id}.json"
         p.write_text(json.dumps({
@@ -145,6 +185,21 @@ class Programme:
         if not p.exists():
             return None
         d = json.loads(p.read_text())
+        if os.environ.get("RUNE_PC_CORE", "auto") != "python":
+            try:
+                from rune import pc_loader
+                if pc_loader.available():
+                    out = pc_loader.load_doc(d)
+                    if out is None:
+                        return None
+                    prog = cls(programme_id=out["programme_id"],
+                               objective=out["objective"],
+                               fallbacks=out.get("fallbacks", []))
+                    for s in out["stages"]:
+                        prog.stages.append(ProgrammeStage(**s))
+                    return prog
+            except Exception:
+                pass  # Python oracle below; never crash the caller
         prog = cls(programme_id=d["programme_id"],
                    objective=d["objective"], fallbacks=d.get("fallbacks", []))
         for s in d["stages"]:
@@ -158,6 +213,19 @@ def run_programme(prog: Programme, stage_runner, *,
     provided by the caller — must return a state for the stage. Between
     EVERY stage: persist → continuation canary → next. Returns the
     transition log (the 15M proof artifact)."""
+    if os.environ.get("RUNE_PC_CORE", "auto") != "python":
+        try:
+            from rune import pc_loader
+            if pc_loader.available():
+                out = pc_loader.run_programme(
+                    prog, stage_runner, max_transitions=max_transitions)
+                by_id = {s.stage_id: s for s in prog.stages}
+                for sid, state in out["states"].items():
+                    if sid in by_id:
+                        by_id[sid].state = state
+                return out["log"]
+        except Exception:
+            pass  # Python oracle below; never crash the caller
     log: list[dict] = []
     for _ in range(max_transitions):
         prog_canary = prog.continuation_canary()
@@ -202,6 +270,26 @@ def scaffolding_stage_runner(interface, *, scaffold_store: Path,
     runs under the event bridge and the driver resumes on the EVENT;
     canonical verification still decides completion truth. Stages without
     work complete as governance-only transitions."""
+    if os.environ.get("RUNE_PC_CORE", "auto") != "python":
+        try:
+            from rune import pc_loader
+            if pc_loader.available():
+                from rune import event_wait as ew  # optional dependency
+
+                cfg = {"scaffold_store": str(scaffold_store),
+                       "daemon_url": daemon_url or "",
+                       "watchdog_secs": watchdog_secs,
+                       "project_id": getattr(interface.scaffold,
+                                             "project_id", "")}
+
+                def run_kernel(stage: ProgrammeStage) -> str:
+                    out = pc_loader.stage_run(stage, cfg, interface)
+                    stage.result.update(out["result"])
+                    return out["state"]
+
+                return run_kernel
+        except Exception:
+            pass  # Python oracle below; never crash the caller
     from rune import event_wait as ew  # local import: optional dependency
 
     def _phase(contract_id: str) -> str:
@@ -321,6 +409,27 @@ def run_construction_programme(prog: Programme, interface, *,
     run_programme with the Scaffolding-bound runner; persists the
     programme around every transition so a restart resumes from the
     durable record, never a transcript. Returns the transition log."""
+    if os.environ.get("RUNE_PC_CORE", "auto") != "python":
+        try:
+            from rune import pc_loader
+            if pc_loader.available():
+                cfg = {"scaffold_store": str(scaffold_store),
+                       "daemon_url": daemon_url or "",
+                       "watchdog_secs": watchdog_secs,
+                       "project_id": getattr(interface.scaffold,
+                                             "project_id", "")}
+                out = pc_loader.run_construction(
+                    prog, cfg, interface, max_transitions=max_transitions)
+                by_id = {s.stage_id: s for s in prog.stages}
+                for sid, state in out["states"].items():
+                    if sid in by_id:
+                        by_id[sid].state = state
+                for sid, result in out["results"].items():
+                    if sid in by_id:
+                        by_id[sid].result.update(result)
+                return out["log"]
+        except Exception:
+            pass  # Python oracle below; never crash the caller
     runner = scaffolding_stage_runner(
         interface, scaffold_store=scaffold_store,
         daemon_url=daemon_url, watchdog_secs=watchdog_secs)

src/rust/programme_continuation/src/lib.rs

@@ -1,28 +1,805 @@
-//! programme_continuation — stage-to-stage continuation (W6).
-use serde::{Deserialize, Serialize};
+//! programme_continuation — the PROGRAMME CONTINUATION CANARY (T15 §15M,
+//! T17 §17D/§17E) — FULL parity port of src/rune/programme_continuation.py
+//! (dichotomy wave, #2479 tranche).
+//!
+//! "task done" is not "programme done". After ANY stage completes the
+//! continuation canary decides: the next eligible stage starts
+//! immediately (no user re-prompt, no voluntary stop), blocked branches
+//! fall back to pre-authorized work, and every stop carries an honest
+//! verdict.
+//!
+//! Port scope = ALL 11 oracle functions: ProgrammeStage.to_dict,
+//! Programme.continuation_canary / eligible_stages / save / load,
+//! run_programme, _programme_base, scaffolding_stage_runner (the `run`
+//! body), its `_phase` helper, and run_construction_programme. The
+//! seams the oracle leaves to the caller — the Scaffolding interface
+//! (load/save/contract/transition/close), the event bridge, the work
+//! callable, the stage runner, and the clock — cross the FFI as JSON
+//! ops served by the Python host (pc_loader.OpsProxy). The kernel is
+//! otherwise pure: drivers, canary verdicts, eligibility, persistence
+//! doc build, and every transition decision are FULL Rust.
+//!
+//! FFI (doc-envelope law): every export returns
+//! {"ok": true, "value": ...} / {"ok": false, "err": ...}. Host-driven
+//! exports take (arg, op_fn, user_data); op_fn is invoked with a JSON
+//! op {"op": name, "args": {...}} and returns the JSON result. No
+//! free-string export — callers free through capability_scoring's
+//! cs_free_string (sibling-crate symbol, already in-process).
 
-#[derive(Debug, Clone, Serialize, Deserialize, Default)]
+use serde::{Deserialize, Serialize};
+use serde_json::{json, Map, Value};
+use std::collections::HashSet;
+
+// ── stage states (name-preserving with the oracle) ──────────────────────
+pub const PENDING: &str = "PENDING";
+pub const ELIGIBLE: &str = "ELIGIBLE";
+pub const WAITING_EVENT: &str = "WAITING_EVENT";
+pub const COMPLETE: &str = "COMPLETE";
+pub const BLOCKED: &str = "BLOCKED";
+pub const APPROVAL_REQUIRED: &str = "APPROVAL_REQUIRED";
+pub const FAILED: &str = "FAILED";
+
+pub const TERMINAL: &[&str] = &[COMPLETE, BLOCKED, APPROVAL_REQUIRED, FAILED];
+
+const DEFAULT_DAEMON_URL: &str = "http://127.0.0.1:8791";
+pub const DEFAULT_PROGRAMME_AREA: &str = "programmes";
+pub const DEFAULT_WATCHDOG_SECS: i64 = 900;
+
+// ── ops host seam ───────────────────────────────────────────────────────
+
+/// The seam the oracle leaves to the caller. Each call is one JSON op;
+/// the host serves it against the live objects and returns the result.
+pub trait OpsHost {
+    fn call(&mut self, op: &Value) -> Result<Value, String>;
+}
+
+// ── data model (name-preserving with the dataclasses) ───────────────────
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(default)]
+pub struct ProgrammeStage {
+    pub stage_id: String,
+    pub objective: String,
+    pub state: String,
+    pub depends_on: Vec<String>,
+    pub priority: i64,
+    pub result: Map<String, Value>,
+}
+
+impl Default for ProgrammeStage {
+    fn default() -> Self {
+        ProgrammeStage {
+            stage_id: String::new(),
+            objective: String::new(),
+            state: PENDING.to_string(),
+            depends_on: Vec::new(),
+            priority: 0,
+            result: Map::new(),
+        }
+    }
+}
+
+impl ProgrammeStage {
+    /// `to_dict` — the oracle omits `result` (durable record shape).
+    pub fn to_dict(&self) -> Value {
+        json!({
+            "stage_id": self.stage_id,
+            "objective": self.objective,
+            "state": self.state,
+            "depends_on": self.depends_on,
+            "priority": self.priority,
+        })
+    }
+
+    pub fn result_without_work(&self) -> Value {
+        let mut out = self.result.clone();
+        out.remove("work");
+        Value::Object(out)
+    }
+}
+
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+#[serde(default)]
 pub struct Programme {
     pub programme_id: String,
     pub objective: String,
-    pub stages: Vec<serde_json::Value>,
+    pub stages: Vec<ProgrammeStage>,
+    pub fallbacks: Vec<String>,
 }
 
-pub fn run_construction_programme(
-    programme: &Programme,
-    closed_contract_id: &str,
-) -> serde_json::Value {
-    // Find the closed stage and determine the next eligible stage
-    let stage_ids: Vec<&str> = programme
-        .stages
-        .iter()
-        .filter_map(|s| s.get("stage_id").and_then(|id| id.as_str()))
-        .collect();
-    let idx = stage_ids.iter().position(|s| *s == closed_contract_id);
-    match idx {
-        Some(i) if i + 1 < stage_ids.len() => {
-            serde_json::json!({"verdict": "PROGRAMME_OPEN", "next_action": ["start", stage_ids[i+1]]})
+impl Programme {
+    /// `continuation_canary` — 'task done' ≠ 'programme done'. Returns
+    /// the verdict AND the next action; the runner never has to guess
+    /// whether to stop. next_action mirrors the oracle's tuple as a
+    /// 2-element array (null when stopping).
+    pub fn continuation_canary(&self) -> Value {
+        let complete: Vec<&ProgrammeStage> =
+            self.stages.iter().filter(|s| s.state == COMPLETE).collect();
+        if complete.len() == self.stages.len() {
+            return json!({
+                "verdict": "PROGRAMME_COMPLETE",
+                "next_action": Value::Null,
+                "reason": format!("all {} stages complete", self.stages.len()),
+            });
         }
-        _ => serde_json::json!({"verdict": "PROGRAMME_COMPLETE", "next_action": null}),
+        let eligible = self.eligible_stages();
+        if !eligible.is_empty() {
+            let nxt = eligible[0];
+            return json!({
+                "verdict": "PROGRAMME_OPEN",
+                "next_action": ["start", nxt.stage_id],
+                "reason": format!(
+                    "{}/{} complete; {} eligible — continue immediately",
+                    complete.len(), self.stages.len(), eligible.len()),
+            });
+        }
+        let waiting: Vec<&ProgrammeStage> =
+            self.stages.iter().filter(|s| s.state == WAITING_EVENT).collect();
+        if !waiting.is_empty() {
+            return json!({
+                "verdict": "PROGRAMME_OPEN_WAITING",
+                "next_action": ["await_event", waiting[0].stage_id],
+                "reason": "event-driven wait active — no polling",
+            });
+        }
+        let blocked: Vec<&ProgrammeStage> =
+            self.stages.iter().filter(|s| s.state == BLOCKED).collect();
+        if !blocked.is_empty() && !self.fallbacks.is_empty() {
+            return json!({
+                "verdict": "PROGRAMME_OPEN",
+                "next_action": ["fallback", self.fallbacks[0]],
+                "reason": format!(
+                    "{} blocked; pre-authorized fallback available", blocked.len()),
+            });
+        }
+        if !blocked.is_empty() {
+            return json!({
+                "verdict": "ALL_REMAINING_BLOCKED",
+                "next_action": Value::Null,
+                "reason": format!("{} stage(s) blocked, no fallback", blocked.len()),
+            });
+        }
+        let approval: Vec<&ProgrammeStage> =
+            self.stages.iter().filter(|s| s.state == APPROVAL_REQUIRED).collect();
+        if !approval.is_empty() {
+            return json!({
+                "verdict": "APPROVAL_REQUIRED",
+                "next_action": Value::Null,
+                "reason": format!(
+                    "{} stage(s) need user authorization", approval.len()),
+            });
+        }
+        // T17 §17E: all stages terminal but some FAILED — the programme
+        // may still close (programme-level acceptance decides), never
+        // hang in UNKNOWN_INTERRUPTION.
+        if self.stages.iter().all(|s| TERMINAL.contains(&s.state.as_str())) {
+            let failed: Vec<String> = self.stages.iter()
+                .filter(|s| s.state == FAILED)
+                .map(|s| s.stage_id.clone())
+                .collect();
+            // Python list-repr fidelity: ['a', 'b'] — single quotes
+            let failed_repr = format!(
+                "[{}]",
+                failed.iter().map(|f| format!("'{f}'")).collect::<Vec<_>>().join(", "));
+            return json!({
+                "verdict": "PROGRAMME_COMPLETE_WITH_FAILURES",
+                "next_action": Value::Null,
+                "reason": format!(
+                    "all stages terminal; failed: {failed_repr} — \
+                     programme-level acceptance decides"),
+            });
+        }
+        json!({
+            "verdict": "UNKNOWN_INTERRUPTION",
+            "next_action": Value::Null,
+            "reason": "no stage runnable and none terminal",
+        })
+    }
+
+    /// `eligible_stages` — PENDING stages whose dependencies are
+    /// COMPLETE, priority-first (ties by stage_id). ELIGIBLE counts as
+    /// runnable too — an eligible stage nobody started is exactly the
+    /// case the continuation canary exists to catch.
+    pub fn eligible_stages(&self) -> Vec<&ProgrammeStage> {
+        let done: HashSet<&str> = self.stages.iter()
+            .filter(|s| s.state == COMPLETE)
+            .map(|s| s.stage_id.as_str())
+            .collect();
+        let mut out: Vec<&ProgrammeStage> = self.stages.iter()
+            .filter(|s| (s.state == PENDING || s.state == ELIGIBLE)
+                && s.depends_on.iter().all(|d| done.contains(d.as_str())))
+            .collect();
+        out.sort_by(|a, b| (a.priority, a.stage_id.as_str())
+            .cmp(&(b.priority, b.stage_id.as_str())));
+        out
+    }
+
+    /// `save` — the durable record doc (kernel builds it; the Python
+    /// side writes the file with json.dumps(..., indent=1)). saved_at
+    /// comes from the host clock in oracle format %Y-%m-%dT%H:%M:%SZ.
+    pub fn save_doc(&self, saved_at: &str) -> Value {
+        json!({
+            "programme_id": self.programme_id,
+            "objective": self.objective,
+            "stages": self.stages.iter().map(|s| s.to_dict()).collect::<Vec<_>>(),
+            "fallbacks": self.fallbacks,
+            "saved_at": saved_at,
+        })
+    }
+
+    /// `load` — rebuild a Programme from the parsed durable record
+    /// (null doc = file missing → None in the oracle).
+    pub fn load(doc: Option<&Value>) -> Option<Programme> {
+        let doc = doc?;
+        Some(Programme {
+            programme_id: str_of(doc, "programme_id"),
+            objective: str_of(doc, "objective"),
+            stages: doc.get("stages").and_then(|v| v.as_array())
+                .map(|arr| arr.iter().map(ProgrammeStage::from_doc_stage).collect())
+                .unwrap_or_default(),
+            fallbacks: doc.get("fallbacks").and_then(|v| v.as_array())
+                .map(|arr| arr.iter().filter_map(|v| v.as_str().map(String::from)).collect())
+                .unwrap_or_default(),
+        })
     }
 }
+
+impl ProgrammeStage {
+    /// Stage dict from a durable record — ProgrammeStage(**s): the
+    /// record carries no `result` (to_dict omits it); the kernel's
+    /// from-payload parse accepts an optional result map.
+    fn from_doc_stage(v: &Value) -> ProgrammeStage {
+        ProgrammeStage {
+            stage_id: str_of(v, "stage_id"),
+            objective: str_of(v, "objective"),
+            state: str_of(v, "state"),
+            depends_on: v.get("depends_on").and_then(|v| v.as_array())
+                .map(|a| a.iter().filter_map(|x| x.as_str().map(String::from)).collect())
+                .unwrap_or_default(),
+            priority: v.get("priority").and_then(|v| v.as_i64()).unwrap_or(0),
+            result: v.get("result").and_then(|v| v.as_object()).cloned()
+                .unwrap_or_default(),
+        }
+    }
+}
+
+fn str_of(v: &Value, key: &str) -> String {
+    v.get(key).and_then(|x| x.as_str()).unwrap_or("").to_string()
+}
+
+// ── Python value semantics the oracle leans on ──────────────────────────
+
+fn truthy(v: Option<&Value>) -> bool {
+    match v {
+        None | Some(Value::Null) => false,
+        Some(Value::Bool(b)) => *b,
+        Some(Value::Number(n)) => n.as_f64().map(|f| f != 0.0).unwrap_or(false),
+        Some(Value::String(s)) => !s.is_empty(),
+        Some(Value::Array(a)) => !a.is_empty(),
+        Some(Value::Object(o)) => !o.is_empty(),
+    }
+}
+
+/// Python str() of a value inside f-strings (None → "None").
+fn py_str(v: Option<&Value>) -> String {
+    match v {
+        None | Some(Value::Null) => "None".to_string(),
+        Some(Value::Bool(b)) => if *b { "True".into() } else { "False".into() },
+        Some(Value::String(s)) => s.clone(),
+        Some(Value::Number(n)) => n.to_string(),
+        Some(other) => other.to_string(),
+    }
+}
+
+/// `x.get(k, default)` — key-present (even null) wins over the default.
+fn get_or_default(v: &Value, key: &str, default: &str) -> Value {
+    match v.get(key) {
+        Some(found) => found.clone(),
+        None => json!(default),
+    }
+}
+
+/// `x.get(k) or default` — Python falsiness wins over the default.
+fn or_str(v: Option<&Value>, default: &str) -> String {
+    match v {
+        Some(s) if truthy(Some(s)) => match s {
+            Value::String(t) => t.clone(),
+            other => other.to_string(),
+        },
+        _ => default.to_string(),
+    }
+}
+
+/// Python [:200] — first 200 characters, not bytes.
+fn trunc_chars(s: &str, n: usize) -> String {
+    s.chars().take(n).collect()
+}
+
+// ── run_programme (15M) ─────────────────────────────────────────────────
+
+/// `run_programme` — drive one programme to a legitimate stop. The
+/// stage_runner crosses the seam as a `run_stage` op (the host calls the
+/// caller-provided runner); between EVERY stage: canary → next. The
+/// returned transition log is the 15M proof artifact.
+pub fn run_programme(
+    prog: &mut Programme,
+    max_transitions: i64,
+    host: &mut dyn OpsHost,
+) -> Result<Vec<Value>, String> {
+    let mut log: Vec<Value> = Vec::new();
+    for _ in 0..max_transitions {
+        let canary = prog.continuation_canary();
+        log.push(json!({"canary": canary}));
+        let Some(action) = canary.get("next_action").filter(|v| !v.is_null()) else {
+            break; // legitimate stop — reason recorded in the log
+        };
+        let kind = action.get(0).and_then(|v| v.as_str()).unwrap_or("").to_string();
+        let target = action.get(1).and_then(|v| v.as_str()).unwrap_or("").to_string();
+        if kind == "fallback" {
+            log.push(json!({"fallback_started": target}));
+            break;
+        }
+        let idx = prog.stages.iter().position(|s| s.stage_id == target)
+            .ok_or_else(|| format!("StopIteration: no stage {target}"))?;
+        let res = host.call(&json!({"op": "run_stage", "args": {
+            "stage_id": prog.stages[idx].stage_id,
+            "stage_state": prog.stages[idx].state,
+        }}))?;
+        let new_state = res.get("state").and_then(|v| v.as_str())
+            .unwrap_or("").to_string();
+        prog.stages[idx].state = new_state;
+        log.push(json!({"stage": target, "new_state": prog.stages[idx].state}));
+    }
+    Ok(log)
+}
+
+// ── T17 §17D: production wiring — Scaffolding-integrated driver ─────────
+
+/// `_programme_base` — the canonical store's programme area.
+pub fn programme_base(scaffold_store: &str) -> String {
+    format!("{scaffold_store}/{DEFAULT_PROGRAMME_AREA}")
+}
+
+fn cfg_str<'a>(cfg: &'a Value, key: &str, default: &'a str) -> &'a str {
+    cfg.get(key).and_then(|v| v.as_str()).unwrap_or(default)
+}
+
+fn cfg_i64(cfg: &Value, key: &str, default: i64) -> i64 {
+    cfg.get(key).and_then(|v| v.as_i64()).unwrap_or(default)
+}
+
+/// `_phase` — ALWAYS re-read: transition() saves through its own doc
+/// copy, so a locally-held contract dict goes stale after any
+/// transition (T17 17E live finding — the draft→approved→building
+/// chain silently stopped after the first step on a stale reference).
+fn _phase(host: &mut dyn OpsHost, contract_id: &str) -> Result<String, String> {
+    let _doc = host.call(&json!({"op": "scaffold_load", "args": {}}))?;
+    let c = host.call(&json!({"op": "contract", "args": {"contract_id": contract_id}}))?;
+    Ok(c.get("contract")
+        .and_then(|v| v.get("phase"))
+        .and_then(|v| v.as_str())
+        .unwrap_or("")
+        .to_string())
+}
+
+fn contract_lookup(host: &mut dyn OpsHost, contract_id: &str) -> Result<Value, String> {
+    host.call(&json!({"op": "contract", "args": {"contract_id": contract_id}}))
+}
+
+fn transition_op(
+    host: &mut dyn OpsHost,
+    contract_id: &str,
+    phase: &str,
+    evidence_ref: &str,
+) -> Result<Value, String> {
+    host.call(&json!({"op": "transition", "args": {
+        "contract_id": contract_id, "phase": phase, "evidence_ref": evidence_ref,
+    }}))
+}
+
+fn now_op(host: &mut dyn OpsHost) -> Result<Value, String> {
+    host.call(&json!({"op": "now", "args": {}}))
+}
+
+/// Append a checkpoint to the contract entity the kernel located via
+/// entity_id, inside the doc the kernel holds from scaffold_load (the
+/// oracle mutates the LIVE contract dict through the same reference).
+fn append_checkpoint(doc: &mut Value, entity_id: &str, checkpoint: Value) -> bool {
+    if let Some(entities) = doc.get_mut("entities").and_then(|v| v.as_array_mut()) {
+        for e in entities.iter_mut() {
+            if e.get("id").and_then(|v| v.as_str()) == Some(entity_id) {
+                let obj = e.as_object_mut().unwrap();
+                let cps = obj.entry("checkpoints")
+                    .or_insert_with(|| Value::Array(Vec::new()));
+                if let Some(arr) = cps.as_array_mut() {
+                    arr.push(checkpoint);
+                    return true;
+                }
+                return false;
+            }
+        }
+    }
+    false
+}
+
+/// The stage runner body (oracle's nested `run` inside
+/// scaffolding_stage_runner): drive ONE stage through the Construction
+/// Interface lifecycle to a legitimate terminal state.
+pub fn run(
+    stage: &mut ProgrammeStage,
+    cfg: &Value,
+    host: &mut dyn OpsHost,
+) -> Result<Value, String> {
+    let contract_id = stage.stage_id.clone();
+    let mut phase = _phase(host, &contract_id)?;
+    if phase.is_empty() {
+        stage.result.insert("error".into(), json!(format!("no contract {contract_id}")));
+        return Ok(json!(FAILED));
+    }
+    if phase == "closed" {
+        return Ok(json!(COMPLETE)); // idempotent re-entry (duplicate signal case)
+    }
+    if phase == "draft" {
+        transition_op(host, &contract_id, "approved", "programme-stage-start")?;
+    }
+    phase = _phase(host, &contract_id)?;
+    if phase == "approved" {
+        transition_op(host, &contract_id, "building", "programme-stage-start")?;
+    }
+    if stage.result.get("work").is_none() {
+        transition_op(host, &contract_id, "verifying", "governance-only-stage")?;
+        host.call(&json!({"op": "close", "args": {
+            "contract_id": contract_id,
+            "observed_evidence": [{"source": "programme-driver",
+                "detail": "governance-only stage: no work, closure by \
+                           programme acceptance"}],
+            "scope_files": [],
+        }}))?;
+        return Ok(json!(COMPLETE));
+    }
+    // work is either a callable (interface, stage) → outcome dict, or a
+    // literal outcome spec dict ({"ok", "evidence", "async_cmd"?}) — the
+    // host resolves either behind the `work` op.
+    let work_res = host.call(&json!({"op": "work", "args": {
+        "stage_id": stage.stage_id, "stage_state": stage.state,
+    }}))?;
+    let mut outcome = work_res.get("outcome").cloned().unwrap_or_else(|| json!({}));
+    if truthy(outcome.get("async_cmd")) {
+        let watchdog_secs = cfg_i64(cfg, "watchdog_secs", DEFAULT_WATCHDOG_SECS);
+        let now = now_op(host)?;
+        let epoch = now.get("epoch").and_then(|v| v.as_f64()).unwrap_or(0.0) as i64;
+        let op = format!("stage-{contract_id}-{epoch}");
+        let project = cfg_str(cfg, "project_id", "");
+        host.call(&json!({"op": "register_process_wait", "args": {
+            "op": op, "contract_id": contract_id, "project": project,
+            "objective": stage.objective,
+            "watched_for": "stage work completion event",
+            "watchdog_secs": watchdog_secs,
+        }}))?;
+        let daemon_url = cfg_str(cfg, "daemon_url", "");
+        let daemon = if daemon_url.is_empty() { DEFAULT_DAEMON_URL } else { daemon_url };
+        let notifier = host.call(&json!({"op": "run_builder_under_events", "args": {
+            "async_cmd": outcome.get("async_cmd").cloned().unwrap_or(Value::Null),
+            "op": op, "daemon_url": daemon,
+        }}))?;
+        let pid = notifier.get("pid").and_then(|v| v.as_i64()).unwrap_or(0);
+        stage.result.insert("operation_id".into(), json!(op));
+        stage.result.insert("notifier_pgid".into(), json!(pid));
+        let result = host.call(&json!({"op": "await_resume_event", "args": {
+            "op": op, "watchdog_secs": watchdog_secs, "pgid": pid,
+        }}))?;
+        let kind = result.get("kind").and_then(|v| v.as_str()).unwrap_or("");
+        if kind != "event" {
+            stage.result.insert("error".into(),
+                json!(format!("stage event wait: {kind}")));
+            transition_op(host, &contract_id, "failed", "stage-event-lost")?;
+            return Ok(json!(FAILED));
+        }
+        let rec = result.get("wait").cloned().unwrap_or(Value::Null);
+        outcome["evidence"] = json!(format!(
+            "event resume; wait {} verdict={}",
+            py_str(rec.get("wait_id")),
+            py_str(rec.get("verdict")),
+        ));
+    }
+    if !truthy(outcome.get("ok")) {
+        stage.result.insert("error".into(),
+            get_or_default(&outcome, "evidence", "work failed").clone());
+        // truthful failure: record the evidence ON the contract and
+        // drive it to the failed phase via the legal transition map
+        // (building→failed is legal; no exception swallowing)
+        let mut doc = host.call(&json!({"op": "scaffold_load", "args": {}}))?;
+        let c = contract_lookup(host, &contract_id)?;
+        if truthy(Some(c.get("contract").unwrap_or(&Value::Null))) {
+            let now = now_op(host)?;
+            let saved_at = now.get("now").and_then(|v| v.as_str()).unwrap_or("");
+            let note = trunc_chars(
+                &or_str(outcome.get("evidence"), "work failed"), 200);
+            let checkpoint = json!({
+                "at": saved_at,
+                "phase": c["contract"].get("phase").and_then(|v| v.as_str()).unwrap_or(""),
+                "evidence": "programme-stage-failed",
+                "note": note,
+            });
+            let entity_id = c.get("entity_id").and_then(|v| v.as_str()).unwrap_or("");
+            append_checkpoint(&mut doc, entity_id, checkpoint);
+            host.call(&json!({"op": "scaffold_save", "args": {"doc": doc}}))?;
+            if c["contract"].get("phase").and_then(|v| v.as_str()).unwrap_or("") == "building" {
+                transition_op(host, &contract_id, "failed", "programme-stage-failed")?;
+            }
+        }
+        return Ok(json!(FAILED));
+    }
+    // completion truth: work evidence recorded ON the contract, then
+    // closed (evidence lives on the durable object, not the log)
+    let phase = _phase(host, &contract_id)?;
+    if phase == "building" {
+        transition_op(host, &contract_id, "verifying", "programme-stage-verify")?;
+    }
+    let mut doc = host.call(&json!({"op": "scaffold_load", "args": {}}))?;
+    let c = contract_lookup(host, &contract_id)?;
+    if truthy(Some(c.get("contract").unwrap_or(&Value::Null))) {
+        let now = now_op(host)?;
+        let saved_at = now.get("now").and_then(|v| v.as_str()).unwrap_or("");
+        let note = trunc_chars(
+            &or_str(outcome.get("evidence"), "work completed"), 200);
+        let checkpoint = json!({
+            "at": saved_at,
+            "phase": "verifying",
+            "evidence": "programme-stage-work",
+            "note": note,
+        });
+        let entity_id = c.get("entity_id").and_then(|v| v.as_str()).unwrap_or("");
+        append_checkpoint(&mut doc, entity_id, checkpoint);
+        host.call(&json!({"op": "scaffold_save", "args": {"doc": doc}}))?;
+    }
+    let detail = get_or_default(&outcome, "evidence", "ok").clone();
+    host.call(&json!({"op": "close", "args": {
+        "contract_id": contract_id,
+        "observed_evidence": [{"source": "programme-driver", "detail": detail}],
+        "scope_files": [],
+    }}))?;
+    Ok(json!(COMPLETE))
+}
+
+/// `scaffolding_stage_runner` — a stage_runner for run_programme bound
+/// to the Construction Interface: partially-applied [`run`] over the
+/// caller's cfg (the oracle returns the nested closure).
+pub fn scaffolding_stage_runner(
+    cfg: Value,
+) -> impl FnMut(&mut ProgrammeStage, &mut dyn OpsHost) -> Result<Value, String> {
+    move |stage, host| run(stage, &cfg, host)
+}
+
+/// `run_construction_programme` — canonical multi-stage construction
+/// continuation (T17 §17D): run_programme with the Scaffolding-bound
+/// runner; persists the programme around every transition so a restart
+/// resumes from the durable record, never a transcript. Returns the
+/// transition log.
+pub fn run_construction_programme(
+    prog: &mut Programme,
+    cfg: &Value,
+    max_transitions: i64,
+    host: &mut dyn OpsHost,
+) -> Result<Vec<Value>, String> {
+    let mut log: Vec<Value> = Vec::new();
+    for _ in 0..max_transitions {
+        save_programme_op(prog, cfg, host)?;
+        let canary = prog.continuation_canary();
+        log.push(json!({"canary": canary}));
+        let Some(action) = canary.get("next_action").filter(|v| !v.is_null()) else {
+            break;
+        };
+        let kind = action.get(0).and_then(|v| v.as_str()).unwrap_or("").to_string();
+        let target = action.get(1).and_then(|v| v.as_str()).unwrap_or("").to_string();
+        if kind == "fallback" {
+            log.push(json!({"fallback_started": target}));
+            break;
+        }
+        let idx = prog.stages.iter().position(|s| s.stage_id == target)
+            .ok_or_else(|| format!("StopIteration: no stage {target}"))?;
+        if prog.stages[idx].state == COMPLETE {
+            continue; // duplicate continuation signal: idempotent
+        }
+        prog.stages[idx].state = ELIGIBLE.to_string();
+        let new_state = run(&mut prog.stages[idx], cfg, host)?;
+        let new_state = new_state.as_str().unwrap_or("").to_string();
+        prog.stages[idx].state = new_state.clone();
+        log.push(json!({"stage": target, "new_state": new_state}));
+        save_programme_op(prog, cfg, host)?;
+    }
+    save_programme_op(prog, cfg, host)?;
+    Ok(log)
+}
+
+fn save_programme_op(
+    prog: &Programme,
+    cfg: &Value,
+    host: &mut dyn OpsHost,
+) -> Result<(), String> {
+    let now = now_op(host)?;
+    let saved_at = now.get("now").and_then(|v| v.as_str()).unwrap_or("");
+    let doc = prog.save_doc(saved_at);
+    host.call(&json!({"op": "save_programme", "args": {
+        "base": programme_base(cfg_str(cfg, "scaffold_store", "")),
+        "doc": doc,
+    }}))?;
+    Ok(())
+}
+
+// ── FFI (doc-envelope law) ──────────────────────────────────────────────
+
+use std::ffi::{CStr, CString};
+use std::os::raw::{c_char, c_void};
+
+/// Host callback: receives a JSON op, returns a JSON result.
+pub type OpFn = unsafe extern "C" fn(op: *const c_char, user_data: *mut c_void) -> *mut c_char;
+
+struct HostCb {
+    f: OpFn,
+    ud: *mut c_void,
+}
+
+impl OpsHost for HostCb {
+    fn call(&mut self, op: &Value) -> Result<Value, String> {
+        let s = CString::new(serde_json::to_string(op).map_err(|e| e.to_string())?)
+            .map_err(|e| e.to_string())?;
+        let p = unsafe { (self.f)(s.as_ptr(), self.ud) };
+        if p.is_null() {
+            return Err("op host returned null".into());
+        }
+        let out = unsafe { CStr::from_ptr(p) }.to_str().map_err(|e| e.to_string())?;
+        let v: Value = serde_json::from_str(out).map_err(|e| e.to_string())?;
+        if let Some(err) = v.get("__host_error__").and_then(|x| x.as_str()) {
+            return Err(err.to_string());
+        }
+        Ok(v)
+    }
+}
+
+fn read_arg(ptr: *const c_char) -> Result<Value, String> {
+    if ptr.is_null() {
+        return Err("null argument pointer".into());
+    }
+    let x = unsafe { CStr::from_ptr(ptr) }.to_str().map_err(|e| e.to_string())?;
+    serde_json::from_str(x).map_err(|e| e.to_string())
+}
+
+fn envelope(r: Result<Value, String>) -> *mut c_char {
+    let v = match r {
+        Ok(x) => json!({"ok": true, "value": x}),
+        Err(e) => json!({"ok": false, "err": e}),
+    };
+    CString::new(serde_json::to_string(&v).unwrap_or_default())
+        .map(|c| c.into_raw())
+        .unwrap_or(std::ptr::null_mut())
+}
+
+fn parse_prog(v: &Value) -> Result<Programme, String> {
+    serde_json::from_value(v.clone()).map_err(|e| format!("bad programme payload: {e}"))
+}
+
+/// pc_canary({"prog": {...}}) → {"verdict", "next_action", "reason"}.
+#[no_mangle]
+pub extern "C" fn pc_canary(arg: *const c_char) -> *mut c_char {
+    envelope(read_arg(arg).and_then(|a| {
+        let prog: Programme = parse_prog(a.get("prog").unwrap_or(&Value::Null))?;
+        Ok(prog.continuation_canary())
+    }))
+}
+
+/// pc_eligible({"prog": {...}}) → ["stage_id", ...] priority-first.
+#[no_mangle]
+pub extern "C" fn pc_eligible(arg: *const c_char) -> *mut c_char {
+    envelope(read_arg(arg).and_then(|a| {
+        let prog: Programme = parse_prog(a.get("prog").unwrap_or(&Value::Null))?;
+        Ok(Value::Array(
+            prog.eligible_stages().iter().map(|s| json!(s.stage_id)).collect(),
+        ))
+    }))
+}
+
+/// pc_stage_to_dict({"stage": {...}}) → the durable stage dict.
+#[no_mangle]
+pub extern "C" fn pc_stage_to_dict(arg: *const c_char) -> *mut c_char {
+    envelope(read_arg(arg).and_then(|a| {
+        let stage: ProgrammeStage =
+            serde_json::from_value(a.get("stage").cloned().unwrap_or(Value::Null))
+                .map_err(|e| format!("bad stage payload: {e}"))?;
+        Ok(stage.to_dict())
+    }))
+}
+
+/// pc_save_doc({"prog": {...}, "saved_at": "..."} → the durable
+/// programme record doc (the Python side writes the file).
+#[no_mangle]
+pub extern "C" fn pc_save_doc(arg: *const c_char) -> *mut c_char {
+    envelope(read_arg(arg).and_then(|a| {
+        let prog: Programme = parse_prog(a.get("prog").unwrap_or(&Value::Null))?;
+        let saved_at = a.get("saved_at").and_then(|v| v.as_str()).unwrap_or("");
+        Ok(prog.save_doc(saved_at))
+    }))
+}
+
+/// pc_load({"doc": {...} | null}) → programme dict or null.
+#[no_mangle]
+pub extern "C" fn pc_load(arg: *const c_char) -> *mut c_char {
+    envelope(read_arg(arg).map(|a| {
+        match Programme::load(a.get("doc")) {
+            Some(p) => serde_json::to_value(&p).unwrap_or(Value::Null),
+            None => Value::Null,
+        }
+    }))
+}
+
+/// pc_run_programme({"prog", "max_transitions"}, op_fn, user_data)
+/// → {"log": [...], "states": {...}}. The stage runner arrives via the
+/// `run_stage` op.
+#[no_mangle]
+pub extern "C" fn pc_run_programme(
+    arg: *const c_char,
+    op_fn: OpFn,
+    user_data: *mut c_void,
+) -> *mut c_char {
+    let r = read_arg(arg).and_then(|a| {
+        let mut prog: Programme = parse_prog(a.get("prog").unwrap_or(&Value::Null))?;
+        let max_transitions = a.get("max_transitions").and_then(|v| v.as_i64()).unwrap_or(100);
+        let mut host = HostCb { f: op_fn, ud: user_data };
+        let log = run_programme(&mut prog, max_transitions, &mut host)?;
+        let states: Map<String, Value> = prog.stages.iter()
+            .map(|s| (s.stage_id.clone(), json!(s.state))).collect();
+        Ok(json!({"log": log, "states": states}))
+    });
+    envelope(r)
+}
+
+/// pc_stage_run({"cfg", "stage"}, op_fn, user_data)
+/// → {"state": "...", "result": {...}} — one scaffolding stage run.
+#[no_mangle]
+pub extern "C" fn pc_stage_run(
+    arg: *const c_char,
+    op_fn: OpFn,
+    user_data: *mut c_void,
+) -> *mut c_char {
+    let r = read_arg(arg).and_then(|a| {
+        let mut stage: ProgrammeStage =
+            serde_json::from_value(a.get("stage").cloned().unwrap_or(Value::Null))
+                .map_err(|e| format!("bad stage payload: {e}"))?;
+        let cfg = a.get("cfg").cloned().unwrap_or_else(|| json!({}));
+        let mut host = HostCb { f: op_fn, ud: user_data };
+        let state = run(&mut stage, &cfg, &mut host)?;
+        Ok(json!({"state": state, "result": stage.result_without_work()}))
+    });
+    envelope(r)
+}
+
+/// pc_run_construction({"prog", "cfg", "max_transitions"}, op_fn, user_data)
+/// → {"log": [...], "states": {...}, "results": {...}}.
+#[no_mangle]
+pub extern "C" fn pc_run_construction(
+    arg: *const c_char,
+    op_fn: OpFn,
+    user_data: *mut c_void,
+) -> *mut c_char {
+    let r = read_arg(arg).and_then(|a| {
+        let mut prog: Programme = parse_prog(a.get("prog").unwrap_or(&Value::Null))?;
+        let cfg = a.get("cfg").cloned().unwrap_or_else(|| json!({}));
+        let max_transitions = a.get("max_transitions").and_then(|v| v.as_i64()).unwrap_or(100);
+        let mut host = HostCb { f: op_fn, ud: user_data };
+        let log = run_construction_programme(&mut prog, &cfg, max_transitions, &mut host)?;
+        let states: Map<String, Value> = prog.stages.iter()
+            .map(|s| (s.stage_id.clone(), json!(s.state))).collect();
+        let results: Map<String, Value> = prog.stages.iter()
+            .map(|s| (s.stage_id.clone(), s.result_without_work())).collect();
+        Ok(json!({"log": log, "states": states, "results": results}))
+    });
+    envelope(r)
+}
+
+// NOTE: no pc_free_string export — sibling crates sharing the Python
+// process already export identical free symbols; a duplicate breaks the
+// cdylib link. Callers free through capability_scoring's cs_free_string.
+
+#[path = "pc_battery.rs"]
+#[cfg(test)]
+mod pc_battery;

src/rust/programme_continuation/src/pc_battery.rs

@@ -0,0 +1,198 @@
+//! programme_continuation branch battery — parity vs the frozen
+//! Python-oracle truth (/tmp/pc_truth.json: 16 canary + 5 eligible +
+//! 2 to_dict + 2 save + 2 load + 5 run_programme + 8 stage_run +
+//! 6 run_construction across every function and branch, frozen
+//! 2026-10-11 with RUNE_PC_CORE=python pinned). Embedded VERBATIM —
+//! never transcribed.
+//!
+//! Interface-driven cases replay the oracle's RECORDED OPS: the kernel
+//! must issue each op (name + args) in exactly the order the oracle
+//! did, then the recorded result is fed back — a hermetic, byte-frozen
+//! replay of the whole driver contract.
+
+use crate::{run_construction_programme, run_programme, Programme, ProgrammeStage, OpsHost};
+use serde_json::{json, Value};
+use std::collections::VecDeque;
+
+const TRUTH: &str = include_str!("frozen_pc_truth.json");
+
+/// Replay host: the kernel's op requests must match the oracle's
+/// recorded ops EXACTLY (name + args); recorded results are fed back.
+struct Replay {
+    ops: VecDeque<Value>,
+    case: String,
+    n: usize,
+}
+
+impl Replay {
+    fn new(ops: &[Value], case: &str) -> Self {
+        Replay { ops: ops.iter().cloned().collect(), case: case.to_string(), n: 0 }
+    }
+}
+
+impl OpsHost for Replay {
+    fn call(&mut self, req: &Value) -> Result<Value, String> {
+        self.n += 1;
+        let rec = self.ops.pop_front().ok_or_else(|| {
+            format!("case {}: kernel issued op #{} the oracle never did: {}",
+                    self.case, self.n, req)
+        })?;
+        assert_eq!(
+            rec.get("op"), req.get("op"),
+            "case {}: op #{} name mismatch (kernel vs oracle)",
+            self.case, self.n
+        );
+        assert_eq!(
+            rec.get("args"), req.get("args"),
+            "case {}: op #{} ({:?}) args mismatch (kernel vs oracle)",
+            self.case, self.n, req.get("op")
+        );
+        // The real close() runs the T5.15 programme-continuation wire
+        // internally (mark stage COMPLETE + re-save the programme file).
+        // Those host-side effects are recorded tagged host_internal and
+        // ride after the close op; the host consumes them here — they
+        // are invisible to the kernel's op stream by construction.
+        if req.get("op").and_then(|v| v.as_str()) == Some("close") {
+            while self.ops.front().map(|o| o.get("host_internal") == Some(&json!(true)))
+                .unwrap_or(false)
+            {
+                self.ops.pop_front();
+            }
+        }
+        Ok(rec.get("result").cloned().unwrap_or(Value::Null))
+    }
+}
+
+fn truth() -> Value {
+    serde_json::from_str(TRUTH).unwrap()
+}
+
+fn prog_of(v: &Value) -> Programme {
+    serde_json::from_value(v.clone()).unwrap()
+}
+
+fn stage_of(v: &Value) -> ProgrammeStage {
+    serde_json::from_value(v.clone()).unwrap()
+}
+
+#[test]
+fn canary_parity_battery() {
+    let t = truth();
+    for case in t["canary"].as_array().unwrap() {
+        let prog = prog_of(&case["input"]);
+        let got = prog.continuation_canary();
+        assert_eq!(
+            &got, &case["expected"],
+            "canary case {:?} input states {:?}",
+            case["name"].as_str().unwrap(),
+            prog.stages.iter().map(|s| s.state.as_str()).collect::<Vec<_>>()
+        );
+    }
+}
+
+#[test]
+fn eligible_parity_battery() {
+    let t = truth();
+    for case in t["eligible"].as_array().unwrap() {
+        let prog = prog_of(&case["input"]);
+        let got: Value = Value::Array(
+            prog.eligible_stages().iter().map(|s| json!(s.stage_id)).collect(),
+        );
+        assert_eq!(
+            &got, &case["expected"],
+            "eligible case {:?}",
+            case["name"].as_str().unwrap()
+        );
+    }
+}
+
+#[test]
+fn to_dict_parity_battery() {
+    let t = truth();
+    for case in t["to_dict"].as_array().unwrap() {
+        let stage = stage_of(&case["input"]);
+        let got = stage.to_dict();
+        assert_eq!(&got, &case["expected"], "to_dict case {:?}", case["name"].as_str().unwrap());
+    }
+}
+
+#[test]
+fn save_parity_battery() {
+    let t = truth();
+    for case in t["save"].as_array().unwrap() {
+        let prog = prog_of(&case["input"]["prog"]);
+        let got = prog.save_doc(case["input"]["saved_at"].as_str().unwrap());
+        assert_eq!(
+            &got, &case["expected_doc"],
+            "save case {:?} (doc; byte format json.dumps(indent=1) on the Python side)",
+            case["name"].as_str().unwrap()
+        );
+    }
+}
+
+#[test]
+fn load_parity_battery() {
+    let t = truth();
+    for case in t["load"].as_array().unwrap() {
+        let doc = &case["input"]["doc"];
+        let got = match Programme::load(if doc.is_null() { None } else { Some(doc) }) {
+            Some(p) => serde_json::to_value(&p).unwrap(),
+            None => Value::Null,
+        };
+        assert_eq!(&got, &case["expected"], "load case {:?}", case["name"].as_str().unwrap());
+    }
+}
+
+#[test]
+fn run_programme_parity_battery() {
+    let t = truth();
+    for case in t["run_programme"].as_array().unwrap() {
+        let name = case["name"].as_str().unwrap().to_string();
+        let mut prog = prog_of(&case["input"]["prog"]);
+        let max = case["input"]["max_transitions"].as_i64().unwrap_or(100);
+        let mut host = Replay::new(case["expected"]["ops"].as_array().unwrap(), &name);
+        let log = run_programme(&mut prog, max, &mut host)
+            .unwrap_or_else(|e| panic!("case {name}: kernel error: {e}"));
+        let states: Value = prog.stages.iter()
+            .map(|s| (s.stage_id.clone(), json!(s.state))).collect();
+        assert_eq!(json!(log), case["expected"]["log"], "run_programme case {name} log");
+        assert_eq!(json!(states), case["expected"]["states"], "run_programme case {name} states");
+    }
+}
+
+#[test]
+fn stage_run_parity_battery() {
+    let t = truth();
+    for case in t["stage_run"].as_array().unwrap() {
+        let name = case["name"].as_str().unwrap().to_string();
+        let mut stage = stage_of(&case["input"]["stage"]);
+        let cfg = case["input"]["cfg"].clone();
+        let mut host = Replay::new(case["expected"]["ops"].as_array().unwrap(), &name);
+        let state = crate::run(&mut stage, &cfg, &mut host)
+            .unwrap_or_else(|e| panic!("case {name}: kernel error: {e}"));
+        let result = stage.result_without_work();
+        assert_eq!(json!(state), case["expected"]["state"], "stage_run case {name} state");
+        assert_eq!(result, case["expected"]["result"], "stage_run case {name} result");
+    }
+}
+
+#[test]
+fn run_construction_parity_battery() {
+    let t = truth();
+    for case in t["run_construction"].as_array().unwrap() {
+        let name = case["name"].as_str().unwrap().to_string();
+        let mut prog = prog_of(&case["input"]["prog"]);
+        let cfg = case["input"]["cfg"].clone();
+        let max = case["input"]["max_transitions"].as_i64().unwrap_or(100);
+        let mut host = Replay::new(case["expected"]["ops"].as_array().unwrap(), &name);
+        let log = run_construction_programme(&mut prog, &cfg, max, &mut host)
+            .unwrap_or_else(|e| panic!("case {name}: kernel error: {e}"));
+        let states: Value = prog.stages.iter()
+            .map(|s| (s.stage_id.clone(), json!(s.state))).collect();
+        let results: Value = prog.stages.iter()
+            .map(|s| (s.stage_id.clone(), s.result_without_work())).collect();
+        assert_eq!(json!(log), case["expected"]["log"], "run_construction case {name} log");
+        assert_eq!(json!(states), case["expected"]["states"], "run_construction case {name} states");
+        assert_eq!(json!(results), case["expected"]["results"], "run_construction case {name} results");
+    }
+}