Python 交给 Node 的 JSON，怎样保住编号与整数合同

Python 工具查到一条记录，Node 服务负责展示确认页。两边都能读写 JSON，字段名也完全一样，订单号仍可能在中途改变。把 9007199254740993 作为 JSON 数字发给 Node，再序列化回 Python，本次实验收到的是 9007199254740992。这一轮没有语法错误，也没有抛出异常。

这里用一份虚构的订单预览记录，把这个边界缩到可以完整复制的大小：四个字段、两个原生子进程、一份共享的原始文本夹具。最终交付的是能在 Python 和 Node 两端运行的同一份线格式合同。示例不调用模型、Agent SDK、网络接口或真实订单系统，也没有支付操作。

已有 JSON Schema 或运行时类型校验的项目，可以把这个实验放在反序列化入口旁边。它检查的是字段变成宿主语言的值之前和之后分别发生了什么：编号是否丢位、数字写法是否被抹平、重复键是否已经覆盖。只检查解析后的对象，可能已经错过了需要拒绝的证据。

order_id 在本例中只用来辨认记录，不参与算术，因此把它定义为 1 到 32 位 ASCII 数字字符串。"0009007199254740993" 中的三个前导零也是身份的一部分。不要先把编号读成 JavaScript Number，再调用 String() 补救；舍入已经发生，字符串化无法找回原来的那一位。

金额字段叫 amount_cents，单位在字段名里固定为分，使用 0 到 1,000,000,000 的整数。这个上限只是实验的业务约束，刻意小于 JavaScript 安全整数上限；它没有提供真实业务金额政策，也没有涵盖税费、汇率、分摊或累计求和。币种固定为 "CNY"，不能只换币种字符串就假设其他货币采用同一规则。

ECMAScript 的安全整数定义解释了为什么边界要另行检查。本次探针里，2⁵³ 和 2⁵³ + 2 都能原值返回，但 Number.isSafeInteger() 为 false。不能把“超出安全整数范围”写成“每一个数都会立即改变”。真正危险的是相邻整数不再都能区别：本例的 2⁵³ + 1 已经与另一个值合并。

合法记录如下。字段顺序和 JSON 允许的空白不影响结果；输出不承诺与输入逐字节相同。

{"version":1,"order_id":"0009007199254740993","amount_cents":12345,"currency":"CNY"}

• version：JSON 整数 token，值只能是 1
• order_id：解码后的字符串匹配 [0-9]{1,32}；全角数字不接受
• amount_cents：JSON 整数 token，值在 0 到 1,000,000,000 之间，含端点；不接受布尔值或数字字符串
• currency：字符串 "CNY"
• 恰好这四个字段，不允许缺少、未知字段、重复的解码后键名或嵌套值
• 输入文本重新编码为 UTF-8 后最多 4096 字节；CLI 先读完整个标准输入，这不是流式内存限制

“整数 token”还约束写法：允许 0、12345，拒绝 1.0 和 1e2，即使后两者在数学上能表示整数。JSON 语法允许这些写法，拒绝它们是本协议的额外决定。-0 被视为零，输出可以规范化成 0；订单号字符串里的前导零则不能删掉。JSON 字符串的合法转义允许使用，比较依据是解码后的字符。

单看 Number.isSafeInteger(JSON.parse("1.00000000000000001"))，本次 Node 返回 true。原因是 JSON.parse 先把它舍入成 Number 1，然后校验的已经是这个整数。因此，若合同要求输入文本必须使用整数写法，不能仅靠解析后的安全整数判断。

还有两个容易混淆的入口。Python 默认 json.dumps() 可以产生 NaN、Infinity 和 -Infinity；本次 Node 的 JSON.parse 对三者都抛出 SyntaxError。Python 出口设置 allow_nan=False 会先拒绝这些值。反方向，Node 默认 JSON.stringify 会把非有限 Number 写成 null，直接序列化未定制的 BigInt 则抛出 TypeError。已有 replacer 或 toJSON 的程序可能改变这个默认行为，探针没有使用它们。

1e400 又不同：它是合法的 JSON 数字 token，在这次两个运行时的默认解析中都成为无穷值；Node 再输出时变成 null。Python 的 parse_constant 针对 NaN 等特殊常量，不能单独拦住这样的指数溢出。本文直接通过 parse_float 拒绝所有小数点或指数数字 token，Node 则检查保留下来的原文。完整探针放在后文，可重新运行这些例子。

{"amount_cents":1,"amount_cents":2} 在两端默认解析后只留下 2。把第二个键写成 "amount_\u0063ents" 也一样：两个拼写解码后是同一个名字。RFC 8259 第 4 节建议对象成员名唯一，并指出重复成员的接收行为存在互操作差异。本文将它收紧为必须拒绝。

Python 用 object_pairs_hook 接收键值对序列，在构造字典前检测重复。Node 普通 reviver 的访问发生得太晚，探针中它只看到最后的 amount_cents = 2，不能从剩余对象推回被覆盖的历史。

本次安装的 Node v24.19.0 确实向 primitive reviver 提供了 context.source，探针读到了原始数字文本 9007199254740993。这个能力可以帮助检查仍保留的数值写法，但不能补回已被覆盖的重复键。因此本例没有另加一套依赖它的版本分支：先让内置 JSON.parse 检查整个文本语法，再用有限的扁平 token 扫描检查键出现次数及数字写法。当前 ECMAScript 在线规范是持续更新的文档，不能拿它的版本标签当成任意 Node 版本的实测结论。

这个扫描只接受四字段扁平对象，不递归，不负责修复错误 JSON，也不是可以提取出来用于任意输入的通用解析库。遇到嵌套值就拒绝。生产系统如果需要完整 JSON 功能，应选用经过审查、明确支持所需 token 和重复键策略的实现，再跑自己的兼容性与安全测试；不要把这几十行教学扫描扩充成通用库。

把下面两个文件保存到同一个空目录。Python 使用标准库；Node 只使用内置的 node:fs。程序从 stdin 读一份记录，成功时向 stdout 写一份 JSON，拒绝时退出码为 2 且不输出成功记录。两端的错误文案没有被定义为公共协议，测试只比较接受与拒绝。

Python 使用精确类型判断，避免 bool 作为 int 子类混入整数。输出的 allow_nan=False 是出口的额外检查；核心仍是输入合同。Node 的正则扫描必须放在完整 JSON.parse 成功之后：语法权威仍是内置解析器，不能把这条正则当成单独的合法性判断。

contract.py

"""A flat teaching wire contract; not a general untrusted-stream decoder."""
import json
import re
import sys

KEYS = {"version", "order_id", "amount_cents", "currency"}


def reject_number(token):
    raise ValueError("non_integer_number")


def unique_pairs(items):
    result = {}
    for key, value in items:
        if key in result:
            raise ValueError("duplicate_key")
        result[key] = value
    return result


def accept(raw):
    if len(raw.encode("utf-8")) > 4096:
        raise ValueError("too_large")
    data = json.loads(raw, object_pairs_hook=unique_pairs,
                      parse_float=reject_number, parse_constant=reject_number)
    if type(data) is not dict or set(data) != KEYS:
        raise ValueError("keys")
    if type(data["version"]) is not int or data["version"] != 1:
        raise ValueError("version")
    if (type(data["order_id"]) is not str
            or re.fullmatch(r"[0-9]{1,32}", data["order_id"]) is None):
        raise ValueError("id")
    if (type(data["amount_cents"]) is not int
            or not 0 <= data["amount_cents"] <= 1_000_000_000):
        raise ValueError("amount")
    if data["currency"] != "CNY":
        raise ValueError("currency")
    return data


if __name__ == "__main__":
    try:
        print(json.dumps(accept(sys.stdin.read()), allow_nan=False,
                         separators=(",", ":")))
    except (ValueError, RecursionError):
        sys.exit(2)

contract.cjs

// Flat four-field teaching protocol. JSON.parse remains the syntax authority.
const fs = require('node:fs');

function accept(raw) {
  if (Buffer.byteLength(raw, 'utf8') > 4096) throw Error('too_large');
  const data = JSON.parse(raw);
  if (data === null || Array.isArray(data) || typeof data !== 'object') {
    throw Error('object');
  }
  // Only run after JSON.parse accepted the complete input.
  // Retain number spellings and every key occurrence; reject nested values.
  const tokens = raw.match(/"(?:\\.|[^"\\])*"|-?\d+(?:\.\d+)?(?:[eE][+-]?\d+)?|true|false|null|[{}\[\]:,]/g);
  let i = 0;
  const seen = new Set();
  if (tokens[i++] !== '{') throw Error('object');
  while (tokens[i] !== '}') {
    const key = JSON.parse(tokens[i++]);
    if (typeof key !== 'string' || seen.has(key)) throw Error('duplicate_key');
    seen.add(key);
    if (tokens[i++] !== ':') throw Error('colon');
    const valueToken = tokens[i++];
    if (valueToken === undefined || ['{', '[', ']', '}', ',', ':'].includes(valueToken)) {
      throw Error('flat_only');
    }
    if (typeof JSON.parse(valueToken) === 'number' && !/^-?\d+$/.test(valueToken)) {
      throw Error('integer_token');
    }
    if (tokens[i] === ',') i++;
    else if (tokens[i] !== '}') throw Error('separator');
  }
  if (++i !== tokens.length) throw Error('trailing');
  if ([...seen].sort().join(',') !== 'amount_cents,currency,order_id,version') {
    throw Error('keys');
  }
  if (data.version !== 1) throw Error('version');
  if (typeof data.order_id !== 'string' || data.order_id.length < 1
      || data.order_id.length > 32 || /[^0-9]/.test(data.order_id)) {
    throw Error('id');
  }
  if (!Number.isSafeInteger(data.amount_cents) || data.amount_cents < 0
      || data.amount_cents > 1000000000) throw Error('amount');
  if (data.currency !== 'CNY') throw Error('currency');
  return data;
}

try {
  process.stdout.write(JSON.stringify(accept(fs.readFileSync(0, 'utf8'))));
} catch (error) {
  process.exitCode = 2;
}

夹具保存 raw 字符串，而不是先解析好的业务对象。这样重复键、指数写法和尾随垃圾都能原样进入两端入口。下面的 27 条记录是固定输入；accept 是事先写下的预期决定。另一个 expected.json 固定样本数量、合法样本名和往返计数，测试不会根据实现当前的输出自动生成期望。

四条合法样本分别是长字符串编号、金额下界、金额上界以及字段重排。其余样本检查数字编号、非 ASCII 编号、错误类型、越界、非整数数字写法、非有限常量、重复键、额外字段、嵌套值、缺字段和尾随文本。这个集合不是完整的边界覆盖清单；不要把样本数量等同于覆盖率。另附 test_edges.py，补查 -0、ID 末尾的四种换行字符、版本布尔值与错误版本、币种、32 位 ID、转义数字，以及 4096/4097 字节文本边界。12 个追加样本分别在两端运行，共 24 次判定；它们不计入主夹具的 54 次。

golden-fixtures.json

