#!/usr/bin/env python3
"""
GenericAgent Web2 Bridge.

Clear split:
1) AgentManager: owns GenericAgent instances, sessions and histories.
2) Transport: HTTP is the command/data channel; WebSocket only pushes small
   session-state notifications.

HTTP API:
  GET    /status
  GET    /config
  POST   /config
  GET    /model-profiles  (+ POST / PUT / DELETE by id)
  GET    /sessions
  POST   /session/new
  GET    /session/{sid}
  DELETE /session/{sid}
  POST   /session/{sid}/prompt
  GET    /session/{sid}/messages?after=0&limit=200
  POST   /session/{sid}/cancel
  POST   /services/start        body: {"id":"frontends/qqapp.py"}
  POST   /services/stop         body: {"id":"frontends/qqapp.py"}
  GET    /services/logs?id=frontends/qqapp.py&tail=200
  GET    /services/panel
  GET    /services/mykey
  POST   /services/mykey       body: {"content":"..."}
  POST   /services/stop-extras   stop conductor + scheduler (127.0.0.1 only)
  POST   /services/start-extras  start conductor + scheduler (127.0.0.1 only)
  POST   /services/bridge/exit    stop managed services, then exit bridge (127.0.0.1 only)

WS API (state sync):
  GET /ws -> on connect sends services.snapshot; service.changed on updates
  {"type":"services.snapshot","services":[...]}
  {"type":"service.changed","service":{...}}
"""
from __future__ import annotations

import asyncio, atexit, contextlib, importlib, json, os, re, subprocess, sys
from collections import Counter, deque
import threading, time, traceback, uuid
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Dict, List, Optional, Set
from aiohttp import web, WSMsgType

APP_DIR = Path(__file__).resolve().parent


def find_default_ga_root() -> Path:
    candidates = [
        APP_DIR / "..",
        APP_DIR / ".." / "..",
        APP_DIR / ".." / "GenericAgent",
        APP_DIR / ".." / ".." / "GenericAgent",
    ]
    for p in candidates:
        root = p.resolve()
        if (root / "agentmain.py").exists():
            return root
    return APP_DIR.parent.parent.resolve()


DEFAULT_GA_ROOT = find_default_ga_root()

_FINAL_INFO_RE = re.compile(r'\n*`{5}\n*\[Info\] Final response to user\.\n*`{5}\s*$')


def strip_final_info_marker(text: Any) -> str:
    return _FINAL_INFO_RE.sub('', str(text or ''))


for _s in (sys.stdout, sys.stderr):
    with contextlib.suppress(Exception):
        _s.reconfigure(encoding="utf-8", errors="replace")


# ---------------------------------------------------------------------------
# Agent management layer
# ---------------------------------------------------------------------------

@dataclass
class Session:
    id: str
    title: str = "New chat"
    cwd: str = ""
    created_at: float = field(default_factory=time.time)
    updated_at: float = field(default_factory=time.time)
    messages: List[dict] = field(default_factory=list)
    msg_seq: int = 0
    partial: Optional[dict] = None
    status: str = "idle"  # idle|running|error|cancelled
    agent: Any = None
    thread: Optional[threading.Thread] = None
    cancel_requested: bool = False
    last_error: str = ""
    pinned: bool = False
    untitled: bool = True
    plan_scan_baseline: int = 0
    plan_path: str = ""
    llm_history: Optional[List[dict]] = None


def _load_plan_baseline(item: dict, msgs: list) -> int:
    """Persisted per-session baseline (tuiapp_v2: set on /continue, not on preset text)."""
    base = int(item.get("plan_scan_baseline", 0) or 0)
    if base >= len(msgs):
        return 0
    return max(0, base)


def _sanitize_desktop_plan_path(session_id: str, plan_path: str) -> str:
    """Desktop: drop shared plan_demo paths so sessions do not read the same file."""
    import plan_state
    p = (plan_path or "").strip()
    if not p:
        return ""
    if plan_state.is_session_scoped_plan_path(p, session_id):
        return p
    return plan_state.default_session_plan_path(session_id)


