diff --git a/backend/api/server.py b/backend/api/server.py index 5c32ebe..5bbbf00 100644 --- a/backend/api/server.py +++ b/backend/api/server.py @@ -3426,6 +3426,16 @@ def _runtime_event_stage(event: dict[str, object]) -> str: return '等待用户 review' if event_type == 'continuation_request': return '继续补全回复' + if event_type == 'runtime_guidance_injected': + return '已吸收运行中引导' + if event_type == 'runtime_guidance_replan_before_tools': + return '引导已触发工具前重规划' + if event_type == 'runtime_guidance_interrupt_tool': + return '引导已中断当前工具' + if event_type == 'runtime_guidance_deferred_after_tool': + return '引导将在工具后重规划' + if event_type == 'runtime_guidance_error': + return '运行中引导处理失败' if event_type in { 'prompt_length_check', 'prompt_length_recovery', @@ -3449,6 +3459,11 @@ _RUN_EVENT_TYPES = { 'final_text_end', 'user_review_required', 'continuation_request', + 'runtime_guidance_injected', + 'runtime_guidance_replan_before_tools', + 'runtime_guidance_interrupt_tool', + 'runtime_guidance_deferred_after_tool', + 'runtime_guidance_error', 'prompt_length_check', 'prompt_length_recovery', 'auto_compact_summary', @@ -3478,12 +3493,19 @@ def _normalize_run_event(event: dict[str, object]) -> dict[str, Any] | None: 'reason', 'continuation_index', 'turn_index', + 'stage', + 'input_ids', + 'count', + 'tool_call_count', + 'tool_names', + 'message', 'strategy', 'projected_input_tokens', 'soft_input_limit_tokens', 'hard_input_limit_tokens', 'exceeds_hard_limit', 'exceeds_soft_limit', + 'killed_processes', } normalized = { key: _json_safe_limited(value, parent_key=key) diff --git a/docs/technical-architecture/12-runtime-guidance-queue.md b/docs/technical-architecture/12-runtime-guidance-queue.md index c896e0b..fdfd020 100644 --- a/docs/technical-architecture/12-runtime-guidance-queue.md +++ b/docs/technical-architecture/12-runtime-guidance-queue.md @@ -28,7 +28,7 @@ -> UI 展示 pending chip -> 用户可编辑、删除、引导 -> 引导: kind=guidance，绑定当前 run_id - -> Agent loop 下一轮开始前消费 guidance + -> Agent loop 在安全插入点消费 guidance -> guidance 以 display=false 的 user message 注入 model_messages ``` @@ -83,10 +83,10 @@ DELETE /api/sessions/{session_id}/input-queue/{item_id} ### Runtime 注入 -`LocalCodingAgent` 提供 `runtime_guidance_provider`，在每轮模型调用前消费 pending guidance： +`LocalCodingAgent` 提供 `runtime_guidance_provider`，在 Agent loop 的安全边界消费 pending guidance： ```text -agent loop boundary +agent loop safe boundary -> consume_session_guidance(session_id, run_id) -> append hidden user message: @@ -96,7 +96,35 @@ agent loop boundary -> 继续模型调用 ``` -这个注入点必须在 tool_result 已经写回之后、下一次模型调用之前，不能插在 tool_use 和 tool_result 中间。 +可用插入点： + +| 插入点 | 时机 | 处理策略 | +|--------|------|----------| +| `before_model` | 每次模型调用前 | 常规消费，适合上一轮工具完成后的补充 | +| `before_tools` | 模型已经给出工具计划，但工具还没开始执行 | 运行时先为上一批未执行工具补 synthetic `tool_result`，标记为 `runtime_guidance_replan`，再注入 guidance，让模型重新判断是否继续原计划、调整参数或换计划 | +| `during_tool_interrupted` | 工具已经开始执行，且用户引导明显要求停止、改目标、换参数、纠错 | 运行时给当前工具传入单工具 interrupt event，并通过 process registry 终止当前工具进程；当前工具结果落盘后注入 guidance，让模型重规划 | +| `after_tool` | 工具执行期间或刚完成后收到补充型 guidance，或工具没有中间输出无法及时中断 | 保留当前工具结果，停止继续执行同批旧工具计划，注入 guidance 让模型判断继续、补充或重跑 | +| `before_finish` | 模型准备给最终回复前 | 注入 guidance，让模型判断是修正最终输出、补充信息，还是转为后续任务 | + +约束： + +- 不把 guidance 插在 `tool_use` 和 `tool_result` 中间。 +- `before_tools` 不直接删除 assistant 的工具计划，而是补一组“未执行、被 runtime 跳过”的 tool result，保证 Anthropic/Bedrock 的消息顺序合法。 +- `during_tool_interrupted` 不复用整轮 run cancel event，而是构造“整轮取消 OR 当前工具中断”的组合 cancel event 传给当前工具，避免把用户引导误判成整轮取消。 +- 立即中断依赖工具合作：bash / python / Jupyter / 远端执行等接入 cancel_event 或 process registry 的工具可以被终止；纯同步且没有中间输出的工具只能在返回后进入 `after_tool` 重规划。 + +### 引导策略判断 + +前端不暴露复杂按钮，用户仍然只点击“引导”。系统内部按安全点自动处理： + +| 用户引导类型 | 默认策略 | +|--------------|----------| +| 工具未开始前的纠偏、改目标、改参数 | `before_tools` 重规划 | +| 工具完成后的补充要求 | `before_model` 注入下一次模型调用 | +| 即将结束前的格式、总结、补充输出要求 | `before_finish` 注入并继续一轮 | +| 已经运行中的长工具纠偏 | 明显停止/改目标/纠错类引导触发 `during_tool_interrupted`；补充输出类引导进入 `after_tool` | + +模型负责在收到 `` 后判断如何吸收：继续原计划、调整计划、说明冲突或转为后续任务；运行时只负责选择合法插入点和维护消息协议。 ## 前端交互 @@ -120,6 +148,7 @@ agent loop boundary 1. 同一个账号同时打开两个 session，分别运行任务，输入不会串台。 2. A session 运行中切到 B session 输入，B 的输入只进入 B 的 queue。 3. A session 运行中输入后刷新，pending chip 仍存在。 -4. 点击“引导”后，下一次模型调用前能消费 guidance，并在活动区记录注入事件。 -5. 点击停止后，对应 run 的进程和 pending queue 都被取消。 -6. 触发 compact 后，UI 仍能从 `display_messages` 回放完整历史；模型只使用 compact 后的 `model_messages`。 +4. 点击“引导”后，最近的安全插入点能消费 guidance，并在活动区记录注入事件。 +5. 如果 guidance 在工具执行前到达，旧工具计划不执行，并产生 `runtime_guidance_replan_before_tools` 活动事件。 +6. 点击停止后，对应 run 的进程和 pending queue 都被取消。 +7. 触发 compact 后，UI 仍能从 `display_messages` 回放完整历史；模型只使用 compact 后的 `model_messages`。 diff --git a/frontend/app/app/api/chat/route.ts b/frontend/app/app/api/chat/route.ts index 0435e5a..4dfbd44 100644 --- a/frontend/app/app/api/chat/route.ts +++ b/frontend/app/app/api/chat/route.ts @@ -441,7 +441,7 @@ async function startClawBackendRun( } } -async function consumeClawStream( +async function _consumeClawStream( body: ReadableStream, writer?: UIMessageStreamWriter, streamState?: StreamState, @@ -622,6 +622,50 @@ function writeRuntimeEvent( : "工具调用完成，等待最终结果汇总。", }); } + if (event.type === "runtime_guidance_injected") { + writeReasoningDelta( + writer, + { + type: "reasoning-delta", + id: "reasoning-1", + delta: "\n已吸收运行中引导。", + }, + streamState, + ); + } + if (event.type === "runtime_guidance_replan_before_tools") { + writeReasoningDelta( + writer, + { + type: "reasoning-delta", + id: "reasoning-1", + delta: "\n引导已触发工具前重规划。", + }, + streamState, + ); + } + if (event.type === "runtime_guidance_interrupt_tool") { + writeReasoningDelta( + writer, + { + type: "reasoning-delta", + id: "reasoning-1", + delta: "\n引导已中断当前工具，准备重规划。", + }, + streamState, + ); + } + if (event.type === "runtime_guidance_deferred_after_tool") { + writeReasoningDelta( + writer, + { + type: "reasoning-delta", + id: "reasoning-1", + delta: "\n引导将在当前工具完成后重规划。", + }, + streamState, + ); + } } function writeReasoningDelta( diff --git a/frontend/app/app/api/claw/training/pipeline/route.ts b/frontend/app/app/api/claw/training/pipeline/route.ts index 3bd10e8..a406ff0 100644 --- a/frontend/app/app/api/claw/training/pipeline/route.ts +++ b/frontend/app/app/api/claw/training/pipeline/route.ts @@ -4,7 +4,8 @@ const CLAW_API_URL = process.env.CLAW_API_URL ?? "http://127.0.0.1:8765"; export async function GET(request: Request) { const account = await getCurrentAccount(); - if (!account) return Response.json({ error: "unauthorized" }, { status: 401 }); + if (!account) + return Response.json({ error: "unauthorized" }, { status: 401 }); const incoming = new URL(request.url); const sessionId = incoming.searchParams.get("session_id") ?? ""; diff --git a/frontend/app/app/api/claw/training/step-detail/route.ts b/frontend/app/app/api/claw/training/step-detail/route.ts index 1184b30..10eb882 100644 --- a/frontend/app/app/api/claw/training/step-detail/route.ts +++ b/frontend/app/app/api/claw/training/step-detail/route.ts @@ -4,7 +4,8 @@ const CLAW_API_URL = process.env.CLAW_API_URL ?? "http://127.0.0.1:8765"; export async function GET(request: Request) { const account = await getCurrentAccount(); - if (!account) return Response.json({ error: "unauthorized" }, { status: 401 }); + if (!account) + return Response.json({ error: "unauthorized" }, { status: 401 }); const incoming = new URL(request.url); const sessionId = incoming.searchParams.get("session_id") ?? ""; diff --git a/frontend/app/components/assistant-ui/activity-panel.tsx b/frontend/app/components/assistant-ui/activity-panel.tsx index 44bd967..cafa143 100644 --- a/frontend/app/components/assistant-ui/activity-panel.tsx +++ b/frontend/app/components/assistant-ui/activity-panel.tsx @@ -1170,6 +1170,18 @@ function summarizeLiveRunEvents( if (event.type === "final_text_start") { lines.push("正在整理回复"); } + if (event.type === "runtime_guidance_injected") { + lines.push("已吸收运行中引导"); + } + if (event.type === "runtime_guidance_replan_before_tools") { + lines.push("引导已触发工具前重规划"); + } + if (event.type === "runtime_guidance_interrupt_tool") { + lines.push("引导已中断当前工具"); + } + if (event.type === "runtime_guidance_deferred_after_tool") { + lines.push("引导将在工具后重规划"); + } } if (runStatus.current_stage && lines.at(-1) !== runStatus.current_stage) { lines.push(runStatus.current_stage); diff --git a/frontend/app/components/assistant-ui/thread-list.tsx b/frontend/app/components/assistant-ui/thread-list.tsx index 21bd034..2c85971 100644 --- a/frontend/app/components/assistant-ui/thread-list.tsx +++ b/frontend/app/components/assistant-ui/thread-list.tsx @@ -1065,6 +1065,18 @@ function summarizeRunEvents(runStatus: ClawActiveRunStatus) { if (event.type === "user_review_required") { lines.push("等待用户 review"); } + if (event.type === "runtime_guidance_injected") { + lines.push("已吸收运行中引导"); + } + if (event.type === "runtime_guidance_replan_before_tools") { + lines.push("引导已触发工具前重规划"); + } + if (event.type === "runtime_guidance_interrupt_tool") { + lines.push("引导已中断当前工具"); + } + if (event.type === "runtime_guidance_deferred_after_tool") { + lines.push("引导将在工具后重规划"); + } } if (runStatus.current_stage && lines.at(-1) !== runStatus.current_stage) { lines.push(runStatus.current_stage); diff --git a/frontend/app/public/doc-assets/technical-docs/12-runtime-guidance-queue.md b/frontend/app/public/doc-assets/technical-docs/12-runtime-guidance-queue.md index c896e0b..fdfd020 100644 --- a/frontend/app/public/doc-assets/technical-docs/12-runtime-guidance-queue.md +++ b/frontend/app/public/doc-assets/technical-docs/12-runtime-guidance-queue.md @@ -28,7 +28,7 @@ -> UI 展示 pending chip -> 用户可编辑、删除、引导 -> 引导: kind=guidance，绑定当前 run_id - -> Agent loop 下一轮开始前消费 guidance + -> Agent loop 在安全插入点消费 guidance -> guidance 以 display=false 的 user message 注入 model_messages ``` @@ -83,10 +83,10 @@ DELETE /api/sessions/{session_id}/input-queue/{item_id} ### Runtime 注入 -`LocalCodingAgent` 提供 `runtime_guidance_provider`，在每轮模型调用前消费 pending guidance： +`LocalCodingAgent` 提供 `runtime_guidance_provider`，在 Agent loop 的安全边界消费 pending guidance： ```text -agent loop boundary +agent loop safe boundary -> consume_session_guidance(session_id, run_id) -> append hidden user message: @@ -96,7 +96,35 @@ agent loop boundary -> 继续模型调用 ``` -这个注入点必须在 tool_result 已经写回之后、下一次模型调用之前，不能插在 tool_use 和 tool_result 中间。 +可用插入点： + +| 插入点 | 时机 | 处理策略 | +|--------|------|----------| +| `before_model` | 每次模型调用前 | 常规消费，适合上一轮工具完成后的补充 | +| `before_tools` | 模型已经给出工具计划，但工具还没开始执行 | 运行时先为上一批未执行工具补 synthetic `tool_result`，标记为 `runtime_guidance_replan`，再注入 guidance，让模型重新判断是否继续原计划、调整参数或换计划 | +| `during_tool_interrupted` | 工具已经开始执行，且用户引导明显要求停止、改目标、换参数、纠错 | 运行时给当前工具传入单工具 interrupt event，并通过 process registry 终止当前工具进程；当前工具结果落盘后注入 guidance，让模型重规划 | +| `after_tool` | 工具执行期间或刚完成后收到补充型 guidance，或工具没有中间输出无法及时中断 | 保留当前工具结果，停止继续执行同批旧工具计划，注入 guidance 让模型判断继续、补充或重跑 | +| `before_finish` | 模型准备给最终回复前 | 注入 guidance，让模型判断是修正最终输出、补充信息，还是转为后续任务 | + +约束： + +- 不把 guidance 插在 `tool_use` 和 `tool_result` 中间。 +- `before_tools` 不直接删除 assistant 的工具计划，而是补一组“未执行、被 runtime 跳过”的 tool result，保证 Anthropic/Bedrock 的消息顺序合法。 +- `during_tool_interrupted` 不复用整轮 run cancel event，而是构造“整轮取消 OR 当前工具中断”的组合 cancel event 传给当前工具，避免把用户引导误判成整轮取消。 +- 立即中断依赖工具合作：bash / python / Jupyter / 远端执行等接入 cancel_event 或 process registry 的工具可以被终止；纯同步且没有中间输出的工具只能在返回后进入 `after_tool` 重规划。 + +### 引导策略判断 + +前端不暴露复杂按钮，用户仍然只点击“引导”。系统内部按安全点自动处理： + +| 用户引导类型 | 默认策略 | +|--------------|----------| +| 工具未开始前的纠偏、改目标、改参数 | `before_tools` 重规划 | +| 工具完成后的补充要求 | `before_model` 注入下一次模型调用 | +| 即将结束前的格式、总结、补充输出要求 | `before_finish` 注入并继续一轮 | +| 已经运行中的长工具纠偏 | 明显停止/改目标/纠错类引导触发 `during_tool_interrupted`；补充输出类引导进入 `after_tool` | + +模型负责在收到 `` 后判断如何吸收：继续原计划、调整计划、说明冲突或转为后续任务；运行时只负责选择合法插入点和维护消息协议。 ## 前端交互 @@ -120,6 +148,7 @@ agent loop boundary 1. 同一个账号同时打开两个 session，分别运行任务，输入不会串台。 2. A session 运行中切到 B session 输入，B 的输入只进入 B 的 queue。 3. A session 运行中输入后刷新，pending chip 仍存在。 -4. 点击“引导”后，下一次模型调用前能消费 guidance，并在活动区记录注入事件。 -5. 点击停止后，对应 run 的进程和 pending queue 都被取消。 -6. 触发 compact 后，UI 仍能从 `display_messages` 回放完整历史；模型只使用 compact 后的 `model_messages`。 +4. 点击“引导”后，最近的安全插入点能消费 guidance，并在活动区记录注入事件。 +5. 如果 guidance 在工具执行前到达，旧工具计划不执行，并产生 `runtime_guidance_replan_before_tools` 活动事件。 +6. 点击停止后，对应 run 的进程和 pending queue 都被取消。 +7. 触发 compact 后，UI 仍能从 `display_messages` 回放完整历史；模型只使用 compact 后的 `model_messages`。 diff --git a/src/agent_runtime.py b/src/agent_runtime.py index 81a5bfe..f1caf83 100644 --- a/src/agent_runtime.py +++ b/src/agent_runtime.py @@ -6,6 +6,7 @@ from datetime import datetime, timezone import json from pathlib import Path import threading +import time from typing import Any, Callable from uuid import uuid4 @@ -269,6 +270,45 @@ def _all_skill_tool_call_ids(transcript: tuple[dict[str, object], ...]) -> list[ return results +class _CombinedCancelEvent: + """Expose a single cancel flag backed by run cancel plus per-tool interrupt.""" + + def __init__( + self, + run_cancel_event: Any | None, + tool_interrupt_event: threading.Event, + ) -> None: + self._run_cancel_event = run_cancel_event + self._tool_interrupt_event = tool_interrupt_event + + def is_set(self) -> bool: + run_cancelled = ( + self._run_cancel_event is not None + and bool(self._run_cancel_event.is_set()) + ) + return run_cancelled or self._tool_interrupt_event.is_set() + + def set(self) -> None: + self._tool_interrupt_event.set() + + def wait(self, timeout: float | None = None) -> bool: + if self.is_set(): + return True + if self._run_cancel_event is None: + return self._tool_interrupt_event.wait(timeout) + if timeout is None: + while not self.is_set(): + self._tool_interrupt_event.wait(0.1) + return True + deadline = time.monotonic() + max(0.0, timeout) + while not self.is_set(): + remaining = deadline - time.monotonic() + if remaining <= 0: + return self.is_set() + self._tool_interrupt_event.wait(min(0.1, remaining)) + return True + + @dataclass class LocalCodingAgent: model_config: ModelConfig @@ -739,6 +779,7 @@ class LocalCodingAgent: session, stream_events, turn_index=turn_index, + stage='before_model', ) self._microcompact_session_if_needed( session, @@ -1036,6 +1077,32 @@ class LocalCodingAgent: self.last_run_result = result return result + if turn.tool_calls: + guidance_items = self._consume_runtime_guidance( + stream_events, + turn_index=turn_index, + stage='before_tools', + ) + if guidance_items: + self._append_skipped_tool_results_for_guidance( + session, + stream_events, + turn=turn, + turn_index=turn_index, + guidance_items=guidance_items, + ) + self._append_runtime_guidance_items( + session, + stream_events, + guidance_items, + turn_index=turn_index, + stage='before_tools', + ) + assistant_response_segments.clear() + continuation_count = 0 + last_content = turn.content + continue + if not turn.tool_calls: assistant_response_segments.append(turn.content) if self._is_empty_truncated_response(turn): @@ -1092,6 +1159,16 @@ class LocalCodingAgent: ) last_content = ''.join(assistant_response_segments) continue + if self._inject_runtime_guidance( + session, + stream_events, + turn_index=turn_index, + stage='before_finish', + ): + assistant_response_segments.clear() + continuation_count = 0 + last_content = turn.content + continue final_output = ''.join(assistant_response_segments) if self._reached_auto_continuation_limit( turn, @@ -1130,6 +1207,7 @@ class LocalCodingAgent: self.last_run_result = result return result + replan_after_tool_guidance = False for tool_call in turn.tool_calls: assistant_response_segments.clear() continuation_count = 0 @@ -1176,6 +1254,8 @@ class LocalCodingAgent: self.last_run_result = result return result tool_result = None + tool_guidance_items: tuple[dict[str, object], ...] = () + tool_guidance_interrupted = False tool_message_index = session.start_tool( name=tool_call.name, tool_call_id=tool_call.id, @@ -1279,11 +1359,19 @@ class LocalCodingAgent: duplicate_skill_counts=duplicate_skill_counts, ) elif tool_result is None: + tool_interrupt_event = threading.Event() + tool_context = replace( + self.tool_context, + cancel_event=_CombinedCancelEvent( + self.tool_context.cancel_event, + tool_interrupt_event, + ), + ) for update in execute_tool_streaming( self.tool_registry, tool_call.name, tool_call.arguments, - self.tool_context, + tool_context, ): if update.kind == 'delta': session.append_tool_delta( @@ -1301,8 +1389,81 @@ class LocalCodingAgent: 'delta': update.content, } ) + if not tool_guidance_items: + pending_guidance = self._consume_runtime_guidance( + stream_events, + turn_index=turn_index, + stage='during_tool', + ) + if pending_guidance: + tool_guidance_items = pending_guidance + if self._runtime_guidance_should_interrupt_tool( + tool_call=tool_call, + guidance_items=pending_guidance, + ): + tool_guidance_interrupted = True + tool_interrupt_event.set() + self._interrupt_running_tool_for_guidance( + stream_events, + turn_index=turn_index, + tool_call=tool_call, + guidance_items=pending_guidance, + ) + else: + stream_events.append( + { + 'type': 'runtime_guidance_deferred_after_tool', + 'turn_index': turn_index, + 'tool_name': tool_call.name, + 'tool_call_id': tool_call.id, + 'input_ids': self._runtime_guidance_item_ids( + pending_guidance + ), + 'message': ( + '用户引导在工具执行中到达，运行时判断无需中断，' + '将等待当前工具完成后重规划' + ), + } + ) continue tool_result = update.result + if tool_result is not None and not tool_guidance_items: + tool_guidance_items = self._consume_runtime_guidance( + stream_events, + turn_index=turn_index, + stage='after_tool', + ) + if tool_guidance_items: + stream_events.append( + { + 'type': 'runtime_guidance_deferred_after_tool', + 'turn_index': turn_index, + 'tool_name': tool_call.name, + 'tool_call_id': tool_call.id, + 'input_ids': self._runtime_guidance_item_ids( + tool_guidance_items + ), + 'message': ( + '用户引导在工具执行期间到达，但当前工具未提供中间输出；' + '已在工具完成后准备重规划' + ), + } + ) + if tool_result is not None and tool_guidance_items: + merged_metadata = dict(tool_result.metadata) + merged_metadata['runtime_guidance_input_ids'] = ( + self._runtime_guidance_item_ids(tool_guidance_items) + ) + if tool_guidance_interrupted: + merged_metadata['runtime_guidance_interrupted_tool'] = True + else: + merged_metadata['runtime_guidance_deferred_after_tool'] = True + tool_result = ToolExecutionResult( + name=tool_result.name, + ok=tool_result.ok, + content=tool_result.content, + metadata=merged_metadata, + ) if tool_result is None: raise RuntimeError(f'Tool executor returned no final result for {tool_call.name}') if ( @@ -1474,6 +1635,23 @@ class LocalCodingAgent: ) if history_entry is not None: file_history.append(history_entry) + if tool_guidance_items: + self._append_runtime_guidance_items( + session, + stream_events, + tool_guidance_items, + turn_index=turn_index, + stage=( + 'during_tool_interrupted' + if tool_guidance_interrupted + else 'after_tool' + ), + ) + assistant_response_segments.clear() + continuation_count = 0 + last_content = turn.content + replan_after_tool_guidance = True + break if tool_result.metadata.get('stop_agent') is True: stop_reason = str( tool_result.metadata.get('stop_reason') or 'tool_requested_stop' @@ -1561,6 +1739,9 @@ class LocalCodingAgent: self.last_run_result = result return result + if replan_after_tool_guidance: + continue + final_output = self._build_max_turns_output( max_turns=self.runtime_config.max_turns, last_content=last_content, @@ -4286,10 +4467,34 @@ class LocalCodingAgent: stream_events: list[dict[str, object]], *, turn_index: int, - ) -> None: + stage: str, + ) -> bool: + items = self._consume_runtime_guidance( + stream_events, + turn_index=turn_index, + stage=stage, + ) + if not items: + return False + self._append_runtime_guidance_items( + session, + stream_events, + items, + turn_index=turn_index, + stage=stage, + ) + return True + + def _consume_runtime_guidance( + self, + stream_events: list[dict[str, object]], + *, + turn_index: int, + stage: str, + ) -> tuple[dict[str, object], ...]: provider = self.runtime_guidance_provider if provider is None: - return + return () try: items = tuple( item @@ -4302,12 +4507,22 @@ class LocalCodingAgent: { 'type': 'runtime_guidance_error', 'turn_index': turn_index, + 'stage': stage, 'error': str(exc)[:500], } ) - return - if not items: - return + return () + return items + + def _append_runtime_guidance_items( + self, + session: AgentSessionState, + stream_events: list[dict[str, object]], + items: tuple[dict[str, object], ...], + *, + turn_index: int, + stage: str, + ) -> None: lines = [ '', ( @@ -4315,6 +4530,26 @@ class LocalCodingAgent: '吸收这些要求；如果它们与当前目标冲突，先说明冲突并选择最合理的继续方式。' ), ] + if stage == 'before_tools': + lines.append( + '这批引导在工具执行前到达，运行时已经跳过上一批尚未执行的工具调用。' + '请重新判断：继续原计划、调整工具参数、改走新计划，或说明冲突。' + ) + elif stage == 'before_finish': + lines.append( + '这批引导在最终回复前到达。请判断它是补充最终输出、修正当前结论，' + '还是应作为后续任务处理。' + ) + elif stage == 'during_tool_interrupted': + lines.append( + '这批引导在工具执行中到达，运行时已经中断当前工具并停止继续执行同批旧工具计划。' + '请结合已有工具输出和这批引导重新规划；如果需要重跑工具，请使用新的参数。' + ) + elif stage == 'after_tool': + lines.append( + '这批引导在工具执行期间或刚完成后到达。当前工具结果已保留，' + '运行时停止继续执行同批旧工具计划，请判断是否基于该结果继续、补充或重跑。' + ) item_ids: list[int] = [] for index, item in enumerate(items, start=1): item_id = item.get('id') @@ -4328,6 +4563,7 @@ class LocalCodingAgent: metadata={ 'kind': 'runtime_guidance', 'turn_index': turn_index, + 'stage': stage, 'queue_item_ids': item_ids, }, message_id=f'runtime_guidance_{turn_index}_{uuid4().hex[:8]}', @@ -4337,12 +4573,174 @@ class LocalCodingAgent: { 'type': 'runtime_guidance_injected', 'turn_index': turn_index, + 'stage': stage, 'input_ids': item_ids, 'count': len(items), 'message': f'已将 {len(items)} 条用户引导注入当前任务', } ) + @staticmethod + def _runtime_guidance_item_ids( + items: tuple[dict[str, object], ...], + ) -> list[int]: + return [ + int(item.get('id')) + for item in items + if isinstance(item.get('id'), int) + and not isinstance(item.get('id'), bool) + ] + + @staticmethod + def _runtime_guidance_should_interrupt_tool( + *, + tool_call: ToolCall, + guidance_items: tuple[dict[str, object], ...], + ) -> bool: + combined = '\n'.join( + str(item.get('content') or '').strip().lower() + for item in guidance_items + ) + if not combined: + return False + defer_markers = ( + '完成后', + '结束后', + '最后', + '顺便', + '同时整理', + '补充输出', + 'finish', + 'after', + 'then', + 'also', + ) + interrupt_markers = ( + '停', + '中断', + '取消', + '暂停', + '别跑', + '别继续', + '不要继续', + '先别', + '错了', + '不对', + '改成', + '换成', + '重新', + '重来', + '不用', + '不是', + '别用', + 'stop', + 'cancel', + 'interrupt', + 'abort', + 'wrong', + 'instead', + 'change to', + 'switch to', + 'restart', + 'redo', + ) + if any(marker in combined for marker in interrupt_markers): + return True + if any(marker in combined for marker in defer_markers): + return False + return tool_call.name in { + 'bash', + 'python_exec', + 'jupyter_exec', + 'remote_bash', + 'remote_python_exec', + 'Agent', + 'delegate_agent', + } and len(combined) >= 120 + + def _interrupt_running_tool_for_guidance( + self, + stream_events: list[dict[str, object]], + *, + turn_index: int, + tool_call: ToolCall, + guidance_items: tuple[dict[str, object], ...], + ) -> None: + process_registry = self.tool_context.process_registry + killed_processes = False + if process_registry is not None: + try: + process_registry.kill_all() + killed_processes = True + except Exception as exc: # noqa: BLE001 + stream_events.append( + { + 'type': 'runtime_guidance_error', + 'turn_index': turn_index, + 'stage': 'during_tool', + 'tool_name': tool_call.name, + 'tool_call_id': tool_call.id, + 'error': str(exc)[:500], + } + ) + stream_events.append( + { + 'type': 'runtime_guidance_interrupt_tool', + 'turn_index': turn_index, + 'tool_name': tool_call.name, + 'tool_call_id': tool_call.id, + 'input_ids': self._runtime_guidance_item_ids(guidance_items), + 'killed_processes': killed_processes, + 'message': '用户引导要求调整当前任务，已中断当前工具并准备重规划', + } + ) + + def _append_skipped_tool_results_for_guidance( + self, + session: AgentSessionState, + stream_events: list[dict[str, object]], + *, + turn: AssistantTurn, + turn_index: int, + guidance_items: tuple[dict[str, object], ...], + ) -> None: + input_ids = [ + int(item.get('id')) + for item in guidance_items + if isinstance(item.get('id'), int) + and not isinstance(item.get('id'), bool) + ] + for tool_call in turn.tool_calls: + tool_result = ToolExecutionResult( + name=tool_call.name, + ok=False, + content=( + '工具尚未执行时收到了用户运行中引导。运行时跳过了这次旧工具调用，' + '请结合后续 runtime guidance 重新规划。' + ), + metadata={ + 'action': 'runtime_guidance_replan', + 'runtime_guidance_skipped': True, + 'runtime_guidance_input_ids': input_ids, + }, + ) + session.append_tool( + tool_call.name, + tool_call.id, + serialize_tool_result(tool_result), + display=False, + ) + stream_events.append( + { + 'type': 'runtime_guidance_replan_before_tools', + 'turn_index': turn_index, + 'input_ids': input_ids, + 'tool_call_count': len(turn.tool_calls), + 'tool_names': [tool_call.name for tool_call in turn.tool_calls], + 'message': '用户引导在工具执行前到达，已跳过旧工具计划并要求模型重规划', + } + ) + def render_system_prompt(self) -> str: prompt_context = self.build_prompt_context() parts = self.build_system_prompt_parts(prompt_context) diff --git a/tests/test_agent_runtime.py b/tests/test_agent_runtime.py index 906fdcc..b2ef871 100644 --- a/tests/test_agent_runtime.py +++ b/tests/test_agent_runtime.py @@ -7,6 +7,7 @@ import unittest from pathlib import Path from unittest.mock import patch +from src.agent_tool_core import ToolStreamUpdate from src.agent_session import AgentMessage from src.agent_runtime import LocalCodingAgent, MAX_AUTO_CONTINUATIONS from src.agent_tools import build_tool_context, default_tool_registry, execute_tool @@ -16,6 +17,7 @@ from src.agent_types import ( BudgetConfig, ModelConfig, OutputSchemaConfig, + ToolExecutionResult, UsageStats, ) from src.compact import CompactionResult @@ -282,9 +284,324 @@ class AgentRuntimeTests(unittest.TestCase): self.assertEqual(result.final_output, 'The file contains hello world.') self.assertEqual(result.tool_calls, 1) - self.assertGreaterEqual(len(result.transcript), 5) + transcript_roles = [entry.get('role') for entry in result.transcript] + self.assertEqual(transcript_roles, ['user', 'assistant', 'tool', 'assistant']) + self.assertIn('I will inspect the file first.', result.transcript[1]['content']) + self.assertIn('hello world', result.transcript[2]['content']) + self.assertIn('The file contains hello world.', result.transcript[3]['content']) self.assertGreaterEqual(len(result.file_history), 0) + def test_runtime_guidance_before_tools_skips_stale_tool_plan(self) -> None: + responses = [ + { + 'choices': [ + { + 'message': { + 'role': 'assistant', + 'content': 'I will inspect the old file.', + 'tool_calls': [ + { + 'id': 'call_1', + 'type': 'function', + 'function': { + 'name': 'read_file', + 'arguments': '{"path": "old.txt"}', + }, + } + ], + }, + 'finish_reason': 'tool_calls', + } + ] + }, + { + 'choices': [ + { + 'message': { + 'role': 'assistant', + 'content': 'Replanned with the guidance.', + }, + 'finish_reason': 'stop', + } + ] + }, + ] + recorded_payloads: list[dict[str, object]] = [] + provider_calls = 0 + + def guidance_provider() -> tuple[dict[str, object], ...]: + nonlocal provider_calls + provider_calls += 1 + if provider_calls == 2: + return ({'id': 7, 'content': 'Use new.txt instead.'},) + return () + + with tempfile.TemporaryDirectory() as tmp_dir: + workspace = Path(tmp_dir) + (workspace / 'old.txt').write_text('old content\n', encoding='utf-8') + with patch( + 'src.openai_compat.request.urlopen', + side_effect=make_recording_urlopen_side_effect(responses, recorded_payloads), + ): + agent = LocalCodingAgent( + model_config=ModelConfig( + model='Qwen/Qwen3-Coder-30B-A3B-Instruct', + base_url='http://127.0.0.1:8000/v1', + ), + runtime_config=AgentRuntimeConfig(cwd=workspace), + ) + agent.runtime_guidance_provider = guidance_provider + result = agent.run('Inspect old.txt') + + self.assertEqual(result.final_output, 'Replanned with the guidance.') + self.assertEqual(result.tool_calls, 0) + self.assertFalse(result.file_history) + replan_events = [ + event + for event in result.events + if event.get('type') == 'runtime_guidance_replan_before_tools' + ] + self.assertEqual(len(replan_events), 1) + injected_events = [ + event + for event in result.events + if event.get('type') == 'runtime_guidance_injected' + ] + self.assertEqual(injected_events[-1].get('stage'), 'before_tools') + self.assertEqual(len(recorded_payloads), 2) + second_messages = recorded_payloads[1]['messages'] + assert isinstance(second_messages, list) + roles = [ + message.get('role') + for message in second_messages + if isinstance(message, dict) + ] + self.assertEqual(roles[-3:], ['assistant', 'tool', 'user']) + self.assertIn('runtime_guidance_replan', str(second_messages[-2])) + self.assertIn('Use new.txt instead.', str(second_messages[-1])) + + def test_runtime_guidance_interrupts_running_streaming_tool(self) -> None: + responses = [ + { + 'choices': [ + { + 'message': { + 'role': 'assistant', + 'content': 'I will run the long tool.', + 'tool_calls': [ + { + 'id': 'call_1', + 'type': 'function', + 'function': { + 'name': 'long_tool', + 'arguments': '{}', + }, + } + ], + }, + 'finish_reason': 'tool_calls', + } + ] + }, + { + 'choices': [ + { + 'message': { + 'role': 'assistant', + 'content': 'Replanned after the interruption.', + }, + 'finish_reason': 'stop', + } + ] + }, + ] + recorded_payloads: list[dict[str, object]] = [] + provider_calls = 0 + cancel_observed_after_guidance: list[bool] = [] + + def guidance_provider() -> tuple[dict[str, object], ...]: + nonlocal provider_calls + provider_calls += 1 + if provider_calls == 3: + return ({'id': 12, 'content': '停一下，改成只生成 3 条。'},) + return () + + def fake_execute_tool_streaming(registry, name, arguments, context): # noqa: ANN001 + self.assertEqual(name, 'long_tool') + yield ToolStreamUpdate(kind='delta', content='working\n', stream='stdout') + cancel_observed_after_guidance.append(bool(context.cancel_event.is_set())) + yield ToolStreamUpdate( + kind='result', + result=ToolExecutionResult( + name=name, + ok=False, + content='Tool saw cancellation.', + metadata={'cancel_seen': bool(context.cancel_event.is_set())}, + ), + ) + + with tempfile.TemporaryDirectory() as tmp_dir: + workspace = Path(tmp_dir) + with ( + patch( + 'src.openai_compat.request.urlopen', + side_effect=make_recording_urlopen_side_effect( + responses, + recorded_payloads, + ), + ), + patch( + 'src.agent_runtime.execute_tool_streaming', + side_effect=fake_execute_tool_streaming, + ), + ): + agent = LocalCodingAgent( + model_config=ModelConfig( + model='Qwen/Qwen3-Coder-30B-A3B-Instruct', + base_url='http://127.0.0.1:8000/v1', + ), + runtime_config=AgentRuntimeConfig(cwd=workspace), + ) + agent.runtime_guidance_provider = guidance_provider + result = agent.run('Run a long tool') + + self.assertEqual(result.final_output, 'Replanned after the interruption.') + self.assertEqual(result.tool_calls, 1) + self.assertEqual(cancel_observed_after_guidance, [True]) + interrupt_events = [ + event + for event in result.events + if event.get('type') == 'runtime_guidance_interrupt_tool' + ] + self.assertEqual(len(interrupt_events), 1) + injected_events = [ + event + for event in result.events + if event.get('type') == 'runtime_guidance_injected' + ] + self.assertEqual(injected_events[-1].get('stage'), 'during_tool_interrupted') + self.assertEqual(len(recorded_payloads), 2) + second_messages = recorded_payloads[1]['messages'] + assert isinstance(second_messages, list) + roles = [ + message.get('role') + for message in second_messages + if isinstance(message, dict) + ] + self.assertEqual(roles[-3:], ['assistant', 'tool', 'user']) + self.assertIn('runtime_guidance_interrupted_tool', str(second_messages[-2])) + self.assertIn('改成只生成 3 条', str(second_messages[-1])) + + def test_runtime_guidance_during_tool_can_defer_until_tool_result(self) -> None: + responses = [ + { + 'choices': [ + { + 'message': { + 'role': 'assistant', + 'content': 'I will run the long tool.', + 'tool_calls': [ + { + 'id': 'call_1', + 'type': 'function', + 'function': { + 'name': 'long_tool', + 'arguments': '{}', + }, + } + ], + }, + 'finish_reason': 'tool_calls', + } + ] + }, + { + 'choices': [ + { + 'message': { + 'role': 'assistant', + 'content': 'Replanned after the tool result.', + }, + 'finish_reason': 'stop', + } + ] + }, + ] + recorded_payloads: list[dict[str, object]] = [] + provider_calls = 0 + cancel_observed_after_guidance: list[bool] = [] + + def guidance_provider() -> tuple[dict[str, object], ...]: + nonlocal provider_calls + provider_calls += 1 + if provider_calls == 3: + return ({'id': 13, 'content': '完成后顺便整理成 Markdown。'},) + return () + + def fake_execute_tool_streaming(registry, name, arguments, context): # noqa: ANN001 + self.assertEqual(name, 'long_tool') + yield ToolStreamUpdate(kind='delta', content='working\n', stream='stdout') + cancel_observed_after_guidance.append(bool(context.cancel_event.is_set())) + yield ToolStreamUpdate( + kind='result', + result=ToolExecutionResult( + name=name, + ok=True, + content='Tool completed.', + metadata={'cancel_seen': bool(context.cancel_event.is_set())}, + ), + ) + + with tempfile.TemporaryDirectory() as tmp_dir: + workspace = Path(tmp_dir) + with ( + patch( + 'src.openai_compat.request.urlopen', + side_effect=make_recording_urlopen_side_effect( + responses, + recorded_payloads, + ), + ), + patch( + 'src.agent_runtime.execute_tool_streaming', + side_effect=fake_execute_tool_streaming, + ), + ): + agent = LocalCodingAgent( + model_config=ModelConfig( + model='Qwen/Qwen3-Coder-30B-A3B-Instruct', + base_url='http://127.0.0.1:8000/v1', + ), + runtime_config=AgentRuntimeConfig(cwd=workspace), + ) + agent.runtime_guidance_provider = guidance_provider + result = agent.run('Run a long tool') + + self.assertEqual(result.final_output, 'Replanned after the tool result.') + self.assertEqual(cancel_observed_after_guidance, [False]) + deferred_events = [ + event + for event in result.events + if event.get('type') == 'runtime_guidance_deferred_after_tool' + ] + self.assertEqual(len(deferred_events), 1) + interrupt_events = [ + event + for event in result.events + if event.get('type') == 'runtime_guidance_interrupt_tool' + ] + self.assertFalse(interrupt_events) + injected_events = [ + event + for event in result.events + if event.get('type') == 'runtime_guidance_injected' + ] + self.assertEqual(injected_events[-1].get('stage'), 'after_tool') + second_messages = recorded_payloads[1]['messages'] + assert isinstance(second_messages, list) + self.assertIn('runtime_guidance_deferred_after_tool', str(second_messages[-2])) + self.assertIn('完成后顺便整理成 Markdown', str(second_messages[-1])) + def test_agent_persists_max_turns_explanation_after_tool_result(self) -> None: responses = [ { @@ -331,6 +648,72 @@ class AgentRuntimeTests(unittest.TestCase): self.assertEqual(result.transcript[-1]['stop_reason'], 'max_turns') self.assertIn('最大步骤上限', result.transcript[-1]['content']) + def test_runtime_guidance_before_finish_continues_instead_of_finishing(self) -> None: + responses = [ + { + 'choices': [ + { + 'message': { + 'role': 'assistant', + 'content': 'Draft answer.', + }, + 'finish_reason': 'stop', + } + ] + }, + { + 'choices': [ + { + 'message': { + 'role': 'assistant', + 'content': 'Revised final answer.', + }, + 'finish_reason': 'stop', + } + ] + }, + ] + recorded_payloads: list[dict[str, object]] = [] + provider_calls = 0 + + def guidance_provider() -> tuple[dict[str, object], ...]: + nonlocal provider_calls + provider_calls += 1 + if provider_calls == 2: + return ({'id': 9, 'content': 'Add the timing summary before finalizing.'},) + return () + + with tempfile.TemporaryDirectory() as tmp_dir: + workspace = Path(tmp_dir) + with patch( + 'src.openai_compat.request.urlopen', + side_effect=make_recording_urlopen_side_effect(responses, recorded_payloads), + ): + agent = LocalCodingAgent( + model_config=ModelConfig( + model='Qwen/Qwen3-Coder-30B-A3B-Instruct', + base_url='http://127.0.0.1:8000/v1', + ), + runtime_config=AgentRuntimeConfig(cwd=workspace), + ) + agent.runtime_guidance_provider = guidance_provider + result = agent.run('Prepare an answer') + + self.assertEqual(result.final_output, 'Revised final answer.') + injected_events = [ + event + for event in result.events + if event.get('type') == 'runtime_guidance_injected' + ] + self.assertEqual(injected_events[-1].get('stage'), 'before_finish') + self.assertEqual(len(recorded_payloads), 2) + second_messages = recorded_payloads[1]['messages'] + assert isinstance(second_messages, list) + self.assertEqual(second_messages[-2].get('role'), 'assistant') + self.assertEqual(second_messages[-2].get('content'), 'Draft answer.') + self.assertEqual(second_messages[-1].get('role'), 'user') + self.assertIn('Add the timing summary', str(second_messages[-1].get('content'))) + def test_agent_stops_after_data_agent_goal_requires_review(self) -> None: responses = [ {