vrr/koboldcpp
2
0
Fork 0
mirror of https://github.com/LostRuins/koboldcpp.git synced 2026-08-23 07:15:39 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge branch 'upstream' into concedo_experimental

Browse source 
# Conflicts:
#	docs/backend/SYCL.md
#	ggml/src/ggml-cpu/ggml-cpu.c
#	ggml/src/ggml-opencl/CMakeLists.txt
#	ggml/src/ggml-opencl/ggml-opencl.cpp
#	ggml/src/ggml-opencl/kernels/cvt.cl
#	ggml/src/ggml-opencl/kernels/gemv_moe_mxfp4_f32_ns.cl
#	ggml/src/ggml-opencl/kernels/gemv_moe_q4_k_f32_ns.cl
#	ggml/src/ggml-sycl/backend.hpp
#	ggml/src/ggml-sycl/common.hpp
#	ggml/src/ggml-sycl/dmmv.cpp
#	ggml/src/ggml-sycl/ggml-sycl.cpp
#	scripts/sync_vendor.py
#	tests/CMakeLists.txt
#	tests/test-alloc.cpp
#	tests/test-backend-ops.cpp
#	tests/test-chat.cpp
#	tests/test-gguf.cpp
#	tests/test-save-load-state.cpp
This commit is contained in:
Concedo 2026-07-13 20:53:37 +08:00
commit fa21872f97
76 changed files with 1399 additions and 1110 deletions
Download patch file Download diff file Expand all files Collapse all files

31
tools/server/README-dev.md
View file

