#!/usr/bin/env python3
"""Inspect and validate Omi's agent-friendly failure-class protocol.
Definitions live in .github/failure-classes/ as one JSON file per semantic
failure-class ID. This CLI never classifies a change from its diff or paths;
it provides structured context and validates the declaration an author chose.
`prepare` may narrow which definitions it *lists* using advisory `scope_hints`;
that is a display convenience which reports itself as such, and the author still
chooses the declaration.
All required validation is local and deterministic. `report` accepts an
explicit event fixture so advisory recurrence reports do not require network
access or mutate definition state.
"""
from __future__ import annotations
import argparse
import json
import re
import subprocess
import sys
from dataclasses import dataclass
from fnmatch import fnmatch
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any
OUTPUT_SCHEMA_VERSION = 1
DEFINITION_SCHEMA_VERSION = 1
DEFINITIONS_RELATIVE_PATH = Path(".github/failure-classes")
FAILURE_CLASS_ID_RE = re.compile(r"^FC-[a-z][a-z0-9]*(?:-[a-z0-9]+)*$")
FIX_SUBJECT_RE = re.compile(r"^fix(?:\([^)]+\))?!?:")
DECLARATION_LINE_RE = re.compile(r"^[ \t]*Failure-Class:[ \t]*([^\r\n]*)[ \t]*$", re.MULTILINE)
# The declaration value is ONE of these literal forms. The syntax used to be taught as
# "Failure-Class: FC-<slug> | new | none", which reads as a pipe-separated value in a
# repo whose other PR-body directives really are pipe-separated
# (Line-Count-Exception: path | 2157 -> 2181 | reason). Authors wrote
# "Failure-Class: FC-my-slug | new", and the error repeated the same ambiguous template
# back at them. Always present the alternatives as separate lines.
DECLARATION_FORMS = (
"Failure-Class: FC-<lower-kebab-slug>",
"Failure-Class: new",
"Failure-Class: none",
)
HTML_COMMENT_RE = re.compile(r"<!--.*?-->", re.DOTALL)
DURATION_RE = re.compile(r"^(\d+)([dh])$")
@dataclass(frozen=True)
class Definition:
"""A parsed failure-class definition and its repository-relative path."""
data: dict[str, Any]
path: Path
@property
def id(self) -> str:
return self.data["id"]
class CliError(Exception):
"""A deterministic input or repository error intended for CLI output."""
def error(code: str, message: str, **details: Any) -> dict[str, Any]:
return {"code": code, "message": message, **details}
def emit(payload: dict[str, Any], output_format: str) -> None:
if output_format == "json":
print(json.dumps(payload, indent=2, sort_keys=True))
return
if not payload.get("ok", True):
print("FAIL:")
for item in payload.get("errors", []):
print(f"- {item['code']}: {item['message']}")
return
command = payload.get("command", "failure-class")
print(f"OK: {command}")
if command == "validate":
validation = payload["validation"]
print(
" "
f"fix commits={validation['has_fix_commit']}; "
f"declaration={validation['declaration'] or 'absent'}"
)
elif command == "explain":
definition = payload["failure_class"]
print(f" {definition['id']}: {definition['violated_contract']}")
elif command == "prepare":
patch = payload["pr_body_patch"]
print(f" patch operation={patch['operation']}")
if patch["text"]:
print(patch["text"], end="" if patch["text"].endswith("\n") else "\n")
elif command == "report":
for item in payload["classes"]:
print(f" {item['id']}: closure_eligible={item['closure_eligible']}")
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
subparsers = parser.add_subparsers(dest="command", required=True)
def add_common_arguments(command: argparse.ArgumentParser) -> None:
command.add_argument("--root", type=Path, default=Path.cwd(), help="Repository root (default: cwd).")
command.add_argument("--format", choices=("json", "text"), default="json")
prepare = subparsers.add_parser("prepare", help="Return a non-destructive PR-body declaration patch.")
add_common_arguments(prepare)
prepare.add_argument("--base", default="origin/main")
prepare.add_argument("--head", default="HEAD")
prepare.add_argument("--pr-body-file", type=Path, required=True)
prepare.add_argument(
"--all-candidates",
action="store_true",
help="List every definition instead of those whose scope_hints overlap this change.",
)
explain = subparsers.add_parser("explain", help="Return one failure-class definition.")
add_common_arguments(explain)
explain.add_argument("failure_class_id")
validate = subparsers.add_parser("validate", help="Validate definitions and a PR-body declaration offline.")
add_common_arguments(validate)
validate.add_argument("--base", default="origin/main")
validate.add_argument("--head", default="HEAD")
validate.add_argument("--pr-body-file", type=Path, required=True)
report = subparsers.add_parser("report", help="Produce an advisory, non-mutating recurrence report.")
add_common_arguments(report)
report.add_argument("--since", default="14d", help="Quiet period, such as 14d or 24h (default: 14d).")
report.add_argument("--events-file", type=Path, help="Local merged-PR event fixture; no network is used.")
report.add_argument("--now", help="UTC ISO-8601 timestamp for deterministic advisory reports.")
return parser.parse_args()
def repository_root(root: Path) -> Path:
root = root.resolve()
if not root.is_dir():
raise CliError(f"repository root does not exist: {root}")
return root
def definitions_directory(root: Path) -> Path:
return root / DEFINITIONS_RELATIVE_PATH
def validate_canonical_prevention_artifact(value: Any, path: Path, root: Path) -> list[dict[str, Any]]:
"""Return errors for the optional guard-artifact field.
The field names the reusable guard surface that makes recurrence of this
class mechanically harder — a checker, a shared fixture, or a behavioral
contract test. Every listed path must exist so a rename cannot silently
leave a class claiming a guard it no longer has.
"""
if (
not isinstance(value, list)
or not value
or any(not isinstance(item, str) or not item.strip() for item in value)
):
return [
error(
"invalid_canonical_prevention_artifact",
"'canonical_prevention_artifact', when present, must be a non-empty array of repository-relative paths",
path=str(path),
)
]
errors: list[dict[str, Any]] = []
if len(set(value)) != len(value):
errors.append(
error(
"invalid_canonical_prevention_artifact",
"'canonical_prevention_artifact' must not contain duplicates",
path=str(path),
)
)
for item in value:
candidate = Path(item)
if candidate.is_absolute() or ".." in candidate.parts:
errors.append(
error(
"invalid_canonical_prevention_artifact",
f"'{item}' must be a repository-relative path without '..'",
path=str(path),
)
)
elif not (root / candidate).exists():
errors.append(
error(
"missing_canonical_prevention_artifact",
f"'canonical_prevention_artifact' path does not exist: {item}",
path=str(path),
)
)
return errors
def validate_definition(data: Any, path: Path, root: Path) -> list[dict[str, Any]]:
"""Return schema errors without allowing malformed files to fail open."""
if not isinstance(data, dict):
return [error("invalid_definition", "definition must be a JSON object", path=str(path))]
required = {
"schema_version",
"id",
"violated_contract",
"canonical_prevention",
"evidence_prs",
"status",
}
allowed = required | {"scope_hints", "dormant_since", "canonical_prevention_artifact"}
errors: list[dict[str, Any]] = []
for key in sorted(required - data.keys()):
errors.append(error("missing_definition_field", f"missing required field '{key}'", path=str(path)))
for key in sorted(data.keys() - allowed):
errors.append(error("unknown_definition_field", f"unknown field '{key}'", path=str(path)))
if data.get("schema_version") != DEFINITION_SCHEMA_VERSION:
errors.append(
error(
"unsupported_definition_schema",
f"schema_version must be {DEFINITION_SCHEMA_VERSION}",
path=str(path),
)
)
failure_class_id = data.get("id")
if not isinstance(failure_class_id, str) or not FAILURE_CLASS_ID_RE.fullmatch(failure_class_id):
errors.append(
error(
"invalid_failure_class_id",
"id must use the semantic form FC-<lower-kebab-slug>",
path=str(path),
)
)
for key in ("violated_contract", "canonical_prevention"):
if not isinstance(data.get(key), str) or not data[key].strip():
errors.append(error("invalid_definition_field", f"'{key}' must be a non-empty string", path=str(path)))
evidence_prs = data.get("evidence_prs")
# May be empty. `evidence_prs` records *merged* PRs, so a class born in the PR that
# fixes its first instance has none to cite: its number does not exist until the PR
# is opened. Requiring non-empty made `Failure-Class: new` unsatisfiable — the author
# had to invent a number or push with --no-verify and backfill. Empty is not missing
# data; the adding commit is the evidence, and `git log -- <definition>` recovers it.
# Nothing enforces recurrence from this field (the guard-artifact ratchet counts
# declarations in git history), so requiring it bought paperwork, not traceability.
if not isinstance(evidence_prs, list) or any(
not isinstance(pr, int) or isinstance(pr, bool) or pr <= 0 for pr in evidence_prs
):
errors.append(
error(
"invalid_evidence_prs",
"'evidence_prs' must be an array of positive PR numbers (may be empty for a "
"class this change introduces)",
path=str(path),
)
)
elif len(set(evidence_prs)) != len(evidence_prs):
errors.append(error("invalid_evidence_prs", "'evidence_prs' must not contain duplicates", path=str(path)))
scope_hints = data.get("scope_hints", [])
if not isinstance(scope_hints, list) or any(not isinstance(hint, str) or not hint.strip() for hint in scope_hints):
errors.append(
error(
"invalid_scope_hints",
"'scope_hints', when present, must be an array of non-empty strings",
path=str(path),
)
)
if "canonical_prevention_artifact" in data:
errors.extend(validate_canonical_prevention_artifact(data["canonical_prevention_artifact"], path, root))
status = data.get("status")
if status not in {"open", "dormant"}:
errors.append(error("invalid_status", "'status' must be 'open' or 'dormant'", path=str(path)))
dormant_since = data.get("dormant_since")
if status == "dormant":
if not isinstance(dormant_since, str):
errors.append(
error(
"missing_dormant_since",
"a dormant class must record an ISO-8601 'dormant_since' timestamp",
path=str(path),
)
)
else:
try:
parse_timestamp(dormant_since)
except CliError as exc:
errors.append(error("invalid_dormant_since", str(exc), path=str(path)))
elif dormant_since is not None:
errors.append(
error(
"unexpected_dormant_since",
"'dormant_since' is only valid while status is 'dormant'",
path=str(path),
)
)
if isinstance(failure_class_id, str) and path.name != f"{failure_class_id}.json":
errors.append(
error(
"definition_filename_mismatch",
f"definition filename must be {failure_class_id}.json",
path=str(path),
)
)
return errors
def load_definitions(root: Path) -> tuple[list[Definition], list[dict[str, Any]]]:
directory = definitions_directory(root)
if not directory.is_dir():
return [], [error("missing_definition_directory", f"missing {DEFINITIONS_RELATIVE_PATH}")]
definitions: list[Definition] = []
errors: list[dict[str, Any]] = []
ids: dict[str, Path] = {}
paths = sorted(directory.glob("*.json"))
if not paths:
errors.append(error("missing_definitions", "at least one failure-class definition is required"))
for path in paths:
try:
data = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
errors.append(error("invalid_definition_json", str(exc), path=str(path)))
continue
errors.extend(validate_definition(data, path, root))
if isinstance(data, dict) and isinstance(data.get("id"), str):
if data["id"] in ids:
errors.append(
error(
"duplicate_failure_class_id",
f"duplicate id '{data['id']}' in {path.name} and {ids[data['id']].name}",
)
)
else:
ids[data["id"]] = path
definitions.append(Definition(data=data, path=path))
return sorted(definitions, key=lambda item: item.id), errors
def clean_pr_body(body: str) -> str:
return HTML_COMMENT_RE.sub("", body)
def declarations_in_body(body: str) -> list[str]:
return [match.group(1).strip() for match in DECLARATION_LINE_RE.finditer(clean_pr_body(body))]
def read_pr_body(path: Path) -> str:
try:
return path.read_text(encoding="utf-8")
except OSError as exc:
raise CliError(f"could not read PR body file {path}: {exc}") from exc
def run_git(root: Path, *args: str) -> str:
result = subprocess.run(
["git", *args],
cwd=root,
check=False,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
)
if result.returncode:
detail = result.stderr.strip() or result.stdout.strip()
raise CliError(f"git {' '.join(args)} failed: {detail}")
return result.stdout.strip()
def merge_base(root: Path, base: str, head: str) -> str:
return run_git(root, "merge-base", base, head)
def commit_subjects(root: Path, base: str, head: str) -> tuple[str, list[str]]:
common = merge_base(root, base, head)
subjects = run_git(root, "log", "--format=%s", f"{common}..{head}")
return common, [subject for subject in subjects.splitlines() if subject]
def changed_paths_in_range(root: Path, common: str, head: str) -> list[str]:
output = run_git(root, "diff", "--name-only", common, head)
return [line for line in output.splitlines() if line]
def candidates_matching_scope(definitions: list[Definition], paths: list[str]) -> list[Definition]:
"""Definitions whose advisory `scope_hints` overlap the change's paths.
This narrows what is *listed*; it does not choose. `scope_hints` stay advisory — a
hit is not a classification, and the author still declares manually. The narrowing
exists because `prepare` printed all definitions in full, and an unreadable registry
gets a new class invented rather than an existing one reused, inflating the very
registry this protocol wants to keep small.
"""
matched: list[Definition] = []
for definition in definitions:
hints = definition.data.get("scope_hints") or []
if any(fnmatch(path, hint) for hint in hints for path in paths):
matched.append(definition)
return matched
def added_definition_paths(root: Path, common: str, head: str) -> list[Path]:
output = run_git(
root,
"diff",
"--name-only",
"--diff-filter=A",
common,
head,
"--",
str(DEFINITIONS_RELATIVE_PATH),
)
return [Path(line) for line in output.splitlines() if line]
def changed_definition_paths(root: Path, common: str, head: str) -> list[Path]:
output = run_git(
root,
"diff",
"--name-only",
"--diff-filter=ACMRD",
common,
head,
"--",
str(DEFINITIONS_RELATIVE_PATH),
)
return [Path(line) for line in output.splitlines() if line]
def definition_json_at_ref(root: Path, ref: str, relative: Path) -> dict[str, Any] | None:
try:
output = run_git(root, "show", f"{ref}:{relative.as_posix()}")
except CliError:
return None
try:
data = json.loads(output)
except json.JSONDecodeError:
return None
return data if isinstance(data, dict) else None
def is_artifact_only_definition_update(root: Path, common: str, relative: Path) -> bool:
"""True when the only JSON delta is adding or updating canonical_prevention_artifact."""
head_path = root / relative
try:
head_data = json.loads(head_path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
return False
if not isinstance(head_data, dict) or "canonical_prevention_artifact" not in head_data:
return False
base_data = definition_json_at_ref(root, common, relative)
if base_data is None:
return False
if base_data.get("canonical_prevention_artifact") == head_data.get("canonical_prevention_artifact"):
return False
for key in set(base_data) | set(head_data):
if key == "canonical_prevention_artifact":
continue
if base_data.get(key) != head_data.get(key):
return False
return True
def instance_fix_registry_changes_allowed(
root: Path,
common: str,
declaration: str,
changed_paths: list[Path],
) -> bool:
"""Instance-fix PRs may record a guard artifact for the declared class only."""
if not changed_paths:
return True
expected = DEFINITIONS_RELATIVE_PATH / f"{declaration}.json"
if set(changed_paths) != {expected}:
return False
return is_artifact_only_definition_update(root, common, expected)
def public_definition(definition: Definition, root: Path) -> dict[str, Any]:
return {**definition.data, "path": str(definition.path.relative_to(root))}
def validate_command(args: argparse.Namespace, root: Path) -> tuple[dict[str, Any], int]:
definitions, errors = load_definitions(root)
body = read_pr_body(args.pr_body_file)
try:
common, subjects = commit_subjects(root, args.base, args.head)
except CliError as exc:
errors.append(error("invalid_git_range", str(exc), base=args.base, head=args.head))
common, subjects = "", []
declarations = declarations_in_body(body)
declaration: str | None = None
if len(declarations) > 1:
errors.append(
error(
"multiple_declarations",
"PR body may contain at most one Failure-Class declaration outside HTML comments",
)
)
elif declarations:
declaration = declarations[0]
has_fix_commit = any(FIX_SUBJECT_RE.match(subject) for subject in subjects)
by_id = {definition.id: definition for definition in definitions}
changed_definition_paths_in_range: list[Path] = []
if common:
try:
changed_definition_paths_in_range = changed_definition_paths(root, common, args.head)
except CliError as exc:
errors.append(error("definition_change_check_failed", str(exc)))
if declaration is None:
if has_fix_commit:
errors.append(
error(
"missing_declaration",
"a commit subject beginning with fix: requires a Failure-Class declaration "
"in the PR body; write exactly one of the accepted forms",
accepted_forms=list(DECLARATION_FORMS),
)
)
elif declaration == "new":
if common:
try:
added_paths = added_definition_paths(root, common, args.head)
except CliError as exc:
errors.append(error("new_definition_check_failed", str(exc)))
else:
added_definitions = [
definition
for definition in definitions
if definition.path.relative_to(root) in added_paths
]
if len(added_definitions) != 1:
errors.append(
error(
"new_definition_required",
"'Failure-Class: new' requires exactly one added, valid class definition "
"in this commit range; add "
".github/failure-classes/FC-<lower-kebab-slug>.json alongside the fix "
"(its 'evidence_prs' may be [] — this PR is the evidence)",
added_paths=[str(path) for path in added_paths],
)
)
elif set(changed_definition_paths_in_range) != set(added_paths):
errors.append(
error(
"new_definition_must_be_only_registry_change",
"'Failure-Class: new' may add one definition but must not modify or remove other definitions",
changed_paths=[str(path) for path in changed_definition_paths_in_range],
)
)
elif declaration == "none":
pass
elif not FAILURE_CLASS_ID_RE.fullmatch(declaration):
# "FC-my-slug | new" is the common miss, because the old template joined its
# alternatives with a character this repo uses as a real field separator
# elsewhere. Name that mistake instead of echoing the template again.
hint = (
"write one form only — '|' separated the alternatives in the old template, "
"it is not part of the value"
if "|" in declaration
else "the value is a single token"
)
errors.append(
error(
"invalid_declaration",
f"'{declaration}' is not a valid Failure-Class declaration; {hint}",
declaration=declaration,
accepted_forms=list(DECLARATION_FORMS),
)
)
elif declaration not in by_id:
errors.append(
error(
"unknown_failure_class",
f"no definition exists for '{declaration}'",
declaration=declaration,
)
)
elif by_id[declaration].data["status"] == "dormant":
errors.append(
error(
"dormant_failure_class_requires_reopen",
f"'{declaration}' is dormant; explicitly reopen its definition before classifying a new instance",
declaration=declaration,
)
)
if (
has_fix_commit
and declaration not in (None, "new", "none")
and declaration in by_id
and changed_definition_paths_in_range
and not instance_fix_registry_changes_allowed(
root, common, declaration, changed_definition_paths_in_range
)
):
errors.append(
error(
"instance_fix_mutates_registry",
"an instance-fix PR must not edit failure-class definitions except to add or update "
"'canonical_prevention_artifact' on the declared class; other registry edits need a "
"separate registry-only lifecycle PR",
changed_paths=[str(path) for path in changed_definition_paths_in_range],
)
)
payload = {
"schema_version": OUTPUT_SCHEMA_VERSION,
"command": "validate",
"ok": not errors,
"errors": errors,
"validation": {
"base": args.base,
"head": args.head,
"merge_base": common or None,
"commit_subjects": subjects,
"has_fix_commit": has_fix_commit,
"declaration": declaration,
"definition_count": len(definitions),
},
}
return payload, 0 if not errors else 1
def pr_body_patch(body: str, declarations: list[str]) -> dict[str, str]:
if declarations:
return {"operation": "none", "text": "", "resulting_pr_body": body}
prefix = "" if not body or body.endswith("\n") else "\n"
text = f"{prefix}Failure-Class: none\n"
return {"operation": "append", "text": text, "resulting_pr_body": body + text}
def prepare_command(args: argparse.Namespace, root: Path) -> tuple[dict[str, Any], int]:
definitions, errors = load_definitions(root)
body = read_pr_body(args.pr_body_file)
try:
common, subjects = commit_subjects(root, args.base, args.head)
except CliError as exc:
errors.append(error("invalid_git_range", str(exc), base=args.base, head=args.head))
common, subjects = "", []
has_fix_commit = any(FIX_SUBJECT_RE.match(subject) for subject in subjects)
declarations = declarations_in_body(body)
shown = definitions
narrowing = "none: every definition is listed"
if not args.all_candidates and common:
try:
paths = changed_paths_in_range(root, common, args.head)
except CliError as exc:
errors.append(error("changed_path_check_failed", str(exc)))
else:
matched = candidates_matching_scope(definitions, paths)
# Never narrow to nothing: an empty list reads as "no class can apply",
# which is a classification this CLI does not make.
if matched:
shown = matched
narrowing = (
"scope_hints overlapping this change's paths; advisory only, not a "
"classification. Pass --all-candidates for the full registry."
)
payload = {
"schema_version": OUTPUT_SCHEMA_VERSION,
"command": "prepare",
"ok": not errors,
"errors": errors,
"requires_declaration": has_fix_commit,
"commit_subjects": subjects,
"merge_base": common or None,
"declaration_template": list(DECLARATION_FORMS),
"pr_body_patch": pr_body_patch(body, declarations),
"advisory_candidates": [
{
"id": definition.id,
"status": definition.data["status"],
"violated_contract": definition.data["violated_contract"],
"canonical_prevention": definition.data["canonical_prevention"],
}
for definition in shown
],
"candidates_shown": len(shown),
"candidates_total": len(definitions),
"candidate_narrowing": narrowing,
"candidate_source": "registry-only; no class was inferred from paths, diffs, or commit text",
"next_action": (
"Choose the declaration manually; replace 'none' in an appended patch if a class "
"applies. Declaring 'new' means adding one "
".github/failure-classes/FC-<lower-kebab-slug>.json in this same change; its "
"'evidence_prs' may be [] because this PR is the evidence."
),
}
return payload, 0 if not errors else 1
def explain_command(args: argparse.Namespace, root: Path) -> tuple[dict[str, Any], int]:
definitions, errors = load_definitions(root)
matches = [definition for definition in definitions if definition.id == args.failure_class_id]
if not errors and not matches:
errors.append(error("unknown_failure_class", f"no definition exists for '{args.failure_class_id}'"))
payload = {
"schema_version": OUTPUT_SCHEMA_VERSION,
"command": "explain",
"ok": not errors,
"errors": errors,
}
if matches:
payload["failure_class"] = public_definition(matches[0], root)
return payload, 0 if not errors else 1
def parse_duration(value: str) -> timedelta:
match = DURATION_RE.fullmatch(value)
if not match:
raise CliError("--since must be a positive duration such as 14d or 24h")
count = int(match.group(1))
if count <= 0:
raise CliError("--since must be greater than zero")
return timedelta(days=count) if match.group(2) == "d" else timedelta(hours=count)
def parse_timestamp(value: str) -> datetime:
try:
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
except ValueError as exc:
raise CliError(f"invalid ISO-8601 timestamp: {value}") from exc
if parsed.tzinfo is None:
raise CliError(f"timestamp must include a timezone: {value}")
return parsed.astimezone(timezone.utc)
def timestamp_string(value: datetime) -> str:
return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
def load_report_events(path: Path | None) -> tuple[list[dict[str, Any]], dict[str, Any], list[dict[str, Any]]]:
if path is None:
return [], {"type": "none"}, [
error(
"no_event_source",
"no event source was supplied; report cannot establish a last reported instance",
)
]
try:
raw = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
raise CliError(f"could not load events fixture {path}: {exc}") from exc
if not isinstance(raw, dict) or raw.get("schema_version") != OUTPUT_SCHEMA_VERSION:
raise CliError(f"events fixture must be a schema_version {OUTPUT_SCHEMA_VERSION} JSON object")
events = raw.get("events")
if not isinstance(events, list):
raise CliError("events fixture must be an array or an object with an 'events' array")
parsed: list[dict[str, Any]] = []
for index, event in enumerate(events):
if not isinstance(event, dict):
raise CliError(f"events[{index}] must be an object")
body = event.get("body")
merged_at = event.get("merged_at")
if not isinstance(body, str) or not isinstance(merged_at, str):
raise CliError(f"events[{index}] requires string fields 'body' and 'merged_at'")
parsed.append(
{
"number": event.get("number"),
"body": body,
"merged_at": parse_timestamp(merged_at),
}
)
return parsed, {"type": "events_fixture", "path": str(path)}, []
def report_command(args: argparse.Namespace, root: Path) -> tuple[dict[str, Any], int]:
definitions, errors = load_definitions(root)
try:
quiet_period = parse_duration(args.since)
as_of = parse_timestamp(args.now) if args.now else datetime.now(timezone.utc)
events, source, warnings = load_report_events(args.events_file)
except CliError as exc:
errors.append(error("invalid_report_input", str(exc)))
quiet_period = timedelta(days=14)
as_of = datetime.now(timezone.utc)
events, source, warnings = [], {"type": "none"}, []
by_id = {definition.id: definition for definition in definitions}
latest: dict[str, dict[str, Any]] = {}
for event in events:
declarations = declarations_in_body(event["body"])
if len(declarations) != 1 or declarations[0] not in by_id:
continue
class_id = declarations[0]
if class_id not in latest or event["merged_at"] > latest[class_id]["merged_at"]:
latest[class_id] = event
cutoff = as_of - quiet_period
classes: list[dict[str, Any]] = []
for definition in definitions:
instance = latest.get(definition.id)
reopen_required = False
if definition.data["status"] == "dormant" and instance is not None:
dormant_since = parse_timestamp(definition.data["dormant_since"])
reopen_required = instance["merged_at"] > dormant_since
closure_eligible = bool(
definition.data["status"] == "open" and instance is not None and instance["merged_at"] <= cutoff
)
if reopen_required:
reason = "a classified recurrence was reported after this class became dormant; explicit reopen required"
elif definition.data["status"] == "dormant":
reason = "already dormant; report never changes state automatically"
elif instance is None:
reason = "no classified instance in the supplied event source"
elif closure_eligible:
reason = "no classified recurrence was reported during the quiet period; maintainer confirmation required"
else:
reason = "a classified instance falls inside the quiet period"
classes.append(
{
"id": definition.id,
"status": definition.data["status"],
"last_reported_instance": (
{"number": instance["number"], "merged_at": timestamp_string(instance["merged_at"])} if instance else None
),
"closure_eligible": closure_eligible,
"reopen_required": reopen_required,
"reason": reason,
}
)
payload = {
"schema_version": OUTPUT_SCHEMA_VERSION,
"command": "report",
"ok": not errors,
"errors": errors,
"advisory": True,
"automatic_state_changes": False,
"as_of": timestamp_string(as_of),
"since": args.since,
"source": source,
"warnings": warnings,
"events_considered": len(events),
"classes": classes,
}
return payload, 0 if not errors else 1
def main() -> int:
args = parse_args()
try:
root = repository_root(args.root)
if args.command == "validate":
payload, exit_code = validate_command(args, root)
elif args.command == "prepare":
payload, exit_code = prepare_command(args, root)
elif args.command == "explain":
payload, exit_code = explain_command(args, root)
elif args.command == "report":
payload, exit_code = report_command(args, root)
else: # argparse makes this unreachable; preserve fail-closed behavior.
raise CliError(f"unsupported command: {args.command}")
except CliError as exc:
payload = {
"schema_version": OUTPUT_SCHEMA_VERSION,
"command": args.command,
"ok": False,
"errors": [error("invalid_input", str(exc))],
}
exit_code = 2
emit(payload, args.format)
return exit_code
if __name__ == "__main__":
raise SystemExit(main())