[
  {
    "name": "long_string_id",
    "raw": "{\"version\": 1, \"order_id\": \"0009007199254740993\", \"amount_cents\": 12345, \"currency\": \"CNY\"}",
    "accept": true
  },
  {
    "name": "amount_0",
    "raw": "{\"version\": 1, \"order_id\": \"0009007199254740993\", \"amount_cents\": 0, \"currency\": \"CNY\"}",
    "accept": true
  },
  {
    "name": "amount_1000000000",
    "raw": "{\"version\": 1, \"order_id\": \"0009007199254740993\", \"amount_cents\": 1000000000, \"currency\": \"CNY\"}",
    "accept": true
  },
  {
    "name": "numeric_id",
    "raw": "{\"version\": 1, \"order_id\": 9007199254740993, \"amount_cents\": 12345, \"currency\": \"CNY\"}",
    "accept": false
  },
  {
    "name": "empty_id",
    "raw": "{\"version\": 1, \"order_id\": \"\", \"amount_cents\": 12345, \"currency\": \"CNY\"}",
    "accept": false
  },
  {
    "name": "unicode_digits",
    "raw": "{\"version\": 1, \"order_id\": \"\\uff11\\uff12\\uff13\", \"amount_cents\": 12345, \"currency\": \"CNY\"}",
    "accept": false
  },
  {
    "name": "long_id",
    "raw": "{\"version\": 1, \"order_id\": \"111111111111111111111111111111111\", \"amount_cents\": 12345, \"currency\": \"CNY\"}",
    "accept": false
  },
  {
    "name": "over_amount",
    "raw": "{\"version\": 1, \"order_id\": \"0009007199254740993\", \"amount_cents\": 1000000001, \"currency\": \"CNY\"}",
    "accept": false
  },
  {
    "name": "unsafe_amount",
    "raw": "{\"version\": 1, \"order_id\": \"0009007199254740993\", \"amount_cents\": 9007199254740993, \"currency\": \"CNY\"}",
    "accept": false
  },
  {
    "name": "negative_amount",
    "raw": "{\"version\": 1, \"order_id\": \"0009007199254740993\", \"amount_cents\": -1, \"currency\": \"CNY\"}",
    "accept": false
  },
  {
    "name": "bool_amount",
    "raw": "{\"version\": 1, \"order_id\": \"0009007199254740993\", \"amount_cents\": true, \"currency\": \"CNY\"}",
    "accept": false
  },
  {
    "name": "null_amount",
    "raw": "{\"version\": 1, \"order_id\": \"0009007199254740993\", \"amount_cents\": null, \"currency\": \"CNY\"}",
    "accept": false
  },
  {
    "name": "string_amount",
    "raw": "{\"version\": 1, \"order_id\": \"0009007199254740993\", \"amount_cents\": \"12345\", \"currency\": \"CNY\"}",
    "accept": false
  },
  {
    "name": "fraction_amount",
    "raw": "{\"version\": 1, \"order_id\": \"0009007199254740993\", \"amount_cents\": 1.25, \"currency\": \"CNY\"}",
    "accept": false
  },
  {
    "name": "nan",
    "raw": "{\"version\": 1, \"order_id\": \"0009007199254740993\", \"amount_cents\": NaN, \"currency\": \"CNY\"}",
    "accept": false
  },
  {
    "name": "infinity",
    "raw": "{\"version\": 1, \"order_id\": \"0009007199254740993\", \"amount_cents\": Infinity, \"currency\": \"CNY\"}",
    "accept": false
  },
  {
    "name": "exp_overflow",
    "raw": "{\"version\": 1, \"order_id\": \"0009007199254740993\", \"amount_cents\": 1e400, \"currency\": \"CNY\"}",
    "accept": false
  },
  {
    "name": "rounded_fraction",
    "raw": "{\"version\": 1, \"order_id\": \"0009007199254740993\", \"amount_cents\": 1.00000000000000001, \"currency\": \"CNY\"}",
    "accept": false
  },
  {
    "name": "integral_float",
    "raw": "{\"version\": 1, \"order_id\": \"0009007199254740993\", \"amount_cents\": 1.0, \"currency\": \"CNY\"}",
    "accept": false
  },
  {
    "name": "exponent_integer",
    "raw": "{\"version\": 1, \"order_id\": \"0009007199254740993\", \"amount_cents\": 1e2, \"currency\": \"CNY\"}",
    "accept": false
  },
  {
    "name": "duplicate",
    "raw": "{\"version\": 1, \"order_id\": \"0009007199254740993\", \"amount_cents\": 12345, \"currency\": \"CNY\",\"amount_cents\":2}",
    "accept": false
  },
  {
    "name": "escaped_duplicate",
    "raw": "{\"version\": 1, \"order_id\": \"0009007199254740993\", \"amount_cents\": 12345, \"currency\": \"CNY\",\"amount_\\u0063ents\":2}",
    "accept": false
  },
  {
    "name": "extra",
    "raw": "{\"version\": 1, \"order_id\": \"0009007199254740993\", \"amount_cents\": 12345, \"currency\": \"CNY\",\"x\":2}",
    "accept": false
  },
  {
    "name": "nested_duplicate",
    "raw": "{\"version\": 1, \"order_id\": \"0009007199254740993\", \"amount_cents\": 12345, \"currency\": \"CNY\",\"x\":{\"a\":1,\"a\":2}}",
    "accept": false
  },
  {
    "name": "missing",
    "raw": "{\"version\": 1, \"order_id\": \"0009007199254740993\", \"amount_cents\": 12345}",
    "accept": false
  },
  {
    "name": "trailing",
    "raw": "{\"version\": 1, \"order_id\": \"0009007199254740993\", \"amount_cents\": 12345, \"currency\": \"CNY\"} garbage",
    "accept": false
  },
  {
    "name": "reordered",
    "raw": "{\"currency\": \"CNY\", \"amount_cents\": 12345, \"order_id\": \"0009007199254740993\", \"version\": 1}",
    "accept": true
  }
]

expected.json

{
  "fixtures": 27,
  "acceptance_checks": 54,
  "accepted_names": ["long_string_id", "amount_0", "amount_1000000000", "reordered"],
  "semantic_roundtrips": 8
}

test_protocol.py

"""Read shared raw fixtures, then run both native CLI implementations."""
import json
import platform
import subprocess
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parent


def require(condition, message):
    if not condition:
        raise RuntimeError(message)


def run(command, raw):
    result = subprocess.run(command, input=raw, text=True, encoding="utf-8",
                            capture_output=True, timeout=5)
    require(result.returncode in (0, 2), f"unexpected exit: {command}: {result}")
    return result