@ -126,15 +126,15 @@ It is opt in via the `X-Conversation-Id` header on `POST /v1/chat/completions`.
The feature lives entirely in `server-stream.{h,cpp}` and rests on three types:
- `stream_session`: a bounded ring buffer (4 MiB cap, oldest bytes drop first) plus a condvar. `append` pushes raw SSE bytes, `read_from` drains from any offset and blocks for live bytes or finalize, `finalize` wakes readers, `cancel` stops the producer. One conv maps to at most one live session.
- `stream_session`: a bounded ring buffer (4 MiB cap, oldest bytes drop first) plus a condvar. `append` pushes raw SSE bytes, `read_from` drains from any offset and blocks for live bytes or finalize, `finalize` wakes readers, `cancel` sets the flag the producer polls. One conv maps to at most one live session.
- `stream_session_manager`: a file-static singleton (`g_stream_sessions`) inside `server-stream.cpp`, owns all sessions keyed by conv id, enforces the one conv one session invariant via `create_or_replace`, and runs a GC thread that drops completed sessions past their TTL. Exposed to main only through `server_stream_session_manager_start/stop`.
- `stream_pipe_producer` / `stream_pipe_consumer`: the write and read ends. The producer owns the session lifetime and finalizes it on destruction; the consumer is read only and never finalizes, so a reader detaching cannot kill a running generation.
The implementation is hidden in `server-stream.cpp` (pimpl). The header exposes only the route handler factories, `server_stream_session_attach_pipe`, `server_stream_aware_should_stop`, `server_stream_conv_id_from_headers` and the GC lifecycle; the session, manager and consumer types stay in the `.cpp`.
The implementation is hidden in `server-stream.cpp` (pimpl). The header exposes only the route handler factories, the `server_res_spipe` response base, `server_stream_conv_id_from_headers` and the GC lifecycle; the session, manager, consumer and the `server_stream_create_spipe` factory stay in the `.cpp`.
Producer side: `server_res_generator` attaches a producer pipe when the header is present. The HTTP content provider mirrors every chunk into the ring before writing it to the socket. While a pipe is attached, `server_stream_aware_should_stop` ignores peer disconnect, so a dropped socket does not stop generation: only an explicit `DELETE` does. When the peer leaves early, `on_complete` calls `close()`, which drains the rest of the generation into the ring on the http worker.
Producer side: `server_res_generator` extends `server_res_spipe`, which keeps all spipe logic out of the generic `server_http_res`. `set_req` attaches a producer when the header is present, and the wrapped `next` tees each chunk into the ring before the socket, so a chunk lost to a dead wire is already buffered. While attached, `should_stop` ignores peer disconnect: only a `DELETE` stops generation. On an early peer drop, `on_complete` drains the tail into the ring on the http worker.
Lifetime safety: the producer pipe holds a shared `alive` flag also captured by the session cancel hook. `~server_res_generator` calls `cleanup()` to clear that hook while the reader is still alive, so a `cancel` arriving during teardown can never call `stop()` on a freed response. This ordering is the most fragile part of the feature: finalizing or destroying the producer before `cleanup()` runs reintroduces a use after free.
Lifetime safety: the session holds no back reference to the response, so `spipe` is a plain `unique_ptr` touched only by the http worker. `cancel` raises an atomic the producer polls; the producer finalizes the session from its destructor, which also runs `~server_response_reader::stop()` to cancel the generation at the queue level. A `DELETE` stops work by raising the flag and letting the worker unwind.
Consumer side: `GET /v1/stream/<conv_id>?from=N` opens a `text/event-stream` that replays buffered bytes from offset `N` and blocks for live bytes, so the browser reattaches like a fresh EventSource. An offset below the dropped prefix returns 400.
@ -235,6 +235,29 @@ That requires `JSON.stringify` when formatted to message content:
}
```
Set `stream: true` in the request body to stream a tool's output as it runs, instead of waiting for it to finish. Only certain tools accept this (for ex. `exec_shell_command`);
returns 404 if tool doesn't support it.
Response is SSE stream, one `data: <json>` line per chunk:
```json
{"chunk": "hello\n"}
```
followed by a final event once the tool returns:
```json
{"done": true}
```
or, if `invoke()` threw:
```json
{"done": true, "error": "..."}
```
There is no `[DONE]` sentinel (unlike `/chat/completions`), the stream ends after the `done`
### Router mode: how child <--> router communicates
Upon spawning a new child process using `subprocess`, both child and router listen to the stdout/stderr (combined)

68
tools/server/server-chat.cpp
View file

@ -431,22 +431,70 @@ json server_chat_convert_anthropic_to_oai(const json & body) {
std::string tool_use_id = json_value(block, "tool_use_id", std::string());
auto result_content = json_value(block, "content", json());
std::string result_text;
if (result_content.is_string()) {
result_text = result_content.get<std::string>();
tool_results.push_back({
{"role", "tool"},
{"tool_call_id", tool_use_id},
{"content", result_content.get<std::string>()}
});
} else if (result_content.is_array()) {
// Single-pass: build both text and content_parts, decide format at the end
std::string result_text;
json content_parts = json::array();
bool has_images = false;
for (const auto & c : result_content) {
if (json_value(c, "type", std::string()) == "text") {
result_text += json_value(c, "text", std::string());
std::string c_type = json_value(c, "type", std::string());
if (c_type == "text") {
std::string text = json_value(c, "text", std::string());
result_text += text;
content_parts.push_back({
{"type", "text"},
{"text", text}
});
} else if (c_type == "image") {
has_images = true;
json source = json_value(c, "source", json::object());
std::string source_type = json_value(source, "type", std::string());
if (source_type == "base64") {
std::string media_type = json_value(source, "media_type", std::string("image/jpeg"));
std::string data = json_value(source, "data", std::string());
std::string url = "data:" + media_type + ";base64," + data;
content_parts.push_back({
{"type", "image_url"},
{"image_url", {{"url", url}}}
});
} else if (source_type == "url") {
content_parts.push_back({
{"type", "image_url"},
{"image_url", {{"url", json_value(source, "url", std::string())}}}
});
}
}
}
}
tool_results.push_back({
{"role", "tool"},
{"tool_call_id", tool_use_id},
{"content", result_text}
});
if (!has_images) {
// Text-only: collapse to a plain string for maximum compatibility
tool_results.push_back({
{"role", "tool"},
{"tool_call_id", tool_use_id},
{"content", result_text}
});
} else {
// Mixed or image-only: use array content parts (OpenAI multimodal tool format)
tool_results.push_back({
{"role", "tool"},
{"tool_call_id", tool_use_id},
{"content", content_parts}
});
}
} else {
tool_results.push_back({
{"role", "tool"},
{"tool_call_id", tool_use_id},
{"content", ""}
});
}
}
}

8
tools/server/server-common.cpp
View file

@ -705,7 +705,8 @@ server_tokens process_mtmd_prompt(mtmd_context * mctx, const std::string & promp
std::vector<server_tokens> inputs;
// multimodal
mtmd_input_text inp_txt = {
prompt.c_str(),
prompt.data(),
prompt.size(),
/* add_special */ true,
/* parse_special */ true,
};
@ -1116,7 +1117,8 @@ json oaicompat_chat_params_parse(
// Reasoning budget: pass parameters through to sampling layer
{
int reasoning_budget = json_value(body, "thinking_budget_tokens", -1);
int reasoning_budget = json_value(body, "reasoning_budget_tokens",
json_value(body, "thinking_budget_tokens", -1));
if (reasoning_budget == -1) {
reasoning_budget = opt.reasoning_budget;
}
@ -1125,7 +1127,7 @@ json oaicompat_chat_params_parse(
llama_params["reasoning_budget_tokens"] = reasoning_budget;
llama_params["reasoning_budget_start_tag"] = chat_params.thinking_start_tag;
llama_params["reasoning_budget_end_tag"] = chat_params.thinking_end_tag;
llama_params["reasoning_budget_message"] = opt.reasoning_budget_message;
llama_params["reasoning_budget_message"] = json_value(body, "reasoning_budget_message", opt.reasoning_budget_message);
llama_params["reasoning_control"] = json_value(body, "reasoning_control", false);
}
}

52
tools/server/server-context.cpp
View file

@ -2290,6 +2290,24 @@ private:
// n_tokens_cur: the number of tokens added to the batch for the current slot
void create_checkpoint(server_slot & slot, const int64_t n_tokens_cur, llama_pos pos_min, llama_pos pos_max) {
const int id_task = slot.task->id;
// evict checkpoints within min-step of a previous checkpoint, unless they were
// created by the current task
int64_t last = -1;
for (auto it = slot.prompt.checkpoints.begin(); it != slot.prompt.checkpoints.end(); ) {
if (it->id_task != id_task && last >= 0 && it->n_tokens <= last + params_base.checkpoint_min_step) {
SLT_TRC(slot, "erasing context checkpoint too close to an earlier one (pos_min = %d, pos_max = %d, n_tokens = %" PRId64 ", size = %.3f MiB)\n",
it->pos_min, it->pos_max, it->n_tokens, (float) it->size() / 1024 / 1024);
it = slot.prompt.checkpoints.erase(it);
continue;
}
last = it->n_tokens;
++it;
}
while (slot.prompt.checkpoints.size() >= (size_t) params_base.n_ctx_checkpoints) {
// make room for the new checkpoint, if needed
const auto & cur = slot.prompt.checkpoints.front();
@ -2302,6 +2320,8 @@ private:
auto & cur = slot.prompt.checkpoints.emplace_back();
cur.id_task = id_task;
// [TAG_CHECKPOINTS_FIX_POS_MIN]
// TODO: here we incorrectly deterimne that the saved checkpoint data covers the [pos_min, pos_max] range
// this is not true for SWA models: https://github.com/ggml-org/llama.cpp/pull/24411#issuecomment-4677983225
@ -3511,7 +3531,10 @@ private:
do_checkpoint = do_checkpoint && !has_mtmd;
// no need to create checkpoints that are too close together, unless it's the last user message
do_checkpoint = do_checkpoint && (slot.prompt.checkpoints.empty() || is_last_user_message || n_tokens_start > slot.prompt.checkpoints.back().n_tokens + params_base.checkpoint_min_step);
do_checkpoint = do_checkpoint && (
slot.prompt.checkpoints.empty() ||
is_last_user_message || near_prompt_end ||
n_tokens_start > slot.prompt.checkpoints.back().n_tokens + params_base.checkpoint_min_step);
SLT_DBG(slot, "main/do_checkpoint = %s, pos_min = %d, pos_max = %d\n", do_checkpoint ? "yes" : "no", pos_min, pos_max);
// note: we create the checkpoint before calling llama_decode(), so the current batch is not
@ -3979,11 +4002,9 @@ server_context_meta server_context::get_meta() const {
};
}
// generator-like API for HTTP response generation
// may have bypass_sleep = true if the task does not use ctx_server
struct server_res_generator : server_http_res {
struct server_res_generator : server_res_spipe {
server_response_reader rd;
server_res_generator(server_queue & queue_tasks, server_response & queue_results, int sleep_idle_seconds, bool bypass_sleep = false)
: rd(queue_tasks, queue_results, HTTP_POLLING_SECONDS) {
@ -3993,15 +4014,6 @@ struct server_res_generator : server_http_res {
queue_tasks.wait_until_no_sleep();
}
}
~server_res_generator() override {
// cleanup() must run while rd is still alive (rd is destroyed after this body returns)
if (spipe) {
spipe->cleanup();
}
}
void stop() override {
rd.stop();
}
void ok(const json & response_data) {
status = 200;
data = safe_json_to_str(response_data);
@ -4039,6 +4051,8 @@ std::unique_ptr<server_res_generator> server_routes::handle_completions_impl(
auto & rd = res->rd;
auto & params = this->params;
res->set_req(&req); // will also set spipe if needed
int32_t sse_ping_interval = params.sse_ping_interval;
try {
@ -4181,7 +4195,7 @@ std::unique_ptr<server_res_generator> server_routes::handle_completions_impl(
}
res->status = 200;
res->content_type = "text/event-stream";
res->next = [res_this = res.get(), res_type, sse_ping_interval, &req](std::string & output) -> bool {
res->set_next([res_this = res.get(), res_type, sse_ping_interval](std::string & output) -> bool {
static auto format_error = [](task_response_type res_type, const json & res_json) {
if (res_type == TASK_RESPONSE_TYPE_ANTHROPIC) {
return format_anthropic_sse({
@ -4193,7 +4207,9 @@ std::unique_ptr<server_res_generator> server_routes::handle_completions_impl(
}
};
auto effective_should_stop = server_stream_aware_should_stop(res_this, req.should_stop);
auto effective_should_stop = [&res_this]() {
return res_this->should_stop();
};
try {
if (effective_should_stop()) {
@ -4284,13 +4300,9 @@ std::unique_ptr<server_res_generator> server_routes::handle_completions_impl(
// terminate on exception
return false;
}
};
});
}
// attach a producer pipe to the response when X-Conversation-Id is present.
// the pipe mirrors SSE chunks into the ring buffer and wires up the cancel hook.
server_stream_session_attach_pipe(*res, req.headers);
return res;
}

19
tools/server/server-http.cpp
View file

@ -1,7 +1,6 @@
#include "common.h"
#include "http.h"
#include "server-http.h"
#include "server-stream.h"
#include "server-common.h"
#include "ui.h"
@ -530,33 +529,20 @@ static void process_handler_response(server_http_req_ptr && request, server_http
std::string chunk;
const bool has_next = response->next(chunk);
if (!chunk.empty()) {
// mirror into the ring buffer first, the session must reflect every SSE chunk
// whether or not the wire write below succeeds
if (response->spipe) {
response->spipe->write(chunk.data(), chunk.size());
}
if (!sink.write(chunk.data(), chunk.size())) {
// peer is gone, stop the wire path here
return false;
}
SRV_DBG("http: streamed chunk: %s\n", chunk.c_str());
}
if (!has_next) {
// producer reached its natural end on the wire, a later close() skips the drain
if (response->spipe) {
response->spipe->done();
}
sink.done();
SRV_DBG("%s", "http: stream ended\n");
}
return has_next;
};
const auto on_complete = [request = q_ptr, response = r_ptr](bool) mutable {
// on a dropped peer, close() drains the rest of the generation into the ring buffer
if (response->spipe) {
response->spipe->close();
}
response.reset(); // spipe destructor finalizes the session if attached
response->on_complete();
response.reset();
request.reset();
};
res.set_chunked_content_provider(content_type, chunked_content_provider, on_complete);
@ -564,6 +550,7 @@ static void process_handler_response(server_http_req_ptr && request, server_http
res.status = response->status;
set_headers(res, response->headers);
res.set_content(response->data, response->content_type);
response->on_complete();
}
}

11
tools/server/server-http.h
View file

@ -11,7 +11,6 @@
#include <unordered_map>
struct common_params;
struct stream_pipe_producer; // defined in server-stream.h
// generator-like API for HTTP response generation
// this object response with one of the 2 modes:
@ -25,19 +24,13 @@ struct server_http_res {
std::string data;
std::map<std::string, std::string> headers;
// if set, the stream survives a client disconnect: the producer pipe keeps draining into the
// ring buffer and finalizes the session on destruction, so no explicit on_stream_end is needed.
// shared_ptr (not unique_ptr) so the forward-declared type is safe to delete here.
std::shared_ptr<stream_pipe_producer> spipe;
std::function<bool(std::string &)> next = nullptr;
bool is_stream() const {
return next != nullptr;
}
// called when the session is cancelled (e.g. DELETE /v1/stream/<conv_id>).
// server_res_generator overrides this to stop its reader; the default is a no-op.
virtual void stop() {}
// fired before req and res are destroyed
virtual void on_complete() {}
virtual ~server_http_res() = default;
};

3
tools/server/server-models.cpp
View file

@ -219,13 +219,14 @@ void server_model_meta::update_caps() {
"LLAMA_ARG_MODEL_URL",
"LLAMA_ARG_MMPROJ",
"LLAMA_ARG_MMPROJ_URL",
"LLAMA_ARG_MMPROJ_AUTO",
"LLAMA_ARG_HF_REPO",
"LLAMA_ARG_HF_REPO_FILE",
});
params.offline = true;
common_models_handler handler = common_models_handler_init(params, LLAMA_EXAMPLE_SERVER);
common_models_handler_apply(handler, params); // note: this won't download the model because offline=true
if (params.mmproj.path.empty()) {
if (params.no_mmproj || params.mmproj.path.empty()) {
multimodal = { false, false };
} else {
multimodal = mtmd_get_cap_from_file(params.mmproj.path.c_str());

146
tools/server/server-stream.cpp
View file

@ -96,8 +96,6 @@ struct stream_session {
size_t dropped_prefix() const; // bytes evicted from the front due to cap
int64_t completed_at() const; // 0 while alive, unix seconds after finalize
void set_stop_producer(std::function<void()> fn);
void cancel();
private:
@ -109,7 +107,6 @@ private:
bool done;
std::atomic<bool> cancelled; // polled lock-free by the should_stop closure, no mu
int64_t completed_ts;
std::function<void()> stop_producer;
};
stream_session::stream_session(std::string conversation_id_, size_t max_bytes_)
: conversation_id(std::move(conversation_id_))
@ -217,26 +214,10 @@ int64_t stream_session::completed_at() const {
return completed_ts;
}
void stream_session::set_stop_producer(std::function<void()> fn) {
std::lock_guard<std::mutex> lock(mu);
stop_producer = std::move(fn);
}
void stream_session::cancel() {
// flip cancelled first so the producer-side server_stream_aware_should_stop can break out of the
// recv() wait even if remove_waiting_task_ids does not notify the condvar (the cancel task
// posted by rd.stop() will eventually notify, but we do not want to depend on that timing)
// the should_stop closure on both the producer and any HTTP reader polls is_cancelled()
// so flipping this is the only signal needed to unwind both sides
cancelled.store(true, std::memory_order_release);
// copy the hook under the lock then invoke outside, the producer side may grab queue locks
// and we do not want to hold our mu across that path
std::function<void()> fn;
{
std::lock_guard<std::mutex> lock(mu);
fn = stop_producer;
}
if (fn) {
fn();
}
}
bool stream_session::is_cancelled() const {
@ -325,8 +306,10 @@ void stream_session_manager::evict_and_cancel(const std::string & conversation_i
s = it->second;
sessions.erase(it);
}
// signal the producer side first so the inference is cancelled at the queue level,
// then finalize, which wakes any pending HTTP reader and lets the drain exit naturally
// cancel first so the producer's on_complete() drain loop and any pending HTTP reader
// observe is_cancelled() and stop pulling further output, then finalize to wake readers
// blocked in read_from(). note: this does not interrupt the underlying generation itself,
// which keeps running to its own natural stop condition (EOS/max_tokens)
s->cancel();
s->finalize();
}
@ -431,65 +414,15 @@ stream_pipe_producer::stream_pipe_producer(stream_session_ptr session)
}
stream_pipe_producer::~stream_pipe_producer() {
cleanup();
session_->finalize();
}
void stream_pipe_producer::cleanup() {
if (!alive_) {
return;
}
alive_->store(false, std::memory_order_release);
session_->set_stop_producer(nullptr);
alive_.reset();
}
bool stream_pipe_producer::write(const char * data, size_t len) {
return session_->append(data, len);
}
void stream_pipe_producer::done() {
done_ = true;
}
void stream_pipe_producer::close() {
// httplib bails its content provider the moment is_peer_alive() goes false, so pump the rest
// of the generation into the ring buffer here. a DELETE flips is_cancelled and cuts it short
if (done_ || session_->is_cancelled()) {
SRV_TRC("stream_pipe close: skip drain (done=%d cancelled=%d) conv=%s\n",
done_ ? 1 : 0, session_->is_cancelled() ? 1 : 0, session_->conversation_id.c_str());
return;
}
SRV_TRC("stream_pipe close: draining conv=%s\n", session_->conversation_id.c_str());
size_t drained = 0;
std::string chunk;
while (true) {
chunk.clear();
bool has_next = res_->next(chunk);
if (!chunk.empty()) {
write(chunk.data(), chunk.size());
drained += chunk.size();
}
if (!has_next) {
break;
}
}
SRV_TRC("stream_pipe close: drain ended conv=%s bytes=%zu\n", session_->conversation_id.c_str(), drained);
}
std::shared_ptr<stream_pipe_producer> stream_pipe_producer::create(stream_session_ptr session,
server_http_res & res) {
auto alive = std::make_shared<std::atomic<bool>>(true);
auto * res_ptr = &res;
session->set_stop_producer([alive, res_ptr]() {
if (alive->load(std::memory_order_acquire)) {
res_ptr->stop();
}
});
auto pipe = std::shared_ptr<stream_pipe_producer>(new stream_pipe_producer(std::move(session)));
pipe->alive_ = std::move(alive);
pipe->res_ = res_ptr;
return pipe;
stream_pipe_producer * stream_pipe_producer::create(stream_session_ptr session) {
return new stream_pipe_producer(std::move(session));
}
// stream_pipe_consumer
@ -661,21 +594,68 @@ std::string server_stream_conv_id_from_headers(const std::map<std::string, std::
return std::string();
}
void server_stream_session_attach_pipe(server_http_res & res, const std::map<std::string, std::string> & headers) {
static stream_pipe_producer * server_stream_create_spipe(const std::map<std::string, std::string> & headers) {
std::string conversation_id = server_stream_conv_id_from_headers(headers);
SRV_TRC("conv_id=%s (empty=%d)\n", conversation_id.c_str(), conversation_id.empty() ? 1 : 0);
if (conversation_id.empty()) {
return;
return nullptr;
}
auto session = g_stream_sessions.create_or_replace(conversation_id);
res.spipe = stream_pipe_producer::create(session, res);
return stream_pipe_producer::create(session);
}
std::function<bool()> server_stream_aware_should_stop(server_http_res * res, std::function<bool()> fallback) {
return [res, fallback = std::move(fallback)]() -> bool {
if (res->spipe) {
return res->spipe->is_cancelled();
//
// server_res_spipe
//
void server_res_spipe::set_req(const server_http_req * req) {
this->req = req;
// optionally attach spipe to the response when X-Conversation-Id is present
spipe.reset(server_stream_create_spipe(req->headers));
}
bool server_res_spipe::conn_alive() {
GGML_ASSERT(req != nullptr);
return !req->should_stop();
}
bool server_res_spipe::should_stop() {
if (spipe) {
// note: if DELETE /v1/stream/<conv_id> is called, is_cancelled() will be true
return spipe->is_cancelled();
} else {
return !conn_alive();
}
}
void server_res_spipe::on_complete() {
if (!spipe || next_finished) {
return;
}
std::string chunk;
while (!spipe->is_cancelled()) {
chunk.clear();
bool has_next = next_orig(chunk);
if (!chunk.empty()) {
spipe->write(chunk.data(), chunk.size());
}
return fallback();
if (!has_next) {
break;
}
}
}
void server_res_spipe::set_next(std::function<bool(std::string &)> next_fn) {
next_orig = std::move(next_fn);
next = [this](std::string & out) {
bool has_next = next_orig(out);
if (spipe) {
// if spipe is set, tee-style pipe input to both HTTP and spipe
spipe->write(out.data(), out.size());
}
if (!has_next) {
next_finished = true;
}
return has_next;
};
}

47
tools/server/server-stream.h
View file

@ -30,36 +30,15 @@ protected:
// producer end: writes chunks into the ring buffer and owns the session lifetime, finalizing it
// on destruction.
//
// lifetime safety: holds a shared_ptr<atomic<bool>> alive also captured by the session's
// stop_producer hook. cleanup() sets alive=false and clears the hook; it must run while the
// response the hook calls stop() on is still alive. ~server_res_generator() does this explicitly.
struct stream_pipe_producer : stream_pipe {
~stream_pipe_producer() override;
bool write(const char * data, size_t len);
// mark the natural end on the wire so a later close() is a no-op
void done();
// on a peer drop, pump the response next() into the ring buffer until done. runs on the http
// worker from on_complete, no-op after done() or cancel
void close();
// disarm the stop hook and drop the alive guard, must run while the response the hook
// references is still alive. idempotent, the destructor calls it too
void cleanup();
// res.stop() is invoked when the session is cancelled, the alive guard ensures stop() is not
// called after cleanup() has run
static std::shared_ptr<stream_pipe_producer> create(stream_session_ptr session, server_http_res & res);
static stream_pipe_producer * create(stream_session_ptr session);
private:
explicit stream_pipe_producer(stream_session_ptr session);
bool done_ = false;
std::shared_ptr<std::atomic<bool>> alive_;
server_http_res * res_ = nullptr;
};
void server_stream_session_manager_start();
@ -73,10 +52,22 @@ server_http_context::handler_t server_stream_make_delete_handler();
// extract the X-Conversation-Id header value (case-insensitive), empty when absent
std::string server_stream_conv_id_from_headers(const std::map<std::string, std::string> & headers);
// on an X-Conversation-Id header, create or replace the session and attach a producer pipe to res
void server_stream_session_attach_pipe(server_http_res & res, const std::map<std::string, std::string> & headers);
// implement tee-style pipe (spipe) for "stream replay" functionality
struct server_res_spipe : server_http_res {
private:
// if set, the stream survives a client disconnect:
// connection kept alive, output is forwarded to spipe and reuse later
std::unique_ptr<stream_pipe_producer> spipe;
// if spipe is set, use this next_orig to implement tee-style pipe
std::function<bool(std::string &)> next_orig;
const server_http_req * req = nullptr;
// set once next_orig reports no more data, so on_complete() doesn't re-drain a finished stream
bool next_finished = false;
// should_stop closure that ignores peer disconnect when a pipe is attached, so only an explicit
// DELETE stops the producer and generation keeps flowing into the ring buffer. without a pipe it
// delegates to fallback, the legacy non-resumable flow
std::function<bool()> server_stream_aware_should_stop(server_http_res * res, std::function<bool()> fallback);
public:
void set_req(const server_http_req * req);
bool conn_alive();
bool should_stop();
void on_complete() override;
void set_next(std::function<bool(std::string &)> next_fn);
};

172
tools/server/server-tools.cpp
View file

@ -12,6 +12,7 @@
#include <climits>
#include <algorithm>
#include <unordered_set>
#include <functional>
namespace fs = std::filesystem;
@ -51,7 +52,13 @@ public:
virtual bool write_file(const std::string & path, const std::string & content) const = 0;
// paths relative to `base`, '/'-separated; sets `err` if `base` isn't a directory
virtual std::vector<std::string> list_files(const std::string & base, std::string & err) const = 0;
virtual exec_result run(const std::vector<std::string> & args, size_t max_output, int timeout_secs) const = 0;
// on_chunk, if set, is called with each chunk of output as it is read (before truncation cuts in);
// returning false terminates the process early (e.g. the client disconnected)
virtual exec_result run(
const std::vector<std::string> & args,
size_t max_output,
int timeout_secs,
const std::function<bool(const std::string &)> & on_chunk = nullptr) const = 0;
};
class tools_io_basic : public tools_io {
@ -123,7 +130,11 @@ public:
return list_files_fallback(base);
}
exec_result run(const std::vector<std::string> & args, size_t max_output, int timeout_secs) const override {
exec_result run(
const std::vector<std::string> & args,
size_t max_output,
int timeout_secs,
const std::function<bool(const std::string &)> & on_chunk = nullptr) const override {
exec_result res;
subprocess_s proc;
@ -164,8 +175,14 @@ public:
size_t len = strlen(buf);
if (output.size() + len <= max_output) {
output.append(buf, len);
if (on_chunk && !on_chunk(std::string(buf, len))) {
subprocess_terminate(&proc);
break;
}
} else {
output.append(buf, max_output - output.size());
size_t remaining = max_output - output.size();
output.append(buf, remaining);
if (on_chunk && remaining > 0) on_chunk(std::string(buf, remaining));
truncated = true;
}
}
@ -287,7 +304,7 @@ struct server_tool_read_file : server_tool {
};
}
json invoke(json params) const override {
json invoke(json params, server_tool::stream *) const override {
std::string path = params.at("path").get<std::string>();
int start_line = json_value(params, "start_line", 1);
int end_line = json_value(params, "end_line", -1); // -1 = no limit
@ -376,7 +393,7 @@ struct server_tool_file_glob_search : server_tool {
};
}
json invoke(json params) const override {
json invoke(json params, server_tool::stream *) const override {
std::string base = params.at("path").get<std::string>();
std::string include = json_value(params, "include", std::string("**"));
std::string exclude = json_value(params, "exclude", std::string(""));
@ -457,7 +474,7 @@ struct server_tool_grep_search : server_tool {
};
}
json invoke(json params) const override {
json invoke(json params, server_tool::stream *) const override {
std::string path = params.at("path").get<std::string>();
std::string pat_str = params.at("pattern").get<std::string>();
std::string include = json_value(params, "include", std::string("**"));
@ -577,6 +594,7 @@ struct server_tool_exec_shell_command : server_tool {
name = "exec_shell_command";
display_name = "Execute shell command";
permission_write = true;
support_stream = true;
}
json get_definition() const override {
@ -598,7 +616,7 @@ struct server_tool_exec_shell_command : server_tool {
};
}
json invoke(json params) const override {
json invoke(json params, server_tool::stream * st) const override {
std::string command = params.at("command").get<std::string>();
int timeout = json_value(params, "timeout", 10);
size_t max_output = (size_t) json_value(params, "max_output_size", (int) SERVER_TOOL_EXEC_SHELL_COMMAND_MAX_OUTPUT_SIZE);
@ -612,7 +630,24 @@ struct server_tool_exec_shell_command : server_tool {
std::vector<std::string> args = {"sh", "-c", command};
#endif
auto io = make_tools_io(params);
auto io = make_tools_io(params);
if (st) {
auto res = io->run(args, max_output, timeout, [st](const std::string & chunk) {
st->push(chunk);
return !st->alive || st->alive();
});
if (st->alive && !st->alive()) {
return json();
}
std::string tail = string_format("\n[exit code: %d]", res.exit_code);
if (res.timed_out) {
tail += " [exit due to timed out]";
}
st->push(tail);
return json();
}
auto res = io->run(args, max_output, timeout);
std::string text_output = res.output;
@ -654,7 +689,7 @@ struct server_tool_write_file : server_tool {
};
}
json invoke(json params) const override {
json invoke(json params, server_tool::stream *) const override {
std::string path = params.at("path").get<std::string>();
std::string content = params.at("content").get<std::string>();
@ -710,7 +745,7 @@ struct server_tool_edit_file : server_tool {
};
}
json invoke(json params) const override {
json invoke(json params, server_tool::stream *) const override {
std::string path = params.at("path").get<std::string>();
const json & edits_json = params.at("edits");
@ -1018,7 +1053,7 @@ struct server_tool_get_datetime : server_tool {
};
}
json invoke(json) const override {
json invoke(json, server_tool::stream *) const override {
auto now = std::chrono::system_clock::now();
auto time = std::chrono::system_clock::to_time_t(now);
@ -1026,6 +1061,59 @@ struct server_tool_get_datetime : server_tool {
}
};
struct server_tool_stream_result : server_task_result {
std::string chunk;
bool done = false;
std::string error_msg;
json to_json() override {
if (!done) {
return {{"chunk", chunk}};
} else {
json result = {{"done", true}};
if (!error_msg.empty()) {
result["error"] = error_msg;
}
return result;
}
}
};
void server_tool::stream::push(const std::string & chunk) {
if (chunk.empty()) return;
auto r = std::make_unique<server_tool_stream_result>();
r->id = id;
r->chunk = chunk;
qr.send(std::move(r));
}
struct server_tools_res : server_http_res {
std::thread worker;
server_response * qr = nullptr; // set only for streaming responses
int id = -1;
~server_tools_res() override {
if (worker.joinable()) {
worker.join();
}
if (qr) {
qr->remove_waiting_task_id(id);
}
}
};
static server_tool & find_tool(std::vector<std::unique_ptr<server_tool>> & tools, const std::string & name, bool require_stream) {
for (auto & t : tools) {
if (t->name == name) {
if (require_stream && !t->support_stream) {
throw std::invalid_argument(string_format("tool \"%s\" does not support stream = true", name.c_str()));
}
return *t;
}
}
throw std::invalid_argument(string_format("unknown tool \"%s\"", name.c_str()));
}
//
// public API
//
@ -1090,16 +1178,63 @@ void server_tools::setup(const std::vector<std::string> & enabled_tools) {
};
handle_post = [this](const server_http_req & req) -> server_http_res_ptr {
auto res = std::make_unique<server_http_res>();
auto res = std::make_unique<server_tools_res>();
try {
json body = json::parse(req.body);
std::string tool_name = body.at("tool").get<std::string>();
json params = body.value("params", json::object());
json result = invoke(tool_name, params);
res->data = safe_json_to_str(result);
bool stream = body.value("stream", false);
server_tool & tool = find_tool(tools, tool_name, stream);
if (stream) {
int id = res_id.fetch_add(1);
queue_res.add_waiting_task_id(id);
res->qr = &queue_res;
res->id = id;
res->worker = std::thread([this, id, &req, &tool, params]() mutable {
server_tool::stream st{queue_res, id, [&req]() {
return !req.should_stop();
}};
auto done = std::make_unique<server_tool_stream_result>();
try {
tool.invoke(params, &st);
} catch (const std::exception & e) {
done->error_msg = e.what();
} catch (...) {
done->error_msg = "An unknown error occurred";
}
done->id = st.id;
done->done = true;
st.qr.send(std::move(done));
});
res->content_type = "text/event-stream";
res->status = 200;
res->next = [this, id](std::string & output) -> bool {
auto result = queue_res.recv(id);
auto * r = dynamic_cast<server_tool_stream_result *>(result.get());
GGML_ASSERT(r != nullptr);
output = "data: " + safe_json_to_str(r->to_json()) + "\n\n";
if (r->done) {
queue_res.remove_waiting_task_id(id);
return false;
}
return true;
};
} else {
json result = tool.invoke(params, nullptr);
res->status = 200;
res->data = safe_json_to_str(result);
}
} catch (const json::exception & e) {
res->status = 400;
res->data = safe_json_to_str(format_error_response(e.what(), ERROR_TYPE_INVALID_REQUEST));
} catch (const std::invalid_argument & e) {
res->status = 404;
res->data = safe_json_to_str(format_error_response(e.what(), ERROR_TYPE_INVALID_REQUEST));
} catch (const std::exception & e) {
SRV_ERR("got exception: %s\n", e.what());
res->status = 500;
@ -1108,12 +1243,3 @@ void server_tools::setup(const std::vector<std::string> & enabled_tools) {
return res;
};
}
json server_tools::invoke(const std::string & name, const json & params) {
for (auto & t : tools) {
if (t->name == name) {
return t->invoke(params);
}
}
return {{"error", "unknown tool: " + name}};
}

19
tools/server/server-tools.h
View file

@ -2,15 +2,27 @@
#include "server-common.h"
#include "server-http.h"
#include "server-queue.h"
#include <atomic>
#include <functional>
struct server_tool {
std::string name;
std::string display_name;
bool permission_write = false;
bool support_stream = false; // if true, output can be streamed
virtual ~server_tool() = default;
virtual json get_definition() const = 0;
virtual json invoke(json params) const = 0;
struct stream {
server_response & qr;
int id;
std::function<bool()> alive;
void push(const std::string & chunk);
};
virtual json invoke(json params, stream * st = nullptr) const = 0;
json to_json() const;
};
@ -18,8 +30,11 @@ struct server_tool {
struct server_tools {
std::vector<std::unique_ptr<server_tool>> tools;
// for streaming
server_response queue_res;
std::atomic<int> res_id{0};
void setup(const std::vector<std::string> & enabled_tools);
json invoke(const std::string & name, const json & params);
server_http_context::handler_t handle_get;
server_http_context::handler_t handle_post;

59
tools/server/tests/unit/test_compat_anthropic.py
View file

@ -402,6 +402,65 @@ def test_anthropic_tool_result_with_text():
assert len(res.body["content"]) > 0
def test_anthropic_tool_result_with_image():
"""Test tool result containing mixed text and image blocks
Verifies that image blocks inside Anthropic tool_result content are
properly converted to OpenAI image_url format rather than being
silently dropped. With a non-multimodal model, the converted image
triggers a clear error message instead of being ignored.
"""
server.jinja = True
server.start()
# Small 1x1 red PNG image in base64 (same as vision tests)
red_pixel_png = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8DwHwAFBQIAX8jx0gAAAABJRU5ErkJggg=="
res = server.make_request("POST", "/v1/messages", data={
"model": "test",
"max_tokens": 100,
"messages": [
{"role": "user", "content": "What is in this image?"},
{
"role": "assistant",
"content": [
{
"type": "tool_use",
"id": "tool_1",
"name": "read",
"input": {"file": "test.png"}
}
]
},
{
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": "tool_1",
"content": [
{"type": "text", "text": "File: test.png"},
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/png",
"data": red_pixel_png
}
}
]
}
]
}
]
})
# Without the fix, image block would cause "unsupported content[].type"
# With the fix, image is converted to image_url but tinyllama doesn't support images
assert res.status_code == 500
assert "image input is not supported" in res.body.get("error", {}).get("message", "").lower()
def test_anthropic_tool_result_error():
"""Test tool result with error flag"""
server.jinja = True

18
tools/server/tests/unit/test_tools_builtin.py
View file

@ -105,6 +105,24 @@ def test_tools_builtin_edit_file_rejects_non_unique_old_text():
os.remove(log_path)
def test_tools_builtin_exec_shell_command_stream():
global server
server.start()
events = list(server.make_stream_request("POST", "/tools", data={
"tool": "exec_shell_command",
"params": {"command": "echo hello"},
"stream": True,
}))
assert len(events) >= 2
assert events[-1]["done"] is True
assert not events[-1].get("error")
chunks = "".join(e["chunk"] for e in events[:-1])
assert "hello" in chunks
assert "[exit code: 0]" in chunks
def test_tools_builtin_edit_file_rejects_overlapping_edits():
global server
server.start()