tm731531/benchmark_runner.py

## benchmark_runner.py
"""
Walsin Gateway Benchmark Runner
================================
跑一組 prompts 對比多個 model(地端 vs cloud),產出 markdown 報表。

需要:
  pip install httpx
  ollama pull qwen3:14b qwen3:4b qwen2.5:0.5b  (any subset)
  export ANTHROPIC_API_KEY=sk-ant-...  (optional, 跑 cloud 對比才需要)

跑法:
  python3 benchmark_runner.py                # 全跑
  python3 benchmark_runner.py --suite swe    # 只跑 SWE-bench Lite 子集
  python3 benchmark_runner.py --suite custom # 只跑自家 prompts
  python3 benchmark_runner.py --models qwen3:14b,qwen3:4b
  python3 benchmark_runner.py --output report.md

輸出:Markdown 報表(每個 model × 每個 prompt 的 latency / output_tokens / 截斷回應)
本工具不做能力打分,只跑出可比較的數據,讓人類自己判斷。
"""
import argparse
import asyncio
import json
import time
from pathlib import Path

import httpx

# ============================================================
# Test suites
# ============================================================
SWE_BENCH_LITE_SUBSET = [
    {
        "id": "swe-1-flask-redirect",
        "category": "bug-fix",
        "prompt": "I have a Flask app where redirect() returns 308 instead of 302. "
                  "What's the most likely cause and one-line fix?",
    },
    {
        "id": "swe-2-django-orm",
        "category": "orm",
        "prompt": "Django QuerySet: how to do a LEFT JOIN and filter on the joined table's NULL? "
                  "Give the ORM expression, not raw SQL.",
    },
    {
        "id": "swe-3-pytest-async",
        "category": "test",
        "prompt": "pytest-asyncio test 一直 fail with 'async def functions are not natively supported'. "
                  "Two minimal fixes (config-based and decorator-based).",
    },
    {
        "id": "swe-4-numpy-broadcast",
        "category": "numerics",
        "prompt": "numpy: I have a (3,) array and (3, 5) array, want elementwise multiply with broadcasting. "
                  "Show the .reshape or np.newaxis solution.",
    },
    {
        "id": "swe-5-typing",
        "category": "typing",
        "prompt": "Python TypedDict 怎麼宣告選擇性欄位 + Required 欄位混用? Give 5-line example.",
    },
]

CUSTOM_PROMPTS = [
    {
        "id": "custom-1-kafka",
        "category": "concept",
        "prompt": "用一句話說明 Kafka 跟 RabbitMQ 的差別。中文 30 字內。",
    },
    {
        "id": "custom-2-osgi",
        "category": "domain",
        "prompt": "iDempiere OSGi bundle 部署常見的 ClassNotFoundException,3 個最常見的根本原因。",
    },
    {
        "id": "custom-3-sse",
        "category": "protocol",
        "prompt": "Anthropic Messages SSE streaming 共有幾個 event 類型?列出來並說明每個的用途,中文。",
    },
    {
        "id": "custom-4-tool-use",
        "category": "agent",
        "prompt": "解釋 LLM tool use 的 input_json_delta 是什麼?為什麼不是直接送完整 JSON?",
    },
    {
        "id": "custom-5-ollama",
        "category": "ops",
        "prompt": "Ollama 模型第一次 load 很慢,有哪 3 種方法可以加速?",
    },
]


# ============================================================
# Runners
# ============================================================
async def run_ollama(model: str, prompt: str, timeout: float = 600.0) -> dict:
    """跑單一 prompt 到 Ollama。回 {ok, latency_s, output, output_tokens, error}。"""
    url = "http://localhost:11434/v1/chat/completions"
    body = {
        "model": model,
        "messages": [{"role": "user", "content": prompt}],
        "stream": False,
        "max_tokens": 1024,
    }
    t0 = time.time()
    try:
        async with httpx.AsyncClient(timeout=timeout) as client:
            r = await client.post(url, json=body)
            data = r.json()
        out = data["choices"][0]["message"]["content"]
        usage = data.get("usage", {})
        return {
            "ok": True,
            "latency_s": round(time.time() - t0, 2),
            "output": out,
            "output_tokens": usage.get("completion_tokens", 0),
            "error": None,
        }
    except Exception as e:
        return {
            "ok": False,
            "latency_s": round(time.time() - t0, 2),
            "output": "",
            "output_tokens": 0,
            "error": f"{type(e).__name__}: {e}",
        }


async def run_anthropic(model: str, prompt: str, api_key: str, timeout: float = 300.0) -> dict:
    """跑單一 prompt 到 Anthropic Messages API。"""
    url = "https://api.anthropic.com/v1/messages"
    headers = {
        "x-api-key": api_key,
        "anthropic-version": "2023-06-01",
        "content-type": "application/json",
    }
    body = {
        "model": model,
        "max_tokens": 1024,
        "messages": [{"role": "user", "content": prompt}],
    }
    t0 = time.time()
    try:
        async with httpx.AsyncClient(timeout=timeout) as client:
            r = await client.post(url, json=body, headers=headers)
            data = r.json()
        out_blocks = data.get("content", [])
        out = "\n".join(b.get("text", "") for b in out_blocks if b.get("type") == "text")
        return {
            "ok": True,
            "latency_s": round(time.time() - t0, 2),
            "output": out,
            "output_tokens": data.get("usage", {}).get("output_tokens", 0),
            "error": None,
        }
    except Exception as e:
        return {
            "ok": False,
            "latency_s": round(time.time() - t0, 2),
            "output": "",
            "output_tokens": 0,
            "error": f"{type(e).__name__}: {e}",
        }


# ============================================================
# Main
# ============================================================
def truncate(s: str, n: int = 200) -> str:
    s = s.replace("\n", " ").strip()
    return s if len(s) <= n else s[:n] + "..."


async def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--suite", choices=["swe", "custom", "all"], default="all")
    ap.add_argument("--models", default="qwen3:14b,qwen3:4b,qwen2.5:0.5b",
                    help="comma-separated Ollama models")
    ap.add_argument("--anthropic-models", default="",
                    help="comma-separated Anthropic models (e.g. claude-opus-4-7),需要 ANTHROPIC_API_KEY")
    ap.add_argument("--output", default="benchmark_report.md")
    ap.add_argument("--timeout", type=float, default=600.0)
    args = ap.parse_args()

    import os
    api_key = os.environ.get("ANTHROPIC_API_KEY", "")
    anthropic_models = [m.strip() for m in args.anthropic_models.split(",") if m.strip()]
    ollama_models = [m.strip() for m in args.models.split(",") if m.strip()]

    if anthropic_models and not api_key:
        print("⚠️  --anthropic-models 設了但 ANTHROPIC_API_KEY 沒設,Anthropic 部分會跳過")
        anthropic_models = []

    suites = {
        "swe": SWE_BENCH_LITE_SUBSET,
        "custom": CUSTOM_PROMPTS,
        "all": SWE_BENCH_LITE_SUBSET + CUSTOM_PROMPTS,
    }
    prompts = suites[args.suite]

    print(f"Running {len(prompts)} prompts × {len(ollama_models) + len(anthropic_models)} models...")
    print(f"Ollama: {ollama_models}")
    print(f"Anthropic: {anthropic_models or '(none)'}")
    print()

    results = []
    for i, p in enumerate(prompts, 1):
        for m in ollama_models:
            print(f"  [{i}/{len(prompts)}] {p['id']} × {m} ...", end="", flush=True)
            r = await run_ollama(m, p["prompt"], args.timeout)
            results.append({"prompt_id": p["id"], "category": p["category"],
                            "model": m, "backend": "ollama", **r})
            print(f" {r['latency_s']}s {'OK' if r['ok'] else 'FAIL'}")

        for m in anthropic_models:
            print(f"  [{i}/{len(prompts)}] {p['id']} × {m} (cloud) ...", end="", flush=True)
            r = await run_anthropic(m, p["prompt"], api_key, args.timeout)
            results.append({"prompt_id": p["id"], "category": p["category"],
                            "model": m, "backend": "anthropic", **r})
            print(f" {r['latency_s']}s {'OK' if r['ok'] else 'FAIL'}")

    # 寫 markdown 報表
    out = ["# Walsin Gateway Benchmark Report", "",
           f"_Generated_: {time.strftime('%Y-%m-%d %H:%M:%S')}", "",
           f"_Prompts_: {len(prompts)} (suite={args.suite})", "",
           f"_Models_: ollama={ollama_models}, anthropic={anthropic_models}", "",
           "## Summary by Model", ""]

    by_model = {}
    for r in results:
        by_model.setdefault(r["model"], []).append(r)

    out.append("| Model | Backend | OK | Avg latency | Avg output_tokens |")
    out.append("|---|---|---|---|---|")
    for m, rs in by_model.items():
        ok = sum(1 for r in rs if r["ok"])
        avg_lat = sum(r["latency_s"] for r in rs) / max(1, len(rs))
        avg_tok = sum(r["output_tokens"] for r in rs) / max(1, len(rs))
        backend = rs[0]["backend"]
        out.append(f"| {m} | {backend} | {ok}/{len(rs)} | {avg_lat:.1f}s | {avg_tok:.0f} |")

    out.append("\n## Detailed Results\n")
    for p in prompts:
        out.append(f"### {p['id']} ({p['category']})\n")
        out.append(f"**Prompt**: {p['prompt']}\n")
        out.append("| Model | Latency | Tokens | Output (truncated) |")
        out.append("|---|---|---|---|")
        for r in [r for r in results if r["prompt_id"] == p["id"]]:
            out.append(f"| {r['model']} | {r['latency_s']}s | {r['output_tokens']} | "
                       f"{truncate(r['output'] if r['ok'] else r['error'])} |")
        out.append("")

    Path(args.output).write_text("\n".join(out))
    print(f"\n✓ Report written: {args.output}")


