jason.woltje
8ebddd6f93
Rocko-authored, Filbert-reviewed inspector (r6 manifest a4a44930...) with full review/build/verdict evidence under docs/plans/reviews. 43/0 selftests, oracle zero-disagreement, foundation checker PASS. Owner A9 acceptance recorded separately.
576 lines
28 KiB
Python
576 lines
28 KiB
Python
#!/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)")
|