def main():
    cases = json.loads((ROOT / "golden-fixtures.json").read_text(encoding="utf-8"))
    expected = json.loads((ROOT / "expected.json").read_text(encoding="utf-8"))
    require(len(cases) == expected["fixtures"], "fixture count")
    require([c["name"] for c in cases if c["accept"]] == expected["accepted_names"],
            "accepted fixture names")
    commands = {"python": [sys.executable, str(ROOT / "contract.py")],
                "node": ["node", str(ROOT / "contract.cjs")]}
    checks, roundtrips = [], []
    for case in cases:
        for runtime, command in commands.items():
            first = run(command, case["raw"])
            require((first.returncode == 0) == case["accept"],
                    f"acceptance mismatch: {case['name']}: {runtime}")
            checks.append({"fixture": case["name"], "runtime": runtime,
                           "accepted": first.returncode == 0, "pass": True})
            if not case["accept"]:
                require(first.stdout == "", "rejected input emitted a payload")
                continue
            other = "node" if runtime == "python" else "python"
            second = run(commands[other], first.stdout)
            require(second.returncode == 0, "roundtrip rejected")
            require(json.loads(first.stdout) == json.loads(case["raw"]), "first changed value")
            require(json.loads(second.stdout) == json.loads(case["raw"]), "second changed value")
            roundtrips.append({"fixture": case["name"],
                               "direction": f"{runtime}->{other}", "pass": True})
    require(len(checks) == expected["acceptance_checks"], "check count")
    require(len(roundtrips) == expected["semantic_roundtrips"], "roundtrip count")
    report = {"python": platform.python_version(),
              "node": subprocess.check_output(["node", "--version"], text=True).strip(),
              "checks": checks, "roundtrips": roundtrips}
    (ROOT / "protocol-results.json").write_text(
        json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    print(f"{len(cases)} shared fixtures; {len(checks)} acceptance checks passed; "
          f"{len(roundtrips)} semantic roundtrips passed")


if __name__ == "__main__":
    main()

test_edges.py

"""Small additional edge regression, separate from the 27 golden fixtures."""
import json
from test_protocol import ROOT, require, run
import sys

base = {"version": 1, "order_id": "1", "amount_cents": 0, "currency": "CNY"}
cases = []
for suffix in ("\n", "\r", "\u2028", "\u2029"):
    cases.append(("id_suffix_" + hex(ord(suffix)),
                  json.dumps(base | {"order_id": "1" + suffix}), False))
for version in (True, 2):
    cases.append(("version_" + str(version), json.dumps(base | {"version": version}), False))
cases.append(("negative_zero", json.dumps(base).replace('"amount_cents": 0',
                                                        '"amount_cents": -0'), True))
cases.append(("wrong_currency", json.dumps(base | {"currency": "USD"}), False))
cases.append(("id_32_digits", json.dumps(base | {"order_id": "0" * 32}), True))
cases.append(("escaped_digit", json.dumps(base).replace('"order_id": "1"',
                                                       '"order_id": "\\u0031"'), True))
raw = json.dumps(base)
cases.append(("bytes_4096", raw + " " * (4096 - len(raw)), True))
cases.append(("bytes_4097", raw + " " * (4097 - len(raw)), False))
for name, raw, expected in cases:
    for command in ([sys.executable, str(ROOT / "contract.py")],
                    ["node", str(ROOT / "contract.cjs")]):
        result = run(command, raw)
        require((result.returncode == 0) == expected, name)
        if expected:
            require(json.loads(result.stdout) == json.loads(raw), name + " changed value")
        else:
            require(result.stdout == "", name + " rejection emitted payload")
print(f"{len(cases)} additional edge fixtures; {len(cases) * 2} acceptance checks passed")

复制前面的六个文件：contract.py、contract.cjs、golden-fixtures.json、expected.json、test_protocol.py 和 test_edges.py。确保 Python 与 Node 已在 PATH 中；本例不需要 pip 或 npm 安装。用以下命令运行：

python --version
node --version
python test_protocol.py
python test_edges.py

本次真实环境是 Python 3.12.14 与 Node v24.19.0，标准输出为：

27 shared fixtures; 54 acceptance checks passed; 8 semantic roundtrips passed
12 additional edge fixtures; 24 acceptance checks passed

54 次判定是 27 条文本分别进入两个独立进程。8 次语义往返是四条合法记录分别走 Python → Node 和 Node → Python，检查每一站接受且字段值与原输入一致。它不比较属性顺序、空白或转义拼写，也不承诺字节级规范化。原始夹具含 accept = false 的样本不会被拿去做“成功回传”。

测试会写出 protocol-results.json，逐项列出运行时、样本与结果，并记录实际版本。它有单个子进程 5 秒的超时，异常退出不会被误算成预期拒绝。要将它接入自己的工程，先把真实业务决定写成新夹具，例如版本演进、负数是否允许、金额上限和空值语义，再分别运行两端。不能为了让当前实现通过而改掉期望。

下面的 probe.py 独立于严格合同，故意使用两端默认 JSON 行为，展示问题发生的位置。它会生成 results.json，内容包括整数边界、前导零字符串、三个非有限值、普通与转义重复键、指数溢出、解析后的舍入小数，以及当前 Node 的 reviver source 探测。文件中全部数据为合成样本。

python probe.py

probe.py

"""Synthetic local cross-process probes. No network, SDK or real data."""
import json, subprocess, sys, platform
from pathlib import Path
ROOT=Path(__file__).resolve().parent
NODE=r'''const fs=require('node:fs');const raw=fs.readFileSync(0,'utf8');try{const v=JSON.parse(raw);process.stdout.write(JSON.stringify(v));}catch(e){process.stderr.write(e.name);process.exit(2);}'''
def node(raw):
 p=subprocess.run(['node','-e',NODE],input=raw,text=True,capture_output=True,check=False)
 return {'returncode':p.returncode,'wire':p.stdout,'error':p.stderr}
results={'python':platform.python_version(),'node':subprocess.check_output(['node','--version'],text=True).strip(),'python_to_node_to_python':[]}
for n in [2**53-1,2**53,2**53+1,2**53+2,-(2**53+1)]:
 raw=json.dumps({'order_id':n,'amount_cents':n},separators=(',',':'))
 r=node(raw);back=json.loads(r['wire']);results['python_to_node_to_python'].append({'sent':n,'node_wire':r['wire'],'received':back['order_id'],'equal':n==back['order_id']})
results['id_string_roundtrip']=node(json.dumps({'order_id':'0009007199254740993','amount_cents':12345}))
results['python_nonfinite']=[]
for x in [float('nan'),float('inf'),float('-inf')]:
 raw=json.dumps({'amount_cents':x})
 try: json.dumps({'amount_cents':x},allow_nan=False); strict='accepted'
 except ValueError: strict='ValueError'
 results['python_nonfinite'].append({'python_wire':raw,'node':node(raw),'allow_nan_false':strict})
results['duplicates']=[]
for raw in ['{"amount_cents":1,"amount_cents":2}','{"amount_cents":1,"amount_\\u0063ents":2}']:
 results['duplicates'].append({'raw':raw,'python':json.loads(raw),'node':node(raw)})
results['valid_json_exponent_overflow']={'raw':'{"amount_cents":1e400}','python_is_infinite':json.loads('{"amount_cents":1e400}')['amount_cents']==float('inf'),'node':node('{"amount_cents":1e400}')}
js=r'''const r={};for(const [k,v] of [['nonfinite',{a:NaN,b:Infinity,c:-Infinity}],['bigint',{id:9007199254740993n}]]){try{r[k]=JSON.stringify(v)}catch(e){r[k]=e.name}}r.safe=[9007199254740991,9007199254740992,9007199254740994].map(n=>({value:n,safe:Number.isSafeInteger(n)}));r.postParseFraction={raw:'1.00000000000000001',value:JSON.parse('1.00000000000000001'),safe:Number.isSafeInteger(JSON.parse('1.00000000000000001'))};let seen=[];JSON.parse('{"amount_cents":1,"amount_cents":2}',(k,v)=>{if(k)seen.push([k,v]);return v;});r.reviverSees=seen;let source;JSON.parse('9007199254740993',(k,v,c)=>{source=c?.source;return v});r.reviverSource=source;process.stdout.write(JSON.stringify(r));'''
results['node_to_python']=json.loads(subprocess.check_output(['node','-e',js],text=True))
text=json.dumps(results,indent=2,allow_nan=False);(ROOT/'results.json').write_text(text+'\n');print(text)

这次只验证已列出的原始夹具和默认行为探针。没有做 fuzz 测试、恶意字节输入审计、非法 UTF-8 一致性检查、流式大小限制、嵌套协议、多版本 Node/Python 矩阵、浏览器运行时或真实 Agent SDK 集成。两个 CLI 都先读完整个 stdin，4096 字节检查不能阻止读入阶段占用过多内存。本文实现不应直接作为不可信网络入口的安全边界。

把它接到 Agent 工具层时，合理的位置是在外部文本进入业务对象的那一层。工具参数校验仍然有用，身份授权、业务权限、执行审批和幂等控制也仍需各自实现。这里成功通过合同，只说明这份预览记录的表示符合约定，并不授权创建订单或付款。

正文和三张解释图可离线阅读，图中没有产品界面截图。测试结果来自实际原生进程；全文网页排版没有经过浏览器 GUI 验证。把验证范围留在结果旁边，下一位维护者才能知道哪些承诺已经有证据，哪些仍需要补测。

参考资料

• RFC 8259：对象成员名与数字的互操作范围
• Python 3.12 json：非有限值、重复名字与解码钩子。查阅页当前标记 3.12.15，本次执行环境为 3.12.14
• ECMAScript：JSON 序列化算法
• ECMAScript：JSON.parse 与 reviver
• ECMAScript：Number.MAX_SAFE_INTEGER

资料核对日期：2026-10-07。在线规范会继续更新，运行结果以本文记录的版本为准。

前面的实现先读完整个 stdin，再检查文本长度。2026-10-13 的补充实验保留原四字段合同，在它前面增加一个按原始字节计数的入口：最多累计 4097 个字节，读到第 4097 个就拒绝。这一步可以在发送者尚未关闭写端时完成。

实验先启动原来的两个 CLI，向各自的管道写入 4097 字节并保留写端。Python 与 Node 在 1 秒观察窗内都仍在运行；关闭写端后，两者均以 2 退出。换用新增入口，相同输入在写端仍打开时已经以 2 退出，stdout 为空。另送一份恰好 4096 字节的合法记录，两个新增入口都继续等待，收到 EOF 后才输出记录并以 0 退出。

这组结果补充了读取阶段的行为。原文的编号、整数 token、重复键和字段值域规则继续生效；上面的原实现与原验证范围也保留，方便对照。所有输入均为本地合成字节，没有连接模型、网络服务或真实订单系统。

假设刚读到 4096 字节，其中已经有完整 JSON 和尾部空格，发送者仍可能继续写第 4097 个字节。只读满上限就交给解析器，会把一个超限输入的合法前缀当成完整记录。额外的一个字节是超限证据：读到它即可拒绝；读不到它且收到 EOF，才能确定完整输入没有超过上限。

循环每次只请求 4097 − 已累计字节数。单次读取返回得少，不代表流已经结束，下一轮继续读。Python 的 os.read最多返回请求数量的字节，到 EOF 返回空 bytes；Node 的 fs.readSync返回实际读取数量。代码据此区分短读和结束，不根据一块数据是否包含完整花括号提前接受。

本例处理的是“一次进程、一份有限记录”。未超限的输入必须有 EOF。发送者只写几字节就停住，或者写满 4096 字节却不关闭写端，入口都可以持续阻塞。测试驱动的 1 秒与 5 秒观察窗没有给 CLI 增加超时；需要总时限、空闲时限或取消能力的服务，必须另行设计并验证。

累计 4095 字节：下次最多请求 2 字节
累计 4096 字节：下次最多请求 1 字节
收到第 4097 字节：拒绝，不再读取
不足或恰好 4096 字节且收到 EOF：严格解码，再验收完整记录

新增两个小文件。Python 直接读取尚未消费过的文件描述符 0；Node 用一个 4097 字节 Buffer 承接同步读取。这两个入口都要求独占新打开的 stdin，不能先用 sys.stdin、sys.stdin.buffer 或 process.stdin 读走一部分，再切换到它们。注入的 read 参数只供受控测试使用，须遵守相同的读取数量约定。

wire_in.py

"""Finite, blocking, single-record stdin gate; no deadline or general parser."""
import os

LIMIT = 4096


def read_utf8(read=None):
    # Use only on a fresh descriptor: never mix with earlier text/buffer reads.
    if read is None:
        read = lambda count: os.read(0, count)
    raw = bytearray()
    while len(raw) <= LIMIT:
        part = read(LIMIT + 1 - len(raw))
        if not part:
            break
        raw.extend(part)
    if len(raw) > LIMIT:
        raise ValueError("wire_too_large")
    if raw.startswith(b"\xef\xbb\xbf"):
        raise ValueError("wire_bom")
    return raw.decode("utf-8", errors="strict")

wire-in.cjs

// Finite, blocking, single-record stdin gate. No deadline or general parser.
const fs = require('node:fs');
const { TextDecoder } = require('node:util');
const LIMIT = 4096;

function readUtf8(read = (buffer, offset, count) =>
  fs.readSync(0, buffer, offset, count, null)) {
  // Use only on a fresh descriptor: never mix with process.stdin consumption.
  const buffer = Buffer.alloc(LIMIT + 1);
  let used = 0;
  while (used <= LIMIT) {
    const count = read(buffer, used, LIMIT + 1 - used);
    if (count === 0) break;
    used += count;
  }
  if (used > LIMIT) throw Error('wire_too_large');
  const raw = buffer.subarray(0, used);
  if (raw.length >= 3 && raw[0] === 0xef && raw[1] === 0xbb && raw[2] === 0xbf) {
    throw Error('wire_bom');
  }
  return new TextDecoder('utf-8', { fatal: true, ignoreBOM: true }).decode(raw);
}

module.exports = { readUtf8 };

长度检查先于解码，随后拒绝开头的 UTF-8 BOM，再严格解码整个已收齐的字节串。Python 使用 strict；Node 使用 fatal: true。Node 的 ignoreBOM: true 表示解码结果保留 BOM，名称容易让人误读。这里已在字节层主动拒绝 BOM，因此不会把它悄悄删掉后继续验收。fatal 支持还取决于 Node 的 ICU 构建，换用其他构建应重新跑非法字节夹具。该选项的含义见 TextDecoder 文档。

RFC 8259 第 8.1 节要求跨系统 JSON 使用 UTF-8，并允许解析方选择忽略 BOM。本节明确选择拒绝开头的 BOM；这是这一份合同的策略。空输入、UTF-16LE 字节与第二份连续 JSON 文档也不会被接纳。某些字节能严格解码为文本，后面的 JSON 或字段检查仍会拒绝它们。

下面是接好入口的两个完整 CLI。Python 只增加导入并替换 stdin 调用；Node 替换导入与 stdin 调用。accept 及其辅助函数逐字保留，输出形式也相同：Python 成功结果末尾有换行，Node 没有；本节的预期拒绝均为退出码 2、stdout 与 stderr 为空。

"""A flat teaching wire contract; not a general untrusted-stream decoder."""
import json
import re
import sys
from wire_in import read_utf8

KEYS = {"version", "order_id", "amount_cents", "currency"}


def reject_number(token):
    raise ValueError("non_integer_number")


def unique_pairs(items):
    result = {}
    for key, value in items:
        if key in result:
            raise ValueError("duplicate_key")
        result[key] = value
    return result


def accept(raw):
    if len(raw.encode("utf-8")) > 4096:
        raise ValueError("too_large")
    data = json.loads(raw, object_pairs_hook=unique_pairs,
                      parse_float=reject_number, parse_constant=reject_number)
    if type(data) is not dict or set(data) != KEYS:
        raise ValueError("keys")
    if type(data["version"]) is not int or data["version"] != 1:
        raise ValueError("version")
    if (type(data["order_id"]) is not str
            or re.fullmatch(r"[0-9]{1,32}", data["order_id"]) is None):
        raise ValueError("id")
    if (type(data["amount_cents"]) is not int
            or not 0 <= data["amount_cents"] <= 1_000_000_000):
        raise ValueError("amount")
    if data["currency"] != "CNY":
        raise ValueError("currency")
    return data


if __name__ == "__main__":
    try:
        print(json.dumps(accept(read_utf8()), allow_nan=False,
                         separators=(",", ":")))
    except (ValueError, RecursionError):
        sys.exit(2)

// Flat four-field teaching protocol. JSON.parse remains the syntax authority.
const { readUtf8 } = require('./wire-in.cjs');

function accept(raw) {
  if (Buffer.byteLength(raw, 'utf8') > 4096) throw Error('too_large');
  const data = JSON.parse(raw);
  if (data === null || Array.isArray(data) || typeof data !== 'object') {
    throw Error('object');
  }
  // Only run after JSON.parse accepted the complete input.
  // Retain number spellings and every key occurrence; reject nested values.
  const tokens = raw.match(/"(?:\\.|[^"\\])*"|-?\d+(?:\.\d+)?(?:[eE][+-]?\d+)?|true|false|null|[{}\[\]:,]/g);
  let i = 0;
  const seen = new Set();
  if (tokens[i++] !== '{') throw Error('object');
  while (tokens[i] !== '}') {
    const key = JSON.parse(tokens[i++]);
    if (typeof key !== 'string' || seen.has(key)) throw Error('duplicate_key');
    seen.add(key);
    if (tokens[i++] !== ':') throw Error('colon');
    const valueToken = tokens[i++];
    if (valueToken === undefined || ['{', '[', ']', '}', ',', ':'].includes(valueToken)) {
      throw Error('flat_only');
    }
    if (typeof JSON.parse(valueToken) === 'number' && !/^-?\d+$/.test(valueToken)) {
      throw Error('integer_token');
    }
    if (tokens[i] === ',') i++;
    else if (tokens[i] !== '}') throw Error('separator');
  }
  if (++i !== tokens.length) throw Error('trailing');
  if ([...seen].sort().join(',') !== 'amount_cents,currency,order_id,version') {
    throw Error('keys');
  }
  if (data.version !== 1) throw Error('version');
  if (typeof data.order_id !== 'string' || data.order_id.length < 1
      || data.order_id.length > 32 || /[^0-9]/.test(data.order_id)) {
    throw Error('id');
  }
  if (!Number.isSafeInteger(data.amount_cents) || data.amount_cents < 0
      || data.amount_cents > 1000000000) throw Error('amount');
  if (data.currency !== 'CNY') throw Error('currency');
  return data;
}

