const std = @import("std"); 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"); pub const ToolRegistry = tool_registry_mod.ToolRegistry; pub const Usage = session_mod.Usage; 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` 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, => 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, }; /// Vtable for receiving streaming events from a Provider. /// /// The lifecycle callbacks (`onMessageStart` ... `onMessageComplete`) return /// `anyerror!void`. Returning an error aborts the in-flight turn: the Provider /// stops streaming, calls `onError(err)`, and propagates `err` out of /// `streamStep`. No partial assistant message is appended to the conversation. /// /// Tool-use identity (`id`, `name`) is delivered via `onToolDetails`, fired /// once per ToolUse block at the earliest moment both fields are known. For /// Anthropic this is immediately after `onBlockStart`, 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 only guarantees are: it fires strictly after the block's /// `onBlockStart`, strictly before its `onBlockComplete`, 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, `onToolDetails` /// (and `onBlockComplete`) never fire for it. /// /// `onError` is the receiver's cleanup hook. It fires exactly once per failed /// turn, whether the error originated in the receiver itself (a write failure) /// or in the Provider (HTTP/parse/stream failure). It is the last callback the /// receiver will see for that turn. `onError` itself cannot fail; receivers /// must swallow secondary failures during cleanup. /// /// `onMessageComplete`'s `usage` argument carries the wire-reported token /// counts for the just-finished assistant turn. Providers fire this exactly /// once per successful turn. `usage` is `null` only when the wire genuinely /// did not deliver any usage information — chiefly OpenAI-compatible proxies /// (OpenRouter, vLLM, some self-hosted backends) that ignore /// `stream_options.include_usage`. Receivers that compute cost should record /// the null case explicitly ("unknown") rather than treating it as zero. pub const ReceiverVTable = struct { onMessageStart: *const fn (*anyopaque, conversation.MessageRole) anyerror!void, onBlockStart: *const fn (*anyopaque, ContentBlockType, usize) anyerror!void, onToolDetails: *const fn (*anyopaque, usize, []const u8, []const u8) anyerror!void, onContentDelta: *const fn (*anyopaque, usize, []const u8) anyerror!void, onBlockComplete: *const fn (*anyopaque, usize, conversation.ContentBlock) anyerror!void, onMessageComplete: *const fn (*anyopaque, conversation.Message, ?Usage) anyerror!void, onError: *const fn (*anyopaque, anyerror) void, /// Fired by the agent after a provider attempt fails and before it /// sleeps for the next attempt. Purely informational: receivers surface /// it (e.g. a dim CLI status line) but cannot abort or alter the retry. /// Never fires for tool-call failures (those become tool results) nor /// for terminal provider errors. May be a no-op. onProviderRetry: *const fn (*anyopaque, ProviderRetryInfo) void, }; pub const Receiver = struct { ptr: *anyopaque, vtable: *const ReceiverVTable, pub fn onMessageStart(self: Receiver, role: conversation.MessageRole) !void { try self.vtable.onMessageStart(self.ptr, role); } pub fn onBlockStart(self: Receiver, block_type: ContentBlockType, index: usize) !void { try self.vtable.onBlockStart(self.ptr, block_type, index); } pub fn onToolDetails(self: Receiver, block_index: usize, id: []const u8, name: []const u8) !void { try self.vtable.onToolDetails(self.ptr, block_index, id, name); } pub fn onContentDelta(self: Receiver, block_index: usize, delta: []const u8) !void { try self.vtable.onContentDelta(self.ptr, block_index, delta); } pub fn onBlockComplete(self: Receiver, block_index: usize, block: conversation.ContentBlock) !void { try self.vtable.onBlockComplete(self.ptr, block_index, block); } pub fn onMessageComplete(self: Receiver, message: conversation.Message, usage: ?Usage) !void { try self.vtable.onMessageComplete(self.ptr, message, usage); } pub fn onError(self: Receiver, err: anyerror) void { self.vtable.onError(self.ptr, err); } pub fn onProviderRetry(self: Receiver, info: ProviderRetryInfo) void { self.vtable.onProviderRetry(self.ptr, info); } }; /// Drive one streaming provider turn against the active config snapshot. /// /// 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 runs 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 taken from `cfg.registry`; the serializers receive /// it directly. pub fn streamStep( allocator: std.mem.Allocator, io: std.Io, cfg: *const config_mod.Config, conv: *conversation.Conversation, receiver: *Receiver, diag: ?*ProviderDiagnostic, ) anyerror!void { // 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 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, }; return req.streamStep(conv, cfg.registry, receiver); }, .anthropic_messages => |*c| { var req: provider_anthropic_messages.AnthropicMessagesRequest = .{ .allocator = allocator, .io = io, .config = c, .http_client = client, .diag = diag, }; return req.streamStep(conv, cfg.registry, receiver); }, } } /// The shape of `streamStep`, 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 StreamFn = *const fn ( allocator: std.mem.Allocator, io: std.Io, cfg: *const config_mod.Config, conv: *conversation.Conversation, receiver: *Receiver, diag: ?*ProviderDiagnostic, ) anyerror!void; 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")); }