fix(board): add first-chunk timeout to _stream_expert_speech

Commit 36b0296 changed expert speech generation from gateway.chat()
(non-streaming) to gateway.chat_stream() (streaming) for progressive
UI output. However, DashScope/DeepSeek via LiteLLM occasionally accept
the streaming request but never emit the first SSE chunk — the async
generator hangs indefinitely with no error and no timeout, freezing
the entire board.

Root cause: `async for chunk in gateway.chat_stream(...)` blocks
forever when the provider silently stalls. The moderator path
(gateway.chat) still works because it's synchronous — that's why
the moderator opening always succeeds but experts hang.

Fix: pull the first chunk via `__anext__()` wrapped in
`asyncio.wait_for(timeout=30s)`. If no chunk arrives within 30s,
close the stream and fall back to non-streaming gateway.chat() +
_replay_stream() (which splits the response into small chunks for
progressive UI rendering). This preserves the "逐个输出" UX while
guaranteeing the board never hangs on a stalled streaming provider.

Add 4 unit tests covering:
- Normal streaming works (existing path)
- First-chunk timeout → fallback to chat()
- Empty stream → fallback to chat()
- chat_stream raising → fallback to chat()

E2E verification: 5-expert board with DashScope completes in ~277s
(status=completed), all experts produce content, no hang.

Same branch as the earlier _handle_llm_gateway fix (commit d0fe661).

src/agentkit/experts/board_orchestrator.py

@@ -311,12 +311,20 @@ class BoardOrchestrator:
         # loop is unchanged.
         return await self._stream_expert_speech(expert, round, prompt)
 
+    # First-chunk timeout for chat_stream. Some OpenAI-compatible providers
+    # (DashScope/DeepSeek via LiteLLM) occasionally accept the request but
+    # never emit the first SSE chunk, causing the board to hang silently.
+    # 30s is the P95 of moderator (non-streaming) chat on this provider —
+    # if streaming can't beat that, we fall back to a working path.
+    _STREAM_FIRST_CHUNK_TIMEOUT_S: float = 30.0
+
     async def _stream_expert_speech(self, expert: Expert, round: int, prompt: str) -> str:
         """Stream an expert's speech via chat_stream, emitting chunks.
 
         Falls back to non-streaming ``chat()`` when ``chat_stream`` is
-        unavailable (e.g. an LLM provider without streaming support) or
+        unavailable (e.g. an LLM provider without streaming support), raises
-        raises before any chunk is produced.
+        before any chunk is produced, or **fails to emit the first chunk
+        within ``_STREAM_FIRST_CHUNK_TIMEOUT_S`` seconds**.
 
         ponytail: when the LLM does not actually stream (returns a single
         big chunk), we still want the UI to see content appearing
@@ -324,22 +332,45 @@ class BoardOrchestrator:
         sentence/line chunks and emit them with a small delay. The
         ``expert_speech_chunk`` event already handles duplicate-sender
         dedup, so emitting many small chunks is safe.
+
+        Regression guard: commit 36b0296 introduced streaming here, but
+        DashScope/DeepSeek via LiteLLM occasionally hang on the first SSE
+        chunk with no error and no timeout. The first-chunk timeout below
+        ensures we fall back to the proven non-streaming path instead of
+        blocking the board indefinitely.
         """
         gateway = self._get_llm_gateway(expert)
         assert gateway is not None  # checked by caller
         total = ""
-        # Emit an opening chunk-less event so the UI can create the streaming
+
-        # placeholder before the first token arrives (keeps the first paint
+        # Try streaming with a hard first-chunk deadline. If the provider
-        # aligned with the streaming indicator).
+        # accepts the request but never emits the first chunk (observed on
+        # DashScope/DeepSeek via LiteLLM), fall through to non-streaming.
+        stream_obj = None
         try:
