007 — CPython json 3.14.5：身分測試早就在跑，而答案取決於你問哪一種相同

{
  "name": "json-accelerator-recut",
  "what_it_is": "A codec with an optional accelerator, where the choice is a value you can re-evaluate and the claim 'the accelerator is not structural' is a written contract the run checks.",
  "examined": {
    "project": "CPython json",
    "version": "3.14.5",
    "license": "PSF-2.0",
    "measured": {
      "json/__init__.py": 365,
      "json/decoder.py": 364,
      "json/encoder.py": 461,
      "json/scanner.py": 73,
      "accelerator": "_json, a built-in C module",
      "selection_shape": "make_scanner = c_make_scanner or py_make_scanner, evaluated at import time"
    },
    "identity_test_run_on_upstream": {
      "values": "21 of 21 encode/decode results byte-identical with and without _json",
      "errors": "1 of 13 error messages differs",
      "the_one": "'\\x' -> C: Invalid \\escape: line 1 column 2 (char 1) / py: Invalid \\escape: 'x': line 1 column 3 (char 2)"
    }
  },
  "the_finding": "CPython already runs the identity test on _json, as a runtime fallback: the module structures itself so the accelerator can be absent. Under the witness 'same value', the accelerator is not structural. Under the witness 'same error message', it is observable. Same code, two witnesses, two answers — which is example 007's finding, confirmed on stdlib code that predates it by two decades.",
  "the_second_finding": "Turning the accelerator off is harder than it looks. `make_scanner = c_make_scanner or py_make_scanner` runs at import time, so setting c_make_scanner = None afterwards changes nothing that has already been bound. My first comparison ran C against C and reported the two implementations identical.",
  "equivalence_contract": {
    "must_be_identical": ["encoded output for every corpus value"],
    "may_differ": ["error message text"],
    "must_not_differ": ["the class of error raised", "whether an input is accepted at all"]
  },
  "sets": {
    "FMS": "this file, including the equivalence contract",
    "SCL": "whether this deployment may use the accelerator at all",
    "SMS": "selection and the equivalence runner's driver",
    "TMS": "two implementations of the same escaper, each importing nothing",
    "DMS": "what the two implementations agreed and disagreed about, by contract clause"
  },
  "non_goals": [
    "Being a JSON library. One escaper, no parser.",
    "Claiming CPython's divergence is a bug. It is a message, the contract permits it, and this entry's point is that the contract has to be written down before that sentence means anything."
  ]
}

{
  "policy_version": "1.0",
  "_comment": [
    "A deployment may refuse the accelerator - for reproducibility across hosts,",
    "or because it is auditing the pure implementation. Upstream has no such",
    "switch: the choice is made at import time by what is installed."
  ],
  "accelerator": "allow",
  "error_text_must_match": false
}

"""What this deployment permits."""
import json
import pathlib

_C = json.loads((pathlib.Path(__file__).parent / "policy.json").read_text(encoding="utf-8"))


def accelerator_allowed():
    return _C["accelerator"] == "allow"


def error_text_must_match():
    return bool(_C["error_text_must_match"])

"""Run one corpus through one implementation and collect what happened."""


def run(module, corpus):
    values, errors = [], []
    for item in corpus:
        try:
            values.append(module.escape(item))
            errors.append(None)
        except Exception as exc:  # noqa: BLE001 - the class IS the observation
            values.append(None)
            errors.append((type(exc).__name__, str(exc)))
    return values, errors

"""Choosing an implementation, and being able to say which one you got.

Upstream binds the choice at import time:

    make_scanner = c_make_scanner or py_make_scanner    # json/scanner.py

That is one line, it is correct, and it has a consequence: after import there is
no way to select the other one, and no way to ask which was taken except by
inspecting the type of an object it produced. Setting c_make_scanner = None
later changes nothing, because the `or` already ran.

Here the choice is made per call from values that are still live, and the
selection is returned beside the result.
"""
import importlib

FAST = "impl/fast"
PLAIN = "impl/plain"


def available():
    """Which implementations this installation actually has."""
    found = []
    for name in (FAST, PLAIN):
        try:
            importlib.import_module(f"TMS.{name.replace('/', '.')}")
            found.append(name)
        except ImportError:
            pass
    return found


def choose(prefer_fast, allowed):
    """Return (name, module). Never raises for a missing accelerator."""
    order = [FAST, PLAIN] if (prefer_fast and allowed) else [PLAIN]
    for name in order:
        try:
            return name, importlib.import_module(f"TMS.{name.replace('/', '.')}")
        except ImportError:
            continue
    raise RuntimeError("no implementation is available at all")

