#!/usr/bin/env python3
"""Checks the HEART 1.0 spec artefacts against each other.

  1. every schema is valid JSON Schema 2020-12
  2. every example validates against its schema
  3. privacy lint: a required property may not be more sensitive than its
     parent, so a redacted view always stays schema-valid
  4. Valve test: filtering each message example down to each clearance gives a
     valid message; the public Pulse equals examples/pulse-public.json
  5. deliberately broken messages are rejected
  6. the operation table in the standard matches the OpenAPI file
  7. ports agree across the standard, OpenAPI and examples
  8. custom definitions are valid and injection-safe: every placeholder is a
     declared input, and string inputs are limited by enum or pattern
  9. every message type has an example, heartprivacy is honest, and every
     state-changing module operation declares a risk and a known scope
  10. the API Index matches the OpenAPI file and test cases resolve
  11. fault-codes/1.0/fault-codes.json equals Part 3's fault-code tables:
      the same codes, H-numbers, areas and meanings, with no duplicates,
      and no event type among them

Usage: python validate.py   (needs: jsonschema[format] referencing)
"""
import copy, json, pathlib, re, sys
from urllib.parse import urljoin
import yaml
from jsonschema import Draft202012Validator, FormatChecker
from referencing import Registry, Resource

ROOT = pathlib.Path(__file__).resolve().parent.parent
SCHEMAS = ROOT / "schemas" / "1.0"
EXAMPLES = ROOT / "examples"
RANK = {"public": 0, "internal": 1, "private": 2, "never-share": 3}
BASE = "https://heartcortex.ai/schemas/1.0/"

docs = {}
for f in sorted(SCHEMAS.glob("*.json")):
    d = json.loads(f.read_text())
    docs[d["$id"]] = d
registry = Registry().with_resources(
    [(k, Resource.from_contents(v)) for k, v in docs.items()])
failures = []


def check(ok, label):
    if not ok:
        failures.append(label)
        print("FAIL", label)


def validator(name):
    return Draft202012Validator(docs[BASE + name], registry=registry,
                                format_checker=FormatChecker())


def errors(name, obj):
    return [f"{'/'.join(map(str, e.path))}: {e.message}"
            for e in validator(name).iter_errors(obj)]


def resolve(schema, base):
    """Follow a $ref, keeping sibling keywords. Returns (schema, base)."""
    while "$ref" in schema:
        url, _, frag = urljoin(base, schema["$ref"]).partition("#")
        target = docs[url]
        for part in filter(None, frag.split("/")):
            target = target[part]
        schema = {**target, **{k: v for k, v in schema.items() if k != "$ref"}}
        base = url
    return schema, base


def level_of(schema, inherited):
    """Annotations may raise the inherited level, never lower it."""
    own = schema.get("x-heart-privacy")
    if own is None:
        return inherited
    if own == "dynamic":      # resolved per value; worst case for the lint
        own = "private"
    return own if RANK[own] >= RANK[inherited] else inherited


def lint(schema, base, parent, path, seen=()):
    schema, base = resolve(schema, base)
    for key in schema.get("required", []):
        sub = schema.get("properties", {}).get(key)
        if sub is None:
            check(False, f"lint {path}: required '{key}' has no definition")
            continue
        sub_resolved, _ = resolve(sub, base)
        lvl = level_of(sub_resolved, parent)
        check(RANK[lvl] <= RANK[parent],
              f"lint {path}/{key}: required but {lvl}, parent is {parent}")
    for key, sub in schema.get("properties", {}).items():
        resolved, sub_base = resolve(sub, base)
        lint(resolved, sub_base, level_of(resolved, parent), f"{path}/{key}")
    if isinstance(schema.get("items"), dict):
        resolved, sub_base = resolve(schema["items"], base)
        lint(resolved, sub_base, parent, f"{path}[]")


DROP = object()


def valve(value, schema, base, inherited, clearance):
    """Reference Valve: returns the filtered value, or DROP."""
    schema, base = resolve(schema, base)
    level = level_of(schema, inherited)
    if RANK[level] > RANK[clearance]:
        return DROP
    if isinstance(value, dict) and "properties" in schema:
        out = {}
        for key, item in value.items():
            # A field the schema does not classify fails closed to private.
            sub = schema["properties"].get(key, {"x-heart-privacy": "private"})
            if sub.get("x-heart-privacy") == "dynamic":
                sub = {"x-heart-privacy": value.get("privacy", "private")}
            kept = valve(item, sub, base, level, clearance)
            if kept is not DROP:
                out[key] = kept
        return out
    if isinstance(value, list) and isinstance(schema.get("items"), dict):
        kept = [valve(i, schema["items"], base, level, clearance) for i in value]
        return [k for k in kept if k is not DROP]
    return value


