path: root/libpanto/src/provider.zig

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");
const stream_mod = @import("stream.zig");
pub const ToolRegistry = tool_registry_mod.ToolRegistry;
pub const Usage = session_mod.Usage;

const EventQueue = stream_mod.EventQueue;

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,
};

/// 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,
    };

    /// 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);
    }
};

/// 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 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();
        },
    }
}

/// 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 "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"));
}