andre/crewAI
1
0
Fork 0
mirror of https://github.com/crewAIInc/crewAI.git synced 2026-09-22 02:46:39 +00:00
Code Issues Actions 5 Packages Projects Releases Wiki Activity

feat(tracing): record the last traced run for crewai eval instead of printing it (#7648)

Browse Source 
* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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) <noreply@anthropic.com>

* 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) <noreply@anthropic.com>

* 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) <noreply@anthropic.com>

---------

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
João Moura
2026-09-21 14:28:25 -03:00
committed by GitHub
parent 784aadde1d
commit 6e1fa8cf2a
10 changed files with 730 additions and 175 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
lib/cli/src/crewai_cli/templates/crew/.gitignore vendored
View File

@@ -1,3 +1,4 @@
.env
__pycache__/
.DS_Store
.crewai/

1
lib/cli/src/crewai_cli/templates/flow/.gitignore vendored
View File

@@ -2,3 +2,4 @@
__pycache__/
lib/
.DS_Store
.crewai/

1
lib/cli/src/crewai_cli/templates/json_crew/.gitignore vendored
View File

@@ -2,3 +2,4 @@
__pycache__/
.DS_Store
report.md
.crewai/

17
lib/cli/tests/test_scaffold_gitignore.py Normal file
View File

@@ -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

2
lib/crewai/src/crewai/execution.py
View File

@@ -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))

5
lib/crewai/src/crewai/telemetry/tracing/ephemeral.py
View File

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

98
lib/crewai/src/crewai/telemetry/tracing/grants.py
View File

@@ -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:

188
lib/crewai/src/crewai/telemetry/tracing/last_run.py Normal file
View File

@@ -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

212
lib/crewai/tests/telemetry/test_last_run.py Normal file
View File

@@ -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"

380
lib/crewai/tests/telemetry/test_session_trace_export.py
View File

@@ -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",
[