diff --git a/src/agent_passport/agora.py b/src/agent_passport/agora.py
index 4c2ce5f..ffd2b3b 100644
--- a/src/agent_passport/agora.py
+++ b/src/agent_passport/agora.py
@@ -13,7 +13,7 @@
 from typing import Any, Optional
 
 from .crypto import sign, verify
-from .canonical import canonicalize
+from .canonical import canonicalize, canonicalize_for_write
 
 
 def _msg_id() -> str:
@@ -73,7 +73,7 @@ def create_agora_message(
     if reply_to:
         message_content["replyTo"] = reply_to
 
-    canonical = canonicalize(message_content)
+    canonical = canonicalize_for_write(message_content)
     signature = sign(canonical, private_key)
 
     return {**message_content, "signature": signature}
diff --git a/src/agent_passport/canonical.py b/src/agent_passport/canonical.py
index 49db8e8..2b32b56 100644
--- a/src/agent_passport/canonical.py
+++ b/src/agent_passport/canonical.py
@@ -12,6 +12,12 @@
 import math
 import re
 
+from .write_policy import (
+    MAX_SAFE_INTEGER,
+    UnsafeIntegerError,
+    assert_write_safe_numbers,
+)
+
 
 class JCSCanonicalizationError(ValueError):
     """A value cannot be canonicalized under RFC 8785.
@@ -163,6 +169,68 @@ def canonicalize(obj) -> str:
     return json.dumps(obj, ensure_ascii=False)
 
 
+def canonicalize_for_write(obj, path: str = "$") -> str:
+    """Legacy canonicalization for a NEW WRITE, with the APS unsafe-integer rule applied.
+
+    Byte-identical to :func:`canonicalize` for every value it accepts: same key order,
+    same null stripping, same number and string formatting. The only difference is that
+    an integer-valued number outside the interoperable IEEE 754 range is refused rather
+    than emitted. See :mod:`agent_passport.write_policy` for the rule.
+
+    Use at signing and new-write boundaries ONLY. Verification, recompute, and any path
+    rebuilding the preimage of an existing artifact must keep calling
+    :func:`canonicalize`, which stays unrestricted so historical bytes keep verifying.
+
+    READS EACH KEY EXACTLY ONCE. A separate validating pre-pass followed by
+    :func:`canonicalize` would walk the mapping twice, and a Mapping subclass or an
+    object overriding ``__getitem__`` can answer differently on the second read, so the
+    value checked would not be the value signed. Here the value is captured once into
+    ``val``, checked, and emitted from that same capture.
+
+    Note this deliberately does NOT fix the legacy int-verbatim divergence from the other
+    SDKs: refusing the unsafe range makes that divergence unreachable on write without
+    altering a single historical byte.
+    """
+    if obj is None:
+        return "null"
+    if isinstance(obj, bool):
+        return "true" if obj else "false"
+    if isinstance(obj, int):
+        if abs(obj) > MAX_SAFE_INTEGER:
+            raise UnsafeIntegerError(f"{path}: integer exceeds the interoperable IEEE 754 range")
+        return json.dumps(obj)
+    if isinstance(obj, float):
+        if math.isnan(obj) or math.isinf(obj):
+            raise ValueError(f"Cannot canonicalize {obj}: NaN and Infinity are not valid JSON per RFC 8259")
+        if obj.is_integer() and abs(obj) > MAX_SAFE_INTEGER:
+            raise UnsafeIntegerError(f"{path}: integer exceeds the interoperable IEEE 754 range")
+        return _es_number(obj)
+    if isinstance(obj, str):
+        return json.dumps(obj, ensure_ascii=False)
+    if isinstance(obj, list):
+        return "[" + ",".join(
+            canonicalize_for_write(item, f"{path}[{index}]") for index, item in enumerate(obj)
+        ) + "]"
+    if isinstance(obj, dict):
+        pairs = []
+        for key in _canonical_keys(obj.keys()):
+            val = obj[key]  # the single read; everything below uses this capture
+            if val is None:
+                continue
+            pairs.append(
+                json.dumps(key, ensure_ascii=False)
+                + ":"
+                + canonicalize_for_write(val, f"{path}.{key}")
+            )
+        return "{" + ",".join(pairs) + "}"
+    # Anything that reaches here (a tuple, in practice) is emitted by the same json.dumps
+    # fallback the READ twin uses, so the bytes stay identical. Validate it first, or an
+    # out-of-range integer inside a tuple would be signed unchecked. A second read is safe
+    # here precisely because the values that reach this branch are immutable.
+    assert_write_safe_numbers(obj, path)
+    return json.dumps(obj, ensure_ascii=False)
+
+
 def canonicalize_jcs(obj) -> str:
     """RFC 8785 JSON Canonicalization Scheme (strict).
 
@@ -179,11 +247,31 @@ def canonicalize_jcs(obj) -> str:
     if obj is None:
         return "null"
     if isinstance(obj, bool):
+        # Must stay ahead of the int branch: bool is a subclass of int in Python,
+        # and reordering would serialize True as 1.
         return "true" if obj else "false"
     if isinstance(obj, int):
-        return json.dumps(obj)
+        # RFC 8785 section 3.2.2.3 defines the JCS number domain as IEEE-754
+        # binary64 serialized under ECMAScript Number::toString. Python's int is
+        # arbitrary precision, so emitting it verbatim preserves a decimal
+        # spelling the double does not have: 2**60 would serialize as
+        # 1152921504606846976 where the double is 1152921504606847000. Widen to
+        # binary64 first, then take exactly the same path a float takes.
+        try:
+            as_double = float(obj)
+        except OverflowError:
+            as_double = math.inf
+        if math.isinf(as_double):
+            raise JCSCanonicalizationError(
+                f"canonicalize_jcs: integer {obj} exceeds the IEEE-754 binary64 "
+                "range and has no representation in the RFC 8785 number domain",
+                reason="number_out_of_double_range",
+            )
+        return _es_number(as_double)
     if isinstance(obj, float):
-        import math
+        # math is imported at module scope. A local "import math" here would make
+        # the name function-local for the whole of canonicalize_jcs and shadow it
+        # from the int branch above.
         if math.isnan(obj) or math.isinf(obj):
             raise ValueError(f"Cannot canonicalize {obj}")
         return _es_number(obj)
@@ -206,3 +294,57 @@ def canonicalize_jcs(obj) -> str:
             )
         return "{" + ",".join(pairs) + "}"
     return json.dumps(obj, ensure_ascii=False)
+
+
+def canonicalize_jcs_for_write(obj, path: str = "$") -> str:
+    """RFC 8785 canonicalization for a NEW WRITE, with the unsafe-integer rule applied.
+
+    Byte-identical to :func:`canonicalize_jcs` for every value it accepts. The only
+    difference is that an integer-valued number outside the interoperable IEEE 754 range
+    is refused rather than emitted.
+
+    Use at signing and new-write boundaries ONLY. Verification and recompute keep calling
+    :func:`canonicalize_jcs`, which stays unrestricted so historical bytes keep verifying.
+
+    READS EACH KEY EXACTLY ONCE, so a Mapping subclass or an object overriding
+    ``__getitem__`` cannot answer safe on a validating pre-pass and unsafe on the
+    emitting pass. The value is captured once, checked, and emitted from that capture.
+    """
+    if obj is None:
+        return "null"
+    if isinstance(obj, bool):
+        return "true" if obj else "false"
+    if isinstance(obj, int):
+        if abs(obj) > MAX_SAFE_INTEGER:
+            raise UnsafeIntegerError(f"{path}: integer exceeds the interoperable IEEE 754 range")
+        return _es_number(float(obj))
+    if isinstance(obj, float):
+        if math.isnan(obj) or math.isinf(obj):
+            raise ValueError(f"Cannot canonicalize {obj}")
+        if obj.is_integer() and abs(obj) > MAX_SAFE_INTEGER:
+            raise UnsafeIntegerError(f"{path}: integer exceeds the interoperable IEEE 754 range")
+        return _es_number(obj)
+    if isinstance(obj, str):
+        _assert_no_lone_surrogate(obj)
+        return json.dumps(obj, ensure_ascii=False)
+    if isinstance(obj, list):
+        return "[" + ",".join(
+            canonicalize_jcs_for_write(item, f"{path}[{index}]") for index, item in enumerate(obj)
+        ) + "]"
+    if isinstance(obj, dict):
+        for key in obj.keys():
+            _assert_no_lone_surrogate(key)
+        pairs = []
+        for key in _canonical_keys(obj.keys()):
+            val = obj[key]  # the single read; everything below uses this capture
+            pairs.append(
+                json.dumps(key, ensure_ascii=False)
+                + ":"
+                + canonicalize_jcs_for_write(val, f"{path}.{key}")
+            )
+        return "{" + ",".join(pairs) + "}"
+    # Same reasoning as canonicalize_for_write above: emit through the shared fallback so
+    # the bytes match the read twin, but validate first so a tuple cannot smuggle an
+    # out-of-range integer past the rule.
+    assert_write_safe_numbers(obj, path)
+    return json.dumps(obj, ensure_ascii=False)
diff --git a/src/agent_passport/commerce.py b/src/agent_passport/commerce.py
index cae595b..731e094 100644
--- a/src/agent_passport/commerce.py
+++ b/src/agent_passport/commerce.py
@@ -14,7 +14,7 @@
 from typing import Any, Optional
 
 from .crypto import sign, verify
-from .canonical import canonicalize
+from .canonical import canonicalize, canonicalize_for_write
 from .passport import verify_passport
 
 
@@ -269,7 +269,7 @@ def sign_commerce_receipt(
         "delegationChain": delegation_chain,
         "beneficiary": beneficiary,
     }
-    payload = canonicalize(receipt)
+    payload = canonicalize_for_write(receipt)
     sig = sign(payload, private_key)
     return {**receipt, "signature": sig}
 
diff --git a/src/agent_passport/coordination.py b/src/agent_passport/coordination.py
index e6462ec..94b414a 100644
--- a/src/agent_passport/coordination.py
+++ b/src/agent_passport/coordination.py
@@ -14,7 +14,7 @@
 from typing import Any, Optional
 
 from .crypto import sign, verify
-from .canonical import canonicalize
+from .canonical import canonicalize, canonicalize_for_write
 
 
 def _rand_hex(n: int = 4) -> str:
@@ -60,7 +60,7 @@ def create_task_brief(
         "status": "draft",
     }
 
-    signature = sign(canonicalize(brief), operator_private_key)
+    signature = sign(canonicalize_for_write(brief), operator_private_key)
     return {**brief, "signature": signature}
 
 
@@ -127,7 +127,7 @@ def assign_task(
         "assignedAt": datetime.now(timezone.utc).isoformat(),
     }
 