def filter_message(msg, clearance):
    name = msg["type"].split(".")[1] + ".schema.json"
    out = copy.deepcopy(msg)
    # A message's root is the classification baseline; fields say the rest.
    out["data"] = valve(msg["data"], docs[BASE + name], BASE + name, "public", clearance)
    out["heartredacted"] = out["data"] != msg["data"]
    if not out["heartredacted"]:
        del out["heartredacted"]
    # heartprivacy is the highest level actually present after filtering.
    out["heartprivacy"] = next(
        c for c in ("public", "internal", "private")
        if valve(out["data"], docs[BASE + name], BASE + name, "public", c) == out["data"])
    return out


# 1. schemas
for url, doc in docs.items():
    Draft202012Validator.check_schema(doc)
print(f"schemas: {len(docs)} valid 2020-12 documents")

# 2. examples
PLAIN = {"reason-request": "reason-request", "reason-response": "reason-response",
         "ask-request": "ask-request", "invoke-request": "invoke-request",
         "decision-request": "decision-request", "subscription": "subscription",
         "problem": "problem", "batch-request": "batch-request",
         "fleet-request": "fleet-request", "native-request": "native-request",
         "serve-request": "serve-request", "serve-response": "serve-response",
         "signal-table": "signal-table", "consultation": "consultation",
         "lifecycle-request": "lifecycle-request", "brain-lease": "brain-lease",
         "brain-signal": "brain-signal", "brain-offline-request": "brain-offline-request",
         "presence-method": "presence-method"}
messages = {}
for f in sorted(EXAMPLES.glob("*.json")):
    obj = json.loads(f.read_text())
    target = PLAIN.get(f.stem, "envelope")
    errs = errors(target + ".schema.json", obj)
    check(not errs, f"example {f.name}: {errs}")
    if target == "envelope":
        messages[f.stem] = obj
print(f"examples: {len(list(EXAMPLES.glob('*.json')))} checked")

# 3. privacy lint
# Proposals and Trials are never public: their lowest level is internal.
for name, root in (("pulse", "public"), ("card", "public"), ("rhythm", "public"), ("arrhythmia", "public"), ("action", "public"),
                   ("receipt", "public"), ("evolution", "public"), ("rule", "public"), ("control", "public"),
                   ("proposal", "internal"), ("trial", "internal")):
    url = BASE + name + ".schema.json"
    lint(docs[url], url, root, name)
print("privacy lint: done")

# 4. Valve
# Proposals and Trials are never public: a public caller gets no view at all.
NEVER_PUBLIC = {"heart.proposal", "heart.trial"}
for stem, msg in messages.items():
    for clearance in ("public", "internal", "private"):
        if clearance == "public" and msg["type"] in NEVER_PUBLIC:
            continue
        view = filter_message(msg, clearance)
        errs = errors("envelope.schema.json", view)
        check(not errs, f"valve {stem}@{clearance}: {errs}")
expected = json.loads((EXAMPLES / "pulse-public.json").read_text())
check(filter_message(messages["pulse-internal"], "public") == expected,
      "valve: public view of pulse-internal differs from pulse-public.json")
leak = copy.deepcopy(messages["pulse-internal"])
leak["data"]["secretField"] = "x"
check("secretField" not in filter_message(leak, "internal")["data"]
      and "secretField" in filter_message(leak, "private")["data"],
      "valve: unclassified field survived filtering")
print("valve: views valid at every clearance; unclassified fields dropped")

# 5. negative cases
base = messages["pulse-internal"]
def mutate(fn):
    m = copy.deepcopy(base); fn(m); return m
