fix(agent-server): propagate subscriber init errors (OpenHands#3200)

Author: VascoSch92

openhands-agent-server/openhands/agent_server/conversation_service.py

@@ -993,28 +993,32 @@ async def _start_event_service(self, stored: StoredConversation) -> EventService
             cipher=self.cipher,
             owner_instance_id=self.owner_instance_id,
         )
-        # Create subscribers...
-        await event_service.subscribe_to_events(_EventSubscriber(service=event_service))
-        if stored.autotitle and stored.title is None:
-            await event_service.subscribe_to_events(
-                AutoTitleSubscriber(service=event_service)
-            )
-        asyncio.gather(
-            *[
-                event_service.subscribe_to_events(
-                    WebhookSubscriber(
-                        conversation_id=stored.id,
-                        service=event_service,
-                        spec=webhook_spec,
-                        session_api_key=self.session_api_key,
-                    )
-                )
-                for webhook_spec in self.webhook_specs
-            ]
-        )
 
         try:
             await event_service.start()
+            # Register subscribers after start() so subscribe_to_events runs
+            # its initial-state push synchronously and any failure surfaces to
+            # the caller instead of being silently logged on a later publish.
+            await event_service.subscribe_to_events(
+                _EventSubscriber(service=event_service)
+            )
+            if stored.autotitle and stored.title is None:
+                await event_service.subscribe_to_events(
+                    AutoTitleSubscriber(service=event_service)
+                )
+            await asyncio.gather(
+                *[
+                    event_service.subscribe_to_events(
+                        WebhookSubscriber(
+                            conversation_id=stored.id,
+                            service=event_service,
+                            spec=webhook_spec,
+                            session_api_key=self.session_api_key,
+                        )
+                    )
+                    for webhook_spec in self.webhook_specs
+                ]
+            )
             # Save metadata immediately after successful start to ensure persistence
             # even if the system is not shut down gracefully
             await event_service.save_meta()

openhands-agent-server/openhands/agent_server/event_service.py

@@ -424,10 +424,7 @@ async def subscribe_to_events(self, subscriber: Subscriber[Event]) -> UUID:
                     f"Initial state push to subscriber {subscriber_id} timed "
                     f"out after {INITIAL_STATE_PUSH_TIMEOUT_SECONDS}s."
                 )
-            except Exception as e:
-                logger.error(
-                    f"Error sending initial state to subscriber {subscriber_id}: {e}"
-                )
+            # Non-timeout errors propagate to caller (e.g. webhook failures).
 
         return subscriber_id
 

tests/agent_server/test_webhook_subscriber.py

@@ -1304,19 +1304,6 @@ async def test_timer_cancelled_when_buffer_full(
         subscriber._post_events.assert_called_once()
 
 
-@pytest.mark.xfail(
-    strict=True,
-    reason=(
-        "Subscribe-time errors from a subscriber's initial __call__ never "
-        "reach the start_conversation caller. Two swallow sites in series: "
-        "event_service.py:411-416 wraps the initial state sync in "
-        "try/except + logger.error (no re-raise), and "
-        "conversation_service.py:862 fires asyncio.gather() on the webhook "
-        "subscribes without awaiting. Both must be fixed for init errors "
-        "to surface. "
-        "Tracked in https://github.com/OpenHands/software-agent-sdk/issues/3121."
-    ),
-)
 @pytest.mark.timeout(30)
 async def test_webhook_subscribe_errors_surface(tmp_path, monkeypatch):
     persist = tmp_path / "persist"