tegwick
6289ecbb68
audit-core rollout made origin /readyz 200. Remaining gates are the T03-only Flex Auth package, no approval:create requester for these eight acts, and a live KeyCape subject. Attach writes bound copies from a created receipt; it does not bind. Assistant: grok Assistant-Session: 01a09dc1-b21e-77e1-919e-fcad2f82b267
118 lines
4.9 KiB
Python
118 lines
4.9 KiB
Python
"""Attach native approval ids to compact-sitting drafts (INFD-WP-0002-T03).
|
|
|
|
Requires an exact live KeyCape subject and eight unapproved human-control
|
|
receipts. Writes bound memo copies; does not present, bind, create approvals,
|
|
or load the live store.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
from dataclasses import replace
|
|
from pathlib import Path
|
|
|
|
from informed_decision.memo import Principal
|
|
from informed_decision.records import dumps, memo_from
|
|
|
|
ROOT = Path(__file__).resolve().parents[1] / "docs" / "batches" / "2026-09-14"
|
|
DIGEST = re.compile(r"^sha256:[0-9a-f]{64}$")
|
|
APPROVAL_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$")
|
|
|
|
|
|
def _principal(value: str) -> Principal:
|
|
if not value or any(c.isspace() for c in value) or len(value) > 256:
|
|
raise ValueError("exact human subject required")
|
|
if value == "pending-human-session":
|
|
raise ValueError("live key-cape subject required")
|
|
return Principal(id=value, kind="person", display_name=value, role="reviewer")
|
|
|
|
|
|
def _approval(row: dict, memo_id: str) -> dict:
|
|
if row.get("memo_id") != memo_id:
|
|
raise ValueError(f"receipt memo_id mismatch for {memo_id}")
|
|
approval = row.get("approval") or {}
|
|
binding = approval.get("binding") or {}
|
|
digest = binding.get("digest")
|
|
if approval.get("status") != "requested" or approval.get("entries"):
|
|
raise ValueError(f"unapproved human-control request required for {memo_id}")
|
|
if binding.get("human_control") is not True:
|
|
raise ValueError(f"human_control required for {memo_id}")
|
|
if not isinstance(approval.get("id"), str) or not APPROVAL_ID.fullmatch(approval["id"]):
|
|
raise ValueError(f"native approval id required for {memo_id}")
|
|
if not isinstance(digest, str) or not DIGEST.fullmatch(digest):
|
|
raise ValueError(f"native binding digest required for {memo_id}")
|
|
return approval
|
|
|
|
|
|
def attach(principal: str, receipt: dict, root: Path = ROOT) -> dict:
|
|
actor = _principal(principal)
|
|
if receipt.get("status") != "created":
|
|
raise ValueError("created native receipt required")
|
|
sitting = json.loads((root / "sitting.json").read_text())
|
|
expected = []
|
|
for name in ("credentials", "decisions"):
|
|
index = json.loads((root / name / "index.json").read_text())
|
|
for row in index["ordinal"]:
|
|
expected.append((name, row["memo_id"], row["memo"], row["packet"]))
|
|
requests = receipt.get("requests") or []
|
|
if len(requests) != len(expected):
|
|
raise ValueError("receipt must cover every sitting memo once")
|
|
by_id = {row.get("memo_id"): row for row in requests}
|
|
if set(by_id) != {memo_id for _, memo_id, _, _ in expected}:
|
|
raise ValueError("receipt memo set must match the sitting")
|
|
prepared = []
|
|
for name, memo_id, memo_name, packet_name in expected:
|
|
approval = _approval(by_id[memo_id], memo_id)
|
|
memo = memo_from(json.loads((root / name / memo_name).read_text()))
|
|
if memo.approval_id is not None or memo.binding.principal.id != "pending-human-session":
|
|
raise ValueError(f"unsigned draft required for {memo_id}")
|
|
bound = replace(
|
|
memo,
|
|
binding=replace(memo.binding, principal=actor),
|
|
approval_id=approval["id"],
|
|
approval_binding_digest=approval["binding"]["digest"],
|
|
)
|
|
prepared.append((name, memo_id, memo_name, packet_name, bound))
|
|
bound_root = root / "bound"
|
|
bound_root.mkdir(parents=True, exist_ok=True)
|
|
written = []
|
|
for name, memo_id, memo_name, packet_name, bound in prepared:
|
|
path = bound_root / memo_name
|
|
path.write_text(dumps(bound) + "\n", encoding="utf-8")
|
|
packet_src = root / name / packet_name
|
|
packet_dst = bound_root / packet_name
|
|
packet_dst.write_bytes(packet_src.read_bytes())
|
|
written.append(
|
|
{
|
|
"memo_id": memo_id,
|
|
"approval_id": bound.approval_id,
|
|
"memo": path.name,
|
|
"packet": packet_dst.name,
|
|
}
|
|
)
|
|
index = {
|
|
"kind": "informed-decision-bound-sitting",
|
|
"sitting_id": sitting["id"],
|
|
"principal": actor.id,
|
|
"submitted": False,
|
|
"agent_disposition_forbidden": True,
|
|
"memos": written,
|
|
}
|
|
(bound_root / "index.json").write_text(json.dumps(index, indent=2, ensure_ascii=False) + "\n")
|
|
return index
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--principal", required=True, help="exact live KeyCape subject")
|
|
parser.add_argument("--receipt", type=Path, required=True, help="native created-approval receipt")
|
|
args = parser.parse_args()
|
|
receipt = json.loads(args.receipt.read_text())
|
|
index = attach(args.principal, receipt)
|
|
print(json.dumps({"status": "bound_copies_written", "count": len(index["memos"])}, indent=2))
|
|
print("No presentations, dispositions, or approval entries created.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|