try {
  process.stdout.write(JSON.stringify(accept(readUtf8())));
} catch (error) {
  process.exitCode = 2;
}

4097 限制的是该入口累计的原始载荷字节数，以及每次发给底层读函数的剩余额度。Node 此处显式分配 4097 字节，Python 还有 bytearray 容量、临时 bytes 和后续字符串；内核管道、运行时内部缓冲、解析对象与进程 RSS 都有额外成本。本节没有测量峰值 RSS，不能把这个数字当作整个进程的硬内存上限。

读函数异常会直接离开入口，绝不会作为 EOF 返回一段残缺文本。现有 Python CLI 只把 ValueError 和 RecursionError 归为退出码 2；底层 OSError 等运行故障仍可能有其他非零退出码和诊断。Node 保留原来的 catch。示例没有统一系统 I/O 错误协议，也没有验证关闭 stdout、权限故障或非阻塞描述符。接入方应保留异常证据，不能把所有非零退出一律统计成合同拒绝。

最小输入固定为 62 个 ASCII 字节。make_wire_fixtures.py 只负责构造输入，将每份字节串的 Base64 和字节数写入 wire-fixtures.json。wire-expected.json 独立列出接受、拒绝名称和精确成功输出，生成器不读实现输出，也不改期望。保存下列完整生成器并执行一次，就能得到表中每一个字节。

"""Deterministic raw bytes. Expected decisions live separately, not in this generator."""
import base64
import json
from pathlib import Path
ROOT = Path(__file__).resolve().parent
BASE = b'{"version":1,"order_id":"1","amount_cents":0,"currency":"CNY"}'
assert len(BASE) == 62

