zk-data-agent/backend/api/server.py

"""FastAPI server backing the local web GUI.

Wraps a single global :class:`LocalCodingAgent`, exposes JSON endpoints for
chat, slash commands, and saved sessions, and serves the static SPA.
"""

from __future__ import annotations

import asyncio
import csv
import json
import os
import queue
import re
import shutil
import signal
import subprocess
import threading
import time
import venv
from dataclasses import dataclass, field, replace
from pathlib import Path
from threading import Lock, RLock
from typing import Any
from urllib import error, request
from uuid import uuid4

from fastapi import FastAPI, HTTPException
from fastapi.responses import FileResponse, JSONResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel, Field

from src.agent_session import AgentMessage, AgentSessionState
from src.agent_runtime import LocalCodingAgent
from src.agent_slash_commands import get_slash_command_specs
from src.agent_types import (
    AgentPermissions,
    AgentRuntimeConfig,
    ModelConfig,
)
from src.bundled_skills import ALWAYS_ENABLED_HIDDEN_SKILL_NAMES, get_bundled_skills
from src.data_agent_inputs import DataAgentInputError, load_input_sources
from src.jupyter_runtime import (
    DEFAULT_JUPYTER_WORKSPACE_ROOT,
    JupyterRuntimeError,
    JupyterRuntimeManager,
)
from src.mcp_runtime import MCPRuntime, MCPServerProfile
from src.openai_compat import OpenAICompatClient, OpenAICompatError
from src.session_store import (
    DEFAULT_AGENT_SESSION_DIR,
    StoredAgentSession,
    deserialize_runtime_config,
    load_agent_session,
    save_agent_session,
    serialize_model_config,
    serialize_runtime_config,
)
from src.token_budget import calculate_token_budget


STATIC_DIR = Path(__file__).resolve().parents[2] / 'frontend' / 'legacy-static'
API_TOOL_CONTENT_MAX_CHARS = 20000
FEISHU_NPM_REGISTRY = 'https://pkgs.d.xiaomi.net/artifactory/api/npm/mi-npm/'
FEISHU_MCP_PACKAGE = '@mi/feishu-mcp-pro@latest'
FEISHU_MCP_SERVER_NAME = 'feishu-mcp-pro'
FEISHU_DOCUMENT_SUFFIXES = {'.docx', '.md', '.txt'}
FEISHU_SPREADSHEET_SUFFIXES = {'.csv', '.xlsx'}
FEISHU_SUPPORTED_DOCUMENT_SUFFIXES = FEISHU_DOCUMENT_SUFFIXES | FEISHU_SPREADSHEET_SUFFIXES
FEISHU_UNSUPPORTED_DOCUMENT_SUFFIXES = {'.json', '.jsonl'}
FEISHU_DOC_MARKDOWN_MAX_CHARS = 100_000
FEISHU_SPREADSHEET_MAX_ROWS = 5000
FEISHU_SPREADSHEET_MAX_COLS = 200
FEISHU_SPREADSHEET_MAX_CELL_CHARS = 5000
FEISHU_SPREADSHEET_WRITE_BATCH_ROWS = 500
FEISHU_ONLINE_DOCS_FILENAME = 'online-docs.json'


@dataclass
class FeishuLoginProcess:
    process: subprocess.Popen[str]
    started_at: float
    output: list[str] = field(default_factory=list)
    login_url: str | None = None
    user_code: str | None = None


_FEISHU_LOGIN_PROCESSES: dict[str, FeishuLoginProcess] = {}
_FEISHU_LOGIN_LOCK = threading.Lock()
_FEISHU_DOC_MAP_LOCK = threading.Lock()

VALIDATED_CHAT_MODEL_PROVIDERS = {
    # 这些 provider 已验证可以在当前 WebUI 中使用。
    # 部分模型族会在客户端按模型 id 切换到专用兼容路由。
    'azure_openai',
    'hunyuan',
    'minimax',
    'moonshot',
    'ppio',
    'siliconflow',
    'tongyi',
    'vertex_ai',
    'volcengine_maas',
    'wenxin',
    'xiaomi',
    'zhipuai',
}

UNSUPPORTED_TOOL_CHAT_MODEL_IDS = {
    # 这些模型来自 /models，但带 tools/tool_choice 的最小 Agent 请求实测失败。
    'azure_openai/gpt-5.5',
    'azure_openai/grok-3',
    'azure_openai/grok-3-mini',
    'minimax/abab6.5t-chat',
    'ppio/gemini-2.0-flash-20250609',
    'siliconflow/deepseek-ai/deepseek-ocr',
    'siliconflow/deepseek-ai/deepseek-r1-distill-qwen-14b',
    'siliconflow/deepseek-ai/deepseek-r1-distill-qwen-32b',
    'siliconflow/deepseek-ai/deepseek-r1-distill-qwen-7b',
    'siliconflow/qwen/qwen2.5-coder-32b-instruct',
    'siliconflow/qwen/qwen2.5-vl-32b-instruct',
    'siliconflow/qwen/qwen2.5-vl-72b-instruct',
    'siliconflow/qwen/qwen3-235b-a22b',
    'siliconflow/qwen/qwen3-235b-a22b-thinking-2507',
    'siliconflow/qwen/qwen3-30b-a3b-instruct-2507',
    'siliconflow/qwen/qwen3-vl-8b-instruct',
    'siliconflow/qwen/qwen3-vl-8b-thinking',
    'siliconflow/thudm/glm-4-9b-chat',
    'siliconflow/thudm/glm-4.1v-9b-thinking',
    'siliconflow/zai-org/glm-4.5',
    'tongyi/deepseek-r1-distill-qwen-14b',
    'tongyi/deepseek-r1-distill-qwen-32b',
    'tongyi/qwen-mt-plus',
    'tongyi/qwen-mt-turbo',
    'tongyi/qwen-plus-0919',
    'tongyi/qwen2.5-0.5b-instruct',
    'tongyi/qwen2.5-1.5b-instruct',
    'tongyi/qwen3-0.6b',
    'tongyi/qwen3-1.7b',
    'tongyi/qwen3-30b-a3b',
    'tongyi/qwen3-4b',
    'tongyi/qwen3-8b',
    'vertex_ai/gemini-2.0-flash-001',
    'vertex_ai/gemini-2.0-flash-lite-001',
    'vertex_ai/gemini-2.5-computer-use-preview-10-2025',
    'vertex_ai/gemini-2.5-flash',
    'vertex_ai/gemini-2.5-flash-lite',
    'vertex_ai/gemini-2.5-flash-lite-preview-09-2025',
    'vertex_ai/gemini-2.5-pro',
    'vertex_ai/gemini-3-flash-preview',
    'vertex_ai/gemini-3.1-flash-lite-preview',
    'vertex_ai/gemini-3.1-pro-preview',
    'vertex_ai/gemini-3.1-pro-preview-pt',
    'wenxin/ernie-speed-128k',
    'wenxin/ernie-speed-8k',
    'xiaomi/mi-brag-vl',
    'xiaomi/midashenglm-7b-1021-bf16',
    'xiaomi/milm2.1-13b-chat',
    'xiaomi/mimo-vl-7b-rl',
    'xiaomi/mimo-vl-7b-rl-0808',
    'xiaomi/paddleocr-vl-0.9b',
    'xiaomi/qwen-235b-a22b',
    'xiaomi/qwen2.5-72b-instruct-gptq-int4',
    'xiaomi/qwen2.5-vl-72b-instruct-awq',
    'xiaomi/qwen25-coder-7b',
    'xiaomi/qwen3-235b-a22b',
    'xiaomi/qwen3-235b-a22b-instruct-2507',
    'xiaomi/qwen3-32b',
}

# WebUI 的快速入口只展示可以通过对话输入框安全执行的 slash command。
# 这些命令仍然保留在终端 `/` 列表里，但不适合作为 WebUI 快捷项。
WEBUI_HIDDEN_SLASH_COMMANDS = {
    'exit',
    'quit',
    'feedback',
    'bug',
    'upgrade',
    'stickers',
    'chrome',
    'install-github-app',
    'install-slack-app',
    'privacy-settings',
    'mobile',
    'ios',
    'android',
    'desktop',
    'app',
    'vim',
    'theme',
}

WEBUI_SLASH_COMMAND_DESCRIPTIONS_ZH = {
    'help': '查看内置 slash command 帮助。',
    'context': '查看当前会话上下文用量估算。',
    'context-raw': '查看原始环境、用户上下文和系统上下文快照。',
    'token-budget': '查看当前 token 预算窗口、保留量和提示词长度限制。',
    'mcp': '查看本地 MCP 清单和资源数量。',
    'search': '查看搜索运行状态、切换搜索提供方，或执行一次网页搜索。',
    'remote': '查看远程运行状态，或激活远程目标/配置。',
    'worktree': '查看受管 git worktree 状态，或进入/退出当前 worktree 会话。',
    'account': '查看账号运行状态或已配置账号档案。',
    'ask': '查看 ask-user 运行状态或历史记录。',
    'login': '激活本地账号档案或临时身份。',
    'logout': '清除当前本地账号会话。',
    'config': '查看配置状态、有效配置、配置来源或某个配置值。',
    'lsp': '查看 LSP 状态，或执行符号、定义、引用、悬停、调用层级和诊断查询。',
    'remotes': '列出本地远程配置。',
    'ssh': '激活 SSH 远程目标/配置。',
    'teleport': '激活 Teleport 远程目标/配置。',
    'direct-connect': '激活直连远程目标/配置。',
    'deep-link': '激活 deep-link 远程目标/配置。',
    'disconnect': '断开当前远程运行目标。',
    'resources': '列出本地 MCP 资源，可按关键词过滤。',
    'resource': '按 URI 渲染一个本地 MCP 资源。',
    'tasks': '查看本地运行时任务列表，可按状态过滤。',
    'workflows': '列出从工作流清单发现的本地 workflows。',
    'workflow': '查看或运行一个本地 workflow。',
    'triggers': '列出从清单发现的本地远程触发器。',
    'trigger': '查看或运行一个远程触发器。',
    'teams': '列出本地协作团队配置。',
    'team': '查看一个本地协作团队。',
    'messages': '查看所有团队或某个团队的协作消息。',
    'task-next': '查看本地任务列表中的下一个可执行任务。',
    'plan': '查看当前本地运行计划。',
    'task': '按 id 查看一个本地任务。',
    'prompt': '渲染当前生效的系统提示词。',
    'permissions': '查看当前工具权限模式。',
    'hooks': '查看已发现的本地 hook 和 policy 清单。',
    'trust': '查看 workspace 信任模式、受管设置和安全环境变量。',
    'model': '查看或修改当前 Agent 实例使用的模型。',
    'tools': '列出已注册工具，以及当前权限是否允许调用。',
    'agents': '列出本地 Agent 配置，或查看某个 Agent 定义。',
    'memory': '查看当前加载的 CLAUDE.md 记忆包和已发现文件。',
    'status': '查看当前运行时和会话状态摘要。',
    'clear': '清理当前进程中的临时运行状态。',
    'compact': '总结并压缩当前对话，释放上下文空间。',
    'cost': '查看当前会话的总耗时和费用估算。',
    'diff': '查看当前工作区未提交改动。',
    'files': '列出当前会话上下文中已加载的文件。',
    'copy': '把上一条助手回复写入临时文件。',
    'export': '导出当前对话为文本文件。',
    'stats': '查看会话使用统计。',
    'tag': '给当前会话添加或移除可搜索标签。',
    'rename': '重命名当前对话。',
    'branch': '基于当前对话创建一个分支/副本。',
    'effort': '查看或设置模型推理强度。',
    'doctor': '诊断并验证 claw-code 安装和配置。',
    'commit': '创建 git commit。',
    'pr-comments': '读取 GitHub PR 评论。',
    'resume': '恢复之前的对话。',
    'add-dir': '添加新的工作目录。',
    'skills': '列出可用 Skills。',
    'fast': '切换 fast mode。',
    'rewind': '把对话恢复到之前的 checkpoint。',
    'output-style': '已废弃：请使用 /config 修改输出风格。',
    'release-notes': '查看本地 release notes 或变更日志链接。',
    'extra-usage': '查看 extra-usage 配置链接。',
    'passes': '查看 Claude Code guest passes 信息。',
    'rate-limit-options': '查看账号触发限流时的可选处理方式。',
    'reload-plugins': '重新加载本地插件清单并报告数量。',
    'voice': '切换当前 workspace 的 voice mode 设置。',
    'sandbox-toggle': '查看 sandbox 状态，或排除一条命令模式。',
    'keybindings': '打印或创建本地快捷键配置文件。',
    'btw': '向模型快速提一个旁路问题，不修改当前任务状态。',
    'version': '打印当前 Agent 版本。',
    'init': '初始化 CLAUDE.md 代码库说明文件。',
    'ide': '查看检测到的 IDE/终端集成状态。',
    'plugin': '列出已安装插件，或查看插件子命令用法。',
    'remote-env': '列出远程环境，或设置默认配置。',
    'bridge': '查看 remote-control bridge 状态。',
    'remote-setup': '检查 Claude Code on the web 的准备状态。',
}


# ---------------------------------------------------------------------------
# Agent state holder
# ---------------------------------------------------------------------------

