Merge branch 'main' into joaomdmoura/run-inputs-definition-resolution

The declarative-flow path never loaded .env — flow projects (type = "flow")
missed API keys/config that crew projects pick up. The JSON-crew path loads
Path.cwd()/.env with override=True (run_crew._run_json_crew); mirror that at
the top of run_declarative_flow() so flow projects behave the same regardless
of where crewai is installed. Test: run_declarative_flow_loads_project_env.

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/cli.py

@@ -526,8 +526,8 @@ def run(
     inputs: str | None,
 ) -> None:
     """Run the Crew or Flow."""
-    if inputs is not None and definition is None:
-        raise click.UsageError("--inputs requires --definition")
+    # --inputs no longer requires --definition: with no override it resolves the
+    # configured [tool.crewai] flow, same as a bare `crewai run`.
     if trained_agents_file is not None and definition is not None:
         raise click.UsageError("--filename can only be used when running crews")
 

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,657 @@
+"""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 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.
+_CATALOG_TTL = 6 * 3600
+
+#: 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.
+    # Skip Ollama: it's a local, fast-failing server, so re-probing is cheap and
+    # avoids serving suggestions after the server comes up within the TTL.
+    if fallback and provider_key != "ollama":
+        _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 _cache_key(provider_key: str) -> str:
+    """Cache key for a provider's resolved model list.
+
+    Includes the inputs that change what a fetch would return, so a cached
+    entry is only reused when those inputs still match:
+
+    - Ollama lists models from a base URL that can change between runs.
+    - Whether the vendor's API key is present flips between a live fetch and
+      the negatively-cached fallback — so a key added after a no-key call is
+      not shadowed by the cached fallback.
+    """
+    if provider_key == "ollama":
+        return f"ollama@{_ollama_base()}"
+    suffix = "key" if _provider_api_key(provider_key) else "nokey"
+    return f"{provider_key}#{suffix}"
+
+
+def _read_catalog_cache(provider_key: str) -> list[tuple[str, str]] | None:
+    """Return a fresh cached catalog for ``provider_key``, or ``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:
+    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/src/crewai_cli/run_crew.py