"""The "accelerated" escaper: a table lookup built once.

Stands in for _json. Imports nothing.
"""

NAME = "impl/fast"

_TABLE = {c: f"\\u{ord(c):04x}" for c in map(chr, range(0x20))}
_TABLE.update({'"': '\\"', "\\": "\\\\", "\n": "\\n", "\t": "\\t", "\r": "\\r"})


def escape(text):
    if not isinstance(text, str):
        raise TypeError("escape expects str")
    return "".join(_TABLE.get(ch, ch) for ch in text)

"""The plain escaper: branches, no table.

Deliberately identical in output and DIFFERENT in one error message, which is
the divergence measured in CPython 3.14.5 between _json and its fallback.
Imports nothing.
"""

NAME = "impl/plain"

_NAMED = {'"': '\\"', "\\": "\\\\", "\n": "\\n", "\t": "\\t", "\r": "\\r"}


def escape(text):
    if not isinstance(text, str):
        # Upstream's fallback names the offending value; the C path does not.
        raise TypeError(f"escape expects str, got {type(text).__name__}")
    out = []
    for ch in text:
        if ch in _NAMED:
            out.append(_NAMED[ch])
        elif ord(ch) < 0x20:
            out.append(f"\\u{ord(ch):04x}")
        else:
            out.append(ch)
    return "".join(out)

"""What the two implementations agreed and disagreed about, clause by clause.

The point is not "are they the same". It is "are they the same in the ways the
contract says they must be" — because 'the same' is not a property a machine can
check without being told which observations count.
"""


def compare(contract, corpus, a_name, a_result, b_name, b_result):
    a_values, a_errors = a_result
    b_values, b_errors = b_result
    findings = {"identical_values": 0, "differing_values": [], "same_error_class": 0,
                "differing_error_text": [], "differing_error_class": [], "acceptance_differs": []}

    for item, av, bv, ae, be in zip(corpus, a_values, b_values, a_errors, b_errors):
        label = repr(item)[:40]
        if (ae is None) != (be is None):
            findings["acceptance_differs"].append((label, a_name if ae is None else b_name))
            continue
        if ae is None:
            if av == bv:
                findings["identical_values"] += 1
            else:
                findings["differing_values"].append((label, av, bv))
            continue
        if ae[0] != be[0]:
            findings["differing_error_class"].append((label, ae[0], be[0]))
        else:
            findings["same_error_class"] += 1
            if ae[1] != be[1]:
                findings["differing_error_text"].append((label, ae[1], be[1]))
    return findings


def verdict(contract, findings, error_text_must_match):
    """Contract clauses, each answered by the run."""
    clauses = []
    clauses.append(("encoded output identical for every value",
                    not findings["differing_values"],
                    f"{findings['identical_values']} identical, {len(findings['differing_values'])} differing"))
    clauses.append(("the class of error is the same",
                    not findings["differing_error_class"],
                    f"{findings['same_error_class']} same class, {len(findings['differing_error_class'])} differing"))
    clauses.append(("acceptance is the same",
                    not findings["acceptance_differs"],
                    f"{len(findings['acceptance_differs'])} input(s) one accepted and the other did not"))
    # A waived pass must never render as a pass. The first version of this line
    # printed "ok" with "(contract says MAY differ)" appended, and a reader
    # skimming the column saw four greens over an observed difference. The
    # verdict WITHOUT the waiver is now shown beside the one with it, because
    # that is the only thing separating "a declared, temporary exception" from
    # "the system's identity was quietly rewritten".
    text_differs = bool(findings["differing_error_text"])
    unwaived = not text_differs
    waived = unwaived or not error_text_must_match
    label = "error text identical"
    if text_differs and waived:
        label += "  [WAIVED — would FAIL without the contract's may-differ]"
    elif not error_text_must_match:
        label += "  (contract says MAY differ; nothing used the waiver)"
    clauses.append((label, waived, f"{len(findings['differing_error_text'])} message(s) differ"))
    return clauses


def render(a_name, b_name, findings, clauses):
    lines = ["", f"== {a_name} vs {b_name}"]
    for label, ok, detail in clauses:
        lines.append(f"  {'ok ' if ok else '!! '} {label:<52} {detail}")
    if findings["differing_error_text"]:
        lines.append("")
        lines.append("  the observable difference, in full:")
        for label, a, b in findings["differing_error_text"]:
            lines.append(f"    input {label}")
            lines.append(f"      {a_name:<12} {a}")
            lines.append(f"      {b_name:<12} {b}")
    lines.append("")
    lines.append("  under 'same value' the accelerator is NOT structural.")
    lines.append("  under 'same message' it is observable. The contract is what decides.")
    return "\n".join(lines) + "\n"

