from __future__ import annotations
import json
import re
import hashlib
from dataclasses import dataclass
from pathlib import Path
from typing import Any
DECISIONS = {"allowed", "warn", "needs_human", "blocked"}
TERMINAL_BLOCKING_STATES = {
"abandoned",
"duplicate",
"reserved_internal",
"security_private",
}
class SpecRailError(ValueError):
@dataclass(frozen=True)
class PackConfig:
repo: Path
workflow: dict[str, Any]
states: dict[str, Any]
labels: dict[str, Any]
def read_text(path: Path) -> str:
try:
return path.read_text(encoding="utf-8")
except OSError as exc:
raise SpecRailError(f"cannot read {path}: {exc}") from exc
def parse_scalar(value: str) -> Any:
value = value.strip()
if value in {"true", "True"}:
return True
if value in {"false", "False"}:
return False
if value in {"null", "None", "~"}:
return None
if value == "[]":
return []
if value.startswith("[") and value.endswith("]"):
inner = value[1:-1].strip()
if not inner:
return []
return [parse_scalar(item.strip()) for item in inner.split(",")]
if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}:
return value[1:-1]
if re.fullmatch(r"-?[0-9]+", value):
return int(value)
return value
def _significant_lines(text: str) -> list[tuple[int, str]]:
lines: list[tuple[int, str]] = []
for raw in text.splitlines():
if not raw.strip() or raw.lstrip().startswith("#"):
continue
indent = len(raw) - len(raw.lstrip(" "))
lines.append((indent, raw.strip()))
return lines
def parse_yaml_subset(text: str) -> Any:
lines = _significant_lines(text)
def parse_block(index: int, indent: int) -> tuple[Any, int]:
if index >= len(lines):
return {}, index
container: Any = [] if lines[index][1].startswith("- ") else {}
while index < len(lines):
line_indent, content = lines[index]
if line_indent < indent:
break
if line_indent > indent:
raise SpecRailError(f"unexpected indent near: {content}")
if isinstance(container, list):
if not content.startswith("- "):
break
item = content[2:].strip()
if not item:
child, index = parse_block(index + 1, indent + 2)
container.append(child)
else:
container.append(parse_scalar(item))
index += 1
continue
if content.startswith("- "):
break
key, sep, value = content.partition(":")
if not sep:
raise SpecRailError(f"expected key/value near: {content}")
key = key.strip()
value = value.strip()
if value:
container[key] = parse_scalar(value)
index += 1
continue
if index + 1 < len(lines) and lines[index + 1][0] > line_indent:
child, index = parse_block(index + 1, lines[index + 1][0])
container[key] = child
else:
container[key] = {}
index += 1
return container, index
parsed, end = parse_block(0, lines[0][0] if lines else 0)
if end != len(lines):
raise SpecRailError(f"could not parse YAML near: {lines[end][1]}")
return parsed
def load_yaml_file(path: Path) -> Any:
return parse_yaml_subset(read_text(path))
def load_pack(repo: Path) -> PackConfig:
repo = repo.resolve()
return PackConfig(
repo=repo,
workflow=load_yaml_file(repo / "workflow.yaml"),
states=load_yaml_file(repo / "states.yaml"),
labels=load_yaml_file(repo / "labels.yaml"),
)
def state_map(config: PackConfig) -> dict[str, Any]:
states = config.states.get("states")
if not isinstance(states, dict):
raise SpecRailError("states.yaml must contain a states mapping")
return states
def label_groups(config: PackConfig) -> dict[str, list[str]]:
labels = config.labels.get("labels")
if not isinstance(labels, dict):
raise SpecRailError("labels.yaml must contain a labels mapping")
groups: dict[str, list[str]] = {}
for group, values in labels.items():
groups[group] = [str(value) for value in values] if isinstance(values, list) else []
return groups
def action_policy(config: PackConfig) -> dict[str, Any]:
policy = config.workflow.get("action_policy", {})
actions = policy.get("actions", {}) if isinstance(policy, dict) else {}
if not isinstance(actions, dict):
raise SpecRailError("workflow.yaml action_policy.actions must be a mapping")
return actions
def artifact_templates(config: PackConfig) -> dict[str, str]:
artifacts = config.workflow.get("artifacts", {})
if not isinstance(artifacts, dict):
raise SpecRailError("workflow.yaml artifacts must be a mapping")
return {str(key): str(value) for key, value in artifacts.items()}
def work_id_for_issue(issue: int | None) -> str | None:
if issue is None:
return None
return f"GH{issue}"
def render_artifact_path(config: PackConfig, artifact: str, issue: int | None) -> str | None:
template = artifact_templates(config).get(artifact)
if not template:
return None
if issue is None:
return template
return (
template.replace("{issue_number}", str(issue))
.replace("{work_id}", work_id_for_issue(issue) or "")
)
def infer_state(config: PackConfig, state: str | None, labels: list[str]) -> tuple[str | None, list[str]]:
if state:
return state, [f"state provided explicitly: {state}"]
known_states = set(state_map(config))
label_set = {label.strip() for label in labels if label.strip()}
matches = sorted(label_set & known_states)
if len(matches) == 1:
return matches[0], [f"state inferred from label: {matches[0]}"]
if len(matches) > 1:
raise SpecRailError(f"conflicting state labels: {', '.join(matches)}")
return None, []
def validate_state_graph(config: PackConfig) -> list[str]:
errors: list[str] = []
states = state_map(config)
for name, body in states.items():
if not isinstance(body, dict):
errors.append(f"states.yaml: state {name} must be a mapping")
continue
if "owner" not in body:
errors.append(f"states.yaml: state {name} missing owner")
next_states = body.get("next", [])
if body.get("terminal") is True and next_states:
errors.append(f"states.yaml: terminal state {name} must not define next")
if next_states and not isinstance(next_states, list):
errors.append(f"states.yaml: state {name} next must be a list")
continue
for next_state in next_states:
if str(next_state) not in states:
errors.append(f"states.yaml: state {name} references unknown next state {next_state}")
return errors
def validate_labels(config: PackConfig) -> list[str]:
errors: list[str] = []
states = set(state_map(config))
groups = label_groups(config)
for required_group in ["readiness", "outcome", "review"]:
if required_group not in groups:
errors.append(f"labels.yaml: missing label group {required_group}")
for state in ["needs_info", "triaged", "ready_to_spec", "ready_to_implement"]:
if state not in groups.get("readiness", []):
errors.append(f"labels.yaml: readiness labels missing {state}")
for label in groups.get("readiness", []) + groups.get("outcome", []):
if label not in states and label not in {"merged"}:
errors.append(f"labels.yaml: label {label} is not a known state or allowed outcome")
return errors
def validate_action_policy(config: PackConfig) -> list[str]:
errors: list[str] = []
states = set(state_map(config))
actions = action_policy(config)
for route in ["triage_issue", "write_spec", "implement", "review_pr", "fix_ci", "draft_release_note"]:
if route not in actions:
errors.append(f"workflow.yaml: action_policy missing route {route}")
for route, body in actions.items():
if not isinstance(body, dict):
errors.append(f"workflow.yaml: action {route} must be a mapping")
continue
allowed_from = body.get("allowed_from", [])
if not isinstance(allowed_from, list):
errors.append(f"workflow.yaml: action {route} allowed_from must be a list")
continue
for state in allowed_from:
if str(state) not in states:
errors.append(f"workflow.yaml: action {route} references unknown state {state}")
return errors
def validate_template_parity(repo: Path) -> list[str]:
errors: list[str] = []
root = repo / "templates"
zh = root / "zh-CN"
base_files = sorted(path.name for path in root.glob("*.md"))
zh_files = sorted(path.name for path in zh.glob("*.md")) if zh.is_dir() else []
for name in base_files:
if name not in zh_files:
errors.append(f"templates/zh-CN: missing localized template {name}")
for name in zh_files:
if name not in base_files:
errors.append(f"templates/zh-CN/{name}: no matching base template")
stable_tokens = ["GH-", "ready_to_spec", "ready_to_implement"]
for name in ["issue_feature.md", "product_spec.md", "tech_spec.md", "pull_request.md"]:
for rel in [Path("templates") / name, Path("templates/zh-CN") / name]:
path = repo / rel
if not path.is_file():
continue
text = read_text(path)
for token in stable_tokens:
if token in read_text(repo / "templates" / name) and token not in text:
errors.append(f"{rel}: missing stable token {token}")
return errors
def validate_json_schemas(repo: Path) -> list[str]:
errors: list[str] = []
schema_dir = repo / "schemas"
if not schema_dir.is_dir():
return ["missing schemas/ directory"]
for path in sorted(schema_dir.glob("*.schema.json")):
try:
data = json.loads(read_text(path))
except json.JSONDecodeError as exc:
errors.append(f"{path.relative_to(repo)}: invalid JSON: {exc.msg}")
continue
if "$schema" not in data:
errors.append(f"{path.relative_to(repo)}: missing $schema")
if "title" not in data:
errors.append(f"{path.relative_to(repo)}: missing title")
if data.get("type") != "object":
errors.append(f"{path.relative_to(repo)}: top-level type must be object")
return errors
def _frontmatter(text: str) -> dict[str, str] | None:
if not text.startswith("---\n"):
return None
end = text.find("\n---\n", 4)
if end < 0:
return None
raw = text[4:end]
metadata: dict[str, str] = {}
for line in raw.splitlines():
key, sep, value = line.partition(":")
if not sep:
return None
metadata[key.strip()] = value.strip()
return metadata
def _sha256_file(path: Path) -> str:
return hashlib.sha256(path.read_bytes()).hexdigest()
def validate_skills_lock(repo: Path) -> list[str]:
errors: list[str] = []
lock_path = repo / "skills-lock.json"
if not lock_path.is_file():
return ["missing required file: skills-lock.json"]
try:
lock = json.loads(read_text(lock_path))
except json.JSONDecodeError as exc:
return [f"skills-lock.json: invalid JSON: {exc.msg}"]
if not isinstance(lock, dict):
return ["skills-lock.json: top-level value must be an object"]
if lock.get("version") != 1:
errors.append("skills-lock.json: version must be 1")
if lock.get("algorithm") != "sha256":
errors.append("skills-lock.json: algorithm must be sha256")
skills = lock.get("skills")
if not isinstance(skills, list) or not skills:
errors.append("skills-lock.json: skills must be a non-empty list")
return errors
seen_names: set[str] = set()
seen_paths: set[str] = set()
paths: list[str] = []
for index, item in enumerate(skills, start=1):
if not isinstance(item, dict):
errors.append(f"skills-lock.json: skill #{index} must be an object")
continue
name = item.get("name")
rel_path = item.get("path")
digest = item.get("computedHash")
if not isinstance(name, str) or not name.strip():
errors.append(f"skills-lock.json: skill #{index} missing name")
continue
if not isinstance(rel_path, str) or not rel_path.strip():
errors.append(f"skills-lock.json: skill {name} missing path")
continue
if name in seen_names:
errors.append(f"skills-lock.json: duplicate skill name {name}")
if rel_path in seen_paths:
errors.append(f"skills-lock.json: duplicate skill path {rel_path}")
seen_names.add(name)
seen_paths.add(rel_path)
paths.append(rel_path)
path = Path(rel_path)
if path.is_absolute() or ".." in path.parts:
errors.append(f"skills-lock.json: skill {name} path must be repo-relative")
continue
if path.parts[:1] != ("skills",) or path.name != "SKILL.md":
errors.append(f"skills-lock.json: skill {name} path must be skills/<name>/SKILL.md")
continue
if len(path.parts) != 3 or path.parts[1] != name:
errors.append(f"skills-lock.json: skill {name} path must match its name")
skill_path = repo / path
if not skill_path.is_file():
errors.append(f"skills-lock.json: skill file does not exist: {rel_path}")
continue
text = read_text(skill_path)
metadata = _frontmatter(text)
if metadata is None:
errors.append(f"{rel_path}: missing YAML frontmatter")
else:
if set(metadata) != {"name", "description"}:
errors.append(f"{rel_path}: frontmatter must contain only name and description")
if metadata.get("name") != name:
errors.append(f"{rel_path}: frontmatter name must be {name}")
if not metadata.get("description"):
errors.append(f"{rel_path}: description must not be empty")
if not isinstance(digest, str) or not digest.startswith("sha256:"):
errors.append(f"skills-lock.json: skill {name} computedHash must start with sha256:")
elif digest.removeprefix("sha256:") != _sha256_file(skill_path):
errors.append(f"skills-lock.json: skill {name} computedHash mismatch")
if paths != sorted(paths):
errors.append("skills-lock.json: skills must be sorted by path")
skill_files = {
str(path.relative_to(repo))
for path in sorted((repo / "skills").glob("specrail-*/SKILL.md"))
}
missing_from_lock = sorted(skill_files - seen_paths)
extra_in_lock = sorted(
path for path in seen_paths if path.startswith("skills/specrail-") and path not in skill_files
)
for rel_path in missing_from_lock:
errors.append(f"skills-lock.json: missing skill file {rel_path}")
for rel_path in extra_in_lock:
errors.append(f"skills-lock.json: locked skill file missing from repo {rel_path}")
return errors