@dataclass(frozen=True)
class AgentInstanceConfig:
    cwd: Path
    model: str
    base_url: str
    api_key: str
    timeout_seconds: float
    allow_shell: bool
    allow_write: bool


SKILL_SETTINGS_FILENAME = 'skill_settings.json'


class RunProcessRegistry:
    def __init__(self) -> None:
        self._lock = Lock()
        self._processes: set[Any] = set()

    def add(self, process: Any) -> None:
        with self._lock:
            self._processes.add(process)

    def discard(self, process: Any) -> None:
        with self._lock:
            self._processes.discard(process)

    def kill_all(self) -> None:
        with self._lock:
            processes = list(self._processes)
        for process in processes:
            try:
                if process.poll() is None:
                    self._signal_process_tree(process, signal.SIGTERM)
            except OSError:
                continue
        deadline = time.monotonic() + 1.0
        for process in processes:
            try:
                remaining = max(0.0, deadline - time.monotonic())
                process.wait(timeout=remaining)
            except Exception:
                self._signal_process_tree(process, signal.SIGKILL)
                try:
                    process.wait(timeout=1.0)
                except Exception:
                    pass

    @staticmethod
    def _signal_process_tree(process: Any, sig: signal.Signals) -> None:
        try:
            if os.name == 'posix':
                pgid = os.getpgid(process.pid)
                if pgid == process.pid:
                    os.killpg(pgid, sig)
                    return
            if sig == signal.SIGTERM:
                process.terminate()
            else:
                process.kill()
        except ProcessLookupError:
            return
        except OSError:
            try:
                if sig == signal.SIGTERM:
                    process.terminate()
                else:
                    process.kill()
            except OSError:
                pass


@dataclass
class RunRecord:
    run_id: str
    account_key: str
    session_id: str
    status: str
    started_at: float
    updated_at: float
    cancel_event: threading.Event
    process_registry: RunProcessRegistry
    pending_prompt: str = ''
    current_stage: str = ''
    error: str = ''
    events: list[dict[str, Any]] = field(default_factory=list)


class RunManager:
    def __init__(self) -> None:
        self._lock = RLock()
        self._runs: dict[str, RunRecord] = {}
        self._latest_by_session: dict[tuple[str, str], str] = {}

    def start(self, account_key: str, session_id: str, pending_prompt: str = '') -> RunRecord:
        now = time.time()
        record = RunRecord(
            run_id=uuid4().hex,
            account_key=account_key,
            session_id=session_id,
            status='queued',
            started_at=now,
            updated_at=now,
            cancel_event=threading.Event(),
            process_registry=RunProcessRegistry(),
            pending_prompt=pending_prompt,
        )
        with self._lock:
            self._runs[record.run_id] = record
            self._latest_by_session[(account_key, session_id)] = record.run_id
        return record

    def update(self, run_id: str, *, status: str | None = None, stage: str | None = None, error: str | None = None) -> None:
        with self._lock:
            record = self._runs.get(run_id)
            if record is None:
                return
            if status is not None:
                record.status = status
            if stage is not None:
                record.current_stage = stage
            if error is not None:
                record.error = error
            record.updated_at = time.time()

    def record_event(self, run_id: str, event: dict[str, object]) -> None:
        normalized = _normalize_run_event(event)
        if normalized is None:
            return
        normalized['recorded_at'] = time.time()
        with self._lock:
            record = self._runs.get(run_id)
            if record is None:
                return
            record.events.append(normalized)
            if len(record.events) > 160:
                del record.events[: len(record.events) - 160]
            record.updated_at = time.time()

    def finish(self, run_id: str, status: str) -> None:
        self.update(run_id, status=status)

    def cancel_latest(self, account_key: str, session_id: str) -> bool:
        with self._lock:
            record = self._select_session_record(account_key, session_id)
        return self.cancel_record(record)

    def cancel_run(self, run_id: str) -> bool:
        with self._lock:
            record = self._runs.get(run_id)
        return self.cancel_record(record)

    def cancel_record(self, record: RunRecord | None) -> bool:
        if record is None or record.status in {'completed', 'failed', 'cancelled'}:
            return False
        record.cancel_event.set()
        record.process_registry.kill_all()
        self.update(record.run_id, status='cancelled', stage='用户已取消')
        return True

    def snapshot_latest(self, account_key: str, session_id: str) -> dict[str, Any] | None:
        with self._lock:
            record = self._select_session_record(account_key, session_id)
            if record is None:
                return None
            return {
                'run_id': record.run_id,
                'session_id': record.session_id,
                'status': record.status,
                'current_stage': record.current_stage,
                'started_at': record.started_at,
                'updated_at': record.updated_at,
                'pending_prompt': record.pending_prompt,
                'error': record.error,
                'events': [dict(event) for event in record.events],
            }

    def _select_session_record(
        self,
        account_key: str,
        session_id: str,
    ) -> RunRecord | None:
        records = [
            record
            for record in self._runs.values()
            if record.account_key == account_key and record.session_id == session_id
        ]
        running = [
            record
            for record in records
            if record.status == 'running'
        ]
        if running:
            return max(running, key=lambda item: item.updated_at)
        queued = [
            record
            for record in records
            if record.status == 'queued'
        ]
        if queued:
            return max(queued, key=lambda item: item.started_at)
        run_id = self._latest_by_session.get((account_key, session_id))
        return self._runs.get(run_id or '')


class AgentState:
    """Holds account-scoped agent instances, config, and execution locks."""

    def __init__(
        self,
        *,
        cwd: Path,
        model: str,
        base_url: str,
        api_key: str,
        timeout_seconds: float,
        allow_shell: bool,
        allow_write: bool,
        session_directory: Path,
    ) -> None:
        self.session_directory = session_directory
        self._lock = RLock()
        self._agents: dict[str, LocalCodingAgent] = {}
        self._run_locks: dict[str, Lock] = {}
        self._account_configs: dict[str, AgentInstanceConfig] = {}
        self.run_manager = RunManager()
        self.jupyter_runtime_manager = JupyterRuntimeManager()
        self._default_config = AgentInstanceConfig(
            cwd=cwd.resolve(),
            model=model,
            base_url=base_url,
            api_key=api_key,
            timeout_seconds=timeout_seconds,
            allow_shell=allow_shell,
            allow_write=allow_write,
        )

    @property
    def cwd(self) -> Path:
        return self._default_config.cwd

    @property
    def model(self) -> str:
        return self._default_config.model

    @property
    def base_url(self) -> str:
        return self._default_config.base_url

    @property
    def api_key(self) -> str:
        return self._default_config.api_key

    @property
    def allow_shell(self) -> bool:
        return self._default_config.allow_shell

    @property
    def allow_write(self) -> bool:
        return self._default_config.allow_write

    def _account_key(self, account_id: str | None) -> str:
        return _safe_account_id(account_id) if account_id else '__default__'

    def _session_key(self, account_id: str | None, session_id: str | None = None) -> str:
        session_part = _safe_session_id(session_id) or '__shared__'
        return f'{self._account_key(account_id)}:{session_part}'

    def _config_for(self, account_id: str | None) -> AgentInstanceConfig:
        if not account_id:
            return self._default_config
        key = self._account_key(account_id)
        config = self._account_configs.get(key)
        if config is None:
            config = replace(self._default_config)
            self._account_configs[key] = config
        return config

    def config_for(self, account_id: str | None) -> AgentInstanceConfig:
        with self._lock:
            return self._config_for(account_id)

    def _set_config_for(self, account_id: str | None, config: AgentInstanceConfig) -> None:
        if not account_id:
            self._default_config = config
            return
        self._account_configs[self._account_key(account_id)] = config

    def _account_base(self, account_id: str | None) -> Path:
        if not account_id:
            return self.session_directory.parent
        safe_id = _safe_account_id(account_id)
        return (self.session_directory.parent / 'accounts' / safe_id).resolve()

    def account_paths(self, account_id: str | None) -> dict[str, Path]:
        if not account_id:
            base = self.session_directory.parent
            return {
                'base': base,
                'sessions': self.session_directory,
                'scratchpad': self.session_directory,
                'uploads': self.session_directory,
                'outputs': self.session_directory,
                'python_env': base / 'python' / '.venv',
            }
        base = self._account_base(account_id)
        return {
            'base': base,
            'sessions': base / 'sessions',
            'scratchpad': base / 'sessions',
            'uploads': base / 'sessions',
            'outputs': base / 'sessions',
            'python_env': base / 'python' / '.venv',
        }

    def _skill_settings_path(self, account_id: str | None) -> Path:
        return self.account_paths(account_id)['base'] / SKILL_SETTINGS_FILENAME

    def _load_disabled_skill_names(self, account_id: str | None) -> set[str]:
        path = self._skill_settings_path(account_id)
        try:
            payload = json.loads(path.read_text(encoding='utf-8'))
        except (OSError, json.JSONDecodeError):
            return set()
        disabled = payload.get('disabled_skill_names')
        if not isinstance(disabled, list):
            return set()
        return {
            str(name).strip().lower()
            for name in disabled
            if str(name).strip()
        }

    def _save_disabled_skill_names(
        self,
        account_id: str | None,
        disabled_skill_names: set[str],
    ) -> None:
        path = self._skill_settings_path(account_id)
        path.parent.mkdir(parents=True, exist_ok=True)
        payload = {
            'disabled_skill_names': sorted(
                name
                for name in disabled_skill_names
                if name not in ALWAYS_ENABLED_HIDDEN_SKILL_NAMES
            )
        }
        path.write_text(
            json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + '\n',
            encoding='utf-8',
        )

    def enabled_skill_names(self, account_id: str | None) -> tuple[str, ...]:
        config = self._config_for(account_id)
        disabled = self._load_disabled_skill_names(account_id)
        enabled: list[str] = []
        for skill in get_bundled_skills(config.cwd):
            if not skill.user_invocable:
                continue
            lowered = skill.name.lower()
            if (
                lowered in ALWAYS_ENABLED_HIDDEN_SKILL_NAMES
                or lowered not in disabled
            ):
                enabled.append(skill.name)
        return tuple(enabled)

    def set_skill_enabled(
        self,
        account_id: str | None,
        skill_name: str,
        enabled: bool,
    ) -> None:
        with self._lock:
            skill_name = skill_name.strip()
            if not skill_name:
                raise ValueError('skill name must not be empty')
            lowered = skill_name.lower()
            if lowered in ALWAYS_ENABLED_HIDDEN_SKILL_NAMES:
                raise ValueError(
                    f'skill is always enabled and cannot be changed: {skill_name}'
                )
            config = self._config_for(account_id)
            if not any(
                skill.name.lower() == lowered and skill.user_invocable
                for skill in get_bundled_skills(config.cwd)
            ):
                raise ValueError(f'unknown skill: {skill_name}')
            disabled = self._load_disabled_skill_names(account_id)
            if enabled:
                disabled.discard(lowered)
            else:
                disabled.add(lowered)
            self._save_disabled_skill_names(account_id, disabled)
            account_prefix = f'{self._account_key(account_id)}:'
            for key in list(self._agents):
                if key.startswith(account_prefix):
                    self._agents.pop(key, None)

    def set_all_skills_enabled(
        self,
        account_id: str | None,
        enabled: bool,
    ) -> None:
        with self._lock:
            config = self._config_for(account_id)
            configurable_skill_names = {
                skill.name.lower()
                for skill in get_bundled_skills(config.cwd)
                if skill.user_invocable
                and skill.name.lower() not in ALWAYS_ENABLED_HIDDEN_SKILL_NAMES
            }
            disabled = self._load_disabled_skill_names(account_id)
            if enabled:
                disabled.difference_update(configurable_skill_names)
            else:
                disabled.update(configurable_skill_names)
            self._save_disabled_skill_names(account_id, disabled)
            self._clear_agents_for_account(account_id)

    def _clear_agents_for_account(self, account_id: str | None) -> None:
        account_prefix = f'{self._account_key(account_id)}:'
        for key in list(self._agents):
            if key.startswith(account_prefix):
                self._agents.pop(key, None)

    def sync_skills_from_git(self, account_id: str | None) -> dict[str, Any]:
        """Pull the current git branch and refresh account-scoped skill state."""
        with self._lock:
            config = self._config_for(account_id)
            repo = config.cwd
            if not (repo / '.git').exists():
                raise ValueError(f'当前工作目录不是 git 仓库：{repo}')

            def run_git(args: list[str], *, timeout: int = 60) -> subprocess.CompletedProcess[str]:
                env = {
                    **os.environ,
                    'GIT_TERMINAL_PROMPT': '0',
                }
                proc = subprocess.run(
                    ['git', *args],
                    cwd=repo,
                    capture_output=True,
                    text=True,
                    timeout=timeout,
                    env=env,
                )
                if proc.returncode != 0:
                    detail = (proc.stderr or proc.stdout or '').strip()
                    raise ValueError(f'git {" ".join(args)} 失败：{detail}')
                return proc

            dirty = run_git(['status', '--porcelain', '--untracked-files=no']).stdout.strip()
            if dirty:
                raise ValueError('工作区存在未提交的 tracked 改动，已停止同步 Skill。')

            branch = run_git(['branch', '--show-current']).stdout.strip() or 'main'
            before = run_git(['rev-parse', 'HEAD']).stdout.strip()
            run_git(['fetch', 'origin'], timeout=120)
            pull = run_git(['pull', '--ff-only', 'origin', branch], timeout=120)
            after = run_git(['rev-parse', 'HEAD']).stdout.strip()
            changed_files: list[str] = []
            if before != after:
                changed_files = [
                    line.strip()
                    for line in run_git(
                        ['diff', '--name-only', f'{before}..{after}'],
                        timeout=60,
                    ).stdout.splitlines()
                    if line.strip()
                ]
            self._clear_agents_for_account(account_id)
            skills = get_bundled_skills(config.cwd)
            return {
                'branch': branch,
                'before': before,
                'after': after,
                'updated': before != after,
                'changed_files': changed_files,
                'skill_count': sum(1 for skill in skills if skill.user_invocable),
                'message': pull.stdout.strip() or pull.stderr.strip(),
            }

    def _build_agent(self, account_id: str | None = None) -> LocalCodingAgent:
        config = self._config_for(account_id)
        paths = self.account_paths(account_id)
        for directory in paths.values():
            if directory.name != '.venv':
                directory.mkdir(parents=True, exist_ok=True)
        self._ensure_python_env(paths['python_env'])
        permissions = AgentPermissions(
            allow_file_write=config.allow_write,
            allow_shell_commands=config.allow_shell,
        )
        runtime_config = AgentRuntimeConfig(
            cwd=config.cwd,
            permissions=permissions,
            stream_model_responses=True,
            session_directory=paths['sessions'],
            scratchpad_root=paths['scratchpad'],
            python_env_dir=paths['python_env'],
            enabled_skill_names=self.enabled_skill_names(account_id),
        )
        model_config = ModelConfig(
            model=config.model,
            base_url=config.base_url,
            api_key=config.api_key,
            timeout_seconds=config.timeout_seconds,
        )
        return LocalCodingAgent(
            model_config=model_config,
            runtime_config=runtime_config,
        )

    def agent_for(
        self,
        account_id: str | None = None,
        session_id: str | None = None,
    ) -> LocalCodingAgent:
        with self._lock:
            key = self._session_key(account_id, session_id)
            agent = self._agents.get(key)
            if agent is None:
                agent = self._build_agent(account_id)
                self._agents[key] = agent
            return agent

    def run_lock_for(
        self,
        account_id: str | None = None,
        session_id: str | None = None,
    ) -> Lock:
        key = self._session_key(account_id, session_id)
        with self._lock:
            lock = self._run_locks.get(key)
            if lock is None:
                lock = Lock()
                self._run_locks[key] = lock
            return lock

    def update(
        self,
        *,
        model: str | None = None,
        base_url: str | None = None,
        api_key: str | None = None,
        cwd: str | None = None,
        allow_shell: bool | None = None,
        allow_write: bool | None = None,
        account_id: str | None = None,
    ) -> None:
        with self._lock:
            config = self._config_for(account_id)
            if model is not None:
                normalized_model = _normalize_model_id(model)
                if normalized_model is None:
                    raise ValueError('model must not be empty')
                config = replace(config, model=normalized_model)
            if base_url is not None:
                config = replace(config, base_url=base_url)
            if api_key is not None:
                config = replace(config, api_key=api_key)
            if cwd is not None:
                resolved = Path(cwd).expanduser().resolve()
                if not resolved.is_dir():
                    raise ValueError(f'cwd does not exist: {resolved}')
                config = replace(config, cwd=resolved)
            if allow_shell is not None:
                config = replace(config, allow_shell=allow_shell)
            if allow_write is not None:
                config = replace(config, allow_write=allow_write)
            self._set_config_for(account_id, config)
            account_prefix = f'{self._account_key(account_id)}:'
            for key in list(self._agents):
                if key.startswith(account_prefix):
                    self._agents.pop(key, None)

    def snapshot(self, account_id: str | None = None) -> dict[str, Any]:
        with self._lock:
            config = self._config_for(account_id)
            agent = self.agent_for(account_id)
            paths = self.account_paths(account_id)
            return {
                'model': config.model,
                'base_url': config.base_url,
                'cwd': str(config.cwd),
                'account_id': _safe_account_id(account_id) if account_id else None,
                'account_directory': str(paths['base']),
                'session_directory': str(paths['sessions']),
                'upload_directory': str(paths['uploads']),
            'output_directory': str(paths['outputs']),
            'python_env_directory': str(paths['python_env']),
            'allow_shell': config.allow_shell,
                'allow_write': config.allow_write,
                'active_session_id': agent.active_session_id,
            }

    def lock(self) -> Lock:
        return self._lock

    def _ensure_python_env(self, env_dir: Path) -> None:
        python_bin = env_dir / 'bin' / 'python'
        pip_bin = env_dir / 'bin' / 'pip'
        if python_bin.exists() and pip_bin.exists():
            return
        env_dir.parent.mkdir(parents=True, exist_ok=True)
        venv.EnvBuilder(with_pip=True, symlinks=False).create(env_dir)


