#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Cyrene 假后端（测试台）。纯标准库，零第三方依赖。

它的存在理由只有一个：有些客户端行为**只在特定服务端行为下**才验得到，而真后端的行为不可编排、
造场景的代价又高到不可做（例如「WS 只回放最近 50 条、库里却有 120 条」这种洞，用真后端得灌 51 条
消息、每条跑一次模型回合）。这里可以精确、免费、可复现地造出这些场景，并把**客户端到底发了什么**记下来。

它**不是**产品的一部分，也**不是**后端行为的权威来源：所有形状都是照客户端代码解析出来的（每个
字段的出处都写在下面注释与 README 里），不代表真后端真的这么回。

    python fake_backend.py --port 18080

端点一览见 README.md。控制端点一律挂在 `/__control/` 前缀下，不碰 `/api/v1/**`。
"""

import argparse
import base64
import hashlib
import json
import os
import re
import struct
import sys
import threading
import time
import uuid
from datetime import datetime
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, unquote, urlsplit

# 控制台在 Windows 的默认编码（cp936 / cp1252）会把中文和上游帧原样打印时炸掉；
# 这里固定成 UTF-8，让「贴出来的日志」与 out/*.jsonl 是同一份文本。
for _stream in (sys.stdout, sys.stderr):
    try:
        _stream.reconfigure(encoding="utf-8", errors="replace")
    except Exception:
        pass

HERE = os.path.dirname(os.path.abspath(__file__))

MAIN_SESSION_ID = "web"
WS_GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"

MAX_PAGE_SIZE = 200  # 与 MessageRepository.MAX_PAGE_SIZE 一致
DEFAULT_PAGE_SIZE = 50  # 与 MessageRepository.DEFAULT_PAGE_SIZE 一致
TITLE_MAX_CHARS = 80
MESSAGE_CONTENT_MAX_RUNES = 4000

# GET /api/v1/meta/enums 的载荷（逐字取自客户端仓库里「线上抓下来的」夹具：
# app/src/test/kotlin/top/yeij/cyrene/data/api/dto/MetaDtoRealPayloadTest.kt:24，
# 所以这张表的事实来源是客户端仓库，与任何正在运行的实例无关）。
ENUMS_PAYLOAD = {
    "platforms": [
        {"id": "web", "label": "网页"},
        {"id": "kook", "label": "KOOK"},
        {"id": "qq-obv11", "label": "QQ"},
        {"id": "weixin-ilink", "label": "微信"},
        {"id": "telegram", "label": "Telegram"},
        {"id": "webhook", "label": "Webhook"},
    ],
    "moods": [
        {"id": "calm", "label": "平静"},
        {"id": "happy", "label": "开心"},
        {"id": "excited", "label": "兴奋"},
        {"id": "sad", "label": "低落"},
        {"id": "angry", "label": "生气"},
    ],
    "modes": [
        {"id": "chat", "label": "对话"},
        {"id": "code", "label": "代码"},
        {"id": "home", "label": "家居"},
        {"id": "work", "label": "工作"},
        {"id": "docs", "label": "文档"},
        {"id": "voice", "label": "语音"},
        {"id": "vision", "label": "视觉"},
        {"id": "agents", "label": "智能体"},
        {"id": "settings", "label": "设置"},
    ],
    "unknown": "display-raw",
}

# GET /api/v1/personas（UserPersonaDto，PersonaDto.kt:16）。形状取自客户端解码测试的示例载荷
# （PersonaDtoDecodingTest.kt:32），**不是**真后端的真实角色清单。
# 刻意省掉夹具里的 avatar / chatBackground 路径：那些是服务端文件路径，假后端不提供图片，
# 带着它们只会让客户端去要一堆 404 的图。
PERSONAS = [
    {
        "id": "xilian",
        "name": "昔涟",
        "character": "memokeeper",
        "isDefault": True,
        "mode": "chat",
        "memoryNamespace": "xilian",
        "mood": "calm",
    },
    {
        "id": "qiqi",
        "name": "七七",
        "character": "poker",
        "isDefault": False,
        "mode": "code",
        "memoryNamespace": "qiqi",
    },
]

# 会话种子的 createdAt 取自客户端夹具里的真实值（SessionDtoRealPayloadTest.kt:37），
# 排序规则也照它：pinned DESC, type='main' DESC, created_at ASC
SEED_MAIN_CREATED_AT = 1788724031196
SEED_SUB_CREATED_AT = 1791092728237


def now_ms() -> int:
    return int(time.time() * 1000)


def iso_now() -> str:
    return datetime.now().astimezone().isoformat(timespec="milliseconds")


def log(tag: str, message: str) -> None:
    print("[{}] {:<4} {}".format(datetime.now().strftime("%H:%M:%S"), tag, message), flush=True)


def redact(value, keep: int = 8) -> str:
    """凭据类字段只留指纹，绝不把 token 原样写进 out/ 下的文件。"""
    if not value:
        return ""
    digest = hashlib.sha256(str(value).encode("utf-8")).hexdigest()[:keep]
    return "<redacted sha256:{}>".format(digest)


class JsonlLog:
    """一行一条 JSON 的追加日志。每次写都开一次文件，崩了也不丢已写部分。"""

    def __init__(self, path: str):
        self.path = path
        self.lock = threading.Lock()
        os.makedirs(os.path.dirname(path), exist_ok=True)

    def append(self, record: dict) -> None:
        line = json.dumps(record, ensure_ascii=False)
        with self.lock:
            with open(self.path, "a", encoding="utf-8", newline="\n") as handle:
                handle.write(line + "\n")


def envelope(data) -> dict:
    """成功信封：ApiEnvelope.kt:8 —— 成功 `{data}`，失败 `{error:{code,message}}`。"""
    return {"data": data}


def error_envelope(code: str, message: str) -> dict:
    return {"error": {"code": code, "message": message}}


def collapse_whitespace(text: str) -> str:
    return " ".join(text.split())


def truncate60(text: str) -> str:
    """服务端 lastMessage.content：折叠空白 + 截断到 60 字符 + 省略号（SessionDto.kt:83）。"""
    collapsed = collapse_whitespace(text)
    if len(collapsed) <= 60:
        return collapsed
    return collapsed[:60] + "…"


class Session:
    def __init__(self, session_id: str, kind: str, created_at: int, title=None, pinned=False, persona_id=None):
        self.id = session_id
        self.type = kind
        self.created_at = created_at
        self.title = title
        self.pinned = pinned
        self.persona_id = persona_id
        self.read_at = 0
        self.busy_mode = None
        self.messages = []

    def summary(self) -> dict:
        """SessionSummaryDto（SessionDto.kt:16）。title / lastMessage / personaId 缺席即整键省略。"""
        row = {
            "id": self.id,
            "type": self.type,
            "createdAt": self.created_at,
            "pinned": self.pinned,
            "unreadCount": self.unread_count(),
        }
        if self.title is not None:
            row["title"] = self.title
        if self.persona_id:
            row["personaId"] = self.persona_id
        if self.messages:
            last = self.messages[-1]
            row["lastMessage"] = {
                "role": last["role"],
                "content": truncate60(last.get("content", "")),
                "createdAt": last["createdAt"],
            }
        return row

    def unread_count(self) -> int:
        return sum(1 for m in self.messages if m["createdAt"] > self.read_at)


class Store:
    """全部状态都在内存里：不落盘成数据库、不调模型、重启即回到种子状态。"""

    def __init__(self, args):
        self.args = args
        self.lock = threading.RLock()
        self.sessions = {}
        self.message_seq = 0
        self.state_seq = 0
        self.clients = []
        self.upstream_records = []
        self.seen_client_msg_ids = {}
        # 上行落库模式（--hold-replies / POST /__control/hold-replies）：挂起的上行消息一条不落库。
        self.hold_replies = args.hold_replies
        self.held_messages = []
        self.hold_seq = 0
        self.max_upstream_kept = args.keep_records
        self.out_dir = args.out_dir
        self.upstream_log = JsonlLog(os.path.join(args.out_dir, "upstream.jsonl"))
        self.downstream_log = JsonlLog(os.path.join(args.out_dir, "downstream.jsonl"))
        self.http_log = JsonlLog(os.path.join(args.out_dir, "http.jsonl"))
        self.downstream_count = 0
        self.http_count = 0
        self.reset()

    # —— 种子 / 复位 ——

    def reset(self) -> None:
        with self.lock:
            self.sessions = {}
            self.message_seq = 0
            self.state_seq = 0
            self.seen_client_msg_ids = {}
            self.upstream_records = []
            # 挂起的上行消息属于「库里的状态」，一起复位；hold_replies 是**模式**，复位不动它。
            self.held_messages = []
            self.hold_seq = 0
            main = Session(MAIN_SESSION_ID, "main", SEED_MAIN_CREATED_AT)
            sub = Session(
                "web:5eed0000000000000000000000000001",
                "sub",
                SEED_SUB_CREATED_AT,
                pinned=True,
            )
            self.sessions[main.id] = main
            self.sessions[sub.id] = sub

    def ordered_sessions(self):
        with self.lock:
            rows = list(self.sessions.values())
        rows.sort(key=lambda s: (0 if s.pinned else 1, 0 if s.type == "main" else 1, s.created_at))
        return rows

    # —— 消息 ——

    def next_frame_seq(self) -> int:
        with self.lock:
            self.state_seq += 1
            return self.state_seq

    def add_message(
        self,
        session_id: str,
        role: str,
        content: str,
        client_msg_id=None,
        platform=None,
        mode=None,
        mood=None,
        documents=None,
        created_at=None,
        message_id=None,
    ) -> dict:
        """HistoryMessageDto（MessageDto.kt:13）。可选键为空时**整键省略**，与后端 omitempty 一致。

        [message_id] 非空 = 用挂起时刻先占好的号（见 [hold_message]），不再动 `message_seq`。
        """
        with self.lock:
            session = self.sessions.get(session_id)
            if session is None:
                raise KeyError(session_id)
            if message_id is None:
                self.message_seq += 1
                message_id = "msg-{:06d}".format(self.message_seq)
            previous = session.messages[-1]["createdAt"] if session.messages else 0
            stamp = created_at if created_at is not None else max(now_ms(), previous + 1, session.created_at + 1)
            message = {
                "id": message_id,
                "sessionId": session_id,
                "role": role,
                "content": content,
                "createdAt": stamp,
            }
            if client_msg_id:
                message["clientMsgId"] = client_msg_id
            if platform:
                message["platform"] = platform
            if mode:
                message["mode"] = mode
            if mood:
                message["mood"] = mood
            if documents:
                message["documents"] = documents
            session.messages.append(message)
            return message

    def hold_message(self, session_id: str, role: str, content: str, client_msg_id=None, mode=None) -> dict:
        """上行消息**挂起**：不落库、不广播 `message_persisted`（--hold-replies 打开时走这条）。

        为什么要这个模式：默认行为下客户端一发 `chat`，一个往返就收到那条用户消息的
        `message_persisted`，乐观气泡立刻被收走 ⇒ 「服务端把这条**拒掉**」这类场景根本凑不出来
        （手里没有还挂着的待发气泡）。真后端忙时也是先排队 / 先拒绝，不会立刻落库。

        id 在这里**先占号**（与真正落库共用同一个 `message_seq`，不会撞号），`createdAt` 留到
        真正落库那一刻再盖（[persist_held_message]）。幂等键**不在**这一刻认领 —— 与契约一致：
        队列满的判定发生在认领之前（WsEvent.kt:398），所以挂起期间重发同一帧会得到第二条挂起
        记录，而不是 `chat_duplicate`。
        """
        with self.lock:
            self.message_seq += 1
            self.hold_seq += 1
            record = {
                "holdId": "hold-{:06d}".format(self.hold_seq),
                "sessionId": session_id,
                "messageId": "msg-{:06d}".format(self.message_seq),
                "role": role,
                "clientMsgId": client_msg_id,
                "mode": mode,
                "content": content,
                "heldAt": iso_now(),
            }
            self.held_messages.append(record)
            return record

    def persist_held_message(self, hold: dict) -> dict:
        """把一条挂起的上行消息**显式**落库（控制端点 `POST /__control/persist`）。

        落库动作与默认路径逐字一致：进同一个会话、认领幂等键、由调用方广播 `message_persisted`。
        """
        with self.lock:
            message = self.add_message(
                hold["sessionId"],
                hold["role"],
                hold["content"],
                client_msg_id=hold.get("clientMsgId"),
                mode=hold.get("mode"),
                message_id=hold["messageId"],
            )
            if hold.get("clientMsgId"):
                self.seen_client_msg_ids[hold["clientMsgId"]] = message["id"]
            self.held_messages.remove(hold)
            return message

    def held_for(self, session_id=None) -> list:
        with self.lock:
            return [h for h in self.held_messages if session_id is None or h["sessionId"] == session_id]

    def history_replay(self, session_id: str, limit: int) -> list:
        """握手 history 帧的回放：**最近 N 条、升序**。N 与库里的条数之差就是那个洞。"""
        with self.lock:
            session = self.sessions.get(session_id)
            if session is None:
                return []
            return [dict(m) for m in session.messages[-limit:]] if limit > 0 else []

    def find_message(self, message_id: str):
        with self.lock:
            for session in self.sessions.values():
                for message in session.messages:
                    if message["id"] == message_id:
                        return session, message
        return None, None

    # —— 客户端连接 ——

    def add_client(self, client) -> None:
        with self.lock:
            self.clients.append(client)

    def remove_client(self, client) -> None:
        with self.lock:
            if client in self.clients:
                self.clients.remove(client)

    def broadcast(self, frame: dict, origin: str) -> int:
        with self.lock:
            targets = list(self.clients)
        sent = 0
        for client in targets:
            if client.send_frame(frame, origin):
                sent += 1
        return sent

    # —— 记录 ——

    def record_upstream(self, client_label: str, raw: str, frame, parse_error=None) -> dict:
        record = {
            "ts": iso_now(),
            "epochMs": now_ms(),
            "client": client_label,
            "type": (frame or {}).get("type") if isinstance(frame, dict) else None,
            "sessionId": None,
            "raw": raw,
        }
        if isinstance(frame, dict):
            data = frame.get("data")
            if isinstance(data, dict):
                record["sessionId"] = data.get("sessionId")
            record["frame"] = frame
        if parse_error is not None:
            record["parseError"] = parse_error
        self.upstream_log.append(record)
        with self.lock:
            self.upstream_records.append(record)
            if len(self.upstream_records) > self.max_upstream_kept:
                self.upstream_records = self.upstream_records[-self.max_upstream_kept:]
        # 终端上实时可见：这是「客户端到底发没发那一帧」的第一现场。
        log("WS<-", json.dumps(record.get("frame", raw), ensure_ascii=False))
        return record

    def record_downstream(self, client_label: str, origin: str, raw: str) -> None:
        record = {
            "ts": iso_now(),
            "epochMs": now_ms(),
            "to": client_label,
            "origin": origin,
            "raw": raw,
        }
        self.downstream_log.append(record)
        with self.lock:
            self.downstream_count += 1

    def record_http(self, handler_label: str, method: str, path: str, query: dict, headers) -> dict:
        """把收到的请求头**记下来**（鉴权头不校验，但要看得见）。

        `Authorization` 只留指纹：那是客户端当前会话的凭据，即使假后端只看 HTTP 2xx，
        也没有理由把它原文写进一个可能被 commit 的文件。
        """
        recorded_headers = {}
        for key, value in headers.items():
            if key.lower() == "authorization":
                recorded_headers[key] = redact(value)
            else:
                recorded_headers[key] = value
        if "token" in query:
            query = dict(query)
            query["token"] = [redact(v) for v in query["token"]]
        record = {
            "ts": iso_now(),
            "epochMs": now_ms(),
            "client": handler_label,
            "method": method,
            "path": path,
            "query": query,
            "headers": recorded_headers,
        }
        self.http_log.append(record)
        with self.lock:
            self.http_count += 1
        return record

    def state(self) -> dict:
        with self.lock:
            sessions = [
                {
                    "id": s.id,
                    "type": s.type,
                    "title": s.title,
                    "pinned": s.pinned,
                    "messages": len(s.messages),
                    "unreadCount": s.unread_count(),
                    "readAt": s.read_at,
                    "busy": s.busy_mode,
                    "oldestCreatedAt": s.messages[0]["createdAt"] if s.messages else None,
                    "newestCreatedAt": s.messages[-1]["createdAt"] if s.messages else None,
                }
                for s in self.sessions.values()
            ]
            counts = {}
            for record in self.upstream_records:
                key = record.get("type") or "<unparsed>"
                counts[key] = counts.get(key, 0) + 1
            return {
                "port": self.args.port,
                "host": self.args.host,
                "historyReplay": self.args.history_replay,
                "streamChunk": self.args.stream_chunk,
                "autoReply": self.args.auto_reply,
                "interruptAck": self.args.interrupt_ack,
                "holdReplies": self.hold_replies,
                "heldMessages": [dict(h) for h in self.held_messages],
                "outDir": self.out_dir,
                "sessions": sessions,
                "wsClients": len(self.clients),
                "wsUpstreamFrames": len(self.upstream_records),
                "wsUpstreamByType": counts,
                "wsDownstreamFrames": self.downstream_count,
                "httpRequests": self.http_count,
            }


class WsConnection:
    """手写的 WebSocket 连接（RFC 6455 的最小集：文本帧 + close + ping/pong）。

    只做本工具需要的事：收上行文本帧、下发 JSON 帧。不做扩展、不做压缩、不做分片发送
    （收方向支持分片拼接，因为客户端万一发大包会分片）。
    """

    def __init__(self, store: Store, rfile, wfile, client_label: str):
        self.store = store
        self.rfile = rfile
        self.wfile = wfile
        self.label = client_label
        self.write_lock = threading.Lock()
        self.closed = False

    # —— 发送 ——

    def send_frame(self, frame: dict, origin: str) -> bool:
        text = json.dumps(frame, ensure_ascii=False, separators=(",", ":"))
        return self.send_text(text, origin)

    def send_text(self, text: str, origin: str) -> bool:
        if self.closed:
            return False
        payload = text.encode("utf-8")
        header = bytearray()
        header.append(0x80 | 0x1)  # FIN + text
        length = len(payload)
        if length < 126:
            header.append(length)
        elif length < 65536:
            header.append(126)
            header += struct.pack(">H", length)
        else:
            header.append(127)
            header += struct.pack(">Q", length)
        try:
            with self.write_lock:
                self.wfile.write(bytes(header) + payload)
                self.wfile.flush()
        except (BrokenPipeError, ConnectionResetError, OSError, ValueError):
            self.closed = True
            return False
        self.store.record_downstream(self.label, origin, text)
        return True

    def send_close(self, code: int = 1001, reason: str = "server closing") -> None:
        if self.closed:
            return
        payload = struct.pack(">H", code) + reason.encode("utf-8")
        header = bytearray([0x80 | 0x8, len(payload)])
        try:
            with self.write_lock:
                self.wfile.write(bytes(header) + payload)
                self.wfile.flush()
        except (BrokenPipeError, ConnectionResetError, OSError, ValueError):
            pass
        self.closed = True

    # —— 接收 ——

    def _read_exact(self, count: int):
        try:
            data = self.rfile.read(count)
        except (ConnectionResetError, OSError, ValueError):
            return None
        if data is None or len(data) < count:
            return None
        return data

    def run(self) -> None:
        fragment = bytearray()
        fragment_opcode = None
        while not self.closed:
            head = self._read_exact(2)
            if head is None:
                break
            fin = bool(head[0] & 0x80)
            opcode = head[0] & 0x0F
            masked = bool(head[1] & 0x80)
            length = head[1] & 0x7F
            if length == 126:
                extra = self._read_exact(2)
                if extra is None:
                    break
                length = struct.unpack(">H", extra)[0]
            elif length == 127:
                extra = self._read_exact(8)
                if extra is None:
                    break
                length = struct.unpack(">Q", extra)[0]
            mask = self._read_exact(4) if masked else None
            if masked and mask is None:
                break
            payload = self._read_exact(length) if length else b""
            if payload is None:
                break
            if mask:
                payload = bytes(byte ^ mask[i % 4] for i, byte in enumerate(payload))

            if opcode == 0x8:  # close
                self.send_close(1000, "ack")
                break
            if opcode == 0x9:  # ping → pong（OkHttp pingInterval 30s 靠它活着）
                self._send_control(0xA, payload)
                continue
            if opcode == 0xA:  # pong
                continue

            if opcode == 0x0:  # continuation
                fragment += payload
                if fragment_opcode is None:
                    continue
                if not fin:
                    continue
                self._on_text(bytes(fragment), fragment_opcode)
                fragment = bytearray()
                fragment_opcode = None
                continue

            if not fin:
                fragment = bytearray(payload)
                fragment_opcode = opcode
                continue

            if opcode == 0x1:
                self._on_text(bytes(payload), opcode)
            # 二进制帧（0x2）本工具不处理：客户端不会发。

        self.closed = True

    def _send_control(self, opcode: int, payload: bytes) -> None:
        header = bytearray([0x80 | opcode])
        length = len(payload)
        if length < 126:
            header.append(length)
        elif length < 65536:
            header.append(126)
            header += struct.pack(">H", length)
        else:
            header.append(127)
            header += struct.pack(">Q", length)
        try:
            with self.write_lock:
                self.wfile.write(bytes(header) + payload)
                self.wfile.flush()
        except (BrokenPipeError, ConnectionResetError, OSError, ValueError):
            self.closed = True

    def _on_text(self, payload: bytes, opcode: int) -> None:
        try:
            text = payload.decode("utf-8")
        except UnicodeDecodeError:
            self.store.record_upstream(self.label, payload.hex(), None, "invalid utf-8")
            return
        try:
            frame = json.loads(text)
        except json.JSONDecodeError as exc:
            self.store.record_upstream(self.label, text, None, "bad json: {}".format(exc))
            return
        self.store.record_upstream(self.label, text, frame if isinstance(frame, dict) else None)
        handle_upstream(self.store, self, frame)


# ============ 上行帧的处置 ============

def handle_upstream(store: Store, client, frame) -> None:
    if not isinstance(frame, dict):
        return
    kind = frame.get("type")
    data = frame.get("data") if isinstance(frame.get("data"), dict) else {}

    if kind == "chat":
        session_id = data.get("sessionId") or MAIN_SESSION_ID
        client_msg_id = data.get("clientMsgId")
        if client_msg_id and client_msg_id in store.seen_client_msg_ids:
            # 幂等回执（WsEvent.kt:328 chat_duplicate）：不落库、不起回合。
            client.send_frame(
                {
                    "type": "chat_duplicate",
                    "sessionId": session_id,
                    "data": {
                        "sessionId": session_id,
                        "clientMsgId": client_msg_id,
                        "messageId": store.seen_client_msg_ids[client_msg_id],
                    },
                },
                "chat_duplicate",
            )
            return
        if session_id not in store.sessions:
            client.send_frame(
                {
                    "type": "error",
                    "data": {
                        "code": "bad_request",
                        "message": "unknown session {}".format(session_id),
                        "sessionId": session_id,
                    },
                },
                "error",
            )
            return
        try:
            if store.hold_replies:
                # 挂起：**不**落库、**不**广播 message_persisted（这一处是默认路径里的第一个广播点）。
                # 上行帧已经被 record_upstream 记过（out/upstream.jsonl + wsUpstreamByType），
                # 回复路径照旧不受影响。
                hold = store.hold_message(session_id, "user", data.get("text", ""), client_msg_id=client_msg_id)
                log(
                    "hold",
                    "chat 挂起 {} clientMsgId={}（等 /__control/persist）".format(hold["holdId"], client_msg_id),
                )
                if store.args.auto_reply:
                    send_reply_round(store, session_id, "收到：{}".format(data.get("text", "")[:20]))
                return
            message = store.add_message(
                session_id,
                "user",
                data.get("text", ""),
                client_msg_id=client_msg_id,
            )
        except KeyError:
            return
        if client_msg_id:
            store.seen_client_msg_ids[client_msg_id] = message["id"]
        store.broadcast(
            {"type": "message_persisted", "sessionId": session_id, "data": {"message": message}},
            "message_persisted",
        )
        if store.args.auto_reply:
            send_reply_round(store, session_id, "收到：{}".format(data.get("text", "")[:20]))
        return

    if kind == "interrupt":
        session_id = data.get("sessionId") or MAIN_SESSION_ID
        if store.args.interrupt_ack:
            with store.lock:
                session = store.sessions.get(session_id)
                if session is not None:
                    session.busy_mode = None
            store.broadcast(
                {
                    "type": "session_interrupted",
                    "sessionId": session_id,
                    "data": {"sessionId": session_id, "mode": "idle", "dropped": 0},
                },
                "interrupt_ack",
            )
        return
    # 其余上行帧（voice_input / mode_switch / subagent_ask / task_output_query）只记录，不处置。


def split_chunks(text: str, count: int):
    """把整段正文切成至多 [count] 段（尽量等长；文本比 count 短时就按字符切）。"""
    count = max(1, min(count, len(text)))
    if count <= 1:
        return [text]
    size, extra = divmod(len(text), count)
    parts = []
    index = 0
    for position in range(count):
        step = size + (1 if position < extra else 0)
        parts.append(text[index : index + step])
        index += step
    return [part for part in parts if part]


def send_reply_round(store: Store, session_id: str, text: str, chunks: int = 3, mode: str = "default", persist: bool = True):
    """下发一轮助手回复。

    [mode] `chunk` 走**旧形状** `stream_chunk` + `stream_end`（WsEvent.kt:206），
    用来验客户端那条回退分支（ChatViewModel.kt:1468）；`structured` 走 message.start/delta/end。
    `default` 由 `--stream-chunk` 决定。
    """
    if mode == "default":
        mode = "chunk" if store.args.stream_chunk else "structured"
    turn_id = "turn-{:012x}".format(uuid.uuid4().int & 0xFFFFFFFFFFFF)
    seq_base = store.next_frame_seq()
    deltas = split_chunks(text, chunks)

    if mode == "chunk":
        for offset, delta in enumerate(deltas):
            store.broadcast(
                {
                    "type": "stream_chunk",
                    "turnId": turn_id,
                    "seq": seq_base + offset,
                    "sessionId": session_id,
                    "data": {"delta": delta},
                },
                "reply:stream_chunk",
            )
    else:
        store.broadcast(
            {
                "type": "message.start",
                "turnId": turn_id,
                "seq": seq_base,
                "sessionId": session_id,
                "data": {"index": 0},
            },
            "reply:message.start",
        )
        for offset, delta in enumerate(deltas):
            store.broadcast(
                {
                    "type": "message.delta",
                    "turnId": turn_id,
                    "seq": seq_base + offset + 1,
                    "sessionId": session_id,
                    "data": {"delta": delta, "index": 0},
                },
                "reply:message.delta",
            )
        store.broadcast(
            {
                "type": "message.end",
                "turnId": turn_id,
                "seq": seq_base + len(deltas) + 1,
                "sessionId": session_id,
                "data": {"index": 0},
            },
            "reply:message.end",
        )

    store.broadcast(
        {
            "type": "stream_end",
            "turnId": turn_id,
            "seq": seq_base + len(deltas) + 2,
            "sessionId": session_id,
            "data": {"stopReason": "stop", "steps": 1},
        },
        "reply:stream_end",
    )

    if persist:
        try:
            message = store.add_message(session_id, "assistant", text, mode="chat")
        except KeyError:
            return {"turnId": turn_id, "mode": mode, "chunks": len(deltas), "persisted": False}
        store.broadcast(
            {"type": "message_persisted", "sessionId": session_id, "data": {"message": message}},
            "message_persisted",
        )
        return {"turnId": turn_id, "mode": mode, "chunks": len(deltas), "persisted": True, "messageId": message["id"]}
    return {"turnId": turn_id, "mode": mode, "chunks": len(deltas), "persisted": False}


# ============ HTTP ============

MESSAGES_RE = re.compile(r"^/api/v1/sessions/([^/]+)/messages$")
SESSION_RE = re.compile(r"^/api/v1/sessions/([^/]+)$")
SESSION_READ_RE = re.compile(r"^/api/v1/sessions/([^/]+)/read$")
MESSAGE_RE = re.compile(r"^/api/v1/messages/([^/]+)$")


class Handler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"
    server_version = "CyreneFakeBackend"
    store: Store = None

    # —— 基础设施 ——

    def log_message(self, fmt, *args):
        return

    def _label(self) -> str:
        host, port = self.client_address[0], self.client_address[1]
        return "{}:{}".format(host, port)

    def _split(self):
        parts = urlsplit(self.path)
        query = parse_qs(parts.query, keep_blank_values=True)
        return parts.path, query

    def _read_body(self):
        length = self.headers.get("Content-Length")
        if not length:
            return b""
        try:
            size = int(length)
        except ValueError:
            return b""
        if size <= 0:
            return b""
        return self.rfile.read(size) or b""

    def _parse_json_body(self):
        raw = self._read_body()
        if not raw:
            return None, None
        try:
            return json.loads(raw.decode("utf-8")), raw
        except (json.JSONDecodeError, UnicodeDecodeError) as exc:
            return None, str(exc)

    def _send_json(self, status: int, payload) -> None:
        body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
        try:
            self.send_response(status)
            self.send_header("Content-Type", "application/json; charset=utf-8")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)
        except (BrokenPipeError, ConnectionResetError, OSError, ValueError):
            self.close_connection = True

    def _ok(self, data, status: int = 200) -> None:
        self._send_json(status, envelope(data))

    def _fail(self, status: int, code: str, message: str) -> None:
        self._send_json(status, error_envelope(code, message))

    # —— 方法入口 ——

    def do_GET(self):
        self._handle("GET")

    def do_POST(self):
        self._handle("POST")

    def do_PATCH(self):
        self._handle("PATCH")

    def do_DELETE(self):
        self._handle("DELETE")

    def do_OPTIONS(self):
        self._handle("OPTIONS")

    def _handle(self, method: str):
        path, query = self._split()
        self.store.record_http(self._label(), method, path, query, self.headers)
        try:
            if path == "/ws/chat":
                self._handle_ws(method, query)
                return
            if path.startswith("/__control/"):
                self._handle_control(method, path, query)
                return
            if path.startswith("/api/v1/"):
                self._handle_api(method, path, query)
                return
            if path == "/":
                self._ok({"name": "cyrene-fake-backend", "control": "/__control/state"})
                return
            self._fail(404, "not_found", "no such endpoint: {}".format(path))
        except BrokenPipeError:
            self.close_connection = True
        except Exception as exc:  # 假后端绝不该把线程炸掉；把原因如实回给调用方
            log("ERR ", "{} {} -> {}".format(method, path, exc))
            try:
                self._fail(500, "fake_backend_error", repr(exc))
            except Exception:
                self.close_connection = True

    # —— WebSocket 握手 ——

    def _handle_ws(self, method: str, query) -> None:
        if method != "GET":
            self._fail(405, "bad_request", "websocket handshake must be GET")
            return
        upgrade = (self.headers.get("Upgrade") or "").lower()
        key = self.headers.get("Sec-WebSocket-Key")
        version = self.headers.get("Sec-WebSocket-Version")
        if "websocket" not in upgrade or not key:
            self._fail(400, "bad_request", "not a websocket handshake")
            return
        if version not in (None, "13"):
            self._fail(400, "bad_request", "unsupported Sec-WebSocket-Version {}".format(version))
            return

        accept = base64.b64encode(hashlib.sha1((key + WS_GUID).encode("ascii")).digest()).decode("ascii")
        self.send_response(101)
        self.send_header("Upgrade", "websocket")
        self.send_header("Connection", "Upgrade")
        self.send_header("Sec-WebSocket-Accept", accept)
        self.end_headers()
        self.wfile.flush()

        token = (query.get("token") or [""])[0]
        client = WsConnection(self.store, self.rfile, self.wfile, self._label())
        self.store.add_client(client)
        log("WS  ", "handshake 101 client={} token={}".format(self._label(), redact(token) or "<none>"))
        try:
            # 握手后立刻发 history，回放条数由 --history-replay 决定（这就是那个洞）。
            replay = self.store.history_replay(MAIN_SESSION_ID, self.store.args.history_replay)
            client.send_frame(
                {
                    "type": "history",
                    "sessionId": MAIN_SESSION_ID,
                    "seq": self.store.next_frame_seq(),
                    "data": {"messages": replay},
                },
                "history",
            )
            log(
                "WS->",
                "history replayed={} (historyReplay={})".format(len(replay), self.store.args.history_replay),
            )
            client.run()
        finally:
            self.store.remove_client(client)
            self.close_connection = True
            log("WS  ", "closed client={}".format(self._label()))

    # —— /api/v1/** ——

    def _handle_api(self, method: str, path: str, query) -> None:
        # 鉴权：**一律不校验**（README 里明说）。token / X-API-Version 照收不误，已经记进 http.jsonl。
        if path == "/api/v1/auth/login" and method == "POST":
            body, raw = self._parse_json_body()
            self._ok(
                {
                    "token": "fake-token-{}".format(uuid.uuid4().hex[:16]),
                    "expiresAt": int(time.time()) + 86400,  # Unix **秒**（LoginDto.kt:14）
                    "role": "owner",
                    "refreshToken": "fake-refresh-{}".format(uuid.uuid4().hex[:16]),
                }
            )
            return
        if path == "/api/v1/auth/refresh" and method == "POST":
            self._ok(
                {
                    "token": "fake-token-{}".format(uuid.uuid4().hex[:16]),
                    "expiresAt": int(time.time()) + 86400,
                    "role": "owner",
                    "refreshToken": "fake-refresh-{}".format(uuid.uuid4().hex[:16]),
                }
            )
            return
        if path == "/api/v1/auth/logout" and method == "POST":
            self._parse_json_body()
            self._ok({"revoked": True, "revokedCount": 1})
            return
        if path == "/api/v1/meta/enums" and method == "GET":
            self._ok(ENUMS_PAYLOAD)
            return
        if path == "/api/v1/personas" and method == "GET":
            self._ok(PERSONAS)
            return
        if path == "/api/v1/sessions":
            if method == "GET":
                self._ok([s.summary() for s in self.store.ordered_sessions()])
                return
            if method == "POST":
                self._create_session()
                return
        if path == "/api/v1/chat" and method == "POST":
            self._rest_chat()
            return

        match = MESSAGES_RE.match(path)
        if match and method == "GET":
            self._list_messages(unquote(match.group(1)), query)
            return

        match = SESSION_READ_RE.match(path)
        if match and method == "POST":
            self._mark_read(unquote(match.group(1)))
            return

        match = SESSION_RE.match(path)
        if match:
            session_id = unquote(match.group(1))
            if method == "PATCH":
                self._update_session(session_id)
                return
            if method == "DELETE":
                self._delete_session(session_id)
                return

        match = MESSAGE_RE.match(path)
        if match:
            message_id = unquote(match.group(1))
            if method == "PATCH":
                self._update_message(message_id)
                return
            if method == "DELETE":
                self._delete_message(message_id)
                return

        self._fail(404, "not_found", "no such endpoint: {} {}".format(method, path))

    def _create_session(self) -> None:
        body, raw = self._parse_json_body()
        if raw is not None and body is None:
            self._fail(400, "bad_request", "invalid json body")
            return
        persona_id = None
        if isinstance(body, dict):
            persona_id = body.get("personaId")
        known = {p["id"] for p in PERSONAS}
        if persona_id is not None and persona_id not in known:
            self._fail(400, "bad_request", "unknown personaId: {}".format(persona_id))
            return
        session_id = "web:{}".format(uuid.uuid4().hex)
        with self.store.lock:
            self.store.sessions[session_id] = Session(
                session_id,
                "sub",
                now_ms(),
                persona_id=persona_id,
            )
        payload = {"sessionId": session_id}
        if persona_id:
            payload["personaId"] = persona_id
        log("http", "created session {} persona={}".format(session_id, persona_id))
        self._ok(payload, status=201)

    def _update_session(self, session_id: str) -> None:
        body, raw = self._parse_json_body()
        if not isinstance(body, dict):
            self._fail(400, "bad_request", "invalid json body")
            return
        if "personaId" in body:
            # 会话角色只在创建时写入，之后不可变（SessionsApi.kt:46）。
            self._fail(400, "bad_request", "personaId is immutable")
            return
        if not body:
            self._fail(400, "bad_request", "title or pinned is required")
            return
        with self.store.lock:
            session = self.store.sessions.get(session_id)
            if session is None:
                self._fail(404, "not_found", "unknown session {}".format(session_id))
                return
            response = {"id": session_id}
            if "title" in body:
                title = body.get("title")
                if not isinstance(title, str) or not title.strip():
                    self._fail(400, "bad_request", "title is required")
                    return
                if len(title.strip()) > TITLE_MAX_CHARS:
                    self._fail(400, "bad_request", "title too long")
                    return
                session.title = title.strip()
                response["title"] = session.title
            if "pinned" in body:
                session.pinned = bool(body.get("pinned"))
                response["pinned"] = session.pinned
        self._ok(response)

    def _delete_session(self, session_id: str) -> None:
        if session_id == MAIN_SESSION_ID:
            self._fail(403, "forbidden", "main session cannot be deleted")
            return
        with self.store.lock:
            if session_id not in self.store.sessions:
                self._fail(404, "not_found", "unknown session {}".format(session_id))
                return
            del self.store.sessions[session_id]
        self._ok({"deleted": session_id})

    def _list_messages(self, session_id: str, query) -> None:
        with self.store.lock:
            session = self.store.sessions.get(session_id)
            if session is None:
                # 契约 v0.63 的会话级 404（ChatViewModel.kt:1808 按这个码判「会话没了」）。
                self._fail(404, "session_not_found", "unknown session {}".format(session_id))
                return
            messages = [dict(m) for m in session.messages]

        limit_raw = (query.get("limit") or [None])[0]
        before_raw = (query.get("before") or [None])[0]
        after_raw = (query.get("after") or [None])[0]
        platform = (query.get("platform") or [None])[0]
        search = (query.get("q") or [None])[0]

        limit = DEFAULT_PAGE_SIZE
        if limit_raw not in (None, ""):
            try:
                limit = int(limit_raw)
            except ValueError:
                self._fail(400, "bad_request", "limit must be an integer")
                return
            limit = max(1, min(limit, MAX_PAGE_SIZE))

        before = self._parse_cursor(before_raw, "before")
        if before is _BAD:
            return
        after = self._parse_cursor(after_raw, "after")
        if after is _BAD:
            return
        if after is not None and before is not None:
            self._fail(400, "bad_request", "after and before are mutually exclusive")
            return
        if after is not None and search:
            self._fail(400, "bad_request", "after and q are mutually exclusive")
            return

        if platform:
            messages = [m for m in messages if m.get("platform") == platform]
        if search:
            needle = search.lower()
            messages = [m for m in messages if needle in (m.get("content") or "").lower()]

        if after is not None:
            # 正向增量：created_at > after，**升序**（契约 §1 行 6）。
            page = [m for m in sorted(messages, key=lambda m: m["createdAt"]) if m["createdAt"] > after][:limit]
        elif before is not None:
            # 倒序分页：created_at < before，**最新在前**。
            page = [m for m in sorted(messages, key=lambda m: m["createdAt"], reverse=True) if m["createdAt"] < before][
                :limit
            ]
        else:
            page = sorted(messages, key=lambda m: m["createdAt"], reverse=True)[:limit]

        self._ok(page)

    def _parse_cursor(self, raw, name: str):
        if raw in (None, ""):
            return None
        try:
            return int(raw)
        except ValueError:
            self._fail(400, "bad_request", "{} must be a unix-millis integer".format(name))
            return _BAD

    def _mark_read(self, session_id: str) -> None:
        body, raw = self._parse_json_body()
        if raw is not None and body is None:
            self._fail(400, "bad_request", "invalid json body")
            return
        before = None
        if isinstance(body, dict) and body.get("before") is not None:
            candidate = body.get("before")
            if not isinstance(candidate, int) or candidate <= 0:
                self._fail(400, "bad_request", "before must be a positive unix-millis integer")
                return
            before = candidate
        with self.store.lock:
            session = self.store.sessions.get(session_id)
            if session is None:
                self._fail(404, "session_not_found", "unknown session {}".format(session_id))
                return
            # 游标只前进（SQL MAX(read_at, before)），重复 / 乱序都幂等安全。
            session.read_at = max(session.read_at, before if before is not None else now_ms())
            read_at = session.read_at
        # 契约 §3.9：成功后广播 session_read，载荷里**只有 readAt**，会话归属在帧壳上。
        self.store.broadcast(
            {"type": "session_read", "sessionId": session_id, "data": {"readAt": read_at}},
            "session_read",
        )
        self._ok({"sessionId": session_id, "readAt": read_at})

    def _update_message(self, message_id: str) -> None:
        body, raw = self._parse_json_body()
        if not isinstance(body, dict):
            self._fail(400, "bad_request", "invalid json body")
            return
        content = body.get("content")
        if not isinstance(content, str) or not content.strip():
            self._fail(400, "bad_request", "content is required")
            return
        if len(content) > MESSAGE_CONTENT_MAX_RUNES:
            self._fail(400, "bad_request", "content too long")
            return
        session, message = self.store.find_message(message_id)
        if message is None:
            self._fail(404, "not_found", "unknown message {}".format(message_id))
            return
        message["content"] = content
        log("http", "edited {} -> {}".format(message_id, json.dumps(content, ensure_ascii=False)))
        # 契约 §3.8：data **就是**编辑后的完整消息 DTO，没有 message 包一层（WsEvent.kt:678）。
        self.store.broadcast(
            {"type": "message_edited", "sessionId": session.id, "data": message},
            "message_edited",
        )
        self._ok(message)

    def _delete_message(self, message_id: str) -> None:
        with self.store.lock:
            for session in self.store.sessions.values():
                for index, message in enumerate(session.messages):
                    if message["id"] == message_id:
                        del session.messages[index]
                        self.store.broadcast(
                            {"type": "message_deleted", "sessionId": session.id, "data": {"id": message_id}},
                            "message_deleted",
                        )
                        self._ok({"deleted": message_id})
                        return
        self._fail(404, "not_found", "unknown message {}".format(message_id))

    def _rest_chat(self) -> None:
        """`POST /api/v1/chat`（ChatApi.kt:21）——**额外**支持的写路径，形状照 ChatAckDto。"""
        body, raw = self._parse_json_body()
        if not isinstance(body, dict):
            self._fail(400, "bad_request", "invalid json body")
            return
        session_id = body.get("sessionId") or MAIN_SESSION_ID
        if session_id not in self.store.sessions:
            self._fail(404, "session_not_found", "unknown session {}".format(session_id))
            return
        if self.store.hold_replies:
            # 挂起：回执形状逐字不变（`ChatAckDto`），但那条消息**没有**落库、也**没有**广播
            # message_persisted（这是默认路径里的第二个广播点）。`messageId` 是挂起时先占的号，
            # 它会在 /__control/persist 那一刻真的进库。
            hold = self.store.hold_message(
                session_id,
                "user",
                body.get("text", ""),
                client_msg_id=body.get("clientMsgId"),
                mode=body.get("mode"),
            )
            log(
                "hold",
                "rest chat 挂起 {} messageId={}（等 /__control/persist）".format(hold["holdId"], hold["messageId"]),
            )
            if self.store.args.auto_reply:
                send_reply_round(self.store, session_id, "收到：{}".format(body.get("text", "")[:20]))
            self._ok(
                {
                    "stream": True,
                    "queued": False,
                    "queueDepth": 0,
                    "messageId": hold["messageId"],
                    "sessionId": session_id,
                },
                status=202,
            )
            return
        message = self.store.add_message(
            session_id,
            "user",
            body.get("text", ""),
            client_msg_id=body.get("clientMsgId"),
            mode=body.get("mode"),
        )
        self.store.broadcast(
            {"type": "message_persisted", "sessionId": session_id, "data": {"message": message}},
            "message_persisted",
        )
        if self.store.args.auto_reply:
            send_reply_round(self.store, session_id, "收到：{}".format(body.get("text", "")[:20]))
        self._ok(
            {
                "stream": True,
                "queued": False,
                "queueDepth": 0,
                "messageId": message["id"],
                "sessionId": session_id,
            },
            status=202,
        )

    # —— /__control/** ——

    def _handle_control(self, method: str, path: str, query) -> None:
        action = path[len("/__control/") :].strip("/")

        if action == "state" and method in ("GET", "POST"):
            self._ok(self.store.state())
            return

        if action == "upstream" and method in ("GET", "POST"):
            limit = self._int_param(query, "limit", 20)
            with self.store.lock:
                records = self.store.upstream_records[-limit:]
            self._ok({"count": len(records), "records": records})
            return

        if action == "requests" and method in ("GET", "POST"):
            limit = self._int_param(query, "limit", 20)
            records = []
            if os.path.exists(self.store.http_log.path):
                with open(self.store.http_log.path, "r", encoding="utf-8") as handle:
                    records = [json.loads(line) for line in handle if line.strip()]
            self._ok({"count": len(records), "records": records[-limit:]})
            return

        if action == "reset" and method in ("GET", "POST"):
            self.store.reset()
            log("ctl ", "state reset to seed")
            self._ok({"reset": True})
            return

        if action == "messages" and method in ("GET", "POST"):
            session_id = self._param(query, "session", MAIN_SESSION_ID)
            count = self._int_param(query, "count", 1)
            role = self._param(query, "role", "user")
            prefix = self._param(query, "prefix", "seed")
            template = self._param(query, "text", None)
            platform = self._param(query, "platform", None)
            if session_id not in self.store.sessions:
                self._fail(404, "session_not_found", "unknown session {}".format(session_id))
                return
            first_id = last_id = None
            for index in range(1, count + 1):
                content = template.format(n=index) if template else "{} #{}".format(prefix, index)
                message = self.store.add_message(session_id, role, content, platform=platform)
                first_id = first_id or message["id"]
                last_id = message["id"]
            with self.store.lock:
                session = self.store.sessions[session_id]
                total = len(session.messages)
                oldest = session.messages[0]["createdAt"]
                newest = session.messages[-1]["createdAt"]
            log("ctl ", "seeded {} msgs into {} (total={})".format(count, session_id, total))
            self._ok(
                {
                    "session": session_id,
                    "added": count,
                    "total": total,
                    "firstId": first_id,
                    "lastId": last_id,
                    "oldestCreatedAt": oldest,
                    "newestCreatedAt": newest,
                }
            )
            return

        if action == "message" and method in ("GET", "POST"):
            # 取某一条（1 基下标），用来在脚本里**精确**拿游标：例如「回放里最旧那条的前一条」。
            session_id = self._param(query, "session", MAIN_SESSION_ID)
            index = self._int_param(query, "index", 1)
            with self.store.lock:
                session = self.store.sessions.get(session_id)
                if session is None:
                    self._fail(404, "session_not_found", "unknown session {}".format(session_id))
                    return
                total = len(session.messages)
                if index < 1 or index > total:
                    self._fail(404, "not_found", "index {} out of range (1..{})".format(index, total))
                    return
                message = dict(session.messages[index - 1])
            self._ok({"session": session_id, "index": index, "total": total, "message": message})
            return

        if action == "busy" and method in ("GET", "POST"):
            session_id = self._param(query, "session", MAIN_SESSION_ID)
            mode = self._param(query, "mode", "thinking")
            with self.store.lock:
                session = self.store.sessions.get(session_id)
                if session is None:
                    self._fail(404, "session_not_found", "unknown session {}".format(session_id))
                    return
                session.busy_mode = mode
            sent = self.store.broadcast(
                {"type": "session_busy", "sessionId": session_id, "data": {"sessionId": session_id, "mode": mode}},
                "control:busy",
            )
            self._ok({"session": session_id, "mode": mode, "delivered": sent})
            return

        if action == "idle" and method in ("GET", "POST"):
            session_id = self._param(query, "session", MAIN_SESSION_ID)
            mode = self._param(query, "mode", "idle")
            with self.store.lock:
                session = self.store.sessions.get(session_id)
                if session is None:
                    self._fail(404, "session_not_found", "unknown session {}".format(session_id))
                    return
                session.busy_mode = None
            sent = self.store.broadcast(
                {"type": "session_idle", "sessionId": session_id, "data": {"sessionId": session_id, "mode": mode}},
                "control:idle",
            )
            self._ok({"session": session_id, "mode": mode, "delivered": sent})
            return

        if action == "reply" and method in ("GET", "POST"):
            session_id = self._param(query, "session", MAIN_SESSION_ID)
            text = self._param(query, "text", "这是一条来自假后端的回复。")
            chunks = self._int_param(query, "chunks", 3)
            mode = self._param(query, "frame", "default")
            persist = self._param(query, "persist", "1") not in ("0", "false", "no")
            if mode not in ("default", "chunk", "structured"):
                self._fail(400, "bad_request", "frame must be default|chunk|structured")
                return
            if session_id not in self.store.sessions:
                self._fail(404, "session_not_found", "unknown session {}".format(session_id))
                return
            result = send_reply_round(self.store, session_id, text, chunks=chunks, mode=mode, persist=persist)
            result["delivered"] = len(self.store.clients)
            self._ok(result)
            return

        if action == "interrupted" and method in ("GET", "POST"):
            session_id = self._param(query, "session", MAIN_SESSION_ID)
            mode = self._param(query, "mode", "idle")
            dropped = self._int_param(query, "dropped", 0)
            with self.store.lock:
                session = self.store.sessions.get(session_id)
                if session is None:
                    self._fail(404, "session_not_found", "unknown session {}".format(session_id))
                    return
                session.busy_mode = None
            sent = self.store.broadcast(
                {
                    "type": "session_interrupted",
                    "sessionId": session_id,
                    "data": {"sessionId": session_id, "mode": mode, "dropped": dropped},
                },
                "control:interrupted",
            )
            self._ok({"session": session_id, "mode": mode, "dropped": dropped, "delivered": sent})
            return

        if action == "hold-replies" and method in ("GET", "POST"):
            # 运行期开关（等价于启动参数 --hold-replies），用来在同一个进程里做「先挂着 → 再落库」两段。
            on = self._param(query, "on", "1") not in ("0", "false", "no")
            self.store.hold_replies = on
            with self.store.lock:
                held = len(self.store.held_messages)
            log("ctl ", "hold-replies {} (held={})".format("on" if on else "off", held))
            self._ok({"holdReplies": on, "held": held})
            return

        if action == "persist" and method in ("GET", "POST"):
            # 显式补上挂起那一刻没做的那次落库：进库 + 广播 message_persisted（与默认路径同一形状）。
            session_id = self._param(query, "session", None)
            ident = self._param(query, "id", None)
            index = self._int_param(query, "index", None)
            candidates = self.store.held_for(session_id)
            if ident:
                candidates = [
                    hold
                    for hold in candidates
                    if ident in (hold["holdId"], hold["messageId"], hold.get("clientMsgId"))
                ]
            elif index is not None:
                if index < 1 or index > len(candidates):
                    self._fail(404, "not_found", "index {} out of range (1..{})".format(index, len(candidates)))
                    return
                candidates = [candidates[index - 1]]
            else:
                candidates = candidates[:1]  # 不给选择器就按 FIFO 落最旧的那条
            if not candidates:
                self._fail(
                    404,
                    "no_held_message",
                    "no held message matches (hold-replies={}, held={}); see GET /__control/state".format(
                        self.store.hold_replies, len(self.store.held_messages)
                    ),
                )
                return
            hold = candidates[0]
            message = self.store.persist_held_message(hold)
            sent = self.store.broadcast(
                {"type": "message_persisted", "sessionId": message["sessionId"], "data": {"message": message}},
                "message_persisted",
            )
            log("ctl ", "persisted {} as {} (delivered={})".format(hold["holdId"], message["id"], sent))
            self._ok(
                {
                    "persisted": True,
                    "session": message["sessionId"],
                    "holdId": hold["holdId"],
                    "messageId": message["id"],
                    "clientMsgId": hold.get("clientMsgId"),
                    "delivered": sent,
                    "remaining": len(self.store.held_messages),
                    "message": message,
                }
            )
            return

        if action == "pending-full" and method in ("GET", "POST"):
            session_id = self._param(query, "session", MAIN_SESSION_ID)
            mode = self._param(query, "mode", "thinking")
            maximum = self._int_param(query, "max", 5)
            client_msg_id = self._param(query, "clientMsgId", None)
            message_id = self._param(query, "messageId", None)
            data = {"sessionId": session_id, "mode": mode, "max": maximum}
            # 两个可选键**整键缺席**，绝不填空串（WsEvent.kt:798）。
            if client_msg_id:
                data["clientMsgId"] = client_msg_id
            if message_id:
                data["messageId"] = message_id
            sent = self.store.broadcast(
                {"type": "pending_full", "sessionId": session_id, "data": data},
                "control:pending_full",
            )
            self._ok({"delivered": sent, "data": data})
            return

        if action == "error" and method in ("GET", "POST"):
            session_id = self._param(query, "session", MAIN_SESSION_ID)
            code = self._param(query, "code", "chat_busy")
            message = self._param(query, "message", "session queue is full")
            client_msg_id = self._param(query, "clientMsgId", None)
            message_id = self._param(query, "messageId", None)
            data = {"code": code, "message": message, "sessionId": session_id}
            if client_msg_id:
                data["clientMsgId"] = client_msg_id
            if message_id:
                data["messageId"] = message_id
            sent = self.store.broadcast({"type": "error", "data": data}, "control:error")
            self._ok({"delivered": sent, "data": data})
            return

        if action == "frame" and method in ("GET", "POST"):
            body, raw = self._parse_json_body()
            payload = body
            inline = self._param(query, "json", None)
            if payload is None and inline:
                try:
                    payload = json.loads(inline)
                except json.JSONDecodeError as exc:
                    self._fail(400, "bad_request", "json param is not valid json: {}".format(exc))
                    return
            if payload is None and isinstance(raw, str):
                self._fail(400, "bad_request", "body must be a json frame: {}".format(raw))
                return
            if not isinstance(payload, dict) or not payload.get("type"):
                self._fail(400, "bad_request", "a frame needs at least {\"type\": ...}")
                return
            sent = self.store.broadcast(payload, "control:frame")
            self._ok({"delivered": sent, "frame": payload})
            return

        if action == "forget" and method in ("GET", "POST"):
            session_id = self._param(query, "session", None)
            if not session_id:
                self._fail(400, "bad_request", "session is required")
                return
            with self.store.lock:
                if session_id not in self.store.sessions:
                    self._fail(404, "session_not_found", "unknown session {}".format(session_id))
                    return
                del self.store.sessions[session_id]
            log("ctl ", "forgot session {} (its /messages now 404s)".format(session_id))
            self._ok({"forgot": session_id})
            return

        if action == "create" and method in ("GET", "POST"):
            session_id = self._param(query, "session", None)
            if not session_id:
                self._fail(400, "bad_request", "session is required")
                return
            kind = self._param(query, "type", "sub")
            title = self._param(query, "title", None)
            pinned = self._param(query, "pinned", "0") not in ("0", "false", "no")
            persona_id = self._param(query, "personaId", None)
            with self.store.lock:
                self.store.sessions[session_id] = Session(
                    session_id,
                    kind,
                    now_ms(),
                    title=title,
                    pinned=pinned,
                    persona_id=persona_id,
                )
            self._ok({"session": session_id, "type": kind})
            return

        if action == "clear" and method in ("GET", "POST"):
            session_id = self._param(query, "session", MAIN_SESSION_ID)
            with self.store.lock:
                session = self.store.sessions.get(session_id)
                if session is None:
                    self._fail(404, "session_not_found", "unknown session {}".format(session_id))
                    return
                session.messages = []
                session.read_at = 0
            self._ok({"cleared": session_id})
            return

        if action == "ws-close" and method in ("GET", "POST"):
            code = self._int_param(query, "code", 1001)
            reason = self._param(query, "reason", "fake backend closed the socket")
            with self.store.lock:
                targets = list(self.store.clients)
            for client in targets:
                client.send_close(code, reason)
            log("ctl ", "closed {} client sockets (code={})".format(len(targets), code))
            self._ok({"closed": len(targets), "code": code})
            return

        self._fail(404, "not_found", "no such control action: {}".format(action))

    def _param(self, query, name, default):
        values = query.get(name)
        if not values:
            return default
        return values[0]

    def _int_param(self, query, name, default):
        raw = self._param(query, name, None)
        if raw in (None, ""):
            return default
        try:
            return int(raw)
        except ValueError:
            return default


_BAD = object()  # _parse_cursor 的失败哨兵


class Server(ThreadingHTTPServer):
    daemon_threads = True
    allow_reuse_address = True


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="fake_backend.py",
        description="Cyrene 可编程假后端（开发用测试台，不是产品的一部分）",
    )
    parser.add_argument("--host", default="127.0.0.1", help="监听地址（默认 127.0.0.1）")
    parser.add_argument("--port", type=int, default=18080, help="监听端口（默认 18080；别用真后端的 10942）")
    parser.add_argument(
        "--history-replay",
        type=int,
        default=50,
        dest="history_replay",
        help="WS 握手 history 帧回放的条数（默认 50，与真后端一致）。比库里的条数小就是那个 after 洞",
    )
    parser.add_argument(
        "--stream-chunk",
        action="store_true",
        dest="stream_chunk",
        help="回复走旧形状 stream_chunk + stream_end（用来验客户端那条回退分支）",
    )
    parser.add_argument(
        "--auto-reply",
        action="store_true",
        dest="auto_reply",
        help="收到上行 chat 帧后自动回一轮（默认关：不自动回，保持确定性）",
    )
    parser.add_argument(
        "--interrupt-ack",
        action="store_true",
        dest="interrupt_ack",
        help="收到上行 interrupt 帧后回一帧 session_interrupted{mode:idle,dropped:0}（默认关）",
    )
    parser.add_argument(
        "--hold-replies",
        action="store_true",
        dest="hold_replies",
        help="上行消息**只记录、不落库**：收到 chat / POST /api/v1/chat 后不广播 message_persisted，"
        "由 POST /__control/persist 显式补落库（默认关，默认行为逐字不变）。用来复现「忙时被拒 / 排队 / 队列满」",
    )
    parser.add_argument(
        "--out-dir",
        default=os.path.join(HERE, "out"),
        dest="out_dir",
        help="记录目录（默认 <脚本目录>/out）：upstream.jsonl / downstream.jsonl / http.jsonl",
    )
    parser.add_argument(
        "--keep-records",
        type=int,
        default=1000,
        dest="keep_records",
        help="内存里保留的上行帧条数上限（默认 1000；文件里是全量）",
    )
    return parser


def main(argv=None) -> int:
    args = build_parser().parse_args(argv)
    store = Store(args)
    Handler.store = store

    server = Server((args.host, args.port), Handler)
    host, port = server.server_address[0], server.server_address[1]
    print("=" * 72)
    print("Cyrene 假后端已启动  http://{}:{}".format(host, port))
    print("  WS      ws://{}:{}/ws/chat?token=<任意>".format(host, port))
    print("  history 握手回放 {} 条（--history-replay）".format(args.history_replay))
    print("  回复形状 {}".format("stream_chunk 旧形状" if args.stream_chunk else "message.* 结构化"))
    print("  自动回复 {} / interrupt 回执 {}".format(args.auto_reply, args.interrupt_ack))
    print(
        "  上行落库 {}".format(
            "挂起（--hold-replies：等 POST /__control/persist 才落库）"
            if args.hold_replies
            else "立即（收到即落库 + message_persisted）"
        )
    )
    print("  记录目录 {}".format(args.out_dir))
    print("  控制台 GET /__control/state")
    print("=" * 72)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nbye")
    finally:
        server.server_close()
    return 0


if __name__ == "__main__":
    sys.exit(main())