diff --git a/diagnostics/kb_proposal_routes.py b/diagnostics/kb_proposal_routes.py
index 7dcbf09..f209ee6 100644
--- a/diagnostics/kb_proposal_routes.py
+++ b/diagnostics/kb_proposal_routes.py
@@ -31,6 +31,7 @@ for scripts_dir in SCRIPT_DIR_CANDIDATES:
     if str(scripts_dir) not in sys.path:
         sys.path.insert(0, str(scripts_dir))
 
+import kb_proposal_normalize as proposal_normalize  # noqa: E402
 import kb_proposal_review_packet as proposal_review  # noqa: E402
 
 logger = logging.getLogger("argus.kb_proposals")
@@ -83,7 +84,12 @@ def _db_args(filters: dict[str, Any]) -> argparse.Namespace:
 def _load_packets(request: web.Request, filters: dict[str, Any]) -> list[dict[str, Any]]:
     loader = request.app.get(KB_PROPOSAL_LOADER_KEY)
     proposals = loader(_db_args(filters)) if loader else proposal_review.load_from_db(_db_args(filters))
-    return [proposal_review.classify_proposal(proposal) for proposal in proposals]
+    packets = []
+    for proposal in proposals:
+        packet = proposal_review.classify_proposal(proposal)
+        packet["normalization_preview"] = proposal_normalize.normalize_proposal(proposal)
+        packets.append(packet)
+    return packets
 
 
 def _counts(packets: list[dict[str, Any]], key: str) -> dict[str, int]:
@@ -155,9 +161,25 @@ def _packet_card(packet: dict[str, Any]) -> str:
         ("Supersession edges", str(summary.get("supersession_edge_count", 0))),
         ("Missing contract", _join_values(applyability.get("missing_contract") or [])),
     ]
+    normalization = packet.get("normalization_preview") or {}
+    if normalization:
+        rows.extend(
+            [
+                ("Normalization", _badge(normalization.get("normalization_state"))),
+                ("Strict child proposals", str(normalization.get("strict_child_count", 0))),
+                ("Blocked fragments", str(normalization.get("blocked_count", 0))),
+            ]
+        )
     details = "\n".join(
         f"<tr><th>{escape(label)}</th><td>{value}</td></tr>" for label, value in rows
     )
+    normalization_html = ""
+    if normalization:
+        normalization_html = f"""
+  <div class="next-action">
+    <div class="label">Normalization action</div>
+    <p>{escape(str(normalization.get("next_normalization_action") or "No normalization action recorded."))}</p>
+  </div>"""
     return f"""
 <article class="proposal-card">
   <div class="proposal-card-header">
@@ -170,6 +192,7 @@ def _packet_card(packet: dict[str, Any]) -> str:
     <div class="label">Next admin action</div>
     <p>{next_action}</p>
   </div>
+  {normalization_html}
 </article>"""
 
 
