feat(subscription): BridgesLegacyView and pickle BRIDGE_SND shim (V2-P2-007)

Export subscriptions to legacy BRIDGES dict and encode v1 monitor pickle
frames; library-only, not wired to report runtime yet.

src/adn_server/application/subscription/__init__.py

@@ -1,10 +1,12 @@
 """Subscription application helpers."""
 
 from .bridges_export import export_bridges, subscription_to_legacy_row
+from .bridges_legacy_view import BridgesLegacyView
 from .ingress import build_voice_ingress
 from .router import SubscriptionRouter
 
 __all__ = [
+    "BridgesLegacyView",
     "SubscriptionRouter",
     "build_voice_ingress",
     "export_bridges",

src/adn_server/application/subscription/bridges_legacy_view.py

@@ -0,0 +1,19 @@
+"""Legacy BRIDGES view from ``SubscriptionStore`` (v1 monitor shim, D-08)."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from adn_server.application.ports import SubscriptionStore
+from adn_server.application.subscription.bridges_export import export_bridges
+
+
+class BridgesLegacyView:
+    """One-way export: subscriptions → legacy ``BRIDGES`` dict (pickle / ``BRIDGE_SND``)."""
+
+    def __init__(self, store: SubscriptionStore) -> None:
+        self._store = store
+
+    def generate(self, *, now: float | None = None) -> dict[str, list[dict[str, Any]]]:
+        """Build a pickle-compatible ``BRIDGES`` snapshot (legacy ``bridge_master.send_bridge``)."""
+        return export_bridges(self._store, now=now)

src/adn_server/infrastructure/twisted_adapters/report/pickle_legacy.py

@@ -0,0 +1,19 @@
+"""Pickle wire helpers for legacy report v1 monitor shim."""
+
+from __future__ import annotations
+
+import pickle
+from typing import Any
+
+from .opcodes import REPORT_OPCODES
+
+_PICKLE_PROTOCOL = 2
+
+
+def encode_bridge_snd_frame(
+    bridges: dict[str, Any],
+    *,
+    protocol: int = _PICKLE_PROTOCOL,
+) -> bytes:
+    """``BRIDGE_SND`` opcode + ``pickle.dumps(BRIDGES, protocol=2)`` (legacy parity)."""
+    return REPORT_OPCODES["BRIDGE_SND"] + pickle.dumps(bridges, protocol=protocol)

tests/application/test_bridges_legacy_view.py

@@ -0,0 +1,56 @@
+"""BridgesLegacyView: subscription store → pickle BRIDGE_SND shim."""
+
+from __future__ import annotations
+
+import pickle
+
+from adn_server.application.subscription.bridges_export import export_bridges
+from adn_server.application.subscription.bridges_legacy_view import BridgesLegacyView
+from adn_server.infrastructure.twisted_adapters.report.pickle_legacy import encode_bridge_snd_frame
+from adn_server.domain import bytes_3
+from adn_server.domain.subscription import (
+    ActivationPolicy,
+    AudioChannel,
+    Subscription,
+    SubscriptionPhase,
+    SubscriptionRole,
+    SubscriptionState,
+    SystemId,
+    TgId,
+)
+from adn_server.infrastructure.subscription_store import InMemorySubscriptionStore
+from adn_server.infrastructure.twisted_adapters.report.opcodes import REPORT_OPCODES
+
+
+def _sample_store() -> InMemorySubscriptionStore:
+    store = InMemorySubscriptionStore()
+    store.upsert(
+        Subscription(
+            channel=AudioChannel(tgid=TgId(730444), slot=1),
+            system=SystemId("MASTER-A"),
+            target_tgid=TgId(730444),
+            role=SubscriptionRole.SINK,
+            policy=ActivationPolicy.INBAND,
+            state=SubscriptionState(phase=SubscriptionPhase.ACTIVE),
+        )
+    )
+    return store
+
+
+def test_generate_matches_export_bridges():
+    store = _sample_store()
+    view = BridgesLegacyView(store)
+    now = 1_700_000_000.0
+    assert view.generate(now=now) == export_bridges(store, now=now)
+
+
+def test_bridge_snd_frame_is_pickle_protocol_2():
+    store = _sample_store()
+    frame = encode_bridge_snd_frame(BridgesLegacyView(store).generate())
+    assert frame[:1] == REPORT_OPCODES["BRIDGE_SND"]
+    bridges = pickle.loads(frame[1:], encoding="bytes")
+    assert "730444" in bridges
+    row = bridges["730444"][0]
+    assert row["SYSTEM"] == "MASTER-A"
+    assert row["TGID"] == bytes_3(730444)
+    assert isinstance(row["ACTIVE"], bool)