def padded(size, fill=b' '):
    n = size - len(BASE)
    return BASE + fill * (n // len(fill)) + b' ' * (n % len(fill))

cases = {
    'minimal': BASE,
    'ascii_4096': padded(4096),
    'ascii_4097': padded(4097),
    'crlf_4096': padded(4096, b'\r\n'),
    'crlf_4097': padded(4097, b'\r\n'),
    'large_finite': padded(100000),
    'utf8_bom': b'\xef\xbb\xbf' + BASE,
    'invalid_ff': BASE + b'\xff',
    'truncated_utf8': BASE + b'\xe4\xb8',
    'utf16le': BASE.decode('ascii').encode('utf-16-le'),
    'escaped_digit': BASE.replace(b'"1"', b'"\\u0031"'),
    'empty': b'',
    'two_documents': BASE + BASE,
    'multibyte_value': BASE.replace(b'"1"', '"一"'.encode('utf8')),
}
rows = [{'name':name, 'wire_base64':base64.b64encode(raw).decode('ascii'),
         'wire_bytes':len(raw)} for name,raw in cases.items()]
(ROOT/'wire-fixtures.json').write_text(json.dumps(rows, indent=2)+'\n')
print(f'{len(rows)} byte fixtures created')

wire-expected.json

{
  "limit": 4096,
  "accepted_names": ["minimal", "ascii_4096", "crlf_4096", "escaped_digit"],
  "rejected_names": ["ascii_4097", "crlf_4097", "large_finite", "utf8_bom", "invalid_ff", "truncated_utf8", "utf16le", "empty", "two_documents", "multibyte_value"],
  "success_json": "{\"version\":1,\"order_id\":\"1\",\"amount_cents\":0,\"currency\":\"CNY\"}",
  "gate_reject_names": ["ascii_4097", "crlf_4097", "large_finite", "utf8_bom", "invalid_ff", "truncated_utf8"],
  "max_copied_bytes": 4097,
  "fixture_count": 14,
  "acceptance_checks": 28,
  "gate_checks_per_runtime": 42,
  "held_open_4097": {"original": "pending_until_eof", "fixed": "rejected_before_eof"},
  "rejection_returncode": 2
}

单独测试字节入口时，empty、utf16le、two_documents、multibyte_value 都可以得到解码后的字符串；完整 CLI 随后按 JSON 语法或字段规则拒绝。这就是 gate_reject_names 与 rejected_names 分开的原因。不能让“严格 UTF-8 解码成功”替代完整合同。

CRLF 和非法 UTF-8 也曾作为旧入口可能误接受的假设进行核查。本机实际 sys.stdin 保留了 a\r\nb\rc\n 的 7 个字节；原 CLI 的 14 份字节输入、28 次判定与上述预期一致，没有发现这两类误接受。补充入口的直接证据来自超限后的退出时机，不能把已被实验否定的猜测写成旧代码漏洞。

"""Replay identical byte fixtures against fresh-GET inline sources without modifying them."""
import base64
import json
import subprocess
import sys
from pathlib import Path
ROOT=Path(__file__).resolve().parent
cases=json.loads((ROOT/'wire-fixtures.json').read_text())
expected=json.loads((ROOT/'wire-expected.json').read_text())
rows=[]
for case in cases:
    accepted=case['name'] in expected['accepted_names']
    for runtime,command in [('python',[sys.executable,str(ROOT/'original/contract.py')]),
                            ('node',['node',str(ROOT/'original/contract.cjs')])]:
        r=subprocess.run(command,input=base64.b64decode(case['wire_base64']),capture_output=True,timeout=5)
        wanted=(expected['success_json']+('\n' if runtime=='python' else '')).encode() if accepted else b''
        assert r.returncode==(0 if accepted else 2) and r.stdout==wanted and r.stderr==b'',(case['name'],runtime,r)
        rows.append({'name':case['name'],'runtime':runtime,'returncode':r.returncode,
                     'stdout':r.stdout.decode(),'stderr':r.stderr.decode()})
(ROOT/'original-byte-results.json').write_text(json.dumps(rows,indent=2)+'\n')
print('28 finite byte decisions match the fixed independent oracle in the unchanged original CLIs')

"""Record actual process stdin behavior, without changing stream configuration."""
import json
import subprocess
import sys
from pathlib import Path
ROOT=Path(__file__).resolve().parent
raw=b'a\r\nb\rc\n'
code='import sys,json; s=sys.stdin.read(); print(json.dumps({"text":s,"bytes_after_utf8_encode":len(s.encode("utf8")),"newlines":sys.stdin.newlines,"encoding":sys.stdin.encoding,"errors":sys.stdin.errors}))'
r=subprocess.run([sys.executable,'-c',code],input=raw,capture_output=True,timeout=5)
assert r.returncode==0 and not r.stderr
observation=json.loads(r.stdout)
assert observation['text']=='a\r\nb\rc\n' and observation['bytes_after_utf8_encode']==7
(ROOT/'stdin-newline-observation.json').write_text(json.dumps(observation,indent=2)+'\n')
print('Actual OS stdin preserves CRLF on this runtime; no newline-size divergence found')

{
  "text": "a\r\nb\rc\n",
  "bytes_after_utf8_encode": 7,
  "newlines": null,
  "encoding": "utf-8",
  "errors": "surrogateescape"
}

test_wire.py 先把 14 份 bytes 逐份送进 Python 与 Node 原生子进程，共 28 次判定。每次核对退出码、stdout 和 stderr；四种成功输入都必须输出 expected 中那一份规范化记录，拒绝输入不能有成功载荷。测试没有设置 text=True，因此非法字节和 CRLF 都原样进入子进程。

接着，两端分别注入每次最多返回 1、7、4097 字节的读取函数。14 × 3 × 2 共 84 次入口检查，逐次断言请求数量没有超过剩余额度、累计载荷不超过 4097，并检查严格解码与 BOM 策略。能通过入口的文本重新编码后须逐字节等于原输入。注入结果验证循环的分块逻辑；真实管道实验另行验证 EOF 前的退出行为。

test_wire.py

"""Raw bytes enter ordinary child stdin; fixed independent acceptance decisions."""
import base64
import json
import platform
import subprocess
import sys
from pathlib import Path
from wire_in import read_utf8

ROOT = Path(__file__).resolve().parent
expected = json.loads((ROOT/'wire-expected.json').read_text())
cases = json.loads((ROOT/'wire-fixtures.json').read_text())
assert len(cases) == expected['fixture_count']
assert set(c['name'] for c in cases) == set(expected['accepted_names'] + expected['rejected_names'])
commands = {'python': [sys.executable, str(ROOT/'contract.py')],
            'node': ['node', str(ROOT/'contract.cjs')]}
report = {'python': platform.python_version(),
          'node': subprocess.check_output(['node','--version'], text=True).strip(),
          'cli': [], 'python_gate': []}
for case in cases:
    raw = base64.b64decode(case['wire_base64'], validate=True)
    assert len(raw) == case['wire_bytes']
    accepted = case['name'] in expected['accepted_names']
    for runtime, command in commands.items():
        r = subprocess.run(command, input=raw, capture_output=True, timeout=5)
        want = (expected['success_json'] + ('\n' if runtime == 'python' else '')).encode() if accepted else b''
        assert r.returncode == (0 if accepted else 2), (case['name'], runtime, r)
        assert r.stdout == want and r.stderr == b'', (case['name'], runtime, r)
        report['cli'].append({'name':case['name'],'runtime':runtime,'returncode':r.returncode,
                              'stdout':r.stdout.decode(),'stderr':r.stderr.decode()})
    for step in [1, 7, 4097]:
        cursor, calls, requests = 0, 0, []
        def read(requested):
            global cursor, calls
            assert 0 < requested <= 4097 - cursor
            requests.append(requested); calls += 1
            part = raw[cursor:cursor+min(step, requested)]
            cursor += len(part)
            return part
        error = None
        try: result = read_utf8(read)
        except ValueError as exc: error = type(exc).__name__
        reject = case['name'] in expected['gate_reject_names']
        assert bool(error) == reject, case['name']
        assert cursor == min(len(raw), 4097), case['name']
        if not reject: assert result.encode('utf8') == raw
        report['python_gate'].append({'name':case['name'],'step':step,'copied_bytes':cursor,
                                      'calls':calls,'max_requested':max(requests),'rejected':reject})
report['node_gate'] = json.loads(subprocess.check_output(['node','test-gate.cjs'],cwd=ROOT))
assert report['node_gate'] == report['python_gate']
assert len(report['cli']) == expected['acceptance_checks']
assert len(report['python_gate']) == expected['gate_checks_per_runtime']
(ROOT/'wire-results.json').write_text(json.dumps(report,indent=2)+'\n')
print('14 raw byte fixtures; 28 CLI decisions; 84 injected-reader checks passed')

test-gate.cjs

const fs = require('node:fs');
const assert = require('node:assert/strict');
const { readUtf8 } = require('./wire-in.cjs');
const cases = JSON.parse(fs.readFileSync('wire-fixtures.json', 'utf8'));
const expected = JSON.parse(fs.readFileSync('wire-expected.json', 'utf8'));
const rows = [];
for (const c of cases) for (const step of [1, 7, 4097]) {
  const raw = Buffer.from(c.wire_base64, 'base64');
  let cursor = 0, calls = 0, maxRequested = 0, result, error;
  function read(buffer, offset, requested) {
    assert(requested > 0 && requested <= 4097 - cursor);
    maxRequested = Math.max(maxRequested, requested);
    calls++;
    const n = Math.min(step, requested, raw.length - cursor);
    raw.copy(buffer, offset, cursor, cursor + n); cursor += n;
    return n;
  }
  try { result = readUtf8(read); } catch (e) { error = e.name; }
  const rejected = expected.gate_reject_names.includes(c.name);
  assert.equal(Boolean(error), rejected, c.name);
  assert.equal(cursor, Math.min(raw.length, 4097), c.name + ' copied_bytes');
  if (!rejected) assert.deepEqual(Buffer.from(result, 'utf8'), raw, c.name + ' exact bytes');
  rows.push({ name: c.name, step, copied_bytes: cursor, calls, max_requested: maxRequested, rejected });
}
assert.equal(rows.length, expected.gate_checks_per_runtime);
process.stdout.write(JSON.stringify(rows, null, 2));

新增 Python 测试使用 assert，运行时不能加 -O。缺少文件、超时、非预期退出和错误输出都会使测试失败。原有 27 份共享文本夹具、54 次判定、8 次语义往返，以及另外 12 份边界输入、24 次判定，分别在原入口和新增入口上重跑；原 expected.json 与 golden-fixtures.json 保持原样。

probe_pipe.py 自己创建并持有子进程管道。它先检查本机管道容量足以装下至多 4097 字节，再循环写入精确输入，保留 stdin 写端，等待子进程状态。新增入口必须在 5 秒测试窗口内结束；原入口与 4096 字节合法输入则分别观察 1 秒仍运行。之后才关闭写端，读取完整 stdout、stderr 与最终退出码。

本次 6 个案例的父进程写端在观察时均保持打开；原入口的两个超限案例关闭后以 2 退出，新增入口的两个超限案例关闭前已以 2 退出，两个 4096 字节案例关闭后以 0 退出。1 秒未退出只是一个有界观察，结合原代码的整流读取可以解释等待原因；实验没有记录系统调用，也不承诺任意机器在固定毫秒内完成拒绝。

"""Owned local child processes only. No sleeps, tracer, environment overrides or network."""
import base64
import fcntl
import json
import os
import subprocess
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parent
cases = {r['name']:base64.b64decode(r['wire_base64'])
         for r in json.loads((ROOT/'wire-fixtures.json').read_text())}
expected = json.loads((ROOT/'wire-expected.json').read_text())


def probe(runtime, variant, fixture, before_eof):
    path = ROOT / ('original' if variant == 'original' else '') / ('contract.py' if runtime == 'python' else 'contract.cjs')
    command = [sys.executable if runtime == 'python' else 'node', str(path)]
    raw = cases[fixture]
    proc = subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                            stderr=subprocess.PIPE, bufsize=0)
    forced = False
    observed = None
    try:
        # This Linux-only probe supplies at most 4097 bytes, within this pipe capacity.
        capacity = fcntl.fcntl(proc.stdin.fileno(), fcntl.F_GETPIPE_SZ)
        assert capacity >= len(raw)
        sent = 0
        while sent < len(raw):
            sent += os.write(proc.stdin.fileno(), raw[sent:])
        assert sent == len(raw)
        # The writer remains open. A successful write means these exact bytes entered the pipe.
        window = 5 if before_eof == 'exited' else 1
        try:
            proc.wait(timeout=window)
            observed = 'exited'
        except subprocess.TimeoutExpired:
            observed = 'pending'
        assert observed == before_eof, (runtime, variant, fixture, observed)
        if observed == 'exited':
            assert proc.returncode == 2
        writer_was_open_at_observation = not proc.stdin.closed
        proc.stdin.close()
        proc.stdin = None
        stdout, stderr = proc.communicate(timeout=5)
        accepted = fixture == 'ascii_4096'
        wanted = (expected['success_json'] + ('\n' if runtime=='python' else '')).encode() if accepted else b''
        assert proc.returncode == (0 if accepted else 2)
        assert stdout == wanted and stderr == b''
        return {'runtime':runtime,'variant':variant,'fixture':fixture,
                'bytes_written':sent,'pipe_capacity':capacity,
                'observation_window_seconds':window,'before_parent_eof':observed,
                'writer_open_at_observation':writer_was_open_at_observation,
                'after_eof_returncode':proc.returncode,'stdout':stdout.decode(),
                'stderr':stderr.decode(),'forced_kill':forced}
    finally:
        if proc.stdin is not None:
            proc.stdin.close(); proc.stdin = None
        if proc.poll() is None:
            try: proc.communicate(timeout=5)
            except subprocess.TimeoutExpired:
                forced = True
                proc.kill(); proc.communicate(timeout=5)
        if proc.stdout is not None: proc.stdout.close()
        if proc.stderr is not None: proc.stderr.close()

rows=[]
recovery=[]
for runtime in ['python','node']:
    rows.append(probe(runtime,'original','ascii_4097','pending'))
    rows.append(probe(runtime,'fixed','ascii_4097','exited'))
    rows.append(probe(runtime,'fixed','ascii_4096','pending'))
    command = [sys.executable if runtime=='python' else 'node',str(ROOT/('contract.py' if runtime=='python' else 'contract.cjs'))]
    r=subprocess.run(command,input=cases['minimal'],capture_output=True,timeout=5)
    wanted=(expected['success_json'] + ('\n' if runtime=='python' else '')).encode()
    assert r.returncode==0 and r.stdout==wanted and not r.stderr
    recovery.append({'runtime':runtime,'fixture':'minimal','returncode':r.returncode,
                     'stdout':r.stdout.decode(),'stderr':r.stderr.decode()})
(ROOT/'pipe-results.json').write_text(json.dumps({'cases':rows,'fresh_process_recovery_checks':2,'recovery':recovery,
    'limitation':'A pending original is a one-second observation, supported by read-all source semantics; no timing guarantee or syscall observation.'},indent=2)+'\n')
print('6 held-open pipe cases and 2 fresh-process recovery checks passed')

{
  "cases": [
    {
      "runtime": "python",
      "variant": "original",
      "fixture": "ascii_4097",
      "bytes_written": 4097,
      "pipe_capacity": 65536,
      "observation_window_seconds": 1,
      "before_parent_eof": "pending",
      "writer_open_at_observation": true,
      "after_eof_returncode": 2,
      "stdout": "",
      "stderr": "",
      "forced_kill": false
    },
    {
      "runtime": "python",
      "variant": "fixed",
      "fixture": "ascii_4097",
      "bytes_written": 4097,
      "pipe_capacity": 65536,
      "observation_window_seconds": 5,
      "before_parent_eof": "exited",
      "writer_open_at_observation": true,
      "after_eof_returncode": 2,
      "stdout": "",
      "stderr": "",
      "forced_kill": false
    },
    {
      "runtime": "python",
      "variant": "fixed",
      "fixture": "ascii_4096",
      "bytes_written": 4096,
      "pipe_capacity": 65536,
      "observation_window_seconds": 1,
      "before_parent_eof": "pending",
      "writer_open_at_observation": true,
      "after_eof_returncode": 0,
      "stdout": "{\"version\":1,\"order_id\":\"1\",\"amount_cents\":0,\"currency\":\"CNY\"}\n",
      "stderr": "",
      "forced_kill": false
    },
    {
      "runtime": "node",
      "variant": "original",
      "fixture": "ascii_4097",
      "bytes_written": 4097,
      "pipe_capacity": 65536,
      "observation_window_seconds": 1,
      "before_parent_eof": "pending",
      "writer_open_at_observation": true,
      "after_eof_returncode": 2,
      "stdout": "",
      "stderr": "",
      "forced_kill": false
    },
    {
      "runtime": "node",
      "variant": "fixed",
      "fixture": "ascii_4097",
      "bytes_written": 4097,
      "pipe_capacity": 65536,
      "observation_window_seconds": 5,
      "before_parent_eof": "exited",
      "writer_open_at_observation": true,
      "after_eof_returncode": 2,
      "stdout": "",
      "stderr": "",
      "forced_kill": false
    },
    {
      "runtime": "node",
      "variant": "fixed",
      "fixture": "ascii_4096",
      "bytes_written": 4096,
      "pipe_capacity": 65536,
      "observation_window_seconds": 1,
      "before_parent_eof": "pending",
      "writer_open_at_observation": true,
      "after_eof_returncode": 0,
      "stdout": "{\"version\":1,\"order_id\":\"1\",\"amount_cents\":0,\"currency\":\"CNY\"}",
      "stderr": "",
      "forced_kill": false
    }
  ],
  "fresh_process_recovery_checks": 2,
  "recovery": [
    {
      "runtime": "python",
      "fixture": "minimal",
      "returncode": 0,
      "stdout": "{\"version\":1,\"order_id\":\"1\",\"amount_cents\":0,\"currency\":\"CNY\"}\n",
      "stderr": ""
    },
    {
      "runtime": "node",
      "fixture": "minimal",
      "returncode": 0,
      "stdout": "{\"version\":1,\"order_id\":\"1\",\"amount_cents\":0,\"currency\":\"CNY\"}",
      "stderr": ""
    }
  ],
  "limitation": "A pending original is a one-second observation, supported by read-all source semantics; no timing guarantee or syscall observation."
}

