fix: bound M0 publisher envelopes to ingress size

Co-Authored-By: Codex <noreply@openai.com>

Author: Pigbibi

docs/m0_research_publisher_envelope_contract.md

@@ -34,6 +34,14 @@ canonical JSON（键排序、紧凑分隔符、禁止 NaN）计算。台账的
 `generated_at` 和 `computed_at` 必须相同，均为精确到秒的 UTC `Z` 时间戳。
 因此同一 source artifact、metadata 和 `--now` 总会生成字节相同的封套。
 
+尽管 source snapshot 离线输入上限为 2 MiB，生成完成的 canonical envelope
+本身必须不超过 **262,144 bytes（256 KiB）**。这个限制按将要写入和 POST 的
+紧凑 UTF-8 JSON body 的实际字节数计算，而不是字符数、文件系统占用或 source
+snapshot 大小；本地输出文件末尾的换行符不属于 JSON body。超过该上限会以
+`publisher_envelope_size_exceeded` fail closed，既不写本地文件，也不发起网络
+请求。这个上限与 M0 接收端 Worker ingress 一致，避免“本地可生成但接收端无法
+接收”的跨模块失败。
+
 ## 默认离线构建
 
 ```bash

python/scripts/build_m0_research_publisher_envelope.py

@@ -40,6 +40,10 @@
 PUBLISH_URL_ENV = "QSL_M0_RESEARCH_LEDGER_PUBLISH_URL"
 PUBLISH_TOKEN_ENV = "QSL_M0_RESEARCH_LEDGER_PUBLISH_TOKEN"
 MAX_SOURCE_SNAPSHOT_BYTES = 2 * 1024 * 1024
+# The receiving Worker ingress accepts at most 256 KiB.  This is enforced on
+# the actual compact UTF-8 JSON body, not on a Python object estimate, source
+# artifact size, or character count.
+MAX_PUBLISHER_ENVELOPE_BYTES = 256 * 1024
 
 _REPOSITORY = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]*/[A-Za-z0-9][A-Za-z0-9_.-]*$")
 _REVISION = re.compile(r"^[0-9a-f]{40}$")
@@ -69,6 +73,21 @@ def calculate_ledger_sha256(ledger: Mapping[str, Any]) -> str:
     return hashlib.sha256(canonical_json(dict(ledger)).encode("utf-8")).hexdigest()
 
 
+def canonical_envelope_body(envelope: Mapping[str, Any]) -> bytes:
+    """Serialize the exact compact UTF-8 body used for local output and POST."""
+
+    if not isinstance(envelope, Mapping):
+        raise M0ResearchPublisherEnvelopeError("publisher_envelope_invalid")
+    return canonical_json(dict(envelope)).encode("utf-8")
+
+
+def _enforce_publisher_envelope_size(envelope: Mapping[str, Any]) -> None:
+    """Fail closed before a too-large envelope can be written or published."""
+
+    if len(canonical_envelope_body(envelope)) > MAX_PUBLISHER_ENVELOPE_BYTES:
+        raise M0ResearchPublisherEnvelopeError("publisher_envelope_size_exceeded")
+
+
 def _exact_mapping(value: object, fields: frozenset[str], label: str) -> dict[str, Any]:
     if not isinstance(value, Mapping) or set(value) != fields:
         raise M0ResearchPublisherEnvelopeError(f"{label}_keys_invalid")
@@ -270,13 +289,15 @@ def validate_m0_research_publisher_envelope(payload: object) -> dict[str, Any]:
     expected_digest = calculate_ledger_sha256(normalized_ledger)
     if _require_sha256(envelope["ledger_sha256"], "ledger_sha256") != expected_digest:
         raise M0ResearchPublisherEnvelopeError("ledger_sha256_mismatch")
-    return {
+    normalized = {
         "schema_version": PUBLISHER_ENVELOPE_SCHEMA,
         "producer": normalized_producer,
         "source_artifact": normalized_artifact,
         "ledger_sha256": expected_digest,
         "ledger": normalized_ledger,
     }
+    _enforce_publisher_envelope_size(normalized)
+    return normalized
 
 
 def _publish_url_from_environment(environ: Mapping[str, str]) -> tuple[str, str]:
@@ -308,7 +329,7 @@ def publish_m0_research_publisher_envelope(
     url, token = _publish_url_from_environment(os.environ if environ is None else environ)
     request = urllib.request.Request(
         url,
-        data=canonical_json(validated).encode("utf-8"),
+        data=canonical_envelope_body(validated),
         method="POST",
         headers={
             "Authorization": f"Bearer {token}",
@@ -376,7 +397,7 @@ def main(argv: Sequence[str] | None = None) -> int:
         # leave a local file that an operator mistakes for an attempted POST.
         _publish_url_from_environment(os.environ)
     args.output.parent.mkdir(parents=True, exist_ok=True)
-    args.output.write_text(canonical_json(envelope) + "\n", encoding="utf-8")
+    args.output.write_bytes(canonical_envelope_body(envelope) + b"\n")
     if args.publish:
         publish_m0_research_publisher_envelope(envelope)
     print(
@@ -402,13 +423,15 @@ def main(argv: Sequence[str] | None = None) -> int:
 
 __all__ = [
     "MAX_SOURCE_SNAPSHOT_BYTES",
+    "MAX_PUBLISHER_ENVELOPE_BYTES",
     "M0ResearchPublisherEnvelopeError",
     "PUBLISHER_ENVELOPE_SCHEMA",
     "PUBLISH_TOKEN_ENV",
     "PUBLISH_URL_ENV",
     "build_m0_research_publisher_envelope",
     "build_source_artifact_metadata",
     "calculate_ledger_sha256",
+    "canonical_envelope_body",
     "canonical_json",
     "canonical_timestamp",
     "load_source_snapshot",

python/tests/test_build_m0_research_publisher_envelope.py

@@ -44,6 +44,15 @@ def getcode(self):
 
 
 class M0ResearchPublisherEnvelopeTest(unittest.TestCase):
+    def test_schema_declares_the_cross_module_canonical_utf8_body_limit(self):
+        schema = json.loads(
+            (ROOT.parent / "schemas" / "qsl-m0-research-publisher-envelope.v1.schema.json").read_text(
+                encoding="utf-8"
+            )
+        )
+        self.assertEqual(schema["x-qsl-canonical-utf8-max-bytes"], 256 * 1024)
+        self.assertIn("canonical UTF-8 JSON request body", schema["$comment"])
+
     def _snapshot(self) -> dict[str, object]:
         return {
             "schema_version": "qsl_m0_research_source_snapshot.v1",
@@ -147,6 +156,28 @@ def test_build_is_deterministic_hash_bound_and_research_only(self):
         self.assertTrue(first["ledger"]["policy"]["no_order"])
         self.assertEqual(first["ledger_sha256"], publisher.calculate_ledger_sha256(first["ledger"]))
         self.assertEqual(publisher.validate_m0_research_publisher_envelope(first), first)
+        self.assertLessEqual(
+            len(publisher.canonical_envelope_body(first)),
+            publisher.MAX_PUBLISHER_ENVELOPE_BYTES,
+        )
+
+    def test_builder_fails_closed_when_actual_utf8_envelope_body_exceeds_worker_ingress_limit(self):
+        oversized = self._snapshot()
+        hypotheses = []
+        for index in range(500):
+            hypothesis = json.loads(json.dumps(oversized["hypotheses"][0]))
+            hypothesis["hypothesis_id"] = f"m0r-large-{index:03d}"
+            hypothesis["subject"]["identifier"] = f"SOXX-{index:03d}"
+            hypotheses.append(hypothesis)
+        oversized["hypotheses"] = hypotheses
+        with self.assertRaisesRegex(publisher.M0ResearchPublisherEnvelopeError, "publisher_envelope_size_exceeded"):
+            publisher.build_m0_research_publisher_envelope(
+                source_snapshot=oversized,
+                source_artifact=self._artifact("f" * 64),
+                producer_repository="QuantStrategyLab/QuantRuntimeSettings",
+                producer_revision="e" * 40,
+                now="2026-08-21T12:00:00Z",
+            )
 
     def test_envelope_validation_rejects_digest_or_execution_policy_tampering(self):
         envelope = publisher.build_m0_research_publisher_envelope(
@@ -180,12 +211,37 @@ def test_cli_default_is_local_only_and_binds_the_exact_source_bytes(self):
             envelope = json.loads(output.read_text(encoding="utf-8"))
             self.assertEqual(envelope["source_artifact"]["sha256"], sha256)
             self.assertEqual(envelope["ledger_sha256"], publisher.calculate_ledger_sha256(envelope["ledger"]))
+            self.assertEqual(output.read_bytes(), publisher.canonical_envelope_body(envelope) + b"\n")
 
             missing_output = root / "missing.json"
             with self.assertRaisesRegex(publisher.M0ResearchPublisherEnvelopeError, "source_artifact_sha256_mismatch"):
                 publisher.main(self._arguments(source, missing_output, "0" * 64))
             self.assertFalse(missing_output.exists())
 
+    def test_cli_oversize_fails_before_any_write_or_opt_in_publish(self):
+        oversized = self._snapshot()
+        hypotheses = []
+        for index in range(500):
+            hypothesis = json.loads(json.dumps(oversized["hypotheses"][0]))
+            hypothesis["hypothesis_id"] = f"m0r-large-{index:03d}"
+            hypothesis["subject"]["identifier"] = f"SOXX-{index:03d}"
+            hypotheses.append(hypothesis)
+        oversized["hypotheses"] = hypotheses
+        with tempfile.TemporaryDirectory() as temp_dir:
+            root = Path(temp_dir)
+            source = root / "oversized-source.json"
+            output = root / "must-not-exist.json"
+            raw = json.dumps(oversized, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
+            source.write_bytes(raw)
+            arguments = self._arguments(source, output, hashlib.sha256(raw).hexdigest()) + ["--publish"]
+            with patch.object(publisher.urllib.request, "urlopen", side_effect=AssertionError("network called")):
+                with self.assertRaisesRegex(
+                    publisher.M0ResearchPublisherEnvelopeError,
+                    "publisher_envelope_size_exceeded",
+                ):
+                    publisher.main(arguments)
+            self.assertFalse(output.exists())
+
     def test_publish_requires_dedicated_environment_and_never_serializes_token(self):
         envelope = publisher.build_m0_research_publisher_envelope(
             source_snapshot=self._snapshot(),

schemas/qsl-m0-research-publisher-envelope.v1.schema.json

@@ -3,6 +3,8 @@
   "$id": "https://quantstrategylab.github.io/QuantRuntimeSettings/schemas/qsl-m0-research-publisher-envelope.v1.schema.json",
   "title": "QSL M0 Research Publisher Envelope v1",
   "description": "Hash-bound transport for a locally aggregated M0 research ledger. It cannot express allocation, runtime, platform, broker, or execution authority.",
+  "$comment": "The complete canonical UTF-8 JSON request body must be no larger than 262144 bytes. JSON Schema cannot measure a whole serialized document's UTF-8 byte length; publisher and ingress implementations must enforce this bound before write or POST.",
+  "x-qsl-canonical-utf8-max-bytes": 262144,
   "type": "object",
   "additionalProperties": false,
   "required": ["schema_version", "producer", "source_artifact", "ledger_sha256", "ledger"],