Simplify Langfuse instrumentation - use SDK's authoritative turn numbers only

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
Signed-off-by: sallyom <somalley@redhat.com>

Author: 2 people

components/operator/internal/handlers/sessions.go

@@ -140,15 +140,29 @@ func handleAgenticSessionEvent(obj *unstructured.Unstructured) error {
 					log.Printf("Successfully deleted job %s for stopped session", jobName)
 				}
 
-				// IMPORTANT: Do NOT explicitly delete pods here
-				// Job deletion with Foreground propagation will automatically cascade to pods
-				// Explicit pod deletion bypasses TerminationGracePeriodSeconds
-				//
-				// Kubernetes will:
-				// 1. Send SIGTERM to container for graceful shutdown
-				// 2. Wait up to TerminationGracePeriodSeconds (30s default) for graceful exit
-				// 3. Send SIGKILL if still running after grace period
-				log.Printf("Pods for job %s will be deleted automatically by Kubernetes", jobName)
+				// Then, explicitly delete all pods for this job (by job-name label)
+				podSelector := fmt.Sprintf("job-name=%s", jobName)
+				log.Printf("Deleting pods with job-name selector: %s", podSelector)
+				err = config.K8sClient.CoreV1().Pods(sessionNamespace).DeleteCollection(context.TODO(), v1.DeleteOptions{}, v1.ListOptions{
+					LabelSelector: podSelector,
+				})
+				if err != nil && !errors.IsNotFound(err) {
+					log.Printf("Failed to delete pods for job %s: %v (continuing anyway)", jobName, err)
+				} else {
+					log.Printf("Successfully deleted pods for job %s", jobName)
+				}
+
+				// Also delete any pods labeled with this session (in case owner refs are lost)
+				sessionPodSelector := fmt.Sprintf("agentic-session=%s", name)
+				log.Printf("Deleting pods with agentic-session selector: %s", sessionPodSelector)
+				err = config.K8sClient.CoreV1().Pods(sessionNamespace).DeleteCollection(context.TODO(), v1.DeleteOptions{}, v1.ListOptions{
+					LabelSelector: sessionPodSelector,
+				})
+				if err != nil && !errors.IsNotFound(err) {
+					log.Printf("Failed to delete session-labeled pods: %v (continuing anyway)", err)
+				} else {
+					log.Printf("Successfully deleted session-labeled pods")
+				}
 			} else {
 				log.Printf("Job %s already completed (Succeeded: %d, Failed: %d), no cleanup needed", jobName, job.Status.Succeeded, job.Status.Failed)
 			}
