From cc4c24e57fe637f87fc8a2453cb4b13ee8701f4d Mon Sep 17 00:00:00 2001 From: huanghuoguoguo <60681390+huanghuoguoguo@users.noreply.github.com> Date: Wed, 15 Jul 2026 09:10:49 +0800 Subject: [PATCH] fix(agent-runner): unify streamed completion delivery --- docs/agent-runner-pluginization/STATUS.md | 5 ++- .../cases/acp-agent-runner-debug-chat.yaml | 2 +- .../pkg/pipeline/process/handlers/chat.py | 12 ++++++ tests/unit_tests/agent/test_chat_handler.py | 41 +++++++++++++++++++ 4 files changed, 57 insertions(+), 3 deletions(-) diff --git a/docs/agent-runner-pluginization/STATUS.md b/docs/agent-runner-pluginization/STATUS.md index 17a7f1f03..13692f1d8 100644 --- a/docs/agent-runner-pluginization/STATUS.md +++ b/docs/agent-runner-pluginization/STATUS.md @@ -38,14 +38,15 @@ | Runner | 状态 | 最近证据 | | --- | --- | --- | | `plugin:langbot-team/LocalAgent/default` | Unit-pass; Marketplace UI pass; Debug Chat E2E pass | 2026-07-12 隔离 first-run 实例从真实 AgentRunner catalog 安装 `langbot-team/LocalAgent` 0.1.0，Host 注册 `plugin:langbot-team/LocalAgent/default`，Wizard 自动选中并解锁后续操作。2026-07-15 `2026-07-15-08-44-10-770-08-00-sandbox-skill-authoring-edit-existing-e2e` 使用真实 `gpt-5.5` 完成 Skill 创建、注册、同 Query 激活、已激活包编辑与脚本执行；三阶段 UI、浏览器诊断和结构化文件系统检查全部通过，每阶段恰好新增一个 Bot 气泡，p95 14.6 秒、错误率 0。 | -| `plugin:langbot-team/ACPAgentRunner/default` / `plugin:langbot-team/ClaudeCodeAgent/default` / `plugin:langbot-team/CodexAgent/default` | Unit-pass; E2E pending | 通过 runner 仓库单测覆盖 session、run_id 注入和 LangBot MCP gateway；真实 harness E2E 取决于对应运行环境、CLI/daemon 可用性和 provider 登录态。 | +| `plugin:langbot-team/ACPAgentRunner/default` | Unit-pass; Debug Chat E2E pass | 2026-07-15 从本地 0.1.4 发布包安装并注册 PascalCase runner，remote-ssh Claude ACP 通过反向隧道调用 run-scoped `langbot_get_current_event`，97.8 秒返回可见结果；Host 将增量 delta 和 `message.completed` 聚合为一个完整 Bot 气泡。 | +| `plugin:langbot-team/ClaudeCodeAgent/default` / `plugin:langbot-team/CodexAgent/default` | Unit-pass; E2E pending | 通过 runner 仓库单测覆盖 session、run_id 注入和 LangBot MCP gateway；真实 harness E2E 取决于对应运行环境、CLI/daemon 可用性和 provider 登录态。 | | Dify / n8n / Coze / DashScope / Langflow / Tbox / DeerFlow / WeKnora | Unit-pass; credential smoke optional | 2026-06-13 plugin layout / parser tests 通过；真实服务凭据 smoke 非每轮必跑。 | ## Host / SDK 验收状态 | 范围 | 状态 | 最近证据 | | --- | --- | --- | -| LangBot Runtime Control Plane v2 foundation | Unit-pass; EBA release gate 5/5 pass | 2026-07-12 `eba-functional-20260712-release-gate-rerun` 通过 Quick Start 场景筛选、隔离实例 Runner Marketplace 安装、Runner 健康状态、事件路由 dry-run / 合成派发，以及真实 OneBot `group.member_joined` → Agent → `send_group_msg` 链路。 | +| LangBot Runtime Control Plane v2 foundation | Unit-pass; EBA release gate 5/5 pass; AgentRunner preflight pass | 2026-07-12 `eba-functional-20260712-release-gate-rerun` 通过 Quick Start 场景筛选、隔离实例 Runner Marketplace 安装、Runner 健康状态、事件路由 dry-run / 合成派发，以及真实 OneBot `group.member_joined` → Agent → `send_group_msg` 链路。2026-07-15 AgentRunner release preflight 16 项通过、0 warning，behavior matrix 覆盖 5 类结果并通过。 | | Host Skill / native tool integration | Unit-pass; WebUI E2E pass | 2026-07-15 provider / native / Skill / monitoring 定向测试 67 项通过，Pipeline / Chat / Wrapper 定向测试 61 项通过，Skills CLI 105 项通过；真实 Debug Chat 验证 `register_skill` 后同 Query `activate` 成功，监控工具调用不再把 SQL 行误取为字符串，结构化 JSON 文件检查不依赖格式空格，非流式多阶段 runner 结果只生成一个最终 Bot 气泡。 | | SDK AgentRunner control entities / proxy | Unit-pass | 2026-06-23 SDK `tests/api/entities/builtin/agent_runner`、`tests/api/proxies`、`tests/api/test_agent_tools_mcp_bridge.py`、`tests/runtime/plugin/test_mgr_agent_runner.py`、`tests/runtime/test_pull_api_handlers.py`、`tests/runtime/io/handlers/test_plugin_handler.py`、EBA event entities 和 message tests 通过，覆盖 typed entities、AgentRunAPIProxy、MCP bridge、runtime manager 与 pull API handlers。 | diff --git a/skills/skills/langbot-testing/cases/acp-agent-runner-debug-chat.yaml b/skills/skills/langbot-testing/cases/acp-agent-runner-debug-chat.yaml index 7a246267b..d62d4705a 100644 --- a/skills/skills/langbot-testing/cases/acp-agent-runner-debug-chat.yaml +++ b/skills/skills/langbot-testing/cases/acp-agent-runner-debug-chat.yaml @@ -35,7 +35,7 @@ automation_env: automation_pipeline_url_env: LANGBOT_ACP_AGENT_RUNNER_PIPELINE_URL automation_pipeline_name_env: LANGBOT_ACP_AGENT_RUNNER_PIPELINE_NAME automation_expected_runner_id: "plugin:langbot-team/ACPAgentRunner/default" -automation_prompt: "Use the injected LangBot MCP server tool langbot_get_current_event once. If the MCP call succeeds, reply with exactly ACP_AGENT_RUNNER_E2E_OK." +automation_prompt: "Do not launch any background agent, subagent, task, or worker. In this current ACP session, directly call the MCP tool named langbot_get_current_event exactly once and wait for its result. After it returns, reply exactly ACP_AGENT_RUNNER_E2E_OK with no other text." automation_expected_text: "ACP_AGENT_RUNNER_E2E_OK" automation_response_timeout_ms: "300000" setup_automation: diff --git a/src/langbot/pkg/pipeline/process/handlers/chat.py b/src/langbot/pkg/pipeline/process/handlers/chat.py index 7454a6cf0..834da6566 100644 --- a/src/langbot/pkg/pipeline/process/handlers/chat.py +++ b/src/langbot/pkg/pipeline/process/handlers/chat.py @@ -119,6 +119,18 @@ class ChatMessageHandler(handler.MessageHandler): # This replaces direct runner lookup and PluginAgentRunnerWrapper async for result in self.ap.agent_run_orchestrator.run_from_query(query): has_result = True + + if is_stream and isinstance(result, provider_message.MessageChunk): + if result.all_content is not None: + result = result.model_copy(update={'content': result.all_content}) + elif is_stream and isinstance(result, provider_message.Message): + result = provider_message.MessageChunk.model_validate( + { + **result.model_dump(), + 'is_final': True, + } + ) + result.resp_message_id = str(resp_message_id) # For streaming mode, pop previous response before adding new chunk diff --git a/tests/unit_tests/agent/test_chat_handler.py b/tests/unit_tests/agent/test_chat_handler.py index 62cc892df..d6ba5f386 100644 --- a/tests/unit_tests/agent/test_chat_handler.py +++ b/tests/unit_tests/agent/test_chat_handler.py @@ -383,6 +383,47 @@ class TestChatHandlerAsyncBehavior: assert len(query.resp_messages) == 1 assert query.resp_messages[0].content == 'Hello World!' + @pytest.mark.asyncio + async def test_streaming_renders_accumulated_deltas_and_completes_same_message(self): + """Incremental runner deltas and message.completed update one response stream.""" + from langbot.pkg.pipeline.process.handlers.chat import ChatMessageHandler + from langbot.pkg.pipeline import entities + from langbot_plugin.api.entities.builtin.provider import message as provider_message + + chunks = [ + provider_message.MessageChunk(role='assistant', content='Hel', all_content='Hel'), + provider_message.MessageChunk(role='assistant', content='lo', all_content='Hello'), + provider_message.Message(role='assistant', content='Hello'), + ] + orchestrator = MockAgentRunOrchestrator(chunks=chunks) + mock_ap = MockApplication(orchestrator=orchestrator) + mock_ap.plugin_connector.emit_event = AsyncMock(return_value=MockEventContext(prevented=False)) + query = MockQuery() + query.adapter.is_stream = True + handler = ChatMessageHandler(mock_ap) + mock_event = MagicMock() + mock_event.return_value = MagicMock() + + def make_result(*args, **kwargs): + return MagicMock(result_type=kwargs.get('result_type', entities.ResultType.CONTINUE)) + + observed = [] + with ( + patch('langbot.pkg.pipeline.process.handlers.chat.events') as mock_events_module, + patch('langbot.pkg.pipeline.entities.StageProcessResult', side_effect=make_result), + ): + mock_events_module.PersonNormalMessageReceived = mock_event + mock_events_module.GroupNormalMessageReceived = mock_event + + async for _ in handler.handle(query): + current = query.resp_messages[-1] + observed.append((current.content, current.is_final, current.resp_message_id)) + + assert [content for content, _, _ in observed] == ['Hel', 'Hello', 'Hello'] + assert [is_final for _, is_final, _ in observed] == [False, False, True] + assert len({response_id for _, _, response_id in observed}) == 1 + assert all(isinstance(message, provider_message.MessageChunk) for message in query.resp_messages) + @pytest.mark.asyncio async def test_non_streaming_keeps_results_and_yields_once(self): """Non-streaming mode keeps runner results but runs downstream stages once."""