File size: 5,896 Bytes
f2cf4fa | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 | #!/usr/bin/env python3
"""Local consistency checks for the Spine V1 scaffold.
This script does not implement full JSON Schema validation. It validates the
spec bundle at the repository-contract level using only the Python stdlib.
"""
from __future__ import annotations
import json
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
SCHEMAS = ROOT / "schemas"
REGISTRY = ROOT / "registry"
EXAMPLES = ROOT / "examples" / "run_lifecycle"
REQUIRED_SCHEMAS = {
"OperatorContext": "operator_context.schema.json",
"TaskRequest": "task_request.schema.json",
"RoutingDecision": "routing_decision.schema.json",
"RunRecord": "run_record.schema.json",
"ExecutionRequest": "execution_request.schema.json",
"ExecutionStatus": "execution_status.schema.json",
"RouteTrace": "route_trace.schema.json",
"ArtifactManifest": "artifact_manifest.schema.json",
"EvalRecord": "eval_record.schema.json",
"SpineEvent": "spine_event.schema.json",
"SystemManifest": "system_manifest.schema.json",
"HarnessManifest": "harness_manifest.schema.json",
"AdapterManifest": "adapter_manifest.schema.json",
}
def load_json(path: Path):
with path.open("r", encoding="utf-8") as handle:
return json.load(handle)
def require(condition: bool, message: str) -> None:
if not condition:
raise SystemExit(f"VALIDATION FAILED: {message}")
def check_schema_inventory() -> None:
for schema_name, filename in REQUIRED_SCHEMAS.items():
path = SCHEMAS / filename
require(path.exists(), f"missing schema file: {filename}")
data = load_json(path)
require(data["title"] == schema_name, f"{filename} has unexpected title")
def check_registry() -> None:
system_manifest = load_json(REGISTRY / "system_manifest.json")
require(system_manifest["schema_name"] == "SystemManifest", "system manifest schema name mismatch")
require(
sorted(system_manifest["schema_inventory"]) == sorted(REQUIRED_SCHEMAS.keys()),
"system manifest schema inventory is incomplete",
)
harness_ids = set()
for path in sorted((REGISTRY / "harnesses").glob("*.json")):
data = load_json(path)
require(data["schema_name"] == "HarnessManifest", f"{path.name} is not a harness manifest")
require(data["harness_id"] not in harness_ids, f"duplicate harness id: {data['harness_id']}")
harness_ids.add(data["harness_id"])
adapter_ids = set()
for path in sorted((REGISTRY / "adapters").glob("*.json")):
data = load_json(path)
require(data["schema_name"] == "AdapterManifest", f"{path.name} is not an adapter manifest")
require(data["adapter_id"] not in adapter_ids, f"duplicate adapter id: {data['adapter_id']}")
adapter_ids.add(data["adapter_id"])
def check_examples() -> None:
task = load_json(EXAMPLES / "task_request.json")
decision = load_json(EXAMPLES / "routing_decision.json")
run = load_json(EXAMPLES / "run_record.json")
execution = load_json(EXAMPLES / "execution_request.json")
status = load_json(EXAMPLES / "execution_status.json")
trace = load_json(EXAMPLES / "route_trace.json")
artifact = load_json(EXAMPLES / "artifact_manifest.json")
eval_record = load_json(EXAMPLES / "eval_record.json")
events = load_json(EXAMPLES / "events.json")
require(task["schema_name"] == "TaskRequest", "task example schema mismatch")
require(decision["task_id"] == task["task_id"], "decision does not reference task")
require(run["task_id"] == task["task_id"], "run does not reference task")
require(execution["decision_id"] == decision["decision_id"], "execution does not reference decision")
require(execution["run_id"] == run["run_id"], "execution does not reference run")
require(status["execution_id"] == execution["execution_id"], "status does not reference execution")
require(status["run_id"] == run["run_id"], "status does not reference run")
require(trace["execution_id"] == execution["execution_id"], "trace does not reference execution")
require(artifact["run_id"] == run["run_id"], "artifact does not reference run")
require(eval_record["run_id"] == run["run_id"], "eval does not reference run")
require(eval_record["task_id"] == task["task_id"], "eval does not reference task")
require(run["decision_ids"] == [decision["decision_id"]], "run decision ids mismatch")
require(run["execution_ids"] == [execution["execution_id"]], "run execution ids mismatch")
require(run["artifact_ids"] == [artifact["artifact_id"]], "run artifact ids mismatch")
require(run["eval_ids"] == [eval_record["eval_id"]], "run eval ids mismatch")
sequences = [event["sequence"] for event in events]
require(sequences == list(range(1, len(events) + 1)), "event sequence must be contiguous starting at 1")
for event in events:
require(event["schema_name"] == "SpineEvent", f"event {event['event_id']} schema mismatch")
require(event["run_id"] == run["run_id"], f"event {event['event_id']} run id mismatch")
event_ids = [event["event_id"] for event in events]
require(run["event_ids"] == event_ids, "run event ids do not match event log")
def main() -> None:
check_schema_inventory()
check_registry()
check_examples()
print("Spine validation OK")
print(f"Schemas: {len(REQUIRED_SCHEMAS)}")
print(f"Harness manifests: {len(list((REGISTRY / 'harnesses').glob('*.json')))}")
print(f"Adapter manifests: {len(list((REGISTRY / 'adapters').glob('*.json')))}")
print(f"Example files: {len(list(EXAMPLES.glob('*.json')))}")
if __name__ == "__main__":
try:
main()
except SystemExit:
raise
except Exception as exc: # pragma: no cover - defensive catch for local use
raise SystemExit(f"VALIDATION FAILED: unexpected error: {exc}") from exc
|