Merge branch 'fix/private-board-llm' into main

Fixes 2 private board bugs:
1. board_orchestrator: chat_stream first-chunk hang (commit 36b0296 regression)
   - Add 30s first-chunk timeout to _stream_expert_speech
   - Fallback to non-streaming chat() + replay_stream on timeout
   - Fix mid-stream break double-broadcast (P2, review finding)
2. config_driven: _handle_llm_generate prefers _llm_gateway over legacy
   _llm_client (AgentPool only injects gateway)

3 commits, 4 files changed, 5 new tests.
E2E verified: 5-expert board completes in ~277s with DashScope.

src/agentkit/core/config_driven.py

@@ -1358,7 +1358,14 @@ class ConfigDrivenAgent(BaseAgent, EvolutionMixin):
         return gateway
 
     async def _handle_llm_generate(self, task: TaskMessage) -> dict:
-        """LLM 生成模式：渲染 Prompt → 调用 LLM → 解析输出"""
+        """LLM 生成模式：渲染 Prompt → 调用 LLM → 解析输出
+
+        ponytail: 优先使用 ``_llm_gateway``（v2 路径，``AgentPool`` 注入的），
+        仅在 gateway 缺失时回退到 ``_llm_client``（legacy 字段）。原实现只检查
+        ``_llm_client``，导致 ``AgentPool`` 创建的 expert/board agent 在
+        ``_llm_client is None`` 但 ``_llm_gateway`` 已注入时错误降级为
+        ``llm_generate_no_client``。
+        """
         if not self._prompt_template:
             raise ConfigValidationError(
                 agent_name=self.name,
@@ -1371,7 +1378,19 @@ class ConfigDrivenAgent(BaseAgent, EvolutionMixin):
         variables["task_type"] = task.task_type
         messages = self._prompt_template.render(variables=variables)
 
-        # 调用 LLM
+        # v2 路径：_llm_gateway 优先（AgentPool 仅注入 gateway，不设 _llm_client）
+        if self._llm_gateway is not None:
+            llm_params = self._config.llm.copy() if self._config.llm else {}
+            model = llm_params.get("model", "default")
+            response = await self._llm_gateway.chat(
+                messages=messages,
+                model=model,
+                agent_name=self.name,
+                task_type=task.task_type,
+            )
+            return self._parse_llm_response(response.content)
+
+        # 兼容旧路径：_llm_client（legacy）
         if self._llm_client is None:
             # 无 LLM 客户端时返回渲染后的 Prompt（降级模式）
             return {

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
-        raises before any chunk is produced.
+        unavailable (e.g. an LLM provider without streaming support), raises
+        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
-        # aligned with the streaming indicator).
+
+        # Try streaming with a hard first-chunk deadline. If the provider
+        # 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
-            async for chunk in gateway.chat_stream(
+            stream_obj = 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,12 +398,43 @@ 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
         # silent and then dumping the whole text in one frame.
+        #
+        # If we already broadcasted partial content from first_chunk before
+        # the stream broke, do NOT call gateway.chat() — that would broadcast
+        # the full content again and the UI would see duplicated text.
+        # Return the partial content instead; partial is better than duplicated.
+        if total:
+            logger.info(
+                f"Expert '{expert.config.name}' stream broke after {len(total)} chars "
+                f"of partial content — returning partial (avoid double broadcast)"
+            )
+            return total.strip()
         try:
             response = await gateway.chat(
                 messages=[{"role": "user", "content": prompt}],

tests/unit/experts/test_board_orchestrator.py

@@ -343,3 +343,165 @@ 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()
+
+    @pytest.mark.asyncio
+    async def test_stream_breaks_after_first_chunk_returns_partial(self):
+        """首 chunk 成功但后续断流时返回部分内容，不调 chat() 避免双重广播。"""
+
+        class _BreakAfterFirstStream:
+            """首个 chunk 正常返回，第二个 __anext__ 抛异常模拟断流。"""
+
+            def __init__(self):
+                self._first_yielded = False
+
+            async def __anext__(self):
+                if not self._first_yielded:
+                    self._first_yielded = True
+                    chunk = MagicMock()
+                    chunk.content = "部分内容"
+                    return chunk
+                raise RuntimeError("stream broke mid-way")
+
+            def __aiter__(self):
+                return self
+
+            async def aclose(self):
+                pass
+
+        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=_BreakAfterFirstStream())
+        expert.agent._llm_gateway = gateway
+        orchestrator._get_llm_gateway = lambda e: gateway
+
+        with patch.object(
+            orchestrator, "_broadcast_event", new_callable=AsyncMock
+        ) as mock_broadcast:
+            result = await orchestrator._stream_expert_speech(expert, 1, "test prompt")
+
+        # 返回部分内容，而不是完整 fallback 内容
+        assert "部分内容" in result
+        assert "完整 fallback" not in result
+        # 不应调 chat()（避免双重广播）
+        gateway.chat.assert_not_called()
+        # 只广播了 first_chunk 的部分内容
+        assert mock_broadcast.call_count == 1

tests/unit/test_config_driven.py

@@ -144,6 +144,52 @@ class TestConfigDrivenAgent:
         assert result["mode"] == "llm_generate_no_client"
         assert len(result["messages"]) == 2  # system + user
 
+    async def test_llm_generate_with_gateway_only(self):
+        """仅注入 LLMGateway（_llm_client=None）时应正常调用 LLM。
+
+        复现：AgentPool.create_agent() 只注入 llm_gateway，不设 _llm_client。
+        原实现错误降级为 llm_generate_no_client，导致私董会专家返回占位文本。
+        """
+
+        class _StubResponse:
+            def __init__(self, content: str) -> None:
+                self.content = content
+
+        class _StubGateway:
+            def __init__(self) -> None:
+                self.calls: list[dict] = []
+
+            async def chat(self, *, messages, model, agent_name, task_type, **kwargs):
+                self.calls.append(
+                    {
+                        "messages": messages,
+                        "model": model,
+                        "agent_name": agent_name,
+                        "task_type": task_type,
+                    }
+                )
+                return _StubResponse(json.dumps({"title": "FromGateway", "content": "OK"}))
+
+        gateway = _StubGateway()
+        config = AgentConfig.from_dict(_sample_llm_config())
+        agent = ConfigDrivenAgent(config=config, llm_gateway=gateway)
+
+        # 验证前置条件：_llm_client 必须为 None，模拟 AgentPool 注入路径
+        assert agent._llm_client is None
+        assert agent._llm_gateway is gateway
+
+        task = _make_task()
+        result = await agent.handle_task(task)
+
+        # 不应再走 no_client 降级
+        assert "mode" not in result or result.get("mode") != "llm_generate_no_client"
+        assert result["title"] == "FromGateway"
+        assert result["content"] == "OK"
+        # 验证 gateway.chat 被实际调用
+        assert len(gateway.calls) == 1
+        assert gateway.calls[0]["agent_name"] == config.name
+        assert gateway.calls[0]["model"] == "gpt-4"
+
     async def test_llm_generate_with_client(self):
         """有 LLM 客户端时调用 LLM 并解析结果"""