feat(conformance): implement live mode for conformance testing

Enable `adk conformance test --mode=live` to run agent with live model
calls and compare against recorded baselines. This catches regressions when
model behavior changes while maintaining the same comparison logic as replay
mode.

Live mode runs actual model calls (no replay config) but still validates
against existing recordings, making it suitable for regression detection when
model behavior may have changed.

Fixes #7290

Author: joshi-07

src/google/adk/cli/cli_tools_click.py

@@ -646,7 +646,7 @@ def cli_conformance_record(
     show_default=True,
     help=(
         "Test mode: 'replay' verifies against recorded interactions, 'live'"
-        " runs evaluation-based verification."
+        " runs against live model responses."
     ),
 )
 @click.option(
@@ -683,20 +683,23 @@ def cli_conformance_test(
   """Run conformance tests to verify agent behavior consistency.
 
   Validates that agents produce consistent outputs by comparing against recorded
-  interactions or evaluating live execution results.
+  interactions or checking against live model responses.
 
   PATHS can be any number of folder paths. Each folder can either:
   - Contain a spec.yaml file directly (single test case)
   - Contain subdirectories with spec.yaml files (multiple test cases)
 
   If no paths are provided, defaults to searching for the 'tests' folder.
 
-  TEST MODES:
+  MODE DIFFERENCES:
 
-  \b
-  replay  : Verifies agent interactions match previously recorded behaviors
-            exactly. Compares LLM requests/responses and tool calls/results.
-  live    : Runs evaluation-based verification (not yet implemented)
+  Use 'adk conformance test --mode=replay' for fast, deterministic verification
+  against previously recorded interactions. This is ideal for CI/CD pipelines
+  and ensuring changes don't break existing behavior.
+
+  Use 'adk conformance test --mode=live' to verify against live model responses
+  and catch regressions when model behavior may have changed. This runs actual
+  model calls but still compares against recorded baselines.
 
   DIRECTORY STRUCTURE:
 
@@ -706,10 +709,13 @@ def cli_conformance_test(
   category/
     test_name/
       spec.yaml                     # Test specification
-      generated-recordings.yaml     # Recorded interactions (replay mode)
-      generated-session.yaml        # Session data (replay mode)
-      generated-recordings-sse.yaml # Recorded SSE interactions (replay mode)
-      generated-session-sse.yaml    # SSE Session data (replay mode)
+      generated-recordings.yaml     # Recorded interactions (required for both modes)
+      generated-session.yaml        # Session data (required for both modes)
+      generated-recordings-sse.yaml # Recorded SSE interactions (SSE mode)
+      generated-session-sse.yaml    # SSE Session data (SSE mode)
+
+  NOTE: Both replay and live modes require existing recordings. Use
+  'adk conformance record' to generate baseline recordings first.
 
   REPORT GENERATION:
 
@@ -731,7 +737,7 @@ def cli_conformance_test(
   adk conformance test tests/core/description_001
 
   \b
-  # Run in live mode (when available)
+  # Run in live mode to check against live model responses
   adk conformance test --mode=live tests/core
 
   \b

src/google/adk/cli/conformance/cli_test.py

@@ -103,10 +103,10 @@ def _discover_test_cases(self) -> list[TestCase]:
         else:
           raise ValueError(f"Unsupported streaming mode: {self.streaming_mode}")
 
-        # Skip if recordings missing in replay mode
-        if self.mode == "replay" and not recordings_file.exists():
+        # Skip if recordings missing in replay or live mode (need baseline for comparison)
+        if not recordings_file.exists():
           click.secho(
-              f"Skipping {category}/{name}: no recordings",
+              f"Skipping {category}/{name}: no recordings found for comparison",
               fg="yellow",
               err=True,
           )
@@ -129,7 +129,7 @@ async def _run_user_messages(
       session_id: str,
       test_case: TestCase,
   ) -> None:
-    """Run all user messages for a test case."""
+    """Run all user messages for a test case in replay mode."""
     function_call_name_to_id_map = {}
     for user_message_index, user_message in enumerate(
         test_case.test_spec.user_messages
@@ -178,7 +178,7 @@ async def _run_user_messages(
           state_delta=user_message.state_delta,
       )
 
-      # Run the agent but don't collect events here
+      # Run the agent in replay mode
       async for event in self.client.run_agent(
           request,
           mode="replay",
@@ -289,6 +289,114 @@ async def _run_test_case_replay(self, test_case: TestCase) -> _TestResult:
           description=test_case.test_spec.description,
       )
 
+  async def _run_user_messages_live(
+      self,
+      session_id: str,
+      test_case: TestCase,
+  ) -> None:
+    """Run all user messages for a test case in live mode."""
+    function_call_name_to_id_map = {}
+    for user_message_index, user_message in enumerate(
+        test_case.test_spec.user_messages
+    ):
+      # Create content from UserMessage object
+      if user_message.content is not None:
+        content = user_message.content.model_copy(deep=True)
+
+        # If the user provides a function response, it means this is for
+        # long-running tool. Replace the function call ID with the actual
+        # function call ID. This is needed because the function call ID is not
+        # known when writing the test case.
+        if content.parts:
+          for part in content.parts:
+            if part.function_response:
+              name = part.function_response.name
+              if not name:
+                raise ValueError(
+                    "FunctionResponse part is missing a 'name' field."
+                )
+              if (
+                  name not in function_call_name_to_id_map
+                  or not function_call_name_to_id_map[name]
+              ):
+                raise ValueError(
+                    "Function response for"
+                    f" {name} does not match any pending function call."
+                )
+              part.function_response.id = function_call_name_to_id_map[
+                  name
+              ].pop(0)
+      elif user_message.text is not None:
+        content = types.UserContent(parts=[types.Part(text=user_message.text)])
+      else:
+        raise ValueError(
+            f"UserMessage at index {user_message_index} has neither text nor"
+            " content"
+        )
+
+      request = RunAgentRequest(
+          app_name=test_case.test_spec.agent,
+          user_id=self.user_id,
+          session_id=session_id,
+          new_message=content,
+          streaming=self.streaming_mode == StreamingMode.SSE,
+          state_delta=user_message.state_delta,
+      )
+
+      # Run the agent with live model calls (no mode parameter)
+      async for event in self.client.run_agent(request):
+        if getattr(event, "partial", False):
+          continue
+        if event.content and event.content.parts:
+          for part in event.content.parts:
+            if part.function_call:
+              function_call_name_to_id_map.setdefault(
+                  part.function_call.name, []
+              ).append(part.function_call.id)
+
+  async def _run_test_case_live(self, test_case: TestCase) -> _TestResult:
+    """Run a single test case in live mode."""
+    try:
+      # Create session
+      session = await self.client.create_session(
+          app_name=test_case.test_spec.agent,
+          user_id=self.user_id,
+          state=test_case.test_spec.initial_state,
+      )
+
+      # Run each user message with live model calls
+      try:
+        await self._run_user_messages_live(session.id, test_case)
+      except Exception as e:
+        return _TestResult(
+            category=test_case.category,
+            name=test_case.name,
+            success=False,
+            error_message=f"Live execution failed: {e}",
+            description=test_case.test_spec.description,
+        )
+
+      # Validate results against existing recordings
+      result = await self._validate_test_results(session.id, test_case)
+
+      # Clean up session
+      await self.client.delete_session(
+          app_name=test_case.test_spec.agent,
+          user_id=self.user_id,
+          session_id=session.id,
+      )
+
+      return result
+
+    except Exception as e:
+      return _TestResult(
+          category=test_case.category,
+          name=test_case.name,
+          success=False,
+          error_message=f"Test setup failed: {e}",
+          description=test_case.test_spec.description,
+      )
+
   async def run_all_tests(self) -> _ConformanceTestSummary:
     """Run all discovered test cases."""
     test_cases = self._discover_test_cases()
@@ -311,13 +419,14 @@ async def run_all_tests(self) -> _ConformanceTestSummary:
       click.echo(f"Running {test_case.category}/{test_case.name}...", nl=False)
       if self.mode == "replay":
         result = await self._run_test_case_replay(test_case)
+      elif self.mode == "live":
+        result = await self._run_test_case_live(test_case)
       else:
-        # TODO: Implement live mode
         result = _TestResult(
             category=test_case.category,
             name=test_case.name,
             success=False,
-            error_message="Live mode is not implemented yet",
+            error_message=f"Unsupported mode: {self.mode}",
             description=test_case.test_spec.description,
         )
       results.append(result)

tests/unittests/cli/conformance/test_cli_test.py

@@ -14,6 +14,8 @@
 
 """Tests for cli_test.py."""
 
+from __future__ import annotations
+
 from typing import Optional
 from unittest.mock import MagicMock
 
@@ -377,3 +379,78 @@ def test_summary_succeeds_when_every_streaming_mode_passes():
 
   assert result.exit_code == 0
   assert result.output.count("All tests passed!") == 2
+
+
+@pytest.mark.asyncio
+async def test_run_user_messages_live_mode():
+  """Test that live mode runs agent without replay config."""
+  client = MagicMock()
+  captured_requests = []
+
+  async def fake_run_agent(req, **kwargs):
+    captured_requests.append(req)
+    # Live mode should not have mode parameter
+    assert "mode" not in kwargs
+    yield Event(partial=False, content=types.Content(parts=[types.Part.from_text("response")]))
+
+  client.run_agent = fake_run_agent
+
+  runner = ConformanceTestRunner([], client, mode="live")
+  test_case = TestCase(
+      category="cat",
+      name="tc",
+      dir=None,
+      test_spec=TestSpec(
+          description="test live mode",
+          agent="agent",
+          user_messages=[UserMessage(text="test message")],
+      ),
+  )
+  await runner._run_user_messages_live("sess1", test_case)
+
+  assert len(captured_requests) == 1
+  assert captured_requests[0].new_message.parts[0].text == "test message"
+
+
+@pytest.mark.asyncio
+async def test_run_test_case_live():
+  """Test complete live mode test case execution."""
+  client = MagicMock()
+  session = MagicMock()
+  session.id = "test-session"
+
+  async def fake_create_session(**kwargs):
+    return session
+
+  async def fake_get_session(**kwargs):
+    return MagicMock(events=[])
+
+  async def fake_delete_session(**kwargs):
+    pass
+
+  client.create_session = fake_create_session
+  client.get_session = fake_get_session
+  client.delete_session = fake_delete_session
+
+  runner = ConformanceTestRunner([], client, mode="live")
+  test_case = TestCase(
+      category="cat",
+      name="tc",
+      dir=None,
+      test_spec=TestSpec(
+          description="test live mode",
+          agent="agent",
+          user_messages=[UserMessage(text="test message")],
+      ),
+  )
+
+  # Mock the user messages execution
+  runner._run_user_messages_live = MagicMock(return_value=None)
+
+  result = await runner._run_test_case_live(test_case)
+
+  assert result.category == "cat"
+  assert result.name == "tc"
+  # Should fail because no recordings exist for comparison
+  assert not result.success
+  assert "No recorded session found" in result.error_message