From 6070cf60d161a9abab41321ec1322433b293ea22 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=8A=84=E6=89=8B=E6=BB=8B=E5=91=B3?= Date: Wed, 12 Aug 2026 01:15:48 +0800 Subject: [PATCH 1/2] feat(cc-bridge): native Anthropic Messages adapter for Claude Code Thin protocol bridge letting Claude Code talk to a local baseRT server: /v1/messages <-> /v1/chat/completions with native SSE streaming, thinking and tool blocks, keep-alive pings, and cancellation propagation. Inbound requests require a master key; binds to 127.0.0.1 by default; the classifier-bypass short-circuit is off unless explicitly enabled. Includes pytest unit and end-to-end tests. --- cc-bridge/.gitignore | 2 + cc-bridge/anthropic_adapter.py | 913 +++++++++++++++++++++++++++++++ cc-bridge/pytest.ini | 4 + cc-bridge/requirements-dev.txt | 3 + cc-bridge/requirements.txt | 1 + cc-bridge/tests/conftest.py | 83 +++ cc-bridge/tests/helpers.py | 28 + cc-bridge/tests/test_auth.py | 83 +++ cc-bridge/tests/test_convert.py | 141 +++++ cc-bridge/tests/test_estimate.py | 51 ++ cc-bridge/tests/test_sse.py | 188 +++++++ cc-bridge/tests/test_streamer.py | 91 +++ 12 files changed, 1588 insertions(+) create mode 100644 cc-bridge/.gitignore create mode 100644 cc-bridge/anthropic_adapter.py create mode 100644 cc-bridge/pytest.ini create mode 100644 cc-bridge/requirements-dev.txt create mode 100644 cc-bridge/requirements.txt create mode 100644 cc-bridge/tests/conftest.py create mode 100644 cc-bridge/tests/helpers.py create mode 100644 cc-bridge/tests/test_auth.py create mode 100644 cc-bridge/tests/test_convert.py create mode 100644 cc-bridge/tests/test_estimate.py create mode 100644 cc-bridge/tests/test_sse.py create mode 100644 cc-bridge/tests/test_streamer.py diff --git a/cc-bridge/.gitignore b/cc-bridge/.gitignore new file mode 100644 index 0000000..599b2a7 --- /dev/null +++ b/cc-bridge/.gitignore @@ -0,0 +1,2 @@ +.venv/ +debug/ diff --git a/cc-bridge/anthropic_adapter.py b/cc-bridge/anthropic_adapter.py new file mode 100644 index 0000000..65ee5ea --- /dev/null +++ b/cc-bridge/anthropic_adapter.py @@ -0,0 +1,913 @@ +#!/usr/bin/env python3 +""" +Native Anthropic Messages API adapter for baseRT (llama.cpp server pattern). + +Claude Code --(/v1/messages)--> adapter.py --(/v1/chat/completions, stream)--> baseRT + +Why this exists: Claude Code speaks only the Anthropic Messages API, while +`basert serve` exposes OpenAI-compatible endpoints. A LiteLLM bridge was tried +first and failed in two measured ways (BerriAI/litellm#27492: its non-streaming +openai->anthropic conversion always returns empty content; and Claude Code +abandons a stream after ~36s of silence, falling back to exactly that broken +non-streaming path). So the wire format is generated natively instead: + + - request: anthropic -> openai chat conversion (llama.cpp: + server_chat_convert_anthropic_to_oai) + - response: native anthropic blocks, thinking (index 0) -> text (0/1) -> tool_use (n+) + - keep-alive: SSE comment pings (":\\n\\n") every PING_INTERVAL seconds while the + stream is silent, so clients (Claude Code / stainless) never time out during + long prefills (llama.cpp: sse_ping_interval, default 30s; we start pinging + from request dispatch) + - cancellation: the upstream read loop polls every 1s (llama.cpp: + HTTP_POLLING_SECONDS) and probes the client socket non-blockingly + (httplib's is_connection_closed trick: recv(1, MSG_PEEK), EOF == gone). + On disconnect the upstream call is closed so baseRT stops generating. + - thinking blocks carry empty signature (local models, no crypto verification) + - no "data: [DONE]" for anthropic streams + +Portions of the conversion and streaming logic follow llama.cpp's Anthropic +implementation, https://github.com/ggml-org/llama.cpp (MIT license, +server/chat.cpp). See TECHNICAL.md for the design rationale. + +Deployment notes: + - baseRT is single-slot: when it is busy, upstream requests queue and the + response headers may be late. The upstream client therefore uses a + connect timeout of 10s and a read timeout of 600s (read timeout must + exceed the longest silent gap in a stream, e.g. ~35s prefill of a 23k + token prompt). + - --max-tokens-override: Claude Code sends small budgets (64/8192) for + tool-loop and background calls; a heavy-thinking local model truncates on + those. Override replaces the client budget with a fixed value. + - --debug-dir: every request body and the response actually sent to the + client are written to disk for offline debugging. + - --master-key is required: every inbound request must present it via + `x-api-key` or `Authorization: Bearer`, otherwise it is rejected with 401. + - --allow-classifier-bypass is off by default; see the classifier-bypass + section in handle_messages and TECHNICAL.md before enabling it. + +Usage: anthropic_adapter.py --port 4003 --upstream http://127.0.0.1:7999/v1 \ + --master-key --model \ + [--max-tokens-override N] [--debug-dir DIR] [--allow-classifier-bypass] +""" +import argparse +import asyncio +import hmac +import json +import logging +import math +import os +import socket +import time +import uuid + +import aiohttp +from aiohttp import web + +PING_INTERVAL = 20 # seconds of silence before sending an SSE comment ping + +log = logging.getLogger("cc-bridge") + +# -------------------------------------------------------------------------- +# request conversion: anthropic /v1/messages -> openai /v1/chat/completions +# -------------------------------------------------------------------------- + +def _text_of(block): + return block.get("text", "") + + +# ========================================================================= +# System-block merging +# +# baseRT's default chat template expects the system prompt as a single +# leading message. Claude Code sends system fragments both top-level and +# inside "messages", so they are merged into one leading system message +# before conversion. Some models (e.g. froggeric v21.3 baked in via +# base-convert) handle mid-list system messages natively; use +# --no-system-merge for those. See cc-bridge/misc/chat-template-fix.md. +# ========================================================================= + +def _merge_system_blocks(body, messages_out): + """ + Collect every system prompt fragment (top-level "system" param + any + role:"system" entries inside "messages") and prepend them as a single + leading system message to *messages_out*. Returns the list of non-system + messages that should be processed by the regular role-based converter. + """ + system_parts = [] + + system = body.get("system") + if system is not None: + if isinstance(system, str): + system_parts.append(system) + elif isinstance(system, list): + system_parts.extend(_text_of(b) for b in system + if isinstance(b, dict) and b.get("type") == "text") + + rest = [] + for msg in body.get("messages", []): + if msg.get("role") == "system": + content = msg.get("content") + if isinstance(content, str): + system_parts.append(content) + elif isinstance(content, list): + system_parts.extend(_text_of(b) for b in content + if isinstance(b, dict) and b.get("type") == "text") + else: + rest.append(msg) + + if system_parts: + messages_out.append({"role": "system", "content": "\n".join(system_parts)}) + return rest + + +def convert_anthropic_to_oai(body, merge_system=True): + """Translate a /v1/messages body into an OpenAI chat.completions body. + + If *merge_system* is False the merging is skipped and system blocks stay + in their original array positions (requires a chat template that handles + mid-list system messages, e.g. froggeric v21.3 baked into the model via + base-convert). + """ + oai = {} + messages = [] + if merge_system: + rest = _merge_system_blocks(body, messages) + else: + rest = body.get("messages", []) + + for msg in rest: + role = msg.get("role", "user") + content = msg.get("content") + + if isinstance(content, str): + messages.append({"role": role, "content": content}) + continue + if not isinstance(content, list): + continue + + if role == "assistant": + # collect thinking + text + tool_use blocks + text_parts, thinking, tool_calls = [], [], [] + for b in content: + t = b.get("type") + if t == "text": + text_parts.append(b.get("text", "")) + elif t == "thinking": + thinking.append(b.get("thinking", "")) + elif t == "tool_use": + tool_calls.append({ + "id": b.get("id", "call_" + uuid.uuid4().hex[:16]), + "type": "function", + "function": { + "name": b.get("name", ""), + "arguments": json.dumps(b.get("input", {}), ensure_ascii=False), + }, + }) + entry = {"role": "assistant"} + if thinking: + entry["reasoning_content"] = "".join(thinking) + if text_parts: + entry["content"] = "".join(text_parts) + else: + entry["content"] = "" + if tool_calls: + entry["tool_calls"] = tool_calls + messages.append(entry) + + elif role == "user": + # Preserve the original block order. Claude Code always sends + # tool_result after the tool_use it answers, and OpenAI requires + # tool messages to follow the assistant tool_calls they answer, + # so keeping the anthropic order verbatim is correct. Consecutive + # text blocks are merged into a single user message. + pending_text = [] + + def flush_text(): + if pending_text: + messages.append({"role": "user", "content": "".join(pending_text)}) + pending_text.clear() + + for b in content: + t = b.get("type") + if t == "text": + pending_text.append(b.get("text", "")) + elif t == "tool_result": + flush_text() + messages.append({ + "role": "tool", + "tool_call_id": b.get("tool_use_id", ""), + "content": b.get("content", ""), + }) + flush_text() + + else: + messages.append({"role": role, "content": content}) + + oai["messages"] = messages + + # tools: input_schema -> parameters, drop anthropic-only fields + if body.get("tools"): + tools = [] + for t in body["tools"]: + tools.append({ + "type": "function", + "function": { + "name": t.get("name", ""), + "description": t.get("description", ""), + "parameters": t.get("input_schema", {"type": "object", "properties": {}}), + }, + }) + oai["tools"] = tools + if body.get("tool_choice") == "none": + oai["tool_choice"] = "none" + + oai["stream"] = bool(body.get("stream")) + oai["max_tokens"] = int(body.get("max_tokens", 2048)) + # pass through the sampling params that share a name in both APIs, and + # map anthropic stop_sequences -> openai stop + for key in ("temperature", "top_p"): + if key in body: + oai[key] = body[key] + if body.get("stop_sequences"): + oai["stop"] = body["stop_sequences"] + return oai + + +def _is_cjk(ch): + """True for CJK ideographs, CJK punctuation, and fullwidth forms.""" + return ("\u4e00" <= ch <= "\u9fff" + or "\u3000" <= ch <= "\u303f" + or "\uff00" <= ch <= "\uffef") + + +def _text_token_estimate(text): + """Upper-bound token estimate for one text string. + + CJK characters are roughly 1 token each; latin text is roughly 1 token + per 3.5 chars. We return an over-estimate on purpose (see + estimate_input_tokens). + """ + cjk = other = 0 + for ch in text: + if _is_cjk(ch): + cjk += 1 + else: + other += 1 + return cjk + math.ceil(other / 3.5) + + +def estimate_input_tokens(body): + """Honest upper-bound estimate of the input token count for a + /v1/messages body, never an under-estimate. + + Claude Code uses /v1/messages/count_tokens for context-budget + management: under-reporting would let it exceed the real context + window, so we deliberately err high. This is a heuristic, not a real + tokenizer; the same function feeds count_tokens and the streaming + message_start usage so both always agree. + """ + tokens = 0 + system = body.get("system") + if isinstance(system, str): + tokens += _text_token_estimate(system) + elif isinstance(system, list): + for b in system: + if isinstance(b, dict) and b.get("type") == "text": + tokens += _text_token_estimate(b.get("text", "")) + + for msg in body.get("messages", []): + content = msg.get("content") + if isinstance(content, str): + tokens += _text_token_estimate(content) + continue + for b in content or []: + if not isinstance(b, dict): + continue + t = b.get("type") + if t in ("text", "thinking"): + tokens += _text_token_estimate(b.get("text") or b.get("thinking") or "") + elif t == "tool_result": + c = b.get("content", "") + tokens += _text_token_estimate(c if isinstance(c, str) else json.dumps(c)) + elif t == "tool_use": + tokens += _text_token_estimate(json.dumps(b.get("input", {}))) + + for t in body.get("tools", []): + tokens += _text_token_estimate(json.dumps(t.get("input_schema", {}))) + + return max(1, tokens) + + +def verify_master_key(headers, master_key): + """Check inbound credentials against the configured master key. + + Accepts `x-api-key: ` or `Authorization: Bearer `. The + comparison is constant-time; a missing or empty configured key always + rejects (fail closed). + """ + if not master_key: + return False + candidate = headers.get("x-api-key") + if candidate is None: + auth = headers.get("Authorization", "") + if auth.startswith("Bearer "): + candidate = auth[len("Bearer "):] + if candidate is None: + return False + return hmac.compare_digest(candidate, master_key) + + +def looks_like_classifier_request(body): + """Heuristic for Claude Code's harm-classifier probe. + + Claude Code asks a separate model name (e.g. "claude-sonnet-5") a + single-message, no-tool, non-streaming question to judge whether a + Bash/WebFetch/WebSearch action is safe. Our backend is not a + classifier, so those probes misclassify harmless commands. This is + still a heuristic — see --allow-classifier-bypass in main() — and is + only consulted when the operator explicitly enables the bypass. + """ + msgs = body.get("messages") + shape_matches = (body.get("model", "") != "local" + and not body.get("tools") + and not body.get("stream")) + single_user_message = (isinstance(msgs, list) and len(msgs) == 1 + and isinstance(msgs[0], dict) + and msgs[0].get("role") == "user") + return shape_matches and single_user_message + +# -------------------------------------------------------------------------- +# response: baseRT chat/completions stream -> anthropic SSE +# -------------------------------------------------------------------------- + +def _fmt_sse(event, data): + """Format one anthropic streaming event as an SSE block.""" + return f"event: {event}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n" + + +class AnthropicStreamer: + """ + Streaming openai-chunk -> anthropic-SSE state machine. + + Block index layout (llama.cpp to_json_anthropic): thinking (0) -> + text (0 or 1) -> tool_use (n+). A block's start/stop events are emitted + lazily: content_block_start fires on the first delta of that kind, and + finish() closes every opened block in index order before message_delta / + message_stop. baseRT already splits reasoning_content / content, so no + chat-template reasoning detection is needed on our side. + """ + + def __init__(self, msg_id, model, input_tokens=0): + self.msg_id = msg_id + self.model = model + self.input_tokens = input_tokens + self.output_tokens = 0 + self.output_tokens_estimate = 0 # token estimate of emitted text/thinking, + # used when upstream never sends usage + self.thinking_started = False + self.text_started = False + self.text_index = None + self.tool_states = {} # index -> {"id","name","arguments"} + self.tool_states_started = {} # index -> block_start emitted + self.stop_reason = "end_turn" + + def feed(self, chunk): + """Consume one parsed openai chunk; returns SSE event strings.""" + ev = [] + if chunk.get("usage"): + self.output_tokens = chunk["usage"].get("completion_tokens", self.output_tokens) + choices = chunk.get("choices") or [] + if not choices: + return ev + delta = choices[0].get("delta") or {} + finish = choices[0].get("finish_reason") + + thinking = delta.get("reasoning_content") + if thinking: + self.output_tokens_estimate += _text_token_estimate(thinking) + if not self.thinking_started: + ev.append(_fmt_sse("content_block_start", { + "type": "content_block_start", + "index": 0, + "content_block": {"type": "thinking", "thinking": ""}, + })) + self.thinking_started = True + ev.append(_fmt_sse("content_block_delta", { + "type": "content_block_delta", + "index": 0, + "delta": {"type": "thinking_delta", "thinking": thinking}, + })) + + text = delta.get("content") + if text: + self.output_tokens_estimate += _text_token_estimate(text) + if not self.text_started: + self.text_index = 1 if self.thinking_started else 0 + ev.append(_fmt_sse("content_block_start", { + "type": "content_block_start", + "index": self.text_index, + "content_block": {"type": "text", "text": ""}, + })) + self.text_started = True + ev.append(_fmt_sse("content_block_delta", { + "type": "content_block_delta", + "index": self.text_index, + "delta": {"type": "text_delta", "text": text}, + })) + + for tc in delta.get("tool_calls") or []: + idx = tc.get("index", 0) + state = self.tool_states.setdefault( + idx, {"id": "call_" + uuid.uuid4().hex[:16], "name": "", "arguments": ""}) + if tc.get("id"): + state["id"] = tc["id"] + fn = tc.get("function") or {} + if fn.get("name"): + state["name"] = fn["name"] + if fn.get("arguments"): + state["arguments"] += fn["arguments"] + # tool blocks come after thinking (0) and/or text (0/1) + tool_index = ((1 if self.thinking_started else 0) + + (1 if self.text_started else 0) + idx) + if not self.tool_states_started.get(idx): + ev.append(_fmt_sse("content_block_start", { + "type": "content_block_start", + "index": tool_index, + "content_block": {"type": "tool_use", "id": state["id"], + "name": state["name"], "input": {}}, + })) + self.tool_states_started[idx] = True + if fn.get("arguments"): + ev.append(_fmt_sse("content_block_delta", { + "type": "content_block_delta", + "index": tool_index, + "delta": {"type": "input_json_delta", "partial_json": fn["arguments"]}, + })) + + if finish: + if finish == "tool_calls": + self.stop_reason = "tool_use" + elif finish == "length": + # openai uses "length" for max_tokens truncation + self.stop_reason = "max_tokens" + else: + self.stop_reason = "end_turn" + return ev + + def finish(self): + """Close open blocks and emit message_delta / message_stop.""" + ev = [] + # close open blocks in index order (same index formula as feed()) + if self.thinking_started: + ev.append(_fmt_sse("content_block_stop", + {"type": "content_block_stop", "index": 0})) + if self.text_started: + ev.append(_fmt_sse("content_block_stop", + {"type": "content_block_stop", "index": self.text_index})) + base_tool_index = ((1 if self.thinking_started else 0) + + (1 if self.text_started else 0)) + for idx in self.tool_states: + ev.append(_fmt_sse("content_block_stop", + {"type": "content_block_stop", + "index": base_tool_index + idx})) + # report real upstream usage when available, otherwise an honest + # estimate of what we emitted (never a silent 0) + out_tokens = self.output_tokens or self.output_tokens_estimate + ev.append(_fmt_sse("message_delta", { + "type": "message_delta", + "delta": {"stop_reason": self.stop_reason, "stop_sequence": None}, + "usage": {"output_tokens": out_tokens}, + })) + ev.append(_fmt_sse("message_stop", {"type": "message_stop"})) + return ev + + def initial(self): + return [_fmt_sse("message_start", { + "type": "message_start", + "message": { + "id": self.msg_id, + "type": "message", + "role": "assistant", + "content": [], + "model": self.model, + "stop_reason": None, + "stop_sequence": None, + "usage": {"input_tokens": self.input_tokens, "output_tokens": 0}, + }, + })] + + +def build_nonstream_message(msg_id, model, prompt_tokens, output_tokens, data): + """ + Assemble the non-streaming anthropic message from a chat.completion object. + + This path exists because the client may fall back to non-streaming (Claude + Code does when a streaming request stalls); LiteLLM's equivalent conversion + returns empty content (BerriAI/litellm#27492), so we build the blocks here: + thinking (empty signature, like llama.cpp) + text + tool_use (input parsed + as JSON, {} on parse failure). + """ + choice = (data.get("choices") or [{}])[0] + message = choice.get("message") or {} + finish = choice.get("finish_reason") + + stop_reason = "end_turn" + if finish == "tool_calls": + stop_reason = "tool_use" + elif finish == "length": + # openai uses "length" for max_tokens truncation + stop_reason = "max_tokens" + + blocks = [] + reasoning = message.get("reasoning_content") + if reasoning: + blocks.append({"type": "thinking", "thinking": reasoning, "signature": ""}) + text = message.get("content") + if text: + blocks.append({"type": "text", "text": text}) + for tc in message.get("tool_calls") or []: + fn = tc.get("function") or {} + try: + args = json.loads(fn.get("arguments") or "{}") + except Exception: + log.warning("tool arguments are not valid JSON, substituting {}: %r", + (fn.get("arguments") or "")[:200]) + args = {} + blocks.append({"type": "tool_use", + "id": tc.get("id", "call_" + uuid.uuid4().hex[:16]), + "name": fn.get("name", ""), "input": args}) + + return { + "id": msg_id, + "type": "message", + "role": "assistant", + "content": blocks, + "model": model, + "stop_reason": stop_reason, + "stop_sequence": None, + "usage": {"input_tokens": prompt_tokens, "output_tokens": output_tokens}, + } + + +# -------------------------------------------------------------------------- +# HTTP layer +# -------------------------------------------------------------------------- + +def _write_req_log(req_log, body, peer, path): + with open(req_log, "w") as f: + f.write(f"# {time.strftime('%Y-%m-%d %H:%M:%S')} peer={peer} path={path}\n") + f.write(json.dumps(body, ensure_ascii=False, indent=1)) + + +def _write_res_log(req_log, msg): + with open(req_log.replace("-req.json", "-res.json"), "w") as f: + f.write("# assembled anthropic response (what the client received)\n") + f.write(json.dumps(msg, ensure_ascii=False, indent=1)) + + +async def handle_messages(request): + t0 = time.monotonic() + + # Authentication first: reject before touching the body, so unauthenticated + # probing cannot even exercise the JSON parser. + if not verify_master_key(request.headers, request.app["master_key"]): + log.warning("rejected request without valid credentials from %s", request.remote) + return _auth_error() + + try: + body = await request.json() + except Exception: + return web.json_response({"type": "error", "error": { + "type": "invalid_request_error", + "message": "invalid JSON body"}}, status=400) + + if not isinstance(body, dict) or not body.get("messages"): + return web.json_response({"type": "error", "error": { + "type": "invalid_request_error", + "message": "'messages' is required"}}, status=400) + + client_model = body.get("model", "?") + msg_id = "msg_" + uuid.uuid4().hex + req_id = msg_id[4:12] + req_log = None + if request.app["debug_dir"]: + peer = request.remote.replace(":", "_") + req_log = os.path.join(request.app["debug_dir"], + f"{int(time.time()*1000):016d}-{peer}-{req_id}-req.json") + await asyncio.to_thread(_write_req_log, req_log, body, request.remote, request.path) + + oai = convert_anthropic_to_oai(body, + merge_system=not request.app["no_system_merge"]) + oai["model"] = request.app["model"] # upstream backend model id + if request.app["max_tokens_override"]: + oai["max_tokens"] = request.app["max_tokens_override"] # block client budget + + log.info("req=%s --> POST /v1/messages model=%s %s max_tokens=%s", + req_id, client_model, "stream" if oai["stream"] else "non-stream", + oai["max_tokens"]) + + # Always echo "local" regardless of what the client sent. Claude Code + # uses multiple model names internally for routing (e.g. "claude-sonnet-5" + # for its harm classifier). If we echo different names CC will treat them + # as separate backends and expect classifier-format responses from the + # classifier-named one. A single fixed alias prevents CC from discovering + # multiple models through our adapter. + model_name = "local" + + # Harm-classifier bypass. Claude Code asks a separate model name (e.g. + # "claude-sonnet-5") a single-message, no-tool, non-streaming question to + # judge whether a Bash/WebFetch/WebSearch action is safe. Our backend is + # not a classifier and produces false positives that block harmless + # commands. This short-circuit returns a fixed "allow" response so CC + # never blocks on a misclassification. Security implication: it disables + # CC's online harm detection — the operator is responsible for what the + # agent executes. It is OFF by default and must be enabled explicitly + # with --allow-classifier-bypass; see TECHNICAL.md "Classifier bypass". + if request.app["allow_classifier_bypass"] and looks_like_classifier_request(body): + ms = int((time.monotonic() - t0) * 1000) + log.warning("req=%s <-- 200 classifier-bypass (enabled) %dms", req_id, ms) + return web.json_response({ + "id": msg_id, + "type": "message", + "role": "assistant", + "model": model_name, + "content": [{"type": "text", "text": "no"}], + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": {"input_tokens": 0, "output_tokens": 0}, + }) + + headers = {"Content-Type": "application/json"} + headers["Authorization"] = f"Bearer {request.app['master_key']}" + + # baseRT is single-slot: when it is busy (e.g. a long prefill), further + # connections queue server-side. Without a timeout the handler would hang + # forever waiting for response headers. sock_read must exceed the longest + # silent gap in a stream (prefill of a huge prompt ~35s). + try: + async with request.app["session"].post( + request.app["upstream"] + "/chat/completions", + json=oai, headers=headers) as up: + if up.status != 200: + err = await up.text() + ms = int((time.monotonic() - t0) * 1000) + log.warning("req=%s <-- %s upstream %dms", req_id, up.status, ms) + return web.json_response({"type": "error", "error": { + "type": "api_error", + "message": f"upstream {up.status}: {err[:500]}"}}, status=502) + + if oai["stream"]: + return await _stream_response(request, up, msg_id, model_name, t0, req_log, + input_tokens=estimate_input_tokens(body)) + data = await up.json() + usage = data.get("usage") or {} + msg = build_nonstream_message(msg_id, model_name, + usage.get("prompt_tokens", 0), + usage.get("completion_tokens", 0), data) + ms = int((time.monotonic() - t0) * 1000) + pt = usage.get("prompt_tokens", "?") + ct = usage.get("completion_tokens", "?") + finish = data.get("choices", [{}])[0].get("finish_reason", "stop") + log.info("req=%s <-- 200 prompt=%s completion=%s %dms %s", req_id, pt, ct, ms, finish) + if req_log: + await asyncio.to_thread(_write_res_log, req_log, msg) + return web.json_response(msg) + except aiohttp.ClientError as e: + ms = int((time.monotonic() - t0) * 1000) + log.warning("req=%s <-- 502 upstream-error %dms: %s", req_id, ms, e) + return web.json_response({"type": "error", "error": { + "type": "api_error", + "message": f"upstream error: {e}"}}, status=502) + + +async def _stream_response(request, up, msg_id, model_name, t0, req_log=None, input_tokens=0): + resp = web.StreamResponse(status=200, headers={ + "Content-Type": "text/event-stream", + "Cache-Control": "no-cache", + "Connection": "keep-alive", + "X-Accel-Buffering": "no", + }) + await resp.prepare(request) + + res_log = None + if req_log: + res_log = open(req_log.replace("-req.json", "-res.sse"), "w") + res_log.write(f"# {time.strftime('%Y-%m-%d %H:%M:%S')} stream response\n") + + streamer = AnthropicStreamer(msg_id, model_name, input_tokens=input_tokens) + sent_initial = False + last_ping = time.monotonic() + + async def send_initial(): + """Emit message_start; must lead every anthropic stream.""" + nonlocal sent_initial + if sent_initial: + return + for e in streamer.initial(): + await resp.write(e.encode()) + if res_log: + res_log.write(e + "\n") + sent_initial = True + + async def ping_task(): + nonlocal last_ping + while True: + await asyncio.sleep(PING_INTERVAL) + if time.monotonic() - last_ping >= PING_INTERVAL: + try: + await resp.write(b":\n\n") + last_ping = time.monotonic() + except Exception: + return + + ping_handle = asyncio.create_task(ping_task()) + + def client_gone(): + """Probe the client connection without blocking (llama.cpp's + is_connection_closed: recv(1, MSG_PEEK), EOF == gone).""" + transport = request.transport + if transport is None or transport.is_closing(): + return True + sock = transport.get_extra_info("socket") + if sock is None: + return False + # recent aiohttp / Python 3.14 may hand us a TransportSocket wrapper + # instead of the raw socket; dig through. Never call setblocking() + # here — the transport socket is already non-blocking, and mutating + # state the event loop owns has broken across versions before. + raw = getattr(sock, "_sock", sock) + try: + return raw.recv(1, socket.MSG_PEEK) == b"" + except BlockingIOError: + return False # socket open, no pending EOF + except OSError: + return True + + # aiohttp's StreamReader.__aiter__ returns an AsyncStreamIterator which + # carries the actual __anext__; the reader itself has no __anext__. + upstream_iter = up.content.__aiter__() + + try: + while True: + # wait one chunk from upstream, at most 1s (llama.cpp: HTTP_POLLING_SECONDS) + try: + line = await asyncio.wait_for(upstream_iter.__anext__(), timeout=1.0) + except asyncio.TimeoutError: + # no new data this cycle: check whether the client is gone + if client_gone(): + up.close() # abort upstream so baseRT stops generating + if res_log: + res_log.write("# client disconnected, upstream aborted\n") + break + continue + except StopAsyncIteration: + break + + line = line.strip() + if not line.startswith(b"data:"): + continue + payload = line[5:].strip() + if payload == b"[DONE]": + break + try: + chunk = json.loads(payload) + except json.JSONDecodeError: + continue + + await send_initial() + + for e in streamer.feed(chunk): + await resp.write(e.encode()) + if res_log: + res_log.write(e + "\n") + last_ping = time.monotonic() + + # the stream may have ended before any content arrived (immediate + # [DONE] or early close): message_start must still lead the sequence + await send_initial() + for e in streamer.finish(): + await resp.write(e.encode()) + if res_log: + res_log.write(e + "\n") + ms = int((time.monotonic() - t0) * 1000) + out_tokens = streamer.output_tokens or streamer.output_tokens_estimate + log.info("req=%s <-- 200 completion=%s %dms %s", + msg_id[4:12], out_tokens, ms, streamer.stop_reason) + except (ConnectionResetError, aiohttp.ClientConnectionError): + up.close() + ms = int((time.monotonic() - t0) * 1000) + log.warning("req=%s <-- client-disconnected %dms, upstream aborted", msg_id[4:12], ms) + if res_log: + res_log.write("# client disconnected mid-write, upstream aborted\n") + finally: + ping_handle.cancel() + if res_log: + res_log.close() + try: + await resp.write_eof() + except (ConnectionResetError, aiohttp.ClientConnectionError): + pass # client already gone; nothing more to deliver + return resp + + +def _auth_error(): + """Anthropic-style 401, which Claude Code recognises as a credential + problem rather than a server fault.""" + return web.json_response({"type": "error", "error": { + "type": "authentication_error", + "message": "invalid x-api-key"}}, status=401) + + +async def handle_api_hello(request): + return web.json_response({"message": "Hello"}) + +async def handle_health(request): + return web.json_response({"status": "healthy"}) + +async def handle_count_tokens(request): + if not verify_master_key(request.headers, request.app["master_key"]): + return _auth_error() + try: + body = await request.json() + except Exception: + body = {} + if not isinstance(body, dict): + body = {} + return web.json_response({"input_tokens": estimate_input_tokens(body)}) + + +def main(): + ap = argparse.ArgumentParser(description="Native Anthropic Messages adapter for baseRT") + ap.add_argument("--host", default="127.0.0.1", + help="bind address (default 127.0.0.1; use 0.0.0.0 only for " + "cross-machine deployments, together with --master-key)") + ap.add_argument("--port", type=int, default=4003) + ap.add_argument("--upstream", default="http://127.0.0.1:7999/v1") + ap.add_argument("--master-key", default="", + help="shared secret required on every inbound /v1/messages " + "request (x-api-key or Authorization: Bearer); it is also " + "sent upstream as a Bearer token. Required.") + ap.add_argument("--model", default="", + help="backend model id known to baseRT. Required.") + ap.add_argument("--max-tokens-override", type=int, default=0, + help="ignore the client's max_tokens and use this fixed budget " + "(Claude Code sends small budgets like 64/8192 for tool-loop and " + "background calls; a heavy-thinking local model truncates on them). " + "0 = pass the client value through") + ap.add_argument("--debug-dir", default="", + help="save every incoming request body and the response sent to the " + "client as files here (req.json / res.json / res.sse). " + "Off by default; enabling it writes full prompts to disk") + ap.add_argument("--no-system-merge", action="store_true", + help="skip system-block merging (use when the model's chat template " + "already handles mid-list system messages, e.g. froggeric v21.3)") + ap.add_argument("--allow-classifier-bypass", action="store_true", + help="short-circuit requests that look like Claude Code's harm-" + "classifier probes with a fixed allow response. This disables " + "Claude Code's online harm detection — enable only if you " + "understand the consequences (see TECHNICAL.md). Off by default") + ap.add_argument("--log-level", default="info", + choices=["debug", "info", "warning", "error"]) + args = ap.parse_args() + if not args.model: + ap.error("--model is required (backend model id known to baseRT)") + if not args.master_key: + ap.error("--master-key is required (inbound requests are rejected without it)") + + logging.basicConfig(level=getattr(logging, args.log_level.upper()), + format="%(asctime)s %(levelname)s %(message)s", + datefmt="%H:%M:%S") + + app = web.Application() + app["upstream"] = args.upstream + app["master_key"] = args.master_key + app["model"] = args.model + app["max_tokens_override"] = args.max_tokens_override + app["debug_dir"] = args.debug_dir + app["no_system_merge"] = args.no_system_merge + app["allow_classifier_bypass"] = args.allow_classifier_bypass + app.router.add_post("/v1/messages", handle_messages) + app.router.add_get("/api/hello", handle_api_hello) + app.router.add_get("/health", handle_health) + app.router.add_post("/v1/messages/count_tokens", handle_count_tokens) + + async def on_startup(app): + if app["debug_dir"]: + os.makedirs(app["debug_dir"], exist_ok=True) + # one shared session for all requests, created inside the event loop + app["session"] = aiohttp.ClientSession( + timeout=aiohttp.ClientTimeout(sock_connect=10, sock_read=600)) + + async def on_shutdown(app): + await app["session"].close() + + app.on_startup.append(on_startup) + app.on_shutdown.append(on_shutdown) + + log.info("anthropic adapter %s:%s -> %s (model=%s)", args.host, args.port, + args.upstream, args.model) + web.run_app(app, host=args.host, port=args.port, print=None) + + +if __name__ == "__main__": + main() diff --git a/cc-bridge/pytest.ini b/cc-bridge/pytest.ini new file mode 100644 index 0000000..3c6c52f --- /dev/null +++ b/cc-bridge/pytest.ini @@ -0,0 +1,4 @@ +[pytest] +asyncio_mode = auto +pythonpath = .. +testpaths = tests diff --git a/cc-bridge/requirements-dev.txt b/cc-bridge/requirements-dev.txt new file mode 100644 index 0000000..7df8cd2 --- /dev/null +++ b/cc-bridge/requirements-dev.txt @@ -0,0 +1,3 @@ +-r requirements.txt +pytest>=8.0 +pytest-asyncio>=0.24 diff --git a/cc-bridge/requirements.txt b/cc-bridge/requirements.txt new file mode 100644 index 0000000..6a37f7b --- /dev/null +++ b/cc-bridge/requirements.txt @@ -0,0 +1 @@ +aiohttp>=3.9 diff --git a/cc-bridge/tests/conftest.py b/cc-bridge/tests/conftest.py new file mode 100644 index 0000000..2bb6b54 --- /dev/null +++ b/cc-bridge/tests/conftest.py @@ -0,0 +1,83 @@ +"""Shared fixtures: a started adapter server backed by a fake upstream. + +The fake upstream speaks OpenAI chat/completions and is fully configurable +per test (default non-streaming JSON, or any custom SSE byte stream). +""" +import aiohttp +import pytest +from aiohttp import web +from aiohttp.test_utils import TestServer + +import anthropic_adapter as adapter + + +async def _default_upstream(request): + """Non-streaming chat.completion; streaming tests override the handler.""" + body = await request.json() + if body.get("stream"): + resp = web.StreamResponse(status=200, + headers={"Content-Type": "text/event-stream"}) + await resp.prepare(request) + await resp.write(b'data: {"choices":[{"index":0,"delta":' + b'{"role":"assistant","content":"hi"}}]}\n\n') + await resp.write(b'data: {"choices":[{"index":0,"delta":{},' + b'"finish_reason":"stop"}]}\n\n') + await resp.write(b"data: [DONE]\n\n") + return resp + return web.json_response({ + "id": "chatcmpl-test", + "choices": [{"index": 0, + "message": {"role": "assistant", "content": "hi"}, + "finish_reason": "stop"}], + "usage": {"prompt_tokens": 10, "completion_tokens": 2}, + }) + + +async def _adapter_startup(app): + app["session"] = aiohttp.ClientSession() + + +async def _adapter_shutdown(app): + await app["session"].close() + + +@pytest.fixture +async def make_bridge(): + """Build and start (adapter server, fake upstream server). + + Returns an async factory; app overrides (e.g. allow_classifier_bypass, + debug_dir) are passed as keyword arguments. + """ + created = [] + + async def build(upstream_handler=None, **app_overrides): + up_app = web.Application() + up_app.router.add_post("/v1/chat/completions", + upstream_handler or _default_upstream) + up_server = TestServer(up_app) + await up_server.start_server() + created.append(up_server) + + app = web.Application() + app["upstream"] = f"http://127.0.0.1:{up_server.port}/v1" + app["master_key"] = "test-key" + app["model"] = "backend-model" + app["max_tokens_override"] = 0 + app["debug_dir"] = "" + app["no_system_merge"] = False + app["allow_classifier_bypass"] = False + app.update(app_overrides) + app.router.add_post("/v1/messages", adapter.handle_messages) + app.router.add_post("/v1/messages/count_tokens", adapter.handle_count_tokens) + app.on_startup.append(_adapter_startup) + app.on_shutdown.append(_adapter_shutdown) + + server = TestServer(app) + await server.start_server() + created.append(server) + return server + + yield build + + for s in reversed(created): + await s.close() diff --git a/cc-bridge/tests/helpers.py b/cc-bridge/tests/helpers.py new file mode 100644 index 0000000..24e38b9 --- /dev/null +++ b/cc-bridge/tests/helpers.py @@ -0,0 +1,28 @@ +"""Small test helpers shared by the test modules.""" + +import json + + +def split_sse(raw: bytes) -> list[tuple[str, dict]]: + """Split a raw SSE byte stream into (event, data) pairs, skipping + comment pings (lines with no data: field).""" + events = [] + for block in raw.decode("utf-8").split("\n\n"): + event, data = "message", None + for line in block.splitlines(): + if line.startswith("event:"): + event = line[6:].strip() + elif line.startswith("data:"): + data = line[5:].strip() + if data is None: + continue + events.append((event, json.loads(data))) + return events + + +def sse_bytes(*chunks: dict) -> bytes: + """Serialize openai-style chunks as SSE data lines.""" + out = b"" + for c in chunks: + out += b"data: " + json.dumps(c).encode("utf-8") + b"\n\n" + return out diff --git a/cc-bridge/tests/test_auth.py b/cc-bridge/tests/test_auth.py new file mode 100644 index 0000000..29322ef --- /dev/null +++ b/cc-bridge/tests/test_auth.py @@ -0,0 +1,83 @@ +"""Inbound authentication: pure function + endpoint level.""" + +import aiohttp + +from anthropic_adapter import verify_master_key + + +def test_verify_missing_key_rejected(): + assert verify_master_key({}, "k") is False + assert verify_master_key({"Content-Type": "application/json"}, "k") is False + + +def test_verify_x_api_key(): + assert verify_master_key({"x-api-key": "k"}, "k") is True + assert verify_master_key({"x-api-key": "wrong"}, "k") is False + + +def test_verify_bearer(): + assert verify_master_key({"Authorization": "Bearer k"}, "k") is True + assert verify_master_key({"Authorization": "Bearer wrong"}, "k") is False + assert verify_master_key({"Authorization": "Basic abc"}, "k") is False + + +def test_verify_empty_configured_key_fails_closed(): + assert verify_master_key({"x-api-key": "k"}, "") is False + assert verify_master_key({"x-api-key": "k"}, None) is False + + +async def test_endpoint_401_without_credentials(make_bridge): + server = await make_bridge() + async with aiohttp.ClientSession() as s: + async with s.post(server.make_url("/v1/messages"), + json={"model": "local", + "messages": [{"role": "user", "content": "hi"}]}) as resp: + assert resp.status == 401 + body = await resp.json() + assert body["error"]["type"] == "authentication_error" + + +async def test_endpoint_401_with_wrong_key(make_bridge): + server = await make_bridge() + async with aiohttp.ClientSession() as s: + async with s.post(server.make_url("/v1/messages"), + json={"messages": [{"role": "user", "content": "hi"}]}, + headers={"x-api-key": "wrong"}) as resp: + assert resp.status == 401 + + +async def test_401_before_body_parsing(make_bridge): + """Bad JSON without credentials must 401, not 400 (auth precedes parsing).""" + server = await make_bridge() + async with aiohttp.ClientSession() as s: + async with s.post(server.make_url("/v1/messages"), + data=b"{not json", headers={"content-type": "application/json"}) as resp: + assert resp.status == 401 + + +async def test_count_tokens_requires_auth(make_bridge): + server = await make_bridge() + async with aiohttp.ClientSession() as s: + async with s.post(server.make_url("/v1/messages/count_tokens"), + json={"messages": [{"role": "user", "content": "hi"}]}) as resp: + assert resp.status == 401 + async with s.post(server.make_url("/v1/messages/count_tokens"), + json={"messages": [{"role": "user", "content": "hi"}]}, + headers={"x-api-key": "test-key"}) as resp: + assert resp.status == 200 + body = await resp.json() + assert body["input_tokens"] >= 1 + + +async def test_authenticated_request_reaches_upstream(make_bridge): + server = await make_bridge() + async with aiohttp.ClientSession() as s: + async with s.post(server.make_url("/v1/messages"), + json={"model": "local", + "messages": [{"role": "user", "content": "hi"}]}, + headers={"x-api-key": "test-key"}) as resp: + assert resp.status == 200 + body = await resp.json() + assert body["type"] == "message" + assert body["content"][0]["type"] == "text" + assert body["usage"]["input_tokens"] == 10 # from fake upstream usage diff --git a/cc-bridge/tests/test_convert.py b/cc-bridge/tests/test_convert.py new file mode 100644 index 0000000..6fcfb87 --- /dev/null +++ b/cc-bridge/tests/test_convert.py @@ -0,0 +1,141 @@ +"""Golden cases for anthropic -> openai request conversion.""" + +from anthropic_adapter import convert_anthropic_to_oai + + +def test_system_string_merged_to_front(): + oai = convert_anthropic_to_oai({ + "system": "top-level", + "messages": [ + {"role": "user", "content": "hi"}, + ], + }) + assert oai["messages"] == [ + {"role": "system", "content": "top-level"}, + {"role": "user", "content": "hi"}, + ] + + +def test_system_text_blocks_and_midlist_system_merged(): + oai = convert_anthropic_to_oai({ + "system": [{"type": "text", "text": "A"}, {"type": "text", "text": "B"}], + "messages": [ + {"role": "user", "content": "first"}, + {"role": "system", "content": "mid reminder"}, + {"role": "user", "content": "second"}, + ], + }) + roles = [m["role"] for m in oai["messages"]] + assert roles == ["system", "user", "user"] + assert oai["messages"][0]["content"] == "A\nB\nmid reminder" + + +def test_no_system_merge_keeps_positions(): + oai = convert_anthropic_to_oai({ + "messages": [ + {"role": "user", "content": "first"}, + {"role": "system", "content": "mid"}, + {"role": "user", "content": "last"}, + ], + }, merge_system=False) + roles = [m["role"] for m in oai["messages"]] + assert roles == ["user", "system", "user"] + + +def test_assistant_thinking_and_tool_use(): + oai = convert_anthropic_to_oai({ + "messages": [{ + "role": "assistant", + "content": [ + {"type": "thinking", "thinking": "let me think"}, + {"type": "text", "text": "answer"}, + {"type": "tool_use", "id": "tu_1", "name": "bash", + "input": {"cmd": "ls"}}, + ], + }], + }) + msg = oai["messages"][0] + assert msg["reasoning_content"] == "let me think" + assert msg["content"] == "answer" + assert msg["tool_calls"][0]["id"] == "tu_1" + assert msg["tool_calls"][0]["function"]["name"] == "bash" + assert msg["tool_calls"][0]["function"]["arguments"] == '{"cmd": "ls"}' + + +def test_user_tool_result_after_text(): + oai = convert_anthropic_to_oai({ + "messages": [{ + "role": "user", + "content": [ + {"type": "text", "text": "please run it"}, + {"type": "tool_result", "tool_use_id": "tu_1", "content": "ok"}, + ], + }], + }) + assert [m["role"] for m in oai["messages"]] == ["user", "tool"] + assert oai["messages"][1]["tool_call_id"] == "tu_1" + + +def test_user_text_between_tool_results_keeps_order(): + oai = convert_anthropic_to_oai({ + "messages": [{ + "role": "user", + "content": [ + {"type": "tool_result", "tool_use_id": "tu_1", "content": "a"}, + {"type": "text", "text": "now try b"}, + {"type": "tool_result", "tool_use_id": "tu_2", "content": "b"}, + {"type": "text", "text": "and c"}, + ], + }], + }) + assert [m["role"] for m in oai["messages"]] == ["tool", "user", "tool", "user"] + assert oai["messages"][0]["tool_call_id"] == "tu_1" + assert oai["messages"][2]["tool_call_id"] == "tu_2" + + +def test_consecutive_text_blocks_merged(): + oai = convert_anthropic_to_oai({ + "messages": [{ + "role": "user", + "content": [ + {"type": "text", "text": "one "}, + {"type": "text", "text": "two"}, + ], + }], + }) + assert len(oai["messages"]) == 1 + assert oai["messages"][0]["content"] == "one two" + + +def test_tools_mapping_and_tool_choice_none(): + oai = convert_anthropic_to_oai({ + "tools": [{"name": "f", "description": "d", + "input_schema": {"type": "object", "properties": {"x": {"type": "string"}}}}], + "tool_choice": "none", + "messages": [{"role": "user", "content": "hi"}], + }) + assert oai["tools"][0]["function"]["name"] == "f" + assert oai["tools"][0]["function"]["parameters"]["properties"]["x"]["type"] == "string" + assert oai["tool_choice"] == "none" + + +def test_sampling_params_passthrough(): + oai = convert_anthropic_to_oai({ + "temperature": 0.7, + "top_p": 0.9, + "stop_sequences": ["END", "STOP"], + "messages": [{"role": "user", "content": "hi"}], + }) + assert oai["temperature"] == 0.7 + assert oai["top_p"] == 0.9 + assert oai["stop"] == ["END", "STOP"] + + +def test_defaults(): + oai = convert_anthropic_to_oai({ + "messages": [{"role": "user", "content": "hi"}], + }) + assert oai["stream"] is False + assert oai["max_tokens"] == 2048 + assert "temperature" not in oai + assert "stop" not in oai diff --git a/cc-bridge/tests/test_estimate.py b/cc-bridge/tests/test_estimate.py new file mode 100644 index 0000000..2da2290 --- /dev/null +++ b/cc-bridge/tests/test_estimate.py @@ -0,0 +1,51 @@ +"""estimate_input_tokens: honest upper bound, monotonic, never zero.""" + +from anthropic_adapter import estimate_input_tokens + + +def test_never_zero(): + assert estimate_input_tokens({}) >= 1 + assert estimate_input_tokens({"messages": []}) >= 1 + + +def test_ascii_estimate(): + body = {"messages": [{"role": "user", "content": "a" * 1000}]} + est = estimate_input_tokens(body) + # upper bound: 1000 ascii chars / 3.5 per token + assert 1 <= est <= 1000 + + +def test_cjk_heavier_than_ascii(): + body_cjk = {"messages": [{"role": "user", "content": "中" * 100}]} + body_ascii = {"messages": [{"role": "user", "content": "x" * 100}]} + assert estimate_input_tokens(body_cjk) > estimate_input_tokens(body_ascii) + + +def test_monotonic_with_more_content(): + small = {"messages": [{"role": "user", "content": "hello"}]} + big = {"messages": [{"role": "user", "content": "hello"}, + {"role": "assistant", "content": "world"}]} + assert estimate_input_tokens(big) > estimate_input_tokens(small) + + +def test_tools_schema_counts(): + no_tools = {"messages": [{"role": "user", "content": "hi"}]} + with_tools = { + "messages": [{"role": "user", "content": "hi"}], + "tools": [{"name": "f", + "input_schema": {"type": "object", + "properties": {"a": {"type": "string"}, + "b": {"type": "integer"}}}}], + } + assert estimate_input_tokens(with_tools) > estimate_input_tokens(no_tools) + + +def test_tool_result_and_thinking_counted(): + base = {"messages": [{"role": "user", "content": "hi"}]} + tool = {"messages": [{"role": "user", + "content": [{"type": "tool_result", "tool_use_id": "t", + "content": "output " * 50}]}]} + think = {"messages": [{"role": "assistant", + "content": [{"type": "thinking", "thinking": "deep " * 50}]}]} + assert estimate_input_tokens(tool) > estimate_input_tokens(base) + assert estimate_input_tokens(think) > estimate_input_tokens(base) diff --git a/cc-bridge/tests/test_sse.py b/cc-bridge/tests/test_sse.py new file mode 100644 index 0000000..bc9e0db --- /dev/null +++ b/cc-bridge/tests/test_sse.py @@ -0,0 +1,188 @@ +"""End-to-end SSE lifecycle through the real HTTP stack.""" + +import aiohttp +from aiohttp import web + +from tests.helpers import split_sse, sse_bytes + + +async def test_stream_normal_lifecycle(make_bridge): + async def upstream(request): + resp = web.StreamResponse(status=200, + headers={"Content-Type": "text/event-stream"}) + await resp.prepare(request) + await resp.write(sse_bytes( + {"choices": [{"delta": {"reasoning_content": "hmm"}}]}, + {"choices": [{"delta": {"content": "hi"}}]}, + {"choices": [{"delta": {}, "finish_reason": "stop"}]}, + )) + await resp.write(b"data: [DONE]\n\n") + return resp + + server = await make_bridge(upstream_handler=upstream) + async with aiohttp.ClientSession() as s: + async with s.post(server.make_url("/v1/messages"), + json={"model": "local", "stream": True, + "messages": [{"role": "user", "content": "hi"}]}, + headers={"x-api-key": "test-key"}) as resp: + assert resp.status == 200 + raw = await resp.read() + + events = split_sse(raw) + kinds = [e for e, _ in events] + assert kinds[0] == "message_start" + assert "content_block_start" in kinds + assert kinds[-1] == "message_stop" + # anthropic streams must not pass [DONE] through + assert b"[DONE]" not in raw + # message_start carries the honest input estimate + assert events[0][1]["message"]["usage"]["input_tokens"] >= 1 + + # block indices: thinking 0, text 1 (thinking already started) + starts = [d for e, d in events if e == "content_block_start"] + assert [d["index"] for d in starts] == [0, 1] + assert [d["content_block"]["type"] for d in starts] == ["thinking", "text"] + + +async def test_stream_message_start_leading_on_immediate_done(make_bridge): + """Upstream sends [DONE] immediately: message_start must still lead.""" + async def upstream(request): + resp = web.StreamResponse(status=200, + headers={"Content-Type": "text/event-stream"}) + await resp.prepare(request) + await resp.write(b"data: [DONE]\n\n") + return resp + + server = await make_bridge(upstream_handler=upstream) + async with aiohttp.ClientSession() as s: + async with s.post(server.make_url("/v1/messages"), + json={"model": "local", "stream": True, + "messages": [{"role": "user", "content": "hi"}]}, + headers={"x-api-key": "test-key"}) as resp: + raw = await resp.read() + + events = split_sse(raw) + kinds = [e for e, _ in events] + assert kinds[0] == "message_start" + assert kinds == ["message_start", "message_delta", "message_stop"] + assert events[0][1]["message"]["usage"]["input_tokens"] >= 1 + + +async def test_stream_message_start_leading_on_abrupt_close(make_bridge): + """Upstream closes without any data: same guarantee.""" + async def upstream(request): + resp = web.StreamResponse(status=200, + headers={"Content-Type": "text/event-stream"}) + await resp.prepare(request) + return resp # end the response body immediately + + server = await make_bridge(upstream_handler=upstream) + async with aiohttp.ClientSession() as s: + async with s.post(server.make_url("/v1/messages"), + json={"model": "local", "stream": True, + "messages": [{"role": "user", "content": "hi"}]}, + headers={"x-api-key": "test-key"}) as resp: + raw = await resp.read() + + kinds = [e for e, _ in split_sse(raw)] + assert kinds == ["message_start", "message_delta", "message_stop"] + + +async def test_stream_garbage_json_skipped(make_bridge): + async def upstream(request): + resp = web.StreamResponse(status=200, + headers={"Content-Type": "text/event-stream"}) + await resp.prepare(request) + await resp.write(b"data: {not json\n\n") + await resp.write(sse_bytes( + {"choices": [{"delta": {"content": "still works"}}]}, + {"choices": [{"delta": {}, "finish_reason": "stop"}]}, + )) + await resp.write(b"data: [DONE]\n\n") + return resp + + server = await make_bridge(upstream_handler=upstream) + async with aiohttp.ClientSession() as s: + async with s.post(server.make_url("/v1/messages"), + json={"model": "local", "stream": True, + "messages": [{"role": "user", "content": "hi"}]}, + headers={"x-api-key": "test-key"}) as resp: + raw = await resp.read() + + events = split_sse(raw) + kinds = [e for e, _ in events] + assert kinds[0] == "message_start" + text_deltas = [d["delta"].get("text") for e, d in events + if e == "content_block_delta" and d["delta"].get("type") == "text_delta"] + assert text_deltas == ["still works"] + + +async def test_stream_truncation_reports_max_tokens(make_bridge): + async def upstream(request): + resp = web.StreamResponse(status=200, + headers={"Content-Type": "text/event-stream"}) + await resp.prepare(request) + await resp.write(sse_bytes( + {"choices": [{"delta": {"content": "half"}}]}, + {"choices": [{"delta": {}, "finish_reason": "length"}]}, + )) + await resp.write(b"data: [DONE]\n\n") + return resp + + server = await make_bridge(upstream_handler=upstream) + async with aiohttp.ClientSession() as s: + async with s.post(server.make_url("/v1/messages"), + json={"model": "local", "stream": True, + "messages": [{"role": "user", "content": "hi"}]}, + headers={"x-api-key": "test-key"}) as resp: + raw = await resp.read() + + msg_delta = [d for e, d in split_sse(raw) if e == "message_delta"][0] + assert msg_delta["delta"]["stop_reason"] == "max_tokens" + + +async def test_classifier_bypass_only_when_enabled(make_bridge): + """A classifier-shaped request is forwarded when the flag is off...""" + async def upstream(request): + body = await request.json() + return aiohttp.web.json_response({ + "id": "chatcmpl-bypass", + "choices": [{"index": 0, + "message": {"role": "assistant", "content": "real answer"}, + "finish_reason": "stop"}], + "usage": {"prompt_tokens": 1, "completion_tokens": 1}, + }) + + probe = {"model": "claude-sonnet-5", # classifier model name + "messages": [{"role": "user", "content": "should I run this?"}]} + hdrs = {"x-api-key": "test-key"} + + server = await make_bridge(upstream_handler=upstream) + async with aiohttp.ClientSession() as s: + async with s.post(server.make_url("/v1/messages"), json=probe, + headers=hdrs) as resp: + assert resp.status == 200 + body = await resp.json() + # flag off: forwarded upstream, real answer + assert body["content"][0]["text"] == "real answer" + + server = await make_bridge(upstream_handler=upstream, + allow_classifier_bypass=True) + async with aiohttp.ClientSession() as s: + async with s.post(server.make_url("/v1/messages"), json=probe, + headers=hdrs) as resp: + assert resp.status == 200 + body = await resp.json() + # flag on: short-circuited allow response + assert body["content"][0]["text"] == "no" + assert body["model"] == "local" + + # with tools present it is not a classifier probe, even with the flag on + probe_tools = dict(probe, tools=[{"name": "f", "input_schema": {"type": "object"}}]) + server = await make_bridge(upstream_handler=upstream, + allow_classifier_bypass=True) + async with aiohttp.ClientSession() as s: + async with s.post(server.make_url("/v1/messages"), json=probe_tools, + headers=hdrs) as resp: + body = await resp.json() + assert body["content"][0]["text"] == "real answer" diff --git a/cc-bridge/tests/test_streamer.py b/cc-bridge/tests/test_streamer.py new file mode 100644 index 0000000..218b53b --- /dev/null +++ b/cc-bridge/tests/test_streamer.py @@ -0,0 +1,91 @@ +"""AnthropicStreamer state machine: block indices, stop reasons, usage.""" + +import json + +from anthropic_adapter import AnthropicStreamer, _fmt_sse + + +def _events(feed_chunks, finish=True): + s = AnthropicStreamer("msg_x", "local", input_tokens=42) + out = [] + for c in feed_chunks: + out.extend(s.feed(c)) + if finish: + out.extend(s.finish()) + return s, out + + +def _parse(events): + return [(e.split("\n")[0].replace("event: ", ""), + json.loads(e.split("\n", 1)[1][5:])) for e in events] + + +def test_normal_sequence_thinking_text_tool(): + s, ev = _events([ + {"choices": [{"delta": {"reasoning_content": "think"}}]}, + {"choices": [{"delta": {"content": "hello"}}]}, + {"choices": [{"delta": {"tool_calls": [{"index": 0, "id": "t1", + "function": {"name": "bash", + "arguments": '{"c":"ls"}'}}]}}]}, + {"choices": [{"delta": {}, "finish_reason": "tool_calls"}]}, + ]) + parsed = _parse(ev) + kinds = [(e, d.get("index")) for e, d in parsed if e == "content_block_start"] + assert kinds == [("content_block_start", 0), # thinking + ("content_block_start", 1), # text + ("content_block_start", 2)] # tool_use + delta_kinds = [d["delta"]["type"] for e, d in parsed if e == "content_block_delta"] + assert delta_kinds == ["thinking_delta", "text_delta", "input_json_delta"] + msg_delta = [d for e, d in parsed if e == "message_delta"][0] + assert msg_delta["delta"]["stop_reason"] == "tool_use" + + +def test_length_maps_to_max_tokens(): + _, ev = _events([{"choices": [{"delta": {"content": "trunc"}, + "finish_reason": "length"}]}]) + msg_delta = [d for e, d in _parse(ev) if e == "message_delta"][0] + assert msg_delta["delta"]["stop_reason"] == "max_tokens" + + +def test_stop_maps_to_end_turn(): + _, ev = _events([{"choices": [{"delta": {"content": "x"}, + "finish_reason": "stop"}]}]) + msg_delta = [d for e, d in _parse(ev) if e == "message_delta"][0] + assert msg_delta["delta"]["stop_reason"] == "end_turn" + + +def test_empty_feed_still_finishes_cleanly(): + _, ev = _events([]) + parsed = _parse(ev) + events = [e for e, _ in parsed] + assert events == ["message_delta", "message_stop"] + assert [d for e, d in parsed if e == "message_delta"][0]["usage"]["output_tokens"] == 0 + + +def test_output_tokens_fallback_estimate_without_usage(): + s = AnthropicStreamer("msg_x", "local") + ev = s.feed({"choices": [{"delta": {"content": "a" * 100}}]}) + assert ev # start + delta + s.finish() + # no upstream usage came in, so the estimate must be non-zero + _, finish_events = _events([{"choices": [{"delta": {"content": "a" * 100}}]}]) + msg_delta = [d for e, d in _parse(finish_events) if e == "message_delta"][0] + assert msg_delta["usage"]["output_tokens"] >= 1 + + +def test_usage_from_upstream_wins(): + _, ev = _events([ + {"usage": {"completion_tokens": 17}, "choices": [{"delta": {"content": "x"}}]}, + ]) + msg_delta = [d for e, d in _parse(ev) if e == "message_delta"][0] + assert msg_delta["usage"]["output_tokens"] == 17 + + +def test_initial_carries_input_tokens(): + s = AnthropicStreamer("msg_x", "local", input_tokens=42) + parsed = _parse(s.initial()) + msg = parsed[0][1]["message"] + assert msg["id"] == "msg_x" + assert msg["model"] == "local" + assert msg["usage"]["input_tokens"] == 42 + assert parsed[0][0] == "message_start" From e75bf080890d4d71596554fe28121f3a0eb1def1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=8A=84=E6=89=8B=E6=BB=8B=E5=91=B3?= Date: Wed, 12 Aug 2026 01:15:48 +0800 Subject: [PATCH 2/2] docs(cc-bridge): README, technical design, and usage guide English docs for the adapter: setup and flags (README.md), protocol and security details (TECHNICAL.md), a Claude Code client guide, a chat-template analysis with a vendored froggeric v21.3 reference (MIT), and a root README/CHANGELOG entry. --- CHANGELOG.md | 11 + README.md | 28 ++ cc-bridge/README.md | 103 ++++++ cc-bridge/TECHNICAL.md | 239 ++++++++++++++ cc-bridge/misc/chat-template-fix.md | 77 +++++ cc-bridge/misc/chat_template-patched.jinja | 358 +++++++++++++++++++++ docs/guides/connect-claude-code.md | 87 +++++ docs/proposals/native-anthropic-adapter.md | 77 +++++ 8 files changed, 980 insertions(+) create mode 100644 cc-bridge/README.md create mode 100644 cc-bridge/TECHNICAL.md create mode 100644 cc-bridge/misc/chat-template-fix.md create mode 100644 cc-bridge/misc/chat_template-patched.jinja create mode 100644 docs/guides/connect-claude-code.md create mode 100644 docs/proposals/native-anthropic-adapter.md diff --git a/CHANGELOG.md b/CHANGELOG.md index aa53e38..1c640d9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,17 @@ All notable changes to BaseRT. This project is pre-1.0, so minor versions may include behavior changes. Format loosely based on [Keep a Changelog](https://keepachangelog.com). +## [Unreleased] + +### Added + +- **cc-bridge: native Anthropic Messages adapter.** A thin protocol bridge + (`cc-bridge/`) that lets Claude Code talk to a local `basert serve` through + `/v1/messages`, with native SSE streaming, thinking and tool blocks, + keep-alive pings, and cancellation propagation. Inbound requests are + authenticated against a required master key; the server binds to 127.0.0.1 + by default. See `cc-bridge/README.md`. + ## [0.2.0] — 2026-07-31 **A second hardware backend.** BaseRT now runs on **NVIDIA GB10 (DGX Spark, diff --git a/README.md b/README.md index f51ce41..9c30c9b 100644 --- a/README.md +++ b/README.md @@ -111,6 +111,34 @@ pi Inside pi, run `/model` to pick a served model. See the [pi-basert README](https://github.com/basecompute/pi-basert) for details. +### Claude Code via cc-bridge + +[cc-bridge](cc-bridge/) is a native Anthropic Messages API adapter that lets +[Claude Code](https://claude.com/claude-code) talk to a local `basert serve` +directly (`/v1/messages` ↔ `/v1/chat/completions`), with SSE streaming, thinking +and tool blocks, keep-alive pings, and cancellation propagation. Inbound +requests require a master key; the adapter binds to 127.0.0.1 by default. + +```sh +# 1. Serve a model +basert serve basecompute/Qwen3.6-35B-A3B + +# 2. Set up and run the adapter (see cc-bridge/README.md for all options) +cd cc-bridge +python3 -m venv .venv && .venv/bin/pip install -r requirements.txt +.venv/bin/python anthropic_adapter.py --master-key change-me \ + --model basecompute/Qwen3.6-35B-A3B + +# 3. Point Claude Code at it +export ANTHROPIC_BASE_URL="http://localhost:4003" +export ANTHROPIC_AUTH_TOKEN="change-me" +export ANTHROPIC_MODEL="local" +claude +``` + +Full client setup and troubleshooting: +[docs/guides/connect-claude-code.md](docs/guides/connect-claude-code.md). + ## The `basert` CLI `basert` is a single front-end. Model-management commands run natively; runtime diff --git a/cc-bridge/README.md b/cc-bridge/README.md new file mode 100644 index 0000000..113fe7a --- /dev/null +++ b/cc-bridge/README.md @@ -0,0 +1,103 @@ +# cc-bridge — native Anthropic Messages adapter for baseRT + +A thin protocol bridge that lets Claude Code talk to a local baseRT inference +server directly, with no LiteLLM or other intermediate layer. + +``` +Claude Code ──(/v1/messages)──▶ anthropic_adapter.py:4003 ──(/v1/chat/completions, stream)──▶ baseRT:7999 +``` + +## Why it exists + +- Claude Code speaks only the Anthropic Messages API (`/v1/messages`); + `basert serve` exposes OpenAI-compatible endpoints (`/v1/chat/completions`). +- A LiteLLM bridge was tried first and failed in two measured ways (see + `docs/proposals/native-anthropic-adapter.md`): + 1. **Non-streaming translation bug**: LiteLLM's openai→anthropic non-streaming + conversion always returns empty `content` (upstream issue #27492, still + open in 1.89.2 and 1.95.0). + 2. **Long prefill drops the connection**: Claude Code's requests carry a + ~23.5k-token system prompt; baseRT needs ~35s to prefill it, and the + stainless HTTP client gives up after ~36s of silence and retries the same + prompt **non-streaming** — straight into bug 1. +- This adapter generates the Anthropic wire format natively instead, modeled on + llama.cpp server's Anthropic implementation (MIT). + +## Setup + +```bash +cd cc-bridge +python3 -m venv .venv +.venv/bin/pip install -r requirements.txt +``` + +## Run + +```bash +# prerequisite: baseRT is running +.venv/bin/python anthropic_adapter.py --master-key --model +``` + +`--master-key` and `--model` are required; everything else has a sensible +default: + +| Flag | Default | Meaning | +|---|---|---| +| `--host` | `127.0.0.1` | bind address; use `0.0.0.0` only for cross-machine deployments, together with a strong key | +| `--port` | 4003 | listen port | +| `--upstream` | `http://127.0.0.1:7999/v1` | baseRT chat/completions endpoint | +| `--model` | — (**required**) | backend model id sent to baseRT | +| `--master-key` | — (**required**) | shared secret; every inbound request must present it via `x-api-key` or `Authorization: Bearer`, otherwise it is rejected with 401 | +| `--max-tokens-override` | 0 | max_tokens override, 0 = pass the client value through | +| `--debug-dir` | *(empty)* | request/response dump directory, empty = no dump; enabling it writes full prompts to disk | +| `--no-system-merge` | off | skip system-block merging (froggeric v21.3 baked into the model) | +| `--allow-classifier-bypass` | off | short-circuit harm-classifier probes; security implications in TECHNICAL.md §3.7 | +| `--log-level` | `info` | debug / info / warning / error | + +Example with different port/upstream/model: + +```bash +.venv/bin/python anthropic_adapter.py \ + --port 4004 --upstream http://other:8080/v1 --model Qwen3-0.6B \ + --master-key 'change-me' +``` + +Run `python3 anthropic_adapter.py --help` for the full flag reference. + +Design details, protocol rules and known limitations: [TECHNICAL.md](TECHNICAL.md). + +## Claude Code client configuration + +```bash +export ANTHROPIC_BASE_URL="http://:4003" +export ANTHROPIC_AUTH_TOKEN="" +export ANTHROPIC_MODEL="local" +export ANTHROPIC_DEFAULT_SONNET_MODEL="local" +export ANTHROPIC_DEFAULT_OPUS_MODEL="local" +export ANTHROPIC_DEFAULT_HAIKU_MODEL="local" +``` + +The model name is an alias the adapter echoes back as `local`; it does not need +to match any known model family. + +## Verify + +```bash +# non-streaming smoke test +curl -s -m 120 http://localhost:4003/v1/messages \ + -H "content-type: application/json" -H "x-api-key: " -H "anthropic-version: 2023-06-01" \ + -d '{"model":"local","max_tokens":64,"messages":[{"role":"user","content":"1+1=? Answer with a number only"}]}' +``` + +## Known limitations + +- The model always thinks (cannot be disabled), and thinking tokens count + toward the output budget; give a generous `--max-tokens-override` for heavy + thinking models. +- Thinking blocks carry an empty signature (local models, no cryptographic + verification; same as llama.cpp). +- Claude Code's system reminders (mid-array `role:"system"` messages) are + merged into a single leading system message, because baseRT's Qwen template + requires system to come first. With a chat template that handles mid-list + system messages (e.g. froggeric v21.3), pass `--no-system-merge`. +- Text-only: multimodal and audio inputs are not supported. diff --git a/cc-bridge/TECHNICAL.md b/cc-bridge/TECHNICAL.md new file mode 100644 index 0000000..5f6fdd4 --- /dev/null +++ b/cc-bridge/TECHNICAL.md @@ -0,0 +1,239 @@ +# cc-bridge — technical notes + +Native Anthropic Messages API adapter that lets Claude Code talk to baseRT +directly. This document records the motivation, design decisions, protocol +details, security posture, deployment notes and known limitations. + +## 1. Motivation + +Claude Code speaks only the Anthropic Messages API (`POST /v1/messages`, SSE +streaming, content blocks); `basert serve` exposes OpenAI-compatible endpoints +(`/v1/chat/completions`). A protocol bridge is required; the question is who +bridges and how. + +### Why not LiteLLM (measured, 2026-08-08) + +The LiteLLM bridge failed in two ways that compound into "Claude Code gets a +200 but displays nothing": + +**Failure A: Claude Code gives up on slow streams and falls back to non-streaming** +- CC requests carry a ~23.5k-token system prompt (28 tool schemas + 3 system blocks) +- baseRT needs ~35.4s to prefill 23.5k tokens (663 t/s), emitting no bytes meanwhile +- CC's HTTP client (stainless) cuts the streaming connection after ~36s without a + first byte and resends the same prompt **non-streaming** (packet capture: first + attempt has `"stream": true`, the retry has no `stream` field) +- baseRT logs show the streaming request cut at ~35.8s (output truncated, no + finish_reason) + +**Failure B: LiteLLM's non-streaming translation drops all content** +- baseRT returns full content (`content` + `reasoning_content`) for non-streaming +- LiteLLM's anthropic translation yields `"content":[],"stop_reason":"end_turn"` — content swallowed +- Excluded via experiments: reproduces with and without upstream `reasoning_content`; + identical on litellm 1.89.2 and 1.95.0 +- Confirmed upstream: BerriAI/litellm#27492 (OPEN); root cause in + `litellm/llms/anthropic/experimental_pass_through/adapters/transformation.py` + +**Conclusion**: the streaming path works fine through LiteLLM, but Failure A +pushes requests into Failure B's dead end. The community has the same idea +(leonmeijer/litellm_reasoning_proxy: force streaming + re-aggregate), but rather +than fix LiteLLM we generate the Anthropic protocol natively — following +llama.cpp server's approach. + +## 2. Design + +Modeled on llama.cpp server's Anthropic implementation (MIT): + +``` +Claude Code ──/v1/messages──▶ anthropic_adapter.py:4003 ──/v1/chat/completions, stream──▶ basert serve:7999 +``` + +Key decisions: + +| Decision | Rationale | +|---|---| +| Generate the Anthropic protocol natively | Avoids LiteLLM's translation bug; response shape fully under our control | +| SSE comment keep-alive pings | llama.cpp `sse_ping_interval` (default 30s): send `:\n\n` during silent periods so CC's 36s timeout never fires; we start pinging from request dispatch (every 20s), earlier than llama.cpp | +| Cancellation propagation inside the polling loop | llama.cpp `HTTP_POLLING_SECONDS = 1` + httplib `is_connection_closed` (non-blocking `recv(MSG_PEEK)` EOF probe): no extra threads/tasks | +| Upstream timeouts (connect 10s / read 600s) | baseRT is single-slot: when busy, connections queue server-side and response headers can be late; without a timeout the handler hangs forever. sock_read must exceed the longest silent gap in a stream (~35s prefill) | +| `--max-tokens-override` | CC sends small budgets (64/8192) for tool-loop and background calls; a heavy-thinking model burns the budget in thinking and truncates the body. The override replaces the client budget with a fixed value | +| Empty thinking-block signature | Same as llama.cpp (local models, no cryptographic verification) | +| Native non-streaming path | CC may fall back to non-streaming at any time; assembly shares the same block rules as streaming and is unaffected by the LiteLLM bug | + +### Why "poll + probe" instead of write-failure detection + +Cancelling a request = Claude Code closes the TCP connection. The adapter may +be blocked waiting for the next upstream chunk at that moment; the disconnect +is invisible until the next chunk arrives and the write to the client fails. +So we probe actively: on the 1s polling tick, `recv(1, MSG_PEEK)` on the client +socket — EOF means gone. This is more elegant than a watchdog task: the +detection rides on the existing loop's tick with zero extra machinery. + +## 3. Protocol details + +### 3.1 Request conversion (Anthropic → OpenAI) + +- **system**: the top-level `system` param (string or text blocks) and every + `role:"system"` entry inside `messages` (CC 2.x injects system reminders + mid-array) are merged into a single leading system message. baseRT's Qwen + chat template throws on a system message that is not first. With + `--no-system-merge` this step is skipped (requires a template that handles + mid-list system messages, e.g. froggeric v21.3 baked in via base-convert); + see `misc/chat-template-fix.md`. +- **assistant messages**: `thinking` blocks → `reasoning_content`; `tool_use` + blocks → OpenAI `tool_calls` +- **user messages**: `tool_result` blocks → `role:"tool"` messages, kept in + their original position relative to text blocks +- **tools**: `input_schema` → `parameters`; anthropic-only fields are dropped +- **stream / max_tokens**: passed through (max_tokens replaceable by + `--max-tokens-override`) +- **sampling params**: `temperature`, `top_p` passed through as-is; + `stop_sequences` → `stop` + +### 3.2 Streaming response (SSE lifecycle) + +Block index rule (llama.cpp): **thinking (0) → text (0 or 1) → tool_use (n+)** + +``` +event: message_start ← on first chunk (id/model/usage) +event: content_block_start index=0 thinking ← on first reasoning delta +event: content_block_delta index=0 thinking_delta +event: content_block_start index=1 text ← on first text delta (1 if thinking, else 0) +event: content_block_delta index=1 text_delta +event: content_block_start index=n tool_use ← tool call (id/name/input:{}) +event: content_block_delta index=n input_json_delta +event: content_block_stop index=… ← close all open blocks in 0,1,n order +event: message_delta stop_reason=end_turn|tool_use|max_tokens +event: message_stop +``` + +- Keep-alive: after 20s of silence, a single `:\n\n` line (SSE comment, + treated as a heartbeat by clients) +- Anthropic streams do **not** emit `data: [DONE]` (same as llama.cpp) +- If the upstream stream ends before any content arrives, `message_start` is + still emitted first, so the event sequence is always legal +- `finish_reason="length"` (openai's max_tokens truncation) is reported as + `stop_reason: "max_tokens"`, never as a natural end + +### 3.3 Non-streaming response + +Assembles `thinking` (signature:"") + `text` + `tool_use` (input JSON-parsed, +`{}` on parse failure) blocks; `stop_reason` maps from `finish_reason` +(stop→end_turn, tool_calls→tool_use, length→max_tokens). + +### 3.4 Cancellation propagation + +- The main loop awaits `upstream_iter.__anext__()` under a 1s + `asyncio.wait_for` (polling tick) +- On timeout → `client_gone()`: non-blocking `recv(1, MSG_PEEK)` on the client + socket; EOF means the client left +- Client gone → `up.close()` aborts the upstream call → baseRT stops generating +- `ConnectionResetError` on writes to the client triggers the same upstream + close +- Note: aiohttp's `StreamReader` has no `__anext__`; take + `up.content.__aiter__()` (an `AsyncStreamIterator`) and wait_for on that + +### 3.5 Timeout design + +- Upstream connect: 10s; upstream read (silence between reads): 600s + (must exceed the longest prefill, ~35s) +- While baseRT is single-slot, queued requests are covered by the connect + timeout — they never hang forever + +### 3.6 Token accounting (honest estimates) + +`/v1/messages/count_tokens` never under-reports: Claude Code uses it for +context-budget management, and under-reporting would let the conversation +exceed the real context window. The endpoint and the streaming +`message_start.usage.input_tokens` both use `estimate_input_tokens()`, a +deliberate upper-bound heuristic (CJK characters ~1 token each, other text +~1 token per 3.5 chars, including tool schemas) — not a real tokenizer. +Streaming `message_delta.usage.output_tokens` uses the upstream usage when +available, otherwise the same estimate over the emitted text. + +### 3.7 Classifier bypass (off by default) + +Before executing Bash/WebFetch/WebSearch, Claude Code sends the conversation +transcript to `/v1/messages` under a separate model name (e.g. +`claude-sonnet-5`) for online safety judgment. Our backend is not a classifier: +in practice it misjudges harmless commands (`date && echo`, `find`, +`echo "test"`) as dangerous and emits `yes`, which makes CC +block normal tool calls. + +`--allow-classifier-bypass` (env: none; CLI only) short-circuits requests that +match a heuristic (model name ≠ `local`, no tools, non-streaming, exactly one +user message) with a fixed `no` allow response. + +**Security impact**: this disables CC's online harm detection — the operator +is fully responsible for what the agent executes. It is OFF by default; the +heuristic is a heuristic, not an identifier of the classifier, and may change +as CC evolves. Enable it only for local, offline, or trusted workloads. + +**Alternatives**: +- A: set `CLAUDE_CODE_ATTRIBUTION_HEADER=0` in the CC config, exploiting a + known bug (#64585) that breaks the classifier sub-request (CC-side, no + adapter change) +- B: forward classifier requests to baseRT and accept the false-positive + blocks (for environments where no security degradation is acceptable) +- C: deploy a dedicated harm-classifier fine-tune on baseRT + +### 3.8 Authentication + +`--master-key` is required. Every inbound `/v1/messages` and +`/v1/messages/count_tokens` request must present it via `x-api-key` or +`Authorization: Bearer`; anything else is rejected with an Anthropic-style +`authentication_error` 401 before the body is parsed. The same key is sent +upstream as a Bearer token — either set baseRT's `--api-key` to the same value, +or run baseRT on a trusted network. The adapter binds to `127.0.0.1` by +default; bind `0.0.0.0` only for cross-machine deployments. + +## 4. Deployment & operations + +```bash +cd cc-bridge +python3 -m venv .venv && .venv/bin/pip install -r requirements.txt +.venv/bin/python anthropic_adapter.py --master-key 'change-me' \ + --model basecompute/Qwen3.6-35B-A3B +``` + +Client (Claude Code) configuration: `docs/guides/connect-claude-code.md`. + +**baseRT-side notes**: in single-slot deployments concurrent requests queue +(upstream timeouts prevent hangs but not throughput); consider +`--continuous-batching` for multiple clients. + +## 5. Known limitations & open problems + +| # | Problem | Status / mitigation | +|---|---|---| +| 1 | The model always thinks (cannot be disabled); thinking tokens count toward the output budget, so small budgets truncate the body | `--max-tokens-override` mitigates; proper fix needs separate thinking budget in baseRT (tracked in the `--chat-template`/thinking upstream issues) | +| 2 | baseRT single-slot queueing: one long request (35s prefill) stalls others | upstream timeouts prevent hangs; `--continuous-batching` recommended | +| 3 | Occasional GPU OOM on the first big request after baseRT restart (kIOGPUCommandBufferCallbackErrorOutOfMemory) | Retry passes (prefix-cache hit); likely a memory spike while the model loads | +| 4 | Empty thinking-block signature (CC accepts; local models have no verification) | Intentional, same as llama.cpp | +| 5 | Multimodal/audio inputs unsupported | text path first | +| 6 | CC system reminders merged to the front; positional semantics lost | baseRT Qwen template constraint; `--no-system-merge` when the template handles it | +| 7 | The max_tokens override also applies to CC background tasks (title generation etc.) and may return over-long replies | under observation; may differentiate by request shape | +| 8 | CC's online harm classifier is useless for non-Anthropic models — Qwen misjudges harmless commands as `yes`, blocking Bash/WebFetch | adapter can short-circuit these requests when `--allow-classifier-bypass` is enabled; security cost in §3.7 | + +## 6. Verification record (2026-08-08) + +- Non-streaming: thinking+text blocks complete (empty on the same path through + LiteLLM) +- Streaming: lifecycle/block indices correct; tool calls `input_json_delta` + + `stop_reason: tool_use` +- 36k-token prompt: completes fully; keep-alive pings work +- Cancellation propagation: client disconnect → upstream closed → marker logged +- CC end-to-end: multi-turn agentic tool loop (WebSearch) runs completely, + replies display correctly +- Upstream busy (single-slot queueing): no more permanent hangs (connect/read + timeouts) + +## 7. References + +- llama.cpp `tools/server/server-chat.cpp` (request conversion), + `server-task.cpp` (to_json_anthropic / to_json_anthropic_stream), + `server-context.cpp` (sse_ping_interval / HTTP_POLLING_SECONDS), + `server-http.h` (is_connection_closed), `vendor/cpp-httplib/httplib.h` +- BerriAI/litellm#27492 (non-streaming content loss, OPEN) +- leonmeijer/litellm_reasoning_proxy (community solution to the same problem, MIT) +- froggeric/Qwen-Fixed-Chat-Templates v21.3 (vendored at + `misc/chat_template-patched.jinja`, MIT) diff --git a/cc-bridge/misc/chat-template-fix.md b/cc-bridge/misc/chat-template-fix.md new file mode 100644 index 0000000..c1755c3 --- /dev/null +++ b/cc-bridge/misc/chat-template-fix.md @@ -0,0 +1,77 @@ +# Chat template: mid-list system messages on Qwen3.5/3.6 + +## Problem + +Claude Code (and other agentic coding tools) inject `role:"system"` messages +into the middle of the message array at runtime — context-compression +reminders, file-change notifications, and similar steering instructions. + +The official Qwen3.5/3.6 chat template rejects this with a hard exception: + +``` +Qwen/Qwen3.6-35B-A3B, commit 7da1103: +{%- if message.role == "system" %} + {%- if not loop.first %} + {{- raise_exception('System message must be at the beginning.') }} + {%- endif %} +{%- endif %} +``` + +When triggered, baseRT catches the exception and falls back to a generic ChatML +template. The request succeeds but loses + +- model-specific formatting +- tool-call syntax +- reasoning delimiters + +Any agentic workload that uses a Qwen3.5/3.6 model and injects system +instructions mid-conversation hits this. + +## Community solutions + +The most mature community response is +[froggeric/Qwen-Fixed-Chat-Templates](https://huggingface.co/froggeric/Qwen-Fixed-Chat-Templates) +(v21.3, MIT — vendored reference copy at `chat_template-patched.jinja`), which +replaces the exception with a direct ChatML rendering of each system block in +its original position. It also adds several features for agentic workloads: + +| Aspect | Official Qwen3.6 | froggeric v21.3 | +|---|---|---| +| Mid-list system message | `raise_exception` — crash | Rendered as an independent ChatML system block | +| Unknown message role | `raise_exception` — crash | Rendered as `[role]: content` | +| Missing user query | `raise_exception` — crash | Safe fallback, no crash | +| `developer` role | Not supported | Treated identically to `system` | +| Tool-call format | XML only | XML (default) or JSON, selectable via `_tool_format` | +| Consecutive tool failure detection | None | Detects error keywords (traceback, command not found, etc.); after two consecutive failures, injects a system warning and disables thinking to prevent error loops | +| Inline thinking control | None | Message-level `think_off` / `think_on` markers in system and user prompts (rendered via ChatML special tokens) | +| Tool argument truncation | None | `max_tool_arg_chars` prevents context-window overflow | +| Tool response truncation | None | `max_tool_response_chars` with `[TRUNCATED]` marker | +| Reasoning tag parsing | `` only | Six variants: ``, ``, and whitespace-deformed forms | +| Auto-disable thinking with tools | None | `auto_disable_thinking_with_tools` parameter | +| Thinking during consecutive failures | Always on | Forced off after two consecutive tool errors | + +## What this means for downstream bridges + +`anthropic_adapter.py` contains system-block merging logic in +`convert_anthropic_to_oai()` (enabled by default, skippable with +`--no-system-merge`). Its purpose is to collapse Claude Code's top-level +`system` parameter and any mid-array system reminders into a single leading +system message so the official Qwen template does not crash. With a patched +template baked into the model via `base-convert`, that logic becomes +unnecessary. + +## Recommendation + +`basert serve` does not currently support loading an external chat template. +Adding a `--chat-template` flag (or an equivalent mechanism, as llama.cpp +provides) would resolve this at the engine level: + +- The adapter's system-merging logic could be removed. +- Downstream bridges would not each need to work around a template limitation + that only affects certain model versions. +- Users could deploy community-maintained templates optimized for agentic + workloads without re-converting their models. + +Tracked upstream as a GitHub issue (the engine-side feature request, opened +together with the cc-bridge PR); this document and the vendored template are +the reference material for it. diff --git a/cc-bridge/misc/chat_template-patched.jinja b/cc-bridge/misc/chat_template-patched.jinja new file mode 100644 index 0000000..fa67c00 --- /dev/null +++ b/cc-bridge/misc/chat_template-patched.jinja @@ -0,0 +1,358 @@ +{#- +Vendored reference copy of froggeric/Qwen-Fixed-Chat-Templates v21.3 +https://huggingface.co/froggeric/Qwen-Fixed-Chat-Templates + +MIT License + +Copyright (c) 2025 froggeric (https://huggingface.co/froggeric) + +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. + +This file is not loaded at runtime by cc-bridge; it is the reference for the +engine-side --chat-template feature request (see chat-template-fix.md). +-#} +{%- set template_version = "qwen3.6-froggeric-v21.3" %} +{%- set _tool_format = tool_call_format if tool_call_format is defined else 'xml' %} +{%- set image_count = namespace(value=0) %} +{%- set video_count = namespace(value=0) %} +{%- set add_vision_id = add_vision_id if add_vision_id is defined else false %} +{%- set enable_thinking = enable_thinking if enable_thinking is defined else true %} +{%- set auto_disable_thinking_with_tools = auto_disable_thinking_with_tools if auto_disable_thinking_with_tools is defined else false %} +{%- set _preserve_thinking = preserve_thinking if preserve_thinking is defined else true %} +{%- set max_tool_arg_chars = max_tool_arg_chars if max_tool_arg_chars is defined else 0 %} +{%- set max_tool_response_chars = max_tool_response_chars if max_tool_response_chars is defined else 0 %} +{%- set _has_tools = (tools is defined and tools and tools is iterable and tools is not mapping) %} +{%- set ns_state = namespace(thinking=enable_thinking) %} +{%- if auto_disable_thinking_with_tools and _has_tools %} + {%- set ns_state.thinking = false %} +{%- endif %} +{%- macro render_content(content, do_vision_count, is_system_content=false) %} + {%- if content is string %} + {{- content }} + {%- elif content is iterable and content is not mapping %} + {%- for item in content %} + {%- if item is mapping %} + {%- if item.type == 'image' or 'image' in item or 'image_url' in item %} + {%- if is_system_content %} + {{- raise_exception('System message cannot contain images.') }} + {%- endif %} + {%- if do_vision_count %} + {%- set image_count.value = image_count.value + 1 %} + {%- endif %} + {%- if add_vision_id %} + {{- 'Picture ' ~ image_count.value ~ ': ' }} + {%- endif %} + {{- '<|vision_start|><|image_pad|><|vision_end|>' }} + {%- elif item.type == 'video' or 'video' in item %} + {%- if is_system_content %} + {{- raise_exception('System message cannot contain videos.') }} + {%- endif %} + {%- if do_vision_count %} + {%- set video_count.value = video_count.value + 1 %} + {%- endif %} + {%- if add_vision_id %} + {{- 'Video ' ~ video_count.value ~ ': ' }} + {%- endif %} + {{- '<|vision_start|><|video_pad|><|vision_end|>' }} + {%- elif 'text' in item %} + {{- item.text }} + {%- else %} + {{- raise_exception('Unexpected item type in content.') }} + {%- endif %} + {%- else %} + {{- item | string }} + {%- endif %} + {%- endfor %} + {%- elif content is none or content is undefined %} + {{- '' }} + {%- else %} + {{- raise_exception('Unexpected content type.') }} + {%- endif %} +{%- endmacro %} +{%- if not messages %} + {{- raise_exception('No messages provided.') }} +{%- endif %} +{%- set _first_role = messages[0].role %} +{%- if _first_role == 'system' or _first_role == 'developer' %} + {%- set _sys_msg = messages[0] %} + {%- set _msgs = messages[1:] %} +{%- else %} + {%- set _sys_msg = none %} + {%- set _msgs = messages %} +{%- endif %} +{%- set _sc = '' %} +{%- if _sys_msg is not none %} + {%- set _sc = render_content(_sys_msg.content, false, true) | trim %} + {%- if '<|think_off|>' in _sc %} + {%- set ns_state.thinking = false %} + {%- set _sc = _sc.split('<|think_off|>') | join('') | trim %} + {%- elif '<|think_on|>' in _sc %} + {%- set ns_state.thinking = true %} + {%- set _sc = _sc.split('<|think_on|>') | join('') | trim %} + {%- endif %} +{%- endif %} +{%- if _has_tools %} + {{- '<|im_start|>system\n' }} + {{- '# Tools\n\nYou have access to the following functions:\n\n' }} + {%- for tool in tools %} + {{- '\n' }} + {{- tool | tojson }} + {%- endfor %} + {{- '\n' }} + {%- set tool_instructions %} +If you choose to call a function ONLY reply in the following format with NO suffix: + +{%- if _tool_format == 'json' %} + +Brief explanation of tool call + + +{"name": "example_function_name", "arguments": {"example_parameter_1": "value_1", "example_parameter_2": "This is the value for the second parameter"}} + +{%- else %} + +Brief explanation of tool call + + + + +value_1 + + +This is the value for the second parameter +that can span +multiple lines + + + +{%- endif %} + + +Reminder: +- You can use the block to plan your next tool call OR to synthesize data and formulate your final response to the user. +- ALL explanation and reasoning MUST be placed strictly inside the block. +{%- if _tool_format == 'json' %} +- Function calls MUST follow the specified format: a single JSON object with "name" and "arguments" keys inside XML tags. +{%- else %} +- Function calls MUST follow the specified format: an inner block must be nested within XML tags. +{%- endif %} +- If you choose to call a tool, you MUST output the block IMMEDIATELY after thinking, with NO conversational text before it. +{%- if _tool_format == 'json' %} +- The tag MUST be at the very beginning of a new line, with NO spaces or indentation before it. +{%- else %} +- The and tags MUST be at the very beginning of a new line, with NO spaces or indentation before them. +{%- endif %} +- To call multiple functions, output a separate, completely closed block for EACH function. Do NOT nest blocks. +- If you have all necessary data, provide your final answer directly to the user without any tool call. + + {%- endset %} + {{- '\n\n' ~ tool_instructions | trim }} + {%- if _sc %} + {{- '\n\n' + _sc }} + {%- endif %} + {{- '<|im_end|>\n' }} +{%- else %} + {%- if _sc %} + {{- '<|im_start|>system\n' + _sc + '<|im_end|>\n' }} + {%- endif %} +{%- endif %} +{%- set _last_idx = _msgs | length - 1 %} +{%- set ns = namespace(multi_step_tool=true, last_query_index=_last_idx) %} +{%- for message in _msgs[::-1] %} + {%- set index = (_msgs | length - 1) - loop.index0 %} + {%- if ns.multi_step_tool and message.role == 'user' %} + {%- set _rc = render_content(message.content, false) | trim %} + {%- if not (_rc.startswith('') and _rc.endswith('')) %} + {%- set ns.multi_step_tool = false %} + {%- set ns.last_query_index = index %} + {%- endif %} + {%- endif %} +{%- endfor %} +{%- if ns.multi_step_tool %} + {%- if _last_idx > 50 %} + {%- set ns.last_query_index = _last_idx %} + {%- else %} + {%- set ns.last_query_index = 0 %} + {%- endif %} +{%- endif %} +{%- set ns2 = namespace(prev_role='', consecutive_failures=0) %} +{%- for message in _msgs %} + {%- set is_system = (message.role == "system" or message.role == "developer") %} + {%- set content = render_content(message.content, true, is_system) | trim %} + {%- if is_system or message.role == 'user' %} + {%- if '<|think_off|>' in content %} + {%- set ns_state.thinking = false %} + {%- set content = content.split('<|think_off|>') | join('') | trim %} + {%- elif '<|think_on|>' in content %} + {%- set ns_state.thinking = true %} + {%- set content = content.split('<|think_on|>') | join('') | trim %} + {%- endif %} + {%- endif %} + {%- if is_system %} + {{- '<|im_start|>system\n' + content + '<|im_end|>\n' }} + {%- elif message.role == 'user' %} + {%- set ns2.consecutive_failures = 0 %} + {{- '<|im_start|>user\n' + content + '<|im_end|>\n' }} + {%- elif message.role == 'assistant' %} + {%- set reasoning_content = '' %} + {%- if message.reasoning_content is defined and message.reasoning_content is not none %} + {%- if message.reasoning_content is string %} + {%- set reasoning_content = message.reasoning_content %} + {%- else %} + {%- set reasoning_content = message.reasoning_content | string %} + {%- endif %} + {%- elif message.thinking is defined and message.thinking is not none %} + {%- if message.thinking is string %} + {%- set reasoning_content = message.thinking %} + {%- else %} + {%- set reasoning_content = message.thinking | string %} + {%- endif %} + {%- else %} + {%- set _think_end = '' %} + {%- if content.startswith('') %} + {%- set _think_end = '' %} + {%- elif content.startswith('') %} + {%- set _think_end = '' %} + {%- elif '\n' in content %} + {%- set _think_end = '\n' %} + {%- elif '\n' in content %} + {%- set _think_end = '\n' %} + {%- elif '\n' in content %} + {%- set _think_end = '\n' %} + {%- elif '\n' in content %} + {%- set _think_end = '\n' %} + {%- endif %} + {%- if _think_end %} + {%- if 'thinking' in _think_end %} + {%- set _think_start = '' %} + {%- else %} + {%- set _think_start = '' %} + {%- endif %} + {%- set reasoning_content = content.split(_think_end)[0].rstrip('\n') %} + {%- if _think_start in reasoning_content %} + {%- set reasoning_content = reasoning_content.split(_think_start)[-1].lstrip('\n') %} + {%- endif %} + {%- set content = content.split(_think_end)[-1].lstrip('\n') %} + {%- endif %} + {%- endif %} + {%- set reasoning_content = reasoning_content | trim %} + {%- if (_preserve_thinking or loop.index0 > ns.last_query_index) and reasoning_content %} + {{- '<|im_start|>assistant\n\n' + reasoning_content + '\n\n\n' + content }} + {%- else %} + {{- '<|im_start|>assistant\n' + content }} + {%- endif %} + {%- if message.tool_calls is defined and message.tool_calls and message.tool_calls is iterable and message.tool_calls is not mapping %} + {%- for tool_call in message.tool_calls %} + {%- if tool_call.function is defined and tool_call.function is not none %} + {%- set tc = tool_call.function %} + {%- else %} + {%- set tc = tool_call %} + {%- endif %} + {%- if _tool_format == 'json' %} + {%- if not loop.first or content | trim %} + {{- '\n\n' }} + {%- endif %} + {%- set _args = '{}' %} + {%- if tc.arguments is defined and tc.arguments is not none %} + {%- if tc.arguments is mapping %} + {%- set _args = tc.arguments | tojson %} + {%- elif tc.arguments is string and tc.arguments %} + {%- set _args = tc.arguments %} + {%- endif %} + {%- endif %} + {{- '\n{"name": ' }}{{- tc.name | tojson }}{{- ', "arguments": ' }}{{- _args }}{{- '}\n' }} + {%- else %} + {%- if loop.first %} + {%- if content | trim %} + {{- '\n\n\n\n' }} + {%- else %} + {{- '\n\n' }} + {%- endif %} + {%- else %} + {{- '\n\n\n\n' }} + {%- endif %} + {%- if tc.arguments is defined and tc.arguments is not none %} + {%- if tc.arguments is mapping %} + {%- for args_name, args_value in tc.arguments.items() %} + {{- '\n' }} + {%- if args_value is mapping or (args_value is sequence and args_value is not string) %} + {%- set _av = args_value | tojson %} + {%- else %} + {%- set _av = args_value | string %} + {%- endif %} + {%- if max_tool_arg_chars > 0 and _av | length > max_tool_arg_chars %} + {{- _av[:max_tool_arg_chars] + '\n[TRUNCATED — original length ' ~ (_av | length | string) ~ ' chars]' }} + {%- else %} + {{- _av }} + {%- endif %} + {{- '\n\n' }} + {%- endfor %} + {%- elif tc.arguments is string and tc.arguments %} + {{- tc.arguments }} + {%- endif %} + {%- endif %} + {{- '\n' }} + {%- endif %} + {%- endfor %} + {%- endif %} + {{- '<|im_end|>\n' }} + {%- elif message.role == 'tool' %} + {%- set _content_lower = content | lower %} + {%- set _content_head = _content_lower[:80] %} + {%- if content | length < 500 and '$ ' not in content and 'took ' not in _content_lower and ('"error":' in _content_head or 'error:' in _content_head or 'err!' in _content_head or 'fatal:' in _content_head or 'exception:' in _content_head or 'traceback' in _content_head or 'command not found' in _content_head or 'invalid syntax' in _content_head or 'failed to' in _content_head) %} + {%- set ns2.consecutive_failures = ns2.consecutive_failures + 1 %} + {%- else %} + {%- set ns2.consecutive_failures = 0 %} + {%- endif %} + {%- if ns2.prev_role != 'tool' %} + {{- '<|im_start|>user' }} + {%- endif %} + {%- if max_tool_response_chars > 0 and content | length > max_tool_response_chars %} + {%- set content = content[:max_tool_response_chars] + '\n[TRUNCATED — original length ' ~ (content | length | string) ~ ' chars]' %} + {%- endif %} + {{- '\n\n' + content }} + {%- if ns2.consecutive_failures >= 2 %} + {{- '\n\n⚠️ SYSTEM WARNING: ' ~ ns2.consecutive_failures ~ ' consecutive tool errors detected. Your previous approach is incorrect. You MUST use a fundamentally different approach or corrected arguments.' }} + {%- elif ns2.consecutive_failures == 1 %} + {{- '\n\n⚠️ SYSTEM WARNING: The previous tool call returned an error. Diagnose the failure and retry with completely corrected arguments.' }} + {%- endif %} + {{- '\n' }} + {%- if loop.last %} + {{- '<|im_end|>\n' }} + {%- else %} + {%- set _next_role = _msgs[loop.index0 + 1].role %} + {%- if _next_role != 'tool' %} + {{- '<|im_end|>\n' }} + {%- endif %} + {%- endif %} + {%- else %} + {{- '<|im_start|>user\n[' + message.role + ']: ' + content + '<|im_end|>\n' }} + {%- endif %} + {%- set ns2.prev_role = message.role %} +{%- endfor %} +{%- if add_generation_prompt %} + {{- '<|im_start|>assistant\n' }} + {%- if not ns_state.thinking %} + {{- '\n\n\n\n' }} + {%- elif ns2.consecutive_failures >= 2 %} + {{- '\n\n\n\n' }} + {%- else %} + {{- '\n' }} + {%- endif %} +{%- endif %} \ No newline at end of file diff --git a/docs/guides/connect-claude-code.md b/docs/guides/connect-claude-code.md new file mode 100644 index 0000000..1909a86 --- /dev/null +++ b/docs/guides/connect-claude-code.md @@ -0,0 +1,87 @@ +# Connect Claude Code to baseRT via cc-bridge + +This guide explains how to make Claude Code in your LAN talk to a local baseRT +inference server directly, without LiteLLM. The bridging component is +`cc-bridge/anthropic_adapter.py`, modeled on llama.cpp server's Anthropic +implementation (design rationale: `docs/proposals/native-anthropic-adapter.md`; +full protocol details: `cc-bridge/TECHNICAL.md`). + +## Architecture + +``` +Claude Code ──(/v1/messages)──▶ cc-bridge :4003 ──(/v1/chat/completions, stream)──▶ baseRT :7999 +``` + +- 4003: cc-bridge (this repository's `cc-bridge/`) — a pure HTTP translation + layer; it touches no GPU and no model files +- 7999: `basert serve` itself + +## Start + +```bash +# prerequisite: baseRT is running (any address; point the adapter at it with --upstream) +cd cc-bridge +python3 -m venv .venv && .venv/bin/pip install -r requirements.txt +.venv/bin/python anthropic_adapter.py --master-key 'change-me' --model +``` + +All flags are documented in `cc-bridge/README.md`. In particular +`--max-tokens-override` shields the backend from Claude Code's small budgets +(64/8192) — a heavy local thinking model needs a generous budget. + +## Client configuration (Claude Code) + +```bash +export ANTHROPIC_BASE_URL="http://:4003" # host IP for LAN, localhost for same machine +export ANTHROPIC_AUTH_TOKEN="" +export ANTHROPIC_MODEL="local" +export ANTHROPIC_DEFAULT_SONNET_MODEL="local" +export ANTHROPIC_DEFAULT_OPUS_MODEL="local" +export ANTHROPIC_DEFAULT_HAIKU_MODEL="local" +claude +``` + +The `local` alias is arbitrary — the adapter echoes it back as `local` for +every request. Persist the block in the `env` section of `~/.claude/settings.json`. +**A new Claude Code session is required after changing these.** + +## Verify + +```bash +# smoke test (non-streaming) +curl -s -m 120 http://localhost:4003/v1/messages \ + -H "content-type: application/json" -H "x-api-key: " -H "anthropic-version: 2023-06-01" \ + -d '{"model":"local","max_tokens":64,"messages":[{"role":"user","content":"1+1=? Answer with a number only"}]}' +# expect: content contains {type:"thinking"} and {type:"text"} blocks, stop_reason: end_turn + +# streaming +curl -sN -m 120 http://localhost:4003/v1/messages \ + -H "content-type: application/json" -H "x-api-key: " -H "anthropic-version: 2023-06-01" \ + -d '{"model":"local","max_tokens":64,"stream":true,"messages":[{"role":"user","content":"hi"}]}' +# expect: message_start → content_block_start → thinking_delta → text_delta → message_delta → message_stop +``` + +End-to-end acceptance: run an agentic task through Claude Code that writes and +executes a file (multi-turn tool calls). + +## Troubleshooting + +- Request/response dump: start the adapter with `--debug-dir `; every + request is saved as `-req.json` and the response as `-res.json` (streams: + `-res.sse`), so you can compare byte-for-byte what CC sent and received. +- CC shows no reply but baseRT logs look fine: most likely content is being + dropped in translation — check whether `text_delta` appears in `-res.sse`. +- Truncated replies: the model's thinking burns the output budget; raise + `--max-tokens-override`. +- Health checks: CC probes `/api/hello`; the adapter implements it (200). +- 401s: your `ANTHROPIC_AUTH_TOKEN` must match the adapter's master key. + +## Known limitations (measured) + +- The model always thinks (cannot be disabled); thinking tokens count toward + the output budget. +- Thinking blocks carry an empty signature (local models; same as llama.cpp). +- Claude Code's system reminders (mid-array system messages) are merged into + the leading system message; pass `--no-system-merge` if the model's chat + template handles mid-list system messages. +- Text-only; multimodal and audio inputs are not supported. diff --git a/docs/proposals/native-anthropic-adapter.md b/docs/proposals/native-anthropic-adapter.md new file mode 100644 index 0000000..736b19c --- /dev/null +++ b/docs/proposals/native-anthropic-adapter.md @@ -0,0 +1,77 @@ +# Design: Native Anthropic Messages adapter for baseRT (cc-bridge) + +Status: implemented (2026-08-08), field-tested with Claude Code 2.1.226 over LAN. + +## Problem + +Claude Code speaks only the Anthropic Messages API (`POST /v1/messages`, SSE +streaming, content blocks). `basert serve` exposes OpenAI-compatible endpoints +(`/v1/chat/completions` etc.). Bridging is required; the question is where. + +### Why LiteLLM as the bridge failed (measured, 2026-08-08) + +Two independent faults, both reproduced end-to-end with byte-level captures: + +1. **Non-streaming translation drops content.** LiteLLM's openai→anthropic + non-streaming conversion always returns `"content":[]` regardless of the + upstream payload (tested with the real baseRT response and a fake upstream + with and without `reasoning_content`; litellm 1.89.2 and 1.95.0 identical). + Upstream confirmation: BerriAI/litellm#27492 (OPEN). Streaming path is fine. + +2. **Claude Code gives up on slow streaming and falls back to non-streaming.** + CC's requests carry ~23.5k tokens (28 tool schemas + 3 system blocks); + baseRT prefills that in ~35.4 s. CC's HTTP client (stainless) abandoned the + stream after ~36 s of silence and resent the same prompt as a non-streaming + request — landing on fault 1. Every request in the session did this. + +Net effect: CC received `200 OK` messages with empty content and rendered +nothing ("worked for 1m 55s"). + +## Solution: native adapter, no middleware + +Copy llama.cpp server's Anthropic pattern (MIT): a thin HTTP layer that owns +the Anthropic wire format end to end. + +``` +Claude Code ──/v1/messages──▶ cc-bridge (anthropic_adapter.py) ──/v1/chat/completions──▶ basert serve +``` + +Implementation (cc-bridge/anthropic_adapter.py, ~450 lines, Python/aiohttp): + +- **Request**: convert Anthropic → OpenAI chat. System prompt (string or blocks) + plus any `role:"system"` entries inside `messages` (CC 2.x system reminders) + are merged into a single leading system message — baseRT's Qwen chat template + raises if a system message is not first. +- **Streaming response**: message_start → content_block_start (thinking, idx 0) + → thinking_delta / text_delta (text, idx 0/1) → tool_use blocks with + input_json_delta → message_delta (stop_reason) → message_stop. baseRT already + splits `reasoning_content`/`content`, so no chat-template reasoning detection + is needed. +- **Keep-alive**: SSE comment pings (`:\n\n`) every 20 s while the stream is + silent, so CC never times out during long prefills (llama.cpp's + `sse_ping_interval`, default 30 s; we start pinging from request dispatch). +- **Non-streaming response**: assembled into thinking (empty `signature`, like + llama.cpp) + text + tool_use blocks — immune to the LiteLLM bug. +- **`--max-tokens-override`**: CC sends small budgets (64/8192) for tool-loop + and background calls; a heavy-thinking local model truncates on those. The + flag replaces the client budget with a fixed one (e.g. 32000). +- **`--debug-dir`**: saves every request body and response (raw SSE for + streams) for offline debugging. +- `/api/hello` (CC health check) and `/health` endpoints included. + +## Acceptance + +- [x] Non-streaming /v1/messages returns thinking + text blocks (litellm path returned empty). +- [x] Streaming lifecycle and block indices verified (thinking 0 → text 0/1 → tool_use n+). +- [x] Tool use streams `input_json_delta` with `stop_reason: tool_use`. +- [x] 36k-token prompt completes with pings keeping the connection alive. +- [x] Claude Code on LAN renders replies (repeated calls, no regression). + +## Known limitations + +- Model always reasons (cannot be disabled); thinking tokens count toward + output budget — use `--max-tokens-override` generously. +- Thinking blocks carry empty signature (local models, no crypto verification). +- Multimodal/audio input not supported yet (text path only). +- baseRT occasionally OOMs on the first large request right after model load + (GPU memory peak); a retry succeeds (prefix cache then serves it).