Prepare offline attended provisioning checklists and route remaining custody gates
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a10391-6355-7b41-b0fa-3e14e874f091
This commit is contained in:
parent
7c1a5c2516
commit
46eb8bea9e
6 changed files with 221 additions and 4 deletions
|
|
@ -2240,6 +2240,12 @@ def build_parser() -> argparse.ArgumentParser:
|
|||
)
|
||||
applier_apply.set_defaults(func=command_applier_apply)
|
||||
|
||||
from provisioning_window import command_window
|
||||
window = sub.add_parser("window-checklist", help="Render an offline attended-window preparation checklist; never execute")
|
||||
window.add_argument("--window-file", required=True)
|
||||
window.add_argument("--workplan-root", action="append", required=True)
|
||||
window.set_defaults(func=command_window)
|
||||
|
||||
runbook = sub.add_parser(
|
||||
"runbook",
|
||||
help="Render or execute the attended operator apply/verify runbook",
|
||||
|
|
|
|||
80
scripts/provisioning_window.py
Normal file
80
scripts/provisioning_window.py
Normal file
|
|
@ -0,0 +1,80 @@
|
|||
"""Offline, metadata-only attended-window checklist; never executes a CCR."""
|
||||
from pathlib import Path
|
||||
import re
|
||||
from uuid import UUID
|
||||
import yaml
|
||||
|
||||
|
||||
def task_sources(roots, decision_id):
|
||||
selected = {}
|
||||
for root in roots:
|
||||
for path in sorted(Path(root).glob("*.md")):
|
||||
text = path.read_text()
|
||||
front = re.match(r"---\n(.*?)\n---", text, re.S)
|
||||
meta = yaml.safe_load(front.group(1)) if front else {}
|
||||
if meta and (meta.get("flavor") == "residual" or meta.get("status") in {"finished", "archived"}):
|
||||
continue
|
||||
for block in re.findall(r"```task\s*\n(.*?)\n```", text, re.S):
|
||||
task = yaml.safe_load(block)
|
||||
if str(task.get("decision_id", "")) == decision_id and task.get("status") not in {"done", "cancel"}:
|
||||
task_id = task["id"]
|
||||
if task_id in selected:
|
||||
raise ValueError("duplicate task id: " + task_id)
|
||||
selected[task_id] = task
|
||||
return selected
|
||||
|
||||
|
||||
def local_file(repo, value):
|
||||
path = (repo / value).resolve()
|
||||
if not path.is_relative_to(repo.resolve()) or not path.is_file():
|
||||
raise ValueError("missing or outside-repository preparation file")
|
||||
return path
|
||||
|
||||
|
||||
def render_window(window_file, roots, repo):
|
||||
window = yaml.safe_load(Path(window_file).read_text())
|
||||
decision = str(UUID(str(window["decision_id"])))
|
||||
date = str(window["date"])
|
||||
# A dated manifest is preparation, never a scheduled event or approval.
|
||||
if not re.fullmatch(r"\d{4}-\d{2}-\d{2}", date):
|
||||
raise ValueError("window date must be YYYY-MM-DD")
|
||||
tasks = task_sources(roots, decision)
|
||||
entries = window.get("entries", [])
|
||||
by_task = {}
|
||||
for entry in entries:
|
||||
tid = entry["task_id"]
|
||||
if tid in by_task:
|
||||
raise ValueError("duplicate window task")
|
||||
by_task[tid] = entry
|
||||
stale = set(by_task) - set(tasks)
|
||||
if stale:
|
||||
raise ValueError("manifest tasks absent from current decision: " + ", ".join(sorted(stale)))
|
||||
lines = ["# Attended provisioning preparation: " + date,
|
||||
"Decision: " + decision,
|
||||
"Preparation only. Founder window approval and each exact CCR revision remain required."]
|
||||
ready = bool(tasks)
|
||||
for index, tid in enumerate(sorted(tasks), 1):
|
||||
entry = by_task.get(tid)
|
||||
if not entry:
|
||||
lines.append(f"{index}. {tid}: NOT READY — missing CCR/lane/receipt join.")
|
||||
ready = False
|
||||
continue
|
||||
ccr = yaml.safe_load(local_file(repo, entry["ccr"]).read_text())
|
||||
local_file(repo, entry["lane_doc"])
|
||||
local_file(repo, entry["receipt_contract"])
|
||||
eligible = ccr.get("status") == "approved"
|
||||
ready &= eligible
|
||||
label = "REVIEW PREPARED" if eligible else "NOT READY (CCR is not approved; never replay applied custody)"
|
||||
# Deliberately avoid rendering arbitrary CCR contents or secret-bearing values.
|
||||
lines.extend([f"{index}. {tid}: {label}",
|
||||
" CCR: " + entry["ccr"], " Lane: " + entry["lane_doc"],
|
||||
" Owner receipt contract: " + entry["receipt_contract"],
|
||||
" Revalidate exact digest, upstream gates and native readiness before attended execution."])
|
||||
lines.append("Checklist complete: " + str(ready).lower())
|
||||
return "\n".join(lines) + "\n", ready
|
||||
|
||||
|
||||
def command_window(args):
|
||||
output, ready = render_window(args.window_file, args.workplan_root, Path(__file__).resolve().parents[1])
|
||||
print(output, end="")
|
||||
return 0 if ready else 2
|
||||
Loading…
Add table
Add a link
Reference in a new issue