From f83578fdc9264019a1a1cef8c5484a161167d3dd Mon Sep 17 00:00:00 2001 From: t Date: Tue, 7 Jul 2026 11:26:32 -0600 Subject: initial commit, moved libpanto over from the pantograph repo --- src/provider_anthropic_messages.zig | 1267 +++++++++++++++++++++++++++++++++++ 1 file changed, 1267 insertions(+) create mode 100644 src/provider_anthropic_messages.zig (limited to 'src/provider_anthropic_messages.zig') diff --git a/src/provider_anthropic_messages.zig b/src/provider_anthropic_messages.zig new file mode 100644 index 0000000..955991a --- /dev/null +++ b/src/provider_anthropic_messages.zig @@ -0,0 +1,1267 @@ +//! Anthropic Messages API streaming provider. +//! +//! Wire format reference: +//! https://platform.claude.com/docs/en/build-with-claude/streaming +//! +//! Responsibilities: +//! - Convert `Conversation` → request JSON (delegated to anthropic_messages_json.zig) +//! - POST to `{base_url}/v1/messages` with `stream: true` +//! (the provider owns the current `/v1` suffix; a future wire revision +//! would add a new `anthropic_messages_v2` API style rather than guessing +//! from the configured base URL) +//! - Read the chunked body, feed bytes through SSEParser +//! - Parse each event payload, drive a thin assembly loop, and emit Receiver +//! callbacks. Anthropic gives us explicit block boundaries, so no +//! state-machine inference is needed. +//! - 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("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. +pub const AnthropicMessagesRequest = struct { + allocator: Allocator, + io: Io, + config: *const config_mod.AnthropicMessagesConfig, + http_client: *http.Client, + /// Optional diagnostic side-channel; see `OpenAIChatRequest.diag`. + diag: ?*provider_mod.ProviderDiagnostic = null, + + /// 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, + ) !*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), + }; + rr.state.signature_origin = try conversation.SignatureOrigin.init( + self.allocator, + .anthropic_messages, + self.config.base_url, + self.config.model, + ); + errdefer { + rr.parser.deinit(); + rr.state.deinit(); + } + + const trimmed_base = std.mem.trim(u8, self.config.base_url, "/"); + const url = try std.fmt.allocPrint( + self.allocator, + "{s}/v1/messages", + .{trimmed_base}, + ); + defer self.allocator.free(url); + + const uri = try Uri.parse(url); + + const body = try json_mod.serializeRequest(self.allocator, self.config, conv, tools); + defer self.allocator.free(body); + std.log.debug("anthropic_messages => {s}", .{body}); + + // Build headers. Standard Anthropic-compatible backends use + // `x-api-key`. OAuth-backed Anthropic-compatible providers (e.g. + // Copilot) opt into `Authorization: Bearer ...` via + // `config.use_bearer_auth`, which is derived from the configured auth + // family rather than guessed from the URL/headers. The four base + // headers are always present; the interleaved-thinking beta header is + // added only when the config explicitly requests manual extended + // thinking with interleaving. It is intentionally NOT sent for + // `.adaptive` (interleaving is automatic there and the header causes + // 400s on some backends) or `.disabled`. + const use_bearer_auth = self.config.use_bearer_auth; + const auth_value = if (use_bearer_auth) + try std.fmt.allocPrint( + self.allocator, + "Bearer {s}", + .{self.config.api_key}, + ) + else + ""; + defer if (use_bearer_auth) self.allocator.free(auth_value); + const auth_header: http.Header = if (use_bearer_auth) + .{ .name = "authorization", .value = auth_value } + else + .{ .name = "x-api-key", .value = self.config.api_key }; + var headers_buf: [5]http.Header = .{ + .{ .name = "content-type", .value = "application/json" }, + .{ .name = "accept", .value = "text/event-stream" }, + auth_header, + .{ .name = "anthropic-version", .value = self.config.api_version }, + undefined, // slot reserved for the optional beta header + }; + const send_interleaved = self.config.thinking == .enabled and + self.config.thinking_interleaved; + if (send_interleaved) { + headers_buf[4] = .{ + .name = "anthropic-beta", + .value = "interleaved-thinking-2025-05-14", + }; + } + const base_headers = headers_buf[0..if (send_interleaved) @as(usize, 5) else @as(usize, 4)]; + // Merge any provider `extra_headers` onto the base set. Freed at the + // end of `open` — after the request body has been flushed. + const extra_headers = try provider_mod.mergeHeaders( + self.allocator, + base_headers, + self.config.extra_headers, + ); + defer self.allocator.free(extra_headers); + + rr.response = try provider_mod.sendRequest(self.http_client, uri, extra_headers, body, &rr.req); + rr.req_open = true; + errdefer { + rr.req.deinit(); + rr.req_open = false; + } + + // A >=400 status maps to a retryable/terminal provider error. Anthropic + // rejects oversized requests with HTTP 400 + "prompt is too long", + // which `classifyErrorResponse` maps to ContextOverflow (compact+retry). + if (@intFromEnum(rr.response.head.status) >= 400) { + return provider_mod.classifyErrorResponse(self.allocator, &rr.response, &rr.transfer_buf, self.diag, "anthropic_messages"); + } + + 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, + + 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, + + req_open: bool = false, + 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, + .last_error = lastErrorVT, + }; + + fn produceVT(ptr: *anyopaque, out: *EventQueue) anyerror!ProduceStatus { + const self: *ResumableResponse = @ptrCast(@alignCast(ptr)); + return self.produce(out); + } + + fn lastErrorVT(ptr: *anyopaque) ?[]const u8 { + const self: *ResumableResponse = @ptrCast(@alignCast(ptr)); + return self.state.stream_error_message; + } + + 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; + } + + 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. +/// +/// Anthropic gives us explicit block boundaries (`content_block_start` / +/// `content_block_stop`), so we don't need to infer transitions like +/// `provider_openai_chat` does. We just track the currently-open block. +const StreamState = struct { + allocator: Allocator, + started: bool = false, + end_of_stream: bool = false, + finalized: bool = false, + + /// The block currently being assembled (if any). + active: ?ActiveBlock = null, + + /// Assembled blocks for the final message, in stream order. + blocks: std.ArrayList(conversation.ContentBlock) = .empty, + + /// Accumulated token counts. Anthropic reports the input-side counts + /// on `message_start.usage` and the final `output_tokens` (plus + /// possibly updated input-side counts) on `message_delta.usage`. + /// `usage_seen` distinguishes "genuinely all zero" from "never + /// reported" — the former stamps a real `Usage` on + /// `onMessageComplete`, the latter stamps `null`. + usage: provider_mod.Usage = .{}, + usage_seen: bool = false, + signature_origin: ?conversation.SignatureOrigin = null, + stop_reason: ?[]u8 = null, + /// Owned, human-readable description of a mid-stream `error` event + /// (e.g. `"overloaded_error: Overloaded"`), surfaced to the agent via + /// `ProviderStream.lastError` so the retry notice can show *why*. + stream_error_message: ?[]u8 = null, + + const ActiveBlock = struct { + /// Index reported on the wire (Anthropic's content-array index). + wire_index: usize, + kind: BlockKind, + text_buf: conversation.TextualBlock = .empty, + signature: ?[]const u8 = null, + /// Populated for `.tool_use` blocks. Owned by this state until the + /// block closes, at which point ownership transfers to the + /// ToolUseBlock. + tool_id: ?[]u8 = null, + tool_name: ?[]u8 = null, + }; + + const BlockKind = enum { text, thinking, tool_use, unsupported }; + + fn init(allocator: Allocator) StreamState { + return .{ .allocator = allocator }; + } + + fn deinit(self: *StreamState) void { + if (self.active) |*a| { + a.text_buf.deinit(self.allocator); + if (a.signature) |sig| self.allocator.free(sig); + if (a.tool_id) |s| self.allocator.free(s); + if (a.tool_name) |s| self.allocator.free(s); + } + for (self.blocks.items) |*b| b.deinit(self.allocator); + self.blocks.deinit(self.allocator); + if (self.signature_origin) |*o| o.deinit(self.allocator); + if (self.stop_reason) |s| self.allocator.free(s); + if (self.stream_error_message) |s| self.allocator.free(s); + } + + fn ensureStarted(self: *StreamState, out: *EventQueue) !void { + if (self.started) return; + self.started = true; + try out.push(.{ .message_start = .assistant }); + } + + /// Merge a wire-level usage snapshot into the accumulated counts. + /// Missing fields mean "unchanged," not "reset to zero." Marks + /// `usage_seen` so `finalize` delivers a non-null `Usage` to + /// `onMessageComplete`. + fn mergeUsage(self: *StreamState, partial: json_mod.StreamUsage) void { + if (partial.input_tokens) |v| self.usage.input = v; + if (partial.output_tokens) |v| self.usage.output = v; + if (partial.cache_creation_input_tokens) |v| self.usage.cache_write = v; + if (partial.cache_read_input_tokens) |v| self.usage.cache_read = v; + if (partial.input_tokens != null or partial.output_tokens != null or + partial.cache_creation_input_tokens != null or partial.cache_read_input_tokens != null) + { + self.usage_seen = true; + } + } + + fn openBlock( + self: *StreamState, + out: *EventQueue, + wire_index: usize, + kind: BlockKind, + tool_id: ?[]const u8, + tool_name: ?[]const u8, + ) !void { + // Defensive: if a prior block didn't get an explicit stop, drop it. + if (self.active != null) { + self.discardActive(); + } + var ab: ActiveBlock = .{ + .wire_index = wire_index, + .kind = kind, + }; + // For tool_use blocks, capture the identity fields. Anthropic + // delivers both whole on content_block_start. The wire name is + // encoded (`__` for `.`); decode it here so everything downstream + // — onToolDetails, the stored ContentBlock, session logs, and + // dispatch — sees the internal (dotted) name. The decoded form is + // never longer than the wire form. + if (kind == .tool_use) { + if (tool_id) |id| ab.tool_id = try self.allocator.dupe(u8, id); + if (tool_name) |n| { + // Decode `__` -> `.` into an exact-size owned buffer so the + // stored slice is freeable as a whole allocation. + const owned = try self.allocator.alloc(u8, n.len); + errdefer self.allocator.free(owned); + const decoded = tool_registry_mod.decodeName(owned, n); + if (decoded.len == n.len) { + ab.tool_name = owned; + } else { + ab.tool_name = try self.allocator.realloc(owned, decoded.len); + } + } + } + self.active = ab; + const block_type: ?provider_mod.ContentBlockType = switch (kind) { + .text => .Text, + .thinking => .Thinking, + .tool_use => .ToolUse, + .unsupported => null, + }; + if (block_type) |bt| { + 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 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 out.push(.{ .tool_details = .{ + .index = wire_index, + .id = try out.dupeBytes(ab.tool_id.?), + .name = try out.dupeBytes(ab.tool_name.?), + } }); + } + } + } + } + + fn appendTextDelta( + self: *StreamState, + 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); + // 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, + 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 out.push(.{ .content_delta = .{ + .index = a.wire_index, + .delta = try out.dupeBytes(delta), + } }); + } + + fn setSignature(self: *StreamState, sig: []const u8) !void { + const a = &(self.active orelse return); + if (a.signature) |old| self.allocator.free(old); + a.signature = try self.allocator.dupe(u8, sig); + } + + fn setStopReason(self: *StreamState, reason: ?[]const u8) !void { + if (self.stop_reason) |old| self.allocator.free(old); + self.stop_reason = if (reason) |r| try self.allocator.dupe(u8, r) else null; + } + + /// Record a readable description of a mid-stream `error` event, combining + /// the error `kind` and `message` into one owned string (either may be + /// absent). Replaces any previous value. + fn setStreamErrorMessage(self: *StreamState, kind: ?[]const u8, message: ?[]const u8) !void { + if (self.stream_error_message) |old| self.allocator.free(old); + self.stream_error_message = try provider_mod.formatStreamError(self.allocator, kind, message); + } + + /// Close the active block: append it to `blocks` and emit block_complete. + fn closeBlock( + self: *StreamState, + out: *EventQueue, + ) !void { + var a = self.active orelse return; + self.active = null; + + if (a.kind == .unsupported) { + a.text_buf.deinit(self.allocator); + if (a.signature) |sig| self.allocator.free(sig); + if (a.tool_id) |s| self.allocator.free(s); + if (a.tool_name) |s| self.allocator.free(s); + return; + } + // tool_use blocks require both id and name. If either is missing + // (malformed stream), drop the block defensively. + if (a.kind == .tool_use and (a.tool_id == null or a.tool_name == null)) { + a.text_buf.deinit(self.allocator); + if (a.tool_id) |s| self.allocator.free(s); + if (a.tool_name) |s| self.allocator.free(s); + return; + } + + const block: conversation.ContentBlock = switch (a.kind) { + .text => blk: { + if (a.signature) |sig| self.allocator.free(sig); + break :blk .{ .Text = a.text_buf }; + }, + .thinking => .{ .Thinking = .{ + .text = a.text_buf, + .signature = a.signature, + } }, + // An interrupted/malformed tool_use (incomplete or non-object + // input JSON) is preserved as-is. The agent's dispatch path + // detects invalid input and answers it with a synthetic error + // ToolResult in the *following user message* — emitting a + // ToolResult here would wrongly place it in this assistant + // message, which Anthropic rejects. + .tool_use => blk: { + break :blk .{ .ToolUse = .{ + .id = a.tool_id.?, + .name = a.tool_name.?, + .input = a.text_buf, + } }; + }, + .unsupported => unreachable, + }; + + try self.blocks.append(self.allocator, block); + 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. + /// Used when an unexpected `content_block_start` arrives before the + /// previous block closed. + fn discardActive(self: *StreamState) void { + if (self.active) |*a| { + a.text_buf.deinit(self.allocator); + if (a.signature) |sig| self.allocator.free(sig); + if (a.tool_id) |s| self.allocator.free(s); + if (a.tool_name) |s| self.allocator.free(s); + self.active = null; + } + } + + fn finalize( + self: *StreamState, + out: *EventQueue, + conv: *conversation.Conversation, + ) !void { + if (self.finalized) return; + self.finalized = true; + + 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(out); + } + + const moved_blocks = try self.blocks.toOwnedSlice(self.allocator); + defer self.allocator.free(moved_blocks); + + const usage: ?provider_mod.Usage = if (self.usage_seen) self.usage else null; + if (self.signature_origin) |origin| { + try conversation.setThinkingOrigins( + self.allocator, + moved_blocks, + origin.api_style, + origin.base_url, + origin.model, + ); + } + try conv.addAssistantMessage(moved_blocks, usage); + + const msg = conv.messages.items[conv.messages.items.len - 1]; + try out.push(.{ .message_complete = .{ .message = msg, .usage = usage } }); + } +}; + +fn handleEvent( + allocator: Allocator, + payload: []const u8, + state: *StreamState, + out: *EventQueue, +) !void { + var parsed = try json_mod.parseStreamEvent(allocator, payload); + defer parsed.deinit(); + + switch (parsed.event) { + .message_start => |s| { + try state.ensureStarted(out); + state.mergeUsage(s.usage); + }, + .content_block_start => |s| { + try state.ensureStarted(out); + const kind: StreamState.BlockKind = switch (s.kind) { + .text => .text, + .thinking => .thinking, + .tool_use => .tool_use, + .unknown => .unsupported, + }; + 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(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(out, j); + }, + .content_block_stop => |s| { + if (state.active) |a| { + if (a.wire_index == s.index) try state.closeBlock(out); + } + }, + .message_delta => |d| { + try state.setStopReason(d.stop_reason); + state.mergeUsage(d.usage); + }, + .message_stop => { + state.end_of_stream = true; + }, + .ping => {}, + .@"error" => |e| { + if (!@import("builtin").is_test) { + std.log.err("anthropic stream error: {?s}: {?s}", .{ e.kind, e.message }); + } + // Stash a readable description so the agent's retry notice can + // explain *why* the stream failed instead of only showing the + // bare error name. Owned by `state`; freed in `deinit`. + state.setStreamErrorMessage(e.kind, e.message) catch {}; + // Mid-stream error event before the message was committed. Map + // the common overload case to a dedicated retryable error so the + // UI can say "overloaded" rather than "malformed stream"; other + // kinds stay as the generic retryable malformed-stream error. + if (e.kind) |k| { + if (std.mem.eql(u8, k, "overloaded_error")) return error.ProviderOverloaded; + } + return error.ProviderStreamMalformed; + }, + .unknown => { + // Forward-compatible: ignore unknown event types per Anthropic's + // versioning policy. + }, + } +} + +// ----------------------------------------------------------------------------- +// Tests +// ----------------------------------------------------------------------------- + +const testing = std.testing; + +/// 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(Rec) = .empty, + + const Rec = union(enum) { + message_start: conversation.MessageRole, + block_start: struct { + kind: provider_mod.ContentBlockType, + index: usize, + }, + delta: struct { + index: usize, + bytes: []const u8, // owned copy + }, + block_complete: struct { + index: usize, + kind: provider_mod.ContentBlockType, + text: []const u8, // owned copy + signature: ?[]const u8 = null, // owned copy when present + }, + message_complete: ?provider_mod.Usage, + }; + + fn init(allocator: Allocator) EventRecorder { + return .{ .allocator = allocator }; + } + + fn deinit(self: *EventRecorder) void { + for (self.events.items) |ev| { + switch (ev) { + .delta => |d| self.allocator.free(d.bytes), + .block_complete => |b| { + self.allocator.free(b.text); + if (b.signature) |s| self.allocator.free(s); + }, + else => {}, + } + } + self.events.deinit(self.allocator); + } + + /// 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 } }); + }, + .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 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| { + try handleEvent(allocator, payload, &state, &queue); + if (state.end_of_stream) break; + } + try state.finalize(&queue, conv); + + while (queue.pop()) |ev| { + if (rec) |r| try r.record(ev); + } +} + +/// 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 "streams a text-only turn end-to-end" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try addUserText(&conv, "hello"); + + var rec = EventRecorder.init(allocator); + defer rec.deinit(); + + const events = [_][]const u8{ + \\{"type":"message_start","message":{"id":"msg_1","type":"message","role":"assistant","content":[],"model":"claude"}} + , + \\{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}} + , + \\{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hello"}} + , + \\{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"!"}} + , + \\{"type":"content_block_stop","index":0} + , + \\{"type":"message_delta","delta":{"stop_reason":"end_turn"}} + , + \\{"type":"message_stop"} + , + }; + + try runStreamedTurn(allocator, &conv, &rec, &events); + + // Conversation now holds the assistant reply. + try testing.expectEqual(@as(usize, 2), conv.messages.items.len); + try testing.expectEqual(conversation.MessageRole.assistant, conv.messages.items[1].role); + try testing.expectEqual(@as(usize, 1), conv.messages.items[1].content.items.len); + try testing.expectEqualStrings( + "Hello!", + conv.messages.items[1].content.items[0].Text.items, + ); + + // Callback sequence: msg_start, block_start, delta, delta, block_complete, msg_complete. + try testing.expectEqual(@as(usize, 6), rec.events.items.len); + try testing.expectEqual(conversation.MessageRole.assistant, rec.events.items[0].message_start); + try testing.expectEqual(provider_mod.ContentBlockType.Text, rec.events.items[1].block_start.kind); + try testing.expectEqualStrings("Hello", rec.events.items[2].delta.bytes); + try testing.expectEqualStrings("!", rec.events.items[3].delta.bytes); + try testing.expectEqualStrings("Hello!", rec.events.items[4].block_complete.text); + try testing.expect(rec.events.items[5] == .message_complete); + // No usage on the wire — the assertion is structural. + try testing.expect(rec.events.items[5].message_complete == null); +} + +test "anthropic: captures usage from message_start and message_delta on message_complete" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try addUserText(&conv, "hi"); + + var rec = EventRecorder.init(allocator); + defer rec.deinit(); + + const events = [_][]const u8{ + // Initial input-side counts on message_start. + \\{"type":"message_start","message":{"id":"m","type":"message","role":"assistant","content":[],"model":"claude","usage":{"input_tokens":100,"cache_creation_input_tokens":50,"cache_read_input_tokens":200}}} + , + \\{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}} + , + \\{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"ok"}} + , + \\{"type":"content_block_stop","index":0} + , + // Final output count on message_delta. + \\{"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":42}} + , + \\{"type":"message_stop"} + , + }; + + try runStreamedTurn(allocator, &conv, &rec, &events); + + // Find the message_complete event and check its usage payload. + var found: ?provider_mod.Usage = null; + for (rec.events.items) |ev| { + if (ev == .message_complete) found = ev.message_complete; + } + try testing.expect(found != null); + const u = found.?; + try testing.expectEqual(@as(u64, 100), u.input); + try testing.expectEqual(@as(u64, 42), u.output); + try testing.expectEqual(@as(u64, 200), u.cache_read); + try testing.expectEqual(@as(u64, 50), u.cache_write); + try testing.expectEqual(@as(u64, 0), u.reasoning); // Anthropic doesn't split reasoning separately. +} + +test "anthropic: message_complete carries null usage when wire omits it" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try addUserText(&conv, "hi"); + + var rec = EventRecorder.init(allocator); + defer rec.deinit(); + + const events = [_][]const u8{ + \\{"type":"message_start","message":{"id":"m","role":"assistant","content":[],"model":"claude"}} + , + \\{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}} + , + \\{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"hi"}} + , + \\{"type":"content_block_stop","index":0} + , + \\{"type":"message_delta","delta":{"stop_reason":"end_turn"}} + , + \\{"type":"message_stop"} + , + }; + + try runStreamedTurn(allocator, &conv, &rec, &events); + + var saw_complete = false; + for (rec.events.items) |ev| { + if (ev == .message_complete) { + saw_complete = true; + try testing.expect(ev.message_complete == null); + } + } + try testing.expect(saw_complete); +} + +test "captures thinking signature for round-trip" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try addUserText(&conv, "solve"); + + var rec = EventRecorder.init(allocator); + defer rec.deinit(); + + const events = [_][]const u8{ + \\{"type":"message_start","message":{"role":"assistant"}} + , + \\{"type":"content_block_start","index":0,"content_block":{"type":"thinking","thinking":"","signature":""}} + , + \\{"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"step one"}} + , + \\{"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":" step two"}} + , + \\{"type":"content_block_delta","index":0,"delta":{"type":"signature_delta","signature":"EqQBabc"}} + , + \\{"type":"content_block_stop","index":0} + , + \\{"type":"content_block_start","index":1,"content_block":{"type":"text","text":""}} + , + \\{"type":"content_block_delta","index":1,"delta":{"type":"text_delta","text":"answer"}} + , + \\{"type":"content_block_stop","index":1} + , + \\{"type":"message_stop"} + , + }; + + try runStreamedTurn(allocator, &conv, &rec, &events); + + // The assistant message has Thinking + Text, with signature on the Thinking. + const asst = conv.messages.items[1]; + try testing.expectEqual(@as(usize, 2), asst.content.items.len); + try testing.expectEqualStrings("step one step two", asst.content.items[0].Thinking.text.items); + try testing.expectEqualStrings("EqQBabc", asst.content.items[0].Thinking.signature.?); + try testing.expectEqualStrings("answer", asst.content.items[1].Text.items); +} + +test "signature-only thinking block (display omitted)" { + // Anthropic emits a thinking block with only a signature_delta when + // `display: "omitted"` is configured. Verify we still capture the + // signature with empty thinking text. + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try addUserText(&conv, "hi"); + + var rec = EventRecorder.init(allocator); + defer rec.deinit(); + + const events = [_][]const u8{ + \\{"type":"message_start","message":{"role":"assistant"}} + , + \\{"type":"content_block_start","index":0,"content_block":{"type":"thinking","thinking":"","signature":""}} + , + \\{"type":"content_block_delta","index":0,"delta":{"type":"signature_delta","signature":"sig123"}} + , + \\{"type":"content_block_stop","index":0} + , + \\{"type":"content_block_start","index":1,"content_block":{"type":"text","text":""}} + , + \\{"type":"content_block_delta","index":1,"delta":{"type":"text_delta","text":"hi back"}} + , + \\{"type":"content_block_stop","index":1} + , + \\{"type":"message_stop"} + , + }; + + try runStreamedTurn(allocator, &conv, &rec, &events); + + const asst = conv.messages.items[1]; + try testing.expectEqual(@as(usize, 2), asst.content.items.len); + try testing.expectEqualStrings("", asst.content.items[0].Thinking.text.items); + try testing.expectEqualStrings("sig123", asst.content.items[0].Thinking.signature.?); +} + +test "ping and unknown events are ignored" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try addUserText(&conv, "hi"); + + var rec = EventRecorder.init(allocator); + defer rec.deinit(); + + const events = [_][]const u8{ + \\{"type":"message_start","message":{"role":"assistant"}} + , + \\{"type":"ping"} + , + \\{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}} + , + \\{"type":"ping"} + , + \\{"type":"future_event_type","whatever":true} + , + \\{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"ok"}} + , + \\{"type":"content_block_stop","index":0} + , + \\{"type":"message_stop"} + , + }; + + try runStreamedTurn(allocator, &conv, &rec, &events); + + try testing.expectEqualStrings( + "ok", + conv.messages.items[1].content.items[0].Text.items, + ); +} + +test "tool_use blocks are captured with id, name, and assembled input" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try addUserText(&conv, "use a tool"); + + var rec = EventRecorder.init(allocator); + defer rec.deinit(); + + const events = [_][]const u8{ + \\{"type":"message_start","message":{"role":"assistant"}} + , + \\{"type":"content_block_start","index":0,"content_block":{"type":"tool_use","id":"tu_1","name":"calc","input":{}}} + , + \\{"type":"content_block_delta","index":0,"delta":{"type":"input_json_delta","partial_json":"{\"x\":"}} + , + \\{"type":"content_block_delta","index":0,"delta":{"type":"input_json_delta","partial_json":"1}"}} + , + \\{"type":"content_block_stop","index":0} + , + \\{"type":"content_block_start","index":1,"content_block":{"type":"text","text":""}} + , + \\{"type":"content_block_delta","index":1,"delta":{"type":"text_delta","text":"done"}} + , + \\{"type":"content_block_stop","index":1} + , + \\{"type":"message_stop"} + , + }; + + try runStreamedTurn(allocator, &conv, &rec, &events); + + const asst = conv.messages.items[1]; + try testing.expectEqual(@as(usize, 2), asst.content.items.len); + + const tu = asst.content.items[0].ToolUse; + try testing.expectEqualStrings("tu_1", tu.id); + try testing.expectEqualStrings("calc", tu.name); + try testing.expectEqualStrings("{\"x\":1}", tu.input.items); + + try testing.expectEqualStrings("done", asst.content.items[1].Text.items); +} + +test "inbound wire tool name is decoded to dotted form" { + // Anthropic delivers the (wire-encoded) name whole at + // content_block_start; it is decoded to the internal dotted form for + // the conversation, session logs, and dispatch. + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try addUserText(&conv, "use a tool"); + + var rec = EventRecorder.init(allocator); + defer rec.deinit(); + + const events = [_][]const u8{ + \\{"type":"message_start","message":{"role":"assistant"}} + , + \\{"type":"content_block_start","index":0,"content_block":{"type":"tool_use","id":"t1","name":"calc__sum","input":{}}} + , + \\{"type":"content_block_stop","index":0} + , + \\{"type":"message_stop"} + , + }; + + try runStreamedTurn(allocator, &conv, &rec, &events); + + const tu = conv.messages.items[1].content.items[0].ToolUse; + try testing.expectEqualStrings("calc.sum", tu.name); +} + +test "error event propagates as Zig error" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try addUserText(&conv, "hi"); + + var queue = EventQueue.init(allocator); + defer queue.deinit(); + + var state: StreamState = .init(allocator); + defer state.deinit(); + + try handleEvent( + allocator, + \\{"type":"message_start","message":{"role":"assistant"}} + , + &state, + &queue, + ); + + const result = handleEvent( + allocator, + \\{"type":"error","error":{"type":"overloaded_error","message":"too busy"}} + , + &state, + &queue, + ); + // `overloaded_error` maps to the dedicated retryable error, and the + // provider's diagnostic is stashed for the agent's retry notice. + try testing.expectError(error.ProviderOverloaded, result); + try testing.expectEqualStrings("overloaded_error: too busy", state.stream_error_message.?); +} + +test "non-overloaded error event stays malformed and stashes message" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try addUserText(&conv, "hi"); + + var queue = EventQueue.init(allocator); + defer queue.deinit(); + + var state: StreamState = .init(allocator); + defer state.deinit(); + + const result = handleEvent( + allocator, + \\{"type":"error","error":{"type":"api_error","message":"boom"}} + , + &state, + &queue, + ); + try testing.expectError(error.ProviderStreamMalformed, result); + try testing.expectEqualStrings("api_error: boom", state.stream_error_message.?); +} + +test "two streamed turns persist assistant replies in the conversation" { + // Same regression scenario as the openai_chat test, adapted to Anthropic. + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addSystemMessage("Be brief."); + try addUserText(&conv, "hi"); + + var rec = EventRecorder.init(allocator); + defer rec.deinit(); + + const turn1 = [_][]const u8{ + \\{"type":"message_start","message":{"role":"assistant"}} + , + \\{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}} + , + \\{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hi!"}} + , + \\{"type":"content_block_stop","index":0} + , + \\{"type":"message_stop"} + , + }; + try runStreamedTurn(allocator, &conv, &rec, &turn1); + + try addUserText(&conv, "what did you say?"); + + const turn2 = [_][]const u8{ + \\{"type":"message_start","message":{"role":"assistant"}} + , + \\{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}} + , + \\{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"I said: Hi!"}} + , + \\{"type":"content_block_stop","index":0} + , + \\{"type":"message_stop"} + , + }; + try runStreamedTurn(allocator, &conv, &rec, &turn2); + + // system + user + assistant + user + assistant = 5 + try testing.expectEqual(@as(usize, 5), conv.messages.items.len); + try testing.expectEqualStrings( + "I said: Hi!", + conv.messages.items[4].content.items[0].Text.items, + ); +} + +/// Helper: whether this config sends Bearer auth instead of `x-api-key`. +fn headerSliceUsesBearerAuth(cfg: *const config_mod.AnthropicMessagesConfig) bool { + return cfg.use_bearer_auth; +} + +/// Helper: build the header slice exactly as `open` does, given a config, +/// and return whether the interleaved beta header is present. +/// This lets us test the header-selection logic without a live HTTP connection. +fn headerSliceIncludesInterleaved(cfg: *const config_mod.AnthropicMessagesConfig) bool { + const send_interleaved = cfg.thinking == .enabled and cfg.thinking_interleaved; + return send_interleaved; +} + +test "oauth-backed anthropic auth uses bearer auth" { + const cfg: config_mod.AnthropicMessagesConfig = .{ + .api_key = "k", + .base_url = "https://api.individual.githubcopilot.com", + .model = "claude-sonnet-4-5", + .use_bearer_auth = true, + }; + try testing.expect(headerSliceUsesBearerAuth(&cfg)); +} + +test "plain anthropic auth uses x-api-key" { + const cfg: config_mod.AnthropicMessagesConfig = .{ + .api_key = "k", + .base_url = "https://api.anthropic.com", + .model = "claude-sonnet-4-5", + }; + try testing.expect(!headerSliceUsesBearerAuth(&cfg)); +} + +test "interleaved beta header: enabled when thinking=.enabled and interleaved=true" { + const cfg: config_mod.AnthropicMessagesConfig = .{ + .api_key = "k", + .base_url = "u", + .model = "m", + .thinking = .enabled, + .thinking_interleaved = true, + }; + try testing.expect(headerSliceIncludesInterleaved(&cfg)); +} + +test "interleaved beta header: absent when thinking=.enabled and interleaved=false" { + const cfg: config_mod.AnthropicMessagesConfig = .{ + .api_key = "k", + .base_url = "u", + .model = "m", + .thinking = .enabled, + .thinking_interleaved = false, + }; + try testing.expect(!headerSliceIncludesInterleaved(&cfg)); +} + +test "interleaved beta header: absent when thinking=.adaptive even if interleaved=true" { + const cfg: config_mod.AnthropicMessagesConfig = .{ + .api_key = "k", + .base_url = "u", + .model = "m", + .thinking = .adaptive, + .thinking_interleaved = true, + }; + // .adaptive does not send the header; interleaving is automatic there. + try testing.expect(!headerSliceIncludesInterleaved(&cfg)); +} + +test "interleaved beta header: absent when thinking=.disabled" { + const cfg: config_mod.AnthropicMessagesConfig = .{ + .api_key = "k", + .base_url = "u", + .model = "m", + .thinking = .disabled, + .thinking_interleaved = true, + }; + try testing.expect(!headerSliceIncludesInterleaved(&cfg)); +} -- cgit v1.3