# ---------------------------------------------------------------------------
# Request models
# ---------------------------------------------------------------------------

class ChatRequest(BaseModel):
    prompt: str = Field(min_length=1)
    runtime_context: str | None = None
    resume_session_id: str | None = None
    account_id: str | None = None
    session_id: str | None = None


class RunCancelRequest(BaseModel):
    session_id: str = Field(min_length=1)
    account_id: str | None = None
    run_id: str | None = None


class StateUpdate(BaseModel):
    model: str | None = None
    base_url: str | None = None
    api_key: str | None = None
    cwd: str | None = None
    allow_shell: bool | None = None
    allow_write: bool | None = None
    account_id: str | None = None


class ModelListRequest(BaseModel):
    base_url: str | None = None
    api_key: str | None = None
    account_id: str | None = None


class JupyterWorkspaceBindRequest(BaseModel):
    session_id: str = Field(min_length=1)
    base_url: str = Field(min_length=1)
    password: str = Field(min_length=1)
    workspace_root: str = DEFAULT_JUPYTER_WORKSPACE_ROOT
    account_id: str | None = None


class SkillPreferenceUpdate(BaseModel):
    skill: str | None = None
    enabled: bool
    apply_all: bool = False
    account_id: str | None = None


class SkillSyncRequest(BaseModel):
    account_id: str | None = None


class SessionTitleUpdate(BaseModel):
    title: str = Field(min_length=1, max_length=80)


class FeishuAccountRequest(BaseModel):
    account_id: str | None = None


class FeishuOnlineDocRequest(BaseModel):
    path: str = Field(min_length=1)
    title: str | None = None
    folder_token: str | None = None
    account_id: str | None = None


def _append_runtime_context(current: str | None, addition: str) -> str:
    parts = [part.strip() for part in (current, addition) if part and part.strip()]
    return '\n\n'.join(parts)


# ---------------------------------------------------------------------------
# App factory
# ---------------------------------------------------------------------------

