mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-08-10 08:21:54 +00:00
e1b0ca1124
test_all_common_attributes_land_on_exported_spans failed in CI with an IndexError on an empty span list, and only in one shard: the suite runs with OTEL_SDK_DISABLED set, so TracerProvider hands out no-op tracers and an export-based assertion sees zero spans rather than a wrong attribute. It passed only when it happened to run after a test whose fixture flips the variable, which random ordering decides. Adds an otel_enabled fixture that sets the variable for the four tests asserting on exported spans. Three of them predate this branch and had the same latent dependency - they are fixed here because the new test made the ordering hit reachable, and leaving them would keep the required check red. Verified by running every test in the file individually, all of which previously exposed the dependency, and the telemetry suite three times under random ordering. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
779 lines
26 KiB
Python
779 lines
26 KiB
Python
"""Tests for AI coding assistant detection in telemetry."""
|
|
|
|
import os
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from crewai.telemetry.utils import (
|
|
KNOWN_CODING_AGENTS,
|
|
KNOWN_RUNTIME_CONTEXTS,
|
|
detect_coding_agent,
|
|
detect_runtime_context,
|
|
)
|
|
from crewai.utilities.constants import (
|
|
ANTIGRAVITY_ENV_VARS,
|
|
AUGMENT_ENV_VARS,
|
|
CC_ENV_VAR,
|
|
CC_ENV_VARS,
|
|
CLINE_ENV_VARS,
|
|
CODEX_ENV_VARS,
|
|
CODING_AGENT_ENV_MARKERS,
|
|
CURSOR_ENV_VARS,
|
|
GEMINI_CLI_ENV_VARS,
|
|
GENERIC_AGENT_ENV_VARS,
|
|
JUNIE_ENV_VARS,
|
|
OPENCODE_ENV_VARS,
|
|
RUNTIME_CONTEXT_ENV_MARKERS,
|
|
)
|
|
|
|
|
|
# Derived from the shared tables rather than restated, so adding an assistant
|
|
# or runtime there cannot leave these tests silently checking a stale set.
|
|
RUNTIME_MARKERS = tuple(
|
|
var for _, env_vars in RUNTIME_CONTEXT_ENV_MARKERS for var in env_vars
|
|
)
|
|
|
|
ALL_MARKERS = (
|
|
tuple(var for _, env_vars in CODING_AGENT_ENV_MARKERS for var in env_vars)
|
|
+ RUNTIME_MARKERS
|
|
+ GENERIC_AGENT_ENV_VARS
|
|
+ ("TERM_PROGRAM", "TERMINAL_EMULATOR")
|
|
)
|
|
|
|
EVERY_RUNTIME_CASE = [
|
|
(var, context)
|
|
for context, env_vars in RUNTIME_CONTEXT_ENV_MARKERS
|
|
for var in env_vars
|
|
]
|
|
|
|
EVERY_MARKER_CASE = [
|
|
(var, agent) for agent, env_vars in CODING_AGENT_ENV_MARKERS for var in env_vars
|
|
]
|
|
|
|
|
|
@pytest.fixture
|
|
def clean_env(monkeypatch):
|
|
"""Remove every marker so each test starts from a known state."""
|
|
for var in ALL_MARKERS:
|
|
monkeypatch.delenv(var, raising=False)
|
|
return monkeypatch
|
|
|
|
|
|
@pytest.fixture
|
|
def otel_enabled(monkeypatch):
|
|
"""Let the SDK build real spans for tests that assert on exported ones.
|
|
|
|
The suite runs with OTEL_SDK_DISABLED set, which makes TracerProvider hand
|
|
out no-op tracers. An export-based assertion then sees zero spans rather
|
|
than a failed attribute, so it fails only when it happens to run before a
|
|
test whose fixture flips the variable - which random ordering decides.
|
|
"""
|
|
monkeypatch.setenv("OTEL_SDK_DISABLED", "false")
|
|
|
|
|
|
@pytest.fixture
|
|
def isolated_telemetry(monkeypatch):
|
|
"""Build a fresh Telemetry without touching the process-wide singleton.
|
|
|
|
Telemetry is a singleton whose __init__ registers atexit and signal
|
|
handlers. Re-initializing the shared instance would leak state into later
|
|
tests and stack duplicate handlers, so replace _instance for the duration
|
|
of the test and suppress lifecycle registration.
|
|
"""
|
|
from crewai.telemetry.telemetry import Telemetry
|
|
|
|
monkeypatch.setattr(Telemetry, "_instance", None)
|
|
monkeypatch.setattr(Telemetry, "_register_shutdown_handlers", lambda self: None)
|
|
|
|
def build():
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"CREWAI_DISABLE_TELEMETRY": "false",
|
|
"CREWAI_DISABLE_TRACKING": "false",
|
|
"OTEL_SDK_DISABLED": "false",
|
|
},
|
|
):
|
|
return Telemetry()
|
|
|
|
yield build
|
|
|
|
Telemetry._instance = None
|
|
|
|
|
|
@pytest.mark.parametrize(("env_var", "expected"), EVERY_MARKER_CASE)
|
|
def test_detects_every_marker_in_the_shared_table(clean_env, env_var, expected):
|
|
"""Every marker must map to its assistant, including Codex/Cursor extras."""
|
|
clean_env.setenv(env_var, "1")
|
|
assert detect_coding_agent() == expected
|
|
|
|
|
|
def test_shares_the_canonical_marker_sets():
|
|
"""Detection must not maintain a second, narrower set of markers.
|
|
|
|
The env-context events and telemetry previously disagreed: a session
|
|
exposing only CODEX_THREAD_ID was Codex to get_env_context() but unknown
|
|
here. Both now read the same table.
|
|
"""
|
|
by_agent = dict(CODING_AGENT_ENV_MARKERS)
|
|
|
|
assert CC_ENV_VAR in by_agent["claude_code"]
|
|
assert by_agent["codex"] is CODEX_ENV_VARS
|
|
assert by_agent["cursor"] is CURSOR_ENV_VARS
|
|
|
|
|
|
def test_codex_takes_precedence_over_cursor(clean_env):
|
|
"""Codex running inside Cursor must report codex, matching get_env_context().
|
|
|
|
Cursor sets CURSOR_* in every integrated terminal, so checking Cursor first
|
|
would mask any assistant spawned inside it.
|
|
"""
|
|
clean_env.setenv("CURSOR_TRACE_ID", "t-1")
|
|
clean_env.setenv("CODEX_THREAD_ID", "th-1")
|
|
|
|
assert detect_coding_agent() == "codex"
|
|
|
|
|
|
def test_claude_code_takes_precedence_over_cursor(clean_env):
|
|
clean_env.setenv("CURSOR_TRACE_ID", "t-1")
|
|
clean_env.setenv("CLAUDECODE", "1")
|
|
|
|
assert detect_coding_agent() == "claude_code"
|
|
|
|
|
|
def test_precedence_matches_get_env_context(clean_env):
|
|
"""The two signals must agree on which assistant is present."""
|
|
from crewai.events.types.env_events import (
|
|
CCEnvEvent,
|
|
CodexEnvEvent,
|
|
CursorEnvEvent,
|
|
)
|
|
from crewai.utilities import env as env_module
|
|
|
|
event_to_agent = {
|
|
CCEnvEvent: "claude_code",
|
|
CodexEnvEvent: "codex",
|
|
CursorEnvEvent: "cursor",
|
|
}
|
|
|
|
for markers in (
|
|
{"CLAUDECODE": "1"},
|
|
{"CODEX_THREAD_ID": "1"},
|
|
{"CURSOR_TRACE_ID": "1"},
|
|
{"CURSOR_TRACE_ID": "1", "CODEX_CI": "1"},
|
|
{"CURSOR_SANDBOX": "1", "CLAUDECODE": "1"},
|
|
):
|
|
for var in ALL_MARKERS:
|
|
clean_env.delenv(var, raising=False)
|
|
for var, value in markers.items():
|
|
clean_env.setenv(var, value)
|
|
|
|
emitted: list[type] = []
|
|
clean_env.setattr(
|
|
env_module.crewai_event_bus,
|
|
"emit",
|
|
lambda _source, event, sink=emitted: sink.append(type(event)),
|
|
)
|
|
env_module._env_context_emitted.set(False)
|
|
env_module.get_env_context()
|
|
|
|
expected = event_to_agent[emitted[0]]
|
|
assert detect_coding_agent() == expected, markers
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"config_var",
|
|
[
|
|
"AIDER_MODEL",
|
|
"COPILOT_GITHUB_TOKEN",
|
|
"COPILOT_MODEL",
|
|
"GOOSE_PROVIDER",
|
|
],
|
|
)
|
|
def test_config_style_variables_are_not_used_as_markers(config_var):
|
|
"""Persistent user config must never be treated as a session marker.
|
|
|
|
crewai loads dotenv files on normal runs, so a committed AIDER_MODEL or
|
|
GOOSE_PROVIDER would mislabel ordinary human executions. Published
|
|
detection matrices list several of these; they are deliberately excluded
|
|
here rather than copied wholesale.
|
|
"""
|
|
all_vars = {var for _, env_vars in CODING_AGENT_ENV_MARKERS for var in env_vars}
|
|
|
|
assert config_var not in all_vars
|
|
|
|
|
|
def test_hosted_environments_are_not_reported_as_assistants():
|
|
"""REPL_ID marks a hosted environment, not an assistant driving the run."""
|
|
all_vars = {var for _, env_vars in CODING_AGENT_ENV_MARKERS for var in env_vars}
|
|
|
|
assert "REPL_ID" not in all_vars
|
|
|
|
|
|
def test_every_marker_comes_from_a_verified_set():
|
|
"""Guard against reintroducing guessed variable names.
|
|
|
|
A wrong name never matches, so the assistant is silently counted as
|
|
"unknown" while the table implies it is covered - worse than omitting it.
|
|
Adding an assistant means extending the canonical sets, which keeps both
|
|
detection paths in sync.
|
|
"""
|
|
verified = {
|
|
*ANTIGRAVITY_ENV_VARS,
|
|
*AUGMENT_ENV_VARS,
|
|
*CC_ENV_VARS,
|
|
*CLINE_ENV_VARS,
|
|
*CODEX_ENV_VARS,
|
|
*CURSOR_ENV_VARS,
|
|
*GEMINI_CLI_ENV_VARS,
|
|
*JUNIE_ENV_VARS,
|
|
*OPENCODE_ENV_VARS,
|
|
}
|
|
declared = {var for _, env_vars in CODING_AGENT_ENV_MARKERS for var in env_vars}
|
|
|
|
assert declared == verified, (
|
|
"markers must come from the canonical per-assistant sets; "
|
|
f"unverified names present: {sorted(declared - verified)}"
|
|
)
|
|
|
|
|
|
def test_generic_marker_is_the_last_resort(clean_env):
|
|
"""AI_AGENT says an assistant is present without naming which one.
|
|
|
|
A named marker must win, so the generic entry cannot mask a specific one.
|
|
"""
|
|
clean_env.setenv("AI_AGENT", "1")
|
|
assert detect_coding_agent() == "other"
|
|
|
|
clean_env.setenv("CLINE_ACTIVE", "true")
|
|
assert detect_coding_agent() == "cline"
|
|
|
|
|
|
def test_generic_marker_value_is_never_reported(clean_env):
|
|
"""Its value is an arbitrary vendor string, so it is never read."""
|
|
clean_env.setenv("AI_AGENT", "some-unreleased-tool/2.0")
|
|
|
|
assert detect_coding_agent() == "other"
|
|
|
|
|
|
def test_terminal_bound_assistants_outrank_cursor(clean_env):
|
|
"""CURSOR_* is set for every integrated terminal.
|
|
|
|
Checking Cursor first would report cursor for anything spawned inside it,
|
|
the same trap Codex already had to be ordered around.
|
|
"""
|
|
clean_env.setenv("CURSOR_TRACE_ID", "t-1")
|
|
|
|
for marker, expected in (
|
|
("CLINE_ACTIVE", "cline"),
|
|
("GEMINI_CLI", "gemini_cli"),
|
|
("AUGMENT_AGENT", "augment"),
|
|
("OPENCODE_CLIENT", "opencode"),
|
|
):
|
|
clean_env.setenv(marker, "1")
|
|
assert detect_coding_agent() == expected, marker
|
|
clean_env.delenv(marker)
|
|
|
|
|
|
def test_concurrent_attach_registers_the_processor_once(isolated_telemetry, clean_env):
|
|
"""Check-then-act on the provider set must be locked.
|
|
|
|
Crews and flows created from different threads can both reach set_tracer()
|
|
before trace_set flips, and without a lock each would attach its own
|
|
processor to the same provider for the life of the process.
|
|
"""
|
|
import threading
|
|
import time
|
|
|
|
telemetry = isolated_telemetry()
|
|
threads_count = 8
|
|
|
|
class SlowProvider:
|
|
"""Widens the check-then-act window so the race is deterministic.
|
|
|
|
Sleeping inside add_span_processor guarantees every unlocked thread gets
|
|
past the membership check before any of them records the provider.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self.processors: list[object] = []
|
|
|
|
def add_span_processor(self, processor: object) -> None:
|
|
time.sleep(0.05)
|
|
self.processors.append(processor)
|
|
|
|
provider = SlowProvider()
|
|
start = threading.Barrier(threads_count)
|
|
|
|
def attach() -> None:
|
|
start.wait()
|
|
telemetry._attach_common_attributes(provider)
|
|
|
|
threads = [threading.Thread(target=attach) for _ in range(threads_count)]
|
|
for thread in threads:
|
|
thread.start()
|
|
for thread in threads:
|
|
thread.join()
|
|
|
|
assert len(provider.processors) == 1
|
|
|
|
|
|
def test_editor_terminal_requires_exact_value(clean_env):
|
|
clean_env.setenv("TERM_PROGRAM", "vscode")
|
|
assert detect_runtime_context() == "vscode_terminal"
|
|
|
|
clean_env.setenv("TERM_PROGRAM", "iTerm.app")
|
|
assert detect_runtime_context() != "vscode_terminal"
|
|
|
|
|
|
def test_editor_terminal_is_not_reported_as_an_assistant(clean_env):
|
|
"""An editor's terminal says where a process runs, not who drove it."""
|
|
clean_env.setenv("TERM_PROGRAM", "vscode")
|
|
assert detect_coding_agent() == "unknown"
|
|
|
|
|
|
def test_empty_marker_value_is_ignored(clean_env):
|
|
clean_env.setenv("CLAUDECODE", "")
|
|
assert detect_coding_agent() != "claude_code"
|
|
|
|
|
|
@pytest.mark.parametrize(("env_var", "expected"), EVERY_RUNTIME_CASE)
|
|
def test_detects_every_runtime_marker(clean_env, env_var, expected):
|
|
"""Every runtime marker must map to its context."""
|
|
clean_env.setenv(env_var, "1")
|
|
assert detect_runtime_context() == expected
|
|
|
|
|
|
def test_runtime_precedence_prefers_the_most_specific(clean_env):
|
|
"""CI and hosted IDEs usually run in containers; the specific one wins."""
|
|
clean_env.setenv("KUBERNETES_SERVICE_HOST", "10.0.0.1")
|
|
assert detect_runtime_context() == "container"
|
|
|
|
clean_env.setenv("GITHUB_ACTIONS", "true")
|
|
assert detect_runtime_context() == "ci"
|
|
|
|
|
|
def test_an_automated_run_still_reports_an_unknown_assistant(clean_env):
|
|
"""The split must keep the two fields independent.
|
|
|
|
A CI run has no assistant to find, which is different from failing to
|
|
recognize one - the reason they no longer share a field.
|
|
"""
|
|
clean_env.setenv("CI", "true")
|
|
assert detect_runtime_context() == "ci"
|
|
assert detect_coding_agent() == "unknown"
|
|
|
|
|
|
def test_assistant_and_runtime_are_reported_together(clean_env):
|
|
"""An assistant inside CI must not mask either signal."""
|
|
clean_env.setenv("CI", "true")
|
|
clean_env.setenv("CLAUDECODE", "1")
|
|
assert detect_coding_agent() == "claude_code"
|
|
assert detect_runtime_context() == "ci"
|
|
|
|
|
|
def test_falls_back_to_non_interactive_without_tty(clean_env, monkeypatch):
|
|
monkeypatch.setattr("os.path.exists", lambda path: False)
|
|
monkeypatch.setattr("sys.stdout", type("S", (), {"isatty": lambda self: False})())
|
|
assert detect_runtime_context() == "non_interactive"
|
|
|
|
|
|
def test_falls_back_to_interactive_with_tty(clean_env, monkeypatch):
|
|
monkeypatch.setattr("os.path.exists", lambda path: False)
|
|
monkeypatch.setattr("sys.stdout", type("S", (), {"isatty": lambda self: True})())
|
|
assert detect_runtime_context() == "interactive"
|
|
|
|
|
|
def test_dockerenv_marks_a_container(clean_env, monkeypatch):
|
|
"""The container check is the last resort before the TTY fallback."""
|
|
monkeypatch.setattr("os.path.exists", lambda path: path == "/.dockerenv")
|
|
assert detect_runtime_context() == "container"
|
|
|
|
|
|
def test_unmatched_assistant_is_unknown(clean_env, monkeypatch):
|
|
"""No marker means a gap in the table, reported as unknown."""
|
|
monkeypatch.setattr("sys.stdout", type("S", (), {"isatty": lambda self: True})())
|
|
assert detect_coding_agent() == "unknown"
|
|
|
|
|
|
def test_never_returns_env_var_value(clean_env):
|
|
"""The detected name must never leak the environment variable's contents."""
|
|
secret = "sk-super-secret-token"
|
|
clean_env.setenv("CURSOR_TRACE_ID", secret)
|
|
assert secret not in detect_coding_agent()
|
|
|
|
|
|
def test_handles_broken_stdout(clean_env, monkeypatch):
|
|
class BrokenStdout:
|
|
def isatty(self):
|
|
raise ValueError("detached")
|
|
|
|
monkeypatch.setattr("os.path.exists", lambda path: False)
|
|
monkeypatch.setattr("sys.stdout", BrokenStdout())
|
|
assert detect_runtime_context() == "unknown"
|
|
|
|
|
|
def test_result_is_always_a_known_literal(clean_env):
|
|
"""PII guarantee: the return value can only ever be a known literal.
|
|
|
|
Every marker is set to a value that would be catastrophic to emit, and the
|
|
result must still come from the fixed vocabulary.
|
|
"""
|
|
sensitive = "/Users/jane.doe/secrets/api-key-sk-live-1234"
|
|
|
|
for var in ALL_MARKERS:
|
|
clean_env.setenv(var, sensitive)
|
|
agent = detect_coding_agent()
|
|
context = detect_runtime_context()
|
|
assert agent in KNOWN_CODING_AGENTS
|
|
assert context in KNOWN_RUNTIME_CONTEXTS
|
|
assert sensitive not in agent
|
|
assert sensitive not in context
|
|
clean_env.delenv(var, raising=False)
|
|
|
|
|
|
def test_known_agents_contains_no_pii_shaped_values():
|
|
"""Every possible emitted value is a short, opaque identifier."""
|
|
for name in KNOWN_CODING_AGENTS | KNOWN_RUNTIME_CONTEXTS:
|
|
assert name.replace("_", "").isalnum(), name
|
|
assert len(name) <= 32, name
|
|
|
|
|
|
def test_coding_agent_lands_on_every_exported_span(clean_env, otel_enabled):
|
|
"""End-to-end: the attribute must appear as a *span attribute* on any span.
|
|
|
|
It cannot be a Resource attribute - the ingestion pipeline preserves only
|
|
serviceName from the resource, so anything else set there is dropped before
|
|
it reaches storage. This test exports through a real TracerProvider and
|
|
asserts the attribute survives on arbitrary spans.
|
|
"""
|
|
from opentelemetry.sdk.trace import TracerProvider
|
|
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
|
|
from opentelemetry.sdk.trace.export.in_memory_span_exporter import (
|
|
InMemorySpanExporter,
|
|
)
|
|
|
|
from crewai.telemetry.telemetry import CommonAttributesSpanProcessor
|
|
|
|
exporter = InMemorySpanExporter()
|
|
provider = TracerProvider()
|
|
provider.add_span_processor(
|
|
CommonAttributesSpanProcessor({"coding_agent": "claude_code"})
|
|
)
|
|
provider.add_span_processor(SimpleSpanProcessor(exporter))
|
|
|
|
tracer = provider.get_tracer("crewai.telemetry")
|
|
for name in ("Crew Created", "Task Execution", "Tool Usage", "Feature Usage"):
|
|
span = tracer.start_span(name)
|
|
span.end()
|
|
|
|
exported = exporter.get_finished_spans()
|
|
assert len(exported) == 4
|
|
for span in exported:
|
|
assert span.attributes["coding_agent"] == "claude_code", span.name
|
|
|
|
# It must be a span attribute, not a resource attribute, or ingestion drops it.
|
|
assert "coding_agent" not in exported[0].resource.attributes
|
|
|
|
|
|
def test_common_attributes_processor_never_breaks_span_creation(clean_env, otel_enabled):
|
|
"""A failure applying attributes must not propagate into user execution."""
|
|
from crewai.telemetry.telemetry import CommonAttributesSpanProcessor
|
|
|
|
class ExplodingSpan:
|
|
def set_attributes(self, _):
|
|
raise RuntimeError("boom")
|
|
|
|
CommonAttributesSpanProcessor({"coding_agent": "cursor"}).on_start(
|
|
ExplodingSpan() # type: ignore[arg-type]
|
|
)
|
|
|
|
|
|
def test_coding_agent_span_emits_once(isolated_telemetry, clean_env, monkeypatch):
|
|
clean_env.setenv("CLAUDECODE", "1")
|
|
|
|
telemetry = isolated_telemetry()
|
|
|
|
emitted: list[str] = []
|
|
monkeypatch.setattr(telemetry, "feature_usage_span", emitted.append)
|
|
|
|
telemetry.coding_agent_span()
|
|
telemetry.coding_agent_span()
|
|
telemetry.coding_agent_span()
|
|
|
|
assert emitted == ["coding_agent:claude_code"]
|
|
|
|
|
|
def test_attribute_survives_an_externally_installed_provider(
|
|
isolated_telemetry, clean_env, otel_enabled
|
|
):
|
|
"""Spans must keep coding_agent when the app installs its own provider.
|
|
|
|
set_tracer() leaves an existing non-proxy provider in place, and telemetry
|
|
methods resolve their tracer through the global provider - so attaching the
|
|
processor only to our own provider would drop the attribute entirely in any
|
|
already-instrumented application.
|
|
"""
|
|
from opentelemetry import trace as ot
|
|
from opentelemetry.sdk.trace import TracerProvider
|
|
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
|
|
from opentelemetry.sdk.trace.export.in_memory_span_exporter import (
|
|
InMemorySpanExporter,
|
|
)
|
|
|
|
clean_env.setenv("CLAUDECODE", "1")
|
|
|
|
exporter = InMemorySpanExporter()
|
|
app_provider = TracerProvider()
|
|
app_provider.add_span_processor(SimpleSpanProcessor(exporter))
|
|
|
|
with patch.object(ot, "get_tracer_provider", return_value=app_provider):
|
|
telemetry = isolated_telemetry()
|
|
telemetry.set_tracer()
|
|
|
|
span = app_provider.get_tracer("crewai.telemetry").start_span("Crew Created")
|
|
span.end()
|
|
|
|
exported = exporter.get_finished_spans()
|
|
assert len(exported) == 1
|
|
assert exported[0].attributes["coding_agent"] == "claude_code"
|
|
|
|
|
|
def test_attaching_common_attributes_is_idempotent(isolated_telemetry, clean_env):
|
|
"""Repeated set_tracer() calls must not stack duplicate processors."""
|
|
from opentelemetry.sdk.trace import TracerProvider
|
|
|
|
provider = TracerProvider()
|
|
telemetry = isolated_telemetry()
|
|
|
|
before = len(provider._active_span_processor._span_processors)
|
|
telemetry._attach_common_attributes(provider)
|
|
telemetry._attach_common_attributes(provider)
|
|
after = len(provider._active_span_processor._span_processors)
|
|
|
|
assert after - before == 1
|
|
|
|
|
|
def test_attaching_to_a_provider_without_processors_is_safe(isolated_telemetry):
|
|
"""A NoOp provider has no add_span_processor; this must not raise."""
|
|
telemetry = isolated_telemetry()
|
|
|
|
telemetry._attach_common_attributes(object())
|
|
|
|
|
|
def _common_attributes(monkeypatch, project_id=None):
|
|
"""Build the process-wide span attributes with a stubbed project id."""
|
|
from crewai.telemetry.telemetry import Telemetry
|
|
|
|
monkeypatch.setattr(
|
|
"crewai.telemetry.telemetry.get_project_id", lambda *a, **k: project_id
|
|
)
|
|
telemetry = Telemetry.__new__(Telemetry)
|
|
telemetry._common_attributes = None
|
|
return telemetry._common_span_attributes()
|
|
|
|
|
|
def test_common_attributes_carry_agent_and_runtime(clean_env, monkeypatch):
|
|
"""Both fields ride on every span, independently of each other."""
|
|
clean_env.setenv("CLAUDECODE", "1")
|
|
clean_env.setenv("GITHUB_ACTIONS", "true")
|
|
|
|
attributes = _common_attributes(monkeypatch)
|
|
|
|
assert attributes["coding_agent"] == "claude_code"
|
|
assert attributes["runtime_context"] == "ci"
|
|
|
|
|
|
def test_common_attributes_include_project_id_when_declared(clean_env, monkeypatch):
|
|
attributes = _common_attributes(monkeypatch, project_id="proj-123")
|
|
|
|
assert attributes["project_id"] == "proj-123"
|
|
|
|
|
|
def test_project_id_is_omitted_when_absent(clean_env, monkeypatch):
|
|
"""Projects without an id must not report a placeholder."""
|
|
attributes = _common_attributes(monkeypatch, project_id=None)
|
|
|
|
assert "project_id" not in attributes
|
|
|
|
|
|
def test_project_id_lookup_never_breaks_telemetry(clean_env, monkeypatch):
|
|
"""A failed lookup degrades to omitting the attribute."""
|
|
from crewai.telemetry.telemetry import Telemetry
|
|
|
|
def boom(*args, **kwargs):
|
|
raise OSError("unreadable")
|
|
|
|
monkeypatch.setattr("crewai.telemetry.telemetry.get_project_id", boom)
|
|
telemetry = Telemetry.__new__(Telemetry)
|
|
telemetry._common_attributes = None
|
|
|
|
attributes = telemetry._common_span_attributes()
|
|
|
|
assert "project_id" not in attributes
|
|
assert "coding_agent" in attributes
|
|
|
|
|
|
def test_common_attributes_are_computed_once(clean_env, monkeypatch):
|
|
"""The project file must not be re-read for each provider."""
|
|
from crewai.telemetry.telemetry import Telemetry
|
|
|
|
calls = []
|
|
|
|
def counting_get_project_id(*args, **kwargs):
|
|
calls.append(1)
|
|
return "proj-123"
|
|
|
|
monkeypatch.setattr(
|
|
"crewai.telemetry.telemetry.get_project_id", counting_get_project_id
|
|
)
|
|
telemetry = Telemetry.__new__(Telemetry)
|
|
telemetry._common_attributes = None
|
|
|
|
first = telemetry._common_span_attributes()
|
|
second = telemetry._common_span_attributes()
|
|
|
|
assert first is second
|
|
assert len(calls) == 1
|
|
|
|
|
|
def test_all_common_attributes_land_on_exported_spans(clean_env, monkeypatch, otel_enabled):
|
|
"""End-to-end: every common attribute survives onto arbitrary spans."""
|
|
from opentelemetry.sdk.trace import TracerProvider
|
|
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
|
|
from opentelemetry.sdk.trace.export.in_memory_span_exporter import (
|
|
InMemorySpanExporter,
|
|
)
|
|
|
|
from crewai.telemetry.telemetry import CommonAttributesSpanProcessor
|
|
|
|
clean_env.setenv("CLAUDECODE", "1")
|
|
clean_env.setenv("CI", "true")
|
|
attributes = _common_attributes(monkeypatch, project_id="proj-123")
|
|
|
|
exporter = InMemorySpanExporter()
|
|
provider = TracerProvider()
|
|
provider.add_span_processor(CommonAttributesSpanProcessor(attributes))
|
|
provider.add_span_processor(SimpleSpanProcessor(exporter))
|
|
|
|
provider.get_tracer("test").start_span("Feature Usage").end()
|
|
provider.force_flush()
|
|
|
|
exported = dict(exporter.get_finished_spans()[0].attributes)
|
|
assert exported["coding_agent"] == "claude_code"
|
|
assert exported["runtime_context"] == "ci"
|
|
assert exported["project_id"] == "proj-123"
|
|
|
|
|
|
def test_runtime_markers_are_detected_by_presence(clean_env, monkeypatch):
|
|
"""An empty value still means the platform set the marker.
|
|
|
|
Some platforms export a bare `CI=`; truthiness checks would drop those
|
|
runs to the TTY fallback and mislabel them as ordinary local executions.
|
|
"""
|
|
monkeypatch.setattr("os.path.exists", lambda path: False)
|
|
clean_env.setenv("CI", "")
|
|
|
|
assert detect_runtime_context() == "ci"
|
|
|
|
|
|
def test_managed_platforms_are_not_reported_as_serverless(clean_env):
|
|
"""Long-lived managed platforms must not claim the serverless label.
|
|
|
|
DYNO and WEBSITE_INSTANCE_ID mark Heroku dynos and Azure App Service
|
|
instances, which are containers rather than per-invocation functions.
|
|
"""
|
|
clean_env.setenv("DYNO", "web.1")
|
|
assert detect_runtime_context() == "paas"
|
|
|
|
clean_env.delenv("DYNO")
|
|
clean_env.setenv("WEBSITE_INSTANCE_ID", "abc123")
|
|
assert detect_runtime_context() == "paas"
|
|
|
|
|
|
def test_serverless_markers_still_win_over_paas(clean_env):
|
|
clean_env.setenv("DYNO", "web.1")
|
|
clean_env.setenv("AWS_LAMBDA_FUNCTION_NAME", "my-fn")
|
|
|
|
assert detect_runtime_context() == "serverless"
|
|
|
|
|
|
def test_generic_marker_is_detected_by_presence(clean_env):
|
|
"""An empty AI_AGENT still says an assistant is present.
|
|
|
|
The named markers keep truthiness, where an empty value means the tool set
|
|
a placeholder rather than claiming the session.
|
|
"""
|
|
clean_env.setenv("AI_AGENT", "")
|
|
|
|
assert detect_coding_agent() == "other"
|
|
|
|
|
|
def test_azure_functions_are_not_reported_as_paas(clean_env):
|
|
"""Azure Functions run on the App Service host and inherit its marker."""
|
|
clean_env.setenv("WEBSITE_INSTANCE_ID", "abc123")
|
|
clean_env.setenv("FUNCTIONS_WORKER_RUNTIME", "python")
|
|
|
|
assert detect_runtime_context() == "serverless"
|
|
|
|
|
|
def test_env_context_precedence_matches_the_shared_table(clean_env):
|
|
"""Both detection paths must agree on which assistant is present.
|
|
|
|
get_env_context previously restated precedence, so a marker added for
|
|
telemetry was invisible here and the two disagreed.
|
|
"""
|
|
from crewai.events.types.env_events import (
|
|
CCEnvEvent,
|
|
CodexEnvEvent,
|
|
CursorEnvEvent,
|
|
DefaultEnvEvent,
|
|
)
|
|
from crewai.utilities import env as env_module
|
|
|
|
agent_to_event = {
|
|
"claude_code": CCEnvEvent,
|
|
"codex": CodexEnvEvent,
|
|
"cursor": CursorEnvEvent,
|
|
}
|
|
|
|
for agent, env_vars in CODING_AGENT_ENV_MARKERS:
|
|
for var in env_vars:
|
|
for other in ALL_MARKERS:
|
|
clean_env.delenv(other, raising=False)
|
|
clean_env.setenv(var, "1")
|
|
|
|
emitted: list[type] = []
|
|
clean_env.setattr(
|
|
env_module.crewai_event_bus,
|
|
"emit",
|
|
lambda _source, event, sink=emitted: sink.append(type(event)),
|
|
)
|
|
env_module._env_context_emitted.set(False)
|
|
env_module.get_env_context()
|
|
|
|
assert detect_coding_agent() == agent, var
|
|
assert emitted[0] is agent_to_event.get(agent, DefaultEnvEvent), var
|
|
|
|
|
|
def test_assistant_inside_cursor_agrees_across_both_paths(clean_env):
|
|
"""Cursor sets CURSOR_* in every terminal, including for other assistants."""
|
|
from crewai.events.types.env_events import DefaultEnvEvent
|
|
from crewai.utilities import env as env_module
|
|
|
|
clean_env.setenv("CURSOR_TRACE_ID", "t-1")
|
|
clean_env.setenv("CLINE_ACTIVE", "true")
|
|
|
|
emitted: list[type] = []
|
|
clean_env.setattr(
|
|
env_module.crewai_event_bus,
|
|
"emit",
|
|
lambda _source, event, sink=emitted: sink.append(type(event)),
|
|
)
|
|
env_module._env_context_emitted.set(False)
|
|
env_module.get_env_context()
|
|
|
|
assert detect_coding_agent() == "cline"
|
|
assert emitted[0] is DefaultEnvEvent
|