"""
=============================================================================
CS329Z · 前置讲 P2 共享模块：net_lab.py —— 本机上的一个玩具 API、一个 HTTP 客户端、JSON Schema 校验
=============================================================================

对应讲义：lectures/prereq2_http_apis_protocols.md

【这个文件要回答的问题】
    智能体调用一个 LLM、调用一个外部 API、连接一个 MCP 服务器，底下都是"把一段文字发给某个地址、
    等它回一段文字"。这个文件在**你自己的电脑上**把这件事从零搭一遍，所以每一层都能看见、能改：

    ToyAPI / serve()  ：一个用标准库写的 HTTP 服务器（虚构公司的目录与邮件 API），
                        支持 GET / POST、查询参数、认证、JSON 响应、故意的限速和错误、SSE 流式输出
    raw_request()      ：用**裸 socket** 发一次 HTTP 请求，把请求和响应的原始字节打印出来
    HttpClient         ：一个"能用"的客户端：超时、重试（指数退避 + 尊重 Retry-After）、
                         认证头、用量记账、错误分类
    validate(schema)   ：JSON Schema 的一个子集校验器（够用来校验工具调用的参数）
    parse_sse()        ：解析 Server-Sent Events 流（LLM 的流式输出、MCP 的 HTTP 传输都用它）

【为什么不直接用 requests / httpx？】
    标准库的 urllib 已经够用，而且**少一层魔法**：超时、重试、错误分类都要你自己写一遍，
    写完你就知道 SDK 替你做了什么（第 2 讲会用真正的 SDK）。生产代码请用成熟的库。

【关于"真实的网络"】
    这个文件只连 127.0.0.1（本机）。服务器用端口 0 启动，让操作系统分配一个空闲端口——
    所以不会和你电脑上别的程序冲突，也不需要联网。

【运行】python3 net_lab.py   # 自检：起一个服务器，发几个请求，校验几个参数
【依赖】只用 Python 标准库。
"""

import json
import re
import socket
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass, field
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

API_TOKEN = "demo-token-1234"           # 玩具 API 的"密钥"（真实的密钥不要写进代码，见讲义 §4.1）
PROTOCOL_VERSION = "2025-06-18"         # MCP 的协议版本（第 04、05 号脚本用）


# ============================================================================ 1. 玩具数据
def toy_data():
    """
    虚构公司 Cardinal 的一点数据：5 个员工、12 封邮件。

    【返回】dict：{"employees": [...], "emails": [...]}
    【思路】数据写死在代码里（不是随机生成），所以每次运行、每台机器上完全一样。
    """
    employees = [
        {"id": "e-01", "name": "Ana Reyes", "team": "firmware", "office_days": 3},
        {"id": "e-02", "name": "Bo Chen", "team": "firmware", "office_days": 3},
        {"id": "e-03", "name": "Cy Okafor", "team": "analytics", "office_days": 2},
        {"id": "e-04", "name": "Di Novak", "team": "analytics", "office_days": 2},
        {"id": "e-05", "name": "Ed Brandt", "team": "legal", "office_days": 4},
    ]
    subjects = ["server room temperature alert", "Q3 roadmap review", "parental leave policy update",
                "invoice 8812 overdue", "all-hands slides", "firmware spec v4 draft",
                "office move timeline", "expense report rejected", "security training reminder",
                "Thistledown site visit", "laptop refresh program", "on-call rotation swap"]
    emails = []
    for i, subj in enumerate(subjects):
        emails.append({"id": f"em-{i:03d}", "from": employees[i % 5]["id"],
                       "date": f"2026-06-{10 + i % 5:02d}", "subject": subj,
                       "urgent": i % 4 == 0})
    return {"employees": employees, "emails": emails}


