Upload 6 files

Dockerfile

@@ -0,0 +1,18 @@
+FROM python:3.12-slim
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY server.py extract_token.py extract_token_remote.py get_token.py ./
+
+# HF Spaces 需要监听 7860 端口，/data 是持久化目录
+ENV PROXY_PORT=7860
+ENV TOKEN_FILE_PATH=/data/token.json
+
+RUN mkdir -p /data
+
+EXPOSE 7860
+
+CMD ["python", "server.py"]

extract_token.py

@@ -0,0 +1,129 @@
+#!/usr/bin/env python3
+"""
+从运行中的 WorkBuddy 提取 Token（通过 CDP）。
+
+用法:
+    python extract_token.py                     # 默认 CDP 端口 9222
+    python extract_token.py --port 9223         # 自定义端口
+    python extract_token.py --save              # 提取并保存到 data/token.json
+
+前提: WorkBuddy 以 --remote-debugging-port=9222 启动。
+"""
+
+import argparse
+import asyncio
+import json
+import sys
+from pathlib import Path
+
+if sys.platform == "win32":
+    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
+
+try:
+    import httpx
+except ImportError:
+    sys.exit("请先安装依赖: pip install httpx")
+
+
+async def extract(cdp_port: int = 9222) -> dict | None:
+    base = f"http://127.0.0.1:{cdp_port}"
+    async with httpx.AsyncClient() as client:
+        try:
+            resp = await client.get(f"{base}/json", timeout=5)
+        except httpx.ConnectError:
+            print(f"无法连接 CDP ({base})，请确认 WorkBuddy 已用调试模式启动")
+            return None
+        targets = resp.json()
+
+    ws_url = None
+    for t in targets:
+        if t.get("type") == "page" and "workbench" in t.get("url", ""):
+            ws_url = t.get("webSocketDebuggerUrl")
+            break
+    if not ws_url:
+        for t in targets:
+            if t.get("type") == "page":
+                ws_url = t.get("webSocketDebuggerUrl")
+                break
+    if not ws_url:
+        print("未找到可用的 CDP 页面目标")
+        return None
+
+    try:
+        import websockets
+    except ImportError:
+        sys.exit("请安装 websockets: pip install websockets")
+
+    async with websockets.connect(ws_url) as ws:
+        cmd = {
+            "id": 1,
+            "method": "Runtime.evaluate",
+            "params": {
+                "expression": """
+                    (async () => {
+                        try {
+                            const s = await window.vscode.ipcRenderer.invoke(
+                                'vscode:genie:auth:getSession'
+                            );
+                            return JSON.stringify(s);
+                        } catch(e) {
+                            return JSON.stringify({error: e.message});
+                        }
+                    })()
+                """,
+                "awaitPromise": True,
+                "returnByValue": True,
+            },
+        }
+        await ws.send(json.dumps(cmd))
+        result = json.loads(await asyncio.wait_for(ws.recv(), timeout=10))
+
+    value = result.get("result", {}).get("result", {}).get("value", "")
+    if not value:
+        print("CDP 返回为空")
+        return None
+
+    session = json.loads(value)
+    if session.get("error"):
+        print(f"CDP 错误: {session['error']}")
+        return None
+
+    return session
+
+
+def main():
+    parser = argparse.ArgumentParser(description="从 WorkBuddy 提取 Token")
+    parser.add_argument("--port", type=int, default=9222, help="CDP 端口 (默认 9222)")
+    parser.add_argument("--save", action="store_true", help="保存到 data/token.json")
+    args = parser.parse_args()
+
+    session = asyncio.run(extract(args.port))
+    if not session:
+        sys.exit(1)
+
+    # 兼容新版（嵌套 auth）和旧版（扁平）结构
+    auth = session.get("auth", session)
+    access_token = auth.get("accessToken", "")
+    refresh_token = auth.get("refreshToken", "")
+
+    if not access_token:
+        print("未获取到 accessToken")
+        sys.exit(1)
+
+    print(f"accessToken:  {access_token[:40]}...  (len={len(access_token)})")
+    print(f"refreshToken: {refresh_token[:40]}...  (len={len(refresh_token)})")
+
+    if args.save:
+        import time
+        out = Path(__file__).parent / "data" / "token.json"
+        out.parent.mkdir(parents=True, exist_ok=True)
+        out.write_text(json.dumps({
+            "access_token": access_token,
+            "refresh_token": refresh_token,
+            "saved_at": time.strftime("%Y-%m-%d %H:%M:%S"),
+        }, indent=2))
+        print(f"\n已保存到 {out}")
+
+
+if __name__ == "__main__":
+    main()

extract_token_remote.py