BAD = {
    "never-share on the wire": lambda m: m.update(heartprivacy="never-share"),
    "missing specversion": lambda m: m.pop("specversion"),
    "unknown status": lambda m: m["data"].update(status="sad"),
    "bad extension attr version": lambda m: m.update(heartversion="2.0"),
    "pulse without status": lambda m: m["data"].pop("status"),
    "bad source": lambda m: m.update(source="https://x.example"),
    "bad timestamp": lambda m: m.update(time="yesterday"),
    "unknown heart type": lambda m: m.update(type="heart.madeup"),
}
# A Proposal that cannot be checked or undone must be refused (Core Standard, Section 7.3), and a learned
# Playbook must say where it came from and when it expires (Section 7.6).
BAD_DATA = {
    "proposal without undoIf": ("proposal", lambda d: d.pop("undoIf")),
    "proposal without expect": ("proposal", lambda d: d.pop("expect")),
    "proposal with empty undoIf": ("proposal", lambda d: d.update(undoIf={})),
    "check with unknown op": ("proposal", lambda d: d["expect"][0].update(op="~")),
    "learned rule without evidence": ("rule-learned", lambda d: d.pop("evidence")),
    "adopted learned rule without reviewBy": ("rule-learned", lambda d: d.update(status="adopted", adoptedAt="2026-11-12T15:00:00Z")),
    "learned rule without undoIf": ("rule-learned", lambda d: d["then"].pop("undoIf")),
    "adopted rule without adoptedAt": ("rule", lambda d: d.pop("adoptedAt")),
    "rule with an unknown field": ("rule", lambda d: d.update(physical=True)),
    "rule with old vital trigger": ("rule", lambda d: d.update(when={"vital": "x.y", "above": 1})),
    "envelope with zero maxStep": ("control", lambda d: d["allowlist"][1]["envelope"]["arguments"]["ramp_rate"].update(maxStep=0)),
    "envelope argument without max": ("control", lambda d: d["allowlist"][1]["envelope"]["arguments"]["ramp_rate"].pop("max")),
    "old flat envelope": ("control", lambda d: d["allowlist"][1].update(envelope={"min": 1, "max": 3})),
    "approval timeout under a minute": ("control", lambda d: d.update(approvalTimeoutSec=30)),
    "trial in an unknown state": ("trial", lambda d: d.update(state="awaiting")),
    "trial with both proposal and rule": ("trial", lambda d: d.update(ruleId="learned-gentle-ramp")),
    "proposal with previous values": ("proposal", lambda d: d.update(previous={"ramp_rate": 2.0})),
    "check with unknown statistic": ("proposal", lambda d: d["expect"][0].update(feature="rate")),
    "allowlist entry for nobody": ("control", lambda d: d["allowlist"][0].update(initiators=[])),
}
for label, (stem, fn) in BAD_DATA.items():
    m = copy.deepcopy(messages[stem]); fn(m["data"])
    check(bool(errors("envelope.schema.json", m)) or bool(errors(m["dataschema"].rsplit("/", 1)[1], m["data"])), f"negative accepted: {label}")
c = json.loads((EXAMPLES / "consultation.json").read_text()); c["features"][0]["samples"] = [1, 2, 3]
check(bool(errors("consultation.schema.json", c)), "negative accepted: consultation with raw samples")
c = json.loads((EXAMPLES / "consultation.json").read_text()); c["features"][0]["stats"]["last"] = 1
check(bool(errors("consultation.schema.json", c)), "negative accepted: feature with old 'last' statistic")
BAD.update({k: None for k in BAD_DATA}); BAD["consultation with raw samples"] = None; BAD["feature with last"] = None
for label, fn in BAD.items():
    if fn is not None:
        check(bool(errors("envelope.schema.json", mutate(fn))), f"negative accepted: {label}")
print(f"negative: {len(BAD)} broken messages tried")

for attr in [k for m in messages.values() for k in m if k.startswith("heart")]:
    check(attr.isalnum() and attr.islower() and len(attr) <= 20,
          f"CloudEvents attribute name invalid: {attr}")

# 6. standard vs OpenAPI
api = yaml.safe_load((ROOT / "openapi" / "heart-api-1.0.yaml").read_text())
in_api = {(m.upper(), path, op["x-heart-level"]) for path, item in api["paths"].items()
          for m, op in item.items() if m in ("get", "post", "put", "patch", "delete")}
text = (ROOT / "HEART-API-Standard-1.0.md").read_text()
rows = re.findall(r"^\| `(GET|POST|PUT|PATCH|DELETE) (\S+)` \| .+? \| ([A-Za-z ]+?)(?: only)? \|$", text, re.M)
rows = [(m, p, lvl.replace(" ", "-")) for m, p, lvl in rows]
in_doc = {(m, p, lvl.lower()) for m, p, lvl in rows}
check(in_api == in_doc, f"operation table mismatch: {in_api ^ in_doc}")
print(f"operations: {len(in_api)} in OpenAPI, {len(in_doc)} in the standard")

