From d3c677fa374fd18615b64fbb4dfc315b55bcf616 Mon Sep 17 00:00:00 2001 From: mdheller <21163552+mdheller@users.noreply.github.com> Date: Sun, 19 Apr 2026 12:33:50 -0400 Subject: [PATCH 1/2] ci(validate): add control-plane wrapper $id resolution test --- scripts/validate_control_plane_wrapper_ids.py | 102 ++++++++++++++++++ 1 file changed, 102 insertions(+) create mode 100644 scripts/validate_control_plane_wrapper_ids.py diff --git a/scripts/validate_control_plane_wrapper_ids.py b/scripts/validate_control_plane_wrapper_ids.py new file mode 100644 index 0000000..68f77fb --- /dev/null +++ b/scripts/validate_control_plane_wrapper_ids.py @@ -0,0 +1,102 @@ +#!/usr/bin/env python3 +"""Validate that control-plane canonical wrapper `$id` values resolve correctly. + +This is a portability guardrail: + +- Control-plane wrapper schemas live in `schemas/control-plane/*.json` and have + canonical `$id` values under `https://schemas.srcos.ai/v2/control-plane/...`. +- The wrappers `allOf`-wrap legacy `*.schema.json` files. + +We validate that a minimal instance can be validated against a schema that `$ref`s +one of the canonical wrapper `$id` values. + +Exit: +- 0 on success +- 1 on failure +""" + +from __future__ import annotations + +import glob +import json +import os +import sys +from pathlib import Path + + +def load_json(path: str) -> dict: + with open(path, encoding="utf-8") as f: + return json.load(f) + + +def main() -> int: + repo = Path(__file__).resolve().parents[1] + schema_dir = repo / "schemas" + cp_dir = schema_dir / "control-plane" + + registry: dict[str, dict] = {} + + # Load top-level schemas + for p in glob.glob(str(schema_dir / "*.json")): + s = load_json(p) + sid = s.get("$id") + if sid: + registry[sid] = s + base = os.path.basename(p) + registry[base] = s + registry[f"./{base}"] = s + + # Load control-plane schemas (wrappers + legacy) + for p in glob.glob(str(cp_dir / "*.json")) + glob.glob(str(cp_dir / "*.schema.json")): + s = load_json(p) + sid = s.get("$id") + if sid: + registry[sid] = s + base = os.path.basename(p) + registry[base] = s + registry[f"./{base}"] = s + + from jsonschema import RefResolver, validate + + base_uri = f"file://{schema_dir.resolve()}/" + + class LocalRegistry(RefResolver): + def resolve_remote(self, uri): + clean = uri.split("#")[0] + if clean in registry: + return registry[clean] + name = os.path.basename(clean) + if name in registry: + return registry[name] + return super().resolve_remote(uri) + + resolver = LocalRegistry(base_uri=base_uri, referrer={}, store=registry) + + # Minimal canonical wrapper `$id` resolution test: IncidentEvent + wrapper_id = "https://schemas.srcos.ai/v2/control-plane/IncidentEvent.json" + + schema_ref = { + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$ref": wrapper_id, + } + + instance = { + "event_id": "evt_ci_0001", + "event_name": "incident.freeze", + "occurred_at": "2026-04-19T00:00:00Z", + "actor": {"kind": "service", "id": "ci"}, + "status": "succeeded", + } + + try: + validate(instance, schema_ref, resolver=resolver) + except Exception as e: + print(f"FAIL: canonical wrapper $id resolution failed: {e}", file=sys.stderr) + return 1 + + print("OK: control-plane canonical wrapper $id resolution") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) From 33bec8f29690ffc215de7251e30fac3150eb980d Mon Sep 17 00:00:00 2001 From: mdheller <21163552+mdheller@users.noreply.github.com> Date: Sun, 19 Apr 2026 12:35:01 -0400 Subject: [PATCH 2/2] ci(validate): preload control-plane schemas and test wrapper $id resolution --- .github/workflows/validate.yml | 21 ++++++++++++++------- 1 file changed, 14 insertions(+), 7 deletions(-) diff --git a/.github/workflows/validate.yml b/.github/workflows/validate.yml index 7713bf7..849d2ff 100644 --- a/.github/workflows/validate.yml +++ b/.github/workflows/validate.yml @@ -46,6 +46,10 @@ jobs: run: | python3 scripts/check_duplicate_schema_ids.py + - name: Guardrail: control-plane canonical wrapper $id resolution + run: | + python3 scripts/validate_control_plane_wrapper_ids.py + - name: Validate examples against schemas run: | python3 - << 'EOF' @@ -54,17 +58,24 @@ jobs: # Pre-load all schemas into a registry keyed by both $id and local path schema_dir = 'schemas' registry = {} - for schema_file in glob.glob(f'{schema_dir}/*.json'): + + def add_schema(schema_file: str): with open(schema_file) as f: schema = json.load(f) - # Register by $id if '$id' in schema: registry[schema['$id']] = schema - # Register by local relative path (used in $ref) base = os.path.basename(schema_file) registry[f'./{base}'] = schema registry[base] = schema + # Top-level schemas + for schema_file in glob.glob(f'{schema_dir}/*.json'): + add_schema(schema_file) + + # Control-plane wrappers + legacy schemas + for schema_file in glob.glob(f'{schema_dir}/control-plane/*.json') + glob.glob(f'{schema_dir}/control-plane/*.schema.json'): + add_schema(schema_file) + from jsonschema import RefResolver, validate, ValidationError failed = 0 @@ -95,16 +106,12 @@ jobs: class LocalRegistry(RefResolver): def resolve_remote(self, uri): - # Strip fragment clean = uri.split('#')[0] - # Try registry first if clean in registry: return registry[clean] - # Try by basename name = os.path.basename(clean) if name in registry: return registry[name] - # Fall back to file:// resolution return super().resolve_remote(uri) resolver = LocalRegistry(base_uri=base_uri, referrer=schema, store=registry)