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_openai_chat.zig | 80 +++-------------------------------- 1 file changed, 7 insertions(+), 73 deletions(-) (limited to 'libpanto/src/provider_openai_chat.zig') diff --git a/libpanto/src/provider_openai_chat.zig b/libpanto/src/provider_openai_chat.zig index 3f01e7f..f653ff6 100644 --- a/libpanto/src/provider_openai_chat.zig +++ b/libpanto/src/provider_openai_chat.zig @@ -22,20 +22,11 @@ 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; -} +const decodeNameInPlace = provider_mod.decodeNameInPlace; /// Active streaming block type tracked by the state machine. Mirrors the /// `ContentBlock` union variants but adds `.none` for "no block open yet". @@ -123,67 +114,18 @@ pub const OpenAIChatRequest = struct { // 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.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; } - 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); - + // A >=400 status classifies 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. if (@intFromEnum(rr.response.head.status) >= 400) { - // `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 = provider_mod.retryAfterFromHead(rr.response.head); - // 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); - // 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); - // 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("openai_chat HTTP {d} (recoverable auth): {s}", .{ status, err_buf.items }); - } else { - std.log.err("openai_chat HTTP {d}: {s}", .{ status, err_buf.items }); - } - if (self.diag) |d| { - d.status_code = status; - d.retry_after_ms = retry_after_ms; - } - return classified; + return provider_mod.classifyErrorResponse(self.allocator, &rr.response, &rr.transfer_buf, self.diag, "openai_chat"); } // Bind the streaming body reader. Valid for the lifetime of `rr` @@ -422,15 +364,7 @@ const StreamState = struct { /// 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 = null; - self.stream_error_message = if (kind != null and message != null) - try std.fmt.allocPrint(self.allocator, "{s}: {s}", .{ kind.?, message.? }) - else if (kind) |k| - try self.allocator.dupe(u8, k) - else if (message) |m| - try self.allocator.dupe(u8, m) - else - null; + self.stream_error_message = try provider_mod.formatStreamError(self.allocator, kind, message); } /// Close the active text/thinking block (if any) and emit -- cgit v1.3