JOJO
0aa37b83ee
1. lifecycle.ts handleTaskStopped: 移除'再次按下停止按钮以结束后台任务'弹窗。 后台任务停止不再走停止按钮二次点击路径。 2. 主对话空闲+后台任务跑时输入栏依然为停止按钮: 根因 - App.vue 传给 InputComposer 的 streaming-message 用的是 composerStreamingForInput (= composerBusy && !displayLockEngaged)， 只要有后台任务 taskInProgress=true 就让 streamingMessage=true。 改为传 streamingUi (= streamingMessage || hasPendingToolActions)， 只看主对话自身是否在流式输出，与后台任务无关。 3. 后端 chat_flow_task_main.py 用户主动停止时发的 task_stopped 事件没带 has_running_* 字段，导致前端误判为完全空闲。 补全字段（has_running_sub_agents/has_running_background_commands/ has_running_multi_agent），前端用此信息决定是否保持对话运行态。
2516 lines
108 KiB
Python
2516 lines
108 KiB
Python
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import time
|
||
import re
|
||
import zipfile
|
||
from collections import defaultdict, Counter, deque
|
||
from datetime import datetime, timedelta
|
||
from pathlib import Path
|
||
from typing import Any, Dict, List, Optional, Tuple
|
||
|
||
from werkzeug.utils import secure_filename
|
||
|
||
from config import (
|
||
OUTPUT_FORMATS,
|
||
AUTO_FIX_TOOL_CALL,
|
||
AUTO_FIX_MAX_ATTEMPTS,
|
||
MAX_ITERATIONS_PER_TASK,
|
||
MAX_CONSECUTIVE_SAME_TOOL,
|
||
MAX_TOTAL_TOOL_CALLS,
|
||
TOOL_CALL_COOLDOWN,
|
||
MAX_UPLOAD_SIZE,
|
||
DEFAULT_CONVERSATIONS_LIMIT,
|
||
MAX_CONVERSATIONS_LIMIT,
|
||
CONVERSATIONS_DIR,
|
||
DEFAULT_RESPONSE_MAX_TOKENS,
|
||
DEFAULT_PROJECT_PATH,
|
||
LOGS_DIR,
|
||
AGENT_VERSION,
|
||
THINKING_FAST_INTERVAL,
|
||
PROJECT_MAX_STORAGE_MB,
|
||
PROJECT_MAX_STORAGE_BYTES,
|
||
UPLOAD_SCAN_LOG_SUBDIR,
|
||
)
|
||
from modules.personalization_manager import (
|
||
load_personalization_config,
|
||
save_personalization_config,
|
||
THINKING_INTERVAL_MIN,
|
||
THINKING_INTERVAL_MAX,
|
||
resolve_context_compression_settings,
|
||
)
|
||
from modules.skill_hint_manager import SkillHintManager
|
||
from modules.upload_security import UploadSecurityError
|
||
from modules.user_manager import UserWorkspace
|
||
from modules.usage_tracker import QUOTA_DEFAULTS
|
||
from modules.sub_agent import TERMINAL_STATUSES
|
||
from modules.multi_agent.debug_logger import ma_debug
|
||
from modules.versioning_manager import ConversationVersioningManager, VersioningError
|
||
from modules.shallow_versioning import ShallowVersioningManager
|
||
from core.web_terminal import WebTerminal
|
||
from utils.tool_result_formatter import format_tool_result_for_context
|
||
from utils.conversation_manager import ConversationManager
|
||
from config.model_profiles import get_model_context_window, get_model_profile
|
||
|
||
from .auth_helpers import api_login_required, resolve_admin_policy, get_current_user_record, get_current_username
|
||
from .context import with_terminal, get_gui_manager, get_upload_guard, build_upload_error_response, ensure_conversation_loaded, reset_system_state, get_user_resources, get_or_create_usage_tracker
|
||
from .work_timer import finalize_conversation_work_timer
|
||
from .utils_common import (
|
||
build_review_lines,
|
||
debug_log,
|
||
log_backend_chunk,
|
||
log_frontend_chunk,
|
||
log_streaming_debug_entry,
|
||
brief_log,
|
||
DEBUG_LOG_FILE,
|
||
CHUNK_BACKEND_LOG_FILE,
|
||
CHUNK_FRONTEND_LOG_FILE,
|
||
STREAMING_DEBUG_LOG_FILE,
|
||
)
|
||
from .security import rate_limited, compact_web_search_result, consume_socket_token, prune_socket_tokens, validate_csrf_request, requires_csrf_protection, get_csrf_token
|
||
from .monitor import cache_monitor_snapshot, get_cached_monitor_snapshot
|
||
from .extensions import socketio
|
||
from .state import (
|
||
MONITOR_FILE_TOOLS,
|
||
MONITOR_MEMORY_TOOLS,
|
||
MONITOR_SNAPSHOT_CHAR_LIMIT,
|
||
MONITOR_MEMORY_ENTRY_LIMIT,
|
||
RATE_LIMIT_BUCKETS,
|
||
FAILURE_TRACKERS,
|
||
pending_socket_tokens,
|
||
usage_trackers,
|
||
MONITOR_SNAPSHOT_CACHE,
|
||
MONITOR_SNAPSHOT_CACHE_LIMIT,
|
||
PROJECT_STORAGE_CACHE,
|
||
PROJECT_STORAGE_CACHE_TTL_SECONDS,
|
||
RECENT_UPLOAD_EVENT_LIMIT,
|
||
RECENT_UPLOAD_FEED_LIMIT,
|
||
THINKING_FAILURE_KEYWORDS,
|
||
TITLE_PROMPT_PATH,
|
||
get_last_active_ts,
|
||
user_manager,
|
||
container_manager,
|
||
custom_tool_registry,
|
||
user_terminals,
|
||
terminal_rooms,
|
||
connection_users,
|
||
stop_flags,
|
||
active_polling_tasks,
|
||
get_stop_flag,
|
||
set_stop_flag,
|
||
clear_stop_flag,
|
||
)
|
||
from .chat_flow_helpers import (
|
||
detect_malformed_tool_call as _detect_malformed_tool_call,
|
||
detect_tool_failure,
|
||
get_thinking_state,
|
||
mark_force_thinking as _mark_force_thinking,
|
||
mark_suppress_thinking,
|
||
apply_thinking_schedule as _apply_thinking_schedule,
|
||
update_thinking_after_call as _update_thinking_after_call,
|
||
maybe_mark_failure_from_message as _maybe_mark_failure_from_message,
|
||
generate_conversation_title_background as _generate_conversation_title_background,
|
||
)
|
||
|
||
|
||
from .chat_flow_runner_helpers import (
|
||
extract_intent_from_partial,
|
||
resolve_monitor_path,
|
||
resolve_monitor_memory,
|
||
capture_monitor_snapshot,
|
||
)
|
||
|
||
|
||
from .chat_flow_runtime import (
|
||
generate_conversation_title_background,
|
||
mark_force_thinking,
|
||
apply_thinking_schedule,
|
||
update_thinking_after_call,
|
||
maybe_mark_failure_from_message,
|
||
detect_malformed_tool_call,
|
||
)
|
||
|
||
from .chat_flow_task_support import (
|
||
process_sub_agent_updates,
|
||
process_background_command_updates,
|
||
process_multi_agent_master_messages,
|
||
inject_multi_agent_master_message,
|
||
_auto_message_type_for_multi_agent_subtype,
|
||
)
|
||
from .chat_flow_tool_loop import execute_tool_calls
|
||
from .chat_flow_stream_loop import run_streaming_attempts
|
||
from .deep_compression import run_deep_compression
|
||
from .goal_flow import (
|
||
maybe_start_goal,
|
||
inject_goal_prompt,
|
||
handle_goal_after_turn,
|
||
stop_goal_user_cancel,
|
||
goal_is_active,
|
||
emit_goal_progress,
|
||
)
|
||
|
||
_VALID_USER_MESSAGE_SOURCES = {
|
||
"user",
|
||
"guidance",
|
||
"notify",
|
||
"presend",
|
||
"sub_agent",
|
||
"background_command",
|
||
"goal",
|
||
"goal_prompt",
|
||
"goal_review",
|
||
"compression",
|
||
"compression_handoff",
|
||
"permission",
|
||
"sandbox",
|
||
"skill",
|
||
}
|
||
|
||
|
||
def _user_message_ui_defaults(source: str, *, auto_user_message_event: bool = False) -> Dict[str, Any]:
|
||
normalized = str(source or "").strip().lower()
|
||
if normalized in {"skill", "goal_prompt", "permission", "sandbox"}:
|
||
return {"visibility": "hidden", "starts_work": False}
|
||
if normalized in {"guidance", "notify"}:
|
||
return {"visibility": "compact", "starts_work": False}
|
||
# 后台完成通知 / 运行期压缩续接：本质属于当前这轮工作的延续，
|
||
# 不应暂停上一轮计时器、也不应开启新的工作头与计时器。
|
||
if normalized in {"sub_agent", "background_command", "compression", "compression_handoff"}:
|
||
return {"visibility": "compact", "starts_work": False}
|
||
# 目标审核续命：确实开启新一轮工作，保持 starts_work=True。
|
||
if normalized == "goal_review":
|
||
return {"visibility": "compact", "starts_work": True}
|
||
if normalized == "presend":
|
||
return {"visibility": "chat", "starts_work": True}
|
||
if normalized == "goal":
|
||
return {"visibility": "compact", "starts_work": True}
|
||
return {"visibility": "chat", "starts_work": normalized == "user" and not auto_user_message_event}
|
||
|
||
|
||
def _extract_media_path(item: Any) -> str:
|
||
if isinstance(item, dict):
|
||
return str(item.get("path") or "").strip()
|
||
return str(item or "").strip()
|
||
|
||
|
||
def _build_media_paths_message(images: Optional[List[Any]], videos: Optional[List[Any]]) -> Optional[str]:
|
||
"""为本次消息附带的图片/视频生成一条 hidden user 消息文本。"""
|
||
paths: List[str] = []
|
||
for item in images or []:
|
||
path = _extract_media_path(item)
|
||
if path and path not in paths:
|
||
paths.append(path)
|
||
for item in videos or []:
|
||
path = _extract_media_path(item)
|
||
if path and path not in paths:
|
||
paths.append(path)
|
||
if not paths:
|
||
return None
|
||
lines = ["[系统通知|media_paths]", "本次消息附带的媒体文件路径："]
|
||
for path in paths:
|
||
lines.append(f"- `{path}`")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def _should_skip_versioning_for_message(
|
||
*,
|
||
message: str,
|
||
auto_user_message_event: bool,
|
||
) -> bool:
|
||
if auto_user_message_event:
|
||
return True
|
||
text = (message or "").strip()
|
||
if not text:
|
||
return True
|
||
if text.startswith("这是一句系统自动发送的user消息"):
|
||
return True
|
||
return False
|
||
|
||
|
||
def _prepare_hidden_versioning_baseline_for_first_input(
|
||
*,
|
||
web_terminal,
|
||
workspace,
|
||
conversation_id: str,
|
||
message: str,
|
||
auto_user_message_event: bool,
|
||
) -> None:
|
||
"""Ensure first-run baseline is committed (hidden, no checkpoint row)."""
|
||
try:
|
||
if _should_skip_versioning_for_message(
|
||
message=message,
|
||
auto_user_message_event=auto_user_message_event,
|
||
):
|
||
return
|
||
cm = getattr(getattr(web_terminal, "context_manager", None), "conversation_manager", None)
|
||
if not cm:
|
||
return
|
||
conv_data = cm.load_conversation(conversation_id) or {}
|
||
versioning_meta = ((conv_data.get("metadata") or {}).get("versioning") or {})
|
||
if not bool(versioning_meta.get("enabled", False)):
|
||
return
|
||
backup_mode = str(versioning_meta.get("backup_mode") or "shallow").strip().lower()
|
||
if backup_mode == "full":
|
||
tracking_mode = ConversationVersioningManager.normalize_tracking_mode(
|
||
versioning_meta.get("tracking_mode")
|
||
)
|
||
if (
|
||
tracking_mode == ConversationVersioningManager.TRACKING_MODE_WORKSPACE_AND_CONVERSATION
|
||
and not bool(getattr(web_terminal, "username", None) == "host")
|
||
):
|
||
return
|
||
manager = ConversationVersioningManager(
|
||
project_path=workspace.project_path,
|
||
data_dir=workspace.data_dir,
|
||
conversation_id=conversation_id,
|
||
)
|
||
baseline = manager.ensure_baseline_for_first_input(tracking_mode=tracking_mode)
|
||
debug_log(
|
||
f"[Versioning][Baseline] conv={conversation_id} "
|
||
f"tracking_mode={tracking_mode} "
|
||
f"created={baseline.get('created')} skipped={baseline.get('skipped')} "
|
||
f"reason={baseline.get('reason')} head={baseline.get('head')}"
|
||
)
|
||
# shallow 模式下不需要完整 workspace baseline，由工具调用前的 track_edit 处理
|
||
except VersioningError as exc:
|
||
debug_log(f"[Versioning] 创建首轮基线失败: {exc}")
|
||
except Exception as exc:
|
||
debug_log(f"[Versioning] 创建首轮基线异常: {exc}")
|
||
|
||
|
||
def _record_hidden_versioning_checkpoint_after_run(
|
||
*,
|
||
web_terminal,
|
||
workspace,
|
||
conversation_id: str,
|
||
message: str,
|
||
message_index: int,
|
||
auto_user_message_event: bool,
|
||
run_status: str = "completed",
|
||
) -> None:
|
||
"""Hidden snapshot after current manual user input run finished."""
|
||
try:
|
||
if _should_skip_versioning_for_message(
|
||
message=message,
|
||
auto_user_message_event=auto_user_message_event,
|
||
):
|
||
return
|
||
cm = getattr(getattr(web_terminal, "context_manager", None), "conversation_manager", None)
|
||
if not cm:
|
||
return
|
||
conv_data = cm.load_conversation(conversation_id) or {}
|
||
versioning_meta = ((conv_data.get("metadata") or {}).get("versioning") or {})
|
||
if not bool(versioning_meta.get("enabled", False)):
|
||
return
|
||
tracking_mode = ConversationVersioningManager.normalize_tracking_mode(
|
||
versioning_meta.get("tracking_mode")
|
||
)
|
||
if (
|
||
tracking_mode == ConversationVersioningManager.TRACKING_MODE_WORKSPACE_AND_CONVERSATION
|
||
and not bool(getattr(web_terminal, "username", None) == "host")
|
||
):
|
||
return
|
||
snapshot_messages = conv_data.get("messages") or []
|
||
# 找到当前用户消息的消息ID（用于浅备份快照）
|
||
message_id = None
|
||
try:
|
||
target_msg = snapshot_messages[int(message_index)] if 0 <= int(message_index) < len(snapshot_messages) else None
|
||
if target_msg and target_msg.get("role") == "user":
|
||
message_id = target_msg.get("message_id") or target_msg.get("id")
|
||
except Exception:
|
||
message_id = None
|
||
|
||
backup_mode = str(versioning_meta.get("backup_mode") or "shallow").strip().lower()
|
||
if backup_mode == "full":
|
||
snapshot_payload = {
|
||
"conversation_id": conversation_id,
|
||
"title": conv_data.get("title"),
|
||
"metadata": conv_data.get("metadata") or {},
|
||
"messages": snapshot_messages,
|
||
"message_index": int(message_index),
|
||
"run_status": str(run_status or "completed"),
|
||
}
|
||
manager = ConversationVersioningManager(
|
||
project_path=workspace.project_path,
|
||
data_dir=workspace.data_dir,
|
||
conversation_id=conversation_id,
|
||
)
|
||
row = manager.create_checkpoint(
|
||
message=message,
|
||
message_index=message_index,
|
||
workspace_path=str(workspace.project_path),
|
||
conversation_snapshot=snapshot_payload,
|
||
run_status=str(run_status or "completed"),
|
||
tracking_mode=tracking_mode,
|
||
)
|
||
debug_log(
|
||
f"[Versioning][Checkpoint] conv={conversation_id} seq={row.get('seq')} "
|
||
f"tracking_mode={tracking_mode} "
|
||
f"msg_index={message_index} snapshot_messages={len(snapshot_messages)} "
|
||
f"tree_hash={row.get('tree_hash')} changed={row.get('changed')} status={row.get('run_status')}"
|
||
)
|
||
cm.update_conversation_metadata(
|
||
conversation_id,
|
||
{
|
||
"versioning": {
|
||
"enabled": True,
|
||
"mode": "overwrite",
|
||
"tracking_mode": tracking_mode,
|
||
"backup_mode": "full",
|
||
"last_commit": row.get("tree_hash"),
|
||
"last_input_seq": int(row.get("seq") or 0),
|
||
"updated_at": datetime.now().isoformat(),
|
||
}
|
||
},
|
||
)
|
||
else:
|
||
# 浅备份：每个用户消息结束时对 tracked files 做快照
|
||
shallow_files: List[Dict[str, Any]] = []
|
||
shallow_insertions = 0
|
||
shallow_deletions = 0
|
||
if message_id:
|
||
shallow_manager = ShallowVersioningManager(
|
||
project_path=workspace.project_path,
|
||
data_dir=workspace.data_dir,
|
||
conversation_id=conversation_id,
|
||
)
|
||
snapshot = shallow_manager.make_snapshot(str(message_id))
|
||
shallow_files = shallow_manager.get_file_diff_stats(str(message_id))
|
||
stats = shallow_manager.get_diff_stats(str(message_id)) or {}
|
||
shallow_insertions = int(stats.get("insertions") or 0)
|
||
shallow_deletions = int(stats.get("deletions") or 0)
|
||
debug_log(
|
||
f"[Versioning][ShallowSnapshot] conv={conversation_id} "
|
||
f"msg_id={message_id} tracked_files={snapshot.get('tracked_files_count')}"
|
||
)
|
||
|
||
# 同时写入一个轻量 ConversationVersioningManager 检查点行，
|
||
# 让前端版本管理弹窗能统一列出回溯点。
|
||
manager = ConversationVersioningManager(
|
||
project_path=workspace.project_path,
|
||
data_dir=workspace.data_dir,
|
||
conversation_id=conversation_id,
|
||
)
|
||
snapshot_payload = {
|
||
"conversation_id": conversation_id,
|
||
"title": conv_data.get("title"),
|
||
"metadata": conv_data.get("metadata") or {},
|
||
"messages": snapshot_messages,
|
||
"message_index": int(message_index),
|
||
"run_status": str(run_status or "completed"),
|
||
"shallow_message_id": str(message_id) if message_id else None,
|
||
}
|
||
row = manager.create_checkpoint(
|
||
message=message,
|
||
message_index=int(message_index),
|
||
workspace_path=str(workspace.project_path),
|
||
conversation_snapshot=snapshot_payload,
|
||
run_status=str(run_status or "completed"),
|
||
tracking_mode=tracking_mode,
|
||
diff_summary={
|
||
"insertions": shallow_insertions,
|
||
"deletions": shallow_deletions,
|
||
"files_changed": len(shallow_files),
|
||
"files": shallow_files,
|
||
},
|
||
)
|
||
debug_log(
|
||
f"[Versioning][ShallowCheckpoint] conv={conversation_id} seq={row.get('seq')} "
|
||
f"msg_index={message_index} files={len(shallow_files)} "
|
||
f"insertions={shallow_insertions} deletions={shallow_deletions}"
|
||
)
|
||
cm.update_conversation_metadata(
|
||
conversation_id,
|
||
{
|
||
"versioning": {
|
||
"enabled": True,
|
||
"mode": "overwrite",
|
||
"tracking_mode": tracking_mode,
|
||
"backup_mode": "shallow",
|
||
"last_commit": row.get("tree_hash"),
|
||
"last_input_seq": int(row.get("seq") or 0),
|
||
"updated_at": datetime.now().isoformat(),
|
||
}
|
||
},
|
||
)
|
||
except VersioningError as exc:
|
||
debug_log(f"[Versioning] 记录快照失败: {exc}")
|
||
except Exception as exc:
|
||
debug_log(f"[Versioning] 记录快照异常: {exc}")
|
||
finally:
|
||
# 当前输入处理结束，清除浅备份消息 ID，避免后续无消息上下文时的错误归属。
|
||
try:
|
||
ctx = getattr(web_terminal, "context_manager", None)
|
||
if ctx is not None and hasattr(ctx, "current_shallow_message_id"):
|
||
ctx.current_shallow_message_id = None
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _persist_and_echo_preceding_notice(
|
||
*,
|
||
web_terminal,
|
||
sender,
|
||
conversation_id: str,
|
||
message: str,
|
||
payload: Dict[str, Any],
|
||
) -> None:
|
||
"""预写一条「前置完成通知」：写入对话历史 + socketio 回显（不触发新一轮工作）。
|
||
|
||
用于「通知池」一次取出多条时，前 N-1 条随同一后续任务一起呈现。
|
||
这些通知必须落历史，否则模型看不到、刷新后也会丢失。
|
||
"""
|
||
message = (message or "").strip()
|
||
if not message:
|
||
return
|
||
src = str(payload.get("message_source") or "sub_agent").strip().lower()
|
||
if src not in _VALID_USER_MESSAGE_SOURCES:
|
||
src = "sub_agent"
|
||
ui_defaults = _user_message_ui_defaults(src, auto_user_message_event=True)
|
||
# 前置通知渲染为完成态气泡：不开启独立工作计时器
|
||
ui_defaults = {**ui_defaults, "starts_work": False}
|
||
metadata = {
|
||
"message_source": src,
|
||
"is_auto_generated": True,
|
||
"auto_message_type": "completion_notice",
|
||
**ui_defaults,
|
||
}
|
||
try:
|
||
cm = getattr(web_terminal, "context_manager", None)
|
||
if cm is not None:
|
||
cm.add_conversation("user", message, metadata=metadata)
|
||
except Exception as exc:
|
||
echo_payload = {
|
||
"message": message,
|
||
"conversation_id": conversation_id,
|
||
"message_source": src,
|
||
**ui_defaults,
|
||
**{k: v for k, v in (payload or {}).items() if k not in {"message", "conversation_id"}},
|
||
}
|
||
echo_payload["starts_work"] = False
|
||
echo_payload["metadata"] = {**metadata}
|
||
sender("user_message", echo_payload)
|
||
except Exception as exc:
|
||
pass
|
||
|
||
async def _dispatch_completion_user_notice(
|
||
*,
|
||
web_terminal,
|
||
workspace,
|
||
sender,
|
||
client_sid: str,
|
||
username: str,
|
||
conversation_id: str,
|
||
user_message: str,
|
||
extra_payload: Optional[Dict[str, Any]] = None,
|
||
preceding_notices: Optional[List[Dict[str, Any]]] = None,
|
||
):
|
||
"""复用子智能体完成后的 user 代发机制。
|
||
|
||
preceding_notices：本批「通知池」里除最后一条以外的前置通知列表，
|
||
每项为 {"message": str, "payload": dict}。这些会先于触发消息写入历史/事件流，
|
||
但不各自触发新一轮工作；仅最后一条 user_message 作为后续任务的触发消息。
|
||
"""
|
||
extra_payload = extra_payload or {}
|
||
preceding_notices = preceding_notices or []
|
||
message_source = str(extra_payload.get("message_source") or "").strip().lower()
|
||
if not message_source:
|
||
if extra_payload.get("runtime_mode_notice"):
|
||
message_source = "notify"
|
||
elif extra_payload.get("runtime_guidance"):
|
||
message_source = "guidance"
|
||
elif extra_payload.get("background_command_notice"):
|
||
message_source = "background_command"
|
||
elif extra_payload.get("sub_agent_notice"):
|
||
message_source = "sub_agent"
|
||
if message_source not in _VALID_USER_MESSAGE_SOURCES:
|
||
message_source = "user"
|
||
|
||
# 先把前置通知写入历史并 socketio 回显（在线客户端即时可见）。
|
||
# 轮询客户端则通过后续任务事件流回放（见 _run_chat_task 的 preceding_user_notices 注入）。
|
||
for item in preceding_notices:
|
||
_persist_and_echo_preceding_notice(
|
||
web_terminal=web_terminal,
|
||
sender=sender,
|
||
conversation_id=conversation_id,
|
||
message=str(item.get("message") or ""),
|
||
payload=dict(item.get("payload") or {}),
|
||
)
|
||
|
||
try:
|
||
from .tasks import task_manager
|
||
workspace_id = getattr(workspace, "workspace_id", None) or "default"
|
||
host_mode = bool(getattr(workspace, "username", None) == "host")
|
||
session_data = {
|
||
"username": username,
|
||
"role": getattr(web_terminal, "user_role", "user"),
|
||
"is_api_user": getattr(web_terminal, "user_role", "") == "api",
|
||
"host_mode": host_mode,
|
||
"host_workspace_id": workspace_id if host_mode else None,
|
||
"workspace_id": workspace_id,
|
||
"run_mode": getattr(web_terminal, "run_mode", None),
|
||
"thinking_mode": getattr(web_terminal, "thinking_mode", None),
|
||
"model_key": getattr(web_terminal, "model_key", None),
|
||
"message_source": message_source,
|
||
}
|
||
ui_defaults = _user_message_ui_defaults(
|
||
message_source,
|
||
auto_user_message_event=True,
|
||
)
|
||
# 关键：通知类后台任务需要把 user_message 写入任务事件流，
|
||
# 否则前端轮询只会看到 AI/tool 事件，看不到 user_message。
|
||
session_data["auto_user_message_event"] = True
|
||
session_data["auto_user_message_payload"] = {
|
||
**dict(extra_payload or {}),
|
||
**ui_defaults,
|
||
"timestamp": datetime.now().isoformat(),
|
||
}
|
||
# 轮询客户端通过后续任务事件流回放前置通知（在线客户端已由上面的 socketio 回显覆盖）。
|
||
if preceding_notices:
|
||
session_data["preceding_user_notices"] = [
|
||
{
|
||
"message": str(item.get("message") or ""),
|
||
"payload": dict(item.get("payload") or {}),
|
||
}
|
||
for item in preceding_notices
|
||
if str(item.get("message") or "").strip()
|
||
]
|
||
ma_debug(
|
||
"dispatch_completion_create_chat_task",
|
||
conversation_id=conversation_id,
|
||
user_message_preview=user_message[:300],
|
||
preceding_count=len(preceding_notices),
|
||
)
|
||
rec = task_manager.create_chat_task(
|
||
username,
|
||
workspace_id,
|
||
user_message,
|
||
[],
|
||
conversation_id,
|
||
model_key=session_data.get("model_key"),
|
||
thinking_mode=session_data.get("thinking_mode"),
|
||
run_mode=session_data.get("run_mode"),
|
||
session_data=session_data,
|
||
)
|
||
ma_debug(
|
||
"dispatch_completion_task_created",
|
||
conversation_id=conversation_id,
|
||
task_id=getattr(rec, "task_id", None),
|
||
)
|
||
payload = {
|
||
'message': user_message,
|
||
'conversation_id': conversation_id,
|
||
'task_id': rec.task_id,
|
||
'message_source': message_source,
|
||
'timestamp': datetime.now().isoformat(),
|
||
**ui_defaults,
|
||
}
|
||
payload.update(extra_payload)
|
||
payload["metadata"] = {
|
||
"message_source": message_source,
|
||
**ui_defaults,
|
||
**(payload.get("metadata") or {}),
|
||
}
|
||
sender('user_message', payload)
|
||
return
|
||
except Exception as e:
|
||
debug_log(f"[CompletionNotice] 创建后台消息任务失败，回退直接执行: {e}")
|
||
|
||
payload = {
|
||
'message': user_message,
|
||
'conversation_id': conversation_id,
|
||
'message_source': message_source,
|
||
'timestamp': datetime.now().isoformat(),
|
||
}
|
||
payload.update(_user_message_ui_defaults(message_source, auto_user_message_event=True))
|
||
payload.update(extra_payload)
|
||
payload["metadata"] = {
|
||
"message_source": message_source,
|
||
"visibility": payload.get("visibility"),
|
||
"starts_work": payload.get("starts_work"),
|
||
**(payload.get("metadata") or {}),
|
||
}
|
||
sender('user_message', payload)
|
||
try:
|
||
task_handle = asyncio.create_task(handle_task_with_sender(
|
||
terminal=web_terminal,
|
||
workspace=workspace,
|
||
message=user_message,
|
||
images=[],
|
||
sender=sender,
|
||
client_sid=client_sid,
|
||
username=username,
|
||
videos=[],
|
||
auto_user_message_event=True,
|
||
))
|
||
await task_handle
|
||
except Exception as inner_exc:
|
||
debug_log(f"[CompletionNotice] 回退处理 user_message 失败: {inner_exc}")
|
||
|
||
|
||
def _build_shared_waiting_payload(items: List[Dict[str, Any]]) -> Dict[str, Any]:
|
||
"""构建与子智能体一致的 waiting 事件载荷结构。"""
|
||
normalized = []
|
||
for item in items or []:
|
||
summary = item.get('summary') or item.get('command') or '后台任务'
|
||
normalized.append({
|
||
'agent_id': item.get('agent_id') or item.get('command_id') or item.get('task_id'),
|
||
'summary': summary,
|
||
})
|
||
return {
|
||
'count': len(normalized),
|
||
'tasks': normalized,
|
||
}
|
||
|
||
def _build_sub_agent_notice_text(update: Dict[str, Any], *, summary_override: Optional[str] = None) -> str:
|
||
"""根据子智能体完成结果构建一条 user 通知文本（统一 [系统通知|sub_agent] 前缀）。
|
||
|
||
update 通常是 task 的 final_result，其中 system_message 已经是一条完整的格式化通知
|
||
（含 ✅/❌ 前缀、统计、运行秒数、summary、交付目录）。这里直接复用它，只补统一前缀，
|
||
不再在外层重复拼 agent/运行秒数/交付目录，避免内容出现两遍。
|
||
"""
|
||
system_message = (update.get("system_message") or "").strip()
|
||
if not system_message:
|
||
# 兜底：极少数情况下 final_result 没有 system_message，用最小信息拼一条
|
||
agent_id = update.get("agent_id")
|
||
summary = summary_override or update.get("summary") or update.get("message") or ""
|
||
deliverables_dir = update.get("deliverables_dir", "") or ""
|
||
lines = [f"子智能体{agent_id} ({summary}) 已完成任务。"]
|
||
msg = update.get("message")
|
||
if msg:
|
||
lines.append(str(msg))
|
||
if deliverables_dir:
|
||
lines.append(f"交付目录：{deliverables_dir}")
|
||
system_message = "\n\n".join(lines)
|
||
return f"[系统通知|sub_agent]\n{system_message}"
|
||
|
||
|
||
def _collect_pending_completion_notices(*, web_terminal, conversation_id: str) -> List[Dict[str, Any]]:
|
||
"""从子智能体 + 后台 run_command 两路统一取出所有待通知项（已按时间排序）。
|
||
|
||
返回的每一项形如：
|
||
{
|
||
"kind": "sub_agent" | "background_command",
|
||
"message": <user 通知文本>,
|
||
"payload": <前端事件 extra payload>,
|
||
"sort_key": <用于跨两路排序的时间戳>,
|
||
}
|
||
取出时即就地标记 notified，确保不会被下一轮重复消费。
|
||
"""
|
||
notices: List[Dict[str, Any]] = []
|
||
|
||
# 1) 子智能体
|
||
sub_manager = getattr(web_terminal, "sub_agent_manager", None)
|
||
if sub_manager:
|
||
if not hasattr(web_terminal, "_announced_sub_agent_tasks"):
|
||
web_terminal._announced_sub_agent_tasks = set()
|
||
# 关键：先 reconcile 刷新真实状态，然后**直接遍历 tasks** 找终态+未通知的任务。
|
||
# 不依赖 poll_updates() —— 它只返回「本次调用里恰好 running→terminal」的跃迁项，
|
||
# 任务一旦被提前 reconcile 成 completed，跃迁就丢了，poll_updates 永远返回空。
|
||
try:
|
||
sub_manager.reconcile_task_states(conversation_id=conversation_id)
|
||
except Exception:
|
||
pass
|
||
candidates = [
|
||
t for t in sub_manager.tasks.values()
|
||
if isinstance(t, dict)
|
||
and t.get("conversation_id") == conversation_id
|
||
and t.get("status") in TERMINAL_STATUSES # completed/failed/timeout（不含 terminated）
|
||
and not t.get("notified")
|
||
and t.get("task_id") not in web_terminal._announced_sub_agent_tasks
|
||
and not t.get("multi_agent_mode") # 多智能体任务走独立注入路径，不在这里发传统通知
|
||
]
|
||
candidates.sort(key=lambda t: t.get("updated_at") or t.get("created_at") or 0)
|
||
for task_info in candidates:
|
||
task_id = task_info.get("task_id")
|
||
final_result = task_info.get("final_result")
|
||
if not isinstance(final_result, dict):
|
||
# 兜底：状态已是终态但 final_result 缺失，主动取一次
|
||
try:
|
||
final_result = sub_manager._check_task_status(task_info)
|
||
except Exception:
|
||
pass
|
||
if not isinstance(final_result, dict):
|
||
final_result = {}
|
||
# 就地标记已通知
|
||
if task_id:
|
||
web_terminal._announced_sub_agent_tasks.add(task_id)
|
||
task_info["notified"] = True
|
||
task_info["updated_at"] = time.time()
|
||
|
||
agent_id = task_info.get("agent_id")
|
||
summary = task_info.get("summary") or ""
|
||
notice_text = _build_sub_agent_notice_text(
|
||
final_result,
|
||
summary_override=summary,
|
||
)
|
||
notices.append({
|
||
"kind": "sub_agent",
|
||
"message": notice_text,
|
||
"payload": {
|
||
"sub_agent_notice": True,
|
||
"message_source": "sub_agent",
|
||
"agent_id": agent_id,
|
||
"task_id": task_id,
|
||
},
|
||
"sort_key": task_info.get("updated_at") or task_info.get("created_at") or time.time(),
|
||
})
|
||
try:
|
||
sub_manager._save_state()
|
||
except Exception:
|
||
pass
|
||
|
||
# 2) 后台 run_command
|
||
bg_manager = getattr(web_terminal, "background_command_manager", None)
|
||
if bg_manager:
|
||
try:
|
||
bg_updates = bg_manager.poll_updates(conversation_id=conversation_id)
|
||
except Exception:
|
||
bg_updates = []
|
||
for update in bg_updates or []:
|
||
command_id = update.get("command_id")
|
||
command = update.get("command") or ""
|
||
output = update.get("output") or ""
|
||
return_code = update.get("return_code")
|
||
try:
|
||
bg_manager.mark_notified(str(command_id))
|
||
except Exception:
|
||
pass
|
||
# 统一 [系统通知|background_command] 前缀；正文给出命令、退出码、输出
|
||
header = "后台指令已完成。"
|
||
if command:
|
||
header += f"\n\n命令：{command}"
|
||
if return_code is not None:
|
||
header += f"\n退出码：{return_code}"
|
||
body = output if output else "[no_output]"
|
||
message_text = f"[系统通知|background_command]\n{header}\n\n输出：\n{body}"
|
||
notices.append({
|
||
"kind": "background_command",
|
||
"message": message_text,
|
||
"payload": {
|
||
# 与子智能体完成通知完全复用同一前端通道/处理逻辑
|
||
"sub_agent_notice": True,
|
||
"message_source": "background_command",
|
||
"background_command_notice": True,
|
||
},
|
||
"sort_key": update.get("updated_at") or time.time(),
|
||
})
|
||
|
||
notices.sort(key=lambda item: item.get("sort_key") or 0)
|
||
return notices
|
||
|
||
|
||
def _has_pending_completion_work(*, web_terminal, conversation_id: str) -> bool:
|
||
"""是否还有运行中或待通知的传统后台任务（子智能体/后台 run_command）。
|
||
|
||
多智能体模式下的 pending 消息由独立的 poll_multi_agent_notifications 处理，
|
||
不再混入本函数。
|
||
"""
|
||
sub_manager = getattr(web_terminal, "sub_agent_manager", None)
|
||
if sub_manager:
|
||
announced = getattr(web_terminal, "_announced_sub_agent_tasks", set())
|
||
has_running_non_ma = False
|
||
has_unnotified_non_ma = False
|
||
for task in sub_manager.tasks.values():
|
||
if not isinstance(task, dict):
|
||
continue
|
||
if task.get("conversation_id") != conversation_id:
|
||
continue
|
||
status = task.get("status")
|
||
multi_agent_flag = task.get("multi_agent_mode") or False
|
||
if status not in TERMINAL_STATUSES.union({"terminated"}):
|
||
if not multi_agent_flag and task.get("run_in_background"):
|
||
has_running_non_ma = True
|
||
continue
|
||
if not multi_agent_flag and task.get("run_in_background") and (task.get("task_id") not in announced) and not task.get("notified"):
|
||
has_unnotified_non_ma = True
|
||
if has_running_non_ma or has_unnotified_non_ma:
|
||
return True
|
||
bg_manager = getattr(web_terminal, "background_command_manager", None)
|
||
if bg_manager:
|
||
try:
|
||
if bg_manager.has_pending_for_conversation(conversation_id):
|
||
return True
|
||
except Exception:
|
||
pass
|
||
return False
|
||
|
||
|
||
async def poll_completion_notifications(*, web_terminal, workspace, conversation_id, client_sid, username):
|
||
"""统一的“通知池”轮询器。
|
||
|
||
把子智能体完成通知与后台 run_command 完成通知合并到同一条链路：
|
||
- 每轮把两路所有待通知项一次性取出（池化），按时间排序；
|
||
- 这一批里前 N-1 条作为「预写通知」随同一个后续任务一起注入（写历史 + 事件流 + socketio），
|
||
不各自触发新一轮工作；
|
||
- 仅最后 1 条作为后续 chat task 的触发消息，启动一次「工作 → 停止」循环；
|
||
- 该后续任务结束后回到 handle_task_with_sender 结尾重新 spawn 本轮询器，继续消费剩余通知。
|
||
|
||
这样多个后台任务同时完成时，会被合并成一次（而非每条都触发一轮停止-再工作）。
|
||
"""
|
||
from .extensions import socketio
|
||
|
||
sub_manager = getattr(web_terminal, "sub_agent_manager", None)
|
||
bg_manager = getattr(web_terminal, "background_command_manager", None)
|
||
if not sub_manager and not bg_manager:
|
||
return
|
||
if not hasattr(web_terminal, "_announced_sub_agent_tasks"):
|
||
web_terminal._announced_sub_agent_tasks = set()
|
||
|
||
max_wait_time = 3600 # 最多等待 1 小时
|
||
start_wait = time.time()
|
||
|
||
def sender(event_type, data):
|
||
try:
|
||
socketio.emit(event_type, data, room=f"user_{username}")
|
||
except Exception:
|
||
pass
|
||
|
||
loop_count = 0
|
||
ma_debug("poll_completion_notifications_start", conversation_id=conversation_id)
|
||
while (time.time() - start_wait) < max_wait_time:
|
||
loop_count += 1
|
||
# 检查停止标志
|
||
client_stop_info = get_stop_flag(client_sid, username)
|
||
if client_stop_info:
|
||
stop_requested = client_stop_info.get('stop', False) if isinstance(client_stop_info, dict) else client_stop_info
|
||
if stop_requested:
|
||
ma_debug("poll_completion_notifications_stop_requested", conversation_id=conversation_id)
|
||
break
|
||
|
||
# 若主对话仍在工具循环中，暂不消费完成事件，避免抢占 system 消息插入
|
||
if getattr(web_terminal, "_tool_loop_active", False):
|
||
ma_debug("poll_completion_notifications_wait_tool_loop", conversation_id=conversation_id)
|
||
await asyncio.sleep(1)
|
||
continue
|
||
|
||
# 多智能体模式：主对话任务仍在运行时，由主循环自己消费 pending 消息
|
||
if getattr(web_terminal, "_multi_agent_main_task_active", False):
|
||
ma_debug("poll_completion_notifications_wait_main_task", conversation_id=conversation_id)
|
||
await asyncio.sleep(1)
|
||
continue
|
||
|
||
ma_debug("poll_completion_notifications_collect", conversation_id=conversation_id, loop_count=loop_count)
|
||
notices = _collect_pending_completion_notices(
|
||
web_terminal=web_terminal,
|
||
conversation_id=conversation_id,
|
||
)
|
||
ma_debug("poll_completion_notifications_collected", conversation_id=conversation_id, notice_count=len(notices))
|
||
|
||
if notices:
|
||
# 取出后剩余是否还有未完成/未通知的后台工作（决定前端是否保持等待态）
|
||
has_remaining = _has_pending_completion_work(
|
||
web_terminal=web_terminal,
|
||
conversation_id=conversation_id,
|
||
)
|
||
|
||
preceding = notices[:-1]
|
||
last_notice = notices[-1]
|
||
|
||
# 前 N-1 条预写通知：随后续任务注入历史/事件流，不单独触发工作
|
||
preceding_payload: List[Dict[str, Any]] = []
|
||
for item in preceding:
|
||
payload = dict(item.get("payload") or {})
|
||
payload["starts_work"] = False # 预写通知渲染为完成态气泡，不起独立计时器
|
||
preceding_payload.append({
|
||
"message": item.get("message") or "",
|
||
"payload": payload,
|
||
})
|
||
|
||
last_payload = dict(last_notice.get("payload") or {})
|
||
last_payload.update({
|
||
"has_running_sub_agents": has_remaining,
|
||
"has_running_background_commands": has_remaining,
|
||
"remaining_count": 1 if has_remaining else 0,
|
||
})
|
||
|
||
try:
|
||
await _dispatch_completion_user_notice(
|
||
web_terminal=web_terminal,
|
||
workspace=workspace,
|
||
sender=sender,
|
||
client_sid=client_sid,
|
||
username=username,
|
||
conversation_id=conversation_id,
|
||
user_message=last_notice.get("message") or "",
|
||
extra_payload=last_payload,
|
||
preceding_notices=preceding_payload,
|
||
)
|
||
except Exception:
|
||
pass
|
||
|
||
# 本轮已派发一个后续任务，由该任务结束后重新 spawn 本轮询器继续消费剩余通知
|
||
return
|
||
|
||
# 没有待通知项：若也没有运行中的后台工作，结束轮询
|
||
pending = _has_pending_completion_work(web_terminal=web_terminal, conversation_id=conversation_id)
|
||
if not pending:
|
||
break
|
||
|
||
await asyncio.sleep(5)
|
||
|
||
|
||
async def poll_multi_agent_notifications(*, web_terminal, workspace, conversation_id, client_sid, username):
|
||
"""多智能体模式专用通知池。
|
||
|
||
只在主任务结束后启动，负责两件事：
|
||
1. 若主任务已空闲且 MultiAgentState 中有 pending_master_messages，
|
||
一次性取出并触发新一轮主任务工作——即使仍有 running 子智能体也立即派发
|
||
（主对话空闲时消息池有消息就该全部插入，设计中属情况2）。
|
||
2. 若 pending 为空且仍有 running 实例，循环等待下一次 push 新输出。
|
||
|
||
重要：旧实现要求所有 running 实例退出再 drain，导致 ask_master 阻塞中的
|
||
子智能体永远等不到主对话回答而超时——现已修复为池优先。
|
||
|
||
与 poll_completion_notifications 完全分离，避免多智能体消息和传统后台
|
||
通知竞争 task_manager 的单工作区互斥。
|
||
"""
|
||
from .extensions import socketio
|
||
|
||
max_wait_time = 3600
|
||
start_wait = time.time()
|
||
|
||
def sender(event_type, data):
|
||
try:
|
||
socketio.emit(event_type, data, room=f"user_{username}")
|
||
except Exception:
|
||
pass
|
||
|
||
import threading as _threading
|
||
ma_debug(
|
||
"poll_ma_notifications_start",
|
||
conversation_id=conversation_id,
|
||
thread_id=_threading.get_ident(),
|
||
multi_agent_mode=getattr(web_terminal, "multi_agent_mode", False),
|
||
main_task_active_at_start=getattr(web_terminal, "_multi_agent_main_task_active", False),
|
||
)
|
||
|
||
tick_count = 0
|
||
while (time.time() - start_wait) < max_wait_time:
|
||
tick_count += 1
|
||
# 检查停止标志
|
||
client_stop_info = get_stop_flag(client_sid, username)
|
||
if client_stop_info:
|
||
stop_requested = client_stop_info.get('stop', False) if isinstance(client_stop_info, dict) else client_stop_info
|
||
if stop_requested:
|
||
ma_debug("poll_ma_notifications_stop_requested", conversation_id=conversation_id, tick=tick_count)
|
||
break
|
||
|
||
manager = getattr(web_terminal, "sub_agent_manager", None)
|
||
if not manager:
|
||
ma_debug("poll_ma_no_manager", conversation_id=conversation_id, tick=tick_count)
|
||
break
|
||
|
||
state = manager.get_multi_agent_state(conversation_id)
|
||
if not state:
|
||
ma_debug("poll_ma_no_state", conversation_id=conversation_id, tick=tick_count)
|
||
break
|
||
|
||
# 主任务又被唤醒了（用户发送新消息等），让主循环自己处理，本通知池退出
|
||
if getattr(web_terminal, "_multi_agent_main_task_active", False):
|
||
ma_debug("poll_ma_notifications_main_task_active", conversation_id=conversation_id, tick=tick_count)
|
||
break
|
||
|
||
# 状态快照
|
||
all_insts = state.list_all()
|
||
statuses = [a.status for a in all_insts]
|
||
pending_count_now = len(state.pending_master_messages)
|
||
main_active = getattr(web_terminal, "_multi_agent_main_task_active", False)
|
||
ma_debug(
|
||
"poll_ma_tick",
|
||
conversation_id=conversation_id,
|
||
tick=tick_count,
|
||
instance_count=len(all_insts),
|
||
statuses=statuses,
|
||
pending_count=pending_count_now,
|
||
main_active=main_active,
|
||
)
|
||
|
||
# 1) 先检查 pending 池——主对话空闲时只要池里有消息就立即派发（情况2）。
|
||
if state.has_pending_master_messages():
|
||
pending = state.drain_master_messages()
|
||
ma_debug(
|
||
"poll_ma_notifications_dispatch",
|
||
conversation_id=conversation_id,
|
||
count=len(pending),
|
||
running_count=len([a for a in all_insts if a.status == "running"]),
|
||
previews=[str(m)[:200] for m in pending],
|
||
tick=tick_count,
|
||
)
|
||
try:
|
||
await _dispatch_multi_agent_idle_messages(
|
||
web_terminal=web_terminal,
|
||
workspace=workspace,
|
||
sender=sender,
|
||
client_sid=client_sid,
|
||
username=username,
|
||
conversation_id=conversation_id,
|
||
messages=pending,
|
||
)
|
||
ma_debug("poll_ma_dispatch_success", conversation_id=conversation_id, count=len(pending), tick=tick_count)
|
||
except Exception as exc:
|
||
debug_log(f"[MultiAgent] 分发 idle 消息失败: {exc}")
|
||
ma_debug(
|
||
"poll_ma_notifications_dispatch_error",
|
||
conversation_id=conversation_id,
|
||
error=str(exc),
|
||
error_type=type(exc).__name__,
|
||
tick=tick_count,
|
||
)
|
||
# 分发后退出，由新任务结束后再决定是否继续轮询
|
||
return
|
||
|
||
# 2) 没有 pending 消息：检查还有哪些子智能体在运行，决定退出还是继续等
|
||
running_agents = [a for a in all_insts if a.status == "running"]
|
||
if not running_agents:
|
||
ma_debug("poll_ma_notifications_no_work_end", conversation_id=conversation_id, tick=tick_count)
|
||
break
|
||
|
||
ma_debug(
|
||
"poll_ma_notifications_wait_running",
|
||
conversation_id=conversation_id,
|
||
running_count=len(running_agents),
|
||
tick=tick_count,
|
||
)
|
||
await asyncio.sleep(0.5)
|
||
|
||
ma_debug("poll_ma_notifications_end", conversation_id=conversation_id, tick=tick_count)
|
||
|
||
# poll 退出时，如果主智能体也空闲且没有 pending 消息，发送 task_complete 通知前端对话变空闲。
|
||
# 否则前端会一直保持 taskInProgress=true，MultiAgentTaskProbe 会持续轮询但找不到任务。
|
||
if not getattr(web_terminal, "_multi_agent_main_task_active", False):
|
||
state = manager.get_multi_agent_state(conversation_id) if manager else None
|
||
still_running = state and any(a.status == "running" for a in state.list_all())
|
||
still_pending = state and state.has_pending_master_messages()
|
||
if not still_running and not still_pending:
|
||
sender('task_complete', {
|
||
'total_iterations': 0,
|
||
'total_tool_calls': 0,
|
||
'auto_fix_attempts': 0,
|
||
'has_running_sub_agents': False,
|
||
'has_running_background_commands': False,
|
||
'has_running_multi_agent': False,
|
||
'pending_runtime_guidance_messages': [],
|
||
'multi_agent_idle_signal': True,
|
||
})
|
||
ma_debug("poll_ma_idle_signal_sent", conversation_id=conversation_id, tick=tick_count)
|
||
|
||
|
||
async def _dispatch_multi_agent_idle_messages(
|
||
*,
|
||
web_terminal,
|
||
workspace,
|
||
sender,
|
||
client_sid,
|
||
username,
|
||
conversation_id,
|
||
messages: List[str],
|
||
):
|
||
"""主智能体空闲时，把多智能体 pending 消息作为新一轮用户消息分发。
|
||
|
||
所有消息先持久化到对话历史；前 N-1 条作为前置通知通过 socketio 推送；
|
||
最后一条创建 task_type="notice" 任务触发 Team Leader 新一轮工作。
|
||
"""
|
||
ma_debug(
|
||
"dispatch_ma_idle_enter",
|
||
conversation_id=conversation_id,
|
||
count=len(messages) if messages else 0,
|
||
previews=[str(m)[:150] for m in (messages or [])],
|
||
)
|
||
if not messages:
|
||
ma_debug("dispatch_ma_idle_no_messages", conversation_id=conversation_id)
|
||
return
|
||
|
||
from modules.multi_agent.state import parse_multi_agent_message
|
||
from .tasks import task_manager
|
||
|
||
parsed_messages = []
|
||
for msg_text in messages:
|
||
parsed = parse_multi_agent_message(msg_text) or {}
|
||
parsed_messages.append({
|
||
"text": msg_text,
|
||
"display_name": parsed.get("display_name"),
|
||
"subtype": parsed.get("subtype"),
|
||
})
|
||
|
||
# 1) 前置通知全部写入历史 + emit给在线客户端（未后一条会由后续任务自己持久化，
|
||
# 避免重复产生两条相同 user 消息导致刷新后被渲染两遍）。
|
||
ma_debug(
|
||
"dispatch_ma_idle_preceding",
|
||
conversation_id=conversation_id,
|
||
preceding_count=max(0, len(parsed_messages) - 1),
|
||
)
|
||
for item in parsed_messages[:-1]:
|
||
inject_multi_agent_master_message(
|
||
web_terminal=web_terminal,
|
||
messages=None,
|
||
text=item["text"],
|
||
sender=sender,
|
||
conversation_id=conversation_id,
|
||
inline=False,
|
||
)
|
||
|
||
# 1.5) 最后一条只 emit 给在线客户端，不在这里持久化（后续 task handle_task_with_sender
|
||
# 会以这条文本作为 message 创建任务，并写入历史，从而避免重复写入。）
|
||
last = parsed_messages[-1]
|
||
message_source = "sub_agent"
|
||
last_emit_payload = {
|
||
"message": last["text"],
|
||
"content": last["text"],
|
||
"conversation_id": conversation_id,
|
||
"message_source": message_source,
|
||
"sub_agent_notice": True,
|
||
"multi_agent_message": True,
|
||
"multi_agent_display_name": last["display_name"],
|
||
"multi_agent_subtype": last["subtype"],
|
||
"auto_message_type": _auto_message_type_for_multi_agent_subtype(last["subtype"]),
|
||
"visibility": "chat",
|
||
"starts_work": False,
|
||
"metadata": {
|
||
"message_source": message_source,
|
||
"auto_message_type": _auto_message_type_for_multi_agent_subtype(last["subtype"]),
|
||
"multi_agent_message": True,
|
||
"multi_agent_display_name": last["display_name"],
|
||
"multi_agent_subtype": last["subtype"],
|
||
"visibility": "chat",
|
||
"starts_work": False,
|
||
},
|
||
"timestamp": datetime.now().isoformat(),
|
||
}
|
||
try:
|
||
sender("user_message", last_emit_payload)
|
||
ma_debug(
|
||
"dispatch_ma_idle_last_emit",
|
||
conversation_id=conversation_id,
|
||
message_preview=last["text"][:200],
|
||
)
|
||
except Exception as exc:
|
||
ma_debug(
|
||
"dispatch_ma_idle_last_emit_error",
|
||
conversation_id=conversation_id,
|
||
error=str(exc),
|
||
)
|
||
|
||
# 2) 前置通知的 socketio emit 已在上述注入 inject_multi_agent_master_message 内
|
||
# 同时完成，避免在多智能体消息场景重复发送相同事件。
|
||
|
||
# 3) 最后一条触发新一轮工作
|
||
ui_defaults = dict(_user_message_ui_defaults(message_source, auto_user_message_event=True))
|
||
ui_defaults["starts_work"] = False
|
||
|
||
workspace_id = getattr(workspace, "workspace_id", None) or "default"
|
||
host_mode = bool(getattr(workspace, "username", None) == "host")
|
||
session_data = {
|
||
"username": username,
|
||
"role": getattr(web_terminal, "user_role", "user"),
|
||
"is_api_user": getattr(web_terminal, "user_role", "") == "api",
|
||
"host_mode": host_mode,
|
||
"host_workspace_id": workspace_id if host_mode else None,
|
||
"workspace_id": workspace_id,
|
||
"run_mode": getattr(web_terminal, "run_mode", None),
|
||
"thinking_mode": getattr(web_terminal, "thinking_mode", None),
|
||
"model_key": getattr(web_terminal, "model_key", None),
|
||
"message_source": message_source,
|
||
}
|
||
ma_auto_type = _auto_message_type_for_multi_agent_subtype(last["subtype"])
|
||
# 重要：ui_defaults 默认会给出 visibility="compact"/starts_work=False，
|
||
# 多智能体主消息是正常聊天消息，必须在 **ui_defaults 之后覆写为 chat，
|
||
# 否则后端历史 metadata.visibility=compact 会让加载的消息走通知渲染。
|
||
session_data["auto_user_message_event"] = True
|
||
session_data["auto_user_message_payload"] = {
|
||
**ui_defaults,
|
||
"message_source": message_source,
|
||
"sub_agent_notice": True,
|
||
"multi_agent_message": True,
|
||
"multi_agent_display_name": last["display_name"],
|
||
"multi_agent_subtype": last["subtype"],
|
||
"auto_message_type": ma_auto_type,
|
||
"starts_work": False,
|
||
"visibility": "chat",
|
||
"timestamp": datetime.now().isoformat(),
|
||
}
|
||
ma_debug(
|
||
"dispatch_ma_idle_before_create_task",
|
||
conversation_id=conversation_id,
|
||
last_text_preview=last["text"][:300],
|
||
workspace_id=workspace_id,
|
||
username=username,
|
||
)
|
||
session_data["auto_user_message_payload"].setdefault("metadata", {
|
||
**ui_defaults,
|
||
"message_source": message_source,
|
||
"auto_message_type": ma_auto_type,
|
||
"multi_agent_message": True,
|
||
"multi_agent_display_name": last["display_name"],
|
||
"multi_agent_subtype": last["subtype"],
|
||
"starts_work": False,
|
||
"visibility": "chat",
|
||
})
|
||
if len(parsed_messages) > 1:
|
||
session_data["preceding_user_notices"] = [
|
||
{
|
||
"message": item["text"],
|
||
"payload": {
|
||
"message_source": "sub_agent",
|
||
"sub_agent_notice": True,
|
||
"multi_agent_message": True,
|
||
"multi_agent_display_name": item["display_name"],
|
||
"multi_agent_subtype": item["subtype"],
|
||
"auto_message_type": _auto_message_type_for_multi_agent_subtype(item["subtype"]),
|
||
"visibility": "chat",
|
||
"starts_work": False,
|
||
},
|
||
}
|
||
for item in parsed_messages[:-1]
|
||
]
|
||
|
||
try:
|
||
rec = task_manager.create_chat_task(
|
||
username,
|
||
workspace_id,
|
||
last["text"],
|
||
[],
|
||
conversation_id,
|
||
model_key=session_data.get("model_key"),
|
||
thinking_mode=session_data.get("thinking_mode"),
|
||
run_mode=session_data.get("run_mode"),
|
||
session_data=session_data,
|
||
task_type="notice",
|
||
)
|
||
except Exception as exc:
|
||
ma_debug(
|
||
"dispatch_ma_idle_create_task_exception",
|
||
conversation_id=conversation_id,
|
||
error=str(exc),
|
||
error_type=type(exc).__name__,
|
||
)
|
||
raise
|
||
task_id_created = getattr(rec, "task_id", None)
|
||
ma_debug("dispatch_ma_idle_task_created", conversation_id=conversation_id, task_id=task_id_created)
|
||
|
||
payload = {
|
||
"message": last["text"],
|
||
"content": last["text"],
|
||
"conversation_id": conversation_id,
|
||
"task_id": getattr(rec, "task_id", None),
|
||
"message_source": message_source,
|
||
"sub_agent_notice": True,
|
||
"multi_agent_message": True,
|
||
"multi_agent_display_name": last["display_name"],
|
||
"multi_agent_subtype": last["subtype"],
|
||
"visibility": "chat",
|
||
"starts_work": False,
|
||
"auto_message_type": _auto_message_type_for_multi_agent_subtype(last["subtype"]),
|
||
"metadata": {
|
||
"message_source": message_source,
|
||
"auto_message_type": _auto_message_type_for_multi_agent_subtype(last["subtype"]),
|
||
"multi_agent_message": True,
|
||
"multi_agent_display_name": last["display_name"],
|
||
"multi_agent_subtype": last["subtype"],
|
||
"visibility": "chat",
|
||
"starts_work": False,
|
||
},
|
||
"timestamp": datetime.now().isoformat(),
|
||
}
|
||
sender("user_message", payload)
|
||
ma_debug(
|
||
"dispatch_ma_idle_sender_user_message",
|
||
conversation_id=conversation_id,
|
||
task_id=getattr(rec, "task_id", None),
|
||
payload_message_preview=last["text"][:200],
|
||
)
|
||
ma_debug(
|
||
"dispatch_ma_idle_exit_ok",
|
||
conversation_id=conversation_id,
|
||
count=len(parsed_messages),
|
||
task_id=getattr(rec, "task_id", None),
|
||
)
|
||
|
||
|
||
async def handle_task_with_sender(
|
||
terminal: WebTerminal,
|
||
workspace: UserWorkspace,
|
||
message,
|
||
images,
|
||
sender,
|
||
client_sid,
|
||
username: str,
|
||
videos=None,
|
||
auto_user_message_event: bool = False,
|
||
):
|
||
"""处理任务并发送消息 - 集成token统计版本"""
|
||
from .extensions import socketio
|
||
|
||
web_terminal = terminal
|
||
conversation_id = getattr(web_terminal.context_manager, "current_conversation_id", None)
|
||
# 多智能体模式：标记主对话任务正在运行，供后台通知池判断是否可以安全消费
|
||
if getattr(web_terminal, "multi_agent_mode", False):
|
||
web_terminal._multi_agent_main_task_active = True
|
||
manager = getattr(web_terminal, "sub_agent_manager", None)
|
||
pending_count = 0
|
||
if manager:
|
||
state = manager.get_multi_agent_state(conversation_id)
|
||
if state:
|
||
pending_count = len(state.pending_master_messages)
|
||
ma_debug(
|
||
"handle_task_with_sender_start",
|
||
conversation_id=conversation_id,
|
||
terminal_id=id(web_terminal),
|
||
manager_id=id(manager) if manager else None,
|
||
message_preview=str(message)[:300] if message else None,
|
||
pending_master_messages_count=pending_count,
|
||
)
|
||
videos = videos or []
|
||
raw_sender = sender
|
||
|
||
def sender(event_type, data):
|
||
"""为关键事件补充会话标识，便于前端定位报错归属。"""
|
||
if not isinstance(data, dict):
|
||
raw_sender(event_type, data)
|
||
return
|
||
payload = dict(data)
|
||
current_conv = conversation_id or getattr(web_terminal.context_manager, "current_conversation_id", None)
|
||
|
||
# 为所有事件添加 conversation_id，确保前端能正确匹配
|
||
if current_conv and event_type not in {"connect", "disconnect", "system_ready"}:
|
||
payload.setdefault("conversation_id", current_conv)
|
||
|
||
# 调试信息：记录关键事件
|
||
if event_type in {"user_message", "ai_message_start", "text_start", "text_chunk", "tool_preparing"}:
|
||
debug_log(f"[SENDER] 发送事件: {event_type}, conversation_id={current_conv}, data_keys={list(payload.keys())}")
|
||
|
||
# 为关键事件添加额外的标识信息
|
||
if event_type in {"error", "quota_exceeded", "task_stopped", "task_complete"}:
|
||
task_id = getattr(web_terminal, "task_id", None) or client_sid
|
||
if task_id:
|
||
payload.setdefault("task_id", task_id)
|
||
if client_sid:
|
||
payload.setdefault("client_sid", client_sid)
|
||
|
||
raw_sender(event_type, payload)
|
||
|
||
# 如果是思考模式，重置状态
|
||
if web_terminal.thinking_mode:
|
||
web_terminal.api_client.start_new_task(force_deep=web_terminal.deep_thinking_mode)
|
||
state = get_thinking_state(web_terminal)
|
||
state["fast_streak"] = 0
|
||
state["force_next"] = False
|
||
state["suppress_next"] = False
|
||
|
||
# 添加到对话历史
|
||
user_work_started_at = datetime.now().isoformat()
|
||
user_message_index = -1
|
||
current_work_user_message_index = -1
|
||
finalized_work_message_indices: set[int] = set()
|
||
history_len_before = len(getattr(web_terminal.context_manager, "conversation_history", []) or [])
|
||
is_first_user_message = history_len_before == 0
|
||
# 构建 user 消息来源与 metadata
|
||
source_from_terminal = str(getattr(web_terminal, "_current_user_message_source", "user") or "user").strip().lower()
|
||
user_message_source = source_from_terminal or "user"
|
||
multi_agent_meta: Dict[str, Any] = {}
|
||
if auto_user_message_event:
|
||
auto_payload = getattr(web_terminal, "_auto_user_message_payload", None)
|
||
if isinstance(auto_payload, dict):
|
||
src = str(auto_payload.get("message_source") or "").strip().lower()
|
||
if src:
|
||
user_message_source = src
|
||
elif auto_payload.get("runtime_mode_notice"):
|
||
user_message_source = "notify"
|
||
elif auto_payload.get("runtime_guidance"):
|
||
user_message_source = "guidance"
|
||
elif auto_payload.get("background_command_notice"):
|
||
user_message_source = "background_command"
|
||
elif auto_payload.get("sub_agent_notice"):
|
||
user_message_source = "sub_agent"
|
||
# 保留多智能体元数据，避免前端把子智能体输出渲染成普通 user 消息
|
||
if auto_payload.get("multi_agent_message"):
|
||
multi_agent_meta["multi_agent_message"] = True
|
||
multi_agent_meta["multi_agent_display_name"] = auto_payload.get("multi_agent_display_name")
|
||
multi_agent_meta["multi_agent_subtype"] = auto_payload.get("multi_agent_subtype")
|
||
if user_message_source not in _VALID_USER_MESSAGE_SOURCES:
|
||
user_message_source = "user"
|
||
|
||
# 构建user消息metadata
|
||
user_message_metadata = {
|
||
"message_source": user_message_source,
|
||
}
|
||
user_message_metadata.update(
|
||
_user_message_ui_defaults(
|
||
user_message_source,
|
||
auto_user_message_event=bool(auto_user_message_event),
|
||
)
|
||
)
|
||
if user_message_metadata.get("starts_work") is True:
|
||
user_message_metadata["work_timer"] = {
|
||
"status": "working",
|
||
"started_at": user_work_started_at
|
||
}
|
||
# 如果是自动发送的user消息（子智能体/后台命令完成通知），添加标记
|
||
if auto_user_message_event:
|
||
user_message_metadata["is_auto_generated"] = True
|
||
if multi_agent_meta.get("multi_agent_message"):
|
||
# 多智能体消息：恢复专用 auto_message_type 与显示字段
|
||
user_message_metadata["auto_message_type"] = _auto_message_type_for_multi_agent_subtype(
|
||
multi_agent_meta.get("multi_agent_subtype")
|
||
)
|
||
user_message_metadata.update(multi_agent_meta)
|
||
# 多智能体消息是正常聊天消息，走多智能体动态专用渲染，
|
||
# 不应走 _user_message_ui_defaults 默认给出的 visibility="compact"（通知渲染）。
|
||
user_message_metadata["visibility"] = "chat"
|
||
# 多智能体消息触发的是主智能体对子智能体输出的续答，不应显示新的
|
||
# assistant 回复头部（Astrion/工作时间），因此 starts_work=false。
|
||
# 前端通过单独的 multi_agent_message 标记恢复轮询。
|
||
user_message_metadata["starts_work"] = False
|
||
else:
|
||
user_message_metadata["auto_message_type"] = "completion_notice"
|
||
saved_user_message = web_terminal.context_manager.add_conversation(
|
||
"user",
|
||
message,
|
||
images=images,
|
||
videos=videos,
|
||
metadata=user_message_metadata
|
||
)
|
||
# 为浅备份 track_edit 提供当前用户消息 ID，避免文件修改被归到上一个输入。
|
||
try:
|
||
web_terminal.context_manager.current_shallow_message_id = (
|
||
saved_user_message.get("message_id") if isinstance(saved_user_message, dict) else None
|
||
)
|
||
except Exception:
|
||
web_terminal.context_manager.current_shallow_message_id = None
|
||
try:
|
||
user_message_index = len(getattr(web_terminal.context_manager, "conversation_history", []) or []) - 1
|
||
except Exception:
|
||
user_message_index = -1
|
||
current_work_user_message_index = user_message_index
|
||
skill_context_messages = getattr(web_terminal, "_skill_context_messages", None)
|
||
if not auto_user_message_event and isinstance(skill_context_messages, list):
|
||
for skill_item in skill_context_messages:
|
||
if not isinstance(skill_item, dict):
|
||
continue
|
||
skill_content = str(skill_item.get("content") or "")
|
||
if not skill_content:
|
||
continue
|
||
skill_name = str(skill_item.get("name") or "").strip()
|
||
skill_path = str(skill_item.get("path") or "").strip()
|
||
web_terminal.context_manager.add_conversation(
|
||
"user",
|
||
f"[系统通知|skill]\n{skill_content}",
|
||
metadata={
|
||
"source": "skill",
|
||
"message_source": "skill",
|
||
"visibility": "hidden",
|
||
"starts_work": False,
|
||
"hidden": True,
|
||
"skill_name": skill_name,
|
||
"skill_path": skill_path,
|
||
}
|
||
)
|
||
if not auto_user_message_event:
|
||
media_paths_message = _build_media_paths_message(images, videos)
|
||
if media_paths_message:
|
||
web_terminal.context_manager.add_conversation(
|
||
"user",
|
||
media_paths_message,
|
||
metadata={
|
||
"source": "media_paths",
|
||
"message_source": "media_paths",
|
||
"visibility": "hidden",
|
||
"starts_work": False,
|
||
"hidden": True,
|
||
}
|
||
)
|
||
try:
|
||
sender(
|
||
'user_message',
|
||
{
|
||
"message": message,
|
||
"images": (saved_user_message or {}).get("images") or images or [],
|
||
"videos": (saved_user_message or {}).get("videos") or videos or [],
|
||
"media_refs": (saved_user_message or {}).get("media_refs") or [],
|
||
"message_source": user_message_source,
|
||
"visibility": user_message_metadata.get("visibility"),
|
||
"starts_work": user_message_metadata.get("starts_work"),
|
||
"metadata": user_message_metadata,
|
||
"conversation_id": conversation_id,
|
||
"timestamp": (saved_user_message or {}).get("timestamp") or user_work_started_at,
|
||
},
|
||
)
|
||
except Exception as exc:
|
||
debug_log(f"[TaskFlow] 发送 user_message 回显失败: {exc}")
|
||
_prepare_hidden_versioning_baseline_for_first_input(
|
||
web_terminal=web_terminal,
|
||
workspace=workspace,
|
||
conversation_id=conversation_id,
|
||
message=message,
|
||
auto_user_message_event=bool(auto_user_message_event),
|
||
)
|
||
|
||
def finalize_user_work_timer(index: Optional[int] = None):
|
||
target_index = current_work_user_message_index if index is None else index
|
||
if target_index in finalized_work_message_indices:
|
||
return
|
||
history = getattr(web_terminal.context_manager, "conversation_history", None) or []
|
||
if target_index < 0 or target_index >= len(history):
|
||
return
|
||
target_msg = history[target_index] or {}
|
||
if target_msg.get("role") != "user":
|
||
return
|
||
metadata = target_msg.get("metadata") or {}
|
||
timer = metadata.get("work_timer")
|
||
if not isinstance(timer, dict):
|
||
return
|
||
started_at = timer.get("started_at") or target_msg.get("timestamp") or user_work_started_at
|
||
start_ts = None
|
||
try:
|
||
start_ts = datetime.fromisoformat(str(started_at).replace("Z", "+00:00")).timestamp()
|
||
except Exception:
|
||
start_ts = None
|
||
now_ts = time.time()
|
||
duration_ms = int(max(0.0, (now_ts - start_ts) * 1000.0)) if start_ts is not None else 0
|
||
timer.update({
|
||
"status": "completed",
|
||
"started_at": started_at,
|
||
"finished_at": datetime.now().isoformat(),
|
||
"duration_ms": duration_ms
|
||
})
|
||
metadata["work_timer"] = timer
|
||
target_msg["metadata"] = metadata
|
||
history[target_index] = target_msg
|
||
web_terminal.context_manager.auto_save_conversation(force=True)
|
||
finalized_work_message_indices.add(target_index)
|
||
|
||
def switch_current_work_timer_to_latest_user():
|
||
nonlocal current_work_user_message_index
|
||
history = getattr(web_terminal.context_manager, "conversation_history", None) or []
|
||
for idx in range(len(history) - 1, -1, -1):
|
||
msg = history[idx] or {}
|
||
if msg.get("role") == "user":
|
||
metadata = msg.get("metadata") or {}
|
||
if isinstance(metadata.get("work_timer"), dict):
|
||
current_work_user_message_index = idx
|
||
return
|
||
|
||
versioning_checkpoint_recorded = False
|
||
|
||
def finalize_run_versioning_checkpoint(run_status: str = "completed"):
|
||
nonlocal versioning_checkpoint_recorded
|
||
if versioning_checkpoint_recorded:
|
||
return
|
||
_record_hidden_versioning_checkpoint_after_run(
|
||
web_terminal=web_terminal,
|
||
workspace=workspace,
|
||
conversation_id=conversation_id,
|
||
message=message,
|
||
message_index=user_message_index,
|
||
auto_user_message_event=bool(auto_user_message_event),
|
||
run_status=run_status,
|
||
)
|
||
versioning_checkpoint_recorded = True
|
||
|
||
# Skill 提示系统：检测关键词并在用户消息之后插入 system 消息
|
||
try:
|
||
personal_config = load_personalization_config(workspace.data_dir)
|
||
skill_hints_enabled = personal_config.get("skill_hints_enabled", False)
|
||
|
||
if skill_hints_enabled and message:
|
||
hint_manager = SkillHintManager()
|
||
hint_manager.set_enabled(True)
|
||
hint_messages = hint_manager.build_hint_messages(message)
|
||
|
||
# 将提示消息插入到对话历史中（在用户消息之后）
|
||
for hint_msg in hint_messages:
|
||
debug_log(f"[Skill Hints] 插入提示消息: {hint_msg['content'][:100]}")
|
||
web_terminal.context_manager.add_conversation(
|
||
"system",
|
||
hint_msg["content"]
|
||
)
|
||
# 验证插入后的消息
|
||
last_msg = web_terminal.context_manager.conversation_history[-1]
|
||
debug_log(f"[Skill Hints] 插入后验证 - role: {last_msg.get('role')}, content: {last_msg.get('content')[:100]}")
|
||
except Exception as exc:
|
||
debug_log(f"Skill hints 处理失败: {exc}")
|
||
|
||
if is_first_user_message and getattr(web_terminal, "context_manager", None):
|
||
try:
|
||
personal_config = load_personalization_config(workspace.data_dir)
|
||
except Exception:
|
||
personal_config = {}
|
||
auto_title_enabled = personal_config.get("auto_generate_title", True)
|
||
skip_auto_title_generation = bool(
|
||
(web_terminal.context_manager.conversation_metadata or {}).get("skip_auto_title_generation", False)
|
||
)
|
||
if auto_title_enabled and not skip_auto_title_generation:
|
||
conv_id = getattr(web_terminal.context_manager, "current_conversation_id", None)
|
||
socketio.start_background_task(
|
||
generate_conversation_title_background,
|
||
web_terminal,
|
||
conv_id,
|
||
message,
|
||
username
|
||
)
|
||
|
||
# 自动深层压缩（用户输入后触发）
|
||
try:
|
||
personal_config = load_personalization_config(workspace.data_dir)
|
||
except Exception:
|
||
personal_config = {}
|
||
compression_settings = resolve_context_compression_settings(personal_config)
|
||
auto_deep_enabled = bool(personal_config.get("auto_deep_compress_enabled", False))
|
||
current_tokens_for_deep = web_terminal.context_manager.get_current_context_tokens(conversation_id)
|
||
if (
|
||
auto_deep_enabled
|
||
and current_tokens_for_deep > compression_settings["deep_trigger_tokens"]
|
||
and not web_terminal.context_manager.is_compression_in_progress()
|
||
):
|
||
web_terminal.context_manager._set_meta_flag("is_ultra_long_conversation", True)
|
||
sender('compression_state', {
|
||
"conversation_id": conversation_id,
|
||
"in_progress": True,
|
||
"mode": "auto",
|
||
"stage": "queued"
|
||
})
|
||
deep_result = await run_deep_compression(
|
||
web_terminal=web_terminal,
|
||
workspace=workspace,
|
||
conversation_id=conversation_id,
|
||
mode="auto",
|
||
sender=sender,
|
||
)
|
||
if not deep_result.get("success"):
|
||
sender('error', {
|
||
"message": deep_result.get("error") or "自动深层压缩失败",
|
||
"conversation_id": conversation_id,
|
||
})
|
||
else:
|
||
guide_message = (deep_result.get("guide_message") or "").strip()
|
||
if guide_message:
|
||
finalize_user_work_timer()
|
||
finalize_run_versioning_checkpoint("auto_deep_compress_handoff")
|
||
# 压缩后的续接引导语单独标记来源，使其以 compact 形式展示，
|
||
# 而不是混入普通用户消息流。
|
||
setattr(web_terminal, "_current_user_message_source", "compression_handoff")
|
||
try:
|
||
await handle_task_with_sender(
|
||
web_terminal,
|
||
workspace,
|
||
guide_message,
|
||
[],
|
||
sender,
|
||
client_sid,
|
||
username,
|
||
[]
|
||
)
|
||
finally:
|
||
setattr(web_terminal, "_current_user_message_source", "user")
|
||
return
|
||
finalize_user_work_timer()
|
||
finalize_run_versioning_checkpoint("auto_deep_compress_end")
|
||
return
|
||
|
||
# === 移除：不在这里计算输入token，改为在每次API调用前计算 ===
|
||
|
||
# 构建上下文和消息（用于API调用）
|
||
context = web_terminal.build_context()
|
||
messages = web_terminal.build_messages(context, message)
|
||
tools = web_terminal.define_tools()
|
||
|
||
# === 目标模式（Goal Mode）启动 / 续注入 ===
|
||
# 1) 用户请求开启目标模式 → 无条件覆盖工作区旧目标状态并启动新目标。
|
||
# 2) 工作区已存在活动目标（含压缩后重入、子智能体通知重入）→ 重新注入提示词，
|
||
# 确保主模型在长运行/压缩后仍“知道自己处于目标模式”。
|
||
try:
|
||
if bool(getattr(web_terminal, "_goal_mode_requested", False)):
|
||
maybe_start_goal(
|
||
web_terminal=web_terminal,
|
||
workspace=workspace,
|
||
conversation_id=conversation_id,
|
||
goal_text=message,
|
||
current_tool_calls=0,
|
||
)
|
||
# 本标记只表示“本次用户显式开启目标模式”。目标状态已写入
|
||
# goal_state.json 后必须立即消费掉，否则自动深层压缩后的递归
|
||
# handoff 会再次进入 handle_task_with_sender，把压缩引导语误当作
|
||
# 新目标并重置目标文本、轮数、token、工具次数和开始时间。
|
||
setattr(web_terminal, "_goal_mode_requested", False)
|
||
if goal_is_active(workspace):
|
||
inject_goal_prompt(
|
||
web_terminal=web_terminal,
|
||
messages=messages,
|
||
sender=sender,
|
||
conversation_id=conversation_id,
|
||
)
|
||
emit_goal_progress(
|
||
web_terminal=web_terminal,
|
||
workspace=workspace,
|
||
sender=sender,
|
||
total_tool_calls=0,
|
||
)
|
||
except Exception as exc:
|
||
debug_log(f"[Goal] 启动/注入失败: {exc}")
|
||
|
||
try:
|
||
profile = get_model_profile(getattr(web_terminal, "model_key", None))
|
||
web_terminal.apply_model_profile(profile)
|
||
except Exception as exc:
|
||
debug_log(f"更新模型配置失败: {exc}")
|
||
|
||
# === 上下文预算与安全校验（避免超出模型上下文） ===
|
||
max_context_tokens = get_model_context_window(getattr(web_terminal, "model_key", None))
|
||
current_tokens = web_terminal.context_manager.get_current_context_tokens(conversation_id)
|
||
# 提前同步给底层客户端，动态收缩 max_tokens
|
||
web_terminal.api_client.update_context_budget(current_tokens, max_context_tokens)
|
||
if max_context_tokens:
|
||
if current_tokens >= max_context_tokens:
|
||
err_msg = (
|
||
f"当前对话上下文已达 {current_tokens} tokens，超过模型上限 "
|
||
f"{max_context_tokens}，请先使用压缩功能或清理对话后再试。"
|
||
)
|
||
debug_log(err_msg)
|
||
web_terminal.context_manager.add_conversation("system", err_msg)
|
||
sender('error', {
|
||
'message': err_msg,
|
||
'status_code': 400,
|
||
'error_type': 'context_overflow'
|
||
})
|
||
finalize_user_work_timer()
|
||
finalize_run_versioning_checkpoint("context_overflow")
|
||
return
|
||
usage_percent = (current_tokens / max_context_tokens) * 100
|
||
warned = web_terminal.context_manager.conversation_metadata.get("context_warning_sent", False)
|
||
if usage_percent >= 70 and not warned:
|
||
warn_msg = (
|
||
f"当前对话上下文约占 {usage_percent:.1f}%（{current_tokens}/{max_context_tokens}），"
|
||
"建议使用压缩功能。"
|
||
)
|
||
web_terminal.context_manager.conversation_metadata["context_warning_sent"] = True
|
||
web_terminal.context_manager.auto_save_conversation(force=True)
|
||
sender('context_warning', {
|
||
'title': '上下文过长',
|
||
'message': warn_msg,
|
||
'type': 'warning',
|
||
'conversation_id': conversation_id
|
||
})
|
||
|
||
# 开始新的AI消息
|
||
sender('ai_message_start', {})
|
||
|
||
# 增量保存相关变量
|
||
accumulated_response = "" # 累积的响应内容
|
||
is_first_iteration = True # 是否是第一次迭代
|
||
|
||
# 统计和限制变量
|
||
total_iterations = 0
|
||
total_tool_calls = 0
|
||
consecutive_same_tool = defaultdict(int)
|
||
last_tool_name = ""
|
||
auto_fix_attempts = 0
|
||
last_tool_call_time = 0
|
||
detected_tool_intent: Dict[str, str] = {}
|
||
# 目标模式：记录本“目标轮段”内主模型是否调用过工具（空转保护用）。
|
||
# 每次续命注入后重置为 False，直到下一次 no-tool-call break。
|
||
goal_segment_made_tool_call = False
|
||
|
||
# 设置最大迭代次数（API 可覆盖）；None 表示不限制
|
||
max_iterations_override = getattr(web_terminal, "max_iterations_override", None)
|
||
max_iterations = max_iterations_override if max_iterations_override is not None else MAX_ITERATIONS_PER_TASK
|
||
max_api_retries = 4
|
||
retry_delay_seconds = 10
|
||
|
||
|
||
iteration = 0
|
||
while max_iterations is None or iteration < max_iterations:
|
||
# 检查停止标志
|
||
stop_entry = get_stop_flag(client_sid, username)
|
||
if stop_entry and stop_entry.get('stop'):
|
||
debug_log(f"[Task] 检测到停止标志，退出循环")
|
||
try:
|
||
stop_goal_user_cancel(
|
||
web_terminal=web_terminal,
|
||
workspace=workspace,
|
||
sender=sender,
|
||
total_tool_calls=total_tool_calls,
|
||
)
|
||
except Exception as exc:
|
||
debug_log(f"[Goal] 用户停止时停止目标失败: {exc}")
|
||
# 计算后台任务状态，前端用这个字段决定是否保持对话运行态、输入区是否释放
|
||
_stopped_has_bg = False
|
||
_stopped_has_bg_cmd = False
|
||
_stopped_has_ma = False
|
||
_manager = getattr(web_terminal, "sub_agent_manager", None)
|
||
if _manager:
|
||
try:
|
||
_manager.reconcile_task_states(conversation_id=conversation_id)
|
||
except Exception:
|
||
pass
|
||
if getattr(web_terminal, "multi_agent_mode", False):
|
||
_state = _manager.get_multi_agent_state(conversation_id)
|
||
if _state:
|
||
_stopped_has_ma = any(a.status == "running" for a in _state.list_all())
|
||
for _t in _manager.tasks.values():
|
||
if _t.get("conversation_id") != conversation_id:
|
||
continue
|
||
if _t.get("status") in TERMINAL_STATUSES.union({"terminated"}):
|
||
continue
|
||
if _t.get("multi_agent_mode"):
|
||
continue
|
||
_stopped_has_bg = True
|
||
break
|
||
_bg_mgr = getattr(web_terminal, "background_command_manager", None)
|
||
if _bg_mgr and conversation_id:
|
||
try:
|
||
_bg_mgr.reconcile_stale_records(conversation_id=conversation_id)
|
||
except Exception:
|
||
pass
|
||
if _bg_mgr.list_waiting_items(conversation_id):
|
||
_stopped_has_bg_cmd = True
|
||
sender('task_stopped', {
|
||
'message': '任务已停止',
|
||
'reason': 'user_requested',
|
||
'has_running_sub_agents': _stopped_has_bg,
|
||
'has_running_background_commands': _stopped_has_bg_cmd,
|
||
'has_running_multi_agent': _stopped_has_ma,
|
||
})
|
||
break
|
||
|
||
current_iteration = iteration + 1
|
||
iteration += 1
|
||
total_iterations += 1
|
||
iteration_limit_label = max_iterations if max_iterations is not None else "∞"
|
||
debug_log(f"\n--- 迭代 {current_iteration}/{iteration_limit_label} 开始 ---")
|
||
|
||
# 检查是否超过总工具调用限制
|
||
if MAX_TOTAL_TOOL_CALLS is not None and total_tool_calls >= MAX_TOTAL_TOOL_CALLS:
|
||
debug_log(f"已达到最大工具调用次数限制 ({MAX_TOTAL_TOOL_CALLS})")
|
||
sender('system_message', {
|
||
'content': f'⚠️ 已达到最大工具调用次数限制 ({MAX_TOTAL_TOOL_CALLS})，任务结束。'
|
||
})
|
||
mark_force_thinking(web_terminal, reason="tool_limit")
|
||
break
|
||
|
||
apply_thinking_schedule(web_terminal)
|
||
|
||
full_response = ""
|
||
tool_calls = []
|
||
current_thinking = ""
|
||
detected_tools = {}
|
||
last_usage_payload = None
|
||
|
||
# 状态标志
|
||
in_thinking = False
|
||
thinking_started = False
|
||
thinking_ended = False
|
||
text_started = False
|
||
text_has_content = False
|
||
text_streaming = False
|
||
text_chunk_index = 0
|
||
last_text_chunk_time: Optional[float] = None
|
||
|
||
# 计数器
|
||
chunk_count = 0
|
||
reasoning_chunks = 0
|
||
content_chunks = 0
|
||
tool_chunks = 0
|
||
last_finish_reason = None
|
||
|
||
thinking_expected = web_terminal.api_client.get_current_thinking_mode()
|
||
debug_log(f"思考模式: {thinking_expected}")
|
||
quota_allowed = True
|
||
quota_info = {}
|
||
if hasattr(web_terminal, "record_model_call"):
|
||
quota_allowed, quota_info = web_terminal.record_model_call(bool(thinking_expected))
|
||
if not quota_allowed:
|
||
quota_type = 'thinking' if thinking_expected else 'fast'
|
||
socketio.emit('quota_notice', {
|
||
'type': quota_type,
|
||
'reset_at': quota_info.get('reset_at'),
|
||
'limit': quota_info.get('limit'),
|
||
'count': quota_info.get('count')
|
||
}, room=f"user_{getattr(web_terminal, 'username', '')}")
|
||
sender('quota_exceeded', {
|
||
'type': quota_type,
|
||
'reset_at': quota_info.get('reset_at')
|
||
})
|
||
sender('error', {
|
||
'message': "配额已达到上限，暂时无法继续调用模型。",
|
||
'quota': quota_info
|
||
})
|
||
finalize_user_work_timer()
|
||
finalize_run_versioning_checkpoint("quota_exceeded")
|
||
return
|
||
|
||
tool_call_limit_label = MAX_TOTAL_TOOL_CALLS if MAX_TOTAL_TOOL_CALLS is not None else "∞"
|
||
print(f"[API] 第{current_iteration}次调用 (总工具调用: {total_tool_calls}/{tool_call_limit_label})")
|
||
|
||
stream_result = await run_streaming_attempts(
|
||
web_terminal=web_terminal,
|
||
messages=messages,
|
||
tools=tools,
|
||
sender=sender,
|
||
client_sid=client_sid,
|
||
username=username,
|
||
conversation_id=conversation_id,
|
||
current_iteration=current_iteration,
|
||
max_api_retries=max_api_retries,
|
||
retry_delay_seconds=retry_delay_seconds,
|
||
detected_tool_intent=detected_tool_intent,
|
||
full_response=full_response,
|
||
tool_calls=tool_calls,
|
||
current_thinking=current_thinking,
|
||
detected_tools=detected_tools,
|
||
last_usage_payload=last_usage_payload,
|
||
in_thinking=in_thinking,
|
||
thinking_started=thinking_started,
|
||
thinking_ended=thinking_ended,
|
||
text_started=text_started,
|
||
text_has_content=text_has_content,
|
||
text_streaming=text_streaming,
|
||
text_chunk_index=text_chunk_index,
|
||
last_text_chunk_time=last_text_chunk_time,
|
||
chunk_count=chunk_count,
|
||
reasoning_chunks=reasoning_chunks,
|
||
content_chunks=content_chunks,
|
||
tool_chunks=tool_chunks,
|
||
last_finish_reason=last_finish_reason,
|
||
accumulated_response=accumulated_response,
|
||
)
|
||
if stream_result.get("stopped"):
|
||
finalize_user_work_timer()
|
||
finalize_run_versioning_checkpoint("stopped")
|
||
return
|
||
|
||
full_response = stream_result["full_response"]
|
||
tool_calls = stream_result["tool_calls"]
|
||
current_thinking = stream_result["current_thinking"]
|
||
detected_tools = stream_result["detected_tools"]
|
||
last_usage_payload = stream_result["last_usage_payload"]
|
||
in_thinking = stream_result["in_thinking"]
|
||
thinking_started = stream_result["thinking_started"]
|
||
thinking_ended = stream_result["thinking_ended"]
|
||
text_started = stream_result["text_started"]
|
||
text_has_content = stream_result["text_has_content"]
|
||
text_streaming = stream_result["text_streaming"]
|
||
text_chunk_index = stream_result["text_chunk_index"]
|
||
last_text_chunk_time = stream_result["last_text_chunk_time"]
|
||
chunk_count = stream_result["chunk_count"]
|
||
reasoning_chunks = stream_result["reasoning_chunks"]
|
||
content_chunks = stream_result["content_chunks"]
|
||
tool_chunks = stream_result["tool_chunks"]
|
||
last_finish_reason = stream_result["last_finish_reason"]
|
||
accumulated_response = stream_result["accumulated_response"]
|
||
|
||
# 流结束后的处理
|
||
debug_log(f"\n流结束统计:")
|
||
debug_log(f" 总chunks: {chunk_count}")
|
||
debug_log(f" 思考chunks: {reasoning_chunks}")
|
||
debug_log(f" 内容chunks: {content_chunks}")
|
||
debug_log(f" 工具chunks: {tool_chunks}")
|
||
debug_log(f" 收集到的思考: {len(current_thinking)} 字符")
|
||
debug_log(f" 收集到的正文: {len(full_response)} 字符")
|
||
debug_log(f" 收集到的工具: {len(tool_calls)} 个")
|
||
|
||
# 结束未完成的流
|
||
if in_thinking and not thinking_ended:
|
||
sender('thinking_end', {'full_content': current_thinking})
|
||
await asyncio.sleep(0.1)
|
||
|
||
|
||
# 确保text_end事件被发送
|
||
if text_started and text_has_content:
|
||
debug_log(f"发送text_end事件，完整内容长度: {len(full_response)}")
|
||
sender('text_end', {'full_content': full_response})
|
||
await asyncio.sleep(0.1)
|
||
text_streaming = False
|
||
|
||
if full_response.strip():
|
||
debug_log(f"流式文本内容长度: {len(full_response)} 字符")
|
||
|
||
if web_terminal.api_client.last_call_used_thinking and current_thinking:
|
||
web_terminal.api_client.current_task_thinking = current_thinking or ""
|
||
if web_terminal.api_client.current_task_first_call:
|
||
web_terminal.api_client.current_task_first_call = False
|
||
update_thinking_after_call(web_terminal)
|
||
|
||
# 检测是否有格式错误的工具调用
|
||
if not tool_calls and full_response and AUTO_FIX_TOOL_CALL:
|
||
if detect_malformed_tool_call(full_response):
|
||
auto_fix_attempts += 1
|
||
|
||
if auto_fix_attempts <= AUTO_FIX_MAX_ATTEMPTS:
|
||
debug_log(f"检测到格式错误的工具调用，尝试自动修复 (尝试 {auto_fix_attempts}/{AUTO_FIX_MAX_ATTEMPTS})")
|
||
|
||
fix_message = "你使用了错误的格式输出工具调用。请使用正确的工具调用格式而不是直接输出JSON。根据当前进度继续执行任务。"
|
||
|
||
sender('system_message', {
|
||
'content': f'⚠️ 自动修复: {fix_message}'
|
||
})
|
||
maybe_mark_failure_from_message(web_terminal, f'⚠️ 自动修复: {fix_message}')
|
||
|
||
messages.append({
|
||
"role": "user",
|
||
"content": fix_message
|
||
})
|
||
|
||
await asyncio.sleep(1)
|
||
continue
|
||
else:
|
||
debug_log(f"自动修复尝试已达上限 ({AUTO_FIX_MAX_ATTEMPTS})")
|
||
sender('system_message', {
|
||
'content': f'⌘ 工具调用格式错误，自动修复失败。请手动检查并重试。'
|
||
})
|
||
maybe_mark_failure_from_message(web_terminal, '⌘ 工具调用格式错误，自动修复失败。请手动检查并重试。')
|
||
break
|
||
|
||
# 构建助手消息（用于API继续对话）
|
||
assistant_content_parts = []
|
||
|
||
if full_response:
|
||
assistant_content_parts.append(full_response)
|
||
|
||
assistant_content = "\n".join(assistant_content_parts) if assistant_content_parts else ""
|
||
|
||
# 添加到消息历史（用于API继续对话，不保存到文件）
|
||
assistant_message = {
|
||
"role": "assistant",
|
||
"content": assistant_content,
|
||
# thinking 模式下，reasoning_content 需要原样回传；即使该轮为空字符串也保留字段
|
||
"reasoning_content": current_thinking or "",
|
||
}
|
||
if tool_calls:
|
||
assistant_message["tool_calls"] = tool_calls
|
||
|
||
messages.append(assistant_message)
|
||
if assistant_content or current_thinking or tool_calls:
|
||
web_terminal.context_manager.add_conversation(
|
||
"assistant",
|
||
assistant_content,
|
||
tool_calls=tool_calls if tool_calls else None,
|
||
reasoning_content=current_thinking or ""
|
||
)
|
||
|
||
# 为下一轮迭代重置流状态标志，但保留 full_response 供上面保存使用
|
||
text_streaming = False
|
||
text_started = False
|
||
text_has_content = False
|
||
full_response = ""
|
||
|
||
if not tool_calls:
|
||
debug_log("没有工具调用，结束迭代")
|
||
# === 目标模式：turn 结束拦截 ===
|
||
try:
|
||
if goal_is_active(workspace):
|
||
goal_result = await handle_goal_after_turn(
|
||
web_terminal=web_terminal,
|
||
workspace=workspace,
|
||
messages=messages,
|
||
sender=sender,
|
||
conversation_id=conversation_id,
|
||
assistant_content=assistant_content,
|
||
made_tool_call=goal_segment_made_tool_call,
|
||
total_tool_calls=total_tool_calls,
|
||
)
|
||
action = goal_result.get("action")
|
||
if action == "continue":
|
||
# 已注入续命 user 消息：上一段 assistant work 到此结束，
|
||
# 后续计时切换到刚注入的 goal_review 消息。
|
||
finalize_user_work_timer()
|
||
switch_current_work_timer_to_latest_user()
|
||
goal_segment_made_tool_call = False
|
||
is_first_iteration = False
|
||
debug_log(f"[Goal] 续命：{goal_result.get('message', '')[:80]}")
|
||
continue
|
||
if action == "done":
|
||
debug_log("[Goal] 目标已达成，结束任务")
|
||
elif action == "stop":
|
||
debug_log(f"[Goal] 目标停止：{goal_result.get('reason')}")
|
||
except Exception as exc:
|
||
debug_log(f"[Goal] turn 结束处理失败: {exc}")
|
||
|
||
# 多智能体模式：没有工具调用时，先消费子智能体待转发到主对话的消息；
|
||
# 如果有新消息注入，继续迭代让 Team Leader 响应，而不是直接结束任务。
|
||
if getattr(web_terminal, "multi_agent_mode", False):
|
||
injected_count = await process_multi_agent_master_messages(
|
||
messages=messages,
|
||
inline=False,
|
||
web_terminal=web_terminal,
|
||
sender=sender,
|
||
debug_log=debug_log,
|
||
)
|
||
ma_debug(
|
||
"no_tool_call_turn_process_ma_messages",
|
||
conversation_id=conversation_id,
|
||
injected_count=injected_count,
|
||
)
|
||
if injected_count:
|
||
debug_log(f"[MultiAgent] no-tool-call turn 注入 {injected_count} 条子智能体消息，继续迭代")
|
||
is_first_iteration = False
|
||
continue
|
||
|
||
break
|
||
|
||
# 目标模式：本轮段确实产生了工具调用
|
||
goal_segment_made_tool_call = True
|
||
|
||
# 检查连续相同工具调用
|
||
for tc in tool_calls:
|
||
tool_name = tc["function"]["name"]
|
||
|
||
if tool_name == last_tool_name:
|
||
consecutive_same_tool[tool_name] += 1
|
||
|
||
if (
|
||
MAX_CONSECUTIVE_SAME_TOOL is not None
|
||
and consecutive_same_tool[tool_name] >= MAX_CONSECUTIVE_SAME_TOOL
|
||
):
|
||
debug_log(f"警告: 连续调用相同工具 {tool_name} 已达 {MAX_CONSECUTIVE_SAME_TOOL} 次")
|
||
sender('system_message', {
|
||
'content': f'⚠️ 检测到重复调用 {tool_name} 工具 {MAX_CONSECUTIVE_SAME_TOOL} 次，可能存在循环。'
|
||
})
|
||
maybe_mark_failure_from_message(web_terminal, f'⚠️ 检测到重复调用 {tool_name} 工具 {MAX_CONSECUTIVE_SAME_TOOL} 次，可能存在循环。')
|
||
|
||
if consecutive_same_tool[tool_name] >= MAX_CONSECUTIVE_SAME_TOOL + 2:
|
||
debug_log(f"终止: 工具 {tool_name} 调用次数过多")
|
||
sender('system_message', {
|
||
'content': f'⌘ 工具 {tool_name} 重复调用过多，任务终止。'
|
||
})
|
||
maybe_mark_failure_from_message(web_terminal, f'⌘ 工具 {tool_name} 重复调用过多，任务终止。')
|
||
break
|
||
else:
|
||
consecutive_same_tool.clear()
|
||
consecutive_same_tool[tool_name] = 1
|
||
|
||
last_tool_name = tool_name
|
||
|
||
# 更新统计
|
||
total_tool_calls += len(tool_calls)
|
||
try:
|
||
if goal_is_active(workspace):
|
||
emit_goal_progress(
|
||
web_terminal=web_terminal,
|
||
workspace=workspace,
|
||
sender=sender,
|
||
total_tool_calls=total_tool_calls,
|
||
)
|
||
except Exception as exc:
|
||
debug_log(f"[Goal] 进度广播失败: {exc}")
|
||
|
||
# 执行每个工具
|
||
tool_loop_result = await execute_tool_calls(
|
||
web_terminal=web_terminal,
|
||
tool_calls=tool_calls,
|
||
sender=sender,
|
||
messages=messages,
|
||
client_sid=client_sid,
|
||
username=username,
|
||
iteration=iteration,
|
||
conversation_id=conversation_id,
|
||
last_tool_call_time=last_tool_call_time,
|
||
process_sub_agent_updates=process_sub_agent_updates,
|
||
process_background_command_updates=process_background_command_updates,
|
||
process_multi_agent_master_messages=process_multi_agent_master_messages,
|
||
maybe_mark_failure_from_message=maybe_mark_failure_from_message,
|
||
mark_force_thinking=mark_force_thinking,
|
||
get_stop_flag=get_stop_flag,
|
||
clear_stop_flag=clear_stop_flag,
|
||
workspace=workspace,
|
||
)
|
||
last_tool_call_time = tool_loop_result.get("last_tool_call_time", last_tool_call_time)
|
||
if tool_loop_result.get("stopped"):
|
||
finalize_user_work_timer()
|
||
finalize_run_versioning_checkpoint("stopped")
|
||
return
|
||
if tool_loop_result.get("approval_rejected"):
|
||
sender('task_stopped', {
|
||
'message': tool_loop_result.get("approval_message") or '操作被用户拒绝',
|
||
'reason': 'approval_rejected',
|
||
'conversation_id': conversation_id,
|
||
'has_running_sub_agents': False,
|
||
'has_running_background_commands': False,
|
||
'has_running_multi_agent': False
|
||
})
|
||
finalize_user_work_timer()
|
||
finalize_run_versioning_checkpoint("approval_rejected")
|
||
return
|
||
if tool_loop_result.get("deep_compressed"):
|
||
deep_result = tool_loop_result.get("deep_result") or {}
|
||
guide_message = (deep_result.get("guide_message") or "").strip()
|
||
if deep_result.get("success") and guide_message:
|
||
finalize_user_work_timer()
|
||
finalize_run_versioning_checkpoint("deep_compress_handoff")
|
||
# 压缩后的续接引导语单独标记来源，使其以 compact 形式展示，
|
||
# 而不是混入普通用户消息流。
|
||
setattr(web_terminal, "_current_user_message_source", "compression_handoff")
|
||
try:
|
||
await handle_task_with_sender(
|
||
web_terminal,
|
||
workspace,
|
||
guide_message,
|
||
[],
|
||
sender,
|
||
client_sid,
|
||
username,
|
||
[]
|
||
)
|
||
finally:
|
||
setattr(web_terminal, "_current_user_message_source", "user")
|
||
return
|
||
finalize_user_work_timer()
|
||
finalize_run_versioning_checkpoint("deep_compress_end")
|
||
return
|
||
|
||
# 标记不再是第一次迭代
|
||
is_first_iteration = False
|
||
|
||
# 多智能体模式：在进入下一轮模型调用前，消费子智能体最新输出到主对话
|
||
if getattr(web_terminal, "multi_agent_mode", False):
|
||
await process_multi_agent_master_messages(
|
||
messages=messages,
|
||
inline=False,
|
||
web_terminal=web_terminal,
|
||
sender=sender,
|
||
debug_log=debug_log,
|
||
)
|
||
|
||
|
||
# 最终统计
|
||
# 多智能体模式：主对话任务结束，解除活跃标记
|
||
if getattr(web_terminal, "multi_agent_mode", False):
|
||
web_terminal._multi_agent_main_task_active = False
|
||
|
||
debug_log(f"\n{'='*40}")
|
||
debug_log(f"任务完成统计:")
|
||
debug_log(f" 总迭代次数: {total_iterations}")
|
||
debug_log(f" 总工具调用: {total_tool_calls}")
|
||
debug_log(f" 自动修复尝试: {auto_fix_attempts}")
|
||
debug_log(f" 累积响应: {len(accumulated_response)} 字符")
|
||
debug_log(f"{'='*40}\n")
|
||
|
||
# 检查是否有后台运行的子智能体或待通知的完成任务
|
||
manager = getattr(web_terminal, "sub_agent_manager", None)
|
||
has_running_sub_agents = False
|
||
has_running_multi_agent = False
|
||
bg_manager = getattr(web_terminal, "background_command_manager", None)
|
||
has_running_background_commands = False
|
||
if manager:
|
||
try:
|
||
manager.reconcile_task_states(conversation_id=conversation_id)
|
||
except Exception as exc:
|
||
debug_log(f"[SubAgent] reconcile_task_states failed: {exc}")
|
||
if not hasattr(web_terminal, "_announced_sub_agent_tasks"):
|
||
web_terminal._announced_sub_agent_tasks = set()
|
||
|
||
# 多智能体模式：检测是否还有真正运行中的实例，以及是否有尚未消费的 pending 消息。
|
||
# idle 实例已完成当前输出、保留上下文等待后续指令；pending 消息会在主任务
|
||
# 结束后由 poll_multi_agent_notifications 继续消费并触发新一轮工作。
|
||
has_pending_ma_messages = False
|
||
if getattr(web_terminal, "multi_agent_mode", False):
|
||
state = manager.get_multi_agent_state(conversation_id)
|
||
if state:
|
||
has_running_multi_agent = any(a.status == "running" for a in state.list_all())
|
||
has_pending_ma_messages = state.has_pending_master_messages()
|
||
|
||
running_tasks = [
|
||
task for task in manager.tasks.values()
|
||
if task.get("status") not in TERMINAL_STATUSES.union({"terminated"})
|
||
and task.get("run_in_background")
|
||
and not task.get("multi_agent_mode")
|
||
and task.get("conversation_id") == conversation_id
|
||
]
|
||
pending_notice_tasks = [
|
||
task for task in manager.tasks.values()
|
||
if task.get("status") in TERMINAL_STATUSES.union({"terminated"})
|
||
and task.get("run_in_background")
|
||
and not task.get("multi_agent_mode")
|
||
and task.get("conversation_id") == conversation_id
|
||
and task.get("task_id") not in web_terminal._announced_sub_agent_tasks
|
||
]
|
||
|
||
if running_tasks or pending_notice_tasks:
|
||
has_running_sub_agents = True
|
||
notify_tasks = running_tasks + pending_notice_tasks
|
||
debug_log(f"[SubAgent] 后台子智能体等待: running={len(running_tasks)} pending_notice={len(pending_notice_tasks)}")
|
||
# 先通知前端：有子智能体在运行/待通知，保持等待状态
|
||
sender('sub_agent_waiting', {
|
||
'count': len(notify_tasks),
|
||
'tasks': [{'agent_id': t.get('agent_id'), 'summary': t.get('summary')} for t in notify_tasks]
|
||
})
|
||
|
||
# 检查是否有后台 run_command 或待通知任务
|
||
if bg_manager and conversation_id:
|
||
try:
|
||
bg_manager.reconcile_stale_records(conversation_id=conversation_id)
|
||
except Exception as exc:
|
||
debug_log(f"[BgCmdDebug] reconcile_stale_records failed: {exc}")
|
||
waiting_items = bg_manager.list_waiting_items(conversation_id)
|
||
if waiting_items:
|
||
has_running_background_commands = True
|
||
# 与子智能体完全复用同一 waiting 事件（前端已有稳定处理链路）
|
||
sender('sub_agent_waiting', _build_shared_waiting_payload(waiting_items))
|
||
|
||
# 传统后台任务通知池：只处理 run_in_background=True 的子智能体 / 后台 run_command。
|
||
needs_completion_poll = (
|
||
has_running_sub_agents
|
||
or has_running_background_commands
|
||
or _has_pending_completion_work(
|
||
web_terminal=web_terminal, conversation_id=conversation_id
|
||
)
|
||
)
|
||
if needs_completion_poll:
|
||
ma_debug(
|
||
"completion_poll_start",
|
||
conversation_id=conversation_id,
|
||
has_running_sub_agents=has_running_sub_agents,
|
||
has_running_background_commands=has_running_background_commands,
|
||
)
|
||
def run_completion_poll():
|
||
import asyncio
|
||
loop = asyncio.new_event_loop()
|
||
asyncio.set_event_loop(loop)
|
||
try:
|
||
loop.run_until_complete(poll_completion_notifications(
|
||
web_terminal=web_terminal,
|
||
workspace=workspace,
|
||
conversation_id=conversation_id,
|
||
client_sid=client_sid,
|
||
username=username
|
||
))
|
||
finally:
|
||
loop.close()
|
||
|
||
socketio.start_background_task(run_completion_poll)
|
||
|
||
# 多智能体模式独立通知池：处理 running 实例 / idle 实例待消费的 pending 消息。
|
||
# 与传统后台任务完全分离，避免两者竞争 create_chat_task 的单工作区互斥。
|
||
ma_mode_flag = bool(getattr(web_terminal, "multi_agent_mode", False))
|
||
needs_ma_poll = ma_mode_flag and (has_running_multi_agent or has_pending_ma_messages)
|
||
ma_debug(
|
||
"ma_poll_decision",
|
||
conversation_id=conversation_id,
|
||
multi_agent_mode=ma_mode_flag,
|
||
has_running_multi_agent=has_running_multi_agent,
|
||
has_pending_ma_messages=has_pending_ma_messages,
|
||
needs_ma_poll=needs_ma_poll,
|
||
)
|
||
if needs_ma_poll:
|
||
ma_debug(
|
||
"ma_poll_start",
|
||
conversation_id=conversation_id,
|
||
has_running_multi_agent=has_running_multi_agent,
|
||
has_pending_ma_messages=has_pending_ma_messages,
|
||
)
|
||
def run_ma_poll():
|
||
import asyncio
|
||
import threading as _t
|
||
ma_debug(
|
||
"ma_poll_thread_enter",
|
||
conversation_id=conversation_id,
|
||
thread_id=_t.get_ident(),
|
||
)
|
||
loop = asyncio.new_event_loop()
|
||
asyncio.set_event_loop(loop)
|
||
try:
|
||
loop.run_until_complete(poll_multi_agent_notifications(
|
||
web_terminal=web_terminal,
|
||
workspace=workspace,
|
||
conversation_id=conversation_id,
|
||
client_sid=client_sid,
|
||
username=username
|
||
))
|
||
except Exception as exc:
|
||
ma_debug(
|
||
"ma_poll_thread_exception",
|
||
conversation_id=conversation_id,
|
||
error=str(exc),
|
||
error_type=type(exc).__name__,
|
||
)
|
||
else:
|
||
ma_debug("ma_poll_thread_exit_ok", conversation_id=conversation_id)
|
||
finally:
|
||
loop.close()
|
||
ma_debug("ma_poll_thread_finally", conversation_id=conversation_id)
|
||
|
||
socketio.start_background_task(run_ma_poll)
|
||
|
||
if not needs_completion_poll and not needs_ma_poll:
|
||
ma_debug(
|
||
"completion_poll_skip",
|
||
conversation_id=conversation_id,
|
||
reason="no_running_background_or_multi_agent",
|
||
)
|
||
|
||
has_running_completion_jobs = has_running_sub_agents or has_running_background_commands
|
||
|
||
# 任务结束时清理残留的 pending runtime mode（权限/执行环境/网络权限）。
|
||
# 场景：用户在模型最后一次输出（无工具调用）后、task 仍为 running 状态时切换了
|
||
# 权限/执行环境，此时会进入 queue_*_change 路径只设置 pending，但 task 随即自然
|
||
# 结束，不会进入 execute_tool_calls → apply_pending_runtime_mode_changes。
|
||
# 若不在此清理，pending 会残留到下一次工作首次工具调用时才被应用，
|
||
# 并把"变更通知"作为 user 消息注入到新一轮对话中（bug）。
|
||
# 正确行为：task 结束时立即应用 pending（等同空闲期切换），但不生成通知。
|
||
try:
|
||
if hasattr(web_terminal, "apply_pending_runtime_mode_changes"):
|
||
residual_notices = web_terminal.apply_pending_runtime_mode_changes() or []
|
||
if residual_notices:
|
||
debug_log(
|
||
"[RuntimeMode] task 结束时应用残留 pending runtime mode，丢弃通知: "
|
||
f"{[n.get('source') if isinstance(n, dict) else n for n in residual_notices]}"
|
||
)
|
||
except Exception as exc:
|
||
debug_log(f"[RuntimeMode] task 结束时清理 pending runtime mode 失败: {exc}")
|
||
|
||
pending_runtime_guidance_messages: List[str] = []
|
||
try:
|
||
from .tasks import task_manager
|
||
|
||
raw_items = task_manager.consume_runtime_guidance_for_injection(
|
||
username=username,
|
||
task_id=client_sid,
|
||
)
|
||
# 权限/执行环境/网络权限变更通知在任务完成时不应该触发新一轮工作，
|
||
# 只保留真正的引导消息（压缩续接等），丢弃变更通知。
|
||
runtime_skip_sources = {"权限变更", "执行环境变更", "网络权限变更", "notify"}
|
||
for item in (raw_items or []):
|
||
if isinstance(item, dict):
|
||
src = str(item.get("source") or "").strip().lower()
|
||
text = str(item.get("text") or "").strip()
|
||
if src in runtime_skip_sources:
|
||
continue
|
||
if text:
|
||
pending_runtime_guidance_messages.append(text)
|
||
else:
|
||
text = str(item or "").strip()
|
||
if text:
|
||
pending_runtime_guidance_messages.append(text)
|
||
except Exception as exc:
|
||
debug_log(f"[RuntimeGuidance] 读取剩余引导消息失败: {exc}")
|
||
pending_runtime_guidance_messages = []
|
||
|
||
# 发送完成事件（如果有后台完成任务在运行，前端会保持等待状态）
|
||
if not has_running_completion_jobs:
|
||
# 对话真正空闲：把 work_timer 持久化到后端，避免刷新后恢复成「工作中」
|
||
try:
|
||
if web_terminal and conversation_id:
|
||
finalize_conversation_work_timer(
|
||
web_terminal,
|
||
conversation_id,
|
||
finished_at=datetime.now().isoformat(),
|
||
)
|
||
except Exception:
|
||
pass
|
||
finalize_user_work_timer()
|
||
finalize_run_versioning_checkpoint("completed")
|
||
else:
|
||
finalize_run_versioning_checkpoint("waiting_background")
|
||
sender('task_complete', {
|
||
'total_iterations': total_iterations,
|
||
'total_tool_calls': total_tool_calls,
|
||
'auto_fix_attempts': auto_fix_attempts,
|
||
# 沿用子智能体字段，确保前端直接走已验证通路
|
||
'has_running_sub_agents': has_running_completion_jobs,
|
||
'has_running_background_commands': has_running_background_commands,
|
||
# 多智能体模式：只要还有 running 实例或尚未消费的 pending 消息，
|
||
# 前端就应保持运行态并继续轮询。
|
||
'has_running_multi_agent': has_running_multi_agent or has_pending_ma_messages,
|
||
'pending_runtime_guidance_messages': pending_runtime_guidance_messages,
|
||
})
|