if __name__ == "__main__":
    asyncio.run(main())

## demo_record.sh
#!/bin/bash
# Walsin Gateway 60-second demo recording script
# ============================================================
# 用法:
#   asciinema rec -c "bash demo_record.sh" walsin-demo.cast
#   asciinema upload walsin-demo.cast    # (可選)上傳分享
# ============================================================
# 不裝 asciinema 也能直接跑(看 terminal output):
#   bash demo_record.sh
# ============================================================
# 前提:
#   - Gateway 已啟動: python3 gateway_v2_cc.py &
#   - Ollama 已 pull qwen3:14b 跟 qwen2.5:0.5b(任一)
#   - curl + python3 已裝
# ============================================================
set -e

GW="http://localhost:4000"
KEY="${MASTER_KEY:-sk-walsin-test}"
SLEEP_BETWEEN=2

# v2.3 pre-flight check
if ! curl -s -o /dev/null --max-time 3 "$GW/health"; then
    printf "\033[31m❌ Gateway 沒在 %s 跑。先啟動:\033[0m\n" "$GW"
    echo "   cd ~/walsin-gateway-demo"
    echo "   python3 gateway_v2_cc.py &"
    echo ""
    echo "   或設 MASTER_KEY:"
    echo "   MASTER_KEY=sk-your-key python3 gateway_v2_cc.py &"
    exit 1
fi

cyan()  { printf "\033[36m%s\033[0m\n" "$1"; }
green() { printf "\033[32m%s\033[0m\n" "$1"; }
yellow(){ printf "\033[33m%s\033[0m\n" "$1"; }
gray()  { printf "\033[90m%s\033[0m\n" "$1"; }

cyan "=========================================="
cyan " Walsin Gateway 60s Demo"
cyan "=========================================="
echo
gray "目標:展示 Gateway 看 prompt 自動分級路由"
echo
sleep $SLEEP_BETWEEN

# Step 1: Health check
yellow "[Step 1] Health check"
echo '$ curl $GW/health'
curl -s "$GW/health" | python3 -m json.tool
echo
sleep $SLEEP_BETWEEN

# Step 2: C 級 prompt
yellow "[Step 2] C 級 prompt(技術問題,應走 cloud / fallback 地端)"
echo '$ curl ... -d {"messages":[{"role":"user","content":"What is Kafka?"}]}'
RESP=$(curl -s -i "$GW/v1/messages" \
  -H "x-api-key: $KEY" \
  -H "content-type: application/json" \
  -H "anthropic-version: 2023-06-01" \
  -d '{
    "model": "claude-opus-4-7",
    "max_tokens": 80,
    "messages": [{"role":"user","content":"用 20 字說 Kafka 是什麼?"}]
  }')

green "  Headers:"
echo "$RESP" | grep -i "x-gateway" | sed 's/^/    /'
green "  Response:"
echo "$RESP" | tail -1 | python3 -c "
import sys, json
try:
    r = json.loads(sys.stdin.read())
    if 'content' in r:
        for b in r['content']:
            if b.get('type') == 'text':
                print('   ', b['text'][:200])
    else:
        print('   ', r)
except Exception as e:
    print('    (parse error:', e, ')')
"
echo
sleep $SLEEP_BETWEEN

# Step 3: A 級 prompt
yellow "[Step 3] A 級 prompt(命中字典,強制地端,即使有 cloud key)"
echo '$ curl ... -d {"messages":[{"role":"user","content":"[client_alpha] 出問題"}]}'
RESP=$(curl -s -i "$GW/v1/messages" \
  -H "x-api-key: $KEY" \
  -H "content-type: application/json" \
  -H "anthropic-version: 2023-06-01" \
  --max-time 60 \
  -d '{
    "model": "claude-opus-4-7",
    "max_tokens": 80,
    "messages": [{"role":"user","content":"[client_alpha] 出問題了,30 字內怎修?"}]
  }')

green "  Headers:"
echo "$RESP" | grep -i "x-gateway" | sed 's/^/    /'
green "  Response:"
echo "$RESP" | tail -1 | python3 -c "
import sys, json
try:
    r = json.loads(sys.stdin.read())
    if 'content' in r:
        for b in r['content']:
            if b.get('type') == 'text':
                print('   ', b['text'][:200])
    else:
        print('   ', r)
except Exception as e:
    print('    (parse error:', e, ')')
"
echo
sleep $SLEEP_BETWEEN

# Step 4: Auth fail
yellow "[Step 4] 沒帶 key 401(v2.1 安全修正)"
echo '$ curl $GW/v1/messages  (no auth header)'
curl -s -o /dev/null -w "  HTTP %{http_code}\n" "$GW/v1/messages" \
  -X POST -H "content-type: application/json" \
  -d '{"model":"x","max_tokens":10,"messages":[{"role":"user","content":"hi"}]}'
echo
sleep 1

cyan "=========================================="
green "  Demo done. 看到 x-gateway-decision 在 header 裡"
green "  A 級永遠地端,C 級可走 cloud,B 級完整 fallback"
green "  搬離方法論在 Mini PC 上 work。"
cyan "=========================================="

## gateway_v2_cc.py
"""
Walsin Gateway v2.3 — Self-review 後定版(P0 全清)
========================================================
v1(80 行):用 /v1/chat/completions(OpenAI),CC 打 /v1/messages 會 404
v2.0(364 行):Anthropic 原生 endpoint + 完整翻譯層
v2.1(502 行):修 3 個 review#2 點出的 bug
  - Auth 邏輯反轉 → 無條件檢查
  - Streaming tool use 不工作 → 加 tool_calls delta
  - stop_reason 寫死 → 真實 finish_reason 映射
v2.2(620 行):TODO 全部 close(B 級實作 + tiktoken + benchmark + demo)
v2.3(本版,674 行):self-review 找出 7 個漏洞,全清:
  - 🔴 Auth substring match 漏洞 → secrets.compare_digest 精確比對
  - 🔴 SSE 透傳格式錯 → aiter_lines 改 aiter_bytes 直通
  - 🔴 Sanitize 漏 tool_use input / tool_result content → 改遞迴
  - 🟡 MASTER_KEY hardcoded default → 環境變數沒設給 warning
  - 🟡 demo_record.sh 缺 pre-flight check
  - 🟡 /health 沒回報 ollama 狀態
  - 🟡 B 級走 cloud 沒標 sanitized → 加 X-Gateway-Sanitized header

設計原則:
  - A 級:強制地端,即使 ANTHROPIC_API_KEY 設了也不走 cloud
  - B 級:地端優先 → 地端死了 sanitize 後 cloud → sanitize 沒命中拒絕
  - C 級:cloud 優先,沒 key 才 fallback 地端

需求:
  pip install fastapi uvicorn httpx tiktoken
  ollama pull qwen3:14b   # A/B 級主處理(地端最強)
  export ANTHROPIC_API_KEY=sk-ant-... (optional,不設則純地端 + B 級 fallback 失效)

啟動:
  python3 gateway_v2_cc.py

接 CC:
  export ANTHROPIC_BASE_URL=http://localhost:4000
  export ANTHROPIC_AUTH_TOKEN=sk-walsin-test
  claude
"""
import os
import re
import json
import uuid
import copy
import secrets
import logging
from typing import Optional
import httpx
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse, StreamingResponse

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
log = logging.getLogger("gateway-v2")

# ============================================================
# 配置 — A/B/C 三級分類字典 + B 級脫敏字典
# ============================================================
# 設計原則(雙維度:敏感度 × 可用性):
#   A 級:強制地端,即使 ANTHROPIC_API_KEY 設了也不走 cloud
#         理由:資料外洩風險 > 模型能力差異
#   B 級:地端優先,地端失敗 fallback 脫敏後 cloud
#         理由:能脫敏 → 可降級用 cloud 求能力
#   C 級:cloud 優先,沒 key 才 fallback 地端
#         理由:純技術問題,無敏感度
#
# v2.2:B 級完整實作 — 試地端 → 失敗則 sanitize → fallback cloud

A_KEYWORDS = [
    "[client_alpha]", "[project_xxx]", "BOM_FORMULA",
    # A 級:絕對不能上 cloud(就算脫敏也不行,全程地端)
]

B_KEYWORDS = [
    "[internal_process]", "[vendor_quote]", "[employee_name]",
    # B 級:能脫敏就上 cloud,脫敏失敗就退回地端
]

# B 級脫敏映射(命中關鍵字 → 替換為 placeholder)
SANITIZE_MAP = {
    r"\[internal_process\]": "[PROCESS]",
    r"\[vendor_quote\]": "[QUOTE]",
    r"\[employee_name\]": "[PERSON]",
    # 通用 PII patterns
    r"\b[\w.+-]+@[\w-]+\.[\w.-]+\b": "[EMAIL]",
    r"\b(?:\d{1,3}\.){3}\d{1,3}\b": "[IP]",
    r"\b\d{4}-\d{4}-\d{4}-\d{4}\b": "[CARD]",
}
SANITIZE_REGEX = [(re.compile(p, re.IGNORECASE), repl) for p, repl in SANITIZE_MAP.items()]

A_PATTERN = re.compile("|".join(re.escape(k) for k in A_KEYWORDS), re.IGNORECASE) if A_KEYWORDS else None
B_PATTERN = re.compile("|".join(re.escape(k) for k in B_KEYWORDS), re.IGNORECASE) if B_KEYWORDS else None

