fix(gateway): a restored lane whose receiving bot is offline delivers nowhere

Composing PR-4 (intake/delivery split) with PR-5 (persisted transport_profile): the delivery
fallback to the runtime profile's unique adapter is only for sources with NO identity. A pinned
identity names the receiving bot; if that bot has no adapter it is offline and the lane fails
closed, never answering from the runtime profile's bot. Also: tests and docs reference the split
helper, not the removed _adapter_for_source.

gateway/authz_mixin.py

@@ -300,6 +300,17 @@ class GatewayAuthorizationMixin:
         adapter = self._intake_adapter_for(source)
         if adapter is not None:
             return adapter
+        # A pinned identity NAMES the receiving bot (live or restored from ``transport_profile``).
+        # If that bot has no adapter right now it is offline: fail closed rather than fall through to
+        # the runtime profile's bot — that fallthrough is the "restored lane answers from the wrong
+        # bot" row the identity exists to close. Only an identity-less source (legacy row, bare
+        # fixture) uses the unique-owner heuristic below.
+        from gateway.session_identity import identity_of
+        identity = identity_of(source)
+        if identity is not None and identity.multiplexed and not identity.transport_inferred:
+            return None
+        # No identity, or one whose transport was only inferred (hand-built source, pre-column row):
+        # the unique owner of ``(platform, runtime_profile)`` delivers.
         # ``getattr``: test fixtures build bare SimpleNamespace sources without ``profile``.
         return self._authorization_adapter(getattr(source, "platform", None), getattr(source, "profile", None))
 

gateway/session_identity.py

@@ -57,6 +57,11 @@ class RoutingIdentity:
     # Receiving adapter; None for restored/synthetic sources (no live provenance → fail closed).
     # Provenance, not identity: two events from the same bot share one identity.
     transport: Optional[weakref.ref] = field(default=None, compare=False, hash=False)
+    # True when nothing named the receiving bot (no live adapter, no persisted transport_profile,
+    # no explicit hint) and ``transport_profile`` is the primary by default. A hand-built or
+    # pre-column source. Delivery may still fall back to the runtime profile's unique adapter for
+    # these; an identity whose transport is KNOWN (live or restored) never does.
+    transport_inferred: bool = field(default=False, compare=False, hash=False)
 
     @property
     def namespace(self) -> str:
@@ -221,6 +226,7 @@ def resolve_identity(
             source._transport_adapter_ref = weakref.ref(adapter)
         _registered, owner_profile = runner._owning_profile(adapter, platform)
     transport_name = _name(transport_profile) or _name(owner_profile) or primary_profile
+    transport_inferred = adapter is None and _name(transport_profile) is None and _name(owner_profile) is None
     transport_ref = weakref.ref(adapter) if adapter is not None else None
 
     if not multiplexed:
@@ -262,6 +268,6 @@ def resolve_identity(
     identity = RoutingIdentity(
         transport_profile=transport_name, runtime_profile=runtime_name,
         authorization_home=authorization_home, runtime_home=runtime_home,
-        multiplexed=True, transport=transport_ref)
+        multiplexed=True, transport=transport_ref, transport_inferred=transport_inferred)
     setattr(source, _IDENTITY_ATTR, identity)
     return identity

tests/gateway/test_session_identity_restore.py

@@ -108,12 +108,12 @@ def test_restored_lane_delivers_through_the_bot_that_received_it_never_the_defau
     assert (restored_identity.transport_profile, restored_identity.runtime_profile) == ("team_b", "ops")
     assert restored_identity.authorization_home == mux.home / "profiles" / "team_b"
     assert restored_identity.runtime_home == mux.home / "profiles" / "ops"
-    assert fresh.runner._adapter_for_source(source) is fresh.team_b
+    assert fresh.runner._delivery_adapter_for(source) is fresh.team_b
     assert fresh.runner._adapter_profile_for_source(source) == "team_b"
     assert fresh.runner._authorization_home_for_source(source) == mux.home / "profiles" / "team_b"
     # Fail closed: team_b's bot did not reconnect → nothing delivers; the default bot never does.
     fresh.runner._profile_adapters["team_b"] = {}
-    assert fresh.runner._adapter_for_source(source) is None
+    assert fresh.runner._delivery_adapter_for(source) is None
 
     # The satellite lane (shared default bot, runtime ops) keeps its default-bot egress.
     shared = mux.primary.build_source(chat_id="72719239", chat_type="dm", user_id="72719239")
@@ -122,7 +122,7 @@ def test_restored_lane_delivers_through_the_bot_that_received_it_never_the_defau
     assert shared_entry.transport_profile == "default"
     fresh2, shared_source, _ = _restart(mux, shared_entry)
     assert identity_of(shared_source).transport_profile == "default"
-    assert fresh2.runner._adapter_for_source(shared_source) is fresh2.primary
+    assert fresh2.runner._delivery_adapter_for(shared_source) is fresh2.primary
 
     # A routing entry written before the column existed: nothing is pinned, old chain unchanged.
     legacy = entry.to_dict()
@@ -130,7 +130,7 @@ def test_restored_lane_delivers_through_the_bot_that_received_it_never_the_defau
     fresh3 = _runner(mux.home)
     legacy_source = fresh3.runner._restored_source(SessionEntry.from_dict(legacy))
     assert identity_of(legacy_source) is None
-    assert fresh3.runner._adapter_for_source(legacy_source) is fresh3.primary  # the heuristic, as before
+    assert fresh3.runner._delivery_adapter_for(legacy_source) is fresh3.primary  # the heuristic, as before
 
 
 def test_standalone_gateway_persists_nothing_and_keys_stay_agent_main(tmp_path, monkeypatch):
@@ -150,5 +150,5 @@ def test_standalone_gateway_persists_nothing_and_keys_stay_agent_main(tmp_path,
     fresh = _runner(home, multiplex=False)
     restored = fresh.runner._restored_source(SessionEntry.from_dict(entry.to_dict()))
     assert identity_of(restored) is None
-    assert fresh.runner._adapter_for_source(restored) is fresh.primary
+    assert fresh.runner._delivery_adapter_for(restored) is fresh.primary
     assert fresh.runner._session_key_for_source(restored) == "agent:main:telegram:dm:4040"

website/docs/developer-guide/multiplexing-gateway.md

@@ -227,7 +227,7 @@ it. `tests/gateway/test_multiplex_transport_matrix.py` asserts every row.
 The routing entry persists `transport_profile` next to the key (and the
 `sessions.transport_profile` column in `state.db`), so after a restart a
 revived lane still knows which bot received it: `_restored_source(entry)`
-re-pins a `RoutingIdentity` with no live adapter and `_adapter_for_source`
+re-pins a `RoutingIdentity` with no live adapter and `_delivery_adapter_for`
 delivers through that bot's adapter or fails closed — a satellite routed
 through the default bot keeps answering from the default bot, a lane owned by
 a secondary never falls back to the default bot's credential. Entries written