#1890·agent-zero

Bug Fix Made for OpenCode-go X-Header Patch Update

Author: jamiedukCreated Sep 8, 2026Updated Sep 8, 2026

gedit /tmp/model_config.py

import os
from copy import deepcopy

import models
from helpers import defer, plugins, files
from helpers.extension import call_extensions_async
from helpers import yaml as yaml_helper
from helpers.providers import get_provider_config, get_providers

PRESETS_FILE="presets.yaml"
FALLBACK_PRESETS_FILE="mode_presets_fallback.yaml"
PROVIDER_METADATA_FILE="provider_metadata.yaml"
DEFAULT_PRESET_NAME="Default"
DEFAULT_VISION_TIMEOUT_SECONDS=300
DEFAULT_VISION_MAX_TOKENS=2000
MODEL_PRESET_CONFIG_KEY="model_preset"
PRESET_SCOPE_GLOBAL="global"
PRESET_SCOPE_PROJECT="project"
PRESET_SLOT_CONFIG_SECTIONS={
    "chat": "chat_model",
    "vision": "vision_model",
    "utility": "utility_model",
    "embedding": "embedding_model",
}
MODEL_SLOT_PRESET_REPLACE_FIELDS={"kwargs"}
IMPLICIT_PRESET_SLOT_DEFAULTS={
    "vision": {
        "vision": True,
        "max_embeds": 10,
        "timeout": DEFAULT_VISION_TIMEOUT_SECONDS,
        "max_tokens": DEFAULT_VISION_MAX_TOKENS,
        "override_main": False,
        "rl_requests": 0,
        "rl_input": 0,
        "rl_output": 0,
        "kwargs": {},
    },
    "utility": {
        "ctx_length": 128000,
        "ctx_input": 0.7,
        "rl_requests": 0,
        "rl_input": 0,
        "rl_output": 0,
        "kwargs": {},
    },
    "embedding": {
        "rl_requests": 0,
        "rl_input": 0,
        "kwargs": {},
    },
}
LOCAL_PROVIDERS={"ollama", "lm_studio", "llama_cpp", "omlx", "vllm"}
LOCAL_EMBEDDING={"huggingface"}
_PROVIDER_METADATA_CACHE: dict | None=None


def _get_provider_metadata_path() -> str:
    plugin_dir=plugins.find_plugin_dir("_model_config")
    return files.get_abs_path(plugin_dir, PROVIDER_METADATA_FILE) if plugin_dir else ""


def get_provider_metadata(model_type: str="chat", provider: str="") -> dict:
    """Get plugin-owned provider metadata that does not belong in conf/model_providers.yaml."""
    global _PROVIDER_METADATA_CACHE
    if _PROVIDER_METADATA_CACHE is None:
        path=_get_provider_metadata_path()
        if path and files.exists(path):
            data=yaml_helper.loads(files.read_file(path))
            _PROVIDER_METADATA_CACHE=data if isinstance(data, dict) else {}
        else:
            _PROVIDER_METADATA_CACHE={}

    section=_PROVIDER_METADATA_CACHE.get(model_type, {})
    if not isinstance(section, dict):
        return {}
    meta=section.get(str(provider or "").strip().lower(), {})
    return meta if isinstance(meta, dict) else {}


def _model_type_for_label(label: str) -> str:
    return "embedding" if label == "Embedding Model" else "chat"


def provider_requires_api_key(provider: str, model_type: str="chat") -> bool:
    provider_id=str(provider or "").strip().lower()
    if not provider_id:
        return False
    cfg=get_provider_config(model_type, provider_id) or get_provider_config("chat", provider_id) or {}
    meta=get_provider_metadata(model_type, provider_id) or get_provider_metadata("chat", provider_id)
    mode=str(meta.get("api_key_mode") or cfg.get("api_key_mode") or "required").strip().lower()
    return mode not in {"none", "optional", "oauth"}


