diff --git a/lib/cli/src/crewai_cli/templates/crew/.gitignore b/lib/cli/src/crewai_cli/templates/crew/.gitignore index 7279347af..3d3629134 100644 --- a/lib/cli/src/crewai_cli/templates/crew/.gitignore +++ b/lib/cli/src/crewai_cli/templates/crew/.gitignore @@ -1,3 +1,4 @@ .env __pycache__/ .DS_Store +.crewai/ diff --git a/lib/cli/src/crewai_cli/templates/flow/.gitignore b/lib/cli/src/crewai_cli/templates/flow/.gitignore index 3b6f1bec0..161f444dc 100644 --- a/lib/cli/src/crewai_cli/templates/flow/.gitignore +++ b/lib/cli/src/crewai_cli/templates/flow/.gitignore @@ -2,3 +2,4 @@ __pycache__/ lib/ .DS_Store +.crewai/ diff --git a/lib/cli/src/crewai_cli/templates/json_crew/.gitignore b/lib/cli/src/crewai_cli/templates/json_crew/.gitignore index 012e44489..e0e581eb5 100644 --- a/lib/cli/src/crewai_cli/templates/json_crew/.gitignore +++ b/lib/cli/src/crewai_cli/templates/json_crew/.gitignore @@ -2,3 +2,4 @@ __pycache__/ .DS_Store report.md +.crewai/ diff --git a/lib/cli/tests/test_scaffold_gitignore.py b/lib/cli/tests/test_scaffold_gitignore.py new file mode 100644 index 000000000..a1c8cf941 --- /dev/null +++ b/lib/cli/tests/test_scaffold_gitignore.py @@ -0,0 +1,17 @@ +"""Every runnable scaffold ignores `.crewai/`, where crewAI records the last run.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + + +TEMPLATES = Path(__file__).resolve().parents[1] / "src" / "crewai_cli" / "templates" + + +@pytest.mark.parametrize("template", ["crew", "flow", "json_crew", "declarative_flow"]) +def test_the_scaffold_gitignore_covers_the_crewai_directory(template): + lines = (TEMPLATES / template / ".gitignore").read_text(encoding="utf-8").splitlines() + assert ".crewai/" in lines + assert ".env" in lines diff --git a/lib/crewai/src/crewai/execution.py b/lib/crewai/src/crewai/execution.py index ceda6969e..b0cde19de 100644 --- a/lib/crewai/src/crewai/execution.py +++ b/lib/crewai/src/crewai/execution.py @@ -164,7 +164,7 @@ def _start_tracing(execution_uuid: str, tracing: bool | None) -> None: def finish_authenticated_trace() -> None: if session.shutdown(): - exporter.show_trace_summary() + exporter.record_export() stack.callback(finish_authenticated_trace) _activate_tracing(ExecutionTrace(session, stack)) diff --git a/lib/crewai/src/crewai/telemetry/tracing/ephemeral.py b/lib/crewai/src/crewai/telemetry/tracing/ephemeral.py index 67c263cf0..0c160a895 100644 --- a/lib/crewai/src/crewai/telemetry/tracing/ephemeral.py +++ b/lib/crewai/src/crewai/telemetry/tracing/ephemeral.py @@ -143,7 +143,10 @@ class EphemeralSpanBuffer(SpanExporter): logger.warning("Ephemeral trace export failed; buffer discarded") finally: exporter.shutdown() - exporter.show_trace_summary() + if ( + not self._dropped + ): # a truncated trace is not recorded: the grader could not read it whole + exporter.record_export() except TraceGrantError as error: logger.warning( "Ephemeral trace grant failed (HTTP %s); buffer discarded", diff --git a/lib/crewai/src/crewai/telemetry/tracing/grants.py b/lib/crewai/src/crewai/telemetry/tracing/grants.py index 7025eae22..9678c3097 100644 --- a/lib/crewai/src/crewai/telemetry/tracing/grants.py +++ b/lib/crewai/src/crewai/telemetry/tracing/grants.py @@ -16,6 +16,7 @@ from crewai_core.plus_api import PlusAPI from opentelemetry.exporter.otlp.proto.common.trace_encoder import encode_spans from opentelemetry.sdk.trace import ReadableSpan from opentelemetry.sdk.trace.export import SpanExportResult, SpanExporter +from rich.console import Console from rich.style import Style from rich.text import Text @@ -25,7 +26,7 @@ from crewai.events.listeners.tracing.utils import ( is_tui_mode, should_suppress_tracing_messages, ) -from crewai.events.utils.console_formatter import ConsoleFormatter +from crewai.telemetry.tracing import last_run from crewai.telemetry.tracing.session import MAX_EXPORT_BATCH_SIZE, otlp_exporter @@ -172,7 +173,13 @@ class TraceGrantClient: class GrantSpanExporter(SpanExporter): - """Refresh a grant before exporting; delegate OTLP transport to the shared path.""" + """Refresh a grant before exporting; delegate OTLP transport to the shared path. + + Once the run's spans have reached Wharf, ``record_export`` writes the + project's ``.crewai/last_run.json`` (``last_run``) so ``crewai eval`` can + find the run without an id being shown to anyone, and prints AMP's viewer + link once for whoever wants to look at the run. + """ def __init__(self, client: TraceGrantClient, grant: TraceGrant): self._client = client @@ -182,8 +189,9 @@ class GrantSpanExporter(SpanExporter): self._trace_url = grant.trace_url self._exported = False self._export_failed = False - self._summary_shown = False - self._suppress_output = should_suppress_tracing_messages() or is_tui_mode() + self._recorded = False + self._first_start_ns: int | None = None + self._last_end_ns: int | None = None @staticmethod def _exporter(grant: TraceGrant) -> SpanExporter: @@ -200,9 +208,28 @@ class GrantSpanExporter(SpanExporter): self._export_failed = True raise self._export_failed |= result != SpanExportResult.SUCCESS - self._exported |= bool(spans) and result == SpanExportResult.SUCCESS + if spans and result == SpanExportResult.SUCCESS: + self._exported = True + self._note_span_times(spans) return result + def _note_span_times(self, spans: Sequence[ReadableSpan]) -> None: + """The run's first start and last end, across every exported batch.""" + starts = [s.start_time for s in spans if s.start_time is not None] + ends = [s.end_time for s in spans if s.end_time is not None] + if starts: + first = min(starts) + self._first_start_ns = ( + first + if self._first_start_ns is None + else min(self._first_start_ns, first) + ) + if ends: + last = max(ends) + self._last_end_ns = ( + last if self._last_end_ns is None else max(self._last_end_ns, last) + ) + def _export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult: result = SpanExportResult.SUCCESS pending = [ @@ -240,6 +267,7 @@ class GrantSpanExporter(SpanExporter): exporter = self._exporter(grant) self._delegate.shutdown() self._delegate, self._grant = exporter, grant + # A renewed grant may carry the viewer URL the first one lacked. self._trace_url = grant.trace_url or self._trace_url if self._delegate.export(batch) != SpanExportResult.SUCCESS: return SpanExportResult.FAILURE @@ -250,32 +278,46 @@ class GrantSpanExporter(SpanExporter): ) return result - def show_trace_summary(self) -> None: - """Display the exported execution UUID and AMP's optional viewer link once.""" + def record_export(self) -> None: + """Record the run for `crewai eval` once its spans have reached Wharf, + and say where to look at it. + + Only a run whose every export succeeded is recorded — a partial export + would name a run the grader cannot read whole — and only such a run + gets a link, for the same reason. + """ with self._lock: - if ( - not self._exported - or self._export_failed - or self._summary_shown - or self._suppress_output - or should_suppress_tracing_messages() - or is_tui_mode() - ): + if not self._exported or self._export_failed or self._recorded: return - self._summary_shown = True - content = Text() - content.append("Traces exported\n", style="green bold") - content.append("Execution trace ID: ", style="white") - content.append(self._grant.execution_uuid, style="green") - if self._trace_url: - content.append("\n\nView traces:\n", style="white bold") - content.append( - self._trace_url, - style=Style(color="cyan", underline=True, link=self._trace_url), - ) - ConsoleFormatter(verbose=True).print_panel( - content, "🔗 Execution Traces", "green" + self._recorded = True + execution_uuid = self._grant.execution_uuid + api = getattr(self._client, "_api", None) + last_run.record_last_run( + execution_id=execution_uuid, + tier=getattr(self._client, "_tier", None), + started_at_ns=self._first_start_ns, + finished_at_ns=self._last_end_ns, + amp_base_url=getattr(api, "base_url", None), ) + logger.debug("Traces exported for execution %s", execution_uuid) + self._show_trace_link() + + def _show_trace_link(self) -> None: + """One line, once: where to see the run that was just exported. + + The execution id stays out of it — `crewai eval` reads that from the + record — but whoever wants to open the trace gets AMP's viewer link. + Silent when AMP granted no viewer URL, under the TUI, and wherever + tracing messages are suppressed. + """ + if not self._trace_url or should_suppress_tracing_messages() or is_tui_mode(): + return + line = Text("View traces: ", style="white") + line.append( + self._trace_url, + style=Style(color="cyan", underline=True, link=self._trace_url), + ) + Console().print(line) def shutdown(self) -> None: with self._lock: diff --git a/lib/crewai/src/crewai/telemetry/tracing/last_run.py b/lib/crewai/src/crewai/telemetry/tracing/last_run.py new file mode 100644 index 000000000..53e137dbd --- /dev/null +++ b/lib/crewai/src/crewai/telemetry/tracing/last_run.py @@ -0,0 +1,188 @@ +"""The last traced run of a project, recorded for `crewai eval`. + +Once a run's spans have reached Wharf, crewAI writes ``.crewai/last_run.json`` +in the project directory: the execution id, when the run started and ended, +and whether it was traced anonymously or under an account. Nothing is +printed — the id is internal — and ``crewai eval`` reads it back to evaluate +the run without the user pasting anything. + +Inside a deployment nothing is recorded, and no check here is what stops it: +the platform kicks off with ``tracing`` off, so crewAI never starts a trace +session of its own, never builds a ``GrantSpanExporter``, and never reaches +this module. +""" + +from __future__ import annotations + +from collections.abc import Iterator +import contextlib +from datetime import datetime, timezone +import json +import logging +import os +from pathlib import Path +import tempfile +from typing import Any + + +try: # POSIX only; without it the write below simply is not serialised + import fcntl +except ImportError: # pragma: no cover - Windows + fcntl = None # type: ignore[assignment] + +logger = logging.getLogger(__name__) + +LAST_RUN_DIR = ".crewai" +LAST_RUN_FILE = "last_run.json" +LOCK_FILE = "last_run.lock" + + +def project_dir() -> Path: + """The run's project: the working directory, as every project-local path in crewAI.""" + return Path.cwd() + + +def recording_enabled() -> bool: + """Off under the test suite, so kickoff-level tests leave no file behind. + + Nothing else is checked. A deployment is kept out by not getting here at + all, and the platform's integration token is NOT a deployment marker — it + is a credential `crewai create crew` writes into a project's own `.env` + for platform tools, so treating it as one would stop recording the runs of + every developer who uses them.""" + return os.environ.get("CREWAI_TESTING", "").lower() != "true" + + +def last_run_path(directory: Path | None = None) -> Path: + return (directory or project_dir()) / LAST_RUN_DIR / LAST_RUN_FILE + + +def _iso(nanoseconds: int | None) -> str | None: + if nanoseconds is None: + return None + return datetime.fromtimestamp(nanoseconds / 1e9, tz=timezone.utc).isoformat( + timespec="milliseconds" + ) + + +def record_last_run( + *, + execution_id: str, + tier: str | None, + started_at_ns: int | None, + finished_at_ns: int | None, + amp_base_url: str | None, +) -> Path | None: + """Write the record atomically; the path, or None when recording is off + or the write failed. Never raises — a run is never failed by this. + + "Last" means the run that FINISHED last: a record already there for a run + with a later `finished_at` is kept, so two crews finishing together in one + project leave the later-finished one whichever writer gets to the file + last. With no completion time to compare, the write stands.""" + if not recording_enabled(): + return None + record: dict[str, Any] = { + "execution_id": execution_id, + "tier": tier, + "started_at": _iso(started_at_ns), + "finished_at": _iso(finished_at_ns), + "recorded_at": datetime.now(timezone.utc).isoformat(timespec="milliseconds"), + "amp_base_url": amp_base_url, + } + path: Path | None = None + try: + path = last_run_path() + path.parent.mkdir(parents=True, exist_ok=True) + # A temporary file of its own per writer: two crews finishing together in one + # project must not write through the same name, or one record is lost. + descriptor, temporary = tempfile.mkstemp( + prefix=f".{path.name}.", suffix=".tmp", dir=path.parent + ) + try: + with os.fdopen(descriptor, "w", encoding="utf-8") as handle: + handle.write(json.dumps(record, indent=2) + "\n") + # Reading what is there, deciding, and replacing are one step: another + # process must not slip a newer record in between, or this write would + # compare against a record that is already gone and overwrite it. + with _exclusive(path.parent): + if _keep(record, read_last_run(path.parent.parent)): + os.replace(temporary, path) + else: + logger.debug( + "A run that finished later is already recorded in %s", path + ) + os.unlink(temporary) + except OSError: + with contextlib.suppress(OSError): + os.unlink(temporary) + raise + except OSError as error: # a vanished cwd fails last_run_path() too; the run is never failed by this + logger.debug( + "Could not record the last run in %s: %s", + path or ".crewai/last_run.json", + error, + ) + return None + return path + + +@contextlib.contextmanager +def _exclusive(directory: Path) -> Iterator[None]: + """One writer at a time in this project, across processes: a lock file beside + the record, held over the read, the comparison and the replace. + + The locking never fails a write. No `flock` at all (Windows), a lock file + that cannot be opened, or a filesystem that refuses the lock (some network + mounts) each leave the write to go ahead unserialised — the record is a + convenience pointer for `crewai eval`, and having it unserialised beats not + having it.""" + if fcntl is None: + yield + return + try: + handle = open(directory / LOCK_FILE, "a") + except OSError: + yield + return + locked = False + try: + try: + fcntl.flock(handle, fcntl.LOCK_EX) + locked = True + except OSError as error: + logger.debug("Could not lock %s: %s", directory / LOCK_FILE, error) + yield + finally: + if locked: + with contextlib.suppress(OSError): + fcntl.flock(handle, fcntl.LOCK_UN) + handle.close() + + +def _keep(record: dict[str, Any], existing: dict[str, Any] | None) -> bool: + """Does RECORD replace EXISTING? Only a record we can PROVE finished later + holds the file: a different run with a `finished_at` after ours. Everything + else — no record there, the same run recorded again, a missing completion + time on either side, a tie to the millisecond — leaves the write to stand, + so the last run recorded is the one a reader gets.""" + if not existing or existing.get("execution_id") == record["execution_id"]: + return True + ours, theirs = record["finished_at"], existing.get("finished_at") + if not isinstance(ours, str) or not isinstance(theirs, str): + return True # nothing to order by: the write stands + # Both ISO 8601 in UTC at the same precision, so text order is time order; + # a tie means two runs finished in the same millisecond and either is a fair "last run". + return ours >= theirs + + +def read_last_run(directory: Path | None = None) -> dict[str, Any] | None: + """The record, or None when there is none or it cannot be read.""" + path = last_run_path(directory) + 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 + return loaded diff --git a/lib/crewai/tests/telemetry/test_last_run.py b/lib/crewai/tests/telemetry/test_last_run.py new file mode 100644 index 000000000..946044fca --- /dev/null +++ b/lib/crewai/tests/telemetry/test_last_run.py @@ -0,0 +1,212 @@ +"""`.crewai/last_run.json`: the run `crewai eval` evaluates by default.""" + +from __future__ import annotations + +import json +import threading +import time + +import pytest + +from crewai.telemetry.tracing import last_run + + +def _leftovers(project): + """Anything in .crewai that is neither the record nor its lock file.""" + directory = project / ".crewai" + keep = {last_run.LAST_RUN_FILE, last_run.LOCK_FILE} + return [child.name for child in directory.iterdir() if child.name not in keep] if directory.exists() else [] + + +@pytest.fixture +def project(monkeypatch, tmp_path): + monkeypatch.setattr(last_run, "project_dir", lambda: tmp_path) + monkeypatch.setattr(last_run, "recording_enabled", lambda: True) + return tmp_path + + +def test_a_run_is_recorded_atomically_and_read_back(project): + path = last_run.record_last_run( + execution_id="6f31fe1a-20bd-4bfe-a011-25d6b9341f62", + tier="ephemeral", + started_at_ns=1_758_240_000_000_000_000, + finished_at_ns=1_758_240_009_500_000_000, + amp_base_url="https://app.crewai.com", + ) + assert path == project / ".crewai" / "last_run.json" + assert not _leftovers(project) # no temporary file left behind + written = json.loads(path.read_text(encoding="utf-8")) + assert written["execution_id"] == "6f31fe1a-20bd-4bfe-a011-25d6b9341f62" + assert written["tier"] == "ephemeral" + assert written["started_at"] == "2025-09-19T00:00:00.000+00:00" + assert written["finished_at"] == "2025-09-19T00:00:09.500+00:00" + assert written["amp_base_url"] == "https://app.crewai.com" + assert written["recorded_at"] + assert last_run.read_last_run(project) == written + + +def test_the_newest_run_replaces_the_previous_one(project): + last_run.record_last_run(execution_id="first", tier=None, started_at_ns=None, finished_at_ns=None, amp_base_url=None) + last_run.record_last_run(execution_id="second", tier="authenticated", started_at_ns=None, finished_at_ns=None, amp_base_url=None) + written = last_run.read_last_run(project) + assert written is not None and written["execution_id"] == "second" + assert written["started_at"] is None and written["finished_at"] is None + + +def test_nothing_is_recorded_under_the_test_suite(monkeypatch, tmp_path): + monkeypatch.setattr(last_run, "project_dir", lambda: tmp_path) + monkeypatch.setenv("CREWAI_TESTING", "true") + assert last_run.record_last_run(execution_id="x", tier=None, started_at_ns=None, finished_at_ns=None, amp_base_url=None) is None + assert not (tmp_path / ".crewai").exists() + + +def test_a_project_using_platform_tools_is_still_recorded(monkeypatch, tmp_path): + """`crewai create crew` writes CREWAI_PLATFORM_INTEGRATION_TOKEN into the project's + own .env, so it says "this developer uses platform tools", never "this is a + deployment". Reading it as a deployment marker would leave those users with no + run for `crewai eval` to find. + + Deliberately NOT the `project` fixture: that one stubs `recording_enabled`, which + is the very thing under test here.""" + monkeypatch.setattr(last_run, "project_dir", lambda: tmp_path) + monkeypatch.delenv("CREWAI_TESTING", raising=False) + monkeypatch.setenv("CREWAI_PLATFORM_INTEGRATION_TOKEN", "a token from the project's .env") + + assert last_run.recording_enabled() is True # the real guard, not the fixture's stub + path = last_run.record_last_run(execution_id="local", tier="authenticated", started_at_ns=None, finished_at_ns=None, amp_base_url=None) + + assert path is not None + assert last_run.read_last_run(tmp_path)["execution_id"] == "local" + + +def test_a_missing_or_broken_record_reads_as_none(project): + assert last_run.read_last_run(project) is None + (project / ".crewai").mkdir() + (project / ".crewai" / "last_run.json").write_text("not json", encoding="utf-8") + assert last_run.read_last_run(project) is None + (project / ".crewai" / "last_run.json").write_text(json.dumps({"tier": "ephemeral"}), encoding="utf-8") + assert last_run.read_last_run(project) is None # no execution id: no run + + +def test_a_write_failure_never_raises(project, monkeypatch): + (project / ".crewai").write_text("a file where the directory should be", encoding="utf-8") + assert last_run.record_last_run(execution_id="x", tier=None, started_at_ns=None, finished_at_ns=None, amp_base_url=None) is None + + +def test_each_writer_uses_a_temporary_file_of_its_own(project, monkeypatch): + """Two crews finishing together in one project: neither may replace the other's temporary file.""" + replaced: list[str] = [] + real_replace = last_run.os.replace + monkeypatch.setattr(last_run.os, "replace", lambda src, dst: replaced.append(str(src)) or real_replace(src, dst)) + for execution_id in ("first", "second"): + last_run.record_last_run(execution_id=execution_id, tier=None, started_at_ns=None, finished_at_ns=None, amp_base_url=None) + assert len(replaced) == 2 and replaced[0] != replaced[1] + names = [source.rsplit("/", 1)[-1] for source in replaced] + assert all(name.startswith(".last_run.json.") and name.endswith(".tmp") for name in names) + assert not _leftovers(project) + + +def test_a_failed_write_leaves_no_temporary_file(project, monkeypatch): + monkeypatch.setattr(last_run.os, "replace", lambda src, dst: (_ for _ in ()).throw(OSError("disk full"))) + assert last_run.record_last_run(execution_id="x", tier=None, started_at_ns=None, finished_at_ns=None, amp_base_url=None) is None + assert not _leftovers(project) and not (project / ".crewai" / last_run.LAST_RUN_FILE).exists() + + +def test_the_run_that_finished_last_stays_recorded_whichever_writer_comes_last(project): + """Two crews finish together; the OLDER run's writer gets to the file after the newer one did.""" + second = 1_000_000_000 + base = 1_758_240_000 * second + last_run.record_last_run(execution_id="newer", tier="ephemeral", started_at_ns=base, finished_at_ns=base + 30 * second, amp_base_url=None) + kept = last_run.record_last_run(execution_id="older", tier="ephemeral", started_at_ns=base, finished_at_ns=base + 10 * second, amp_base_url=None) + written = last_run.read_last_run(project) + assert kept == project / ".crewai" / "last_run.json" + assert written is not None and written["execution_id"] == "newer" + assert not _leftovers(project) # the loser's temporary file is gone + + # The same run recorded again (a refreshed grant) and a run that finished later both replace it. + last_run.record_last_run(execution_id="newer", tier="authenticated", started_at_ns=base, finished_at_ns=base + 30 * second, amp_base_url=None) + assert last_run.read_last_run(project)["tier"] == "authenticated" + last_run.record_last_run(execution_id="newest", tier="ephemeral", started_at_ns=base, finished_at_ns=base + 40 * second, amp_base_url=None) + assert last_run.read_last_run(project)["execution_id"] == "newest" + + # A tie to the millisecond: either run is a fair "last run", so the write stands. + last_run.record_last_run(execution_id="same-instant", tier="ephemeral", started_at_ns=base, finished_at_ns=base + 40 * second + 400_000, amp_base_url=None) + assert last_run.read_last_run(project)["execution_id"] == "same-instant" + + # An OLDER run still loses that tie-free comparison, however late its writer arrives. + last_run.record_last_run(execution_id="stale", tier="ephemeral", started_at_ns=base, finished_at_ns=base + 5 * second, amp_base_url=None) + assert last_run.read_last_run(project)["execution_id"] == "same-instant" + + # A record without a comparable time never blocks the run just finished. + (project / ".crewai" / "last_run.json").write_text(json.dumps({"execution_id": "legacy"}), encoding="utf-8") + last_run.record_last_run(execution_id="fresh", tier=None, started_at_ns=None, finished_at_ns=None, amp_base_url=None) + assert last_run.read_last_run(project)["execution_id"] == "fresh" + + +def test_writers_are_serialised_so_a_stale_read_cannot_overwrite_a_newer_run(project, monkeypatch): + """Two crews finish at once. Reading the file, deciding and replacing it are one + step, so no writer can decide against a record another writer has already replaced.""" + second = 1_000_000_000 + base = 1_758_240_000 * second + depth = 0 + overlaps = [] + guard = threading.Lock() + real_keep = last_run._keep + + def slow_keep(record, existing): + nonlocal depth + with guard: + depth += 1 + if depth > 1: + overlaps.append(depth) + time.sleep(0.05) # the window a stale read would live in + try: + return real_keep(record, existing) + finally: + with guard: + depth -= 1 + + monkeypatch.setattr(last_run, "_keep", slow_keep) + runs = [("oldest", 10), ("newest", 40), ("middle", 20)] + writers = [ + threading.Thread( + target=last_run.record_last_run, + kwargs={"execution_id": name, "tier": "ephemeral", "started_at_ns": base, + "finished_at_ns": base + offset * second, "amp_base_url": None}, + name=name, + ) + for name, offset in runs + ] + for writer in writers: + writer.start() + for writer in writers: + writer.join(10) + + assert overlaps == [] # never two writers inside the read-decide-replace region + written = last_run.read_last_run(project) + assert written is not None and written["execution_id"] == "newest" + assert not _leftovers(project) + + +def test_a_write_still_happens_where_the_platform_has_no_file_locking(project, monkeypatch): + """Windows has no flock: the record is a convenience pointer, never worth failing a run over.""" + monkeypatch.setattr(last_run, "fcntl", None) + + path = last_run.record_last_run(execution_id="unlocked", tier=None, started_at_ns=None, finished_at_ns=None, amp_base_url=None) + + assert path is not None + assert last_run.read_last_run(project)["execution_id"] == "unlocked" + assert not (project / ".crewai" / last_run.LOCK_FILE).exists() + + +@pytest.mark.parametrize("failure", [OSError(45, "Operation not supported"), PermissionError(13, "Permission denied")]) +def test_a_filesystem_that_refuses_the_lock_still_records_the_run(project, monkeypatch, failure): + """Some network mounts have no working flock. Unserialised beats not recorded.""" + monkeypatch.setattr( + last_run.fcntl, "flock", lambda handle, operation: (_ for _ in ()).throw(failure) + ) + + path = last_run.record_last_run(execution_id="unlockable", tier=None, started_at_ns=None, finished_at_ns=None, amp_base_url=None) + + assert path is not None + assert last_run.read_last_run(project)["execution_id"] == "unlockable" diff --git a/lib/crewai/tests/telemetry/test_session_trace_export.py b/lib/crewai/tests/telemetry/test_session_trace_export.py index 865824e19..db2ac76d6 100644 --- a/lib/crewai/tests/telemetry/test_session_trace_export.py +++ b/lib/crewai/tests/telemetry/test_session_trace_export.py @@ -131,6 +131,241 @@ def record(session, name="execute flow"): ).end() +@pytest.fixture +def recorded_runs(monkeypatch, tmp_path): + """Recording on, into tmp_path (the suite's CREWAI_TESTING turns it off).""" + from crewai.telemetry.tracing import last_run + + monkeypatch.setattr(last_run, "project_dir", lambda: tmp_path) + monkeypatch.setattr(last_run, "recording_enabled", lambda: True) + return tmp_path + + +def _last_run(directory): + from crewai.telemetry.tracing import last_run + + return last_run.read_last_run(directory) + + +@pytest.mark.parametrize( + ("authenticated", "approved", "status", "suppression", "recorded"), + [ + (True, True, 200, None, True), + (False, True, 200, None, True), + (False, False, 200, None, False), + (True, True, 401, None, False), + (False, True, 401, None, False), + (True, True, 200, "messages", True), + (False, True, 200, "messages", True), + (True, True, 200, "tui", True), + (False, True, 200, "tui", True), + ], +) +def test_the_viewer_link_is_shown_the_id_is_not_and_a_successful_export_is_recorded( + collector, + recorded_runs, + monkeypatch, + capsys, + authenticated, + approved, + status, + suppression, + recorded, +): + from crewai.events.listeners.tracing.utils import ( + set_suppress_tracing_messages, + set_tui_mode, + ) + from crewai.execution import begin_execution, end_execution, get_execution_uuid + from crewai.telemetry.tracing.context import get_trace_session + + url = collector.url + "/crewai_plus/otel_traces/run?access_code=secret&x=[value]" + collector.grant_override = {"trace_url": url} + collector.export_status = status + if authenticated: + monkeypatch.setenv("CREWAI_USER_PAT", "synthetic-pat") + seen = {} + + def run(): + if suppression == "messages": + set_suppress_tracing_messages(True) + elif suppression == "tui": + set_tui_mode(True) + with trace_consent(lambda: approved): + token = begin_execution(tracing=True) + try: + seen["uuid"] = get_execution_uuid() + session = get_trace_session() + record(session) + nested = begin_execution(tracing=True) + end_execution(nested) + finally: + end_execution(token) + + copy_context().run(run) + output = capsys.readouterr().out + # The id is internal — `crewai eval` reads it from the record — so it is never + # printed, nor is the old "Traces exported" panel around it. + assert seen["uuid"] not in output + assert "Execution trace ID:" not in output and "Traces exported" not in output + # The viewer link IS shown, once, for whoever wants to look at the run: only for + # a run that was exported whole, and not where tracing messages are suppressed. + shown = recorded and suppression is None + assert ("View traces:" in output) == shown + assert (url in output.replace("\n", "")) == shown + assert len(collector.batches) == int(authenticated or approved) + # A run whose spans reached Wharf is recorded for `crewai eval`, silently, + # in the TUI and under message suppression too; a failed export is not. + written = _last_run(recorded_runs) + if recorded: + assert written is not None + assert written["execution_id"] == seen["uuid"] + assert written["tier"] == ("authenticated" if authenticated else "ephemeral") + assert written["started_at"] and written["finished_at"] and written["recorded_at"] + assert written["amp_base_url"] == collector.url + else: + assert written is None + + +@pytest.mark.parametrize("first_status", [200, 401]) +def test_a_deferred_run_is_recorded_once_at_finalization_and_a_failed_export_never( + collector, recorded_runs, monkeypatch, capsys, first_status +): + from crewai.execution import begin_execution, end_execution + from crewai.telemetry.tracing.context import get_trace_session + + monkeypatch.setenv("CREWAI_USER_PAT", "synthetic-pat") + collector.export_status = first_status + token = begin_execution(tracing=True) + try: + session = get_trace_session() + record(session) + session.flush() + finally: + lifetime = end_execution(token, defer=True) + assert _last_run(recorded_runs) is None # not finished yet + + collector.export_status = 200 + token = begin_execution(tracing=True, trace_session=lifetime) + try: + record(get_trace_session()) + finally: + end_execution(token) + lifetime.finish() + assert capsys.readouterr().out == "" # this grant carries no viewer URL, so there is nothing to show + # A run one of whose exports failed is never recorded: the grader could not read it whole. + assert (_last_run(recorded_runs) is not None) == (first_status == 200) + + +@pytest.mark.parametrize("credential", [None, "pat"]) +def test_a_refreshed_grant_still_records_the_run_once( + collector, recorded_runs, monkeypatch, capsys, credential +): + from crewai.telemetry.tracing import last_run + + writes = [] + real_record = last_run.record_last_run + monkeypatch.setattr( + last_run, + "record_last_run", + lambda **fields: (writes.append(fields["execution_id"]), real_record(**fields))[1], + ) + collector.grant_override = {"trace_url": collector.url + "/crewai_plus/otel_traces/original"} + client = TraceGrantClient(credential) + grant = replace( + client.create(str(uuid4())), + expires_at=datetime.now(timezone.utc) + timedelta(seconds=1), + ) + collector.grant_override = {} + exporter = GrantSpanExporter(client, grant) + session = TraceSession(grant.execution_uuid, [exporter]) + try: + record(session) + finally: + session.shutdown() + exporter.record_export() + exporter.record_export() + assert writes == [grant.execution_uuid] # written once, however often the exporter is told the run ended + shown = capsys.readouterr().out + assert shown.count("View traces:") == 1 # and the link with it, once + assert grant.execution_uuid not in shown + written = _last_run(recorded_runs) + assert written is not None and written["execution_id"] == grant.execution_uuid + assert written["tier"] == ("authenticated" if credential else "ephemeral") + assert len(collector.grants) == 2 + assert all( + payload["include_trace_url"] is True for _, _, payload in collector.grants + ) + + +def test_the_recorded_window_is_the_first_start_and_last_end_across_batches( + collector, recorded_runs +): + """Batches reach the exporter in any order; the record spans them all.""" + from crewai.telemetry.tracing import last_run + + memory = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(memory)) + tracer = provider.get_tracer("window") + second = 1_000_000_000 + base = 1_700_000_000 * second + for name, start, end in (("late", 10, 12), ("early", 0, 5), ("middle", 3, 9)): + tracer.start_span(name, start_time=base + start * second).end(end_time=base + end * second) + provider.shutdown() + by_name = {span.name: span for span in memory.get_finished_spans()} + client = TraceGrantClient(None) + exporter = GrantSpanExporter(client, client.create(str(uuid4()))) + try: + assert exporter.export([by_name["late"]]) == SpanExportResult.SUCCESS + assert exporter.export([by_name["early"], by_name["middle"]]) == SpanExportResult.SUCCESS + finally: + exporter.shutdown() + exporter.record_export() + written = _last_run(recorded_runs) + assert written is not None + assert written["started_at"] == last_run._iso(base) + assert written["finished_at"] == last_run._iso(base + 12 * second) + + +def test_a_truncated_ephemeral_trace_is_uploaded_but_not_recorded( + collector, recorded_runs, monkeypatch +): + """The buffer dropped spans at its cap: the grader could not read the run whole.""" + monkeypatch.setenv("CREWAI_EPHEMERAL_TRACE_MAX_SPANS", "1") + buffer = EphemeralSpanBuffer() + monkeypatch.setattr(ephemeral, "EphemeralSpanBuffer", lambda: buffer) + with trace_consent(lambda: True), ephemeral_tracing(str(uuid4())) as session: + record(session, "first") + record(session, "second") + assert buffer._dropped == 1 + assert len(collector.batches) == 1 # what survived was still shared + assert _last_run(recorded_runs) is None + + +@pytest.mark.parametrize("authenticated", [True, False]) +def test_an_export_without_a_viewer_url_is_recorded_all_the_same( + collector, recorded_runs, monkeypatch, capsys, authenticated +): + from crewai.execution import begin_execution, end_execution, get_execution_uuid + from crewai.telemetry.tracing.context import get_trace_session + + if authenticated: + monkeypatch.setenv("CREWAI_USER_PAT", "synthetic-pat") + with trace_consent(lambda: True): + token = begin_execution(tracing=True) + try: + execution_uuid = get_execution_uuid() + record(get_trace_session()) + finally: + end_execution(token) + # AMP granted no viewer URL, so there is no link to show — and never the id. + assert capsys.readouterr().out == "" + written = _last_run(recorded_runs) + assert written is not None and written["execution_id"] == execution_uuid + assert len(collector.batches) == 1 + + def test_credential_precedence_and_missing_login(monkeypatch): monkeypatch.setenv("CREWAI_USER_PAT", "pat") monkeypatch.setenv("CREWAI_PLATFORM_INTEGRATION_TOKEN", "integration") @@ -214,151 +449,6 @@ def test_unusable_optional_viewer_url_does_not_break_grant(collector, value): assert TraceGrantClient("pat").create(str(uuid4())).trace_url is None -@pytest.mark.parametrize( - ("authenticated", "approved", "status", "suppression", "shows_link"), - [ - (True, True, 200, None, True), - (False, True, 200, None, True), - (False, False, 200, None, False), - (True, True, 401, None, False), - (False, True, 401, None, False), - (True, True, 200, "messages", False), - (False, True, 200, "messages", False), - (True, True, 200, "tui", False), - (False, True, 200, "tui", False), - ], -) -def test_viewer_link_is_printed_once_after_successful_execution_export( - collector, - monkeypatch, - capsys, - authenticated, - approved, - status, - suppression, - shows_link, -): - from crewai.events.listeners.tracing.utils import ( - set_suppress_tracing_messages, - set_tui_mode, - ) - from crewai.execution import begin_execution, end_execution - from crewai.telemetry.tracing.context import get_trace_session - - url = collector.url + "/crewai_plus/otel_traces/run?access_code=secret&x=[value]" - collector.grant_override = {"trace_url": url} - collector.export_status = status - if authenticated: - monkeypatch.setenv("CREWAI_USER_PAT", "synthetic-pat") - - def run(): - if suppression == "messages": - set_suppress_tracing_messages(True) - elif suppression == "tui": - set_tui_mode(True) - with trace_consent(lambda: approved): - token = begin_execution(tracing=True) - try: - session = get_trace_session() - record(session) - nested = begin_execution(tracing=True) - end_execution(nested) - assert "View traces:" not in capsys.readouterr().out - finally: - end_execution(token) - - copy_context().run(run) - output = capsys.readouterr().out - assert output.count(url) == int(shows_link) - assert output.count("View traces:") == int(shows_link) - assert output.count("Execution trace ID:") == int(shows_link) - assert len(collector.batches) == int(authenticated or approved) - - -@pytest.mark.parametrize("first_status", [200, 401]) -def test_deferred_trace_link_waits_for_finalization_and_remembers_export_failures( - collector, monkeypatch, capsys, first_status -): - from crewai.execution import begin_execution, end_execution - from crewai.telemetry.tracing.context import get_trace_session - - monkeypatch.setenv("CREWAI_USER_PAT", "synthetic-pat") - url = collector.url + "/crewai_plus/otel_traces/run" - collector.grant_override = {"trace_url": url} - collector.export_status = first_status - token = begin_execution(tracing=True) - try: - session = get_trace_session() - record(session) - session.flush() - finally: - lifetime = end_execution(token, defer=True) - assert url not in capsys.readouterr().out - - collector.export_status = 200 - token = begin_execution(tracing=True, trace_session=lifetime) - try: - record(get_trace_session()) - finally: - end_execution(token) - lifetime.finish() - assert capsys.readouterr().out.count(url) == int(first_status == 200) - - -@pytest.mark.parametrize("updated_url", [False, True]) -@pytest.mark.parametrize("credential", [None, "pat"]) -def test_refreshed_grant_preserves_or_updates_viewer_url( - collector, capsys, updated_url, credential -): - original = collector.url + "/crewai_plus/otel_traces/original" - renewed = collector.url + "/crewai_plus/otel_traces/renewed" - collector.grant_override = {"trace_url": original} - client = TraceGrantClient(credential) - grant = replace( - client.create(str(uuid4())), - expires_at=datetime.now(timezone.utc) + timedelta(seconds=1), - ) - collector.grant_override = {"trace_url": renewed} if updated_url else {} - exporter = GrantSpanExporter(client, grant) - session = TraceSession(grant.execution_uuid, [exporter]) - try: - record(session) - finally: - session.shutdown() - exporter.show_trace_summary() - exporter.show_trace_summary() - output = capsys.readouterr().out - assert output.count("View traces:") == 1 - assert (renewed if updated_url else original) in output - assert len(collector.grants) == 2 - assert all( - payload["include_trace_url"] is True for _, _, payload in collector.grants - ) - - -@pytest.mark.parametrize("authenticated", [True, False]) -def test_export_without_viewer_url_still_shows_execution_uuid( - collector, monkeypatch, capsys, authenticated -): - from crewai.execution import begin_execution, end_execution, get_execution_uuid - from crewai.telemetry.tracing.context import get_trace_session - - if authenticated: - monkeypatch.setenv("CREWAI_USER_PAT", "synthetic-pat") - with trace_consent(lambda: True): - token = begin_execution(tracing=True) - try: - execution_uuid = get_execution_uuid() - record(get_trace_session()) - assert "Execution trace ID:" not in capsys.readouterr().out - finally: - end_execution(token) - output = capsys.readouterr().out - assert output.count(f"Execution trace ID: {execution_uuid}") == 1 - assert "View traces:" not in output - assert len(collector.batches) == 1 - - @pytest.mark.parametrize( "override", [