From 2d18a4f4aba4720335ab8656a564e592337b3abd Mon Sep 17 00:00:00 2001 From: Alfred Date: Thu, 9 Jul 2026 12:00:02 +0800 Subject: [PATCH] Add non-destructive provider sync and multimodal gateway routing --- .../src-tauri/src/commands.rs | 20 +- apps/codex-plus-manager/src/App.tsx | 140 +++++++++-- .../src/components/ProviderPresetSelector.tsx | 10 +- .../src/model-windows.test.ts | 4 + apps/codex-plus-manager/src/presets.ts | 5 + crates/codex-plus-core/src/ccs_import.rs | 4 + crates/codex-plus-core/src/protocol_proxy.rs | 226 +++++++++++++++++- crates/codex-plus-core/src/provider_import.rs | 4 + crates/codex-plus-core/src/settings.rs | 106 ++++++++ crates/codex-plus-core/tests/launcher.rs | 4 + .../codex-plus-core/tests/protocol_proxy.rs | 184 +++++++++++++- crates/codex-plus-data/src/lib.rs | 5 +- crates/codex-plus-data/src/provider_sync.rs | 134 +++++++++-- crates/codex-plus-data/tests/provider_sync.rs | 165 +++++++++++-- 14 files changed, 944 insertions(+), 67 deletions(-) diff --git a/apps/codex-plus-manager/src-tauri/src/commands.rs b/apps/codex-plus-manager/src-tauri/src/commands.rs index 4456c8546..c292bece3 100644 --- a/apps/codex-plus-manager/src-tauri/src/commands.rs +++ b/apps/codex-plus-manager/src-tauri/src/commands.rs @@ -1160,13 +1160,25 @@ fn merge_manual_provider_sync_targets( } #[tauri::command] -pub async fn sync_providers_now(target_provider: Option) -> CommandResult { +pub async fn sync_providers_now( + target_provider: Option, + rewrite_provider: Option, +) -> CommandResult { let target_provider = target_provider .map(|value| value.trim().to_string()) .filter(|value| !value.is_empty()); let target_for_settings = target_provider.clone(); + let mode = if rewrite_provider.unwrap_or(false) { + codex_plus_data::ProviderSyncMode::RewriteProvider + } else { + codex_plus_data::ProviderSyncMode::NonDestructive + }; let result = tauri::async_runtime::spawn_blocking(move || { - codex_plus_data::run_provider_sync_with_target(None, target_provider.as_deref()) + codex_plus_data::run_provider_sync_with_options( + None, + target_provider.as_deref(), + codex_plus_data::ProviderSyncOptions { mode }, + ) }) .await .map_err(|error| anyhow::anyhow!("provider sync task failed: {error}")); @@ -1188,17 +1200,21 @@ pub async fn sync_providers_now(target_provider: Option) -> CommandResul ), json!({ "syncStatus": sync.status, + "syncMode": sync.mode, "targetProvider": sync.target_provider, "changedSessionFiles": sync.changed_session_files, + "wouldRewriteSessionFiles": sync.would_rewrite_session_files, "skippedLockedRolloutFiles": sync.skipped_locked_rollout_files, "sqliteRowsUpdated": sync.sqlite_rows_updated, "sqliteProviderRowsUpdated": sync.sqlite_provider_rows_updated, + "wouldUpdateProviderRows": sync.would_update_provider_rows, "sqliteUserEventRowsUpdated": sync.sqlite_user_event_rows_updated, "sqliteCwdRowsUpdated": sync.sqlite_cwd_rows_updated, "updatedWorkspaceRoots": sync.updated_workspace_roots, "encryptedContentWarning": sync.encrypted_content_warning, "backupDir": sync.backup_dir, "syncMessage": sync.message, + "destructiveRewritePerformed": sync.destructive_rewrite_performed, }), ) } diff --git a/apps/codex-plus-manager/src/App.tsx b/apps/codex-plus-manager/src/App.tsx index 11a2c709b..a9fa06403 100644 --- a/apps/codex-plus-manager/src/App.tsx +++ b/apps/codex-plus-manager/src/App.tsx @@ -131,6 +131,7 @@ type BackendSettings = { codexAppPath: string; codexExtraArgs: string[]; providerSyncEnabled: boolean; + providerSyncMode: "discoveryOnly" | "nonDestructive" | "rewriteProvider"; providerSyncSavedProviders: string[]; providerSyncManualProviders: string[]; providerSyncLastSelectedProvider: string; @@ -196,6 +197,10 @@ export type RelayProfile = { upstreamBaseUrl: string; apiKey: string; protocol: RelayProtocol; + routeMode: RelayRouteMode; + visionModel: string; + visionBaseUrl: string; + blockGptOnMultimodalGateway: boolean; relayMode: RelayMode; officialMixApiKey: boolean; testModel: string; @@ -256,6 +261,7 @@ type CodexContextEntries = { }; type RelayProtocol = "responses" | "chatCompletions"; +type RelayRouteMode = "direct" | "multimodalGatewayCompat"; type RelayMode = "official" | "mixedApi" | "pureApi" | "aggregate"; const PROTOCOL_PROXY_BASE_URL = "http://127.0.0.1:57321/v1"; const CHAT_UPSTREAM_BASE_URL_KEY = "codex_plus_chat_base_url"; @@ -467,15 +473,19 @@ type RemoveEnvConflictsResult = CommandResult<{ type ProviderSyncPayload = { syncStatus?: string; + syncMode?: "discovery_only" | "non_destructive" | "rewrite_provider"; targetProvider?: string; changedSessionFiles?: number; + wouldRewriteSessionFiles?: number; skippedLockedRolloutFiles?: string[]; sqliteRowsUpdated?: number; sqliteProviderRowsUpdated?: number; + wouldUpdateProviderRows?: number; sqliteUserEventRowsUpdated?: number; sqliteCwdRowsUpdated?: number; updatedWorkspaceRoots?: number; encryptedContentWarning?: string | null; + destructiveRewritePerformed?: boolean; }; type ProviderSyncTargetSource = "config" | "rollout" | "sqlite" | "manual"; @@ -585,8 +595,20 @@ function providerSyncProgressMessage(result: CommandResult) const changed = result.changedSessionFiles ?? 0; const rows = result.sqliteRowsUpdated ?? 0; const target = result.targetProvider || t("当前 provider"); + const mode = result.syncMode || "non_destructive"; const skipped = result.skippedLockedRolloutFiles?.length ?? 0; const skippedText = skipped ? tf("，跳过 {0} 个占用文件", [skipped]) : ""; + if (mode !== "rewrite_provider") { + const wouldRewrite = result.wouldRewriteSessionFiles ?? 0; + const wouldUpdateProviders = result.wouldUpdateProviderRows ?? 0; + return tf("已安全同步到 {0}：更新 {1} 行索引，未重写历史 provider；发现 {2} 个会话文件和 {3} 行 provider 可显式修复{4}。", [ + target, + rows, + wouldRewrite, + wouldUpdateProviders, + skippedText, + ]); + } return tf("已同步到 {0}：修复 {1} 个会话文件，更新 {2} 行索引{3}。", [target, changed, rows, skippedText]); } @@ -653,6 +675,7 @@ const defaultSettings: BackendSettings = { codexAppPath: "", codexExtraArgs: [], providerSyncEnabled: false, + providerSyncMode: "nonDestructive", providerSyncSavedProviders: [], providerSyncManualProviders: [], providerSyncLastSelectedProvider: "", @@ -706,6 +729,10 @@ const defaultSettings: BackendSettings = { upstreamBaseUrl: "", apiKey: "", protocol: "responses", + routeMode: "direct", + visionModel: "", + visionBaseUrl: "", + blockGptOnMultimodalGateway: true, relayMode: "official", officialMixApiKey: false, testModel: "", @@ -1448,12 +1475,14 @@ export function App() { return result; }; - const syncProvidersNow = async () => { + const syncProvidersNow = async (rewriteProvider = false) => { if (providerSyncProgress.active) return; setProviderSyncProgress({ active: true, percent: 12, - message: selectedProviderSyncTarget ? tf("正在同步到 {0}…", [selectedProviderSyncTarget]) : t("正在扫描历史会话与索引…"), + message: selectedProviderSyncTarget + ? tf("正在安全同步到 {0}…", [selectedProviderSyncTarget]) + : t("正在扫描历史会话与索引…"), result: null, }); const progressTimer = window.setInterval(() => { @@ -1462,14 +1491,14 @@ export function App() { return { ...current, percent: Math.min(88, current.percent + 8), - message: current.percent < 40 ? t("正在检查会话 provider 标记…") : t("正在写入修复与备份…"), + message: current.percent < 40 ? t("正在检查会话 provider 标记…") : t("正在写入安全修复与备份…"), }; }); }, 350); try { const targetProvider = selectedProviderSyncTarget || undefined; const result = await run(() => - call>("sync_providers_now", { targetProvider }), + call>("sync_providers_now", { targetProvider, rewriteProvider }), ); if (result) { setProviderSyncProgress({ @@ -1489,12 +1518,12 @@ export function App() { setSettingsForm(next); } await refreshProviderSyncTargets(true); - showNotice(t("历史会话修复"), result.message, result.status); + showNotice(rewriteProvider ? t("历史 provider 重写") : t("历史会话安全同步"), result.message, result.status); } else { setProviderSyncProgress({ active: false, percent: 100, - message: t("历史会话修复失败，请查看错误提示后重试。"), + message: t("历史会话同步失败，请查看错误提示后重试。"), result: null, }); } @@ -2226,7 +2255,7 @@ type Actions = { clearCodexAppPath: () => Promise; chooseImageOverlayPath: () => Promise; saveManualCodexAppPath: () => Promise; - syncProvidersNow: () => Promise; + syncProvidersNow: (rewriteProvider?: boolean) => Promise; refreshProviderSyncTargets: (silent?: boolean) => Promise; setProviderSyncTarget: (provider: string) => void; setLaunchMode: (launchMode: LaunchMode) => Promise; @@ -3056,12 +3085,16 @@ function SessionsScreen({ +
- {providerSyncProgress.active ? t("正在修复历史会话") : t("历史会话修复进度")} + {providerSyncProgress.active ? t("正在同步历史会话") : t("历史会话同步进度")} {providerSyncProgress.percent}%
- {t("删除会创建本地备份；如果 Codex App 正在使用该会话，建议先关闭对应会话窗口再操作。")} + {t("安全同步只修复索引可见性字段，不会改写历史会话 provider；重写历史 provider 会创建备份，但只应作为显式修复操作使用。")}
- + @@ -4189,7 +4222,7 @@ function RelayProfileEditor({
+ + +

+ {t("通用多模态网关会把 Responses 请求转换为兼容文本请求，并把图片分析结果注入上下文。")} +

+ + {profile.routeMode === "multimodalGatewayCompat" ? ( + <> + + updateDraft({ visionModel: event.currentTarget.value })} + placeholder={t("例如 qwen-vl-max")} + /> + + + updateDraft({ visionBaseUrl: event.currentTarget.value })} + placeholder={t("留空则复用上游 Base URL")} + /> + + + + + + ) : null}
) : null} {showApiFields ? ( @@ -5958,6 +6032,10 @@ function normalizeSettings(settings: BackendSettings): BackendSettings { upstreamBaseUrl: settings.relayBaseUrl || defaultSettings.relayBaseUrl, apiKey: settings.relayApiKey || "", protocol: "responses" as RelayProtocol, + routeMode: "direct" as RelayRouteMode, + visionModel: "", + visionBaseUrl: "", + blockGptOnMultimodalGateway: true, relayMode: "official" as RelayMode, officialMixApiKey: false, testModel: "", @@ -6024,6 +6102,10 @@ function normalizeRelayProfile(profile: RelayProfile, defaultContextSelection = upstreamBaseUrl: "", apiKey: "", protocol: "responses", + routeMode: "direct", + visionModel: "", + visionBaseUrl: "", + blockGptOnMultimodalGateway: true, relayMode: "aggregate", officialMixApiKey: false, testModel: profile.testModel || "", @@ -6051,6 +6133,10 @@ function normalizeRelayProfile(profile: RelayProfile, defaultContextSelection = upstreamBaseUrl: profile.upstreamBaseUrl || profile.baseUrl || "", apiKey: profile.apiKey || "", protocol: profile.protocol === "chatCompletions" ? "chatCompletions" : "responses", + routeMode: profile.routeMode === "multimodalGatewayCompat" ? "multimodalGatewayCompat" : "direct", + visionModel: profile.visionModel || "", + visionBaseUrl: profile.visionBaseUrl || "", + blockGptOnMultimodalGateway: profile.blockGptOnMultimodalGateway !== false, relayMode, officialMixApiKey, testModel: profile.testModel || "", @@ -6244,10 +6330,14 @@ function buildRelayConfigToml( options: { includeBearerToken: boolean }, ): string { const baseUrl = profile.protocol === "chatCompletions" ? PROTOCOL_PROXY_BASE_URL : profile.baseUrl.trim(); + const upstreamBaseUrl = profile.upstreamBaseUrl.trim() || profile.baseUrl.trim(); const apiKey = profile.apiKey.trim(); const rootLines = [ profile.model.trim() ? `model = "${tomlString(profile.model.trim())}"` : null, 'model_provider = "custom"', + profile.protocol === "chatCompletions" && upstreamBaseUrl + ? `${CHAT_UPSTREAM_BASE_URL_KEY} = "${tomlString(upstreamBaseUrl)}"` + : null, "", ].filter((line): line is string => line !== null); return [ @@ -6315,6 +6405,9 @@ function applyRelayProfilePatchToFiles( options: { allowGenerateFiles?: boolean } = {}, ): RelayProfile { let next: RelayProfile = { ...profile, ...patch }; + if (patch.routeMode === "multimodalGatewayCompat") { + next.protocol = "chatCompletions"; + } if (isAggregateRelayProfile(next)) { return normalizeAggregateRelayProfile(next, null); } @@ -6345,10 +6438,13 @@ function applyRelayProfilePatchToFiles( if ("upstreamBaseUrl" in patch) { next.baseUrl = patch.upstreamBaseUrl || ""; } - if ("baseUrl" in patch || "upstreamBaseUrl" in patch || "protocol" in patch) { + if ("baseUrl" in patch || "upstreamBaseUrl" in patch || "protocol" in patch || "routeMode" in patch) { const baseUrlForConfig = next.protocol === "chatCompletions" ? PROTOCOL_PROXY_BASE_URL : next.upstreamBaseUrl || next.baseUrl; next.configContents = setCodexProviderStringKey(next.configContents, "base_url", baseUrlForConfig); - next.configContents = removeRootTomlKey(next.configContents, CHAT_UPSTREAM_BASE_URL_KEY); + const upstreamBaseUrl = next.upstreamBaseUrl || next.baseUrl; + next.configContents = next.protocol === "chatCompletions" && upstreamBaseUrl + ? setRootTomlStringKey(next.configContents, CHAT_UPSTREAM_BASE_URL_KEY, upstreamBaseUrl) + : removeRootTomlKey(next.configContents, CHAT_UPSTREAM_BASE_URL_KEY); } if ("contextWindow" in patch) { next.configContents = setRootTomlIntKey(next.configContents, "model_context_window", patch.contextWindow || ""); @@ -6689,6 +6785,10 @@ function createRelayProfile(settings: BackendSettings): RelayProfile { upstreamBaseUrl: defaultSettings.relayBaseUrl, apiKey: "", protocol: "responses" as RelayProtocol, + routeMode: "direct" as RelayRouteMode, + visionModel: "", + visionBaseUrl: "", + blockGptOnMultimodalGateway: true, relayMode: "official" as RelayMode, officialMixApiKey: false, testModel: "", @@ -6719,6 +6819,10 @@ function createAggregateRelayProfile(settings: BackendSettings): RelayProfile { upstreamBaseUrl: "", apiKey: "", protocol: "responses", + routeMode: "direct", + visionModel: "", + visionBaseUrl: "", + blockGptOnMultimodalGateway: true, relayMode: "aggregate", officialMixApiKey: false, testModel: "", @@ -6848,6 +6952,10 @@ function normalizeAggregateRelayProfile(profile: RelayProfile, settings: Backend upstreamBaseUrl: "", apiKey: "", protocol: "responses", + routeMode: "direct", + visionModel: "", + visionBaseUrl: "", + blockGptOnMultimodalGateway: true, relayMode: "aggregate", officialMixApiKey: false, configContents: "", diff --git a/apps/codex-plus-manager/src/components/ProviderPresetSelector.tsx b/apps/codex-plus-manager/src/components/ProviderPresetSelector.tsx index bc86b7463..94dce6a6e 100644 --- a/apps/codex-plus-manager/src/components/ProviderPresetSelector.tsx +++ b/apps/codex-plus-manager/src/components/ProviderPresetSelector.tsx @@ -1,5 +1,5 @@ import { useState, useMemo } from "react"; -import type { ProviderPreset, RelayProtocol } from "../presets"; +import type { ProviderPreset, RelayProtocol, RelayRouteMode } from "../presets"; import { PRESETS } from "../presets"; import { t, tf } from "@/i18n"; @@ -11,6 +11,10 @@ export type RelayProfile = { upstreamBaseUrl: string; apiKey: string; protocol: RelayProtocol; + routeMode: RelayRouteMode; + visionModel: string; + visionBaseUrl: string; + blockGptOnMultimodalGateway: boolean; relayMode: string; officialMixApiKey: boolean; testModel: string; @@ -43,6 +47,10 @@ export function createPresetPatch(preset: ProviderPreset): PresetPatch { baseUrl: preset.baseUrl, upstreamBaseUrl: preset.baseUrl, protocol: preset.protocol, + routeMode: preset.routeMode ?? "direct", + visionModel: preset.visionModel ?? "", + visionBaseUrl: preset.visionBaseUrl ?? "", + blockGptOnMultimodalGateway: preset.blockGptOnMultimodalGateway ?? true, model: preset.model, testModel: preset.model, modelList: preset.modelList?.join("\n") ?? "", diff --git a/apps/codex-plus-manager/src/model-windows.test.ts b/apps/codex-plus-manager/src/model-windows.test.ts index 9f995cea7..e7fc43134 100644 --- a/apps/codex-plus-manager/src/model-windows.test.ts +++ b/apps/codex-plus-manager/src/model-windows.test.ts @@ -19,6 +19,10 @@ const _profileTypeCheck: RelayProfile = { upstreamBaseUrl: "", apiKey: "", protocol: "responses", + routeMode: "direct", + visionModel: "", + visionBaseUrl: "", + blockGptOnMultimodalGateway: true, relayMode: "official", officialMixApiKey: false, testModel: "", diff --git a/apps/codex-plus-manager/src/presets.ts b/apps/codex-plus-manager/src/presets.ts index 07b40e050..34bef38ad 100644 --- a/apps/codex-plus-manager/src/presets.ts +++ b/apps/codex-plus-manager/src/presets.ts @@ -10,6 +10,7 @@ export type PresetCategory = "official" | "aggregator" | "third_party" | "cn_official"; export type RelayProtocol = "responses" | "chatCompletions"; +export type RelayRouteMode = "direct" | "multimodalGatewayCompat"; export interface ProviderPreset { id: string; @@ -19,6 +20,10 @@ export interface ProviderPreset { category: PresetCategory; baseUrl: string; protocol: RelayProtocol; + routeMode?: RelayRouteMode; + visionModel?: string; + visionBaseUrl?: string; + blockGptOnMultimodalGateway?: boolean; model: string; modelList?: string[]; } diff --git a/crates/codex-plus-core/src/ccs_import.rs b/crates/codex-plus-core/src/ccs_import.rs index 0770e1f88..e3fdf7db1 100644 --- a/crates/codex-plus-core/src/ccs_import.rs +++ b/crates/codex-plus-core/src/ccs_import.rs @@ -86,6 +86,10 @@ pub fn relay_profile_from_ccs( upstream_base_url: provider.base_url.clone(), api_key: provider.api_key.clone(), protocol: provider.protocol, + route_mode: Default::default(), + vision_model: String::new(), + vision_base_url: String::new(), + block_gpt_on_multimodal_gateway: true, relay_mode: RelayMode::PureApi, official_mix_api_key: false, test_model: String::new(), diff --git a/crates/codex-plus-core/src/protocol_proxy.rs b/crates/codex-plus-core/src/protocol_proxy.rs index 94178394a..3c7e9bc17 100644 --- a/crates/codex-plus-core/src/protocol_proxy.rs +++ b/crates/codex-plus-core/src/protocol_proxy.rs @@ -10,7 +10,7 @@ use anyhow::Context; use serde_json::{Value, json}; use crate::relay_rotation::{RotationContext, RotationEvent}; -use crate::settings::{RelayProtocol, SettingsStore}; +use crate::settings::{RelayProtocol, RelayRouteMode, SettingsStore}; pub const DEFAULT_PROTOCOL_PROXY_PORT: u16 = 57321; const UPSTREAM_CONNECT_TIMEOUT: Duration = Duration::from_secs(5); @@ -217,6 +217,95 @@ pub fn responses_to_chat_completions(body: Value) -> anyhow::Result { Ok(result) } +pub fn validate_multimodal_gateway_model(model: &str) -> anyhow::Result<()> { + let normalized = model.trim().to_ascii_lowercase(); + if normalized.starts_with("gpt-") || normalized.starts_with("openai/gpt-") { + anyhow::bail!( + "Multimodal gateway does not route OpenAI GPT models; use an official/OpenAI-compatible provider for {model}" + ); + } + Ok(()) +} + +pub fn multimodal_gateway_responses_to_chat_completions( + body: Value, + image_analyses: &[String], +) -> anyhow::Result { + let model = body.get("model").and_then(Value::as_str).unwrap_or(""); + validate_multimodal_gateway_model(model)?; + responses_to_chat_completions(inject_multimodal_gateway_image_analyses( + body, + image_analyses, + )) +} + +async fn multimodal_gateway_image_analyses( + relay: &crate::settings::RelayProfile, + body: &Value, +) -> anyhow::Result> { + let image_urls = collect_responses_image_urls(body.get("input").unwrap_or(&Value::Null)); + if image_urls.is_empty() { + return Ok(Vec::new()); + } + if relay.vision_model.trim().is_empty() { + return Ok(vec![ + "Image analysis unavailable.".to_string(); + image_urls.len() + ]); + } + + let base_url = relay + .vision_base_url + .trim() + .trim_end_matches('/') + .to_string(); + let base_url = if base_url.is_empty() { + relay.base_url.trim().trim_end_matches('/').to_string() + } else { + base_url + }; + let endpoint = chat_completions_url(&base_url); + let client = upstream_http_client()?; + let mut analyses = Vec::with_capacity(image_urls.len()); + for image_url in image_urls { + let request = json!({ + "model": relay.vision_model.trim(), + "messages": [{ + "role": "user", + "content": [ + { "type": "text", "text": "Describe the image for a coding assistant. Be concise and factual." }, + { "type": "image_url", "image_url": image_url } + ] + }], + "stream": false + }); + let response = send_upstream_request_with_header_timeout( + client + .post(&endpoint) + .bearer_auth(relay.api_key.trim()) + .header(reqwest::header::CONTENT_TYPE, "application/json") + .json(&request), + UPSTREAM_HEADER_TIMEOUT, + ) + .await?; + let status = response.status().as_u16(); + let content_type = response + .headers() + .get(reqwest::header::CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .unwrap_or("") + .to_string(); + let bytes = response.bytes().await?; + if !(200..300).contains(&status) { + let error = responses_error_from_upstream(status, &content_type, &bytes); + anyhow::bail!("vision upstream failed: {}", error); + } + let value: Value = serde_json::from_slice(&bytes)?; + analyses.push(extract_vision_analysis_text(&value)); + } + Ok(analyses) +} + pub fn chat_completion_to_response(body: Value) -> anyhow::Result { chat_completion_to_response_with_context(body, &CodexToolContext::default(), None) } @@ -514,7 +603,7 @@ async fn open_responses_proxy_request_with_settings_and_user_agent( for (attempt, relay) in relays.into_iter().enumerate() { validate_upstream(&relay)?; let (endpoint, upstream_body, wire_api) = - upstream_request_parts(&relay, request_json.clone())?; + upstream_request_parts(&relay, request_json.clone()).await?; let has_more_candidates = attempt + 1 < relay_count; let header_timeout = response_header_timeout(is_stream); let _ = crate::diagnostic_log::append_diagnostic_log( @@ -730,7 +819,7 @@ fn response_header_timeout(is_stream: bool) -> Duration { } } -fn upstream_request_parts( +async fn upstream_request_parts( relay: &crate::settings::RelayProfile, request_json: Value, ) -> anyhow::Result<(String, Value, UpstreamWireApi)> { @@ -742,7 +831,12 @@ fn upstream_request_parts( )), RelayProtocol::ChatCompletions => Ok(( chat_completions_url(&relay.base_url), - responses_to_chat_completions(request_json)?, + if relay.route_mode == RelayRouteMode::MultimodalGatewayCompat { + let image_analyses = multimodal_gateway_image_analyses(relay, &request_json).await?; + multimodal_gateway_responses_to_chat_completions(request_json, &image_analyses)? + } else { + responses_to_chat_completions(request_json)? + }, UpstreamWireApi::ChatCompletions, )), } @@ -2213,6 +2307,130 @@ fn responses_content_to_chat_content(_role: &str, content: &Value) -> Value { Value::Array(chat_parts) } +fn inject_multimodal_gateway_image_analyses(mut body: Value, image_analyses: &[String]) -> Value { + let mut analysis_index = 0usize; + if let Some(input) = body.get_mut("input") { + inject_image_analyses_into_responses_input(input, image_analyses, &mut analysis_index); + } + body +} + +fn collect_responses_image_urls(input: &Value) -> Vec { + let mut image_urls = Vec::new(); + collect_responses_image_urls_into(input, &mut image_urls); + image_urls +} + +fn collect_responses_image_urls_into(value: &Value, image_urls: &mut Vec) { + match value { + Value::Array(items) => { + for item in items { + collect_responses_image_urls_into(item, image_urls); + } + } + Value::Object(object) => { + if object.get("type").and_then(Value::as_str) == Some("input_image") { + if let Some(image_url) = object.get("image_url") { + let image_url = if image_url.is_object() { + image_url.clone() + } else { + json!({ "url": image_url.as_str().unwrap_or_default() }) + }; + image_urls.push(image_url); + } + } + if let Some(content) = object.get("content") { + collect_responses_image_urls_into(content, image_urls); + } + } + _ => {} + } +} + +fn extract_vision_analysis_text(value: &Value) -> String { + if let Some(text) = value + .get("choices") + .and_then(Value::as_array) + .and_then(|choices| choices.first()) + .and_then(|choice| choice.get("message")) + .and_then(|message| message.get("content")) + .and_then(Value::as_str) + .filter(|text| !text.trim().is_empty()) + { + return text.trim().to_string(); + } + + if let Some(output) = value.get("output").and_then(Value::as_array) { + let text = output + .iter() + .filter_map(|item| item.get("content").and_then(Value::as_array)) + .flat_map(|parts| parts.iter()) + .filter_map(|part| { + part.get("text") + .and_then(Value::as_str) + .or_else(|| part.get("output_text").and_then(Value::as_str)) + }) + .filter(|text| !text.trim().is_empty()) + .collect::>() + .join("\n\n"); + if !text.trim().is_empty() { + return text.trim().to_string(); + } + } + + "Image analysis unavailable.".to_string() +} + +fn inject_image_analyses_into_responses_input( + input: &mut Value, + image_analyses: &[String], + analysis_index: &mut usize, +) { + match input { + Value::Array(items) => { + for item in items { + inject_image_analyses_into_responses_input(item, image_analyses, analysis_index); + } + } + Value::Object(object) => { + if let Some(content) = object.get_mut("content") { + inject_image_analyses_into_responses_content(content, image_analyses, analysis_index); + } + } + _ => {} + } +} + +fn inject_image_analyses_into_responses_content( + content: &mut Value, + image_analyses: &[String], + analysis_index: &mut usize, +) { + let Some(parts) = content.as_array_mut() else { + return; + }; + + let mut rewritten = Vec::with_capacity(parts.len()); + for part in std::mem::take(parts) { + if part.get("type").and_then(Value::as_str) == Some("input_image") { + let label = *analysis_index + 1; + let analysis = image_analyses + .get(*analysis_index) + .map(String::as_str) + .filter(|value| !value.trim().is_empty()) + .unwrap_or("Image analysis unavailable."); + *analysis_index += 1; + rewritten.push(json!({ + "type": "input_text", + "text": format!("\n[IMAGE ANALYSIS {label}]\n{analysis}") + })); + } else { + rewritten.push(part); + } + } + *parts = rewritten; +} + fn responses_history_function_name(item: &Value) -> String { let name = item.get("name").and_then(Value::as_str).unwrap_or(""); let namespace = item.get("namespace").and_then(Value::as_str).unwrap_or(""); diff --git a/crates/codex-plus-core/src/provider_import.rs b/crates/codex-plus-core/src/provider_import.rs index 26d67caf0..1bbe7bb10 100644 --- a/crates/codex-plus-core/src/provider_import.rs +++ b/crates/codex-plus-core/src/provider_import.rs @@ -196,6 +196,10 @@ fn relay_profile_from_request( upstream_base_url: request.base_url.clone(), api_key: request.api_key.clone(), protocol: relay_protocol(&request.wire_api), + route_mode: Default::default(), + vision_model: String::new(), + vision_base_url: String::new(), + block_gpt_on_multimodal_gateway: true, relay_mode: relay_mode(&request.relay_mode), official_mix_api_key: false, test_model: String::new(), diff --git a/crates/codex-plus-core/src/settings.rs b/crates/codex-plus-core/src/settings.rs index e54611b61..d76cfd5f7 100644 --- a/crates/codex-plus-core/src/settings.rs +++ b/crates/codex-plus-core/src/settings.rs @@ -57,6 +57,22 @@ pub struct RelayProfile { pub api_key: String, #[serde(default)] pub protocol: RelayProtocol, + #[serde(rename = "routeMode", default)] + pub route_mode: RelayRouteMode, + #[serde( + rename = "visionModel", + default, + skip_serializing_if = "String::is_empty" + )] + pub vision_model: String, + #[serde( + rename = "visionBaseUrl", + default, + skip_serializing_if = "String::is_empty" + )] + pub vision_base_url: String, + #[serde(rename = "blockGptOnMultimodalGateway", default = "default_true")] + pub block_gpt_on_multimodal_gateway: bool, #[serde(rename = "relayMode", default)] pub relay_mode: RelayMode, #[serde(rename = "officialMixApiKey", default)] @@ -135,6 +151,10 @@ impl Default for RelayProfile { upstream_base_url: String::new(), api_key: String::new(), protocol: RelayProtocol::Responses, + route_mode: RelayRouteMode::Direct, + vision_model: String::new(), + vision_base_url: String::new(), + block_gpt_on_multimodal_gateway: true, relay_mode: RelayMode::Official, official_mix_api_key: false, test_model: String::new(), @@ -169,6 +189,14 @@ pub enum RelayProtocol { ChatCompletions, } +#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize, Default)] +#[serde(rename_all = "camelCase")] +pub enum RelayRouteMode { + #[default] + Direct, + MultimodalGatewayCompat, +} + #[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize, Default)] #[serde(rename_all = "camelCase")] pub enum RelayMode { @@ -179,6 +207,15 @@ pub enum RelayMode { Aggregate, } +#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize, Default)] +#[serde(rename_all = "camelCase")] +pub enum ProviderSyncMode { + DiscoveryOnly, + #[default] + NonDestructive, + RewriteProvider, +} + #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] pub struct BackendSettings { #[serde(rename = "codexAppPath", default)] @@ -187,6 +224,8 @@ pub struct BackendSettings { pub codex_extra_args: Vec, #[serde(rename = "providerSyncEnabled", default)] pub provider_sync_enabled: bool, + #[serde(rename = "providerSyncMode", default)] + pub provider_sync_mode: ProviderSyncMode, #[serde(rename = "providerSyncSavedProviders", default)] pub provider_sync_saved_providers: Vec, #[serde(rename = "providerSyncManualProviders", default)] @@ -325,6 +364,7 @@ impl Default for BackendSettings { codex_app_path: String::new(), codex_extra_args: Vec::new(), provider_sync_enabled: false, + provider_sync_mode: ProviderSyncMode::NonDestructive, provider_sync_saved_providers: Vec::new(), provider_sync_manual_providers: Vec::new(), provider_sync_last_selected_provider: String::new(), @@ -403,6 +443,10 @@ impl BackendSettings { }, api_key: self.relay_api_key.clone(), protocol: RelayProtocol::Responses, + route_mode: RelayRouteMode::Direct, + vision_model: String::new(), + vision_base_url: String::new(), + block_gpt_on_multimodal_gateway: true, relay_mode: RelayMode::MixedApi, official_mix_api_key: true, test_model: String::new(), @@ -448,6 +492,10 @@ impl BackendSettings { }, api_key: self.relay_api_key.clone(), protocol: RelayProtocol::Responses, + route_mode: RelayRouteMode::Direct, + vision_model: String::new(), + vision_base_url: String::new(), + block_gpt_on_multimodal_gateway: true, relay_mode: RelayMode::Official, official_mix_api_key: false, test_model: String::new(), @@ -759,6 +807,11 @@ fn merge_known_setting_fields(target: &mut Map, source: &Map(value.clone()).is_ok() { + target.insert("providerSyncMode".to_string(), value.clone()); + } + } if let Some(value) = source.get("relayProfilesEnabled").and_then(Value::as_bool) { target.insert("relayProfilesEnabled".to_string(), Value::Bool(value)); } @@ -1217,6 +1270,7 @@ mod tests { fn settings_default_matches_expected_behavior() { let settings = BackendSettings::default(); assert!(!settings.provider_sync_enabled); + assert_eq!(settings.provider_sync_mode, ProviderSyncMode::NonDestructive); assert!(settings.relay_profiles_enabled); assert!(settings.enhancements_enabled); assert!(!settings.computer_use_guard_enabled); @@ -1238,6 +1292,7 @@ mod tests { assert_eq!(settings.relay_base_url, default_relay_base_url()); assert!(settings.relay_api_key.is_empty()); assert_eq!(settings.relay_profiles[0].relay_mode, RelayMode::Official); + assert_eq!(settings.relay_profiles[0].route_mode, RelayRouteMode::Direct); assert!(settings.relay_common_config_contents.is_empty()); assert_eq!(settings.relay_test_model, default_relay_test_model()); assert!(!settings.codex_app_stepwise_enabled); @@ -1263,6 +1318,7 @@ mod tests { .unwrap(); assert_eq!(settings.codex_app_path, r"C:\Portable\Codex\app"); assert!(settings.provider_sync_enabled); + assert_eq!(settings.provider_sync_mode, ProviderSyncMode::NonDestructive); assert!(settings.codex_goals_enabled); assert_eq!(settings.relay_base_url, default_relay_base_url()); assert!(settings.codex_extra_args.is_empty()); @@ -2102,6 +2158,56 @@ experimental_bearer_token = "sk-existing""# assert_eq!(saved["customField"], json!({"nested": true})); } + #[test] + fn settings_store_update_persists_provider_sync_mode() { + let dir = temp_dir(); + let path = dir.join("settings.json"); + let store = SettingsStore::new(path.clone()); + std::fs::write( + &path, + r#"{"providerSyncEnabled":true,"providerSyncMode":"nonDestructive","customField":{"nested":true}}"#, + ) + .unwrap(); + + let updated = store + .update(json!({ + "providerSyncMode": "rewriteProvider" + })) + .unwrap(); + let saved: Value = serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap(); + + assert_eq!(updated.provider_sync_mode, ProviderSyncMode::RewriteProvider); + assert_eq!(saved["providerSyncMode"], json!("rewriteProvider")); + assert_eq!(saved["customField"], json!({"nested": true})); + } + + #[test] + fn relay_profile_deserializes_multimodal_gateway_route_fields() { + let settings: BackendSettings = serde_json::from_str( + r#"{ + "relayProfiles": [{ + "id": "gateway", + "name": "Gateway", + "baseUrl": "https://gateway.example/v1", + "apiKey": "sk-test", + "protocol": "chatCompletions", + "routeMode": "multimodalGatewayCompat", + "visionModel": "qwen-vl-max", + "visionBaseUrl": "https://vision.example/v1", + "blockGptOnMultimodalGateway": true + }], + "activeRelayId": "gateway" + }"#, + ) + .unwrap(); + + let profile = settings.active_relay_profile(); + assert_eq!(profile.route_mode, RelayRouteMode::MultimodalGatewayCompat); + assert_eq!(profile.vision_model, "qwen-vl-max"); + assert_eq!(profile.vision_base_url, "https://vision.example/v1"); + assert!(profile.block_gpt_on_multimodal_gateway); + } + #[test] fn settings_store_update_persists_codex_extra_args_and_preserves_unknown_fields() { let dir = temp_dir(); diff --git a/crates/codex-plus-core/tests/launcher.rs b/crates/codex-plus-core/tests/launcher.rs index d2df5f0b7..83a14a40d 100644 --- a/crates/codex-plus-core/tests/launcher.rs +++ b/crates/codex-plus-core/tests/launcher.rs @@ -1082,6 +1082,10 @@ async fn launch_starts_helper_when_chat_protocol_proxy_is_enabled() { upstream_base_url: "https://chat-only.example.test/v1".to_string(), api_key: "sk-test".to_string(), protocol: RelayProtocol::ChatCompletions, + route_mode: Default::default(), + vision_model: String::new(), + vision_base_url: String::new(), + block_gpt_on_multimodal_gateway: true, relay_mode: codex_plus_core::settings::RelayMode::MixedApi, official_mix_api_key: false, test_model: String::new(), diff --git a/crates/codex-plus-core/tests/protocol_proxy.rs b/crates/codex-plus-core/tests/protocol_proxy.rs index 795796cad..5ec7582ef 100644 --- a/crates/codex-plus-core/tests/protocol_proxy.rs +++ b/crates/codex-plus-core/tests/protocol_proxy.rs @@ -1,16 +1,18 @@ use codex_plus_core::protocol_proxy::{ ChatSseToResponsesConverter, chat_completion_to_response, chat_completion_to_response_with_request, chat_completions_url, chat_sse_to_responses_sse, - chat_sse_to_responses_sse_with_request, is_chat_completions_proxy_path, is_models_proxy_path, - is_responses_proxy_path, models_url, open_chat_completions_proxy_request, + chat_sse_to_responses_sse_with_request, is_chat_completions_proxy_path, + is_models_proxy_path, is_responses_proxy_path, models_url, + multimodal_gateway_responses_to_chat_completions, open_chat_completions_proxy_request, open_models_proxy_request, open_responses_proxy_request, open_responses_proxy_request_with_settings, responses_error_from_upstream, responses_to_chat_completions, send_upstream_request_with_header_timeout, upstream_header_timeout, upstream_http_client, upstream_stream_header_timeout, + validate_multimodal_gateway_model, }; use codex_plus_core::settings::{ AggregateRelayMember, AggregateRelayProfile, AggregateRelayStrategy, BackendSettings, - RelayMode, RelayProfile, + RelayMode, RelayProfile, RelayRouteMode, }; use serde_json::json; use std::io::{Read, Write}; @@ -208,6 +210,58 @@ fn responses_request_maps_developer_role_to_system_for_chat_upstream() { ); } +#[test] +fn multimodal_gateway_rejects_openai_gpt_models() { + let error = validate_multimodal_gateway_model("gpt-5.5").unwrap_err(); + assert!( + error + .to_string() + .contains("Multimodal gateway does not route OpenAI GPT models") + ); + + validate_multimodal_gateway_model("glm-5.2").unwrap(); + validate_multimodal_gateway_model("deepseek-v4-pro").unwrap(); +} + +#[test] +fn multimodal_gateway_removes_images_and_injects_analysis_text() { + let converted = multimodal_gateway_responses_to_chat_completions( + json!({ + "model": "glm-5.2", + "input": [ + { + "type": "message", + "role": "developer", + "content": [{ "type": "input_text", "text": "Be precise." }] + }, + { + "type": "message", + "role": "user", + "content": [ + { "type": "input_text", "text": "What is in this image?" }, + { "type": "input_image", "image_url": "data:image/png;base64,abc" } + ] + } + ] + }), + &["A screenshot of a project sidebar.".to_string()], + ) + .unwrap(); + + assert_eq!(converted["messages"][0]["role"], "system"); + assert_eq!(converted["messages"][0]["content"], "Be precise."); + assert_eq!(converted["messages"][1]["role"], "user"); + assert_eq!( + converted["messages"][1]["content"], + "What is in this image?\n\n[IMAGE ANALYSIS 1]\nA screenshot of a project sidebar." + ); + assert!( + !serde_json::to_string(&converted) + .unwrap() + .contains("image_url") + ); +} + #[test] fn responses_request_collapses_system_messages_to_head_for_strict_chat_upstreams() { let converted = responses_to_chat_completions(json!({ @@ -1550,6 +1604,115 @@ async fn responses_proxy_passes_through_original_user_agent_when_unconfigured() assert_eq!(request.user_agent, "Original-Codex-UA/1.0"); } +#[tokio::test] +async fn responses_proxy_uses_multimodal_gateway_route_mode_for_chat_upstream() { + let settings = { + let server = spawn_chat_server(); + let settings = BackendSettings { + relay_profiles: vec![RelayProfile { + id: "gateway".to_string(), + name: "Gateway".to_string(), + base_url: server.base_url.clone(), + api_key: "sk-test".to_string(), + protocol: codex_plus_core::settings::RelayProtocol::ChatCompletions, + relay_mode: RelayMode::MixedApi, + route_mode: RelayRouteMode::MultimodalGatewayCompat, + ..RelayProfile::default() + }], + active_relay_id: "gateway".to_string(), + ..BackendSettings::default() + }; + (settings, server) + }; + let (settings, server) = settings; + + let upstream = open_responses_proxy_request_with_settings( + r#"{"model":"glm-5.2","input":[{"type":"message","role":"user","content":[{"type":"input_text","text":"describe"},{"type":"input_image","image_url":"data:image/png;base64,abc"}]}],"stream":false}"#, + settings, + ) + .await + .unwrap(); + assert_eq!(upstream.status_code, 200); + + let request = server.finish(); + let body: serde_json::Value = serde_json::from_str(&request.body).unwrap(); + assert_eq!(body["messages"][0]["content"], "describe\n\n[IMAGE ANALYSIS 1]\nImage analysis unavailable."); + assert!(!request.body.contains("image_url")); +} + +#[tokio::test] +async fn multimodal_gateway_calls_configured_vision_upstream_before_main_request() { + let main_server = spawn_chat_server(); + let vision_server = spawn_chat_server_with_body( + r#"{"id":"chatcmpl-vision","object":"chat.completion","choices":[{"message":{"role":"assistant","content":"Detected screenshot"}}]}"#, + ); + let settings = BackendSettings { + relay_profiles: vec![RelayProfile { + id: "gateway".to_string(), + name: "Gateway".to_string(), + base_url: main_server.base_url.clone(), + vision_base_url: vision_server.base_url.clone(), + vision_model: "qwen-vl-max".to_string(), + api_key: "sk-test".to_string(), + protocol: codex_plus_core::settings::RelayProtocol::ChatCompletions, + relay_mode: RelayMode::MixedApi, + route_mode: RelayRouteMode::MultimodalGatewayCompat, + ..RelayProfile::default() + }], + active_relay_id: "gateway".to_string(), + ..BackendSettings::default() + }; + + let upstream = open_responses_proxy_request_with_settings( + r#"{"model":"glm-5.2","input":[{"type":"message","role":"user","content":[{"type":"input_image","image_url":"data:image/png;base64,abc"}]}],"stream":false}"#, + settings, + ) + .await + .unwrap(); + assert_eq!(upstream.status_code, 200); + + let vision_request = vision_server.finish(); + assert!(vision_request.body.contains("qwen-vl-max")); + assert!(vision_request.body.contains("image_url")); + let main_request = main_server.finish(); + assert!(main_request.body.contains("Detected screenshot")); + assert!(!main_request.body.contains("data:image/png")); +} + +#[tokio::test] +async fn responses_proxy_blocks_gpt_models_in_multimodal_gateway_route_mode() { + let settings = BackendSettings { + relay_profiles: vec![RelayProfile { + id: "gateway".to_string(), + name: "Gateway".to_string(), + base_url: "http://127.0.0.1:9/v1".to_string(), + api_key: "sk-test".to_string(), + protocol: codex_plus_core::settings::RelayProtocol::ChatCompletions, + relay_mode: RelayMode::MixedApi, + route_mode: RelayRouteMode::MultimodalGatewayCompat, + ..RelayProfile::default() + }], + active_relay_id: "gateway".to_string(), + ..BackendSettings::default() + }; + + let error = match open_responses_proxy_request_with_settings( + r#"{"model":"gpt-5.5","input":"hello","stream":false}"#, + settings, + ) + .await + { + Ok(_) => panic!("expected multimodal gateway to reject GPT models"), + Err(error) => error, + }; + + assert!( + error + .to_string() + .contains("Multimodal gateway does not route OpenAI GPT models") + ); +} + #[tokio::test] async fn models_proxy_passes_through_original_user_agent_when_unconfigured() { let _lock = settings_path_test_lock().lock().unwrap(); @@ -1623,9 +1786,14 @@ impl ChatServer { struct ChatRequest { user_agent: String, + body: String, } fn spawn_chat_server() -> ChatServer { + spawn_chat_server_with_body(r#"{"id":"chatcmpl-test","object":"chat.completion","choices":[]}"#) +} + +fn spawn_chat_server_with_body(body: &'static str) -> ChatServer { let listener = TcpListener::bind(("127.0.0.1", 0)).unwrap(); let address = listener.local_addr().unwrap(); let base_url = format!("http://{address}/v1"); @@ -1657,6 +1825,10 @@ fn spawn_chat_server() -> ChatServer { } }; let request = String::from_utf8_lossy(&buffer[..bytes]).to_string(); + let request_body = request + .split_once("\r\n\r\n") + .map(|(_, body)| body.to_string()) + .unwrap_or_default(); let user_agent = request .lines() .find_map(|line| { @@ -1666,14 +1838,16 @@ fn spawn_chat_server() -> ChatServer { }) }) .unwrap_or_default(); - let body = r#"{"id":"chatcmpl-test","object":"chat.completion","choices":[]}"#; let response = format!( "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", body.len(), body ); stream.write_all(response.as_bytes()).unwrap(); - ChatRequest { user_agent } + ChatRequest { + user_agent, + body: request_body, + } }); ChatServer { base_url, handle } } diff --git a/crates/codex-plus-data/src/lib.rs b/crates/codex-plus-data/src/lib.rs index 4ea138dcc..d7cca1dd0 100644 --- a/crates/codex-plus-data/src/lib.rs +++ b/crates/codex-plus-data/src/lib.rs @@ -6,8 +6,9 @@ pub mod storage; pub use backup::BackupStore; pub use markdown::{MarkdownExportService, export_markdown_from_paths}; pub use provider_sync::{ - ProviderSyncResult, ProviderSyncStatus, ProviderSyncTargetList, ProviderSyncTargetOption, - ProviderSyncTargetSource, load_provider_sync_targets, run_provider_sync, + ProviderSyncMode, ProviderSyncOptions, ProviderSyncResult, ProviderSyncStatus, + ProviderSyncTargetList, ProviderSyncTargetOption, ProviderSyncTargetSource, + load_provider_sync_targets, run_provider_sync, run_provider_sync_with_options, run_provider_sync_with_target, }; pub use storage::{ diff --git a/crates/codex-plus-data/src/provider_sync.rs b/crates/codex-plus-data/src/provider_sync.rs index e4f7f443c..4a0719f6a 100644 --- a/crates/codex-plus-data/src/provider_sync.rs +++ b/crates/codex-plus-data/src/provider_sync.rs @@ -18,20 +18,43 @@ pub enum ProviderSyncStatus { Synced, } +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ProviderSyncMode { + DiscoveryOnly, + NonDestructive, + RewriteProvider, +} + +impl Default for ProviderSyncMode { + fn default() -> Self { + Self::NonDestructive + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)] +pub struct ProviderSyncOptions { + pub mode: ProviderSyncMode, +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct ProviderSyncResult { pub status: ProviderSyncStatus, pub message: String, + pub mode: ProviderSyncMode, pub target_provider: String, pub backup_dir: Option, pub changed_session_files: usize, + pub would_rewrite_session_files: usize, pub skipped_locked_rollout_files: Vec, pub sqlite_rows_updated: usize, pub sqlite_provider_rows_updated: usize, + pub would_update_provider_rows: usize, pub sqlite_user_event_rows_updated: usize, pub sqlite_cwd_rows_updated: usize, pub updated_workspace_roots: usize, pub encrypted_content_warning: Option, + pub destructive_rewrite_performed: bool, } #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)] @@ -123,6 +146,18 @@ pub fn run_provider_sync(codex_home: Option<&Path>) -> ProviderSyncResult { pub fn run_provider_sync_with_target( codex_home: Option<&Path>, explicit_target_provider: Option<&str>, +) -> ProviderSyncResult { + run_provider_sync_with_options( + codex_home, + explicit_target_provider, + ProviderSyncOptions::default(), + ) +} + +pub fn run_provider_sync_with_options( + codex_home: Option<&Path>, + explicit_target_provider: Option<&str>, + options: ProviderSyncOptions, ) -> ProviderSyncResult { let home = codex_home .map(Path::to_path_buf) @@ -135,6 +170,7 @@ pub fn run_provider_sync_with_target( None, 0, 0, + options.mode, ); } let target_provider = @@ -148,6 +184,7 @@ pub fn run_provider_sync_with_target( None, 0, 0, + options.mode, ); } }; @@ -160,6 +197,7 @@ pub fn run_provider_sync_with_target( None, 0, 0, + options.mode, ); } let sync_result = (|| -> anyhow::Result { @@ -187,7 +225,7 @@ pub fn run_provider_sync_with_target( .filter(|(thread_id, _)| !projectless_thread_ids.contains(thread_id)) .collect::>(); let sqlite_paths = codex_plus_core::codex_sqlite::codex_session_db_paths_from_home(&home); - let sqlite_update_count = count_sqlite_updates_for_paths( + let sqlite_update_counts = count_sqlite_updates_for_paths( &sqlite_paths, &target_provider, &thread_ids_with_user_events, @@ -195,7 +233,27 @@ pub fn run_provider_sync_with_target( )?; let global_state_update_count = count_global_state_updates(&home.join(".codex-global-state.json"))?; - if rewrite_changes.is_empty() && sqlite_update_count == 0 && global_state_update_count == 0 + let would_rewrite_session_files = rewrite_changes.len(); + let safe_sqlite_update_count = + sqlite_update_counts.user_event_rows + sqlite_update_counts.cwd_rows; + let actual_sqlite_update_count = match options.mode { + ProviderSyncMode::DiscoveryOnly => 0, + ProviderSyncMode::NonDestructive => safe_sqlite_update_count, + ProviderSyncMode::RewriteProvider => sqlite_update_counts.total(), + }; + let actual_global_state_update_count = match options.mode { + ProviderSyncMode::DiscoveryOnly => 0, + ProviderSyncMode::NonDestructive | ProviderSyncMode::RewriteProvider => { + global_state_update_count + } + }; + let actual_session_rewrite_count = match options.mode { + ProviderSyncMode::RewriteProvider => rewrite_changes.len(), + ProviderSyncMode::DiscoveryOnly | ProviderSyncMode::NonDestructive => 0, + }; + if actual_session_rewrite_count == 0 + && actual_sqlite_update_count == 0 + && actual_global_state_update_count == 0 { let mut synced = result( ProviderSyncStatus::Synced, @@ -204,22 +262,37 @@ pub fn run_provider_sync_with_target( None, 0, 0, + options.mode, ); + synced.would_rewrite_session_files = would_rewrite_session_files; + synced.would_update_provider_rows = sqlite_update_counts.provider_rows; synced.skipped_locked_rollout_files = collected.skipped_locked_rollout_files; synced.encrypted_content_warning = encrypted_content_warning; return Ok(synced); } - let backup_dir = create_backup(&home, &target_provider, &rewrite_changes)?; - let applied = apply_session_changes(&rewrite_changes)?; + let session_changes_to_apply = match options.mode { + ProviderSyncMode::RewriteProvider => rewrite_changes.as_slice(), + ProviderSyncMode::DiscoveryOnly | ProviderSyncMode::NonDestructive => &[], + }; + let backup_dir = if options.mode == ProviderSyncMode::DiscoveryOnly { + None + } else { + Some(create_backup(&home, &target_provider, session_changes_to_apply)?) + }; + let applied = apply_session_changes(session_changes_to_apply)?; let apply_result = (|| -> anyhow::Result<(SqliteUpdateCounts, usize)> { let sqlite_updates = apply_sqlite_update_for_paths( &sqlite_paths, &target_provider, &thread_ids_with_user_events, &cwd_by_thread_id, + options.mode == ProviderSyncMode::RewriteProvider, )?; - let updated_workspace_roots = - apply_global_state_update(&home.join(".codex-global-state.json"))?; + let updated_workspace_roots = if options.mode == ProviderSyncMode::DiscoveryOnly { + 0 + } else { + apply_global_state_update(&home.join(".codex-global-state.json"))? + }; prune_backups(&home)?; Ok((sqlite_updates, updated_workspace_roots)) })(); @@ -234,10 +307,13 @@ pub fn run_provider_sync_with_target( ProviderSyncStatus::Synced, "Provider sync complete", &target_provider, - Some(backup_dir), + backup_dir, applied.changes.len(), sqlite_updates.total(), + options.mode, ); + synced.would_rewrite_session_files = would_rewrite_session_files; + synced.would_update_provider_rows = sqlite_update_counts.provider_rows; synced.skipped_locked_rollout_files = collected.skipped_locked_rollout_files; synced .skipped_locked_rollout_files @@ -249,6 +325,8 @@ pub fn run_provider_sync_with_target( synced.sqlite_cwd_rows_updated = sqlite_updates.cwd_rows; synced.updated_workspace_roots = updated_workspace_roots; synced.encrypted_content_warning = encrypted_content_warning; + synced.destructive_rewrite_performed = options.mode == ProviderSyncMode::RewriteProvider + && (applied.changes.len() > 0 || sqlite_updates.provider_rows > 0); Ok(synced) })(); let _ = release_lock(&lock_dir); @@ -260,6 +338,7 @@ pub fn run_provider_sync_with_target( None, 0, 0, + options.mode, ) }) } @@ -271,20 +350,25 @@ fn result( backup_dir: Option, changed_session_files: usize, sqlite_rows_updated: usize, + mode: ProviderSyncMode, ) -> ProviderSyncResult { ProviderSyncResult { status, message: message.into(), + mode, target_provider: target_provider.to_string(), backup_dir, changed_session_files, + would_rewrite_session_files: 0, skipped_locked_rollout_files: Vec::new(), sqlite_rows_updated, sqlite_provider_rows_updated: 0, + would_update_provider_rows: 0, sqlite_user_event_rows_updated: 0, sqlite_cwd_rows_updated: 0, updated_workspace_roots: 0, encrypted_content_warning: None, + destructive_rewrite_performed: false, } } @@ -832,23 +916,24 @@ fn count_sqlite_updates( target_provider: &str, user_event_thread_ids: &HashSet, cwd_by_thread_id: &HashMap, -) -> anyhow::Result { +) -> anyhow::Result { if !path.exists() { - return Ok(0); + return Ok(SqliteUpdateCounts::default()); } let db = Connection::open(path)?; let columns = table_columns(&db, "threads")?; if !columns.contains("model_provider") { - return Ok(0); + return Ok(SqliteUpdateCounts::default()); } - let mut total: usize = db.query_row( + let mut counts = SqliteUpdateCounts::default(); + counts.provider_rows = db.query_row( "SELECT COUNT(*) FROM threads WHERE COALESCE(model_provider, '') <> ?1", [target_provider], |row| row.get::<_, i64>(0), )? as usize; if columns.contains("has_user_event") { for thread_id in user_event_thread_ids { - total += db.query_row( + counts.user_event_rows += db.query_row( "SELECT COUNT(*) FROM threads WHERE id = ?1 AND COALESCE(has_user_event, 0) <> 1", [thread_id], |row| row.get::<_, i64>(0), @@ -857,14 +942,14 @@ fn count_sqlite_updates( } if columns.contains("cwd") { for (thread_id, cwd) in cwd_by_thread_id { - total += db.query_row( + counts.cwd_rows += db.query_row( "SELECT COUNT(*) FROM threads WHERE id = ?1 AND COALESCE(cwd, '') <> ?2", (thread_id, cwd), |row| row.get::<_, i64>(0), )? as usize; } } - Ok(total) + Ok(counts) } fn count_sqlite_updates_for_paths( @@ -872,15 +957,15 @@ fn count_sqlite_updates_for_paths( target_provider: &str, user_event_thread_ids: &HashSet, cwd_by_thread_id: &HashMap, -) -> anyhow::Result { - let mut total = 0; +) -> anyhow::Result { + let mut total = SqliteUpdateCounts::default(); for path in paths { - total += count_sqlite_updates( + total.add(count_sqlite_updates( path, target_provider, user_event_thread_ids, cwd_by_thread_id, - )?; + )?); } Ok(total) } @@ -890,6 +975,7 @@ fn apply_sqlite_update( target_provider: &str, user_event_thread_ids: &HashSet, cwd_by_thread_id: &HashMap, + rewrite_provider: bool, ) -> anyhow::Result { if !path.exists() { return Ok(SqliteUpdateCounts::default()); @@ -901,10 +987,12 @@ fn apply_sqlite_update( } let tx = db.transaction()?; let mut counts = SqliteUpdateCounts::default(); - counts.provider_rows = tx.execute( - "UPDATE threads SET model_provider = ?1 WHERE COALESCE(model_provider, '') <> ?1", - [target_provider], - )?; + if rewrite_provider { + counts.provider_rows = tx.execute( + "UPDATE threads SET model_provider = ?1 WHERE COALESCE(model_provider, '') <> ?1", + [target_provider], + )?; + } if columns.contains("has_user_event") { for thread_id in user_event_thread_ids { counts.user_event_rows += tx.execute( @@ -930,6 +1018,7 @@ fn apply_sqlite_update_for_paths( target_provider: &str, user_event_thread_ids: &HashSet, cwd_by_thread_id: &HashMap, + rewrite_provider: bool, ) -> anyhow::Result { let mut total = SqliteUpdateCounts::default(); for path in paths { @@ -938,6 +1027,7 @@ fn apply_sqlite_update_for_paths( target_provider, user_event_thread_ids, cwd_by_thread_id, + rewrite_provider, )?); } Ok(total) diff --git a/crates/codex-plus-data/tests/provider_sync.rs b/crates/codex-plus-data/tests/provider_sync.rs index 258493964..a54a5adb5 100644 --- a/crates/codex-plus-data/tests/provider_sync.rs +++ b/crates/codex-plus-data/tests/provider_sync.rs @@ -1,5 +1,6 @@ use codex_plus_data::{ - ProviderSyncStatus, ProviderSyncTargetSource, load_provider_sync_targets, run_provider_sync, + ProviderSyncMode, ProviderSyncOptions, ProviderSyncStatus, ProviderSyncTargetSource, + load_provider_sync_targets, run_provider_sync, run_provider_sync_with_options, run_provider_sync_with_target, }; use rusqlite::Connection; @@ -74,6 +75,140 @@ fn create_state_db_with_providers(path: &Path, rows: &[(&str, &str, i64)]) { } } +fn run_provider_rewrite_sync(home: &Path) -> codex_plus_data::ProviderSyncResult { + run_provider_sync_with_options( + Some(home), + None, + ProviderSyncOptions { + mode: ProviderSyncMode::RewriteProvider, + }, + ) +} + +fn run_provider_rewrite_sync_with_target( + home: &Path, + target_provider: &str, +) -> codex_plus_data::ProviderSyncResult { + run_provider_sync_with_options( + Some(home), + Some(target_provider), + ProviderSyncOptions { + mode: ProviderSyncMode::RewriteProvider, + }, + ) +} + +#[test] +fn provider_sync_non_destructive_does_not_rewrite_rollout_model_provider() { + let tmp = tempdir().unwrap(); + let home = tmp.path().join(".codex"); + fs::create_dir(&home).unwrap(); + fs::write( + home.join("config.toml"), + r#"model_provider = "custom_gateway" + +[model_providers.openai] +name = "OpenAI" + +[model_providers.custom_gateway] +name = "Custom Gateway" +"#, + ) + .unwrap(); + let rollout = home.join("sessions/2026/rollout-openai.jsonl"); + write_rollout(&rollout, "openai", "thread-1", "C:/workspace"); + create_state_db(&home.join("state_5.sqlite")); + + let result = run_provider_sync(Some(&home)); + + assert_eq!(result.status, ProviderSyncStatus::Synced); + assert_eq!(result.mode, ProviderSyncMode::NonDestructive); + assert_eq!(result.changed_session_files, 0); + assert_eq!(result.would_rewrite_session_files, 1); + assert!(!result.destructive_rewrite_performed); + let first: serde_json::Value = serde_json::from_str( + fs::read_to_string(&rollout) + .unwrap() + .lines() + .next() + .unwrap(), + ) + .unwrap(); + assert_eq!(first["payload"]["model_provider"], "openai"); +} + +#[test] +fn provider_sync_non_destructive_does_not_rewrite_sqlite_provider() { + let tmp = tempdir().unwrap(); + let home = tmp.path().join(".codex"); + fs::create_dir(&home).unwrap(); + fs::write(home.join("config.toml"), "model_provider = \"custom_gateway\"\n").unwrap(); + write_rollout( + &home.join("sessions/2026/rollout-openai.jsonl"), + "openai", + "thread-1", + "C:/workspace", + ); + create_state_db(&home.join("state_5.sqlite")); + + let result = run_provider_sync(Some(&home)); + + assert_eq!(result.status, ProviderSyncStatus::Synced); + assert_eq!(result.sqlite_provider_rows_updated, 0); + assert_eq!(result.would_update_provider_rows, 1); + let db = Connection::open(home.join("state_5.sqlite")).unwrap(); + let provider: String = db + .query_row( + "SELECT model_provider FROM threads WHERE id = 'thread-1'", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(provider, "old-provider"); +} + +#[test] +fn provider_sync_rewrite_mode_preserves_existing_destructive_behavior() { + let tmp = tempdir().unwrap(); + let home = tmp.path().join(".codex"); + fs::create_dir(&home).unwrap(); + fs::write(home.join("config.toml"), "model_provider = \"custom_gateway\"\n").unwrap(); + let rollout = home.join("sessions/2026/rollout-openai.jsonl"); + write_rollout(&rollout, "openai", "thread-1", "C:/workspace"); + create_state_db(&home.join("state_5.sqlite")); + + let result = run_provider_sync_with_options( + Some(&home), + None, + ProviderSyncOptions { + mode: ProviderSyncMode::RewriteProvider, + }, + ); + + assert_eq!(result.status, ProviderSyncStatus::Synced); + assert_eq!(result.mode, ProviderSyncMode::RewriteProvider); + assert_eq!(result.changed_session_files, 1); + assert!(result.destructive_rewrite_performed); + let first: serde_json::Value = serde_json::from_str( + fs::read_to_string(&rollout) + .unwrap() + .lines() + .next() + .unwrap(), + ) + .unwrap(); + assert_eq!(first["payload"]["model_provider"], "custom_gateway"); + let db = Connection::open(home.join("state_5.sqlite")).unwrap(); + let provider: String = db + .query_row( + "SELECT model_provider FROM threads WHERE id = 'thread-1'", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(provider, "custom_gateway"); +} + #[test] fn provider_sync_targets_merge_config_rollout_sqlite_and_sort_current_first() { let tmp = tempdir().unwrap(); @@ -174,7 +309,7 @@ experimental_bearer_token = "sk-test" write_rollout(&rollout, "openai", "thread-1", "C:/workspace"); create_state_db(&home.join("state_5.sqlite")); - let result = run_provider_sync(Some(&home)); + let result = run_provider_rewrite_sync(&home); assert_eq!(result.status, ProviderSyncStatus::Synced); assert_eq!(result.target_provider, "custom"); @@ -215,7 +350,7 @@ fn provider_sync_rewrites_all_session_meta_model_providers() { ); create_state_db(&home.join("state_5.sqlite")); - let result = run_provider_sync(Some(&home)); + let result = run_provider_rewrite_sync(&home); assert_eq!(result.status, ProviderSyncStatus::Synced); assert_eq!(result.target_provider, "apigather"); @@ -271,7 +406,7 @@ fn provider_sync_updates_rollout_sqlite_visibility_and_creates_backup() { write_rollout(&rollout, "openai", "thread-1", "C:/workspace"); create_state_db(&home.join("state_5.sqlite")); - let result = run_provider_sync(Some(&home)); + let result = run_provider_rewrite_sync(&home); assert_eq!(result.status, ProviderSyncStatus::Synced); assert_eq!(result.target_provider, "apigather"); @@ -324,7 +459,7 @@ fn provider_sync_updates_new_codex_sqlite_directory_db() { let db_path = sqlite_dir.join("codex-dev.db"); create_state_db(&db_path); - let result = run_provider_sync(Some(&home)); + let result = run_provider_rewrite_sync(&home); assert_eq!(result.status, ProviderSyncStatus::Synced); assert_eq!(result.sqlite_rows_updated, 3); @@ -364,7 +499,7 @@ fn provider_sync_backup_metadata_contains_reference_fields_and_managed_marker() ); create_state_db(&home.join("state_5.sqlite")); - let result = run_provider_sync(Some(&home)); + let result = run_provider_rewrite_sync(&home); assert_eq!(result.status, ProviderSyncStatus::Synced); let backup_dir = result.backup_dir.unwrap(); @@ -396,7 +531,7 @@ fn provider_sync_explicit_target_overrides_config_without_switching_config() { write_rollout(&rollout, "openai", "thread-1", "C:/workspace"); create_state_db(&home.join("state_5.sqlite")); - let result = run_provider_sync_with_target(Some(&home), Some("custom")); + let result = run_provider_rewrite_sync_with_target(&home, "custom"); assert_eq!(result.status, ProviderSyncStatus::Synced); assert_eq!(result.target_provider, "custom"); @@ -467,7 +602,7 @@ fn provider_sync_repairs_sqlite_when_rollout_provider_matches_and_normalizes_pat ) .unwrap(); - let result = run_provider_sync(Some(&home)); + let result = run_provider_rewrite_sync(&home); assert_eq!(result.status, ProviderSyncStatus::Synced); assert_eq!(result.changed_session_files, 0); @@ -519,7 +654,7 @@ fn provider_sync_does_not_restore_cwd_for_projectless_threads() { ) .unwrap(); - let result = run_provider_sync(Some(&home)); + let result = run_provider_rewrite_sync(&home); assert_eq!(result.status, ProviderSyncStatus::Synced); assert_eq!(result.sqlite_cwd_rows_updated, 0); @@ -562,7 +697,7 @@ fn provider_sync_normalizes_open_in_target_preferences_per_path() { ) .unwrap(); - let result = run_provider_sync(Some(&home)); + let result = run_provider_rewrite_sync(&home); assert_eq!(result.status, ProviderSyncStatus::Synced); let state: serde_json::Value = @@ -607,7 +742,7 @@ fn provider_sync_restores_rollout_first_line_when_later_step_fails() { .unwrap(); drop(db); - let result = run_provider_sync(Some(&home)); + let result = run_provider_rewrite_sync(&home); assert_eq!(result.status, ProviderSyncStatus::Skipped); assert!(result.message.contains("Provider sync skipped")); @@ -650,7 +785,7 @@ fn provider_sync_rolls_back_sqlite_provider_update_when_later_update_fails() { .unwrap(); drop(db); - let result = run_provider_sync(Some(&home)); + let result = run_provider_rewrite_sync(&home); assert_eq!(result.status, ProviderSyncStatus::Skipped); let db = Connection::open(home.join("state_5.sqlite")).unwrap(); @@ -714,7 +849,7 @@ fn provider_sync_skips_when_home_missing_or_lock_exists_and_prunes_backups() { fs::create_dir(&home).unwrap(); fs::create_dir_all(home.join("tmp/provider-sync.lock")).unwrap(); fs::write(home.join("config.toml"), "model_provider = \"apigather\"\n").unwrap(); - let result = run_provider_sync(Some(&home)); + let result = run_provider_rewrite_sync(&home); assert_eq!(result.status, ProviderSyncStatus::Skipped); assert!(result.message.to_lowercase().contains("lock")); @@ -735,7 +870,7 @@ fn provider_sync_skips_when_home_missing_or_lock_exists_and_prunes_backups() { "thread-1", "C:/workspace", ); - let result = run_provider_sync(Some(&home)); + let result = run_provider_rewrite_sync(&home); assert_eq!(result.status, ProviderSyncStatus::Synced); let backups = fs::read_dir(&backup_root) .unwrap() @@ -761,7 +896,7 @@ fn provider_sync_preserves_rollout_mtime() { let mtime_before = fs::metadata(&rollout).unwrap().modified().unwrap(); - let result = run_provider_sync(Some(&home)); + let result = run_provider_rewrite_sync(&home); assert_eq!(result.status, ProviderSyncStatus::Synced); assert_eq!(result.changed_session_files, 1);