fix(OMN-15742): bounded reconnect backoff + DEGRADED bus status for gateway forwarder

src/omnibase_infra/nodes/node_bus_forwarder_effect/contract.yaml

@@ -74,6 +74,10 @@ config:
       lag_threshold_messages: 500
       lag_threshold_seconds: 120
       drain_deadline_seconds: 30
+      reconnect_backoff_initial_seconds: 1
+      reconnect_backoff_max_seconds: 30
+      reconnect_backoff_jitter_seconds: 0.5
+      degraded_after_seconds: 60
     dedupe:
       key: "envelope_id"
       store: "sqlite"
@@ -121,7 +125,7 @@ metadata:
   description: "Hybrid gateway local runtime bus forwarder"
   author: "OmniNode Team"
   created: "2026-06-10"
-  updated: "2026-07-14"
+  updated: "2026-08-08"
   tags:
     - gateway
     - event-bus

src/omnibase_infra/nodes/node_bus_forwarder_effect/models/model_gateway_forwarder_config.py

@@ -42,6 +42,10 @@ class ModelGatewayForwarderConfig(BaseModel):
     dedupe_retention_hours: int = Field(default=24, ge=24)
     forward_retry_initial_seconds: float = Field(default=1.0, gt=0)
     forward_retry_max_seconds: float = Field(default=30.0, gt=0)
+    reconnect_backoff_initial_seconds: float = Field(default=1.0, gt=0)
+    reconnect_backoff_max_seconds: float = Field(default=30.0, gt=0)
+    reconnect_backoff_jitter_seconds: float = Field(default=0.5, ge=0)
+    degraded_after_seconds: int = Field(default=60, ge=1)
 
     @field_validator("dedupe_store_path")
     @classmethod
@@ -71,4 +75,9 @@ def _validate_liveness_windows(self) -> ModelGatewayForwarderConfig:
                 "canary.topic must be dedicated and must not appear in "
                 "mirror_topics.inbound or mirror_topics.outbound"
             )
+        if self.reconnect_backoff_max_seconds < self.reconnect_backoff_initial_seconds:
+            raise ValueError(
+                "reconnect_backoff_max_seconds must be greater than or equal to "
+                "reconnect_backoff_initial_seconds"
+            )
         return self

src/omnibase_infra/nodes/node_bus_forwarder_effect/models/model_gateway_heartbeat.py

@@ -7,21 +7,30 @@
 from datetime import datetime
 from typing import Literal
 
-from pydantic import BaseModel, ConfigDict
+from pydantic import BaseModel, ConfigDict, Field
 
 
 class ModelGatewayHeartbeat(BaseModel):
-    """Observable proof that one tenant-scoped edge is attached and active."""
+    """Observable proof of one tenant-scoped edge's liveness/reconnect state.
+
+    ``status="degraded"`` is emitted by the runtime reconnect-supervision
+    loop (``runtime/gateway_forwarder.py``) once a cloud-leg delivery
+    failure has persisted past the contract-declared
+    ``degraded_after_seconds`` window; ``consecutive_failures``/``detail``
+    are only populated on that path.
+    """
 
     model_config = ConfigDict(frozen=True, extra="forbid")
 
     # Heartbeat tenant_id is the config-bound DNS-safe slug.
     tenant_id: str
     # Canonical MSK principal is t-<tenant UUID hex>, not a UUID field.
     principal_id: str
-    status: Literal["active"] = "active"
+    status: Literal["active", "degraded"] = "active"
     emitted_at: datetime
     local_transport_flavor: Literal["containerized", "lightweight"]
+    consecutive_failures: int = Field(default=0, ge=0)
+    detail: str = ""
 
 
 __all__ = ["ModelGatewayHeartbeat"]

src/omnibase_infra/nodes/node_bus_forwarder_effect/services/service_gateway_forwarder.py

@@ -13,7 +13,7 @@
 import logging
 from collections.abc import Awaitable, Callable
 from datetime import UTC, datetime
-from typing import Protocol
+from typing import Literal, Protocol
 from uuid import uuid4
 
 from omnibase_core.models.core.model_envelope_metadata import ModelEnvelopeMetadata
@@ -34,6 +34,45 @@
 
 logger = logging.getLogger(__name__)
 
+# ``gateway_direction`` tag values that mark an envelope as local-bus-only --
+# the forwarder's own outbound consumer loop (NodeGatewayDelivery polling the
+# SAME transport local_bus publishes into, see runtime/gateway_forwarder.py)
+# must skip these rather than re-forward them to cloud (OMN-15570/OMN-15742
+# reconciliation finding D1). "cloud-to-local" is an inbound-transformed
+# envelope publish_status/publish_heartbeat never emit and needs the same
+# skip already established for the inbound leg; "local-mirror" marks a
+# direct local-only publish (DEGRADED status, and the G3 heartbeat local
+# mirror) that was never meant to leave this cluster at all.
+_LOCAL_ONLY_DIRECTIONS = frozenset({"cloud-to-local", "local-mirror"})
+
+
+def _stamp_local_only(
+    envelope: ModelEventEnvelope[dict[str, object]],
+) -> ModelEventEnvelope[dict[str, object]]:
+    """Tag a local-bus-only publish so the outbound consumer never re-forwards it.
+
+    Both the DEGRADED status publish (``ServiceGatewayForwarder.publish_status``)
+    and the G3 heartbeat local mirror (``publish_heartbeat``) go directly onto
+    the local bus's canonical outbound topic -- exactly the topic the
+    forwarder's own outbound consumer polls (``NodeGatewayDelivery`` on
+    ``local_consumer``, the SAME transport object ``local_bus`` publishes into
+    in the real runtime). Without this tag, ``_forward_outbound_message``'s
+    loopback skip does not match an untagged envelope, so it falls through to
+    ``_prepare_outbound`` -- a second cloud publish per heartbeat tick, or a
+    DEGRADED status leak to cloud (OMN-15570/OMN-15742 reconciliation finding
+    D1). Module-level (not a method) to stay under the class's method-count
+    pattern threshold.
+    """
+    metadata = envelope.metadata.model_copy(
+        update={
+            "tags": {
+                **envelope.metadata.tags,
+                "gateway_direction": "local-mirror",
+            }
+        }
+    )
+    return envelope.model_copy(update={"metadata": metadata})
+
 
 class ProtocolGatewayPublisher(Protocol):
     """Destination publish boundary shared by push and pull transports."""
