From 44c274f11a74a324f343a437d61682e2525cc96c Mon Sep 17 00:00:00 2001 From: Luhaozhu Date: Thu, 13 Aug 2026 04:45:12 -0400 Subject: [PATCH 1/2] feat(chat): add mid-run steering and stable history replay --- document/en/api/overview.md | 28 +- document/en/modules/chat.md | 33 ++- document/zh-CN/api/overview.md | 17 +- document/zh-CN/modules/chat.md | 19 +- src/backend/api/routes/v1/chat_runs.py | 68 ++++- src/backend/api/routes/v1/chats.py | 4 - src/backend/core/chat/tool_log.py | 48 +++- src/backend/core/llm/agent_factory.py | 21 +- src/backend/core/llm/chat_models.py | 109 +++++++- src/backend/core/llm/middlewares.py | 115 +++++++- .../core/services/chat_steer_service.py | 76 ++++++ .../orchestration/batch_orchestrator.py | 15 +- .../orchestration/chat_run_executor.py | 133 +++++++-- src/backend/orchestration/streaming.py | 154 +++++++---- src/backend/orchestration/workflow.py | 52 ++-- .../tests/chat/test_stream_event_builders.py | 31 ++- src/backend/tests/llm/test_openai_provider.py | 73 ++++- .../tests/orchestration/test_chat_steer.py | 199 ++++++++++++++ .../test_missing_tool_call_synthesis.py | 4 +- .../scripts/test-chat-stream-segments.ts | 116 ++++++++ src/frontend/src/App.tsx | 20 +- src/frontend/src/api.ts | 28 ++ src/frontend/src/components/chat/ChatArea.tsx | 8 +- .../src/components/chat/InputArea.tsx | 65 ++++- .../src/components/chat/QueuedMessageCard.tsx | 159 +++++++++++ src/frontend/src/components/chat/index.ts | 1 + .../src/components/tool/ToolCallRow.tsx | 6 +- .../src/components/tool/ToolRunShell.tsx | 40 +-- src/frontend/src/hooks/chatStream.ts | 114 +++++++- src/frontend/src/hooks/useChatInit.ts | 9 +- src/frontend/src/hooks/useStreaming.ts | 258 +++++++++++++++++- src/frontend/src/i18n/en/chat.ts | 14 + src/frontend/src/i18n/en/tool.ts | 1 - src/frontend/src/stores/chatStore.ts | 33 +++ src/frontend/src/styles/chat.css | 133 +++++++++ src/frontend/src/styles/tool.css | 17 +- src/frontend/src/types.ts | 7 +- src/frontend/src/utils/index.ts | 1 + src/frontend/src/utils/segments.ts | 113 +++----- src/frontend/src/utils/toolRunState.ts | 7 + 40 files changed, 2021 insertions(+), 328 deletions(-) create mode 100644 src/backend/core/services/chat_steer_service.py create mode 100644 src/backend/tests/orchestration/test_chat_steer.py create mode 100644 src/frontend/src/components/chat/QueuedMessageCard.tsx create mode 100644 src/frontend/src/utils/toolRunState.ts diff --git a/document/en/api/overview.md b/document/en/api/overview.md index 50b11e92..f790295d 100644 --- a/document/en/api/overview.md +++ b/document/en/api/overview.md @@ -1,6 +1,6 @@ # API Overview -> Last updated: 2026-07-28 +> Last updated: August 12, 2026 The HugAgentOS backend is a FastAPI application; all business endpoints live under the `/v1/*` prefix. In a production deployment, Nginx strips the `/api/` prefix before forwarding to the backend (see `src/frontend/default.conf.template`), so **the full browser-facing path is `/api/v1/...`**, while hitting the backend container directly uses `/v1/...`. Examples in this document use the local development address `http://localhost:3000/api`. @@ -112,7 +112,8 @@ The request body is a `ChatRequest` (`src/backend/api/schemas.py`): `chat_id` an | `tool_call_start` | The model starts constructing a tool call | `tool_name`, `tool_display_name`, `tool_id` | | `tool_call_delta` | Incremental tool-argument JSON (batched by the backend) | `tool_name`, `tool_id`, `arguments_delta` | | `tool_call` | Arguments are complete and execution is about to start | `tool_name`, `tool_display_name`, `tool_args`, `tool_id`, `subagent_name?` | -| `tool_result` | Tool returns | `tool_name`, `result` (JSON), `tool_id`, `citations` (citation items) | +| `tool_result` | Tool returns | `tool_name`, `result` (JSON), `tool_id`, `status`, `citations` (citation items) | +| `steer_applied` | A follow-up entered the context at a safe ReAct boundary | `steer_id`, `message`, `message_id`, `chat_id` | | `tool_pending` | Waiting fallback when the provider exposes no parseable deltas | `reason` (e.g. `llm_buffering`) | | `file_confirm` | A tool is suspended awaiting user confirmation of a "My Space" write | `confirm_id`, `op`, `logical_path`, `message`, `expired`; the stream stays open — the user confirms out-of-band via `POST /v1/chats/{chat_id}/file-confirm` and the tool resumes | | `batch_confirm` | A batch-execution plan awaits user confirmation | `plan_id`, `total`, `preview`, `default_template`, `placeholder_keys`; confirm via `POST /v1/batch/{plan_id}/confirm` | @@ -145,7 +146,7 @@ data: [DONE] `[ref:tool_name-N]` markers in the answer text are parsed into `citations` items by `orchestration/citations.py`; the frontend renders them as citation badges (see [Chat module](../modules/chat.md)). -### Resume and cancel +### Resume, Steer, and cancel The answer is generated by a background run; the SSE connection merely *follows* it — disconnecting does not stop generation: @@ -157,9 +158,26 @@ curl -N "http://localhost:3000/api/v1/chats/stream/run_9f8e7d?from_offset=0" \ # Cancel generation curl -X POST http://localhost:3000/api/v1/chat-runs/run_9f8e7d/cancel \ -H "Authorization: Bearer sk-jx-xxxxxxxx" + +# Add a plain-text instruction at the next safe ReAct boundary +curl -X POST http://localhost:3000/api/v1/chat-runs/run_9f8e7d/steer \ + -H "Authorization: Bearer sk-jx-xxxxxxxx" \ + -H "Content-Type: application/json" \ + -d '{"steer_id":"steer_001","message":"Do not download it. Compare both options instead."}' ``` -Resume validates run ownership: a non-owner gets 403, a missing run gets 404. `GET /v1/chats/{chat_id}/active-run` reports whether a chat currently has a run in progress (the frontend uses this to reconnect after a page refresh). A run silent for longer than `CHAT_RUN_INACTIVITY_TIMEOUT_SEC` (default 600 s) is considered dead and terminated. +Resume, Steer, and cancel validate run ownership: a non-owner gets 403, and a +missing run gets 404. Steer accepts only an active regular-chat run. Redis +hands the instruction to the worker. If a tool is running, the worker injects +the instruction after that tool result enters the context and before the next +model call. If the next tool batch hasn't started, the worker interrupts the +old call before injecting the instruction. Both paths confirm delivery with a +`steer_applied` event. `DELETE /v1/chat-runs/{run_id}/steer/{steer_id}` +withdraws an instruction that hasn't been consumed. +`GET /v1/chats/{chat_id}/active-run` reports whether a chat currently has a run +in progress, which the frontend uses to reconnect after a page refresh. A run +silent for longer than `CHAT_RUN_INACTIVITY_TIMEOUT_SEC` (default 600 seconds) +is considered dead and terminated. ### Other SSE endpoints @@ -186,7 +204,7 @@ Auth column legend: "User" = session cookie or personal API key (`get_current_us | Group | Module (`api/routes/v1/`) | Prefix | Representative endpoints | Auth | |---|---|---|---|---| | Chat & messages | `chats.py` | `/v1/chats` | `POST /stream` (SSE), `GET /stream/{run_id}` (resume), `POST /send` (non-streaming), `GET /`, `GET /{chat_id}/messages`, `POST /{chat_id}/share` | User | -| Chat & messages | `chat_runs.py` | `/v1/chat-runs` | `POST /{run_id}/cancel` | User | +| Chat & messages | `chat_runs.py` | `/v1/chat-runs` | `POST /{run_id}/cancel`, `POST /{run_id}/steer`, `DELETE /{run_id}/steer/{steer_id}` | User | | Chat & messages | `chat_shares.py` | `/v1/chat-shares` | `POST /`, `GET /{share_id}`, `POST /{share_id}/revoke` | User | | Chat & messages | `summary.py` | `/v1/summary` | `POST /` (chat title summarization) | User | | Chat & messages | `classify.py` | `/v1/classify` | `POST /` (business-topic classification) | User | diff --git a/document/en/modules/chat.md b/document/en/modules/chat.md index ffbf542e..414497e9 100644 --- a/document/en/modules/chat.md +++ b/document/en/modules/chat.md @@ -42,9 +42,29 @@ Every sent message creates a `ChatRun` and a background task (`orchestration/cha | Resume after refresh / disconnect | `GET /v1/chats/stream/{run_id}?from_offset=N` | | Probe for an in-flight run | `GET /v1/chats/{chat_id}/active-run` | | Cancel a run (kills the background task) | `POST /v1/chat-runs/{run_id}/cancel` | +| Add an instruction at the next safe ReAct boundary | `POST /v1/chat-runs/{run_id}/steer` | +| Withdraw an instruction that hasn't taken effect | `DELETE /v1/chat-runs/{run_id}/steer/{steer_id}` | Defensive machinery: a `: heartbeat` SSE comment line every 15 silent seconds (keeps nginx `proxy_read_timeout` and other proxies from cutting the stream); an inactivity watchdog fails the run if the workflow produces no chunk for 600 s (`CHAT_RUN_INACTIVITY_TIMEOUT_SEC`); a periodic reaper collects over-age running runs; `recover_orphan_runs()` cleans up leftovers at startup. +### Mid-run follow-ups, Steer, and the stop shortcut + +While a regular chat is generating, the composer continues accepting the next +message. Sending it creates a queued card above the composer. You can edit the +card from its more menu or delete it. If you don't select **Steer**, the client +sends the message as the next turn after the current answer finishes. + +When you select **Steer**, Redis hands the plain-text instruction to the active +run. If the instruction arrives while a tool is running, `SteerMiddleware` +atomically consumes it after that tool result enters the context and before the +next model call, appends the real user message, and lets the model replan. If +the instruction arrives earlier, the middleware interrupts the old tool call +before it starts and enters the same replanning flow. A `steer_applied` SSE +event confirms delivery. Messages containing attachments, skills, plugins, or +sub-agents wait and send normally after the current answer. Pressing `Esc` +cancels the run for the chat visible on the current page. If a card editor or +dialog already consumes `Esc`, it doesn't stop the run. + ### Agent construction highlights (core/llm/agent_factory.py) `create_agent_executor()` is the shared factory for every mode (main chat, plan, batch, sub-agents, automation): @@ -52,7 +72,7 @@ Defensive machinery: a `: heartbeat` SSE comment line every 15 silent seconds (k - **MCP tools**: after the three-layer filter of catalog + per-user overrides + request context (see [Capability Center](catalog.md)), stable servers reuse the process-level connection pool (`core/llm/mcp_pool.py`); per-request servers (e.g. `retrieve_dataset_content`, which needs per-request HTTP headers) are spawned fresh; the user's self-added private MCP servers are merged in with owner isolation. - **Skills**: registered as AgentScope Agent Skills via `core/agent_skills/loader.py`, with `view_text_file` allow-listed to read SKILL.md files (see [Agent Skills](agent-skills.md)). - **File / sandbox tools**: `bash`, `sandbox_put_artifact`, `sandbox_get_artifact` are always registered; Read/Edit/Write/Glob/Grep/Delete/Move/mkdir plus the MySpace tools are gated by `CODE_CAPABILITY_ENABLED` and share one `ReadStateTracker` to keep the "must Read before Edit" invariant. -- **Middlewares** (onion model, `core/llm/middlewares.py`): `DynamicModelMiddleware` (switches the model per chat_mode, see [Model Providers](model-providers.md)), `FileContextMiddleware` (injects uploaded/historical file context), `WorkspacePinHintMiddleware`, `GoalAnchorReminderMiddleware`, `FinishPinGuardMiddleware`. +- **Middlewares** (onion model, `core/llm/middlewares.py`): `DynamicModelMiddleware` (switches the model per chat_mode, see [Model Providers](model-providers.md)), `FileContextMiddleware` (injects uploaded/historical file context), `SteerMiddleware` (injects follow-ups after tool results and before the next reasoning round), `WorkspacePinHintMiddleware`, `GoalAnchorReminderMiddleware`, `FinishPinGuardMiddleware`. - **Context compression**: `ContextConfig(trigger_ratio=0.6, tool_result_limit=20000)` plus a structured Chinese compression prompt designed to produce a *resumable ReAct workflow* summary; if the compression call itself fails, `JxOpenAIChatModel.generate_structured_output` returns an L3 synthetic summary so the reply never crashes. - **Permissions**: every registered tool gets a native `PermissionRule(ALLOW)` seed, preserving AgentScope's built-in dangerous-operation checks (no blanket BYPASS). - **Iteration caps**: main agent defaults to `max_iters=50`, isolated sub-agents to 10. @@ -69,7 +89,8 @@ Defensive machinery: a `: heartbeat` SSE comment line every 15 silent seconds (k | `tool_call_start` | Tool-call construction starts; the frontend opens one card by stable ID | `tool_name`, `tool_display_name`, `tool_id` | | `tool_call_delta` | Incremental argument JSON appended to the same card | `tool_name`, `tool_id`, `arguments_delta` | | `tool_call` | Arguments are complete and execution is about to start | `tool_name`, `tool_display_name`, `tool_args`, `tool_id`, plus `subagent_name` for sub-agent calls | -| `tool_result` | Tool invocation result | `tool_name`, `result`, `tool_id`, `citations[]` | +| `tool_result` | Tool invocation result | `tool_name`, `result`, `tool_id`, `status`, `citations[]` | +| `steer_applied` | A mid-run instruction entered the ReAct context | `steer_id`, `message`, `message_id`, `chat_id` | | `subagent_event` | Child execution details nested under the parent `call_subagent` card | `parent_tool_id`, `sub_type`, `agent_name`, plus child tool or content fields | | `ontology_activation` / `ontology_gate` / `ontology_review` | Ontology-governance state, separate from model reasoning | workflow activation, gate decision, and committee status or verdict | | `tool_pending` | Waiting fallback when the provider exposes no parseable argument deltas | `reason` | @@ -112,6 +133,12 @@ event, the frontend replaces the body in place, and the database stores only the reviewed final answer. It persists `ontology_governance` with the assistant message so the module remains available after a history refresh. +History replay recognizes both reasoning protocols. Inline-reasoning models may +emit `reasoningbody`, while the backend normalizes a structured reasoning +field to `reasoningbody`. The frontend places reasoning and tool +calls in one process area and renders all visible body text as one continuous +Markdown block. History rendering doesn't split the body at character offsets. + ## Citation system (Evidence Anchors) Citations make every fact in the answer traceable back to a specific tool result. Numbering authority belongs to a single backend source of truth — the model only **copies** ids, never computes them. The chain has four segments: @@ -280,4 +307,4 @@ The same orchestration foundation also powers: response regeneration (`POST /v1/ | Oversized-result offloading | `src/backend/core/llm/offloader.py` | | Chat sharing | `src/backend/api/routes/v1/chat_shares.py` | | Follow-up generation | `src/backend/orchestration/followups.py` | -| Frontend stream parsing | `src/frontend/src/hooks/chatStream.ts` | +| Frontend stream parsing / follow-up queue | `src/frontend/src/hooks/chatStream.ts`, `useStreaming.ts`, `components/chat/QueuedMessageCard.tsx` | diff --git a/document/zh-CN/api/overview.md b/document/zh-CN/api/overview.md index 981f2608..3c0de222 100644 --- a/document/zh-CN/api/overview.md +++ b/document/zh-CN/api/overview.md @@ -1,6 +1,6 @@ # API 总览 -> 最后更新：2026-07-28 +> 最后更新：2026-08-12 HugAgentOS 后端是一个 FastAPI 应用，所有业务接口挂在 `/v1/*` 前缀下。生产部署中 Nginx 把 `/api/` 前缀剥掉后转发给后端（见 `src/frontend/default.conf.template`），因此**浏览器侧的完整路径是 `/api/v1/...`**，直接访问后端容器则是 `/v1/...`。本文示例统一使用本地开发地址 `http://localhost:3000/api`。 @@ -112,7 +112,8 @@ curl -N http://localhost:3000/api/v1/chats/stream \ | `tool_call_start` | 模型开始构造工具调用 | `tool_name`、`tool_display_name`、`tool_id` | | `tool_call_delta` | 工具参数 JSON 增量（后端合批后下发） | `tool_name`、`tool_id`、`arguments_delta` | | `tool_call` | 参数完整、即将执行工具 | `tool_name`、`tool_display_name`、`tool_args`、`tool_id`、`subagent_name?` | -| `tool_result` | 工具返回结果 | `tool_name`、`result`（JSON）、`tool_id`、`citations`（引用项列表） | +| `tool_result` | 工具返回结果 | `tool_name`、`result`（JSON）、`tool_id`、`status`、`citations`（引用项列表） | +| `steer_applied` | 追加指令已在安全 ReAct 边界注入 | `steer_id`、`message`、`message_id`、`chat_id` | | `tool_pending` | 提供商未暴露可解析增量时的等待兜底 | `reason`（如 `llm_buffering`） | | `file_confirm` | 工具挂起等待用户确认「我的空间」写操作 | `confirm_id`、`op`、`logical_path`、`message`、`expired`；流不结束，用户带外 `POST /v1/chats/{chat_id}/file-confirm` 后续跑 | | `batch_confirm` | 批量执行计划等待用户确认 | `plan_id`、`total`、`preview`、`default_template`、`placeholder_keys`；确认走 `POST /v1/batch/{plan_id}/confirm` | @@ -145,7 +146,7 @@ data: [DONE] 正文中的 `[ref:tool_name-N]` 引用标记由 `orchestration/citations.py` 解析为 `citations` 项，前端据此渲染角标（见 [对话模块](../modules/chat.md)）。 -### 续播与取消 +### 续播、Steer 与取消 回答在后台 run 中执行，SSE 只是「跟随」——断开连接不会终止生成： @@ -157,9 +158,15 @@ curl -N "http://localhost:3000/api/v1/chats/stream/run_9f8e7d?from_offset=0" \ # 主动取消生成 curl -X POST http://localhost:3000/api/v1/chat-runs/run_9f8e7d/cancel \ -H "Authorization: Bearer sk-jx-xxxxxxxx" + +# 在下一次安全 ReAct 边界追加纯文本指令 +curl -X POST http://localhost:3000/api/v1/chat-runs/run_9f8e7d/steer \ + -H "Authorization: Bearer sk-jx-xxxxxxxx" \ + -H "Content-Type: application/json" \ + -d '{"steer_id":"steer_001","message":"先不要下载，改为比较两个方案"}' ``` -续播校验 run 归属：非属主返回 403、run 不存在返回 404。`GET /v1/chats/{chat_id}/active-run` 可查询某会话当前是否有进行中的 run（前端刷新页面后据此重连）。run 静默超过 `CHAT_RUN_INACTIVITY_TIMEOUT_SEC`（默认 600 秒）会被判定为僵死并终止。 +续播、Steer 和取消都会校验 run 归属：非属主返回 403、run 不存在返回 404。Steer 只支持进行中的普通对话 run；指令通过 Redis 交给执行进程。若工具正在执行，指令会在该工具结果进入上下文后、下一轮模型推理前注入；若下一批工具尚未开始，则先中止旧调用再注入。两种路径都以 `steer_applied` 事件确认。`DELETE /v1/chat-runs/{run_id}/steer/{steer_id}` 可撤回尚未被消费的指令。`GET /v1/chats/{chat_id}/active-run` 可查询某会话当前是否有进行中的 run（前端刷新页面后据此重连）。run 静默超过 `CHAT_RUN_INACTIVITY_TIMEOUT_SEC`（默认 600 秒）会被判定为僵死并终止。 ### 其他 SSE 端点 @@ -186,7 +193,7 @@ curl -X POST http://localhost:3000/api/v1/chat-runs/run_9f8e7d/cancel \ | 分组 | 模块（`api/routes/v1/`） | 前缀 | 代表端点 | 鉴权 | |---|---|---|---|---| | 会话与消息 | `chats.py` | `/v1/chats` | `POST /stream`（SSE）、`GET /stream/{run_id}`（续播）、`POST /send`（非流式）、`GET /`、`GET /{chat_id}/messages`、`POST /{chat_id}/share` | 用户 | -| 会话与消息 | `chat_runs.py` | `/v1/chat-runs` | `POST /{run_id}/cancel` | 用户 | +| 会话与消息 | `chat_runs.py` | `/v1/chat-runs` | `POST /{run_id}/cancel`、`POST /{run_id}/steer`、`DELETE /{run_id}/steer/{steer_id}` | 用户 | | 会话与消息 | `chat_shares.py` | `/v1/chat-shares` | `POST /`、`GET /{share_id}`、`POST /{share_id}/revoke` | 用户 | | 会话与消息 | `summary.py` | `/v1/summary` | `POST /`（会话标题摘要） | 用户 | | 会话与消息 | `classify.py` | `/v1/classify` | `POST /`（业务主题分类） | 用户 | diff --git a/document/zh-CN/modules/chat.md b/document/zh-CN/modules/chat.md index 2056fbb9..41851dac 100644 --- a/document/zh-CN/modules/chat.md +++ b/document/zh-CN/modules/chat.md @@ -42,9 +42,17 @@ SSE follower：chat_run_executor.follow_run_as_sse() | 刷新/断线后续播 | `GET /v1/chats/stream/{run_id}?from_offset=N` | | 探测会话进行中的 run | `GET /v1/chats/{chat_id}/active-run` | | 取消 run（真正杀后台任务） | `POST /v1/chat-runs/{run_id}/cancel` | +| 在下一次安全 ReAct 边界追加指令 | `POST /v1/chat-runs/{run_id}/steer` | +| 撤回尚未生效的追加指令 | `DELETE /v1/chat-runs/{run_id}/steer/{steer_id}` | 防御机制：静默 15 秒写一行 `: heartbeat` SSE 注释（防 nginx `proxy_read_timeout` 掐流）；workflow 600 秒无任何 chunk 触发看门狗判 failed（`CHAT_RUN_INACTIVITY_TIMEOUT_SEC`）；周期 reaper 把超龄 running run 收成 failed；启动钩子 `recover_orphan_runs()` 清理重启遗留。 +### 运行中追加、Steer 与快捷停止 + +普通对话正在生成时，输入框仍可接收下一条消息。发送后，消息先显示在输入框上方的待发送卡片中；用户可通过更多菜单编辑，也可删除。若不执行 **Steer**，当前回答结束后会自动把这条消息作为下一轮发送。 + +选择 **Steer** 后，后端通过 Redis 把纯文本指令交给当前 run。若指令在工具执行期间到达，`SteerMiddleware` 会在该轮工具结果进入上下文后、下一轮模型推理前原子消费并插入真实用户消息，让模型立即重新规划；若指令更早到达，则在下一批工具开始前中止尚未执行的旧工具调用，再进入同一重规划流程。`steer_applied` SSE 事件确认指令已生效。包含附件、技能、插件或子智能体的消息不会走中途注入，而是在当前回答结束后正常发送。按 `Esc` 会取消当前页面正在显示的会话 run；编辑卡片或弹窗已消费 `Esc` 时，不会误停任务。 + ### Agent 构建要点（core/llm/agent_factory.py） `create_agent_executor()` 是所有模式（主对话、计划、批量、子智能体、自动化）共用的工厂： @@ -52,7 +60,7 @@ SSE follower：chat_run_executor.follow_run_as_sse() - **MCP 工具**：经 catalog + 用户覆盖 + 请求上下文三层过滤后（见 [能力目录](catalog.md)），stable 服务复用进程级连接池（`core/llm/mcp_pool.py`），per-request 服务（如 `retrieve_dataset_content` 需带每请求 HTTP header）每次新建；用户自助添加的私有 MCP 按 owner 现查合入。 - **技能**：经 `core/agent_skills/loader.py` 注册为 AgentScope Agent Skills，并放行 `view_text_file` 读取 SKILL.md（详见 [技能系统](agent-skills.md)）。 - **文件/沙箱工具**：`bash`、`sandbox_put_artifact`、`sandbox_get_artifact` 无条件注册；Read/Edit/Write/Glob/Grep/Delete/Move/mkdir + MySpace 工具受 `CODE_CAPABILITY_ENABLED` 门控，共享同一个 `ReadStateTracker` 维持「先 Read 才能 Edit」不变量。 -- **中间件**（洋葱模型，`core/llm/middlewares.py`）：`DynamicModelMiddleware`（按 chat_mode 切模型，见 [模型接入](model-providers.md)）、`FileContextMiddleware`（注入上传/历史文件上下文）、`WorkspacePinHintMiddleware`、`GoalAnchorReminderMiddleware`、`FinishPinGuardMiddleware`。 +- **中间件**（洋葱模型，`core/llm/middlewares.py`）：`DynamicModelMiddleware`（按 chat_mode 切模型，见 [模型接入](model-providers.md)）、`FileContextMiddleware`（注入上传/历史文件上下文）、`SteerMiddleware`（工具结果之后、下一轮推理之前注入追加指令）、`WorkspacePinHintMiddleware`、`GoalAnchorReminderMiddleware`、`FinishPinGuardMiddleware`。 - **上下文压缩**：`ContextConfig(trigger_ratio=0.6, tool_result_limit=20000)` + 结构化中文「可恢复 ReAct 工作流」压缩提示词；压缩调用失败时由 `JxOpenAIChatModel.generate_structured_output` 返回 L3 占位摘要兜底。 - **权限**：所有已注册工具 seed 原生 `PermissionRule(ALLOW)`，保留 AgentScope 内置工具的危险操作检查（不使用一刀切 BYPASS）。 - **迭代上限**：主智能体默认 `max_iters=50`，隔离子智能体默认 10。 @@ -69,7 +77,8 @@ SSE follower：chat_run_executor.follow_run_as_sse() | `tool_call_start` | 开始构造一次工具调用；前端按稳定 ID 创建一张卡片 | `tool_name`, `tool_display_name`, `tool_id` | | `tool_call_delta` | 工具参数 JSON 增量；前端在同一卡片中追加 | `tool_name`, `tool_id`, `arguments_delta` | | `tool_call` | 工具参数已完整、即将执行 | `tool_name`, `tool_display_name`, `tool_args`, `tool_id`，调子智能体时附 `subagent_name` | -| `tool_result` | 工具调用结果 | `tool_name`, `result`, `tool_id`, `citations[]` | +| `tool_result` | 工具调用结果 | `tool_name`, `result`, `tool_id`, `status`, `citations[]` | +| `steer_applied` | 运行中追加指令已注入 ReAct 上下文 | `steer_id`, `message`, `message_id`, `chat_id` | | `subagent_event` | 子智能体内部过程，挂在父 `call_subagent` 卡片下 | `parent_tool_id`, `sub_type`, `agent_name`，以及内部工具或内容字段 | | `ontology_activation` / `ontology_gate` / `ontology_review` | 本体治理状态，不属于模型思考 | 工作流、门禁决策、委员会状态与结论 | | `tool_pending` | 提供商没有暴露可解析参数增量时的等待兜底 | `reason` | @@ -102,6 +111,10 @@ data: [DONE] `meta` 之后，`chat_run_executor.py` 持久化助手消息、回填 artifact，并起后台任务生成追问问题（`orchestration/followups.py`，结果写进消息 `extra_data.follow_up_questions`，前端经 `GET /v1/chats/{chat_id}/messages/{message_id}/followups` 拉取）。本体事件在前端汇总为独立的“领域本体治理”模块，不再写入或显示在“思考过程”中。模型草稿保持逐 token 流式展示；委员会仅在实际修订答案时发送一次 `content_replace`，前端原位替换正文，数据库只保存评审后的最终答案。`ontology_governance` 随助手消息持久化，刷新历史会话后仍可回显。 +历史回放先识别两种思考协议：内联思考模型可能只输出 +`reasoning正文`，结构化 reasoning 字段则由后端归一化为 +`reasoning正文`。前端把思考与工具调用放入统一过程区，并把全部可见正文合并为一个连续的 Markdown 块；历史渲染不再按正文字符位置切分。 + ## 引用系统（Citations · 证据锚点） 引用让回答里的每个事实可溯源到具体工具结果。编号权收归后端唯一真源——模型只**复制**编号、不做任何计算，链路分四段： @@ -219,4 +232,4 @@ data: [DONE] | 超长结果 offload | `src/backend/core/llm/offloader.py` | | 会话分享 | `src/backend/api/routes/v1/chat_shares.py` | | 追问生成 | `src/backend/orchestration/followups.py` | -| 前端流式解析 | `src/frontend/src/hooks/chatStream.ts` | +| 前端流式解析 / 追加消息 | `src/frontend/src/hooks/chatStream.ts`，`useStreaming.ts`，`components/chat/QueuedMessageCard.tsx` | diff --git a/src/backend/api/routes/v1/chat_runs.py b/src/backend/api/routes/v1/chat_runs.py index dc724e54..77d79da3 100644 --- a/src/backend/api/routes/v1/chat_runs.py +++ b/src/backend/api/routes/v1/chat_runs.py @@ -1,17 +1,31 @@ -"""Chat Run management API — currently only exposes cancel; list/detail are not needed for now.""" +"""Chat Run management API — cancel and mid-run steering.""" -from fastapi import APIRouter, Depends, HTTPException - -from core.auth.backend import get_current_user, UserContext +from core.auth.backend import UserContext, get_current_user from core.infra.logging import get_logger from core.infra.responses import success_response +from core.services.chat_steer_service import put_pending_steer, remove_pending_steer +from fastapi import APIRouter, Depends, HTTPException from orchestration import chat_run_executor +from pydantic import BaseModel, Field, field_validator logger = get_logger(__name__) router = APIRouter(prefix="/v1/chat-runs", tags=["ChatRuns"]) +class SteerChatRunRequest(BaseModel): + steer_id: str = Field(..., min_length=1, max_length=64, description="前端待发送卡片 ID") + message: str = Field(..., min_length=1, max_length=10000, description="追加给当前 run 的指令") + + @field_validator("message") + @classmethod + def validate_message(cls, value: str) -> str: + value = value.strip() + if not value: + raise ValueError("message cannot be empty") + return value + + @router.post("/{run_id}/cancel", summary="取消正在执行的 run（真正杀掉后台任务）") async def cancel_chat_run( run_id: str, @@ -25,3 +39,49 @@ async def cancel_chat_run( except chat_run_executor.ChatRunPermissionDenied: raise HTTPException(status_code=403, detail="无权取消该 run") return success_response(data={"run_id": run_id, "cancelled": cancelled}) + + +@router.post("/{run_id}/steer", summary="在下一次安全 ReAct 边界追加指令") +async def steer_chat_run( + run_id: str, + body: SteerChatRunRequest, + user: UserContext = Depends(get_current_user), +): + """Queue one instruction for the live ReAct loop's next safe boundary.""" + run = chat_run_executor.get_run(run_id) + if run is None: + raise HTTPException(status_code=404, detail="run not found") + if run.user_id != user.user_id: + raise HTTPException(status_code=403, detail="无权操作该 run") + if run.status not in ("pending", "running"): + raise HTTPException(status_code=409, detail="run 已结束") + payload = run.request_payload if isinstance(run.request_payload, dict) else {} + if payload.get("kind", "chat") != "chat": + raise HTTPException(status_code=409, detail="当前运行模式不支持 Steer") + + message = body.message.strip() + await put_pending_steer( + run_id, + { + "steer_id": body.steer_id, + "message": message, + "run_id": run_id, + "chat_id": run.chat_id, + }, + ) + return success_response(data={"run_id": run_id, "steer_id": body.steer_id, "queued": True}) + + +@router.delete("/{run_id}/steer/{steer_id}", summary="撤回尚未生效的追加指令") +async def withdraw_chat_run_steer( + run_id: str, + steer_id: str, + user: UserContext = Depends(get_current_user), +): + run = chat_run_executor.get_run(run_id) + if run is None: + raise HTTPException(status_code=404, detail="run not found") + if run.user_id != user.user_id: + raise HTTPException(status_code=403, detail="无权操作该 run") + removed = await remove_pending_steer(run_id, steer_id) + return success_response(data={"run_id": run_id, "steer_id": steer_id, "removed": removed}) diff --git a/src/backend/api/routes/v1/chats.py b/src/backend/api/routes/v1/chats.py index 07e729f2..f4fb1761 100644 --- a/src/backend/api/routes/v1/chats.py +++ b/src/backend/api/routes/v1/chats.py @@ -1348,10 +1348,6 @@ def _flush_thinking() -> None: elif chunk_type == "tool_call": _flush_thinking() _tc_evt = build_tool_call_event(chunk, chat_id, tool_calls_log) - # 记录该工具卡片出现时正文的累计长度：历史重建按此偏移把 - # 「文本 ↔ 工具卡片」按流式原顺序交错（问题15：刷新后内容与实时不一致）。 - for _tc in tool_calls_log: - _tc.setdefault("content_offset", len(full_response)) yield f"data: {json.dumps(_tc_evt, ensure_ascii=False)}\n\n" elif chunk_type == "tool_call_start": _flush_thinking() diff --git a/src/backend/core/chat/tool_log.py b/src/backend/core/chat/tool_log.py index 8ee97d70..218a9fcc 100644 --- a/src/backend/core/chat/tool_log.py +++ b/src/backend/core/chat/tool_log.py @@ -88,6 +88,17 @@ def build_tool_result_event(chunk: dict, chat_id: str, tool_calls_log: list) -> tid = chunk.get("tool_id") tn = chunk.get("tool_name") res = chunk.get("result", {}) + raw_status = str(chunk.get("status") or "success").lower() + # A steer interrupts the pending tool *before* execution so the model can + # re-plan with the user's new instruction. It is a normal control-flow + # boundary, not a failed tool call: preserve the distinct status for audit + # and let the UI render it neutrally instead of showing a red error cross. + if raw_status == "interrupted": + status = "interrupted" + elif raw_status in {"error", "denied"}: + status = "error" + else: + status = "success" evt: Dict[str, Any] = { "type": "tool_result", "tool_name": tn, @@ -95,12 +106,13 @@ def build_tool_result_event(chunk: dict, chat_id: str, tool_calls_log: list) -> "tool_id": tid, "chat_id": chat_id, "citations": chunk.get("citations", []), + "status": status, } if chunk.get("subagent_name"): evt["subagent_name"] = chunk["subagent_name"] if chunk.get("scope"): evt["scope"] = chunk["scope"] - attach_tool_result(tool_calls_log, tid, tn, res) + attach_tool_result(tool_calls_log, tid, tn, res, status=status) return evt @@ -118,22 +130,29 @@ def upsert_tool_call(tool_calls_log: list, tc: dict) -> None: tool_calls_log.append(tc) -def attach_tool_result(tool_calls_log: list, tid: str, tn: str, res: Any) -> None: +def attach_tool_result( + tool_calls_log: list, + tid: str, + tn: str, + res: Any, + *, + status: str = "success", +) -> None: """Attach a tool_result to the matching tool_call entry in the log.""" for tc in tool_calls_log: if tid and tc.get("tool_id") == tid: - tc["result"], tc["status"] = res, "success" + tc["result"], tc["status"] = res, status return if tn and tc.get("tool_name") == tn and "result" not in tc: - tc["result"], tc["status"] = res, "success" + tc["result"], tc["status"] = res, status return if tid or tn: - tool_calls_log.append({"tool_name": tn, "tool_id": tid, "result": res, "status": "success"}) + tool_calls_log.append({"tool_name": tn, "tool_id": tid, "result": res, "status": status}) # Persistence caps for sub-agent sub-steps (prevent a single call_subagent's sub_steps from growing unbounded and bloating the message row). -_SUBSTEP_OUTPUT_CAP = 16000 # max characters stored per sub-tool result -_SUBSTEP_MAX_STEPS = 200 # max sub-steps stored per call_subagent card +_SUBSTEP_OUTPUT_CAP = 16000 # max characters stored per sub-tool result +_SUBSTEP_MAX_STEPS = 200 # max sub-steps stored per call_subagent card def _upsert_tool_step(steps: list, tid: Any, name: str, patch: dict) -> None: @@ -145,8 +164,9 @@ def _upsert_tool_step(steps: list, tid: Any, name: str, patch: dict) -> None: s["name"] = name return if len(steps) < _SUBSTEP_MAX_STEPS: - steps.append({"kind": "tool", "toolId": tid, "name": name or "tool", - "status": "running", **patch}) + steps.append( + {"kind": "tool", "toolId": tid, "name": name or "tool", "status": "running", **patch} + ) def attach_subagent_step(tool_calls_log: list, parent_tool_id: str, ev: dict) -> None: @@ -179,16 +199,18 @@ def attach_subagent_step(tool_calls_log: list, parent_tool_id: str, ev: dict) -> if st == "tool_call": inp = ev.get("input") - _upsert_tool_step(steps, ev.get("tool_id"), ev.get("tool_name"), - {"input": inp} if inp is not None else {}) + _upsert_tool_step( + steps, ev.get("tool_id"), ev.get("tool_name"), {"input": inp} if inp is not None else {} + ) elif st == "tool_result": out = ev.get("output") if isinstance(out, str) and len(out) > _SUBSTEP_OUTPUT_CAP: out = out[:_SUBSTEP_OUTPUT_CAP] + "…（已截断）" status = "error" if ev.get("status") == "error" else "success" - _upsert_tool_step(steps, ev.get("tool_id"), ev.get("tool_name"), - {"output": out, "status": status}) + _upsert_tool_step( + steps, ev.get("tool_id"), ev.get("tool_name"), {"output": out, "status": status} + ) elif st == "thinking": delta = ev.get("delta") or "" diff --git a/src/backend/core/llm/agent_factory.py b/src/backend/core/llm/agent_factory.py index dd551706..3473b6a2 100644 --- a/src/backend/core/llm/agent_factory.py +++ b/src/backend/core/llm/agent_factory.py @@ -24,15 +24,16 @@ from core.llm.mcp_pool import MCPConnectionPool from core.llm.middlewares import ( ActingToolCallIdMiddleware, - CitationAnchorMiddleware, AgentRuntimeState, + CitationAnchorMiddleware, DynamicModelMiddleware, FileContextMiddleware, FinishPinGuardMiddleware, GoalAnchorReminderMiddleware, IterBudgetReminderMiddleware, - StallInterventionMiddleware, OntologyGateMiddleware, + StallInterventionMiddleware, + SteerMiddleware, WorkspacePinHintMiddleware, ) from core.llm.providers.registry import get_spec, split_provider_extra @@ -40,9 +41,9 @@ from core.llm.tools import ( ReadStateTracker, register_bash, + register_channel_attachment, register_delete, register_edit, - register_channel_attachment, register_get_data_context, register_glob, register_grep, @@ -730,9 +731,7 @@ def _elapsed(): if not route.task_types or str(chat_mode or "chat") in route.task_types } visible_subagents = [ - agent - for agent in visible_subagents - if str(agent.get("agent_id") or "") in routed + agent for agent in visible_subagents if str(agent.get("agent_id") or "") in routed ] # The profile's tool allowlist, applied. It can only narrow: the candidate @@ -752,7 +751,9 @@ def _elapsed(): enabled_mcp_ids = [mcp_id for mcp_id in current_mcp if mcp_id in allowed_tools] _log.info( "[factory] profile %s narrowed MCP servers %d → %d", - profile.profile_id, len(current_mcp), len(enabled_mcp_ids), + profile.profile_id, + len(current_mcp), + len(enabled_mcp_ids), ) # ── Runtime Binder (GCE ticket 03) ────────────────────────────────────── @@ -1025,9 +1026,7 @@ async def _connect_http(key: str, cfg: dict): # evolution-authored ids — but the exposure gate is applied here anyway # rather than relying on that. A gate that only covers the paths we # happened to think of is not a gate. - skill_ids_to_register = _filter_skill_ids_for_user( - skill_ids_to_register, current_user_id - ) + skill_ids_to_register = _filter_skill_ids_for_user(skill_ids_to_register, current_user_id) # Note: a subagent's (user_agent) enabled_skill_ids is always a list ([] # when unconfigured) and never hits the None fallback above — i.e. "a # subagent with no skills configured has no skills"; strictly per its own @@ -1864,6 +1863,7 @@ def _build_toolkit() -> Toolkit: model_pinned=_subagent_model_pinned, user_id=current_user_id, chat_id=chat_id, + run_id=run_id, ontology_enabled=bool(_ontology_runtime.get("enabled")), ontology_runtime=_ontology_runtime, permission_context=PermissionContext(), @@ -1872,6 +1872,7 @@ def _build_toolkit() -> Toolkit: _middlewares: list = [ DynamicModelMiddleware(), # on_reply: switch models by chat_mode FileContextMiddleware(), # on_reply: inject file context + SteerMiddleware(), # on_acting/on_reasoning: inject queued user steer before tool I/O WorkspacePinHintMiddleware(), # on_reasoning: remind to pin IterBudgetReminderMiddleware(), # on_reasoning: inject a wrap-up reminder near max_iters # on_acting: the active profile's intervention rules, applied to *this* diff --git a/src/backend/core/llm/chat_models.py b/src/backend/core/llm/chat_models.py index 85fd8f6f..940623a1 100755 --- a/src/backend/core/llm/chat_models.py +++ b/src/backend/core/llm/chat_models.py @@ -35,12 +35,13 @@ from agentscope.message import Msg from agentscope.model import ChatModelBase, ChatResponse, OpenAIChatModel from agentscope.tool._types import ToolChoice - -from prompts.prompt_config import ModelConfig - -from core.llm.providers._fallback import L3_SYNTHETIC_METADATA, StructuredFallbackMixin # noqa: F401 +from core.llm.providers._fallback import ( # noqa: F401 + L3_SYNTHETIC_METADATA, + StructuredFallbackMixin, +) from core.llm.providers.registry import get_spec, split_provider_extra from core.llm.providers.vendor_models import build_litellm_model, build_native_model +from prompts.prompt_config import ModelConfig logger = logging.getLogger(__name__) @@ -52,6 +53,77 @@ # final error surface well within the run's lifetime. STREAM_READ_TIMEOUT_S: float = float(os.getenv("LLM_STREAM_READ_TIMEOUT_S", "600")) +_MULTIMODAL_CONTENT_TYPES = frozenset( + { + "audio", + "image", + "image_url", + "input_audio", + "input_image", + } +) + + +def _is_multimodal_unsupported_error(exc: Exception) -> bool: + """Whether an OpenAI-compatible endpoint explicitly rejected media input.""" + message = str(exc).lower() + if "not a multimodal model" in message: + return True + mentions_media = any(word in message for word in ("image", "audio", "multimodal")) + rejects_media = any( + phrase in message + for phrase in ( + "does not support", + "doesn't support", + "not supported", + "unsupported", + ) + ) + return mentions_media and rejects_media + + +def _without_multimodal_content(messages: list[dict[str, Any]]) -> tuple[list[dict[str, Any]], int]: + """Copy formatted messages while replacing unsupported media blocks with text.""" + sanitized: list[dict[str, Any]] = [] + removed = 0 + fallback_text = ( + "工具返回了图片或音频，但当前模型不支持直接读取该媒体。" + "请依据工具结果中已有的文字、元数据和图注继续完成回答，不要因此中止。" + ) + + for message in messages: + content = message.get("content") + if not isinstance(content, list): + sanitized.append(message) + continue + + kept: list[Any] = [] + removed_from_message = 0 + for block in content: + block_type = str(block.get("type", "")).lower() if isinstance(block, dict) else "" + if block_type in _MULTIMODAL_CONTENT_TYPES: + removed += 1 + removed_from_message += 1 + else: + kept.append(block) + + if not removed_from_message: + sanitized.append(message) + continue + + # AgentScope promotes multimodal tool outputs into a synthetic + # ``system-reminder`` user message. Its remaining identifier prose is + # meaningless once the media block is removed, so replace the whole + # reminder. For ordinary user messages, retain their accompanying text + # and append the same explicit degradation notice. + if message.get("name") == "system-reminder": + kept = [{"type": "text", "text": fallback_text}] + else: + kept.append({"type": "text", "text": fallback_text}) + sanitized.append({**message, "content": kept}) + + return sanitized, removed + def _build_chat_template_kwargs( *, @@ -173,12 +245,32 @@ async def _call_api( # type: ignore[override] kwargs["stream_options"] = {"include_usage": True} start_datetime = datetime.now() - response = await client.chat.completions.create(**kwargs) + try: + response = await client.chat.completions.create(**kwargs) + except Exception as exc: + # MCP tools may return image DataBlocks alongside useful JSON text + # (for example get_paper_figures). AgentScope promotes those blocks + # into an OpenAI image_url message for the next ReAct round. A + # text-only compatible endpoint rejects the whole request with a + # 400, which used to terminate the run immediately after the tool + # succeeded. Preserve multimodal input by default; only after an + # endpoint explicitly rejects it, retry once with the media blocks + # removed while retaining the tool's text/metadata/captions. + if not _is_multimodal_unsupported_error(exc): + raise + fallback_messages, removed = _without_multimodal_content(formatted_messages) + if not removed: + raise + logger.warning( + "Model %s rejected multimodal input; retrying without %d media block(s)", + model_name, + removed, + ) + kwargs["messages"] = fallback_messages + response = await client.chat.completions.create(**kwargs) audio_cfg = kwargs.get("audio") - audio_fmt = ( - audio_cfg.get("format", "wav") if isinstance(audio_cfg, dict) else "wav" - ) + audio_fmt = audio_cfg.get("format", "wav") if isinstance(audio_cfg, dict) else "wav" if self.stream: return self._parse_stream_response(start_datetime, response, audio_fmt) return self._parse_completion_response(start_datetime, response, audio_fmt) @@ -328,6 +420,7 @@ def _resolve_or_dummy(role_key: str): """Resolve config from DB, return None if not available.""" try: from core.services.model_config import ModelConfigService + return ModelConfigService.get_instance().resolve(role_key) except Exception as exc: logger.warning("ModelConfigService unavailable for role '%s': %s", role_key, exc) diff --git a/src/backend/core/llm/middlewares.py b/src/backend/core/llm/middlewares.py index 83d1e296..9057db62 100644 --- a/src/backend/core/llm/middlewares.py +++ b/src/backend/core/llm/middlewares.py @@ -145,9 +145,7 @@ async def on_acting(self, agent: Agent, input_kwargs: dict, next_handler): # no # 保证编号在该 agent 的整条流里唯一 allocator = resolve_allocator(agent) full_text = "".join((b.text or "") for b in text_blocks) - new_text, items = annotate_tool_result( - tool_name, tool_id, full_text, allocator - ) + new_text, items = annotate_tool_result(tool_name, tool_id, full_text, allocator) if items: allocator.register(tool_id, items) final.content = [TextBlock(type="text", text=new_text)] @@ -159,7 +157,8 @@ async def on_acting(self, agent: Agent, input_kwargs: dict, next_handler): # no annotated = True except Exception: # noqa: BLE001 logger.warning( - "[citation-anchor] middleware annotate failed tool=%s", tool_name, + "[citation-anchor] middleware annotate failed tool=%s", + tool_name, exc_info=True, ) @@ -182,6 +181,7 @@ class AgentRuntimeState(AgentState): model_pinned: bool = False user_id: str | None = None chat_id: str | None = None + run_id: str | None = None enable_thinking: bool = True chat_mode: str | None = None uploaded_files: List[dict] = Field(default_factory=list) @@ -189,6 +189,10 @@ class AgentRuntimeState(AgentState): user_message_text: str = "" ontology_enabled: bool = False ontology_runtime: dict = Field(default_factory=dict) + # Set by SteerMiddleware immediately before the next reasoning round. The + # streaming adapter turns it into one ``steer_applied`` event, then clears + # it. Keeping this on the typed runtime state avoids process-global queues. + steer_delivery: dict | None = None def apply_request_context(self, context: dict, user_message_text: str) -> None: """Populate per-request runtime fields from the request ``context`` dict (replaces the 1.x agent._jx_context). @@ -201,6 +205,7 @@ def apply_request_context(self, context: dict, user_message_text: str) -> None: self.model_provider_id = str(context.get("model_provider_id", "") or "") self.user_id = str(context.get("user_id", "") or "") or None self.chat_id = str(context.get("chat_id", "") or "") or None + self.run_id = str(context.get("run_id", "") or "") or None self.enable_thinking = bool(context.get("enable_thinking", True)) cm = str(context.get("chat_mode") or "").lower() or None if cm: @@ -213,6 +218,94 @@ def apply_request_context(self, context: dict, user_message_text: str) -> None: self.ontology_runtime = runtime if isinstance(runtime, dict) else {} +class SteerMiddleware(MiddlewareBase): + """Deliver a queued user instruction at the next safe ReAct boundary. + + ``on_reasoning`` is the primary insertion point: AgentScope calls it after + the previous tool results have entered context and immediately before the + next model call. ``on_acting`` is the earlier fallback for a steer that + arrives while the model is constructing its next tool call; it marks that + not-yet-started tool batch interrupted so the next reasoning round can + replan. Both paths preserve a valid assistant-tool-result-user order. + """ + + def __init__(self) -> None: + self._lock = asyncio.Lock() + self._delivery: dict | None = None + self._interrupted_tools = False + + async def on_acting(self, agent: Agent, input_kwargs: dict, next_handler): # noqa: ANN001 + run_id = str(getattr(agent.state, "run_id", "") or "") + if not run_id: + async for item in next_handler(**input_kwargs): + yield item + return + + async with self._lock: + if self._delivery is None: + from core.services.chat_steer_service import take_pending_steer + + self._delivery = await take_pending_steer(run_id) + self._interrupted_tools = self._delivery is not None + delivery = self._delivery + + if delivery is None: + async for item in next_handler(**input_kwargs): + yield item + return + + notice = "用户追加了新指令；本工具调用已在执行前中止，等待模型按新指令重新规划。" + block = TextBlock(type="text", text=notice) + yield ToolChunk(content=[block], state=ToolResultState.INTERRUPTED) + yield ToolResponse(content=[block], state=ToolResultState.INTERRUPTED) + + async def on_reasoning(self, agent: Agent, input_kwargs: dict, next_handler): # noqa: ANN001 + run_id = str(getattr(agent.state, "run_id", "") or "") + async with self._lock: + # A steer can arrive while a long-running tool is executing. There + # is no later on_acting hook in that round, so poll again here, + # after AgentScope saved the tool result and before it starts the + # next model call. This is the normal "insert after this tool" + # path; without it the steer waits until another tool call (or the + # whole run) finishes. + if self._delivery is None and run_id: + from core.services.chat_steer_service import take_pending_steer + + self._delivery = await take_pending_steer(run_id) + delivery = self._delivery + interrupted_tools = self._interrupted_tools + self._delivery = None + self._interrupted_tools = False + + if delivery is not None: + message = str(delivery.get("message") or "").strip() + if message: + agent.state.context.append( + Msg( + name="user", + role="user", + content=[ + TextBlock( + type="text", + text=( + "[用户在当前执行中追加的新指令]\n" + f"{message}\n" + + ( + "请立即按这条新指令调整后续计划；不要继续已经被中止的旧工具调用。" + if interrupted_tools + else "请立即按这条新指令调整后续计划；上一轮工具结果已经完成，可按需使用。" + ) + ), + ) + ], + ) + ) + agent.state.steer_delivery = dict(delivery) + + async for item in next_handler(**input_kwargs): + yield item + + class OntologyGateMiddleware(MiddlewareBase): """L-a deterministic gate: validate every visible tool call without an LLM.""" @@ -834,8 +927,10 @@ async def on_acting(self, agent: Agent, input_kwargs: dict, next_handler): # no tool_call = input_kwargs.get("tool_call") tool_name = str(getattr(tool_call, "name", "") or "") try: - signature = (tool_name, json.dumps(getattr(tool_call, "input", None), sort_keys=True, - default=str)) + signature = ( + tool_name, + json.dumps(getattr(tool_call, "input", None), sort_keys=True, default=str), + ) except Exception: # noqa: BLE001 signature = (tool_name, "") @@ -882,7 +977,8 @@ def _maybe_intervene(self, agent: Agent) -> None: applicable = [ rule for rule in self._rules - if self._signals.get(getattr(rule, "signal", ""), 0) >= int(getattr(rule, "threshold", 0) or 0) + if self._signals.get(getattr(rule, "signal", ""), 0) + >= int(getattr(rule, "threshold", 0) or 0) and int(getattr(rule, "threshold", 0) or 0) > 0 ] if not applicable: @@ -935,7 +1031,9 @@ def _maybe_intervene(self, agent: Agent) -> None: ], ) ) - logger.info("[stall-intervention] %s >= %s -> %s", signal, getattr(rule, "threshold", 0), action) + logger.info( + "[stall-intervention] %s >= %s -> %s", signal, getattr(rule, "threshold", 0), action + ) # ── FinishPinGuard ───────────────────────────────────────────────────────── @@ -969,6 +1067,7 @@ async def on_reasoning(self, agent: Agent, input_kwargs: dict, next_handler): __all__ = [ "AgentRuntimeState", + "SteerMiddleware", "DynamicModelMiddleware", "FileContextMiddleware", "WorkspacePinHintMiddleware", diff --git a/src/backend/core/services/chat_steer_service.py b/src/backend/core/services/chat_steer_service.py new file mode 100644 index 00000000..ebcdc323 --- /dev/null +++ b/src/backend/core/services/chat_steer_service.py @@ -0,0 +1,76 @@ +"""Redis-backed pending steer instructions for live chat runs. + +The API process and the worker consuming a ``ChatRun`` may be different +processes, so the hand-off cannot rely on an in-memory queue. A run accepts at +most one pending instruction; editing/re-submitting the same queued card simply +replaces the value until the execution middleware atomically consumes it. +""" + +from __future__ import annotations + +import json +from typing import Any, Dict, Optional + +from core.infra.redis import get_redis +from redis.exceptions import WatchError + +_STEER_KEY = "jx:chat:run:{run_id}:steer" +_STEER_TTL_SECONDS = 3600 + + +def _key(run_id: str) -> str: + return _STEER_KEY.format(run_id=run_id) + + +async def put_pending_steer(run_id: str, payload: Dict[str, Any]) -> None: + """Create or replace the single pending steer instruction for ``run_id``.""" + await get_redis().set( + _key(run_id), + json.dumps(payload, ensure_ascii=False), + ex=_STEER_TTL_SECONDS, + ) + + +async def take_pending_steer(run_id: str) -> Optional[Dict[str, Any]]: + """Atomically consume the pending steer instruction, if one exists.""" + redis = get_redis() + try: + raw = await redis.getdel(_key(run_id)) + except AttributeError: # pragma: no cover - compatibility with old redis clients + raw = await redis.get(_key(run_id)) + if raw is not None: + await redis.delete(_key(run_id)) + if not raw: + return None + try: + value = json.loads(raw) + except (TypeError, json.JSONDecodeError): + return None + return value if isinstance(value, dict) else None + + +async def remove_pending_steer(run_id: str, steer_id: str) -> bool: + """Withdraw a still-pending instruction without deleting a newer one.""" + redis = get_redis() + key = _key(run_id) + async with redis.pipeline(transaction=True) as pipe: + while True: + try: + await pipe.watch(key) + raw = await pipe.get(key) + if not raw: + return False + try: + value = json.loads(raw) + except (TypeError, json.JSONDecodeError): + return False + if not isinstance(value, dict) or value.get("steer_id") != steer_id: + return False + pipe.multi() + pipe.delete(key) + result = await pipe.execute() + return bool(result and result[0]) + except WatchError: + # The API may have replaced the queued card between GET and + # DELETE. Re-read instead of deleting that newer instruction. + continue diff --git a/src/backend/orchestration/batch_orchestrator.py b/src/backend/orchestration/batch_orchestrator.py index e346d829..8516f6a0 100644 --- a/src/backend/orchestration/batch_orchestrator.py +++ b/src/backend/orchestration/batch_orchestrator.py @@ -453,12 +453,15 @@ async def _run_item_via_workflow( tool_name = payload.get("name", "unknown") tool_args = payload.get("args", {}) tool_id = payload.get("id", "") - _upsert_tool_call(tool_calls_log, { - "tool_name": tool_name, - "tool_display_name": TOOL_DISPLAY_NAMES.get(tool_name, tool_name), - "tool_args": tool_args if isinstance(tool_args, dict) else {}, - "tool_id": tool_id, - }) + _upsert_tool_call( + tool_calls_log, + { + "tool_name": tool_name, + "tool_display_name": TOOL_DISPLAY_NAMES.get(tool_name, tool_name), + "tool_args": tool_args if isinstance(tool_args, dict) else {}, + "tool_id": tool_id, + }, + ) elif event_type == "tool_result": tool_name = payload.get("name", "unknown") tool_id = payload.get("id", "") diff --git a/src/backend/orchestration/chat_run_executor.py b/src/backend/orchestration/chat_run_executor.py index 7b1501a5..66ffdc06 100644 --- a/src/backend/orchestration/chat_run_executor.py +++ b/src/backend/orchestration/chat_run_executor.py @@ -366,6 +366,11 @@ async def _emit(event: Dict[str, Any]) -> None: from core.llm import workspace as _workspace_mod full_response = "" + # A steer starts a new visible assistant segment in the same backend run. + # Each segment gets its own message id so persisted history stays in the + # same chronological order the user saw while streaming. + current_message_id = message_id + latest_user_message = raw_user_message metadata: Dict[str, Any] = {} tool_calls_log: list = [] _workspace_mod.init_state() @@ -494,13 +499,100 @@ def _flush_thinking() -> None: } ) + elif chunk_type == "steer_applied": + _flush_thinking() + steer_id = str(chunk.get("steer_id") or "")[:64] + steer_message = str(chunk.get("message") or "").strip() + steer_message_id = ( + f"msg_{uuid.uuid5(uuid.NAMESPACE_URL, f'{run_id}:{steer_id}').hex[:16]}" + if steer_id + else f"msg_{uuid.uuid4().hex[:16]}" + ) + next_assistant_message_id = ( + f"msg_{uuid.uuid5(uuid.NAMESPACE_URL, f'{run_id}:{steer_id}:assistant').hex[:16]}" + if steer_id + else f"msg_{uuid.uuid4().hex[:16]}" + ) + had_assistant_output = bool(full_response or tool_calls_log) + if steer_message: + # Close the visible assistant segment before inserting the + # user's steer. Persisting in this order is what keeps a + # refresh from moving every mid-run user message above the + # whole assistant response. + with SessionLocal() as db: + chat_service = ChatService(db) + if had_assistant_output: + chat_service.add_message( + chat_id=chat_id, + role="assistant", + content=full_response, + model=model_name, + tool_calls=tool_calls_log if tool_calls_log else None, + message_id=current_message_id, + extra_data={ + "timestamp": now_iso(), + "is_markdown": bool( + "\n" in full_response + or "```" in full_response + or "**" in full_response + ), + "message_id": current_message_id, + "run_id": run_id, + "steer_segment": True, + "duration_ms": int( + (time.monotonic() - _run_started_monotonic) * 1000 + ), + }, + ) + # Persist once the middleware has actually injected the + # instruction. A deterministic id makes replay safe. + chat_service.upsert_message( + chat_id=chat_id, + role="user", + content=steer_message, + message_id=steer_message_id, + extra_data={ + "timestamp": now_iso(), + "steer": True, + "run_id": run_id, + "steer_id": steer_id, + }, + ) + latest_user_message = steer_message + await _emit( + { + "type": "steer_applied", + "chat_id": chat_id, + "run_id": run_id, + "steer_id": steer_id, + "message": steer_message, + "message_id": steer_message_id, + "previous_assistant_message_id": ( + current_message_id if had_assistant_output else None + ), + "next_assistant_message_id": next_assistant_message_id, + "had_assistant_output": had_assistant_output, + } + ) + current_message_id = next_assistant_message_id + # The workflow reads this dict again when it emits the final + # evolution/memory settlement marker, so keep that marker bound + # to the post-steer assistant segment too. + context["message_id"] = current_message_id + full_response = "" + tool_calls_log = [] + _thinking_parts.clear() + metadata = {} + try: + from core.services.log_service import set_current_message_id + + set_current_message_id(current_message_id) + except Exception: # pragma: no cover - logging must never fail a run + pass + elif chunk_type == "tool_call": _flush_thinking() _tc_evt = build_tool_call_event(chunk, chat_id, tool_calls_log) - # 记录该工具卡片出现时正文的累计长度：历史重建按此偏移把 - # 「文本 ↔ 工具卡片」按流式原顺序交错（问题15：刷新后内容与实时不一致）。 - for _tc in tool_calls_log: - _tc.setdefault("content_offset", len(full_response)) await _emit(_tc_evt) elif chunk_type == "tool_call_start": @@ -648,7 +740,7 @@ def _flush_thinking() -> None: "warnings": chunk.get("warnings", []), "is_markdown": chunk.get("is_markdown", False), "chat_id": chat_id, - "message_id": message_id, + "message_id": current_message_id, "citations": chunk.get("citations", []), "workspace_files": _ws_files, "ontology_governance": chunk.get("ontology_governance"), @@ -667,7 +759,7 @@ def _flush_thinking() -> None: "artifacts": metadata.get("artifacts", []), "warnings": metadata.get("warnings", []), "citations": metadata.get("citations", []), - "message_id": message_id, + "message_id": current_message_id, "workspace_files": _ws_files, "duration_ms": int((time.monotonic() - _run_started_monotonic) * 1000), } @@ -684,7 +776,7 @@ def _flush_thinking() -> None: model=model_name, tool_calls=tool_calls_log if tool_calls_log else None, usage=usage_payload, - message_id=message_id, + message_id=current_message_id, extra_data=_persist_extra, ) # Build a ProjectScope from the workflow context and pass it @@ -718,9 +810,9 @@ def _flush_thinking() -> None: if not seen_batch_confirm: _spawn_followup_task( chat_id=chat_id, - user_msg=raw_user_message, + user_msg=latest_user_message, response=full_response, - msg_id=message_id, + msg_id=current_message_id, ) # End-of-turn compaction: when real token usage crosses the @@ -766,7 +858,7 @@ def _flush_thinking() -> None: role="assistant", content="", model=model_name, - message_id=message_id, + message_id=current_message_id, error={"error": str(exc), "timestamp": _utcnow().isoformat()}, ) except Exception: @@ -1313,15 +1405,18 @@ async def start_autonomous_loop_run( from core.services.loop_service import LoopService as _LoopSvc with SessionLocal() as db: - _LoopSvc(db).save_start_params(loop_id, { - "model_name": model_name, - "model_provider_id": model_provider_id, - "evaluator_model": evaluator_model, - "worker_max_iters": worker_max_iters, - "hitl_enabled": hitl_enabled, - "enable_thinking": enable_thinking, - "chat_mode": chat_mode, - }) + _LoopSvc(db).save_start_params( + loop_id, + { + "model_name": model_name, + "model_provider_id": model_provider_id, + "evaluator_model": evaluator_model, + "worker_max_iters": worker_max_iters, + "hitl_enabled": hitl_enabled, + "enable_thinking": enable_thinking, + "chat_mode": chat_mode, + }, + ) except Exception: # noqa: BLE001 - 参数存档失败不阻塞启动 logger.warning("loop start_params persist failed", exc_info=True) run = _create_run_record(chat_id=chat_id, user_id=user_id, request_payload=request_payload) diff --git a/src/backend/orchestration/streaming.py b/src/backend/orchestration/streaming.py index d5c4c77a..4fc9363c 100644 --- a/src/backend/orchestration/streaming.py +++ b/src/backend/orchestration/streaming.py @@ -49,10 +49,9 @@ from agentscope.agent import Agent from agentscope.mcp import MCPClient from agentscope.message import Msg - -from core.services import log_service as log_writer from core.infra.logging import LogContext from core.llm.message_compat import session_to_msgs +from core.services import log_service as log_writer logger = logging.getLogger(__name__) @@ -113,7 +112,7 @@ def _strip_thinking_answer(raw: str, enable_thinking: bool, in_thinking: bool) - return raw, False last_end = raw.rfind("") if last_end != -1: - return raw[last_end + len(""):], False + return raw[last_end + len("") :], False if "" in raw or in_thinking: return "", True return raw, False @@ -232,11 +231,12 @@ async def stream( user_msg: Optional[Msg] = None if last_user_content: from core.llm.message_compat import _wrap_content - user_msg = Msg(name="user", role="user", - content=_wrap_content(last_user_content)) + + user_msg = Msg(name="user", role="user", content=_wrap_content(last_user_content)) # myspace write-confirmation gate (in-house ContextVar, distinct from 2.0 native HITL) from core.llm.tools import _myspace_confirm as _mc + _confirm_chat_id = st.chat_id or None def _drain_confirm_signals() -> list: @@ -258,6 +258,7 @@ def _drain_confirm_signals() -> list: # Subagent streaming bypass: register this run's event_q so call_subagent (separate thread) # can deliver the subagent's thinking/tool_call/... events back into this main queue in real time. from core.llm import _subagent_stream + _subagent_stream.attach(st.chat_id, asyncio.get_running_loop(), event_q) async def _produce(): @@ -266,6 +267,7 @@ async def _produce(): await event_q.put(("ev", ev)) except BaseException as e: # noqa: BLE001 import traceback + logger.error("Agent reply_stream failed: %r\n%s", e, traceback.format_exc()) await event_q.put(("err", e)) finally: @@ -319,6 +321,14 @@ async def _produce(): yield ("subagent_event", payload) continue # kind == "ev" + steer_delivery = getattr(agent.state, "steer_delivery", None) + if isinstance(steer_delivery, dict): + # SteerMiddleware appends the user instruction immediately + # before this next reasoning round. Emit one acknowledgement + # before mapping ModelCallStart so the queued-card UI can + # switch from "waiting" to "applied" deterministically. + agent.state.steer_delivery = None + yield ("steer_applied", dict(steer_delivery)) reasoning_protocol = self._take_reasoning_protocol() if reasoning_protocol is not None: yield ("reasoning_protocol", reasoning_protocol) @@ -352,23 +362,25 @@ async def _produce(): try: _started_mono = _rec.get("started_monotonic") _dur = int((time.monotonic() - _started_mono) * 1000) if _started_mono else None - log_writer.schedule_tool_call_write({ - # user_id/chat_id are taken explicitly from agent.state — contextvars are - # unreliable in the stream() generator frame (the agent runs in the context - # snapshot of _produce's create_task, while tool results are written in the - # generator's consumer frame; the two contexts don't sync), so _context_ids - # cannot be relied on. - "user_id": st.user_id or None, - "chat_id": st.chat_id or None, - "tool_name": _rec.get("tool_name", "unknown"), - "tool_call_id": _tid, - "tool_args": _rec.get("tool_args"), - "tool_result": None, - "status": "failed", - "error_message": "no tool_result received (stream ended)", - "duration_ms": _dur, - "started_at": _rec.get("started_at"), - }) + log_writer.schedule_tool_call_write( + { + # user_id/chat_id are taken explicitly from agent.state — contextvars are + # unreliable in the stream() generator frame (the agent runs in the context + # snapshot of _produce's create_task, while tool results are written in the + # generator's consumer frame; the two contexts don't sync), so _context_ids + # cannot be relied on. + "user_id": st.user_id or None, + "chat_id": st.chat_id or None, + "tool_name": _rec.get("tool_name", "unknown"), + "tool_call_id": _tid, + "tool_args": _rec.get("tool_args"), + "tool_result": None, + "status": "failed", + "error_message": "no tool_result received (stream ended)", + "duration_ms": _dur, + "started_at": _rec.get("started_at"), + } + ) except Exception: logger.debug("pending tool_call flush failed", exc_info=True) self._pending_tool_calls.clear() @@ -379,6 +391,7 @@ async def _produce(): except Exception: pass if not prod_task.done(): + async def _wait(): try: await asyncio.wait_for(asyncio.shield(prod_task), timeout=10) @@ -390,6 +403,7 @@ async def _wait(): pass except Exception: pass + asyncio.create_task(_wait()) async def _map_event(self, ev: Any) -> AsyncIterator[Tuple[str, Any]]: @@ -415,8 +429,9 @@ async def _map_event(self, ev: Any) -> AsyncIterator[Tuple[str, Any]]: ) if answer and answer != self._emitted_answer: out = ( - answer[len(self._emitted_answer):] - if answer.startswith(self._emitted_answer) else answer + answer[len(self._emitted_answer) :] + if answer.startswith(self._emitted_answer) + else answer ) if out: yield ("text_delta", out) @@ -484,6 +499,7 @@ async def _map_event(self, ev: Any) -> AsyncIterator[Tuple[str, Any]]: {"name": name, "id": tid, "delta": pending_delta}, ) import json + try: args = json.loads(args_str) if args_str else {} except json.JSONDecodeError: @@ -493,6 +509,7 @@ async def _map_event(self, ev: Any) -> AsyncIterator[Tuple[str, Any]]: rec["tool_args"] = args try: from orchestration.tool_callbacks import note_tool_call + note_tool_call(self.__dict__.setdefault("_tool_warn_state", {}), name, args) except Exception: # noqa: BLE001 pass @@ -501,36 +518,52 @@ async def _map_event(self, ev: Any) -> AsyncIterator[Tuple[str, Any]]: if nm == "ToolResultTextDeltaEvent": tid = getattr(ev, "tool_call_id", "") or "" - self._tool_result_buf[tid] = self._tool_result_buf.get(tid, "") + (getattr(ev, "delta", "") or "") + self._tool_result_buf[tid] = self._tool_result_buf.get(tid, "") + ( + getattr(ev, "delta", "") or "" + ) return if nm == "ToolResultEndEvent": tid = getattr(ev, "tool_call_id", "") or "" content = self._tool_result_buf.pop(tid, "") - state = str(getattr(ev, "state", "") or "") + raw_state = getattr(ev, "state", "") or "" + state = str(getattr(raw_state, "value", raw_state) or "") pending = self._pending_tool_calls.pop(tid, None) - name = (getattr(ev, "tool_call_name", "") or (pending or {}).get("tool_name") - or self._tool_name_buf.get(tid) or "unknown") + name = ( + getattr(ev, "tool_call_name", "") + or (pending or {}).get("tool_name") + or self._tool_name_buf.get(tid) + or "unknown" + ) try: - is_error = state == "error" or _looks_like_tool_error(content) + is_error = state in {"error", "denied", "interrupted"} or _looks_like_tool_error( + content + ) started_mono = pending.get("started_monotonic") if pending else None - duration_ms = int((time.monotonic() - started_mono) * 1000) if started_mono else None - log_writer.schedule_tool_call_write({ - # See the matching comment in _produce: carry user_id/chat_id explicitly, never rely on contextvars. - "user_id": self.agent.state.user_id or None, - "chat_id": self.agent.state.chat_id or None, - "tool_name": (pending or {}).get("tool_name") or name, - "tool_call_id": tid, - "tool_args": (pending or {}).get("tool_args"), - "tool_result": content, - "status": "failed" if is_error else "success", - "error_message": content if is_error else None, - "duration_ms": duration_ms, - "started_at": (pending or {}).get("started_at"), - }) + duration_ms = ( + int((time.monotonic() - started_mono) * 1000) if started_mono else None + ) + log_writer.schedule_tool_call_write( + { + # See the matching comment in _produce: carry user_id/chat_id explicitly, never rely on contextvars. + "user_id": self.agent.state.user_id or None, + "chat_id": self.agent.state.chat_id or None, + "tool_name": (pending or {}).get("tool_name") or name, + "tool_call_id": tid, + "tool_args": (pending or {}).get("tool_args"), + "tool_result": content, + "status": "failed" if is_error else "success", + "error_message": content if is_error else None, + "duration_ms": duration_ms, + "started_at": (pending or {}).get("started_at"), + } + ) except Exception: # noqa: BLE001 logger.debug("tool_call log persist failed", exc_info=True) - yield ("tool_result", {"name": name, "id": tid, "content": content}) + yield ( + "tool_result", + {"name": name, "id": tid, "content": content, "status": state}, + ) return if nm == "ModelCallStartEvent": @@ -540,10 +573,12 @@ async def _map_event(self, ev: Any) -> AsyncIterator[Tuple[str, Any]]: if nm == "ModelCallEndEvent": self._model_call_inflight = False - self._usage_records.append({ - "prompt_tokens": int(getattr(ev, "input_tokens", 0) or 0), - "completion_tokens": int(getattr(ev, "output_tokens", 0) or 0), - }) + self._usage_records.append( + { + "prompt_tokens": int(getattr(ev, "input_tokens", 0) or 0), + "completion_tokens": int(getattr(ev, "output_tokens", 0) or 0), + } + ) # 首轮权威判定：思考模式下整轮正文没有出现 → 该模型不内联思考 # （结构化 reasoning 通道，或本轮确实没思考）。补发协议标记，前端据此 # 把误当思考缓冲/展示的正文重归正文区（bug：无思考时正文进思考块）。 @@ -568,14 +603,20 @@ async def _map_event(self, ev: Any) -> AsyncIterator[Tuple[str, Any]]: if nm == "RequireUserConfirmEvent": # 2.0 native HITL (distinct from the myspace gate); our tools default to ALLOW, so this rarely triggers. try: - yield ("file_confirm", { - "reply_id": getattr(ev, "reply_id", ""), - "tool_calls": [ - {"id": getattr(tc, "id", ""), "name": getattr(tc, "name", ""), - "input": getattr(tc, "input", "")} - for tc in (getattr(ev, "tool_calls", []) or []) - ], - }) + yield ( + "file_confirm", + { + "reply_id": getattr(ev, "reply_id", ""), + "tool_calls": [ + { + "id": getattr(tc, "id", ""), + "name": getattr(tc, "name", ""), + "input": getattr(tc, "input", ""), + } + for tc in (getattr(ev, "tool_calls", []) or []) + ], + }, + ) except Exception: # noqa: BLE001 pass return @@ -593,6 +634,7 @@ async def _map_event(self, ev: Any) -> AsyncIterator[Tuple[str, Any]]: async def shutdown(self): """Close transient (per-request) MCP clients.""" from core.llm.mcp_manager import close_clients + try: await close_clients(self.mcp_clients) except Exception as exc: diff --git a/src/backend/orchestration/workflow.py b/src/backend/orchestration/workflow.py index 72515b33..17c8bae7 100644 --- a/src/backend/orchestration/workflow.py +++ b/src/backend/orchestration/workflow.py @@ -65,9 +65,7 @@ def _extract_project_ctx(context: Dict[str, Any]) -> Optional[Dict[str, Any]]: return {k: context.get(k) for k in _PROJECT_CTX_KEYS} -def _resolve_agent_model_runtime( - agent: Any, fallback_model_name: Any = "" -) -> Tuple[str, int]: +def _resolve_agent_model_runtime(agent: Any, fallback_model_name: Any = "") -> Tuple[str, int]: """Return the effective model name and context window baked into an agent. AgentScope chat models expose the upstream name as ``.model``. Some provider @@ -1186,9 +1184,7 @@ def run_chat_workflow( ) _workflow_turbo = ( - _direct_user_agent is None - and _workflow_chat_mode == "turbo" - and not _workflow_batch_chat + _direct_user_agent is None and _workflow_chat_mode == "turbo" and not _workflow_batch_chat ) if _workflow_turbo: # 极速模式下子智能体仅在本轮被显式委派/@ 时入场（此路径无文本 @ 解析）。 @@ -1243,9 +1239,7 @@ async def _run(): except Exception as exc: # noqa: BLE001 logger.warning("[workflow] set agent.state failed: %s", exc) - _actual_model, _ctx_window = _resolve_agent_model_runtime( - agent, _workflow_model_name - ) + _actual_model, _ctx_window = _resolve_agent_model_runtime(agent, _workflow_model_name) # PreTurn compaction safety net (symmetric with the streaming path # — both workflow entry points protect themselves, and future new @@ -1268,9 +1262,7 @@ async def _run(): if history and history[-1].get("role") in ("user", "human"): history.pop() - _ctx_mgr = ContextWindowManager( - ContextBudget(model_context_window=_ctx_window) - ) + _ctx_mgr = ContextWindowManager(ContextBudget(model_context_window=_ctx_window)) history = _ctx_mgr.trim_history(history) if history: @@ -1525,9 +1517,7 @@ async def _finish_direct_log( # 证据锚点发号器：跨轮续号；创建后绑到 agent 上（见下方 attach_allocator）， # 中间件与本函数由此共享同一个计数器 _anchor_allocator = AnchorAllocator( - await asyncio.to_thread( - anchor_start_for_chat, str(context.get("chat_id") or "") or None - ) + await asyncio.to_thread(anchor_start_for_chat, str(context.get("chat_id") or "") or None) ) try: @@ -1603,12 +1593,8 @@ async def _finish_direct_log( # checkpoint system of its own); over budget it is trimmed directly to # the token budget (layer-C compression of oversized user messages # still happens inside manage_context). - _actual_model, _ctx_window = _resolve_agent_model_runtime( - agent, _stream_model_name - ) - ctx_manager = ContextWindowManager( - ContextBudget(model_context_window=_ctx_window) - ) + _actual_model, _ctx_window = _resolve_agent_model_runtime(agent, _stream_model_name) + ctx_manager = ContextWindowManager(ContextBudget(model_context_window=_ctx_window)) trimmed, dropped_messages = ctx_manager.manage_context(session_messages) if dropped_messages: logger.warning( @@ -1662,6 +1648,9 @@ async def _finish_direct_log( elif event_type == "reasoning_protocol": yield {"type": "thinking", **payload} + elif event_type == "steer_applied": + yield {"type": "steer_applied", **(payload or {})} + elif event_type == "thinking_delta": yield {"type": "thinking", "delta": payload} @@ -1827,6 +1816,7 @@ async def _finish_direct_log( "result": tool_result_json, "tool_id": tool_id, "citations": cit_dicts, + "status": payload.get("status", "success"), } elif event_type in ("heartbeat", "model_progress"): @@ -2134,9 +2124,7 @@ def _assemble_episode_background( bundle = getattr(agent, "_jx_asset_bundle", None) if agent is not None else None - sink = TraceSink( - run_id=run_id, message_id=message_id, chat_id=chat_id, user_id=user_id - ) + sink = TraceSink(run_id=run_id, message_id=message_id, chat_id=chat_id, user_id=user_id) # The rendered memory block has already discarded ids, scores and ranks; # this is the only place they still exist. retrieval = get_last_retrieval(memory_task) @@ -2160,9 +2148,7 @@ def _assemble_episode_background( if selection is not None: from core.evolution.events import EV_SKILL_SELECTED - sink.append( - EV_SKILL_SELECTED, selection.to_event_payload(), asset_kind="skill" - ) + sink.append(EV_SKILL_SELECTED, selection.to_event_payload(), asset_kind="skill") sink.flush() # Stamp the user's contribution choice onto the episode. Doing it at @@ -2322,9 +2308,7 @@ async def astream_chat_workflow( # 证据锚点发号器：跨轮续号；创建后绑到 agent 上（见下方 attach_allocator）， # 中间件与本函数由此共享同一个计数器 _anchor_allocator = AnchorAllocator( - await asyncio.to_thread( - anchor_start_for_chat, str(context.get("chat_id") or "") or None - ) + await asyncio.to_thread(anchor_start_for_chat, str(context.get("chat_id") or "") or None) ) try: @@ -2517,9 +2501,7 @@ async def astream_chat_workflow( # object do we fall back to resolve — unconfigured raises, fail loud, # never silently run with the wrong window. # preturn and manage_context below share the same value. - _actual_model, _ctx_window = _resolve_agent_model_runtime( - agent, _stream_model_name - ) + _actual_model, _ctx_window = _resolve_agent_model_runtime(agent, _stream_model_name) try: from core.services.compaction_service import maybe_run_pre_turn_compaction @@ -2604,6 +2586,9 @@ async def astream_chat_workflow( elif event_type == "reasoning_protocol": yield {"type": "thinking", **payload} + elif event_type == "steer_applied": + yield {"type": "steer_applied", **(payload or {})} + elif event_type == "thinking_delta": yield {"type": "thinking", "delta": payload} @@ -2819,6 +2804,7 @@ async def astream_chat_workflow( "result": tool_result_json, "tool_id": tool_id, "citations": cit_dicts, + "status": payload.get("status", "success"), **({"subagent_name": _tr_sa_name} if _tr_sa_name else {}), } diff --git a/src/backend/tests/chat/test_stream_event_builders.py b/src/backend/tests/chat/test_stream_event_builders.py index dbc3ca6f..371bb818 100644 --- a/src/backend/tests/chat/test_stream_event_builders.py +++ b/src/backend/tests/chat/test_stream_event_builders.py @@ -86,12 +86,14 @@ def test_tool_call_event_and_log_upsert(): "chat_id": "c1", } # The builder upserts the tool_call into the log (without chat_id/type). - assert log == [{ - "tool_name": "bash", - "tool_display_name": "Bash", - "tool_args": {"command": "ls"}, - "tool_id": "t1", - }] + assert log == [ + { + "tool_name": "bash", + "tool_display_name": "Bash", + "tool_args": {"command": "ls"}, + "tool_id": "t1", + } + ] def test_tool_call_event_subagent_passthrough(): @@ -120,6 +122,7 @@ def test_tool_result_event_and_log_attach(): "tool_id": "t1", "chat_id": "c1", "citations": [{"n": 1}], + "status": "success", } # The result is attached onto the matching log entry. assert log[0]["result"] == {"stdout": "ok"} @@ -131,3 +134,19 @@ def test_tool_result_event_defaults_when_missing(): assert evt["result"] == {} assert evt["citations"] == [] assert "subagent_name" not in evt + + +def test_interrupted_tool_result_keeps_neutral_status(): + log = [{"tool_name": "bash", "tool_id": "t1"}] + evt = build_tool_result_event( + { + "tool_id": "t1", + "tool_name": "bash", + "result": {"message": "interrupted by steer"}, + "status": "interrupted", + }, + "c1", + log, + ) + assert evt["status"] == "interrupted" + assert log[0]["status"] == "interrupted" diff --git a/src/backend/tests/llm/test_openai_provider.py b/src/backend/tests/llm/test_openai_provider.py index c476d037..7755b8c7 100644 --- a/src/backend/tests/llm/test_openai_provider.py +++ b/src/backend/tests/llm/test_openai_provider.py @@ -1,5 +1,16 @@ -"""Tests for the dedicated OpenAI/Codex provider preset.""" +"""Tests for OpenAI-compatible model providers.""" +from types import SimpleNamespace + +import pytest +from agentscope.message import ( + Base64Source, + DataBlock, + Msg, + TextBlock, + ToolResultBlock, + ToolResultState, +) from core.llm.chat_models import make_chat_model from core.llm.providers.registry import get_spec, to_frontend_schema @@ -48,3 +59,63 @@ def test_generic_compatible_provider_keeps_existing_reasoning_transport(): "reasoning_effort": "high", } assert model.structured_reasoning is False + + +@pytest.mark.asyncio +async def test_text_only_model_retries_tool_image_result_without_multimodal_blocks(monkeypatch): + """A figure-returning MCP must not end the ReAct loop on a text-only model.""" + model = make_chat_model( + model="deepseekv4-flash", + temperature=0.0, + max_tokens=32, + timeout=10, + base_url="http://model.test/api/v1", + api_key="test-key", + provider="openai_compatible", + stream=False, + context_size=4096, + ) + tool_result = ToolResultBlock( + id="call-figures", + name="get_paper_figures", + output=[ + TextBlock(type="text", text='{"title":"paper","figures":[{"caption":"architecture"}]}'), + DataBlock( + type="data", + source=Base64Source( + type="base64", + media_type="image/png", + data="QUJD", + ), + ), + ], + state=ToolResultState.SUCCESS, + ) + messages = [Msg(name="assistant", role="assistant", content=[tool_result])] + + calls: list[dict] = [] + expected_response = object() + + async def create(**kwargs): + calls.append(kwargs) + if len(calls) == 1: + raise RuntimeError("deepseekv4-flash is not a multimodal model") + return expected_response + + fake_client = SimpleNamespace( + chat=SimpleNamespace(completions=SimpleNamespace(create=create)), + ) + monkeypatch.setattr(model, "_build_client", lambda: fake_client) + monkeypatch.setattr( + model, + "_parse_completion_response", + lambda _started_at, response, _audio_format: response, + ) + + response = await model._call_api("deepseekv4-flash", messages) + + assert response is expected_response + assert len(calls) == 2 + assert "image_url" in str(calls[0]["messages"]) + assert "image_url" not in str(calls[1]["messages"]) + assert "architecture" in str(calls[1]["messages"]) diff --git a/src/backend/tests/orchestration/test_chat_steer.py b/src/backend/tests/orchestration/test_chat_steer.py new file mode 100644 index 00000000..fc522df2 --- /dev/null +++ b/src/backend/tests/orchestration/test_chat_steer.py @@ -0,0 +1,199 @@ +"""Mid-run chat steering: Redis hand-off and ReAct middleware behavior.""" + +from types import SimpleNamespace + +import fakeredis.aioredis +import pytest +from agentscope.message import ToolResultState +from agentscope.tool._response import ToolChunk +from core.llm.middlewares import SteerMiddleware +from core.services import chat_steer_service +from orchestration import chat_run_executor as executor + + +@pytest.mark.asyncio +async def test_pending_steer_round_trip_is_single_consumer(monkeypatch): + redis = fakeredis.aioredis.FakeRedis(decode_responses=True) + monkeypatch.setattr(chat_steer_service, "get_redis", lambda: redis) + + payload = {"steer_id": "s1", "message": "换一种实现", "run_id": "r1"} + await chat_steer_service.put_pending_steer("r1", payload) + + assert await chat_steer_service.take_pending_steer("r1") == payload + assert await chat_steer_service.take_pending_steer("r1") is None + + +@pytest.mark.asyncio +async def test_withdraw_only_removes_matching_steer(monkeypatch): + redis = fakeredis.aioredis.FakeRedis(decode_responses=True) + monkeypatch.setattr(chat_steer_service, "get_redis", lambda: redis) + await chat_steer_service.put_pending_steer( + "r1", + {"steer_id": "newer", "message": "保留这条"}, + ) + + assert await chat_steer_service.remove_pending_steer("r1", "older") is False + assert (await chat_steer_service.take_pending_steer("r1"))["steer_id"] == "newer" + + +@pytest.mark.asyncio +async def test_middleware_interrupts_tool_then_injects_user_instruction(monkeypatch): + delivery = {"steer_id": "s1", "message": "先停下来，改查第二个方案"} + + async def take_pending_steer(run_id: str): + assert run_id == "r1" + return delivery + + monkeypatch.setattr(chat_steer_service, "take_pending_steer", take_pending_steer) + middleware = SteerMiddleware() + agent = SimpleNamespace(state=SimpleNamespace(run_id="r1", context=[], steer_delivery=None)) + tool_executed = False + + async def tool_handler(**_kwargs): + nonlocal tool_executed + tool_executed = True + yield ToolChunk(content=[], state=ToolResultState.SUCCESS) + + results = [item async for item in middleware.on_acting(agent, {}, tool_handler)] + + assert tool_executed is False + assert results[-1].state == ToolResultState.INTERRUPTED + + async def model_handler(**_kwargs): + yield "model-started" + + model_events = [item async for item in middleware.on_reasoning(agent, {}, model_handler)] + + assert model_events == ["model-started"] + assert agent.state.steer_delivery == delivery + assert agent.state.context[-1].role == "user" + assert "改查第二个方案" in agent.state.context[-1].content[0].text + + +@pytest.mark.asyncio +async def test_middleware_injects_steer_received_while_tool_was_running(monkeypatch): + delivery = {"steer_id": "s2", "message": "工具完成后，改做新的问题"} + polls = 0 + + async def take_pending_steer(run_id: str): + nonlocal polls + assert run_id == "r1" + polls += 1 + # No instruction existed when the tool started. It arrived while that + # tool was running and must be picked up before the next model call. + return None if polls == 1 else delivery + + monkeypatch.setattr(chat_steer_service, "take_pending_steer", take_pending_steer) + middleware = SteerMiddleware() + agent = SimpleNamespace(state=SimpleNamespace(run_id="r1", context=[], steer_delivery=None)) + tool_executed = False + + async def tool_handler(**_kwargs): + nonlocal tool_executed + tool_executed = True + yield ToolChunk(content=[], state=ToolResultState.SUCCESS) + + results = [item async for item in middleware.on_acting(agent, {}, tool_handler)] + assert tool_executed is True + assert results[-1].state == ToolResultState.SUCCESS + + async def model_handler(**_kwargs): + yield "next-model-started" + + model_events = [item async for item in middleware.on_reasoning(agent, {}, model_handler)] + + assert polls == 2 + assert model_events == ["next-model-started"] + assert agent.state.steer_delivery == delivery + assert agent.state.context[-1].role == "user" + assert "工具完成后，改做新的问题" in agent.state.context[-1].content[0].text + assert "上一轮工具结果已经完成" in agent.state.context[-1].content[0].text + + +@pytest.mark.asyncio +async def test_middleware_is_pass_through_without_pending_steer(monkeypatch): + async def take_pending_steer(_run_id: str): + return None + + monkeypatch.setattr(chat_steer_service, "take_pending_steer", take_pending_steer) + middleware = SteerMiddleware() + agent = SimpleNamespace(state=SimpleNamespace(run_id="r1", context=[], steer_delivery=None)) + + async def tool_handler(**_kwargs): + yield ToolChunk(content=[], state=ToolResultState.SUCCESS) + + results = [item async for item in middleware.on_acting(agent, {}, tool_handler)] + assert results[-1].state == ToolResultState.SUCCESS + + +@pytest.mark.asyncio +async def test_executor_persists_steer_at_the_stream_boundary(monkeypatch): + persisted = [] + emitted = [] + + class FakeSession: + def __enter__(self): + return self + + def __exit__(self, *_args): + return False + + class FakeChatService: + def __init__(self, _db): + pass + + def add_message(self, **kwargs): + persisted.append((kwargs["role"], kwargs["content"], kwargs["message_id"])) + + def upsert_message(self, **kwargs): + persisted.append((kwargs["role"], kwargs["content"], kwargs["message_id"])) + + async def fake_workflow(**_kwargs): + yield {"type": "content", "delta": "前半段"} + yield {"type": "steer_applied", "steer_id": "s1", "message": "改用第二种方案"} + yield {"type": "content", "delta": "后半段"} + yield {"type": "meta", "is_markdown": False, "usage": {}} + + async def fake_xadd(_run_id, _offset, event): + emitted.append(dict(event)) + + monkeypatch.setattr(executor, "SessionLocal", lambda: FakeSession()) + monkeypatch.setattr(executor, "ChatService", FakeChatService) + monkeypatch.setattr(executor, "astream_chat_workflow", fake_workflow) + monkeypatch.setattr(executor, "_xadd_event", fake_xadd) + monkeypatch.setattr(executor, "_update_run_status", lambda *_args, **_kwargs: None) + monkeypatch.setattr(executor, "_finalize_run", lambda *_args, **_kwargs: None) + monkeypatch.setattr(executor, "_spawn_followup_task", lambda **_kwargs: None) + monkeypatch.setattr(executor, "_spawn_compaction_task", lambda **_kwargs: None) + + async def fake_expire(_run_id): + return None + + monkeypatch.setattr(executor, "_expire_stream", fake_expire) + monkeypatch.setattr( + "core.services.artifact_service.persist_artifacts", + lambda *_args, **_kwargs: None, + ) + + context = {"chat_id": "c1", "user_id": "u1"} + await executor._run_workflow( + run_id="r1", + chat_id="c1", + user_id="u1", + message_id="m1", + session_messages=[], + effective_user_message="原始问题", + raw_user_message="原始问题", + context=context, + model_name="test-model", + ) + + assert [(role, content) for role, content, _ in persisted] == [ + ("assistant", "前半段"), + ("user", "改用第二种方案"), + ("assistant", "后半段"), + ] + steer_event = next(event for event in emitted if event.get("type") == "steer_applied") + assert steer_event["previous_assistant_message_id"] == "m1" + assert persisted[-1][2] == steer_event["next_assistant_message_id"] + assert context["message_id"] == steer_event["next_assistant_message_id"] diff --git a/src/backend/tests/orchestration/test_missing_tool_call_synthesis.py b/src/backend/tests/orchestration/test_missing_tool_call_synthesis.py index 84f077e1..1cdbf98b 100644 --- a/src/backend/tests/orchestration/test_missing_tool_call_synthesis.py +++ b/src/backend/tests/orchestration/test_missing_tool_call_synthesis.py @@ -66,8 +66,8 @@ def test_unknown_tool_falls_back_to_its_raw_name(): def test_persisted_log_entry_is_complete_instead_of_the_bare_fallback(): """Without the synthesized call the log entry came from attach_tool_result's - append branch — no display name, no args, and (downstream) no content_offset, - which knocked the whole message off the offset-ordered history replay.""" + append branch with no display name or args, leaving an incomplete history + card that could not be matched to the original call.""" log: list = [] upsert_tool_call(log, {"tool_name": "sibling", "tool_display_name": "同伴", "tool_args": {"a": 1}, "tool_id": "sib"}) diff --git a/src/frontend/scripts/test-chat-stream-segments.ts b/src/frontend/scripts/test-chat-stream-segments.ts index 3b89a72f..30a40f16 100644 --- a/src/frontend/scripts/test-chat-stream-segments.ts +++ b/src/frontend/scripts/test-chat-stream-segments.ts @@ -8,11 +8,21 @@ import { restoreDeferredThinkingTextFragment, } from '../src/utils/streamSegments'; import { extractCodeFromStreamingArgs } from '../src/utils/codeExecParser'; +import { buildHistorySegments } from '../src/utils/segments'; +import { getToolRunInitialOpen } from '../src/utils/toolRunState'; function tool(toolIndex: number): MessageSegment { return { type: 'tool', toolIndex }; } +{ + // Live tool details start expanded. A refreshed/history message is mounted + // as non-streaming and must start collapsed. + assert.equal(getToolRunInitialOpen(true), true); + assert.equal(getToolRunInitialOpen(false), false); + assert.equal(getToolRunInitialOpen(undefined), false); +} + { const segments: MessageSegment[] = [tool(0), { type: 'text', content: '数' }]; let deferred = deferThinkingTextFragmentBeforeTool(segments, true, undefined); @@ -108,4 +118,110 @@ function tool(toolIndex: number): MessageSegment { }); } +{ + // When there are more tool calls than persisted thinking blocks, every + // unmatched tool still happened before the final answer and must not be + // appended below it after refresh. + const toolCalls = [0, 1, 2].map((i) => ({ + id: `tool-${i}`, + name: 'demo', + status: 'success' as const, + })); + const { segments, cleanContent } = buildHistorySegments( + '分析任务最终回答', + toolCalls, + ); + + assert.deepEqual(segments?.map((segment) => segment.type), [ + 'thinking', + 'tool', + 'tool', + 'tool', + 'text', + ]); + assert.equal(cleanContent, '最终回答'); +} + +{ + // Persisted offsets are not a valid rendering coordinate: backend content + // includes reasoning markup and the live UI may merge streamed fragments. + // History must keep the visible answer as one Markdown block instead of + // splitting it around every tool call. + const phases = [ + '我来', + '帮你查找最新的自进化相关文章。首先让我确认一下当前可访问的项目。', + '当前', + '只有一个项目「agent harness」。让我在这个项目里查找自进化相关的最新文章。', + '项目里有', + '1276 篇论文。让我用多种方式检索自进化相关内容。', + ]; + const content = phases.join(''); + const toolCalls = phases.slice(0, -1).map((_, i) => ({ + id: `tool-${i}`, + name: 'demo', + status: 'success' as const, + })); + const { segments, cleanContent } = buildHistorySegments(content, toolCalls); + assert.equal(cleanContent, content); + assert.deepEqual(segments?.map((segment) => segment.type), [ + 'tool', 'tool', 'tool', 'tool', 'tool', 'text', + ]); + assert.equal(segments?.at(-1)?.content, content); +} + +{ + // Visible narration emitted between reasoning rounds must survive history + // cleanup, while still rendering as one answer block. + const toolCalls = [0, 1].map((i) => ({ + id: `thinking-tool-${i}`, + name: 'demo', + status: 'success' as const, + })); + const { segments, cleanContent } = buildHistorySegments( + '第一段思考一第二段思考二第三段', + toolCalls, + ); + + assert.equal(cleanContent, '第一段第二段第三段'); + assert.deepEqual(segments, [ + { type: 'thinking', content: '思考一' }, + tool(0), + { type: 'thinking', content: '思考二' }, + tool(1), + { type: 'text', content: '第一段第二段第三段' }, + ]); +} + +{ + // Inline-reasoning providers may omit the opening tag. Everything before + // the orphan closing tag is reasoning; only the suffix is visible body. + const { segments, cleanContent } = buildHistorySegments( + '先分析用户问题，再决定调用工具这是最终回答。', + [{ id: 'inline-tool', name: 'demo', status: 'success' }], + ); + + assert.equal(cleanContent, '这是最终回答。'); + assert.deepEqual(segments, [ + { type: 'thinking', content: '先分析用户问题，再决定调用工具' }, + tool(0), + { type: 'text', content: '这是最终回答。' }, + ]); +} + +{ + // Structured reasoning arrives through a separate reasoning field/SSE + // event and is persisted in a paired block by the backend. + const { segments, cleanContent } = buildHistorySegments( + '结构化 reasoning 字段里的内容这是结构化模型的正文。', + [{ id: 'structured-tool', name: 'demo', status: 'success' }], + ); + + assert.equal(cleanContent, '这是结构化模型的正文。'); + assert.deepEqual(segments, [ + { type: 'thinking', content: '结构化 reasoning 字段里的内容' }, + tool(0), + { type: 'text', content: '这是结构化模型的正文。' }, + ]); +} + console.log('chat stream segment tests passed'); diff --git a/src/frontend/src/App.tsx b/src/frontend/src/App.tsx index 6c6b7fb7..0eab6f65 100755 --- a/src/frontend/src/App.tsx +++ b/src/frontend/src/App.tsx @@ -334,10 +334,26 @@ export default function App() { } = useChatActions(effectiveApiUrl); // ── Streaming hook ── - const { send: rawSend, abort, handleFileSelect, removeFile, regenerate, editAndResend, resumeRunIfAny, cancelAndResumeBatch, continueLoop } = useStreaming( + const { send: rawSend, abort, activateQueuedMessage, discardQueuedMessage, handleFileSelect, removeFile, regenerate, editAndResend, resumeRunIfAny, cancelAndResumeBatch, continueLoop } = useStreaming( effectiveApiUrl, generateSummary, generateClassification, ); + // Codex-style stop shortcut: Escape only targets the chat currently visible + // in this tab. Popup/menu handlers can preventDefault first and retain their + // normal close behavior without accidentally cancelling the run. + useEffect(() => { + const handleEscape = (event: KeyboardEvent) => { + if (event.key !== 'Escape' || event.defaultPrevented || event.isComposing) return; + if (useCatalogStore.getState().panel !== 'chat') return; + const state = useChatStore.getState(); + if (!state.sendingChatIds.has(state.currentChatId)) return; + event.preventDefault(); + abort(state.currentChatId); + }; + window.addEventListener('keydown', handleEscape); + return () => window.removeEventListener('keydown', handleEscape); + }, [abort]); + // ── Fetch the project list once after login: used to resolve names for the chat-header // "project name / title" breadcrumb (sessions from the backend only carry projectId; // the project list is needed to look up the name) ── @@ -800,6 +816,8 @@ export default function App() { { + await apiRequest(`/v1/chat-runs/${encodeURIComponent(runId)}/steer`, { + method: 'POST', + headers: { 'Content-Type': 'application/json', ...chatTargetHeaders(chatId) }, + body: JSON.stringify({ steer_id: steerId, message: content }), + }); +} + +/** Withdraw an instruction that has not yet reached a tool boundary. */ +export async function withdrawChatRunSteer( + runId: string, + steerId: string, + chatId?: string, +): Promise { + const wrapped = await apiRequest( + `/v1/chat-runs/${encodeURIComponent(runId)}/steer/${encodeURIComponent(steerId)}`, + { method: 'DELETE', headers: { ...chatTargetHeaders(chatId) } }, + ); + const data = unwrapData<{ removed?: boolean }>(wrapped); + return data?.removed === true; +} + /** * Discover whether a chat has an in-flight backend run (for resume after refresh). * Returns null if no active run. diff --git a/src/frontend/src/components/chat/ChatArea.tsx b/src/frontend/src/components/chat/ChatArea.tsx index 5f665473..7e1efc9f 100644 --- a/src/frontend/src/components/chat/ChatArea.tsx +++ b/src/frontend/src/components/chat/ChatArea.tsx @@ -83,6 +83,8 @@ function BatchPanelsForChat({ chatId }: { chatId: string }) { interface ChatAreaProps { send: (text?: string) => void; abort?: () => void; + activateQueuedMessage?: (chatId?: string) => Promise; + discardQueuedMessage?: (chatId?: string) => Promise; continueLoop?: (chatId?: string) => void; exportChatRecord: (id: string) => Promise; createChatShare: ( @@ -102,7 +104,7 @@ interface ChatAreaProps { } export function ChatArea({ - send, abort, continueLoop, exportChatRecord, createChatShare, onCapabilityClick, handleFileSelect, removeFile, + send, abort, activateQueuedMessage, discardQueuedMessage, continueLoop, exportChatRecord, createChatShare, onCapabilityClick, handleFileSelect, removeFile, regenerate, editAndResend, inputRef, fileInputRef, chatListRef, messagesEndRef, }: ChatAreaProps) { @@ -354,6 +356,8 @@ export function ChatArea({ fileInputRef={fileInputRef} send={() => send()} abort={abort} + activateQueuedMessage={activateQueuedMessage} + discardQueuedMessage={discardQueuedMessage} continueLoop={continueLoop} handleFileSelect={handleFileSelect} removeFile={removeFile} @@ -589,6 +593,8 @@ export function ChatArea({ fileInputRef={fileInputRef} send={() => send()} abort={abort} + activateQueuedMessage={activateQueuedMessage} + discardQueuedMessage={discardQueuedMessage} continueLoop={continueLoop} handleFileSelect={handleFileSelect} removeFile={removeFile} diff --git a/src/frontend/src/components/chat/InputArea.tsx b/src/frontend/src/components/chat/InputArea.tsx index 7eb63fdd..3df05fe8 100644 --- a/src/frontend/src/components/chat/InputArea.tsx +++ b/src/frontend/src/components/chat/InputArea.tsx @@ -26,6 +26,7 @@ import { DropOverlay } from '../common/DropOverlay'; import LocalApprovalPill from './LocalApprovalPill'; import DeploymentSwitcher from './DeploymentSwitcher'; import { ContextGauge } from './ContextGauge'; +import { QueuedMessageCard } from './QueuedMessageCard'; import { t } from '../../i18n'; interface InputAreaProps { @@ -33,6 +34,8 @@ interface InputAreaProps { fileInputRef: React.RefObject; send: () => void; abort?: () => void; + activateQueuedMessage?: (chatId?: string) => Promise; + discardQueuedMessage?: (chatId?: string) => Promise; continueLoop?: (chatId?: string) => void; handleFileSelect: (e: React.ChangeEvent, ref: React.RefObject) => void; removeFile: (index: number) => void; @@ -189,7 +192,7 @@ function clearEditorIfOnlyBrowserEmptyNodes(editor: HTMLElement) { // ── Component ─────────────────────────────────────────────────────────── export function InputArea({ - inputRef, fileInputRef, send, abort, continueLoop, handleFileSelect, removeFile, + inputRef, fileInputRef, send, abort, activateQueuedMessage, discardQueuedMessage, continueLoop, handleFileSelect, removeFile, placeholder = t('请输入你的问题，按Enter发送，Shift+Enter换行'), mobilePlaceholder, rows: _rows = 3, @@ -205,6 +208,7 @@ export function InputArea({ activeSkill, setActiveSkill, activePlugin, setActivePlugin, activeMention, setActiveMention, planMode, setPlanMode, loopMode, setLoopMode, currentChat, enterChatMode, currentChatId, bindChatProject, unbindChatProject, + queuedMessages, updateQueuedMessage, activeRuns, } = useChatStore(); // Autonomous-loop capability bit (enabled by default): without permission the "autonomous loop" toggle is hidden const loopCapEnabled = useAuthStore((s) => s.authUser?.can_run_autonomous_loop); @@ -593,6 +597,28 @@ export function InputArea({ const showPlaceholder = !input.trim() && !activeMention && !activeSkill && !activePlugin && !isComposing; const hasAttachments = uploadedFiles.length > 0 || importedSpaceFiles.length > 0; + // A project-detail composer starts a separate chat and deliberately ignores + // the currently selected chat's run state; do not leak that chat's queue + // into this independent composer either. + const queuedMessage = forceSendMode ? undefined : queuedMessages[currentChatId]; + const canSteerQueued = !!activeRuns[currentChatId]?.runId + && !hasAttachments + && !activeSkill + && !activePlugin + && !activeMention; + + // A terminal run can race with the steer response. Never leave the card in + // an impossible "waiting for a tool boundary" state once this chat is idle. + useEffect(() => { + if (!sending && queuedMessage?.status === 'steering') { + updateQueuedMessage(currentChatId, (current) => ({ + ...current, + status: 'queued', + })); + } + }, [currentChatId, queuedMessage?.status, sending, updateQueuedMessage]); + + const showStopButton = sending && !input.trim(); // 拖文件到输入区直接作为附件上传，复用点击"浏览"的同一条 handleFileSelect 管线 // （它只读 e.target.files，合成一个最小 change 事件即可）。 @@ -609,6 +635,29 @@ export function InputArea({ {/* 项目页 composer 不显示云端/本机切换：会话在哪执行由项目本身决定（云端项目在云端、 本地项目在本机），不在项目内提供切换入口 */} {!projectComposer && } + + {queuedMessage && ( + + { void activateQueuedMessage?.(currentChatId); }} + onDelete={() => { void discardQueuedMessage?.(currentChatId); }} + onEdit={(content) => updateQueuedMessage(currentChatId, (current) => ({ + ...current, + content, + }))} + /> + + )} + {hasAttachments && (
@@ -1113,18 +1162,18 @@ export function InputArea({ ); })()} - {/* Send ↔ abort: single button + icon crossfade (button hover/active scaling is - done in CSS, motion only animates the inner icon, so they never conflict) */} + {/* While a run is active, an empty composer keeps the stop button; typing switches + back to send so Enter/click can queue a follow-up without cancelling the run. */}