Adopt spec v0.2: schema, semantic validator, operating modes; workplans updated

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: sonnet
Assistant-Process: 295952@bnt-lap001
Assistant-Session: e93f64ad-516c-46eb-9666-aad8d300c477

rtut/__main__.py

@@ -9,6 +9,8 @@ from . import pack as P
 
 
 def _packs(root: str) -> list[P.Pack]:
+    if Path(root).is_file():
+        return [P.load(Path(root))]
     return [P.load(p) for p in P.discover(Path(root))]
 
 

rtut/pack.py

@@ -26,6 +26,10 @@ class Pack:
     def id(self) -> str:
         return str(self.data.get("id", self.path.parent.name))
 
+    @property
+    def spec(self) -> str:
+        return str(self.data.get("spec", ""))
+
     @property
     def ok(self) -> bool:
         return not self.errors
@@ -97,6 +101,9 @@ def load(path: Path) -> Pack:
         return Pack(path, {}, [f"invalid YAML: {e}"])
     if not isinstance(data, dict):
         return Pack(path, {}, ["manifest must be a mapping"])
+    if data.get("spec") == "runbook-pack/v0.2":
+        from . import v2
+        return Pack(path, data, v2.validate(data))
     errs = _schema_errors(data)
     if not errs:
         errs = _semantic_errors(data)

rtut/run.py

@@ -66,6 +66,8 @@ class Run:
 def create(pack: Pack, operator: str, params: dict[str, str] | None = None) -> Run:
     if not pack.ok or pack.data.get("engine") != "native":
         raise RunError("only valid native packs can be run")
+    if pack.spec != "runbook-pack/v0.1":
+        raise RunError(f"{pack.spec} packs cannot be run yet (RBT-WP-0007)")
     operator = _clean(operator.strip(), "operator")
     if not re.fullmatch(r"\S+", operator):
         raise RunError("operator must be one word (name or handle)")

rtut/ui.py

@@ -48,6 +48,9 @@ class App:
             d = p.data
             if not p.ok:
                 rows.append(f"<li class=bad>{e(p.id)}: invalid ({e(p.errors[0])})</li>")
+            elif d.get("spec") != "runbook-pack/v0.1" and d["engine"] == "native":
+                rows.append(f"<li>{e(d['title'])} <span class=pill>{e(d.get('spec'))}</span> "
+                            "<span class=note>cannot be run yet (RBT-WP-0007)</span></li>")
             elif d["engine"] != "native":
                 rows.append(f"<li>{e(d['title'])} <span class=pill>legacy console</span> "
                             f"<span class=note>start with: python -m rtut launch {e(p.path)}</span></li>")

rtut/v2.py

