zach/z-bar-qt
1
1
Fork 0
Code Issues 23 Pull Requests 3 Actions Packages Projects 1 Releases Wiki Activity

fix: format and apply lint fixes in python script

Browse Source
This commit is contained in:
zach
2026-09-30 15:01:28 +02:00
parent ca51f660ad
commit 83bc153ee3
1 changed files with 21 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/mock-llama-server.py
+21 -117
View File
@@ -1,64 +1,5 @@
#!/usr/bin/env python3
"""Mock llama-server for exercising zshell's LLM error handling.
Implements the OpenAI-compatible subset of llama-server that the C++
backend talks to, and lets you trigger specific failure modes from the
QML chat UI by typing a keyword as your message.
Wire format is taken from ggml-org/llama.cpp (tools/server):
- SSE events are "data: <json>\n\n" (note the space after "data:").
- A clean stream ends with "data: [DONE]\n\n".
- Errors are {"error": {"code", "message", "type"}}.
- A stream that dies mid-way sends the error event, then closes
WITHOUT [DONE].
- A request that errors before the first token is a plain HTTP 4xx/5xx
with the error JSON body (not a stream).
- /props is a JSON object; n_ctx lives under
default_generation_settings.n_ctx.
Run:
python3 scripts/mock-llama-server.py [--port 8080]
Then point zshell's LLM endpoint at http://localhost:8080.
Trigger words (type as the chat message, case-insensitive):
normal -> healthy stream, ends with [DONE] (also the default)
instant error -> 500 before any token ("server_error")
context overflow -> 400 "exceeds the available context size (2048 ...)";
backend retries once at 2048, then fails
error during -> streams a few tokens, then an SSE error event,
closes without [DONE]
no done -> streams a few tokens, closes without [DONE] and
without an error (server killed mid-response)
200 error -> HTTP 200 with a plain JSON error body (not SSE)
tool call -> streams reasoning, then a tool_calls round calling
readfile on /etc/os-release (finish_reason
"tool_calls"); after you approve (or deny) the
permission dialog, the follow-up round that
carries the tool result gets a normal summary
answer. Exercises the tool permission dialog,
tool execution and the multi-round transcript.
multiple tool -> two tool calls in one round (readfile + webfetch
calls of this mock's own / endpoint); approve both at
once
unknown tool -> a tool call for a tool the client does not have;
approving it yields an "unknown tool" error
result
Control endpoint (for the non-chat routes / global state), via curl:
curl -XPOST localhost:8080/_mock -d '{"loading": true}'
/props, /v1/models, /health all answer 503 "Loading model"
curl -XPOST localhost:8080/_mock -d '{"no_models": true}'
/v1/models returns an empty list
curl -XPOST localhost:8080/_mock -d '{"chat_mode": "error_during"}'
force a chat behaviour regardless of the message text
(values: normal|instant_error|context_overflow|error_during|
no_done|200_error|tool_call|tool_call_multi|
tool_call_unknown, or null to go back to keywords)
curl -XPOST localhost:8080/_mock -d '{"reset": true}'
"""
import argparse
import contextlib
import json
import threading
import time
@@ -67,17 +8,14 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
MODEL_ID = "mock-model"
CTX_SIZE = 8192
# Set in main(); used by the tool-call behaviours so webfetch can point
# at this server itself.
_host = "127.0.0.1"
_port = 8080
# Mutable, shared across handler threads.
_state_lock = threading.Lock()
_state = {
"loading": False, # 503 "Loading model" on /props, /v1/models, /health
"no_models": False, # /v1/models returns an empty list
"chat_mode": None, # force a chat behaviour, overriding keywords
"loading": False,
"no_models": False,
"chat_mode": None,
}
@@ -122,7 +60,6 @@ def _error(code, message, etype):
def _sse(obj):
# Exact llama.cpp format_oai_sse shape: "data: " + compact json + "\n\n".
return ("data: " + json.dumps(obj, separators=(",", ":")) + "\n\n").encode()
@@ -130,9 +67,6 @@ def _sse_done():
return b"data: [DONE]\n\n"
# Chat behaviours. Each returns ("json", status, body_bytes) for a
# non-streaming answer, or ("sse", 200, None) and then writes to the
# provided stream writer.
def _behaviour_for(text):
t = text.lower()
if "instant error" in t:
@@ -145,7 +79,6 @@ def _behaviour_for(text):
return "no_done"
if "200 error" in t:
return "200_error"
# Order matters: "multiple tool calls" contains "tool call".
if "two tool calls" in t or "two tool call" in t:
return "tool_call_duo"
if "some tool calls" in t:
@@ -183,11 +116,6 @@ def _tool_call_delta(index, call_id=None, name=None, arguments=None):
def _stream_tool_calls(w, cid, reasoning, calls):
# One tool round: optional reasoning, then the calls (llama.cpp
# streams the id/name with the first fragment of each call and the
# arguments as successive fragments), then finish_reason
# "tool_calls". The client concatenates the argument fragments and
# shows the permission dialog on stream end.
w(_sse(_chunk(cid, {"role": "assistant"})))
if reasoning:
for piece in reasoning:
@@ -222,8 +150,6 @@ def _tool_result_snippets(messages, limit=160):
def _stream_tool_round_done(w, cid, messages):
# Follow-up round: the client sent the tool results back; answer
# with a plain content summary so the turn completes.
w(_sse(_chunk(cid, {"role": "assistant"})))
parts = ["mock:", "tool", "round", "complete."]
for i, snippet in enumerate(_tool_result_snippets(messages), 1):
@@ -252,8 +178,6 @@ def _stream_error_during(w, cid, words):
for word in words:
w(_sse(_chunk(cid, {"content": word + " "})))
time.sleep(0.05)
# Mid-stream failure: error event, then the connection closes without
# [DONE] (llama.cpp terminates the stream on a task error).
w(_sse(_error(500, "Mock: generation failed mid-stream", "server_error")))
@@ -262,7 +186,6 @@ def _stream_no_done(w, cid, words):
for word in words:
w(_sse(_chunk(cid, {"content": word + " "})))
time.sleep(0.05)
# Clean FIN, no [DONE], no error: server died mid-response.
def _last_user_text(messages):
@@ -270,7 +193,6 @@ def _last_user_text(messages):
if msg.get("role") == "user":
content = msg.get("content")
if isinstance(content, list):
# multimodal shape: pull out the text parts
parts = []
for part in content:
if isinstance(part, dict) and part.get("type") == "text":
@@ -284,7 +206,6 @@ class Handler(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"
server_version = "MockLlama/1.0"
# --- low-level response helpers -----------------------------------
def _send_json(self, status, obj):
body = json.dumps(obj, separators=(",", ":")).encode()
self.send_response(status)
@@ -295,7 +216,6 @@ class Handler(BaseHTTPRequestHandler):
self.wfile.write(body)
def _send_loading_error(self):
# llama.cpp middleware: server not ready -> 503 on every route.
self._send_json(503, _error(503, "Loading model", "unavailable_error"))
def _not_ready(self):
@@ -313,12 +233,9 @@ class Handler(BaseHTTPRequestHandler):
except (BrokenPipeError, ConnectionResetError):
pass
finally:
try:
with contextlib.suppress(BrokenPipeError, ConnectionResetError):
self.wfile.flush()
except (BrokenPipeError, ConnectionResetError):
pass
# --- routing -------------------------------------------------------
def do_GET(self):
path = self.path.split("?", 1)[0].rstrip("/") or "/"
if path in ("/health", "/v1/health"):
@@ -382,7 +299,6 @@ class Handler(BaseHTTPRequestHandler):
return self._send_json(404, _error(404, "not found", "not_found_error"))
# --- /_mock control ------------------------------------------------
def _mock_control(self, body):
with _state_lock:
if body.get("reset"):
@@ -399,7 +315,6 @@ class Handler(BaseHTTPRequestHandler):
self.log_message("control -> %s", json.dumps(snapshot))
return self._send_json(200, {"ok": True, "state": snapshot})
# --- /chat/completions --------------------------------------------
def _chat(self, body):
text = _last_user_text(body.get("messages"))
with _state_lock:
@@ -411,7 +326,6 @@ class Handler(BaseHTTPRequestHandler):
self.log_message("chat mode=%s stream=%s text=%r", mode, stream, text)
if mode == "instant_error":
# Error before the first token: plain HTTP error, not a stream.
return self._send_json(
500, _error(500, "Mock: instant server error", "server_error")
)
@@ -428,15 +342,11 @@ class Handler(BaseHTTPRequestHandler):
)
if mode == "200_error":
# 200 with a plain JSON error body (not SSE) -- the case the
# data: filter would otherwise swallow.
return self._send_json(
200, _error(400, "Mock: 200-with-json-error", "server_error")
)
if mode in _TOOL_MODES:
# Continuation round: the previous round's tool results came
# back through the client (role: "tool" messages).
if any(m.get("role") == "tool" for m in body.get("messages") or []):
return self._stream(
200,
@@ -467,7 +377,7 @@ class Handler(BaseHTTPRequestHandler):
"call_mock_2",
"webfetch",
[
'{"url": "http://%s:%d/", ' % (_host, _port),
f'{{"url": "http://{_host}:{_port}/", ',
'"format": "text"}',
],
),
@@ -487,7 +397,7 @@ class Handler(BaseHTTPRequestHandler):
"call_mock_2",
"webfetch",
[
'{"url": "http://%s:%d/", ' % (_host, _port),
f'{{"url": "http://{_host}:{_port}/", ',
'"format": "text"}',
],
),
@@ -500,7 +410,7 @@ class Handler(BaseHTTPRequestHandler):
"call_mock_4",
"webfetch",
[
'{"url": "http://%s:%d/", ' % (_host, _port),
f'{{"url": "http://{_host}:{_port}/", ',
'"format": "text"}',
],
),
@@ -513,7 +423,7 @@ class Handler(BaseHTTPRequestHandler):
"call_mock_6",
"webfetch",
[
'{"url": "http://%s:%d/", ' % (_host, _port),
f'{{"url": "http://{_host}:{_port}/", ',
'"format": "text"}',
],
),
@@ -533,7 +443,7 @@ class Handler(BaseHTTPRequestHandler):
"call_mock_2",
"webfetch",
[
'{"url": "http://%s:%d/", ' % (_host, _port),
f'{{"url": "http://{_host}:{_port}/", ',
'"format": "text"}',
],
),
@@ -546,7 +456,7 @@ class Handler(BaseHTTPRequestHandler):
"call_mock_4",
"webfetch",
[
'{"url": "http://%s:%d/", ' % (_host, _port),
f'{{"url": "http://{_host}:{_port}/", ',
'"format": "text"}',
],
),
@@ -559,7 +469,7 @@ class Handler(BaseHTTPRequestHandler):
"call_mock_6",
"webfetch",
[
'{"url": "http://%s:%d/", ' % (_host, _port),
f'{{"url": "http://{_host}:{_port}/", ',
'"format": "text"}',
],
),
@@ -572,7 +482,7 @@ class Handler(BaseHTTPRequestHandler):
"call_mock_2",
"webfetch",
[
'{"url": "http://%s:%d/", ' % (_host, _port),
f'{{"url": "http://{_host}:{_port}/", ',
'"format": "text"}',
],
),
@@ -585,7 +495,7 @@ class Handler(BaseHTTPRequestHandler):
"call_mock_4",
"webfetch",
[
'{"url": "http://%s:%d/", ' % (_host, _port),
f'{{"url": "http://{_host}:{_port}/", ',
'"format": "text"}',
],
),
@@ -598,7 +508,7 @@ class Handler(BaseHTTPRequestHandler):
"call_mock_6",
"webfetch",
[
'{"url": "http://%s:%d/", ' % (_host, _port),
f'{{"url": "http://{_host}:{_port}/", ',
'"format": "text"}',
],
),
@@ -611,7 +521,7 @@ class Handler(BaseHTTPRequestHandler):
"call_mock_2",
"webfetch",
[
'{"url": "http://%s:%d/", ' % (_host, _port),
f'{{"url": "http://{_host}:{_port}/", ',
'"format": "text"}',
],
),
@@ -624,7 +534,7 @@ class Handler(BaseHTTPRequestHandler):
"call_mock_4",
"webfetch",
[
'{"url": "http://%s:%d/", ' % (_host, _port),
f'{{"url": "http://{_host}:{_port}/", ',
'"format": "text"}',
],
),
@@ -637,7 +547,7 @@ class Handler(BaseHTTPRequestHandler):
"call_mock_6",
"webfetch",
[
'{"url": "http://%s:%d/", ' % (_host, _port),
f'{{"url": "http://{_host}:{_port}/", ',
'"format": "text"}',
],
),
@@ -646,7 +556,7 @@ class Handler(BaseHTTPRequestHandler):
"Two independent tool calls, ",
"batched together.",
]
else: # tool_call_unknown
else:
calls = [
(
"call_mock_3",
@@ -663,7 +573,6 @@ class Handler(BaseHTTPRequestHandler):
)
if not stream:
# Non-streaming success: single chat.completion object.
content = " ".join(words)
return self._send_json(
200,
@@ -699,10 +608,7 @@ class Handler(BaseHTTPRequestHandler):
return self._stream(200, lambda w: _stream_normal(w, cid, words))
def log_message(self, format, *args):
# Compact single-line logging to stdout.
print(
"%s - %s" % (self.log_date_time_string(), format % args), flush=True
)
print(f"{self.log_date_time_string()} - {format % args}", flush=True)
def main():
@@ -725,10 +631,8 @@ def main():
"type trigger words in the chat UI, or curl /_mock to change state",
flush=True,
)
try:
with contextlib.suppress(KeyboardInterrupt):
server.serve_forever()
except KeyboardInterrupt:
pass
if __name__ == "__main__":