def _get_presets_path(project_name: str | None=None) -> str:
    """Return the user-editable presets path for the requested scope."""
    if project_name:
        return plugins.determine_plugin_asset_path(
            "_model_config", project_name, "", PRESETS_FILE
        )
    return files.get_abs_path(files.USER_DIR, files.PLUGINS_DIR, "_model_config", PRESETS_FILE)


def _get_fallback_presets_path() -> str:
    """Return the plugin-local fallback used when no saved presets exist."""
    plugin_dir=plugins.find_plugin_dir("_model_config")
    return files.get_abs_path(plugin_dir, FALLBACK_PRESETS_FILE) if plugin_dir else ""


def get_config(agent=None, project_name=None, agent_profile=None):
    """Get the resolved model config for an agent or selected scope."""
    config=plugins.get_plugin_config(
        "_model_config",
        agent=agent,
        project_name=project_name,
        agent_profile=agent_profile,
    ) or {}
    # The plugin hook resolves selection-only config. Keep this boundary robust
    # when hooks are disabled by tests or embedding applications.
    if any(section in config for section in PRESET_SLOT_CONFIG_SECTIONS.values()):
        return config
    return resolve_config_settings(config)


def get_configured_preset_name(agent=None, project_name=None, agent_profile=None) -> str:
    """Return the valid scoped preset selection, falling back to Default."""
    config=plugins.get_plugin_config(
        "_model_config",
        agent=agent,
        project_name=project_name,
        agent_profile=agent_profile,
    ) or {}
    name=str(config.get(MODEL_PRESET_CONFIG_KEY) or DEFAULT_PRESET_NAME).strip()
    return name if resolve_preset(name) else DEFAULT_PRESET_NAME


def preset_to_config(preset: dict) -> dict:
    """Convert a complete preset into the legacy runtime model-config shape."""
    config: dict={}
    for slot, section in PRESET_SLOT_CONFIG_SECTIONS.items():
        slot_config=preset.get(slot) if isinstance(preset, dict) else None
        config[section]=(
            _strip_ui_fields(slot_config, strip_api_key=False)
            if isinstance(slot_config, dict)
            else {}
        )
    return config


def config_to_preset(config: dict, name: str=DEFAULT_PRESET_NAME) -> dict:
    """Convert legacy full model config into a preset without UI/API-key fields."""
    preset={"name": str(name or "").strip()}
    for slot, section in PRESET_SLOT_CONFIG_SECTIONS.items():
        slot_config=config.get(section) if isinstance(config, dict) else None
        if isinstance(slot_config, dict):
            preset[slot]=_strip_ui_fields(slot_config, strip_api_key=True)
    return preset


def resolve_config_settings(settings: dict | None) -> dict:
    """Resolve selection-only settings to the complete runtime config shape."""
    raw=settings if isinstance(settings, dict) else {}
    selected_name=str(raw.get(MODEL_PRESET_CONFIG_KEY) or DEFAULT_PRESET_NAME).strip()
    default_preset=resolve_preset(DEFAULT_PRESET_NAME) or {"name": DEFAULT_PRESET_NAME}
    selected=resolve_preset(selected_name) or default_preset
    config=preset_to_config(default_preset)
    if selected.get("name") != DEFAULT_PRESET_NAME:
        config=build_config_from_preset(selected, config, strip_api_key=False)
    config[MODEL_PRESET_CONFIG_KEY]=str(selected.get("name") or DEFAULT_PRESET_NAME)
    # Retained as a read-only compatibility flag for integrations that still
    # inspect it. The switcher is always available in the unified preset model.
    config["allow_chat_override"]=True
    return config


def has_project_config(project_name: str) -> bool:
    path=plugins.determine_plugin_asset_path(
        "_model_config", project_name, "", plugins.CONFIG_FILE_NAME
    )
    return files.exists(path)


def load_project_llm_data(project_name: str) -> dict:
    """Build the preset-selection payload shown in Project Settings."""
    project_config_exists=has_project_config(project_name)
    preset_name=get_configured_preset_name(project_name=project_name)
    return {
        "has_project_config": project_config_exists,
        "selected_preset": {
            "scope": PRESET_SCOPE_GLOBAL,
            "project_name": "",
            "name": preset_name,
        },
        "presets": get_combined_presets(),
        "global_presets": get_presets(),
        "project_presets": [],
    }


