242 lines
8.2 KiB
Python
242 lines
8.2 KiB
Python
#!/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 = (
|
|
"<!-- MACHINE_STATE_BEGIN -->\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"
|
|
"<!-- MACHINE_STATE_END -->"
|
|
)
|
|
if "<!-- MACHINE_STATE_BEGIN -->" in text:
|
|
text = re.sub(
|
|
r"<!-- MACHINE_STATE_BEGIN -->.*?<!-- MACHINE_STATE_END -->",
|
|
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())
|