refactor(pipeline): route comparison identity through typed constructors

docs/plans/hash-boundary-registry.yaml

@@ -529,41 +529,41 @@ entries:
   call: hash_payload
   occurrence: 0
   classification: content-hash
-  note: 'feeds session_content_hash / revision projection (2026-07-09 hash-boundary census (docs/audits/2026-07-09-hash-boundary-census.md), Table 1 row 1).'
+  note: 'per-message content hash (full, unstripped payload) -- feeds session_content_hash / revision projection (2026-07-09 hash-boundary census (docs/audits/2026-07-09-hash-boundary-census.md), Table 1 row 1).'
 - path: polylogue/pipeline/ids.py
   function: '<module>.session_revision_projection'
   call: hash_payload
   occurrence: 1
   classification: content-hash
-  note: 'feeds session_content_hash / revision projection (2026-07-09 hash-boundary census (docs/audits/2026-07-09-hash-boundary-census.md), Table 1 row 1).'
+  note: 'event_hashes -- feeds session_content_hash / revision projection, unstripped and order-preserving (polylogue-nuec).'
 - path: polylogue/pipeline/ids.py
   function: '<module>.session_revision_projection'
   call: hash_payload
   occurrence: 2
   classification: content-hash
-  note: 'feeds session_content_hash / revision projection (2026-07-09 hash-boundary census (docs/audits/2026-07-09-hash-boundary-census.md), Table 1 row 1).'
+  note: 'per-event content hash (measurement excluded via _event_content_payload''s allowlist), feeding event_contents (polylogue-aggz, polylogue-nuec).'
 - path: polylogue/pipeline/ids.py
-  function: '<module>.session_revision_projection'
+  function: '<module>.message_identity_hash'
   call: hash_payload
-  occurrence: 3
+  occurrence: 0
   classification: content-hash
-  note: 'event_hashes -- feeds session_content_hash / revision projection, unstripped and order-preserving (polylogue-nuec).'
+  note: 'the typed constructor for message comparison identity (polylogue-aggz) -- fixed keyword-only signature (id only) is the sole path into message identity, replacing a dict-key-list projection (polylogue-aggz constructor chokepoint).'
 - path: polylogue/pipeline/ids.py
-  function: '<module>.session_revision_projection'
+  function: '<module>.attachment_identity_hash'
   call: hash_payload
-  occurrence: 4
+  occurrence: 0
   classification: content-hash
-  note: 'event base identity (event type + anchoring message), content-derived and position-independent, feeding event_contents (polylogue-aggz, polylogue-nuec).'
+  note: 'the typed constructor for attachment comparison identity (polylogue-aggz, polylogue-bu1i, polylogue-d8al, polylogue-hith) -- fixed keyword-only signature (message_id/name/mime_type only) is the sole path into attachment identity.'
 - path: polylogue/pipeline/ids.py
-  function: '<module>.session_revision_projection'
+  function: '<module>.event_base_identity_hash'
   call: hash_payload
-  occurrence: 5
+  occurrence: 0
   classification: content-hash
-  note: 'per-event content hash (measurement excluded via _event_content_payload''s allowlist), feeding event_contents (polylogue-aggz, polylogue-nuec).'
+  note: 'the typed constructor for an event''s position-independent base identity (polylogue-aggz, polylogue-nuec) -- fixed keyword-only signature (event_type/source_message_provider_id only) is the sole path into event base identity.'
 - path: polylogue/pipeline/ids.py
-  function: '<module>.session_revision_projection'
+  function: '<module>.event_canonical_identity_hash'
   call: hash_payload
-  occurrence: 6
+  occurrence: 0
   classification: content-hash
   note: 'canonical event identity fold (base identity + content) used only when the base identity is locally ambiguous within one revision -- still content-derived, never the array index (polylogue-aggz).'
 - path: polylogue/pipeline/ids.py

polylogue/archive/session_revision_membership.py

@@ -68,7 +68,7 @@ def _content_by_identity(contents: frozenset[tuple[bytes, bytes]]) -> dict[bytes
     Identity is not always injective: two acquired attachments on one
     message can share one identity (same ``message_id``/``name``/
     ``mime_type``, different bytes -- the accepted limit documented on
-    ``_ATTACHMENT_IDENTITY_FIELDS``, not a new one). Collapsing such a
+    ``attachment_identity_hash``, not a new one). Collapsing such a
     collision to a single arbitrary content hash (a plain ``dict``, last
     value wins) would let a real content conflict compare as equal instead.
     Grouping into a set per identity means a collision always degrades to

polylogue/pipeline/ids.py