def save_project_llm_settings(project_name: str, llm_data: object) -> None:
    """Persist only a project preset selection from Project Settings."""
    if not isinstance(llm_data, dict):
        return
    selected_preset=llm_data.get("selected_preset")
    if not isinstance(selected_preset, dict):
        return
    name=str(selected_preset.get("name") or "").strip()
    if resolve_preset(name):
        if not has_project_config(project_name) and name == get_configured_preset_name():
            return
        previous_embedding=get_config(project_name=project_name).get(
            "embedding_model",
            {},
        )
        plugins.save_plugin_config(
            "_model_config",
            project_name,
            "",
            {MODEL_PRESET_CONFIG_KEY: name},
        )
        current_embedding=get_config(project_name=project_name).get(
            "embedding_model",
            {},
        )
        if previous_embedding != current_embedding:
            defer.DeferredTask().start_task(
                call_extensions_async,
                "embedding_model_changed",
            )


def _load_presets_from_path(path: str) -> list | None:
    if files.exists(path):
        try:
            data=yaml_helper.loads(files.read_file(path))
        except Exception:
            return None
        if isinstance(data, list):
            return data
    return None


def _strip_ui_fields(value: dict, *, strip_api_key: bool) -> dict:
    cleaned=deepcopy(value)
    for key in list(cleaned.keys()):
        if key.startswith("_"):
            cleaned.pop(key, None)
    if strip_api_key:
        cleaned.pop("api_key", None)
    return cleaned


def _preset_default_values_equal(value, default) -> bool:
    if isinstance(default, float):
        try:
            return float(value) == default
        except (TypeError, ValueError):
            return False
    return value == default


def _strip_implicit_preset_defaults(slot: str, slot_config: dict) -> dict:
    cleaned=deepcopy(slot_config)
    defaults=IMPLICIT_PRESET_SLOT_DEFAULTS.get(slot, {})
    for key, default in defaults.items():
        if key in cleaned and _preset_default_values_equal(cleaned[key], default):
            cleaned.pop(key, None)
    return cleaned


def _clean_preset_for_file(preset: dict) -> dict:
    name=str(preset.get("name", "") or "").strip()
    if name.casefold() == DEFAULT_PRESET_NAME.casefold():
        name=DEFAULT_PRESET_NAME
    cleaned={
        "name": name,
    }
    has_named_slots=any(
        isinstance(preset.get(slot), dict) for slot in PRESET_SLOT_CONFIG_SECTIONS
    )
    for slot in PRESET_SLOT_CONFIG_SECTIONS:
        slot_config=preset.get(slot)
        if isinstance(slot_config, dict):
            slot_clean=_strip_ui_fields(slot_config, strip_api_key=True)
            cleaned[slot]=(
                slot_clean
                if name == DEFAULT_PRESET_NAME
                else _strip_implicit_preset_defaults(slot, slot_clean)
            )
    # Very old presets stored the main model directly beside ``name``. Preserve
    # those definitions while bringing them into the canonical slot schema.
    if not has_named_slots and _slot_has_identity(preset):
        raw_chat={
            key: value
            for key, value in preset.items()
            if key not in {"name", "scope", "project_name"}
        }
        raw_chat["name"]=name
        cleaned["chat"]=_strip_ui_fields(raw_chat, strip_api_key=True)
    return cleaned


def clean_presets_for_file(presets: list) -> list:
    """Return presets without API/UI metadata, preserving the plain YAML schema."""
    cleaned=[]
    for preset in presets:
        if isinstance(preset, dict):
            cleaned.append(_clean_preset_for_file(preset))
    return cleaned


