Merge branch 'main' into viditostwal/oss-170-bedrock-acall-fallback

lib/cli/src/crewai_cli/templates/crew/.gitignore

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

lib/cli/src/crewai_cli/templates/flow/.gitignore

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

lib/cli/src/crewai_cli/templates/json_crew/.gitignore

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

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

lib/crewai-tools/src/crewai_tools/aws/s3/reader_tool.py

@@ -1,3 +1,4 @@
+from contextlib import suppress
 import os
 
 from crewai.tools import BaseTool
@@ -38,8 +39,13 @@ class S3ReaderTool(BaseTool):
             )
 
             response = s3.get_object(Bucket=bucket_name, Key=object_key)
-            result: str = response["Body"].read().decode("utf-8")
-            return result
+            body = response["Body"]
+            try:
+                result: str = body.read().decode("utf-8")
+                return result
+            finally:
+                with suppress(Exception):
+                    body.close()
 
         except ClientError as e:
             return f"Error reading file from S3: {e!s}"

lib/crewai-tools/src/crewai_tools/tools/selenium_scraping_tool/selenium_scraping_tool.py

@@ -142,9 +142,6 @@ class SeleniumScrapingTool(BaseTool):
             return "\n".join(content)
         except Exception as e:
             return f"Error scraping website: {e!s}"
