fix(cli): key model-catalog cache by exact API key, shorten TTL, skip Ollama

Follow-ups to #6462's caching:

1. Key the catalog cache by the exact API key (via a short, non-reversible
   sha256 digest — never the key itself), not just key-present vs absent.
   Switching to a different key for the same provider now misses the previous
   account's entry and refetches, instead of showing the old account's models.

2. Never cache local providers (Ollama). /api/tags is fast and installed
   models change out-of-band, so caching could keep offering a model the user
   just deleted until the entry expired. _is_cacheable() gates both cache read
   and write; the picker now re-probes every call and reflects what's installed.

3. Shorten the dynamic catalog TTL from 6h to 5m — a stale list (new/removed
   models, account changes) is worse than a ~1s refetch, and the cache only
   needs to spare repeated fetches within a wizard session.

Tests: distinct-key cache entries, digest never stores the raw key, Ollama not
cached (reflects deletions / never written), and dynamic TTL expiry.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RBYGqJHC2TMC6fonFziuuh

lib/cli/src/crewai_cli/create_json_crew.py

@@ -14,6 +14,7 @@ from rich.text import Text
 
 from crewai_cli.constants import ENV_VARS
 from crewai_cli.git import initialize_if_git_available
+from crewai_cli.model_catalog import get_provider_models
 from crewai_cli.tui_picker import pick_many, pick_one
 from crewai_cli.utils import (
     enable_prompt_line_editing,
@@ -42,41 +43,50 @@ _PROVIDERS: list[tuple[str, str]] = [
     ("watson", "IBM watsonx"),
 ]
 
+# Curated offline fallback / label source. The picker prefers models pulled
+# live from the vendor's own API via ``model_catalog.get_provider_models``;
+# this list is the hand-verified backstop used when no API key is available.
+# Keep entries to real, current model ids — last verified against each vendor's
+# official model docs on 2026-07-05.
 _PROVIDER_MODELS: dict[str, list[tuple[str, str]]] = {
     "openai": [
         ("gpt-5.5", "GPT-5.5"),
         ("gpt-5.5-pro", "GPT-5.5 Pro"),
         ("gpt-5.4", "GPT-5.4"),
-        ("o4-mini", "o4-mini"),
+        ("gpt-5.4-mini", "GPT-5.4 Mini"),
+        ("gpt-5.2", "GPT-5.2"),
         ("gpt-4.1", "GPT-4.1"),
-        ("gpt-4.1-mini", "GPT-4.1 Mini"),
     ],
     "anthropic": [
-        ("claude-opus-4-6", "Claude Opus 4.6"),
+        ("claude-fable-5", "Claude Fable 5"),
+        ("claude-opus-4-8", "Claude Opus 4.8"),
+        ("claude-sonnet-5", "Claude Sonnet 5"),
+        ("claude-opus-4-7", "Claude Opus 4.7"),
+        ("claude-haiku-4-5", "Claude Haiku 4.5"),
         ("claude-sonnet-4-6", "Claude Sonnet 4.6"),
-        ("claude-haiku-4-5-20251001", "Claude Haiku 4.5"),
-        ("claude-3-7-sonnet-20250219", "Claude 3.7 Sonnet"),
-        ("claude-3-5-sonnet-20241022", "Claude 3.5 Sonnet"),
     ],
     "gemini": [
-        ("gemini-3-pro-preview", "Gemini 3 Pro (preview)"),
-        ("gemini-2.5-pro-exp-03-25", "Gemini 2.5 Pro"),
-        ("gemini-2.5-flash-preview-04-17", "Gemini 2.5 Flash"),
-        ("gemini-2.0-flash-001", "Gemini 2.0 Flash"),
-        ("gemini-1.5-pro", "Gemini 1.5 Pro"),
+        ("gemini-3.5-flash", "Gemini 3.5 Flash"),
+        ("gemini-3.1-pro-preview", "Gemini 3.1 Pro (preview)"),
+        ("gemini-3-flash-preview", "Gemini 3 Flash (preview)"),
+        ("gemini-2.5-pro", "Gemini 2.5 Pro"),
+        ("gemini-2.5-flash", "Gemini 2.5 Flash"),
+        ("gemini-2.5-flash-lite", "Gemini 2.5 Flash Lite"),
     ],
     "groq": [
+        ("meta-llama/llama-4-maverick-17b-128e-instruct", "Llama 4 Maverick"),
+        ("meta-llama/llama-4-scout-17b-16e-instruct", "Llama 4 Scout"),
+        ("openai/gpt-oss-120b", "GPT-OSS 120B"),
+        ("qwen/qwen3-32b", "Qwen3 32B"),
+        ("moonshotai/kimi-k2-instruct-0905", "Kimi K2"),
         ("llama-3.3-70b-versatile", "Llama 3.3 70B"),
-        ("llama-3.1-70b-versatile", "Llama 3.1 70B"),
-        ("llama-3.1-8b-instant", "Llama 3.1 8B"),
-        ("deepseek-r1-distill-llama-70b", "DeepSeek R1 70B"),
-        ("mixtral-8x7b-32768", "Mixtral 8x7B"),
     ],
     "ollama": [
         ("llama3.3", "Llama 3.3"),
-        ("llama3.1", "Llama 3.1"),
+        ("qwen3", "Qwen 3"),
         ("deepseek-r1", "DeepSeek R1"),
-        ("qwen2.5", "Qwen 2.5"),
+        ("gpt-oss", "GPT-OSS"),
+        ("gemma3", "Gemma 3"),
         ("mistral", "Mistral"),
     ],
 }
@@ -758,7 +768,9 @@ def _select_model() -> str:
     provider_key, provider_name = _PROVIDERS[p_idx]
     click.secho(f"  → {provider_name}", fg="green")
 
-    models = _PROVIDER_MODELS.get(provider_key, [])
+    # Prefer the latest models pulled live from the vendor / LiteLLM; the
+    # curated ``_PROVIDER_MODELS`` entry is the offline fallback and label source.
+    models = get_provider_models(provider_key, _PROVIDER_MODELS.get(provider_key, []))
     if not models:
         custom = click.prompt(
             click.style(f"  Enter model name for {provider_key}/", fg="cyan"),

lib/cli/src/crewai_cli/model_catalog.py

@@ -0,0 +1,673 @@
+"""Dynamic model catalog for the crew-creation wizard.
+
+Resolves the models to offer for a given provider using a three-tier strategy:
+
+1. **Vendor API** - when the provider's API key is already present in the
+   environment, query the vendor's own model-listing endpoint. This is the only
+   source that reliably reflects the *latest* models (real release dates /
+   display names, straight from the vendor).
+2. **Curated hardcoded fallback** - the hand-verified list baked into the
+   wizard, used when no API key is available. Authoritative but frozen, so it is
+   refreshed periodically.
+3. **LiteLLM feed** - the community ``model_prices_and_context_window.json`` the
+   CLI already caches. Only used for providers with *no* curated list: the feed
+   lags real releases badly (it can miss a vendor's newest models entirely), so
+   it must never preempt the curated fallback.
+
+Every tier is best-effort: any network error, timeout, missing key, or empty
+result quietly falls through to the next tier, and the caller's hardcoded list
+is always the final backstop. The picker never blocks for long — network calls
+use a short timeout and successful results are cached.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+import contextlib
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import time
+from typing import Any
+
+import certifi
+import httpx
+
+from crewai_cli.constants import JSON_URL
+
+
+# ── Tunables ─────────────────────────────────────────────────────
+
+#: How many models to surface per provider.
+MAX_MODELS = 8
+
+#: Timeout (seconds) for any network call made while resolving models.
+_TIMEOUT = 6.0
+
+#: How long a resolved (dynamic) catalog stays fresh before we refetch. Kept
+#: short: it only spares the picker repeated fetches within a wizard session,
+#: and a stale list (new/removed models, account changes) is worse than a ~1s
+#: refetch. Local providers (Ollama) are not cached at all — see _is_cacheable.
+_CATALOG_TTL = 300
+
+#: How long a fallback result is cached after a failed/empty fetch. Short, so a
+#: newly-added API key takes effect soon, but long enough to spare the picker a
+#: repeated timeout-prone network attempt on every call within one session.
+_NEGATIVE_TTL = 300
+
+#: How long the shared LiteLLM feed cache stays fresh.
+_LITELLM_TTL = 24 * 3600
+
+#: Env vars that may hold each provider's API key, in priority order. A
+#: provider with an empty tuple (e.g. local Ollama) needs no key. Gemini accepts
+#: either name, matching crewai's own Gemini provider.
+_PROVIDER_KEY_ENV: dict[str, tuple[str, ...]] = {
+    "openai": ("OPENAI_API_KEY",),
+    "anthropic": ("ANTHROPIC_API_KEY",),
+    "gemini": ("GEMINI_API_KEY", "GOOGLE_API_KEY"),
+    "groq": ("GROQ_API_KEY",),
+    "cerebras": ("CEREBRAS_API_KEY",),
+    "ollama": (),
+}
+
+
+def _provider_api_key(provider_key: str) -> str | None:
+    """First non-empty API key found among the provider's env vars."""
+    for env in _PROVIDER_KEY_ENV.get(provider_key, ()):
+        value = os.environ.get(env)
+        if value:
+            return value
+    return None
+
+
+# Substrings that mark a model id as *not* a chat/completion model. Used to
+# filter noisy OpenAI-compatible ``/models`` listings.
+_NON_CHAT_MARKERS = (
+    "embedding",
+    "embed",
+    "whisper",
+    "tts",
+    "audio",
+    "transcribe",
+    "realtime",
+    "dall-e",
+    "dalle",
+    "image",
+    "moderation",
+    "similarity",
+    "-edit",
+    "davinci-002",
+    "babbage-002",
+    "computer-use",
+    "guard",
+)
+
+_ACRONYMS = {
+    "gpt": "GPT",
+    "ai": "AI",
+    "nim": "NIM",
+    "llm": "LLM",
+    "hd": "HD",
+    "us": "US",
+    "eu": "EU",
+    "oss": "OSS",
+    "it": "IT",
+}
+
+# Tokens with non-title-case brand capitalization.
+_BRAND_TOKENS = {
+    "deepseek": "DeepSeek",
+    "chatgpt": "ChatGPT",
+    "qwq": "QwQ",
+}
+
+
+# ── Public API ───────────────────────────────────────────────────
+
+
+def get_provider_models(
+    provider_key: str, fallback: list[tuple[str, str]]
+) -> list[tuple[str, str]]:
+    """Return ``(model_id, label)`` pairs for ``provider_key``, newest first.
+
+    Tries the vendor API (if a key is in the environment) first, since it is the
+    only reliably-fresh source. When no key is available it returns the curated
+    ``fallback`` verbatim — the LiteLLM feed is consulted **only** for providers
+    with no curated list, because the feed lags real releases and would
+    otherwise surface a staler list than the hand-verified fallback. Never
+    raises: any failure degrades to the next tier.
+
+    Args:
+        provider_key: Short provider identifier, e.g. ``"anthropic"``.
+        fallback: Curated ``(model_id, label)`` pairs to use as the backstop and
+            to source friendly labels for known models.
+
+    Returns:
+        Up to :data:`MAX_MODELS` ``(model_id, label)`` pairs. Falls back to
+        ``fallback`` verbatim when no fresher list can be resolved.
+    """
+    cached = _read_catalog_cache(provider_key)
+    if cached is not None:
+        return cached
+
+    label_map = {model_id: label for model_id, label in fallback}
+
+    # A non-None vendor result is authoritative — even when empty (e.g. a
+    # reachable Ollama with no models installed): show that rather than
+    # hardcoded suggestions the crew can't actually run. The picker handles an
+    # empty list by prompting for manual entry.
+    vendor = _from_vendor(provider_key)
+    if vendor is not None:
+        result = _finalize(vendor, label_map)
+        if result:
+            _write_catalog_cache(provider_key, result, source="dynamic")
+        return result
+
+    # Vendor tier unavailable. The LiteLLM feed lags real releases, so only
+    # reach for it when we have no curated fallback — never override the fallback.
+    entries = _from_litellm(provider_key) if not fallback else None
+    result = _finalize(entries, label_map) if entries else []
+    if result:
+        _write_catalog_cache(provider_key, result, source="dynamic")
+        return result
+
+    # Nothing fresher than the curated list. Cache it briefly (negative cache)
+    # so a failed vendor/LiteLLM fetch isn't retried on every subsequent call.
+    # (_write_catalog_cache skips non-cacheable providers like Ollama.)
+    if fallback:
+        _write_catalog_cache(provider_key, fallback, source="fallback")
+    return fallback
+
+
+# ── Tier 1: vendor APIs ──────────────────────────────────────────
+
+
+def _from_vendor(provider_key: str) -> list[dict[str, Any]] | None:
+    """Fetch models from the vendor.
+
+    Returns the model list on a successful fetch — **including an empty list**,
+    which is meaningful (e.g. a reachable Ollama server with nothing installed).
+    Returns ``None`` only when the vendor tier is unavailable: no fetcher, no
+    API key, or the request failed.
+    """
+    fetcher = _VENDOR_FETCHERS.get(provider_key)
+    if fetcher is None:
+        return None
+
+    api_key = _provider_api_key(provider_key)
+    if _PROVIDER_KEY_ENV.get(provider_key) and not api_key:
+        # Provider needs a key and none is set — skip to the next tier.
+        return None
+
+    try:
+        return fetcher(api_key)
+    except Exception:
+        # Network error, auth failure, unexpected payload — degrade quietly.
+        return None
+
+
+def _fetch_openai(api_key: str | None) -> list[dict[str, Any]]:
+    return _fetch_openai_compatible("https://api.openai.com/v1", api_key)
+
+
+def _fetch_groq(api_key: str | None) -> list[dict[str, Any]]:
+    return _fetch_openai_compatible("https://api.groq.com/openai/v1", api_key)
+
+
+def _fetch_cerebras(api_key: str | None) -> list[dict[str, Any]]:
+    return _fetch_openai_compatible("https://api.cerebras.ai/v1", api_key)
+
+
+def _fetch_openai_compatible(
+    base_url: str, api_key: str | None
+) -> list[dict[str, Any]]:
+    """Parse an OpenAI-shaped ``GET /models`` response."""
+    data = _http_get_json(
+        f"{base_url}/models",
+        headers={"Authorization": f"Bearer {api_key}"},
+    )
+    entries: list[dict[str, Any]] = []
+    for item in data.get("data", []):
+        model_id = item.get("id")
+        if not model_id or not _is_chat_model(model_id) or _is_fine_tune(model_id):
+            continue
+        created = _as_float(item.get("created"))
+        entries.append(_entry(model_id, _humanize(model_id), created=created))
+    return entries
+
+
+def _fetch_anthropic(api_key: str | None) -> list[dict[str, Any]]:
+    data = _http_get_json(
+        "https://api.anthropic.com/v1/models",
+        headers={"x-api-key": api_key or "", "anthropic-version": "2023-06-01"},
+    )
+    entries: list[dict[str, Any]] = []
+    for item in data.get("data", []):
+        model_id = item.get("id")
+        if not model_id:
+            continue
+        label = item.get("display_name") or _humanize(model_id)
+        created = _parse_iso(item.get("created_at"))
+        entries.append(_entry(model_id, label, created=created))
+    return entries
+
+
+def _fetch_gemini(api_key: str | None) -> list[dict[str, Any]]:
+    entries: list[dict[str, Any]] = []
+    params: dict[str, Any] = {"key": api_key or "", "pageSize": 200}
+    # models.list is paginated and not guaranteed newest-first, so walk pages
+    # (bounded) to see the full set — _finalize does the sort + truncation.
+    for _ in range(10):
+        try:
+            data = _http_get_json(
+                "https://generativelanguage.googleapis.com/v1beta/models",
+                params=params,
+            )
+        except Exception:
+            # Later-page failure: keep the models already gathered. First-page
+            # failure (nothing gathered yet) is a real outage — re-raise so the
+            # caller falls back to the curated list rather than mistaking it for
+            # a successful empty result.
+            if entries:
+                break
+            raise
+        for item in data.get("models", []):
+            methods = item.get("supportedGenerationMethods") or []
+            if "generateContent" not in methods:
+                continue
+            name = (item.get("name") or "").removeprefix("models/")
+            if not name or not _is_chat_model(name) or "aqa" in name:
+                continue
+            label = item.get("displayName") or _humanize(name)
+            # Gemini has no timestamp; rank by the version in name/version.
+            version_hint = f"{name} {item.get('version') or ''}"
+            entries.append(_entry(name, label, version_hint=version_hint))
+        token = data.get("nextPageToken")
+        if not token:
+            break
+        params = {"key": api_key or "", "pageSize": 200, "pageToken": token}
+    return entries
+
+
+def _ollama_base() -> str:
+    """Resolve the Ollama server base URL from the environment.
+
+    Checks ``OLLAMA_API_BASE`` / ``API_BASE`` (what LiteLLM and the generated
+    crew use) first, then ``OLLAMA_HOST`` (the Ollama runtime convention), so a
+    user who only set ``OLLAMA_HOST`` sees models from the right server.
+    """
+    base = (
+        os.environ.get("OLLAMA_API_BASE")
+        or os.environ.get("API_BASE")
+        or os.environ.get("OLLAMA_HOST")
+        or "http://localhost:11434"
+    ).strip()
+    # OLLAMA_HOST is often scheme-less (e.g. "127.0.0.1:11434").
+    if "://" not in base:
+        base = f"http://{base}"
+    return base.rstrip("/")
+
+
+def _fetch_ollama(_api_key: str | None) -> list[dict[str, Any]]:
+    """List models installed on the local Ollama server (no API key)."""
+    data = _http_get_json(f"{_ollama_base()}/api/tags")
+    entries: list[dict[str, Any]] = []
+    for item in data.get("models", []):
+        model_id = item.get("model") or item.get("name")
+        if not model_id or not _is_chat_model(model_id) or _is_fine_tune(model_id):
+            # /api/tags lists everything installed, including embedding models.
+            continue
+        # Ollama returns an ISO 8601 modified_at we can rank by.
+        created = _parse_iso(item.get("modified_at"))
+        entries.append(_entry(model_id, _humanize(model_id), created=created))
+    return entries
+
+
+_VENDOR_FETCHERS: dict[str, Callable[[str | None], list[dict[str, Any]]]] = {
+    "openai": _fetch_openai,
+    "anthropic": _fetch_anthropic,
+    "gemini": _fetch_gemini,
+    "groq": _fetch_groq,
+    "cerebras": _fetch_cerebras,
+    "ollama": _fetch_ollama,
+}
+
+
+# ── Tier 2: LiteLLM feed ─────────────────────────────────────────
+
+# Process-level memo so a single CLI run attempts the LiteLLM download at most
+# once — repeated picker calls otherwise each incur a multi-second timeout when
+# the feed is stale/unreachable. Reset via _reset_litellm_memo() in tests.
+_UNSET: Any = object()
+_litellm_memo: Any = _UNSET
+
+
+def _reset_litellm_memo() -> None:
+    """Clear the process-level LiteLLM memo (test hook)."""
+    global _litellm_memo
+    _litellm_memo = _UNSET
+
+
+def _from_litellm(provider_key: str) -> list[dict[str, Any]] | None:
+    """Build chat-model entries for ``provider_key`` from the LiteLLM feed."""
+    data = _load_litellm_data()
+    # A corrupt feed (non-mapping JSON root) must not crash the picker.
+    if not isinstance(data, dict):
+        return None
+
+    entries: list[dict[str, Any]] = []
+    for model_name, props in data.items():
+        if not isinstance(props, dict):
+            continue
+        # `litellm_provider` can be present-but-null in the feed; coerce before
+        # string ops so a null value is skipped rather than raising.
+        if (props.get("litellm_provider") or "").strip().lower() != provider_key:
+            continue
+        if props.get("mode") != "chat":
+            continue
+        # LiteLLM keys are sometimes prefixed with the provider; the picker
+        # re-adds ``provider/`` itself, so strip a leading one to avoid dupes.
+        model_id = model_name
+        if model_id.startswith(f"{provider_key}/"):
+            model_id = model_id[len(provider_key) + 1 :]
+        if not model_id:
+            continue
+        entries.append(_entry(model_id, _humanize(model_id), version_hint=model_id))
+    return entries or None
+
+
+def _load_litellm_data() -> dict[str, Any] | None:
+    """Return the LiteLLM feed, memoized once per process (see _litellm_memo)."""
+    global _litellm_memo
+    if _litellm_memo is _UNSET:
+        _litellm_memo = _fetch_litellm_data()
+    memoized: dict[str, Any] | None = _litellm_memo
+    return memoized
+
+
+def _fetch_litellm_data() -> dict[str, Any] | None:
+    """Read the cached LiteLLM feed, fetching it once if the cache is cold."""
+    cache_file = _litellm_cache_file()
+    fresh = (
+        cache_file.exists()
+        and (time.time() - cache_file.stat().st_mtime) < _LITELLM_TTL
+    )
+    if fresh:
+        data = _read_json(cache_file)
+        # A corrupt/non-mapping fresh cache must not block a recoverable
+        # download — only short-circuit on a usable mapping.
+        if isinstance(data, dict) and data:
+            return data
+
+    try:
+        data = _http_get_json(JSON_URL)
+    except Exception:
+        # Fall back to a stale cache if we have one, else give up on this tier.
+        return _read_json(cache_file)
+
+    # Best-effort cache write; a failure (e.g. read-only home) is non-fatal
+    # since we already hold the freshly-fetched data.
+    with contextlib.suppress(OSError):
+        cache_file.parent.mkdir(parents=True, exist_ok=True)
+        cache_file.write_text(json.dumps(data), encoding="utf-8")
+    return data
+
+
+# ── Ranking + labelling ──────────────────────────────────────────
+
+
+def _finalize(
+    entries: list[dict[str, Any]], label_map: dict[str, str]
+) -> list[tuple[str, str]]:
+    """Sort newest-first, dedupe, relabel with curated names, and truncate."""
+    entries.sort(key=lambda e: e["sort"], reverse=True)
+    seen: set[str] = set()
+    out: list[tuple[str, str]] = []
+    for entry in entries:
+        model_id = entry["id"]
+        if model_id in seen:
+            continue
+        seen.add(model_id)
+        label = label_map.get(model_id) or entry["label"]
+        out.append((model_id, label))
+        if len(out) >= MAX_MODELS:
+            break
+    return out
+
+
+def _entry(
+    model_id: str,
+    label: str,
+    *,
+    created: float = 0.0,
+    version_hint: str | None = None,
+) -> dict[str, Any]:
+    """Build a rankable catalog entry.
+
+    ``sort`` is a comparable tuple ``(created, date_int, version_tuple)`` so a
+    real vendor timestamp wins, then a date embedded in the id, then the numeric
+    version. Types line up positionally, so entries compare cleanly.
+    """
+    date_int, version = _version_key(version_hint or model_id)
+    return {
+        "id": model_id,
+        "label": label,
+        "sort": (created, date_int, version),
+    }
+
+
+_DATE_RE = re.compile(r"(20\d{2})[-_]?(0[1-9]|1[0-2])[-_]?(0[1-9]|[12]\d|3[01])")
+_NUM_RE = re.compile(r"\d+")
+
+
+def _version_key(text: str) -> tuple[int, tuple[int, ...]]:
+    """Extract a ``(date_int, version_tuple)`` sort key from a model id.
+
+    A trailing/embedded ``YYYYMMDD`` (or ``YYYY-MM-DD``) becomes ``date_int``;
+    remaining numbers become the version tuple. ``claude-opus-4-6`` → version
+    ``(4, 6)``; ``claude-3-5-sonnet-20241022`` → date ``20241022`` version
+    ``(3, 5)``.
+    """
+    text = text or ""
+    date_int = 0
+    match = _DATE_RE.search(text)
+    if match:
+        date_int = int(match.group(1) + match.group(2) + match.group(3))
+        text = _DATE_RE.sub(" ", text)
+    version = tuple(int(n) for n in _NUM_RE.findall(text)[:4])
+    return date_int, version
+
+
+def _is_chat_model(model_id: str) -> bool:
+    """Heuristically reject embedding/audio/image/etc. models by their id."""
+    lowered = model_id.lower()
+    return not any(marker in lowered for marker in _NON_CHAT_MARKERS)
+
+
+def _is_fine_tune(model_id: str) -> bool:
+    """A user fine-tune or training checkpoint (``ft:...`` / ``...:ckpt-step-N``).
+
+    These are account-specific artifacts: they clutter the picker, crowd out the
+    foundation models (their recent ``created`` timestamps rank them first), and
+    humanize into unreadable labels. Excluded from the auto-list; a user who
+    wants one can still enter it via the picker's "Other" option.
+    """
+    lowered = model_id.lower()
+    return lowered.startswith("ft:") or ":ckpt" in lowered
+
+
+_SIZE_RE = re.compile(r"^\d+(?:\.\d+)?[bmk]$")  # 8b, 70b, 1.5b, 120m, 32k
+_OSERIES_RE = re.compile(r"^o\d+$")  # o1, o3, o4 — kept lowercase (OpenAI brand)
+
+
+def _humanize(model_id: str) -> str:
+    """Derive a readable label from a raw model id.
+
+    Best-effort only — vendor display names and the curated label map take
+    precedence. Drops embedded dates and applies light casing so raw ids read
+    cleanly: ``gpt-oss-120b`` → ``GPT OSS 120B``, ``qwen3-32b`` → ``Qwen3 32B``,
+    ``deepseek-r1:671b`` → ``DeepSeek R1 671B``, ``o3-mini`` → ``o3 Mini``.
+    """
+    base = model_id.split("/")[-1]
+    # Drop embedded release dates — they're noise in a label, and the picker
+    # already shows the full model id alongside it.
+    base = _DATE_RE.sub(" ", base)
+    words: list[str] = []
+    # Split on separators including ``:`` so Ollama tags (llama3.3:70b) read well.
+    for part in re.split(r"[-_\s:]+", base):
+        if not part:
+            continue
+        low = part.lower()
+        if low in _ACRONYMS:
+            words.append(_ACRONYMS[low])
+        elif low in _BRAND_TOKENS:
+            words.append(_BRAND_TOKENS[low])
+        elif _SIZE_RE.match(low):
+            words.append(low[:-1] + low[-1].upper())  # 70b -> 70B
+        elif _OSERIES_RE.match(low):
+            words.append(low)  # o3 stays lowercase
+        elif part[0].isalpha():
+            # Capitalize the leading letter, preserve the rest (so a fused
+            # family+version keeps its digits): qwen3 -> Qwen3, mini -> Mini.
+            words.append(part[0].upper() + part[1:])
+        else:
+            words.append(part)  # starts with a digit (4o, 4.1, 0905) — leave as-is
+    return " ".join(words) or base
+
+
+# ── HTTP + parsing helpers ───────────────────────────────────────
+
+
+def _http_get_json(
+    url: str,
+    *,
+    headers: dict[str, str] | None = None,
+    params: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """GET ``url`` and return parsed JSON, with a short timeout and TLS verify."""
+    ssl_config = os.environ.get("SSL_CERT_FILE") or certifi.where()
+    response = httpx.get(
+        url,
+        headers=headers,
+        params=params,
+        timeout=_TIMEOUT,
+        verify=ssl_config,
+        follow_redirects=True,
+    )
+    response.raise_for_status()
+    result: dict[str, Any] = response.json()
+    return result
+
+
+def _parse_iso(value: Any) -> float:
+    """Parse an ISO 8601 timestamp to an epoch float; ``0.0`` on failure."""
+    if not value or not isinstance(value, str):
+        return 0.0
+    from datetime import datetime
+
+    try:
+        return datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp()
+    except ValueError:
+        return 0.0
+
+
+def _as_float(value: Any) -> float:
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        return 0.0
+
+
+def _read_json(path: Path) -> dict[str, Any] | None:
+    try:
+        data: dict[str, Any] = json.loads(path.read_text(encoding="utf-8"))
+        return data
+    except (OSError, json.JSONDecodeError):
+        return None
+
+
+# ── Caching ──────────────────────────────────────────────────────
+
+
+def _cache_dir() -> Path:
+    return Path.home() / ".crewai"
+
+
+def _catalog_cache_file() -> Path:
+    return _cache_dir() / "model_catalog_cache.json"
+
+
+def _litellm_cache_file() -> Path:
+    # Shared with crewai_cli.provider so both flows warm the same cache.
+    return _cache_dir() / "provider_cache.json"
+
+
+def _is_cacheable(provider_key: str) -> bool:
+    """Whether a provider's resolved catalog may be cached.
+
+    Ollama is a local server (``/api/tags`` is fast), and its installed models
+    change out-of-band, so it is never cached — the picker re-probes every call
+    and always reflects what is currently installed.
+    """
+    return provider_key != "ollama"
+
+
+def _cache_key(provider_key: str) -> str:
+    """Cache key for a provider's resolved model list.
+
+    Keyed by the exact API key (via a short, non-reversible digest — never the
+    key itself), so switching to a different key for the same provider misses
+    the previous account's cached entry and refetches. Absent key -> ``#nokey``,
+    which also keeps a negatively-cached no-key fallback from shadowing a run
+    after a key is added.
+    """
+    api_key = _provider_api_key(provider_key)
+    if not api_key:
+        return f"{provider_key}#nokey"
+    digest = hashlib.sha256(api_key.encode("utf-8")).hexdigest()[:12]
+    return f"{provider_key}#{digest}"
+
+
+def _read_catalog_cache(provider_key: str) -> list[tuple[str, str]] | None:
+    """Return a fresh cached catalog for ``provider_key``, or ``None``."""
+    if not _is_cacheable(provider_key):
+        return None
+    payload = _read_json(_catalog_cache_file())
+    if not isinstance(payload, dict):
+        return None
+    entry = payload.get(_cache_key(provider_key))
+    if not isinstance(entry, dict):
+        return None
+    # Fallback (negative) entries expire fast; dynamic ones live the full TTL.
+    ttl = _NEGATIVE_TTL if entry.get("source") == "fallback" else _CATALOG_TTL
+    if (time.time() - _as_float(entry.get("ts"))) >= ttl:
+        return None
+    models = entry.get("models")
+    if not isinstance(models, list) or not models:
+        return None
+    try:
+        return [(str(m[0]), str(m[1])) for m in models]
+    except (IndexError, TypeError):
+        return None
+
+
+def _write_catalog_cache(
+    provider_key: str, models: list[tuple[str, str]], *, source: str
+) -> None:
+    if not _is_cacheable(provider_key):
+        return
+    cache_file = _catalog_cache_file()
+    payload = _read_json(cache_file)
+    if not isinstance(payload, dict):
+        payload = {}
+    payload[_cache_key(provider_key)] = {
+        "ts": time.time(),
+        "source": source,
+        "models": [[model_id, label] for model_id, label in models],
+    }
+    # Best-effort cache write; a failure (e.g. read-only home) is non-fatal.
+    with contextlib.suppress(OSError):
+        cache_file.parent.mkdir(parents=True, exist_ok=True)
+        cache_file.write_text(json.dumps(payload), encoding="utf-8")

lib/cli/tests/test_model_catalog.py

@@ -0,0 +1,606 @@
+"""Tests for the dynamic model catalog used by the crew-creation wizard."""
+
+from __future__ import annotations
+
+import json
+import time
+
+import pytest
+
+import crewai_cli.model_catalog as mc
+
+_ALL_KEY_ENVS = [
+    "OPENAI_API_KEY",
+    "ANTHROPIC_API_KEY",
+    "GEMINI_API_KEY",
+    "GOOGLE_API_KEY",
+    "GROQ_API_KEY",
+    "CEREBRAS_API_KEY",
+    "OLLAMA_API_BASE",
+    "API_BASE",
+    "OLLAMA_HOST",
+]
+
+FALLBACK_ANTHROPIC = [
+    ("claude-opus-4-6", "Claude Opus 4.6"),
+    ("claude-sonnet-4-6", "Claude Sonnet 4.6"),
+]
+
+
+@pytest.fixture(autouse=True)
+def isolated_env(monkeypatch, tmp_path):
+    """Point the cache at a temp dir and clear provider keys for every test."""
+    monkeypatch.setattr(mc, "_cache_dir", lambda: tmp_path)
+    mc._reset_litellm_memo()  # clear the process-level LiteLLM memo per test
+    for key in _ALL_KEY_ENVS:
+        monkeypatch.delenv(key, raising=False)
+
+
+# ── version / label helpers ──────────────────────────────────────
+
+
+def test_version_key_parses_embedded_date():
+    date_int, version = mc._version_key("claude-3-5-sonnet-20241022")
+    assert date_int == 20241022
+    assert version == (3, 5)
+
+
+def test_version_key_parses_dashed_date():
+    date_int, _ = mc._version_key("gpt-4o-2024-08-06")
+    assert date_int == 20240806
+
+
+def test_version_key_version_only():
+    date_int, version = mc._version_key("claude-opus-4-6")
+    assert date_int == 0
+    assert version == (4, 6)
+
+
+def test_version_key_ranks_newer_higher():
+    older = mc._version_key("claude-sonnet-4-5")
+    newer = mc._version_key("claude-sonnet-4-6")
+    assert newer > older
+
+
+def test_is_chat_model_rejects_non_chat():
+    assert mc._is_chat_model("gpt-4.1-mini")
+    assert not mc._is_chat_model("text-embedding-3-large")
+    assert not mc._is_chat_model("whisper-1")
+    assert not mc._is_chat_model("dall-e-3")
+
+
+def test_search_substring_not_treated_as_non_chat():
+    # 'search' must not drop legitimate completion models: a token like
+    # *-search-preview, or 'research' (which contains 'search' as a substring).
+    assert mc._is_chat_model("gpt-4o-search-preview")
+    assert mc._is_chat_model("o3-deep-research")
+    # genuine non-chat markers still filter
+    assert not mc._is_chat_model("text-embedding-3-large")
+
+
+def test_humanize():
+    assert mc._humanize("gpt-4.1-mini") == "GPT 4.1 Mini"
+    assert mc._humanize("anthropic/claude-opus-4-6") == "Claude Opus 4 6"
+    # size suffixes uppercased, acronyms/brands cased, o-series preserved, ':' split
+    assert mc._humanize("openai/gpt-oss-120b") == "GPT OSS 120B"
+    assert mc._humanize("qwen/qwen3-32b") == "Qwen3 32B"
+    assert mc._humanize("deepseek-r1-distill-llama-70b") == "DeepSeek R1 Distill Llama 70B"
+    assert mc._humanize("o3-mini") == "o3 Mini"
+    assert mc._humanize("chatgpt-4o-latest") == "ChatGPT 4o Latest"
+    assert mc._humanize("llama3.3:70b") == "Llama3.3 70B"
+    assert mc._humanize("gemma2-9b-it") == "Gemma2 9B IT"
+
+
+# ── vendor tier ──────────────────────────────────────────────────
+
+
+def test_vendor_anthropic_ranks_by_date_and_uses_display_name(monkeypatch):
+    monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-test")
+    payload = {
+        "data": [
+            {
+                "id": "claude-3-5-sonnet-20240620",
+                "display_name": "Claude 3.5 Sonnet (old)",
+                "created_at": "2024-06-20T00:00:00Z",
+            },
+            {
+                "id": "claude-opus-4-6",
+                "display_name": "Claude Opus 4.6",
+                "created_at": "2026-02-01T00:00:00Z",
+            },
+            {
+                "id": "claude-haiku-4-5-20251001",
+                "display_name": "Claude Haiku 4.5",
+                "created_at": "2025-10-01T00:00:00Z",
+            },
+        ]
+    }
+    monkeypatch.setattr(mc, "_http_get_json", lambda *a, **k: payload)
+
+    models = mc.get_provider_models("anthropic", FALLBACK_ANTHROPIC)
+
+    # Newest first by created_at, display names preserved.
+    assert models[0] == ("claude-opus-4-6", "Claude Opus 4.6")
+    assert models[1] == ("claude-haiku-4-5-20251001", "Claude Haiku 4.5")
+    assert models[2] == ("claude-3-5-sonnet-20240620", "Claude 3.5 Sonnet (old)")
+
+
+def test_vendor_openai_filters_non_chat_models(monkeypatch):
+    monkeypatch.setenv("OPENAI_API_KEY", "sk-test")
+    payload = {
+        "data": [
+            {"id": "gpt-4.1", "created": 1_700_000_000},
+            {"id": "text-embedding-3-large", "created": 1_800_000_000},
+            {"id": "whisper-1", "created": 1_800_000_000},
+            {"id": "gpt-5.5", "created": 1_750_000_000},
+        ]
+    }
+    monkeypatch.setattr(mc, "_http_get_json", lambda *a, **k: payload)
+
+    models = mc.get_provider_models("openai", [])
+    ids = [m for m, _ in models]
+
+    assert ids == ["gpt-5.5", "gpt-4.1"]  # embeddings/whisper dropped, newest first
+
+
+def test_vendor_gemini_requires_generate_content(monkeypatch):
+    monkeypatch.setenv("GEMINI_API_KEY", "key")
+    payload = {
+        "models": [
+            {
+                "name": "models/gemini-2.5-pro",
+                "displayName": "Gemini 2.5 Pro",
+                "supportedGenerationMethods": ["generateContent"],
+            },
+            {
+                "name": "models/text-embedding-004",
+                "displayName": "Embedding",
+                "supportedGenerationMethods": ["embedContent"],
+            },
+            {
+                "name": "models/gemini-1.5-pro",
+                "displayName": "Gemini 1.5 Pro",
+                "supportedGenerationMethods": ["generateContent"],
+            },
+        ]
+    }
+    monkeypatch.setattr(mc, "_http_get_json", lambda *a, **k: payload)
+
+    models = mc.get_provider_models("gemini", [])
+    ids = [m for m, _ in models]
+
+    # "models/" prefix stripped, embedding excluded, newer version first.
+    assert ids == ["gemini-2.5-pro", "gemini-1.5-pro"]
+
+
+def test_openai_excludes_fine_tunes_and_checkpoints(monkeypatch):
+    # Fine-tunes/checkpoints have recent `created` timestamps and would otherwise
+    # crowd out (and rank above) the base models — they must be excluded so the
+    # picker shows clean foundation models.
+    monkeypatch.setenv("OPENAI_API_KEY", "sk-test")
+    payload = {
+        "data": [
+            {"id": "ft:gpt-4o-mini-2024-07-18:crewai::DyJG86uF", "created": 1_900_000_000},
+            {
+                "id": "ft:gpt-4o-mini-2024-07-18:crewai::DyJG7Q9N:ckpt-step-84",
+                "created": 1_900_000_001,
+            },
+            {"id": "gpt-5.5", "created": 1_800_000_000},
+            {"id": "gpt-4.1", "created": 1_700_000_000},
+        ]
+    }
+    monkeypatch.setattr(mc, "_http_get_json", lambda *a, **k: payload)
+
+    ids = [m for m, _ in mc.get_provider_models("openai", [])]
+    assert ids == ["gpt-5.5", "gpt-4.1"]  # fine-tunes + checkpoints dropped
+
+
+def test_vendor_gemini_paginates(monkeypatch):
+    monkeypatch.setenv("GEMINI_API_KEY", "key")
+    pages = {
+        None: {
+            "models": [
+                {
+                    "name": "models/gemini-3.5-flash",
+                    "displayName": "Gemini 3.5 Flash",
+                    "supportedGenerationMethods": ["generateContent"],
+                }
+            ],
+            "nextPageToken": "p2",
+        },
+        "p2": {
+            "models": [
+                {
+                    "name": "models/gemini-2.5-pro",
+                    "displayName": "Gemini 2.5 Pro",
+                    "supportedGenerationMethods": ["generateContent"],
+                }
+            ]
+        },
+    }
+
+    def fetch(url, headers=None, params=None):
+        return pages[(params or {}).get("pageToken")]
+
+    monkeypatch.setattr(mc, "_http_get_json", fetch)
+
+    ids = sorted(m for m, _ in mc.get_provider_models("gemini", []))
+    # Both pages contributed (newest-first ranking is _finalize's job).
+    assert ids == ["gemini-2.5-pro", "gemini-3.5-flash"]
+
+
+def test_vendor_gemini_first_page_error_uses_fallback(monkeypatch):
+    # A total (first-page) Gemini failure with a key set must fall back to the
+    # curated list, not be mistaken for a successful empty result.
+    monkeypatch.setenv("GEMINI_API_KEY", "key")
+
+    def boom(*a, **k):
+        raise RuntimeError("gemini down")
+
+    monkeypatch.setattr(mc, "_http_get_json", boom)
+    models = mc.get_provider_models("gemini", [("gemini-x", "Gemini X")])
+    assert models == [("gemini-x", "Gemini X")]
+
+
+def test_vendor_gemini_keeps_partial_on_later_page_error(monkeypatch):
+    monkeypatch.setenv("GEMINI_API_KEY", "key")
+
+    def fetch(url, headers=None, params=None):
+        if (params or {}).get("pageToken"):
+            raise RuntimeError("page 2 down")
+        return {
+            "models": [
+                {
+                    "name": "models/gemini-3.5-flash",
+                    "displayName": "Gemini 3.5 Flash",
+                    "supportedGenerationMethods": ["generateContent"],
+                }
+            ],
+            "nextPageToken": "p2",
+        }
+
+    monkeypatch.setattr(mc, "_http_get_json", fetch)
+
+    # Page-1 models are kept; the later-page error doesn't force the fallback.
+    models = mc.get_provider_models("gemini", [("fallback-x", "Fallback X")])
+    assert [m for m, _ in models] == ["gemini-3.5-flash"]
+
+
+def test_ollama_empty_response_not_filled_with_fallback(monkeypatch):
+    # A reachable Ollama with nothing installed -> empty (manual entry), not the
+    # curated suggestions the crew can't actually run.
+    monkeypatch.setattr(mc, "_http_get_json", lambda *a, **k: {"models": []})
+    assert mc.get_provider_models("ollama", [("llama3.3", "Llama 3.3")]) == []
+
+
+def test_ollama_unreachable_uses_fallback(monkeypatch):
+    # Server down (fetch raises) is different from empty -> fall back to suggestions.
+    def boom(*a, **k):
+        raise RuntimeError("connection refused")
+
+    monkeypatch.setattr(mc, "_http_get_json", boom)
+    models = mc.get_provider_models("ollama", [("llama3.3", "Llama 3.3")])
+    assert models == [("llama3.3", "Llama 3.3")]
+
+
+def test_ollama_excludes_embedding_models(monkeypatch):
+    # /api/tags lists everything installed, including embeddings — filter them.
+    monkeypatch.setattr(
+        mc,
+        "_http_get_json",
+        lambda *a, **k: {
+            "models": [
+                {"model": "llama3.3:70b"},
+                {"model": "nomic-embed-text"},
+                {"model": "mxbai-embed-large"},
+            ]
+        },
+    )
+    ids = [m for m, _ in mc.get_provider_models("ollama", [])]
+    assert ids == ["llama3.3:70b"]
+
+
+def test_ollama_base_honors_ollama_host(monkeypatch):
+    # OLLAMA_HOST (scheme-less runtime convention) is resolved with a scheme.
+    monkeypatch.setenv("OLLAMA_HOST", "10.0.0.5:11434")
+    assert mc._ollama_base() == "http://10.0.0.5:11434"
+
+
+def test_ollama_recovery_not_blocked_by_negative_cache(monkeypatch):
+    # Ollama down -> fallback, but not negatively cached; once the server is up
+    # the next call fetches live models rather than serving suggestions.
+    calls = {"n": 0}
+
+    def flaky(*a, **k):
+        calls["n"] += 1
+        if calls["n"] == 1:
+            raise RuntimeError("connection refused")
+        return {"models": [{"model": "llama-installed"}]}
+
+    monkeypatch.setattr(mc, "_http_get_json", flaky)
+    first = mc.get_provider_models("ollama", [("llama3.3", "Llama 3.3")])
+    assert first == [("llama3.3", "Llama 3.3")]  # down -> fallback (not cached)
+    second = mc.get_provider_models("ollama", [("llama3.3", "Llama 3.3")])
+    assert [m for m, _ in second] == ["llama-installed"]  # recovered live
+
+
+def test_gemini_honors_google_api_key(monkeypatch):
+    # GOOGLE_API_KEY (equivalent to GEMINI_API_KEY in crewai) enables the live tier.
+    monkeypatch.setenv("GOOGLE_API_KEY", "key")
+    monkeypatch.setattr(
+        mc,
+        "_http_get_json",
+        lambda *a, **k: {
+            "models": [
+                {
+                    "name": "models/gemini-3.5-flash",
+                    "displayName": "Gemini 3.5 Flash",
+                    "supportedGenerationMethods": ["generateContent"],
+                }
+            ]
+        },
+    )
+    models = mc.get_provider_models("gemini", [("gemini-x", "Gemini X")])
+    assert [m for m, _ in models] == ["gemini-3.5-flash"]  # live, not fallback
+
+
+def test_curated_label_overrides_raw_vendor_label(monkeypatch):
+    monkeypatch.setenv("OPENAI_API_KEY", "sk-test")
+    payload = {"data": [{"id": "gpt-5.5", "created": 1}]}
+    monkeypatch.setattr(mc, "_http_get_json", lambda *a, **k: payload)
+
+    models = mc.get_provider_models("openai", [("gpt-5.5", "GPT-5.5 (curated)")])
+    assert models == [("gpt-5.5", "GPT-5.5 (curated)")]
+
+
+def test_truncates_to_max_models(monkeypatch):
+    monkeypatch.setenv("OPENAI_API_KEY", "sk-test")
+    payload = {
+        "data": [{"id": f"gpt-test-{i}", "created": i} for i in range(20)]
+    }
+    monkeypatch.setattr(mc, "_http_get_json", lambda *a, **k: payload)
+
+    models = mc.get_provider_models("openai", [])
+    assert len(models) == mc.MAX_MODELS
+
+
+# ── litellm tier ─────────────────────────────────────────────────
+
+
+def test_litellm_tier_for_uncurated_provider(monkeypatch):
+    # A provider with no curated fallback ([]) -> the LiteLLM feed is consulted.
+    litellm_data = {
+        "claude-opus-4-6": {"litellm_provider": "anthropic", "mode": "chat"},
+        "claude-sonnet-4-5": {"litellm_provider": "anthropic", "mode": "chat"},
+        "voyage-embed": {"litellm_provider": "anthropic", "mode": "embedding"},
+        "gpt-4.1": {"litellm_provider": "openai", "mode": "chat"},
+    }
+    mc._litellm_cache_file().write_text(json.dumps(litellm_data), encoding="utf-8")
+
+    models = mc.get_provider_models("anthropic", [])  # empty == uncurated
+    ids = [m for m, _ in models]
+
+    # Only anthropic chat models, embedding + other providers excluded.
+    assert ids == ["claude-opus-4-6", "claude-sonnet-4-5"]
+
+
+def test_null_litellm_provider_does_not_crash(monkeypatch):
+    # A present-but-null litellm_provider must be skipped, not raise.
+    litellm_data = {
+        "weird-model": {"litellm_provider": None, "mode": "chat"},
+        "anthropic.claude-v2": {"litellm_provider": "bedrock", "mode": "chat"},
+    }
+    mc._litellm_cache_file().write_text(json.dumps(litellm_data), encoding="utf-8")
+
+    models = mc.get_provider_models("bedrock", [])
+    assert [m for m, _ in models] == ["anthropic.claude-v2"]
+
+
+def test_litellm_strips_provider_prefix(monkeypatch):
+    litellm_data = {
+        "gemini/gemini-1.5-pro": {"litellm_provider": "gemini", "mode": "chat"},
+    }
+    mc._litellm_cache_file().write_text(json.dumps(litellm_data), encoding="utf-8")
+
+    models = mc.get_provider_models("gemini", [])
+    assert models == [("gemini-1.5-pro", "Gemini 1.5 Pro")]
+
+
+# ── fallback + caching ───────────────────────────────────────────
+
+
+def test_falls_back_when_everything_fails(monkeypatch):
+    # No key, no litellm cache, network raises -> curated fallback verbatim.
+    def boom(*a, **k):
+        raise RuntimeError("network down")
+
+    monkeypatch.setattr(mc, "_http_get_json", boom)
+    models = mc.get_provider_models("anthropic", FALLBACK_ANTHROPIC)
+    assert models == FALLBACK_ANTHROPIC
+
+
+def test_result_is_cached(monkeypatch):
+    monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-test")
+    calls = {"n": 0}
+
+    def fetch(*a, **k):
+        calls["n"] += 1
+        return {"data": [{"id": "claude-opus-4-6", "created_at": "2026-01-01T00:00:00Z"}]}
+
+    monkeypatch.setattr(mc, "_http_get_json", fetch)
+
+    first = mc.get_provider_models("anthropic", FALLBACK_ANTHROPIC)
+    # Second call must hit the cache and not touch the network again.
+    monkeypatch.setattr(mc, "_http_get_json", lambda *a, **k: pytest.fail("refetched"))
+    second = mc.get_provider_models("anthropic", FALLBACK_ANTHROPIC)
+
+    assert first == second
+    assert calls["n"] == 1
+
+
+def test_curated_fallback_preferred_over_litellm(monkeypatch):
+    # The feed lags real releases, so a non-empty curated fallback must win even
+    # when a fresh LiteLLM cache is present (regression: Anthropic's feed lacked
+    # Fable 5 / Opus 4.8 / Sonnet 5).
+    monkeypatch.setattr(mc, "_http_get_json", lambda *a, **k: pytest.fail("no net"))
+    litellm_data = {
+        "claude-opus-4-6": {"litellm_provider": "anthropic", "mode": "chat"},
+    }
+    mc._litellm_cache_file().write_text(json.dumps(litellm_data), encoding="utf-8")
+
+    models = mc.get_provider_models("anthropic", FALLBACK_ANTHROPIC)
+    assert models == FALLBACK_ANTHROPIC
+
+
+def test_added_key_bypasses_negative_cache(monkeypatch):
+    # A no-key call negatively-caches the fallback; adding a key afterwards must
+    # fetch live models rather than serve the cached fallback (distinct cache key).
+    first = mc.get_provider_models("openai", [("gpt-x", "GPT X")])
+    assert first == [("gpt-x", "GPT X")]  # no key -> fallback
+
+    monkeypatch.setenv("OPENAI_API_KEY", "sk-test")
+    monkeypatch.setattr(
+        mc, "_http_get_json", lambda *a, **k: {"data": [{"id": "gpt-5.5", "created": 1}]}
+    )
+    second = mc.get_provider_models("openai", [("gpt-x", "GPT X")])
+    assert [m for m, _ in second] == ["gpt-5.5"]  # live fetch, not cached fallback
+
+
+def test_invalid_litellm_cache_falls_through_to_download(monkeypatch):
+    # A corrupt-but-fresh cache must neither crash the picker nor block a
+    # recoverable download — it falls through and refetches.
+    mc._litellm_cache_file().write_text("[1, 2, 3]", encoding="utf-8")
+    monkeypatch.setattr(
+        mc,
+        "_http_get_json",
+        lambda *a, **k: {
+            "anthropic.claude-v2": {"litellm_provider": "bedrock", "mode": "chat"}
+        },
+    )
+    models = mc.get_provider_models("bedrock", [])
+    assert [m for m, _ in models] == ["anthropic.claude-v2"]  # recovered via download
+
+
+def test_litellm_fetch_attempted_once_per_process(monkeypatch):
+    # With no cache and a failing download, the feed is fetched at most once per
+    # process — repeated lookups (across providers) must not re-hit the network.
+    calls = {"n": 0}
+
+    def boom(*a, **k):
+        calls["n"] += 1
+        raise RuntimeError("offline")
+
+    monkeypatch.setattr(mc, "_http_get_json", boom)
+    mc.get_provider_models("bedrock", [])
+    mc.get_provider_models("azure", [])
+    assert calls["n"] == 1  # memoized after the first failed attempt
+
+
+def test_litellm_fills_uncurated_bedrock(monkeypatch):
+    # No vendor fetcher and no curated fallback -> LiteLLM feed fills the gap.
+    monkeypatch.setattr(mc, "_http_get_json", lambda *a, **k: pytest.fail("no net"))
+    litellm_data = {
+        "anthropic.claude-v2": {"litellm_provider": "bedrock", "mode": "chat"},
+    }
+    mc._litellm_cache_file().write_text(json.dumps(litellm_data), encoding="utf-8")
+
+    models = mc.get_provider_models("bedrock", [])
+    assert models == [("anthropic.claude-v2", "Anthropic.claude V2")]
+
+
+def test_failed_fetch_is_negatively_cached(monkeypatch):
+    # A failed vendor fetch must not be retried on every call — the fallback is
+    # cached briefly so the picker doesn't re-hit the timeout-prone endpoint.
+    monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-test")
+    calls = {"n": 0}
+
+    def boom(*a, **k):
+        calls["n"] += 1
+        raise RuntimeError("down")
+
+    monkeypatch.setattr(mc, "_http_get_json", boom)
+    first = mc.get_provider_models("anthropic", FALLBACK_ANTHROPIC)
+    second = mc.get_provider_models("anthropic", FALLBACK_ANTHROPIC)
+
+    assert first == second == FALLBACK_ANTHROPIC
+    assert calls["n"] == 1  # second call served from the negative cache
+
+
+def test_bad_cache_json_does_not_crash(monkeypatch):
+    # A corrupt cache whose root is not a mapping must not raise (get_provider_models
+    # is documented to never raise).
+    mc._catalog_cache_file().write_text("[1, 2, 3]", encoding="utf-8")
+
+    models = mc.get_provider_models("anthropic", FALLBACK_ANTHROPIC)
+    assert models == FALLBACK_ANTHROPIC
+
+
+def test_ollama_is_not_cached_reflects_installed_changes(monkeypatch):
+    # Ollama is local and never cached: the picker re-probes /api/tags on every
+    # call, so a model deleted locally drops out immediately (no stale entry).
+    responses = iter(
+        [
+            {"models": [{"model": "llama3.3"}, {"model": "qwen3"}]},
+            {"models": [{"model": "llama3.3"}]},  # qwen3 deleted between calls
+        ]
+    )
+    monkeypatch.setattr(mc, "_http_get_json", lambda *a, **k: next(responses))
+
+    first = {m for m, _ in mc.get_provider_models("ollama", [])}
+    second = {m for m, _ in mc.get_provider_models("ollama", [])}
+
+    assert first == {"llama3.3", "qwen3"}
+    assert second == {"llama3.3"}  # re-probed, not served from a stale cache
+
+
+def test_ollama_never_written_to_catalog_cache(monkeypatch):
+    monkeypatch.setattr(
+        mc, "_http_get_json", lambda *a, **k: {"models": [{"model": "llama3.3"}]}
+    )
+    mc.get_provider_models("ollama", [])
+    assert not mc._catalog_cache_file().exists()
+
+
+def test_different_api_key_uses_separate_cache_entry(monkeypatch):
+    # Cache is keyed by the exact key: switching keys must refetch, not serve
+    # the previous account's cached list.
+    monkeypatch.setenv("OPENAI_API_KEY", "sk-account-A")
+    monkeypatch.setattr(
+        mc, "_http_get_json", lambda *a, **k: {"data": [{"id": "gpt-5.5", "created": 1}]}
+    )
+    assert [m for m, _ in mc.get_provider_models("openai", [])] == ["gpt-5.5"]
+
+    monkeypatch.setenv("OPENAI_API_KEY", "sk-account-B")
+    monkeypatch.setattr(
+        mc, "_http_get_json", lambda *a, **k: {"data": [{"id": "gpt-4.1", "created": 1}]}
+    )
+    # New key -> distinct cache entry -> refetch, not the account-A cache.
+    assert [m for m, _ in mc.get_provider_models("openai", [])] == ["gpt-4.1"]
+
+
+def test_cache_key_hashes_key_and_never_stores_it(monkeypatch):
+    monkeypatch.setenv("OPENAI_API_KEY", "sk-super-secret")
+    key = mc._cache_key("openai")
+    assert key.startswith("openai#") and key != "openai#nokey"
+    assert "sk-super-secret" not in key  # only a digest, never the raw key
+
+
+def test_dynamic_cache_expires_after_catalog_ttl(monkeypatch):
+    # A dynamic entry older than the (now short) catalog TTL is not served.
+    monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-test")
+    entry_key = mc._cache_key("anthropic")
+    stale_ts = time.time() - (mc._CATALOG_TTL + 5)
+    mc._catalog_cache_file().write_text(
+        json.dumps(
+            {
+                entry_key: {
+                    "ts": stale_ts,
+                    "source": "dynamic",
+                    "models": [["stale-model", "Stale"]],
+                }
+            }
+        ),
+        encoding="utf-8",
+    )
+
+    assert mc._read_catalog_cache("anthropic") is None

lib/cli/tests/test_plus_api.py

@@ -1,8 +1,6 @@
 import os
 import unittest
-from unittest.mock import ANY, AsyncMock, MagicMock, patch
-
-import pytest
+from unittest.mock import ANY, MagicMock, patch
 
 from crewai_cli.plus_api import PlusAPI
 
@@ -343,28 +341,23 @@ class TestPlusAPI(unittest.TestCase):
         )
 
 
-@pytest.mark.asyncio
-@patch("httpx.AsyncClient")
-async def test_get_agent(mock_async_client_class):
+@patch("crewai_core.plus_api.PlusAPI._make_request")
+def test_get_agent(mock_make_request):
     api = PlusAPI("test_api_key")
     mock_response = MagicMock()
-    mock_client_instance = AsyncMock()
-    mock_client_instance.get.return_value = mock_response
-    mock_async_client_class.return_value.__aenter__.return_value = mock_client_instance
+    mock_make_request.return_value = mock_response
 
-    response = await api.get_agent("test_agent_handle")
+    response = api.get_agent("test_agent_handle")
 
-    mock_client_instance.get.assert_called_once_with(
-        f"{api.base_url}/crewai_plus/api/v1/agents/test_agent_handle",
-        headers=api.headers,
+    mock_make_request.assert_called_once_with(
+        "GET", "/crewai_plus/api/v1/agents/test_agent_handle"
     )
     assert response == mock_response
 
 
-@pytest.mark.asyncio
-@patch("httpx.AsyncClient")
+@patch("crewai_core.plus_api.PlusAPI._make_request")
 @patch("crewai_core.plus_api.Settings")
-async def test_get_agent_with_org_uuid(mock_settings_class, mock_async_client_class):
+def test_get_agent_with_org_uuid(mock_settings_class, mock_make_request):
     org_uuid = "test-org-uuid"
     mock_settings = MagicMock()
     mock_settings.org_uuid = org_uuid
@@ -374,15 +367,12 @@ async def test_get_agent_with_org_uuid(mock_settings_class, mock_async_client_cl
     api = PlusAPI("test_api_key")
 
     mock_response = MagicMock()
-    mock_client_instance = AsyncMock()
-    mock_client_instance.get.return_value = mock_response
-    mock_async_client_class.return_value.__aenter__.return_value = mock_client_instance
+    mock_make_request.return_value = mock_response
 
-    response = await api.get_agent("test_agent_handle")
+    response = api.get_agent("test_agent_handle")
 
-    mock_client_instance.get.assert_called_once_with(
-        f"{api.base_url}/crewai_plus/api/v1/agents/test_agent_handle",
-        headers=api.headers,
+    mock_make_request.assert_called_once_with(
+        "GET", "/crewai_plus/api/v1/agents/test_agent_handle"
     )
     assert "X-Crewai-Organization-Id" in api.headers
     assert api.headers["X-Crewai-Organization-Id"] == org_uuid

lib/crewai-core/src/crewai_core/plus_api.py

@@ -232,10 +232,8 @@ class PlusAPI:
     def get_tool(self, handle: str) -> httpx.Response:
         return self._make_request("GET", f"{self.TOOLS_RESOURCE}/{handle}")
 
-    async def get_agent(self, handle: str) -> httpx.Response:
-        url = urljoin(self.base_url, f"{self.AGENTS_RESOURCE}/{handle}")
-        async with httpx.AsyncClient() as client:
-            return await client.get(url, headers=cast(dict[str, str], self.headers))
+    def get_agent(self, handle: str) -> httpx.Response:
+        return self._make_request("GET", f"{self.AGENTS_RESOURCE}/{handle}")
 
     def publish_tool(
         self,

lib/crewai-core/src/crewai_core/telemetry.py

@@ -264,10 +264,12 @@ class Telemetry:
 
     def flow_creation_span(self, flow_name: str) -> None:
         """Records the creation of a new flow."""
+        from crewai_core.version import get_crewai_version
 
         def _operation() -> None:
             tracer = trace.get_tracer("crewai.telemetry")
             span = tracer.start_span("Flow Creation")
+            self._add_attribute(span, "crewai_version", get_crewai_version())
             self._add_attribute(span, "flow_name", flow_name)
             close_span(span)
 

lib/crewai-core/tests/test_smoke.py

@@ -233,3 +233,31 @@ def test_core_telemetry_records_feature_usage(
     tracer.start_span.assert_called_once_with("Feature Usage")
     span.set_attribute.assert_any_call("feature", "cli_usage:view_traces")
     span.end.assert_called_once()
+
+
+def test_core_telemetry_records_flow_creation_version(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from crewai_core.telemetry import Telemetry
+
+    Telemetry._instance = None
+    monkeypatch.delenv("OTEL_SDK_DISABLED", raising=False)
+    monkeypatch.delenv("CREWAI_DISABLE_TELEMETRY", raising=False)
+    monkeypatch.delenv("CREWAI_DISABLE_TRACKING", raising=False)
+    monkeypatch.setattr("crewai_core.version.get_crewai_version", lambda: "1.0.0")
+
+    tracer = Mock()
+    span = Mock()
+    tracer.start_span.return_value = span
+    monkeypatch.setattr(
+        "crewai_core.telemetry.trace.get_tracer",
+        lambda _name: tracer,
+    )
+
+    telemetry = Telemetry()
+    telemetry.flow_creation_span("ResearchFlow")
+
+    tracer.start_span.assert_called_once_with("Flow Creation")
+    span.set_attribute.assert_any_call("crewai_version", "1.0.0")
+    span.set_attribute.assert_any_call("flow_name", "ResearchFlow")
+    span.end.assert_called_once()

lib/crewai-tools/src/crewai_tools/tools/tavily_extractor_tool/tavily_extractor_tool.py

@@ -87,9 +87,11 @@ class TavilyExtractorTool(BaseTool):
         """
         super().__init__(**kwargs)
         if TAVILY_AVAILABLE:
-            self.client = TavilyClient(api_key=self.api_key, proxies=self.proxies)
+            self.client = TavilyClient(
+                api_key=self.api_key, proxies=self.proxies, client_name="crewai"
+            )
             self.async_client = AsyncTavilyClient(
-                api_key=self.api_key, proxies=self.proxies
+                api_key=self.api_key, proxies=self.proxies, client_name="crewai"
             )
         else:
             try:

lib/crewai-tools/src/crewai_tools/tools/tavily_get_research_tool/tavily_get_research_tool.py

@@ -54,8 +54,10 @@ class TavilyGetResearchTool(BaseTool):
         super().__init__(**kwargs)
         if TAVILY_AVAILABLE:
             api_key = os.getenv("TAVILY_API_KEY")
-            self._client = TavilyClient(api_key=api_key)
-            self._async_client = AsyncTavilyClient(api_key=api_key)
+            self._client = TavilyClient(api_key=api_key, client_name="crewai")
+            self._async_client = AsyncTavilyClient(
+                api_key=api_key, client_name="crewai"
+            )
         else:
             try:
                 import subprocess

lib/crewai-tools/src/crewai_tools/tools/tavily_research_tool/tavily_research_tool.py

@@ -90,8 +90,10 @@ class TavilyResearchTool(BaseTool):
         super().__init__(**kwargs)
         if TAVILY_AVAILABLE:
             api_key = os.getenv("TAVILY_API_KEY")
-            self._client = TavilyClient(api_key=api_key)
-            self._async_client = AsyncTavilyClient(api_key=api_key)
+            self._client = TavilyClient(api_key=api_key, client_name="crewai")
+            self._async_client = AsyncTavilyClient(
+                api_key=api_key, client_name="crewai"
+            )
         else:
             try:
                 import subprocess

lib/crewai-tools/src/crewai_tools/tools/tavily_search_tool/tavily_search_tool.py

@@ -115,9 +115,11 @@ class TavilySearchTool(BaseTool):
     def __init__(self, **kwargs: Any):
         super().__init__(**kwargs)
         if TAVILY_AVAILABLE:
-            self.client = TavilyClient(api_key=self.api_key, proxies=self.proxies)
+            self.client = TavilyClient(
+                api_key=self.api_key, proxies=self.proxies, client_name="crewai"
+            )
             self.async_client = AsyncTavilyClient(
-                api_key=self.api_key, proxies=self.proxies
+                api_key=self.api_key, proxies=self.proxies, client_name="crewai"
             )
         else:
             try:

lib/crewai/src/crewai/experimental/agent_executor.py

@@ -106,6 +106,7 @@ from crewai.utilities.planning_types import (
     TodoItem,
     TodoList,
 )
+from crewai.utilities.prompts import StandardPromptResult, SystemPromptResult
 from crewai.utilities.step_execution_context import StepExecutionContext, StepResult
 from crewai.utilities.string_utils import sanitize_tool_name
 from crewai.utilities.tool_utils import execute_tool_and_check_finality
@@ -118,7 +119,6 @@ if TYPE_CHECKING:
     from crewai.agents.tools_handler import ToolsHandler
     from crewai.llms.base_llm import BaseLLM
     from crewai.tools.tool_types import ToolResult
-    from crewai.utilities.prompts import StandardPromptResult, SystemPromptResult
 
 _RouteT = TypeVar("_RouteT", bound=str)
 
@@ -218,6 +218,7 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
     _instance_id: str = PrivateAttr(default_factory=lambda: str(uuid4())[:8])
     _step_executor: Any = PrivateAttr(default=None)
     _planner_observer: PlannerObserver | None = PrivateAttr(default=None)
+    _is_feedback_iteration: bool = PrivateAttr(default=False)
 
     @model_validator(mode="after")
     def _setup_executor(self) -> Self:
@@ -296,6 +297,33 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
         """Set state messages."""
         self._state.messages = value
 
+    def _setup_messages(self, inputs: dict[str, Any]) -> None:
+        """Set up messages for the agent execution."""
+        provider = get_provider()
+        if provider.setup_messages(cast("ExecutorContext", self)):
+            return
+
+        from crewai.llms.cache import mark_cache_breakpoint
+
+        if isinstance(self.prompt, SystemPromptResult):
+            system_prompt = self._format_prompt(self.prompt["system"], inputs)
+            user_prompt = self._format_prompt(self.prompt["user"], inputs)
+            self.state.messages.append(
+                mark_cache_breakpoint(
+                    format_message_for_llm(system_prompt, role="system")
+                )
+            )
+            self.state.messages.append(
+                mark_cache_breakpoint(format_message_for_llm(user_prompt))
+            )
+        elif isinstance(self.prompt, StandardPromptResult):
+            user_prompt = self._format_prompt(self.prompt["prompt"], inputs)
+            self.state.messages.append(
+                mark_cache_breakpoint(format_message_for_llm(user_prompt))
+            )
+
+        provider.post_setup_messages(cast("ExecutorContext", self))
+
     @property
     def ask_for_human_input(self) -> bool:
         """Compatibility property - returns state ask_for_human_input."""
@@ -314,6 +342,8 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
         enabled on the agent, it generates a plan before execution begins.
         The plan is stored in state and todos are created from the steps.
         """
+        if self._is_feedback_iteration:
+            return
         if not getattr(self.agent, "planning_enabled", False):
             return
 
@@ -2761,27 +2791,7 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
                     "AgentExecutor.llm or .prompt is unset; the executor was "
                     "not fully restored or initialized before execution."
                 )
-            if "system" in self.prompt:
-                from crewai.llms.cache import mark_cache_breakpoint
-
-                prompt = cast("SystemPromptResult", self.prompt)
-                system_prompt = self._format_prompt(prompt["system"], inputs)
-                user_prompt = self._format_prompt(prompt["user"], inputs)
-                self.state.messages.append(
-                    mark_cache_breakpoint(
-                        format_message_for_llm(system_prompt, role="system")
-                    )
-                )
-                self.state.messages.append(
-                    mark_cache_breakpoint(format_message_for_llm(user_prompt))
-                )
-            else:
-                from crewai.llms.cache import mark_cache_breakpoint
-
-                user_prompt = self._format_prompt(self.prompt["prompt"], inputs)
-                self.state.messages.append(
-                    mark_cache_breakpoint(format_message_for_llm(user_prompt))
-                )
+            self._setup_messages(inputs)
 
             self._inject_files_from_inputs(inputs)
 
@@ -2867,27 +2877,7 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
                     "AgentExecutor.llm or .prompt is unset; the executor was "
                     "not fully restored or initialized before execution."
                 )
-            if "system" in self.prompt:
-                from crewai.llms.cache import mark_cache_breakpoint
-
-                prompt = cast("SystemPromptResult", self.prompt)
-                system_prompt = self._format_prompt(prompt["system"], inputs)
-                user_prompt = self._format_prompt(prompt["user"], inputs)
-                self.state.messages.append(
-                    mark_cache_breakpoint(
-                        format_message_for_llm(system_prompt, role="system")
-                    )
-                )
-                self.state.messages.append(
-                    mark_cache_breakpoint(format_message_for_llm(user_prompt))
-                )
-            else:
-                from crewai.llms.cache import mark_cache_breakpoint
-
-                user_prompt = self._format_prompt(self.prompt["prompt"], inputs)
-                self.state.messages.append(
-                    mark_cache_breakpoint(format_message_for_llm(user_prompt))
-                )
+            self._setup_messages(inputs)
 
             await self._ainject_files_from_inputs(inputs)
 
@@ -3169,8 +3159,13 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
         Returns:
             Final answer after feedback.
         """
+        self.messages = self.state.messages
         provider = get_provider()
-        return provider.handle_feedback(formatted_answer, cast("ExecutorContext", self))
+        final_answer = provider.handle_feedback(
+            formatted_answer, cast("ExecutorContext", self)
+        )
+        self._complete_feedback(final_answer)
+        return final_answer
 
     async def _ahandle_human_feedback(
         self, formatted_answer: AgentFinish
@@ -3183,10 +3178,63 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
         Returns:
             Final answer after feedback.
         """
+        self.messages = self.state.messages
         provider = get_provider()
-        return await provider.handle_feedback_async(
+        final_answer = await provider.handle_feedback_async(
             formatted_answer, cast("AsyncExecutorContext", self)
         )
+        self._complete_feedback(final_answer)
+        return final_answer
+
+    def _complete_feedback(self, final_answer: AgentFinish) -> None:
+        """Mark the final reviewed answer as the completed executor state."""
+        self.state.current_answer = final_answer
+        self.state.is_finished = True
+        self.state.ask_for_human_input = False
+        self._finalize_called = True
+
+    def _prepare_feedback_iteration(self) -> None:
+        """Reset flow completion state before rerunning with feedback."""
+        self._finalize_called = False
+        self._is_feedback_iteration = True
+        self.state.current_answer = None
+        self.state.is_finished = False
+        self.state.iterations = 0
+        self.state.use_native_tools = False
+        self.state.pending_tool_calls = []
+        self.state.plan = None
+        self.state.plan_ready = False
+        self.state.todos = TodoList()
+        self.state.replan_count = 0
+        self.state.last_replan_reason = None
+        self.state.observations = {}
+        self.state.execution_log = []
+
+    def _invoke_loop(self) -> AgentFinish:
+        """Re-run the executor flow using the existing feedback messages."""
+        self._prepare_feedback_iteration()
+        try:
+            self.kickoff()
+        finally:
+            self._is_feedback_iteration = False
+
+        if not isinstance(self.state.current_answer, AgentFinish):
+            raise RuntimeError("Agent execution ended without reaching a final answer.")
+
+        return self.state.current_answer
+
+    async def _ainvoke_loop(self) -> AgentFinish:
+        """Re-run the executor flow asynchronously using feedback messages."""
+        self._prepare_feedback_iteration()
+        try:
+            await self.kickoff_async()
+        finally:
+            self._is_feedback_iteration = False
+
+        if not isinstance(self.state.current_answer, AgentFinish):
+            raise RuntimeError("Agent execution ended without reaching a final answer.")
+
+        return self.state.current_answer
 
     def _is_training_mode(self) -> bool:
         """Check if training mode is active.
@@ -3196,6 +3244,12 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
         """
         return bool(self.crew and self.crew._train)
 
+    def _format_feedback_message(self, feedback: str) -> LLMMessage:
+        """Format human feedback as an LLM message."""
+        return format_message_for_llm(
+            I18N_DEFAULT.slice("feedback_instructions").format(feedback=feedback)
+        )
+
 
 # Backward compatibility alias (deprecated)
 CrewAgentExecutorFlow = AgentExecutor

lib/crewai/src/crewai/flow/runtime/_actions.py

@@ -138,11 +138,12 @@ class CrewAction:
 
         local_context = _pop_local_context(kwargs)
         if self.definition.from_declaration is not None:
-            crew, default_inputs = load_crew(
+            crew, default_inputs = await asyncio.to_thread(
+                load_crew,
                 _resolve_crew_declaration(
                     self.definition.from_declaration,
                     base_dir=self.flow._definition.source_dir,
-                )
+                ),
             )
             input_template = {**default_inputs, **(self.definition.inputs or {})}
         else:
@@ -155,7 +156,9 @@ class CrewAction:
                 **crew_definition.inputs,
                 **(self.definition.inputs or {}),
             }
-            crew, _ = load_crew_from_definition(crew_definition, source="crew action")
+            crew, _ = await asyncio.to_thread(
+                load_crew_from_definition, crew_definition, source="crew action"
+            )
 
         inputs = Expression.from_flow(
             cast(ExpressionData, input_template),
@@ -184,7 +187,8 @@ class AgentAction:
         if not isinstance(rendered_input, str):
             raise ValueError("agent input must render to a string")
 
-        agent, response_format = load_agent_from_definition(
+        agent, response_format = await asyncio.to_thread(
+            load_agent_from_definition,
             self.definition.with_,
             source="agent action",
         )

lib/crewai/src/crewai/project/crew_definition.py

@@ -66,21 +66,24 @@ class CrewAgentDefinition(BaseModel):
 
     model_config = ConfigDict(extra="allow")
 
-    role: str = Field(
+    role: str | None = Field(
+        default=None,
         description=(
             "Crew agent role. Crew inputs are interpolated with `{name}` "
             "placeholders such as `{topic}`; this is not CEL."
         ),
         examples=["Research analyst"],
     )
-    goal: str = Field(
+    goal: str | None = Field(
+        default=None,
         description=(
             "Crew agent goal. Crew inputs are interpolated with `{name}` "
             "placeholders such as `{topic}`; this is not CEL."
         ),
         examples=["Research {topic}"],
     )
-    backstory: str = Field(
+    backstory: str | None = Field(
+        default=None,
         description=(
             "Crew agent backstory. Crew inputs are interpolated with `{name}` "
             "placeholders such as `{topic}`; this is not CEL."
@@ -92,6 +95,15 @@ class CrewAgentDefinition(BaseModel):
         description="Optional built-in type or Python reference used to load the agent.",
         examples=["agent", {"python": "my_project.agents.ResearchAgent"}],
     )
+    from_repository: str | None = Field(
+        default=None,
+        description=(
+            "Agent repository name to load. Repository values supply missing "
+            "agent configuration; explicitly provided local fields override the "
+            "repository values."
+        ),
+        examples=["researcher"],
+    )
     settings: dict[str, Any] = Field(
         default_factory=dict,
         description="Additional agent settings passed to the loader.",
@@ -183,15 +195,18 @@ class CrewAgentDefinition(BaseModel):
 class AgentDefinition(CrewAgentDefinition):
     """Inline individual agent definition used outside of a crew."""
 
-    role: str = Field(
+    role: str | None = Field(
+        default=None,
         description="Individual agent role used by a Flow agent action outside of a crew.",
         examples=["Support specialist"],
     )
-    goal: str = Field(
+    goal: str | None = Field(
+        default=None,
         description="Individual agent goal for the Flow agent action outside of a crew.",
         examples=["Draft a concise customer reply"],
     )
-    backstory: str = Field(
+    backstory: str | None = Field(
+        default=None,
         description=(
             "Individual agent backstory used to shape behavior outside of a crew."
         ),

lib/crewai/src/crewai/project/json_loader.py

@@ -978,9 +978,10 @@ def _agent_kwargs_from_definition(
         extra_allowed,
         skip_unknown=skip_unknown,
     )
-    for required in ("role", "goal", "backstory"):
-        if required not in defn:
-            errors.append(f"{path}: missing required field '{required}'")
+    if not defn.get("from_repository"):
+        for required in ("role", "goal", "backstory"):
+            if defn.get(required) is None:
+                errors.append(f"{path}: missing required field '{required}'")
 
     settings = defn.get("settings", {})
     if settings is None:

lib/crewai/src/crewai/telemetry/telemetry.py

@@ -949,6 +949,7 @@ class Telemetry:
         def _operation() -> None:
             tracer = trace.get_tracer("crewai.telemetry")
             span = tracer.start_span("Flow Creation")
+            self._add_attribute(span, "crewai_version", version("crewai"))
             self._add_attribute(span, "flow_name", flow_name)
             close_span(span)
 

lib/crewai/src/crewai/utilities/agent_utils.py

@@ -1125,7 +1125,7 @@ def load_agent_from_repository(from_repository: str) -> dict[str, Any]:
 
             client = PlusAPI(api_key=get_auth_token())
         _print_current_organization()
-        response = asyncio.run(client.get_agent(from_repository))
+        response = client.get_agent(from_repository)
         if response.status_code == 404:
             raise AgentRepositoryError(
                 f"Agent {from_repository} does not exist, make sure the name is correct or the agent is available on your organization."
@@ -1158,6 +1158,8 @@ def load_agent_from_repository(from_repository: str) -> dict[str, Any]:
                         raise AgentRepositoryError(
                             f"Tool {tool['name']} could not be loaded: {e}"
                         ) from e
+            elif key == "skills" and value == []:
+                continue
             else:
                 attributes[key] = value
     return attributes

lib/crewai/tests/agents/test_agent.py

@@ -3,13 +3,14 @@
 import os
 import threading
 from unittest import mock
-from unittest.mock import MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 import warnings
 
 from crewai.agents.crew_agent_executor import AgentFinish, CrewAgentExecutor
 from crewai.constants import DEFAULT_LLM_MODEL
 from crewai.events.event_bus import crewai_event_bus
 from crewai.events.types.tool_usage_events import ToolUsageFinishedEvent
+from crewai.experimental.agent_executor import AgentExecutor
 from crewai.knowledge.knowledge import Knowledge
 from crewai.knowledge.knowledge_config import KnowledgeConfig
 from crewai.knowledge.source.base_knowledge_source import BaseKnowledgeSource
@@ -802,6 +803,97 @@ def test_agent_human_input():
         assert output.strip().lower() == "hello"
 
 
+def test_agent_default_executor_human_input():
+    from crewai.core.providers.human_input import SyncHumanInputProvider
+
+    agent = Agent(
+        role="test role",
+        goal="test goal",
+        backstory="test backstory",
+    )
+    task = Task(
+        agent=agent,
+        description="Say the word: Hi",
+        expected_output="The word: Hi",
+        human_input=True,
+    )
+    answers = iter(
+        [
+            AgentFinish(output="Hi", thought="", text="Hi"),
+            AgentFinish(output="Hello", thought="", text="Hello"),
+        ]
+    )
+    feedback_responses = iter(["Don't say hi, say Hello instead!", ""])
+
+    def kickoff_side_effect(executor, *_args, **_kwargs):
+        executor.state.current_answer = next(answers)
+        executor.state.is_finished = True
+
+    with (
+        patch.object(
+            SyncHumanInputProvider,
+            "_prompt_input",
+            side_effect=lambda *_args, **_kwargs: next(feedback_responses),
+        ) as mock_prompt_input,
+        patch.object(
+            AgentExecutor, "kickoff", autospec=True, side_effect=kickoff_side_effect
+        ) as mock_kickoff,
+    ):
+        output = agent.execute_task(task)
+
+    assert output == "Hello"
+    assert mock_prompt_input.call_count == 2
+    assert mock_kickoff.call_count == 2
+
+
+@pytest.mark.asyncio
+async def test_agent_default_executor_async_human_input():
+    from crewai.core.providers.human_input import SyncHumanInputProvider
+
+    agent = Agent(
+        role="test role",
+        goal="test goal",
+        backstory="test backstory",
+    )
+    task = Task(
+        agent=agent,
+        description="Say the word: Hi",
+        expected_output="The word: Hi",
+        human_input=True,
+    )
+    answers = iter(
+        [
+            AgentFinish(output="Hi", thought="", text="Hi"),
+            AgentFinish(output="Hello", thought="", text="Hello"),
+        ]
+    )
+    feedback_responses = iter(["Don't say hi, say Hello instead!", ""])
+
+    async def kickoff_side_effect(executor, *_args, **_kwargs):
+        executor.state.current_answer = next(answers)
+        executor.state.is_finished = True
+
+    with (
+        patch.object(
+            SyncHumanInputProvider,
+            "_prompt_input_async",
+            new_callable=AsyncMock,
+            side_effect=lambda *_args, **_kwargs: next(feedback_responses),
+        ) as mock_prompt_input,
+        patch.object(
+            AgentExecutor,
+            "kickoff_async",
+            autospec=True,
+            side_effect=kickoff_side_effect,
+        ) as mock_kickoff,
+    ):
+        output = await agent.aexecute_task(task)
+
+    assert output == "Hello"
+    assert mock_prompt_input.await_count == 2
+    assert mock_kickoff.await_count == 2
+
+
 def test_interpolate_inputs():
     agent = Agent(
         role="{topic} specialist",
@@ -2243,6 +2335,27 @@ def test_agent_from_repository_override_attributes(mock_get_agent, mock_get_auth
     assert isinstance(agent.tools[0], SerperDevTool)
 
 
+@patch("crewai.plus_api.PlusAPI.get_agent")
+def test_agent_from_repository_ignores_empty_skills(
+    mock_get_agent, mock_get_auth_token
+):
+    mock_get_response = MagicMock()
+    mock_get_response.status_code = 200
+    mock_get_response.json.return_value = {
+        "role": "test role",
+        "goal": "test goal",
+        "backstory": "test backstory",
+        "tools": [],
+        "skills": [],
+    }
+    mock_get_agent.return_value = mock_get_response
+
+    agent = Agent(from_repository="test_agent")
+
+    assert agent.role == "test role"
+    assert agent.skills is None
+
+
 @patch("crewai.plus_api.PlusAPI.get_agent")
 def test_agent_from_repository_with_invalid_tools(mock_get_agent, mock_get_auth_token):
     mock_get_response = MagicMock()

lib/crewai/tests/agents/test_agent_executor.py

@@ -18,6 +18,7 @@ import pytest
 from pydantic import BaseModel
 
 from crewai.agents.tools_handler import ToolsHandler as _ToolsHandler
+from crewai.core.providers.human_input import SyncHumanInputProvider
 from crewai.agents.step_executor import StepExecutor
 
 
@@ -27,6 +28,13 @@ def _build_executor(**kwargs: Any) -> AgentExecutor:
     Uses model_construct to skip Pydantic validators so plain Mock()
     objects are accepted for typed fields like llm, agent, crew, task.
     """
+    prompt = kwargs.get("prompt")
+    if isinstance(prompt, dict):
+        if "system" in prompt:
+            kwargs["prompt"] = SystemPromptResult(**prompt)
+        else:
+            kwargs["prompt"] = StandardPromptResult(**prompt)
+
     executor = AgentExecutor.model_construct(**kwargs)
     executor._state = AgentExecutorState()
     executor._methods = {}
@@ -50,6 +58,7 @@ def _build_executor(**kwargs: Any) -> AgentExecutor:
     executor._last_context_error = None
     executor._step_executor = None
     executor._planner_observer = None
+    executor._is_feedback_iteration = False
     return executor
 from crewai.agents.planner_observer import PlannerObserver
 from crewai.experimental.agent_executor import (
@@ -68,7 +77,8 @@ from crewai.events.types.tool_usage_events import (
 )
 from crewai.tools.tool_types import ToolResult
 from crewai.utilities.step_execution_context import StepExecutionContext
-from crewai.utilities.planning_types import TodoItem
+from crewai.utilities.planning_types import TodoItem, TodoList
+from crewai.utilities.prompts import StandardPromptResult, SystemPromptResult
 from crewai.utilities.file_store import clear_files, clear_task_files, store_files
 from crewai_files import TextFile
 
@@ -119,6 +129,189 @@ class TestAgentExecutor:
     class StructuredResult(BaseModel):
         value: str
 
+    def test_setup_messages_calls_human_input_provider_hooks(self):
+        """Message setup should preserve the HumanInputProvider hook contract."""
+        executor = _build_executor(
+            prompt=StandardPromptResult(prompt="Original task: {input}"),
+        )
+        provider = Mock()
+        provider.setup_messages.return_value = False
+
+        def post_setup(context: AgentExecutor) -> None:
+            context.messages.append(
+                {"role": "system", "content": "provider post setup"}
+            )
+
+        provider.post_setup_messages.side_effect = post_setup
+
+        with patch(
+            "crewai.experimental.agent_executor.get_provider", return_value=provider
+        ):
+            executor._setup_messages(
+                {"input": "draft this", "tool_names": "", "tools": ""}
+            )
+
+        provider.setup_messages.assert_called_once_with(executor)
+        provider.post_setup_messages.assert_called_once_with(executor)
+        assert executor.state.messages[0]["role"] == "user"
+        assert executor.state.messages[0]["content"] == "Original task: draft this"
+        assert executor.state.messages[1] == {
+            "role": "system",
+            "content": "provider post setup",
+        }
+
+    def test_setup_messages_can_be_owned_by_human_input_provider(self):
+        """Providers can skip standard prompt setup by returning True."""
+        executor = _build_executor(
+            prompt=StandardPromptResult(prompt="Original task: {input}"),
+        )
+        provider = Mock()
+
+        def setup(context: AgentExecutor) -> bool:
+            context.messages.append({"role": "user", "content": "provider message"})
+            return True
+
+        provider.setup_messages.side_effect = setup
+
+        with patch(
+            "crewai.experimental.agent_executor.get_provider", return_value=provider
+        ):
+            executor._setup_messages(
+                {"input": "draft this", "tool_names": "", "tools": ""}
+            )
+
+        provider.setup_messages.assert_called_once_with(executor)
+        provider.post_setup_messages.assert_not_called()
+        assert executor.state.messages == [
+            {"role": "user", "content": "provider message"}
+        ]
+
+    def test_human_feedback_reruns_flow_with_state_messages(self):
+        """Human feedback should use AgentExecutor state messages."""
+        executor = _build_executor(agent=SimpleNamespace(verbose=False), crew=None)
+        executor.state.messages = [{"role": "user", "content": "original task"}]
+        executor.state.current_answer = AgentFinish(
+            thought="", output="draft", text="draft"
+        )
+        executor.state.is_finished = True
+        executor._finalize_called = True
+        executor.ask_for_human_input = True
+        executor.state.iterations = executor.max_iter
+        executor.state.plan = "completed plan"
+        executor.state.plan_ready = True
+        executor.state.todos = TodoList(
+            items=[TodoItem(step_number=1, description="Done", status="completed")]
+        )
+
+        improved_answer = AgentFinish(thought="", output="improved", text="improved")
+        feedback_responses = iter(["make it friendlier", ""])
+
+        def finish_feedback_iteration(*_args: Any, **_kwargs: Any) -> None:
+            assert executor._is_feedback_iteration is True
+            assert executor.state.iterations == 0
+            assert executor.state.plan is None
+            assert executor.state.todos.items == []
+            executor.state.current_answer = improved_answer
+            executor.state.is_finished = True
+
+        with (
+            patch.object(
+                SyncHumanInputProvider,
+                "_prompt_input",
+                side_effect=lambda *_args, **_kwargs: next(feedback_responses),
+            ) as mock_prompt_input,
+            patch.object(
+                AgentExecutor, "kickoff", side_effect=finish_feedback_iteration
+            ) as mock_kickoff,
+        ):
+            result = executor._handle_human_feedback(
+                AgentFinish(thought="", output="draft", text="draft")
+            )
+
+        assert result is improved_answer
+        assert mock_prompt_input.call_count == 2
+        mock_kickoff.assert_called_once()
+        assert executor.messages is executor.state.messages
+        assert "make it friendlier" in executor.state.messages[-1]["content"]
+        assert executor.ask_for_human_input is False
+        assert executor.state.current_answer is improved_answer
+        assert executor.state.is_finished is True
+        assert executor._finalize_called is True
+        assert executor._is_feedback_iteration is False
+
+    @pytest.mark.asyncio
+    async def test_async_human_feedback_reruns_flow_with_state_messages(self):
+        """Async human feedback should use AgentExecutor state messages."""
+        executor = _build_executor(agent=SimpleNamespace(verbose=False), crew=None)
+        executor.state.messages = [{"role": "user", "content": "original task"}]
+        executor.state.current_answer = AgentFinish(
+            thought="", output="draft", text="draft"
+        )
+        executor.state.is_finished = True
+        executor._finalize_called = True
+        executor.ask_for_human_input = True
+        executor.state.iterations = executor.max_iter
+        executor.state.plan = "completed plan"
+        executor.state.plan_ready = True
+        executor.state.todos = TodoList(
+            items=[TodoItem(step_number=1, description="Done", status="completed")]
+        )
+
+        improved_answer = AgentFinish(thought="", output="improved", text="improved")
+        feedback_responses = iter(["make it friendlier", ""])
+
+        async def finish_feedback_iteration(*_args: Any, **_kwargs: Any) -> None:
+            assert executor._is_feedback_iteration is True
+            assert executor.state.iterations == 0
+            assert executor.state.plan is None
+            assert executor.state.todos.items == []
+            executor.state.current_answer = improved_answer
+            executor.state.is_finished = True
+
+        with (
+            patch.object(
+                SyncHumanInputProvider,
+                "_prompt_input_async",
+                new_callable=AsyncMock,
+                side_effect=lambda *_args, **_kwargs: next(feedback_responses),
+            ) as mock_prompt_input,
+            patch.object(
+                AgentExecutor,
+                "kickoff_async",
+                new_callable=AsyncMock,
+                side_effect=finish_feedback_iteration,
+            ) as mock_kickoff,
+        ):
+            result = await executor._ahandle_human_feedback(
+                AgentFinish(thought="", output="draft", text="draft")
+            )
+
+        assert result is improved_answer
+        assert mock_prompt_input.await_count == 2
+        mock_kickoff.assert_awaited_once()
+        assert executor.messages is executor.state.messages
+        assert "make it friendlier" in executor.state.messages[-1]["content"]
+        assert executor.ask_for_human_input is False
+        assert executor.state.current_answer is improved_answer
+        assert executor.state.is_finished is True
+        assert executor._finalize_called is True
+        assert executor._is_feedback_iteration is False
+
+    def test_feedback_iteration_skips_plan_generation(self):
+        """Feedback reruns should reason over feedback without regenerating a plan."""
+        executor = _build_executor(
+            agent=SimpleNamespace(planning_enabled=True, verbose=False),
+            task=SimpleNamespace(),
+        )
+        executor._is_feedback_iteration = True
+
+        with patch("crewai.utilities.reasoning_handler.AgentReasoning") as reasoning:
+            executor.generate_plan()
+
+        reasoning.assert_not_called()
+        assert executor.state.plan is None
+        assert executor.state.todos.items == []
+
     def test_inject_files_from_crew_task_store(self):
         """Crew-level input_files should attach to the LLM user message."""
         crew_id = uuid4()

lib/crewai/tests/cli/test_plus_api.py

@@ -1,8 +1,6 @@
 import os
 import unittest
-from unittest.mock import ANY, AsyncMock, MagicMock, patch
-
-import pytest
+from unittest.mock import ANY, MagicMock, patch
 
 from crewai.plus_api import PlusAPI
 
@@ -396,28 +394,23 @@ class TestPlusAPI(unittest.TestCase):
         )
 
 
-@pytest.mark.asyncio
-@patch("httpx.AsyncClient")
-async def test_get_agent(mock_async_client_class):
+@patch("crewai_core.plus_api.PlusAPI._make_request")
+def test_get_agent(mock_make_request):
     api = PlusAPI("test_api_key")
     mock_response = MagicMock()
-    mock_client_instance = AsyncMock()
-    mock_client_instance.get.return_value = mock_response
-    mock_async_client_class.return_value.__aenter__.return_value = mock_client_instance
+    mock_make_request.return_value = mock_response
 
-    response = await api.get_agent("test_agent_handle")
+    response = api.get_agent("test_agent_handle")
 
-    mock_client_instance.get.assert_called_once_with(
-        f"{api.base_url}/crewai_plus/api/v1/agents/test_agent_handle",
-        headers=api.headers,
+    mock_make_request.assert_called_once_with(
+        "GET", "/crewai_plus/api/v1/agents/test_agent_handle"
     )
     assert response == mock_response
 
 
-@pytest.mark.asyncio
-@patch("httpx.AsyncClient")
+@patch("crewai_core.plus_api.PlusAPI._make_request")
 @patch("crewai_core.plus_api.Settings")
-async def test_get_agent_with_org_uuid(mock_settings_class, mock_async_client_class):
+def test_get_agent_with_org_uuid(mock_settings_class, mock_make_request):
     org_uuid = "test-org-uuid"
     mock_settings = MagicMock()
     mock_settings.org_uuid = org_uuid
@@ -427,15 +420,12 @@ async def test_get_agent_with_org_uuid(mock_settings_class, mock_async_client_cl
     api = PlusAPI("test_api_key")
 
     mock_response = MagicMock()
-    mock_client_instance = AsyncMock()
-    mock_client_instance.get.return_value = mock_response
-    mock_async_client_class.return_value.__aenter__.return_value = mock_client_instance
+    mock_make_request.return_value = mock_response
 
-    response = await api.get_agent("test_agent_handle")
+    response = api.get_agent("test_agent_handle")
 
-    mock_client_instance.get.assert_called_once_with(
-        f"{api.base_url}/crewai_plus/api/v1/agents/test_agent_handle",
-        headers=api.headers,
+    mock_make_request.assert_called_once_with(
+        "GET", "/crewai_plus/api/v1/agents/test_agent_handle"
     )
     assert "X-Crewai-Organization-Id" in api.headers
     assert api.headers["X-Crewai-Organization-Id"] == org_uuid

lib/crewai/tests/project/test_json_loader.py

@@ -355,6 +355,24 @@ class TestLoadAgent:
         with pytest.raises(Exception):
             load_agent(agent_file)
 
+    @pytest.mark.parametrize("field", ["role", "goal", "backstory"])
+    def test_load_agent_rejects_null_required_fields(
+        self, tmp_path: Path, field: str
+    ):
+        agent_def = {
+            "role": "Researcher",
+            "goal": "Find information",
+            "backstory": "Expert researcher.",
+        }
+        agent_def[field] = None
+        agent_file = tmp_path / "agent.json"
+        agent_file.write_text(json.dumps(agent_def))
+
+        with pytest.raises(
+            JSONProjectValidationError, match=f"missing required field '{field}'"
+        ):
+            load_agent(agent_file)
+
     def test_load_agent_file_not_found(self):
         with pytest.raises(FileNotFoundError):
             load_agent(Path("/nonexistent/agent.json"))

lib/crewai/tests/telemetry/test_telemetry.py

@@ -96,6 +96,32 @@ def test_flow_execution_span_records_crewai_version():
     span.set_attribute.assert_any_call("flow_name", "ResearchFlow")
 
 
+def test_flow_creation_span_records_crewai_version():
+    tracer = Mock()
+    span = Mock()
+    tracer.start_span.return_value = span
+
+    with (
+        patch.dict(
+            os.environ,
+            {
+                "CREWAI_DISABLE_TELEMETRY": "false",
+                "CREWAI_DISABLE_TRACKING": "false",
+                "OTEL_SDK_DISABLED": "false",
+            },
+        ),
+        patch("crewai.telemetry.telemetry.TracerProvider"),
+        patch("crewai.telemetry.telemetry.trace.get_tracer", return_value=tracer),
+        patch("crewai.telemetry.telemetry.version", return_value="9.9.9"),
+    ):
+        telemetry = Telemetry()
+        telemetry.flow_creation_span("ResearchFlow")
+
+    tracer.start_span.assert_called_once_with("Flow Creation")
+    span.set_attribute.assert_any_call("crewai_version", "9.9.9")
+    span.set_attribute.assert_any_call("flow_name", "ResearchFlow")
+
+
 @patch("crewai.telemetry.telemetry.logger.error")
 @patch(
     "opentelemetry.exporter.otlp.proto.http.trace_exporter.OTLPSpanExporter.export",

lib/crewai/tests/test_crew.py

@@ -2908,12 +2908,6 @@ def test_manager_agent_with_tools_raises_exception(researcher, writer):
         crew.kickoff()
 
 
-@pytest.mark.xfail(
-    strict=True,
-    reason="crew.train() relies on CrewAgentExecutor._format_feedback_message; "
-    "AgentExecutor (the new default) does not implement training feedback yet. "
-    "Remove this xfail once training is migrated to AgentExecutor.",
-)
 @pytest.mark.vcr()
 def test_crew_train_success(researcher, writer, monkeypatch):
     task = Task(

lib/crewai/tests/test_flow_from_definition.py

@@ -1163,6 +1163,139 @@ methods:
     }
 
 
+def test_agent_action_runs_repository_yaml_definition(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    from crewai import Agent
+    from crewai.plus_api import PlusAPI
+
+    fetched_agents: list[str] = []
+
+    class FakeResponse:
+        status_code = 200
+        text = ""
+
+        def json(self) -> dict[str, Any]:
+            return {
+                "role": "Repository specialist",
+                "goal": "Answer support questions",
+                "backstory": "Loaded from the agent repository.",
+                "max_iter": 3,
+                "tools": [],
+            }
+
+    def fake_get_agent(self: PlusAPI, handle: str) -> FakeResponse:
+        fetched_agents.append(handle)
+        return FakeResponse()
+
+    async def fake_kickoff_async(
+        self: Agent, messages: str, **_kwargs: Any
+    ) -> dict[str, Any]:
+        return {"agent": self.role, "input": messages, "max_iter": self.max_iter}
+
+    monkeypatch.setattr("crewai.auth.token.get_auth_token", lambda: "test-token")
+    monkeypatch.setattr(PlusAPI, "get_agent", fake_get_agent)
+    monkeypatch.setattr(Agent, "kickoff_async", fake_kickoff_async)
+
+    yaml_str = """
+schema: crewai.flow/v1
+name: AgentFlow
+methods:
+  answer:
+    do:
+      call: agent
+      with:
+        from_repository: support_specialist
+        input: "${state.question}"
+    start: true
+"""
+
+    flow = Flow.from_declaration(contents=yaml_str)
+
+    assert flow.kickoff(inputs={"question": "What is CrewAI?"}) == {
+        "agent": "Repository specialist",
+        "input": "What is CrewAI?",
+        "max_iter": 3,
+    }
+    assert fetched_agents == ["support_specialist"]
+
+
+def test_agent_action_repository_fetch_does_not_block_event_loop(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    from crewai import Agent
+    from crewai.plus_api import PlusAPI
+
+    loop_marker_ran = threading.Event()
+    fetch_started = threading.Event()
+    release_fetch = threading.Event()
+    fetch_saw_loop_marker = False
+
+    class FakeResponse:
+        status_code = 200
+        text = ""
+
+        def json(self) -> dict[str, Any]:
+            return {
+                "role": "Repository specialist",
+                "goal": "Answer support questions",
+                "backstory": "Loaded from the agent repository.",
+                "tools": [],
+            }
+
+    def fake_get_agent(self: PlusAPI, handle: str) -> FakeResponse:
+        nonlocal fetch_saw_loop_marker
+        fetch_started.set()
+        release_fetch.wait(timeout=1)
+        fetch_saw_loop_marker = loop_marker_ran.is_set()
+        return FakeResponse()
+
+    async def fake_kickoff_async(
+        self: Agent, messages: str, **_kwargs: Any
+    ) -> str:
+        return f"{self.role}:{messages}"
+
+    monkeypatch.setattr("crewai.auth.token.get_auth_token", lambda: "test-token")
+    monkeypatch.setattr(PlusAPI, "get_agent", fake_get_agent)
+    monkeypatch.setattr(Agent, "kickoff_async", fake_kickoff_async)
+
+    yaml_str = """
+schema: crewai.flow/v1
+name: AgentFlow
+methods:
+  answer:
+    do:
+      call: agent
+      with:
+        from_repository: support_specialist
+        input: "${state.question}"
+    start: true
+"""
+
+    flow = Flow.from_declaration(contents=yaml_str)
+
+    async def run_flow() -> str:
+        async def mark_loop_progress() -> None:
+            while not fetch_started.is_set():
+                await asyncio.sleep(0)
+            loop_marker_ran.set()
+            release_fetch.set()
+
+        marker_task = asyncio.create_task(mark_loop_progress())
+        kickoff_task = asyncio.create_task(
+            flow.kickoff_async(inputs={"question": "What is CrewAI?"})
+        )
+        try:
+            result = await asyncio.wait_for(kickoff_task, timeout=2)
+            await asyncio.wait_for(marker_task, timeout=2)
+            return result
+        finally:
+            release_fetch.set()
+
+    assert asyncio.run(run_flow()) == "Repository specialist:What is CrewAI?"
+    assert fetch_saw_loop_marker
+
+
 def test_agent_action_renders_text_custom_expression_input(
     monkeypatch: pytest.MonkeyPatch,
 ):
@@ -1281,6 +1414,7 @@ def test_agent_action_json_schema_describes_inline_agent_definitions():
         "role",
         "goal",
         "backstory",
+        "from_repository",
         "settings",
         "llm",
         "input",
@@ -1385,6 +1519,167 @@ methods:
     }
 
 
+def test_crew_action_runs_repository_agent_yaml_definition(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    from crewai import Crew
+    from crewai.plus_api import PlusAPI
+
+    fetched_agents: list[str] = []
+
+    class FakeResponse:
+        status_code = 200
+        text = ""
+
+        def json(self) -> dict[str, Any]:
+            return {
+                "role": "Repository researcher",
+                "goal": "Research {topic}",
+                "backstory": "Loaded from the agent repository.",
+                "max_iter": 5,
+                "tools": [],
+            }
+
+    def fake_get_agent(self: PlusAPI, handle: str) -> FakeResponse:
+        fetched_agents.append(handle)
+        return FakeResponse()
+
+    async def fake_kickoff_async(
+        self: Crew, inputs: dict[str, Any] | None = None, **_kwargs: Any
+    ) -> dict[str, Any]:
+        return {
+            "crew": self.name,
+            "agents": [
+                {"role": agent.role, "max_iter": agent.max_iter}
+                for agent in self.agents
+            ],
+            "tasks": [task.description for task in self.tasks],
+            "inputs": inputs,
+        }
+
+    monkeypatch.setattr("crewai.auth.token.get_auth_token", lambda: "test-token")
+    monkeypatch.setattr(PlusAPI, "get_agent", fake_get_agent)
+    monkeypatch.setattr(Crew, "kickoff_async", fake_kickoff_async)
+
+    yaml_str = """
+schema: crewai.flow/v1
+name: CrewFlow
+methods:
+  research:
+    do:
+      call: crew
+      with:
+        name: inline_research
+        agents:
+          researcher:
+            from_repository: researcher
+        tasks:
+          - name: research_task
+            description: Research {topic}
+            expected_output: Findings about {topic}
+            agent: researcher
+      inputs:
+        topic: "${state.topic}"
+    start: true
+"""
+
+    flow = Flow.from_declaration(contents=yaml_str)
+
+    assert flow.kickoff(inputs={"topic": "AI"}) == {
+        "crew": "inline_research",
+        "agents": [{"role": "Repository researcher", "max_iter": 5}],
+        "tasks": ["Research {topic}"],
+        "inputs": {"topic": "AI"},
+    }
+    assert fetched_agents == ["researcher"]
+
+
+def test_crew_action_repository_fetch_does_not_block_event_loop(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    from crewai import Crew
+    from crewai.plus_api import PlusAPI
+
+    loop_marker_ran = threading.Event()
+    fetch_started = threading.Event()
+    release_fetch = threading.Event()
+    fetch_saw_loop_marker = False
+
+    class FakeResponse:
+        status_code = 200
+        text = ""
+
+        def json(self) -> dict[str, Any]:
+            return {
+                "role": "Repository researcher",
+                "goal": "Research {topic}",
+                "backstory": "Loaded from the agent repository.",
+                "tools": [],
+            }
+
+    def fake_get_agent(self: PlusAPI, handle: str) -> FakeResponse:
+        nonlocal fetch_saw_loop_marker
+        fetch_started.set()
+        release_fetch.wait(timeout=1)
+        fetch_saw_loop_marker = loop_marker_ran.is_set()
+        return FakeResponse()
+
+    async def fake_kickoff_async(
+        self: Crew, inputs: dict[str, Any] | None = None, **_kwargs: Any
+    ) -> dict[str, Any]:
+        return {"agents": [agent.role for agent in self.agents], "inputs": inputs}
+
+    monkeypatch.setattr("crewai.auth.token.get_auth_token", lambda: "test-token")
+    monkeypatch.setattr(PlusAPI, "get_agent", fake_get_agent)
+    monkeypatch.setattr(Crew, "kickoff_async", fake_kickoff_async)
+
+    yaml_str = """
+schema: crewai.flow/v1
+name: CrewFlow
+methods:
+  research:
+    do:
+      call: crew
+      with:
+        agents:
+          researcher:
+            from_repository: researcher
+        tasks:
+          - description: Research {topic}
+            expected_output: Findings about {topic}
+            agent: researcher
+      inputs:
+        topic: "${state.topic}"
+    start: true
+"""
+
+    flow = Flow.from_declaration(contents=yaml_str)
+
+    async def run_flow() -> dict[str, Any]:
+        async def mark_loop_progress() -> None:
+            while not fetch_started.is_set():
+                await asyncio.sleep(0)
+            loop_marker_ran.set()
+            release_fetch.set()
+
+        marker_task = asyncio.create_task(mark_loop_progress())
+        kickoff_task = asyncio.create_task(
+            flow.kickoff_async(inputs={"topic": "AI"})
+        )
+        try:
+            result = await asyncio.wait_for(kickoff_task, timeout=2)
+            await asyncio.wait_for(marker_task, timeout=2)
+            return result
+        finally:
+            release_fetch.set()
+
+    assert asyncio.run(run_flow()) == {
+        "agents": ["Repository researcher"],
+        "inputs": {"topic": "AI"},
+    }
+    assert fetch_saw_loop_marker
+
+
 def test_crew_action_interpolates_runtime_strings_and_lists(
     monkeypatch: pytest.MonkeyPatch,
 ):
@@ -1709,6 +2004,7 @@ def test_crew_action_json_schema_describes_inline_crew_definitions():
         "role",
         "goal",
         "backstory",
+        "from_repository",
         "settings",
         "llm",
         "tools",
@@ -1728,36 +2024,45 @@ def test_crew_action_json_schema_describes_inline_crew_definitions():
 
 
 def test_crew_action_rejects_incomplete_inline_agent_definition():
-    with pytest.raises(ValidationError, match="goal"):
-        FlowDefinition.from_declaration(contents=
-            {
-                "schema": "crewai.flow/v1",
-                "name": "CrewFlow",
-                "methods": {
-                    "research": {
-                        "start": True,
-                        "do": {
-                            "call": "crew",
-                            "with": {
-                                "agents": {
-                                    "researcher": {
-                                        "role": "Researcher",
-                                        "backstory": "Knows things.",
-                                    }
-                                },
-                                "tasks": [
-                                    {
-                                        "description": "Research",
-                                        "expected_output": "Findings",
-                                        "agent": "researcher",
-                                    }
-                                ],
+    from crewai.project.crew_loader import load_crew_from_definition
+    from crewai.project.json_loader import JSONProjectValidationError
+
+    definition = FlowDefinition.from_declaration(contents=
+        {
+            "schema": "crewai.flow/v1",
+            "name": "CrewFlow",
+            "methods": {
+                "research": {
+                    "start": True,
+                    "do": {
+                        "call": "crew",
+                        "with": {
+                            "agents": {
+                                "researcher": {
+                                    "role": "Researcher",
+                                    "backstory": "Knows things.",
+                                }
                             },
+                            "tasks": [
+                                {
+                                    "description": "Research",
+                                    "expected_output": "Findings",
+                                    "agent": "researcher",
+                                }
+                            ],
                         },
-                    }
-                },
-            }
-        )
+                    },
+                }
+            },
+        }
+    )
+    crew_definition = definition.methods["research"].do.with_
+    assert crew_definition.agents["researcher"].goal is None
+
+    with pytest.raises(
+        JSONProjectValidationError, match="missing required field 'goal'"
+    ):
+        load_crew_from_definition(crew_definition, source="crew action")
 
 
 def test_crew_action_rejects_python_ref_field():

lib/crewai/tests/test_flow_human_input_integration.py

@@ -138,4 +138,27 @@ class TestFlowHumanInputIntegration:
                 for call in call_args
                 if call[0]
             )
-            assert training_panel_found
+            assert training_panel_found
+
+    @patch("builtins.input", return_value="please make it warmer")
+    def test_non_empty_input_prints_processing_feedback(self, mock_input):
+        """Non-empty input should be displayed as feedback to process."""
+        provider = SyncHumanInputProvider()
+        crew = MagicMock()
+        crew._train = False
+
+        formatter = event_listener.formatter
+
+        with (
+            patch.object(formatter, "pause_live_updates"),
+            patch.object(formatter, "resume_live_updates"),
+            patch.object(formatter.console, "print") as mock_console_print,
+        ):
+            result = provider._prompt_input(crew)
+
+        assert result == "please make it warmer"
+        mock_input.assert_called_once()
+        printed_text = "\n".join(
+            str(call.args[0]) for call in mock_console_print.call_args_list
+        )
+        assert "Processing your feedback" in printed_text