@@ -77,10 +116,12 @@ def __init__(
     async def _forward_outbound_message(self, message: object) -> None:
         source_topic = self._message_topic(message)
         envelope = self._decode_message(message)
-        if envelope.metadata.tags.get("gateway_direction") == "cloud-to-local":
+        direction = envelope.metadata.tags.get("gateway_direction")
+        if direction in _LOCAL_ONLY_DIRECTIONS:
             logger.debug(
-                "Skipping gateway loopback on local topic %s",
+                "Skipping gateway loopback on local topic %s (direction=%s)",
                 source_topic,
+                direction,
             )
             return
         transformed, wire_topic = self._prepare_outbound(envelope, source_topic)
@@ -100,7 +141,8 @@ def validate_outbound_message(self, message: object) -> None:
         """Validate an outbound trust-boundary message without publishing it."""
         source_topic = self._message_topic(message)
         envelope = self._decode_message(message)
-        if envelope.metadata.tags.get("gateway_direction") != "cloud-to-local":
+        direction = envelope.metadata.tags.get("gateway_direction")
+        if direction not in _LOCAL_ONLY_DIRECTIONS:
             self._prepare_outbound(envelope, source_topic)
 
     async def _consume_inbound_message(self, message: object) -> None:
@@ -140,16 +182,25 @@ def decode_message(
         """Decode the canonical envelope used as the durable dedupe key source."""
         return cls._decode_message(message)
 
-    async def publish_heartbeat(self) -> None:
-        """Publish one tenant-scoped liveness event onto the cloud wire topic."""
+    def _build_status_envelope(
+        self,
+        status: Literal["active", "degraded"],
+        *,
+        consecutive_failures: int = 0,
+        detail: str = "",
+    ) -> tuple[ModelEventEnvelope[dict[str, object]], str]:
+        """Build the heartbeat/status envelope plus its canonical topic."""
         identity = self._config.tenant_identity
         now = datetime.now(UTC)
         envelope_id = uuid4()
         heartbeat = ModelGatewayHeartbeat(
             tenant_id=identity.tenant_slug,
             principal_id=identity.principal_id,
+            status=status,
             emitted_at=now,
             local_transport_flavor=self._config.local_transport_flavor,
+            consecutive_failures=consecutive_failures,
+            detail=detail,
         )
         envelope = ModelEventEnvelope[dict[str, object]](
             envelope_id=envelope_id,
@@ -170,6 +221,16 @@ async def publish_heartbeat(self) -> None:
             for topic in self._config.mirror_topics.outbound
             if topic.endswith(".gateway-heartbeat.v1")
         )
+        return envelope, canonical_topic
+
+    async def publish_heartbeat(self) -> None:
+        """Publish one tenant-scoped liveness event onto the cloud wire topic."""
+        identity = self._config.tenant_identity
+        envelope, canonical_topic = self._build_status_envelope("active")
+        # OMN-15740: _prepare_outbound returns (transformed_envelope, wire_topic) --
+        # the transform-seam handler delegation G0 wired in, not the pre-G0 single
+        # return this call site used to unpack. Keep G0's tuple return; G2 only
+        # adds the reconnect-supervision status publish below.
         transformed, wire_topic = self._prepare_outbound(envelope, canonical_topic)
         await self._publish_with_delivery_retry(
             bus=self._cloud_bus,
@@ -178,6 +239,35 @@ async def publish_heartbeat(self) -> None:
             value=self._encode_envelope(transformed),
         )
 
+    async def publish_status(
+        self,
+        status: Literal["active", "degraded"],
+        *,
+        consecutive_failures: int = 0,
+        detail: str = "",
+    ) -> None:
+        """Publish a reconnect-supervision status event onto the LOCAL bus.
+
+        Unlike ``publish_heartbeat`` (which crosses the cloud wire),
+        reconnect-supervision status -- most importantly a ``DEGRADED``
+        transition -- must stay observable while the cloud leg that caused
+        it is itself unreachable, so this publishes on the local bus using
+        the same heartbeat topic instead of the cloud leg.
+        """
+        identity = self._config.tenant_identity
+        envelope, canonical_topic = self._build_status_envelope(
+            status,
+            consecutive_failures=consecutive_failures,
+            detail=detail,
+        )
+        local_only = _stamp_local_only(envelope)
+        await self._publish_with_delivery_retry(
+            bus=self._local_bus,
+            topic=canonical_topic,
+            key=str(identity.tenant_id).encode("utf-8"),
+            value=self._encode_envelope(local_only),
+        )
+
     async def _publish_with_delivery_retry(
         self,
         *,