Catalog fluid-telegram-operator (draft) and tenant-engine sender projection; case plan env spec and case-ref

- fluid-telegram-operator: pointer-only draft entry and playbook from ops-mason's
  MASON-WP-0005-T06 proposal; activation waits for the owner-run member and
  non-member checks and the salt create-once probe.
- audit-core-senders: tenant-engine recorded as a verified exact-path consumer
  projection (MASON-WP-0007-T06). The registry lane itself stays draft.
- warden case plan: {param:env} format spec (net-kingdom 850907a), other specs
  and conversions refused; prints case-ref <case_id>@sha256:<file digest>.
- Workload-join unknown count 21 -> 22 (test and pep-stance.yaml).

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

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 427672@bnt-lap001
Assistant-Session: a0a2367e-4697-440c-a4e5-9713c009f6d8

src/warden/case_plan.py

@@ -13,6 +13,7 @@ parameter whose type declares ``sensitivity: secret`` is refused outright.
 """
 from __future__ import annotations
 
+import hashlib
 import re
 import string
 from dataclasses import dataclass, field
@@ -56,10 +57,12 @@ class CasePlan:
     skipped: List[str] = field(default_factory=list)
     attended_sessions: List[str] = field(default_factory=list)
     verification: List[dict] = field(default_factory=list)
+    case_ref: Optional[str] = None
 
     def as_dict(self) -> dict:
         return {
             "case": self.case_id,
+            "case_ref": self.case_ref,
             "type": self.type_ref,
             "steps": [s.as_dict() for s in self.steps],
             "skipped": self.skipped,
@@ -76,6 +79,15 @@ def load_case(path: Path) -> dict:
     return doc
 
 
+def case_ref(case_id: str, case_bytes: bytes) -> str:
+    """``<case_id>@sha256:<hex>`` over the case file's exact bytes.
+
+    The reference an owner entry point records for a run (RPF-WP-0054-T01). The
+    owner re-derives it from the file rather than trusting the caller.
+    """
+    return f"{case_id}@sha256:{hashlib.sha256(case_bytes).hexdigest()}"
+
+
 def default_type_path(case_path: Path, type_ref: str) -> Path:
     """net-kingdom layout: security-cases/<case>.yaml beside security-case-types/<id>/type.yaml."""
     type_id = type_ref.split("@", 1)[0]
@@ -94,14 +106,31 @@ def _applies(when: Optional[str], params: Dict[str, str]) -> bool:
     return str(params[name]).lower() == expected.lower()
 
 
+#: Format specs a template may apply, matching net-kingdom's validator
+#: (``FORMAT_SPECS``): ``{client_id:env}`` is upper case with ``-`` as ``_``.
+FORMAT_SPECS = ("", "env")
+
+
+class _Param(str):
+    def __format__(self, spec: str) -> str:
+        if spec == "env":
+            return self.upper().replace("-", "_")
+        return str.__format__(self, spec)
+
+
 def _render(text: Optional[str], params: Dict[str, str], where: str) -> Optional[str]:
     if text is None:
         return None
-    names = {f for _, f, _, _ in string.Formatter().parse(text) if f}
-    missing = sorted(names - set(params))
+    fields = [(f, spec or "", conv) for _, f, spec, conv in string.Formatter().parse(text) if f]
+    bad = sorted({f"{f}:{spec}" for f, spec, _ in fields if spec not in FORMAT_SPECS})
+    if bad:
+        raise CaseError(f"{where}: unknown format spec in {bad}; only {{param:env}} is supported")
+    if any(conv for _, _, conv in fields):
+        raise CaseError(f"{where}: conversions such as !r are not supported in templates")
+    missing = sorted({f for f, _, _ in fields} - set(params))
     if missing:
         raise CaseError(f"{where}: parameters {missing} are not set by the case")
-    return text.format(**params)
+    return text.format_map({k: _Param(v) for k, v in params.items()})
 
 
 def plan_case(case_doc: dict, type_doc: dict, catalog: Optional[Catalog]) -> CasePlan:
@@ -178,7 +207,10 @@ def _line(f: Finding) -> str:
 
 def format_plan(plan: CasePlan) -> str:
     labels = {"warden": "warden (caller proxy)", "attended": "ATTENDED", "owner": "owner"}
-    lines = [f"case {plan.case_id} ({plan.type_ref}): plan only, nothing executed", ""]
+    lines = [f"case {plan.case_id} ({plan.type_ref}): plan only, nothing executed"]
+    if plan.case_ref:
+        lines.append(f"case-ref: {plan.case_ref}")
+    lines.append("")
     for n, s in enumerate(plan.steps, 1):
         lane = f" lane={s.lane}" if s.lane else ""
         lines.append(f"{n}. {s.id} [{labels[s.mode]}] owner={s.owner}{lane} values={s.value_handling}")

src/warden/cli.py

@@ -2255,13 +2255,14 @@ def case_plan(
     output_json: Annotated[bool, typer.Option("--json", help="Output JSON")] = False,
 ) -> None:
     """Print a case's run (warden, attended, owner steps and verification); executes nothing (exit 1 if refused)."""
-    from warden.case_plan import CaseError, default_type_path, format_plan, load_case, plan_case
+    from warden.case_plan import CaseError, case_ref, default_type_path, format_plan, load_case, plan_case
     from warden.case_type import load_type
 
     try:
         case_doc = load_case(case_file)
         type_path = type_file or default_type_path(case_file, str((case_doc.get("spec") or {}).get("type", "")))
         plan = plan_case(case_doc, load_type(type_path), _load_catalog())
+        plan.case_ref = case_ref(plan.case_id, case_file.read_bytes())
     except CaseError as exc:
         err.print(str(exc), markup=False)
         raise typer.Exit(1)
@@ -2271,4 +2272,4 @@ def case_plan(
     if output_json:
         print(json.dumps(plan.as_dict(), indent=2))
     else:
-        console.print(format_plan(plan), markup=False)
+        console.print(format_plan(plan), markup=False, soft_wrap=True)