Cat llm

cat_llm_server.py

#!/usr/bin/env python3
# Copyright (c) 2026 Max Maton
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
"""OpenAI-compatible chat/completions server that simulates a cat walking on a keyboard.

Two paws walk a 2D QWERTY grid with gait and momentum. Printable keys produce text;
Enter ends the message; F-keys invoke tools from the request (F1 = first tool, extra
rows F13+ when there are more than 12 tools). String tool arguments use the same walk.

Binds to 127.0.0.1:8013 by default. No third-party dependencies.

Run:

    python cat_llm_server.py

Point the editor LLM base URL at http://127.0.0.1:8013/v1

Smoke tests:

    # Non-streaming text
    curl -s http://127.0.0.1:8013/v1/chat/completions \\
      -H 'Content-Type: application/json' \\
      -d '{"model":"cat","messages":[{"role":"user","content":"hi"}],"stream":false}'

    # Tool call
    curl -s http://127.0.0.1:8013/v1/chat/completions \\
      -H 'Content-Type: application/json' \\
      -d '{"model":"cat","messages":[{"role":"user","content":"hi"}],"tools":[{"type":"function","function":{"name":"grep","parameters":{"type":"object","properties":{"pattern":{"type":"string"},"case_sensitive":{"type":"boolean","default":false}},"required":["pattern"]}}}],"stream":false}'

    # Model list (llama.cpp uses GET /models; OpenAI clients use GET /v1/models)
    curl -s http://127.0.0.1:8013/models

    # Tokenize (llama.cpp POST /tokenize)
    curl -s http://127.0.0.1:8013/tokenize \\
      -H 'Content-Type: application/json' \\
      -d '{"content": "hello", "with_pieces": true}'

    # Apply chat template (llama.cpp POST /apply-template)
    curl -s http://127.0.0.1:8013/apply-template \\
      -H 'Content-Type: application/json' \\
      -d '{"messages": [{"role": "user", "content": "Hello!"}]}'

    # Streaming
    curl -N http://127.0.0.1:8013/v1/chat/completions \\
      -H 'Content-Type: application/json' \\
      -d '{"model":"cat","messages":[{"role":"user","content":"hi"}],"stream":true}'

Optional: --seed N for reproducibility; request header X-Cat-Seed overrides per call.
"""

from __future__ import annotations

import argparse
import json
import random
import secrets
import sys
import time
from dataclasses import dataclass, field
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Any, Iterator, Literal
from urllib.parse import urlparse

LISTEN_HOST = "127.0.0.1"
LISTEN_PORT = 8013
FKEYS_PER_ROW = 12
MAX_PAW_DISTANCE = 4
MOMENTUM_CONTINUE_PROB = 0.75
SAME_PAW_TWICE_PROB = 0.15
TRAILING_PRESS_PROB = 0.30
DEFAULT_MAX_KEY_EVENTS = 512
DEFAULT_MODEL = "cat-on-keyboard"
BOS_TOKEN_ID = 1
BOS_TOKEN_PIECE = "<s>"
# Fake vocab: one token per Unicode code point (stable, no real tokenizer).
_TOKEN_ID_BASE = 256
_IM_START = "<|im_start|>"
_IM_END = ""

# US QWERTY rows (F-row prepended / extended at runtime). None = empty cell.
_BASE_CHAR_ROWS: list[list[str | None]] = [
    list("`1234567890-="),
    list("qwertyuiop[]\\"),
    list("asdfghjkl;'"),
    list("zxcvbnm,./"),
    [None, None, " ", " ", " ", " ", " ", " ", " ", None, None, "Enter"],
]

_DIRECTIONS: tuple[tuple[int, int], ...] = (
    (-1, -1),
    (-1, 0),
    (-1, 1),
    (0, -1),
    (0, 1),
    (1, -1),
    (1, 0),
    (1, 1),
)


