From 538a5d926fa626a00cc3dc12c555f11f82867efd Mon Sep 17 00:00:00 2001 From: t Date: Mon, 22 Jun 2026 18:18:28 -0600 Subject: libpanto ponytail-audit simplifications --- libpanto/src/provider.zig | 121 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 121 insertions(+) (limited to 'libpanto/src/provider.zig') 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", -- cgit v1.3