def create_app(state: AgentState) -> FastAPI:
    app = FastAPI(title='Claw Code GUI', version='1.0')

    # ------------- static + index ------------------------------------------
    app.mount(
        '/static',
        StaticFiles(directory=str(STATIC_DIR)),
        name='static',
    )

    @app.get('/', include_in_schema=False)
    async def root() -> FileResponse:
        return FileResponse(STATIC_DIR / 'index.html')

    # ------------- info ------------------------------------------------------
    @app.get('/api/state')
    async def get_state(account_id: str | None = None) -> dict[str, Any]:
        return state.snapshot(account_id)

    @app.post('/api/state')
    async def post_state(payload: StateUpdate) -> dict[str, Any]:
        try:
            state.update(**payload.model_dump(exclude_none=True))
        except ValueError as exc:
            raise HTTPException(status_code=400, detail=str(exc))
        return state.snapshot(payload.account_id)

    @app.get('/api/jupyter/session')
    async def get_jupyter_session(
        session_id: str,
        account_id: str | None = None,
    ) -> dict[str, Any]:
        safe_session_id = _safe_session_id(session_id)
        if not safe_session_id:
            raise HTTPException(status_code=400, detail='session_id is required')
        account_key = state._account_key(account_id)
        return state.jupyter_runtime_manager.session_payload(
            account_key,
            safe_session_id,
        )

    @app.post('/api/jupyter/bind-session')
    async def bind_jupyter_session(
        payload: JupyterWorkspaceBindRequest,
    ) -> dict[str, Any]:
        safe_session_id = _safe_session_id(payload.session_id)
        if not safe_session_id:
            raise HTTPException(status_code=400, detail='session_id is required')
        account_key = state._account_key(payload.account_id)
        try:
            runtime = await asyncio.to_thread(
                state.jupyter_runtime_manager.bind_session,
                account_id=account_key,
                session_id=safe_session_id,
                base_url=payload.base_url,
                password=payload.password,
                workspace_root=payload.workspace_root,
            )
        except JupyterRuntimeError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        except Exception as exc:
            raise HTTPException(
                status_code=502,
                detail=f'Unable to connect Jupyter runtime: {exc}',
            ) from exc
        return runtime.to_dict()

    @app.get('/api/slash-commands')
    async def list_slash_commands() -> list[dict[str, Any]]:
        commands: list[dict[str, Any]] = []
        for spec in get_slash_command_specs():
            if any(name in WEBUI_HIDDEN_SLASH_COMMANDS for name in spec.names):
                continue
            commands.append(
                {
                    'names': list(spec.names),
                    'primary': spec.names[0],
                    'description': WEBUI_SLASH_COMMAND_DESCRIPTIONS_ZH.get(
                        spec.names[0],
                        spec.description,
                    ),
                    'source': 'slash_command',
                    'webui_supported': True,
                }
            )
        return commands

    @app.get('/api/skills')
    async def list_skills(account_id: str | None = None) -> list[dict[str, Any]]:
        config = state.config_for(account_id)
        disabled = state._load_disabled_skill_names(account_id)
        return [
            {
                'name': skill.name,
                'description': skill.description,
                'when_to_use': skill.when_to_use,
                'aliases': list(skill.aliases),
                'allowed_tools': list(skill.allowed_tools),
                'enabled': skill.name.lower() not in disabled,
                'configurable': skill.name.lower() not in ALWAYS_ENABLED_HIDDEN_SKILL_NAMES,
            }
            for skill in get_bundled_skills(config.cwd)
            if skill.user_invocable
            and skill.name.lower() not in ALWAYS_ENABLED_HIDDEN_SKILL_NAMES
        ]

    @app.patch('/api/skills')
    async def update_skill_preference(payload: SkillPreferenceUpdate) -> list[dict[str, Any]]:
        try:
            if payload.apply_all:
                state.set_all_skills_enabled(payload.account_id, payload.enabled)
            elif payload.skill:
                state.set_skill_enabled(payload.account_id, payload.skill, payload.enabled)
            else:
                raise ValueError('skill must be provided unless apply_all is true')
        except ValueError as exc:
            raise HTTPException(status_code=400, detail=str(exc))
        return await list_skills(payload.account_id)

    @app.post('/api/skills/sync')
    async def sync_skills(payload: SkillSyncRequest) -> dict[str, Any]:
        try:
            result = await asyncio.to_thread(
                state.sync_skills_from_git,
                payload.account_id,
            )
        except (ValueError, subprocess.TimeoutExpired) as exc:
            raise HTTPException(status_code=400, detail=str(exc))
        result['skills'] = await list_skills(payload.account_id)
        return result

    @app.get('/api/models')
    async def list_models_get(account_id: str | None = None) -> dict[str, Any]:
        config = state.config_for(account_id)
        return _list_backend_models(config.base_url, config.api_key)

    @app.post('/api/models')
    async def list_models_post(payload: ModelListRequest) -> dict[str, Any]:
        config = state.config_for(payload.account_id)
        return _list_backend_models(
            payload.base_url or config.base_url,
            payload.api_key or config.api_key,
        )

    # ------------- Feishu integration ---------------------------------------
    @app.get('/api/integrations/feishu/status')
    async def feishu_status(account_id: str | None = None) -> dict[str, Any]:
        return _feishu_status_payload(state, account_id)

    @app.post('/api/integrations/feishu/login')
    async def feishu_login(payload: FeishuAccountRequest) -> dict[str, Any]:
        return _start_feishu_login(state, payload.account_id)

    @app.post('/api/integrations/feishu/logout')
    async def feishu_logout(payload: FeishuAccountRequest) -> dict[str, Any]:
        _stop_feishu_login(payload.account_id)
        paths = _feishu_paths(state, payload.account_id)
        result = _run_feishu_cli(paths, ['logout'], timeout_seconds=30)
        status = _feishu_status_payload(state, payload.account_id)
        status['logout_output'] = result['output']
        return status

    @app.post('/api/files/online-doc')
    async def create_feishu_online_doc(payload: FeishuOnlineDocRequest) -> dict[str, Any]:
        file_path = _resolve_account_file(state, payload.account_id, payload.path)
        suffix = file_path.suffix.lower()
        if suffix in FEISHU_UNSUPPORTED_DOCUMENT_SUFFIXES:
            raise HTTPException(
                status_code=400,
                detail='当前先不支持 json/jsonl 转在线文档。',
            )
        if suffix not in FEISHU_SUPPORTED_DOCUMENT_SUFFIXES:
            raise HTTPException(
                status_code=400,
                detail='当前仅支持 md、txt、csv、docx、xlsx 转在线文档。',
            )
        status = _feishu_status_payload(state, payload.account_id)
        if not status.get('logged_in'):
            raise HTTPException(
                status_code=409,
                detail={
                    'code': 'feishu_not_logged_in',
                    'message': '当前账号还没有完成飞书授权，请先授权后再转换。',
                    'status': status,
                },
            )
        paths = _feishu_paths(state, payload.account_id)
        runtime = MCPRuntime(servers=(_feishu_server_profile(paths),))
        title = _clean_feishu_doc_title(payload.title or file_path.stem)
        kind = 'sheet' if suffix in FEISHU_SPREADSHEET_SUFFIXES else 'doc'
        try:
            if kind == 'sheet':
                rendered, metadata = _create_feishu_online_spreadsheet(
                    runtime,
                    file_path,
                    title=title,
                    folder_token=payload.folder_token,
                )
            else:
                markdown = _convert_file_to_feishu_markdown(file_path, title=title)
                arguments: dict[str, Any] = {'title': title, 'markdown': markdown}
                if payload.folder_token and payload.folder_token.strip():
                    arguments['folder_token'] = payload.folder_token.strip()
                rendered, metadata = runtime.call_tool(
                    'doc_create',
                    arguments=arguments,
                    server_name=FEISHU_MCP_SERVER_NAME,
                    max_chars=API_TOOL_CONTENT_MAX_CHARS,
                    timeout_seconds=60.0,
                )
        except DataAgentInputError as exc:
            raise HTTPException(status_code=400, detail=str(exc))
        except OSError as exc:
            raise HTTPException(status_code=400, detail=f'读取文件失败: {exc}')
        except Exception as exc:
            target_name = '飞书表格' if kind == 'sheet' else '飞书文档'
            raise HTTPException(status_code=502, detail=f'{target_name}创建失败: {exc}')
        url = _extract_first_url(rendered)
        if url:
            _record_feishu_online_doc(
                state,
                payload.account_id,
                file_path=file_path,
                title=title,
                url=url,
                kind=kind,
            )
        return {
            'ok': True,
            'title': title,
            'url': url,
            'kind': kind,
            'file_path': str(file_path),
            'file_type': suffix.lstrip('.'),
            'result': rendered,
            'metadata': metadata,
        }

    # ------------- sessions --------------------------------------------------
    @app.get('/api/sessions')
    async def list_sessions(account_id: str | None = None) -> list[dict[str, Any]]:
        directory = state.account_paths(account_id)['sessions']
        if not directory.exists():
            return []
        results: list[dict[str, Any]] = []
        for path in sorted(_iter_session_files(directory), key=_session_mtime, reverse=True):
            try:
                data = json.loads(path.read_text(encoding='utf-8'))
            except (OSError, json.JSONDecodeError):
                continue
            session_id = str(data.get('session_id', _session_id_from_path(path)))
            usage = data.get('usage') if isinstance(data.get('usage'), dict) else {}
            model_config = (
                data.get('model_config')
                if isinstance(data.get('model_config'), dict)
                else {}
            )
            messages = data.get('messages') or []
            title = data.get('title')
            preview = ''
            for msg in messages:
                if isinstance(msg, dict) and msg.get('role') == 'user':
                    content = msg.get('content', '')
                    if isinstance(content, str) and not _is_internal_message(content):
                        preview = _strip_session_context(content)[:120]
                        break
            results.append(
                {
                    'session_id': session_id,
                    'turns': data.get('turns', 0),
                    'tool_calls': data.get('tool_calls', 0),
                    'preview': title if isinstance(title, str) and title.strip() else preview,
                    'modified_at': _session_mtime(path),
                    'model': model_config.get('model'),
                    'usage': usage,
                }
            )
        return results

    @app.get('/api/sessions/{session_id}')
    async def get_session(session_id: str, account_id: str | None = None) -> dict[str, Any]:
        directory = state.account_paths(account_id)['sessions']
        safe_id = _safe_session_id(session_id)
        if safe_id is None:
            raise HTTPException(status_code=404, detail='Session not found')
        try:
            stored = load_agent_session(safe_id, directory=directory)
        except FileNotFoundError:
            raise HTTPException(status_code=404, detail='Session not found')
        return _serialize_stored_session(stored)

    @app.delete('/api/sessions/{session_id}')
    async def delete_session(session_id: str, account_id: str | None = None) -> dict[str, Any]:
        directory = state.account_paths(account_id)['sessions']
        safe_id = _safe_session_id(session_id)
        if safe_id is None or not _delete_session_files(directory, safe_id):
            raise HTTPException(status_code=404, detail='Session not found')
        return {'deleted': True, 'session_id': safe_id}

    @app.patch('/api/sessions/{session_id}')
    async def update_session(
        session_id: str,
        payload: SessionTitleUpdate,
        account_id: str | None = None,
    ) -> dict[str, Any]:
        directory = state.account_paths(account_id)['sessions']
        safe_id = _safe_session_id(session_id)
        title = _clean_manual_session_title(payload.title)
        if safe_id is None or title is None:
            raise HTTPException(status_code=400, detail='Invalid session title')
        if not _update_session_title(directory, safe_id, title):
            raise HTTPException(status_code=404, detail='Session not found')
        return {'session_id': safe_id, 'title': title}

    @app.get('/api/context-budget')
    async def get_context_budget(
        session_id: str | None = None,
        account_id: str | None = None,
    ) -> dict[str, Any]:
        with state.lock():
            if session_id:
                directory = state.account_paths(account_id)['sessions']
                safe_id = _safe_session_id(session_id)
                if safe_id is None:
                    raise HTTPException(status_code=404, detail='Session not found')
                try:
                    stored = load_agent_session(safe_id, directory=directory)
                except FileNotFoundError:
                    raise HTTPException(status_code=404, detail='Session not found')
                session = _session_state_from_stored(stored)
                model = str(stored.model_config.get('model') or state.model)
                runtime_config = deserialize_runtime_config(stored.runtime_config)
            else:
                agent = state.agent_for(account_id)
                session = agent.last_session or agent.build_session()
                model = agent.model_config.model
                runtime_config = agent.runtime_config

            snapshot = calculate_token_budget(
                session=session,
                model=model,
                budget_config=runtime_config.budget_config,
                output_schema=runtime_config.output_schema,
            )
        return _serialize_token_budget(snapshot)

    @app.get('/api/runs/latest')
    async def latest_run(
        session_id: str,
        account_id: str | None = None,
    ) -> dict[str, Any]:
        safe_id = _safe_session_id(session_id)
        if safe_id is None:
            raise HTTPException(status_code=400, detail='Invalid session id')
        snapshot = state.run_manager.snapshot_latest(
            state._account_key(account_id),
            safe_id,
        )
        if snapshot is not None:
            return snapshot
        directory = state.account_paths(account_id)['sessions']
        try:
            stored = load_agent_session(safe_id, directory=directory)
        except FileNotFoundError:
            return {'session_id': safe_id, 'status': 'idle'}
        if _stored_session_has_incomplete_tail(stored):
            return {
                'session_id': safe_id,
                'status': 'interrupted',
                'current_stage': '上次运行已中断，后台没有正在执行的进程',
            }
        return {'session_id': safe_id, 'status': 'idle'}

    @app.post('/api/runs/cancel')
    async def cancel_run(payload: RunCancelRequest) -> dict[str, Any]:
        safe_id = _safe_session_id(payload.session_id)
        if safe_id is None:
            raise HTTPException(status_code=400, detail='Invalid session id')
        cancelled = (
            state.run_manager.cancel_run(payload.run_id)
            if payload.run_id
            else state.run_manager.cancel_latest(
                state._account_key(payload.account_id),
                safe_id,
            )
        )
        if cancelled:
            _mark_session_interrupted(
                state.account_paths(payload.account_id)['sessions'],
                safe_id,
                status='cancelled',
            )
        return {'session_id': safe_id, 'cancelled': cancelled}

    def _run_chat_payload(
        request: ChatRequest,
        event_sink: Any | None = None,
    ) -> dict[str, Any]:
        requested_session_id = _safe_session_id(
            request.resume_session_id or request.session_id
        ) or uuid4().hex
        account_key = state._account_key(request.account_id)
        prompt = request.prompt.strip()
        run_record = state.run_manager.start(account_key, requested_session_id, prompt)
        run_lock = state.run_lock_for(request.account_id, requested_session_id)
        agent = state.agent_for(request.account_id, requested_session_id)
        config = state.config_for(request.account_id)
        session_directory = state.account_paths(request.account_id)['sessions']
        jupyter_runtime = state.jupyter_runtime_manager.get(
            account_key,
            requested_session_id,
        )
        runtime_context = request.runtime_context
        if jupyter_runtime is not None:
            runtime_context = _append_runtime_context(
                runtime_context,
                jupyter_runtime.render_context(),
            )

        def emit_agent_event(event: dict[str, object]) -> None:
            stage = _runtime_event_stage(event)
            if stage:
                state.run_manager.update(run_record.run_id, stage=stage)
            state.run_manager.record_event(run_record.run_id, event)
            _emit_runtime_event(event_sink, event)

        previous_title = _read_session_title(
            session_directory,
            requested_session_id,
        )
        fallback_title = None
        if request.resume_session_id is None:
            fallback_title = _save_in_progress_session(
                directory=session_directory,
                agent=agent,
                session_id=requested_session_id,
                prompt=prompt,
            )
        if run_lock.locked():
            queued_event = {
                'type': 'run_queued',
                'run_id': run_record.run_id,
                'session_id': requested_session_id or '',
                'account_id': request.account_id or '',
            }
            state.run_manager.record_event(run_record.run_id, queued_event)
            _emit_runtime_event(
                event_sink,
                queued_event,
            )
        with run_lock:
            if run_record.cancel_event.is_set():
                state.run_manager.finish(run_record.run_id, 'cancelled')
                return {
                    'final_output': '已取消排队中的请求。',
                    'turns': 0,
                    'tool_calls': 0,
                    'transcript': (),
                    'session_id': requested_session_id,
                    'usage': {},
                    'total_cost_usd': 0.0,
                    'stop_reason': 'cancelled',
                }
            state.run_manager.update(run_record.run_id, status='running', stage='后端开始执行')
            started_event = {
                'type': 'run_started',
                'run_id': run_record.run_id,
                'started_at': run_record.started_at,
                'session_id': requested_session_id or '',
                'account_id': request.account_id or '',
            }
            state.run_manager.record_event(run_record.run_id, started_event)
            _emit_runtime_event(event_sink, started_event)
            agent.tool_context = replace(
                agent.tool_context,
                cancel_event=run_record.cancel_event,
                process_registry=run_record.process_registry,
                jupyter_runtime=jupyter_runtime,
            )
            started_at = time.perf_counter()
            try:
                try:
                    if request.resume_session_id is not None:
                        try:
                            stored = load_agent_session(
                                requested_session_id,
                                directory=session_directory,
                            )
                        except FileNotFoundError:
                            raise HTTPException(
                                status_code=404,
                                detail='Session to resume not found',
                            )
                        stored = _sanitize_stored_session_for_resume(stored)
                        result = agent.resume(
                            prompt,
                            stored,
                            runtime_context=runtime_context,
                            event_sink=emit_agent_event,
                        )
                    else:
                        result = agent.run(
                            prompt,
                            session_id=requested_session_id,
                            runtime_context=runtime_context,
                            event_sink=emit_agent_event,
                        )
                except Exception as exc:
                    _mark_session_interrupted(
                        session_directory,
                        requested_session_id,
                        status=(
                            'cancelled'
                            if run_record.cancel_event.is_set()
                            else 'failed'
                        ),
                        detail=str(exc),
                    )
                    state.run_manager.update(
                        run_record.run_id,
                        status=(
                            'cancelled'
                            if run_record.cancel_event.is_set()
                            else 'failed'
                        ),
                        error=str(exc),
                    )
                    raise
            finally:
                agent.tool_context = replace(
                    agent.tool_context,
                    cancel_event=None,
                    process_registry=None,
                    jupyter_runtime=None,
                )
            elapsed_ms = max(0, int((time.perf_counter() - started_at) * 1000))
            payload = _serialize_run_result(result)
            payload['elapsed_ms'] = elapsed_ms
            if result.session_id:
                _annotate_last_assistant_elapsed(
                    session_directory,
                    result.session_id,
                    elapsed_ms,
                )
                _ensure_session_title(
                    session_directory,
                    result.session_id,
                    model=config.model,
                    base_url=config.base_url,
                    api_key=config.api_key,
                    previous_title=previous_title,
                    fallback_title=fallback_title,
                )
                _annotate_transcript_elapsed(payload, elapsed_ms)
            state.run_manager.finish(
                run_record.run_id,
                'cancelled' if run_record.cancel_event.is_set() else 'completed',
            )
            if run_record.cancel_event.is_set() and result.session_id:
                _mark_session_interrupted(
                    session_directory,
                    result.session_id,
                    status='cancelled',
                )
            return payload

    # ------------- chat ------------------------------------------------------
    @app.post('/api/chat')
    async def chat(request: ChatRequest) -> dict[str, Any]:
        prompt = request.prompt.strip()
        if not prompt:
            raise HTTPException(status_code=400, detail='Prompt is empty')

        def _run() -> dict[str, Any]:
            return _run_chat_payload(request)

        try:
            payload = await asyncio.to_thread(_run)
        except HTTPException:
            raise
        except Exception as exc:  # surface the error in the UI
            return JSONResponse(
                status_code=500,
                content={
                    'error': str(exc),
                    'error_type': type(exc).__name__,
                },
            )
        return payload

    @app.post('/api/chat/stream')
    async def chat_stream(request: ChatRequest) -> StreamingResponse:
        prompt = request.prompt.strip()
        if not prompt:
            raise HTTPException(status_code=400, detail='Prompt is empty')

        event_queue: queue.Queue[dict[str, Any] | None] = queue.Queue()

        def emit_event(event: dict[str, object]) -> None:
            event_queue.put({'kind': 'event', 'event': event})

        def worker() -> None:
            try:
                payload = _run_chat_payload(request, event_sink=emit_event)
            except HTTPException as exc:
                event_queue.put(
                    {
                        'kind': 'error',
                        'status_code': exc.status_code,
                        'detail': exc.detail,
                    }
                )
            except Exception as exc:
                event_queue.put(
                    {
                        'kind': 'error',
                        'status_code': 500,
                        'error': str(exc),
                        'error_type': type(exc).__name__,
                    }
                )
            else:
                event_queue.put({'kind': 'result', 'payload': payload})
            finally:
                event_queue.put(None)

        threading.Thread(target=worker, daemon=True).start()

        stream_started = time.perf_counter()

        def generate() -> Any:
            while True:
                try:
                    item = event_queue.get(timeout=15)
                except queue.Empty:
                    yield json.dumps(
                        {
                            'kind': 'event',
                            'event': {
                                'type': 'server_heartbeat',
                                'elapsed_ms': max(
                                    0,
                                    int((time.perf_counter() - stream_started) * 1000),
                                ),
                            },
                        },
                        ensure_ascii=False,
                    ) + '\n'
                    continue
                if item is None:
                    break
                yield json.dumps(item, ensure_ascii=False) + '\n'

        return StreamingResponse(generate(), media_type='application/x-ndjson')

    @app.post('/api/clear')
    async def clear_state(account_id: str | None = None) -> dict[str, Any]:
        with state.lock():
            state.agent_for(account_id).clear_runtime_state()
        return state.snapshot(account_id)

    return app


