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..5a2f8c2b3 100644 --- a/src/builtins/providers.zig +++ b/src/builtins/providers.zig @@ -6,6 +6,10 @@ 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 openai_compatible = @import("../gateway/openai_compatible.zig"); +const openai_compatible_models = @import("../gateway/openai_compatible_models.zig"); const provider_catalog = @import("../core/auth/provider_catalog.zig"); pub const native = provider_set.Set{ @@ -26,4 +30,16 @@ 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, + }, + .openai_compatible = .{ + .presentation = provider_catalog.find(.openai_compatible), + .agent_stream = openai_compatible.agent_stream_provider, + .cli_model_catalog = openai_compatible_models.cli_model_catalog_provider, + .model_catalog = openai_compatible_models.model_catalog_provider, + }, }; diff --git a/src/core/app/app_agent_runtime.zig b/src/core/app/app_agent_runtime.zig index 02645d783..421d0eda8 100644 --- a/src/core/app/app_agent_runtime.zig +++ b/src/core/app/app_agent_runtime.zig @@ -1064,6 +1064,14 @@ pub fn Runtime(comptime App: type) type { .agent_stream = tool_context.agent_stream_provider, .permission_reviewer = tool_context.permission_reviewer_provider, }, + .anthropic = .{ + .agent_stream = tool_context.agent_stream_provider, + .permission_reviewer = null, + }, + .openai_compatible = .{ + .agent_stream = tool_context.agent_stream_provider, + .permission_reviewer = 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..d5f177652 100644 --- a/src/core/app/app_auth_runtime.zig +++ b/src/core/app/app_auth_runtime.zig @@ -59,6 +59,8 @@ pub fn Runtime(comptime App: type) type { const required_source: credentials.Source = switch (provider) { .codex => .chatgpt_subscription, .grok => .grok_subscription, + .anthropic => .anthropic_api_key, + .openai_compatible => .stored_key, .gateway => app.auth.credentialSource() orelse .fx_login, }; const route_change = app.auth.selectForProvider(app.alloc, provider) catch |err| switch (err) { @@ -778,6 +780,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(.{ @@ -1397,7 +1400,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, .openai_compatible => unreachable, } try std.testing.expectEqual(@as(usize, 0), app.auth.start_count); diff --git a/src/core/app/app_lifecycle.zig b/src/core/app/app_lifecycle.zig index 37ea7ef85..ad3a678ba 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,6 +422,7 @@ fn loadStartupStateFromOwnedWorkspace( mode, state.provider, settings.credential_source, + settings.anthropic_api_key, ); state.credential = resolution.credential; state.stored_key_status = resolution.stored_key_status; @@ -1113,12 +1124,23 @@ fn configuredProviderSelection( const provider = settings.provider orelse .gateway; const model = settings.models.get(provider) orelse switch (provider) { .gateway => default_model, - .codex => return error.CodexModelNotSelected, - .grok => return error.GrokModelNotSelected, + .codex, .grok, .anthropic, .openai_compatible => switchModelFromEnv() orelse switch (provider) { + .codex => return error.CodexModelNotSelected, + .grok => return error.GrokModelNotSelected, + .anthropic => return error.AnthropicModelNotSelected, + .openai_compatible => return error.OpenAiModelNotSelected, + .gateway => unreachable, + }, }; 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"); @@ -1151,6 +1173,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 { diff --git a/src/core/auth/auth_runtime.zig b/src/core/auth/auth_runtime.zig index d96b188b8..1ad4f53cf 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,24 @@ pub const Runtime = struct { self, loadRuntimeCredentialSource, ), + .anthropic => if (self.credentialSource() == .anthropic_api_key) + false + else + self.selectSourceWithLoader( + alloc, + .anthropic_api_key, + self, + loadRuntimeCredentialSource, + ), + .openai_compatible => if (self.credentialSource() == .stored_key) + false + else + self.selectSourceWithLoader( + alloc, + .stored_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..0d63525c7 100644 --- a/src/core/auth/auth_transition.zig +++ b/src/core/auth/auth_transition.zig @@ -73,6 +73,8 @@ pub fn signInCompletion( .{ .switch_provider = .grok } else .{ .activate_source = .grok_subscription }, + .anthropic => .{ .activate_source = .anthropic_api_key }, + .openai_compatible => .{ .activate_source = .stored_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/auth/credentials.zig b/src/core/auth/credentials.zig index 563a11c04..2c3f76971 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,41 @@ 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_openai_credential_message = "fx needs an OpenAI-compatible API key. Set OPENAI_API_KEY (or LITELLM_API_KEY), or save a key with fx api-key."; +pub const missing_openai_interactive_credential_message = missing_openai_credential_message; 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, + .openai_compatible => if (interactive) + missing_openai_interactive_credential_message + else + missing_openai_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 +313,7 @@ pub fn resolveForProvider( mode: LoadMode, provider: model_provider.ProviderId, preferred: ?Source, + profile_anthropic_api_key: ?[]const u8, ) !Resolution { switch (provider) { .codex => { @@ -298,6 +331,8 @@ pub fn resolveForProvider( return .{ .credential = credential }; }, .gateway => {}, + .anthropic => return .{ .credential = try loadAnthropicApiKeyCredential(alloc, profile_anthropic_api_key) }, + .openai_compatible => return .{ .credential = try loadOpenAiCompatibleApiKeyCredential(alloc, secret_store) }, } return resolvePreferring( alloc, @@ -403,6 +438,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 +455,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 +464,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 +490,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 +520,45 @@ 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 loadOpenAiCompatibleApiKeyCredential(alloc: std.mem.Allocator, secret_store: host.SecretStore) !?Credential { + if (nonEmptyEnvValue("OPENAI_API_KEY")) |value| { + return .{ .token = try alloc.dupe(u8, value), .source = .stored_key }; + } + if (nonEmptyEnvValue("LITELLM_API_KEY")) |value| { + return .{ .token = try alloc.dupe(u8, value), .source = .stored_key }; + } + if (secret_store.isDisabled()) return null; + const value = (try secret_store.load(alloc)) orelse return null; + return .{ .token = value, .source = .stored_key }; +} + fn loadStoredKeyCredential( alloc: std.mem.Allocator, secret_store: host.SecretStore, @@ -661,6 +756,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 +1244,40 @@ 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, + 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..0af88fd10 100644 --- a/src/core/auth/provider_catalog.zig +++ b/src/core/auth/provider_catalog.zig @@ -37,6 +37,23 @@ 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, + }, + .{ + .id = .openai_compatible, + .slug = "openai-compatible", + .aliases = &.{"openai"}, + .name = "OpenAI Compatible", + .route_name = "OpenAI Compatible", + .description = "OpenAI-compatible Chat Completions API key via OPENAI_API_KEY", + .subscription = false, + }, }; pub fn parse(value: []const u8) ?model_provider.ProviderId { @@ -52,18 +69,23 @@ 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.expect(parse("openai-codex") == null); + try std.testing.expectEqual(model_provider.ProviderId.anthropic, parse("anthropic").?); + try std.testing.expectEqual(model_provider.ProviderId.openai_compatible, parse("openai-compatible").?); + try std.testing.expectEqual(model_provider.ProviderId.openai_compatible, parse("openai").?); 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); + try std.testing.expect(!find(.openai_compatible).subscription); } + +pub fn label(id: model_provider.ProviderId) []const u8 { + return find(id).route_name; +} + 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..88c1074e2 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,8 @@ 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", + .openai_compatible => "OpenAI Compatible is already selected.\n", }); return true; } @@ -722,6 +725,7 @@ fn activateProviderSelection( .refresh_if_needed, target, settings.credential_source, + null, ); } if (resolution.credential == null and target == .grok and caller == .provider_command) { @@ -738,6 +742,7 @@ fn activateProviderSelection( .refresh_if_needed, target, settings.credential_source, + null, ); } @@ -749,6 +754,8 @@ fn activateProviderSelection( switch (target) { .codex => "Codex credential is unavailable", .grok => "Grok credential is unavailable", + .anthropic => "Anthropic credential is unavailable", + .openai_compatible => "OpenAI Compatible credential is unavailable", .gateway => "configure a Gateway credential first", }, ); @@ -758,6 +765,8 @@ 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", + .openai_compatible => "OpenAI Compatible model catalog is unavailable", .gateway => "Gateway model catalog is unavailable", }); return false; @@ -803,13 +812,15 @@ 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, .openai_compatible => 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", + .openai_compatible => "Provider set to OpenAI Compatible.\n", }); } return true; @@ -986,6 +997,14 @@ 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; + }, + .openai_compatible => { + try writeStderr(deps, "fx login: OpenAI Compatible uses API keys; set OPENAI_API_KEY or run fx api-key openai\n"); + return .handled_failure; + }, } return .handled_success; }, @@ -1180,6 +1199,8 @@ 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", + .openai_compatible => "fx models: OpenAI Compatible model catalog is unavailable\n", }); return .handled_failure; }; diff --git a/src/core/config/config_runtime.zig b/src/core/config/config_runtime.zig index 0f9753637..3fae99dbc 100644 --- a/src/core/config/config_runtime.zig +++ b/src/core/config/config_runtime.zig @@ -39,6 +39,10 @@ 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, + openai_model: ?[]u8 = null, + grok_model: ?[]u8 = null, permission_mode: ?types.PermissionMode = null, credential_source: ?types.CredentialSource = null, yolo_acknowledged: ?bool = null, @@ -63,9 +67,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 +608,8 @@ fn isProfileOnlySettingKey(key: []const u8) bool { "models", "provider", "codex_model", + "anthropic_model", + "openai_model", "grok_model", "effort", "fast_mode", @@ -617,6 +626,7 @@ fn isProfileOnlySettingKey(key: []const u8) bool { "update_channel", "permission_mode", "credential_source", + "anthropic_api_key", "yolo_acknowledged", "permission", "additional_directories", @@ -1363,6 +1373,18 @@ 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("openai_model")) |model_value| { + if (model_value != .string) return error.InvalidOpenAiModelType; + settings_store.validateModel(model_value.string) catch return error.InvalidOpenAiModelValue; + settings.openai_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 +1414,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 +1565,33 @@ 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.openai_model) |value| { + if (target.openai_model) |current| alloc.free(current); + target.openai_model = value; + incoming.openai_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..0d8c4d38a 100644 --- a/src/core/config/model_provider.zig +++ b/src/core/config/model_provider.zig @@ -5,6 +5,8 @@ pub const ProviderId = enum { gateway, codex, grok, + anthropic, + openai_compatible, }; pub const ProviderSelection = struct { @@ -16,15 +18,29 @@ 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; + if (std.ascii.eqlIgnoreCase(value, "openai-compatible") or std.ascii.eqlIgnoreCase(value, "openai")) return .openai_compatible; return null; } +pub fn label(provider: ProviderId) []const u8 { + return switch (provider) { + .gateway => "Vercel AI Gateway", + .codex => "Codex subscription", + .grok => "Grok subscription", + .anthropic => "Anthropic", + .openai_compatible => "OpenAI Compatible", + }; +} + 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, + .openai_compatible => selected == .stored_key, }; } @@ -38,12 +54,22 @@ 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)); + try std.testing.expect(authorizesCredential(.openai_compatible, .stored_key)); + try std.testing.expect(!authorizesCredential(.openai_compatible, .anthropic_api_key)); + try std.testing.expect(!authorizesCredential(.openai_compatible, null)); } -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.expectEqual(ProviderId.openai_compatible, parse("openai-compatible").?); + try std.testing.expectEqual(ProviderId.openai_compatible, parse("OpenAI").?); 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..a801df990 100644 --- a/src/core/config/settings_store.zig +++ b/src/core/config/settings_store.zig @@ -93,6 +93,10 @@ 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, + openai_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 +118,10 @@ 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.openai_model == null and self.permission_mode == null and self.credential_source == null and !self.clear_credential_source and @@ -986,6 +994,10 @@ 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.openai_model) |value| application.changed = try putString(arena, &root.object, "openai_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")) { @@ -1552,6 +1564,8 @@ fn putModelPreference( .gateway => "model", .codex => "codex_model", .grok => "grok_model", + .anthropic => "anthropic_model", + .openai_compatible => "openai_model", }; if (root.contains(legacy_key)) { _ = root.orderedRemove(legacy_key); @@ -1814,6 +1828,14 @@ 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("openai_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/gateway/openai_json.zig b/src/core/gateway/openai_json.zig new file mode 100644 index 000000000..83aa539c7 --- /dev/null +++ b/src/core/gateway/openai_json.zig @@ -0,0 +1,190 @@ +const std = @import("std"); +const openai_tools = @import("openai_tools.zig"); +const types = @import("../shared/types.zig"); + +const Allocator = std.mem.Allocator; +const ChatMessage = types.ChatMessage; +const ToolCall = types.ToolCall; +const BuildBudget = stream_provider.BuildBudget; + +const stream_provider = @import("../agent/stream_provider.zig"); + +fn roleName(role: types.ChatRole) []const u8 { + return switch (role) { + .system => "system", + .user => "user", + .assistant => "assistant", + .tool => "tool", + }; +} + +fn validateToolMessageHistory(alloc: Allocator, messages: []const ChatMessage) !void { + for (messages) |message| { + if (message.role != .tool) continue; + if (message.tool_call_id == null or message.tool_call_id.?.len == 0) { + _ = alloc; + return error.InvalidToolMessageHistory; + } + } +} + +pub fn buildChatCompletionsBody( + alloc: Allocator, + model: []const u8, + gateway_tools_json: ?[]const u8, + messages: []const ChatMessage, + tool_choice: types.ToolChoice, + max_output_tokens: ?u32, + budget: ?BuildBudget, +) ![]u8 { + if (budget) |active| { + if (active.cancel_flag) |flag| if (flag.load(.seq_cst)) return error.Cancelled; + _ = active.deadline; + } + try validateToolMessageHistory(alloc, messages); + + const tools_json = try openai_tools.convertGatewayToolsJson(alloc, gateway_tools_json); + defer alloc.free(tools_json); + + var out: std.Io.Writer.Allocating = .init(alloc); + errdefer out.deinit(); + + try out.writer.writeAll("{\"model\":"); + try std.json.Stringify.value(model, .{}, &out.writer); + try out.writer.writeAll(",\"stream\":true,\"messages\":["); + for (messages, 0..) |message, index| { + if (index > 0) try out.writer.writeByte(','); + try writeOpenAiMessage(&out.writer, message); + } + try out.writer.writeAll("],\"tools\":"); + try out.writer.writeAll(tools_json); + try out.writer.writeAll(",\"tool_choice\":"); + try std.json.Stringify.value(tool_choice.label(), .{}, &out.writer); + if (max_output_tokens) |limit| { + try out.writer.writeAll(",\"max_tokens\":"); + try std.json.Stringify.value(limit, .{}, &out.writer); + } + try out.writer.writeByte('}'); + return try out.toOwnedSlice(); +} + +pub fn buildRequiredToolChatCompletionsBody( + alloc: Allocator, + model: []const u8, + gateway_tools_json: []const u8, + messages: []const ChatMessage, + max_output_tokens: ?u32, + budget: ?BuildBudget, +) ![]u8 { + if (budget) |active| try active.check(); + try validateToolMessageHistory(alloc, messages); + + const tools_json = try openai_tools.convertGatewayToolsJson(alloc, gateway_tools_json); + defer alloc.free(tools_json); + + var out: std.Io.Writer.Allocating = .init(alloc); + errdefer out.deinit(); + + try out.writer.writeAll("{\"model\":"); + try std.json.Stringify.value(model, .{}, &out.writer); + try out.writer.writeAll(",\"stream\":true,\"messages\":["); + for (messages, 0..) |message, index| { + if (budget) |active| try active.check(); + if (index > 0) try out.writer.writeByte(','); + try writeOpenAiMessage(&out.writer, message); + } + try out.writer.writeAll("],\"tools\":"); + try out.writer.writeAll(tools_json); + try out.writer.writeAll(",\"tool_choice\":\"required\""); + if (max_output_tokens) |limit| { + try out.writer.writeAll(",\"max_tokens\":"); + try std.json.Stringify.value(limit, .{}, &out.writer); + } + try out.writer.writeByte('}'); + return try out.toOwnedSlice(); +} + +fn writeOpenAiMessage(writer: *std.Io.Writer, message: ChatMessage) !void { + try writer.writeAll("{\"role\":"); + try std.json.Stringify.value(roleName(message.role), .{}, writer); + + switch (message.role) { + .system, .user => { + if (message.images.len != 0) return error.VisionUnavailable; + try writer.writeAll(",\"content\":"); + if (message.content) |content| { + try std.json.Stringify.value(content, .{}, writer); + } else { + try writer.writeAll("\"\""); + } + }, + .assistant => { + if (message.content) |content| { + try writer.writeAll(",\"content\":"); + try std.json.Stringify.value(content, .{}, writer); + } + if (message.tool_calls.len > 0) { + try writer.writeAll(",\"tool_calls\":["); + for (message.tool_calls, 0..) |call, index| { + if (index > 0) try writer.writeByte(','); + try writer.writeAll("{\"id\":"); + try std.json.Stringify.value(call.id, .{}, writer); + try writer.writeAll(",\"type\":\"function\",\"function\":{"); + try writer.writeAll("\"name\":"); + try std.json.Stringify.value(call.name, .{}, writer); + try writer.writeAll(",\"arguments\":"); + try std.json.Stringify.value(call.arguments_json, .{}, writer); + try writer.writeAll("}}"); + } + try writer.writeByte(']'); + } + }, + .tool => { + try writer.writeAll(",\"tool_call_id\":"); + if (message.tool_call_id) |tool_call_id| { + try std.json.Stringify.value(tool_call_id, .{}, writer); + } else { + try writer.writeAll("\"\""); + } + try writer.writeAll(",\"content\":"); + if (message.content) |content| { + try std.json.Stringify.value(content, .{}, writer); + } else { + try writer.writeAll("\"\""); + } + }, + } + try writer.writeByte('}'); +} + +test "buildChatCompletionsBody emits OpenAI chat request" { + const alloc = std.testing.allocator; + const messages = [_]ChatMessage{ + .{ .role = .user, .content = "hello" }, + }; + const tools = + \\[{"type":"function","name":"read_file","description":"Read","inputSchema":{"type":"object"}}] + ; + const body = try buildChatCompletionsBody(alloc, "gpt-4o", tools, &messages, .auto, null, null); + defer alloc.free(body); + try std.testing.expect(std.mem.find(u8, body, "\"stream\":true") != null); + try std.testing.expect(std.mem.find(u8, body, "\"messages\"") != null); + try std.testing.expect(std.mem.find(u8, body, "\"parameters\"") != null); + try std.testing.expect(std.mem.find(u8, body, "\"prompt\"") == null); +} + +test "buildChatCompletionsBody rejects vision attachments" { + const alloc = std.testing.allocator; + const image = types.ImageAttachment{ + .path = @constCast("photo.png"), + .media_type = @constCast("image/png"), + }; + const messages = [_]ChatMessage{ + .{ .role = .user, .content = "look", .images = &.{image} }, + }; + const tools = "[]"; + try std.testing.expectError( + error.VisionUnavailable, + buildChatCompletionsBody(alloc, "gpt-4o", tools, &messages, .auto, null, null), + ); +} diff --git a/src/core/gateway/openai_tools.zig b/src/core/gateway/openai_tools.zig new file mode 100644 index 000000000..4eaba9d8c --- /dev/null +++ b/src/core/gateway/openai_tools.zig @@ -0,0 +1,126 @@ +const std = @import("std"); + +const Allocator = std.mem.Allocator; + +/// Converts a Gateway function-tool JSON envelope into OpenAI Chat Completions tools. +pub fn convertGatewayToolsJson(alloc: Allocator, gateway_tools_json_opt: ?[]const u8) ![]u8 { + const gateway_tools_json = gateway_tools_json_opt orelse return alloc.dupe(u8, "[]"); + const trimmed = std.mem.trim(u8, gateway_tools_json, " \n\r\t"); + if (trimmed.len == 0) return alloc.dupe(u8, "[]"); + if (trimmed[0] != '[') return error.InvalidToolArguments; + + var parsed = try std.json.parseFromSlice(std.json.Value, alloc, trimmed, .{}); + defer parsed.deinit(); + if (parsed.value != .array) return error.InvalidToolArguments; + + var out: std.Io.Writer.Allocating = .init(alloc); + errdefer out.deinit(); + try out.writer.writeByte('['); + var first = true; + for (parsed.value.array.items) |tool| { + if (tool != .object) continue; + const tool_type = tool.object.get("type") orelse continue; + if (tool_type != .string) continue; + if (std.mem.eql(u8, tool_type.string, "provider")) continue; + if (!std.mem.eql(u8, tool_type.string, "function")) continue; + + const name = tool.object.get("name") orelse continue; + if (name != .string) continue; + const description = tool.object.get("description"); + const input_schema = tool.object.get("inputSchema") orelse continue; + + if (!first) try out.writer.writeByte(','); + first = false; + + try out.writer.writeAll("{\"type\":\"function\",\"function\":{"); + try out.writer.writeAll("\"name\":"); + try std.json.Stringify.value(name.string, .{}, &out.writer); + try out.writer.writeAll(",\"description\":"); + if (description) |desc| { + if (desc == .string) { + try std.json.Stringify.value(desc.string, .{}, &out.writer); + } else { + try out.writer.writeAll("\"\""); + } + } else { + try out.writer.writeAll("\"\""); + } + try out.writer.writeAll(",\"parameters\":"); + try std.json.Stringify.value(input_schema, .{}, &out.writer); + try out.writer.writeAll("}}"); + } + try out.writer.writeByte(']'); + return try out.toOwnedSlice(); +} + +/// Converts a Gateway function-tool JSON envelope into flat Responses API tools. +pub fn convertGatewayToolsToResponsesJson(alloc: Allocator, gateway_tools_json: []const u8) ![]u8 { + const trimmed = std.mem.trim(u8, gateway_tools_json, " \n\r\t"); + if (trimmed.len == 0) return alloc.dupe(u8, "[]"); + if (trimmed[0] != '[') return error.InvalidToolArguments; + + var parsed = try std.json.parseFromSlice(std.json.Value, alloc, trimmed, .{}); + defer parsed.deinit(); + if (parsed.value != .array) return error.InvalidToolArguments; + + var out: std.Io.Writer.Allocating = .init(alloc); + errdefer out.deinit(); + try out.writer.writeByte('['); + var first = true; + for (parsed.value.array.items) |tool| { + if (tool != .object) continue; + const tool_type = tool.object.get("type") orelse continue; + if (tool_type != .string) continue; + if (std.mem.eql(u8, tool_type.string, "provider")) continue; + if (!std.mem.eql(u8, tool_type.string, "function")) continue; + + const name = tool.object.get("name") orelse continue; + if (name != .string) continue; + const description = tool.object.get("description"); + const input_schema = tool.object.get("inputSchema") orelse continue; + + if (!first) try out.writer.writeByte(','); + first = false; + + try out.writer.writeAll("{\"type\":\"function\",\"name\":"); + try std.json.Stringify.value(name.string, .{}, &out.writer); + try out.writer.writeAll(",\"description\":"); + if (description) |desc| { + if (desc == .string) { + try std.json.Stringify.value(desc.string, .{}, &out.writer); + } else { + try out.writer.writeAll("\"\""); + } + } else { + try out.writer.writeAll("\"\""); + } + try out.writer.writeAll(",\"parameters\":"); + try std.json.Stringify.value(input_schema, .{}, &out.writer); + try out.writer.writeByte('}'); + } + try out.writer.writeByte(']'); + return try out.toOwnedSlice(); +} + +test "convertGatewayToolsJson maps function envelope to OpenAI tools" { + const alloc = std.testing.allocator; + const gateway = + \\[{"type":"function","name":"read_file","description":"Read a file","inputSchema":{"type":"object","properties":{"path":{"type":"string"}}}}] + ; + const openai = try convertGatewayToolsJson(alloc, gateway); + defer alloc.free(openai); + try std.testing.expect(std.mem.find(u8, openai, "\"parameters\"") != null); + try std.testing.expect(std.mem.find(u8, openai, "inputSchema") == null); + try std.testing.expect(std.mem.find(u8, openai, "read_file") != null); +} + +test "convertGatewayToolsJson drops provider-executed tools" { + const alloc = std.testing.allocator; + const gateway = + \\[{"type":"provider","id":"gateway.perplexity_search","name":"perplexity_search"},{"type":"function","name":"read_file","description":"Read","inputSchema":{"type":"object"}}] + ; + const openai = try convertGatewayToolsJson(alloc, gateway); + defer alloc.free(openai); + try std.testing.expect(std.mem.find(u8, openai, "perplexity_search") == null); + try std.testing.expect(std.mem.find(u8, openai, "read_file") != null); +} diff --git a/src/core/gateway/openai_transport.zig b/src/core/gateway/openai_transport.zig new file mode 100644 index 000000000..8108da7c1 --- /dev/null +++ b/src/core/gateway/openai_transport.zig @@ -0,0 +1,153 @@ +const std = @import("std"); +const io_mod = @import("../shared/io.zig"); + +pub const gateway_chat_url_env = "FX_GATEWAY_CHAT_URL"; + +pub const openai_api_key_env = "OPENAI_API_KEY"; +pub const litellm_api_key_env = "LITELLM_API_KEY"; +pub const openai_base_url_env = "FX_OPENAI_BASE_URL"; +pub const default_base_url = "https://api.openai.com/v1"; +pub const models_path = "/v1/models"; +pub const chat_completions_suffix = "/chat/completions"; + +pub const e2e_openai_chat_url_env = "FX_E2E_OPENAI_CHAT_URL"; +pub const e2e_openai_models_url_env = "FX_E2E_OPENAI_MODELS_URL"; + +pub const OpenAiSettings = struct { + openai_base_url: ?[]const u8 = null, + openai_api_key: ?[]const u8 = null, +}; + +fn nonEmptyEnv(name: []const u8) ?[]const u8 { + const raw = io_mod.getenv(name) orelse return null; + if (std.mem.trim(u8, raw, " \t\r\n").len == 0) return null; + return raw; +} + +pub fn resolveOpenAiBaseUrlFromSettings(settings: OpenAiSettings) []const u8 { + if (nonEmptyEnv(openai_base_url_env)) |value| return value; + if (settings.openai_base_url) |url| { + if (std.mem.trim(u8, url, " \t\r\n").len > 0) return url; + } + return default_base_url; +} + +pub fn openAiApiKeyConfigured(settings: OpenAiSettings) bool { + if (nonEmptyEnv(openai_api_key_env) != null) return true; + if (nonEmptyEnv(litellm_api_key_env) != null) return true; + if (settings.openai_api_key) |key| { + if (std.mem.trim(u8, key, " \t\r\n").len > 0) return true; + } + return false; +} + +pub fn resolveGatewayChatUrl(fallback: []const u8, override: ?[]const u8) []const u8 { + const candidate = override orelse return fallback; + if (!isLoopbackHttpUrl(candidate)) return fallback; + return candidate; +} + +fn isLoopbackHttpUrl(url: []const u8) bool { + const uri = std.Uri.parse(url) catch return false; + if (!std.ascii.eqlIgnoreCase(uri.scheme, "http") or + uri.user != null or + uri.password != null or + uri.port == null) + { + return false; + } + + const host_component = uri.host orelse return false; + var host_buf: [std.Io.net.HostName.max_len]u8 = undefined; + const host = host_component.toRaw(&host_buf) catch return false; + return std.mem.eql(u8, host, "127.0.0.1") or + std.ascii.eqlIgnoreCase(host, "localhost") or + std.mem.eql(u8, host, "[::1]"); +} + +pub fn selectE2eWireUrl( + e2e_env: ?[]const u8, + fallback: []const u8, +) []const u8 { + const override = e2e_env orelse return fallback; + if (!isLoopbackHttpUrl(override)) return fallback; + return override; +} + +pub fn formatChatUrl(buf: []u8, base_url: []const u8) ![]const u8 { + const trimmed = std.mem.trimEnd(u8, base_url, "/"); + if (std.mem.endsWith(u8, trimmed, chat_completions_suffix)) { + if (trimmed.len > buf.len) return error.PathTooLong; + @memcpy(buf[0..trimmed.len], trimmed); + return buf[0..trimmed.len]; + } + const path_suffix = if (trimmed.len == 0) chat_completions_suffix else "chat/completions"; + const total = if (trimmed.len == 0) + path_suffix.len + else + trimmed.len + 1 + path_suffix.len; + if (total > buf.len) return error.PathTooLong; + if (trimmed.len == 0) { + @memcpy(buf[0..path_suffix.len], path_suffix); + return buf[0..path_suffix.len]; + } + @memcpy(buf[0..trimmed.len], trimmed); + buf[trimmed.len] = '/'; + @memcpy(buf[trimmed.len + 1 .. trimmed.len + 1 + path_suffix.len], path_suffix); + return buf[0 .. trimmed.len + 1 + path_suffix.len]; +} + +pub const max_streamed_tool_index: usize = 64; + +pub fn formatModelsUrl(alloc: std.mem.Allocator, base_url: []const u8) ![]u8 { + const trimmed = std.mem.trimEnd(u8, base_url, "/"); + if (std.mem.endsWith(u8, trimmed, "/models")) { + return alloc.dupe(u8, trimmed); + } + const suffix = if (std.mem.endsWith(u8, trimmed, "/v1")) "/models" else models_path; + if (std.mem.endsWith(u8, trimmed, "/")) { + return std.fmt.allocPrint(alloc, "{s}{s}", .{ trimmed, suffix[1..] }); + } + return std.fmt.allocPrint(alloc, "{s}{s}", .{ trimmed, suffix }); +} + +test "resolveOpenAiBaseUrlFromSettings uses profile when env unset" { + const stable = try stableOpenAiTransportTestEnviron(); + io_mod.setEnvironMap(stable); + + try std.testing.expectEqualStrings( + "https://litellm.example/v1", + resolveOpenAiBaseUrlFromSettings(.{ .openai_base_url = "https://litellm.example/v1" }), + ); +} + +test "formatChatUrl composes base and suffix" { + var buf: [128]u8 = undefined; + const official = try formatChatUrl(&buf, "https://api.openai.com/v1"); + try std.testing.expectEqualStrings("https://api.openai.com/v1/chat/completions", official); + + const ollama = try formatChatUrl(&buf, "http://127.0.0.1:11434/v1/"); + try std.testing.expectEqualStrings("http://127.0.0.1:11434/v1/chat/completions", ollama); + + const already = try formatChatUrl(&buf, "http://127.0.0.1:1/v1/chat/completions"); + try std.testing.expectEqualStrings("http://127.0.0.1:1/v1/chat/completions", already); +} + +test "formatModelsUrl composes OpenAI models endpoint" { + const alloc = std.testing.allocator; + const url = try formatModelsUrl(alloc, "https://api.openai.com/v1"); + defer alloc.free(url); + try std.testing.expectEqualStrings("https://api.openai.com/v1/models", url); +} + +var stable_openai_transport_test_environ: ?*std.process.Environ.Map = null; + +fn stableOpenAiTransportTestEnviron() !*const std.process.Environ.Map { + if (stable_openai_transport_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_openai_transport_test_environ = map; + return map; +} diff --git a/src/core/gateway/provider_set.zig b/src/core/gateway/provider_set.zig index 97d921acb..9f42bd76d 100644 --- a/src/core/gateway/provider_set.zig +++ b/src/core/gateway/provider_set.zig @@ -51,12 +51,16 @@ pub const Set = struct { gateway: Bundle, codex: Bundle, grok: Bundle, + anthropic: Bundle, + openai_compatible: 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, + .openai_compatible => self.openai_compatible, }; } @@ -65,6 +69,8 @@ pub const Set = struct { .gateway = self.gateway.deferred_usage, .codex = self.codex.deferred_usage, .grok = self.grok.deferred_usage, + .anthropic = self.anthropic.deferred_usage, + .openai_compatible = self.openai_compatible.deferred_usage, }; } }; @@ -74,6 +80,8 @@ pub fn gateway_only(gateway: Bundle) Set { .gateway = gateway, .codex = .{}, .grok = .{}, + .anthropic = .{}, + .openai_compatible = .{}, }; } @@ -144,7 +152,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 = .{}, .openai_compatible = .{} }; 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..8693a891a 100644 --- a/src/core/output/output_contracts.zig +++ b/src/core/output/output_contracts.zig @@ -852,6 +852,8 @@ pub const ModelListSnapshot = struct { .gateway => "gateway", .codex => provider_catalog.label(.codex), .grok => provider_catalog.label(.grok), + .anthropic => provider_catalog.label(.anthropic), + .openai_compatible => provider_catalog.label(.openai_compatible), }; } diff --git a/src/core/session/generation_usage_provider.zig b/src/core/session/generation_usage_provider.zig index 19ad021b1..d74f5a1b5 100644 --- a/src/core/session/generation_usage_provider.zig +++ b/src/core/session/generation_usage_provider.zig @@ -85,6 +85,8 @@ pub const Set = struct { gateway: ?Provider = null, codex: ?Provider = null, grok: ?Provider = null, + anthropic: ?Provider = null, + openai_compatible: ?Provider = null, pub fn gatewayOnly(provider: Provider) Set { return .{ .gateway = provider }; @@ -95,6 +97,8 @@ pub const Set = struct { .gateway => self.gateway, .codex => self.codex, .grok => self.grok, + .anthropic => self.anthropic, + .openai_compatible => self.openai_compatible, }; } }; diff --git a/src/core/session/session_usage.zig b/src/core/session/session_usage.zig index 1d354e746..f8dc02be1 100644 --- a/src/core/session/session_usage.zig +++ b/src/core/session/session_usage.zig @@ -3272,6 +3272,8 @@ fn exactUsageOrigin(provider: model_provider.ProviderId) []const u8 { .gateway => "exact/gateway", .codex => "exact/codex", .grok => "exact/grok", + .anthropic => "exact/anthropic", + .openai_compatible => "exact/openai_compatible", }; } 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, 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 new file mode 100644 index 000000000..2ba8cb55f --- /dev/null +++ b/src/gateway/anthropic.zig @@ -0,0 +1,1040 @@ +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 model_tool_schema = @import("../core/tooling/model_tool_schema.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{ + .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; + } +} + +pub fn buildRequest( + alloc: Allocator, + request: stream_provider.RequestData, +) ![]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.tools); + 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, + tools: stream_provider.ToolSelection, +) !usize { + var count: usize = 0; + var tools_out: std.Io.Writer.Allocating = .init(alloc); + defer tools_out.deinit(); + try tools_out.writer.writeAll(",\"tools\":["); + + 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 (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 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 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; +} + +fn streamCompletion( + _: ?*anyopaque, + alloc: Allocator, + request: stream_provider.ModelRequest, +) !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.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); +} + +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.ModelRequest) !stream_provider.Result { + if (request.cancel_flag.load(.seq_cst)) return error.Cancelled; + 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; + } 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.credential.secret }; + 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.credential.secret, + .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; + } + } + try request.admission.admit(); + 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 = 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(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 .{ .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, + &events, + EventBridge.content, + EventBridge.toolStart, + EventBridge.reasoning, + EventBridge.toolInput, + request.cancel_flag, + request.content_capture_limit, + ); + 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, + 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.ModelCompletion { + 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 buildRequest(std.testing.allocator, .{ + .model = "claude-sonnet-4-5", + .messages = &messages, + .tools = .{ .additional_functions = &.{.{ .name = "read_file", .description = "Read" }} }, + .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\",\"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 buildRequest(std.testing.allocator, .{ + .model = "claude-sonnet-4-5", + .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 buildRequest(std.testing.allocator, .{ + .model = "claude-sonnet-4-5", + .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.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)); + 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; +} diff --git a/src/gateway/openai_client.zig b/src/gateway/openai_client.zig new file mode 100644 index 000000000..d91d5741f --- /dev/null +++ b/src/gateway/openai_client.zig @@ -0,0 +1,400 @@ +const std = @import("std"); +const openai_transport = @import("../core/gateway/openai_transport.zig"); +const types = @import("../core/shared/types.zig"); +const stream_provider = @import("../core/agent/stream_provider.zig"); + +const Allocator = std.mem.Allocator; +const max_sse_event_line_bytes: usize = 4 * 1024 * 1024; + +/// Consumes an OpenAI chat-completions SSE stream and reduces chunk events +/// into a model completion, emitting deltas through the provider event sink. +pub fn consumeOpenAiSse( + alloc: Allocator, + reader: anytype, + events: *stream_provider.EventSink, + cancel_flag: *std.atomic.Value(bool), + content_capture_limit: ?usize, +) !types.ModelCompletion { + var content_buf: std.ArrayList(u8) = .empty; + errdefer content_buf.deinit(alloc); + + var streamed_tools: std.ArrayList(StreamedToolCall) = .empty; + defer { + for (streamed_tools.items) |*tool| tool.deinit(alloc); + streamed_tools.deinit(alloc); + } + + var finish_reason: ?types.ProviderFinishReason = null; + + var event_reader = SseEventReader{ .max_line_bytes = max_sse_event_line_bytes }; + defer event_reader.deinit(alloc); + + while (true) { + if (cancel_flag.load(.seq_cst)) return error.Cancelled; + + const event = try event_reader.next(alloc, reader); + defer event_reader.releaseLine(); + + switch (event) { + .data => |json_text| { + try handleOpenAiChunk( + alloc, + json_text, + &content_buf, + &streamed_tools, + &finish_reason, + events, + content_capture_limit, + ); + }, + .done => break, + .ignored => continue, + .read_failed => { + if (cancel_flag.load(.seq_cst)) return error.Cancelled; + return error.ReadFailed; + }, + .eof => break, + } + } + + if (finish_reason == null) { + if (streamed_tools.items.len > 0) { + finish_reason = .tool_calls; + } else if (content_buf.items.len > 0) { + finish_reason = .stop; + } + } + + const owned_content: ?[]u8 = if (content_buf.items.len > 0) try content_buf.toOwnedSlice(alloc) else null; + if (owned_content != null) content_buf = .empty; + errdefer if (owned_content) |value| alloc.free(value); + + const owned_tools: []types.ToolCall = if (streamed_tools.items.len > 0) + try alloc.alloc(types.ToolCall, streamed_tools.items.len) + else + &.{}; + errdefer if (owned_tools.len > 0) alloc.free(owned_tools); + const initialized: usize = 0; + _ = &initialized; + errdefer for (owned_tools[0..initialized]) |call| { + alloc.free(call.id); + alloc.free(call.name); + alloc.free(call.arguments_json); + }; + var count: usize = 0; + for (streamed_tools.items) |tool| { + if (tool.id.items.len == 0 or tool.name.items.len == 0) continue; + owned_tools[count] = try dupeStreamedToolCall(alloc, tool); + count += 1; + } + + return .{ + .content = owned_content, + .tool_calls = owned_tools[0..count], + .finish_reason = finish_reason orelse if (count > 0) .tool_calls else .stop, + }; +} + +fn emitContent(events: *stream_provider.EventSink, chunk: []const u8) void { + events.emit(.{ .content_delta = chunk }); +} + +fn emitToolStart(events: *stream_provider.EventSink, id: []const u8, name: []const u8) void { + events.emit(.{ .tool_started = .{ .id = id, .name = name, .label = null } }); +} + +fn emitToolInput(events: *stream_provider.EventSink, chunk: []const u8) void { + events.emit(.{ .tool_input_delta = chunk }); +} + +const SseEvent = union(enum) { + data: []const u8, + done, + ignored, + read_failed, + eof, +}; + +const SseEventReader = struct { + pending_line: std.ArrayList(u8) = .empty, + max_line_bytes: usize, + + fn deinit(self: *@This(), alloc: Allocator) void { + self.pending_line.deinit(alloc); + } + + fn releaseLine(self: *@This()) void { + self.pending_line.clearRetainingCapacity(); + } + + fn next(self: *@This(), alloc: Allocator, reader: anytype) !SseEvent { + const line = switch (try self.readLine(alloc, reader)) { + .line => |value| value, + .read_failed => return .read_failed, + .eof => return .eof, + }; + + const trimmed = std.mem.trimEnd(u8, line, "\r"); + if (trimmed.len == 0) return .ignored; + if (trimmed[0] == ':') return .ignored; + const data_prefix = "data: "; + if (!std.mem.startsWith(u8, trimmed, data_prefix)) return .ignored; + const json_text = trimmed[data_prefix.len..]; + if (std.mem.eql(u8, json_text, "[DONE]")) return .done; + return .{ .data = json_text }; + } + + fn readLine(self: *@This(), alloc: Allocator, reader: anytype) !union(enum) { + line: []const u8, + read_failed, + eof, + } { + while (true) { + const fragment = reader.takeDelimiter('\n') catch |err| switch (err) { + error.StreamTooLong => { + const buffered = reader.buffered(); + if (buffered.len == 0) return error.OpenAiSseReadStalled; + if (buffered.len > self.max_line_bytes - self.pending_line.items.len) { + return error.OpenAiSseEventTooLarge; + } + try self.pending_line.appendSlice(alloc, buffered); + reader.tossBuffered(); + continue; + }, + error.ReadFailed => return .read_failed, + } orelse { + if (self.pending_line.items.len > 0) return .{ .line = self.pending_line.items }; + return .eof; + }; + + if (fragment.len > self.max_line_bytes - self.pending_line.items.len) { + return error.OpenAiSseEventTooLarge; + } + if (self.pending_line.items.len == 0) return .{ .line = fragment }; + try self.pending_line.appendSlice(alloc, fragment); + return .{ .line = self.pending_line.items }; + } + } +}; + +const StreamedToolCall = struct { + id: std.ArrayList(u8) = .empty, + name: std.ArrayList(u8) = .empty, + arguments: std.ArrayList(u8) = .empty, + announced: bool = false, + + fn deinit(self: *@This(), alloc: Allocator) void { + self.id.deinit(alloc); + self.name.deinit(alloc); + self.arguments.deinit(alloc); + } +}; + +fn dupeStreamedToolCall(alloc: Allocator, tool: StreamedToolCall) !types.ToolCall { + const args = if (tool.arguments.items.len == 0) "{}" else tool.arguments.items; + if (try types.ToolArgumentIntegrity.classifySerialized(alloc, args) == .malformed_json) { + return error.InvalidOpenAiResponse; + } + const id = try alloc.dupe(u8, tool.id.items); + errdefer alloc.free(id); + const name = try alloc.dupe(u8, tool.name.items); + errdefer alloc.free(name); + const arguments_json = try alloc.dupe(u8, args); + errdefer alloc.free(arguments_json); + return .{ + .id = id, + .name = name, + .arguments_json = arguments_json, + }; +} + +fn handleOpenAiChunk( + alloc: Allocator, + json_text: []const u8, + content_buf: *std.ArrayList(u8), + streamed_tools: *std.ArrayList(StreamedToolCall), + finish_reason: *?types.ProviderFinishReason, + events: *stream_provider.EventSink, + content_capture_limit: ?usize, +) !void { + var parsed = std.json.parseFromSlice(std.json.Value, alloc, json_text, .{}) catch return; + defer parsed.deinit(); + if (parsed.value != .object) return; + const choices = parsed.value.object.get("choices") orelse return; + if (choices != .array or choices.array.items.len == 0) return; + const choice = choices.array.items[0]; + if (choice != .object) return; + + if (choice.object.get("finish_reason")) |reason_value| { + if (reason_value == .string and reason_value.string.len > 0) { + finish_reason.* = types.ProviderFinishReason.parse_legacy(reason_value.string); + } + } + + const delta = choice.object.get("delta") orelse return; + if (delta != .object) return; + + if (delta.object.get("content")) |content_value| { + if (content_value == .string and content_value.string.len > 0) { + const retained = if (content_capture_limit) |limit| + content_value.string[0..@min(content_value.string.len, limit -| content_buf.items.len)] + else + content_value.string; + try content_buf.appendSlice(alloc, retained); + emitContent(events, content_value.string); + } + } + + const tool_calls = delta.object.get("tool_calls") orelse return; + if (tool_calls != .array) return; + + for (tool_calls.array.items) |tool_call| { + if (tool_call != .object) continue; + const index_value = tool_call.object.get("index") orelse continue; + const index = validatedStreamedToolIndex(index_value) orelse return error.InvalidOpenAiResponse; + while (streamed_tools.items.len <= index) { + try streamed_tools.append(alloc, .{}); + } + const record = &streamed_tools.items[index]; + + if (tool_call.object.get("id")) |id_value| { + if (id_value == .string and id_value.string.len > 0) { + record.id.clearRetainingCapacity(); + try record.id.appendSlice(alloc, id_value.string); + } + } + + const function = tool_call.object.get("function") orelse continue; + if (function != .object) continue; + + if (function.object.get("name")) |name_value| { + if (name_value == .string and name_value.string.len > 0) { + record.name.clearRetainingCapacity(); + try record.name.appendSlice(alloc, name_value.string); + if (!record.announced and record.id.items.len > 0) { + record.announced = true; + emitToolStart(events, record.id.items, record.name.items); + } + } + } + + if (function.object.get("arguments")) |args_value| { + if (args_value == .string and args_value.string.len > 0) { + try record.arguments.appendSlice(alloc, args_value.string); + emitToolInput(events, args_value.string); + } + } + } +} + +fn validatedStreamedToolIndex(index_value: std.json.Value) ?usize { + if (index_value != .integer) return null; + if (index_value.integer < 0) return null; + const index = std.math.cast(usize, index_value.integer) orelse return null; + if (index > openai_transport.max_streamed_tool_index) return null; + return index; +} + +const RecordedToolStart = struct { + id: []u8, + name: []u8, +}; + +const TestEventSink = struct { + sink: stream_provider.EventSink, + contents: std.ArrayList(u8) = .empty, + tool_inputs: std.ArrayList([]u8) = .empty, + tool_starts: std.ArrayList(RecordedToolStart) = .empty, + + fn init() TestEventSink { + return .{ .sink = .{ + .context = undefined, + .emit_fn = emit, + } }; + } + + fn deinit(self: *TestEventSink) void { + const alloc = std.testing.allocator; + self.contents.deinit(alloc); + for (self.tool_inputs.items) |item| alloc.free(item); + self.tool_inputs.deinit(alloc); + for (self.tool_starts.items) |item| { + alloc.free(item.id); + alloc.free(item.name); + } + self.tool_starts.deinit(alloc); + } + + fn emit(context: *anyopaque, event: stream_provider.Event) void { + const self: *TestEventSink = @ptrCast(@alignCast(context)); + const alloc = std.testing.allocator; + switch (event) { + .content_delta => |chunk| self.contents.appendSlice(alloc, chunk) catch {}, + .tool_input_delta => |chunk| { + const owned = alloc.dupe(u8, chunk) catch return; + self.tool_inputs.append(alloc, owned) catch { + alloc.free(owned); + }; + }, + .tool_started => |start| { + self.tool_starts.append(alloc, .{ + .id = alloc.dupe(u8, start.id) catch return, + .name = alloc.dupe(u8, start.name) catch return, + }) catch {}; + }, + .reasoning_delta => {}, + } + } +}; + +test "handleOpenAiChunk accumulates streamed tool call fragments" { + const testing = std.testing; + var content: std.ArrayList(u8) = .empty; + defer content.deinit(testing.allocator); + var tools: std.ArrayList(StreamedToolCall) = .empty; + defer { + for (tools.items) |*tool| tool.deinit(testing.allocator); + tools.deinit(testing.allocator); + } + var finish: ?types.ProviderFinishReason = null; + + var sink = TestEventSink.init(); + defer sink.deinit(); + sink.sink.context = @ptrCast(&sink); + + const chunk1 = + \\{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"read_file","arguments":"{\"path\":"}}]}}]} + ; + try handleOpenAiChunk( + testing.allocator, + chunk1, + &content, + &tools, + &finish, + &sink.sink, + null, + ); + + const chunk2 = + \\{"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"\"/tmp\"}"}}]}}]} + ; + try handleOpenAiChunk( + testing.allocator, + chunk2, + &content, + &tools, + &finish, + &sink.sink, + null, + ); + + try testing.expectEqual(@as(usize, 1), tools.items.len); + try testing.expectEqualStrings("call_1", tools.items[0].id.items); + try testing.expectEqualStrings("read_file", tools.items[0].name.items); + try testing.expectEqualStrings("{\"path\":\"/tmp\"}", tools.items[0].arguments.items); + try testing.expectEqual(@as(usize, 1), sink.tool_starts.items.len); + try testing.expectEqualStrings("call_1", sink.tool_starts.items[0].id); + try testing.expectEqual(@as(usize, 2), sink.tool_inputs.items.len); + try testing.expectEqualStrings("{\"path\":", sink.tool_inputs.items[0]); + try testing.expectEqualStrings("\"/tmp\"}", sink.tool_inputs.items[1]); +} diff --git a/src/gateway/openai_compatible.zig b/src/gateway/openai_compatible.zig new file mode 100644 index 000000000..4f2256e3b --- /dev/null +++ b/src/gateway/openai_compatible.zig @@ -0,0 +1,252 @@ +const std = @import("std"); +const openai_json = @import("../core/gateway/openai_json.zig"); +const stream_provider = @import("../core/agent/stream_provider.zig"); +const io_mod = @import("../core/shared/io.zig"); +const gateway_client = @import("client.zig"); +const secret = @import("../core/auth/secret.zig"); +const openai_client = @import("openai_client.zig"); + +const Allocator = std.mem.Allocator; +const max_error_body_bytes: usize = 1024 * 1024; +const transfer_buffer_bytes: usize = 256 * 1024; +const connect_timeout_ms: i64 = 30_000; +const default_base_url = "https://api.openai.com/v1"; +pub const base_url_env = "FX_OPENAI_BASE_URL"; +pub const e2e_endpoint_env = "FX_E2E_OPENAI_CHAT_URL"; + +pub const agent_stream_provider = stream_provider.Provider{ + .context = null, + .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 chatUrl(alloc: Allocator) ![]u8 { + const base = std.mem.trimEnd(u8, resolveBaseUrl(), "/"); + if (std.mem.endsWith(u8, base, "/chat/completions")) return alloc.dupe(u8, base); + return std.fmt.allocPrint(alloc, "{s}/chat/completions", .{base}); +} + +fn validateModel(model: []const u8) !void { + if (model.len == 0 or model.len > 1024) return error.InvalidOpenAiModel; + for (model) |byte| { + if (byte <= 0x20 or byte == 0x7f) return error.InvalidOpenAiModel; + } +} + +pub fn buildRequest( + alloc: Allocator, + request: stream_provider.RequestData, +) ![]u8 { + try validateModel(request.model); + if (request.budget) |budget| { + if (budget.cancel_flag) |flag| if (flag.load(.seq_cst)) return error.Cancelled; + _ = budget.deadline; + } + return openai_json.buildChatCompletionsBody( + alloc, + request.model, + null, + request.messages, + request.tool_choice, + request.max_output_tokens, + if (request.budget) |budget| + .{ .deadline = budget.deadline, .cancel_flag = budget.cancel_flag } + else + null, + ); +} + +fn streamCompletion( + _: ?*anyopaque, + alloc: Allocator, + request: stream_provider.ModelRequest, +) !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.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); +} + +const OpenedRequest = struct { + request: ?std.http.Client.Request, + + pub fn deinit(self: *OpenedRequest, _: Allocator) void { + if (self.request) |*req| req.deinit(); + self.request = null; + } + + pub fn take(self: *OpenedRequest) std.http.Client.Request { + const req = self.request.?; + self.request = null; + return req; + } +}; + +const OpenRequestOperation = struct { + client: *std.http.Client, + uri: std.Uri, + auth_header: []const u8, + + pub fn run(self: *@This()) !OpenedRequest { + return .{ .request = try self.client.request(.POST, self.uri, .{ + .headers = .{ + .content_type = .{ .override = "application/json" }, + .authorization = .{ .override = self.auth_header }, + .accept_encoding = .omit, + .user_agent = .{ .override = gateway_client.user_agent }, + }, + .extra_headers = &.{ + .{ .name = "accept", .value = "text/event-stream" }, + }, + .keep_alive = false, + .redirect_behavior = .unhandled, + }) }; + } +}; + +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| { + if (source != .stored_key) return error.OpenAiCompatibleCredentialRequired; + } else { + return error.OpenAiCompatibleCredentialRequired; + } + 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.InvalidE2EOpenAiEndpoint; + break :endpoint override; + } else try chatUrl(alloc); + defer if (io_mod.getenv(e2e_endpoint_env) == null) alloc.free(@constCast(request_endpoint)); + const uri = try std.Uri.parse(request_endpoint); + + const auth_header = try std.fmt.allocPrint(alloc, "Bearer {s}", .{request.credential.secret}); + defer secret.zeroAndFree(alloc, @constCast(auth_header)); + + var client: std.http.Client = .{ .allocator = alloc, .io = io_mod.getIo() }; + defer client.deinit(); + var open_operation = OpenRequestOperation{ + .client = &client, + .uri = uri, + .auth_header = auth_header, + }; + 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; + } + } + try request.admission.admit(); + 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 = 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(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, "OpenAI-compatible 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, "OpenAI-compatible error response exceeded the local limit"); + } else bounded_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 openai_client.consumeOpenAiSse( + alloc, + reader, + &events, + request.cancel_flag, + request.content_capture_limit, + ); + return .{ .completed = .{ + .completion = completion, + .ownership = .owned, + } }; +} + +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, + }; +} diff --git a/src/gateway/openai_compatible_models.zig b/src/gateway/openai_compatible_models.zig new file mode 100644 index 000000000..269dc3d11 --- /dev/null +++ b/src/gateway/openai_compatible_models.zig @@ -0,0 +1,250 @@ +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 openai_transport = @import("../core/gateway/openai_transport.zig"); +const secret = @import("../core/auth/secret.zig"); +const io_mod = @import("../core/shared/io.zig"); +const gateway_client = @import("client.zig"); +const openai_compatible = @import("openai_compatible.zig"); + +const Allocator = std.mem.Allocator; +const max_catalog_models: usize = 512; +const max_model_id_bytes: usize = 1024; +const max_catalog_bytes: usize = 4 * 1024 * 1024; +const fetch_timeout_ms: i64 = 30_000; + +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() != .stored_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.OpenAiModelCatalogTooLarge) return .{ .category = .malformed_response }; + return .{ .category = .transport, .retryable = true }; +} + +const FetchResponse = struct { + status: std.http.Status, + body: []u8, + + pub fn deinit(self: *FetchResponse, alloc: Allocator) void { + secret.zeroAndFree(alloc, self.body); + self.* = undefined; + } +}; + +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 secret.zeroAndFree(self.alloc, auth_header); + const body_buffer = try self.alloc.alloc(u8, max_catalog_bytes + 1); + defer secret.zeroAndFree(self.alloc, body_buffer); + var response_writer = std.Io.Writer.fixed(body_buffer); + 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 = &.{ + .{ .name = "accept", .value = "application/json" }, + }, + .response_writer = &response_writer, + .redirect_behavior = .unhandled, + }) catch |err| switch (err) { + error.WriteFailed => return error.OpenAiModelCatalogTooLarge, + else => return err, + }; + const body = response_writer.buffered(); + if (body.len > max_catalog_bytes) return error.OpenAiModelCatalogTooLarge; + return .{ + .status = result.status, + .body = try self.alloc.dupe(u8, body), + }; + } +}; + +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.InvalidE2EOpenAiModelsEndpoint; + return alloc.dupe(u8, override); + } + const formatted = try openai_transport.formatModelsUrl(alloc, openai_compatible.resolveBaseUrl()); + errdefer alloc.free(formatted); + return formatted; +} + +pub const e2e_models_endpoint_env = "FX_E2E_OPENAI_MODELS_URL"; + +fn parseCatalog( + alloc: Allocator, + json_text: []const u8, +) !std.ArrayList(model_catalog.ModelCatalogEntry) { + var parsed = try std.json.parseFromSlice(std.json.Value, alloc, json_text, .{}); + defer parsed.deinit(); + if (parsed.value != .object) return error.InvalidOpenAiModelCatalog; + const models_value = parsed.value.object.get("data") orelse + return error.InvalidOpenAiModelCatalog; + if (models_value != .array or models_value.array.items.len > max_catalog_models) { + return error.InvalidOpenAiModelCatalog; + } + + var catalog: std.ArrayList(model_catalog.ModelCatalogEntry) = .empty; + errdefer model_catalog.freeModelCatalog(alloc, &catalog); + for (models_value.array.items) |value| { + if (value != .object) return error.InvalidOpenAiModelCatalog; + const id = try requiredString(value.object, "id"); + try validateModelId(id); + const owned_id = try alloc.dupe(u8, id); + errdefer alloc.free(owned_id); + const model_type = try alloc.dupe(u8, "language"); + errdefer alloc.free(model_type); + try catalog.append(alloc, .{ + .id = owned_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.InvalidOpenAiModelCatalog; + if (value != .string or value.string.len == 0) return error.InvalidOpenAiModelCatalog; + return value.string; +} + +fn validateModelId(id: []const u8) !void { + if (id.len == 0 or id.len > max_model_id_bytes) return error.InvalidOpenAiModelCatalog; + for (id) |byte| { + if (byte <= 0x20 or byte == 0x7f) return error.InvalidOpenAiModelCatalog; + } +} + +test "OpenAI-compatible catalog parser accepts standard models payload" { + const alloc = std.testing.allocator; + const json = + \\{"data":[ + \\ {"id":"gpt-test","object":"model"}, + \\ {"id":"gpt-4o","object":"model"} + \\]} + ; + var catalog = try parseCatalog(alloc, json); + defer model_catalog.freeModelCatalog(alloc, &catalog); + try std.testing.expectEqual(@as(usize, 2), catalog.items.len); + try std.testing.expectEqualStrings("gpt-test", catalog.items[0].id); +} + +test "OpenAI-compatible catalog rejects missing credentials" { + const result = try fetchCatalogForProvider(null, std.testing.allocator, .{ + .access = .{ .public_only = .no_credential }, + .endpoint = "/v1/models", + }); + try std.testing.expect(result.failure.category == .authentication); +} 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;