# web_server.py - Web服务器（修复版 - 确保text_end事件正确发送 + 停止功能）

import asyncio
import json
import os
import sys
import re
import threading
from typing import Dict, List, Optional, Callable, Any, Tuple
from flask import Flask, request, jsonify, send_from_directory, session, redirect, send_file, abort
from flask_socketio import SocketIO, emit, join_room, leave_room, disconnect
from flask_cors import CORS
from werkzeug.exceptions import RequestEntityTooLarge
from pathlib import Path
from io import BytesIO
import zipfile
import argparse
from functools import wraps
from datetime import timedelta
import time
from datetime import datetime
from collections import defaultdict, deque, Counter
from config.model_profiles import get_model_profile
from modules import admin_policy_manager, balance_client
from werkzeug.utils import secure_filename
from werkzeug.routing import BaseConverter
import secrets
import logging
import hmac
import mimetypes

# ==========================================
# 回顾文件生成辅助
# ==========================================

def _sanitize_filename_component(text: str) -> str:
    safe = (text or "untitled").strip()
    safe = re.sub(r'[\\/:*?"<>|]+', '_', safe)
    return safe or "untitled"


def build_review_lines(messages, limit=None):
    """
    将对话消息序列拍平成简化文本。
    保留 user / assistant / system 以及 assistant 内的 tool 调用与 tool 消息。
    limit 为正整数时，最多返回该数量的行（用于预览）。
    """
    lines = []

    def append_line(text: str):
        lines.append(text.rstrip())

    def extract_text(content):
        # content 可能是字符串、列表（OpenAI 新结构）或字典
        if isinstance(content, str):
            return content
        if isinstance(content, list):
            parts = []
            for item in content:
                if isinstance(item, dict) and item.get("type") == "text":
                    parts.append(item.get("text") or "")
                elif isinstance(item, str):
                    parts.append(item)
            return "".join(parts)
        if isinstance(content, dict):
            return content.get("text") or ""
        return ""

    def append_tool_call(name, args):
        try:
            args_text = json.dumps(args, ensure_ascii=False)
        except Exception:
            args_text = str(args)
        append_line(f"tool_call：{name} {args_text}")

    for msg in messages or []:
        role = msg.get("role")
        base_content_raw = msg.get("content") if isinstance(msg.get("content"), (str, list, dict)) else msg.get("text") or ""
        base_content = extract_text(base_content_raw)

        if role in ("user", "assistant", "system"):
            append_line(f"{role}：{base_content}")

        if role == "tool":
            append_line(f"tool：{extract_text(base_content_raw)}")

        if role == "assistant":
            # actions 格式
            actions = msg.get("actions") or []
            for action in actions:
                if action.get("type") != "tool":
                    continue
                tool = action.get("tool") or {}
                name = tool.get("name") or "tool"
                args = tool.get("arguments")
                if args is None:
                    args = tool.get("argumentSnapshot")
                try:
                    args_text = json.dumps(args, ensure_ascii=False)
                except Exception:
                    args_text = str(args)
                append_line(f"tool_call：{name} {args_text}")

                tool_content = tool.get("content")
                if tool_content is None:
                    if isinstance(tool.get("result"), str):
                        tool_content = tool.get("result")
                    elif tool.get("result") is not None:
                        try:
                            tool_content = json.dumps(tool.get("result"), ensure_ascii=False)
                        except Exception:
                            tool_content = str(tool.get("result"))
                    elif tool.get("message"):
                        tool_content = tool.get("message")
                    else:
                        tool_content = ""
                append_line(f"tool：{tool_content}")

                if isinstance(limit, int) and limit > 0 and len(lines) >= limit:
                    return lines[:limit]

            # OpenAI 风格 tool_calls
            tool_calls = msg.get("tool_calls") or []
            for tc in tool_calls:
                fn = tc.get("function") or {}
                name = fn.get("name") or "tool"
                args_raw = fn.get("arguments")
                try:
                    args_obj = json.loads(args_raw) if isinstance(args_raw, str) else args_raw
                except Exception:
                    args_obj = args_raw
                append_tool_call(name, args_obj)
                # tool 结果在单独的 tool 消息
                if isinstance(limit, int) and limit > 0 and len(lines) >= limit:
                    return lines[:limit]

            # content 内嵌 tool_call（部分供应商）
            if isinstance(base_content_raw, list):
                for item in base_content_raw:
                    if isinstance(item, dict) and item.get("type") == "tool_call":
                        fn = item.get("function") or {}
                        name = fn.get("name") or "tool"
                        args_raw = fn.get("arguments")
                        try:
                            args_obj = json.loads(args_raw) if isinstance(args_raw, str) else args_raw
                        except Exception:
                            args_obj = args_raw
                        append_tool_call(name, args_obj)
                        if isinstance(limit, int) and limit > 0 and len(lines) >= limit:
                            return lines[:limit]

        if isinstance(limit, int) and limit > 0 and len(lines) >= limit:
            return lines[:limit]

    return lines if limit is None else lines[:limit]

# 控制台输出策略：默认静默，只保留简要事件
_ORIGINAL_PRINT = print
ENABLE_VERBOSE_CONSOLE = False


def brief_log(message: str):
    """始终输出的简要日志（模型输出/工具调用等关键事件）"""
    try:
        _ORIGINAL_PRINT(message)
    except Exception:
        pass


if not ENABLE_VERBOSE_CONSOLE:
    import builtins

    def _silent_print(*args, **kwargs):
        return

    builtins.print = _silent_print

# 抑制 Flask/Werkzeug 访问日志，只保留 brief_log 输出
logging.getLogger('werkzeug').setLevel(logging.ERROR)
logging.getLogger('werkzeug').disabled = True
for noisy_logger in ('engineio.server', 'socketio.server'):
    logging.getLogger(noisy_logger).setLevel(logging.ERROR)
    logging.getLogger(noisy_logger).disabled = True
# 静音子智能体模块错误日志（交由 brief_log 或前端提示处理）
sub_agent_logger = logging.getLogger('modules.sub_agent_manager')
sub_agent_logger.setLevel(logging.CRITICAL)
sub_agent_logger.disabled = True
sub_agent_logger.propagate = False
for h in list(sub_agent_logger.handlers):
    sub_agent_logger.removeHandler(h)

# 添加项目根目录到Python路径
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from core.web_terminal import WebTerminal
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,
    MAX_ACTIVE_USER_CONTAINERS,
    PROJECT_MAX_STORAGE_MB,
    PROJECT_MAX_STORAGE_BYTES,
    UPLOAD_SCAN_LOG_SUBDIR,
)
from modules.user_manager import UserManager, UserWorkspace
from modules.gui_file_manager import GuiFileManager
from modules.upload_security import UploadQuarantineManager, UploadSecurityError
from modules.personalization_manager import (
    load_personalization_config,
    save_personalization_config,
    THINKING_INTERVAL_MIN,
    THINKING_INTERVAL_MAX,
)
from modules.user_container_manager import UserContainerManager
from modules.usage_tracker import UsageTracker, QUOTA_DEFAULTS
from utils.tool_result_formatter import format_tool_result_for_context
from utils.conversation_manager import ConversationManager
from utils.api_client import DeepSeekClient

app = Flask(__name__, static_folder='static')
app.config['MAX_CONTENT_LENGTH'] = MAX_UPLOAD_SIZE
_secret_key = os.environ.get("WEB_SECRET_KEY") or os.environ.get("SECRET_KEY")
if not _secret_key:
    _secret_key = secrets.token_hex(32)
    print("[security] WEB_SECRET_KEY 未设置，已生成临时密钥（重启后失效）。")
app.config['SECRET_KEY'] = _secret_key
app.config['PERMANENT_SESSION_LIFETIME'] = timedelta(hours=12)
_cookie_secure_env = (os.environ.get("WEB_COOKIE_SECURE") or "").strip().lower()
app.config['SESSION_COOKIE_SAMESITE'] = os.environ.get("WEB_COOKIE_SAMESITE", "Strict")
app.config['SESSION_COOKIE_SECURE'] = _cookie_secure_env in {"1", "true", "yes"}
app.config['SESSION_COOKIE_HTTPONLY'] = True
CORS(app)

socketio = SocketIO(app, cors_allowed_origins="*", async_mode='threading', logger=False, engineio_logger=False)


class EndpointFilter(logging.Filter):
    """过滤掉噪声请求日志。"""
    BLOCK_PATTERNS = (
        "GET /api/project-storage",
        "GET /api/container-status",
    )

    def filter(self, record: logging.LogRecord) -> bool:
        message = record.getMessage()
        return not any(pattern in message for pattern in self.BLOCK_PATTERNS)


logging.getLogger('werkzeug').addFilter(EndpointFilter())


class ConversationIdConverter(BaseConverter):
    regex = r'(?:conv_)?\d{8}_\d{6}_\d{3}'


app.url_map.converters['conv'] = ConversationIdConverter

user_manager = UserManager()
container_manager = UserContainerManager()
user_terminals: Dict[str, WebTerminal] = {}
terminal_rooms: Dict[str, set] = {}
connection_users: Dict[str, str] = {}
stop_flags: Dict[str, Dict[str, Any]] = {}

MONITOR_FILE_TOOLS = {'append_to_file', 'modify_file', 'write_file_diff'}
MONITOR_MEMORY_TOOLS = {'update_memory'}
MONITOR_SNAPSHOT_CHAR_LIMIT = 60000
MONITOR_MEMORY_ENTRY_LIMIT = 256
RATE_LIMIT_BUCKETS: Dict[str, deque] = defaultdict(deque)
FAILURE_TRACKERS: Dict[str, Dict[str, float]] = {}
pending_socket_tokens: Dict[str, Dict[str, Any]] = {}
usage_trackers: Dict[str, UsageTracker] = {}

MONITOR_SNAPSHOT_CACHE: Dict[str, Dict[str, Any]] = {}
MONITOR_SNAPSHOT_CACHE_LIMIT = 120

ADMIN_ASSET_DIR = (Path(app.static_folder) / 'admin_dashboard').resolve()
RECENT_UPLOAD_EVENT_LIMIT = 150
RECENT_UPLOAD_FEED_LIMIT = 60

DEFAULT_PORT = 8091
THINKING_FAILURE_KEYWORDS = ["⚠️", "🛑", "失败", "错误", "异常", "终止", "error", "failed", "未完成", "超时", "强制"]
CSRF_HEADER_NAME = "X-CSRF-Token"
CSRF_SESSION_KEY = "_csrf_token"
CSRF_SAFE_METHODS = {"GET", "HEAD", "OPTIONS", "TRACE"}
CSRF_PROTECTED_PATHS = {"/login", "/register", "/logout"}
CSRF_PROTECTED_PREFIXES = ("/api/",)
CSRF_EXEMPT_PATHS = {"/api/csrf-token"}
FAILED_LOGIN_LIMIT = 5
FAILED_LOGIN_LOCK_SECONDS = 300
SOCKET_TOKEN_TTL_SECONDS = 45
PROJECT_STORAGE_CACHE: Dict[str, Dict[str, Any]] = {}
PROJECT_STORAGE_CACHE_TTL_SECONDS = float(os.environ.get("PROJECT_STORAGE_CACHE_TTL", "30"))
USER_IDLE_TIMEOUT_SECONDS = int(os.environ.get("USER_IDLE_TIMEOUT_SECONDS", "900"))
LAST_ACTIVE_FILE = Path(LOGS_DIR).expanduser().resolve() / "last_active.json"
_last_active_lock = threading.Lock()
_last_active_cache: Dict[str, float] = {}
_idle_reaper_started = False
TITLE_PROMPT_PATH = Path(__file__).resolve().parent / "prompts" / "title_generation_prompt.txt"


def sanitize_filename_preserve_unicode(filename: str) -> str:
    """在保留中文等字符的同时，移除危险字符和路径成分"""
    if not filename:
        return ""

    cleaned = filename.strip().replace("\x00", "")
    if not cleaned:
        return ""

    # 去除路径成分
    cleaned = cleaned.replace("\\", "/").split("/")[-1]
    # 替换不安全符号
    cleaned = re.sub(r'[<>:"\\|?*\n\r\t]', "_", cleaned)
    # 去掉前后的点避免隐藏文件/穿越
    cleaned = cleaned.strip(". ")

    if not cleaned:
        return ""

    # Windows/Unix 通用文件名长度安全上限
    return cleaned[:255]


def _load_last_active_cache():
    """从持久化文件加载最近活跃时间，失败时保持空缓存。"""
    try:
        LAST_ACTIVE_FILE.parent.mkdir(parents=True, exist_ok=True)
        if not LAST_ACTIVE_FILE.exists():
            return
        data = json.loads(LAST_ACTIVE_FILE.read_text(encoding="utf-8"))
        if isinstance(data, dict):
            for user, ts in data.items():
                try:
                    _last_active_cache[user] = float(ts)
                except (TypeError, ValueError):
                    continue
    except Exception:
        # 读取失败时忽略，避免影响启动
        pass


def _persist_last_active_cache():
    """原子写入最近活跃时间缓存。"""
    try:
        tmp = LAST_ACTIVE_FILE.with_suffix(".tmp")
        tmp.write_text(json.dumps(_last_active_cache, ensure_ascii=False, indent=2), encoding="utf-8")
        tmp.replace(LAST_ACTIVE_FILE)
    except Exception:
        # 写入失败不影响主流程，记录即可
        debug_log("[IdleReaper] 写入 last_active 文件失败")


def record_user_activity(username: Optional[str], ts: Optional[float] = None):
    """记录用户最近活跃时间，刷新容器 handle 并持久化。"""
    if not username:
        return
    now = ts or time.time()
    with _last_active_lock:
        _last_active_cache[username] = now
        _persist_last_active_cache()
    handle = container_manager.get_handle(username)
    if handle:
        handle.touch()


def get_last_active_ts(username: str, fallback: Optional[float] = None) -> Optional[float]:
    """获取用户最近活跃时间，优先缓存，其次回退值。"""
    with _last_active_lock:
        cached = _last_active_cache.get(username)
    if cached is not None:
        return float(cached)
    return float(fallback) if fallback is not None else None


def idle_reaper_loop():
    """后台轮询：长时间无消息则回收用户容器。"""
    while True:
        try:
            now = time.time()
            handle_map = container_manager.list_containers()
            for username, handle in list(handle_map.items()):
                last_ts = get_last_active_ts(username, handle.get("last_active"))
                if not last_ts:
                    continue
                if now - last_ts >= USER_IDLE_TIMEOUT_SECONDS:
                    debug_log(f"[IdleReaper] 回收容器: {username} (idle {int(now - last_ts)}s)")
                    container_manager.release_container(username, reason="idle_timeout")
            time.sleep(60)
        except Exception as exc:
            debug_log(f"[IdleReaper] 后台循环异常: {exc}")
            time.sleep(60)


def start_background_jobs():
    """启动一次性的后台任务（容器空闲回收）。"""
    global _idle_reaper_started
    if _idle_reaper_started:
        return
    _idle_reaper_started = True
    _load_last_active_cache()
    socketio.start_background_task(idle_reaper_loop)


async def _generate_title_async(user_message: str) -> Optional[str]:
    """使用快速模型生成对话标题。"""
    if not user_message:
        return None
    client = DeepSeekClient(thinking_mode=False, web_mode=True)
    try:
        prompt_text = TITLE_PROMPT_PATH.read_text(encoding="utf-8")
    except Exception:
        prompt_text = "生成一个简洁的、3-5个词的标题，并包含单个emoji，使用用户的语言，直接输出标题。"
    messages = [
        {"role": "system", "content": prompt_text},
        {"role": "user", "content": user_message}
    ]
    try:
        async for resp in client.chat(messages, tools=[], stream=False):
            try:
                content = resp.get("choices", [{}])[0].get("message", {}).get("content")
                if content:
                    return " ".join(str(content).strip().split())
            except Exception:
                continue
    except Exception as exc:
        debug_log(f"[TitleGen] 生成标题异常: {exc}")
    return None


def generate_conversation_title_background(web_terminal: WebTerminal, conversation_id: str, user_message: str, username: str):
    """在后台生成对话标题并更新索引、推送给前端。"""
    if not conversation_id or not user_message:
        return

    async def _runner():
        title = await _generate_title_async(user_message)
        if not title:
            return
        # 限长，避免标题过长
        safe_title = title[:80]
        ok = False
        try:
            ok = web_terminal.context_manager.conversation_manager.update_conversation_title(conversation_id, safe_title)
        except Exception as exc:
            debug_log(f"[TitleGen] 保存标题失败: {exc}")
        if not ok:
            return
        try:
            socketio.emit('conversation_changed', {
                'conversation_id': conversation_id,
                'title': safe_title
            }, room=f"user_{username}")
            socketio.emit('conversation_list_update', {
                'action': 'updated',
                'conversation_id': conversation_id
            }, room=f"user_{username}")
        except Exception as exc:
            debug_log(f"[TitleGen] 推送标题更新失败: {exc}")

    try:
        asyncio.run(_runner())
    except Exception as exc:
        debug_log(f"[TitleGen] 任务执行失败: {exc}")

def cache_monitor_snapshot(execution_id: Optional[str], stage: str, snapshot: Optional[Dict[str, Any]]):
    """缓存工具执行前/后的文件快照。"""
    if not execution_id or not snapshot or not snapshot.get('content'):
        return
    normalized_stage = 'after' if stage == 'after' else 'before'
    entry = MONITOR_SNAPSHOT_CACHE.get(execution_id) or {
        'before': None,
        'after': None,
        'path': snapshot.get('path'),
        'timestamp': 0.0
    }
    entry[normalized_stage] = {
        'path': snapshot.get('path'),
        'content': snapshot.get('content'),
        'lines': snapshot.get('lines') if snapshot.get('lines') is not None else None
    }
    entry['path'] = snapshot.get('path') or entry.get('path')
    entry['timestamp'] = time.time()
    MONITOR_SNAPSHOT_CACHE[execution_id] = entry
    if len(MONITOR_SNAPSHOT_CACHE) > MONITOR_SNAPSHOT_CACHE_LIMIT:
        try:
            oldest_key = min(
                MONITOR_SNAPSHOT_CACHE.keys(),
                key=lambda key: MONITOR_SNAPSHOT_CACHE[key].get('timestamp', 0.0)
            )
            MONITOR_SNAPSHOT_CACHE.pop(oldest_key, None)
        except ValueError:
            pass


def get_cached_monitor_snapshot(execution_id: Optional[str], stage: str) -> Optional[Dict[str, Any]]:
    if not execution_id:
        return None
    entry = MONITOR_SNAPSHOT_CACHE.get(execution_id)
    if not entry:
        return None
    normalized_stage = 'after' if stage == 'after' else 'before'
    snapshot = entry.get(normalized_stage)
    if snapshot and snapshot.get('content'):
        return snapshot
    return None


def get_client_ip() -> str:
    """获取客户端IP，支持 X-Forwarded-For."""
    forwarded = request.headers.get("X-Forwarded-For")
    if forwarded:
        return forwarded.split(",")[0].strip()
    return request.remote_addr or "unknown"


def resolve_identifier(scope: str = "ip", identifier: Optional[str] = None, kwargs: Optional[Dict[str, Any]] = None) -> str:
    if identifier:
        return identifier
    if scope == "user":
        if kwargs:
            username = kwargs.get('username')
            if username:
                return username
        username = get_current_username()
        if username:
            return username
    return get_client_ip()


def check_rate_limit(action: str, limit: int, window_seconds: int, identifier: Optional[str]) -> Tuple[bool, int]:
    """针对指定动作进行简单的滑动窗口限频。"""
    bucket_key = f"{action}:{identifier or 'anonymous'}"
    bucket = RATE_LIMIT_BUCKETS[bucket_key]
    now = time.time()
    while bucket and now - bucket[0] > window_seconds:
        bucket.popleft()
    if len(bucket) >= limit:
        retry_after = window_seconds - int(now - bucket[0])
        return True, max(retry_after, 1)
    bucket.append(now)
    return False, 0


def rate_limited(action: str, limit: int, window_seconds: int, scope: str = "ip", error_message: Optional[str] = None):
    """装饰器：为路由增加速率限制。"""
    def decorator(func):
        @wraps(func)
        def wrapped(*args, **kwargs):
            identifier = resolve_identifier(scope, kwargs=kwargs)
            limited, retry_after = check_rate_limit(action, limit, window_seconds, identifier)
            if limited:
                message = error_message or "请求过于频繁，请稍后再试。"
                return jsonify({
                    "success": False,
                    "error": message,
                    "retry_after": retry_after
                }), 429
            return func(*args, **kwargs)
        return wrapped
    return decorator


def register_failure(action: str, limit: int, lock_seconds: int, scope: str = "ip", identifier: Optional[str] = None, kwargs: Optional[Dict[str, Any]] = None) -> int:
    """记录失败次数，超过阈值后触发锁定。"""
    ident = resolve_identifier(scope, identifier, kwargs)
    key = f"{action}:{ident}"
    now = time.time()
    entry = FAILURE_TRACKERS.setdefault(key, {"count": 0, "blocked_until": 0})
    blocked_until = entry.get("blocked_until", 0)
    if blocked_until and blocked_until > now:
        return int(blocked_until - now)
    entry["count"] = entry.get("count", 0) + 1
    if entry["count"] >= limit:
        entry["count"] = 0
        entry["blocked_until"] = now + lock_seconds
        return lock_seconds
    return 0


def is_action_blocked(action: str, scope: str = "ip", identifier: Optional[str] = None, kwargs: Optional[Dict[str, Any]] = None) -> Tuple[bool, int]:
    ident = resolve_identifier(scope, identifier, kwargs)
    key = f"{action}:{ident}"
    entry = FAILURE_TRACKERS.get(key)
    if not entry:
        return False, 0
    now = time.time()
    blocked_until = entry.get("blocked_until", 0)
    if blocked_until and blocked_until > now:
        return True, int(blocked_until - now)
    return False, 0


def clear_failures(action: str, scope: str = "ip", identifier: Optional[str] = None, kwargs: Optional[Dict[str, Any]] = None):
    ident = resolve_identifier(scope, identifier, kwargs)
    key = f"{action}:{ident}"
    FAILURE_TRACKERS.pop(key, None)


def get_csrf_token(force_new: bool = False) -> str:
    token = session.get(CSRF_SESSION_KEY)
    if force_new or not token:
        token = secrets.token_urlsafe(32)
        session[CSRF_SESSION_KEY] = token
    return token


def requires_csrf_protection(path: str) -> bool:
    if path in CSRF_EXEMPT_PATHS:
        return False
    if path in CSRF_PROTECTED_PATHS:
        return True
    return any(path.startswith(prefix) for prefix in CSRF_PROTECTED_PREFIXES)


def validate_csrf_request() -> bool:
    expected = session.get(CSRF_SESSION_KEY)
    provided = request.headers.get(CSRF_HEADER_NAME) or request.form.get("csrf_token")
    if not expected or not provided:
        return False
    try:
        return hmac.compare_digest(str(provided), str(expected))
    except Exception:
        return False


def prune_socket_tokens(now: Optional[float] = None):
    current = now or time.time()
    for token, meta in list(pending_socket_tokens.items()):
        if meta.get("expires_at", 0) <= current:
            pending_socket_tokens.pop(token, None)


def consume_socket_token(token_value: Optional[str], username: Optional[str]) -> bool:
    if not token_value or not username:
        return False
    prune_socket_tokens()
    token_meta = pending_socket_tokens.pop(token_value, None)
    if not token_meta:
        return False
    if token_meta.get("username") != username:
        return False
    if token_meta.get("expires_at", 0) <= time.time():
        return False
    fingerprint = token_meta.get("fingerprint") or ""
    request_fp = (request.headers.get("User-Agent") or "")[:128]
    if fingerprint and request_fp and not hmac.compare_digest(fingerprint, request_fp):
        return False
    return True


def format_tool_result_notice(tool_name: str, tool_call_id: Optional[str], content: str) -> str:
    """将工具执行结果转为系统消息文本，方便在对话中回传。"""
    header = f"[工具结果] {tool_name}"
    if tool_call_id:
        header += f" (tool_call_id={tool_call_id})"
    body = (content or "").strip()
    if not body:
        body = "（无附加输出）"
    return f"{header}\n{body}"


def compact_web_search_result(result_data: Dict[str, Any]) -> Dict[str, Any]:
    """提取 web_search 结果中前端展示所需的关键字段，避免持久化时丢失列表。"""
    if not isinstance(result_data, dict):
        return {"success": False, "error": "invalid search result"}

    compact: Dict[str, Any] = {
        "success": bool(result_data.get("success")),
        "summary": result_data.get("summary"),
        "query": result_data.get("query"),
        "filters": result_data.get("filters") or {},
        "total_results": result_data.get("total_results", 0)
    }

    # 仅保留前端需要渲染的字段，避免巨大正文导致历史加载时缺失
    items: List[Dict[str, Any]] = []
    for item in result_data.get("results") or []:
        if not isinstance(item, dict):
            continue
        items.append({
            "index": item.get("index"),
            "title": item.get("title") or item.get("name"),
            "url": item.get("url")
        })

    compact["results"] = items

    if not compact.get("success") and result_data.get("error"):
        compact["error"] = result_data.get("error")

    return compact

