Files
neuron-tai/scripts/ralph_prd_schema.py
2026-07-16 23:01:55 +03:00

577 lines
22 KiB
Python

#!/usr/bin/env python3
"""Canonical schema, validator, and generator for Ralph `prd.json` backlogs.
`prd.json` is the single authoritative source for a Ralph-managed backlog.
This module defines the field shape every story must have, validates
structural and cross-story invariants (unique IDs/titles, resolvable
`dependsOn`, acyclic dependency graph, `blocks` derived correctly), and
renders the same fields into the generated Markdown/Gitea-issue view used
by `.scratch/<feature>/issues/*.md`.
Generation is one-directional: prd.json -> Markdown / Gitea payload. There
is no Markdown -> prd.json parser, so a generated artifact can never feed
`passes` (or any other field) back into the authoritative source. Every
generated Markdown document must start with the authority disclaimer
(`authority_disclaimer()`, sourced from prd.json's own `metadataSchema` when
present) so a stray edit of the generated file cannot be mistaken for a
source-of-truth change. The shared "quality gates" boilerplate is likewise
sourced from prd.json's `qualityGates`, not duplicated as a Python literal.
Example:
python scripts/ralph_prd_schema.py validate .scratch/distributed-gguf-runtime/prd.json
python scripts/ralph_prd_schema.py render .scratch/distributed-gguf-runtime/prd.json DGR-018
"""
from __future__ import annotations
import argparse
import json
import re
import sys
from pathlib import Path
from typing import Any
REPO_ROOT = Path(__file__).resolve().parents[1]
AUTHORITY_DISCLAIMER = (
"<!-- GENERATED FROM prd.json — DO NOT EDIT AS AN INDEPENDENT SOURCE. "
"prd.json IS AUTHORITATIVE. -->"
)
EXECUTION_MODES = {"AFK", "HITL"}
EVIDENCE_CLASSES = {"model-free", "fixture", "real-model", "real-hardware", "release"}
HARDWARE_FLAGS = {"none", "optional", "required"}
UPSTREAM_FLAGS = {"yes", "no", "conditional"}
TRIAGE_VALUES = {"ready-for-agent", "ready-for-human"}
# Canonical field registry: name -> (required, expected python type(s)).
# This is the single definition of "what a story is" referenced by both
# validation and documentation; keep it in sync with any schema change.
STORY_FIELDS: dict[str, tuple[bool, tuple[type, ...]]] = {
"id": (True, (str,)),
"title": (True, (str,)),
"priority": (True, (int,)),
"milestone": (True, (str,)),
"executionMode": (True, (str,)),
"labels": (True, (list,)),
"evidenceClass": (True, (str,)),
"evidencePath": (True, (str,)),
"hardware": (True, (str,)),
"model": (True, (str,)),
"upstream": (True, (str,)),
"dependsOn": (True, (list,)),
"triage": (True, (str,)),
"description": (True, (str,)),
"acceptanceCriteria": (True, (list,)),
"passes": (True, (bool,)),
"notes": (True, (str,)),
"blocks": (True, (list,)),
"completionNotes": (False, (str,)),
}
_ID_RE = re.compile(r"^[A-Z][A-Z0-9]*-\d+$")
class PrdValidationError(ValueError):
"""Raised with all accumulated validation errors, one per line."""
def __init__(self, errors: list[str]):
super().__init__("\n".join(errors))
self.errors = errors
def load_prd(path: Path) -> dict[str, Any]:
try:
text = path.read_text(encoding="utf-8")
except FileNotFoundError as exc:
raise PrdValidationError([f"PRD not found: {path}"]) from exc
try:
data = json.loads(text)
except json.JSONDecodeError as exc:
raise PrdValidationError([f"Invalid JSON in {path}: {exc}"]) from exc
if not isinstance(data, dict):
raise PrdValidationError([f"{path}: top-level PRD document must be an object"])
return data
def stories_of(data: dict[str, Any]) -> list[dict[str, Any]]:
stories = data.get("userStories")
if not isinstance(stories, list):
return []
return [s for s in stories if isinstance(s, dict)]
def derive_type(labels: list[str]) -> str | None:
"""Derive the story's `type` from its `type:` label, if present.
Release-gate stories (labelled `gate:hitl`) intentionally carry no
`type:` label; they derive the synthetic type `release-gate` instead.
"""
types = [label.split(":", 1)[1] for label in labels if label.startswith("type:")]
if types:
return types[0]
if any(label.startswith("gate:") for label in labels):
return "release-gate"
return None
def _label_prefix_counts(labels: list[str], prefix: str) -> int:
return sum(1 for label in labels if label.startswith(prefix))
def validate_schema(data: dict[str, Any]) -> list[str]:
"""Structural validation: required top-level keys and per-story field shape."""
errors: list[str] = []
for key in ("name", "description", "branchName", "userStories"):
if key not in data:
errors.append(f"prd: missing top-level field '{key}'")
stories = data.get("userStories")
if not isinstance(stories, list):
errors.append("prd: 'userStories' must be a list")
return errors
for index, story in enumerate(stories):
label = f"userStories[{index}]"
if not isinstance(story, dict):
errors.append(f"{label}: story must be an object")
continue
sid = story.get("id", label)
for field, (required, types) in STORY_FIELDS.items():
if field not in story:
if required:
errors.append(f"{sid}: missing required field '{field}'")
continue
value = story[field]
if types == (bool,) and isinstance(value, bool):
continue
if not isinstance(value, types):
errors.append(
f"{sid}: field '{field}' must be {types}, got {type(value).__name__}"
)
if isinstance(story.get("id"), str) and not _ID_RE.match(story["id"]):
errors.append(f"{sid}: id '{story['id']}' does not match PREFIX-NUMBER convention")
if isinstance(story.get("executionMode"), str) and story["executionMode"] not in EXECUTION_MODES:
errors.append(f"{sid}: executionMode '{story['executionMode']}' not in {sorted(EXECUTION_MODES)}")
if isinstance(story.get("evidenceClass"), str) and story["evidenceClass"] not in EVIDENCE_CLASSES:
errors.append(f"{sid}: evidenceClass '{story['evidenceClass']}' not in {sorted(EVIDENCE_CLASSES)}")
if isinstance(story.get("hardware"), str) and story["hardware"] not in HARDWARE_FLAGS:
errors.append(f"{sid}: hardware '{story['hardware']}' not in {sorted(HARDWARE_FLAGS)}")
if isinstance(story.get("upstream"), str) and story["upstream"] not in UPSTREAM_FLAGS:
errors.append(f"{sid}: upstream '{story['upstream']}' not in {sorted(UPSTREAM_FLAGS)}")
if isinstance(story.get("triage"), str) and story["triage"] not in TRIAGE_VALUES:
errors.append(f"{sid}: triage '{story['triage']}' not in {sorted(TRIAGE_VALUES)}")
labels = story.get("labels")
if isinstance(labels, list):
if not all(isinstance(label, str) for label in labels):
errors.append(f"{sid}: labels must all be strings")
elif _label_prefix_counts(labels, "type:") > 1:
errors.append(f"{sid}: at most one 'type:' label allowed")
elif _label_prefix_counts(labels, "priority:") > 1:
errors.append(f"{sid}: at most one 'priority:' label allowed")
depends_on = story.get("dependsOn")
if isinstance(depends_on, list) and not all(isinstance(dep, str) for dep in depends_on):
errors.append(f"{sid}: dependsOn entries must all be strings")
acceptance = story.get("acceptanceCriteria")
if isinstance(acceptance, list):
if not acceptance:
errors.append(f"{sid}: acceptanceCriteria must be non-empty")
elif not all(isinstance(item, str) and item.strip() for item in acceptance):
errors.append(f"{sid}: acceptanceCriteria entries must be non-empty strings")
return errors
def _find_cycle(story_ids: set[str], depends_on: dict[str, list[str]]) -> list[str] | None:
WHITE, GRAY, BLACK = 0, 1, 2
color = {sid: WHITE for sid in story_ids}
path: list[str] = []
def dfs(node: str) -> list[str] | None:
color[node] = GRAY
path.append(node)
for dep in depends_on.get(node, []):
if dep not in color:
continue
if color[dep] == GRAY:
cycle_start = path.index(dep)
return path[cycle_start:] + [dep]
if color[dep] == WHITE:
found = dfs(dep)
if found:
return found
path.pop()
color[node] = BLACK
return None
for sid in sorted(story_ids):
if color[sid] == WHITE:
cycle = dfs(sid)
if cycle:
return cycle
return None
def validate_semantics(data: dict[str, Any]) -> list[str]:
"""Cross-story validation: uniqueness, dependency graph, derived `blocks`."""
errors: list[str] = []
stories = stories_of(data)
ids = [s.get("id") for s in stories if isinstance(s.get("id"), str)]
seen_ids: set[str] = set()
for sid in ids:
if sid in seen_ids:
errors.append(f"duplicate story id: '{sid}'")
seen_ids.add(sid)
titles = [s.get("title") for s in stories if isinstance(s.get("title"), str)]
seen_titles: set[str] = set()
for title in titles:
if title in seen_titles:
errors.append(f"duplicate story title: '{title}'")
seen_titles.add(title)
by_id = {s["id"]: s for s in stories if isinstance(s.get("id"), str)}
depends_on: dict[str, list[str]] = {}
for story in stories:
sid = story.get("id")
if not isinstance(sid, str):
continue
deps = story.get("dependsOn") if isinstance(story.get("dependsOn"), list) else []
deps = [d for d in deps if isinstance(d, str)]
depends_on[sid] = deps
if sid in deps:
errors.append(f"{sid}: cannot depend on itself")
for dep in deps:
if dep not in by_id:
errors.append(f"{sid}: depends on unknown story '{dep}'")
cycle = _find_cycle(set(by_id), depends_on)
if cycle:
errors.append("dependency cycle detected: " + " -> ".join(cycle))
derived_blocks: dict[str, list[str]] = {sid: [] for sid in by_id}
for sid, deps in depends_on.items():
for dep in deps:
if dep in derived_blocks:
derived_blocks[dep].append(sid)
for sid in derived_blocks:
derived_blocks[sid] = sorted(derived_blocks[sid])
for story in stories:
sid = story.get("id")
if not isinstance(sid, str):
continue
given = story.get("blocks") if isinstance(story.get("blocks"), list) else []
given_sorted = sorted(d for d in given if isinstance(d, str))
expected = derived_blocks.get(sid, [])
if given_sorted != expected:
errors.append(
f"{sid}: 'blocks' does not match derived dependents "
f"(got {given_sorted}, expected {expected})"
)
evidence_path = story.get("evidencePath")
if isinstance(evidence_path, str):
expected_path = f".scratch/distributed-gguf-runtime/evidence/{sid}/README.md"
if evidence_path != expected_path:
errors.append(
f"{sid}: evidencePath '{evidence_path}' does not match convention "
f"'{expected_path}'"
)
return errors
def validate_fresh_backlog(data: dict[str, Any]) -> list[str]:
"""Extra invariant for a freshly generated (not-yet-started) backlog.
A fresh backlog must not claim any completion credit: every story
starts `passes: false`. This is distinct from validating an in-flight
backlog (use validate_schema + validate_semantics for that), because a
real backlog legitimately accumulates `passes: true` stories over time.
"""
errors = validate_backlog(data)
for story in stories_of(data):
if story.get("passes") is not False:
errors.append(
f"{story.get('id')}: fresh backlog requires passes=false, "
f"got {story.get('passes')!r}"
)
return errors
def validate_backlog(data: dict[str, Any]) -> list[str]:
"""Full validation for an in-progress backlog: structure, semantics, and
(when present) prd.json's self-documented metadataSchema/qualityGates."""
return (
validate_schema(data)
+ validate_semantics(data)
+ validate_metadata_schema_consistency(data)
)
def validate_metadata_schema_consistency(data: dict[str, Any]) -> list[str]:
"""If prd.json documents its own `metadataSchema`/`qualityGates`, verify
that self-description hasn't drifted from what this validator actually
enforces. Absent entirely, this is a no-op (a minimal fixture PRD is not
required to carry self-documentation to be structurally valid).
"""
errors: list[str] = []
ms = data.get("metadataSchema")
if ms is None:
return errors
enum_checks = {
"triageValues": TRIAGE_VALUES,
"executionModeValues": EXECUTION_MODES,
"evidenceClassValues": EVIDENCE_CLASSES,
"hardwareValues": HARDWARE_FLAGS,
"upstreamValues": UPSTREAM_FLAGS,
}
for key, expected in enum_checks.items():
declared = ms.get(key)
if declared is None:
errors.append(f"metadataSchema: missing '{key}'")
continue
if set(declared) != expected:
errors.append(
f"metadataSchema.{key} {sorted(declared)} does not match "
f"enforced values {sorted(expected)}"
)
required_in_code = {field for field, (required, _) in STORY_FIELDS.items() if required}
declared_required = set(ms.get("requiredStoryFields", []))
if declared_required != required_in_code:
errors.append(
f"metadataSchema.requiredStoryFields {sorted(declared_required)} does not "
f"match code-enforced required fields {sorted(required_in_code)}"
)
optional_in_code = {field for field, (required, _) in STORY_FIELDS.items() if not required}
declared_optional = set(ms.get("optionalStoryFields", []))
if declared_optional != optional_in_code:
errors.append(
f"metadataSchema.optionalStoryFields {sorted(declared_optional)} does not "
f"match code-enforced optional fields {sorted(optional_in_code)}"
)
if not ms.get("generatedArtifactDisclaimer"):
errors.append("metadataSchema: missing 'generatedArtifactDisclaimer'")
if "qualityGates" not in data:
errors.append("prd: metadataSchema is documented but top-level 'qualityGates' is missing")
elif not quality_gate_bullets(data):
errors.append("qualityGates: no bullets declared under universal/native/realModelHardware/scope")
return errors
def quality_gate_bullets(data: dict[str, Any]) -> list[str]:
"""Flatten prd.json's `qualityGates` into the bullet order used by the
generated "Shared quality gates" Markdown section."""
gates = data.get("qualityGates", {})
bullets: list[str] = []
for key in ("universal", "native", "realModelHardware", "scope"):
bullets.extend(gates.get(key, []))
return bullets
def authority_disclaimer(data: dict[str, Any]) -> str:
"""The exact disclaimer generated Markdown/Gitea bodies must start with.
Sourced from prd.json's `metadataSchema.generatedArtifactDisclaimer` when
present, falling back to the module default for minimal fixtures.
"""
return data.get("metadataSchema", {}).get("generatedArtifactDisclaimer", AUTHORITY_DISCLAIMER)
def check_generated_markdown_authority(text: str, disclaimer: str = AUTHORITY_DISCLAIMER) -> list[str]:
"""Reject generated Markdown that fails to defer authority to prd.json."""
errors: list[str] = []
first_line = text.splitlines()[0] if text else ""
if first_line != disclaimer:
errors.append(
"generated Markdown must start with the authority disclaimer: "
f"expected {disclaimer!r}, got {first_line!r}"
)
lowered = text.lower()
for claim in ("this file is authoritative", "this document is authoritative", "markdown is authoritative"):
if claim in lowered:
errors.append(f"generated Markdown claims authority over prd.json: found '{claim}'")
return errors
def slugify(title: str) -> str:
slug = re.sub(r"[^a-z0-9]+", "-", title.lower()).strip("-")
return re.sub(r"-{2,}", "-", slug)
def issue_number(story_id: str) -> str:
match = re.search(r"(\d+)$", story_id)
if not match:
raise ValueError(f"story id '{story_id}' has no trailing number")
return match.group(1)
def issue_filename(story: dict[str, Any]) -> str:
return f"{issue_number(story['id'])}-{slugify(story['title'])}.md"
# Fallback bullets used only when a minimal fixture PRD has no qualityGates
# block of its own; the real backlog always sources these from prd.json via
# quality_gate_bullets() so this list is a default, not a duplicate of truth.
_DEFAULT_QUALITY_GATE_BULLETS = [
"Targeted deterministic tests pass; Python changes also pass `python -m compileall packages tests`.",
"`git diff --check` passes.",
"Default tests are model-download-free, API-credit-free, and GPU-free.",
"Evidence README records exact changed files, commands/results, limitations, and dependency handoff; no fabricated evidence or inherited completion credit.",
]
def render_issue_markdown(story: dict[str, Any], data: dict[str, Any] | None = None) -> str:
"""Render the canonical generated Markdown view of a story.
`data` supplies the shared quality-gates bullets and authority disclaimer
from prd.json's own `qualityGates`/`metadataSchema`; omit it only for
fixtures that don't model those fields. This is byte-for-byte identical
to the convention already used by `.scratch/distributed-gguf-runtime/
issues/*.md`; see the round-trip tests in tests/test_ralph_prd_schema.py.
"""
data = data or {}
sid = story["id"]
passes = story["passes"]
deps = story.get("dependsOn", [])
blocks = sorted(story.get("blocks", []))
deps_text = ", ".join(f"`{d}`" for d in deps) if deps else "None"
blocks_text = ", ".join(f"`{b}`" for b in blocks) if blocks else "None"
labels_text = ", ".join(f"`{label}`" for label in story.get("labels", []))
if passes:
status_line = "completed; `passes: true`"
else:
status_line = f"specification only; `{story['triage']}`; `passes: false`"
checkbox = "x" if passes else " "
acceptance_lines = "\n".join(
f"- [{checkbox}] {criterion}" for criterion in story["acceptanceCriteria"]
)
if passes:
evidence_handoff = (
f"Verified evidence: `{story['evidencePath']}`. Legacy evidence remains "
"provenance only and grants no implementation completion credit."
)
else:
evidence_handoff = (
f"Write and verify `{story['evidencePath']}`. Until every criterion and "
f"applicable gate has real evidence, this story remains `passes: false`. "
"Legacy evidence is provenance only, not completion credit."
)
description = story["description"]
gate_bullets = quality_gate_bullets(data) if "qualityGates" in data else _DEFAULT_QUALITY_GATE_BULLETS
gates_section = "## Shared quality gates\n\n" + "\n".join(f"- {bullet}" for bullet in gate_bullets)
lines = [
authority_disclaimer(data),
f"# {sid}: {story['title']}",
"",
f"- **Status / triage:** {status_line}",
f"- **Execution mode:** `{story['executionMode']}`",
f"- **Milestone:** `{story['milestone']}`",
f"- **Dependencies:** {deps_text}",
f"- **Blocks (derived):** {blocks_text}",
f"- **Labels:** {labels_text}",
f"- **Evidence class:** `{story['evidenceClass']}`",
f"- **Hardware:** `{story['hardware']}`",
f"- **Model:** `{story['model']}`",
f"- **Upstream:** `{story['upstream']}`",
"",
"## Objective / description",
"",
description,
"",
"## Acceptance criteria",
"",
acceptance_lines,
"",
gates_section,
"",
"## Evidence handoff",
"",
evidence_handoff,
"",
]
return "\n".join(lines[:-1]) + "\n"
def to_gitea_issue_payload(story: dict[str, Any], data: dict[str, Any] | None = None) -> dict[str, Any]:
"""Structured payload matching the Gitea create-issue API shape.
`body` is exactly `render_issue_markdown`'s output, so the payload
carries no information that isn't losslessly derivable from prd.json.
"""
return {
"title": f"{story['id']}: {story['title']}",
"body": render_issue_markdown(story, data),
"labels": list(story.get("labels", [])),
"milestone": story["milestone"],
}
def _cmd_validate(args: argparse.Namespace) -> int:
data = load_prd(args.prd)
errors = validate_fresh_backlog(data) if args.fresh else validate_backlog(data)
if errors:
for error in errors:
print(f"ERROR: {error}", file=sys.stderr)
print(f"{len(errors)} validation error(s).", file=sys.stderr)
return 1
print(f"OK: {len(stories_of(data))} stories validated.")
return 0
def _cmd_render(args: argparse.Namespace) -> int:
data = load_prd(args.prd)
by_id = {s["id"]: s for s in stories_of(data)}
story = by_id.get(args.story_id)
if story is None:
print(f"Unknown story id: {args.story_id}", file=sys.stderr)
return 1
print(render_issue_markdown(story, data), end="")
return 0
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description=__doc__)
subparsers = parser.add_subparsers(dest="command", required=True)
validate = subparsers.add_parser("validate", help="Validate a prd.json against the canonical schema")
validate.add_argument("prd", type=Path)
validate.add_argument("--fresh", action="store_true", help="Also require passes=false on every story")
validate.set_defaults(func=_cmd_validate)
render = subparsers.add_parser("render", help="Render the generated Markdown view of one story")
render.add_argument("prd", type=Path)
render.add_argument("story_id")
render.set_defaults(func=_cmd_render)
return parser
def main(argv: list[str]) -> int:
parser = build_parser()
args = parser.parse_args(argv[1:])
return args.func(args)
if __name__ == "__main__":
raise SystemExit(main(sys.argv))