refactor(llmobs): abstract llmobs from llama_index integration code

.claude/skills/apm-integrations/references/implementation-guide.md

@@ -65,6 +65,8 @@ it's patched. **Without this entry, integration config settings are silently ign
 Create `ddtrace/llmobs/_integrations/{name}.py` subclassing `BaseLLMIntegration`.
 Read `ddtrace/llmobs/_integrations/anthropic.py` for the canonical pattern.
 Register in `ddtrace/llmobs/_integrations/__init__.py` (import + `__all__`).
+Connect it to the contrib with `LlmEvents` subscribers in `ddtrace/llmobs/_contrib/{name}/`
+(see `ddtrace/llmobs/_contrib/anthropic/`) so the contrib does not import `ddtrace.llmobs`.
 
 See the **llmobs-integrations** skill for the full LLM-specific implementation guide.
 

.claude/skills/llmobs-integrations/SKILL.md

@@ -19,22 +19,24 @@ LLMObs integrations enable Datadog LLM Observability for AI/LLM libraries. They
 
 LLMObs integrations consist of two cooperating layers:
 
-1. **Patch Layer** (`ddtrace/contrib/internal/{name}/patch.py`) -- wraps library functions. Standard request/response LLM integrations construct `LlmRequestEvent` and use `core.context_with_event()` so the LLM tracing subscriber owns span lifecycle and LLMObs tag extraction.
+1. **Patch Layer** (`ddtrace/contrib/internal/{name}/patch.py`) -- wraps library functions. Standard request/response LLM integrations construct `LlmRequestEvent` and use `core.context_with_event()` so the LLM tracing subscriber owns span lifecycle. The patch layer must not import from `ddtrace.llmobs`.
 2. **Integration Layer** (`ddtrace/llmobs/_integrations/{name}.py`) -- extends `BaseLLMIntegration`, implements `_set_base_span_tags()` and `_llmobs_set_tags()` to extract and set provider-specific messages, tools, metadata, and token metrics.
+3. **LLMObs Subscribers** (`ddtrace/llmobs/_contrib/{name}/`) -- subscribe to the `LlmEvents` span lifecycle events (`SPAN_STARTING`, `SPAN_STARTED`, `SPAN_FINISHING`) that `LlmTracingSubscriber` dispatches, filter on `ctx.event.component`, and call the integration layer. `listen_integrations()` in `ddtrace/llmobs/_contrib/__init__.py` registers them when the library is patched.
 
-Both layers must work together. The patch layer identifies the operation and passes request/response data through the event; the integration layer controls what data is extracted.
+The layers must work together. The patch layer identifies the operation and passes request/response data through the event; the subscribers connect the event to the integration layer, which controls what data is extracted.
 
 ## Active Patch Patterns
 
-- **Event-based request spans**: Use `LlmRequestEvent` with `core.context_with_event()` for new standard request/response LLM integrations. Anthropic is the canonical reference. This is the preferred pattern.
+- **Event-based request spans with LLMObs subscribers**: Use `LlmRequestEvent` with `core.context_with_event()`, leave `llmobs_integration` unset, and add subscribers under `ddtrace/llmobs/_contrib/{name}/`. Anthropic is the canonical reference. This is the preferred pattern.
+- **Event-based request spans with `llmobs_integration`**: Older event-based integrations pass `llmobs_integration=integration` on the event, and `LlmTracingSubscriber` calls the integration directly. This makes the contrib import LLMObs code; do not use it for new work.
 - **Direct integration spans**: Some existing or specialized integrations still call `integration.trace()` and `integration.llmobs_set_tags()` directly, especially for child spans, agent/tool spans, or integrations not yet migrated. Google GenAI, OpenAI tool spans, and Claude Agent SDK are useful references.
 
 ## Key Files
 
 | Purpose | File |
 |---------|------|
 | Base LLM integration class | `ddtrace/llmobs/_integrations/base.py` (`BaseLLMIntegration`) |
-| Stream handler base classes | `ddtrace/llmobs/_integrations/base_stream_handler.py` (`BaseStreamHandler`, `StreamHandler`, `AsyncStreamHandler`) |
+| Stream handler base classes | `ddtrace/contrib/internal/stream_handler.py` (`BaseStreamHandler`, `StreamHandler`, `AsyncStreamHandler`) |
 | Shared utilities | `ddtrace/llmobs/_integrations/utils.py` |
 | LLMObs annotation helper | `ddtrace/llmobs/_utils.py` (`_annotate_llmobs_span_data`) |
 | LLMObs constants | `ddtrace/llmobs/_constants.py` |