class AgentManager:
    def __init__(self):
        self.lock = threading.RLock()
        self.ga_root = str(DEFAULT_GA_ROOT)
        self.config: Dict[str, Any] = {}
        self.sessions: Dict[str, Session] = {}
        self.active_session_id: Optional[str] = None
        self._sessions_file = Path(self.ga_root) / "temp" / "desktop_sessions.json"
        self._load_sessions()

    @property
    def mykey_path(self) -> str:
        return str(Path(self.ga_root) / "mykey.py")

    def _persist(self):
        try:
            self._sessions_file.parent.mkdir(parents=True, exist_ok=True)
            arr = []
            with self.lock:
                for s in self.sessions.values():
                    llm_hist = None
                    if s.agent and hasattr(s.agent, 'llmclient'):
                        try: llm_hist = s.agent.llmclient.backend.history
                        except Exception: pass
                    if llm_hist is None:
                        llm_hist = s.llm_history
                    arr.append({"id": s.id, "title": s.title, "cwd": s.cwd,
                                "created_at": s.created_at, "updated_at": s.updated_at,
                                "messages": s.messages, "msg_seq": s.msg_seq,
                                "pinned": s.pinned, "untitled": s.untitled,
                                "plan_scan_baseline": s.plan_scan_baseline,
                                "plan_path": s.plan_path or "",
                                "llm_history": llm_hist})
            self._sessions_file.write_text(json.dumps(arr, ensure_ascii=False, default=str), encoding="utf-8")
        except Exception as e:
            print(f"[bridge] persist sessions failed: {e}", file=sys.stderr)

    def _load_sessions(self):
        try:
            if not self._sessions_file.exists():
                return
            arr = json.loads(self._sessions_file.read_text(encoding="utf-8"))
            for item in arr:
                msgs = item.get("messages", [])
                sess = Session(id=item["id"], title=item.get("title", "New chat"),
                               cwd=item.get("cwd", self.ga_root),
                               created_at=item.get("created_at", time.time()),
                               updated_at=item.get("updated_at", time.time()),
                               messages=msgs,
                               msg_seq=item.get("msg_seq", 0),
                               pinned=item.get("pinned", False),
                               untitled=item.get("untitled", True),
                               plan_scan_baseline=_load_plan_baseline(item, msgs),
                               plan_path=_sanitize_desktop_plan_path(
                                   item["id"], item.get("plan_path") or ""),
                               status="idle", agent=None,
                               llm_history=item.get("llm_history"))
                self.sessions[sess.id] = sess
            if self.sessions:
                self.active_session_id = max(self.sessions.values(), key=lambda s: s.updated_at).id
        except Exception as e:
            print(f"[bridge] load sessions failed: {e}", file=sys.stderr)

    def _mykey_file(self) -> Path:
        p = Path(self.ga_root) / "mykey.py"
        if not p.exists():
            tpl = Path(self.ga_root) / "mykey_template.py"
            p.write_text(tpl.read_text(encoding="utf-8") if tpl.exists() else "", encoding="utf-8")
        return p

    @staticmethod
    def _next_native_var(text: str, protocol: str) -> str:
        # 协议必选(由前端下拉强制),不再用 apibase 兜底瞎猜
        proto = str(protocol or "").strip().lower()
        if proto == "claude":
            prefix = "native_claude_config"
        elif proto in ("oai", "openai"):
            prefix = "native_oai_config"
        else:
            raise ValueError("protocol is required: choose 'oai' or 'claude'")
        nums = [0]
        if re.search(rf"^{prefix}\s*=", text, re.M):
            nums.append(0)
        nums.extend(int(m.group(1)) for m in re.finditer(rf"^{prefix}(\d+)\s*=", text, re.M))
        n = max(nums) + 1
        return prefix if n == 1 and not re.search(rf"^{prefix}\s*=", text, re.M) else f"{prefix}{n}"

    @staticmethod
    def _format_py_dict(d: dict) -> str:
        lines = [f"    '{k}': {json.dumps(v, ensure_ascii=False)}," if isinstance(v, str) else f"    '{k}': {v}," for k, v in d.items()]
        return "{\n" + "\n".join(lines) + "\n}"

    def _invalidate_mykey_cache(self) -> None:
        self.ensure_ga_import_path()
        sys.modules.pop("mykey", None)
        with contextlib.suppress(Exception):
            import llmcore
            llmcore._mykey_mtime = None

    def _profile_keys(self) -> List[str]:
        self.ensure_ga_import_path()
        from llmcore import reload_mykeys
        return [k for k in reload_mykeys()[0] if any(x in k for x in ("api", "config", "cookie"))]

    def _profile_at(self, profile_id: int) -> tuple[str, dict]:
        keys = self._profile_keys()
        if profile_id < 0 or profile_id >= len(keys):
            raise ValueError("profile not found")
        var = keys[profile_id]
        if "mixin" in var:
            raise ValueError("mixin profiles not supported here")
        from llmcore import reload_mykeys
        cfg = reload_mykeys()[0].get(var)
        if not isinstance(cfg, dict):
            raise ValueError("profile not editable")
        return var, dict(cfg)

    @staticmethod
    def _find_var_block_span(text: str, var_name: str) -> Optional[tuple[int, int]]:
        m = re.search(rf"^{re.escape(var_name)}\s*=\s*\{{", text, re.M)
        if not m:
            return None
        start, i, depth = m.start(), m.end() - 1, 0
        while i < len(text):
            if text[i] == "{":
                depth += 1
            elif text[i] == "}":
                depth -= 1
                if depth == 0:
                    end = i + 1
                    while end < len(text) and text[end] in "\r\n":
                        end += 1
                    return start, end
            i += 1
        return None

    def _patch_var_block(self, text: str, var: str, cfg: Optional[dict] = None) -> str:
        if not (span := self._find_var_block_span(text, var)):
            raise ValueError(f"config block not found: {var}")
        s, e = span
        if cfg is None:
            return text[:s].rstrip() + "\n" + text[e:].lstrip("\n")
        return text[:s] + f"{var} = {self._format_py_dict(cfg)}\n" + text[e:]

    def _build_cfg(self, data: dict, existing: Optional[dict] = None, *, require_key: bool = True) -> dict:
        apibase, model = str(data.get("apibase") or "").strip(), str(data.get("model") or "").strip()
        if not apibase or not model:
            raise ValueError("apibase and model are required")
        apikey = str(data.get("apikey") or "").strip() or str((existing or {}).get("apikey") or "").strip()
        if require_key and not apikey:
            raise ValueError("apikey is required")
        # 从 existing 起步：保留表单未覆盖的高级字段（proxy / temperature / api_mode /
        # reasoning_effort / fake_cc_system_prompt / thinking_type …），避免 GUI 编辑时丢失
        cfg: Dict[str, Any] = dict(existing or {})
        cfg.update({"apikey": apikey, "apibase": apibase, "model": model})
        if "name" in data:
            name = str(data.get("name") or "").strip()
            if name:
                cfg["name"] = name
            else:
                cfg.pop("name", None)
        for k in ("max_retries", "connect_timeout", "read_timeout"):
            if data.get(k) is not None and str(data.get(k)).strip() != "":
                cfg[k] = int(data[k])
        # 流式开关：默认 True 不写（保持 mykey 干净），仅显式非流式才落 'stream': False
        if "stream" in data:
            s = data["stream"]
            stream = s if isinstance(s, bool) else str(s).strip().lower() not in ("false", "0", "no", "off")
            if stream:
                cfg.pop("stream", None)
            else:
                cfg["stream"] = False
        return cfg

    def _save_mykey_text(self, text: str) -> list:
        self._mykey_file().write_text(text, encoding="utf-8")
        self._invalidate_mykey_cache()
        self._reload_live_agents()
        return self.list_model_profiles()

    def _reload_live_agents(self) -> None:
        """mykey.py 改动后，强制所有活着的会话 agent 重建 LLM session，让新 key/模型
        立即生效（无需重启）。重建保留对话 history（agentmain 内部用 oldhistory 接回）。

        纯 bridge 侧实现，不改 agentmain：每次调 agent.load_llm_sessions() 前，把
        llmcore 的全局 mtime 标志清空（与 _invalidate_mykey_cache 同一手法），使其内部
        reload_mykeys() 报告 changed=True、从而真正重建——否则刷新模型列表等路径会先
        消费掉变更标志，常驻 agent 的 load_llm_sessions 会因 changed=False 跳过重建。"""
        self.ensure_ga_import_path()
        try:
            import llmcore
        except Exception:
            return
        with self.lock:
            agents = [s.agent for s in self.sessions.values() if getattr(s, "agent", None) is not None]
        for agent in agents:
            fn = getattr(agent, "load_llm_sessions", None)
            if not callable(fn):
                continue
            try:
                llmcore._mykey_mtime = None   # 让本次 reload_mykeys() 视为“已变更”，触发真正重建
                fn()
            except Exception as e:
                print(f"[bridge] reload live agent failed: {e}", file=sys.stderr)

    def add_model_profile(self, data: dict) -> dict:
        cfg = self._build_cfg(data)
        text = self._mykey_file().read_text(encoding="utf-8")
        var = self._next_native_var(text, data.get("protocol", ""))
        profiles = self._save_mykey_text(text.rstrip() + f"\n{var} = {self._format_py_dict(cfg)}\n")
        return {"varName": var, "profileId": profiles[-1]["id"] if profiles else 0, "profiles": profiles}

    def get_model_profile(self, profile_id: int) -> dict:
        var, cfg = self._profile_at(profile_id)
        ks = ("model", "apibase", "apikey", "name", "max_retries", "connect_timeout", "read_timeout")
        out = {"id": profile_id, "varName": var, **{k: cfg.get(k, d) for k, d in zip(ks, ("", "", "", "", 5, 15, 300))}}
        out["stream"] = cfg.get("stream", True)
        return out

    def update_model_profile(self, profile_id: int, data: dict) -> dict:
        var, existing = self._profile_at(profile_id)
        text = self._mykey_file().read_text(encoding="utf-8")
        profiles = self._save_mykey_text(self._patch_var_block(text, var, self._build_cfg(data, existing, require_key=False)))
        return {"varName": var, "profileId": profile_id, "profiles": profiles}

    def delete_model_profile(self, profile_id: int) -> dict:
        if len(self._profile_keys()) <= 1:
            raise ValueError("cannot delete the last profile")
        var, cfg = self._profile_at(profile_id)
        text = self._patch_var_block(self._mykey_file().read_text(encoding="utf-8"), var).rstrip() + "\n"
        # 顺手把它从聚合渠道里摘掉，避免 llm_nos 残留指向已删除的模型（会让 Mixin 构建失败）
        name = str(cfg.get("name") or cfg.get("model") or "").strip()
        keys, mk = self._mykey_vars()
        mvar, mcfg = self._mixin_entry(keys, mk)
        if mcfg and mvar is not None and name in [str(m) for m in (mcfg.get("llm_nos") or [])]:
            mcfg = {**mcfg, "llm_nos": [str(m) for m in (mcfg.get("llm_nos") or []) if str(m) != name]}
            if self._find_var_block_span(text, mvar):
                text = self._patch_var_block(text, mvar, mcfg)
        profiles = self._save_mykey_text(text)
        return {"profileId": profile_id, "profiles": profiles}

    def ensure_ga_import_path(self) -> Path:
        root = Path(self.ga_root).resolve()
        if str(root) not in sys.path:
            sys.path.insert(0, str(root))
        return root

    def make_agent(self, sess: Session):
        root = self.ensure_ga_import_path()
        try: import cost_tracker; cost_tracker.install()
        except Exception: pass
        old_cwd = os.getcwd()
        try:
            os.chdir(sess.cwd or str(root))
            agentmain = importlib.import_module("agentmain")
            GA = getattr(agentmain, "GenericAgent")
            agent = GA()
            agent.inc_out = True
            agent.verbose = True
            threading.Thread(target=agent.run, daemon=True, name=f"GA-{sess.id}").start()
            return agent
        finally:
            with contextlib.suppress(Exception):
                os.chdir(old_cwd)

    @staticmethod
    def _base_display_name(var: str, cfg: Optional[dict]) -> str:
        c = cfg or {}
        return str(c.get("name") or c.get("model") or var)

    def _mykey_vars(self):
        """(keys, mk)：mykey 里的模型变量名（按定义顺序，与 agentmain.llmclients 索引
        一一对齐）和原始 dict。过滤规则与 _profile_keys / load_llm_sessions 完全一致，
        因此 id == llmclients 下标，前端选中 llmNo 能正确激活对应 client。"""
        self._mykey_file()   # 确保 mykey.py 存在（首次从模板生成空配置），否则全新安装时
                             # reload_mykeys 找不到 mykey 会返回空，空聚合渠道就不显示了
        self.ensure_ga_import_path()
        from llmcore import reload_mykeys
        mk = reload_mykeys()[0]
        keys = [k for k in mk if any(x in k for x in ("api", "config", "cookie"))]
        return keys, mk

    def _mixin_entry(self, keys, mk):
        """返回 (mixin_var, mixin_cfg_dict) 或 (None, None)。单一主聚合渠道，只取第一个。"""
        for k in keys:
            if "mixin" in k and isinstance(mk.get(k), dict):
                return k, dict(mk[k])
        return None, None

    def list_model_profiles(self):
        """直接读 mykey.py 结构（不依赖能否成功构建出 client），这样空聚合渠道、
        未填 key 的模型也能如实展示。聚合渠道(kind=mixin)带 members；基本模型
        (kind=native)带 inMixin/group。"""
        try:
            keys, mk = self._mykey_vars()
        except Exception as e:
            print(f"get model profiles failed: {e}", file=sys.stderr)
            return []
        active = self.config.get("llmNo", 0)
        # collect all mixin members for inMixin check
        all_mixin_members: set = set()
        for k in keys:
            if "mixin" in k:
                c = mk.get(k) if isinstance(mk.get(k), dict) else {}
                all_mixin_members.update(str(m) for m in (c.get("llm_nos") or []))
        out = []
        for i, k in enumerate(keys):
            cfg = mk.get(k) if isinstance(mk.get(k), dict) else {}
            if "mixin" in k:
                mems = [str(m) for m in (cfg.get("llm_nos") or [])]
                out.append({"id": i, "varName": k, "kind": "mixin", "name": "",
                            "members": mems, "active": i == active})
            else:
                name = self._base_display_name(k, cfg)
                out.append({"id": i, "varName": k, "kind": "native", "name": name,
                            "model": cfg.get("model", ""),
                            "group": "native" if "native" in k else "std",
                            "inMixin": name in all_mixin_members, "active": i == active})
        return out

    def add_to_mixin(self, profile_id: int) -> dict:
        """把一个基本模型加入主聚合渠道：把它的 name 追加进 mixin_config['llm_nos']。
        坑1：校验 Native 一致性（聚合内必须全 Native 或全非 Native）。
        坑2：加入前若该模型没有显式 name，先把 name 写进它的配置块（保证引用稳定）。"""
        var, cfg = self._profile_at(profile_id)   # 对 mixin 会抛错（只接受 native）
        name = str(cfg.get("name") or cfg.get("model") or "").strip()
        if not name:
            raise ValueError("this model needs a name or model before joining the channel")
        keys, mk = self._mykey_vars()
        mvar, mcfg = self._mixin_entry(keys, mk)
        new_is_native = "native" in var
        name2var = {self._base_display_name(k, mk.get(k) if isinstance(mk.get(k), dict) else {}): k
                    for k in keys if "mixin" not in k}
        existing = [str(m) for m in (mcfg.get("llm_nos") or [])] if mcfg else []
        for m in existing:
            mv = name2var.get(m)
            if mv is not None and ("native" in mv) != new_is_native:
                raise ValueError("aggregation channel requires all-Native or all-non-Native models")
        text = self._mykey_file().read_text(encoding="utf-8")
        if not cfg.get("name"):
            text = self._patch_var_block(text, var, {**cfg, "name": name})
        if mcfg is None:
            mcfg, mvar, existing = {"llm_nos": [], "max_retries": 10, "base_delay": 0.5}, "mixin_config", []
        if name not in existing:
            existing.append(name)
        mcfg = {**mcfg, "llm_nos": existing}
        if self._find_var_block_span(text, mvar):
            text = self._patch_var_block(text, mvar, mcfg)
        else:
            text = text.rstrip() + f"\n{mvar} = {self._format_py_dict(mcfg)}\n"
        return {"profiles": self._save_mykey_text(text)}

    def remove_from_mixin(self, profile_id: int) -> dict:
        """把一个基本模型移出主聚合渠道。"""
        var, cfg = self._profile_at(profile_id)
        name = str(cfg.get("name") or cfg.get("model") or "").strip()
        keys, mk = self._mykey_vars()
        mvar, mcfg = self._mixin_entry(keys, mk)
        if not mcfg or mvar is None:
            return {"profiles": self.list_model_profiles()}
        members = [str(m) for m in (mcfg.get("llm_nos") or []) if str(m) != name]
        mcfg = {**mcfg, "llm_nos": members}
        text = self._patch_var_block(self._mykey_file().read_text(encoding="utf-8"), mvar, mcfg)
        return {"profiles": self._save_mykey_text(text)}

    def reorder_mixin(self, members: list) -> dict:
        """按前端拖拽后的顺序重写主渠道组 llm_nos。只接受当前成员的重排，不增删。"""
        keys, mk = self._mykey_vars()
        mvar, mcfg = self._mixin_entry(keys, mk)
        if not mcfg or mvar is None:
            raise ValueError("mixin channel not found")
        old = [str(m) for m in (mcfg.get("llm_nos") or [])]
        new = [str(m) for m in (members or [])]
        if len(new) != len(old) or Counter(new) != Counter(old):
            raise ValueError("reorder must contain the same channel members")
        if new == old:
            return {"profiles": self.list_model_profiles()}
        mcfg = {**mcfg, "llm_nos": new}
        text = self._patch_var_block(self._mykey_file().read_text(encoding="utf-8"), mvar, mcfg)
        return {"profiles": self._save_mykey_text(text)}

    @staticmethod
    def _live_model(sess: Session) -> Optional[dict]:
        """该会话 agent 当前真正在用的模型（渠道组会随故障转移变化）。
        agent 还没建（没跑过 turn）时返回 None，前端回退到静态显示。"""
        ag = getattr(sess, "agent", None)
        if ag is None:
            return None
        try:
            back = ag.llmclient.backend
            if "Mixin" in type(back).__name__:
                return {"current": back.current_name, "isMixin": True}
            return {"current": back.name, "isMixin": False}
        except Exception:
            return None

    def snapshot(self, sess: Session, include_messages: bool = True) -> dict:
        out = {
            "sessionId": sess.id,
            "id": sess.id,
            "title": sess.title,
            "cwd": sess.cwd,
            "status": sess.status,
            "createdAt": sess.created_at,
            "updatedAt": sess.updated_at,
            "lastError": sess.last_error,
            "msgSeq": sess.msg_seq,
            "pinned": sess.pinned,
            "untitled": sess.untitled,
            "model": self._live_model(sess),
        }
        if include_messages:
            out["messages"] = list(sess.messages)
            out["partial"] = dict(sess.partial) if sess.partial else None
        return out

    def add_message(self, sess: Session, role: str, content: str, **extra) -> dict:
        sess.msg_seq += 1
        msg = {"id": sess.msg_seq, "role": role, "content": content, "ts": time.time()}
        msg.update(extra)
        sess.messages.append(msg)
        sess.updated_at = time.time()
        if role == "user" and content.strip() and sess.title == "New chat":
            sess.title = content.strip().replace("\n", " ")[:40]
        self._persist()
        return msg

    def create_session(self, cwd: Optional[str] = None) -> Session:
        sid = "sess-" + uuid.uuid4().hex[:12]
        sess = Session(id=sid, cwd=str(cwd or self.ga_root))
        with self.lock:
            self.sessions[sid] = sess
            self.active_session_id = sid
        emit_session_state(sess, "created")
        self._persist()
        return sess

    def get_session(self, sid: str) -> Session:
        with self.lock:
            sess = self.sessions.get(sid)
            if not sess:
                raise web.HTTPNotFound(text=json.dumps({"error": f"session not found: {sid}"}, ensure_ascii=False), content_type="application/json")
            return sess

    def delete_session(self, sid: str) -> dict:
        with self.lock:
            sess = self.sessions.pop(sid, None)
            if not sess:
                raise web.HTTPNotFound(text=json.dumps({"error": f"session not found: {sid}"}, ensure_ascii=False), content_type="application/json")
            if self.active_session_id == sid:
                self.active_session_id = next(iter(self.sessions), None)
            if sess.agent and hasattr(sess.agent, "abort"):
                with contextlib.suppress(Exception):
                    sess.agent.abort()
        emit_session_state(sess, "closed")
        self._persist()
        _purge_session_uploads(sid)
        return {"ok": True, "sessionId": sid}

    def submit_prompt(self, sid: str, prompt: Any, images: Optional[list] = None, llm_no: Optional[int] = None, display: Optional[str] = None, files_meta: Optional[list] = None, image_metas: Optional[list] = None) -> dict:
        prompt, image_ids = normalize_prompt(prompt, images)
        if llm_no is not None:
            self.config["llmNo"] = int(llm_no)
        with self.lock:
            sess = self.sessions.get(sid)
            if not sess:
                raise web.HTTPNotFound(text=json.dumps({"error": f"session not found: {sid}"}, ensure_ascii=False), content_type="application/json")
            if sess.status == "running":
                raise web.HTTPConflict(text=json.dumps({"error": "session is already running"}, ensure_ascii=False), content_type="application/json")
            extra = {}
            if image_ids:
                extra["image_ids"] = image_ids
            if isinstance(display, str) and display.strip() and display != prompt:
                extra["display"] = display
            if files_meta:
                extra["files"] = files_meta
            if image_metas:
                extra["images"] = image_metas
            user_msg = self.add_message(sess, "user", prompt, **extra)
            import plan_state
            if plan_state.is_plan_preset_prompt(prompt):
                plan_state.bind_plan_session(sess, prompt)
                self._persist()
            sess.status = "running"
            sess.cancel_requested = False
            sess.last_error = ""
            sess.partial = {"id": sess.msg_seq + 1, "role": "assistant", "content": "", "ts": time.time(), "partial": True,
                            "curr_turn": 0, "turn_segs": []}  # turn_segs[i]=第i轮全文(权威结构化,前端按轮渲染);content保留双轨兜底
            t = threading.Thread(target=self.run_agent_turn, args=(sess, prompt, None, llm_no), daemon=True, name=f"Turn-{sid}")
            sess.thread = t
            t.start()
            seq = sess.msg_seq
        emit_session_state(sess, "running")
        return {"ok": True, "sessionId": sid, "accepted": True, "userMessageId": user_msg["id"], "seq": seq}

    def run_agent_turn(self, sess: Session, prompt: str, images: Optional[list] = None, llm_no: Optional[int] = None):
        try:
            if sess.agent is None:
                sess.agent = self.make_agent(sess)
            agent = sess.agent
            no = self.config.get("llmNo") if llm_no is None else llm_no
            if no is not None and hasattr(agent, "next_llm"):
                with contextlib.suppress(Exception):
                    agent.next_llm(int(no))
            full = ""
            done_outputs = None  # done时agent给的全量轮文本(turn_resps.copy())
            if hasattr(agent, "put_task"):
                display_q = agent.put_task(prompt, images=images or [])
                pieces = []
                import queue as _queue
                while True:
                    if sess.cancel_requested:
                        break
                    try:
                        item = display_q.get(timeout=1.0)
                    except _queue.Empty:
                        continue
                    if isinstance(item, dict):
                        if item.get("next"):
                            text = str(item["next"])
                            pieces.append(text)
                            with self.lock:
                                if sess.partial is not None:
                                    sess.partial["content"] = "".join(pieces) if getattr(agent, "inc_out", False) else text
                                    sess.partial["ts"] = time.time()
                                    sess.updated_at = time.time()
                                    # 轨道2: bridge 归一化为前端直接可渲染的 0 基 turn_segs；outputs=turn_resps[-2:]
                                    _t = int(item.get("turn", 0) or 0)
                                    _outs = item.get("outputs") or []
                                    _idx = max(0, _t - 1)
                                    sess.partial["curr_turn"] = _idx
                                    _segs = sess.partial["turn_segs"]
                                    while len(_segs) <= _idx:
                                        _segs.append("")
                                    if _outs:
                                        _segs[_idx] = str(_outs[-1])
                                        if len(_outs) >= 2 and _idx >= 1:
                                            _segs[_idx - 1] = str(_outs[-2])
                        if "done" in item:
                            full = strip_final_info_marker(item.get("done") or "")
                            done_outputs = item.get("outputs")  # done时=turn_resps.copy()全量轮
                            if done_outputs:
                                done_outputs = [strip_final_info_marker(s) for s in done_outputs]
                                with self.lock:
                                    if sess.partial is not None:
                                        sess.partial["content"] = full
                                        sess.partial["ts"] = time.time()
                                        sess.partial["updatedAt"] = sess.partial["ts"] if "updatedAt" in sess.partial else sess.partial.get("updatedAt")
                                        sess.partial["curr_turn"] = max(0, len(done_outputs) - 1)
                                        sess.partial["turn_segs"] = list(done_outputs)
                                        sess.updated_at = time.time()
                            break
                    else:
                        pieces.append(str(item))
                if not full and pieces:
                    full = pieces[-1] if not getattr(agent, "inc_out", False) else "".join(pieces)
            else:
                full = "GenericAgent object has no put_task method"
            if not full:
                full = "(completed)"
            if sess.cancel_requested:
                with self.lock:
                    sess.partial = None
                    # Ensure status stays cancelled (don't overwrite)
                    if sess.status != "cancelled":
                        sess.status = "cancelled"
                    sess.updated_at = time.time()
                emit_session_state(sess, "cancelled")
                return
            with self.lock:
                sess.partial = None
                full = strip_final_info_marker(full)
                if done_outputs:
                    done_outputs = [strip_final_info_marker(s) for s in done_outputs]
                import plan_state
                plan_state.sync_plan_path_from_text(sess, full, sess.cwd or self.ga_root)
                # 轨道2: 落库时带结构化全量轮(权威turn_segs),前端按轮渲染;content保留兜底
                _final_segs = [str(s) for s in done_outputs] if done_outputs else None
                if _final_segs:
                    self.add_message(sess, "assistant", full, turn_segs=_final_segs)
                else:
                    self.add_message(sess, "assistant", full)
                try: sess.llm_history = json.loads(json.dumps(agent.llmclient.backend.history, ensure_ascii=False, default=str))
                except Exception: pass
                sess.status = "idle"
                sess.last_error = ""
            emit_session_state(sess, "idle")
        except Exception as e:
            tb = traceback.format_exc()
            with self.lock:
                sess.partial = None
                sess.status = "error"
                sess.last_error = str(e)
                self.add_message(sess, "error", str(e))
            print(tb, file=sys.stderr)
            emit_session_state(sess, "error")

    def messages(self, sid: str, after: int = 0, limit: int = 200) -> dict:
        with self.lock:
            sess = self.sessions.get(sid)
            if not sess:
                raise web.HTTPNotFound(text=json.dumps({"error": f"session not found: {sid}"}, ensure_ascii=False), content_type="application/json")
            msgs = [m for m in sess.messages if int(m.get("id", 0)) > after]
            if limit > 0:
                msgs = msgs[-limit:]
            import plan_state
            return {
                "sessionId": sid,
                "status": sess.status,
                "messages": msgs,
                "partial": dict(sess.partial) if sess.partial else None,
                "plan": plan_state.desktop_plan_payload_from_session(sess, self.ga_root),
                "msgSeq": sess.msg_seq,
                "updatedAt": sess.updated_at,
                "lastError": sess.last_error,
                "model": self._live_model(sess),
            }

    def plan_snapshot(self, sid: str) -> dict:
        with self.lock:
            sess = self.sessions.get(sid)
            if not sess:
                raise web.HTTPNotFound(text=json.dumps({"error": f"session not found: {sid}"}, ensure_ascii=False), content_type="application/json")
            import plan_state
            return {
                "sessionId": sid,
                "plan": plan_state.desktop_plan_payload_from_session(sess, self.ga_root),
            }

    def cancel(self, sid: str) -> dict:
        with self.lock:
            sess = self.sessions.get(sid)
            if not sess:
                raise web.HTTPNotFound(text=json.dumps({"error": f"session not found: {sid}"}, ensure_ascii=False), content_type="application/json")
            sess.cancel_requested = True
            if sess.agent and hasattr(sess.agent, "abort"):
                with contextlib.suppress(Exception):
                    sess.agent.abort()
            partial_text = ""
            if sess.partial:
                partial_text = (sess.partial.get("content") or "").strip()
            if partial_text:
                self.add_message(sess, "assistant", partial_text, stopped=True)
            sess.status = "cancelled"
            sess.partial = None
            sess.updated_at = time.time()
        emit_session_state(sess, "cancelled")
        return {"ok": True, "sessionId": sid}

    def restore_context(self, sid: str) -> dict:
        with self.lock:
            sess = self.sessions.get(sid)
            if not sess:
                raise web.HTTPNotFound(text=json.dumps({"error": f"session not found: {sid}"}, ensure_ascii=False), content_type="application/json")
            if sess.agent is not None:
                return {"ok": True, "sessionId": sid, "restored": False, "reason": "agent already alive"}
        agent = self.make_agent(sess)
        if sess.llm_history:
            try:
                agent.llmclient.backend.history = sess.llm_history
            except Exception as e:
                print(f"[bridge] restore llm_history failed: {e}", file=sys.stderr)
        else:
            history = []
            for m in sess.messages:
                role = m.get("role")
                content = m.get("content", "")
                if role == "user":
                    history.append({"role": "user", "content": [{"type": "text", "text": content}]})
                elif role == "assistant":
                    history.append({"role": "assistant", "content": [{"type": "text", "text": content}]})
            if history:
                try:
                    agent.llmclient.backend.history = history
                except Exception as e:
                    print(f"[bridge] inject history failed: {e}", file=sys.stderr)
        with self.lock:
            sess.agent = agent
            sess.status = "idle"
        return {"ok": True, "sessionId": sid, "restored": True, "messageCount": len(sess.llm_history or sess.messages)}


