Spaces:
Sleeping
Sleeping
File size: 5,499 Bytes
41016fc | 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 | from __future__ import annotations
import json
from pathlib import Path
from typing import Any
from jsonschema import Draft202012Validator, FormatChecker
from referencing import Registry, Resource
from .claim_ladder import LEVEL_ORDER
from .consent_gate import comparison_gate, public_exhibit_gate
from .temporal_ratchet import validate_temporal_ratchet
SCHEMA_FILES = {
"creator_trace_capsule": "creator_trace_capsule_schema_v0_1_0.json",
"temporal_event": "temporal_event_schema_v0_1_0.json",
"evidence_asset_manifest": "evidence_asset_manifest_schema_v0_1_0.json",
"consent_receipt": "consent_receipt_schema_v0_1_0.json",
"causal_arc_packet": "causal_arc_packet_schema_v0_1_0.json",
"cross_account_comparison_node": "cross_account_comparison_node_schema_v0_1_0.json",
"metric_receipt": "metric_receipt_schema_v0_2_0.json",
"holographic_comparison": "holographic_comparison_schema_v0_2_0.json",
}
def _load_schema(schema_dir: Path, filename: str) -> dict[str, Any]:
return json.loads((schema_dir / filename).read_text(encoding="utf-8"))
def build_registry(schema_dir: str | Path) -> tuple[Registry, dict[str, dict[str, Any]]]:
root = Path(schema_dir)
schemas = {key: _load_schema(root, filename) for key, filename in SCHEMA_FILES.items()}
registry = Registry()
# Register each schema under its literal filename, declared $id, and the
# absolute URI produced when the creator capsule resolves local references.
for key, filename in SCHEMA_FILES.items():
schema = schemas[key]
resource = Resource.from_contents(schema)
uris = {
filename,
schema.get("$id", filename),
f"https://primordial.example/schema/{filename}",
}
for uri in uris:
registry = registry.with_resource(uri, resource)
return registry, schemas
def schema_errors(instance: Any, schema_name: str, schema_dir: str | Path) -> list[str]:
registry, schemas = build_registry(schema_dir)
validator = Draft202012Validator(schemas[schema_name], registry=registry, format_checker=FormatChecker())
errors = sorted(validator.iter_errors(instance), key=lambda e: list(e.absolute_path))
rendered: list[str] = []
for error in errors:
path = "/".join(str(p) for p in error.absolute_path) or "<root>"
rendered.append(f"{path}: {error.message}")
return rendered
def semantic_capsule_errors(capsule: dict[str, Any]) -> list[str]:
errors: list[str] = []
events = capsule.get("temporal_events", [])
errors.extend(validate_temporal_ratchet(events))
content_topology = capsule.get("content_topology", {})
if content_topology.get("system_inferred_labels") not in ([], None):
errors.append("system_inferred_labels must remain empty.")
asset_ids = {a.get("asset_id") for a in capsule.get("evidence_assets", [])}
event_ids = {e.get("event_id") for e in events}
for asset in capsule.get("evidence_assets", []):
if asset.get("execution_blocked") is not True:
errors.append(f"{asset.get('asset_id')}: execution_blocked must be true.")
for linked in asset.get("linked_event_ids", []) or []:
if linked not in event_ids:
errors.append(f"{asset.get('asset_id')}: linked event {linked!r} does not exist.")
for claim in capsule.get("claims", []):
rank = LEVEL_ORDER.get(claim.get("claim_level", ""), 99)
if rank >= 5 and claim.get("state") != "BLOCKED":
errors.append(f"{claim.get('claim_id')}: L5/L6 claims must be BLOCKED in this prototype.")
if rank >= 2 and claim.get("human_review_required") is not True:
errors.append(f"{claim.get('claim_id')}: L2-L6 claims require human_review_required=true.")
for basis in claim.get("basis", []) or []:
if basis.startswith("ASSET_") and basis not in asset_ids:
errors.append(f"{claim.get('claim_id')}: claim basis {basis!r} is not a known asset.")
review_pack = capsule.get("review_pack", {})
blockers = review_pack.get("closure_blockers", []) or []
if capsule.get("loop_state") == "CLOSED_FOR_CURRENT_SCOPE" and blockers:
errors.append("False closure: CLOSED_FOR_CURRENT_SCOPE cannot coexist with closure blockers.")
consent = capsule.get("consent", {})
if consent.get("publication_requires_additional_review") is not True:
errors.append("publication_requires_additional_review must be true.")
return errors
def validate_capsule(capsule: dict[str, Any], schema_dir: str | Path) -> list[str]:
return schema_errors(capsule, "creator_trace_capsule", schema_dir) + semantic_capsule_errors(capsule)
def validate_causal_arc(packet: dict[str, Any], schema_dir: str | Path) -> list[str]:
errors = schema_errors(packet, "causal_arc_packet", schema_dir)
if packet.get("human_review_required") is not True:
errors.append("human_review_required must be true.")
return errors
def gate_receipt(consent_state: str, scope: str) -> dict[str, str | bool]:
comparison_ok, comparison_reason = comparison_gate(consent_state, scope, human_review_approved=False)
public_ok, public_reason = public_exhibit_gate(consent_state, scope, publication_review_approved=False)
return {
"comparison_allowed_now": comparison_ok,
"comparison_state": comparison_reason,
"public_exhibit_allowed_now": public_ok,
"public_exhibit_state": public_reason,
}
|