1 files changed, 121 insertions, 0 deletions
diff --git a/libpanto/src/provider.zig b/libpanto/src/provider.zig
index 3c3b52e..e12ff59 100644
--- a/libpanto/src/provider.zig
+++ b/libpanto/src/provider.zig
@@ -1,4 +1,6 @@
 const std = @import("std");
+const http = std.http;
+const Uri = std.Uri;
 
 const config_mod = @import("config.zig");
 const conversation = @import("conversation.zig");
@@ -10,6 +12,109 @@ pub const Usage = session_mod.Usage;
 
 const EventQueue = stream_mod.EventQueue;
 
+/// Open a streaming `POST` to `uri` carrying `body`, with the shared transport
+/// options every provider uses (identity encoding so gzip can't buffer SSE
+/// frames, no keep-alive, no redirects). `req` must point at pinned storage —
+/// the body writer and `receiveHead` borrow it, and the caller keeps it for
+/// the response's lifetime. On success the request is left open (the caller
+/// owns it) and the received response head is returned; on failure the request
+/// is cleaned up before returning the error.
+pub fn sendRequest(
+    client: *http.Client,
+    uri: Uri,
+    extra_headers: []const http.Header,
+    body: []const u8,
+    req: *http.Client.Request,
+) !http.Client.Response {
+    req.* = try 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,
+    });
+    errdefer req.deinit();
+
+    req.transfer_encoding = .{ .content_length = body.len };
+    var send_buf: [4096]u8 = undefined;
+    var bw = try req.sendBodyUnflushed(&send_buf);
+    try bw.writer.writeAll(body);
+    try bw.end();
+    try req.connection.?.flush();
+
+    var redirect_buf: [1024]u8 = undefined;
+    return try req.receiveHead(&redirect_buf);
+}
+
+/// Handle a >=400 provider response: capture `Retry-After`, drain the body
+/// (capped at 16 KiB) for diagnostics, classify the status, log it (demoting
+/// recoverable auth failures to `.debug`), stash status + retry into `diag`,
+/// and return the classified error for the caller to propagate. `transfer_buf`
+/// backs the drain reader; `name` is the provider's log prefix.
+pub fn classifyErrorResponse(
+    allocator: std.mem.Allocator,
+    response: *http.Client.Response,
+    transfer_buf: []u8,
+    diag: ?*ProviderDiagnostic,
+    name: []const u8,
+) ProviderError {
+    // `head.bytes` (which `iterateHeaders` walks) points into the connection
+    // read buffer and is invalidated the moment the body stream is
+    // initialized below. Capture Retry-After first.
+    const retry_after_ms = retryAfterFromHead(response.head);
+    const body_reader = response.reader(transfer_buf);
+    var err_buf: std.ArrayList(u8) = .empty;
+    defer err_buf.deinit(allocator);
+    var tmp: [1024]u8 = undefined;
+    while (true) {
+        const n = body_reader.readSliceShort(&tmp) catch break;
+        if (n == 0) break;
+        err_buf.appendSlice(allocator, tmp[0..n]) catch break;
+        if (err_buf.items.len > 16 * 1024) break;
+    }
+    const status: u16 = @intFromEnum(response.head.status);
+    const classified = classifyHttpStatus(status, err_buf.items);
+    // 401/403 is routinely recovered by the turn-runner's forced token
+    // refresh + reopen; demote it to `.debug` (still in the debug log) so a
+    // transparent refresh doesn't surface a scary error line. The retry layer
+    // raises a hard error only if recovery ultimately fails.
+    if (classified == error.ProviderAuthFailed) {
+        std.log.debug("{s} HTTP {d} (recoverable auth): {s}", .{ name, status, err_buf.items });
+    } else {
+        std.log.err("{s} HTTP {d}: {s}", .{ name, status, err_buf.items });
+    }
+    if (diag) |d| {
+        d.status_code = status;
+        d.retry_after_ms = retry_after_ms;
+    }
+    return classified;
+}
+
+/// 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 `__`.
+pub 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;
+}
+
+/// Combine a stream error's `kind` and `message` into one owned, human-readable
+/// string (either may be absent). Returns null when both are absent. Caller
+/// owns the result.
+pub fn formatStreamError(
+    allocator: std.mem.Allocator,
+    kind: ?[]const u8,
+    message: ?[]const u8,
+) std.mem.Allocator.Error!?[]u8 {
+    if (kind != null and message != null)
+        return try std.fmt.allocPrint(allocator, "{s}: {s}", .{ kind.?, message.? });
+    if (kind) |k| return try allocator.dupe(u8, k);
+    if (message) |m| return try allocator.dupe(u8, m);
+    return null;
+}
+
 pub const ContentBlockType = enum {
     Text,
     Thinking,
@@ -44,6 +149,22 @@ pub fn mergeHeaders(
     return out;
 }
 
+/// Splice a pre-encoded JSON value into the current stringifier position.
+/// Used to embed a tool's `input_schema` (and a replayed tool_use `input`)
+/// verbatim into a request body. On parse failure, emit `{}` so we never
+/// produce invalid wire JSON — an empty object is the correct degenerate
+/// value on the wire for both uses.
+pub fn writeRawJson(s: *std.json.Stringify, raw: []const u8) !void {
+    var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
+    defer arena.deinit();
+    const parsed = std.json.parseFromSlice(std.json.Value, arena.allocator(), raw, .{}) catch {
+        try s.beginObject();
+        try s.endObject();
+        return;
+    };
+    try s.write(parsed.value);
+}
+
 pub fn isContextOverflowBody(body: []const u8) bool {
     const markers = [_][]const u8{
         "context_length_exceeded",