Maturity from run history, pack revision digests, attribute sets, projects and templates (RBT-WP-0008)

Also fixes run lists and scores mixing runs of different packs in one repository.

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/attrs.py

@@ -0,0 +1,142 @@
+"""Attribute sets: named, composable, non-secret attribute collections with layered resolution.
+
+An attribute has a *requirement* (required | optional), declared, and a *state*, derived with the same
+words as run datapoints: nil (no value), set, err (invalid for a consuming runbook). `modified` marks a
+value that a later layer replaced. Every value keeps its source, so a wrong value is traceable.
+"""
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass, field
+from pathlib import Path
+
+import yaml
+from jsonschema import Draft202012Validator
+
+from . import pack as P
+from .pack import SECRET_RE
+
+SCHEMA_DIR = Path(__file__).resolve().parent.parent / "schema"
+
+
+def _schema_errors(data: dict, schema_file: str) -> list[str]:
+    v = Draft202012Validator(json.loads((SCHEMA_DIR / schema_file).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)))]
+
+
+def normalize(attrs: dict | None) -> dict[str, dict]:
+    """scalar or {value, requirement, note} -> {value, requirement, note}"""
+    out = {}
+    for k, v in (attrs or {}).items():
+        out[k] = dict(v) if isinstance(v, dict) else {"value": v}
+        out[k].setdefault("value", None)
+        out[k].setdefault("requirement", None)
+    return out
+
+
+@dataclass
+class Doc:
+    path: Path
+    data: dict
+    errors: list[str] = field(default_factory=list)
+
+    @property
+    def id(self) -> str:
+        return str(self.data.get("id", self.path.stem))
+
+    @property
+    def ok(self) -> bool:
+        return not self.errors
+
+
+def _load(path: Path, schema_file: str) -> Doc:
+    try:
+        data = yaml.safe_load(Path(path).read_text())
+    except yaml.YAMLError as e:
+        return Doc(Path(path), {}, [f"invalid YAML: {e}"])
+    if not isinstance(data, dict):
+        return Doc(Path(path), {}, ["must be a mapping"])
+    errs = _schema_errors(data, schema_file)
+    if not errs and SECRET_RE.search(json.dumps(data)):
+        errs.append("contains secret-looking marker")
+    return Doc(Path(path), data, errs)
+
+
+def load_set(path: Path) -> Doc:
+    return _load(path, "attribute-set.schema.json")
+
+
+class Registry:
+    """Everything found under one or more repository roots."""
+
+    def __init__(self, roots: list[Path]):
+        self.roots = [Path(r) for r in roots]
+        self.sets: dict[str, Doc] = {}
+        self.projects: dict[str, Doc] = {}
+        self.templates: dict[str, Doc] = {}
+        self.packs: dict[str, P.Pack] = {}
+        for r in self.roots:
+            for f in sorted(r.glob("attribute-sets/*.yaml")):
+                d = load_set(f); self.sets[d.id] = d
+            for f in sorted(r.glob("projects/*/project.yaml")):
+                d = _load(f, "project.schema.json"); self.projects[d.id] = d
+            for f in sorted(r.glob("project-templates/*.yaml")):
+                d = _load(f, "project.schema.json"); self.templates[d.id] = d
+            for f in P.discover(r):
+                pk = P.load(f); self.packs[pk.id] = pk
+
+
+def set_layers(reg: Registry, set_id: str, _seen: tuple = ()) -> list[tuple[str, dict]]:
+    """Layers of an attribute set, its `extends` first. Raises on unknown sets and cycles."""
+    if set_id in _seen:
+        raise ValueError("attribute set cycle: " + " -> ".join(_seen + (set_id,)))
+    doc = reg.sets.get(set_id)
+    if doc is None:
+        raise ValueError(f"unknown attribute set {set_id}")
+    if not doc.ok:
+        raise ValueError(f"attribute set {set_id} is invalid: {doc.errors[0]}")
+    layers: list[tuple[str, dict]] = []
+    for base in doc.data.get("extends", []):
+        layers += set_layers(reg, base, _seen + (set_id,))
+    layers.append((f"set:{set_id}", normalize(doc.data["attributes"])))
+    return layers
+
+
+def fold(layers: list[tuple[str, dict]]) -> dict[str, dict]:
+    """Later layers win. Keeps provenance and marks a value replaced by a later layer as modified."""
+    out: dict[str, dict] = {}
+    for source, attrs in layers:
+        for name, a in attrs.items():
+            cur = out.setdefault(name, {"value": None, "requirement": None, "source": None, "modified": False,
+                                        "history": []})
+            if a.get("requirement"):
+                cur["requirement"] = a["requirement"]
+            if a.get("value") is not None:
+                if cur["value"] is not None and cur["value"] != a["value"]:
+                    cur["modified"] = True
+                cur["value"] = a["value"]
+                cur["source"] = source
+                cur["history"].append({"source": source, "value": a["value"]})
+    for cur in out.values():
+        cur["state"] = "set" if cur["value"] is not None else "nil"
+    return out
+
+
+def check_value(pack: P.Pack, name: str, value) -> str | None:
+    """None if the value is valid for the pack's datapoint/parameter `name`, else the reason."""
+    defs = {d["id"]: d for d in (pack.data.get("datapoints") or pack.data.get("parameters") or [])}
+    d = defs.get(name)
+    if d is None:
+        return None
+    v = str(value).lower() if isinstance(value, bool) else str(value)
+    import re
+    if d["type"] == "int" and not re.fullmatch(r"-?\d+", v):
+        return "not an integer"
+    if d["type"] == "bool" and v not in ("true", "false"):
+        return "must be true or false"
+    try:
+        P.render("{{%s}}" % name, {name: v}, pack)
+    except (ValueError, KeyError) as e:
+        return str(e)
+    return None