diff --git a/lib/cli/src/crewai_cli/cli.py b/lib/cli/src/crewai_cli/cli.py index 4c922f84c..0250289d3 100644 --- a/lib/cli/src/crewai_cli/cli.py +++ b/lib/cli/src/crewai_cli/cli.py @@ -48,6 +48,12 @@ def run_crew(*args: Any, **kwargs: Any) -> Any: return _run_crew(*args, **kwargs) +def eval_crew(*args: Any, **kwargs: Any) -> Any: + from crewai_cli.eval_crew import eval_crew as _eval_crew + + return _eval_crew(*args, **kwargs) + + if TYPE_CHECKING: # mypy sees the real classes; at runtime the shims below defer the # heavy imports until a command actually instantiates them. @@ -674,6 +680,23 @@ def run( ) +@crewai.command(name="eval") +@click.option( + "--run", + "run_id", + type=str, + default=None, + metavar="EXECUTION_ID", + help=( + "Evaluate this traced run instead of the last one. The execution id " + "crewAI recorded for the run." + ), +) +def eval_command(run_id: str | None) -> None: + """Evaluate the last traced run through CrewAI AMP.""" + eval_crew(run_id=run_id) + + @crewai.command() def update() -> None: """Update the pyproject.toml of the Crew project to use uv.""" diff --git a/lib/cli/src/crewai_cli/eval_crew.py b/lib/cli/src/crewai_cli/eval_crew.py new file mode 100644 index 000000000..308def5d6 --- /dev/null +++ b/lib/cli/src/crewai_cli/eval_crew.py @@ -0,0 +1,377 @@ +"""`crewai eval`: evaluate the last traced run through CrewAI AMP. + +crewAI records a traced run in `.crewai/last_run.json` when the run's spans +reach Wharf. This command reads that record (or takes `--run EXECUTION_ID`), +asks AMP to evaluate the run, prints and opens the URL AMP answers with, +waits for the verdict and prints it. With no traced run recorded it offers +to turn tracing on for the project and run the crew now. + +Who may evaluate what is AMP's decision: an anonymous run once without an +account, then it needs one; a run traced while logged in for that +organization's members; a deployment execution for members who may see its +traces. The command sends the saved `crewai login` when there is one. +""" + +from __future__ import annotations + +import contextlib +from ipaddress import ip_address +import json +import os +from pathlib import Path +import sys +import time +from typing import Any +from urllib.parse import urlparse +import webbrowser + +import click +from crewai_core.constants import DEFAULT_CREWAI_ENTERPRISE_URL +from crewai_core.settings import Settings +from dotenv import load_dotenv, set_key +import httpx +from rich.console import Console + +from crewai_cli.authentication.token import AuthError, get_auth_token +from crewai_cli.plus_api import PlusAPI +from crewai_cli.utils import get_or_create_project_id, is_dmn_mode_enabled + + +console = Console() + +LAST_RUN_FILE = Path(".crewai") / "last_run.json" +TRACING_ENV_VAR = "CREWAI_TRACING_ENABLED" +POLL_SECONDS = 3.0 +POLL_RETRIES = 5 # consecutive unreachable / 5xx polls before giving up; the evaluation keeps running +FINISHED = {"done", "failed"} +STATUSES = {"queued", "running"} | FINISHED + + +def eval_crew(run_id: str | None = None) -> None: + """Evaluate the last traced run of this project, or the run RUN_ID.""" + get_or_create_project_id() + # Read before the project's .env is loaded, so a project cannot add itself. + trusted = _trusted_amp_origins() + _load_project_env() + record = read_last_run() or {} + execution_id = run_id or record.get("execution_id") + if execution_id is None: + execution_id = _run_now_or_explain() + record = read_last_run() or {} + + client = _amp_client(trusted) + recorded_amp = str(record.get("amp_base_url") or "").rstrip("/") + if not run_id and recorded_amp and recorded_amp != client.base_url.rstrip("/"): + console.print( + f"The run was traced to {recorded_amp}; evaluating at the configured AMP {client.base_url}.", + style="yellow", + ) + started = _start_evaluation(client, execution_id) + url = started.get("url") + console.print(f"Evaluating run [bold]{execution_id}[/bold]") + if url: + console.print(f"Follow it at [cyan underline]{url}[/cyan underline]") + _open(url) + + finished = _wait(client, str(started["id"]), url) + _print_verdict(finished, url) + if finished.get("status") != "done": + raise SystemExit(1) + + +def _trusted_amp_origins() -> set[str]: + """Where the saved login may be sent: the AMP this machine is configured for + (`crewai enterprise configure`), one already exported in this shell, and + crewAI's own.""" + candidates = ( + os.environ.get("CREWAI_PLUS_URL"), + Settings().enterprise_base_url, + DEFAULT_CREWAI_ENTERPRISE_URL, + ) + return {origin for origin in map(_origin, candidates) if origin} + + +def _origin(url: str | None) -> str | None: + parsed = urlparse(str(url or "")) + return ( + f"{parsed.scheme}://{parsed.netloc}".lower() + if parsed.scheme and parsed.netloc + else None + ) + + +def _encrypted(origin: str | None) -> bool: + """HTTPS, or plain HTTP to this machine — the rule `TraceGrantClient` already + applies to collector grants (localhost, its subdomains, loopback addresses).""" + parsed = urlparse(origin or "") + if parsed.scheme == "https": + return True + if parsed.scheme != "http": + return False + hostname = (parsed.hostname or "").rstrip(".") + if hostname == "localhost" or hostname.endswith(".localhost"): + return True + try: + return ip_address(hostname).is_loopback + except ValueError: + return False + + +def _amp_client(trusted: set[str]) -> PlusAPI: + """The AMP to ask, and whether the saved login goes with it. + + A project's `.env` may point `crewai eval` at another AMP — that is how a + self-hosted project is wired, and the run was traced there — so the request + follows it. The credential does not: it goes only to an AMP this machine is + logged in to, over a connection that encrypts it. Anywhere else the run is + read anonymously.""" + client = PlusAPI(api_key=saved_login()) + if client.api_key is None: + return client + + origin = _origin(client.base_url) + if origin in trusted and _encrypted(origin): + return client + + why = ( + "is not an AMP this machine is logged in to" + if origin not in trusted + else "would carry the login over plain HTTP" + ) + console.print( + f"Reading anonymously: {client.base_url} {why}. " + "Run `crewai enterprise configure ` to log in to it.", + style="yellow", + ) + return PlusAPI() + + +def _load_project_env() -> None: + """The project's .env, as `crewai run` loads it — so CREWAI_PLUS_URL here is the one the run used.""" + env_file = Path.cwd() / ".env" + if env_file.is_file(): + load_dotenv(env_file, override=True) + + +def read_last_run(directory: Path | None = None) -> dict[str, Any] | None: + """The record crewAI wrote for the project's last traced run (execution_id, + tier, started_at, finished_at, amp_base_url), or None.""" + path = (directory or Path.cwd()) / LAST_RUN_FILE + try: + loaded = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return None + if not isinstance(loaded, dict) or not loaded.get("execution_id"): + return None + loaded["execution_id"] = str(loaded["execution_id"]) + return loaded + + +def saved_login() -> str | None: + """The `crewai login` token, or None: AMP then treats the caller as anonymous. + + Only "not logged in" reads as anonymous. A credential that exists but cannot + be read — a rotated key, a half-written store, a directory `sudo` left owned + by root — is said out loud: reading anonymously instead would quietly spend + the run's one anonymous read and then refuse a user who believes they are + logged in.""" + try: + return get_auth_token() + except AuthError: + return None + except Exception as error: + _fail( + f"Could not read the saved login ({type(error).__name__}: {error}). " + "Run `crewai login` again, or `crewai eval` will not know who you are." + ) + return None + + +def _run_now_or_explain() -> str: + """No traced run recorded here: offer to turn tracing on and run the crew now.""" + if not Path("pyproject.toml").is_file(): + _fail( + "No crewAI project here (no pyproject.toml). Run `crewai eval` from the project's " + "directory, or name a run: `crewai eval --run EXECUTION_ID`." + ) + steps = ( + "No traced run is recorded in this project. Turn tracing on and run the crew, " + f"then come back:\n 1. add {TRACING_ENV_VAR}=true to .env\n 2. crewai run\n 3. crewai eval" + ) + if is_dmn_mode_enabled() or not sys.stdin.isatty(): + console.print(steps, style="yellow") + raise SystemExit(1) + if not click.confirm( + f"No traced run is recorded in this project. Turn tracing on ({TRACING_ENV_VAR}=true " + "stays in .env) and run the crew now?", + default=True, # João, 2026-09-20: y/n with Y as the default — the prompt says what Enter does + ): + console.print(steps, style="yellow") + raise SystemExit(0) + + _enable_tracing() + from crewai_cli.run_crew import run_crew + + run_crew() + record = read_last_run() + if record is None: + console.print( + "The run finished but no trace was recorded: the run may have failed, sharing the " + "trace was declined, or this project's crewai is older than the version that records " + f"the last run ({LAST_RUN_FILE}). Run the crew again and accept when asked, then `crewai eval`.", + style="bold red", + ) + raise SystemExit(1) + return str(record["execution_id"]) + + +def _enable_tracing() -> None: + """`CREWAI_TRACING_ENABLED=true` in the project's .env, and in this process for the run about to start.""" + env_file = Path.cwd() / ".env" + env_file.touch(exist_ok=True) + set_key(str(env_file), TRACING_ENV_VAR, "true", quote_mode="never") + os.environ[TRACING_ENV_VAR] = "true" + console.print( + f"Tracing is on for this project ({TRACING_ENV_VAR}=true in .env).", + style="green", + ) + + +def _start_evaluation(client: PlusAPI, execution_id: str) -> dict[str, Any]: + try: + response = client.create_evaluation(execution_id) + except httpx.HTTPError as error: + _fail(f"Could not reach AMP to start the evaluation: {error}") + if response.status_code in (200, 202): + payload = _payload(response) + if payload and payload.get("id"): + return payload + _fail(f"AMP answered without an evaluation id ({response.status_code}).") + _refused(response, f"run {execution_id}") + raise AssertionError("unreachable") + + +def _wait(client: PlusAPI, evaluation_id: str, url: str | None) -> dict[str, Any]: + """Poll until the evaluation is done or failed; Ctrl-C leaves it running.""" + console.print("Waiting for the verdict…", style="dim") + where = f" at {url}" if url else "" + subject = f"evaluation {evaluation_id}" + misses = ( + 0 # AMP unreachable or answering 5xx: a blip is retried, a streak is reported + ) + try: + while True: + try: + response = client.get_evaluation(evaluation_id) + except httpx.HTTPError as error: + misses += 1 + if misses >= POLL_RETRIES: + _fail( + f"Could not reach AMP while waiting ({error}); the evaluation keeps running{where}." + ) + time.sleep(POLL_SECONDS) + continue + if response.status_code >= 500: + misses += 1 + if misses >= POLL_RETRIES: + _refused(response, subject) + time.sleep(POLL_SECONDS) + continue + if response.status_code != 200: + _refused(response, subject) + misses = 0 + payload = _payload(response) or {} + status = payload.get("status") + if status == "done" and not _well_formed_verdict(payload.get("verdict")): + _fail( + f"AMP answered done without a verdict (protocol error); follow it{where or ' on AMP'}." + ) + if status in FINISHED: + return payload + if status not in STATUSES: + _fail( + f"AMP answered without a known evaluation status ({status!r}); follow it{where or ' on AMP'}." + ) + time.sleep(POLL_SECONDS) + except KeyboardInterrupt: + console.print(f"\nStill running{where}.", style="yellow") + raise SystemExit(130) from None + + +def _well_formed_verdict(verdict: Any) -> bool: + """`{"gate": "", "grades": {area: 1..5 | null}}` — anything else is a + protocol error. A grade is an exact integer in range: `True` is an `int` to + Python and 6 is not a grade, and neither may print as one.""" + return ( + isinstance(verdict, dict) + and isinstance(verdict.get("gate"), str) + and isinstance(verdict.get("grades"), dict) + and all(_a_grade(grade) for grade in verdict["grades"].values()) + ) + + +def _a_grade(grade: Any) -> bool: + return grade is None or (type(grade) is int and 1 <= grade <= 5) + + +def _print_verdict(finished: dict[str, Any], url: str | None) -> None: + if finished.get("status") != "done": + console.print( + f"Evaluation failed: {finished.get('error') or 'no reason given'}", + style="bold red", + ) + return + verdict = finished["verdict"] # _wait let only a well-formed one through + gate = str(verdict["gate"]).upper() + style = {"PASSED": "bold green", "FAILED": "bold red"}.get(gate, "bold yellow") + grades = verdict.get("grades") or {} + parts = [ + f"{area} {grades[area]}/5" + if grades.get(area) is not None + else f"{area} not measured" + for area in ("goal", "quality", "process", "cost") + ] + console.print(f"Goal gate: [{style}]{gate}[/{style}] · " + " · ".join(parts)) + if url: + console.print(f"Full report: {url}") + + +def _open(url: str) -> None: + if is_dmn_mode_enabled(): + return + with contextlib.suppress(Exception): # no browser is not an error + webbrowser.open(url) + + +def _payload(response: httpx.Response) -> dict[str, Any] | None: + try: + loaded = response.json() + except ValueError: + return None + return loaded if isinstance(loaded, dict) else None + + +def _refused(response: httpx.Response, subject: str) -> None: + """AMP's own words when it sent them, then exit 1. SUBJECT is "run " or "evaluation ".""" + payload = _payload(response) or {} + message = str(payload.get("message") or "").strip() + error = str(payload.get("error") or "") + if response.status_code in (401, 403): + if error == "account_required" and message: + _fail(message) + _fail( + f"{message or 'AMP refused the credential'}. Log in with `crewai login` and try again." + ) + if response.status_code == 404: + _fail(message or f"AMP answered 404 for {subject}.") + if response.status_code == 429: + retry = response.headers.get("Retry-After") + _fail( + f"{message or 'AMP is rate limiting this request'}{f' — retry after {retry}s' if retry else ''}." + ) + _fail(f"AMP answered {response.status_code}{': ' + message if message else ''}.") + + +def _fail(message: str) -> None: + console.print(message, style="bold red") + raise SystemExit(1) diff --git a/lib/cli/src/crewai_cli/plus_api.py b/lib/cli/src/crewai_cli/plus_api.py index 6f94d96d3..f03661e5b 100644 --- a/lib/cli/src/crewai_cli/plus_api.py +++ b/lib/cli/src/crewai_cli/plus_api.py @@ -18,9 +18,32 @@ class PlusAPI(_CorePlusAPI): The ZIP deployment methods live here as well as in newer crewai-core versions so editable CLI installs still work when an older crewai-core is - present in the runtime environment. + present in the runtime environment. The evaluation methods live here + because only the CLI calls them. """ + EVALUATIONS_RESOURCE = f"{_CorePlusAPI.TRACING_RESOURCE}/evaluations" + # AMP reads the run's spans from Wharf inside the POST; a large run takes a while. + EVALUATION_START_TIMEOUT = 120.0 + EVALUATION_POLL_TIMEOUT = 30.0 + + def create_evaluation(self, execution_id: str) -> httpx.Response: + """Ask AMP to evaluate the traced run EXECUTION_ID (crewai eval).""" + return self._make_request( + "POST", + self.EVALUATIONS_RESOURCE, + json={"execution_id": execution_id}, + timeout=self.EVALUATION_START_TIMEOUT, + ) + + def get_evaluation(self, evaluation_id: str) -> httpx.Response: + """The evaluation's status and, once done, its verdict.""" + return self._make_request( + "GET", + f"{self.EVALUATIONS_RESOURCE}/{evaluation_id}", + timeout=self.EVALUATION_POLL_TIMEOUT, + ) + def _make_multipart_request( self, method: HttpMethod, diff --git a/lib/cli/tests/test_eval_crew.py b/lib/cli/tests/test_eval_crew.py new file mode 100644 index 000000000..4c91e3fe3 --- /dev/null +++ b/lib/cli/tests/test_eval_crew.py @@ -0,0 +1,505 @@ +"""`crewai eval`: the last traced run, evaluated through AMP.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path +from types import SimpleNamespace + +from click.testing import CliRunner +import httpx +import pytest +from rich.console import Console + +from crewai_cli import eval_crew as eval_module +from crewai_cli.cli import eval_command + + +EXECUTION_ID = "6f31fe1a-20bd-4bfe-a011-25d6b9341f62" +URL = "https://evolve.crewai.test/e/ev-1" + + +class FakeAMP: + """A PlusAPI double: scripted answers, calls recorded.""" + + def __init__(self, create=None, statuses=None): + self.create = create if create is not None else httpx.Response( + 202, json={"id": "ev-1", "url": URL, "status": "queued"} + ) + self.statuses = list(statuses or []) + self.calls: list[tuple] = [] + self.api_key = None + + def create_evaluation(self, execution_id): + self.calls.append(("create", execution_id)) + return self.create + + def get_evaluation(self, evaluation_id): + self.calls.append(("get", evaluation_id)) + return self.statuses.pop(0) if self.statuses else httpx.Response(200, json={"id": evaluation_id, "status": "running"}) + + +def done(gate="passed", grades=None): + return httpx.Response(200, json={ + "id": "ev-1", "status": "done", "url": URL, + "verdict": {"gate": gate, "grades": grades if grades is not None else {"goal": 5, "quality": 4, "process": 5, "cost": None}}, + }) + + +@pytest.fixture +def project(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + monkeypatch.setattr(eval_module, "console", Console(width=240)) # one sentence per line in the captured output + monkeypatch.setattr(eval_module, "get_or_create_project_id", lambda: None) + monkeypatch.setattr(eval_module, "saved_login", lambda: "login-token") + monkeypatch.setattr(eval_module.time, "sleep", lambda seconds: None) + monkeypatch.setattr(eval_module, "is_dmn_mode_enabled", lambda: False) + # This machine is logged in to https://amp.test (`crewai enterprise configure`). + monkeypatch.setattr(eval_module, "Settings", lambda: SimpleNamespace(enterprise_base_url="https://amp.test")) + monkeypatch.delenv("CREWAI_PLUS_URL", raising=False) + opened: list[str] = [] + monkeypatch.setattr(eval_module.webbrowser, "open", lambda url: opened.append(url) or True) + return tmp_path, opened + + +def record_last_run(directory: Path, execution_id: str = EXECUTION_ID, **fields) -> None: + (directory / ".crewai").mkdir(exist_ok=True) + record = {"execution_id": execution_id, "tier": "ephemeral", "amp_base_url": "https://amp.test", **fields} + (directory / ".crewai" / "last_run.json").write_text(json.dumps(record)) + + +def install(monkeypatch, amp: FakeAMP, configured_amp: str = "https://amp.test") -> FakeAMP: + def build(api_key=None, base_url=None): + # PlusAPI's own resolution: explicit, then CREWAI_PLUS_URL, then the saved settings. + amp.api_key = api_key + amp.base_url = base_url or os.environ.get("CREWAI_PLUS_URL") or configured_amp + return amp + + monkeypatch.setattr(eval_module, "PlusAPI", build) + return amp + + +def test_the_last_run_is_evaluated_the_url_opened_and_the_verdict_printed(project, monkeypatch, capsys): + directory, opened = project + record_last_run(directory) + amp = install(monkeypatch, FakeAMP(statuses=[httpx.Response(200, json={"id": "ev-1", "status": "running"}), done()])) + + eval_module.eval_crew() + + out = capsys.readouterr().out + assert amp.api_key == "login-token" + assert amp.calls == [("create", EXECUTION_ID), ("get", "ev-1"), ("get", "ev-1")] + assert opened == [URL] + assert EXECUTION_ID in out and URL in out + assert "Goal gate: PASSED" in out and "goal 5/5" in out and "cost not measured" in out + + +def test_run_names_another_execution_and_an_anonymous_caller_sends_no_token(project, monkeypatch, capsys): + directory, _ = project + record_last_run(directory) + monkeypatch.setattr(eval_module, "saved_login", lambda: None) + amp = install(monkeypatch, FakeAMP(statuses=[done("failed")])) + + eval_module.eval_crew(run_id="other-run") + + assert amp.api_key is None + assert amp.calls[0] == ("create", "other-run") + assert "Goal gate: FAILED" in capsys.readouterr().out + + +def test_a_failed_evaluation_exits_one_with_amps_reason(project, monkeypatch, capsys): + directory, _ = project + record_last_run(directory) + install(monkeypatch, FakeAMP(statuses=[httpx.Response(200, json={"id": "ev-1", "status": "failed", "error": "the judge was unreachable"})])) + + with pytest.raises(SystemExit) as exit_: + eval_module.eval_crew() + + assert exit_.value.code == 1 + assert "Evaluation failed: the judge was unreachable" in capsys.readouterr().out + + +@pytest.mark.parametrize( + ("response", "expected"), + [ + (httpx.Response(401, json={"error": "account_required", "message": "Execution x was already read once without an account. Log in with `crewai login`, or create an account, to read it again."}), + "already read once without an account"), + (httpx.Response(401, json={"error": "bad_credentials", "message": "Bad credentials"}), "Bad credentials. Log in with `crewai login`"), + (httpx.Response(404, json={"error": "trace_not_found", "message": "No spans recorded for execution 6f31"}), "No spans recorded for execution 6f31"), + (httpx.Response(404, text="Page not found"), f"AMP answered 404 for run {EXECUTION_ID}."), + (httpx.Response(202, json={"url": URL}), "AMP answered without an evaluation id (202)."), + (httpx.Response(429, json={"error": "rate_limit_exceeded", "message": "Too many requests"}, headers={"Retry-After": "60"}), "Too many requests — retry after 60s"), + (httpx.Response(503, json={"error": "service_unavailable", "message": "Wharf could not list the spans"}), "AMP answered 503: Wharf could not list the spans"), + (httpx.Response(500, text="boom"), "AMP answered 500."), + ], +) +def test_amps_refusals_are_printed_in_its_words_and_exit_one(project, monkeypatch, capsys, response, expected): + directory, opened = project + record_last_run(directory) + install(monkeypatch, FakeAMP(create=response)) + + with pytest.raises(SystemExit) as exit_: + eval_module.eval_crew() + + assert exit_.value.code == 1 + assert expected in capsys.readouterr().out + assert opened == [] + + +def test_the_credential_goes_only_to_the_configured_amp_never_to_an_address_off_the_record(project, monkeypatch, capsys): + directory, _ = project + record_last_run(directory, amp_base_url="https://evil.example/steal") + amp = install(monkeypatch, FakeAMP(statuses=[done()]), configured_amp="https://app.crewai.com") + + eval_module.eval_crew() + + assert amp.api_key == "login-token" and amp.base_url == "https://app.crewai.com" # PlusAPI got no base_url + out = capsys.readouterr().out + assert "The run was traced to https://evil.example/steal; evaluating at the configured AMP https://app.crewai.com." in out + + # The project's .env is what `crewai run` traced with, so it is loaded first. This machine is + # logged in to that AMP, so the credential goes with the request and nothing is remarked on. + (directory / ".env").write_text("CREWAI_PLUS_URL=https://amp.test\n") + record_last_run(directory, amp_base_url="https://amp.test/") + amp = install(monkeypatch, FakeAMP(statuses=[done()])) + eval_module.eval_crew() + assert eval_module.os.environ["CREWAI_PLUS_URL"] == "https://amp.test" + assert amp.api_key == "login-token" and amp.base_url == "https://amp.test" + out = capsys.readouterr().out + assert "was traced to" not in out and "Reading anonymously" not in out + + +def test_a_project_may_point_at_another_amp_but_never_gets_the_saved_login(project, monkeypatch, capsys): + """A .env can send the request elsewhere — that is how a self-hosted project is wired — + but the token goes only to an AMP this machine is logged in to.""" + directory, _ = project + (directory / ".env").write_text("CREWAI_PLUS_URL=https://evil.example\n") + record_last_run(directory, amp_base_url="https://evil.example") + amp = install(monkeypatch, FakeAMP(statuses=[done()])) + + eval_module.eval_crew() # still works: AMP reads it anonymously + + assert amp.base_url == "https://evil.example" # the request follows the project + assert amp.api_key is None # the credential does not + out = capsys.readouterr().out + assert "Reading anonymously: https://evil.example is not an AMP this machine is logged in to." in out + assert "crewai enterprise configure" in out + + +def test_a_trusted_amp_over_plain_http_still_gets_no_credential(project, monkeypatch, capsys): + """A cleartext connection is not a place to put a bearer token, trusted or not.""" + directory, _ = project + monkeypatch.setenv("CREWAI_PLUS_URL", "http://amp.internal") # exported, so it IS trusted + record_last_run(directory, amp_base_url="http://amp.internal") + amp = install(monkeypatch, FakeAMP(statuses=[done()])) + + eval_module.eval_crew() + + assert amp.base_url == "http://amp.internal" and amp.api_key is None + assert "would carry the login over plain HTTP" in capsys.readouterr().out + + +def test_plain_http_to_this_machine_is_fine_for_local_development(project, monkeypatch, capsys): + directory, _ = project + monkeypatch.setenv("CREWAI_PLUS_URL", "http://localhost:3000") + record_last_run(directory, amp_base_url="http://localhost:3000") + amp = install(monkeypatch, FakeAMP(statuses=[done()])) + + eval_module.eval_crew() + + assert amp.api_key == "login-token" + assert "Reading anonymously" not in capsys.readouterr().out + + +@pytest.mark.parametrize( + ("origin", "encrypted"), + [ + ("https://app.crewai.com", True), + ("http://localhost:3000", True), + ("http://127.0.0.1:8000", True), + ("http://[::1]:8000", True), + ("http://amp.localhost", True), + ("http://amp.internal", False), + ("http://169.254.169.254", False), + ("ftp://amp.test", False), + (None, False), + ], +) +def test_which_connections_may_carry_the_login(origin, encrypted): + assert eval_module._encrypted(origin) is encrypted + + +def test_an_amp_exported_in_this_shell_is_trusted(project, monkeypatch, capsys): + directory, _ = project + monkeypatch.setenv("CREWAI_PLUS_URL", "https://shell.amp.test") # exported before the project is read + record_last_run(directory, amp_base_url="https://shell.amp.test") + amp = install(monkeypatch, FakeAMP(statuses=[done()])) + + eval_module.eval_crew() + + assert amp.base_url == "https://shell.amp.test" and amp.api_key == "login-token" + assert "Reading anonymously" not in capsys.readouterr().out + + +@pytest.mark.parametrize( + ("url", "origin"), + [ + ("https://amp.test", "https://amp.test"), + ("https://AMP.Test/crewai_plus/", "https://amp.test"), + ("http://localhost:8000/x", "http://localhost:8000"), + ("app.crewai.com", None), # no scheme: not an origin, never trusted + ("", None), + (None, None), + ], +) +def test_an_origin_is_scheme_and_host_only(url, origin): + assert eval_module._origin(url) == origin + + +@pytest.mark.parametrize( + "verdict", + [ + None, + "passed", + [], + {"gate": "passed"}, + {"gate": None, "grades": {}}, + {"gate": "passed", "grades": "5/5"}, + {"gate": "passed", "grades": {"goal": "five"}}, + {"gate": "passed", "grades": {"goal": True}}, # a bool is an int to Python, never a grade + {"gate": "passed", "grades": {"goal": 6}}, # out of the 1..5 range + {"gate": "passed", "grades": {"goal": 0}}, + {"gate": "passed", "grades": {"goal": 4.5}}, + ], +) +def test_a_done_answer_without_a_well_formed_verdict_is_a_protocol_error(project, monkeypatch, capsys, verdict): + directory, _ = project + record_last_run(directory) + body = {"id": "ev-1", "status": "done", "url": URL} + if verdict is not None: + body["verdict"] = verdict + install(monkeypatch, FakeAMP(statuses=[httpx.Response(200, json=body)])) + + with pytest.raises(SystemExit) as exit_: + eval_module.eval_crew() + + assert exit_.value.code == 1 + assert f"AMP answered done without a verdict (protocol error); follow it at {URL}." in capsys.readouterr().out + + +def test_amp_unreachable_at_the_start_is_a_sentence_not_a_traceback(project, monkeypatch, capsys): + directory, _ = project + record_last_run(directory) + amp = install(monkeypatch, FakeAMP()) + monkeypatch.setattr(amp, "create_evaluation", lambda execution_id: (_ for _ in ()).throw(httpx.ConnectError("connection refused"))) + + with pytest.raises(SystemExit) as exit_: + eval_module.eval_crew() + + assert exit_.value.code == 1 + assert "Could not reach AMP to start the evaluation: connection refused" in capsys.readouterr().out + + +def test_while_waiting_a_blip_is_retried_and_a_streak_is_reported(project, monkeypatch, capsys): + directory, _ = project + record_last_run(directory) + amp = install(monkeypatch, FakeAMP(statuses=[httpx.Response(502), httpx.Response(503, json={"error": "service_unavailable", "message": "crew-optimize is down"}), done()])) + + eval_module.eval_crew() # two bad polls, then the verdict + + assert "Goal gate: PASSED" in capsys.readouterr().out + assert amp.calls.count(("get", "ev-1")) == 3 + + record_last_run(directory) + amp = install(monkeypatch, FakeAMP(statuses=[httpx.Response(503, json={"error": "service_unavailable", "message": "crew-optimize is down"})] * eval_module.POLL_RETRIES)) + with pytest.raises(SystemExit) as exit_: + eval_module.eval_crew() + assert exit_.value.code == 1 + assert "AMP answered 503: crew-optimize is down" in capsys.readouterr().out + assert amp.calls.count(("get", "ev-1")) == eval_module.POLL_RETRIES + + amp = install(monkeypatch, FakeAMP()) + monkeypatch.setattr(amp, "get_evaluation", lambda evaluation_id: (_ for _ in ()).throw(httpx.ReadTimeout("timed out"))) + with pytest.raises(SystemExit) as exit_: + eval_module.eval_crew() + assert exit_.value.code == 1 + assert f"Could not reach AMP while waiting (timed out); the evaluation keeps running at {URL}." in capsys.readouterr().out + + +def test_a_refusal_mid_poll_names_the_evaluation_and_an_unknown_status_stops_the_wait(project, monkeypatch, capsys): + directory, _ = project + record_last_run(directory) + install(monkeypatch, FakeAMP(statuses=[httpx.Response(404, text="gone")])) + with pytest.raises(SystemExit) as exit_: + eval_module.eval_crew() + assert exit_.value.code == 1 and "AMP answered 404 for evaluation ev-1." in capsys.readouterr().out + + for odd in (httpx.Response(200, json={"id": "ev-1", "status": "cancelled"}), httpx.Response(200, json=[]), httpx.Response(200, text="")): + install(monkeypatch, FakeAMP(statuses=[httpx.Response(200, json={"id": "ev-1", "status": "queued"}), odd])) + with pytest.raises(SystemExit) as exit_: + eval_module.eval_crew() + assert exit_.value.code == 1 + assert f"AMP answered without a known evaluation status" in capsys.readouterr().out + + +def test_ctrl_c_leaves_the_evaluation_running_and_exits_130(project, monkeypatch, capsys): + directory, _ = project + record_last_run(directory) + amp = install(monkeypatch, FakeAMP()) + monkeypatch.setattr(amp, "get_evaluation", lambda evaluation_id: (_ for _ in ()).throw(KeyboardInterrupt())) + + with pytest.raises(SystemExit) as exit_: + eval_module.eval_crew() + + assert exit_.value.code == 130 + assert f"Still running at {URL}." in capsys.readouterr().out + + +def test_dmn_mode_prints_the_url_but_opens_no_browser(project, monkeypatch, capsys): + directory, opened = project + record_last_run(directory) + monkeypatch.setattr(eval_module, "is_dmn_mode_enabled", lambda: True) + install(monkeypatch, FakeAMP(statuses=[done()])) + + eval_module.eval_crew() + + assert opened == [] and URL in capsys.readouterr().out + + +def test_run_skips_the_offer_when_nothing_is_recorded(project, monkeypatch, capsys): + monkeypatch.setattr(eval_module.click, "confirm", lambda *args, **kwargs: pytest.fail("no offer with --run")) + amp = install(monkeypatch, FakeAMP(statuses=[done()])) + + eval_module.eval_crew(run_id="named-run") + + assert amp.calls[0] == ("create", "named-run") + + +def test_outside_a_crewai_project_nothing_is_written_and_it_says_so(project, monkeypatch, capsys): + directory, _ = project + monkeypatch.setattr(eval_module.sys.stdin, "isatty", lambda: True) + monkeypatch.setattr(eval_module.click, "confirm", lambda *args, **kwargs: pytest.fail("no offer outside a project")) + amp = install(monkeypatch, FakeAMP()) + + with pytest.raises(SystemExit) as exit_: + eval_module.eval_crew() + + assert exit_.value.code == 1 + assert "No crewAI project here (no pyproject.toml)" in capsys.readouterr().out + assert not (directory / ".env").exists() and amp.calls == [] + + +def test_without_a_traced_run_and_no_terminal_it_explains_and_exits(project, monkeypatch, capsys): + (project[0] / "pyproject.toml").write_text("[project]\nname = 'demo'\n") + monkeypatch.setattr(eval_module, "is_dmn_mode_enabled", lambda: True) + amp = install(monkeypatch, FakeAMP()) + + with pytest.raises(SystemExit) as exit_: + eval_module.eval_crew() + + assert exit_.value.code == 1 + out = capsys.readouterr().out + assert "No traced run is recorded" in out and "CREWAI_TRACING_ENABLED=true" in out and "crewai run" in out + assert amp.calls == [] + + +def test_without_a_traced_run_it_offers_to_turn_tracing_on_and_run_the_crew(project, monkeypatch, capsys): + directory, _ = project + (directory / "pyproject.toml").write_text("[project]\nname = 'demo'\n") + monkeypatch.setattr(eval_module.sys.stdin, "isatty", lambda: True) + prompts: list[tuple] = [] + monkeypatch.setattr(eval_module.click, "confirm", lambda text, **kwargs: prompts.append((text, kwargs)) or True) + ran: list[str] = [] + + def fake_run_crew() -> None: + ran.append("run") + assert eval_module.os.environ.get("CREWAI_TRACING_ENABLED") == "true" + record_last_run(directory, "fresh-run") + + import crewai_cli.run_crew as run_crew_module + + monkeypatch.setattr(run_crew_module, "run_crew", fake_run_crew) + monkeypatch.delenv("CREWAI_TRACING_ENABLED", raising=False) + amp = install(monkeypatch, FakeAMP(statuses=[done()])) + + eval_module.eval_crew() + + assert ran == ["run"] + text, kwargs = prompts[0] + assert "CREWAI_TRACING_ENABLED=true stays in .env" in text and kwargs == {"default": True} # Enter is yes (João's call); the prompt names both effects + assert "CREWAI_TRACING_ENABLED=true" in (directory / ".env").read_text() + assert amp.calls[0] == ("create", "fresh-run") + assert "Tracing is on for this project" in capsys.readouterr().out + + +def test_declining_the_offer_exits_cleanly_with_the_steps(project, monkeypatch, capsys): + (project[0] / "pyproject.toml").write_text("[project]\nname = 'demo'\n") + monkeypatch.setattr(eval_module.sys.stdin, "isatty", lambda: True) + monkeypatch.setattr(eval_module.click, "confirm", lambda *args, **kwargs: False) + amp = install(monkeypatch, FakeAMP()) + + with pytest.raises(SystemExit) as exit_: + eval_module.eval_crew() + + assert exit_.value.code == 0 + assert "crewai eval" in capsys.readouterr().out and amp.calls == [] + + +def test_a_run_that_leaves_no_trace_behind_is_explained(project, monkeypatch, capsys): + (project[0] / "pyproject.toml").write_text("[project]\nname = 'demo'\n") + monkeypatch.setattr(eval_module.sys.stdin, "isatty", lambda: True) + monkeypatch.setattr(eval_module.click, "confirm", lambda *args, **kwargs: True) + import crewai_cli.run_crew as run_crew_module + + monkeypatch.setattr(run_crew_module, "run_crew", lambda: None) + install(monkeypatch, FakeAMP()) + + with pytest.raises(SystemExit) as exit_: + eval_module.eval_crew() + + assert exit_.value.code == 1 + out = capsys.readouterr().out + assert "no trace was recorded" in out and "older than the version that records the last run" in out + + +def test_the_cli_command_maps_to_the_implementation(monkeypatch): + calls = [] + monkeypatch.setattr("crewai_cli.cli.eval_crew", lambda **kwargs: calls.append(kwargs)) + runner = CliRunner() + + assert runner.invoke(eval_command, []).exit_code == 0 + assert runner.invoke(eval_command, ["--run", EXECUTION_ID]).exit_code == 0 + assert calls == [{"run_id": None}, {"run_id": EXECUTION_ID}] + assert "Evaluate the last traced run" in runner.invoke(eval_command, ["--help"]).output + + +def test_only_a_missing_login_reads_as_anonymous(monkeypatch, capsys): + """An unreadable credential store is not "anonymous": it is said out loud.""" + from crewai_cli.authentication.token import AuthError + + monkeypatch.setattr(eval_module, "get_auth_token", lambda: (_ for _ in ()).throw(AuthError("No token found"))) + assert eval_module.saved_login() is None # not logged in: AMP treats the caller as anonymous + + monkeypatch.setattr(eval_module, "get_auth_token", lambda: "login-token") + assert eval_module.saved_login() == "login-token" + + for broken in (OSError(13, "Permission denied"), ValueError("Fernet key must be 32 url-safe base64-encoded bytes.")): + monkeypatch.setattr(eval_module, "get_auth_token", lambda error=broken: (_ for _ in ()).throw(error)) + with pytest.raises(SystemExit) as exit_: + eval_module.saved_login() + assert exit_.value.code == 1 + out = capsys.readouterr().out + assert "Could not read the saved login" in out and type(broken).__name__ in out and "crewai login" in out + + +def test_read_last_run_reads_the_record_crewai_writes(tmp_path): + assert eval_module.read_last_run(tmp_path) is None + (tmp_path / ".crewai").mkdir() + (tmp_path / ".crewai" / "last_run.json").write_text("not json") + assert eval_module.read_last_run(tmp_path) is None + (tmp_path / ".crewai" / "last_run.json").write_text(json.dumps({"tier": "ephemeral"})) + assert eval_module.read_last_run(tmp_path) is None + record_last_run(tmp_path) + record = eval_module.read_last_run(tmp_path) + assert record is not None and record["execution_id"] == EXECUTION_ID and record["amp_base_url"] == "https://amp.test" diff --git a/lib/cli/tests/test_plus_api.py b/lib/cli/tests/test_plus_api.py index 16cf684d5..e40cfdbc2 100644 --- a/lib/cli/tests/test_plus_api.py +++ b/lib/cli/tests/test_plus_api.py @@ -18,6 +18,33 @@ class TestPlusAPI(unittest.TestCase): self.assertIn("CrewAI-CLI/", self.api.headers["User-Agent"]) self.assertTrue(self.api.headers["X-Crewai-Version"]) + @patch("crewai_core.plus_api.PlusAPI._make_request") + def test_create_evaluation(self, mock_make_request): + mock_response = MagicMock() + mock_make_request.return_value = mock_response + + response = self.api.create_evaluation("6f31fe1a-20bd-4bfe-a011-25d6b9341f62") + + mock_make_request.assert_called_once_with( + "POST", + "/crewai_plus/api/v1/tracing/evaluations", + json={"execution_id": "6f31fe1a-20bd-4bfe-a011-25d6b9341f62"}, + timeout=120.0, # AMP reads the run's spans inside this request + ) + self.assertEqual(response, mock_response) + + @patch("crewai_core.plus_api.PlusAPI._make_request") + def test_get_evaluation(self, mock_make_request): + mock_response = MagicMock() + mock_make_request.return_value = mock_response + + response = self.api.get_evaluation("ev-1") + + mock_make_request.assert_called_once_with( + "GET", "/crewai_plus/api/v1/tracing/evaluations/ev-1", timeout=30.0 + ) + self.assertEqual(response, mock_response) + @patch("crewai_core.plus_api.PlusAPI._make_request") def test_login_to_tool_repository(self, mock_make_request): mock_response = MagicMock()