@@ -299,9 +313,9 @@ func handleAgenticSessionEvent(obj *unstructured.Unstructured) error {
 
 	// Hardcoded secret names (convention over configuration)
 	const runnerSecretsName = "ambient-runner-secrets"               // ANTHROPIC_API_KEY only (ignored when Vertex enabled)
-	const integrationSecretsName = "ambient-non-vertex-integrations" // GIT_*, JIRA_*, custom keys (optional, NO Langfuse keys)
+	const integrationSecretsName = "ambient-non-vertex-integrations" // GIT_*, JIRA_*, custom keys (optional)
 
-	// Check if integration secrets exist (user-provided integrations like GIT_TOKEN, JIRA_*)
+	// Check if integration secrets exist (optional)
 	integrationSecretsExist := false
 	if _, err := config.K8sClient.CoreV1().Secrets(sessionNamespace).Get(context.TODO(), integrationSecretsName, v1.GetOptions{}); err == nil {
 		integrationSecretsExist = true
@@ -383,8 +397,6 @@ func handleAgenticSessionEvent(obj *unstructured.Unstructured) error {
 				},
 				Spec: corev1.PodSpec{
 					RestartPolicy: corev1.RestartPolicyNever,
-					// Default grace period for graceful shutdown
-					TerminationGracePeriodSeconds: int64Ptr(30),
 					// Explicitly set service account for pod creation permissions
 					AutomountServiceAccountToken: boolPtr(false),
 					Volumes: []corev1.Volume{
@@ -611,7 +623,7 @@ func handleAgenticSessionEvent(obj *unstructured.Unstructured) error {
 							}(),
 
 							// Import secrets as environment variables
-							// - integrationSecretsName: Only if exists (GIT_TOKEN, JIRA_*, custom keys - NO Langfuse keys)
+							// - integrationSecretsName: Only if exists (GIT_TOKEN, JIRA_*, custom keys)
 							// - runnerSecretsName: Only when Vertex disabled (ANTHROPIC_API_KEY)
 							// - ambient-langfuse-keys: Platform-wide Langfuse observability (LANGFUSE_PUBLIC_KEY, LANGFUSE_SECRET_KEY, LANGFUSE_HOST, LANGFUSE_ENABLED)
 							EnvFrom: func() []corev1.EnvFromSource {
@@ -641,10 +653,6 @@ func handleAgenticSessionEvent(obj *unstructured.Unstructured) error {
 									log.Printf("Skipping runner secrets '%s' for session %s (Vertex enabled)", runnerSecretsName, name)
 								}
 
-								// Note: Platform-wide Langfuse observability keys are injected via explicit Env entries above
-								// (LANGFUSE_* env vars from ambient-admin-langfuse-secret Secret, platform-admin managed)
-								// EnvFrom is intentionally NOT used here to prevent automatic exposure of future secret keys
-
 								return sources
 							}(),
 

components/runners/claude-code-runner/observability.py

@@ -4,11 +4,12 @@
 Provides Langfuse LLM observability for Claude sessions with trace structure:
 
 1. Turn Traces (top-level generations):
-   - Each turn is a separate trace
-   - Named: claude_turn_1, claude_turn_2, etc.
-   - Contains authoritative usage data from Claude SDK
+   - ONE trace per turn (SDK sends multiple AssistantMessages during streaming, but guard prevents duplicates)
+   - Named: "claude_interaction" (turn number stored in metadata)
+   - First AssistantMessage creates trace, subsequent ones ignored until end_turn() clears it
+   - Final trace contains authoritative turn number and usage data from ResultMessage
    - Canonical format with separate cache token tracking for accurate cost
-   - All turns grouped by session_id via propagate_attributes()
+   - All traces grouped by session_id via propagate_attributes()
 
 2. Tool Spans (observations within turn traces):
    - Named: tool_Read, tool_Write, tool_Bash, etc.
@@ -18,21 +19,20 @@
 
 Architecture:
 - Session-based grouping via propagate_attributes() with session_id and user_id
-- Each turn is an independent trace (not nested under a session trace)
+- Each turn creates ONE independent trace (not nested under session)
 - Langfuse automatically aggregates tokens and costs across all traces with same session_id
-- Filter by session_id, user_id, or model in Langfuse UI
+- Filter by session_id, user_id, model, or metadata.turn in Langfuse UI
+- Sessions can be paused/resumed: each turn creates a trace regardless of session lifecycle
 
 Trace Hierarchy:
-claude_turn_1 (trace - generation)
+claude_interaction (trace - generation, metadata: {turn: 1})
 ├── tool_Read (observation - span)
 └── tool_Write (observation - span)
 
-claude_turn_2 (trace - generation)
+claude_interaction (trace - generation, metadata: {turn: 2})
 └── tool_Bash (observation - span)
 
-claude_turn_3 (trace - generation)
-
-Usage Format (turn-level only):
+Usage Format:
 {
     "input": int,  # Regular input tokens
     "output": int,  # Output tokens
@@ -183,71 +183,73 @@ async def initialize(self, prompt: str, namespace: str, model: str = None) -> bo
             self._propagate_ctx = None
             return False
 
-    def start_turn(self, turn_count: int, model: str, user_input: str | None = None) -> None:
+    def start_turn(self, model: str, user_input: str | None = None) -> None:
         """Start tracking a new turn as a top-level trace.
 
         Creates the turn generation as a TRACE (not an observation) so that each turn
         appears as a separate trace in Langfuse. Tools will be observations within the trace.
 
-        Note: Cannot use 'with' context managers due to async streaming architecture.
+        Prevents duplicate traces when SDK sends multiple AssistantMessages per turn during
+        streaming. Only the first AssistantMessage creates a trace; subsequent ones are ignored
+        until end_turn() clears the current trace.
+
+        Cannot use 'with' context managers due to async streaming architecture.
         Messages arrive asynchronously (AssistantMessage → ToolUseBlocks → ResultMessage)
         and the turn context must stay open across multiple async loop iterations.
 
         Args:
-            turn_count: Current turn number
             model: Model name (e.g., "claude-3-5-sonnet-20241022")
             user_input: Optional actual user input/prompt (if available)
         """
         if not self.langfuse_client:
             return
 
-        # Guard: Don't create a new turn if one is already active
-        # This prevents duplicate traces when AssistantMessage arrives multiple times
+        # Guard: Prevent creating duplicate traces for the same turn
+        # SDK sends multiple AssistantMessages during streaming - only create trace once
         if self._current_turn_generation:
-            logging.debug(f"Langfuse: Turn already active, skipping start_turn for turn {turn_count}")
+            logging.debug("Langfuse: Trace already active for current turn, skipping duplicate start_turn")
             return
 
         try:
-            # Build metadata
-            metadata = {"turn": turn_count}
-
             # Use pending initial prompt for turn 1 if available
-            if user_input is None and turn_count == 1 and self._pending_initial_prompt:
+            if user_input is None and self._pending_initial_prompt:
                 user_input = self._pending_initial_prompt
                 self._pending_initial_prompt = None  # Clear after use
-                logging.debug("Langfuse: Using pending initial prompt for turn 1")
+                logging.debug("Langfuse: Using pending initial prompt")
 
-            # Use actual user input if provided, otherwise use placeholder
+            # Use actual user input if provided, otherwise use generic placeholder
             if user_input:
                 input_content = [{"role": "user", "content": user_input}]
-                logging.info(f"Langfuse: Starting turn {turn_count} trace with model={model} and actual user input")
+                logging.info(f"Langfuse: Starting turn trace with model={model} and actual user input")
             else:
-                input_content = [{"role": "user", "content": f"Turn {turn_count}"}]
-                logging.info(f"Langfuse: Starting turn {turn_count} trace with model={model}")
+                input_content = [{"role": "user", "content": "User input"}]
+                logging.info(f"Langfuse: Starting turn trace with model={model}")
 
             # Create generation as a TRACE using start_as_current_observation()
-            # This makes claude_turn_X a top-level trace, not an observation
+            # Name doesn't include turn number - that will be added to metadata in end_turn()
+            # This makes the trace a top-level observation, not nested
             # Tools will automatically become child observations of this trace
             self._current_turn_ctx = self.langfuse_client.start_as_current_observation(
                 as_type="generation",
-                name=f"claude_turn_{turn_count}",
+                name="claude_interaction",  # Generic name, turn number added in metadata
                 input=input_content,
                 model=model,
-                metadata=metadata,
+                metadata={},  # Turn number will be added in end_turn()
             )
             self._current_turn_generation = self._current_turn_ctx.__enter__()
-            logging.debug(f"Langfuse: Turn {turn_count} trace started, ready for tool spans")
+            logging.info(f"Langfuse: Created new trace (model={model})")
 
         except Exception as e:
             logging.error(f"Langfuse: Failed to start turn: {e}", exc_info=True)
 
     def end_turn(self, turn_count: int, message: Any, usage: dict | None = None) -> None:
         """Complete turn tracking with output and usage data (called when ResultMessage arrives).
 
-        Updates the turn generation with the assistant's output and usage metrics, then closes it.
+        Updates the turn generation with the assistant's output, usage metrics, and SDK's
+        authoritative turn number in metadata, then closes it.
 
         Args:
-            turn_count: Current turn number
+            turn_count: Current turn number (from SDK's authoritative num_turns in ResultMessage)
             message: AssistantMessage from Claude SDK
             usage: Usage dict from ResultMessage with input_tokens, output_tokens, cache tokens, etc.
         """
@@ -291,8 +293,12 @@ def end_turn(self, turn_count: int, message: Any, usage: dict | None = None) ->
                 if cache_creation > 0:
                     usage_details_dict["cache_creation_input_tokens"] = cache_creation
 
-            # Update with output and usage_details (SDK v3 requires 'usage_details' parameter)
-            update_params = {"output": output_text}
+            # Update with output, usage_details, and turn number in metadata
+            # SDK v3 requires 'usage_details' parameter for usage tracking
+            update_params = {
+                "output": output_text,
+                "metadata": {"turn": turn_count}  # Add SDK's authoritative turn number
+            }
             if usage_details_dict:
                 update_params["usage_details"] = usage_details_dict
             self._current_turn_generation.update(**update_params)
@@ -310,7 +316,7 @@ def end_turn(self, turn_count: int, message: Any, usage: dict | None = None) ->
             if self.langfuse_client:
                 try:
                     self.langfuse_client.flush()
-                    logging.debug(f"Langfuse: Flushed turn {turn_count} data")
+                    logging.info(f"Langfuse: Flushed turn {turn_count} data")
                 except Exception as e:
                     logging.warning(f"Langfuse: Flush failed after turn {turn_count}: {e}")
 

components/runners/claude-code-runner/tests/test_duplicate_turn_prevention.py

@@ -0,0 +1,152 @@
+"""Unit tests for duplicate turn prevention in observability module."""
+
+import pytest
+from unittest.mock import Mock, patch, MagicMock
+from observability import ObservabilityManager
+
+
+class TestDuplicateTurnPrevention:
+    """Tests for preventing duplicate trace creation."""
+
+    @pytest.mark.asyncio
+    async def test_multiple_assistant_messages_same_turn_no_duplicates(self):
+        """Test that multiple AssistantMessages for the same turn don't create duplicate traces."""
+        manager = ObservabilityManager(
+            session_id="test-session", user_id="user-1", user_name="Test User"
+        )
+
+        # Mock Langfuse client
+        mock_client = Mock()
+        mock_ctx = Mock()
+        mock_generation = Mock()
+        mock_ctx.__enter__ = Mock(return_value=mock_generation)
+        mock_ctx.__exit__ = Mock()
+        mock_client.start_as_current_observation = Mock(return_value=mock_ctx)
+
+        manager.langfuse_client = mock_client
+
+        # Simulate first AssistantMessage - should create trace
+        manager.start_turn("claude-sonnet-4-5")
+        assert manager._current_turn_generation is not None
+        assert mock_client.start_as_current_observation.call_count == 1
+
+        # Simulate second AssistantMessage for SAME turn - should skip
+        manager.start_turn("claude-sonnet-4-5")
+        # Should still be 1 call (no new trace created)
+        assert mock_client.start_as_current_observation.call_count == 1
+
+        # Simulate third AssistantMessage for SAME turn - should skip
+        manager.start_turn("claude-sonnet-4-5")
+        # Should still be 1 call
+        assert mock_client.start_as_current_observation.call_count == 1
+
+    @pytest.mark.asyncio
+    async def test_sequential_turns_create_separate_traces(self):
+        """Test that sequential turns create separate traces."""
+        manager = ObservabilityManager(
+            session_id="test-session", user_id="user-1", user_name="Test User"
+        )
+
+        # Mock Langfuse client
+        mock_client = Mock()
+        mock_ctx = Mock()
+        mock_generation = Mock()
+        mock_ctx.__enter__ = Mock(return_value=mock_generation)
+        mock_ctx.__exit__ = Mock()
+        mock_client.start_as_current_observation = Mock(return_value=mock_ctx)
+
+        manager.langfuse_client = mock_client
+
+        # Turn 1
+        manager.start_turn("claude-sonnet-4-5")
+        assert manager._current_turn_generation is not None
+        assert mock_client.start_as_current_observation.call_count == 1
+
+        # End turn 1 (clear generation)
+        manager._current_turn_generation = None
+        manager._current_turn_ctx = None
+
+        # Turn 2 - should create new trace
+        manager.start_turn("claude-sonnet-4-5")
+        assert manager._current_turn_generation is not None
+        assert mock_client.start_as_current_observation.call_count == 2
+
+        # End turn 2
+        manager._current_turn_generation = None
+        manager._current_turn_ctx = None
+
+        # Turn 3 - should create new trace
+        manager.start_turn("claude-sonnet-4-5")
+        assert manager._current_turn_generation is not None
+        assert mock_client.start_as_current_observation.call_count == 3
+
+    @pytest.mark.asyncio
+    async def test_end_turn_adds_turn_number_to_metadata(self):
+        """Test that end_turn adds SDK's authoritative turn number to metadata."""
+        manager = ObservabilityManager(
+            session_id="test-session", user_id="user-1", user_name="Test User"
+        )
+
+        # Mock Langfuse client and generation
+        mock_client = Mock()
+        mock_generation = Mock()
+        mock_ctx = Mock()
+        mock_ctx.__exit__ = Mock()
+
+        manager.langfuse_client = mock_client
+        manager._current_turn_generation = mock_generation
+        manager._current_turn_ctx = mock_ctx
+
+        # Create mock AssistantMessage
+        mock_message = MagicMock()
+        mock_message.content = []
+
+        # End turn with SDK's turn number
+        manager.end_turn(5, mock_message, usage={"input_tokens": 100, "output_tokens": 50})
+
+        # Check that update was called with turn number in metadata
+        mock_generation.update.assert_called_once()
+        call_kwargs = mock_generation.update.call_args[1]
+        assert "metadata" in call_kwargs
+        assert call_kwargs["metadata"]["turn"] == 5
+
+    @pytest.mark.asyncio
+    async def test_no_prediction_just_sdk_turn_count(self):
+        """Test that we use SDK's authoritative turn count, not predictions."""
+        manager = ObservabilityManager(
+            session_id="test-session", user_id="user-1", user_name="Test User"
+        )
+
+        # Mock Langfuse client
+        mock_client = Mock()
+        mock_ctx = Mock()
+        mock_generation = Mock()
+        mock_ctx.__enter__ = Mock(return_value=mock_generation)
+        mock_ctx.__exit__ = Mock()
+        mock_client.start_as_current_observation = Mock(return_value=mock_ctx)
+        mock_client.flush = Mock()
+
+        manager.langfuse_client = mock_client
+
+        # Start turn without specifying turn number
+        manager.start_turn("claude-sonnet-4-5")
+        assert manager._current_turn_generation is not None
+        assert mock_client.start_as_current_observation.call_count == 1
+
+        # Second AssistantMessage arrives
+        manager.start_turn("claude-sonnet-4-5")
+        # Should be skipped - turn already active
+        assert mock_client.start_as_current_observation.call_count == 1
+
+        # SDK ResultMessage arrives with authoritative num_turns=2
+        mock_message = MagicMock()
+        mock_message.content = []
+
+        manager.end_turn(2, mock_message, usage={"input_tokens": 100, "output_tokens": 50})
+
+        # Check turn number was added to metadata
+        call_kwargs = mock_generation.update.call_args[1]
+        assert call_kwargs["metadata"]["turn"] == 2
+
+        # Should have called flush
+        assert mock_client.flush.call_count == 1