fix(opal-server): retrieve webhook task exceptions before dropping them

The webhook task sweep dropped finished tasks without inspecting
t.exception(), and stop() only gathered self._tasks — so an error raised
inside trigger() never reached the app logger, surfacing only as
asyncio's generic "Task exception was never retrieved" at GC time. Log
failed trigger tasks at ERROR during the sweep and include webhook tasks
in the stop() gather.

Also de-flake the cleanup test: await the freshly scheduled trigger task
directly instead of relying on an asyncio.sleep(0) scheduler tick, and
add a test asserting the failure of a trigger task is retrieved and
logged.

Addresses review comments:
- #923 (comment) (@zeevmoney)
- #923 (comment) (@zeevmoney)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: Zivxx

packages/opal-server/opal_server/policy/watcher/task.py

@@ -30,7 +30,12 @@ async def _on_webhook(self, topic: Topic, data: Any):
         logger.info(f"Webhook listener triggered ({len(self._webhook_tasks)})")
         # Rebuild rather than remove-while-iterating: list.remove() inside a
         # `for t in self._webhook_tasks` loop skips the element after each removal,
-        # so finished tasks accumulate.
+        # so finished tasks accumulate. Retrieve exceptions before dropping the
+        # references — otherwise a failed trigger() is only reported by asyncio's
+        # generic "Task exception was never retrieved" at GC time.
+        for t in self._webhook_tasks:
+            if t.done() and not t.cancelled() and t.exception() is not None:
+                logger.error(f"Webhook trigger task failed: {t.exception()!r}")
         self._webhook_tasks = [t for t in self._webhook_tasks if not t.done()]
         self._webhook_tasks.append(asyncio.create_task(self.trigger(topic, data)))
 
@@ -72,7 +77,7 @@ async def stop(self):
         for task in self._tasks + self._webhook_tasks:
             if not task.done():
                 task.cancel()
-        await asyncio.gather(*self._tasks, return_exceptions=True)
+        await asyncio.gather(*self._tasks, *self._webhook_tasks, return_exceptions=True)
 
     async def trigger(self, topic: Topic, data: Any):
         """Triggers the policy watcher from outside to check for changes (git

packages/opal-server/opal_server/tests/webhook_task_cleanup_test.py

@@ -22,7 +22,6 @@ async def _done():
     w._webhook_tasks = list(finished)
 
     await w._on_webhook("webhook", None)
-    await asyncio.sleep(0)  # let the newly created trigger task finish
 
     # all 3 done ones removed...
     remaining_done = [t for t in w._webhook_tasks if t in finished]
@@ -32,4 +31,38 @@ async def _done():
     assert len(w._webhook_tasks) == 1
     assert len(survivors) == 1, f"new trigger task not scheduled: {w._webhook_tasks}"
 
-    await asyncio.gather(*survivors)  # drain the dangling task
+    # Await the trigger task directly (not sleep(0)) so the test doesn't depend
+    # on scheduler tick ordering.
+    await asyncio.gather(*survivors)
+
+
+class _FailingWatcher(BasePolicyWatcherTask):
+    async def trigger(self, topic, data):
+        raise RuntimeError("trigger blew up")
+
+
+@pytest.mark.asyncio
+async def test_failed_trigger_exception_is_retrieved_and_logged():
+    """Sweeping a failed trigger task must retrieve and log its exception, not
+    silently drop the reference (asyncio's 'exception was never retrieved')."""
+    from opal_common.logger import logger as opal_logger
+
+    w = _FailingWatcher(pubsub_endpoint=None)
+
+    await w._on_webhook("webhook", None)  # schedules a trigger that raises
+    failed = list(w._webhook_tasks)
+    await asyncio.gather(*failed, return_exceptions=True)
+
+    records = []
+    sink_id = opal_logger.add(lambda m: records.append(str(m)), level="ERROR")
+    try:
+        await w._on_webhook("webhook", None)  # sweep must log the failure
+    finally:
+        opal_logger.remove(sink_id)
+
+    assert failed[0] not in w._webhook_tasks, "failed task not swept"
+    assert any(
+        "Webhook trigger task failed" in r and "trigger blew up" in r for r in records
+    ), f"failure not logged: {records}"
+
+    await asyncio.gather(*w._webhook_tasks, return_exceptions=True)