-            streamed_chunk_count = 0
+            stream_obj = gateway.chat_stream(
-            async for chunk in gateway.chat_stream(
                 messages=[{"role": "user", "content": prompt}],
                 model="default",
-            ):
+            )
+            # Defensive: provider returning a coroutine instead of an async
+            # generator indicates an implementation bug — raise so the
+            # except below picks up the non-streaming fallback.
+            if asyncio.iscoroutine(stream_obj):
+                raise TypeError("chat_stream returned a coroutine, not an async generator")
+
+            # Pull the first chunk with a timeout. If it arrives, the
+            # remainder of the stream is trusted to flow normally.
+            first_chunk = await asyncio.wait_for(
+                stream_obj.__anext__(),
+                timeout=self._STREAM_FIRST_CHUNK_TIMEOUT_S,
+            )
+            streamed_chunk_count = 0
+
+            async def _emit(chunk) -> None:
+                nonlocal total, streamed_chunk_count
                 delta = chunk.content or ""
                 if not delta:
-                    continue
+                    return
                 total += delta
                 streamed_chunk_count += 1
                 await self._broadcast_event(
@@ -353,6 +384,10 @@ class BoardOrchestrator:
                         "role": "expert",
                     },
                 )
+
+            await _emit(first_chunk)
+            async for chunk in stream_obj:
+                await _emit(chunk)
             # If the LLM "streamed" but only delivered one big chunk, still
             # let the UI see content arrive progressively.
             if streamed_chunk_count <= 1 and total:
@@ -363,8 +398,28 @@ class BoardOrchestrator:
                 f"Provider for '{expert.config.name}' lacks chat_stream, "
                 f"falling back to non-streaming: {e}"
             )
+        except asyncio.TimeoutError:
+            logger.warning(
+                f"Expert '{expert.config.name}' stream produced no chunk in "
+                f"{self._STREAM_FIRST_CHUNK_TIMEOUT_S}s — falling back to "
+                f"non-streaming (provider may be DashScope/DeepSeek via LiteLLM)"
+            )
+            # Close the partial stream to avoid resource leak
+            if stream_obj is not None and hasattr(stream_obj, "aclose"):
+                try:
+                    await stream_obj.aclose()
+                except Exception:  # noqa: BLE001 — best-effort cleanup
+                    pass
+        except StopAsyncIteration:
+            # Empty stream — no chunks at all, fall through to fallback
+            logger.info(f"Expert '{expert.config.name}' stream produced no chunks")
         except Exception as e:
             logger.warning(f"Expert '{expert.config.name}' stream failed: {e}")
+            if stream_obj is not None and hasattr(stream_obj, "aclose"):
+                try:
+                    await stream_obj.aclose()
+                except Exception:  # noqa: BLE001 — best-effort cleanup
+                    pass
 
         # Fallback: non-streaming path. Emit the whole content as small
         # chunks so the UI still renders progressively rather than going

tests/unit/experts/test_board_orchestrator.py

@@ -343,3 +343,119 @@ class TestBoardOrchestratorBroadcast:
 
         # 不应抛出异常
         await orchestrator._broadcast_event("board_started", {"topic": "测试"})