# 创建调试日志文件
DEBUG_LOG_FILE = Path(LOGS_DIR).expanduser().resolve() / "debug_stream.log"
CHUNK_BACKEND_LOG_FILE = Path(LOGS_DIR).expanduser().resolve() / "chunk_backend.log"
CHUNK_FRONTEND_LOG_FILE = Path(LOGS_DIR).expanduser().resolve() / "chunk_frontend.log"
STREAMING_DEBUG_LOG_FILE = Path(LOGS_DIR).expanduser().resolve() / "streaming_debug.log"
UPLOAD_FOLDER_NAME = "user_upload"


def is_logged_in() -> bool:
    return session.get('username') is not None


def login_required(view_func):
    @wraps(view_func)
    def wrapped(*args, **kwargs):
        if not is_logged_in():
            return redirect('/login')
        return view_func(*args, **kwargs)

    return wrapped


def api_login_required(view_func):
    @wraps(view_func)
    def wrapped(*args, **kwargs):
        if not is_logged_in():
            return jsonify({"error": "Unauthorized"}), 401
        return view_func(*args, **kwargs)

    return wrapped


def get_current_username() -> Optional[str]:
    return session.get('username')


def get_current_user_record():
    username = get_current_username()
    if not username:
        return None
    return user_manager.get_user(username)


def get_current_user_role(record=None) -> str:
    role = session.get('role')
    if role:
        return role
    if record is None:
        record = get_current_user_record()
    return (record.role if record and record.role else 'user')


def is_admin_user(record=None) -> bool:
    role = get_current_user_role(record)
    return isinstance(role, str) and role.lower() == 'admin'

def resolve_admin_policy(record=None) -> Dict[str, Any]:
    """获取当前用户生效的管理员策略。"""
    if record is None:
        record = get_current_user_record()
    username = record.username if record else None
    role = get_current_user_role(record)
    invite_code = getattr(record, "invite_code", None)
    try:
        return admin_policy_manager.get_effective_policy(username, role, invite_code)
    except Exception as exc:
        debug_log(f"[admin_policy] 加载失败: {exc}")
        return admin_policy_manager.get_effective_policy(username, role, invite_code)


def admin_required(view_func):
    @wraps(view_func)
    def wrapped(*args, **kwargs):
        record = get_current_user_record()
        if not record or not is_admin_user(record):
            return redirect('/new')
        return view_func(*args, **kwargs)

    return wrapped


def admin_api_required(view_func):
    @wraps(view_func)
    def wrapped(*args, **kwargs):
        record = get_current_user_record()
        if not record or not is_admin_user(record):
            return jsonify({"success": False, "error": "需要管理员权限"}), 403
        return view_func(*args, **kwargs)

    return wrapped


def make_terminal_callback(username: str):
    """生成面向指定用户的广播函数"""
    def _callback(event_type, data):
        try:
            socketio.emit(event_type, data, room=f"user_{username}")
        except Exception as exc:
            debug_log(f"广播事件失败 ({username}): {event_type} - {exc}")
    return _callback


def attach_user_broadcast(terminal: WebTerminal, username: str):
    """确保终端的广播函数指向当前用户的房间"""
    callback = make_terminal_callback(username)
    terminal.message_callback = callback
    if terminal.terminal_manager:
        terminal.terminal_manager.broadcast = callback


def get_user_resources(username: Optional[str] = None) -> Tuple[Optional[WebTerminal], Optional[UserWorkspace]]:
    username = (username or get_current_username())
    if not username:
        return None, None
    workspace = user_manager.ensure_user_workspace(username)
    container_handle = container_manager.ensure_container(username, str(workspace.project_path))
    usage_tracker = get_or_create_usage_tracker(username, workspace)
    terminal = user_terminals.get(username)
    if not terminal:
        run_mode = session.get('run_mode')
        thinking_mode_flag = session.get('thinking_mode')
        if run_mode not in {"fast", "thinking", "deep"}:
            preferred_run_mode = None
            try:
                personal_config = load_personalization_config(workspace.data_dir)
                candidate_mode = (personal_config or {}).get('default_run_mode')
                if isinstance(candidate_mode, str) and candidate_mode.lower() in {"fast", "thinking", "deep"}:
                    preferred_run_mode = candidate_mode.lower()
            except Exception as exc:
                debug_log(f"[UserInit] 加载个性化偏好失败: {exc}")

            if preferred_run_mode:
                run_mode = preferred_run_mode
                thinking_mode_flag = preferred_run_mode != "fast"
            elif thinking_mode_flag:
                run_mode = "deep"
            else:
                run_mode = "fast"
        thinking_mode = run_mode != "fast"
        terminal = WebTerminal(
            project_path=str(workspace.project_path),
            thinking_mode=thinking_mode,
            run_mode=run_mode,
            message_callback=make_terminal_callback(username),
            data_dir=str(workspace.data_dir),
            container_session=container_handle,
            usage_tracker=usage_tracker
        )
        if terminal.terminal_manager:
            terminal.terminal_manager.broadcast = terminal.message_callback
        user_terminals[username] = terminal
        terminal.username = username
        terminal.quota_update_callback = lambda metric=None: emit_user_quota_update(username)
        session['run_mode'] = terminal.run_mode
        session['thinking_mode'] = terminal.thinking_mode
    else:
        terminal.update_container_session(container_handle)
        attach_user_broadcast(terminal, username)
        terminal.username = username
        terminal.quota_update_callback = lambda metric=None: emit_user_quota_update(username)

    # 应用管理员策略（工具分类、强制开关、模型禁用）
    try:
        from core.tool_config import ToolCategory

        policy = resolve_admin_policy(user_manager.get_user(username))
        categories_map = {
            cid: ToolCategory(
                label=cat.get("label") or cid,
                tools=list(cat.get("tools") or []),
                default_enabled=bool(cat.get("default_enabled", True)),
                silent_when_disabled=bool(cat.get("silent_when_disabled", False)),
            )
            for cid, cat in policy.get("categories", {}).items()
        }
        forced_states = policy.get("forced_category_states") or {}
        disabled_models = policy.get("disabled_models") or []
        terminal.set_admin_policy(categories_map, forced_states, disabled_models)
        terminal.admin_policy_ui_blocks = policy.get("ui_blocks") or {}
        terminal.admin_policy_version = policy.get("updated_at")
        # 若当前模型被禁用，则回退到第一个可用模型
        if terminal.model_key in disabled_models:
            for candidate in ["kimi", "deepseek", "qwen3-vl-plus", "qwen3-max"]:
                if candidate not in disabled_models:
                    try:
                        terminal.set_model(candidate)
                        session["model_key"] = terminal.model_key
                        break
                    except Exception:
                        continue
    except Exception as exc:
        debug_log(f"[admin_policy] 应用失败: {exc}")
    return terminal, workspace


def get_or_create_usage_tracker(username: Optional[str], workspace: Optional[UserWorkspace] = None) -> Optional[UsageTracker]:
    if not username:
        return None
    tracker = usage_trackers.get(username)
    if tracker:
        return tracker
    if workspace is None:
        workspace = user_manager.ensure_user_workspace(username)
    record = user_manager.get_user(username)
    role = getattr(record, "role", "user") if record else "user"
    tracker = UsageTracker(str(workspace.data_dir), role=role or "user")
    usage_trackers[username] = tracker
    return tracker


def emit_user_quota_update(username: Optional[str]):
    if not username:
        return
    tracker = get_or_create_usage_tracker(username)
    if not tracker:
        return
    try:
        snapshot = tracker.get_quota_snapshot()
        socketio.emit('quota_update', {
            'quotas': snapshot
        }, room=f"user_{username}")
    except Exception:
        pass