"""The island test, and the live measurement of CPython's json accelerator.

    python src/island_test.py

Section 4 is the one that produced this entry: it turns _json off, proves the
switch actually happened, and then asks what changed.
"""
import importlib
import pathlib
import sys

HERE = pathlib.Path(__file__).parent
sys.path.insert(0, str(HERE))

from DMS import equivalence  # noqa: E402
from SCL import policy  # noqa: E402
from SMS import drive, select  # noqa: E402

FAILURES = []


def report(label, ok, detail=""):
    print(f"  {'PASS' if ok else 'FAIL'}  {label}{' - ' + detail if detail else ''}")
    if not ok:
        FAILURES.append(label)


print("\n== 1. each implementation is an island")
for name in ("fast", "plain"):
    module = importlib.import_module(f"TMS.impl.{name}")
    source = (HERE / "TMS" / "impl" / f"{name}.py").read_text(encoding="utf-8")
    report(f"impl/{name} escapes with no sibling loaded", module.escape('a"b') == 'a\\"b',
           module.escape('a"b'))
    report(f"impl/{name} imports nothing", "import " not in source)

print("\n== 2. the selection is a value, not a name bound at import time")
report("both implementations are available", len(select.available()) == 2,
       ", ".join(select.available()))
fast_name, _ = select.choose(prefer_fast=True, allowed=True)
plain_name, _ = select.choose(prefer_fast=False, allowed=True)
refused_name, _ = select.choose(prefer_fast=True, allowed=False)
report("preferring the accelerator selects it", fast_name == "impl/fast")
report("not preferring it selects the other", plain_name == "impl/plain")
report("SCL refusing it selects the other too", refused_name == "impl/plain",
       "upstream has no equivalent switch — the choice is made by what is installed")

print("\n== 3. the comparison refuses to compare something with itself")
{
}
findings = equivalence.compare(
    {}, ["x"], "impl/fast", drive.run(importlib.import_module("TMS.impl.fast"), ["x"]),
    "impl/fast", drive.run(importlib.import_module("TMS.impl.fast"), ["x"]))
report("comparing an implementation with itself finds no difference",
       findings["identical_values"] == 1 and not findings["differing_values"],
       "which is why main.py refuses to report it as an equivalence result")
main_source = (HERE / "main.py").read_text(encoding="utf-8")
report("and main.py has that refusal in code, not in prose",
       "REFUSING to report an equivalence" in main_source
       and "if fast_name == plain_name:" in main_source)

print("\n== 4. measured against CPython json 3.14.5 itself")
import json  # noqa: E402
import json.decoder  # noqa: E402
import json.encoder  # noqa: E402
import json.scanner  # noqa: E402

report("this interpreter is the examined version", sys.version.split()[0] == "3.14.5",
       sys.version.split()[0])
report("the C accelerator is present", importlib.util.find_spec("_json") is not None)

CORPUS = [{"a": 1, "b": [1, 2, {"c": None}]}, {"u": "héllo — 中文"}, [], {},
          {"n": [1.5, -0.0, 1e300]}, {"k": True, "l": None}]
BAD = ['{"a": }', '[1,]', '{"a" 1}', '"unterminated', '{1: 2}', '[1 2]', 'nul',
       '{"a": 1', '[', '1.', '"\\x"', '{"a": 01}']


def snapshot():
    decoder = json.decoder.JSONDecoder()
    encoder = json.encoder.JSONEncoder(sort_keys=True)
    values = [encoder.encode(obj) for obj in CORPUS]
    values += [repr(decoder.decode(encoder.encode(obj))) for obj in CORPUS]
    errors = []
    for text in BAD:
        try:
            decoder.decode(text)
            errors.append((text, None, "accepted"))
        except Exception as exc:  # noqa: BLE001
            errors.append((text, type(exc).__name__, str(exc)))
    return values, errors, type(decoder.scan_once).__module__


c_values, c_errors, c_scanner = snapshot()

# Setting c_make_scanner alone does NOT switch the decoder: json/scanner.py runs
# `make_scanner = c_make_scanner or py_make_scanner` at import time, so the name
# is already bound. My first measurement did exactly this and compared C with C.
json.scanner.c_make_scanner = None
_, _, still = snapshot()
report("clearing c_make_scanner alone does not switch the scanner", still == c_scanner,
       f"still {still} — the `or` already ran at import time")

