diff --git a/CHANGELOG.md b/CHANGELOG.md
index 984be25440..e7ba5fef36 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,8 @@ Each release's page on GitHub has its notes and measurements. See [the 1.0.0 rel
## Unreleased
+- In one-call tool mode, a stream sends prose held after a hidden call envelope at the finish, so the streamed content equals the whole reply; other streams are unchanged (#2049). Thanks to @devopsfortunato.
+
- GLM tool calls stream their arguments as the model writes them, as Qwen-style XML calls already do, and the joined pieces equal the whole reply's arguments (#2050). Thanks to @devopsfortunato.
- Kolibri-1 on CUDA keeps its full layers' caches in pages of the core KV pool, the same way, with identical tokens when the store is off (#1620, part of #1657). Thanks to @jschmied.
diff --git a/zig/src/server/chat.zig b/zig/src/server/chat.zig
index 18f9847772..71fed1c7e0 100644
--- a/zig/src/server/chat.zig
+++ b/zig/src/server/chat.zig
@@ -242,7 +242,7 @@ pub fn generate(srv: *Server, cx: *Cx, prepared: Prepared, sink: ?Sink, gone: an
};
srv.release(preparing, null); // a background request waits only while a foreground one prepares
preparing = false;
- var gen: Generation = .{ .srv = srv, .a = a, .box = &box, .id = id, .reply_id = input.id, .seed = if (request.sampling) |v| v.seed else null, .sink = sink, .thinking = prepared.reasoning or (srv.info.reasoning and reply_text.isChannel(srv.markers)), .stops = .{ .strings = stops_opt.strings }, .ignore_eos = stops_opt.ignore_eos, .max_tokens = request.max_tokens, .tools = input.tools, .token_ids = prepared.token_ids, .lp = if (request.logprobs != null) .{} else null, .loop_stop = request.loop_stop };
+ var gen: Generation = .{ .srv = srv, .a = a, .box = &box, .id = id, .reply_id = input.id, .seed = if (request.sampling) |v| v.seed else null, .sink = sink, .thinking = prepared.reasoning or (srv.info.reasoning and reply_text.isChannel(srv.markers)), .stops = .{ .strings = stops_opt.strings }, .ignore_eos = stops_opt.ignore_eos, .max_tokens = request.max_tokens, .tools = input.tools, .token_ids = prepared.token_ids, .lp = if (request.logprobs != null) .{} else null, .loop_stop = request.loop_stop, .single = flag(input.fields, "tool_call_single") orelse false };
defer srv.noteRequest(prepared.prompt_len, box.cached orelse 0, gen.collected.items.len, box.stats.drafted, box.stats.accepted, box.stats.rounds, received, gen.first_ns, gen.last_ns, box.stats.prefill_seconds);
errdefer if (!gen.engine_done) gen.cancel(); // the engine writes to the mailbox until it says finished
const result: Failure!Reply = blk: {
@@ -285,6 +285,7 @@ const Generation = struct {
stops: reply_text.Stops,
ignore_eos: bool,
max_tokens: u32,
+ single: bool, // parallel_tool_calls: false: a cut call is no call, so its prose ends as the whole reply's does
token_ids: bool, // "return_token_ids": the response's tensorfold block carries the reply's token ids
tools: []const Value,
calls: ?tool_stream.Streamer = null,
@@ -497,7 +498,7 @@ const Generation = struct {
const thought = g.streamed_reasoning.text.items;
if (reasoning) |r| if (std.mem.startsWith(u8, r, thought) and r.len > thought.len)
try g.emit(sink, try deltaOf(a, "reasoning_content", r[thought.len..]));
- const shown = if (g.calls != null) try tool_stream.endProse(a, content, if (g.thinking) content else text, g.tools, reason) else content;
+ const shown = if (g.calls != null) try tool_stream.endProse(a, content, if (g.thinking) content else text, g.tools, reason, g.single) else content;
const sent = g.streamed.text.items;
const rest = if (std.mem.startsWith(u8, shown, sent) and shown.len > sent.len) shown[sent.len..] else "";
if (g.lp) |*l| (if (try l.delta(g.srv, a, g.thinking, g.ignore_eos, g.collected.items, rest, shown.len, true)) |d| try g.emit(sink, d)) else if (rest.len > 0) try g.emit(sink, .{ .string = rest });
diff --git a/zig/src/server/root.zig b/zig/src/server/root.zig
index 57442102bc..794cc76e0d 100644
--- a/zig/src/server/root.zig
+++ b/zig/src/server/root.zig
@@ -94,6 +94,7 @@ test {
_ = @import("tool_cut_test.zig");
_ = @import("tool_glm_stream_test.zig");
_ = @import("tool_glm_messages_test.zig");
+ _ = @import("tool_single_stream_test.zig");
_ = @import("logprobs_test.zig");
_ = @import("think_split_test.zig");
_ = @import("deepseek_reply_test.zig");
diff --git a/zig/src/server/tool_single_stream_test.zig b/zig/src/server/tool_single_stream_test.zig
new file mode 100644
index 0000000000..93165d8431
--- /dev/null
+++ b/zig/src/server/tool_single_stream_test.zig
@@ -0,0 +1,292 @@
+//! #1396: with parallel_tool_calls false a stream sends the whole reply's content, calls, arguments and finish reason.
+const std = @import("std");
+const api = @import("engine_api");
+const json = @import("json");
+const model_text = @import("model_text.zig");
+const openai = @import("openai.zig");
+const Server = @import("server.zig").Server;
+const Conn = @import("http_conn.zig").Conn;
+const pre = @import("stream_preflight_test.zig");
+const Allocator = std.mem.Allocator;
+
+/// Bytes as tokens; every prompt renders as "prompt" (6 tokens), as the preflight harness's text does.
+const ByteText = struct {
+ fn text(t: *@This()) model_text.Text {
+ return .{ .ctx = t, .vtable = &.{ .encode = encode, .decode = decode, .token_id = tokenId, .token_string = tokenString, .vocab_size = vocabSize, .eos_ids = eosIds, .render = render, .template_source = templateSource } };
+ }
+
+ fn encode(_: *anyopaque, a: Allocator, input: []const u8, _: bool) model_text.Error![]u32 {
+ const ids = try a.alloc(u32, input.len);
+ for (input, ids) |byte, *id| id.* = byte;
+ return ids;
+ }
+
+ fn decode(_: *anyopaque, a: Allocator, ids: []const u32) std.mem.Allocator.Error![]u8 {
+ const decoded = try a.alloc(u8, ids.len);
+ for (ids, decoded) |id, *byte| byte.* = @intCast(id);
+ return decoded;
+ }
+
+ fn tokenId(_: *anyopaque, _: []const u8) ?u32 {
+ return null;
+ }
+
+ fn tokenString(_: *anyopaque, a: Allocator, id: u32) std.mem.Allocator.Error![]u8 {
+ return std.fmt.allocPrint(a, "{d}", .{id});
+ }
+
+ fn vocabSize(_: *anyopaque) u32 {
+ return 256;
+ }
+
+ fn eosIds(_: *anyopaque) []const u32 {
+ return &.{};
+ }
+
+ fn render(_: *anyopaque, a: Allocator, _: json.Value, _: model_text.RenderOptions, _: *[]const u8) model_text.Error!model_text.Rendered {
+ return .{ .text = try (a.dupe(u8, "prompt")) };
+ }
+
+ fn templateSource(_: *anyopaque) []const u8 {
+ return "";
+ }
+};
+
+/// What a client got: a whole reply's status, or a stream's opening, with its content, calls, arguments and finish.
+const Got = struct {
+ opened: bool = false,
+ status: ?u16 = null,
+ calls: usize = 0,
+ content: std.ArrayList(u8) = .empty,
+ chunks: std.ArrayList([]const u8) = .empty, // every content delta as sent, in order
+ arguments: std.ArrayList(u8) = .empty,
+ finish: [16]u8 = undefined,
+ finish_len: usize = 0,
+ content_at_call: ?usize = null, // the content a stream had sent when its first call delta arrived
+
+ fn deinit(g: *Got) void {
+ for (g.chunks.items) |c| std.testing.allocator.free(c);
+ g.chunks.deinit(std.testing.allocator);
+ g.content.deinit(std.testing.allocator);
+ g.arguments.deinit(std.testing.allocator);
+ }
+
+ fn reason(g: *const Got) []const u8 {
+ return g.finish[0..g.finish_len];
+ }
+
+ fn keepFinish(g: *Got, choice: json.Value) void {
+ const f = choice.strField("finish_reason") orelse return;
+ g.finish_len = @min(f.len, g.finish.len);
+ @memcpy(g.finish[0..g.finish_len], f[0..g.finish_len]);
+ }
+
+ fn keepCall(g: *Got, call: json.Value) error{Closed}!void {
+ const f = call.get("function") orelse return;
+ const args = f.strField("arguments") orelse return;
+ g.arguments.appendSlice(std.testing.allocator, args) catch return error.Closed;
+ }
+
+ fn out(g: *Got) openai.Out {
+ return .{ .ctx = g, .vt = &.{ .open = open, .event = event, .reply = reply } };
+ }
+
+ fn open(ctx: *anyopaque, _: usize) error{Closed}!void {
+ const g: *Got = @ptrCast(@alignCast(ctx));
+ g.opened = true;
+ }
+
+ fn event(ctx: *anyopaque, payload: ?json.Value) error{Closed}!void {
+ const g: *Got = @ptrCast(@alignCast(ctx));
+ const p = payload orelse return;
+ const choices = p.get("choices") orelse return;
+ if (choices != .array or choices.array.len == 0) return;
+ const choice = choices.array[0];
+ g.keepFinish(choice);
+ const delta = choice.get("delta") orelse return;
+ if (delta.get("content")) |t| if (t == .string) {
+ g.content.appendSlice(std.testing.allocator, t.string) catch return error.Closed;
+ const one = std.testing.allocator.dupe(u8, t.string) catch return error.Closed;
+ g.chunks.append(std.testing.allocator, one) catch return error.Closed;
+ };
+ if (delta.get("tool_calls")) |t| if (t == .array) for (t.array) |call| {
+ if (call.get("id") != null) {
+ g.calls += 1;
+ if (g.content_at_call == null) g.content_at_call = g.content.items.len;
+ }
+ try g.keepCall(call);
+ };
+ }
+
+ fn reply(ctx: *anyopaque, status: u16, payload: json.Value) void {
+ const g: *Got = @ptrCast(@alignCast(ctx));
+ g.status = status;
+ const choices = payload.get("choices") orelse return;
+ if (choices != .array or choices.array.len == 0) return;
+ const choice = choices.array[0];
+ g.keepFinish(choice);
+ const message = choice.get("message") orelse return;
+ if (message.get("content")) |t| if (t == .string) g.content.appendSlice(std.testing.allocator, t.string) catch {};
+ if (message.get("tool_calls")) |t| if (t == .array) {
+ g.calls += t.array.len;
+ for (t.array) |call| g.keepCall(call) catch {};
+ };
+ }
+};
+
+/// The reply to ``body``: whole, or a stream whose content, calls, arguments and finish the sink keeps.
+fn send(engine: api.Engine, body: []const u8) !Got {
+ var text: ByteText = .{};
+ var srv = try Server.init(std.testing.allocator, std.testing.io, engine, text.text(), .{
+ .served_name = "test-model",
+ .model_ids = &.{"test-model"},
+ .enable_thinking = false,
+ .use_drafts = false,
+ }, null);
+ defer srv.deinit();
+ var arena = std.heap.ArenaAllocator.init(std.testing.allocator);
+ defer arena.deinit();
+ const a = arena.allocator();
+ const raw = (try json.parse(a, body)).ok;
+ var conn: Conn = .{ .fd = -1, .peer = "", .buf = &.{}, .gpa = a };
+ var got: Got = .{};
+ errdefer got.deinit();
+ openai.run(srv, a, got.out(), .{ .conn = &conn }, true, raw);
+ return got;
+}
+
+const head = "{\"model\":\"test-model\",\"messages\":[{\"role\":\"user\",\"content\":\"x\"}]," ++
+ "\"tools\":[{\"type\":\"function\",\"function\":{\"name\":\"lookup\",\"parameters\":{\"type\":\"object\"," ++
+ "\"properties\":{\"q\":{\"type\":\"string\"}}}}}]";
+
+/// A reply, the tokens its stream is cut at, and what both replies must send.
+const Case = struct {
+ reply: []const u8,
+ cut: ?usize = null,
+ content: ?[]const u8 = null,
+ calls: usize = 0,
+ arguments: []const u8 = "",
+ finish: []const u8 = "stop",
+ prose_before_calls: ?usize = null,
+ single: bool = true,
+};
+
+/// ``c`` whole and streamed, batched and not: both replies agree with each other and with ``c``.
+fn expectAgrees(c: Case) !void {
+ var arena: std.heap.ArenaAllocator = .init(std.testing.allocator);
+ defer arena.deinit();
+ const a = arena.allocator();
+ const tokens = try pre.TestText.encode(undefined, a, c.reply, false);
+ const body = try std.fmt.allocPrint(a, "{s}{s},\"max_tokens\":{d}", .{ head, if (c.single) ",\"parallel_tool_calls\":false" else "", c.cut orelse 2000 });
+ for ([_]bool{ false, true }) |batched| {
+ var backend: pre.CallEngine = .{ .tokens = tokens, .batched = batched };
+ var whole = try send(backend.engine(), try std.mem.concat(a, u8, &.{ body, ",\"stream\":false}" }));
+ defer whole.deinit();
+ var streamed = try send(backend.engine(), try std.mem.concat(a, u8, &.{ body, ",\"stream\":true}" }));
+ defer streamed.deinit();
+ try std.testing.expectEqual(@as(?u16, 200), whole.status);
+ try std.testing.expect(streamed.opened and streamed.status == null);
+ if (c.content) |content| try std.testing.expectEqualStrings(content, whole.content.items);
+ try std.testing.expectEqual(c.calls, whole.calls);
+ try std.testing.expectEqualStrings(c.arguments, whole.arguments.items);
+ try std.testing.expectEqualStrings(c.finish, whole.reason());
+ try std.testing.expectEqualStrings(whole.content.items, streamed.content.items);
+ try std.testing.expectEqual(whole.calls, streamed.calls);
+ try std.testing.expectEqualStrings(whole.arguments.items, streamed.arguments.items);
+ try std.testing.expectEqualStrings(whole.reason(), streamed.reason());
+ if (c.prose_before_calls) |n| try std.testing.expectEqual(n, streamed.content_at_call.?);
+ }
+}
+
+test "single: a rejected whole envelope stays in the streamed content as in the whole reply" {
+ const named = "Before.{\"name\":\"other\",\"arguments\":{}}Tail.";
+ const upper = "Before.{\"name\":\"other\",\"arguments\":{}}Tail.";
+ const mixed = "Before.{\"name\":\"other\",\"arguments\":{}}Tail.";
+ try expectAgrees(.{ .reply = named, .content = named });
+ try expectAgrees(.{ .reply = upper, .content = upper });
+ try expectAgrees(.{ .reply = mixed, .content = mixed });
+ const invalid = "Before.{invalid}Tail.";
+ try expectAgrees(.{ .reply = invalid, .content = invalid });
+}
+
+test "single: a cut reply ends its prose as the whole reply does" {
+ const first = "{\"name\":\"other\",\"arguments\":{}}Between.";
+ const cut = first ++ "lookupqabc";
+ try expectAgrees(.{ .reply = cut, .cut = cut.len, .content = first, .finish = "length" });
+ const progress = "\nWriting it.\n";
+ const xml = progress ++ "\n\n\nThe histor";
+ const glm = progress ++ "lookupqThe histor";
+ try expectAgrees(.{ .reply = xml ++ "y", .cut = xml.len, .content = progress, .finish = "length" });
+ try expectAgrees(.{ .reply = glm ++ "y", .cut = glm.len, .content = progress, .finish = "length" });
+}
+
+test "an unclosed DSML block stays in both replies, with or without a trailing newline, in either mode" {
+ const spaced = "Done.<\u{ff5c}DSML\u{ff5c} calls><\u{ff5c}DSML\u{ff5c} invoke name=\"lookup\"><\u{ff5c}DSML\u{ff5c} parameter name=\"q\" string=\"true\">x";
+ const compact = "Done.<\u{ff5c}DSML\u{ff5c}tool_calls><\u{ff5c}DSML\u{ff5c}invoke name=\"lookup\"><\u{ff5c}DSML\u{ff5c}parameter name=\"q\" string=\"true\">x";
+ inline for ([_]bool{ true, false }) |single| {
+ // a single-call reply's content is the parse's and stripped; with many calls it is the reply's own text
+ const end = if (single) "" else "\n";
+ try expectAgrees(.{ .reply = spaced, .content = spaced, .single = single });
+ try expectAgrees(.{ .reply = compact, .content = compact, .single = single });
+ try expectAgrees(.{ .reply = spaced ++ "\n", .content = spaced ++ end, .single = single });
+ try expectAgrees(.{ .reply = compact ++ "\n", .content = compact ++ end, .single = single });
+ }
+}
+
+test "single: an accepted call's prose streams before it and its tail after it" {
+ const before = "Before.";
+ const tail = "Tail.";
+ const call = "{\"name\":\"lookup\",\"arguments\":{\"q\":\"abc\"}}";
+ try expectAgrees(.{
+ .reply = before ++ call ++ tail,
+ .content = before ++ tail,
+ .calls = 1,
+ .arguments = "{\"q\":\"abc\"}",
+ .finish = "tool_calls",
+ .prose_before_calls = before.len,
+ });
+}
+
+/// The reply to ``reply`` whole: the content a stream's joined deltas must equal.
+fn wholeReply(a: Allocator, reply: []const u8) !Got {
+ const tokens = try pre.TestText.encode(undefined, a, reply, false);
+ var backend: pre.CallEngine = .{ .tokens = tokens };
+ const body = try std.fmt.allocPrint(a, "{s},\"max_tokens\":2000", .{head});
+ return send(backend.engine(), try std.mem.concat(a, u8, &.{ body, ",\"stream\":false}" }));
+}
+
+/// What one stream sends for ``reply``: default mode, or one-call mode with ``single``, batched or per token.
+fn streamedReply(a: Allocator, reply: []const u8, single: bool, batched: bool) !Got {
+ const tokens = try pre.TestText.encode(undefined, a, reply, false);
+ var backend: pre.CallEngine = .{ .tokens = tokens, .batched = batched };
+ const body = try std.fmt.allocPrint(a, "{s}{s},\"max_tokens\":2000", .{ head, if (single) ",\"parallel_tool_calls\":false" else "" });
+ return send(backend.engine(), try std.mem.concat(a, u8, &.{ body, ",\"stream\":true}" }));
+}
+
+/// The content chunks one stream sent, in order.
+fn expectChunks(want: []const []const u8, got: []const []const u8) !void {
+ try std.testing.expectEqual(want.len, got.len);
+ for (want, got) |one, seen| try std.testing.expectEqualStrings(one, seen);
+}
+
+test "a call-free reply ending in a newline streams the whole reply's content, chunk for chunk" {
+ const reply = "abc\n";
+ // a per-token engine sends the text byte by byte and then the newline; batched, the text and then its newline
+ const byte_by_byte = [_][]const u8{ "a", "b", "c", "\n" };
+ const in_one_chunk = [_][]const u8{ "abc", "\n" };
+ for ([_]bool{ false, true }) |batched| {
+ var arena: std.heap.ArenaAllocator = .init(std.testing.allocator);
+ defer arena.deinit();
+ const a = arena.allocator();
+ var whole = try wholeReply(a, reply);
+ defer whole.deinit();
+ try std.testing.expectEqualStrings(reply, whole.content.items);
+ for ([_]bool{ false, true }) |single| { // one-call mode sends this reply in the same chunks
+ var streamed = try streamedReply(a, reply, single, batched);
+ defer streamed.deinit();
+ const want: []const []const u8 = if (batched) in_one_chunk[0..] else byte_by_byte[0..];
+ try expectChunks(want, streamed.chunks.items);
+ try std.testing.expectEqualStrings(whole.content.items, streamed.content.items);
+ }
+ }
+}
diff --git a/zig/src/server/tool_stream.zig b/zig/src/server/tool_stream.zig
index efefe97d18..e322c4e753 100644
--- a/zig/src/server/tool_stream.zig
+++ b/zig/src/server/tool_stream.zig
@@ -347,8 +347,9 @@ pub const Streamer = struct {
}
};
-/// A streamed reply's final prose: ``cutProse`` when max_tokens left a call open, else ``finishedProse``.
-pub fn endProse(a: Allocator, content: []const u8, text: []const u8, tools: []const Value, reason: []const u8) Allocator.Error![]const u8 {
+/// A streamed reply's final prose: ``cutProse`` for a cut call, else ``finishedProse``; one-call mode rstrips it.
+pub fn endProse(a: Allocator, content: []const u8, text: []const u8, tools: []const Value, reason: []const u8, single: bool) Allocator.Error![]const u8 {
+ if (single) return reply_text.pyRstrip(try reply_text.finishedProse(a, content));
const cut = std.mem.eql(u8, reason, "length") and (try cutCalls(a, text, tools)) != null;
return if (cut) reply_text.cutProse(a, content) else reply_text.finishedProse(a, content);
}
@@ -407,24 +408,28 @@ fn nameLike(s: []const u8) bool {
pub const TextFilter = struct {
pending: std.ArrayList(u8) = .empty,
hidden: ?[]const u8 = null,
+ deferred: bool = false, // a closed envelope was hidden: later prose waits for the reply's final content
pub fn feed(f: *TextFilter, a: Allocator, text: []const u8) Allocator.Error![]const u8 {
+ if (f.deferred) return "";
try f.pending.appendSlice(a, text);
var out: std.ArrayList(u8) = .empty;
while (f.pending.items.len > 0) {
const p = f.pending.items;
if (f.hidden) |name| {
const close = try std.mem.concat(a, u8, &.{ "", name, ">" });
- const end = std.ascii.findIgnoreCase(p, close) orelse {
+ if (std.ascii.findIgnoreCase(p, close) == null) {
const keep = reply_text.charCount(close) - 1;
const chars = reply_text.charCount(p);
const tail = if (chars > keep) reply_text.afterChars(p, chars - keep) else p;
try f.reset(a, tail);
break;
- };
- try f.reset(a, p[end + close.len ..]);
+ }
+ // The whole reply may keep this envelope as text, so what follows cannot overtake it.
f.hidden = null;
- continue;
+ f.deferred = true;
+ f.pending.clearRetainingCapacity();
+ break;
}
const start = std.mem.indexOfScalar(u8, p, '<') orelse {
try out.appendSlice(a, p);
@@ -462,7 +467,7 @@ pub const TextFilter = struct {
}
pub fn finish(f: *TextFilter) []const u8 {
- const tail = if (f.hidden != null) "" else f.pending.items;
+ const tail = if (f.hidden != null or f.deferred) "" else f.pending.items;
f.pending = .empty;
return tail;
}
@@ -514,11 +519,12 @@ pub const Policy = struct {
}
/// Held-back malformed or unoffered call envelopes kept as reply text: ``final`` past the prose ``sent``.
- pub fn kept(_: *const Policy, final: []const u8, sent: []const u8) ?[]const u8 {
+ pub fn kept(p: *const Policy, final: []const u8, sent: []const u8) ?[]const u8 {
// Replies without parsed calls retain their leading whitespace; parsed content may be stripped.
const shown = if (std.mem.startsWith(u8, final, sent)) sent else reply_text.pyLstrip(sent);
if (!std.mem.startsWith(u8, final, shown)) return null;
const rest = final[shown.len..];
+ if (p.single) return if (rest.len > 0) rest else null;
return if (reply_text.pyStrip(rest).len > 0) rest else null;
}
};