diff --git a/samcli/lib/telemetry/agent_detector.py b/samcli/lib/telemetry/agent_detector.py new file mode 100644 index 00000000000..84a81a6f76d --- /dev/null +++ b/samcli/lib/telemetry/agent_detector.py @@ -0,0 +1,107 @@ +""" +Module used for detecting whether SAMCLI is executed by an AI agent. +""" + +import os +from enum import Enum, auto +from typing import Callable, Dict, Mapping, Optional, Union + + +class Agent(Enum): + ClaudeCode = auto() + Codex = auto() + Cursor = auto() + GeminiCLI = auto() + Antigravity = auto() + Kiro = auto() + OpenCode = auto() + GitHubCopilot = auto() + + +def _is_codex(environ: Mapping) -> bool: + """ + Whether it is running in Codex (OpenAI's Codex CLI). Codex sets a family of + CODEX_* variables rather than one canonical var (e.g. CODEX_SANDBOX, CODEX_CI, + CODEX_THREAD_ID), so match any name starting with "CODEX_". + """ + return any(key.startswith("CODEX_") for key in environ) + + +def _is_kiro(environ: Mapping) -> bool: + """ + Whether it is running in Kiro (the Kiro IDE, via TERM_PROGRAM=kiro, or the Kiro + CLI, the renamed Amazon Q Developer CLI, via AWS_EXECUTION_ENV). AWS_EXECUTION_ENV + is shared with AWS Lambda/CloudShell/CodeBuild (e.g. "AWS_Lambda_python3.12"), so + it is substring-matched on "amazonq"/"kiro", not checked for presence, to avoid + misattributing those environments. + + NOTE: the "amazonq" substring also matches the genuine Amazon Q Developer CLI. + Amazon Q is deferred for now, so its CLI runs are attributed to kiro until a + dedicated Amazon Q agent is added (which must be ordered AFTER Kiro in the enum). + """ + if environ.get("TERM_PROGRAM", "") == "kiro": + return True + aws_execution_env: str = environ.get("AWS_EXECUTION_ENV", "").lower() + return "amazonq" in aws_execution_env or "kiro" in aws_execution_env + + +_ENV_VAR_OR_CALLABLE_BY_AGENT: Dict[Agent, Union[str, Callable[[Mapping], bool]]] = { + # https://docs.anthropic.com/en/docs/claude-code + Agent.ClaudeCode: "CLAUDECODE", + # https://developers.openai.com/codex/cli + Agent.Codex: _is_codex, + # match presence, not value: the value is not guaranteed stable + Agent.Cursor: "CURSOR_AGENT", + # exact name, not a prefix, so GEMINI_CLI_HOME / GEMINI_CLI_NO_RELAUNCH don't match + Agent.GeminiCLI: "GEMINI_CLI", + # Google Antigravity CLI (agy); replaces Gemini CLI for consumer users + Agent.Antigravity: "ANTIGRAVITY_AGENT", + Agent.Kiro: _is_kiro, + # OpenCode sets OPENCODE=1 in its root CLI middleware + Agent.OpenCode: "OPENCODE", + Agent.GitHubCopilot: "COPILOT_AGENT_SESSION_ID", +} + + +def _is_agent(agent: Agent, environ: Mapping) -> bool: + """ + Check whether sam-cli is run by a particular AI agent based on certain environment variables. + + Parameters + ---------- + agent + an enum Agent object indicating which AI agent to check against. + environ + the mapping to look for environment variables, for example, os.environ. + + Returns + ------- + bool + A boolean indicating whether there are environment variables matching the agent. + """ + env_var_or_callable = _ENV_VAR_OR_CALLABLE_BY_AGENT[agent] + if isinstance(env_var_or_callable, str): + return env_var_or_callable in environ + + # it is a callable, use the return value + return env_var_or_callable(environ) + + +class AgentDetector: + _agent: Optional[Agent] + + def __init__(self): + try: + self._agent: Optional[Agent] = next(agent for agent in Agent if _is_agent(agent, os.environ)) + except StopIteration: + self._agent = None + + def agent(self) -> Optional[Agent]: + """ + Identify which AI agent SAM CLI is running in. + Returns + ------- + Agent + an optional Agent enum indicating the AI agent. + """ + return self._agent diff --git a/samcli/lib/telemetry/user_agent.py b/samcli/lib/telemetry/user_agent.py index c6b6e6bee96..8e5cc042688 100644 --- a/samcli/lib/telemetry/user_agent.py +++ b/samcli/lib/telemetry/user_agent.py @@ -4,10 +4,34 @@ import os import re -from typing import Optional +from typing import Dict, Optional + +from samcli.lib.telemetry.agent_detector import Agent, AgentDetector USER_AGENT_ENV_VAR = "AWS_TOOLING_USER_AGENT" +# Placeholder version emitted for every detected AI agent. Agent CLIs (e.g. Claude +# Code) do not reliably expose their own version through the environment, so we use +# a fixed "1.0" and rely on the agent name for grouping. Bump this only if the +# emitted user-agent format itself needs to change. +AGENT_USER_AGENT_VERSION = "1.0" + +# Stable, lowercase-with-dashes names for detected AI agents. These are combined +# with AGENT_USER_AGENT_VERSION to build the user-agent string (e.g. +# "claude-code/1.0") that lands in the queryable "userAgent" telemetry column, so +# they must remain stable across releases to keep downstream queries valid. When +# adding an agent here, also add its detection entry in agent_detector.py. +_USER_AGENT_NAME_BY_AGENT: Dict[Agent, str] = { + Agent.ClaudeCode: "claude-code", + Agent.Codex: "codex", + Agent.Cursor: "cursor", + Agent.GeminiCLI: "gemini-cli", + Agent.Antigravity: "antigravity", + Agent.Kiro: "kiro", + Agent.OpenCode: "opencode", + Agent.GitHubCopilot: "github-copilot", +} + # Should accept format: ${AGENT_NAME}/${AGENT_VERSION} # AWS_Toolkit-For-VSCode/1.62.0 # AWS-Toolkit-For-JetBrains/1.60-223 @@ -16,7 +40,42 @@ def get_user_agent_string() -> Optional[str]: + """ + Return the user-agent string for telemetry, or None if none is available. + + Precedence: + 1. AWS_TOOLING_USER_AGENT, when set to a value matching + ACCEPTED_USER_AGENT_FORMAT. This is set by AWS toolkits (e.g. the VS Code + and JetBrains toolkits) and always wins when present, so a toolkit + invocation is still attributed to the toolkit even if it happens to run + inside an AI agent. + 2. A detected AI agent (e.g. Claude Code), emitted as "/1.0". + This is used only as a fallback when the toolkit user-agent above is not + set to a valid value. + """ user_agent = os.environ.get(USER_AGENT_ENV_VAR, "").strip() if user_agent and ACCEPTED_USER_AGENT_FORMAT.match(user_agent): return user_agent + + return _get_agent_user_agent_string() + + +def _get_agent_user_agent_string() -> Optional[str]: + """ + Build a "/1.0" user-agent string for the detected AI agent, or None + when no agent is detected (or the detected agent has no configured name). + """ + agent = AgentDetector().agent() + if agent is None: + return None + + agent_name = _USER_AGENT_NAME_BY_AGENT.get(agent) + if not agent_name: + return None + + agent_user_agent = f"{agent_name}/{AGENT_USER_AGENT_VERSION}" + # Validate against the same format toolkit strings must satisfy, so a + # misconfigured agent name never emits a malformed user-agent value. + if ACCEPTED_USER_AGENT_FORMAT.match(agent_user_agent): + return agent_user_agent return None diff --git a/tests/unit/lib/telemetry/test_agent_detector.py b/tests/unit/lib/telemetry/test_agent_detector.py new file mode 100644 index 00000000000..4cb98a01959 --- /dev/null +++ b/tests/unit/lib/telemetry/test_agent_detector.py @@ -0,0 +1,92 @@ +from unittest import TestCase +from unittest.mock import patch + +from parameterized import parameterized + +from samcli.lib.telemetry.agent_detector import Agent, AgentDetector, _is_agent + + +class TestAgentDetector(TestCase): + @parameterized.expand( + [ + (Agent.ClaudeCode, "CLAUDECODE", "1"), + (Agent.Codex, "CODEX_SANDBOX", "seatbelt"), + (Agent.Cursor, "CURSOR_AGENT", "1"), + (Agent.GeminiCLI, "GEMINI_CLI", "1"), + (Agent.Antigravity, "ANTIGRAVITY_AGENT", "1"), + (Agent.Kiro, "TERM_PROGRAM", "kiro"), + (Agent.Kiro, "AWS_EXECUTION_ENV", "AmazonQ-For-CLI Version/1.13.3"), + (Agent.OpenCode, "OPENCODE", "1"), + (Agent.GitHubCopilot, "COPILOT_AGENT_SESSION_ID", "0b8f6e2c-1a3d-4c9e-8f7a-2b1c0d9e8f7a"), + ] + ) + def test_is_agent(self, agent, env_var, env_var_value): + self.assertTrue(_is_agent(agent, {env_var: env_var_value})) + + @parameterized.expand( + [ + (Agent.ClaudeCode, "NOT_CLAUDECODE", "1"), + (Agent.Codex, "NOT_CODEX_SANDBOX", "seatbelt"), + (Agent.Cursor, "NOT_CURSOR_AGENT", "1"), + # exact-name rule: GEMINI_CLI_HOME must not match Agent.GeminiCLI + (Agent.GeminiCLI, "GEMINI_CLI_HOME", "1"), + # shared-variable rule: a Lambda AWS_EXECUTION_ENV must not match Agent.Kiro + (Agent.Kiro, "AWS_EXECUTION_ENV", "AWS_Lambda_python3.12"), + # too-generic var: AGENT alone (no OPENCODE) must not match Agent.OpenCode + (Agent.OpenCode, "AGENT", "1"), + ] + ) + def test_is_not_agent(self, agent, env_var, env_var_value): + self.assertFalse(_is_agent(agent, {env_var: env_var_value})) + + @patch.dict("samcli.lib.telemetry.agent_detector.os.environ", {"CLAUDECODE": "1"}, clear=True) + def test_detector_identifies_claude_code(self): + self.assertEqual(AgentDetector().agent(), Agent.ClaudeCode) + + @patch.dict("samcli.lib.telemetry.agent_detector.os.environ", {"CODEX_SANDBOX": "seatbelt"}, clear=True) + def test_detector_identifies_codex(self): + self.assertEqual(AgentDetector().agent(), Agent.Codex) + + @patch.dict("samcli.lib.telemetry.agent_detector.os.environ", {"CURSOR_AGENT": "1"}, clear=True) + def test_detector_identifies_cursor(self): + self.assertEqual(AgentDetector().agent(), Agent.Cursor) + + @patch.dict("samcli.lib.telemetry.agent_detector.os.environ", {"GEMINI_CLI": "1"}, clear=True) + def test_detector_identifies_gemini_cli(self): + self.assertEqual(AgentDetector().agent(), Agent.GeminiCLI) + + @patch.dict("samcli.lib.telemetry.agent_detector.os.environ", {"TERM_PROGRAM": "kiro"}, clear=True) + def test_detector_identifies_kiro_ide(self): + self.assertEqual(AgentDetector().agent(), Agent.Kiro) + + @patch.dict( + "samcli.lib.telemetry.agent_detector.os.environ", + {"AWS_EXECUTION_ENV": "AmazonQ-For-CLI Version/1.13.3"}, + clear=True, + ) + def test_detector_identifies_kiro_cli(self): + self.assertEqual(AgentDetector().agent(), Agent.Kiro) + + @patch.dict( + "samcli.lib.telemetry.agent_detector.os.environ", + {"AWS_EXECUTION_ENV": "AWS_Lambda_python3.12"}, + clear=True, + ) + def test_detector_does_not_identify_lambda_as_kiro(self): + self.assertIsNone(AgentDetector().agent()) + + @patch.dict("samcli.lib.telemetry.agent_detector.os.environ", {"OPENCODE": "1"}, clear=True) + def test_detector_identifies_opencode(self): + self.assertEqual(AgentDetector().agent(), Agent.OpenCode) + + @patch.dict( + "samcli.lib.telemetry.agent_detector.os.environ", + {"COPILOT_AGENT_SESSION_ID": "0b8f6e2c-1a3d-4c9e-8f7a-2b1c0d9e8f7a"}, + clear=True, + ) + def test_detector_identifies_github_copilot(self): + self.assertEqual(AgentDetector().agent(), Agent.GitHubCopilot) + + @patch.dict("samcli.lib.telemetry.agent_detector.os.environ", {}, clear=True) + def test_detector_returns_none_when_no_agent(self): + self.assertIsNone(AgentDetector().agent()) diff --git a/tests/unit/lib/telemetry/test_event.py b/tests/unit/lib/telemetry/test_event.py index 3735940d5dd..0554af8c349 100644 --- a/tests/unit/lib/telemetry/test_event.py +++ b/tests/unit/lib/telemetry/test_event.py @@ -130,7 +130,12 @@ def test_track_event(self, event_mock, lock_mock): lock_mock.__exit__.assert_called() @patch("samcli.lib.telemetry.event.Telemetry") - def test_events_get_sent(self, telemetry_mock): + @patch("samcli.lib.telemetry.metric.get_user_agent_string") + def test_events_get_sent(self, get_user_agent_mock, telemetry_mock): + # A detected AI agent adds a "userAgent" field to every metric's common + # attributes; pin it to a known value so the exact-equality assertion below + # verifies the field is emitted rather than breaking on it. + get_user_agent_mock.return_value = "claude-code/1.0" # Create fake emit to capture tracked events dummy_telemetry = Mock() emitted_events = [] @@ -168,6 +173,7 @@ def test_events_get_sent(self, telemetry_mock): "samcliVersion": ANY, "osPlatform": ANY, "commandName": ANY, + "userAgent": "claude-code/1.0", "metricSpecificAttributes": { "containerEngine": ANY, "events": [ diff --git a/tests/unit/lib/telemetry/test_user_agent.py b/tests/unit/lib/telemetry/test_user_agent.py index 6fb14a83dd5..a4bf54f2331 100644 --- a/tests/unit/lib/telemetry/test_user_agent.py +++ b/tests/unit/lib/telemetry/test_user_agent.py @@ -3,7 +3,11 @@ from parameterized import parameterized -from samcli.lib.telemetry.user_agent import USER_AGENT_ENV_VAR, get_user_agent_string +from samcli.lib.telemetry.user_agent import ( + ACCEPTED_USER_AGENT_FORMAT, + USER_AGENT_ENV_VAR, + get_user_agent_string, +) class TestUserAgent(TestCase): @@ -43,3 +47,48 @@ def test_user_agent_without_env_var(self): @patch("samcli.lib.telemetry.user_agent.os.environ", {USER_AGENT_ENV_VAR: ""}) def test_user_agent_with_empty_env_var(self): self.assertEqual(get_user_agent_string(), None) + + +class TestUserAgentAgentFallback(TestCase): + # NOTE: user_agent.py and agent_detector.py both read os.environ, which is a + # single shared object, so one patch.dict on os.environ controls the env seen + # by both the toolkit lookup and the agent detector. clear=True keeps the + # ambient CLAUDECODE of the running shell from leaking into these tests. + @patch.dict("samcli.lib.telemetry.user_agent.os.environ", {"CLAUDECODE": "1"}, clear=True) + def test_detected_agent_is_used_when_no_toolkit_user_agent(self): + # No AWS_TOOLING_USER_AGENT set, Claude Code detected -> agent string is emitted. + self.assertEqual(get_user_agent_string(), "claude-code/1.0") + + @parameterized.expand( + [ + ("AWS_Toolkit-For-VSCode/1.62.0",), + ("AWS-Toolkit-For-JetBrains/1.60-223",), + ] + ) + def test_toolkit_user_agent_takes_precedence_over_detected_agent(self, toolkit_user_agent): + # A valid AWS_TOOLING_USER_AGENT wins even when an agent is detected. + with patch.dict( + "samcli.lib.telemetry.user_agent.os.environ", + {"CLAUDECODE": "1", USER_AGENT_ENV_VAR: toolkit_user_agent}, + clear=True, + ): + self.assertEqual(get_user_agent_string(), toolkit_user_agent) + + @patch.dict( + "samcli.lib.telemetry.user_agent.os.environ", + {"CLAUDECODE": "1", USER_AGENT_ENV_VAR: "invalid_value"}, + clear=True, + ) + def test_detected_agent_is_used_when_toolkit_user_agent_is_invalid(self): + # An invalid AWS_TOOLING_USER_AGENT falls through to the detected agent. + self.assertEqual(get_user_agent_string(), "claude-code/1.0") + + @patch.dict("samcli.lib.telemetry.user_agent.os.environ", {}, clear=True) + def test_none_when_no_toolkit_user_agent_and_no_agent(self): + self.assertEqual(get_user_agent_string(), None) + + @patch.dict("samcli.lib.telemetry.user_agent.os.environ", {"CLAUDECODE": "1"}, clear=True) + def test_emitted_agent_string_matches_accepted_format(self): + # Guard against emitting a bare "ClaudeCode": the fallback must satisfy the + # same regex the toolkit user-agent must satisfy. + self.assertIsNotNone(ACCEPTED_USER_AGENT_FORMAT.match(get_user_agent_string()))