fix(orchestrator): propagate dispatcher and watcher failures

src/prime_rl/orchestrator/orchestrator.py

@@ -62,6 +62,7 @@
 from prime_rl.orchestrator.utils import (
     get_weight_dir,
     intercept_vf_logging,
+    raise_for_failed_component_tasks,
     save_rollouts,
     set_default_executor,
     setup_policy_inference_pool,
@@ -455,6 +456,7 @@ async def main_loop(self) -> None:
         to the train / eval sink. Both sinks return a finalized batch (or
         ``None``) from ``add()``; we just dispatch on the result."""
         while not self.stopped.is_set():
+            raise_for_failed_component_tasks(self.component_tasks)
             if self.draining and self.dispatcher.is_idle:
                 get_logger().info("Pipeline drained, exiting main loop")
                 self.stopped.set()

src/prime_rl/orchestrator/utils.py

@@ -19,6 +19,16 @@
 )
 
 
+def raise_for_failed_component_tasks(tasks: list[asyncio.Task]) -> None:
+    """Raise a named error for any failed long-lived orchestrator task."""
+    for task in tasks:
+        if not task.done() or task.cancelled():
+            continue
+        exception = task.exception()
+        if exception is not None:
+            raise RuntimeError(f"Orchestrator component task {task.get_name()!r} failed") from exception
+
+
 async def setup_policy_inference_pool(*, config: OrchestratorConfig, tokenizer):
     """Build the live policy inference pool + matching renderer. Returns
     ``(renderer, inference_pool)``.

tests/unit/orchestrator/test_orchestrator_setup.py

@@ -2,9 +2,26 @@
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, patch
 
+import pytest
 from renderers import Qwen3VLRendererConfig
 
-from prime_rl.orchestrator.utils import setup_policy_inference_pool
+from prime_rl.orchestrator.utils import raise_for_failed_component_tasks, setup_policy_inference_pool
+
+
+def test_raise_for_failed_component_tasks_preserves_cause():
+    async def run() -> None:
+        async def fail_weight_update() -> None:
+            raise ConnectionError("weight update failed")
+
+        task = asyncio.create_task(fail_weight_update(), name="watcher")
+        await asyncio.sleep(0)
+
+        with pytest.raises(RuntimeError, match="component task 'watcher' failed") as exc_info:
+            raise_for_failed_component_tasks([task])
+
+        assert isinstance(exc_info.value.__cause__, ConnectionError)
+
+    asyncio.run(run())
 
 
 def test_setup_policy_inference_pool_uses_renderer_when_enabled():