diff --git a/scripts/kb_proposal_normalize.py b/scripts/kb_proposal_normalize.py
new file mode 100644
index 0000000..74127da
--- /dev/null
+++ b/scripts/kb_proposal_normalize.py
@@ -0,0 +1,345 @@
+#!/usr/bin/env python3
+"""Preview strict apply-payload normalization for staged KB proposals.
+
+Leo and reviewers can approve rich proposal intent before it is mechanically
+safe to apply. The apply worker intentionally accepts only strict contracts:
+``add_edge``, ``attach_evidence``, and ``revise_strategy`` proposals that carry
+``payload.apply_payload`` with canonical IDs.
+
+This helper is read-only by default. It inspects a freeform proposal and reports:
+
+* strict child proposals that can be emitted because all canonical IDs exist in
+  the proposal payload;
+* blocked fragments that still require claim/source creation, source dedup, or a
+  schema decision before canonical apply is possible.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import re
+import sys
+from pathlib import Path
+from typing import Any
+
+HERE = Path(__file__).resolve().parent
+sys.path.insert(0, str(HERE))
+
+import apply_proposal as ap  # noqa: E402
+import kb_proposal_review_packet as review_packet  # noqa: E402
+
+UUID_RE = re.compile(
+    r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$"
+)
+
+
+def _payload_dict(payload: dict[str, Any], key: str) -> dict[str, Any]:
+    value = payload.get(key)
+    return value if isinstance(value, dict) else {}
+
+
+def _payload_list(payload: dict[str, Any], key: str) -> list[Any]:
+    value = payload.get(key)
+    return value if isinstance(value, list) else []
+
+
+def _is_uuid(value: Any) -> bool:
+    return isinstance(value, str) and bool(UUID_RE.match(value))
+
+
+def _first_uuid(fragment: dict[str, Any], keys: tuple[str, ...]) -> str | None:
+    for key in keys:
+        value = fragment.get(key)
+        if _is_uuid(value):
+            return value
+    return None
+
+
+def _fingerprint(value: Any) -> str:
+    raw = json.dumps(value, sort_keys=True, separators=(",", ":"))
+    return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:16]
+
+
+def _compact_fragment(fragment: Any) -> dict[str, Any]:
+    if not isinstance(fragment, dict):
+        return {"value_type": type(fragment).__name__, "fingerprint": _fingerprint(fragment)}
+    keys = [
+        "claim_key",
+        "source_key",
+        "headline",
+        "title",
+        "edge_type",
+        "target",
+        "target_claim_id",
+        "to_claim_id",
+        "to_existing_claim",
+        "from_claim_id",
+        "from_claim",
+        "claim_id",
+        "source_id",
+    ]
+    compact = {key: fragment[key] for key in keys if key in fragment}
+    compact["fingerprint"] = _fingerprint(fragment)
+    return compact
+
+
+def _blocked(kind: str, reason: str, missing: list[str], fragment: Any) -> dict[str, Any]:
+    return {
+        "kind": kind,
+        "reason": reason,
+        "missing": missing,
+        "fragment": _compact_fragment(fragment),
+    }
+
+
+def _edge_candidate(fragment: dict[str, Any], default_from_claim: str | None = None) -> dict[str, Any] | None:
+    edge_type = fragment.get("edge_type")
+    from_claim = _first_uuid(fragment, ("from_claim", "from_claim_id", "from_claim_uuid")) or default_from_claim
+    to_claim = _first_uuid(
+        fragment,
+        (
+            "to_claim",
+            "to_claim_id",
+            "target_claim_id",
+            "target_claim",
+            "to_existing_claim",
+            "existing_claim_id",
+        ),
+    )
+    if from_claim and to_claim and edge_type:
+        result: dict[str, Any] = {"from_claim": from_claim, "to_claim": to_claim, "edge_type": edge_type}
+        if fragment.get("weight") is not None:
+            result["weight"] = fragment.get("weight")
+        return result
+    return None
+
+
+def _evidence_candidate(fragment: dict[str, Any], default_claim_id: str | None = None) -> dict[str, Any] | None:
+    claim_id = _first_uuid(fragment, ("claim_id", "target_claim_id", "to_claim_id")) or default_claim_id
+    source_id = _first_uuid(fragment, ("source_id", "source_uuid", "canonical_source_id"))
+    if claim_id and source_id:
+        result: dict[str, Any] = {"claim_id": claim_id, "source_id": source_id}
+        if fragment.get("role"):
+            result["role"] = fragment.get("role")
+        if fragment.get("weight") is not None:
+            result["weight"] = fragment.get("weight")
+        return result
+    return None
+
+
+def _strict_child(kind: str, apply_payload: dict[str, Any], source_fragment: Any) -> dict[str, Any]:
+    return {
+        "proposal_type": kind,
+        "payload": {"apply_payload": apply_payload},
+        "source_fragment_fingerprint": _fingerprint(source_fragment),
+    }
+
+
+def _normalize_edges(payload: dict[str, Any]) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
+    strict: list[dict[str, Any]] = []
+    blocked: list[dict[str, Any]] = []
+
+    old_claim_id = _first_uuid(_payload_dict(payload, "old_claim"), ("id", "claim_id"))
+
+    edge_sources: list[tuple[dict[str, Any], str | None]] = []
+    for edge in _payload_list(payload, "proposed_supersession_edges"):
+        if isinstance(edge, dict):
+            edge_sources.append((edge, None))
+    for candidate in _payload_list(payload, "claim_candidates"):
+        if not isinstance(candidate, dict):
+            continue
+        candidate_claim_id = _first_uuid(candidate, ("claim_id", "id", "canonical_claim_id"))
+        for edge in _payload_list(candidate, "edges"):
+            if isinstance(edge, dict):
+                edge_sources.append((edge, candidate_claim_id))
+        if candidate_claim_id and old_claim_id:
+            edge_sources.append(
+                (
+                    {
+                        "from_claim": candidate_claim_id,
+                        "to_claim": old_claim_id,
+                        "edge_type": "supersedes_component_of",
+                    },
+                    None,
+                )
+            )
+
+    seen = set()
+    for fragment, default_from_claim in edge_sources:
+        candidate = _edge_candidate(fragment, default_from_claim)
+        if candidate:
+            key = json.dumps(candidate, sort_keys=True)
+            if key not in seen:
+                seen.add(key)
+                strict.append(_strict_child("add_edge", candidate, fragment))
+            continue
+        missing = []
+        if not fragment.get("edge_type"):
+            missing.append("edge_type")
+        if not (_first_uuid(fragment, ("from_claim", "from_claim_id", "from_claim_uuid")) or default_from_claim):
+            missing.append("canonical from_claim")
+        if not _first_uuid(
+            fragment,
+            ("to_claim", "to_claim_id", "target_claim_id", "target_claim", "to_existing_claim", "existing_claim_id"),
+        ):
+            missing.append("canonical to_claim")
+        blocked.append(_blocked("edge", "edge fragment is not a strict add_edge contract", missing, fragment))
+
+    return strict, blocked
+
+
+def _normalize_evidence(payload: dict[str, Any]) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
+    evidence_rows: list[dict[str, Any]] = []
+    blocked: list[dict[str, Any]] = []
+
+    evidence_sources: list[tuple[dict[str, Any], str | None]] = []
+    for evidence in _payload_list(payload, "evidence"):
+        if isinstance(evidence, dict):
+            evidence_sources.append((evidence, None))
+    for evidence in _payload_list(payload, "evidence_candidates"):
+        if isinstance(evidence, dict):
+            evidence_sources.append((evidence, None))
+    for candidate in _payload_list(payload, "claim_candidates"):
+        if not isinstance(candidate, dict):
+            continue
+        candidate_claim_id = _first_uuid(candidate, ("claim_id", "id", "canonical_claim_id"))
+        for evidence in _payload_list(candidate, "evidence"):
+            if isinstance(evidence, dict):
+                evidence_sources.append((evidence, candidate_claim_id))
+
+    seen = set()
+    for fragment, default_claim_id in evidence_sources:
+        candidate = _evidence_candidate(fragment, default_claim_id)
+        if candidate:
+            key = json.dumps(candidate, sort_keys=True)
+            if key not in seen:
+                seen.add(key)
+                evidence_rows.append(candidate)
+            continue
+        missing = []
+        if not (_first_uuid(fragment, ("claim_id", "target_claim_id", "to_claim_id")) or default_claim_id):
+            missing.append("canonical claim_id")
+        if not _first_uuid(fragment, ("source_id", "source_uuid", "canonical_source_id")):
+            missing.append("canonical source_id")
+        blocked.append(_blocked("evidence", "evidence fragment is not a strict attach_evidence contract", missing, fragment))
+
+    strict = [_strict_child("attach_evidence", {"evidence": evidence_rows}, evidence_rows)] if evidence_rows else []
+    return strict, blocked
+
+
+def _normalize_strategy(proposal: dict[str, Any], payload: dict[str, Any]) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
+    if proposal.get("proposal_type") != "revise_strategy":
+        return [], []
+    strategy_payload = payload.get("apply_payload") or {
+        key: payload.get(key) for key in ("agent_id", "strategy", "strategy_nodes") if key in payload
+    }
+    has_contract = all(strategy_payload.get(key) for key in ("agent_id", "strategy", "strategy_nodes"))
+    if has_contract:
+        return [_strict_child("revise_strategy", strategy_payload, payload)], []
+    return [], [
+        _blocked(
+            "strategy",
+            "strategy revision needs agent_id, strategy, and strategy_nodes",
+            [key for key in ("agent_id", "strategy", "strategy_nodes") if not strategy_payload.get(key)],
+            payload,
+        )
+    ]
+
+
+def _dedupe_children(children: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    seen = set()
+    out = []
+    for child in children:
+        key = json.dumps({"proposal_type": child["proposal_type"], "payload": child["payload"]}, sort_keys=True)
+        if key in seen:
+            continue
+        seen.add(key)
+        out.append(child)
+    return out
+
+
+def normalize_proposal(proposal: dict[str, Any]) -> dict[str, Any]:
+    payload = proposal.get("payload") or {}
+    if not isinstance(payload, dict):
+        payload = {}
+
+    if payload.get("apply_payload") and proposal.get("proposal_type") in ap.APPLYABLE_TYPES:
+        return {
+            "proposal_id": proposal.get("id"),
+            "normalization_state": "already_strict_apply_payload",
+            "strict_child_proposals": [],
+            "blocked_fragments": [],
+            "blocked_count": 0,
+            "strict_child_count": 0,
+            "next_normalization_action": "Proposal already carries a strict apply_payload; use dry-run/apply review.",
+        }
+
+    strict_children: list[dict[str, Any]] = []
+    blocked_fragments: list[dict[str, Any]] = []
+    for strict, blocked in (
+        _normalize_edges(payload),
+        _normalize_evidence(payload),
+        _normalize_strategy(proposal, payload),
+    ):
+        strict_children.extend(strict)
+        blocked_fragments.extend(blocked)
+
+    strict_children = _dedupe_children(strict_children)
+    if strict_children and not blocked_fragments:
+        state = "strict_children_ready"
+        next_action = "Review generated strict child proposal JSON, then stage as approved apply_payload proposals."
+    elif strict_children:
+        state = "partial_strict_children_ready"
+        next_action = "Stage only the strict children after review; resolve blocked fragments separately."
+    elif blocked_fragments:
+        state = "blocked_missing_canonical_ids"
+        next_action = "Create or identify canonical claim/source rows, then rerun normalization."
+    else:
+        state = "no_supported_apply_fragments"
+        next_action = "Rewrite as a supported add_edge, attach_evidence, or revise_strategy apply contract."
+
+    return {
+        "proposal_id": proposal.get("id"),
+        "normalization_state": state,
+        "strict_child_proposals": strict_children,
+        "blocked_fragments": blocked_fragments,
+        "blocked_count": len(blocked_fragments),
+        "strict_child_count": len(strict_children),
+        "next_normalization_action": next_action,
+    }
+
+
+def load_from_input(path: str) -> list[dict[str, Any]]:
+    return review_packet.load_from_input(path)
+
+
+def load_from_db(args: argparse.Namespace) -> list[dict[str, Any]]:
+    return review_packet.load_from_db(args)
+
+
+def parse_args(argv: list[str]) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--input-json", help="normalize proposal JSON from a file")
+    parser.add_argument("--proposal-id", help="load one live proposal from Postgres")
+    parser.add_argument("--status", default="approved", help="load live proposals by status")
+    parser.add_argument("--limit", type=int, default=20)
+    parser.add_argument("--secrets-file", default=ap.DEFAULT_SECRETS_FILE)
+    parser.add_argument("--container", default=ap.DEFAULT_CONTAINER)
+    parser.add_argument("--db", default=ap.DEFAULT_DB)
+    parser.add_argument("--host", default=ap.DEFAULT_HOST)
+    parser.add_argument("--role", default=ap.DEFAULT_ROLE)
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parse_args(sys.argv[1:] if argv is None else argv)
+    proposals = load_from_input(args.input_json) if args.input_json else load_from_db(args)
+    previews = [normalize_proposal(proposal) for proposal in proposals]
+    print(json.dumps({"normalization_previews": previews}, indent=2, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/tests/test_kb_proposal_normalize.py b/tests/test_kb_proposal_normalize.py
new file mode 100644
index 0000000..d6e6607
--- /dev/null
+++ b/tests/test_kb_proposal_normalize.py
@@ -0,0 +1,131 @@
+"""Tests for scripts/kb_proposal_normalize.py."""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(REPO_ROOT / "scripts"))
+
+import kb_proposal_normalize as norm  # noqa: E402
+
+
+def test_freeform_claim_split_is_blocked_until_canonical_ids_exist():
+    preview = norm.normalize_proposal(
+        {
+            "id": "p-freeform",
+            "proposal_type": "attach_evidence",
+            "status": "approved",
+            "payload": {
+                "old_claim": {"id": "d0000000-0000-0000-0000-000000000005"},
+                "claim_candidates": [
+                    {
+                        "claim_key": "new_claim_key",
+                        "headline": "New claim headline",
+                        "edges": [
+                            {
+                                "edge_type": "supersedes_component_of",
+                                "target_claim_id": "d0000000-0000-0000-0000-000000000005",
+                            }
+                        ],
+                        "evidence": [{"source_key": "human_review"}],
+                    }
+                ],
+                "source_candidates": [{"source_key": "human_review"}],
+            },
+        }
+    )
+
+    assert preview["normalization_state"] == "blocked_missing_canonical_ids"
+    assert preview["strict_child_count"] == 0
+    assert preview["blocked_count"] == 2
+    assert preview["blocked_fragments"][0]["missing"] == ["canonical from_claim"]
+    assert preview["blocked_fragments"][1]["missing"] == ["canonical claim_id", "canonical source_id"]
+    assert "Create or identify canonical claim/source rows" in preview["next_normalization_action"]
+
+
+def test_exact_edge_fragment_emits_strict_add_edge_child():
+    preview = norm.normalize_proposal(
+        {
+            "id": "p-edge",
+            "proposal_type": "attach_evidence",
+            "status": "approved",
+            "payload": {
+                "proposed_supersession_edges": [
+                    {
+                        "from_claim": "11111111-1111-1111-1111-111111111111",
+                        "to_claim_id": "22222222-2222-2222-2222-222222222222",
+                        "edge_type": "supports",
+                        "weight": 0.7,
+                    }
+                ]
+            },
+        }
+    )
+
+    assert preview["normalization_state"] == "strict_children_ready"
+    assert preview["blocked_count"] == 0
+    assert preview["strict_child_count"] == 1
+    child = preview["strict_child_proposals"][0]
+    assert child["proposal_type"] == "add_edge"
+    assert child["payload"]["apply_payload"] == {
+        "from_claim": "11111111-1111-1111-1111-111111111111",
+        "to_claim": "22222222-2222-2222-2222-222222222222",
+        "edge_type": "supports",
+        "weight": 0.7,
+    }
+
+
+def test_exact_evidence_fragments_emit_one_attach_evidence_child():
+    preview = norm.normalize_proposal(
+        {
+            "id": "p-evidence",
+            "proposal_type": "attach_evidence",
+            "status": "approved",
+            "payload": {
+                "evidence": [
+                    {
+                        "claim_id": "33333333-3333-3333-3333-333333333333",
+                        "source_id": "44444444-4444-4444-4444-444444444444",
+                        "role": "grounds",
+                        "weight": 0.9,
+                    }
+                ]
+            },
+        }
+    )
+
+    assert preview["normalization_state"] == "strict_children_ready"
+    assert preview["strict_child_count"] == 1
+    child = preview["strict_child_proposals"][0]
+    assert child["proposal_type"] == "attach_evidence"
+    assert child["payload"]["apply_payload"]["evidence"] == [
+        {
+            "claim_id": "33333333-3333-3333-3333-333333333333",
+            "source_id": "44444444-4444-4444-4444-444444444444",
+            "role": "grounds",
+            "weight": 0.9,
+        }
+    ]
+
+
+def test_existing_apply_payload_is_already_strict():
+    preview = norm.normalize_proposal(
+        {
+            "id": "p-strict",
+            "proposal_type": "add_edge",
+            "status": "approved",
+            "payload": {
+                "apply_payload": {
+                    "from_claim": "11111111-1111-1111-1111-111111111111",
+                    "to_claim": "22222222-2222-2222-2222-222222222222",
+                    "edge_type": "supports",
+                }
+            },
+        }
+    )
+
+    assert preview["normalization_state"] == "already_strict_apply_payload"
+    assert preview["strict_child_count"] == 0
+    assert preview["blocked_count"] == 0
diff --git a/tests/test_kb_proposal_routes.py b/tests/test_kb_proposal_routes.py
index 03e717c..ddf28f0 100644
--- a/tests/test_kb_proposal_routes.py
+++ b/tests/test_kb_proposal_routes.py
@@ -82,11 +82,15 @@ def test_api_kb_proposals_returns_review_packets_and_counts():
     assert data["review_state_counts"]["approved_needs_apply_payload"] == 1
     assert data["review_state_counts"]["approved_applyable"] == 1
     assert data["packets"][0]["identity_and_authorization"]["source_ref"] == "telegram:leo:m3taversal:claim-review"
+    assert data["packets"][0]["normalization_preview"]["normalization_state"] == "blocked_missing_canonical_ids"
+    assert data["packets"][1]["normalization_preview"]["normalization_state"] == "already_strict_apply_payload"
 
 
 def test_page_renders_applyability_and_next_admin_action():
+    packet = routes.proposal_review.classify_proposal(_approved_freeform_proposal())
+    packet["normalization_preview"] = routes.proposal_normalize.normalize_proposal(_approved_freeform_proposal())
     data = routes.build_kb_proposal_response(
-        [routes.proposal_review.classify_proposal(_approved_freeform_proposal())],
+        [packet],
         {"status": "approved", "proposal_id": "", "limit": 20},
     )
 
@@ -97,6 +101,8 @@ def test_page_renders_applyability_and_next_admin_action():
     assert "strict payload.apply_payload" in html
     assert "claim-row normalization" in html
     assert "normalize into one or more strict apply_payload" in html
+    assert "Normalization" in html
+    assert "blocked_missing_canonical_ids" in html
 
 
 def test_api_kb_proposals_supports_proposal_id_filter():