import base64


def normalize_prompt(prompt: Any, images: Optional[list] = None):
    """Flatten a prompt (str or content-part list) to plain text.

    Image/file attachments are handled by the frontend, which inlines the
    uploaded file path into the prompt text (see expandFilePlaceholders) and
    sends path-only metadata via files/imageMetas — so no per-prompt image
    persistence happens here. The `images` arg is accepted for backward compat
    and ignored; the returned image-id list is always empty.
    """
    if isinstance(prompt, list):
        text_parts = []
        for part in prompt:
            if isinstance(part, str):
                text_parts.append(part)
            elif isinstance(part, dict) and part.get("type") in ("text", "input_text"):
                text_parts.append(str(part.get("text") or part.get("content") or ""))
        prompt = "\n".join([p for p in text_parts if p])

    return str(prompt or ""), []


manager = AgentManager()


# ---------------------------------------------------------------------------
# Transport layer: WS state push
# ---------------------------------------------------------------------------

class WsHub:
    def __init__(self):
        self.websockets: Set[web.WebSocketResponse] = set()
        self.loop: Optional[asyncio.AbstractEventLoop] = None

    def emit(self, obj: dict):
        if self.loop and self.loop.is_running():
            asyncio.run_coroutine_threadsafe(self._broadcast(obj), self.loop)

    async def _broadcast(self, obj: dict):
        data = json.dumps(obj, ensure_ascii=False, default=str)
        dead = set()
        for ws in list(self.websockets):
            try:
                await ws.send_str(data)
            except Exception:
                dead.add(ws)
        self.websockets.difference_update(dead)