@dataclass(frozen=True)
class KeyCell:
    kind: Literal["empty", "char", "enter", "fkey"]
    value: str | int = ""

    @staticmethod
    def from_label(label: str | None) -> KeyCell:
        if label is None:
            return KeyCell("empty")
        if label == "Enter":
            return KeyCell("enter")
        if label.startswith("F") and label[1:].isdigit():
            return KeyCell("fkey", int(label[1:]))
        return KeyCell("char", label)


@dataclass
class ToolSpec:
    name: str
    parameters: dict[str, Any]
    fkey: int


@dataclass
class KeyPressEvent:
    kind: Literal["char", "tool", "enter"]
    value: str | ToolSpec = ""


@dataclass
class KeyboardGrid:
    rows: list[list[KeyCell]] = field(default_factory=list)

    @classmethod
    def build(cls, num_tools: int) -> KeyboardGrid:
        extra_f_rows = max(0, (num_tools - 1) // FKEYS_PER_ROW)
        grid_rows: list[list[KeyCell]] = []
        for row_idx in range(extra_f_rows, 0, -1):
            start = row_idx * FKEYS_PER_ROW + 1
            labels = [f"F{i}" for i in range(start, start + FKEYS_PER_ROW)]
            grid_rows.append([KeyCell.from_label(l) for l in labels])
        f_labels = [f"F{i}" for i in range(1, FKEYS_PER_ROW + 1)]
        grid_rows.append([KeyCell.from_label(l) for l in f_labels])
        for char_row in _BASE_CHAR_ROWS:
            grid_rows.append([KeyCell.from_label(c) for c in char_row])
        return cls(rows=grid_rows)

    @property
    def height(self) -> int:
        return len(self.rows)

    @property
    def width(self) -> int:
        return max((len(r) for r in self.rows), default=0)

    def cell(self, row: int, col: int) -> KeyCell:
        if row < 0 or row >= self.height:
            return KeyCell("empty")
        r = self.rows[row]
        if col < 0 or col >= len(r):
            return KeyCell("empty")
        return r[col]

    def letter_rows(self) -> range:
        """Row indices for the main typing area (after F-key rows)."""
        f_row_count = 0
        for row in self.rows:
            if row and row[0].kind == "fkey":
                f_row_count += 1
            else:
                break
        return range(f_row_count, self.height)


def _manhattan(a: tuple[int, int], b: tuple[int, int]) -> int:
    return abs(a[0] - b[0]) + abs(a[1] - b[1])


def _clamp_step(row: int, col: int, dr: int, dc: int, grid: KeyboardGrid) -> tuple[int, int]:
    nr = max(0, min(grid.height - 1, row + dr))
    nc = max(0, min(grid.width - 1, col + dc))
    return nr, nc


def _nearest_non_empty(grid: KeyboardGrid, row: int, col: int) -> tuple[int, int]:
    if grid.cell(row, col).kind != "empty":
        return row, col
    for dist in range(1, 6):
        for dr in range(-dist, dist + 1):
            for dc in range(-dist, dist + 1):
                if abs(dr) + abs(dc) != dist:
                    continue
                r, c = row + dr, col + dc
                if grid.cell(r, c).kind != "empty":
                    return r, c
    return row, col


def _step_toward(src: tuple[int, int], dst: tuple[int, int]) -> tuple[int, int]:
    sr, sc = src
    dr = dst[0] - sr
    dc = dst[1] - sc
    if dr == 0 and dc == 0:
        return src
    step_r = 0 if dr == 0 else (1 if dr > 0 else -1)
    step_c = 0 if dc == 0 else (1 if dc > 0 else -1)
    if abs(dr) >= abs(dc):
        return sr + step_r, sc
    return sr, sc + step_c


class CatWalkSimulator:
    def __init__(
        self,
        rng: random.Random,
        grid: KeyboardGrid,
        tool_by_fkey: dict[int, ToolSpec],
        allowed_fkeys: set[int] | None = None,
    ) -> None:
        self.rng = rng
        self.grid = grid
        self.tool_by_fkey = tool_by_fkey
        self.allowed_fkeys = allowed_fkeys
        self.left_paw: tuple[int, int] = (0, 0)
        self.right_paw: tuple[int, int] = (0, 0)
        self.leading_paw: Literal["left", "right"] = "left"
        self.heading: tuple[int, int] = (0, 1)

    def type_until_enter(self, max_events: int = DEFAULT_MAX_KEY_EVENTS) -> str:
        chars: list[str] = []
        for ev in self.walk_until_enter(max_events):
            if ev.kind == "char":
                chars.append(str(ev.value))
        return "".join(chars)

    def walk_until_enter(self, max_events: int = DEFAULT_MAX_KEY_EVENTS) -> list[KeyPressEvent]:
        self._init_paws()
        events: list[KeyPressEvent] = []
        count = 0
        while count < max_events:
            for ev in self._step():
                if ev.kind == "enter":
                    return events
                events.append(ev)
                count += 1
                if count >= max_events:
                    return events
        return events

    def simulate_message(
        self,
        max_events: int = DEFAULT_MAX_KEY_EVENTS,
    ) -> tuple[str | None, list[dict[str, Any]]]:
        raw = self.walk_until_enter(max_events)
        content_parts: list[str] = []
        tool_calls: list[dict[str, Any]] = []
        for ev in raw:
            if ev.kind == "char":
                content_parts.append(str(ev.value))
            elif ev.kind == "tool":
                spec = ev.value
                assert isinstance(spec, ToolSpec)
                args = generate_args_from_schema(spec.parameters, self.rng, self.grid, self.tool_by_fkey)
                tool_calls.append(
                    {
                        "id": f"call_{secrets.token_hex(8)}",
                        "type": "function",
                        "function": {
                            "name": spec.name,
                            "arguments": json.dumps(args, ensure_ascii=False),
                        },
                    }
                )
        content = "".join(content_parts)
        return (content if content else None), tool_calls

    def _init_paws(self) -> None:
        letter = list(self.grid.letter_rows())
        if not letter:
            letter = [0]
        row = self.rng.choice(letter)
        col = self.rng.randrange(self.grid.width)
        left = _nearest_non_empty(self.grid, row, col)
        offset_c = self.rng.choice([-2, -1, 1, 2])
        offset_r = self.rng.choice([0, 0, -1, 1])
        right = _nearest_non_empty(self.grid, row + offset_r, col + offset_c)
        if _manhattan(left, right) > MAX_PAW_DISTANCE:
            right = _nearest_non_empty(self.grid, row, col + self.rng.choice([-1, 1]))
        self.left_paw = left
        self.right_paw = right
        self.leading_paw = self.rng.choice(["left", "right"])
        self.heading = self.rng.choice(_DIRECTIONS)

    def _pick_direction(self) -> tuple[int, int]:
        if self.rng.random() < MOMENTUM_CONTINUE_PROB and self.heading != (0, 0):
            return self.heading
        # Bias toward letter rows (away from top F-rows).
        f_rows = sum(1 for r in self.grid.rows if r and r[0].kind == "fkey")
        candidates = list(_DIRECTIONS)
        weights = []
        for dr, dc in candidates:
            w = 1.0
            paw = self.left_paw if self.leading_paw == "left" else self.right_paw
            nr, nc = _clamp_step(paw[0], paw[1], dr, dc, self.grid)
            if nr >= f_rows + 1:
                w += 1.5
            if nr < f_rows:
                w *= 0.5
            weights.append(w)
        return self.rng.choices(candidates, weights=weights, k=1)[0]

    def _resolve_press(self, row: int, col: int) -> KeyPressEvent | None:
        cell = self.grid.cell(row, col)
        if cell.kind == "empty":
            return None
        if cell.kind == "enter":
            return KeyPressEvent("enter")
        if cell.kind == "char":
            return KeyPressEvent("char", cell.value)
        if cell.kind == "fkey":
            fnum = int(cell.value)
            if self.allowed_fkeys is not None and fnum not in self.allowed_fkeys:
                return None
            spec = self.tool_by_fkey.get(fnum)
            if spec is None:
                return None
            return KeyPressEvent("tool", spec)
        return None

    def _step(self) -> list[KeyPressEvent]:
        if self.rng.random() >= SAME_PAW_TWICE_PROB:
            self.leading_paw = "right" if self.leading_paw == "left" else "left"

        direction = self._pick_direction()
        self.heading = direction
        dr, dc = direction

        if self.leading_paw == "left":
            step_paw, trail_paw = "left", "right"
            step_pos = self.left_paw
        else:
            step_paw, trail_paw = "right", "left"
            step_pos = self.right_paw

        sr, sc = step_pos
        new_step = _clamp_step(sr, sc, dr, dc, self.grid)
        if step_paw == "left":
            self.left_paw = new_step
        else:
            self.right_paw = new_step

        trail_pos = self.left_paw if trail_paw == "left" else self.right_paw
        step_pos = self.left_paw if step_paw == "left" else self.right_paw

        if _manhattan(step_pos, trail_pos) > MAX_PAW_DISTANCE:
            shuffled = _step_toward(trail_pos, step_pos)
            shuffled = (
                max(0, min(self.grid.height - 1, shuffled[0])),
                max(0, min(self.grid.width - 1, shuffled[1])),
            )
            if trail_paw == "left":
                self.left_paw = shuffled
            else:
                self.right_paw = shuffled

        events: list[KeyPressEvent] = []
        press = self._resolve_press(step_pos[0], step_pos[1])
        if press is not None:
            if press.kind == "enter":
                return [press]
            events.append(press)

        trail_pos = self.left_paw if trail_paw == "left" else self.right_paw
        if self.rng.random() < TRAILING_PRESS_PROB:
            trail_press = self._resolve_press(trail_pos[0], trail_pos[1])
            if trail_press is not None:
                if trail_press.kind == "enter":
                    return events + [trail_press]
                events.append(trail_press)

        return events


def generate_args_from_schema(
    schema: dict[str, Any] | None,
    rng: random.Random,
    grid: KeyboardGrid,
    tool_by_fkey: dict[int, ToolSpec],
) -> dict[str, Any]:
    if not isinstance(schema, dict):
        return {}
    if schema.get("type") == "object" or "properties" in schema:
        return _object_from_schema(schema, rng, grid, tool_by_fkey)
    return {}


def _object_from_schema(
    schema: dict[str, Any],
    rng: random.Random,
    grid: KeyboardGrid,
    tool_by_fkey: dict[int, ToolSpec],
) -> dict[str, Any]:
    if isinstance(schema.get("example"), dict):
        return dict(schema["example"])
    props = schema.get("properties")
    if not isinstance(props, dict):
        return {}
    required_raw = schema.get("required")
    required = set(required_raw) if isinstance(required_raw, list) else set()
    out: dict[str, Any] = {}
    for key, spec in props.items():
        if not isinstance(spec, dict):
            continue
        if "default" in spec:
            out[key] = spec["default"]
            continue
        if key not in required:
            continue
        out[key] = _value_from_property(spec, rng, grid, tool_by_fkey)
    return out


def _value_from_property(
    spec: dict[str, Any],
    rng: random.Random,
    grid: KeyboardGrid,
    tool_by_fkey: dict[int, ToolSpec],
) -> Any:
    if "example" in spec:
        return spec["example"]
    if "oneOf" in spec and isinstance(spec["oneOf"], list) and spec["oneOf"]:
        branch = rng.choice(spec["oneOf"])
        if isinstance(branch, dict):
            return _value_from_property(branch, rng, grid, tool_by_fkey)
    if "anyOf" in spec and isinstance(spec["anyOf"], list) and spec["anyOf"]:
        branch = rng.choice(spec["anyOf"])
        if isinstance(branch, dict):
            return _value_from_property(branch, rng, grid, tool_by_fkey)

    if "enum" in spec and isinstance(spec["enum"], list) and spec["enum"]:
        return rng.choice(spec["enum"])

    typ = spec.get("type")
    if typ == "boolean":
        return rng.choice([True, False])
    if typ == "integer":
        if "minimum" in spec and "maximum" in spec:
            lo, hi = int(spec["minimum"]), int(spec["maximum"])
            if lo > hi:
                lo, hi = hi, lo
            return rng.randint(lo, hi)
        return rng.randint(0, 100)
    if typ == "number":
        if "minimum" in spec and "maximum" in spec:
            lo, hi = float(spec["minimum"]), float(spec["maximum"])
            if lo > hi:
                lo, hi = hi, lo
            return rng.uniform(lo, hi)
        return float(rng.randint(0, 100))
    if typ == "string":
        sim = CatWalkSimulator(rng, grid, tool_by_fkey)
        return sim.type_until_enter(max_events=128)
    if typ == "object":
        return _object_from_schema(spec, rng, grid, tool_by_fkey)
    if typ == "array":
        items = spec.get("items")
        if not isinstance(items, dict):
            return []
        length = rng.randint(0, 2)
        return [_value_from_property(items, rng, grid, tool_by_fkey) for _ in range(length)]
    return ""


def parse_tools(tools_raw: Any) -> list[ToolSpec]:
    if not isinstance(tools_raw, list):
        return []
    out: list[ToolSpec] = []
    for i, t in enumerate(tools_raw):
        if not isinstance(t, dict) or t.get("type") != "function":
            continue
        fn = t.get("function")
        if not isinstance(fn, dict):
            continue
        name = str(fn.get("name") or "").strip()
        if not name:
            continue
        params = fn.get("parameters")
        if not isinstance(params, dict):
            params = {}
        out.append(ToolSpec(name=name, parameters=params, fkey=i + 1))
    return out


def resolve_tool_choice(
    tool_choice: Any,
    tools: list[ToolSpec],
) -> tuple[dict[int, ToolSpec], set[int] | None]:
    """Return (fkey->ToolSpec map, allowed fkeys or None for all mapped)."""
    if not tools:
        return {}, set()

    by_fkey = {t.fkey: t for t in tools}
    by_name = {t.name: t for t in tools}

    if tool_choice is None or tool_choice == "auto":
        return by_fkey, None

    if tool_choice == "none":
        return by_fkey, set()

    if tool_choice == "required":
        return by_fkey, None

    if isinstance(tool_choice, dict):
        fn = tool_choice.get("function")
        if isinstance(fn, dict):
            name = str(fn.get("name") or "").strip()
            if name in by_name:
                t = by_name[name]
                return by_fkey, {t.fkey}
        if tool_choice.get("type") == "function" and isinstance(fn, dict):
            name = str(fn.get("name") or "").strip()
            if name in by_name:
                t = by_name[name]
                return by_fkey, {t.fkey}

    return by_fkey, None


def run_simulation(
    rng: random.Random,
    tools: list[ToolSpec],
    tool_choice: Any,
    max_events: int,
) -> tuple[str | None, list[dict[str, Any]]]:
    tool_by_fkey, allowed = resolve_tool_choice(tool_choice, tools)
    grid = KeyboardGrid.build(len(tools))
    sim = CatWalkSimulator(rng, grid, tool_by_fkey, allowed_fkeys=allowed)
    return sim.simulate_message(max_events=max_events)


def _completion_id() -> str:
    return f"chatcmpl-{secrets.token_hex(12)}"


def build_completion_body(
    model: str,
    content: str | None,
    tool_calls: list[dict[str, Any]],
) -> dict[str, Any]:
    completion_tokens = len(content or "") + len(tool_calls)
    has_tools_only = bool(tool_calls) and not (content or "").strip()
    finish_reason = "tool_calls" if has_tools_only else "stop"
    message: dict[str, Any] = {"role": "assistant", "content": content}
    if tool_calls:
        message["tool_calls"] = tool_calls
    return {
        "id": _completion_id(),
        "object": "chat.completion",
        "created": int(time.time()),
        "model": model,
        "choices": [
            {
                "index": 0,
                "message": message,
                "finish_reason": finish_reason,
            }
        ],
        "usage": {
            "prompt_tokens": 0,
            "completion_tokens": completion_tokens,
            "total_tokens": completion_tokens,
        },
    }


def iter_sse_chunks(
    model: str,
    content: str | None,
    tool_calls: list[dict[str, Any]],
) -> Iterator[str]:
    cid = _completion_id()
    created = int(time.time())
    base = {
        "id": cid,
        "object": "chat.completion.chunk",
        "created": created,
        "model": model,
    }

    def chunk(delta: dict[str, Any], finish: str | None = None) -> str:
        choice: dict[str, Any] = {"index": 0, "delta": delta}
        if finish is not None:
            choice["finish_reason"] = finish
        else:
            choice["finish_reason"] = None
        payload = {**base, "choices": [choice]}
        return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"

    yield chunk({"role": "assistant"})
    for ch in content or "":
        yield chunk({"content": ch})
    if tool_calls:
        yield chunk({"tool_calls": tool_calls})
    has_tools_only = bool(tool_calls) and not (content or "").strip()
    finish_reason = "tool_calls" if has_tools_only else "stop"
    yield chunk({}, finish=finish_reason)
    yield "data: [DONE]\n\n"


def _parse_max_tokens(body: dict[str, Any]) -> int:
    mt = body.get("max_tokens")
    if mt is None:
        return DEFAULT_MAX_KEY_EVENTS
    try:
        n = int(mt)
        return max(1, min(n, DEFAULT_MAX_KEY_EVENTS * 4))
    except (TypeError, ValueError):
        return DEFAULT_MAX_KEY_EVENTS


def _request_rng(handler: BaseHTTPRequestHandler, default_seed: int | None) -> random.Random:
    hdr = handler.headers.get("X-Cat-Seed")
    if hdr is not None:
        try:
            return random.Random(int(hdr))
        except ValueError:
            pass
    if default_seed is not None:
        return random.Random(default_seed ^ random.randrange(1 << 30))
    return random.Random(random.randrange(1 << 30))


def _models_body() -> dict[str, Any]:
    """OpenAI + llama.cpp-compatible model list (GET /models or GET /v1/models)."""
    return {
        "object": "list",
        "data": [
            {
                "id": DEFAULT_MODEL,
                "object": "model",
                "created": int(time.time()),
                "owned_by": "llamacpp",
                "aliases": [],
                "tags": [],
            }
        ],
    }


def _send_json(handler: BaseHTTPRequestHandler, status: int, body: object) -> None:
    data = json.dumps(body, ensure_ascii=False).encode("utf-8")
    handler.send_response(status)
    handler.send_header("Content-Type", "application/json; charset=utf-8")
    handler.send_header("Content-Length", str(len(data)))
    handler.end_headers()
    handler.wfile.write(data)


def _send_error_json(handler: BaseHTTPRequestHandler, status: int, message: str) -> None:
    _send_json(
        handler,
        status,
        {"error": {"message": message, "type": "invalid_request_error"}},
    )


def _read_json_body(handler: BaseHTTPRequestHandler) -> tuple[dict[str, Any] | None, str | None]:
    length = int(handler.headers.get("Content-Length", "0") or "0")
    raw = handler.rfile.read(length) if length > 0 else b""
    try:
        body = json.loads(raw.decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError) as e:
        return None, f"invalid JSON body: {e}"
    if not isinstance(body, dict):
        return None, "request body must be a JSON object"
    return body, None


def _char_token_id(ch: str) -> int:
    return _TOKEN_ID_BASE + ord(ch)


def _tokenize_content(body: dict[str, Any]) -> tuple[dict[str, Any] | None, str | None]:
    """llama.cpp POST /tokenize — mock one-token-per-character encoding."""
    if "content" not in body:
        return None, "missing required field: content"
    content = body["content"]
    if content is None:
        return None, "content must be a string"
    if not isinstance(content, str):
        return None, "content must be a string"

    add_special = bool(body.get("add_special", False))
    with_pieces = bool(body.get("with_pieces", False))
    # parse_special is accepted for API compatibility; mock tokenizer has no special strings.

    tokens: list[Any] = []
    if add_special:
        if with_pieces:
            tokens.append({"id": BOS_TOKEN_ID, "piece": BOS_TOKEN_PIECE})
        else:
            tokens.append(BOS_TOKEN_ID)

    for ch in content:
        tid = _char_token_id(ch)
        if with_pieces:
            tokens.append({"id": tid, "piece": ch})
        else:
            tokens.append(tid)

    return {"tokens": tokens}, None


def _message_text(content: Any) -> str:
    if content is None:
        return ""
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        parts: list[str] = []
        for part in content:
            if isinstance(part, dict) and part.get("type") == "text":
                parts.append(str(part.get("text") or ""))
            elif isinstance(part, str):
                parts.append(part)
        return "".join(parts)
    return str(content)


def _format_tool_calls(tool_calls: Any) -> str:
    if not isinstance(tool_calls, list):
        return ""
    blocks: list[str] = []
    for tc in tool_calls:
        if not isinstance(tc, dict):
            continue
        fn = tc.get("function")
        if not isinstance(fn, dict):
            continue
        name = str(fn.get("name") or "")
        args_raw = fn.get("arguments")
        if isinstance(args_raw, dict):
            args_json = json.dumps(args_raw, ensure_ascii=False)
        elif isinstance(args_raw, str) and args_raw.strip():
            try:
                args_json = json.dumps(json.loads(args_raw), ensure_ascii=False)
            except json.JSONDecodeError:
                args_json = json.dumps(args_raw, ensure_ascii=False)
        else:
            args_json = "{}"
        blocks.append(f'<tool_call>\n{{"name": {json.dumps(name)}, "arguments": {args_json}}}\n</tool_call>')
    return "\n".join(blocks)


def _apply_template(body: dict[str, Any]) -> tuple[dict[str, Any] | None, str | None]:
    """llama.cpp POST /apply-template — mock ChatML formatting."""
    messages = body.get("messages")
    if messages is None:
        return None, "missing required field: messages"
    if not isinstance(messages, list):
        return None, "messages must be an array"

    add_generation_prompt = bool(body.get("add_generation_prompt", True))
    parts: list[str] = []

    for msg in messages:
        if not isinstance(msg, dict):
            return None, "each message must be an object"
        role = str(msg.get("role") or "user")
        text = _message_text(msg.get("content"))

        if role == "assistant":
            block_parts: list[str] = []
            reasoning = msg.get("reasoning_content")
            if isinstance(reasoning, str) and reasoning.strip():
                block_parts.append(f"\n{reasoning}\n")
            if text:
                block_parts.append(text)
            tc_text = _format_tool_calls(msg.get("tool_calls"))
            if tc_text:
                block_parts.append(tc_text)
            block = "\n".join(block_parts)
        elif role == "tool":
            call_id = str(msg.get("tool_call_id") or "")
            block = f'<tool_response id={json.dumps(call_id)}>\n{text}\n</tool_response>'
        else:
            block = text

        parts.append(f"{_IM_START}{role}\n{block}{_IM_END}\n")

    prompt = "".join(parts)
    if add_generation_prompt:
        last_role = str(messages[-1].get("role") or "") if messages else ""
        if not messages or last_role != "assistant":
            prompt += f"{_IM_START}assistant\n"

    return {"prompt": prompt}, None


def _handle_chat_completions(
    handler: BaseHTTPRequestHandler,
    body: dict[str, Any],
    default_seed: int | None,
) -> None:
    model = str(body.get("model") or DEFAULT_MODEL)
    tools = parse_tools(body.get("tools"))
    tool_choice = body.get("tool_choice", "auto")
    max_events = _parse_max_tokens(body)
    rng = _request_rng(handler, default_seed)
    content, tool_calls = run_simulation(rng, tools, tool_choice, max_events)

    if body.get("stream"):
        handler.send_response(HTTPStatus.OK)
        handler.send_header("Content-Type", "text/event-stream; charset=utf-8")
        handler.send_header("Cache-Control", "no-cache")
        handler.send_header("Connection", "keep-alive")
        handler.end_headers()
        try:
            for part in iter_sse_chunks(model, content, tool_calls):
                handler.wfile.write(part.encode("utf-8"))
                handler.wfile.flush()
        except BrokenPipeError:
            pass
        return

    _send_json(handler, HTTPStatus.OK, build_completion_body(model, content, tool_calls))


def make_handler(default_seed: int | None) -> type[BaseHTTPRequestHandler]:
    class CatLLMHandler(BaseHTTPRequestHandler):
        server_version = "CatLLM/1.0"

        def log_message(self, fmt: str, *args: object) -> None:
            sys.stderr.write("%s - [%s] %s\n" % (self.address_string(), self.log_date_time_string(), fmt % args))

        def do_GET(self) -> None:
            path = urlparse(self.path).path.rstrip("/") or "/"
            if path == "/":
                _send_json(
                    self,
                    HTTPStatus.OK,
                    {"service": "cat-llm-server", "openai_compatible": True},
                )
                return
            if path in ("/models", "/v1/models"):
                _send_json(self, HTTPStatus.OK, _models_body())
                return
            _send_error_json(self, HTTPStatus.NOT_FOUND, f"unknown route: {path}")

        def do_POST(self) -> None:
            path = urlparse(self.path).path.rstrip("/")
            body, err = _read_json_body(self)
            if err is not None:
                _send_error_json(self, HTTPStatus.BAD_REQUEST, err)
                return

            if path == "/tokenize":
                result, err = _tokenize_content(body)
                if err is not None:
                    _send_error_json(self, HTTPStatus.BAD_REQUEST, err)
                    return
                _send_json(self, HTTPStatus.OK, result)
                return

            if path == "/apply-template":
                result, err = _apply_template(body)
                if err is not None:
                    _send_error_json(self, HTTPStatus.BAD_REQUEST, err)
                    return
                _send_json(self, HTTPStatus.OK, result)
                return

            if path == "/v1/chat/completions":
                _handle_chat_completions(self, body, default_seed)
                return

            _send_error_json(self, HTTPStatus.NOT_FOUND, f"unknown route: {path}")

    return CatLLMHandler


def main() -> int:
    p = argparse.ArgumentParser(
        description="OpenAI-compatible LLM server simulating a cat walking on a keyboard.",
    )
    p.add_argument("--host", default=LISTEN_HOST, help=f"bind host (default: {LISTEN_HOST})")
    p.add_argument("--port", type=int, default=LISTEN_PORT, help=f"bind port (default: {LISTEN_PORT})")
    p.add_argument(
        "--seed",
        type=int,
        default=None,
        help="optional RNG seed base (per-request X-Cat-Seed header overrides)",
    )
    args = p.parse_args()

    Handler = make_handler(args.seed)
    httpd = ThreadingHTTPServer((args.host, args.port), Handler)
    print(f"LISTENING=http://{args.host}:{args.port}", flush=True)
    print(f"OPENAI_BASE=http://{args.host}:{args.port}/v1", flush=True)
    if args.seed is not None:
        print(f"CAT_SEED={args.seed}", flush=True)
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        print("\nShutting down.", file=sys.stderr)
        return 0
    finally:
        httpd.server_close()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())