46eb8bea9e
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a10391-6355-7b41-b0fa-3e14e874f091
80 lines
3.5 KiB
Python
80 lines
3.5 KiB
Python
"""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
|