#!/usr/bin/env python3
"""Differential schema oracle for the foundation synthetic inspector (verification only).

Run: python3 scripts/foundation/verify-schema.py [--fixtures <dir>]

Compares the inspector's record-schema verdict (via the test-only Node bridge
scripts/foundation/validate-record.mjs) against the pinned candidate schema
evaluated by the explicitly selected Python jsonschema 4.26.0, over:
  * docs/plans/foundation-v1-candidate/records.fixtures.json cases and path typeCases,
  * every record inside the checked-in inspector fixture bundles,
  * deterministic charter §10.5 mutations (type substitution, null-versus-absent,
    integer bounds, boolean/1.0 numerics, order-insensitive uniqueItems, BMP/non-BMP
    length boundaries, UTF-8 path byte boundaries, escaped surrogates).

Pinned hashes are checked first. A missing or wrong-version jsonschema FAILS; nothing
is installed and nothing is skipped. The platform behaviour the pinned checker depends
on (strftime %Y padding) is witnessed explicitly. The schema verdict column and the
strict bundle/profile verdict column are reported separately. ANY schema-column
disagreement between the pinned checker and the inspector FAILS this oracle; there is
no waiver list. A passing finite corpus is compatibility evidence, not proof of
schema equivalence.
"""
import hashlib
import json
import subprocess
import sys
import tempfile
import unicodedata
from copy import deepcopy
from datetime import datetime
from pathlib import Path

REPO = Path(__file__).resolve().parent.parent.parent
CANDIDATE = REPO / "docs" / "plans" / "foundation-v1-candidate"
BRIDGE = REPO / "scripts" / "foundation" / "validate-record.mjs"
DEFAULT_FIXTURES = REPO / "scripts" / "foundation" / "fixtures"

PINNED = {
    "records.schema.json": "05774aaf6943cb69c113e39ff1c29676a2a230ca7bf665c50dbcaa8049672af6",
    "check.py": "82564a7d3200afcdda0850a9454cac6e6cd6a76687d2162c13cf214d7eac4607",
    "records.fixtures.json": "d433d06da5cd38baf9e51c8857244ee70375db3b68e02a5325a6d1c2cc47da85",
}
REQUIRED_JSONSCHEMA = "4.26.0"



def fail(msg):
    print(f"FAIL: {msg}")
    sys.exit(1)


def sha256(path):
    return hashlib.sha256(path.read_bytes()).hexdigest()


# --- gate 1: pinned inputs ------------------------------------------------------------
for name, digest in PINNED.items():
    actual = sha256(CANDIDATE / name)
    if actual != digest:
        fail(f"pinned hash mismatch for {name}: {actual}")
print(f"PASS: {len(PINNED)} pinned candidate inputs match")

# --- gate 2: explicit oracle dependency ------------------------------------------------
try:
    from importlib.metadata import version as package_version
    from jsonschema import Draft202012Validator, FormatChecker
    jsonschema_version = package_version("jsonschema")
except Exception as exc:  # environment gate: fail, never install or skip
    fail(f"jsonschema is not importable ({exc}); install nothing, select the pinned environment")
if jsonschema_version != REQUIRED_JSONSCHEMA:
    fail(f"jsonschema {jsonschema_version} is not the pinned {REQUIRED_JSONSCHEMA}")
print(f"PASS: python {sys.version.split()[0]} jsonschema {jsonschema_version}")

# --- gate 3: platform witness for the pinned checker's date-time round-trip ------------
# check.py accepts a date-time only if strptime -> strftime reproduces the input text.
# On the measured platform (CPython 3.12.8, glibc 2.44) strftime("%Y") is not
# zero-padded below year 1000, so the pinned checker refuses years 0001..0999 and the
# inspector was aligned to that measured behaviour. A platform that pads %Y would make
# the pinned checker accept those years; that is a platform boundary, reported here
# explicitly rather than surfacing as an unexplained disagreement below.
strftime_year_999 = datetime(999, 1, 1).strftime("%Y")
if strftime_year_999 != "999":
    fail(f"platform boundary: strftime('%Y') for year 999 is {strftime_year_999!r}, not '999'; "
         "the pinned checker's calendar verdicts differ from the measured platform")
print(f"platform witness: strftime('%Y') for year 999 -> {strftime_year_999!r} (pinned checker refuses years 0001..0999)")