# ============================================================================ 2. 一个玩具 HTTP 服务器
class ToyAPIHandler(BaseHTTPRequestHandler):
    """
    处理一个 HTTP 请求。每个请求都会调用这个类的 do_GET / do_POST 方法。

    【路由】
        GET  /health                     不需要认证，返回 {"ok": true}
        GET  /employees?team=...&limit=N 需要认证，返回员工列表
        GET  /emails?urgent=true&limit=N 需要认证，返回邮件列表
        POST /digest                     需要认证，请求体是 JSON {"date": "...", "max_items": N}
        GET  /slow?seconds=0.3           需要认证，故意慢（演示超时）
        GET  /flaky?fail_times=2         需要认证，前 N 次返回 503（演示重试）
        GET  /limited                    需要认证，第 2 次之后返回 429 + Retry-After（演示限速）
        GET  /stream?n=5                 需要认证，用 SSE 一个词一个词地返回（演示流式输出）
    【认证】除 /health 外都要求 `Authorization: Bearer <API_TOKEN>`，否则 401。
    【Python 小知识】BaseHTTPRequestHandler 把一个 HTTP 请求解析好之后，按方法名调用 do_GET / do_POST；
        self.path 是路径（含查询串），self.headers 是头部，self.rfile / self.wfile 是读写请求体 / 响应体的流。
    """

    server_version = "ToyAPI/1.0"
    protocol_version = "HTTP/1.1"       # 支持 keep-alive，需要每次都写对 Content-Length

    # -------------------------------------------------- 工具方法
    def _send_json(self, obj, status=200, extra_headers=()):
        """把一个 Python 对象序列化成 JSON，连头部一起写回去。"""
        body = json.dumps(obj, ensure_ascii=False, sort_keys=True).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        for k, v in extra_headers:
            self.send_header(k, v)
        self.end_headers()
        self.wfile.write(body)

    def _error(self, status, type_, message):
        """按一个固定的格式返回错误（像真实 API 那样，让客户端能程序化地处理）。"""
        self._send_json({"error": {"type": type_, "message": message}}, status=status)

    def _authorized(self):
        """检查 Authorization 头。【返回】bool"""
        got = self.headers.get("Authorization", "")
        return got == f"Bearer {API_TOKEN}"

    def _query(self):
        """解析查询参数。【返回】(路径, dict)：同名参数只取第一个值"""
        parsed = urllib.parse.urlparse(self.path)
        q = {k: v[0] for k, v in urllib.parse.parse_qs(parsed.query).items()}
        return parsed.path, q

    def _read_body(self):
        """
        读请求体。

        【思路】HTTP 的请求体不会自己"结束"——必须靠 Content-Length 头知道要读多少字节。
        【返回】(obj, error)：解析成功返回 (对象, None)，失败返回 (None, 错误说明)
        """
        n = int(self.headers.get("Content-Length") or 0)
        raw = self.rfile.read(n) if n else b""
        if not raw:
            return None, "request body is empty"
        try:
            return json.loads(raw.decode("utf-8")), None
        except (UnicodeDecodeError, json.JSONDecodeError) as e:
            return None, f"body is not valid JSON: {e}"

    def log_message(self, fmt, *args):
        """默认会把每个请求打印到 stderr；我们把它记在服务器对象上，想看的时候再打印。"""
        self.server.access_log.append(f"{self.command} {self.path}")

    # -------------------------------------------------- GET
    def do_GET(self):
        path, q = self._query()
        data = self.server.data
        if path == "/health":
            return self._send_json({"ok": True})
        if not self._authorized():
            return self._error(401, "authentication_error", "missing or invalid bearer token")
        if path == "/employees":
            rows = data["employees"]
            if "team" in q:
                rows = [r for r in rows if r["team"] == q["team"]]
            return self._send_json({"employees": rows[:int(q.get("limit", 50))], "total": len(rows)})
        if path == "/emails":
            rows = data["emails"]
            if q.get("urgent") == "true":
                rows = [r for r in rows if r["urgent"]]
            return self._send_json({"emails": rows[:int(q.get("limit", 50))], "total": len(rows)})
        if path == "/slow":
            time.sleep(float(q.get("seconds", 0.3)))
            return self._send_json({"slept": float(q.get("seconds", 0.3))})
        if path == "/flaky":
            key = ("flaky", q.get("fail_times", "2"))
            self.server.counters[key] = self.server.counters.get(key, 0) + 1
            if self.server.counters[key] <= int(q.get("fail_times", 2)):
                return self._error(503, "api_error", "service temporarily unavailable")
            return self._send_json({"attempt": self.server.counters[key]})
        if path == "/limited":
            self.server.counters["limited"] = self.server.counters.get("limited", 0) + 1
            if self.server.counters["limited"] > 2:
                body = json.dumps({"error": {"type": "rate_limit_error", "message": "too many requests"}}).encode()
                self.send_response(429)
                self.send_header("Content-Type", "application/json")
                self.send_header("Retry-After", "1")
                self.send_header("Content-Length", str(len(body)))
                self.end_headers()
                return self.wfile.write(body)
            return self._send_json({"request_number": self.server.counters["limited"]})
        if path == "/stream":
            return self._stream(int(q.get("n", 5)))
        return self._error(404, "not_found_error", f"no such path: {path}")

    def _stream(self, n):
        """
        用 Server-Sent Events 把一句话一个词一个词地发出去。

        【思路】SSE 是一种很简单的流式格式：每条消息是几行 `字段: 值`，空行表示一条消息结束。
            服务器不能事先知道总长度，所以用 `Transfer-Encoding: chunked`（分块传输）。
            LLM API 的流式输出（`stream: true`）和 MCP 的 HTTP 传输都用这个格式。
        【参数】n：发几个词
        """
        words = ["Cardinal", "Energy", "has", "five", "employees", "on", "record", "today"]
        self.send_response(200)
        self.send_header("Content-Type", "text/event-stream")
        self.send_header("Cache-Control", "no-cache")
        self.send_header("Transfer-Encoding", "chunked")
        self.end_headers()
        for i in range(min(n, len(words))):
            self._chunk(f"event: delta\ndata: {json.dumps({'index': i, 'text': words[i]})}\n\n")
        self._chunk("event: done\ndata: {}\n\n")
        self._chunk("")                                  # 长度为 0 的块 = 流结束

    def _chunk(self, text):
        """
        写一个 chunked 编码的块：先写一行十六进制的长度，再写内容，再写一个空行。

        【Python 小知识】f"{n:x}" 把整数写成十六进制。
        """
        raw = text.encode("utf-8")
        self.wfile.write(f"{len(raw):x}\r\n".encode() + raw + b"\r\n")
        self.wfile.flush()

    # -------------------------------------------------- POST
    def do_POST(self):
        path, _ = self._query()
        if not self._authorized():
            return self._error(401, "authentication_error", "missing or invalid bearer token")
        if path != "/digest":
            return self._error(404, "not_found_error", f"no such path: {path}")
        body, err = self._read_body()
        if err:
            return self._error(400, "invalid_request_error", err)
        ok, why = validate(DIGEST_SCHEMA, body)
        if not ok:
            return self._error(400, "invalid_request_error", why)
        rows = [e for e in self.server.data["emails"] if e["date"] == body["date"]]
        rows = sorted(rows, key=lambda e: (not e["urgent"], e["id"]))[:body.get("max_items", 3)]
        return self._send_json({"date": body["date"], "items": [e["subject"] for e in rows],
                                "urgent_count": sum(e["urgent"] for e in rows)})