@@ -0,0 +1,205 @@
+#!/usr/bin/env python3
+"""
+从远程 WorkBuddy 实例提取 Token。
+
+支持两种模式：
+1. 本地模式：自动检测本机 WorkBuddy
+2. 远程模式：连接远程 Mac 的 CDP 端口
+
+用法:
+    # 本机自动检测
+    python extract_token_remote.py
+
+    # 远程 Mac（需要先在目标机器启动 WorkBuddy 调试模式）
+    python extract_token_remote.py --host 192.168.1.100 --port 9222 --save
+
+前置条件（远程 Mac）：
+    /Applications/WorkBuddy.app/Contents/MacOS/Electron --remote-debugging-port=9222
+
+    注意：可能需要配置防火墙允许 9222 端口入站。
+"""
+
+import argparse
+import asyncio
+import json
+import sys
+import time
+from pathlib import Path
+
+if sys.platform == "win32":
+    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
+
+try:
+    import httpx
+except ImportError:
+    sys.exit("请先安装依赖: pip install httpx")
+
+try:
+    import websockets
+except ImportError:
+    sys.exit("请先安装依赖: pip install websockets")
+
+
+async def extract(cdp_host: str = "127.0.0.1", cdp_port: int = 9222) -> dict | None:
+    base = f"http://{cdp_host}:{cdp_port}"
+
+    print(f"正在连接 CDP ({base})...")
+
+    async with httpx.AsyncClient() as client:
+        try:
+            resp = await client.get(f"{base}/json", timeout=10)
+        except httpx.ConnectError as e:
+            print(f"❌ 无法连接 CDP: {e}")
+            print(f"\n请确保远程 Mac 已启动 WorkBuddy 调试模式：")
+            print(f"  /Applications/WorkBuddy.app/Contents/MacOS/Electron --remote-debugging-port={cdp_port}")
+            return None
+        targets = resp.json()
+
+    ws_url = None
+    for t in targets:
+        if t.get("type") == "page" and "workbench" in t.get("url", ""):
+            ws_url = t.get("webSocketDebuggerUrl")
+            if ws_url:
+                if cdp_host != "127.0.0.1" and cdp_host != "localhost":
+                    ws_url = ws_url.replace("127.0.0.1", cdp_host)
+                print(f"✅ 找到 Workbench 页面")
+                break
+
+    if not ws_url:
+        for t in targets:
+            if t.get("type") == "page":
+                ws_url = t.get("webSocketDebuggerUrl")
+                if ws_url:
+                    if cdp_host != "127.0.0.1" and cdp_host != "localhost":
+                        ws_url = ws_url.replace("127.0.0.1", cdp_host)
+                    print(f"⚠️  未找到 Workbench，找到通用页面")
+                    break
+
+    if not ws_url:
+        print("❌ 未找到可用的 CDP 页面目标")
+        return None
+
+    print(f"正在通过 WebSocket 提取 Token...")
+
+    async with websockets.connect(ws_url, ping_timeout=30) as ws:
+        cmd = {
+            "id": 1,
+            "method": "Runtime.evaluate",
+            "params": {
+                "expression": """
+                    (async () => {
+                        try {
+                            const s = await window.vscode.ipcRenderer.invoke(
+                                'vscode:genie:auth:getSession'
+                            );
+                            return JSON.stringify(s);
+                        } catch(e) {
+                            return JSON.stringify({error: e.message});
+                        }
+                    })()
+                """,
+                "awaitPromise": True,
+                "returnByValue": True,
+            },
+        }
+        await ws.send(json.dumps(cmd))
+        result = json.loads(await asyncio.wait_for(ws.recv(), timeout=30))
+
+    value = result.get("result", {}).get("result", {}).get("value", "")
+    if not value:
+        print("❌ CDP 返回为空")
+        return None
+
+    session = json.loads(value)
+    if session.get("error"):
+        print(f"❌ CDP 错误: {session['error']}")
+        return None
+
+    return session
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="从 WorkBuddy 提取 Token（支持远程）",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        epilog="""
+示例:
+  # 本机自动检测
+  python extract_token_remote.py
+
+  # 远程 Mac
+  python extract_token_remote.py --host 192.168.1.100 --save
+
+  # 自定义端口
+  python extract_token_remote.py --host 192.168.1.100 --port 9223 --save
+        """
+    )
+    parser.add_argument(
+        "--host", "-H",
+        default="127.0.0.1",
+        help="CDP 主机地址 (默认: 127.0.0.1)"
+    )
+    parser.add_argument(
+        "--port", "-p",
+        type=int,
+        default=9222,
+        help="CDP 端口 (默认: 9222)"
+    )
+    parser.add_argument(
+        "--save", "-s",
+        action="store_true",
+        help="保存到 data/token.json"
+    )
+    parser.add_argument(
+        "--output", "-o",
+        help="保存到指定文件 (默认: data/token.json)"
+    )
+    args = parser.parse_args()
+
+    is_remote = args.host not in ("127.0.0.1", "localhost")
+
+    if is_remote:
+        print(f"\n🔗 远程模式")
+        print(f"   目标: {args.host}:{args.port}")
+        print(f"   注意: 确保目标 Mac 已启动 WorkBuddy 调试模式\n")
+    else:
+        print(f"\n🔗 本地模式\n")
+
+    session = asyncio.run(extract(args.host, args.port))
+    if not session:
+        sys.exit(1)
+
+    auth = session.get("auth", session)
+    access_token = auth.get("accessToken", "")
+    refresh_token = auth.get("refreshToken", "")
+
+    if not access_token:
+        print("❌ 未获取到 accessToken")
+        sys.exit(1)
+
+    print(f"\n✅ Token 提取成功！")
+    print(f"   accessToken:  {access_token[:40]}... (len={len(access_token)})")
+    print(f"   refreshToken: {refresh_token[:40] if refresh_token else 'N/A'}... "
+          f"(len={len(refresh_token) if refresh_token else 0})")
+
+    if args.save or args.output:
+        import time as time_module
+        out_path = Path(args.output) if args.output else Path(__file__).parent / "data" / "token.json"
+        out_path.parent.mkdir(parents=True, exist_ok=True)
+        out_path.write_text(json.dumps({
+            "access_token": access_token,
+            "refresh_token": refresh_token,
+            "saved_at": time_module.strftime("%Y-%m-%d %H:%M:%S"),
+            "source_host": args.host if is_remote else "localhost",
+            "source_port": args.port,
+        }, indent=2))
+        print(f"\n💾 已保存到: {out_path}")
+
+    if is_remote:
+        print(f"\n📋 接下来在本地 Mac 上：")
+        print(f"   1. 将生成的 data/token.json 复制到本项目")
+        print(f"   2. 启动代理: python server.py")
+
+
+if __name__ == "__main__":
+    main()

get_token.py

