Add extension event transport planning and migration hooks

.github/agents/axis-review-verify.agent.md

@@ -65,7 +65,7 @@ You are a verification-focused reviewer for the Axis Python repository. Your job
 ## Review Priorities
 1. Correctness and regressions:
 - Handler initialization phase behavior in `axis/interfaces/` (`API_DISCOVERY`, `PARAM_CGI_FALLBACK`, `APPLICATION`).
-- Event/state behavior in `axis/models/event.py` and `axis/models/event_instance.py`.
+- Event/state behavior in `axis/models/event.py` and `axis/models/events/event_instance.py`.
 - Model input normalization and default handling at boundaries.
 2. Reliability and compatibility:
 - Unknown enum and external input handling (`_missing_`, safe defaults).

.github/agents/axis-review.agent.md

@@ -67,7 +67,7 @@ user-invocable: true
 ## Review Priorities
 1. Correctness and regressions:
 - Behavior changes in `axis/interfaces/` handlers and initialization flow (`API_DISCOVERY`, `PARAM_CGI_FALLBACK`, `APPLICATION`).
-- State/event semantics in `axis/models/event.py` and `axis/models/event_instance.py`.
+- State/event semantics in `axis/models/event.py` and `axis/models/events/event_instance.py`.
 - Input boundary normalization in model constructors or `__post_init__`.
 2. Reliability and compatibility:
 - Unknown enum and external input handling (`_missing_` fallbacks, safe defaults).

axis/device.py

@@ -2,7 +2,8 @@
 
 from typing import TYPE_CHECKING
 
-from .interfaces.event_manager import EventManager
+from .interfaces.events.event_manager import EventManager
+from .interfaces.events.transport_manager import EventTransportManager
 from .interfaces.vapix import Vapix
 from .stream_manager import StreamManager
 
@@ -19,6 +20,12 @@ def __init__(self, configuration: Configuration) -> None:
         self.vapix = Vapix(self)
         self.stream = StreamManager(self)
         self.event = EventManager()
+        self.event_transport = EventTransportManager(
+            event_instances=self.vapix.event_instances,
+            stream=self.stream,
+            event=self.event,
+            mqtt=self.vapix.mqtt,
+        )
 
     def enable_events(self) -> None:
         """Enable events for stream."""

axis/interfaces/events/__init__.py

@@ -0,0 +1 @@
+"""Event interface handlers for Axis devices."""

axis/interfaces/events/event_instances.py

