diff --git a/README.md b/README.md
index d599643393..2dadeb09fa 100644
--- a/README.md
+++ b/README.md
@@ -198,6 +198,8 @@ Optional: add an ordered **Fallback Models** list under **Model Config**. It
 applies to every connected client. A failed request may reach and consume usage
 from more than one provider before succeeding.
 
+For another API endpoint, open **Providers → Custom providers → Add provider**. Enter its API base URL, optional key, and API format. Leave **Model IDs** empty to discover models automatically, or enter one upstream model ID per line. Save, then select a model in **Model Config**.
+
 <details>
 <summary><strong>Provider catalog</strong></summary>
 
diff --git a/e2e/conftest.py b/e2e/conftest.py
index 5d7bdff642..2368f22f06 100644
--- a/e2e/conftest.py
+++ b/e2e/conftest.py
@@ -117,6 +117,7 @@ async def stream_responses(
         response_model: str | None = None,
         reasoning: ReasoningPolicy = DEFAULT_REASONING_POLICY,
         request_headers: Mapping[str, str] | None = None,
+        model_info: ProviderModelInfo | None = None,
     ) -> AsyncIterator[str]:
         if False:
             yield ""
diff --git a/e2e/test_custom_providers.py b/e2e/test_custom_providers.py
new file mode 100644
index 0000000000..3ef6fc3cd2
--- /dev/null
+++ b/e2e/test_custom_providers.py
@@ -0,0 +1,122 @@
+"""Custom endpoint configuration through the real Admin Apply flow."""
+
+import pytest
+from playwright.sync_api import Page, expect
+
+
+@pytest.fixture(autouse=True)
+def no_page_errors(page: Page):
+    errors = []
+    page.on("pageerror", lambda error: errors.append(str(error)))
+    yield
+    assert errors == []
+
+
+def add_provider(page: Page, name: str) -> str:
+    page.get_by_role("button", name="Add provider", exact=True).click()
+    dialog = page.locator("#providerDialog")
+    dialog.get_by_label("Name", exact=True).fill(name)
+    dialog.get_by_label("Base URL", exact=True).fill("https://gateway.example/team/v1")
+    dialog.get_by_label("API key", exact=True).fill("example-custom-key")
+    dialog.get_by_label("Model IDs", exact=True).fill("org/model-one\nmodel-two")
+    page.locator("#saveProvider").click()
+    expect(dialog).not_to_be_visible()
+    card = page.locator("#providers-custom .provider-card").filter(
+        has=page.get_by_text(name, exact=True)
+    )
+    expect(card).to_be_visible()
+    expect(page.locator("#addCustomProvider")).to_be_focused()
+    provider_id = card.get_attribute("data-provider")
+    assert provider_id is not None
+    return provider_id
+
+
+def test_custom_provider_save_rename_and_remove_preserve_unrelated_draft(
+    page: Page, admin_base_url: str
+):
+    page.goto(admin_base_url + "/admin")
+    expect(page.locator("#messageArea")).to_have_text("")
+    page.locator("#field-HTTP_READ_TIMEOUT").fill("181")
+    provider_id = add_provider(page, "Work gateway")
+    expect(page.locator("#field-HTTP_READ_TIMEOUT")).to_have_value("181")
+    card = page.locator(f'[data-provider="{provider_id}"]')
+    expect(card.locator(".provider-check-result")).to_have_text("2 models available")
+    card.get_by_role("button", name="Edit", exact=True).click()
+    dialog = page.locator("#providerDialog")
+    expect(dialog.get_by_label("API key", exact=True)).to_have_value("")
+    dialog.get_by_label("Name", exact=True).fill("Renamed gateway")
+    page.locator("#saveProvider").click()
+    expect(dialog).not_to_be_visible()
+    expect(card.locator(".provider-title")).to_have_text("Renamed gateway")
+    expect(card.get_by_role("button", name="Edit", exact=True)).to_be_focused()
+    card.get_by_role("button", name="Edit", exact=True).click()
+    dialog.get_by_role("button", name="Remove provider", exact=True).click()
+    expect(dialog).not_to_be_visible()
+    expect(card).to_have_count(0)
+    expect(page.locator("#addCustomProvider")).to_be_focused()
+    expect(page.locator("#field-HTTP_READ_TIMEOUT")).to_have_value("181")
+
+
+def test_custom_provider_sorting_selection_and_blocked_removal(
+    page: Page, admin_base_url: str
+):
+    page.goto(admin_base_url + "/admin")
+    add_provider(page, "Zulu gateway")
+    selected = add_provider(page, "Alpha gateway")
+    expect(page.locator("#providers-custom .provider-title")).to_have_text(
+        ["Alpha gateway", "Zulu gateway"]
+    )
+    page.get_by_role("button", name="Model Config", exact=True).click()
+    model = page.locator("#field-MODEL_HAIKU")
+    model.fill("Alpha gateway")
+    page.get_by_role("option", name="Alpha gateway/org/model-one", exact=True).click()
+    expect(model).to_have_value(f"{selected}/org/model-one")
+    page.locator("#applyButton").click()
+    expect(page.locator("#dirtyState")).to_have_text("No changes")
+    page.get_by_role("button", name="Providers", exact=True).click()
+    page.locator(f'[data-provider="{selected}"]').get_by_role(
+        "button", name="Edit", exact=True
+    ).click()
+    page.get_by_role("button", name="Remove provider", exact=True).click()
+    expect(page.locator("#providerMessage")).to_contain_text("MODEL")
+    expect(page.locator("#providerDialog")).to_be_visible()
+
+
+def test_custom_dialog_cancel_and_format_changes(page: Page, admin_base_url: str):
+    page.goto(admin_base_url + "/admin")
+    page.get_by_role("button", name="Add provider", exact=True).click()
+    dialog = page.locator("#providerDialog")
+    dialog.get_by_label("Reasoning format", exact=True).select_option("thinking")
+    dialog.get_by_label("Reasoning history format", exact=True).select_option(
+        "reasoning_content"
+    )
+    dialog.get_by_label("API format", exact=True).select_option("anthropic_messages")
+    expect(dialog.get_by_label("Reasoning format", exact=True)).to_have_value(
+        "provider_default"
+    )
+    expect(
+        dialog.get_by_label("Reasoning history format", exact=True)
+    ).not_to_be_visible()
+    page.keyboard.press("Escape")
+    expect(dialog).not_to_be_visible()
+    expect(page.locator("#providers-custom .provider-card")).to_have_count(0)
+    expect(page.locator("#addCustomProvider")).to_be_focused()
+
+
+def test_duplicate_custom_provider_keeps_failed_save_draft(
+    page: Page, admin_base_url: str
+):
+    page.goto(admin_base_url + "/admin")
+    add_provider(page, "Gateway")
+    page.get_by_role("button", name="Add provider", exact=True).click()
+    dialog = page.locator("#providerDialog")
+    dialog.get_by_label("Name", exact=True).fill("Gateway")
+    dialog.get_by_label("Base URL", exact=True).fill("https://another.example/v1")
+    page.locator("#saveProvider").click()
+    expect(page.locator("#providerMessage")).to_contain_text("already exists")
+    expect(dialog).to_be_visible()
+    expect(dialog.get_by_label("Base URL", exact=True)).to_have_value(
+        "https://another.example/v1"
+    )
+    expect(page.locator("#saveProvider")).to_be_enabled()
+    expect(page.locator("#providers-custom .provider-card")).to_have_count(1)
diff --git a/e2e/test_provider_modals.py b/e2e/test_provider_modals.py
index 7a66a8dd32..74e2d8ce3b 100644
--- a/e2e/test_provider_modals.py
+++ b/e2e/test_provider_modals.py
@@ -82,6 +82,7 @@ def test_provider_groups_sort_each_subgroup_and_keep_setup_separate_from_health(
         "OAuth providers",
         "Cloud providers",
         "Local providers",
+        "Custom providers",
     ]
     for group in ("cloud", "local"):
         for subgroup in ("configured", "unconfigured"):
diff --git a/src/free_claude_code/api/admin_routes.py b/src/free_claude_code/api/admin_routes.py
index e2f8f34beb..698a4f0c39 100644
--- a/src/free_claude_code/api/admin_routes.py
+++ b/src/free_claude_code/api/admin_routes.py
@@ -22,6 +22,7 @@
 from free_claude_code.application.errors import ApplicationError
 from free_claude_code.application.model_catalog import read_model_catalog
 from free_claude_code.application.model_metadata import ProviderModelRefreshResult
+from free_claude_code.config.admin.custom_providers import CustomProviderMutation
 from free_claude_code.config.admin.manifest import FIELD_BY_KEY
 from free_claude_code.config.provider_catalog import (
     PROVIDER_CATALOG,
@@ -76,6 +77,7 @@ class AdminConfigPayload(BaseModel):
     """Partial config update submitted by the admin UI."""
 
     values: JsonObject = Field(default_factory=dict)
+    custom_provider: CustomProviderMutation | None = None
 
 
 class ConnectedAccountLoginPayload(BaseModel):
@@ -136,6 +138,10 @@ async def apply_admin_config(
     services: ApiServices = Depends(get_services),
 ):
     require_loopback_admin(request)
+    if payload.custom_provider is not None:
+        return await services.admin.apply_admin_config(
+            _filtered_values(payload.values), payload.custom_provider
+        )
     result = await services.admin.apply_admin_config(_filtered_values(payload.values))
     return result
 
@@ -421,13 +427,16 @@ def _model_options(
     services: ApiServices,
     *,
     refresh_result: ProviderModelRefreshResult | None = None,
-) -> dict[str, list[str]]:
+) -> JsonObject:
     catalog = read_model_catalog(services.requests)
     failed_provider_ids = (
         refresh_result.failed_provider_ids if refresh_result is not None else ()
     )
     return {
         "models": [model.provider_model_ref for model in catalog.models],
+        "model_labels": {
+            model.provider_model_ref: model.display_name for model in catalog.models
+        },
         "failed_providers": list(failed_provider_ids),
     }
 
diff --git a/src/free_claude_code/api/admin_static/admin.js b/src/free_claude_code/api/admin_static/admin.js
index 71f33d238d..4eeab6461d 100644
--- a/src/free_claude_code/api/admin_static/admin.js
+++ b/src/free_claude_code/api/admin_static/admin.js
@@ -11,6 +11,8 @@ const state = {
   authStatuses: new Map(),
   providerChecks: new Map(),
   providerId: null,
+  customProvider: null,
+  modelLabels: new Map(),
   localStatusRequest: null,
   startup: null,
   startupRequest: null,
@@ -209,7 +211,10 @@ async function load({ providersOnly = false } = {}) {
   state.startupRequest = null;
   showMessage("Loading admin config");
   const config = await api("/admin/api/config");
+  config.custom_providers ||= [];
+  config.provider_status.push(...config.custom_providers.map((provider) => ({ ...provider, kind: "custom", status: "configured", settings_keys: [], missing_configuration_keys: [] })));
   state.config = config;
+  window.CodeSessions.setProviderNames(config.custom_providers);
   state.startup = null;
   state.fields = new Map(config.fields.map((field) => [field.key, field]));
   if (!providersOnly) renderNav();
@@ -334,6 +339,25 @@ function renderProviders(providerStatus) {
     });
     container.appendChild(group);
   });
+  const custom = document.createElement("section");
+  custom.className = "provider-strip";
+  custom.dataset.providerGroup = "custom";
+  const heading = document.createElement("h3");
+  heading.textContent = "Custom providers";
+  const add = authButton("Add provider", () => openCustomProviderDialog(), "primary-button");
+  add.id = "addCustomProvider";
+  add.disabled = !!state.config.custom_providers_locked;
+  const empty = document.createElement("p");
+  empty.textContent = "No custom providers yet.";
+  empty.hidden = state.config.custom_providers.length > 0;
+  const grid = document.createElement("div");
+  grid.className = "provider-grid";
+  grid.id = "providers-custom";
+  const header = document.createElement("div");
+  header.className = "strip-header";
+  header.append(heading, add);
+  custom.append(header, empty, grid);
+  container.appendChild(custom);
   providerStatus.forEach(updateProviderCard);
 }
 
