3"""Notes-to-plan parser: a strict Markdown schema, validated and topologically ordered.
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.
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.
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.
21Nothing here runs a command, touches the network, or writes outside a path the
25from __future__
import annotations
30from dataclasses
import dataclass, field
31from pathlib
import Path
33from work_git
import WorkError
34from work_tracker
import tracker_schema
35from work_workspace
import KEY_RE
38PLAN_SCHEMA_VERSION = 1
41ITEM_BULLETS = (
"labels",
"priority",
"track",
"status",
"depends-on",
"estimate")
44CONFIG_BULLETS = (
"statuses",
"tracks")
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$")
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]$")
65BIDI_FORMAT_CONTROLS = frozenset(
70 *range(0x202A, 0x202F),
71 *range(0x2066, 0x206A),
77UNICODE_LINE_SEPARATORS = frozenset({0x2028, 0x2029})
90class PlanError(WorkError):
91 """One or more notes-schema violations, collected together."""
93 def __init__(self, problems: list[str]) ->
None:
94 """Store the collected problems and build a single-line summary.
97 problems: Human-readable violations, already carrying line numbers
98 where a line is meaningful.
100 self.problems = list(problems)
101 super().__init__(f
"{len(self.problems)} problem(s) in the notes file")
106 """One heading and everything that followed it, before validation."""
112 body: list[str] = field(default_factory=list)
113 meta: list[tuple[int, str, str]] = field(default_factory=list)
116@dataclass(frozen=True)
118 """One validated epic or issue."""
126 labels: tuple[str, ...]
130 depends_on: tuple[str, ...]
133 def to_dict(self) -> dict[str, object]:
134 """Return the JSON-serialisable form of this node.
137 A mapping with every field, using null for absent optional values.
145 "labels": list(self.labels),
146 "priority": self.priority,
148 "status": self.status,
149 "depends_on": list(self.depends_on),
150 "estimate": self.estimate,
154@dataclass(frozen=True)
156 """A validated, ordered plan."""
163 statuses: tuple[str, ...]
164 tracks: tuple[str, ...]
165 nodes: tuple[Node, ...]
166 order: tuple[str, ...]
168 def by_key(self) -> dict[str, Node]:
169 """Return the nodes indexed by key.
172 A mapping from key to node.
174 return {node.key: node
for node
in self.nodes}
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:
184 or DEL <= codepoint < C1_LIMIT
185 or codepoint
in UNICODE_LINE_SEPARATORS
186 or codepoint
in BIDI_FORMAT_CONTROLS
190def _notes_control_problems(text: str) -> list[str]:
191 """Locate controls before line-oriented parsing can reinterpret them."""
192 problems: list[str] = []
195 if _is_disallowed_control(value, allow_newline=
True):
196 problems.append(f
"line {line}: disallowed control U+{ord(value):04X} in notes")
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),
210 for node
in plan.nodes:
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),
224 (
"priority", node.priority),
225 (
"track", node.track),
226 (
"status", node.status),
227 (
"estimate", node.estimate),
232 f
"{name} contains disallowed control U+{ord(value):04X}"
233 for name, text, allow_newline
in fields
235 if _is_disallowed_control(value, allow_newline=allow_newline)
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)
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.
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.
254 The new section, or None when the heading was rejected.
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)
267 f
"line {number}: unrecognised heading. Expected one of "
268 "'# Plan: <title>', '## Config', '## Epic: <key> -- <title>', "
269 "'### Issue: <key> -- <title>'"
274def _absorb(section: _Section, number: int, line: str, problems: list[str]) ->
None:
275 """Add one non-heading line to the section it belongs to.
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.
285 section.body.append(
"")
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()))
293 f
"line {number}: body text is not allowed after the metadata bullets; "
294 "move it above the first bullet"
297 section.body.append(line.rstrip())
300def _scan(text: str) -> tuple[list[_Section], list[str]]:
301 """Split the notes into sections without validating any of their contents.
304 text: The whole notes file.
307 The sections in document order, and any structural problems found.
309 sections: list[_Section] = []
310 problems: list[str] = []
311 current: _Section |
None =
None
312 for number, raw
in enumerate(text.splitlines(), start=1):
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)
322 problems.append(f
"line {number}: text appears before the '# Plan:' header")
324 _absorb(current, number, line, problems)
325 return sections, problems
328def _split_list(value: str) -> tuple[str, ...]:
329 """Split a comma-separated bullet value into stripped, non-empty parts.
332 value: The raw text after the bullet colon.
337 return tuple(part.strip()
for part
in value.split(
",")
if part.strip())
341 section: _Section, allowed: tuple[str, ...], problems: list[str]
343 """Validate a section's bullets against ``allowed`` and return them by name.
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.
351 A mapping from bullet name to its raw value.
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}")
360 problems.append(f
"line {number}: metadata bullet '{name}' is repeated")
367 sections: list[_Section], problems: list[str]
368) -> tuple[tuple[str, ...], tuple[str, ...]]:
369 """Read the optional ``## Config`` section, falling back to the defaults.
372 sections: Every scanned section.
373 problems: Collector appended to when more than one Config appears.
376 A two-element tuple of the accepted statuses and the accepted tracks.
378 configs = [item
for item
in sections
if item.kind == KIND_CONFIG]
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
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
392def _check_membership(
393 node_meta: dict[str, str], line: int, plan_bits: dict[str, tuple[str, ...]], problems: list[str]
395 """Validate the priority, status and track values of one item.
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.
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}")
416def _check_required_metadata(
419 meta: dict[str, str],
422 """Require the board fields and mirrored priority/epic labels CLAUDE.md mandates."""
423 required = (
"status",
"track",
"priority",
"labels")
425 f
"line {section.line}: {section.kind} '{section.key}' requires {name}"
427 if not meta.get(name)
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]:
434 f
"line {section.line}: labels must contain exactly one "
435 f
"{expected_priority} matching priority"
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]:
441 f
"line {section.line}: labels must contain exactly one "
442 f
"{expected_epic} matching ownership"
447 section: _Section, epic: str |
None, plan_bits: dict[str, tuple[str, ...]], problems: list[str]
449 """Validate one epic or issue section and build its node.
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.
458 The node, built even when problems were recorded so that later checks
459 (dependencies, cycles) can still report everything in one pass.
461 if not KEY_RE.match(section.key):
463 f
"line {section.line}: key '{section.key}' must match "
464 "a lowercase slug of 1 to 63 characters, letters, digits and dashes"
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)
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"),
488 sections: list[_Section], plan_bits: dict[str, tuple[str, ...]], problems: list[str]
490 """Walk the sections in order, attaching issues to the epic above them.
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.
498 The nodes in document order.
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):
506 if section.kind == KIND_EPIC:
510 f
"line {section.line}: issue '{section.key}' has no enclosing "
511 "'## Epic:' section above it"
513 if section.key
in seen:
515 f
"line {section.line}: key '{section.key}' is already used at line "
516 f
"{seen[section.key]}"
519 seen[section.key] = section.line
521 _build_node(section,
None if section.kind == KIND_EPIC
else epic, plan_bits, problems)
526def _check_dependencies(nodes: list[Node], problems: list[str]) ->
None:
527 """Validate that every dependency names a real, different key.
530 nodes: Every built node.
531 problems: Collector appended to for unknown and self dependencies.
533 known = {node.key
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}'")
547def _next_targets(graph: dict[str, tuple[str, ...]], node: str) -> list[str]:
548 """Return ``node``'s dependencies ordered so that ``pop()`` yields the smallest.
551 graph: Mapping from key to the keys it depends on.
552 node: The key whose edges are wanted.
555 The targets in reverse lexicographic order, to be consumed from the end.
557 return sorted(graph.get(node, ()), reverse=
True)
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.
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
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.
577 The closing cycle path, or None when this subtree has none.
580 pending: list[tuple[str, list[str]]] = [(start, _next_targets(graph, start))]
584 node, remaining = pending[-1]
590 target = remaining.pop()
591 if target
not in graph:
593 colour = state.get(target, _WHITE)
595 return [*path[path.index(target) :], target]
597 state[target] = _GREY
599 pending.append((target, _next_targets(graph, target)))
603def find_cycle(graph: dict[str, tuple[str, ...]]) -> list[str] |
None:
604 """Return one concrete dependency cycle, or None when the graph is acyclic.
607 graph: Mapping from key to the keys it depends on.
610 A path whose first and last element are the same key, or None.
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:
621def topological_order(nodes: list[Node], problems: list[str]) -> tuple[str, ...]:
622 """Order the nodes so every dependency precedes its dependants.
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.
629 nodes: Every built node.
630 problems: Collector appended to with one concrete cycle path when the
631 graph does not fully drain.
634 The ordered keys, empty when a cycle was reported.
636 known = {node.key
for node
in nodes}
641 *(target
for target
in node.depends_on
if target
in known),
642 *([node.epic]
if node.epic
in known
else []),
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]
655 order: list[str] = []
657 key = heapq.heappop(ready)
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}")
671def parse_plan(text: str) -> Plan:
672 """Parse and validate a notes file into an ordered plan.
675 text: The whole notes file, as read from disk.
681 PlanError: One or more schema violations. Every problem found is
682 carried, not just the first.
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]
689 problems.append(
"missing required '# Plan: <title>' header")
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)
699 problems.append(
"plan must contain at least one epic or issue")
700 _check_dependencies(nodes, problems)
701 order = topological_order(nodes, problems)
703 raise PlanError(problems)
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,
717def load_plan(path: Path) -> Plan:
718 """Read a notes file from disk and parse it.
721 path: The notes file.
727 PlanError: The file could not be read, or violates the schema.
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:
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})"
739 raise PlanError(problems)
from exc
740 return parse_plan(text)
743def plan_to_json(plan: Plan) -> str:
744 """Render the plan as deterministic JSON.
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.
750 plan: The validated plan.
753 JSON text with sorted object keys and a trailing newline.
755 _require_safe_plan(plan)
757 "schema_version": PLAN_SCHEMA_VERSION,
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),
767 "nodes": [node.to_dict()
for node
in plan.nodes],
768 "order": list(plan.order),
770 return json.dumps(payload, indent=2, sort_keys=
True) +
"\n"
773def render_summary(plan: Plan) -> str:
774 """Render the human-readable summary of a plan.
777 plan: The validated plan.
780 Multi-line text ending in a newline.
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
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)}",
794 for position, key
in enumerate(plan.order, start=1):
796 lines.append(f
" {position:3d}. [{node.kind}] {node.key} -- {node.title}")
798 lines.append(f
" depends-on: {', '.join(node.depends_on)}")
799 return "\n".join(lines) +
"\n"