"""Windows tray launcher for DeepSeek Harness (DSH).

Starts DSH Web in the background from the cloned repository directory and
offers Open, Restart, Stop, and Exit actions from the system tray.
"""

from __future__ import annotations

import os
import json
import socket
import subprocess
import sys
import threading
import time
import webbrowser
import winreg
import re
import base64
import secrets
import mimetypes
import math
from datetime import datetime
from pathlib import Path
from tkinter import BooleanVar, END, Text, Tk, StringVar, Toplevel
from tkinter import filedialog, ttk
from urllib import error as urlerror
from urllib.parse import urlparse, urlunparse
from urllib import request as urlrequest


import pystray
import psutil
from PIL import Image, ImageDraw


BIND_HOST = "127.0.0.1"
TRAY_VERSION = "3.9.12"
DEFAULT_WEB_URL = "http://localhost:3080"
DEFAULT_PROJECT_DIR = Path.home() / "Documents" / "deepseek-harness"
PROJECT_DIR = Path(os.environ.get("DSH_PROJECT_DIR", DEFAULT_PROJECT_DIR))
STARTUP_FILE = Path(os.environ["APPDATA"]) / "Microsoft" / "Windows" / "Start Menu" / "Programs" / "Startup" / "Start DeepSeek Harness Tray.cmd"
TRAY_HOME = Path(os.environ["APPDATA"]) / "DSHTray"
TRAY_CONFIG = TRAY_HOME / "settings.json"
DSH_HOME = TRAY_HOME / "dsh-home"
NODE_DIR = Path(os.environ.get("ProgramFiles", r"C:\Program Files")) / "nodejs"
NODE_EXE = NODE_DIR / "node.exe"
NPM_CMD = NODE_DIR / "npm.cmd"
DSH_CMD = Path(os.environ["APPDATA"]) / "npm" / "dsh.cmd"
DSH_ENTRY = Path(os.environ["APPDATA"]) / "npm" / "node_modules" / "@deepseek-ai" / "dsh" / "lib" / "bin.js"
DSH_PACKAGE = Path(os.environ["APPDATA"]) / "npm" / "node_modules" / "@deepseek-ai" / "dsh" / "package.json"
DSH_SOURCE_DIR = Path.home() / "Documents" / "deepseek-harness"
DSH_BUILD_LOG = DSH_SOURCE_DIR / "auto-continue-build.log"
DSH_RUNTIME_LOG = TRAY_HOME / "dsh-startup.log"
DSH_CUSTOM_CLI = DSH_SOURCE_DIR / "apps" / "cli" / "lib" / "bin.js"
AUTO_CONTINUE_SOURCE = DSH_SOURCE_DIR / "packages" / "core" / "agent-loop" / "src" / "agent.ts"
HEADLESS_TELEMETRY_SOURCE = DSH_SOURCE_DIR / "packages" / "bundle" / "headless" / "src" / "index.ts"
TELEGRAM_HISTORY = TRAY_HOME / "telegram-task-history.json"
TELEGRAM_OUTPUT_DIR = TRAY_HOME / "telegram-results"
TELEGRAM_IMAGE_DIR = TRAY_HOME / "telegram-images"
TELEGRAM_DAILY_METRICS = TRAY_HOME / "telegram-daily-metrics.json"
DSH_WEB_METRICS = TRAY_HOME / "dsh-web-metrics.jsonl"
TELEGRAM_CHINESE_RESPONSE_RULE = (
    "请使用简体中文回答用户。回答应清楚、完整、自然；代码、命令、文件路径、专有名词和必要的原文可保留原语言。"
    "不要输出内部推理过程。"
)

DEFAULT_SETTINGS = {
    "web_url": DEFAULT_WEB_URL,
    "provider": "lmstudio",
    "model": "",
    "base_url": "http://127.0.0.1:1234/v1",
    # LM Studio accepts a harmless non-empty bearer value by default. DSH's
    # OpenAI-compatible client still requires one before it sends a request.
    "api_key": "lm-studio",
    "timeout": 30,
    "max_tokens": 102400,
    "vision_enabled": False,
    "telegram_enabled": False,
    # Keep public source generic. Existing users retain their saved owner ID.
    "telegram_user_id": "",
    "telegram_allowed_user_ids": "",
    "telegram_task_timeout": 7200,
    "telegram_daily_status_enabled": False,
    "telegram_daily_status_time": "09:00",
    "telegram_last_daily_status_date": "",
    "auto_continue": True,
    "auto_continue_limit": 1_000_000,
    "auto_update_dsh": False,
    "project_dir": str(PROJECT_DIR),
}


class TaskResourceSampler:
    """Collect machine-wide peak resources while a Telegram task is running."""
    def __init__(self) -> None:
        self.stop_event = threading.Event()
        self.thread: threading.Thread | None = None
        self.max_cpu_percent = 0.0
        self.max_dram_used_mib = 0.0
        self.max_vram_used_mib: float | None = None
        self.max_gpu_temp_c: float | None = None

    def start(self) -> None:
        psutil.cpu_percent(interval=None)
        self.thread = threading.Thread(target=self._sample, daemon=True)
        self.thread.start()

    def _sample_nvidia(self) -> None:
        try:
            result = subprocess.run(
                ["nvidia-smi", "--query-gpu=memory.used,temperature.gpu", "--format=csv,noheader,nounits"],
                stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, check=False,
                creationflags=subprocess.CREATE_NO_WINDOW, timeout=3,
            )
            if result.returncode != 0:
                return
            rows = [line.strip() for line in result.stdout.decode("utf-8", errors="replace").splitlines() if line.strip()]
            vram_values: list[float] = []
            temp_values: list[float] = []
            for row in rows:
                values = [value.strip() for value in row.split(",")]
                if len(values) >= 2:
                    try:
                        vram_values.append(float(values[0]))
                        temp_values.append(float(values[1]))
                    except ValueError:
                        continue
            if vram_values:
                self.max_vram_used_mib = max(self.max_vram_used_mib or 0.0, sum(vram_values))
            if temp_values:
                self.max_gpu_temp_c = max(self.max_gpu_temp_c or 0.0, max(temp_values))
        except (OSError, subprocess.TimeoutExpired):
            return

    def _sample(self) -> None:
        while not self.stop_event.is_set():
            self.max_cpu_percent = max(self.max_cpu_percent, psutil.cpu_percent(interval=None))
            memory = psutil.virtual_memory()
            self.max_dram_used_mib = max(self.max_dram_used_mib, memory.used / (1024 * 1024))
            self._sample_nvidia()
            self.stop_event.wait(1)

    def finish(self) -> dict[str, float | None]:
        self.stop_event.set()
        if self.thread is not None:
            self.thread.join(timeout=5)
        return {
            "cpu_peak_percent": round(self.max_cpu_percent, 1),
            "dram_peak_mib": round(self.max_dram_used_mib, 1),
            "vram_peak_mib": round(self.max_vram_used_mib, 1) if self.max_vram_used_mib is not None else None,
            "gpu_temp_peak_c": round(self.max_gpu_temp_c, 1) if self.max_gpu_temp_c is not None else None,
        }