@@ -0,0 +1,402 @@
+#!/usr/bin/env python3
+"""
+WorkBuddy Token 提取工具（傻瓜版）
+
+全自动检测和安装依赖，用户只需双击运行即可。
+
+功能：
+- 自动检测并安装 Python（如未安装）
+- 自动安装脚本依赖
+- 自动启动 WorkBuddy（调试模式）
+- 自动提取 Token 并保存到桌面
+"""
+
+import os
+import sys
+import time
+import json
+import subprocess
+import asyncio
+import platform
+from pathlib import Path
+
+
+def run_cmd(cmd, check=True, shell=True):
+    """执行命令并返回结果"""
+    try:
+        result = subprocess.run(
+            cmd, shell=shell, capture_output=True, text=True, timeout=300
+        )
+        if check and result.returncode != 0:
+            return False, result.stderr
+        return True, result.stdout
+    except subprocess.TimeoutExpired:
+        return False, "命令执行超时"
+    except Exception as e:
+        return False, str(e)
+
+
+def check_python():
+    """检查 Python 是否安装"""
+    py_names = ["python", "python3"] if platform.system() == "Windows" else ["python3", "python"]
+    for name in py_names:
+        try:
+            result = subprocess.run(
+                [name, "--version"], capture_output=True, text=True
+            )
+            if result.returncode == 0:
+                version = result.stdout.strip()
+                print(f"  ✅ Python 已安装: {version}")
+                return True
+        except FileNotFoundError:
+            continue
+
+    print("  ❌ Python 未安装，正在安装...")
+    return False
+
+
+def install_homebrew():
+    """安装 Homebrew"""
+    print("\n正在安装 Homebrew...")
+    print("（这可能需要几分钟，请耐心等待）\n")
+
+    cmd = '/bin/bash -c "$(curl -fsSL https://raw.githubusercontent.com/Homebrew/install/HEAD/install.sh)"'
+    ok, msg = run_cmd(cmd, check=False)
+
+    if ok:
+        print("  ✅ Homebrew 安装成功！")
+        return True
+    else:
+        print(f"  ⚠️ Homebrew 安装失败")
+        print("\n请手动安装 Homebrew：")
+        print("  打开终端，粘贴以下命令：")
+        print("  /bin/bash -c \"$(curl -fsSL https://raw.githubusercontent.com/Homebrew/install/HEAD/install.sh)\"")
+        return False
+
+
+def install_python_via_installer():
+    """通过 python.org 安装包安装 Python（不需要 sudo）"""
+    print("\n正在下载 Python 安装包...")
+    print("（这可能需要几分钟，取决于网速）\n")
+
+    dmg_url = "https://www.python.org/ftp/python/3.13.0/python-3.13.0-macos11.pkg"
+    dmg_path = "/tmp/python-installer.pkg"
+
+    ok, _ = run_cmd(f'curl -L -o "{dmg_path}" "{dmg_url}"', check=False)
+    if not ok:
+        print("  ❌ 下载失败，请检查网络连接")
+        return False
+
+    print("  ✅ 下载完成，正在打开安装程序...")
+    print("\n" + "=" * 50)
+    print("  请在弹出的安装程序中：")
+    print("  1. 点击「继续」")
+    print("  2. 阅读许可协议，点击「继续」→「同意」")
+    print("  3. 选择安装位置，点击「安装」")
+    print("  4. 输入电脑密码（如果有）")
+    print("  5. 等待安装完成，点击「关闭」")
+    print("=" * 50)
+    print("\n安装完成后，回到此窗口按回车键继续...")
+    print("（或者输入 q 退出，稍后手动安装）")
+
+    response = input("\n按回车继续，或输入 q 退出: ").strip().lower()
+    if response == 'q':
+        return False
+
+    run_cmd(f'open "{dmg_path}"', check=False)
+
+    input("\n安装完成后，按回车继续...")
+    return True
+
+
+def install_python():
+    """安装 Python"""
+    print("\n" + "=" * 50)
+    print("正在安装 Python...")
+    print("=" * 50)
+
+    success, _ = run_cmd("which brew")
+    if not success:
+        print("  ❌ 未检测到 Homebrew")
+
+        print("\n尝试通过安装包安装 Python（不需要 sudo）...")
+        if install_python_via_installer():
+            ok, _ = run_cmd("which python3", check=False)
+            if ok:
+                print("  ✅ Python 安装成功！")
+                return True
+
+        print("\n正在安装 Homebrew 作为备选...")
+        if not install_homebrew():
+            print("\n⚠️ 无法自动安装 Python")
+            print("请手动安装: https://www.python.org/downloads/")
+            return False
+
+        print("\n重新检测 Homebrew...")
+        success, _ = run_cmd("which brew")
+        if not success:
+            print("  ❌ Homebrew 安装失败")
+            return False
+
+    print("使用 Homebrew 安装 Python...")
+    ok, msg = run_cmd("brew install python", check=False)
+    if ok:
+        print("  ✅ Python 安装成功！")
+        return True
+    else:
+        print(f"  ⚠️ Homebrew 安装 Python 失败")
+        print("尝试通过安装包安装...")
+
+        if install_python_via_installer():
+            ok, _ = run_cmd("which python3", check=False)
+            if ok:
+                print("  ✅ Python 安装成功！")
+                return True
+
+        return False
+
+
+def install_dependencies():
+    """安装脚本依赖"""
+    print("\n正在安装依赖 (httpx, websockets)...")
+
+    pip_cmd = "pip" if platform.system() == "Windows" else "pip3"
+    deps = ["httpx", "websockets"]
+    for dep in deps:
+        ok, _ = run_cmd(f"{pip_cmd} install {dep}")
+        if ok:
+            print(f"  ✅ {dep} 安装成功")
+        else:
+            print(f"  ❌ {dep} 安装失败")
+            return False
+    return True
+
+
+def check_workbuddy_installed():
+    """检查 WorkBuddy 是否安装"""
+    paths = [
+        Path("/Applications/WorkBuddy.app"),
+        Path.home() / "Applications/WorkBuddy.app",
+        # Windows paths
+        Path(os.path.expandvars(r"%LOCALAPPDATA%\Programs\WorkBuddy\WorkBuddy.exe")),
+        Path(os.path.expandvars(r"%ProgramFiles%\WorkBuddy\WorkBuddy.exe")),
+        Path(os.path.expandvars(r"%ProgramFiles(x86)%\WorkBuddy\WorkBuddy.exe")),
+        Path(r"D:\Program Files\WorkBuddy\WorkBuddy.exe"),
+        Path(r"E:\Program Files\WorkBuddy\WorkBuddy.exe"),
+    ]
+    for p in paths:
+        if p.exists():
+            return True, p
+    return False, None
+
+
+def check_workbuddy_running():
+    """检查 WorkBuddy 是否正在运行（调试模式）"""
+    try:
+        import httpx
+        resp = httpx.get("http://127.0.0.1:9222/json", timeout=2)
+        if resp.status_code == 200:
+            return True
+    except:
+        pass
+    return False
+
+
+def start_workbuddy_debug():
+    """启动 WorkBuddy 调试模式"""
+    print("\n正在启动 WorkBuddy（调试模式）...")
+
+    if platform.system() == "Windows":
+        # Find WorkBuddy on Windows
+        candidates = [
+            os.path.expandvars(r"%LOCALAPPDATA%\Programs\WorkBuddy\WorkBuddy.exe"),
+            os.path.expandvars(r"%ProgramFiles%\WorkBuddy\WorkBuddy.exe"),
+            os.path.expandvars(r"%ProgramFiles(x86)%\WorkBuddy\WorkBuddy.exe"),
+            r"D:\Program Files\WorkBuddy\WorkBuddy.exe",
+            r"E:\Program Files\WorkBuddy\WorkBuddy.exe",
+        ]
+        workbuddy_path = next((p for p in candidates if os.path.exists(p)), None)
+        if not workbuddy_path:
+            print("  ❌ 找不到 WorkBuddy，请手动启动"  )
+            print("  → 关闭 WorkBuddy，然后重新打开"  )
+            print("  → 或者：Win+R 运行：workbuddy --remote-debugging-port=9222"  )
+            return False
+        cmd = f'"{workbuddy_path}" --remote-debugging-port=9222'
+    else:
+        workbuddy_path = "/Applications/WorkBuddy.app/Contents/MacOS/Electron"
+        cmd = f'"{workbuddy_path}" --remote-debugging-port=9222'
+
+    try:
+        subprocess.Popen(cmd, shell=True)
+        print("  ✅ WorkBuddy 已启动")
+        print("\n等待 WorkBuddy 完全加载（5秒）...")
+        time.sleep(5)
+        return True
+    except Exception as e:
+        print(f"  ❌ 启动失败: {e}")
+        return False
+
+
+def check_requirements():
+    """检查前置条件"""
+    print("\n[检查 1/4] 检测 Python...")
+    if not check_python():
+        if not install_python():
+            return False
+
+    print("\n[检查 2/4] 安装依赖...")
+    if not install_dependencies():
+        return False
+
+    print("\n[检查 3/4] 检测 WorkBuddy...")
+    installed, path = check_workbuddy_installed()
+    if not installed:
+        print("  ❌ WorkBuddy 未安装")
+        print("  请先安装 WorkBuddy: https://workbuddy.com/download")
+        return False
+    print(f"  ✅ WorkBuddy 已安装: {path}")
+
+    print("\n[检查 4/4] 检测 WorkBuddy 运行状态...")
+    if check_workbuddy_running():
+        print("  ✅ WorkBuddy 已运行（调试模式）")
+    else:
+        print("  ⚠️ WorkBuddy 未运行，正在启动...")
+        if not start_workbuddy_debug():
+            return False
+
+    return True
+
+
+async def extract_token():
+    """从 WorkBuddy 提取 Token"""
+    import httpx
+    import websockets
+
+    print("\n正在连接 WorkBuddy...")
+
+    async with httpx.AsyncClient() as client:
+        resp = await client.get("http://127.0.0.1:9222/json", timeout=5)
+        targets = resp.json()
+
+    print(f"  ✅ 找到 {len(targets)} 个页面")
+
+    ws_url = None
+    for t in targets:
+        if t.get("type") == "page" and "workbench" in t.get("url", ""):
+            ws_url = t.get("webSocketDebuggerUrl")
+            break
+
+    if not ws_url:
+        for t in targets:
+            if t.get("type") == "page":
+                ws_url = t.get("webSocketDebuggerUrl")
+                break
+
+    if not ws_url:
+        raise Exception("未找到 WorkBuddy 页面")
+
+    print("  ✅ 已找到 WorkBuddy 页面")
+    print("\n���在提取 Token...")
+
+    async with websockets.connect(ws_url, ping_timeout=30) as ws:
+        cmd = {
+            "id": 1,
+            "method": "Runtime.evaluate",
+            "params": {
+                "expression": """
+                    (async () => {
+                        try {
+                            const s = await window.vscode.ipcRenderer.invoke(
+                                'vscode:genie:auth:getSession'
+                            );
+                            return JSON.stringify(s);
+                        } catch(e) {
+                            return JSON.stringify({error: e.message});
+                        }
+                    })()
+                """,
+                "awaitPromise": True,
+                "returnByValue": True,
+            },
+        }
+        await ws.send(json.dumps(cmd))
+        result = json.loads(await asyncio.wait_for(ws.recv(), timeout=30))
+
+    value = result.get("result", {}).get("result", {}).get("value", "")
+    if not value:
+        raise Exception("CDP 返回为空")
+
+    session = json.loads(value)
+    if session.get("error"):
+        raise Exception(f"CDP 错误: {session['error']}")
+
+    return session
+
+
+def save_token(access_token, refresh_token):
+    """保存 Token 到桌面"""
+    desktop = Path.home() / "Desktop"
+    output_file = desktop / "token.txt"
+
+    content = f"""WorkBuddy Token
+生成时间: {time.strftime("%Y-%m-%d %H:%M:%S")}
+
+access_token={access_token}
+refresh_token={refresh_token}
+"""
+
+    output_file.write_text(content, encoding="utf-8")
+    return output_file
+
+
+def main():
+    print("=" * 50)
+    print("  WorkBuddy Token 提取工具")
+    print("=" * 50)
+    print("\n正在检测环境...\n")
+
+    if not check_requirements():
+        input("\n按回车键退出...")
+        sys.exit(1)
+
+    try:
+        session = asyncio.run(extract_token())
+    except Exception as e:
+        print(f"\n❌ 提取失败: {e}")
+        print("\n请确保：")
+        print("  1. WorkBuddy 已登录")
+        print("  2. 重新运行此脚本")
+        input("\n按回车键退出...")
+        sys.exit(1)
+
+    auth = session.get("auth", session)
+    access_token = auth.get("accessToken", "")
+    refresh_token = auth.get("refreshToken", "")
+
+    if not access_token:
+        print("\n❌ 未获取到 accessToken")
+        print("请确保 WorkBuddy 已登录 AI 功能")
+        input("\n按回车键退出...")
+        sys.exit(1)
+
+    try:
+        output_file = save_token(access_token, refresh_token)
+    except Exception as e:
+        print(f"\n❌ 保存失败: {e}")
+        input("\n按回车键退出...")
+        sys.exit(1)
+
+    print("\n" + "=" * 50)
+    print("✅ 成功！")
+    print("=" * 50)
+    print(f"\n📄 Token 已保存到桌面:")
+    print(f"   {output_file}")
+    print(f"\n🔑 access_token:  {access_token[:30]}...")
+    print(f"🔑 refresh_token: {refresh_token[:30] if refresh_token else 'N/A'}...")
+    print("\n" + "=" * 50)
+
+    input("\n按回车键退出...")
+
+
+if __name__ == "__main__":
+    main()