# ---------------------------------------------------------------------------
# Serialization helpers
# ---------------------------------------------------------------------------

def _serialize_run_result(result: Any) -> dict[str, Any]:
    return {
        'final_output': result.final_output,
        'turns': result.turns,
        'tool_calls': result.tool_calls,
        'transcript': [_normalize_transcript_entry(entry) for entry in result.transcript],
        'session_id': result.session_id,
        'usage': result.usage.to_dict(),
        'total_cost_usd': result.total_cost_usd,
        'stop_reason': result.stop_reason,
    }


def _normalize_transcript_entry(entry: dict[str, Any]) -> dict[str, Any]:
    content = entry.get('content', '')
    if entry.get('role') == 'tool' and isinstance(content, str):
        content = _truncate_api_text(content, API_TOOL_CONTENT_MAX_CHARS)
    out: dict[str, Any] = {
        'role': entry.get('role', ''),
        'content': content,
    }
    for key in ('name', 'tool_call_id', 'tool_calls', 'metadata', 'message_id'):
        if key in entry and entry[key] not in (None, '', [], {}):
            out[key] = entry[key]
    for key in ('state', 'stop_reason'):
        if key in entry and entry[key] not in (None, ''):
            out[key] = entry[key]
    return out


def _truncate_api_text(text: str, limit: int) -> str:
    if len(text) <= limit:
        return text
    head = text[: limit // 2]
    tail = text[-(limit // 2) :]
    return f'{head}\n...[truncated for api response]...\n{tail}'


def _serialize_stored_session(stored: StoredAgentSession) -> dict[str, Any]:
    return {
        'session_id': stored.session_id,
        'turns': stored.turns,
        'tool_calls': stored.tool_calls,
        'messages': [_normalize_transcript_entry(dict(m)) for m in stored.messages],
        'usage': stored.usage,
        'total_cost_usd': stored.total_cost_usd,
        'model': stored.model_config.get('model'),
    }


def _stored_session_has_incomplete_tail(stored: StoredAgentSession) -> bool:
    _, changed = _trim_incomplete_message_tail(stored.messages)
    if changed:
        return True
    budget_state = stored.budget_state if isinstance(stored.budget_state, dict) else {}
    return budget_state.get('status') in {'queued', 'running'}


def _mark_session_interrupted(
    directory: Path,
    session_id: str,
    *,
    status: str,
    detail: str = '',
) -> None:
    try:
        stored = load_agent_session(session_id, directory=directory)
    except (FileNotFoundError, OSError, json.JSONDecodeError):
        return
    trimmed, changed = _trim_incomplete_message_tail(stored.messages)
    budget_state = (
        dict(stored.budget_state)
        if isinstance(stored.budget_state, dict)
        else {}
    )
    if not changed and budget_state.get('status') not in {'queued', 'running'}:
        return
    messages = list(trimmed)
    if not _last_message_is_run_status(messages, status):
        messages.append(
            {
                'role': 'assistant',
                'content': _interrupted_session_message(status),
                'state': 'final',
                'stop_reason': status,
                'metadata': {
                    'kind': 'run_status',
                    'status': status,
                    'detail': detail[:500],
                    'created_at_ms': int(time.time() * 1000),
                },
            }
        )
    budget_state['status'] = status
    budget_state['interrupted_at'] = int(time.time())
    save_agent_session(
        replace(
            stored,
            messages=tuple(messages),
            budget_state=budget_state,
        ),
        directory=directory,
    )


def _sanitize_stored_session_for_resume(
    stored: StoredAgentSession,
) -> StoredAgentSession:
    trimmed, changed = _trim_incomplete_message_tail(stored.messages)
    messages = _without_run_status_messages(trimmed)
    if len(messages) != len(trimmed):
        changed = True
    if not changed:
        return stored
    budget_state = (
        dict(stored.budget_state)
        if isinstance(stored.budget_state, dict)
        else {}
    )
    budget_state['status'] = 'interrupted'
    return replace(
        stored,
        messages=messages,
        budget_state=budget_state,
    )


def _trim_incomplete_message_tail(
    messages: tuple[dict[str, Any], ...] | tuple[Any, ...],
) -> tuple[tuple[dict[str, Any], ...], bool]:
    trimmed = [dict(message) for message in messages if isinstance(message, dict)]
    changed = False
    while trimmed and _is_incomplete_session_message(trimmed[-1]):
        trimmed.pop()
        changed = True
    while trimmed and _assistant_has_trailing_tool_call(trimmed[-1]):
        trimmed.pop()
        changed = True
    return tuple(trimmed), changed


def _is_incomplete_session_message(message: dict[str, Any]) -> bool:
    state = message.get('state')
    if isinstance(state, str) and state not in {'', 'final'}:
        return True
    metadata = message.get('metadata')
    if isinstance(metadata, dict):
        if (
            message.get('role') != 'user'
            and metadata.get('placeholder') is True
            and metadata.get('status') == 'running'
        ):
            return True
        if metadata.get('phase') in {'starting', 'running'} and message.get('role') == 'tool':
            return True
    return False


def _assistant_has_trailing_tool_call(message: dict[str, Any]) -> bool:
    if message.get('role') != 'assistant':
        return False
    tool_calls = message.get('tool_calls')
    return isinstance(tool_calls, list) and bool(tool_calls)


def _without_run_status_messages(
    messages: tuple[dict[str, Any], ...],
) -> tuple[dict[str, Any], ...]:
    return tuple(
        message
        for message in messages
        if not _is_run_status_message(message)
    )


def _is_run_status_message(message: dict[str, Any]) -> bool:
    metadata = message.get('metadata')
    return isinstance(metadata, dict) and metadata.get('kind') == 'run_status'


def _last_message_is_run_status(messages: list[dict[str, Any]], status: str) -> bool:
    if not messages:
        return False
    metadata = messages[-1].get('metadata')
    return (
        isinstance(metadata, dict)
        and metadata.get('kind') == 'run_status'
        and metadata.get('status') == status
    )


def _interrupted_session_message(status: str) -> str:
    if status == 'cancelled':
        return '上一次任务已取消，后台没有正在执行的进程。你可以继续回复，或重新发起任务。'
    if status == 'failed':
        return '上一次任务异常中断，后台没有正在执行的进程。你可以继续回复，或重新发起任务。'
    return '上一次任务已中断，后台没有正在执行的进程。你可以继续回复，或重新发起任务。'


def _runtime_event_stage(event: dict[str, object]) -> str:
    event_type = event.get('type')
    if event_type == 'tool_start':
        stage_note = str(event.get('assistant_content') or '').strip()
        if stage_note.startswith(('进度：', '进度:')):
            return stage_note
        tool_name = str(event.get('tool_name') or '').strip()
        return f'调用工具 {tool_name}' if tool_name else '正在调用工具'
    if event_type == 'tool_result':
        tool_name = str(event.get('tool_name') or '').strip()
        return f'工具完成 {tool_name}' if tool_name else '工具调用完成'
    if event_type == 'final_text_start':
        return '正在整理回复'
    if event_type == 'final_text_end':
        return '回复整理完成'
    if event_type == 'user_review_required':
        return '等待用户 review'
    if event_type == 'continuation_request':
        return '继续补全回复'
    if event_type in {
        'prompt_length_check',
        'prompt_length_recovery',
        'auto_compact_summary',
        'auto_compact_circuit_breaker',
    }:
        return '整理上下文'
    if event_type == 'task_budget_exceeded':
        return '达到任务预算限制'
    return ''


_RUN_EVENT_TYPES = {
    'run_queued',
    'run_started',
    'tool_start',
    'tool_result',
    'final_text_start',
    'final_text_end',
    'user_review_required',
    'continuation_request',
    'prompt_length_check',
    'prompt_length_recovery',
    'auto_compact_summary',
    'auto_compact_circuit_breaker',
    'task_budget_exceeded',
}


def _normalize_run_event(event: dict[str, object]) -> dict[str, Any] | None:
    event_type = event.get('type')
    if not isinstance(event_type, str) or event_type not in _RUN_EVENT_TYPES:
        return None
    allowed = {
        'type',
        'run_id',
        'session_id',
        'account_id',
        'tool_name',
        'tool_call_id',
        'arguments',
        'assistant_content',
        'message_id',
        'ok',
        'metadata',
        'reason',
        'continuation_index',
        'turn_index',
        'strategy',
        'projected_input_tokens',
        'soft_input_limit_tokens',
        'hard_input_limit_tokens',
        'exceeds_hard_limit',
        'exceeds_soft_limit',
    }
    normalized = {
        key: _json_safe_limited(value)
        for key, value in event.items()
        if key in allowed and value is not None
    }
    normalized['type'] = event_type
    return normalized


def _json_safe_limited(value: Any, *, depth: int = 0) -> Any:
    if depth > 4:
        return str(value)[:500]
    if isinstance(value, str):
        return value[:2000]
    if isinstance(value, (int, float, bool)) or value is None:
        return value
    if isinstance(value, dict):
        return {
            str(key)[:200]: _json_safe_limited(item, depth=depth + 1)
            for key, item in list(value.items())[:80]
        }
    if isinstance(value, (list, tuple)):
        return [_json_safe_limited(item, depth=depth + 1) for item in value[:80]]
    try:
        json.dumps(value)
        return value
    except TypeError:
        return str(value)[:1000]


def _emit_runtime_event(
    event_sink: Any | None,
    event: dict[str, object],
) -> None:
    if event_sink is None:
        return
    event_sink(event)


def _save_in_progress_session(
    *,
    directory: Path,
    agent: LocalCodingAgent,
    session_id: str | None,
    prompt: str,
) -> str | None:
    safe_id = _safe_session_id(session_id)
    if safe_id is None or not prompt:
        return None
    if _session_json_path(directory, safe_id).exists():
        return None

    # 新会话的第一轮执行可能很久。先写一个运行中占位，避免刷新页面后找不到会话。
    initial_title = _derive_initial_session_title(prompt)
    scratchpad_directory = (
        agent.runtime_config.scratchpad_root / safe_id / 'scratchpad'
    ).resolve()
    try:
        scratchpad_directory.mkdir(parents=True, exist_ok=True)
        session = agent.build_session(None, scratchpad_directory=scratchpad_directory)
        session.append_user(
            prompt,
            metadata={'status': 'running', 'placeholder': True},
            message_id='user_pending_0',
        )
        stored = StoredAgentSession(
            session_id=safe_id,
            model_config=serialize_model_config(agent.model_config),
            runtime_config=serialize_runtime_config(agent.runtime_config),
            system_prompt_parts=session.system_prompt_parts,
            user_context=dict(session.user_context),
            system_context=dict(session.system_context),
            messages=session.transcript(),
            turns=0,
            tool_calls=0,
            usage={},
            total_cost_usd=0.0,
            file_history=(),
            budget_state={'status': 'running'},
            plugin_state={},
            scratchpad_directory=str(scratchpad_directory),
        )
        saved_path = save_agent_session(stored, directory=directory)
        if initial_title:
            try:
                data = json.loads(saved_path.read_text(encoding='utf-8'))
            except (OSError, json.JSONDecodeError):
                return initial_title
            data['title'] = initial_title
            data['title_source'] = 'first_message'
            data['title_generated_at'] = int(time.time())
            _write_session_metadata(saved_path, data)
        return initial_title
    except OSError:
        return None


def _derive_initial_session_title(prompt: str) -> str | None:
    # 第一条消息刚发出时先给侧边栏一个可读标题，后续再由模型摘要精修。
    stripped = _strip_session_context(prompt)
    stripped = re.sub(r'\s+', ' ', stripped).strip()
    stripped = stripped.strip('"\'“”‘’`')
    if not stripped:
        return None
    title = _clean_session_title(stripped)
    return title or None


def _session_state_from_stored(stored: StoredAgentSession) -> AgentSessionState:
    return AgentSessionState(
        system_prompt_parts=stored.system_prompt_parts,
        user_context=stored.user_context,
        system_context=stored.system_context,
        messages=[
            AgentMessage.from_openai_message(message)
            for message in stored.messages
            if isinstance(message, dict)
        ],
    )


def _serialize_token_budget(snapshot: Any) -> dict[str, Any]:
    return {
        'model': snapshot.model,
        'context_window_tokens': snapshot.context_window_tokens,
        'projected_input_tokens': snapshot.projected_input_tokens,
        'message_tokens': snapshot.message_tokens,
        'chat_overhead_tokens': snapshot.chat_overhead_tokens,
        'reserved_output_tokens': snapshot.reserved_output_tokens,
        'reserved_compaction_buffer_tokens': snapshot.reserved_compaction_buffer_tokens,
        'reserved_schema_tokens': snapshot.reserved_schema_tokens,
        'hard_input_limit_tokens': snapshot.hard_input_limit_tokens,
        'soft_input_limit_tokens': snapshot.soft_input_limit_tokens,
        'overflow_tokens': snapshot.overflow_tokens,
        'soft_overflow_tokens': snapshot.soft_overflow_tokens,
        'exceeds_hard_limit': snapshot.exceeds_hard_limit,
        'exceeds_soft_limit': snapshot.exceeds_soft_limit,
        'token_counter_backend': snapshot.token_counter_backend,
        'token_counter_source': snapshot.token_counter_source,
        'token_counter_accurate': snapshot.token_counter_accurate,
    }


def _list_backend_models(base_url: str, api_key: str) -> dict[str, Any]:
    req = request.Request(
        _join_url(base_url, '/models'),
        headers={
            'Authorization': f'Bearer {api_key}',
            'api-key': api_key,
            'Content-Type': 'application/json',
        },
        method='GET',
    )
    try:
        with request.urlopen(req, timeout=10) as response:
            payload = json.loads(response.read().decode('utf-8'))
    except error.HTTPError as exc:
        detail = exc.read().decode('utf-8', errors='replace')
        raise HTTPException(
            status_code=502,
            detail=f'HTTP {exc.code} from model backend: {detail}',
        ) from exc
    except (error.URLError, OSError, json.JSONDecodeError) as exc:
        raise HTTPException(
            status_code=502,
            detail=f'Unable to list models from {base_url}: {exc}',
        ) from exc

    models = _normalize_model_list(payload)
    return {
        'models': models,
        'raw_count': _raw_model_count(payload),
        'filtered_count': len(models),
    }


def _normalize_model_list(payload: Any) -> list[dict[str, str]]:
    data = payload.get('data') if isinstance(payload, dict) else payload
    if not isinstance(data, list):
        return []
    models_by_key: dict[str, dict[str, str]] = {}
    for item in data:
        if isinstance(item, str):
            normalized = _normalize_model_id(item)
            if normalized is not None:
                models_by_key.setdefault(normalized.lower(), {'id': normalized})
            continue
        if not isinstance(item, dict):
            continue
        model_id = item.get('id') or item.get('model') or item.get('name')
        if not isinstance(model_id, str) or not model_id:
            continue
        model_type = _optional_model_string(item.get('model_type') or item.get('type'))
        if not _is_llm_model(model_id, model_type):
            continue
        provider = _optional_model_string(
            item.get('owned_by') or item.get('provider') or item.get('owner')
        )
        if provider and provider not in VALIDATED_CHAT_MODEL_PROVIDERS:
            continue
        normalized = _normalize_model_id(model_id, provider=provider)
        if normalized is not None and normalized.lower() in UNSUPPORTED_TOOL_CHAT_MODEL_IDS:
            continue
        if normalized is not None:
            entry = {'id': normalized}
            if provider:
                entry['provider'] = provider
            if model_type:
                entry['model_type'] = model_type
            models_by_key.setdefault(normalized.lower(), entry)
    return sorted(models_by_key.values(), key=lambda item: item['id'].lower())


def _raw_model_count(payload: Any) -> int:
    data = payload.get('data') if isinstance(payload, dict) else payload
    return len(data) if isinstance(data, list) else 0


def _optional_model_string(value: Any) -> str | None:
    if not isinstance(value, str):
        return None
    stripped = value.strip()
    return stripped or None


def _normalize_model_id(model: str, *, provider: str | None = None) -> str | None:
    value = model.strip()
    if not value:
        return None
    if provider and not value.lower().startswith(f'{provider.lower()}/'):
        return f'{provider}/{value}'
    return value


def _is_llm_model(model_id: str, model_type: str | None) -> bool:
    lower = model_id.strip().lower()
    blocked_fragments = (
        'asr',
        'audio',
        'embedding',
        'image_generation',
        'rerank',
        'speech',
        'text2image',
        'transcribe',
        'translation',
        'tts',
        'voiceclone',
        'voicedesign',
    )
    if any(fragment in lower for fragment in blocked_fragments):
        return False
    if model_type:
        return model_type.strip().lower() == 'llm'
    return True


def _join_url(base_url: str, suffix: str) -> str:
    return f'{base_url.rstrip("/")}/{suffix.lstrip("/")}'


def _annotate_transcript_elapsed(payload: dict[str, Any], elapsed_ms: int) -> None:
    transcript = payload.get('transcript')
    if not isinstance(transcript, list):
        return
    for entry in reversed(transcript):
        if not isinstance(entry, dict) or entry.get('role') != 'assistant':
            continue
        metadata = entry.get('metadata')
        if not isinstance(metadata, dict):
            metadata = {}
            entry['metadata'] = metadata
        metadata['elapsed_ms'] = elapsed_ms
        return


def _annotate_last_assistant_elapsed(
    directory: Path,
    session_id: str,
    elapsed_ms: int,
) -> None:
    path = _session_json_path(directory, session_id)
    try:
        data = json.loads(path.read_text(encoding='utf-8'))
    except (OSError, json.JSONDecodeError):
        return
    messages = data.get('messages')
    if not isinstance(messages, list):
        return
    for message in reversed(messages):
        if not isinstance(message, dict) or message.get('role') != 'assistant':
            continue
        metadata = message.get('metadata')
        if not isinstance(metadata, dict):
            metadata = {}
            message['metadata'] = metadata
        metadata['elapsed_ms'] = elapsed_ms
        try:
            path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding='utf-8')
        except OSError:
            return
        return


def _read_session_title(directory: Path, session_id: str | None) -> str | None:
    if not session_id:
        return None
    path = _session_json_path(directory, session_id)
    try:
        data = json.loads(path.read_text(encoding='utf-8'))
    except (OSError, json.JSONDecodeError):
        return None
    title = data.get('title')
    if isinstance(title, str) and title.strip():
        return title.strip()
    return None


def _ensure_session_title(
    directory: Path,
    session_id: str,
    *,
    model: str,
    base_url: str,
    api_key: str,
    previous_title: str | None = None,
    fallback_title: str | None = None,
) -> None:
    path = _session_json_path(directory, session_id)
    try:
        data = json.loads(path.read_text(encoding='utf-8'))
    except (OSError, json.JSONDecodeError):
        return
    if previous_title:
        data['title'] = previous_title
        _write_session_metadata(path, data)
        return
    title = data.get('title')
    title_source = data.get('title_source')
    if (
        isinstance(title, str)
        and title.strip()
        and title_source != 'first_message'
    ):
        return
    messages = data.get('messages')
    if not isinstance(messages, list):
        if fallback_title and not (isinstance(title, str) and title.strip()):
            data['title'] = fallback_title
            data['title_source'] = 'first_message'
            data['title_generated_at'] = int(time.time())
            _write_session_metadata(path, data)
        return
    user_messages = _session_user_messages(messages)
    if not user_messages:
        if fallback_title and not (isinstance(title, str) and title.strip()):
            data['title'] = fallback_title
            data['title_source'] = 'first_message'
            data['title_generated_at'] = int(time.time())
            _write_session_metadata(path, data)
        return
    generated = _generate_session_title(
        user_messages,
        model=model,
        base_url=base_url,
        api_key=api_key,
    )
    if not generated:
        if fallback_title and not (isinstance(title, str) and title.strip()):
            data['title'] = fallback_title
            data['title_source'] = 'first_message'
            data['title_generated_at'] = int(time.time())
            _write_session_metadata(path, data)
        return
    data['title'] = generated
    data['title_source'] = 'llm'
    data['title_generated_at'] = int(time.time())
    _write_session_metadata(path, data)


def _session_user_messages(messages: list[Any]) -> list[str]:
    user_messages: list[str] = []
    for message in messages:
        if not isinstance(message, dict) or message.get('role') != 'user':
            continue
        content = message.get('content')
        if not isinstance(content, str) or _is_internal_message(content):
            continue
        stripped = _strip_session_context(content)
        if stripped:
            user_messages.append(stripped)
    return user_messages


def _generate_session_title(
    user_messages: list[str],
    *,
    model: str,
    base_url: str,
    api_key: str,
) -> str | None:
    conversation = '\n'.join(
        f'用户第{index}轮：{message[:240]}'
        for index, message in enumerate(user_messages[-6:], start=1)
    )
    payload = {
        'model': model,
        'temperature': 0.2,
        'max_tokens': 32,
        'messages': [
            {
                'role': 'system',
                'content': (
                    '你是会话标题生成器。请根据用户对话生成一个中文短标题，'
                    '要求 4 到 12 个字，只输出标题，不要解释，不要引号。'
                ),
            },
            {'role': 'user', 'content': conversation},
        ],
    }
    client = OpenAICompatClient(
        ModelConfig(
            model=model,
            base_url=base_url,
            api_key=api_key,
            temperature=0.2,
        )
    )
    try:
        turn = client.complete(payload['messages'], tools=[])
    except OpenAICompatError:
        return None
    return _clean_session_title(turn.content)


def _clean_session_title(value: str) -> str | None:
    title = ' '.join(value.strip().strip('"\'“”‘’`').split())
    title = title.rstrip('。.!！?？')
    if not title:
        return None
    return title[:24].rstrip()


def _clean_manual_session_title(value: str) -> str | None:
    title = ' '.join(value.strip().strip('"\'“”‘’`').split())
    if not title:
        return None
    return title[:80]


def _write_session_metadata(path: Path, data: dict[str, Any]) -> None:
    try:
        path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding='utf-8')
    except OSError:
        return


def _session_json_path(directory: Path, session_id: str) -> Path:
    path = directory / session_id / 'session.json'
    if path.exists():
        return path
    return directory / f'{session_id}.json'


def _feishu_paths(state: AgentState, account_id: str | None) -> dict[str, Path]:
    base = state.account_paths(account_id)['base'] / 'integrations' / 'feishu'
    paths = {
        'root': base,
        'home': base / 'home',
        'config': base / 'config',
        'npm_cache': base / 'npm-cache',
    }
    for path in paths.values():
        path.mkdir(parents=True, exist_ok=True)
    return paths


def _feishu_command() -> list[str]:
    return [
        _feishu_npx_executable(),
        '-y',
        f'--registry={FEISHU_NPM_REGISTRY}',
        FEISHU_MCP_PACKAGE,
    ]


def _feishu_npx_executable() -> str:
    explicit = os.environ.get('CLAW_NPX_BIN')
    if explicit and Path(explicit).expanduser().exists():
        return str(Path(explicit).expanduser())
    node_bin_dir = os.environ.get('CLAW_NODE_BIN_DIR')
    if node_bin_dir:
        candidate = Path(node_bin_dir).expanduser() / 'npx'
        if candidate.exists():
            return str(candidate)
    return shutil.which('npx') or 'npx'


def _feishu_env(paths: dict[str, Path]) -> dict[str, str]:
    env = {
        'HOME': str(paths['home']),
        'XDG_CONFIG_HOME': str(paths['config']),
        'npm_config_cache': str(paths['npm_cache']),
        'npm_config_update_notifier': 'false',
        'NO_UPDATE_NOTIFIER': 'true',
        'FEISHU_LOGIN_MODE': 'devicecode',
    }
    node_bin_dir = os.environ.get('CLAW_NODE_BIN_DIR')
    if node_bin_dir:
        env['PATH'] = f'{node_bin_dir}{os.pathsep}{os.environ.get("PATH", "")}'
    return env


def _run_feishu_cli(
    paths: dict[str, Path],
    args: list[str],
    *,
    timeout_seconds: float,
) -> dict[str, Any]:
    env = os.environ.copy()
    env.update(_feishu_env(paths))
    try:
        completed = subprocess.run(
            [*_feishu_command(), *args],
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
            cwd=str(paths['root']),
            env=env,
            timeout=timeout_seconds,
            check=False,
        )
        return {
            'returncode': completed.returncode,
            'output': completed.stdout.strip(),
        }
    except FileNotFoundError as exc:
        return {
            'returncode': 127,
            'output': f'缺少命令: {exc.filename}',
        }
    except subprocess.TimeoutExpired as exc:
        output = exc.stdout or ''
        return {
            'returncode': 124,
            'output': str(output).strip() or '飞书状态检查超时',
        }


def _feishu_status_payload(state: AgentState, account_id: str | None) -> dict[str, Any]:
    paths = _feishu_paths(state, account_id)
    _reap_feishu_login(account_id)
    result = _run_feishu_cli(paths, ['status'], timeout_seconds=30)
    output = str(result.get('output') or '')
    lowered = output.lower()
    logged_in = result.get('returncode') == 0 and not any(
        marker in lowered
        for marker in ('not logged in', '未登录', 'no credentials')
    )
    pending = _feishu_login_snapshot(account_id)
    payload: dict[str, Any] = {
        'logged_in': logged_in,
        'status': 'logged_in' if logged_in else 'not_logged_in',
        'output': output,
    }
    if pending:
        payload['login'] = pending
        if not logged_in:
            payload['status'] = 'login_pending'
    if not logged_in and result.get('returncode') not in (0, 1):
        payload['status'] = 'error'
        payload['error'] = output or '无法获取飞书登录状态'
    return payload


def _start_feishu_login(state: AgentState, account_id: str | None) -> dict[str, Any]:
    status = _feishu_status_payload(state, account_id)
    if status.get('logged_in'):
        return status

    key = _feishu_login_key(account_id)
    with _FEISHU_LOGIN_LOCK:
        existing = _FEISHU_LOGIN_PROCESSES.get(key)
        if existing is not None and existing.process.poll() is None:
            return {
                'logged_in': False,
                'status': 'login_pending',
                'login': _feishu_login_snapshot_unlocked(existing),
            }
        _FEISHU_LOGIN_PROCESSES.pop(key, None)

    paths = _feishu_paths(state, account_id)
    env = os.environ.copy()
    env.update(_feishu_env(paths))
    try:
        process = subprocess.Popen(
            [*_feishu_command(), 'login', '--device'],
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
            bufsize=1,
            cwd=str(paths['root']),
            env=env,
        )
    except FileNotFoundError as exc:
        raise HTTPException(status_code=500, detail=f'缺少命令: {exc.filename}')

    entry = FeishuLoginProcess(process=process, started_at=time.time())
    with _FEISHU_LOGIN_LOCK:
        _FEISHU_LOGIN_PROCESSES[key] = entry

    reader = threading.Thread(
        target=_read_feishu_login_output,
        args=(key, entry),
        name=f'feishu-login-{key}',
        daemon=True,
    )
    reader.start()

    deadline = time.time() + 5.0
    while time.time() < deadline:
        snapshot = _feishu_login_snapshot(account_id)
        if snapshot and snapshot.get('login_url'):
            return {
                'logged_in': False,
                'status': 'login_pending',
                'login': snapshot,
            }
        if process.poll() is not None:
            break
        time.sleep(0.1)

    snapshot = _feishu_login_snapshot(account_id)
    return {
        'logged_in': False,
        'status': 'login_pending' if process.poll() is None else 'login_failed',
        'login': snapshot,
    }


def _stop_feishu_login(account_id: str | None) -> None:
    key = _feishu_login_key(account_id)
    with _FEISHU_LOGIN_LOCK:
        entry = _FEISHU_LOGIN_PROCESSES.pop(key, None)
    if entry is None:
        return
    process = entry.process
    if process.poll() is None:
        process.terminate()
        try:
            process.wait(timeout=2.0)
        except subprocess.TimeoutExpired:
            process.kill()
            process.wait(timeout=2.0)


def _reap_feishu_login(account_id: str | None) -> None:
    key = _feishu_login_key(account_id)
    with _FEISHU_LOGIN_LOCK:
        entry = _FEISHU_LOGIN_PROCESSES.get(key)
        if entry is not None and entry.process.poll() is not None:
            _FEISHU_LOGIN_PROCESSES.pop(key, None)


def _read_feishu_login_output(key: str, entry: FeishuLoginProcess) -> None:
    stream = entry.process.stdout
    if stream is None:
        return
    for line in stream:
        with _FEISHU_LOGIN_LOCK:
            current = _FEISHU_LOGIN_PROCESSES.get(key)
            if current is not entry:
                return
            entry.output.append(line.rstrip())
            text = '\n'.join(entry.output)
            login_url, user_code = _parse_feishu_login_details(text)
            if login_url:
                entry.login_url = login_url
            if user_code:
                entry.user_code = user_code


def _feishu_login_snapshot(account_id: str | None) -> dict[str, Any] | None:
    key = _feishu_login_key(account_id)
    with _FEISHU_LOGIN_LOCK:
        entry = _FEISHU_LOGIN_PROCESSES.get(key)
        if entry is None:
            return None
        return _feishu_login_snapshot_unlocked(entry)


def _feishu_login_snapshot_unlocked(entry: FeishuLoginProcess) -> dict[str, Any]:
    output = '\n'.join(entry.output[-20:])
    if not entry.login_url or not entry.user_code:
        login_url, user_code = _parse_feishu_login_details(output)
        entry.login_url = entry.login_url or login_url
        entry.user_code = entry.user_code or user_code
    return {
        'running': entry.process.poll() is None,
        'started_at': entry.started_at,
        'elapsed_seconds': max(0, int(time.time() - entry.started_at)),
        'login_url': entry.login_url,
        'user_code': entry.user_code,
        'output': output,
    }


def _parse_feishu_login_details(text: str) -> tuple[str | None, str | None]:
    urls = re.findall(r'https?://[^\s]+', text)
    login_url = next((url.rstrip('.,;') for url in urls if 'feishu' in url.lower()), None)
    user_code = None
    match = re.search(r'user_code=([A-Za-z0-9_-]+)', login_url or '')
    if match:
        user_code = match.group(1)
    if not user_code:
        match = re.search(r'\b([A-Z0-9]{4}-[A-Z0-9]{4})\b', text)
        if match:
            user_code = match.group(1)
    return login_url, user_code


def _feishu_login_key(account_id: str | None) -> str:
    return _safe_account_id(account_id)


def _feishu_server_profile(paths: dict[str, Path]) -> MCPServerProfile:
    return MCPServerProfile(
        name=FEISHU_MCP_SERVER_NAME,
        source_manifest='builtin:feishu-mcp-pro',
        transport='stdio',
        command='npx',
        args=(
            '-y',
            f'--registry={FEISHU_NPM_REGISTRY}',
            FEISHU_MCP_PACKAGE,
        ),
        env=_feishu_env(paths),
        cwd=str(paths['root']),
        description='账号隔离的飞书 MCP Pro。',
    )


def _record_feishu_online_doc(
    state: AgentState,
    account_id: str | None,
    *,
    file_path: Path,
    title: str,
    url: str,
    kind: str = 'doc',
) -> None:
    paths = _feishu_paths(state, account_id)
    map_path = paths['root'] / FEISHU_ONLINE_DOCS_FILENAME
    now = int(time.time())
    clean_url = _clean_url(url)
    with _FEISHU_DOC_MAP_LOCK:
        try:
            payload = json.loads(map_path.read_text(encoding='utf-8'))
        except (OSError, json.JSONDecodeError):
            payload = {}
        files = payload.get('files') if isinstance(payload, dict) else None
        if not isinstance(files, dict):
            files = {}
        previous = files.get(str(file_path))
        created_at = (
            previous.get('created_at')
            if isinstance(previous, dict) and isinstance(previous.get('created_at'), int)
            else now
        )
        files[str(file_path)] = {
            'url': clean_url,
            'title': title,
            'kind': kind,
            'file_path': str(file_path),
            'created_at': created_at,
            'updated_at': now,
        }
        map_path.write_text(
            json.dumps({'files': files}, ensure_ascii=False, indent=2),
            encoding='utf-8',
        )


def _resolve_account_file(state: AgentState, account_id: str | None, raw_path: str) -> Path:
    path = Path(raw_path).expanduser().resolve()
    account_base = state.account_paths(account_id)['base'].resolve()
    try:
        path.relative_to(account_base)
    except ValueError:
        raise HTTPException(status_code=403, detail='文件不在当前账号目录内')
    if not path.exists():
        raise HTTPException(status_code=404, detail='文件不存在')
    if not path.is_file():
        raise HTTPException(status_code=400, detail='目标不是文件')
    return path


def _clean_feishu_doc_title(value: str) -> str:
    title = re.sub(r'[\r\n\t/\\]+', ' ', value).strip()
    return title[:80].strip() or 'Claw 生成文档'


def _create_feishu_online_spreadsheet(
    runtime: MCPRuntime,
    file_path: Path,
    *,
    title: str,
    folder_token: str | None = None,
) -> tuple[str, dict[str, Any]]:
    sheets = _load_file_as_feishu_sheets(file_path)
    create_params: dict[str, Any] = {'title': title}
    if folder_token and folder_token.strip():
        create_params['folder_token'] = folder_token.strip()
    rendered, metadata = _call_feishu_mcp_tool_checked(
        runtime,
        'sheet_ops',
        arguments={'action': 'create', 'params': create_params},
        max_chars=API_TOOL_CONTENT_MAX_CHARS,
        timeout_seconds=60.0,
    )
    spreadsheet_token = _extract_feishu_spreadsheet_token(rendered)
    if not spreadsheet_token:
        raise DataAgentInputError('飞书表格已创建，但没有解析到 spreadsheet token')

    for index, sheet_data in enumerate(sheets):
        rows = sheet_data['rows']
        if not rows:
            continue
        if index == 0:
            # 新建飞书表格默认会带一个名为 Sheet1 的工作表。
            # 不从 spreadsheet meta 里取顶层 title，避免把文件名误当成工作表名。
            sheet_ref = 'Sheet1'
        else:
            sheet_ref = _create_feishu_worksheet(
                runtime,
                spreadsheet_token,
                title=str(sheet_data.get('title') or f'Sheet{index + 1}'),
                index=index,
            )
        _write_feishu_sheet_rows(runtime, spreadsheet_token, sheet_ref, rows)
    return rendered, metadata


def _call_feishu_mcp_tool_checked(
    runtime: MCPRuntime,
    tool_name: str,
    *,
    arguments: dict[str, Any],
    max_chars: int,
    timeout_seconds: float,
) -> tuple[str, dict[str, Any]]:
    rendered, metadata = runtime.call_tool(
        tool_name,
        arguments=arguments,
        server_name=FEISHU_MCP_SERVER_NAME,
        max_chars=max_chars,
        timeout_seconds=timeout_seconds,
    )
    error_message = _extract_feishu_mcp_error(rendered)
    if metadata.get('is_error') or error_message:
        raise DataAgentInputError(error_message or f'飞书 MCP 工具 {tool_name} 调用失败')
    return rendered, metadata


def _extract_feishu_mcp_error(text: str) -> str | None:
    payload = _parse_first_json_object(text)
    if not isinstance(payload, dict):
        return None
    error_value = payload.get('error')
    if not error_value:
        return None
    suggestion = payload.get('suggestion')
    code = payload.get('code')
    parts = [str(error_value)]
    if code is not None:
        parts.append(f'code={code}')
    if suggestion:
        parts.append(str(suggestion))
    return '；'.join(parts)


def _load_file_as_feishu_sheets(file_path: Path) -> list[dict[str, Any]]:
    suffix = file_path.suffix.lower()
    if suffix == '.csv':
        rows = _load_csv_rows_for_feishu(file_path)
        return [{'title': _clean_feishu_sheet_title(file_path.stem) or 'Sheet1', 'rows': rows}]
    if suffix == '.xlsx':
        return _load_xlsx_rows_for_feishu(file_path)
    raise DataAgentInputError(f'不支持转为飞书表格的文件类型: {suffix}')


def _load_csv_rows_for_feishu(file_path: Path) -> list[list[str]]:
    rows: list[list[str]] = []
    with file_path.open('r', encoding='utf-8-sig', errors='replace', newline='') as handle:
        reader = csv.reader(handle)
        for row_index, row in enumerate(reader):
            if row_index >= FEISHU_SPREADSHEET_MAX_ROWS:
                break
            rows.append(_clean_feishu_sheet_row(row))
    normalized_rows = _normalize_feishu_sheet_rows(rows)
    if not normalized_rows:
        raise DataAgentInputError('没有解析到可转成飞书表格的数据')
    return normalized_rows


def _load_xlsx_rows_for_feishu(file_path: Path) -> list[dict[str, Any]]:
    try:
        import openpyxl  # type: ignore[import-not-found]
    except ImportError as exc:
        raise DataAgentInputError('解析 xlsx 需要 openpyxl') from exc
    workbook = openpyxl.load_workbook(file_path, data_only=True, read_only=True)
    sheets: list[dict[str, Any]] = []
    for index, sheet in enumerate(workbook.worksheets):
        if hasattr(sheet, 'reset_dimensions'):
            sheet.reset_dimensions()
        rows: list[list[str]] = []
        for row_index, row in enumerate(sheet.iter_rows(values_only=True)):
            if row_index >= FEISHU_SPREADSHEET_MAX_ROWS:
                break
            rows.append(_clean_feishu_sheet_row(row))
        normalized_rows = _normalize_feishu_sheet_rows(rows)
        if normalized_rows:
            sheets.append(
                {
                    'title': _clean_feishu_sheet_title(str(sheet.title)) or f'Sheet{index + 1}',
                    'rows': normalized_rows,
                }
            )
    if not sheets:
        raise DataAgentInputError('没有解析到可转成飞书表格的数据')
    return sheets


def _clean_feishu_sheet_row(row: Any) -> list[str]:
    if not isinstance(row, (list, tuple)):
        return []
    return [_clean_feishu_sheet_cell(cell) for cell in row[:FEISHU_SPREADSHEET_MAX_COLS]]


def _clean_feishu_sheet_cell(value: Any) -> str:
    if value is None:
        return ''
    text = str(value)
    if len(text) > FEISHU_SPREADSHEET_MAX_CELL_CHARS:
        return text[:FEISHU_SPREADSHEET_MAX_CELL_CHARS]
    return text


def _normalize_feishu_sheet_rows(rows: list[list[str]]) -> list[list[str]]:
    while rows and not any(cell for cell in rows[-1]):
        rows.pop()
    if not rows:
        return []
    width = max((len(row) for row in rows), default=0)
    width = min(max(width, 1), FEISHU_SPREADSHEET_MAX_COLS)
    return [row[:width] + [''] * max(0, width - len(row)) for row in rows]


def _clean_feishu_sheet_title(value: str) -> str:
    title = re.sub(r'[\r\n\t/\\?*\[\]:]+', ' ', value).strip()
    return title[:80].strip()


def _extract_feishu_spreadsheet_token(text: str) -> str | None:
    url = _extract_first_url(text)
    candidates = [item for item in (url, text) if item]
    for candidate in candidates:
        match = re.search(r'/sheets/([A-Za-z0-9]+)', candidate)
        if match:
            return match.group(1)
    payload = _parse_first_json_object(text)
    token = _find_first_string_value(
        payload,
        {'spreadsheet_token', 'spreadsheetToken', 'token'},
    )
    return token.strip() if token else None


def _discover_feishu_sheet_refs(runtime: MCPRuntime, spreadsheet_token: str) -> list[str]:
    try:
        rendered, _metadata = _call_feishu_mcp_tool_checked(
            runtime,
            'sheet_ops',
            arguments={'action': 'meta', 'params': {'url_or_token': spreadsheet_token}},
            max_chars=API_TOOL_CONTENT_MAX_CHARS,
            timeout_seconds=60.0,
        )
    except Exception:
        return []
    refs = _extract_feishu_sheet_refs(rendered)
    return refs


def _extract_feishu_sheet_refs(text: str) -> list[str]:
    payload = _parse_first_json_object(text)
    refs: list[str] = []

    def visit(value: Any) -> None:
        if isinstance(value, dict):
            raw_id = value.get('sheet_id') or value.get('sheetId')
            raw_title = value.get('title') or value.get('name')
            if isinstance(raw_id, str) and raw_id.strip():
                refs.append(raw_id.strip())
            elif (
                isinstance(raw_title, str)
                and raw_title.strip()
                and ('index' in value or 'row_count' in value or 'col_count' in value)
            ):
                refs.append(raw_title.strip())
            for child in value.values():
                visit(child)
        elif isinstance(value, list):
            for child in value:
                visit(child)

    visit(payload)
    return _dedupe_strings([ref for ref in refs if ref])


def _dedupe_strings(values: list[str]) -> list[str]:
    seen: set[str] = set()
    result: list[str] = []
    for value in values:
        if value in seen:
            continue
        seen.add(value)
        result.append(value)
    return result


def _create_feishu_worksheet(
    runtime: MCPRuntime,
    spreadsheet_token: str,
    *,
    title: str,
    index: int,
) -> str:
    sheet_title = _clean_feishu_sheet_title(title) or f'Sheet{index + 1}'
    rendered, _metadata = _call_feishu_mcp_tool_checked(
        runtime,
        'sheet_ops',
        arguments={
            'action': 'add_sheet',
            'params': {
                'spreadsheet_token': spreadsheet_token,
                'title': sheet_title,
                'index': index,
            },
        },
        max_chars=API_TOOL_CONTENT_MAX_CHARS,
        timeout_seconds=60.0,
    )
    refs = _extract_feishu_sheet_refs(rendered)
    return refs[0] if refs else sheet_title


def _write_feishu_sheet_rows(
    runtime: MCPRuntime,
    spreadsheet_token: str,
    sheet_ref: str,
    rows: list[list[str]],
) -> None:
    if not rows:
        return
    width = max(len(row) for row in rows)
    end_column = _spreadsheet_column_name(width)
    for start in range(0, len(rows), FEISHU_SPREADSHEET_WRITE_BATCH_ROWS):
        chunk = rows[start : start + FEISHU_SPREADSHEET_WRITE_BATCH_ROWS]
        start_row = start + 1
        end_row = start + len(chunk)
        _call_feishu_mcp_tool_checked(
            runtime,
            'sheet_ops',
            arguments={
                'action': 'write',
                'params': {
                    'spreadsheet_token': spreadsheet_token,
                    'range': f'{sheet_ref}!A{start_row}:{end_column}{end_row}',
                    'values': json.dumps(chunk, ensure_ascii=False),
                },
            },
            max_chars=API_TOOL_CONTENT_MAX_CHARS,
            timeout_seconds=60.0,
        )


def _spreadsheet_column_name(index: int) -> str:
    if index <= 0:
        raise ValueError('index must be positive')
    name = ''
    current = index
    while current:
        current, remainder = divmod(current - 1, 26)
        name = chr(ord('A') + remainder) + name
    return name


def _parse_first_json_object(text: str) -> Any:
    stripped = text.strip()
    candidates = [stripped]
    json_match = re.search(r'(\{.*\})', stripped, flags=re.DOTALL)
    if json_match:
        candidates.append(json_match.group(1))
    for candidate in candidates:
        try:
            return json.loads(candidate)
        except json.JSONDecodeError:
            continue
    return None


def _find_first_string_value(value: Any, keys: set[str]) -> str | None:
    if isinstance(value, dict):
        for key in keys:
            item = value.get(key)
            if isinstance(item, str) and item.strip():
                return item
        for child in value.values():
            found = _find_first_string_value(child, keys)
            if found:
                return found
    elif isinstance(value, list):
        for child in value:
            found = _find_first_string_value(child, keys)
            if found:
                return found
    return None


def _convert_file_to_feishu_markdown(file_path: Path, *, title: str) -> str:
    suffix = file_path.suffix.lower()
    if suffix == '.md':
        body = file_path.read_text(encoding='utf-8', errors='replace')
        return _limit_feishu_markdown(f'# {title}\n\n> 来源文件：{file_path.name}\n\n{body}')
    if suffix == '.txt':
        body = file_path.read_text(encoding='utf-8', errors='replace')
        return _limit_feishu_markdown(f'# {title}\n\n> 来源文件：{file_path.name}\n\n{body}')

    loaded = load_input_sources(
        file_path.parent,
        [file_path.name],
        max_files=1,
        max_paragraphs_per_file=200,
        max_tables_per_file=30,
        max_rows_per_table=120,
        max_cell_chars=500,
    )
    sources = loaded.get('sources')
    if not isinstance(sources, list) or not sources:
        raise DataAgentInputError('没有解析到可转成在线文档的内容')
    source = sources[0]
    lines = [
        f'# {title}',
        '',
        f'> 来源文件：{file_path.name}',
        f'> 文件类型：{suffix.lstrip(".")}',
    ]
    warnings = source.get('warnings')
    if isinstance(warnings, list) and warnings:
        lines.append(f'> 解析提示：{"；".join(str(item) for item in warnings)}')
    for paragraph in source.get('paragraphs', []):
        if not isinstance(paragraph, dict):
            continue
        text = str(paragraph.get('text') or '').strip()
        if text:
            lines.extend(['', text])
    for table in source.get('tables', []):
        if not isinstance(table, dict):
            continue
        rows = table.get('rows')
        if not isinstance(rows, list) or not rows:
            continue
        title_text = str(table.get('title') or '表格').strip() or '表格'
        lines.extend(['', f'## {title_text}', ''])
        lines.append(_render_markdown_table(rows))
        row_count = table.get('row_count')
        if isinstance(row_count, int) and row_count > len(rows):
            lines.append(f'\n> 仅展示前 {len(rows)} 行，原表约 {row_count} 行。')
    markdown = '\n'.join(lines).strip()
    if not markdown:
        raise DataAgentInputError('没有解析到可转成在线文档的内容')
    return _limit_feishu_markdown(markdown)


def _render_markdown_table(raw_rows: list[Any]) -> str:
    rows = [
        [str(cell) for cell in row]
        for row in raw_rows
        if isinstance(row, list)
    ]
    if not rows:
        return ''
    width = max(len(row) for row in rows)
    normalized = [row + [''] * (width - len(row)) for row in rows]
    header = normalized[0] if any(cell.strip() for cell in normalized[0]) else [
        f'列{index + 1}' for index in range(width)
    ]
    body = normalized[1:] if header is normalized[0] else normalized
    lines = [
        '| ' + ' | '.join(_escape_markdown_table_cell(cell) for cell in header) + ' |',
        '| ' + ' | '.join('---' for _ in range(width)) + ' |',
    ]
    for row in body:
        lines.append('| ' + ' | '.join(_escape_markdown_table_cell(cell) for cell in row) + ' |')
    return '\n'.join(lines)


def _escape_markdown_table_cell(value: str) -> str:
    return value.replace('\\', '\\\\').replace('|', '\\|').replace('\n', ' ').strip()


def _limit_feishu_markdown(markdown: str) -> str:
    if len(markdown) <= FEISHU_DOC_MARKDOWN_MAX_CHARS:
        return markdown
    suffix = '\n\n> 内容较长，已截断后写入在线文档。'
    return markdown[: FEISHU_DOC_MARKDOWN_MAX_CHARS - len(suffix)].rstrip() + suffix


def _extract_first_url(text: str) -> str | None:
    match = re.search(r'https?://[^\s<>\]\)\"\'“”‘’]+', text)
    return _clean_url(match.group(0)) if match else None


def _clean_url(value: str) -> str:
    return value.strip().rstrip('.,;，。；、"\'“”‘’')


def _safe_account_id(account_id: str | None) -> str:
    normalized = re.sub(r'[^a-zA-Z0-9._-]+', '_', (account_id or '').strip())
    return normalized[:80] or 'default'


def _is_internal_message(content: str) -> bool:
    return content.lstrip().startswith('<system-reminder>')


def _strip_session_context(content: str) -> str:
    marker = '\n\n[当前会话目录]'
    if marker in content:
        return content.split(marker, 1)[0].strip()
    marker = '\n[当前会话目录]'
    if marker in content:
        return content.split(marker, 1)[0].strip()
    return content.strip()


def _iter_session_files(directory: Path) -> list[Path]:
    if not directory.exists():
        return []
    by_session_id: dict[str, Path] = {}
    for path in directory.glob('*.json'):
        by_session_id[_session_id_from_path(path)] = path
    for path in directory.glob('*/session.json'):
        # 新目录格式包含 input/output/scratchpad，优先级高于旧的平铺 json。
        by_session_id[_session_id_from_path(path)] = path
    return list(by_session_id.values())


def _delete_session_files(directory: Path, session_id: str) -> bool:
    deleted = False
    # 新目录格式：每个 session 独立目录，里面包含 session/input/output。
    nested = directory / session_id
    if nested.exists():
        shutil.rmtree(nested)
        deleted = True
    # 兼容早期平铺 session json，避免历史列表删除后旧数据又出现。
    legacy = directory / f'{session_id}.json'
    if legacy.exists():
        legacy.unlink()
        deleted = True
    return deleted


def _update_session_title(directory: Path, session_id: str, title: str) -> bool:
    path = _session_json_path(directory, session_id)
    if not path.exists():
        return False
    try:
        data = json.loads(path.read_text(encoding='utf-8'))
    except (OSError, json.JSONDecodeError):
        return False
    data['title'] = title
    data['title_source'] = 'manual'
    data['title_updated_at'] = int(time.time())
    _write_session_metadata(path, data)
    return True


def _session_mtime(path: Path) -> float:
    try:
        return path.stat().st_mtime
    except OSError:
        return 0.0


def _session_id_from_path(path: Path) -> str:
    if path.name == 'session.json':
        return path.parent.name
    return path.stem


def _safe_session_id(session_id: str | None) -> str | None:
    if not session_id:
        return None
    normalized = re.sub(r'[^a-zA-Z0-9._-]+', '_', session_id.strip())
    return normalized[:120] or None