From 1e1f9c17c85ddc27ac70bdfa90193fbcbc03ca75 Mon Sep 17 00:00:00 2001 From: Madhav Donthula Date: Mon, 13 Jul 2026 16:53:14 -0700 Subject: [PATCH 1/2] feat: detect AI agent invocations in telemetry Add an AgentDetector, modeled on the existing CICDDetector, that identifies when SAM CLI is invoked by an AI coding agent by reading environment variables the agent sets, and records it via the existing userAgent telemetry field (emitted as "/1.0"). This lets us distinguish agent-driven usage from human and CI/CD usage. Detected agents and their signals: - Claude Code CLAUDECODE present - Codex any CODEX_* var present - Cursor CURSOR_AGENT present - Gemini CLI GEMINI_CLI present - Kiro TERM_PROGRAM == kiro, or AWS_EXECUTION_ENV contains amazonq/kiro - OpenCode OPENCODE present - GitHub Copilot COPILOT_AGENT_SESSION_ID present, or GITHUB_COPILOT_CLI_MODE == "true" Shared variables (AWS_EXECUTION_ENV, TERM_PROGRAM) are substring or exact-value matched, never presence-checked, to avoid misattributing Lambda/CloudShell/CodeBuild or generic VS Code environments as agents. GitHub Copilot detection deliberately avoids any GITHUB_* / GITHUB_ACTION key so it never false-matches ordinary GitHub Actions CI, which the existing CICDDetector already handles. An existing AWS_TOOLING_USER_AGENT (AWS toolkit) takes precedence over a detected agent. Amazon Q is intentionally deferred; until it is added after Kiro in resolution order, Amazon Q CLI runs resolve to kiro. Includes unit tests for each agent, the exact-name/too-generic guards (GEMINI_CLI, OPENCODE vs AGENT), the AWS_EXECUTION_ENV and GITHUB_ACTION false-positive guards, and toolkit-takes-precedence behavior. --- samcli/lib/telemetry/agent_detector.py | 104 ++++++++++++++++++ samcli/lib/telemetry/user_agent.py | 60 +++++++++- .../unit/lib/telemetry/test_agent_detector.py | 91 +++++++++++++++ tests/unit/lib/telemetry/test_event.py | 8 +- tests/unit/lib/telemetry/test_user_agent.py | 51 ++++++++- 5 files changed, 311 insertions(+), 3 deletions(-) create mode 100644 samcli/lib/telemetry/agent_detector.py create mode 100644 tests/unit/lib/telemetry/test_agent_detector.py diff --git a/samcli/lib/telemetry/agent_detector.py b/samcli/lib/telemetry/agent_detector.py new file mode 100644 index 00000000000..1d9f224e735 --- /dev/null +++ b/samcli/lib/telemetry/agent_detector.py @@ -0,0 +1,104 @@ +""" +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() + 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", + 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..83540beb6ed 100644 --- a/samcli/lib/telemetry/user_agent.py +++ b/samcli/lib/telemetry/user_agent.py @@ -4,10 +4,33 @@ 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.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 +39,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..0e5db8f5dfc --- /dev/null +++ b/tests/unit/lib/telemetry/test_agent_detector.py @@ -0,0 +1,91 @@ +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.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())) From 272cae1ae1fbe27dfe85cf7048a30b4142f83684 Mon Sep 17 00:00:00 2001 From: Madhav Donthula Date: Tue, 21 Jul 2026 12:10:25 -0700 Subject: [PATCH 2/2] feat: add Antigravity (Google agy) to AI agent detection Detect the Antigravity CLI via the ANTIGRAVITY_AGENT env var, emitting antigravity/1.0 in telemetry. Antigravity replaces Gemini CLI for consumer users. --- samcli/lib/telemetry/agent_detector.py | 3 +++ samcli/lib/telemetry/user_agent.py | 1 + tests/unit/lib/telemetry/test_agent_detector.py | 1 + 3 files changed, 5 insertions(+) diff --git a/samcli/lib/telemetry/agent_detector.py b/samcli/lib/telemetry/agent_detector.py index 1d9f224e735..84a81a6f76d 100644 --- a/samcli/lib/telemetry/agent_detector.py +++ b/samcli/lib/telemetry/agent_detector.py @@ -12,6 +12,7 @@ class Agent(Enum): Codex = auto() Cursor = auto() GeminiCLI = auto() + Antigravity = auto() Kiro = auto() OpenCode = auto() GitHubCopilot = auto() @@ -53,6 +54,8 @@ def _is_kiro(environ: Mapping) -> bool: 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", diff --git a/samcli/lib/telemetry/user_agent.py b/samcli/lib/telemetry/user_agent.py index 83540beb6ed..8e5cc042688 100644 --- a/samcli/lib/telemetry/user_agent.py +++ b/samcli/lib/telemetry/user_agent.py @@ -26,6 +26,7 @@ Agent.Codex: "codex", Agent.Cursor: "cursor", Agent.GeminiCLI: "gemini-cli", + Agent.Antigravity: "antigravity", Agent.Kiro: "kiro", Agent.OpenCode: "opencode", Agent.GitHubCopilot: "github-copilot", diff --git a/tests/unit/lib/telemetry/test_agent_detector.py b/tests/unit/lib/telemetry/test_agent_detector.py index 0e5db8f5dfc..4cb98a01959 100644 --- a/tests/unit/lib/telemetry/test_agent_detector.py +++ b/tests/unit/lib/telemetry/test_agent_detector.py @@ -13,6 +13,7 @@ class TestAgentDetector(TestCase): (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"),