Envoy backend protocol gaps

PLUGINS.md

@@ -165,7 +165,6 @@ Actions represent mutations and configuration changes triggered before or during
 | `get_exclusive_site_group(site_id)` | Retrieves or creates a site group exclusive to the specified site. |
 | `create_site_control(site_control_group_id, control: SiteControlWrite)` | Schedules a DERControl (DOE limits, start/duration). |
 | `cancel_active_site_controls()` | Cancels all active and scheduled DERControls across all groups. |
-| `delete_site_controls_for_group(site_control_group_id)` | Deletes controls belonging to a specific group. |
 | `set_site_control_default(site_control_group_id, default: SiteControlGroupDefaultWrite)` | Applies or clears DefaultDERControl limits for a DERProgram. |
 | `update_runtime_config(config: RuntimeConfigWrite)` | Applies runtime configuration updates (e.g., server features, timing). |
 | `delete_all_site_control_groups()` | Removes all DERPrograms, DERControls, and DefaultDERControls. |

src/cactus_runner/plugin/backends/common.py

@@ -366,17 +366,6 @@ async def cancel_active_site_controls(
         """
         ...
 
-    async def delete_site_controls_for_group(self, site_control_group_id: str) -> None:
-        """Delete all controls for the given group, if can be differentiated from cancelling.
-
-        Args:
-            site_control_group_id: id of group the controls will belong to
-
-        Raises:
-            Exception: If a cancellation request fails.
-        """
-        ...
-
     # ------------------------------------------------------------------
     # Default DER Controls
     # ------------------------------------------------------------------

src/cactus_runner/plugin/backends/envoy/backend.py

@@ -545,6 +545,24 @@ async def get_site_control_responses(self) -> Sequence[dtos.SiteControlResponse]
     # Subscriptions
     # ------------------------------------------------------------------
 
+    async def get_subscription(self, subscription_id: str) -> dtos.Subscription | None:
+        """Returns a single subscription by ID.
+
+        Args:
+            subscription_id: The string-encoded subscription identifier.
+
+        Returns:
+            The matching subscription, or None if no subscription with that ID exists.
+        """
+        stmt = select(Subscription).where(Subscription.subscription_id == int(subscription_id))
+
+        async with self._session_factory() as session:
+            subscription = (await session.execute(stmt)).scalar_one_or_none()
+
+        if subscription is None:
+            return None
+        return mappers.map_envoy_subscription_to_dto(subscription)
+
     async def get_subscriptions(
         self,
         aggregator_client_id: str | None = None,

tests/unit/app/test_check.py

@@ -2601,11 +2601,45 @@ async def test_check_all_notifications_transmitted_failure_logs(pg_base_config,
         backend = EnvoyBackend(session_factory=lambda: session, admin_client=mock_admin_client)
         actual = await check_all_notifications_transmitted(backend)
         assert_check_result(actual, False)
+        assert actual.description is not None
+        assert "not presented by the runner backend" in actual.description, "No Subscription exists for the log"
 
     # Currently not relying on admin api for checks. This may change.
     mock_admin_client.assert_not_called()
 
 
+@pytest.mark.anyio
+async def test_check_all_notifications_transmitted_failure_reports_notification_uri(pg_base_config):
+    """check_all_notifications_transmitted should report the notification_uri of the failing subscription"""
+
+    async with generate_async_session(pg_base_config) as session:
+        agg = (await session.execute(select(Aggregator).where(Aggregator.aggregator_id == 1))).scalar_one()
+        session.add(
+            generate_class_instance(
+                Subscription, subscription_id=1, aggregator=agg, scoped_site_id=None, notification_uri="https://dev/1"
+            )
+        )
+        session.add(
+            generate_class_instance(
+                TransmitNotificationLog,
+                transmit_notification_log_id=None,
+                subscription_id_snapshot=1,
+                http_status_code=500,
+            )
+        )
+        await session.commit()
+
+    mock_admin_client = mock.Mock(spec=EnvoyAdminClient)
+    async with generate_async_session(pg_base_config) as session:
+        backend = EnvoyBackend(session_factory=lambda: session, admin_client=mock_admin_client)
+        actual = await check_all_notifications_transmitted(backend)
+        assert_check_result(actual, False)
+        assert actual.description is not None
+        assert "https://dev/1" in actual.description
+
+    mock_admin_client.assert_not_called()
+
+
 @pytest.mark.parametrize("input_val", [-1, {"a": 2}, 99999998, [1, 2, 3], "abc123"])
 def test_response_type_to_string_bad_values(input_val):
     output = response_type_to_string(input_val)

tests/unit/plugin/backends/test_envoy.py

@@ -1,3 +1,4 @@
+import inspect
 from datetime import UTC, datetime
 from unittest.mock import AsyncMock, Mock
 
@@ -14,14 +15,63 @@
     SiteGroup,
     SiteReading,
     SiteReadingType,
+    Subscription,
 )
+from envoy.server.model.aggregator import Aggregator
 from envoy.server.model.archive import ArchiveDynamicOperatingEnvelope
+from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from cactus_runner.plugin.backends.common import RunnerBackend
 from cactus_runner.plugin.backends.envoy import EnvoyAdminClient, EnvoyBackend
 from cactus_runner.plugin.backends.envoy.resolver import EnvoyResolver
 
 
+def test_envoy_backend_implements_runner_backend_protocol() -> None:
+    """Ensure EnvoyBackend defines every method required by RunnerBackend."""
+    protocol_methods = {
+        name for name, member in vars(RunnerBackend).items() if inspect.isfunction(member) and not name.startswith("_")
+    }
+    envoy_backend_methods = {name for name, member in vars(EnvoyBackend).items() if inspect.isfunction(member)}
+
+    missing_methods = sorted(protocol_methods - envoy_backend_methods)
+    assert not missing_methods, f"EnvoyBackend is missing RunnerBackend methods: {missing_methods}"
+
+
+@pytest.mark.anyio
+async def test_get_subscription(pg_base_config, envoy_admin_client) -> None:
+    """Tests that get_subscription returns the matching subscription by ID, or None if it doesn't exist."""
+    async with generate_async_session(pg_base_config) as session:
+        agg = (await session.execute(select(Aggregator).where(Aggregator.aggregator_id == 1))).scalar_one()
+        session.add(
+            generate_class_instance(
+                Subscription, subscription_id=1, aggregator=agg, scoped_site_id=None, notification_uri="https://dev/1"
+            )
+        )
+        session.add(
+            generate_class_instance(
+                Subscription,
+                seed=101,
+                subscription_id=2,
+                aggregator=agg,
+                scoped_site_id=None,
+                notification_uri="https://dev/2",
+            )
+        )
+        await session.commit()
+
+    async with generate_async_session(pg_base_config) as session:
+        backend = EnvoyBackend(session_factory=lambda: session, admin_client=envoy_admin_client)
+        sub = await backend.get_subscription("2")
+        missing = await backend.get_subscription("999")
+
+    assert sub is not None
+    assert sub.subscription_id == "2"
+    assert sub.client_aggregator_id == "1"
+    assert sub.notification_uri == "https://dev/2"
+    assert missing is None
+
+
 @pytest.mark.anyio
 async def test_get_site_controls(pg_base_config, envoy_admin_client) -> None:
     """Tests that the get_site_controls returns all controls as expected."""