class ToyServer(ThreadingHTTPServer):
    """一个会在后台线程里跑的 HTTP 服务器，外加一点共享状态（数据、计数器、访问日志）。"""

    daemon_threads = True

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.data = toy_data()
        self.counters = {}
        self.access_log = []


class serve:
    """
    在本机起一个玩具 API，用完自动关掉。

    【用法】
        with serve() as base:          # base 形如 "http://127.0.0.1:54321"
            ...
    【思路】端口写 0，让操作系统挑一个空闲端口（避免和别的程序冲突）；服务器跑在后台线程里，
        主线程照常发请求。__exit__ 里关掉它。
    【Python 小知识】定义了 __enter__ / __exit__ 的类可以用在 with 语句里，离开 with 块时自动清理。
    """

    def __init__(self, host="127.0.0.1"):
        self.httpd = ToyServer((host, 0), ToyAPIHandler)
        self.base = f"http://{host}:{self.httpd.server_address[1]}"
        self.thread = threading.Thread(target=self.httpd.serve_forever, daemon=True)

    def __enter__(self):
        self.thread.start()
        return self.base

    def __exit__(self, *exc):
        self.httpd.shutdown()
        self.httpd.server_close()
        self.thread.join(timeout=2)
        return False


# ============================================================================ 3. 裸 socket：看见 HTTP 的字节
def raw_request(base, raw_text, timeout=3.0, read_bytes=65536):
    """
    用一个 TCP 连接发送**原始的** HTTP 请求文本，返回服务器回的原始字节。

    【思路】HTTP 说到底就是一段有格式的文字：第一行是"方法 路径 版本"，接着是若干行头部，
        一个空行，然后（可选的）请求体。这个函数让你亲手写这段文字，看服务器怎么回。
    【参数】base："http://host:port"；raw_text：请求文本（函数会把 \\n 换成 HTTP 要求的 \\r\\n）；
        timeout：秒；read_bytes：最多读多少字节
    【返回】str（响应的原始文本）
    【边界】只支持 http（不支持 https，那需要 TLS）；服务器不关连接时靠 timeout 收场。
    """
    host, port = urllib.parse.urlparse(base).netloc.split(":")
    text = raw_text.replace("\r\n", "\n").replace("\n", "\r\n")
    with socket.create_connection((host, int(port)), timeout=timeout) as s:
        s.sendall(text.encode("utf-8"))
        chunks = []
        try:
            while len(b"".join(chunks)) < read_bytes:
                b = s.recv(4096)
                if not b:
                    break
                chunks.append(b)
        except socket.timeout:
            pass
    return b"".join(chunks).decode("utf-8", errors="replace")