hub = WsHub()


# ---------------------------------------------------------------------------
# Service management (hub.pyw core + WS notify)
# ---------------------------------------------------------------------------

_SKIP = frozenset({"goal_mode.py", "chatapp_common.py", "tuiapp.py", "qtapp.py"})
BRIDGE_ID = "__bridge__"

_SERVICE_KEYS: Dict[str, tuple] = {
    "frontends/qqapp.py": ("qq_app_id", "qq_app_secret"),
    "frontends/dcapp.py": ("discord_bot_token",),
    "frontends/dingtalkapp.py": ("dingtalk_client_id", "dingtalk_client_secret"),
    "frontends/fsapp.py": ("fs_app_id", "fs_app_secret"),
    "frontends/tgapp.py": ("tg_bot_token",),
    "frontends/wecomapp.py": ("wecom_bot_id", "wecom_secret"),
}


def _load_mykeys(ga_root: Path) -> dict:
    if not (ga_root / "mykey.py").exists():
        return {}
    root = str(ga_root.resolve())
    if root not in sys.path:
        sys.path.insert(0, root)
    import mykey as mk
    importlib.reload(mk)
    return {k: v for k, v in vars(mk).items() if not k.startswith("_")}


def discover_im_services(ga_root: Path) -> List[dict]:
    out: List[dict] = []
    d = ga_root / "frontends"
    if not d.is_dir():
        return out
    for f in sorted(os.listdir(d)):
        if "app" not in f or not f.endswith(".py") or f in _SKIP or "stapp" in f or "tuiapp" in f:
            continue
        rel = f"frontends/{f}"
        out.append({"id": rel, "cmd": [sys.executable, str(d / f)]})
    return out