@@ -116,7 +118,8 @@ Note two already-shipped integrations predate this key: bedrock and the claude-a
 - **Streaming** must use `BaseStreamHandler`/`AsyncStreamHandler` -- never consume streams directly
 - **Event-based patch wrappers** should not call `span.set_exc_info()`, `span.finish()`, or `integration.llmobs_set_tags()` directly; the tracing subscriber handles that when the event ends
 - **Direct integration spans** must keep `integration.llmobs_set_tags()` and span lifecycle handling aligned with the closest current reference
-- **Integration instance** must be stored on the module: `module._datadog_integration = MyLibIntegration(integration_config=config.mylib)`
+- **Integration instance**: subscriber-based integrations build it lazily in the LLMObs subscriber (see `ddtrace/llmobs/_contrib/anthropic/subscribers.py`); older integrations store it on the module as `module._datadog_integration = MyLibIntegration(integration_config=config.mylib)`
+- **Subscriber registration**: subscriber-based integrations hook `{name}.patch`/`{name}.unpatch` core events in `listen_integrations()` and stay registered while LLMObs is disabled, because they also set the APM shadow tags
 
 ## Message Types
 

.claude/skills/llmobs-integrations/references/failure-modes.md

@@ -14,11 +14,11 @@ Comprehensive debugging guide for all known LLMObs integration failure modes. Ea
 **Causes:**
 1. `submit_to_llmobs=True` not set on `LlmRequestEvent` for event-based patch code
 2. `ctx.dispatch_ended_event()` not called, so `LlmTracingSubscriber` never calls `integration.llmobs_set_tags()`
-3. Integration not instantiated -- `module._datadog_integration` is `None`
+3. Integration not instantiated -- `module._datadog_integration` is `None`, or (for subscriber-based integrations) the `ddtrace/llmobs/_contrib/{name}` subscribers were never registered because `listen_integrations()` did not run before `patch()`
 4. `llmobs_enabled` returns `False` -- LLMObs not configured in tracer config
 
 **Fix:**
-- Verify `LlmRequestEvent(..., submit_to_llmobs=True, llmobs_integration=integration, request_kwargs=kwargs, ...)` in event-based patch wrappers
+- Verify `LlmRequestEvent(..., submit_to_llmobs=True, request_kwargs=kwargs, ...)` in event-based patch wrappers, plus either registered LLMObs subscribers or `llmobs_integration=integration` for older integrations
 - Verify success and error paths call `ctx.dispatch_ended_event(...)`
 - Verify `patch()` stores integration: `module._datadog_integration = MyIntegration(integration_config=config.mylib)`
 - Check `DD_LLMOBS_ENABLED=1` is set

.claude/skills/llmobs-integrations/references/implementation-guide.md

@@ -5,8 +5,9 @@ Follow the **apm-integrations** skill's [Implementation Guide](../../apm-integra
 ## Design: Two-Layer Architecture
 
 LLM integrations use `BaseLLMIntegration` as a second layer on top of a standard APM integration:
-- Patch code creates a subclass instance and stores it on the module: `module._datadog_integration = MyLibIntegration(integration_config=config.mylib)`
-- Standard request/response patch code constructs `LlmRequestEvent` and uses `core.context_with_event()`; `LlmTracingSubscriber` manages span lifecycle and calls `integration.llmobs_set_tags()` when the event ends
+- Standard request/response patch code constructs `LlmRequestEvent` and uses `core.context_with_event()`; `LlmTracingSubscriber` manages span lifecycle and dispatches `LlmEvents.SPAN_STARTING`, `SPAN_STARTED`, and `SPAN_FINISHING` with the execution context
+- LLMObs subscribers in `ddtrace/llmobs/_contrib/{name}/` listen to those events, build the `BaseLLMIntegration` subclass lazily, and call it to set the span type, base tags, and LLMObs tags, so the contrib never imports `ddtrace.llmobs`
+- Older integrations instead store the instance on the module (`module._datadog_integration = ...`) and pass it as `LlmRequestEvent(llmobs_integration=...)`; `LlmTracingSubscriber` then calls it directly
 - The `BaseLLMIntegration` subclass in `ddtrace/llmobs/_integrations/` handles provider-specific message, token, metadata, and tool extraction
 - Some existing or specialized integrations still call `integration.trace()` directly for direct child spans; follow the closest current reference before using that pattern
 
