diff options
Diffstat (limited to 'libpanto/src/provider_anthropic_messages.zig')
| -rw-r--r-- | libpanto/src/provider_anthropic_messages.zig | 448 |
1 files changed, 247 insertions, 201 deletions
diff --git a/libpanto/src/provider_anthropic_messages.zig b/libpanto/src/provider_anthropic_messages.zig index 55efb5f..4bd146a 100644 --- a/libpanto/src/provider_anthropic_messages.zig +++ b/libpanto/src/provider_anthropic_messages.zig @@ -20,11 +20,15 @@ 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("anthropic_messages_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; + /// A single Anthropic Messages streaming request. Transient: constructed /// per `streamStep`, holds only borrowed state (allocator, io, the global /// HTTP client, and the active config). Carries nothing across requests. @@ -36,26 +40,29 @@ pub const AnthropicMessagesRequest = struct { /// Optional diagnostic side-channel; see `OpenAIChatRequest.diag`. diag: ?*provider_mod.ProviderDiagnostic = null, - pub fn streamStep( + /// Open the streaming HTTP request and return a heap-allocated resumable + /// response. Reads response headers (classifying any >=400 status) but + /// does not pump the body — that happens lazily in + /// `ResumableResponse.produce`. On success the caller owns the returned + /// `*ResumableResponse` and must `deinit` it. + pub fn open( self: *AnthropicMessagesRequest, conv: *conversation.Conversation, tools: *const provider_mod.ToolRegistry, - receiver: *provider_mod.Receiver, - ) !void { - // Outer wrapper guarantees `onError` is called exactly once if - // anything fails. - self.streamStepInner(conv, tools, receiver) catch |err| { - receiver.onError(err); - return err; + ) !*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(); + } - fn streamStepInner( - self: *AnthropicMessagesRequest, - conv: *conversation.Conversation, - tools: *const provider_mod.ToolRegistry, - receiver: *provider_mod.Receiver, - ) !void { const url = try std.fmt.allocPrint( self.allocator, "{s}/v1/messages", @@ -75,7 +82,7 @@ pub const AnthropicMessagesRequest = struct { .{ .name = "anthropic-version", .value = self.config.api_version }, }; - var req = try self.http_client.request(.POST, uri, .{ + 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. @@ -83,22 +90,25 @@ pub const AnthropicMessagesRequest = struct { .keep_alive = false, .redirect_behavior = .not_allowed, }); - defer req.deinit(); + rr.req_open = true; + errdefer { + rr.req.deinit(); + rr.req_open = false; + } - req.transfer_encoding = .{ .content_length = body.len }; + rr.req.transfer_encoding = .{ .content_length = body.len }; var send_buf: [4096]u8 = undefined; - var bw = try req.sendBodyUnflushed(&send_buf); + var bw = try rr.req.sendBodyUnflushed(&send_buf); try bw.writer.writeAll(body); try bw.end(); - try req.connection.?.flush(); + try rr.req.connection.?.flush(); var redirect_buf: [1024]u8 = undefined; - var response = try req.receiveHead(&redirect_buf); + rr.response = try rr.req.receiveHead(&redirect_buf); - if (@intFromEnum(response.head.status) >= 400) { - var transfer_buf: [4096]u8 = undefined; - const body_reader = response.reader(&transfer_buf); + if (@intFromEnum(rr.response.head.status) >= 400) { + 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; @@ -108,7 +118,7 @@ pub const AnthropicMessagesRequest = struct { try err_buf.appendSlice(self.allocator, tmp[0..n]); if (err_buf.items.len > 16 * 1024) break; } - const status: u16 = @intFromEnum(response.head.status); + const status: u16 = @intFromEnum(rr.response.head.status); std.log.err("anthropic_messages HTTP {d}: {s}", .{ status, err_buf.items }); // Anthropic rejects oversized requests with HTTP 400 and a // "prompt is too long" message; `classifyHttpStatus` maps that to @@ -117,52 +127,104 @@ pub const AnthropicMessagesRequest = struct { const classified = provider_mod.classifyHttpStatus(status, err_buf.items); if (self.diag) |d| { d.status_code = status; - d.retry_after_ms = provider_mod.retryAfterFromHead(response.head); + d.retry_after_ms = provider_mod.retryAfterFromHead(rr.response.head); } return classified; } - var transfer_buf: [4096]u8 = undefined; - const body_reader = response.reader(&transfer_buf); + rr.body_reader = rr.response.reader(&rr.transfer_buf); + return rr; + } +}; + +/// A resumable Anthropic Messages streaming response. Owns the pinned HTTP +/// request/response, the body reader's transfer buffer, the `SSEParser`, and +/// the block-assembly `StreamState`. 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, - var parser = sse_mod.SSEParser.init(self.allocator); - defer parser.deinit(); + req: http.Client.Request = undefined, + response: http.Client.Response = undefined, + transfer_buf: [4096]u8 = undefined, + body_reader: *std.Io.Reader = undefined, + chunk: [4096]u8 = undefined, - var state: StreamState = .init(self.allocator); - defer state.deinit(); + req_open: bool = false, + done: bool = false, - // Use `readVec` so we return to the event loop as soon as *any* - // bytes arrive, rather than waiting for the buffer to fill. - // `readSliceShort` blocks until EOF or full, which defeats streaming. - // - // Per `std.Io.Reader.readVec` docs: `n == 0` does NOT mean EOF; - // EOF is signalled only via `error.EndOfStream`. Breaking on - // `n == 0` truncates the response mid-stream. - var chunk: [4096]u8 = undefined; - var vecs: [1][]u8 = .{&chunk}; - while (true) { - const n = body_reader.readVec(&vecs) catch |err| switch (err) { - error.EndOfStream => break, - // Transport failure before the message completed: retryable. - else => return error.ProviderStreamMalformed, - }; - if (n == 0) continue; + pub const ProduceStatus = provider_mod.ProviderStream.ProduceStatus; - const events = try parser.feed(chunk[0..n]); - defer parser.freeEvents(events); + /// 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 }; + } - for (events) |ev_payload| { - std.log.debug("anthropic_messages <= {s}", .{ev_payload}); - try handleEvent(self.allocator, ev_payload, &state, receiver); - if (state.end_of_stream) { - try state.finalize(receiver, conv); - return; - } + 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 `message_stop` (or EOF) is reached and the + /// assistant message has been committed + a final `message_complete` + /// pushed. + 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 an explicit message_stop. Finalize anyway. + error.EndOfStream => { + try self.finishStream(out); + return .response_complete; + }, + // Transport failure before the message completed: retryable. + 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("anthropic_messages <= {s}", .{ev_payload}); + try handleEvent(self.allocator, ev_payload, &self.state, out); + if (self.state.end_of_stream) { + try self.finishStream(out); + return .response_complete; } } + return .more; + } - // Stream ended without an explicit message_stop. Finalize anyway. - try state.finalize(receiver, conv); + fn finishStream(self: *ResumableResponse, out: *EventQueue) !void { + if (self.done) return; + self.done = true; + try self.state.finalize(out, self.conv); } }; @@ -224,10 +286,10 @@ const StreamState = struct { if (self.stop_reason) |s| self.allocator.free(s); } - fn ensureStarted(self: *StreamState, receiver: *provider_mod.Receiver) !void { + fn ensureStarted(self: *StreamState, out: *EventQueue) !void { if (self.started) return; self.started = true; - try receiver.onMessageStart(.assistant); + try out.push(.{ .message_start = .assistant }); } /// Merge a wire-level usage snapshot into the accumulated counts. @@ -248,7 +310,7 @@ const StreamState = struct { fn openBlock( self: *StreamState, - receiver: *provider_mod.Receiver, + out: *EventQueue, wire_index: usize, kind: BlockKind, tool_id: ?[]const u8, @@ -291,14 +353,20 @@ const StreamState = struct { .unsupported => null, }; if (block_type) |bt| { - try receiver.onBlockStart(bt, wire_index); + try out.push(.{ .block_start = .{ .block_type = bt, .index = wire_index } }); // Anthropic delivers tool id+name whole on content_block_start, - // so we can fire onToolDetails immediately — before any arg + // so we can fire tool_details immediately — before any arg // deltas. If the wire was malformed and either field is // missing, skip: closeBlock will drop the block defensively. + // id/name are owned by `ab` (stable) but we dupe into the queue + // arena for a uniform borrow lifetime. if (kind == .tool_use) { if (ab.tool_id != null and ab.tool_name != null) { - try receiver.onToolDetails(wire_index, ab.tool_id.?, ab.tool_name.?); + try out.push(.{ .tool_details = .{ + .index = wire_index, + .id = try out.dupeBytes(ab.tool_id.?), + .name = try out.dupeBytes(ab.tool_name.?), + } }); } } } @@ -306,26 +374,34 @@ const StreamState = struct { fn appendTextDelta( self: *StreamState, - receiver: *provider_mod.Receiver, + out: *EventQueue, delta: []const u8, ) !void { const a = &(self.active orelse return); if (a.kind == .unsupported) return; try a.text_buf.appendSlice(self.allocator, delta); - try receiver.onContentDelta(a.wire_index, delta); + // Dupe into the queue arena: `delta` borrows the transient SSE/JSON + // payload that `produce` frees before `next()` reads the queue. + try out.push(.{ .content_delta = .{ + .index = a.wire_index, + .delta = try out.dupeBytes(delta), + } }); } /// Append a chunk of the streamed JSON arguments for the active /// tool_use block. No-op if the active block isn't a tool_use. fn appendInputJsonDelta( self: *StreamState, - receiver: *provider_mod.Receiver, + out: *EventQueue, delta: []const u8, ) !void { const a = &(self.active orelse return); if (a.kind != .tool_use) return; try a.text_buf.appendSlice(self.allocator, delta); - try receiver.onContentDelta(a.wire_index, delta); + try out.push(.{ .content_delta = .{ + .index = a.wire_index, + .delta = try out.dupeBytes(delta), + } }); } fn setSignature(self: *StreamState, sig: []const u8) !void { @@ -339,10 +415,10 @@ const StreamState = struct { self.stop_reason = if (reason) |r| try self.allocator.dupe(u8, r) else null; } - /// Close the active block: append it to `blocks` and emit onBlockComplete. + /// Close the active block: append it to `blocks` and emit block_complete. fn closeBlock( self: *StreamState, - receiver: *provider_mod.Receiver, + out: *EventQueue, ) !void { var a = self.active orelse return; self.active = null; @@ -389,7 +465,10 @@ const StreamState = struct { }; try self.blocks.append(self.allocator, block); - try receiver.onBlockComplete(a.wire_index, self.blocks.items[self.blocks.items.len - 1]); + try out.push(.{ .block_complete = .{ + .index = a.wire_index, + .block = self.blocks.items[self.blocks.items.len - 1], + } }); } /// Drop the active block without emitting a completion callback. @@ -407,7 +486,7 @@ const StreamState = struct { fn finalize( self: *StreamState, - receiver: *provider_mod.Receiver, + out: *EventQueue, conv: *conversation.Conversation, ) !void { if (self.finalized) return; @@ -416,7 +495,7 @@ const StreamState = struct { if (self.active != null) { // Preserve an interrupted tool call so the agent can answer it // with a synthetic error ToolResult instead of invoking it. - try self.closeBlock(receiver); + try self.closeBlock(out); } const moved_blocks = try self.blocks.toOwnedSlice(self.allocator); @@ -426,7 +505,7 @@ const StreamState = struct { try conv.addAssistantMessageWithUsage(moved_blocks, usage); const msg = conv.messages.items[conv.messages.items.len - 1]; - try receiver.onMessageComplete(msg, usage); + try out.push(.{ .message_complete = .{ .message = msg, .usage = usage } }); } }; @@ -434,35 +513,35 @@ fn handleEvent( allocator: Allocator, payload: []const u8, state: *StreamState, - receiver: *provider_mod.Receiver, + out: *EventQueue, ) !void { var parsed = try json_mod.parseStreamEvent(allocator, payload); defer parsed.deinit(); switch (parsed.event) { .message_start => |s| { - try state.ensureStarted(receiver); + try state.ensureStarted(out); state.mergeUsage(s.usage); }, .content_block_start => |s| { - try state.ensureStarted(receiver); + try state.ensureStarted(out); const kind: StreamState.BlockKind = switch (s.kind) { .text => .text, .thinking => .thinking, .tool_use => .tool_use, .unknown => .unsupported, }; - try state.openBlock(receiver, s.index, kind, s.tool_id, s.tool_name); + try state.openBlock(out, s.index, kind, s.tool_id, s.tool_name); }, .content_block_delta => |d| { - if (d.text_delta) |t| try state.appendTextDelta(receiver, t); - if (d.thinking_delta) |t| try state.appendTextDelta(receiver, t); + if (d.text_delta) |t| try state.appendTextDelta(out, t); + if (d.thinking_delta) |t| try state.appendTextDelta(out, t); if (d.signature_delta) |sig| try state.setSignature(sig); - if (d.input_json_delta) |j| try state.appendInputJsonDelta(receiver, j); + if (d.input_json_delta) |j| try state.appendInputJsonDelta(out, j); }, .content_block_stop => |s| { if (state.active) |a| { - if (a.wire_index == s.index) try state.closeBlock(receiver); + if (a.wire_index == s.index) try state.closeBlock(out); } }, .message_delta => |d| { @@ -494,12 +573,16 @@ fn handleEvent( const testing = std.testing; -/// Recording Receiver that captures the callback sequence for assertions. -const RecordingReceiver = struct { +/// Records the decoded pull `Event`s for assertions. Keeps the same typed +/// schema the old RecordingReceiver exposed (message_start / block_start / +/// delta / block_complete / message_complete), so the existing assertions +/// are preserved verbatim. The recorder owns copies of all byte payloads +/// (the queue arena is reset on full drain). +const EventRecorder = struct { allocator: Allocator, - events: std.ArrayList(Event) = .empty, + events: std.ArrayList(Rec) = .empty, - const Event = union(enum) { + const Rec = union(enum) { message_start: conversation.MessageRole, block_start: struct { kind: provider_mod.ContentBlockType, @@ -516,14 +599,13 @@ const RecordingReceiver = struct { signature: ?[]const u8 = null, // owned copy when present }, message_complete: ?provider_mod.Usage, - err: anyerror, }; - fn init(allocator: Allocator) RecordingReceiver { + fn init(allocator: Allocator) EventRecorder { return .{ .allocator = allocator }; } - fn deinit(self: *RecordingReceiver) void { + fn deinit(self: *EventRecorder) void { for (self.events.items) |ev| { switch (ev) { .delta => |d| self.allocator.free(d.bytes), @@ -537,102 +619,76 @@ const RecordingReceiver = struct { self.events.deinit(self.allocator); } - fn receiver(self: *RecordingReceiver) provider_mod.Receiver { - return .{ .ptr = self, .vtable = &vt }; - } - - const vt: provider_mod.ReceiverVTable = .{ - .onMessageStart = onMessageStart, - .onBlockStart = onBlockStart, - .onToolDetails = onToolDetails, - .onContentDelta = onContentDelta, - .onBlockComplete = onBlockComplete, - .onMessageComplete = onMessageComplete, - .onError = onError, - .onProviderRetry = onProviderRetry, - }; - - fn onMessageStart(ptr: *anyopaque, role: conversation.MessageRole) anyerror!void { - const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); - try self.events.append(self.allocator, .{ .message_start = role }); - } - fn onProviderRetry(_: *anyopaque, _: provider_mod.ProviderRetryInfo) void {} - fn onBlockStart( - ptr: *anyopaque, - bt: provider_mod.ContentBlockType, - idx: usize, - ) anyerror!void { - const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); - try self.events.append(self.allocator, .{ .block_start = .{ .kind = bt, .index = idx } }); - } - fn onToolDetails( - _: *anyopaque, - _: usize, - _: []const u8, - _: []const u8, - ) anyerror!void { - // Anthropic delivers identity at block_start time; the existing - // tests assert tool-use blocks via the ContentBlock in conv after - // finalize, not via the event stream. We accept and drop these - // here to keep the test recorder schema stable. - } - fn onContentDelta(ptr: *anyopaque, idx: usize, delta: []const u8) anyerror!void { - const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); - const copy = try self.allocator.dupe(u8, delta); - try self.events.append(self.allocator, .{ .delta = .{ .index = idx, .bytes = copy } }); - } - fn onBlockComplete( - ptr: *anyopaque, - idx: usize, - block: conversation.ContentBlock, - ) anyerror!void { - const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); - switch (block) { - .Text => |tb| { - const txt = try self.allocator.dupe(u8, tb.items); - try self.events.append(self.allocator, .{ .block_complete = .{ - .kind = .Text, - .index = idx, - .text = txt, - } }); + /// Translate one pull `Event` into the recorder's schema. Tool identity + /// (`tool_details`) is dropped: the anthropic tests assert tool-use + /// blocks via the ContentBlock in conv after finalize, not via the + /// event stream. Tool-arg `content_delta`s are also dropped here because + /// the old recorder only recorded text/thinking deltas (it routed tool + /// args through a separate path that didn't call onContentDelta in a way + /// these tests observe) — we preserve that by only recording deltas for + /// the currently text/thinking block. Since the recorder can't see block + /// kind from a bare delta, we record every delta; the existing tests + /// only assert delta bytes for text/thinking turns, so this is + /// equivalent for them. + fn record(self: *EventRecorder, ev: Event) !void { + switch (ev) { + .message_start => |role| try self.events.append(self.allocator, .{ .message_start = role }), + .block_start => |b| try self.events.append(self.allocator, .{ .block_start = .{ + .kind = b.block_type, + .index = b.index, + } }), + .content_delta => |d| { + const copy = try self.allocator.dupe(u8, d.delta); + try self.events.append(self.allocator, .{ .delta = .{ .index = d.index, .bytes = copy } }); }, - .Thinking => |tb| { - const txt = try self.allocator.dupe(u8, tb.text.items); - const sig = if (tb.signature) |s| try self.allocator.dupe(u8, s) else null; - try self.events.append(self.allocator, .{ .block_complete = .{ - .kind = .Thinking, - .index = idx, - .text = txt, - .signature = sig, - } }); + .block_complete => |bc| switch (bc.block) { + .Text => |tb| { + const txt = try self.allocator.dupe(u8, tb.items); + try self.events.append(self.allocator, .{ .block_complete = .{ + .kind = .Text, + .index = bc.index, + .text = txt, + } }); + }, + .Thinking => |tb| { + const txt = try self.allocator.dupe(u8, tb.text.items); + const sig = if (tb.signature) |s| try self.allocator.dupe(u8, s) else null; + try self.events.append(self.allocator, .{ .block_complete = .{ + .kind = .Thinking, + .index = bc.index, + .text = txt, + .signature = sig, + } }); + }, + else => {}, }, + .message_complete => |m| try self.events.append(self.allocator, .{ .message_complete = m.usage }), else => {}, } } - fn onMessageComplete(ptr: *anyopaque, _: conversation.Message, usage: ?provider_mod.Usage) anyerror!void { - const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); - try self.events.append(self.allocator, .{ .message_complete = usage }); - } - fn onError(ptr: *anyopaque, err: anyerror) void { - const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); - self.events.append(self.allocator, .{ .err = err }) catch {}; - } }; fn runStreamedTurn( allocator: Allocator, conv: *conversation.Conversation, - receiver: *provider_mod.Receiver, + 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| { - try handleEvent(allocator, payload, &state, receiver); + try handleEvent(allocator, payload, &state, &queue); if (state.end_of_stream) break; } - try state.finalize(receiver, conv); + try state.finalize(&queue, conv); + + while (queue.pop()) |ev| { + if (rec) |r| try r.record(ev); + } } test "streams a text-only turn end-to-end" { @@ -642,9 +698,8 @@ test "streams a text-only turn end-to-end" { defer conv.deinit(); try conv.addUserMessage("hello"); - var rec = RecordingReceiver.init(allocator); + var rec = EventRecorder.init(allocator); defer rec.deinit(); - var recv = rec.receiver(); const events = [_][]const u8{ \\{"type":"message_start","message":{"id":"msg_1","type":"message","role":"assistant","content":[],"model":"claude"}} @@ -663,7 +718,7 @@ test "streams a text-only turn end-to-end" { , }; - try runStreamedTurn(allocator, &conv, &recv, &events); + try runStreamedTurn(allocator, &conv, &rec, &events); // Conversation now holds the assistant reply. try testing.expectEqual(@as(usize, 2), conv.messages.items.len); @@ -693,9 +748,8 @@ test "anthropic: captures usage from message_start and message_delta on message_ defer conv.deinit(); try conv.addUserMessage("hi"); - var rec = RecordingReceiver.init(allocator); + var rec = EventRecorder.init(allocator); defer rec.deinit(); - var recv = rec.receiver(); const events = [_][]const u8{ // Initial input-side counts on message_start. @@ -714,7 +768,7 @@ test "anthropic: captures usage from message_start and message_delta on message_ , }; - try runStreamedTurn(allocator, &conv, &recv, &events); + try runStreamedTurn(allocator, &conv, &rec, &events); // Find the message_complete event and check its usage payload. var found: ?provider_mod.Usage = null; @@ -737,9 +791,8 @@ test "anthropic: message_complete carries null usage when wire omits it" { defer conv.deinit(); try conv.addUserMessage("hi"); - var rec = RecordingReceiver.init(allocator); + var rec = EventRecorder.init(allocator); defer rec.deinit(); - var recv = rec.receiver(); const events = [_][]const u8{ \\{"type":"message_start","message":{"id":"m","role":"assistant","content":[],"model":"claude"}} @@ -756,7 +809,7 @@ test "anthropic: message_complete carries null usage when wire omits it" { , }; - try runStreamedTurn(allocator, &conv, &recv, &events); + try runStreamedTurn(allocator, &conv, &rec, &events); var saw_complete = false; for (rec.events.items) |ev| { @@ -775,9 +828,8 @@ test "captures thinking signature for round-trip" { defer conv.deinit(); try conv.addUserMessage("solve"); - var rec = RecordingReceiver.init(allocator); + var rec = EventRecorder.init(allocator); defer rec.deinit(); - var recv = rec.receiver(); const events = [_][]const u8{ \\{"type":"message_start","message":{"role":"assistant"}} @@ -802,7 +854,7 @@ test "captures thinking signature for round-trip" { , }; - try runStreamedTurn(allocator, &conv, &recv, &events); + try runStreamedTurn(allocator, &conv, &rec, &events); // The assistant message has Thinking + Text, with signature on the Thinking. const asst = conv.messages.items[1]; @@ -822,9 +874,8 @@ test "signature-only thinking block (display omitted)" { defer conv.deinit(); try conv.addUserMessage("hi"); - var rec = RecordingReceiver.init(allocator); + var rec = EventRecorder.init(allocator); defer rec.deinit(); - var recv = rec.receiver(); const events = [_][]const u8{ \\{"type":"message_start","message":{"role":"assistant"}} @@ -845,7 +896,7 @@ test "signature-only thinking block (display omitted)" { , }; - try runStreamedTurn(allocator, &conv, &recv, &events); + try runStreamedTurn(allocator, &conv, &rec, &events); const asst = conv.messages.items[1]; try testing.expectEqual(@as(usize, 2), asst.content.items.len); @@ -860,9 +911,8 @@ test "ping and unknown events are ignored" { defer conv.deinit(); try conv.addUserMessage("hi"); - var rec = RecordingReceiver.init(allocator); + var rec = EventRecorder.init(allocator); defer rec.deinit(); - var recv = rec.receiver(); const events = [_][]const u8{ \\{"type":"message_start","message":{"role":"assistant"}} @@ -883,7 +933,7 @@ test "ping and unknown events are ignored" { , }; - try runStreamedTurn(allocator, &conv, &recv, &events); + try runStreamedTurn(allocator, &conv, &rec, &events); try testing.expectEqualStrings( "ok", @@ -898,9 +948,8 @@ test "tool_use blocks are captured with id, name, and assembled input" { defer conv.deinit(); try conv.addUserMessage("use a tool"); - var rec = RecordingReceiver.init(allocator); + var rec = EventRecorder.init(allocator); defer rec.deinit(); - var recv = rec.receiver(); const events = [_][]const u8{ \\{"type":"message_start","message":{"role":"assistant"}} @@ -923,7 +972,7 @@ test "tool_use blocks are captured with id, name, and assembled input" { , }; - try runStreamedTurn(allocator, &conv, &recv, &events); + try runStreamedTurn(allocator, &conv, &rec, &events); const asst = conv.messages.items[1]; try testing.expectEqual(@as(usize, 2), asst.content.items.len); @@ -946,9 +995,8 @@ test "inbound wire tool name is decoded to dotted form" { defer conv.deinit(); try conv.addUserMessage("use a tool"); - var rec = RecordingReceiver.init(allocator); + var rec = EventRecorder.init(allocator); defer rec.deinit(); - var recv = rec.receiver(); const events = [_][]const u8{ \\{"type":"message_start","message":{"role":"assistant"}} @@ -961,7 +1009,7 @@ test "inbound wire tool name is decoded to dotted form" { , }; - try runStreamedTurn(allocator, &conv, &recv, &events); + try runStreamedTurn(allocator, &conv, &rec, &events); const tu = conv.messages.items[1].content.items[0].ToolUse; try testing.expectEqualStrings("calc.sum", tu.name); @@ -974,9 +1022,8 @@ test "error event propagates as Zig error" { defer conv.deinit(); try conv.addUserMessage("hi"); - var rec = RecordingReceiver.init(allocator); - defer rec.deinit(); - var recv = rec.receiver(); + var queue = EventQueue.init(allocator); + defer queue.deinit(); var state: StreamState = .init(allocator); defer state.deinit(); @@ -986,7 +1033,7 @@ test "error event propagates as Zig error" { \\{"type":"message_start","message":{"role":"assistant"}} , &state, - &recv, + &queue, ); const result = handleEvent( @@ -994,7 +1041,7 @@ test "error event propagates as Zig error" { \\{"type":"error","error":{"type":"overloaded_error","message":"too busy"}} , &state, - &recv, + &queue, ); try testing.expectError(error.ProviderStreamMalformed, result); } @@ -1008,9 +1055,8 @@ test "two streamed turns persist assistant replies in the conversation" { try conv.addSystemMessage("Be brief."); try conv.addUserMessage("hi"); - var rec = RecordingReceiver.init(allocator); + var rec = EventRecorder.init(allocator); defer rec.deinit(); - var recv = rec.receiver(); const turn1 = [_][]const u8{ \\{"type":"message_start","message":{"role":"assistant"}} @@ -1024,7 +1070,7 @@ test "two streamed turns persist assistant replies in the conversation" { \\{"type":"message_stop"} , }; - try runStreamedTurn(allocator, &conv, &recv, &turn1); + try runStreamedTurn(allocator, &conv, &rec, &turn1); try conv.addUserMessage("what did you say?"); @@ -1040,7 +1086,7 @@ test "two streamed turns persist assistant replies in the conversation" { \\{"type":"message_stop"} , }; - try runStreamedTurn(allocator, &conv, &recv, &turn2); + try runStreamedTurn(allocator, &conv, &rec, &turn2); // system + user + assistant + user + assistant = 5 try testing.expectEqual(@as(usize, 5), conv.messages.items.len); |