class DSHTrayApp:
    def __init__(self) -> None:
        self.process: subprocess.Popen[bytes] | None = None
        self.secure_web_url: str | None = None
        self.telegram_thread: threading.Thread | None = None
        self.telegram_stop = threading.Event()
        self.telegram_offset: int | None = None
        self.telegram_task: subprocess.Popen[bytes] | None = None
        self.telegram_task_info: dict[str, object] | None = None
        self.telegram_queue: list[dict[str, object]] = []
        self.telegram_task_cancel = threading.Event()
        self.telegram_task_lock = threading.Lock()
        self.telegram_confirmations: dict[str, dict[str, object]] = {}
        self.telegram_input_modes: dict[tuple[str, str], str] = {}
        self.lock = threading.Lock()
        self.icon = pystray.Icon("DeepSeekHarness", self.make_icon("starting"), "DSH: starting")

    @staticmethod
    def make_icon(state: str) -> Image.Image:
        colors = {"starting": "#e6a700", "running": "#1f9d55", "stopped": "#7a7a7a", "error": "#c53030"}
        image = Image.new("RGBA", (64, 64), (0, 0, 0, 0))
        draw = ImageDraw.Draw(image)
        draw.ellipse((4, 4, 60, 60), fill="#141414")
        draw.ellipse((11, 11, 53, 53), fill=colors[state])
        draw.text((25, 16), "D", fill="white", stroke_width=1, stroke_fill="#141414")
        return image

    @staticmethod
    def web_endpoint(web_url: object) -> tuple[str, int]:
        """Validate the local browser address and return its normalised URL and port."""
        raw_url = str(web_url).strip().rstrip("/")
        parsed = urlparse(raw_url)
        if parsed.scheme != "http" or not parsed.hostname or parsed.path not in ("", "/") or parsed.query or parsed.fragment:
            raise ValueError("Open Address must be a plain local HTTP address, for example http://localhost:3080")
        if parsed.hostname.lower() not in {"localhost", "127.0.0.1", "::1"}:
            raise ValueError("Open Address must use localhost, 127.0.0.1, or ::1")
        try:
            port = parsed.port or 80
        except ValueError as exc:
            raise ValueError("Open Address has an invalid port") from exc
        if not 1 <= port <= 65535:
            raise ValueError("Open Address has an invalid port")
        return raw_url, port

    def browser_url_for_token(self, token_url: str) -> str:
        """Use the configured local browser hostname while preserving DSH's token."""
        try:
            preferred, _port = self.web_endpoint(self.load_settings()["web_url"])
            target = urlparse(token_url)
            browser = urlparse(preferred)
            if target.hostname not in {"127.0.0.1", "localhost", "::1"} or browser.hostname is None:
                return token_url
            host = browser.hostname
            netloc = host if target.port in (None, 80) else f"{host}:{target.port}"
            return urlunparse((target.scheme, netloc, target.path, target.params, target.query, target.fragment))
        except (ValueError, TypeError):
            return token_url

    @staticmethod
    def port_open(port: int) -> bool:
        try:
            with socket.create_connection((BIND_HOST, port), timeout=0.5):
                return True
        except OSError:
            return False

    @staticmethod
    def listener_pid(port: int) -> int | None:
        """Return the TCP listener PID for a local port, if Windows reports one."""
        try:
            result = subprocess.run(
                ["netstat", "-ano", "-p", "tcp"],
                stdout=subprocess.PIPE,
                stderr=subprocess.DEVNULL,
                check=False,
                creationflags=subprocess.CREATE_NO_WINDOW,
            )
            pattern = re.compile(rf"^\s*TCP\s+\S+:{port}\s+\S+\s+LISTENING\s+(\d+)\s*$", re.IGNORECASE)
            for line in result.stdout.decode("utf-8", errors="replace").splitlines():
                match = pattern.match(line)
                if match:
                    return int(match.group(1))
        except (OSError, ValueError):
            pass
        return None

    @staticmethod
    def is_dsh_process(pid: int) -> bool:
        """Avoid stopping an unrelated program that happens to use the DSH port."""
        try:
            command = (
                "(Get-CimInstance Win32_Process -Filter \"ProcessId = "
                f"{pid}\").CommandLine"
            )
            result = subprocess.run(
                ["powershell.exe", "-NoProfile", "-Command", command],
                stdout=subprocess.PIPE,
                stderr=subprocess.DEVNULL,
                check=False,
                creationflags=subprocess.CREATE_NO_WINDOW,
            )
            value = result.stdout.decode("utf-8", errors="replace").lower()
            return "deepseek-harness" in value or "@deepseek-ai\\dsh" in value
        except OSError:
            return False

    def stop_unowned_dsh_listener(self, port: int) -> bool:
        pid = self.listener_pid(port)
        if pid is None or not self.is_dsh_process(pid):
            return False
        try:
            subprocess.run(
                ["taskkill", "/PID", str(pid), "/T", "/F"],
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                check=False,
                creationflags=subprocess.CREATE_NO_WINDOW,
                timeout=5,
            )
            return True
        except (OSError, subprocess.TimeoutExpired):
            return False

    def set_status(self, state: str, title: str) -> None:
        self.icon.icon = self.make_icon(state)
        self.icon.title = title

    @staticmethod
    def position_near_tray(window: Tk | Toplevel) -> None:
        """Place a new dialog above the usual bottom-right Windows tray area."""
        class Rect(__import__("ctypes").Structure):
            _fields_ = [("left", __import__("ctypes").c_long), ("top", __import__("ctypes").c_long),
                       ("right", __import__("ctypes").c_long), ("bottom", __import__("ctypes").c_long)]

        window.update_idletasks()
        work_area = Rect()
        # SPI_GETWORKAREA returns the desktop area above the taskbar.
        if not __import__("ctypes").windll.user32.SystemParametersInfoW(0x0030, 0, __import__("ctypes").byref(work_area), 0):
            return
        width = max(window.winfo_reqwidth(), window.winfo_width())
        height = max(window.winfo_reqheight(), window.winfo_height())
        x = max(work_area.left + 10, work_area.right - width - 18)
        y = max(work_area.top + 10, work_area.bottom - height - 18)
        window.geometry(f"+{x}+{y}")

    @staticmethod
    def build_summary() -> str:
        if DSH_CUSTOM_CLI.is_file():
            return "Custom DSH build is ready."
        if not DSH_BUILD_LOG.is_file():
            return "No custom DSH build has been started."
        try:
            tail = DSH_BUILD_LOG.read_text(encoding="utf-8", errors="replace")[-12_000:].lower()
        except OSError:
            return "Build log is temporarily unavailable."
        if "command failed with exit code" in tail or " failed" in tail or "error:" in tail:
            return "Build failed — open Build Status for the error details."
        return "Custom DSH build is running or awaiting completion."

    def show_build_status(self) -> None:
        def create_window() -> None:
            root = Tk()
            root.withdraw()
            window = Toplevel(root)
            window.title("Custom DSH Build Status")
            window.geometry("760x460")
            window.minsize(560, 300)
            window.columnconfigure(0, weight=1)
            window.rowconfigure(1, weight=1)
            summary = StringVar()
            ttk.Label(window, textvariable=summary, padding=(12, 10, 12, 4), wraplength=700).grid(row=0, column=0, sticky="ew")
            output = Text(window, wrap="word", font=("Cascadia Mono", 9))
            output.grid(row=1, column=0, sticky="nsew", padx=12, pady=(4, 8))
            output.configure(state="disabled")

            def refresh() -> None:
                summary.set(self.build_summary())
                try:
                    contents = DSH_BUILD_LOG.read_text(encoding="utf-8", errors="replace")[-30_000:] if DSH_BUILD_LOG.is_file() else "No build log yet.\n"
                except OSError as exc:
                    contents = f"Cannot read build log: {exc}\n"
                if DSH_RUNTIME_LOG.is_file():
                    try:
                        contents += "\n\n--- Latest DSH startup output ---\n" + DSH_RUNTIME_LOG.read_text(encoding="utf-8", errors="replace")[-12_000:]
                    except OSError:
                        pass
                output.configure(state="normal")
                output.delete("1.0", END)
                output.insert("1.0", contents)
                output.see(END)
                output.configure(state="disabled")

            controls = ttk.Frame(window, padding=(12, 0, 12, 12))
            controls.grid(row=2, column=0, sticky="e")
            ttk.Button(controls, text="Refresh", command=refresh).grid(row=0, column=0, padx=4)
            ttk.Button(controls, text="Close", command=root.destroy).grid(row=0, column=1, padx=4)
            refresh()
            self.position_near_tray(window)
            window.protocol("WM_DELETE_WINDOW", root.destroy)
            window.mainloop()

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

    def dsh_installed(self) -> bool:
        """Check the global package files first, then fall back to npm metadata."""
        if DSH_CMD.is_file() and DSH_ENTRY.is_file():
            return True
        try:
            result = subprocess.run(
                [str(NPM_CMD), "list", "--global", "@deepseek-ai/dsh", "--depth=0", "--json"],
                stdout=subprocess.PIPE,
                stderr=subprocess.DEVNULL,
                check=False,
                creationflags=subprocess.CREATE_NO_WINDOW,
                env=self.node_environment(),
            )
            packages = json.loads(result.stdout.decode("utf-8", errors="replace"))
            return "@deepseek-ai/dsh" in packages.get("dependencies", {})
        except (FileNotFoundError, json.JSONDecodeError):
            return False

    @staticmethod
    def package_version(package_file: Path) -> str:
        try:
            version = json.loads(package_file.read_text(encoding="utf-8")).get("version")
            return str(version) if version else "unknown"
        except (OSError, json.JSONDecodeError):
            return "not installed"

    def official_dsh_version(self) -> str:
        return self.package_version(DSH_PACKAGE)

    def custom_dsh_version(self) -> str:
        return self.package_version(DSH_SOURCE_DIR / "apps" / "cli" / "package.json")

    @staticmethod
    def custom_dsh_ready() -> bool:
        return NODE_EXE.is_file() and DSH_CUSTOM_CLI.is_file()

    @staticmethod
    def auto_continue_build_ready() -> bool:
        """Confirm the custom CLI was built after the auto-continuation patch."""
        if not DSHTrayApp.custom_dsh_ready() or not AUTO_CONTINUE_SOURCE.is_file() or not HEADLESS_TELEMETRY_SOURCE.is_file():
            return False
        try:
            source = AUTO_CONTINUE_SOURCE.read_text(encoding="utf-8", errors="replace")
            telemetry = HEADLESS_TELEMETRY_SOURCE.read_text(encoding="utf-8", errors="replace")
            return (
                "DSH_AUTO_CONTINUE_OUTPUT_TOKENS" in source
                and "dsh: telemetry:" in telemetry
                and DSH_CUSTOM_CLI.stat().st_mtime >= max(AUTO_CONTINUE_SOURCE.stat().st_mtime, HEADLESS_TELEMETRY_SOURCE.stat().st_mtime)
            )
        except OSError:
            return False

    def version_summary(self) -> str:
        custom = self.custom_dsh_version()
        build_state = "ready" if self.auto_continue_build_ready() else "not built"
        return f"Tray v{TRAY_VERSION}   |   Official DSH v{self.official_dsh_version()}   |   Custom DSH v{custom} ({build_state})"

    @staticmethod
    def confirm(message: str, title: str) -> bool:
        return __import__("ctypes").windll.user32.MessageBoxW(
            None, message, title, 0x00000004 | 0x00000030
        ) == 6  # Yes/No + warning icon; IDYES

    @staticmethod
    def show_message(message: str, title: str, error: bool = False) -> None:
        def create_window() -> None:
            root = Tk()
            root.title(title)
            root.resizable(False, False)
            frame = ttk.Frame(root, padding=18)
            frame.grid(sticky="nsew")
            icon = "✖" if error else "ℹ"
            ttk.Label(frame, text=icon, font=("Segoe UI", 24)).grid(row=0, column=0, sticky="n", padx=(0, 12))
            ttk.Label(frame, text=message, justify="left", wraplength=420).grid(row=0, column=1, sticky="w")
            ttk.Button(frame, text="OK", command=root.destroy).grid(row=1, column=1, sticky="e", pady=(16, 0))
            DSHTrayApp.position_near_tray(root)
            root.protocol("WM_DELETE_WINDOW", root.destroy)
            root.mainloop()

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

    @staticmethod
    def confirm_async(message: str, title: str, on_yes: callable) -> None:
        """Use a Tk window instead of a modal Win32 box from a tray callback."""
        def create_window() -> None:
            root = Tk()
            root.title(title)
            root.resizable(False, False)
            frame = ttk.Frame(root, padding=18)
            frame.grid(sticky="nsew")
            ttk.Label(frame, text="?", font=("Segoe UI", 24)).grid(row=0, column=0, sticky="n", padx=(0, 12))
            ttk.Label(frame, text=message, justify="left", wraplength=420).grid(row=0, column=1, sticky="w")

            def finish(approved: bool) -> None:
                root.destroy()
                if approved:
                    on_yes()

            controls = ttk.Frame(frame)
            controls.grid(row=1, column=1, sticky="e", pady=(16, 0))
            ttk.Button(controls, text="Yes", command=lambda: finish(True), width=10).grid(row=0, column=0, padx=4)
            ttk.Button(controls, text="No", command=lambda: finish(False), width=10).grid(row=0, column=1, padx=4)
            DSHTrayApp.position_near_tray(root)
            root.protocol("WM_DELETE_WINDOW", lambda: finish(False))
            root.mainloop()

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

    @staticmethod
    def node_environment() -> dict[str, str]:
        environment = dict(os.environ)
        environment["PATH"] = f"{NODE_DIR};{DSH_CMD.parent};{environment.get('PATH', '')}"
        return environment

    def run_npm_operation(self, action: str, commands: list[list[str]]) -> None:
        def worker() -> None:
            self.set_status("starting", f"DSH: {action.lower()} in progress")
            try:
                last_output = ""
                for command in commands:
                    result = subprocess.run(
                        command,
                        stdout=subprocess.PIPE,
                        stderr=subprocess.STDOUT,
                        check=False,
                        creationflags=subprocess.CREATE_NO_WINDOW,
                        env=self.node_environment(),
                    )
                    last_output = result.stdout.decode("utf-8", errors="replace").strip()
                    if result.returncode != 0:
                        raise RuntimeError(last_output or "npm returned an error")
                if action == "Uninstall":
                    self.set_status("stopped", "DSH: not installed")
                    completed = "DeepSeek Harness was uninstalled."
                else:
                    self.set_status("stopped", "DSH: installed and stopped")
                    completed = f"DeepSeek Harness {action.lower()} completed."
                self.icon.notify(f"DeepSeek Harness {action.lower()} completed.", "DSH Tray")
                self.show_message(completed, "DSH Tray")
            except (FileNotFoundError, RuntimeError) as exc:
                self.set_status("error", f"DSH: {action.lower()} failed")
                self.icon.notify("Check that Node.js and npm are available.", "DSH Tray")
                detail = str(exc).strip()
                if len(detail) > 1_500:
                    detail = detail[-1_500:]
                self.show_message(f"{action} failed.\n\n{detail}", "DSH Tray", error=True)

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

    def upgrade_official_dsh(self, status_callback: callable | None = None, launch_after: bool = False) -> None:
        """Update only the official global package; never replace the custom source tree."""
        def report(message: str) -> None:
            if status_callback is not None:
                status_callback(message)

        def worker() -> None:
            self.set_status("starting", "DSH: checking for updates")
            report("Checking and upgrading the official DSH package…")
            try:
                result = subprocess.run(
                    [str(NPM_CMD), "install", "--global", "@deepseek-ai/dsh@latest"],
                    stdout=subprocess.PIPE,
                    stderr=subprocess.STDOUT,
                    check=False,
                    creationflags=subprocess.CREATE_NO_WINDOW,
                    env=self.node_environment(),
                )
                if result.returncode != 0 or not self.dsh_installed():
                    output = result.stdout.decode("utf-8", errors="replace").strip()
                    raise RuntimeError(output or "npm did not complete the upgrade")
                version = self.official_dsh_version()
                self.set_status("stopped", "DSH: update complete")
                report(f"Official DSH is up to date (v{version}).")
                self.icon.notify(f"Official DSH is up to date (v{version}).", "DSH Tray")
            except (FileNotFoundError, RuntimeError) as exc:
                self.set_status("error", "DSH: update failed")
                detail = str(exc).strip()
                report(f"Update failed: {detail[-800:]}")
                self.icon.notify("Official DSH update failed. Open Settings and try again.", "DSH Tray")
            finally:
                if launch_after:
                    self.launch()

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

    def install_with_progress_window(self) -> None:
        """Install DSH in a visible window so long npm work never looks frozen."""
        def create_window() -> None:
            root = Tk()
            root.title("Installing DeepSeek Harness")
            root.resizable(False, False)
            frame = ttk.Frame(root, padding=20)
            frame.grid(sticky="nsew")
            status = StringVar(value="Installing DeepSeek Harness globally…\nThis can take a few minutes.")
            ttk.Label(frame, textvariable=status, justify="left", wraplength=460).grid(row=0, column=0, sticky="w")
            close_button = ttk.Button(frame, text="Close", command=root.destroy, state="disabled")
            close_button.grid(row=1, column=0, sticky="e", pady=(16, 0))

            def worker() -> None:
                self.set_status("starting", "DSH: install in progress")
                try:
                    result = subprocess.run(
                        [str(NPM_CMD), "install", "--global", "@deepseek-ai/dsh"],
                        stdout=subprocess.PIPE,
                        stderr=subprocess.STDOUT,
                        check=False,
                        creationflags=subprocess.CREATE_NO_WINDOW,
                        env=self.node_environment(),
                    )
                    output = result.stdout.decode("utf-8", errors="replace").strip()
                    if result.returncode != 0 or not self.dsh_installed():
                        raise RuntimeError(output or "npm did not complete the installation")
                    self.set_status("stopped", "DSH: installed and stopped")
                    message = "Installation complete.\n\nYou can close this window, then use Open DSH from the tray menu."
                    failed = False
                except (FileNotFoundError, RuntimeError) as exc:
                    self.set_status("error", "DSH: install failed")
                    detail = str(exc).strip()
                    if len(detail) > 1_200:
                        detail = detail[-1_200:]
                    message = f"Installation failed.\n\n{detail}"
                    failed = True

                def finish() -> None:
                    root.title("DSH installation failed" if failed else "DSH installation complete")
                    status.set(message)
                    close_button.configure(state="normal")

                root.after(0, finish)

            threading.Thread(target=worker, daemon=True).start()
            self.position_near_tray(root)
            root.mainloop()

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

    def install_dsh(self, _icon: pystray.Icon | None = None, _item: object | None = None) -> None:
        if self.dsh_installed():
            self.icon.notify("DeepSeek Harness is already installed.", "DSH Tray")
            self.show_message("DeepSeek Harness is already installed globally.\n\nUse Open DSH or Settings… from the tray menu.", "DSH Tray")
            return
        self.confirm_async("Install DeepSeek Harness globally with npm?", "Install DeepSeek Harness", self.install_with_progress_window)

    def repair_dsh(self, _icon: pystray.Icon | None = None, _item: object | None = None) -> None:
        def repair() -> None:
            self.stop()
            self.run_npm_operation(
                "Repair",
                [[str(NPM_CMD), "cache", "verify"], [str(NPM_CMD), "install", "--global", "@deepseek-ai/dsh"]],
            )
        self.confirm_async("Repair DeepSeek Harness by verifying npm's cache and reinstalling the package?", "Repair DeepSeek Harness", repair)

    def reinstall_dsh(self, _icon: pystray.Icon | None = None, _item: object | None = None) -> None:
        def reinstall() -> None:
            self.stop()
            self.run_npm_operation(
                "Reinstall",
                [[str(NPM_CMD), "uninstall", "--global", "@deepseek-ai/dsh"], [str(NPM_CMD), "install", "--global", "@deepseek-ai/dsh"]],
            )
        self.confirm_async("Reinstall DeepSeek Harness? This removes and installs its global npm package.", "Reinstall DeepSeek Harness", reinstall)

    def uninstall_dsh(self, _icon: pystray.Icon | None = None, _item: object | None = None) -> None:
        def remove_dsh() -> None:
            self.stop()
            self.run_npm_operation("Uninstall", [[str(NPM_CMD), "uninstall", "--global", "@deepseek-ai/dsh"]])
        self.confirm_async(
            "Uninstall the globally installed DeepSeek Harness package? Your source folder and tray launcher will be kept.",
            "Uninstall DeepSeek Harness",
            remove_dsh,
        )

    @staticmethod
    def provider_id(value: str) -> str:
        cleaned = "".join(char.lower() if char.isalnum() else "-" for char in value).strip("-")
        return cleaned or "lmstudio"

    def load_settings(self) -> dict[str, object]:
        try:
            saved = json.loads(TRAY_CONFIG.read_text(encoding="utf-8"))
            return DEFAULT_SETTINGS | saved
        except (FileNotFoundError, json.JSONDecodeError):
            return DEFAULT_SETTINGS.copy()

    def save_settings(self, settings: dict[str, object]) -> None:
        TRAY_HOME.mkdir(parents=True, exist_ok=True)
        DSH_HOME.mkdir(parents=True, exist_ok=True)
        project_dir = str(settings["project_dir"]).strip()
        os.environ["DSH_PROJECT_DIR"] = project_dir
        try:
            with winreg.OpenKey(winreg.HKEY_CURRENT_USER, "Environment", 0, winreg.KEY_SET_VALUE) as key:
                winreg.SetValueEx(key, "DSH_PROJECT_DIR", 0, winreg.REG_SZ, project_dir)
        except OSError:
            # The current launcher can still apply the folder even if Windows cannot persist it.
            pass
        TRAY_CONFIG.write_text(json.dumps(settings, indent=2), encoding="utf-8")
        model = str(settings["model"]).strip()
        if not model:
            return
        provider = self.provider_id(str(settings["provider"]))
        base_url = str(settings["base_url"]).rstrip("/")
        try:
            max_tokens = max(1, int(settings.get("max_tokens", 102400)))
        except (TypeError, ValueError):
            max_tokens = 102400
        quote = json.dumps
        settings_yaml = (
            "# Managed by DeepSeek Harness Tray\n"
            "llm-pi-ai:\n"
            "  providers:\n"
            f"    {provider}:\n"
            f"      displayName: {quote(str(settings['provider']))}\n"
            "      api: openai-completions\n"
            "      apiKeyEnv: DSH_TRAY_API_KEY\n"
            f"      baseURL: {quote(base_url)}\n"
            "      models:\n"
            f"        - id: {quote(model)}\n"
            f"          name: {quote(model)}\n"
            + ("          input: [text, image]\n" if bool(settings.get("vision_enabled", False)) else "")
            +
            "          contextWindow: 131072\n"
            f"          maxTokens: {max_tokens}\n"
            "agent-default-model:\n"
            f"  provider: {quote(provider)}\n"
            f"  model: {quote(model)}\n"
        )
        (DSH_HOME / "settings.yaml").write_text(settings_yaml, encoding="utf-8")

    def telegram_api(self, token: str, method: str, payload: dict[str, object]) -> dict[str, object]:
        data = json.dumps(payload).encode("utf-8")
        request = urlrequest.Request(
            f"https://api.telegram.org/bot{token}/{method}", data=data,
            headers={"Content-Type": "application/json"}, method="POST",
        )
        with urlrequest.urlopen(request, timeout=35) as response:
            result = json.loads(response.read().decode("utf-8"))
        if not isinstance(result, dict) or not result.get("ok"):
            raise ValueError("Telegram API request failed")
        return result

    def telegram_send_document(self, token: str, chat_id: object, file_path: Path, caption: str) -> None:
        """Send a Markdown result as a real Telegram document without extra packages."""
        boundary = f"----DSHTray{secrets.token_hex(12)}"
        fields = [("chat_id", str(chat_id)), ("caption", caption)]
        body = bytearray()
        for name, value in fields:
            body.extend(f"--{boundary}\r\nContent-Disposition: form-data; name=\"{name}\"\r\n\r\n{value}\r\n".encode("utf-8"))
        body.extend(
            f"--{boundary}\r\nContent-Disposition: form-data; name=\"document\"; filename=\"{file_path.name}\"\r\n"
            "Content-Type: text/markdown; charset=utf-8\r\n\r\n".encode("utf-8")
        )
        body.extend(file_path.read_bytes())
        body.extend(f"\r\n--{boundary}--\r\n".encode("utf-8"))
        request = urlrequest.Request(
            f"https://api.telegram.org/bot{token}/sendDocument", data=bytes(body),
            headers={"Content-Type": f"multipart/form-data; boundary={boundary}"}, method="POST",
        )
        with urlrequest.urlopen(request, timeout=35) as response:
            result = json.loads(response.read().decode("utf-8"))
        if not isinstance(result, dict) or not result.get("ok"):
            raise ValueError("Telegram document upload failed")

    def telegram_menu_markup(self, sender: str) -> dict[str, object]:
        """Persistent Chinese reply keyboard; sensitive controls are owner-only."""
        rows: list[list[str]] = [
            ["📝 提交任务", "📷 图片分析", "📊 查看状态", "⚙️ 查看设定"],
            ["🗂️ 任务历史", "⏹️ 取消任务", "❓ 帮助", "🙈 隐藏菜单"],
        ]
        if sender == self.telegram_owner_id():
            rows.append(["🔄 重启 DSH", "⛔ 停止 DSH", "👥 白名单", "📅 每日报表"])
        return {"keyboard": rows, "resize_keyboard": True, "is_persistent": True}

    def telegram_send_menu(self, token: str, chat_id: object, sender: str, text: str = "中文功能菜单已开启。请选择一个功能。") -> None:
        self.telegram_api(token, "sendMessage", {
            "chat_id": chat_id,
            "text": text,
            "reply_markup": self.telegram_menu_markup(sender),
        })

    def telegram_hide_menu(self, token: str, chat_id: object) -> None:
        self.telegram_api(token, "sendMessage", {
            "chat_id": chat_id,
            "text": "底部菜单已隐藏。以后发送 /start 或 /help，即可再次显示菜单。",
            "reply_markup": {"remove_keyboard": True},
        })

    @staticmethod
    def telegram_profile_name(message: dict[str, object]) -> str:
        """Use Telegram's visible profile name only; never expose the username or user ID."""
        sender = message.get("from", {})
        if not isinstance(sender, dict):
            return "朋友"
        first = str(sender.get("first_name", "")).strip()
        last = str(sender.get("last_name", "")).strip()
        name = " ".join(part for part in (first, last) if part)
        return name[:80] or "朋友"

    def telegram_owner_id(self) -> str:
        return str(self.load_settings().get("telegram_user_id", "")).strip()

    def telegram_allowed_ids(self) -> set[str]:
        settings = self.load_settings()
        owner = str(settings.get("telegram_user_id", "")).strip()
        extra = str(settings.get("telegram_allowed_user_ids", ""))
        return {value.strip() for value in [owner, *extra.split(",")] if value.strip()}

    def telegram_task_timeout(self, settings: dict[str, object]) -> int:
        try:
            return min(86_400, max(30, int(settings.get("telegram_task_timeout", 7200))))
        except (TypeError, ValueError):
            return 7200

    def telegram_status_text(self) -> str:
        settings = self.load_settings()
        with self.telegram_task_lock:
            active = self.telegram_task_info
            queued = len(self.telegram_queue)
        task = "idle"
        if active:
            started = active.get("started_at")
            if isinstance(started, (float, int)):
                task = f"执行中，已运行 {int(time.monotonic() - started)} 秒"
            else:
                task = "正在启动"
        return (
            f"DSH 状态：{self.icon.title}\n"
            "Bot 状态：运行中\n"
            f"模型：{settings.get('provider', '')} / {settings.get('model', '') or '（尚未选择）'}\n"
            f"视觉分析：{'已启用' if settings.get('vision_enabled') else '未启用'}\n"
            f"Telegram 任务：{task}；等待中：{queued}\n"
            f"任务时限：{self.telegram_task_timeout(settings)} 秒"
        )

    def telegram_settings_text(self) -> str:
        settings = self.load_settings()
        return (
            "DSH Tray 设定（不会显示密钥）\n"
            f"Provider：{settings.get('provider', '')}\n"
            f"模型：{settings.get('model', '') or '（尚未选择）'}\n"
            f"Base URL：{settings.get('base_url', '')}\n"
            f"连接时限：{settings.get('timeout', 30)} 秒\n"
            f"Telegram 任务时限：{self.telegram_task_timeout(settings)} 秒\n"
            f"视觉分析：{'已启用' if settings.get('vision_enabled') else '未启用'}\n"
            f"自动续写：{'已启用' if settings.get('auto_continue') else '未启用'}"
        )

    def load_task_history(self) -> list[dict[str, object]]:
        try:
            saved = json.loads(TELEGRAM_HISTORY.read_text(encoding="utf-8"))
            return saved if isinstance(saved, list) else []
        except (OSError, json.JSONDecodeError):
            return []

    def append_task_history(self, task: dict[str, object], status: str, detail: str = "") -> None:
        TRAY_HOME.mkdir(parents=True, exist_ok=True)
        history = self.load_task_history()
        started = task.get("started_wall", datetime.now().isoformat(timespec="seconds"))
        history.append({
            "time": started,
            "kind": task.get("kind", "ask"),
            "status": status,
            "prompt": str(task.get("prompt", ""))[:240],
            "detail": detail[:240],
            "metrics": task.get("metrics", {}),
        })
        TELEGRAM_HISTORY.write_text(json.dumps(history[-100:], ensure_ascii=False, indent=2), encoding="utf-8")
        self.append_daily_metrics(task, status)

    @staticmethod
    def estimate_tokens(text: str) -> int:
        """A transparent fallback when a provider does not report real usage."""
        units = sum(1.0 if ord(char) > 0x2E7F else 0.25 for char in text)
        return max(1, math.ceil(units))

    def append_daily_metrics(self, task: dict[str, object], status: str) -> None:
        metrics = task.get("metrics")
        if not isinstance(metrics, dict):
            return
        try:
            saved = json.loads(TELEGRAM_DAILY_METRICS.read_text(encoding="utf-8"))
            daily = saved if isinstance(saved, dict) else {}
        except (OSError, json.JSONDecodeError):
            daily = {}
        day = str(task.get("started_wall", datetime.now().isoformat()))[:10]
        entries = daily.get(day, [])
        if not isinstance(entries, list):
            entries = []
        entries.append({
            "source": str(task.get("source", "Telegram")), "status": status,
            "kind": task.get("kind", "ask"), "metrics": metrics,
        })
        daily[day] = entries
        recent_days = sorted(daily)[-90:]
        TELEGRAM_DAILY_METRICS.write_text(
            json.dumps({key: daily[key] for key in recent_days}, ensure_ascii=False, indent=2), encoding="utf-8"
        )

    @staticmethod
    def load_web_daily_metrics(day: str) -> list[dict[str, object]]:
        """Read completed Web turn aggregates written by the custom DSH build."""
        if not DSH_WEB_METRICS.is_file():
            return []
        entries: list[dict[str, object]] = []
        try:
            with DSH_WEB_METRICS.open("r", encoding="utf-8") as handle:
                for line in handle:
                    try:
                        entry = json.loads(line)
                    except json.JSONDecodeError:
                        continue
                    if (isinstance(entry, dict)
                            and str(entry.get("started_wall", ""))[:10] == day
                            and isinstance(entry.get("metrics"), dict)):
                        entries.append(entry)
        except OSError:
            return []
        return entries

    def daily_report_text(self, day: str | None = None) -> str:
        report_day = day or datetime.now().date().isoformat()
        try:
            saved = json.loads(TELEGRAM_DAILY_METRICS.read_text(encoding="utf-8"))
            entries = saved.get(report_day, []) if isinstance(saved, dict) else []
        except (OSError, json.JSONDecodeError):
            entries = []
        if not isinstance(entries, list):
            entries = []
        entries = entries + self.load_web_daily_metrics(report_day)
        if not entries:
            return f"每日 DSH 报表（{report_day}）\n当天尚无 Telegram 或 Web 工作区任务记录。"
        metrics = [entry.get("metrics", {}) for entry in entries if isinstance(entry, dict) and isinstance(entry.get("metrics"), dict)]
        completed = sum(1 for entry in entries if isinstance(entry, dict) and entry.get("status") == "completed")
        failed = len(entries) - completed
        total_input = sum(int(item.get("input_tokens", 0) or 0) for item in metrics)
        total_output = sum(int(item.get("output_tokens", 0) or 0) for item in metrics)
        total_seconds = sum(float(item.get("duration_seconds", 0) or 0) for item in metrics)
        total_steps = sum(int(item.get("steps", 0) or 0) for item in metrics)
        first_token = [float(item["first_token_seconds"]) for item in metrics if isinstance(item.get("first_token_seconds"), (int, float))]
        token_rates = [float(item["output_tokens_per_second"]) for item in metrics if isinstance(item.get("output_tokens_per_second"), (int, float))]
        cpu = [float(item["cpu_peak_percent"]) for item in metrics if isinstance(item.get("cpu_peak_percent"), (int, float))]
        dram = [float(item["dram_peak_mib"]) for item in metrics if isinstance(item.get("dram_peak_mib"), (int, float))]
        vram = [float(item["vram_peak_mib"]) for item in metrics if isinstance(item.get("vram_peak_mib"), (int, float))]
        gpu_temp = [float(item["gpu_temp_peak_c"]) for item in metrics if isinstance(item.get("gpu_temp_peak_c"), (int, float))]
        cached_input = sum(int(item.get("cached_input_tokens", 0) or 0) for item in metrics)
        cache_reported = [item for item in metrics if item.get("cache_source") == "provider"]
        cache_denominator = sum(
            int(item.get("input_tokens", 0) or 0) + int(item.get("cached_input_tokens", 0) or 0)
            for item in cache_reported
        )
        source_counts: dict[str, int] = {}
        for entry in entries:
            if isinstance(entry, dict):
                source = str(entry.get("source", "Telegram"))
                source_counts[source] = source_counts.get(source, 0) + 1
        sources = "｜".join(f"{source} {count}" for source, count in sorted(source_counts.items()))
        exact = all(item.get("token_source") == "provider" for item in metrics if item)
        quality = "Provider 回报（精确）" if exact else "包含估算值"
        duration = f"{total_seconds / 60:.1f} 分钟" if total_seconds >= 60 else f"{total_seconds:.1f} 秒"
        lines = [
            f"每日 DSH 报表｜{report_day}",
            f"来源：{sources}",
            f"任务：{len(entries)}（完成 {completed}，失败／取消 {failed}）",
            f"输入 Token：{total_input:,}｜输出 Token：{total_output:,}（{quality}）",
            (f"缓存命中率：{cached_input / cache_denominator:.1%}（缓存读取 {cached_input:,} Token）"
             if cache_reported and cache_denominator else "缓存命中率：模型未提供缓存 Token 数据"),
            f"LLM 总运行：{duration}｜步骤：{total_steps:,}",
            f"首 Token：{sum(first_token) / len(first_token):.2f} 秒（平均）" if first_token else "首 Token：无可用数据",
            f"生成速度：{sum(token_rates) / len(token_rates):.2f} tok/s（平均）" if token_rates else "生成速度：无可用数据",
            f"CPU 峰值：{max(cpu):.1f}%｜DRAM 峰值：{max(dram):,.0f} MiB" if cpu and dram else "CPU／DRAM：无可用数据",
            f"VRAM 峰值：{max(vram):,.0f} MiB｜GPU 最高温：{max(gpu_temp):.0f}°C" if vram and gpu_temp else "VRAM／GPU 温度：N/A（需要此电脑的 NVIDIA nvidia-smi）",
        ]
        return "\n".join(lines)

    def publish_telegram_result(self, token: str, chat_id: object, answer: str, task: dict[str, object]) -> None:
        """Telegram accepts 4,096 chars per message; leave space for a part label."""
        chunk_size = 3900
        parts = [answer[index:index + chunk_size] for index in range(0, len(answer), chunk_size)] or ["DSH 已完成，但没有返回文字内容。"]
        for index, part in enumerate(parts, start=1):
            prefix = f"结果第 {index}/{len(parts)} 段\n" if len(parts) > 1 else ""
            self.telegram_api(token, "sendMessage", {"chat_id": chat_id, "text": prefix + part})
        if len(parts) > 1:
            TELEGRAM_OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
            file_path = TELEGRAM_OUTPUT_DIR / f"dsh-result-{datetime.now().strftime('%Y%m%d-%H%M%S')}.md"
            file_path.write_text(f"# DSH Telegram 完整结果\n\n{answer}\n", encoding="utf-8")
            self.telegram_send_document(token, chat_id, file_path, "完整结果（Markdown）")

    @staticmethod
    def terminate_process(process: subprocess.Popen[bytes]) -> None:
        try:
            subprocess.run(
                ["taskkill", "/PID", str(process.pid), "/T", "/F"], stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL, check=False, creationflags=subprocess.CREATE_NO_WINDOW, timeout=5,
            )
        except (OSError, subprocess.TimeoutExpired):
            pass

    def enqueue_telegram_task(self, token: str, chat_id: object, sender: str, kind: str, prompt: str, image: Path | None = None) -> str:
        task: dict[str, object] = {
            "token": token, "chat_id": chat_id, "sender": sender, "kind": kind,
            "source": "Telegram", "prompt": prompt, "image": image,
            "started_wall": datetime.now().isoformat(timespec="seconds"),
        }
        with self.telegram_task_lock:
            if self.telegram_task_info is not None:
                if len(self.telegram_queue) >= 1:
                    return "已有任务正在执行，而且唯一的等待位置已满。请先使用 /cancel。"
                self.telegram_queue.append(task)
                return "任务已加入单一等待队列。可使用 /status 查看进度，或用 /cancel 清除。"
            self.telegram_task_info = task
            self.telegram_task_cancel.clear()
        threading.Thread(target=self.run_telegram_task, args=(task,), daemon=True).start()
        return "任务已开始。每分钟会回报一次进度；可使用 /cancel 停止。"

    def finish_telegram_task(self) -> None:
        next_task: dict[str, object] | None = None
        with self.telegram_task_lock:
            self.telegram_task = None
            self.telegram_task_info = None
            self.telegram_task_cancel.clear()
            if self.telegram_queue:
                next_task = self.telegram_queue.pop(0)
                self.telegram_task_info = next_task
        if next_task:
            threading.Thread(target=self.run_telegram_task, args=(next_task,), daemon=True).start()

    def run_dsh_telegram_task(self, task: dict[str, object]) -> str:
        settings = self.load_settings()
        command = [str(NODE_EXE), str(DSH_CUSTOM_CLI)] if bool(settings.get("auto_continue", True)) else [str(DSH_CMD)]
        prompt = f"{TELEGRAM_CHINESE_RESPONSE_RULE}\n\n用户任务：\n{str(task['prompt'])}"
        command.extend(["--profile", "headless", prompt])
        started = time.monotonic()
        task["metrics"] = {
            "input_tokens": self.estimate_tokens(prompt), "output_tokens": 0, "token_source": "estimated",
            "cached_input_tokens": 0, "cache_source": "unavailable",
            "steps": 0, "first_token_seconds": None, "output_tokens_per_second": None,
        }
        sampler = TaskResourceSampler()
        sampler.start()
        environment = self.node_environment()
        environment["DSH_HOME"] = str(DSH_HOME)
        environment["DSH_TRAY_API_KEY"] = str(settings.get("api_key", "")).strip() or "lm-studio"
        try:
            process = subprocess.Popen(
                command, cwd=str(settings["project_dir"]), env=environment, stdout=subprocess.PIPE,
                stderr=subprocess.PIPE, creationflags=subprocess.CREATE_NO_WINDOW,
            )
            with self.telegram_task_lock:
                self.telegram_task = process
                if self.telegram_task_info is task:
                    task["started_at"] = time.monotonic()
            output = bytearray()
            diagnostics = bytearray()

            def read_stream(stream: object, destination: bytearray, diagnostic: bool = False) -> None:
                if stream is None or not hasattr(stream, "readline"):
                    return
                for line in iter(stream.readline, b""):
                    destination.extend(line)
                    if diagnostic and line.startswith(b"dsh: telemetry:"):
                        try:
                            value = json.loads(line[len(b"dsh: telemetry:"):].decode("utf-8", errors="replace"))
                            if isinstance(value, dict):
                                task["dsh_telemetry"] = value
                        except json.JSONDecodeError:
                            pass
                    elif not diagnostic and line and task["metrics"].get("first_token_seconds") is None:
                        task["metrics"]["first_token_seconds"] = round(time.monotonic() - started, 3)

            result_reader = threading.Thread(target=read_stream, args=(process.stdout, output), daemon=True)
            diagnostics_reader = threading.Thread(target=read_stream, args=(process.stderr, diagnostics, True), daemon=True)
            result_reader.start()
            diagnostics_reader.start()
            deadline = time.monotonic() + self.telegram_task_timeout(settings)
            last_progress = time.monotonic()
            while process.poll() is None:
                if self.telegram_task_cancel.is_set():
                    self.terminate_process(process)
                    raise RuntimeError("任务已取消。")
                if time.monotonic() >= deadline:
                    self.terminate_process(process)
                    raise RuntimeError(f"任务超过 {self.telegram_task_timeout(settings)} 秒时限，已停止。")
                if time.monotonic() - last_progress >= 60:
                    elapsed = int(time.monotonic() - float(task.get("started_at", time.monotonic())))
                    self.telegram_api(str(task["token"]), "sendMessage", {"chat_id": task["chat_id"], "text": f"进度：DSH 任务仍在执行中（已运行 {elapsed} 秒）。"})
                    last_progress = time.monotonic()
                time.sleep(1)
            result_reader.join(timeout=10)
            diagnostics_reader.join(timeout=10)
            if process.returncode not in (0, None):
                detail = bytes(diagnostics).decode("utf-8", errors="replace").strip()
                raise RuntimeError(detail[-700:] or "DSH 指令执行失败。")
            answer = bytes(output).decode("utf-8", errors="replace").strip() or "DSH 已完成，但没有返回文字内容。"
            telemetry = task.get("dsh_telemetry")
            if isinstance(telemetry, dict):
                task["metrics"].update({
                    "steps": int(telemetry.get("steps", 0) or 0),
                    "first_token_seconds": telemetry.get("firstTokenSeconds"),
                })
            if isinstance(telemetry, dict) and bool(telemetry.get("usageReported", False)):
                task["metrics"].update({
                    "input_tokens": int(telemetry.get("inputTokens", 0) or 0),
                    "output_tokens": int(telemetry.get("outputTokens", 0) or 0), "token_source": "provider",
                })
                if bool(telemetry.get("cacheUsageReported", False)):
                    task["metrics"].update({
                        "cached_input_tokens": int(telemetry.get("cacheReadTokens", 0) or 0),
                        "cache_source": "provider",
                    })
            else:
                task["metrics"]["output_tokens"] = self.estimate_tokens(answer)
                task["metrics"]["input_tokens"] = self.estimate_tokens(prompt)
                task["metrics"]["steps"] = max(1, int(task["metrics"].get("steps", 0) or 0))
            return answer
        finally:
            task["metrics"].update(sampler.finish())
            duration = max(0.001, time.monotonic() - started)
            task["metrics"]["duration_seconds"] = round(duration, 3)
            output_tokens = float(task["metrics"].get("output_tokens", 0) or 0)
            first_token = task["metrics"].get("first_token_seconds")
            generation = duration - float(first_token) if isinstance(first_token, (int, float)) else duration
            task["metrics"]["output_tokens_per_second"] = round(output_tokens / max(0.001, generation), 3) if output_tokens else None

    def download_telegram_photo(self, token: str, photo: dict[str, object]) -> Path:
        file_id = str(photo.get("file_id", ""))
        if not file_id:
            raise ValueError("Telegram photo did not contain a file ID")
        result = self.telegram_api(token, "getFile", {"file_id": file_id}).get("result", {})
        remote_path = str(result.get("file_path", "")) if isinstance(result, dict) else ""
        if not remote_path:
            raise ValueError("Telegram could not provide the image file")
        with urlrequest.urlopen(f"https://api.telegram.org/file/bot{token}/{remote_path}", timeout=60) as response:
            content = response.read(20 * 1024 * 1024 + 1)
        if len(content) > 20 * 1024 * 1024:
            raise ValueError("Image is larger than the 20 MB safety limit")
        TELEGRAM_IMAGE_DIR.mkdir(parents=True, exist_ok=True)
        suffix = Path(remote_path).suffix.lower() or ".jpg"
        file_path = TELEGRAM_IMAGE_DIR / f"telegram-image-{datetime.now().strftime('%Y%m%d-%H%M%S')}-{secrets.token_hex(3)}{suffix}"
        file_path.write_bytes(content)
        return file_path

    def run_telegram_vision_task(self, task: dict[str, object]) -> str:
        settings = self.load_settings()
        started = time.monotonic()
        sampler = TaskResourceSampler()
        sampler.start()
        task["metrics"] = {
            "input_tokens": 0, "output_tokens": 0, "token_source": "estimated", "steps": 1,
            "cached_input_tokens": 0, "cache_source": "unavailable",
            "first_token_seconds": None, "output_tokens_per_second": None,
        }
        if not bool(settings.get("vision_enabled", False)):
            raise RuntimeError("视觉分析尚未启用。请先在 Settings 启用 Enable Vision / Image Analysis。")
        model = str(settings.get("model", "")).strip()
        if not model:
            raise RuntimeError("请先在 Settings 选择支援视觉分析的模型。")
        image = task.get("image")
        if not isinstance(image, Path) or not image.is_file():
            raise RuntimeError("The Telegram image file is unavailable.")
        mime = mimetypes.guess_type(image.name)[0] or "image/jpeg"
        encoded = base64.b64encode(image.read_bytes()).decode("ascii")
        user_prompt = str(task.get("prompt", "")).strip() or "请仔细分析这张图片，并描述重要细节。"
        prompt = f"{TELEGRAM_CHINESE_RESPONSE_RULE}\n\n用户请求：\n{user_prompt}"
        payload = {
            "model": model,
            "messages": [{"role": "user", "content": [
                {"type": "text", "text": prompt},
                {"type": "image_url", "image_url": {"url": f"data:{mime};base64,{encoded}"}},
            ]}],
        }
        endpoint = str(settings.get("base_url", "")).rstrip("/") + "/chat/completions"
        headers = {"Content-Type": "application/json"}
        api_key = str(settings.get("api_key", "")).strip()
        if api_key:
            headers["Authorization"] = f"Bearer {api_key}"
        request = urlrequest.Request(endpoint, data=json.dumps(payload).encode("utf-8"), headers=headers, method="POST")
        timeout = max(30, min(self.telegram_task_timeout(settings), 7200))
        with self.telegram_task_lock:
            if self.telegram_task_info is task:
                task["started_at"] = time.monotonic()
        try:
            with urlrequest.urlopen(request, timeout=timeout) as response:
                result = json.loads(response.read().decode("utf-8"))
            choices = result.get("choices", []) if isinstance(result, dict) else []
            message = choices[0].get("message", {}) if choices and isinstance(choices[0], dict) else {}
            content = message.get("content", "") if isinstance(message, dict) else ""
            if isinstance(content, list):
                content = "\n".join(str(item.get("text", "")) for item in content if isinstance(item, dict))
            if not isinstance(content, str) or not content.strip():
                raise RuntimeError("The selected model returned no vision text.")
            usage = result.get("usage", {}) if isinstance(result, dict) else {}
            if isinstance(usage, dict) and isinstance(usage.get("prompt_tokens"), int):
                task["metrics"].update({
                    "input_tokens": usage.get("prompt_tokens", 0), "output_tokens": usage.get("completion_tokens", 0),
                    "token_source": "provider",
                })
                details = usage.get("prompt_tokens_details", {})
                if isinstance(details, dict) and isinstance(details.get("cached_tokens"), int):
                    task["metrics"].update({
                        "cached_input_tokens": details["cached_tokens"], "cache_source": "provider",
                    })
            else:
                task["metrics"].update({"input_tokens": self.estimate_tokens(prompt), "output_tokens": self.estimate_tokens(content)})
            return content.strip()
        finally:
            task["metrics"].update(sampler.finish())
            duration = max(0.001, time.monotonic() - started)
            task["metrics"]["duration_seconds"] = round(duration, 3)
            task["metrics"]["output_tokens_per_second"] = round(float(task["metrics"].get("output_tokens", 0) or 0) / duration, 3)

    def run_telegram_task(self, task: dict[str, object]) -> None:
        token, chat_id = str(task["token"]), task["chat_id"]
        try:
            answer = self.run_telegram_vision_task(task) if task.get("kind") == "vision" else self.run_dsh_telegram_task(task)
            if self.telegram_task_cancel.is_set():
                raise RuntimeError("任务已取消。")
            self.publish_telegram_result(token, chat_id, answer, task)
            self.append_task_history(task, "completed")
        except Exception as exc:
            message = str(exc)[:700]
            self.telegram_api(token, "sendMessage", {"chat_id": chat_id, "text": f"任务失败：{message}"})
            self.append_task_history(task, "cancelled" if "取消" in message or "cancelled" in message.lower() else "failed", message)
        finally:
            self.finish_telegram_task()

    def telegram_cancel(self, sender: str) -> str:
        with self.telegram_task_lock:
            active = self.telegram_task_info
            if active is None:
                return "目前没有正在执行或等待中的任务。"
            if sender != self.telegram_owner_id() and str(active.get("sender", "")) != sender:
                return "只有任务发起者或 Telegram Owner 可以取消这个任务。"
            self.telegram_task_cancel.set()
            self.telegram_queue.clear()
        return "已请求取消。正在运行的 DSH 任务会停止，等待队列也已清空。"

    def telegram_history_text(self) -> str:
        history = self.load_task_history()[-10:]
        if not history:
            return "目前还没有 Telegram 任务历史。"
        lines = ["最近的 Telegram 任务："]
        for entry in reversed(history):
            lines.append(f"• {entry.get('time', '')} — {entry.get('kind', 'ask')} — {entry.get('status', '')}: {entry.get('prompt', '')}")
        return "\n".join(lines)

    def telegram_whitelist_command(self, text: str, sender: str) -> str:
        if sender != self.telegram_owner_id():
            return "只有 Telegram Owner 可以管理白名单。"
        settings = self.load_settings()
        tokens = text.split()
        extras = {value.strip() for value in str(settings.get("telegram_allowed_user_ids", "")).split(",") if value.strip()}
        if len(tokens) == 1 or tokens[1].lower() == "list":
            return "Owner：" + self.telegram_owner_id() + "\n额外允许用户：" + (", ".join(sorted(extras)) or "（无）")
        if len(tokens) < 3 or not tokens[2].isdigit():
            return "用法：/whitelist list、/whitelist add <数字-user-id> 或 /whitelist remove <数字-user-id>。"
        target, action = tokens[2], tokens[1].lower()
        if target == self.telegram_owner_id():
            return "Owner 永远拥有权限，不能在这里移除。"
        if action == "add":
            extras.add(target)
        elif action in {"remove", "delete"}:
            extras.discard(target)
        else:
            return "请使用 add 或 remove。"
        settings["telegram_allowed_user_ids"] = ", ".join(sorted(extras))
        self.save_settings(settings)
        return f"白名单已更新。额外允许用户：{', '.join(sorted(extras)) or '（无）'}"

    def telegram_confirmation(self, token: str, chat_id: object, sender: str, action: str) -> None:
        if sender != self.telegram_owner_id():
            self.telegram_api(token, "sendMessage", {"chat_id": chat_id, "text": "只有 Telegram Owner 可以控制 DSH。"})
            return
        nonce = secrets.token_urlsafe(8)
        self.telegram_confirmations[nonce] = {"action": action, "chat_id": chat_id, "expires": time.monotonic() + 60}
        label = "重新启动 DSH" if action == "restart" else "停止 DSH"
        self.telegram_api(token, "sendMessage", {
            "chat_id": chat_id,
            "text": f"确认要{label}吗？此确认会在 60 秒后失效。",
            "reply_markup": {"inline_keyboard": [[
                {"text": f"确认{label}", "callback_data": f"dsh:{action}:{nonce}"},
                {"text": "取消", "callback_data": f"dsh:cancel:{nonce}"},
            ]]},
        })

    def handle_telegram_callback(self, token: str, callback: dict[str, object]) -> None:
        sender = str(callback.get("from", {}).get("id", "")) if isinstance(callback.get("from"), dict) else ""
        callback_id = str(callback.get("id", ""))
        data = str(callback.get("data", ""))
        message = callback.get("message", {})
        chat_id = message.get("chat", {}).get("id") if isinstance(message, dict) and isinstance(message.get("chat"), dict) else None
        parts = data.split(":", 2)
        if len(parts) != 3 or parts[0] != "dsh" or sender != self.telegram_owner_id() or chat_id is None:
            self.telegram_api(token, "answerCallbackQuery", {"callback_query_id": callback_id, "text": "没有权限。"})
            return
        action, nonce = parts[1], parts[2]
        confirmation = self.telegram_confirmations.pop(nonce, None)
        if not confirmation or confirmation.get("chat_id") != chat_id or float(confirmation.get("expires", 0)) < time.monotonic():
            self.telegram_api(token, "answerCallbackQuery", {"callback_query_id": callback_id, "text": "确认已过期。"})
            return
        if action == "cancel":
            self.telegram_api(token, "answerCallbackQuery", {"callback_query_id": callback_id, "text": "已取消。"})
            return
        if action not in {"restart", "stop"} or confirmation.get("action") != action:
            self.telegram_api(token, "answerCallbackQuery", {"callback_query_id": callback_id, "text": "无效确认。"})
            return
        self.telegram_api(token, "answerCallbackQuery", {"callback_query_id": callback_id, "text": "已确认。"})
        threading.Thread(target=self.restart if action == "restart" else self.stop, daemon=True).start()
        self.telegram_api(token, "sendMessage", {"chat_id": chat_id, "text": f"已请求 {'重新启动' if action == 'restart' else '停止'} DSH。"})

    def maybe_send_daily_status(self, token: str) -> None:
        settings = self.load_settings()
        if not bool(settings.get("telegram_daily_status_enabled", False)):
            return
        now = datetime.now()
        expected = str(settings.get("telegram_daily_status_time", "09:00"))
        if now.strftime("%H:%M") != expected or str(settings.get("telegram_last_daily_status_date", "")) == now.date().isoformat():
            return
        owner = self.telegram_owner_id()
        if not owner:
            return
        self.telegram_api(token, "sendMessage", {"chat_id": owner, "text": self.daily_report_text() + "\n\n当前状态\n" + self.telegram_status_text()})
        settings["telegram_last_daily_status_date"] = now.date().isoformat()
        self.save_settings(settings)

    def telegram_worker(self) -> None:
        token = os.environ.get("DSH_TELEGRAM_BOT_TOKEN", "").strip()
        if not token:
            self.set_status("error", "DSH: Telegram Token not set")
            return
        while not self.telegram_stop.is_set():
            try:
                self.maybe_send_daily_status(token)
                payload: dict[str, object] = {"timeout": 25, "allowed_updates": ["message", "callback_query"]}
                if self.telegram_offset is not None: payload["offset"] = self.telegram_offset
                updates = self.telegram_api(token, "getUpdates", payload).get("result", [])
                for update in updates if isinstance(updates, list) else []:
                    if not isinstance(update, dict): continue
                    self.telegram_offset = int(update.get("update_id", 0)) + 1
                    callback = update.get("callback_query")
                    if isinstance(callback, dict):
                        self.handle_telegram_callback(token, callback)
                        continue
                    message = update.get("message", {})
                    sender = str(message.get("from", {}).get("id", "")) if isinstance(message, dict) else ""
                    chat_id = message.get("chat", {}).get("id") if isinstance(message, dict) else None
                    text = str(message.get("text", "")).strip() if isinstance(message, dict) else ""
                    if chat_id is None or sender not in self.telegram_allowed_ids(): continue
                    menu_key = (str(chat_id), sender)
                    photos = message.get("photo", []) if isinstance(message, dict) else []
                    if isinstance(photos, list) and photos:
                        try:
                            largest = max((photo for photo in photos if isinstance(photo, dict)), key=lambda photo: int(photo.get("file_size", 0)))
                            image = self.download_telegram_photo(token, largest)
                            self.telegram_input_modes.pop(menu_key, None)
                            reply = self.enqueue_telegram_task(token, chat_id, sender, "vision", text, image)
                        except Exception as exc:
                            reply = f"无法启动图片分析：{str(exc)[:500]}"
                        self.telegram_api(token, "sendMessage", {"chat_id": chat_id, "text": reply})
                        continue
                    if text == "📝 提交任务":
                        self.telegram_input_modes[menu_key] = "ask"
                        self.telegram_api(token, "sendMessage", {"chat_id": chat_id, "text": "请直接输入你要交给 DSH 的任务内容。"})
                        continue
                    if text == "📷 图片分析":
                        self.telegram_input_modes[menu_key] = "vision"
                        self.telegram_api(token, "sendMessage", {"chat_id": chat_id, "text": "请发送一张图片；你可以在图片 caption 写下分析要求。"})
                        continue
                    if text == "🙈 隐藏菜单":
                        self.telegram_hide_menu(token, chat_id)
                        continue
                    button_commands = {
                        "📊 查看状态": "/status", "⚙️ 查看设定": "/settings", "🗂️ 任务历史": "/history",
                        "⏹️ 取消任务": "/cancel", "🔄 重启 DSH": "/restart", "⛔ 停止 DSH": "/stop",
                        "👥 白名单": "/whitelist list", "❓ 帮助": "/help", "📅 每日报表": "/daily",
                    }
                    text = button_commands.get(text, text)
                    pending_mode = self.telegram_input_modes.get(menu_key)
                    if pending_mode == "ask" and text and not text.startswith("/"):
                        self.telegram_input_modes.pop(menu_key, None)
                        reply = self.enqueue_telegram_task(token, chat_id, sender, "ask", text)
                        self.telegram_api(token, "sendMessage", {"chat_id": chat_id, "text": reply})
                        continue
                    if pending_mode == "vision" and text and not text.startswith("/"):
                        self.telegram_api(token, "sendMessage", {"chat_id": chat_id, "text": "请直接发送图片；可在图片 caption 写下你的分析要求。"})
                        continue
                    command = text.split(maxsplit=1)[0].split("@", 1)[0].lower() if text else ""
                    argument = text[len(text.split(maxsplit=1)[0]):].strip() if text else ""
                    if command == "/status": reply = self.telegram_status_text()
                    elif command == "/settings": reply = self.telegram_settings_text()
                    elif command == "/daily": reply = self.daily_report_text()
                    elif command == "/help" or command == "/start":
                        name = self.telegram_profile_name(message) if isinstance(message, dict) else "朋友"
                        self.telegram_send_menu(
                            token, chat_id, sender,
                            f"你好，{name}！请问有什么可以帮到你？\n\n"
                            "你可以提交文字任务、传送图片进行分析，或选择下方功能。输入 /daily 可查看今天的统计报表。"
                            "长输出会每 3,900 字符分段发送，并附上 Markdown 档案。",
                        )
                        continue
                    elif command == "/ask":
                        reply = self.enqueue_telegram_task(token, chat_id, sender, "ask", argument) if argument else "请在 /ask 后面输入任务内容。"
                    elif command == "/cancel": reply = self.telegram_cancel(sender)
                    elif command == "/history": reply = self.telegram_history_text()
                    elif command == "/whitelist": reply = self.telegram_whitelist_command(text, sender)
                    elif command in {"/restart", "/stop"}:
                        self.telegram_confirmation(token, chat_id, sender, command[1:])
                        continue
                    else: reply = "无法识别这个指令。请发送 /help。"
                    self.telegram_api(token, "sendMessage", {"chat_id": chat_id, "text": reply})
            except Exception:
                time.sleep(5)

    def start_telegram_bot(self) -> None:
        if self.telegram_thread and self.telegram_thread.is_alive():
            if not self.telegram_stop.is_set():
                return
            previous_worker = self.telegram_thread

            def start_after_stop() -> None:
                previous_worker.join(timeout=30)
                if not previous_worker.is_alive():
                    self.start_telegram_bot()

            threading.Thread(target=start_after_stop, daemon=True).start()
            return
        if not bool(self.load_settings().get("telegram_enabled", False)): return
        self.telegram_stop.clear()
        self.telegram_thread = threading.Thread(target=self.telegram_worker, daemon=True)
        self.telegram_thread.start()

    def stop_telegram_bot(self) -> None:
        """Stop polling without stopping DSH itself; the next enable starts a fresh worker."""
        self.telegram_stop.set()

    def show_telegram_setup_guide(self, parent: Tk | Toplevel) -> None:
        guide = Toplevel(parent)
        guide.title("Telegram Bot Setup Guide")
        guide.resizable(True, True)
        guide.minsize(620, 420)
        guide.columnconfigure(0, weight=1)
        guide.rowconfigure(0, weight=1)
        text = Text(guide, wrap="word", font=("Segoe UI", 10), padx=14, pady=12)
        text.grid(row=0, column=0, sticky="nsew")
        text.insert("1.0", """Telegram Bot Setup — Step by Step

1. Open Telegram and search for @BotFather.
2. Send /newbot, then follow the prompts to choose a name and username.
3. BotFather gives you a Bot Token. Keep it private — never paste it into GitHub or send it to anyone.
4. On this Windows PC, open PowerShell and run:

   [Environment]::SetEnvironmentVariable(
     \"DSH_TELEGRAM_BOT_TOKEN\",
     \"paste-your-BotFather-token-here\",
     \"User\"
   )

5. Close and reopen the Tray application so it can read the new environment variable.
6. In Telegram, open your new bot, press Start, and send /start once.
7. Get your numeric Telegram user ID in PowerShell:

   $result = Invoke-RestMethod "https://api.telegram.org/bot$env:DSH_TELEGRAM_BOT_TOKEN/getUpdates?timeout=30"
   $result.result.message.from.id

   The number printed is your Telegram user ID. Enter it in Telegram Owner User ID.
8. Return to @BotFather and send /setjoingroups, select your bot, then choose Disable. This prevents the bot from being added to groups.
9. In this Settings window, tick Enable Telegram Bot, then click Save.
10. Send /status to your bot. It should reply with the current DSH state.

11. Test DSH Headless before using /ask. For the Tray's LM Studio configuration, run:

   $env:DSH_HOME = "$env:APPDATA\\DSHTray\\dsh-home"
   $env:DSH_TRAY_API_KEY = "lm-studio"
   dsh --profile headless "Reply only: Headless test successful."

   If you see MISSING_CREDENTIAL for deepseek-official, DSH is using its default official route instead of the Tray's LM Studio settings. Run the two environment-variable lines above in the same PowerShell window, then repeat the test.

Switching back to the official DeepSeek API

• Open DSH Web, go to Models, choose the official DeepSeek provider, and enter your official API key there.
• Or set it for the current PowerShell session:

   $env:DEEPSEEK_API_KEY = "your-official-DeepSeek-key"

• For a permanent user-level variable, use:

   [Environment]::SetEnvironmentVariable("DEEPSEEK_API_KEY", "your-official-DeepSeek-key", "User")

• Restart the Tray after changing providers. Do not share or commit the API key.

Security tips

• Keep the bot private; do not add it to groups.
• The owner can add or remove additional allowed user IDs with /whitelist.
• The Bot Token is read from Windows environment variables, not stored in this app's Settings or source code.
""")
        text.configure(state="disabled")
        controls = ttk.Frame(guide, padding=10)
        controls.grid(row=1, column=0, sticky="e")
        ttk.Button(controls, text="Close", command=guide.destroy).grid(row=0, column=0)
        self.position_near_tray(guide)

    def discover_models(self, settings: dict[str, object]) -> tuple[bool, list[str], str]:
        base_url = str(settings["base_url"]).rstrip("/")
        try:
            timeout = max(1, int(settings["timeout"]))
        except (TypeError, ValueError):
            timeout = 30
        if not base_url.startswith(("http://", "https://")):
            return False, [], "Base URL must start with http:// or https://"
        try:
            api_key = str(settings.get("api_key", "")).strip()
            headers = {"Authorization": f"Bearer {api_key}"} if api_key else {}
            with urlrequest.urlopen(urlrequest.Request(f"{base_url}/models", headers=headers), timeout=timeout) as response:
                payload = json.loads(response.read().decode("utf-8", errors="replace"))
                entries = payload.get("data", []) if isinstance(payload, dict) else []
                models = sorted(
                    {str(entry["id"]) for entry in entries if isinstance(entry, dict) and entry.get("id")}
                )
                return True, models, f"Connected — {len(models)} model(s) reported."
        except (urlerror.URLError, TimeoutError, ValueError, json.JSONDecodeError) as exc:
            return False, [], f"Connection failed: {exc}"

    @staticmethod
    def lmstudio_native_url(base_url: object, path: str) -> str:
        """Translate an OpenAI-compatible /v1 URL to LM Studio's native API."""
        parsed = urlparse(str(base_url).strip())
        if parsed.scheme not in {"http", "https"} or not parsed.netloc:
            raise ValueError("Base URL must be a complete HTTP URL")
        return f"{parsed.scheme}://{parsed.netloc}/api/v1{path}"

    def lmstudio_native_request(self, settings: dict[str, object], path: str, payload: dict[str, object] | None = None) -> dict[str, object]:
        url = self.lmstudio_native_url(settings["base_url"], path)
        api_key = str(settings.get("api_key", "")).strip()
        headers = {"Accept": "application/json"}
        if api_key:
            headers["Authorization"] = f"Bearer {api_key}"
        data = None if payload is None else json.dumps(payload).encode("utf-8")
        if data is not None:
            headers["Content-Type"] = "application/json"
        request = urlrequest.Request(url, data=data, headers=headers, method="GET" if data is None else "POST")
        timeout = max(1, int(settings.get("timeout", 30)))
        with urlrequest.urlopen(request, timeout=timeout) as response:
            decoded = json.loads(response.read().decode("utf-8", errors="replace"))
        if not isinstance(decoded, dict):
            raise ValueError("LM Studio returned an unexpected response")
        return decoded

    def lmstudio_local_models(self, settings: dict[str, object]) -> list[str]:
        response = self.lmstudio_native_request(settings, "/models")
        entries = response.get("models", [])
        if not isinstance(entries, list):
            raise ValueError("LM Studio returned no model list")
        return sorted(str(item["key"]) for item in entries if isinstance(item, dict) and item.get("type") == "llm" and isinstance(item.get("key"), str))

    def lmstudio_load_model(self, settings: dict[str, object]) -> str:
        model = str(settings.get("model", "")).strip()
        if not model:
            raise ValueError("Choose a model first. Use List Local Models to see downloaded models.")
        response = self.lmstudio_native_request(settings, "/models/load", {"model": model})
        return f"Loaded: {response.get('instance_id', model)}"

    def lmstudio_unload_model(self, settings: dict[str, object]) -> str:
        model = str(settings.get("model", "")).strip()
        if not model:
            raise ValueError("Choose the loaded model to unload first.")
        response = self.lmstudio_native_request(settings, "/models")
        for item in response.get("models", []):
            if not isinstance(item, dict):
                continue
            instances = item.get("loaded_instances", [])
            matching_instance = next((instance for instance in instances if isinstance(instance, dict) and instance.get("id") == model), None)
            if item.get("key") != model and matching_instance is None:
                continue
            if not instances:
                raise ValueError("That model is not currently loaded.")
            instance = matching_instance or instances[0]
            if not isinstance(instance, dict) or not isinstance(instance.get("id"), str):
                raise ValueError("LM Studio did not return a usable loaded instance.")
            unloaded = self.lmstudio_native_request(settings, "/models/unload", {"instance_id": instance["id"]})
            return f"Unloaded: {unloaded.get('instance_id', instance['id'])}"
        raise ValueError("The selected model was not found in LM Studio.")

    def test_connection(self, settings: dict[str, object]) -> tuple[bool, str]:
        ok, models, message = self.discover_models(settings)
        return ok, message

    def show_settings(self, _icon: pystray.Icon | None = None, _item: object | None = None) -> None:
        def create_window() -> None:
            settings = self.load_settings()
            root = Tk()
            root.withdraw()
            window = Toplevel(root)
            window.title("DeepSeek Harness Tray Settings")
            window.resizable(True, True)
            window.minsize(520, 260)
            window.columnconfigure(0, weight=1)
            window.rowconfigure(0, weight=1)
            frame = ttk.Frame(window, padding=16)
            frame.grid(sticky="nsew")
            frame.columnconfigure(1, weight=1)
            ttk.Label(frame, text=self.version_summary(), foreground="#555555").grid(
                row=0, column=0, columnspan=2, sticky="w", pady=(0, 10)
            )
            auto_build_status = ttk.Label(frame)
            auto_build_status.grid(row=1, column=0, columnspan=2, sticky="w", pady=(0, 8))

            def refresh_auto_continue_build_status() -> None:
                if self.auto_continue_build_ready():
                    auto_build_status.configure(
                        text="✓ Auto Continue on Token Limit: custom build is ready",
                        foreground="#16803c",
                    )
                else:
                    auto_build_status.configure(
                        text="✗ Auto Continue on Token Limit: custom build is not ready",
                        foreground="#b42318",
                    )

            refresh_auto_continue_build_status()
            fields = [
                ("Open Address", "web_url"),
                ("Provider", "provider"),
                ("Model", "model"),
                ("Base URL", "base_url"),
                ("API Key", "api_key"),
                ("Timeout (seconds)", "timeout"),
                ("Max Output Tokens", "max_tokens"),
                ("Enable Vision / Image Analysis", "vision_enabled"),
                ("Enable Telegram Bot", "telegram_enabled"),
                ("Telegram Owner User ID", "telegram_user_id"),
                ("Telegram Allowed User IDs (comma separated)", "telegram_allowed_user_ids"),
                ("Telegram Task Timeout (seconds)", "telegram_task_timeout"),
                ("Daily Telegram Status", "telegram_daily_status_enabled"),
                ("Daily Status Time (HH:MM)", "telegram_daily_status_time"),
                ("Auto Continue on Token Limit", "auto_continue"),
                ("Auto Continue Limit", "auto_continue_limit"),
                ("Automatically Update Official DSH", "auto_update_dsh"),
                ("Project Folder", "project_dir"),
            ]
            variables: dict[str, StringVar] = {}
            auto_continue = BooleanVar(value=bool(settings["auto_continue"]))
            vision_enabled = BooleanVar(value=bool(settings["vision_enabled"]))
            telegram_enabled = BooleanVar(value=bool(settings["telegram_enabled"]))
            daily_status_enabled = BooleanVar(value=bool(settings["telegram_daily_status_enabled"]))
            model_picker: ttk.Combobox | None = None
            first_field_row = 2
            for field_row, (label, key) in enumerate(fields, start=first_field_row):
                row = field_row
                ttk.Label(frame, text=label).grid(row=row, column=0, sticky="w", padx=(0, 10), pady=5)
                variable = StringVar(value=str(settings[key]))
                variables[key] = variable
                width = 44 if key == "base_url" else 30
                if key == "model":
                    model_picker = ttk.Combobox(frame, textvariable=variable, width=42)
                    model_picker.grid(row=row, column=1, sticky="ew", pady=5)
                elif key == "api_key":
                    ttk.Entry(frame, textvariable=variable, width=width, show="•").grid(row=row, column=1, sticky="ew", pady=5)
                elif key == "auto_continue":
                    ttk.Checkbutton(frame, variable=auto_continue, text="Continue automatically after an output limit").grid(row=row, column=1, sticky="w", pady=5)
                elif key == "vision_enabled":
                    ttk.Checkbutton(
                        frame,
                        variable=vision_enabled,
                        text="This selected model supports image input",
                    ).grid(row=row, column=1, sticky="w", pady=5)
                elif key == "telegram_enabled":
                    telegram_controls = ttk.Frame(frame)
                    telegram_controls.grid(row=row, column=1, sticky="w", pady=5)
                    ttk.Checkbutton(telegram_controls, variable=telegram_enabled, text="Use DSH_TELEGRAM_BOT_TOKEN from Windows environment").grid(row=0, column=0)
                    ttk.Button(telegram_controls, text="Setup Guide…", command=lambda: self.show_telegram_setup_guide(window)).grid(row=0, column=1, padx=(10, 0))
                elif key == "telegram_daily_status_enabled":
                    ttk.Checkbutton(frame, variable=daily_status_enabled, text="Send the owner a scheduled DSH / Bot / model daily report").grid(row=row, column=1, sticky="w", pady=5)
                elif key == "auto_update_dsh":
                    auto_update = BooleanVar(value=bool(settings["auto_update_dsh"]))
                    ttk.Checkbutton(frame, variable=auto_update, text="Check for official DSH updates whenever the tray starts").grid(row=row, column=1, sticky="w", pady=5)
                elif key == "project_dir":
                    folder_controls = ttk.Frame(frame)
                    folder_controls.grid(row=row, column=1, sticky="ew", pady=5)
                    folder_controls.columnconfigure(0, weight=1)
                    ttk.Entry(folder_controls, textvariable=variable, width=44).grid(row=0, column=0, sticky="ew")

                    def browse_folder(target: StringVar = variable) -> None:
                        selected = filedialog.askdirectory(
                            parent=window,
                            initialdir=target.get().strip() or str(Path.home() / "Documents"),
                            title="Choose DSH Project Folder",
                        )
                        if selected:
                            target.set(selected)

                    ttk.Button(folder_controls, text="Browse…", command=browse_folder).grid(row=0, column=1, padx=(8, 0))
                else:
                    ttk.Entry(frame, textvariable=variable, width=width).grid(row=row, column=1, sticky="ew", pady=5)
            status_row = first_field_row + len(fields)
            ttk.Label(
                frame,
                text="Telegram long output is sent automatically in 3,900-character parts, with a full Markdown file attached.",
                foreground="#555555",
                wraplength=520,
            ).grid(row=status_row, column=0, columnspan=2, sticky="w", pady=(8, 2))
            status_row += 1
            status = StringVar(value="Telegram Task Timeout defaults to 7,200 seconds (2 hours); the upper limit is 86,400 seconds.")
            ttk.Label(frame, textvariable=status, wraplength=520).grid(row=status_row, column=0, columnspan=2, sticky="w", pady=(10, 8))
            folder_path_variables: list[StringVar] = []

            def folder_row(row: int, label: str, folder: Path) -> None:
                ttk.Label(frame, text=label).grid(row=row, column=0, sticky="w", padx=(0, 10), pady=5)
                path_var = StringVar(value=str(folder))
                folder_path_variables.append(path_var)
                controls = ttk.Frame(frame)
                controls.grid(row=row, column=1, sticky="ew", pady=5)
                controls.columnconfigure(0, weight=1)
                entry = ttk.Entry(controls, textvariable=path_var, state="readonly")
                entry.grid(row=0, column=0, sticky="ew")

                def open_folder(target: Path = folder) -> None:
                    if target.is_dir():
                        subprocess.Popen(["explorer.exe", str(target)], creationflags=subprocess.CREATE_NO_WINDOW)
                    else:
                        status.set(f"Folder was not found: {target}")

                ttk.Button(controls, text="Open Folder", command=open_folder).grid(row=0, column=1, padx=(8, 0))

            folder_row(status_row + 1, "Official DSH Folder", DSH_PACKAGE.parent)
            folder_row(status_row + 2, "Auto Continue Custom Build Folder", DSH_SOURCE_DIR)

            def current_settings() -> dict[str, object]:
                try:
                    timeout = max(1, int(variables["timeout"].get().strip()))
                except ValueError:
                    timeout = 30
                try:
                    max_tokens = max(1, int(variables["max_tokens"].get().strip()))
                except ValueError:
                    max_tokens = 102400
                try:
                    auto_continue_limit = max(1, int(variables["auto_continue_limit"].get().strip()))
                except ValueError:
                    auto_continue_limit = 1_000_000
                try:
                    telegram_task_timeout = min(86_400, max(30, int(variables["telegram_task_timeout"].get().strip())))
                except ValueError:
                    telegram_task_timeout = 7200
                return {
                    "web_url": variables["web_url"].get().strip(),
                    "provider": variables["provider"].get().strip() or "lmstudio",
                    "model": variables["model"].get().strip(),
                    "base_url": variables["base_url"].get().strip(),
                    "api_key": variables["api_key"].get().strip(),
                    "timeout": timeout,
                    "max_tokens": max_tokens,
                    "vision_enabled": vision_enabled.get(),
                    "telegram_enabled": telegram_enabled.get(),
                    "telegram_user_id": variables["telegram_user_id"].get().strip(),
                    "telegram_allowed_user_ids": variables["telegram_allowed_user_ids"].get().strip(),
                    "telegram_task_timeout": telegram_task_timeout,
                    "telegram_daily_status_enabled": daily_status_enabled.get(),
                    "telegram_daily_status_time": variables["telegram_daily_status_time"].get().strip(),
                    "telegram_last_daily_status_date": str(settings.get("telegram_last_daily_status_date", "")),
                    "auto_continue": auto_continue.get(),
                    "auto_continue_limit": auto_continue_limit,
                    "auto_update_dsh": auto_update.get(),
                    "project_dir": variables["project_dir"].get().strip(),
                }

            def save() -> None:
                values = current_settings()
                try:
                    self.web_endpoint(values["web_url"])
                except ValueError as exc:
                    status.set(str(exc))
                    return
                if not str(values["base_url"]).startswith(("http://", "https://")):
                    status.set("Base URL must start with http:// or https://")
                    return
                if not str(values["api_key"]):
                    status.set("API Key is required. LM Studio can use the default: lm-studio")
                    return
                if int(values["max_tokens"]) > 131072:
                    status.set("Max Output Tokens cannot exceed the configured 131072 context window.")
                    return
                if int(values["auto_continue_limit"]) > 1_000_000:
                    status.set("Auto Continue Limit cannot exceed 1,000,000 tokens.")
                    return
                if not re.fullmatch(r"(?:[01]\\d|2[0-3]):[0-5]\\d", str(values["telegram_daily_status_time"])):
                    status.set("Daily Status Time must use 24-hour HH:MM format, for example 09:00.")
                    return
                if not Path(str(values["project_dir"])).is_dir():
                    status.set("Project Folder must be an existing folder.")
                    return
                self.save_settings(values)
                if values["telegram_enabled"]:
                    self.start_telegram_bot()
                else:
                    self.stop_telegram_bot()
                refresh_auto_continue_build_status()
                status.set("Saved. Telegram output is sent in 3,900-character parts; long results also include a Markdown file.")

            def upgrade_now() -> None:
                values = current_settings()
                if bool(values["auto_continue"]):
                    status.set("Official DSH upgrades are paused while Auto Continue is enabled, so the custom build is not replaced.")
                    return
                self.upgrade_official_dsh(lambda message: window.after(0, status.set, message))

            def test() -> None:
                values = current_settings()
                status.set("Testing connection…")

                def run_test() -> None:
                    ok, models, message = self.discover_models(values)

                    def update() -> None:
                        if ok and model_picker is not None:
                            model_picker["values"] = models
                            if models and not variables["model"].get().strip():
                                variables["model"].set(models[0])
                        status.set(message)

                    window.after(0, update)

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

            def refresh_models() -> None:
                status.set("Loading models…")

                def load() -> None:
                    ok, models, message = self.discover_models(current_settings())

                    def update() -> None:
                        if ok and model_picker is not None:
                            model_picker["values"] = models
                            if models and not variables["model"].get().strip():
                                variables["model"].set(models[0])
                        status.set(message)

                    window.after(0, update)

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

            def run_native_action(action: str) -> None:
                values = current_settings()
                status.set(f"{action}…")

                def run() -> None:
                    try:
                        if action == "Listing local models":
                            models = self.lmstudio_local_models(values)
                            message = f"Found {len(models)} downloaded LLM model(s). Choose one, then Load Model."
                        elif action == "Loading model":
                            models = []
                            message = self.lmstudio_load_model(values)
                        else:
                            models = []
                            message = self.lmstudio_unload_model(values)
                    except (urlerror.URLError, TimeoutError, ValueError, json.JSONDecodeError) as exc:
                        models = []
                        message = f"{action} failed: {exc}"

                    def update() -> None:
                        if models and model_picker is not None:
                            model_picker["values"] = models
                        status.set(message)

                    window.after(0, update)

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

            controls = ttk.Frame(frame)
            controls.grid(row=status_row + 3, column=0, columnspan=2, sticky="e", pady=(4, 0))
            ttk.Button(controls, text="Use Documents", command=lambda: variables["project_dir"].set(str(Path.home() / "Documents"))).grid(row=0, column=0, padx=4)
            ttk.Button(controls, text="Save", command=save).grid(row=0, column=1, padx=4)
            ttk.Button(controls, text="Refresh Models", command=refresh_models).grid(row=0, column=2, padx=4)
            ttk.Button(controls, text="Test Connection", command=test).grid(row=0, column=3, padx=4)
            ttk.Button(controls, text="Build Status…", command=self.show_build_status).grid(row=0, column=4, padx=4)
            ttk.Button(controls, text="Check & Upgrade DSH", command=upgrade_now).grid(row=0, column=5, padx=4)
            ttk.Button(controls, text="Close", command=root.destroy).grid(row=0, column=6, padx=4)
            model_controls = ttk.Frame(frame)
            model_controls.grid(row=status_row + 4, column=0, columnspan=2, sticky="e", pady=(6, 0))
            ttk.Button(model_controls, text="List Local Models", command=lambda: run_native_action("Listing local models")).grid(row=0, column=0, padx=4)
            ttk.Button(model_controls, text="Load Model", command=lambda: run_native_action("Loading model")).grid(row=0, column=1, padx=4)
            ttk.Button(model_controls, text="Unload Model", command=lambda: run_native_action("Unloading model")).grid(row=0, column=2, padx=4)
            self.position_near_tray(window)
            window.protocol("WM_DELETE_WINDOW", root.destroy)
            window.mainloop()

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

    def launch(self) -> None:
        with self.lock:
            settings = self.load_settings()
            try:
                _web_url, web_port = self.web_endpoint(settings["web_url"])
            except ValueError:
                self.set_status("error", "DSH: invalid Open Address")
                return
            if self.port_open(web_port):
                if self.secure_web_url:
                    self.set_status("running", "DSH: running")
                else:
                    self.set_status("error", f"DSH: port {web_port} is occupied")
                    self.icon.notify("A previous DSH or another app is using this port. Choose Restart DSH.", "DSH Tray")
                return
            project_dir = Path(str(settings["project_dir"])).expanduser()
            if not project_dir.is_dir():
                self.set_status("error", f"DSH: folder not found")
                return
            use_custom_dsh = bool(settings.get("auto_continue", True))
            if use_custom_dsh and not self.custom_dsh_ready():
                self.set_status("error", "DSH: custom build required")
                self.icon.notify("Auto Continue needs the custom DSH build. Open Settings → Build Status.", "DSH Tray")
                return
            if not use_custom_dsh and not self.dsh_installed():
                self.set_status("error", "DSH: not installed")
                self.icon.notify("Choose Install DeepSeek Harness from the tray menu.", "DSH Tray")
                return
            self.set_status("starting", "DSH: starting")
            try:
                launch_env = self.node_environment()
                launch_env["DSH_HOME"] = str(DSH_HOME)
                launch_env["DSH_TRAY_WEB_METRICS_FILE"] = str(DSH_WEB_METRICS)
                launch_env["DSH_TRAY_API_KEY"] = str(settings.get("api_key", "")).strip() or "lm-studio"
                launch_env["DSH_AUTO_CONTINUE_OUTPUT_TOKENS"] = (
                    str(settings.get("auto_continue_limit", 1_000_000))
                    if bool(settings.get("auto_continue", True)) else "0"
                )
                # Always bind DSH to the local machine. The separately
                # configurable browser URL defaults to localhost, whose Origin
                # is accepted by DSH's privileged workspace-picker endpoint.
                command = [str(NODE_EXE), str(DSH_CUSTOM_CLI)] if use_custom_dsh else [str(DSH_CMD)]
                command.extend(["web", "--no-open", "--host", BIND_HOST, "--port", str(web_port)])
                self.secure_web_url = None
                self.process = subprocess.Popen(
                    command,
                    cwd=project_dir,
                    env=launch_env,
                    stdin=subprocess.DEVNULL,
                    stdout=subprocess.PIPE,
                    stderr=subprocess.STDOUT,
                    creationflags=subprocess.CREATE_NO_WINDOW,
                )
            except FileNotFoundError:
                self.set_status("error", "DSH: npx was not found")
                return
        threading.Thread(target=self.wait_for_server, args=(web_port,), daemon=True).start()
        threading.Thread(target=self.read_secure_web_url, daemon=True).start()

    def read_secure_web_url(self) -> None:
        """Open the one-time token URL printed by current DSH Web releases."""
        process = self.process
        if process is None or process.stdout is None:
            return
        deadline = time.monotonic() + 60
        recent: list[str] = []
        try:
            while time.monotonic() < deadline:
                line = process.stdout.readline()
                if not line:
                    break
                message = line.decode("utf-8", errors="replace").strip()
                recent.append(message)
                recent = recent[-80:]
                match = re.match(r"^dsh web:\s+(http://\S+?)(?:\s+\(LAN:|$)", message)
                if match:
                    self.secure_web_url = self.browser_url_for_token(match.group(1))
                    webbrowser.open(self.secure_web_url)
                    return
        except OSError:
            return
        if process.poll() is not None:
            try:
                TRAY_HOME.mkdir(parents=True, exist_ok=True)
                DSH_RUNTIME_LOG.write_text("\n".join(recent), encoding="utf-8")
            except OSError:
                pass
            self.set_status("error", "DSH: failed to start")
            self.icon.notify("DSH stopped during startup. Open Settings → Build Status for details.", "DSH Tray")

    def wait_for_server(self, web_port: int) -> None:
        for _ in range(45):
            if self.port_open(web_port):
                self.set_status("running", "DSH: running")
                return
            time.sleep(1)
        self.set_status("error", "DSH: did not start")

    def stop(self) -> None:
        with self.lock:
            if self.process and self.process.poll() is None:
                try:
                    subprocess.run(
                        ["taskkill", "/PID", str(self.process.pid), "/T", "/F"],
                        stdout=subprocess.DEVNULL,
                        stderr=subprocess.DEVNULL,
                        check=False,
                        creationflags=subprocess.CREATE_NO_WINDOW,
                        timeout=5,
                    )
                except subprocess.TimeoutExpired:
                    pass
            self.process = None
            try:
                _web_url, web_port = self.web_endpoint(self.load_settings()["web_url"])
            except ValueError:
                web_port = 3080
            stopped_previous = self.stop_unowned_dsh_listener(web_port)
            self.set_status("stopped", "DSH: stopped" if stopped_previous or not self.port_open(web_port) else f"DSH: port {web_port} is in use")

    def force_exit(self) -> None:
        """Ensure the windowless tray process terminates after the icon loop stops."""
        self.icon.stop()

        def terminate() -> None:
            time.sleep(0.25)
            os._exit(0)

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

    def open_browser(self, _icon: pystray.Icon | None = None, _item: object | None = None) -> None:
        try:
            web_url, web_port = self.web_endpoint(self.load_settings()["web_url"])
        except ValueError as exc:
            self.show_message(str(exc), "DSH Tray", error=True)
            return
        if self.port_open(web_port):
            if self.secure_web_url:
                webbrowser.open(self.secure_web_url)
            else:
                self.show_message(
                    "DSH is already running, but this Tray session does not have its secure browser link. "
                    "Choose Restart DSH once, then the authenticated page will open automatically.",
                    "DSH Tray",
                )
            return
        self.launch()

    def open_when_ready(self, web_url: str, web_port: int) -> None:
        for _ in range(45):
            if self.port_open(web_port):
                webbrowser.open(web_url)
                return
            time.sleep(1)

    def restart(self, _icon: pystray.Icon | None = None, _item: object | None = None) -> None:
        self.stop()
        self.launch()

    def stop_from_menu(self, _icon: pystray.Icon | None = None, _item: object | None = None) -> None:
        self.stop()

    def exit(self, _icon: pystray.Icon | None = None, _item: object | None = None) -> None:
        self.stop()
        self.force_exit()

    def uninstall(self, _icon: pystray.Icon | None = None, _item: object | None = None) -> None:
        """Remove only this launcher and its Windows startup entry."""
        self.confirm_async(
            "This will stop DSH, remove its Windows startup entry, and delete this tray launcher. "
            "Your DeepSeek Harness source folder will not be changed. Continue?",
            "Uninstall DeepSeek Harness Tray",
            self.complete_tray_uninstall,
        )

    def complete_tray_uninstall(self) -> None:
        launcher_path = Path(sys.executable if getattr(sys, "frozen", False) else __file__).resolve()
        cleanup_script = Path(os.environ["TEMP"]) / "uninstall_deepseek_harness_tray.cmd"
        cleanup_script.write_text(
            "@echo off\n"
            "timeout /t 2 /nobreak >nul\n"
            f'del /f /q "{STARTUP_FILE}" >nul 2>&1\n'
            f'del /f /q "{launcher_path}" >nul 2>&1\n'
            'del /f /q "%~f0" >nul 2>&1\n',
            encoding="ascii",
        )
        subprocess.Popen(
            ["cmd.exe", "/c", str(cleanup_script)],
            creationflags=subprocess.CREATE_NO_WINDOW,
        )
        self.stop()
        self.force_exit()

    def run(self) -> None:
        self.icon.menu = pystray.Menu(
            pystray.MenuItem("Open DSH", self.open_browser, default=True),
            pystray.MenuItem("Settings…", self.show_settings),
            pystray.Menu.SEPARATOR,
            pystray.MenuItem("Install DeepSeek Harness", self.install_dsh, enabled=lambda item: not self.dsh_installed()),
            pystray.MenuItem("Repair DeepSeek Harness", self.repair_dsh, enabled=lambda item: self.dsh_installed()),
            pystray.MenuItem("Reinstall DeepSeek Harness", self.reinstall_dsh, enabled=lambda item: self.dsh_installed()),
            pystray.MenuItem("Uninstall DeepSeek Harness", self.uninstall_dsh, enabled=lambda item: self.dsh_installed()),
            pystray.Menu.SEPARATOR,
            pystray.MenuItem("Restart DSH", self.restart),
            pystray.MenuItem("Stop DSH", self.stop_from_menu),
            pystray.Menu.SEPARATOR,
            pystray.MenuItem("Uninstall Tray Launcher", self.uninstall),
            pystray.MenuItem("Exit", self.exit),
        )
        settings = self.load_settings()
        self.start_telegram_bot()
        # The custom auto-continuation work is built from the local source
        # checkout. Do not overwrite that workflow with an official npm update.
        if bool(settings.get("auto_update_dsh", False)) and not bool(settings.get("auto_continue", True)):
            self.upgrade_official_dsh(launch_after=True)
        else:
            threading.Thread(target=self.launch, daemon=True).start()
        self.icon.run()


if __name__ == "__main__":
    DSHTrayApp().run()