@@ -16,6 +17,7 @@ This separation keeps APM patching decoupled from LLMObs data extraction.
 
 An LLM integration is an APM integration with an extra layer. You do everything in the apm-integrations guide, but:
 - **Step 1 (patch module)**: Use `LlmRequestEvent` with `core.context_with_event()` for standard request/response LLM integrations
+- **Step 1b (LLMObs subscribers)**: Add `ddtrace/llmobs/_contrib/{name}/` subscribers and register them from `listen_integrations()` in `ddtrace/llmobs/_contrib/__init__.py`
 - **Step 3 (LLMObs integration)**: Create the `BaseLLMIntegration` subclass that handles provider-specific message, tool, and token extraction (this guide)
 - **Step 4 (test environment)**: Use `tests/llmobs/suitespec.yml`; add `vcrpy` only when the suite uses vcrpy cassettes and follow nearby version pins
 - **Step 5 (tests)**: Add `test_{name}_llmobs.py` in addition to the APM `test_{name}.py`, using the right transport pattern for the integration and `assert_llmobs_span_data(_get_llmobs_data_metastruct(span), ...)`
@@ -90,22 +92,33 @@ Register in `ddtrace/llmobs/_integrations/__init__.py` (import + `__all__` entry
 
 ## Step 1 Expanded: Patch Layer (`LlmRequestEvent`)
 
-Standard LLM integrations should use `LlmRequestEvent` from `ddtrace/contrib/_events/llm.py` with `core.context_with_event()`. Read `ddtrace/contrib/internal/anthropic/patch.py` for the current event-based pattern. The patch layer constructs the event, stores the response on `event.response`, and calls `ctx.dispatch_ended_event()`; `ddtrace/_trace/subscribers/llm.py` handles span creation, base tags, LLMObs extraction, errors, and span finish under the hood.
+Standard LLM integrations should use `LlmRequestEvent` from `ddtrace/contrib/_events/llm.py` with `core.context_with_event()`. Read `ddtrace/contrib/internal/anthropic/patch.py` for the current event-based pattern. The patch layer constructs the event, stores the response on `event.response`, and calls `ctx.dispatch_ended_event()`; `ddtrace/_trace/subscribers/llm.py` handles span creation, errors, and span finish, and dispatches `LlmEvents` so LLMObs subscribers can add base tags and LLMObs data.
 
 Key points:
-- Construct `LlmRequestEvent(..., llmobs_integration=integration, submit_to_llmobs=True, request_kwargs=kwargs, ...)`
+- Construct `LlmRequestEvent(..., submit_to_llmobs=True, request_kwargs=kwargs, ...)` and leave `llmobs_integration` unset
+- Set APM tags the contrib owns (for example `{name}.request.model`) through the event's `tags`, not in the `BaseLLMIntegration`
+- Shared stream helpers live in `ddtrace/contrib/internal/stream_handler.py`; pass `None` as the integration when the contrib has none
 - The event/subscriber path owns span creation and finishing; patch wrappers should not call `tracer.trace()`, `integration.trace()`, or create spans directly for standard request spans
 - Use `with core.context_with_event(event, dispatch_end_event=False) as ctx:` when streaming or when the wrapper needs to dispatch the ended event manually
 - For non-streaming success, set `event.response = resp` and call `ctx.dispatch_ended_event()`
 - For errors, call `ctx.dispatch_ended_event(*sys.exc_info())` and re-raise; do not call `span.set_exc_info()` or `span.finish()` directly in the patch wrapper
-- LLMObs tag setting is handled by `LlmTracingSubscriber`, not directly in the patch wrapper
+- LLMObs tag setting is handled by the LLMObs subscribers, not directly in the patch wrapper
+
+## Step 1b Expanded: LLMObs Subscribers
+
+Read `ddtrace/llmobs/_contrib/anthropic/` for the pattern. `LlmTracingSubscriber` dispatches three events with the `ExecutionContext`. They are shared by every LLM integration, so each handler returns early unless `ctx.event.component` matches:
+- `LlmEvents.SPAN_STARTING` runs before the span exists. This is the only point where `event.span_type` can still be set to `SpanTypes.LLM`, which `LLMObs._on_span_start` needs at creation.
+- `LlmEvents.SPAN_STARTED` runs after the span is created. Use it for `_set_base_span_tags()`, `_annotate_integration_tag()`, and `_stamp_llmobs_span_kind_at_start()`.
+- `LlmEvents.SPAN_FINISHING` runs before the span is finished. Call `integration.llmobs_set_tags()` here.
+
+Subscribers set `auto_register = False`. Register them from `listen_integrations()` in `ddtrace/llmobs/_contrib/__init__.py` on the `{name}.patch` core event, and unregister them on `{name}.unpatch`. `listen_integrations()` runs from both the product's `post_preload` and `LLMObs.enable()`. Test fixtures that call `patch()` directly should call `listen_integrations()` first.
 - The async variant is identical but uses `async def` / `await`
 
 Some older or specialized integrations still call `integration.trace()` and `integration.llmobs_set_tags()` directly. Use that pattern when modifying an existing integration that already does so, when the closest current reference uses it (for example Google GenAI), or when the behavior requires direct child spans (for example OpenAI MCP tool spans or agent/tool child spans).
 
 ## Streaming
 
-Subclass `StreamHandler`/`AsyncStreamHandler` from `ddtrace/llmobs/_integrations/base_stream_handler.py`:
+Subclass `StreamHandler`/`AsyncStreamHandler` from `ddtrace/contrib/internal/stream_handler.py`:
 
 - `initialize_chunk_storage()` — set up accumulators for content, usage, role
 - `process_chunk(chunk)` — accumulate text, tool blocks, usage from each chunk
@@ -212,7 +225,8 @@ In addition to the full checklist in the apm-integrations [Implementation Guide]
 
 - [ ] `ddtrace/llmobs/_integrations/{name}.py` — `BaseLLMIntegration` subclass
 - [ ] `ddtrace/llmobs/_integrations/__init__.py` — import + `__all__` entry
-- [ ] `ddtrace/contrib/internal/{name}/patch.py` — uses `LlmRequestEvent` + `core.context_with_event()` for standard LLM request spans (see anthropic for pattern)
+- [ ] `ddtrace/contrib/internal/{name}/patch.py` — uses `LlmRequestEvent` + `core.context_with_event()` for standard LLM request spans, with no `ddtrace.llmobs` imports (see anthropic for pattern)
+- [ ] `ddtrace/llmobs/_contrib/{name}/` — `LlmEvents` subscribers, registered from `listen_integrations()` in `ddtrace/llmobs/_contrib/__init__.py`
 - [ ] `tests/llmobs/suitespec.yml` — LLMObs test suite entry
 - [ ] Test dependencies match the suite style; include `vcrpy` only when cassette replay is used
 - [ ] `docs/index.rst` — add integration to the docs index

ddtrace/_trace/subscribers/llm.py

@@ -3,6 +3,7 @@
 
 from ddtrace._trace.subscribers._base import TracingSubscriber
 from ddtrace.constants import SPAN_KIND
+from ddtrace.contrib._events.llm import LlmEvents
 from ddtrace.contrib._events.llm import LlmRequestEvent
 from ddtrace.internal import core
 from ddtrace.internal.constants import COMPONENT
@@ -23,11 +24,17 @@ class LlmTracingSubscriber(TracingSubscriber["LlmRequestEvent"]):
 
     Handles span creation, base tag setting, proxy detection,
     and LLMObs tag extraction. Provider-specific logic is delegated
-    to the integration object carried by the event.
+    to the integration object carried by the event, or, for events that
+    carry none, to subscribers of the LlmEvents span lifecycle events.
     """
 
     event_names = (LlmRequestEvent.event_name,)
 
+    @classmethod
+    def _on_context_started(cls, ctx: core.ExecutionContext["LlmRequestEvent"]) -> None:
+        core.dispatch(LlmEvents.SPAN_STARTING.value, (ctx,))
+        super()._on_context_started(ctx)
+
     @classmethod
     def on_started(cls, ctx: core.ExecutionContext["LlmRequestEvent"]) -> None:
         event: LlmRequestEvent = ctx.event
@@ -40,6 +47,10 @@ def on_started(cls, ctx: core.ExecutionContext["LlmRequestEvent"]) -> None:
         span._remove_attribute(COMPONENT)
         span._remove_attribute(SPAN_KIND)
 
+        if event.llmobs_integration is None:
+            core.dispatch(LlmEvents.SPAN_STARTED.value, (ctx,))
+            return
+
         event.llmobs_integration._set_base_span_tags(
             span,
             model=event.model,
@@ -67,6 +78,9 @@ def on_ended(
         dispatch_ended_event().
         """
         event: LlmRequestEvent = ctx.event
+        if event.llmobs_integration is None:
+            core.dispatch(LlmEvents.SPAN_FINISHING.value, (ctx,))
+            return
         event.llmobs_integration.llmobs_set_tags(
             span_from_context(ctx),
             args=[],

ddtrace/aiguard/_streaming.py

@@ -22,10 +22,10 @@
 import wrapt
 
 from ddtrace.aiguard._context import is_aiguard_context_active
+from ddtrace.contrib.internal.stream_handler import AsyncStreamHandler
+from ddtrace.contrib.internal.stream_handler import BaseStreamHandler
 import ddtrace.internal.logger as ddlogger
 from ddtrace.internal.settings.aiguard import aiguard_config
-from ddtrace.llmobs._integrations.base_stream_handler import AsyncStreamHandler
-from ddtrace.llmobs._integrations.base_stream_handler import BaseStreamHandler
 
 
 logger = ddlogger.get_logger(__name__)
@@ -146,7 +146,7 @@ def __next__(self) -> Any:
     # ------------------------------------------------------------------
     # Context-manager protocol
     #
-    # TracedStream.__enter__() (base_stream_handler.py) has two branches:
+    # TracedStream.__enter__() (ddtrace/contrib/internal/stream_handler.py) has two branches:
     #   - non-manager (raw Stream): returns ``self`` (the TracedStream).
     #   - manager (MessageStreamManager): returns a NEW TracedStream
     #     wrapping the inner MessageStream.

ddtrace/contrib/_events/llm.py

@@ -2,6 +2,7 @@
 
 from dataclasses import dataclass
 from dataclasses import field
+from enum import Enum
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import Optional
@@ -16,6 +17,18 @@
     from ddtrace.llmobs._integrations.base import BaseLLMIntegration
 
 
+class LlmEvents(Enum):
+    LLM_REQUEST = "llm.request"
+    # Dispatched by LlmTracingSubscriber with the ExecutionContext so products can
+    # take part in the span lifecycle without the contrib importing them:
+    # SPAN_STARTING fires before the span is created (the only point where
+    # event.span_type can still change), SPAN_STARTED right after, and
+    # SPAN_FINISHING before the span is finished.
+    SPAN_STARTING = "llm.request.span_starting"
+    SPAN_STARTED = "llm.request.span_started"
+    SPAN_FINISHING = "llm.request.span_finishing"
+
+
 @dataclass
 class LlmRequestEvent(TracingEvent):
     """LLM request event for all LLM integrations.
@@ -25,12 +38,13 @@ class LlmRequestEvent(TracingEvent):
     (_set_base_span_tags, llmobs_set_tags).
     """
 
-    event_name = "llm.request"
+    event_name = LlmEvents.LLM_REQUEST.value
     span_kind = SpanKind.CLIENT
 
     provider: str = event_field()
     model: Optional[str] = event_field(default=None)
-    llmobs_integration: BaseLLMIntegration = event_field()
+    # Integrations that have moved to LlmEvents subscribers leave this unset.
+    llmobs_integration: Optional[BaseLLMIntegration] = event_field(default=None)
     request_kwargs: dict[str, Any] = event_field(default_factory=dict)
     submit_to_llmobs: bool = event_field(default=False)
     instance: Optional[Any] = event_field(default=None)
@@ -43,4 +57,10 @@ class LlmRequestEvent(TracingEvent):
 
     def __post_init__(self) -> None:
         self.operation_name = f"{self.component}.request"
-        self.span_type = SpanTypes.LLM if (self.submit_to_llmobs and self.llmobs_integration.llmobs_enabled) else None
+        self.span_type = (
+            SpanTypes.LLM
+            if (
+                self.submit_to_llmobs and self.llmobs_integration is not None and self.llmobs_integration.llmobs_enabled
+            )
+            else None
+        )