From 6e1fa8cf2aaedab8748dd5c507c27e8830914bd1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Moura?= Date: Mon, 21 Sep 2026 14:28:25 -0300 Subject: [PATCH] feat(tracing): record the last traced run for crewai eval instead of printing it (#7648) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(tracing): record the last traced run for crewai eval instead of printing it After a traced run, crewAI printed a panel with the execution id and a viewer link. The id is internal, and the panel exposed it for no purpose a user has. Nothing is printed now. Instead, once the run's spans have reached Wharf, crewAI writes `.crewai/last_run.json` in the project: the execution id, when the run started and ended, whether it was traced anonymously or under an account, and the AMP base url. `crewai eval` reads it back, so the user evaluates their last run without pasting anything. GrantSpanExporter.record_export replaces show_trace_summary at the two finish points (an authenticated run's shutdown, an anonymous run's share). Only a run whose every export succeeded is recorded — a partial export would name a run the grader could not read whole. Recording is silent in the TUI and under message suppression too, off under the test suite, and inert inside a deployment, where the platform binds the execution before crewAI's own tracing starts. The record is written atomically and a write failure never fails the run. The crew, flow and json_crew scaffolds now ignore `.crewai/` like the declarative flow scaffold already did. Co-Authored-By: Claude Fable 5.1 * fix(tracing): never record a truncated ephemeral run; resolve the record path inside the guard; pin the window and the once - An ephemeral buffer that dropped spans at its cap still uploaded the rest and recorded the run. The grader could not read that run whole, so it is no longer recorded (the upload still happens). Test added. - last_run_path() ran before the try in record_last_run; a cwd that vanished mid-run would have raised out of a function documented never to raise. It is inside now, with the same debug log and None. - Tests: the "recorded once" check counted an unchanged mtime, which proves nothing on a coarse filesystem — it now counts calls to record_last_run; the first-start/last-end fold across export batches had no value test — batches now arrive out of order and the ISO window is asserted. - Docs: the conversational-flows comment said "one trace link" for output that no longer exists; tracing.mdx now names .crewai/last_run.json, what it holds, who reads it, and when it is not written. en, ar, ko, pt-BR. 85 passed (test_session_trace_export + test_last_run); ruff and mypy clean. Co-Authored-By: Claude Fable 5.1 * fix(tracing): a temporary file of its own per writer; the docs name deployments - record_last_run wrote every record through one `.tmp` name, so two crews finishing together in one project could replace each other's temporary file and one record was lost. Each write now goes through tempfile.mkstemp in the same directory, then os.replace; a failed write removes its own temporary file. Tests: two writers use two names and leave only last_run.json; a failed replace leaves nothing behind. - tracing.mdx (en, ar, ko, pt-BR): nothing is recorded inside a deployment, where the platform owns the trace. 87 passed (test_last_run + test_session_trace_export); ruff and mypy clean. Co-Authored-By: Claude Fable 5.1 * fix(tracing): the run that finished last stays recorded, whichever writer comes last Two crews finishing together in one project: the older run's writer could reach the file after the newer one and leave the older run as "the last". record_last_run now compares finished_at (recorded_at when absent) with the record already there and keeps the later-finished run; the same run recorded again (a refreshed grant) and a record without a comparable time never block the run just finished. No lock file: the file is a convenience pointer and the compare closes the ordering, leaving a microsecond window between read and replace that two runs finishing in the same instant could hit — either is then a fair "last run". Test: the older writer replaces after the newer one; the newer stays. 88 passed. Co-Authored-By: Claude Fable 5.1 * fix(tracing): a same-millisecond tie keeps the record already there; nothing is recorded inside a deployment - The times in the record are stored to the millisecond; two runs finishing in the same millisecond compared equal and the later writer won, which could leave the older run recorded. A tie now keeps what is there. No finer order is worth a field in the file. - recording_enabled() is false when the platform's integration token is present: a deployment normally never reaches this code (the host binds the trace before crewAI would start its own), but a container that did not would have written a file the platform never reads, against what the docs promise. Test added. 89 passed; ruff and mypy clean. Co-Authored-By: Claude Fable 5.1 * fix(tracing): only a run that provably finished later holds the record My previous commit compared `finished_at` and fell back to `recorded_at`, which broke two existing tests on CI: two records written in the same millisecond with no completion time compared equal, so the second write was dropped and the older run stayed. Locally the two writes happened to land in different milliseconds, so the suite passed — a timing-dependent bug, not a CI quirk. The rule is now narrow: a write stands unless the record already there is a DIFFERENT run with a later `finished_at`. No completion time on either side, the same run recorded again, or a tie to the millisecond all leave the write to stand, so "the last run recorded" stays what a reader gets. The case the rule exists for is unchanged: the older-finished writer arriving last does not replace the newer one. 485 passed in lib/crewai/tests/telemetry (test_last_run run five times for timing); ruff and mypy clean. Co-Authored-By: Claude Opus 5 (1M context) * fix(tracing): serialise the write across processes; drop the platform-token guard, which was not a deployment test - Reading the record, deciding and replacing it are now one step, held under a lock file beside the record, so a writer can no longer decide against a record another process has already replaced. Where the platform has no flock the write goes ahead unserialised: the record is a convenience pointer for `crewai eval`, never worth failing a run over. The regression drives three writers at once and fails without the lock. - The deployment guard I added yesterday read CREWAI_PLATFORM_INTEGRATION_TOKEN, which is not a deployment marker: `crewai create crew` writes that token into a project's own .env for platform tools, and crew_base loads it, so the guard would have stopped recording for every developer who uses them. It is also unnecessary — a deployment kicks off with tracing off, so crewAI never builds the exporter and never reaches this module. Guard removed, with a test that a project carrying that token is still recorded. - The module docstring claimed the platform "binds the execution before crewAI would start its own tracing". It does not; it sets tracing off at kickoff. Docstring and the four tracing docs now say the real reason. 487 passed in lib/crewai/tests/telemetry; ruff and mypy clean. Co-Authored-By: Claude Opus 5 (1M context) * fix(tracing): a refused lock still records the run; the platform-token test exercises the real guard Two Cursor findings on the locking commit. - If flock itself failed — some network mounts refuse it — the OSError escaped _exclusive, the outer handler deleted the temporary file and record_last_run returned None, so the run was not recorded at all. A refused lock is now logged and the write goes ahead unserialised, matching what already happened when the lock file could not be opened or flock does not exist. Unserialised beats not recorded. - test_a_project_using_platform_tools_is_still_recorded used the `project` fixture, which stubs recording_enabled — the very thing under test — so the regression it guards could have come back green. It now stubs only project_dir and asserts the real guard. 13 in test_last_run, 489 across lib/crewai/tests/telemetry. Co-Authored-By: Claude Opus 5 (1M context) --------- Co-authored-by: Claude Fable 5.1 --- .../src/crewai_cli/templates/crew/.gitignore | 1 + .../src/crewai_cli/templates/flow/.gitignore | 1 + .../crewai_cli/templates/json_crew/.gitignore | 1 + lib/cli/tests/test_scaffold_gitignore.py | 17 + lib/crewai/src/crewai/execution.py | 2 +- .../src/crewai/telemetry/tracing/ephemeral.py | 5 +- .../src/crewai/telemetry/tracing/grants.py | 98 +++-- .../src/crewai/telemetry/tracing/last_run.py | 188 +++++++++ lib/crewai/tests/telemetry/test_last_run.py | 212 ++++++++++ .../telemetry/test_session_trace_export.py | 380 +++++++++++------- 10 files changed, 730 insertions(+), 175 deletions(-) create mode 100644 lib/cli/tests/test_scaffold_gitignore.py create mode 100644 lib/crewai/src/crewai/telemetry/tracing/last_run.py create mode 100644 lib/crewai/tests/telemetry/test_last_run.py 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", [