def discover_extra_services(ga_root: Path) -> List[dict]:
    out: List[dict] = []
    sched = ga_root / "reflect" / "scheduler.py"
    if sched.is_file():
        out.append({
            "id": "reflect/scheduler.py",
            "cmd": [sys.executable, "agentmain.py", "--reflect", "reflect/scheduler.py"],
        })
    # conductor 跟 scheduler 一样,bridge 启动时自动拉起。--no-browser 是关键:
    # conductor.py 默认会用 webbrowser.open 在用户浏览器弹一个 8900 端口 UI,
    # 桌面版自启时不需要这个独立 UI(用户从「指挥家」页直接访问)。
    conductor = ga_root / "frontends" / "conductor.py"
    if conductor.is_file():
        out.append({
            "id": "frontends/conductor.py",
            "cmd": [sys.executable, "frontends/conductor.py", "--no-browser"],
        })
    return out


def _mem_mb(pid: Optional[int]) -> Optional[int]:
    if not pid:
        return None
    if sys.platform == "win32":
        import ctypes
        from ctypes import wintypes
        class PROCESS_MEMORY_COUNTERS(ctypes.Structure):
            _fields_ = [
                ("cb", wintypes.DWORD), ("PageFaultCount", wintypes.DWORD),
                ("PeakWorkingSetSize", ctypes.c_size_t), ("WorkingSetSize", ctypes.c_size_t),
                ("QuotaPeakPagedPoolUsage", ctypes.c_size_t), ("QuotaPagedPoolUsage", ctypes.c_size_t),
                ("QuotaPeakNonPagedPoolUsage", ctypes.c_size_t), ("QuotaNonPagedPoolUsage", ctypes.c_size_t),
                ("PagefileUsage", ctypes.c_size_t), ("PeakPagefileUsage", ctypes.c_size_t),
            ]
        counters = PROCESS_MEMORY_COUNTERS()
        counters.cb = ctypes.sizeof(PROCESS_MEMORY_COUNTERS)
        h = ctypes.windll.kernel32.OpenProcess(0x1000, False, pid)
        if not h:
            return None
        ok = ctypes.windll.psapi.GetProcessMemoryInfo(h, ctypes.byref(counters), counters.cb)
        ctypes.windll.kernel32.CloseHandle(h)
        return round(counters.WorkingSetSize / 1024 / 1024) if ok else None
    status = Path(f"/proc/{pid}/status")
    if status.is_file():
        for line in status.read_text(encoding="utf-8", errors="replace").splitlines():
            if line.startswith("VmRSS:"):
                return round(int(line.split()[1]) / 1024)
    return None


def _cpu_pct(pid: Optional[int]) -> Optional[float]:
    if not pid:
        return None
    try:
        import psutil
        return round(psutil.Process(pid).cpu_percent(0) or 0, 1)
    except Exception:
        return None


class ServiceManager:
    """hub.pyw ServiceManager + HTTP/WS glue."""

    def __init__(self, ga_root: str, emit_fn):
        self.ga_root = Path(ga_root)
        self.procs: Dict[str, subprocess.Popen] = {}
        self.buffers: Dict[str, deque] = {}
        self._emit = emit_fn
        im = discover_im_services(self.ga_root)
        extra = discover_extra_services(self.ga_root)
        self._im_catalog = {s["id"]: s for s in im}
        self._catalog = {**self._im_catalog, **{s["id"]: s for s in extra}}
        self._stopping: Set[str] = set()

    def _is_configured(self, sid: str) -> bool:
        keys = _SERVICE_KEYS.get(sid)
        if not keys:
            return True
        mykeys = _load_mykeys(self.ga_root)
        return all(str(mykeys.get(k) or "").strip() for k in keys)

    def _log_tail(self, sid: str, n: int = 3) -> str:
        buf = self.buffers.get(sid)
        if not buf:
            return ""
        lines = [ln.strip() for ln in list(buf)[-n:] if ln.strip()]
        return lines[-1][:300] if lines else ""

    def _state(self, sid: str, *, err: str = "") -> dict:
        proc = self.procs.get(sid)
        running = proc is not None and proc.poll() is None
        status = "running" if running else "offline"
        last_error = err
        if proc is not None and proc.poll() is not None:
            if sid in self._stopping:
                status, last_error = "offline", ""
            else:
                status = "error"
                last_error = err or self._log_tail(sid) or f"exit code {proc.returncode}"
        elif err:
            status, running = "error", False
        return {
            "id": sid,
            "status": status,
            "running": running,
            "pid": proc.pid if running else None,
            "lastError": last_error,
        }

    def list_state(self) -> List[dict]:
        return [self._state(sid) for sid in sorted(self._im_catalog)]

    def _bridge_state(self) -> dict:
        pid = os.getpid()
        port = int(os.environ.get("BRIDGE_PORT", "14168"))
        return {
            "id": BRIDGE_ID,
            "name": f"bridge (:{port})",
            "status": "running",
            "running": True,
            "pid": pid,
            "memMb": _mem_mb(pid),
            "cpuPct": _cpu_pct(pid),
            "managed": False,
            "lastError": "",
        }

    def list_panel_state(self) -> List[dict]:
        out = [self._bridge_state()]
        for sid in sorted(self._catalog, key=lambda s: (s in self._im_catalog, s)):
            item = self._state(sid)
            item["name"] = sid
            item["memMb"] = _mem_mb(item.get("pid"))
            item["cpuPct"] = _cpu_pct(item.get("pid"))
            item["managed"] = True
            out.append(item)
        return out

    def _notify(self, sid: str, *, err: str = "") -> None:
        self._emit({"type": "service.changed", "service": self._state(sid, err=err)})

    def _wait_started(self, proc: subprocess.Popen, timeout: float = 2.0) -> None:
        deadline = time.time() + timeout
        while time.time() < deadline:
            if proc.poll() is not None:
                return
            time.sleep(0.1)

    def _reader(self, sid: str, proc: subprocess.Popen) -> None:
        assert proc.stdout is not None
        for line in proc.stdout:
            buf = self.buffers.get(sid)
            if buf is not None:
                buf.append(line)
        self._notify(sid)

    def start_service(self, sid: str) -> dict:
        svc = self._catalog.get(sid)
        if not svc:
            raise KeyError(sid)
        proc = self.procs.get(sid)
        if proc is not None and proc.poll() is None:
            return {"ok": True, "service": self._state(sid)}
        if not self._is_configured(sid):
            keys = ", ".join(_SERVICE_KEYS.get(sid, ()))
            err = f"not configured in mykey.py ({keys})"
            self._notify(sid, err=err)
            return {"ok": False, "error": "not_configured", "service": self._state(sid, err=err)}
        self.buffers[sid] = deque(maxlen=500)
        env = {**os.environ, "PYTHONUNBUFFERED": "1", "PYTHONIOENCODING": "utf-8"}
        kw: Dict[str, Any] = dict(
            cwd=str(self.ga_root), stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
            text=True, encoding="utf-8", errors="replace", bufsize=1, env=env,
        )
        if sys.platform == "win32":
            kw["creationflags"] = subprocess.CREATE_NO_WINDOW
        proc = subprocess.Popen(svc["cmd"], **kw)
        self.procs[sid] = proc
        threading.Thread(target=self._reader, args=(sid, proc), daemon=True).start()
        self._wait_started(proc)
        item = self._state(sid)
        self._notify(sid)
        if item["status"] == "error":
            return {"ok": False, "error": item["lastError"] or "start_failed", "service": item}
        return {"ok": True, "service": item}

    def autostart_extras(self) -> None:
        """Auto-start non-IM services on bridge boot. Currently:
          - reflect/scheduler.py (drives L4 archive cron every 12h).
        IM services stay manual (need explicit mykey.py config + user opt-in)."""
        for sid in sorted(set(self._catalog) - set(self._im_catalog)):
            try:
                res = self.start_service(sid)
                tag = "ok" if res.get("ok") else f"fail: {res.get('error')}"
            except Exception as e:
                tag = f"exception {type(e).__name__}: {e}"
            print(f"[autostart] {sid}: {tag}", file=sys.stderr)

    def stop_all_extras(self) -> None:
        for sid in sorted(set(self._catalog) - set(self._im_catalog)):
            with contextlib.suppress(Exception):
                self.stop_service(sid)

    def stop_service(self, sid: str) -> dict:
        if sid not in self._catalog:
            raise KeyError(sid)
        self._stopping.add(sid)
        proc = self.procs.get(sid)
        if proc and proc.poll() is None:
            proc.terminate()
            proc.wait()
        self.procs.pop(sid, None)
        self._stopping.discard(sid)
        item = self._state(sid)
        self._notify(sid)
        return {"ok": True, "service": item}

    def read_logs(self, sid: str, tail: int = 200) -> dict:
        if sid == BRIDGE_ID:
            return {"ok": True, "lines": [f"GenericAgent bridge pid={os.getpid()}"]}
        if sid not in self._catalog:
            raise KeyError(sid)
        tail = max(1, min(int(tail or 200), 2000))
        buf = self.buffers.get(sid)
        lines = [ln.rstrip("\n") for ln in list(buf or [])[-tail:]]
        return {"ok": True, "lines": lines}