@@ -343,7 +367,7 @@ function updateProviderCard(provider) {
   const kind = oauth ? "oauth" : provider.kind === "local" ? "local" : "cloud";
   const configured = oauth ? status?.connected : provider.status === "configured";
   const subgroup = oauth && configured == null ? "loading" : configured ? "configured" : "unconfigured";
-  const grid = byId(`providers-${kind}-${subgroup}`);
+  const grid = byId(provider.kind === "custom" ? "providers-custom" : `providers-${kind}-${subgroup}`);
   let card = document.querySelector(`[data-provider="${provider.provider_id}"]`);
   if (!card) {
     card = document.createElement("article");
@@ -362,12 +386,12 @@ function updateProviderCard(provider) {
   website.rel = "noopener noreferrer";
   const logo = document.createElement("img");
   logo.className = "provider-logo";
-  logo.src = new URL(`providers/${provider.logo_filename}`, adminAssetBase);
+  if (provider.logo_filename) logo.src = new URL(`providers/${provider.logo_filename}`, adminAssetBase);
   logo.alt = "";
   logo.width = 20;
   logo.height = 20;
   website.append(name, logo);
-  title.appendChild(website);
+  title.appendChild(provider.kind === "custom" ? name : website);
   const meta = document.createElement("span");
   meta.className = "provider-meta";
   meta.hidden = !oauth;
@@ -378,7 +402,7 @@ function updateProviderCard(provider) {
   const actions = document.createElement("div");
   actions.className = "provider-actions";
   if (oauth) populateConnectedAccountActions(provider, status, actions);
-  if (provider.settings_keys?.length) {
+  if (provider.kind === "custom" || provider.settings_keys?.length) {
     const edit = oauth || configured;
     const settings = authButton(edit ? "Edit" : "Configure", () => openProviderDialog(provider.provider_id), edit ? "secondary-button" : "primary-button");
     settings.dataset.providerSettings = "true";
@@ -401,6 +425,8 @@ function updateProviderCard(provider) {
 }
 
 function openProviderDialog(providerId) {
+  if (connectedAccountDescriptor(providerId)?.kind === "custom") return openCustomProviderDialog(providerId);
+  state.customProvider = null;
   state.providerId = providerId;
   const provider = connectedAccountDescriptor(providerId);
   byId("providerDialogTitle").textContent = connectedAccountName(provider);
@@ -429,7 +455,70 @@ function openProviderDialog(providerId) {
   first?.focus();
 }
 
+const CUSTOM_LABELS = {
+  provider_default: "Provider default", openai_effort: "OpenAI effort", limited_effort: "Limited effort (low / medium / high)",
+  reasoning_object: "Reasoning object", thinking: "Thinking on/off", chat_template: "Chat-template thinking",
+  native_responses: "Native reasoning", messages_manual: "Token-budget thinking", messages_adaptive: "Adaptive thinking",
+};
+
+function openCustomProviderDialog(providerId = null) {
+  const provider = state.config.custom_providers.find((item) => item.provider_id === providerId) || {
+    display_name: "", base_url: "", api_key: null, api_format: "openai_chat", reasoning_format: "provider_default", reasoning_history_format: "disabled", model_ids: [],
+  };
+  state.providerId = providerId || "__custom_new__";
+  state.customProvider = provider;
+  byId("providerDialogTitle").textContent = providerId ? provider.display_name : "Add provider";
+  byId("providerMessage").textContent = "";
+  byId("providerDialogCheck").hidden = true;
+  const fields = byId("providerFields");
+  fields.replaceChildren();
+  const formats = (state.config.custom_reasoning_formats || {})[provider.api_format] || ["provider_default"];
+  const definitions = [
+    ["display_name", "Name", "string", [], ""],
+    ["base_url", "Base URL", "string", [], "Include the API path, for example https://gateway.example/v1."],
+    ["api_key", "API key", "secret", [], "Optional for endpoints that do not require a key."],
+    ["api_format", "API format", "select", [["openai_chat", "Chat Completions"], ["openai_responses", "Responses"], ["anthropic_messages", "Anthropic Messages"]], "Select the API exposed by your endpoint."],
+    ["reasoning_format", "Reasoning format", "select", formats.map((value) => [value, CUSTOM_LABELS[value]]), "How FCC sends reasoning controls. Provider default leaves computation to the upstream."],
+    ["reasoning_history_format", "Reasoning history format", "select", [["disabled", "Text context"], ["reasoning_content", "reasoning_content"], ["reasoning", "reasoning"], ["think_tags", "Think tags"]], "How previous reasoning is sent to a Chat Completions endpoint."],
+    ["model_ids", "Model IDs", "textarea", [], "Optional, one upstream model ID per line. Leave empty to discover models automatically."],
+  ];
+  for (const [key, label, type, options, description] of definitions) {
+    fields.appendChild(renderField({ key, label, type, options: options.map(([value, label]) => ({value, label})), description,
+      value: key === "model_ids" ? provider.model_ids.join("\n") : provider[key], nullable: key === "api_key", secret: key === "api_key",
+      configured: key === "api_key" && !!provider.api_key, locked: !!state.config.custom_providers_locked, source: "managed_env" }));
+  }
+  const format = byId("field-api_format");
+  const reasoning = byId("field-reasoning_format");
+  const history = byId("field-reasoning_history_format");
+  history.closest(".field").hidden = format.value !== "openai_chat";
+  format.addEventListener("change", () => {
+    const allowed = state.config.custom_reasoning_formats[format.value];
+    const previous = reasoning.value;
+    reasoning.replaceChildren(...allowed.map((value) => option(value, CUSTOM_LABELS[value])));
+    reasoning.value = allowed.includes(previous) ? previous : "provider_default";
+    history.closest(".field").hidden = format.value !== "openai_chat";
+    if (format.value !== "openai_chat") history.value = "disabled";
+    updateDirtyState();
+  });
+  const actions = byId("providerDialogActions");
+  actions.replaceChildren();
+  if (providerId && !state.config.custom_providers_locked) {
+    if (!provider.model_ids.length) actions.appendChild(authButton("Refresh models", (button) => testProvider(providerId, button), "secondary-button"));
+    actions.appendChild(authButton("Remove provider", () => apply(providerId, "delete"), "danger-button"));
+    const modelConfig = document.createElement("a");
+    modelConfig.href = "/admin/model_config";
+    modelConfig.textContent = "Model Config";
+    modelConfig.addEventListener("click", (event) => { event.preventDefault(); byId("providerDialog").close(); navigateToView("model_config"); });
+    actions.appendChild(modelConfig);
+  }
+  byId("saveProvider").hidden = false;
+  updateDirtyState();
+  byId("providerDialog").showModal();
+  byId("field-display_name").focus();
+}
+
 function renderProviderDialogActions(provider) {
+  if (state.customProvider) return;
   if (state.providerId !== provider.provider_id) return;
   const actions = byId("providerDialogActions");
   actions.replaceChildren();
@@ -855,6 +944,7 @@ function renderField(field) {
 }
 
 function inputForField(field) {
+  if (field.type === "textarea") { const input = document.createElement("textarea"); input.rows = 3; input.value = field.value || ""; return input; }
   if (field.type === "boolean") {
     const input = document.createElement("input");
     input.type = "checkbox";
@@ -914,6 +1004,7 @@ function createModelCombobox(input, field) {
   return new window.FccModelCombobox(input, {
     listboxId: `model-options-${field.key}`,
     label: field.label,
+    displayValue: (value) => state.modelLabels.get(value) || value,
     values: () =>
       field.type === "optional_model"
         ? ["None", ...state.modelOptions]
@@ -1219,14 +1310,22 @@ function showRestartNotice() {
   }
 }
 
-async function apply(providerId = null) {
+async function apply(providerId = null, customAction = null) {
   if (state.applying) return;
   if (state.restart) {
     await reconnectAfterRestart();
     return;
   }
   const values = changedValues(providerId ? byId("providerFields") : byId("adminViews"));
-  if (!Object.keys(values).length) return;
+  const custom = providerId && state.customProvider;
+  let mutation = null;
+  if (custom) {
+    const action = customAction || (custom.provider_id ? "update" : "create");
+    if ("model_ids" in values) values.model_ids = values.model_ids.split(/\r?\n/).map((id) => id.trim()).filter(Boolean);
+    mutation = { action, values: action === "delete" ? {} : values };
+    if (custom.provider_id) mutation.provider_id = custom.provider_id;
+  }
+  if (!Object.keys(values).length && mutation?.action !== "delete") return;
   const checkingKeys = Object.keys(values).some((key) => {
     const field = state.fields.get(key);
     return field?.secret && field.section === "providers" && values[key] !== null;
@@ -1238,7 +1337,7 @@ async function apply(providerId = null) {
   try {
     const result = await api("/admin/api/config/apply", {
       method: "POST",
-      body: JSON.stringify({ values }),
+      body: JSON.stringify(mutation ? { custom_provider: mutation } : { values }),
     });
     const checks = result.credential_checks || [];
     if (!result.applied) {
@@ -1273,7 +1372,10 @@ async function apply(providerId = null) {
       rejectedField.scrollIntoView({ block: "center", behavior: "instant" });
       rejectedField.focus();
     } else if (providerId && applied) {
-      document.querySelector(`[data-provider="${providerId}"] [data-provider-settings]`)?.focus({ preventScroll: true });
+      const focus = mutation && mutation.action !== "update"
+        ? byId("addCustomProvider")
+        : document.querySelector(`[data-provider="${providerId}"] [data-provider-settings]`);
+      focus?.focus({ preventScroll: true });
     }
   }
 }
@@ -1385,7 +1487,7 @@ async function loadModelOptions(refresh = false) {
   const result = await api("/admin/api/models" + (refresh ? "/refresh" : ""), {
     method: refresh ? "POST" : "GET",
   });
-  if (request === state.modelOptionsRequest && config === state.config) setModelOptions(result.models);
+  if (request === state.modelOptionsRequest && config === state.config) { state.modelLabels = new Map(Object.entries(result.model_labels || {})); setModelOptions(result.models); }
   if (refresh && window.CodeSessions) await window.CodeSessions.refresh();
   return result;
 }
@@ -1459,6 +1561,8 @@ byId("providerDialog").addEventListener("click", (event) => {
 byId("providerDialog").addEventListener("close", () => {
   if (byId("providerDialog").open) return;
   const providerId = state.providerId;
+  const wasCustom = !!state.customProvider;
+  state.customProvider = null;
   state.providerId = null;
   byId("providerFields").replaceChildren();
   updateDirtyState();
diff --git a/src/free_claude_code/api/admin_static/code_sessions.js b/src/free_claude_code/api/admin_static/code_sessions.js
index 8d4a900a0f..a4c8fe4d47 100644
--- a/src/free_claude_code/api/admin_static/code_sessions.js
+++ b/src/free_claude_code/api/admin_static/code_sessions.js
@@ -2,6 +2,7 @@
   "use strict";
   const UI = window.SessionUI;
   const modelComboboxes = new Set();
+  let providerNames = new Map();
   let modelControl,
     reasoningControl,
     providerControl,
@@ -1282,7 +1283,7 @@
     const providers = [...new Set(catalog.map((model) => model.provider_id))];
     if (provider && !providers.includes(provider)) providers.push(provider);
     providerControl.update(
-      providers.map((value) => [value, value]),
+      providers.map((value) => [value, providerNames.get(value) || value]),
       provider,
     );
     providerControl.select.disabled = disabled;
@@ -1661,6 +1662,9 @@
   }
   window.addEventListener("pagehide", deactivate);
   window.CodeSessions = {
+    setProviderNames(providers) {
+      providerNames = new Map(providers.map((provider) => [provider.provider_id, provider.display_name]));
+    },
     initialize(client) {
       api = client;
       root = document.getElementById("codeRoot");
diff --git a/src/free_claude_code/api/admin_static/model_combobox.js b/src/free_claude_code/api/admin_static/model_combobox.js
index ad65cd2d64..57be41c4eb 100644
--- a/src/free_claude_code/api/admin_static/model_combobox.js
+++ b/src/free_claude_code/api/admin_static/model_combobox.js
@@ -8,11 +8,13 @@
         values,
         emptyMessage,
         registry,
+        displayValue = (value) => value,
         onSelect = null,
         onClose = null,
       },
     ) {
       this.input = input;
+      this.displayValue = displayValue;
       this.getValues = values;
       this.getEmptyMessage = emptyMessage;
       this.registry = registry;
@@ -106,7 +108,7 @@
       const allValues = this.getValues();
       const values = normalizedQuery
         ? allValues.filter((value) =>
-            value.toLocaleLowerCase().includes(normalizedQuery),
+            (value + " " + this.displayValue(value)).toLocaleLowerCase().includes(normalizedQuery),
           )
         : allValues;
       this.listbox.replaceChildren();
@@ -127,7 +129,7 @@
         optionEl.id = `${this.listbox.id}-option-${index}`;
         optionEl.dataset.value = value;
         optionEl.setAttribute("role", "option");
-        optionEl.textContent = value;
+        optionEl.textContent = this.displayValue(value);
         this.listbox.appendChild(optionEl);
       });
       const selectedIndex = values.indexOf(this.input.value);
diff --git a/src/free_claude_code/api/dependencies.py b/src/free_claude_code/api/dependencies.py
index e900fdbb8d..b848470a2c 100644
--- a/src/free_claude_code/api/dependencies.py
+++ b/src/free_claude_code/api/dependencies.py
@@ -7,7 +7,6 @@
 
 from free_claude_code.application.errors import UnknownProviderError
 from free_claude_code.application.ports import ProviderPort, RequestRuntimeLease
-from free_claude_code.config.provider_catalog import PROVIDER_CATALOG
 from free_claude_code.config.settings import Settings
 
 from .ports import ApiServices
@@ -36,7 +35,7 @@ async def resolve_provider(
         logger.error(
             "Unknown provider_type: '{}'. Supported: {}",
             provider_type,
-            ", ".join(f"'{key}'" for key in PROVIDER_CATALOG),
+            ", ".join(f"'{key}'" for key in lease.settings.provider_ids),
         )
         raise
     if should_log_init:
diff --git a/src/free_claude_code/api/handlers/responses.py b/src/free_claude_code/api/handlers/responses.py
index ff0df5092c..abd90f2e6f 100644
--- a/src/free_claude_code/api/handlers/responses.py
+++ b/src/free_claude_code/api/handlers/responses.py
@@ -17,7 +17,7 @@
 )
 from free_claude_code.application.errors import ApplicationError, InvalidRequestError
 from free_claude_code.application.execution import ProviderExecutor
-from free_claude_code.application.ports import ProviderResolver
+from free_claude_code.application.ports import ModelInfoLookup, ProviderResolver
 from free_claude_code.application.routing import ModelRouter
 from free_claude_code.config.settings import Settings
 from free_claude_code.core.diagnostics import safe_exception_message
@@ -43,6 +43,7 @@ def __init__(
         provider_executor: ProviderExecutor | None = None,
         generation_id: int | None = None,
         request_headers: Mapping[str, str] | None = None,
+        model_info_lookup: ModelInfoLookup | None = None,
     ) -> None:
         self._settings = settings
         self._model_router = model_router or ModelRouter(settings)
@@ -52,6 +53,7 @@ def __init__(
             generation_id=generation_id,
             log_raw_payloads=settings.log_raw_api_payloads,
             request_headers=request_headers,
+            model_info_lookup=model_info_lookup,
         )
 
     async def create(
diff --git a/src/free_claude_code/api/ports.py b/src/free_claude_code/api/ports.py
index 983c30c5a1..8e4c4cdbd6 100644
--- a/src/free_claude_code/api/ports.py
+++ b/src/free_claude_code/api/ports.py
@@ -12,6 +12,7 @@
 from free_claude_code.application.model_metadata import ProviderModelRefreshResult
 from free_claude_code.application.ports import RequestRuntimePort, TaskController
 from free_claude_code.application.web_tools.ports import WebToolsPort
+from free_claude_code.config.admin.custom_providers import CustomProviderMutation
 from free_claude_code.config.admin.state import ConfigInputValue, ValueState
 from free_claude_code.core.json_types import JsonObject
 
@@ -20,7 +21,9 @@ class AdminRuntimePort(Protocol):
     """Runtime operations exposed by the local Admin API."""
 
     async def apply_admin_config(
-        self, updates: Mapping[str, ConfigInputValue]
+        self,
+        updates: Mapping[str, ConfigInputValue],
+        custom_provider: CustomProviderMutation | None = None,
     ) -> JsonObject: ...
 
     async def admin_config(self) -> JsonObject: ...
diff --git a/src/free_claude_code/api/routes.py b/src/free_claude_code/api/routes.py
index 2e1d2c2ded..4b1cd65570 100644
--- a/src/free_claude_code/api/routes.py
+++ b/src/free_claude_code/api/routes.py
@@ -96,6 +96,7 @@ async def _create_responses_response(
             provider_resolver=_provider_resolver(lease),
             generation_id=lease.generation_id,
             request_headers=request_headers,
+            model_info_lookup=lease.model_info,
         )
         response = await handler.create(request_data, request_id=request_id)
     except ApplicationError as exc:
diff --git a/src/free_claude_code/application/execution.py b/src/free_claude_code/application/execution.py
index 91f3f1f632..0181bddb73 100644
--- a/src/free_claude_code/application/execution.py
+++ b/src/free_claude_code/application/execution.py
@@ -248,6 +248,9 @@ async def open_candidate(
                 response_model=routed.resolved.original_model,
                 reasoning=routed.reasoning,
                 request_headers=self._request_headers,
+                model_info=self._model_info_lookup(
+                    target.provider_id, target.provider_model
+                ),
             )
 
         raw_input = routed.request.input
diff --git a/src/free_claude_code/application/model_catalog.py b/src/free_claude_code/application/model_catalog.py
index d1646d3f53..8eef9982a3 100644
--- a/src/free_claude_code/application/model_catalog.py
+++ b/src/free_claude_code/application/model_catalog.py
@@ -1,7 +1,7 @@
 """Application model inventory and presentation order, independent of clients."""
 
 from collections.abc import Sequence
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from typing import TYPE_CHECKING
 
 from free_claude_code.config.model_refs import (
@@ -58,11 +58,32 @@ def read_model_catalog(
     for info in runtime.cached_prefixed_model_infos():
         if info.model_id not in models:
             models[info.model_id] = _catalog_model(info.model_id, info)
+    for ref, model in tuple(models.items()):
+        provider_id, model_id = split_provider_model_ref(ref)
+        if definition := settings.custom_provider(provider_id):
+            models[ref] = replace(
+                model,
+                display_name=f"{definition.display_name}/{model_id}"
+                + (" (no thinking)" if model.supports_reasoning is False else ""),
+            )
+
+    def order(model: CatalogModel):
+        provider_id, model_id = split_provider_model_ref(model.provider_model_ref)
+        definition = settings.custom_provider(provider_id)
+        name = definition.display_name if definition else provider_id
+        return (
+            name.casefold(),
+            model_id.casefold(),
+            name,
+            model_id,
+            model.provider_model_ref,
+        )
+
     return ModelCatalog(
         models=tuple(
             sorted(
                 models.values(),
-                key=lambda model: model_order_key(model.provider_model_ref),
+                key=order,
             )
         ),
         default_model_id=models[settings.model].wire_slug,
diff --git a/src/free_claude_code/application/ports.py b/src/free_claude_code/application/ports.py
index d7d5b67c28..fbcf2a56a8 100644
--- a/src/free_claude_code/application/ports.py
+++ b/src/free_claude_code/application/ports.py
@@ -37,6 +37,7 @@ def stream_responses(
         response_model: str,
         reasoning: ReasoningPolicy,
         request_headers: Mapping[str, str] | None = None,
+        model_info: ProviderModelInfo | None = None,
     ) -> AsyncIterator[str]: ...
 
 
diff --git a/src/free_claude_code/application/routing.py b/src/free_claude_code/application/routing.py
index cc2afd8b3b..00336cda07 100644
--- a/src/free_claude_code/application/routing.py
+++ b/src/free_claude_code/application/routing.py
@@ -13,10 +13,6 @@
     parse_model_name,
     parse_provider_type,
 )
-from free_claude_code.config.provider_catalog import (
-    PROVIDER_CATALOG,
-    SUPPORTED_PROVIDER_IDS,
-)
 from free_claude_code.config.reasoning import ReasoningPreference
 from free_claude_code.config.settings import Settings
 from free_claude_code.core.anthropic import MessagesRequest, TokenCountRequest
@@ -149,10 +145,11 @@ def _fallback_targets(
         )
         return tuple(target for target in configured if target != primary)
 
-    @staticmethod
-    def _validate_provider_id(provider_id: str) -> None:
-        if provider_id not in PROVIDER_CATALOG:
-            raise UnknownProviderError.for_provider(provider_id, PROVIDER_CATALOG)
+    def _validate_provider_id(self, provider_id: str) -> None:
+        if provider_id not in self._settings.provider_ids:
+            raise UnknownProviderError.for_provider(
+                provider_id, self._settings.provider_ids
+            )
 
     def _direct_provider_model(
         self, model_name: str
@@ -173,13 +170,15 @@ def _direct_provider_model(
                 decoded.force_reasoning_off if decoded is not None else False,
             )
         if decoded is not None:
-            if decoded.provider_id not in SUPPORTED_PROVIDER_IDS:
-                if model_name.partition("/")[0] in {
+            if decoded.provider_id not in self._settings.provider_ids:
+                if decoded.provider_id.startswith("custom_") or model_name.partition(
+                    "/"
+                )[0] in {
                     DESKTOP_MODEL_PREFIX,
                     DESKTOP_NO_THINKING_PREFIX,
                 }:
                     raise UnknownProviderError.for_provider(
-                        decoded.provider_id, PROVIDER_CATALOG
+                        decoded.provider_id, self._settings.provider_ids
                     )
                 return None, None, False
             return (
@@ -191,7 +190,11 @@ def _direct_provider_model(
         provider_id, separator, provider_model = model_name.partition("/")
         if not separator:
             return None, None, False
-        if provider_id not in SUPPORTED_PROVIDER_IDS:
+        if provider_id not in self._settings.provider_ids:
+            if provider_id.startswith("custom_"):
+                raise UnknownProviderError.for_provider(
+                    provider_id, self._settings.provider_ids
+                )
             return None, None, False
         if not provider_model:
             return None, None, False
diff --git a/src/free_claude_code/config/admin/custom_providers.py b/src/free_claude_code/config/admin/custom_providers.py
new file mode 100644
index 0000000000..b2f319903d
--- /dev/null
+++ b/src/free_claude_code/config/admin/custom_providers.py
@@ -0,0 +1,101 @@
+"""Keyed custom-provider edits merged into a fresh configuration snapshot."""
+
+from typing import Literal
+from uuid import uuid4
+
+from pydantic import BaseModel, ConfigDict, Field, ValidationError
+
+from free_claude_code.config.custom_providers import (
+    CUSTOM_PROVIDERS_ENV,
+    CustomProviderDefinition,
+    encode_custom_providers,
+)
+from free_claude_code.config.loader import ConfigSource, ManagedConfigSnapshot
+from free_claude_code.core.json_types import JsonObject
+
+
+class CustomProviderMutation(BaseModel):
+    model_config = ConfigDict(extra="forbid", hide_input_in_errors=True)
+    action: Literal["create", "update", "delete"]
+    provider_id: str | None = None
+    values: JsonObject = Field(default_factory=dict, repr=False)
+
+
+def merge_custom_provider(
+    target: dict[str, str],
+    snapshot: ManagedConfigSnapshot,
+    mutation: CustomProviderMutation,
+) -> str:
+    if snapshot.sources["custom_providers"] == ConfigSource.PROCESS:
+        raise ValueError("Custom providers are managed by the process environment")
+    definitions = {
+        item.provider_id: item for item in snapshot.settings.custom_providers
+    }
+    provider_id = mutation.provider_id
+    if mutation.action == "create":
+        if provider_id is not None:
+            raise ValueError("Provider IDs are assigned by FCC")
+        provider_id = "custom_" + uuid4().hex
+        values = {}
+    else:
+        if provider_id is None:
+            raise ValueError("A provider ID is required")
+        if mutation.action == "delete":
+            if mutation.values:
+                raise ValueError("Removal does not accept field edits")
+            references = [
+                key
+                for key in (
+                    "MODEL",
+                    "MODEL_FABLE",
+                    "MODEL_OPUS",
+                    "MODEL_SONNET",
+                    "MODEL_HAIKU",
+                    "MODEL_FALLBACKS",
+                )
+                if any(
+                    ref.strip().startswith(provider_id + "/")
+                    for ref in snapshot.process.get(key, target.get(key, "")).split(",")
+                )
+            ]
+            if references:
+                raise ValueError(
+                    "Change these Model Config settings before removing the provider: "
+                    + ", ".join(references)
+                )
+            definitions.pop(provider_id, None)
+            target[CUSTOM_PROVIDERS_ENV] = encode_custom_providers(
+                tuple(definitions.values())
+            )
+            return provider_id
+        if provider_id not in definitions:
+            raise ValueError("This custom provider no longer exists")
+        values = definitions[provider_id].private_values()
+    if "provider_id" in mutation.values:
+        raise ValueError("Provider IDs cannot be edited")
+    changes = dict(mutation.values)
+    if changes.get("api_key") == "********" or (
+        isinstance(changes.get("api_key"), str) and not str(changes["api_key"]).strip()
+    ):
+        changes.pop("api_key")
+    try:
+        definition = CustomProviderDefinition.model_validate(
+            values | changes | {"provider_id": provider_id}
+        )
+    except ValidationError as exc:
+        errors = exc.errors(include_input=False, include_context=False)
+        raise ValueError(
+            "; ".join(
+                f"{'.'.join(map(str, error['loc']))}: {error['msg']}"
+                for error in errors
+            )
+        ) from None
+    if any(
+        item.provider_id != provider_id
+        and item.display_name.casefold() == definition.display_name.casefold()
+        for item in definitions.values()
+    ):
+        raise ValueError("A custom provider with this name already exists")
+    definitions[provider_id] = definition
+    target[CUSTOM_PROVIDERS_ENV] = encode_custom_providers(tuple(definitions.values()))
+    return provider_id
diff --git a/src/free_claude_code/config/admin/persistence.py b/src/free_claude_code/config/admin/persistence.py
index 7d5bb48617..877be50195 100644
--- a/src/free_claude_code/config/admin/persistence.py
+++ b/src/free_claude_code/config/admin/persistence.py
@@ -18,6 +18,7 @@
 from free_claude_code.config.settings import Settings
 from free_claude_code.core.json_types import JsonObject
 
+from .custom_providers import CustomProviderMutation, merge_custom_provider
 from .manifest import FIELD_BY_KEY, FIELDS
 from .state import ConfigInputValue
 from .validation import settings_from_values
@@ -39,6 +40,7 @@ class PreparedAdminUpdate:
     pending_fields: tuple[str, ...]
     path: Path
     changed_keys: tuple[str, ...] = ()
+    custom_provider_id: str | None = None
 
     @property
     def valid(self) -> bool:
@@ -61,6 +63,7 @@ def applied_response(self) -> JsonObject:
                 "pending_fields": [],
             }
         return {
+            "custom_provider_id": self.custom_provider_id,
             "applied": True,
             "valid": True,
             "errors": [],
@@ -129,11 +132,18 @@ def prepare_admin_update(
     updates: Mapping[str, ConfigInputValue],
     snapshot: ManagedConfigSnapshot,
     active_settings: Settings,
+    custom_provider: CustomProviderMutation | None = None,
 ) -> PreparedAdminUpdate:
     """Validate an update and construct its prospective Settings snapshot."""
 
     update_errors = _update_protocol_errors(updates)
     target_values = target_values_with_updates(updates, snapshot)
+    custom_id = None
+    if custom_provider is not None:
+        try:
+            custom_id = merge_custom_provider(target_values, snapshot, custom_provider)
+        except ValueError as exc:
+            update_errors = (*update_errors, str(exc))
     settings, settings_errors = settings_from_values(target_values, snapshot.process)
     errors = (
         *update_errors,
@@ -164,6 +174,7 @@ def prepare_admin_update(
         pending_fields=pending_fields,
         path=snapshot.path,
         changed_keys=changed_keys,
+        custom_provider_id=custom_id,
     )
 
 
diff --git a/src/free_claude_code/config/admin/values.py b/src/free_claude_code/config/admin/values.py
index 3069d448cc..37f537fe49 100644
--- a/src/free_claude_code/config/admin/values.py
+++ b/src/free_claude_code/config/admin/values.py
@@ -2,6 +2,7 @@
 
 from enum import Enum
 
+from free_claude_code.config.custom_providers import REASONING_FORMATS
 from free_claude_code.config.loader import ConfigSource, ManagedConfigSnapshot
 from free_claude_code.core.json_types import JsonObject
 
@@ -112,4 +113,14 @@ def load_config_response(snapshot: ManagedConfigSnapshot) -> JsonObject:
         "fields": fields,
         "paths": {"managed": str(snapshot.path)},
         "provider_status": provider_config_status(state),
+        "custom_providers": [
+            item.model_dump(mode="json")
+            | {"api_key": MASKED_SECRET if item.api_key else None}
+            for item in snapshot.settings.custom_providers
+        ],
+        "custom_providers_locked": snapshot.sources["custom_providers"]
+        == ConfigSource.PROCESS,
+        "custom_reasoning_formats": {
+            key: list(values) for key, values in REASONING_FORMATS.items()
+        },
     }
diff --git a/src/free_claude_code/config/custom_providers.py b/src/free_claude_code/config/custom_providers.py
new file mode 100644
index 0000000000..eb2767a1b4
--- /dev/null
+++ b/src/free_claude_code/config/custom_providers.py
@@ -0,0 +1,119 @@
+"""Saved custom endpoint definitions, independent of provider resources."""
+
+import json
+from typing import Literal
+
+from pydantic import (
+    AnyHttpUrl,
+    BaseModel,
+    ConfigDict,
+    Field,
+    SecretStr,
+    TypeAdapter,
+    field_validator,
+    model_validator,
+)
+
+CUSTOM_PROVIDERS_ENV = "FCC_CUSTOM_PROVIDERS"
+REASONING_FORMATS = {
+    "openai_chat": (
+        "provider_default",
+        "openai_effort",
+        "limited_effort",
+        "reasoning_object",
+        "thinking",
+        "chat_template",
+    ),
+    "openai_responses": ("provider_default", "native_responses"),
+    "anthropic_messages": ("provider_default", "messages_manual", "messages_adaptive"),
+}
+
+
+class CustomProviderDefinition(BaseModel):
+    model_config = ConfigDict(frozen=True, extra="forbid", hide_input_in_errors=True)
+
+    provider_id: str = Field(pattern=r"^custom_[0-9a-f]{32}$")
+    display_name: str = Field(min_length=1)
+    base_url: str
+    api_key: SecretStr | None = Field(default=None, repr=False)
+    api_format: Literal["openai_chat", "openai_responses", "anthropic_messages"] = (
+        "openai_chat"
+    )
+    reasoning_format: str = "provider_default"
+    reasoning_history_format: Literal[
+        "disabled", "reasoning_content", "reasoning", "think_tags"
+    ] = "disabled"
+    model_ids: tuple[str, ...] = ()
+
+    @field_validator("display_name", mode="before")
+    @classmethod
+    def trim_name(cls, value: object) -> object:
+        return value.strip() if isinstance(value, str) else value
+
+    @field_validator("base_url")
+    @classmethod
+    def validate_base(cls, value: str) -> str:
+        try:
+            url = TypeAdapter(AnyHttpUrl).validate_python(value.strip())
+        except ValueError:
+            raise ValueError("Base URL must be an HTTP or HTTPS API address") from None
+        if (
+            url.username
+            or url.password
+            or url.query is not None
+            or url.fragment is not None
+        ):
+            raise ValueError(
+                "Base URL cannot contain credentials, a query or a fragment"
+            )
+        return str(url).rstrip("/")
+
+    @field_validator("api_key", mode="before")
+    @classmethod
+    def empty_key(cls, value: object) -> object:
+        return value.strip() or None if isinstance(value, str) else value
+
+    @field_validator("model_ids")
+    @classmethod
+    def normalize_models(cls, value: tuple[str, ...]) -> tuple[str, ...]:
+        return tuple(dict.fromkeys(model.strip() for model in value if model.strip()))
+
+    @model_validator(mode="after")
+    def validate_formats(self) -> CustomProviderDefinition:
+        if self.reasoning_format not in REASONING_FORMATS[self.api_format]:
+            raise ValueError("Reasoning format is not supported by this API format")
+        if (
+            self.api_format != "openai_chat"
+            and self.reasoning_history_format != "disabled"
+        ):
+            raise ValueError(
+                "Reasoning history format applies only to Chat Completions"
+            )
+        return self
+
+    def private_values(self) -> dict:
+        return self.model_dump(mode="json") | {
+            "api_key": self.api_key.get_secret_value() if self.api_key else None,
+        }
+
+    def same_inventory(self, other: CustomProviderDefinition) -> bool:
+        return self.model_dump(exclude={"display_name"}) == other.model_dump(
+            exclude={"display_name"}
+        )
+
+
+def decode_custom_providers(value: object) -> object:
+    if isinstance(value, str):
+        try:
+            return json.loads(value) if value.strip() else []
+        except ValueError:
+            raise ValueError("Custom providers must be a JSON array") from None
+    return value
+
+
+def encode_custom_providers(definitions: tuple[CustomProviderDefinition, ...]) -> str:
+    return json.dumps(
+        [item.private_values() for item in definitions],
+        ensure_ascii=False,
+        separators=(",", ":"),
+    )
diff --git a/src/free_claude_code/config/env_migrations.py b/src/free_claude_code/config/env_migrations.py
index 6a251dad9b..e62d8a041f 100644
--- a/src/free_claude_code/config/env_migrations.py
+++ b/src/free_claude_code/config/env_migrations.py
@@ -173,7 +173,7 @@ def secret_env_keys() -> frozenset[str]:
         for name in Settings.model_fields
         if name.endswith(("_api_key", "_proxy", "_token"))
     }
-    secret_attrs.update({"telegram_proxy_url", "proxy_auth_token"})
+    secret_attrs.update({"telegram_proxy_url", "proxy_auth_token", "custom_providers"})
     return frozenset(
         str(Settings.model_fields[name].validation_alias) for name in secret_attrs
     )
diff --git a/src/free_claude_code/config/settings.py b/src/free_claude_code/config/settings.py
index a0bf711187..6f3fa9a119 100644
--- a/src/free_claude_code/config/settings.py
+++ b/src/free_claude_code/config/settings.py
@@ -13,6 +13,7 @@
 )
 
 from .constants import DEFAULT_MODEL, HTTP_CONNECT_TIMEOUT_DEFAULT
+from .custom_providers import CustomProviderDefinition, decode_custom_providers
 from .model_refs import parse_model_fallbacks
 from .nim import NimSettings
 from .provider_catalog import (
@@ -50,9 +51,6 @@ def _validate_model_ref(value: str) -> str:
             f"Valid providers: {', '.join(SUPPORTED_PROVIDER_IDS)}. "
             "Format: provider_type/model/name"
         )
-    if provider not in SUPPORTED_PROVIDER_IDS:
-        supported = ", ".join(f"'{item}'" for item in SUPPORTED_PROVIDER_IDS)
-        raise ValueError(f"Invalid provider: '{provider}'. Supported: {supported}")
     if not model:
         raise ValueError("Model reference must include a non-empty model suffix.")
     return value
@@ -65,8 +63,49 @@ class Settings(BaseModel):
         validate_default=True,
         populate_by_name=True,
         extra="ignore",
+        hide_input_in_errors=True,
     )
 
+    custom_providers: Annotated[
+        tuple[CustomProviderDefinition, ...], BeforeValidator(decode_custom_providers)
+    ] = Field(default=(), validation_alias="FCC_CUSTOM_PROVIDERS")
+
+    @property
+    def provider_ids(self) -> tuple[str, ...]:
+        return (
+            *SUPPORTED_PROVIDER_IDS,
+            *(item.provider_id for item in self.custom_providers),
+        )
+
+    def custom_provider(self, provider_id: str) -> CustomProviderDefinition | None:
+        return next(
+            (item for item in self.custom_providers if item.provider_id == provider_id),
+            None,
+        )
+
+    @model_validator(mode="after")
+    def validate_provider_references(self) -> Settings:
+        ids = [item.provider_id for item in self.custom_providers]
+        names = [item.display_name.casefold() for item in self.custom_providers]
+        if len(ids) != len(set(ids)) or len(names) != len(set(names)):
+            raise ValueError("Custom provider names and IDs must be unique")
+        for field in (
+            "model",
+            "model_fable",
+            "model_opus",
+            "model_sonnet",
+            "model_haiku",
+            "model_fallbacks",
+        ):
+            value = getattr(self, field)
+            refs = value if isinstance(value, tuple) else (value,) if value else ()
+            for ref in refs:
+                if ref.partition("/")[0] not in self.provider_ids:
+                    raise ValueError(
+                        f"{field.upper()}: Invalid provider in model reference"
+                    )
+        return self
+
     # ==================== OpenAI Platform API ====================
     openai_api_key: OptionalNonEmptyString = Field(
         default=None, validation_alias="OPENAI_API_KEY"
diff --git a/src/free_claude_code/providers/anthropic_messages/request_policy.py b/src/free_claude_code/providers/anthropic_messages/request_policy.py
index 7189442af2..23b6b56dd3 100644
--- a/src/free_claude_code/providers/anthropic_messages/request_policy.py
+++ b/src/free_claude_code/providers/anthropic_messages/request_policy.py
@@ -43,9 +43,12 @@ def resolve_messages_options(
     capabilities: MessagesModelCapabilities = MessagesModelCapabilities(),
     thinking: ThinkingConfig | None = None,
     output_effort: object = None,
+    preserve_native_controls: bool = False,
 ) -> NativeMessagesOptions:
     """Encode supported controls without conflating effort and thinking mode."""
 
+    if preserve_native_controls:
+        reasoning = ReasoningPolicy.provider_default()
     if output_effort is not None:
         if not isinstance(
             output_effort, str
@@ -109,6 +112,10 @@ def resolve_messages_options(
                 and not reasoning.requests_reasoning
             ):
                 exact = thinking.budget_tokens
+            if preserve_native_controls and exact is None:
+                raise NativeMessagesError(
+                    "Native manual thinking requires a token budget."
+                )
             derived = max(_MIN_THINKING_BUDGET, reasoning.numeric_budget_tokens or 2048)
             effective = exact if exact is not None else min(derived, limit - 1)
             if effective < _MIN_THINKING_BUDGET or effective >= limit:
@@ -153,6 +160,7 @@ def resolve_messages_options(
             and supported is None
             and native_mode is None
             and mode is None
+            and not preserve_native_controls
         ):
             raise NativeMessagesError(
                 "This model does not advertise output effort support."
diff --git a/src/free_claude_code/providers/anthropic_messages/transport.py b/src/free_claude_code/providers/anthropic_messages/transport.py
index 4579c7cfd5..7710ac7d31 100644
--- a/src/free_claude_code/providers/anthropic_messages/transport.py
+++ b/src/free_claude_code/providers/anthropic_messages/transport.py
@@ -4,6 +4,7 @@
 import json
 import sys
 from collections.abc import AsyncIterator, Callable, Mapping
+from dataclasses import replace
 from functools import partial
 from typing import cast
 
@@ -103,26 +104,31 @@ def __init__(
         self._read_timeout_s = read_timeout_s
         self._capabilities = capabilities
 
+    def _effective_capabilities(
+        self, model_info: ProviderModelInfo | None
+    ) -> MessagesModelCapabilities:
+        if model_info is None or model_info.max_output_tokens is None:
+            return self._capabilities
+        cap = model_info.max_output_tokens
+        if self._capabilities.max_output_tokens is not None:
+            cap = min(cap, self._capabilities.max_output_tokens)
+        return replace(self._capabilities, max_output_tokens=cap)
+
     def _messages_body(
         self,
         request: MessagesRequest,
         reasoning: ReasoningPolicy,
-        model_info: ProviderModelInfo | None = None,
+        capabilities: MessagesModelCapabilities,
+        preserve_native_controls: bool,
     ) -> PreparedMessagesRequest:
         validate_history(request.model_dump(mode="json"))
-        request, reasoning = prepare_messages_reasoning(
-            request,
-            reasoning,
-            model_info=model_info,
-            can_disable=True,
-            normal_max_tokens=DEFAULT_MESSAGES_OUTPUT_TOKENS,
-        )
         try:
             options = resolve_messages_options(
                 model=request.model,
                 max_tokens=request.max_tokens,
                 reasoning=reasoning,
-                capabilities=self._capabilities,
+                capabilities=capabilities,
+                preserve_native_controls=preserve_native_controls,
                 thinking=request.thinking,
                 output_effort=request.output_config.get("effort")
                 if request.output_config
@@ -133,7 +139,10 @@ def _messages_body(
             raise InvalidRequestError(str(error)) from error
 
     def _responses_body(
-        self, request: OpenAIResponsesRequest, reasoning: ReasoningPolicy
+        self,
+        request: OpenAIResponsesRequest,
+        reasoning: ReasoningPolicy,
+        capabilities: MessagesModelCapabilities,
     ) -> ResponsesMessagesRequest:
         validate_history(request.model_dump(mode="json"))
         try:
@@ -141,7 +150,7 @@ def _responses_body(
                 model=request.model,
                 max_tokens=request.max_output_tokens,
                 reasoning=reasoning,
-                capabilities=self._capabilities,
+                capabilities=capabilities,
                 output_effort=request.reasoning.get("effort")
                 if request.reasoning
                 else None,
@@ -159,7 +168,11 @@ def stream_messages(
         response_model: str | None = None,
         reasoning: ReasoningPolicy = DEFAULT_REASONING_POLICY,
         model_info: ProviderModelInfo | None = None,
+        preserve_native_controls: bool = False,
     ) -> AsyncIterator[str]:
+        capabilities = self._effective_capabilities(model_info)
+        if preserve_native_controls:
+            reasoning = ReasoningPolicy.provider_default()
         prepared_request, wire_reasoning = prepare_messages_reasoning(
             request,
             reasoning,
@@ -167,13 +180,15 @@ def stream_messages(
             can_disable=True,
             normal_max_tokens=DEFAULT_MESSAGES_OUTPUT_TOKENS,
         )
-        prepared = self._messages_body(prepared_request, wire_reasoning)
+        prepared = self._messages_body(
+            prepared_request, wire_reasoning, capabilities, preserve_native_controls
+        )
         correction = (
             ReasoningCorrection(
                 (("thinking",),),
                 "max_tokens",
                 DEFAULT_MESSAGES_OUTPUT_TOKENS,
-                self._capabilities.max_output_tokens,
+                capabilities.max_output_tokens,
             )
             if reasoning.control is ReasoningControl.PREFER_OFF
             and wire_reasoning.control is ReasoningControl.OFF
@@ -198,8 +213,11 @@ def stream_responses(
         request_id: str | None = None,
         response_model: str | None = None,
         reasoning: ReasoningPolicy = DEFAULT_REASONING_POLICY,
+        model_info: ProviderModelInfo | None = None,
     ) -> AsyncIterator[str]:
-        prepared = self._responses_body(request, reasoning)
+        prepared = self._responses_body(
+            request, reasoning, self._effective_capabilities(model_info)
+        )
         return self._stream(
             prepared.body,
             betas=(),
@@ -299,7 +317,7 @@ async def _run(
                         dict.fromkeys([*filter(None, existing.split(",")), *betas])
                     )
                 base_url = endpoint.base_url.rstrip("/")
-                path = "/messages" if base_url.endswith("/v1") else "/v1/messages"
+                path = "/messages"
                 response = scope.retain(
                     await self._client.send(
                         self._client.build_request(
diff --git a/src/free_claude_code/providers/base.py b/src/free_claude_code/providers/base.py
index 283f77662e..ecf11ab817 100644
--- a/src/free_claude_code/providers/base.py
+++ b/src/free_claude_code/providers/base.py
@@ -69,5 +69,6 @@ def stream_responses(
         response_model: str | None = None,
         reasoning: ReasoningPolicy = DEFAULT_REASONING_POLICY,
         request_headers: Mapping[str, str] | None = None,
+        model_info: ProviderModelInfo | None = None,
     ) -> AsyncIterator[str]:
         """Validate the request before yielding OpenAI Responses SSE events."""
diff --git a/src/free_claude_code/providers/custom.py b/src/free_claude_code/providers/custom.py
new file mode 100644
index 0000000000..07f51f8a68
--- /dev/null
+++ b/src/free_claude_code/providers/custom.py
@@ -0,0 +1,380 @@
+"""Custom endpoint resource ownership using FCC's existing wire transports."""
+
+from collections.abc import AsyncIterator, Mapping
+
+import httpx
+import httpx2
+
+from free_claude_code.application.errors import InvalidRequestError
+from free_claude_code.application.model_metadata import ProviderModelInfo
+from free_claude_code.config.custom_providers import CustomProviderDefinition
+from free_claude_code.core.anthropic import MessagesRequest, ReasoningReplayMode
+from free_claude_code.core.model_capabilities import ModelInputModality
+from free_claude_code.core.openai_responses import (
+    OpenAIResponsesRequest,
+    responses_reasoning_policy,
+)
+from free_claude_code.core.reasoning import (
+    DEFAULT_REASONING_POLICY,
+    ReasoningEffort,
+    ReasoningPolicy,
+)
+from free_claude_code.providers.admission import (
+    ProviderAdmissionController,
+    ProviderOperationKind,
+)
+from free_claude_code.providers.anthropic_messages.request_policy import (
+    MessagesModelCapabilities,
+)
+from free_claude_code.providers.anthropic_messages.transport import (
+    AnthropicMessagesTransport,
+)
+from free_claude_code.providers.base import BaseProvider, ProviderConfig
+from free_claude_code.providers.endpoint_types import HttpEndpoint
+from free_claude_code.providers.model_listing import (
+    ModelListResponseError,
+    extract_openai_model_infos,
+    model_infos_from_ids,
+)
+from free_claude_code.providers.openai_chat import (
+    NO_REASONING,
+    ChatTemplateReasoning,
+    NamedEffortReasoning,
+    OpenAIChatBehavior,
+    OpenAIChatProfile,
+    OpenAIChatRequestPolicy,
+    OpenAIChatTransport,
+    ReasoningEncoder,
+    ReasoningObject,
+    ThinkingObjectReasoning,
+    create_chat_client,
+)
+from free_claude_code.providers.openai_client import OpenAIRequestClient
+from free_claude_code.providers.openai_responses import OpenAIResponsesTransport
+
+
+def _reasoning_encoder(name: str) -> ReasoningEncoder:
+    efforts = tuple(
+        (effort, "xhigh" if effort is ReasoningEffort.MAX else effort.value)
+        for effort in ReasoningEffort
+    )
+    limited = tuple(
+        (
+            effort,
+            "low"
+            if effort is ReasoningEffort.MINIMAL
+            else "high"
+            if effort in {ReasoningEffort.XHIGH, ReasoningEffort.MAX}
+            else effort.value,
+        )
+        for effort in ReasoningEffort
+    )
+    return {
+        "provider_default": NO_REASONING,
+        "openai_effort": NamedEffortReasoning(
+            efforts, disabled_value="none", enabled_value="medium"
+        ),
+        "limited_effort": NamedEffortReasoning(limited, enabled_value="medium"),
+        "reasoning_object": ReasoningObject(limited),
+        "thinking": ThinkingObjectReasoning({"type": "enabled"}, {"type": "disabled"}),
+        "chat_template": ChatTemplateReasoning("enable_thinking"),
+    }[name]
+
+
+async def _omit_cookies(request: httpx.Request) -> None:
+    request.headers.pop("cookie", None)
+
+
+class CustomProvider(BaseProvider):
+    def __init__(
+        self,
+        config: ProviderConfig,
+        *,
+        definition: CustomProviderDefinition,
+        admission: ProviderAdmissionController,
+        openai_transport: httpx2.AsyncBaseTransport | None = None,
+        messages_transport: httpx.AsyncBaseTransport | None = None,
+    ):
+        super().__init__(config)
+        self._definition = definition
+        self._admission = admission
+        self._snapshot = HttpEndpoint(
+            base_url=config.base_url, api_key=config.api_key, headers={}
+        )
+        self._openai_transport = openai_transport
+        self._sdk = None
+        self._http = None
+        self._chat = None
+        self._responses = None
+        self._messages = None
+        if definition.api_format == "anthropic_messages":
+            self._http = httpx.AsyncClient(
+                timeout=httpx.Timeout(
+                    config.http_read_timeout,
+                    connect=config.http_connect_timeout,
+                    write=config.http_write_timeout,
+                ),
+                transport=messages_transport,
+                follow_redirects=False,
+                event_hooks={"request": [_omit_cookies]},
+            )
+            mode = definition.reasoning_format
+            capabilities = MessagesModelCapabilities(
+                adaptive_thinking="unsupported"
+                if mode == "messages_manual"
+                else "required"
+                if mode == "messages_adaptive"
+                else None,
+                supports_output_effort=False
+                if mode == "messages_manual"
+                else True
+                if mode == "messages_adaptive"
+                else None,
+                supported_efforts=("low", "medium", "high", "xhigh", "max")
+                if mode == "messages_adaptive"
+                else None,
+            )
+            self._messages = AnthropicMessagesTransport(
+                client=self._http,
+                admission=admission,
+                provider_name=definition.provider_id,
+                replay_scope=definition.provider_id,
+                read_timeout_s=config.http_read_timeout,
+                capabilities=capabilities,
+            )
+        else:
+
+            async def credential() -> str:
+                return config.api_key or ""
+
+            self._sdk = create_chat_client(
+                config,
+                base_url=config.base_url,
+                provider_name=definition.provider_id,
+                api_key_provider=credential,
+            )
+            if definition.api_format == "openai_chat":
+                profile = OpenAIChatProfile(
+                    OpenAIChatRequestPolicy(
+                        definition.provider_id,
+                        ReasoningReplayMode(definition.reasoning_history_format),
+                    ),
+                    _reasoning_encoder(definition.reasoning_format),
+                    reasoning_delta_fallback_field="reasoning",
+                )
+                self._chat = OpenAIChatTransport(
+                    client=self._sdk,
+                    admission=admission,
+                    behavior=OpenAIChatBehavior(profile),
+                    read_timeout_s=config.http_read_timeout,
+                    log_raw_sse_events=config.log_raw_sse_events,
+                    log_api_error_tracebacks=config.log_api_error_tracebacks,
+                    endpoint_transport=openai_transport,
+                )
+            else:
+                self._responses = OpenAIResponsesTransport(
+                    client=self._sdk,
+                    admission=admission,
+                    provider_name=definition.provider_id,
+                    read_timeout_s=config.http_read_timeout,
+                    log_raw_sse_events=config.log_raw_sse_events,
+                    endpoint_transport=openai_transport,
+                )
+
+    async def endpoint(self, *, force_refresh: bool = False) -> HttpEndpoint:
+        return self._snapshot
+
+    async def cleanup(self) -> None:
+        if self._sdk is not None:
+            await self._sdk.close()
+        if self._http is not None:
+            await self._http.aclose()
+        if self._openai_transport is not None:
+            await self._openai_transport.aclose()
+
+    def _reasoning(self, reasoning: ReasoningPolicy) -> ReasoningPolicy:
+        mode = self._definition.reasoning_format
+        if mode == "provider_default":
+            return DEFAULT_REASONING_POLICY
+        if reasoning.budget_tokens is not None and mode in {
+            "openai_effort",
+            "limited_effort",
+            "native_responses",
+            "messages_adaptive",
+        }:
+            raise InvalidRequestError(
+                "The selected reasoning format cannot represent an exact token budget"
+            )
+        return reasoning
+
+    def stream_messages(
+        self,
+        request: MessagesRequest,
+        input_tokens: int = 0,
+        *,
+        request_id: str | None = None,
+        response_model: str | None = None,
+        reasoning: ReasoningPolicy = DEFAULT_REASONING_POLICY,
+        request_headers: Mapping[str, str] | None = None,
+        model_info: ProviderModelInfo | None = None,
+    ) -> AsyncIterator[str]:
+        reasoning = self._reasoning(reasoning)
+        if (
+            self._definition.reasoning_format == "provider_default"
+            and self._messages is None
+        ):
+            request = request.model_copy(
+                update={
+                    "thinking": None,
+                    "output_config": {
+                        key: value
+                        for key, value in (request.output_config or {}).items()
+                        if key != "effort"
+                    },
+                }
+            )
+        if self._messages is not None:
+            return self._messages.stream_messages(
+                request,
+                endpoint_context=self,
+                request_id=request_id,
+                response_model=response_model,
+                reasoning=reasoning,
+                model_info=model_info,
+                preserve_native_controls=self._definition.reasoning_format
+                == "provider_default",
+            )
+        transport = self._chat if self._chat is not None else self._responses
+        assert transport is not None
+        return transport.stream_messages(
+            request,
+            input_tokens=input_tokens,
+            endpoint_context=self,
+            request_id=request_id,
+            response_model=response_model or request.model,
+            reasoning=reasoning,
+            model_info=model_info,
+        )
+
+    def stream_responses(
+        self,
+        request: OpenAIResponsesRequest,
+        input_tokens: int = 0,
+        *,
+        request_id: str | None = None,
+        response_model: str | None = None,
+        reasoning: ReasoningPolicy = DEFAULT_REASONING_POLICY,
+        request_headers: Mapping[str, str] | None = None,
+        model_info: ProviderModelInfo | None = None,
+    ) -> AsyncIterator[str]:
+        reasoning = (
+            responses_reasoning_policy(request.reasoning)
+            if self._definition.reasoning_format == "provider_default"
+            and self._responses is not None
+            else self._reasoning(reasoning)
+        )
+        if (
+            self._definition.reasoning_format == "provider_default"
+            and self._responses is None
+        ):
+            request = request.model_copy(
+                update={
+                    "reasoning": {
+                        key: value
+                        for key, value in (request.reasoning or {}).items()
+                        if key != "effort"
+                    }
+                }
+            )
+        if self._messages is not None:
+            return self._messages.stream_responses(
+                request,
+                endpoint_context=self,
+                request_id=request_id,
+                response_model=response_model,
+                reasoning=reasoning,
+                model_info=model_info,
+            )
+        transport = self._chat if self._chat is not None else self._responses
+        assert transport is not None
+        return transport.stream_responses(
+            request,
+            input_tokens=input_tokens,
+            endpoint_context=self,
+            request_id=request_id,
+            response_model=response_model or request.model,
+            reasoning=reasoning,
+        )
+
+    async def list_model_infos(self) -> frozenset[ProviderModelInfo]:
+        if self._definition.model_ids:
+            return model_infos_from_ids(self._definition.model_ids)
+        if self._sdk is not None:
+            binding = OpenAIRequestClient(self._openai_transport)
+            try:
+                client = binding.for_endpoint(self._sdk, self._snapshot)
+                payload = await self._admission.start_execution().run_call(
+                    lambda: client.models.list(extra_headers=binding.openai_headers()),
+                    operation_kind=ProviderOperationKind.MODEL_DISCOVERY,
+                )
+                return extract_openai_model_infos(
+                    payload, provider_name=self._definition.provider_id
+                )
+            finally:
+                await binding.aclose()
+        assert self._http is not None
+        headers = {"anthropic-version": "2023-06-01"}
+        if self._config.api_key:
+            headers["x-api-key"] = self._config.api_key
+        params = {"limit": "1000"}
+        seen: set[str] = set()
+        infos: dict[str, ProviderModelInfo] = {}
+        for _ in range(100):
+            response = await self._admission.start_execution().run_call(
+                lambda params=params: self._fetch_models(headers, params),
+                operation_kind=ProviderOperationKind.MODEL_DISCOVERY,
+            )
+            if not isinstance(response, dict) or not isinstance(
+                response.get("has_more"), bool
+            ):
+                raise ModelListResponseError("Invalid Messages model-list page")
+            page = extract_openai_model_infos(
+                response,
+                provider_name=self._definition.provider_id,
+                thinking_boolean_path=("capabilities", "thinking", "supported"),
+                fixed_input_modalities=frozenset({ModelInputModality.TEXT}),
+                input_modality_boolean_paths=(
+                    (
+                        ModelInputModality.IMAGE,
+                        ("capabilities", "image_input", "supported"),
+                    ),
+                ),
+                context_window_tokens_path=("max_input_tokens",),
+                max_output_tokens_path=("max_tokens",),
+            )
+            infos.update((info.model_id, info) for info in page)
+            if not response["has_more"]:
+                return frozenset(infos.values())
+            cursor = response.get("last_id")
+            if (
+                not isinstance(cursor, str)
+                or not cursor
+                or cursor in seen
+                or cursor not in {info.model_id for info in page}
+            ):
+                raise ModelListResponseError(
+                    "Messages model-list cursor did not advance"
+                )
+            seen.add(cursor)
+            params = {"limit": "1000", "after_id": cursor}
+        raise ModelListResponseError("Messages model-list exceeded 100 pages")
+
+    async def _fetch_models(
+        self, headers: dict[str, str], params: dict[str, str]
+    ) -> object:
+        assert self._http is not None
+        response = await self._http.get(
+            self._config.base_url + "/models", headers=headers, params=params
+        )
+        response.raise_for_status()
+        return response.json()
diff --git a/src/free_claude_code/providers/github_copilot/provider.py b/src/free_claude_code/providers/github_copilot/provider.py
index a816949c5d..c78815aec1 100644
--- a/src/free_claude_code/providers/github_copilot/provider.py
+++ b/src/free_claude_code/providers/github_copilot/provider.py
@@ -4,6 +4,7 @@
 import sys
 from collections.abc import AsyncIterator, Mapping
 from contextlib import suppress
+from dataclasses import replace
 
 import httpx
 import httpx2
@@ -58,7 +59,11 @@ def __init__(self, context: EndpointContext, client: httpx.AsyncClient) -> None:
     async def endpoint(self, *, force_refresh: bool = False) -> HttpEndpoint:
         snapshot = await self._context.endpoint(force_refresh=force_refresh)
         self._client.cookies.clear()
-        return snapshot
+        base_url = snapshot.base_url.rstrip("/")
+        return replace(
+            snapshot,
+            base_url=base_url if base_url.endswith("/v1") else base_url + "/v1",
+        )
 
 
 class GitHubCopilotProvider(BaseProvider):
@@ -156,6 +161,7 @@ def stream_responses(
         response_model: str | None = None,
         reasoning: ReasoningPolicy = DEFAULT_REASONING_POLICY,
         request_headers: Mapping[str, str] | None = None,
+        model_info: ProviderModelInfo | None = None,
     ) -> AsyncIterator[str]:
         self._check_model(request.model)
         return self._dispatch(
diff --git a/src/free_claude_code/providers/lmstudio/client.py b/src/free_claude_code/providers/lmstudio/client.py
index 794385edee..3967efebea 100644
--- a/src/free_claude_code/providers/lmstudio/client.py
+++ b/src/free_claude_code/providers/lmstudio/client.py
@@ -126,6 +126,7 @@ def stream_responses(
         reasoning: ReasoningPolicy = DEFAULT_REASONING_POLICY,
         endpoint_context: EndpointContext | None = None,
         request_headers: Mapping[str, str] | None = None,
+        model_info: ProviderModelInfo | None = None,
     ) -> AsyncIterator[str]:
         stream = super().stream_responses(
             request,
@@ -135,6 +136,7 @@ def stream_responses(
             reasoning=reasoning,
             endpoint_context=endpoint_context,
             request_headers=request_headers,
+            model_info=model_info,
         )
         return self._stream_with_context_budget(
             stream,
diff --git a/src/free_claude_code/providers/openai_api/provider.py b/src/free_claude_code/providers/openai_api/provider.py
index cc5f8ea9f2..5e17e55d43 100644
--- a/src/free_claude_code/providers/openai_api/provider.py
+++ b/src/free_claude_code/providers/openai_api/provider.py
@@ -140,6 +140,7 @@ def stream_responses(
         response_model: str | None = None,
         reasoning: ReasoningPolicy = DEFAULT_REASONING_POLICY,
         request_headers: Mapping[str, str] | None = None,
+        model_info: ProviderModelInfo | None = None,
     ) -> AsyncIterator[str]:
         return self._responses.stream_responses(
             request,
diff --git a/src/free_claude_code/providers/openai_chat/__init__.py b/src/free_claude_code/providers/openai_chat/__init__.py
index f555d0ea4a..11f40ba1b8 100644
--- a/src/free_claude_code/providers/openai_chat/__init__.py
+++ b/src/free_claude_code/providers/openai_chat/__init__.py
@@ -16,7 +16,9 @@
     NO_REASONING,
     ChatTemplateReasoning,
     NamedEffortReasoning,
+    ReasoningEncoder,
     ReasoningObject,
+    ThinkingObjectReasoning,
 )
 from .request_policy import (
     OpenAIChatRequestPolicy,
@@ -60,7 +62,9 @@ def create_openai_chat_provider(
     "OpenAIChatRequestPolicy",
     "OpenAIChatTransport",
     "OpenAIModelListing",
+    "ReasoningEncoder",
     "ReasoningObject",
+    "ThinkingObjectReasoning",
     "apply_openai_chat_body_policy",
     "build_openai_chat_request_body",
     "create_chat_client",
diff --git a/src/free_claude_code/providers/openai_chat/provider.py b/src/free_claude_code/providers/openai_chat/provider.py
index 86fcb8bfdd..1f366dec60 100644
--- a/src/free_claude_code/providers/openai_chat/provider.py
+++ b/src/free_claude_code/providers/openai_chat/provider.py
@@ -233,6 +233,7 @@ def stream_responses(
         reasoning: ReasoningPolicy = DEFAULT_REASONING_POLICY,
         endpoint_context: EndpointContext | None = None,
         request_headers: Mapping[str, str] | None = None,
+        model_info: ProviderModelInfo | None = None,
     ) -> AsyncIterator[str]:
         return self._chat.stream_responses(
             request,
diff --git a/src/free_claude_code/providers/openai_codex/provider.py b/src/free_claude_code/providers/openai_codex/provider.py
index 8c20bb02a1..24dedca6d1 100644
--- a/src/free_claude_code/providers/openai_codex/provider.py
+++ b/src/free_claude_code/providers/openai_codex/provider.py
@@ -187,6 +187,7 @@ def stream_responses(
         response_model: str | None = None,
         reasoning: ReasoningPolicy = DEFAULT_REASONING_POLICY,
         request_headers: Mapping[str, str] | None = None,
+        model_info: ProviderModelInfo | None = None,
     ) -> AsyncIterator[str]:
         return self._responses.stream_responses(
             request,
diff --git a/src/free_claude_code/providers/opencode/provider.py b/src/free_claude_code/providers/opencode/provider.py
index da51eec919..fdad199361 100644
--- a/src/free_claude_code/providers/opencode/provider.py
+++ b/src/free_claude_code/providers/opencode/provider.py
@@ -248,6 +248,7 @@ def stream_responses(
         reasoning: ReasoningPolicy = DEFAULT_REASONING_POLICY,
         endpoint_context: EndpointContext | None = None,
         request_headers: Mapping[str, str] | None = None,
+        model_info: ProviderModelInfo | None = None,
     ) -> AsyncIterator[str]:
         return self._dispatch_responses_stream(
             request,
diff --git a/src/free_claude_code/providers/runtime/config.py b/src/free_claude_code/providers/runtime/config.py
index 1e17cb5a4f..a36e9b921c 100644
--- a/src/free_claude_code/providers/runtime/config.py
+++ b/src/free_claude_code/providers/runtime/config.py
@@ -1,6 +1,7 @@
 """Provider configuration construction from neutral catalog metadata."""
 
 from free_claude_code.application.errors import ApplicationUnavailableError
+from free_claude_code.config.custom_providers import CustomProviderDefinition
 from free_claude_code.config.provider_catalog import ProviderDescriptor
 from free_claude_code.config.settings import Settings
 from free_claude_code.providers.base import ProviderConfig
@@ -89,3 +90,21 @@ def build_provider_config(
         log_raw_sse_events=settings.log_raw_sse_events,
         log_api_error_tracebacks=settings.log_api_error_tracebacks,
     )
+
+
+def build_custom_provider_config(
+    definition: CustomProviderDefinition, settings: Settings
+) -> ProviderConfig:
+    return ProviderConfig(
+        api_key=definition.api_key.get_secret_value() if definition.api_key else None,
+        base_url=definition.base_url,
+        proxy=None,
+        rate_limit=settings.provider_rate_limit,
+        rate_window=settings.provider_rate_window,
+        max_concurrency=settings.provider_max_concurrency,
+        http_read_timeout=settings.http_read_timeout,
+        http_write_timeout=settings.http_write_timeout,
+        http_connect_timeout=settings.http_connect_timeout,
+        log_raw_sse_events=settings.log_raw_sse_events,
+        log_api_error_tracebacks=settings.log_api_error_tracebacks,
+    )
diff --git a/src/free_claude_code/providers/runtime/discovery.py b/src/free_claude_code/providers/runtime/discovery.py
index b47b67470a..69c2b8e9ca 100644
--- a/src/free_claude_code/providers/runtime/discovery.py
+++ b/src/free_claude_code/providers/runtime/discovery.py
@@ -44,7 +44,9 @@ def model_cache_provider_ids_for_settings(
     )
     available = set(configured) | set(connected_provider_ids)
     return tuple(
-        provider_id for provider_id in PROVIDER_CATALOG if provider_id in available
+        provider_id
+        for provider_id in settings.provider_ids
+        if provider_id in available or settings.custom_provider(provider_id) is not None
     )
 
 
@@ -59,5 +61,7 @@ def model_list_provider_ids_for_settings(
         for provider_id in model_cache_provider_ids_for_settings(
             settings, connected_provider_ids
         )
-        if not PROVIDER_CATALOG[provider_id].local or provider_id in referenced_ids
+        if settings.custom_provider(provider_id) is not None
+        or not PROVIDER_CATALOG[provider_id].local
+        or provider_id in referenced_ids
     )
diff --git a/src/free_claude_code/providers/runtime/factory.py b/src/free_claude_code/providers/runtime/factory.py
index 87f5fd91e7..cd08d7b9f5 100644
--- a/src/free_claude_code/providers/runtime/factory.py
+++ b/src/free_claude_code/providers/runtime/factory.py
@@ -7,6 +7,7 @@
     ApplicationUnavailableError,
     UnknownProviderError,
 )
+from free_claude_code.config.custom_providers import CustomProviderDefinition
 from free_claude_code.config.provider_catalog import PROVIDER_CATALOG
 from free_claude_code.config.settings import Settings
 from free_claude_code.providers.admission import ProviderAdmissionController
@@ -249,12 +250,31 @@ def _required_setting(settings: Settings, attr_name: str) -> str:
 def prepare_provider(
     provider_id: str,
     provider_loaders: Mapping[str, Callable[[], ProviderFactory]],
+    custom_definition: CustomProviderDefinition | None = None,
 ) -> Callable[[Settings], BaseProvider]:
     """Load implementation modules in a worker; return a loop-owned constructor."""
 
     # The SDK lazily imports these on first client resource access. Keep that
     # work in this loader, before constructing clients on their owner loop.
     importlib.import_module("openai.resources")
+    if custom_definition is not None:
+        from free_claude_code.providers.custom import CustomProvider
+
+        from .config import build_custom_provider_config
+
+        def construct_custom(settings: Settings) -> BaseProvider:
+            config = build_custom_provider_config(custom_definition, settings)
+            admission = ProviderAdmissionController(
+                provider_name=provider_id,
+                rate_limit=config.rate_limit,
+                rate_window=config.rate_window,
+                max_concurrency=config.max_concurrency,
+            )
+            return CustomProvider(
+                config, definition=custom_definition, admission=admission
+            )
+
+        return construct_custom
     descriptor = PROVIDER_CATALOG.get(provider_id)
     if descriptor is None:
         raise UnknownProviderError.for_provider(provider_id, PROVIDER_CATALOG)
diff --git a/src/free_claude_code/providers/runtime/model_cache.py b/src/free_claude_code/providers/runtime/model_cache.py
index ccd4bd40af..4452c342a9 100644
--- a/src/free_claude_code/providers/runtime/model_cache.py
+++ b/src/free_claude_code/providers/runtime/model_cache.py
@@ -78,7 +78,10 @@ def cached_model_info(
     def cached_prefixed_model_infos(self) -> tuple[ProviderModelInfo, ...]:
         """Return cached provider models with user-selectable prefixed ids."""
         infos: list[ProviderModelInfo] = []
-        for provider_id in SUPPORTED_PROVIDER_IDS:
+        for provider_id in (
+            *SUPPORTED_PROVIDER_IDS,
+            *sorted(self._available_provider_ids.difference(SUPPORTED_PROVIDER_IDS)),
+        ):
             provider_infos = self._model_infos_by_provider.get(provider_id, {})
             infos.extend(
                 replace(info, model_id=f"{provider_id}/{info.model_id}")
diff --git a/src/free_claude_code/providers/runtime/runtime.py b/src/free_claude_code/providers/runtime/runtime.py
index d60bffa9ae..0a2ebd09e7 100644
--- a/src/free_claude_code/providers/runtime/runtime.py
+++ b/src/free_claude_code/providers/runtime/runtime.py
@@ -18,11 +18,17 @@
 
 
 def _load_constructor(
-    provider_id: str, provider_loaders: Mapping[str, ProviderLoader]
+    provider_id: str, provider_loaders: Mapping[str, ProviderLoader], settings: Settings
 ) -> Callable[[Settings], BaseProvider]:
     from .factory import prepare_provider
 
-    return prepare_provider(provider_id, provider_loaders)
+    return prepare_provider(
+        provider_id,
+        provider_loaders,
+        settings.custom_provider(provider_id)
+        if provider_id.startswith("custom_")
+        else None,
+    )
 
 
 async def create_provider(
@@ -32,7 +38,7 @@ async def create_provider(
     provider_loaders: Mapping[str, ProviderLoader] | None = None,
 ) -> BaseProvider:
     constructor = await run_sync_owned(
-        partial(_load_constructor, provider_id, provider_loaders or {})
+        partial(_load_constructor, provider_id, provider_loaders or {}, settings)
     )
     return constructor(settings)
 
diff --git a/src/free_claude_code/runtime/application.py b/src/free_claude_code/runtime/application.py
index 627556a116..4b6c0abff2 100644
--- a/src/free_claude_code/runtime/application.py
+++ b/src/free_claude_code/runtime/application.py
@@ -27,6 +27,7 @@
 )
 from free_claude_code.application.model_metadata import ProviderModelRefreshResult
 from free_claude_code.application.ports import StopResult
+from free_claude_code.config.admin.custom_providers import CustomProviderMutation
 from free_claude_code.config.admin.persistence import (
     PreparedAdminUpdate,
 )
@@ -334,6 +335,7 @@ async def pick_folder(self, initial_path: str | None) -> str | None:
     async def apply_admin_config(
         self,
         updates: Mapping[str, ConfigInputValue],
+        custom_provider: CustomProviderMutation | None = None,
     ) -> JsonObject:
         """Apply one validated config update without splitting runtime ownership."""
         caller = asyncio.current_task()
@@ -344,7 +346,9 @@ async def apply_admin_config(
                 raise ApplicationUnavailableError(
                     "Configuration runtime is shutting down."
                 )
-            prepared = await self._configuration.prepare(updates, self.settings)
+            prepared = await self._configuration.prepare(
+                updates, self.settings, custom_provider
+            )
             if not prepared.valid:
                 return prepared.applied_response() | {"credential_checks": []}
             assert prepared.settings is not None
diff --git a/src/free_claude_code/runtime/configuration.py b/src/free_claude_code/runtime/configuration.py
index e0198e2f87..5592113a39 100644
--- a/src/free_claude_code/runtime/configuration.py
+++ b/src/free_claude_code/runtime/configuration.py
@@ -4,6 +4,7 @@
 
 from anyio import CapacityLimiter, to_thread
 
+from free_claude_code.config.admin.custom_providers import CustomProviderMutation
 from free_claude_code.config.admin.persistence import (
     PreparedAdminUpdate,
     prepare_admin_update,
@@ -37,12 +38,15 @@ async def admin_values(self) -> ValueState:
         return load_value_state(snapshot)
 
     async def prepare(
-        self, updates: Mapping[str, ConfigInputValue], active_settings: Settings
+        self,
+        updates: Mapping[str, ConfigInputValue],
+        active_settings: Settings,
+        custom_provider: CustomProviderMutation | None = None,
     ) -> PreparedAdminUpdate:
         snapshot = await to_thread.run_sync(
             self._store.read, limiter=self._worker_limiter
         )
-        return prepare_admin_update(updates, snapshot, active_settings)
+        return prepare_admin_update(updates, snapshot, active_settings, custom_provider)
 
     async def commit(self, prepared: PreparedAdminUpdate) -> JsonObject:
         if not prepared.valid:
diff --git a/src/free_claude_code/runtime/diagnostics.py b/src/free_claude_code/runtime/diagnostics.py
index 623078e25e..043a2b0867 100644
--- a/src/free_claude_code/runtime/diagnostics.py
+++ b/src/free_claude_code/runtime/diagnostics.py
@@ -135,7 +135,7 @@ def classified_settings() -> set[str]:
         set(RUNTIME_FIELDS + ROUTING_FIELDS + REASONING_FIELDS + WEB_FIELDS)
         | EXCLUDED_FIELDS
         | URL_FIELDS
-        | {"nim", "vertex_location"}
+        | {"nim", "vertex_location", "custom_providers"}
     )
 
 
@@ -179,6 +179,18 @@ def settings_report(settings: Settings, accounts: dict[str, str]) -> dict[str, A
         if name == "vertex":
             row["location"] = settings.vertex_location
         providers[name] = row
+    for definition in settings.custom_providers:
+        providers[definition.provider_id] = {
+            "display_name": definition.display_name,
+            "configured": True,
+            "referenced": definition.provider_id in referenced,
+            "base_url": _url_summary(definition.base_url),
+            "api_key_configured": bool(definition.api_key),
+            "api_format": definition.api_format,
+            "reasoning_format": definition.reasoning_format,
+            "reasoning_history_format": definition.reasoning_history_format,
+            "model_ids": list(definition.model_ids),
+        }
     runtime = {name: getattr(settings, name) for name in RUNTIME_FIELDS}
     runtime["telegram_proxy"] = _url_summary(settings.telegram_proxy_url)
     runtime["nim"] = {name: getattr(settings.nim, name) for name in NIM_FIELDS}
diff --git a/src/free_claude_code/runtime/provider_manager.py b/src/free_claude_code/runtime/provider_manager.py
index 076265bf69..71df8dbd3c 100644
--- a/src/free_claude_code/runtime/provider_manager.py
+++ b/src/free_claude_code/runtime/provider_manager.py
@@ -21,6 +21,7 @@
 from free_claude_code.core.token_estimation import initialize_token_estimation
 from free_claude_code.core.trace import trace_event
 from free_claude_code.providers.base import BaseProvider
+from free_claude_code.providers.model_listing import model_infos_from_ids
 from free_claude_code.providers.runtime.discovery import (
     _provider_query_failure_reason,
     model_cache_provider_ids_for_settings,
@@ -64,6 +65,12 @@ class _ProviderGeneration:
     file_state: str = "starting"
 
     def __post_init__(self) -> None:
+        for definition in self.settings.custom_providers:
+            if definition.model_ids:
+                self.cache.cache_model_infos(
+                    definition.provider_id, model_infos_from_ids(definition.model_ids)
+                )
+                self.initialized.add(definition.provider_id)
         self.drained.set()
 
     def snapshot(self) -> ModelCatalogSnapshot:
@@ -259,6 +266,10 @@ def _catalog_task(
     async def _discover_provider(
         self, generation: _ProviderGeneration, provider_id: str
     ) -> ProviderModelRefreshResult:
+        definition = generation.settings.custom_provider(provider_id)
+        if definition is not None and definition.model_ids:
+            generation.initialized.add(provider_id)
+            return ProviderModelRefreshResult(refreshed_provider_ids=(provider_id,))
         try:
             provider = await generation.runtime.resolve_provider(provider_id)
             infos = await provider.list_model_infos()
@@ -531,15 +542,21 @@ async def replace(
             assert candidate_runtime is not None
             async with self._publication_lock:
                 previous = self._current
+                cache = previous.cache.copy(
+                    model_cache_provider_ids_for_settings(
+                        settings, self._connected_provider_ids()
+                    )
+                )
+                for definition in settings.custom_providers:
+                    old = previous.settings.custom_provider(definition.provider_id)
+                    if old is None or not definition.same_inventory(old):
+                        cache.remove_provider(definition.provider_id)
+                        cache.add_provider(definition.provider_id)
                 candidate = _ProviderGeneration(
                     generation_id=candidate_id,
                     settings=settings,
                     runtime=candidate_runtime,
-                    cache=previous.cache.copy(
-                        model_cache_provider_ids_for_settings(
-                            settings, self._connected_provider_ids()
-                        )
-                    ),
+                    cache=cache,
                 )
                 self._current = candidate
                 self._catalog_revision += 1
diff --git a/tests/api/model_fallback_support.py b/tests/api/model_fallback_support.py
index 1f71b81f94..16bcb3e08d 100644
--- a/tests/api/model_fallback_support.py
+++ b/tests/api/model_fallback_support.py
@@ -238,6 +238,7 @@ async def stream_responses(
         response_model: str | None = None,
         reasoning: ReasoningPolicy,
         request_headers: Mapping[str, str] | None = None,
+        model_info: ProviderModelInfo | None = None,
     ) -> AsyncIterator[str]:
         del input_tokens, request_id, reasoning
         if self._validation_error is not None:
diff --git a/tests/api/test_admin.py b/tests/api/test_admin.py
index f14212a7bd..93e7ab9c0c 100644
--- a/tests/api/test_admin.py
+++ b/tests/api/test_admin.py
@@ -660,7 +660,6 @@ def test_admin_static_model_combobox_owns_dropdown_and_search_behavior():
     assert 'this.toggle.className = "model-combobox-toggle"' in combobox_script
     assert "class FccModelCombobox" in combobox_script
     assert 'input.addEventListener("click", () => this.open())' in combobox_script
-    assert "value.toLocaleLowerCase().includes(normalizedQuery)" in combobox_script
     assert 'event.key === "ArrowDown" || event.key === "ArrowUp"' in combobox_script
     assert "this.setActive(this.visibleOptions.length - 1)" in combobox_script
     assert 'event.key === "Enter"' in combobox_script
@@ -847,7 +846,9 @@ def test_admin_models_include_configured_and_cached_canonical_slugs():
     response = _local_client(app).get("/admin/api/models")
 
     assert response.status_code == 200
-    assert response.json() == {
+    assert {
+        key: value for key, value in response.json().items() if key != "model_labels"
+    } == {
         "models": [
             "nvidia_nim/configured-model",
             "open_router/anthropic/configured-opus",
@@ -876,7 +877,9 @@ async def refresh_models() -> ProviderModelRefreshResult:
     response = _local_client(app).post("/admin/api/models/refresh")
 
     assert response.status_code == 200
-    assert response.json() == {
+    assert {
+        key: value for key, value in response.json().items() if key != "model_labels"
+    } == {
         "models": ["deepseek/deepseek-chat", "deepseek/deepseek-reasoner"],
         "failed_providers": [],
     }
@@ -898,7 +901,9 @@ def test_admin_model_refresh_reports_partial_provider_failures():
     response = _local_client(app).post("/admin/api/models/refresh")
 
     assert response.status_code == 200
-    assert response.json() == {
+    assert {
+        key: value for key, value in response.json().items() if key != "model_labels"
+    } == {
         "models": ["deepseek/deepseek-chat"],
         "failed_providers": ["open_router"],
     }
diff --git a/tests/api/test_api_handlers.py b/tests/api/test_api_handlers.py
index 44919beeb1..836eb715fd 100644
--- a/tests/api/test_api_handlers.py
+++ b/tests/api/test_api_handlers.py
@@ -86,6 +86,7 @@ async def stream_responses(
         response_model: str | None = None,
         reasoning: ReasoningPolicy,
         request_headers: Mapping[str, str] | None = None,
+        model_info: ProviderModelInfo | None = None,
     ) -> AsyncIterator[str]:
         self.responses_requests.append(request)
         self.stream_kwargs.append(
diff --git a/tests/api/test_execution_failure_contract.py b/tests/api/test_execution_failure_contract.py
index 689720a50d..871d592680 100644
--- a/tests/api/test_execution_failure_contract.py
+++ b/tests/api/test_execution_failure_contract.py
@@ -127,6 +127,7 @@ async def stream_responses(
         response_model: str,
         reasoning: ReasoningPolicy,
         request_headers: Mapping[str, str] | None = None,
+        model_info: ProviderModelInfo | None = None,
     ) -> AsyncIterator[str]:
         del input_tokens, request_id, response_model, reasoning
         try:
diff --git a/tests/api/test_openai_responses.py b/tests/api/test_openai_responses.py
index 58b6467499..983231d7b3 100644
--- a/tests/api/test_openai_responses.py
+++ b/tests/api/test_openai_responses.py
@@ -2,10 +2,12 @@
 from collections.abc import AsyncIterator
 from unittest.mock import MagicMock, patch
 
+import httpx
 import pytest
 from fastapi.testclient import TestClient
 
 from free_claude_code.application.errors import InvalidRequestError
+from free_claude_code.application.model_metadata import ProviderModelInfo
 from free_claude_code.config.constants import DEFAULT_MODEL
 from free_claude_code.config.settings import Settings
 from free_claude_code.core.anthropic import ReasoningReplayMode
@@ -24,14 +26,49 @@
     OpenAIChatProvider,
     OpenAIChatRequestPolicy,
 )
-from tests.api.support import create_test_app
+from tests.api.support import create_test_app, provider_manager_for_app
 from tests.providers.support import immediate_admission, make_provider_config
+from tests.providers.test_custom_provider import completion, definition
+from tests.providers.test_custom_provider import provider as custom_provider
 
 _PUBLIC_MODEL = "nvidia_nim/test-model"
 _UPSTREAM_MODEL = "test-model"
 _RESPONSE_ID = "resp_test"
 
 
+def test_responses_route_carries_generation_model_limit_to_messages_egress():
+    bodies = []
+
+    def reply(request):
+        bodies.append(json.loads(request.content))
+        return httpx.Response(
+            200,
+            text=completion("anthropic_messages"),
+            headers={"content-type": "text/event-stream"},
+        )
+
+    config = definition(api_format="anthropic_messages")
+    instance = custom_provider(config, messages_handler=reply)
+    app = create_test_app(
+        Settings(custom_providers=(config,)), providers={config.provider_id: instance}
+    )
+    provider_manager_for_app(app).cache_model_infos(
+        config.provider_id, {ProviderModelInfo("m", max_output_tokens=4096)}
+    )
+    with TestClient(app) as client:
+        response = client.post(
+            "/v1/responses",
+            json={
+                "model": f"{config.provider_id}/m",
+                "input": "Hello",
+                "max_output_tokens": 8192,
+            },
+        )
+    assert response.status_code == 200
+    assert "response.completed" in response.text
+    assert bodies[0]["max_tokens"] == 4096
+
+
 class FakeProvider:
     def __init__(self, chunks: list[str]) -> None:
         self.chunks = chunks
diff --git a/tests/api/test_request_lifetime.py b/tests/api/test_request_lifetime.py
index 98019b4c60..31023983d5 100644
--- a/tests/api/test_request_lifetime.py
+++ b/tests/api/test_request_lifetime.py
@@ -430,6 +430,7 @@ async def stream_responses(
         response_model: str,
         reasoning: ReasoningPolicy,
         request_headers: Mapping[str, str] | None = None,
+        model_info: ProviderModelInfo | None = None,
     ) -> AsyncIterator[str]:
         del input_tokens, response_model, reasoning
         self.request_ids.append(request_id)
diff --git a/tests/api/test_web_server_tools.py b/tests/api/test_web_server_tools.py
index bb8dc44fbc..65b5510c39 100644
--- a/tests/api/test_web_server_tools.py
+++ b/tests/api/test_web_server_tools.py
@@ -184,6 +184,7 @@ async def stream_responses(
         response_model: str,
         reasoning: ReasoningPolicy,
         request_headers: Mapping[str, str] | None = None,
+        model_info: ProviderModelInfo | None = None,
     ) -> AsyncIterator[str]:
         raise AssertionError("Web-search selection received a Responses request")
         yield ""
diff --git a/tests/application/test_execution.py b/tests/application/test_execution.py
index d6fd1544ae..8426b33132 100644
--- a/tests/application/test_execution.py
+++ b/tests/application/test_execution.py
@@ -60,6 +60,7 @@ async def stream_responses(
         response_model: str,
         reasoning: ReasoningPolicy,
         request_headers: Mapping[str, str] | None = None,
+        model_info: ProviderModelInfo | None = None,
     ) -> AsyncIterator[str]:
         raise AssertionError("Messages test provider received a Responses request")
         yield ""
@@ -112,6 +113,7 @@ async def stream_responses(
         response_model: str,
         reasoning: ReasoningPolicy,
         request_headers: Mapping[str, str] | None = None,
+        model_info: ProviderModelInfo | None = None,
     ) -> AsyncIterator[str]:
         self.stream_calls.append(
             {
@@ -334,8 +336,13 @@ def _routed_request(
 
 
 @pytest.mark.asyncio
-async def test_fallback_receives_its_own_cached_capability_in_stream():
+@pytest.mark.parametrize("ingress", ["messages", "responses"])
+@pytest.mark.parametrize("unknown_fallback", [False, True])
+async def test_fallback_receives_its_own_cached_capability_in_stream(
+    ingress, unknown_fallback
+):
     seen = []
+    resolved = set()
 
     class MetadataProvider(ControlledProvider):
         async def stream_messages(self, request, input_tokens=0, **kwargs):
@@ -343,33 +350,45 @@ async def stream_messages(self, request, input_tokens=0, **kwargs):
             async for event in super().stream_messages(request, input_tokens, **kwargs):
                 yield event
 
+        stream_responses = stream_messages
+
     primary_info = ProviderModelInfo(
         "provider/provider-model", reasoning_capability=ReasoningCapability.NONE
     )
-    fallback_info = ProviderModelInfo(
-        "fallback/fallback-model", reasoning_capability=ReasoningCapability.REQUIRED
+    fallback_info = (
+        ProviderModelInfo(
+            "fallback/fallback-model", reasoning_capability=ReasoningCapability.REQUIRED
+        )
+        if not unknown_fallback
+        else None
     )
     primary = MetadataProvider(
         [ExecutionFailure(FailureKind.UNAVAILABLE, 503, "unavailable", True)]
     )
     fallback = MetadataProvider(["verdict"])
     providers = {"provider": primary, "fallback": fallback}
+
+    async def resolve(name):
+        resolved.add(name)
+        return providers[name]
+
+    def lookup(provider_id, model_id):
+        assert provider_id in resolved
+        return {
+            "provider/provider-model": primary_info,
+            "fallback/fallback-model": fallback_info,
+        }.get(f"{provider_id}/{model_id}")
+
     executor = ProviderExecutor(
-        AsyncMock(side_effect=lambda name: providers[name]),
+        resolve,
         progress_timeout_seconds=10,
-        model_info_lookup=lambda provider_id, model_id: next(
-            (
-                info
-                for info in (primary_info, fallback_info)
-                if info.model_id == f"{provider_id}/{model_id}"
-            ),
-            None,
-        ),
+        model_info_lookup=lookup,
     )
+    route = _routed_request if ingress == "messages" else _routed_responses_request
     output = [
         event
-        async for event in executor.stream_messages(
-            _routed_request(_target("fallback", "fallback-model")),
+        async for event in getattr(executor, f"stream_{ingress}")(
+            route(_target("fallback", "fallback-model")),
             raw_log_payload={},
             request_id="capability-fallback",
         )
diff --git a/tests/application/test_model_catalog.py b/tests/application/test_model_catalog.py
index 68386023ac..684842eef6 100644
--- a/tests/application/test_model_catalog.py
+++ b/tests/application/test_model_catalog.py
@@ -7,9 +7,45 @@
 )
 from free_claude_code.application.model_metadata import ProviderModelInfo
 from free_claude_code.application.ports import ModelCatalogSnapshot
+from free_claude_code.config.custom_providers import CustomProviderDefinition
 from free_claude_code.config.settings import Settings
 
 
+def test_custom_display_names_sort_as_labels_without_changing_model_identity():
+    names = ["Team/West", "/Gateway", "Team", "z"]
+    definitions = tuple(
+        CustomProviderDefinition(
+            provider_id=f"custom_{index:032x}",
+            display_name=name,
+            base_url="https://gateway.example/v1",
+        )
+        for index, name in enumerate(names)
+    )
+    settings = Settings(
+        custom_providers=definitions, model=f"{definitions[0].provider_id}/Z"
+    )
+    infos = tuple(
+        ProviderModelInfo(f"{definition.provider_id}/{model}")
+        for definition in definitions
+        for model in ("Z", "org/m")
+    )
+    snapshot = ModelCatalogSnapshot(settings, infos)
+    catalog = read_model_catalog(snapshot)
+    assert [model.display_name for model in catalog.models] == [
+        f"{name}/{model}"
+        for name in ("/Gateway", "Team", "Team/West", "z")
+        for model in ("org/m", "Z")
+    ]
+    assert {model.provider_model_ref for model in catalog.models} == {
+        info.model_id for info in infos
+    }
+    assert catalog.default_model_id == settings.model
+    assert (
+        read_model_catalog(replace(snapshot, model_infos=tuple(reversed(infos))))
+        == catalog
+    )
+
+
 def test_catalog_sorts_components_and_preserves_exact_identity_and_default():
     settings = Settings.model_construct(
         model="open_router/Zulu",
diff --git a/tests/config/test_custom_provider_mutations.py b/tests/config/test_custom_provider_mutations.py
new file mode 100644
index 0000000000..46fe9f20ff
--- /dev/null
+++ b/tests/config/test_custom_provider_mutations.py
@@ -0,0 +1,117 @@
+import json
+from pathlib import Path
+
+from free_claude_code.config.admin.custom_providers import CustomProviderMutation
+from free_claude_code.config.admin.persistence import prepare_admin_update
+from free_claude_code.config.admin.values import load_config_response
+from free_claude_code.config.custom_providers import CUSTOM_PROVIDERS_ENV
+from free_claude_code.config.loader import (
+    ManagedConfigSnapshot,
+    compose_settings_snapshot,
+)
+from free_claude_code.runtime.diagnostics import settings_report
+
+
+def snapshot(values=None, process=None):
+    values, process = values or {}, process or {}
+    result = compose_settings_snapshot(values, process)
+    return ManagedConfigSnapshot(
+        result.settings, result.sources, values, process, Path("unused.env")
+    )
+
+
+def mutate(state, action, provider_id=None, values=None, routes=None):
+    return prepare_admin_update(
+        routes or {},
+        state,
+        state.settings,
+        CustomProviderMutation(
+            action=action, provider_id=provider_id, values=values or {}
+        ),
+    )
+
+
+def created():
+    original = snapshot()
+    result = mutate(
+        original,
+        "create",
+        values={
+            "display_name": "Gateway",
+            "base_url": "https://example.org/private/path",
+            "api_key": "private-api-key",
+            "model_ids": ["one"],
+        },
+    )
+    assert result.valid
+    return snapshot(result.target_values), result.custom_provider_id
+
+
+def test_mutations_preserve_other_providers_secrets_and_routing_identity():
+    state, first = created()
+    second = mutate(
+        state,
+        "create",
+        values={"display_name": "Second", "base_url": "http://localhost:9999/api"},
+    )
+    state = snapshot(second.target_values)
+    edited = mutate(
+        state, "update", first, {"display_name": "Renamed", "api_key": "********"}
+    )
+    assert edited.valid and len(edited.settings.custom_providers) == 2
+    original = edited.settings.custom_provider(first)
+    assert original.display_name == "Renamed"
+    assert original.api_key.get_secret_value() == "private-api-key"
+    cleared = mutate(snapshot(edited.target_values), "update", first, {"api_key": None})
+    assert cleared.settings.custom_provider(first).api_key is None
+
+
+def test_deletion_requires_reassignment_and_is_idempotent():
+    state, provider_id = created()
+    selected = snapshot(state.managed | {"MODEL": f"{provider_id}/one"})
+    rejected = mutate(selected, "delete", provider_id)
+    assert not rejected.valid and "MODEL" in str(rejected.errors)
+    removed = mutate(
+        selected, "delete", provider_id, routes={"MODEL": state.settings.model}
+    )
+    assert removed.valid and not removed.settings.custom_providers
+    assert mutate(snapshot(removed.target_values), "delete", provider_id).valid
+    assert not mutate(
+        snapshot(removed.target_values),
+        "update",
+        provider_id,
+        {"display_name": "stale"},
+    ).valid
+
+
+def test_invalid_duplicate_or_locked_mutation_does_not_expose_keys():
+    state, provider_id = created()
+    duplicate = mutate(
+        state,
+        "create",
+        values={"display_name": "GATEWAY", "base_url": "http://localhost"},
+    )
+    assert not duplicate.valid
+    invalid = mutate(
+        state,
+        "update",
+        provider_id,
+        {"base_url": "https://user:private-api-key@example.org"},
+    )
+    assert not invalid.valid
+    assert "private-api-key" not in json.dumps(invalid.applied_response())
+    locked = snapshot(
+        state.managed, {CUSTOM_PROVIDERS_ENV: state.managed[CUSTOM_PROVIDERS_ENV]}
+    )
+    assert not mutate(locked, "delete", provider_id).valid
+    public = load_config_response(state)
+    rows = public["custom_providers"]
+    assert isinstance(rows, list) and isinstance(rows[0], dict)
+    assert rows[0]["api_key"] == "********"
+    assert "private-api-key" not in json.dumps(public)
+    doctor = json.dumps(settings_report(state.settings, {}))
+    assert (
+        "Gateway" in doctor
+        and "private-api-key" not in doctor
+        and "/private/path" not in doctor
+    )
diff --git a/tests/config/test_custom_providers.py b/tests/config/test_custom_providers.py
new file mode 100644
index 0000000000..e90b1115da
--- /dev/null
+++ b/tests/config/test_custom_providers.py
@@ -0,0 +1,78 @@
+import json
+
+import pytest
+from pydantic import ValidationError
+
+from free_claude_code.application.errors import UnknownProviderError
+from free_claude_code.application.routing import ModelRouter
+from free_claude_code.config.custom_providers import (
+    CustomProviderDefinition,
+    encode_custom_providers,
+)
+from free_claude_code.config.settings import Settings
+
+ID = "custom_" + "a" * 32
+
+
+def definition(**changes):
+    return CustomProviderDefinition.model_validate(
+        {
+            "provider_id": ID,
+            "display_name": "My Gateway",
+            "base_url": "https://gateway.example/api/v2/",
+            "api_key": "private-key",
+            **changes,
+        }
+    )
+
+
+def test_private_storage_round_trip_and_routing():
+    original = definition(model_ids=[" org/model ", "org/model"])
+    encoded = encode_custom_providers((original,))
+    settings = Settings.model_validate(
+        {"FCC_CUSTOM_PROVIDERS": encoded, "MODEL": f"{ID}/org/model"}
+    )
+    assert settings.custom_providers == (original,)
+    assert original.model_ids == ("org/model",)
+    assert original.base_url == "https://gateway.example/api/v2"
+    assert "private-key" not in repr(original)
+    assert "private-key" not in original.model_dump_json()
+    assert json.loads(encoded)[0]["api_key"] == "private-key"
+    assert ModelRouter(settings).resolve(settings.model).primary.provider_id == ID
+
+
+@pytest.mark.parametrize(
+    "url",
+    [
+        "ftp://example.org",
+        "https://u:p@example.org",
+        "https://example.org?key=secret",
+        "https://example.org/#fragment",
+    ],
+)
+def test_invalid_api_bases_are_rejected(url):
+    with pytest.raises(ValidationError):
+        definition(base_url=url)
+
+
+@pytest.mark.parametrize("prefix", ["", "anthropic/", "claude-3-freecc-no-thinking/"])
+def test_unknown_custom_reference_never_falls_back(prefix):
+    with pytest.raises(UnknownProviderError):
+        ModelRouter(Settings()).resolve(f"{prefix}{ID}/missing")
+
+
+def test_duplicate_names_and_missing_route_targets_are_rejected():
+    with pytest.raises(ValidationError):
+        Settings(
+            custom_providers=[
+                definition(),
+                definition(provider_id="custom_" + "b" * 32, display_name="MY GATEWAY"),
+            ]
+        )
+    with pytest.raises(ValidationError):
+        Settings(model=f"{ID}/missing")
+
+
+def test_format_and_reasoning_must_match():
+    with pytest.raises(ValidationError):
+        definition(api_format="anthropic_messages", reasoning_format="openai_effort")
diff --git a/tests/providers/test_anthropic_messages_transport.py b/tests/providers/test_anthropic_messages_transport.py
index a94685846a..32465702aa 100644
--- a/tests/providers/test_anthropic_messages_transport.py
+++ b/tests/providers/test_anthropic_messages_transport.py
@@ -8,6 +8,7 @@
 import pytest
 
 from free_claude_code.application.errors import InvalidRequestError
+from free_claude_code.application.model_metadata import ProviderModelInfo
 from free_claude_code.core.anthropic.models import MessagesRequest
 from free_claude_code.core.anthropic.stream_contracts import parse_sse_text
 from free_claude_code.core.failures import ExecutionFailure, FailureKind
@@ -15,6 +16,9 @@
 from free_claude_code.core.openai_responses import OpenAIResponsesRequest
 from free_claude_code.core.reasoning import ReasoningPolicy
 from free_claude_code.providers.admission import ProviderAdmissionController
+from free_claude_code.providers.anthropic_messages.request_policy import (
+    MessagesModelCapabilities,
+)
 from free_claude_code.providers.anthropic_messages.transport import (
     AnthropicMessagesTransport,
 )
@@ -45,7 +49,10 @@
     ],
     ids=["mandatory", "param", "loc"],
 )
-async def test_tolerant_classifier_corrects_required_reasoning(stream_error, error):
+@pytest.mark.parametrize("cap", [None, 4096])
+async def test_tolerant_classifier_corrects_required_reasoning(
+    stream_error, error, cap
+):
     bodies = []
 
     def handler(request):
@@ -79,11 +86,12 @@ def handler(request):
                 ),
                 endpoint_context=Endpoint(),
                 reasoning=ReasoningPolicy.prefer_off(),
+                model_info=ProviderModelInfo("route", max_output_tokens=cap),
             )
         ]
     assert "<severity>0</severity>" in "".join(output)
     assert len(bodies) == 2
-    assert bodies[0]["max_tokens"] == 8192
+    assert [body["max_tokens"] for body in bodies] == [cap or 8192, cap or 8192]
     assert "thinking" not in bodies[1]
 
 
@@ -243,7 +251,9 @@ def _events(text: str = "hello", stop: str = "end_turn") -> list[JsonObject]:
 
 
 def _transport(
-    client: httpx.AsyncClient, admission: ProviderAdmissionController | None = None
+    client: httpx.AsyncClient,
+    admission: ProviderAdmissionController | None = None,
+    capabilities: MessagesModelCapabilities = MessagesModelCapabilities(),
 ) -> AnthropicMessagesTransport:
     return AnthropicMessagesTransport(
         client=client,
@@ -251,9 +261,46 @@ def _transport(
         provider_name="TEST",
         replay_scope="test/messages",
         read_timeout_s=3,
+        capabilities=capabilities,
     )
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "profile_cap,model_cap,expected",
+    [(2048, 4096, 2048), (4096, 2048, 2048), (4096, None, 4096)],
+)
+async def test_model_metadata_cannot_relax_declared_profile_cap(
+    profile_cap, model_cap, expected
+):
+    bodies = []
+
+    def reply(request):
+        bodies.append(json.loads(request.content))
+        return httpx.Response(
+            200,
+            headers={"content-type": "text/event-stream"},
+            stream=Wire([_sse(*_events())]),
+        )
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(reply)) as client:
+        transport = _transport(
+            client,
+            capabilities=MessagesModelCapabilities(max_output_tokens=profile_cap),
+        )
+        assert [
+            item
+            async for item in transport.stream_responses(
+                OpenAIResponsesRequest(
+                    model="native", input="hi", max_output_tokens=8192
+                ),
+                endpoint_context=Endpoint(),
+                model_info=ProviderModelInfo("native", max_output_tokens=model_cap),
+            )
+        ]
+    assert bodies[0]["max_tokens"] == expected
+
+
 def _stream(
     transport: AnthropicMessagesTransport, endpoint: Endpoint, *, responses: bool
 ) -> AsyncIterator[str]:
@@ -308,7 +355,7 @@ def handler(request: httpx.Request) -> httpx.Response:
         )
         assert not client.is_closed
     assert wire.closed and len(requests) == 1
-    assert requests[0].url.path == "/v1/messages"
+    assert requests[0].url.path == ("/v1/messages" if versioned_base else "/messages")
     assert requests[0].headers["Authorization"] == "Bearer original"
     assert requests[0].headers["anthropic-version"] == "2023-06-01"
     assert requests[0].headers.get_list("anthropic-version") == ["2023-06-01"]
diff --git a/tests/providers/test_custom_provider.py b/tests/providers/test_custom_provider.py
new file mode 100644
index 0000000000..78ba6c94dd
--- /dev/null
+++ b/tests/providers/test_custom_provider.py
@@ -0,0 +1,718 @@
+import asyncio
+import json
+
+import httpx
+import httpx2
+import pytest
+
+from free_claude_code.application.errors import InvalidRequestError
+from free_claude_code.application.model_metadata import ProviderModelInfo
+from free_claude_code.config.custom_providers import CustomProviderDefinition
+from free_claude_code.config.settings import Settings
+from free_claude_code.core.anthropic import MessagesRequest
+from free_claude_code.core.model_capabilities import ModelInputModality
+from free_claude_code.core.openai_responses import OpenAIResponsesRequest
+from free_claude_code.core.reasoning import ReasoningEffort, ReasoningPolicy
+from free_claude_code.providers.admission import ProviderAdmissionController
+from free_claude_code.providers.custom import CustomProvider
+from free_claude_code.providers.runtime.config import build_custom_provider_config
+
+pytestmark = pytest.mark.asyncio
+
+
+@pytest.mark.parametrize(
+    "thinking",
+    [
+        None,
+        {"type": "adaptive", "display": "omitted"},
+        {"type": "enabled", "budget_tokens": 1536},
+    ],
+)
+async def test_native_default_preserves_controls_without_advertised_support(thinking):
+    bodies = []
+
+    def reply(request):
+        bodies.append(json.loads(request.content))
+        return httpx.Response(
+            200,
+            text=completion("anthropic_messages"),
+            headers={"content-type": "text/event-stream"},
+        )
+
+    instance = provider(
+        definition(api_format="anthropic_messages"), messages_handler=reply
+    )
+    request = MessagesRequest(
+        model="m",
+        max_tokens=4096,
+        messages=[{"role": "user", "content": "Hello"}],
+        thinking=thinking,
+        output_config={
+            "effort": "high",
+            "format": {"type": "json_schema", "schema": {"type": "object"}},
+        },
+    )
+    before = request.model_dump()
+    try:
+        assert [
+            item
+            async for item in instance.stream_messages(
+                request, reasoning=ReasoningPolicy.off()
+            )
+        ]
+        assert bodies[0].get("thinking") == thinking
+        assert bodies[0]["output_config"] == request.output_config
+        assert request.model_dump() == before
+    finally:
+        await instance.cleanup()
+
+
+@pytest.mark.parametrize("ingress", ["messages", "responses"])
+@pytest.mark.parametrize(
+    "limit,cap,expected",
+    [(2048, 4096, 2048), (8192, 4096, 4096), (None, 4096, 4096), (8192, None, 8192)],
+)
+async def test_messages_egress_obeys_request_model_limit(ingress, limit, cap, expected):
+    bodies = []
+
+    def reply(request):
+        bodies.append(json.loads(request.content))
+        return httpx.Response(
+            200,
+            text=completion("anthropic_messages"),
+            headers={"content-type": "text/event-stream"},
+        )
+
+    instance = provider(
+        definition(api_format="anthropic_messages"), messages_handler=reply
+    )
+    info = ProviderModelInfo("m", max_output_tokens=cap) if cap else None
+    try:
+        if ingress == "messages":
+            request = MessagesRequest(
+                model="m",
+                max_tokens=limit,
+                messages=[{"role": "user", "content": "Hello"}],
+            )
+            stream = instance.stream_messages(request, model_info=info)
+        else:
+            request = OpenAIResponsesRequest(
+                model="m",
+                max_output_tokens=limit,
+                input="Hello",
+                reasoning={"effort": "high"},
+            )
+            stream = instance.stream_responses(request, model_info=info)
+        before = request.model_dump()
+        assert [item async for item in stream]
+        assert bodies[0]["max_tokens"] == expected
+        assert "thinking" not in bodies[0]
+        assert "effort" not in bodies[0].get("output_config", {})
+        assert request.model_dump() == before
+    finally:
+        await instance.cleanup()
+
+
+@pytest.mark.parametrize("ingress", ["messages", "responses"])
+@pytest.mark.parametrize("exact", [False, True])
+@pytest.mark.parametrize("cap", [1024, 4096])
+async def test_manual_preset_respects_cap_before_resolving_budget(ingress, exact, cap):
+    bodies = []
+
+    def reply(request):
+        bodies.append(json.loads(request.content))
+        return httpx.Response(
+            200,
+            text=completion("anthropic_messages"),
+            headers={"content-type": "text/event-stream"},
+        )
+
+    instance = provider(
+        definition(api_format="anthropic_messages", reasoning_format="messages_manual"),
+        messages_handler=reply,
+    )
+    policy = (
+        ReasoningPolicy.on(budget_tokens=4096)
+        if exact
+        else ReasoningPolicy.on(effort=ReasoningEffort.HIGH)
+    )
+    info = ProviderModelInfo("m", max_output_tokens=cap)
+
+    async def consume():
+        if ingress == "messages":
+            stream = instance.stream_messages(
+                MessagesRequest(
+                    model="m",
+                    max_tokens=8192,
+                    messages=[{"role": "user", "content": "Hello"}],
+                ),
+                reasoning=policy,
+                model_info=info,
+            )
+        else:
+            stream = instance.stream_responses(
+                OpenAIResponsesRequest(
+                    model="m", max_output_tokens=8192, input="Hello"
+                ),
+                reasoning=policy,
+                model_info=info,
+            )
+        return [item async for item in stream]
+
+    try:
+        if exact or cap == 1024:
+            with pytest.raises(InvalidRequestError, match="below max_tokens"):
+                await consume()
+            assert not bodies
+        else:
+            assert await consume()
+            assert bodies[0]["max_tokens"] == cap
+            assert 1024 <= bodies[0]["thinking"]["budget_tokens"] < cap
+    finally:
+        await instance.cleanup()
+
+
+@pytest.mark.parametrize(
+    "thinking,error",
+    [
+        ({"type": "enabled"}, "budget"),
+        ({"type": "enabled", "budget_tokens": 6000}, "below max_tokens"),
+    ],
+)
+async def test_native_default_does_not_invent_or_shrink_exact_budget(thinking, error):
+    bodies = []
+    instance = provider(
+        definition(api_format="anthropic_messages"),
+        messages_handler=lambda request: bodies.append(request),
+    )
+    try:
+        with pytest.raises(InvalidRequestError, match=error):
+            stream = instance.stream_messages(
+                MessagesRequest(
+                    model="m",
+                    max_tokens=8192,
+                    thinking=thinking,
+                    messages=[{"role": "user", "content": "Hello"}],
+                ),
+                model_info=ProviderModelInfo("m", max_output_tokens=4096),
+            )
+            [item async for item in stream]
+        assert not bodies
+    finally:
+        await instance.cleanup()
+
+
+async def test_concurrent_models_keep_independent_limits():
+    bodies = []
+    both_started = asyncio.Event()
+
+    async def reply(request):
+        bodies.append(json.loads(request.content))
+        if len(bodies) == 2:
+            both_started.set()
+        await asyncio.wait_for(both_started.wait(), timeout=2)
+        return httpx.Response(
+            200,
+            text=completion("anthropic_messages"),
+            headers={"content-type": "text/event-stream"},
+        )
+
+    instance = provider(
+        definition(api_format="anthropic_messages"), messages_handler=reply
+    )
+
+    async def call(model, cap):
+        stream = instance.stream_messages(
+            MessagesRequest(
+                model=model,
+                max_tokens=8192,
+                messages=[{"role": "user", "content": "Hello"}],
+            ),
+            model_info=ProviderModelInfo(model, max_output_tokens=cap),
+        )
+        return [item async for item in stream]
+
+    try:
+        await asyncio.gather(call("small", 2048), call("large", 6000))
+        assert {body["model"]: body["max_tokens"] for body in bodies} == {
+            "small": 2048,
+            "large": 6000,
+        }
+    finally:
+        await instance.cleanup()
+
+
+def definition(**changes):
+    return CustomProviderDefinition.model_validate(
+        {
+            "provider_id": "custom_" + "a" * 32,
+            "display_name": "Gateway",
+            "base_url": "https://custom.example/team/api",
+            **changes,
+        }
+    )
+
+
+def provider(config, *, openai_handler=None, messages_handler=None):
+    settings = Settings(provider_rate_limit=1000)
+    return CustomProvider(
+        build_custom_provider_config(config, settings),
+        definition=config,
+        admission=ProviderAdmissionController(
+            provider_name=config.provider_id,
+            rate_limit=1000,
+            rate_window=1,
+            max_concurrency=2,
+        ),
+        openai_transport=httpx2.MockTransport(openai_handler)
+        if openai_handler
+        else None,
+        messages_transport=httpx.MockTransport(messages_handler)
+        if messages_handler
+        else None,
+    )
+
+
+def event(name, data):
+    return f"event: {name}\ndata: {json.dumps(data)}\n\n"
+
+
+def completion(api_format):
+    if api_format == "openai_chat":
+        data = {
+            "id": "chat",
+            "object": "chat.completion.chunk",
+            "created": 0,
+            "model": "m",
+            "choices": [
+                {"index": 0, "delta": {"content": "Hello"}, "finish_reason": "stop"}
+            ],
+        }
+        return f"data: {json.dumps(data)}\n\ndata: [DONE]\n\n"
+    if api_format == "openai_responses":
+        message = {
+            "id": "msg",
+            "type": "message",
+            "role": "assistant",
+            "status": "completed",
+            "content": [{"type": "output_text", "text": "Hello", "annotations": []}],
+        }
+        return (
+            event(
+                "response.output_item.added",
+                {
+                    "type": "response.output_item.added",
+                    "output_index": 0,
+                    "item": {**message, "content": []},
+                },
+            )
+            + event(
+                "response.output_text.delta",
+                {
+                    "type": "response.output_text.delta",
+                    "item_id": "msg",
+                    "output_index": 0,
+                    "content_index": 0,
+                    "delta": "Hello",
+                },
+            )
+            + event(
+                "response.completed",
+                {
+                    "type": "response.completed",
+                    "response": {
+                        "id": "resp",
+                        "object": "response",
+                        "created_at": 0,
+                        "status": "completed",
+                        "model": "m",
+                        "output": [message],
+                        "usage": {
+                            "input_tokens": 1,
+                            "output_tokens": 1,
+                            "total_tokens": 2,
+                        },
+                    },
+                },
+            )
+        )
+    return "".join(
+        [
+            event(
+                "message_start",
+                {
+                    "type": "message_start",
+                    "message": {
+                        "id": "msg",
+                        "type": "message",
+                        "role": "assistant",
+                        "model": "m",
+                        "content": [],
+                        "stop_reason": None,
+                        "usage": {"input_tokens": 1, "output_tokens": 0},
+                    },
+                },
+            ),
+            event(
+                "content_block_start",
+                {
+                    "type": "content_block_start",
+                    "index": 0,
+                    "content_block": {"type": "text", "text": ""},
+                },
+            ),
+            event(
+                "content_block_delta",
+                {
+                    "type": "content_block_delta",
+                    "index": 0,
+                    "delta": {"type": "text_delta", "text": "Hello"},
+                },
+            ),
+            event("content_block_stop", {"type": "content_block_stop", "index": 0}),
+            event(
+                "message_delta",
+                {
+                    "type": "message_delta",
+                    "delta": {"stop_reason": "end_turn"},
+                    "usage": {"output_tokens": 1},
+                },
+            ),
+            event("message_stop", {"type": "message_stop"}),
+        ]
+    )
+
+
+@pytest.mark.parametrize(
+    "api_format,path",
+    [
+        ("openai_chat", "chat/completions"),
+        ("openai_responses", "responses"),
+        ("anthropic_messages", "messages"),
+    ],
+)
+@pytest.mark.parametrize("ingress", ["messages", "responses"])
+@pytest.mark.parametrize("key", [None, "custom-key"])
+async def test_six_paths_use_exact_endpoint_and_credentials(
+    api_format, path, ingress, key, monkeypatch
+):
+    monkeypatch.setenv("OPENAI_API_KEY", "unrelated-key")
+    monkeypatch.setenv("OPENAI_ORG_ID", "unrelated-org")
+    monkeypatch.setenv("OPENAI_PROJECT_ID", "unrelated-project")
+    requests = []
+
+    def reply(request):
+        requests.append(request)
+        cls = httpx.Response if api_format == "anthropic_messages" else httpx2.Response
+        return cls(
+            200,
+            text=completion(api_format),
+            headers={"content-type": "text/event-stream"},
+        )
+
+    instance = provider(
+        definition(api_format=api_format, api_key=key),
+        openai_handler=reply if api_format != "anthropic_messages" else None,
+        messages_handler=reply if api_format == "anthropic_messages" else None,
+    )
+    try:
+        policy = ReasoningPolicy.on(effort=ReasoningEffort.MEDIUM)
+        if ingress == "messages":
+            request = MessagesRequest(
+                model="m",
+                max_tokens=100,
+                messages=[{"role": "user", "content": "Hello"}],
+            )
+            stream = instance.stream_messages(request, reasoning=policy)
+        else:
+            request = OpenAIResponsesRequest(
+                model="m", input="Hello", reasoning={"effort": "medium"}
+            )
+            stream = instance.stream_responses(request, reasoning=policy)
+        output = "".join([chunk async for chunk in stream])
+        assert "Hello" in output
+        assert len(requests) == 1
+        sent = requests[0]
+        assert str(sent.url) == f"https://custom.example/team/api/{path}"
+        assert sent.headers.get("x-api-key") == (
+            key if api_format == "anthropic_messages" else None
+        )
+        assert sent.headers.get("authorization") == (
+            f"Bearer {key}" if key and api_format != "anthropic_messages" else None
+        )
+        assert "openai-organization" not in sent.headers
+        assert "openai-project" not in sent.headers
+        body = json.loads(sent.content)
+        if ingress == "responses" and api_format == "openai_responses":
+            assert body["reasoning"]["effort"] == "medium"
+        else:
+            assert "reasoning_effort" not in body
+            assert not body.get("reasoning")
+            assert "thinking" not in body
+        if ingress == "responses":
+            assert isinstance(request, OpenAIResponsesRequest)
+            assert request.reasoning == {"effort": "medium"}
+    finally:
+        await instance.cleanup()
+
+
+@pytest.mark.parametrize(
+    "api_format", ["openai_chat", "openai_responses", "anthropic_messages"]
+)
+async def test_discovery_is_bound_and_manual_inventory_needs_no_network(
+    api_format, monkeypatch
+):
+    monkeypatch.setenv("OPENAI_API_KEY", "do-not-send")
+    requests = []
+
+    def reply(request):
+        requests.append(request)
+        cls = httpx.Response if api_format == "anthropic_messages" else httpx2.Response
+        return cls(200, json={"data": [{"id": "org/model"}], "has_more": False})
+
+    instance = provider(
+        definition(api_format=api_format),
+        openai_handler=reply if api_format != "anthropic_messages" else None,
+        messages_handler=reply if api_format == "anthropic_messages" else None,
+    )
+    try:
+        assert {info.model_id for info in await instance.list_model_infos()} == {
+            "org/model"
+        }
+        assert requests[0].url.path == "/team/api/models"
+        assert "authorization" not in requests[0].headers
+        assert "x-api-key" not in requests[0].headers
+    finally:
+        await instance.cleanup()
+    requests.clear()
+    manual = provider(
+        definition(api_format=api_format, model_ids=["manual"]),
+        openai_handler=reply if api_format != "anthropic_messages" else None,
+        messages_handler=reply if api_format == "anthropic_messages" else None,
+    )
+    try:
+        assert {info.model_id for info in await manual.list_model_infos()} == {"manual"}
+        assert not requests
+    finally:
+        await manual.cleanup()
+
+
+@pytest.mark.parametrize(
+    "api_format,mode,field,expected",
+    [
+        ("openai_chat", "openai_effort", "reasoning_effort", "high"),
+        ("openai_chat", "limited_effort", "reasoning_effort", "high"),
+        ("openai_chat", "reasoning_object", "reasoning", {"effort": "high"}),
+        ("openai_chat", "thinking", "thinking", {"type": "enabled"}),
+        (
+            "openai_chat",
+            "chat_template",
+            "chat_template_kwargs",
+            {"enable_thinking": True},
+        ),
+        (
+            "openai_responses",
+            "native_responses",
+            "reasoning",
+            {"effort": "high", "summary": "auto"},
+        ),
+        (
+            "anthropic_messages",
+            "messages_manual",
+            "thinking",
+            {"type": "enabled", "budget_tokens": 2048},
+        ),
+        ("anthropic_messages", "messages_adaptive", "thinking", {"type": "adaptive"}),
+    ],
+)
+async def test_reasoning_presets_encode_controls_on_the_wire(
+    api_format, mode, field, expected
+):
+    bodies = []
+
+    def reply(request):
+        bodies.append(json.loads(request.content))
+        cls = httpx.Response if api_format == "anthropic_messages" else httpx2.Response
+        return cls(
+            200,
+            text=completion(api_format),
+            headers={"content-type": "text/event-stream"},
+        )
+
+    instance = provider(
+        definition(api_format=api_format, reasoning_format=mode),
+        openai_handler=reply if api_format != "anthropic_messages" else None,
+        messages_handler=reply if api_format == "anthropic_messages" else None,
+    )
+    try:
+        request = MessagesRequest(
+            model="m", max_tokens=4096, messages=[{"role": "user", "content": "Hello"}]
+        )
+        assert [
+            chunk
+            async for chunk in instance.stream_messages(
+                request, reasoning=ReasoningPolicy.on(effort=ReasoningEffort.HIGH)
+            )
+        ]
+        assert bodies[0][field] == expected
+    finally:
+        await instance.cleanup()
+
+
+@pytest.mark.parametrize(
+    "api_format", ["openai_chat", "openai_responses", "anthropic_messages"]
+)
+@pytest.mark.parametrize("ingress", ["messages", "responses"])
+async def test_tool_continuation_keeps_tool_results_and_images(api_format, ingress):
+    bodies = []
+
+    def reply(request):
+        bodies.append(json.loads(request.content))
+        cls = httpx.Response if api_format == "anthropic_messages" else httpx2.Response
+        return cls(
+            200,
+            text=completion(api_format),
+            headers={"content-type": "text/event-stream"},
+        )
+
+    instance = provider(
+        definition(api_format=api_format),
+        openai_handler=reply if api_format != "anthropic_messages" else None,
+        messages_handler=reply if api_format == "anthropic_messages" else None,
+    )
+    try:
+        if ingress == "messages":
+            request = MessagesRequest(
+                model="m",
+                max_tokens=100,
+                tools=[
+                    {
+                        "name": "Read",
+                        "input_schema": {"type": "object", "properties": {}},
+                    }
+                ],
+                messages=[
+                    {
+                        "role": "assistant",
+                        "content": [
+                            {
+                                "type": "tool_use",
+                                "id": "call_read",
+                                "name": "Read",
+                                "input": {},
+                            }
+                        ],
+                    },
+                    {
+                        "role": "user",
+                        "content": [
+                            {
+                                "type": "tool_result",
+                                "tool_use_id": "call_read",
+                                "content": [
+                                    {"type": "text", "text": "tool-result-token"},
+                                    {
+                                        "type": "image",
+                                        "source": {
+                                            "type": "base64",
+                                            "media_type": "image/png",
+                                            "data": "aGVsbG8=",
+                                        },
+                                    },
+                                ],
+                            }
+                        ],
+                    },
+                ],
+            )
+            chunks = instance.stream_messages(request)
+        else:
+            request = OpenAIResponsesRequest(
+                model="m",
+                tools=[
+                    {
+                        "type": "function",
+                        "name": "Read",
+                        "parameters": {"type": "object", "properties": {}},
+                    }
+                ],
+                input=[
+                    {
+                        "type": "function_call",
+                        "call_id": "call_read",
+                        "name": "Read",
+                        "arguments": "{}",
+                    },
+                    {
+                        "type": "function_call_output",
+                        "call_id": "call_read",
+                        "output": [
+                            {"type": "input_text", "text": "tool-result-token"},
+                            {
+                                "type": "input_image",
+                                "image_url": "data:image/png;base64,aGVsbG8=",
+                            },
+                        ],
+                    },
+                ],
+            )
+            chunks = instance.stream_responses(request)
+        assert [chunk async for chunk in chunks]
+        body = json.dumps(bodies[0])
+        assert (
+            "tool-result-token" in body and "aGVsbG8=" in body and "call_read" in body
+        )
+    finally:
+        await instance.cleanup()
+
+
+async def test_messages_discovery_rejects_repeating_cursor_without_partial_results():
+    calls = []
+
+    def reply(request):
+        calls.append(request)
+        return httpx.Response(
+            200, json={"data": [{"id": "same"}], "has_more": True, "last_id": "same"}
+        )
+
+    instance = provider(
+        definition(api_format="anthropic_messages"), messages_handler=reply
+    )
+    try:
+        with pytest.raises(ValueError, match="cursor did not advance"):
+            await instance.list_model_infos()
+        assert len(calls) == 2
+        assert calls[1].url.params["after_id"] == "same"
+    finally:
+        await instance.cleanup()
+
+
+@pytest.mark.parametrize("vision", [True, False])
+async def test_messages_discovery_retains_text_and_advertised_limits(vision):
+    instance = provider(
+        definition(api_format="anthropic_messages"),
+        messages_handler=lambda request: httpx.Response(
+            200,
+            json={
+                "data": [
+                    {
+                        "id": "model",
+                        "max_input_tokens": 200000,
+                        "max_tokens": 32000,
+                        "capabilities": {"image_input": {"supported": vision}},
+                    }
+                ],
+                "has_more": False,
+            },
+        ),
+    )
+    try:
+        (info,) = await instance.list_model_infos()
+        expected = {ModelInputModality.TEXT}
+        if vision:
+            expected.add(ModelInputModality.IMAGE)
+        assert info.input_modalities == frozenset(expected)
+        assert info.context_window_tokens == 200000
+        assert info.max_output_tokens == 32000
+    finally:
+        await instance.cleanup()
diff --git a/tests/providers/test_messages_request_policy.py b/tests/providers/test_messages_request_policy.py
index 5914f0d9c1..fda7375bb2 100644
--- a/tests/providers/test_messages_request_policy.py
+++ b/tests/providers/test_messages_request_policy.py
@@ -131,7 +131,10 @@ def test_manual_effort_uses_budget_and_minimum() -> None:
 
 
 @pytest.mark.parametrize("raw", ["bogus", 1, {"mode": "high"}])
-def test_client_on_cannot_hide_invalid_output_effort(raw: object) -> None:
+@pytest.mark.parametrize("preserve_native_controls", [False, True])
+def test_client_on_cannot_hide_invalid_output_effort(
+    raw: object, preserve_native_controls: bool
+) -> None:
     with pytest.raises(NativeMessagesError, match="client output effort"):
         resolve_messages_options(
             model="m",
@@ -139,6 +142,7 @@ def test_client_on_cannot_hide_invalid_output_effort(raw: object) -> None:
             output_effort=raw,
             reasoning=ReasoningPolicy.on(budget_tokens=1536),
             thinking=ThinkingConfig(type="enabled", budget_tokens=1536),
+            preserve_native_controls=preserve_native_controls,
         )
 
 
diff --git a/tests/providers/test_model_discovery.py b/tests/providers/test_model_discovery.py
index 5e958e9d9e..3a6565386c 100644
--- a/tests/providers/test_model_discovery.py
+++ b/tests/providers/test_model_discovery.py
@@ -393,6 +393,7 @@ async def stream_responses(
         response_model: str | None = None,
         reasoning: ReasoningPolicy = DEFAULT_REASONING_POLICY,
         request_headers: Mapping[str, str] | None = None,
+        model_info: ProviderModelInfo | None = None,
     ) -> AsyncIterator[str]:
         if False:
             yield ""
diff --git a/tests/runtime/test_application_runtime.py b/tests/runtime/test_application_runtime.py
index ff03abc32f..31b1d051cb 100644
--- a/tests/runtime/test_application_runtime.py
+++ b/tests/runtime/test_application_runtime.py
@@ -96,6 +96,7 @@ async def stream_responses(
         response_model: str | None = None,
         reasoning: ReasoningPolicy = DEFAULT_REASONING_POLICY,
         request_headers: Mapping[str, str] | None = None,
+        model_info: ProviderModelInfo | None = None,
     ) -> AsyncIterator[str]:
         if False:
             yield ""
diff --git a/tests/runtime/test_custom_provider_lifecycle.py b/tests/runtime/test_custom_provider_lifecycle.py
new file mode 100644
index 0000000000..96b7835200
--- /dev/null
+++ b/tests/runtime/test_custom_provider_lifecycle.py
@@ -0,0 +1,95 @@
+import asyncio
+
+import pytest
+
+from free_claude_code.application.model_catalog import read_model_catalog
+from free_claude_code.application.model_metadata import ProviderModelInfo
+from free_claude_code.config.custom_providers import CustomProviderDefinition
+from free_claude_code.config.settings import Settings
+from free_claude_code.providers.custom import CustomProvider
+from free_claude_code.providers.runtime import ProviderRuntime
+from free_claude_code.runtime.provider_manager import ProviderRuntimeManager
+from tests.runtime.test_provider_manager import FakeRuntime, RuntimeFactory
+
+pytestmark = pytest.mark.asyncio
+ID = "custom_" + "a" * 32
+
+
+def settings(**changes):
+    definition = CustomProviderDefinition.model_validate(
+        {
+            "provider_id": ID,
+            "display_name": "Gateway",
+            "base_url": "https://example.org/v1",
+            **changes,
+        }
+    )
+    return Settings(custom_providers=(definition,))
+
+
+async def commit():
+    pass
+
+
+async def test_runtime_constructs_one_custom_owner_lazily():
+    runtime = ProviderRuntime(settings(model_ids=["manual"]))
+    try:
+        assert not runtime.is_cached(ID)
+        one, two = await asyncio.gather(
+            runtime.resolve_provider(ID), runtime.resolve_provider(ID)
+        )
+        assert one is two and isinstance(one, CustomProvider)
+        assert {info.model_id for info in await one.list_model_infos()} == {"manual"}
+    finally:
+        await runtime.cleanup()
+
+
+async def test_manual_inventory_is_immediate_and_does_not_construct_client():
+    factory = RuntimeFactory()
+    manager = ProviderRuntimeManager(
+        settings(model_ids=["manual"]), runtime_factory=factory
+    )
+    try:
+        catalog = read_model_catalog(manager)
+        assert any(
+            item.provider_model_ref == f"{ID}/manual"
+            and item.display_name == "Gateway/manual"
+            for item in catalog.models
+        )
+        await manager.refresh_provider(ID)
+        factory.runtimes[0].provider.list_model_infos.assert_not_called()
+    finally:
+        await manager.close()
+
+
+async def test_changed_endpoint_cannot_inherit_catalog_but_rename_can():
+    factory = RuntimeFactory()
+    release = asyncio.Event()
+
+    async def wait():
+        await release.wait()
+        return frozenset()
+
+    def blocked(settings):
+        runtime = factory(settings)
+        assert isinstance(runtime, FakeRuntime)
+        runtime.provider.list_model_infos.side_effect = wait
+        return runtime
+
+    manager = ProviderRuntimeManager(settings(), runtime_factory=blocked)
+    manager.cache_model_infos(ID, [ProviderModelInfo("old")])
+    try:
+        await manager.replace(settings(display_name="Renamed"), commit=commit)
+        assert manager.cached_model_info(ID, "old") is not None
+        lease = await manager.acquire()
+        await manager.replace(
+            settings(base_url="https://other.example/v2"), commit=commit
+        )
+        assert manager.cached_model_info(ID, "old") is None
+        retained = lease.settings.custom_provider(ID)
+        assert retained is not None and retained.base_url == "https://example.org/v1"
+        await lease.release()
+        assert factory.runtimes[-2].cleanup_calls == 1
+    finally:
+        release.set()
+        await manager.close()
diff --git a/tests/runtime/test_provider_manager.py b/tests/runtime/test_provider_manager.py
index f336e24ce3..616b977d7c 100644
--- a/tests/runtime/test_provider_manager.py
+++ b/tests/runtime/test_provider_manager.py
@@ -815,16 +815,25 @@ async def test_generation_lease_keeps_its_model_metadata_after_replacement() ->
     await lease.resolve_provider("open_router")
 
     await manager.replace(
-        _settings("nvidia_nim/two"),
+        first_settings,
         commit=AsyncMock(),
     )
 
+    factory.runtimes[-1].provider.list_model_infos.return_value = frozenset(
+        {ProviderModelInfo("old-model", max_output_tokens=16384)}
+    )
+    new_lease = await manager.acquire()
+    await new_lease.resolve_provider("open_router")
+    new_info = new_lease.model_info("open_router", "old-model")
+    assert new_info is not None and new_info.max_output_tokens == 16384
+
     assert lease.model_info("open_router", "old-model") == ProviderModelInfo(
         "open_router/old-model",
         context_window_tokens=32_000,
         max_output_tokens=4_096,
     )
     await lease.release()
+    await new_lease.release()
     await manager.close()