# 7. ports
ports = api["servers"][0]["variables"]["port"]
check(ports["default"] == "43278" and "7685" in ports["enum"], "OpenAPI ports wrong")
check("`43278/tcp`" in text and "`7685/tcp`" in text, "standard ports missing")
card = messages["card"]
check(":43278/" in card["data"]["endpoints"]["gateway"], "card gateway port wrong")
bad_alias = copy.deepcopy(card)
bad_alias["data"]["system"]["alias"] = "has.dot"
check(bool(errors("envelope.schema.json", bad_alias)), "alias with a dot accepted")
public_card = filter_message(
    {**card, "data": {**card["data"], "endpoints": {"socket": "/run/heart/x.sock"}}}, "public")
check("socket" not in public_card["data"]["endpoints"], "socket path leaked to public")
print("ports: consistent; aliases cannot look like system IDs; socket paths stay internal")

# 8. custom definitions
PLACEHOLDER = re.compile(r"\{\{\s*([^}]+?)\s*\}\}")


def injection_errors(d):
    found = []
    props = d.get("inputSchema", {}).get("properties", {})
    for ph in PLACEHOLDER.findall(json.dumps(d["native"]["request"])):
        if ph == "value" and d["kind"] == "property":
            continue
        name = ph.removeprefix("arguments.")
        if not ph.startswith("arguments.") or name not in props:
            found.append(f"undeclared placeholder {ph}")
        elif props[name].get("type") == "string" and not (
                "enum" in props[name] or "pattern" in props[name]):
            found.append(f"unconstrained string input {name}")
    native = d["native"]
    if native["protocol"] == "process" and not isinstance(native["request"].get("argv"), list):
        found.append("process adapter without argv list")
    return found


defs = sorted((EXAMPLES / "custom").glob("*.json"))
protocols = set()
for f in defs:
    d = json.loads(f.read_text())
    errs = errors("custom-definition.schema.json", d) + injection_errors(d)
    check(not errs, f"custom {f.name}: {errs}")
    protocols.add(d["native"]["protocol"])
    check(d["native"]["protocol"] in text, f"protocol not in standard: {d['native']['protocol']}")
base_def = json.loads(defs[0].read_text())
evil = copy.deepcopy(json.loads((EXAMPLES / "custom" / "process-edge-rotate_logs.json").read_text()))
evil["native"]["request"]["argv"].append("{{arguments.path}}")
check(bool(injection_errors(evil)), "undeclared placeholder accepted")
evil["inputSchema"]["properties"]["path"] = {"type": "string"}
check(bool(injection_errors(evil)), "free-text string placeholder accepted")
print(f"custom: {len(defs)} definitions across {len(protocols)} protocols, injection rules hold")

# 9. coverage
types = set(docs[BASE + "common.schema.json"]["$defs"]["messageType"]["anyOf"][0]["enum"])
have = {m["type"] for m in messages.values()}
check(types <= have, f"message types without an example: {types - have}")
for stem, msg in messages.items():
    honest = next((c for c in ("public", "internal", "private")
                   if filter_message(msg, c)["data"] == msg["data"]), "private")
    check(msg["heartprivacy"] == honest,
          f"{stem}: heartprivacy is {msg['heartprivacy']}, content is {honest}")
scopes = set(api["components"]["securitySchemes"]["oauth2"]["flows"]["clientCredentials"]["scopes"])
modules = set(docs[BASE + "common.schema.json"]["$defs"]["module"]["enum"])
seen = set()
for path, item in api["paths"].items():
    for method, op in item.items():
        if method not in ("get", "post", "put", "patch", "delete"):
            continue
        level = op["x-heart-level"]
        if level.startswith("module-"):
            seen.add(level[7:])
            if method != "get":
                check("x-heart-risk" in op, f"no risk: {method} {path}")
        for req in op.get("security", []):
            check(set(req.get("oauth2", [])) <= scopes, f"unknown scope on {path}")
check(seen == modules, f"module mismatch: {seen ^ modules}")
print(f"coverage: {len(types)} message types, {len(modules)} modules, "
      f"{sum(1 for i in api['paths'].values() for k in i if k in ('get','post','put','patch','delete'))} operations")

# 10. HEART Core: the index must be current and the test cases must resolve
index = json.loads((ROOT / "core" / "api-index-1.0.json").read_text())
errs = errors("api-index.schema.json", index)
check(not errs, f"api index: {errs[:3]}")
indexed = {o["id"] for o in index["operations"]}
in_openapi = {op["operationId"] for item in api["paths"].values()
              for k, op in item.items() if k in ("get", "post", "put", "patch", "delete")}
