From 02c4bc2d7f83f73f282bb542505c98446bc41bce Mon Sep 17 00:00:00 2001 From: Colton Padden Date: Mon, 31 Aug 2026 13:56:15 -0400 Subject: [PATCH] Add direct OpenAI Responses provider --- README.md | 13 +- src/builtins/providers.zig | 6 + src/core/app/app_auth_runtime.zig | 118 ++++-- src/core/app/app_lifecycle.zig | 1 + src/core/auth/auth_runtime.zig | 23 +- src/core/auth/auth_transition.zig | 8 + src/core/auth/credential_authority.zig | 1 + src/core/auth/credentials.zig | 9 + src/core/auth/provider_catalog.zig | 9 + src/core/cli/cli_ask.zig | 4 +- src/core/cli/cli_surface.zig | 43 ++- src/core/config/model_provider.zig | 8 +- src/core/config/settings_store.zig | 1 + src/core/gateway/provider_set.zig | 4 + src/core/output/output_contracts.zig | 1 + .../session/generation_usage_provider.zig | 2 + src/core/session/session_usage.zig | 1 + src/core/shared/types.zig | 1 + src/gateway/openai.zig | 171 +++++++++ src/gateway/openai_codex.zig | 335 +++--------------- src/gateway/responses_sse.zig | 107 ++++++ src/gateway/responses_transport.zig | 215 +++++++++++ src/main.zig | 1 + src/ui/footer/model_menu_presentation.zig | 1 + 24 files changed, 760 insertions(+), 323 deletions(-) create mode 100644 src/gateway/openai.zig create mode 100644 src/gateway/responses_sse.zig create mode 100644 src/gateway/responses_transport.zig diff --git a/README.md b/README.md index 37f120eb0..4e1552224 100644 --- a/README.md +++ b/README.md @@ -33,6 +33,17 @@ Sign in with Vercel AI Gateway: fx login ``` +Or use the OpenAI Responses API, including a compatible HTTPS or local loopback endpoint: + +```bash +export OPENAI_API_KEY=your_key +export OPENAI_BASE_URL=https://api.openai.com/v1 +FX_MODEL=gpt-5.4 fx provider openai +fx +``` + +`OPENAI_BASE_URL` defaults to `https://api.openai.com/v1`. The OpenAI route accepts HTTPS endpoints and loopback HTTP endpoints, uses `OPENAI_API_KEY` only for that route, and does not send Vercel or subscription credentials to the configured endpoint. Configure `models.openai` in `~/.fx/settings.json` instead of `FX_MODEL` when you want to select the model persistently without the command above. OpenAI-compatible model discovery is not assumed; `fx models` reports that limitation for this provider. + Or use an eligible ChatGPT subscription through OpenAI Codex OAuth: ```bash @@ -47,7 +58,7 @@ fx login grok fx ``` -`fx login codex` and `fx login grok` select that provider and a model from its authenticated catalog. Inside fx, open `/setup` and choose **Model provider** to move between Gateway, Codex, and Grok. `/model` lists the active provider's fetched models. Subscription model IDs are the raw IDs returned by each authenticated catalog. Use `/logout codex` or `/logout grok` to remove that subscription session without affecting other providers; choosing it again from **Model provider** starts sign-in. +`fx login codex` and `fx login grok` select that provider and a model from its authenticated catalog. Inside fx, open `/setup` and choose **Model provider** to move between Gateway, OpenAI, Codex, and Grok. `/model` lists the active provider's fetched models. Subscription model IDs are the raw IDs returned by each authenticated catalog. Use `/logout codex` or `/logout grok` to remove that subscription session without affecting other providers; choosing it again from **Model provider** starts sign-in. The OpenAI Codex route uses ChatGPT subscription access directly and never sends its OAuth token to Vercel AI Gateway. The session is stored privately at `~/.fx/chatgpt-auth.json` and refreshed when needed. On supported Codex models, `/fast` requests OpenAI's priority service tier and consumes ChatGPT credits at the higher Fast mode rate. diff --git a/src/builtins/providers.zig b/src/builtins/providers.zig index ac63905b1..d7d111f6c 100644 --- a/src/builtins/providers.zig +++ b/src/builtins/providers.zig @@ -1,5 +1,6 @@ const provider_set = @import("../core/gateway/provider_set.zig"); const gateway = @import("gateway.zig"); +const openai = @import("../gateway/openai.zig"); const openai_codex = @import("../gateway/openai_codex.zig"); const openai_codex_models = @import("../gateway/openai_codex_models.zig"); const openai_codex_permission_reviewer = @import("../gateway/openai_codex_permission_reviewer.zig"); @@ -10,6 +11,11 @@ const provider_catalog = @import("../core/auth/provider_catalog.zig"); pub const native = provider_set.Set{ .gateway = gateway.provider_bundle, + .openai = .{ + .presentation = provider_catalog.find(.openai), + .fallback_model_capabilities_fn = openai.fallback_capabilities, + .agent_stream = openai.agent_stream_provider, + }, .codex = .{ .presentation = provider_catalog.find(.codex), .auth_strategy = .chatgpt, diff --git a/src/core/app/app_auth_runtime.zig b/src/core/app/app_auth_runtime.zig index 3cfa808ff..b4f62a68a 100644 --- a/src/core/app/app_auth_runtime.zig +++ b/src/core/app/app_auth_runtime.zig @@ -1,5 +1,6 @@ const std = @import("std"); const config_runtime = @import("../config/config_runtime.zig"); +const settings_store = @import("../config/settings_store.zig"); const debug_trace = @import("../shared/debug_trace.zig"); const host = @import("../hosts/host.zig"); const runtime_profile = @import("../hosts/runtime_profile.zig"); @@ -57,6 +58,7 @@ pub fn Runtime(comptime App: type) type { { const provider = provider_runtime.provider(app); const required_source: credentials.Source = switch (provider) { + .openai => .openai_api_key, .codex => .chatgpt_subscription, .grok => .grok_subscription, .gateway => app.auth.credentialSource() orelse .fx_login, @@ -71,7 +73,9 @@ pub fn Runtime(comptime App: type) type { try app.writeDomainNotice(.{ .topic = "auth", .tone = .warning, - .body = if (provider == .grok) + .body = if (provider == .openai) + credentials.missing_openai_interactive_credential_message + else if (provider == .grok) credentials.missing_grok_interactive_credential_message else if (provider == .codex) credentials.missing_chatgpt_interactive_credential_message @@ -133,7 +137,7 @@ pub fn Runtime(comptime App: type) type { try writeAuthNotice(app, .{ .topic = "auth", .tone = .warning, - .body = "Usage: /logout [vercel|codex|grok]", + .body = "Usage: /logout [vercel|openai|codex|grok]", }); return; }; @@ -152,6 +156,14 @@ pub fn Runtime(comptime App: type) type { .active_source = app.auth.credentialSource(), .available_sources = provider_inventory, }); + if (logout_provider == .openai) { + try writeAuthNotice(app, .{ + .topic = "auth", + .tone = .neutral, + .body = "OpenAI uses the process-owned OPENAI_API_KEY. Unset it and restart fx to disconnect.", + }); + return; + } if (logout_provider == .grok) { const outcome = grok_oauth.logout(app.alloc, app.auth.oauthTransport()) catch { try writeAuthNotice(app, .{ @@ -805,6 +817,8 @@ pub fn Runtime(comptime App: type) type { .tone = .warning, .body = if (intent == .post_oauth) "Subscription sign-in completed, but its saved credential is unavailable. The current provider is unchanged." + else if (target == .openai) + credentials.missing_openai_interactive_credential_message else if (target == .codex) "Run fx login codex, then try switching again." else if (target == .grok) @@ -828,40 +842,91 @@ pub fn Runtime(comptime App: type) type { return; } - const access = credentials.catalogAccessForCredentialAndAccount( - credential.source, - credential.token, - credential.gatewayTeam(), - credential.accountId(), - ); - const fetched = app.fetchProviderCatalog(target, access) catch |err| { - debug_trace.logf("provider", "catalog preparation failed provider={t} err={s}", .{ target, @errorName(err) }); + var settings = config_runtime.loadMergedSettings(app.alloc, app.workspace_root) catch |err| { + debug_trace.logf("provider", "settings load failed err={s}", .{@errorName(err)}); try app.writeDomainNotice(.{ .topic = "provider", .tone = .@"error", .body = providerFailureMessage( intent, - "Could not load the target provider catalog. The current provider is unchanged.", - "Subscription sign-in completed, but its model catalog could not be loaded. The current provider is unchanged.", + "Could not load the saved provider model. The current provider is unchanged.", + "Subscription sign-in completed, but its saved provider model could not be loaded. The current provider is unchanged.", ), }, true); return; }; - var catalog = switch (fetched) { - .catalog => |catalog| catalog, - .failure => |failure| { - debug_trace.logf("provider", "catalog rejected provider={t} category={t}", .{ target, failure.category }); + defer settings.deinit(app.alloc); + + const access = credentials.catalogAccessForCredentialAndAccount( + credential.source, + credential.token, + credential.gatewayTeam(), + credential.accountId(), + ); + var catalog = if (target == .openai) catalog: { + const model = io_mod.getenv("FX_MODEL") orelse settings.models.get(.openai) orelse { + try app.writeDomainNotice(.{ + .topic = "provider", + .tone = .warning, + .body = "Configure models.openai or FX_MODEL before selecting OpenAI.", + }, true); + return; + }; + settings_store.validateModel(model) catch { + try app.writeDomainNotice(.{ + .topic = "provider", + .tone = .warning, + .body = "The configured OpenAI model is invalid. The current provider is unchanged.", + }, true); + return; + }; + var configured: std.ArrayList(model_catalog.ModelCatalogEntry) = .empty; + errdefer model_catalog.freeModelCatalog(app.alloc, &configured); + const id = try app.alloc.dupe(u8, model); + const model_type = app.alloc.dupe(u8, "language") catch |err| { + app.alloc.free(id); + return err; + }; + configured.append(app.alloc, .{ + .id = id, + .model_type = model_type, + .has_tool_use = true, + }) catch |err| { + app.alloc.free(id); + app.alloc.free(model_type); + return err; + }; + break :catalog configured; + } else catalog: { + const fetched = app.fetchProviderCatalog(target, access) catch |err| { + debug_trace.logf("provider", "catalog preparation failed provider={t} err={s}", .{ target, @errorName(err) }); try app.writeDomainNotice(.{ .topic = "provider", .tone = .@"error", .body = providerFailureMessage( intent, - "The target provider catalog could not be validated. The current provider is unchanged.", - "Subscription sign-in completed, but its model catalog could not be validated. The current provider is unchanged.", + "Could not load the target provider catalog. The current provider is unchanged.", + "Subscription sign-in completed, but its model catalog could not be loaded. The current provider is unchanged.", ), }, true); return; - }, + }; + break :catalog switch (fetched) { + .catalog => |loaded| loaded, + .failure => |failure| { + debug_trace.logf("provider", "catalog rejected provider={t} category={t}", .{ target, failure.category }); + try app.writeDomainNotice(.{ + .topic = "provider", + .tone = .@"error", + .body = providerFailureMessage( + intent, + "The target provider catalog could not be validated. The current provider is unchanged.", + "Subscription sign-in completed, but its model catalog could not be validated. The current provider is unchanged.", + ), + }, true); + return; + }, + }; }; defer model_catalog.freeModelCatalog(app.alloc, &catalog); if (catalog.items.len == 0) { @@ -877,20 +942,6 @@ pub fn Runtime(comptime App: type) type { return; } - var settings = config_runtime.loadMergedSettings(app.alloc, app.workspace_root) catch |err| { - debug_trace.logf("provider", "settings load failed err={s}", .{@errorName(err)}); - try app.writeDomainNotice(.{ - .topic = "provider", - .tone = .@"error", - .body = providerFailureMessage( - intent, - "Could not load the saved provider model. The current provider is unchanged.", - "Subscription sign-in completed, but its saved provider model could not be loaded. The current provider is unchanged.", - ), - }, true); - return; - }; - defer settings.deinit(app.alloc); const saved_model = settings.models.get(target); const current_model = if (intent == .post_oauth and current == target) provider_runtime.model(app) @@ -1395,6 +1446,7 @@ test "interactive subscription sign-in rejects active and queued work before OAu app.worker.queued_prompts = case.queued_prompts; switch (provider) { + .openai => unreachable, .codex => try Runtime(BusySignInApp).beginChatGptSignIn(&app), .grok => try Runtime(BusySignInApp).beginGrokSignIn(&app), .gateway => unreachable, diff --git a/src/core/app/app_lifecycle.zig b/src/core/app/app_lifecycle.zig index 37ea7ef85..ba272b85f 100644 --- a/src/core/app/app_lifecycle.zig +++ b/src/core/app/app_lifecycle.zig @@ -1113,6 +1113,7 @@ fn configuredProviderSelection( const provider = settings.provider orelse .gateway; const model = settings.models.get(provider) orelse switch (provider) { .gateway => default_model, + .openai => return error.OpenAIModelNotSelected, .codex => return error.CodexModelNotSelected, .grok => return error.GrokModelNotSelected, }; diff --git a/src/core/auth/auth_runtime.zig b/src/core/auth/auth_runtime.zig index d96b188b8..fb9256dbe 100644 --- a/src/core/auth/auth_runtime.zig +++ b/src/core/auth/auth_runtime.zig @@ -604,6 +604,12 @@ pub const StatusSnapshot = struct { pub fn missingHelp(self: StatusSnapshot, surface: MissingHelpSurface) ?[]const u8 { if (self.active_source != null) return null; if (self.stored_key_status == .unavailable) return credentials.unreadable_store_message; + if (self.required_source == .openai_api_key) { + return switch (surface) { + .cli => credentials.missing_openai_credential_message, + .interactive => credentials.missing_openai_interactive_credential_message, + }; + } if (self.required_source == .chatgpt_subscription) { return switch (surface) { .cli => credentials.missing_chatgpt_credential_message, @@ -695,7 +701,7 @@ pub fn loadStatusSnapshotForProvider( }, }; const resolved_source = if (resolution.credential) |credential| credential.source else null; - var gateway_connected = resolved_source != null and resolved_source != .chatgpt_subscription and resolved_source != .grok_subscription; + var gateway_connected = resolved_source != null and resolved_source != .openai_api_key and resolved_source != .chatgpt_subscription and resolved_source != .grok_subscription; const gateway_probe_required = provider == .codex or provider == .grok or resolved_source == .chatgpt_subscription or resolved_source == .grok_subscription; if (gateway_probe_required) { @@ -726,7 +732,9 @@ pub fn loadStatusSnapshotForProvider( }; } return .{ - .required_source = if (provider == .codex) + .required_source = if (provider == .openai) + .openai_api_key + else if (provider == .codex) .chatgpt_subscription else if (provider == .grok) .grok_subscription @@ -1618,6 +1626,15 @@ pub const Runtime = struct { provider: model_provider.ProviderId, ) !?bool { return switch (provider) { + .openai => if (self.credentialSource() == .openai_api_key) + false + else + self.selectSourceWithLoader( + alloc, + .openai_api_key, + self, + loadRuntimeCredentialSource, + ), .codex => if (self.credentialSource() == .chatgpt_subscription) false else @@ -1636,7 +1653,7 @@ pub const Runtime = struct { self, loadRuntimeCredentialSource, ), - .gateway => if (self.credentialSource() != .chatgpt_subscription and self.credentialSource() != .grok_subscription) + .gateway => if (self.credentialSource() != .openai_api_key and self.credentialSource() != .chatgpt_subscription and self.credentialSource() != .grok_subscription) false else @as(?bool, try self.reselectByPrecedenceWithDeps( diff --git a/src/core/auth/auth_transition.zig b/src/core/auth/auth_transition.zig index 84abed350..586e4d3d3 100644 --- a/src/core/auth/auth_transition.zig +++ b/src/core/auth/auth_transition.zig @@ -39,6 +39,7 @@ pub const LogoutFacts = struct { pub fn decideLogoutProvider(facts: LogoutFacts) model_provider.ProviderId { if (facts.requested) |provider| return provider; + if (facts.selected == .openai or facts.active_source == .openai_api_key) return .openai; if (facts.selected == .grok or facts.active_source == .grok_subscription) return .grok; if (facts.selected == .codex or facts.active_source == .chatgpt_subscription) return .codex; @@ -65,6 +66,7 @@ pub fn signInCompletion( ) SignInCompletionAction { return switch (provider) { .gateway => .vercel, + .openai => .{ .activate_source = .openai_api_key }, .codex => if (provider_routing_supported) .{ .switch_provider = .codex } else @@ -108,6 +110,12 @@ test "provider switch and logout decisions are pure and provider keyed" { .active_source = .chatgpt_subscription, .available_sources = inventory, })); + try std.testing.expectEqual(model_provider.ProviderId.openai, decideLogoutProvider(.{ + .requested = null, + .selected = .openai, + .active_source = .openai_api_key, + .available_sources = inventory, + })); } test "sign in completion selects routing or credential activation without effects" { diff --git a/src/core/auth/credential_authority.zig b/src/core/auth/credential_authority.zig index a031d3c42..9fc816ad9 100644 --- a/src/core/auth/credential_authority.zig +++ b/src/core/auth/credential_authority.zig @@ -23,6 +23,7 @@ pub fn derive( switch (source) { .vercel_oidc_token, .ai_gateway_api_key, + .openai_api_key, .fx_login, .stored_key, => hash.update("\x00slot\x00"), diff --git a/src/core/auth/credentials.zig b/src/core/auth/credentials.zig index 563a11c04..30d6656c7 100644 --- a/src/core/auth/credentials.zig +++ b/src/core/auth/credentials.zig @@ -40,6 +40,7 @@ pub const CatalogPublicOnlyReason = std.meta.Tag(CatalogPublicOnly); pub const CatalogAuthenticatedSource = enum { vercel_oidc_token, ai_gateway_api_key, + openai_api_key, fx_login, stored_key, chatgpt_subscription, @@ -49,6 +50,7 @@ pub const CatalogAuthenticatedSource = enum { return switch (self) { .vercel_oidc_token => .vercel_oidc_token, .ai_gateway_api_key => .ai_gateway_api_key, + .openai_api_key => .openai_api_key, .fx_login => .fx_login, .stored_key => .stored_key, .chatgpt_subscription => .chatgpt_subscription, @@ -165,6 +167,7 @@ pub fn catalogAccessForCredentialAndAccount( const authenticated_source: CatalogAuthenticatedSource = switch (selected_source) { .vercel_oidc_token => .vercel_oidc_token, .ai_gateway_api_key => .ai_gateway_api_key, + .openai_api_key => .openai_api_key, .stored_key => .stored_key, .chatgpt_subscription => .chatgpt_subscription, .grok_subscription => .grok_subscription, @@ -198,6 +201,8 @@ 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_openai_credential_message = "fx needs an OpenAI API key. Set OPENAI_API_KEY."; +pub const missing_openai_interactive_credential_message = "fx needs an OpenAI API key. Set OPENAI_API_KEY."; 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."; @@ -283,6 +288,7 @@ pub fn resolveForProvider( preferred: ?Source, ) !Resolution { switch (provider) { + .openai => return .{ .credential = try loadSource(alloc, transport, secret_store, .openai_api_key) }, .codex => { const credential = switch (mode) { .stored => try loadStoredChatGptCredential(alloc), @@ -407,6 +413,7 @@ pub fn loadSource( return switch (source) { .vercel_oidc_token => loadEnvCredential(alloc, "VERCEL_OIDC_TOKEN", source), .ai_gateway_api_key => loadEnvCredential(alloc, "AI_GATEWAY_API_KEY", source), + .openai_api_key => loadEnvCredential(alloc, "OPENAI_API_KEY", source), .fx_login => loadFxLoginCredential(alloc, transport), .stored_key => loadStoredKeyCredential(alloc, secret_store), .chatgpt_subscription => loadChatGptCredential(alloc, transport, .if_needed), @@ -422,6 +429,7 @@ pub fn sourceExists( return switch (source) { .vercel_oidc_token => nonEmptyEnvValue("VERCEL_OIDC_TOKEN") != null, .ai_gateway_api_key => nonEmptyEnvValue("AI_GATEWAY_API_KEY") != null, + .openai_api_key => nonEmptyEnvValue("OPENAI_API_KEY") != null, .fx_login => blk: { const loaded = oauth_session.load(alloc) catch |err| switch (err) { error.OutOfMemory => return err, @@ -658,6 +666,7 @@ pub fn sourceLabel(source: Source) []const u8 { return switch (source) { .vercel_oidc_token => "VERCEL_OIDC_TOKEN", .ai_gateway_api_key => "AI_GATEWAY_API_KEY", + .openai_api_key => "OPENAI_API_KEY", .fx_login => "fx login", .stored_key => "stored API key (" ++ stored_key_backend_label ++ ")", .chatgpt_subscription => "Codex subscription", diff --git a/src/core/auth/provider_catalog.zig b/src/core/auth/provider_catalog.zig index d949aa7ff..c0d55ef50 100644 --- a/src/core/auth/provider_catalog.zig +++ b/src/core/auth/provider_catalog.zig @@ -21,6 +21,14 @@ pub const entries = [_]Entry{ .description = "Vercel account or AI Gateway billing", .subscription = false, }, + .{ + .id = .openai, + .slug = "openai", + .name = "OpenAI API", + .route_name = "OpenAI API", + .description = "OpenAI API key and Responses-compatible endpoint", + .subscription = false, + }, .{ .id = .codex, .slug = "codex", @@ -59,6 +67,7 @@ pub fn label(id: model_provider.ProviderId) []const u8 { 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.openai, parse("openai").?); 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); diff --git a/src/core/cli/cli_ask.zig b/src/core/cli/cli_ask.zig index 2ee7dc03e..a1dbe87fb 100644 --- a/src/core/cli/cli_ask.zig +++ b/src/core/cli/cli_ask.zig @@ -1404,7 +1404,9 @@ fn missingCredentialResult( options: RunOptions, provider: model_provider.ProviderId, ) !PromptRunResult { - const message = if (provider == .codex) + const message = if (provider == .openai) + credentials.missing_openai_credential_message + else if (provider == .codex) credentials.missing_chatgpt_credential_message else if (provider == .grok) credentials.missing_grok_credential_message diff --git a/src/core/cli/cli_surface.zig b/src/core/cli/cli_surface.zig index dc4f4f915..a2b8dfb1b 100644 --- a/src/core/cli/cli_surface.zig +++ b/src/core/cli/cli_surface.zig @@ -701,6 +701,7 @@ fn activateProviderSelection( if (caller == .provider_command and already_selected and resolution.credential != null) { try writeStdout(deps, switch (target) { .gateway => "Gateway is already selected.\n", + .openai => "OpenAI is already selected.\n", .codex => "Codex is already selected.\n", .grok => "Grok is already selected.\n", }); @@ -747,6 +748,7 @@ fn activateProviderSelection( deps, caller, switch (target) { + .openai => "OPENAI_API_KEY is unavailable", .codex => "Codex credential is unavailable", .grok => "Grok credential is unavailable", .gateway => "configure a Gateway credential first", @@ -754,8 +756,30 @@ fn activateProviderSelection( ); return false; }; + if (target == .openai) { + const selected_model = settings.models.get(.openai) orelse io_mod.getenv("FX_MODEL") orelse { + try writeProviderActivationError(alloc, deps, caller, "configure models.openai or FX_MODEL first"); + return false; + }; + var attempt = config_runtime.attemptUserPreferences(alloc, .{ + .provider = .openai, + .model_preference = .{ .provider = .openai, .model = selected_model }, + }); + defer attempt.deinit(alloc); + switch (attempt) { + .failure => |failure| { + debug_trace.logf("config", "OpenAI provider selection persistence failed err={s}", .{@errorName(failure.err)}); + try writeProviderActivationError(alloc, deps, caller, "failed to save OpenAI provider selection"); + return false; + }, + .outcome => {}, + } + if (caller == .provider_command) try writeStdout(deps, "Provider set to OpenAI.\n"); + return true; + } const catalog_provider = cfg.provider_set.select(target).model_catalog orelse { try writeProviderActivationError(alloc, deps, caller, switch (target) { + .openai => unreachable, .codex => "Codex model catalog is unavailable", .grok => "Grok model catalog is unavailable", .gateway => "Gateway model catalog is unavailable", @@ -801,6 +825,7 @@ fn activateProviderSelection( .outcome => {}, } if (performed_login) |provider| switch (provider) { + .openai => unreachable, .codex => try writeStdout(deps, "Signed in with Codex.\n"), .grok => try writeStdout(deps, "Signed in with Grok.\n"), .gateway => unreachable, @@ -808,6 +833,7 @@ fn activateProviderSelection( if (caller == .provider_command) { try writeStdout(deps, switch (target) { .gateway => "Provider set to Gateway.\n", + .openai => unreachable, .codex => "Provider set to Codex.\n", .grok => "Provider set to Grok.\n", }); @@ -932,12 +958,16 @@ fn runNonInteractiveWithDeps( .issue => |rest| return runGithubWorkflow(alloc, rest, cfg, global_args.modifiers, deps, .issue), .login => |rest| { const maybe_login_provider = parseLoginProvider(rest) catch { - try writeStderr(deps, "usage: fx login [vercel|codex|grok]\n"); + try writeStderr(deps, "usage: fx login [vercel|openai|codex|grok]\n"); return .handled_failure; }; // Preserve the original `fx login` behavior for scripts and users. const login_provider = maybe_login_provider orelse .gateway; switch (login_provider) { + .openai => { + try writeStderr(deps, "fx login: OpenAI uses OPENAI_API_KEY; set it and select the OpenAI provider\n"); + return .handled_failure; + }, .gateway => login_flow.runLogin( alloc, cfg.gateway_provider.oauth_transport, @@ -991,11 +1021,15 @@ fn runNonInteractiveWithDeps( }, .logout => |rest| { const maybe_login_provider = parseLoginProvider(rest) catch { - try writeStderr(deps, "usage: fx logout [vercel|codex|grok]\n"); + try writeStderr(deps, "usage: fx logout [vercel|openai|codex|grok]\n"); return .handled_failure; }; // Preserve the original `fx logout` behavior for scripts and users. const login_provider = maybe_login_provider orelse .gateway; + if (login_provider == .openai) { + try writeStdout(deps, "OpenAI uses the process-owned OPENAI_API_KEY; unset it to disconnect.\n"); + return .handled_success; + } if (login_provider == .codex) { const outcome = chatgpt_oauth.logout() catch { try writeStderr(deps, "fx logout: failed to durably remove saved Codex login\n"); @@ -1080,11 +1114,11 @@ fn runNonInteractiveWithDeps( }, .provider => |rest| { if (rest.len != 1) { - try writeStderr(deps, "usage: fx provider \n"); + try writeStderr(deps, "usage: fx provider \n"); return .handled_failure; } const target = model_provider.parse(rest[0]) orelse { - try writeStderr(deps, "fx provider: expected gateway, codex, or grok\n"); + try writeStderr(deps, "fx provider: expected gateway, openai, codex, or grok\n"); return .handled_failure; }; return if (try activateProviderSelection(alloc, cfg, deps, target, .provider_command)) @@ -1178,6 +1212,7 @@ fn runNonInteractiveWithDeps( const catalog_provider = cfg.provider_set.select(startup.provider).cli_model_catalog orelse { try writeStderr(deps, switch (startup.provider) { .gateway => "fx models: Gateway model catalog is unavailable\n", + .openai => "fx models: OpenAI model discovery is unavailable; configure models.openai or FX_MODEL\n", .codex => "fx models: Codex model catalog is unavailable\n", .grok => "fx models: Grok model catalog is unavailable\n", }); diff --git a/src/core/config/model_provider.zig b/src/core/config/model_provider.zig index 93b0168a7..557ee0f84 100644 --- a/src/core/config/model_provider.zig +++ b/src/core/config/model_provider.zig @@ -3,6 +3,7 @@ const types = @import("../shared/types.zig"); pub const ProviderId = enum { gateway, + openai, codex, grok, }; @@ -14,6 +15,7 @@ pub const ProviderSelection = struct { pub fn parse(value: []const u8) ?ProviderId { if (std.ascii.eqlIgnoreCase(value, "gateway")) return .gateway; + if (std.ascii.eqlIgnoreCase(value, "openai")) return .openai; if (std.ascii.eqlIgnoreCase(value, "codex")) return .codex; if (std.ascii.eqlIgnoreCase(value, "grok")) return .grok; return null; @@ -22,7 +24,8 @@ pub fn parse(value: []const u8) ?ProviderId { 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 != .openai_api_key and selected != .chatgpt_subscription and selected != .grok_subscription, + .openai => selected == .openai_api_key, .codex => selected == .chatgpt_subscription, .grok => selected == .grok_subscription, }; @@ -31,6 +34,8 @@ pub fn authorizesCredential(provider: ProviderId, source: ?types.CredentialSourc test "explicit providers authorize only their own credential origins" { try std.testing.expect(authorizesCredential(.gateway, .ai_gateway_api_key)); try std.testing.expect(authorizesCredential(.gateway, .fx_login)); + try std.testing.expect(authorizesCredential(.openai, .openai_api_key)); + try std.testing.expect(!authorizesCredential(.openai, .ai_gateway_api_key)); try std.testing.expect(!authorizesCredential(.gateway, .chatgpt_subscription)); try std.testing.expect(authorizesCredential(.codex, .chatgpt_subscription)); try std.testing.expect(!authorizesCredential(.codex, .ai_gateway_api_key)); @@ -42,6 +47,7 @@ test "explicit providers authorize only their own credential origins" { test "provider parsing exposes gateway codex and grok" { try std.testing.expectEqual(ProviderId.gateway, parse("gateway").?); + try std.testing.expectEqual(ProviderId.openai, parse("OPENAI").?); try std.testing.expectEqual(ProviderId.codex, parse("CODEX").?); try std.testing.expectEqual(ProviderId.grok, parse("GROK").?); try std.testing.expect(parse("openai-codex") == null); diff --git a/src/core/config/settings_store.zig b/src/core/config/settings_store.zig index 5cb8e756a..7542db0b1 100644 --- a/src/core/config/settings_store.zig +++ b/src/core/config/settings_store.zig @@ -1550,6 +1550,7 @@ fn putModelPreference( changed = try putString(arena, models, @tagName(preference.provider), preference.model) or changed; const legacy_key = switch (preference.provider) { .gateway => "model", + .openai => "openai_model", .codex => "codex_model", .grok => "grok_model", }; diff --git a/src/core/gateway/provider_set.zig b/src/core/gateway/provider_set.zig index 97d921acb..2157954f2 100644 --- a/src/core/gateway/provider_set.zig +++ b/src/core/gateway/provider_set.zig @@ -49,12 +49,14 @@ fn emptyModelCapabilities(_: []const u8) model_capabilities.Capabilities { pub const Set = struct { gateway: Bundle, + openai: Bundle = .{}, codex: Bundle, grok: Bundle, pub fn select(self: Set, provider: model_provider.ProviderId) Bundle { return switch (provider) { .gateway => self.gateway, + .openai => self.openai, .codex => self.codex, .grok => self.grok, }; @@ -63,6 +65,7 @@ pub const Set = struct { pub fn deferredUsageProviders(self: Set) generation_usage_provider.Set { return .{ .gateway = self.gateway.deferred_usage, + .openai = self.openai.deferred_usage, .codex = self.codex.deferred_usage, .grok = self.grok.deferred_usage, }; @@ -72,6 +75,7 @@ pub const Set = struct { pub fn gateway_only(gateway: Bundle) Set { return .{ .gateway = gateway, + .openai = .{}, .codex = .{}, .grok = .{}, }; diff --git a/src/core/output/output_contracts.zig b/src/core/output/output_contracts.zig index 271346b91..99e84c062 100644 --- a/src/core/output/output_contracts.zig +++ b/src/core/output/output_contracts.zig @@ -850,6 +850,7 @@ pub const ModelListSnapshot = struct { fn emptyCatalogProviderName(self: ModelListSnapshot) []const u8 { return switch (self.provider) { .gateway => "gateway", + .openai => provider_catalog.label(.openai), .codex => provider_catalog.label(.codex), .grok => provider_catalog.label(.grok), }; diff --git a/src/core/session/generation_usage_provider.zig b/src/core/session/generation_usage_provider.zig index 19ad021b1..db5b56b37 100644 --- a/src/core/session/generation_usage_provider.zig +++ b/src/core/session/generation_usage_provider.zig @@ -83,6 +83,7 @@ pub const unavailable_provider = Provider{ pub const Set = struct { gateway: ?Provider = null, + openai: ?Provider = null, codex: ?Provider = null, grok: ?Provider = null, @@ -93,6 +94,7 @@ pub const Set = struct { pub fn select(self: Set, provider: model_provider.ProviderId) ?Provider { return switch (provider) { .gateway => self.gateway, + .openai => self.openai, .codex => self.codex, .grok => self.grok, }; diff --git a/src/core/session/session_usage.zig b/src/core/session/session_usage.zig index 1d354e746..8de3b7e8c 100644 --- a/src/core/session/session_usage.zig +++ b/src/core/session/session_usage.zig @@ -3270,6 +3270,7 @@ fn canonicalExactGenerationId( fn exactUsageOrigin(provider: model_provider.ProviderId) []const u8 { return switch (provider) { .gateway => "exact/gateway", + .openai => "exact/openai", .codex => "exact/codex", .grok => "exact/grok", }; diff --git a/src/core/shared/types.zig b/src/core/shared/types.zig index 3901bc51d..3a3b70fb4 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, + openai_api_key, fx_login, stored_key, chatgpt_subscription, diff --git a/src/gateway/openai.zig b/src/gateway/openai.zig new file mode 100644 index 000000000..d2bc09c6a --- /dev/null +++ b/src/gateway/openai.zig @@ -0,0 +1,171 @@ +const std = @import("std"); +const stream_provider = @import("../core/agent/stream_provider.zig"); +const model_capabilities = @import("../core/config/model_capabilities.zig"); +const io_mod = @import("../core/shared/io.zig"); +const gateway_client = @import("client.zig"); +const responses_protocol = @import("responses_protocol.zig"); +const responses_transport = @import("responses_transport.zig"); + +const Allocator = std.mem.Allocator; + +pub const default_base_url = "https://api.openai.com/v1"; +pub const base_url_env = "OPENAI_BASE_URL"; +const max_error_body_bytes: usize = 1024 * 1024; +const max_sse_line_bytes: usize = 32 * 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 max_provider_state_bytes: usize = 4 * 1024 * 1024; + +pub const agent_stream_provider = stream_provider.Provider{ + .stream_fn = stream_completion, +}; + +pub fn fallback_capabilities(_: []const u8) model_capabilities.Capabilities { + return .{ .supports_tool_use = true }; +} + +/// The caller owns the returned Responses endpoint. +pub fn responses_url(alloc: Allocator, base_url: []const u8) ![]u8 { + const trimmed = std.mem.trim(u8, base_url, " \t\r\n"); + const without_trailing_slash = std.mem.trimEnd(u8, trimmed, "/"); + if (without_trailing_slash.len == 0) return error.InvalidOpenAIBaseUrl; + const uri = std.Uri.parse(without_trailing_slash) catch return error.InvalidOpenAIBaseUrl; + if (uri.host == null or uri.user != null or uri.password != null or uri.query != null or uri.fragment != null) { + return error.InvalidOpenAIBaseUrl; + } + if (!std.ascii.eqlIgnoreCase(uri.scheme, "https") and !gateway_client.isLoopbackHttpUrl(without_trailing_slash)) { + return error.InvalidOpenAIBaseUrl; + } + return std.fmt.allocPrint(alloc, "{s}/responses", .{without_trailing_slash}); +} + +fn configured_base_url() []const u8 { + return io_mod.getenv(base_url_env) orelse default_base_url; +} + +fn validate_model(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 build_request(alloc: Allocator, request: stream_provider.RequestData) ![]u8 { + try validate_model(request.model); + if (request.budget) |budget| { + if (budget.cancel_flag) |flag| if (flag.load(.seq_cst)) return error.Cancelled; + _ = budget.deadline; + } + + var instructions: std.Io.Writer.Allocating = .init(alloc); + defer instructions.deinit(); + for (request.messages) |message| { + if (message.role != .system) continue; + const text = message.content orelse continue; + if (text.len == 0) continue; + if (instructions.written().len > 0) try instructions.writer.writeAll("\n\n"); + try instructions.writer.writeAll(text); + } + if (instructions.written().len == 0) try instructions.writer.writeAll("You are a helpful assistant."); + + 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.writeAll(",\"store\":false,\"stream\":true,\"instructions\":"); + try std.json.Stringify.value(instructions.written(), .{}, writer); + try writer.writeAll(",\"input\":["); + try responses_protocol.writeInput(writer, alloc, request.messages, request.verified_images, .{ + .tool_calls = max_tool_calls, + .tool_identity_bytes = max_tool_identity_bytes, + .tool_arguments_bytes = max_tool_arguments_bytes, + .provider_state_bytes = max_provider_state_bytes, + }); + try writer.writeByte(']'); + _ = try responses_protocol.writeTools(writer, alloc, request.tools); + try writer.writeAll(",\"tool_choice\":"); + try std.json.Stringify.value(request.tool_choice.label(), .{}, writer); + try writer.writeAll(",\"parallel_tool_calls\":true,\"include\":[\"reasoning.encrypted_content\"]"); + if (request.max_output_tokens) |limit| try writer.print(",\"max_output_tokens\":{d}", .{limit}); + if (request.response_format) |format| { + if (format.schema != .object) return error.InvalidStructuredResponseSchema; + try writer.writeAll(",\"text\":{\"format\":{\"type\":\"json_schema\",\"name\":"); + try std.json.Stringify.value(format.name, .{}, writer); + try writer.writeAll(",\"description\":"); + try std.json.Stringify.value(format.description, .{}, writer); + try writer.writeAll(",\"schema\":"); + try std.json.Stringify.value(format.schema, .{}, writer); + try writer.writeAll(",\"strict\":true}}"); + } + if (request.provider_options.reasoning) |effort| { + try writer.writeAll(",\"reasoning\":{\"effort\":"); + try std.json.Stringify.value(effort.label(), .{}, writer); + try writer.writeByte('}'); + } + try writer.writeByte('}'); + return out.toOwnedSlice(); +} + +fn stream_completion(_: ?*anyopaque, alloc: Allocator, request: stream_provider.ModelRequest) !stream_provider.Result { + if (request.credential.source != .openai_api_key) { + return stream_provider.failResult(error.OpenAIApiKeyRequired); + } + const payload = try build_request(alloc, request.data()); + defer alloc.free(payload); + const endpoint = try responses_url(alloc, configured_base_url()); + defer alloc.free(endpoint); + return responses_transport.stream(alloc, request, payload, .{ + .endpoint = endpoint, + .extra_headers = &.{.{ .name = "accept", .value = "text/event-stream" }}, + .max_error_body_bytes = max_error_body_bytes, + .error_limit_message = "OpenAI error response exceeded the local limit", + .stream_limits = .{ + .line_bytes = max_sse_line_bytes, + .aggregate_bytes = max_sse_aggregate_bytes, + .events = max_sse_events, + .tool_calls = max_tool_calls, + .tool_identity_bytes = max_tool_identity_bytes, + .tool_arguments_bytes = max_tool_arguments_bytes, + .provider_state_bytes = max_provider_state_bytes, + }, + }); +} + +test "OpenAI base URL appends Responses endpoint and rejects unsafe origins" { + const url = try responses_url(std.testing.allocator, "https://proxy.example/v1/"); + defer std.testing.allocator.free(url); + try std.testing.expectEqualStrings("https://proxy.example/v1/responses", url); + const loopback = try responses_url(std.testing.allocator, "http://127.0.0.1:43123/v1"); + defer std.testing.allocator.free(loopback); + try std.testing.expectEqualStrings("http://127.0.0.1:43123/v1/responses", loopback); + try std.testing.expectError(error.InvalidOpenAIBaseUrl, responses_url(std.testing.allocator, "http://proxy.example/v1")); + try std.testing.expectError(error.InvalidOpenAIBaseUrl, responses_url(std.testing.allocator, "https://key@proxy.example/v1")); + try std.testing.expectError(error.InvalidOpenAIBaseUrl, responses_url(std.testing.allocator, "https://proxy.example/v1?target=other")); +} + +test "OpenAI request rejects unsafe model identifiers" { + try std.testing.expectError(error.InvalidOpenAIModel, build_request(std.testing.allocator, .{ + .model = "gpt model", + .messages = &.{}, + .tool_choice = .auto, + .provider_options = .{}, + })); +} + +test "OpenAI Responses request includes direct API limits and replay state" { + const body = try build_request(std.testing.allocator, .{ + .model = "gpt-5.4", + .messages = &.{.{ .role = .user, .content = "Hello" }}, + .tool_choice = .auto, + .provider_options = .{}, + .max_output_tokens = 400, + }); + defer std.testing.allocator.free(body); + try std.testing.expect(std.mem.find(u8, body, "\"max_output_tokens\":400") != null); + try std.testing.expect(std.mem.find(u8, body, "\"model\":\"gpt-5.4\"") != null); + try std.testing.expect(std.mem.find(u8, body, "\"include\":[\"reasoning.encrypted_content\"]") != null); +} diff --git a/src/gateway/openai_codex.zig b/src/gateway/openai_codex.zig index 95e3356ed..00c61d819 100644 --- a/src/gateway/openai_codex.zig +++ b/src/gateway/openai_codex.zig @@ -1,12 +1,13 @@ const std = @import("std"); const chatgpt_oauth = @import("../core/auth/chatgpt_oauth.zig"); const image_attachments = @import("../core/images/image_attachments.zig"); -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 responses_protocol = @import("responses_protocol.zig"); +const responses_sse = @import("responses_sse.zig"); +const responses_transport = @import("responses_transport.zig"); const model_tool_schema = @import("../core/tooling/model_tool_schema.zig"); const Allocator = std.mem.Allocator; @@ -20,10 +21,10 @@ const max_tool_calls: usize = 128; const max_tool_identity_bytes: usize = 1024; const max_tool_arguments_bytes: usize = 4 * 1024 * 1024; const max_provider_state_bytes: usize = 4 * 1024 * 1024; -const transfer_buffer_bytes: usize = 256 * 1024; const connect_timeout_ms: i64 = 30_000; const CodexLimits = struct { + line_bytes: usize = max_sse_line_bytes, aggregate_bytes: usize = max_sse_aggregate_bytes, events: usize = max_sse_events, tool_calls: usize = max_tool_calls, @@ -140,281 +141,74 @@ fn streamCompletion( try validateModel(request.model); const payload = try buildRequest(alloc, request.data()); defer alloc.free(payload); - return streamPrepared(alloc, request, payload) catch |err| { - if (request.cancel_flag.load(.seq_cst)) return stream_provider.failResult(error.Cancelled); - request.attempt_evidence.network_failure = gateway_client.networkFailureEvidence(err, request.delivery.load()); - return err; - }; + return streamPrepared(alloc, request, payload); } -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, - auth_header: []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 = .{ .override = self.auth_header }, - .accept_encoding = .omit, - .user_agent = .{ .override = gateway_client.user_agent }, - }, - .extra_headers = self.extra_headers, - .keep_alive = false, - .redirect_behavior = .unhandled, - }) }; - } -}; - pub fn streamPrepared( alloc: Allocator, request: stream_provider.ModelRequest, payload: []const u8, ) !stream_provider.Result { - if (request.cancel_flag.load(.seq_cst)) return stream_provider.failResult(error.Cancelled); const account_id = try chatgpt_oauth.extractAccountId(alloc, request.credential.secret); defer alloc.free(account_id); - const auth_header = try std.fmt.allocPrint(alloc, "Bearer {s}", .{request.credential.secret}); - defer secret.zeroAndFree(alloc, auth_header); const request_endpoint = if (io_mod.getenv(e2e_endpoint_env)) |override| endpoint: { if (!gateway_client.isLoopbackHttpUrl(override)) { return stream_provider.failResult(error.InvalidE2EOpenAICodexEndpoint); } break :endpoint override; } else endpoint; - const uri = try std.Uri.parse(request_endpoint); - - var extra_headers_buf: [7]std.http.Header = undefined; - var extra_count: usize = 0; - extra_headers_buf[extra_count] = .{ .name = "chatgpt-account-id", .value = account_id }; - extra_count += 1; - extra_headers_buf[extra_count] = .{ .name = "originator", .value = "fx" }; - extra_count += 1; - extra_headers_buf[extra_count] = .{ .name = "OpenAI-Beta", .value = "responses=experimental" }; - extra_count += 1; - extra_headers_buf[extra_count] = .{ .name = "accept", .value = "text/event-stream" }; - extra_count += 1; + + var headers: [7]std.http.Header = undefined; + var count: usize = 0; + headers[count] = .{ .name = "chatgpt-account-id", .value = account_id }; + count += 1; + headers[count] = .{ .name = "originator", .value = "fx" }; + count += 1; + headers[count] = .{ .name = "OpenAI-Beta", .value = "responses=experimental" }; + count += 1; + headers[count] = .{ .name = "accept", .value = "text/event-stream" }; + count += 1; if (request.session_id) |session_id| if (session_id.len > 0) { - extra_headers_buf[extra_count] = .{ .name = "session-id", .value = session_id }; - extra_count += 1; - extra_headers_buf[extra_count] = .{ .name = "x-client-request-id", .value = session_id }; - extra_count += 1; + headers[count] = .{ .name = "session-id", .value = session_id }; + count += 1; + headers[count] = .{ .name = "x-client-request-id", .value = session_id }; + count += 1; }; - 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, - .extra_headers = extra_headers_buf[0..extra_count], - }; - const connect_deadline = std.Io.Clock.Timestamp.fromNow(io_mod.getIo(), .{ - .clock = .awake, - .raw = .fromMilliseconds(connect_timeout_ms), + var result = try responses_transport.stream(alloc, request, payload, .{ + .endpoint = request_endpoint, + .extra_headers = headers[0..count], + .max_error_body_bytes = max_error_body_bytes, + .error_limit_message = "OpenAI Codex error response exceeded the local limit", + .connect_timeout_ms = connect_timeout_ms, + .stream_limits = .{ + .line_bytes = max_sse_line_bytes, + .aggregate_bytes = max_sse_aggregate_bytes, + .events = max_sse_events, + .tool_calls = max_tool_calls, + .tool_identity_bytes = max_tool_identity_bytes, + .tool_arguments_bytes = max_tool_arguments_bytes, + .provider_state_bytes = max_provider_state_bytes, + }, }); - try request.admission.admit(); - var opened = try gateway_client.runBoundedHttpOperation( - OpenedRequest, + errdefer result.deinit(alloc); + switch (result) { + .failed => return result, + .completed => {}, + } + const completion = &result.completed.completion; + if (completion.generation_id == null) return result; + completion.billing = try responses_protocol.buildSubscriptionBilling( 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| - 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 body = reader.allocRemaining(alloc, .limited(max_error_body_bytes)) catch |err| switch (err) { - error.StreamTooLong => try alloc.dupe(u8, "OpenAI Codex error response exceeded the local limit"), - else => return err, - }; - 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; - var completion = try consumeSse( - alloc, - reader, - &events, - EventBridge.content, - EventBridge.toolStart, - EventBridge.reasoning, - EventBridge.toolInput, - request.cancel_flag, - request.content_capture_limit, - .{}, - ); - errdefer { - var owned = stream_provider.Result{ .completed = .{ - .completion = completion, - .ownership = .owned, - } }; - owned.deinit(alloc); - } - const usage_outcome: stream_provider.UsageOutcome = usage: { - if (completion.generation_id == null) { - break :usage .{ .unavailable = .possibly_billed }; - } - completion.billing = try responses_protocol.buildSubscriptionBilling( - alloc, - .codex, - request.model, - @max(io_mod.milliTimestamp(), 0), - completion.usage, - ) orelse break :usage .{ .unavailable = .possibly_billed }; - break :usage .{ .exact = .codex }; - }; - return .{ .completed = .{ - .completion = completion, - .usage = usage_outcome, - .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 } }); - } -}; - -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, - }; + .codex, + request.model, + @max(io_mod.milliTimestamp(), 0), + completion.usage, + ) orelse return result; + result.completed.usage = .{ .exact = .codex }; + return result; } -const SseReader = struct { - pending_line: std.ArrayList(u8) = .empty, - - fn deinit(self: *SseReader, alloc: Allocator) void { - self.pending_line.deinit(alloc); - } - - fn release(self: *SseReader) void { - self.pending_line.clearRetainingCapacity(); - } - - fn next(self: *SseReader, alloc: Allocator, reader: anytype) !?[]const u8 { - while (true) { - const line = try self.readLine(alloc, reader) orelse return null; - const trimmed = std.mem.trim(u8, line, " \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 (std.mem.eql(u8, data, "[DONE]")) return null; - return data; - } - } - - fn readLine(self: *SseReader, alloc: Allocator, reader: anytype) !?[]const u8 { - while (true) { - const fragment = reader.takeDelimiter('\n') catch |err| switch (err) { - error.StreamTooLong => { - const buffered = reader.buffered(); - if (buffered.len == 0) return error.OpenAICodexSseReadStalled; - if (buffered.len > max_sse_line_bytes - self.pending_line.items.len) { - return error.OpenAICodexSseEventTooLarge; - } - try self.pending_line.appendSlice(alloc, buffered); - reader.tossBuffered(); - continue; - }, - error.ReadFailed => return error.ReadFailed, - } orelse { - if (self.pending_line.items.len > 0) return self.pending_line.items; - return null; - }; - if (fragment.len > max_sse_line_bytes - self.pending_line.items.len) { - return error.OpenAICodexSseEventTooLarge; - } - if (self.pending_line.items.len == 0) return fragment; - try self.pending_line.appendSlice(alloc, fragment); - return self.pending_line.items; - } - } -}; - fn consumeSse( alloc: Allocator, reader: anytype, @@ -427,42 +221,23 @@ fn consumeSse( content_capture_limit: ?usize, limits: CodexLimits, ) !types.ModelCompletion { - var reducer = responses_protocol.Reducer.init(alloc); - defer reducer.deinit(alloc); - var sse: SseReader = .{}; - defer sse.deinit(alloc); - const callbacks = responses_protocol.StreamCallbacks{ + return responses_sse.consume(alloc, reader, .{ .context = callback_ctx, .on_content = on_content_chunk, .on_tool_start = on_tool_start, .on_reasoning = on_reasoning_chunk, .on_tool_input = on_tool_input_chunk, - }; - const stream_limits = responses_protocol.StreamLimits{ + }, cancel_flag, content_capture_limit, .{ + .line_bytes = limits.line_bytes, .aggregate_bytes = limits.aggregate_bytes, .events = limits.events, .tool_calls = limits.tool_calls, .tool_identity_bytes = limits.tool_identity_bytes, .tool_arguments_bytes = limits.tool_arguments_bytes, .provider_state_bytes = limits.provider_state_bytes, - }; - while (try sse.next(alloc, reader)) |json_text| { - defer sse.release(); - if (reducer.applyJson( - alloc, - json_text, - callbacks, - cancel_flag, - content_capture_limit, - stream_limits, - ) catch |err| return mapReducerError(err)) break; - } - return reducer.finish(alloc, cancel_flag, stream_limits) catch |err| - return mapReducerError(err); -} - -fn mapReducerError(err: anyerror) anyerror { - return switch (err) { + }) catch |err| return switch (err) { + error.ResponsesSseReadStalled => error.OpenAICodexSseReadStalled, + error.ResponsesSseEventTooLarge => error.OpenAICodexSseEventTooLarge, error.InvalidEvent => error.InvalidOpenAICodexSseEvent, error.ResponseFailed => error.OpenAICodexResponseFailed, error.StreamIncomplete => error.OpenAICodexStreamIncomplete, diff --git a/src/gateway/responses_sse.zig b/src/gateway/responses_sse.zig new file mode 100644 index 000000000..699a424a6 --- /dev/null +++ b/src/gateway/responses_sse.zig @@ -0,0 +1,107 @@ +const std = @import("std"); +const stream_provider = @import("../core/agent/stream_provider.zig"); +const types = @import("../core/shared/types.zig"); +const responses_protocol = @import("responses_protocol.zig"); + +const Allocator = std.mem.Allocator; + +pub const Limits = struct { + line_bytes: usize, + aggregate_bytes: usize, + events: usize, + tool_calls: usize, + tool_identity_bytes: usize, + tool_arguments_bytes: usize, + provider_state_bytes: usize, +}; + +const Reader = struct { + pending_line: std.ArrayList(u8) = .empty, + + fn deinit(self: *Reader, alloc: Allocator) void { + self.pending_line.deinit(alloc); + } + + fn release(self: *Reader) void { + self.pending_line.clearRetainingCapacity(); + } + + fn next(self: *Reader, alloc: Allocator, reader: anytype, max_line_bytes: usize) !?[]const u8 { + while (true) { + const line = try self.readLine(alloc, reader, max_line_bytes) orelse return null; + const trimmed = std.mem.trim(u8, line, " \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 (std.mem.eql(u8, data, "[DONE]")) return null; + return data; + } + } + + fn readLine(self: *Reader, alloc: Allocator, reader: anytype, max_line_bytes: usize) !?[]const u8 { + while (true) { + const fragment = reader.takeDelimiter('\n') catch |err| switch (err) { + error.StreamTooLong => { + const buffered = reader.buffered(); + if (buffered.len == 0) return error.ResponsesSseReadStalled; + if (buffered.len > max_line_bytes -| self.pending_line.items.len) { + return error.ResponsesSseEventTooLarge; + } + try self.pending_line.appendSlice(alloc, buffered); + reader.tossBuffered(); + continue; + }, + error.ReadFailed => return error.ReadFailed, + } orelse { + if (self.pending_line.items.len > 0) return self.pending_line.items; + return null; + }; + if (fragment.len > max_line_bytes -| self.pending_line.items.len) { + return error.ResponsesSseEventTooLarge; + } + if (self.pending_line.items.len == 0) return fragment; + try self.pending_line.appendSlice(alloc, fragment); + return self.pending_line.items; + } + } +}; + +pub fn consume( + alloc: Allocator, + reader: anytype, + callbacks: responses_protocol.StreamCallbacks, + cancel_flag: *std.atomic.Value(bool), + content_capture_limit: ?usize, + limits: Limits, +) !types.ModelCompletion { + var reducer = responses_protocol.Reducer.init(alloc); + defer reducer.deinit(alloc); + var sse: Reader = .{}; + defer sse.deinit(alloc); + const stream_limits = responses_protocol.StreamLimits{ + .aggregate_bytes = limits.aggregate_bytes, + .events = limits.events, + .tool_calls = limits.tool_calls, + .tool_identity_bytes = limits.tool_identity_bytes, + .tool_arguments_bytes = limits.tool_arguments_bytes, + .provider_state_bytes = limits.provider_state_bytes, + }; + while (try sse.next(alloc, reader, limits.line_bytes)) |json_text| { + defer sse.release(); + if (try reducer.applyJson( + alloc, + json_text, + callbacks, + cancel_flag, + content_capture_limit, + stream_limits, + )) break; + } + return reducer.finish(alloc, cancel_flag, stream_limits); +} diff --git a/src/gateway/responses_transport.zig b/src/gateway/responses_transport.zig new file mode 100644 index 000000000..872851227 --- /dev/null +++ b/src/gateway/responses_transport.zig @@ -0,0 +1,215 @@ +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 gateway_client = @import("client.zig"); +const responses_protocol = @import("responses_protocol.zig"); +const responses_sse = @import("responses_sse.zig"); + +const Allocator = std.mem.Allocator; + +pub const Config = struct { + endpoint: []const u8, + extra_headers: []const std.http.Header = &.{}, + max_error_body_bytes: usize, + error_limit_message: []const u8, + connect_timeout_ms: i64 = 30_000, + stream_limits: responses_sse.Limits, +}; + +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, + authorization: []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 = .{ .override = self.authorization }, + .accept_encoding = .omit, + .user_agent = .{ .override = gateway_client.user_agent }, + }, + .extra_headers = self.extra_headers, + .keep_alive = false, + .redirect_behavior = .unhandled, + }) }; + } +}; + +pub fn stream( + alloc: Allocator, + request: stream_provider.ModelRequest, + payload: []const u8, + config: Config, +) !stream_provider.Result { + var result = stream_inner(alloc, request, payload, config) catch |err| { + if (request.cancel_flag.load(.seq_cst)) return stream_provider.failResult(error.Cancelled); + if (deadline_expired(request.deadline)) return stream_provider.failResult(error.Timeout); + request.attempt_evidence.network_failure = gateway_client.networkFailureEvidence(err, request.delivery.load()); + return err; + }; + if (deadline_expired(request.deadline)) { + result.deinit(alloc); + return stream_provider.failResult(error.Timeout); + } + return result; +} + +fn stream_inner( + alloc: Allocator, + request: stream_provider.ModelRequest, + payload: []const u8, + config: Config, +) !stream_provider.Result { + if (request.cancel_flag.load(.seq_cst)) return stream_provider.failResult(error.Cancelled); + const uri = try std.Uri.parse(config.endpoint); + const authorization = try std.fmt.allocPrint(alloc, "Bearer {s}", .{request.credential.secret}); + defer secret.zeroAndFree(alloc, authorization); + + var client: std.http.Client = .{ .allocator = alloc, .io = io_mod.getIo() }; + defer client.deinit(); + var open_operation = OpenRequestOperation{ + .client = &client, + .uri = uri, + .authorization = authorization, + .extra_headers = config.extra_headers, + }; + var connect_deadline = std.Io.Clock.Timestamp.fromNow(io_mod.getIo(), .{ + .clock = .awake, + .raw = .fromMilliseconds(config.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 detail = reader.allocRemaining(alloc, .limited(config.max_error_body_bytes)) catch |err| switch (err) { + error.StreamTooLong => try alloc.dupe(u8, config.error_limit_message), + else => return err, + }; + return .{ .failed = .{ + .kind = failure_kind(response.head.status), + .detail = detail, + .ownership = .owned, + } }; + } + + var transfer: [256 * 1024]u8 = undefined; + var events = request.events; + const completion = try responses_sse.consume(alloc, response.reader(&transfer), .{ + .context = &events, + .on_content = EventBridge.content, + .on_tool_start = EventBridge.tool_start, + .on_reasoning = EventBridge.reasoning, + .on_tool_input = EventBridge.tool_input, + }, request.cancel_flag, request.content_capture_limit, config.stream_limits); + return .{ .completed = .{ + .completion = completion, + .usage = .{ .unavailable = .possibly_billed }, + .ownership = .owned, + } }; +} + +fn deadline_expired(deadline: ?std.Io.Clock.Timestamp) bool { + const value = deadline orelse return false; + const now = std.Io.Clock.Timestamp.now(io_mod.getIo(), .awake); + return !std.Io.Clock.Timestamp.compare(now, .lt, value); +} + +const EventBridge = struct { + fn sink(raw: *anyopaque) *stream_provider.EventSink { + return @ptrCast(@alignCast(raw)); + } + + fn content(raw: *anyopaque, value: []const u8) void { + sink(raw).emit(.{ .content_delta = value }); + } + + fn reasoning(raw: *anyopaque, value: []const u8) void { + sink(raw).emit(.{ .reasoning_delta = value }); + } + + fn tool_input(raw: *anyopaque, value: []const u8) void { + sink(raw).emit(.{ .tool_input_delta = value }); + } + + fn tool_start(raw: *anyopaque, id: []const u8, name: []const u8, label: ?[]const u8) void { + sink(raw).emit(.{ .tool_started = .{ .id = id, .name = name, .label = label } }); + } +}; + +fn failure_kind(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/main.zig b/src/main.zig index 018c31d50..89c3458aa 100644 --- a/src/main.zig +++ b/src/main.zig @@ -4206,6 +4206,7 @@ test { _ = @import("core/auth/provider_catalog.zig"); _ = @import("gateway/openai_codex_models.zig"); _ = @import("gateway/openai_codex.zig"); + _ = @import("gateway/openai.zig"); _ = @import("gateway/openai_codex_permission_reviewer.zig"); _ = @import("core/auth/grok_session.zig"); _ = @import("core/auth/grok_oauth.zig"); diff --git a/src/ui/footer/model_menu_presentation.zig b/src/ui/footer/model_menu_presentation.zig index f566b2f7c..9d8d73626 100644 --- a/src/ui/footer/model_menu_presentation.zig +++ b/src/ui/footer/model_menu_presentation.zig @@ -414,6 +414,7 @@ fn loadedCatalogStatusText(state: model_cache_runtime.ModelMenuCatalogState) ?[] return switch (source) { .fx_login => "Gateway catalog: authenticated with fx login.", .ai_gateway_api_key => "Note: Gateway catalog is authenticated with an API key", + .openai_api_key => "OpenAI catalog: authenticated with OPENAI_API_KEY.", .vercel_oidc_token => "Gateway catalog: authenticated with the Vercel session.", .stored_key => "Gateway catalog: authenticated with the stored API key.", .chatgpt_subscription => "Codex catalog: authenticated with a subscription.",