@@ -0,0 +1,213 @@
+"""Validation for runbook-pack/v0.2: JSON Schema plus semantic checks."""
+from __future__ import annotations
+
+import json
+import re
+from pathlib import Path
+
+from jsonschema import Draft202012Validator
+
+from .pack import PLACEHOLDER, SECRET_RE
+
+SCHEMA_PATH = Path(__file__).resolve().parent.parent / "schema" / "runbook-pack-v0.2.schema.json"
+MAX_DEPTH = 3
+
+
+def schema_errors(data: dict) -> list[str]:
+    v = Draft202012Validator(json.loads(SCHEMA_PATH.read_text()))
+    return [f"{'/'.join(map(str, e.absolute_path)) or '<root>'}: {e.message}"
+            for e in sorted(v.iter_errors(data), key=lambda e: list(map(str, e.absolute_path)))]
+
+
+class _Model:
+    """Flattened view of a v0.2 pack used by the semantic checks."""
+
+    def __init__(self, data: dict):
+        self.data = data
+        self.errs: list[str] = []
+        self.flow: dict[str, dict] = {}     # stages, steps, gates share one id space
+        self.kind: dict[str, str] = {}
+        self.parent: dict[str, str | None] = {}
+        self.children: dict[str, list[str]] = {}
+        self.alts: dict[str, str] = {}      # alternative id -> parent step id
+        self.depth_ok = True
+
+    def add(self, id_: str, kind: str, node: dict, parent: str | None) -> None:
+        if id_ in self.flow or id_ in self.alts:
+            self.errs.append(f"duplicate id: {id_}")
+            return                       # keep the first definition; never overwrite the tree
+        self.flow[id_] = node
+        self.kind[id_] = kind
+        self.parent[id_] = parent
+        if parent:
+            self.children.setdefault(parent, []).append(id_)
+
+    def walk(self, stages: list[dict], parent: str | None, depth: int) -> None:
+        for st in stages:
+            if depth > MAX_DEPTH:
+                self.errs.append(f"stage {st['id']}: nesting deeper than {MAX_DEPTH} levels")
+            self.add(st["id"], "stage", st, parent)
+            for s in st.get("steps", []):
+                self.add(s["id"], "step", s, st["id"])
+                for a in s.get("one_of", []):
+                    if a["id"] in self.flow or a["id"] in self.alts:
+                        self.errs.append(f"duplicate id: {a['id']}")
+                    self.alts[a["id"]] = s["id"]
+            for g in st.get("gates", []):
+                self.add(g["id"], "gate", g, st["id"])
+            self.walk(st.get("stages", []), st["id"], depth + 1)
+
+    def resolve(self, ref: str) -> str | None:
+        """requires reference -> flow id, or None if unresolved."""
+        if ref.startswith("gate:"):
+            g = ref[5:]
+            return g if self.kind.get(g) == "gate" else None
+        return ref if self.kind.get(ref) in ("stage", "step") else None
+
+
+def _check_refs(m: _Model, data: dict) -> None:
+    actors = {a["id"] for a in data.get("actors", [])}
+    dps = {d["id"]: d for d in data.get("datapoints", [])}
+    arts = {a["id"] for a in data.get("artefacts", [])}
+    steps = {i for i, k in m.kind.items() if k == "step"} | set(m.alts)
+    gates = {i for i, k in m.kind.items() if k == "gate"}
+    stages = {i for i, k in m.kind.items() if k == "stage"}
+    for label, ids in (("actor", [a["id"] for a in data.get("actors", [])]),
+                       ("datapoint", list(dps)), ("artefact", [a["id"] for a in data.get("artefacts", [])])):
+        if len(ids) != len(set(ids)):
+            m.errs.append(f"duplicate {label} id")
+
+    def check_ck(ck: dict, where: str, uses: set[str]) -> None:
+        exp = ck.get("expect", "manual")
+        if exp in ("exit-0", "output-contains") and not ck.get("command"):
+            m.errs.append(f"{where}: verify expect {exp} needs a command")
+        if exp == "output-contains" and not ck.get("contains"):
+            m.errs.append(f"{where}: verify output-contains needs 'contains'")
+        for d in ck.get("datapoints_ok", []):
+            if d not in dps:
+                m.errs.append(f"{where}: datapoints_ok names unknown datapoint {d}")
+        for t in (ck.get("command", ""), ck.get("contains", "")):
+            for name in PLACEHOLDER.findall(t or ""):
+                if name not in dps:
+                    m.errs.append(f"{where}: undefined placeholder {{{{{name}}}}}")
+                elif uses and name not in uses:
+                    m.errs.append(f"{where}: placeholder {name} not listed in uses")
+
+    def check_text(text: str, where: str, uses: set[str]) -> None:
+        for name in PLACEHOLDER.findall(text or ""):
+            if name not in dps:
+                m.errs.append(f"{where}: undefined placeholder {{{{{name}}}}}")
+            elif name not in uses:
+                m.errs.append(f"{where}: placeholder {name} not listed in uses")
+
+    for id_, node in m.flow.items():
+        kind = m.kind[id_]
+        if kind == "step":
+            if node["actor"] not in actors:
+                m.errs.append(f"step {id_}: unknown actor {node['actor']}")
+            uses = set(node.get("uses", []))
+            for u in uses:
+                if u not in dps:
+                    m.errs.append(f"step {id_}: uses unknown datapoint {u}")
+            for p in node.get("produces", []):
+                if p not in dps and p not in arts:
+                    m.errs.append(f"step {id_}: produces unknown datapoint or artefact {p}")
+            parts = [(node, id_)] + [(a, a["id"]) for a in node.get("one_of", [])]
+            for part, pid in parts:
+                check_text(part.get("command", ""), f"step {pid}", uses)
+                check_text(part.get("rollback", ""), f"step {pid}", uses)
+                if "verify" in part:
+                    check_ck(part["verify"], f"step {pid}", uses)
+                if part.get("reversible", node.get("reversible", True)) is False:
+                    if not (node.get("confirm_phrase") or data.get("confirm_phrase")):
+                        m.errs.append(f"step {pid}: irreversible step needs a confirm_phrase (step or pack)")
+                for k, vals in (part.get("recommended_when") or {}).items():
+                    if k not in dps:
+                        m.errs.append(f"step {pid}: recommended_when names unknown datapoint {k}")
+                    elif dps[k].get("values"):
+                        for v in vals:
+                            if v not in dps[k]["values"]:
+                                m.errs.append(f"step {pid}: recommended_when {k}={v} not in values")
+            if node.get("one_of") and node.get("command"):
+                m.errs.append(f"step {id_}: a step with one_of must not also have its own command")
+        elif kind == "gate":
+            by = (node.get("waivable") or {}).get("by")
+            if by and by not in actors:
+                m.errs.append(f"gate {id_}: waivable.by names unknown actor {by}")
+            for ck in node["checks"]:
+                check_ck(ck, f"gate {id_}", set())
+    for d in dps.values():
+        if d["type"] == "enum" and not d.get("values"):
+            m.errs.append(f"datapoint {d['id']}: enum without values")
+        dv = d.get("default")
+        if dv is not None:
+            if d.get("values") and dv not in d["values"]:
+                m.errs.append(f"datapoint {d['id']}: default not in values")
+            if d.get("pattern") and isinstance(dv, str) and not re.fullmatch(d["pattern"], dv):
+                m.errs.append(f"datapoint {d['id']}: default violates pattern")
+        if d.get("ask_in") and d["ask_in"] not in stages:
+            m.errs.append(f"datapoint {d['id']}: ask_in names unknown stage {d['ask_in']}")
+        if d.get("produced_by") and d["produced_by"] not in steps:
+            m.errs.append(f"datapoint {d['id']}: produced_by names unknown step {d['produced_by']}")
+        if d.get("confirmed_by") and d["confirmed_by"] not in steps | gates:
+            m.errs.append(f"datapoint {d['id']}: confirmed_by names unknown step or gate")
+    for a in data.get("artefacts", []):
+        if a["location"]["datapoint"] not in dps:
+            m.errs.append(f"artefact {a['id']}: location names unknown datapoint")
+        if a["produced_by"] not in steps:
+            m.errs.append(f"artefact {a['id']}: produced_by names unknown step")
+        if a.get("lives_until") and a["lives_until"] not in steps:
+            m.errs.append(f"artefact {a['id']}: lives_until names unknown step")
+
+
+def _check_graph(m: _Model) -> None:
+    deps: dict[str, set[str]] = {i: set() for i in m.flow}
+    for id_, node in m.flow.items():
+        for ref in node.get("requires", []):
+            r = m.resolve(ref)
+            if r is None:
+                m.errs.append(f"{m.kind[id_]} {id_}: requires unknown item {ref}")
+            else:
+                deps[id_].add(r)
+        # a node also waits for whatever its enclosing stages wait for
+        p, seen = m.parent[id_], set()
+        while p and p not in seen:
+            seen.add(p)
+            for ref in m.flow[p].get("requires", []):
+                r = m.resolve(ref)
+                if r:
+                    deps[id_].add(r)
+            p = m.parent[p]
+        # a stage is finished when its contents are
+        for c in m.children.get(id_, []):
+            deps[id_].add(c)
+    state: dict[str, int] = {}
+
+    def visit(n: str, path: list[str]) -> None:
+        if state.get(n) == 2:
+            return
+        if state.get(n) == 1:
+            m.errs.append("requires cycle: " + " -> ".join(path[path.index(n):] + [n]))
+            return
+        state[n] = 1
+        for d in sorted(deps[n]):
+            visit(d, path + [n])
+        state[n] = 2
+
+    for n in sorted(deps):
+        visit(n, [])
+
+
+def validate(data: dict) -> list[str]:
+    errs = schema_errors(data)
+    if errs:
+        return errs
+    if data["engine"] == "legacy-command":
+        return []
+    m = _Model(data)
+    m.walk(data["stages"], None, 1)
+    _check_refs(m, data)
+    _check_graph(m)
+    if SECRET_RE.search(json.dumps(data)):
+        m.errs.append("contains secret-looking marker")
+    return sorted(set(m.errs))