测试的 finally 负责关闭自己持有的写端、收集子进程输出，并在异常情况下终止仍未结束的自有子进程后回收。正常通过的六个案例没有强制终止。每种运行时完成上述实验后，再启动一个新进程发送最小合法记录，核对完整成功输出。这两次恢复检查说明干净的新进程仍能正常处理输入，未验证同一常驻进程复用失败状态。

超限后不要继续向已经结束的消费端写入，也不要把旧输入截成 4096 字节重送。先停止这次发送，关闭并回收自己拥有的管道和进程，再修正数据或合同，按新的一次记录重新启动。多记录长连接需要额外的帧边界与错误恢复设计，不能直接沿用本节的 EOF 协议。

test_mutants.py 在临时目录复制所需文件，分别修改 Python 与 Node 的四处关键行为：删掉第 4097 字节探测、把严格解码改成替换字符、删掉 BOM 检查、把第一次短读当作结束。先确认未修改的基线通过，每次再仅改一种运行时，并运行同一份 test_wire.py 与独立预期。测试必须以 1 退出，诊断包含预先指定的样本名和 AssertionError；语法、导入、模块缺失、引用错误或超时不能算作检出。临时目录会自动清理，交付文件不被覆盖。

删掉额外字节探测时，ascii_4097 的合法前缀会被误接受；放宽解码与删除 BOM 检查时，入口级断言分别在 invalid_ff 与 utf8_bom 失败，即使后面的字段检查仍拒绝完整 CLI；第一次短读就退出时，1 字节分块的 minimal 案例在累计长度检查失败。后两种拒绝层次必须单独验收，否则仅看最终退出码可能漏掉入口退化。

"""Run unchanged independent oracle against deliberately broken gate modules."""
import json
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path
ROOT=Path(__file__).resolve().parent
py=(ROOT/'wire_in.py').read_text()
js=(ROOT/'wire-in.cjs').read_text()
mutants = {
 'no_sentinel': (
  py.replace('while len(raw) <= LIMIT:', 'while len(raw) < LIMIT:').replace('LIMIT + 1 - len(raw)', 'LIMIT - len(raw)'),
  js.replace('Buffer.alloc(LIMIT + 1)', 'Buffer.alloc(LIMIT)').replace('while (used <= LIMIT)', 'while (used < LIMIT)').replace('LIMIT + 1 - used', 'LIMIT - used')),
 'lossy_decode': (py.replace('errors="strict"','errors="replace"'), js.replace('fatal: true','fatal: false')),
 'bom_gate_removed': (
  py.replace('    if raw.startswith(b"\\xef\\xbb\\xbf"):\n        raise ValueError("wire_bom")\n',''),
  js.replace("  if (raw.length >= 3 && raw[0] === 0xef && raw[1] === 0xbb && raw[2] === 0xbf) {\n    throw Error('wire_bom');\n  }\n",'')),
 'first_chunk_as_eof': (py.replace('        raw.extend(part)','        raw.extend(part)\n        break'),js.replace('    used += count;', '    used += count;\n    break;')),
}
files=['contract.py','contract.cjs','wire_in.py','wire-in.cjs','test_wire.py','test-gate.cjs','wire-fixtures.json','wire-expected.json']
baseline=subprocess.run([sys.executable,str(ROOT/'test_wire.py')],cwd=ROOT,capture_output=True,text=True,timeout=15)
assert baseline.returncode == 0 and baseline.stderr == '', baseline
reasons={'no_sentinel':'ascii_4097','lossy_decode':'invalid_ff','bom_gate_removed':'utf8_bom','first_chunk_as_eof':'minimal'}
rows=[]
for name,(bad_py,bad_js) in mutants.items():
 for runtime,bad in [('python',bad_py),('node',bad_js)]:
  assert bad != (py if runtime=='python' else js)
  with tempfile.TemporaryDirectory(dir=ROOT, prefix='mutant-') as tmp:
   p=Path(tmp)
   for f in files: shutil.copy2(ROOT/f,p/f)
   (p/('wire_in.py' if runtime=='python' else 'wire-in.cjs')).write_text(bad)
   r=subprocess.run([sys.executable,str(p/'test_wire.py')],cwd=p,capture_output=True,text=True,timeout=15)
   assert r.returncode == 1, (name,runtime,'wrong failure',r)
   assert 'AssertionError' in r.stderr and reasons[name] in r.stderr, (name,runtime,'wrong assertion',r)
   forbidden=['SyntaxError','ModuleNotFoundError','ImportError','Cannot find module','ReferenceError','TimeoutExpired']
   assert not any(s in r.stderr for s in forbidden), (name,runtime,'infrastructure failure',r)
   rows.append({'mutation':name,'runtime':runtime,'killed':True,'returncode':r.returncode,'expected_assertion':reasons[name],'baseline_passed':True,'stderr':r.stderr})
(ROOT/'mutation-results.json').write_text(json.dumps(rows,indent=2)+'\n')
print('8 gate mutants rejected by unchanged independent oracle')

[
  {
    "mutation": "no_sentinel",
    "runtime": "python",
    "killed": true,
    "returncode": 1,
    "expected_assertion": "ascii_4097",
    "baseline_passed": true,
    "stderr": "Traceback (most recent call last):\n  File \"source/mutant-TEMP/test_wire.py\", line 27, in <module>\n    assert r.returncode == (0 if accepted else 2), (case['name'], runtime, r)\n           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\nAssertionError: ('ascii_4097', 'python', CompletedProcess(args=['python', 'source/mutant-TEMP/contract.py'], returncode=0, stdout=b'{\"version\":1,\"order_id\":\"1\",\"amount_cents\":0,\"currency\":\"CNY\"}\\n', stderr=b''))\n"
  },
  {
    "mutation": "no_sentinel",
    "runtime": "node",
    "killed": true,
    "returncode": 1,
    "expected_assertion": "ascii_4097",
    "baseline_passed": true,
    "stderr": "Traceback (most recent call last):\n  File \"source/mutant-TEMP/test_wire.py\", line 27, in <module>\n    assert r.returncode == (0 if accepted else 2), (case['name'], runtime, r)\n           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\nAssertionError: ('ascii_4097', 'node', CompletedProcess(args=['node', 'source/mutant-TEMP/contract.cjs'], returncode=0, stdout=b'{\"version\":1,\"order_id\":\"1\",\"amount_cents\":0,\"currency\":\"CNY\"}', stderr=b''))\n"
  },
  {
    "mutation": "lossy_decode",
    "runtime": "python",
    "killed": true,
    "returncode": 1,
    "expected_assertion": "invalid_ff",
    "baseline_passed": true,
    "stderr": "Traceback (most recent call last):\n  File \"source/mutant-TEMP/test_wire.py\", line 44, in <module>\n    assert bool(error) == reject, case['name']\n           ^^^^^^^^^^^^^^^^^^^^^\nAssertionError: invalid_ff\n"
  },
  {
    "mutation": "lossy_decode",
    "runtime": "node",
    "killed": true,
    "returncode": 1,
    "expected_assertion": "invalid_ff",
    "baseline_passed": true,
    "stderr": "node:assert:152\n  throw new AssertionError(obj);\n  ^\n\nAssertionError [ERR_ASSERTION]: invalid_ff\n\nfalse !== true\n\n    at Object.<anonymous> (source/mutant-TEMP/test-gate.cjs:20:10)\n    at Module._compile (node:internal/modules/cjs/loader:1872:14)\n    at Object..js (node:internal/modules/cjs/loader:2003:10)\n    at Module.load (node:internal/modules/cjs/loader:1594:32)\n    at Module._load (node:internal/modules/cjs/loader:1396:12)\n    at wrapModuleLoad (node:internal/modules/cjs/loader:255:19)\n    at Module.executeUserEntryPoint [as runMain] (node:internal/modules/run_main:154:5)\n    at node:internal/main/run_main_module:33:47 {\n  generatedMessage: false,\n  code: 'ERR_ASSERTION',\n  actual: false,\n  expected: true,\n  operator: 'strictEqual',\n  diff: 'simple'\n}\n\nNode.js v24.19.0\nTraceback (most recent call last):\n  File \"source/mutant-TEMP/test_wire.py\", line 49, in <module>\n    report['node_gate'] = json.loads(subprocess.check_output(['node','test-gate.cjs'],cwd=ROOT))\n                                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n  File \"PYTHON_STDLIB/subprocess.py\", line 466, in check_output\n    return run(*popenargs, stdout=PIPE, timeout=timeout, check=True,\n           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n  File \"PYTHON_STDLIB/subprocess.py\", line 571, in run\n    raise CalledProcessError(retcode, process.args,\nsubprocess.CalledProcessError: Command '['node', 'test-gate.cjs']' returned non-zero exit status 1.\n"
  },
  {
    "mutation": "bom_gate_removed",
    "runtime": "python",
    "killed": true,
    "returncode": 1,
    "expected_assertion": "utf8_bom",
    "baseline_passed": true,
    "stderr": "Traceback (most recent call last):\n  File \"source/mutant-TEMP/test_wire.py\", line 44, in <module>\n    assert bool(error) == reject, case['name']\n           ^^^^^^^^^^^^^^^^^^^^^\nAssertionError: utf8_bom\n"
  },
  {
    "mutation": "bom_gate_removed",
    "runtime": "node",
    "killed": true,
    "returncode": 1,
    "expected_assertion": "utf8_bom",
    "baseline_passed": true,
    "stderr": "node:assert:152\n  throw new AssertionError(obj);\n  ^\n\nAssertionError [ERR_ASSERTION]: utf8_bom\n\nfalse !== true\n\n    at Object.<anonymous> (source/mutant-TEMP/test-gate.cjs:20:10)\n    at Module._compile (node:internal/modules/cjs/loader:1872:14)\n    at Object..js (node:internal/modules/cjs/loader:2003:10)\n    at Module.load (node:internal/modules/cjs/loader:1594:32)\n    at Module._load (node:internal/modules/cjs/loader:1396:12)\n    at wrapModuleLoad (node:internal/modules/cjs/loader:255:19)\n    at Module.executeUserEntryPoint [as runMain] (node:internal/modules/run_main:154:5)\n    at node:internal/main/run_main_module:33:47 {\n  generatedMessage: false,\n  code: 'ERR_ASSERTION',\n  actual: false,\n  expected: true,\n  operator: 'strictEqual',\n  diff: 'simple'\n}\n\nNode.js v24.19.0\nTraceback (most recent call last):\n  File \"source/mutant-TEMP/test_wire.py\", line 49, in <module>\n    report['node_gate'] = json.loads(subprocess.check_output(['node','test-gate.cjs'],cwd=ROOT))\n                                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n  File \"PYTHON_STDLIB/subprocess.py\", line 466, in check_output\n    return run(*popenargs, stdout=PIPE, timeout=timeout, check=True,\n           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n  File \"PYTHON_STDLIB/subprocess.py\", line 571, in run\n    raise CalledProcessError(retcode, process.args,\nsubprocess.CalledProcessError: Command '['node', 'test-gate.cjs']' returned non-zero exit status 1.\n"
  },
  {
    "mutation": "first_chunk_as_eof",
    "runtime": "python",
    "killed": true,
    "returncode": 1,
    "expected_assertion": "minimal",
    "baseline_passed": true,
    "stderr": "Traceback (most recent call last):\n  File \"source/mutant-TEMP/test_wire.py\", line 45, in <module>\n    assert cursor == min(len(raw), 4097), case['name']\n           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\nAssertionError: minimal\n"
  },
  {
    "mutation": "first_chunk_as_eof",
    "runtime": "node",
    "killed": true,
    "returncode": 1,
    "expected_assertion": "minimal",
    "baseline_passed": true,
    "stderr": "node:assert:152\n  throw new AssertionError(obj);\n  ^\n\nAssertionError [ERR_ASSERTION]: minimal copied_bytes\n\n1 !== 62\n\n    at Object.<anonymous> (source/mutant-TEMP/test-gate.cjs:21:10)\n    at Module._compile (node:internal/modules/cjs/loader:1872:14)\n    at Object..js (node:internal/modules/cjs/loader:2003:10)\n    at Module.load (node:internal/modules/cjs/loader:1594:32)\n    at Module._load (node:internal/modules/cjs/loader:1396:12)\n    at wrapModuleLoad (node:internal/modules/cjs/loader:255:19)\n    at Module.executeUserEntryPoint [as runMain] (node:internal/modules/run_main:154:5)\n    at node:internal/main/run_main_module:33:47 {\n  generatedMessage: false,\n  code: 'ERR_ASSERTION',\n  actual: 1,\n  expected: 62,\n  operator: 'strictEqual',\n  diff: 'simple'\n}\n\nNode.js v24.19.0\nTraceback (most recent call last):\n  File \"source/mutant-TEMP/test_wire.py\", line 49, in <module>\n    report['node_gate'] = json.loads(subprocess.check_output(['node','test-gate.cjs'],cwd=ROOT))\n                                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n  File \"PYTHON_STDLIB/subprocess.py\", line 466, in check_output\n    return run(*popenargs, stdout=PIPE, timeout=timeout, check=True,\n           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n  File \"PYTHON_STDLIB/subprocess.py\", line 571, in run\n    raise CalledProcessError(retcode, process.args,\nsubprocess.CalledProcessError: Command '['node', 'test-gate.cjs']' returned non-zero exit status 1.\n"
  }
]