@@ -209,21 +209,22 @@ def _message_hash_payload(message: ParsedMessage, message_id: str) -> dict[str,
     return payload
 
 
-#: The one field of a message hash payload that answers *which message is
-#: this*, as opposed to *what does it currently say*. A provider's own
-#: message id is stable across re-exports even when the export's array
-#: ordering is not (polylogue-c429).
-_MESSAGE_IDENTITY_FIELDS = ("id",)
-
-
-def _message_identity_payload(payload: dict[str, JSONValue]) -> dict[str, JSONValue]:
-    """Project the order-independent identity of one message payload.
-
-    Reads the already-normalized value out of ``_message_hash_payload`` rather
-    than re-deriving it, mirroring ``_attachment_identity_payload``'s single
-    normalization site.
+def message_identity_hash(*, id: str) -> bytes:
+    """The sole constructor of a message's comparison identity (polylogue-aggz).
+
+    A provider's own message id is stable across re-exports even when the
+    export's array ordering is not (polylogue-c429) -- it is the only
+    content field that answers *which message is this*, as opposed to *what
+    does it currently say*.
+
+    This is a fixed keyword-only signature, not a dict projected by a list
+    of field names: passing ``role``/``text``/``timestamp``/anything else is
+    a ``TypeError`` at the call boundary, not a value that has to be
+    remembered and stripped. Extending what a message's comparison identity
+    covers requires editing this signature -- an explicit, reviewable
+    decision, never a side effect of a parser gaining a new field.
     """
-    return {field: payload[field] for field in _MESSAGE_IDENTITY_FIELDS}
+    return bytes.fromhex(hash_payload({"id": id}))
 
 
 #: Fields of an attachment hash payload that answer *which attachment is
@@ -248,7 +249,19 @@ def _message_identity_payload(payload: dict[str, JSONValue]) -> dict[str, JSONVa
 #: genuinely distinct attachments that share one message/name/media-type and
 #: carry no bytes on either side of a comparison are indistinguishable by any
 #: signal this projection can offer.
-_ATTACHMENT_IDENTITY_FIELDS = ("message_id", "name", "mime_type")
+
+
+def attachment_identity_hash(*, message_id: JSONValue, name: JSONValue, mime_type: JSONValue) -> bytes:
+    """The sole constructor of an attachment's comparison identity (polylogue-aggz).
+
+    Fixed to (anchoring message, name, media type) -- content-derived and
+    never the provider's own attachment id (polylogue-d8al, polylogue-hith)
+    or acquisition state such as ``size_bytes``/inline bytes
+    (polylogue-bu1i). Those fields are not parameters here; passing them
+    (e.g. spreading a full attachment payload dict as ``**kwargs``) is a
+    ``TypeError``, not a value this function has to remember to strip.
+    """
+    return bytes.fromhex(hash_payload({"message_id": message_id, "name": name, "mime_type": mime_type}))
 
 
 def _attachment_hash_payload(attachment: ParsedAttachment) -> dict[str, JSONValue]:
@@ -265,16 +278,6 @@ def _attachment_hash_payload(attachment: ParsedAttachment) -> dict[str, JSONValu
     return payload
 
 
-def _attachment_identity_payload(payload: dict[str, JSONValue]) -> dict[str, JSONValue]:
-    """Project the acquisition-independent identity of one attachment payload.
-
-    Reads the already-normalized values out of ``_attachment_hash_payload``
-    rather than re-deriving them, so there is exactly one normalization site and
-    identity can never drift from the content hash it is paired with.
-    """
-    return {field: payload[field] for field in _ATTACHMENT_IDENTITY_FIELDS}
-
-
 #: `generation_lifecycle` payload keys that are provider-reported measurement,
 #: not identity, when the event's own payload declares them non-durable via
 #: ``duration_semantics == "provider_reported_elapsed"``. ChatGPT re-derives
@@ -372,12 +375,29 @@ def _event_content_payload(event: ParsedSessionEvent) -> dict[str, JSONValue]:
 #: depends on what else happens to be in the set. Still content-derived
 #: (each block's own content, including any content-intrinsic field such as
 #: ``block_index``, already differs), never the array position.
-_EVENT_BASE_IDENTITY_FIELDS = ("event_type", "source_message_provider_id")
+def event_base_identity_hash(*, event_type: JSONValue, source_message_provider_id: JSONValue) -> bytes:
+    """The sole constructor of an event's position-independent base identity.
+
+    Anchoring message plus event type only -- content-derived, never the
+    array index and never provider-reported measurement (polylogue-nuec),
+    which is not a parameter here. Fixed keyword-only signature: passing a
+    whole event content payload as ``**kwargs`` (which also carries
+    ``timestamp``/``payload``) is a ``TypeError``.
+    """
+    return bytes.fromhex(
+        hash_payload({"event_type": event_type, "source_message_provider_id": source_message_provider_id})
+    )
 
 
-def _event_base_identity_payload(payload: dict[str, JSONValue]) -> dict[str, JSONValue]:
-    """Project the position-independent base correlation key of one event payload."""
-    return {field: payload[field] for field in _EVENT_BASE_IDENTITY_FIELDS}
+def event_canonical_identity_hash(*, base_identity: bytes, content_hash: bytes) -> bytes:
+    """Fold an event's base identity with its own content hash.
+
+    Used only when a base identity (event type + anchoring message) may be
+    shared by more than one event within one revision (e.g. multiple
+    ``chatgpt_block_metadata`` events on the same message, one per block) --
+    still content-derived, never the array index (polylogue-aggz).
+    """
+    return bytes.fromhex(hash_payload({"base_identity": base_identity.hex(), "content": content_hash.hex()}))
 
 
 def _session_hash_payload(
@@ -510,14 +530,18 @@ def session_revision_projection(convo: ParsedSession) -> SessionRevisionProjecti
     message_contents: set[tuple[bytes, bytes]] = set()
     message_hashes: list[bytes] = []
     for payload in messages_payload:
-        identity = bytes.fromhex(hash_payload(_message_identity_payload(payload)))
+        message_native_id = payload["id"]
+        assert isinstance(message_native_id, str)  # built as str above, never anything else
+        identity = message_identity_hash(id=message_native_id)
         content = bytes.fromhex(hash_payload(payload))
         message_contents.add((identity, content))
         message_hashes.append(content)
     attachment_identities: set[bytes] = set()
     attachment_contents: set[tuple[bytes, bytes]] = set()
     for payload in attachments_payload:
-        identity = bytes.fromhex(hash_payload(_attachment_identity_payload(payload)))
+        identity = attachment_identity_hash(
+            message_id=payload["message_id"], name=payload["name"], mime_type=payload["mime_type"]
+        )
         inline_content_hash = payload.get("inline_content_hash")
         attachment_identities.add(identity)
         if isinstance(inline_content_hash, str):
@@ -528,7 +552,12 @@ def session_revision_projection(convo: ParsedSession) -> SessionRevisionProjecti
     for payload, event in zip(session_events_payload, convo.session_events, strict=True):
         event_hashes.append(bytes.fromhex(hash_payload(payload)))
         content_payload = _event_content_payload(event)
-        event_base_identities.append(bytes.fromhex(hash_payload(_event_base_identity_payload(content_payload))))
+        event_base_identities.append(
+            event_base_identity_hash(
+                event_type=content_payload["event_type"],
+                source_message_provider_id=content_payload["source_message_provider_id"],
+            )
+        )
         event_content_hashes.append(bytes.fromhex(hash_payload(content_payload)))
     event_contents: set[tuple[bytes, bytes]] = set()
     for base_identity, content_hash in zip(event_base_identities, event_content_hashes, strict=True):
@@ -549,9 +578,7 @@ def session_revision_projection(convo: ParsedSession) -> SessionRevisionProjecti
         # content-intrinsic block_index), and true duplicates (same base
         # identity, same content, whether or not any sibling exists)
         # correctly collapse to one set entry either way.
-        canonical_identity = bytes.fromhex(
-            hash_payload({"base_identity": base_identity.hex(), "content": content_hash.hex()})
-        )
+        canonical_identity = event_canonical_identity_hash(base_identity=base_identity, content_hash=content_hash)
         event_contents.add((canonical_identity, content_hash))
     return SessionRevisionProjection(
         session_hash=bytes.fromhex(session_hash_hex),

tests/unit/pipeline/test_pipeline_ids.py

@@ -12,6 +12,10 @@
     _message_hash_payload,
     _normalize_for_hash,
     _session_hash_payload,
+    attachment_identity_hash,
+    event_base_identity_hash,
+    event_canonical_identity_hash,
+    message_identity_hash,
     session_content_hash,
     session_id,
     session_revision_projection,
@@ -289,3 +293,125 @@ def test_session_revision_projection_matches_independent_recomputation() -> None
         if "inline_content_hash" in p
     )
     assert list(projection.event_hashes) == [bytes.fromhex(hash_payload(p)) for p in independent_event_payloads]
+
+
+# --- polylogue-aggz: typed identity constructor -----------------------------
+#
+# These tests prove the constructor property structurally, not just by
+# example: acquisition state, provider-reported measurement, and any field a
+# parser might add in the future cannot reach comparison identity, because
+# the identity constructors are fixed keyword-only functions -- passing
+# anything outside their declared parameters is a TypeError at the call
+# boundary, not a value that has to be remembered and stripped.
+
+
+def test_message_identity_hash_rejects_non_content_fields() -> None:
+    """The message identity constructor accepts only ``id`` -- nothing else.
+
+    Attempting to pass ``text``/``timestamp``/``role`` (what a message says,
+    not which message it is) is rejected by the function signature itself,
+    not by a runtime filter someone has to remember to apply.
+    """
+    with pytest.raises(TypeError):
+        message_identity_hash(id="m1", text="hello")  # type: ignore[call-arg]
+
+
+def test_attachment_identity_hash_rejects_acquisition_state() -> None:
+    """Acquisition state (polylogue-bu1i) cannot reach attachment identity.
+
+    ``size_bytes`` and ``inline_content_hash`` describe whether an
+    attachment's bytes have been acquired, not which attachment it is --
+    passing them is a TypeError, proving the exclusion is structural rather
+    than a convention encoded in a list of dict keys.
+    """
+    with pytest.raises(TypeError):
+        attachment_identity_hash(  # type: ignore[call-arg]
+            message_id="m1", name="f.txt", mime_type="text/plain", size_bytes=3
+        )
+    with pytest.raises(TypeError):
+        attachment_identity_hash(  # type: ignore[call-arg]
+            message_id="m1", name="f.txt", mime_type="text/plain", inline_content_hash="deadbeef"
+        )
+
+
+def test_attachment_identity_hash_rejects_full_payload_spread() -> None:
+    """A parser adding a new field to the payload dict cannot silently enter
+    identity: spreading the *entire* hash-stable attachment payload (as a
+    real future parser change might attempt, e.g. after adding a brand new
+    ``upload_origin`` or ``caption`` field to what gets hashed) into the
+    identity constructor is rejected outright, because the payload carries
+    keys (``id``, ``size_bytes``, and whatever new field a parser adds) that
+    are simply not parameters of ``attachment_identity_hash``.
+    """
+    attachment = ParsedAttachment(
+        provider_attachment_id="a1", message_provider_id="m1", name="f.txt", mime_type="text/plain", size_bytes=3
+    )
+    full_payload = _attachment_hash_payload(attachment)
+    # Simulate a parser adding a brand-new, never-seen-before field to the
+    # hash-stable payload -- the exact failure shape polylogue-bu1i/-nuec
+    # were: a NEW field silently entering identity because the extraction
+    # took "everything except a denylist" rather than an explicit allowlist.
+    full_payload["totally_new_provider_field"] = "unclassified-value"
+    with pytest.raises(TypeError):
+        attachment_identity_hash(**full_payload)
+
+
+def test_event_base_identity_hash_rejects_measurement_fields() -> None:
+    """Provider-reported measurement (polylogue-nuec) cannot reach event identity."""
+    with pytest.raises(TypeError):
+        event_base_identity_hash(  # type: ignore[call-arg]
+            event_type="generation_lifecycle",
+            source_message_provider_id="m1",
+            payload={"elapsed_duration_ms": 13000},
+        )
+
+
+def test_identity_constructors_ignore_new_payload_fields_when_called_correctly() -> None:
+    """Adding a new field to a parser fixture leaves identity unaffected.
+
+    Two attachments differing only in a field that is not one of the three
+    named identity parameters -- here, acquisition state plus a synthetic
+    "field a future parser might add" -- still produce identical identity,
+    because the constructor was never given that field to begin with.
+    """
+    acquired = ParsedAttachment(
+        provider_attachment_id="a1",
+        message_provider_id="m1",
+        name="f.txt",
+        mime_type="text/plain",
+        size_bytes=3,
+        inline_bytes=b"abc",
+    )
+    unacquired = ParsedAttachment(
+        provider_attachment_id="a1-different-provider-id",
+        message_provider_id="m1",
+        name="f.txt",
+        mime_type="text/plain",
+        size_bytes=None,
+    )
+    acquired_payload = _attachment_hash_payload(acquired)
+    unacquired_payload = _attachment_hash_payload(unacquired)
+    identity_acquired = attachment_identity_hash(
+        message_id=acquired_payload["message_id"],
+        name=acquired_payload["name"],
+        mime_type=acquired_payload["mime_type"],
+    )
+    identity_unacquired = attachment_identity_hash(
+        message_id=unacquired_payload["message_id"],
+        name=unacquired_payload["name"],
+        mime_type=unacquired_payload["mime_type"],
+    )
+    assert identity_acquired == identity_unacquired
+
+
+def test_event_canonical_identity_hash_folds_base_and_content() -> None:
+    """The canonical fold is a pure function of the two hashes it is given."""
+    base = event_base_identity_hash(event_type="chatgpt_block_metadata", source_message_provider_id="m1")
+    content_a = bytes.fromhex(hash_payload({"block_index": 0}))
+    content_b = bytes.fromhex(hash_payload({"block_index": 1}))
+    folded_a = event_canonical_identity_hash(base_identity=base, content_hash=content_a)
+    folded_b = event_canonical_identity_hash(base_identity=base, content_hash=content_b)
+    # Same base identity, different content -> different canonical identity
+    # (this is precisely what lets two distinct same-type-same-anchor events
+    # coexist as separate set entries -- polylogue-aggz).
+    assert folded_a != folded_b