//! OpenAI Chat Completions streaming provider. //! //! Wire format reference: https://platform.openai.com/docs/api-reference/chat/streaming //! //! Responsibilities: //! - Convert `Conversation` → request JSON (delegated to openai_chat_json.zig) //! - POST to `{base_url}/chat/completions` with `stream: true` //! - Read the chunked body, feed bytes through SSEParser //! - Parse each event payload, drive the block boundary state machine, //! and emit Receiver callbacks //! - Assemble the final Message and emit onMessageComplete const std = @import("std"); const Allocator = std.mem.Allocator; const Io = std.Io; const http = std.http; const Uri = std.Uri; const conversation = @import("conversation.zig"); const provider_mod = @import("provider.zig"); const stream_mod = @import("stream.zig"); const sse_mod = @import("sse.zig"); const json_mod = @import("openai_chat_json.zig"); const config_mod = @import("config.zig"); const tool_registry_mod = @import("tool_registry.zig"); const Event = stream_mod.Event; const EventQueue = stream_mod.EventQueue; /// Decode a wire tool name (`__` -> `.`) in place within an assembled /// name buffer. Decoding only ever shrinks the buffer (reads stay ahead /// of writes), so aliasing src/dst is safe; we then truncate to the /// decoded length. Unambiguous because internal names never contain a /// literal `__` (enforced at registration). fn decodeNameInPlace(name_buf: *conversation.TextualBlock) void { const decoded = tool_registry_mod.decodeName(name_buf.items, name_buf.items); name_buf.items.len = decoded.len; } /// Active streaming block type tracked by the state machine. Mirrors the /// `ContentBlock` union variants but adds `.none` for "no block open yet". const ActiveBlock = enum { none, text, thinking, tool_use }; /// A single OpenAI Chat streaming request. Transient: constructed per /// `streamStep`, holds only borrowed state (allocator, io, the global HTTP /// client, and the active config). Carries nothing across requests, so it /// is created inline by the free `streamStep` entry point below. pub const OpenAIChatRequest = struct { allocator: Allocator, io: Io, config: *const config_mod.OpenAIChatConfig, http_client: *http.Client, /// Optional diagnostic side-channel. When non-null, classified failures /// stash the HTTP status code and any `Retry-After` here for the agent's /// retry policy. Strings written here are not owned by the diagnostic; /// they live only as long as this request object. diag: ?*provider_mod.ProviderDiagnostic = null, /// Open the streaming HTTP request and return a heap-allocated /// resumable response. Performs the POST and reads response headers /// (classifying any >=400 status into a provider error), but does NOT /// pump the body — that happens lazily in `ResumableResponse.produce`. /// /// On success the caller owns the returned `*ResumableResponse` and must /// `deinit` it. On failure nothing is allocated. pub fn open( self: *OpenAIChatRequest, conv: *conversation.Conversation, tools: *const provider_mod.ToolRegistry, ) !*ResumableResponse { const rr = try self.allocator.create(ResumableResponse); errdefer self.allocator.destroy(rr); rr.* = .{ .allocator = self.allocator, .conv = conv, .parser = sse_mod.SSEParser.init(self.allocator), .state = .init(self.allocator), }; errdefer { rr.parser.deinit(); rr.state.deinit(); } // Build URL: "{base_url}/chat/completions" const url = try std.fmt.allocPrint( self.allocator, "{s}/chat/completions", .{self.config.base_url}, ); defer self.allocator.free(url); const uri = try Uri.parse(url); // Build the request body. const body = try json_mod.serializeRequest(self.allocator, self.config, conv, tools); defer self.allocator.free(body); // Auth header const auth_value = try std.fmt.allocPrint( self.allocator, "Bearer {s}", .{self.config.api_key}, ); defer self.allocator.free(auth_value); const extra_headers = [_]http.Header{ .{ .name = "content-type", .value = "application/json" }, .{ .name = "accept", .value = "text/event-stream" }, .{ .name = "authorization", .value = auth_value }, }; // Open the request. We can't use `fetch()` because it buffers the // response; we want to stream the body as it arrives. The request // is moved into the heap struct so the body reader (which borrows // `&rr.response`) stays valid across `produce` calls. rr.req = try self.http_client.request(.POST, uri, .{ .extra_headers = &extra_headers, // Disable compression: gzip buffers small SSE frames, defeating // the streaming property we paid for `stream: true` to get. .headers = .{ .accept_encoding = .{ .override = "identity" } }, .keep_alive = false, .redirect_behavior = .not_allowed, }); rr.req_open = true; errdefer { rr.req.deinit(); rr.req_open = false; } rr.req.transfer_encoding = .{ .content_length = body.len }; var send_buf: [4096]u8 = undefined; var bw = try rr.req.sendBodyUnflushed(&send_buf); try bw.writer.writeAll(body); try bw.end(); try rr.req.connection.?.flush(); // Receive response headers. var redirect_buf: [1024]u8 = undefined; rr.response = try rr.req.receiveHead(&redirect_buf); if (@intFromEnum(rr.response.head.status) >= 400) { // Drain body for diagnostics. const body_reader = rr.response.reader(&rr.transfer_buf); var err_buf: std.ArrayList(u8) = .empty; defer err_buf.deinit(self.allocator); var tmp: [1024]u8 = undefined; while (true) { const n = body_reader.readSliceShort(&tmp) catch break; if (n == 0) break; try err_buf.appendSlice(self.allocator, tmp[0..n]); if (err_buf.items.len > 16 * 1024) break; } const status: u16 = @intFromEnum(rr.response.head.status); std.log.err("openai_chat HTTP {d}: {s}", .{ status, err_buf.items }); // Classify the status into a retryable/terminal provider error. // HTTP 400 with a context marker becomes `ContextOverflow` so the // caller can compact and retry rather than hard-fail. const classified = provider_mod.classifyHttpStatus(status, err_buf.items); if (self.diag) |d| { d.status_code = status; d.retry_after_ms = provider_mod.retryAfterFromHead(rr.response.head); } return classified; } // Bind the streaming body reader. Valid for the lifetime of `rr` // (it borrows `&rr.response` and `&rr.transfer_buf`, both pinned). rr.body_reader = rr.response.reader(&rr.transfer_buf); return rr; } }; /// A resumable OpenAI Chat streaming response. Owns the pinned HTTP /// request/response, the body reader's transfer buffer, the `SSEParser`, /// and the block-assembly `StreamState`. `produce` pumps just enough bytes /// to emit one or more events into the queue, or reports the response is /// complete (its assistant message committed to the conversation). /// /// Must be heap-allocated and never moved: `body_reader` borrows /// `&self.response`. pub const ResumableResponse = struct { allocator: Allocator, conv: *conversation.Conversation, parser: sse_mod.SSEParser, state: StreamState, req: http.Client.Request = undefined, response: http.Client.Response = undefined, /// Transfer buffer backing `body_reader`. Pinned in the heap struct. transfer_buf: [4096]u8 = undefined, /// The streaming body reader, bound in `open` after a 2xx response. body_reader: *std.Io.Reader = undefined, /// Chunk scratch for `readVec`. chunk: [4096]u8 = undefined, /// True once `req` has been initialized (so `deinit` knows to free it). req_open: bool = false, /// Set once the response is fully decoded and finalized. done: bool = false, pub const ProduceStatus = provider_mod.ProviderStream.ProduceStatus; /// Wrap this response in the provider-agnostic `ProviderStream` the agent /// loop drives. pub fn providerStream(self: *ResumableResponse) provider_mod.ProviderStream { return .{ .ptr = self, .vtable = &vtable }; } const vtable: provider_mod.ProviderStream.VTable = .{ .produce = produceVT, .deinit = deinitVT, }; fn produceVT(ptr: *anyopaque, out: *EventQueue) anyerror!ProduceStatus { const self: *ResumableResponse = @ptrCast(@alignCast(ptr)); return self.produce(out); } fn deinitVT(ptr: *anyopaque) void { const self: *ResumableResponse = @ptrCast(@alignCast(ptr)); self.deinit(); } pub fn deinit(self: *ResumableResponse) void { if (self.req_open) self.req.deinit(); self.parser.deinit(); self.state.deinit(); self.allocator.destroy(self); } /// Pump the response: read one chunk, feed it through the SSE parser, /// and decode each SSE event into zero or more `Event`s appended to /// `out`. Returns `.more` if the caller should pump again, or /// `.response_complete` once the terminal (`[DONE]` or EOF) has been /// reached and the assistant message has been committed + a final /// `message_complete` pushed. /// /// Reading and finalizing here means a single `produce` call may push /// several events; the `Stream` drains the queue before pumping again. pub fn produce(self: *ResumableResponse, out: *EventQueue) !ProduceStatus { if (self.done) return .response_complete; var vecs: [1][]u8 = .{&self.chunk}; const n = self.body_reader.readVec(&vecs) catch |err| switch (err) { // Stream ended without [DONE]. Some servers and proxies omit it // (or drop the trailing usage chunk). Finalize with whatever // we've got — usage will be null in that case, which is fine. error.EndOfStream => { try self.finishStream(out); return .response_complete; }, // A transport read failure mid-stream (reset, TLS, timeout) // before `[DONE]` means no assistant message was committed. // Surface it as a retryable malformed-stream error. else => return error.ProviderStreamMalformed, }; if (n == 0) return .more; const events = try self.parser.feed(self.chunk[0..n]); defer self.parser.freeEvents(events); for (events) |ev_payload| { std.log.debug("openai_chat <= {s}", .{ev_payload}); if (std.mem.eql(u8, ev_payload, "[DONE]")) { try self.finishStream(out); return .response_complete; } try handleEvent(self.allocator, ev_payload, &self.state, out); // Note: we do NOT bail when state.end_of_stream is set. // OpenAI emits the terminating `usage` chunk *after* the // chunk carrying finish_reason, then sends `[DONE]`. If // we returned on finish_reason we'd never capture usage. // `[DONE]` is the authoritative end-of-stream marker. } return .more; } fn finishStream(self: *ResumableResponse, out: *EventQueue) !void { if (self.done) return; self.done = true; try self.state.finalize(out, self.conv); } }; /// State maintained across the streaming response: which block is currently /// being assembled, accumulated content, and the assistant message being /// built up for the final `onMessageComplete` callback. /// /// We model the assistant message as a sequence of blocks, exactly one of /// which is active at a time. Text/thinking transitions are inferred from /// which field a delta carries. Tool_use blocks arrive as a per-call wire /// `index`; the OpenAI Chat Completions streaming spec does not formally /// promise that all fragments for a given index arrive contiguously, but in /// practice every well-behaved backend (and the official Node SDK's own /// reassembly logic) treats a delta for a new index as the implicit close /// of the previous one. We do the same: seeing a delta for an index that /// differs from `current_tool_index` closes the prior tool_use and opens a /// new one. `finish_reason` closes the last still-open tool_use. A delta /// arriving for an index that has already been closed is a degenerate /// backend behavior (e.g. vLLM with speculative decoding under some /// configurations) — we log an error and drop the fragment. const StreamState = struct { allocator: Allocator, started: bool = false, /// Set when the wire stream signals end-of-turn (finish_reason or [DONE]). /// Tells the outer read loop to stop pulling more events. end_of_stream: bool = false, /// Set once `finalize` has run, to make it idempotent. finalized: bool = false, active: ActiveBlock = .none, /// Block index reported to the receiver. Increments per block boundary. block_index: usize = 0, /// Buffer for the currently-streaming text/thinking block. Owned by /// this state until the block is completed, at which point ownership /// transfers to the assembled Message. current_buf: conversation.TextualBlock = .empty, /// Assembled blocks for the final message, in stream order. blocks: std.ArrayList(conversation.ContentBlock) = .empty, /// The currently-streaming tool_use, if any. Closed when a delta for /// a different wire index arrives, or at finalize. active_tool: ?ToolUseInProgress = null, /// Wire index of `active_tool` (when non-null). current_tool_index: ?usize = null, /// Wire indices that have already been closed. Used solely to detect /// (and report) the degenerate case of a delta arriving for an index /// whose block we've already emitted. closed_tool_indices: std.AutoHashMap(usize, void), /// Token counts from the terminating chunk's `usage` block. Only /// populated when the server sent `usage` (i.e. the request used /// `stream_options.include_usage: true` AND the server honored it). usage: ?provider_mod.Usage = null, const ToolUseInProgress = struct { /// Block index emitted to the receiver for this tool call's /// onBlockStart / onContentDelta / onBlockComplete callbacks. block_index: usize, /// id/name are buffered as TextualBlocks because lenient providers /// (OpenRouter passthroughs, some self-hosted backends) may stream /// either field as fragments across multiple deltas. OpenAI itself /// sends them whole on the first delta, but the structural cost of /// supporting fragments is small and worth the robustness. id_buf: conversation.TextualBlock = .empty, name_buf: conversation.TextualBlock = .empty, arguments: conversation.TextualBlock = .empty, /// Set once we've emitted `onBlockStart(.ToolUse, ...)` for this /// block. We defer until either the first argument fragment /// arrives or the block is closed — not for identity reasons /// (identity is no longer passed at start) but to keep block /// indices clean: a tool_call that turns out to lack id or name /// is dropped silently rather than producing an empty /// start/complete pair. started: bool = false, /// Set once we've emitted `onToolDetails` for this block. Fired /// as soon as both id and name are non-empty, which may be on /// the first delta (the common case) or partway through arg /// deltas (fragmented-identity providers). details_emitted: bool = false, fn deinit(self: *ToolUseInProgress, allocator: Allocator) void { self.id_buf.deinit(allocator); self.name_buf.deinit(allocator); self.arguments.deinit(allocator); } }; fn init(allocator: Allocator) StreamState { return .{ .allocator = allocator, .closed_tool_indices = std.AutoHashMap(usize, void).init(allocator), }; } fn deinit(self: *StreamState) void { self.current_buf.deinit(self.allocator); for (self.blocks.items) |*b| b.deinit(self.allocator); self.blocks.deinit(self.allocator); if (self.active_tool) |*tu| tu.deinit(self.allocator); self.closed_tool_indices.deinit(); } /// Close the active text/thinking block (if any) and emit /// onBlockComplete. Ownership of `current_buf` transfers into the /// appended block. fn closeActive(self: *StreamState, out: *EventQueue) !void { if (self.active == .none) return; const block: conversation.ContentBlock = switch (self.active) { .text => .{ .Text = self.current_buf }, .thinking => .{ .Thinking = .{ .text = self.current_buf } }, .tool_use, .none => unreachable, }; self.current_buf = .empty; try self.blocks.append(self.allocator, block); try out.push(.{ .block_complete = .{ .index = self.block_index, .block = self.blocks.items[self.blocks.items.len - 1], } }); self.active = .none; } /// Open a new text/thinking block, possibly closing a prior one. fn openBlock( self: *StreamState, new_active: ActiveBlock, out: *EventQueue, ) !void { std.debug.assert(new_active == .text or new_active == .thinking); if (self.active == new_active) return; if (self.active != .none) { try self.closeActive(out); self.block_index += 1; } self.active = new_active; const block_type: provider_mod.ContentBlockType = switch (new_active) { .text => .Text, .thinking => .Thinking, .tool_use, .none => unreachable, }; try out.push(.{ .block_start = .{ .block_type = block_type, .index = self.block_index } }); } fn appendDelta( self: *StreamState, out: *EventQueue, delta: []const u8, ) !void { try self.current_buf.appendSlice(self.allocator, delta); // Dupe into the queue arena: the raw `delta` borrows the transient // SSE payload that `produce` frees before `next()` reads the queue. try out.push(.{ .content_delta = .{ .index = self.block_index, .delta = try out.dupeBytes(delta), } }); } /// Apply one streaming tool_call delta. Opens a new tool_use on the /// first sight of a wire index, closing any prior tool_use (or active /// text/thinking block) first. A delta for an already-closed index is /// a malformed stream — we log and drop it. fn applyToolCallDelta( self: *StreamState, out: *EventQueue, d: json_mod.ToolCallDelta, ) !void { // Degenerate backend: a delta arrived for an index whose block we // already finalized. Drop the fragment so we don't reopen a closed // block, but log loudly enough to make this diagnosable. if (self.closed_tool_indices.contains(d.index)) { if (!@import("builtin").is_test) { std.log.err( "openai_chat: dropping tool_call delta for already-closed wire index {d} (non-contiguous tool_call stream); id={?s} name={?s} args={?s}", .{ d.index, d.id, d.name, d.arguments }, ); } return; } // Wire-index change closes the previously-active tool_use. This is // the only signal openai_chat gives us for mid-stream tool_use // boundaries; see the StreamState doc-comment for the rationale. if (self.current_tool_index) |cur| { if (cur != d.index) try self.closeActiveTool(out); } if (self.active_tool == null) { // Opening a new tool_use. First close any open text/thinking // block so the tool_use gets its own block_index. if (self.active != .none) { try self.closeActive(out); self.block_index += 1; } self.active_tool = .{ .block_index = self.block_index }; self.current_tool_index = d.index; self.block_index += 1; } const tu = &self.active_tool.?; // Append identity fragments. Most providers send id+name whole on // the first delta and never repeat them, but appending is the only // correct behavior across the full range of OpenAI-compatible // backends — some chunk these strings. if (d.id) |s| try tu.id_buf.appendSlice(self.allocator, s); if (d.name) |s| try tu.name_buf.appendSlice(self.allocator, s); // Defer `onBlockStart` until args begin. The first argument // fragment is our signal that identity is likely settled enough // to render. If the block closes before any args arrive (zero-arg // tool), `closeActiveTool` emits the start there. if (d.arguments) |a| { try self.emitStartIfNeeded(out, tu); // Fire `tool_details` as soon as both id and name are // known. We can't know identity is *final* until the block // closes (a later delta could append more bytes), but in // practice OpenAI sends each whole on the first delta. A // pathological backend that streams id/name across many // chunks would have us emit a truncated value here. We // accept that trade-off: consumers that need the canonical // value can read it from the assembled ContentBlock at // block_complete. try self.emitDetailsIfReady(out, tu); try tu.arguments.appendSlice(self.allocator, a); // Dupe into the queue arena (the SSE payload is freed before // `next()` reads the queue). try out.push(.{ .content_delta = .{ .index = tu.block_index, .delta = try out.dupeBytes(a), } }); } else { // Identity-only chunk (no args yet). Still try to emit // details, in case both fields are now populated. if (tu.started) try self.emitDetailsIfReady(out, tu); } } /// Fire `tool_details` once both id and name are non-empty. No-op if /// already fired or if either field is still empty. Requires that /// `block_start` has already been emitted. Slices are duped into the /// queue arena because `id_buf`/`name_buf` may still grow (and realloc) /// on later fragments. fn emitDetailsIfReady( self: *StreamState, out: *EventQueue, tu: *ToolUseInProgress, ) !void { _ = self; if (tu.details_emitted) return; if (!tu.started) return; if (tu.id_buf.items.len == 0 or tu.name_buf.items.len == 0) return; tu.details_emitted = true; try out.push(.{ .tool_details = .{ .index = tu.block_index, .id = try out.dupeBytes(tu.id_buf.items), .name = try out.dupeBytes(tu.name_buf.items), } }); } /// Close the currently-active tool_use (if any), emitting block_start /// (if it wasn't already), block_complete, and recording the wire /// index as closed. No-op if there's no active tool_use. fn closeActiveTool(self: *StreamState, out: *EventQueue) !void { var tu = self.active_tool orelse return; self.active_tool = null; const wire_index = self.current_tool_index.?; self.current_tool_index = null; try self.closed_tool_indices.put(wire_index, {}); // Drop entries lacking id or name. The stream closed the block // before the provider sent enough to identify which tool was // being called — there's nothing we can dispatch. if (tu.id_buf.items.len == 0 or tu.name_buf.items.len == 0) { if (!@import("builtin").is_test) { std.log.err( "openai_chat: dropping incomplete tool_use at wire index {d}: id={d} bytes, name=\"{s}\", args={d} bytes", .{ wire_index, tu.id_buf.items.len, tu.name_buf.items, tu.arguments.items.len, }, ); } tu.deinit(self.allocator); return; } // The model echoes the wire-encoded tool name (`__` for `.`). // Decode in place now that the full name is assembled, so the // conversation, receiver callbacks, and dispatch all see the // internal (dotted) name. Decoding never grows the buffer. decodeNameInPlace(&tu.name_buf); // If no arguments ever arrived, we haven't emitted block_start // yet — do it now so the consumer sees a balanced start/complete. try self.emitStartIfNeeded(out, &tu); // Last chance to fire details if a fragmented-identity provider // only finished id/name accumulation at the very end. try self.emitDetailsIfReady(out, &tu); const id_owned = try tu.id_buf.toOwnedSlice(self.allocator); const name_owned = try tu.name_buf.toOwnedSlice(self.allocator); const block: conversation.ContentBlock = .{ .ToolUse = .{ .id = id_owned, .name = name_owned, .input = tu.arguments, } }; // Ownership has moved into `block`; clear the local before it // goes out of scope so deinit doesn't double-free. tu.arguments = .empty; try self.blocks.append(self.allocator, block); try out.push(.{ .block_complete = .{ .index = tu.block_index, .block = self.blocks.items[self.blocks.items.len - 1], } }); } /// Emit `block_start(.ToolUse, ...)` once per in-progress tool use. /// Callers must invoke this before the first `content_delta` or /// `block_complete` for the block. Identity (id/name) is *not* passed at /// start — consumers get identity from `tool_details` or the assembled /// ContentBlock at block_complete time. fn emitStartIfNeeded( self: *StreamState, out: *EventQueue, tu: *ToolUseInProgress, ) !void { _ = self; if (tu.started) return; tu.started = true; try out.push(.{ .block_start = .{ .block_type = .ToolUse, .index = tu.block_index } }); } /// End the stream: close any open text/thinking block, close the still- /// active tool_use (if any), then commit the assembled assistant /// Message to the conversation and push the terminal `message_complete`. fn finalize( self: *StreamState, out: *EventQueue, conv: *conversation.Conversation, ) !void { if (self.finalized) return; self.finalized = true; try self.closeActive(out); try self.closeActiveTool(out); // Move blocks into a fresh conversation message. const moved_blocks = try self.blocks.toOwnedSlice(self.allocator); defer self.allocator.free(moved_blocks); try conv.addAssistantMessage(moved_blocks, self.usage); const msg = conv.messages.items[conv.messages.items.len - 1]; try out.push(.{ .message_complete = .{ .message = msg, .usage = self.usage } }); } }; fn handleEvent( allocator: Allocator, payload: []const u8, state: *StreamState, out: *EventQueue, ) !void { var parsed = try json_mod.parseStreamEvent(allocator, payload); defer parsed.deinit(); const d = parsed.delta; // Usage block arrives in the terminating chunk (after finish_reason, // with an empty `choices` array). Capture it; `finalize` delivers it // as part of `onMessageComplete`. OpenAI bills `prompt_tokens` as // the *total* input including cached tokens; we split them so // callers don't have to. if (d.usage) |u| { const prompt: u64 = u.prompt_tokens orelse 0; const cached: u64 = u.cached_prompt_tokens orelse 0; const fresh: u64 = if (cached > prompt) 0 else prompt - cached; state.usage = .{ .input = fresh, .output = u.completion_tokens orelse 0, .cache_read = cached, .cache_write = 0, // OpenAI doesn't bill a cache-write premium. .reasoning = u.reasoning_tokens orelse 0, }; } // Mid-stream provider error: some OpenAI-compatible endpoints (and // OpenAI itself on rare transient failures) return HTTP 200 with an // error embedded in the SSE stream. Treat the turn as failed. if (d.error_message != null or d.error_type != null) { if (!@import("builtin").is_test) { std.log.err("openai_chat stream error: {?s}: {?s}", .{ d.error_type, d.error_message, }); } return error.ProviderStreamMalformed; } if (!state.started and d.role != null) { state.started = true; try out.push(.{ .message_start = .assistant }); } if (d.reasoning_content) |rc| { if (!state.started) { state.started = true; try out.push(.{ .message_start = .assistant }); } try state.openBlock(.thinking, out); try state.appendDelta(out, rc); } if (d.content) |c| { if (!state.started) { state.started = true; try out.push(.{ .message_start = .assistant }); } try state.openBlock(.text, out); try state.appendDelta(out, c); } if (d.tool_calls.len > 0) { if (!state.started) { state.started = true; try out.push(.{ .message_start = .assistant }); } for (d.tool_calls) |tc| try state.applyToolCallDelta(out, tc); } if (d.finish_reason) |_| { state.end_of_stream = true; } } // ----------------------------------------------------------------------------- // Tests // ----------------------------------------------------------------------------- const testing = std.testing; /// Feed a sequence of SSE event payloads through the state machine as if /// they had been delivered by the wire, finalizing into `conv`. The decoded /// `Event`s are recorded as compact strings (the same schema the old /// RecordingReceiver used) so callback-ordering assertions are preserved. fn runStreamedTurn( allocator: Allocator, conv: *conversation.Conversation, rec: ?*EventRecorder, events: []const []const u8, ) !void { var state: StreamState = .init(allocator); defer state.deinit(); var queue = EventQueue.init(allocator); defer queue.deinit(); for (events) |payload| { if (std.mem.eql(u8, payload, "[DONE]")) break; // Process every chunk through to [DONE], including the // post-finish_reason usage chunk. Mirrors the production pump in // ResumableResponse.produce. try handleEvent(allocator, payload, &state, &queue); } try state.finalize(&queue, conv); // Drain into the recorder before the arena resets. The queue holds all // events from this turn; popping records each, and the final null-pop // resets the arena. while (queue.pop()) |ev| { if (rec) |r| try r.record(ev); } } /// Records decoded `Event`s as compact strings for ordering assertions. const EventRecorder = struct { allocator: Allocator, events: std.ArrayList([]const u8) = .empty, fn deinit(self: *EventRecorder) void { for (self.events.items) |e| self.allocator.free(e); self.events.deinit(self.allocator); } fn push(self: *EventRecorder, comptime fmt: []const u8, args: anytype) !void { const owned = try std.fmt.allocPrint(self.allocator, fmt, args); try self.events.append(self.allocator, owned); } fn record(self: *EventRecorder, ev: Event) !void { switch (ev) { .message_start => try self.push("msg_start", .{}), .block_start => |b| try self.push("block_start[{d}]:{s}", .{ b.index, @tagName(b.block_type) }), .tool_details => |t| try self.push("tool_details[{d}]:{s}:{s}", .{ t.index, t.id, t.name }), .content_delta => |d| try self.push("delta[{d}]:{s}", .{ d.index, d.delta }), .block_complete => |b| try self.push("block_complete[{d}]", .{b.index}), .message_complete => |m| { if (m.usage) |u| { try self.push( "msg_complete[usage:in={d},out={d},cr={d},cw={d},rsn={d}]", .{ u.input, u.output, u.cache_read, u.cache_write, u.reasoning }, ); } else { try self.push("msg_complete[usage:null]", .{}); } }, else => {}, } } }; /// Test helper: append a single-text user message. `addUserMessage` now /// takes a block slice (symmetric with `addAssistantMessage`); this wraps /// the common plain-text case the tests below use. fn addUserText(conv: *conversation.Conversation, text: []const u8) !void { const tb = try conversation.textualBlockFromSlice(conv.allocator, text); var block: conversation.ContentBlock = .{ .Text = tb }; errdefer block.deinit(conv.allocator); try conv.addUserMessage(&.{block}); } test "two streamed turns persist assistant replies in the conversation" { // Regression test for the bug where `finish_reason` arrived before // `[DONE]` and `finalize` early-returned without appending the assistant // message, so follow-up turns were sent without prior responses. const allocator = testing.allocator; var conv = conversation.Conversation.init(allocator); defer conv.deinit(); try conv.addSystemMessage("You are a helpful assistant."); try addUserText(&conv, "hello!"); const turn1 = [_][]const u8{ \\{"choices":[{"delta":{"role":"assistant"}}]} , \\{"choices":[{"delta":{"content":"Hello! "}}]} , \\{"choices":[{"delta":{"content":"How can I help you today?"}}]} , \\{"choices":[{"delta":{},"finish_reason":"stop"}]} , "[DONE]", }; try runStreamedTurn(allocator, &conv, null, &turn1); try testing.expectEqual(@as(usize, 3), conv.messages.items.len); try testing.expectEqual(conversation.MessageRole.assistant, conv.messages.items[2].role); try testing.expectEqualStrings( "Hello! How can I help you today?", conv.messages.items[2].content.items[0].Text.items, ); // Second user turn: the assistant must still see its prior response. try addUserText(&conv, "how did you respond to my greeting just now?"); const turn2 = [_][]const u8{ \\{"choices":[{"delta":{"role":"assistant"}}]} , \\{"choices":[{"delta":{"content":"I replied: \"Hello! How can I help you today?\""}}]} , \\{"choices":[{"delta":{},"finish_reason":"stop"}]} , "[DONE]", }; try runStreamedTurn(allocator, &conv, null, &turn2); // System + user + assistant + user + assistant = 5 messages. try testing.expectEqual(@as(usize, 5), conv.messages.items.len); try testing.expectEqual(conversation.MessageRole.assistant, conv.messages.items[4].role); try testing.expectEqualStrings( "I replied: \"Hello! How can I help you today?\"", conv.messages.items[4].content.items[0].Text.items, ); } test "openai_chat: terminating usage chunk lands on message_complete with split cache_read" { const allocator = testing.allocator; var conv = conversation.Conversation.init(allocator); defer conv.deinit(); try addUserText(&conv, "hi"); var rec = EventRecorder{ .allocator = allocator }; defer rec.deinit(); const events = [_][]const u8{ \\{"choices":[{"delta":{"role":"assistant"}}]} , \\{"choices":[{"delta":{"content":"hi"}}]} , \\{"choices":[{"delta":{},"finish_reason":"stop"}]} , // OpenAI's terminating chunk: empty choices, top-level usage. \\{"choices":[],"usage":{"prompt_tokens":150,"completion_tokens":42,"prompt_tokens_details":{"cached_tokens":120},"completion_tokens_details":{"reasoning_tokens":18}}} , "[DONE]", }; try runStreamedTurn(allocator, &conv, &rec, &events); var found: ?[]const u8 = null; for (rec.events.items) |s| { if (std.mem.startsWith(u8, s, "msg_complete[")) found = s; } try testing.expect(found != null); // 150 prompt - 120 cached = 30 fresh input. try testing.expectEqualStrings("msg_complete[usage:in=30,out=42,cr=120,cw=0,rsn=18]", found.?); } test "openai_chat: omitted stream usage yields null on message_complete" { const allocator = testing.allocator; var conv = conversation.Conversation.init(allocator); defer conv.deinit(); try addUserText(&conv, "hi"); var rec = EventRecorder{ .allocator = allocator }; defer rec.deinit(); const events = [_][]const u8{ \\{"choices":[{"delta":{"role":"assistant"}}]} , \\{"choices":[{"delta":{"content":"hi"}}]} , \\{"choices":[{"delta":{},"finish_reason":"stop"}]} , "[DONE]", }; try runStreamedTurn(allocator, &conv, &rec, &events); var found: ?[]const u8 = null; for (rec.events.items) |s| { if (std.mem.startsWith(u8, s, "msg_complete")) found = s; } try testing.expect(found != null); try testing.expectEqualStrings("msg_complete[usage:null]", found.?); } test "fragmented tool_call id and name are reassembled" { // Lenient OpenAI-compatible providers occasionally split `id` and // `function.name` across multiple deltas instead of sending them whole // on the first chunk. Verify the state machine appends both correctly // and emits a complete identity to the receiver. const allocator = testing.allocator; var conv = conversation.Conversation.init(allocator); defer conv.deinit(); try addUserText(&conv, "call something"); const events = [_][]const u8{ \\{"choices":[{"delta":{"role":"assistant"}}]} , \\{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_","type":"function","function":{"name":"pi"}}]}}]} , \\{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"xyz","function":{"name":"ng"}}]}}]} , \\{"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\"host\":\"a.com\"}"}}]}}]} , \\{"choices":[{"delta":{},"finish_reason":"tool_calls"}]} , "[DONE]", }; try runStreamedTurn(allocator, &conv, null, &events); const asst = conv.messages.items[1]; try testing.expectEqual(@as(usize, 1), asst.content.items.len); const tu = asst.content.items[0].ToolUse; try testing.expectEqualStrings("call_xyz", tu.id); try testing.expectEqualStrings("ping", tu.name); try testing.expectEqualStrings("{\"host\":\"a.com\"}", tu.input.items); } test "inbound wire tool name is decoded to dotted form (even split across __)" { // The model echoes the wire name it was given (`std__read`). It is // decoded to the internal `std.read` for the conversation/session/ // dispatch. The decode happens after full assembly, so a `__` split // across two deltas (`std_` + `_read`) decodes correctly. const allocator = testing.allocator; var conv = conversation.Conversation.init(allocator); defer conv.deinit(); try addUserText(&conv, "read a file"); const events = [_][]const u8{ \\{"choices":[{"delta":{"role":"assistant"}}]} , \\{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"c1","type":"function","function":{"name":"std_"}}]}}]} , \\{"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"name":"_read"}}]}}]} , \\{"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{}"}}]}}]} , \\{"choices":[{"delta":{},"finish_reason":"tool_calls"}]} , "[DONE]", }; try runStreamedTurn(allocator, &conv, null, &events); const tu = conv.messages.items[1].content.items[0].ToolUse; try testing.expectEqualStrings("std.read", tu.name); } test "parallel tool_calls emit one complete start/delta/complete cycle per block" { // Regression test: previously, the OpenAI provider deferred ALL // tool_use onBlockComplete callbacks to finalize, so a four-tool // parallel batch produced start/start/start/start/delta*/complete/ // complete/complete/complete — the receiver couldn't render each tool // as its own discrete block. With the new contiguity-driven close-on- // next-index logic, each tool_use should produce a contiguous // start → delta(s) → complete trio. const allocator = testing.allocator; var conv = conversation.Conversation.init(allocator); defer conv.deinit(); try addUserText(&conv, "ping four hosts"); var rec: EventRecorder = .{ .allocator = allocator }; defer rec.deinit(); const events = [_][]const u8{ \\{"choices":[{"delta":{"role":"assistant"}}]} , \\{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"c0","type":"function","function":{"name":"ping","arguments":"{\"host\":\"a\"}"}}]}}]} , \\{"choices":[{"delta":{"tool_calls":[{"index":1,"id":"c1","type":"function","function":{"name":"ping","arguments":"{\"host\":\"b\"}"}}]}}]} , \\{"choices":[{"delta":{"tool_calls":[{"index":2,"id":"c2","type":"function","function":{"name":"ping","arguments":"{\"host\":\"c\"}"}}]}}]} , \\{"choices":[{"delta":{"tool_calls":[{"index":3,"id":"c3","type":"function","function":{"name":"ping","arguments":"{\"host\":\"d\"}"}}]}}]} , \\{"choices":[{"delta":{},"finish_reason":"tool_calls"}]} , "[DONE]", }; try runStreamedTurn(allocator, &conv, &rec, &events); const expected = [_][]const u8{ "msg_start", "block_start[0]:ToolUse", "tool_details[0]:c0:ping", "delta[0]:{\"host\":\"a\"}", "block_complete[0]", "block_start[1]:ToolUse", "tool_details[1]:c1:ping", "delta[1]:{\"host\":\"b\"}", "block_complete[1]", "block_start[2]:ToolUse", "tool_details[2]:c2:ping", "delta[2]:{\"host\":\"c\"}", "block_complete[2]", "block_start[3]:ToolUse", "tool_details[3]:c3:ping", "delta[3]:{\"host\":\"d\"}", "block_complete[3]", // No usage chunk in this fixture (older test data) — record // shows null. "msg_complete[usage:null]", }; // Identity arrives in the assembled ContentBlock at completion time. const asst = conv.messages.items[1]; try testing.expectEqual(@as(usize, 4), asst.content.items.len); for (asst.content.items) |b| { try testing.expectEqualStrings("ping", b.ToolUse.name); } try testing.expectEqual(expected.len, rec.events.items.len); for (expected, rec.events.items) |want, got| { try testing.expectEqualStrings(want, got); } } test "non-contiguous tool_call deltas: re-emission of a closed index is dropped" { // Degenerate backend behavior: a delta for an already-closed wire // index. We must not reopen the block; instead drop the fragment and // log. The successfully-closed prior blocks remain intact. const allocator = testing.allocator; var conv = conversation.Conversation.init(allocator); defer conv.deinit(); try addUserText(&conv, "go"); var rec: EventRecorder = .{ .allocator = allocator }; defer rec.deinit(); const events = [_][]const u8{ \\{"choices":[{"delta":{"role":"assistant"}}]} , \\{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"c0","type":"function","function":{"name":"ping","arguments":"{\"x\":1}"}}]}}]} , \\{"choices":[{"delta":{"tool_calls":[{"index":1,"id":"c1","type":"function","function":{"name":"ping","arguments":"{\"y\":2}"}}]}}]} , // Delta for already-closed index 0: must be dropped. \\{"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":",extra"}}]}}]} , \\{"choices":[{"delta":{},"finish_reason":"tool_calls"}]} , "[DONE]", }; try runStreamedTurn(allocator, &conv, &rec, &events); // Two well-formed tool_use blocks in the final message, args unaffected // by the dropped fragment. const asst = conv.messages.items[1]; try testing.expectEqual(@as(usize, 2), asst.content.items.len); try testing.expectEqualStrings("{\"x\":1}", asst.content.items[0].ToolUse.input.items); try testing.expectEqualStrings("{\"y\":2}", asst.content.items[1].ToolUse.input.items); // Callback sequence: index 0 closed cleanly before any stray delta. // There must be exactly one block_complete[0] in the event log // (i.e. the stray delta did not produce a second open/close cycle). var n_complete_0: usize = 0; for (rec.events.items) |e| { if (std.mem.eql(u8, e, "block_complete[0]")) n_complete_0 += 1; } try testing.expectEqual(@as(usize, 1), n_complete_0); } test "onToolDetails fires after id+name complete, even mid-arg-stream" { // Fragmented-identity provider: id arrives split across two chunks, // and an arg fragment appears between them. `onToolDetails` must // wait until both id and name are non-empty (i.e. on the chunk that // completes id), and must fire exactly once, before block_complete. const allocator = testing.allocator; var conv = conversation.Conversation.init(allocator); defer conv.deinit(); try addUserText(&conv, "go"); var rec: EventRecorder = .{ .allocator = allocator }; defer rec.deinit(); const events = [_][]const u8{ \\{"choices":[{"delta":{"role":"assistant"}}]} , // Identity-only chunk: name arrives whole, id starts. No args yet, // so onBlockStart hasn't fired and details can't either. \\{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_","type":"function","function":{"name":"ping"}}]}}]} , // First arg chunk: onBlockStart fires. id is still "call_" — not // empty — and name is non-empty, so onToolDetails fires here // with whatever id we have so far (`call_`). \\{"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\"x\":"}}]}}]} , // Mid-stream id completion + second arg chunk. onToolDetails has // already fired so it does NOT fire again, even though id grew. // The final ContentBlock will carry the full "call_xyz" id. \\{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"xyz","function":{"arguments":"1}"}}]}}]} , \\{"choices":[{"delta":{},"finish_reason":"tool_calls"}]} , "[DONE]", }; try runStreamedTurn(allocator, &conv, &rec, &events); // Exactly one tool_details event, fired with the id-prefix that was // current at first-args-arrival, and ordered between block_start and // block_complete. var n_details: usize = 0; var details_pos: ?usize = null; var block_start_pos: ?usize = null; var block_complete_pos: ?usize = null; for (rec.events.items, 0..) |e, i| { if (std.mem.startsWith(u8, e, "tool_details[")) { n_details += 1; details_pos = i; try testing.expectEqualStrings("tool_details[0]:call_:ping", e); } else if (std.mem.eql(u8, e, "block_start[0]:ToolUse")) { block_start_pos = i; } else if (std.mem.eql(u8, e, "block_complete[0]")) { block_complete_pos = i; } } try testing.expectEqual(@as(usize, 1), n_details); try testing.expect(block_start_pos.? < details_pos.?); try testing.expect(details_pos.? < block_complete_pos.?); // Final ContentBlock has the full id assembled from both fragments. const asst = conv.messages.items[1]; try testing.expectEqualStrings("call_xyz", asst.content.items[0].ToolUse.id); try testing.expectEqualStrings("ping", asst.content.items[0].ToolUse.name); try testing.expectEqualStrings("{\"x\":1}", asst.content.items[0].ToolUse.input.items); } test "tool_call with no arguments still finalizes a well-formed ToolUse" { // Some providers may emit a tool call with no arguments at all (e.g. a // zero-arg tool). The state machine should still emit onBlockStart // exactly once at finalize time and produce a ToolUse with empty input. const allocator = testing.allocator; var conv = conversation.Conversation.init(allocator); defer conv.deinit(); try addUserText(&conv, "ring it"); const events = [_][]const u8{ \\{"choices":[{"delta":{"role":"assistant"}}]} , \\{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"c1","type":"function","function":{"name":"ring"}}]}}]} , \\{"choices":[{"delta":{},"finish_reason":"tool_calls"}]} , "[DONE]", }; try runStreamedTurn(allocator, &conv, null, &events); const asst = conv.messages.items[1]; try testing.expectEqual(@as(usize, 1), asst.content.items.len); const tu = asst.content.items[0].ToolUse; try testing.expectEqualStrings("c1", tu.id); try testing.expectEqualStrings("ring", tu.name); try testing.expectEqual(@as(usize, 0), tu.input.items.len); }