这些变异用于检查特定断言是否能拦住已知退化，数量不能代表安全覆盖率。没有进行 fuzz、穷举字节序列、压力测试或跨版本兼容性验证。

公开离线包的 source/original 保留原文七个文件，source 顶层保存接好字节入口的版本、原回归和新增测试。手工复制时，可先把上文原文文件保存进 original，再把原 expected.json、golden-fixtures.json、test_protocol.py、test_edges.py 复制到 source 顶层，补齐本节各代码块。先运行 make_wire_fixtures.py 生成原始字节输入，再运行测试。

下载完整源码、固定输入与样例输出 ZIP。离线完整包同时包含全文与四幅解释图。

源码 ZIP 根目录附有 SHA256SUMS，记录包内 32 个源码、预期、说明与样例输出文件的 SHA-256，不包含清单自身。解压后，先在包含 README.txt 的目录执行下列校验，再进入 source 运行测试。本轮 Linux 实测 32 项校验全部通过；在独立测试副本中改动 README.txt 后，校验以退出码 1 报告该文件失败。这个清单只用于核对文件内容，不是签名，也不证明下载来源可信或程序安全。校验失败时先检查是否修改了包内文件，并重新下载核对；不要通过重新生成清单把变化当成通过。

sha256sum --check SHA256SUMS

# 解压后进入 source；已有 Python 与 Node 即可，不需要 pip 或 npm
cd source
python --version
node --version
# 原入口及原合同
python -B original/test_protocol.py
python -B original/test_edges.py
# 新入口复用原合同
python -B test_protocol.py
python -B test_edges.py
# 新增输入、分块、定向变异与持有写端的真实管道
python -B make_wire_fixtures.py
python -B probe_original_bytes.py
python -B test_wire.py
python -B test_mutants.py
python -B probe_pipe.py
python -B probe_newlines.py

本次 Python 3.12.14、Node v24.19.0 的输出依次如下。pipe 探针使用 Linux 的 F_GETPIPE_SZ，本次容量为 65536 字节；该探针没有在 macOS 或 Windows 上运行。换行观察同样只描述这台机器的实际 stdin 设置。

27 shared fixtures; 54 acceptance checks passed; 8 semantic roundtrips passed
12 additional edge fixtures; 24 acceptance checks passed
27 shared fixtures; 54 acceptance checks passed; 8 semantic roundtrips passed
12 additional edge fixtures; 24 acceptance checks passed
14 byte fixtures created
28 finite byte decisions match the fixed independent oracle in the unchanged original CLIs
14 raw byte fixtures; 28 CLI decisions; 84 injected-reader checks passed
8 gate mutants rejected by unchanged independent oracle
6 held-open pipe cases and 2 fresh-process recovery checks passed
Actual OS stdin preserves CRLF on this runtime; no newline-size divergence found

sample-output 保存本次完整 JSON 结果；其中变异诊断仅将绝对路径和随机临时目录名改成包内占位名称，断言内容和退出结果保留。正式运行生成的结果文件留在 source 或 source/original 中，便于与失败输入一起保存。完整结果中的 54、24、28、84 分属不同测试层，不应合并成一项笼统的“用例数”。