# ============================================================================ 4. 一个"能用"的 HTTP 客户端
class ApiError(Exception):
    """一次 API 调用失败。【属性】status：HTTP 状态码（网络错误时为 None）；type_：错误类型；retryable：能不能重试"""

    def __init__(self, message, status=None, type_="unknown", retryable=False):
        super().__init__(message)
        self.status = status
        self.type_ = type_
        self.retryable = retryable


RETRYABLE_STATUS = (408, 409, 429, 500, 502, 503, 504, 529)


@dataclass
class Attempt:
    """一次尝试的记录（用来看重试到底发生了什么）。"""
    method: str
    path: str
    status: int = None
    error: str = None
    slept: float = 0.0


@dataclass
class HttpClient:
    """
    一个最小但"能用"的 API 客户端：认证、超时、重试、记账。

    【字段】
        base       ：服务器地址
        token      ：bearer token（放在 Authorization 头里）
        timeout    ：每次尝试最多等几秒
        max_retries：最多重试几次（0 = 不重试）
        base_delay ：退避的基准秒数（第 k 次重试等 base_delay × 2^k）
        sleep      ：睡眠函数，默认 time.sleep；测试时可以换成一个"假睡"（只记录时长）
    【用法】c = HttpClient(base, API_TOKEN); c.get("/employees", team="firmware")
    """
    base: str
    token: str = None
    timeout: float = 3.0
    max_retries: int = 0
    base_delay: float = 0.1
    sleep: callable = time.sleep
    attempts: list = field(default_factory=list)
    calls: int = 0
    bytes_in: int = 0

    def request(self, method, path, params=None, body=None, headers=None, raw=False):
        """
        发一次请求（必要时重试）。

        【参数】
            method："GET" / "POST"；path："/employees"；params：查询参数 dict；
            body：要作为 JSON 发送的对象（None = 没有请求体）；headers：额外的头部；
            raw：True 时返回 (status, headers, 原始字节)，不解析 JSON
        【返回】解析后的 JSON 对象（raw=True 时见上）
        【边界】
            - 4xx（除 408 / 409 / 429）不重试：请求本身有问题，重试多少次都一样
            - 429 / 5xx / 网络错误会重试；429 带 Retry-After 时按它说的等
            - 重试次数用完后抛 ApiError
        【工程联系】这三件事（超时、重试、尊重 Retry-After）是任何 LLM / 工具 API 客户端的标配；
            官方 SDK 默认就会重试连接错误、408、409、429 和 5xx（见讲义 §5.3）。
        """
        url = self.base + path
        if params:
            url += "?" + urllib.parse.urlencode({k: v for k, v in params.items() if v is not None})
        data = None if body is None else json.dumps(body).encode("utf-8")
        hdrs = {"Accept": "application/json"}
        if self.token:
            hdrs["Authorization"] = f"Bearer {self.token}"
        if data is not None:
            hdrs["Content-Type"] = "application/json"
        hdrs.update(headers or {})

        last = None
        for k in range(self.max_retries + 1):
            rec = Attempt(method, path)
            self.attempts.append(rec)
            self.calls += 1
            req = urllib.request.Request(url, data=data, headers=hdrs, method=method)
            try:
                with urllib.request.urlopen(req, timeout=self.timeout) as resp:
                    payload = resp.read()
                    rec.status = resp.status
                    self.bytes_in += len(payload)
                    if raw:
                        return resp.status, dict(resp.headers), payload
                    return json.loads(payload.decode("utf-8")) if payload else None
            except urllib.error.HTTPError as e:              # 服务器回了 4xx / 5xx
                payload = e.read()
                rec.status = e.code
                detail = _error_detail(payload)
                last = ApiError(f"HTTP {e.code}: {detail}", e.code, _error_type(payload),
                                e.code in RETRYABLE_STATUS)
                rec.error = f"{e.code} {detail}"[:60]
                retry_after = e.headers.get("Retry-After")
            except (urllib.error.URLError, socket.timeout, TimeoutError) as e:   # 连不上 / 超时
                # urlopen 超时在 Python 里表现为 URLError 包着一个 timeout，所以要往里看一层
                inner = getattr(e, "reason", e)
                is_timeout = isinstance(e, (socket.timeout, TimeoutError)) or \
                    isinstance(inner, (socket.timeout, TimeoutError))
                kind = "timeout_error" if is_timeout else "connection_error"
                last = ApiError(f"{kind}: {inner}", None, kind, True)
                rec.error = f"{kind}: {inner}"[:60]
                retry_after = None
            if not last.retryable or k == self.max_retries:
                raise last
            delay = float(retry_after) if retry_after else self.base_delay * (2 ** k)
            rec.slept = delay
            self.sleep(delay)
        raise last                                           # 理论上到不了这里

    def stream(self, path, params=None):
        """
        发一个请求，**边收边出** SSE 事件（生成器）。

        【思路】不等整个响应收完就处理：每读到一个空行，就说明一条消息完整了，立刻交出去。
            这正是 LLM 流式输出的价值——第一个字来得早，用户不用干等。
        【参数】path；params：查询参数
        【返回】生成器，逐个产出 dict（形如 {"event": "delta", "data": "..."}）
        【边界】服务器不发空行（例如一直在"思考"）时会一直等到 timeout；真实客户端还要处理
            心跳注释行和断线重连（讲义 §6.3）。
        【Python 小知识】函数里出现 yield，它就变成一个"生成器"：调用它不会立刻执行，
            而是每次 for 循环取一个值时才继续往下跑。
        """
        url = self.base + path + ("?" + urllib.parse.urlencode(params) if params else "")
        hdrs = {"Accept": "text/event-stream"}
        if self.token:
            hdrs["Authorization"] = f"Bearer {self.token}"
        req = urllib.request.Request(url, headers=hdrs, method="GET")
        self.calls += 1
        buf = ""
        with urllib.request.urlopen(req, timeout=self.timeout) as resp:
            for raw_line in resp:                        # 逐行读：库已经解掉了分块编码
                line = raw_line.decode("utf-8")
                self.bytes_in += len(raw_line)
                buf += line
                if buf.endswith("\n\n") or buf.endswith("\r\n\r\n"):
                    for ev in parse_sse(buf):
                        yield ev
                    buf = ""
        for ev in parse_sse(buf):                        # 流被截断时，把最后一条不完整的也交出去
            yield ev

    def get(self, path, **params):
        """GET 的快捷写法：c.get("/emails", urgent="true", limit=3)"""
        return self.request("GET", path, params=params)

    def post(self, path, body):
        """POST 的快捷写法：c.post("/digest", {"date": "2026-06-11"})"""
        return self.request("POST", path, body=body)