services = ServiceManager(str(DEFAULT_GA_ROOT), hub.emit)


def _bridge_shutdown_services() -> None:
    with contextlib.suppress(Exception):
        services.stop_all_extras()


atexit.register(_bridge_shutdown_services)


def emit_session_state(sess: Session, state_name: str):
    hub.emit({
        "type": "session-state",
        "sessionId": sess.id,
        "state": state_name,
        "status": sess.status,
        "seq": sess.msg_seq,
        "updatedAt": sess.updated_at,
        "title": sess.title,
    })


async def ws_handler(request):
    ws = web.WebSocketResponse(heartbeat=30)
    await ws.prepare(request)
    hub.websockets.add(ws)
    await ws.send_str(json.dumps({
        "type": "bridge-ready",
        "gaRoot": manager.ga_root,
        "mykeyPath": manager.mykey_path,
        "http": True,
        "wsEventsOnly": True,
    }, ensure_ascii=False))
    await ws.send_str(json.dumps({
        "type": "services.snapshot",
        "services": services.list_state(),
    }, ensure_ascii=False, default=str))
    async for msg in ws:
        if msg.type == WSMsgType.TEXT:
            # WS is intentionally not a data/command channel anymore.
            with contextlib.suppress(Exception):
                data = json.loads(msg.data)
                if data.get("action") == "ping":
                    await ws.send_str(json.dumps({"type": "pong", "ts": time.time()}, ensure_ascii=False))
    hub.websockets.discard(ws)
    return ws


# ---------------------------------------------------------------------------
# Transport layer: HTTP command/data API
# ---------------------------------------------------------------------------

def cors_headers():
    return {
        "Access-Control-Allow-Origin": "*",
        "Access-Control-Allow-Methods": "GET,POST,PUT,DELETE,OPTIONS",
        "Access-Control-Allow-Headers": "Content-Type",
    }


@web.middleware
async def cors_middleware(request, handler):
    if request.method == "OPTIONS":
        return web.Response(status=204, headers=cors_headers())
    resp = await handler(request)
    for k, v in cors_headers().items():
        resp.headers[k] = v
    return resp


def json_ok(data: dict, status: int = 200):
    return web.json_response(data, status=status, headers=cors_headers(), dumps=lambda x: json.dumps(x, ensure_ascii=False, default=str))


async def read_json(request) -> dict:
    if request.can_read_body:
        try:
            data = await request.json()
            return data if isinstance(data, dict) else {}
        except Exception:
            return {}
    return {}


async def status_handler(request):
    return json_ok({
        "ok": True,
        "running": True,
        "ready": True,
        "gaRoot": manager.ga_root,
        "mykeyPath": manager.mykey_path,
        "sessionCount": len(manager.sessions),
        "activeSessionId": manager.active_session_id,
        "ws": "/ws",
        "transport": {"http": True, "wsEventsOnly": True},
    })


_SETTINGS = Path.home() / ".ga_desktop_settings.json"
_UI_KEYS = ("lang", "theme", "appearance", "plain", "llmNo", "fontSize")


def _desktop_ui() -> dict:
    try:
        ui = json.loads(_SETTINGS.read_text(encoding="utf-8")).get("ui")
        return dict(ui) if isinstance(ui, dict) else {}
    except Exception:
        return {}


async def get_config_handler(request):
    profiles = manager.list_model_profiles()
    active = next((p["id"] for p in profiles if p.get("active")), manager.config.get("llmNo", 0))
    cfg = dict(manager.config)
    if "llmNo" not in cfg:
        cfg["llmNo"] = active
    cfg.update(_desktop_ui())
    return json_ok({"gaRoot": manager.ga_root, "mykeyPath": manager.mykey_path, "config": cfg})


async def save_config_handler(request):
    data = await read_json(request)
    cfg = data.get("config", data)
    if isinstance(cfg, dict):
        patch = {k: cfg[k] for k in _UI_KEYS if k in cfg}
        if patch:
            try:
                doc = json.loads(_SETTINGS.read_text(encoding="utf-8")) if _SETTINGS.is_file() else {}
                if not isinstance(doc, dict):
                    doc = {}
                ui = doc["ui"] if isinstance(doc.get("ui"), dict) else {}
                ui.update(patch)
                doc["ui"] = ui
                _SETTINGS.write_text(json.dumps(doc, ensure_ascii=False, indent=2), encoding="utf-8")
            except Exception as e:
                print(f"[bridge] save ui prefs failed: {e}", file=sys.stderr)
        manager.config.update(cfg)
    return json_ok({"ok": True, "gaRoot": manager.ga_root, "mykeyPath": manager.mykey_path, "config": manager.config})


async def model_profiles_handler(request):
    try:
        pid = request.match_info.get("id")
        if pid is not None:
            profile_id = int(pid)
            if request.method == "GET":
                return json_ok({"profile": manager.get_model_profile(profile_id)})
            if request.method == "PUT":
                return json_ok({"ok": True, **manager.update_model_profile(profile_id, await read_json(request))})
            if request.method == "DELETE":
                return json_ok({"ok": True, **manager.delete_model_profile(profile_id)})
            return json_ok({"ok": False, "error": "method not allowed"}, status=405)
        if request.method == "POST":
            return json_ok({"ok": True, **manager.add_model_profile(await read_json(request))})
        return json_ok({"profiles": manager.list_model_profiles()})
    except ValueError as e:
        return json_ok({"ok": False, "error": str(e)}, status=400)
    except Exception as e:
        return json_ok({"ok": False, "error": str(e)}, status=500)


async def mixin_handler(request):
    """聚合渠道成员管理：POST 加入 / DELETE 移出 主聚合渠道。"""
    try:
        profile_id = int(request.match_info.get("id"))
        if request.method == "POST":
            return json_ok({"ok": True, **manager.add_to_mixin(profile_id)})
        if request.method == "DELETE":
            return json_ok({"ok": True, **manager.remove_from_mixin(profile_id)})
        return json_ok({"ok": False, "error": "method not allowed"}, status=405)
    except ValueError as e:
        return json_ok({"ok": False, "error": str(e)}, status=400)
    except Exception as e:
        return json_ok({"ok": False, "error": str(e)}, status=500)


async def mixin_order_handler(request):
    """渠道组成员拖拽排序：PUT {members:[name,...]}。"""
    try:
        data = await read_json(request)
        return json_ok({"ok": True, **manager.reorder_mixin(data.get("members") or [])})
    except ValueError as e:
        return json_ok({"ok": False, "error": str(e)}, status=400)
    except Exception as e:
        return json_ok({"ok": False, "error": str(e)}, status=500)


async def list_sessions_handler(request):
    with manager.lock:
        sessions = [manager.snapshot(s, include_messages=False) for s in manager.sessions.values()]
    return json_ok({"sessions": sessions, "activeSessionId": manager.active_session_id})


async def new_session_handler(request):
    data = await read_json(request)
    sess = manager.create_session(cwd=data.get("cwd") or data.get("path"))
    return json_ok({"ok": True, "sessionId": sess.id, "session": manager.snapshot(sess)}, status=201)


async def get_session_handler(request):
    sid = request.match_info["sid"]
    sess = manager.get_session(sid)
    return json_ok({"sessionId": sid, "session": manager.snapshot(sess), "messages": list(sess.messages), "partial": sess.partial})


async def delete_session_handler(request):
    sid = request.match_info["sid"]
    return json_ok(manager.delete_session(sid))


