diff --git a/CHANGELOG.md b/CHANGELOG.md index 08ab38f7fd..e22fd2326a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,10 @@ Each release's page on GitHub has its notes and measurements. See [the 1.0.0 rel ## Unreleased +- DeepSeek-V4.1 DSML calls whose first parameter is the long one now stream from the invoke name instead of waiting for that parameter to close, as XML and GLM calls already do; a block that breaks after its name leaves the call it began while the whole reply keeps its text (#2191, #1975). + +- DeepSeek-V4.1 DSML calls now stream from the invoke name, DSML's own tag, unlike GLM and XML calls, which have none; before, the header waited for the first parameter to close, so a call whose first parameter is the long one now streams while it is written. A DSML call malformed after its name may stream before the whole reply keeps it as text (#2191, #1975). + - Vision: image patch preparation is rewritten from published model cards and tested behaviour, with byte-equal grid sizes, normalized pixels and cache identities in 55,503 host cases (#2271). - GLM on CUDA: the recipe notes that `TF_GLM_CHUNK_SOLO=4096` made prompts 12-14% slower with NVIDIA's NVFP4 checkpoint on two GB10s at `--parallel 1`, so leave it unset there (#2288). diff --git a/zig/src/server/stream_dsml_test.zig b/zig/src/server/stream_dsml_test.zig index 25d6e7713e..ff2c316fee 100644 --- a/zig/src/server/stream_dsml_test.zig +++ b/zig/src/server/stream_dsml_test.zig @@ -65,8 +65,14 @@ test "unfinished and malformed DSML never emit their complete nested call in con defer arena.deinit(); const a = arena.allocator(); for (spellings) |sp| { - const prefix = try std.mem.concat(a, u8, &.{ sp.prefix, "" }); - for ([_][]const u8{ prefix, try std.mem.concat(a, u8, &.{ prefix, sp.unclosed }) }) |raw| { + const nested = ""; + const prefix = try std.mem.concat(a, u8, &.{ sp.prefix, nested }); + // the whole reply keeps each block as text; the stream sent the header, the nested XML as its value + for ([_][2][]const u8{ + .{ prefix, "{\"q\":\"" ++ nested }, + .{ try std.mem.concat(a, u8, &.{ prefix, sp.unclosed }), "{\"q\":\"" ++ nested ++ "\"" }, + }) |case| { + const raw = case[0]; const tokens = try TestText.encode(undefined, a, raw, false); for ([_]bool{ false, true }) |batched| { var backend: CallEngine = .{ .tokens = tokens, .batched = batched }; @@ -75,8 +81,8 @@ test "unfinished and malformed DSML never emit their complete nested call in con try std.testing.expectEqual(@as(?u16, 200), whole.status); try std.testing.expect(streamed.opened and streamed.status == null); try std.testing.expectEqual(@as(usize, 0), whole.calls); - try std.testing.expectEqual(@as(usize, 0), streamed.calls); - try std.testing.expectEqual(@as(usize, 0), streamed.arguments_len); + try std.testing.expectEqual(@as(usize, 1), streamed.calls); + try std.testing.expectEqualStrings(case[1], streamed.arguments[0..streamed.arguments_len]); try std.testing.expectEqual(@as(i64, 0), streamed.preparing); } } @@ -147,19 +153,22 @@ test "DSML spellings mixed in one block stay text in whole and streamed consumer defer arena.deinit(); const a = arena.allocator(); const v41 = spellings[1]; - const mixed = [_][]const u8{ - try std.mem.concat(a, u8, &.{ v41.prefix, "a", spellings[0].suffix }), - try std.mem.concat(a, u8, &.{ spellings[0].prefix, "a", v41.suffix }), - "<｜DSML｜ calls><｜DSML｜invoke name=\"lookup\"><｜DSML｜parameter name=\"q\" string=\"true\">a", + // a block whose opener spelling and closers disagree: the whole reply keeps it as text, the stream has the header + const cases = [_]struct { []const u8, usize, []const u8 }{ + .{ try std.mem.concat(a, u8, &.{ v41.prefix, "a", spellings[0].suffix }), 1, "{\"q\":\"a" }, + .{ try std.mem.concat(a, u8, &.{ spellings[0].prefix, "a", v41.suffix }), 1, "{\"q\":\"a" }, + .{ "<｜DSML｜ calls><｜DSML｜invoke name=\"lookup\"><｜DSML｜parameter name=\"q\" string=\"true\">a", 0, "" }, }; - for (mixed) |raw| { + for (cases) |case| { + const raw = case[0]; const tokens = try TestText.encode(undefined, a, raw, false); for ([_]bool{ false, true }) |batched| { var backend: CallEngine = .{ .tokens = tokens, .batched = batched }; const whole = try sendWith(backend.engine(), .chat, dsml_body ++ ",\"stream\":false}", .{}); const streamed = try sendWith(backend.engine(), .chat, dsml_body ++ ",\"stream\":true}", .{}); try std.testing.expectEqual(@as(usize, 0), whole.calls); - try std.testing.expectEqual(@as(usize, 0), streamed.calls); + try std.testing.expectEqual(case[1], streamed.calls); + try std.testing.expectEqualStrings(case[2], streamed.arguments[0..streamed.arguments_len]); try std.testing.expectEqualStrings(raw[0..@min(raw.len, whole.text.len)], whole.content()); try std.testing.expectEqualStrings(whole.content(), streamed.content()); try std.testing.expectEqual(.stop, whole.finish.?); diff --git a/zig/src/server/tool_dsml_reply_test.zig b/zig/src/server/tool_dsml_reply_test.zig index 3f6be50338..0fee519393 100644 --- a/zig/src/server/tool_dsml_reply_test.zig +++ b/zig/src/server/tool_dsml_reply_test.zig @@ -67,7 +67,7 @@ test "DSML calls agree whole and streamed in every spelling, with prose, typed v }; } -test "a DSML block that breaks before its first header fails as it did: no call, its text kept, whole and streamed" { +test "a DSML block that breaks before its invoke name fails as it did: no call, its text kept, whole and streamed" { var arena: std.heap.ArenaAllocator = .init(std.testing.allocator); defer arena.deinit(); const a = arena.allocator(); @@ -90,7 +90,10 @@ test "a DSML call max_tokens cuts after its header returns cut whole and streame const reply = try std.mem.concat(a, u8, &.{ "Writing it.\n", try t.block(a, sp, &.{try t.invoke(a, sp, "write_file", &.{ .{ "path", true, "notes.txt" }, .{ "content", true, "The history of the bicycle" } })}) }); const cases = [_]struct { usize, usize, []const u8 }{ .{ std.mem.indexOf(u8, reply, " the bicycle").?, 1, "{\"path\":\"notes.txt\",\"content\":\"The history of" }, - .{ std.mem.indexOf(u8, reply, "tes.txt").?, 0, "" }, + // cut inside the first parameter: its header is out, so the cut reply returns the call it began + .{ std.mem.indexOf(u8, reply, "tes.txt").?, 1, "{\"path\":\"no" }, + // cut inside the invoke name: no header went out, so both replies make no call + .{ std.mem.indexOf(u8, reply, "write_file").?, 0, "" }, }; for (cases) |case| { const body = try chatBody(a, t.tools_text, try std.fmt.allocPrint(a, ",\"max_tokens\":{d}", .{case[0]})); diff --git a/zig/src/server/tool_dsml_stream.zig b/zig/src/server/tool_dsml_stream.zig index 953514cae4..64f7277495 100644 --- a/zig/src/server/tool_dsml_stream.zig +++ b/zig/src/server/tool_dsml_stream.zig @@ -1,4 +1,4 @@ -//! DSML invokes as written: a call goes out once its first parameter is whole in the block's spelling, then streams. +//! DSML invokes as written: a call goes out at its invoke name, then its values' JSON bytes as they arrive. const std = @import("std"); const json = @import("json"); const reply_text = @import("reply_text.zig"); @@ -23,12 +23,9 @@ pub const Dsml = struct { at: usize = 0, // where the stage's reading resumes from: usize = 0, // the open name's, key's or value's first byte name: []const u8 = "", // the open invoke's offered name - header: bool = false, // the open invoke's header went out - params: usize = 0, // the open invoke's whole parameters sent: usize = 0, // the block's headers that went out typed: bool = false, // the open value is JSON (``string="false"``) value_sent: usize = 0, // the open value's next byte to send - held: std.ArrayList(u8) = .empty, // the open invoke's arguments, written before its header }; /// The reply's ``text`` so far: the deltas the open DSML block adds, and what the Streamer should do next. @@ -36,7 +33,7 @@ pub fn feed(s: anytype, text: []const u8, out: *std.ArrayList(Value)) Allocator. while (true) { const next = switch (s.dsml.stage) { .invokes => invokes(s, text), - .name => try name(s, text), + .name => try name(s, text, out), .params => try params(s, text, out), .key => try key(s, text, out), .value => try value(s, text, out), @@ -48,7 +45,7 @@ pub fn feed(s: anytype, text: []const u8, out: *std.ArrayList(Value)) Allocator. /// The cut the reply's end left inside a streamed value: the bytes held back for a closer that never came. pub fn cut(s: anytype, text: []const u8, out: *std.ArrayList(Value)) Allocator.Error!void { const d = &s.dsml; - if (d.stage != .value or !d.header) return; + if (d.stage != .value) return; const end = text.len - closerTail(text[@min(d.from, text.len)..], d.sp); try sendValue(s, text, end, out); if (d.typed) try send(s, out, try s.a.dupe(u8, s.buffer.items)); // the value's JSON never closed @@ -77,8 +74,8 @@ fn invokes(s: anytype, text: []const u8) Next { return if (open == .part or close == .part) .wait else .rejected; } -/// An invoke's name: the bytes to the first quote, which ``>`` must follow, naming an offered tool. -fn name(s: anytype, text: []const u8) Allocator.Error!Next { +/// An invoke's name: the bytes to the first quote, then ``>``; an offered name sends its header. +fn name(s: anytype, text: []const u8, out: *std.ArrayList(Value)) Allocator.Error!Next { const d = &s.dsml; const q = std.mem.indexOfScalarPos(u8, text, d.at, '"'); if (has(text, d.from, d.at, q orelse text.len, d.sp.close)) return .rejected; // the block closed in the name @@ -91,10 +88,8 @@ fn name(s: anytype, text: []const u8) Allocator.Error!Next { if (text[at + 1] != '>') return .rejected; const written = reply_text.pyStrip(text[d.from..at]); d.name = s.known.get(try std.ascii.allocLowerString(s.a, written)) orelse return .rejected; - d.header = false; - d.params = 0; - d.held.clearRetainingCapacity(); - s.args_open = false; + try s.begin(d.name, out); + d.sent += 1; d.at = at + 2; d.stage = .params; return .more; @@ -107,14 +102,12 @@ fn params(s: anytype, text: []const u8, out: *std.ArrayList(Value)) Allocator.Er const param = tagAt(text, i, d.sp.param_open); const close = tagAt(text, i, d.sp.invoke_close); if (param == .whole) { - if (d.params > 0) try decide(s, out); // a whole parameter, then this spelling's next opener d.from = i + d.sp.param_open.len; d.at = d.from; d.stage = .key; return .more; } if (close == .whole) { - try decide(s, out); try send(s, out, if (s.args_open) "}" else "{}"); d.at = i + d.sp.invoke_close.len; d.stage = .invokes; @@ -172,28 +165,14 @@ fn value(s: anytype, text: []const u8, out: *std.ArrayList(Value)) Allocator.Err }; try send(s, out, try json.stringify(s.a, v, .{ .ascii = false, .compact = true })); } else try send(s, out, "\""); - d.params += 1; d.at = e + d.sp.param_close.len; d.stage = .params; return .more; } -/// The open invoke's header, once, then the arguments written before it as one piece. -fn decide(s: anytype, out: *std.ArrayList(Value)) Allocator.Error!void { - const d = &s.dsml; - if (d.header) return; - const args_open = s.args_open; - try s.begin(d.name, out); - s.args_open = args_open; - d.header = true; - d.sent += 1; - if (d.held.items.len > 0) try out.append(s.a, try s.args(try s.a.dupe(u8, d.held.items))); - d.held.clearRetainingCapacity(); -} - -/// An arguments piece: out under the open header, else held until the header goes out. +/// An arguments piece: out under the invoke's header, which went out before any key or value wrote one. fn send(s: anytype, out: *std.ArrayList(Value), piece: []const u8) Allocator.Error!void { - if (s.dsml.header) try out.append(s.a, try s.args(piece)) else try s.dsml.held.appendSlice(s.a, piece); + try out.append(s.a, try s.args(piece)); } /// The value's bytes from ``value_sent`` to ``end``: escaped into the arguments, or into the buffer JSON decodes. diff --git a/zig/src/server/tool_dsml_stream_test.zig b/zig/src/server/tool_dsml_stream_test.zig index 0a5e07c041..468b77bee7 100644 --- a/zig/src/server/tool_dsml_stream_test.zig +++ b/zig/src/server/tool_dsml_stream_test.zig @@ -198,8 +198,21 @@ test "seeded DSML blocks stream every call before their closer and join as the w try std.testing.expect(valid > 360); } -/// Blocks that break before their first call's header could go out: the whole reply keeps each as text. +/// Blocks that break before the invoke's name is whole: no header can go out, so nothing streams. pub fn earlyBreaks(a: Allocator, sp: *const Spelling) ![]const []const u8 { + const q = "\" string=\"true\">"; + var out: std.ArrayList([]const u8) = .empty; + for ([_][]const []const u8{ + &.{ sp.open, sp.invoke_open, "launch\">", sp.param_open, "q", q, "a", sp.param_close, sp.invoke_close, sp.close }, // not offered + &.{ sp.open, "x", sp.invoke_open, "lookup\">", sp.invoke_close, sp.close }, // text before the invoke + &.{ sp.open, sp.invoke_open, "lookup\" >", sp.invoke_close, sp.close }, // the name's quote without > + &.{ sp.open, "\n", sp.close }, // no invoke + }) |parts| try out.append(a, try std.mem.concat(a, u8, parts)); + return out.items; +} + +/// Blocks whose invoke is whole and offered, then break: the header stands while the whole reply keeps the text. +pub fn lateBreaks(a: Allocator, sp: *const Spelling) ![]const []const u8 { const q = "\" string=\"true\">"; // the grammar's spelling shares tool_calls' invoke tags and calls' parameter tags: another spelling per tag const other_invoke = &spellings[if (sp == &spellings[1]) 0 else 1]; @@ -207,9 +220,6 @@ pub fn earlyBreaks(a: Allocator, sp: *const Spelling) ![]const []const u8 { const first = try std.mem.concat(a, u8, &.{ sp.open, sp.invoke_open, "lookup\">", sp.param_open, "q", q, "a", sp.param_close }); var out: std.ArrayList([]const u8) = .empty; for ([_][]const []const u8{ - &.{ sp.open, sp.invoke_open, "launch\">", sp.param_open, "q", q, "a", sp.param_close, sp.invoke_close, sp.close }, // not offered - &.{ sp.open, "x", sp.invoke_open, "lookup\">", sp.invoke_close, sp.close }, // text before the invoke - &.{ sp.open, sp.invoke_open, "lookup\" >", sp.invoke_close, sp.close }, // the name's quote without > &.{ sp.open, sp.invoke_open, "lookup\">", sp.param_open, "q\" string=\"yes\">a", sp.param_close, sp.invoke_close, sp.close }, &.{ sp.open, sp.invoke_open, "lookup\">", sp.param_open, "q\">a", sp.param_close, sp.invoke_close, sp.close }, &.{ sp.open, sp.invoke_open, "lookup\">", sp.param_open, "n\" string=\"false\">{oops", sp.param_close, sp.param_open, "q", q, "b", sp.param_close, sp.invoke_close, sp.close }, @@ -218,14 +228,13 @@ pub fn earlyBreaks(a: Allocator, sp: *const Spelling) ![]const []const u8 { &.{ first, other_invoke.invoke_close, sp.close }, // another spelling's invoke closer &.{ first, "x", sp.param_open, "n", q, "b", sp.param_close, sp.invoke_close, sp.close }, // text between parameters &.{ sp.open, sp.invoke_open, "lookup\">", sp.param_open, "q", q, "a", sp.invoke_close, "b", sp.param_close, sp.invoke_close, sp.close }, - &.{ sp.open, "\n", sp.close }, // no invoke &.{ sp.open, sp.invoke_open, "lookup\">", other.param_open, "q", q, "a", other.param_close, sp.invoke_close, sp.close }, &.{ sp.open, sp.invoke_open, "lookup\">", sp.param_open, "q", q, "a", other.param_close, other_invoke.invoke_close, other.close }, }) |parts| try out.append(a, try std.mem.concat(a, u8, parts)); return out.items; } -test "a DSML block that breaks before its first header sends nothing at any byte, as the whole reply makes no call" { +test "a DSML block that breaks before its invoke name sends nothing at any byte, as the whole reply makes no call" { var arena: std.heap.ArenaAllocator = .init(std.testing.allocator); defer arena.deinit(); const a = arena.allocator(); @@ -244,6 +253,20 @@ test "a DSML block that breaks before its first header sends nothing at any byte }; } +test "a DSML block that breaks after its invoke name: the header stands while the whole reply keeps the text" { + var arena: std.heap.ArenaAllocator = .init(std.testing.allocator); + defer arena.deinit(); + const a = arena.allocator(); + const tools = try toolsOf(a); + for (spellings) |*sp| for (try lateBreaks(a, sp)) |text| { + try std.testing.expect((try tool_parse.parse(a, text, tools, null)).calls == null); + const deltas = try bytewise(a, tools, text); + // the header the invoke name earned is out and cannot be taken back; no second call is invented + try std.testing.expectEqualStrings("lookup ", try names(a, deltas)); + try std.testing.expect(deltas.len > 0); + }; +} + test "a DSML block that breaks after a header went out: the whole reply keeps it as text, the stream sent the call" { var arena: std.heap.ArenaAllocator = .init(std.testing.allocator); defer arena.deinit(); @@ -264,6 +287,33 @@ test "a DSML block that breaks after a header went out: the whole reply keeps it } } +test "a call whose first parameter is long streams while it is written, in every spelling" { + var arena: std.heap.ArenaAllocator = .init(std.testing.allocator); + defer arena.deinit(); + const a = arena.allocator(); + const tools = try toolsOf(a); + const value = try a.alloc(u8, 2000); + @memset(value, 'x'); + for (spellings) |*sp| { + const text = try block(a, sp, &.{try invoke(a, sp, "write_file", &.{.{ "content", true, value }})}); + const closer = std.mem.lastIndexOf(u8, text, sp.param_close).?; + var s = try Streamer.init(a, tools); + var out: std.ArrayList(Value) = .empty; + var early: usize = 0; + var at: usize = 0; + while (at < text.len) : (at += 3) { + const end = @min(at + 3, text.len); + try s.feed(text[0..end], &out); + if (end < closer) early = out.items.len; + } + // the header leaves at the name, so the value's pieces go out before its parameter closes + try std.testing.expectEqualStrings("write_file ", try names(a, out.items)); + try std.testing.expect(early > 2); + try std.testing.expectEqualStrings((try whole(a, tools, text)).args, try joined(a, out.items)); + try std.testing.expect(!s.open()); + } +} + test "a repeated DSML key streams both pairs, as XML and GLM do; the whole reply keeps the last" { var arena: std.heap.ArenaAllocator = .init(std.testing.allocator); defer arena.deinit(); @@ -274,7 +324,7 @@ test "a repeated DSML key streams both pairs, as XML and GLM do; the whole reply try std.testing.expectEqualStrings("{\"q\":\"b\"}", (try whole(a, tools, text)).args); } -test "a cut DSML value replays from the stream; before its first header the stream makes no call" { +test "a cut DSML value replays from the stream; a cut before the invoke name makes no call" { var arena: std.heap.ArenaAllocator = .init(std.testing.allocator); defer arena.deinit(); const a = arena.allocator(); @@ -285,7 +335,9 @@ test "a cut DSML value replays from the stream; before its first header the stre .{ try std.mem.concat(a, u8, &.{ head, "content\" string=\"true\">The history of12" }), "{\"path\":\"notes.txt\",\"limit\":12", true }, .{ try std.mem.concat(a, u8, &.{ head, "content\" str" }), "{\"path\":\"notes.txt\"", true }, - .{ try std.mem.concat(a, u8, &.{ sp.open, sp.invoke_open, "write_file\">", sp.param_open, "path\" string=\"true\">notes" }), "", false }, + .{ try std.mem.concat(a, u8, &.{ sp.open, sp.invoke_open, "write_file\">", sp.param_open, "path\" string=\"true\">notes" }), "{\"path\":\"notes", true }, + // cut inside the invoke's name: no header went out, so the stream makes no call + .{ try std.mem.concat(a, u8, &.{ sp.open, sp.invoke_open, "write_fi" }), "", false }, }; for (cases) |case| { var s = try Streamer.init(a, tools); diff --git a/zig/src/server/tool_single_stream_test.zig b/zig/src/server/tool_single_stream_test.zig index 93165d8431..953d19b276 100644 --- a/zig/src/server/tool_single_stream_test.zig +++ b/zig/src/server/tool_single_stream_test.zig @@ -220,16 +220,41 @@ test "single: a cut reply ends its prose as the whole reply does" { 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" { +test "an unclosed DSML block streams the call it began while both replies keep its text" { 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 }); + inline for ([_][3][]const u8{ + .{ spaced, spaced, "{\"q\":\"x" }, + .{ compact, compact, "{\"q\":\"x" }, + .{ spaced ++ "\n", spaced ++ end, "{\"q\":\"x\\n" }, + .{ compact ++ "\n", compact ++ end, "{\"q\":\"x\\n" }, + }) |pair| { + var arena: std.heap.ArenaAllocator = .init(std.testing.allocator); + defer arena.deinit(); + const a = arena.allocator(); + const tokens = try pre.TestText.encode(undefined, a, pair[0], false); + const body = try std.fmt.allocPrint(a, "{s}{s},\"max_tokens\":2000", .{ head, if (single) ",\"parallel_tool_calls\":false" else "" }); + 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); + // the whole reply keeps the unclosed block as text and makes no call + try std.testing.expectEqualStrings(pair[1], whole.content.items); + try std.testing.expectEqual(@as(usize, 0), whole.calls); + try std.testing.expectEqualStrings(whole.content.items, streamed.content.items); + // the stream shows the call the name began, but the one-call policy withholds an unclosed one + try std.testing.expectEqual(@as(usize, if (single) 0 else 1), streamed.calls); + if (single) try std.testing.expectEqualStrings("", streamed.arguments.items) else try std.testing.expectEqualStrings(pair[2], streamed.arguments.items); + try std.testing.expectEqualStrings(whole.reason(), streamed.reason()); + } + } } } diff --git a/zig/src/server/tool_stream_test.zig b/zig/src/server/tool_stream_test.zig index ff56c85d5f..b1177cec1e 100644 --- a/zig/src/server/tool_stream_test.zig +++ b/zig/src/server/tool_stream_test.zig @@ -180,33 +180,49 @@ test "DSML containment survives every byte boundary and preserves adjacent call }; } -test "unfinished DSML retains containment through byte feeds and final closeCall" { - const literals = [_][]const u8{ - "", - "", - "{\"name\":\"lookup\",\"arguments\":{}", +test "an unfinished DSML block streams the call it began, its nested XML as argument text" { + // the nested XML literal, and the arguments its block's value streams up to the reply's end + const literals = [_][2][]const u8{ + .{ "", "{\"q\":\"" }, + .{ "", "{\"q\":\"" }, + .{ "{\"name\":\"lookup\",\"arguments\":{}", "{\"q\":\"{\\\"name\\\":\\\"lookup\\\",\\\"arguments\\\":{}" }, }; for (dsml) |sp| for (literals) |literal| { var arena: std.heap.ArenaAllocator = .init(std.testing.allocator); defer arena.deinit(); const a = arena.allocator(); const tools = (try json.parse(a, "[{\"type\":\"function\",\"function\":{\"name\":\"lookup\"}}]")).ok.array; - const text = try std.mem.concat(a, u8, &.{ sp[0], literal }); + const text = try std.mem.concat(a, u8, &.{ sp[0], literal[0] }); + const name_end = std.mem.indexOf(u8, text, "\">").? + 2; // the byte the invoke's name ends on var s = try Streamer.init(a, tools); var out: std.ArrayList(Value) = .empty; + var last: usize = 0; for (0..text.len + 1) |n| { try s.feed(text[0..n], &out); - try std.testing.expectEqual(@as(usize, 0), out.items.len); + // nothing before the name's ``>``, the header at it, then only its arguments' pieces + if (n <= name_end) try std.testing.expectEqual(@as(usize, @intFromBool(n == name_end)), out.items.len); + try std.testing.expect(out.items.len >= last); // a longer prefix only ever adds + last = out.items.len; } const tail = try tool_parse.closeCall(a, text, tools); try std.testing.expectEqualStrings("", tail); try s.feed(try std.mem.concat(a, u8, &.{ text, tail }), &out); - try std.testing.expectEqual(@as(usize, 0), out.items.len); - try std.testing.expect(!s.streamed); + try std.testing.expectEqual(last, out.items.len); // the empty closeCall sends no delta of its own + // the invoke's header and its value stream; the nested XML is that value's text, never a call of its own + var headers: usize = 0; + for (out.items) |delta| for (delta.get("tool_calls").?.array) |call| { + if (call.get("id") == null) continue; + try std.testing.expectEqual(@as(i64, @intCast(headers)), call.get("index").?.int64().?); + try std.testing.expectEqualStrings("lookup", call.get("function").?.strField("name").?); + headers += 1; + }; + try std.testing.expectEqual(@as(usize, 1), headers); + try std.testing.expectEqualStrings(literal[1], try sentArguments(a, out.items)); + try std.testing.expect(s.streamed); }; } -test "malformed closed DSML masks its nested XML and resumes the next legitimate call" { +test "a malformed DSML block's header stands: its nested XML is argument text and the next call arrives" { const legacy = ""; inline for (.{ .{ dsml[0][0], "" }, .{ dsml[1][0], "" } }) |sp| { const malformed = sp[0] ++ legacy ++ sp[1]; @@ -217,16 +233,21 @@ test "malformed closed DSML masks its nested XML and resumes the next legitimate const tools = (try json.parse(a, "[{\"type\":\"function\",\"function\":{\"name\":\"lookup\"}}]")).ok.array; var s = try Streamer.init(a, tools); var out: std.ArrayList(Value) = .empty; - for (0..malformed.len + 1) |n| { + const name_end = std.mem.indexOf(u8, text, "\">").? + 2; // the byte the invoke's name ends on + var last: usize = 0; + for (0..text.len + 1) |n| { try s.feed(text[0..n], &out); - try std.testing.expectEqual(@as(usize, 0), out.items.len); + // nothing before the name's ``>``, the header at it, then only pieces + if (n <= name_end) try std.testing.expectEqual(@as(usize, @intFromBool(n == name_end)), out.items.len); + try std.testing.expect(out.items.len >= last); + last = out.items.len; } - for (malformed.len + 1..text.len + 1) |n| try s.feed(text[0..n], &out); var calls: usize = 0; for (out.items) |delta| for (delta.get("tool_calls").?.array) |call| { - if (call.get("id") != null) calls += 1; + calls += @intFromBool(call.get("id") != null); }; - try std.testing.expectEqual(@as(usize, 1), calls); - try std.testing.expectEqualStrings("{}", try sentArguments(a, out.items)); + // the call the block began, then the next one: the nested XML stays the block's argument text + try std.testing.expectEqual(@as(usize, 2), calls); + try std.testing.expectEqualStrings("{\"q\":\"\"{}", try sentArguments(a, out.items)); } }