-        finally:
-            if self.driver is not None:
-                self.driver.close()
 
     def _get_content(
         self, css_element: str | None, return_html: bool | None
@@ -207,5 +204,6 @@ class SeleniumScrapingTool(BaseTool):
             time.sleep(sleep_time)
 
     def close(self) -> None:
+        """End the browser session so the process does not linger."""
         if self.driver is not None:
-            self.driver.close()
+            self.driver.quit()

lib/crewai-tools/tests/tools/selenium_scraping_tool_test.py

@@ -69,7 +69,8 @@ def test_scrape_without_css_selector(_mocked_chrome_driver):
     assert "test content" in result
     mock_driver.get.assert_called_once_with("https://example.com")
     mock_driver.find_element.assert_called_with("tag name", "body")
-    mock_driver.close.assert_called_once()
+    mock_driver.close.assert_not_called()
+    mock_driver.quit.assert_not_called()
 
 
 @patch("selenium.webdriver.Chrome")
@@ -83,7 +84,8 @@ def test_scrape_with_css_selector(_mocked_chrome_driver):
     assert "test content in a specific div" in result
     mock_driver.get.assert_called_once_with("https://example.com")
     mock_driver.find_elements.assert_called_with("css selector", "div.test")
-    mock_driver.close.assert_called_once()
+    mock_driver.close.assert_not_called()
+    mock_driver.quit.assert_not_called()
 
 
 @patch("selenium.webdriver.Chrome")
@@ -97,7 +99,8 @@ def test_scrape_with_return_html_true(_mocked_chrome_driver):
     assert html_content in result
     mock_driver.get.assert_called_once_with("https://example.com")
     mock_driver.find_element.assert_called_with("tag name", "body")
-    mock_driver.close.assert_called_once()
+    mock_driver.close.assert_not_called()
+    mock_driver.quit.assert_not_called()
 
 
 @patch("selenium.webdriver.Chrome")
@@ -111,7 +114,8 @@ def test_scrape_with_return_html_false(_mocked_chrome_driver):
     assert "HTML content" in result
     mock_driver.get.assert_called_once_with("https://example.com")
     mock_driver.find_element.assert_called_with("tag name", "body")
-    mock_driver.close.assert_called_once()
+    mock_driver.close.assert_not_called()
+    mock_driver.quit.assert_not_called()
 
 
 @patch("selenium.webdriver.Chrome")
@@ -121,7 +125,8 @@ def test_scrape_with_driver_error(_mocked_chrome_driver):
     tool = initialize_tool_with(mock_driver)
     result = tool._run(website_url="https://example.com")
     assert result == "Error scraping website: WebDriver error occurred"
-    mock_driver.close.assert_called_once()
+    mock_driver.close.assert_not_called()
+    mock_driver.quit.assert_not_called()
 
 
 @patch("selenium.webdriver.Chrome")
@@ -129,3 +134,71 @@ def test_initialization_with_driver(_mocked_chrome_driver):
     mock_driver = MagicMock()
     tool = initialize_tool_with(mock_driver)
     assert tool.driver == mock_driver
+
+
+class FakeWindowDriver:
+    """Mimics a real chromedriver: closing the only window ends navigation."""
+
+    def __init__(self) -> None:
+        self.get_calls = 0
+        self.window_closed = False
+        self.quit_calls = 0
+
+    def get(self, url: str) -> None:
+        if self.window_closed:
+            raise Exception("no such window: target window already closed")
+        self.get_calls += 1
+
+    def find_element(self, by: str, selector: str):
+        if self.window_closed:
+            raise Exception("no such window: target window already closed")
+
+        class Element:
+            text = "body content"
+
+            def get_attribute(self, name: str) -> str:
+                return "<html><body>body content</body></html>"
+
+        return Element()
+
+    def find_elements(self, by: str, selector: str):
+        if self.window_closed:
+            raise Exception("no such window: target window already closed")
+        return [self.find_element(by, selector)]
+
+    def close(self) -> None:
+        self.window_closed = True
+
+    def quit(self) -> None:
+        self.quit_calls += 1
+
+
+@patch("selenium.webdriver.Chrome")
+def test_driver_stays_reusable_across_runs(_mocked_chrome_driver):
+    """The driver must survive a run so the tool can be called again.
+
+    Real chromedriver raises 'no such window' once the only window has been
+    closed, so closing the driver after every run made every call after the
+    first fail.
+    """
+    driver = FakeWindowDriver()
+    tool = SeleniumScrapingTool(driver=driver, wait_time=0)
+
+    first = tool._run(website_url="https://example.com")
+    second = tool._run(website_url="https://example.com", css_element="div.test")
+
+    assert "body content" in first
+    assert "body content" in second
+    assert driver.get_calls == 2
+    assert not driver.window_closed
+
+
+@patch("selenium.webdriver.Chrome")
+def test_close_ends_the_session(_mocked_chrome_driver):
+    """close() must end the session so the browser process does not linger."""
+    driver = FakeWindowDriver()
+    tool = SeleniumScrapingTool(driver=driver, wait_time=0)
+
+    tool.close()
+
+    assert driver.quit_calls == 1

lib/crewai-tools/tests/tools/test_s3_reader_tool.py

@@ -0,0 +1,107 @@
+"""Tests for the S3 reader tool."""
+
+import sys
+from types import ModuleType
+from unittest.mock import Mock, patch
+
+import pytest
+
+from crewai_tools.aws.s3.reader_tool import S3ReaderTool
+
+
+def _boto_modules(client: Mock) -> dict[str, ModuleType]:
+    """Build minimal boto modules for exercising the lazy imports."""
+    boto3 = ModuleType("boto3")
+    boto3.client = Mock(return_value=client)  # type: ignore[attr-defined]
+
+    botocore = ModuleType("botocore")
+    exceptions = ModuleType("botocore.exceptions")
+
+    class ClientError(Exception):
+        pass
+
+    exceptions.ClientError = ClientError  # type: ignore[attr-defined]
+    botocore.exceptions = exceptions  # type: ignore[attr-defined]
+    return {
+        "boto3": boto3,
+        "botocore": botocore,
+        "botocore.exceptions": exceptions,
+    }
+
+
+def test_s3_reader_closes_response_body() -> None:
+    """Release the streaming response after a successful read."""
+    body = Mock()
+    body.read.return_value = b"hello"
+    client = Mock()
+    client.get_object.return_value = {"Body": body}
+
+    with patch.dict(sys.modules, _boto_modules(client)):
+        result = S3ReaderTool()._run("s3://bucket/key.txt")
+
+    assert result == "hello"
+    body.close.assert_called_once_with()
+
+
+def test_s3_reader_closes_response_body_after_decode_error() -> None:
+    """Release the streaming response when UTF-8 decoding fails."""
+    body = Mock()
+    body.read.return_value = b"\xff"
+    client = Mock()
+    client.get_object.return_value = {"Body": body}
+
+    with (
+        patch.dict(sys.modules, _boto_modules(client)),
+        pytest.raises(UnicodeDecodeError),
+    ):
+        S3ReaderTool()._run("s3://bucket/key.txt")
+
+    body.close.assert_called_once_with()
+
+
+def test_s3_reader_closes_response_body_after_read_error() -> None:
+    """Release the streaming response when reading the body fails."""
+    body = Mock()
+    body.read.side_effect = OSError("connection reset")
+    client = Mock()
+    client.get_object.return_value = {"Body": body}
+
+    with (
+        patch.dict(sys.modules, _boto_modules(client)),
+        pytest.raises(OSError, match="connection reset"),
+    ):
+        S3ReaderTool()._run("s3://bucket/key.txt")
+
+    body.close.assert_called_once_with()
+
+
+def test_s3_reader_preserves_result_when_close_fails() -> None:
+    """Keep a successful read result when best-effort cleanup fails."""
+    body = Mock()
+    body.read.return_value = b"hello"
+    body.close.side_effect = OSError("close failed")
+    client = Mock()
+    client.get_object.return_value = {"Body": body}
+
+    with patch.dict(sys.modules, _boto_modules(client)):
+        result = S3ReaderTool()._run("s3://bucket/key.txt")
+
+    assert result == "hello"
+    body.close.assert_called_once_with()
+
+
+def test_s3_reader_preserves_read_error_when_close_fails() -> None:
+    """Keep the primary read error when best-effort cleanup also fails."""
+    body = Mock()
+    body.read.side_effect = OSError("connection reset")
+    body.close.side_effect = RuntimeError("close failed")
+    client = Mock()
+    client.get_object.return_value = {"Body": body}
+
+    with (
+        patch.dict(sys.modules, _boto_modules(client)),
+        pytest.raises(OSError, match="connection reset"),
+    ):
+        S3ReaderTool()._run("s3://bucket/key.txt")
+
+    body.close.assert_called_once_with()

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

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

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:

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

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"

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