def validate_presets(presets: list, *, require_default: bool=True) -> list:
    """Validate and clean the durable global preset collection."""
    if not isinstance(presets, list):
        raise ValueError("Presets must be a list.")

    cleaned: list[dict]=[]
    seen: set[str]=set()
    for raw in presets:
        if not isinstance(raw, dict):
            raise ValueError("Every preset must be an object.")
        preset=_clean_preset_for_file(raw)
        name=str(preset.get("name") or "").strip()
        if not name:
            raise ValueError("Preset names cannot be empty.")
        normalized=name.casefold()
        if normalized in seen:
            raise ValueError(f"Preset names must be unique: '{name}'.")
        if normalized == DEFAULT_PRESET_NAME.casefold():
            preset["name"]=DEFAULT_PRESET_NAME
            for slot, label in (
                ("chat", "main"),
                ("utility", "utility"),
                ("embedding", "embedding"),
            ):
                if not _slot_has_identity(preset.get(slot) or {}):
                    raise ValueError(
                        f"The Default preset requires a {label} model."
                    )
        seen.add(normalized)
        cleaned.append(preset)

    default_index=next(
        (i for i, preset in enumerate(cleaned) if preset["name"] == DEFAULT_PRESET_NAME),
        None,
    )
    if require_default and default_index is None:
        raise ValueError("The Default preset cannot be deleted or renamed.")
    if default_index not in (None, 0):
        cleaned.insert(0, cleaned.pop(default_index))
    return cleaned


def normalize_config_for_save(config: dict) -> dict:
    """Remove UI-only fields and inline API keys before storing scoped config."""
    cleaned=deepcopy(config or {})
    for section_name in (
        "chat_model",
        "vision_model",
        "utility_model",
        "embedding_model",
    ):
        section=cleaned.get(section_name)
        if isinstance(section, dict):
            cleaned[section_name]=_strip_ui_fields(section, strip_api_key=True)
    return cleaned


def _legacy_default_preset() -> dict | None:
    """Build Default from a pre-v2 global config when startup migration has not run."""
    path=plugins.determine_plugin_asset_path(
        "_model_config", "", "", plugins.CONFIG_FILE_NAME
    )
    if not files.exists(path):
        return None
    try:
        raw=files.read_file_json(path)
    except Exception:
        return None
    if not isinstance(raw, dict) or not any(
        section in raw for section in PRESET_SLOT_CONFIG_SECTIONS.values()
    ):
        return None
    return config_to_preset(raw, DEFAULT_PRESET_NAME)


def parse_preset_collection(text: str) -> list:
    """Parse, validate, and sanitize a preset YAML document."""
    return validate_presets(yaml_helper.loads(text))


def _fallback_presets() -> list:
    path=_get_fallback_presets_path()
    if not files.exists(path):
        return []
    try:
        return parse_preset_collection(files.read_file(path))
    except Exception:
        return []


def _ensure_default_preset(presets: list) -> list:
    result=[deepcopy(preset) for preset in presets if isinstance(preset, dict)]
    legacy_default=_legacy_default_preset()
    bundled_default=next(
        (
            deepcopy(preset)
            for preset in _fallback_presets()
            if isinstance(preset, dict)
            and str(preset.get("name") or "").strip().casefold()
            == DEFAULT_PRESET_NAME.casefold()
        ),
        None,
    )
    fallback_default=bundled_default or {"name": DEFAULT_PRESET_NAME}
    if legacy_default:
        for slot in PRESET_SLOT_CONFIG_SECTIONS:
            legacy_slot=legacy_default.get(slot)
            if _slot_has_identity(legacy_slot or {}):
                fallback_default[slot]=deepcopy(legacy_slot)
    default_index=next(
        (
            i
            for i, preset in enumerate(result)
            if str(preset.get("name") or "").strip().casefold()
            == DEFAULT_PRESET_NAME.casefold()
        ),
        None,
    )
    if default_index is not None:
        result[default_index]["name"]=DEFAULT_PRESET_NAME
        for slot in PRESET_SLOT_CONFIG_SECTIONS:
            if not _slot_has_identity(result[default_index].get(slot) or {}):
                fallback_slot=fallback_default.get(slot)
                if isinstance(fallback_slot, dict):
                    result[default_index][slot]=deepcopy(fallback_slot)
        if default_index:
            result.insert(0, result.pop(default_index))
        return result

    result.insert(0, fallback_default)
    return result