json.scanner.make_scanner = json.scanner.py_make_scanner
json.encoder.c_make_encoder = None
json.decoder.scanstring = json.decoder.py_scanstring
json.encoder.encode_basestring_ascii = json.encoder.py_encode_basestring_ascii
json.encoder.encode_basestring = json.encoder.py_encode_basestring
py_values, py_errors, py_scanner = snapshot()

report("rebinding make_scanner does switch it", py_scanner != c_scanner,
       f"{c_scanner} -> {py_scanner}")
report("values are byte-identical with and without the accelerator",
       c_values == py_values, f"{len(c_values)} strings compared")

same_class = [c for c, p in zip(c_errors, py_errors) if c[1] == p[1]]
text_differs = [(c[0], c[2], p[2]) for c, p in zip(c_errors, py_errors) if c[1] == p[1] and c[2] != p[2]]
report("every input is accepted or rejected the same way",
       len(same_class) == len(BAD), f"{len(same_class)} of {len(BAD)} same error class")
report("and exactly one error MESSAGE differs", len(text_differs) == 1,
       "; ".join(f"{t!r}" for t, _, _ in text_differs))
for text, c, p in text_differs:
    print(f"        C : {c}")
    print(f"        py: {p}")
report("so 'the accelerator is not structural' is true only under a stated witness",
       c_values == py_values and len(text_differs) == 1,
       "same value: not structural. same message: observable.")

print("\n== 5. the checks can fail")
report("a comparison of two DIFFERENT implementations does find the difference",
       len(equivalence.compare({}, [42], "a", drive.run(importlib.import_module("TMS.impl.fast"), [42]),
                               "b", drive.run(importlib.import_module("TMS.impl.plain"), [42]))
           ["differing_error_text"]) == 1,
       "evaluated, not asserted")
report("the contract clause for error text can be made binding",
       any("MAY differ" not in label for label, _, _ in
           equivalence.verdict({}, {"differing_values": [], "identical_values": 0,
                                    "same_error_class": 0, "differing_error_class": [],
                                    "acceptance_differs": [], "differing_error_text": [("x", "a", "b")]},
                               True)),
       "policy.error_text_must_match() would turn today's pass into a failure")
report("this deployment does not require matching text", not policy.error_text_must_match())

print()
if FAILURES:
    print(f"  {len(FAILURES)} check(s) failed: {', '.join(FAILURES)}")
    raise SystemExit(1)
print("  island test passed")

"""Compare the two implementations against the written contract.

    python src/main.py
"""
import json
import pathlib
import sys

sys.path.insert(0, str(pathlib.Path(__file__).parent))

from DMS import equivalence  # noqa: E402
from SCL import policy  # noqa: E402
from SMS import drive, select  # noqa: E402

CORPUS = [
    "plain",
    'quote " and backslash \\',
    "tab\there",
    "newline\nhere",
    "control\x00\x1f",
    "unicode héllo 中文",
    "",
    42,          # not a str: both must raise
    None,        # not a str: both must raise
]


def main():
    manifest = json.loads(
        (pathlib.Path(__file__).parent / "FMS" / "manifest.json").read_text(encoding="utf-8"))
    contract = manifest["equivalence_contract"]

    print(f"\n  available implementations : {', '.join(select.available())}")
    print(f"  SCL allows the accelerator: {policy.accelerator_allowed()}")

    fast_name, fast = select.choose(prefer_fast=True, allowed=policy.accelerator_allowed())
    plain_name, plain = select.choose(prefer_fast=False, allowed=policy.accelerator_allowed())
    print(f"  chosen with preference    : {fast_name}")
    print(f"  chosen without            : {plain_name}")

    if fast_name == plain_name:
        print("\n  REFUSING to report an equivalence: both sides are the same implementation.")
        print("  This is the check my first measurement of CPython did not have, and it")
        print("  reported the C scanner identical to itself.")
        return 1

    findings = equivalence.compare(
        contract, CORPUS,
        fast_name, drive.run(fast, CORPUS),
        plain_name, drive.run(plain, CORPUS),
    )
    clauses = equivalence.verdict(contract, findings, policy.error_text_must_match())
    sys.stdout.write(equivalence.render(fast_name, plain_name, findings, clauses))
    return 0 if all(ok for _, ok, _ in clauses) else 2


if __name__ == "__main__":
    raise SystemExit(main())