diff options
| author | t <t@tjp.lol> | 2026-07-07 11:29:23 -0600 |
|---|---|---|
| committer | t <t@tjp.lol> | 2026-07-07 12:15:28 -0600 |
| commit | 621d7fee0ace729f8d44126032d2c6e13f72ee7f (patch) | |
| tree | a2938b0a9e5c0930e8644721abbe94875df9ff08 /libpanto/src/provider.zig | |
| parent | 0fa6d4209ff9b4a95e7d1955887aa4c73ee3423c (diff) | |
Remove the in-repo libpanto sources and binding projects from pantograph.
Consume libpantograph through the Zig package URL at
code.tjp.lol/libpantograph.git, including the Lua module artifact used by
CLI extensions.
Diffstat (limited to 'libpanto/src/provider.zig')
| -rw-r--r-- | libpanto/src/provider.zig | 554 |
1 files changed, 0 insertions, 554 deletions
diff --git a/libpanto/src/provider.zig b/libpanto/src/provider.zig deleted file mode 100644 index e12ff59..0000000 --- a/libpanto/src/provider.zig +++ /dev/null @@ -1,554 +0,0 @@ -const std = @import("std"); -const http = std.http; -const Uri = std.Uri; - -const config_mod = @import("config.zig"); -const conversation = @import("conversation.zig"); -const tool_registry_mod = @import("tool_registry.zig"); -const session_mod = @import("session.zig"); -const stream_mod = @import("stream.zig"); -pub const ToolRegistry = tool_registry_mod.ToolRegistry; -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, - ToolUse, - ToolResult, -}; - -/// Heuristic detector for provider context-overflow rejections, applied to -/// an HTTP 400 error response body. Both OpenAI-compatible and Anthropic -/// APIs reject oversized requests on the input side with HTTP 400 and a -/// recognizable message; matching it lets the agent compact and retry -/// instead of surfacing a hard error. -/// -/// Markers (case-sensitive substrings, as the wire emits them): -/// - OpenAI: `context_length_exceeded`, `maximum context length` -/// - Anthropic: `prompt is too long` -/// Merge a provider's built-in request headers with caller-supplied -/// `extra_headers` (config `Header`s) into one `[]std.http.Header`, allocated -/// with `alloc`. The caller owns the result and must free it once the request -/// has been sent. `base` comes first; `extra` is appended in order, so an -/// `extra` header with the same name as a base header is sent as a second -/// occurrence (provider-identity headers in practice never collide with the -/// fixed content-type/accept/authorization set). -pub fn mergeHeaders( - alloc: std.mem.Allocator, - base: []const std.http.Header, - extra: []const config_mod.Header, -) ![]std.http.Header { - const out = try alloc.alloc(std.http.Header, base.len + extra.len); - @memcpy(out[0..base.len], base); - for (extra, 0..) |h, i| out[base.len + i] = .{ .name = h.name, .value = h.value }; - 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", - "maximum context length", - "prompt is too long", - "context window", - }; - for (markers) |m| { - if (std.mem.indexOf(u8, body, m) != null) return true; - } - return false; -} - -/// Distinct provider/API failure classes the agent needs in order to decide -/// between retrying, compacting, and hard-failing. The transport/stream layer -/// maps HTTP status codes and connection failures onto these so the agent's -/// retry policy can switch on a stable, provider-agnostic name rather than a -/// broad `error.HttpError`. -/// -/// Zig errors cannot carry payloads, so status code, `Retry-After`, and the -/// provider's diagnostic message ride alongside via `ProviderDiagnostic` (an -/// out-parameter the provider fills before returning the error). -pub const ProviderError = error{ - /// HTTP 429. Retryable; honor `Retry-After` when present. - ProviderRateLimited, - /// HTTP 503 or a server explicitly signalling unavailability. Retryable. - ProviderUnavailable, - /// HTTP 500/502/504. Retryable. - ProviderServerError, - /// Connection reset, DNS/connect failure, TLS failure, request timeout - /// (HTTP 408), or other transport-level failure before a response. Also - /// covers retryable conflict/not-ready statuses (409, 425). Retryable. - ProviderTransport, - /// The provider stream ended or was malformed before a complete - /// assistant message was committed. Retryable (subject to policy). - ProviderStreamMalformed, - /// HTTP 401/403. Not retryable — a credentials/permissions problem. - ProviderAuthFailed, - /// HTTP 400 (other than context overflow). Not retryable — the request - /// itself is malformed. - ProviderBadRequest, - /// HTTP 404 shaped as an unknown-model error. Not retryable. - ProviderModelNotFound, - /// The input context exceeds the model's window (HTTP 400 + a recognized - /// context marker). Handled by one-shot compaction, not ordinary retry. - ContextOverflow, -}; - -/// Side-channel for the payload a `ProviderError` cannot carry. The provider -/// fills the relevant fields immediately before returning a classified error; -/// the agent reads them to drive backoff and retry notifications. Reset to -/// `.{}` before each provider attempt. -pub const ProviderDiagnostic = struct { - /// The HTTP status code, when the failure carried one. - status_code: ?u16 = null, - /// Parsed `Retry-After` delay in milliseconds, when the provider sent it. - retry_after_ms: ?u64 = null, - /// The provider's diagnostic message (borrowed/owned per caller; the - /// agent treats it as borrowed for the lifetime of the failed attempt). - message: ?[]const u8 = null, - - pub fn reset(self: *ProviderDiagnostic) void { - self.* = .{}; - } -}; - -/// True for the provider errors the agent's retry policy may retry. Context -/// overflow is deliberately excluded — it has a separate one-shot compaction -/// path. Auth, bad-request, and model-not-found are terminal. -pub fn isRetryableProviderError(err: anyerror) bool { - return switch (err) { - error.ProviderRateLimited, - error.ProviderUnavailable, - error.ProviderServerError, - error.ProviderTransport, - error.ProviderStreamMalformed, - error.ProviderOverloaded, - => true, - else => false, - }; -} - -/// Map an HTTP status code from a provider response onto a `ProviderError`. -/// The `body` is inspected only for the 400 case, to separate context -/// overflow (compact-and-retry) from an ordinary bad request (hard-fail). -/// -/// Caller is responsible for stashing the status code (and any `Retry-After`) -/// into a `ProviderDiagnostic`; this function only chooses the error name. -pub fn classifyHttpStatus(status: u16, body: []const u8) ProviderError { - return switch (status) { - 400 => if (isContextOverflowBody(body)) error.ContextOverflow else error.ProviderBadRequest, - 401, 403 => error.ProviderAuthFailed, - 404 => error.ProviderModelNotFound, - 408 => error.ProviderTransport, - 409, 425 => error.ProviderTransport, - 429 => error.ProviderRateLimited, - 503 => error.ProviderUnavailable, - 500, 502, 504 => error.ProviderServerError, - else => if (status >= 500) error.ProviderServerError else error.ProviderBadRequest, - }; -} - -/// Parse an HTTP `Retry-After` header value into milliseconds. Supports the -/// delta-seconds form (`"120"`); the HTTP-date form is not parsed and returns -/// null (the agent then falls back to its computed backoff). Returns null for -/// empty or unparseable values. -pub fn parseRetryAfterMs(value: []const u8) ?u64 { - const trimmed = std.mem.trim(u8, value, " \t\r\n"); - if (trimmed.len == 0) return null; - const secs = std.fmt.parseInt(u64, trimmed, 10) catch return null; - return secs *| std.time.ms_per_s; -} - -/// Find a `Retry-After` header (case-insensitive) in an HTTP response head -/// and parse it into milliseconds. `head` is a `std.http.Client.Response.Head` -/// (taken as `anytype` to avoid importing the http types here). Returns null -/// when absent or unparseable. -pub fn retryAfterFromHead(head: anytype) ?u64 { - var it = head.iterateHeaders(); - while (it.next()) |h| { - if (std.ascii.eqlIgnoreCase(h.name, "retry-after")) { - return parseRetryAfterMs(h.value); - } - } - return null; -} - -/// Details handed to the receiver before the agent sleeps for a provider -/// retry. Purely a UI/runtime event — never persisted to the session. -pub const ProviderRetryInfo = struct { - /// The attempt that just failed, 1-based (1 = the initial attempt). - attempt: usize, - /// Total attempts the policy will make, including the first. - max_attempts: usize, - /// How long the agent will sleep before the next attempt, in ms. - delay_ms: u64, - /// The classified error that triggered the retry. - err: anyerror, - /// HTTP status code, when known. - status_code: ?u16 = null, - /// Provider-sent `Retry-After`, when present, in ms. - retry_after_ms: ?u64 = null, - /// Provider diagnostic message, when known. - message: ?[]const u8 = null, - /// True when the retry is a context-overflow compaction attempt rather - /// than an ordinary backoff retry. Compaction retries report - /// `delay_ms == 0`. - compaction: bool = false, -}; - -/// A resumable provider streaming response: the pull-side projection of one -/// provider HTTP turn. It wraps the per-provider `ResumableResponse` (which -/// owns the pinned HTTP request/response, body reader, `SSEParser`, and -/// decode state) behind a tag so the agent loop can pump any provider -/// uniformly. -/// -/// `produce(out)` reads just enough bytes to append one or more `Event`s to -/// `out` (or reach response-complete), so the `Stream` drains the queue -/// before pumping again. On response completion the assistant message has -/// been committed to the conversation and a terminal `message_complete` -/// pushed. -/// -/// Tool-use identity (`id`, `name`) is delivered via a `tool_details` event, -/// pushed once per ToolUse block at the earliest moment both fields are -/// known. For Anthropic this is immediately after `block_start`, before any -/// deltas. For OpenAI Chat Completions this may be partway through the -/// arg-deltas, since the wire protocol can split `id` and `name` across -/// multiple streaming chunks. The guarantees: it fires strictly after the -/// block's `block_start`, strictly before its `block_complete`, and at most -/// once per ToolUse block. It never fires for non-ToolUse blocks. If a -/// tool_use block is dropped because identity never fully arrived, neither -/// `tool_details` nor `block_complete` fire for it. -/// -/// The terminal `message_complete`'s `usage` carries the wire-reported token -/// counts for the just-finished assistant message. It is `null` only when the -/// wire genuinely delivered no usage — chiefly OpenAI-compatible proxies -/// (OpenRouter, vLLM, some self-hosted backends) that ignore -/// `stream_options.include_usage`. Consumers that compute cost should record -/// the null case explicitly ("unknown") rather than treating it as zero. -pub const ProviderStream = struct { - ptr: *anyopaque, - vtable: *const VTable, - - pub const ProduceStatus = enum { more, response_complete }; - - pub const VTable = struct { - /// Pump the response, appending decoded events to `out`. Returns - /// `.more` (pump again) or `.response_complete` (the assistant - /// message is committed; a terminal `message_complete` was pushed). - produce: *const fn (*anyopaque, *EventQueue) anyerror!ProduceStatus, - /// Free the response and any owned state. - deinit: *const fn (*anyopaque) void, - /// Optional: after a failed `produce`, return the provider's - /// diagnostic message for the failure (e.g. an Anthropic - /// `overloaded_error` message), borrowed for the lifetime of the - /// response. Null when the provider has nothing to add beyond the - /// classified error name. - last_error: ?*const fn (*anyopaque) ?[]const u8 = null, - }; - - /// Pump the response, appending decoded events to `out`. Errors are - /// genuine failures (transport/parse/provider). - pub fn produce(self: ProviderStream, out: *EventQueue) anyerror!ProduceStatus { - return self.vtable.produce(self.ptr, out); - } - - pub fn deinit(self: ProviderStream) void { - self.vtable.deinit(self.ptr); - } - - /// The provider's diagnostic message for the most recent `produce` - /// failure, if any. Borrowed for the lifetime of the response. - pub fn lastError(self: ProviderStream) ?[]const u8 { - const f = self.vtable.last_error orelse return null; - return f(self.ptr); - } -}; - -/// Open one streaming provider turn against the active config snapshot, -/// returning a resumable `ProviderStream`. Performs the POST and reads -/// response headers (classifying any >=400 status into a provider error), -/// but does not pump the body — that happens lazily via -/// `ProviderStream.produce`. -/// -/// This is the single dispatch point: it switches on `cfg.provider`'s -/// `APIStyle` tag, builds a transient per-request object bound to the -/// process-global HTTP client, and opens it. There is no persistent provider -/// object — every turn re-reads `cfg`, so swapping the agent's -/// `*const Config` between turns changes provider, model, base_url, and the -/// visible tool set with no transport teardown. -/// -/// The tool registry is supplied by the caller (the `Agent` owns it now, -/// not `cfg`); the serializers receive it directly. On success the caller -/// owns the returned `ProviderStream` and must `deinit` it. -pub fn openStream( - allocator: std.mem.Allocator, - io: std.Io, - cfg: *const config_mod.Config, - registry: *const ToolRegistry, - conv: *conversation.Conversation, - diag: ?*ProviderDiagnostic, -) anyerror!ProviderStream { - // Imported lazily to break the circular module graph: - // provider.zig <- provider_openai_chat.zig <- provider.zig. - const provider_openai_chat = @import("provider_openai_chat.zig"); - const provider_anthropic_messages = @import("provider_anthropic_messages.zig"); - const provider_openai_responses = @import("provider_openai_responses.zig"); - const client = config_mod.httpClient(); - switch (cfg.provider) { - .openai_chat => |*c| { - var req: provider_openai_chat.OpenAIChatRequest = .{ - .allocator = allocator, - .io = io, - .config = c, - .http_client = client, - .diag = diag, - }; - const rr = try req.open(conv, registry); - return rr.providerStream(); - }, - .anthropic_messages => |*c| { - var req: provider_anthropic_messages.AnthropicMessagesRequest = .{ - .allocator = allocator, - .io = io, - .config = c, - .http_client = client, - .diag = diag, - }; - const rr = try req.open(conv, registry); - return rr.providerStream(); - }, - .openai_responses => |*c| { - var req: provider_openai_responses.OpenAIResponsesRequest = .{ - .allocator = allocator, - .io = io, - .config = c, - .http_client = client, - .diag = diag, - }; - const rr = try req.open(conv, registry); - return rr.providerStream(); - }, - .openai_codex_responses => |*c| { - var req: provider_openai_responses.OpenAIResponsesRequest = .{ - .allocator = allocator, - .io = io, - .config = c, - .dialect = .codex, - .http_client = client, - .diag = diag, - }; - const rr = try req.open(conv, registry); - return rr.providerStream(); - }, - } -} - -/// The shape of `openStream`, exposed as a function-pointer type so the agent -/// can carry an injectable seam (real dispatch in production, a stub in -/// tests) without resurrecting a per-provider vtable. -pub const OpenStreamFn = *const fn ( - allocator: std.mem.Allocator, - io: std.Io, - cfg: *const config_mod.Config, - registry: *const ToolRegistry, - conv: *conversation.Conversation, - diag: ?*ProviderDiagnostic, -) anyerror!ProviderStream; - -test "mergeHeaders - base first, extra appended, converted to http.Header" { - const t2 = std.testing; - const base = [_]std.http.Header{ - .{ .name = "content-type", .value = "application/json" }, - .{ .name = "authorization", .value = "Bearer x" }, - }; - const extra = [_]config_mod.Header{ - .{ .name = "Copilot-Integration-Id", .value = "vscode-chat" }, - .{ .name = "X-Initiator", .value = "user" }, - }; - const merged = try mergeHeaders(t2.allocator, &base, &extra); - defer t2.allocator.free(merged); - try t2.expectEqual(@as(usize, 4), merged.len); - try t2.expectEqualStrings("content-type", merged[0].name); - try t2.expectEqualStrings("Copilot-Integration-Id", merged[2].name); - try t2.expectEqualStrings("user", merged[3].value); -} - -test "mergeHeaders - empty extra yields a copy of base" { - const t2 = std.testing; - const base = [_]std.http.Header{.{ .name = "accept", .value = "text/event-stream" }}; - const merged = try mergeHeaders(t2.allocator, &base, &.{}); - defer t2.allocator.free(merged); - try t2.expectEqual(@as(usize, 1), merged.len); - try t2.expectEqualStrings("accept", merged[0].name); -} - -test "isContextOverflowBody - matches known markers, rejects others" { - const t2 = std.testing; - try t2.expect(isContextOverflowBody("{\"error\":{\"code\":\"context_length_exceeded\"}}")); - try t2.expect(isContextOverflowBody("This model's maximum context length is 8192 tokens")); - try t2.expect(isContextOverflowBody("prompt is too long: 250000 tokens > 200000 maximum")); - try t2.expect(!isContextOverflowBody("{\"error\":{\"code\":\"invalid_api_key\"}}")); - try t2.expect(!isContextOverflowBody("rate limit exceeded")); -} - -test "classifyHttpStatus - maps statuses to provider errors" { - const t2 = std.testing; - try t2.expectEqual(error.ContextOverflow, classifyHttpStatus(400, "prompt is too long")); - try t2.expectEqual(error.ProviderBadRequest, classifyHttpStatus(400, "bad json")); - try t2.expectEqual(error.ProviderAuthFailed, classifyHttpStatus(401, "")); - try t2.expectEqual(error.ProviderAuthFailed, classifyHttpStatus(403, "")); - try t2.expectEqual(error.ProviderModelNotFound, classifyHttpStatus(404, "")); - try t2.expectEqual(error.ProviderTransport, classifyHttpStatus(408, "")); - try t2.expectEqual(error.ProviderTransport, classifyHttpStatus(409, "")); - try t2.expectEqual(error.ProviderTransport, classifyHttpStatus(425, "")); - try t2.expectEqual(error.ProviderRateLimited, classifyHttpStatus(429, "")); - try t2.expectEqual(error.ProviderServerError, classifyHttpStatus(500, "")); - try t2.expectEqual(error.ProviderServerError, classifyHttpStatus(502, "")); - try t2.expectEqual(error.ProviderUnavailable, classifyHttpStatus(503, "")); - try t2.expectEqual(error.ProviderServerError, classifyHttpStatus(504, "")); - try t2.expectEqual(error.ProviderServerError, classifyHttpStatus(599, "")); - try t2.expectEqual(error.ProviderBadRequest, classifyHttpStatus(418, "")); -} - -test "isRetryableProviderError - retryable vs terminal" { - const t2 = std.testing; - try t2.expect(isRetryableProviderError(error.ProviderRateLimited)); - try t2.expect(isRetryableProviderError(error.ProviderUnavailable)); - try t2.expect(isRetryableProviderError(error.ProviderServerError)); - try t2.expect(isRetryableProviderError(error.ProviderTransport)); - try t2.expect(isRetryableProviderError(error.ProviderStreamMalformed)); - try t2.expect(!isRetryableProviderError(error.ProviderAuthFailed)); - try t2.expect(!isRetryableProviderError(error.ProviderBadRequest)); - try t2.expect(!isRetryableProviderError(error.ProviderModelNotFound)); - try t2.expect(!isRetryableProviderError(error.ContextOverflow)); - try t2.expect(!isRetryableProviderError(error.Canceled)); -} - -test "parseRetryAfterMs - delta-seconds and rejects" { - const t2 = std.testing; - try t2.expectEqual(@as(?u64, 120_000), parseRetryAfterMs("120")); - try t2.expectEqual(@as(?u64, 0), parseRetryAfterMs("0")); - try t2.expectEqual(@as(?u64, 2_000), parseRetryAfterMs(" 2 ")); - try t2.expectEqual(@as(?u64, null), parseRetryAfterMs("")); - // HTTP-date form is not parsed. - try t2.expectEqual(@as(?u64, null), parseRetryAfterMs("Wed, 21 Oct 2015 07:28:00 GMT")); -} |
