feat(OMN-18116): originate the causal edge and carry it on the wire

docker/runners/runner-image.lock.json

@@ -1,12 +1,12 @@
 {
   "base_image_digest": "sha256:3ba65aa20f86a0fad9df2b2c259c613df006b2e6d0bfcc8a146afb8c525a9751",
   "gh_version": "2.67.0",
-  "identity_digest": "1350647c095e75b220eaa18df9715abc",
+  "identity_digest": "6282d0dc7f380ceef69e30e91f7bd3b5",
   "image_version": 7,
   "kubectl_version": "1.32.1",
   "python_version": "3.12",
   "runner_version": "2.336.0",
-  "shared_env_digest": "f5af1a6e4eeb41cb1e22650f",
+  "shared_env_digest": "34a4296fdbc1413808263263",
   "shared_env_install_args": "--frozen --all-extras --all-groups --no-install-project",
   "uv_version": "0.6.14"
 }

pyproject.toml

@@ -33,7 +33,7 @@ dependencies = [
     # below pulls the merged core #1439 (terminal delegation classes) AND the
     # merged core #1440 (runtime_fanout_resolver, OMN-14403) pre-release --
     # #1440 is a strict descendant of #1439 on core dev.
-    "omnibase-core==0.47.7",
+    "omnibase-core==0.47.8",
     "omnibase-spi==0.23.3",
     # OMN-14190: canonical OCC stamp schema lives in omnibase_compat (OMN-14223).
     "omnibase-compat==0.5.7",
@@ -240,7 +240,7 @@ dev = [
 
 [tool.uv]
 override-dependencies = [
-    "omnibase-core==0.47.7",
+    "omnibase-core==0.47.8",
     "omnibase-spi==0.23.3",
     "omnibase-compat==0.5.7",
 ]

src/omnibase_infra/event_bus/adapters/adapter_protocol_event_publisher_kafka.py

@@ -60,6 +60,9 @@
 from omnibase_core.types import JsonType
 from omnibase_infra.enums import EnumInfraTransportType
 from omnibase_infra.errors import InfraUnavailableError
+from omnibase_infra.event_bus.envelope_header_identity import (
+    header_identity_fields_from_envelope,
+)
 from omnibase_infra.event_bus.models.model_event_headers import ModelEventHeaders
 from omnibase_infra.event_bus.testing.model_publisher_metrics import (
     ModelPublisherMetrics,
@@ -283,11 +286,21 @@ async def publish(
 
             # Build headers explicitly so EventBusKafka.publish() does not
             # fall back to minting its own, independent correlation_id.
+            # OMN-18116: this seam already refused to let the header
+            # correlation diverge from the envelope's (OMN-14962). The same
+            # reasoning applies to IDENTITY and to the causal edge, and until
+            # now it did not: `message_id` was left to its uuid4 default, which
+            # is the value `event_ledger` records as the hop's identity.
+            # `resolved_correlation_id` still wins over the envelope's, because
+            # the block above may have just minted and stamped it.
             headers = ModelEventHeaders(
                 source=self._service_name,
                 event_type=event_type,
-                correlation_id=resolved_correlation_id,
                 timestamp=start_time,
+                **{
+                    **header_identity_fields_from_envelope(envelope),
+                    "correlation_id": resolved_correlation_id,
+                },
             )
 
             # Publish to underlying bus

src/omnibase_infra/event_bus/envelope_header_identity.py

@@ -0,0 +1,60 @@
+# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc.
+# SPDX-License-Identifier: MIT
+"""Bind wire-header identity and the causal edge to the envelope (OMN-18116).
+
+``ModelEventHeaders`` defaults ``message_id`` and ``correlation_id`` to a fresh
+``uuid4``. Every envelope publish seam built its headers without overriding
+them, so the identity travelling on the wire was unrelated to the identity
+inside the envelope it described.
+
+That is not cosmetic. ``event_ledger`` -- the surface a chain replay reads back
+-- populates its per-hop identity column from the ``message_id`` HEADER, not
+from the envelope body (``HandlerLedgerProjection``). Measured read-only on the
+dev lane over a two-hour window: of 737 ledger rows whose decoded body carried
+an ``envelope_id``, the column matched the body on **0** and differed on
+**737**. ``correlation_id`` agreed on 4151 rows only because OMN-14962 forced
+that agreement, for correlation and nothing else.
+
+A causal edge is checkable only if the identity it references is the identity
+the reader reads back, so identity and edge are bound together, here, once. The
+alternative -- repeating the rule at each seam -- is the drift the two separate
+``ModelEventHeaders`` declarations in this codebase already demonstrate.
+
+Absence is preserved rather than defaulted. A payload that is not an envelope
+gets no keys back and keeps the model's own minting behaviour; an envelope with
+no parent gets no ``parent_message_id``, which is the checkable statement that
+the hop is a chain HEAD.
+"""
+
+from __future__ import annotations
+
+from uuid import UUID
+
+__all__ = ["header_identity_fields_from_envelope"]
+
+
+def header_identity_fields_from_envelope(envelope: object) -> dict[str, UUID]:
+    """Return the header fields an envelope determines, and only those.
+
+    Args:
+        envelope: The object about to be published. Anything without the
+            attributes below contributes nothing, so non-envelope payloads are
+            unaffected.
+
+    Returns:
+        A mapping suitable for splatting into ``ModelEventHeaders``. A key is
+        present only when the envelope actually carries that value, so an
+        absent parent never becomes an explicit ``None`` the model would have
+        to interpret, and never becomes an invented UUID.
+    """
+    fields: dict[str, UUID] = {}
+    envelope_id = getattr(envelope, "envelope_id", None)
+    if isinstance(envelope_id, UUID):
+        fields["message_id"] = envelope_id
+    correlation_id = getattr(envelope, "correlation_id", None)
+    if isinstance(correlation_id, UUID):
+        fields["correlation_id"] = correlation_id
+    parent_envelope_id = getattr(envelope, "parent_envelope_id", None)
+    if isinstance(parent_envelope_id, UUID):
+        fields["parent_message_id"] = parent_envelope_id
+    return fields

src/omnibase_infra/event_bus/event_bus_inmemory.py

@@ -67,6 +67,9 @@
     ModelInfraErrorContext,
     ProtocolConfigurationError,
 )
+from omnibase_infra.event_bus.envelope_header_identity import (
+    header_identity_fields_from_envelope,
+)
 from omnibase_infra.event_bus.models import (
     ModelEventHeaders,
     ModelEventMessage,
@@ -323,11 +326,17 @@ async def publish_envelope(
                 value=str(type(envelope)),
             ) from e
 
+        # OMN-18116: identity and the causal edge come from the envelope, not
+        # from the header model's uuid4 defaults. Applied on this bus too, so a
+        # chain assembled from an in-memory run reads the same as one assembled
+        # from the broker -- a rule that holds on only one transport is a rule
+        # a verifier cannot rely on.
         headers = ModelEventHeaders(
             source=f"{self._environment}.{self._group}",
             event_type=topic,
             content_type="application/json",
             timestamp=datetime.now(UTC),
+            **header_identity_fields_from_envelope(envelope),
         )
         await self.publish(topic, key, value, headers)
 

src/omnibase_infra/event_bus/event_bus_kafka.py

@@ -3145,6 +3145,13 @@ def _model_headers_to_kafka(
             kafka_headers.append(
                 ("parent_span_id", headers.parent_span_id.encode("utf-8"))
             )
+        # OMN-18116: the causal edge. Conditional like every other optional
+        # header, so a chain HEAD emits no key at all rather than an empty
+        # string -- an absent edge and a blank one must not read alike.
+        if headers.parent_message_id is not None:
+            kafka_headers.append(
+                ("parent_message_id", str(headers.parent_message_id).encode("utf-8"))
+            )
         if headers.operation_name:
             kafka_headers.append(
                 ("operation_name", headers.operation_name.encode("utf-8"))
@@ -3176,6 +3183,28 @@ def _parse_uuid_header(value: str | None) -> UUID:
                 pass
         return uuid4()
 
+    @staticmethod
+    def _parse_optional_uuid_header(value: str | None) -> UUID | None:
+        """Parse an optional UUID header, returning None rather than minting one.
+
+        OMN-18116: this is the causal-edge counterpart to
+        ``_parse_uuid_header``. That method mints a fresh UUID when the header
+        is absent or malformed, which is correct for an identity every message
+        must have. For an EDGE it would be a fabrication: a chain head would
+        acquire an invented parent, and a verifier's re-derivation would then
+        be comparing against a number nothing produced. Absent means absent.
+        """
+        if not value:
+            return None
+        try:
+            return UUID(value)
+        except (ValueError, AttributeError):
+            logger.warning(
+                "Malformed parent_message_id header %r, recording no causal edge",
+                value,
+            )
+            return None
+
     @staticmethod
     def _parse_int_header(value: str | None, default: int, field_name: str) -> int:
         """Parse an integer header value, logging a warning and returning default on failure."""
@@ -3216,6 +3245,16 @@ def _kafka_headers_to_model(
 
         correlation_id = self._parse_uuid_header(headers_dict.get("correlation_id"))
         message_id = self._parse_uuid_header(headers_dict.get("message_id"))
+        # OMN-18116: the causal edge, parsed WITHOUT the mint-on-absence
+        # fallback the two above use. `_parse_uuid_header` invents a UUID when
+        # a header is missing or malformed, which is right for an identity that
+        # must always exist and catastrophic for an edge: it would fabricate a
+        # parent for a chain head and turn a broken chain into a verifiable-
+        # looking one. Absent or unparseable means None, which is a chain head
+        # or a refusal, never an invention.
+        parent_message_id = self._parse_optional_uuid_header(
+            headers_dict.get("parent_message_id")
+        )
 
         # Parse timestamp from ISO format string to datetime (with fallback to now)
         timestamp_str = headers_dict.get("timestamp")
@@ -3301,6 +3340,7 @@ def _kafka_headers_to_model(
             trace_id=headers_dict.get("trace_id"),
             span_id=headers_dict.get("span_id"),
             parent_span_id=headers_dict.get("parent_span_id"),
+            parent_message_id=parent_message_id,
             operation_name=headers_dict.get("operation_name"),
             priority=priority,
             routing_key=headers_dict.get("routing_key"),

src/omnibase_infra/event_bus/mixin_kafka_broadcast.py

@@ -36,6 +36,9 @@ def __init__(self, config):
 from datetime import UTC, datetime
 from typing import Protocol, runtime_checkable
 
+from omnibase_infra.event_bus.envelope_header_identity import (
+    header_identity_fields_from_envelope,
+)
 from omnibase_infra.event_bus.models import ModelEventHeaders
 from omnibase_infra.event_bus.models.model_publish_receipt import (
     ModelPublishReceipt,
@@ -189,11 +192,15 @@ async def publish_envelope(
 
         value = json.dumps(envelope_dict).encode("utf-8")
 
+        # OMN-18116: the wire headers take their IDENTITY and their CAUSAL
+        # EDGE from the envelope instead of minting their own. The rule and the
+        # measurement behind it live in `envelope_header_identity`.
         headers = ModelEventHeaders(
             source=self._environment,
             event_type=topic,
             content_type="application/json",
             timestamp=datetime.now(UTC),
+            **header_identity_fields_from_envelope(envelope),
         )
 
         await self.publish(topic, key, value, headers)

src/omnibase_infra/event_bus/models/model_event_headers.py

@@ -69,6 +69,18 @@ class ModelEventHeaders(BaseModel):
     content_type: str = Field(default="application/json")
     correlation_id: UUID = Field(default_factory=uuid4)
     message_id: UUID = Field(default_factory=uuid4)
+    # OMN-18116: the CAUSAL EDGE on the wire -- the `message_id` of the
+    # message whose consumption caused this one to be published.
+    #
+    # This is the transport-side half of `ModelEventEnvelope.parent_envelope_id`.
+    # It exists because the readback surface a chain replay reads --
+    # `event_ledger` -- populates its per-hop identity column from THIS
+    # header (`handler_ledger_projection`), not from the envelope body. An
+    # edge recorded only in the body would be compared against a column
+    # holding an unrelated identity and would fail every hop.
+    #
+    # `None` is the checkable statement that this message is a chain HEAD.
+    parent_message_id: UUID | None = Field(default=None)
     # Timestamps - MUST be explicitly injected (no default_factory for testability)
     timestamp: datetime = Field(
         ..., description="Message creation timestamp (must be explicitly provided)"

src/omnibase_infra/runtime/service_dispatch_result_applier.py

@@ -65,6 +65,9 @@
     apply_failure_terminal_guard,
     declared_failure_terminal_topics,
 )
+from omnibase_infra.runtime.dispatch_envelope_context import (
+    current_dispatch_envelope,
+)
 from omnibase_infra.topics import topic_keys
 from omnibase_infra.topics.service_topic_registry import ServiceTopicRegistry
 from omnibase_infra.utils import derive_event_type_from_topic, sanitize_error_message
@@ -705,6 +708,27 @@ async def apply(
 
         # Phase 2: Publish output events AFTER projection and intents have committed.
         if result.output_events:
+            # OMN-18116: THE causal-edge origination site.
+            #
+            # Every event this runtime publishes as a consequence of consuming
+            # another one is built here, and the consumed envelope is already
+            # bound to this dispatch on a contextvar. Until now that parent
+            # identity was in scope at the publish site and discarded, so a
+            # chain read back later was a set of hops sharing a correlation id
+            # with no recorded statement of what caused what -- and a
+            # "replay green" verdict over it could only be a claim.
+            #
+            # No handler sets this. The canonical handler signature is
+            # `handle(request: ModelX) -> ModelY` and never sees an envelope;
+            # origination belongs in the runtime adapter, which is why there is
+            # exactly one site rather than one per node.
+            #
+            # None here is a STATEMENT: nothing was consumed, so every event of
+            # this dispatch is a chain head. A verifier can check that.
+            consumed_envelope = current_dispatch_envelope()
+            parent_envelope_id = (
+                consumed_envelope.envelope_id if consumed_envelope is not None else None
+            )
             for idx, output_event in enumerate(result.output_events):
                 try:
                     publish_payload = self._publish_payload_for_output_event(
@@ -719,11 +743,30 @@ async def apply(
                         effective_correlation_id,
                         f"{type(output_event).__name__}:{idx}",
                     )
+                    # A self-edge is refused by the envelope model, and
+                    # refusing it here too keeps a pathological equality from
+                    # taking the publish down. Equality means the deterministic
+                    # uuid5 collided with the consumed envelope's own id, which
+                    # is a real anomaly worth naming rather than a case to
+                    # silently paper over -- so it is logged and the hop is
+                    # recorded as a head, which reads as a broken chain to a
+                    # verifier instead of as a valid self-caused one.
+                    edge = parent_envelope_id
+                    if edge is not None and edge == deterministic_id:
+                        logger.warning(
+                            "Refusing a self-referential causal edge: the consumed "
+                            "envelope id equals this output's deterministic id "
+                            "(correlation_id=%s, envelope_id=%s)",
+                            str(effective_correlation_id),
+                            str(deterministic_id),
+                        )
+                        edge = None
                     output_envelope: ModelEventEnvelope[BaseModel] = ModelEventEnvelope(
                         envelope_id=deterministic_id,
                         payload=publish_payload,
                         correlation_id=effective_correlation_id,
                         envelope_timestamp=self._clock(),
+                        parent_envelope_id=edge,
                     )
 
                     # Extract partition key for per-entity ordering.

src/omnibase_infra/runtime/version_compatibility.py

@@ -175,7 +175,7 @@ def _build_matrix_from_pyproject(
 _FALLBACK_MATRIX: list[VersionConstraint] = [
     VersionConstraint(
         package="omnibase_core",
-        min_version="0.47.7",
+        min_version="0.47.8",
         max_version="0.48.0",
     ),
     VersionConstraint(

tests/fixtures/seams/core_release/0.46.11_expected_symbols.json

@@ -8,7 +8,7 @@
     "Update procedure: bump core, re-run the seam test, and record the new",
     "surface here only after confirming every consumer still type-checks."
   ],
-  "core_version": "0.47.7",
+  "core_version": "0.47.8",
   "source": "pypi",
   "modules": {
     "omnibase_core.models.delegation.wire": [