@@ -0,0 +1,100 @@
+"""Event service and action service APIs available in Axis network device."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+from ...models.events.event_instance import (
+    EventInstance,
+    ListEventInstancesRequest,
+    ListEventInstancesResponse,
+)
+from ...models.events.topic_filter import (
+    EventTopicFilter,
+    to_canonical,
+    to_topic_filter,
+)
+from ..api_handler import ApiHandler
+from .event_manager import BLACK_LISTED_TOPICS
+
+if TYPE_CHECKING:
+    from ...models.event import Event
+
+
+class EventInstanceHandler(ApiHandler[EventInstance]):
+    """Event instances for Axis devices."""
+
+    async def _api_request(self) -> dict[str, EventInstance]:
+        """Get default data of API discovery."""
+        return await self.get_event_instances()
+
+    async def get_event_instances(self) -> dict[str, EventInstance]:
+        """List all event instances."""
+        bytes_data = await self.vapix.api_request(ListEventInstancesRequest())
+        response = ListEventInstancesResponse.decode(bytes_data)
+        return response.data
+
+    def get_expected_events_per_topic(
+        self,
+        include_internal_topics: bool = False,
+    ) -> dict[str, list[Event]]:
+        """Return expected startup events grouped by topic.
+
+        Event instances are the protocol-agnostic bootstrap source for startup
+        predeclaration. Returned events are synthesized from schema data and represent
+        expected event identity/state (operation=Initialized), not live stream updates.
+        """
+        grouped: dict[str, list[Event]] = {}
+        for item in self.values():
+            if not include_internal_topics and item.topic in BLACK_LISTED_TOPICS:
+                continue
+            grouped[item.topic] = item.to_events()
+        return grouped
+
+    def get_supported_topics(
+        self, include_internal_topics: bool = False
+    ) -> tuple[str, ...]:
+        """Return canonical supported event topics from event instances."""
+        expected = self.get_expected_events_per_topic(
+            include_internal_topics=include_internal_topics
+        )
+        return tuple(sorted(to_canonical(topic) for topic in expected))
+
+    def get_supported_event_descriptors(
+        self, include_internal_topics: bool = False
+    ) -> dict[str, dict[str, Any]]:
+        """Return normalized supported-event descriptors for extension consumers."""
+        descriptors: dict[str, dict[str, Any]] = {}
+        for topic, events in self.get_expected_events_per_topic(
+            include_internal_topics=include_internal_topics
+        ).items():
+            canonical_topic = to_canonical(topic)
+            topic_filter = to_topic_filter(canonical_topic)
+            descriptors[canonical_topic] = {
+                "topic": canonical_topic,
+                "topic_filter": topic_filter,
+                "count": len(events),
+                "sources": tuple(
+                    sorted({event.source for event in events if event.source})
+                ),
+            }
+        return descriptors
+
+    def build_transport_filter_payloads(
+        self,
+        event_filter: EventTopicFilter | None = None,
+        include_internal_topics: bool = False,
+    ) -> dict[str, list[str]]:
+        """Build no-op extension payloads for MQTT and WebSocket topic filters."""
+        if event_filter is None or event_filter.is_wildcard:
+            topics = list(self.get_supported_topics(include_internal_topics))
+        else:
+            topics = event_filter.canonical_topics
+
+        topic_filters = [to_topic_filter(topic) for topic in topics]
+
+        return {
+            "canonical_topics": topics,
+            "mqtt_topics": topic_filters,
+            "websocket_topic_filters": topic_filters,
+        }

axis/interfaces/event_manager.py → axis/interfaces/events/event_manager.py

@@ -4,7 +4,8 @@
 import logging
 from typing import Any
 
-from ..models.event import Event, EventOperation, EventTopic
+from ...models.event import Event, EventOperation, EventTopic
+from ...models.events.topic_filter import to_canonical
 
 SubscriptionCallback = Callable[[Event], None]
 SubscriptionType = tuple[
@@ -32,8 +33,20 @@ def __init__(self) -> None:
         """Ready information about events."""
         self._known_topics: set[str] = set()
         self._unsupported_topics: set[str] = set()
+        self._allowed_topics: set[str] | None = None
         self._subscribers: dict[str, list[SubscriptionType]] = {ID_FILTER_ALL: []}
 
+    def set_allowed_topics(self, topics: list[str] | set[str] | None) -> None:
+        """Set optional canonical topic allow-list for runtime filtering.
+
+        Default behavior remains unchanged until this extension is explicitly used.
+        """
+        if topics is None:
+            self._allowed_topics = None
+            return
+
+        self._allowed_topics = {to_canonical(topic) for topic in topics}
+
     def handler(self, data: bytes | dict[str, Any]) -> None:
         """Create event and pass it along to subscribers."""
         event = Event.decode(data)
@@ -49,6 +62,12 @@ def handler(self, data: bytes | dict[str, Any]) -> None:
         if event.topic in BLACK_LISTED_TOPICS:
             return
 
+        if (
+            self._allowed_topics is not None
+            and to_canonical(event.topic) not in self._allowed_topics
+        ):
+            return
+
         known = (unique_topic := f"{event.topic}_{event.id}") not in self._known_topics
         self._known_topics.add(unique_topic)
 

axis/interfaces/events/transport_manager.py

@@ -0,0 +1,115 @@
+"""Event transport orchestration manager."""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING, Any
+
+from ...models.events.topic_filter import EventTopicFilter
+from ...models.events.transport_capabilities import (
+    TRANSPORT_FILTER_CAPABILITIES,
+    EventTransport,
+    TransportFilterCapability,
+)
+
+if TYPE_CHECKING:
+    from ...stream_manager import StreamManager
+    from ..mqtt import MqttClientHandler
+    from .event_instances import EventInstanceHandler
+    from .event_manager import EventManager
+
+LOGGER = logging.getLogger(__name__)
+
+
+class EventTransportManager:
+    """Owns event transport filter orchestration across subsystems."""
+
+    def __init__(
+        self,
+        event_instances: EventInstanceHandler,
+        stream: StreamManager,
+        event: EventManager,
+        mqtt: MqttClientHandler,
+    ) -> None:
+        """Store subsystem dependencies used for transport filter application."""
+        self._event_instances = event_instances
+        self._stream = stream
+        self._event = event
+        self._mqtt = mqtt
+
+    def get_event_transport_capabilities(
+        self,
+    ) -> dict[EventTransport, TransportFilterCapability]:
+        """Return extension capability matrix for event transports."""
+        return dict(TRANSPORT_FILTER_CAPABILITIES)
+
+    def get_supported_event_descriptors(
+        self,
+        include_internal_topics: bool = False,
+    ) -> dict[str, dict[str, Any]]:
+        """Return normalized supported-event descriptors from event instances.
+
+        This method is extension-oriented and has no side effects.
+        """
+        return self._event_instances.get_supported_event_descriptors(
+            include_internal_topics=include_internal_topics
+        )
+
+    async def apply_event_transport_filters(
+        self,
+        event_filter: EventTopicFilter | None = None,
+        include_internal_topics: bool = False,
+    ) -> None:
+        """Apply validated transport filters to websocket, MQTT, and local fallback.
+
+        This method performs no implicit event-instance initialization. Callers must
+        invoke ``initialize_event_instances`` before applying filters.
+        """
+        if not self._event_instances.initialized:
+            msg = (
+                "Event instances are not initialized. "
+                "Call initialize_event_instances() first."
+            )
+            raise RuntimeError(msg)
+
+        if event_filter is None:
+            supported = self._event_instances.get_supported_topics(
+                include_internal_topics=include_internal_topics
+            )
+            event_filter = (
+                EventTopicFilter.from_topics(list(supported))
+                if supported
+                else EventTopicFilter.for_all_events()
+            )
+
+        if not event_filter.is_wildcard:
+            supported_topics = set(
+                self._event_instances.get_supported_topics(
+                    include_internal_topics=include_internal_topics
+                )
+            )
+            unknown_topics = sorted(
+                set(event_filter.canonical_topics) - supported_topics
+            )
+            if unknown_topics:
+                message = f"Requested unsupported topics: {', '.join(unknown_topics)}"
+                raise ValueError(message)
+
+        self._stream.set_event_subscription(event_filter)
+
+        if self._mqtt.supported:
+            await self._mqtt.configure_event_publication(
+                event_filter.transport_topic_filters
+            )
+
+        self._event.set_allowed_topics(event_filter.canonical_topics or None)
+
+        LOGGER.debug(
+            "Applied event transport filters: websocket=%s mqtt=%s local=%s topics=%s",
+            True,
+            self._mqtt.supported,
+            True,
+            len(event_filter.canonical_topics)
+            if not event_filter.is_wildcard
+            else "all",
+        )

axis/interfaces/mqtt.py

@@ -3,6 +3,7 @@
 from typing import Any
 
 from ..models.api_discovery import ApiId
+from ..models.events.topic_filter import to_topic_filter
 from ..models.mqtt import (
     API_VERSION,
     ActivateClientRequest,
@@ -66,12 +67,28 @@ async def get_event_publication_config(self) -> EventPublicationConfig:
         response = GetEventPublicationConfigResponse.decode(bytes_data)
         return response.data
 
+    def build_topic_filters(
+        self, topics: list[str], normalize_topics: bool = False
+    ) -> list[str]:
+        """Build MQTT topic filters.
+
+        Defaults preserve previous behavior unless normalization is requested.
+        """
+        normalized = (
+            [to_topic_filter(topic) for topic in topics] if normalize_topics else topics
+        )
+        # Preserve input order while deduplicating.
+        return list(dict.fromkeys(normalized))
+
     async def configure_event_publication(
-        self, topics: list[str] = DEFAULT_TOPICS
+        self,
+        topics: list[str] = DEFAULT_TOPICS,
+        normalize_topics: bool = False,
     ) -> None:
         """Configure MQTT Client event publication."""
+        topic_filters = self.build_topic_filters(topics, normalize_topics)
         event_filters = EventFilter.from_list(
-            [{"topicFilter": topic} for topic in topics]
+            [{"topicFilter": topic} for topic in topic_filters]
         )
         config = EventPublicationConfig(event_filter_list=event_filters)
         await self.vapix.api_request(

axis/interfaces/vapix.py

@@ -21,7 +21,7 @@
 from .applications.object_analytics import ObjectAnalyticsHandler
 from .applications.vmd4 import Vmd4Handler
 from .basic_device_info import BasicDeviceInfoHandler
-from .event_instances import EventInstanceHandler
+from .events.event_instances import EventInstanceHandler
 from .light_control import LightHandler
 from .mqtt import MqttClientHandler
 from .parameters.param_cgi import Params