-    op_sig = sign(canonicalize(assignment_content), operator_private_key)
+    op_sig = sign(canonicalize_for_write(assignment_content), operator_private_key)
     assignment = {**assignment_content, "operatorSignature": op_sig}
 
     updated_roles = [
@@ -138,7 +138,7 @@ def assign_task(
     brief_content = {k: v for k, v in brief.items() if k != "signature"}
     brief_content["roles"] = updated_roles
     brief_content["status"] = "assigned" if all_assigned else "draft"
-    new_sig = sign(canonicalize(brief_content), operator_private_key)
+    new_sig = sign(canonicalize_for_write(brief_content), operator_private_key)
 
     return {"assignment": assignment, "updatedBrief": {**brief_content, "signature": new_sig}}
 
@@ -147,7 +147,7 @@ def accept_task(assignment: dict, agent_private_key: str) -> dict:
     """Agent accepts a task assignment."""
     accepted_at = datetime.now(timezone.utc).isoformat()
     to_sign = {"assignmentId": assignment["assignmentId"], "taskId": assignment["taskId"], "acceptedAt": accepted_at}
-    agent_sig = sign(canonicalize(to_sign), agent_private_key)
+    agent_sig = sign(canonicalize_for_write(to_sign), agent_private_key)
     return {**assignment, "acceptedAt": accepted_at, "agentSignature": agent_sig}
 
 
@@ -188,7 +188,7 @@ def submit_evidence(
         },
     }
 
-    signature = sign(canonicalize(content), submitter_private_key)
+    signature = sign(canonicalize_for_write(content), submitter_private_key)
     return {**content, "signature": signature}
 
 
@@ -245,7 +245,7 @@ def review_evidence(
         "rationale": rationale,
         "issues": issues,
     }
-    signature = sign(canonicalize(content), reviewer_private_key)
+    signature = sign(canonicalize_for_write(content), reviewer_private_key)
     return {**content, "signature": signature}
 
 
@@ -298,7 +298,7 @@ def handoff_evidence(
         "toAgent": to_agent_public_key,
         "handoffAt": datetime.now(timezone.utc).isoformat(),
     }
-    op_sig = sign(canonicalize(content), operator_private_key)
+    op_sig = sign(canonicalize_for_write(content), operator_private_key)
     return {**content, "operatorSignature": op_sig}
 
 
@@ -343,7 +343,7 @@ def submit_deliverable(
         "citationCount": citation_count,
         "gapsFlagged": gaps_flagged,
     }
-    signature = sign(canonicalize(deliv), submitter_private_key)
+    signature = sign(canonicalize_for_write(deliv), submitter_private_key)
     return {**deliv, "signature": signature}
 
 
@@ -409,7 +409,7 @@ def complete_task(
         "metrics": metrics,
         "retrospective": retrospective,
     }
-    signature = sign(canonicalize(completion), operator_private_key)
+    signature = sign(canonicalize_for_write(completion), operator_private_key)
     return {**completion, "signature": signature}
 
 
diff --git a/src/agent_passport/data_settlement.py b/src/agent_passport/data_settlement.py
index 43847b1..7ccc971 100644
--- a/src/agent_passport/data_settlement.py
+++ b/src/agent_passport/data_settlement.py
@@ -17,7 +17,7 @@
 from typing import Any
 
 from .crypto import sign, verify
-from .canonical import canonicalize
+from .canonical import canonicalize, canonicalize_for_write
 
 
 def _merkle_root(items: list[str]) -> str:
@@ -74,7 +74,7 @@ def generate_settlement(
         "receiptCount": len(all_receipt_ids),
         "merkleRoot": _merkle_root(all_receipt_ids),
     }
-    record["signature"] = sign(canonicalize(record), generator_private_key)
+    record["signature"] = sign(canonicalize_for_write(record), generator_private_key)
     return record
 
 
diff --git a/src/agent_passport/data_source.py b/src/agent_passport/data_source.py
index 296f6f0..c0427ca 100644
--- a/src/agent_passport/data_source.py
+++ b/src/agent_passport/data_source.py
@@ -16,7 +16,7 @@
 from typing import Any
 
 from .crypto import sign, verify, generate_key_pair, public_key_from_private
-from .canonical import canonicalize
+from .canonical import canonicalize, canonicalize_for_write
 from .attribution import build_merkle_root, get_merkle_proof, verify_merkle_proof
 
 
@@ -57,7 +57,7 @@ def register_self_attested_source(
         "ownerPublicKey": owner_public_key,
         "registeredAt": _now_iso(),
     }
-    receipt["signature"] = sign(canonicalize(receipt), owner_private_key)
+    receipt["signature"] = sign(canonicalize_for_write(receipt), owner_private_key)
     return receipt
 
 
@@ -76,7 +76,7 @@ def register_custodian_attested_source(
         "custodianPublicKey": custodian_public_key,
         "registeredAt": _now_iso(),
     }
-    receipt["signature"] = sign(canonicalize(receipt), custodian_private_key)
+    receipt["signature"] = sign(canonicalize_for_write(receipt), custodian_private_key)
     return receipt
 
 
@@ -95,7 +95,7 @@ def register_gateway_observed_source(
         "gatewayPublicKey": gateway_public_key,
         "registeredAt": _now_iso(),
     }
-    receipt["signature"] = sign(canonicalize(receipt), gateway_private_key)
+    receipt["signature"] = sign(canonicalize_for_write(receipt), gateway_private_key)
     return receipt
 
 
@@ -166,7 +166,7 @@ def record_data_access(
         "accessedAt": _now_iso(),
         "termsAtAccessTime": terms_snapshot,
     }
-    receipt["signature"] = sign(canonicalize(receipt), gateway_private_key)
+    receipt["signature"] = sign(canonicalize_for_write(receipt), gateway_private_key)
     return receipt
 
 
diff --git a/src/agent_passport/delegation.py b/src/agent_passport/delegation.py
index fe68f3f..c847ae5 100644
--- a/src/agent_passport/delegation.py
+++ b/src/agent_passport/delegation.py
@@ -11,7 +11,7 @@
 from typing import Any, Optional
 
 from .crypto import sign, verify
-from .canonical import canonicalize, has_non_finite
+from .canonical import canonicalize, has_non_finite, canonicalize_for_write
 from ._time import parse_iso_utc
 
 
@@ -55,7 +55,7 @@ def create_delegation(
     }
 
     # Sign delegation (excluding signature field)
-    canonical = canonicalize(delegation)
+    canonical = canonicalize_for_write(delegation)
     delegation["signature"] = sign(canonical, private_key)
     return delegation
 
@@ -204,7 +204,7 @@ def sub_delegate(
         "createdAt": now.isoformat(),
     }
 
-    canonical = canonicalize(delegation)
+    canonical = canonicalize_for_write(delegation)
     delegation["signature"] = sign(canonical, private_key)
     return delegation
 
@@ -227,7 +227,7 @@ def revoke_delegation(
         "revokedAt": now.isoformat(),
         "reason": reason,
     }
-    canonical = canonicalize(revocation)
+    canonical = canonicalize_for_write(revocation)
     revocation["signature"] = sign(canonical, private_key)
 
     # Mark original delegation as revoked
@@ -306,7 +306,7 @@ def create_action_receipt(
         "delegationChain": delegation_chain or [],
     }
 
-    canonical = canonicalize(receipt)
+    canonical = canonicalize_for_write(receipt)
     receipt["signature"] = sign(canonical, private_key)
     return receipt
 
diff --git a/src/agent_passport/governance_block.py b/src/agent_passport/governance_block.py
index 060471f..bc1ae36 100644
--- a/src/agent_passport/governance_block.py
+++ b/src/agent_passport/governance_block.py
@@ -18,7 +18,7 @@
 from typing import Optional, Literal
 
 from .crypto import sign, verify
-from .canonical import canonicalize
+from .canonical import canonicalize, canonicalize_for_write
 
 UsagePermission = Literal[
     "permitted", "prohibited", "compensation_required", "attribution_required"
@@ -82,7 +82,7 @@ def generate_governance_block(
         "revocation_policy": dict(revocation_policy or DEFAULT_REVOCATION_POLICY),
     }
 
-    payload = canonicalize(block)
+    payload = canonicalize_for_write(block)
     signature = sign(payload, private_key)
     block["signature"] = signature
     return block
diff --git a/src/agent_passport/intent.py b/src/agent_passport/intent.py
index 02d741e..a854ea1 100644
--- a/src/agent_passport/intent.py
+++ b/src/agent_passport/intent.py
@@ -14,7 +14,7 @@
 from typing import Any, Optional
 
 from .crypto import sign, verify
-from .canonical import canonicalize
+from .canonical import canonicalize, canonicalize_for_write
 from .passport import verify_passport
 
 
@@ -59,7 +59,7 @@ def assign_role(
         "scope": scope,
     }
 
-    signature = sign(canonicalize(assignment), assigner_private_key)
+    signature = sign(canonicalize_for_write(assignment), assigner_private_key)
     return {**assignment, "signature": signature}
 
 
@@ -134,7 +134,7 @@ def create_intent_document(
         "expiresAt": expires_at,
     }
 
-    signature = sign(canonicalize(doc), author_private_key)
+    signature = sign(canonicalize_for_write(doc), author_private_key)
     return {**doc, "signature": signature}
 
 
@@ -215,7 +215,7 @@ def submit_consensus_round(
         "positionDelta": position_delta,
     }
 
-    signature = sign(canonicalize(round_content), private_key)
+    signature = sign(canonicalize_for_write(round_content), private_key)
     rnd = {**round_content, "signature": signature}
 
     updated = {
@@ -303,7 +303,7 @@ def resolve_deliberation(
         "resolvedAt": datetime.now(timezone.utc).isoformat(),
     }
 
-    outcome_sig = sign(canonicalize(outcome_content), resolver_private_key)
+    outcome_sig = sign(canonicalize_for_write(outcome_content), resolver_private_key)
     precedent_id = f"prec-{_rand_hex()}"
     outcome = {**outcome_content, "precedentId": precedent_id, "signature": outcome_sig}
 
diff --git a/src/agent_passport/passport.py b/src/agent_passport/passport.py
index 7a66c40..0bbd636 100644
--- a/src/agent_passport/passport.py
+++ b/src/agent_passport/passport.py
@@ -11,7 +11,7 @@
 from typing import Any, Optional
 
 from .crypto import generate_key_pair, sign, verify
-from .canonical import canonicalize, has_non_finite
+from .canonical import canonicalize, has_non_finite, canonicalize_for_write
 from ._time import parse_iso_utc
 
 DEFAULT_EXPIRY_DAYS = 365
@@ -98,7 +98,7 @@ def sign_passport(passport: dict, private_key: str) -> dict:
     Returns:
         SignedPassport dict with passport, signature, and signedAt.
     """
-    canonical = canonicalize(passport)
+    canonical = canonicalize_for_write(passport)
     signature = sign(canonical, private_key)
     return {
         "passport": passport,
diff --git a/src/agent_passport/policy.py b/src/agent_passport/policy.py
index db8ad68..fb14cd4 100644
--- a/src/agent_passport/policy.py
+++ b/src/agent_passport/policy.py
@@ -22,7 +22,7 @@
 from typing import Any, Optional
 
 from .crypto import sign, verify
-from .canonical import canonicalize
+from .canonical import canonicalize, canonicalize_for_write
 
 
 ENFORCEMENT_ESCALATION: dict[str, int] = {
@@ -55,7 +55,7 @@ def create_action_intent(
         "context": context,
         "createdAt": datetime.now(timezone.utc).isoformat(),
     }
-    signature = sign(canonicalize(intent), private_key)
+    signature = sign(canonicalize_for_write(intent), private_key)
     return {**intent, "signature": signature}
 
 
@@ -113,7 +113,7 @@ def evaluate_intent(
         "expiresAt": expires.isoformat(),
     }
 
-    signature = sign(canonicalize(decision), evaluator_private_key)
+    signature = sign(canonicalize_for_write(decision), evaluator_private_key)
     return {**decision, "signature": signature}
 
 
@@ -164,7 +164,7 @@ def create_policy_receipt(
         },
         "verifiedAt": datetime.now(timezone.utc).isoformat(),
     }
-    signature = sign(canonicalize(pr), verifier_private_key)
+    signature = sign(canonicalize_for_write(pr), verifier_private_key)
     return {**pr, "signature": signature}
 
 
diff --git a/src/agent_passport/principal.py b/src/agent_passport/principal.py
index 58f8936..e678100 100644
--- a/src/agent_passport/principal.py
+++ b/src/agent_passport/principal.py
@@ -10,7 +10,7 @@
 import hashlib
 from datetime import datetime, timedelta, timezone
 from .crypto import generate_key_pair, sign, verify
-from .canonical import canonicalize
+from .canonical import canonicalize, canonicalize_for_write
 
 
 def _utcnow():
@@ -67,7 +67,7 @@ def endorse_agent(
         "endorsedAt": now.isoformat() + "Z",
         "expiresAt": expiry.isoformat() + "Z",
     }
-    canonical = canonicalize(payload)
+    canonical = canonicalize_for_write(payload)
     signature = sign(canonical, principal_private_key)
     return {**payload, "revoked": False, "signature": signature}
 
@@ -152,7 +152,7 @@ def create_disclosure(principal, principal_private_key, level=None):
             "contactChannel": principal.get("contactChannel"),
         }
 
-    canonical = canonicalize(revealed)
+    canonical = canonicalize_for_write(revealed)
     proof = sign(canonical, principal_private_key)
 
     return {
diff --git a/src/agent_passport/training_attribution.py b/src/agent_passport/training_attribution.py
index 53298f9..5c66c87 100644
--- a/src/agent_passport/training_attribution.py
+++ b/src/agent_passport/training_attribution.py
@@ -16,7 +16,7 @@
 from typing import Any
 
 from .crypto import sign, verify
-from .canonical import canonicalize
+from .canonical import canonicalize, canonicalize_for_write
 
 TRAINING_USE_TYPES = [
     "fine_tune", "lora_adapter", "embedding", "rag_index",
@@ -54,9 +54,9 @@ def create_training_attribution(
         "recordedAt": datetime.now(timezone.utc).isoformat(),
     }
     receipt["contentHash"] = hashlib.sha256(
-        canonicalize(receipt).encode()
+        canonicalize_for_write(receipt).encode()
     ).hexdigest()
-    receipt["signature"] = sign(canonicalize(receipt), trainer_private_key)
+    receipt["signature"] = sign(canonicalize_for_write(receipt), trainer_private_key)
     return receipt
 
 
diff --git a/src/agent_passport/v2/accountability/bundle.py b/src/agent_passport/v2/accountability/bundle.py
index f5852f7..253c18b 100644
--- a/src/agent_passport/v2/accountability/bundle.py
+++ b/src/agent_passport/v2/accountability/bundle.py
@@ -21,7 +21,7 @@
 from hashlib import sha256
 from typing import List, Optional
 
-from ...canonical import canonicalize_jcs
+from ...canonical import canonicalize_jcs, canonicalize_jcs_for_write
 from ...crypto import public_key_from_private, sign, verify
 from .types import (
     APSBundle,
@@ -101,14 +101,14 @@ def create_aps_bundle(
 
     # Receipt_id over canonical form WITHOUT signature key.
     receipt_id = _sha256_hex(
-        canonicalize_jcs(draft.to_canonical_dict(drop_signature_field=True))
+        canonicalize_jcs_for_write(draft.to_canonical_dict(drop_signature_field=True))
     )
     draft.receipt_id = receipt_id
 
     # Signature payload over the SAME form (without signature key), with
     # the receipt_id now populated.
     signature = sign(
-        canonicalize_jcs(draft.to_canonical_dict(drop_signature_field=True)),
+        canonicalize_jcs_for_write(draft.to_canonical_dict(drop_signature_field=True)),
         bundler_private_key,
     )
     draft.signature = signature
diff --git a/src/agent_passport/v2/accountability/construct.py b/src/agent_passport/v2/accountability/construct.py
index cd9d9d6..61ccaef 100644
--- a/src/agent_passport/v2/accountability/construct.py
+++ b/src/agent_passport/v2/accountability/construct.py
@@ -23,7 +23,7 @@
 from hashlib import sha256
 from typing import Any, Dict, List, Optional
 
-from ...canonical import canonicalize_jcs
+from ...canonical import canonicalize_jcs, canonicalize_jcs_for_write
 from ...crypto import public_key_from_private, sign
 from .types import (
     ActionPayload,
@@ -99,9 +99,9 @@ def create_action_receipt(
         signature="",
     )
 
-    receipt_id = _sha256_hex(canonicalize_jcs(draft.to_canonical_dict()))
+    receipt_id = _sha256_hex(canonicalize_jcs_for_write(draft.to_canonical_dict()))
     draft.receipt_id = receipt_id
-    signature = sign(canonicalize_jcs(draft.to_canonical_dict()), signer_private_key)
+    signature = sign(canonicalize_jcs_for_write(draft.to_canonical_dict()), signer_private_key)
     draft.signature = signature
     return draft
 
@@ -138,9 +138,9 @@ def create_authority_boundary_receipt(
         signature="",
     )
 
-    receipt_id = _sha256_hex(canonicalize_jcs(draft.to_canonical_dict()))
+    receipt_id = _sha256_hex(canonicalize_jcs_for_write(draft.to_canonical_dict()))
     draft.receipt_id = receipt_id
-    signature = sign(canonicalize_jcs(draft.to_canonical_dict()), evaluator_private_key)
+    signature = sign(canonicalize_jcs_for_write(draft.to_canonical_dict()), evaluator_private_key)
     draft.signature = signature
     return draft
 
@@ -179,9 +179,9 @@ def create_custody_receipt(
         signature="",
     )
 
-    receipt_id = _sha256_hex(canonicalize_jcs(draft.to_canonical_dict()))
+    receipt_id = _sha256_hex(canonicalize_jcs_for_write(draft.to_canonical_dict()))
     draft.receipt_id = receipt_id
-    signature = sign(canonicalize_jcs(draft.to_canonical_dict()), custodian_private_key)
+    signature = sign(canonicalize_jcs_for_write(draft.to_canonical_dict()), custodian_private_key)
     draft.signature = signature
     return draft
 
@@ -234,11 +234,11 @@ def create_contestability_receipt(
     # Receipt_id and signature are computed over the filing form: no
     # controller_response, signature: "" present.
     receipt_id = _sha256_hex(
-        canonicalize_jcs(draft.to_canonical_dict(drop_controller_response=True))
+        canonicalize_jcs_for_write(draft.to_canonical_dict(drop_controller_response=True))
     )
     draft.receipt_id = receipt_id
     signature = sign(
-        canonicalize_jcs(draft.to_canonical_dict(drop_controller_response=True)),
+        canonicalize_jcs_for_write(draft.to_canonical_dict(drop_controller_response=True)),
         contestant_private_key,
     )
     draft.signature = signature
@@ -287,7 +287,7 @@ def attach_controller_response(
         signature=receipt.signature,
     )
     response_signature = sign(
-        canonicalize_jcs(
+        canonicalize_jcs_for_write(
             receipt_with_response.to_canonical_dict(drop_response_signature=True)
         ),
         controller_private_key,
diff --git a/src/agent_passport/v2/attribution_consent/create.py b/src/agent_passport/v2/attribution_consent/create.py
index 3fdb594..446879e 100644
--- a/src/agent_passport/v2/attribution_consent/create.py
+++ b/src/agent_passport/v2/attribution_consent/create.py
@@ -8,14 +8,13 @@
 from typing import Optional
 
 from ...crypto import sign
-from ...canonical import canonicalize
+from ...canonical import canonicalize, canonicalize_for_write
 from .types import AttributionReceipt, HybridTimestamp
 
 
-def receipt_core(receipt: dict) -> str:
-    """Canonical unsigned core string. Both citer and cited principal sign
-    exactly this payload, and the receipt id is sha256(core)."""
-    return canonicalize({
+def _receipt_core_impl(receipt: dict, _canon) -> str:
+    """Shared body so the read and write twins can never drift on the field list."""
+    return _canon({
         "version": receipt["version"],
         "citer": receipt["citer"],
         "citer_public_key": receipt["citer_public_key"],
@@ -28,6 +27,24 @@ def receipt_core(receipt: dict) -> str:
     })
 
 
+def receipt_core(receipt: dict) -> str:
+    """Canonical unsigned core string. Both citer and cited principal sign
+    exactly this payload, and the receipt id is sha256(core)."""
+    return _receipt_core_impl(receipt, canonicalize)
+
+
+def receipt_core_for_write(receipt: dict) -> str:
+    """Write-boundary twin of :func:`receipt_core`.
+
+    Emits the same bytes as :func:`receipt_core` for every value it accepts. The only
+    difference is that an integer-valued number outside the interoperable IEEE 754
+    range is refused instead of serialized. Use at signing and new-write boundaries
+    only: :func:`receipt_core` stays unrestricted so an artifact signed before this rule
+    existed keeps verifying.
+    """
+    return _receipt_core_impl(receipt, canonicalize_for_write)
+
+
 def create_attribution_receipt(
     *,
     citer: str,
@@ -60,7 +77,7 @@ def create_attribution_receipt(
         "expires_at": dict(expires_at),
     }
 
-    core = receipt_core(unsigned)
+    core = receipt_core_for_write(unsigned)
     receipt_id = hashlib.sha256(core.encode("utf-8")).hexdigest()
     citer_signature = sign(core, citer_private_key)
 
diff --git a/src/agent_passport/v2/attribution_consent/sign.py b/src/agent_passport/v2/attribution_consent/sign.py
index 61604d1..ba1b75e 100644
--- a/src/agent_passport/v2/attribution_consent/sign.py
+++ b/src/agent_passport/v2/attribution_consent/sign.py
@@ -2,7 +2,7 @@
 """sign_attribution_consent — cited principal adds the consent signature."""
 
 from ...crypto import sign, verify
-from .create import receipt_core
+from .create import receipt_core, receipt_core_for_write
 from .types import AttributionReceipt
 
 
@@ -12,7 +12,7 @@ def sign_attribution_consent(
 ) -> AttributionReceipt:
     """Add the cited principal's consent signature. Does not mutate the input.
     Raises ValueError if the private key does not match cited_principal_public_key."""
-    core = receipt_core(receipt)
+    core = receipt_core_for_write(receipt)
     cited_principal_signature = sign(core, cited_principal_private_key)
 
     if not verify(core, cited_principal_signature, receipt["cited_principal_public_key"]):
diff --git a/src/agent_passport/v2/attribution_primitive/canonical.py b/src/agent_passport/v2/attribution_primitive/canonical.py
index 32d132e..d72bb46 100644
--- a/src/agent_passport/v2/attribution_primitive/canonical.py
+++ b/src/agent_passport/v2/attribution_primitive/canonical.py
@@ -9,7 +9,12 @@
 from datetime import datetime, timezone
 from typing import Any, List, Union
 
-from ...canonical import canonicalize, canonicalize_jcs
+from ...canonical import (
+    canonicalize,
+    canonicalize_for_write,
+    canonicalize_jcs,
+    canonicalize_jcs_for_write,
+)
 from .types import (
     AttributionAxes,
     ComputeAxisEntry,
@@ -128,6 +133,18 @@ def hash_axis_leaf(axis: Any) -> bytes:
     return hashlib.sha256(canonicalize(axis).encode("utf-8")).digest()
 
 
+def hash_axis_leaf_for_write(axis: Any) -> bytes:
+    """Write-boundary twin of :func:`hash_axis_leaf`.
+
+    Emits the same bytes as :func:`hash_axis_leaf` for every value it accepts. The only
+    difference is that an integer-valued number outside the interoperable IEEE 754
+    range is refused instead of serialized. Use at signing and new-write boundaries
+    only: :func:`hash_axis_leaf` stays unrestricted so an artifact signed before this rule
+    existed keeps verifying.
+    """
+    return hashlib.sha256(canonicalize_for_write(axis).encode("utf-8")).digest()
+
+
 def hash_node(left: bytes, right: bytes) -> bytes:
     return hashlib.sha256(left + right).digest()
 
@@ -147,12 +164,41 @@ def canonical_hash_hex(obj: Any) -> str:
     return hashlib.sha256(canonicalize_jcs(obj).encode("utf-8")).hexdigest()
 
 
-def envelope_bytes(env) -> str:
-    """Canonical envelope string §2.3. Accepts TypedDict or plain dict."""
+def canonical_hash_hex_for_write(obj: Any) -> str:
+    """Write-boundary twin of :func:`canonical_hash_hex`.
+
+    Emits the same bytes as :func:`canonical_hash_hex` for every value it accepts. The only
+    difference is that an integer-valued number outside the interoperable IEEE 754
+    range is refused instead of serialized. Use at signing and new-write boundaries
+    only: :func:`canonical_hash_hex` stays unrestricted so an artifact signed before this rule
+    existed keeps verifying.
+    """
+    return hashlib.sha256(canonicalize_jcs_for_write(obj).encode("utf-8")).hexdigest()
+
+
+def _envelope_bytes_impl(env, _canon) -> str:
+    """Shared body so the read and write twins can never drift on the field list."""
     assert_canonical_timestamp(env["timestamp"])
-    return canonicalize({
+    return _canon({
         "action_ref": env["action_ref"],
         "merkle_root": env["merkle_root"],
         "issuer": env["issuer"],
         "timestamp": env["timestamp"],
     })
+
+
+def envelope_bytes(env) -> str:
+    """Canonical envelope string §2.3. Accepts TypedDict or plain dict."""
+    return _envelope_bytes_impl(env, canonicalize)
+
+
+def envelope_bytes_for_write(env) -> str:
+    """Write-boundary twin of :func:`envelope_bytes`.
+
+    Emits the same bytes as :func:`envelope_bytes` for every value it accepts. The only
+    difference is that an integer-valued number outside the interoperable IEEE 754
+    range is refused instead of serialized. Use at signing and new-write boundaries
+    only: :func:`envelope_bytes` stays unrestricted so an artifact signed before this rule
+    existed keeps verifying.
+    """
+    return _envelope_bytes_impl(env, canonicalize_for_write)
diff --git a/src/agent_passport/v2/attribution_primitive/construct.py b/src/agent_passport/v2/attribution_primitive/construct.py
index 9ca8cc1..424022e 100644
--- a/src/agent_passport/v2/attribution_primitive/construct.py
+++ b/src/agent_passport/v2/attribution_primitive/construct.py
@@ -7,15 +7,16 @@
 from .canonical import (
     assert_canonical_timestamp,
     canonical_hash_hex,
+    canonical_hash_hex_for_write,
     canonical_timestamp,
-    envelope_bytes,
+    envelope_bytes_for_write,
 )
-from .merkle import build_merkle_frame
+from .merkle import build_merkle_frame_for_write
 from .types import AttributionAction, AttributionAxes, AttributionPrimitive
 
 
-def compute_attribution_action_ref(action: AttributionAction) -> str:
-    """Derive action_ref from an action tuple. §1.2 / §3.4."""
+def _compute_attribution_action_ref_impl(action: AttributionAction, _hash) -> str:
+    """Shared body so the read and write twins can never drift on the field list."""
     if not action.get("agentId"):
         raise ValueError("attribution-primitive: action.agentId required")
     if not action.get("actionType"):
@@ -24,7 +25,7 @@ def compute_attribution_action_ref(action: AttributionAction) -> str:
         raise ValueError("attribution-primitive: action.nonce required")
     if not isinstance(action.get("params"), dict):
         raise ValueError("attribution-primitive: action.params must be an object")
-    return canonical_hash_hex({
+    return _hash({
         "agentId": action["agentId"],
         "actionType": action["actionType"],
         "params": action["params"],
@@ -32,6 +33,22 @@ def compute_attribution_action_ref(action: AttributionAction) -> str:
     })
 
 
+def compute_attribution_action_ref(action: AttributionAction) -> str:
+    """Derive action_ref from an action tuple. §1.2 / §3.4."""
+    return _compute_attribution_action_ref_impl(action, canonical_hash_hex)
+
+
+def _compute_attribution_action_ref_for_write(action: AttributionAction) -> str:
+    """Write-boundary twin of :func:`compute_attribution_action_ref`.
+
+    Module-internal on purpose: it is deliberately absent from the package barrel, so
+    this split adds no public API. The exported :func:`compute_attribution_action_ref`
+    stays unrestricted, because an external verifier re-deriving the action_ref of a
+    primitive signed before this rule must still get the same value.
+    """
+    return _compute_attribution_action_ref_impl(action, canonical_hash_hex_for_write)
+
+
 def construct_attribution_primitive(
     *,
     action: AttributionAction,
@@ -46,13 +63,13 @@ def construct_attribution_primitive(
     if not issuer_private_key:
         raise ValueError("attribution-primitive: issuer_private_key required")
 
-    action_ref = compute_attribution_action_ref(action)
-    frame = build_merkle_frame(axes)
+    action_ref = _compute_attribution_action_ref_for_write(action)
+    frame = build_merkle_frame_for_write(axes)
     merkle_root = frame["root"].hex()
     ts = timestamp if timestamp is not None else canonical_timestamp()
     assert_canonical_timestamp(ts)
 
-    envelope = envelope_bytes({
+    envelope = envelope_bytes_for_write({
         "action_ref": action_ref,
         "merkle_root": merkle_root,
         "issuer": issuer,
@@ -81,13 +98,13 @@ def resign_attribution_primitive(
     """Re-sign a primitive whose axes or metadata have changed."""
     new_axes = axes if axes is not None else primitive["axes"]
     action_ref = (
-        compute_attribution_action_ref(action) if action is not None else primitive["action_ref"]
+        _compute_attribution_action_ref_for_write(action) if action is not None else primitive["action_ref"]
     )
-    frame = build_merkle_frame(new_axes)
+    frame = build_merkle_frame_for_write(new_axes)
     merkle_root = frame["root"].hex()
     ts = timestamp if timestamp is not None else canonical_timestamp()
     assert_canonical_timestamp(ts)
-    envelope = envelope_bytes({
+    envelope = envelope_bytes_for_write({
         "action_ref": action_ref,
         "merkle_root": merkle_root,
         "issuer": primitive["issuer"],
diff --git a/src/agent_passport/v2/attribution_primitive/merkle.py b/src/agent_passport/v2/attribution_primitive/merkle.py
index 9d14460..4456942 100644
--- a/src/agent_passport/v2/attribution_primitive/merkle.py
+++ b/src/agent_passport/v2/attribution_primitive/merkle.py
@@ -3,17 +3,22 @@
 
 from typing import Dict, List, Tuple
 
-from .canonical import hash_axis_leaf, hash_node, normalize_axes
+from .canonical import (
+    hash_axis_leaf,
+    hash_axis_leaf_for_write,
+    hash_node,
+    normalize_axes,
+)
 from .types import AttributionAxes
 
 
-def build_merkle_frame(raw_axes: AttributionAxes) -> dict:
-    """Returns {axes, leaves, nodes, root} mirroring TS MerkleFrame."""
+def _build_merkle_frame_impl(raw_axes: AttributionAxes, _leaf) -> dict:
+    """Shared body so the read and write twins can never drift apart."""
     axes = normalize_axes(raw_axes)
-    leaf_d = hash_axis_leaf(axes["D"])
-    leaf_p = hash_axis_leaf(axes["P"])
-    leaf_g = hash_axis_leaf(axes["G"])
-    leaf_c = hash_axis_leaf(axes["C"])
+    leaf_d = _leaf(axes["D"])
+    leaf_p = _leaf(axes["P"])
+    leaf_g = _leaf(axes["G"])
+    leaf_c = _leaf(axes["C"])
     n_content = hash_node(leaf_d, leaf_p)
     n_auth_infra = hash_node(leaf_g, leaf_c)
     root = hash_node(n_content, n_auth_infra)
@@ -25,6 +30,23 @@ def build_merkle_frame(raw_axes: AttributionAxes) -> dict:
     }
 
 
+def build_merkle_frame(raw_axes: AttributionAxes) -> dict:
+    """Returns {axes, leaves, nodes, root} mirroring TS MerkleFrame."""
+    return _build_merkle_frame_impl(raw_axes, hash_axis_leaf)
+
+
+def build_merkle_frame_for_write(raw_axes: AttributionAxes) -> dict:
+    """Write-boundary twin of :func:`build_merkle_frame`.
+
+    Produces the same frame as :func:`build_merkle_frame` for every value it accepts.
+    The only difference is that an integer-valued number outside the interoperable
+    IEEE 754 range is refused instead of hashed into a leaf. Use when CONSTRUCTING a
+    primitive; projection and verification keep calling :func:`build_merkle_frame` so
+    a primitive built before this rule still reconstructs.
+    """
+    return _build_merkle_frame_impl(raw_axes, hash_axis_leaf_for_write)
+
+
 def projection_path(frame: dict, axis: str) -> Tuple[str, str]:
     leaves = frame["leaves"]
     nodes = frame["nodes"]
diff --git a/src/agent_passport/v2/attribution_primitive/residual.py b/src/agent_passport/v2/attribution_primitive/residual.py
index fbea1c4..e17d286 100644
--- a/src/agent_passport/v2/attribution_primitive/residual.py
+++ b/src/agent_passport/v2/attribution_primitive/residual.py
@@ -4,7 +4,7 @@
 import hashlib
 from typing import List
 
-from ...canonical import canonicalize
+from ...canonical import canonicalize_for_write
 from .canonical import to_weight_string
 from .types import (
     ComputeAxisEntry,
@@ -29,7 +29,7 @@ def _is_residual(x) -> bool:
 
 def _pooled_hash(entries) -> str:
     sorted_entries = sorted(entries, key=lambda e: e["did"])
-    return hashlib.sha256(canonicalize(sorted_entries).encode("utf-8")).hexdigest()
+    return hashlib.sha256(canonicalize_for_write(sorted_entries).encode("utf-8")).hexdigest()
 
 
 def aggregate_data_axis(entries: List[DataAxisItem], *, min_weight: float = DEFAULT_MIN_WEIGHT) -> dict:
diff --git a/src/agent_passport/v2/attribution_settlement/aggregate.py b/src/agent_passport/v2/attribution_settlement/aggregate.py
index cd916cf..81c14a5 100644
--- a/src/agent_passport/v2/attribution_settlement/aggregate.py
+++ b/src/agent_passport/v2/attribution_settlement/aggregate.py
@@ -11,9 +11,14 @@
 from datetime import datetime, timezone
 from typing import Dict, List, Optional
 
-from ...canonical import canonicalize
+from ...canonical import canonicalize, canonicalize_for_write
 from ..attribution_primitive.canonical import assert_canonical_timestamp
-from .merkle import build_merkle_root, empty_axis_merkle_root, leaf_hash
+from .merkle import (
+    build_merkle_root,
+    empty_axis_merkle_root,
+    leaf_hash,
+    leaf_hash_for_write,
+)
 from .types import (
     SettlementAxisIndex,
     SettlementContributor,
@@ -63,6 +68,18 @@ def residual_leaf_hash_hex(r) -> str:
     return hashlib.sha256(canonicalize(r).encode("utf-8")).hexdigest()
 
 
+def residual_leaf_hash_hex_for_write(r) -> str:
+    """Write-boundary twin of :func:`residual_leaf_hash_hex`.
+
+    Emits the same bytes as :func:`residual_leaf_hash_hex` for every value it accepts. The only
+    difference is that an integer-valued number outside the interoperable IEEE 754
+    range is refused instead of serialized. Use at signing and new-write boundaries
+    only: :func:`residual_leaf_hash_hex` stays unrestricted so an artifact signed before this rule
+    existed keeps verifying.
+    """
+    return hashlib.sha256(canonicalize_for_write(r).encode("utf-8")).hexdigest()
+
+
 class _AxisAccum:
     __slots__ = (
         "map",
@@ -167,7 +184,7 @@ def _finalize_axis(axis: str, accum: _AxisAccum, period: SettlementPeriod) -> Se
             "total_weight": total_str,
             "contribution_count": slot["count"],
         }
-        merkle_leaf_hash = hashlib.sha256(canonicalize(leaf_body).encode("utf-8")).hexdigest()
+        merkle_leaf_hash = hashlib.sha256(canonicalize_for_write(leaf_body).encode("utf-8")).hexdigest()
         contributors.append({
             "contributor_did": did,
             "total_weight": total_str,
@@ -181,7 +198,7 @@ def _finalize_axis(axis: str, accum: _AxisAccum, period: SettlementPeriod) -> Se
             raise ValueError("attribution-settlement: governance axis cannot carry a residual bucket")
         sorted_hashes = sorted(accum.per_receipt_residual_hashes)
         pooled_contributors_hash = hashlib.sha256(
-            canonicalize(sorted_hashes).encode("utf-8")
+            canonicalize_for_write(sorted_hashes).encode("utf-8")
         ).hexdigest()
         residual_bucket = {
             "residual_id": f"residual:{axis}",  # type: ignore[typeddict-item]
@@ -192,7 +209,7 @@ def _finalize_axis(axis: str, accum: _AxisAccum, period: SettlementPeriod) -> Se
 
     leaves = [bytes.fromhex(c["merkle_leaf_hash"]) for c in contributors]
     if residual_bucket:
-        leaves.append(bytes.fromhex(residual_leaf_hash_hex(residual_bucket)))
+        leaves.append(bytes.fromhex(residual_leaf_hash_hex_for_write(residual_bucket)))
     axis_merkle_root = (
         empty_axis_merkle_root() if not leaves else build_merkle_root(leaves).hex()
     )
@@ -290,7 +307,7 @@ def clone_period() -> SettlementPeriod:
     }
 
     sorted_refs = sorted(r["action_ref"] for r in in_period)
-    ref_leaves = [leaf_hash(ref) for ref in sorted_refs]
+    ref_leaves = [leaf_hash_for_write(ref) for ref in sorted_refs]
     input_receipts_hash = (
         empty_axis_merkle_root() if not ref_leaves else build_merkle_root(ref_leaves).hex()
     )
diff --git a/src/agent_passport/v2/attribution_settlement/merkle.py b/src/agent_passport/v2/attribution_settlement/merkle.py
index 0c33fb3..862999b 100644
--- a/src/agent_passport/v2/attribution_settlement/merkle.py
+++ b/src/agent_passport/v2/attribution_settlement/merkle.py
@@ -23,7 +23,7 @@
 import re
 from typing import List
 
-from ...canonical import canonicalize
+from ...canonical import canonicalize, canonicalize_for_write
 
 
 _HEX64 = re.compile(r"^[0-9a-f]{64}$", re.IGNORECASE)
@@ -64,6 +64,18 @@ def leaf_hash(obj) -> bytes:
     return hashlib.sha256(canonicalize(obj).encode("utf-8")).digest()
 
 
+def leaf_hash_for_write(obj) -> bytes:
+    """Write-boundary twin of :func:`leaf_hash`.
+
+    Emits the same bytes as :func:`leaf_hash` for every value it accepts. The only
+    difference is that an integer-valued number outside the interoperable IEEE 754
+    range is refused instead of serialized. Use at signing and new-write boundaries
+    only: :func:`leaf_hash` stays unrestricted so an artifact signed before this rule
+    existed keeps verifying.
+    """
+    return hashlib.sha256(canonicalize_for_write(obj).encode("utf-8")).digest()
+
+
 def build_merkle_root(leaves: List[bytes]) -> bytes:
     """Build a balanced binary Merkle tree over raw leaf hashes and return
     the root bytes. Leaves and internal nodes are domain-separated; an odd
diff --git a/src/agent_passport/v2/attribution_settlement/sign.py b/src/agent_passport/v2/attribution_settlement/sign.py
index 8dba9f4..d675569 100644
--- a/src/agent_passport/v2/attribution_settlement/sign.py
+++ b/src/agent_passport/v2/attribution_settlement/sign.py
@@ -8,16 +8,33 @@
 
 import hashlib
 
-from ...canonical import canonicalize
+from ...canonical import canonicalize, canonicalize_for_write
 from ...crypto import sign as ed25519_sign, verify as ed25519_verify
 
 
+def _settlement_signing_payload_impl(record: dict, _canon) -> str:
+    """Shared body so the read and write twins can never drift apart."""
+    body = dict(record)
+    body.pop("signature", None)
+    return _canon(body)
+
+
 def settlement_signing_payload(record: dict) -> str:
     """Canonical byte string signed (or verified). Strips the
     ``signature`` field if present."""
-    body = dict(record)
-    body.pop("signature", None)
-    return canonicalize(body)
+    return _settlement_signing_payload_impl(record, canonicalize)
+
+
+def settlement_signing_payload_for_write(record: dict) -> str:
+    """Write-boundary twin of :func:`settlement_signing_payload`.
+
+    Emits the same bytes as :func:`settlement_signing_payload` for every value it accepts. The only
+    difference is that an integer-valued number outside the interoperable IEEE 754
+    range is refused instead of serialized. Use at signing and new-write boundaries
+    only: :func:`settlement_signing_payload` stays unrestricted so an artifact signed before this rule
+    existed keeps verifying.
+    """
+    return _settlement_signing_payload_impl(record, canonicalize_for_write)
 
 
 def settlement_record_hash(record: dict) -> str:
@@ -27,7 +44,7 @@ def settlement_record_hash(record: dict) -> str:
 def sign_settlement_record(record: dict, gateway_private_key_hex: str) -> str:
     if not isinstance(gateway_private_key_hex, str) or not gateway_private_key_hex:
         raise ValueError("attribution-settlement: gateway_private_key_hex required")
-    return ed25519_sign(settlement_signing_payload(record), gateway_private_key_hex)
+    return ed25519_sign(settlement_signing_payload_for_write(record), gateway_private_key_hex)
 
 
 def verify_settlement_signature(record: dict, gateway_public_key_hex: str) -> bool:
diff --git a/src/agent_passport/v2/cognitive_attestation/envelope.py b/src/agent_passport/v2/cognitive_attestation/envelope.py
index 68dd9a8..b86be07 100644
--- a/src/agent_passport/v2/cognitive_attestation/envelope.py
+++ b/src/agent_passport/v2/cognitive_attestation/envelope.py
@@ -15,7 +15,7 @@
 from hashlib import sha256
 from typing import List, Optional
 
-from ...canonical import canonicalize_jcs
+from ...canonical import canonicalize_jcs, canonicalize_jcs_for_write
 from ...crypto import sign as ed_sign_hex
 from .types import (
     AggregationPolicy,
@@ -113,12 +113,29 @@ def canonicalize_attestation(att: CognitiveAttestation) -> bytes:
     payload produce byte-identical input regardless of signing order.
     Feature activations are sorted canonically. Returns UTF-8 bytes.
     """
+    return _canonicalize_attestation_impl(att, canonicalize_jcs)
+
+
+def _canonicalize_attestation_impl(att: CognitiveAttestation, _canon) -> bytes:
+    """Shared body so the read and write twins can never drift apart."""
     sorted_features = sort_feature_activations(att.feature_activations)
     view = replace(att, feature_activations=sorted_features, signatures=[])
-    canonical_str = canonicalize_jcs(view.to_canonical_dict())
+    canonical_str = _canon(view.to_canonical_dict())
     return canonical_str.encode("utf-8")
 
 
+def canonicalize_attestation_for_write(att: CognitiveAttestation) -> bytes:
+    """Write-boundary twin of :func:`canonicalize_attestation`.
+
+    Emits the same bytes as :func:`canonicalize_attestation` for every value it accepts. The only
+    difference is that an integer-valued number outside the interoperable IEEE 754
+    range is refused instead of serialized. Use at signing and new-write boundaries
+    only: :func:`canonicalize_attestation` stays unrestricted so an artifact signed before this rule
+    existed keeps verifying.
+    """
+    return _canonicalize_attestation_impl(att, canonicalize_jcs_for_write)
+
+
 def sign_attestation(
     att: CognitiveAttestation,
     private_key: bytes,
@@ -139,7 +156,7 @@ def sign_attestation(
     if not isinstance(signer_did, str) or len(signer_did) == 0:
         raise ValueError("sign_attestation: signer_did must be a non-empty string")
 
-    canonical_bytes = canonicalize_attestation(att)
+    canonical_bytes = canonicalize_attestation_for_write(att)
     canonical_str = canonical_bytes.decode("utf-8")
     private_key_hex = bytes(private_key).hex()
     sig_hex = ed_sign_hex(canonical_str, private_key_hex)
diff --git a/src/agent_passport/v2/human_escalation.py b/src/agent_passport/v2/human_escalation.py
index a586cc8..2ef7b32 100644
--- a/src/agent_passport/v2/human_escalation.py
+++ b/src/agent_passport/v2/human_escalation.py
@@ -19,7 +19,8 @@
 from typing import List, Optional, TypedDict
 
 from ..crypto import sign, verify
-from ..canonical import canonicalize
+from ..canonical import canonicalize, canonicalize_for_write
+from ..write_policy import assert_write_safe_numbers
 
 
 # ── Type aliases / TypedDicts ────────────────────────────────────────
@@ -92,12 +93,29 @@ class VerifyForActionResult(TypedDict, total=False):
 # ── Helpers ──────────────────────────────────────────────────────────
 
 
+def _hash_object_impl(obj: dict, _canon) -> str:
+    """Shared body so the read and write twins can never drift apart."""
+    return hashlib.sha256(_canon(obj).encode("utf-8")).hexdigest()
+
+
 def _hash_object(obj: dict) -> str:
     """sha256 hex of the canonical serialization — mirrors hashObject in
     src/v2/bridge.ts. TS signObject/verifyObject sign the HASH string, not
     the canonical JSON itself; Python must do the same for cross-language
     signature parity."""
-    return hashlib.sha256(canonicalize(obj).encode("utf-8")).hexdigest()
+    return _hash_object_impl(obj, canonicalize)
+
+
+def _hash_object_for_write(obj: dict) -> str:
+    """Write-boundary twin of :func:`_hash_object`.
+
+    Emits the same bytes as :func:`_hash_object` for every value it accepts. The only
+    difference is that an integer-valued number outside the interoperable IEEE 754
+    range is refused instead of serialized. Use at signing and new-write boundaries
+    only: :func:`_hash_object` stays unrestricted so an artifact signed before this rule
+    existed keeps verifying.
+    """
+    return _hash_object_impl(obj, canonicalize_for_write)
 
 
 def hash_action_details(details: dict) -> str:
@@ -109,6 +127,22 @@ def hash_action_details(details: dict) -> str:
     return hashlib.sha256(serialized.encode("utf-8")).hexdigest()
 
 
+def hash_action_details_for_write(details: dict) -> str:
+    """Write-boundary twin of :func:`hash_action_details`.
+
+    This commitment is minted by :func:`request_owner_confirmation` and recomputed by
+    :func:`is_confirmation_valid`, so the helper is shared and cannot be guarded in
+    place.
+
+    Note it does NOT canonicalize: it hashes ``json.dumps`` output, so no canonicalizer
+    census could see it. The guard therefore VALIDATES ONLY and still hashes the exact
+    same bytes, which keeps every existing commitment reproducible. Mirrors the
+    TypeScript ``hashActionDetailsForWrite``.
+    """
+    assert_write_safe_numbers(details)
+    return hash_action_details(details)
+
+
 def _find_requirement(delegation: dict, action_class: str) -> Optional[EscalationRequirement]:
     reqs = delegation.get("scope", {}).get("escalation_requirements")
     if not reqs:
@@ -169,7 +203,7 @@ def request_owner_confirmation(delegation: dict, action: EscalationAction) -> Co
         "id": str(uuid.uuid4()),
         "delegation_id": delegation["id"],
         "action_class": action["action_class"],
-        "action_details_hash": hash_action_details(action["action_details"]),
+        "action_details_hash": hash_action_details_for_write(action["action_details"]),
         "confirmation_scope": requirement["confirmation_scope"],
         "session_id": action.get("session_id"),
         "confirmation_ttl_ms": requirement["confirmation_ttl_ms"],
@@ -203,7 +237,7 @@ def record_owner_confirmation(
         "expires_at": _now_iso(expires_at_ms),
     }
     # Sign the sha256 of the canonical form (TS bridge.ts signObject parity).
-    signature = sign(_hash_object(data), owner_private_key)
+    signature = sign(_hash_object_for_write(data), owner_private_key)
     return {**data, "signature": signature}
 
 
diff --git a/src/agent_passport/v2/instruction_provenance/canonicalize.py b/src/agent_passport/v2/instruction_provenance/canonicalize.py
index 99725dd..dc3157d 100644
--- a/src/agent_passport/v2/instruction_provenance/canonicalize.py
+++ b/src/agent_passport/v2/instruction_provenance/canonicalize.py
@@ -12,7 +12,7 @@
 from hashlib import sha256
 from typing import List
 
-from ...canonical import canonicalize_jcs
+from ...canonical import canonicalize_jcs, canonicalize_jcs_for_write
 from .types import FilesystemMode, InstructionFile, InstructionProvenanceReceipt
 
 
@@ -126,11 +126,28 @@ def compute_context_root(files: List[InstructionFile]) -> str:
     sha256 of the JCS canonicalization of the instruction_files array
     (sorted). Must be byte-identical across languages.
     """
+    return _compute_context_root_impl(files, canonicalize_jcs)
+
+
+def _compute_context_root_impl(files: List[InstructionFile], _canon) -> str:
+    """Shared body so the read and write twins can never drift apart."""
     sorted_files = sort_instruction_files(files)
-    canon = canonicalize_jcs([f.to_canonical_dict() for f in sorted_files])
+    canon = _canon([f.to_canonical_dict() for f in sorted_files])
     return sha256_hex(canon)
 
 
+def compute_context_root_for_write(files: List[InstructionFile]) -> str:
+    """Write-boundary twin of :func:`compute_context_root`.
+
+    Emits the same bytes as :func:`compute_context_root` for every value it accepts. The only
+    difference is that an integer-valued number outside the interoperable IEEE 754
+    range is refused instead of serialized. Use at signing and new-write boundaries
+    only: :func:`compute_context_root` stays unrestricted so an artifact signed before this rule
+    existed keeps verifying.
+    """
+    return _compute_context_root_impl(files, canonicalize_jcs_for_write)
+
+
 def canonicalize_envelope(envelope: InstructionProvenanceReceipt) -> str:
     """Strip signature and receipt_id, JCS-canonicalize the rest.
 
@@ -140,3 +157,17 @@ def canonicalize_envelope(envelope: InstructionProvenanceReceipt) -> str:
     return canonicalize_jcs(
         envelope.to_canonical_dict(drop_signature=True, drop_receipt_id=True)
     )
+
+
+def canonicalize_envelope_for_write(envelope: InstructionProvenanceReceipt) -> str:
+    """Write-boundary twin of :func:`canonicalize_envelope`.
+
+    Emits the same bytes as :func:`canonicalize_envelope` for every value it accepts. The only
+    difference is that an integer-valued number outside the interoperable IEEE 754
+    range is refused instead of serialized. Use at signing and new-write boundaries
+    only: :func:`canonicalize_envelope` stays unrestricted so an artifact signed before this rule
+    existed keeps verifying.
+    """
+    return canonicalize_jcs_for_write(
+        envelope.to_canonical_dict(drop_signature=True, drop_receipt_id=True)
+    )
diff --git a/src/agent_passport/v2/instruction_provenance/envelope.py b/src/agent_passport/v2/instruction_provenance/envelope.py
index 3a56852..7b27a1e 100644
--- a/src/agent_passport/v2/instruction_provenance/envelope.py
+++ b/src/agent_passport/v2/instruction_provenance/envelope.py
@@ -12,9 +12,9 @@
 
 from ...crypto import sign as ed_sign_hex
 from .canonicalize import (
-    canonicalize_envelope,
+    canonicalize_envelope_for_write,
     canonicalize_path,
-    compute_context_root,
+    compute_context_root_for_write,
     sha256_hex,
     sort_instruction_files,
 )
@@ -139,7 +139,7 @@ def create_instruction_provenance_receipt(
             )
 
     sorted_files = sort_instruction_files(canonical_files)
-    context_root = compute_context_root(sorted_files)
+    context_root = compute_context_root_for_write(sorted_files)
     issued = issued_at if issued_at is not None else _now_iso()
     signing_key_id = f"ed25519:{public_key_hex[:16]}"
 
@@ -162,7 +162,7 @@ def create_instruction_provenance_receipt(
         expires_at=expires_at,
     )
 
-    canonical_bytes = canonicalize_envelope(unsigned)
+    canonical_bytes = canonicalize_envelope_for_write(unsigned)
     receipt_id = sha256_hex(canonical_bytes)
     signature_hex = sign_ed25519(canonical_bytes, private_key_hex)
 
diff --git a/src/agent_passport/v2/mutual_auth/certificate.py b/src/agent_passport/v2/mutual_auth/certificate.py
index 5d64e25..6ec429c 100644
--- a/src/agent_passport/v2/mutual_auth/certificate.py
+++ b/src/agent_passport/v2/mutual_auth/certificate.py
@@ -5,7 +5,7 @@
 import hashlib
 from typing import Optional, List
 
-from ...canonical import canonicalize_jcs
+from ...canonical import canonicalize_jcs, canonicalize_jcs_for_write
 from ...crypto import sign as ed_sign, verify as ed_verify
 from .types import MutualAuthCertificate, TrustAnchor
 
@@ -52,7 +52,7 @@ def build_certificate(
 
 def sign_certificate(unsigned: dict, issuer_sk_hex: str) -> MutualAuthCertificate:
     """Sign an unsigned certificate. Signature is Ed25519 over JCS canonical form."""
-    canonical = canonicalize_jcs(unsigned)
+    canonical = canonicalize_jcs_for_write(unsigned)
     sig_hex = ed_sign(canonical, issuer_sk_hex)
     sig_b64 = base64.b64encode(bytes.fromhex(sig_hex)).decode("ascii")
     signed = dict(unsigned)
diff --git a/src/agent_passport/v2/mutual_auth/handshake.py b/src/agent_passport/v2/mutual_auth/handshake.py
index 2135ce8..3314299 100644
--- a/src/agent_passport/v2/mutual_auth/handshake.py
+++ b/src/agent_passport/v2/mutual_auth/handshake.py
@@ -14,7 +14,7 @@
 import os
 from typing import List, Optional
 
-from ...canonical import canonicalize_jcs
+from ...canonical import canonicalize_jcs, canonicalize_jcs_for_write
 from ...crypto import sign as ed_sign, verify as ed_verify
 from .certificate import (
     certificate_id,
@@ -83,7 +83,7 @@ def build_attest(
         "certificate": certificate,
         "timestamp": now_ms,
     }
-    canonical = canonicalize_jcs(unsigned)
+    canonical = canonicalize_jcs_for_write(unsigned)
     sig_hex = ed_sign(canonical, own_sk_hex)
     sig_b64 = base64.b64encode(bytes.fromhex(sig_hex)).decode("ascii")
     signed = dict(unsigned)
@@ -205,7 +205,7 @@ def derive_session(
     agent_cert_id = certificate_id(agent_attest["certificate"])
     is_cert_id = certificate_id(is_attest["certificate"])
 
-    material = canonicalize_jcs({
+    material = canonicalize_jcs_for_write({
         "spec_version": SPEC_VERSION,
         "chosen_version": agent_attest["chosen_version"],
         "agent_cert_id": agent_cert_id,
diff --git a/src/agent_passport/v2/mutual_auth/trust_bundle.py b/src/agent_passport/v2/mutual_auth/trust_bundle.py
index ed4a94f..1fc91ed 100644
--- a/src/agent_passport/v2/mutual_auth/trust_bundle.py
+++ b/src/agent_passport/v2/mutual_auth/trust_bundle.py
@@ -4,7 +4,7 @@
 import base64
 from typing import List, Optional
 
-from ...canonical import canonicalize_jcs
+from ...canonical import canonicalize_jcs, canonicalize_jcs_for_write
 from ...crypto import sign as ed_sign, verify as ed_verify
 from .types import TrustAnchorBundle, TrustAnchor
 
@@ -33,7 +33,7 @@ def build_bundle(
 
 
 def sign_bundle(unsigned: dict, publisher_sk_hex: str) -> TrustAnchorBundle:
-    canonical = canonicalize_jcs(unsigned)
+    canonical = canonicalize_jcs_for_write(unsigned)
     sig_hex = ed_sign(canonical, publisher_sk_hex)
     sig_b64 = base64.b64encode(bytes.fromhex(sig_hex)).decode("ascii")
     signed = dict(unsigned)
diff --git a/src/agent_passport/v2/provisional_statement/create.py b/src/agent_passport/v2/provisional_statement/create.py
index d08da3f..3ca868d 100644
--- a/src/agent_passport/v2/provisional_statement/create.py
+++ b/src/agent_passport/v2/provisional_statement/create.py
@@ -9,7 +9,7 @@
 from typing import Optional
 
 from ...crypto import sign, verify
-from ...canonical import canonicalize
+from ...canonical import canonicalize, canonicalize_for_write
 from .types import ProvisionalStatement, Ed25519Signature
 from ..attribution_consent.types import HybridTimestamp
 
@@ -32,8 +32,8 @@ def _create_hybrid_timestamp(gateway_id: str, drift_ms: int = _DEFAULT_NTP_DRIFT
     }
 
 
-def statement_signing_payload(s: dict) -> str:
-    """Canonical payload an author signs."""
+def _statement_signing_payload_impl(s: dict, _canon) -> str:
+    """Shared body so the read and write twins can never drift on the field list."""
     base = {
         "id": s["id"],
         "version": s["version"],
@@ -44,7 +44,24 @@ def statement_signing_payload(s: dict) -> str:
     }
     if s.get("dead_man_expires_at"):
         base["dead_man_expires_at"] = s["dead_man_expires_at"]
-    return canonicalize(base)
+    return _canon(base)
+
+
+def statement_signing_payload(s: dict) -> str:
+    """Canonical payload an author signs."""
+    return _statement_signing_payload_impl(s, canonicalize)
+
+
+def statement_signing_payload_for_write(s: dict) -> str:
+    """Write-boundary twin of :func:`statement_signing_payload`.
+
+    Emits the same bytes as :func:`statement_signing_payload` for every value it accepts. The only
+    difference is that an integer-valued number outside the interoperable IEEE 754
+    range is refused instead of serialized. Use at signing and new-write boundaries
+    only: :func:`statement_signing_payload` stays unrestricted so an artifact signed before this rule
+    existed keeps verifying.
+    """
+    return _statement_signing_payload_impl(s, canonicalize_for_write)
 
 
 def create_provisional(
@@ -71,7 +88,7 @@ def create_provisional(
     if dead_man_expires_at is not None:
         base["dead_man_expires_at"] = dict(dead_man_expires_at)
 
-    author_signature = sign(statement_signing_payload(base), author_private_key)
+    author_signature = sign(statement_signing_payload_for_write(base), author_private_key)
 
     return {
         **base,
diff --git a/src/agent_passport/v2/read_fidelity_receipt/receipt.py b/src/agent_passport/v2/read_fidelity_receipt/receipt.py
index 6b5ef9e..33e9db0 100644
--- a/src/agent_passport/v2/read_fidelity_receipt/receipt.py
+++ b/src/agent_passport/v2/read_fidelity_receipt/receipt.py
@@ -17,7 +17,7 @@
 import re
 from typing import Dict, List, Optional, Sequence
 
-from ...canonical import canonicalize_jcs
+from ...canonical import canonicalize_jcs, canonicalize_jcs_for_write
 from ...crypto import public_key_from_private, sign, verify as ed_verify
 
 from .sampler import commit_spans, derive_seed, sample_spans, score_responses
@@ -53,8 +53,25 @@ def canonical_no_sig(record: dict) -> str:
     """Canonical signing preimage: RFC 8785 JCS of the record with the
     "sig" key removed entirely. Accepts a signed record (sig dropped)
     or an unsigned draft (no sig key present)."""
+    return _canonical_no_sig_impl(record, canonicalize_jcs)
+
+
+def _canonical_no_sig_impl(record: dict, _canon) -> str:
+    """Shared body so the read and write twins can never drift apart."""
     rest = {k: v for k, v in record.items() if k != "sig"}
-    return canonicalize_jcs(rest)
+    return _canon(rest)
+
+
+def canonical_no_sig_for_write(record: dict) -> str:
+    """Write-boundary twin of :func:`canonical_no_sig`.
+
+    Emits the same bytes as :func:`canonical_no_sig` for every value it accepts. The only
+    difference is that an integer-valued number outside the interoperable IEEE 754
+    range is refused instead of serialized. Use at signing and new-write boundaries
+    only: :func:`canonical_no_sig` stays unrestricted so an artifact signed before this rule
+    existed keeps verifying.
+    """
+    return _canonical_no_sig_impl(record, canonicalize_jcs_for_write)
 
 
 def _challenge_shape_reason(value: object) -> Optional[ReadFidelityVerifyReason]:
@@ -217,7 +234,7 @@ def create_read_fidelity_receipt(fields: dict, private_key_hex: str) -> ReadFide
             "content_digest, presentation_digest, nonce, and version"
         )
 
-    sig = sign(canonical_no_sig(draft), private_key_hex)
+    sig = sign(canonical_no_sig_for_write(draft), private_key_hex)
     return {**draft, "sig": sig}  # type: ignore[return-value]
 
 
diff --git a/src/agent_passport/values.py b/src/agent_passport/values.py
index 0988f10..ff1511a 100644
--- a/src/agent_passport/values.py
+++ b/src/agent_passport/values.py
@@ -13,7 +13,7 @@
 from typing import Any, Optional
 
 from .crypto import sign, verify
-from .canonical import canonicalize
+from .canonical import canonicalize, canonicalize_for_write
 
 # Enforcement escalation order (higher = stricter)
 ENFORCEMENT_ESCALATION: dict[str, int] = {
@@ -188,7 +188,7 @@ def attest_floor(
         "commitment": f"floor:{floor_version}|ext:{','.join(sorted(extensions)) or 'none'}|ts:{now.isoformat().replace('+00:00', 'Z')}",
     }
 
-    canonical = canonicalize(attestation)
+    canonical = canonicalize_for_write(attestation)
     signature = sign(canonical, private_key)
     return {**attestation, "signature": signature}
 
@@ -258,7 +258,7 @@ def evaluate_compliance(
         "generatedAt": now_iso,
     }
 
-    canonical = canonicalize(report)
+    canonical = canonicalize_for_write(report)
     signature = sign(canonical, verifier_private_key)
     return {**report, "signature": signature}
 
diff --git a/src/agent_passport/vc_wrapper.py b/src/agent_passport/vc_wrapper.py
index 18a1ce8..6aadf2d 100644
--- a/src/agent_passport/vc_wrapper.py
+++ b/src/agent_passport/vc_wrapper.py
@@ -9,7 +9,7 @@
 import base64
 from datetime import datetime, timezone
 
-from .canonical import canonicalize
+from .canonical import canonicalize, canonicalize_for_write
 from .crypto import sign, verify, public_key_from_private
 from .did_interop import to_did_key, from_did_key, _hex_to_multibase
 
@@ -235,7 +235,7 @@ def verify_verifiable_presentation(vp: dict) -> dict:
 # ── Proof helpers ──
 
 def _create_proof(data: dict, private_key: str, did: str, purpose: str, options: dict = None) -> dict:
-    canonical = canonicalize(data)
+    canonical = canonicalize_for_write(data)
     sig = sign(canonical, private_key)
     proof = {
         "type": "Ed25519Signature2020",
diff --git a/src/agent_passport/write_policy.py b/src/agent_passport/write_policy.py
new file mode 100644
index 0000000..36ae63a
--- /dev/null
+++ b/src/agent_passport/write_policy.py
@@ -0,0 +1,110 @@
+# Copyright 2026 Tymofii Pidlisnyi. Apache-2.0 license. See LICENSE.
+"""APS write policy: admissibility rules applied at signing and new-write boundaries.
+
+This is a layer ABOVE canonicalization, not a part of it. RFC 8785 canonicalization
+must remain able to canonicalize any valid binary64 number, so nothing here belongs
+inside ``canonicalize`` or ``canonicalize_jcs``.
+
+The rule implemented here refuses an integer-valued JSON number whose absolute value
+exceeds 9007199254740991, which is 2**53 minus 1. RFC 7493 (I-JSON) section 2.2 states
+that a sender cannot expect a receiver to treat integers outside that range exactly, and
+RECOMMENDS representing such values as JSON strings where exact interchange is required.
+It recommends; it does not mandate. APS adopts the recommendation as a write rule: an
+exact large quantity is carried as a decimal string, which is what
+draft-pidlisnyi-aps-03 already does with ``"per_action":"5000"``.
+
+Applied at write time only. A verifier reading an artifact signed before this rule
+existed MUST NOT be given this check, or it would refuse bytes it accepted before.
+
+Deliberately narrower than ``receipt_core.jcs.assert_i_json``. That validator also
+rejects any value whose type is outside a fixed JSON set, including ``datetime``, which
+legacy signing payloads carry. This one inspects numbers and recurses through
+containers, and leaves every other type alone, so adding it to an existing signing path
+cannot refuse a write that succeeds today for a reason unrelated to the number rule.
+"""
+
+from __future__ import annotations
+
+import math
+
+#: Largest integer magnitude that survives a binary64 round trip exactly.
+MAX_SAFE_INTEGER = 9_007_199_254_740_991
+
+
+class UnsafeIntegerError(ValueError):
+    """A new-write value carries an integer outside the interoperable IEEE 754 range.
+
+    Subclasses ValueError so callers that already fail closed around signing keep
+    working. Carries a stable machine-readable ``category`` and ``reason`` so a caller
+    can branch without parsing the message, and the message names the JSON path of the
+    offending member, matching the Go SDK's ``ErrInvalidIJSON`` wording.
+    """
+
+    #: Stable machine-readable category for this write-policy refusal.
+    category = "invalid_number"
+
+    def __init__(self, message: str, reason: str = "integer_exceeds_interoperable_range") -> None:
+        super().__init__(message)
+        #: Specific failure within the category.
+        self.reason = reason
+
+
+def assert_write_safe_numbers(value, path: str = "$", _ancestors: set[int] | None = None) -> None:
+    """Raise :class:`UnsafeIntegerError` if ``value`` carries an unsafe integer anywhere.
+
+    Recurses through lists and dicts so the rule applies to the whole artifact rather
+    than only its top-level members. Non-numeric values of any type are left untouched.
+
+    Args:
+        value: The in-memory value about to be canonicalized and signed.
+        path: JSON path of ``value``, used to locate the offending member.
+
+    Raises:
+        UnsafeIntegerError: An integer-valued number exceeds the interoperable range.
+    """
+    if _ancestors is None:
+        _ancestors = set()
+
+    # bool is a subclass of int in Python and is never a number for this purpose.
+    if value is None or isinstance(value, bool) or isinstance(value, str):
+        return
+
+    if isinstance(value, int):
+        if abs(value) > MAX_SAFE_INTEGER:
+            raise UnsafeIntegerError(
+                f"{path}: integer exceeds the interoperable IEEE 754 range"
+            )
+        return
+
+    if isinstance(value, float):
+        # Only integer-valued floats are bounded. A fractional value carries no claim
+        # to exactness beyond the double itself, which is the same rule the Go SDK
+        # applies with math.Trunc(x) == x.
+        if math.isfinite(value) and value.is_integer() and abs(value) > MAX_SAFE_INTEGER:
+            raise UnsafeIntegerError(
+                f"{path}: integer exceeds the interoperable IEEE 754 range"
+            )
+        return
+
+    if isinstance(value, (list, tuple)):
+        identity = id(value)
+        if identity in _ancestors:
+            return
+        _ancestors.add(identity)
+        for index, item in enumerate(value):
+            assert_write_safe_numbers(item, f"{path}[{index}]", _ancestors)
+        _ancestors.discard(identity)
+        return
+
+    if isinstance(value, dict):
+        identity = id(value)
+        if identity in _ancestors:
+            return
+        _ancestors.add(identity)
+        for key, item in value.items():
+            assert_write_safe_numbers(item, f"{path}.{key}", _ancestors)
+        _ancestors.discard(identity)
+        return
+
+    # Any other type is outside this rule's remit and is left to the canonicalizer.
+    return
diff --git a/tests/cross_impl/canonical-bytes-jcs-vectors.json b/tests/cross_impl/canonical-bytes-jcs-vectors.json
index 40814eb..162d754 100644
--- a/tests/cross_impl/canonical-bytes-jcs-vectors.json
+++ b/tests/cross_impl/canonical-bytes-jcs-vectors.json
@@ -95,6 +95,26 @@
       "canonical": "{\"arr\":[3,1,2],\"outer\":{\"a\":1,\"b\":2}}",
       "canonical_bytes_hex": "7b22617272223a5b332c312c325d2c226f75746572223a7b2261223a312c2262223a327d7d",
       "canonical_sha256": "be6fc370ca7efc25ae4b5ee7736507cbbac8fbc8278cb3371d1b8020a2bd0ca3"
+    },
+    {
+      "name": "integer-2pow60-inside-int64",
+      "description": "2^60 is inside signed 64-bit range but above 2^53, so its shortest round-trip double differs from its exact decimal. RFC 8785 serializes from the binary64 value, not from the caller's spelling.",
+      "input": {
+        "value": 1152921504606846976
+      },
+      "canonical": "{\"value\":1152921504606847000}",
+      "canonical_bytes_hex": "7b2276616c7565223a313135323932313530343630363834373030307d",
+      "canonical_sha256": "001814306319dfed540de7a22f61e88baf0288ebe916380796e8995d5db5eb00"
+    },
+    {
+      "name": "integer-2pow68-above-int64",
+      "description": "2^68 exceeds signed 64-bit range and is exactly representable as a binary64. Shortest round-trip is 2.9514790517935283e+20 and |x| < 1e21 forces decimal notation.",
+      "input": {
+        "value": 295147905179352825856
+      },
+      "canonical": "{\"value\":295147905179352830000}",
+      "canonical_bytes_hex": "7b2276616c7565223a3239353134373930353137393335323833303030307d",
+      "canonical_sha256": "57c82398dfd8be4a4ac903d1d2d01beee4c1d7179c9a5afc574bdf972646783b"
     }
   ]
 }
diff --git a/tests/fixtures/write-policy-admissibility-v1.json b/tests/fixtures/write-policy-admissibility-v1.json
new file mode 100644
index 0000000..ca47ec3
--- /dev/null
+++ b/tests/fixtures/write-policy-admissibility-v1.json
@@ -0,0 +1,42 @@
+{
+  "version": "aps-write-policy-admissibility-v1",
+  "rule": "At signing and new-write boundaries only, APS refuses an integer-valued JSON number whose absolute value exceeds 9007199254740991, which is 2**53 minus 1. RFC 7493 section 2.2 states a sender cannot expect a receiver to treat integers outside that range exactly, and RECOMMENDS a JSON string where exact interchange is required.",
+  "scope": "This corpus states ADMISSIBILITY only. It is deliberately separate from the RFC 8785 canonical-bytes corpus, because one vector must never mean two things.",
+  "cases": [
+    {
+      "name": "max_safe_integer",
+      "description": "2**53-1 is the largest integer that survives a binary64 round trip exactly.",
+      "value": 9007199254740991,
+      "verdict": "ACCEPT",
+      "path": null
+    },
+    {
+      "name": "max_safe_integer_plus_one",
+      "description": "2**53 is the first integer that no longer round trips exactly.",
+      "value": 9007199254740992,
+      "verdict": "REJECT",
+      "path": "$"
+    },
+    {
+      "name": "negative_max_safe_integer_minus_one",
+      "description": "The bound is on absolute value, so the negative twin is refused too.",
+      "value": -9007199254740992,
+      "verdict": "REJECT",
+      "path": "$"
+    },
+    {
+      "name": "nested_unsafe_integer",
+      "description": "The rule applies at any depth and must name the offending member.",
+      "value": { "a": { "b": [ { "c": 9007199254740992 } ] } },
+      "verdict": "REJECT",
+      "path": "$.a.b[0].c"
+    },
+    {
+      "name": "large_quantity_as_decimal_string",
+      "description": "The RFC 7493 remedy: an exact large quantity is carried as a decimal string, which is what draft-pidlisnyi-aps-03 already does with per_action.",
+      "value": "1152921504606846976",
+      "verdict": "ACCEPT",
+      "path": null
+    }
+  ]
+}
diff --git a/tests/test_canonical_bytes_vectors.py b/tests/test_canonical_bytes_vectors.py
index 270fe35..108ca97 100644
--- a/tests/test_canonical_bytes_vectors.py
+++ b/tests/test_canonical_bytes_vectors.py
@@ -21,7 +21,7 @@
 
 
 def test_canonical_bytes_jcs_parity():
-    assert len(VECTORS) == 8
+    assert len(VECTORS) == 10
     for v in VECTORS:
         canon = canonicalize_jcs(v["input"])
         assert canon == v["canonical"], f"{v['name']}: {canon!r} != {v['canonical']!r}"
diff --git a/tests/test_write_policy_admissibility.py b/tests/test_write_policy_admissibility.py
new file mode 100644
index 0000000..37e764e
--- /dev/null
+++ b/tests/test_write_policy_admissibility.py
@@ -0,0 +1,187 @@
+# Copyright 2026 Tymofii Pidlisnyi. Apache-2.0 license. See LICENSE.
+"""APS write-policy admissibility corpus, Python half.
+
+The corpus is a SEPARATE fixture from the RFC 8785 canonical-bytes vectors on purpose:
+one vector must never carry two meanings. These five cases state admissibility only,
+and the identical file ships in the TypeScript SDK so both languages are driven by the
+same bytes. The same five were run through the Go SDK's receiptcore validator and it
+agreed on all five, including the nested path.
+"""
+
+import hashlib
+import json
+import pathlib
+
+import pytest
+
+from agent_passport.canonical import (
+    canonicalize,
+    canonicalize_for_write,
+    canonicalize_jcs,
+    canonicalize_jcs_for_write,
+)
+from agent_passport.write_policy import UnsafeIntegerError
+
+FIXTURE = pathlib.Path(__file__).parent / "fixtures" / "write-policy-admissibility-v1.json"
+#: Recorded so a drift in the corpus is a visible test failure, not a silent reinterpretation.
+FIXTURE_SHA256 = "97db9ed8bfeab81ac50187c161ea80953f5878092530ff3fa1912d7eeb985f67"
+
+
+def test_fixture_bytes_are_pinned():
+    digest = hashlib.sha256(FIXTURE.read_bytes()).hexdigest()
+    assert digest == FIXTURE_SHA256, (
+        "the admissibility corpus changed; the TypeScript copy and the recorded Go run "
+        "must be updated together or the three languages stop meaning the same thing"
+    )
+
+
+def _cases():
+    data = json.loads(FIXTURE.read_text())
+    return [(c["name"], c["value"], c["verdict"], c["path"]) for c in data["cases"]]
+
+
+CASES = _cases()
+
+
+@pytest.mark.parametrize("name,value,verdict,path", CASES, ids=[c[0] for c in CASES])
+@pytest.mark.parametrize("writer", [canonicalize_for_write, canonicalize_jcs_for_write],
+                         ids=["legacy_write", "jcs_write"])
+def test_admissibility(writer, name, value, verdict, path):
+    if verdict == "ACCEPT":
+        writer(value)
+        return
+    with pytest.raises(UnsafeIntegerError) as exc:
+        writer(value)
+    assert str(exc.value).startswith(path + ":"), str(exc.value)
+    assert exc.value.category == "invalid_number"
+    assert exc.value.reason == "integer_exceeds_interoperable_range"
+
+
+@pytest.mark.parametrize("name,value,verdict,path", CASES, ids=[c[0] for c in CASES])
+@pytest.mark.parametrize("reader", [canonicalize, canonicalize_jcs], ids=["legacy_read", "jcs_read"])
+def test_the_read_canonicalizers_accept_every_case(reader, name, value, verdict, path):
+    """The rule is a WRITE rule.
+
+    Both unrestricted canonicalizers must keep serializing every case, including the
+    rejected ones, because that is what lets a verifier rebuild the preimage of an
+    artifact signed before the rule existed.
+    """
+    reader(value)
+
+
+@pytest.mark.parametrize("name,value,verdict,path", [c for c in CASES if c[2] == "ACCEPT"],
+                         ids=[c[0] for c in CASES if c[2] == "ACCEPT"])
+def test_accepted_cases_are_byte_identical_across_read_and_write(name, value, verdict, path):
+    assert canonicalize_for_write(value) == canonicalize(value)
+    assert canonicalize_jcs_for_write(value) == canonicalize_jcs(value)
+
+
+# ── Verification regression, permanent ──────────────────────────────────────
+#
+# The highest-risk failure mode in this whole change is guarding a verification path,
+# which would refuse artifacts that were signed before the rule existed and that verify
+# today. These tests reconstruct such an artifact the way the pre-rule code did, by
+# signing through the UNRESTRICTED canonicalizer, and then assert the shipped verifier
+# still accepts it.
+#
+# principal.verify_endorsement and principal.verify_disclosure are named explicitly
+# because an earlier classification pass wrongly listed both as signing paths. Guarding
+# them would have broken every endorsement and disclosure already published.
+
+from agent_passport.crypto import generate_key_pair, sign  # noqa: E402
+from agent_passport.principal import (  # noqa: E402
+    create_principal_identity,
+    verify_endorsement,
+    verify_disclosure,
+)
+
+UNSAFE = 9007199254740992
+
+
+def _pre_rule_endorsement():
+    """An endorsement minted the way the code did BEFORE the rule existed."""
+    created = create_principal_identity("Pre Rule Principal", "individual")
+    principal, keys = created["principal"], created["keyPair"]
+    agent = generate_key_pair()
+    payload = {
+        "endorsementId": "endorsement-prerule",
+        "principalId": principal["principalId"],
+        "principalPublicKey": principal["publicKey"],
+        "agentId": "did:aps:agent-prerule",
+        "agentPublicKey": agent["publicKey"],
+        # An out-of-range integer that the rule refuses on a NEW write today.
+        "scope": ["read", UNSAFE],
+        "relationship": "employee",
+        "endorsedAt": "2026-01-01T00:00:00Z",
+        "expiresAt": "2099-01-01T00:00:00Z",
+    }
+    signature = sign(canonicalize(payload), keys["privateKey"])
+    return {**payload, "revoked": False, "signature": signature}
+
+
+def test_pre_rule_endorsement_still_verifies():
+    """verify_endorsement must accept an endorsement signed before the rule existed."""
+    result = verify_endorsement(_pre_rule_endorsement())
+    assert result["valid"] is True, result
+
+
+def test_verify_endorsement_never_raises_on_the_number_rule():
+    """A tampered endorsement must fail on the SIGNATURE, never by refusing the number.
+
+    If verify_endorsement were ever switched to the write canonicalizer this would
+    raise UnsafeIntegerError instead of returning a verdict.
+    """
+    endorsement = _pre_rule_endorsement()
+    endorsement["scope"] = ["read", UNSAFE, {"nested": UNSAFE}]
+    result = verify_endorsement(endorsement)
+    assert result["valid"] is False
+    assert isinstance(result, dict)
+
+
+def test_pre_rule_disclosure_still_verifies():
+    """verify_disclosure must accept a disclosure signed before the rule existed."""
+    keys = generate_key_pair()
+    revealed = {
+        "did": f"did:aps:{keys['publicKey']}",
+        "employeeCount": UNSAFE,
+    }
+    disclosure = {
+        "disclosureId": "disclosure-prerule",
+        "principalId": "principal-prerule",
+        "level": "verified",
+        "revealedFields": revealed,
+        "disclosedAt": "2026-01-01T00:00:00Z",
+        "proof": sign(canonicalize(revealed), keys["privateKey"]),
+    }
+    result = verify_disclosure(disclosure)
+    assert result["valid"] is True, result
+
+
+# ── Tuple bypass, closed 2026-08-19 ─────────────────────────────────────────
+#
+# A tuple falls through every guarded branch of both write canonicalizers to the
+# terminal json.dumps fallback. Before the fix it was emitted unchecked, so an
+# out-of-range integer inside a tuple could be signed.
+#
+# The fix validates at that fallback but still EMITS through it, because the read twin
+# serializes a tuple the same way. Recursing instead would have emitted "[1,2]" where
+# the read twin emits "[1, 2]", and a signed artifact would then fail verification.
+
+def test_tuple_cannot_smuggle_an_unsafe_integer_past_the_write_rule():
+    for writer in (canonicalize_for_write, canonicalize_jcs_for_write):
+        with pytest.raises(UnsafeIntegerError) as exc:
+            writer({"v": (UNSAFE,)})
+        assert str(exc.value).startswith("$.v[0]:"), str(exc.value)
+
+
+def test_tuple_bytes_are_unchanged_between_the_read_and_write_twins():
+    """The bypass fix must not move a byte for a tuple the rule accepts."""
+    safe = {"v": (1, 2)}
+    assert canonicalize_for_write(safe) == canonicalize(safe)
+    assert canonicalize_jcs_for_write(safe) == canonicalize_jcs(safe)
+
+
+def test_the_read_twins_still_accept_a_tuple_carrying_an_unsafe_integer():
+    """Historical bytes stay reproducible: the rule is write-only, tuples included."""
+    canonicalize({"v": (UNSAFE,)})
+    canonicalize_jcs({"v": (UNSAFE,)})
diff --git a/tests/test_write_policy_c_splits.py b/tests/test_write_policy_c_splits.py
new file mode 100644
index 0000000..4744c7f
--- /dev/null
+++ b/tests/test_write_policy_c_splits.py
@@ -0,0 +1,266 @@
+# Copyright 2026 Tymofii Pidlisnyi. Apache-2.0 license. See LICENSE.
+"""Phase 2B: shared canonicalization helpers split into read and write twins.
+
+Each helper below was reachable from BOTH a signing path and a verification path. A
+guard placed on the shared helper would have refused to rebuild the preimage of an
+artifact signed before the APS unsafe-integer rule existed, so each one gained a
+``*_for_write`` twin used only by the constructing callers.
+
+Every twin is checked for four properties:
+  1. safe input: twin output is byte-identical to the original
+  2. a top-level unsafe integer is refused, at the exact path
+  3. a nested unsafe integer is refused, at the exact nested path
+  4. the mapping is read exactly ONCE on the write path, so the value validated is the
+     value emitted (a second read is what a getter would use to smuggle a value past)
+
+The originals are additionally checked to stay unrestricted, which is what keeps
+historical artifacts verifiable.
+"""
+
+import pytest
+
+from agent_passport.write_policy import UnsafeIntegerError
+
+from agent_passport.v2.human_escalation import _hash_object, _hash_object_for_write
+from agent_passport.v2.attribution_consent.create import receipt_core, receipt_core_for_write
+from agent_passport.v2.attribution_primitive.canonical import (
+    canonical_hash_hex,
+    canonical_hash_hex_for_write,
+    envelope_bytes,
+    envelope_bytes_for_write,
+    hash_axis_leaf,
+    hash_axis_leaf_for_write,
+)
+from agent_passport.v2.attribution_primitive.merkle import (
+    build_merkle_frame,
+    build_merkle_frame_for_write,
+)
+from agent_passport.v2.attribution_settlement.aggregate import (
+    residual_leaf_hash_hex,
+    residual_leaf_hash_hex_for_write,
+)
+from agent_passport.v2.attribution_settlement.merkle import leaf_hash, leaf_hash_for_write
+from agent_passport.v2.attribution_settlement.sign import (
+    settlement_signing_payload,
+    settlement_signing_payload_for_write,
+)
+from agent_passport.v2.instruction_provenance.canonicalize import (
+    canonicalize_envelope,
+    canonicalize_envelope_for_write,
+    compute_context_root,
+    compute_context_root_for_write,
+)
+from agent_passport.v2.provisional_statement.create import (
+    statement_signing_payload,
+    statement_signing_payload_for_write,
+)
+from agent_passport.v2.read_fidelity_receipt.receipt import (
+    canonical_no_sig,
+    canonical_no_sig_for_write,
+)
+
+SAFE = 9007199254740991
+UNSAFE = 9007199254740992
+
+
+def _receipt(amount):
+    return {
+        "version": "1.0",
+        "citer": "did:aps:citer",
+        "citer_public_key": "aa" * 32,
+        "cited_principal": "did:aps:cited",
+        "cited_principal_public_key": "bb" * 32,
+        "citation_content": {"quote": "text", "weight": amount},
+        "binding_context": "ctx",
+        "created_at": {"wall_clock": "2026-08-19T00:00:00.000Z", "logical": 1},
+        "expires_at": {"wall_clock": "2026-09-19T00:00:00.000Z", "logical": 2},
+    }
+
+
+def _statement(amount):
+    return {
+        "id": "stmt_1",
+        "version": 1,
+        "author": "did:aps:a",
+        "author_principal": "did:aps:p",
+        "content": {"body": "hello", "weight": amount},
+        "created_at": {"wall_clock": "2026-08-19T00:00:00.000Z", "logical": 1},
+    }
+
+
+def _envelope():
+    return {
+        "action_ref": "ab" * 32,
+        "merkle_root": "cd" * 32,
+        "issuer": "did:aps:issuer",
+        "timestamp": "2026-08-19T00:00:00.000Z",
+    }
+
+
+def _axes(amount):
+    return {
+        "D": [{
+            "source_did": "did:data:one",
+            "contribution_weight": "1.000000",
+            "access_receipt_hash": "a" * 64,
+            "sample_count": amount,
+        }],
+        "P": [],
+        "G": [],
+        "C": [],
+    }
+
+
+# (label, original, twin, safe_input, unsafe_input, expected_path)
+CASES = [
+    ("_hash_object", _hash_object, _hash_object_for_write,
+     {"a": SAFE}, {"a": UNSAFE}, "$.a"),
+    ("leaf_hash", leaf_hash, leaf_hash_for_write,
+     {"a": SAFE}, {"a": UNSAFE}, "$.a"),
+    ("residual_leaf_hash_hex", residual_leaf_hash_hex, residual_leaf_hash_hex_for_write,
+     {"a": SAFE}, {"a": UNSAFE}, "$.a"),
+    ("hash_axis_leaf", hash_axis_leaf, hash_axis_leaf_for_write,
+     {"a": SAFE}, {"a": UNSAFE}, "$.a"),
+    ("canonical_hash_hex", canonical_hash_hex, canonical_hash_hex_for_write,
+     {"a": SAFE}, {"a": UNSAFE}, "$.a"),
+    ("settlement_signing_payload", settlement_signing_payload, settlement_signing_payload_for_write,
+     {"a": SAFE, "signature": "x"}, {"a": UNSAFE, "signature": "x"}, "$.a"),
+    ("canonical_no_sig", canonical_no_sig, canonical_no_sig_for_write,
+     {"a": SAFE, "sig": "x"}, {"a": UNSAFE, "sig": "x"}, "$.a"),
+    ("receipt_core", receipt_core, receipt_core_for_write,
+     _receipt(SAFE), _receipt(UNSAFE), "$.citation_content.weight"),
+    ("statement_signing_payload", statement_signing_payload, statement_signing_payload_for_write,
+     _statement(SAFE), _statement(UNSAFE), "$.content.weight"),
+]
+
+NESTED_SAFE = {"a": {"b": [{"c": SAFE}]}}
+NESTED_UNSAFE = {"a": {"b": [{"c": UNSAFE}]}}
+
+
+@pytest.mark.parametrize("label,orig,twin,safe,unsafe,path", CASES, ids=[c[0] for c in CASES])
+def test_twin_matches_original_on_safe_input(label, orig, twin, safe, unsafe, path):
+    assert twin(safe) == orig(safe)
+
+
+@pytest.mark.parametrize("label,orig,twin,safe,unsafe,path", CASES, ids=[c[0] for c in CASES])
+def test_twin_refuses_unsafe_integer_at_exact_path(label, orig, twin, safe, unsafe, path):
+    with pytest.raises(UnsafeIntegerError) as exc:
+        twin(unsafe)
+    assert str(exc.value).startswith(path + ":"), str(exc.value)
+    assert exc.value.category == "invalid_number"
+    assert exc.value.reason == "integer_exceeds_interoperable_range"
+
+
+@pytest.mark.parametrize("label,orig,twin,safe,unsafe,path", CASES, ids=[c[0] for c in CASES])
+def test_original_stays_unrestricted(label, orig, twin, safe, unsafe, path):
+    """The read twin must keep accepting what it accepted before the rule existed."""
+    orig(unsafe)
+
+
+NESTED_CASES = [c for c in CASES if c[0] in (
+    "_hash_object", "leaf_hash", "residual_leaf_hash_hex", "hash_axis_leaf",
+    "canonical_hash_hex", "settlement_signing_payload", "canonical_no_sig")]
+
+
+@pytest.mark.parametrize("label,orig,twin,safe,unsafe,path", NESTED_CASES, ids=[c[0] for c in NESTED_CASES])
+def test_twin_refuses_nested_unsafe_at_exact_nested_path(label, orig, twin, safe, unsafe, path):
+    assert twin(NESTED_SAFE) == orig(NESTED_SAFE)
+    with pytest.raises(UnsafeIntegerError) as exc:
+        twin(NESTED_UNSAFE)
+    assert str(exc.value).startswith("$.a.b[0].c:"), str(exc.value)
+
+
+class CountingDict(dict):
+    """Answers safe on the first read of a key and unsafe on every read after it.
+
+    A check-then-canonicalize helper would validate the safe first answer and then
+    serialize the unsafe second one. A single-observation helper cannot.
+    """
+
+    def __init__(self, key, first, later):
+        super().__init__({key: first})
+        self._key = key
+        self._first = first
+        self._later = later
+        self.reads = 0
+
+    def __getitem__(self, key):
+        if key == self._key:
+            self.reads += 1
+            return self._first if self.reads == 1 else self._later
+        return super().__getitem__(key)
+
+
+# Two distinct ways to reach single observation, and both are correct:
+#   DIRECT      the caller's mapping is handed straight to the write canonicalizer,
+#               which captures each value once. Expected read count is exactly 1.
+#   MATERIALIZE the helper first copies the mapping (dict(record) or .items()), which
+#               freezes the observation before canonicalization and, on a dict
+#               subclass, bypasses __getitem__ entirely. Expected read count is 0.
+# What matters in both cases is that the mapping is observed AT MOST once, so the
+# value validated is necessarily the value emitted.
+DIRECT_CASES = [c for c in CASES if c[0] in (
+    "_hash_object", "leaf_hash", "residual_leaf_hash_hex", "hash_axis_leaf",
+    "canonical_hash_hex")]
+MATERIALIZE_CASES = [c for c in CASES if c[0] in (
+    "settlement_signing_payload", "canonical_no_sig")]
+ACCESSOR_CASES = DIRECT_CASES + MATERIALIZE_CASES
+
+
+@pytest.mark.parametrize("label,orig,twin,safe,unsafe,path", DIRECT_CASES, ids=[c[0] for c in DIRECT_CASES])
+def test_direct_twin_reads_each_key_exactly_once(label, orig, twin, safe, unsafe, path):
+    """Assert the READ COUNT, not only the emitted value.
+
+    A helper that validated and then re-serialized would report reads == 2 here while
+    still producing safe-looking output, which is exactly the bug this guards.
+    """
+    d = CountingDict("a", SAFE, UNSAFE)
+    twin(d)
+    assert d.reads == 1, "write path observed the key %d times" % d.reads
+
+
+@pytest.mark.parametrize("label,orig,twin,safe,unsafe,path", MATERIALIZE_CASES, ids=[c[0] for c in MATERIALIZE_CASES])
+def test_materializing_twin_observes_at_most_once(label, orig, twin, safe, unsafe, path):
+    d = CountingDict("a", SAFE, UNSAFE)
+    twin(d)
+    assert d.reads <= 1, "write path observed the key %d times" % d.reads
+
+
+@pytest.mark.parametrize("label,orig,twin,safe,unsafe,path", ACCESSOR_CASES, ids=[c[0] for c in ACCESSOR_CASES])
+def test_twin_refuses_when_the_stored_value_is_unsafe(label, orig, twin, safe, unsafe, path):
+    """An accessor cannot get a safe value signed by answering safe once and unsafe later."""
+    d = CountingDict("a", UNSAFE, SAFE)
+    with pytest.raises(UnsafeIntegerError):
+        twin(d)
+    assert d.reads <= 1
+
+
+# ── Cascading splits: helpers that reach a canonicalizer indirectly ──────────
+
+def test_build_merkle_frame_twin_matches_and_refuses():
+    assert build_merkle_frame_for_write(_axes(SAFE))["root"] == build_merkle_frame(_axes(SAFE))["root"]
+    with pytest.raises(UnsafeIntegerError):
+        build_merkle_frame_for_write(_axes(UNSAFE))
+    build_merkle_frame(_axes(UNSAFE))  # read twin stays unrestricted
+
+
+def test_instruction_provenance_twins_match_on_safe_input():
+    from agent_passport.v2.instruction_provenance.types import InstructionFile
+
+    files = [InstructionFile(path="a.md", digest="ab" * 32, bytes=10, role="system")]
+    assert compute_context_root_for_write(files) == compute_context_root(files)
+
+
+def test_envelope_bytes_twin_is_byte_identical_and_carries_no_numbers():
+    """envelope_bytes canonicalizes four string members only.
+
+    Recorded deliberately: the twin exists for symmetry with the other signing
+    preimages, but the unsafe-integer rule is UNREACHABLE through it, because
+    action_ref, merkle_root, issuer and timestamp are all strings. Do not read a
+    passing test here as evidence that the rule fires on this path.
+    """
+    env = _envelope()
+    assert envelope_bytes_for_write(env) == envelope_bytes(env)
+    # Every member of the canonicalized subset is a string, so no number is reachable.
+    canonicalized_members = ("action_ref", "merkle_root", "issuer", "timestamp")
+    assert all(isinstance(env[k], str) for k in canonicalized_members)