def with_terminal(func):
    """注入用户专属终端和工作区"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        username = get_current_username()
        try:
            terminal, workspace = get_user_resources(username)
        except RuntimeError as exc:
            return jsonify({"error": str(exc), "code": "resource_busy"}), 503
        if not terminal or not workspace:
            return jsonify({"error": "System not initialized"}), 503
        kwargs.update({
            'terminal': terminal,
            'workspace': workspace,
            'username': username
        })
        return func(*args, **kwargs)
    return wrapper


def get_terminal_for_sid(sid: str) -> Tuple[Optional[str], Optional[WebTerminal], Optional[UserWorkspace]]:
    username = connection_users.get(sid)
    if not username:
        return None, None, None
    try:
        terminal, workspace = get_user_resources(username)
    except RuntimeError:
        return username, None, None
    return username, terminal, workspace


def get_gui_manager(workspace: UserWorkspace) -> GuiFileManager:
    """构建 GUI 文件管理器"""
    return GuiFileManager(str(workspace.project_path))


def get_upload_guard(workspace: UserWorkspace) -> UploadQuarantineManager:
    """构建上传隔离管理器"""
    return UploadQuarantineManager(workspace)


def build_upload_error_response(exc: UploadSecurityError):
    status = 400
    if exc.code in {"scanner_missing", "scanner_unavailable"}:
        status = 500
    return jsonify({
        "success": False,
        "error": str(exc),
        "code": exc.code,
    }), status


def ensure_conversation_loaded(terminal: WebTerminal, conversation_id: Optional[str]) -> Tuple[str, bool]:
    """确保终端加载指定对话，若无则创建新的"""
    created_new = False
    if not conversation_id:
        # 不显式传入运行模式，优先回到个性化/默认配置
        result = terminal.create_new_conversation()
        if not result.get("success"):
            raise RuntimeError(result.get("message", "创建对话失败"))
        conversation_id = result["conversation_id"]
        session['run_mode'] = terminal.run_mode
        session['thinking_mode'] = terminal.thinking_mode
        created_new = True
    else:
        conversation_id = conversation_id if conversation_id.startswith('conv_') else f"conv_{conversation_id}"
        current_id = terminal.context_manager.current_conversation_id
        if current_id != conversation_id:
            load_result = terminal.load_conversation(conversation_id)
            if not load_result.get("success"):
                raise RuntimeError(load_result.get("message", "对话加载失败"))
            # 切换到对话记录的运行模式
            try:
                conv_data = terminal.context_manager.conversation_manager.load_conversation(conversation_id) or {}
                meta = conv_data.get("metadata", {}) or {}
                run_mode_meta = meta.get("run_mode")
                if run_mode_meta:
                    terminal.set_run_mode(run_mode_meta)
                elif meta.get("thinking_mode"):
                    terminal.set_run_mode("thinking")
                else:
                    terminal.set_run_mode("fast")
                if terminal.thinking_mode:
                    terminal.api_client.start_new_task(force_deep=terminal.deep_thinking_mode)
                else:
                    terminal.api_client.start_new_task()
                session['run_mode'] = terminal.run_mode
                session['thinking_mode'] = terminal.thinking_mode
            except Exception:
                pass
    return conversation_id, created_new

def reset_system_state(terminal: Optional[WebTerminal]):
    """完整重置系统状态，确保停止后能正常开始新任务"""
    if not terminal:
        return
    
    try:
        # 1. 重置API客户端状态
        if hasattr(terminal, 'api_client') and terminal.api_client:
            debug_log("重置API客户端状态")
            terminal.api_client.start_new_task(force_deep=getattr(terminal, "deep_thinking_mode", False))
        
        # 2. 重置主终端会话状态
        if hasattr(terminal, 'current_session_id'):
            terminal.current_session_id += 1  # 开始新会话
            debug_log(f"重置会话ID为: {terminal.current_session_id}")
        
        # 3. 清理读取文件跟踪器
            debug_log("清理文件读取跟踪器")
        
        # 4. 重置Web特有的状态属性
        web_attrs = ['streamingMessage', 'currentMessageIndex', 'preparingTools', 'activeTools']
        for attr in web_attrs:
            if hasattr(terminal, attr):
                if attr in ['streamingMessage']:
                    setattr(terminal, attr, False)
                elif attr in ['currentMessageIndex']:
                    setattr(terminal, attr, -1)
                elif attr in ['preparingTools', 'activeTools'] and hasattr(getattr(terminal, attr), 'clear'):
                    getattr(terminal, attr).clear()
        
        debug_log("系统状态重置完成")
        
    except Exception as e:
        debug_log(f"状态重置过程中出现错误: {e}")
        import traceback
        debug_log(f"错误详情: {traceback.format_exc()}")


def _write_log(file_path: Path, message: str) -> None:
    file_path.parent.mkdir(parents=True, exist_ok=True)
    with file_path.open('a', encoding='utf-8') as f:
        timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]
        f.write(f"[{timestamp}] {message}\n")


def debug_log(message):
    """写入调试日志"""
    _write_log(DEBUG_LOG_FILE, message)


def log_backend_chunk(conversation_id: str, iteration: int, chunk_index: int, elapsed: float, char_len: int, content_preview: str):
    preview = content_preview.replace('\n', '\\n')
    _write_log(
        CHUNK_BACKEND_LOG_FILE,
        f"conv={conversation_id or 'unknown'} iter={iteration} chunk={chunk_index} elapsed={elapsed:.3f}s len={char_len} preview={preview}"
    )


def log_frontend_chunk(conversation_id: str, chunk_index: int, elapsed: float, char_len: int, client_ts: float):
    _write_log(
        CHUNK_FRONTEND_LOG_FILE,
        f"conv={conversation_id or 'unknown'} chunk={chunk_index} elapsed={elapsed:.3f}s len={char_len} client_ts={client_ts}"
    )


def log_streaming_debug_entry(data: Dict[str, Any]):
    try:
        serialized = json.dumps(data, ensure_ascii=False)
    except Exception:
        serialized = str(data)
    _write_log(STREAMING_DEBUG_LOG_FILE, serialized)


def get_thinking_state(terminal: WebTerminal) -> Dict[str, Any]:
    """获取（或初始化）思考调度状态。"""
    state = getattr(terminal, "_thinking_state", None)
    if not state:
        state = {"fast_streak": 0, "force_next": False, "suppress_next": False}
        terminal._thinking_state = state
    return state


def mark_force_thinking(terminal: WebTerminal, reason: str = ""):
    """标记下一次API调用必须使用思考模型。"""
    if getattr(terminal, "deep_thinking_mode", False):
        return
    if not getattr(terminal, "thinking_mode", False):
        return
    state = get_thinking_state(terminal)
    state["force_next"] = True
    if reason:
        debug_log(f"[Thinking] 下次强制思考，原因: {reason}")


def mark_suppress_thinking(terminal: WebTerminal):
    """标记下一次API调用必须跳过思考模型（例如写入窗口）。"""
    if getattr(terminal, "deep_thinking_mode", False):
        return
    if not getattr(terminal, "thinking_mode", False):
        return
    state = get_thinking_state(terminal)
    state["suppress_next"] = True


def apply_thinking_schedule(terminal: WebTerminal):
    """根据当前状态配置API客户端的思考/快速模式。"""
    client = terminal.api_client
    if getattr(terminal, "deep_thinking_mode", False):
        client.force_thinking_next_call = False
        client.skip_thinking_next_call = False
        return
    if not getattr(terminal, "thinking_mode", False):
        client.force_thinking_next_call = False
        client.skip_thinking_next_call = False
        return
    state = get_thinking_state(terminal)
    awaiting_writes = getattr(terminal, "pending_append_request", None) or getattr(terminal, "pending_modify_request", None)
    if awaiting_writes:
        client.skip_thinking_next_call = True
        state["suppress_next"] = False
        debug_log("[Thinking] 检测到写入窗口请求，跳过思考。")
        return
    if state.get("suppress_next"):
        client.skip_thinking_next_call = True
        state["suppress_next"] = False
        debug_log("[Thinking] 由于写入窗口，下一次跳过思考。")
        return
    if state.get("force_next"):
        client.force_thinking_next_call = True
        state["force_next"] = False
        state["fast_streak"] = 0
        debug_log("[Thinking] 响应失败，下一次强制思考。")
        return
    custom_interval = getattr(terminal, "thinking_fast_interval", THINKING_FAST_INTERVAL)
    interval = max(0, custom_interval or 0)
    if interval > 0:
        allowed_fast = max(0, interval - 1)
        if state.get("fast_streak", 0) >= allowed_fast:
            client.force_thinking_next_call = True
            state["fast_streak"] = 0
            if allowed_fast == 0:
                debug_log("[Thinking] 频率=1，持续思考。")
            else:
                debug_log(f"[Thinking] 快速模式已连续 {allowed_fast} 次，下一次强制思考。")
            return
    client.force_thinking_next_call = False
    client.skip_thinking_next_call = False


def update_thinking_after_call(terminal: WebTerminal):
    """一次API调用完成后更新快速计数。"""
    if getattr(terminal, "deep_thinking_mode", False):
        state = get_thinking_state(terminal)
        state["fast_streak"] = 0
        return
    if not getattr(terminal, "thinking_mode", False):
        return
    state = get_thinking_state(terminal)
    if terminal.api_client.last_call_used_thinking:
        state["fast_streak"] = 0
    else:
        state["fast_streak"] = state.get("fast_streak", 0) + 1
        debug_log(f"[Thinking] 快速模式计数: {state['fast_streak']}")


def maybe_mark_failure_from_message(terminal: WebTerminal, content: Optional[str]):
    """根据system消息内容判断是否需要强制思考。"""
    if not content:
        return
    normalized = content.lower()
    if any(keyword.lower() in normalized for keyword in THINKING_FAILURE_KEYWORDS):
        mark_force_thinking(terminal, reason="system_message")


def detect_tool_failure(result_data: Any) -> bool:
    """识别工具返回结果是否代表失败。"""
    if not isinstance(result_data, dict):
        return False
    if result_data.get("success") is False:
        return True
    status = str(result_data.get("status", "")).lower()
    if status in {"failed", "error"}:
        return True
    error_msg = result_data.get("error")
    if isinstance(error_msg, str) and error_msg.strip():
        return True
    return False

# 终端广播回调函数
def terminal_broadcast(event_type, data):
    """广播终端事件到所有订阅者"""
    try:
        # 对于全局事件，发送给所有连接的客户端
        if event_type in ('token_update', 'todo_updated'):
            socketio.emit(event_type, data)  # 全局广播，不限制房间
            debug_log(f"全局广播{event_type}: {data}")
        else:
            # 其他终端事件发送到终端订阅者房间
            socketio.emit(event_type, data, room='terminal_subscribers')
            
            # 如果是特定会话的事件，也发送到该会话的专属房间
            if 'session' in data:
                session_room = f"terminal_{data['session']}"
                socketio.emit(event_type, data, room=session_room)
        
        debug_log(f"终端广播: {event_type} - {data}")
    except Exception as e:
        debug_log(f"终端广播错误: {e}")


@app.route('/api/csrf-token', methods=['GET'])
def issue_csrf_token():
    """提供 CSRF token，供前端写操作附带。"""
    token = get_csrf_token()
    response = jsonify({"success": True, "token": token})
    response.headers['Cache-Control'] = 'no-store'
    return response


@app.before_request
def enforce_csrf_token():
    method = (request.method or "GET").upper()
    if method in CSRF_SAFE_METHODS:
        return
    if not requires_csrf_protection(request.path):
        return
    if validate_csrf_request():
        return
    return jsonify({"success": False, "error": "CSRF validation failed"}), 403


@app.after_request
def apply_security_headers(response):
    response.headers.setdefault("X-Frame-Options", "SAMEORIGIN")
    response.headers.setdefault("X-Content-Type-Options", "nosniff")
    response.headers.setdefault("Referrer-Policy", "strict-origin-when-cross-origin")
    if response.mimetype == "application/json":
        response.headers.setdefault("Cache-Control", "no-store")
    if app.config.get("SESSION_COOKIE_SECURE"):
        response.headers.setdefault("Strict-Transport-Security", "max-age=31536000; includeSubDomains")
    return response

@app.route('/login', methods=['GET', 'POST'])
def login():
    """登录页面与认证"""
    if request.method == 'GET':
        if is_logged_in():
            return redirect('/new')
        if not container_manager.has_capacity():
            return app.send_static_file('resource_busy.html'), 503
        return app.send_static_file('login.html')

    data = request.get_json() or {}
    email = (data.get('email') or '').strip()
    password = data.get('password') or ''
    client_ip = get_client_ip()
    limited, retry_after = check_rate_limit("login", 10, 60, client_ip)
    if limited:
        return jsonify({
            "success": False,
            "error": "登录请求过于频繁，请稍后再试。",
            "retry_after": retry_after
        }), 429

    blocked, block_for = is_action_blocked("login", identifier=client_ip)
    if blocked:
        return jsonify({
            "success": False,
            "error": f"尝试次数过多，请 {block_for} 秒后重试。",
            "retry_after": block_for
        }), 429

    record = user_manager.authenticate(email, password)
    if not record:
        wait_seconds = register_failure("login", FAILED_LOGIN_LIMIT, FAILED_LOGIN_LOCK_SECONDS, identifier=client_ip)
        error_payload = {
            "success": False,
            "error": "账号或密码错误"
        }
        status_code = 401
        if wait_seconds:
            error_payload.update({
                "error": f"尝试次数过多，请 {wait_seconds} 秒后重试。",
                "retry_after": wait_seconds
            })
            status_code = 429
        return jsonify(error_payload), status_code

    workspace = user_manager.ensure_user_workspace(record.username)
    preferred_run_mode = None
    try:
        personal_config = load_personalization_config(workspace.data_dir)
        candidate_mode = (personal_config or {}).get('default_run_mode')
        if isinstance(candidate_mode, str):
            normalized_mode = candidate_mode.lower()
            if normalized_mode in {"fast", "thinking", "deep"}:
                preferred_run_mode = normalized_mode
    except Exception as exc:
        debug_log(f"加载个性化偏好失败: {exc}")

    session['logged_in'] = True
    session['username'] = record.username
    session['role'] = record.role or 'user'
    default_thinking = app.config.get('DEFAULT_THINKING_MODE', False)
    session['thinking_mode'] = default_thinking
    session['run_mode'] = app.config.get('DEFAULT_RUN_MODE', "deep" if default_thinking else "fast")
    if preferred_run_mode:
        session['run_mode'] = preferred_run_mode
        session['thinking_mode'] = preferred_run_mode != 'fast'
    session.permanent = True
    clear_failures("login", identifier=client_ip)
    try:
        container_manager.ensure_container(record.username, str(workspace.project_path))
    except RuntimeError as exc:
        session.clear()
        return jsonify({"success": False, "error": str(exc), "code": "resource_busy"}), 503
    record_user_activity(record.username)
    get_csrf_token(force_new=True)
    return jsonify({"success": True})


@app.route('/register', methods=['GET', 'POST'])
def register():
    """注册新用户（需要邀请码）"""
    if request.method == 'GET':
        if is_logged_in():
            return redirect('/new')
        return app.send_static_file('register.html')
    
    data = request.get_json() or {}
    username = (data.get('username') or '').strip()
    email = (data.get('email') or '').strip()
    password = data.get('password') or ''
    invite_code = (data.get('invite_code') or '').strip()
    
    limited, retry_after = check_rate_limit("register", 5, 300, get_client_ip())
    if limited:
        return jsonify({
            "success": False,
            "error": "注册请求过于频繁，请稍后再试。",
            "retry_after": retry_after
        }), 429
    
    try:
        user_manager.register_user(username, email, password, invite_code)
        return jsonify({"success": True})
    except ValueError as exc:
        return jsonify({"success": False, "error": str(exc)}), 400
    except Exception as exc:
        return jsonify({"success": False, "error": str(exc)}), 500


@app.route('/logout', methods=['POST'])
def logout():
    """退出登录"""
    username = session.get('username')
    session.clear()
    if username and username in user_terminals:
        user_terminals.pop(username, None)
    if username:
        container_manager.release_container(username, reason="logout")
        for token_value, meta in list(pending_socket_tokens.items()):
            if meta.get("username") == username:
                pending_socket_tokens.pop(token_value, None)
    return jsonify({"success": True})


@app.route('/')
@login_required
def index():
    """主页 -> 重定向到 /new"""
    return redirect('/new')


@app.route('/new')
@login_required
def new_page():
    return app.send_static_file('index.html')


@app.route('/<conv:conversation_id>')
@login_required
def conversation_page(conversation_id):
    return app.send_static_file('index.html')


@app.route('/terminal')
@login_required
def terminal_page():
    """终端监控页面"""
    policy = resolve_admin_policy(get_current_user_record())
    if policy.get("ui_blocks", {}).get("block_realtime_terminal"):
        return "实时终端已被管理员禁用", 403
    return app.send_static_file('terminal.html')


@app.route('/file-manager')
@login_required
def gui_file_manager_page():
    """桌面式文件管理器页面"""
    policy = resolve_admin_policy(get_current_user_record())
    if policy.get("ui_blocks", {}).get("block_file_manager"):
        return "文件管理器已被管理员禁用", 403
    return send_from_directory(Path(app.static_folder) / 'file_manager', 'index.html')


@app.route('/file-manager/editor')
@login_required
def gui_file_editor_page():
    """GUI 文件编辑器页面"""
    return send_from_directory(Path(app.static_folder) / 'file_manager', 'editor.html')


@app.route('/file-preview/<path:relative_path>')
@login_required
@with_terminal
def gui_file_preview(relative_path: str, terminal: WebTerminal, workspace: UserWorkspace, username: str):
    manager = get_gui_manager(workspace)
    try:
        target = manager.prepare_download(relative_path)
        if not target.is_file():
            return "预览仅支持文件", 400
        return send_from_directory(
            directory=target.parent,
            path=target.name,
            mimetype='text/html'
        )
    except Exception as exc:
        return f"无法预览文件: {exc}", 400


@app.route('/admin/monitor')
@login_required
@admin_required
def admin_monitor_page():
    """管理员监控页面入口"""
    return send_from_directory(str(ADMIN_ASSET_DIR), 'index.html')

@app.route('/admin/policy')
@login_required
@admin_required
def admin_policy_page():
    """管理员策略配置页面"""
    return send_from_directory(Path(app.static_folder) / 'admin_policy', 'index.html')


@app.route('/api/admin/balance', methods=['GET'])
@login_required
@admin_required
def admin_balance_api():
    """查询第三方账户余额（Kimi/DeepSeek/Qwen）。"""
    data = balance_client.fetch_all_balances()
    return jsonify({"success": True, "data": data})


@app.route('/admin/assets/<path:filename>')
@login_required
@admin_required
def admin_asset_file(filename: str):
    return send_from_directory(str(ADMIN_ASSET_DIR), filename)


@app.route('/user_upload/<path:filename>')
@login_required
def serve_user_upload(filename: str):
    """
    直接向前端暴露当前登录用户的上传目录文件，用于 <show_image src="/user_upload/..."> 等场景。
    - 仅登录用户可访问
    - 路径穿越校验：目标必须位于用户自己的 uploads_dir 内
    """
    user = get_current_user_record()
    if not user:
        return redirect('/login')

    workspace = user_manager.ensure_user_workspace(user.username)
    uploads_dir = workspace.uploads_dir.resolve()

    target = (uploads_dir / filename).resolve()
    try:
        target.relative_to(uploads_dir)
    except ValueError:
        abort(403)

    if not target.exists() or not target.is_file():
        abort(404)

    return send_from_directory(str(uploads_dir), str(target.relative_to(uploads_dir)))


@app.route('/workspace/<path:filename>')
@login_required
def serve_workspace_file(filename: str):
    """
    暴露当前登录用户项目目录下的文件（主要用于图片展示）。
    - 仅登录用户可访问自己的项目文件
    - 路径穿越校验：目标必须位于用户自己的 project_path 内
    - 非图片直接拒绝，避免误暴露其他文件
    """
    user = get_current_user_record()
    if not user:
        return redirect('/login')

    workspace = user_manager.ensure_user_workspace(user.username)
    project_root = workspace.project_path.resolve()

    target = (project_root / filename).resolve()
    try:
        target.relative_to(project_root)
    except ValueError:
        abort(403)

    if not target.exists() or not target.is_file():
        abort(404)

    mime_type, _ = mimetypes.guess_type(str(target))
    if not mime_type or not mime_type.startswith("image/"):
        abort(415)

    return send_from_directory(str(target.parent), target.name)


@app.route('/static/<path:filename>')
def static_files(filename):
    """提供静态文件"""
    if filename.startswith('admin_dashboard'):
        abort(404)
    return send_from_directory('static', filename)

@app.route('/api/status')
@api_login_required
@with_terminal
def get_status(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    """获取系统状态（增强版：包含对话信息）"""
    status = terminal.get_status()
    
    # 添加终端状态信息
    if terminal.terminal_manager:
        terminal_status = terminal.terminal_manager.list_terminals()
        status['terminals'] = terminal_status
    
    # 【新增】添加当前对话的详细信息
    try:
        current_conv = terminal.context_manager.current_conversation_id
        status['conversation'] = status.get('conversation', {})
        status['conversation']['current_id'] = current_conv
        if current_conv and not current_conv.startswith('temp_'):
            current_conv_data = terminal.context_manager.conversation_manager.load_conversation(current_conv)
            if current_conv_data:
                status['conversation']['title'] = current_conv_data.get('title', '未知对话')
                status['conversation']['created_at'] = current_conv_data.get('created_at')
                status['conversation']['updated_at'] = current_conv_data.get('updated_at')
    except Exception as e:
        print(f"[Status] 获取当前对话信息失败: {e}")
    
    status['project_path'] = str(workspace.project_path)
    try:
        status['container'] = container_manager.get_container_status(username)
    except Exception as exc:
        status['container'] = {"success": False, "error": str(exc)}
    status['version'] = AGENT_VERSION
    try:
        policy = resolve_admin_policy(user_manager.get_user(username))
        status['admin_policy'] = {
            "ui_blocks": policy.get("ui_blocks") or {},
            "disabled_models": policy.get("disabled_models") or [],
            "forced_category_states": policy.get("forced_category_states") or {},
            "version": policy.get("updated_at"),
        }
    except Exception as exc:
        debug_log(f"[status] 附加管理员策略失败: {exc}")
    return jsonify(status)

@app.route('/api/container-status')
@api_login_required
@with_terminal
def get_container_status_api(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    """轮询容器状态（供前端用量面板定时刷新）。"""
    try:
        status = container_manager.get_container_status(username)
        return jsonify({"success": True, "data": status})
    except Exception as exc:
        return jsonify({"success": False, "error": str(exc)}), 500

@app.route('/api/project-storage')
@api_login_required
@with_terminal
def get_project_storage(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    """获取项目目录占用情况，供前端轮询。"""
    now = time.time()
    cache_entry = PROJECT_STORAGE_CACHE.get(username)
    if cache_entry and (now - cache_entry.get("ts", 0)) < PROJECT_STORAGE_CACHE_TTL_SECONDS:
        return jsonify({"success": True, "data": cache_entry["data"]})
    try:
        file_manager = getattr(terminal, 'file_manager', None)
        if not file_manager:
            return jsonify({"success": False, "error": "文件管理器未初始化"}), 500
        used_bytes = file_manager._get_project_size()
        limit_bytes = PROJECT_MAX_STORAGE_MB * 1024 * 1024 if PROJECT_MAX_STORAGE_MB else None
        usage_percent = (used_bytes / limit_bytes * 100) if limit_bytes else None
        data = {
            "used_bytes": used_bytes,
            "limit_bytes": limit_bytes,
            "limit_label": f"{PROJECT_MAX_STORAGE_MB}MB" if PROJECT_MAX_STORAGE_MB else "未限制",
            "usage_percent": usage_percent
        }
        PROJECT_STORAGE_CACHE[username] = {"ts": now, "data": data}
        return jsonify({"success": True, "data": data})
    except Exception as exc:
        stale = PROJECT_STORAGE_CACHE.get(username)
        if stale:
            return jsonify({"success": True, "data": stale.get("data"), "stale": True}), 200
        return jsonify({"success": False, "error": str(exc)}), 500


@app.route('/api/admin/dashboard')
@api_login_required
@admin_api_required
def admin_dashboard_snapshot_api():
    try:
        snapshot = build_admin_dashboard_snapshot()
        return jsonify({"success": True, "data": snapshot})
    except Exception as exc:
        logging.exception("Failed to build admin dashboard")
        return jsonify({"success": False, "error": str(exc)}), 500

@app.route('/api/admin/policy', methods=['GET', 'POST'])
@api_login_required
@admin_api_required
def admin_policy_api():
    if request.method == 'GET':
        try:
            data = admin_policy_manager.load_policy()
            defaults = admin_policy_manager.describe_defaults()
            return jsonify({"success": True, "data": data, "defaults": defaults})
        except Exception as exc:
            return jsonify({"success": False, "error": str(exc)}), 500
    # POST 更新
    payload = request.get_json() or {}
    target_type = payload.get("target_type")
    target_value = payload.get("target_value") or ""
    config = payload.get("config") or {}
    try:
        saved = admin_policy_manager.save_scope_policy(target_type, target_value, config)
        return jsonify({"success": True, "data": saved})
    except ValueError as exc:
        return jsonify({"success": False, "error": str(exc)}), 400
    except Exception as exc:
        return jsonify({"success": False, "error": str(exc)}), 500

@app.route('/api/effective-policy', methods=['GET'])
@api_login_required
def effective_policy_api():
    record = get_current_user_record()
    policy = resolve_admin_policy(record)
    return jsonify({"success": True, "data": policy})


@app.route('/api/usage', methods=['GET'])
@api_login_required
def get_usage_stats():
    """返回当前用户的模型/搜索调用统计。"""
    username = get_current_username()
    tracker = get_or_create_usage_tracker(username)
    if not tracker:
        return jsonify({"success": False, "error": "未找到用户"}), 404
    return jsonify({
        "success": True,
        "data": tracker.get_stats()
    })

@app.route('/api/thinking-mode', methods=['POST'])
@api_login_required
@with_terminal
@rate_limited("thinking_mode_toggle", 15, 60, scope="user")
def update_thinking_mode(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    """切换思考模式"""
    try:
        data = request.get_json() or {}
        requested_mode = data.get('mode')
        if requested_mode in {"fast", "thinking", "deep"}:
            target_mode = requested_mode
        elif 'thinking_mode' in data:
            target_mode = "thinking" if bool(data.get('thinking_mode')) else "fast"
        else:
            target_mode = terminal.run_mode
        terminal.set_run_mode(target_mode)
        if terminal.thinking_mode:
            terminal.api_client.start_new_task(force_deep=terminal.deep_thinking_mode)
        else:
            terminal.api_client.start_new_task()
        session['thinking_mode'] = terminal.thinking_mode
        session['run_mode'] = terminal.run_mode
        # 更新当前对话的元数据
        ctx = terminal.context_manager
        if ctx.current_conversation_id:
            try:
                ctx.conversation_manager.save_conversation(
                    conversation_id=ctx.current_conversation_id,
                    messages=ctx.conversation_history,
                    project_path=str(ctx.project_path),
                    todo_list=ctx.todo_list,
                    thinking_mode=terminal.thinking_mode,
                    run_mode=terminal.run_mode,
                    model_key=getattr(terminal, "model_key", None)
                )
            except Exception as exc:
                print(f"[API] 保存思考模式到对话失败: {exc}")
        
        status = terminal.get_status()
        socketio.emit('status_update', status, room=f"user_{username}")
        
        return jsonify({
            "success": True,
            "data": {
                "thinking_mode": terminal.thinking_mode,
                "mode": terminal.run_mode
            }
        })
    except Exception as exc:
        print(f"[API] 切换思考模式失败: {exc}")
        code = 400 if isinstance(exc, ValueError) else 500
        return jsonify({
            "success": False,
            "error": str(exc),
            "message": "切换思考模式时发生异常"
        }), code


@app.route('/api/model', methods=['POST'])
@api_login_required
@with_terminal
@rate_limited("model_switch", 10, 60, scope="user")
def update_model(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    """切换基础模型（快速/思考模型组合）。"""
    try:
        data = request.get_json() or {}
        model_key = data.get("model_key")
        if not model_key:
            return jsonify({"success": False, "error": "缺少 model_key"}), 400

        # 管理员禁用模型校验
        policy = resolve_admin_policy(get_current_user_record())
        disabled_models = set(policy.get("disabled_models") or [])
        if model_key in disabled_models:
            return jsonify({
                "success": False,
                "error": "该模型已被管理员禁用",
                "message": "被管理员强制禁用"
            }), 403

        terminal.set_model(model_key)
        # fast-only 时 run_mode 可能被强制为 fast
        session["model_key"] = terminal.model_key
        session["run_mode"] = terminal.run_mode
        session["thinking_mode"] = terminal.thinking_mode

        # 更新当前对话元数据
        ctx = terminal.context_manager
        if ctx.current_conversation_id:
            try:
                ctx.conversation_manager.save_conversation(
                    conversation_id=ctx.current_conversation_id,
                    messages=ctx.conversation_history,
                    project_path=str(ctx.project_path),
                    todo_list=ctx.todo_list,
                    thinking_mode=terminal.thinking_mode,
                    run_mode=terminal.run_mode,
                    model_key=terminal.model_key,
                    has_images=getattr(ctx, "has_images", False)
                )
            except Exception as exc:
                print(f"[API] 保存模型到对话失败: {exc}")

        status = terminal.get_status()
        socketio.emit('status_update', status, room=f"user_{username}")

        return jsonify({
            "success": True,
            "data": {
                "model_key": terminal.model_key,
                "run_mode": terminal.run_mode,
                "thinking_mode": terminal.thinking_mode
            }
        })
    except Exception as exc:
        print(f"[API] 切换模型失败: {exc}")
        code = 400 if isinstance(exc, ValueError) else 500
        return jsonify({"success": False, "error": str(exc), "message": str(exc)}), code


@app.route('/api/personalization', methods=['GET'])
@api_login_required
@with_terminal
def get_personalization_settings(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    """获取个性化配置"""
    try:
        policy = resolve_admin_policy(get_current_user_record())
        if policy.get("ui_blocks", {}).get("block_personal_space"):
            return jsonify({"success": False, "error": "个人空间已被管理员禁用"}), 403
        data = load_personalization_config(workspace.data_dir)
        return jsonify({
            "success": True,
            "data": data,
            "tool_categories": terminal.get_tool_settings_snapshot(),
            "thinking_interval_default": THINKING_FAST_INTERVAL,
            "thinking_interval_range": {
                "min": THINKING_INTERVAL_MIN,
                "max": THINKING_INTERVAL_MAX
            }
        })
    except Exception as exc:
        return jsonify({"success": False, "error": str(exc)}), 500


@app.route('/api/personalization', methods=['POST'])
@api_login_required
@with_terminal
@rate_limited("personalization_update", 20, 300, scope="user")
def update_personalization_settings(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    """更新个性化配置"""
    payload = request.get_json() or {}
    try:
        policy = resolve_admin_policy(get_current_user_record())
        if policy.get("ui_blocks", {}).get("block_personal_space"):
            return jsonify({"success": False, "error": "个人空间已被管理员禁用"}), 403
        config = save_personalization_config(workspace.data_dir, payload)
        try:
            terminal.apply_personalization_preferences(config)
            session['run_mode'] = terminal.run_mode
            session['thinking_mode'] = terminal.thinking_mode
            ctx = getattr(terminal, 'context_manager', None)
            if ctx and getattr(ctx, 'current_conversation_id', None):
                try:
                    ctx.conversation_manager.save_conversation(
                        conversation_id=ctx.current_conversation_id,
                        messages=ctx.conversation_history,
                        project_path=str(ctx.project_path),
                        todo_list=ctx.todo_list,
                        thinking_mode=terminal.thinking_mode,
                        run_mode=terminal.run_mode
                    )
                except Exception as meta_exc:
                    debug_log(f"应用个性化偏好失败: 同步对话元数据异常 {meta_exc}")
            try:
                status = terminal.get_status()
                socketio.emit('status_update', status, room=f"user_{username}")
            except Exception as status_exc:
                debug_log(f"广播个性化状态失败: {status_exc}")
        except Exception as exc:
            debug_log(f"应用个性化偏好失败: {exc}")
        return jsonify({
            "success": True,
            "data": config,
            "tool_categories": terminal.get_tool_settings_snapshot(),
            "thinking_interval_default": THINKING_FAST_INTERVAL,
            "thinking_interval_range": {
                "min": THINKING_INTERVAL_MIN,
                "max": THINKING_INTERVAL_MAX
            }
        })
    except ValueError as exc:
        return jsonify({"success": False, "error": str(exc)}), 400
    except Exception as exc:
        return jsonify({"success": False, "error": str(exc)}), 500

@app.route('/api/files')
@api_login_required
@with_terminal
def get_files(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    """获取文件树"""
    policy = resolve_admin_policy(get_current_user_record())
    if policy.get("ui_blocks", {}).get("collapse_workspace") or policy.get("ui_blocks", {}).get("block_file_manager"):
        return jsonify({"success": False, "error": "文件浏览已被管理员禁用"}), 403
    structure = terminal.context_manager.get_project_structure()
    return jsonify(structure)


# ==========================================
# 新版 GUI 文件管理器 API
# ==========================================


def _format_entry(entry) -> Dict[str, Any]:
    return {
        "name": entry.name,
        "path": entry.path,
        "type": entry.type,
        "size": entry.size,
        "modified_at": entry.modified_at,
        "extension": entry.extension,
        "is_editable": entry.is_editable,
    }


@app.route('/api/gui/files/entries', methods=['GET'])
@api_login_required
@with_terminal
def gui_list_entries(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    """列出指定目录内容"""
    policy = resolve_admin_policy(get_current_user_record())
    if policy.get("ui_blocks", {}).get("block_file_manager"):
        return jsonify({"success": False, "error": "文件管理已被管理员禁用"}), 403
    relative_path = request.args.get('path') or ""
    manager = get_gui_manager(workspace)
    try:
        resolved_path, entries = manager.list_directory(relative_path)
        breadcrumb = manager.breadcrumb(resolved_path)
        return jsonify({
            "success": True,
            "data": {
                "path": resolved_path,
                "breadcrumb": breadcrumb,
                "items": [_format_entry(entry) for entry in entries]
            }
        })
    except Exception as exc:
        return jsonify({
            "success": False,
            "error": str(exc)
        }), 400


@app.route('/api/gui/files/create', methods=['POST'])
@api_login_required
@with_terminal
@rate_limited("gui_file_create", 30, 60, scope="user")
def gui_create_entry(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    payload = request.get_json() or {}
    parent = payload.get('path') or ""
    name = payload.get('name') or ""
    entry_type = payload.get('type') or "file"
    policy = resolve_admin_policy(get_current_user_record())
    if policy.get("ui_blocks", {}).get("block_file_manager"):
        return jsonify({"success": False, "error": "文件管理已被管理员禁用"}), 403
    manager = get_gui_manager(workspace)
    try:
        new_path = manager.create_entry(parent, name, entry_type)
        return jsonify({
            "success": True,
            "path": new_path
        })
    except Exception as exc:
        return jsonify({
            "success": False,
            "error": str(exc)
        }), 400


@app.route('/api/gui/files/delete', methods=['POST'])
@api_login_required
@with_terminal
@rate_limited("gui_file_delete", 30, 60, scope="user")
def gui_delete_entries(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    payload = request.get_json() or {}
    paths = payload.get('paths') or []
    policy = resolve_admin_policy(get_current_user_record())
    if policy.get("ui_blocks", {}).get("block_file_manager"):
        return jsonify({"success": False, "error": "文件管理已被管理员禁用"}), 403
    manager = get_gui_manager(workspace)
    try:
        result = manager.delete_entries(paths)
        return jsonify({
            "success": True,
            "result": result
        })
    except Exception as exc:
        return jsonify({
            "success": False,
            "error": str(exc)
        }), 400


@app.route('/api/gui/files/rename', methods=['POST'])
@api_login_required
@with_terminal
@rate_limited("gui_file_rename", 30, 60, scope="user")
def gui_rename_entry(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    payload = request.get_json() or {}
    path = payload.get('path')
    new_name = payload.get('new_name')
    if not path or not new_name:
        return jsonify({"success": False, "error": "缺少 path 或 new_name"}), 400
    policy = resolve_admin_policy(get_current_user_record())
    if policy.get("ui_blocks", {}).get("block_file_manager"):
        return jsonify({"success": False, "error": "文件管理已被管理员禁用"}), 403
    manager = get_gui_manager(workspace)
    try:
        new_path = manager.rename_entry(path, new_name)
        return jsonify({
            "success": True,
            "path": new_path
        })
    except Exception as exc:
        return jsonify({
            "success": False,
            "error": str(exc)
        }), 400


@app.route('/api/gui/files/copy', methods=['POST'])
@api_login_required
@with_terminal
@rate_limited("gui_file_copy", 40, 120, scope="user")
def gui_copy_entries(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    payload = request.get_json() or {}
    paths = payload.get('paths') or []
    target_dir = payload.get('target_dir') or ""
    policy = resolve_admin_policy(get_current_user_record())
    if policy.get("ui_blocks", {}).get("block_file_manager"):
        return jsonify({"success": False, "error": "文件管理已被管理员禁用"}), 403
    manager = get_gui_manager(workspace)
    try:
        result = manager.copy_entries(paths, target_dir)
        return jsonify({
            "success": True,
            "result": result
        })
    except Exception as exc:
        return jsonify({
            "success": False,
            "error": str(exc)
        }), 400


@app.route('/api/gui/files/move', methods=['POST'])
@api_login_required
@with_terminal
@rate_limited("gui_file_move", 40, 120, scope="user")
def gui_move_entries(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    payload = request.get_json() or {}
    paths = payload.get('paths') or []
    target_dir = payload.get('target_dir') or ""
    manager = get_gui_manager(workspace)
    try:
        result = manager.move_entries(paths, target_dir)
        return jsonify({
            "success": True,
            "result": result
        })
    except Exception as exc:
        return jsonify({
            "success": False,
            "error": str(exc)
        }), 400


@app.route('/api/gui/files/upload', methods=['POST'])
@api_login_required
@with_terminal
@rate_limited("gui_file_upload", 10, 300, scope="user")
def gui_upload_entry(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    policy = resolve_admin_policy(get_current_user_record())
    if policy.get("ui_blocks", {}).get("block_upload"):
        return jsonify({"success": False, "error": "文件上传已被管理员禁用"}), 403
    if 'file' not in request.files:
        return jsonify({"success": False, "error": "未找到文件"}), 400
    file_obj = request.files['file']
    if not file_obj or not file_obj.filename:
        return jsonify({"success": False, "error": "文件名为空"}), 400
    current_dir = request.form.get('path') or ""
    raw_name = request.form.get('filename') or file_obj.filename
    filename = sanitize_filename_preserve_unicode(raw_name) or secure_filename(raw_name)
    if not filename:
        return jsonify({"success": False, "error": "非法文件名"}), 400
    manager = get_gui_manager(workspace)
    try:
        target_path = manager.prepare_upload(current_dir, filename)
    except Exception as exc:
        return jsonify({"success": False, "error": str(exc)}), 400

    try:
        relative_path = manager._to_relative(target_path)
    except Exception as exc:
        return jsonify({"success": False, "error": str(exc)}), 400

    guard = get_upload_guard(workspace)
    try:
        result = guard.process_upload(
            file_obj,
            target_path,
            username=username,
            source="web_gui",
            original_name=raw_name,
            relative_path=relative_path,
        )
    except UploadSecurityError as exc:
        return build_upload_error_response(exc)
    except Exception as exc:
        return jsonify({"success": False, "error": f"保存文件失败: {exc}"}), 500

    metadata = result.get("metadata", {})
    return jsonify({
        "success": True,
        "path": relative_path,
        "filename": target_path.name,
        "scan": metadata.get("scan"),
        "sha256": metadata.get("sha256"),
        "size": metadata.get("size"),
    })


@app.route('/api/gui/files/download', methods=['GET'])
@api_login_required
@with_terminal
def gui_download_entry(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    path = request.args.get('path')
    if not path:
        return jsonify({"success": False, "error": "缺少 path"}), 400
    manager = get_gui_manager(workspace)
    try:
        target = manager.prepare_download(path)
        if target.is_dir():
            memory_file = BytesIO()
            with zipfile.ZipFile(memory_file, mode='w', compression=zipfile.ZIP_DEFLATED) as zf:
                for root, dirs, files in os.walk(target):
                    for file in files:
                        full_path = Path(root) / file
                        arcname = manager._to_relative(full_path)
                        zf.write(full_path, arcname=arcname)
            memory_file.seek(0)
            download_name = f"{target.name}.zip"
            return send_file(memory_file, as_attachment=True, download_name=download_name, mimetype='application/zip')
        return send_file(target, as_attachment=True, download_name=target.name)
    except Exception as exc:
        return jsonify({"success": False, "error": str(exc)}), 400


@app.route('/api/gui/files/download/batch', methods=['POST'])
@api_login_required
@with_terminal
def gui_download_batch(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    payload = request.get_json() or {}
    paths = payload.get('paths') or []
    if not paths:
        return jsonify({"success": False, "error": "缺少待下载的路径"}), 400
    manager = get_gui_manager(workspace)
    try:
        memory_file = BytesIO()
        with zipfile.ZipFile(memory_file, mode='w', compression=zipfile.ZIP_DEFLATED) as zf:
            for rel in paths:
                target = manager.prepare_download(rel)
                arc_base = rel.strip('/') or target.name
                if target.is_dir():
                    for root, _, files in os.walk(target):
                        for file in files:
                            full_path = Path(root) / file
                            relative_sub = full_path.relative_to(target)
                            arcname = Path(arc_base) / relative_sub
                            zf.write(full_path, arcname=str(arcname))
                else:
                    zf.write(target, arcname=arc_base)
        memory_file.seek(0)
        download_name = f"selected_{len(paths)}.zip"
        return send_file(memory_file, as_attachment=True, download_name=download_name, mimetype='application/zip')
    except Exception as exc:
        return jsonify({"success": False, "error": str(exc)}), 400


@app.route('/api/gui/files/text', methods=['GET', 'POST'])
@api_login_required
@with_terminal
def gui_text_entry(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    manager = get_gui_manager(workspace)
    if request.method == 'GET':
        path = request.args.get('path')
        if not path:
            return jsonify({"success": False, "error": "缺少 path"}), 400
        try:
            content, modified = manager.read_text(path)
            return jsonify({
                "success": True,
                "path": path,
                "content": content,
                "modified_at": modified
            })
        except Exception as exc:
            return jsonify({"success": False, "error": str(exc)}), 400

    payload = request.get_json() or {}
    path = payload.get('path')
    content = payload.get('content')
    if path is None or content is None:
        return jsonify({"success": False, "error": "缺少 path 或 content"}), 400
    try:
        result = manager.write_text(path, content)
        return jsonify({"success": True, "data": result})
    except Exception as exc:
        return jsonify({"success": False, "error": str(exc)}), 400


@app.route('/api/memory', methods=['GET'])
@api_login_required
@with_terminal
def api_memory_entries(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    """返回主/任务记忆条目列表，供虚拟显示器加载"""
    memory_type = request.args.get('type', 'main')
    if memory_type not in ('main', 'task'):
        return jsonify({"success": False, "error": "type 必须是 main 或 task"}), 400
    try:
        entries = terminal.memory_manager._read_entries(memory_type)  # type: ignore
        return jsonify({"success": True, "type": memory_type, "entries": entries})
    except Exception as exc:
        return jsonify({"success": False, "error": str(exc)}), 500


@app.route('/api/gui/monitor_snapshot', methods=['GET'])
@api_login_required
def get_monitor_snapshot_api():
    execution_id = request.args.get('executionId') or request.args.get('execution_id') or request.args.get('id')
    if not execution_id:
        return jsonify({
            'success': False,
            'error': '缺少 executionId 参数'
        }), 400
    stage = (request.args.get('stage') or 'before').lower()
    if stage not in {'before', 'after'}:
        stage = 'before'
    snapshot = get_cached_monitor_snapshot(execution_id, stage)
    if not snapshot:
        return jsonify({
            'success': False,
            'error': '未找到对应快照'
        }), 404
    return jsonify({
        'success': True,
        'snapshot': snapshot,
        'stage': stage
    })

@app.route('/api/focused')
@api_login_required
@with_terminal
def get_focused_files(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    """获取聚焦文件"""
    focused = {}
    for path, content in terminal.focused_files.items():
        focused[path] = {
            "content": content,
            "size": len(content),
            "lines": content.count('\n') + 1
        }
    return jsonify(focused)

@app.route('/api/todo-list')
@api_login_required
@with_terminal
def get_todo_list(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    """获取当前待办列表"""
    todo_snapshot = terminal.context_manager.get_todo_snapshot()
    return jsonify({
        "success": True,
        "data": todo_snapshot
    })

@app.route('/api/upload', methods=['POST'])
@api_login_required
@with_terminal
@rate_limited("legacy_upload", 20, 300, scope="user")
def upload_file(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    """处理前端文件上传请求"""
    policy = resolve_admin_policy(get_current_user_record())
    if policy.get("ui_blocks", {}).get("block_upload"):
        return jsonify({
            "success": False,
            "error": "文件上传已被管理员禁用",
            "message": "被管理员禁用上传"
        }), 403
    if 'file' not in request.files:
        return jsonify({
            "success": False,
            "error": "未找到文件",
            "message": "请求中缺少文件字段"
        }), 400

    uploaded_file = request.files['file']
    original_name = (request.form.get('filename') or '').strip()

    if not uploaded_file or not uploaded_file.filename or uploaded_file.filename.strip() == '':
        return jsonify({
            "success": False,
            "error": "文件名为空",
            "message": "请选择要上传的文件"
        }), 400

    raw_name = original_name or uploaded_file.filename
    filename = sanitize_filename_preserve_unicode(raw_name)
    if not filename:
        filename = secure_filename(raw_name)
    if not filename:
        return jsonify({
            "success": False,
            "error": "非法文件名",
            "message": "文件名包含不支持的字符"
        }), 400

    file_manager = getattr(terminal, 'file_manager', None)
    if file_manager is None:
        return jsonify({
            "success": False,
            "error": "文件管理器未初始化"
        }), 500

    target_folder_relative = UPLOAD_FOLDER_NAME
    valid_folder, folder_error, folder_path = file_manager._validate_path(target_folder_relative)
    if not valid_folder:
        return jsonify({
            "success": False,
            "error": folder_error
        }), 400

    try:
        folder_path.mkdir(parents=True, exist_ok=True)
    except Exception as exc:
        return jsonify({
            "success": False,
            "error": f"创建上传目录失败: {exc}"
        }), 500

    target_relative = str(Path(target_folder_relative) / filename)
    valid_file, file_error, target_full_path = file_manager._validate_path(target_relative)
    if not valid_file:
        return jsonify({
            "success": False,
            "error": file_error
        }), 400

    final_path = target_full_path
    if final_path.exists():
        stem = final_path.stem
        suffix = final_path.suffix
        counter = 1

        while final_path.exists():
            candidate_name = f"{stem}_{counter}{suffix}"
            target_relative = str(Path(target_folder_relative) / candidate_name)
            valid_file, file_error, candidate_path = file_manager._validate_path(target_relative)
            if not valid_file:
                return jsonify({
                    "success": False,
                    "error": file_error
                }), 400
            final_path = candidate_path
            counter += 1

    try:
        relative_path = str(final_path.relative_to(workspace.project_path))
    except Exception as exc:
        return jsonify({
            "success": False,
            "error": f"路径解析失败: {exc}"
        }), 400

    guard = get_upload_guard(workspace)
    try:
        result = guard.process_upload(
            uploaded_file,
            final_path,
            username=username,
            source="legacy_upload",
            original_name=raw_name,
            relative_path=relative_path,
        )
    except UploadSecurityError as exc:
        return build_upload_error_response(exc)
    except Exception as exc:
        return jsonify({
            "success": False,
            "error": f"保存文件失败: {exc}"
        }), 500

    metadata = result.get("metadata", {})
    print(f"{OUTPUT_FORMATS['file']} 上传文件: {relative_path}")

    return jsonify({
        "success": True,
        "path": relative_path,
        "filename": final_path.name,
        "folder": target_folder_relative,
        "scan": metadata.get("scan"),
        "sha256": metadata.get("sha256"),
        "size": metadata.get("size"),
    })


@app.errorhandler(RequestEntityTooLarge)
def handle_file_too_large(error):
    """全局捕获上传超大小"""
    size_mb = MAX_UPLOAD_SIZE / (1024 * 1024)
    return jsonify({
        "success": False,
        "error": "文件过大",
        "message": f"单个文件大小不可超过 {size_mb:.1f} MB"
    }), 413


@app.route('/api/download/file')
@api_login_required
@with_terminal
def download_file_api(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    """下载单个文件"""
    path = (request.args.get('path') or '').strip()
    if not path:
        return jsonify({"success": False, "error": "缺少路径参数"}), 400

    valid, error, full_path = terminal.file_manager._validate_path(path)
    if not valid or full_path is None:
        return jsonify({"success": False, "error": error or "路径校验失败"}), 400
    if not full_path.exists() or not full_path.is_file():
        return jsonify({"success": False, "error": "文件不存在"}), 404

    return send_file(
        full_path,
        as_attachment=True,
        download_name=full_path.name
    )


@app.route('/api/download/folder')
@api_login_required
@with_terminal
def download_folder_api(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    """打包并下载文件夹"""
    path = (request.args.get('path') or '').strip()
    if not path:
        return jsonify({"success": False, "error": "缺少路径参数"}), 400

    valid, error, full_path = terminal.file_manager._validate_path(path)
    if not valid or full_path is None:
        return jsonify({"success": False, "error": error or "路径校验失败"}), 400
    if not full_path.exists() or not full_path.is_dir():
        return jsonify({"success": False, "error": "文件夹不存在"}), 404

    buffer = BytesIO()
    folder_name = Path(path).name or full_path.name or "archive"

    with zipfile.ZipFile(buffer, 'w', zipfile.ZIP_DEFLATED) as zip_buffer:
        # 确保目录本身被包含
        zip_buffer.write(full_path, arcname=folder_name + '/')

        for item in full_path.rglob('*'):
            relative_name = Path(folder_name) / item.relative_to(full_path)
            if item.is_dir():
                zip_buffer.write(item, arcname=str(relative_name) + '/')
            else:
                zip_buffer.write(item, arcname=str(relative_name))

    buffer.seek(0)
    return send_file(
        buffer,
        mimetype='application/zip',
        as_attachment=True,
        download_name=f"{folder_name}.zip"
    )

@app.route('/api/tool-settings', methods=['GET', 'POST'])
@api_login_required
@with_terminal
def tool_settings(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    """获取或更新工具启用状态"""
    if request.method == 'GET':
        snapshot = terminal.get_tool_settings_snapshot()
        return jsonify({
            "success": True,
            "categories": snapshot
        })

    data = request.get_json() or {}
    category = data.get('category')
    if category is None:
        return jsonify({
            "success": False,
            "error": "缺少类别参数",
            "message": "请求体需要提供 category 字段"
        }), 400

    if 'enabled' not in data:
        return jsonify({
            "success": False,
            "error": "缺少启用状态",
            "message": "请求体需要提供 enabled 字段"
        }), 400

    try:
        policy = resolve_admin_policy(get_current_user_record())
        if policy.get("ui_blocks", {}).get("block_tool_toggle"):
            return jsonify({
                "success": False,
                "error": "工具开关已被管理员禁用",
                "message": "被管理员强制禁用"
            }), 403
        enabled = bool(data['enabled'])
        forced = getattr(terminal, "admin_forced_category_states", {}) or {}
        if isinstance(forced.get(category), bool) and forced[category] != enabled:
            return jsonify({
                "success": False,
                "error": "该工具类别已被管理员强制为启用/禁用，无法修改",
                "message": "被管理员强制启用/禁用"
            }), 403
        terminal.set_tool_category_enabled(category, enabled)
        snapshot = terminal.get_tool_settings_snapshot()
        socketio.emit('tool_settings_updated', {
            'categories': snapshot
        }, room=f"user_{username}")
        return jsonify({
            "success": True,
            "categories": snapshot
        })
    except ValueError as exc:
        return jsonify({
            "success": False,
            "error": str(exc)
        }), 400

@app.route('/api/terminals')
@api_login_required
@with_terminal
def get_terminals(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    """获取终端会话列表"""
    policy = resolve_admin_policy(get_current_user_record())
    if policy.get("ui_blocks", {}).get("block_realtime_terminal"):
        return jsonify({"success": False, "error": "实时终端已被管理员禁用"}), 403
    if terminal.terminal_manager:
        result = terminal.terminal_manager.list_terminals()
        return jsonify(result)
    else:
        return jsonify({"sessions": [], "active": None, "total": 0})


@app.route('/api/socket-token', methods=['GET'])
@api_login_required
def issue_socket_token():
    """生成一次性 WebSocket token，供握手阶段使用。"""
    username = get_current_username()
    prune_socket_tokens()
    now = time.time()
    for token_value, meta in list(pending_socket_tokens.items()):
        if meta.get("username") == username:
            pending_socket_tokens.pop(token_value, None)
    token_value = secrets.token_urlsafe(32)
    pending_socket_tokens[token_value] = {
        "username": username,
        "expires_at": now + SOCKET_TOKEN_TTL_SECONDS,
        "fingerprint": (request.headers.get('User-Agent') or '')[:128],
    }
    return jsonify({
        "success": True,
        "token": token_value,
        "expires_in": SOCKET_TOKEN_TTL_SECONDS
    })

@socketio.on('connect')
def handle_connect(auth):
    """客户端连接"""
    print(f"[WebSocket] 客户端连接: {request.sid}")
    username = get_current_username()
    token_value = (auth or {}).get('socket_token') if isinstance(auth, dict) else None
    if not username or not consume_socket_token(token_value, username):
        emit('error', {'message': '未登录或连接凭证无效'})
        disconnect()
        return
    
    emit('connected', {'status': 'Connected to server'})
    connection_users[request.sid] = username
    
    # 清理可能存在的停止标志和状态
    stop_flags.pop(request.sid, None)
    
    join_room(f"user_{username}")
    join_room(f"user_{username}_terminal")
    if request.sid not in terminal_rooms:
        terminal_rooms[request.sid] = set()
    terminal_rooms[request.sid].update({f"user_{username}", f"user_{username}_terminal"})
    
    terminal, workspace = get_user_resources(username)
    if terminal:
        reset_system_state(terminal)
        emit('system_ready', {
            'project_path': str(workspace.project_path),
            'thinking_mode': bool(getattr(terminal, "thinking_mode", False)),
            'version': AGENT_VERSION
        }, room=request.sid)
        
        if terminal.terminal_manager:
            terminals = terminal.terminal_manager.get_terminal_list()
            emit('terminal_list_update', {
                'terminals': terminals,
                'active': terminal.terminal_manager.active_terminal
            }, room=request.sid)
            
            if terminal.terminal_manager.active_terminal:
                for name, term in terminal.terminal_manager.terminals.items():
                    emit('terminal_started', {
                        'session': name,
                        'working_dir': str(term.working_dir),
                        'shell': term.shell_command,
                        'time': term.start_time.isoformat() if term.start_time else None
                    }, room=request.sid)

@socketio.on('disconnect')
def handle_disconnect():
    """客户端断开"""
    print(f"[WebSocket] 客户端断开: {request.sid}")
    username = connection_users.pop(request.sid, None)
    task_info = stop_flags.get(request.sid)
    if isinstance(task_info, dict):
        task_info['stop'] = True
        pending_task = task_info.get('task')
        if pending_task and not pending_task.done():
            debug_log(f"disconnect: cancel task for {request.sid}")
            pending_task.cancel()
        terminal = task_info.get('terminal')
        if terminal:
            reset_system_state(terminal)
    
    # 清理停止标志
    stop_flags.pop(request.sid, None)
    
    # 从所有房间移除
    for room in list(terminal_rooms.get(request.sid, [])):
        leave_room(room)
    if request.sid in terminal_rooms:
        del terminal_rooms[request.sid]
    
    if username:
        leave_room(f"user_{username}")
        leave_room(f"user_{username}_terminal")

@socketio.on('stop_task')
def handle_stop_task():
    """处理停止任务请求"""
    print(f"[停止] 收到停止请求: {request.sid}")
    
    task_info = stop_flags.get(request.sid)
    if not isinstance(task_info, dict):
        task_info = {'stop': False, 'task': None, 'terminal': None}
        stop_flags[request.sid] = task_info
    
    if task_info.get('task') and not task_info['task'].done():
        debug_log(f"正在取消任务: {request.sid}")
        task_info['task'].cancel()
    
    task_info['stop'] = True
    if task_info.get('terminal'):
        reset_system_state(task_info['terminal'])
    
    emit('stop_requested', {
        'message': '停止请求已接收，正在取消任务...'
    })

@socketio.on('terminal_subscribe')
def handle_terminal_subscribe(data):
    """订阅终端事件"""
    session_name = data.get('session')
    subscribe_all = data.get('all', False)

    username, terminal, _ = get_terminal_for_sid(request.sid)
    if not username or not terminal or not terminal.terminal_manager:
        emit('error', {'message': 'Terminal system not initialized'})
        return
    policy = resolve_admin_policy(user_manager.get_user(username))
    if policy.get("ui_blocks", {}).get("block_realtime_terminal"):
        emit('error', {'message': '实时终端已被管理员禁用'})
        return
    
    if request.sid not in terminal_rooms:
        terminal_rooms[request.sid] = set()
    
    if subscribe_all:
        # 订阅所有终端事件
        room_name = f"user_{username}_terminal"
        join_room(room_name)
        terminal_rooms[request.sid].add(room_name)
        print(f"[Terminal] {request.sid} 订阅所有终端事件")
        
        # 发送当前终端状态
        emit('terminal_subscribed', {
            'type': 'all',
            'terminals': terminal.terminal_manager.get_terminal_list()
        })
    elif session_name:
        # 订阅特定终端会话
        room_name = f'user_{username}_terminal_{session_name}'
        join_room(room_name)
        terminal_rooms[request.sid].add(room_name)
        print(f"[Terminal] {request.sid} 订阅终端: {session_name}")
        
        # 发送该终端的当前输出
        output_result = terminal.terminal_manager.get_terminal_output(session_name, 100)
        if output_result['success']:
            emit('terminal_history', {
                'session': session_name,
                'output': output_result['output']
            })

@socketio.on('terminal_unsubscribe')
def handle_terminal_unsubscribe(data):
    """取消订阅终端事件"""
    session_name = data.get('session')
    username = connection_users.get(request.sid)
    
    if session_name:
        room_name = f'user_{username}_terminal_{session_name}' if username else f'terminal_{session_name}'
        leave_room(room_name)
        if request.sid in terminal_rooms:
            terminal_rooms[request.sid].discard(room_name)
        print(f"[Terminal] {request.sid} 取消订阅终端: {session_name}")

@socketio.on('get_terminal_output')
def handle_get_terminal_output(data):
    """获取终端输出历史"""
    session_name = data.get('session')
    lines = data.get('lines', 50)
    
    username, terminal, _ = get_terminal_for_sid(request.sid)
    if not terminal or not terminal.terminal_manager:
        emit('error', {'message': 'Terminal system not initialized'})
        return
    policy = resolve_admin_policy(user_manager.get_user(username))
    if policy.get("ui_blocks", {}).get("block_realtime_terminal"):
        emit('error', {'message': '实时终端已被管理员禁用'})
        return
    
    result = terminal.terminal_manager.get_terminal_output(session_name, lines)
    
    if result['success']:
        emit('terminal_output_history', {
            'session': session_name,
            'output': result['output'],
            'is_interactive': result.get('is_interactive', False),
            'last_command': result.get('last_command', '')
        })
    else:
        emit('error', {'message': result['error']})

@socketio.on('send_message')
def handle_message(data):
    """处理用户消息"""
    username, terminal, workspace = get_terminal_for_sid(request.sid)
    if not terminal:
        emit('error', {'message': 'System not initialized'})
        return
    
    message = (data.get('message') or '').strip()
    images = data.get('images') or []
    if not message and not images:
        emit('error', {'message': '消息不能为空'})
        return
    if images and getattr(terminal, "model_key", None) != "qwen3-vl-plus":
        emit('error', {'message': '当前模型不支持图片，请切换到 Qwen-VL'})
        return
    
    print(f"[WebSocket] 收到消息: {message}")
    debug_log(f"\n{'='*80}\n新任务开始: {message}\n{'='*80}")
    record_user_activity(username)
    
    requested_conversation_id = data.get('conversation_id')
    try:
        conversation_id, created_new = ensure_conversation_loaded(terminal, requested_conversation_id)
    except RuntimeError as exc:
        emit('error', {'message': str(exc)})
        return
    try:
        conv_data = terminal.context_manager.conversation_manager.load_conversation(conversation_id) or {}
    except Exception:
        conv_data = {}
    title = conv_data.get('title', '新对话')

    socketio.emit('conversation_resolved', {
        'conversation_id': conversation_id,
        'title': title,
        'created': created_new
    }, room=request.sid)

    if created_new:
        socketio.emit('conversation_list_update', {
            'action': 'created',
            'conversation_id': conversation_id
        }, room=f"user_{username}")
        socketio.emit('conversation_changed', {
            'conversation_id': conversation_id,
            'title': title
        }, room=request.sid)
    
    client_sid = request.sid

    def send_to_client(event_type, data):
        """发送消息到客户端"""
        socketio.emit(event_type, data, room=client_sid)

    # 模型活动事件：用于刷新“在线”心跳（回复/工具调用都算活动）
    activity_events = {
        'ai_message_start', 'thinking_start', 'thinking_chunk', 'thinking_end',
        'text_start', 'text_chunk', 'text_end',
        'tool_hint', 'tool_preparing', 'tool_start', 'update_action',
        'append_payload', 'modify_payload', 'system_message',
        'task_complete'
    }
    last_model_activity = 0.0

    def send_with_activity(event_type, data):
        """模型产生输出或调用工具时刷新活跃时间，防止长回复被误判下线。"""
        nonlocal last_model_activity
        if event_type in activity_events:
            now = time.time()
            # 轻量节流：1 秒内多次事件只记一次
            if now - last_model_activity >= 1.0:
                record_user_activity(username)
                last_model_activity = now
        send_to_client(event_type, data)

    # 传递客户端ID
    images = data.get('images') or []
    socketio.start_background_task(process_message_task, terminal, message, images, send_with_activity, client_sid, workspace, username)


@socketio.on('client_chunk_log')
def handle_client_chunk_log(data):
    """前端chunk日志上报"""
    conversation_id = data.get('conversation_id')
    chunk_index = int(data.get('index') or data.get('chunk_index') or 0)
    elapsed = float(data.get('elapsed') or 0.0)
    length = int(data.get('length') or len(data.get('content') or ""))
    client_ts = float(data.get('ts') or 0.0)
    log_frontend_chunk(conversation_id, chunk_index, elapsed, length, client_ts)


@socketio.on('client_stream_debug_log')
def handle_client_stream_debug_log(data):
    """前端流式调试日志"""
    if not isinstance(data, dict):
        return
    entry = dict(data)
    entry.setdefault('server_ts', time.time())
    log_streaming_debug_entry(entry)

# 在 web_server.py 中添加以下对话管理API接口
# 添加在现有路由之后，@socketio 事件处理之前

# ==========================================
# 对话管理API接口
# ==========================================

@app.route('/api/conversations', methods=['GET'])
@api_login_required
@with_terminal
def get_conversations(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    """获取对话列表"""
    try:
        # 获取查询参数
        limit = request.args.get('limit', 20, type=int)
        offset = request.args.get('offset', 0, type=int)
        
        # 限制参数范围
        limit = max(1, min(limit, 100))  # 限制在1-100之间
        offset = max(0, offset)
        
        result = terminal.get_conversations_list(limit=limit, offset=offset)
        
        if result["success"]:
            return jsonify({
                "success": True,
                "data": result["data"]
            })
        else:
            return jsonify({
                "success": False,
                "error": result.get("error", "Unknown error"),
                "message": result.get("message", "获取对话列表失败")
            }), 500
            
    except Exception as e:
        print(f"[API] 获取对话列表错误: {e}")
        return jsonify({
            "success": False,
            "error": str(e),
            "message": "获取对话列表时发生异常"
        }), 500

@app.route('/api/conversations', methods=['POST'])
@api_login_required
@with_terminal
def create_conversation(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    """创建新对话"""
    try:
        data = request.get_json() or {}
        # 前端现在期望“新建对话”回到用户配置的默认模型/模式，
        # 只有当客户端显式要求保留当前模式时才使用传入值。
        preserve_mode = bool(data.get('preserve_mode'))
        thinking_mode = data.get('thinking_mode') if preserve_mode and 'thinking_mode' in data else None
        run_mode = data.get('mode') if preserve_mode and 'mode' in data else None
        
        result = terminal.create_new_conversation(thinking_mode=thinking_mode, run_mode=run_mode)
        
        if result["success"]:
            session['run_mode'] = terminal.run_mode
            session['thinking_mode'] = terminal.thinking_mode
            # 广播对话列表更新事件
            socketio.emit('conversation_list_update', {
                'action': 'created',
                'conversation_id': result["conversation_id"]
            }, room=f"user_{username}")
            
            # 广播当前对话切换事件
            socketio.emit('conversation_changed', {
                'conversation_id': result["conversation_id"],
                'title': "新对话"
            }, room=f"user_{username}")
            
            return jsonify(result), 201
        else:
            return jsonify(result), 500
            
    except Exception as e:
        print(f"[API] 创建对话错误: {e}")
        return jsonify({
            "success": False,
            "error": str(e),
            "message": "创建对话时发生异常"
        }), 500

@app.route('/api/conversations/<conversation_id>', methods=['GET'])
@api_login_required
@with_terminal
def get_conversation_info(terminal: WebTerminal, workspace: UserWorkspace, username: str, conversation_id):
    """获取特定对话信息"""
    try:
        # 通过ConversationManager直接获取对话数据
        conversation_data = terminal.context_manager.conversation_manager.load_conversation(conversation_id)
        
        if conversation_data:
            # 提取关键信息，不返回完整消息内容（避免数据量过大）
            info = {
                "id": conversation_data["id"],
                "title": conversation_data["title"],
                "created_at": conversation_data["created_at"],
                "updated_at": conversation_data["updated_at"],
                "metadata": conversation_data["metadata"],
                "messages_count": len(conversation_data.get("messages", []))
            }
            
            return jsonify({
                "success": True,
                "data": info
            })
        else:
            return jsonify({
                "success": False,
                "error": "Conversation not found",
                "message": f"对话 {conversation_id} 不存在"
            }), 404
            
    except Exception as e:
        print(f"[API] 获取对话信息错误: {e}")
        return jsonify({
            "success": False,
            "error": str(e),
            "message": "获取对话信息时发生异常"
        }), 500

@app.route('/api/conversations/<conversation_id>/load', methods=['PUT'])
@api_login_required
@with_terminal
def load_conversation(conversation_id, terminal: WebTerminal, workspace: UserWorkspace, username: str):
    """加载特定对话"""
    try:
        result = terminal.load_conversation(conversation_id)
        
        if result["success"]:
            # 广播对话切换事件
            socketio.emit('conversation_changed', {
                'conversation_id': conversation_id,
                'title': result.get("title", "未知对话"),
                'messages_count': result.get("messages_count", 0)
            }, room=f"user_{username}")
            
            # 广播系统状态更新（因为当前对话改变了）
            status = terminal.get_status()
            socketio.emit('status_update', status, room=f"user_{username}")
            
            # 清理和重置相关UI状态
            socketio.emit('conversation_loaded', {
                'conversation_id': conversation_id,
                'clear_ui': True  # 提示前端清理当前UI状态
            }, room=f"user_{username}")
            
            return jsonify(result)
        else:
            return jsonify(result), 404 if "不存在" in result.get("message", "") else 500
            
    except Exception as e:
        print(f"[API] 加载对话错误: {e}")
        return jsonify({
            "success": False,
            "error": str(e),
            "message": "加载对话时发生异常"
        }), 500

@app.route('/api/conversations/<conversation_id>', methods=['DELETE'])
@api_login_required
@with_terminal
def delete_conversation(conversation_id, terminal: WebTerminal, workspace: UserWorkspace, username: str):
    """删除特定对话"""
    try:
        # 检查是否是当前对话
        is_current = (terminal.context_manager.current_conversation_id == conversation_id)
        
        result = terminal.delete_conversation(conversation_id)
        
        if result["success"]:
            # 广播对话列表更新事件
            socketio.emit('conversation_list_update', {
                'action': 'deleted',
                'conversation_id': conversation_id
            }, room=f"user_{username}")
            
            # 如果删除的是当前对话，广播对话清空事件
            if is_current:
                socketio.emit('conversation_changed', {
                    'conversation_id': None,
                    'title': None,
                    'cleared': True
                }, room=f"user_{username}")
                
                # 更新系统状态
                status = terminal.get_status()
                socketio.emit('status_update', status, room=f"user_{username}")
            
            return jsonify(result)
        else:
            return jsonify(result), 404 if "不存在" in result.get("message", "") else 500
            
    except Exception as e:
        print(f"[API] 删除对话错误: {e}")
        return jsonify({
            "success": False,
            "error": str(e),
            "message": "删除对话时发生异常"
        }), 500

@app.route('/api/conversations/search', methods=['GET'])
@api_login_required
@with_terminal
def search_conversations(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    """搜索对话"""
    try:
        query = request.args.get('q', '').strip()
        limit = request.args.get('limit', 20, type=int)
        
        if not query:
            return jsonify({
                "success": False,
                "error": "Missing query parameter",
                "message": "请提供搜索关键词"
            }), 400
        
        # 限制参数范围
        limit = max(1, min(limit, 50))
        
        result = terminal.search_conversations(query, limit)
        
        return jsonify({
            "success": True,
            "data": {
                "results": result["results"],
                "count": result["count"],
                "query": query
            }
        })
            
    except Exception as e:
        print(f"[API] 搜索对话错误: {e}")
        return jsonify({
            "success": False,
            "error": str(e),
            "message": "搜索对话时发生异常"
        }), 500

@app.route('/api/conversations/<conversation_id>/messages', methods=['GET'])
@api_login_required
@with_terminal
def get_conversation_messages(conversation_id, terminal: WebTerminal, workspace: UserWorkspace, username: str):
    """获取对话的消息历史（可选功能，用于调试或详细查看）"""
    try:
        # 获取完整对话数据
        conversation_data = terminal.context_manager.conversation_manager.load_conversation(conversation_id)
        
        if conversation_data:
            messages = conversation_data.get("messages", [])
            
            # 可选：限制消息数量，避免返回过多数据
            limit = request.args.get('limit', type=int)
            if limit:
                messages = messages[-limit:]  # 获取最后N条消息
            
            return jsonify({
                "success": True,
                "data": {
                    "conversation_id": conversation_id,
                    "messages": messages,
                    "total_count": len(conversation_data.get("messages", []))
                }
            })
        else:
            return jsonify({
                "success": False,
                "error": "Conversation not found",
                "message": f"对话 {conversation_id} 不存在"
            }), 404
            
    except Exception as e:
        print(f"[API] 获取对话消息错误: {e}")
        return jsonify({
            "success": False,
            "error": str(e),
            "message": "获取对话消息时发生异常"
        }), 500


@app.route('/api/conversations/<conversation_id>/compress', methods=['POST'])
@api_login_required
@with_terminal
def compress_conversation(conversation_id, terminal: WebTerminal, workspace: UserWorkspace, username: str):
    """压缩指定对话的大体积消息，生成压缩版新对话"""
    try:
        policy = resolve_admin_policy(get_current_user_record())
        if policy.get("ui_blocks", {}).get("block_compress_conversation"):
            return jsonify({"success": False, "error": "压缩对话已被管理员禁用"}), 403
        normalized_id = conversation_id if conversation_id.startswith('conv_') else f"conv_{conversation_id}"
        result = terminal.context_manager.compress_conversation(normalized_id)

        if not result.get("success"):
            status_code = 404 if "不存在" in result.get("error", "") else 400
            return jsonify(result), status_code

        new_conversation_id = result["compressed_conversation_id"]

        load_result = terminal.load_conversation(new_conversation_id)

        if load_result.get("success"):
            socketio.emit('conversation_list_update', {
                'action': 'compressed',
                'conversation_id': new_conversation_id
            }, room=f"user_{username}")
            socketio.emit('conversation_changed', {
                'conversation_id': new_conversation_id,
                'title': load_result.get('title', '压缩后的对话'),
                'messages_count': load_result.get('messages_count', 0)
            }, room=f"user_{username}")
            socketio.emit('conversation_loaded', {
                'conversation_id': new_conversation_id,
                'clear_ui': True
            }, room=f"user_{username}")

        response_payload = {
            "success": True,
            "compressed_conversation_id": new_conversation_id,
            "compressed_types": result.get("compressed_types", []),
            "system_message": result.get("system_message"),
            "load_result": load_result
        }

        return jsonify(response_payload)

    except Exception as e:
        print(f"[API] 压缩对话错误: {e}")
        return jsonify({
            "success": False,
            "error": str(e),
            "message": "压缩对话时发生异常"
        }), 500


@app.route('/api/sub_agents', methods=['GET'])
@api_login_required
@with_terminal
def list_sub_agents(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    """返回当前对话的子智能体任务列表。"""
    manager = getattr(terminal, "sub_agent_manager", None)
    if not manager:
        return jsonify({"success": True, "data": []})
    try:
        conversation_id = terminal.context_manager.current_conversation_id
        data = manager.get_overview(conversation_id=conversation_id)
        return jsonify({"success": True, "data": data})
    except Exception as exc:
        return jsonify({"success": False, "error": str(exc)}), 500


@app.route('/api/conversations/<conversation_id>/duplicate', methods=['POST'])
@api_login_required
@with_terminal
def duplicate_conversation(conversation_id, terminal: WebTerminal, workspace: UserWorkspace, username: str):
    """复制指定对话，生成新的对话副本"""

    try:
        result = terminal.context_manager.duplicate_conversation(conversation_id)

        if not result.get("success"):
            status_code = 404 if "不存在" in result.get("error", "") else 400
            return jsonify(result), status_code

        new_conversation_id = result["duplicate_conversation_id"]
        load_result = terminal.load_conversation(new_conversation_id)

        if load_result.get("success"):
            socketio.emit('conversation_list_update', {
                'action': 'duplicated',
                'conversation_id': new_conversation_id
            }, room=f"user_{username}")
            socketio.emit('conversation_changed', {
                'conversation_id': new_conversation_id,
                'title': load_result.get('title', '复制的对话'),
                'messages_count': load_result.get('messages_count', 0)
            }, room=f"user_{username}")
            socketio.emit('conversation_loaded', {
                'conversation_id': new_conversation_id,
                'clear_ui': True
            }, room=f"user_{username}")

        response_payload = {
            "success": True,
            "duplicate_conversation_id": new_conversation_id,
            "load_result": load_result
        }

        return jsonify(response_payload)

    except Exception as e:
        print(f"[API] 复制对话错误: {e}")
        return jsonify({
            "success": False,
            "error": str(e),
            "message": "复制对话时发生异常"
        }), 500


@app.route('/api/conversations/<conversation_id>/review_preview', methods=['GET'])
@api_login_required
@with_terminal
def review_conversation_preview(conversation_id, terminal: WebTerminal, workspace: UserWorkspace, username: str):
    """生成对话回顾预览（不落盘，只返回前若干行文本）"""
    policy = resolve_admin_policy(get_current_user_record())
    if policy.get("ui_blocks", {}).get("block_conversation_review"):
        return jsonify({"success": False, "error": "对话引用已被管理员禁用"}), 403
    try:
        current_id = terminal.context_manager.current_conversation_id
        if conversation_id == current_id:
            return jsonify({
                "success": False,
                "message": "无法引用当前对话"
            }), 400

        conversation_data = terminal.context_manager.conversation_manager.load_conversation(conversation_id)
        if not conversation_data:
            return jsonify({
                "success": False,
                "error": "Conversation not found",
                "message": f"对话 {conversation_id} 不存在"
            }), 404

        limit = request.args.get('limit', default=20, type=int) or 20
        lines = build_review_lines(conversation_data.get("messages", []), limit=limit)

        return jsonify({
            "success": True,
            "data": {
                "preview": lines,
                "count": len(lines)
            }
        })
    except Exception as e:
        print(f"[API] 对话回顾预览错误: {e}")
        return jsonify({
            "success": False,
            "error": str(e),
            "message": "生成预览时发生异常"
        }), 500


@app.route('/api/conversations/<conversation_id>/review', methods=['POST'])
@api_login_required
@with_terminal
def review_conversation(conversation_id, terminal: WebTerminal, workspace: UserWorkspace, username: str):
    """生成完整对话回顾 Markdown 文件"""
    policy = resolve_admin_policy(get_current_user_record())
    if policy.get("ui_blocks", {}).get("block_conversation_review"):
        return jsonify({"success": False, "error": "对话引用已被管理员禁用"}), 403
    try:
        current_id = terminal.context_manager.current_conversation_id
        if conversation_id == current_id:
            return jsonify({
                "success": False,
                "message": "无法引用当前对话"
            }), 400

        conversation_data = terminal.context_manager.conversation_manager.load_conversation(conversation_id)
        if not conversation_data:
            return jsonify({
                "success": False,
                "error": "Conversation not found",
                "message": f"对话 {conversation_id} 不存在"
            }), 404

        messages = conversation_data.get("messages", [])
        lines = build_review_lines(messages)
        content = "\n".join(lines) + "\n"
        char_count = len(content)

        uploads_dir = workspace.uploads_dir / "review"
        uploads_dir.mkdir(parents=True, exist_ok=True)

        title = conversation_data.get("title") or "untitled"
        safe_title = _sanitize_filename_component(title)
        timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
        filename = f"review_{safe_title}_{timestamp}.md"
        target = uploads_dir / filename

        target.write_text(content, encoding='utf-8')

        return jsonify({
            "success": True,
            "data": {
                "path": f"user_upload/review/{filename}",
                "char_count": char_count
            }
        })
    except Exception as e:
        print(f"[API] 对话回顾生成错误: {e}")
        return jsonify({
            "success": False,
            "error": str(e),
            "message": "生成对话回顾时发生异常"
        }), 500

@app.route('/api/conversations/statistics', methods=['GET'])
@api_login_required
@with_terminal
def get_conversations_statistics(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    """获取对话统计信息"""
    try:
        stats = terminal.context_manager.get_conversation_statistics()
        
        return jsonify({
            "success": True,
            "data": stats
        })
            
    except Exception as e:
        print(f"[API] 获取对话统计错误: {e}")
        return jsonify({
            "success": False,
            "error": str(e),
            "message": "获取对话统计时发生异常"
        }), 500

@app.route('/api/conversations/current', methods=['GET'])
@api_login_required
@with_terminal
def get_current_conversation(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    """获取当前对话信息"""
    current_id = terminal.context_manager.current_conversation_id
    
    # 如果是临时ID，返回空的对话信息
    if not current_id or current_id.startswith('temp_'):
        return jsonify({
            "success": True,
            "data": {
                "id": current_id,
                "title": "新对话",
                "messages_count": 0,
                "is_temporary": True
            }
        })
    
    # 如果是真实的对话ID，查找对话数据
    try:
        conversation_data = terminal.context_manager.conversation_manager.load_conversation(current_id)
        if conversation_data:
            return jsonify({
                "success": True,
                "data": {
                    "id": current_id,
                    "title": conversation_data.get("title", "未知对话"),
                    "messages_count": len(conversation_data.get("messages", [])),
                    "is_temporary": False
                }
            })
        else:
            return jsonify({
                "success": False,
                "error": "对话不存在"
            }), 404
            
    except Exception as e:
        print(f"[API] 获取当前对话错误: {e}")
        return jsonify({
            "success": False,
            "error": str(e)
        }), 500
    
def process_message_task(terminal: WebTerminal, message: str, images, sender, client_sid, workspace: UserWorkspace, username: str):
    """在后台处理消息任务"""
    try:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        
        # 创建可取消的任务
        task = loop.create_task(handle_task_with_sender(terminal, workspace, message, images, sender, client_sid, username))
        
        entry = stop_flags.get(client_sid)
        if not isinstance(entry, dict):
            entry = {'stop': False, 'task': None, 'terminal': None}
            stop_flags[client_sid] = entry
        entry['stop'] = False
        entry['task'] = task
        entry['terminal'] = terminal
        
        try:
            loop.run_until_complete(task)
        except asyncio.CancelledError:
            debug_log(f"任务 {client_sid} 被成功取消")
            sender('task_stopped', {
                'message': '任务已停止',
                'reason': 'user_requested'
            })
            reset_system_state(terminal)
        
        loop.close()
    except Exception as e:
        # 【新增】错误时确保对话状态不丢失
        try:
            if terminal and terminal.context_manager:
                # 尝试保存当前对话状态
                terminal.context_manager.auto_save_conversation()
                debug_log("错误恢复：对话状态已保存")
        except Exception as save_error:
            debug_log(f"错误恢复：保存对话状态失败: {save_error}")
        
        # 原有的错误处理逻辑
        print(f"[Task] 错误: {e}")
        debug_log(f"任务处理错误: {e}")
        import traceback
        traceback.print_exc()
        sender('error', {'message': str(e)})
        sender('task_complete', {
            'total_iterations': 0,
            'total_tool_calls': 0,
            'auto_fix_attempts': 0,
            'error': str(e)
        })

    finally:
        # 清理任务引用
        stop_flags.pop(client_sid, None)

def detect_malformed_tool_call(text):
    """检测文本中是否包含格式错误的工具调用"""
    # 检测多种可能的工具调用格式
    patterns = [
        r'执行工具[:：]\s*\w+<.*?tool.*?sep.*?>',  # 执行工具: xxx<｜tool▼sep｜>
        r'<\|?tool[_▼]?call[_▼]?start\|?>',  # <｜tool_call_start｜>
        r'```tool[_\s]?call',  # ```tool_call 或 ```tool call
        r'{\s*"tool":\s*"[^"]+",\s*"arguments"',  # JSON格式的工具调用
        r'function_calls?:\s*\[?\s*{',  # function_call: [{
    ]
    
    for pattern in patterns:
        if re.search(pattern, text, re.IGNORECASE):
            return True
    
    # 检测特定的工具名称后跟JSON
    tool_names = ['create_file', 'read_file', 'write_file_diff', 'delete_file', 
                  'terminal_session', 'terminal_input', 'web_search', 
                  'extract_webpage', 'save_webpage',
                  'run_python', 'run_command', 'focus_file', 'unfocus_file', 'sleep']
    for tool in tool_names:
        if tool in text and '{' in text:
            # 可能是工具调用但格式错误
            return True
            
    return False

async def handle_task_with_sender(terminal: WebTerminal, workspace: UserWorkspace, message, images, sender, client_sid, username: str):
    """处理任务并发送消息 - 集成token统计版本"""
    web_terminal = terminal
    conversation_id = getattr(web_terminal.context_manager, "current_conversation_id", None)
    
    # 如果是思考模式，重置状态
    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
    
    # 添加到对话历史
    history_len_before = len(getattr(web_terminal.context_manager, "conversation_history", []) or [])
    is_first_user_message = history_len_before == 0
    web_terminal.context_manager.add_conversation("user", message, images=images)
    
    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)
        if auto_title_enabled:
            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
            )
    
    # === 移除：不在这里计算输入token，改为在每次API调用前计算 ===
    
    # 构建上下文和消息（用于API调用）
    context = web_terminal.build_context()
    messages = web_terminal.build_messages(context, message)
    tools = web_terminal.define_tools()
    
    # 开始新的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] = {}
    
    # 设置最大迭代次数
    max_iterations = MAX_ITERATIONS_PER_TASK
    
    pending_append = None  # {"path": str, "tool_call_id": str, "buffer": str, ...}
    append_probe_buffer = ""
    pending_modify = None  # {"path": str, "tool_call_id": str, "buffer": str, ...}
    modify_probe_buffer = ""

    def extract_intent_from_partial(arg_str: str) -> Optional[str]:
        """从不完整的JSON字符串中粗略提取 intent 字段，容错用于流式阶段。"""
        if not arg_str or "intent" not in arg_str:
            return None
        import re
        # 匹配 "intent": "xxx" 形式，允许前面有换行或空格；宽松匹配未闭合的引号
        match = re.search(r'"intent"\s*:\s*"([^"]{0,128})', arg_str, re.IGNORECASE | re.DOTALL)
        if match:
            return match.group(1)
        return None

    def resolve_monitor_path(args: Dict[str, Any], fallback: Optional[str] = None) -> Optional[str]:
        candidates = [
            args.get('path'),
            args.get('target_path'),
            args.get('file_path'),
            args.get('destination_path'),
            fallback
        ]
        for candidate in candidates:
            if isinstance(candidate, str):
                trimmed = candidate.strip()
                if trimmed:
                    return trimmed
        return None


    def resolve_monitor_memory(entries: Any) -> Optional[List[str]]:
        if isinstance(entries, list):
            return [str(item) for item in entries][:MONITOR_MEMORY_ENTRY_LIMIT]
        return None

    def capture_monitor_snapshot(path: Optional[str]) -> Optional[Dict[str, Any]]:
        if not path:
            return None
        try:
            read_result = web_terminal.file_manager.read_file(path)
        except Exception as exc:
            debug_log(f"[MonitorSnapshot] 读取文件失败: {path} ({exc})")
            return None
        if not isinstance(read_result, dict) or not read_result.get('success'):
            return None
        content = read_result.get('content')
        if not isinstance(content, str):
            content = ''
        if len(content) > MONITOR_SNAPSHOT_CHAR_LIMIT:
            content = content[:MONITOR_SNAPSHOT_CHAR_LIMIT]
        return {
            'path': read_result.get('path') or path,
            'content': content
        }
    
    async def finalize_pending_append(response_text: str, stream_completed: bool, finish_reason: str = None) -> Dict:
        """在流式输出结束后处理追加写入"""
        nonlocal pending_append, append_probe_buffer
        
        result = {
            "handled": False,
            "success": False,
            "summary": None,
            "summary_message": None,
            "tool_content": None,
            "tool_call_id": None,
            "path": None,
            "forced": False,
            "error": None,
            "assistant_content": response_text,
            "lines": 0,
            "bytes": 0,
            "finish_reason": finish_reason,
            "appended_content": "",
            "assistant_metadata": None
        }
        
        if not pending_append:
            return result
        
        state = pending_append
        path = state.get("path")
        tool_call_id = state.get("tool_call_id")
        buffer = state.get("buffer", "")
        start_marker = state.get("start_marker")
        end_marker = state.get("end_marker")
        start_idx = state.get("content_start")
        end_idx = state.get("end_index")
        
        display_id = state.get("display_id")
        
        result.update({
            "handled": True,
            "path": path,
            "tool_call_id": tool_call_id,
            "display_id": display_id
        })
        
        if path is None or tool_call_id is None:
            error_msg = "append_to_file 状态不完整，缺少路径或ID。"
            debug_log(error_msg)
            result["error"] = error_msg
            result["summary_message"] = error_msg
            result["tool_content"] = json.dumps({
                "success": False,
                "error": error_msg
            }, ensure_ascii=False)
            if display_id:
                sender('update_action', {
                    'id': display_id,
                    'status': 'failed',
                    'preparing_id': tool_call_id,
                    'message': error_msg
                })
            pending_append = None
            return result
        
        if start_idx is None:
            error_msg = f"未检测到格式正确的开始标识 {start_marker}。"
            debug_log(error_msg)
            result["error"] = error_msg
            result["summary_message"] = error_msg
            result["tool_content"] = json.dumps({
                "success": False,
                "path": path,
                "error": error_msg
            }, ensure_ascii=False)
            if display_id:
                sender('update_action', {
                    'id': display_id,
                    'status': 'failed',
                    'preparing_id': tool_call_id,
                    'message': error_msg
                })
            pending_append = None
            return result
        
        forced = False
        if end_idx is None:
            forced = True
            # 查找下一个<<<，否则使用整个缓冲结尾
            remaining = buffer[start_idx:]
            next_marker = remaining.find("<<<", len(end_marker))
            if next_marker != -1:
                end_idx = start_idx + next_marker
            else:
                end_idx = len(buffer)
        
        content = buffer[start_idx:end_idx]
        if content.startswith('\n'):
            content = content[1:]
        
        if not content:
            error_msg = "未检测到需要追加的内容，请严格按照<<<APPEND:path>>>...<<<END_APPEND>>>格式输出。"
            debug_log(error_msg)
            result["error"] = error_msg
            result["forced"] = forced
            result["tool_content"] = json.dumps({
                "success": False,
                "path": path,
                "error": error_msg
            }, ensure_ascii=False)
            if display_id:
                sender('update_action', {
                    'id': display_id,
                    'status': 'failed',
                    'preparing_id': tool_call_id,
                    'message': error_msg
                })
            pending_append = None
            return result
        
        assistant_message_lines = []
        if start_marker:
            assistant_message_lines.append(start_marker)
        assistant_message_lines.append(content)
        if not forced and end_marker:
            assistant_message_lines.append(end_marker)
        assistant_message_text = "\n".join(assistant_message_lines)
        result["assistant_content"] = assistant_message_text
        assistant_metadata = {
            "append_payload": {
                "path": path,
                "tool_call_id": tool_call_id,
                "forced": forced,
                "has_end_marker": not forced
            }
        }
        result["assistant_metadata"] = assistant_metadata
        
        write_result = web_terminal.file_manager.append_file(path, content)
        if write_result.get("success"):
            bytes_written = len(content.encode('utf-8'))
            line_count = content.count('\n')
            if content and not content.endswith('\n'):
                line_count += 1
            
            summary = f"已向 {path} 追加 {line_count} 行（{bytes_written} 字节）"
            if forced:
                summary += "。未检测到 <<<END_APPEND>>> 标记，系统已在流结束处完成写入。如内容未完成，请重新调用 append_to_file 并按标准格式补充；如已完成，可继续后续步骤。"
            
            result.update({
                "success": True,
                "summary": summary,
                "summary_message": summary,
                "forced": forced,
                "lines": line_count,
                "bytes": bytes_written,
                "appended_content": content,
                "tool_content": json.dumps({
                    "success": True,
                    "path": path,
                    "lines": line_count,
                    "bytes": bytes_written,
                    "forced": forced,
                    "message": summary,
                    "finish_reason": finish_reason
                }, ensure_ascii=False)
            })
            
            assistant_meta_payload = result["assistant_metadata"]["append_payload"]
            assistant_meta_payload["lines"] = line_count
            assistant_meta_payload["bytes"] = bytes_written
            assistant_meta_payload["success"] = True
            
            summary_payload = {
                "success": True,
                "path": path,
                "lines": line_count,
                "bytes": bytes_written,
                "forced": forced,
                "message": summary
            }
            
            if display_id:
                sender('update_action', {
                    'id': display_id,
                    'status': 'completed',
                    'result': summary_payload,
                    'preparing_id': tool_call_id,
                    'message': summary
                })
            
            # 更新聚焦文件内容
            if path in web_terminal.focused_files:
                refreshed = web_terminal.file_manager.read_file(path)
                if refreshed.get("success"):
                    web_terminal.focused_files[path] = refreshed["content"]
                    debug_log(f"聚焦文件已刷新: {path}")
            
            debug_log(f"追加写入完成: {summary}")
        else:
            error_msg = write_result.get("error", "追加写入失败")
            result.update({
                "error": error_msg,
                "summary_message": error_msg,
                "forced": forced,
                "appended_content": content,
                "tool_content": json.dumps({
                    "success": False,
                    "path": path,
                    "error": error_msg,
                    "finish_reason": finish_reason
                }, ensure_ascii=False)
            })
            debug_log(f"追加写入失败: {error_msg}")
            
            if result["assistant_metadata"]:
                assistant_meta_payload = result["assistant_metadata"]["append_payload"]
                assistant_meta_payload["lines"] = content.count('\n') + (0 if content.endswith('\n') or not content else 1)
                assistant_meta_payload["bytes"] = len(content.encode('utf-8'))
                assistant_meta_payload["success"] = False
            
            failure_payload = {
                "success": False,
                "path": path,
                "error": error_msg,
                "forced": forced
            }
            
            if display_id:
                sender('update_action', {
                    'id': display_id,
                    'status': 'completed',
                    'result': failure_payload,
                    'preparing_id': tool_call_id,
                    'message': error_msg
                })
        
        pending_append = None
        append_probe_buffer = ""
        if hasattr(web_terminal, "pending_append_request"):
            web_terminal.pending_append_request = None
        return result
    
    async def finalize_pending_modify(response_text: str, stream_completed: bool, finish_reason: str = None) -> Dict:
        """在流式输出结束后处理修改写入"""
        nonlocal pending_modify, modify_probe_buffer
        
        result = {
            "handled": False,
            "success": False,
            "path": None,
            "tool_call_id": None,
            "display_id": None,
            "total_blocks": 0,
            "completed_blocks": [],
            "failed_blocks": [],
            "forced": False,
            "details": [],
            "error": None,
            "assistant_content": response_text,
            "assistant_metadata": None,
            "tool_content": None,
            "summary_message": None,
            "finish_reason": finish_reason
        }
        
        if not pending_modify:
            return result
        
        state = pending_modify
        path = state.get("path")
        tool_call_id = state.get("tool_call_id")
        display_id = state.get("display_id")
        start_marker = state.get("start_marker")
        end_marker = state.get("end_marker")
        buffer = state.get("buffer", "")
        raw_buffer = state.get("raw_buffer", "")
        end_index = state.get("end_index")
        
        result.update({
            "handled": True,
            "path": path,
            "tool_call_id": tool_call_id,
            "display_id": display_id
        })
        
        if not state.get("start_seen"):
            error_msg = "未检测到格式正确的 <<<MODIFY:path>>> 标记。"
            debug_log(error_msg)
            result["error"] = error_msg
            result["summary_message"] = error_msg
            result["tool_content"] = json.dumps({
                "success": False,
                "path": path,
                "error": error_msg,
                "finish_reason": finish_reason
            }, ensure_ascii=False)
            if display_id:
                sender('update_action', {
                    'id': display_id,
                    'status': 'failed',
                    'preparing_id': tool_call_id,
                    'message': error_msg
                })
            if hasattr(web_terminal, "pending_modify_request"):
                web_terminal.pending_modify_request = None
            pending_modify = None
            modify_probe_buffer = ""
            return result
        
        forced = end_index is None
        apply_text = buffer if forced else buffer[:end_index]
        raw_content = raw_buffer if forced else raw_buffer[:len(start_marker) + end_index + len(end_marker)]
        if raw_content:
            result["assistant_content"] = raw_content
        
        blocks_info = []
        block_reports = {}
        detected_indices = set()
        block_pattern = re.compile(r"\[replace:(\d+)\](.*?)\[/replace\]", re.DOTALL)
        structure_warnings: List[str] = []
        structure_detail_entries: List[Dict] = []

        def record_structure_warning(message: str, hint: Optional[str] = None):
            """记录结构性缺陷，便于给出更具体的反馈。"""
            if message in structure_warnings:
                return
            structure_warnings.append(message)
            structure_detail_entries.append({
                "index": 0,
                "status": "failed",
                "reason": message,
                "removed_lines": 0,
                "added_lines": 0,
                "hint": hint or "请严格按照模板输出：[replace:n] + <<OLD>>/<<NEW>> + [/replace]，并使用 <<<END_MODIFY>>> 收尾。"
            })
        
        def extract_segment(body: str, tag: str):
            marker = f"<<{tag}>>"
            end_tag = "<<END>>"
            start_pos = body.find(marker)
            if start_pos == -1:
                return None, f"缺少 {marker}"
            start_pos += len(marker)
            if body[start_pos:start_pos+2] == "\r\n":
                start_pos += 2
            elif body[start_pos:start_pos+1] == "\n":
                start_pos += 1
            end_pos = body.find(end_tag, start_pos)
            if end_pos == -1:
                return None, f"缺少 {end_tag}"
            segment = body[start_pos:end_pos]
            return segment, None
        
        for match in block_pattern.finditer(apply_text):
            try:
                index = int(match.group(1))
            except ValueError:
                continue
            body = match.group(2)
            if index in detected_indices:
                continue
            detected_indices.add(index)
            block_reports[index] = {
                "index": index,
                "status": "pending",
                "reason": None,
                "removed_lines": 0,
                "added_lines": 0,
                "hint": None
            }
            old_content, old_error = extract_segment(body, "OLD")
            new_content, new_error = extract_segment(body, "NEW")
            if old_error or new_error:
                reason = old_error or new_error
                block_reports[index]["status"] = "failed"
                block_reports[index]["reason"] = reason
            blocks_info.append({
                "index": index,
                "old": old_content,
                "new": new_content,
                "error": old_error or new_error
            })
        
        if not blocks_info:
            has_replace_start = bool(re.search(r"\[replace:\s*\d+\]", apply_text))
            has_replace_end = "[/replace]" in apply_text
            has_old_tag = "<<OLD>>" in apply_text
            has_new_tag = "<<NEW>>" in apply_text

            if has_replace_start and not has_replace_end:
                record_structure_warning("检测到 [replace:n] 标记但缺少对应的 [/replace] 结束标记。")
            if has_replace_end and not has_replace_start:
                record_structure_warning("检测到 [/replace] 结束标记但缺少对应的 [replace:n] 起始标记。")

            old_tags = len(re.findall(r"<<OLD>>", apply_text))
            completed_old_tags = len(re.findall(r"<<OLD>>[\s\S]*?<<END>>", apply_text))
            if old_tags and completed_old_tags < old_tags:
                record_structure_warning("检测到 <<OLD>> 段落但未看到对应的 <<END>> 结束标记。")

            new_tags = len(re.findall(r"<<NEW>>", apply_text))
            completed_new_tags = len(re.findall(r"<<NEW>>[\s\S]*?<<END>>", apply_text))
            if new_tags and completed_new_tags < new_tags:
                record_structure_warning("检测到 <<NEW>> 段落但未看到对应的 <<END>> 结束标记。")

            if (has_replace_start or has_replace_end or has_old_tag or has_new_tag) and not structure_warnings:
                record_structure_warning("检测到部分补丁标记，但整体结构不完整，请严格按照模板填写所有标记。")

        total_blocks = len(blocks_info)
        result["total_blocks"] = total_blocks
        if forced:
            debug_log("未检测到 <<<END_MODIFY>>>，将在流结束处执行已识别的修改块。")
            result["forced"] = True
        
        blocks_to_apply = [
            {"index": block["index"], "old": block["old"], "new": block["new"]}
            for block in blocks_info
            if block["error"] is None and block["old"] is not None and block["new"] is not None
        ]
        
        # 记录格式残缺的块
        for block in blocks_info:
            if block["error"]:
                idx = block["index"]
                block_reports[idx]["status"] = "failed"
                block_reports[idx]["reason"] = block["error"]
                block_reports[idx]["hint"] = "请检查补丁块的 OLD/NEW 标记是否完整，必要时复用 terminal_snapshot 或终端命令重新调整。"
        
        apply_result = {}
        if blocks_to_apply:
            apply_result = web_terminal.file_manager.apply_modify_blocks(path, blocks_to_apply)
        else:
            apply_result = {"success": False, "completed": [], "failed": [], "results": [], "write_performed": False, "error": None}
        
        block_result_map = {item["index"]: item for item in apply_result.get("results", [])}
        
        for block in blocks_info:
            idx = block["index"]
            report = block_reports.get(idx)
            if report is None:
                continue
            if report["status"] == "failed":
                continue
            block_apply = block_result_map.get(idx)
            if not block_apply:
                report["status"] = "failed"
                report["reason"] = "未执行，可能未找到匹配原文"
                report["hint"] = report.get("hint") or "请确认 OLD 文本与文件内容完全一致；若多次失败，可改用终端命令/Python 进行精准替换。"
                continue
            status = block_apply.get("status")
            report["removed_lines"] = block_apply.get("removed_lines", 0)
            report["added_lines"] = block_apply.get("added_lines", 0)
            if block_apply.get("hint"):
                report["hint"] = block_apply.get("hint")
            if status == "success":
                report["status"] = "completed"
            elif status == "not_found":
                report["status"] = "failed"
                report["reason"] = block_apply.get("reason") or "未找到匹配的原文"
                if not report.get("hint"):
                    report["hint"] = "请使用 terminal_snapshot/grep -n 校验原文，或在说明后改用 run_command/python 精确替换。"
            else:
                report["status"] = "failed"
                report["reason"] = block_apply.get("reason") or "替换失败"
                if not report.get("hint"):
                    report["hint"] = block_apply.get("hint") or "若多次尝试仍失败，可考虑利用终端命令或 Python 小脚本完成此次修改。"
        
        completed_blocks = sorted([idx for idx, rep in block_reports.items() if rep["status"] == "completed"])
        failed_blocks = sorted([idx for idx, rep in block_reports.items() if rep["status"] != "completed"])
        
        result["completed_blocks"] = completed_blocks
        result["failed_blocks"] = failed_blocks
        details = sorted(block_reports.values(), key=lambda x: x["index"])
        if structure_detail_entries:
            details = structure_detail_entries + details
        result["details"] = details
        
        summary_parts = []
        if total_blocks == 0:
            summary_parts.append("未检测到有效的修改块，未执行任何修改。")
            summary_parts.extend(structure_warnings)
        else:
            if not completed_blocks and failed_blocks:
                summary_parts.append(f"共检测到 {total_blocks} 个修改块，全部未执行。")
            elif completed_blocks and not failed_blocks:
                summary_parts.append(f"共 {total_blocks} 个修改块全部完成。")
            else:
                summary_parts.append(
                    f"共检测到 {total_blocks} 个修改块，其中成功 {len(completed_blocks)} 个，失败 {len(failed_blocks)} 个。"
                )
        if forced:
            summary_parts.append("未检测到 <<<END_MODIFY>>> 标记，系统已在流结束处执行补丁。")
        if apply_result.get("error"):
            summary_parts.append(apply_result["error"])
        
        matching_note = "提示：补丁匹配基于完整文本，包含注释和空白符，请确保 <<<OLD>>> 段落与文件内容逐字一致。如果修改成功，请忽略，如果失败，请明确原文后再次尝试。"
        summary_parts.append(matching_note)
        summary_message = " ".join(summary_parts).strip()
        result["summary_message"] = summary_message
        result["success"] = bool(completed_blocks) and not failed_blocks and apply_result.get("error") is None
        
        tool_payload = {
            "success": result["success"],
            "path": path,
            "total_blocks": total_blocks,
            "completed": completed_blocks,
            "failed": [
                {
                    "index": rep["index"],
                    "reason": rep.get("reason"),
                    "hint": rep.get("hint")
                }
                for rep in result["details"] if rep["status"] != "completed"
            ],
            "forced": forced,
            "message": summary_message,
            "finish_reason": finish_reason,
            "details": result["details"]
        }
        if apply_result.get("error"):
            tool_payload["error"] = apply_result["error"]
        
        result["tool_content"] = json.dumps(tool_payload, ensure_ascii=False)
        result["assistant_metadata"] = {
            "modify_payload": {
                "path": path,
                "total_blocks": total_blocks,
                "completed": completed_blocks,
                "failed": failed_blocks,
                "forced": forced,
                "details": result["details"]
            }
        }
        
        if display_id:
            sender('update_action', {
                'id': display_id,
                'status': 'completed' if result["success"] else 'failed',
                'result': tool_payload,
                'preparing_id': tool_call_id,
                'message': summary_message
            })
        
        if path in web_terminal.focused_files and tool_payload.get("success"):
            refreshed = web_terminal.file_manager.read_file(path)
            if refreshed.get("success"):
                web_terminal.focused_files[path] = refreshed["content"]
                debug_log(f"聚焦文件已刷新: {path}")
        
        pending_modify = None
        modify_probe_buffer = ""
        if hasattr(web_terminal, "pending_modify_request"):
            web_terminal.pending_modify_request = None
        return result

    async def process_sub_agent_updates(
        messages: List[Dict],
        inline: bool = False,
        after_tool_call_id: Optional[str] = None
    ):
        """轮询子智能体任务并通知前端，并把结果插入当前对话上下文。"""
        manager = getattr(web_terminal, "sub_agent_manager", None)
        if not manager:
            return
        try:
            updates = manager.poll_updates()
            debug_log(f"[SubAgent] poll inline={inline} updates={len(updates)}")
        except Exception as exc:
            debug_log(f"子智能体状态检查失败: {exc}")
            return
        for update in updates:
            message = update.get("system_message")
            if not message:
                continue
            task_id = update.get("task_id")
            debug_log(f"[SubAgent] update task={task_id} inline={inline} msg={message}")
            web_terminal._record_sub_agent_message(message, task_id, inline=inline)
            debug_log(f"[SubAgent] recorded task={task_id}, 计算插入位置")

            insert_index = len(messages)
            if after_tool_call_id:
                for idx, msg in enumerate(messages):
                    if msg.get("role") == "tool" and msg.get("tool_call_id") == after_tool_call_id:
                        insert_index = idx + 1
                        break

            messages.insert(insert_index, {
                "role": "system",
                "content": message,
                "metadata": {"sub_agent_notice": True, "inline": inline, "task_id": task_id}
            })
            debug_log(f"[SubAgent] 插入系统消息位置: {insert_index}")
            sender('system_message', {
                'content': message,
                'inline': inline
            })
            maybe_mark_failure_from_message(web_terminal, message)
    
    for iteration in range(max_iterations):
        total_iterations += 1
        debug_log(f"\n--- 迭代 {iteration + 1}/{max_iterations} 开始 ---")
        
        # 检查是否超过总工具调用限制
        if 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
        append_break_triggered = False
        append_result = {"handled": False}
        modify_break_triggered = False
        modify_result = {"handled": False}
        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
            })
            return
        
        print(f"[API] 第{iteration + 1}次调用 (总工具调用: {total_tool_calls}/{MAX_TOTAL_TOOL_CALLS})")
        
        # 收集流式响应
        async for chunk in web_terminal.api_client.chat(messages, tools, stream=True):
            chunk_count += 1
            
            # 检查停止标志
            client_stop_info = stop_flags.get(client_sid)
            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:
                    debug_log(f"检测到停止请求，中断流处理")
                    if pending_append:
                        append_result = await finalize_pending_append(full_response, False, finish_reason="user_stop")
                        break
                    if pending_modify:
                        modify_result = await finalize_pending_modify(full_response, False, finish_reason="user_stop")
                        break
            
            # 先尝试记录 usage（有些平台会在最后一个 chunk 里携带 usage 但 choices 为空）
            usage_info = chunk.get("usage")
            if usage_info:
                last_usage_payload = usage_info

            if "choices" not in chunk:
                debug_log(f"Chunk {chunk_count}: 无choices字段")
                continue
            if not chunk.get("choices"):
                debug_log(f"Chunk {chunk_count}: choices为空列表")
                continue
            choice = chunk["choices"][0]
            if not usage_info and isinstance(choice, dict) and choice.get("usage"):
                # 兼容部分供应商将 usage 放在 choice 内的格式（例如部分 Kimi/Qwen 返回）
                last_usage_payload = choice.get("usage")
            delta = choice.get("delta", {})
            finish_reason = choice.get("finish_reason")
            if finish_reason:
                last_finish_reason = finish_reason
            
            # 处理思考内容
            if "reasoning_content" in delta:
                reasoning_content = delta["reasoning_content"]
                if reasoning_content:
                    reasoning_chunks += 1
                    debug_log(f"  思考内容 #{reasoning_chunks}: {len(reasoning_content)} 字符")
                    
                    if not thinking_started:
                        in_thinking = True
                        thinking_started = True
                        sender('thinking_start', {})
                        await asyncio.sleep(0.05)
                    
                    current_thinking += reasoning_content
                    sender('thinking_chunk', {'content': reasoning_content})
            
            # 处理正常内容
            if "content" in delta:
                content = delta["content"]
                if content:
                    content_chunks += 1
                    debug_log(f"  正式内容 #{content_chunks}: {repr(content[:100] if content else 'None')}")
                    
                    # 通过文本内容提前检测工具调用意图
                    if not detected_tools:
                        # 检测常见的工具调用模式
                        tool_patterns = [
                            (r'(创建|新建|生成).*(文件|file)', 'create_file'),
                            (r'(读取|查看|打开).*(文件|file)', 'read_file'),
                            (r'(修改|编辑|更新).*(文件|file)', 'write_file_diff'),
                            (r'(删除|移除).*(文件|file)', 'delete_file'),
                            (r'(搜索|查找|search)', 'web_search'),
                            (r'(执行|运行).*(Python|python|代码)', 'run_python'),
                            (r'(执行|运行).*(命令|command)', 'run_command'),
                            (r'(等待|sleep|延迟)', 'sleep'),
                            (r'(聚焦|focus).*(文件|file)', 'focus_file'),
                            (r'(终端|terminal|会话|session)', 'terminal_session'),
                        ]
                        
                        for pattern, tool_name in tool_patterns:
                            if re.search(pattern, content, re.IGNORECASE):
                                early_tool_id = f"early_{tool_name}_{time.time()}"
                                if early_tool_id not in detected_tools:
                                    sender('tool_hint', {
                                        'id': early_tool_id,
                                        'name': tool_name,
                                        'message': f'检测到可能需要调用 {tool_name}...',
                                        'confidence': 'low',
                                        'conversation_id': conversation_id
                                    })
                                    detected_tools[early_tool_id] = tool_name
                                    debug_log(f"    ⚡ 提前检测到工具意图: {tool_name}")
                                    break
                    
                    if in_thinking and not thinking_ended:
                        in_thinking = False
                        thinking_ended = True
                        sender('thinking_end', {'full_content': current_thinking})
                        await asyncio.sleep(0.1)
                        
                    
                    expecting_modify = bool(pending_modify) or bool(getattr(web_terminal, "pending_modify_request", None))
                    expecting_append = bool(pending_append) or bool(getattr(web_terminal, "pending_append_request", None))

                    if pending_modify:
                        if not pending_modify.get("start_seen"):
                            probe_buffer = pending_modify.get("probe_buffer", "") + content
                            if len(probe_buffer) > 10000:
                                probe_buffer = probe_buffer[-10000:]
                            marker = pending_modify.get("start_marker")
                            marker_index = probe_buffer.find(marker)
                            if marker_index == -1:
                                pending_modify["probe_buffer"] = probe_buffer
                                continue
                            after_marker = marker_index + len(marker)
                            remainder = probe_buffer[after_marker:]
                            pending_modify["buffer"] = remainder
                            pending_modify["raw_buffer"] = marker + remainder
                            pending_modify["start_seen"] = True
                            pending_modify["detected_blocks"] = set()
                            pending_modify["probe_buffer"] = ""
                            if pending_modify.get("display_id"):
                                sender('update_action', {
                                    'id': pending_modify["display_id"],
                                    'status': 'running',
                                    'preparing_id': pending_modify.get("tool_call_id"),
                                    'message': f"正在修改 {pending_modify['path']}..."
                                })
                        else:
                            pending_modify["buffer"] += content
                            pending_modify["raw_buffer"] += content
                        
                        if pending_modify.get("start_seen"):
                            block_text = pending_modify["buffer"]
                            for match in re.finditer(r"\[replace:(\d+)\]", block_text):
                                try:
                                    block_index = int(match.group(1))
                                except ValueError:
                                    continue
                                detected_blocks = pending_modify.setdefault("detected_blocks", set())
                                if block_index not in detected_blocks:
                                    detected_blocks.add(block_index)
                                    if pending_modify.get("display_id"):
                                        sender('update_action', {
                                            'id': pending_modify["display_id"],
                                            'status': 'running',
                                            'preparing_id': pending_modify.get("tool_call_id"),
                                            'message': f"正在对 {pending_modify['path']} 进行第 {block_index} 处修改..."
                                        })
                        
                        if pending_modify.get("start_seen"):
                            end_pos = pending_modify["buffer"].find(pending_modify["end_marker"])
                            if end_pos != -1:
                                pending_modify["end_index"] = end_pos
                                modify_break_triggered = True
                                debug_log("检测到<<<END_MODIFY>>>，即将终止流式输出并应用修改")
                                break
                        continue
                    elif expecting_modify:
                        modify_probe_buffer += content
                        if len(modify_probe_buffer) > 10000:
                            modify_probe_buffer = modify_probe_buffer[-10000:]
                        
                        marker_match = re.search(r"<<<MODIFY:\s*([\s\S]*?)>>>", modify_probe_buffer)
                        if marker_match:
                            detected_raw_path = marker_match.group(1)
                            detected_path = detected_raw_path.strip()
                            marker_full = marker_match.group(0)
                            after_marker_index = modify_probe_buffer.find(marker_full) + len(marker_full)
                            remainder = modify_probe_buffer[after_marker_index:]
                            modify_probe_buffer = ""
                            
                            if not detected_path:
                                debug_log("检测到 MODIFY 起始标记但路径为空，忽略。")
                                continue
                            
                            pending_modify = {
                                "path": detected_path,
                                "tool_call_id": None,
                                "buffer": remainder,
                                "raw_buffer": marker_full + remainder,
                                "start_marker": marker_full,
                                "end_marker": "<<<END_MODIFY>>>",
                                "start_seen": True,
                                "end_index": None,
                                "display_id": None,
                                "detected_blocks": set()
                            }
                            if hasattr(web_terminal, "pending_modify_request"):
                                web_terminal.pending_modify_request = {"path": detected_path}
                            debug_log(f"直接检测到modify起始标记，构建修改缓冲: {detected_path}")
                            
                            end_pos = pending_modify["buffer"].find(pending_modify["end_marker"])
                            if end_pos != -1:
                                pending_modify["end_index"] = end_pos
                                modify_break_triggered = True
                                debug_log("检测到<<<END_MODIFY>>>，即将终止流式输出并应用修改")
                                break
                        continue
                    
                    if pending_append:
                        pending_append["buffer"] += content
                        
                        if pending_append.get("content_start") is None:
                            marker_index = pending_append["buffer"].find(pending_append["start_marker"])
                            if marker_index != -1:
                                pending_append["content_start"] = marker_index + len(pending_append["start_marker"])
                                debug_log(f"检测到追加起始标识: {pending_append['start_marker']}")
                        
                        if pending_append.get("content_start") is not None:
                            end_index = pending_append["buffer"].find(
                                pending_append["end_marker"],
                                pending_append["content_start"]
                            )
                            if end_index != -1:
                                pending_append["end_index"] = end_index
                                append_break_triggered = True
                                debug_log("检测到<<<END_APPEND>>>，即将终止流式输出并写入文件")
                                break
                        
                        # 继续累积追加内容
                        continue
                    elif expecting_append:
                        append_probe_buffer += content
                        # 限制缓冲区大小防止过长
                        if len(append_probe_buffer) > 10000:
                            append_probe_buffer = append_probe_buffer[-10000:]

                        marker_match = re.search(r"<<<APPEND:\s*([\s\S]*?)>>>", append_probe_buffer)
                        if marker_match:
                            detected_raw_path = marker_match.group(1)
                            detected_path = detected_raw_path.strip()
                            if not detected_path:
                                append_probe_buffer = append_probe_buffer[marker_match.end():]
                                continue
                            marker_full = marker_match.group(0)
                            after_marker_index = append_probe_buffer.find(marker_full) + len(marker_full)
                            remainder = append_probe_buffer[after_marker_index:]
                            append_probe_buffer = ""
                            pending_append = {
                                "path": detected_path,
                                "tool_call_id": None,
                                "buffer": remainder,
                                "start_marker": marker_full,
                                "end_marker": "<<<END_APPEND>>>",
                                "content_start": 0,
                                "end_index": None,
                                "display_id": None
                            }
                            if hasattr(web_terminal, "pending_append_request"):
                                web_terminal.pending_append_request = {"path": detected_path}
                            debug_log(f"直接检测到append起始标记，构建追加缓冲: {detected_path}")
                            # 检查是否立即包含结束标记
                            if pending_append["buffer"]:
                                end_index = pending_append["buffer"].find(pending_append["end_marker"], pending_append["content_start"])
                                if end_index != -1:
                                    pending_append["end_index"] = end_index
                                    append_break_triggered = True
                                    debug_log("检测到<<<END_APPEND>>>，即将终止流式输出并写入文件")
                                    break
                            continue
                    
                    if not text_started:
                        text_started = True
                        text_streaming = True
                        sender('text_start', {})
                        brief_log("模型输出了内容")
                        await asyncio.sleep(0.05)
                    
                    if not pending_append:
                        full_response += content
                        accumulated_response += content
                        text_has_content = True
                        emit_time = time.time()
                        elapsed = 0.0 if last_text_chunk_time is None else emit_time - last_text_chunk_time
                        last_text_chunk_time = emit_time
                        text_chunk_index += 1
                        log_backend_chunk(
                            conversation_id,
                            iteration + 1,
                            text_chunk_index,
                            elapsed,
                            len(content),
                            content[:32]
                        )
                        sender('text_chunk', {
                            'content': content,
                            'index': text_chunk_index,
                            'elapsed': elapsed
                        })
            
            # 收集工具调用 - 实时发送准备状态
            if "tool_calls" in delta:
                tool_chunks += 1
                for tc in delta["tool_calls"]:
                    found = False
                    for existing in tool_calls:
                        if existing.get("index") == tc.get("index"):
                            if "function" in tc and "arguments" in tc["function"]:
                                arg_chunk = tc["function"]["arguments"]
                                existing_fn = existing.get("function", {})
                                existing_args = existing_fn.get("arguments", "")
                                existing_fn["arguments"] = (existing_args or "") + arg_chunk
                                existing["function"] = existing_fn

                                combined_args = existing_fn.get("arguments", "")
                                tool_id = existing.get("id") or tc.get("id")
                                tool_name = (
                                    existing_fn.get("name")
                                    or tc.get("function", {}).get("name", "")
                                )
                                intent_value = extract_intent_from_partial(combined_args)
                                if (
                                    intent_value
                                    and tool_id
                                    and detected_tool_intent.get(tool_id) != intent_value
                                ):
                                    detected_tool_intent[tool_id] = intent_value
                                    brief_log(f"[intent] 增量提取 {tool_name}: {intent_value}")
                                    sender('tool_intent', {
                                        'id': tool_id,
                                        'name': tool_name,
                                        'intent': intent_value,
                                        'conversation_id': conversation_id
                                    })
                                    debug_log(f"    发送工具意图: {tool_name} -> {intent_value}")
                                    await asyncio.sleep(0.01)
                            found = True
                            break
                    
                    if not found and tc.get("id"):
                        tool_id = tc["id"]
                        tool_name = tc.get("function", {}).get("name", "")
                        arguments_str = tc.get("function", {}).get("arguments", "") or ""
                        
                        # 新工具检测到，立即发送准备事件
                        if tool_id not in detected_tools and tool_name:
                            detected_tools[tool_id] = tool_name
                            
                            # 尝试提前提取 intent
                            intent_value = None
                            if arguments_str:
                                intent_value = extract_intent_from_partial(arguments_str)
                                if intent_value:
                                    detected_tool_intent[tool_id] = intent_value
                                    brief_log(f"[intent] 预提取 {tool_name}: {intent_value}")
                            
                            # 立即发送工具准备中事件
                            brief_log(f"[tool] 准备调用 {tool_name} (id={tool_id}) intent={intent_value or '-'}")
                            sender('tool_preparing', {
                                'id': tool_id,
                                'name': tool_name,
                                'message': f'准备调用 {tool_name}...',
                                'intent': intent_value,
                                'conversation_id': conversation_id
                            })
                            debug_log(f"    发送工具准备事件: {tool_name}")
                            await asyncio.sleep(0.1)
                        
                        tool_calls.append({
                            "id": tool_id,
                            "index": tc.get("index"),
                            "type": "function",
                            "function": {
                                "name": tool_name,
                                "arguments": arguments_str
                            }
                        })
                        # 尝试从增量参数中抽取 intent，并单独推送
                        if tool_id and arguments_str:
                            intent_value = extract_intent_from_partial(arguments_str)
                            if intent_value and detected_tool_intent.get(tool_id) != intent_value:
                                detected_tool_intent[tool_id] = intent_value
                                sender('tool_intent', {
                                    'id': tool_id,
                                    'name': tool_name,
                                    'intent': intent_value,
                                    'conversation_id': conversation_id
                                })
                                debug_log(f"    发送工具意图: {tool_name} -> {intent_value}")
                                await asyncio.sleep(0.01)
                        debug_log(f"    新工具: {tool_name}")
        
        # 检查是否被停止
        client_stop_info = stop_flags.get(client_sid)
        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:
                debug_log("任务在流处理完成后检测到停止状态")
                return
        
        # === API响应完成后只计算输出token ===
        if last_usage_payload:
            try:
                web_terminal.context_manager.apply_usage_statistics(last_usage_payload)
                debug_log(
                    f"Usage统计: prompt={last_usage_payload.get('prompt_tokens', 0)}, "
                    f"completion={last_usage_payload.get('completion_tokens', 0)}, "
                    f"total={last_usage_payload.get('total_tokens', 0)}"
                )
            except Exception as e:
                debug_log(f"Usage统计更新失败: {e}")
        else:
            debug_log("未获取到usage字段，跳过token统计更新")
        
        # 流结束后的处理
        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 not append_result["handled"] and pending_append:
            append_result = await finalize_pending_append(full_response, True, finish_reason=last_finish_reason)
        if not modify_result["handled"] and pending_modify:
            modify_result = await finalize_pending_modify(full_response, True, finish_reason=last_finish_reason)
        
        # 结束未完成的流
        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 and not append_result["handled"] and not modify_result["handled"]:
            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 append_result["handled"]:
            append_metadata = append_result.get("assistant_metadata")
            append_content_text = append_result.get("assistant_content")
            if append_content_text:
                web_terminal.context_manager.add_conversation(
                    "assistant",
                    append_content_text,
                    metadata=append_metadata
                )
                debug_log("💾 增量保存：追加正文快照")
                
                payload_info = append_metadata.get("append_payload") if append_metadata else {}
                sender('append_payload', {
                    'path': payload_info.get("path") or append_result.get("path"),
                    'forced': payload_info.get("forced", False),
                    'lines': payload_info.get("lines"),
                    'bytes': payload_info.get("bytes"),
                    'tool_call_id': payload_info.get("tool_call_id") or append_result.get("tool_call_id"),
                    'success': payload_info.get("success", append_result.get("success", False)),
                    'conversation_id': conversation_id
                })
            
            if append_result["tool_content"]:
                tool_call_id = append_result.get("tool_call_id") or f"append_{int(time.time() * 1000)}"
                system_notice = format_tool_result_notice("append_to_file", tool_call_id, append_result["tool_content"])
                web_terminal.context_manager.add_conversation("system", system_notice)
                append_result["tool_call_id"] = tool_call_id
                debug_log("💾 增量保存：append_to_file 工具结果（system 通知）")
            
            finish_reason = append_result.get("finish_reason")
            path_for_prompt = append_result.get("path")
            need_follow_prompt = (
                finish_reason == "length" or
                append_result.get("forced") or
                not append_result.get("success")
            )

            if need_follow_prompt and path_for_prompt:
                prompt_lines = [
                    f"append_to_file 在处理 {path_for_prompt} 时未完成，需要重新发起写入。"
                ]
                if finish_reason == "length":
                    prompt_lines.append(
                        "上一次输出达到系统单次输出上限，已写入的内容已保存。"
                    )
                if append_result.get("forced"):
                    prompt_lines.append(
                        "收到的内容缺少 <<<END_APPEND>>> 标记，系统依据流式结束位置落盘。"
                    )
                if not append_result.get("success"):
                    prompt_lines.append("系统未能识别有效的追加标记。")
                prompt_lines.append(
                    "请再次调用 append_to_file 工具获取新的写入窗口，并在工具调用的输出中遵循以下格式："
                )
                prompt_lines.append(f"<<<APPEND:{path_for_prompt}>>>")
                prompt_lines.append("...填写剩余正文，如内容已完成可留空...")
                prompt_lines.append("<<<END_APPEND>>>")
                prompt_lines.append("不要在普通回复中粘贴上述标记，必须通过 append_to_file 工具发送。")
                follow_prompt = "\n".join(prompt_lines)
                messages.append({
                    "role": "system",
                    "content": follow_prompt
                })
                web_terminal.context_manager.add_conversation("system", follow_prompt)
                debug_log("已注入追加任务提示")

        if append_result["handled"] and append_result.get("forced") and append_result.get("success"):
            mark_force_thinking(web_terminal, reason="append_forced_finish")
        if append_result["handled"] and not append_result.get("success"):
            sender('system_message', {
                'content': f'⚠️ 追加写入失败：{append_result.get("error")}'
            })
            maybe_mark_failure_from_message(web_terminal, f'⚠️ 追加写入失败：{append_result.get("error")}')
            mark_force_thinking(web_terminal, reason="append_failed")
        
        if modify_result["handled"]:
            modify_metadata = modify_result.get("assistant_metadata")
            modify_content_text = modify_result.get("assistant_content")
            if modify_content_text:
                web_terminal.context_manager.add_conversation(
                    "assistant",
                    modify_content_text,
                    metadata=modify_metadata
                )
                debug_log("💾 增量保存：修改正文快照")
                
                payload_info = modify_metadata.get("modify_payload") if modify_metadata else {}
                sender('modify_payload', {
                    'path': payload_info.get("path") or modify_result.get("path"),
                    'total': payload_info.get("total_blocks") or modify_result.get("total_blocks"),
                    'completed': payload_info.get("completed") or modify_result.get("completed_blocks"),
                    'failed': payload_info.get("failed") or modify_result.get("failed_blocks"),
                    'forced': payload_info.get("forced", modify_result.get("forced", False)),
                    'success': modify_result.get("success", False),
                    'conversation_id': conversation_id
                })
            
            if modify_result["tool_content"]:
                tool_call_id = modify_result.get("tool_call_id") or f"modify_{int(time.time() * 1000)}"
                system_notice = format_tool_result_notice("modify_file", tool_call_id, modify_result["tool_content"])
                web_terminal.context_manager.add_conversation("system", system_notice)
                modify_result["tool_call_id"] = tool_call_id
                debug_log("💾 增量保存：modify_file 工具结果（system 通知）")
            
            path_for_prompt = modify_result.get("path")
            failed_blocks = modify_result.get("failed_blocks") or []
            need_follow_prompt = modify_result.get("forced") or bool(failed_blocks)
            
            if need_follow_prompt and path_for_prompt:
                prompt_lines = [
                    f"modify_file 在处理 {path_for_prompt} 时未完成，需要重新发起补丁。"
                ]
                if modify_result.get("forced"):
                    prompt_lines.append(
                        "刚才的内容缺少 <<<END_MODIFY>>> 标记，系统仅应用了已识别的部分。"
                    )
                if failed_blocks:
                    failed_text = "、".join(str(idx) for idx in failed_blocks)
                    prompt_lines.append(f"以下补丁未成功：第 {failed_text} 处。")
                prompt_lines.append(
                    "请再次调用 modify_file 工具，并在新的工具调用中按以下模板提供完整补丁："
                )
                prompt_lines.append(f"<<<MODIFY:{path_for_prompt}>>>")
                prompt_lines.append("[replace:序号]")
                prompt_lines.append("<<OLD>>")
                prompt_lines.append("...原文（必须逐字匹配，包含全部缩进、空格和换行）...")
                prompt_lines.append("<<END>>")
                prompt_lines.append("<<NEW>>")
                prompt_lines.append("...新内容，可留空表示清空，注意保持结构完整...")
                prompt_lines.append("<<END>>")
                prompt_lines.append("[/replace]")
                prompt_lines.append("<<<END_MODIFY>>>")
                prompt_lines.append("请勿在普通回复中直接粘贴补丁，必须通过 modify_file 工具发送。")
                follow_prompt = "\n".join(prompt_lines)
                messages.append({
                    "role": "system",
                    "content": follow_prompt
                })
                web_terminal.context_manager.add_conversation("system", follow_prompt)
                debug_log("已注入修改任务提示")

            if modify_result["handled"] and modify_result.get("failed_blocks"):
                mark_force_thinking(web_terminal, reason="modify_partial_failure")
            if modify_result["handled"] and modify_result.get("forced") and modify_result.get("success"):
                mark_force_thinking(web_terminal, reason="modify_forced_finish")
            if modify_result["handled"] and not modify_result.get("success"):
                error_message = modify_result.get("summary_message") or modify_result.get("error") or "修改操作未成功，请根据提示重新执行。"
                sender('system_message', {
                    'content': f'⚠️ 修改操作存在未完成的内容：{error_message}'
                })
                maybe_mark_failure_from_message(web_terminal, f'⚠️ 修改操作存在未完成的内容：{error_message}')
                mark_force_thinking(web_terminal, reason="modify_failed")
            
        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 and not append_result["handled"] and not modify_result["handled"]:
            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)
        elif append_result["handled"] and append_result["assistant_content"]:
            assistant_content_parts.append(append_result["assistant_content"])
        elif modify_result["handled"] and modify_result.get("assistant_content"):
            assistant_content_parts.append(modify_result["assistant_content"])
        
        assistant_content = "\n".join(assistant_content_parts) if assistant_content_parts else ""
        
        # 添加到消息历史（用于API继续对话，不保存到文件）
        assistant_message = {
            "role": "assistant",
            "content": assistant_content,
            "tool_calls": tool_calls
        }
        if current_thinking:
            assistant_message["reasoning_content"] = current_thinking
        
        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 None
            )
        
        # 为下一轮迭代重置流状态标志，但保留 full_response 供上面保存使用
        text_streaming = False
        text_started = False
        text_has_content = False
        full_response = ""
        
        if append_result["handled"] and append_result.get("tool_content"):
            tool_call_id = append_result.get("tool_call_id") or f"append_{int(time.time() * 1000)}"
            system_notice = format_tool_result_notice("append_to_file", tool_call_id, append_result["tool_content"])
            messages.append({
                "role": "system",
                "content": system_notice
            })
            append_result["tool_call_id"] = tool_call_id
            debug_log("已将 append_to_file 工具结果以 system 形式追加到对话上下文")
        if modify_result["handled"] and modify_result.get("tool_content"):
            tool_call_id = modify_result.get("tool_call_id") or f"modify_{int(time.time() * 1000)}"
            system_notice = format_tool_result_notice("modify_file", tool_call_id, modify_result["tool_content"])
            messages.append({
                "role": "system",
                "content": system_notice
            })
            modify_result["tool_call_id"] = tool_call_id
            debug_log("已将 modify_file 工具结果以 system 形式追加到对话上下文")
        
        force_continue = append_result["handled"] or modify_result["handled"]
        if force_continue:
            if append_result["handled"]:
                debug_log("append_to_file 已处理，继续下一轮以让模型返回确认回复")
            elif modify_result["handled"]:
                debug_log("modify_file 已处理，继续下一轮以让模型返回确认回复")
            else:
                debug_log("补丁处理完成，继续下一轮以获取模型回复")
            continue

        if not tool_calls:
            debug_log("没有工具调用，结束迭代")
            break
        
        # 检查连续相同工具调用
        for tc in tool_calls:
            tool_name = tc["function"]["name"]
            
            if tool_name == last_tool_name:
                consecutive_same_tool[tool_name] += 1
                
                if 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)
        
        # 执行每个工具
        for tool_call in tool_calls:
            # 检查停止标志
            client_stop_info = stop_flags.get(client_sid)
            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:
                    debug_log("在工具调用过程中检测到停止状态") 
                    return
            
            # 工具调用间隔控制
            current_time = time.time()
            if last_tool_call_time > 0:
                elapsed = current_time - last_tool_call_time
                if elapsed < TOOL_CALL_COOLDOWN:
                    await asyncio.sleep(TOOL_CALL_COOLDOWN - elapsed)
            last_tool_call_time = time.time()
            
            function_name = tool_call["function"]["name"]
            arguments_str = tool_call["function"]["arguments"]
            tool_call_id = tool_call["id"]
            

            debug_log(f"准备解析JSON，工具: {function_name}, 参数长度: {len(arguments_str)}")
            debug_log(f"JSON参数前200字符: {arguments_str[:200]}")
            debug_log(f"JSON参数后200字符: {arguments_str[-200:]}")
            
            # 使用改进的参数解析方法
            if hasattr(web_terminal, 'api_client') and hasattr(web_terminal.api_client, '_safe_tool_arguments_parse'):
                success, arguments, error_msg = web_terminal.api_client._safe_tool_arguments_parse(arguments_str, function_name)
                if not success:
                    debug_log(f"安全解析失败: {error_msg}")
                    error_text = f'工具参数解析失败: {error_msg}'
                    error_payload = {
                        "success": False,
                        "error": error_text,
                        "error_type": "parameter_format_error",
                        "tool_name": function_name,
                        "tool_call_id": tool_call_id,
                        "message": error_text
                    }
                    sender('error', {'message': error_text})
                    sender('update_action', {
                        'preparing_id': tool_call_id,
                        'status': 'completed',
                        'result': error_payload,
                        'message': error_text
                    })
                    error_content = json.dumps(error_payload, ensure_ascii=False)
                    web_terminal.context_manager.add_conversation(
                        "tool",
                        error_content,
                        tool_call_id=tool_call_id,
                        name=function_name
                    )
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call_id,
                        "name": function_name,
                        "content": error_content
                    })
                    continue
                debug_log(f"使用安全解析成功，参数键: {list(arguments.keys())}")
            else:
                # 回退到带有基本修复逻辑的解析
                try:
                    arguments = json.loads(arguments_str) if arguments_str.strip() else {}
                    debug_log(f"直接JSON解析成功，参数键: {list(arguments.keys())}")
                except json.JSONDecodeError as e:
                    debug_log(f"原始JSON解析失败: {e}")
                    # 尝试基本的JSON修复
                    repaired_str = arguments_str.strip()
                    repair_attempts = []
                    
                    # 修复1: 未闭合字符串
                    if repaired_str.count('"') % 2 == 1:
                        repaired_str += '"'
                        repair_attempts.append("添加闭合引号")
                    
                    # 修复2: 未闭合JSON对象
                    if repaired_str.startswith('{') and not repaired_str.rstrip().endswith('}'):
                        repaired_str = repaired_str.rstrip() + '}'
                        repair_attempts.append("添加闭合括号")
                    
                    # 修复3: 截断的JSON（移除不完整的最后一个键值对）
                    if not repair_attempts:  # 如果前面的修复都没用上
                        last_comma = repaired_str.rfind(',')
                        if last_comma > 0:
                            repaired_str = repaired_str[:last_comma] + '}'
                            repair_attempts.append("移除不完整的键值对")
                    
                    # 尝试解析修复后的JSON
                    try:
                        arguments = json.loads(repaired_str)
                        debug_log(f"JSON修复成功: {', '.join(repair_attempts)}")
                        debug_log(f"修复后参数键: {list(arguments.keys())}")
                    except json.JSONDecodeError as repair_error:
                        debug_log(f"JSON修复也失败: {repair_error}")
                        debug_log(f"修复尝试: {repair_attempts}")
                        debug_log(f"修复后内容前100字符: {repaired_str[:100]}")
                        error_text = f'工具参数解析失败: {e}'
                        error_payload = {
                            "success": False,
                            "error": error_text,
                            "error_type": "parameter_format_error",
                            "tool_name": function_name,
                            "tool_call_id": tool_call_id,
                            "message": error_text
                        }
                        sender('error', {'message': error_text})
                        sender('update_action', {
                            'preparing_id': tool_call_id,
                            'status': 'completed',
                            'result': error_payload,
                            'message': error_text
                        })
                        error_content = json.dumps(error_payload, ensure_ascii=False)
                        web_terminal.context_manager.add_conversation(
                            "tool",
                            error_content,
                            tool_call_id=tool_call_id,
                            name=function_name
                        )
                        messages.append({
                            "role": "tool",
                            "tool_call_id": tool_call_id,
                            "name": function_name,
                            "content": error_content
                        })
                        continue            
            
            debug_log(f"执行工具: {function_name} (ID: {tool_call_id})")
            
            # 发送工具开始事件
            tool_display_id = f"tool_{iteration}_{function_name}_{time.time()}"
            monitor_snapshot = None
            snapshot_path = None
            memory_snapshot_type = None
            if function_name in MONITOR_FILE_TOOLS:
                snapshot_path = resolve_monitor_path(arguments)
                monitor_snapshot = capture_monitor_snapshot(snapshot_path)
                if monitor_snapshot:
                    cache_monitor_snapshot(tool_display_id, 'before', monitor_snapshot)
            elif function_name in MONITOR_MEMORY_TOOLS:
                memory_snapshot_type = (arguments.get('memory_type') or 'main').lower()
                before_entries = None
                try:
                    before_entries = resolve_monitor_memory(web_terminal.memory_manager._read_entries(memory_snapshot_type))
                except Exception as exc:
                    debug_log(f"[MonitorSnapshot] 读取记忆失败: {memory_snapshot_type} ({exc})")
                if before_entries is not None:
                    monitor_snapshot = {
                        'memory_type': memory_snapshot_type,
                        'entries': before_entries
                    }
                    cache_monitor_snapshot(tool_display_id, 'before', monitor_snapshot)
            
            sender('tool_start', {
                'id': tool_display_id,
                'name': function_name,
                'arguments': arguments,
                'preparing_id': tool_call_id,
                'monitor_snapshot': monitor_snapshot,
                'conversation_id': conversation_id
            })
            brief_log(f"调用了工具: {function_name}")
            
            await asyncio.sleep(0.3)
            start_time = time.time()
            
            # 执行工具
            tool_result = await web_terminal.handle_tool_call(function_name, arguments)
            debug_log(f"工具结果: {tool_result[:200]}...")
            
            execution_time = time.time() - start_time
            if execution_time < 1.5:
                await asyncio.sleep(1.5 - execution_time)
            
            # 更新工具状态
            result_data = {}
            try:
                result_data = json.loads(tool_result)
            except:
                result_data = {'output': tool_result}
            tool_failed = detect_tool_failure(result_data)

            action_status = 'completed'
            action_message = None
            awaiting_flag = False
            
            if function_name == "write_file_diff":
                diff_path = result_data.get("path") or arguments.get("path")
                summary = result_data.get("summary") or result_data.get("message")
                if summary:
                    action_message = summary
                debug_log(f"write_file_diff 执行完成: {summary or '无摘要'}")
            
            if function_name == "wait_sub_agent":
                system_msg = result_data.get("system_message")
                if system_msg:
                    messages.append({
                        "role": "system",
                        "content": system_msg
                    })
                    sender('system_message', {
                        'content': system_msg,
                        'inline': False
                    })
                    maybe_mark_failure_from_message(web_terminal, system_msg)
            monitor_snapshot_after = None
            if function_name in MONITOR_FILE_TOOLS:
                result_path = None
                if isinstance(result_data, dict):
                    result_path = resolve_monitor_path(result_data)
                    if not result_path:
                        candidate_path = result_data.get('path')
                        if isinstance(candidate_path, str) and candidate_path.strip():
                            result_path = candidate_path.strip()
                if not result_path:
                    result_path = resolve_monitor_path(arguments, snapshot_path) or snapshot_path
                monitor_snapshot_after = capture_monitor_snapshot(result_path)
            elif function_name in MONITOR_MEMORY_TOOLS:
                memory_after_type = str(
                    arguments.get('memory_type')
                    or (isinstance(result_data, dict) and result_data.get('memory_type'))
                    or memory_snapshot_type
                    or 'main'
                ).lower()
                after_entries = None
                try:
                    after_entries = resolve_monitor_memory(web_terminal.memory_manager._read_entries(memory_after_type))
                except Exception as exc:
                    debug_log(f"[MonitorSnapshot] 读取记忆失败(after): {memory_after_type} ({exc})")
                if after_entries is not None:
                    monitor_snapshot_after = {
                        'memory_type': memory_after_type,
                        'entries': after_entries
                    }
            
            update_payload = {
                'id': tool_display_id,
                'status': action_status,
                'result': result_data,
                'preparing_id': tool_call_id,
                'conversation_id': conversation_id
            }
            if action_message:
                update_payload['message'] = action_message
            if awaiting_flag:
                update_payload['awaiting_content'] = True
            if monitor_snapshot_after:
                update_payload['monitor_snapshot_after'] = monitor_snapshot_after
                cache_monitor_snapshot(tool_display_id, 'after', monitor_snapshot_after)
            
            sender('update_action', update_payload)
            
            # 更新UI状态
            if function_name in ['focus_file', 'unfocus_file', 'write_file_diff']:
                sender('focused_files_update', web_terminal.get_focused_files_info())
            
            if function_name in ['create_file', 'delete_file', 'rename_file', 'create_folder']:
                structure = web_terminal.context_manager.get_project_structure()
                sender('file_tree_update', structure)
            
            # ===== 增量保存：立即保存工具结果 =====
            metadata_payload = None
            if isinstance(result_data, dict):
                # 特殊处理 web_search：保留可供前端渲染的精简结构，以便历史记录复现搜索结果
                if function_name == "web_search":
                    try:
                        tool_result_content = json.dumps(compact_web_search_result(result_data), ensure_ascii=False)
                    except Exception:
                        tool_result_content = tool_result
                else:
                    tool_result_content = format_tool_result_for_context(function_name, result_data, tool_result)
                metadata_payload = {"tool_payload": result_data}
            else:
                tool_result_content = tool_result
            
            # 立即保存工具结果
            web_terminal.context_manager.add_conversation(
                "tool",
                tool_result_content,
                tool_call_id=tool_call_id,
                name=function_name,
                metadata=metadata_payload
            )
            debug_log(f"💾 增量保存：工具结果 {function_name}")
            system_message = result_data.get("system_message") if isinstance(result_data, dict) else None
            if system_message:
                web_terminal._record_sub_agent_message(system_message, result_data.get("task_id"), inline=False)
                maybe_mark_failure_from_message(web_terminal, system_message)
            todo_note = result_data.get("system_note") if isinstance(result_data, dict) else None
            if todo_note:
                web_terminal.context_manager.add_conversation("system", todo_note)
                maybe_mark_failure_from_message(web_terminal, todo_note)
            
            # 添加到消息历史（用于API继续对话）
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call_id,
                "name": function_name,
                "content": tool_result_content
            })

            # 处理图片注入：必须紧跟在对应的 tool 消息之后，且工具成功时才插入
            if (
                function_name == "view_image"
                and getattr(web_terminal, "pending_image_view", None)
                and not tool_failed
                and (isinstance(result_data, dict) and result_data.get("success") is not False)
            ):
                inj = web_terminal.pending_image_view
                web_terminal.pending_image_view = None
                injected_text = "这是一条系统控制发送的信息，并非用户主动发送，目的是返回你需要查看的图片。"
                # 记录到对话历史
                web_terminal.context_manager.add_conversation(
                    "user",
                    injected_text,
                    images=[inj["path"]],
                    metadata={"system_injected_image": True}
                )
                # 同步到当前消息列表（直接带多模态 content），保证顺序为 tool_call -> tool -> (系统代发)user
                content_payload = web_terminal.context_manager._build_content_with_images(
                    injected_text,
                    [inj["path"]]
                )
                messages.append({
                    "role": "user",
                    "content": content_payload,
                    "metadata": {"system_injected_image": True}
                })
                # 提示前端
                sender('system_message', {
                    'content': f'系统已按模型请求插入图片: {inj.get("path")}'
                })

            if function_name != 'write_file_diff':
                await process_sub_agent_updates(messages, inline=True, after_tool_call_id=tool_call_id)
            
            await asyncio.sleep(0.2)
            
            if tool_failed:
                mark_force_thinking(web_terminal, reason=f"{function_name}_failed")
        
        # 标记不再是第一次迭代
        is_first_iteration = 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")
    
    # 发送完成事件
    sender('task_complete', {
        'total_iterations': total_iterations,
        'total_tool_calls': total_tool_calls,
        'auto_fix_attempts': auto_fix_attempts
    })
    
@socketio.on('send_command')
def handle_command(data):
    """处理系统命令"""
    command = data.get('command', '')
    
    username, terminal, _ = get_terminal_for_sid(request.sid)
    if not terminal:
        emit('error', {'message': 'System not initialized'})
        return
    record_user_activity(username)
    
    if command.startswith('/'):
        command = command[1:]
    
    parts = command.split(maxsplit=1)
    cmd = parts[0].lower()
    
    if cmd == "clear":
        terminal.context_manager.conversation_history.clear()
        if terminal.thinking_mode:
            terminal.api_client.start_new_task(force_deep=terminal.deep_thinking_mode)
        emit('command_result', {
            'command': cmd,
            'success': True,
            'message': '对话已清除'
        })
    elif cmd == "status":
        status = terminal.get_status()
        # 添加终端状态
        if terminal.terminal_manager:
            terminal_status = terminal.terminal_manager.list_terminals()
            status['terminals'] = terminal_status
        emit('command_result', {
            'command': cmd,
            'success': True,
            'data': status
        })
    elif cmd == "terminals":
        # 列出终端会话
        if terminal.terminal_manager:
            result = terminal.terminal_manager.list_terminals()
            emit('command_result', {
                'command': cmd,
                'success': True,
                'data': result
            })
        else:
            emit('command_result', {
                'command': cmd,
                'success': False,
                'message': '终端系统未初始化'
            })
    else:
        emit('command_result', {
            'command': cmd,
            'success': False,
            'message': f'未知命令: {cmd}'
        })

@app.route('/api/conversations/<conversation_id>/token-statistics', methods=['GET'])
@api_login_required
@with_terminal
def get_conversation_token_statistics(conversation_id, terminal: WebTerminal, workspace: UserWorkspace, username: str):
    """获取特定对话的token统计"""
    try:
        stats = terminal.context_manager.get_conversation_token_statistics(conversation_id)
        
        if stats:
            return jsonify({
                "success": True,
                "data": stats
            })
        else:
            return jsonify({
                "success": False,
                "error": "Conversation not found",
                "message": f"对话 {conversation_id} 不存在"
            }), 404
            
    except Exception as e:
        print(f"[API] 获取token统计错误: {e}")
        return jsonify({
            "success": False,
            "error": str(e),
            "message": "获取token统计时发生异常"
        }), 500


@app.route('/api/conversations/<conversation_id>/tokens', methods=['GET'])
@api_login_required
@with_terminal
def get_conversation_tokens(conversation_id, terminal: WebTerminal, workspace: UserWorkspace, username: str):
    """获取对话的当前完整上下文token数（包含所有动态内容）"""
    try:
        current_tokens = terminal.context_manager.get_current_context_tokens(conversation_id)
        return jsonify({
            "success": True,
            "data": {
                "total_tokens": current_tokens
            }
        })
    except Exception as e:
        return jsonify({
            "success": False,
            "error": str(e)
        }), 500

def calculate_directory_size(root: Path) -> int:
    if not root.exists():
        return 0
    total = 0
    stack = [root]
    while stack:
        current = stack.pop()
        try:
            with os.scandir(current) as iterator:
                for entry in iterator:
                    try:
                        if entry.is_symlink():
                            continue
                        if entry.is_file(follow_symlinks=False):
                            total += entry.stat(follow_symlinks=False).st_size
                        elif entry.is_dir(follow_symlinks=False):
                            stack.append(Path(entry.path))
                    except (OSError, FileNotFoundError, PermissionError):
                        continue
        except (NotADirectoryError, FileNotFoundError, PermissionError, OSError):
            continue
    return total


def iso_datetime_from_epoch(epoch: Optional[float]) -> Optional[str]:
    if not epoch:
        return None
    try:
        return datetime.utcfromtimestamp(epoch).replace(microsecond=0).isoformat() + "Z"
    except (ValueError, OSError):
        return None


def compute_workspace_storage(workspace: UserWorkspace) -> Dict[str, Any]:
    project_bytes = calculate_directory_size(workspace.project_path)
    data_bytes = calculate_directory_size(workspace.data_dir)
    logs_bytes = calculate_directory_size(workspace.logs_dir)
    quarantine_bytes = calculate_directory_size(workspace.quarantine_dir)
    uploads_bytes = calculate_directory_size(workspace.uploads_dir)
    backups_bytes = calculate_directory_size(workspace.data_dir / "backups")
    usage_percent = None
    if PROJECT_MAX_STORAGE_BYTES:
        usage_percent = round(project_bytes / PROJECT_MAX_STORAGE_BYTES * 100, 2) if project_bytes else 0.0
    status = "ok"
    if usage_percent is not None:
        if usage_percent >= 95:
            status = "critical"
        elif usage_percent >= 80:
            status = "warning"
    return {
        "project_bytes": project_bytes,
        "data_bytes": data_bytes,
        "logs_bytes": logs_bytes,
        "quarantine_bytes": quarantine_bytes,
        "uploads_bytes": uploads_bytes,
        "backups_bytes": backups_bytes,
        "total_bytes": project_bytes + data_bytes + logs_bytes + quarantine_bytes,
        "limit_bytes": PROJECT_MAX_STORAGE_BYTES,
        "usage_percent": usage_percent,
        "status": status,
    }


def collect_usage_snapshot(username: str, workspace: UserWorkspace, role: Optional[str]) -> Dict[str, Any]:
    tracker = get_or_create_usage_tracker(username, workspace)
    stats = tracker.get_stats()
    quotas = stats.get("quotas") or {}
    windows = stats.get("windows") or {}
    snapshot: Dict[str, Any] = {}
    for metric in ("fast", "thinking", "search"):
        window_meta = windows.get(metric) or {}
        quota_meta = quotas.get(metric) or {}
        default_limit = QUOTA_DEFAULTS.get("default", {}).get(metric, {}).get("limit", 0)
        snapshot[metric] = {
            "count": int(window_meta.get("count", 0) or 0),
            "window_start": window_meta.get("window_start"),
            "reset_at": window_meta.get("reset_at") or quota_meta.get("reset_at"),
            "limit": quota_meta.get("limit", default_limit),
        }
    snapshot["role"] = role or quotas.get("role") or "user"
    return snapshot


def _read_token_totals_file(workspace: UserWorkspace) -> Dict[str, int]:
    path = workspace.data_dir / "token_totals.json"
    if not path.exists():
        return {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0}
    try:
        with open(path, 'r', encoding='utf-8') as fh:
            payload = json.load(fh) or {}
        input_tokens = int(payload.get("input_tokens") or payload.get("total_input_tokens") or 0)
        output_tokens = int(payload.get("output_tokens") or payload.get("total_output_tokens") or 0)
        total_tokens = int(payload.get("total_tokens") or (input_tokens + output_tokens))
        return {
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "total_tokens": total_tokens,
        }
    except (OSError, json.JSONDecodeError, ValueError) as exc:
        print(f"[admin] 解析 token_totals.json 失败 ({workspace.username}): {exc}")
        return {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0}


def _collect_conversation_token_totals(workspace: UserWorkspace) -> Dict[str, int]:
    try:
        manager = ConversationManager(base_dir=workspace.data_dir)
        stats = manager.get_statistics() or {}
        token_stats = stats.get("token_statistics") or {}
        input_tokens = int(token_stats.get("total_input_tokens") or 0)
        output_tokens = int(token_stats.get("total_output_tokens") or 0)
        total_tokens = int(token_stats.get("total_tokens") or (input_tokens + output_tokens))
        return {
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "total_tokens": total_tokens,
        }
    except Exception as exc:
        print(f"[admin] 读取 legacy token 统计失败 ({workspace.username}): {exc}")
        return {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0}


def collect_user_token_statistics(workspace: UserWorkspace) -> Dict[str, int]:
    """汇总单个用户在所有对话中的token累计数据。"""
    file_totals = _read_token_totals_file(workspace)
    legacy_totals = _collect_conversation_token_totals(workspace)
    return {
        "input_tokens": max(file_totals["input_tokens"], legacy_totals["input_tokens"]),
        "output_tokens": max(file_totals["output_tokens"], legacy_totals["output_tokens"]),
        "total_tokens": max(file_totals["total_tokens"], legacy_totals["total_tokens"]),
    }


def compute_usage_leaders(users: List[Dict[str, Any]], metric: str, top_n: int = 5) -> List[Dict[str, Any]]:
    ranked = sorted(
        (
            {
                "username": entry["username"],
                "count": entry.get("usage", {}).get(metric, {}).get("count", 0),
                "limit": entry.get("usage", {}).get(metric, {}).get("limit"),
            }
            for entry in users
        ),
        key=lambda item: item["count"],
        reverse=True,
    )
    return [row for row in ranked[:top_n] if row["count"]]


def collect_user_snapshots(handle_map: Dict[str, Dict[str, Any]]) -> Dict[str, Any]:
    user_map = user_manager.list_users()
    items: List[Dict[str, Any]] = []
    role_counter: Counter = Counter()
    usage_totals = {"fast": 0, "thinking": 0, "search": 0}
    token_totals = {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0}
    storage_total_bytes = 0
    quarantine_total_bytes = 0
    now = time.time()

    for username, record in user_map.items():
        workspace = user_manager.ensure_user_workspace(username)
        storage = compute_workspace_storage(workspace)
        usage = collect_usage_snapshot(username, workspace, record.role)
        tokens = collect_user_token_statistics(workspace)
        storage_total_bytes += storage["total_bytes"]
        quarantine_total_bytes += storage["quarantine_bytes"]
        for metric in usage_totals:
            usage_totals[metric] += usage.get(metric, {}).get("count", 0)
        for key in token_totals:
            token_totals[key] += tokens.get(key, 0)
        normalized_role = (record.role or "user").lower()
        role_counter[normalized_role] += 1
        handle = handle_map.get(username)
        handle_last = handle.get("last_active") if handle else None
        last_active = get_last_active_ts(username, handle_last)
        idle_seconds = max(0.0, now - last_active) if last_active else None
        items.append({
            "username": username,
            "email": record.email,
            "role": record.role or "user",
            "created_at": record.created_at,
            "invite_code": record.invite_code,
            "storage": storage,
            "usage": usage,
            "tokens": tokens,
            "workspace": {
                "project_path": str(workspace.project_path),
                "data_dir": str(workspace.data_dir),
                "logs_dir": str(workspace.logs_dir),
                "uploads_dir": str(workspace.uploads_dir),
            },
            "status": {
                "online": handle is not None,
                "container_mode": handle.get("mode") if handle else None,
                "last_active": iso_datetime_from_epoch(last_active),
                "idle_seconds": idle_seconds,
            },
        })

    items.sort(key=lambda entry: entry["username"])
    return {
        "items": items,
        "roles": dict(role_counter),
        "usage_totals": usage_totals,
        "token_totals": token_totals,
        "storage_total_bytes": storage_total_bytes,
        "quarantine_total_bytes": quarantine_total_bytes,
        "active_users": sum(1 for entry in items if entry["status"]["online"]),
        "total_users": len(items),
    }


def collect_container_snapshots(handle_map: Dict[str, Dict[str, Any]]) -> Dict[str, Any]:
    items: List[Dict[str, Any]] = []
    cpu_values: List[float] = []
    mem_percent_values: List[float] = []
    total_mem_used = 0
    total_mem_limit = 0
    total_net_rx = 0
    total_net_tx = 0
    docker_count = 0
    failure_count = 0
    now = time.time()

    for username, handle in sorted(handle_map.items()):
        try:
            status = container_manager.get_container_status(username)
        except Exception as exc:
            status = {
                "username": username,
                "mode": handle.get("mode"),
                "error": str(exc),
                "workspace_path": handle.get("workspace_path"),
            }
        stats = status.get("stats") or {}
        state = status.get("state") or {}
        if status.get("mode") == "docker":
            docker_count += 1
        last_active = get_last_active_ts(username, handle.get("last_active"))
        idle_seconds = max(0.0, now - last_active) if last_active else None
        entry = {
            "username": username,
            "mode": status.get("mode", handle.get("mode")),
            "workspace_path": status.get("workspace_path") or handle.get("workspace_path"),
            "container_name": status.get("container_name") or handle.get("container_name"),
            "created_at": iso_datetime_from_epoch(status.get("created_at") or handle.get("created_at")),
            "last_active": iso_datetime_from_epoch(status.get("last_active") or last_active),
            "idle_seconds": idle_seconds,
            "stats": stats,
            "state": state,
            "error": status.get("error"),
        }
        if entry["error"] or (state and not state.get("running", True)):
            failure_count += 1
        mem_info = stats.get("memory") or {}
        net_info = stats.get("net_io") or {}
        cpu_val = stats.get("cpu_percent")
        mem_percent = mem_info.get("percent")
        mem_used = mem_info.get("used_bytes")
        mem_limit = mem_info.get("limit_bytes")
        rx_bytes = net_info.get("rx_bytes")
        tx_bytes = net_info.get("tx_bytes")
        if isinstance(cpu_val, (int, float)):
            cpu_values.append(cpu_val)
        if isinstance(mem_percent, (int, float)):
            mem_percent_values.append(mem_percent)
        if isinstance(mem_used, (int, float)):
            total_mem_used += mem_used
        if isinstance(mem_limit, (int, float)):
            total_mem_limit += mem_limit
        if isinstance(rx_bytes, (int, float)):
            total_net_rx += rx_bytes
        if isinstance(tx_bytes, (int, float)):
            total_net_tx += tx_bytes
        items.append(entry)

    active_total = len(handle_map)
    summary = {
        "active": active_total,
        "docker": docker_count,
        "host": active_total - docker_count,
        "issues": failure_count,
        "max_containers": container_manager.max_containers,
        "available_slots": max(0, container_manager.max_containers - active_total) if container_manager.max_containers > 0 else None,
        "avg_cpu_percent": round(sum(cpu_values) / len(cpu_values), 2) if cpu_values else None,
        "avg_mem_percent": round(sum(mem_percent_values) / len(mem_percent_values), 2) if mem_percent_values else None,
        "total_mem_used_bytes": total_mem_used,
        "total_mem_limit_bytes": total_mem_limit,
        "net_rx_bytes": total_net_rx,
        "net_tx_bytes": total_net_tx,
    }
    return {"items": items, "summary": summary}


def parse_upload_line(line: str) -> Optional[Dict[str, Any]]:
    marker = "UPLOAD_AUDIT "
    idx = line.find(marker)
    if idx == -1:
        return None
    payload = line[idx + len(marker):].strip()
    try:
        data = json.loads(payload)
    except json.JSONDecodeError:
        return None
    timestamp_value = data.get("timestamp")
    timestamp_dt = None
    if isinstance(timestamp_value, str):
        try:
            timestamp_dt = datetime.fromisoformat(timestamp_value)
        except ValueError:
            timestamp_dt = None
    data["_dt"] = timestamp_dt
    return data


def collect_upload_events(limit: int = RECENT_UPLOAD_EVENT_LIMIT) -> List[Dict[str, Any]]:
    base_dir = (Path(LOGS_DIR).expanduser().resolve() / UPLOAD_SCAN_LOG_SUBDIR).resolve()
    events: List[Dict[str, Any]] = []
    if not base_dir.exists():
        return []
    for log_file in sorted(base_dir.glob('*.log')):
        buffer: deque = deque(maxlen=limit)
        try:
            with open(log_file, 'r', encoding='utf-8') as fh:
                for line in fh:
                    if 'UPLOAD_AUDIT' not in line:
                        continue
                    buffer.append(line.strip())
        except OSError:
            continue
        for raw in buffer:
            event = parse_upload_line(raw)
            if event:
                events.append(event)
    events.sort(key=lambda item: item.get('_dt') or datetime.min, reverse=True)
    return events[:limit]


def summarize_upload_events(events: List[Dict[str, Any]], quarantine_total_bytes: int) -> Dict[str, Any]:
    now = datetime.utcnow()
    cutoff = now - timedelta(hours=24)
    last_24h = [evt for evt in events if evt.get('_dt') and evt['_dt'] >= cutoff]
    accepted_24h = sum(1 for evt in last_24h if evt.get('accepted'))
    blocked_24h = len(last_24h) - accepted_24h
    skipped_24h = sum(1 for evt in last_24h if ((evt.get('scan') or {}).get('status') == 'skipped'))
    source_counter = Counter((evt.get('source') or 'unknown') for evt in events)
    sanitized_events: List[Dict[str, Any]] = []
    for evt in events[:RECENT_UPLOAD_FEED_LIMIT]:
        sanitized_events.append({k: v for k, v in evt.items() if k != '_dt'})
    return {
        "stats": {
            "total_tracked": len(events),
            "last_24h": len(last_24h),
            "accepted_last_24h": accepted_24h,
            "blocked_last_24h": blocked_24h,
            "skipped_scan_last_24h": skipped_24h,
            "quarantine_bytes": quarantine_total_bytes,
        },
        "recent_events": sanitized_events,
        "sources": [{"source": src, "count": count} for src, count in source_counter.most_common()],
    }


def summarize_invite_codes(codes: List[Dict[str, Any]]) -> Dict[str, int]:
    active = consumed = unlimited = 0
    for code in codes:
        remaining = code.get('remaining')
        if remaining is None:
            unlimited += 1
        elif remaining > 0:
            active += 1
        else:
            consumed += 1
    return {
        "total": len(codes),
        "active": active,
        "consumed": consumed,
        "unlimited": unlimited,
    }


def build_admin_dashboard_snapshot() -> Dict[str, Any]:
    handle_map = container_manager.list_containers()
    user_data = collect_user_snapshots(handle_map)
    container_data = collect_container_snapshots(handle_map)
    invite_codes = user_manager.list_invite_codes()
    upload_events = collect_upload_events()
    uploads_summary = summarize_upload_events(upload_events, user_data['quarantine_total_bytes'])
    overview = {
        "generated_at": datetime.utcnow().replace(microsecond=0).isoformat() + "Z",
        "totals": {
            "users": user_data['total_users'],
            "active_users": user_data['active_users'],
            "containers_active": container_data['summary']['active'],
            "containers_max": container_data['summary']['max_containers'],
            "available_container_slots": container_data['summary']['available_slots'],
        },
        "roles": user_data['roles'],
        "usage_totals": user_data['usage_totals'],
        "token_totals": user_data['token_totals'],
        "usage_leaders": {
            metric: compute_usage_leaders(user_data['items'], metric)
            for metric in ("fast", "thinking", "search")
        },
        "storage": {
            "total_bytes": user_data['storage_total_bytes'],
            "per_user_limit_bytes": PROJECT_MAX_STORAGE_BYTES,
            "project_max_mb": PROJECT_MAX_STORAGE_MB,
            "warning_users": [
                {
                    "username": entry['username'],
                    "usage_percent": entry['storage']['usage_percent'],
                    "status": entry['storage']['status'],
                }
                for entry in user_data['items']
                if entry['storage']['status'] != 'ok'
            ],
        },
        "containers": container_data['summary'],
        "invites": summarize_invite_codes(invite_codes),
        "uploads": uploads_summary['stats'],
    }
    return {
        "generated_at": overview['generated_at'],
        "overview": overview,
        "users": user_data['items'],
        "containers": container_data['items'],
        "invites": {
            "summary": summarize_invite_codes(invite_codes),
            "codes": invite_codes,
        },
        "uploads": uploads_summary,
    }


def initialize_system(path: str, thinking_mode: bool = False):
    """初始化系统（多用户版本仅负责写日志和配置）"""
    # 清空或创建调试日志
    DEBUG_LOG_FILE.parent.mkdir(parents=True, exist_ok=True)
    with DEBUG_LOG_FILE.open('w', encoding='utf-8') as f:
        f.write(f"调试日志开始 - {datetime.now()}\n")
        f.write(f"项目路径: {path}\n")
        f.write(f"思考模式: {'思考模式' if thinking_mode else '快速模式'}\n")
        f.write(f"自动修复: {'开启' if AUTO_FIX_TOOL_CALL else '关闭'}\n")
        f.write(f"最大迭代: {MAX_ITERATIONS_PER_TASK}\n")
        f.write(f"最大工具调用: {MAX_TOTAL_TOOL_CALLS}\n")
        f.write("="*80 + "\n")
    
    print(f"[Init] 初始化Web系统...")
    print(f"[Init] 项目路径: {path}")
    print(f"[Init] 运行模式: {'思考模式（首次思考，后续快速）' if thinking_mode else '快速模式（无思考）'}")
    print(f"[Init] 自动修复: {'开启' if AUTO_FIX_TOOL_CALL else '关闭'}")
    print(f"[Init] 调试日志: {DEBUG_LOG_FILE}")
    
    app.config['DEFAULT_THINKING_MODE'] = thinking_mode
    app.config['DEFAULT_RUN_MODE'] = "thinking" if thinking_mode else "fast"
    print(f"{OUTPUT_FORMATS['success']} Web系统初始化完成（多用户模式）")

def run_server(path: str, thinking_mode: bool = False, port: int = DEFAULT_PORT, debug: bool = False):
    """运行Web服务器"""
    initialize_system(path, thinking_mode)
    start_background_jobs()
    socketio.run(
        app,
        host='0.0.0.0',
        port=port,
        debug=debug,
        use_reloader=debug,
        allow_unsafe_werkzeug=True
    )


def parse_arguments():
    parser = argparse.ArgumentParser(description="AI Agent Web Server")
    parser.add_argument(
        "--path",
        default=str(Path(DEFAULT_PROJECT_PATH).resolve()),
        help="项目工作目录（默认使用 config.DEFAULT_PROJECT_PATH）"
    )
    parser.add_argument(
        "--port",
        type=int,
        default=DEFAULT_PORT,
        help=f"监听端口（默认 {DEFAULT_PORT}）"
    )
    parser.add_argument(
        "--debug",
        action="store_true",
        help="开发模式，启用 Flask/Socket.IO 热重载"
    )
    parser.add_argument(
        "--thinking-mode",
        action="store_true",
        help="启用思考模式（首次请求使用 reasoning）"
    )
    return parser.parse_args()


if __name__ == "__main__":
    args = parse_arguments()
    run_server(
        path=args.path,
        thinking_mode=args.thinking_mode,
        port=args.port,
        debug=args.debug
    )
@app.route('/resource_busy')
def resource_busy_page():
    return app.send_static_file('resource_busy.html'), 503