OLLAMA_URL = os.environ.get("OLLAMA_URL", "http://localhost:11434")
MODEL_A_LEVEL = os.environ.get("MODEL_A_LEVEL", "qwen3:14b")     # A 級用強模型
MODEL_B_LEVEL = os.environ.get("MODEL_B_LEVEL", "qwen3:14b")     # B 級先試地端
ANTHROPIC_API_KEY = os.environ.get("ANTHROPIC_API_KEY")           # 沒設 → C 級也走地端
ANTHROPIC_FALLBACK_MODEL = os.environ.get("ANTHROPIC_FALLBACK_MODEL", "qwen3:14b")
LOCAL_HEALTH_TIMEOUT = float(os.environ.get("LOCAL_HEALTH_TIMEOUT", "5.0"))  # B 級 試地端 healthcheck timeout
B_LEVEL_CLOUD_FALLBACK = os.environ.get("B_LEVEL_CLOUD_FALLBACK", "1") == "1"  # 關掉 = B 級永遠地端

# v2.3 Master key — 強制環境變數,沒設給警告(原本 hardcoded default 是壞習慣)
MASTER_KEY = os.environ.get("MASTER_KEY")
if not MASTER_KEY:
    MASTER_KEY = "sk-walsin-test"
    log.warning(
        "⚠️  MASTER_KEY 環境變數沒設,用 insecure default 'sk-walsin-test'。"
        "上線前必設 MASTER_KEY=<random>。"
    )

# Tiktoken approximation(對 Qwen 偏估,但比 chunk count 準很多)
try:
    import tiktoken
    _TIKTOKEN_ENC = tiktoken.encoding_for_model("gpt-4")
    def count_tokens(text: str) -> int:
        return len(_TIKTOKEN_ENC.encode(text)) if text else 0