def _error_detail(payload):
    """从错误响应体里取出人能读的说明；不是预期格式时返回原文的前 80 个字符。"""
    try:
        return json.loads(payload.decode("utf-8"))["error"]["message"]
    except Exception:
        return payload.decode("utf-8", errors="replace")[:80]


def _error_type(payload):
    """从错误响应体里取出机器可读的错误类型。"""
    try:
        return json.loads(payload.decode("utf-8"))["error"]["type"]
    except Exception:
        return "unknown"


# ============================================================================ 5. JSON Schema 的一个子集
DIGEST_SCHEMA = {
    "type": "object",
    "properties": {
        "date": {"type": "string", "pattern": r"^\d{4}-\d{2}-\d{2}$"},
        "max_items": {"type": "integer", "minimum": 1, "maximum": 20},
        "teams": {"type": "array", "items": {"type": "string"}},
    },
    "required": ["date"],
    "additionalProperties": False,
}

_TYPES = {"object": dict, "array": list, "string": str, "number": (int, float), "integer": int,
          "boolean": bool, "null": type(None)}


def validate(schema, value, path="$"):
    """
    按一个 JSON Schema（的子集）校验一个值。

    【思路】工具调用的参数是模型"写"出来的，可能缺字段、类型不对、多了字段、超出范围。
        校验器的作用是**在执行之前**挡住它们，并给出一句能让模型改正的说明。
        支持的关键字：type、properties、required、additionalProperties、enum、
        minimum / maximum、minLength / maxLength、pattern、items。
    【参数】schema：schema 字典；value：要校验的值；path：出错时显示的位置（递归时用）
    【返回】(ok: bool, reason: str)。ok 为真时 reason 是 "ok"
    【边界】
        - 不支持 $ref、oneOf/anyOf、format 等：这是教学用的子集。真实项目请用 jsonschema 库
          （原课 HW1 要求用 jsonschema.Draft202012Validator）。
        - **布尔值是整数的子类**（Python 里 isinstance(True, int) 为真），所以 integer 要单独排除 bool。
    """
    t = schema.get("type")
    if t:
        expected = _TYPES[t]
        if t in ("integer", "number") and isinstance(value, bool):
            return False, f"{path}: expected {t}, got boolean"
        if not isinstance(value, expected):
            return False, f"{path}: expected {t}, got {type(value).__name__}"
    if "enum" in schema and value not in schema["enum"]:
        return False, f"{path}: must be one of {schema['enum']}, got {value!r}"
    if isinstance(value, str):
        if "minLength" in schema and len(value) < schema["minLength"]:
            return False, f"{path}: shorter than minLength {schema['minLength']}"
        if "maxLength" in schema and len(value) > schema["maxLength"]:
            return False, f"{path}: longer than maxLength {schema['maxLength']}"
        if "pattern" in schema and not re.search(schema["pattern"], value):
            return False, f"{path}: {value!r} does not match pattern {schema['pattern']}"
    if isinstance(value, (int, float)) and not isinstance(value, bool):
        if "minimum" in schema and value < schema["minimum"]:
            return False, f"{path}: {value} is below minimum {schema['minimum']}"
        if "maximum" in schema and value > schema["maximum"]:
            return False, f"{path}: {value} is above maximum {schema['maximum']}"
    if isinstance(value, dict):
        for key in schema.get("required", []):
            if key not in value:
                return False, f"{path}: missing required property {key!r}"
        props = schema.get("properties", {})
        if schema.get("additionalProperties") is False:
            extra = sorted(set(value) - set(props))
            if extra:
                return False, f"{path}: unexpected properties {extra} (allowed: {sorted(props)})"
        for key, sub in props.items():
            if key in value:
                ok, why = validate(sub, value[key], f"{path}.{key}")
                if not ok:
                    return ok, why
    if isinstance(value, list) and "items" in schema:
        for i, item in enumerate(value):
            ok, why = validate(schema["items"], item, f"{path}[{i}]")
            if not ok:
                return ok, why
    return True, "ok"


