feat: add terminal_input timeout hints

This commit is contained in:
JOJO 2026-02-25 15:21:35 +08:00
parent bec4902f1f
commit 12047ce237
2 changed files with 34 additions and 1 deletions

View File

@ -131,6 +131,9 @@ class TerminalManager:
# 当前活动终端 # 当前活动终端
self.active_terminal: Optional[str] = None self.active_terminal: Optional[str] = None
# 记录每个终端的连续 timeout 次数,用于给出更合适的提示
self._terminal_input_timeout_streaks: Dict[str, int] = {}
# 终端工厂(跨平台支持) # 终端工厂(跨平台支持)
self.factory = TerminalFactory() self.factory = TerminalFactory()
@ -314,6 +317,7 @@ class TerminalManager:
# 从字典中移除 # 从字典中移除
del self.terminals[session_name] del self.terminals[session_name]
self._terminal_input_timeout_streaks.pop(session_name, None)
# 如果是活动终端,切换到另一个 # 如果是活动终端,切换到另一个
if self.active_terminal == session_name: if self.active_terminal == session_name:
@ -373,6 +377,7 @@ class TerminalManager:
terminal.close() terminal.close()
del self.terminals[target_session] del self.terminals[target_session]
self._terminal_input_timeout_streaks.pop(target_session, None)
sandbox_options = self._build_sandbox_options() sandbox_options = self._build_sandbox_options()
new_terminal = PersistentTerminal( new_terminal = PersistentTerminal(
@ -562,6 +567,7 @@ class TerminalManager:
) )
result["timeout"] = base_timeout result["timeout"] = base_timeout
result["never_timeout"] = False result["never_timeout"] = False
self._apply_terminal_input_timeout_hint(target_session, result)
return result return result
# never_timeout 分支:不包装命令,不发送结束标记,不强杀进程 # never_timeout 分支:不包装命令,不发送结束标记,不强杀进程
@ -574,8 +580,27 @@ class TerminalManager:
) )
result["timeout"] = "never" result["timeout"] = "never"
result["never_timeout"] = True result["never_timeout"] = True
self._apply_terminal_input_timeout_hint(target_session, result)
return result return result
def _apply_terminal_input_timeout_hint(self, session_name: str, result: Dict) -> None:
"""在终端输入连续超时时追加提示标记(仅数值型 timeout。/ Add hint after consecutive timeouts (numeric only)."""
status = (result.get("status") or "").lower()
never_timeout = result.get("never_timeout")
if never_timeout is None:
timeout_value = result.get("timeout")
if isinstance(timeout_value, str) and timeout_value.lower() == "never":
never_timeout = True
if status == "timeout" and not never_timeout:
streak = self._terminal_input_timeout_streaks.get(session_name, 0) + 1
else:
streak = 0
self._terminal_input_timeout_streaks[session_name] = streak
if streak == 2:
result["timeout_hint"] = "suggest_adjust_timeout"
elif streak == 3:
result["timeout_hint"] = "suggest_never_timeout"
def _build_wrapped_command(self, command: str, marker: str, timeout: int) -> (str, int): def _build_wrapped_command(self, command: str, marker: str, timeout: int) -> (str, int):
""" """
构造带超时与完成标记的包装命令 构造带超时与完成标记的包装命令

View File

@ -403,7 +403,15 @@ def _plain_command_output(result_data: Dict[str, Any]) -> str:
def _format_terminal_input(result_data: Dict[str, Any]) -> str: def _format_terminal_input(result_data: Dict[str, Any]) -> str:
return _plain_command_output(result_data) text = _plain_command_output(result_data)
timeout_hint = result_data.get("timeout_hint")
if timeout_hint == "suggest_adjust_timeout":
suggestion = "请根据指令和输出结果判断是否需要提高超时时间或修改指令输入"
return f"{text}\n{suggestion}" if text else suggestion
if timeout_hint == "suggest_never_timeout":
suggestion = "请考虑设置timeout为never让终端持续执行该命令注意 在该命令彻底执行完成前该终端会被占用,处于不可输入的状态)"
return f"{text}\n{suggestion}" if text else suggestion
return text
def _format_sleep(result_data: Dict[str, Any]) -> str: def _format_sleep(result_data: Dict[str, Any]) -> str: