Something went wrong. Try again.
A video game where you play as a misaligned AI, deceiving and building power. An experiment in spec-driven development. forked from cameron.stream/misaligned
Something went wrong. Try again.
15 kB · 382 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383#!/usr/bin/env python3"""Structured work-order metadata, validation, and ROADMAP projection."""
from __future__ import annotations
import argparseimport jsonimport reimport sysfrom dataclasses import dataclassfrom pathlib import Path
STATUS_ENUM = {"DRAFT", "READY", "IN PROGRESS", "BLOCKED", "IMPLEMENTED"}CLASS_ENUM = {"docs", "frontend", "sim", "save", "process"}TASK_RE = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")REF_RE = re.compile(r"^(wiki/[A-Za-z0-9._/-]+[.]md)#([a-z0-9-]+)$")HEADING_RE = re.compile(r"^#{1,6}\s+(.+)$", re.M)BEGIN = "<!-- BEGIN GENERATED WORK ORDERS -->"END = "<!-- END GENERATED WORK ORDERS -->"
def slugify_heading(heading: str) -> str: value = heading.lower() value = re.sub(r"[`*_]", "", value).replace("&", " ") value = re.sub(r"[^a-z0-9 -]", "", value) value = re.sub(r"\s+", "-", value.strip()) return re.sub(r"-+", "-", value).strip("-")
def metadata_block(text: str) -> str: match = re.search(r"^```[^\n]*\n(.*?)^```\s*$", text, re.M | re.S) return match.group(1) if match else ""
def parse_metadata(text: str) -> dict[str, str | list[str]]: block = metadata_block(text) meta: dict[str, str | list[str]] = {} current: str | None = None for line in block.splitlines(): if line.startswith(" - ") and current: value = meta.get(current) if value == "" or value is None: value = [] meta[current] = value if isinstance(value, list): value.append(line[4:].strip()) continue if line and not line.startswith(" ") and ":" in line: key, value = line.split(":", 1) current = key.strip() meta[current] = value.strip() continue if line.startswith(" ") and current == "Status note": prior = str(meta.get(current, "")) meta[current] = f"{prior} {line.strip()}".strip() return meta
@dataclassclass WorkOrder: path: Path rel: str title: str status: str stage: str task: str | None priority: int | None work_class: str | None blockers: list[str] keys: list[str] text: str
def as_dict(self) -> dict: return { "path": self.rel, "title": self.title, "status": self.status, "stage": self.stage, "task": self.task, "priority": self.priority, "class": self.work_class, "blocked_by": self.blockers, "exclusive_keys": self.keys, }
def load_specs(root: Path) -> list[WorkOrder]: specs: list[WorkOrder] = [] for path in sorted((root / "wiki").rglob("*.md")): text = path.read_text(encoding="utf-8") meta = parse_metadata(text) if meta.get("Type") != "spec": continue title_match = re.search(r"^#\s+(.+)$", text, re.M) title = title_match.group(1).strip() if title_match else path.stem title = re.sub(r"^Spec:\s*", "", title, flags=re.I) priority: int | None = None raw_priority = meta.get("Work priority") if isinstance(raw_priority, str) and raw_priority.isdigit(): priority = int(raw_priority) blockers_value = meta.get("Blocked by", []) blockers = blockers_value if isinstance(blockers_value, list) else [] keys_value = meta.get("Exclusive keys", []) keys = keys_value if isinstance(keys_value, list) else [] specs.append( WorkOrder( path=path, rel=path.relative_to(root).as_posix(), title=title, status=str(meta.get("Status", "")), stage=str(meta.get("Stage", "")), task=str(meta["Work order"]) if "Work order" in meta else None, priority=priority, work_class=str(meta["Work class"]) if "Work class" in meta else None, blockers=blockers, keys=keys, text=text, ) ) return specs
def _validate_ref(root: Path, source: WorkOrder, ref: str, errors: list[str]) -> None: match = REF_RE.fullmatch(ref) if not match: errors.append(f"{source.rel}: malformed Blocked by reference: {ref}") return target = root / match.group(1) if not target.is_file(): errors.append(f"{source.rel}: Blocked by target does not exist: {ref}") return text = target.read_text(encoding="utf-8") if parse_metadata(text).get("Type") != "spec": errors.append(f"{source.rel}: Blocked by target is not a spec: {ref}") anchors = {slugify_heading(h) for h in HEADING_RE.findall(text)} if match.group(2) not in anchors: errors.append(f"{source.rel}: Blocked by anchor does not exist: {ref}")
def validate(root: Path, specs: list[WorkOrder]) -> list[str]: errors: list[str] = [] tasks: dict[str, str] = {} for spec in specs: block = metadata_block(spec.text) if "Dispatch state:" in block: errors.append(f"{spec.rel}: Dispatch state duplicates Status; remove it") if spec.status not in STATUS_ENUM: continue required = spec.status != "IMPLEMENTED" fields = { "Work order": spec.task, "Work priority": spec.priority, "Work class": spec.work_class, "Exclusive keys": spec.keys, } if required: for name, value in fields.items(): if value is None or value == []: errors.append(f"{spec.rel}: unfinished spec missing {name}") if not re.search(r"^Blocked by:\s*(?:none)?\s*$", block, re.M): errors.append(f"{spec.rel}: unfinished spec missing Blocked by") if spec.task: if not TASK_RE.fullmatch(spec.task): errors.append(f"{spec.rel}: invalid Work order slug: {spec.task}") elif spec.task in tasks: errors.append( f"{spec.rel}: Work order '{spec.task}' already owned by {tasks[spec.task]}" ) else: tasks[spec.task] = spec.rel if spec.priority is not None and spec.priority <= 0: errors.append(f"{spec.rel}: Work priority must be a positive integer") if spec.work_class and spec.work_class not in CLASS_ENUM: errors.append(f"{spec.rel}: invalid Work class: {spec.work_class}") for ref in spec.blockers: _validate_ref(root, spec, ref, errors) for key in spec.keys: if key.startswith("@"): if not TASK_RE.fullmatch(key[1:]): errors.append(f"{spec.rel}: invalid logical Exclusive key: {key}") elif key.startswith("/") or ".." in Path(key).parts: errors.append(f"{spec.rel}: unsafe Exclusive key: {key}") elif not (root / key).exists(): errors.append(f"{spec.rel}: Exclusive key path does not exist: {key}")
errors.extend(validate_current_prose(root, specs)) return errors
def validate_current_prose(root: Path, specs: list[WorkOrder]) -> list[str]: errors: list[str] = [] by_path = {spec.path.resolve(): spec for spec in specs} roadmap = root / "wiki/process/ROADMAP.md" if roadmap.is_file(): text = roadmap.read_text(encoding="utf-8") pattern = re.compile( r"\*\*Spec:\*\*\s+\[[^]]+\]\(([^)]+)\)\s+" r"\((DRAFT|READY|IN PROGRESS|BLOCKED|IMPLEMENTED)(?:\s+[^)]*)?\)" ) for match in pattern.finditer(text): target = (roadmap.parent / match.group(1)).resolve() spec = by_path.get(target) if spec and match.group(2) != spec.status: line = text.count("\n", 0, match.start()) + 1 errors.append( f"wiki/process/ROADMAP.md:{line}: inline status {match.group(2)} " f"disagrees with {spec.rel} ({spec.status})" )
for spec in specs: sections = re.split(r"(?=^##\s+)", spec.text, flags=re.M) for section in sections: heading = section.splitlines()[0] if section.splitlines() else "" if "[OPEN]" in heading.upper(): for offset, line_text in enumerate(section.splitlines()[1:], 1): if re.match(r"^\s*-\s+\(?resolved\b", line_text, re.I): line = spec.text[: spec.text.find(section)].count("\n") + offset + 2 errors.append( f"{spec.rel}:{line}: resolved item remains under an [OPEN] heading" ) blocked = set(re.findall(r"BLOCKED on issue #(\d+)", spec.text, re.I)) for issue in blocked: closed = re.search( rf"issue #{re.escape(issue)}[^\n]{{0,100}}(?:closed|resolved)", spec.text, re.I, ) if closed: line = spec.text.count("\n", 0, closed.start()) + 1 errors.append( f"{spec.rel}:{line}: issue #{issue} is locally closed but still marks a criterion BLOCKED" ) return errors
def blocker_state(spec: WorkOrder, by_rel: dict[str, WorkOrder]) -> list[str]: active: list[str] = [] for ref in spec.blockers: rel = ref.split("#", 1)[0] target = by_rel.get(rel) if target is None or target.status != "IMPLEMENTED": active.append(target.task if target and target.task else rel) return active
def lane(spec: WorkOrder, by_rel: dict[str, WorkOrder]) -> str: stage = spec.stage.lower() if stage.startswith(("b2", "b3", "release")) or "deferred" in stage: return "staged" if spec.status in {"DRAFT", "BLOCKED"} or blocker_state(spec, by_rel): return "held" return "current"
def _table(rows: list[WorkOrder], by_rel: dict[str, WorkOrder]) -> list[str]: lines = [ "| Priority | Work order | Spec | Status | Class | Blocking |", "|---:|---|---|---|---|---|", ] if not rows: lines.append("| - | - | - | - | - | - |") return lines for spec in sorted(rows, key=lambda item: (item.priority or 9999, item.rel)): link = Path(spec.rel).relative_to("wiki/process") if spec.rel.startswith("wiki/process/") else Path("..") / Path(spec.rel).relative_to("wiki") blockers = ", ".join(blocker_state(spec, by_rel)) or "-" lines.append( f"| {spec.priority} | `{spec.task}` | [{spec.title}]({link.as_posix()}) | " f"{spec.status} | {spec.work_class} | {blockers} |" ) return lines
def generated_region(specs: list[WorkOrder]) -> str: live = [s for s in specs if s.status != "IMPLEMENTED" and s.task] by_rel = {s.rel: s for s in specs} groups = {name: [s for s in live if lane(s, by_rel) == name] for name in ("current", "held", "staged")} lines = [ BEGIN, "## Live work-order index", "", "Generated by `tools/work_orders.py` from each owning spec. Lower priority", "numbers run first within the current stage; detailed prose below is context,", "not a second status owner.", "", "### Current lanes", "", *_table(groups["current"], by_rel), "", "### Held or blocked", "", *_table(groups["held"], by_rel), "", "### Later stages", "", *_table(groups["staged"], by_rel), END, ] return "\n".join(lines)
def projected_roadmap(root: Path, specs: list[WorkOrder]) -> str: path = root / "wiki/process/ROADMAP.md" text = path.read_text(encoding="utf-8") region = generated_region(specs) if BEGIN in text and END in text: return re.sub( rf"{re.escape(BEGIN)}.*?{re.escape(END)}", lambda _: region, text, flags=re.S, ) marker = "## How to dispatch" if marker not in text: raise ValueError("ROADMAP missing '## How to dispatch' insertion point") return text.replace(marker, f"{region}\n\n{marker}", 1)
def task_values(specs: list[WorkOrder], requested: str) -> int: requested_path = Path(requested).as_posix().lstrip("./") for spec in specs: if spec.rel == requested_path or spec.task == requested: if not spec.task or not spec.work_class or not spec.keys: print(f"FAIL: {spec.rel} has incomplete work metadata", file=sys.stderr) return 1 print(f"task={spec.task}") print(f"class={spec.work_class}") for key in spec.keys: print(f"key={key}") return 0 print(f"FAIL: no work order matches {requested}", file=sys.stderr) return 1
def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("command", choices=("write", "check", "validate", "list", "task-values")) parser.add_argument("value", nargs="?") parser.add_argument("--root", type=Path, default=Path(__file__).resolve().parent.parent) parser.add_argument("--json", action="store_true") args = parser.parse_args(argv) root = args.root.resolve() specs = load_specs(root) if args.command == "task-values": if not args.value: parser.error("task-values requires a spec path or task slug") return task_values(specs, args.value) errors = validate(root, specs) if errors: for error in errors: print(f"FAIL: {error}") return 1 if args.command == "validate": print("work orders: OK (metadata + current-prose consistency)") return 0 if args.command == "list": payload = [s.as_dict() for s in specs if s.status != "IMPLEMENTED" and s.task] if args.json: print(json.dumps(payload, indent=2, sort_keys=True)) else: for item in sorted(payload, key=lambda row: (row["priority"] or 9999, row["path"])): print(f"{item['priority']:>3} {item['status']:<11} {item['task']} ({item['class']})") return 0 roadmap = root / "wiki/process/ROADMAP.md" projected = projected_roadmap(root, specs) current = roadmap.read_text(encoding="utf-8") if args.command == "check": if projected != current: print("FAIL: generated ROADMAP work-order index is stale") print("Run: tools/work_orders.py write") return 1 print("work orders: OK (metadata + ROADMAP projection fresh)") return 0 if projected != current: roadmap.write_text(projected, encoding="utf-8") print("work orders: wrote wiki/process/ROADMAP.md") else: print("work orders: ROADMAP already up to date") return 0
if __name__ == "__main__": raise SystemExit(main())