# ============================================================================ 6. SSE 解析
def parse_sse(text):
    """
    解析 Server-Sent Events 流。

    【思路】SSE 的格式极简：每条消息由几行 `字段: 值` 组成，**空行**表示一条消息结束。
        常用字段：event（消息类型）、data（内容，可以有多行，用换行拼起来）、id、retry。
        以 ":" 开头的行是注释（常用来做心跳）。
    【参数】text：收到的完整文本（或任意一段，只要在空行处切开）
    【返回】list[dict]：每条消息形如 {"event": "delta", "data": "{...}"}
    【边界】最后一条消息如果没有空行结尾，也会被返回（真实的流可能被截断）。
    """
    events, cur = [], {}
    for line in text.replace("\r\n", "\n").split("\n"):
        if not line.strip():
            if cur:
                events.append(cur)
                cur = {}
            continue
        if line.startswith(":"):
            continue
        field, _, val = line.partition(":")
        val = val[1:] if val.startswith(" ") else val
        cur[field] = cur[field] + "\n" + val if field in cur else val
    if cur:
        events.append(cur)
    return events


def strip_chunked(text):
    """
    把 chunked 传输编码的响应体还原成原文。

    【思路】分块传输的每个块是：一行十六进制长度、CRLF、内容、CRLF；长度为 0 的块表示结束。
    【参数】text：响应体部分的文本（不含 HTTP 头）
    【返回】str
    【边界】不处理块的扩展参数（`;` 后面的东西）；遇到不认识的长度行就停下并返回已解出的内容。
    【什么时候需要它】只在你**自己读 socket**（raw_request）时需要：成熟的 HTTP 库（包括标准库的
        urllib / http.client）会自动帮你解掉分块编码，所以 HttpClient 拿到的已经是原文了。
    """
    out, i = [], 0
    while i < len(text):
        j = text.find("\r\n", i)
        if j < 0:
            break
        try:
            n = int(text[i:j].split(";")[0], 16)
        except ValueError:
            break
        if n == 0:
            break
        out.append(text[j + 2:j + 2 + n])
        i = j + 2 + n + 2
    return "".join(out)


if __name__ == "__main__":
    with serve() as base:
        c = HttpClient(base, API_TOKEN)
        print("GET /health   →", HttpClient(base).get("/health"))
        print("GET /employees?team=firmware →", c.get("/employees", team="firmware")["employees"])
        print("POST /digest  →", c.post("/digest", {"date": "2026-06-11", "max_items": 2}))
        try:
            HttpClient(base).get("/emails")
        except ApiError as e:
            print("没有 token    →", e, "| retryable:", e.retryable)
        words = [json.loads(ev["data"])["text"] for ev in c.stream("/stream", {"n": 3})
                 if ev.get("event") == "delta"]
        print("GET /stream   →", words)
    print("\nJSON Schema 校验：")
    for v in [{"date": "2026-06-11"}, {"date": "June 11"}, {"max_items": 2}, {"date": "2026-06-11", "max_items": 99},
              {"date": "2026-06-11", "limit": 2}]:
        print(f"  {str(v):52} → {validate(DIGEST_SCHEMA, v)}")