check(indexed == in_openapi, f"api index is stale: {indexed ^ in_openapi}")
custom_names = {json.loads(f.read_text())["name"] for f in defs}
cases = json.loads((ROOT / "core" / "understanding-cases.json").read_text())["cases"]
for case in cases:
    check(case["target"] is None or case["target"] in indexed | custom_names,
          f"case target unknown: {case['target']}")
print(f"core: index covers all {len(indexed)} operations; {len(cases)} test cases resolve")

# 11. fault codes: the data file behind the fault-code pages equals Part 3's tables
part3 = (ROOT / "HEART-Part-3-Information-Model-1.0.md").read_text()
sec = lambda a, b: part3[part3.index(a):part3.index(b)]
p3_areas = re.findall(r"^\| `([a-z]+)` \| (\d\d) \| (.+?) \|$", sec("### 5.2 How a code is built", "### 5.3"), re.M)
p3_codes = re.findall(r"^\| `([a-z0-9]+\.[a-z0-9-]+)` \| (H-\d{5}) \| (.+?) \|$", sec("### 5.4 The standard fault codes", "### 5.5"), re.M)
p3_events = {e for row in re.findall(r"^\| (`[^|]+) \| .+? \|$", sec("### 5.8 Event types", "\n## 6."), re.M)
             for e in re.findall(r"`([a-z0-9.-]+)`", row)}
fc = json.loads((ROOT / "fault-codes" / "1.0" / "fault-codes.json").read_text())
check(fc.get("format") == "heart-fault-codes/1", "fault-codes.json: format is not heart-fault-codes/1")
strip_refs = lambda m: re.sub(r"\s*\((Part \d|HEART Core|API Standard|Core Standard)[^)]*\)", "", m).strip()
check([(a["area"], a["number"]) for a in fc["areas"]] == [(a, n) for a, n, _ in p3_areas], "fault-codes.json: areas differ from Part 3, Section 5.2")
area_no = {a: n for a, n, _ in p3_areas}
check(len(set(area_no.values())) == len(area_no), "Part 3: two areas share a number")
for name, values in (("code", [c for c, _, _ in p3_codes]), ("H-number", [h for _, h, _ in p3_codes])):
    dup = {v for v in values if values.count(v) > 1}
    check(not dup, f"Part 3: duplicate {name}: {dup}")
for name, key in (("code", "code"), ("H-number", "h")):
    values = [c[key] for c in fc["codes"]]
    dup = {v for v in values if values.count(v) > 1}
    check(not dup, f"fault-codes.json: duplicate {name}: {dup}")
check([(c["code"], c["h"]) for c in fc["codes"]] == [(c, h) for c, h, _ in p3_codes],
      f"fault-codes.json: codes or H-numbers differ from Part 3, Section 5.4: "
      f"{set((c['code'], c['h']) for c in fc['codes']) ^ set((c, h) for c, h, _ in p3_codes)}")
p3_meaning = {c: strip_refs(m) for c, _, m in p3_codes}
known = {c["code"] for c in fc["codes"]}
for c in fc["codes"]:
    where = f"fault-codes.json {c['code']}"
    check(c["area"] == c["code"].split(".")[0] and c["area"] in area_no, f"{where}: wrong area")
    check(re.fullmatch(r"H-\d{5}", c["h"]) and c["h"][2:4] == area_no.get(c["area"]), f"{where}: H-number {c['h']} is not in its area's range")
    check(c["meaning"] == p3_meaning.get(c["code"]), f"{where}: meaning differs from Part 3")
    check(c["severity"] in ("info", "warning", "error", "critical"), f"{where}: unknown severity")
    check(3 <= len(c["causes"]) <= 6 and 3 <= len(c["fixes"]) <= 7, f"{where}: needs 3 to 6 causes and 3 to 7 fixes")
    check(all(isinstance(c.get(k), str) and c[k].strip() for k in ("title", "heartDoes", "whenToCallService")), f"{where}: empty text")
    check(c["appliesTo"] and all(r in known and r != c["code"] for r in c["related"]), f"{where}: unknown related code or no appliesTo")
check(not (known & p3_events), f"event types listed as fault codes: {known & p3_events}")
check({"system.started", "component.recovered", "tending.stopped", "circulation.received"} <= p3_events, "Part 3, Section 5.8: event types missing")
print(f"fault codes: {len(known)} codes in {len(area_no)} areas match Part 3; {len(p3_events)} event types kept apart")

print("\nFAILED: %d" % len(failures) if failures else "\nALL CHECKS PASSED")
sys.exit(1 if failures else 0)