fix(loongsuite): isolate advice and stream failures

instrumentation-loongsuite/loongsuite-instrumentation-google-adk/CHANGELOG.md

@@ -12,6 +12,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Capture `gen_ai.skill.*` attributes on Google ADK SkillToolset
   `load_skill` and `load_skill_resource` execute-tool spans.
 
+### Changed
+
+- Align the supported Python and CI matrix with Google ADK's Python 3.10
+  runtime floor.
+
 ## Version 0.7.0 (2026-07-03)
 
 There are no changelog entries for this release.

instrumentation-loongsuite/loongsuite-instrumentation-google-adk/pyproject.toml

@@ -8,7 +8,7 @@ dynamic = ["version"]
 description = "LoongSuite instrumentation for Google Agent Development Kit (ADK)"
 readme = "README.md"
 license = "Apache-2.0"
-requires-python = ">=3.9"
+requires-python = ">=3.10"
 authors = [
   { name = "Huxing Zhang", email = "huxing.zhx@alibaba-inc.com" },
   { name = "Minghui Zhang", email = "zmh1625lumian@gmail.com" },
@@ -20,7 +20,6 @@ classifiers = [
   "License :: OSI Approved :: Apache Software License",
   "Programming Language :: Python",
   "Programming Language :: Python :: 3",
-  "Programming Language :: Python :: 3.9",
   "Programming Language :: Python :: 3.10",
   "Programming Language :: Python :: 3.11",
   "Programming Language :: Python :: 3.12",

instrumentation-loongsuite/loongsuite-instrumentation-litellm/CHANGELOG.md

@@ -7,6 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+### Fixed
+
+- Route streamed chunk accumulation and finalization callbacks through shared
+  fail-open advice while keeping stream iteration and lifecycle handling in the
+  LiteLLM wrapper.
+- Isolate prepare, response mapping, success, and error advice from application
+  completion calls, and detach streaming context before cross-task consumption.
+
 ## Version 0.7.0 (2026-07-03)
 
 There are no changelog entries for this release.

instrumentation-loongsuite/loongsuite-instrumentation-litellm/src/opentelemetry/instrumentation/litellm/_stream_wrapper.py

@@ -18,13 +18,15 @@
 
 import logging
 import timeit
+from threading import Lock
 from typing import Any, Iterator, Optional
 
 from opentelemetry.instrumentation.litellm._utils import (
     extract_litellm_text_parts,
     get_litellm_value,
     parse_tool_call_arguments,
 )
+from opentelemetry.util.genai import hook_advice
 from opentelemetry.util.genai.types import (
     OutputMessage,
     Reasoning,
@@ -35,6 +37,23 @@
 logger = logging.getLogger(__name__)
 
 
+@hook_advice("litellm", "stream_chunk")
+def _record_stream_chunk(accumulator: Any, chunk: Any) -> None:
+    """Record probe-owned chunk state without affecting stream delivery."""
+    accumulator.record_chunk(chunk)
+
+
+@hook_advice("litellm", "stream_finalize")
+def _invoke_stream_callback(
+    callback: Any,
+    span: Any,
+    last_chunk: Any,
+    error: Optional[BaseException],
+) -> None:
+    """Run the external finalization callback as fail-open probe advice."""
+    callback(span, last_chunk, error)
+
+
 class _StreamAccumulator:
     """Accumulate LiteLLM streaming deltas into GenAI output messages."""
 
@@ -200,7 +219,7 @@ class StreamWrapper:
 
     def __init__(
         self,
-        stream: Iterator,
+        stream: Iterator[Any],
         span: Any,
         callback: callable,
         invocation: Any = None,
@@ -212,6 +231,7 @@ def __init__(
         self.last_chunk = None  # Only keep last chunk to avoid memory leak
         self.chunk_count = 0
         self._finalized = False
+        self._finalize_lock = Lock()
 
     def __iter__(self):
         return self
@@ -220,7 +240,7 @@ def __next__(self):
         try:
             chunk = next(self.stream)
 
-            self._accumulator.record_chunk(chunk)
+            _record_stream_chunk(self._accumulator, chunk)
 
             # Only keep the last chunk (contains usage info)
             self.last_chunk = chunk
@@ -231,9 +251,13 @@ def __next__(self):
             # Stream ended normally, finalize span
             self._finalize()
             raise
-        except Exception as e:
+        except GeneratorExit:
+            # Generator close is an early, successful termination signal.
+            self._finalize()
+            raise
+        except BaseException as e:
             # Error during streaming
-            logger.debug(f"Error during streaming: {e}")
+            logger.debug("Error during streaming: %s", e, exc_info=True)
             self._finalize(error=e)
             raise
 
@@ -269,8 +293,8 @@ def __del__(self):
 
         try:
             self._finalize()
-        except Exception as exc:
-            logger.debug("Error finalizing unclosed LiteLLM stream: %s", exc)
+        except Exception:
+            pass
 
     def _close_stream(self) -> None:
         close = getattr(self.stream, "close", None)
@@ -280,27 +304,30 @@ def _close_stream(self) -> None:
         try:
             close()
         except Exception as exc:
-            logger.debug("Error closing LiteLLM stream: %s", exc)
+            logger.debug(
+                "Error closing LiteLLM stream: %s", exc, exc_info=True
+            )
 
-    def _finalize(self, error: Optional[Exception] = None):
+    def _finalize(self, error: Optional[BaseException] = None):
         """Finalize the span with data from last chunk."""
-        if self._finalized:
-            return
-
-        self._finalized = True
-        self._close_stream()
+        with self._finalize_lock:
+            if self._finalized:
+                return
+            self._finalized = True
         try:
-            # Call the callback with only the last chunk
-            # Note: The callback is responsible for calling handler.stop_llm()
-            # or handler.fail_llm().
-            # which will end the span. We no longer call span.end() here.
-            if self.callback:
-                self.callback(self.span, self.last_chunk, error)
-
-            # Clear reference to avoid holding memory
-            self.last_chunk = None
-        except Exception as e:
-            logger.debug(f"Error finalizing stream: {e}")
+            self._close_stream()
+        finally:
+            try:
+                # The callback is probe advice and must not affect stream cleanup.
+                if self.callback:
+                    _invoke_stream_callback(
+                        self.callback,
+                        self.span,
+                        self.last_chunk,
+                        error,
+                    )
+            finally:
+                self.last_chunk = None
 
     def get_output_messages(self) -> list[OutputMessage]:
         return self._accumulator.get_output_messages()
@@ -336,7 +363,9 @@ def __init__(
         self.last_chunk = None  # Only keep last chunk to avoid memory leak
         self.chunk_count = 0
         self._finalized = False
+        self._finalize_lock = Lock()
         self._stream_exhausted = False
+        self._stream_closed = False
 
     def __aiter__(self):
         # Return an async generator that wraps the stream and ensures finalization
@@ -353,7 +382,7 @@ async def _wrapped_iteration(self):
         error = None
         try:
             async for chunk in self.stream:
-                self._accumulator.record_chunk(chunk)
+                _record_stream_chunk(self._accumulator, chunk)
 
                 # Only keep the last chunk (contains usage info)
                 self.last_chunk = chunk
@@ -363,85 +392,117 @@ async def _wrapped_iteration(self):
 
             # Stream exhausted normally
             logger.debug(
-                f"AsyncStreamWrapper: Stream completed (chunks: {self.chunk_count})"
+                "AsyncStreamWrapper: Stream completed (chunks: %s)",
+                self.chunk_count,
             )
-        except Exception as e:
+        except GeneratorExit:
+            # ``aclose()`` injects GeneratorExit into this wrapper generator.
+            raise
+        except BaseException as e:
             # Error during streaming
-            logger.debug(f"AsyncStreamWrapper: Error during streaming: {e}")
+            logger.debug(
+                "AsyncStreamWrapper: Error during streaming: %s",
+                e,
+                exc_info=True,
+            )
             error = e
             raise
         finally:
             # Always finalize, whether completed normally, with error, or closed early
-            await self._aclose_stream()
-            self._finalize(error=error)
+            try:
+                await self._aclose_stream()
+            finally:
+                self._finalize(error=error)
 
     async def __aenter__(self):
         """Support async context manager protocol."""
         return self
 
     async def __aexit__(self, exc_type, exc_val, exc_tb):
         """Ensure finalization on async context exit."""
-        await self._aclose_stream()
-        if exc_type is not None:
-            # Exception occurred during iteration
-            self._finalize(error=exc_val)
-        else:
-            # Normal exit (may have completed or early terminated)
-            self._finalize()
+        try:
+            await self._aclose_stream()
+        finally:
+            if exc_type is not None:
+                # Exception occurred during iteration
+                self._finalize(error=exc_val)
+            else:
+                # Normal exit (may have completed or early terminated)
+                self._finalize()
         return False
 
     async def aclose(self):
         """Explicitly close and finalize the async stream."""
-        await self._aclose_stream()
-        self._finalize()
+        try:
+            await self._aclose_stream()
+        finally:
+            self._finalize()
 
     def close(self):
         """Synchronous close method for compatibility."""
-        self._close_stream()
-        self._finalize()
+        try:
+            self._close_stream()
+        finally:
+            self._finalize()
 
     def _close_stream(self) -> None:
+        if self._stream_closed:
+            return
+
+        self._stream_closed = self._close_sync_stream()
+
+    def _close_sync_stream(self) -> bool:
         close = getattr(self.stream, "close", None)
         if not callable(close):
-            return
+            return False
 
         try:
             close()
         except Exception as exc:
-            logger.debug("Error closing LiteLLM async stream: %s", exc)
+            logger.debug(
+                "Error closing LiteLLM async stream: %s",
+                exc,
+                exc_info=True,
+            )
+            return False
+        return True
 
     async def _aclose_stream(self) -> None:
+        if self._stream_closed:
+            return
+
         aclose = getattr(self.stream, "aclose", None)
         if callable(aclose):
             try:
                 await aclose()
-                return
             except Exception as exc:
-                logger.debug("Error closing LiteLLM async stream: %s", exc)
+                logger.debug(
+                    "Error closing LiteLLM async stream: %s",
+                    exc,
+                    exc_info=True,
+                )
+            else:
+                self._stream_closed = True
+                return
 
-        self._close_stream()
+        self._stream_closed = self._close_sync_stream()
 
-    def _finalize(self, error: Optional[Exception] = None):
+    def _finalize(self, error: Optional[BaseException] = None):
         """Finalize the span with data from last chunk."""
-        if self._finalized:
-            return
-
-        self._finalized = True
+        with self._finalize_lock:
+            if self._finalized:
+                return
+            self._finalized = True
         try:
-            # Call the callback with only the last chunk
-            # Note: The callback is responsible for calling handler.stop_llm()
-            # or handler.fail_llm().
-            # which will end the span. We no longer call span.end() here.
             if self.callback:
-                try:
-                    self.callback(self.span, self.last_chunk, error)
-                except Exception as callback_error:
-                    logger.debug(f"Error in stream callback: {callback_error}")
-
-            # Clear reference to avoid holding memory
+                _invoke_stream_callback(
+                    self.callback,
+                    self.span,
+                    self.last_chunk,
+                    error,
+                )
+        finally:
             self.last_chunk = None
-        except Exception as e:
-            logger.debug(f"Error finalizing async stream: {e}")
 
     def get_output_messages(self) -> list[OutputMessage]:
         return self._accumulator.get_output_messages()