MASON-WP-0008: generate the construction-plan existing-structure survey

scripts/plan-survey.py runs under the ops-mason-build grant and prints a
pasteable section 2: relevant secrets/auth/policy/role listings with counts,
policy stanzas covering the target for name-matched policies, KV metadata
existence, matching ops-warden catalog entries, timestamp, BAO_ADDR, token
display_name and a draft yes/no/partial answer. Metadata only: an allowlist
refuses kv get, */data/* and role-id/secret-id before any process starts; a
non-grant token is refused and the token is never printed. Denied queries
degrade to 'not permitted by ops-mason-build'. Tests use a fake bao runner.

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

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 1170398@bnt-lap001
Assistant-Session: 622cd705-cb31-4322-ba08-dd42e44fa583

AGENTS.md

@@ -250,6 +250,11 @@ and every action lands in the audit log as you.
 ./scripts/bao-session.sh revoke          # end it immediately
 ```
 
+With a grant in place, generate a construction plan's section 2 with
+`./scripts/plan-survey.py --kv-path <path> --policy <name> --approle <role>
+--consumer <repo> [--keyword <word>]` (metadata only; see
+`docs/construction-plan-format.md` § 2).
+
 `grant` mints a **separate** token into `~/.claude-bao-token` — your own
 `~/.vault-token` is untouched, and the two revoke independently. The agent uses
 it as `BAO_TOKEN=$(cat ~/.claude-bao-token) bao <cmd>`.

docs/construction-plan-format.md

@@ -60,6 +60,37 @@ anything new:
   minting a new one, because the scoping already covered the new
   consumer.)
 
+Generate the section rather than gathering it by hand (MASON-WP-0008). The
+founder mints the scoped grant; the agent then runs the survey with it:
+
+```bash
+./scripts/bao-session.sh grant <task>       # founder: ops-mason-build, 45m, claude-<task>
+./scripts/plan-survey.py \
+    --kv-path platform/workloads/<consumer>/<name> \
+    --policy workload-kv-read-<consumer>-<name> \
+    --approle <role> [--k8s-role <role>] \
+    --consumer <consumer-repo> [--keyword <word> ...] \
+    [--catalog ~/ops-warden/registry/routing/catalog.yaml]
+```
+
+It prints a Markdown block headed `## 2. Existing-structure survey`: the
+relevant lines and counts of `bao secrets list`, `bao auth list`,
+`bao policy list` and `bao list auth/<mount>/role` (approle and kubernetes
+mounts), `bao policy read` stanzas covering the target for name-matched
+policies only, KV metadata existence for the target and its siblings,
+matching ops-warden catalog entries, a UTC timestamp, `BAO_ADDR`, the token
+`display_name`, and a draft yes/no/partial answer with reasons. Edit the
+draft answer into a decision; it is generated evidence, not a review.
+
+The survey refuses to run without a grant, refuses a token whose policies
+are anything but `ops-mason-build` (plus `default`), and never prints the
+token. Every bao call passes an allowlist that rejects `kv get` and any
+`*/data/*` path before a process starts. A query the grant does not permit
+is reported as `not permitted by ops-mason-build` and listed under
+"incomplete" rather than read as an empty answer. Policies with names that
+do not match the demand are not read, so a broadly named policy covering
+the path is not detected; the output says so.
+
 ### 3. Proposed changes
 
 A list, each item tagged `create` / `modify` / `tear-down`, each with a

scripts/plan-survey.py

@@ -0,0 +1,503 @@
+#!/usr/bin/env python3
+"""plan-survey — section 2 of a construction plan, generated from live metadata.
+
+    ./scripts/bao-session.sh grant <task>
+    ./scripts/plan-survey.py --kv-path platform/workloads/<consumer>/<name> \\
+        --policy <proposed-policy> [--approle <role>] [--k8s-role <role>] \\
+        --consumer <repo> [--keyword <word> ...] [--catalog <catalog.yaml>]
+
+Prints a Markdown block to paste as "## 2. Existing-structure survey": the
+relevant lines of `bao secrets list`, `bao auth list`, `bao policy list` and the
+auth-role listings, KV metadata existence for the target path, matching
+ops-warden routing catalog entries, and a draft answer to "does an existing lane
+already satisfy this demand?".
+
+Metadata only. It runs under the ops-mason-build grant (~/.claude-bao-token),
+refuses any other token, and every bao invocation passes `allowed_bao_call`, an
+allowlist that rejects `kv get` and every `*/data/*` path before a process starts.
+Policy bodies are read with `bao policy read` (policy text, never a secret value)
+and only for name-matched candidates. The token itself is never printed.
+"""
+
+from __future__ import annotations
+
+import argparse
+import datetime as _dt
+import json
+import os
+import re
+import subprocess
+import sys
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Callable
+
+GRANT_FILE = "~/.claude-bao-token"
+GRANT_POLICY = "ops-mason-build"
+DEFAULT_ADDR = "http://127.0.0.1:18200"
+DEFAULT_CATALOG = "~/ops-warden/registry/routing/catalog.yaml"
+MAX_POLICY_READS = 15
+NOT_PERMITTED = "not permitted by ops-mason-build"
+# Path segments too generic to identify a consumer.
+GENERIC = {"platform", "workloads", "operators", "secret", "data", "metadata",
+           "kv", "read", "workload", "role", "token", "auth", "policy"}
+ROLE_FIELDS = ("token_policies", "policies", "token_ttl", "token_max_ttl",
+               "token_num_uses", "secret_id_ttl", "bound_service_account_names",
+               "bound_service_account_namespaces", "audience")
+
+
+class SurveyError(RuntimeError):
+    """The survey could not be completed; never report this as an empty store."""
+
+
+@dataclass
+class BaoResult:
+    status: str  # ok | denied | missing | error
+    data: object = None
+    text: str = ""
+
+
+Runner = Callable[[list[str], dict], tuple[int, str, str]]
+
+
+# --- metadata-only guard -----------------------------------------------------
+
+def allowed_bao_call(args: list[str]) -> bool:
+    """Allowlist of the bao invocations the survey may make. Flags are ignored."""
+    pos = [a for a in args if not a.startswith("-")]
+    if any("/data/" in a or a.startswith("data/") or a.endswith("/data") for a in pos):
+        return False
+    if any(seg in ("role-id", "secret-id", "secret-id-accessor") for a in pos for seg in a.split("/")):
+        return False
+    shape = tuple(pos[:3])
+    if pos in (["token", "lookup"], ["secrets", "list"], ["auth", "list"], ["policy", "list"]):
+        return True
+    if len(pos) == 3 and shape[:2] == ("policy", "read"):
+        return True
+    if len(pos) == 3 and shape[:2] == ("kv", "list"):
+        return True
+    if len(pos) == 4 and shape == ("kv", "metadata", "get"):
+        return True
+    if len(pos) == 2 and pos[0] == "list":
+        parts = pos[1].strip("/").split("/")
+        return len(parts) == 3 and parts[0] == "auth" and parts[2] == "role"
+    if len(pos) == 2 and pos[0] == "read":
+        parts = pos[1].strip("/").split("/")
+        return len(parts) == 4 and parts[0] == "auth" and parts[2] == "role"
+    return False
+
+
+def subprocess_runner(argv: list[str], env: dict) -> tuple[int, str, str]:
+    try:
+        p = subprocess.run(argv, capture_output=True, text=True, timeout=30, env=env)
+    except (OSError, subprocess.TimeoutExpired) as exc:
+        raise SurveyError("bao CLI unavailable or timed out; survey was not run") from exc
+    return p.returncode, p.stdout, p.stderr
+
+
+class Bao:
+    def __init__(self, runner: Runner, env: dict):
+        self.runner = runner
+        self.env = env
+
+    def call(self, *args: str, as_json: bool = True) -> BaoResult:
+        args_l = list(args)
+        if not allowed_bao_call(args_l):
+            raise SurveyError(f"refused non-metadata bao call: bao {' '.join(args_l[:2])}")
+        # The CLI rejects flags after a positional argument: -format goes right
+        # after the subcommand words.
+        if as_json:
+            words = 3 if args_l[:2] == ["kv", "metadata"] else (1 if args_l[0] in ("list", "read") else 2)
+            args_l = args_l[:words] + ["-format=json"] + args_l[words:]
+        rc, out, err = self.runner(["bao", *args_l], self.env)
+        if rc != 0:
+            low = (err or "").lower()
+            if "permission denied" in low or "403" in low:
+                return BaoResult("denied")
+            if "no value found" in low or "404" in low or (not out.strip() and rc == 2):
+                return BaoResult("missing")
+            return BaoResult("error")
+        if not as_json:
+            return BaoResult("ok", text=out)
+        try:
+            return BaoResult("ok", data=json.loads(out) if out.strip() else None)
+        except json.JSONDecodeError:
+            return BaoResult("error")
+
+
+def grant_env(environ: dict | None = None) -> dict:
+    env = dict(os.environ if environ is None else environ)
+    env.setdefault("BAO_ADDR", DEFAULT_ADDR)
+    grant = os.path.expanduser(GRANT_FILE)
+    if not env.get("BAO_TOKEN") and os.path.exists(grant):
+        with open(grant) as f:
+            env["BAO_TOKEN"] = f.read().strip()
+    if not env.get("BAO_TOKEN"):
+        raise SurveyError("no ops-mason-build grant (~/.claude-bao-token absent and BAO_TOKEN unset); "
+                          "ask the founder to run ./scripts/bao-session.sh grant <task>")
+    return env
+
+
+# --- policy text -------------------------------------------------------------
+
+_STANZA = re.compile(r'path\s+"([^"]+)"\s*\{(.*?)\}', re.S)
+_CAPS = re.compile(r'capabilities\s*=\s*\[([^\]]*)\]')
+
+
+def policy_stanzas(hcl: str) -> list[tuple[str, list[str]]]:
+    out = []
+    for path, body in _STANZA.findall(hcl):
+        m = _CAPS.search(body)
+        caps = re.findall(r'"([^"]+)"', m.group(1)) if m else []
+        out.append((path, caps))
+    return out
+
+
+def path_matches(pattern: str, path: str) -> bool:
+    regex = ""
+    for i, ch in enumerate(pattern):
+        if ch == "*" and i == len(pattern) - 1:
+            regex += ".*"
+        elif ch == "+":
+            regex += "[^/]+"
+        else:
+            regex += re.escape(ch)
+    return re.fullmatch(regex, path) is not None
+
+
+def kv_api_paths(kv_path: str, mount: str | None) -> tuple[str, str]:
+    mount = (mount or kv_path.split("/")[0]).strip("/")
+    rest = kv_path.strip("/")[len(mount):].strip("/")
+    return f"{mount}/data/{rest}", f"{mount}/metadata/{rest}"
+
+
+# --- catalog -----------------------------------------------------------------
+
+def load_catalog(path: Path) -> list[dict]:
+    if not path.exists():
+        raise SurveyError(f"ops-warden catalog not found at {path}; pass --catalog")
+    try:
+        import yaml  # type: ignore
+    except ImportError as exc:  # pragma: no cover - PyYAML is a project dependency
+        raise SurveyError("PyYAML unavailable; cannot read the ops-warden catalog") from exc
+    doc = yaml.safe_load(path.read_text())
+    entries = doc.get("entries") if isinstance(doc, dict) else None
+    if not isinstance(entries, list):
+        raise SurveyError(f"{path} has no entries list")
+    return [e for e in entries if isinstance(e, dict)]
+
+
+def catalog_matches(entries: list[dict], terms: list[str]) -> list[tuple[dict, list[str]]]:
+    out = []
+    for e in entries:
+        hay = " ".join(str(e.get(k, "")) for k in
+                       ("id", "title", "owner_repo", "subsystem", "path_template", "policy_ref",
+                        "auth_method", "canon_ref")).lower()
+        hay += " " + " ".join(str(k) for k in e.get("need_keywords") or []).lower()
+        hits = [t for t in terms if t.lower() in hay]
+        if hits:
+            out.append((e, hits))
+    return out
+
+
+# --- survey ------------------------------------------------------------------
+
+@dataclass
+class Demand:
+    kv_path: str
+    policy: str
+    consumer: str
+    approle: str | None = None
+    k8s_role: str | None = None
+    keywords: list[str] = field(default_factory=list)
+
+    def terms(self) -> list[str]:
+        segs = [s for s in self.kv_path.strip("/").split("/")[1:]]
+        raw = [self.consumer, self.approle, self.k8s_role, *self.keywords, *segs]
+        seen: list[str] = []
+        for t in raw:
+            t = (t or "").strip().lower()
+            if len(t) >= 3 and t not in GENERIC and t not in seen:
+                seen.append(t)
+        return seen
+
+
+def _fmt_ts(now: _dt.datetime) -> str:
+    return now.astimezone(_dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def survey(demand: Demand, bao: Bao, catalog_path: Path, now: _dt.datetime) -> tuple[str, str]:
+    terms = demand.terms()
+    names = {n for n in (demand.policy, demand.approle, demand.k8s_role) if n}
+
+    def relevant(name: str) -> list[str]:
+        low = name.lower()
+        return [t for t in terms if t in low] + (["proposed name"] if name.strip("/") in names else [])
+
+    # token identity — display_name and policies only, never id/accessor
+    tok = bao.call("token", "lookup")
+    if tok.status != "ok" or not isinstance(tok.data, dict) or not isinstance(tok.data.get("data"), dict):
+        raise SurveyError("grant token lookup failed (expired, revoked or unreachable); "
+                          "re-run ./scripts/bao-session.sh grant <task>")
+    td = tok.data["data"]
+    policies = sorted(str(p) for p in td.get("policies") or [])
+    if GRANT_POLICY not in policies or set(policies) - {GRANT_POLICY, "default"}:
+        raise SurveyError(f"token is not an {GRANT_POLICY} grant (policies: {', '.join(policies) or 'none'}); "
+                          "refusing to survey with a broader or different session")
+    display = str(td.get("display_name") or "-")
+
+    lines: list[str] = []
+    findings: dict[str, list[str]] = {"yes": [], "partial": [], "gaps": []}
+    L = lines.append
+
+    L("## 2. Existing-structure survey")
+    L("")
+    L(f"Generated by `scripts/plan-survey.py` at {_fmt_ts(now)} against "
+      f"`BAO_ADDR={bao.env.get('BAO_ADDR')}`, token display_name `{display}` "
+      f"(policies: {', '.join(policies)}). Metadata only: no `kv get` and no "
+      "`*/data/*` read was issued; policy bodies were read with `bao policy read` "
+      "for name-matched candidates only.")
+    L("")
+    L(f"Demand: KV path `{demand.kv_path}`, proposed policy `{demand.policy}`, "
+      f"AppRole `{demand.approle or '-'}`, Kubernetes role `{demand.k8s_role or '-'}`, "
+      f"consumer `{demand.consumer}`. Match terms: {', '.join(f'`{t}`' for t in terms) or '-'}.")
+    L("")
+
+    def block(title: str, body: list[str]) -> None:
+        L(f"### {title}")
+        L("")
+        L("```text")
+        lines.extend(body or ["(no relevant entries)"])
+        L("```")
+        L("")
+
+    # secrets list
+    mounts = bao.call("secrets", "list")
+    target_mount = None
+    if mounts.status == "ok" and isinstance(mounts.data, dict):
+        rows = []
+        for path, m in sorted(mounts.data.items()):
+            m = m if isinstance(m, dict) else {}
+            if demand.kv_path.strip("/").startswith(path.strip("/") + "/") and (
+                    target_mount is None or len(path) > len(target_mount)):
+                target_mount = path.strip("/")
+        for path, m in sorted(mounts.data.items()):
+            m = m if isinstance(m, dict) else {}
+            hit = path.strip("/") == target_mount or relevant(path)
+            if hit:
+                ver = (m.get("options") or {}).get("version", "") if isinstance(m.get("options"), dict) else ""
+                rows.append(f"{path:<24} {m.get('type', '?'):<10} "
+                            f"{'version=' + str(ver) if ver else '':<10} {m.get('description', '')}".rstrip())
+        block(f"`bao secrets list` ({len(mounts.data)} mounts, {len(rows)} relevant)", rows)
+        if target_mount is None:
+            findings["gaps"].append(f"no secrets engine is mounted at a prefix of `{demand.kv_path}`; "
+                                    "enabling one is a railiance-platform act")
+    else:
+        block("`bao secrets list`", [_status_line(mounts)])
+        findings["gaps"].append(f"`bao secrets list`: {_status_line(mounts)}")
+
+    # auth list
+    auth = bao.call("auth", "list")
+    role_mounts: list[tuple[str, str]] = []
+    if auth.status == "ok" and isinstance(auth.data, dict):
+        rows = []
+        for path, m in sorted(auth.data.items()):
+            m = m if isinstance(m, dict) else {}
+            typ = str(m.get("type", "?"))
+            if typ in ("approle", "kubernetes"):
+                role_mounts.append((path.strip("/"), typ))
+            if typ in ("approle", "kubernetes") or relevant(path):
+                rows.append(f"{path:<24} {typ:<12} {m.get('description', '')}".rstrip())
+        block(f"`bao auth list` ({len(auth.data)} auth mounts, {len(rows)} relevant)", rows)
+    else:
+        block("`bao auth list`", [_status_line(auth)])
+        findings["gaps"].append(f"`bao auth list`: {_status_line(auth)}")
+        role_mounts = [("approle", "approle")] + ([("kubernetes", "kubernetes")] if demand.k8s_role else [])
+
+    # policy list
+    plist = bao.call("policy", "list")
+    candidates: list[str] = []
+    if plist.status == "ok" and isinstance(plist.data, list):
+        rows = []
+        for name in sorted(str(p) for p in plist.data):
+            why = relevant(name)
+            if why:
+                candidates.append(name)
+                rows.append(f"{name:<56} matches: {', '.join(why)}")
+        block(f"`bao policy list` ({len(plist.data)} policies, {len(rows)} relevant)", rows)
+        if demand.policy in plist.data:
+            findings["partial"].append(f"proposed policy name `{demand.policy}` already exists")
+    else:
+        block("`bao policy list`", [_status_line(plist)])
+        findings["gaps"].append(f"`bao policy list`: {_status_line(plist)}")
+
+    # auth roles
+    bound: dict[str, list[str]] = {}
+    for mount, typ in role_mounts:
+        res = bao.call("list", f"auth/{mount}/role")
+        title = f"`bao list auth/{mount}/role` ({typ})"
+        if res.status == "missing":
+            block(f"{title} (0 roles)", [])
+            continue
+        if res.status != "ok" or not isinstance(res.data, list):
+            block(title, [_status_line(res)])
+            findings["gaps"].append(f"`bao list auth/{mount}/role`: {_status_line(res)}")
+            continue
+        rows = []
+        for role in sorted(str(r) for r in res.data):
+            why = relevant(role)
+            if not why:
+                continue
+            rows.append(f"{role:<48} matches: {', '.join(why)}")
+            if role in names:
+                findings["partial"].append(f"proposed role name `{role}` already exists on auth/{mount}")
+            detail = bao.call("read", f"auth/{mount}/role/{role}")
+            if detail.status == "ok" and isinstance(detail.data, dict) and isinstance(detail.data.get("data"), dict):
+                d = detail.data["data"]
+                for k in ROLE_FIELDS:
+                    if d.get(k) not in (None, "", []):
+                        rows.append(f"    {k}={d[k]}")
+                pols = d.get("token_policies") or d.get("policies") or []
+                for p in pols if isinstance(pols, list) else []:
+                    bound.setdefault(str(p), []).append(f"auth/{mount}/role/{role}")
+            else:
+                rows.append(f"    (role read: {_status_line(detail)})")
+        block(f"{title} ({len(res.data)} roles, {sum(1 for r in rows if not r.startswith(' '))} relevant)", rows)
+
+    # policy bodies for name-matched candidates
+    data_path, meta_path = kv_api_paths(demand.kv_path, target_mount)
+    granting: list[str] = []
+    rows = []
+    for name in candidates[:MAX_POLICY_READS]:
+        res = bao.call("policy", "read", name, as_json=False)
+        if res.status != "ok":
+            rows.append(f"{name}: {_status_line(res)}")
+            if res.status == "denied":
+                findings["gaps"].append(f"`bao policy read {name}`: {NOT_PERMITTED}")
+            continue
+        hits = [(p, c) for p, c in policy_stanzas(res.text)
+                if path_matches(p, data_path) or path_matches(p, meta_path)]
+        if hits:
+            for p, c in hits:
+                rows.append(f'{name}: path "{p}" capabilities={c}')
+            if any(path_matches(p, data_path) and "read" in c and "deny" not in c for p, c in hits):
+                granting.append(name)
+        else:
+            rows.append(f"{name}: no stanza covers {data_path}")
+    if len(candidates) > MAX_POLICY_READS:
+        rows.append(f"({len(candidates) - MAX_POLICY_READS} further candidates not read; narrow --keyword)")
+    block(f"`bao policy read` of name-matched candidates (target `{data_path}`)", rows)
+
+    # KV metadata for the target and its siblings
+    rows = []
+    meta = bao.call("kv", "metadata", "get", demand.kv_path)
+    kv_exists = False
+    if meta.status == "ok" and isinstance(meta.data, dict) and isinstance(meta.data.get("data"), dict):
+        kv_exists = True
+        d = meta.data["data"]
+        rows.append(f"{demand.kv_path}: exists, current_version={d.get('current_version', '?')}, "
+                    f"created={str(d.get('created_time', ''))[:10]}, updated={str(d.get('updated_time', ''))[:10]}")
+        cm = d.get("custom_metadata") or {}
+        for k in ("description", "owner", "used_by"):
+            if isinstance(cm, dict) and cm.get(k):
+                rows.append(f"    {k}: {cm[k]}")
+        findings["partial"].append(f"KV path `{demand.kv_path}` already exists (v{d.get('current_version', '?')})")
+    elif meta.status == "missing":
+        rows.append(f"{demand.kv_path}: no metadata (path unused)")
+    else:
+        rows.append(f"{demand.kv_path}: {_status_line(meta)}")
+        findings["gaps"].append(f"KV metadata for `{demand.kv_path}`: {_status_line(meta)}")
+    parent = demand.kv_path.strip("/").rsplit("/", 1)[0]
+    sib = bao.call("kv", "list", parent)
+    if sib.status == "ok" and isinstance(sib.data, list):
+        rows.append(f"{parent}/: {', '.join(str(s) for s in sib.data) or '(empty)'}")
+    else:
+        rows.append(f"{parent}/: {_status_line(sib)}")
+    block("`bao kv metadata get` / `bao kv list` (metadata only)", rows)
+
+    # ops-warden catalog
+    try:
+        entries = load_catalog(catalog_path)
+        cat_terms = terms + [t for t in (demand.kv_path, demand.policy) if t not in terms]
+        matches = catalog_matches(entries, cat_terms)
+        rows = []
+        for e, hits in matches:
+            rows.append(f"{e.get('id')}  owner={e.get('owner_repo')}  status={e.get('status')}  "
+                        f"matches: {', '.join(hits)}")
+            for k in ("path_template", "policy_ref", "auth_method"):
+                if e.get(k):
+                    rows.append(f"    {k}: {e[k]}")
+            if demand.kv_path in str(e.get("path_template", "")) or demand.policy == e.get("policy_ref"):
+                findings["partial"].append(f"ops-warden catalog entry `{e.get('id')}` already routes this path/policy")
+        block(f"ops-warden routing catalog `{catalog_path}` ({len(entries)} entries, {len(matches)} relevant)", rows)
+    except SurveyError as exc:
+        block("ops-warden routing catalog", [str(exc)])
+        findings["gaps"].append(str(exc))
+
+    # verdict
+    satisfied = [p for p in granting if p in bound]
+    if satisfied:
+        verdict = "yes"
+        for p in satisfied:
+            findings["yes"].append(f"policy `{p}` grants read on `{data_path}` and is bound by "
+                                   f"{', '.join(bound[p])}: bind to the existing lane instead of creating one")
+    elif granting or findings["partial"]:
+        verdict = "partial"
+        for p in granting:
+            findings["partial"].insert(0, f"policy `{p}` already grants read on `{data_path}` "
+                                          "but no surveyed role binds it: reuse the policy, add only a role")
+    else:
+        verdict = "no"
+    L("### Does an existing lane already satisfy this demand?")
+    L("")
+    L(f"**Draft answer: {verdict}.** Review before committing; this is generated, not a decision.")
+    L("")
+    for r in findings["yes"] + findings["partial"]:
+        L(f"- {r}")
+    if verdict == "no":
+        L(f"- no surveyed policy grants `{data_path}`, the proposed names are free, the KV path is unused "
+          "and no catalog entry routes it")
+    L(f"- policy grants were checked only for name-matched candidates; a policy with an unrelated name "
+      f"that covers `{data_path}` is not detected")
+    for g in findings["gaps"]:
+        L(f"- incomplete: {g}")
+    L("")
+    return "\n".join(lines), verdict
+
+
+def _status_line(res: BaoResult) -> str:
+    return {"denied": NOT_PERMITTED, "missing": "not found",
+            "error": "request failed (output suppressed)"}.get(res.status, res.status)
+
+
+def parse_args(argv: list[str]) -> argparse.Namespace:
+    p = argparse.ArgumentParser(prog="plan-survey", description=__doc__.split("\n\n")[0])
+    p.add_argument("--kv-path", required=True, help="target KV path, e.g. platform/workloads/x/y")
+    p.add_argument("--policy", required=True, help="proposed policy name")
+    p.add_argument("--approle", help="proposed AppRole name")
+    p.add_argument("--k8s-role", help="proposed Kubernetes-auth role name")
+    p.add_argument("--consumer", required=True, help="consumer repo")
+    p.add_argument("--keyword", action="append", default=[], help="extra match term (repeatable)")
+    p.add_argument("--catalog", default=DEFAULT_CATALOG, help="ops-warden routing catalog path")
+    return p.parse_args(argv)
+
+
+def main(argv: list[str] | None = None, runner: Runner = subprocess_runner,
+         environ: dict | None = None, now: _dt.datetime | None = None) -> int:
+    a = parse_args(sys.argv[1:] if argv is None else argv)
+    if "/data/" in a.kv_path or a.kv_path.startswith("/"):
+        raise SurveyError("--kv-path is the logical KV path (mount/path), not an API data path")
+    demand = Demand(a.kv_path.strip("/"), a.policy, a.consumer, a.approle, a.k8s_role, a.keyword)
+    bao = Bao(runner, grant_env(environ))
+    text, _ = survey(demand, bao, Path(os.path.expanduser(a.catalog)),
+                     now or _dt.datetime.now(_dt.timezone.utc))
+    print(text)
+    return 0
+
+
+if __name__ == "__main__":
+    try:
+        raise SystemExit(main())
+    except SurveyError as exc:
+        print(f"ERROR: {exc}", file=sys.stderr)
+        raise SystemExit(2)

tests/test_plan_survey.py

@@ -0,0 +1,216 @@
+"""plan-survey reads metadata only, never prints a token, and fails loudly."""
+import datetime as dt
+import importlib.util
+import json
+import sys
+from pathlib import Path
+
+import pytest
+
+spec = importlib.util.spec_from_file_location("plan_survey", Path(__file__).parents[1] / "scripts/plan-survey.py")
+survey = importlib.util.module_from_spec(spec)
+sys.modules["plan_survey"] = survey
+spec.loader.exec_module(survey)
+
+TOKEN = "hvs.SYNTHETICgrantTOKENvalue0123456789"
+NOW = dt.datetime(2026, 10, 5, 12, 0, tzinfo=dt.timezone.utc)
+ARGS = ["--kv-path", "platform/workloads/inbound-library/imap",
+        "--policy", "workload-kv-read-inbound-library-imap",
+        "--approle", "inbound-library", "--consumer", "inbound-library", "--keyword", "imap"]
+
+CATALOG = """
+version: 1
+entries:
+  - id: inbound-library-imap
+    title: inbound-library IMAP mailbox
+    need_keywords: [inbound-library, imap]
+    owner_repo: ops-mason
+    path_template: platform/workloads/inbound-library/imap
+    status: draft
+  - id: npm-publish-lane
+    title: something else
+    need_keywords: [npm]
+    owner_repo: railiance-platform
+    status: active
+"""
+
+POLICY_GRANTING = '''path "platform/data/workloads/inbound-library/imap" {
+  capabilities = ["read"]
+}
+path "platform/metadata/workloads/inbound-library/imap" {
+  capabilities = ["read"]
+}
+'''
+
+
+def lookup(policies=("default", "ops-mason-build")):
+    return {"data": {"id": TOKEN, "accessor": "acc-SECRET", "display_name": "token-claude-mason-wp-0008",
+                     "policies": list(policies), "ttl": 2000}}
+
+
+class FakeBao:
+    """Records every argv; answers from a table keyed by positional words."""
+
+    def __init__(self, table=None, denied=(), missing=()):
+        self.calls = []
+        self.table = {
+            ("token", "lookup"): lookup(),
+            ("secrets", "list"): {"platform/": {"type": "kv", "options": {"version": "2"}, "description": "workloads"},
+                                  "sys/": {"type": "system"}},
+            ("auth", "list"): {"approle/": {"type": "approle"}, "kubernetes/": {"type": "kubernetes"},
+                               "token/": {"type": "token"}},
+            ("policy", "list"): ["default", "root", "ops-mason-build", "workload-kv-read-inbound-library-imap",
+                                 "workload-kv-read-other"],
+            ("list", "auth/approle/role"): ["inbound-library", "rein-openweights"],
+            ("list", "auth/kubernetes/role"): ["external-secrets-activity-core"],
+            ("read", "auth/approle/role/inbound-library"): {"data": {
+                "token_policies": ["workload-kv-read-inbound-library-imap"], "token_ttl": 900}},
+            ("policy", "read", "workload-kv-read-inbound-library-imap"): POLICY_GRANTING,
+            ("kv", "metadata", "get", "platform/workloads/inbound-library/imap"): {"data": {
+                "current_version": 1, "created_time": "2026-10-01T00:00:00Z",
+                "custom_metadata": {"description": "IMAP login for inbound-library"}}},
+            ("kv", "list", "platform/workloads/inbound-library"): ["imap"],
+        }
+        self.table.update(table or {})
+        self.denied, self.missing = set(denied), set(missing)
+
+    def __call__(self, argv, env):
+        assert argv[0] == "bao"
+        assert env["BAO_TOKEN"] == TOKEN
+        self.calls.append(argv)
+        key = tuple(a for a in argv[1:] if not a.startswith("-"))
+        if key in self.denied:
+            return 2, "", "Error: permission denied"
+        if key in self.missing or key not in self.table:
+            return 2, "", "No value found"
+        value = self.table[key]
+        return 0, value if isinstance(value, str) else json.dumps(value), ""
+
+
+@pytest.fixture
+def catalog(tmp_path):
+    p = tmp_path / "catalog.yaml"
+    p.write_text(CATALOG)
+    return p
+
+
+def run(fake, catalog, capsys, args=ARGS):
+    rc = survey.main([*args, "--catalog", str(catalog)], runner=fake,
+                     environ={"BAO_TOKEN": TOKEN, "BAO_ADDR": "http://127.0.0.1:18200"}, now=NOW)
+    return rc, capsys.readouterr().out
+
+
+def assert_metadata_only(calls):
+    for argv in calls:
+        pos = [a for a in argv[1:] if not a.startswith("-")]
+        assert not (pos[:1] == ["kv"] and "get" in pos and "metadata" not in pos), argv
+        assert pos[:2] != ["kv", "get"], argv
+        assert not any("/data/" in a or a.startswith("data/") for a in pos), argv
+        assert not any(seg in ("role-id", "secret-id") for a in pos for seg in a.split("/")), argv
+
+
+def test_existing_lane_found_and_only_metadata_read(catalog, capsys):
+    fake = FakeBao()
+    rc, out = run(fake, catalog, capsys)
+    assert rc == 0
+    assert_metadata_only(fake.calls)
+    assert "## 2. Existing-structure survey" in out
+    assert "**Draft answer: yes.**" in out
+    assert "token-claude-mason-wp-0008" in out and "BAO_ADDR=http://127.0.0.1:18200" in out
+    assert "2026-10-05T12:00:00Z" in out
+    assert "(5 policies, 1 relevant)" in out
+    assert "inbound-library-imap  owner=ops-mason" in out and "npm-publish-lane" not in out
+    assert "rein-openweights" not in out
+    assert TOKEN not in out and "acc-SECRET" not in out
+
+
+def test_no_existing_lane(catalog, capsys, tmp_path):
+    empty = tmp_path / "empty.yaml"
+    empty.write_text("version: 1\nentries: []\n")
+    fake = FakeBao(table={("policy", "list"): ["default"], ("list", "auth/approle/role"): ["other"]},
+                   missing={("kv", "metadata", "get", "platform/workloads/inbound-library/imap"),
+                            ("kv", "list", "platform/workloads/inbound-library")})
+    rc, out = run(fake, empty, capsys)
+    assert rc == 0
+    assert "**Draft answer: no.**" in out
+    assert "no metadata (path unused)" in out
+    assert_metadata_only(fake.calls)
+
+
+def test_policy_without_bound_role_is_partial(catalog, capsys):
+    fake = FakeBao(table={("list", "auth/approle/role"): ["other"]})
+    rc, out = run(fake, catalog, capsys)
+    assert "**Draft answer: partial.**" in out
+    assert "no surveyed role binds it" in out
+
+
+def test_missing_capability_degrades_and_is_reported(catalog, capsys):
+    fake = FakeBao(denied={("list", "auth/approle/role"), ("list", "auth/kubernetes/role"),
+                           ("policy", "read", "workload-kv-read-inbound-library-imap")})
+    rc, out = run(fake, catalog, capsys)
+    assert rc == 0
+    assert "not permitted by ops-mason-build" in out
+    assert "incomplete:" in out
+
+
+def test_no_grant_fails_loudly(monkeypatch, tmp_path):
+    monkeypatch.setattr(survey.os.path, "expanduser", lambda p: str(tmp_path / "absent"))
+    with pytest.raises(survey.SurveyError, match="no ops-mason-build grant"):
+        survey.main(ARGS, runner=FakeBao(), environ={})
+
+
+@pytest.mark.parametrize("policies", [("root",), ("default", "ops-mason-build", "platform-admin"), ("default",)])
+def test_non_grant_token_is_refused(catalog, policies):
+    fake = FakeBao(table={("token", "lookup"): lookup(policies)})
+    with pytest.raises(survey.SurveyError, match="not an ops-mason-build grant") as exc:
+        survey.main([*ARGS, "--catalog", str(catalog)], runner=fake, environ={"BAO_TOKEN": TOKEN})
+    assert TOKEN not in str(exc.value)
+    assert len(fake.calls) == 1
+
+
+def test_expired_grant_fails_loudly(catalog):
+    fake = FakeBao(denied={("token", "lookup")})
+    with pytest.raises(survey.SurveyError, match="lookup failed"):
+        survey.main([*ARGS, "--catalog", str(catalog)], runner=fake, environ={"BAO_TOKEN": TOKEN})
+
+
+def test_script_exits_nonzero_without_grant(tmp_path):
+    import os
+    import subprocess
+    env = {k: v for k, v in os.environ.items() if k not in ("BAO_TOKEN", "VAULT_TOKEN")}
+    env["HOME"] = str(tmp_path)
+    p = subprocess.run([sys.executable, str(Path(__file__).parents[1] / "scripts/plan-survey.py"), *ARGS],
+                       capture_output=True, text=True, env=env)
+    assert p.returncode == 2
+    assert "ERROR: no ops-mason-build grant" in p.stderr
+
+
+@pytest.mark.parametrize("args", [
+    ["kv", "get", "platform/workloads/x"],
+    ["kv", "get", "-field=password", "platform/workloads/x"],
+    ["read", "platform/data/workloads/x"],
+    ["list", "platform/data/workloads"],
+    ["kv", "metadata", "get", "platform/data/x"],
+    ["read", "auth/approle/role/x/role-id"],
+    ["write", "auth/approle/role/x/secret-id"],
+    ["read", "sys/raw/x"],
+    ["policy", "write", "x", "-"],
+    ["token", "create"],
+])
+def test_guard_refuses_value_reads_and_writes(args):
+    assert not survey.allowed_bao_call(args)
+    bao = survey.Bao(FakeBao(), {"BAO_TOKEN": TOKEN})
+    with pytest.raises(survey.SurveyError, match="refused"):
+        bao.call(*args)
+
+
+def test_kv_path_must_not_be_an_api_data_path(catalog):
+    with pytest.raises(survey.SurveyError, match="logical KV path"):
+        survey.main(["--kv-path", "platform/data/workloads/x", "--policy", "p", "--consumer", "c"],
+                    runner=FakeBao(), environ={"BAO_TOKEN": TOKEN})
+
+
+def test_policy_glob_matching():
+    assert survey.path_matches("platform/data/workloads/*", "platform/data/workloads/a/b")
+    assert survey.path_matches("platform/data/+/a/b", "platform/data/workloads/a/b")
+    assert not survey.path_matches("platform/data/workloads/a", "platform/data/workloads/a/b")

workplans/MASON-WP-0008-plan-survey.md

@@ -0,0 +1,112 @@
+---
+id: MASON-WP-0008
+type: workplan
+title: "Generate the construction-plan existing-structure survey from live metadata"
+domain: infotech
+repo: ops-mason
+status: active
+flavor: implementation
+owner: claude
+topic_slug: netkingdom
+created: "2026-10-05"
+updated: "2026-10-05"
+---
+
+# MASON-WP-0008 — Generate the existing-structure survey
+
+## Goal
+
+Section 2 of a construction plan (`docs/construction-plan-format.md`) needs
+actual OpenBao output: relevant policy, auth and secrets-engine entries, the
+matching ops-warden routing catalog entries, and an explicit answer to "does an
+existing lane already satisfy this demand?". Today the drafting agent gathers
+this by hand. Make it reproducible with one command that runs under the
+`ops-mason-build` grant.
+
+Founder decision: ops-warden routes, ops-mason surveys. ops-warden gets a
+pointer-only catalog entry (WARDEN-WP-0042), no survey code and no token.
+
+## Boundaries
+
+- Metadata only. No `kv get`, no `*/data/*` read, no role-id or secret-id read;
+  enforced by an allowlist in the script and asserted by tests.
+- `policies/ops-mason-build.hcl` is not changed. A query the grant does not
+  permit is reported as `not permitted by ops-mason-build`, not widened.
+- No live OpenBao call in this workplan's implementation; tests use a fake bao.
+
+## Tasks
+
+## Implement scripts/plan-survey.py
+
+```task
+id: MASON-WP-0008-T01
+status: done
+priority: high
+```
+
+Script in the `custody-inventory.py` idiom (stdlib plus PyYAML, grant token
+from `~/.claude-bao-token`, `BAO_ADDR` default `http://127.0.0.1:18200`, loud
+failure). Takes KV path, proposed policy, AppRole and/or Kubernetes role,
+consumer and keywords. Runs `bao token lookup` (display_name and policies only;
+refuses any token other than `ops-mason-build` plus `default`),
+`bao secrets list`, `bao auth list`, `bao policy list`,
+`bao list auth/<mount>/role` for approle and kubernetes mounts, `bao read` of
+name-matched role configs, `bao policy read` for name-matched policies,
+`bao kv metadata get` on the target and `bao kv list` on its parent. Greps the
+ops-warden catalog (`--catalog`, default
+`~/ops-warden/registry/routing/catalog.yaml`). Emits a pasteable section 2 with
+counts, timestamp, `BAO_ADDR`, display_name and a draft yes/no/partial answer.
+
+## Unit tests with a fake bao runner
+
+```task
+id: MASON-WP-0008-T02
+status: done
+priority: high
+```
+
+`tests/test_plan_survey.py`: yes/no/partial verdicts, denied capabilities
+degrade and are listed as incomplete, no grant and non-grant tokens fail
+loudly, the token never appears in output, every recorded call is
+metadata-only, and the guard refuses `kv get`, `*/data/*`, role-id/secret-id
+and write calls.
+
+## Document the command
+
+```task
+id: MASON-WP-0008-T03
+status: done
+priority: medium
+```
+
+`docs/construction-plan-format.md` § 2 (grant, then survey) and the OpenBao
+Access section of `AGENTS.md`.
+
+## Route to it from ops-warden
+
+```task
+id: MASON-WP-0008-T04
+status: done
+priority: medium
+```
+
+Pointer-only catalog entry `openbao-structure-survey` in ops-warden
+(WARDEN-WP-0042): owner ops-mason, `warden_executes: false`, points to the
+grant, this script and the format doc.
+
+## First live survey: inbound-library IMAP lane
+
+```task
+id: MASON-WP-0008-T05
+status: wait
+priority: medium
+blocking_reason: "Needs a founder-minted ops-mason-build grant (./scripts/bao-session.sh grant MASON-WP-0008-T05); agents do not log in or grant."
+```
+
+Run against the real store for `platform/workloads/inbound-library/imap`,
+policy `workload-kv-read-inbound-library-imap`, AppRole `inbound-library`,
+consumer `inbound-library`, and check that `bao list auth/<mount>/role`,
+role reads and `bao policy read` behave as the policy file implies
+(`auth/approle/role/*`, `auth/kubernetes/role/*`, `sys/policies/acl/*`).
+Record any `not permitted by ops-mason-build` line as a founder decision,
+not a policy edit.