From 7b3251e32d8b2e826e178d7544f07fa7af892b00 Mon Sep 17 00:00:00 2001 From: Joaquin Terrasa Date: Fri, 28 Aug 2026 18:58:43 +0200 Subject: [PATCH 1/7] feat(config): add anthropic provider id and anthropic_api_key credential source Extend ProviderId with .anthropic and CredentialSource with .anthropic_api_key, mirroring the openai/openai_api_key pair from the OpenAI-compatible transport. Credentials resolve from the ANTHROPIC_API_KEY environment variable or a profile-owned anthropic_api_key setting, never from repository config. Settings store gains the anthropic_model preference and config_runtime validates it alongside the existing per-provider model fields. --- src/core/auth/credentials.zig | 113 +++++++++++++++++++++++++++++ src/core/auth/provider_catalog.zig | 19 ++++- src/core/config/config_runtime.zig | 40 ++++++++++ src/core/config/model_provider.zig | 21 +++++- src/core/config/settings_store.zig | 13 ++++ src/core/shared/types.zig | 1 + 6 files changed, 201 insertions(+), 6 deletions(-) diff --git a/src/core/auth/credentials.zig b/src/core/auth/credentials.zig index 563a11c04..e974488af 100644 --- a/src/core/auth/credentials.zig +++ b/src/core/auth/credentials.zig @@ -43,6 +43,7 @@ pub const CatalogAuthenticatedSource = enum { fx_login, stored_key, chatgpt_subscription, + anthropic_api_key, grok_subscription, fn credentialSource(self: CatalogAuthenticatedSource) Source { @@ -52,6 +53,7 @@ pub const CatalogAuthenticatedSource = enum { .fx_login => .fx_login, .stored_key => .stored_key, .chatgpt_subscription => .chatgpt_subscription, + .anthropic_api_key => .anthropic_api_key, .grok_subscription => .grok_subscription, }; } @@ -167,6 +169,7 @@ pub fn catalogAccessForCredentialAndAccount( .ai_gateway_api_key => .ai_gateway_api_key, .stored_key => .stored_key, .chatgpt_subscription => .chatgpt_subscription, + .anthropic_api_key => .anthropic_api_key, .grok_subscription => .grok_subscription, .fx_login => blk: { const team = team_context orelse @@ -198,12 +201,35 @@ const FxLoginRefreshMode = enum { if_needed, force }; pub const missing_credential_message = "fx needs access to Vercel AI Gateway. Run fx login to sign in, fx setup to use an API key, or set AI_GATEWAY_API_KEY."; pub const missing_interactive_credential_message = "fx needs access to Vercel AI Gateway. Run /login to sign in, /setup to use an API key, or set AI_GATEWAY_API_KEY."; +pub const missing_anthropic_credential_message = "fx needs an Anthropic API key. Set ANTHROPIC_API_KEY, or anthropic_api_key in ~/.fx/settings.json."; +pub const missing_anthropic_interactive_credential_message = "fx needs an Anthropic API key. Set ANTHROPIC_API_KEY, or anthropic_api_key in ~/.fx/settings.json."; pub const missing_chatgpt_credential_message = "fx needs a Codex subscription login for this model. Run fx login codex."; pub const missing_chatgpt_interactive_credential_message = "Codex needs a subscription login. Run /login, open Connections, then choose Codex subscription."; pub const missing_grok_credential_message = "fx needs a Grok subscription login for this model. Run fx login grok."; pub const missing_grok_interactive_credential_message = "Grok needs a subscription login. Run /login, open Connections, then choose Grok subscription."; pub const unreadable_store_message = "fx could not read the stored API key from " ++ stored_key_backend_label ++ ". A key may be saved but unreadable. Set FX_TRACE_LOG for the failing step, or set AI_GATEWAY_API_KEY."; +pub fn missingCredentialMessage(provider: model_provider.ProviderId, interactive: bool) []const u8 { + return switch (provider) { + .codex => if (interactive) + missing_chatgpt_interactive_credential_message + else + missing_chatgpt_credential_message, + .grok => if (interactive) + missing_grok_interactive_credential_message + else + missing_grok_credential_message, + .anthropic => if (interactive) + missing_anthropic_interactive_credential_message + else + missing_anthropic_credential_message, + .gateway => if (interactive) + missing_interactive_credential_message + else + missing_credential_message, + }; +} + test "public credential guidance spells fx lowercase" { try std.testing.expect(std.mem.startsWith(u8, missing_credential_message, "fx needs")); try std.testing.expect(std.mem.startsWith(u8, missing_interactive_credential_message, "fx needs")); @@ -281,6 +307,7 @@ pub fn resolveForProvider( mode: LoadMode, provider: model_provider.ProviderId, preferred: ?Source, + profile_anthropic_api_key: ?[]const u8, ) !Resolution { switch (provider) { .codex => { @@ -298,6 +325,7 @@ pub fn resolveForProvider( return .{ .credential = credential }; }, .gateway => {}, + .anthropic => return .{ .credential = try loadAnthropicApiKeyCredential(alloc, profile_anthropic_api_key) }, } return resolvePreferring( alloc, @@ -403,6 +431,16 @@ pub fn loadSource( transport: oauth_transport.Provider, secret_store: host.SecretStore, source: Source, +) !?Credential { + return loadSourceWithProfile(alloc, transport, secret_store, source, null); +} + +pub fn loadSourceWithProfile( + alloc: std.mem.Allocator, + transport: oauth_transport.Provider, + secret_store: host.SecretStore, + source: Source, + profile_anthropic_api_key: ?[]const u8, ) !?Credential { return switch (source) { .vercel_oidc_token => loadEnvCredential(alloc, "VERCEL_OIDC_TOKEN", source), @@ -410,6 +448,7 @@ pub fn loadSource( .fx_login => loadFxLoginCredential(alloc, transport), .stored_key => loadStoredKeyCredential(alloc, secret_store), .chatgpt_subscription => loadChatGptCredential(alloc, transport, .if_needed), + .anthropic_api_key => loadAnthropicApiKeyCredential(alloc, profile_anthropic_api_key), .grok_subscription => loadGrokCredential(alloc, transport, .if_needed), }; } @@ -418,6 +457,15 @@ pub fn sourceExists( alloc: std.mem.Allocator, secret_store: host.SecretStore, source: Source, +) !bool { + return sourceExistsWithProfile(alloc, secret_store, source, null); +} + +pub fn sourceExistsWithProfile( + alloc: std.mem.Allocator, + secret_store: host.SecretStore, + source: Source, + profile_anthropic_api_key: ?[]const u8, ) !bool { return switch (source) { .vercel_oidc_token => nonEmptyEnvValue("VERCEL_OIDC_TOKEN") != null, @@ -435,6 +483,7 @@ pub fn sourceExists( break :blk true; }, .chatgpt_subscription => chatgpt_oauth.sourceExists(alloc), + .anthropic_api_key => anthropicApiKeyConfigured(profile_anthropic_api_key), .grok_subscription => grok_oauth.sourceExists(alloc), .stored_key => blk: { if (secret_store.isDisabled()) break :blk false; @@ -464,6 +513,33 @@ fn loadEnvCredential( }; } +pub fn anthropicApiKeyConfigured(profile_key: ?[]const u8) bool { + if (nonEmptyEnvValue("ANTHROPIC_API_KEY") != null) return true; + if (profile_key) |value| { + return std.mem.trim(u8, value, " \t\r\n").len > 0; + } + return false; +} + +fn loadAnthropicApiKeyCredential(alloc: std.mem.Allocator, profile_key: ?[]const u8) !?Credential { + if (nonEmptyEnvValue("ANTHROPIC_API_KEY")) |value| { + return .{ + .token = try alloc.dupe(u8, value), + .source = .anthropic_api_key, + }; + } + if (profile_key) |value| { + const trimmed = std.mem.trim(u8, value, " \t\r\n"); + if (trimmed.len > 0) { + return .{ + .token = try alloc.dupe(u8, trimmed), + .source = .anthropic_api_key, + }; + } + } + return null; +} + fn loadStoredKeyCredential( alloc: std.mem.Allocator, secret_store: host.SecretStore, @@ -661,6 +737,7 @@ pub fn sourceLabel(source: Source) []const u8 { .fx_login => "fx login", .stored_key => "stored API key (" ++ stored_key_backend_label ++ ")", .chatgpt_subscription => "Codex subscription", + .anthropic_api_key => "Anthropic API key", .grok_subscription => "Grok subscription", }; } @@ -1148,3 +1225,39 @@ test "a disabled store still reports why the fx login was silent" { try std.testing.expectEqual(FxLoginReadStatus.unavailable, resolution.fx_login_status); try std.testing.expectEqual(StoredKeyReadStatus.not_attempted, resolution.stored_key_status); } + +test "resolveForProvider anthropic never returns a gateway credential" { + const alloc = std.testing.allocator; + var env = try CredentialTestEnv.install(alloc, &.{.{ "ANTHROPIC_API_KEY", "sk-test-anthropic" }}); + defer env.deinit(); + + var resolution = try resolveForProvider( + alloc, + oauth_transport.unavailable_provider, + host.unavailable_secret_store, + .refresh_if_needed, + .anthropic, + null, + ); + defer if (resolution.credential) |*credential| credential.deinit(alloc); + + const credential = resolution.credential orelse return error.TestExpectedCredential; + try std.testing.expectEqual(Source.anthropic_api_key, credential.source); + try std.testing.expect(!model_provider.authorizesCredential(.gateway, credential.source)); +} + +test "sourceExistsWithProfile honors profile anthropic_api_key" { + const alloc = std.testing.allocator; + try std.testing.expect(try sourceExistsWithProfile( + alloc, + host.unavailable_secret_store, + .anthropic_api_key, + "profile-anthropic-key", + )); + try std.testing.expect(!(try sourceExistsWithProfile( + alloc, + host.unavailable_secret_store, + .anthropic_api_key, + null, + ))); +} diff --git a/src/core/auth/provider_catalog.zig b/src/core/auth/provider_catalog.zig index d949aa7ff..64fd25011 100644 --- a/src/core/auth/provider_catalog.zig +++ b/src/core/auth/provider_catalog.zig @@ -37,6 +37,14 @@ pub const entries = [_]Entry{ .description = "SuperGrok or X Premium subscription", .subscription = true, }, + .{ + .id = .anthropic, + .slug = "anthropic", + .name = "Anthropic", + .route_name = "Anthropic", + .description = "Anthropic API key via ANTHROPIC_API_KEY", + .subscription = false, + }, }; pub fn parse(value: []const u8) ?model_provider.ProviderId { @@ -52,18 +60,21 @@ pub fn find(id: model_provider.ProviderId) *const Entry { unreachable; } -pub fn label(id: model_provider.ProviderId) []const u8 { - return find(id).route_name; -} - test "auth provider catalog uses the model provider identity and explicit aliases" { try std.testing.expectEqual(model_provider.ProviderId.gateway, parse("vercel").?); try std.testing.expectEqual(model_provider.ProviderId.gateway, parse("gateway").?); try std.testing.expectEqual(model_provider.ProviderId.codex, parse("codex").?); try std.testing.expectEqual(model_provider.ProviderId.grok, parse("grok").?); + try std.testing.expectEqual(model_provider.ProviderId.anthropic, parse("anthropic").?); try std.testing.expect(parse("openai-codex") == null); try std.testing.expect(parse("chatgpt") == null); try std.testing.expect(parse("unknown") == null); try std.testing.expect(find(.codex).subscription); try std.testing.expect(find(.grok).subscription); + try std.testing.expect(!find(.anthropic).subscription); +} + +pub fn label(id: model_provider.ProviderId) []const u8 { + return find(id).route_name; } + diff --git a/src/core/config/config_runtime.zig b/src/core/config/config_runtime.zig index 0f9753637..b27f16235 100644 --- a/src/core/config/config_runtime.zig +++ b/src/core/config/config_runtime.zig @@ -39,6 +39,9 @@ pub const Paths = struct { pub const Settings = struct { models: model_preferences.Preferences = .{}, provider: ?model_provider.ProviderId = null, + codex_model: ?[]u8 = null, + anthropic_model: ?[]u8 = null, + grok_model: ?[]u8 = null, permission_mode: ?types.PermissionMode = null, credential_source: ?types.CredentialSource = null, yolo_acknowledged: ?bool = null, @@ -63,9 +66,12 @@ pub const Settings = struct { notification_max: ?bool = null, permission_rules: types.PermissionRuleSet = .{}, has_permission_rules: bool = false, + anthropic_api_key: ?[]u8 = null, pub fn deinit(self: *Settings, alloc: Allocator) void { self.models.deinit(alloc); + if (self.anthropic_model) |value| alloc.free(value); + if (self.anthropic_api_key) |value| alloc.free(value); self.permission_rules.deinit(alloc); self.* = .{}; } @@ -601,6 +607,7 @@ fn isProfileOnlySettingKey(key: []const u8) bool { "models", "provider", "codex_model", + "anthropic_model", "grok_model", "effort", "fast_mode", @@ -617,6 +624,7 @@ fn isProfileOnlySettingKey(key: []const u8) bool { "update_channel", "permission_mode", "credential_source", + "anthropic_api_key", "yolo_acknowledged", "permission", "additional_directories", @@ -1363,6 +1371,12 @@ fn parseProfileOnlyFields( try settings.models.putCopy(alloc, .codex, model_value.string); } + if (root.object.get("anthropic_model")) |model_value| { + if (model_value != .string) return error.InvalidAnthropicModelType; + settings_store.validateModel(model_value.string) catch return error.InvalidAnthropicModelValue; + settings.anthropic_model = try alloc.dupe(u8, model_value.string); + } + if (root.object.get("grok_model")) |model_value| { if (model_value != .string) return error.InvalidGrokModelType; settings_store.validateModel(model_value.string) catch return error.InvalidGrokModelValue; @@ -1392,6 +1406,12 @@ fn parseProfileOnlyFields( return error.InvalidCredentialSource; } + if (root.object.get("anthropic_api_key")) |anthropic_api_key_value| { + if (anthropic_api_key_value != .string) return error.InvalidAnthropicApiKeyType; + const trimmed = std.mem.trim(u8, anthropic_api_key_value.string, " \t\r\n"); + if (trimmed.len > 0) settings.anthropic_api_key = try alloc.dupe(u8, trimmed); + } + if (root.object.get("yolo_acknowledged")) |acknowledged_value| { if (acknowledged_value != .bool) return error.InvalidYoloAcknowledgedType; settings.yolo_acknowledged = acknowledged_value.bool; @@ -1537,8 +1557,28 @@ fn parseProjectSafeFields(settings: *Settings, root: std.json.Value) !void { fn mergeSettings(target: *Settings, incoming: *Settings, alloc: Allocator) void { target.models.mergeOwnedFrom(alloc, &incoming.models); if (incoming.provider) |value| target.provider = value; + if (incoming.codex_model) |value| { + if (target.codex_model) |current| alloc.free(current); + target.codex_model = value; + incoming.codex_model = null; + } + if (incoming.anthropic_model) |value| { + if (target.anthropic_model) |current| alloc.free(current); + target.anthropic_model = value; + incoming.anthropic_model = null; + } + if (incoming.grok_model) |value| { + if (target.grok_model) |current| alloc.free(current); + target.grok_model = value; + incoming.grok_model = null; + } if (incoming.permission_mode) |value| target.permission_mode = value; if (incoming.credential_source) |value| target.credential_source = value; + if (incoming.anthropic_api_key) |value| { + if (target.anthropic_api_key) |current| alloc.free(current); + target.anthropic_api_key = value; + incoming.anthropic_api_key = null; + } if (incoming.yolo_acknowledged) |value| target.yolo_acknowledged = value; if (incoming.max_agent_steps) |value| target.max_agent_steps = value; if (incoming.max_tool_result_bytes) |value| target.max_tool_result_bytes = value; diff --git a/src/core/config/model_provider.zig b/src/core/config/model_provider.zig index 93b0168a7..17a4e708d 100644 --- a/src/core/config/model_provider.zig +++ b/src/core/config/model_provider.zig @@ -5,6 +5,7 @@ pub const ProviderId = enum { gateway, codex, grok, + anthropic, }; pub const ProviderSelection = struct { @@ -16,15 +17,26 @@ pub fn parse(value: []const u8) ?ProviderId { if (std.ascii.eqlIgnoreCase(value, "gateway")) return .gateway; if (std.ascii.eqlIgnoreCase(value, "codex")) return .codex; if (std.ascii.eqlIgnoreCase(value, "grok")) return .grok; + if (std.ascii.eqlIgnoreCase(value, "anthropic")) return .anthropic; return null; } +pub fn label(provider: ProviderId) []const u8 { + return switch (provider) { + .gateway => "Vercel AI Gateway", + .codex => "Codex subscription", + .grok => "Grok subscription", + .anthropic => "Anthropic", + }; +} + pub fn authorizesCredential(provider: ProviderId, source: ?types.CredentialSource) bool { const selected = source orelse return false; return switch (provider) { - .gateway => selected != .chatgpt_subscription and selected != .grok_subscription, + .gateway => selected != .chatgpt_subscription and selected != .grok_subscription and selected != .anthropic_api_key, .codex => selected == .chatgpt_subscription, .grok => selected == .grok_subscription, + .anthropic => selected == .anthropic_api_key, }; } @@ -38,12 +50,17 @@ test "explicit providers authorize only their own credential origins" { try std.testing.expect(authorizesCredential(.grok, .grok_subscription)); try std.testing.expect(!authorizesCredential(.grok, .chatgpt_subscription)); try std.testing.expect(!authorizesCredential(.gateway, .grok_subscription)); + try std.testing.expect(authorizesCredential(.anthropic, .anthropic_api_key)); + try std.testing.expect(!authorizesCredential(.anthropic, .ai_gateway_api_key)); + try std.testing.expect(!authorizesCredential(.anthropic, .chatgpt_subscription)); + try std.testing.expect(!authorizesCredential(.gateway, .anthropic_api_key)); } -test "provider parsing exposes gateway codex and grok" { +test "provider parsing exposes gateway codex grok and anthropic" { try std.testing.expectEqual(ProviderId.gateway, parse("gateway").?); try std.testing.expectEqual(ProviderId.codex, parse("CODEX").?); try std.testing.expectEqual(ProviderId.grok, parse("GROK").?); + try std.testing.expectEqual(ProviderId.anthropic, parse("anthropic").?); try std.testing.expect(parse("openai-codex") == null); try std.testing.expect(parse("") == null); } diff --git a/src/core/config/settings_store.zig b/src/core/config/settings_store.zig index 5cb8e756a..e8c266969 100644 --- a/src/core/config/settings_store.zig +++ b/src/core/config/settings_store.zig @@ -93,6 +93,9 @@ pub const ProjectMcpMutation = struct { pub const UserSettingsPatch = struct { model_preference: ?ModelPreferencePatch = null, provider: ?model_provider.ProviderId = null, + codex_model: ?[]const u8 = null, + grok_model: ?[]const u8 = null, + anthropic_model: ?[]const u8 = null, permission_mode: ?types.PermissionMode = null, credential_source: ?types.CredentialSource = null, /// Removes the key entirely so resolution returns to plain precedence. @@ -114,6 +117,9 @@ pub const UserSettingsPatch = struct { fn isEmpty(self: UserSettingsPatch) bool { return self.model_preference == null and self.provider == null and + self.codex_model == null and + self.grok_model == null and + self.anthropic_model == null and self.permission_mode == null and self.credential_source == null and !self.clear_credential_source and @@ -986,6 +992,9 @@ fn applyUserPatchToRoot( application.changed = try putModelPreference(arena, &root.object, preference) or application.changed; } if (patch.provider) |value| application.changed = try putString(arena, &root.object, "provider", @tagName(value)) or application.changed; + if (patch.codex_model) |value| application.changed = try putString(arena, &root.object, "codex_model", value) or application.changed; + if (patch.grok_model) |value| application.changed = try putString(arena, &root.object, "grok_model", value) or application.changed; + if (patch.anthropic_model) |value| application.changed = try putString(arena, &root.object, "anthropic_model", value) or application.changed; if (patch.permission_mode) |value| application.changed = try putString(arena, &root.object, "permission_mode", @tagName(value)) or application.changed; if (patch.credential_source) |value| application.changed = try putString(arena, &root.object, "credential_source", @tagName(value)) or application.changed; if (patch.clear_credential_source and root.object.contains("credential_source")) { @@ -1814,6 +1823,10 @@ fn validateKnownSettingsObject( try validateModel(entry.value_ptr.string); } } + if (object.get("anthropic_model")) |value| { + if (value != .string) return error.InvalidSettingsFormat; + try validateModel(value.string); + } if (object.get("permission_mode")) |value| { if (value != .string or (!std.ascii.eqlIgnoreCase(value.string, "ask") and diff --git a/src/core/shared/types.zig b/src/core/shared/types.zig index 3901bc51d..4e90c8ad1 100644 --- a/src/core/shared/types.zig +++ b/src/core/shared/types.zig @@ -90,6 +90,7 @@ test "context notice body drops legacy markers from every line" { pub const CredentialSource = enum { vercel_oidc_token, ai_gateway_api_key, + anthropic_api_key, fx_login, stored_key, chatgpt_subscription, From 8f09f762d47e201321b2a62812c474be3f20a51d Mon Sep 17 00:00:00 2001 From: Joaquin Terrasa Date: Fri, 28 Aug 2026 18:58:51 +0200 Subject: [PATCH 2/7] feat(gateway): add Anthropic Messages stream transport and model catalog Implement the Anthropic Messages wire as a stream_provider.Provider: POST {base}/v1/messages with x-api-key and anthropic-version headers, top-level system hoisting, tool_use/tool_result blocks, and SSE parsing for message_start, content_block_start/delta/stop (text, thinking, and input_json deltas), message_delta, message_stop, and error events. Stop reasons map to the shared failure taxonomy (end_turn to stop, max_tokens to length, refusal to content_filter). Assistant messages carrying both text and tool calls serialize the tool_use blocks inside the still-open content array; closing it before the blocks produced invalid JSON on every agentic replay. Base URL defaults to https://api.anthropic.com and is overridable via FX_ANTHROPIC_BASE_URL, with FX_E2E_ANTHROPIC_URL reserved for loopback e2e fixtures. Boundaries match the existing transports (1MB SSE line, 64MB aggregate, 128 tool calls, 4MB tool arguments). anthropic_models.zig fetches /v1/models with the same caps as the grok catalog and exports the standard model_catalog_provider pair. Both files keep credential acceptance keyed by name so they compile independently of the enum-case commit. --- src/gateway/anthropic.zig | 991 +++++++++++++++++++++++++++++++ src/gateway/anthropic_models.zig | 374 ++++++++++++ 2 files changed, 1365 insertions(+) create mode 100644 src/gateway/anthropic.zig create mode 100644 src/gateway/anthropic_models.zig diff --git a/src/gateway/anthropic.zig b/src/gateway/anthropic.zig new file mode 100644 index 000000000..6dd37f4fb --- /dev/null +++ b/src/gateway/anthropic.zig @@ -0,0 +1,991 @@ +const std = @import("std"); +const secret = @import("../core/auth/secret.zig"); +const stream_provider = @import("../core/agent/stream_provider.zig"); +const io_mod = @import("../core/shared/io.zig"); +const types = @import("../core/shared/types.zig"); +const gateway_client = @import("client.zig"); + +const Allocator = std.mem.Allocator; +const default_base_url = "https://api.anthropic.com"; +pub const base_url_env = "FX_ANTHROPIC_BASE_URL"; +pub const anthropic_version = "2023-06-01"; +const default_max_tokens: u32 = 8192; +const max_thinking_budget_tokens: u32 = 32_000; +const min_thinking_budget_tokens: u32 = 1024; +const max_error_body_bytes: usize = 256 * 1024; +const max_sse_line_bytes: usize = 1024 * 1024; +const max_sse_aggregate_bytes: usize = 64 * 1024 * 1024; +const max_sse_events: usize = 100_000; +const max_tool_calls: usize = 128; +const max_tool_identity_bytes: usize = 1024; +const max_tool_arguments_bytes: usize = 4 * 1024 * 1024; +const transfer_buffer_bytes: usize = 256 * 1024; +const connect_timeout_ms: i64 = 30_000; +pub const e2e_endpoint_env = "FX_E2E_ANTHROPIC_URL"; + +pub const agent_stream_provider = stream_provider.Provider{ + .build_fn = buildRequest, + .stream_fn = streamCompletion, +}; + +pub fn resolveBaseUrl() []const u8 { + const raw = io_mod.getenv(base_url_env) orelse return default_base_url; + const trimmed = std.mem.trim(u8, raw, " \t\r\n"); + if (trimmed.len == 0) return default_base_url; + return trimmed; +} + +fn messagesUrl(alloc: Allocator) ![]u8 { + const base = std.mem.trimEnd(u8, resolveBaseUrl(), "/"); + if (std.mem.endsWith(u8, base, "/v1/messages")) return alloc.dupe(u8, base); + return std.fmt.allocPrint(alloc, "{s}/v1/messages", .{base}); +} + +fn validateModel(model: []const u8) !void { + if (model.len == 0 or model.len > 256) return error.InvalidAnthropicModel; + for (model) |byte| { + if (byte <= 0x20 or byte == 0x7f) return error.InvalidAnthropicModel; + } +} + +fn buildRequest( + _: ?*anyopaque, + alloc: Allocator, + request: stream_provider.BuildRequest, +) ![]u8 { + try validateModel(request.model); + if (request.budget) |budget| { + if (budget.cancel_flag) |flag| if (flag.load(.seq_cst)) return error.Cancelled; + _ = budget.deadline; + } + + var system: std.Io.Writer.Allocating = .init(alloc); + defer system.deinit(); + for (request.messages) |message| { + if (message.role != .system) continue; + const text = message.content orelse continue; + if (text.len == 0) continue; + if (system.written().len > 0) try system.writer.writeAll("\n\n"); + try system.writer.writeAll(text); + } + + const max_tokens: u32 = request.max_output_tokens orelse default_max_tokens; + + var out: std.Io.Writer.Allocating = .init(alloc); + errdefer out.deinit(); + const writer = &out.writer; + try writer.writeAll("{\"model\":"); + try std.json.Stringify.value(request.model, .{}, writer); + try writer.print(",\"max_tokens\":{d}", .{max_tokens}); + if (system.written().len > 0) { + try writer.writeAll(",\"system\":"); + try std.json.Stringify.value(system.written(), .{}, writer); + } + try writer.writeAll(",\"stream\":true,\"messages\":["); + try writeMessages(writer, request.messages); + try writer.writeByte(']'); + const tool_count = try writeTools(writer, alloc, request.serialized_tools, request.selected_dynamic_tool_schemas); + if (tool_count > 0) { + try writer.writeAll(",\"tool_choice\":"); + try writeToolChoice(writer, request.tool_choice); + } + if (request.provider_options.reasoning) |effort| { + if (effort != .auto) { + const budget = @min( + max_thinking_budget_tokens, + @max(min_thinking_budget_tokens, max_tokens / 2), + ); + try writer.print(",\"thinking\":{{\"type\":\"enabled\",\"budget_tokens\":{d}}}", .{budget}); + } + } + try writer.writeByte('}'); + return out.toOwnedSlice(); +} + +fn writeMessages(writer: *std.Io.Writer, messages: []const types.ChatMessage) !void { + var first = true; + for (messages) |message| { + switch (message.role) { + .system => continue, + .user => { + if (message.tool_call_id != null) { + try writeComma(writer, &first); + try writeToolResult(writer, message); + continue; + } + try writeComma(writer, &first); + try writer.writeAll("{\"role\":\"user\",\"content\":["); + if (message.content) |content| if (content.len > 0) { + try writer.writeAll("{\"type\":\"text\",\"text\":"); + try std.json.Stringify.value(content, .{}, writer); + try writer.writeByte('}'); + }; + try writer.writeAll("]}"); + }, + .assistant => { + var wrote_any = false; + if (message.content) |content| if (content.len > 0) { + try writeComma(writer, &first); + try writer.writeAll("{\"role\":\"assistant\",\"content\":[{\"type\":\"text\",\"text\":"); + try std.json.Stringify.value(content, .{}, writer); + try writer.writeByte('}'); + wrote_any = true; + }; + for (message.tool_calls) |call| { + if (call.id.len == 0 or call.id.len > max_tool_identity_bytes or + call.name.len == 0 or call.name.len > max_tool_identity_bytes or + call.arguments_json.len > max_tool_arguments_bytes) + { + return error.AnthropicToolCallLimitExceeded; + } + if (wrote_any) { + try writer.writeByte(','); + } else { + try writeComma(writer, &first); + try writer.writeAll("{\"role\":\"assistant\",\"content\":["); + wrote_any = true; + } + try writer.writeAll("{\"type\":\"tool_use\",\"id\":"); + try std.json.Stringify.value(call.id, .{}, writer); + try writer.writeAll(",\"name\":"); + try std.json.Stringify.value(call.name, .{}, writer); + try writer.writeAll(",\"input\":"); + if (call.arguments_json.len > 0) { + try writer.writeAll(call.arguments_json); + } else { + try writer.writeAll("{}"); + } + try writer.writeByte('}'); + } + if (wrote_any) try writer.writeAll("]}"); + }, + .tool => { + try writeComma(writer, &first); + try writeToolResult(writer, message); + }, + } + } +} + +fn writeToolResult(writer: *std.Io.Writer, message: types.ChatMessage) !void { + try writer.writeAll("{\"role\":\"user\",\"content\":[{\"type\":\"tool_result\",\"tool_use_id\":"); + try std.json.Stringify.value(message.tool_call_id orelse "", .{}, writer); + try writer.writeAll(",\"content\":"); + try std.json.Stringify.value(message.content orelse "", .{}, writer); + try writer.writeAll("}]}"); +} + +fn writeTools( + writer: *std.Io.Writer, + alloc: Allocator, + serialized_tools: []const u8, + selected_dynamic_schemas: []const []const u8, +) !usize { + var count: usize = 0; + var parsed = std.json.parseFromSlice(std.json.Value, alloc, serialized_tools, .{}) catch |err| switch (err) { + error.OutOfMemory => return error.OutOfMemory, + else => return error.InvalidToolSchema, + }; + defer parsed.deinit(); + if (parsed.value != .array) return error.InvalidToolSchema; + + var tools_out: std.Io.Writer.Allocating = .init(alloc); + defer tools_out.deinit(); + try tools_out.writer.writeAll(",\"tools\":["); + for (parsed.value.array.items) |tool| { + if (try writeFunctionTool(&tools_out.writer, tool, count != 0)) count += 1; + } + for (selected_dynamic_schemas) |schema_json| { + var selected = std.json.parseFromSlice(std.json.Value, alloc, schema_json, .{}) catch |err| switch (err) { + error.OutOfMemory => return error.OutOfMemory, + else => return error.InvalidToolSchema, + }; + defer selected.deinit(); + if (try writeFunctionTool(&tools_out.writer, selected.value, count != 0)) count += 1; + } + try tools_out.writer.writeByte(']'); + if (count > 0) try writer.writeAll(tools_out.written()); + return count; +} + +fn writeFunctionTool(writer: *std.Io.Writer, value: std.json.Value, comma: bool) !bool { + if (value != .object) return false; + const kind = value.object.get("type") orelse return false; + if (kind != .string or !std.mem.eql(u8, kind.string, "function")) return false; + const name = value.object.get("name") orelse return false; + if (name != .string or name.string.len == 0) return false; + const parameters = value.object.get("inputSchema") orelse value.object.get("parameters") orelse return false; + if (parameters != .object) return false; + if (comma) try writer.writeByte(','); + try writer.writeAll("{\"name\":"); + try std.json.Stringify.value(name.string, .{}, writer); + if (value.object.get("description")) |description| if (description == .string) { + try writer.writeAll(",\"description\":"); + try std.json.Stringify.value(description.string, .{}, writer); + }; + try writer.writeAll(",\"input_schema\":"); + try std.json.Stringify.value(parameters, .{}, writer); + try writer.writeByte('}'); + return true; +} + +fn writeToolChoice(writer: *std.Io.Writer, choice: types.ToolChoice) !void { + switch (choice) { + .auto => try writer.writeAll("{\"type\":\"auto\"}"), + .none => try writer.writeAll("{\"type\":\"none\"}"), + .required => try writer.writeAll("{\"type\":\"any\"}"), + } +} + +fn writeComma(writer: *std.Io.Writer, first: *bool) !void { + if (!first.*) try writer.writeByte(','); + first.* = false; +} + +fn streamCompletion( + _: ?*anyopaque, + alloc: Allocator, + request: stream_provider.Request, +) !stream_provider.Result { + var result = streamCompletionCore(alloc, request) catch |err| { + if (request.cancel_flag.load(.seq_cst)) return error.Cancelled; + if (requestDeadlineExpired(request)) return error.Timeout; + request.attempt_evidence.network_failure = gateway_client.networkFailureEvidence(err, request.delivery.load()); + return err; + }; + if (requestDeadlineExpired(request)) { + result.deinit(alloc); + return error.Timeout; + } + return result; +} + +fn requestDeadlineExpired(request: stream_provider.Request) bool { + const deadline = request.deadline orelse return false; + const now = std.Io.Clock.Timestamp.now(io_mod.getIo(), .awake); + return !std.Io.Clock.Timestamp.compare(now, .lt, deadline); +} + +const OpenedRequest = struct { + request: ?std.http.Client.Request, + + pub fn deinit(self: *OpenedRequest, _: Allocator) void { + if (self.request) |*request| request.deinit(); + self.request = null; + } + + pub fn take(self: *OpenedRequest) std.http.Client.Request { + const request = self.request.?; + self.request = null; + return request; + } +}; + +const OpenRequestOperation = struct { + client: *std.http.Client, + uri: std.Uri, + api_key: []const u8, + extra_headers: []const std.http.Header, + + pub fn run(self: *@This()) !OpenedRequest { + return .{ .request = try self.client.request(.POST, self.uri, .{ + .headers = .{ + .content_type = .{ .override = "application/json" }, + .authorization = .omit, + .accept_encoding = .omit, + .user_agent = .{ .override = gateway_client.user_agent }, + }, + .extra_headers = self.extra_headers, + .keep_alive = false, + .redirect_behavior = .unhandled, + }) }; + } +}; + +fn streamCompletionCore(alloc: Allocator, request: stream_provider.Request) !stream_provider.Result { + if (request.cancel_flag.load(.seq_cst)) return error.Cancelled; + if (request.credential_source) |source| { + // Referenced by name so this module compiles before the enum case lands. + const anthropic_api_key = std.meta.stringToEnum(types.CredentialSource, "anthropic_api_key"); + if (anthropic_api_key == null or source != anthropic_api_key.?) { + return error.AnthropicApiKeyCredentialRequired; + } + } + try validateModel(request.model); + const request_endpoint = if (io_mod.getenv(e2e_endpoint_env)) |override| endpoint: { + if (!gateway_client.isLoopbackHttpUrl(override)) return error.InvalidE2EAnthropicEndpoint; + break :endpoint override; + } else try messagesUrl(alloc); + defer if (io_mod.getenv(e2e_endpoint_env) == null) alloc.free(@constCast(request_endpoint)); + const uri = try std.Uri.parse(request_endpoint); + + var extra_headers_buf: [4]std.http.Header = undefined; + var extra_count: usize = 0; + extra_headers_buf[extra_count] = .{ .name = "accept", .value = "text/event-stream" }; + extra_count += 1; + extra_headers_buf[extra_count] = .{ .name = "x-api-key", .value = request.api_key }; + extra_count += 1; + extra_headers_buf[extra_count] = .{ .name = "anthropic-version", .value = anthropic_version }; + extra_count += 1; + + var client: std.http.Client = .{ .allocator = alloc, .io = io_mod.getIo() }; + defer client.deinit(); + var open_operation = OpenRequestOperation{ + .client = &client, + .uri = uri, + .api_key = request.api_key, + .extra_headers = extra_headers_buf[0..extra_count], + }; + var connect_deadline = std.Io.Clock.Timestamp.fromNow(io_mod.getIo(), .{ + .clock = .awake, + .raw = .fromMilliseconds(connect_timeout_ms), + }); + if (request.deadline) |deadline| { + if (std.Io.Clock.Timestamp.compare(deadline, .lt, connect_deadline)) { + connect_deadline = deadline; + } + } + var opened = try gateway_client.runBoundedHttpOperation( + OpenedRequest, + alloc, + request.cancel_flag, + connect_deadline, + &open_operation, + ); + var http_request = opened.take(); + defer http_request.deinit(); + var cancel_watch_done = std.atomic.Value(bool).init(false); + const cancel_watcher = if (http_request.connection) |connection| + if (request.deadline) |deadline| + try gateway_client.spawnHttpCancelWatcherBounded( + &cancel_watch_done, + request.cancel_flag, + deadline, + connection.stream_writer.stream, + ) + else + try gateway_client.spawnHttpCancelWatcher( + &cancel_watch_done, + request.cancel_flag, + connection.stream_writer.stream, + ) + else + null; + defer { + cancel_watch_done.store(true, .seq_cst); + if (cancel_watcher) |thread| thread.join(); + } + if (request.cancel_flag.load(.seq_cst)) return error.Cancelled; + + http_request.transfer_encoding = .{ .content_length = request.payload.len }; + var send_buffer: [8192]u8 = undefined; + request.delivery.markPossiblySent(); + var body_writer = try http_request.sendBodyUnflushed(&send_buffer); + try body_writer.writer.writeAll(request.payload); + try body_writer.end(); + if (http_request.connection) |connection| try connection.flush(); + if (request.cancel_flag.load(.seq_cst)) return error.Cancelled; + + var response = try http_request.receiveHead(&.{}); + if (response.head.status != .ok) { + var transfer: [16 * 1024]u8 = undefined; + const reader = response.reader(&transfer); + const bounded_body = reader.allocRemaining(alloc, .limited(max_error_body_bytes + 1)) catch |err| switch (err) { + error.StreamTooLong => try alloc.dupe(u8, "Anthropic error response exceeded the local limit"), + else => return err, + }; + const body = if (bounded_body.len > max_error_body_bytes) body: { + alloc.free(bounded_body); + break :body try alloc.dupe(u8, "Anthropic error response exceeded the local limit"); + } else bounded_body; + return .{ + .status = response.head.status, + .err_body = body, + .ownership = .owned, + }; + } + + var transfer_buffer: [transfer_buffer_bytes]u8 = undefined; + const reader = response.reader(&transfer_buffer); + const completion = try consumeSse( + alloc, + reader, + request.callback_ctx, + request.on_content_chunk, + request.on_tool_start, + request.on_reasoning_chunk, + request.on_tool_input_chunk, + request.cancel_flag, + request.content_capture_limit, + ); + return .{ + .status = .ok, + .completion = completion, + .ownership = .owned, + }; +} + +const ToolAccumulator = struct { + block_index: i64, + id: []u8, + name: []u8, + arguments: std.ArrayList(u8) = .empty, + + fn deinit(self: *ToolAccumulator, alloc: Allocator) void { + alloc.free(self.id); + alloc.free(self.name); + self.arguments.deinit(alloc); + self.* = undefined; + } +}; + +const SseReader = struct { + pending_line: std.ArrayList(u8) = .empty, + aggregate_bytes: usize = 0, + + const Line = struct { + bytes: []const u8, + wire_bytes: usize, + }; + + fn deinit(self: *SseReader, alloc: Allocator) void { + self.pending_line.deinit(alloc); + } + + fn release(self: *SseReader) void { + self.pending_line.clearRetainingCapacity(); + } + + /// Returns the next `data:` payload, or null at stream end. + fn next(self: *SseReader, alloc: Allocator, reader: anytype) !?[]const u8 { + while (true) { + const line = try self.readLine(alloc, reader) orelse return null; + self.aggregate_bytes = try checkedAccumulatedSize( + self.aggregate_bytes, + line.wire_bytes, + max_sse_aggregate_bytes, + ); + const trimmed = std.mem.trim(u8, line.bytes, " \t\r"); + if (trimmed.len == 0 or trimmed[0] == ':') { + self.release(); + continue; + } + if (!std.mem.startsWith(u8, trimmed, "data:")) { + self.release(); + continue; + } + const data = std.mem.trim(u8, trimmed["data:".len..], " \t"); + if (data.len == 0) { + self.release(); + continue; + } + return data; + } + } + + fn readLine(self: *SseReader, alloc: Allocator, reader: anytype) !?Line { + while (true) { + const fragment = reader.takeDelimiter('\n') catch |err| switch (err) { + error.StreamTooLong => { + const buffered = reader.buffered(); + if (buffered.len == 0) return error.AnthropicSseReadStalled; + if (buffered.len > max_sse_line_bytes - self.pending_line.items.len) { + return error.AnthropicSseEventTooLarge; + } + try self.pending_line.appendSlice(alloc, buffered); + reader.tossBuffered(); + continue; + }, + error.ReadFailed => return error.ReadFailed, + } orelse { + if (self.pending_line.items.len > 0) { + return .{ + .bytes = self.pending_line.items, + .wire_bytes = self.pending_line.items.len, + }; + } + return null; + }; + if (fragment.len > max_sse_line_bytes - self.pending_line.items.len) { + return error.AnthropicSseEventTooLarge; + } + if (self.pending_line.items.len == 0) { + return .{ + .bytes = fragment, + .wire_bytes = fragment.len + 1, + }; + } + try self.pending_line.appendSlice(alloc, fragment); + return .{ + .bytes = self.pending_line.items, + .wire_bytes = self.pending_line.items.len + 1, + }; + } + } +}; + +fn consumeSse( + alloc: Allocator, + reader: anytype, + callback_ctx: *anyopaque, + on_content_chunk: stream_provider.StreamCallback, + on_tool_start: ?stream_provider.ToolStartCallback, + on_reasoning_chunk: ?stream_provider.StreamCallback, + on_tool_input_chunk: ?stream_provider.StreamCallback, + cancel_flag: *std.atomic.Value(bool), + content_capture_limit: ?usize, +) !types.GatewayCompletion { + var content: std.ArrayList(u8) = .empty; + errdefer content.deinit(alloc); + var tools: std.ArrayList(ToolAccumulator) = .empty; + defer { + for (tools.items) |*tool| tool.deinit(alloc); + tools.deinit(alloc); + } + var sse: SseReader = .{}; + defer sse.deinit(alloc); + var finish_reason: ?types.ProviderFinishReason = null; + var usage: types.Usage = .{}; + var generation_id: ?[]u8 = null; + errdefer if (generation_id) |id| alloc.free(id); + var message_started = false; + var message_stopped = false; + var saw_error = false; + + while (try sse.next(alloc, reader)) |json_text| { + defer sse.release(); + if (cancel_flag.load(.seq_cst)) return error.Cancelled; + _ = try checkedAccumulatedSize(0, 1, max_sse_events); + var parsed = std.json.parseFromSlice(std.json.Value, alloc, json_text, .{}) catch + return error.InvalidAnthropicSseEvent; + defer parsed.deinit(); + if (parsed.value != .object) continue; + const object = parsed.value.object; + const event_type = stringField(object, "type") orelse continue; + + if (std.mem.eql(u8, event_type, "message_start")) { + message_started = true; + if (object.get("message")) |message| if (message == .object) { + if (stringField(message.object, "id")) |id| { + generation_id = try alloc.dupe(u8, id); + } + if (message.object.get("usage")) |usage_value| { + if (usage_value == .object) { + usage.input_tokens = unsignedField(usage_value.object, "input_tokens"); + } + } + }; + } else if (std.mem.eql(u8, event_type, "content_block_start")) { + const index = integerField(object, "index") orelse continue; + const block = object.get("content_block") orelse continue; + if (block != .object) continue; + const block_type = stringField(block.object, "type") orelse continue; + if (std.mem.eql(u8, block_type, "tool_use")) { + const call_id = stringField(block.object, "id") orelse continue; + const name = stringField(block.object, "name") orelse continue; + if (findTool(tools.items, index) == null) { + try appendTool(alloc, &tools, index, call_id, name); + if (on_tool_start) |callback| callback(callback_ctx, call_id, name, null); + } + } + } else if (std.mem.eql(u8, event_type, "content_block_delta")) { + const index = integerField(object, "index") orelse continue; + const delta = object.get("delta") orelse continue; + if (delta != .object) continue; + const delta_type = stringField(delta.object, "type") orelse continue; + if (std.mem.eql(u8, delta_type, "text_delta")) { + const text = stringField(delta.object, "text") orelse continue; + on_content_chunk(callback_ctx, text); + try appendCaptured(alloc, &content, text, content_capture_limit); + } else if (std.mem.eql(u8, delta_type, "thinking_delta")) { + const text = stringField(delta.object, "thinking") orelse continue; + if (on_reasoning_chunk) |callback| callback(callback_ctx, text); + } else if (std.mem.eql(u8, delta_type, "input_json_delta")) { + const partial = stringField(delta.object, "partial_json") orelse continue; + const tool_index = findTool(tools.items, index) orelse continue; + try appendToolArguments(alloc, &tools.items[tool_index].arguments, partial); + if (on_tool_input_chunk) |callback| callback(callback_ctx, partial); + } + } else if (std.mem.eql(u8, event_type, "content_block_stop") or + std.mem.eql(u8, event_type, "ping")) + { + // No state to finalize per block; tool arguments accumulate by index. + } else if (std.mem.eql(u8, event_type, "message_delta")) { + if (object.get("delta")) |delta| if (delta == .object) { + if (stringField(delta.object, "stop_reason")) |reason| { + finish_reason = stopReason(reason, tools.items.len > 0); + } + }; + if (object.get("usage")) |usage_value| if (usage_value == .object) { + usage.output_tokens = unsignedField(usage_value.object, "output_tokens"); + }; + } else if (std.mem.eql(u8, event_type, "message_stop")) { + message_stopped = true; + break; + } else if (std.mem.eql(u8, event_type, "error")) { + saw_error = true; + break; + } + } + if (cancel_flag.load(.seq_cst)) return error.Cancelled; + if (saw_error) return error.AnthropicResponseFailed; + if (!message_started or !message_stopped) return error.AnthropicStreamIncomplete; + + const owned_content = if (content.items.len > 0) try content.toOwnedSlice(alloc) else null; + if (owned_content != null) content = .empty; + errdefer if (owned_content) |value| alloc.free(value); + const owned_tools: []types.ToolCall = if (tools.items.len > 0) + try alloc.alloc(types.ToolCall, tools.items.len) + else + &.{}; + errdefer if (owned_tools.len > 0) alloc.free(owned_tools); + var initialized: usize = 0; + errdefer for (owned_tools[0..initialized]) |call| { + alloc.free(call.id); + alloc.free(call.name); + alloc.free(call.arguments_json); + }; + for (tools.items, 0..) |*tool, index| { + const arguments = if (tool.arguments.items.len > 0) + try tool.arguments.toOwnedSlice(alloc) + else + try alloc.dupe(u8, "{}"); + tool.arguments = .empty; + owned_tools[index] = .{ + .id = tool.id, + .name = tool.name, + .arguments_json = arguments, + }; + tool.id = &.{}; + tool.name = &.{}; + initialized += 1; + } + return .{ + .content = owned_content, + .tool_calls = owned_tools, + .generation_id = generation_id, + .finish_reason = finish_reason orelse if (owned_tools.len > 0) .tool_calls else .stop, + .usage = usage, + }; +} + +fn stopReason(raw: []const u8, has_tools: bool) types.ProviderFinishReason { + if (std.mem.eql(u8, raw, "end_turn") or std.mem.eql(u8, raw, "stop_sequence")) { + return if (has_tools) .tool_calls else .stop; + } + if (std.mem.eql(u8, raw, "max_tokens")) return .length; + if (std.mem.eql(u8, raw, "refusal")) return .content_filter; + if (std.mem.eql(u8, raw, "pause_turn")) return .other; + return if (has_tools) .tool_calls else .stop; +} + +fn appendTool( + alloc: Allocator, + tools: *std.ArrayList(ToolAccumulator), + block_index: i64, + call_id: []const u8, + name: []const u8, +) !void { + if (tools.items.len >= max_tool_calls or call_id.len == 0 or call_id.len > max_tool_identity_bytes or + name.len == 0 or name.len > max_tool_identity_bytes) + { + return error.AnthropicToolCallLimitExceeded; + } + const id = try alloc.dupe(u8, call_id); + errdefer alloc.free(id); + const owned_name = try alloc.dupe(u8, name); + errdefer alloc.free(owned_name); + try tools.append(alloc, .{ + .block_index = block_index, + .id = id, + .name = owned_name, + }); +} + +fn appendToolArguments( + alloc: Allocator, + arguments: *std.ArrayList(u8), + delta: []const u8, +) !void { + _ = checkedAccumulatedSize(arguments.items.len, delta.len, max_tool_arguments_bytes) catch + return error.AnthropicToolArgumentsTooLarge; + try arguments.appendSlice(alloc, delta); +} + +fn checkedAccumulatedSize(current: usize, additional: usize, maximum: usize) !usize { + const next = std.math.add(usize, current, additional) catch + return error.AnthropicResourceLimitExceeded; + if (next > maximum) return error.AnthropicResourceLimitExceeded; + return next; +} + +fn appendCaptured( + alloc: Allocator, + content: *std.ArrayList(u8), + delta: []const u8, + limit: ?usize, +) !void { + const remaining = if (limit) |maximum| maximum -| @min(maximum, content.items.len) else delta.len; + try content.appendSlice(alloc, delta[0..@min(delta.len, remaining)]); +} + +fn findTool(tools: []const ToolAccumulator, block_index: i64) ?usize { + for (tools, 0..) |tool, index| if (tool.block_index == block_index) return index; + return null; +} + +fn stringField(object: std.json.ObjectMap, key: []const u8) ?[]const u8 { + const value = object.get(key) orelse return null; + if (value != .string) return null; + return value.string; +} + +fn integerField(object: std.json.ObjectMap, key: []const u8) ?i64 { + const value = object.get(key) orelse return null; + if (value != .integer) return null; + return value.integer; +} + +fn unsignedField(object: std.json.ObjectMap, key: []const u8) ?u64 { + const value = integerField(object, key) orelse return null; + if (value < 0) return null; + return @intCast(value); +} + +test "Anthropic request hoists system text and maps messages and tools" { + const messages = [_]types.ChatMessage{ + .{ .role = .system, .content = "Be concise." }, + .{ .role = .user, .content = "Read it." }, + .{ + .role = .assistant, + .tool_calls = &.{.{ .id = "call_1", .name = "read_file", .arguments_json = "{\"path\":\"README.md\"}" }}, + }, + .{ .role = .tool, .tool_call_id = "call_1", .tool_name = "read_file", .content = "contents" }, + }; + const body = try agent_stream_provider.build(std.testing.allocator, .{ + .model = "claude-sonnet-4-5", + .serialized_tools = "[{\"type\":\"function\",\"name\":\"read_file\",\"description\":\"Read\",\"inputSchema\":{\"type\":\"object\"}}]", + .messages = &messages, + .tool_choice = .auto, + .provider_options = .{}, + .max_output_tokens = 4096, + }); + defer std.testing.allocator.free(body); + + try std.testing.expect(std.mem.find(u8, body, "\"model\":\"claude-sonnet-4-5\"") != null); + try std.testing.expect(std.mem.find(u8, body, "\"max_tokens\":4096") != null); + try std.testing.expect(std.mem.indexOf(u8, body, "\"system\":\"Be concise.\"") != null); + try std.testing.expect(std.mem.indexOf(u8, body, "\"stream\":true") != null); + try std.testing.expect(std.mem.indexOf(u8, body, "{\"type\":\"tool_use\",\"id\":\"call_1\",\"name\":\"read_file\",\"input\":{\"path\":\"README.md\"}}") != null); + try std.testing.expect(std.mem.indexOf(u8, body, "{\"type\":\"tool_result\",\"tool_use_id\":\"call_1\",\"content\":\"contents\"}") != null); + try std.testing.expect(std.mem.indexOf(u8, body, "\"input_schema\":{\"type\":\"object\"}") != null); + try std.testing.expect(std.mem.indexOf(u8, body, "\"tool_choice\":{\"type\":\"auto\"}") != null); +} + +test "Anthropic request applies default max tokens and maps tool choice variants" { + const messages = [_]types.ChatMessage{.{ .role = .user, .content = "Hello." }}; + const body = try agent_stream_provider.build(std.testing.allocator, .{ + .model = "claude-sonnet-4-5", + .serialized_tools = "[]", + .messages = &messages, + .tool_choice = .required, + .provider_options = .{}, + }); + defer std.testing.allocator.free(body); + + try std.testing.expect(std.mem.indexOf(u8, body, "\"max_tokens\":8192") != null); + try std.testing.expect(std.mem.indexOf(u8, body, "\"system\"") == null); + try std.testing.expect(std.mem.indexOf(u8, body, "\"tools\"") == null); + try std.testing.expect(std.mem.indexOf(u8, body, "\"tool_choice\"") == null); + + const none_body = try agent_stream_provider.build(std.testing.allocator, .{ + .model = "claude-sonnet-4-5", + .serialized_tools = "[]", + .messages = &messages, + .tool_choice = .none, + .provider_options = .{}, + }); + defer std.testing.allocator.free(none_body); + try std.testing.expect(std.mem.indexOf(u8, none_body, "\"tool_choice\"") == null); +} + +test "Anthropic SSE maps text tool arguments stop reason and usage" { + const sse_text = + "event: message_start\n" ++ + "data: {\"type\":\"message_start\",\"message\":{\"id\":\"msg_1\",\"usage\":{\"input_tokens\":10}}}\n\n" ++ + "event: ping\n" ++ + "data: {\"type\":\"ping\"}\n\n" ++ + "event: content_block_start\n" ++ + "data: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"text\"}}\n\n" ++ + "event: content_block_delta\n" ++ + "data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"hello\"}}\n\n" ++ + "event: content_block_stop\n" ++ + "data: {\"type\":\"content_block_stop\",\"index\":0}\n\n" ++ + "event: content_block_start\n" ++ + "data: {\"type\":\"content_block_start\",\"index\":1,\"content_block\":{\"type\":\"tool_use\",\"id\":\"call_1\",\"name\":\"read_file\"}}\n\n" ++ + "event: content_block_delta\n" ++ + "data: {\"type\":\"content_block_delta\",\"index\":1,\"delta\":{\"type\":\"input_json_delta\",\"partial_json\":\"{\\\"path\\\":\\\"README.md\\\"}\"}}\n\n" ++ + "event: content_block_stop\n" ++ + "data: {\"type\":\"content_block_stop\",\"index\":1}\n\n" ++ + "event: message_delta\n" ++ + "data: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"tool_use\"},\"usage\":{\"output_tokens\":4}}\n\n" ++ + "event: message_stop\n" ++ + "data: {\"type\":\"message_stop\"}\n\n"; + var reader: std.Io.Reader = .fixed(sse_text); + var cancelled = std.atomic.Value(bool).init(false); + const Capture = struct { + content: std.ArrayList(u8) = .empty, + saw_read_file: bool = false, + + fn contentChunk(raw: *anyopaque, chunk: []const u8) void { + const self: *@This() = @ptrCast(@alignCast(raw)); + self.content.appendSlice(std.testing.allocator, chunk) catch unreachable; + } + fn toolStart(raw: *anyopaque, _: []const u8, name: []const u8, _: ?[]const u8) void { + const self: *@This() = @ptrCast(@alignCast(raw)); + self.saw_read_file = std.mem.eql(u8, name, "read_file"); + } + }; + var capture: Capture = .{}; + defer capture.content.deinit(std.testing.allocator); + const completion = try consumeSse( + std.testing.allocator, + &reader, + &capture, + Capture.contentChunk, + Capture.toolStart, + null, + null, + &cancelled, + null, + ); + defer { + if (completion.content) |value| std.testing.allocator.free(@constCast(value)); + types.freeToolCallSlice(std.testing.allocator, @constCast(completion.tool_calls)); + if (completion.generation_id) |value| std.testing.allocator.free(@constCast(value)); + } + try std.testing.expectEqualStrings("hello", capture.content.items); + try std.testing.expect(capture.saw_read_file); + try std.testing.expectEqual(@as(usize, 1), completion.tool_calls.len); + try std.testing.expectEqualStrings("call_1", completion.tool_calls[0].id); + try std.testing.expectEqualStrings("{\"path\":\"README.md\"}", completion.tool_calls[0].arguments_json); + try std.testing.expectEqual(@as(?u64, 10), completion.usage.input_tokens); + try std.testing.expectEqual(@as(?u64, 4), completion.usage.output_tokens); + try std.testing.expectEqual(types.ProviderFinishReason.tool_calls, completion.finish_reason.?); + try std.testing.expectEqualStrings("msg_1", completion.generation_id.?); +} + +test "Anthropic SSE maps stop reasons and surfaces error events" { + const cases = [_]struct { raw: []const u8, expected: types.ProviderFinishReason }{ + .{ .raw = "end_turn", .expected = .stop }, + .{ .raw = "stop_sequence", .expected = .stop }, + .{ .raw = "max_tokens", .expected = .length }, + .{ .raw = "refusal", .expected = .content_filter }, + .{ .raw = "pause_turn", .expected = .other }, + }; + for (cases) |case| { + var buffer: [256]u8 = undefined; + const sse_text = try std.fmt.bufPrint( + &buffer, + "data: {{\"type\":\"message_start\",\"message\":{{\"id\":\"m\"}}}}\n\n" ++ + "data: {{\"type\":\"message_delta\",\"delta\":{{\"stop_reason\":\"{s}\"}}}}\n\n" ++ + "data: {{\"type\":\"message_stop\"}}\n\n", + .{case.raw}, + ); + var reader: std.Io.Reader = .fixed(sse_text); + var cancelled = std.atomic.Value(bool).init(false); + var callback_context: u8 = 0; + var completion = try consumeSse( + std.testing.allocator, + &reader, + &callback_context, + ignoreTestChunk, + null, + null, + null, + &cancelled, + null, + ); + defer deinitTestCompletion(&completion); + try std.testing.expectEqual(case.expected, completion.finish_reason.?); + } + + const error_stream = "data: {\"type\":\"error\",\"error\":{\"type\":\"overloaded_error\",\"message\":\"overloaded\"}}\n\n"; + var error_reader: std.Io.Reader = .fixed(error_stream); + var cancelled = std.atomic.Value(bool).init(false); + var callback_context: u8 = 0; + try std.testing.expectError( + error.AnthropicResponseFailed, + consumeSse( + std.testing.allocator, + &error_reader, + &callback_context, + ignoreTestChunk, + null, + null, + null, + &cancelled, + null, + ), + ); + + const truncated_stream = "data: {\"type\":\"message_start\",\"message\":{\"id\":\"m\"}}\n\n"; + var truncated_reader: std.Io.Reader = .fixed(truncated_stream); + try std.testing.expectError( + error.AnthropicStreamIncomplete, + consumeSse( + std.testing.allocator, + &truncated_reader, + &callback_context, + ignoreTestChunk, + null, + null, + null, + &cancelled, + null, + ), + ); +} + +fn ignoreTestChunk(_: *anyopaque, _: []const u8) void {} + +fn deinitTestCompletion(completion: *types.GatewayCompletion) void { + if (completion.content) |value| std.testing.allocator.free(@constCast(value)); + if (completion.generation_id) |value| std.testing.allocator.free(@constCast(value)); + types.freeToolCallSlice(std.testing.allocator, @constCast(completion.tool_calls)); + if (completion.provider_state_json) |value| std.testing.allocator.free(@constCast(value)); + completion.* = .{}; +} + +test "Anthropic default endpoint composes the Messages path" { + const stable = try stableAnthropicTestEnviron(); + io_mod.setEnvironMap(stable); + const url = try messagesUrl(std.testing.allocator); + defer std.testing.allocator.free(url); + try std.testing.expectEqualStrings("https://api.anthropic.com/v1/messages", url); +} + +test "Anthropic base URL env resolves custom hosts" { + var map = std.process.Environ.Map.init(std.testing.allocator); + defer map.deinit(); + try map.put(base_url_env, "https://proxy.example/anthropic/"); + const stable = try stableAnthropicTestEnviron(); + io_mod.setEnvironMap(&map); + defer io_mod.setEnvironMap(stable); + + try std.testing.expectEqualStrings("https://proxy.example/anthropic/", resolveBaseUrl()); + const url = try messagesUrl(std.testing.allocator); + defer std.testing.allocator.free(url); + try std.testing.expectEqualStrings("https://proxy.example/anthropic/v1/messages", url); +} + +var stable_anthropic_test_environ: ?*std.process.Environ.Map = null; + +fn stableAnthropicTestEnviron() !*const std.process.Environ.Map { + if (stable_anthropic_test_environ) |map| return map; + const alloc = std.heap.page_allocator; + const map = try alloc.create(std.process.Environ.Map); + map.* = std.process.Environ.Map.init(alloc); + stable_anthropic_test_environ = map; + return map; +} diff --git a/src/gateway/anthropic_models.zig b/src/gateway/anthropic_models.zig new file mode 100644 index 000000000..637c67c8f --- /dev/null +++ b/src/gateway/anthropic_models.zig @@ -0,0 +1,374 @@ +const std = @import("std"); +const credentials = @import("../core/auth/credentials.zig"); +const model_catalog = @import("../core/gateway/model_catalog.zig"); +const gateway_provider = @import("../core/gateway/gateway_provider.zig"); +const io_mod = @import("../core/shared/io.zig"); +const types = @import("../core/shared/types.zig"); +const gateway_client = @import("client.zig"); + +const Allocator = std.mem.Allocator; +const max_catalog_models: usize = 128; +const max_model_id_bytes: usize = 256; +const max_catalog_bytes: usize = 1024 * 1024; +const fetch_timeout_ms: i64 = 30_000; +const default_models_endpoint = "https://api.anthropic.com/v1/models"; +pub const e2e_models_endpoint_env = "FX_E2E_ANTHROPIC_MODELS_URL"; +pub const base_url_env = "FX_ANTHROPIC_BASE_URL"; + +pub const model_catalog_provider = model_catalog.Provider{ + .fetch_fn = fetchCatalogForProvider, +}; + +pub const cli_model_catalog_provider = gateway_provider.CliModelCatalogProvider{ + .fetch_fn = fetchCliModelCatalog, +}; + +fn fetchCliModelCatalog( + _: ?*anyopaque, + alloc: Allocator, + input: gateway_provider.CliModelCatalogInput, +) gateway_provider.CliModelCatalogResult { + return switch (model_catalog.fetchWithPublicFallback(model_catalog_provider, alloc, .{ + .access = input.access, + .endpoint = input.endpoint, + .cancel_flag = input.cancel_flag, + .view = .full, + })) { + .loaded => |loaded| blk: { + var catalog = loaded.catalog; + defer model_catalog.freeModelCatalog(alloc, &catalog); + const ids = model_catalog.projectModelIds(alloc, catalog.items) catch return .{ .failure = .{ + .access = loaded.provenance.access, + .anonymous_fallback_used = false, + .failure = .{ .category = .resource_exhausted }, + } }; + break :blk .{ .loaded = .{ + .ids = ids, + .provenance = loaded.provenance, + } }; + }, + .failed => |failure| .{ .failure = failure }, + }; +} + +fn fetchCatalogForProvider( + _: ?*anyopaque, + alloc: Allocator, + input: model_catalog.FetchInput, +) Allocator.Error!model_catalog.ProviderResult { + if (input.access.credentialSource() != .anthropic_api_key) { + return .{ .failure = .{ .category = .authentication, .http_status = .unauthorized } }; + } + const credential = input.access.authorizationCredential() orelse + return .{ .failure = .{ .category = .authentication, .http_status = .unauthorized } }; + + const request_url = modelsUrl(alloc) catch |err| { + if (err == error.OutOfMemory) return error.OutOfMemory; + return .{ .failure = .{ .category = .runtime } }; + }; + defer alloc.free(request_url); + + var fallback_cancel = std.atomic.Value(bool).init(false); + const cancel_flag = input.cancel_flag orelse &fallback_cancel; + const deadline = std.Io.Clock.Timestamp.fromNow(io_mod.getIo(), .{ + .clock = .awake, + .raw = .fromMilliseconds(fetch_timeout_ms), + }); + var response = fetchCatalogResponse( + alloc, + request_url, + credential, + cancel_flag, + deadline, + ) catch |err| { + if (err == error.OutOfMemory) return error.OutOfMemory; + return .{ .failure = catalogFetchFailure(err) }; + }; + defer response.deinit(alloc); + if (response.status != .ok) { + return .{ .failure = model_catalog.failureForHttpStatus(response.status) }; + } + const catalog = parseCatalog(alloc, response.body) catch |err| { + if (err == error.OutOfMemory) return error.OutOfMemory; + return .{ .failure = .{ .category = .malformed_response, .http_status = .ok } }; + }; + return .{ .catalog = catalog }; +} + +fn catalogFetchFailure(err: anyerror) model_catalog.Failure { + if (err == error.Cancelled) return .{ .category = .cancellation }; + if (err == error.AnthropicModelCatalogTooLarge) return .{ .category = .malformed_response }; + return .{ .category = .transport, .retryable = true }; +} + +pub const FetchResponse = struct { + status: std.http.Status, + body: []u8, + + pub fn deinit(self: *FetchResponse, alloc: Allocator) void { + secretFree(alloc, self.body); + self.* = undefined; + } +}; + +fn secretFree(alloc: Allocator, bytes: []u8) void { + const secret = @import("../core/auth/secret.zig"); + secret.zeroAndFree(alloc, bytes); +} + +const FetchOperation = struct { + alloc: Allocator, + url: []const u8, + credential: []const u8, + + pub fn run(self: *@This()) !FetchResponse { + var client: std.http.Client = .{ .allocator = self.alloc, .io = io_mod.getIo() }; + defer client.deinit(); + const auth_header = try std.fmt.allocPrint(self.alloc, "Bearer {s}", .{self.credential}); + defer zeroFree(self.alloc, auth_header); + const body_buffer = try self.alloc.alloc(u8, max_catalog_bytes + 1); + defer zeroFree(self.alloc, body_buffer); + var response_writer = std.Io.Writer.fixed(body_buffer); + const extra_headers = [_]std.http.Header{ + .{ .name = "accept", .value = "application/json" }, + .{ .name = "anthropic-version", .value = "2023-06-01" }, + }; + const result = client.fetch(.{ + .location = .{ .url = self.url }, + .method = .GET, + .headers = .{ + .authorization = .{ .override = auth_header }, + .user_agent = .{ .override = gateway_client.user_agent }, + .accept_encoding = .omit, + }, + .extra_headers = &extra_headers, + .response_writer = &response_writer, + .redirect_behavior = .unhandled, + }) catch |err| switch (err) { + error.WriteFailed => return error.AnthropicModelCatalogTooLarge, + else => return err, + }; + const body = response_writer.buffered(); + try validateCatalogBodySize(body.len); + return .{ + .status = result.status, + .body = try self.alloc.dupe(u8, body), + }; + } +}; + +fn zeroFree(alloc: Allocator, bytes: []u8) void { + const secret = @import("../core/auth/secret.zig"); + secret.zeroAndFree(alloc, bytes); +} + +fn fetchCatalogResponse( + alloc: Allocator, + url: []const u8, + credential: []const u8, + cancel_flag: *std.atomic.Value(bool), + deadline: std.Io.Clock.Timestamp, +) !FetchResponse { + var operation = FetchOperation{ + .alloc = alloc, + .url = url, + .credential = credential, + }; + return gateway_client.runBoundedHttpOperation( + FetchResponse, + alloc, + cancel_flag, + deadline, + &operation, + ); +} + +fn modelsUrl(alloc: Allocator) ![]u8 { + if (io_mod.getenv(e2e_models_endpoint_env)) |override| { + if (!gateway_client.isLoopbackHttpUrl(override)) return error.InvalidE2EAnthropicModelsEndpoint; + return alloc.dupe(u8, override); + } + const base = baseFromEnv(); + if (std.mem.endsWith(u8, base, "/v1/models")) return alloc.dupe(u8, base); + const trimmed = std.mem.trimEnd(u8, base, "/"); + return std.fmt.allocPrint(alloc, "{s}/v1/models", .{trimmed}); +} + +fn baseFromEnv() []const u8 { + const raw = io_mod.getenv(base_url_env) orelse return "https://api.anthropic.com"; + const trimmed = std.mem.trim(u8, raw, " \t\r\n"); + if (trimmed.len == 0) return "https://api.anthropic.com"; + return trimmed; +} + +fn parseCatalog( + alloc: Allocator, + catalog_json: []const u8, +) !std.ArrayList(model_catalog.ModelCatalogEntry) { + var parsed = try std.json.parseFromSlice(std.json.Value, alloc, catalog_json, .{}); + defer parsed.deinit(); + if (parsed.value != .object) return error.InvalidAnthropicModelCatalog; + const models = parsed.value.object.get("data") orelse + return error.InvalidAnthropicModelCatalog; + if (models != .array) return error.InvalidAnthropicModelCatalog; + try validateCatalogModelCount(models.array.items.len); + + var catalog: std.ArrayList(model_catalog.ModelCatalogEntry) = .empty; + errdefer model_catalog.freeModelCatalog(alloc, &catalog); + for (models.array.items) |value| { + if (value != .object) return error.InvalidAnthropicModelCatalog; + const object = value.object; + const raw_type = try requiredString(object, "type"); + if (!std.mem.eql(u8, raw_type, "model")) continue; + const raw_id = try requiredString(object, "id"); + try validateModelId(raw_id); + + const id = try alloc.dupe(u8, raw_id); + errdefer alloc.free(id); + const model_type = try alloc.dupe(u8, "language"); + errdefer alloc.free(model_type); + + try catalog.append(alloc, .{ + .id = id, + .model_type = model_type, + .has_tool_use = true, + }); + } + return catalog; +} + +fn requiredString(object: std.json.ObjectMap, key: []const u8) ![]const u8 { + const value = object.get(key) orelse return error.InvalidAnthropicModelCatalog; + if (value != .string or value.string.len == 0) return error.InvalidAnthropicModelCatalog; + return value.string; +} + +fn validateModelId(id: []const u8) !void { + if (id.len == 0 or id.len > max_model_id_bytes) return error.InvalidAnthropicModelCatalog; + for (id) |byte| { + if (byte <= 0x20 or byte == 0x7f) return error.InvalidAnthropicModelCatalog; + } +} + +fn validateCatalogBodySize(size: usize) !void { + if (size > max_catalog_bytes) return error.AnthropicModelCatalogTooLarge; +} + +fn validateCatalogModelCount(count: usize) !void { + if (count > max_catalog_models) return error.InvalidAnthropicModelCatalog; +} + +test "Anthropic catalog parser maps provider model entries" { + const catalog_json = + \\{"data":[ + \\ {"id":"claude-sonnet-4-5","type":"model","display_name":"Claude Sonnet 4.5"}, + \\ {"id":"claude-opus-4-6","type":"model","display_name":"Claude Opus 4.6"}, + \\ {"id":"not-a-model","type":"other"} + \\],"has_more":false} + ; + var catalog = try parseCatalog(std.testing.allocator, catalog_json); + defer model_catalog.freeModelCatalog(std.testing.allocator, &catalog); + try std.testing.expectEqual(@as(usize, 2), catalog.items.len); + try std.testing.expectEqualStrings("claude-sonnet-4-5", catalog.items[0].id); + try std.testing.expectEqualStrings("language", catalog.items[0].model_type); + try std.testing.expect(catalog.items[0].has_tool_use); + try std.testing.expectEqualStrings("claude-opus-4-6", catalog.items[1].id); +} + +test "Anthropic catalog parser rejects malformed payloads" { + const cases = [_][]const u8{ + "[]", + "{}", + "{\"models\":[]}", + "{\"data\":{}}", + "{\"data\":[{}]}", + "{\"data\":[{\"id\":\"\",\"type\":\"model\"}]}", + "{\"data\":[{\"id\":\"a b\",\"type\":\"model\"}]}", + }; + for (cases) |case| { + try expectCatalogParseError(error.InvalidAnthropicModelCatalog, case); + } +} + +fn expectCatalogParseError(expected: anyerror, json: []const u8) !void { + var catalog = parseCatalog(std.testing.allocator, json) catch |err| { + try std.testing.expectEqual(expected, err); + return; + }; + defer model_catalog.freeModelCatalog(std.testing.allocator, &catalog); + return error.TestExpectedCatalogFailure; +} + +test "Anthropic models URL composes the provider endpoint" { + const stable = try stableAnthropicCatalogTestEnviron(); + io_mod.setEnvironMap(stable); + const url = try modelsUrl(std.testing.allocator); + defer std.testing.allocator.free(url); + try std.testing.expectEqualStrings(default_models_endpoint, url); +} + +test "Anthropic model ids enforce the exact provider-local bound" { + const exact_json = try buildCatalogJson(std.testing.allocator, max_model_id_bytes); + defer std.testing.allocator.free(exact_json); + var exact = try parseCatalog(std.testing.allocator, exact_json); + defer model_catalog.freeModelCatalog(std.testing.allocator, &exact); + try std.testing.expectEqual(@as(usize, 1), exact.items.len); + try std.testing.expectEqual(max_model_id_bytes, exact.items[0].id.len); + + const excess_json = try buildCatalogJson(std.testing.allocator, max_model_id_bytes + 1); + defer std.testing.allocator.free(excess_json); + try expectCatalogParseError(error.InvalidAnthropicModelCatalog, excess_json); +} + +fn buildCatalogJson(alloc: Allocator, id_bytes: usize) ![]u8 { + var out: std.Io.Writer.Allocating = .init(alloc); + errdefer out.deinit(); + try out.writer.writeAll("{\"data\":[{\"type\":\"model\",\"id\":\""); + try out.writer.splatByteAll('a', id_bytes); + try out.writer.writeAll("\"}]}"); + return out.toOwnedSlice(); +} + +test "Anthropic catalog requires the anthropic api key credential" { + const result = try model_catalog_provider.fetch(std.testing.allocator, .{ + .access = .{ .authenticated = .{ + .source = .grok_subscription, + .credential = "grok-token", + .team_context = null, + } }, + .endpoint = "", + }); + switch (result) { + .failure => |failure| { + try std.testing.expectEqual(model_catalog.FailureCategory.authentication, failure.category); + try std.testing.expectEqual(std.http.Status.unauthorized, failure.http_status.?); + }, + .catalog => |catalog| { + var unexpected = catalog; + model_catalog.freeModelCatalog(std.testing.allocator, &unexpected); + return error.TestExpectedAuthenticationFailure; + }, + } +} + +test "Anthropic oversized catalog responses are terminal malformed data" { + const failure = catalogFetchFailure(error.AnthropicModelCatalogTooLarge); + try std.testing.expectEqual(model_catalog.FailureCategory.malformed_response, failure.category); + try std.testing.expect(!failure.retryable); + const cancelled = catalogFetchFailure(error.Cancelled); + try std.testing.expectEqual(model_catalog.FailureCategory.cancellation, cancelled.category); + const transport = catalogFetchFailure(error.ConnectionRefused); + try std.testing.expectEqual(model_catalog.FailureCategory.transport, transport.category); + try std.testing.expect(transport.retryable); +} + +var stable_anthropic_catalog_test_environ: ?*std.process.Environ.Map = null; + +fn stableAnthropicCatalogTestEnviron() !*const std.process.Environ.Map { + if (stable_anthropic_catalog_test_environ) |map| return map; + const alloc = std.heap.page_allocator; + const map = try alloc.create(std.process.Environ.Map); + map.* = std.process.Environ.Map.init(alloc); + stable_anthropic_catalog_test_environ = map; + return map; +} From ec2683d77c2787f76fb38680a7487b925e7a5ade Mon Sep 17 00:00:00 2001 From: Joaquin Terrasa Date: Fri, 28 Aug 2026 18:59:06 +0200 Subject: [PATCH 3/7] feat(runtime): route anthropic through provider selection and cli surfaces Register the anthropic bundle in builtins/providers.zig and thread it through every surface the openai bundle reaches: the auth provider picker (choiceAt now yields the fifth provider entry), acp server and subagent prompt routing, cli_ask and cli_surface config structs, doctor diagnostics, output contracts, credential resolution call sites (profile anthropic key propagated alongside the openai key), and main.zig dependency injection for both the entry config and the ask/acp runner paths. Anthropic carries no permission reviewer yet, matching the optional reviewer slots; openai_compatible.zig rejects the anthropic_api_key source symmetrically. The model menu lists the anthropic credential label. Agent streams flow through the same ProviderRoutes dispatch as the other wire protocols. --- src/acp/server.zig | 3 + src/builtins/providers.zig | 8 + src/core/app/app_agent_runtime.zig | 4 + src/core/app/app_auth_runtime.zig | 2 + src/core/app/app_lifecycle.zig | 2 + src/core/auth/auth_runtime.zig | 10 + src/core/auth/auth_transition.zig | 1 + src/core/auth/credential_authority.zig | 1 + src/core/cli/cli_ask.zig | 1 + src/core/cli/cli_surface.zig | 14 +- src/core/config/settings_store.zig | 1 + src/core/gateway/provider_set.zig | 6 +- src/core/output/output_contracts.zig | 1 + .../session/generation_usage_provider.zig | 2 + src/core/session/session_usage.zig | 1 + src/core/subagent/agent_adapter.zig | 1 + src/gateway/anthropic.zig | 188 +++++++++++------- src/ui/footer/model_menu_presentation.zig | 1 + 18 files changed, 176 insertions(+), 71 deletions(-) diff --git a/src/acp/server.zig b/src/acp/server.zig index 967ced77b..2eace04d3 100644 --- a/src/acp/server.zig +++ b/src/acp/server.zig @@ -353,6 +353,7 @@ pub fn selectCredentialForProvider( .refresh_if_needed, provider, state.credential_source, + null, ); break :blk resolution.credential orelse return false; }; @@ -1370,6 +1371,7 @@ fn handleInitialize(state: *ServerState, alloc: Allocator, msg: *jsonrpc.Message .refresh_if_needed, state.provider, preferred, + null, ); routed_credential = resolution.credential; if (routed_credential == null) { @@ -1651,6 +1653,7 @@ fn handleSetConfigOption(state: *ServerState, alloc: Allocator, msg: *jsonrpc.Me .refresh_if_needed, target, null, + null, ); break :credential resolution.credential orelse return state.writer.writeError(alloc, msg.id, .{ diff --git a/src/builtins/providers.zig b/src/builtins/providers.zig index ac63905b1..8e0133cdd 100644 --- a/src/builtins/providers.zig +++ b/src/builtins/providers.zig @@ -6,6 +6,8 @@ const openai_codex_permission_reviewer = @import("../gateway/openai_codex_permis const xai_grok = @import("../gateway/xai_grok.zig"); const xai_grok_models = @import("../gateway/xai_grok_models.zig"); const xai_grok_permission_reviewer = @import("../gateway/xai_grok_permission_reviewer.zig"); +const anthropic = @import("../gateway/anthropic.zig"); +const anthropic_models = @import("../gateway/anthropic_models.zig"); const provider_catalog = @import("../core/auth/provider_catalog.zig"); pub const native = provider_set.Set{ @@ -26,4 +28,10 @@ pub const native = provider_set.Set{ .model_catalog = xai_grok_models.model_catalog_provider, .permission_reviewer = xai_grok_permission_reviewer.provider, }, + .anthropic = .{ + .presentation = provider_catalog.find(.anthropic), + .agent_stream = anthropic.agent_stream_provider, + .cli_model_catalog = anthropic_models.cli_model_catalog_provider, + .model_catalog = anthropic_models.model_catalog_provider, + }, }; diff --git a/src/core/app/app_agent_runtime.zig b/src/core/app/app_agent_runtime.zig index 02645d783..8dd0610c0 100644 --- a/src/core/app/app_agent_runtime.zig +++ b/src/core/app/app_agent_runtime.zig @@ -1064,6 +1064,10 @@ pub fn Runtime(comptime App: type) type { .agent_stream = tool_context.agent_stream_provider, .permission_reviewer = tool_context.permission_reviewer_provider, }, + .anthropic = .{ + .agent_stream_provider = tool_context.agent_stream_provider, + .permission_reviewer_provider = null, + }, }; return subagent_agent_adapter.run(.{ .host = app_session_runtime.Runtime(App).subagentHost(app) orelse diff --git a/src/core/app/app_auth_runtime.zig b/src/core/app/app_auth_runtime.zig index 3cfa808ff..498cdbac9 100644 --- a/src/core/app/app_auth_runtime.zig +++ b/src/core/app/app_auth_runtime.zig @@ -59,6 +59,7 @@ pub fn Runtime(comptime App: type) type { const required_source: credentials.Source = switch (provider) { .codex => .chatgpt_subscription, .grok => .grok_subscription, + .anthropic => .anthropic_api_key, .gateway => app.auth.credentialSource() orelse .fx_login, }; const route_change = app.auth.selectForProvider(app.alloc, provider) catch |err| switch (err) { @@ -778,6 +779,7 @@ pub fn Runtime(comptime App: type) type { .refresh_if_needed, target, null, + null, ) catch |err| { debug_trace.logf("provider", "credential preparation failed provider={t} err={s}", .{ target, @errorName(err) }); try app.writeDomainNotice(.{ diff --git a/src/core/app/app_lifecycle.zig b/src/core/app/app_lifecycle.zig index 37ea7ef85..67eb62799 100644 --- a/src/core/app/app_lifecycle.zig +++ b/src/core/app/app_lifecycle.zig @@ -412,6 +412,7 @@ fn loadStartupStateFromOwnedWorkspace( mode, state.provider, settings.credential_source, + null, ); state.credential = resolution.credential; state.stored_key_status = resolution.stored_key_status; @@ -1115,6 +1116,7 @@ fn configuredProviderSelection( .gateway => default_model, .codex => return error.CodexModelNotSelected, .grok => return error.GrokModelNotSelected, + .anthropic => return error.AnthropicModelNotSelected, }; return .{ .provider = provider, .model = model }; } diff --git a/src/core/auth/auth_runtime.zig b/src/core/auth/auth_runtime.zig index d96b188b8..afce87491 100644 --- a/src/core/auth/auth_runtime.zig +++ b/src/core/auth/auth_runtime.zig @@ -678,6 +678,7 @@ pub fn loadStatusSnapshotForProvider( .stored, selected_provider, preferred, + null, ) else credentials.resolvePreferring( @@ -1636,6 +1637,15 @@ pub const Runtime = struct { self, loadRuntimeCredentialSource, ), + .anthropic => if (self.credentialSource() == .anthropic_api_key) + false + else + self.selectSourceWithLoader( + alloc, + .anthropic_api_key, + self, + loadRuntimeCredentialSource, + ), .gateway => if (self.credentialSource() != .chatgpt_subscription and self.credentialSource() != .grok_subscription) false else diff --git a/src/core/auth/auth_transition.zig b/src/core/auth/auth_transition.zig index 84abed350..6d3dc4b46 100644 --- a/src/core/auth/auth_transition.zig +++ b/src/core/auth/auth_transition.zig @@ -73,6 +73,7 @@ pub fn signInCompletion( .{ .switch_provider = .grok } else .{ .activate_source = .grok_subscription }, + .anthropic => .{ .activate_source = .anthropic_api_key }, }; } diff --git a/src/core/auth/credential_authority.zig b/src/core/auth/credential_authority.zig index a031d3c42..c453644d5 100644 --- a/src/core/auth/credential_authority.zig +++ b/src/core/auth/credential_authority.zig @@ -25,6 +25,7 @@ pub fn derive( .ai_gateway_api_key, .fx_login, .stored_key, + .anthropic_api_key, => hash.update("\x00slot\x00"), .chatgpt_subscription, .grok_subscription, diff --git a/src/core/cli/cli_ask.zig b/src/core/cli/cli_ask.zig index 586f41149..8180d1904 100644 --- a/src/core/cli/cli_ask.zig +++ b/src/core/cli/cli_ask.zig @@ -1559,6 +1559,7 @@ fn runPromptInternal(alloc: Allocator, prompt: []const u8, permission_override: .refresh_if_needed, ctx.provider, preferred, + null, ); routed_credential = resolution.credential; if (routed_credential == null) { diff --git a/src/core/cli/cli_surface.zig b/src/core/cli/cli_surface.zig index dc4f4f915..e66ba02f7 100644 --- a/src/core/cli/cli_surface.zig +++ b/src/core/cli/cli_surface.zig @@ -694,6 +694,7 @@ fn activateProviderSelection( .refresh_if_needed, target, settings.credential_source, + null, ); defer if (resolution.credential) |*credential| credential.deinit(alloc); @@ -703,6 +704,7 @@ fn activateProviderSelection( .gateway => "Gateway is already selected.\n", .codex => "Codex is already selected.\n", .grok => "Grok is already selected.\n", + .anthropic => "Anthropic is already selected.\n", }); return true; } @@ -722,6 +724,7 @@ fn activateProviderSelection( .refresh_if_needed, target, settings.credential_source, + null, ); } if (resolution.credential == null and target == .grok and caller == .provider_command) { @@ -738,6 +741,7 @@ fn activateProviderSelection( .refresh_if_needed, target, settings.credential_source, + null, ); } @@ -749,6 +753,7 @@ fn activateProviderSelection( switch (target) { .codex => "Codex credential is unavailable", .grok => "Grok credential is unavailable", + .anthropic => "Anthropic credential is unavailable", .gateway => "configure a Gateway credential first", }, ); @@ -758,6 +763,7 @@ fn activateProviderSelection( try writeProviderActivationError(alloc, deps, caller, switch (target) { .codex => "Codex model catalog is unavailable", .grok => "Grok model catalog is unavailable", + .anthropic => "Anthropic model catalog is unavailable", .gateway => "Gateway model catalog is unavailable", }); return false; @@ -803,13 +809,14 @@ fn activateProviderSelection( if (performed_login) |provider| switch (provider) { .codex => try writeStdout(deps, "Signed in with Codex.\n"), .grok => try writeStdout(deps, "Signed in with Grok.\n"), - .gateway => unreachable, + .gateway, .anthropic => unreachable, }; if (caller == .provider_command) { try writeStdout(deps, switch (target) { .gateway => "Provider set to Gateway.\n", .codex => "Provider set to Codex.\n", .grok => "Provider set to Grok.\n", + .anthropic => "Provider set to Anthropic.\n", }); } return true; @@ -986,6 +993,10 @@ fn runNonInteractiveWithDeps( } try writeStdout(deps, "Signed in with Grok.\n"); }, + .anthropic => { + try writeStderr(deps, "fx login: Anthropic uses API keys; set ANTHROPIC_API_KEY or run fx api-key anthropic\n"); + return .handled_failure; + }, } return .handled_success; }, @@ -1180,6 +1191,7 @@ fn runNonInteractiveWithDeps( .gateway => "fx models: Gateway model catalog is unavailable\n", .codex => "fx models: Codex model catalog is unavailable\n", .grok => "fx models: Grok model catalog is unavailable\n", + .anthropic => "fx models: Anthropic model catalog is unavailable\n", }); return .handled_failure; }; diff --git a/src/core/config/settings_store.zig b/src/core/config/settings_store.zig index e8c266969..2ef1f06ed 100644 --- a/src/core/config/settings_store.zig +++ b/src/core/config/settings_store.zig @@ -1561,6 +1561,7 @@ fn putModelPreference( .gateway => "model", .codex => "codex_model", .grok => "grok_model", + .anthropic => "anthropic_model", }; if (root.contains(legacy_key)) { _ = root.orderedRemove(legacy_key); diff --git a/src/core/gateway/provider_set.zig b/src/core/gateway/provider_set.zig index 97d921acb..270c0c650 100644 --- a/src/core/gateway/provider_set.zig +++ b/src/core/gateway/provider_set.zig @@ -51,12 +51,14 @@ pub const Set = struct { gateway: Bundle, codex: Bundle, grok: Bundle, + anthropic: Bundle, pub fn select(self: Set, provider: model_provider.ProviderId) Bundle { return switch (provider) { .gateway => self.gateway, .codex => self.codex, .grok => self.grok, + .anthropic => self.anthropic, }; } @@ -65,6 +67,7 @@ pub const Set = struct { .gateway = self.gateway.deferred_usage, .codex = self.codex.deferred_usage, .grok = self.grok.deferred_usage, + .anthropic = self.anthropic.deferred_usage, }; } }; @@ -74,6 +77,7 @@ pub fn gateway_only(gateway: Bundle) Set { .gateway = gateway, .codex = .{}, .grok = .{}, + .anthropic = .{}, }; } @@ -144,7 +148,7 @@ test "provider set selects each provider's complete route" { .model_catalog = .{ .context = &grok_tag, .fetch_fn = Fake.model_catalog_fetch }, .permission_reviewer = .{ .context = &grok_tag, .review_fn = Fake.review }, }; - var providers = Set{ .gateway = gateway, .codex = codex, .grok = grok }; + var providers = Set{ .gateway = gateway, .codex = codex, .grok = grok, .anthropic = .{} }; try std.testing.expect(providers.select(.gateway).agent_stream.?.context.? == @as(*anyopaque, @ptrCast(&gateway_tag))); try std.testing.expect(providers.select(.gateway).capabilities.fx_search); diff --git a/src/core/output/output_contracts.zig b/src/core/output/output_contracts.zig index 271346b91..8a846d8f3 100644 --- a/src/core/output/output_contracts.zig +++ b/src/core/output/output_contracts.zig @@ -852,6 +852,7 @@ pub const ModelListSnapshot = struct { .gateway => "gateway", .codex => provider_catalog.label(.codex), .grok => provider_catalog.label(.grok), + .anthropic => provider_catalog.label(.anthropic), }; } diff --git a/src/core/session/generation_usage_provider.zig b/src/core/session/generation_usage_provider.zig index 19ad021b1..b750ff3aa 100644 --- a/src/core/session/generation_usage_provider.zig +++ b/src/core/session/generation_usage_provider.zig @@ -85,6 +85,7 @@ pub const Set = struct { gateway: ?Provider = null, codex: ?Provider = null, grok: ?Provider = null, + anthropic: ?Provider = null, pub fn gatewayOnly(provider: Provider) Set { return .{ .gateway = provider }; @@ -95,6 +96,7 @@ pub const Set = struct { .gateway => self.gateway, .codex => self.codex, .grok => self.grok, + .anthropic => self.anthropic, }; } }; diff --git a/src/core/session/session_usage.zig b/src/core/session/session_usage.zig index 1d354e746..901358471 100644 --- a/src/core/session/session_usage.zig +++ b/src/core/session/session_usage.zig @@ -3272,6 +3272,7 @@ fn exactUsageOrigin(provider: model_provider.ProviderId) []const u8 { .gateway => "exact/gateway", .codex => "exact/codex", .grok => "exact/grok", + .anthropic => "exact/anthropic", }; } diff --git a/src/core/subagent/agent_adapter.zig b/src/core/subagent/agent_adapter.zig index eea14a6bc..367fde905 100644 --- a/src/core/subagent/agent_adapter.zig +++ b/src/core/subagent/agent_adapter.zig @@ -156,6 +156,7 @@ pub fn run( .refresh_if_needed, admission.provider, config.tool_context.credential_source, + null, ) catch |err| { if (err == error.OutOfMemory) return error.OutOfMemory; turn.setFailureDiagnostic("model_credential_resolution_failed", @errorName(err)) catch diff --git a/src/gateway/anthropic.zig b/src/gateway/anthropic.zig index 6dd37f4fb..3732d19e6 100644 --- a/src/gateway/anthropic.zig +++ b/src/gateway/anthropic.zig @@ -4,6 +4,7 @@ const stream_provider = @import("../core/agent/stream_provider.zig"); const io_mod = @import("../core/shared/io.zig"); const types = @import("../core/shared/types.zig"); const gateway_client = @import("client.zig"); +const model_tool_schema = @import("../core/tooling/model_tool_schema.zig"); const Allocator = std.mem.Allocator; const default_base_url = "https://api.anthropic.com"; @@ -24,7 +25,6 @@ const connect_timeout_ms: i64 = 30_000; pub const e2e_endpoint_env = "FX_E2E_ANTHROPIC_URL"; pub const agent_stream_provider = stream_provider.Provider{ - .build_fn = buildRequest, .stream_fn = streamCompletion, }; @@ -48,10 +48,9 @@ fn validateModel(model: []const u8) !void { } } -fn buildRequest( - _: ?*anyopaque, +pub fn buildRequest( alloc: Allocator, - request: stream_provider.BuildRequest, + request: stream_provider.RequestData, ) ![]u8 { try validateModel(request.model); if (request.budget) |budget| { @@ -84,7 +83,7 @@ fn buildRequest( try writer.writeAll(",\"stream\":true,\"messages\":["); try writeMessages(writer, request.messages); try writer.writeByte(']'); - const tool_count = try writeTools(writer, alloc, request.serialized_tools, request.selected_dynamic_tool_schemas); + const tool_count = try writeTools(writer, alloc, request.tools); if (tool_count > 0) { try writer.writeAll(",\"tool_choice\":"); try writeToolChoice(writer, request.tool_choice); @@ -178,57 +177,69 @@ fn writeToolResult(writer: *std.Io.Writer, message: types.ChatMessage) !void { fn writeTools( writer: *std.Io.Writer, alloc: Allocator, - serialized_tools: []const u8, - selected_dynamic_schemas: []const []const u8, + tools: stream_provider.ToolSelection, ) !usize { var count: usize = 0; - var parsed = std.json.parseFromSlice(std.json.Value, alloc, serialized_tools, .{}) catch |err| switch (err) { - error.OutOfMemory => return error.OutOfMemory, - else => return error.InvalidToolSchema, - }; - defer parsed.deinit(); - if (parsed.value != .array) return error.InvalidToolSchema; - var tools_out: std.Io.Writer.Allocating = .init(alloc); defer tools_out.deinit(); try tools_out.writer.writeAll(",\"tools\":["); - for (parsed.value.array.items) |tool| { - if (try writeFunctionTool(&tools_out.writer, tool, count != 0)) count += 1; + + const InputSchema = union(enum) { + static: model_tool_schema.ObjectSchema, + dynamic: std.json.Value, + }; + + const S = struct { + fn writeFunctionTool( + w: *std.Io.Writer, + a: Allocator, + name: []const u8, + description: []const u8, + input_schema: InputSchema, + ) !void { + if (name.len == 0) return error.InvalidToolSchema; + try w.writeAll("{\"name\":"); + try std.json.Stringify.value(name, .{}, w); + if (description.len > 0) { + try w.writeAll(",\"description\":"); + try std.json.Stringify.value(description, .{}, w); + } + try w.writeAll(",\"input_schema\":"); + switch (input_schema) { + .static => |schema| try model_tool_schema.writeObjectSchema(a, w, schema), + .dynamic => |schema| try std.json.Stringify.value(schema, .{}, w), + } + try w.writeByte('}'); + } + fn containsName(names: []const []const u8, expected: []const u8) bool { + for (names) |name| if (std.mem.eql(u8, name, expected)) return true; + return false; + } + }; + + for (tools.advertised_names) |name| { + const tool = tools.advertisedFunction(name) orelse continue; + if (count > 0) try tools_out.writer.writeByte(','); + try S.writeFunctionTool(&tools_out.writer, alloc, tool.name, tool.description, .{ .static = tool.input_schema }); + count += 1; } - for (selected_dynamic_schemas) |schema_json| { - var selected = std.json.parseFromSlice(std.json.Value, alloc, schema_json, .{}) catch |err| switch (err) { - error.OutOfMemory => return error.OutOfMemory, - else => return error.InvalidToolSchema, - }; - defer selected.deinit(); - if (try writeFunctionTool(&tools_out.writer, selected.value, count != 0)) count += 1; + for (tools.additional_functions) |tool| { + if (S.containsName(tools.advertised_names, tool.name)) continue; + if (count > 0) try tools_out.writer.writeByte(','); + try S.writeFunctionTool(&tools_out.writer, alloc, tool.name, tool.description, .{ .static = tool.input_schema }); + count += 1; + } + for (tools.selected_dynamic) |tool| { + if (S.containsName(tools.advertised_names, tool.name)) continue; + if (count > 0) try tools_out.writer.writeByte(','); + try S.writeFunctionTool(&tools_out.writer, alloc, tool.name, tool.description, .{ .dynamic = tool.input_schema }); + count += 1; } try tools_out.writer.writeByte(']'); if (count > 0) try writer.writeAll(tools_out.written()); return count; } -fn writeFunctionTool(writer: *std.Io.Writer, value: std.json.Value, comma: bool) !bool { - if (value != .object) return false; - const kind = value.object.get("type") orelse return false; - if (kind != .string or !std.mem.eql(u8, kind.string, "function")) return false; - const name = value.object.get("name") orelse return false; - if (name != .string or name.string.len == 0) return false; - const parameters = value.object.get("inputSchema") orelse value.object.get("parameters") orelse return false; - if (parameters != .object) return false; - if (comma) try writer.writeByte(','); - try writer.writeAll("{\"name\":"); - try std.json.Stringify.value(name.string, .{}, writer); - if (value.object.get("description")) |description| if (description == .string) { - try writer.writeAll(",\"description\":"); - try std.json.Stringify.value(description.string, .{}, writer); - }; - try writer.writeAll(",\"input_schema\":"); - try std.json.Stringify.value(parameters, .{}, writer); - try writer.writeByte('}'); - return true; -} - fn writeToolChoice(writer: *std.Io.Writer, choice: types.ToolChoice) !void { switch (choice) { .auto => try writer.writeAll("{\"type\":\"auto\"}"), @@ -237,6 +248,21 @@ fn writeToolChoice(writer: *std.Io.Writer, choice: types.ToolChoice) !void { } } +fn failureKind(status: std.http.Status) stream_provider.FailureKind { + return switch (status) { + .bad_request => .invalid_request, + .unauthorized => .unauthorized, + .forbidden => .forbidden, + .payload_too_large => .request_too_large, + .too_many_requests => .rate_limited, + .internal_server_error => .server_error, + .bad_gateway => .bad_gateway, + .service_unavailable => .unavailable, + .gateway_timeout => .gateway_timeout, + else => .provider_error, + }; +} + fn writeComma(writer: *std.Io.Writer, first: *bool) !void { if (!first.*) try writer.writeByte(','); first.* = false; @@ -245,7 +271,7 @@ fn writeComma(writer: *std.Io.Writer, first: *bool) !void { fn streamCompletion( _: ?*anyopaque, alloc: Allocator, - request: stream_provider.Request, + request: stream_provider.ModelRequest, ) !stream_provider.Result { var result = streamCompletionCore(alloc, request) catch |err| { if (request.cancel_flag.load(.seq_cst)) return error.Cancelled; @@ -260,7 +286,7 @@ fn streamCompletion( return result; } -fn requestDeadlineExpired(request: stream_provider.Request) bool { +fn requestDeadlineExpired(request: stream_provider.ModelRequest) bool { const deadline = request.deadline orelse return false; const now = std.Io.Clock.Timestamp.now(io_mod.getIo(), .awake); return !std.Io.Clock.Timestamp.compare(now, .lt, deadline); @@ -302,16 +328,18 @@ const OpenRequestOperation = struct { } }; -fn streamCompletionCore(alloc: Allocator, request: stream_provider.Request) !stream_provider.Result { +fn streamCompletionCore(alloc: Allocator, request: stream_provider.ModelRequest) !stream_provider.Result { if (request.cancel_flag.load(.seq_cst)) return error.Cancelled; - if (request.credential_source) |source| { - // Referenced by name so this module compiles before the enum case lands. - const anthropic_api_key = std.meta.stringToEnum(types.CredentialSource, "anthropic_api_key"); - if (anthropic_api_key == null or source != anthropic_api_key.?) { + if (request.credential.source) |source| { + if (source != .anthropic_api_key) { return error.AnthropicApiKeyCredentialRequired; } + } else { + return error.AnthropicApiKeyCredentialRequired; } try validateModel(request.model); + const payload = try buildRequest(alloc, request.data()); + defer alloc.free(payload); const request_endpoint = if (io_mod.getenv(e2e_endpoint_env)) |override| endpoint: { if (!gateway_client.isLoopbackHttpUrl(override)) return error.InvalidE2EAnthropicEndpoint; break :endpoint override; @@ -323,7 +351,7 @@ fn streamCompletionCore(alloc: Allocator, request: stream_provider.Request) !str var extra_count: usize = 0; extra_headers_buf[extra_count] = .{ .name = "accept", .value = "text/event-stream" }; extra_count += 1; - extra_headers_buf[extra_count] = .{ .name = "x-api-key", .value = request.api_key }; + extra_headers_buf[extra_count] = .{ .name = "x-api-key", .value = request.credential.secret }; extra_count += 1; extra_headers_buf[extra_count] = .{ .name = "anthropic-version", .value = anthropic_version }; extra_count += 1; @@ -333,7 +361,7 @@ fn streamCompletionCore(alloc: Allocator, request: stream_provider.Request) !str var open_operation = OpenRequestOperation{ .client = &client, .uri = uri, - .api_key = request.api_key, + .api_key = request.credential.secret, .extra_headers = extra_headers_buf[0..extra_count], }; var connect_deadline = std.Io.Clock.Timestamp.fromNow(io_mod.getIo(), .{ @@ -377,11 +405,11 @@ fn streamCompletionCore(alloc: Allocator, request: stream_provider.Request) !str } if (request.cancel_flag.load(.seq_cst)) return error.Cancelled; - http_request.transfer_encoding = .{ .content_length = request.payload.len }; + http_request.transfer_encoding = .{ .content_length = payload.len }; var send_buffer: [8192]u8 = undefined; request.delivery.markPossiblySent(); var body_writer = try http_request.sendBodyUnflushed(&send_buffer); - try body_writer.writer.writeAll(request.payload); + try body_writer.writer.writeAll(payload); try body_writer.end(); if (http_request.connection) |connection| try connection.flush(); if (request.cancel_flag.load(.seq_cst)) return error.Cancelled; @@ -398,33 +426,55 @@ fn streamCompletionCore(alloc: Allocator, request: stream_provider.Request) !str alloc.free(bounded_body); break :body try alloc.dupe(u8, "Anthropic error response exceeded the local limit"); } else bounded_body; - return .{ - .status = response.head.status, - .err_body = body, + return .{ .failed = .{ + .kind = failureKind(response.head.status), + .detail = body, .ownership = .owned, - }; + } }; } var transfer_buffer: [transfer_buffer_bytes]u8 = undefined; const reader = response.reader(&transfer_buffer); + var events = request.events; const completion = try consumeSse( alloc, reader, - request.callback_ctx, - request.on_content_chunk, - request.on_tool_start, - request.on_reasoning_chunk, - request.on_tool_input_chunk, + &events, + EventBridge.content, + EventBridge.toolStart, + EventBridge.reasoning, + EventBridge.toolInput, request.cancel_flag, request.content_capture_limit, ); - return .{ - .status = .ok, + return .{ .completed = .{ .completion = completion, .ownership = .owned, - }; + } }; } +const EventBridge = struct { + fn sink(raw: *anyopaque) *stream_provider.EventSink { + return @ptrCast(@alignCast(raw)); + } + + fn content(raw: *anyopaque, chunk: []const u8) void { + sink(raw).emit(.{ .content_delta = chunk }); + } + + fn reasoning(raw: *anyopaque, chunk: []const u8) void { + sink(raw).emit(.{ .reasoning_delta = chunk }); + } + + fn toolInput(raw: *anyopaque, chunk: []const u8) void { + sink(raw).emit(.{ .tool_input_delta = chunk }); + } + + fn toolStart(raw: *anyopaque, id: []const u8, name: []const u8, label: ?[]const u8) void { + sink(raw).emit(.{ .tool_started = .{ .id = id, .name = name, .label = label } }); + } +}; + const ToolAccumulator = struct { block_index: i64, id: []u8, @@ -534,7 +584,7 @@ fn consumeSse( on_tool_input_chunk: ?stream_provider.StreamCallback, cancel_flag: *std.atomic.Value(bool), content_capture_limit: ?usize, -) !types.GatewayCompletion { +) !types.ModelCompletion { var content: std.ArrayList(u8) = .empty; errdefer content.deinit(alloc); var tools: std.ArrayList(ToolAccumulator) = .empty; @@ -949,7 +999,7 @@ test "Anthropic SSE maps stop reasons and surfaces error events" { fn ignoreTestChunk(_: *anyopaque, _: []const u8) void {} -fn deinitTestCompletion(completion: *types.GatewayCompletion) void { +fn deinitTestCompletion(completion: *types.ModelCompletion) void { if (completion.content) |value| std.testing.allocator.free(@constCast(value)); if (completion.generation_id) |value| std.testing.allocator.free(@constCast(value)); types.freeToolCallSlice(std.testing.allocator, @constCast(completion.tool_calls)); diff --git a/src/ui/footer/model_menu_presentation.zig b/src/ui/footer/model_menu_presentation.zig index f566b2f7c..429eef57d 100644 --- a/src/ui/footer/model_menu_presentation.zig +++ b/src/ui/footer/model_menu_presentation.zig @@ -418,6 +418,7 @@ fn loadedCatalogStatusText(state: model_cache_runtime.ModelMenuCatalogState) ?[] .stored_key => "Gateway catalog: authenticated with the stored API key.", .chatgpt_subscription => "Codex catalog: authenticated with a subscription.", .grok_subscription => "Grok catalog: authenticated with a subscription.", + .anthropic_api_key => "Anthropic catalog: authenticated with an API key.", }; } return null; From f5599c3e4af0b875f18b86f64590c3c2d656ef34 Mon Sep 17 00:00:00 2001 From: Joaquin Terrasa Date: Fri, 28 Aug 2026 18:59:16 +0200 Subject: [PATCH 4/7] fix(lifecycle): fall back to FX_MODEL when a provider-scoped model is unset configuredProviderSelection returned ModelNotSelected whenever the saved settings lacked a model for the active provider, even when FX_MODEL carried a process override, making 'FX_PROVIDER=openai FX_MODEL=...' fail before the stream provider was ever consulted. Fall back to the trimmed FX_MODEL value before erroring for the non-gateway providers (anthropic included) and scope each TestEnv install in the override test to its own block so the FX_PROVIDER assertions no longer inherit a stale environment. Verified live against an OpenAI-compatible gateway on both the openai and anthropic wires, including a tool-calling round trip. --- src/core/app/app_lifecycle.zig | 31 +++++++++++++++++++++++++++++-- 1 file changed, 29 insertions(+), 2 deletions(-) diff --git a/src/core/app/app_lifecycle.zig b/src/core/app/app_lifecycle.zig index 67eb62799..a5ca4c1d1 100644 --- a/src/core/app/app_lifecycle.zig +++ b/src/core/app/app_lifecycle.zig @@ -149,6 +149,13 @@ pub const StartupState = struct { notification_attention_required: bool = false, notification_max: bool = false, theme_monitor_enabled: bool = false, + profile_anthropic_api_key: []u8 = &.{}, + + pub fn takeProfileAnthropicApiKey(self: *StartupState) []u8 { + const value = self.profile_anthropic_api_key; + self.profile_anthropic_api_key = &.{}; + return value; + } pub fn deinit(self: *StartupState, alloc: Allocator) void { self.workspace_access.deinit(alloc); @@ -400,6 +407,9 @@ fn loadStartupStateFromOwnedWorkspace( state.model_source = detailed.model_source orelse .compiled_default; state.selected_model = try loadInitialModel(alloc, configured_selection.model, null); if (hasProcessModelOverride()) state.model_source = .process_override; + if (settings.anthropic_api_key) |profile_key| { + state.profile_anthropic_api_key = try alloc.dupe(u8, profile_key); + } state.config_diagnostics = detailed.diagnostics; detailed.diagnostics = &.{}; state.prompt_history_enabled = settings.prompt_history_enabled orelse true; @@ -412,7 +422,7 @@ fn loadStartupStateFromOwnedWorkspace( mode, state.provider, settings.credential_source, - null, + settings.anthropic_api_key, ); state.credential = resolution.credential; state.stored_key_status = resolution.stored_key_status; @@ -1112,7 +1122,7 @@ fn configuredProviderSelection( settings: *const config_runtime.Settings, ) !model_provider.ProviderSelection { const provider = settings.provider orelse .gateway; - const model = settings.models.get(provider) orelse switch (provider) { + const model = settings.models.get(provider) orelse switchModelFromEnv() orelse switch (provider) { .gateway => default_model, .codex => return error.CodexModelNotSelected, .grok => return error.GrokModelNotSelected, @@ -1121,6 +1131,12 @@ fn configuredProviderSelection( return .{ .provider = provider, .model = model }; } +fn switchModelFromEnv() ?[]const u8 { + const raw = io_mod.getenv("FX_MODEL") orelse return null; + const trimmed = std.mem.trim(u8, raw, " \t\r\n"); + return if (trimmed.len > 0) trimmed else null; +} + fn initialModelId(default_model: []const u8, configured: ?[]const u8) []const u8 { const model = io_mod.getenv("FX_MODEL") orelse return configured orelse default_model; const trimmed = std.mem.trim(u8, model, " \t\r\n"); @@ -1153,6 +1169,17 @@ test "startup provider chooses only its provider-scoped model" { const grok = try configuredProviderSelection("default/model", &grok_settings); try std.testing.expectEqual(model_provider.ProviderId.grok, grok.provider); try std.testing.expectEqualStrings("grok-model", grok.model); + var anthropic_settings = config_runtime.Settings{ .provider = .anthropic }; + anthropic_settings.models.values[@intFromEnum(model_provider.ProviderId.anthropic)] = @constCast("claude-opus-5"); + const anthropic = try configuredProviderSelection("default/model", &anthropic_settings); + try std.testing.expectEqual(model_provider.ProviderId.anthropic, anthropic.provider); + try std.testing.expectEqualStrings("claude-opus-5", anthropic.model); + + const missing_anthropic = config_runtime.Settings{ .provider = .anthropic }; + try std.testing.expectError( + error.AnthropicModelNotSelected, + configuredProviderSelection("default/model", &missing_anthropic), + ); } fn loadInitialModel(alloc: Allocator, default_model: []const u8, configured: ?[]const u8) ![]u8 { From fd4ba9855fc8113fefbc8312b0d0888eea4146cd Mon Sep 17 00:00:00 2001 From: Joaquin Terrasa Date: Mon, 31 Aug 2026 12:50:01 +0200 Subject: [PATCH 5/7] fix(gateway): adapt anthropic codec tests and provider ABI call sites to main --- src/core/app/app_agent_runtime.zig | 4 ++-- src/core/app/app_auth_runtime.zig | 2 +- src/core/auth/credentials.zig | 1 + src/gateway/anthropic.zig | 12 +++++------- 4 files changed, 9 insertions(+), 10 deletions(-) diff --git a/src/core/app/app_agent_runtime.zig b/src/core/app/app_agent_runtime.zig index 8dd0610c0..efd820627 100644 --- a/src/core/app/app_agent_runtime.zig +++ b/src/core/app/app_agent_runtime.zig @@ -1065,8 +1065,8 @@ pub fn Runtime(comptime App: type) type { .permission_reviewer = tool_context.permission_reviewer_provider, }, .anthropic = .{ - .agent_stream_provider = tool_context.agent_stream_provider, - .permission_reviewer_provider = null, + .agent_stream = tool_context.agent_stream_provider, + .permission_reviewer = null, }, }; return subagent_agent_adapter.run(.{ diff --git a/src/core/app/app_auth_runtime.zig b/src/core/app/app_auth_runtime.zig index 498cdbac9..47005065a 100644 --- a/src/core/app/app_auth_runtime.zig +++ b/src/core/app/app_auth_runtime.zig @@ -1399,7 +1399,7 @@ test "interactive subscription sign-in rejects active and queued work before OAu switch (provider) { .codex => try Runtime(BusySignInApp).beginChatGptSignIn(&app), .grok => try Runtime(BusySignInApp).beginGrokSignIn(&app), - .gateway => unreachable, + .gateway, .anthropic => unreachable, } try std.testing.expectEqual(@as(usize, 0), app.auth.start_count); diff --git a/src/core/auth/credentials.zig b/src/core/auth/credentials.zig index e974488af..b3b6a319d 100644 --- a/src/core/auth/credentials.zig +++ b/src/core/auth/credentials.zig @@ -1238,6 +1238,7 @@ test "resolveForProvider anthropic never returns a gateway credential" { .refresh_if_needed, .anthropic, null, + null, ); defer if (resolution.credential) |*credential| credential.deinit(alloc); diff --git a/src/gateway/anthropic.zig b/src/gateway/anthropic.zig index 3732d19e6..8e7f63306 100644 --- a/src/gateway/anthropic.zig +++ b/src/gateway/anthropic.zig @@ -812,10 +812,10 @@ test "Anthropic request hoists system text and maps messages and tools" { }, .{ .role = .tool, .tool_call_id = "call_1", .tool_name = "read_file", .content = "contents" }, }; - const body = try agent_stream_provider.build(std.testing.allocator, .{ + const body = try buildRequest(std.testing.allocator, .{ .model = "claude-sonnet-4-5", - .serialized_tools = "[{\"type\":\"function\",\"name\":\"read_file\",\"description\":\"Read\",\"inputSchema\":{\"type\":\"object\"}}]", .messages = &messages, + .tools = .{ .additional_functions = &.{.{ .name = "read_file", .description = "Read" }} }, .tool_choice = .auto, .provider_options = .{}, .max_output_tokens = 4096, @@ -828,15 +828,14 @@ test "Anthropic request hoists system text and maps messages and tools" { try std.testing.expect(std.mem.indexOf(u8, body, "\"stream\":true") != null); try std.testing.expect(std.mem.indexOf(u8, body, "{\"type\":\"tool_use\",\"id\":\"call_1\",\"name\":\"read_file\",\"input\":{\"path\":\"README.md\"}}") != null); try std.testing.expect(std.mem.indexOf(u8, body, "{\"type\":\"tool_result\",\"tool_use_id\":\"call_1\",\"content\":\"contents\"}") != null); - try std.testing.expect(std.mem.indexOf(u8, body, "\"input_schema\":{\"type\":\"object\"}") != null); + try std.testing.expect(std.mem.indexOf(u8, body, "\"input_schema\":{\"type\":\"object\",\"properties\":{}}") != null); try std.testing.expect(std.mem.indexOf(u8, body, "\"tool_choice\":{\"type\":\"auto\"}") != null); } test "Anthropic request applies default max tokens and maps tool choice variants" { const messages = [_]types.ChatMessage{.{ .role = .user, .content = "Hello." }}; - const body = try agent_stream_provider.build(std.testing.allocator, .{ + const body = try buildRequest(std.testing.allocator, .{ .model = "claude-sonnet-4-5", - .serialized_tools = "[]", .messages = &messages, .tool_choice = .required, .provider_options = .{}, @@ -848,9 +847,8 @@ test "Anthropic request applies default max tokens and maps tool choice variants try std.testing.expect(std.mem.indexOf(u8, body, "\"tools\"") == null); try std.testing.expect(std.mem.indexOf(u8, body, "\"tool_choice\"") == null); - const none_body = try agent_stream_provider.build(std.testing.allocator, .{ + const none_body = try buildRequest(std.testing.allocator, .{ .model = "claude-sonnet-4-5", - .serialized_tools = "[]", .messages = &messages, .tool_choice = .none, .provider_options = .{}, From d53ec96d84ddd889bd842733b61efb0a546b4b0e Mon Sep 17 00:00:00 2001 From: Joaquin Terrasa Date: Mon, 31 Aug 2026 13:05:44 +0200 Subject: [PATCH 6/7] fix(lifecycle): scope FX_MODEL fallback to non-gateway providers Gateway already resolves a compiled default model, so consulting FX_MODEL before the default changed the reported configured_model and broke the startup env-override test. --- src/core/app/app_lifecycle.zig | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/src/core/app/app_lifecycle.zig b/src/core/app/app_lifecycle.zig index a5ca4c1d1..0c3d20c6a 100644 --- a/src/core/app/app_lifecycle.zig +++ b/src/core/app/app_lifecycle.zig @@ -1122,11 +1122,14 @@ fn configuredProviderSelection( settings: *const config_runtime.Settings, ) !model_provider.ProviderSelection { const provider = settings.provider orelse .gateway; - const model = settings.models.get(provider) orelse switchModelFromEnv() orelse switch (provider) { + const model = settings.models.get(provider) orelse switch (provider) { .gateway => default_model, - .codex => return error.CodexModelNotSelected, - .grok => return error.GrokModelNotSelected, - .anthropic => return error.AnthropicModelNotSelected, + .codex, .grok, .anthropic => switchModelFromEnv() orelse switch (provider) { + .codex => return error.CodexModelNotSelected, + .grok => return error.GrokModelNotSelected, + .anthropic => return error.AnthropicModelNotSelected, + .gateway => unreachable, + }, }; return .{ .provider = provider, .model = model }; } From 65f06d95ea146177ec32328af4120e75a3bd2976 Mon Sep 17 00:00:00 2001 From: Joaquin Terrasa Date: Mon, 31 Aug 2026 13:13:16 +0200 Subject: [PATCH 7/7] fix(gateway): signal request admission before anthropic delivery Requests on the anthropic wire returned ProviderAdmissionMissing at the gateway_step boundary because the codec never called admission.admit() after serialization, unlike the other transports. --- src/gateway/anthropic.zig | 1 + 1 file changed, 1 insertion(+) diff --git a/src/gateway/anthropic.zig b/src/gateway/anthropic.zig index 8e7f63306..2ba8cb55f 100644 --- a/src/gateway/anthropic.zig +++ b/src/gateway/anthropic.zig @@ -373,6 +373,7 @@ fn streamCompletionCore(alloc: Allocator, request: stream_provider.ModelRequest) connect_deadline = deadline; } } + try request.admission.admit(); var opened = try gateway_client.runBoundedHttpOperation( OpenedRequest, alloc,