async def patch_session_handler(request):
    sid = request.match_info["sid"]
    sess = manager.get_session(sid)
    data = await read_json(request)
    if "title" in data:
        sess.title = data["title"]
        sess.untitled = False
    if "pinned" in data:
        sess.pinned = bool(data["pinned"])
    if "untitled" in data:
        sess.untitled = bool(data["untitled"])
    if "plan_scan_baseline" in data:
        sess.plan_scan_baseline = int(data["plan_scan_baseline"])
    sess.updated_at = time.time()
    manager._persist()
    return json_ok({"ok": True, "session": manager.snapshot(sess, include_messages=False)})


async def prompt_handler(request):
    sid = request.match_info["sid"]
    data = await read_json(request)
    prompt = data.get("prompt", data.get("content", data.get("message", "")))
    images = data.get("images") or []
    display = data.get("display")
    files_meta = data.get("files") or []        # 非图片附件 [{name, path}]
    image_metas = data.get("imageMetas") or []   # 图片附件 [{name, path}]（不含 dataUrl）
    llm_no = data.get("llmNo")
    if llm_no is not None:
        llm_no = int(llm_no)
    return json_ok(manager.submit_prompt(sid, prompt, images, llm_no=llm_no, display=display,
                                          files_meta=files_meta, image_metas=image_metas))


async def messages_handler(request):
    sid = request.match_info["sid"]
    after = int(request.query.get("after") or request.query.get("afterId") or 0)
    limit = int(request.query.get("limit") or 200)
    return json_ok(manager.messages(sid, after=after, limit=limit))


async def cancel_handler(request):
    sid = request.match_info["sid"]
    return json_ok(manager.cancel(sid))


async def restore_handler(request):
    sid = request.match_info["sid"]
    return json_ok(manager.restore_context(sid))


async def plan_handler(request):
    sid = request.match_info["sid"]
    return json_ok(manager.plan_snapshot(sid))


async def path_open_handler(request):
    data = await read_json(request)
    kind = data.get("kind", "")
    mode = data.get("mode", "open")
    if kind == "mykey":
        target = Path(manager.ga_root) / "mykey.py"
        if not target.exists():
            template = Path(manager.ga_root) / "mykey_template.py"
            target = template if template.exists() else target
    elif kind == "mykeyTemplate":
        target = Path(manager.ga_root) / "mykey_template.py"
    elif kind == "upload":
        raw = Path(data.get("path") or "")
        try:
            resolved = raw.resolve()
            upload_root = _WEB_UPLOAD_DIR.resolve()
            resolved.relative_to(upload_root)
        except (ValueError, OSError):
            return json_ok({"ok": False, "error": "path not in upload dir"}, status=403)
        target = resolved
    else:
        target = Path(data.get("path") or data.get("target") or manager.ga_root)
    target = target.resolve()
    if not target.exists():
        return json_ok({"ok": False, "error": f"File not found: {target}"}, status=404)
    try:
        if mode == "reveal":
            _reveal_path_in_file_manager(target)
        elif kind == "upload":
            _open_path_default(target)  # 用户文件用系统默认程序(open 动词),避免 edit 动词 fallback 记事本
        else:
            _open_path_in_editor(target)  # mykey 等配置文件仍用编辑器(edit 动词)
    except OSError as e:
        return json_ok({"ok": False, "error": str(e), "path": str(target)}, status=500)
    return json_ok({"ok": True, "path": str(target)})


# File attachments live under GA's own temp dir (gitignored), NOT the OS temp
# dir, so they survive bridge restarts. Instead of wiping everything on startup,
# we keep files for UPLOAD_RETENTION_DAYS and only sweep stale ones.
_WEB_UPLOAD_DIR = Path(DEFAULT_GA_ROOT) / "temp" / "desktop_uploads"
_WEB_UPLOAD_DIR.mkdir(parents=True, exist_ok=True)

UPLOAD_RETENTION_DAYS = 30


def _safe_session_dir(sid: str) -> str:
    """Sanitize a session id into a safe single-level folder name."""
    s = re.sub(r"[^A-Za-z0-9_-]", "", str(sid or ""))
    return s or "_misc"


def _session_upload_dir(sid: str) -> Path:
    """Per-session upload subdir under desktop_uploads/, created on demand."""
    d = _WEB_UPLOAD_DIR / _safe_session_dir(sid)
    d.mkdir(parents=True, exist_ok=True)
    return d


def _purge_session_uploads(sid: str) -> None:
    """Best-effort: drop a session's whole upload subdir when the session is deleted."""
    import shutil
    with contextlib.suppress(Exception):
        shutil.rmtree(_WEB_UPLOAD_DIR / _safe_session_dir(sid), ignore_errors=True)


def _sweep_stale_uploads(retention_days: int = UPLOAD_RETENTION_DAYS) -> None:
    """Best-effort: delete uploaded files older than retention_days (by mtime),
    then drop empty session subdirs. Replaces the old wholesale rmtree-on-startup
    so attachments persist across restarts while temp storage can't grow forever."""
    cutoff = time.time() - retention_days * 86400
    try:
        for f in _WEB_UPLOAD_DIR.rglob("*"):
            try:
                if f.is_file() and f.stat().st_mtime < cutoff:
                    f.unlink()
            except OSError:
                pass
        for d in _WEB_UPLOAD_DIR.iterdir():
            try:
                if d.is_dir() and not any(d.iterdir()):
                    d.rmdir()
            except OSError:
                pass
    except OSError:
        pass


_sweep_stale_uploads()


async def upload_handler(request):
    """Save a file uploaded by the web client and return its absolute path.
    Body: {name: "<original filename>", dataUrl: "data:<mime>;base64,<...>", sid: "<session id>"}
    Files are grouped per session under desktop_uploads/<sid>/ so deleting a
    session can purge its attachments. Missing sid falls back to a _misc bucket.
    Returns: {ok: true, path: "<abs path>"}
    """
    try:
        data = await request.json()
        if not isinstance(data, dict):
            data = {}
    except web.HTTPRequestEntityTooLarge:
        return json_ok({"ok": False, "error": "file too large for bridge body limit"})
    except Exception as e:
        return json_ok({"ok": False, "error": f"invalid request: {e}"})
    name = (data.get("name") or "file").strip().replace("/", "_").replace("\\", "_")
    data_url = data.get("dataUrl") or ""
    if "," in data_url:
        b64 = data_url.split(",", 1)[1]
    else:
        b64 = data_url
    try:
        blob = base64.b64decode(b64)
    except Exception as e:
        return json_ok({"ok": False, "error": f"decode failed: {e}"})
    if not blob:
        return json_ok({"ok": False, "error": "empty file"})
    safe_name = name or "file"
    fpath = _session_upload_dir(data.get("sid") or "") / f"{uuid.uuid4().hex[:12]}__{safe_name}"
    fpath.write_bytes(blob)
    return json_ok({"ok": True, "path": str(fpath)})


async def upload_delete_handler(request):
    """Delete a previously-uploaded file. Path must live under _WEB_UPLOAD_DIR."""
    data = await read_json(request)
    raw = data.get("path") or ""
    try:
        target = Path(raw).resolve()
        upload_root = _WEB_UPLOAD_DIR.resolve()
        if upload_root not in target.parents:
            return json_ok({"ok": False, "error": "path outside upload dir"})
        if target.exists():
            target.unlink()
        return json_ok({"ok": True})
    except Exception as e:
        return json_ok({"ok": False, "error": str(e)})


async def upload_raw_handler(request):
    """Stream an uploaded file. inline by default (browser preview / <img>),
    ?download=1 forces a download. Path must live under _WEB_UPLOAD_DIR
    (whitelist — prevents path traversal). Works for remote browsers too,
    so it covers both 'preview after refresh' and 'download from remote'."""
    import mimetypes
    from urllib.parse import quote
    raw = request.query.get("path", "")
    try:
        target = Path(raw).resolve()
        target.relative_to(_WEB_UPLOAD_DIR.resolve())
    except (ValueError, OSError):
        return web.Response(status=403, text="path not in upload dir")
    if not target.is_file():
        return web.Response(status=404, text="file not found")
    ctype = mimetypes.guess_type(target.name)[0] or "application/octet-stream"
    disp = "attachment" if request.query.get("download") in ("1", "true") else "inline"
    orig_name = target.name.split("__", 1)[-1]  # 去掉 <uuid>__ 前缀，还原原始文件名
    return web.Response(
        body=target.read_bytes(),
        content_type=ctype,
        headers={
            "Content-Disposition": f"{disp}; filename*=UTF-8''{quote(orig_name)}",
            "Cache-Control": "no-cache",
        },
    )


def _open_path_in_editor(target: Path) -> None:
    """Open a file in the user's editor; Windows .py often has no default association."""
    import platform
    path = str(target.resolve())
    if platform.system() == "Windows":
        try:
            os.startfile(path, "edit")
            return
        except OSError:
            pass
        for cmd in (["notepad.exe", path], ["cursor.cmd", path], ["code.cmd", path], ["cursor", path], ["code", path]):
            try:
                subprocess.Popen(cmd, close_fds=True)
                return
            except (FileNotFoundError, OSError):
                continue
        raise OSError(f"No editor available to open: {path}")
    if platform.system() == "Darwin":
        subprocess.Popen(["open", path])
        return
    subprocess.Popen(["xdg-open", path])