def get_presets(project_name: str | None=None) -> list:
    """Get global presets with the required Default preset first."""
    if project_name:
        return get_project_presets(project_name)

    path=_get_presets_path()
    presets=_load_presets_from_path(path)
    if presets is not None:
        return _ensure_default_preset(presets)

    # Fall back to the repository-shipped offline collection.
    return _ensure_default_preset(_fallback_presets())


def get_project_presets(project_name: str) -> list:
    """Load legacy project presets for migration/compatibility only."""
    return _load_presets_from_path(_get_presets_path(project_name)) or []


def _with_preset_metadata(preset: dict, scope: str, project_name: str="") -> dict:
    item=deepcopy(preset)
    item["scope"]=scope
    item["project_name"]=project_name if scope == PRESET_SCOPE_PROJECT else ""
    item["name"]=str(item.get("name", "") or "")
    return item


def get_combined_presets(project_name: str | None=None) -> list:
    """Get global presets with API metadata (project definitions are retired)."""
    return [
        _with_preset_metadata(preset, PRESET_SCOPE_GLOBAL)
        for preset in get_presets()
        if isinstance(preset, dict)
    ]


def save_presets(presets: list, project_name: str | None=None) -> None:
    """Save global presets while enforcing the immutable Default identity."""
    if project_name:
        raise ValueError("Project-specific preset definitions are no longer supported.")
    cleaned=validate_presets(presets)
    path=_get_presets_path(project_name)
    files.write_file(path, yaml_helper.dumps(cleaned))


def update_preset_from_config(name: str, config: dict) -> dict:
    """Replace one global preset's model slots from a legacy config payload."""
    target=resolve_preset(name)
    if not target:
        raise ValueError(f"Preset '{name}' was not found.")
    canonical_name=str(target.get("name") or DEFAULT_PRESET_NAME)
    replacement=config_to_preset(config, canonical_name)
    if canonical_name == DEFAULT_PRESET_NAME:
        for slot in PRESET_SLOT_CONFIG_SECTIONS:
            if not _slot_has_identity(replacement.get(slot) or {}):
                current_slot=target.get(slot)
                if isinstance(current_slot, dict):
                    replacement[slot]=deepcopy(current_slot)
    presets=get_presets()
    updated=False
    for index, preset in enumerate(presets):
        if str(preset.get("name") or "").casefold() == canonical_name.casefold():
            presets[index]=replacement
            updated=True
            break
    if not updated:
        raise ValueError(f"Preset '{name}' was not found.")
    save_presets(presets)
    return replacement


def reset_presets(project_name: str | None=None) -> list:
    """Delete user presets for the scope. Global reset falls back to bundled defaults."""
    if project_name:
        raise ValueError("Project-specific preset definitions are no longer supported.")
    path=_get_presets_path(project_name)
    if os.path.exists(path):
        os.remove(path)
    return get_presets()


def resolve_preset(
    name: str,
    *,
    scope: str=PRESET_SCOPE_GLOBAL,
    project_name: str | None=None,
) -> dict | None:
    """Resolve a preset by explicit scope so same-name presets are unambiguous."""
    if scope == PRESET_SCOPE_PROJECT:
        return None
    presets=get_presets()

    for p in presets:
        if str(p.get("name") or "").casefold() == str(name or "").strip().casefold():
            return p
    return None


def resolve_preset_selection(selection: dict | str, project_name: str | None=None) -> dict | None:
    """Resolve a UI/API preset selection payload to a preset dict."""
    if isinstance(selection, str):
        return resolve_preset(selection)
    if not isinstance(selection, dict):
        return None

    scope=str(selection.get("scope") or PRESET_SCOPE_GLOBAL)
    if scope == "current":
        return None
    name=str(selection.get("name") or "")
    selected_project=