@@ -571,9 +571,7 @@ def run_crew(
         definition: Optional path to a declarative Flow definition.
         inputs: Optional JSON object passed to a declarative Flow.
     """
-    if inputs is not None and definition is None:
-        raise click.UsageError("--inputs requires --definition")
-
+    # --definition is a pure override: run that flow directly.
     if definition is not None:
         _run_explicit_declarative_flow(
             definition=definition,
@@ -584,6 +582,7 @@ def run_crew(
 
     pyproject_data = read_toml()
     if json_crew_definition := configured_project_json_crew(pyproject_data):
+        _reject_inputs_for_non_flow(inputs)
         _run_json_crew_in_project_env(
             trained_agents_file=trained_agents_file,
             crew_path=json_crew_definition,
@@ -594,18 +593,27 @@ def run_crew(
     project_type = get_crewai_project_type(pyproject_data)
 
     if project_type == "flow":
+        # No --definition: resolve the configured [tool.crewai] flow — the same
+        # resolution as a bare `crewai run` — and pass --inputs straight through.
         _run_flow_project(
             pyproject_data=pyproject_data,
             trained_agents_file=trained_agents_file,
+            inputs=inputs,
         )
         return
 
+    _reject_inputs_for_non_flow(inputs)
     _run_classic_crew_project(
         pyproject_data=pyproject_data,
         trained_agents_file=trained_agents_file,
     )
 
 
+def _reject_inputs_for_non_flow(inputs: str | None) -> None:
+    if inputs is not None:
+        raise click.UsageError("--inputs is only supported for declarative flows")
+
+
 def _run_explicit_declarative_flow(
     definition: str, inputs: str | None, trained_agents_file: str | None
 ) -> None:
@@ -618,7 +626,9 @@ def _run_explicit_declarative_flow(
 
 
 def _run_flow_project(
-    pyproject_data: dict[str, Any], trained_agents_file: str | None
+    pyproject_data: dict[str, Any],
+    trained_agents_file: str | None,
+    inputs: str | None = None,
 ) -> None:
     if trained_agents_file is not None:
         raise click.UsageError("--filename can only be used when running crews")
@@ -629,9 +639,16 @@ def _run_flow_project(
     )
 
     if definition := configured_project_declarative_flow(pyproject_data):
-        run_declarative_flow_in_project_env(definition=definition)
+        run_declarative_flow_in_project_env(definition=definition, inputs=inputs)
         return
 
+    # No configured declarative flow definition to resolve inputs against.
+    if inputs is not None:
+        raise click.UsageError(
+            "--inputs requires a declarative flow definition "
+            "([tool.crewai].definition) or --definition"
+        )
+
     from crewai_cli.kickoff_flow import (
         _load_conversational_flow_from_kickoff_script,
         _run_conversational_flow_tui,

lib/cli/src/crewai_cli/run_declarative_flow.py

@@ -1,15 +1,21 @@
 from __future__ import annotations
 
+import difflib
 import json
 from pathlib import Path
 import subprocess
+import sys
 from typing import Any
 
 import click
 from crewai_core.project import ProjectDefinitionError, configured_project_definition
 from pydantic import ValidationError
 
-from crewai_cli.utils import build_env_with_all_tool_credentials
+from crewai_cli.utils import (
+    build_env_with_all_tool_credentials,
+    enable_prompt_line_editing,
+    is_dmn_mode_enabled,
+)
 
 
 def run_declarative_flow_in_project_env(
@@ -20,10 +26,13 @@ def run_declarative_flow_in_project_env(
         run_declarative_flow(definition=definition, inputs=inputs)
         return
 
+    # Re-run inside the project env (so the flow loads with the project's deps).
+    # The configured definition is re-resolved there; forward --inputs so the
+    # in-env run kicks off with the same values instead of losing them.
+    command = ["uv", "run", "crewai", "run"]
     if inputs is not None:
-        raise click.UsageError("--inputs is only supported with --definition")
-
-    _execute_declarative_flow_command(["uv", "run", "crewai", "run"])
+        command += ["--inputs", inputs]
+    _execute_declarative_flow_command(command)
 
 
 def plot_declarative_flow_in_project_env(definition: str | Path) -> None:
@@ -36,12 +45,29 @@ def plot_declarative_flow_in_project_env(definition: str | Path) -> None:
 
 
 def run_declarative_flow(definition: str | Path, inputs: str | None = None) -> None:
-    """Run a declarative flow from a definition path."""
-    parsed_inputs = _parse_inputs(inputs)
+    """Run a declarative flow from a definition path.
+
+    Inputs come from one place: the flow's own state schema. Any ``--inputs``
+    JSON is layered on top as an override, missing required fields are prompted
+    for interactively, and everything is validated against the schema before
+    kickoff — so a bare ``crewai run`` on a configured flow just works.
+    """
+    # Load the project's .env before kickoff, mirroring the JSON-crew path
+    # (run_crew._run_json_crew) so flow projects pick up API keys/config the
+    # same way regardless of where crewai is installed.
+    from dotenv import load_dotenv
+
+    env_file = Path.cwd() / ".env"
+    if env_file.exists():
+        load_dotenv(env_file, override=True)
+
+    provided = _parse_inputs(inputs) or {}
+
+    flow = load_declarative_flow(definition)
+    resolved_inputs = _resolve_flow_inputs(flow, provided)
 
     try:
-        flow = load_declarative_flow(definition)
-        result = flow.kickoff(inputs=parsed_inputs)
+        result = flow.kickoff(inputs=resolved_inputs or None)
     except Exception as exc:
         click.echo(
             f"An error occurred while running the declarative flow: {exc}", err=True
@@ -51,6 +77,167 @@ def run_declarative_flow(definition: str | Path, inputs: str | None = None) -> N
     click.echo(_format_result(result))
 
 
+def _resolve_flow_inputs(flow: Any, provided: dict[str, Any]) -> dict[str, Any]:
+    """Resolve kickoff inputs from the flow's state schema.
+
+    Warns on unknown keys, prompts for missing required fields (unless
+    non-interactive), and validates types before kickoff. Exits with a pointed
+    message when a required input is still missing or an input is invalid.
+    """
+    schema = _flow_state_schema(flow)
+    if schema is None:
+        # dict / unschematized state — nothing to derive; pass inputs through.
+        return dict(provided)
+
+    # ``id`` signals a persistence restore: kickoff hydrates the full state from
+    # storage, so required fields may come from the restored state rather than
+    # --inputs. Forward the inputs unchanged instead of prompting/erroring for
+    # fields the resume will supply.
+    if "id" in provided:
+        return dict(provided)
+
+    properties = {
+        name: spec
+        for name, spec in (schema.get("properties") or {}).items()
+        if name != "id"
+    }
+    state_model = type(flow.state)
+    defaults = _flow_state_defaults(flow)
+
+    # Unknown keys are almost always typos — warn and drop them (they'd fail
+    # structured-state validation at kickoff anyway).
+    collected: dict[str, Any] = {}
+    for key, value in provided.items():
+        if key in properties:
+            collected[key] = value
+            continue
+        suggestion = _closest_key(key, properties)
+        hint = f" Did you mean '{suggestion}'?" if suggestion else ""
+        click.secho(
+            f"  Ignoring unknown input '{key}' — not in the flow's state schema.{hint}",
+            fg="yellow",
+            err=True,
+        )
+
+    missing = _missing_required(state_model, {**defaults, **collected})
+    if missing and _is_interactive():
+        collected.update(_prompt_for_flow_inputs(missing, properties))
+        missing = _missing_required(state_model, {**defaults, **collected})
+
+    if missing:
+        for name in missing:
+            description = (properties.get(name) or {}).get("description")
+            suffix = f" — {description}" if description else ""
+            click.secho(
+                f"  Missing required input '{name}'{suffix}", fg="red", err=True
+            )
+        raise SystemExit(1)
+
+    _validate_flow_inputs(state_model, {**defaults, **collected})
+    return collected
+
+
+def _is_interactive() -> bool:
+    """Prompt only in an interactive terminal, never in non-interactive mode."""
+    return not is_dmn_mode_enabled() and sys.stdin.isatty()
+
+
+def _flow_state_schema(flow: Any) -> dict[str, Any] | None:
+    """Return the flow's state JSON schema, or ``None`` for dict/plain state."""
+    state = getattr(flow, "state", None)
+    if state is None or isinstance(state, dict):
+        return None
+    model_json_schema = getattr(type(state), "model_json_schema", None)
+    if not callable(model_json_schema):
+        return None
+    try:
+        schema = model_json_schema()
+    except Exception:
+        return None
+    return schema if isinstance(schema, dict) else None
+
+
+def _flow_state_defaults(flow: Any) -> dict[str, Any]:
+    """Declared state defaults (``state.default``) from the flow definition."""
+    state_definition = getattr(getattr(flow, "_definition", None), "state", None)
+    default = getattr(state_definition, "default", None)
+    return dict(default) if isinstance(default, dict) else {}
+
+
+def _missing_required(state_model: Any, values: dict[str, Any]) -> list[str]:
+    """Required state fields not satisfied by ``values`` (defaults + inputs)."""
+    try:
+        state_model.model_validate(values)
+    except ValidationError as exc:
+        return [
+            str(error["loc"][0])
+            for error in exc.errors()
+            if error.get("type") == "missing" and error.get("loc")
+        ]
+    return []
+
+
+def _validate_flow_inputs(state_model: Any, values: dict[str, Any]) -> None:
+    """Validate inputs against the state schema; exit with pointed type errors."""
+    try:
+        state_model.model_validate(values)
+    except ValidationError as exc:
+        for error in exc.errors():
+            location = ".".join(str(part) for part in error.get("loc", ()))
+            click.secho(
+                f"  Invalid input '{location}': {error.get('msg')}", fg="red", err=True
+            )
+        raise SystemExit(1) from exc
+
+
+def _prompt_for_flow_inputs(
+    missing: list[str], properties: dict[str, Any]
+) -> dict[str, Any]:
+    """Prompt for each missing required field, showing its schema description."""
+    enable_prompt_line_editing()
+    # Prompt chrome goes to stderr so stdout carries only the flow result.
+    click.echo(err=True)
+    click.secho("  Flow inputs", fg="cyan", bold=True, err=True)
+    click.secho("  This flow needs the following to run.", dim=True, err=True)
+
+    collected: dict[str, Any] = {}
+    for name in missing:
+        spec = properties.get(name) or {}
+        description = spec.get("description")
+        if description:
+            click.secho(f"  {description}", dim=True, err=True)
+        raw = click.prompt(
+            click.style(f"  {name}", fg="cyan"),
+            prompt_suffix=click.style(" > ", fg="bright_white"),
+        )
+        collected[name] = _coerce_input(raw, spec)
+    return collected
+
+
+def _coerce_input(raw: str, spec: dict[str, Any]) -> Any:
+    """Best-effort coerce a prompted string to the field's JSON-schema type."""
+    field_type = spec.get("type")
+    if field_type == "integer":
+        try:
+            return int(raw)
+        except ValueError:
+            return raw
+    if field_type == "number":
+        try:
+            return float(raw)
+        except ValueError:
+            return raw
+    if field_type == "boolean":
+        return raw.strip().lower() in {"1", "true", "yes", "y", "on"}
+    return raw
+
+
+def _closest_key(key: str, properties: dict[str, Any]) -> str | None:
+    """Nearest schema field name to a likely typo, if one is close enough."""
+    matches = difflib.get_close_matches(key, list(properties), n=1, cutoff=0.7)
+    return matches[0] if matches else None
+
+
 def plot_declarative_flow(definition: str | Path) -> None:
     """Plot a declarative flow from a definition path."""
     try:

lib/cli/tests/test_cli.py

@@ -151,12 +151,14 @@ def test_run_with_definition_uses_project_runner(run_crew, runner):
 
 
 @mock.patch("crewai_cli.cli.run_crew")
-def test_run_rejects_inputs_without_definition(run_crew, runner):
+def test_run_inputs_without_definition_calls_run_crew(run_crew, runner):
+    # --inputs no longer requires --definition; the resolution happens in run_crew.
     result = runner.invoke(run, ["--inputs", '{"topic":"AI"}'])
 
-    assert result.exit_code == 2
-    assert "Error: --inputs requires --definition" in result.output
-    run_crew.assert_not_called()
+    assert result.exit_code == 0
+    run_crew.assert_called_once_with(
+        trained_agents_file=None, definition=None, inputs='{"topic":"AI"}'
+    )
 
 
 @mock.patch("crewai_cli.cli.run_crew")

lib/cli/tests/test_model_catalog.py

@@ -0,0 +1,551 @@
+"""Tests for the dynamic model catalog used by the crew-creation wizard."""
+
+from __future__ import annotations
+
+import json
+
+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_cache_keyed_by_base(monkeypatch):
+    # Changing OLLAMA_API_BASE must not serve the previous host's cached models.
+    monkeypatch.setenv("OLLAMA_API_BASE", "http://host-a:11434")
+    monkeypatch.setattr(
+        mc, "_http_get_json", lambda *a, **k: {"models": [{"model": "llama-a"}]}
+    )
+    first = mc.get_provider_models("ollama", [])
+    assert [m for m, _ in first] == ["llama-a"]
+
+    monkeypatch.setenv("OLLAMA_API_BASE", "http://host-b:11434")
+    monkeypatch.setattr(
+        mc, "_http_get_json", lambda *a, **k: {"models": [{"model": "llama-b"}]}
+    )
+    second = mc.get_provider_models("ollama", [])
+    assert [m for m, _ in second] == ["llama-b"]  # not the host-a cache

lib/cli/tests/test_run_crew.py

@@ -682,11 +682,48 @@ def test_configured_project_json_crew_ignores_missing_pyproject(
     assert run_crew_module.configured_project_json_crew() is None
 
 
-def test_run_crew_rejects_inputs_without_definition():
+def test_run_crew_inputs_without_definition_rejected_for_non_flow(monkeypatch):
+    # --inputs is flow-only; in a non-flow project it now errors clearly instead
+    # of the old "--inputs requires --definition".
+    monkeypatch.setattr(run_crew_module, "read_toml", lambda *a, **k: {})
+    monkeypatch.setattr(
+        run_crew_module, "configured_project_json_crew", lambda *a, **k: None
+    )
+    monkeypatch.setattr(
+        run_crew_module, "_warn_if_old_poetry_project", lambda *a, **k: None
+    )
+    monkeypatch.setattr(run_crew_module, "get_crewai_project_type", lambda *a, **k: "crew")
+
     with pytest.raises(click.UsageError) as exc_info:
         run_crew_module.run_crew(inputs='{"topic":"AI"}')
 
-    assert "--inputs requires --definition" in exc_info.value.message
+    assert "--inputs is only supported for declarative flows" in exc_info.value.message
+
+
+def test_run_crew_inputs_without_definition_resolves_configured_flow(monkeypatch):
+    # --inputs with no --definition resolves the configured [tool.crewai] flow,
+    # exactly like a bare `crewai run`, and forwards the inputs.
+    import crewai_cli.run_declarative_flow as rdf
+
+    calls: dict[str, object] = {}
+    monkeypatch.setattr(run_crew_module, "read_toml", lambda *a, **k: {})
+    monkeypatch.setattr(
+        run_crew_module, "configured_project_json_crew", lambda *a, **k: None
+    )
+    monkeypatch.setattr(
+        run_crew_module, "_warn_if_old_poetry_project", lambda *a, **k: None
+    )
+    monkeypatch.setattr(run_crew_module, "get_crewai_project_type", lambda *a, **k: "flow")
+    monkeypatch.setattr(
+        rdf, "configured_project_declarative_flow", lambda *a, **k: Path("flow.yaml")
+    )
+    monkeypatch.setattr(
+        rdf, "run_declarative_flow_in_project_env", lambda **kw: calls.update(kw)
+    )
+
+    run_crew_module.run_crew(inputs='{"topic":"AI"}')
+
+    assert calls == {"definition": Path("flow.yaml"), "inputs": '{"topic":"AI"}'}
 
 
 def test_run_crew_rejects_filename_with_explicit_definition():

lib/cli/tests/test_run_declarative_flow.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import os
 from pathlib import Path
 
 import pytest
@@ -146,3 +147,236 @@ def test_run_declarative_flow_in_process_inside_uv(
     )
 
     assert capsys.readouterr().out == "AI\n"
+
+
+def test_run_declarative_flow_in_project_env_forwards_inputs(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    subprocess_calls = []
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.delenv("UV_RUN_RECURSION_DEPTH", raising=False)
+    (tmp_path / "pyproject.toml").write_text("[project]\nname = 'demo'\n")
+    monkeypatch.setattr(
+        run_declarative_flow_module,
+        "build_env_with_all_tool_credentials",
+        lambda: {},
+    )
+    monkeypatch.setattr(
+        run_declarative_flow_module.subprocess,
+        "run",
+        lambda command, **kwargs: subprocess_calls.append(command),
+    )
+
+    run_declarative_flow_module.run_declarative_flow_in_project_env(
+        "flow.yaml", '{"topic":"AI"}'
+    )
+
+    # --inputs is forwarded to the in-env run instead of being rejected.
+    assert subprocess_calls == [
+        ["uv", "run", "crewai", "run", "--inputs", '{"topic":"AI"}']
+    ]
+
+
+# ── Schema-driven inputs: prompt, validate, override ────────────────
+
+REQUIRED_FLOW_YAML = """\
+schema: crewai.flow/v1
+name: RequiredInputFlow
+config:
+  suppress_flow_events: true
+state:
+  type: json_schema
+  json_schema:
+    type: object
+    properties:
+      prospect_email:
+        type: string
+        description: Email address of the prospect to research
+    required: [prospect_email]
+methods:
+  begin:
+    start: true
+    do:
+      call: expression
+      expr: state.prospect_email
+"""
+
+DEFAULTS_FLOW_YAML = """\
+schema: crewai.flow/v1
+name: DefaultsFlow
+config:
+  suppress_flow_events: true
+state:
+  type: json_schema
+  json_schema:
+    type: object
+    properties:
+      topic: {type: string}
+      audience: {type: string}
+    required: [topic, audience]
+  default:
+    topic: AI
+methods:
+  begin:
+    start: true
+    do:
+      call: expression
+      expr: state.audience
+"""
+
+TYPED_FLOW_YAML = """\
+schema: crewai.flow/v1
+name: TypedFlow
+config:
+  suppress_flow_events: true
+state:
+  type: json_schema
+  json_schema:
+    type: object
+    properties:
+      count: {type: integer}
+    required: [count]
+methods:
+  begin:
+    start: true
+    do:
+      call: expression
+      expr: state.count
+"""
+
+
+def _write(tmp_path: Path, contents: str) -> Path:
+    path = tmp_path / "flow.yaml"
+    path.write_text(contents, encoding="utf-8")
+    return path
+
+
+def test_inputs_flag_satisfies_required_field(
+    tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    path = _write(tmp_path, REQUIRED_FLOW_YAML)
+
+    run_declarative_flow_module.run_declarative_flow(
+        str(path), '{"prospect_email":"a@b.com"}'
+    )
+
+    assert capsys.readouterr().out == "a@b.com\n"
+
+
+def test_missing_required_reports_pointed_error(
+    tmp_path: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(run_declarative_flow_module, "_is_interactive", lambda: False)
+    path = _write(tmp_path, REQUIRED_FLOW_YAML)
+
+    with pytest.raises(SystemExit):
+        run_declarative_flow_module.run_declarative_flow(str(path))
+
+    assert (
+        "Missing required input 'prospect_email' — "
+        "Email address of the prospect to research" in capsys.readouterr().err
+    )
+
+
+def test_prompts_for_missing_required_when_interactive(
+    tmp_path: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    path = _write(tmp_path, REQUIRED_FLOW_YAML)
+    monkeypatch.setattr(run_declarative_flow_module, "_is_interactive", lambda: True)
+    prompted: list[str] = []
+
+    def fake_prompt(text: str, **kwargs: object) -> str:
+        prompted.append(text)
+        return "typed@example.com"
+
+    monkeypatch.setattr(run_declarative_flow_module.click, "prompt", fake_prompt)
+
+    run_declarative_flow_module.run_declarative_flow(str(path))
+
+    assert capsys.readouterr().out == "typed@example.com\n"
+    assert any("prospect_email" in text for text in prompted)
+
+
+def test_defaults_satisfy_required_and_are_not_prompted(
+    tmp_path: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(run_declarative_flow_module, "_is_interactive", lambda: False)
+    path = _write(tmp_path, DEFAULTS_FLOW_YAML)
+
+    with pytest.raises(SystemExit):
+        run_declarative_flow_module.run_declarative_flow(str(path))
+
+    err = capsys.readouterr().err
+    # topic has a state default -> satisfied; only audience is missing.
+    assert "Missing required input 'audience'" in err
+    assert "'topic'" not in err
+
+
+def test_warns_on_unknown_input_with_suggestion(
+    tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    path = _write(tmp_path, REQUIRED_FLOW_YAML)
+
+    run_declarative_flow_module.run_declarative_flow(
+        str(path), '{"prospect_email":"a@b.com","prospect_emai":"typo"}'
+    )
+
+    captured = capsys.readouterr()
+    assert captured.out == "a@b.com\n"
+    assert "Ignoring unknown input 'prospect_emai'" in captured.err
+    assert "Did you mean 'prospect_email'?" in captured.err
+
+
+def test_validates_input_types_before_kickoff(
+    tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    path = _write(tmp_path, TYPED_FLOW_YAML)
+
+    with pytest.raises(SystemExit):
+        run_declarative_flow_module.run_declarative_flow(str(path), '{"count":"nope"}')
+
+    assert "Invalid input 'count'" in capsys.readouterr().err
+
+
+def test_reserved_id_input_is_forwarded_not_dropped(
+    tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    # `id` is a reserved kickoff key (persistence restore); it must pass through
+    # instead of being flagged as an unknown key and dropped.
+    path = _write(tmp_path, REQUIRED_FLOW_YAML)
+
+    run_declarative_flow_module.run_declarative_flow(
+        str(path), '{"id":"run-123","prospect_email":"a@b.com"}'
+    )
+
+    captured = capsys.readouterr()
+    assert captured.out == "a@b.com\n"
+    assert "Ignoring unknown input 'id'" not in captured.err
+
+
+def test_run_declarative_flow_loads_project_env(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    # Flow projects must pick up the project's .env, like crew projects do,
+    # overriding any pre-existing value.
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.setenv("DECL_FLOW_ENV_PROBE", "old")
+    (tmp_path / ".env").write_text("DECL_FLOW_ENV_PROBE=from_dotenv\n", encoding="utf-8")
+    path = _write(tmp_path, REQUIRED_FLOW_YAML)
+
+    run_declarative_flow_module.run_declarative_flow(
+        str(path), '{"prospect_email":"a@b.com"}'
+    )
+
+    assert os.environ["DECL_FLOW_ENV_PROBE"] == "from_dotenv"
+
+
+def test_id_only_input_skips_required_validation(tmp_path: Path) -> None:
+    # Resume via `crewai run --inputs '{"id":"..."}'` must not be blocked by the
+    # required-field check: kickoff hydrates required state from persistence.
+    path = _write(tmp_path, REQUIRED_FLOW_YAML)
+    flow = run_declarative_flow_module.load_declarative_flow(str(path))
+
+    resolved = run_declarative_flow_module._resolve_flow_inputs(flow, {"id": "run-123"})
+
+    assert resolved == {"id": "run-123"}

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/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/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",

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/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_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