def _reveal_path_in_file_manager(target: Path) -> None:
    """Open the system file manager and select/highlight the target file."""
    import platform
    path = str(target.resolve())
    if platform.system() == "Windows":
        subprocess.Popen(["explorer", "/select,", path])
        return
    if platform.system() == "Darwin":
        subprocess.Popen(["open", "-R", path])
        return
    # Linux: no universal "select file" command; fall back to opening parent dir
    subprocess.Popen(["xdg-open", str(target.parent)])


def _open_path_default(target: Path) -> None:
    """Open a file with the OS default app (default 'open' verb).

    For user uploads. Unlike _open_path_in_editor (which uses Windows' 'edit'
    verb and falls back to Notepad), this respects each file type's registered
    default app — PDF viewer, Word, archive tool, etc. — so binaries like pdf
    or docx no longer land in Notepad as garbage."""
    import platform
    path = str(target.resolve())
    if platform.system() == "Windows":
        os.startfile(path)  # default "open" verb = double-click behavior
        return
    if platform.system() == "Darwin":
        subprocess.Popen(["open", path])
        return
    subprocess.Popen(["xdg-open", path])


def _mykey_file() -> Path:
    root = Path(manager.ga_root)
    target = root / "mykey.py"
    if not target.is_file():
        template = root / "mykey_template.py"
        if template.is_file():
            target.write_text(template.read_text(encoding="utf-8"), encoding="utf-8")
    return target


async def mykey_get_handler(request):
    target = _mykey_file()
    content = target.read_text(encoding="utf-8") if target.is_file() else ""
    return json_ok({"content": content, "path": str(target)})


async def mykey_save_handler(request):
    data = await read_json(request)
    content = data.get("content")
    if content is None:
        return json_ok({"ok": False, "error": "missing_content"}, status=400)
    try:
        profiles = manager._save_mykey_text(str(content))
    except Exception as e:
        return json_ok({"ok": False, "error": str(e)}, status=400)
    return json_ok({"ok": True, "path": str(manager._mykey_file()), "profiles": profiles})


async def service_start_handler(request):
    body = await read_json(request)
    sid = body.get("id") or request.query.get("id")
    if not sid:
        return json_ok({"ok": False, "error": "missing_id"}, status=400)
    result = services.start_service(sid)
    if not result.get("ok"):
        return json_ok(result, status=400)
    return json_ok(result)


async def service_stop_handler(request):
    body = await read_json(request)
    sid = body.get("id") or request.query.get("id")
    if not sid:
        return json_ok({"ok": False, "error": "missing_id"}, status=400)
    return json_ok(services.stop_service(sid))


async def service_logs_handler(request):
    sid = request.query.get("id")
    if not sid:
        return json_ok({"ok": False, "error": "missing_id"}, status=400)
    tail = int(request.query.get("tail") or 200)
    return json_ok(services.read_logs(sid, tail=tail))


async def service_panel_handler(request):
    return json_ok({"services": services.list_panel_state()})


def _is_local_peer(peer: str) -> bool:
    p = (peer or "").strip()
    return p in ("127.0.0.1", "::1") or p.startswith("::ffff:127.0.0.1")


async def stop_extras_handler(request):
    if not _is_local_peer(request.remote or ""):
        return json_ok({"ok": False, "error": "forbidden"}, status=403)
    services.stop_all_extras()
    return json_ok({"ok": True})


async def start_extras_handler(request):
    if not _is_local_peer(request.remote or ""):
        return json_ok({"ok": False, "error": "forbidden"}, status=403)
    services.autostart_extras()
    return json_ok({"ok": True})


async def identity_handler(request):
    return json_ok({"ga_root": str(DEFAULT_GA_ROOT), "app_dir": str(APP_DIR), "pid": os.getpid(),
                    "build_id": os.environ.get("GA_BUILD_ID", "")})


def _exit_bridge() -> None:
    with contextlib.suppress(Exception):
        services.stop_all_extras()
    threading.Timer(0.4, lambda: os._exit(0)).start()


async def bridge_exit_handler(request):
    if not _is_local_peer(request.remote or ""):
        return json_ok({"ok": False, "error": "forbidden"}, status=403)
    _exit_bridge()
    return json_ok({"ok": True})


async def token_stats_handler(request):
    try:
        sys.path.insert(0, str(APP_DIR)) if str(APP_DIR) not in sys.path else None
        import cost_tracker
        trackers = cost_tracker.all_trackers()
        records = []
        for k, v in trackers.items():
            model = ''
            sid = k.replace('GA-', '')
            with manager.lock:
                sess = manager.sessions.get(sid)
            if sess and sess.agent:
                try: model = sess.agent.get_llm_name(model=True) or ''
                except Exception: pass
            records.append({"thread": k, "input": v.input, "output": v.output,
                            "cacheCreate": v.cache_create, "cacheRead": v.cache_read, "model": model})
    except Exception:
        records = []
    return json_ok({"records": records})


_TOKEN_HISTORY_FILE = None

def _tok_file() -> Path:
    global _TOKEN_HISTORY_FILE
    if _TOKEN_HISTORY_FILE is None:
        _TOKEN_HISTORY_FILE = Path(manager.ga_root) / "temp" / "desktop_token_history.json"
    return _TOKEN_HISTORY_FILE

async def get_token_history_handler(request):
    f = _tok_file()
    if f.is_file():
        try:
            data = json.loads(f.read_text(encoding="utf-8"))
            return json_ok(data)
        except Exception:
            pass
    return json_ok({"history": [], "snap": {}})

async def post_token_history_handler(request):
    data = await read_json(request)
    f = _tok_file()
    f.parent.mkdir(parents=True, exist_ok=True)
    f.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
    return json_ok({"ok": True})


def create_app():
    app = web.Application(middlewares=[cors_middleware], client_max_size=500 * 1024 * 1024)
    app.router.add_get("/ws", ws_handler)
    app.router.add_get("/status", status_handler)
    app.router.add_get("/config", get_config_handler)
    app.router.add_post("/config", save_config_handler)
    app.router.add_get("/model-profiles", model_profiles_handler)
    app.router.add_post("/model-profiles", model_profiles_handler)
    app.router.add_put("/model-profiles/mixin/order", mixin_order_handler)
    app.router.add_post("/model-profiles/{id}/mixin", mixin_handler)
    app.router.add_delete("/model-profiles/{id}/mixin", mixin_handler)
    app.router.add_get("/model-profiles/{id}", model_profiles_handler)
    app.router.add_put("/model-profiles/{id}", model_profiles_handler)
    app.router.add_delete("/model-profiles/{id}", model_profiles_handler)
    app.router.add_get("/sessions", list_sessions_handler)
    app.router.add_post("/session/new", new_session_handler)
    app.router.add_get("/session/{sid}", get_session_handler)
    app.router.add_delete("/session/{sid}", delete_session_handler)
    app.router.add_patch("/session/{sid}", patch_session_handler)
    app.router.add_post("/session/{sid}/prompt", prompt_handler)
    app.router.add_get("/session/{sid}/messages", messages_handler)
    app.router.add_get("/session/{sid}/plan", plan_handler)
    app.router.add_post("/session/{sid}/cancel", cancel_handler)
    app.router.add_post("/session/{sid}/restore", restore_handler)
    app.router.add_post("/path/open", path_open_handler)
    app.router.add_post("/upload", upload_handler)
    app.router.add_delete("/upload", upload_delete_handler)
    app.router.add_get("/upload/raw", upload_raw_handler)
    app.router.add_get("/token-stats", token_stats_handler)
    app.router.add_get("/token-history", get_token_history_handler)
    app.router.add_post("/token-history", post_token_history_handler)
    app.router.add_post("/services/start", service_start_handler)
    app.router.add_post("/services/stop", service_stop_handler)
    app.router.add_get("/services/logs", service_logs_handler)
    app.router.add_get("/services/panel", service_panel_handler)
    app.router.add_get("/services/mykey", mykey_get_handler)
    app.router.add_post("/services/mykey", mykey_save_handler)
    app.router.add_post("/services/stop-extras", stop_extras_handler)
    app.router.add_post("/services/start-extras", start_extras_handler)
    app.router.add_get("/services/identity", identity_handler)
    app.router.add_post("/services/bridge/exit", bridge_exit_handler)

    # Serve static frontend (desktop/static/)
    static_dir = APP_DIR / "desktop" / "static"

    async def index_handler(request):
        return web.FileResponse(
            static_dir / "index.html",
            headers={"Cache-Control": "no-cache, no-store, must-revalidate"},
        )

    app.router.add_get("/", index_handler)
    app.router.add_static("/", static_dir, show_index=False)

    async def on_startup(app):
        hub.loop = asyncio.get_running_loop()
        services.autostart_extras()

    async def on_shutdown(app):
        services.stop_all_extras()

    app.on_startup.append(on_startup)
    app.on_shutdown.append(on_shutdown)
    return app


if __name__ == "__main__":
    host = os.environ.get("BRIDGE_HOST", "127.0.0.1")
    port = int(os.environ.get("BRIDGE_PORT", "14168"))
    print(f"GenericAgent Web2 bridge: http://{host}:{port}  ws://{host}:{port}/ws", file=sys.stderr)
    web.run_app(create_app(), host=host, port=port, print=None)