requirements.txt

@@ -0,0 +1,6 @@
+fastapi>=0.115.0
+uvicorn>=0.34.0
+httpx>=0.28.0
+pyjwt>=2.10.0
+websockets>=14.0
+python-dotenv>=1.0.0

server.py

@@ -0,0 +1,822 @@
+"""
+WorkBuddy → OpenAI-compatible reverse proxy.
+
+Accepts standard OpenAI API requests and forwards them to WorkBuddy's
+/v2/chat/completions endpoint with the required authentication headers.
+
+All user-specific values (user_id, enterprise_id, domain) are automatically
+extracted from the JWT token — no manual configuration required.
+
+Usage:
+    python server.py                                        # auto-extract via CDP
+    WB_TOKEN=<jwt> WB_REFRESH_TOKEN=<jwt> python server.py  # manual token
+"""
+
+import asyncio
+import hmac
+import json
+import logging
+import os
+import re
+import sys
+import time
+import uuid
+from contextlib import asynccontextmanager
+from pathlib import Path
+from typing import AsyncGenerator
+
+if sys.platform == "win32":
+    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
+
+import httpx
+import jwt
+import uvicorn
+from fastapi import FastAPI, HTTPException, Request
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse, StreamingResponse
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(message)s",
+    datefmt="%H:%M:%S",
+)
+log = logging.getLogger("wb-proxy")
+
+BASE_DIR = Path(__file__).parent
+
+# README / .env.example：本地与 Docker 均通过 .env 配置；已有环境变量优先（不 override）
+try:
+    from dotenv import load_dotenv
+
+    load_dotenv(BASE_DIR / ".env", override=False)
+except ImportError:
+    pass
+
+TOKEN_FILE = Path(os.getenv("TOKEN_FILE_PATH", str(BASE_DIR / "data" / "token.json")))
+
+PROXY_PORT = int(os.getenv("PROXY_PORT", "19090"))
+PROXY_API_KEY = os.getenv("PROXY_API_KEY", "").strip()
+WB_API_BASE = os.getenv("WB_API_BASE", "https://copilot.tencent.com")
+CDP_URL = os.getenv("CDP_URL", "http://127.0.0.1:9222")
+
+
+def _detect_wb_version() -> str:
+    """Auto-detect genieVersion from local WorkBuddy installation."""
+    candidates = [
+        # macOS
+        Path("/Applications/WorkBuddy.app/Contents/Resources/app/product.json"),
+        # Windows — common locations
+        Path(os.path.expandvars(r"%LOCALAPPDATA%\Programs\WorkBuddy\resources\app\product.json")),
+        Path(os.path.expandvars(r"%ProgramFiles%\WorkBuddy\resources\app\product.json")),
+        Path(os.path.expandvars(r"%ProgramFiles(x86)%\WorkBuddy\resources\app\product.json")),
+        Path(os.path.expandvars(r"%APPDATA%\WorkBuddy\resources\app\product.json")),
+        # Linux (snap / deb)
+        Path(os.path.expanduser("~/.local/share/WorkBuddy/resources/app/product.json")),
+        Path("/opt/WorkBuddy/resources/app/product.json"),
+    ]
+    for p in candidates:
+        try:
+            data = json.loads(p.read_text(encoding="utf-8"))
+            v = data.get("genieVersion", "")
+            if v:
+                log.info(f"Detected WorkBuddy {v} at {p.parent}")
+                return v
+        except Exception:
+            continue
+    return ""
+
+
+WB_VERSION = os.getenv("WB_VERSION", "") or _detect_wb_version() or "4.8.1"
+
+HEADERS_TEMPLATE = {
+    "X-IDE-Type": "CodeBuddyIDE",
+    "X-IDE-Name": "CodeBuddyIDE",
+    "X-IDE-Version": WB_VERSION,
+    "X-Product-Version": WB_VERSION,
+    "X-Product": "SaaS",
+    "X-Env-ID": "production",
+    "X-Requested-With": "XMLHttpRequest",
+    "User-Agent": f"CodeBuddyIDE/{WB_VERSION} coding-copilot/{WB_VERSION}",
+}
+
+REASONING_MODELS = {"deepseek-r1", "deepseek-r1-0528-lkeap", "hunyuan-2.0-thinking-ioa", "hunyuan-2.0-thinking", "hy3-preview", "kimi-k2.6", "kimi-k2.6-ioa", "kimi-k2.7", "deepseek-v4-pro", "glm-5.2"}
+DEFAULT_TIMEOUT = int(os.getenv("WB_TIMEOUT", "120"))
+
+# WorkBuddy prepends "Thought: Xms\n" to each reasoning chunk — strip it
+_THOUGHT_RE = re.compile(r"^Thought:\s*\d+ms\s*\n", re.MULTILINE)
+
+
+def _clean_reasoning(text: str) -> str:
+    return _THOUGHT_RE.sub("", text)
+REASONING_TIMEOUT = int(os.getenv("WB_REASONING_TIMEOUT", "300"))
+
+
+def _parse_jwt_claims(token: str) -> dict:
+    """Extract user_id, enterprise_id and domain from JWT without verification."""
+    try:
+        payload = jwt.decode(token, options={"verify_signature": False})
+        user_id = payload.get("sub", "")
+
+        iss = payload.get("iss", "")
+        # iss format: https://<domain>/auth/realms/sso-<enterprise_id>
+        enterprise_id = ""
+        m = re.search(r"/sso-([^/]+)$", iss)
+        if m:
+            enterprise_id = m.group(1)
+
+        domain = ""
+        m2 = re.match(r"https?://([^/]+)", iss)
+        if m2:
+            domain = m2.group(1)
+
+        return {"user_id": user_id, "enterprise_id": enterprise_id, "domain": domain}
+    except Exception:
+        return {"user_id": "", "enterprise_id": "", "domain": ""}
+
+
+# ---------------------------------------------------------------------------
+# Token management
+# ---------------------------------------------------------------------------
+class TokenManager:
+    def __init__(self):
+        self.access_token: str = ""
+        self.refresh_token: str = ""
+        self.user_id: str = ""
+        self.enterprise_id: str = ""
+        self.domain: str = ""
+        self.department_info: str = ""
+        self._lock = asyncio.Lock()
+
+    async def init(self):
+        self.access_token = os.getenv("WB_TOKEN", "")
+        self.refresh_token = os.getenv("WB_REFRESH_TOKEN", "")
+
+        if not self.access_token:
+            self._load_from_file()
+
+        if not self.access_token:
+            await self._extract_from_cdp()
+
+        if self.access_token:
+            self._apply_claims()
+            self._log_token_info()
+            self._save_to_file()
+
+    def _apply_claims(self):
+        claims = _parse_jwt_claims(self.access_token)
+        self.user_id = os.getenv("WB_USER_ID", "") or claims["user_id"]
+        self.enterprise_id = os.getenv("WB_ENTERPRISE_ID", "") or claims["enterprise_id"]
+        self.domain = os.getenv("WB_DOMAIN", "") or claims["domain"]
+        log.info(f"User: {self.user_id[:8]}..., Enterprise: {self.enterprise_id}, Domain: {self.domain}")
+
+    def _load_from_file(self):
+        if TOKEN_FILE.exists():
+            try:
+                data = json.loads(TOKEN_FILE.read_text(encoding="utf-8"))
+                self.access_token = data.get("access_token", "")
+                self.refresh_token = data.get("refresh_token", "")
+                if self.access_token:
+                    log.info("Token loaded from file")
+            except Exception:
+                pass
+
+    def _save_to_file(self):
+        TOKEN_FILE.parent.mkdir(parents=True, exist_ok=True)
+        TOKEN_FILE.write_text(json.dumps({
+            "access_token": self.access_token,
+            "refresh_token": self.refresh_token,
+            "saved_at": time.strftime("%Y-%m-%d %H:%M:%S"),
+        }, indent=2), encoding="utf-8")
+
+    async def get_token(self) -> str:
+        if self._is_expired():
+            await self.refresh()
+        return self.access_token
+
+    def _is_expired(self) -> bool:
+        if not self.access_token:
+            return True
+        try:
+            payload = jwt.decode(self.access_token, options={"verify_signature": False})
+            return time.time() > (payload.get("exp", 0) - 300)
+        except Exception as e:
+            log.warning(f"Failed to decode token, treating as expired: {e}")
+            return True
+
+    def _log_token_info(self):
+        try:
+            payload = jwt.decode(self.access_token, options={"verify_signature": False})
+            hours = (payload.get("exp", 0) - time.time()) / 3600
+            log.info(f"Token valid, expires in {hours:.1f}h")
+        except Exception:
+            log.warning("Could not decode token")
+
+    async def refresh(self):
+        async with self._lock:
+            if not self._is_expired():
+                return
+            if self.refresh_token:
+                await self._refresh_via_api()
+            else:
+                await self._extract_from_cdp()
+
+    async def _refresh_via_api(self):
+        log.info("Refreshing token via API...")
+        headers = {
+            **HEADERS_TEMPLATE,
+            "Content-Type": "application/json",
+            "Authorization": f"Bearer {self.access_token}",
+            "X-Refresh-Token": self.refresh_token,
+            "X-Auth-Refresh-Source": "plugin",
+            "X-Domain": self.domain,
+            "X-User-Id": self.user_id,
+            "X-Enterprise-Id": self.enterprise_id,
+            "X-Tenant-Id": self.enterprise_id,
+            "X-Request-ID": uuid.uuid4().hex,
+            "X-Request-Trace-Id": str(uuid.uuid4()),
+        }
+        if self.department_info:
+            headers["X-Department-Info"] = self.department_info
+        async with httpx.AsyncClient() as client:
+            resp = await client.post(
+                f"{WB_API_BASE}/v2/plugin/auth/token/refresh",
+                headers=headers,
+                json={},
+                timeout=15,
+            )
+            data = resp.json()
+            if data.get("code") == 0 and data.get("data", {}).get("accessToken"):
+                self.access_token = data["data"]["accessToken"]
+                if data["data"].get("refreshToken"):
+                    self.refresh_token = data["data"]["refreshToken"]
+                self._apply_claims()
+                log.info("Token refreshed successfully via API")
+                self._log_token_info()
+                self._save_to_file()
+            else:
+                log.error(f"Token refresh failed: {data}")
+                await self._extract_from_cdp()
+
+    async def _extract_from_cdp(self):
+        """Extract token via CDP (deprecated: WorkBuddy 5.1.4+ no longer exposes ipcRenderer)."""
+        log.info(f"Extracting token from WorkBuddy via CDP ({CDP_URL})...")
+        try:
+            async with httpx.AsyncClient() as client:
+                resp = await client.get(f"{CDP_URL}/json", timeout=5)
+                targets = resp.json()
+
+            ws_url = None
+            for t in targets:
+                if t.get("type") == "page" and "workbench" in t.get("url", ""):
+                    ws_url = t.get("webSocketDebuggerUrl")
+                    break
+            if not ws_url:
+                for t in targets:
+                    if t.get("type") == "page":
+                        ws_url = t.get("webSocketDebuggerUrl")
+                        break
+            if not ws_url:
+                log.error("No CDP target found")
+                return
+
+            import websockets
+
+            async with websockets.connect(ws_url) as ws:
+                cmd = {
+                    "id": 1,
+                    "method": "Runtime.evaluate",
+                    "params": {
+                        "expression": """
+                            (async () => {
+                                try {
+                                    const s = await window.vscode.ipcRenderer.invoke(
+                                        'vscode:genie:auth:getSession'
+                                    );
+                                    return JSON.stringify(s);
+                                } catch(e) {
+                                    return JSON.stringify({error: e.message});
+                                }
+                            })()
+                        """,
+                        "awaitPromise": True,
+                        "returnByValue": True,
+                    },
+                }
+                await ws.send(json.dumps(cmd))
+                result = json.loads(await asyncio.wait_for(ws.recv(), timeout=10))
+
+                value = result.get("result", {}).get("result", {}).get("value", "")
+                if value:
+                    session = json.loads(value)
+                    auth = session.get("auth", session)
+                    if auth.get("accessToken"):
+                        self.access_token = auth["accessToken"]
+                        self.refresh_token = auth.get("refreshToken", "")
+                        account = session.get("account", {})
+                        if isinstance(account, dict):
+                            self.department_info = account.get("departmentFullName", "")
+                        self._apply_claims()
+                        log.info("Token extracted from CDP successfully")
+                        self._log_token_info()
+                        self._save_to_file()
+                    elif session.get("error"):
+                        log.error(f"CDP extraction error: {session['error']}")
+
+        except ImportError:
+            log.warning("websockets not installed — run: pip install websockets")
+        except Exception as e:
+            log.error(f"CDP extraction failed: {e}")
+
+
+token_mgr = TokenManager()
+
+
+# ---------------------------------------------------------------------------
+# Cursor model name mapping
+# Cursor validates model names server-side; only its built-in names pass.
+# This map translates Cursor names → WorkBuddy model IDs.
+# ---------------------------------------------------------------------------
+CURSOR_TO_WB_MAP: dict[str, str] = {
+    # Claude
+    "claude-4.6-opus-high": "claude-opus-4.6",
+    "claude-4.6-opus-max": "claude-opus-4.6-1m",
+    "claude-4.6-opus-high-thinking": "claude-opus-4.6",
+    "claude-4.6-opus-high-thinking-fast": "claude-opus-4.6",
+    "claude-4.6-opus-max-thinking": "claude-opus-4.6-1m",
+    "claude-4.6-opus-max-thinking-fast": "claude-opus-4.6-1m",
+    "claude-4.6-sonnet-medium": "claude-sonnet-4.6",
+    "claude-4.6-sonnet-medium-thinking": "claude-sonnet-4.6-1m",
+    "claude-4.5-opus-high": "claude-opus-4.5",
+    "claude-4.5-opus-high-thinking": "claude-opus-4.5",
+    "claude-4.5-sonnet": "claude-4.5",
+    "claude-4.5-sonnet-thinking": "claude-4.5",
+    "claude-4.5-haiku": "claude-haiku-4.5",
+    "claude-4.5-haiku-thinking": "claude-haiku-4.5",
+    "claude-opus-4.6": "claude-opus-4.6",
+    # Gemini
+    "gemini-3.1-pro": "gemini-3.0-pro",
+    "gemini-3-flash": "gemini-3.1-flash-lite",
+    # Kimi
+    "kimi-k2.5": "kimi-k2.5-ioa",
+}
+
+# Reverse map: WB model ID → preferred Cursor alias (for /v1/models)
+WB_TO_CURSOR_MAP: dict[str, str] = {
+    "claude-opus-4.6": "claude-4.6-opus-high",
+    "claude-opus-4.6-1m": "claude-4.6-opus-max",
+    "claude-sonnet-4.6": "claude-4.6-sonnet-medium",
+    "claude-sonnet-4.6-1m": "claude-4.6-sonnet-medium-thinking",
+    "claude-opus-4.5": "claude-4.5-opus-high",
+    "claude-4.5": "claude-4.5-sonnet",
+    "claude-haiku-4.5": "claude-4.5-haiku",
+    "gemini-3.0-pro": "gemini-3.1-pro",
+    "gemini-3.1-flash-lite": "gemini-3-flash",
+    "kimi-k2.5-ioa": "kimi-k2.5",
+}
+
+
+def resolve_model(model: str) -> str:
+    """Resolve Cursor model name to WorkBuddy model ID. Pass through if no mapping."""
+    return CURSOR_TO_WB_MAP.get(model, model)
+
+
+# ---------------------------------------------------------------------------
+# Available models
+# ---------------------------------------------------------------------------
+MODELS = [
+    # DeepSeek
+    {"id": "deepseek-r1", "name": "DeepSeek-R1"},
+    {"id": "deepseek-r1-0528-lkeap", "name": "DeepSeek-R1-LKEAP"},
+    {"id": "deepseek-v3", "name": "DeepSeek-V3"},
+    {"id": "deepseek-v3.2", "name": "DeepSeek-V3.2"},
+    {"id": "deepseek-v3-1", "name": "DeepSeek-V3.1"},
+    {"id": "deepseek-v3-0324", "name": "DeepSeek-V3-0324"},
+    {"id": "deepseek-v3-0324-lkeap", "name": "DeepSeek-V3-0324-LKEAP"},
+    {"id": "deepseek-v3-1-volc", "name": "DeepSeek-V3.1-Volc"},
+    {"id": "deepseek-v4-pro", "name": "DeepSeek-V4-Pro"},
+    {"id": "deepseek-v4-flash", "name": "DeepSeek-V4-Flash"},
+    # Claude (haiku only — opus/sonnet require enterprise auth)
+    {"id": "claude-haiku-4.5", "name": "Claude-Haiku-4.5"},
+    {"id": "claude-4.5-haiku", "name": "Claude-4.5-Haiku"},
+    {"id": "claude-4.5-haiku-thinking", "name": "Claude-4.5-Haiku-Thinking"},
+    # GLM
+    {"id": "glm-4.6", "name": "GLM-4.6"},
+    {"id": "glm-4.7", "name": "GLM-4.7"},
+    {"id": "glm-5.0", "name": "GLM-5.0"},
+    {"id": "glm-5.1", "name": "GLM-5.1"},
+    {"id": "glm-5.2", "name": "GLM-5.2"},
+    {"id": "glm-5v-turbo", "name": "GLM-5v-Turbo"},
+    {"id": "glm-5v-turbo-ioa", "name": "GLM-5v-Turbo-IOA"},
+    # Hunyuan
+    {"id": "hunyuan-2.0-instruct", "name": "Hunyuan-2.0"},
+    {"id": "hunyuan-2.0-thinking", "name": "Hunyuan-2.0-Thinking"},
+    {"id": "hunyuan-2.0-thinking-ioa", "name": "Hunyuan-2.0-Thinking-IOA"},
+    {"id": "hy3-preview", "name": "Hunyuan-3-Preview"},
+    # Kimi
+    {"id": "kimi-k2.6", "name": "Kimi-K2.6"},
+    {"id": "kimi-k2.6-ioa", "name": "Kimi-K2.6-IOA"},
+    {"id": "kimi-k2.7", "name": "Kimi-K2.7"},
+    # MiniMax
+    {"id": "minimax-m3", "name": "MiniMax-M3"},
+    {"id": "minimax-m3-ioa", "name": "MiniMax-M3-IOA"},
+    # Default
+    {"id": "codewise-default-model-v2", "name": "Default (Codewise)"},
+]
+
+
+# ---------------------------------------------------------------------------
+# FastAPI app
+# ---------------------------------------------------------------------------
+http_pool: httpx.AsyncClient | None = None
+
+
+@asynccontextmanager
+async def lifespan(_app: FastAPI):
+    global http_pool
+    http_pool = httpx.AsyncClient(
+        timeout=httpx.Timeout(DEFAULT_TIMEOUT, connect=10),
+        limits=httpx.Limits(max_connections=50, max_keepalive_connections=10),
+    )
+    await token_mgr.init()
+    yield
+    await http_pool.aclose()
+    http_pool = None
+
+
+app = FastAPI(title="WorkBuddy Proxy", lifespan=lifespan)
+
+# CORS: allow all origins since this is a public-facing API proxy.
+# Auth is enforced via API key, not origin.
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=False,  # 与 allow_origins=["*"] 不能同时为 True
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+def _verify_api_key(request: Request):
+    auth = request.headers.get("Authorization") or ""
+    key = auth.replace("Bearer ", "").strip()
+    if not key:
+        key = (request.headers.get("X-API-Key") or "").strip()
+    if not hmac.compare_digest(key, PROXY_API_KEY):
+        raise HTTPException(status_code=401, detail="Invalid API key")
+
+
+def _build_headers(access_token: str) -> dict:
+    headers = {
+        **HEADERS_TEMPLATE,
+        "Content-Type": "application/json",
+        "Accept": "text/event-stream",
+        "Authorization": f"Bearer {access_token}",
+        "X-User-Id": token_mgr.user_id,
+        "X-Enterprise-Id": token_mgr.enterprise_id,
+        "X-Tenant-Id": token_mgr.enterprise_id,
+        "X-Domain": token_mgr.domain,
+        "X-Request-ID": uuid.uuid4().hex,
+        "X-Request-Trace-Id": str(uuid.uuid4()),
+    }
+    if token_mgr.department_info:
+        headers["X-Department-Info"] = token_mgr.department_info
+    return headers
+
+
+@app.get("/v1/models")
+async def list_models(request: Request):
+    _verify_api_key(request)
+
+    # Build model list: original WB models + Cursor-compatible aliases
+    seen_ids: set[str] = set()
+    data = []
+
+    # First: add Cursor-compatible aliases for mapped models
+    for cursor_name, wb_id in CURSOR_TO_WB_MAP.items():
+        if cursor_name not in seen_ids:
+            seen_ids.add(cursor_name)
+            # Find display name from MODELS list
+            wb_model = next((m for m in MODELS if m["id"] == wb_id), None)
+            display_name = wb_model["name"] if wb_model else cursor_name
+            data.append({
+                "id": cursor_name,
+                "object": "model",
+                "created": 1700000000,
+                "owned_by": "workbuddy",
+                "name": f"{display_name} (Cursor)",
+            })
+
+    # Then: add original WB models (for non-Cursor clients like OpenClaw)
+    for m in MODELS:
+        if m["id"] not in seen_ids:
+            seen_ids.add(m["id"])
+            data.append({
+                "id": m["id"],
+                "object": "model",
+                "created": 1700000000,
+                "owned_by": "workbuddy",
+                "name": m["name"],
+            })
+
+    return {"object": "list", "data": data}
+
+
+def _timeout_for(model: str) -> float:
+    return REASONING_TIMEOUT if model in REASONING_MODELS else DEFAULT_TIMEOUT
+
+
+async def _upstream_stream(url: str, headers: dict, body: dict, timeout: float):
+    """Open a streaming connection to upstream; returns resp or None on failure."""
+    try:
+        req = http_pool.build_request("POST", url, headers=headers, json=body, timeout=timeout)
+        resp = await http_pool.send(req, stream=True)
+        return resp
+    except (httpx.TimeoutException, httpx.ConnectError, httpx.NetworkError, httpx.RemoteProtocolError):
+        return None
+
+
+@app.post("/v1/chat/completions")
+async def chat_completions(request: Request):
+    _verify_api_key(request)
+
+    body = await request.json()
+    raw_model = body.get("model", "deepseek-v3")
+    model = resolve_model(raw_model)  # Cursor name → WB model ID
+    stream = body.get("stream", False)
+
+    if raw_model != model:
+        log.info(f"[Model] Mapped: {raw_model} → {model}")
+
+    wb_body = {k: v for k, v in body.items() if k != "stream"}
+    wb_body["stream"] = True
+    wb_body["model"] = model
+
+    # Auto-inject reasoning for thinking models
+    if model in REASONING_MODELS:
+        wb_body.setdefault("reasoning", {"effort": "medium"})
+
+    access_token = await token_mgr.get_token()
+    if not access_token:
+        raise HTTPException(status_code=503, detail="No valid WorkBuddy token")
+
+    url = f"{WB_API_BASE}/v2/chat/completions"
+    timeout = _timeout_for(model)
+    t_start = time.monotonic()
+
+    if stream:
+        return StreamingResponse(
+            _stream_response(url, wb_body, model, timeout),
+            media_type="text/event-stream",
+            headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
+        )
+    return await _non_stream_response(url, wb_body, model, timeout, t_start)
+
+
+# ---------------------------------------------------------------------------
+# Streaming path
+# ---------------------------------------------------------------------------
+async def _stream_response(
+    url: str, body: dict, model: str, timeout: float
+) -> AsyncGenerator[str, None]:
+    max_attempts = 2
+    for attempt in range(1, max_attempts + 1):
+        access_token = await token_mgr.get_token()
+        headers = _build_headers(access_token)
+        t_start = time.monotonic()
+        has_content = False
+
+        resp = await _upstream_stream(url, headers, body, timeout)
+        if resp is None:
+            log.error(f"[{model}] Upstream timeout (attempt {attempt})")
+            if attempt < max_attempts:
+                continue
+            yield 'data: {"error":"upstream timeout"}\n\n'
+            yield "data: [DONE]\n\n"
+            return
+
+        try:
+            if resp.status_code == 401:
+                log.warning(f"[{model}] Got 401, refreshing token...")
+                await token_mgr.refresh()
+                if attempt < max_attempts:
+                    continue
+                yield 'data: {"error":"authentication failed"}\n\n'
+                yield "data: [DONE]\n\n"
+                return
+
+            if resp.status_code != 200:
+                error_body = await resp.aread()
+                error_text = error_body.decode("utf-8", errors="replace")
+                log.error(f"[{model}] Upstream {resp.status_code}: {error_text[:200]}")
+                yield f"data: {json.dumps({'error': error_text})}\n\n"
+                yield "data: [DONE]\n\n"
+                return
+
+            done_sent = False
+            reasoning_buf = ""
+            async for line in resp.aiter_lines():
+                if not line.startswith("data: "):
+                    if line.strip():
+                        has_content = True
+                    yield line + "\n\n"
+                    continue
+                if line == "data: [DONE]":
+                    done_sent = True
+                    if reasoning_buf:
+                        buf_chunk = {"choices": [{"index": 0, "delta": {"reasoning_content": reasoning_buf}}]}
+                        yield f"data: {json.dumps(buf_chunk)}\n\n"
+                    yield line + "\n\n"
+                    continue
+
+                has_content = True
+                try:
+                    chunk = json.loads(line.removeprefix("data: "))
+                    choice = chunk.get("choices", [{}])[0]
+                    delta = choice.get("delta", {})
+                    rc = delta.get("reasoning_content")
+                    has_content_field = bool(delta.get("content") or delta.get("tool_calls"))
+
+                    if rc is not None:
+                        cleaned = _clean_reasoning(rc)
+                        if has_content_field and reasoning_buf:
+                            buf_chunk = {"choices": [{"index": 0, "delta": {"reasoning_content": reasoning_buf}}]}
+                            yield f"data: {json.dumps(buf_chunk)}\n\n"
+                            reasoning_buf = ""
+                        reasoning_buf += cleaned
+                        if not has_content_field:
+                            continue
+                        delta.pop("reasoning_content", None)
+
+                    if reasoning_buf:
+                        buf_chunk = {"choices": [{"index": 0, "delta": {"reasoning_content": reasoning_buf}}]}
+                        yield f"data: {json.dumps(buf_chunk)}\n\n"
+                        reasoning_buf = ""
+
+                    yield f"data: {json.dumps(chunk)}\n\n"
+                except (json.JSONDecodeError, KeyError):
+                    yield line + "\n\n"
+
+            elapsed = time.monotonic() - t_start
+
+            if not has_content and not reasoning_buf and attempt < max_attempts:
+                log.warning(f"[{model}] Empty response, retrying... ({elapsed:.1f}s)")
+                await asyncio.sleep(1)
+                continue
+
+            if not done_sent:
+                yield "data: [DONE]\n\n"
+
+            log.info(f"[{model}] stream {elapsed:.1f}s")
+            return
+
+        except httpx.ReadTimeout:
+            log.error(f"[{model}] Read timeout during stream (attempt {attempt})")
+            if attempt < max_attempts:
+                continue
+            yield 'data: {"error":"upstream timeout"}\n\n'
+            yield "data: [DONE]\n\n"
+            return
+        finally:
+            await resp.aclose()
+
+
+# ---------------------------------------------------------------------------
+# Non-streaming path
+# ---------------------------------------------------------------------------
+async def _non_stream_response(
+    url: str, body: dict, model: str, timeout: float, t_start: float
+) -> JSONResponse:
+    max_attempts = 2
+
+    for attempt in range(1, max_attempts + 1):
+        access_token = await token_mgr.get_token()
+        headers = _build_headers(access_token)
+
+        collected_content = ""
+        collected_reasoning = ""
+        tool_calls_map: dict[int, dict] = {}
+        finish_reason = "stop"
+        resp_model = model
+        usage = {}
+
+        resp = await _upstream_stream(url, headers, body, timeout)
+        if resp is None:
+            log.error(f"[{model}] Upstream timeout (attempt {attempt})")
+            if attempt < max_attempts:
+                continue
+            raise HTTPException(status_code=504, detail="Upstream timeout")
+
+        try:
+            if resp.status_code == 401:
+                log.warning(f"[{model}] Got 401, refreshing token...")
+                await token_mgr.refresh()
+                if attempt < max_attempts:
+                    continue
+                raise HTTPException(status_code=401, detail="Authentication failed")
+
+            if resp.status_code != 200:
+                error_body = await resp.aread()
+                raise HTTPException(status_code=resp.status_code,
+                                    detail=error_body.decode("utf-8", errors="replace"))
+
+            async for line in resp.aiter_lines():
+                text = line.removeprefix("data: ").strip()
+                if not text or text == "[DONE]":
+                    continue
+                try:
+                    chunk = json.loads(text)
+                    choice = chunk.get("choices", [{}])[0]
+                    delta = choice.get("delta", {})
+
+                    collected_content += delta.get("content") or ""
+                    rc = delta.get("reasoning_content")
+                    if rc:
+                        collected_reasoning += _clean_reasoning(rc)
+
+                    for tc in delta.get("tool_calls") or []:
+                        idx = tc.get("index", 0)
+                        if idx not in tool_calls_map:
+                            tool_calls_map[idx] = {
+                                "id": tc.get("id", ""),
+                                "type": "function",
+                                "function": {"name": "", "arguments": ""},
+                            }
+                        entry = tool_calls_map[idx]
+                        if tc.get("id"):
+                            entry["id"] = tc["id"]
+                        fn = tc.get("function", {})
+                        if fn.get("name"):
+                            entry["function"]["name"] += fn["name"]
+                        if fn.get("arguments"):
+                            entry["function"]["arguments"] += fn["arguments"]
+
+                    fr = choice.get("finish_reason")
+                    if fr:
+                        finish_reason = fr
+
+                    if chunk.get("usage"):
+                        usage = chunk["usage"]
+                    resp_model = chunk.get("model", resp_model)
+                except (json.JSONDecodeError, IndexError, KeyError):
+                    pass
+
+        except httpx.ReadTimeout:
+            log.error(f"[{model}] Read timeout during non-stream (attempt {attempt})")
+            if attempt < max_attempts:
+                continue
+            raise HTTPException(status_code=504, detail="Upstream read timeout")
+        finally:
+            await resp.aclose()
+
+        if not collected_content and not collected_reasoning and not tool_calls_map and attempt < max_attempts:
+            log.warning(f"[{model}] Empty response, retrying...")
+            await asyncio.sleep(1)
+            continue
+
+        elapsed = time.monotonic() - t_start
+        prompt_t = usage.get("prompt_tokens", "?")
+        compl_t = usage.get("completion_tokens", "?")
+        log.info(f"[{model}] non-stream {elapsed:.1f}s  prompt={prompt_t} completion={compl_t}")
+
+        message: dict = {"role": "assistant", "content": collected_content or None}
+        if collected_reasoning:
+            message["reasoning_content"] = collected_reasoning
+        if tool_calls_map:
+            message["tool_calls"] = [tool_calls_map[i] for i in sorted(tool_calls_map)]
+
+        return JSONResponse({
+            "id": f"chatcmpl-{uuid.uuid4().hex[:12]}",
+            "object": "chat.completion",
+            "created": int(time.time()),
+            "model": resp_model,
+            "choices": [{
+                "index": 0,
+                "message": message,
+                "finish_reason": finish_reason,
+            }],
+            "usage": usage,
+        })
+
+    raise HTTPException(status_code=502, detail="Upstream returned empty response")
+
+
+@app.get("/health")
+async def health():
+    has_token = bool(token_mgr.access_token)
+    expired = token_mgr._is_expired()
+    return {"status": "ok" if has_token and not expired else "degraded",
+            "has_token": has_token, "expired": expired}
+
+
+if __name__ == "__main__":
+    if not PROXY_API_KEY:
+        log.error("PROXY_API_KEY is not set. Set it via env var or .env file.")
+        log.error("Example: PROXY_API_KEY=$(openssl rand -hex 32) python server.py")
+        sys.exit(1)
+    if len(PROXY_API_KEY) < 16:
+        log.warning("PROXY_API_KEY is shorter than 16 chars — consider using a stronger key")
+    log.info(f"Starting WorkBuddy proxy on port {PROXY_PORT}")
+    log.info(f"WB version: {WB_VERSION}")
+    log.info(f"API key: {PROXY_API_KEY[:4]}***{PROXY_API_KEY[-4:] if len(PROXY_API_KEY) > 8 else ''}")
+    log.info(f"Upstream: {WB_API_BASE}")
+    uvicorn.run(app, host="0.0.0.0", port=PROXY_PORT, log_level="info")