{
  "python": "3.12.14",
  "node": "v24.19.0",
  "cli": [
    {
      "name": "minimal",
      "runtime": "python",
      "returncode": 0,
      "stdout": "{\"version\":1,\"order_id\":\"1\",\"amount_cents\":0,\"currency\":\"CNY\"}\n",
      "stderr": ""
    },
    {
      "name": "minimal",
      "runtime": "node",
      "returncode": 0,
      "stdout": "{\"version\":1,\"order_id\":\"1\",\"amount_cents\":0,\"currency\":\"CNY\"}",
      "stderr": ""
    },
    {
      "name": "ascii_4096",
      "runtime": "python",
      "returncode": 0,
      "stdout": "{\"version\":1,\"order_id\":\"1\",\"amount_cents\":0,\"currency\":\"CNY\"}\n",
      "stderr": ""
    },
    {
      "name": "ascii_4096",
      "runtime": "node",
      "returncode": 0,
      "stdout": "{\"version\":1,\"order_id\":\"1\",\"amount_cents\":0,\"currency\":\"CNY\"}",
      "stderr": ""
    },
    {
      "name": "ascii_4097",
      "runtime": "python",
      "returncode": 2,
      "stdout": "",
      "stderr": ""
    },
    {
      "name": "ascii_4097",
      "runtime": "node",
      "returncode": 2,
      "stdout": "",
      "stderr": ""
    },
    {
      "name": "crlf_4096",
      "runtime": "python",
      "returncode": 0,
      "stdout": "{\"version\":1,\"order_id\":\"1\",\"amount_cents\":0,\"currency\":\"CNY\"}\n",
      "stderr": ""
    },
    {
      "name": "crlf_4096",
      "runtime": "node",
      "returncode": 0,
      "stdout": "{\"version\":1,\"order_id\":\"1\",\"amount_cents\":0,\"currency\":\"CNY\"}",
      "stderr": ""
    },
    {
      "name": "crlf_4097",
      "runtime": "python",
      "returncode": 2,
      "stdout": "",
      "stderr": ""
    },
    {
      "name": "crlf_4097",
      "runtime": "node",
      "returncode": 2,
      "stdout": "",
      "stderr": ""
    },
    {
      "name": "large_finite",
      "runtime": "python",
      "returncode": 2,
      "stdout": "",
      "stderr": ""
    },
    {
      "name": "large_finite",
      "runtime": "node",
      "returncode": 2,
      "stdout": "",
      "stderr": ""
    },
    {
      "name": "utf8_bom",
      "runtime": "python",
      "returncode": 2,
      "stdout": "",
      "stderr": ""
    },
    {
      "name": "utf8_bom",
      "runtime": "node",
      "returncode": 2,
      "stdout": "",
      "stderr": ""
    },
    {
      "name": "invalid_ff",
      "runtime": "python",
      "returncode": 2,
      "stdout": "",
      "stderr": ""
    },
    {
      "name": "invalid_ff",
      "runtime": "node",
      "returncode": 2,
      "stdout": "",
      "stderr": ""
    },
    {
      "name": "truncated_utf8",
      "runtime": "python",
      "returncode": 2,
      "stdout": "",
      "stderr": ""
    },
    {
      "name": "truncated_utf8",
      "runtime": "node",
      "returncode": 2,
      "stdout": "",
      "stderr": ""
    },
    {
      "name": "utf16le",
      "runtime": "python",
      "returncode": 2,
      "stdout": "",
      "stderr": ""
    },
    {
      "name": "utf16le",
      "runtime": "node",
      "returncode": 2,
      "stdout": "",
      "stderr": ""
    },
    {
      "name": "escaped_digit",
      "runtime": "python",
      "returncode": 0,
      "stdout": "{\"version\":1,\"order_id\":\"1\",\"amount_cents\":0,\"currency\":\"CNY\"}\n",
      "stderr": ""
    },
    {
      "name": "escaped_digit",
      "runtime": "node",
      "returncode": 0,
      "stdout": "{\"version\":1,\"order_id\":\"1\",\"amount_cents\":0,\"currency\":\"CNY\"}",
      "stderr": ""
    },
    {
      "name": "empty",
      "runtime": "python",
      "returncode": 2,
      "stdout": "",
      "stderr": ""
    },
    {
      "name": "empty",
      "runtime": "node",
      "returncode": 2,
      "stdout": "",
      "stderr": ""
    },
    {
      "name": "two_documents",
      "runtime": "python",
      "returncode": 2,
      "stdout": "",
      "stderr": ""
    },
    {
      "name": "two_documents",
      "runtime": "node",
      "returncode": 2,
      "stdout": "",
      "stderr": ""
    },
    {
      "name": "multibyte_value",
      "runtime": "python",
      "returncode": 2,
      "stdout": "",
      "stderr": ""
    },
    {
      "name": "multibyte_value",
      "runtime": "node",
      "returncode": 2,
      "stdout": "",
      "stderr": ""
    }
  ],
  "python_gate": [
    {
      "name": "minimal",
      "step": 1,
      "copied_bytes": 62,
      "calls": 63,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "minimal",
      "step": 7,
      "copied_bytes": 62,
      "calls": 10,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "minimal",
      "step": 4097,
      "copied_bytes": 62,
      "calls": 2,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "ascii_4096",
      "step": 1,
      "copied_bytes": 4096,
      "calls": 4097,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "ascii_4096",
      "step": 7,
      "copied_bytes": 4096,
      "calls": 587,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "ascii_4096",
      "step": 4097,
      "copied_bytes": 4096,
      "calls": 2,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "ascii_4097",
      "step": 1,
      "copied_bytes": 4097,
      "calls": 4097,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "ascii_4097",
      "step": 7,
      "copied_bytes": 4097,
      "calls": 586,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "ascii_4097",
      "step": 4097,
      "copied_bytes": 4097,
      "calls": 1,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "crlf_4096",
      "step": 1,
      "copied_bytes": 4096,
      "calls": 4097,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "crlf_4096",
      "step": 7,
      "copied_bytes": 4096,
      "calls": 587,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "crlf_4096",
      "step": 4097,
      "copied_bytes": 4096,
      "calls": 2,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "crlf_4097",
      "step": 1,
      "copied_bytes": 4097,
      "calls": 4097,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "crlf_4097",
      "step": 7,
      "copied_bytes": 4097,
      "calls": 586,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "crlf_4097",
      "step": 4097,
      "copied_bytes": 4097,
      "calls": 1,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "large_finite",
      "step": 1,
      "copied_bytes": 4097,
      "calls": 4097,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "large_finite",
      "step": 7,
      "copied_bytes": 4097,
      "calls": 586,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "large_finite",
      "step": 4097,
      "copied_bytes": 4097,
      "calls": 1,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "utf8_bom",
      "step": 1,
      "copied_bytes": 65,
      "calls": 66,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "utf8_bom",
      "step": 7,
      "copied_bytes": 65,
      "calls": 11,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "utf8_bom",
      "step": 4097,
      "copied_bytes": 65,
      "calls": 2,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "invalid_ff",
      "step": 1,
      "copied_bytes": 63,
      "calls": 64,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "invalid_ff",
      "step": 7,
      "copied_bytes": 63,
      "calls": 10,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "invalid_ff",
      "step": 4097,
      "copied_bytes": 63,
      "calls": 2,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "truncated_utf8",
      "step": 1,
      "copied_bytes": 64,
      "calls": 65,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "truncated_utf8",
      "step": 7,
      "copied_bytes": 64,
      "calls": 11,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "truncated_utf8",
      "step": 4097,
      "copied_bytes": 64,
      "calls": 2,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "utf16le",
      "step": 1,
      "copied_bytes": 124,
      "calls": 125,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "utf16le",
      "step": 7,
      "copied_bytes": 124,
      "calls": 19,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "utf16le",
      "step": 4097,
      "copied_bytes": 124,
      "calls": 2,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "escaped_digit",
      "step": 1,
      "copied_bytes": 67,
      "calls": 68,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "escaped_digit",
      "step": 7,
      "copied_bytes": 67,
      "calls": 11,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "escaped_digit",
      "step": 4097,
      "copied_bytes": 67,
      "calls": 2,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "empty",
      "step": 1,
      "copied_bytes": 0,
      "calls": 1,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "empty",
      "step": 7,
      "copied_bytes": 0,
      "calls": 1,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "empty",
      "step": 4097,
      "copied_bytes": 0,
      "calls": 1,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "two_documents",
      "step": 1,
      "copied_bytes": 124,
      "calls": 125,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "two_documents",
      "step": 7,
      "copied_bytes": 124,
      "calls": 19,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "two_documents",
      "step": 4097,
      "copied_bytes": 124,
      "calls": 2,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "multibyte_value",
      "step": 1,
      "copied_bytes": 64,
      "calls": 65,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "multibyte_value",
      "step": 7,
      "copied_bytes": 64,
      "calls": 11,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "multibyte_value",
      "step": 4097,
      "copied_bytes": 64,
      "calls": 2,
      "max_requested": 4097,
      "rejected": false
    }
  ],
  "node_gate": [
    {
      "name": "minimal",
      "step": 1,
      "copied_bytes": 62,
      "calls": 63,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "minimal",
      "step": 7,
      "copied_bytes": 62,
      "calls": 10,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "minimal",
      "step": 4097,
      "copied_bytes": 62,
      "calls": 2,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "ascii_4096",
      "step": 1,
      "copied_bytes": 4096,
      "calls": 4097,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "ascii_4096",
      "step": 7,
      "copied_bytes": 4096,
      "calls": 587,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "ascii_4096",
      "step": 4097,
      "copied_bytes": 4096,
      "calls": 2,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "ascii_4097",
      "step": 1,
      "copied_bytes": 4097,
      "calls": 4097,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "ascii_4097",
      "step": 7,
      "copied_bytes": 4097,
      "calls": 586,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "ascii_4097",
      "step": 4097,
      "copied_bytes": 4097,
      "calls": 1,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "crlf_4096",
      "step": 1,
      "copied_bytes": 4096,
      "calls": 4097,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "crlf_4096",
      "step": 7,
      "copied_bytes": 4096,
      "calls": 587,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "crlf_4096",
      "step": 4097,
      "copied_bytes": 4096,
      "calls": 2,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "crlf_4097",
      "step": 1,
      "copied_bytes": 4097,
      "calls": 4097,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "crlf_4097",
      "step": 7,
      "copied_bytes": 4097,
      "calls": 586,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "crlf_4097",
      "step": 4097,
      "copied_bytes": 4097,
      "calls": 1,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "large_finite",
      "step": 1,
      "copied_bytes": 4097,
      "calls": 4097,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "large_finite",
      "step": 7,
      "copied_bytes": 4097,
      "calls": 586,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "large_finite",
      "step": 4097,
      "copied_bytes": 4097,
      "calls": 1,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "utf8_bom",
      "step": 1,
      "copied_bytes": 65,
      "calls": 66,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "utf8_bom",
      "step": 7,
      "copied_bytes": 65,
      "calls": 11,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "utf8_bom",
      "step": 4097,
      "copied_bytes": 65,
      "calls": 2,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "invalid_ff",
      "step": 1,
      "copied_bytes": 63,
      "calls": 64,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "invalid_ff",
      "step": 7,
      "copied_bytes": 63,
      "calls": 10,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "invalid_ff",
      "step": 4097,
      "copied_bytes": 63,
      "calls": 2,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "truncated_utf8",
      "step": 1,
      "copied_bytes": 64,
      "calls": 65,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "truncated_utf8",
      "step": 7,
      "copied_bytes": 64,
      "calls": 11,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "truncated_utf8",
      "step": 4097,
      "copied_bytes": 64,
      "calls": 2,
      "max_requested": 4097,
      "rejected": true
    },
    {
      "name": "utf16le",
      "step": 1,
      "copied_bytes": 124,
      "calls": 125,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "utf16le",
      "step": 7,
      "copied_bytes": 124,
      "calls": 19,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "utf16le",
      "step": 4097,
      "copied_bytes": 124,
      "calls": 2,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "escaped_digit",
      "step": 1,
      "copied_bytes": 67,
      "calls": 68,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "escaped_digit",
      "step": 7,
      "copied_bytes": 67,
      "calls": 11,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "escaped_digit",
      "step": 4097,
      "copied_bytes": 67,
      "calls": 2,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "empty",
      "step": 1,
      "copied_bytes": 0,
      "calls": 1,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "empty",
      "step": 7,
      "copied_bytes": 0,
      "calls": 1,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "empty",
      "step": 4097,
      "copied_bytes": 0,
      "calls": 1,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "two_documents",
      "step": 1,
      "copied_bytes": 124,
      "calls": 125,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "two_documents",
      "step": 7,
      "copied_bytes": 124,
      "calls": 19,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "two_documents",
      "step": 4097,
      "copied_bytes": 124,
      "calls": 2,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "multibyte_value",
      "step": 1,
      "copied_bytes": 64,
      "calls": 65,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "multibyte_value",
      "step": 7,
      "copied_bytes": 64,
      "calls": 11,
      "max_requested": 4097,
      "rejected": false
    },
    {
      "name": "multibyte_value",
      "step": 4097,
      "copied_bytes": 64,
      "calls": 2,
      "max_requested": 4097,
      "rejected": false
    }
  ]
}

退出码 2 且无成功载荷时，先保留输入字节数和失败类型，再检查上限、编码、BOM、JSON 语法与字段合同。不要通过 errors=replace、静默去 BOM 或截断字节让同一坏输入通过。遇到测试超时、I/O 异常、解释器缺失或其他退出码，应保留完整 stderr 和版本信息，按运行故障处理。需要调整合同的场景，先明确新预期，再一起修改两端和测试。

本节验证了两个本地阻塞 CLI 的原始字节入口：超限输入可以在 EOF 前被拒绝，未超限输入须收齐后严格解码，再交给原有四字段验收器。源码、从正文重新提取的代码和公开 ZIP 解压后的版本分别复跑；图 4 单独渲染检查。全文 HTML 只做静态结构与内联代码检查，没有启动浏览器做排版验收。

入口没有处理网络超时、慢速持续输入、取消信号、并发配额、常驻连接帧边界、Windows stdin 模式、完整 Unicode 安全审计或生产服务集成。本次也没有 RSS 测量或系统调用跟踪证据。保留的上限检查和扁平 token 扫描继续受原文教学范围约束，真实外部接口还需要宿主层的资源预算、鉴权和业务权限检查。

资料核对日期为 2026-10-13。Python 在线页面当前标记 3.12.15，Node fs 与 TextDecoder 资料固定到 v24.19.0 文档；这些页面说明 API 语义，本文运行结论以实际记录的 Python 3.12.14 与 Node v24.19.0 为准。

• Python：os.read 的数量与 EOF 约定
• Node：fs.readSync 的偏移、长度与 bytesRead
• Node：TextDecoder 的 fatal 与 ignoreBOM
• RFC 8259：UTF-8 与 BOM 策略