+
+
+# ── BoardOrchestrator._stream_expert_speech 测试 ─────────
+
+
+class TestStreamExpertSpeechFallback:
+    """_stream_expert_speech 首 chunk 超时 fallback 测试。
+
+    回归 commit 36b0296：该 commit 把 expert 发言从 gateway.chat() 改为
+    gateway.chat_stream() 以实现"逐个输出"UI 体验，但 DashScope/DeepSeek
+    via LiteLLM 偶尔不返回首个 SSE chunk，导致 board 整体 hang。
+    修复：首 chunk 30s 超时 → fallback 到非流式 chat() + _replay_stream。
+    """
+
+    @pytest.mark.asyncio
+    async def test_stream_normal_works(self):
+        """chat_stream 正常返回 chunks 时走流式路径。"""
+        team = BoardTeam()
+        orchestrator = BoardOrchestrator(team=team)
+        expert = _make_mock_expert("tester", is_lead=False)
+
+        # chat_stream 立刻返回 chunks
+        gateway = _make_mock_gateway("正常流式回复")
+        expert.agent._llm_gateway = gateway
+        orchestrator._get_llm_gateway = lambda e: gateway
+
+        with patch.object(orchestrator, "_broadcast_event", new_callable=AsyncMock):
+            result = await orchestrator._stream_expert_speech(expert, 1, "test prompt")
+
+        assert "正常流式回复" in result
+        # chat_stream 应被调用（流式路径）
+        gateway.chat_stream.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_stream_first_chunk_timeout_falls_back_to_chat(self):
+        """chat_stream 首 chunk 超时时 fallback 到 gateway.chat()。"""
+
+        class _HungStream:
+            """模拟 DashScope via LiteLLM 不返回首 chunk 的 hang。"""
+
+            async def __anext__(self):
+                # 永远不返回 — 触发 wait_for 超时
+                import asyncio as _asyncio
+
+                await _asyncio.Event().wait()  # 永久阻塞
+                raise StopAsyncIteration  # pragma: no cover
+
+            def __aiter__(self):
+                return self
+
+            async def aclose(self):
+                pass
+
+        team = BoardTeam()
+        # 把超时调小让测试快速通过
+        orchestrator = BoardOrchestrator(team=team)
+        orchestrator._STREAM_FIRST_CHUNK_TIMEOUT_S = 0.1
+        expert = _make_mock_expert("tester", is_lead=False)
+
+        gateway = _make_mock_gateway("fallback 内容")
+        # chat_stream 返回 hang 的 stream
+        gateway.chat_stream = MagicMock(return_value=_HungStream())
+        expert.agent._llm_gateway = gateway
+        orchestrator._get_llm_gateway = lambda e: gateway
+
+        with patch.object(orchestrator, "_broadcast_event", new_callable=AsyncMock):
+            result = await orchestrator._stream_expert_speech(expert, 1, "test prompt")
+
+        # 应该 fallback 到 chat() 并返回内容
+        assert "fallback 内容" in result
+        gateway.chat.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_stream_empty_falls_back_to_chat(self):
+        """chat_stream 立即结束（无 chunks）时 fallback 到 chat()。"""
+
+        async def _empty_stream():
+            return
+            yield  # pragma: no cover — make this an async generator
+
+        team = BoardTeam()
+        orchestrator = BoardOrchestrator(team=team)
+        expert = _make_mock_expert("tester", is_lead=False)
+
+        gateway = _make_mock_gateway("fallback 内容")
+        gateway.chat_stream = MagicMock(return_value=_empty_stream())
+        expert.agent._llm_gateway = gateway
+        orchestrator._get_llm_gateway = lambda e: gateway
+
+        with patch.object(orchestrator, "_broadcast_event", new_callable=AsyncMock):
+            result = await orchestrator._stream_expert_speech(expert, 1, "test prompt")
+
+        assert "fallback 内容" in result
+        gateway.chat.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_stream_chat_stream_raising_falls_back_to_chat(self):
+        """chat_stream 抛异常时 fallback 到 chat()。"""
+
+        def _raising_stream(*a, **kw):
+            raise RuntimeError("provider error")
+
+        team = BoardTeam()
+        orchestrator = BoardOrchestrator(team=team)
+        expert = _make_mock_expert("tester", is_lead=False)
+
+        gateway = _make_mock_gateway("fallback 内容")
+        gateway.chat_stream = MagicMock(side_effect=_raising_stream)
+        expert.agent._llm_gateway = gateway
+        orchestrator._get_llm_gateway = lambda e: gateway
+
+        with patch.object(orchestrator, "_broadcast_event", new_callable=AsyncMock):
+            result = await orchestrator._stream_expert_speech(expert, 1, "test prompt")
+
+        assert "fallback 内容" in result
+        gateway.chat.assert_called_once()