MASON-WP-0004-T02: draft custody descriptions and a metadata-only apply tool

custody-describe.py drafts the five custody fields from CCRs, the routing
catalog and the owner proposals (no OpenBao access), and applies accepted
entries with bao kv metadata patch only, silently. 29/38 drafts complete;
gaps left empty for owners. 4 tests.

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

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 327203@bnt-lap001
Assistant-Session: 55028ae3-4914-4bd1-8f5c-815f3527aef0

scripts/custody-describe.py

@@ -0,0 +1,152 @@
+#!/usr/bin/env python3
+"""custody-describe — draft and apply the five custody fields (MASON-WP-0004-T02).
+
+    ./scripts/custody-describe.py draft  > manifests/custody-descriptions.yaml
+    ./scripts/custody-describe.py check  manifests/custody-descriptions.yaml
+    ./scripts/custody-describe.py apply  manifests/custody-descriptions.yaml   # founder
+
+`draft` reads only documents: railiance-platform CCRs, the ops-warden routing
+catalog and the owner proposals. It never calls OpenBao. Every entry it writes
+is `review: draft`; a field it cannot source stays empty rather than guessed,
+because a confidently wrong `used_by` is worse than an empty one.
+
+`apply` writes custom_metadata only (`bao kv metadata patch`), for entries
+marked `review: accepted` with all five fields filled. It never reads or writes
+a value, prints nothing but paths and booleans, and runs under the founder's
+admin session (warden access openbao-platform-admin-login --exec), since
+ops-mason-survey cannot write metadata.
+"""
+
+from __future__ import annotations
+
+import glob
+import os
+import re
+import subprocess
+import sys
+
+import yaml
+
+REQUIRED = ("description", "owner", "used_by", "rotation", "on_loss")
+MAX_VALUE = 512  # OpenBao custom_metadata value limit
+HOME = os.path.expanduser("~")
+CCR_DIR = f"{HOME}/railiance-platform/credential-change-requests"
+CATALOG = f"{HOME}/ops-warden/registry/routing/catalog.yaml"
+PROPOSALS = os.path.join(os.path.dirname(__file__), "..", "docs", "credential-owner-proposals-2026-10-09.md")
+
+
+def _clip(text: str | None) -> str:
+    text = " ".join(str(text or "").split())
+    return text if len(text) <= MAX_VALUE else text[: MAX_VALUE - 1] + "…"
+
+
+def _first_sentence(text: str | None) -> str:
+    text = " ".join(str(text or "").split())
+    m = re.match(r"(.+?\.)(\s|$)", text)
+    return _clip(m.group(1) if m else text)
+
+
+def _proposals() -> dict[str, str]:
+    """Path -> proposed owner, from section C of the owner proposals."""
+    owners = {}
+    for line in open(PROPOSALS):
+        m = re.match(r"\| `([^`]+)` \| [^|]+ \| ([a-z0-9-]+) \(consumer", line)
+        if m:
+            owners[m.group(1)] = m.group(2)
+    return owners
+
+
+def _ccrs() -> dict[str, list[tuple[str, dict]]]:
+    by_path: dict[str, list[tuple[str, dict]]] = {}
+    for f in sorted(glob.glob(f"{CCR_DIR}/CCR-*.yaml")):
+        d = yaml.safe_load(open(f))
+        if not isinstance(d, dict):
+            continue
+        bao = d.get("openbao") or {}
+        path = bao.get("kv_path") or bao.get("path")
+        if path:
+            by_path.setdefault(path, []).append((os.path.basename(f), d))
+    return by_path
+
+
+def _catalog() -> dict[str, dict]:
+    cat = yaml.safe_load(open(CATALOG))
+    entries = cat.get("entries") if isinstance(cat, dict) else cat
+    return {str(e.get("path_template")): e for e in entries or [] if e.get("path_template")}
+
+
+def draft() -> dict:
+    ccrs, catalog, out = _ccrs(), _catalog(), {}
+    for path, owner in _proposals().items():
+        entry = {"review": "draft", "sources": []}
+        fields = dict.fromkeys(REQUIRED, "")
+        fields["owner"] = owner
+        if path in ccrs:
+            # The first CCR placed the path; later ones add readers.
+            name, d = ccrs[path][0]
+            fields["description"] = _first_sentence((d.get("target") or {}).get("purpose") or d.get("title"))
+            fields["used_by"] = _clip((d.get("delivery") or {}).get("target"))
+            refs = [f"railiance-platform/credential-change-requests/{n}#lifecycle.rotate" for n, _ in ccrs[path]]
+            fields["rotation"] = _clip("; ".join(refs[:2]))
+            rotate = (d.get("lifecycle") or {}).get("rotate")
+            if rotate:
+                fields["on_loss"] = _clip(f"Re-issue per rotation: {rotate}")
+            entry["sources"] += [n for n, _ in ccrs[path]]
+        cat = catalog.get(path)
+        if cat:
+            fields["description"] = fields["description"] or _first_sentence(cat.get("title"))
+            rot = cat.get("rotation") or {}
+            if not fields["rotation"] and cat.get("wiki_ref"):
+                fields["rotation"] = _clip(f"ops-warden/{cat['wiki_ref']}")
+            if not fields["on_loss"] and rot.get("steps"):
+                fields["on_loss"] = _clip(f"{rot.get('method', '')} by {rot.get('owner', '')}: {rot['steps'][0]}")
+            entry["sources"].append(f"ops-warden catalog {cat.get('id')}")
+        entry.update(fields)
+        entry["missing"] = [k for k in REQUIRED if not entry[k]]
+        out[path] = entry
+    return out
+
+
+def check(doc: dict) -> int:
+    bad = 0
+    for path, e in doc.items():
+        missing = [k for k in REQUIRED if not e.get(k)]
+        long = [k for k in REQUIRED if len(str(e.get(k, ""))) > MAX_VALUE]
+        state = e.get("review")
+        if missing or long or state != "accepted":
+            bad += 1
+        print(f"{'ok ' if not (missing or long) and state == 'accepted' else '-- '}{path}  review={state}"
+              f"{'  missing=' + ','.join(missing) if missing else ''}{'  too-long=' + ','.join(long) if long else ''}")
+    print(f"{len(doc) - bad}/{len(doc)} ready to apply")
+    return 0 if bad == 0 else 1
+
+
+def apply(doc: dict) -> int:
+    failed = 0
+    for path, e in doc.items():
+        if e.get("review") != "accepted" or any(not e.get(k) for k in REQUIRED):
+            print(f"skip {path}")
+            continue
+        mount, rest = path.split("/", 1)
+        args = ["bao", "kv", "metadata", "patch", f"-mount={mount}"]
+        args += [f"-custom-metadata={k}={_clip(e[k])}" for k in REQUIRED]
+        args.append(rest)
+        ok = subprocess.run(args, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL).returncode == 0
+        failed += not ok
+        print(f"{'patched' if ok else 'FAILED '} {path}")
+    return 1 if failed else 0
+
+
+def main(argv: list[str]) -> int:
+    if len(argv) >= 1 and argv[0] == "draft":
+        yaml.safe_dump(draft(), sys.stdout, sort_keys=True, allow_unicode=True, width=100)
+        return 0
+    if len(argv) == 2 and argv[0] in ("check", "apply"):
+        doc = yaml.safe_load(open(argv[1])) or {}
+        return check(doc) if argv[0] == "check" else apply(doc)
+    print("usage: custody-describe.py draft | check FILE | apply FILE", file=sys.stderr)
+    return 2
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv[1:]))