Slim final output and rule decision payloads

content_agent/business_modules/result_source_lookup.py

@@ -44,12 +44,14 @@ def run(
         discovered_content_items,
         decision_by_target_id,
         media_by_platform_content_id,
+        paths_by_content_id,
     )
     reject_records = _build_reject_records(
         policy_run_id,
         discovered_content_items,
         decision_by_target_id,
         media_by_platform_content_id,
+        paths_by_content_id,
     )
     author_assets, author_asset_rows, author_role_rows = _build_author_assets(
         run_id,
@@ -101,9 +103,7 @@ def run(
                 "decision_reason_code": decision["decision_reason_code"],
                 "search_query_effect_status": decision["search_query_effect_status"],
                 "score": decision.get("score"),
-                "scorecard": decision.get("scorecard", {}),
-                "decision_replay_data": decision.get("decision_replay_data", {}),
-                "source_evidence": decision["source_evidence"],
+                "source_evidence_ref": _source_evidence_ref(decision),
                 **_v4_explanation_field(decision),
             }
             for decision in decisions
@@ -181,10 +181,7 @@ def _build_content_assets(
                     "rule_pack_version": decision["rule_pack_version"],
                     "strategy_version": decision["strategy_version"],
                     "source_path_record_ids": path_ids,
-                    "source_evidence": {
-                        **decision["source_evidence"],
-                        "source_path_record_ids": path_ids,
-                    },
+                    "source_evidence_ref": _source_evidence_ref(decision, path_ids),
                     "content_media_status": media_by_platform_content_id[
                         platform_content_id
                     ]["content_media_status"],
@@ -225,10 +222,7 @@ def _build_review_records(
                     "strategy_version": decision["strategy_version"],
                     "decision_reason_code": decision["decision_reason_code"],
                     "source_path_record_ids": path_ids,
-                    "source_evidence": {
-                        **decision["source_evidence"],
-                        "source_path_record_ids": path_ids,
-                    },
+                    "source_evidence_ref": _source_evidence_ref(decision, path_ids),
                     "content_media_status": media_by_platform_content_id[
                         platform_content_id
                     ]["content_media_status"],
@@ -245,6 +239,7 @@ def _build_reject_records(
     discovered_content_items: list[dict[str, Any]],
     decision_by_target_id: dict[str, dict[str, Any]],
     media_by_platform_content_id: dict[str, dict[str, Any]],
+    paths_by_content_id: dict[str, list[str]],
 ) -> list[dict[str, Any]]:
     reject_records: list[dict[str, Any]] = []
     for item in discovered_content_items:
@@ -252,6 +247,7 @@ def _build_reject_records(
         decision = decision_by_target_id[platform_content_id]
         if decision["decision_action"] != "REJECT_CONTENT":
             continue
+        path_ids = paths_by_content_id.get(platform_content_id, [])
         reject_records.append(
             _with_v4_explanation(
                 {
@@ -259,8 +255,9 @@ def _build_reject_records(
                     "policy_run_id": policy_run_id,
                     "main_decision_reason_code": decision["decision_reason_code"],
                     "decision_id": decision["decision_id"],
+                    "source_path_record_ids": path_ids,
                     "media_snapshot": _media_snapshot(media_by_platform_content_id.get(platform_content_id)),
-                    "source_evidence": decision["source_evidence"],
+                    "source_evidence_ref": _source_evidence_ref(decision, path_ids),
                 },
                 decision,
             )
@@ -273,6 +270,7 @@ def _build_technical_retry_records(
     discovered_content_items: list[dict[str, Any]],
     decision_by_target_id: dict[str, dict[str, Any]],
     media_by_platform_content_id: dict[str, dict[str, Any]],
+    paths_by_content_id: dict[str, list[str]],
 ) -> list[dict[str, Any]]:
     retry_records: list[dict[str, Any]] = []
     for item in discovered_content_items:
@@ -280,6 +278,7 @@ def _build_technical_retry_records(
         decision = decision_by_target_id[platform_content_id]
         if decision["decision_action"] != "TECHNICAL_RETRY_REQUIRED":
             continue
+        path_ids = paths_by_content_id.get(platform_content_id, [])
         retry_records.append(
             _with_v4_explanation(
                 {
@@ -288,8 +287,9 @@ def _build_technical_retry_records(
                     "main_decision_reason_code": decision["decision_reason_code"],
                     "decision_id": decision["decision_id"],
                     "technical_retry_status": "retry_required",
+                    "source_path_record_ids": path_ids,
                     "media_snapshot": _media_snapshot(media_by_platform_content_id.get(platform_content_id)),
-                    "source_evidence": decision["source_evidence"],
+                    "source_evidence_ref": _source_evidence_ref(decision, path_ids),
                 },
                 decision,
             )
@@ -312,6 +312,19 @@ def _media_snapshot(media: dict[str, Any] | None) -> dict[str, Any]:
     return {key: value for key, value in snapshot.items() if value is not None}
 
 
+def _source_evidence_ref(
+    decision: dict[str, Any],
+    source_path_record_ids: list[str] | None = None,
+) -> dict[str, Any]:
+    ref = {
+        "decision_id": decision.get("decision_id"),
+        "decision_target_id": decision.get("decision_target_id"),
+    }
+    if source_path_record_ids is not None:
+        ref["source_path_record_ids"] = source_path_record_ids
+    return {key: value for key, value in ref.items() if value is not None}
+
+
 def _with_v4_explanation(
     record: dict[str, Any],
     decision: dict[str, Any],
@@ -338,14 +351,12 @@ def _v4_decision_explanation(decision: dict[str, Any]) -> dict[str, Any]:
         "query_relevance_score": scorecard.get("query_relevance_score"),
         "platform_performance_score": scorecard.get("platform_performance_score"),
         "score": decision.get("score"),
-        "platform_performance_components": scorecard.get("platform_performance_components", []),
         "missing_observable_fields": scorecard.get("missing_observable_fields", []),
         "decision_action": decision.get("decision_action"),
         "decision_reason_code": decision.get("decision_reason_code"),
         "search_query_effect_status": decision.get("search_query_effect_status"),
         "allow_walk": replay_data.get("allow_walk"),
         "allow_walk_reason": replay_data.get("allow_walk_reason"),
-        "walk_gate_snapshot": replay_data.get("walk_gate_snapshot"),
     }
     for optional_field in [
         "score_missing",

content_agent/business_modules/rule_judgment/evaluator.py

@@ -17,6 +17,25 @@ V4_REVIEW_REASON = "v4_score_review_needed"
 V4_REJECT_REASON = "v4_query_or_score_below_threshold"
 V4_TECHNICAL_RETRY_REASON = "v4_technical_retry_needed"
 V4_TECHNICAL_RETRY_ACTION = "TECHNICAL_RETRY_REQUIRED"
+RULE_DECISION_RAW_PAYLOAD_EXCLUDE_KEYS = {
+    "run_id",
+    "policy_run_id",
+    "decision_id",
+    "policy_bundle_id",
+    "rule_pack_id",
+    "rule_pack_version",
+    "strategy_version",
+    "decision_target_type",
+    "decision_target_id",
+    "triggered_blocking_rules",
+    "scorecard",
+    "score",
+    "decision_action",
+    "decision_reason_code",
+    "search_query_effect_status",
+    "source_evidence",
+    "decision_replay_data",
+}
 # M9B:抖音 50+ 子分。仅当 bundle 带 content_audience_50plus 块(抖音)时计入;
 # 非抖音 bundle 无块 → 维持旧 0.5/0.5(零改动)。权重在 evaluator 计算,不进 rule pack
 # 维度(config 合同要求 active 维度恰为 query_relevance/platform_performance)。
@@ -385,7 +404,7 @@ def _build_decision(
             **replay_marker,
         },
     }
-    return with_raw_payload(decision)
+    return with_raw_payload(decision, exclude_keys=RULE_DECISION_RAW_PAYLOAD_EXCLUDE_KEYS)
 
 
 def _evaluate_hard_gates(bundle: dict[str, Any], gates: list[dict[str, Any]]) -> list[dict[str, Any]]:

content_agent/business_modules/run_record/validation.py

@@ -173,12 +173,18 @@ def compute_final_output_completeness(
     for record in final_output.get("reject_records", []):
         if record.get("decision_id") not in decision_ids:
             findings.append(f"reject_record missing decision: {record.get('decision_target_id')}")
+        if set(record.get("source_path_record_ids", [])) - path_ids:
+            findings.append(f"reject_record missing paths: {record.get('decision_target_id')}")
 
     for record in final_output.get("technical_retry_records", []):
         if record.get("decision_id") not in decision_ids:
             findings.append(
                 f"technical_retry_record missing decision: {record.get('decision_target_id')}"
             )
+        if set(record.get("source_path_record_ids", [])) - path_ids:
+            findings.append(
+                f"technical_retry_record missing paths: {record.get('decision_target_id')}"
+            )
 
     final_decision_ids = {
         record.get("decision_id") for record in final_output.get("decision_records", [])
@@ -485,6 +491,9 @@ def _check_references(data: dict[str, Any], findings: list[dict[str, Any]]) -> N
                     "missing_decision_ref",
                     f"{section} has unknown decision_id: {row.get('decision_id')}",
                 )
+            for path_id in row.get("source_path_record_ids", []):
+                if path_id not in path_ids:
+                    _fail(findings, "missing_path_ref", f"{section} has unknown path_id: {path_id}")
     for author_asset in final_output.get("author_assets", []):
         for decision_id in author_asset.get("decision_ids", []):
             if decision_id not in decision_ids:
@@ -545,45 +554,65 @@ def _check_source_evidence(data: dict[str, Any], findings: list[dict[str, Any]])
             f"decision {decision.get('decision_id')}",
         )
 
-    final_output = data.get("final_output.json", {})
-    for asset in final_output.get("content_assets", []):
-        _check_one_source_evidence(
-            findings,
-            asset.get("source_evidence") or {},
-            evidence_pack,
-            f"asset {asset.get('platform_content_id')}",
-        )
-    for record in final_output.get("reject_records", []):
-        _check_one_source_evidence(
-            findings,
-            record.get("source_evidence") or {},
-            evidence_pack,
-            f"reject {record.get('decision_target_id')}",
-        )
-    for record in final_output.get("review_records", []):
-        _check_one_source_evidence(
-            findings,
-            record.get("source_evidence") or {},
-            evidence_pack,
-            f"review {record.get('platform_content_id')}",
-        )
-    for record in final_output.get("technical_retry_records", []):
-        _check_one_source_evidence(
-            findings,
-            record.get("source_evidence") or {},
-            evidence_pack,
-            f"technical_retry {record.get('decision_target_id')}",
-        )
-    for record in final_output.get("decision_records", []):
-        _check_one_source_evidence(
-            findings,
-            record.get("source_evidence") or {},
-            evidence_pack,
-            f"decision record {record.get('decision_id')}",
-        )
+    _check_final_output_source_evidence_refs(data, findings)
     _check_final_decision_coverage(data, findings)
 
 
+def _check_final_output_source_evidence_refs(
+    data: dict[str, Any],
+    findings: list[dict[str, Any]],
+) -> None:
+    decisions_by_id = {
+        decision.get("decision_id"): decision
+        for decision in data.get("rule_decisions.jsonl", [])
+        if decision.get("decision_id")
+    }
+    final_output = data.get("final_output.json", {})
+    for section in [
+        "content_assets",
+        "reject_records",
+        "review_records",
+        "technical_retry_records",
+        "decision_records",
+    ]:
+        for record in final_output.get(section, []):
+            decision_id = record.get("decision_id")
+            decision = decisions_by_id.get(decision_id)
+            if not decision:
+                continue
+            ref = record.get("source_evidence_ref")
+            if ref is None:
+                if "source_evidence" in record:
+                    continue
+                _fail(
+                    findings,
+                    "source_evidence_ref_missing",
+                    f"{section} {decision_id} missing source_evidence_ref",
+                )
+                continue
+            if not isinstance(ref, dict):
+                _fail(
+                    findings,
+                    "source_evidence_ref_invalid",
+                    f"{section} {decision_id} has invalid source_evidence_ref",
+                )
+                continue
+            if ref.get("decision_id") != decision_id:
+                _fail(
+                    findings,
+                    "source_evidence_ref_mismatch",
+                    f"{section} {decision_id} ref points to wrong decision",
+                )
+            target_id = record.get("platform_content_id") or record.get("decision_target_id")
+            expected_target = decision.get("decision_target_id")
+            if target_id and expected_target and ref.get("decision_target_id") != expected_target:
+                _fail(
+                    findings,
+                    "source_evidence_ref_mismatch",
+                    f"{section} {decision_id} ref points to wrong target",
+                )
+
+
 def _check_one_source_evidence(
     findings: list[dict[str, Any]],
     source_evidence: dict[str, Any],
@@ -1054,6 +1083,7 @@ def _final_output_records_by_decision_id(
         "content_assets",
         "review_records",
         "reject_records",
+        "technical_retry_records",
         "decision_records",
     ]:
         for record in final_output.get(section, []) or []:
@@ -1097,7 +1127,6 @@ def _check_v4_explanation_record(
         "decision_reason_code": decision.get("decision_reason_code"),
         "allow_walk": replay_data.get("allow_walk"),
         "allow_walk_reason": replay_data.get("allow_walk_reason"),
-        "walk_gate_snapshot": replay_data.get("walk_gate_snapshot"),
     }
     for field_name, expected in expected_values.items():
         if expected is None and field_name not in explanation:

content_agent/integrations/database_runtime.py

@@ -4,12 +4,13 @@ import json
 import os
 from dataclasses import dataclass
 from datetime import datetime, timezone
+from decimal import Decimal
 from pathlib import Path
 from typing import Any, Callable
 
 import pymysql
 
-from content_agent.constants import DB_SCHEMA_VERSION
+from content_agent.constants import DB_SCHEMA_VERSION, RUNTIME_RECORD_SCHEMA_VERSION
 
 
 ConnectionFactory = Callable[[], Any]
@@ -127,6 +128,27 @@ JSONL_UPSERT_KEYS = {
     "run_events.jsonl": ("run_id", "policy_run_id", "event_id"),
 }
 
+RULE_DECISION_RUNTIME_COLUMNS = (
+    "run_id",
+    "policy_run_id",
+    "decision_id",
+    "policy_bundle_id",
+    "rule_pack_id",
+    "rule_pack_version",
+    "strategy_version",
+    "decision_target_type",
+    "decision_target_id",
+    "decision_action",
+    "decision_reason_code",
+    "search_query_effect_status",
+    "score",
+    "triggered_blocking_rules",
+    "scorecard",
+    "source_evidence",
+    "decision_replay_data",
+    "raw_payload",
+)
+
 
 @dataclass(frozen=True)
 class ContentSupplyDbConfig:
@@ -262,6 +284,13 @@ class DatabaseRuntimeStore:
 
     def read_jsonl(self, run_id: str, filename: str) -> list[dict[str, Any]]:
         table = _table_for_runtime_file(filename)
+        if filename == "rule_decisions.jsonl":
+            column_sql = ", ".join(f"`{column}`" for column in RULE_DECISION_RUNTIME_COLUMNS)
+            rows = self._fetch_all(
+                f"SELECT {column_sql} FROM `{table}` WHERE `run_id` = %s ORDER BY `id`",
+                (run_id,),
+            )
+            return [_runtime_rule_decision(row) for row in rows]
         rows = self._fetch_all(
             f"SELECT `raw_payload` FROM `{table}` WHERE `run_id` = %s ORDER BY `id`",
             (run_id,),
@@ -970,6 +999,41 @@ def _runtime_payload(payload: dict[str, Any]) -> dict[str, Any]:
     return {**payload, "raw_payload": dict(payload)}
 
 
+def _runtime_rule_decision(row: dict[str, Any]) -> dict[str, Any]:
+    raw_payload = _decode_json_payload(row.get("raw_payload")) or {}
+    if not isinstance(raw_payload, dict):
+        raw_payload = {}
+    formal: dict[str, Any] = {}
+    for column in RULE_DECISION_RUNTIME_COLUMNS:
+        if column == "raw_payload":
+            continue
+        value = row.get(column)
+        if column in JSON_COLUMNS_BY_TABLE["content_agent_rule_decisions"]:
+            value = _decode_json_payload(value)
+        formal[column] = _runtime_json_safe_value(value)
+    return {
+        "record_schema_version": raw_payload.get(
+            "record_schema_version",
+            RUNTIME_RECORD_SCHEMA_VERSION,
+        ),
+        **raw_payload,
+        **formal,
+        "raw_payload": raw_payload,
+    }
+
+
+def _runtime_json_safe_value(value: Any) -> Any:
+    if isinstance(value, Decimal):
+        return int(value) if value == value.to_integral_value() else float(value)
+    if isinstance(value, datetime):
+        return value.isoformat()
+    if isinstance(value, list):
+        return [_runtime_json_safe_value(item) for item in value]
+    if isinstance(value, dict):
+        return {key: _runtime_json_safe_value(child) for key, child in value.items()}
+    return value
+
+
 def _datetime_value(value: Any) -> Any:
     if value is None or isinstance(value, datetime):
         return value

content_agent/record_payload.py

@@ -1,9 +1,15 @@
 from __future__ import annotations
 
+from collections.abc import Iterable
 from copy import deepcopy
 from typing import Any
 
 
-def with_raw_payload(record: dict[str, Any]) -> dict[str, Any]:
-    payload = {key: deepcopy(value) for key, value in record.items() if key != "raw_payload"}
+def with_raw_payload(
+    record: dict[str, Any],
+    *,
+    exclude_keys: Iterable[str] = (),
+) -> dict[str, Any]:
+    excluded = {"raw_payload", *exclude_keys}
+    payload = {key: deepcopy(value) for key, value in record.items() if key not in excluded}
     return {**record, "raw_payload": payload}

tests/test_database_runtime.py

@@ -1,5 +1,6 @@
 import json
 import re
+from decimal import Decimal
 from pathlib import Path
 
 from content_agent.integrations.database_runtime import (
@@ -909,6 +910,58 @@ def test_database_runtime_read_jsonl_reconstructs_runtime_payload():
     assert rows[0]["raw_payload"]["search_query_id"] == "q_001"
 
 
+def test_database_runtime_read_jsonl_reconstructs_rule_decision_from_formal_columns():
+    connection = FakeConnection()
+    connection.select_all_result = [
+        {
+            "run_id": "run_001",
+            "policy_run_id": "policy_run_001",
+            "decision_id": "d_001",
+            "policy_bundle_id": "bundle_001",
+            "rule_pack_id": "rule_pack_001",
+            "rule_pack_version": "4.0.0",
+            "strategy_version": "V4",
+            "decision_target_type": "content",
+            "decision_target_id": "content_001",
+            "decision_action": "ADD_TO_CONTENT_POOL",
+            "decision_reason_code": "v4_query_and_platform_pass",
+            "search_query_effect_status": "success",
+            "score": Decimal("75.50"),
+            "triggered_blocking_rules": json.dumps([]),
+            "scorecard": json.dumps({"schema_version": "v4_scorecard.v1", "total_score": 75.5}),
+            "source_evidence": json.dumps({"source_post_id": "post_001"}),
+            "decision_replay_data": json.dumps({"allow_walk": True}),
+            "raw_payload": json.dumps(
+                {
+                    "record_schema_version": "runtime_record.v1",
+                    "strategy_id": "strategy_001",
+                    "policy_bundle_hash": "hash_001",
+                }
+            ),
+        }
+    ]
+    store = DatabaseRuntimeStore(_config(), connection_factory=lambda: connection)
+
+    rows = store.read_jsonl("run_001", "rule_decisions.jsonl")
+
+    sql, params = connection.statements[-1]
+    assert "FROM `content_agent_rule_decisions`" in sql
+    assert "`scorecard`" in sql
+    assert params == ["run_001"]
+    row = rows[0]
+    assert row["record_schema_version"] == "runtime_record.v1"
+    assert row["decision_id"] == "d_001"
+    assert row["score"] == 75.5
+    assert row["scorecard"]["schema_version"] == "v4_scorecard.v1"
+    assert row["source_evidence"]["source_post_id"] == "post_001"
+    assert row["decision_replay_data"]["allow_walk"] is True
+    assert row["raw_payload"] == {
+        "record_schema_version": "runtime_record.v1",
+        "strategy_id": "strategy_001",
+        "policy_bundle_hash": "hash_001",
+    }
+
+
 def test_database_runtime_rejects_forbidden_raw_payload_keys_in_lists():
     connection = FakeConnection()
     store = DatabaseRuntimeStore(_config(), connection_factory=lambda: connection)

tests/test_p7_final_output.py

@@ -34,8 +34,17 @@ def test_reject_records_carry_source_evidence_refs(tmp_path):
     service, run_id = _start_mock_run(tmp_path)
 
     final_output = service.read_json(run_id, "final_output.json")
-    assert final_output["reject_records"][0]["source_evidence"]["source_post_id"]
-    assert final_output["reject_records"][0]["source_evidence"]["discovered_platform_content_id"]
+    reject = final_output["reject_records"][0]
+    assert "source_evidence" not in reject
+    assert reject["source_evidence_ref"]["decision_id"] == reject["decision_id"]
+    assert reject["source_evidence_ref"]["decision_target_id"] == reject["decision_target_id"]
+    decisions = {
+        decision["decision_id"]: decision
+        for decision in service.read_jsonl(run_id, "rule_decisions.jsonl")
+    }
+    source_evidence = decisions[reject["decision_id"]]["source_evidence"]
+    assert source_evidence["source_post_id"]
+    assert source_evidence["discovered_platform_content_id"]
     assert service.validate_run(run_id)["status"] == "pass"
 
 
@@ -57,19 +66,20 @@ def test_final_output_carries_v4_explanation_records(tmp_path):
     v4_decision_records = [
         record
         for record in final_output["decision_records"]
-        if (record.get("scorecard") or {}).get("schema_version") == "v4_scorecard.v1"
+        if (record.get("v4_explanation") or {}).get("scorecard_schema_version") == "v4_scorecard.v1"
     ]
 
     assert v4_decision_records
     for record in v4_decision_records:
+        assert "scorecard" not in record
+        assert "decision_replay_data" not in record
+        assert "source_evidence" not in record
         explanation = record["v4_explanation"]
         assert explanation["schema_version"] == "v4_decision_explanation.v1"
         assert explanation["scorecard_schema_version"] == "v4_scorecard.v1"
-        assert explanation["query_relevance_score"] == record["scorecard"]["query_relevance_score"]
-        assert explanation["platform_performance_score"] == record["scorecard"]["platform_performance_score"]
         assert explanation["score"] == record["score"]
         assert "allow_walk" in explanation
-        assert "walk_gate_snapshot" in explanation
+        assert "walk_gate_snapshot" not in explanation
 
     v4_ids = {record["decision_id"] for record in v4_decision_records}
     section_records = (

tests/test_runtime_files.py

@@ -60,9 +60,11 @@ def test_runtime_files_are_parseable_and_consistent(tmp_path):
     recall_evidence = service.read_jsonl(run_id, "pattern_recall_evidence.jsonl")
     walk_actions = service.read_jsonl(run_id, "walk_actions.jsonl")
     for row in [*items, *media_records, *recall_evidence, *decisions, *walk_actions, *paths]:
+        assert row["run_id"] == run_id
         assert row["policy_run_id"] == policy_run_id
         assert row["record_schema_version"] == "runtime_record.v1"
-        assert row["raw_payload"]["run_id"] == run_id
+        if "run_id" in row["raw_payload"]:
+            assert row["raw_payload"]["run_id"] == run_id
     assert {row["walk_status"] for row in walk_actions} <= {
         "success",
         "pending",
@@ -419,12 +421,19 @@ def test_runtime_validation_catches_missing_final_decision_record(tmp_path):
 def test_runtime_validation_catches_platform_content_id_source_pollution(tmp_path):
     service, run_id = _start_mock_run(tmp_path)
 
-    final_output_path = service.runtime.run_dir(run_id) / "final_output.json"
-    final_output = json.loads(final_output_path.read_text(encoding="utf-8"))
-    source_evidence = final_output["decision_records"][0]["source_evidence"]
+    decisions_path = service.runtime.run_dir(run_id) / "rule_decisions.jsonl"
+    decisions = [
+        json.loads(line)
+        for line in decisions_path.read_text(encoding="utf-8").splitlines()
+        if line.strip()
+    ]
+    source_evidence = decisions[0]["source_evidence"]
     source_evidence["source_post_id"] = source_evidence["discovered_platform_content_id"]
-    final_output_path.write_text(
-        json.dumps(final_output, ensure_ascii=False, indent=2) + "\n",
+    decisions_path.write_text(
+        "".join(
+            json.dumps(decision, ensure_ascii=False, separators=(",", ":")) + "\n"
+            for decision in decisions
+        ),
         encoding="utf-8",
     )
 

tests/test_v4_m3_scoring_replay.py

@@ -119,8 +119,9 @@ def test_v4_m3_db_runtime_preserves_scoring_json_containers(tmp_path):
     decision_raw = json.loads(decision_row["raw_payload"])
     assert scorecard["schema_version"] == "v4_scorecard.v1"
     assert replay_data["allow_walk"] in {True, False}
-    assert decision_raw["scorecard"] == scorecard
-    assert decision_raw["decision_replay_data"] == replay_data
+    assert "scorecard" not in decision_raw
+    assert "decision_replay_data" not in decision_raw
+    assert decision_raw["record_schema_version"] == "runtime_record.v1"
 
 
 def _table_name(sql: str) -> str: