#!/usr/bin/env python3 from __future__ import annotations import argparse import json import re import sys from pathlib import Path from typing import Any _ROOT_DIR = Path(__file__).resolve().parents[1] if str(_ROOT_DIR) not in sys.path: sys.path.insert(0, str(_ROOT_DIR)) from scripts.packlib import ( # noqa: E402 ROOT, PackError, brussels_now, load_plan, load_state, packet_map, select_next_packet, ) TRANSITIONS = { "pending": {"in_progress", "blocked"}, "in_progress": {"completed", "blocked", "failed"}, "failed": {"in_progress", "blocked"}, "blocked": {"pending", "in_progress"}, "completed": set(), } def update_current_state_markdown(state: dict[str, Any], plan: dict[str, Any]) -> None: path = ROOT / "CURRENT_STATE.md" text = path.read_text(encoding="utf-8") if path.exists() else "# Current State\n" statuses = state["packets"] completed = [pid for pid, status in statuses.items() if status == "completed"] in_progress = [pid for pid, status in statuses.items() if status == "in_progress"] try: selection = select_next_packet(plan, state) next_packet = selection.packet except PackError: next_packet = None if all(status == "completed" for status in statuses.values()): stage = ( "All planned v1.0 work packets completed; release evidence must remain authoritative." ) next_line = "None — all packets completed." elif in_progress: stage = f"Implementation in progress: {in_progress[0]}." packet = packet_map(plan)[in_progress[0]] next_line = f"`{in_progress[0]}` — {packet.get('title')}." elif next_packet: stage = "Implementation active; no packet currently marked in progress." next_line = f"`{next_packet['id']}` — {next_packet.get('title')}." else: stage = "Implementation blocked; inspect BUILD_STATE.json." next_line = "No ready packet." date = brussels_now().split("T", 1)[0] replacements = { r"\*\*State date:\*\*.*": f"**State date:** {date}", r"\*\*Current stage:\*\*.*": f"**Current stage:** {stage}", r"\*\*Next packet:\*\*.*": f"**Next packet:** {next_line}", } for pattern, replacement in replacements.items(): if re.search(pattern, text): text = re.sub(pattern, replacement, text, count=1) else: text = text.rstrip() + "\n\n" + replacement + "\n" block = ( "\n" "## Machine state summary\n\n" f"- Completed packets: {len(completed)}/{len(statuses)}\n" f"- In progress: {', '.join(in_progress) if in_progress else 'none'}\n" f"- Blocking issues: {len(state.get('blocking_issues', []))}\n" f"- External inputs: {len(state.get('external_inputs', []))}\n" f"- Last state update: `{state['last_updated']}`\n" "" ) if "" in text: text = re.sub( r".*?", block, text, flags=re.DOTALL, ) else: text = text.rstrip() + "\n\n" + block + "\n" path.write_text(text.rstrip() + "\n", encoding="utf-8") def main() -> int: parser = argparse.ArgumentParser(description="Update packet state deterministically.") parser.add_argument("packet", help="work packet id, for example WP-020") parser.add_argument( "status", choices=["pending", "in_progress", "completed", "blocked", "failed"], ) parser.add_argument("--reason", help="required for blocked/failed; concise and factual") parser.add_argument( "--evidence", action="append", default=[], help="repository-relative evidence path; repeatable", ) parser.add_argument( "--external-input", action="append", default=[], help="external input description; repeatable", ) parser.add_argument("--commit", help="last verified commit SHA") parser.add_argument( "--force", action="store_true", help="allow an otherwise invalid transition; record a decision manually", ) args = parser.parse_args() try: plan = load_plan() state = load_state() packets = packet_map(plan) except PackError as exc: print(f"ERROR: {exc}", file=sys.stderr) return 2 if args.packet not in packets: print(f"ERROR: unknown packet {args.packet}", file=sys.stderr) return 2 old = state["packets"].get(args.packet) if old == args.status: print(f"No change: {args.packet} already {args.status}") return 0 if not args.force and args.status not in TRANSITIONS.get(old, set()): print( f"ERROR: invalid transition {args.packet}: {old} -> {args.status}", file=sys.stderr, ) return 3 if args.status in {"blocked", "failed"} and not args.reason: print("ERROR: --reason is required for blocked or failed", file=sys.stderr) return 3 if args.status == "in_progress": other = [ pid for pid, status in state["packets"].items() if status == "in_progress" and pid != args.packet ] if other: print( f"ERROR: another packet is already in progress: {other}", file=sys.stderr, ) return 3 unmet = [ dep for dep in packets[args.packet].get("deps", []) if state["packets"].get(dep) != "completed" ] if unmet and not args.force: print( f"ERROR: unmet dependencies for {args.packet}: {unmet}", file=sys.stderr, ) return 3 now = brussels_now() state["packets"][args.packet] = args.status state["last_updated"] = now if args.commit: state["last_verified_commit"] = args.commit if args.status == "in_progress": state["current_packet"] = args.packet state["status"] = "building" elif args.status == "completed": state["status"] = "building" elif args.status in {"blocked", "failed"}: state["status"] = args.status issue = { "packet": args.packet, "status": args.status, "reason": args.reason, "recordedAt": now, } if issue not in state.setdefault("blocking_issues", []): state["blocking_issues"].append(issue) elif args.status == "pending": state["status"] = "building" for evidence in args.evidence: path = ROOT / evidence if not path.exists(): print( f"ERROR: evidence path does not exist: {evidence}", file=sys.stderr, ) return 3 entry = { "packet": args.packet, "path": evidence, "recordedAt": now, } if entry not in state.setdefault("evidence_index", []): state["evidence_index"].append(entry) for description in args.external_input: entry = { "packet": args.packet, "description": description, "recordedAt": now, "status": "open", } if entry not in state.setdefault("external_inputs", []): state["external_inputs"].append(entry) if args.status in {"in_progress", "completed", "pending"}: state["blocking_issues"] = [ item for item in state.get("blocking_issues", []) if not (isinstance(item, dict) and item.get("packet") == args.packet) ] selection = select_next_packet(plan, state) if selection.packet: state["current_packet"] = selection.packet["id"] elif all(status == "completed" for status in state["packets"].values()): state["current_packet"] = None state["status"] = "completed" (ROOT / "BUILD_STATE.json").write_text(json.dumps(state, indent=2) + "\n", encoding="utf-8") update_current_state_markdown(state, plan) print(f"Updated {args.packet}: {old} -> {args.status}; next={state.get('current_packet')}") return 0 if __name__ == "__main__": sys.exit(main())