# --- oracle setup: identical to the pinned check.py -----------------------------------
def unique_object(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise ValueError(f"duplicate JSON key: {key}")
        result[key] = value
    return result


def reject_constant(value):
    raise ValueError(f"non-JSON numeric constant: {value}")


def loads(text):
    return json.loads(text, object_pairs_hook=unique_object, parse_constant=reject_constant)


formats = FormatChecker()


@formats.checks("date-time")
def utc_milliseconds(value):
    if not isinstance(value, str):
        return True
    try:
        parsed = datetime.strptime(value, "%Y-%m-%dT%H:%M:%S.%fZ")
    except ValueError:
        return False
    return parsed.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z" == value


@formats.checks("mosaic-relative-path")
def relative_path(value):
    if not isinstance(value, str):
        return True
    try:
        size = len(value.encode("utf-8"))
    except UnicodeEncodeError:
        return False
    return (
        0 < size <= 4096
        and not any(unicodedata.category(c) in {"Cc", "Cf", "Cs"} for c in value)
        and "\\" not in value
        and all(part not in {"", ".", ".."} for part in value.split("/"))
    )


schema = loads((CANDIDATE / "records.schema.json").read_text())
Draft202012Validator.check_schema(schema)
validator = Draft202012Validator(schema, format_checker=formats)

# Strict profile oracle (addendum FI-C2-1). An in-memory variant of the pinned schema in
# which the three typed grammars end in Python's \Z (end of string only) instead of "$"
# (end of string, or before exactly one final newline under re.search). The pinned bytes
# are untouched; the variant exists only to decide, independently of the inspector, which
# schema-valid values the strict production profile must refuse.
PROFILE_DEFS = ("id", "runtimeId", "digest")
strict_schema = deepcopy(schema)
for definition in PROFILE_DEFS:
    pinned_pattern = strict_schema["$defs"][definition]["pattern"]
    if not pinned_pattern.endswith("$") or pinned_pattern.endswith("\\$"):
        fail(f"$defs/{definition} pattern is not end-anchored as pinned: {pinned_pattern!r}")
    strict_schema["$defs"][definition]["pattern"] = pinned_pattern[:-1] + "\\Z"
Draft202012Validator.check_schema(strict_schema)
strict_validator = Draft202012Validator(strict_schema, format_checker=formats)


def python_verdict(raw):
    try:
        doc = loads(raw)
    except ValueError:
        return "parse-error"
    return validator.is_valid(doc)


def strict_profile_verdict(raw):
    """Pinned-schema-valid AND valid under the end-of-string variant (the profile column's oracle)."""
    doc = loads(raw)
    return validator.is_valid(doc) and strict_validator.is_valid(doc)


# --- corpus ----------------------------------------------------------------------------
corpus = []          # {name, raw}
strict_only = {}     # name -> expected python verdict when the strict parser rejects
names = set()


def add(name, raw, python_expected=None):
    if name in names:
        fail(f"duplicate corpus name {name}")
    names.add(name)
    corpus.append({"name": name, "raw": raw})
    if python_expected is not None:
        strict_only[name] = python_expected


expected_columns = {}   # name -> (pinned schema verdict, strict profile verdict), asserted below


def probe(name, raw, schema_expected, profile_expected):
    """A named probe whose pinned-schema and strict-profile verdicts are declared up front."""
    add(name, raw)
    expected_columns[name] = (schema_expected, profile_expected)


def dumps(doc):
    return json.dumps(doc, ensure_ascii=False, separators=(",", ":"))


record_fixtures = loads((CANDIDATE / "records.fixtures.json").read_text())
bases = {}
for case in record_fixtures["cases"]:
    doc = case["document"]
    if validator.is_valid(doc) != case["schemaValid"]:
        fail(f"pinned fixture expectation mismatch: {case['name']}")
    add(f"fixture-{case['name']}", dumps(doc))
    if case["schemaValid"] and doc["kind"] not in bases:
        bases[doc["kind"]] = doc
print(f"PASS: {len(record_fixtures['cases'])} pinned record fixtures agree with the pinned expectations")

base_task = deepcopy(bases["task"])


def task_with_path(path_value):
    doc = deepcopy(base_task)
    doc["payload"]["restrictions"] = {
        "operations": ["work.read"], "readPaths": [{"root": "workspace", "path": path_value}],
        "writePaths": [], "network": "none", "endpointRefs": [],
    }
    return doc


for case in record_fixtures["typeCases"]:
    if case["definition"] != "relativePath":
        continue
    over_parser_bound = len(case["value"].encode("utf-8", "surrogatepass")) > 4096
    add(f"typecase-{case['name']}", dumps(task_with_path(case["value"])),
        python_expected=case["schemaValid"] if over_parser_bound else None)

fixtures_dir = DEFAULT_FIXTURES
if "--fixtures" in sys.argv:
    fixtures_dir = Path(sys.argv[sys.argv.index("--fixtures") + 1])
bundle_files = sorted((fixtures_dir / "bundles").glob("*.json"))
if not bundle_files:
    fail(f"no fixture bundles under {fixtures_dir}")
seen_raw = set()
bundle_records = 0
for bf in bundle_files:
    bundle = loads(bf.read_text())
    for i, rec in enumerate(bundle.get("records", [])):
        raw = dumps(rec)
        if raw in seen_raw:
            continue
        seen_raw.add(raw)
        bundle_records += 1
        add(f"bundle-{bf.stem}-{i}", raw)

# §10.5 deterministic mutations (supported kinds only; unsupported kinds are never
# payload-judged by the inspector, so their mutations would not be differential).
SUPPORTED_KINDS = ["agent-definition", "project", "workspace", "registration", "mission", "task", "assignment", "decision"]
SUBSTITUTES = [("int", 1), ("str", "x"), ("bool", True), ("null", None), ("array", []), ("object", {})]
for kind, base in sorted(bases.items()):
    if kind not in SUPPORTED_KINDS:
        continue
    for field in list(base.keys()):
        for label, sub in SUBSTITUTES:
            doc = deepcopy(base)
            doc[field] = sub
            add(f"mut-{kind}-env-{field}-{label}", dumps(doc))
        doc = deepcopy(base)
        del doc[field]
        add(f"mut-{kind}-env-{field}-absent", dumps(doc))
    for field in list(base["payload"].keys()):
        for label, sub in SUBSTITUTES:
            doc = deepcopy(base)
            doc["payload"][field] = sub
            add(f"mut-{kind}-payload-{field}-{label}", dumps(doc))
        doc = deepcopy(base)
        del doc["payload"][field]
        add(f"mut-{kind}-payload-{field}-absent", dumps(doc))

for label, value in [("zero", 0), ("one", 1), ("two", 2), ("neg", -1), ("max", 2**53 - 1),
                     ("true", True), ("false", False), ("string", "1")]:
    doc = deepcopy(base_task)
    doc["revision"] = value
    if value == 2:
        doc["supersedes"] = {"kind": "task", "id": doc["id"], "scope": doc["scope"], "revision": 1}
    add(f"mut-revision-{label}", dumps(doc))
for label, value in [("zero", 0), ("two", 2), ("true", True), ("string", "1")]:
    doc = deepcopy(base_task)
    doc["schemaVersion"] = value
    add(f"mut-schemaversion-{label}", dumps(doc))

# Strict-only lexical cases: the inspector refuses at parse time; Python judges the value.
raw_base = dumps(base_task)
assert raw_base.startswith('{"schemaVersion":1,') and raw_base.count('"schemaVersion":1') == 1


def envelope_revision(replacement):
    # The first "revision" key in the serialized envelope is the record's own.
    assert '"revision":1,"createdAt"' in raw_base
    return raw_base.replace('"revision":1,"createdAt"', f'"revision":{replacement},"createdAt"', 1)


add("strict-revision-1.0", envelope_revision("1.0"), python_expected=True)
add("strict-schemaversion-1.0", raw_base.replace('"schemaVersion":1', '"schemaVersion":1.0', 1), python_expected=True)
add("strict-revision-1e0", envelope_revision("1e0"), python_expected=True)
add("strict-revision-negative-zero", envelope_revision("-0"), python_expected=False)
add("strict-revision-2^53", envelope_revision(str(2**53)), python_expected=False)
add("strict-revision-huge", envelope_revision("123456789012345678901234567890"), python_expected=False)
add("strict-revision-leading-zero", envelope_revision("01"), python_expected="parse-error")
add("strict-duplicate-key", envelope_revision("1,\"revision\":1"), python_expected="parse-error")
add("strict-nan", envelope_revision("NaN"), python_expected="parse-error")
add("strict-trailing-content", raw_base + "{}", python_expected="parse-error")

# uniqueItems: nested objects compare order-insensitively in both implementations.
doc = deepcopy(base_task)
doc["payload"]["criteria"] = [{"id": "c1", "text": "done"}, {"text": "done", "id": "c1"}]
add("mut-unique-criteria-reordered-duplicate", dumps(doc))
doc = deepcopy(base_task)
doc["payload"]["criteria"] = [{"id": "c1", "text": "done"}, {"id": "c1", "text": "other"}]
add("mut-unique-criteria-distinct", dumps(doc))
doc = deepcopy(base_task)
dep = {"kind": "task", "id": "dep-1", "scope": doc["scope"], "revision": 1}
dep2 = {"revision": 1, "scope": deepcopy(doc["scope"]), "id": "dep-1", "kind": "task"}
doc["payload"]["dependencies"] = [dep, dep2]
add("mut-unique-dependencies-reordered-duplicate", dumps(doc))
base_decision = deepcopy(bases["decision"])
doc = deepcopy(base_decision)
doc["payload"]["subjectRefs"] = [deepcopy(doc["payload"]["subjectRefs"][0]), dict(reversed(list(doc["payload"]["subjectRefs"][0].items())))]
add("mut-unique-subjectrefs-reordered-duplicate", dumps(doc))
doc = deepcopy(base_task)
doc["payload"]["restrictions"] = {"operations": ["work.read", "work.read"], "readPaths": [], "writePaths": [], "network": "none", "endpointRefs": []}
add("mut-unique-operations-duplicate", dumps(doc))

# BMP / non-BMP code-point length boundaries (schema maxLength counts code points).
base_project = deepcopy(bases["project"])
for label, ch in [("bmp", "é"), ("nonbmp", "😀"), ("ascii", "a")]:
    for n in (1, 128, 129):
        doc = deepcopy(base_project)
        doc["payload"]["displayName"] = ch * n
        add(f"mut-displayname-{label}-{n}", dumps(doc))
    for n in (4000, 4001):
        # Non-ASCII 4000-code-point strings exceed the strict parser's 4096-byte
        # string bound: strict-only, with the schema's own code-point answer declared.
        expect = (n <= 4000) if len(ch.encode("utf-8")) * n > 4096 else None
        doc = deepcopy(base_task)
        doc["payload"]["purpose"] = ch * n
        add(f"mut-purpose-{label}-{n}", dumps(doc), python_expected=expect)
        doc = deepcopy(base_task)
        doc["payload"]["criteria"] = [{"id": "c1", "text": ch * n}]
        add(f"mut-criterion-text-{label}-{n}", dumps(doc), python_expected=expect)
doc = deepcopy(base_project)
doc["payload"]["displayName"] = ""
add("mut-displayname-empty", dumps(doc))
doc = deepcopy(base_task)
doc["payload"]["criteria"] = [{"id": "c1", "text": ""}]
add("mut-criterion-text-empty", dumps(doc))

# UTF-8 byte boundaries for relative paths (format counts bytes, maxLength counts code points).
for label, ch, per in [("ascii", "a", 1), ("bmp", "é", 2), ("nonbmp", "😀", 4)]:
    for n in (4096 // per, 4096 // per + 1):
        over = n * per > 4096  # beyond the parser's byte bound: strict-only, schema rejects too
        add(f"mut-path-{label}-{n}x{per}", dumps(task_with_path(ch * n)), python_expected=False if over else None)
add("mut-path-5000-ascii", dumps(task_with_path("a" * 5000)), python_expected=False)
add("mut-path-zwsp", dumps(task_with_path("a\u200bb")))
add("mut-path-line-separator", dumps(task_with_path("a\u2028b")))
add("mut-path-del", dumps(task_with_path("a\x7fb")))
add("mut-path-c1-control", dumps(task_with_path("a\x85b")))
add("mut-path-tab", dumps(task_with_path("a\tb")))
add("mut-path-empty", dumps(task_with_path("")))
add("mut-path-null-root", dumps(task_with_path(None)))
add("mut-path-int", dumps(task_with_path(7)))

# Escaped surrogate cases (raw JSON text, not Python strings).
raw_project = dumps(base_project)
marker = f'"displayName":{json.dumps(base_project["payload"]["displayName"], ensure_ascii=False)}'
assert raw_project.count(marker) == 1
add("escape-paired-surrogate-displayname", raw_project.replace(marker, '"displayName":"\\ud83d\\ude00"'))
add("escape-lone-high-surrogate-displayname", raw_project.replace(marker, '"displayName":"\\ud800x"'), python_expected=True)
add("escape-lone-low-surrogate-displayname", raw_project.replace(marker, '"displayName":"x\\udc00"'), python_expected=True)
add("escape-reversed-surrogates-displayname", raw_project.replace(marker, '"displayName":"\\udc00\\ud800"'), python_expected=True)
raw_path = dumps(task_with_path("PATHMARK"))
add("escape-paired-surrogate-path", raw_path.replace('"PATHMARK"', '"a\\ud83d\\ude00b"'))
add("escape-lone-surrogate-path", raw_path.replace('"PATHMARK"', '"a\\ud800b"'), python_expected=False)
add("escape-unicode-control-path", raw_path.replace('"PATHMARK"', '"a\\u0001b"'))
add("escape-unicode-solidus-path", raw_path.replace('"PATHMARK"', '"a\\/b"'))
add("escape-unicode-backslash-path", raw_path.replace('"PATHMARK"', '"a\\\\b"'))

# Pattern-anchored typed strings and line endings (addendum FI-C2-1; verdict FI-FILBERT-5).
# The pinned checker evaluates "pattern" through Python `re.search`, whose "$" also
# matches before exactly one final "\n". The inspector's schema column must reproduce
# that verdict (schema agreement, never a waived disagreement) while its strict profile
# refuses the value before admission; every other line-ending form fails the pattern in
# both implementations. The four original witnesses are preserved by name; the matrix
# below covers every affected type family, and each expectation is asserted.
doc = deepcopy(base_task)
doc["id"] = doc["id"] + "\n"
probe("pattern-id-trailing-newline", dumps(doc), True, False)
doc = deepcopy(base_task)
doc["scope"]["projectId"] = doc["scope"]["projectId"] + "\n"
probe("pattern-scope-project-id-trailing-newline", dumps(doc), True, False)
doc = deepcopy(base_task)
doc["authorizationRef"] = doc["authorizationRef"] + "\n"
probe("pattern-authorization-ref-trailing-newline", dumps(doc), True, False)
doc = deepcopy(base_project)
doc["payload"]["policyRef"]["digest"] = doc["payload"]["policyRef"]["digest"] + "\n"
probe("pattern-digest-trailing-newline", dumps(doc), True, False)

NEWLINE_MATRIX = [  # label, mutation, pinned schema verdict, strict profile verdict
    ("valid", lambda v: v, True, True),
    ("one-final-lf", lambda v: v + "\n", True, False),
    ("two-final-lf", lambda v: v + "\n\n", False, False),
    ("crlf", lambda v: v + "\r\n", False, False),
    ("cr", lambda v: v + "\r", False, False),
    ("interior-lf", lambda v: v[:3] + "\n" + v[3:], False, False),
    ("u2028", lambda v: v + "\u2028", False, False),
    ("u2029", lambda v: v + "\u2029", False, False),
]
base_mission = deepcopy(bases["mission"])
base_agent = deepcopy(bases["agent-definition"])
MATRIX_SITES = [  # site label, base document, key path to the typed value
    ("id-record-id", base_task, ["id"]),
    ("id-scope-project-id", base_task, ["scope", "projectId"]),
    ("id-actor-principal-id", base_task, ["createdBy", "principalId"]),
    ("id-registry-ref-id", base_project, ["payload", "policyRef", "id"]),
    ("id-criterion-id", base_mission, ["payload", "criteria", 0, "id"]),
    ("id-agent-type", base_agent, ["payload", "agentType"]),
    ("runtime-id-authorization-ref", base_task, ["authorizationRef"]),
    ("digest-registry-ref-digest", base_project, ["payload", "policyRef", "digest"]),
]
for site, base_doc, key_path in MATRIX_SITES:
    for label, mutate, schema_expected, profile_expected in NEWLINE_MATRIX:
        doc = deepcopy(base_doc)
        holder = doc
        for key in key_path[:-1]:
            holder = holder[key]
        holder[key_path[-1]] = mutate(holder[key_path[-1]])
        probe(f"pattern-matrix-{site}-{label}", dumps(doc), schema_expected, profile_expected)
# time/path keep their existing format semantics: a final newline fails in both.
doc = deepcopy(base_task)
doc["createdAt"] = doc["createdAt"] + "\n"
probe("time-trailing-newline", dumps(doc), False, False)  # strptime rejects the newline; both refuse
probe("path-trailing-newline", dumps(task_with_path("src/main.mjs\n")), False, False)
# Free-form text is not a typed grammar: escaped newlines stay schema- and profile-valid.
doc = deepcopy(base_mission)
doc["payload"]["objective"] = "line one\nline two\n"
probe("free-text-objective-newlines", dumps(doc), True, True)
doc = deepcopy(base_task)
doc["payload"]["purpose"] = "multi\r\nline\u2028text"
probe("free-text-purpose-newlines", dumps(doc), True, True)
doc = deepcopy(base_mission)
doc["payload"]["criteria"] = [{"id": "c1", "text": "done\n"}]
probe("free-text-criterion-newline", dumps(doc), True, True)

# Calendar semantics from the pinned checker.
for label, value in [("feb-30", "2026-02-30T03:00:00.000Z"), ("leap-ok", "2028-02-29T03:00:00.000Z"),
                     ("leap-bad", "2027-02-29T03:00:00.000Z"), ("hour-24", "2026-09-06T24:00:00.000Z"),
                     ("sec-60", "2026-06-30T23:59:60.000Z"), ("month-13", "2026-13-01T00:00:00.000Z"),
                     ("year-0000", "0000-01-01T00:00:00.000Z"), ("no-millis", "2026-09-06T03:00:00Z"),
                     ("offset", "2026-09-06T03:00:00.000+00:00"), ("year-1000", "1000-01-01T00:00:00.000Z"),
                     ("year-9999", "9999-12-31T23:59:59.999Z"),
                     # pinned checker round-trip boundary (gate 3 witness): both must refuse
                     ("year-0001", "0001-01-01T00:00:00.000Z"), ("year-0999", "0999-12-31T00:00:00.000Z"),
                     ("year-0100", "0100-06-15T12:00:00.000Z")]:
    doc = deepcopy(base_task)
    doc["createdAt"] = value
    add(f"calendar-{label}", dumps(doc))

# --- run the bridge ---------------------------------------------------------------------
node_version = subprocess.run(["node", "--version"], check=True, capture_output=True, text=True).stdout.strip()
with tempfile.TemporaryDirectory(prefix="verify-schema-") as tmp:
    corpus_path = Path(tmp) / "corpus.json"
    corpus_path.write_text(json.dumps(corpus, ensure_ascii=True))
    proc = subprocess.run(["node", str(BRIDGE), str(corpus_path)], capture_output=True, text=True)
if proc.returncode != 0:
    fail(f"bridge exited {proc.returncode}: {proc.stderr.strip()[:500]}")
bridge = json.loads(proc.stdout)
if [b["name"] for b in bridge] != [c["name"] for c in corpus]:
    fail("bridge output does not align with the corpus")

# --- compare ----------------------------------------------------------------------------
agree_valid = 0
agree_invalid = 0
disagreements = {}   # name -> (python_schema_valid, inspector_schema_valid, inspector_profile_valid, rule)
profile_valid = 0
profile_invalid = 0
profile_refusals = []   # schema-valid in both, refused only by the strict profile
strict_checked = 0
not_assessed = 0
problems = []
columns = []
seen_expected = set()
for entry, judged in zip(corpus, bridge):
    name = entry["name"]
    py = python_verdict(entry["raw"])
    node_schema = judged["schemaValid"]
    strict = strict_profile_verdict(entry["raw"]) if py is True else False
    columns.append((name, py, strict, node_schema, judged["profileValid"], judged["parse"], judged["rule"]))
    if name in expected_columns:
        seen_expected.add(name)
        if judged["parse"] != "ok":
            problems.append(f"{name}: named probe rejected by the strict parser ({judged['parse']})")
            continue
        if (py, strict) != expected_columns[name]:
            problems.append(f"{name}: pinned checker (schema {py}, strict profile {strict}) != declared {expected_columns[name]}")
            continue
    if judged["parse"] != "ok":
        if name not in strict_only:
            problems.append(f"{name}: strict parser rejected ({judged['parse']}) but no python expectation declared")
            continue
        if py != strict_only[name]:
            problems.append(f"{name}: python {py!r} != expected {strict_only[name]!r} (strict-only case)")
            continue
        strict_checked += 1
        continue
    if name in strict_only:
        problems.append(f"{name}: expected strict rejection but parser accepted")
        continue
    if py == "parse-error":
        problems.append(f"{name}: python parse error but strict parser accepted")
        continue
    if judged["verdict"] == "unsupported-kind":
        # Kind gate only: the inspector never judges these payloads (charter §8).
        not_assessed += 1
        continue
    if not isinstance(node_schema, bool):
        problems.append(f"{name}: inspector schema column is {node_schema!r} for a parsed, supported-kind record")
        continue
    # Profile column (strict bundle/profile verdict) is reported separately from the
    # schema column and never substitutes for it.
    if node_schema:
        if judged["profileValid"]:
            profile_valid += 1
        else:
            profile_invalid += 1
    if py == node_schema:
        if py:
            agree_valid += 1
        else:
            agree_invalid += 1
        # Profile column oracle: schema-valid AND end-of-string-valid (independent Python judgement).
        if judged["profileValid"] != (py and strict):
            problems.append(f"PROFILE {name}: inspector profile {judged['profileValid']} / "
                            f"pinned schema {py} with strict profile {strict} (rule {judged['rule']})")
        elif py and not strict:
            profile_refusals.append(name)
            if judged["rule"] != "profile-pattern-mismatch" or not judged["path"]:
                problems.append(f"PROFILE {name}: strict-only refusal must carry rule profile-pattern-mismatch "
                                f"and a path, got {judged['rule']!r} at {judged['path']!r}")
    else:
        disagreements[name] = (py, node_schema, judged["profileValid"], judged["rule"])

missing_probes = sorted(set(expected_columns) - seen_expected)
if missing_probes:
    problems.append(f"named probes not judged: {missing_probes}")
if not profile_refusals:
    problems.append("no strict-only profile refusal observed: the profile column is not being exercised")
expected_refusals = sorted(n for n, (schema_expected, profile_expected) in expected_columns.items()
                           if schema_expected and not profile_expected)
if not set(expected_refusals) <= set(profile_refusals):
    problems.append(f"declared profile refusals not observed: {sorted(set(expected_refusals) - set(profile_refusals))}")

for n in sorted(disagreements):
    py, node_schema, prof, rule = disagreements[n]
    problems.append(f"DISAGREEMENT {n}: pinned-checker schema {py} / inspector schema {node_schema} "
                    f"(inspector profile {prof}, rule {rule})")

if "--columns" in sys.argv:
    for row in columns:
        print("COL", *row)

print(f"node {node_version}; corpus {len(corpus)} records "
      f"({len(record_fixtures['cases'])} pinned fixtures, {bundle_records} unique bundle records, "
      f"{len(corpus) - len(record_fixtures['cases']) - bundle_records} typeCase/mutation/lexical cases)")
print(f"schema column: agree-valid {agree_valid}, agree-invalid {agree_invalid}, "
      f"DISAGREEMENTS {len(disagreements)}; strict-only (parser-bound) cases: {strict_checked}; "
      f"unsupported-kind records not schema-assessed by the inspector: {not_assessed}")
print(f"profile column (schema-valid records only): profile-valid {profile_valid}, profile-invalid {profile_invalid}")
print(f"profile refusals asserted: {len(profile_refusals)} schema-agreed-valid records refused only by the strict "
      f"typed-string profile (rule profile-pattern-mismatch), {len(expected_refusals)} declared by name; "
      f"{len(expected_columns)} named probes verified against declared schema/profile columns")
if problems:
    for p in problems:
        print(f"  problem: {p}")
    fail(f"{len(problems)} differential problems")
print("PASS: differential schema oracle (finite corpus; compatibility evidence, not equivalence proof)")