except ImportError:
    log.warning("tiktoken 未安裝,token usage 會用粗估(每 4 chars ≈ 1 token)")
    def count_tokens(text: str) -> int:
        return max(1, len(text) // 4) if text else 0

app = FastAPI()

# ============================================================
# Classifier
# ============================================================
def extract_text(messages, system=None):
    """抽出所有文字內容(用於分類)。"""
    parts = []
    if isinstance(system, str):
        parts.append(system)
    elif isinstance(system, list):
        for s in system:
            if isinstance(s, dict) and s.get("type") == "text":
                parts.append(s.get("text", ""))
    for msg in messages:
        c = msg.get("content", "")
        if isinstance(c, str):
            parts.append(c)
        elif isinstance(c, list):
            for x in c:
                if isinstance(x, dict) and x.get("type") == "text":
                    parts.append(x.get("text", ""))
    return "\n".join(parts)


def classify(messages, system=None):
    """三級分類:A → 強制地端,B → 地端優先,C → cloud 優先。"""
    text = extract_text(messages, system)
    if A_PATTERN:
        m = A_PATTERN.search(text)
        if m:
            return ("A", m.group())
    if B_PATTERN:
        m = B_PATTERN.search(text)
        if m:
            return ("B", m.group())
    return ("C", None)


# ============================================================
# Sanitization (B 級 cloud fallback 前用)
# ============================================================
def sanitize_string(text: str) -> tuple[str, bool]:
    """脫敏文字。回傳 (脫敏後文字, 是否有命中替換)。"""
    if not text:
        return text, False
    hit = False
    out = text
    # 先擋 A 級字典(B 級 fallback 時不能讓 A 級 keyword 漏出去)
    if A_PATTERN:
        new = A_PATTERN.sub("[A_REDACTED]", out)
        if new != out:
            hit = True
            out = new
    # 再脫敏 B 級對應 placeholder
    for pat, repl in SANITIZE_REGEX:
        new = pat.sub(repl, out)
        if new != out:
            hit = True
            out = new
    return out, hit


def _sanitize_value(v):
    """v2.3 遞迴 sanitize — 處理 str / dict / list,涵蓋 tool_use input、tool_result content 等任意巢狀。
    回 (sanitized_value, hit)。"""
    if isinstance(v, str):
        return sanitize_string(v)
    if isinstance(v, dict):
        out = {}
        any_hit = False
        for k, vv in v.items():
            nv, h = _sanitize_value(vv)
            out[k] = nv
            any_hit = any_hit or h
        return out, any_hit
    if isinstance(v, list):
        out = []
        any_hit = False
        for item in v:
            ni, h = _sanitize_value(item)
            out.append(ni)
            any_hit = any_hit or h
        return out, any_hit
    # int / float / bool / None — 不動
    return v, False


def sanitize_anthropic_body(body: dict) -> tuple[dict, bool]:
    """對 Anthropic Messages body 脫敏。
    v2.3:遞迴處理 system / messages 全部,涵蓋:
      - text content blocks
      - tool_use input(JSON dict)
      - tool_result content(string or list of blocks)
      - 任意巢狀 dict / list
    """
    out = copy.deepcopy(body)
    any_hit = False

    if "system" in out:
        new_sys, h = _sanitize_value(out["system"])
        out["system"] = new_sys
        any_hit = any_hit or h

    if "messages" in out:
        new_msgs, h = _sanitize_value(out["messages"])
        out["messages"] = new_msgs
        any_hit = any_hit or h

    return out, any_hit


def _check_master_key(auth_header: str) -> bool:
    """v2.3 安全:精確比對 master_key,避免 substring match 漏洞。
    支援 'Bearer XXX' / 'XXX' / 純 x-api-key。timing-safe 比對。"""
    if not auth_header:
        return False
    parts = auth_header.split(None, 1)
    if len(parts) == 2 and parts[0].lower() == "bearer":
        provided = parts[1]
    else:
        provided = parts[0] if len(parts) == 1 else auth_header
    return secrets.compare_digest(provided.encode(), MASTER_KEY.encode())


async def ollama_alive() -> bool:
    """檢查地端 Ollama 是否健在(B 級 fallback 判斷)。"""
    try:
        async with httpx.AsyncClient(timeout=LOCAL_HEALTH_TIMEOUT) as client:
            r = await client.get(f"{OLLAMA_URL}/api/tags")
            return r.status_code == 200
    except Exception:
        return False


# ============================================================
# Anthropic ↔ OpenAI Translation
# ============================================================
def anthropic_to_openai_request(body: dict, target_model: str) -> dict:
    """Anthropic Messages → OpenAI Chat Completions。"""
    openai_messages = []

    # Anthropic system 欄位 → OpenAI system message
    sys = body.get("system")
    if isinstance(sys, str) and sys:
        openai_messages.append({"role": "system", "content": sys})
    elif isinstance(sys, list):
        sys_text = "\n".join(s.get("text", "") for s in sys if s.get("type") == "text")
        if sys_text:
            openai_messages.append({"role": "system", "content": sys_text})

    # Anthropic messages → OpenAI messages
    for msg in body.get("messages", []):
        role = msg.get("role", "user")
        content = msg.get("content", "")

        # 字串 content 直接搬
        if isinstance(content, str):
            openai_messages.append({"role": role, "content": content})
            continue

        # list content(text + tool_use + tool_result)
        if isinstance(content, list):
            text_parts = []
            tool_calls = []
            tool_results = []
            for block in content:
                btype = block.get("type")
                if btype == "text":
                    text_parts.append(block.get("text", ""))
                elif btype == "tool_use":
                    tool_calls.append({
                        "id": block.get("id", ""),
                        "type": "function",
                        "function": {
                            "name": block.get("name", ""),
                            "arguments": json.dumps(block.get("input", {})),
                        },
                    })
                elif btype == "tool_result":
                    # OpenAI 要 tool result 變成獨立 message
                    result_content = block.get("content", "")
                    if isinstance(result_content, list):
                        result_content = "\n".join(
                            x.get("text", "") for x in result_content if x.get("type") == "text"
                        )
                    tool_results.append({
                        "role": "tool",
                        "tool_call_id": block.get("tool_use_id", ""),
                        "content": str(result_content),
                    })

            # assemble
            msg_out = {"role": role}
            if text_parts:
                msg_out["content"] = "\n".join(text_parts)
            if tool_calls:
                msg_out["tool_calls"] = tool_calls
                msg_out.setdefault("content", None)
            if msg_out.get("content") is not None or tool_calls:
                openai_messages.append(msg_out)
            for tr in tool_results:
                openai_messages.append(tr)

    out = {
        "model": target_model,
        "messages": openai_messages,
        "max_tokens": body.get("max_tokens", 4096),
        "stream": body.get("stream", False),
    }
    if "temperature" in body:
        out["temperature"] = body["temperature"]

    # Anthropic tools → OpenAI tools
    if body.get("tools"):
        out["tools"] = [
            {
                "type": "function",
                "function": {
                    "name": t["name"],
                    "description": t.get("description", ""),
                    "parameters": t.get("input_schema", {}),
                },
            }
            for t in body["tools"]
        ]
    return out


def openai_to_anthropic_response(openai_resp: dict, original_model: str) -> dict:
    """OpenAI Chat 回應 → Anthropic Messages 回應格式。"""
    choice = openai_resp.get("choices", [{}])[0]
    msg = choice.get("message", {})

    content_blocks = []

    # text content
    if msg.get("content"):
        content_blocks.append({"type": "text", "text": msg["content"]})

    # tool_calls → tool_use blocks
    for tc in msg.get("tool_calls", []) or []:
        fn = tc.get("function", {})
        try:
            tool_input = json.loads(fn.get("arguments", "{}"))
        except json.JSONDecodeError:
            tool_input = {}
        content_blocks.append({
            "type": "tool_use",
            "id": tc.get("id", f"toolu_{uuid.uuid4().hex[:16]}"),
            "name": fn.get("name", ""),
            "input": tool_input,
        })

    finish = choice.get("finish_reason", "stop")
    stop_reason = {
        "stop": "end_turn",
        "length": "max_tokens",
        "tool_calls": "tool_use",
    }.get(finish, "end_turn")

    usage = openai_resp.get("usage", {})
    return {
        "id": openai_resp.get("id", f"msg_{uuid.uuid4().hex[:16]}"),
        "type": "message",
        "role": "assistant",
        "model": original_model,
        "content": content_blocks or [{"type": "text", "text": ""}],
        "stop_reason": stop_reason,
        "stop_sequence": None,
        "usage": {
            "input_tokens": usage.get("prompt_tokens", 0),
            "output_tokens": usage.get("completion_tokens", 0),
        },
    }


# ============================================================
# SSE Streaming Translation (OpenAI → Anthropic)
# v2.1 修正:
#   - tool_calls delta 累積 + 對映 Anthropic content_block_start (tool_use)
#     + input_json_delta(原本只翻譯 text,tool 會卡死 CC)
#   - stop_reason 用真實 finish_reason 映射(原本寫死 end_turn)
#   - content blocks 改 lazy open(只在真有內容時才送 start)
# ============================================================
def _sse(event: str, payload: dict) -> str:
    return f"event: {event}\ndata: {json.dumps(payload)}\n\n"


async def stream_anthropic_from_openai(openai_stream, msg_id: str, model: str):
    """OpenAI SSE → Anthropic SSE。處理 text + tool_use 兩種 content block。"""
    # 開頭固定送 message_start
    yield _sse("message_start", {
        "type": "message_start",
        "message": {
            "id": msg_id, "type": "message", "role": "assistant", "model": model,
            "content": [], "stop_reason": None,
            "usage": {"input_tokens": 0, "output_tokens": 0},
        },
    })

    # block 狀態管理(lazy open)
    next_block_index = 0
    text_block_index: Optional[int] = None      # text block 的 index(若已開)
    tool_blocks: dict = {}                       # OpenAI tool_call.index -> {our_index, id, name, args_buffer}
    finish_reason = "stop"
    accumulated_text = ""                        # 累積 text 用於 token 計算

    async for line in openai_stream:
        line = line.strip()
        if not line or not line.startswith("data:"):
            continue
        data_str = line[5:].strip()
        if data_str == "[DONE]":
            break
        try:
            data = json.loads(data_str)
        except json.JSONDecodeError:
            continue

        choice = data.get("choices", [{}])[0]
        delta = choice.get("delta", {}) or {}
        if choice.get("finish_reason"):
            finish_reason = choice["finish_reason"]

        # text content delta
        text = delta.get("content")
        if text:
            if text_block_index is None:
                # lazy open text block
                text_block_index = next_block_index
                next_block_index += 1
                yield _sse("content_block_start", {
                    "type": "content_block_start",
                    "index": text_block_index,
                    "content_block": {"type": "text", "text": ""},
                })
            # v2.2:用 tiktoken 估算真實 token,而不是 chunk count
            accumulated_text += text
            yield _sse("content_block_delta", {
                "type": "content_block_delta",
                "index": text_block_index,
                "delta": {"type": "text_delta", "text": text},
            })

        # tool_calls delta(關鍵修正:CC 的 Read/Edit/Bash 都需要這個)
        for tc in delta.get("tool_calls") or []:
            tc_idx = tc.get("index", 0)
            fn = tc.get("function", {}) or {}

            if tc_idx not in tool_blocks:
                # 新 tool_call:close 之前的 text block(if open)
                if text_block_index is not None:
                    yield _sse("content_block_stop", {
                        "type": "content_block_stop", "index": text_block_index,
                    })
                    text_block_index = None

                # open 新 tool_use block
                our_idx = next_block_index
                next_block_index += 1
                tool_blocks[tc_idx] = {
                    "our_index": our_idx,
                    "id": tc.get("id") or f"toolu_{uuid.uuid4().hex[:16]}",
                    "name": fn.get("name", ""),
                    "args_buffer": "",
                }
                yield _sse("content_block_start", {
                    "type": "content_block_start",
                    "index": our_idx,
                    "content_block": {
                        "type": "tool_use",
                        "id": tool_blocks[tc_idx]["id"],
                        "name": tool_blocks[tc_idx]["name"],
                        "input": {},
                    },
                })

            # arguments delta(partial JSON)
            args_delta = fn.get("arguments")
            if args_delta:
                tool_blocks[tc_idx]["args_buffer"] += args_delta
                yield _sse("content_block_delta", {
                    "type": "content_block_delta",
                    "index": tool_blocks[tc_idx]["our_index"],
                    "delta": {"type": "input_json_delta", "partial_json": args_delta},
                })

    # close 所有開著的 blocks
    if text_block_index is not None:
        yield _sse("content_block_stop", {
            "type": "content_block_stop", "index": text_block_index,
        })
    for tc_data in tool_blocks.values():
        yield _sse("content_block_stop", {
            "type": "content_block_stop", "index": tc_data["our_index"],
        })

    # 真實 stop_reason(不再寫死 end_turn)
    stop_reason = {
        "stop": "end_turn",
        "length": "max_tokens",
        "tool_calls": "tool_use",
        "function_call": "tool_use",
        "content_filter": "end_turn",
    }.get(finish_reason, "end_turn")

    # v2.2:tiktoken 估算 output_tokens(text + tool args 都算)
    output_tokens = count_tokens(accumulated_text)
    for tc_data in tool_blocks.values():
        output_tokens += count_tokens(tc_data["args_buffer"])

    yield _sse("message_delta", {
        "type": "message_delta",
        "delta": {"stop_reason": stop_reason, "stop_sequence": None},
        "usage": {"output_tokens": output_tokens},
    })
    yield _sse("message_stop", {"type": "message_stop"})


# ============================================================
# Backend Forwarders
# ============================================================
async def forward_to_ollama(anthropic_body: dict, target_model: str, original_model: str):
    """A 級 → 翻譯成 OpenAI format,forward to Ollama。"""
    openai_body = anthropic_to_openai_request(anthropic_body, target_model)
    is_stream = openai_body.get("stream", False)

    if is_stream:
        async def gen():
            async with httpx.AsyncClient(timeout=600) as client:
                async with client.stream(
                    "POST",
                    f"{OLLAMA_URL}/v1/chat/completions",
                    json=openai_body,
                ) as r:
                    msg_id = f"msg_{uuid.uuid4().hex[:16]}"
                    async for chunk in stream_anthropic_from_openai(r.aiter_lines(), msg_id, original_model):
                        yield chunk
        return StreamingResponse(gen(), media_type="text/event-stream",
                                 headers={"X-Gateway-Decision": "A_LOCAL", "X-Gateway-Model": target_model})
    else:
        async with httpx.AsyncClient(timeout=600) as client:
            r = await client.post(f"{OLLAMA_URL}/v1/chat/completions", json=openai_body)
            data = r.json()
        anthropic_resp = openai_to_anthropic_response(data, original_model)
        return JSONResponse(anthropic_resp,
                            headers={"X-Gateway-Decision": "A_LOCAL", "X-Gateway-Model": target_model})


async def forward_to_anthropic(anthropic_body: dict, request: Request, original_model: str):
    """C 級 → 直接 proxy 到 api.anthropic.com(streaming + non-streaming)。"""
    if not ANTHROPIC_API_KEY:
        # 沒 key → fallback 走地端 14b
        log.warning("[C-LEVEL] ANTHROPIC_API_KEY 沒設 → fallback to local %s", ANTHROPIC_FALLBACK_MODEL)
        return await forward_to_ollama(anthropic_body, ANTHROPIC_FALLBACK_MODEL, original_model)

    headers = {
        "x-api-key": ANTHROPIC_API_KEY,
        "anthropic-version": request.headers.get("anthropic-version", "2023-06-01"),
        "content-type": "application/json",
    }
    is_stream = anthropic_body.get("stream", False)

    if is_stream:
        # v2.3:用 aiter_bytes 直通 — aiter_lines 會剝掉 \n\n event 結尾,SSE 格式會壞
        async def gen():
            async with httpx.AsyncClient(timeout=600) as client:
                async with client.stream(
                    "POST", "https://api.anthropic.com/v1/messages",
                    json=anthropic_body, headers=headers,
                ) as r:
                    async for chunk in r.aiter_bytes():
                        yield chunk
        return StreamingResponse(gen(), media_type="text/event-stream",
                                 headers={"X-Gateway-Decision": "C_CLOUD", "X-Gateway-Model": original_model})
    else:
        async with httpx.AsyncClient(timeout=600) as client:
            r = await client.post("https://api.anthropic.com/v1/messages",
                                  json=anthropic_body, headers=headers)
            return JSONResponse(r.json(), status_code=r.status_code,
                                headers={"X-Gateway-Decision": "C_CLOUD", "X-Gateway-Model": original_model})


# ============================================================
# Main Endpoint
# ============================================================
@app.post("/v1/messages")
async def messages(request: Request):
    # v2.3 安全:精確 token 比對(避免 substring match 漏洞,例如 'sk-test' 通過 'sk-test-extra')
    # 兼容兩種 header:Authorization: Bearer XXX / x-api-key: XXX
    auth = (
        request.headers.get("authorization", "")
        or "Bearer " + (request.headers.get("x-api-key") or "")
    )
    if not _check_master_key(auth):
        raise HTTPException(401, "bad master key")

    body = await request.json()
    original_model = body.get("model", "claude-opus-4-7")
    decision, keyword = classify(body.get("messages", []), body.get("system"))

    # 雙維度決策:A 級永遠地端(即使有 cloud key);B 級地端優先 → fallback 脫敏 cloud;C 級 cloud 優先
    if decision == "A":
        log.warning(f"[A-LEVEL] 命中 '{keyword}' → 強制地端 {MODEL_A_LEVEL}(即使有 cloud key 也不走)")
        return await forward_to_ollama(body, MODEL_A_LEVEL, original_model)

    elif decision == "B":
        # v2.2 完整 B 級實作:試地端 → 失敗 sanitize → cloud(若 cloud key 在且 fallback 啟用)
        if await ollama_alive():
            log.warning(f"[B-LEVEL] 命中 '{keyword}' → 地端 {MODEL_B_LEVEL}(地端健在,優先地端)")
            return await forward_to_ollama(body, MODEL_B_LEVEL, original_model)

        # 地端死了,看能不能 fallback cloud
        if not (ANTHROPIC_API_KEY and B_LEVEL_CLOUD_FALLBACK):
            log.error(f"[B-LEVEL] 地端不可用 + 無 cloud fallback,拒絕請求")
            raise HTTPException(503, "B-level: local unavailable, cloud fallback disabled")

        sanitized_body, hit = sanitize_anthropic_body(body)
        if not hit:
            # 地端死了 + 沒有任何 sanitize 命中(意外狀況,B 級命中字典但脫敏沒抓到)
            log.error(f"[B-LEVEL] 地端不可用 + 脫敏未命中(B 字典 vs 脫敏字典不一致?)拒絕")
            raise HTTPException(500, "B-level: local down + sanitization mismatch")

        log.warning(f"[B-LEVEL] 命中 '{keyword}' → 地端不可用 → 脫敏後 fallback cloud(已 sanitize)")
        # v2.3:讓 client 知道資料被脫敏(透明度)
        resp = await forward_to_anthropic(sanitized_body, request, original_model)
        try:
            resp.headers["X-Gateway-Sanitized"] = "1"
        except Exception:
            pass  # streaming response header 已固定
        return resp

    else:
        log.info(f"[C-LEVEL] → cloud {original_model}" if ANTHROPIC_API_KEY else f"[C-LEVEL] no key → local {ANTHROPIC_FALLBACK_MODEL}")
        return await forward_to_anthropic(body, request, original_model)


@app.get("/health")
async def health():
    # v2.3:回報 ollama 狀態給 monitoring
    ollama_ok = await ollama_alive()
    return {
        "status": "ok",
        "ollama": "alive" if ollama_ok else "down",
        "a_level_model": MODEL_A_LEVEL,
        "b_level_model": MODEL_B_LEVEL,
        "b_cloud_fallback": B_LEVEL_CLOUD_FALLBACK,
        "c_level": "cloud" if ANTHROPIC_API_KEY else f"local fallback ({ANTHROPIC_FALLBACK_MODEL})",
        "keywords_count": len(A_KEYWORDS),
    }


if __name__ == "__main__":
    import uvicorn
    log.info(f"Walsin Gateway v2 starting...")
    log.info(f"  A-LEVEL model: {MODEL_A_LEVEL}")
    log.info(f"  C-LEVEL: {'cloud Claude' if ANTHROPIC_API_KEY else f'local fallback {ANTHROPIC_FALLBACK_MODEL}'}")
    log.info(f"  A_KEYWORDS: {len(A_KEYWORDS)} entries")
    uvicorn.run(app, host="0.0.0.0", port=4000, log_level="info")

## gist-readme.md

      
    Raw
  

              gist-readme.md
            
          
    Walsin Gateway v2.3 — Self-Review 後定版

本地 AI Gateway,真能接 Claude Code。從 v1(80 行,矛盾) → v2.0(364 行) → v2.1(502 行) → v2.2(620 行) → v2.3(674 行)。
完整方法論:腦子系統實證篇
三天五次迭代


版本
行數
觸發
結果


v1
80
初版
描述跟 code 矛盾


v2.0
364
review#1
文字自洽,無 code


v2.0 Gist
394
code 上線
1 安全 + 2 功能 bug


v2.1
502
review#2
bug 修完,4 TODO 待補


v2.2
620
TODO close
B 級實作 + tiktoken + benchmark + demo


v2.3
674
self-review
7 個漏洞清完 + 24 個 pytest 全綠


v2.3 self-review 找到的 7 個漏洞(全清)

🔴 P0:

Auth substring match 漏洞 — MASTER_KEY not in auth 太寬鬆,sk-test-extra 也通過 → 改 secrets.compare_digest 精確比對
SSE 透傳格式錯 — aiter_lines + "\n" 會剝掉 \n\n event 結尾 → 改 aiter_bytes 直通
Sanitize 漏 tool_use input + tool_result content — 只處理 text block → 改遞迴處理任意巢狀

🟡 P1:
4. MASTER_KEY 預設 hardcoded → 環境變數沒設給 warning
5. demo_record.sh 缺 pre-flight → 加 health check + 沒啟動友善提示
6. /health 沒回報 ollama 狀態 → 加 ollama: alive/down + b_level_model
7. B 級走 cloud 沒標 sanitized → 加 X-Gateway-Sanitized: 1 header
24 個 pytest 全綠

pip install --user pytest pytest-asyncio
MASTER_KEY=sk-test-secret python3 -m pytest test_gateway.py -v
# 24 passed in 0.61s
涵蓋:

classify(6 tests):A/B/C 分級 + system 欄位 + list content + A 優先 B
_check_master_key(7 tests):含 substring extra/prefix 漏洞防禦驗證
Sanitization(8 tests):含 tool_use input + tool_result content 遞迴
翻譯(3 tests):system / tool_use / tool_result

設計原則(雙維度:敏感度 × 可用性)


級別
主路由
Fallback
關鍵保護


A 級
地端最強(14b/32b/80B)
沒 fallback
即使有 cloud key 也不走 cloud


B 級
地端優先
地端死 → sanitize → cloud(標 X-Gateway-Sanitized)
不能脫敏寧願 503


C 級
cloud 優先
沒 key → 地端
純技術問題


檔案清單(Gist 5 個檔)


gateway_v2_cc.py(674 行)— 核心 Gateway
test_gateway.py(180 行)— pytest 24 個 test
benchmark_runner.py(258 行)— 多 model 對比 runner
demo_record.sh(125 行)— asciinema 60 秒 demo
gist-readme.md— 本 README

5 步驟啟動(production 級)

# 1. 拉模型
ollama pull qwen3:14b

# 2. 裝 Python 套件
pip install --user fastapi uvicorn httpx tiktoken

# 3. 跑 Gateway(必設 MASTER_KEY)
MASTER_KEY=sk-$(openssl rand -hex 16) python3 gateway_v2_cc.py &

# 4. CC 切過去
export ANTHROPIC_BASE_URL=http://localhost:4000
export ANTHROPIC_AUTH_TOKEN=$MASTER_KEY  # 跟上面一致

# 5. 用
claude
License

MIT — 拿去用。
相關文章

腦子系統九部曲 — 從個人到萬人企業的搬離方法論

  
## test_gateway.py
"""
Walsin Gateway 基本 unit test
============================
跑法:
  pip install --user pytest pytest-asyncio
  cd ~/walsin-gateway-demo
  python3 -m pytest test_gateway.py -v

涵蓋 4 個關鍵函式:
  1. classify — A/B/C 分級邏輯
  2. _check_master_key — auth 精確比對(防 substring 漏洞)
  3. _sanitize_value / sanitize_anthropic_body — 遞迴脫敏(含 tool_use)
  4. anthropic_to_openai_request — 格式翻譯
"""
import os
import sys

# 確保 import gateway 時不會 boot uvicorn(只 import,不 run)
os.environ.setdefault("MASTER_KEY", "sk-test-secret")

# 從 gateway_v2_cc 模組 import 要測的函式
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from gateway_v2_cc import (
    classify,
    _check_master_key,
    _sanitize_value,
    sanitize_anthropic_body,
    anthropic_to_openai_request,
    sanitize_string,
)


# ============================================================
# classify
# ============================================================
class TestClassify:
    def test_C_level_default(self):
        msgs = [{"role": "user", "content": "What is Kafka?"}]
        decision, kw = classify(msgs)
        assert decision == "C"
        assert kw is None

    def test_A_level_keyword_match(self):
        msgs = [{"role": "user", "content": "[client_alpha] 出問題了"}]
        decision, kw = classify(msgs)
        assert decision == "A"
        assert "client_alpha" in kw.lower()

    def test_A_level_in_system(self):
        msgs = [{"role": "user", "content": "hello"}]
        decision, kw = classify(msgs, system="你正在處理 [client_alpha] 案子")
        assert decision == "A"

    def test_A_level_in_list_content(self):
        msgs = [{"role": "user", "content": [
            {"type": "text", "text": "hello"},
            {"type": "text", "text": "BOM_FORMULA stuff"},
        ]}]
        decision, kw = classify(msgs)
        assert decision == "A"

    def test_B_level_match(self):
        msgs = [{"role": "user", "content": "[internal_process] 是什麼"}]
        decision, kw = classify(msgs)
        assert decision == "B"

    def test_A_takes_precedence_over_B(self):
        # 同時命中 A + B,A 優先
        msgs = [{"role": "user", "content": "[client_alpha] 跟 [internal_process] 都來"}]
        decision, kw = classify(msgs)
        assert decision == "A"


# ============================================================
# _check_master_key (v2.3 安全修正)
# ============================================================
class TestMasterKey:
    def test_correct_bearer(self):
        assert _check_master_key("Bearer sk-test-secret") is True

    def test_correct_bare(self):
        assert _check_master_key("sk-test-secret") is True

    def test_empty(self):
        assert _check_master_key("") is False

    def test_wrong(self):
        assert _check_master_key("Bearer wrong") is False

    def test_substring_extra_suffix_blocked(self):
        # v2.3 修正點:'sk-test-secret-extra' 不能通過(原 v2.1 substring match 會通過)
        assert _check_master_key("Bearer sk-test-secret-extra") is False

    def test_substring_prefix_blocked(self):
        assert _check_master_key("Bearer prefix-sk-test-secret") is False

    def test_lower_case_bearer(self):
        assert _check_master_key("bearer sk-test-secret") is True


# ============================================================
# Sanitization (v2.3 遞迴版)
# ============================================================
class TestSanitization:
    def test_string_email(self):
        out, hit = sanitize_string("聯絡 alice@example.com 處理")
        assert "[EMAIL]" in out
        assert "alice@example.com" not in out
        assert hit is True

    def test_string_ip(self):
        out, hit = sanitize_string("server 192.168.1.42 down")
        assert "[IP]" in out
        assert hit is True

    def test_string_no_hit(self):
        out, hit = sanitize_string("just normal text")
        assert out == "just normal text"
        assert hit is False

    def test_recursive_dict(self):
        v, hit = _sanitize_value({"name": "alice@example.com", "ok": 1})
        assert v["name"] == "[EMAIL]"
        assert v["ok"] == 1
        assert hit is True

    def test_recursive_list(self):
        v, hit = _sanitize_value(["alice@example.com", "no hit"])
        assert v[0] == "[EMAIL]"
        assert v[1] == "no hit"
        assert hit is True

    def test_anthropic_body_text_block(self):
        body = {
            "model": "claude-opus-4-7",
            "messages": [
                {"role": "user", "content": [
                    {"type": "text", "text": "聯絡 admin@walsin.com"},
                ]},
            ],
        }
        out, hit = sanitize_anthropic_body(body)
        assert hit is True
        assert "[EMAIL]" in out["messages"][0]["content"][0]["text"]
        # 原 body 不被改(deepcopy)
        assert "@walsin.com" in body["messages"][0]["content"][0]["text"]

    def test_anthropic_body_tool_use_input_v23(self):
        # v2.3 重點修正:tool_use 的 input 也要 sanitize
        body = {
            "messages": [
                {"role": "assistant", "content": [
                    {"type": "tool_use", "id": "x", "name": "lookup",
                     "input": {"email": "alice@example.com"}},
                ]},
            ],
        }
        out, hit = sanitize_anthropic_body(body)
        assert hit is True
        assert out["messages"][0]["content"][0]["input"]["email"] == "[EMAIL]"

    def test_anthropic_body_tool_result_v23(self):
        # v2.3 重點修正:tool_result 的 content 也要 sanitize
        body = {
            "messages": [
                {"role": "user", "content": [
                    {"type": "tool_result", "tool_use_id": "x",
                     "content": "result has alice@example.com"},
                ]},
            ],
        }
        out, hit = sanitize_anthropic_body(body)
        assert hit is True
        assert "[EMAIL]" in out["messages"][0]["content"][0]["content"]


# ============================================================
# Anthropic → OpenAI translation
# ============================================================
class TestRequestTranslation:
    def test_system_string_to_message(self):
        body = {
            "model": "claude-opus-4-7",
            "system": "You are helpful",
            "messages": [{"role": "user", "content": "hi"}],
            "max_tokens": 100,
        }
        out = anthropic_to_openai_request(body, "qwen3:14b")
        assert out["messages"][0] == {"role": "system", "content": "You are helpful"}
        assert out["messages"][1]["role"] == "user"
        assert out["model"] == "qwen3:14b"

    def test_tool_use_to_tool_calls(self):
        body = {
            "messages": [
                {"role": "assistant", "content": [
                    {"type": "text", "text": "Looking up..."},
                    {"type": "tool_use", "id": "tu_1", "name": "lookup",
                     "input": {"q": "x"}},
                ]},
            ],
        }
        out = anthropic_to_openai_request(body, "qwen3:14b")
        msg = out["messages"][0]
        assert "tool_calls" in msg
        assert msg["tool_calls"][0]["id"] == "tu_1"
        assert msg["tool_calls"][0]["function"]["name"] == "lookup"

    def test_tool_result_becomes_separate_message(self):
        body = {
            "messages": [
                {"role": "user", "content": [
                    {"type": "tool_result", "tool_use_id": "tu_1", "content": "the result"},
                ]},
            ],
        }
        out = anthropic_to_openai_request(body, "qwen3:14b")
        # tool result 變成獨立 role:tool message
        tool_msgs = [m for m in out["messages"] if m["role"] == "tool"]
        assert len(tool_msgs) == 1
        assert tool_msgs[0]["tool_call_id"] == "tu_1"


if __name__ == "__main__":
    import pytest
    sys.exit(pytest.main([__file__, "-v"]))
	"""
	Walsin Gateway Benchmark Runner
	================================
	跑一組 prompts 對比多個 model(地端 vs cloud),產出 markdown 報表。

	需要:
	pip install httpx
	ollama pull qwen3:14b qwen3:4b qwen2.5:0.5b (any subset)
	export ANTHROPIC_API_KEY=sk-ant-... (optional, 跑 cloud 對比才需要)

	跑法:
	python3 benchmark_runner.py # 全跑
	python3 benchmark_runner.py --suite swe # 只跑 SWE-bench Lite 子集
	python3 benchmark_runner.py --suite custom # 只跑自家 prompts
	python3 benchmark_runner.py --models qwen3:14b,qwen3:4b
	python3 benchmark_runner.py --output report.md

	輸出:Markdown 報表(每個 model × 每個 prompt 的 latency / output_tokens / 截斷回應)
	本工具不做能力打分,只跑出可比較的數據,讓人類自己判斷。
	"""
	import argparse
	import asyncio
	import json
	import time
	from pathlib import Path

	import httpx

	# ============================================================
	# Test suites
	# ============================================================
	SWE_BENCH_LITE_SUBSET = [
	{
	"id": "swe-1-flask-redirect",
	"category": "bug-fix",
	"prompt": "I have a Flask app where redirect() returns 308 instead of 302. "
	"What's the most likely cause and one-line fix?",
	},
	{
	"id": "swe-2-django-orm",
	"category": "orm",
	"prompt": "Django QuerySet: how to do a LEFT JOIN and filter on the joined table's NULL? "
	"Give the ORM expression, not raw SQL.",
	},
	{
	"id": "swe-3-pytest-async",
	"category": "test",
	"prompt": "pytest-asyncio test 一直 fail with 'async def functions are not natively supported'. "
	"Two minimal fixes (config-based and decorator-based).",
	},
	{
	"id": "swe-4-numpy-broadcast",
	"category": "numerics",
	"prompt": "numpy: I have a (3,) array and (3, 5) array, want elementwise multiply with broadcasting. "
	"Show the .reshape or np.newaxis solution.",
	},
	{
	"id": "swe-5-typing",
	"category": "typing",
	"prompt": "Python TypedDict 怎麼宣告選擇性欄位 + Required 欄位混用? Give 5-line example.",
	},
	]

	CUSTOM_PROMPTS = [
	{
	"id": "custom-1-kafka",
	"category": "concept",
	"prompt": "用一句話說明 Kafka 跟 RabbitMQ 的差別。中文 30 字內。",
	},
	{
	"id": "custom-2-osgi",
	"category": "domain",
	"prompt": "iDempiere OSGi bundle 部署常見的 ClassNotFoundException,3 個最常見的根本原因。",
	},
	{
	"id": "custom-3-sse",
	"category": "protocol",
	"prompt": "Anthropic Messages SSE streaming 共有幾個 event 類型?列出來並說明每個的用途,中文。",
	},
	{
	"id": "custom-4-tool-use",
	"category": "agent",
	"prompt": "解釋 LLM tool use 的 input_json_delta 是什麼?為什麼不是直接送完整 JSON?",
	},
	{
	"id": "custom-5-ollama",
	"category": "ops",
	"prompt": "Ollama 模型第一次 load 很慢,有哪 3 種方法可以加速?",
	},
	]


	# ============================================================
	# Runners
	# ============================================================
	async def run_ollama(model: str, prompt: str, timeout: float = 600.0) -> dict:
	"""跑單一 prompt 到 Ollama。回 {ok, latency_s, output, output_tokens, error}。"""
	url = "http://localhost:11434/v1/chat/completions"
	body = {
	"model": model,
	"messages": [{"role": "user", "content": prompt}],
	"stream": False,
	"max_tokens": 1024,
	}
	t0 = time.time()
	try:
	async with httpx.AsyncClient(timeout=timeout) as client:
	r = await client.post(url, json=body)
	data = r.json()
	out = data["choices"][0]["message"]["content"]
	usage = data.get("usage", {})
	return {
	"ok": True,
	"latency_s": round(time.time() - t0, 2),
	"output": out,
	"output_tokens": usage.get("completion_tokens", 0),
	"error": None,
	}
	except Exception as e:
	return {
	"ok": False,
	"latency_s": round(time.time() - t0, 2),
	"output": "",
	"output_tokens": 0,
	"error": f"{type(e).__name__}: {e}",
	}


	async def run_anthropic(model: str, prompt: str, api_key: str, timeout: float = 300.0) -> dict:
	"""跑單一 prompt 到 Anthropic Messages API。"""
	url = "https://api.anthropic.com/v1/messages"
	headers = {
	"x-api-key": api_key,
	"anthropic-version": "2023-06-01",
	"content-type": "application/json",
	}
	body = {
	"model": model,
	"max_tokens": 1024,
	"messages": [{"role": "user", "content": prompt}],
	}
	t0 = time.time()
	try:
	async with httpx.AsyncClient(timeout=timeout) as client:
	r = await client.post(url, json=body, headers=headers)
	data = r.json()
	out_blocks = data.get("content", [])
	out = "\n".join(b.get("text", "") for b in out_blocks if b.get("type") == "text")
	return {
	"ok": True,
	"latency_s": round(time.time() - t0, 2),
	"output": out,
	"output_tokens": data.get("usage", {}).get("output_tokens", 0),
	"error": None,
	}
	except Exception as e:
	return {
	"ok": False,
	"latency_s": round(time.time() - t0, 2),
	"output": "",
	"output_tokens": 0,
	"error": f"{type(e).__name__}: {e}",
	}


	# ============================================================
	# Main
	# ============================================================
	def truncate(s: str, n: int = 200) -> str:
	s = s.replace("\n", " ").strip()
	return s if len(s) <= n else s[:n] + "..."


	async def main():
	ap = argparse.ArgumentParser()
	ap.add_argument("--suite", choices=["swe", "custom", "all"], default="all")
	ap.add_argument("--models", default="qwen3:14b,qwen3:4b,qwen2.5:0.5b",
	help="comma-separated Ollama models")
	ap.add_argument("--anthropic-models", default="",
	help="comma-separated Anthropic models (e.g. claude-opus-4-7),需要 ANTHROPIC_API_KEY")
	ap.add_argument("--output", default="benchmark_report.md")
	ap.add_argument("--timeout", type=float, default=600.0)
	args = ap.parse_args()

	import os
	api_key = os.environ.get("ANTHROPIC_API_KEY", "")
	anthropic_models = [m.strip() for m in args.anthropic_models.split(",") if m.strip()]
	ollama_models = [m.strip() for m in args.models.split(",") if m.strip()]

	if anthropic_models and not api_key:
	print("⚠️ --anthropic-models 設了但 ANTHROPIC_API_KEY 沒設,Anthropic 部分會跳過")
	anthropic_models = []

	suites = {
	"swe": SWE_BENCH_LITE_SUBSET,
	"custom": CUSTOM_PROMPTS,
	"all": SWE_BENCH_LITE_SUBSET + CUSTOM_PROMPTS,
	}
	prompts = suites[args.suite]

	print(f"Running {len(prompts)} prompts × {len(ollama_models) + len(anthropic_models)} models...")
	print(f"Ollama: {ollama_models}")
	print(f"Anthropic: {anthropic_models or '(none)'}")
	print()

	results = []
	for i, p in enumerate(prompts, 1):
	for m in ollama_models:
	print(f" [{i}/{len(prompts)}] {p['id']} × {m} ...", end="", flush=True)
	r = await run_ollama(m, p["prompt"], args.timeout)
	results.append({"prompt_id": p["id"], "category": p["category"],
	"model": m, "backend": "ollama", **r})
	print(f" {r['latency_s']}s {'OK' if r['ok'] else 'FAIL'}")

	for m in anthropic_models:
	print(f" [{i}/{len(prompts)}] {p['id']} × {m} (cloud) ...", end="", flush=True)
	r = await run_anthropic(m, p["prompt"], api_key, args.timeout)
	results.append({"prompt_id": p["id"], "category": p["category"],
	"model": m, "backend": "anthropic", **r})
	print(f" {r['latency_s']}s {'OK' if r['ok'] else 'FAIL'}")

	# 寫 markdown 報表
	out = ["# Walsin Gateway Benchmark Report", "",
	f"_Generated_: {time.strftime('%Y-%m-%d %H:%M:%S')}", "",
	f"_Prompts_: {len(prompts)} (suite={args.suite})", "",
	f"_Models_: ollama={ollama_models}, anthropic={anthropic_models}", "",
	"## Summary by Model", ""]

	by_model = {}
	for r in results:
	by_model.setdefault(r["model"], []).append(r)

	out.append("\| Model \| Backend \| OK \| Avg latency \| Avg output_tokens \|")
	out.append("\|---\|---\|---\|---\|---\|")
	for m, rs in by_model.items():
	ok = sum(1 for r in rs if r["ok"])
	avg_lat = sum(r["latency_s"] for r in rs) / max(1, len(rs))
	avg_tok = sum(r["output_tokens"] for r in rs) / max(1, len(rs))
	backend = rs[0]["backend"]
	out.append(f"\| {m} \| {backend} \| {ok}/{len(rs)} \| {avg_lat:.1f}s \| {avg_tok:.0f} \|")

	out.append("\n## Detailed Results\n")
	for p in prompts:
	out.append(f"### {p['id']} ({p['category']})\n")
	out.append(f"Prompt: {p['prompt']}\n")
	out.append("\| Model \| Latency \| Tokens \| Output (truncated) \|")
	out.append("\|---\|---\|---\|---\|")
	for r in [r for r in results if r["prompt_id"] == p["id"]]:
	out.append(f"\| {r['model']} \| {r['latency_s']}s \| {r['output_tokens']} \| "
	f"{truncate(r['output'] if r['ok'] else r['error'])} \|")
	out.append("")

	Path(args.output).write_text("\n".join(out))
	print(f"\n✓ Report written: {args.output}")


	if __name__ == "__main__":
	asyncio.run(main())
	#!/bin/bash
	# Walsin Gateway 60-second demo recording script
	# ============================================================
	# 用法:
	# asciinema rec -c "bash demo_record.sh" walsin-demo.cast
	# asciinema upload walsin-demo.cast # (可選)上傳分享
	# ============================================================
	# 不裝 asciinema 也能直接跑(看 terminal output):
	# bash demo_record.sh
	# ============================================================
	# 前提:
	# - Gateway 已啟動: python3 gateway_v2_cc.py &
	# - Ollama 已 pull qwen3:14b 跟 qwen2.5:0.5b(任一)
	# - curl + python3 已裝
	# ============================================================
	set -e

	GW="http://localhost:4000"
	KEY="${MASTER_KEY:-sk-walsin-test}"
	SLEEP_BETWEEN=2

	# v2.3 pre-flight check
	if ! curl -s -o /dev/null --max-time 3 "$GW/health"; then
	printf "\033[31m❌ Gateway 沒在 %s 跑。先啟動:\033[0m\n" "$GW"
	echo " cd ~/walsin-gateway-demo"
	echo " python3 gateway_v2_cc.py &"
	echo ""
	echo " 或設 MASTER_KEY:"
	echo " MASTER_KEY=sk-your-key python3 gateway_v2_cc.py &"
	exit 1
	fi

	cyan() { printf "\033[36m%s\033[0m\n" "$1"; }
	green() { printf "\033[32m%s\033[0m\n" "$1"; }
	yellow(){ printf "\033[33m%s\033[0m\n" "$1"; }
	gray() { printf "\033[90m%s\033[0m\n" "$1"; }

	cyan "=========================================="
	cyan " Walsin Gateway 60s Demo"
	cyan "=========================================="
	echo
	gray "目標:展示 Gateway 看 prompt 自動分級路由"
	echo
	sleep $SLEEP_BETWEEN

	# Step 1: Health check
	yellow "[Step 1] Health check"
	echo '$ curl $GW/health'
	curl -s "$GW/health" \| python3 -m json.tool
	echo
	sleep $SLEEP_BETWEEN

	# Step 2: C 級 prompt
	yellow "[Step 2] C 級 prompt(技術問題,應走 cloud / fallback 地端)"
	echo '$ curl ... -d {"messages":[{"role":"user","content":"What is Kafka?"}]}'
	RESP=$(curl -s -i "$GW/v1/messages" \
	-H "x-api-key: $KEY" \
	-H "content-type: application/json" \
	-H "anthropic-version: 2023-06-01" \
	-d '{
	"model": "claude-opus-4-7",
	"max_tokens": 80,
	"messages": [{"role":"user","content":"用 20 字說 Kafka 是什麼?"}]
	}')

	green " Headers:"
	echo "$RESP" \| grep -i "x-gateway" \| sed 's/^/ /'
	green " Response:"
	echo "$RESP" \| tail -1 \| python3 -c "
	import sys, json
	try:
	r = json.loads(sys.stdin.read())
	if 'content' in r:
	for b in r['content']:
	if b.get('type') == 'text':
	print(' ', b['text'][:200])
	else:
	print(' ', r)
	except Exception as e:
	print(' (parse error:', e, ')')
	"
	echo
	sleep $SLEEP_BETWEEN

	# Step 3: A 級 prompt
	yellow "[Step 3] A 級 prompt(命中字典,強制地端,即使有 cloud key)"
	echo '$ curl ... -d {"messages":[{"role":"user","content":"[client_alpha] 出問題"}]}'
	RESP=$(curl -s -i "$GW/v1/messages" \
	-H "x-api-key: $KEY" \
	-H "content-type: application/json" \
	-H "anthropic-version: 2023-06-01" \
	--max-time 60 \
	-d '{
	"model": "claude-opus-4-7",
	"max_tokens": 80,
	"messages": [{"role":"user","content":"[client_alpha] 出問題了,30 字內怎修?"}]
	}')

	green " Headers:"
	echo "$RESP" \| grep -i "x-gateway" \| sed 's/^/ /'
	green " Response:"
	echo "$RESP" \| tail -1 \| python3 -c "
	import sys, json
	try:
	r = json.loads(sys.stdin.read())
	if 'content' in r:
	for b in r['content']:
	if b.get('type') == 'text':
	print(' ', b['text'][:200])
	else:
	print(' ', r)
	except Exception as e:
	print(' (parse error:', e, ')')
	"
	echo
	sleep $SLEEP_BETWEEN

	# Step 4: Auth fail
	yellow "[Step 4] 沒帶 key 401(v2.1 安全修正)"
	echo '$ curl $GW/v1/messages (no auth header)'
	curl -s -o /dev/null -w " HTTP %{http_code}\n" "$GW/v1/messages" \
	-X POST -H "content-type: application/json" \
	-d '{"model":"x","max_tokens":10,"messages":[{"role":"user","content":"hi"}]}'
	echo
	sleep 1

	cyan "=========================================="
	green " Demo done. 看到 x-gateway-decision 在 header 裡"
	green " A 級永遠地端,C 級可走 cloud,B 級完整 fallback"
	green " 搬離方法論在 Mini PC 上 work。"
	cyan "=========================================="
版本	行數	觸發	結果
v1	80	初版	描述跟 code 矛盾
v2.0	364	review#1	文字自洽,無 code
v2.0 Gist	394	code 上線	1 安全 + 2 功能 bug
v2.1	502	review#2	bug 修完,4 TODO 待補
v2.2	620	TODO close	B 級實作 + tiktoken + benchmark + demo
v2.3	674	self-review	7 個漏洞清完 + 24 個 pytest 全綠
級別	主路由	Fallback	關鍵保護
A 級	地端最強(14b/32b/80B)	沒 fallback	即使有 cloud key 也不走 cloud
B 級	地端優先	地端死 → sanitize → cloud(標 X-Gateway-Sanitized)	不能脫敏寧願 503
C 級	cloud 優先	沒 key → 地端	純技術問題
	"""
	Walsin Gateway 基本 unit test
	============================
	跑法:
	pip install --user pytest pytest-asyncio
	cd ~/walsin-gateway-demo
	python3 -m pytest test_gateway.py -v

	涵蓋 4 個關鍵函式:
	1. classify — A/B/C 分級邏輯
	2. _check_master_key — auth 精確比對(防 substring 漏洞)
	3. _sanitize_value / sanitize_anthropic_body — 遞迴脫敏(含 tool_use)
	4. anthropic_to_openai_request — 格式翻譯
	"""
	import os
	import sys

	# 確保 import gateway 時不會 boot uvicorn(只 import,不 run)
	os.environ.setdefault("MASTER_KEY", "sk-test-secret")

	# 從 gateway_v2_cc 模組 import 要測的函式
	sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
	from gateway_v2_cc import (
	classify,
	_check_master_key,
	_sanitize_value,
	sanitize_anthropic_body,
	anthropic_to_openai_request,
	sanitize_string,
	)


	# ============================================================
	# classify
	# ============================================================
	class TestClassify:
	def test_C_level_default(self):
	msgs = [{"role": "user", "content": "What is Kafka?"}]
	decision, kw = classify(msgs)
	assert decision == "C"
	assert kw is None

	def test_A_level_keyword_match(self):
	msgs = [{"role": "user", "content": "[client_alpha] 出問題了"}]
	decision, kw = classify(msgs)
	assert decision == "A"
	assert "client_alpha" in kw.lower()

	def test_A_level_in_system(self):
	msgs = [{"role": "user", "content": "hello"}]
	decision, kw = classify(msgs, system="你正在處理 [client_alpha] 案子")
	assert decision == "A"

	def test_A_level_in_list_content(self):
	msgs = [{"role": "user", "content": [
	{"type": "text", "text": "hello"},
	{"type": "text", "text": "BOM_FORMULA stuff"},
	]}]
	decision, kw = classify(msgs)
	assert decision == "A"

	def test_B_level_match(self):
	msgs = [{"role": "user", "content": "[internal_process] 是什麼"}]
	decision, kw = classify(msgs)
	assert decision == "B"

	def test_A_takes_precedence_over_B(self):
	# 同時命中 A + B,A 優先
	msgs = [{"role": "user", "content": "[client_alpha] 跟 [internal_process] 都來"}]
	decision, kw = classify(msgs)
	assert decision == "A"


	# ============================================================
	# _check_master_key (v2.3 安全修正)
	# ============================================================
	class TestMasterKey:
	def test_correct_bearer(self):
	assert _check_master_key("Bearer sk-test-secret") is True

	def test_correct_bare(self):
	assert _check_master_key("sk-test-secret") is True

	def test_empty(self):
	assert _check_master_key("") is False

	def test_wrong(self):
	assert _check_master_key("Bearer wrong") is False

	def test_substring_extra_suffix_blocked(self):
	# v2.3 修正點:'sk-test-secret-extra' 不能通過(原 v2.1 substring match 會通過)
	assert _check_master_key("Bearer sk-test-secret-extra") is False

	def test_substring_prefix_blocked(self):
	assert _check_master_key("Bearer prefix-sk-test-secret") is False

	def test_lower_case_bearer(self):
	assert _check_master_key("bearer sk-test-secret") is True


	# ============================================================
	# Sanitization (v2.3 遞迴版)
	# ============================================================
	class TestSanitization:
	def test_string_email(self):
	out, hit = sanitize_string("聯絡 alice@example.com 處理")
	assert "[EMAIL]" in out
	assert "alice@example.com" not in out
	assert hit is True

	def test_string_ip(self):
	out, hit = sanitize_string("server 192.168.1.42 down")
	assert "[IP]" in out
	assert hit is True

	def test_string_no_hit(self):
	out, hit = sanitize_string("just normal text")
	assert out == "just normal text"
	assert hit is False

	def test_recursive_dict(self):
	v, hit = _sanitize_value({"name": "alice@example.com", "ok": 1})
	assert v["name"] == "[EMAIL]"
	assert v["ok"] == 1
	assert hit is True

	def test_recursive_list(self):
	v, hit = _sanitize_value(["alice@example.com", "no hit"])
	assert v[0] == "[EMAIL]"
	assert v[1] == "no hit"
	assert hit is True

	def test_anthropic_body_text_block(self):
	body = {
	"model": "claude-opus-4-7",
	"messages": [
	{"role": "user", "content": [
	{"type": "text", "text": "聯絡 admin@walsin.com"},
	]},
	],
	}
	out, hit = sanitize_anthropic_body(body)
	assert hit is True
	assert "[EMAIL]" in out["messages"][0]["content"][0]["text"]
	# 原 body 不被改(deepcopy)
	assert "@walsin.com" in body["messages"][0]["content"][0]["text"]

	def test_anthropic_body_tool_use_input_v23(self):
	# v2.3 重點修正:tool_use 的 input 也要 sanitize
	body = {
	"messages": [
	{"role": "assistant", "content": [
	{"type": "tool_use", "id": "x", "name": "lookup",
	"input": {"email": "alice@example.com"}},
	]},
	],
	}
	out, hit = sanitize_anthropic_body(body)
	assert hit is True
	assert out["messages"][0]["content"][0]["input"]["email"] == "[EMAIL]"

	def test_anthropic_body_tool_result_v23(self):
	# v2.3 重點修正:tool_result 的 content 也要 sanitize
	body = {
	"messages": [
	{"role": "user", "content": [
	{"type": "tool_result", "tool_use_id": "x",
	"content": "result has alice@example.com"},
	]},
	],
	}
	out, hit = sanitize_anthropic_body(body)
	assert hit is True
	assert "[EMAIL]" in out["messages"][0]["content"][0]["content"]


	# ============================================================
	# Anthropic → OpenAI translation
	# ============================================================
	class TestRequestTranslation:
	def test_system_string_to_message(self):
	body = {
	"model": "claude-opus-4-7",
	"system": "You are helpful",
	"messages": [{"role": "user", "content": "hi"}],
	"max_tokens": 100,
	}
	out = anthropic_to_openai_request(body, "qwen3:14b")
	assert out["messages"][0] == {"role": "system", "content": "You are helpful"}
	assert out["messages"][1]["role"] == "user"
	assert out["model"] == "qwen3:14b"

	def test_tool_use_to_tool_calls(self):
	body = {
	"messages": [
	{"role": "assistant", "content": [
	{"type": "text", "text": "Looking up..."},
	{"type": "tool_use", "id": "tu_1", "name": "lookup",
	"input": {"q": "x"}},
	]},
	],
	}
	out = anthropic_to_openai_request(body, "qwen3:14b")
	msg = out["messages"][0]
	assert "tool_calls" in msg
	assert msg["tool_calls"][0]["id"] == "tu_1"
	assert msg["tool_calls"][0]["function"]["name"] == "lookup"

	def test_tool_result_becomes_separate_message(self):
	body = {
	"messages": [
	{"role": "user", "content": [
	{"type": "tool_result", "tool_use_id": "tu_1", "content": "the result"},
	]},
	],
	}
	out = anthropic_to_openai_request(body, "qwen3:14b")
	# tool result 變成獨立 role:tool message
	tool_msgs = [m for m in out["messages"] if m["role"] == "tool"]
	assert len(tool_msgs) == 1
	assert tool_msgs[0]["tool_call_id"] == "tu_1"


	if __name__ == "__main__":
	import pytest
	sys.exit(pytest.main([__file__, "-v"]))