ra8-firmware 0.1.0
Bare-metal firmware for the Renesas RA8 family (RA8D2 / RA8P1)
Loading...
Searching...
No Matches
work_plan.py
Go to the documentation of this file.
1# SPDX-License-Identifier: MIT
2# Copyright (c) 2026 Brighton Sikarskie
3"""Notes-to-plan parser: a strict Markdown schema, validated and topologically ordered.
4
5The input is a plain Markdown file a person wrote while thinking, and it is
6treated as UNTRUSTED DATA throughout. Nothing here evaluates anything, expands
7anything, or interprets HTML; a heading is matched by a regular expression and
8its text is carried as an opaque string all the way to the emitted commands,
9where :func:`shlex.quote` puts it inside one shell word.
10
11The schema is deliberately strict and every violation is collected rather than
12raised at the first one, because a person fixing notes wants the whole list.
13An unrecognised metadata bullet is a hard error naming its line: a silently
14ignored bullet is how a plan comes to claim a priority nobody ever set.
15
16Ordering is Kahn topological sort with a lexicographic tie-break, so the same
17notes always produce byte-identical output. That determinism is the property
18that lets the JSON be diffed, and a cycle is reported as one concrete path
19rather than as a set of unsatisfied keys.
20
21Nothing here runs a command, touches the network, or writes outside a path the
22caller named.
23"""
24
25from __future__ import annotations
26
27import heapq
28import json
29import re
30from dataclasses import dataclass, field
31from pathlib import Path
32
33from work_git import WorkError
34from work_tracker import tracker_schema
35from work_workspace import KEY_RE
36
37#: Bumped whenever the emitted JSON shape changes incompatibly.
38PLAN_SCHEMA_VERSION = 1
39
40#: Metadata bullets recognised under an epic or issue heading.
41ITEM_BULLETS = ("labels", "priority", "track", "status", "depends-on", "estimate")
42
43#: Metadata bullets recognised under ``## Config``.
44CONFIG_BULLETS = ("statuses", "tracks")
45
46HEADING_RE = re.compile(r"^(?P<hashes>#{1,6})\s+(?P<rest>.*?)\s*$")
47PLAN_HEADING_RE = re.compile(r"^Plan:\s*(?P<title>.*)$")
48CONFIG_HEADING_RE = re.compile(r"^Config$")
49#: The separator is EXACTLY one space, two hyphens, one space -- the form the
50#: schema documents. A looser ``\s+--\s+`` accepted ``key -- title`` and
51#: silently produced a different title from the one the author typed, which is
52#: the worst outcome available: not an error, just a quietly wrong plan.
53EPIC_HEADING_RE = re.compile(r"^Epic: (?P<key>[^\s]+) -- (?P<title>.*)$")
54ISSUE_HEADING_RE = re.compile(r"^Issue: (?P<key>[^\s]+) -- (?P<title>.*)$")
55BULLET_RE = re.compile(r"^-\s+(?P<name>[A-Za-z][A-Za-z0-9-]*):\s*(?P<value>.*)$")
56PRIORITY_RE = re.compile(r"^P[0-3]$")
57
58#: Unicode ranges that can alter a terminal without being printable plan data.
59C0_LIMIT = 0x20
60DEL = 0x7F
61C1_LIMIT = 0xA0
62
63#: Unicode's Bidi_Control property from PropList.txt. These characters can
64#: reorder or relabel terminal text without changing the visible data bytes.
65BIDI_FORMAT_CONTROLS = frozenset(
66 {
67 0x061C,
68 0x200E,
69 0x200F,
70 *range(0x202A, 0x202F),
71 *range(0x2066, 0x206A),
72 }
73)
74
75#: ``str.splitlines`` treats these as line boundaries, so they must be refused
76#: before the Markdown parser can mistake data for document structure.
77UNICODE_LINE_SEPARATORS = frozenset({0x2028, 0x2029})
78
79#: Heading depth each section kind is written at.
80LEVEL_PLAN = 1
81LEVEL_SECTION = 2
82LEVEL_ISSUE = 3
83
84KIND_PLAN = "plan"
85KIND_CONFIG = "config"
86KIND_EPIC = "epic"
87KIND_ISSUE = "issue"
88
89
90class PlanError(WorkError):
91 """One or more notes-schema violations, collected together."""
92
93 def __init__(self, problems: list[str]) -> None:
94 """Store the collected problems and build a single-line summary.
95
96 Args:
97 problems: Human-readable violations, already carrying line numbers
98 where a line is meaningful.
99 """
100 self.problems = list(problems)
101 super().__init__(f"{len(self.problems)} problem(s) in the notes file")
102
103
104@dataclass
105class _Section:
106 """One heading and everything that followed it, before validation."""
107
108 kind: str
109 key: str
110 title: str
111 line: int
112 body: list[str] = field(default_factory=list)
113 meta: list[tuple[int, str, str]] = field(default_factory=list)
114
115
116@dataclass(frozen=True)
117class Node:
118 """One validated epic or issue."""
119
120 key: str
121 kind: str
122 title: str
123 epic: str | None
124 line: int
125 body: str
126 labels: tuple[str, ...]
127 priority: str | None
128 track: str | None
129 status: str | None
130 depends_on: tuple[str, ...]
131 estimate: str | None
132
133 def to_dict(self) -> dict[str, object]:
134 """Return the JSON-serialisable form of this node.
135
136 Returns:
137 A mapping with every field, using null for absent optional values.
138 """
139 return {
140 "key": self.key,
141 "kind": self.kind,
142 "title": self.title,
143 "epic": self.epic,
144 "body": self.body,
145 "labels": list(self.labels),
146 "priority": self.priority,
147 "track": self.track,
148 "status": self.status,
149 "depends_on": list(self.depends_on),
150 "estimate": self.estimate,
151 }
152
153
154@dataclass(frozen=True)
155class Plan:
156 """A validated, ordered plan."""
157
158 title: str
159 github_host: str
160 repository: str
161 project_owner: str
162 project_number: int
163 statuses: tuple[str, ...]
164 tracks: tuple[str, ...]
165 nodes: tuple[Node, ...]
166 order: tuple[str, ...]
167
168 def by_key(self) -> dict[str, Node]:
169 """Return the nodes indexed by key.
170
171 Returns:
172 A mapping from key to node.
173 """
174 return {node.key: node for node in self.nodes}
175
176
177def _is_disallowed_control(value: str, *, allow_newline: bool) -> bool:
178 """Return whether one character can alter parsing or terminal display."""
179 codepoint = ord(value)
180 if value == "\n" and allow_newline:
181 return False
182 return (
183 codepoint < C0_LIMIT
184 or DEL <= codepoint < C1_LIMIT
185 or codepoint in UNICODE_LINE_SEPARATORS
186 or codepoint in BIDI_FORMAT_CONTROLS
187 )
188
189
190def _notes_control_problems(text: str) -> list[str]:
191 """Locate controls before line-oriented parsing can reinterpret them."""
192 problems: list[str] = []
193 line = 1
194 for value in text:
195 if _is_disallowed_control(value, allow_newline=True):
196 problems.append(f"line {line}: disallowed control U+{ord(value):04X} in notes")
197 if value == "\n":
198 line += 1
199 return problems
200
201
202def _plan_control_problems(plan: Plan) -> list[str]:
203 """Validate every scalar even when a caller constructed a Plan directly."""
204 fields: list[tuple[str, str, bool]] = [
205 ("plan title", plan.title, False),
206 *(("status", value, False) for value in plan.statuses),
207 *(("track", value, False) for value in plan.tracks),
208 *(("order key", value, False) for value in plan.order),
209 ]
210 for node in plan.nodes:
211 fields.extend(
212 (
213 ("node key", node.key, False),
214 ("node title", node.title, False),
215 ("node body", node.body, True),
216 *(("label", value, False) for value in node.labels),
217 *(("dependency", value, False) for value in node.depends_on),
218 )
219 )
220 fields.extend(
221 (name, value, False)
222 for name, value in (
223 ("epic", node.epic),
224 ("priority", node.priority),
225 ("track", node.track),
226 ("status", node.status),
227 ("estimate", node.estimate),
228 )
229 if value is not None
230 )
231 return [
232 f"{name} contains disallowed control U+{ord(value):04X}"
233 for name, text, allow_newline in fields
234 for value in text
235 if _is_disallowed_control(value, allow_newline=allow_newline)
236 ]
237
238
239def _require_safe_plan(plan: Plan) -> None:
240 """Refuse renderer input that bypassed the notes parser."""
241 if problems := _plan_control_problems(plan):
242 raise PlanError(problems)
243
244
245def _make_section(match: re.Match[str], number: int, problems: list[str]) -> _Section | None:
246 """Turn one heading line into a section, or record why it is not one.
247
248 Args:
249 match: A successful :data:`HEADING_RE` match.
250 number: One-based line number, for diagnostics.
251 problems: Collector appended to on a malformed heading.
252
253 Returns:
254 The new section, or None when the heading was rejected.
255 """
256 level = len(match.group("hashes"))
257 rest = match.group("rest")
258 if level == LEVEL_PLAN and (plan := PLAN_HEADING_RE.match(rest)):
259 return _Section(KIND_PLAN, "", plan.group("title").strip(), number)
260 if level == LEVEL_SECTION and CONFIG_HEADING_RE.match(rest):
261 return _Section(KIND_CONFIG, "", "", number)
262 if level == LEVEL_SECTION and (epic := EPIC_HEADING_RE.match(rest)):
263 return _Section(KIND_EPIC, epic.group("key"), epic.group("title").strip(), number)
264 if level == LEVEL_ISSUE and (issue := ISSUE_HEADING_RE.match(rest)):
265 return _Section(KIND_ISSUE, issue.group("key"), issue.group("title").strip(), number)
266 problems.append(
267 f"line {number}: unrecognised heading. Expected one of "
268 "'# Plan: <title>', '## Config', '## Epic: <key> -- <title>', "
269 "'### Issue: <key> -- <title>'"
270 )
271 return None
272
273
274def _absorb(section: _Section, number: int, line: str, problems: list[str]) -> None:
275 """Add one non-heading line to the section it belongs to.
276
277 Args:
278 section: The section currently being filled.
279 number: One-based line number, for diagnostics.
280 line: The raw line, newline already stripped.
281 problems: Collector appended to when body text follows the metadata.
282 """
283 if not line.strip():
284 if not section.meta:
285 section.body.append("")
286 return
287 bullet = BULLET_RE.match(line)
288 if bullet is not None:
289 section.meta.append((number, bullet.group("name").lower(), bullet.group("value").strip()))
290 return
291 if section.meta:
292 problems.append(
293 f"line {number}: body text is not allowed after the metadata bullets; "
294 "move it above the first bullet"
295 )
296 return
297 section.body.append(line.rstrip())
298
299
300def _scan(text: str) -> tuple[list[_Section], list[str]]:
301 """Split the notes into sections without validating any of their contents.
302
303 Args:
304 text: The whole notes file.
305
306 Returns:
307 The sections in document order, and any structural problems found.
308 """
309 sections: list[_Section] = []
310 problems: list[str] = []
311 current: _Section | None = None
312 for number, raw in enumerate(text.splitlines(), start=1):
313 line = raw.rstrip()
314 heading = HEADING_RE.match(line)
315 if heading is not None:
316 current = _make_section(heading, number, problems)
317 if current is not None:
318 sections.append(current)
319 continue
320 if current is None:
321 if line.strip():
322 problems.append(f"line {number}: text appears before the '# Plan:' header")
323 continue
324 _absorb(current, number, line, problems)
325 return sections, problems
326
327
328def _split_list(value: str) -> tuple[str, ...]:
329 """Split a comma-separated bullet value into stripped, non-empty parts.
330
331 Args:
332 value: The raw text after the bullet colon.
333
334 Returns:
335 The parts, in order.
336 """
337 return tuple(part.strip() for part in value.split(",") if part.strip())
338
339
340def _collect_meta(
341 section: _Section, allowed: tuple[str, ...], problems: list[str]
342) -> dict[str, str]:
343 """Validate a section's bullets against ``allowed`` and return them by name.
344
345 Args:
346 section: The section whose metadata block is being read.
347 allowed: Bullet names recognised in this kind of section.
348 problems: Collector appended to for unknown or duplicated bullets.
349
350 Returns:
351 A mapping from bullet name to its raw value.
352 """
353 found: dict[str, str] = {}
354 for number, name, value in section.meta:
355 if name not in allowed:
356 joined = ", ".join(allowed)
357 problems.append(f"line {number}: unknown metadata bullet '{name}'. Allowed: {joined}")
358 continue
359 if name in found:
360 problems.append(f"line {number}: metadata bullet '{name}' is repeated")
361 continue
362 found[name] = value
363 return found
364
365
366def _config_from(
367 sections: list[_Section], problems: list[str]
368) -> tuple[tuple[str, ...], tuple[str, ...]]:
369 """Read the optional ``## Config`` section, falling back to the defaults.
370
371 Args:
372 sections: Every scanned section.
373 problems: Collector appended to when more than one Config appears.
374
375 Returns:
376 A two-element tuple of the accepted statuses and the accepted tracks.
377 """
378 configs = [item for item in sections if item.kind == KIND_CONFIG]
379 if len(configs) > 1:
380 problems.append(f"line {configs[1].line}: a second '## Config' section is not allowed")
381 authority = tracker_schema()
382 default_statuses = authority.statuses
383 default_tracks = authority.tracks
384 if not configs:
385 return default_statuses, default_tracks
386 meta = _collect_meta(configs[0], CONFIG_BULLETS, problems)
387 statuses = _split_list(meta["statuses"]) if "statuses" in meta else default_statuses
388 tracks = _split_list(meta["tracks"]) if "tracks" in meta else default_tracks
389 return statuses, tracks
390
391
392def _check_membership(
393 node_meta: dict[str, str], line: int, plan_bits: dict[str, tuple[str, ...]], problems: list[str]
394) -> None:
395 """Validate the priority, status and track values of one item.
396
397 Args:
398 node_meta: The item's collected bullets.
399 line: The heading line of the item, for diagnostics.
400 plan_bits: Mapping with ``statuses`` and ``tracks`` allow-lists.
401 problems: Collector appended to for each rejected value.
402 """
403 priority = node_meta.get("priority")
404 if priority is not None and not PRIORITY_RE.match(priority):
405 problems.append(f"line {line}: priority '{priority}' must be one of P0, P1, P2, P3")
406 status = node_meta.get("status")
407 if status is not None and status not in plan_bits["statuses"]:
408 allowed = ", ".join(plan_bits["statuses"])
409 problems.append(f"line {line}: status '{status}' is not one of: {allowed}")
410 track = node_meta.get("track")
411 if track is not None and track not in plan_bits["tracks"]:
412 allowed = ", ".join(plan_bits["tracks"])
413 problems.append(f"line {line}: track '{track}' is not one of: {allowed}")
414
415
416def _check_required_metadata(
417 section: _Section,
418 epic: str | None,
419 meta: dict[str, str],
420 problems: list[str],
421) -> None:
422 """Require the board fields and mirrored priority/epic labels CLAUDE.md mandates."""
423 required = ("status", "track", "priority", "labels")
424 problems.extend(
425 f"line {section.line}: {section.kind} '{section.key}' requires {name}"
426 for name in required
427 if not meta.get(name)
428 )
429 labels = _split_list(meta.get("labels", ""))
430 priority_labels = [label for label in labels if label.startswith("priority:")]
431 expected_priority = f"priority:{meta.get('priority', '')}"
432 if priority_labels != [expected_priority]:
433 problems.append(
434 f"line {section.line}: labels must contain exactly one "
435 f"{expected_priority} matching priority"
436 )
437 expected_epic = f"epic:{section.key if section.kind == KIND_EPIC else epic}"
438 epic_labels = [label for label in labels if label.startswith("epic:")]
439 if epic_labels != [expected_epic]:
440 problems.append(
441 f"line {section.line}: labels must contain exactly one "
442 f"{expected_epic} matching ownership"
443 )
444
445
446def _build_node(
447 section: _Section, epic: str | None, plan_bits: dict[str, tuple[str, ...]], problems: list[str]
448) -> Node:
449 """Validate one epic or issue section and build its node.
450
451 Args:
452 section: The scanned section.
453 epic: The enclosing epic key, or None for an epic itself.
454 plan_bits: Mapping with the ``statuses`` and ``tracks`` allow-lists.
455 problems: Collector appended to for every violation found.
456
457 Returns:
458 The node, built even when problems were recorded so that later checks
459 (dependencies, cycles) can still report everything in one pass.
460 """
461 if not KEY_RE.match(section.key):
462 problems.append(
463 f"line {section.line}: key '{section.key}' must match "
464 "a lowercase slug of 1 to 63 characters, letters, digits and dashes"
465 )
466 if not section.title:
467 problems.append(f"line {section.line}: {section.kind} '{section.key}' has an empty title")
468 meta = _collect_meta(section, ITEM_BULLETS, problems)
469 _check_membership(meta, section.line, plan_bits, problems)
470 _check_required_metadata(section, epic, meta, problems)
471 return Node(
472 key=section.key,
473 kind=section.kind,
474 title=section.title,
475 epic=epic,
476 line=section.line,
477 body="\n".join(section.body).strip(),
478 labels=_split_list(meta.get("labels", "")),
479 priority=meta.get("priority"),
480 track=meta.get("track"),
481 status=meta.get("status"),
482 depends_on=_split_list(meta.get("depends-on", "")),
483 estimate=meta.get("estimate"),
484 )
485
486
487def _build_nodes(
488 sections: list[_Section], plan_bits: dict[str, tuple[str, ...]], problems: list[str]
489) -> list[Node]:
490 """Walk the sections in order, attaching issues to the epic above them.
491
492 Args:
493 sections: Every scanned section.
494 plan_bits: Mapping with the ``statuses`` and ``tracks`` allow-lists.
495 problems: Collector appended to for epic-less issues and duplicate keys.
496
497 Returns:
498 The nodes in document order.
499 """
500 nodes: list[Node] = []
501 seen: dict[str, int] = {}
502 epic: str | None = None
503 for section in sections:
504 if section.kind not in (KIND_EPIC, KIND_ISSUE):
505 continue
506 if section.kind == KIND_EPIC:
507 epic = section.key
508 elif epic is None:
509 problems.append(
510 f"line {section.line}: issue '{section.key}' has no enclosing "
511 "'## Epic:' section above it"
512 )
513 if section.key in seen:
514 problems.append(
515 f"line {section.line}: key '{section.key}' is already used at line "
516 f"{seen[section.key]}"
517 )
518 else:
519 seen[section.key] = section.line
520 nodes.append(
521 _build_node(section, None if section.kind == KIND_EPIC else epic, plan_bits, problems)
522 )
523 return nodes
524
525
526def _check_dependencies(nodes: list[Node], problems: list[str]) -> None:
527 """Validate that every dependency names a real, different key.
528
529 Args:
530 nodes: Every built node.
531 problems: Collector appended to for unknown and self dependencies.
532 """
533 known = {node.key for node in nodes}
534 for node in nodes:
535 for target in node.depends_on:
536 if target == node.key:
537 problems.append(f"line {node.line}: '{node.key}' depends on itself")
538 elif target not in known:
539 problems.append(f"line {node.line}: '{node.key}' depends on unknown key '{target}'")
540
541
542_WHITE = 0
543_GREY = 1
544_BLACK = 2
545
546
547def _next_targets(graph: dict[str, tuple[str, ...]], node: str) -> list[str]:
548 """Return ``node``'s dependencies ordered so that ``pop()`` yields the smallest.
549
550 Args:
551 graph: Mapping from key to the keys it depends on.
552 node: The key whose edges are wanted.
553
554 Returns:
555 The targets in reverse lexicographic order, to be consumed from the end.
556 """
557 return sorted(graph.get(node, ()), reverse=True)
558
559
560def _descend(
561 start: str, graph: dict[str, tuple[str, ...]], state: dict[str, int]
562) -> list[str] | None:
563 """Iteratively depth-first search from ``start`` for the first cycle it closes.
564
565 The traversal carries its own stack rather than using the interpreter's.
566 The graph comes from a notes file, so its depth is chosen by whoever wrote
567 that file: a 2000-key dependency chain is a perfectly ordinary plan and a
568 recursive walk would have raised ``RecursionError`` on it, turning a valid
569 input into a crash.
570
571 Args:
572 start: The key to descend from.
573 graph: Mapping from key to the keys it depends on.
574 state: Shared colouring, mutated in place across calls.
575
576 Returns:
577 The closing cycle path, or None when this subtree has none.
578 """
579 path: list[str] = []
580 pending: list[tuple[str, list[str]]] = [(start, _next_targets(graph, start))]
581 state[start] = _GREY
582 path.append(start)
583 while pending:
584 node, remaining = pending[-1]
585 if not remaining:
586 state[node] = _BLACK
587 pending.pop()
588 path.pop()
589 continue
590 target = remaining.pop()
591 if target not in graph:
592 continue
593 colour = state.get(target, _WHITE)
594 if colour == _GREY:
595 return [*path[path.index(target) :], target]
596 if colour == _WHITE:
597 state[target] = _GREY
598 path.append(target)
599 pending.append((target, _next_targets(graph, target)))
600 return None
601
602
603def find_cycle(graph: dict[str, tuple[str, ...]]) -> list[str] | None:
604 """Return one concrete dependency cycle, or None when the graph is acyclic.
605
606 Args:
607 graph: Mapping from key to the keys it depends on.
608
609 Returns:
610 A path whose first and last element are the same key, or None.
611 """
612 state: dict[str, int] = {}
613 for key in sorted(graph):
614 if state.get(key, _WHITE) == _WHITE:
615 found = _descend(key, graph, state)
616 if found is not None:
617 return found
618 return None
619
620
621def topological_order(nodes: list[Node], problems: list[str]) -> tuple[str, ...]:
622 """Order the nodes so every dependency precedes its dependants.
623
624 Kahn with a min-heap: among the keys whose dependencies are all satisfied,
625 the lexicographically smallest is always taken next, so the order is a
626 function of the notes alone and not of dictionary iteration.
627
628 Args:
629 nodes: Every built node.
630 problems: Collector appended to with one concrete cycle path when the
631 graph does not fully drain.
632
633 Returns:
634 The ordered keys, empty when a cycle was reported.
635 """
636 known = {node.key for node in nodes}
637 graph = {
638 node.key: tuple(
639 dict.fromkeys(
640 [
641 *(target for target in node.depends_on if target in known),
642 *([node.epic] if node.epic in known else []),
643 ]
644 )
645 )
646 for node in nodes
647 }
648 remaining = {key: len(set(deps)) for key, deps in graph.items()}
649 dependants: dict[str, list[str]] = {key: [] for key in graph}
650 for key, deps in graph.items():
651 for target in set(deps):
652 dependants[target].append(key)
653 ready = [key for key, count in remaining.items() if count == 0]
654 heapq.heapify(ready)
655 order: list[str] = []
656 while ready:
657 key = heapq.heappop(ready)
658 order.append(key)
659 for follower in sorted(dependants[key]):
660 remaining[follower] -= 1
661 if remaining[follower] == 0:
662 heapq.heappush(ready, follower)
663 if len(order) != len(graph):
664 cycle = find_cycle({k: v for k, v in graph.items() if remaining[k] > 0})
665 path = " -> ".join(cycle) if cycle else "unresolved"
666 problems.append(f"dependency cycle: {path}")
667 return ()
668 return tuple(order)
669
670
671def parse_plan(text: str) -> Plan:
672 """Parse and validate a notes file into an ordered plan.
673
674 Args:
675 text: The whole notes file, as read from disk.
676
677 Returns:
678 The validated plan.
679
680 Raises:
681 PlanError: One or more schema violations. Every problem found is
682 carried, not just the first.
683 """
684 if control_problems := _notes_control_problems(text):
685 raise PlanError(control_problems)
686 sections, problems = _scan(text)
687 plans = [item for item in sections if item.kind == KIND_PLAN]
688 if not plans:
689 problems.append("missing required '# Plan: <title>' header")
690 elif len(plans) > 1:
691 problems.append(f"line {plans[1].line}: a second '# Plan:' header is not allowed")
692 if plans and not plans[0].title:
693 problems.append(f"line {plans[0].line}: the plan header has an empty title")
694 authority = tracker_schema()
695 statuses, tracks = _config_from(sections, problems)
696 plan_bits = {"statuses": statuses, "tracks": tracks}
697 nodes = _build_nodes(sections, plan_bits, problems)
698 if not nodes:
699 problems.append("plan must contain at least one epic or issue")
700 _check_dependencies(nodes, problems)
701 order = topological_order(nodes, problems)
702 if problems:
703 raise PlanError(problems)
704 return Plan(
705 title=plans[0].title,
706 github_host=authority.github_host,
707 repository=authority.repository,
708 project_owner=authority.project_owner,
709 project_number=authority.project_number,
710 statuses=statuses,
711 tracks=tracks,
712 nodes=tuple(nodes),
713 order=order,
714 )
715
716
717def load_plan(path: Path) -> Plan:
718 """Read a notes file from disk and parse it.
719
720 Args:
721 path: The notes file.
722
723 Returns:
724 The validated plan.
725
726 Raises:
727 PlanError: The file could not be read, or violates the schema.
728 """
729 try:
730 text = path.read_text(encoding="utf-8")
731 except OSError as exc:
732 problems = [f"could not be read: {exc}"]
733 raise PlanError(problems) from exc
734 except UnicodeDecodeError as exc:
735 problems = [
736 f"is not valid UTF-8 text: byte {exc.object[exc.start]:#04x} at offset "
737 f"{exc.start} could not be decoded ({exc.reason})"
738 ]
739 raise PlanError(problems) from exc
740 return parse_plan(text)
741
742
743def plan_to_json(plan: Plan) -> str:
744 """Render the plan as deterministic JSON.
745
746 No timestamp, no host name, no path: the same notes always produce the same
747 bytes, which is what makes the output diffable and testable.
748
749 Args:
750 plan: The validated plan.
751
752 Returns:
753 JSON text with sorted object keys and a trailing newline.
754 """
755 _require_safe_plan(plan)
756 payload = {
757 "schema_version": PLAN_SCHEMA_VERSION,
758 "title": plan.title,
759 "config": {
760 "github_host": plan.github_host,
761 "repository": plan.repository,
762 "project_owner": plan.project_owner,
763 "project_number": plan.project_number,
764 "statuses": list(plan.statuses),
765 "tracks": list(plan.tracks),
766 },
767 "nodes": [node.to_dict() for node in plan.nodes],
768 "order": list(plan.order),
769 }
770 return json.dumps(payload, indent=2, sort_keys=True) + "\n"
771
772
773def render_summary(plan: Plan) -> str:
774 """Render the human-readable summary of a plan.
775
776 Args:
777 plan: The validated plan.
778
779 Returns:
780 Multi-line text ending in a newline.
781 """
782 _require_safe_plan(plan)
783 index = plan.by_key()
784 epics = sum(1 for node in plan.nodes if node.kind == KIND_EPIC)
785 issues = len(plan.nodes) - epics
786 lines = [
787 f"Plan: {plan.title}",
788 f" epics {epics} issues {issues} ordered {len(plan.order)}",
789 f" statuses: {', '.join(plan.statuses)}",
790 f" tracks: {', '.join(plan.tracks)}",
791 "",
792 "Order:",
793 ]
794 for position, key in enumerate(plan.order, start=1):
795 node = index[key]
796 lines.append(f" {position:3d}. [{node.kind}] {node.key} -- {node.title}")
797 if node.depends_on:
798 lines.append(f" depends-on: {', '.join(node.depends_on)}")
799 return "\n".join(lines) + "\n"