fix legacy

.github/workflows/tests-full.yml

@@ -45,6 +45,12 @@ jobs:
             pytest-mark: 'agentops'  # including agentops+litellm tests here
             runs-on: [self-hosted, 1ES.Pool=agl-runner-gpu]
             has-gpu: true
+          # Similar for Weave.
+          - id: weave
+            display-name: Weave
+            pytest-mark: 'weave'
+            runs-on: ubuntu-latest  # No GPU tests for Weave.
+            has-gpu: false
           # Other tests that require GPU
           - id: gpu
             display-name: GPU required
@@ -54,7 +60,7 @@ jobs:
           # Other uncovered tests
           - id: others
             display-name: Others
-            pytest-mark: 'not store and not agentops and not gpu and not llmproxy'
+            pytest-mark: 'not store and not agentops and not weave and not gpu and not llmproxy'
             runs-on: ubuntu-latest
             has-gpu: false
         env:
@@ -83,24 +89,24 @@ jobs:
 
       - name: Sync dependencies (latest, gpu)
         if: matrix.env.setup-script == 'latest' && matrix.mark.has-gpu
-        run: uv sync --frozen --no-default-groups --extra apo --extra mongo --group dev --group agents --group langchain --group torch-gpu-stable
+        run: uv sync --frozen --no-default-groups --extra apo --extra weave --extra mongo --group dev --group agents --group langchain --group torch-gpu-stable
       # Don't install vllm/pytorch on CPU counterparts
       - name: Sync dependencies (latest, cpu)
         if: matrix.env.setup-script == 'latest' && !matrix.mark.has-gpu
-        run: uv sync --frozen --no-default-groups --extra apo --extra mongo --group dev --group agents --group langchain --group core-stable
+        run: uv sync --frozen --no-default-groups --extra apo --extra weave --extra mongo --group dev --group agents --group langchain --group core-stable
       - name: Sync dependencies (stable, gpu)
         if: matrix.env.setup-script == 'stable' && matrix.mark.has-gpu
-        run: uv sync --frozen --no-default-groups --extra apo --extra mongo --group dev --group agents --group langchain --group torch-gpu-${{ matrix.env.setup-script }}
+        run: uv sync --frozen --no-default-groups --extra apo --extra weave --extra mongo --group dev --group agents --group langchain --group torch-gpu-${{ matrix.env.setup-script }}
       - name: Sync dependencies (stable, cpu)
         if: matrix.env.setup-script == 'stable' && !matrix.mark.has-gpu
-        run: uv sync --frozen --no-default-groups --extra apo --extra mongo --group dev --group agents --group langchain --group core-stable
+        run: uv sync --frozen --no-default-groups --extra apo --extra weave --extra mongo --group dev --group agents --group langchain --group core-stable
       # Don't install langchain for legacy dependency because it has conflicts with torch.
       - name: Sync dependencies (legacy, gpu)
         if: matrix.env.setup-script == 'legacy' && matrix.mark.has-gpu
-        run: uv sync --frozen --no-default-groups --extra apo --extra mongo --group dev --group agents --group torch-gpu-legacy
+        run: uv sync --frozen --no-default-groups --extra apo --extra weave --extra mongo --group dev --group agents --group torch-gpu-legacy
       - name: Sync dependencies (legacy, cpu)
         if: matrix.env.setup-script == 'legacy' && !matrix.mark.has-gpu
-        run: uv sync --frozen --no-default-groups --extra apo --extra mongo --group dev --group agents --group core-legacy
+        run: uv sync --frozen --no-default-groups --extra apo --extra weave --extra mongo --group dev --group agents --group core-legacy
 
       - name: Freeze dependencies
         run: |
@@ -270,6 +276,14 @@ jobs:
           python write_traces.py agentops
           sleep 5
 
+      - name: Write Traces with Operations
+        run: |
+          set -euo pipefail
+          source .venv/bin/activate
+          cd examples/minimal
+          python write_traces.py operation
+          sleep 5
+
       - name: Write Traces via Otel Tracer with Client
         run: |
           set -euo pipefail

.github/workflows/tests.yml

@@ -40,6 +40,7 @@ jobs:
         run: |
           uv sync --frozen \
             --extra apo \
+            --extra weave \
             --extra verl \
             --extra mongo \
             --group dev \
@@ -110,6 +111,10 @@ jobs:
       - name: Set source commit for docs
         run: |
           echo "SOURCE_COMMIT=${{ github.sha }}" >> $GITHUB_ENV
+      - name: Verify OpenAPI specification is up-to-date
+        run: |
+          uv run --locked --no-sync python scripts/export_openapi.py
+          git diff --exit-code docs/assets/store-openapi.json
       - name: Build documentation
         run: uv run --locked --no-sync mkdocs build --strict
       - name: Upload docs artifact
@@ -131,6 +136,10 @@ jobs:
           - id: agentops
             display-name: AgentOps
             pytest-mark: 'agentops'
+          # Similar for Weave.
+          - id: weave
+            display-name: Weave
+            pytest-mark: 'weave'
           # litellm proxy tests are slow
           - id: llmproxy
             display-name: LLM proxy
@@ -142,7 +151,7 @@ jobs:
           # unmarked tests: adapter, execution engine, etc.
           - id: others
             display-name: Others
-            pytest-mark: 'not store and not agentops and not llmproxy and not utils'
+            pytest-mark: 'not store and not agentops and not weave and not llmproxy and not utils'
         env:
           - python-version: '3.10'
             setup-script: 'legacy'
@@ -167,10 +176,10 @@ jobs:
         run: uv lock --upgrade
         if: matrix.env.setup-script == 'latest'
       - name: Sync dependencies (latest)
-        run: uv sync --frozen --no-default-groups --extra apo --group dev --group agents --group langchain --group core-stable
+        run: uv sync --frozen --no-default-groups --extra apo --extra weave --group dev --group agents --group langchain --group core-stable
         if: matrix.env.setup-script == 'latest'
       - name: Sync dependencies (stable & legacy)
-        run: uv sync --frozen --no-default-groups --extra apo --group dev --group agents --group langchain --group core-${{ matrix.env.setup-script }}
+        run: uv sync --frozen --no-default-groups --extra apo --extra weave --group dev --group agents --group langchain --group core-${{ matrix.env.setup-script }}
         if: matrix.env.setup-script != 'latest'
       - name: Freeze dependencies
         run: |

.gitignore

@@ -1,7 +1,7 @@
 # Agentlightning specific files
 verl_old
 meta-llama/**
-debug/*.png
+**/debug/*.png
 requirements-freeze*.txt
 /playground
 

.pre-commit-config.yaml

@@ -3,13 +3,14 @@ repos:
     rev: v6.0.0
     hooks:
       - id: end-of-file-fixer
+        exclude: (.*store-openapi\.json$)
       - id: trailing-whitespace
       - id: check-yaml
         exclude: ^mkdocs\.yml$
       - id: check-toml
       - id: check-added-large-files
         args: ["--maxkb=1024"]
-        exclude: (^uv\.lock$)|(^docs/assets/.*\.svg$)
+        exclude: (^uv\.lock$)|(^docs/assets/.*\.svg$)|(.*store-openapi\.json$)
       - id: check-shebang-scripts-are-executable
       - id: detect-private-key
   - repo: https://github.com/pycqa/isort

agentlightning/emitter/__init__.py

@@ -1,5 +1,16 @@
 # Copyright (c) Microsoft. All rights reserved.
 
+"""Convenient helpers for creating spans / traces.
+
+All emitters operate in two modes, switchable via the `propagate` parameter.
+The emitters first [`SpanCreationRequest`][agentlightning.SpanCreationRequest] object, then:
+
+1. When `propagate` is True, this creation request will be propagated to the active tracer
+   and a [`Span`][agentlightning.Span] instance will be created (possibly deferred).
+2. When `propagate` is False, the creation request will be returned directly. Useful for cases
+   when you don't have a tracer but you want to create a creation request for later use.
+"""
+
 from .annotation import emit_annotation, operation
 from .exception import emit_exception
 from .message import emit_message, get_message_value

agentlightning/emitter/annotation.py

@@ -5,7 +5,6 @@
 import asyncio
 import functools
 import inspect
-import json
 import logging
 from types import TracebackType
 from typing import (
@@ -22,19 +21,18 @@ from typing import (
     overload,
 )
 
-from opentelemetry import trace
-from opentelemetry.sdk.trace import ReadableSpan
-from opentelemetry.trace import Status, StatusCode
-
 from agentlightning.semconv import AGL_ANNOTATION, AGL_OPERATION, LightningSpanAttributes
-from agentlightning.utils.otel import flatten_attributes, get_tracer
+from agentlightning.tracer.base import get_active_tracer
+from agentlightning.tracer.dummy import DummyTracer
+from agentlightning.types import SpanCoreFields, SpanRecordingContext, TraceStatus
+from agentlightning.utils.otel import check_attributes_sanity, flatten_attributes, sanitize_attributes
 
 _FnType = TypeVar("_FnType", bound=Callable[..., Any])
 
 logger = logging.getLogger(__name__)
 
 
-def emit_annotation(annotation: Dict[str, Any], propagate: bool = True) -> ReadableSpan:
+def emit_annotation(annotation: Dict[str, Any], propagate: bool = True) -> SpanCoreFields:
     """Emit a new annotation span.
 
     This is the underlying implementation of [`emit_reward`][agentlightning.emit_reward].
@@ -48,62 +46,46 @@ def emit_annotation(annotation: Dict[str, Any], propagate: bool = True) -> Reada
     Args:
         annotation: Dictionary containing annotation key-value pairs.
             Representatives are rewards, tags, and metadata.
-        propagate: Whether to propagate the span to exporters automatically.
+        propagate: Whether to propagate the span to tracers automatically.
     """
-    annotation_attributes = flatten_attributes(annotation)
-    if any(not isinstance(v, (str, int, float, bool, bytes)) for v in annotation_attributes.values()):
-        raise TypeError("All annotation attributes must be primitive types (str, int, float, bool, bytes)")
+    annotation_attributes = flatten_attributes(annotation, expand_leaf_lists=False)
+    check_attributes_sanity(annotation_attributes)
+    sanitized_attributes = sanitize_attributes(annotation_attributes)
+    logger.debug("Emitting annotation span with keys %s", sanitized_attributes.keys())
 
-    # TODO: this should use a tracer from current context rather than the singleton
-    tracer = get_tracer(use_active_span_processor=propagate)
-    span = tracer.start_span(
-        AGL_ANNOTATION,
-        attributes=annotation_attributes,
+    if propagate:
+        tracer = get_active_tracer()
+        if tracer is None:
+            raise RuntimeError("No active tracer found. Cannot emit annotation span.")
+    else:
+        tracer = DummyTracer()
+
+    return tracer.create_span(
+        name=AGL_ANNOTATION,
+        attributes=sanitized_attributes,
+        status=TraceStatus(status_code="OK"),
     )
-    logger.debug("Emitting annotation span with keys %s", annotation_attributes)
-    with span:
-        pass
-    if not isinstance(span, ReadableSpan):
-        raise ValueError(f"Span is not a ReadableSpan: {span}")
-
-    return span
-
-
-def _safe_json_dump(obj: Any) -> str:
-    """Serialize an object to JSON, falling back to ``str(obj)`` if needed.
-
-    Args:
-        obj: Object to be serialized.
-
-    Returns:
-        The JSON-encoded string representation of the object, or its string
-        representation if JSON encoding fails.
-    """
-    try:
-        return json.dumps(obj, default=str, ensure_ascii=False)
-    except Exception:
-        return str(obj)
 
 
 class OperationContext:
     """Context manager and decorator for tracing operations.
 
-    This class manages an OpenTelemetry span for a logical unit of work. It can
-    be used either:
+    This class manages a tracer-backed span for a logical unit of work. It can be
+    used either:
 
     * As a decorator, in which case inputs and outputs are inferred
       automatically from the wrapped function's signature.
     * As a context manager, in which case inputs and outputs can be recorded
-      explicitly via :meth:`set_input` and :meth:`set_output`.
+      explicitly via [`set_input`][agentlightning.emitter.annotation.OperationContext.set_input]
+      and [`set_output`][agentlightning.emitter.annotation.OperationContext.set_output].
 
     Attributes:
         name: Human-readable span name.
         initial_attributes: Attributes applied when the span is created.
-        tracer: OpenTelemetry tracer used to create spans.
-        span: The currently active span, if any.
+        tracer: Tracer implementation used to create spans.
     """
 
-    def __init__(self, name: str, attributes: Dict[str, Any], *, propagate: bool = True) -> None:
+    def __init__(self, name: str, attributes: Dict[str, Any], propagate: bool = True) -> None:
         """Initialize a new operation context.
 
         Args:
@@ -112,12 +94,19 @@ class OperationContext:
                 JSON-serialized where necessary.
             propagate: Whether the span should be sent to active exporters.
         """
-        self.name: str = name
-        self.initial_attributes: Dict[str, Any] = attributes
-        self.propagate: bool = propagate
-        self.tracer: trace.Tracer = get_tracer(use_active_span_processor=propagate)
-        self.span: Optional[trace.Span] = None
-        self._ctx_token: Optional[ContextManager[Any]] = None
+        self.name = name
+        self.initial_attributes = flatten_attributes(attributes, expand_leaf_lists=False)
+        self.propagate = propagate
+        if propagate:
+            tracer = get_active_tracer()
+            if tracer is None:
+                raise RuntimeError("No active tracer found. Cannot trace operation spans.")
+            self.tracer = tracer
+        else:
+            self.tracer = DummyTracer()
+        self._ctx_manager: Optional[ContextManager[SpanRecordingContext]] = None
+        self._recording_context: Optional[SpanRecordingContext] = None
+        self._span: Optional[SpanCoreFields] = None
 
     def __enter__(self) -> "OperationContext":
         """Enter the context manager and start a new span.
@@ -125,15 +114,10 @@ class OperationContext:
         Returns:
             The current :class:`OperationContext` instance with an active span.
         """
-        # 1. Start the span with initial attributes (JSON serialized)
-        sanitized_attrs = {
-            k: _safe_json_dump(v) if not isinstance(v, (str, int, float, bool)) else v
-            for k, v in self.initial_attributes.items()
-        }
-
-        self.span = self.tracer.start_span(self.name, attributes=sanitized_attrs)
-        self._ctx_token = trace.use_span(self.span, end_on_exit=True)
-        self._ctx_token.__enter__()
+        sanitized_attrs = sanitize_attributes(self.initial_attributes)
+        self._ctx_manager = self.tracer.operation_context(self.name, attributes=sanitized_attrs)
+        recording_context = self._ctx_manager.__enter__()
+        self._recording_context = recording_context
         return self
 
     def __exit__(
@@ -142,57 +126,63 @@ class OperationContext:
         exc_val: Optional[BaseException],
         exc_tb: Optional[TracebackType],
     ) -> None:
-        """Exit the context manager and finish the span.
+        """Exit the context manager and finish the span."""
+        if self._ctx_manager:
+            self._ctx_manager.__exit__(exc_type, exc_val, exc_tb)
+        if self._recording_context:
+            self._span = self._recording_context.get_recorded_span()
+        self._ctx_manager = None
+        self._recording_context = None
 
-        Any exception raised inside the context is recorded on the span and the
-        span status is set to error.
-
-        Args:
-            exc_type: Exception type, if an exception occurred.
-            exc_val: Exception instance, if an exception occurred.
-            exc_tb: Traceback object, if an exception occurred.
-        """
-        # 1. Record Exception if present
-        if exc_val and self.span:
-            self.span.record_exception(exc_val)
-            self.span.set_status(Status(StatusCode.ERROR, str(exc_val)))
-
-        # 2. Close span
-        if self._ctx_token:
-            self._ctx_token.__exit__(exc_type, exc_val, exc_tb)
+    def span(self) -> SpanCoreFields:
+        """Get the span that was created by this context manager."""
+        if self._span is None:
+            raise RuntimeError("Span is not ready yet.")
+        return self._span
 
     def set_input(self, *args: Any, **kwargs: Any) -> None:
         """Record input arguments on the current span.
 
-        Positional arguments are stored under the ``input.args`` attribute,
-        and keyword arguments are stored under ``input.<name>`` attributes.
+        Positional arguments are stored under the `input.args.<index>` attributes,
+        and keyword arguments are stored under `input.<name>` attributes.
 
-        This is intended for use inside a ``with operation(...) as op`` block.
+        This is intended for use inside a `with operation(...) as op` block.
 
         Args:
             *args: Positional arguments to record.
             **kwargs: Keyword arguments to record.
         """
-        if not self.span:
-            return
+        if not self._recording_context:
+            raise RuntimeError("No recording context found. Cannot set input.")
 
+        prefix = LightningSpanAttributes.OPERATION_INPUT.value
+        attributes: Dict[str, Any] = {}
         if args:
-            self.span.set_attribute("input.args", _safe_json_dump(args))
+            for idx, value in enumerate(args):
+                flattened = flatten_attributes({str(idx): value})
+                for nested_key, nested_value in flattened.items():
+                    attributes[f"{prefix}.args.{nested_key}"] = nested_value
         if kwargs:
-            for k, v in kwargs.items():
-                self.span.set_attribute(f"input.{k}", _safe_json_dump(v))
+            for key, value in kwargs.items():
+                flattened = flatten_attributes({key: value})
+                for nested_key, nested_value in flattened.items():
+                    attributes[f"{prefix}.{nested_key}"] = nested_value
+        if attributes:
+            self._recording_context.record_attributes(sanitize_attributes(attributes))
 
     def set_output(self, output: Any) -> None:
         """Record the output value on the current span.
 
-        This is intended for use inside a ``with operation(...) as op`` block.
+        This is intended for use inside a `with operation(...) as op` block.
 
         Args:
             output: The output value to record.
         """
-        if not self.span:
-            return
-        self.span.set_attribute("output", _safe_json_dump(output))
+        if not self._recording_context:
+            raise RuntimeError("No recording context found. Cannot set output.")
+
+        flattened = flatten_attributes({LightningSpanAttributes.OPERATION_OUTPUT.value: output})
+        self._recording_context.record_attributes(sanitize_attributes(flattened))
 
     def __call__(self, fn: _FnType) -> _FnType:
         """Wrap a callable so its execution is traced in a span.
@@ -212,60 +202,60 @@ class OperationContext:
 
         sig = inspect.signature(fn)
 
-        def _record_auto_inputs(span: trace.Span, args: Tuple[Any, ...], kwargs: Dict[str, Any]) -> None:
-            """Bind arguments to signature and log them on the span.
+        sanitized_init_attrs = sanitize_attributes(
+            {LightningSpanAttributes.OPERATION_NAME.value: function_name, **self.initial_attributes}
+        )
 
-            Args:
-                span: Span on which to record attributes.
-                args: Positional arguments passed to the wrapped callable.
-                kwargs: Keyword arguments passed to the wrapped callable.
-            """
+        def _record_auto_inputs(
+            recording_ctx: SpanRecordingContext, args: Tuple[Any, ...], kwargs: Dict[str, Any]
+        ) -> None:
+            """Bind arguments to signature and log them on the span."""
+            attributes: Dict[str, Any] = {}
             try:
                 bound = sig.bind(*args, **kwargs)
                 bound.apply_defaults()
-                for k, v in bound.arguments.items():
-                    span.set_attribute(
-                        f"{LightningSpanAttributes.OPERATION_INPUT.value}.{k}",
-                        _safe_json_dump(v),
-                    )
+                for name, value in bound.arguments.items():
+                    parameter = sig.parameters.get(name)
+                    if parameter and parameter.kind is inspect.Parameter.VAR_POSITIONAL:
+                        attr_prefix = f"{LightningSpanAttributes.OPERATION_INPUT.value}.{name}"
+                        for idx, item in enumerate(value):
+                            flattened = flatten_attributes({str(idx): item})
+                            for nested_key, nested_value in flattened.items():
+                                attributes[f"{attr_prefix}.{nested_key}"] = nested_value
+                    else:
+                        flattened = flatten_attributes({name: value})
+                        for nested_key, nested_value in flattened.items():
+                            attributes[f"{LightningSpanAttributes.OPERATION_INPUT.value}.{nested_key}"] = nested_value
             except Exception:
-                span.set_attribute(
-                    f"{LightningSpanAttributes.OPERATION_INPUT.value}.args",
-                    _safe_json_dump(args),
-                )
-                span.set_attribute(
-                    f"{LightningSpanAttributes.OPERATION_INPUT.value}.kwargs",
-                    _safe_json_dump(kwargs),
-                )
+                if args:
+                    for idx, value in enumerate(args):
+                        flattened = flatten_attributes({str(idx): value})
+                        for nested_key, nested_value in flattened.items():
+                            attributes[f"{LightningSpanAttributes.OPERATION_INPUT.value}.args.{nested_key}"] = (
+                                nested_value
+                            )
+                if kwargs:
+                    flattened = flatten_attributes({"kwargs": kwargs})
+                    for nested_key, nested_value in flattened.items():
+                        attributes[f"{LightningSpanAttributes.OPERATION_INPUT.value}.{nested_key}"] = nested_value
+            if attributes:
+                recording_ctx.record_attributes(sanitize_attributes(attributes))
+
+        def _record_auto_outputs(recording_ctx: SpanRecordingContext, result: Any) -> None:
+            """Record the output value on the span."""
+            flattened = flatten_attributes({LightningSpanAttributes.OPERATION_OUTPUT.value: result})
+            recording_ctx.record_attributes(sanitize_attributes(flattened))
 
         if asyncio.iscoroutinefunction(fn) or inspect.iscoroutinefunction(fn):
 
             @functools.wraps(fn)
             async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
                 """Async wrapper that traces the wrapped coroutine."""
-                # Reuse __enter__ logic via 'with self' would share state incorrectly
-                # across concurrent calls. We must create a new span per call.
-                # So we manually reimplement the span logic for the wrapper here.
-
-                sanitized_attrs = {
-                    k: _safe_json_dump(v) if not isinstance(v, (str, int, float, bool)) else v
-                    for k, v in self.initial_attributes.items()
-                }
-
-                with self.tracer.start_as_current_span(self.name, attributes=sanitized_attrs) as span:
-                    span.set_attribute(LightningSpanAttributes.OPERATION_NAME.value, function_name)
-                    _record_auto_inputs(span, args, kwargs)
-                    try:
-                        result = await fn(*args, **kwargs)
-                        span.set_attribute(
-                            LightningSpanAttributes.OPERATION_OUTPUT.value,
-                            _safe_json_dump(result),
-                        )
-                        return result
-                    except Exception as e:
-                        span.record_exception(e)
-                        span.set_status(Status(StatusCode.ERROR, str(e)))
-                        raise
+                with self.tracer.operation_context(self.name, attributes=sanitized_init_attrs) as recording_ctx:
+                    _record_auto_inputs(recording_ctx, args, kwargs)
+                    result = await fn(*args, **kwargs)
+                    _record_auto_outputs(recording_ctx, result)
+                    return result
 
             return cast(_FnType, async_wrapper)
 
@@ -274,25 +264,11 @@ class OperationContext:
             @functools.wraps(fn)
             def sync_wrapper(*args: Any, **kwargs: Any) -> Any:
                 """Sync wrapper that traces the wrapped callable."""
-                sanitized_attrs = {
-                    k: _safe_json_dump(v) if not isinstance(v, (str, int, float, bool)) else v
-                    for k, v in self.initial_attributes.items()
-                }
-
-                with self.tracer.start_as_current_span(self.name, attributes=sanitized_attrs) as span:
-                    span.set_attribute(LightningSpanAttributes.OPERATION_NAME.value, function_name)
-                    _record_auto_inputs(span, args, kwargs)
-                    try:
-                        result = fn(*args, **kwargs)
-                        span.set_attribute(
-                            LightningSpanAttributes.OPERATION_OUTPUT.value,
-                            _safe_json_dump(result),
-                        )
-                        return result
-                    except Exception as e:
-                        span.record_exception(e)
-                        span.set_status(Status(StatusCode.ERROR, str(e)))
-                        raise
+                with self.tracer.operation_context(self.name, attributes=sanitized_init_attrs) as recording_ctx:
+                    _record_auto_inputs(recording_ctx, args, kwargs)
+                    result = fn(*args, **kwargs)
+                    _record_auto_outputs(recording_ctx, result)
+                    return result
 
             return cast(_FnType, sync_wrapper)
 

agentlightning/emitter/exception.py

@@ -1,13 +1,13 @@
 # Copyright (c) Microsoft. All rights reserved.
 
 import logging
-import traceback
 from typing import Any, Dict, Optional
 
-from opentelemetry.semconv.attributes import exception_attributes
-
 from agentlightning.semconv import AGL_EXCEPTION
-from agentlightning.utils.otel import get_tracer
+from agentlightning.tracer.base import get_active_tracer
+from agentlightning.tracer.dummy import DummyTracer
+from agentlightning.types import TraceStatus
+from agentlightning.utils.otel import flatten_attributes, format_exception_attributes, sanitize_attributes
 
 logger = logging.getLogger(__name__)
 
@@ -32,25 +32,23 @@ def emit_exception(
     """
     if not isinstance(exception, BaseException):  # type: ignore
         raise TypeError(f"Expected a BaseException instance, got: {type(exception)}.")
-
-    tracer = get_tracer(use_active_span_processor=propagate)
-    stacktrace = "".join(traceback.format_exception(type(exception), exception, exception.__traceback__))
-    span_attributes = {
-        exception_attributes.EXCEPTION_TYPE: type(exception).__name__,
-        exception_attributes.EXCEPTION_MESSAGE: str(exception),
-        exception_attributes.EXCEPTION_ESCAPED: True,
-    }
-    if stacktrace.strip():
-        span_attributes[exception_attributes.EXCEPTION_STACKTRACE] = stacktrace
+    span_attributes = format_exception_attributes(exception)
 
     if attributes:
-        span_attributes.update(attributes)
+        flattened = flatten_attributes(attributes, expand_leaf_lists=False)
+        span_attributes.update(sanitize_attributes(flattened))
 
-    span = tracer.start_span(
+    logger.debug("Emitting exception span for %s", type(exception).__name__)
+
+    if propagate:
+        tracer = get_active_tracer()
+        if tracer is None:
+            raise RuntimeError("No active tracer found. Cannot emit exception span.")
+    else:
+        tracer = DummyTracer()
+    tracer.create_span(
         AGL_EXCEPTION,
         attributes=span_attributes,
+        # The exception span is successful by itself.
+        status=TraceStatus(status_code="OK"),
     )
-    logger.debug("Emitting exception span for %s", type(exception).__name__)
-    with span:
-        span.record_exception(exception)
-        # We don't set the status of the span here. They have other semantics.

agentlightning/emitter/message.py

@@ -4,8 +4,10 @@ import logging
 from typing import Any, Dict, Optional
 
 from agentlightning.semconv import AGL_MESSAGE, LightningSpanAttributes
-from agentlightning.types import SpanLike
-from agentlightning.utils.otel import get_tracer
+from agentlightning.tracer.base import get_active_tracer
+from agentlightning.tracer.dummy import DummyTracer
+from agentlightning.types import Attributes, SpanLike
+from agentlightning.utils.otel import flatten_attributes, sanitize_attributes
 
 logger = logging.getLogger(__name__)
 
@@ -27,17 +29,21 @@ def emit_message(message: str, attributes: Optional[Dict[str, Any]] = None, prop
     if not isinstance(message, str):  # type: ignore
         raise TypeError(f"Message must be a string or list of strings, got: {type(message)}.")
 
-    tracer = get_tracer(use_active_span_processor=propagate)
-    span_attributes = {LightningSpanAttributes.MESSAGE_BODY.value: message}
+    if propagate:
+        tracer = get_active_tracer()
+        if tracer is None:
+            raise RuntimeError("No active tracer found. Cannot emit message span.")
+    else:
+        tracer = DummyTracer()
+    span_attributes: Attributes = {LightningSpanAttributes.MESSAGE_BODY.value: message}
     if attributes:
-        span_attributes.update(attributes)
-    span = tracer.start_span(
+        flattened = flatten_attributes(attributes, expand_leaf_lists=False)
+        span_attributes.update(sanitize_attributes(flattened))
+    logger.debug("Emitting message span with message: %s", message)
+    tracer.create_span(
         AGL_MESSAGE,
         attributes=span_attributes,
     )
-    logger.debug("Emitting message span with message: %s", message)
-    with span:
-        pass
 
 
 def get_message_value(span: SpanLike) -> Optional[str]:

agentlightning/emitter/object.py

@@ -6,13 +6,15 @@ import logging
 from typing import Any, Dict, Optional
 
 from agentlightning.semconv import AGL_OBJECT, LightningSpanAttributes
-from agentlightning.types import SpanLike
-from agentlightning.utils.otel import full_qualified_name, get_tracer
+from agentlightning.tracer.base import get_active_tracer
+from agentlightning.tracer.dummy import DummyTracer
+from agentlightning.types import SpanCoreFields, SpanLike, TraceStatus
+from agentlightning.utils.otel import flatten_attributes, full_qualified_name, sanitize_attributes
 
 logger = logging.getLogger(__name__)
 
 
-def emit_object(object: Any, attributes: Optional[Dict[str, Any]] = None, propagate: bool = True) -> None:
+def emit_object(object: Any, attributes: Optional[Dict[str, Any]] = None, propagate: bool = True) -> SpanCoreFields:
     """Emit an object's serialized representation as an OpenTelemetry span.
 
     Args:
@@ -25,20 +27,29 @@ def emit_object(object: Any, attributes: Optional[Dict[str, Any]] = None, propag
     """
     span_attributes = encode_object(object)
     if attributes:
-        span_attributes.update(attributes)
-    tracer = get_tracer(use_active_span_processor=propagate)
-    span = tracer.start_span(
-        AGL_OBJECT,
-        attributes=span_attributes,
-    )
+        flattened = flatten_attributes(attributes, expand_leaf_lists=False)
+        span_attributes.update(sanitize_attributes(flattened))
+
     attr_length = 0
     if LightningSpanAttributes.OBJECT_JSON.value in span_attributes:
         attr_length = len(span_attributes[LightningSpanAttributes.OBJECT_JSON.value])
     elif LightningSpanAttributes.OBJECT_LITERAL.value in span_attributes:
         attr_length = len(span_attributes[LightningSpanAttributes.OBJECT_LITERAL.value])
     logger.debug("Emitting object span with payload size %d characters", attr_length)
-    with span:
-        pass
+
+    if propagate:
+        tracer = get_active_tracer()
+        if tracer is None:
+            raise RuntimeError("No active tracer found. Cannot emit annotation span.")
+    else:
+        # Do not actually propagate to any store or tracer backend.
+        tracer = DummyTracer()
+
+    return tracer.create_span(
+        name=AGL_OBJECT,
+        attributes=span_attributes,
+        status=TraceStatus(status_code="OK"),
+    )
 
 
 def encode_object(object: Any) -> Dict[str, Any]:

agentlightning/emitter/reward.py

@@ -20,13 +20,10 @@ from typing import (
     cast,
 )
 
-import agentops
-from agentops.sdk.decorators import operation
-from opentelemetry.sdk.trace import ReadableSpan
 from pydantic import TypeAdapter
 
 from agentlightning.semconv import AGL_ANNOTATION, LightningSpanAttributes, RewardPydanticModel
-from agentlightning.types import SpanLike
+from agentlightning.types import SpanCoreFields, SpanLike
 from agentlightning.utils.otel import filter_and_unflatten_attributes
 
 from .annotation import emit_annotation
@@ -61,6 +58,8 @@ _FnType = TypeVar("_FnType", bound=Callable[..., Any])
 
 def _agentops_initialized() -> bool:
     """Return `True` when the AgentOps client has been configured."""
+    import agentops
+
     return agentops.get_client().initialized
 
 
@@ -81,6 +80,8 @@ def reward(fn: _FnType) -> _FnType:
         Wrapped callable that preserves the original signature.
     """
 
+    from agentops.sdk.decorators import operation
+
     def wrap_result(result: Optional[float]) -> _RewardSpanData:
         """Normalize the reward value into the span payload format."""
         if result is None:
@@ -146,7 +147,7 @@ def emit_reward(
     primary_key: str | None = None,
     attributes: Dict[str, Any] | None = None,
     propagate: bool = True,
-) -> ReadableSpan:
+) -> SpanCoreFields:
     """Emit a reward value as an OpenTelemetry span.
 
     Examples:
@@ -172,11 +173,7 @@ def emit_reward(
         propagate: Whether to propagate the span to exporters automatically.
 
     Returns:
-        Readable span capturing the recorded reward.
-
-    Raises:
-        ValueError: If the provided reward cannot be interpreted as a float or the
-            resulting span is not a [`ReadableSpan`](https://opentelemetry.io/docs/concepts/signals/traces/) instance.
+        Span core fields capturing the recorded reward.
     """
     logger.debug(f"Emitting reward: {reward}")
     reward_dimensions: List[RewardDimension] = []

agentlightning/instrumentation/__init__.py

@@ -119,20 +119,3 @@ def uninstrument_all():
             warnings.warn("agentops_langchain is installed but uninstrument_agentops_langchain could not be imported.")
     else:
         warnings.warn("Agentops-langchain integration is not installed. It's therefore not uninstrumented.")
-
-
-def instrument_weave():
-    if WEAVE_INSTALLED:
-        from .weave import instrument_weave
-
-        instrument_weave()
-
-
-def uninstrument_weave():
-    if WEAVE_INSTALLED:
-        try:
-            from .weave import uninstrument_weave
-
-            uninstrument_weave()
-        except ImportError:
-            warnings.warn("weave is installed but uninstrument_weave could not be imported.")

agentlightning/instrumentation/weave.py

@@ -1,139 +1,492 @@
 # Copyright (c) Microsoft. All rights reserved.
 
-import logging
-import os
-from typing import Any, Callable, Optional
+from __future__ import annotations
 
-import requests
+import logging
+import threading
+from datetime import datetime, timezone
+from typing import Any, Callable, Dict, Iterator, List
+
+import weave.trace.weave_init
+from pydantic import validate_call
+from weave.trace_server import trace_server_interface as tsi
+from weave.trace_server.ids import generate_id
+from weave.trace_server_bindings.client_interface import TraceServerClientInterface
+from weave.trace_server_bindings.models import ServerInfoRes
 
 logger = logging.getLogger(__name__)
 
 __all__ = [
     "instrument_weave",
     "uninstrument_weave",
+    "InMemoryWeaveTraceServer",
 ]
 
+
+class InMemoryWeaveTraceServer(TraceServerClientInterface):
+    """A minimal in-memory implementation of the TraceServerInterface.
+
+    It stores calls and objects in local dictionaries and returns valid Pydantic
+    responses to satisfy the Weave client and FullTraceServerInterface protocol.
+    """
+
+    def __init__(self):
+        # Minimal storage to allow basic querying in tests
+        self.calls: Dict[str, tsi.CallSchema] = {}
+        self.partial_calls: Dict[str, Dict[str, Any]] = {}
+        self.objs: Dict[str, Any] = {}
+        self.files: Dict[str, bytes] = {}
+        self.feedback: List[tsi.FeedbackCreateReq] = []
+
+        self._call_threading_lock = threading.Lock()
+
+    @classmethod
+    def from_env(cls, *args: Any, **kwargs: Any) -> InMemoryWeaveTraceServer:
+        return cls()
+
+    def server_info(self) -> ServerInfoRes:
+        return ServerInfoRes(min_required_weave_python_version="0.52.22")
+
+    def ensure_project_exists(self, entity: str, project: str) -> tsi.EnsureProjectExistsRes:
+        return tsi.EnsureProjectExistsRes(project_name=project)
+
+    # --- Call API ---
+
+    @validate_call
+    def call_start(self, req: tsi.CallStartReq) -> tsi.CallStartRes:
+        # NOTE: It's not necessary that call_end must be called after call_start.
+        request_content = req.start.model_dump(exclude_none=True)
+
+        # If id needs to be generated here, it's very likely we won't be able to find the call later.
+        # This is just to make the type checker happy.
+        call_id = request_content.get("id") or generate_id()
+        trace_id = request_content.get("trace_id") or generate_id()
+        request_content["id"] = call_id
+        request_content["trace_id"] = trace_id
+
+        with self._call_threading_lock:
+            if call_id in self.partial_calls:
+                # call_end has already been called for this call.
+                kwargs = {**request_content, **self.partial_calls[call_id]}
+                self.calls[call_id] = tsi.CallSchema(**kwargs)
+                del self.partial_calls[call_id]
+            else:
+                self.partial_calls[call_id] = request_content
+
+        return tsi.CallStartRes(id=call_id, trace_id=trace_id)
+
+    @validate_call
+    def call_end(self, req: tsi.CallEndReq) -> tsi.CallEndRes:
+        request_content = req.end.model_dump(exclude_none=True)
+        call_id = req.end.id
+
+        with self._call_threading_lock:
+            if call_id in self.partial_calls:
+                # End request always override the start request content.
+                kwargs = {**self.partial_calls[call_id], **request_content}
+                self.calls[call_id] = tsi.CallSchema(**kwargs)
+                del self.partial_calls[call_id]
+            else:
+                self.partial_calls[call_id] = request_content
+        return tsi.CallEndRes()
+
+    @validate_call
+    def call_start_batch(self, req: tsi.CallCreateBatchReq) -> tsi.CallCreateBatchRes:
+        for item in req.batch:
+            if isinstance(item, tsi.CallStartReq):
+                self.call_start(item)
+            elif isinstance(item, tsi.CallEndReq):
+                self.call_end(item)
+        return tsi.CallCreateBatchRes(res=[])
+
+    @validate_call
+    def call_read(self, req: tsi.CallReadReq) -> tsi.CallReadRes:
+        call_data = self.calls.get(req.id)
+        return tsi.CallReadRes(call=call_data)
+
+    @validate_call
+    def calls_query(self, req: tsi.CallsQueryReq) -> tsi.CallsQueryRes:
+        return tsi.CallsQueryRes(calls=list(self.calls_query_stream(req)))
+
+    @validate_call
+    def calls_query_stream(self, req: tsi.CallsQueryReq) -> Iterator[tsi.CallSchema]:
+        yield from self.calls.values()
+
+    @validate_call
+    def calls_delete(self, req: tsi.CallsDeleteReq) -> tsi.CallsDeleteRes:
+        num_deleted = 0
+        for call_id in req.call_ids:
+            if call_id in self.calls:
+                del self.calls[call_id]
+                num_deleted += 1
+        return tsi.CallsDeleteRes(num_deleted=num_deleted)
+
+    @validate_call
+    def call_update(self, req: tsi.CallUpdateReq) -> tsi.CallUpdateRes:
+        return tsi.CallUpdateRes()
+
+    @validate_call
+    def calls_query_stats(self, req: tsi.CallsQueryStatsReq) -> tsi.CallsQueryStatsRes:
+        return tsi.CallsQueryStatsRes(count=len(self.calls))
+
+    # --- Cost API ---
+
+    @validate_call
+    def cost_create(self, req: tsi.CostCreateReq) -> tsi.CostCreateRes:
+        return tsi.CostCreateRes(ids=[(generate_id(), generate_id()) for _ in req.costs])
+
+    @validate_call
+    def cost_query(self, req: tsi.CostQueryReq) -> tsi.CostQueryRes:
+        return tsi.CostQueryRes(results=[])
+
+    @validate_call
+    def cost_purge(self, req: tsi.CostPurgeReq) -> tsi.CostPurgeRes:
+        return tsi.CostPurgeRes()
+
+    # --- Object API (Legacy V1) ---
+
+    @validate_call
+    def obj_create(self, req: tsi.ObjCreateReq) -> tsi.ObjCreateRes:
+        digest = generate_id()
+        self.objs[digest] = req.obj
+        return tsi.ObjCreateRes(digest=digest)
+
+    @validate_call
+    def obj_read(self, req: tsi.ObjReadReq) -> tsi.ObjReadRes:
+        return tsi.ObjReadRes(obj=self.objs.get(req.digest, {}))
+
+    @validate_call
+    def objs_query(self, req: tsi.ObjQueryReq) -> tsi.ObjQueryRes:
+        return tsi.ObjQueryRes(objs=[])
+
+    @validate_call
+    def obj_delete(self, req: tsi.ObjDeleteReq) -> tsi.ObjDeleteRes:
+        return tsi.ObjDeleteRes(num_deleted=0)
+
+    # --- Table API ---
+
+    @validate_call
+    def table_create(self, req: tsi.TableCreateReq) -> tsi.TableCreateRes:
+        return tsi.TableCreateRes(digest=generate_id(), row_digests=[])
+
+    @validate_call
+    def table_create_from_digests(self, req: tsi.TableCreateFromDigestsReq) -> tsi.TableCreateFromDigestsRes:
+        return tsi.TableCreateFromDigestsRes(digest=generate_id())
+
+    @validate_call
+    def table_update(self, req: tsi.TableUpdateReq) -> tsi.TableUpdateRes:
+        return tsi.TableUpdateRes(digest=generate_id(), updated_row_digests=[])
+
+    @validate_call
+    def table_query(self, req: tsi.TableQueryReq) -> tsi.TableQueryRes:
+        return tsi.TableQueryRes(rows=[])
+
+    @validate_call
+    def table_query_stream(self, req: tsi.TableQueryReq) -> Iterator[tsi.TableRowSchema]:
+        yield from []
+
+    @validate_call
+    def table_query_stats(self, req: tsi.TableQueryStatsReq) -> tsi.TableQueryStatsRes:
+        return tsi.TableQueryStatsRes(count=0)
+
+    @validate_call
+    def table_query_stats_batch(self, req: tsi.TableQueryStatsBatchReq) -> tsi.TableQueryStatsBatchRes:
+        return tsi.TableQueryStatsBatchRes(tables=[])
+
+    # --- Ref API ---
+
+    @validate_call
+    def refs_read_batch(self, req: tsi.RefsReadBatchReq) -> tsi.RefsReadBatchRes:
+        return tsi.RefsReadBatchRes(vals=[])
+
+    # --- File API ---
+
+    def file_create(self, req: tsi.FileCreateReq) -> tsi.FileCreateRes:
+        self.files[req.name] = req.content
+        return tsi.FileCreateRes(digest=generate_id())
+
+    def file_content_read(self, req: tsi.FileContentReadReq) -> tsi.FileContentReadRes:
+        return tsi.FileContentReadRes(content=self.files.get(req.digest, b"dummy_content"))
+
+    def files_stats(self, req: tsi.FilesStatsReq) -> tsi.FilesStatsRes:
+        total_size = sum(len(c) for c in self.files.values())
+        return tsi.FilesStatsRes(total_size_bytes=total_size)
+
+    # --- Feedback API ---
+
+    @validate_call
+    def feedback_create(self, req: tsi.FeedbackCreateReq) -> tsi.FeedbackCreateRes:
+        req.id = req.id or generate_id()
+        self.feedback.append(req)
+        return tsi.FeedbackCreateRes(
+            id=req.id,
+            created_at=datetime.now(timezone.utc),
+            wb_user_id="dummy_user",
+            payload=req.payload,
+        )
+
+    def feedback_create_batch(self, req: tsi.FeedbackCreateBatchReq) -> tsi.FeedbackCreateBatchRes:
+        results: List[tsi.FeedbackCreateRes] = []
+        for item in req.batch:
+            res = self.feedback_create(item)
+            results.append(res)
+        return tsi.FeedbackCreateBatchRes(res=results)
+
+    @validate_call
+    def feedback_query(self, req: tsi.FeedbackQueryReq) -> tsi.FeedbackQueryRes:
+        return tsi.FeedbackQueryRes(result=[])
+
+    @validate_call
+    def feedback_purge(self, req: tsi.FeedbackPurgeReq) -> tsi.FeedbackPurgeRes:
+        self.feedback.clear()
+        return tsi.FeedbackPurgeRes()
+
+    @validate_call
+    def feedback_replace(self, req: tsi.FeedbackReplaceReq) -> tsi.FeedbackReplaceRes:
+        return tsi.FeedbackReplaceRes(
+            id=req.id or generate_id(),
+            created_at=datetime.now(timezone.utc),
+            wb_user_id="dummy",
+            payload={},
+        )
+
+    # --- Action API ---
+
+    @validate_call
+    def actions_execute_batch(self, req: tsi.ActionsExecuteBatchReq) -> tsi.ActionsExecuteBatchRes:
+        return tsi.ActionsExecuteBatchRes()
+
+    # --- Execute LLM API ---
+
+    @validate_call
+    def completions_create(self, req: tsi.CompletionsCreateReq) -> tsi.CompletionsCreateRes:
+        return tsi.CompletionsCreateRes(response={"choices": [{"text": "dummy completion"}]})
+
+    @validate_call
+    def completions_create_stream(self, req: tsi.CompletionsCreateReq) -> Iterator[dict[str, Any]]:
+        yield {"choices": [{"text": "dummy "}]}
+        yield {"choices": [{"text": "stream"}]}
+
+    # --- Execute Image Generation API ---
+
+    @validate_call
+    def image_create(self, req: tsi.ImageGenerationCreateReq) -> tsi.ImageGenerationCreateRes:
+        return tsi.ImageGenerationCreateRes(response={})
+
+    # --- Project Statistics API ---
+
+    @validate_call
+    def project_stats(self, req: tsi.ProjectStatsReq) -> tsi.ProjectStatsRes:
+        return tsi.ProjectStatsRes(
+            trace_storage_size_bytes=0,
+            objects_storage_size_bytes=0,
+            tables_storage_size_bytes=0,
+            files_storage_size_bytes=0,
+        )
+
+    # --- Thread API ---
+
+    @validate_call
+    def threads_query_stream(self, req: tsi.ThreadsQueryReq) -> Iterator[tsi.ThreadSchema]:
+        yield from []
+
+    # --- Evaluation API (V1) ---
+
+    @validate_call
+    def evaluate_model(self, req: tsi.EvaluateModelReq) -> tsi.EvaluateModelRes:
+        return tsi.EvaluateModelRes(call_id=generate_id())
+
+    @validate_call
+    def evaluation_status(self, req: tsi.EvaluationStatusReq) -> tsi.EvaluationStatusRes:
+        return tsi.EvaluationStatusRes(status=tsi.EvaluationStatusNotFound())
+
+    # --- OTEL API ---
+
+    def otel_export(self, req: tsi.OtelExportReq) -> tsi.OtelExportRes:
+        return tsi.OtelExportRes()
+
+    # ==========================================
+    # Object Interface (V2 APIs)
+    # ==========================================
+
+    # --- Ops ---
+    def op_create(self, req: tsi.OpCreateReq) -> tsi.OpCreateRes:
+        return tsi.OpCreateRes(digest=generate_id(), object_id=generate_id(), version_index=0)
+
+    def op_read(self, req: tsi.OpReadReq) -> tsi.OpReadRes:
+        return tsi.OpReadRes(op=None)  # type: ignore
+
+    def op_list(self, req: tsi.OpListReq) -> Iterator[tsi.OpReadRes]:
+        yield from []
+
+    def op_delete(self, req: tsi.OpDeleteReq) -> tsi.OpDeleteRes:
+        return tsi.OpDeleteRes(num_deleted=0)
+
+    # --- Datasets ---
+    def dataset_create(self, req: tsi.DatasetCreateReq) -> tsi.DatasetCreateRes:
+        return tsi.DatasetCreateRes(digest=generate_id(), object_id=generate_id(), version_index=0)
+
+    def dataset_read(self, req: tsi.DatasetReadReq) -> tsi.DatasetReadRes:
+        return tsi.DatasetReadRes(dataset=None)  # type: ignore
+
+    def dataset_list(self, req: tsi.DatasetListReq) -> Iterator[tsi.DatasetReadRes]:
+        yield from []
+
+    def dataset_delete(self, req: tsi.DatasetDeleteReq) -> tsi.DatasetDeleteRes:
+        return tsi.DatasetDeleteRes(num_deleted=0)
+
+    # --- Scorers ---
+    def scorer_create(self, req: tsi.ScorerCreateReq) -> tsi.ScorerCreateRes:
+        return tsi.ScorerCreateRes(digest=generate_id(), object_id=generate_id(), version_index=0, scorer=generate_id())
+
+    def scorer_read(self, req: tsi.ScorerReadReq) -> tsi.ScorerReadRes:
+        return tsi.ScorerReadRes(scorer=None)  # type: ignore
+
+    def scorer_list(self, req: tsi.ScorerListReq) -> Iterator[tsi.ScorerReadRes]:
+        yield from []
+
+    def scorer_delete(self, req: tsi.ScorerDeleteReq) -> tsi.ScorerDeleteRes:
+        return tsi.ScorerDeleteRes(num_deleted=0)
+
+    # --- Evaluations (V2) ---
+    def evaluation_create(self, req: tsi.EvaluationCreateReq) -> tsi.EvaluationCreateRes:
+        return tsi.EvaluationCreateRes(
+            digest=generate_id(), object_id=generate_id(), version_index=0, evaluation_ref=generate_id()
+        )
+
+    def evaluation_read(self, req: tsi.EvaluationReadReq) -> tsi.EvaluationReadRes:
+        return tsi.EvaluationReadRes(evaluation=None)  # type: ignore
+
+    def evaluation_list(self, req: tsi.EvaluationListReq) -> Iterator[tsi.EvaluationReadRes]:
+        yield from []
+
+    def evaluation_delete(self, req: tsi.EvaluationDeleteReq) -> tsi.EvaluationDeleteRes:
+        return tsi.EvaluationDeleteRes(num_deleted=0)
+
+    # --- Models ---
+    def model_create(self, req: tsi.ModelCreateReq) -> tsi.ModelCreateRes:
+        return tsi.ModelCreateRes(
+            digest=generate_id(), object_id=generate_id(), version_index=0, model_ref=generate_id()
+        )
+
+    def model_read(self, req: tsi.ModelReadReq) -> tsi.ModelReadRes:
+        return tsi.ModelReadRes(model=None)  # type: ignore
+
+    def model_list(self, req: tsi.ModelListReq) -> Iterator[tsi.ModelReadRes]:
+        yield from []
+
+    def model_delete(self, req: tsi.ModelDeleteReq) -> tsi.ModelDeleteRes:
+        return tsi.ModelDeleteRes(num_deleted=0)
+
+    # --- Evaluation Runs ---
+    def evaluation_run_create(self, req: tsi.EvaluationRunCreateReq) -> tsi.EvaluationRunCreateRes:
+        return tsi.EvaluationRunCreateRes(evaluation_run_id=generate_id())
+
+    def evaluation_run_read(self, req: tsi.EvaluationRunReadReq) -> tsi.EvaluationRunReadRes:
+        return tsi.EvaluationRunReadRes(evaluation_run=None)  # type: ignore
+
+    def evaluation_run_list(self, req: tsi.EvaluationRunListReq) -> Iterator[tsi.EvaluationRunReadRes]:
+        yield from []
+
+    def evaluation_run_delete(self, req: tsi.EvaluationRunDeleteReq) -> tsi.EvaluationRunDeleteRes:
+        return tsi.EvaluationRunDeleteRes(num_deleted=0)
+
+    def evaluation_run_finish(self, req: tsi.EvaluationRunFinishReq) -> tsi.EvaluationRunFinishRes:
+        return tsi.EvaluationRunFinishRes(success=True)
+
+    # --- Predictions ---
+    def prediction_create(self, req: tsi.PredictionCreateReq) -> tsi.PredictionCreateRes:
+        return tsi.PredictionCreateRes(prediction_id=generate_id())
+
+    def prediction_read(self, req: tsi.PredictionReadReq) -> tsi.PredictionReadRes:
+        return tsi.PredictionReadRes(prediction=None)  # type: ignore
+
+    def prediction_list(self, req: tsi.PredictionListReq) -> Iterator[tsi.PredictionReadRes]:
+        yield from []
+
+    def prediction_delete(self, req: tsi.PredictionDeleteReq) -> tsi.PredictionDeleteRes:
+        return tsi.PredictionDeleteRes(num_deleted=0)
+
+    def prediction_finish(self, req: tsi.PredictionFinishReq) -> tsi.PredictionFinishRes:
+        return tsi.PredictionFinishRes(success=True)
+
+    # --- Scores ---
+    def score_create(self, req: tsi.ScoreCreateReq) -> tsi.ScoreCreateRes:
+        return tsi.ScoreCreateRes(score_id=generate_id())
+
+    def score_read(self, req: tsi.ScoreReadReq) -> tsi.ScoreReadRes:
+        return tsi.ScoreReadRes(score=None)  # type: ignore
+
+    def score_list(self, req: tsi.ScoreListReq) -> Iterator[tsi.ScoreReadRes]:
+        yield from []
+
+    def score_delete(self, req: tsi.ScoreDeleteReq) -> tsi.ScoreDeleteRes:
+        return tsi.ScoreDeleteRes(num_deleted=0)
+
+
 # Module-level storage for originals
-_original_default_entity_name_getter: Callable[..., Any] | None = None
-_original_upsert_project_getter: Callable[..., Any] | None = None
-_original_weave_get = False
-_original_weave_post = False
+_original_init_weave_get_server: Callable[..., Any] | None = None
+_original_get_entity_project_from_project_name: Callable[..., Any] | None = None
+_original_get_username: Callable[..., Any] | None = None
 
 
-def instrument_weave():
-    """
-    Patch the Weave/W&B integration to bypass actual network calls for testing.
+def init_weave_get_server_factory(server: InMemoryWeaveTraceServer) -> Callable[..., Any]:
+    # Bypass the usage of Weave remote server
+    def init_weave_get_server(*args: Any, **kwargs: Any) -> InMemoryWeaveTraceServer:
+        return server
 
-    - Mocks HTTP POST/GET requests
-    - Patches wandb.Api methods
-    - Silences Weave logging
-    - Sets dummy WANDB_API_KEY if not provided
-    """
+    return init_weave_get_server
+
+
+def get_entity_project_from_project_name_factory(entity_name: str) -> tuple[str, str]:
+    # Bypass the usage of API
     try:
-        import weave
-        from weave.compat import wandb  # type: ignore
-    except ImportError:
-        logger.warning("Weave or wandb not installed; cannot uninstrument.")
-        return
+        assert _original_get_entity_project_from_project_name is not None
+        return _original_get_entity_project_from_project_name(entity_name)
+    except weave.trace.weave_init.WeaveWandbAuthenticationException:
+        # In case API is not available.
+        return "agl", "weave"
 
-    _weave_tracer_entity_name = "weave_tracer_entity"
 
-    def default_entity_name_getter(_self) -> str:  # type: ignore
-        return _weave_tracer_entity_name
+def get_username() -> str:
+    # Bypass the usage of API
+    try:
+        assert _original_get_username is not None
+        return _original_get_username()
+    except RuntimeError:
+        return "agl"
 
-    def upsert_project_getter(
-        _self, project: str, description: Optional[str] = None, entity: Optional[str] = None  # type: ignore
-    ) -> dict[str, Any]:
-        return {
-            "upsertModel": {
-                "model": {
-                    "name": project,
-                    "description": description or "",
-                    "entity": entity or _weave_tracer_entity_name,
-                }
-            },
-            "project": "weave_tracer_project",
-        }
 
-    # Mock network requests to avoid real HTTP calls
-    def post(url: str, *args: Any, **kwargs: Any) -> requests.Response:
-        response = requests.Response()
-        response.status_code = 200
-        response._content = b'{"digest": "mocked_digest"}'
-        return response
+def instrument_weave(server: InMemoryWeaveTraceServer):
+    """Patch the Weave/W&B integration to bypass actual network calls for testing."""
 
-    def get(url: str, *args: Any, **kwargs: Any) -> requests.Response:
-        response = requests.Response()
-        response.status_code = 200
-        response._content = b'{"min_required_weave_python_version": "0.52.14"}'
-        return response
-
-    # Patch API methods and HTTP requests
-    global _original_default_entity_name_getter
-    global _original_upsert_project_getter
-    global _original_weave_post
-    global _original_weave_get
-    _original_default_entity_name_getter = wandb.Api.default_entity_name  # type: ignore
-    _original_upsert_project_getter = wandb.Api.upsert_project  # type: ignore
-    _original_weave_post = weave.utils.http_requests.session.post  # type: ignore
-    _original_weave_get = weave.utils.http_requests.session.get  # type: ignore
-
-    # Patch API methods and HTTP requests
-    wandb.Api.default_entity_name = default_entity_name_getter  # type: ignore
-    wandb.Api.upsert_project = upsert_project_getter  # type: ignore
-    weave.utils.http_requests.session.post = post  # type: ignore
-    weave.utils.http_requests.session.get = get  # type: ignore
-
-    # Silence Weave logging
-    for name in logging.root.manager.loggerDict:
-        if name.startswith("weave"):
-            logging.getLogger(name).disabled = True
-
-    # Set dummy API key if missing
-    if not os.environ.get("WANDB_API_KEY"):
-        os.environ["WANDB_API_KEY"] = "dumped_api_key_for_weave_tracer"
-
-    # if needed in future tests, enable this and replace WF_TRACE_SERVER_URL to local server
-    # full_url = f"http://127.0.0.1:{_port}"
-    # os.environ["WF_TRACE_SERVER_URL"] = full_url
+    global _original_init_weave_get_server, _original_get_entity_project_from_project_name, _original_get_username
+    _original_init_weave_get_server = weave.trace.weave_init.init_weave_get_server
+    _original_get_entity_project_from_project_name = weave.trace.weave_init.get_entity_project_from_project_name
+    _original_get_username = weave.trace.weave_init.get_username
+    weave.trace.weave_init.init_weave_get_server = init_weave_get_server_factory(server)
+    weave.trace.weave_init.get_entity_project_from_project_name = get_entity_project_from_project_name_factory
+    weave.trace.weave_init.get_username = get_username
 
 
 def uninstrument_weave():
-    """
-    Restore the original Weave/W&B integration methods and HTTP requests.
-    """
-    try:
-        import weave
-        from weave.compat import wandb  # type: ignore
-    except ImportError:
-        logger.warning("Weave or wandb not installed; cannot uninstrument.")
-        return
+    """Restore the original Weave/W&B integration methods and HTTP requests."""
+    global _original_init_weave_get_server, _original_get_entity_project_from_project_name, _original_get_username
 
-    global _original_default_entity_name_getter
-    if _original_default_entity_name_getter is not None:
-        wandb.Api.default_entity_name = _original_default_entity_name_getter  # type: ignore
-        _original_default_entity_name_getter = None
-        logger.info("restored wandb.Api.default_entity_name")
+    if _original_init_weave_get_server is not None:
+        weave.trace.weave_init.init_weave_get_server = _original_init_weave_get_server
+        _original_init_weave_get_server = None
+    else:
+        raise RuntimeError("Weave/W&B integration was not instrumented.")
 
-    global _original_upsert_project_getter
-    if _original_upsert_project_getter is not None:
-        wandb.Api.upsert_project = _original_upsert_project_getter  # type: ignore
-        _original_upsert_project_getter = None
-        logger.info("restored wandb.Api.upsert_project")
+    if _original_get_entity_project_from_project_name is not None:
+        weave.trace.weave_init.get_entity_project_from_project_name = _original_get_entity_project_from_project_name
+        _original_get_entity_project_from_project_name = None
+    else:
+        raise RuntimeError("Weave/W&B integration was not instrumented.")
 
-    global _original_weave_post
-    if _original_weave_post is not None:
-        weave.utils.http_requests.session.post = _original_weave_post  # type: ignore
-        _original_weave_post = None
-        logger.info("restored weave.utils.http_requests.session.post")
-
-    global _original_weave_get
-    if _original_weave_get is not None:
-        weave.utils.http_requests.session.get = _original_weave_get  # type: ignore
-        _original_weave_get = None
-        logger.info("restored weave.utils.http_requests.session.get")
-
-    # Restore Weave logging
-    for name in logging.root.manager.loggerDict:
-        if name.startswith("weave"):
-            logging.getLogger(name).disabled = False
+    if _original_get_username is not None:
+        weave.trace.weave_init.get_username = _original_get_username
+        _original_get_username = None
+    else:
+        raise RuntimeError("Weave/W&B integration was not instrumented.")

agentlightning/runner/agent.py

@@ -43,6 +43,7 @@ from agentlightning.types import (
     RolloutMode,
     RolloutRawResult,
     Span,
+    SpanCoreFields,
 )
 from agentlightning.utils.system_snapshot import system_snapshot
 
@@ -276,7 +277,7 @@ class LitAgentRunner(Runner[T_task]):
         """
         store = self.get_store()
 
-        trace_spans: list[ReadableSpan] | list[Span] = []
+        trace_spans: list[Span] = []
         result_recognized: bool = False
 
         # Case 0: result is None
@@ -295,31 +296,38 @@ class LitAgentRunner(Runner[T_task]):
             # Preserve the existing spans before another span is emitted
             trace_spans = list(self._tracer.get_last_trace())
             # This will NOT emit another span to the tracer
-            reward_span = emit_reward(raw_result, propagate=False)
+            reward_span_core_fields = emit_reward(raw_result, propagate=False)
             # We add it to the store manually
-            await store.add_otel_span(rollout.rollout_id, rollout.attempt.attempt_id, reward_span)
-            trace_spans.append(reward_span)
+            sequence_id = await store.get_next_span_sequence_id(rollout.rollout_id, rollout.attempt.attempt_id)
+            reward_span = Span.from_core_fields(
+                reward_span_core_fields,
+                rollout_id=rollout.rollout_id,
+                attempt_id=rollout.attempt.attempt_id,
+                sequence_id=sequence_id,
+            )
+            await store.add_span(reward_span)
             result_recognized = True
 
-        # Case 2-3: result is a list
+        # Case 2-4: result is a list
         if isinstance(raw_result, list):
-            # For rollout methods that return a list, we assume that the returned spans
-            # are the complete span set from the whole rollout
-            trace_spans = raw_result
-
             # Case 2: result is a list of ReadableSpan (OpenTelemetry spans)
             if len(raw_result) > 0 and all(isinstance(t, ReadableSpan) for t in raw_result):
-                if not isinstance(self._tracer, OtelTracer):
-                    for span in raw_result:
-                        await store.add_otel_span(
-                            rollout.rollout_id, rollout.attempt.attempt_id, cast(ReadableSpan, span)
-                        )
-                else:
+                if isinstance(self._tracer, OtelTracer):
                     logger.warning(
                         f"{self._log_prefix(rollout.rollout_id)} Tracer is already an OpenTelemetry tracer. "
                         "The traces should have already been added to the store. "
-                        "No need to return anything from rollout."
+                        "Returning the traces from the rollout will result in duplicate spans."
                     )
+                for span in raw_result:
+                    added_span = await store.add_otel_span(
+                        rollout.rollout_id, rollout.attempt.attempt_id, cast(ReadableSpan, span)
+                    )
+                    if added_span is not None:
+                        trace_spans.append(added_span)
+                    else:
+                        logger.error(
+                            f"{self._log_prefix(rollout.rollout_id)} Failed to add OpenTelemetry span to the store: {span}"
+                        )
                 result_recognized = True
 
             # Case 3: result is a list of Span (agentlightning spans)
@@ -327,7 +335,25 @@ class LitAgentRunner(Runner[T_task]):
                 # Add the spans directly to the store
                 for span in raw_result:
                     await store.add_span(cast(Span, span))
-                trace_spans = raw_result
+                trace_spans = [cast(Span, span) for span in raw_result]
+                result_recognized = True
+
+            # Case 4: result is a list of SpanCoreFields (agentlightning spans)
+            elif len(raw_result) > 0 and all(isinstance(t, SpanCoreFields) for t in raw_result):
+                # Add the spans directly to the store too, but needs to get sequence id first
+                sequence_ids = await store.get_many_span_sequence_ids(
+                    [(rollout.rollout_id, rollout.attempt.attempt_id) for _ in range(len(raw_result))]
+                )
+                trace_spans = [
+                    Span.from_core_fields(
+                        cast(SpanCoreFields, span_core_fields),
+                        rollout_id=rollout.rollout_id,
+                        attempt_id=rollout.attempt.attempt_id,
+                        sequence_id=sequence_id,
+                    )
+                    for span_core_fields, sequence_id in zip(raw_result, sequence_ids, strict=True)
+                ]
+                await store.add_many_spans(trace_spans)
                 result_recognized = True
 
             # Left over cases for list
@@ -336,7 +362,7 @@ class LitAgentRunner(Runner[T_task]):
                     f"{self._log_prefix(rollout.rollout_id)} The rollout returns an empty list. "
                     "Please check your rollout implementation."
                 )
-                trace_spans = raw_result
+                trace_spans = []
                 result_recognized = True
 
             else:

agentlightning/runner/legacy.py

@@ -12,7 +12,7 @@ from agentlightning.client import AgentLightningClient
 from agentlightning.litagent import LitAgent
 from agentlightning.litagent.litagent import is_v0_1_rollout_api
 from agentlightning.tracer.base import Tracer
-from agentlightning.types import RolloutLegacy, RolloutRawResultLegacy, Triplet
+from agentlightning.types import RolloutLegacy, RolloutRawResultLegacy, Span, SpanLike, Triplet
 
 from .base import Runner
 
@@ -99,7 +99,7 @@ class LegacyAgentRunner(Runner[Any]):
         trace: Any = None
         final_reward: Optional[float] = None
         triplets: Optional[List[Triplet]] = None
-        trace_spans: Optional[List[ReadableSpan]] = None
+        trace_spans: Optional[List[SpanLike]] = None
 
         # Handle different types of results from the agent
         # Case 1: result is a float (final reward)
@@ -108,10 +108,14 @@ class LegacyAgentRunner(Runner[Any]):
         # Case 2: result is a list of Triplets
         if isinstance(result, list) and all(isinstance(t, Triplet) for t in result):
             triplets = result  # type: ignore
-        # Case 3: result is a list of ReadableSpan (OpenTelemetry spans)
-        if isinstance(result, list) and all(isinstance(t, ReadableSpan) for t in result):
+        # Case 3.1: result is a list of ReadableSpan (OpenTelemetry spans)
+        if isinstance(result, list) and all(isinstance(t, (ReadableSpan)) for t in result):
             trace_spans = result  # type: ignore
             trace = [json.loads(readable_span.to_json()) for readable_span in trace_spans]  # type: ignore
+        # Case 3.2: result is a list of Span (Agent-lightning spans)
+        if isinstance(result, list) and all(isinstance(t, Span) for t in result):
+            trace_spans = result  # type: ignore
+            trace = [span.model_dump() for span in trace_spans]  # type: ignore
         # Case 4: result is a list of dict (trace JSON)
         if isinstance(result, list) and all(isinstance(t, dict) for t in result):
             trace = result
@@ -123,10 +127,9 @@ class LegacyAgentRunner(Runner[Any]):
 
         # If the agent has tracing enabled, use the tracer's last trace if not already set
         if self.tracer and (trace is None or trace_spans is None):
-            spans = self.tracer.get_last_trace()
-            if spans:
-                trace = [json.loads(readable_span.to_json()) for readable_span in spans]
-                trace_spans = spans
+            trace_spans = self.tracer.get_last_trace()  # type: ignore
+            if trace_spans:
+                trace = [cast(Span, span).model_dump() for span in trace_spans]
 
         # Always extract triplets from the trace using TracerTraceToTriplet
         if trace_spans:

agentlightning/semconv.py

@@ -53,6 +53,9 @@ class LightningResourceAttributes(Enum):
     SPAN_SEQUENCE_ID = "agentlightning.span_sequence_id"
     """Resource name for span sequence ID in Agent-lightning spans."""
 
+    TRACER_NAME = "agentlightning.tracer.name"
+    """Which tracer is used to create this span."""
+
 
 class LightningSpanAttributes(Enum):
     """Attribute names that commonly appear in Agent-lightning spans.

agentlightning/store/collection_based.py

@@ -15,11 +15,9 @@ from __future__ import annotations
 
 import asyncio
 import functools
-import hashlib
 import inspect
 import logging
 import time
-import uuid
 import warnings
 from collections import defaultdict
 from types import CoroutineType
@@ -61,6 +59,7 @@ from agentlightning.types import (
     Worker,
     WorkerStatus,
 )
+from agentlightning.utils.id import generate_id
 from agentlightning.utils.metrics import MetricsBackend
 
 from .base import (
@@ -195,19 +194,16 @@ def healthcheck_before(func: T_callable) -> T_callable:
 
 
 def _generate_resources_id() -> str:
-    short_id = hashlib.sha1(uuid.uuid4().bytes).hexdigest()[:12]
-    return "rs-" + short_id
+    return "rs-" + generate_id(12)
 
 
 def _generate_rollout_id() -> str:
-    short_id = hashlib.sha1(uuid.uuid4().bytes).hexdigest()[:12]
-    return "ro-" + short_id
+    return "ro-" + generate_id(12)
 
 
 def _generate_attempt_id() -> str:
     """We don't need that long because attempts are limited to rollouts."""
-    short_id = hashlib.sha1(uuid.uuid4().bytes).hexdigest()[:8]
-    return "at-" + short_id
+    return "at-" + generate_id(8)
 
 
 class CollectionBasedLightningStore(LightningStore, Generic[T_collections]):

agentlightning/tracer/__init__.py

@@ -1,8 +1,16 @@
 # Copyright (c) Microsoft. All rights reserved.
 
 from .agentops import AgentOpsTracer
-from .base import Tracer
+from .base import Tracer, clear_active_tracer, get_active_tracer, set_active_tracer
+from .dummy import DummyTracer
 from .otel import OtelTracer
-from .weave import WeaveTracer
 
-__all__ = ["AgentOpsTracer", "Tracer", "OtelTracer", "WeaveTracer"]
+__all__ = [
+    "AgentOpsTracer",
+    "Tracer",
+    "OtelTracer",
+    "DummyTracer",
+    "get_active_tracer",
+    "set_active_tracer",
+    "clear_active_tracer",
+]

agentlightning/tracer/agentops.py

@@ -13,12 +13,13 @@ import agentops.sdk.core
 import opentelemetry.trace as trace_api
 from agentops.sdk.core import TracingCore
 from opentelemetry.sdk.trace import TracerProvider as TracerProviderImpl
-from opentelemetry.trace import get_tracer_provider
 from opentelemetry.trace.status import StatusCode
 
 from agentlightning.instrumentation import instrument_all, uninstrument_all
 from agentlightning.store.base import LightningStore
+from agentlightning.utils.otel import get_span_processors, get_tracer_provider
 
+from .base import with_active_tracer_context
 from .otel import LightningSpanProcessor, OtelTracer
 
 if TYPE_CHECKING:
@@ -79,13 +80,20 @@ class AgentOpsTracer(OtelTracer):
                 agentops.init(auto_start_session=False)  # type: ignore
                 logger.info(f"[Worker {worker_id}] AgentOps client initialized.")
             else:
-                logger.warning(f"[Worker {worker_id}] AgentOps client was already initialized.")
+                logger.warning(f"[Worker {worker_id}] AgentOps client was already initialized. Skip initialization.")
 
-        self._lightning_span_processor = LightningSpanProcessor()
-
-        # TODO: The span processor cannot be deleted once added.
-        # This might be a problem if the tracer is entered and exited multiple times.
-        self._get_tracer_provider().add_span_processor(self._lightning_span_processor)  # type: ignore
+        span_processors = get_span_processors(self._get_tracer_provider(), LightningSpanProcessor)
+        if len(span_processors) > 0:
+            logger.warning(
+                "LightningSpanProcessor already present in TracerProvider. You might have called init_worker() multiple times."
+                "Agent-lightning will try to reuse the existing LightningSpanProcessor."
+            )
+            if len(span_processors) > 1:
+                logger.error("More than one LightningSpanProcessors present in TracerProvider. This should not happen.")
+            self._lightning_span_processor = span_processors[0]
+        else:
+            self._lightning_span_processor = LightningSpanProcessor()
+            self._get_tracer_provider().add_span_processor(self._lightning_span_processor)  # type: ignore
 
     def teardown_worker(self, worker_id: int) -> None:
         super().teardown_worker(worker_id)
@@ -94,6 +102,10 @@ class AgentOpsTracer(OtelTracer):
             self.uninstrument(worker_id)
             logger.info(f"[Worker {worker_id}] Instrumentation removed.")
 
+        # NOTE: The teardown doesn't try to remove the LightningSpanProcessor from the TracerProvider.
+        # Currently there is no stable way to fully restore the AgentOps state to the initial state.
+
+    @with_active_tracer_context
     @asynccontextmanager
     async def trace_context(
         self,
@@ -158,7 +170,6 @@ class AgentOpsTracer(OtelTracer):
                 with self._agentops_trace_context(rollout_id, attempt_id, kwargs):
                     yield trace_api.get_tracer(__name__, tracer_provider=tracer_provider)
         elif store is None and rollout_id is None and attempt_id is None:
-            # TODO: Add tests to cover both paths
             self._disable_native_otlp_exporter()
             with self._lightning_span_processor:
                 with self._agentops_trace_context(None, None, kwargs):

agentlightning/tracer/base.py

@@ -2,14 +2,13 @@
 
 from __future__ import annotations
 
+import functools
 import logging
 from contextlib import contextmanager
-from typing import TYPE_CHECKING, Any, AsyncContextManager, Awaitable, Callable, ContextManager, List, Optional
-
-from opentelemetry.sdk.trace import ReadableSpan
+from typing import TYPE_CHECKING, Any, AsyncContextManager, Awaitable, Callable, ContextManager, List, Optional, TypeVar
 
 from agentlightning.store.base import LightningStore
-from agentlightning.types import ParallelWorkerBase
+from agentlightning.types import Attributes, ParallelWorkerBase, Span, SpanCoreFields, SpanRecordingContext, TraceStatus
 
 if TYPE_CHECKING:
     from langchain_core.callbacks.base import BaseCallbackHandler  # type: ignore
@@ -17,6 +16,14 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 
+T = TypeVar("T")
+
+
+_active_tracer: Optional[Tracer] = None
+
+T_func = Callable[..., Awaitable[Any]]
+
+
 class Tracer(ParallelWorkerBase):
     """
     An abstract base class for tracers.
@@ -98,12 +105,12 @@ class Tracer(ParallelWorkerBase):
         """Internal API for CI backward compatibility."""
         raise NotImplementedError()
 
-    def get_last_trace(self) -> List[ReadableSpan]:
+    def get_last_trace(self) -> List[Span]:
         """
         Retrieves the raw list of captured spans from the most recent trace.
 
         Returns:
-            A list of OpenTelemetry `ReadableSpan` objects.
+            A list of [`Span`][agentlightning.Span] objects collected during the last trace.
         """
         raise NotImplementedError()
 
@@ -124,6 +131,48 @@ class Tracer(ParallelWorkerBase):
         with self._trace_context_sync(name=func.__name__):
             return func(*args, **kwargs)
 
+    def create_span(
+        self,
+        name: str,
+        attributes: Optional[Attributes] = None,
+        timestamp: Optional[float] = None,
+        status: Optional[TraceStatus] = None,
+    ) -> SpanCoreFields:
+        """Notify the tracer that a span should be created here.
+
+        It uses a fire-and-forget approach and doesn't wait for the span to be created.
+
+        Args:
+            name: The name of the span.
+            attributes: The attributes of the span.
+            timestamp: The timestamp of the span.
+            status: The status of the span.
+
+        Returns:
+            The core fields of the span.
+        """
+        raise NotImplementedError()
+
+    def operation_context(
+        self,
+        name: str,
+        attributes: Optional[Attributes] = None,
+        start_time: Optional[float] = None,
+        end_time: Optional[float] = None,
+    ) -> ContextManager[SpanRecordingContext]:
+        """Start to record an operation to a span.
+
+        Args:
+            name: The name of the operation.
+            attributes: The attributes of the operation.
+            start_time: The start time of the operation.
+            end_time: The end time of the operation.
+
+        Returns:
+            A [`SpanRecordingContext`][agentlightning.SpanRecordingContext] for recording the operation on the span.
+        """
+        raise NotImplementedError()
+
     async def trace_run_async(self, func: Callable[..., Awaitable[Any]], *args: Any, **kwargs: Any) -> Any:
         """
         A convenience wrapper to trace the execution of a single asynchronous function.
@@ -175,3 +224,64 @@ class Tracer(ParallelWorkerBase):
                 self.teardown_worker(0)
             if has_init:
                 self.teardown()
+
+
+def set_active_tracer(tracer: Tracer):
+    """Set the active tracer for the current process.
+
+    Args:
+        tracer: The tracer to set as active.
+    """
+    global _active_tracer
+    if _active_tracer is not None:
+        raise ValueError("An active tracer is already set. Cannot set a new one.")
+    _active_tracer = tracer
+
+
+def clear_active_tracer():
+    """Clear the active tracer for the current process."""
+    global _active_tracer
+    _active_tracer = None
+
+
+def get_active_tracer() -> Optional[Tracer]:
+    """Get the active tracer for the current process.
+
+    Returns:
+        The active tracer, or None if no tracer is active.
+    """
+    global _active_tracer
+    return _active_tracer
+
+
+class _ActiveTracerAsyncCM(AsyncContextManager[T]):
+    def __init__(self, tracer: Tracer, inner: AsyncContextManager[T]):
+        self._tracer = tracer
+        self._inner = inner
+
+    async def __aenter__(self) -> T:
+        set_active_tracer(self._tracer)  # will raise if nested
+        try:
+            return await self._inner.__aenter__()
+        except Exception:
+            clear_active_tracer()
+            raise
+
+    async def __aexit__(self, *args: Any, **kwargs: Any) -> Optional[bool]:
+        try:
+            return await self._inner.__aexit__(*args, **kwargs)
+        finally:
+            clear_active_tracer()
+
+
+def with_active_tracer_context(
+    func: Callable[..., AsyncContextManager[T]],
+) -> Callable[..., AsyncContextManager[T]]:
+    """Decorate a method returning an AsyncContextManager so tracer is active for the whole `async with`."""
+
+    @functools.wraps(func)
+    def wrapper(self: Tracer, *args: Any, **kwargs: Any) -> AsyncContextManager[T]:
+        cm = func(self, *args, **kwargs)
+        return _ActiveTracerAsyncCM(self, cm)
+
+    return wrapper

agentlightning/tracer/dummy.py

@@ -0,0 +1,106 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from __future__ import annotations
+
+import logging
+import time
+from contextlib import contextmanager
+from typing import (
+    Iterator,
+    Optional,
+)
+
+from agentlightning.types import (
+    Attributes,
+    SpanCoreFields,
+    SpanRecordingContext,
+    StatusCode,
+    TraceStatus,
+)
+from agentlightning.utils.otel import format_exception_attributes
+
+from .base import Tracer
+
+logger = logging.getLogger(__name__)
+
+
+class DummySpanRecordingContext(SpanRecordingContext):
+    """Context for recording operations on a dummy span, not dependent on any backend tracer."""
+
+    def __init__(self, name: str, attributes: Optional[Attributes] = None, start_time: Optional[float] = None) -> None:
+        self.name = name
+        self.attributes = attributes or {}
+        self.start_time = start_time or time.time()
+        self.end_time = None
+        self.status = TraceStatus(status_code="OK")
+
+    def record_exception(self, exception: BaseException) -> None:
+        self.record_status("ERROR", str(exception))
+        self.record_attributes(format_exception_attributes(exception))
+
+    def record_attributes(self, attributes: Attributes) -> None:
+        self.attributes.update(attributes)
+
+    def record_status(self, status_code: StatusCode, description: Optional[str] = None) -> None:
+        self.status = TraceStatus(status_code=status_code, description=description)
+
+    def finalize(self, end_time: Optional[float] = None) -> None:
+        self.end_time = end_time or time.time()
+
+    def get_recorded_span(self) -> SpanCoreFields:
+        if self.end_time is None:
+            raise ValueError("End time is not set. Call finalize() first.")
+        return SpanCoreFields(
+            name=self.name,
+            attributes=self.attributes,
+            start_time=self.start_time,
+            end_time=self.end_time,
+            status=self.status,
+        )
+
+
+class DummyTracer(Tracer):
+    """A dummy tracer that does not trace anything, but it is compatible with the emitter API.
+
+    It doesn't rely on any backend tracer, and also doesn't use any stores.
+    """
+
+    def create_span(
+        self,
+        name: str,
+        attributes: Optional[Attributes] = None,
+        timestamp: Optional[float] = None,
+        status: Optional[TraceStatus] = None,
+    ) -> SpanCoreFields:
+        if attributes is None:
+            attributes = {}
+        if timestamp is None:
+            timestamp = time.time()
+        if status is None:
+            status = TraceStatus(status_code="OK")
+        return SpanCoreFields(
+            name=name,
+            attributes=attributes,
+            start_time=timestamp,
+            end_time=timestamp,
+            status=status,
+        )
+
+    @contextmanager
+    def operation_context(
+        self,
+        name: str,
+        attributes: Optional[Attributes] = None,
+        start_time: Optional[float] = None,
+        end_time: Optional[float] = None,
+    ) -> Iterator[DummySpanRecordingContext]:
+        start_time = start_time or time.time()
+        recording_context = DummySpanRecordingContext(name, attributes, start_time)
+        try:
+            yield recording_context
+        except Exception as exc:
+            recording_context.record_exception(exc)
+            recording_context.record_status("ERROR", str(exc))
+            raise
+        finally:
+            recording_context.finalize(end_time)

agentlightning/tracer/otel.py

@@ -6,27 +6,69 @@ import asyncio
 import logging
 import threading
 import warnings
-from contextlib import asynccontextmanager
-from typing import Any, AsyncGenerator, Awaitable, List, Optional
+from contextlib import asynccontextmanager, contextmanager
+from typing import Any, AsyncGenerator, Awaitable, Iterator, List, Optional
 
 import opentelemetry.trace as trace_api
-from agentops.sdk.core import BatchSpanProcessor
 from opentelemetry.instrumentation.utils import suppress_instrumentation
 from opentelemetry.sdk.resources import Resource
 from opentelemetry.sdk.trace import ReadableSpan, SpanProcessor
-from opentelemetry.sdk.trace import TracerProvider
 from opentelemetry.sdk.trace import TracerProvider as TracerProviderImpl
-from opentelemetry.sdk.trace.export import SimpleSpanProcessor
+from opentelemetry.sdk.trace.export import BatchSpanProcessor, SimpleSpanProcessor
 
 from agentlightning.semconv import LightningResourceAttributes
 from agentlightning.store.base import LightningStore
+from agentlightning.types import Attributes, Span, SpanCoreFields, SpanRecordingContext, StatusCode, TraceStatus
+from agentlightning.types.tracer import convert_timestamp
 from agentlightning.utils.otel import get_tracer_provider
 from agentlightning.utils.otlp import LightningStoreOTLPExporter
 
-from .base import Tracer
+from .base import Tracer, with_active_tracer_context
 
 logger = logging.getLogger(__name__)
 
+STORE_WRITE_TIMEOUT_SECONDS = 10.0
+
+
+def to_otel_status_code(status_code: StatusCode) -> trace_api.StatusCode:
+    if status_code == "UNSET":
+        return trace_api.StatusCode.UNSET
+    elif status_code == "ERROR":
+        return trace_api.StatusCode.ERROR
+    else:
+        return trace_api.StatusCode.OK
+
+
+class OtelSpanRecordingContext(SpanRecordingContext):
+    def __init__(self, span: trace_api.Span) -> None:
+        self._span = span
+
+    def record_exception(self, exception: BaseException) -> None:
+        self._span.record_exception(exception)
+        self.record_status("ERROR", str(exception))
+
+    def record_attributes(self, attributes: Attributes) -> None:
+        self._span.set_attributes(attributes)
+
+    def record_status(self, status_code: StatusCode, description: Optional[str] = None) -> None:
+        otel_status_code = to_otel_status_code(status_code)
+        self._span.set_status(otel_status_code, description)
+
+    def get_otel_span(self) -> trace_api.Span:
+        return self._span
+
+    def get_recorded_span(self) -> SpanCoreFields:
+        if isinstance(self._span, ReadableSpan):
+            return SpanCoreFields(
+                name=self._span.name,
+                attributes=dict(self._span.attributes) if self._span.attributes else {},
+                start_time=convert_timestamp(self._span.start_time),
+                end_time=convert_timestamp(self._span.end_time),
+                status=TraceStatus.from_opentelemetry(self._span.status),
+            )
+        else:
+            raise ValueError(f"Span is not a ReadableSpan: {self._span}")
+
 
 class OtelTracer(Tracer):
     """Tracer that provides a basic OpenTelemetry tracer provider.
@@ -38,7 +80,7 @@ class OtelTracer(Tracer):
     def __init__(self):
         super().__init__()
         # This provider is only initialized when the worker is initialized.
-        self._tracer_provider: Optional[TracerProvider] = None
+        self._tracer_provider: Optional[trace_api.TracerProvider] = None
         self._lightning_span_processor: Optional[LightningSpanProcessor] = None
         self._simple_span_processor: Optional[SimpleSpanProcessor] = None
         self._otlp_span_exporter: Optional[LightningStoreOTLPExporter] = None
@@ -63,7 +105,7 @@ class OtelTracer(Tracer):
         except RuntimeError:
             logger.debug(f"[Worker {worker_id}] Tracer provider is not initialized by OtelTracer. Initializing it now.")
 
-        self._tracer_provider = TracerProvider()
+        self._tracer_provider = TracerProviderImpl()
         trace_api.set_tracer_provider(self._tracer_provider)
         self._lightning_span_processor = LightningSpanProcessor()
         self._tracer_provider.add_span_processor(self._lightning_span_processor)
@@ -78,6 +120,7 @@ class OtelTracer(Tracer):
         super().teardown_worker(worker_id)
         logger.info(f"[Worker {worker_id}] Tearing down OpenTelemetry tracer does NOT remove the tracer provider.")
 
+    @with_active_tracer_context
     @asynccontextmanager
     async def trace_context(
         self,
@@ -129,12 +172,69 @@ class OtelTracer(Tracer):
         else:
             raise ValueError("rollout_id and attempt_id must be either all provided or all None")
 
-    def get_last_trace(self) -> List[ReadableSpan]:
+    def create_span(
+        self,
+        name: str,
+        attributes: Optional[Attributes] = None,
+        timestamp: Optional[float] = None,
+        status: Optional[TraceStatus] = None,
+    ) -> SpanCoreFields:
+        # Fire the span to the current active tracer provider.
+        tracer_provider = self._get_tracer_provider()
+        tracer = tracer_provider.get_tracer(__name__)
+        span = tracer.start_span(
+            name, attributes=attributes, start_time=int(timestamp * 1_000_000_000) if timestamp else None
+        )
+        if status is not None:
+            span.set_status(to_otel_status_code(status.status_code), status.description)
+        span.end(int(timestamp * 1_000_000_000) if timestamp else None)
+
+        # The span should have been auto-created by now.
+        # Return the core fields of the span.
+        if isinstance(span, ReadableSpan):
+            return SpanCoreFields(
+                name=name,
+                attributes=dict(span.attributes) if span.attributes else {},
+                start_time=convert_timestamp(span.start_time),
+                end_time=convert_timestamp(span.end_time),
+                status=TraceStatus.from_opentelemetry(span.status),
+            )
+        else:
+            raise ValueError(f"Span is not a ReadableSpan: {span}")
+
+    @contextmanager
+    def operation_context(
+        self,
+        name: str,
+        attributes: Optional[Attributes] = None,
+        start_time: Optional[float] = None,
+        end_time: Optional[float] = None,
+    ) -> Iterator[SpanRecordingContext]:
+        if end_time is not None:
+            logger.warning("OpenTelemetry doesn't support customizing the end time of a span. End time is ignored.")
+        # Record the span to the current active tracer provider.
+        tracer_provider = self._get_tracer_provider()
+        tracer = tracer_provider.get_tracer(__name__)
+
+        # Activate the span as the current span within otel.
+        with tracer.start_as_current_span(
+            name, attributes=attributes, start_time=int(start_time * 1_000_000_000) if start_time else None
+        ) as span:
+            recording_context = OtelSpanRecordingContext(span)
+            try:
+                yield recording_context
+            except Exception as exc:
+                recording_context.record_exception(exc)
+                raise
+
+        # No need to retrieve the span here. It's already been sent to otel processor.
+
+    def get_last_trace(self) -> List[Span]:
         """
         Retrieves the raw list of captured spans from the most recent trace.
 
         Returns:
-            A list of OpenTelemetry `ReadableSpan` objects.
+            A list of [`Span`][agentlightning.Span] objects captured during the most recent trace.
         """
         if not self._lightning_span_processor:
             raise RuntimeError("LightningSpanProcessor is not initialized. Call init_worker() first.")
@@ -143,6 +243,8 @@ class OtelTracer(Tracer):
     def _get_tracer_provider(self) -> TracerProviderImpl:
         if self._tracer_provider is None:
             raise RuntimeError("TracerProvider is not initialized. Call init_worker() first.")
+        if not isinstance(self._tracer_provider, TracerProviderImpl):
+            raise TypeError(f"TracerProvider is not a opentelemetry.sdk.trace.TracerProvider: {self._tracer_provider}")
         return self._tracer_provider
 
     def _enable_native_otlp_exporter(self, store: LightningStore, rollout_id: str, attempt_id: str):
@@ -215,12 +317,13 @@ class LightningSpanProcessor(SpanProcessor):
 
     def __init__(self, disable_store_submission: bool = False):
         self._disable_store_submission: bool = disable_store_submission
-        self._spans: List[ReadableSpan] = []
+        self._spans: List[Span] = []
 
         # Store related context and states
         self._store: Optional[LightningStore] = None
         self._rollout_id: Optional[str] = None
         self._attempt_id: Optional[str] = None
+        self._local_sequence_id: int = 0
         self._lock = threading.Lock()
 
         # private asyncio loop running in a daemon thread
@@ -330,13 +433,13 @@ class LightningSpanProcessor(SpanProcessor):
     def force_flush(self, timeout_millis: int = 30000) -> bool:
         return True
 
-    def spans(self) -> List[ReadableSpan]:
+    def spans(self) -> List[Span]:
         """
         Get the list of spans collected by this processor.
         This is useful for debugging and testing purposes.
 
         Returns:
-            List of ReadableSpan objects collected during tracing.
+            List of [`Span`][agentlightning.Span] objects collected during tracing.
         """
         return self._spans
 
@@ -373,12 +476,46 @@ class LightningSpanProcessor(SpanProcessor):
                 # Submit add_otel_span to the event loop and wait for it to complete
                 with suppress_instrumentation():
                     self._ensure_loop()
-                    self._await_in_loop(
+                    uploaded_span = self._await_in_loop(
                         self._store.add_otel_span(self._rollout_id, self._attempt_id, span),
-                        timeout=60.0,
+                        timeout=STORE_WRITE_TIMEOUT_SECONDS,
                     )
+                    if uploaded_span is not None:
+                        self._spans.append(uploaded_span)
+            except TimeoutError:
+                logger.warning(
+                    "Timed out adding span %s to store after %.1f seconds. The span will be stored locally "
+                    "but it's not guaranteed to be persisted.",
+                    span.name,
+                    STORE_WRITE_TIMEOUT_SECONDS,
+                )
+                self._spans.append(
+                    Span.from_opentelemetry(
+                        span,
+                        rollout_id=self._rollout_id,
+                        attempt_id=self._attempt_id,
+                        sequence_id=self._local_sequence_id,
+                    )
+                )
             except Exception:
                 # log; on_end MUST NOT raise
-                logger.exception(f"Error adding span to store: {span.name}")
+                logger.exception(f"Error adding span to store: {span.name}. The span will be store locally only.")
+                self._spans.append(
+                    Span.from_opentelemetry(
+                        span,
+                        rollout_id=self._rollout_id,
+                        attempt_id=self._attempt_id,
+                        sequence_id=self._local_sequence_id,
+                    )
+                )
 
-        self._spans.append(span)
+        else:
+            # Fallback path
+            created_span = Span.from_opentelemetry(
+                span,
+                rollout_id=self._rollout_id or "rollout-dummy",
+                attempt_id=self._attempt_id or "attempt-dummy",
+                sequence_id=self._local_sequence_id,
+            )
+            self._local_sequence_id += 1
+            self._spans.append(created_span)

agentlightning/tracer/weave.py

@@ -2,63 +2,252 @@
 
 from __future__ import annotations
 
+import asyncio
+import concurrent.futures as futures
 import logging
 import os
-from contextlib import asynccontextmanager
-from typing import TYPE_CHECKING, Any, AsyncIterator, Dict, List, Optional, Tuple, Union
+import re
+import weakref
+from contextlib import asynccontextmanager, contextmanager
+from datetime import datetime
+from typing import (
+    Any,
+    AsyncIterator,
+    Callable,
+    Dict,
+    Iterator,
+    List,
+    Optional,
+    cast,
+)
 
-from agentlightning.instrumentation import instrument_weave, uninstrument_weave
+import weave
+from opentelemetry.semconv.attributes import exception_attributes
+from weave.trace.call import Call
+from weave.trace.settings import UserSettings
+from weave.trace.weave_client import WeaveClient
+from weave.trace_server import trace_server_interface as tsi
+from weave.wandb_interface.context import set_wandb_api_context
+
+from agentlightning.instrumentation.weave import InMemoryWeaveTraceServer, instrument_weave, uninstrument_weave
+from agentlightning.semconv import LightningResourceAttributes, LightningSpanAttributes
 from agentlightning.store.base import LightningStore
-from agentlightning.types.tracer import OtelResource, Span, SpanContext, TraceStatus
+from agentlightning.types import (
+    Attributes,
+    OtelResource,
+    Span,
+    SpanContext,
+    SpanCoreFields,
+    SpanRecordingContext,
+    StatusCode,
+    TraceStatus,
+)
+from agentlightning.utils.id import generate_id
+from agentlightning.utils.otel import (
+    filter_and_unflatten_attributes,
+    flatten_attributes,
+    format_exception_attributes,
+    sanitize_attributes,
+)
 
-from .base import Tracer
-
-if TYPE_CHECKING:
-    from weave.trace.call import Call  # type: ignore
-
-JSONPrimitive = Union[str, int, float, bool, None]
+from .base import Tracer, with_active_tracer_context
 
 logger = logging.getLogger(__name__)
 
 
-class WeaveTracer(Tracer):
+def op_name_to_func_name(op_name: str) -> str:
+    """Convert a Weave operation name to a function name.
+
+    Weave operation names look like this: `weave:///xxx/agentlightning.tracer.weave/op/openai.chat.completions.create:019b10be-...-44d74272569c`
     """
-    Tracer implementation using Weave for telemetry and trace logging.
+    match = re.search(r"/([^/:]+):", op_name)
+    if match:
+        return match.group(1)
+    else:
+        return op_name
+
+
+def random_project_name() -> str:
+    return "agl/weave-" + generate_id(12)
+
+
+def get_timestamp_or_throw(date: Optional[datetime], field_name: str) -> float:
+    if date is None:
+        raise ValueError(f"{field_name} is required but not set")
+    return date.timestamp()
+
+
+class WeaveSpanRecordingContext(SpanRecordingContext):
+    """Universal interface for recording operations on a Weave call."""
+
+    def __init__(self, call: Call) -> None:
+        self._call = call
+
+    def record_exception(self, exception: BaseException) -> None:
+        self._call.exception = str(exception)
+        self.record_status("ERROR", str(exception))
+        self.record_attributes(format_exception_attributes(exception))
+
+    def _get_input_from_attributes(self, attributes: Attributes) -> Dict[str, Any]:
+        if LightningSpanAttributes.OPERATION_INPUT.value in attributes:
+            # This can be a very rare case. If it happens, we can just let it throw.
+            return cast(Dict[str, Any], attributes[LightningSpanAttributes.OPERATION_INPUT.value])
+        else:
+            filtered_attributes = filter_and_unflatten_attributes(
+                attributes, LightningSpanAttributes.OPERATION_INPUT.value
+            )
+            if isinstance(filtered_attributes, list):
+                return {str(i): v for i, v in enumerate(filtered_attributes)}
+            else:
+                return filtered_attributes
+
+    def _get_output_from_attributes(self, attributes: Attributes) -> Any:
+        if LightningSpanAttributes.OPERATION_OUTPUT.value in attributes:
+            return attributes[LightningSpanAttributes.OPERATION_OUTPUT.value]
+        else:
+            return filter_and_unflatten_attributes(attributes, LightningSpanAttributes.OPERATION_OUTPUT.value)
+
+    def record_attributes(self, attributes: Attributes) -> None:
+        input_attributes = self._get_input_from_attributes(attributes)
+        if input_attributes:
+            self._call.inputs.update(input_attributes)
+
+        output_attributes = self._get_output_from_attributes(attributes)
+        if output_attributes:
+            if self._call.output is not None:
+                logger.warning(f"Output is already set. It will be overridden: {self._call.output}")
+            self._call.output = output_attributes
+
+        if LightningSpanAttributes.OPERATION_NAME.value in attributes:
+            logger.error(
+                f"Cannot record operation name as an attribute. It will be skipped: {attributes[LightningSpanAttributes.OPERATION_NAME.value]}"
+            )
+
+        # The rest of the attributes are recorded as summary.
+        for key, value in attributes.items():
+            if (
+                not key == LightningSpanAttributes.OPERATION_INPUT.value
+                and not key.startswith(LightningSpanAttributes.OPERATION_INPUT.value + ".")
+                and not key == LightningSpanAttributes.OPERATION_OUTPUT.value
+                and not key.startswith(LightningSpanAttributes.OPERATION_OUTPUT.value + ".")
+                and not key == LightningSpanAttributes.OPERATION_NAME.value
+            ):
+                if self._call.summary is None:
+                    self._call.summary = {}
+                self._call.summary[key] = value
+
+    def record_status(self, status_code: StatusCode, description: Optional[str] = None) -> None:
+        if status_code == "ERROR":
+            if not description:
+                raise ValueError("Description is required when status code is ERROR")
+            self._call.exception = description
+        elif status_code == "OK":
+            self._call.exception = None
+        # Do nothing for other status codes.
+
+    def finalize(self) -> None:
+        # Do nothing
+        pass
+
+    def get_recorded_span(self) -> SpanCoreFields:
+        return SpanCoreFields(
+            name=self._call.op_name,
+            attributes=flatten_attributes(self._call.attributes or {}),
+            start_time=self._call.started_at.timestamp() if self._call.started_at else None,
+            end_time=self._call.ended_at.timestamp() if self._call.ended_at else None,
+            status=TraceStatus(
+                status_code="OK" if self._call.exception is None else "ERROR", description=self._call.exception
+            ),
+        )
+
+
+class WeaveTracerManagedTraceServer(InMemoryWeaveTraceServer):
+    """A managed trace server for WeaveTracer."""
+
+    def __init__(
+        self,
+        partial_call_callback: Callable[[Dict[str, Any]], None],
+        complete_call_callback: Callable[[tsi.CallSchema], None],
+    ):
+        super().__init__()
+        self.partial_call_callback = partial_call_callback
+        self.complete_call_callback = complete_call_callback
+
+    def trigger_callbacks(self, call_id: str) -> None:
+        with self._call_threading_lock:
+            if call_id in self.calls:
+                self.complete_call_callback(self.calls[call_id])
+            elif call_id in self.partial_calls:
+                self.partial_call_callback(self.partial_calls[call_id])
+            else:
+                logger.error(f"Call {call_id} not found in partial_calls or calls")
+
+    def call_start(self, req: tsi.CallStartReq) -> tsi.CallStartRes:
+        try:
+            ret = super().call_start(req)
+            self.trigger_callbacks(ret.id)
+            return ret
+        except Exception:
+            logger.exception(f"Error calling call_start: {req}", exc_info=True)
+            raise
+
+    def call_end(self, req: tsi.CallEndReq) -> tsi.CallEndRes:
+        try:
+            ret = super().call_end(req)
+            self.trigger_callbacks(req.end.id)
+            return ret
+        except Exception:
+            logger.exception(f"Error calling call_end: {req}", exc_info=True)
+            raise
+
+
+class WeaveTracer(Tracer):
+    """Tracer implementation using Weave for telemetry and trace logging.
 
     This replaces AgentOpsTracer with a Weave-based manual trace context. It tracks:
+
     - Function/method calls
     - Input/Output data
     - Exceptions
-    and logs them to Weave Cloud (W&B backend) or optionally bypasses the network for testing.
 
-    Attributes:
-        project_name: Name of the Weave project. Used to initialize the Weave client.
-        _store: Optional LightningStore instance for storing collected spans.
-        instrument_managed: Whether to patch the Weave/W&B integration to bypass actual network calls for testing.
+    and logs them to Weave Cloud (W&B backend) or optionally bypasses the network for testing.
     """
 
     def __init__(
-        self, *, project_name: str | None = None, wandb_api_key: str | None = None, instrument_managed: bool = True
+        self,
+        *,
+        project_name: str | None = None,
+        weave_user_settings: UserSettings | None = None,
+        instrument_managed: bool = True,
     ):
-        """
-        Initialize a WeaveTracer instance.
+        """Initialize a WeaveTracer instance.
 
         Args:
             project_name: Optional project name for Weave; defaults to the current module name.
-            wandb_api_key: Optional W&B API key; sets environment variable if provided.
+            weave_user_settings: Optional UserSettings for Weave.
             instrument_managed: Whether to patch the Weave/W&B integration to bypass actual network calls for testing.
         """
         super().__init__()
-        self.project_name = project_name or __name__
-        self.sequence_id = 0
-        self._store: Optional[LightningStore] = None
+        self.project_name = project_name
         self.instrument_managed = instrument_managed
+        self.weave_user_settings = weave_user_settings or UserSettings(use_server_cache=False)
 
-        if wandb_api_key:
-            os.environ["WANDB_API_KEY"] = wandb_api_key
+        self._store: Optional[LightningStore] = None
+        self._server = WeaveTracerManagedTraceServer(
+            partial_call_callback=self.partial_call_callback, complete_call_callback=self.complete_call_callback
+        )
+
+        self._default_sequence_counter: int = 0
+        self._calls: Dict[str, tsi.CallSchema] = {}  # call_id -> call
+        self._spans: List[Span] = []  # spans in the current trace
+        self._rollout_id: Optional[str] = None
+        self._attempt_id: Optional[str] = None
+        self._partial_call_futures: Dict[str, asyncio.Future[int] | futures.Future[int]] = {}
+        self._complete_call_futures: List[asyncio.Future[None] | futures.Future[None]] = []
+        self._loop: weakref.ReferenceType[asyncio.AbstractEventLoop] | None = None
 
     def instrument(self, worker_id: int):
-        instrument_weave()
+        instrument_weave(self._server)
 
     def uninstrument(self, worker_id: int):
         uninstrument_weave()
@@ -75,22 +264,34 @@ class WeaveTracer(Tracer):
         logger.info(f"[Worker {worker_id}] Setting up Weave tracer...")
         self._store = store
 
-        try:
-            import weave
-        except ImportError:
-            raise RuntimeError("Weave is not installed. Install it to use WeaveTracer.")
-
         # Optionally patch network calls to bypass real Weave/W&B endpoints
         if self.instrument_managed:
             self.instrument(worker_id)
 
-        # Initialize the Weave client if not already initialized
-        if weave.get_client() is None:  # type: ignore
-            try:
-                weave.init(project_name=self.project_name)  # type: ignore
-                logger.info(f"[Worker {worker_id}] Weave client initialized.")
-            except Exception as e:
-                raise RuntimeError(f"Failed to initialize Weave for project '{self.project_name}': {e}")
+        # If WANDB_API_KEY is not set, we need to initialize Weave with a hack
+        if not os.getenv("WANDB_API_KEY"):
+            logger.info("WANDB_API_KEY is not set. Initializing Weave a mock context.")
+            set_wandb_api_context("agl", api_key=None, headers=None, cookies=None)
+        else:
+            logger.debug("WANDB_API_KEY is set. Weave will be initialized automatically.")
+
+        weave_client = weave.get_client()
+        if self.project_name is None:
+            self.project_name = random_project_name()
+
+        if weave_client is not None:
+            logger.warning("Weave client was already initialized. Reentrant calls are at your own risk.")
+            if weave_client.project == self.project_name:
+                logger.error(
+                    f"Weave client was already initialized for the same project '{self.project_name}'. It's very likely that weave won't work correctly."
+                )
+
+        # Init no matter what
+        try:
+            weave.init(project_name=self.project_name, settings=self.weave_user_settings)
+            logger.info(f"[Worker {worker_id}] Weave client initialized.")
+        except Exception as exc:
+            raise RuntimeError(f"Failed to initialize Weave for project '{self.project_name}'") from exc
 
     def teardown_worker(self, worker_id: int):
         """
@@ -105,21 +306,20 @@ class WeaveTracer(Tracer):
             self.uninstrument(worker_id)
             logger.info(f"[Worker {worker_id}] Instrumentation removed.")
 
+    @with_active_tracer_context
     @asynccontextmanager
     async def trace_context(
         self,
         name: Optional[str] = None,
         *,
-        store: Optional[LightningStore] = None,
         rollout_id: Optional[str] = None,
         attempt_id: Optional[str] = None,
+        **kwargs: Any,
     ) -> AsyncIterator[Any]:
-        """
-        Synchronous implementation of the tracing context.
+        """Asynchronous implementation of the tracing context.
 
         Args:
             name: Optional operation name.
-            store: Optional LightningStore instance.
             rollout_id: Optional rollout ID.
             attempt_id: Optional attempt ID.
 
@@ -127,181 +327,337 @@ class WeaveTracer(Tracer):
             ValueError: If store, rollout_id, and attempt_id are inconsistently provided.
             RuntimeError: If Weave is not installed or client is uninitialized.
         """
-        arg_op = name or self.project_name
-        arg_inputs: dict[str, str] | None = {"rollout_id": rollout_id or "", "attempt_id": attempt_id or ""}
 
-        if store is not None and rollout_id is not None and attempt_id is not None:
+        if rollout_id is not None and attempt_id is not None:
             self._rollout_id = rollout_id
             self._attempt_id = attempt_id
-            self._store = store
+        elif rollout_id is None and attempt_id is None:
+            logger.info("No rollout_id or attempt_id provided. Skipping writing to store.")
+            self._rollout_id = self._attempt_id = None
         else:
-            raise ValueError("store, rollout_id, and attempt_id must be either all provided")
+            raise ValueError("rollout_id and attempt_id must be either both provided or both None")
+
+        await self._init_trace_context()
+
+        weave_client = self._get_weave_client()
+
+        if weave_client.server is not self._server:
+            logger.error(
+                "Weave client is not using the correct trace server. You might have multiple WeaveTracer instances running in the same process. "
+                f"Expected {self._server}, got {weave_client.server}"
+            )
+
+        arg_op = name or weave_client.project
+        arg_inputs: dict[str, str] = {}
+        if rollout_id is not None:
+            arg_inputs[LightningResourceAttributes.ROLLOUT_ID.value] = rollout_id
+        if attempt_id is not None:
+            arg_inputs[LightningResourceAttributes.ATTEMPT_ID.value] = attempt_id
 
         try:
-            import datetime
+            # Create a new trace call object in Weave
+            trace_call = weave_client.create_call(  # pyright: ignore[reportUnknownMemberType]
+                op=arg_op, inputs=arg_inputs
+            )
 
-            import weave
-        except ImportError:
-            raise RuntimeError("Weave is not installed. Install it to use WeaveTracer.")
+            try:
+                yield trace_call
+                # Finish trace even if no exception
+                weave_client.finish_call(trace_call)  # pyright: ignore[reportUnknownMemberType]
+            except Exception as exc:
+                # Finish trace and log any exception
+                weave_client.finish_call(trace_call, exception=exc)  # pyright: ignore[reportUnknownMemberType]
+                logger.error(f"Trace failed for rollout_id={rollout_id}, attempt_id={attempt_id}, error={exc}")
+                raise
 
-        weave_client = weave.get_client()  # type: ignore
+        finally:
+            try:
+                weave_client.flush()
+                # It's possible that the call end futures are from a dedicated Weave thread pool,
+                await asyncio.gather(*[asyncio.wrap_future(future) for future in self._complete_call_futures])
+
+            finally:
+                # Mandatory cleanup
+                self._rollout_id = None
+                self._attempt_id = None
+
+    def create_span(
+        self,
+        name: str,
+        attributes: Optional[Attributes] = None,
+        timestamp: Optional[float] = None,
+        status: Optional[TraceStatus] = None,
+    ) -> SpanCoreFields:
+        if timestamp is not None:
+            logger.warning("Weave doesn't support customizing the start time of a call. Timestamp is ignored.")
+        weave_client = self._get_weave_client()
+        trace_call = weave_client.create_call(  # pyright: ignore[reportUnknownMemberType]
+            op=name,
+            attributes=attributes,
+            inputs={},
+        )
+        # Immediately finish the call
+        weave_client.finish_call(trace_call)  # pyright: ignore[reportUnknownMemberType]
+        # We don't wait for the call to be propagated to the server.
+        start_time = trace_call.started_at.timestamp() if trace_call.started_at else None
+        end_time = trace_call.ended_at.timestamp() if trace_call.ended_at else None
+        trace_status = (
+            TraceStatus(status_code="OK")
+            if trace_call.exception is None
+            else TraceStatus(status_code="ERROR", description=trace_call.exception)
+        )
+        return SpanCoreFields(
+            name=name,
+            attributes=flatten_attributes(trace_call.attributes or {}),
+            start_time=start_time,
+            end_time=end_time,
+            status=trace_status,
+        )
+
+    @contextmanager
+    def operation_context(
+        self,
+        name: str,
+        attributes: Optional[Attributes] = None,
+        start_time: Optional[float] = None,
+        end_time: Optional[float] = None,
+    ) -> Iterator[SpanRecordingContext]:
+        if start_time is not None:
+            logger.warning("Weave doesn't support customizing the start time of a call. Timestamp is ignored.")
+        if end_time is not None:
+            logger.warning("Weave doesn't support customizing the end time of a call. Timestamp is ignored.")
+        weave_client = self._get_weave_client()
+        trace_call = weave_client.create_call(  # pyright: ignore[reportUnknownMemberType]
+            op=name,
+            attributes=attributes,
+            inputs={},
+        )
+        recording_context = WeaveSpanRecordingContext(trace_call)
+        try:
+            yield recording_context
+        except Exception as exc:
+            recording_context.record_exception(exc)
+            raise
+        finally:
+            weave_client.finish_call(trace_call)  # pyright: ignore[reportUnknownMemberType]
+
+    async def _init_trace_context(self) -> None:
+        """Initialize the trace context."""
+        self._spans.clear()
+        self._calls.clear()
+        self._partial_call_futures.clear()
+        self._complete_call_futures.clear()
+        self._loop = weakref.ref(asyncio.get_running_loop())
+
+    def _get_weave_client(self) -> WeaveClient:
+        """Get the Weave client."""
+        weave_client = weave.get_client()
         if not weave_client:
             raise RuntimeError("Weave client is not initialized. Call init_worker() first.")
+        return weave_client
 
-        # Create a new trace call object in Weave
-        trace_call = weave_client.create_call(op=arg_op, inputs=arg_inputs)  # type: ignore
-        trace_call.started_at = datetime.datetime.now(tz=datetime.timezone.utc)
+    def _ensure_loop(self) -> tuple[asyncio.AbstractEventLoop, bool]:
+        """Returns a usable event loop and a boolean indicating whether it's the current running loop.
 
-        try:
-            yield trace_call
-        except Exception as e:
-            # Finish trace and log any exception
-            weave_client.finish_call(trace_call, exception=e)  # type: ignore
-            logger.error(f"Trace failed for rollout_id={rollout_id}, attempt_id={attempt_id}, error={e}")
-        finally:
-            # Finish trace even if no exception
-            weave_client.finish_call(trace_call)  # type: ignore
-            await self._on_finish_handler(trace_call)  # type: ignore
-
-    async def _on_finish_handler(self, call: "Call", *args: Any, **kwargs: Any) -> None:  # type: ignore
+        Prefer using the main loop if it's possible. Otherwise, use the current running loop.
         """
-        Handler called when a Weave Call finishes.
+        # Get the current running loop
+        try:
+            running_loop = asyncio.get_running_loop()
+        except RuntimeError:
+            running_loop = None
+
+        # Get the main loop, which can be a different loop
+        if self._loop is not None:
+            main_loop = self._loop()
+        else:
+            main_loop = None
+
+        if main_loop is not None:
+            return main_loop, id(main_loop) == id(running_loop)
+        elif running_loop is not None:
+            return running_loop, True
+        else:
+            raise RuntimeError("No running event loop found. This should not happen.")
+
+    def get_last_trace(self) -> List[Span]:
+        return self._spans
+
+    def partial_call_callback(self, request_content: Dict[str, Any]) -> None:
+        call_id = request_content.get("id")
+        if call_id is None:
+            raise ValueError("Call ID is required even for partial calls")
+
+        if call_id in self._partial_call_futures:
+            raise ValueError(f"Call {call_id} already has a start future")
+
+        # The callback must possibly be called from a dedicated Weave thread pool,
+        # but it should be executed on the main event loop.
+        try:
+            loop, is_current_loop = self._ensure_loop()
+            if is_current_loop:
+                task = loop.create_task(self.partial_call_handler(request_content))
+            else:
+                # Schedule the task on the dedicated loop
+                task = asyncio.run_coroutine_threadsafe(self.partial_call_handler(request_content), loop)
+            self._partial_call_futures[call_id] = task
+        except Exception as exc:
+            logger.exception(f"Error creating call start task: {exc}", exc_info=True)
+
+    def complete_call_callback(self, call: tsi.CallSchema) -> None:
+        try:
+            loop, is_current_loop = self._ensure_loop()
+            if is_current_loop:
+                task = loop.create_task(self.complete_call_handler(call))
+            else:
+                # Schedule the task on the dedicated loop
+                task = asyncio.run_coroutine_threadsafe(self.complete_call_handler(call), loop)
+            self._complete_call_futures.append(task)
+        except Exception as exc:
+            logger.exception(f"Error creating call finish task: {exc}", exc_info=True)
+
+    async def _get_next_sequence_id(self) -> int:
+        """Get the next sequence ID for a span.
+
+        Use store to get the next sequence ID if available, otherwise use a default counter.
+        """
+        if self._rollout_id and self._attempt_id and self._store:
+            return await self._store.get_next_span_sequence_id(self._rollout_id, self._attempt_id)
+        else:
+            self._default_sequence_counter += 1
+            return self._default_sequence_counter
+
+    async def partial_call_handler(self, request_content: Dict[str, Any]) -> int:
+        """Handler called when a Weave Call starts.
+
+        Args:
+            request_content: The partial Weave Call object.
+
+        Returns:
+            The sequence ID for the call.
+        """
+        sequence_id = await self._get_next_sequence_id()
+        return sequence_id
+
+    async def complete_call_handler(self, call: tsi.CallSchema) -> None:
+        """Handler called when a Weave Call finishes.
 
         Converts the call (including nested children) into spans and stores them in LightningStore.
         """
-        spans, self.sequence_id = self.convert_call_to_spans(call, self._rollout_id, self._attempt_id, self.sequence_id)  # type: ignore
+        # Make sure the corresponding call_start_future is complete
+        if call.id in self._partial_call_futures:
+            sequence_id = await asyncio.wrap_future(self._partial_call_futures[call.id])
+            del self._partial_call_futures[call.id]
+        else:
+            # Fetch a new sequence ID as the call_start is somehow missing
+            logger.warning(f"Call {call.id} has no start future. Fetching a new sequence ID.")
+            sequence_id = await self._get_next_sequence_id()
 
+        self._calls[call.id] = call
+
+        span = await self.convert_call_to_span(call, self._rollout_id, self._attempt_id, sequence_id)
+        self._spans.append(span)
         if self._store and self._rollout_id and self._attempt_id:
             try:
-                await self._store.add_many_spans(spans)
-            except Exception as e:
-                logger.exception(f"Error adding span to store: {e}")
+                await self._store.add_span(span)
+            except Exception as exc:
+                logger.exception(f"Error adding span to store: {exc}")
 
-    def convert_call_to_spans(
+    async def convert_call_to_span(
         self,
-        call: "Call",  # type: ignore
+        call: tsi.CallSchema,
         rollout_id: Optional[str] = None,
         attempt_id: Optional[str] = None,
-        seq_start: int = 0,
-    ) -> tuple[List[Span], int]:
-        """
-        Recursively convert a Weave Call (with nested children) into a flat list of Agent Lightning Spans.
+        sequence_id: Optional[int] = None,
+    ) -> Span:
+        """Convert a Weave Call (with nested children) into a Agent-lightning Span.
+
+        `rollout_id` and `attempt_id` are required to attach the spans to the store.
 
         Args:
             call: The Weave Call object.
             rollout_id: Optional rollout ID to attach to spans.
             attempt_id: Optional attempt ID to attach to spans.
-            seq_start: Sequence number to start from.
+            sequence_id: Optional sequence ID to attach to spans.
 
         Returns:
-            Tuple of (list_of_spans, next_sequence_id).
+            List of converted spans.
         """
-        spans: List[Span] = []
-        sequence_id = seq_start
+        rollout_id = rollout_id or "rollout-dummy"
+        attempt_id = attempt_id or "attempt-dummy"
+        sequence_id = sequence_id or 0
 
-        rollout_id = rollout_id or ""  # type: ignore
-        attempt_id = attempt_id or ""  # type: ignore
+        start_ts: float = call.started_at.timestamp()
+        end_ts: Optional[float] = call.ended_at.timestamp() if call.ended_at else None
 
-        start_dt = getattr(call, "started_at", None)  # type: ignore
-        start_ts: Optional[float] = start_dt.timestamp() if start_dt else None
+        if call.exception:
+            status = TraceStatus(status_code="ERROR", description=call.exception)
+        else:
+            status = TraceStatus(status_code="OK")
 
-        end_dt = getattr(call, "ended_at", None)  # type: ignore
-        end_ts: Optional[float] = end_dt.timestamp() if end_dt else None
+        attributes: Dict[str, Any] = {
+            LightningSpanAttributes.OPERATION_NAME.value: call.op_name,
+            # op_name can be possibly overridden by the attributes.
+            **call.attributes,
+        }
+        if call.inputs:
+            attributes[LightningSpanAttributes.OPERATION_INPUT.value] = call.inputs
+        if call.output:
+            attributes[LightningSpanAttributes.OPERATION_OUTPUT.value] = call.output
+        if call.summary:
+            # attributes can be possibly overridden by the summary.
+            attributes.update(call.summary)
+        if call.exception:
+            attributes[exception_attributes.EXCEPTION_MESSAGE] = call.exception
 
-        trace_id = str(getattr(call, "trace_id", None))  # type: ignore
-        span_id = str(getattr(call, "id", None))  # type: ignore
-        parent_id = str(getattr(call, "parent_id", None)) if getattr(call, "parent_id", None) else None  # type: ignore
-
-        exception = getattr(call, "exception", None)  # type: ignore
-        status_code = "ERROR" if exception else "OK"
-
-        def sanitize(
-            inputs: Dict[str, Any],
-            output: Dict[str, Any],
-        ) -> Dict[str, str | JSONPrimitive]:
-            stack: List[Tuple[Any, str]] = [
-                (inputs or {}, "input"),
-                (output or {}, "output"),
-            ]
-
-            attributes: Dict[str, str | JSONPrimitive] = {}
-
-            while stack:
-                value, key = stack.pop()
-
-                if isinstance(value, dict):
-                    for k, v in value.items():  # type: ignore
-                        stack.append((v, f"{key}.{k}"))  # type: ignore
-                elif isinstance(value, (list, tuple)):
-                    for i, v in enumerate(value):  # type: ignore
-                        stack.append((v, f"{key}.{i}"))  # type: ignore
-                else:
-                    if value is None:
-                        attributes[key] = "None"
-                    elif isinstance(value, (str, int, float, bool)):
-                        attributes[key] = value
-                    else:
-                        try:
-                            attributes[key] = str(value)
-                        except Exception:
-                            attributes[key] = "None"
-
-            return attributes
-
-        inputs = getattr(call, "inputs", {})  # type: ignore
-        output = getattr(call, "output", {})  # type: ignore
-        attributes = sanitize(inputs, output)
+        sanitized_attributes = sanitize_attributes(flatten_attributes(attributes, expand_leaf_lists=False))
 
         context = SpanContext(
-            trace_id=trace_id,
-            span_id=span_id,
+            trace_id=call.trace_id,
+            span_id=call.id,
             is_remote=False,
             trace_state={},
         )
 
-        parent_context = (
-            SpanContext(
-                trace_id=trace_id,
-                span_id=parent_id,
-                is_remote=False,
-                trace_state={},
-            )
-            if parent_id
-            else None
-        )
+        # Get context for parent
+        if call.parent_id:
+            parent_call = self._calls.get(call.parent_id)
+            if parent_call:
+                parent_context = SpanContext(
+                    trace_id=parent_call.trace_id,
+                    span_id=parent_call.id,
+                    is_remote=False,
+                    trace_state={},
+                )
+            else:
+                parent_context = None
+        else:
+            parent_context = None
 
         # Build the Span object
-        span = Span(
-            rollout_id=rollout_id or "",
-            attempt_id=attempt_id or "",
+        return Span(
+            rollout_id=rollout_id,
+            attempt_id=attempt_id,
             sequence_id=sequence_id,
-            trace_id=trace_id,
-            span_id=span_id,
-            parent_id=parent_id,
-            name=getattr(call, "func_name", "unknown"),  # type: ignore
-            status=TraceStatus(status_code=status_code),
-            attributes=attributes,  # type: ignore
+            trace_id=call.trace_id,
+            span_id=call.id,
+            parent_id=call.parent_id,
+            name=op_name_to_func_name(call.op_name),
+            status=status,
+            attributes=sanitized_attributes,
             events=[],  # Weave calls do not generate events
             links=[],  # Weave calls do not generate links
             start_time=start_ts,
             end_time=end_ts,
             context=context,
             parent=parent_context,
-            resource=OtelResource(attributes={}, schema_url=""),
+            resource=OtelResource(
+                attributes={
+                    LightningResourceAttributes.ROLLOUT_ID.value: rollout_id,
+                    LightningResourceAttributes.ATTEMPT_ID.value: attempt_id,
+                    LightningResourceAttributes.SPAN_SEQUENCE_ID.value: sequence_id,
+                    LightningResourceAttributes.TRACER_NAME.value: "weave",
+                },
+                schema_url="",
+            ),
         )
-
-        spans.append(span)
-        sequence_id += 1
-
-        children: List["Call"] = getattr(call, "_children", [])  # type: ignore
-        # Recursively process child calls
-        for child in children:  # type: ignore
-            child_spans, sequence_id = self.convert_call_to_spans(  # type: ignore
-                child,  # type: ignore
-                rollout_id=rollout_id,
-                attempt_id=attempt_id,
-                seq_start=sequence_id,
-            )
-            spans.extend(child_spans)
-
-        return spans, sequence_id

agentlightning/types/core.py

@@ -28,7 +28,7 @@ from typing import (
 from opentelemetry.sdk.trace import ReadableSpan
 from pydantic import BaseModel, Field, model_validator
 
-from .tracer import Span
+from .tracer import Span, SpanCoreFields
 
 if TYPE_CHECKING:
     from agentlightning.litagent import LitAgent
@@ -307,6 +307,7 @@ RolloutRawResult = Union[
     float,  # only final reward
     List[ReadableSpan],  # constructed OTEL spans by user
     List[Span],  # constructed Span objects by user
+    List[SpanCoreFields],  # constructed SpanCoreFields objects by user
 ]
 """Rollout result type.
 

agentlightning/types/tracer.py

@@ -2,11 +2,13 @@
 
 from __future__ import annotations
 
+import time
+
 """Data models that mirror OpenTelemetry spans for Agent Lightning."""
 
 import json
 from enum import Enum
-from typing import Any, Dict, List, Optional, Sequence, Union
+from typing import Any, Dict, List, Literal, Optional, Protocol, Sequence, Union
 
 from opentelemetry import trace as trace_api
 from opentelemetry.sdk.resources import Resource
@@ -31,6 +33,9 @@ __all__ = [
     "SpanNames",
     "SpanAttributeNames",
     "SpanLike",
+    "StatusCode",
+    "SpanCoreFields",
+    "SpanRecordingContext",
 ]
 
 
@@ -83,6 +88,8 @@ Attributes = Dict[str, AttributeValue]
 """Mapping from attribute names to their values. Same as OpenTelemetry `Attributes` type."""
 TraceState = Dict[str, str]
 """Mapping from trace state key to its value. Same as OpenTelemetry `TraceState` type."""
+StatusCode = Literal["UNSET", "OK", "ERROR"]
+"""The status code of the span."""
 
 
 class SpanContext(BaseModel):
@@ -115,7 +122,7 @@ class SpanContext(BaseModel):
 class TraceStatus(BaseModel):
     """Serializable variant of `opentelemetry.trace.Status`."""
 
-    status_code: str
+    status_code: StatusCode
     """The status code of the span. Same as OpenTelemetry `Status.status_code` type."""
     description: Optional[str] = None
     """The description of the span. Same as OpenTelemetry `Status.description` type."""
@@ -203,6 +210,44 @@ class OtelResource(BaseModel):
         )
 
 
+class SpanCoreFields(BaseModel):
+    """Core fields of a span. Used by span creators who don't care about the full span model.
+
+    If the spans are managed by some OTel tracer provider, it's not advised to create spans via this path.
+    """
+
+    name: str
+    """The name of the span."""
+    status: TraceStatus
+    """The status of the span."""
+    attributes: Attributes
+    """The attributes of the span."""
+    start_time: Optional[float]
+    """The start time of the span."""
+    end_time: Optional[float]
+    """The end time of the span."""
+
+
+class SpanRecordingContext(Protocol):
+    """Context for recording operations on a span. It doesn't have to finalize the span; the caller will do it."""
+
+    def record_exception(self, exception: BaseException) -> None:
+        """Record an exception on the span."""
+        raise NotImplementedError()
+
+    def record_attributes(self, attributes: Attributes) -> None:
+        """Record attributes on the span."""
+        raise NotImplementedError()
+
+    def record_status(self, status_code: StatusCode, description: Optional[str] = None) -> None:
+        """Record the status of the span."""
+        raise NotImplementedError()
+
+    def get_recorded_span(self) -> SpanCoreFields:
+        """Get the recording of the span."""
+        raise NotImplementedError()
+
+
 class Span(BaseModel):
     """Agent Lightning's canonical span model used for persistence and analytics.
 
@@ -340,6 +385,7 @@ class Span(BaseModel):
         start_time: Optional[float] = None,
         end_time: Optional[float] = None,
         resource: Optional[OtelResource] = None,
+        status: Optional[TraceStatus] = None,
     ) -> "Span":
         """Build a synthetic span from raw attributes.
         Different from the [`from_opentelemetry`][agentlightning.Span.from_opentelemetry] method,
@@ -357,6 +403,7 @@ class Span(BaseModel):
             start_time: Span start timestamp in seconds.
             end_time: Span end timestamp in seconds.
             resource: Explicit resource information to attach to the span.
+            status: Optional status of the span.
 
         Returns:
             [`Span`][agentlightning.Span] populated with the provided attributes.
@@ -384,7 +431,7 @@ class Span(BaseModel):
             name=name or AGL_VIRTUAL,
             resource=resource or OtelResource(attributes={}, schema_url=""),
             attributes=attributes,
-            status=TraceStatus(status_code="OK"),
+            status=status or TraceStatus(status_code="OK"),
             events=[],
             links=[],
             parent=(
@@ -399,6 +446,37 @@ class Span(BaseModel):
             ),
         )
 
+    @classmethod
+    def from_core_fields(
+        cls,
+        core: SpanCoreFields,
+        *,
+        rollout_id: Optional[str] = None,
+        attempt_id: Optional[str] = None,
+        sequence_id: Optional[int] = None,
+    ) -> Span:
+        """Build a span from a core span.
+
+        Args:
+            core: Core span to build from.
+            rollout_id: Optional rollout identifier associated with the span.
+            attempt_id: Optional attempt identifier associated with the span.
+            sequence_id: Optional sequence number to preserve ordering.
+
+        Returns:
+            [`Span`][agentlightning.Span] populated with the provided attributes.
+        """
+        return cls.from_attributes(
+            attributes=core.attributes,
+            rollout_id=rollout_id,
+            attempt_id=attempt_id,
+            sequence_id=sequence_id,
+            name=core.name,
+            start_time=core.start_time or time.time(),
+            end_time=core.end_time,
+            status=core.status,
+        )
+
 
 class SpanNames(str, Enum):
     """Enumerated span names recognised by Agent-lightning. Deprecated in favor of [semconv][agentlightning.semconv]."""

agentlightning/utils/id.py

@@ -0,0 +1,10 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import hashlib
+import uuid
+
+__all__ = ["generate_id"]
+
+
+def generate_id(length: int) -> str:
+    return hashlib.sha1(uuid.uuid4().bytes).hexdigest()[:length]

agentlightning/utils/otel.py

@@ -2,22 +2,25 @@
 
 """Utilities shared for OpenTelemetry span (attributes) support."""
 
+import json
 import logging
-from typing import Any, Dict, List, Sequence, Union, cast
+import traceback
+from typing import Any, Dict, List, Sequence, Type, TypeVar, Union, cast
 from warnings import filterwarnings
 
 import opentelemetry.trace as trace_api
 from agentops.sdk.exporters import OTLPSpanExporter
-from opentelemetry.sdk.trace import ReadableSpan, SpanLimits, SynchronousMultiSpanProcessor, Tracer
+from opentelemetry.sdk.trace import ReadableSpan, SpanLimits, SpanProcessor, SynchronousMultiSpanProcessor, Tracer
 from opentelemetry.sdk.trace import TracerProvider as TracerProviderImpl
 from opentelemetry.sdk.trace.export import BatchSpanProcessor, SimpleSpanProcessor
 from opentelemetry.sdk.util.instrumentation import InstrumentationInfo, InstrumentationScope
+from opentelemetry.semconv.attributes import exception_attributes
 from opentelemetry.trace import get_tracer_provider as otel_get_tracer_provider
 from pydantic import TypeAdapter
 
 from agentlightning.env_var import LightningEnvVar, resolve_bool_env_var
 from agentlightning.semconv import LightningSpanAttributes, LinkAttributes, LinkPydanticModel
-from agentlightning.types import SpanLike
+from agentlightning.types import Attributes, AttributeValue, SpanLike
 from agentlightning.utils.otlp import LightningStoreOTLPExporter
 
 logger = logging.getLogger(__name__)
@@ -37,6 +40,9 @@ __all__ = [
     "unflatten_attributes",
 ]
 
+T_SpanLike = TypeVar("T_SpanLike", bound=SpanLike)
+T_SpanProcessor = TypeVar("T_SpanProcessor", bound=SpanProcessor)
+
 
 def full_qualified_name(obj: type) -> str:
     if str(obj.__module__) == "builtins":
@@ -112,6 +118,25 @@ def get_tracer_provider(inspect: bool = True) -> TracerProviderImpl:
     return tracer_provider
 
 
+def get_span_processors(
+    tracer_provider: TracerProviderImpl, expected_type: Type[T_SpanProcessor]
+) -> List[T_SpanProcessor]:
+    """Get the span processors from the tracer provider.
+
+    Args:
+        tracer_provider: The tracer provider to get the span processors from.
+        expected_type: The type of the span processors to get.
+
+    Returns:
+        A list of span processors of the expected type.
+    """
+    processors: List[T_SpanProcessor] = []
+    for processor in tracer_provider._active_span_processor._span_processors:  # pyright: ignore[reportPrivateUsage]
+        if isinstance(processor, expected_type):
+            processors.append(processor)
+    return processors
+
+
 def get_tracer(use_active_span_processor: bool = True) -> trace_api.Tracer:
     """Resolve the OpenTelemetry tracer configured for Agent Lightning.
 
@@ -166,7 +191,7 @@ def make_tag_attributes(tags: List[str]) -> Dict[str, Any]:
     ["gen_ai.model:gpt-4", "reward.extrinsic"]
     ```
     """
-    return flatten_attributes({LightningSpanAttributes.TAG.value: tags})
+    return flatten_attributes({LightningSpanAttributes.TAG.value: tags}, expand_leaf_lists=True)
 
 
 def extract_tags_from_attributes(attributes: Dict[str, Any]) -> List[str]:
@@ -196,10 +221,10 @@ def make_link_attributes(links: Dict[str, str]) -> Dict[str, Any]:
         if not isinstance(value, str):  # pyright: ignore[reportUnnecessaryIsInstance]
             raise ValueError(f"Link value must be a string, got {type(value)} for key '{key}'")
         link_list.append({LinkAttributes.KEY_MATCH.value: key, LinkAttributes.VALUE_MATCH.value: value})
-    return flatten_attributes({LightningSpanAttributes.LINK.value: link_list})
+    return flatten_attributes({LightningSpanAttributes.LINK.value: link_list}, expand_leaf_lists=True)
 
 
-def query_linked_spans(spans: Sequence[SpanLike], links: List[LinkPydanticModel]) -> List[SpanLike]:
+def query_linked_spans(spans: Sequence[T_SpanLike], links: List[LinkPydanticModel]) -> List[T_SpanLike]:
     """Query spans that are linked by the given link attributes.
 
     Args:
@@ -209,7 +234,7 @@ def query_linked_spans(spans: Sequence[SpanLike], links: List[LinkPydanticModel]
     Returns:
         A list of spans that match the given link attributes.
     """
-    matched_spans: List[SpanLike] = []
+    matched_spans: List[T_SpanLike] = []
 
     for span in spans:
         span_attributes = span.attributes or {}
@@ -294,7 +319,9 @@ def filter_and_unflatten_attributes(attributes: Dict[str, Any], prefix: str) ->
     return unflatten_attributes(stripped_attributes)
 
 
-def flatten_attributes(nested_data: Union[Dict[str, Any], List[Any]]) -> Dict[str, Any]:
+def flatten_attributes(
+    nested_data: Union[Dict[str, Any], List[Any]], *, expand_leaf_lists: bool = False
+) -> Dict[str, Any]:
     """Flatten a nested dictionary or list into a flat dictionary with dotted keys.
 
     This function recursively traverses dictionaries and lists, producing a flat
@@ -303,12 +330,14 @@ def flatten_attributes(nested_data: Union[Dict[str, Any], List[Any]]) -> Dict[st
 
     Example:
 
-        >>> flatten_attributes({"a": {"b": 1, "c": [2, 3]}})
+        >>> flatten_attributes({"a": {"b": 1, "c": [2, 3]}}, expand_leaf_lists=True)
         {"a.b": 1, "a.c.0": 2, "a.c.1": 3}
 
     Args:
-        nested_data: A nested structure composed of dictionaries, lists, or
-            primitive values.
+        nested_data: A nested structure composed of dictionaries, lists, or primitive values.
+        expand_leaf_lists: Whether to expand lists composed only of primitive values.
+            When `False` (the default), lists of str/int/float/bool are treated as
+            leaf values and stored without enumerating their indices.
 
     Returns:
         A flat dictionary mapping dotted-string paths to primitive values.
@@ -316,6 +345,15 @@ def flatten_attributes(nested_data: Union[Dict[str, Any], List[Any]]) -> Dict[st
 
     flat: Dict[str, Any] = {}
 
+    def _primitive_type(value: Any) -> Union[type[str], type[int], type[float], type[bool]]:
+        if isinstance(value, bool):
+            return bool
+        if isinstance(value, int):
+            return int
+        if isinstance(value, float):
+            return float
+        return str
+
     def _walk(value: Any, prefix: str = "") -> None:
         if isinstance(value, dict):
             for k, v in cast(Dict[Any, Any], value).items():
@@ -326,7 +364,22 @@ def flatten_attributes(nested_data: Union[Dict[str, Any], List[Any]]) -> Dict[st
                 new_prefix = f"{prefix}.{k}" if prefix else k
                 _walk(v, new_prefix)
         elif isinstance(value, list):
-            for idx, item in enumerate(cast(List[Any], value)):
+            maybe_list = cast(List[Any], value)
+            is_leaf_candidate = bool(maybe_list) and all(
+                isinstance(item, (str, int, float, bool)) for item in maybe_list
+            )
+            if not expand_leaf_lists and is_leaf_candidate and prefix:
+                primitive_types = {_primitive_type(item) for item in maybe_list}
+                if len(primitive_types) == 1:
+                    flat[prefix] = maybe_list
+                    return
+                logger.warning(
+                    "List attribute '%s' contains mixed primitive types %s; expanding indexed keys instead.",
+                    prefix,
+                    primitive_types,
+                )
+
+            for idx, item in enumerate(maybe_list):
                 new_prefix = f"{prefix}.{idx}" if prefix else str(idx)
                 _walk(item, new_prefix)
         else:
@@ -399,3 +452,80 @@ def unflatten_attributes(flat_data: Dict[str, Any]) -> Union[Dict[str, Any], Lis
         return node
 
     return convert(root)
+
+
+def sanitize_attribute_value(object: Any) -> AttributeValue:
+    """Sanitize an attribute value to be a valid OpenTelemetry attribute value."""
+    if isinstance(object, (str, int, float, bool)):
+        return object
+
+    if isinstance(object, list):
+        try:
+            return sanitize_list_attribute_sanity(cast(List[Any], object))
+        except ValueError as exc:
+            logger.warning(f"Failed to sanitize list attribute. Fallback to JSON serialization: {exc}")
+
+    try:
+        # This include null, dict, etc.
+        serialized = json.dumps(object)
+    except (TypeError, ValueError) as exc:
+        raise ValueError(f"Object must be JSON serializable, got: {type(cast(Any, object))}.") from exc
+    return serialized
+
+
+def sanitize_attributes(attributes: Dict[str, Any]) -> Attributes:
+    """Sanitize a dictionary of attributes to be a valid OpenTelemetry attributes."""
+    result: Attributes = {}
+    for k, v in attributes.items():
+        try:
+            result[k] = sanitize_attribute_value(v)
+        except ValueError as exc:
+            raise ValueError(f"Failed to sanitize attribute '{k}': {exc}") from exc
+    return result
+
+
+def sanitize_list_attribute_sanity(maybe_list: List[Any]) -> AttributeValue:
+    """Try to sanitize a list of attributes to be a valid OpenTelemetry attribute value.
+
+    Raise error if the list contains multiple types of primitive values.
+    """
+    if all(isinstance(item, str) for item in maybe_list):
+        return list[str](maybe_list)
+    if all(isinstance(item, bool) for item in maybe_list):
+        return list[bool](maybe_list)
+    if all(isinstance(item, (int, bool)) for item in maybe_list):
+        return [int(item) for item in maybe_list]
+    if all(isinstance(item, (float, int, bool)) for item in maybe_list):
+        return [float(item) for item in maybe_list]
+
+    list_types: List[Any] = [type(item) for item in maybe_list]
+    raise ValueError(f"List must contain only one type of primitive values, got: {set(list_types)}.")
+
+
+def check_attributes_sanity(attributes: Dict[Any, Any]) -> None:
+    """Check if a dictionary of attributes is a valid OpenTelemetry attributes."""
+    for k, v in attributes.items():
+        if not isinstance(k, str):
+            raise ValueError(f"Attribute key must be a string, got {type(k)} for key '{k}'")
+        if isinstance(v, list):
+            try:
+                sanitize_list_attribute_sanity(cast(List[Any], v))
+            except ValueError as exc:
+                raise ValueError(f"Failed to sanitize list attribute '{k}': {exc}") from exc
+        elif not isinstance(v, (str, int, float, bool)):
+            raise ValueError(
+                f"Attribute value must be a string, int, float, bool, or list of these, got {type(v)} for value '{v}'"
+            )
+
+
+def format_exception_attributes(exception: BaseException) -> Attributes:
+    """Format an exception into a dictionary of attributes."""
+    stacktrace = "".join(traceback.format_exception(type(exception), exception, exception.__traceback__))
+    span_attributes: Attributes = {
+        exception_attributes.EXCEPTION_TYPE: type(exception).__name__,
+        exception_attributes.EXCEPTION_MESSAGE: str(exception),
+        exception_attributes.EXCEPTION_ESCAPED: True,
+    }
+    if stacktrace.strip():
+        span_attributes[exception_attributes.EXCEPTION_STACKTRACE] = stacktrace
+    return span_attributes

agentlightning/utils/otlp.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 import gzip
 import logging
-from typing import Any, Awaitable, Callable, Dict, List, Optional, Sequence, Tuple, Type, TypeVar
+from typing import Any, Awaitable, Callable, Dict, List, Mapping, Optional, Sequence, Tuple, Type, TypeVar
 
 from fastapi import Request, Response
 from google.protobuf import json_format
@@ -39,6 +39,7 @@ from agentlightning.types.tracer import (
     OtelResource,
     Span,
     SpanContext,
+    StatusCode,
     TraceStatus,
     convert_timestamp,
 )
@@ -413,7 +414,7 @@ def _kv_list_to_dict(kvs: Sequence[KeyValue]) -> Attributes:
     return {kv.key: _any_value_to_python(kv.value) for kv in kvs}
 
 
-_STATUS_CODE_MAP = {
+_STATUS_CODE_MAP: Mapping[ProtoStatus.StatusCode.ValueType, StatusCode] = {
     ProtoStatus.STATUS_CODE_UNSET: "UNSET",
     ProtoStatus.STATUS_CODE_OK: "OK",
     ProtoStatus.STATUS_CODE_ERROR: "ERROR",

docs/reference/runner.md

@@ -17,3 +17,13 @@
 ::: agentlightning.OtelTracer
 
 ::: agentlightning.Tracer
+
+::: agentlightning.tracer.weave.WeaveTracer
+
+::: agentlightning.DummyTracer
+
+::: agentlightning.set_active_tracer
+
+::: agentlightning.get_active_tracer
+
+::: agentlightning.clear_active_tracer

docs/reference/types.md

@@ -82,6 +82,10 @@
 
 ::: agentlightning.SpanLike
 
+::: agentlightning.SpanCoreFields
+
+::: agentlightning.SpanRecordingContext
+
 ## Semantic Conventions
 
 ::: agentlightning.semconv

docs/tutorials/write-agents.md

@@ -223,6 +223,9 @@ Here are the primary emitter functions:
 * [`emit_message(message: str)`][agentlightning.emit_message]: Records a simple log message as a span.
 * [`emit_exception(exception: BaseException)`][agentlightning.emit_exception]: Records a Python exception, including its type, message, and stack trace.
 * [`emit_object(obj: Any)`][agentlightning.emit_object]: Records any JSON-serializable object, perfect for structured data.
+
+Each helper accepts nested `attributes` (or keyword arguments, in the case of [`operation`][agentlightning.operation]) and automatically flattens/sanitizes them into dotted OpenTelemetry keys. That means you can pass ordinary dictionaries/lists without pre-processing and still get consistent attribute names such as `meta.tag` across [`emit_annotation`][agentlightning.emit_annotation], [`emit_message`][agentlightning.emit_message], [`emit_object`][agentlightning.emit_object], [`emit_exception`][agentlightning.emit_exception], [`emit_reward`][agentlightning.emit_reward], and [`operation`][agentlightning.operation]. All emitter helpers also support a `propagate` flag; setting `propagate=False` keeps the span local—useful for offline tests—while the default `True` streams spans through the active tracer/exporters.
+
 Let's see an example of an agent using these emitters to provide detailed feedback.
 
 ```python

examples/minimal/write_traces.py

@@ -10,19 +10,37 @@ Prior to running this example with `--use-client` flag, please start a Lightning
 ```bash
 agl store --port 45993 --log-level DEBUG
 ```
+
+The CLI also ships an `operation` mode showing how to record a synthetic operation span with
+[`operation`][agentlightning.operation], build link attributes via
+[`make_link_attributes`][agentlightning.utils.otel.make_link_attributes], tag the
+follow-up reward with [`make_tag_attributes`][agentlightning.utils.otel.make_tag_attributes],
+emit a reward span tied back to that operation, and then verify the recorded spans by
+extracting rewards, tags, and links from the store using `agentlightning.utils.otel` helpers.
 """
 
 import argparse
 import asyncio
+import random
 import time
-from typing import Sequence
+from typing import Any, Dict, List, Sequence
+from uuid import uuid4
 
 from openai import AsyncOpenAI
 from rich.console import Console
 
-from agentlightning import AgentOpsTracer, LightningStoreClient, OtelTracer, Span, emit_reward, setup_logging
+from agentlightning import AgentOpsTracer, LightningStoreClient, OtelTracer, Span, emit_reward, operation, setup_logging
+from agentlightning.semconv import AGL_OPERATION, LightningSpanAttributes
 from agentlightning.store import InMemoryLightningStore
-from agentlightning.utils.otel import get_tracer_provider
+from agentlightning.utils.otel import (
+    extract_links_from_attributes,
+    extract_tags_from_attributes,
+    filter_and_unflatten_attributes,
+    get_tracer_provider,
+    make_link_attributes,
+    make_tag_attributes,
+    query_linked_spans,
+)
 
 console = Console()
 
@@ -173,10 +191,111 @@ async def _verify_agentops_traces(spans: Sequence[Span], use_client: bool = Fals
             assert span.attributes["agentops.span.kind"] == "session"
 
 
+async def send_operation_links(use_client: bool = False) -> None:
+    """Demonstrate operation spans wired to reward annotations and verify the stored spans."""
+
+    tracer = OtelTracer()
+    if not use_client:
+        store = InMemoryLightningStore()
+    else:
+        store = LightningStoreClient("http://localhost:45993")
+    conversation_id = "chat-42"
+    tags: Sequence[str] = ("demo.operation", "reward.positive")
+    reward_value = 0.9
+    operation_id = f"{conversation_id}-{uuid4().hex[:8]}"
+    rollout = await store.start_rollout(input={"origin": "write_traces_operation"})
+
+    with tracer.lifespan(store):
+        async with tracer.trace_context(
+            "operation-demo", store=store, rollout_id=rollout.rollout_id, attempt_id=rollout.attempt.attempt_id
+        ):
+            console.print(f"[operation] recording span conversation={conversation_id} operation_id={operation_id}")
+            with operation(conversation_id=conversation_id, operation_id=operation_id) as op_ctx:
+                op_ctx.set_input(
+                    task={"conversation_id": conversation_id},
+                    metadata={"operation_id": operation_id},
+                )
+                synthetic_payload = {
+                    "operation_id": operation_id,
+                    "status": "ok",
+                    "latency_seconds": round(random.uniform(0.05, 0.2), 3),
+                }
+                await asyncio.sleep(0.05)
+                op_ctx.set_output(synthetic_payload)
+
+            link_attrs = make_link_attributes({"conversation_id": conversation_id, "operation_id": operation_id})
+            tag_attrs = make_tag_attributes(list(tags))
+            emit_reward(
+                reward_value,
+                attributes={**link_attrs, **tag_attrs},
+            )
+
+    spans = await store.query_spans(rollout_id=rollout.rollout_id)
+    console.print(spans)
+    _verify_operation_spans(spans, conversation_id, operation_id, tags, reward_value)
+
+    if isinstance(store, LightningStoreClient):
+        await store.close()
+
+
+def _verify_operation_spans(
+    spans: Sequence[Span],
+    conversation_id: str,
+    operation_id: str,
+    tags: Sequence[str],
+    expected_reward: float,
+) -> None:
+    """Verify spans recorded by the operation demo using OTEL helpers."""
+
+    operation_spans = [span for span in spans if span.name == AGL_OPERATION]
+    if not operation_spans:
+        raise RuntimeError("No operation spans recorded.")
+    console.print(f"[verify] found {len(operation_spans)} operation spans")
+
+    reward_span: Span | None = None
+    reward_payload: List[Dict[str, Any]] = []
+    for span in spans:
+        flattened = dict(span.attributes or {})
+        reward_section = filter_and_unflatten_attributes(flattened, LightningSpanAttributes.REWARD.value)
+        if reward_section:
+            reward_span = span
+            if isinstance(reward_section, list):
+                reward_payload = [dict(item) for item in reward_section]  # type: ignore[arg-type]
+            else:
+                reward_payload = [dict(reward_section)]  # type: ignore[arg-type]
+            break
+
+    if reward_span is None or not reward_payload:
+        raise RuntimeError("No reward span recorded for operation demo.")
+
+    primary_reward = reward_payload[0].get("value")
+    console.print(f"[verify] reward dimensions: {reward_payload}")
+    if primary_reward != expected_reward:
+        raise AssertionError(f"Expected reward {expected_reward}, observed {primary_reward}")
+
+    reward_attributes = dict(reward_span.attributes or {})
+    extracted_tags = extract_tags_from_attributes(reward_attributes)
+    console.print(f"[verify] reward tags: {extracted_tags}")
+    for tag in tags:
+        if tag not in extracted_tags:
+            raise AssertionError(f"Missing tag '{tag}' on reward span")
+
+    link_models = extract_links_from_attributes(reward_attributes)
+    matches = query_linked_spans(operation_spans, link_models)
+    if not matches:
+        raise AssertionError("No operation span matched the reward links")
+    console.print(f"[verify] reward links resolved spans: {[span.span_id for span in matches]}")
+
+    linked_attrs = dict(matches[0].attributes or {})
+    if linked_attrs.get("conversation_id") != conversation_id or linked_attrs.get("operation_id") != operation_id:
+        raise AssertionError("Linked operation span attributes do not match expected identifiers")
+    console.print("[verify] linked operation span attributes validated")
+
+
 def main():
     setup_logging("DEBUG")
     parser = argparse.ArgumentParser()
-    parser.add_argument("mode", choices=["otel", "agentops"])
+    parser.add_argument("mode", choices=["otel", "agentops", "operation"])
     parser.add_argument("--use-client", action="store_true")
     args = parser.parse_args()
 
@@ -184,6 +303,8 @@ def main():
         asyncio.run(send_traces_via_otel(use_client=args.use_client))
     elif args.mode == "agentops":
         asyncio.run(send_traces_via_agentops(use_client=args.use_client))
+    elif args.mode == "operation":
+        asyncio.run(send_operation_links(use_client=args.use_client))
     else:
         raise ValueError(f"Invalid mode: {args.mode}")
 

pyproject.toml

@@ -39,7 +39,7 @@ verl = [
 ]
 
 weave = [
-  "weave",
+  "weave>=0.52.22",
 ]
 
 # Store-related dependencies.
@@ -288,6 +288,8 @@ override-dependencies = [
   # verl's numpy<2.0.0 constraint is related to Docker images, not code incompatibility
   # Lock to 2.3.0 because numba relies on numpy 2.2
   "numpy>=2.0.0,<2.3.0",
+  # vllm relies on setuptools<80, but polyfile-weave depends on setuptools>=80.9.0
+  "setuptools>=80.9.0",
 ]
 
 [tool.uv.sources]
@@ -347,6 +349,7 @@ markers = [
   "openai: tests that require OpenAI API",
   "gpu: tests that require GPU",
   "agentops: tests that require AgentOps",
+  "weave: tests that require Weave",
   "llmproxy: tests that require LiteLLM",
   "mongo: tests that require MongoDB",
   "store: tests for agentlightning.store module",

tests/common/tracer.py

@@ -1,9 +1,16 @@
 # Copyright (c) Microsoft. All rights reserved.
 
+from __future__ import annotations
+
+from typing import Optional
+
 import agentops
 import agentops.sdk.core as agentops_core
 import opentelemetry.trace as trace_api
 
+from agentlightning.tracer.dummy import DummyTracer
+from agentlightning.types import Attributes, SpanCoreFields, TraceStatus
+
 
 # pyright: reportPrivateUsage=false
 def clear_tracer_provider() -> None:
@@ -33,3 +40,22 @@ def clear_agentops_init() -> None:
     """Make agentops.init() runnable again."""
     agentops.get_client().initialized = False
     agentops_core.tracer._initialized = False
+
+
+class RecordingDummyTracer(DummyTracer):
+    """Dummy tracer that captures the most recent span request for assertions."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.last_span: Optional[SpanCoreFields] = None
+
+    def create_span(
+        self,
+        name: str,
+        attributes: Optional[Attributes] = None,
+        timestamp: Optional[float] = None,
+        status: Optional[TraceStatus] = None,
+    ) -> SpanCoreFields:
+        span = super().create_span(name, attributes, timestamp, status)
+        self.last_span = span
+        return span

tests/emitter/test_annotation.py

@@ -2,71 +2,48 @@
 
 from __future__ import annotations
 
-from typing import Any, Dict
+from typing import Dict
 
-import opentelemetry.trace as trace_api
 import pytest
-from opentelemetry.sdk.resources import Resource
-from opentelemetry.sdk.trace import ReadableSpan
-from opentelemetry.trace import TraceFlags
 
 from agentlightning.emitter import annotation as annotation_module
 from agentlightning.emitter.annotation import emit_annotation
 from agentlightning.semconv import AGL_ANNOTATION
 
-
-class DummyReadableSpan(ReadableSpan):
-    def __init__(self) -> None:
-        super().__init__(
-            name="dummy",
-            context=trace_api.SpanContext(
-                trace_id=0x1,
-                span_id=0x2,
-                is_remote=False,
-                trace_flags=TraceFlags(TraceFlags.SAMPLED),
-                trace_state=trace_api.TraceState(),
-            ),
-            resource=Resource.create({}),
-        )
-
-    def __enter__(self) -> "DummyReadableSpan":
-        return self
-
-    def __exit__(self, exc_type: Any, exc: Any, tb: Any) -> bool:
-        return False
+from ..common.tracer import RecordingDummyTracer
 
 
-class DummyTracer:
-    def __init__(self, span: DummyReadableSpan) -> None:
-        self._span = span
-        self.last_name: str | None = None
-        self.last_attributes: Dict[str, Any] | None = None
-
-    def start_span(self, name: str, attributes: Dict[str, Any] | None = None) -> DummyReadableSpan:
-        self.last_name = name
-        self.last_attributes = attributes or {}
-        return self._span
+def _install_tracer(monkeypatch: pytest.MonkeyPatch) -> RecordingDummyTracer:
+    tracer = RecordingDummyTracer()
+    monkeypatch.setattr(annotation_module, "get_active_tracer", lambda: tracer)
+    return tracer
 
 
-def test_emit_annotation_flattens_and_respects_propagation(monkeypatch: pytest.MonkeyPatch) -> None:
-    span = DummyReadableSpan()
-    tracer = DummyTracer(span)
-    captured: Dict[str, Any] = {}
+def test_emit_annotation_flattens_and_sanitizes_attributes(monkeypatch: pytest.MonkeyPatch) -> None:
+    tracer = _install_tracer(monkeypatch)
 
-    def fake_get_tracer(*_: Any, **kwargs: Any) -> DummyTracer:
-        captured["propagate"] = kwargs.get("use_active_span_processor")
-        return tracer
+    result = emit_annotation({"meta": {"tag": "foo"}, "score": 1.5})
 
-    monkeypatch.setattr(annotation_module, "get_tracer", fake_get_tracer)
+    assert result.name == AGL_ANNOTATION
+    assert tracer.last_span is not None
+    assert tracer.last_span.attributes == {"meta.tag": "foo", "score": 1.5}
 
-    result = emit_annotation({"meta": {"tag": "foo"}, "score": 1.5}, propagate=False)
 
-    assert result is span
-    assert captured["propagate"] is False
-    assert tracer.last_name == AGL_ANNOTATION
-    assert tracer.last_attributes == {"meta.tag": "foo", "score": 1.5}
+def test_emit_annotation_propagate_false_bypasses_active_tracer(monkeypatch: pytest.MonkeyPatch) -> None:
+    captured: Dict[str, bool] = {"called": False}
+
+    def fail_get_active_tracer() -> RecordingDummyTracer:
+        captured["called"] = True
+        raise AssertionError("Should not resolve active tracer when propagate is False")
+
+    monkeypatch.setattr(annotation_module, "get_active_tracer", fail_get_active_tracer)
+
+    result = emit_annotation({"score": 1}, propagate=False)
+
+    assert result.name == AGL_ANNOTATION
+    assert captured["called"] is False
 
 
 def test_emit_annotation_rejects_non_primitive_values() -> None:
-    with pytest.raises(TypeError):
+    with pytest.raises(ValueError):
         emit_annotation({"bad": {"set": {1}}})

tests/emitter/test_exception.py

@@ -2,8 +2,6 @@
 
 from __future__ import annotations
 
-from typing import Any, Dict, Optional
-
 import pytest
 from opentelemetry.semconv.attributes import exception_attributes
 
@@ -11,62 +9,47 @@ from agentlightning.emitter import emit_exception
 from agentlightning.emitter import exception as exception_module
 from agentlightning.semconv import AGL_EXCEPTION
 
-
-class DummySpan:
-    def __init__(self) -> None:
-        self.recorded_exception: Optional[Exception] = None
-
-    def __enter__(self) -> "DummySpan":
-        return self
-
-    def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> bool:
-        return False
-
-    def record_exception(self, exception: Exception) -> None:
-        self.recorded_exception = exception
+from ..common.tracer import RecordingDummyTracer
 
 
-class DummyTracer:
-    def __init__(self, span: DummySpan) -> None:
-        self._span = span
-        self.last_name: Optional[str] = None
-        self.last_attributes: Optional[Dict[str, Any]] = None
-
-    def start_span(self, name: str, attributes: Optional[Dict[str, Any]] = None) -> DummySpan:
-        self.last_name = name
-        self.last_attributes = attributes or {}
-        return self._span
-
-
-def _stub_tracer(monkeypatch: pytest.MonkeyPatch, span: DummySpan) -> DummyTracer:
-    tracer = DummyTracer(span)
-
-    def fake_get_tracer(*_: Any, **__: Any) -> DummyTracer:
-        return tracer
-
-    monkeypatch.setattr(exception_module, "get_tracer", fake_get_tracer)
+def _install_tracer(monkeypatch: pytest.MonkeyPatch) -> RecordingDummyTracer:
+    tracer = RecordingDummyTracer()
+    monkeypatch.setattr(exception_module, "get_active_tracer", lambda: tracer)
     return tracer
 
 
 def test_emit_exception_records_exception(monkeypatch: pytest.MonkeyPatch) -> None:
-    span = DummySpan()
-    tracer = _stub_tracer(monkeypatch, span)
+    tracer = _install_tracer(monkeypatch)
+    err = ValueError("boom")
 
-    exc: Optional[Exception] = None
-    try:
-        raise ValueError("boom")
-    except ValueError as err:
-        emit_exception(err)
-        exc = err
+    emit_exception(err)
 
-    assert tracer.last_name == AGL_EXCEPTION
-    assert tracer.last_attributes is not None
-    assert tracer.last_attributes[exception_attributes.EXCEPTION_TYPE] == "ValueError"
-    assert tracer.last_attributes[exception_attributes.EXCEPTION_MESSAGE] == "boom"
-    assert tracer.last_attributes[exception_attributes.EXCEPTION_ESCAPED] is True
-    assert span.recorded_exception is exc
+    assert tracer.last_span is not None
+    assert tracer.last_span.name == AGL_EXCEPTION
+    assert tracer.last_span.attributes[exception_attributes.EXCEPTION_TYPE] == "ValueError"
+    assert tracer.last_span.attributes[exception_attributes.EXCEPTION_MESSAGE] == "boom"
+    assert tracer.last_span.attributes[exception_attributes.EXCEPTION_ESCAPED] is True
 
 
 def test_emit_exception_requires_exception_instance() -> None:
     with pytest.raises(TypeError):
         emit_exception("boom")  # type: ignore[arg-type]
+
+
+def test_emit_exception_flattens_attributes(monkeypatch: pytest.MonkeyPatch) -> None:
+    tracer = _install_tracer(monkeypatch)
+
+    emit_exception(ValueError("boom"), attributes={"meta": {"tag": "foo"}, "labels": ["x"]})
+
+    assert tracer.last_span is not None
+    assert tracer.last_span.attributes["meta.tag"] == "foo"
+    assert tracer.last_span.attributes["labels"] == ["x"]
+
+
+def test_emit_exception_propagate_false(monkeypatch: pytest.MonkeyPatch) -> None:
+    def fail_get_active_tracer() -> RecordingDummyTracer:
+        raise AssertionError("Should not resolve tracer when propagate=False")
+
+    monkeypatch.setattr(exception_module, "get_active_tracer", fail_get_active_tracer)
+
+    emit_exception(ValueError("boom"), propagate=False)

tests/emitter/test_message.py

@@ -13,39 +13,17 @@ from agentlightning.emitter.message import get_message_value
 from agentlightning.semconv import AGL_MESSAGE, LightningSpanAttributes
 from agentlightning.types.tracer import SpanLike
 
+from ..common.tracer import RecordingDummyTracer
+
 
 @dataclass
 class FakeSpan:
     attributes: Optional[Dict[str, Any]]
 
 
-class DummySpan:
-    def __enter__(self) -> "DummySpan":
-        return self
-
-    def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> bool:
-        return False
-
-
-class DummyTracer:
-    def __init__(self, span: DummySpan) -> None:
-        self._span = span
-        self.last_name: Optional[str] = None
-        self.last_attributes: Optional[Dict[str, Any]] = None
-
-    def start_span(self, name: str, attributes: Optional[Dict[str, Any]] = None) -> DummySpan:
-        self.last_name = name
-        self.last_attributes = attributes or {}
-        return self._span
-
-
-def _stub_tracer(monkeypatch: pytest.MonkeyPatch, span: DummySpan) -> DummyTracer:
-    tracer = DummyTracer(span)
-
-    def fake_get_tracer(*_: Any, **__: Any) -> DummyTracer:
-        return tracer
-
-    monkeypatch.setattr(message_module, "get_tracer", fake_get_tracer)
+def _stub_tracer(monkeypatch: pytest.MonkeyPatch) -> RecordingDummyTracer:
+    tracer = RecordingDummyTracer()
+    monkeypatch.setattr(message_module, "get_active_tracer", lambda: tracer)
     return tracer
 
 
@@ -69,15 +47,34 @@ def test_get_message_value_rejects_non_string() -> None:
 
 
 def test_emit_message_valid(monkeypatch: pytest.MonkeyPatch) -> None:
-    span = DummySpan()
-    tracer = _stub_tracer(monkeypatch, span)
+    tracer = _stub_tracer(monkeypatch)
 
     emit_message("hello world")
 
-    assert tracer.last_name == AGL_MESSAGE
-    assert tracer.last_attributes == {LightningSpanAttributes.MESSAGE_BODY.value: "hello world"}
+    assert tracer.last_span is not None
+    assert tracer.last_span.name == AGL_MESSAGE
+    assert tracer.last_span.attributes == {LightningSpanAttributes.MESSAGE_BODY.value: "hello world"}
 
 
 def test_emit_message_requires_string() -> None:
     with pytest.raises(TypeError):
         emit_message(123)  # type: ignore[arg-type]
+
+
+def test_emit_message_flattens_attributes(monkeypatch: pytest.MonkeyPatch) -> None:
+    tracer = _stub_tracer(monkeypatch)
+
+    emit_message("hello", attributes={"meta": {"tag": "foo"}, "labels": ["a", "b"]})
+
+    assert tracer.last_span is not None
+    assert tracer.last_span.attributes["meta.tag"] == "foo"
+    assert tracer.last_span.attributes["labels"] == ["a", "b"]
+
+
+def test_emit_message_propagate_false(monkeypatch: pytest.MonkeyPatch) -> None:
+    def fail_get_active_tracer() -> RecordingDummyTracer:
+        raise AssertionError("Should not resolve tracer when propagate=False")
+
+    monkeypatch.setattr(message_module, "get_active_tracer", fail_get_active_tracer)
+
+    emit_message("local", propagate=False)

tests/emitter/test_object.py

@@ -35,7 +35,7 @@ class DummyTracer:
         self.last_name: Optional[str] = None
         self.last_attributes: Optional[Dict[str, Any]] = None
 
-    def start_span(self, name: str, attributes: Optional[Dict[str, Any]] = None) -> DummySpan:
+    def create_span(self, name: str, attributes: Optional[Dict[str, Any]] = None, **kwargs: Any) -> DummySpan:
         self.last_name = name
         self.last_attributes = attributes or {}
         return self._span
@@ -47,7 +47,7 @@ def _stub_tracer(monkeypatch: pytest.MonkeyPatch, span: DummySpan) -> DummyTrace
     def fake_get_tracer(*_: Any, **__: Any) -> DummyTracer:
         return tracer
 
-    monkeypatch.setattr(object_module, "get_tracer", fake_get_tracer)
+    monkeypatch.setattr(object_module, "get_active_tracer", fake_get_tracer)
     return tracer
 
 
@@ -175,3 +175,23 @@ def test_get_object_value_raises_for_unknown_literal_type() -> None:
 
     with pytest.raises(RuntimeError):
         get_object_value(cast(SpanLike, span))
+
+
+def test_emit_object_flattens_attributes(monkeypatch: pytest.MonkeyPatch) -> None:
+    span = DummySpan()
+    tracer = _stub_tracer(monkeypatch, span)
+
+    emit_object({"foo": "bar"}, attributes={"meta": {"tag": "foo"}, "labels": ["x", "y"]})
+
+    assert tracer.last_attributes is not None
+    assert tracer.last_attributes["meta.tag"] == "foo"
+    assert tracer.last_attributes["labels"] == ["x", "y"]
+
+
+def test_emit_object_propagate_false(monkeypatch: pytest.MonkeyPatch) -> None:
+    def fail_get_active_tracer() -> DummyTracer:
+        raise AssertionError("Should not resolve tracer when propagate=False")
+
+    monkeypatch.setattr(object_module, "get_active_tracer", fail_get_active_tracer)
+
+    emit_object({"foo": "bar"}, propagate=False)

tests/emitter/test_operation.py

@@ -3,9 +3,11 @@
 from __future__ import annotations
 
 import json
+import time
+from contextlib import contextmanager
 from dataclasses import dataclass
 from types import TracebackType
-from typing import Any, Dict, List, Optional, Tuple, Type
+from typing import Any, ContextManager, Dict, Iterator, List, Optional, Tuple, Type
 
 import opentelemetry.trace as trace_api
 import pytest
@@ -15,96 +17,141 @@ from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanE
 from opentelemetry.trace import Status, StatusCode
 
 import agentlightning.emitter.annotation as annotation_module
-from agentlightning.emitter.annotation import _safe_json_dump  # pyright: ignore[reportPrivateUsage]
-from agentlightning.emitter.annotation import (
-    OperationContext,
-    emit_annotation,
-    operation,
-)
+from agentlightning.emitter.annotation import OperationContext, emit_annotation, operation
 from agentlightning.semconv import AGL_ANNOTATION, AGL_OPERATION, LightningSpanAttributes
-from agentlightning.utils.otel import extract_links_from_attributes, make_link_attributes, query_linked_spans
+from agentlightning.tracer.dummy import DummySpanRecordingContext, DummyTracer
+from agentlightning.types import SpanCoreFields, TraceStatus
+from agentlightning.types.tracer import Attributes
+from agentlightning.utils.otel import (
+    extract_links_from_attributes,
+    filter_and_unflatten_attributes,
+    make_link_attributes,
+    query_linked_spans,
+)
 
 
-class RecordingSpan:
+class RecordingTracer:
     def __init__(self) -> None:
-        self.attributes: Dict[str, Any] = {}
-        self.recorded_exceptions: List[BaseException] = []
-        self.statuses: List[Status] = []
+        self._delegate = DummyTracer()
+        self.recordings: List[DummySpanRecordingContext] = []
 
-    def set_attribute(self, key: str, value: Any) -> None:
-        self.attributes[key] = value
-
-    def record_exception(self, exc: BaseException) -> None:
-        self.recorded_exceptions.append(exc)
-
-    def set_status(self, status: Status) -> None:
-        self.statuses.append(status)
-
-
-class DummySpanContextManager:
-    def __init__(self, span: RecordingSpan) -> None:
-        self.span = span
-        self.exit_calls: List[
-            Tuple[Optional[Type[BaseException]], Optional[BaseException], Optional[TracebackType]]
-        ] = []
-
-    def __enter__(self) -> RecordingSpan:
-        return self.span
-
-    def __exit__(
+    def operation_context(
         self,
-        exc_type: Optional[Type[BaseException]],
-        exc_val: Optional[BaseException],
-        exc_tb: Optional[TracebackType],
-    ) -> bool:
-        self.exit_calls.append((exc_type, exc_val, exc_tb))
-        return False
+        name: str,
+        attributes: Optional[Attributes] = None,
+        start_time: Optional[float] = None,
+        end_time: Optional[float] = None,
+    ) -> ContextManager[DummySpanRecordingContext]:
+        parent_ctx = self._delegate.operation_context(name, attributes, start_time, end_time)
+
+        @contextmanager
+        def _wrapper() -> Iterator[DummySpanRecordingContext]:
+            with parent_ctx as recording:
+                self.recordings.append(recording)
+                yield recording
+
+        return _wrapper()
+
+    def create_span(
+        self,
+        name: str,
+        attributes: Optional[Attributes] = None,
+        timestamp: Optional[float] = None,
+        status: Optional[TraceStatus] = None,
+    ) -> SpanCoreFields:
+        return self._delegate.create_span(name, attributes, timestamp, status)
 
 
-class DummyTracer:
-    def __init__(self, start_span_instance: Optional[RecordingSpan] = None) -> None:
-        self._start_span_instance = start_span_instance
-        self.start_span_calls: List[Tuple[str, Dict[str, Any]]] = []
-        self.start_as_current_span_calls: List[Tuple[str, Dict[str, Any], RecordingSpan]] = []
+class OtelSpanRecordingContext:
+    def __init__(self, span: trace_api.Span) -> None:
+        self._span = span
 
-    def start_span(self, name: str, attributes: Optional[Dict[str, Any]] = None) -> RecordingSpan:
-        span = self._start_span_instance or RecordingSpan()
-        self.start_span_calls.append((name, dict(attributes or {})))
-        return span
+    def record_exception(self, exception: BaseException) -> None:
+        self._span.record_exception(exception)
+        self.record_status("ERROR", str(exception))
 
-    def start_as_current_span(
+    def record_attributes(self, attributes: Dict[str, Any]) -> None:
+        for key, value in attributes.items():
+            self._span.set_attribute(key, value)
+
+    def record_status(self, status_code: str, description: Optional[str] = None) -> None:
+        self._span.set_status(Status(StatusCode[status_code], description))  # type: ignore[index]
+
+    def get_recorded_span(self) -> None:
+        raise NotImplementedError()
+
+
+class OtelTracerAdapter:
+    def __init__(self, tracer: trace_api.Tracer) -> None:
+        self._tracer = tracer
+
+    def operation_context(
         self,
         name: str,
         attributes: Optional[Dict[str, Any]] = None,
-    ) -> DummySpanContextManager:
-        span = RecordingSpan()
-        self.start_as_current_span_calls.append((name, dict(attributes or {}), span))
-        return DummySpanContextManager(span)
+        start_time: Optional[float] = None,
+        end_time: Optional[float] = None,
+    ):
+        ctx = self._tracer.start_as_current_span(name, attributes=attributes)
 
+        class _ContextManager:
+            def __enter__(self) -> OtelSpanRecordingContext:
+                span = ctx.__enter__()
+                return OtelSpanRecordingContext(span)
 
-class DummyUseSpan:
-    def __init__(self) -> None:
-        self.calls: List[Tuple[RecordingSpan, bool]] = []
-        self.exit_calls: List[
-            Tuple[Optional[Type[BaseException]], Optional[BaseException], Optional[TracebackType]]
-        ] = []
+            def __exit__(
+                self,
+                exc_type: Optional[Type[BaseException]],
+                exc_val: Optional[BaseException],
+                exc_tb: Optional[TracebackType],
+            ) -> bool:
+                result = ctx.__exit__(exc_type, exc_val, exc_tb)
+                return bool(result)
 
-    def __call__(self, span: RecordingSpan, end_on_exit: bool) -> DummyUseSpan:
-        self.calls.append((span, end_on_exit))
-        self._span = span
-        return self
+        return _ContextManager()
 
-    def __enter__(self) -> None:
-        return None
-
-    def __exit__(
+    def create_span(
         self,
-        exc_type: Optional[Type[BaseException]],
-        exc_val: Optional[BaseException],
-        exc_tb: Optional[TracebackType],
-    ) -> bool:
-        self.exit_calls.append((exc_type, exc_val, exc_tb))
-        return False
+        name: str,
+        attributes: Optional[Dict[str, Any]] = None,
+        timestamp: Optional[float] = None,
+        status: Optional[TraceStatus] = None,
+    ) -> SpanCoreFields:
+        span = self._tracer.start_span(name, attributes=attributes)
+        if status:
+            span.set_status(Status(StatusCode[status.status_code], status.description))  # type: ignore[index]
+        span.end()
+        start = timestamp or time.time()
+        return SpanCoreFields(
+            name=name,
+            attributes=attributes or {},
+            start_time=start,
+            end_time=start,
+            status=status or TraceStatus(status_code="OK"),
+        )
+
+
+def _install_recording_tracer(monkeypatch: pytest.MonkeyPatch) -> RecordingTracer:
+    tracer = RecordingTracer()
+
+    def fake_get_active_tracer() -> RecordingTracer:
+        return tracer
+
+    monkeypatch.setattr(annotation_module, "get_active_tracer", fake_get_active_tracer)
+    return tracer
+
+
+def _resolve_attr(recording: DummySpanRecordingContext, key: str) -> Any:
+    if key in recording.attributes:
+        value = recording.attributes[key]
+    else:
+        value = filter_and_unflatten_attributes(recording.attributes, key)
+    if isinstance(value, str):
+        try:
+            return json.loads(value)
+        except json.JSONDecodeError:
+            return value
+    return value
 
 
 @dataclass
@@ -113,20 +160,8 @@ class ComplexResult:
     marker: str
 
 
-def test_safe_json_dump_handles_recursive_structures() -> None:
-    payload: List[Any] = []
-    payload.append(payload)
-
-    assert _safe_json_dump(payload) == "[[...]]"
-
-
-def test_operation_context_records_inputs_and_outputs(monkeypatch: pytest.MonkeyPatch) -> None:
-    span = RecordingSpan()
-    tracer = DummyTracer(start_span_instance=span)
-    use_span = DummyUseSpan()
-
-    monkeypatch.setattr(annotation_module, "get_tracer", lambda use_active_span_processor=True: tracer)
-    monkeypatch.setattr(annotation_module.trace, "use_span", use_span)
+def test_operation_context_serializes_inputs(monkeypatch: pytest.MonkeyPatch) -> None:
+    tracer = _install_recording_tracer(monkeypatch)
 
     ctx = OperationContext("custom-span", {"meta": {"foo": 1}, "count": 2})
 
@@ -134,63 +169,60 @@ def test_operation_context_records_inputs_and_outputs(monkeypatch: pytest.Monkey
         op.set_input({"payload": 1}, flag=True)
         op.set_output({"success": True})
 
-    assert tracer.start_span_calls
-    start_name, start_attributes = tracer.start_span_calls[0]
-    assert start_name == "custom-span"
-    assert json.loads(start_attributes["meta"]) == {"foo": 1}
-    assert start_attributes["count"] == 2
-
-    assert json.loads(span.attributes["input.args"]) == [{"payload": 1}]
-    assert span.attributes["input.flag"] == "true"
-    assert json.loads(span.attributes["output"]) == {"success": True}
-    assert use_span.calls == [(span, True)]
+    recording = tracer.recordings[-1]
+    assert recording.name == "custom-span"
+    assert recording.attributes["meta.foo"] == 1
+    assert recording.attributes["count"] == 2
+    input_prefix = LightningSpanAttributes.OPERATION_INPUT.value
+    assert _resolve_attr(recording, f"{input_prefix}.args") == [{"payload": 1}]
+    assert recording.attributes[f"{input_prefix}.flag"] is True
+    assert _resolve_attr(recording, LightningSpanAttributes.OPERATION_OUTPUT.value) == {"success": True}
 
 
 def test_operation_context_set_input_supports_multiple_values(monkeypatch: pytest.MonkeyPatch) -> None:
-    span = RecordingSpan()
-    tracer = DummyTracer(start_span_instance=span)
-    use_span = DummyUseSpan()
-
-    monkeypatch.setattr(annotation_module, "get_tracer", lambda use_active_span_processor=True: tracer)
-    monkeypatch.setattr(annotation_module.trace, "use_span", use_span)
+    tracer = _install_recording_tracer(monkeypatch)
 
     ctx = OperationContext("ctx", {})
 
     with ctx as op:
         op.set_input(1, 2, data={"foo": ["bar"]}, flags=[True, False])
 
-    assert json.loads(span.attributes["input.args"]) == [1, 2]
-    assert json.loads(span.attributes["input.data"]) == {"foo": ["bar"]}
-    assert json.loads(span.attributes["input.flags"]) == [True, False]
+    recording = tracer.recordings[-1]
+    input_prefix = LightningSpanAttributes.OPERATION_INPUT.value
+    assert _resolve_attr(recording, f"{input_prefix}.args") == [1, 2]
+    assert _resolve_attr(recording, f"{input_prefix}.data") == {"foo": ["bar"]}
+    assert _resolve_attr(recording, f"{input_prefix}.flags") == [True, False]
 
 
-def test_operation_context_records_non_serializable_output(monkeypatch: pytest.MonkeyPatch) -> None:
-    class Unserializable:
-        def __str__(self) -> str:
-            return "<Unserializable>"
-
-    span = RecordingSpan()
-    tracer = DummyTracer(start_span_instance=span)
-    use_span = DummyUseSpan()
-
-    monkeypatch.setattr(annotation_module, "get_tracer", lambda use_active_span_processor=True: tracer)
-    monkeypatch.setattr(annotation_module.trace, "use_span", use_span)
+def test_operation_context_set_input_expands_positional_attributes(monkeypatch: pytest.MonkeyPatch) -> None:
+    tracer = _install_recording_tracer(monkeypatch)
 
     ctx = OperationContext("ctx", {})
 
     with ctx as op:
-        op.set_output(Unserializable())
+        op.set_input("alpha", "beta")
 
-    assert json.loads(span.attributes["output"]) == "<Unserializable>"
+    recording = tracer.recordings[-1]
+    input_prefix = LightningSpanAttributes.OPERATION_INPUT.value
+    assert recording.attributes[f"{input_prefix}.args.0"] == "alpha"
+    assert recording.attributes[f"{input_prefix}.args.1"] == "beta"
+
+
+def test_operation_context_rejects_non_serializable_output(monkeypatch: pytest.MonkeyPatch) -> None:
+    _install_recording_tracer(monkeypatch)
+
+    class Unserializable:
+        pass
+
+    ctx = OperationContext("ctx", {})
+
+    with pytest.raises(ValueError, match="Object must be JSON serializable"):
+        with ctx as op:
+            op.set_output(Unserializable())
 
 
 def test_operation_context_records_exceptions(monkeypatch: pytest.MonkeyPatch) -> None:
-    span = RecordingSpan()
-    tracer = DummyTracer(start_span_instance=span)
-    use_span = DummyUseSpan()
-
-    monkeypatch.setattr(annotation_module, "get_tracer", lambda use_active_span_processor=True: tracer)
-    monkeypatch.setattr(annotation_module.trace, "use_span", use_span)
+    tracer = _install_recording_tracer(monkeypatch)
 
     ctx = OperationContext("custom-span", {})
 
@@ -198,48 +230,37 @@ def test_operation_context_records_exceptions(monkeypatch: pytest.MonkeyPatch) -
         with ctx:
             raise RuntimeError("boom")
 
-    assert isinstance(span.recorded_exceptions[0], RuntimeError)
-    status = span.statuses[-1]
-    assert status.status_code == StatusCode.ERROR
-    assert status.description == "boom"
-    assert use_span.exit_calls[-1][1].args == ("boom",)  # type: ignore
+    recording = tracer.recordings[-1]
+    assert "exception.type" in recording.attributes
+    assert recording.status.status_code == "ERROR"
+    assert recording.status.description == "boom"
 
 
 def test_operation_factory_context_records_inputs_and_outputs(monkeypatch: pytest.MonkeyPatch) -> None:
-    span = RecordingSpan()
-    tracer = DummyTracer(start_span_instance=span)
-    use_span = DummyUseSpan()
-
-    monkeypatch.setattr(annotation_module, "get_tracer", lambda use_active_span_processor=True: tracer)
-    monkeypatch.setattr(annotation_module.trace, "use_span", use_span)
+    tracer = _install_recording_tracer(monkeypatch)
 
     with operation(tags=["one", "two"]) as ctx:
         ctx.set_input("alpha", meta={"score": 0.5})
         ctx.set_output(["beta", "gamma"])
 
-    start_name, attrs = tracer.start_span_calls[0]
-    assert start_name == AGL_OPERATION
-    assert json.loads(attrs["tags"]) == ["one", "two"]
-    assert json.loads(span.attributes["input.args"]) == ["alpha"]
-    assert json.loads(span.attributes["input.meta"]) == {"score": 0.5}
-    assert json.loads(span.attributes["output"]) == ["beta", "gamma"]
+    recording = tracer.recordings[-1]
+    input_prefix = LightningSpanAttributes.OPERATION_INPUT.value
+    assert recording.name == AGL_OPERATION
+    assert recording.attributes["tags"] == ["one", "two"]
+    assert _resolve_attr(recording, f"{input_prefix}.args") == ["alpha"]
+    assert _resolve_attr(recording, f"{input_prefix}.meta") == {"score": 0.5}
+    assert _resolve_attr(recording, LightningSpanAttributes.OPERATION_OUTPUT.value) == ["beta", "gamma"]
 
 
 def test_operation_factory_uses_standard_span_name(monkeypatch: pytest.MonkeyPatch) -> None:
-    span = RecordingSpan()
-    tracer = DummyTracer(start_span_instance=span)
-    use_span = DummyUseSpan()
-
-    monkeypatch.setattr(annotation_module, "get_tracer", lambda use_active_span_processor=True: tracer)
-    monkeypatch.setattr(annotation_module.trace, "use_span", use_span)
+    tracer = _install_recording_tracer(monkeypatch)
 
     with operation(user={"id": 5}) as ctx:
         ctx.set_output("done")
 
-    assert tracer.start_span_calls
-    start_name, attrs = tracer.start_span_calls[0]
-    assert start_name == AGL_OPERATION
-    assert json.loads(attrs["user"]) == {"id": 5}
+    recording = tracer.recordings[-1]
+    assert recording.name == AGL_OPERATION
+    assert recording.attributes["user.id"] == 5
 
 
 def test_operation_rejects_custom_span_names() -> None:
@@ -248,8 +269,7 @@ def test_operation_rejects_custom_span_names() -> None:
 
 
 def test_operation_decorator_sync_records_span_attributes(monkeypatch: pytest.MonkeyPatch) -> None:
-    tracer = DummyTracer()
-    monkeypatch.setattr(annotation_module, "get_tracer", lambda use_active_span_processor=True: tracer)
+    tracer = _install_recording_tracer(monkeypatch)
 
     @operation(category={"kind": "combine"})
     def combine(data: Dict[str, int], *, meta: Dict[str, str]) -> Dict[str, Any]:
@@ -258,21 +278,19 @@ def test_operation_decorator_sync_records_span_attributes(monkeypatch: pytest.Mo
     result = combine({"value": 1}, meta={"source": "unit"})
 
     assert result == {"joined": {"value": 1, "source": "unit"}}
-    assert tracer.start_as_current_span_calls
-    span_name, span_attributes, span = tracer.start_as_current_span_calls[0]
-    assert span_name == AGL_OPERATION
-    assert json.loads(span_attributes["category"]) == {"kind": "combine"}
+    recording = tracer.recordings[-1]
+    assert recording.name == AGL_OPERATION
+    assert recording.attributes["category.kind"] == "combine"
 
     input_prefix = LightningSpanAttributes.OPERATION_INPUT.value
-    assert json.loads(span.attributes[f"{input_prefix}.data"]) == {"value": 1}
-    assert json.loads(span.attributes[f"{input_prefix}.meta"]) == {"source": "unit"}
-    assert span.attributes[LightningSpanAttributes.OPERATION_NAME.value] == "combine"
-    assert json.loads(span.attributes[LightningSpanAttributes.OPERATION_OUTPUT.value]) == result
+    assert _resolve_attr(recording, f"{input_prefix}.data") == {"value": 1}
+    assert _resolve_attr(recording, f"{input_prefix}.meta") == {"source": "unit"}
+    assert recording.attributes[LightningSpanAttributes.OPERATION_NAME.value] == "combine"
+    assert _resolve_attr(recording, LightningSpanAttributes.OPERATION_OUTPUT.value) == result
 
 
 def test_operation_decorator_handles_complex_signature(monkeypatch: pytest.MonkeyPatch) -> None:
-    tracer = DummyTracer()
-    monkeypatch.setattr(annotation_module, "get_tracer", lambda use_active_span_processor=True: tracer)
+    tracer = _install_recording_tracer(monkeypatch)
 
     @operation()
     def complicated(
@@ -287,25 +305,25 @@ def test_operation_decorator_handles_complex_signature(monkeypatch: pytest.Monke
     ) -> ComplexResult:
         return ComplexResult(values=(first, len(extra), len(rest)), marker=kwonly + kwdefault + required)
 
-    result = complicated(1, "req", 7, 8, 9, kwonly="x", kwdefault="y", tag="value")
+    with pytest.raises(ValueError):
+        complicated(1, "req", 7, 8, 9, kwonly="x", kwdefault="y", tag="value")
 
-    span = tracer.start_as_current_span_calls[0][2]
+    recording = tracer.recordings[-1]
     input_prefix = LightningSpanAttributes.OPERATION_INPUT.value
 
-    assert json.loads(span.attributes[f"{input_prefix}.first"]) == 1
-    assert json.loads(span.attributes[f"{input_prefix}.required"]) == "req"
-    assert json.loads(span.attributes[f"{input_prefix}.default"]) == 7
-    assert json.loads(span.attributes[f"{input_prefix}.extra"]) == [8, 9]
-    assert json.loads(span.attributes[f"{input_prefix}.kwonly"]) == "x"
-    assert json.loads(span.attributes[f"{input_prefix}.kwdefault"]) == "y"
-    assert json.loads(span.attributes[f"{input_prefix}.rest"]) == {"tag": "value"}
-    assert span.attributes[LightningSpanAttributes.OPERATION_NAME.value] == "complicated"
-    assert json.loads(span.attributes[LightningSpanAttributes.OPERATION_OUTPUT.value]) == str(result)
+    assert _resolve_attr(recording, f"{input_prefix}.first") == 1
+    assert _resolve_attr(recording, f"{input_prefix}.required") == "req"
+    assert _resolve_attr(recording, f"{input_prefix}.default") == 7
+    assert _resolve_attr(recording, f"{input_prefix}.extra") == [8, 9]
+    assert _resolve_attr(recording, f"{input_prefix}.kwonly") == "x"
+    assert _resolve_attr(recording, f"{input_prefix}.kwdefault") == "y"
+    assert _resolve_attr(recording, f"{input_prefix}.rest") == {"tag": "value"}
+    assert recording.attributes[LightningSpanAttributes.OPERATION_NAME.value] == "complicated"
+    assert recording.status.status_code == "ERROR"
 
 
 def test_operation_decorator_records_exceptions(monkeypatch: pytest.MonkeyPatch) -> None:
-    tracer = DummyTracer()
-    monkeypatch.setattr(annotation_module, "get_tracer", lambda use_active_span_processor=True: tracer)
+    tracer = _install_recording_tracer(monkeypatch)
 
     @operation()
     def fail(value: int) -> int:
@@ -314,17 +332,14 @@ def test_operation_decorator_records_exceptions(monkeypatch: pytest.MonkeyPatch)
     with pytest.raises(ValueError):
         fail(1)
 
-    span = tracer.start_as_current_span_calls[0][2]
-    assert isinstance(span.recorded_exceptions[0], ValueError)
-    status = span.statuses[-1]
-    assert status.status_code == StatusCode.ERROR
-    assert status.description == "bad input"
+    recording = tracer.recordings[-1]
+    assert recording.status.status_code == "ERROR"
+    assert recording.status.description == "bad input"
 
 
 @pytest.mark.asyncio()
 async def test_operation_async_wrapper_records_attributes(monkeypatch: pytest.MonkeyPatch) -> None:
-    tracer = DummyTracer()
-    monkeypatch.setattr(annotation_module, "get_tracer", lambda use_active_span_processor=True: tracer)
+    tracer = _install_recording_tracer(monkeypatch)
 
     @operation()
     async def echo(payload: Dict[str, Any]) -> Dict[str, Any]:
@@ -333,20 +348,19 @@ async def test_operation_async_wrapper_records_attributes(monkeypatch: pytest.Mo
     result = await echo({"value": 3})
 
     assert result == {"payload": {"value": 3}}
-    span = tracer.start_as_current_span_calls[0][2]
+    recording = tracer.recordings[-1]
     prefix = LightningSpanAttributes.OPERATION_INPUT.value
-    assert json.loads(span.attributes[f"{prefix}.payload"]) == {"value": 3}
-    assert json.loads(span.attributes[LightningSpanAttributes.OPERATION_OUTPUT.value]) == result
+    assert _resolve_attr(recording, f"{prefix}.payload") == {"value": 3}
+    assert _resolve_attr(recording, LightningSpanAttributes.OPERATION_OUTPUT.value) == result
 
 
 def test_operation_span_can_be_resolved_via_annotation_links(monkeypatch: pytest.MonkeyPatch) -> None:
     provider = TracerProvider()
     exporter = InMemorySpanExporter()
     provider.add_span_processor(SimpleSpanProcessor(exporter))
-    tracer = provider.get_tracer(__name__)
+    tracer = OtelTracerAdapter(provider.get_tracer(__name__))
 
-    monkeypatch.setattr(annotation_module, "get_tracer", lambda use_active_span_processor=True: tracer)
-    monkeypatch.setattr(annotation_module, "get_tracer", lambda use_active_span_processor=True: tracer)
+    monkeypatch.setattr(annotation_module, "get_active_tracer", lambda: tracer)
 
     @operation(conversation_id="conv-1")
     def decorated(value: int) -> int:
@@ -373,16 +387,10 @@ def test_operation_span_can_be_resolved_via_annotation_links(monkeypatch: pytest
 
 
 def test_operation_honors_propagate_flag(monkeypatch: pytest.MonkeyPatch) -> None:
-    tracer = DummyTracer()
-    flags: List[bool] = []
-    use_span = DummyUseSpan()
+    def fail_get_active_tracer() -> RecordingTracer:
+        raise AssertionError("get_active_tracer should not be called when propagate=False")
 
-    def fake_get_tracer(use_active_span_processor: bool = True) -> DummyTracer:
-        flags.append(use_active_span_processor)
-        return tracer
-
-    monkeypatch.setattr(annotation_module, "get_tracer", fake_get_tracer)
-    monkeypatch.setattr(annotation_module.trace, "use_span", use_span)
+    monkeypatch.setattr(annotation_module, "get_active_tracer", fail_get_active_tracer)
 
     @operation(propagate=False)
     def decorated(value: int) -> int:
@@ -390,7 +398,15 @@ def test_operation_honors_propagate_flag(monkeypatch: pytest.MonkeyPatch) -> Non
 
     assert decorated(7) == 7
 
-    with operation(propagate=False):
-        pass
+    with operation(propagate=False, value=7) as op:
+        with pytest.raises(RuntimeError):
+            op.span()
 
-    assert flags == [False, False]
+    assert op.span() is not None
+    assert op.span().name == AGL_OPERATION
+    assert op.span().attributes == {"value": 7}
+    assert op.span().status.status_code == "OK"
+    assert op.span().status.description is None
+    assert op.span().start_time is not None
+    assert op.span().end_time is not None
+    assert op.span().start_time < op.span().end_time  # type: ignore

tests/runner/test_agent_runner.py

@@ -22,7 +22,7 @@ from agentlightning.semconv import AGL_ANNOTATION
 from agentlightning.store.base import UNSET, LightningStore, Unset
 from agentlightning.store.memory import InMemoryLightningStore
 from agentlightning.tracer.base import Tracer
-from agentlightning.types import LLM, Hook, NamedResources, PromptTemplate, Rollout, Span, Worker
+from agentlightning.types import LLM, Hook, NamedResources, PromptTemplate, Rollout, Span, SpanCoreFields, Worker
 
 
 @pytest.fixture(scope="module", autouse=True)
@@ -73,8 +73,9 @@ def create_agent_span(
 class DummyTracer(Tracer):
     def __init__(self) -> None:
         super().__init__()
-        self._last_trace: List[ReadableSpan] = []
+        self._last_trace: List[Span] = []
         self._contexts: List[Dict[str, Any]] = []
+        self._sequence_id = 0
 
     def init(self, *args: Any, **kwargs: Any) -> None:
         self._last_trace.clear()
@@ -82,7 +83,7 @@ class DummyTracer(Tracer):
     def teardown(self, *args: Any, **kwargs: Any) -> None:
         self._last_trace.clear()
 
-    def get_last_trace(self) -> List[ReadableSpan]:
+    def get_last_trace(self) -> List[Span]:
         return list(self._last_trace)
 
     @asynccontextmanager
@@ -93,7 +94,7 @@ class DummyTracer(Tracer):
         store: Optional[LightningStore] = None,
         rollout_id: Optional[str] = None,
         attempt_id: Optional[str] = None,
-    ) -> AsyncGenerator[List[ReadableSpan], None]:
+    ) -> AsyncGenerator[List[Span], None]:
         previous = self._contexts[-1] if self._contexts else None
         current = {
             "name": name,
@@ -112,7 +113,15 @@ class DummyTracer(Tracer):
 
     def record_span(self, name: str, attributes: Optional[Dict[str, Any]] = None) -> ReadableSpan:
         span = create_readable_span(name, attributes)
-        self._last_trace.append(span)
+        rollout_id = "rollout-dummy"
+        attempt_id = "attempt-dummy"
+        sequence_id = self._sequence_id
+        self._sequence_id += 1
+        if self._contexts:
+            current = self._contexts[-1]
+            rollout_id = current["rollout_id"]
+            attempt_id = current["attempt_id"]
+        self._last_trace.append(Span.from_opentelemetry(span, rollout_id, attempt_id, sequence_id))
         return span
 
 
@@ -309,7 +318,9 @@ async def test_step_raises_for_invalid_result_type() -> None:
 
 
 @pytest.mark.asyncio
-async def test_readable_spans_return_skip_store_when_tracer_is_otel(monkeypatch: pytest.MonkeyPatch) -> None:
+async def test_readable_spans_return_skip_store_when_tracer_is_otel(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
     class DummyTracerWithOtel(DummyTracer):
         pass
 
@@ -342,12 +353,55 @@ async def test_readable_spans_return_skip_store_when_tracer_is_otel(monkeypatch:
         attempted_rollout, spans
     )
 
-    assert result_spans == spans
-    assert store.add_otel_span_calls == 0
+    assert all(isinstance(span, Span) for span in result_spans)
+    # Warned, but still logged
+    assert [span.name for span in result_spans] == ["otel-span"]
+    assert store.add_otel_span_calls == 1
+    assert "Tracer is already an OpenTelemetry tracer" in caplog.text
 
     teardown_runner(runner)
 
 
+@pytest.mark.asyncio
+async def test_post_process_readable_spans_adds_each_span() -> None:
+    agent = HeartbeatAgent()
+    runner, store, _ = await setup_runner(agent)
+    attempted_rollout = await store.start_rollout(input={"task": "post-process"}, mode="val")
+
+    spans = [create_readable_span("case-2-span-a"), create_readable_span("case-2-span-b")]
+
+    try:
+        result_spans = await runner._post_process_rollout_result(  # pyright: ignore[reportPrivateUsage]
+            attempted_rollout, spans
+        )
+    finally:
+        teardown_runner(runner)
+
+    assert [span.name for span in result_spans] == ["case-2-span-a", "case-2-span-b"]
+    stored_spans = await store.query_spans(attempted_rollout.rollout_id, attempted_rollout.attempt.attempt_id)
+    assert [span.name for span in stored_spans] == ["case-2-span-a", "case-2-span-b"]
+
+
+@pytest.mark.asyncio
+async def test_post_process_span_core_fields_create_spans() -> None:
+    agent = HeartbeatAgent()
+    runner, store, _ = await setup_runner(agent)
+    attempted_rollout = await store.start_rollout(input={"task": "reward-list"}, mode="val")
+
+    span_core_fields = [emit_reward(0.5, propagate=False), emit_reward(-0.2, propagate=False)]
+
+    try:
+        result_spans = await runner._post_process_rollout_result(  # pyright: ignore[reportPrivateUsage]
+            attempted_rollout, span_core_fields
+        )
+    finally:
+        teardown_runner(runner)
+
+    assert all(span.name == AGL_ANNOTATION for span in result_spans)
+    stored_spans = await store.query_spans(attempted_rollout.rollout_id, attempted_rollout.attempt.attempt_id)
+    assert [span.attributes.get("agentlightning.reward.0.value") for span in stored_spans] == [0.5, -0.2]
+
+
 @pytest.mark.asyncio
 async def test_step_handles_non_llm_resource() -> None:
     class PromptAgent(LitAgent[str]):
@@ -585,8 +639,8 @@ async def test_agent_emits_multiple_rewards() -> None:
     class RewardListAgent(LitAgent[Dict[str, Any]]):
         def validation_rollout(
             self, task: Dict[str, Any], resources: Dict[str, Any], rollout: Any
-        ) -> List[ReadableSpan]:
-            return [emit_reward(0.2), emit_reward(0.6)]
+        ) -> List[SpanCoreFields]:
+            return [emit_reward(0.2, propagate=False), emit_reward(0.6, propagate=False)]
 
     agent = RewardListAgent()
     runner, store, _ = await setup_runner(agent)

tests/runner/test_runner_context.py

@@ -5,14 +5,14 @@ from typing import Any, AsyncGenerator, Dict, List, Optional
 
 import pytest
 from opentelemetry import trace as trace_api
-from opentelemetry.sdk.trace import ReadableSpan, TracerProvider
+from opentelemetry.sdk.trace import TracerProvider
 
 from agentlightning.litagent import LitAgent
 from agentlightning.runner import LitAgentRunner
 from agentlightning.store.base import LightningStore
 from agentlightning.store.memory import InMemoryLightningStore
 from agentlightning.tracer.base import Tracer
-from agentlightning.types import LLM, Hook, Rollout
+from agentlightning.types import LLM, Hook, Rollout, Span
 
 from ..common.tracer import clear_tracer_provider
 
@@ -31,7 +31,7 @@ class DummyTracer(Tracer):
 
     def __init__(self) -> None:
         super().__init__()
-        self._last_trace: List[ReadableSpan] = []
+        self._last_trace: List[Span] = []
         self.init_called = False
         self.init_worker_called = False
         self.teardown_called = False
@@ -51,7 +51,7 @@ class DummyTracer(Tracer):
     def teardown_worker(self, worker_id: int, *args: Any, **kwargs: Any) -> None:
         self.teardown_worker_called = True
 
-    def get_last_trace(self) -> List[ReadableSpan]:
+    def get_last_trace(self) -> List[Span]:
         return list(self._last_trace)
 
     @asynccontextmanager
@@ -62,7 +62,7 @@ class DummyTracer(Tracer):
         store: Optional[LightningStore] = None,
         rollout_id: Optional[str] = None,
         attempt_id: Optional[str] = None,
-    ) -> AsyncGenerator[List[ReadableSpan], None]:
+    ) -> AsyncGenerator[List[Span], None]:
         self._last_trace = []
         try:
             yield self._last_trace

tests/tracer/test_agentops.py

@@ -8,6 +8,7 @@ from typing import Any, List, Optional, Union
 
 import agentops
 import pytest
+import pytest_asyncio
 import uvicorn
 from agentops.sdk.core import TraceContext
 from fastapi import FastAPI, Request
@@ -21,12 +22,20 @@ from portpicker import pick_unused_port
 
 from agentlightning.store.base import LightningStore, LightningStoreCapabilities
 from agentlightning.tracer.agentops import AgentOpsTracer
-from agentlightning.types import Span
+from agentlightning.types import Span, TraceStatus
 from agentlightning.utils import otlp
 
 pytestmark = [pytest.mark.agentops]
 
 
+@pytest_asyncio.fixture
+async def agentops_tracer():
+    tracer = AgentOpsTracer()
+    with tracer.lifespan():
+        async with tracer.trace_context():
+            yield tracer
+
+
 class MockOTLPService:
     """A mock OTLP server to capture trace export requests for testing purposes."""
 
@@ -122,9 +131,42 @@ def _func_without_exception():
     pass
 
 
+def test_agentops_tracer_create_span(agentops_tracer: AgentOpsTracer) -> None:
+    status = TraceStatus(status_code="ERROR", description="agentops")
+
+    span = agentops_tracer.create_span(
+        "agentops-span",
+        attributes={"foo": "bar"},
+        status=status,
+        timestamp=42_000.0,
+    )
+
+    assert span.name == "agentops-span"
+    assert span.attributes["foo"] == "bar"
+    assert span.status.status_code == "ERROR"
+    assert span.status.description == "agentops"
+    assert span.start_time is not None
+    assert span.end_time is not None
+
+
+def test_agentops_tracer_operation_context_records_exception(agentops_tracer: AgentOpsTracer) -> None:
+    with pytest.raises(ValueError):
+        with agentops_tracer.operation_context("agentops-op", attributes={"foo": "bar"}) as ctx:
+            raise ValueError("agentops boom")
+
+    recorded_span = ctx.get_recorded_span()  # type: ignore
+    assert recorded_span.name == "agentops-op"
+    assert recorded_span.status.status_code == "ERROR"
+    assert recorded_span.status.description is not None
+    assert "agentops boom" in recorded_span.status.description
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize("with_exception", [True, False])
 async def test_trace_error_status_from_instance(with_exception: bool):
+    import agentlightning
+
+    agentlightning.setup_logging("DEBUG")
     captured_state = {}
     old_end_trace = agentops.end_trace
 

tests/tracer/test_dummy.py

@@ -0,0 +1,87 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from __future__ import annotations
+
+import itertools
+
+import pytest
+
+from agentlightning.tracer import dummy as dummy_module
+from agentlightning.tracer.base import clear_active_tracer, get_active_tracer, set_active_tracer
+from agentlightning.tracer.dummy import DummyTracer
+
+
+def _fake_time_generator(start: float) -> itertools.count[float]:
+    return itertools.count(start=start, step=1)
+
+
+def test_dummy_tracer_create_span_uses_defaults(monkeypatch: pytest.MonkeyPatch) -> None:
+    tracer = DummyTracer()
+    fake_clock = _fake_time_generator(42.0)
+    monkeypatch.setattr(dummy_module.time, "time", lambda: next(fake_clock))
+
+    span = tracer.create_span("dummy-span", attributes={"foo": "bar"})
+
+    assert span.name == "dummy-span"
+    assert span.attributes == {"foo": "bar"}
+    assert span.start_time == 42.0
+    assert span.end_time == 42.0
+    assert span.status.status_code == "OK"
+
+
+def test_dummy_tracer_operation_context_records_span(monkeypatch: pytest.MonkeyPatch) -> None:
+    tracer = DummyTracer()
+    fake_clock = _fake_time_generator(100.0)
+    monkeypatch.setattr(dummy_module.time, "time", lambda: next(fake_clock))
+
+    with tracer.operation_context("dummy-op", attributes={"foo": "bar"}) as ctx:
+        ctx.record_attributes({"bar": "baz"})
+        ctx.record_status("OK")
+
+    recorded_span = ctx.get_recorded_span()
+    assert recorded_span.name == "dummy-op"
+    assert recorded_span.attributes == {"foo": "bar", "bar": "baz"}
+    assert recorded_span.start_time == 100.0
+    assert recorded_span.end_time == 101.0
+    assert recorded_span.status.status_code == "OK"
+
+
+def test_dummy_tracer_operation_context_records_exceptions(monkeypatch: pytest.MonkeyPatch) -> None:
+    tracer = DummyTracer()
+    fake_clock = _fake_time_generator(200.0)
+    monkeypatch.setattr(dummy_module.time, "time", lambda: next(fake_clock))
+
+    with pytest.raises(RuntimeError):
+        with tracer.operation_context("dummy-error") as ctx:
+            raise RuntimeError("boom")
+
+    recorded_span = ctx.get_recorded_span()  # type: ignore
+    assert recorded_span.status.status_code == "ERROR"
+    assert recorded_span.status.description == "boom"
+    assert recorded_span.end_time == 201.0
+
+
+@pytest.fixture(autouse=True)
+def reset_active_tracer():
+    clear_active_tracer()
+    yield
+    clear_active_tracer()
+
+
+def test_set_active_tracer_returns_same_instance() -> None:
+    tracer = DummyTracer()
+    set_active_tracer(tracer)
+    assert get_active_tracer() is tracer
+
+
+def test_set_active_tracer_raises_when_existing() -> None:
+    set_active_tracer(DummyTracer())
+    with pytest.raises(ValueError):
+        set_active_tracer(DummyTracer())
+
+
+def test_clear_active_tracer_removes_current() -> None:
+    tracer = DummyTracer()
+    set_active_tracer(tracer)
+    clear_active_tracer()
+    assert get_active_tracer() is None

tests/tracer/test_integration.py

@@ -22,7 +22,10 @@ import json
 import os
 import pprint
 import re
+import shutil
+import textwrap
 import time
+import warnings
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any, Awaitable, Callable, Dict, List, Literal, Mapping, Optional, Tuple, Union
 
@@ -39,6 +42,7 @@ from autogen_ext.models.openai import OpenAIChatCompletionClient
 from autogen_ext.tools.mcp import McpWorkbench, StdioServerParams
 from fastapi import FastAPI
 from openai import OpenAI
+from opentelemetry.sdk.trace import ReadableSpan
 from pydantic import BaseModel
 
 from agentlightning.adapter.triplet import TracerTraceToTriplet, TraceTree
@@ -675,7 +679,7 @@ def assert_expected_pairs_in_tree(root_tuple: Tuple[str, List[Any]], expected_pa
 @pytest.mark.agentops
 @pytest.mark.parametrize("agent_name", list(AGENT_FUNCTIONS.keys()), ids=str)
 @pytest.mark.asyncio
-async def test_tracer_integration(agent_name: AgentName):
+async def test_tracer_integration_agentops(agent_name: AgentName):
     if ("langchain" in agent_name or "langgraph" in agent_name) and not LANGCHAIN_INSTALLED:
         pytest.skip("LangChain is not installed. Skip langchain related tests.")
 
@@ -684,18 +688,61 @@ async def test_tracer_integration(agent_name: AgentName):
         await _run_tracer_with_agent(settings, tracer, agent_name)
 
 
-async def _run_tracer_with_agent(settings: OpenAISettings, tracer: Tracer, agent_name: AgentName):
+@pytest.mark.weave
+@pytest.mark.parametrize("agent_name", list(AGENT_FUNCTIONS.keys()), ids=str)
+@pytest.mark.asyncio
+async def test_tracer_integration_weave(agent_name: AgentName):
+    if ("langchain" in agent_name or "langgraph" in agent_name) and not LANGCHAIN_INSTALLED:
+        pytest.skip("LangChain is not installed. Skip langchain related tests.")
+
+    from agentlightning.tracer.weave import WeaveTracer
+
+    async with MockOpenAICompatibleServer() as settings:
+        tracer = WeaveTracer()
+        await _run_tracer_with_agent(settings, tracer, agent_name, _skip_assert=True)
+
+
+async def _run_tracer_with_agent(
+    settings: OpenAISettings, tracer: Tracer, agent_name: AgentName, _skip_assert: bool = False
+):
     agent_func = AGENT_FUNCTIONS[agent_name]
 
     with tracer.lifespan():
         async with tracer.trace_context(name=f"test_integration_{agent_name}"):
             await agent_func(settings, tracer)
 
-        last_trace_normalized = [Span.from_opentelemetry(span, "dummy", "dummy", 0) for span in tracer.get_last_trace()]
+        last_trace_normalized = [
+            Span.from_opentelemetry(span, "dummy", "dummy", 0) if isinstance(span, ReadableSpan) else span
+            for span in tracer.get_last_trace()
+        ]
+        for span in last_trace_normalized:
+            print(">>> rollout_id =", span.rollout_id)
+            print("... attempt_id =", span.attempt_id)
+            print("... sequence_id =", span.sequence_id)
+            print("... trace_id =", span.trace_id)
+            print("... span_id =", span.span_id)
+            print("... parent_id =", span.parent_id)
+            print("... name =", span.name)
+            print("... status =", span.status)
+            print(
+                "... attributes =",
+                textwrap.indent(pprint.pformat(span.attributes, width=200, indent=4), "    ").lstrip(),
+            )
         tree = TraceTree.from_spans(last_trace_normalized)
 
+        if shutil.which("dot"):
+            # Visualize the trace tree for debug
+            debug_dir = os.path.join(os.path.dirname(__file__), "debug")
+            os.makedirs(debug_dir, exist_ok=True)
+            tree.visualize(filename=os.path.join(debug_dir, f"{tracer.__class__.__name__}_{agent_name}"))
+        else:
+            warnings.warn("dot is not installed. Skipping trace tree visualization.")
+
         tree.repair_hierarchy()
 
+        if _skip_assert:
+            return
+
         assert_expected_pairs_in_tree(tree.names_tuple(), AGENTOPS_EXPECTED_TREES[agent_name])
 
         triplets = TracerTraceToTriplet().adapt(last_trace_normalized)

tests/tracer/test_otel.py

@@ -28,31 +28,43 @@ from agentlightning.reward import emit_reward, find_reward_spans, get_reward_val
 from agentlightning.store.base import LightningStore
 from agentlightning.tracer.agentops import LightningSpanProcessor
 from agentlightning.tracer.otel import OtelTracer
+from agentlightning.types import TraceStatus
+from agentlightning.types.tracer import Span
 from agentlightning.utils import otlp
 
-from ..common.tracer import clear_agentops_init, clear_tracer_provider
+from ..common.tracer import clear_tracer_provider
 
 
-def create_span(name: str, sampled: bool = True, with_context: bool = True) -> MagicMock:
+def create_span(name: str, sampled: bool = True, with_context: bool = True) -> ReadableSpan:
     """Helper to create mock spans with different properties."""
-    span = MagicMock(spec=ReadableSpan)
-    span.name = name
-    if with_context:
-        span.context = SpanContext(
-            trace_id=hash(name) % (2**64),
-            span_id=hash(name) % (2**64),
-            is_remote=False,
-            trace_flags=TraceFlags(0x01 if sampled else 0x00),
-        )
-    else:
-        span.context = None
+    return ReadableSpan(
+        name=name,
+        context=(
+            SpanContext(
+                trace_id=hash(name) % (2**64),
+                span_id=hash(name) % (2**64),
+                is_remote=False,
+                trace_flags=TraceFlags(0x01 if sampled else 0x00),
+            )
+            if with_context
+            else None
+        ),
+    )
+
+
+async def add_otel_span(
+    rollout_id: str, attempt_id: str, readable_span: ReadableSpan, sequence_id: int | None = None
+) -> Span:
+    span = Span.from_opentelemetry(
+        readable_span, rollout_id=rollout_id, attempt_id=attempt_id, sequence_id=sequence_id or 0
+    )
     return span
 
 
 def create_mock_store(otlp_supported: bool = False) -> MagicMock:
     """Helper to create a mock LightningStore."""
     store = MagicMock(spec=LightningStore)
-    store.add_otel_span = AsyncMock(return_value=None)
+    store.add_otel_span = AsyncMock(side_effect=add_otel_span)
     store.capabilities = {"otlp_traces": otlp_supported}
     store.otlp_traces_endpoint.return_value = "http://store/v1/traces"
     return store
@@ -114,6 +126,51 @@ def store(store_supports_otlp: bool, otlp_server: Optional[Dict[str, Any]]) -> M
     return mock_store
 
 
+@pytest.fixture
+def otel_tracer():
+    clear_tracer_provider()
+    tracer = OtelTracer()
+    with tracer.lifespan():
+        yield tracer
+    clear_tracer_provider()
+
+
+def test_otel_tracer_create_span_records_fields(otel_tracer: OtelTracer) -> None:
+    timestamp = 2000.123
+    status = TraceStatus(status_code="ERROR", description="boom")
+
+    span = otel_tracer.create_span(
+        "otel-span",
+        attributes={"foo": "bar"},
+        timestamp=timestamp,
+        status=status,
+    )
+
+    assert span.name == "otel-span"
+    assert span.attributes["foo"] == "bar"
+    assert span.start_time == pytest.approx(timestamp)  # type: ignore
+    assert span.end_time == pytest.approx(timestamp)  # type: ignore
+    assert span.status.status_code == "ERROR"
+    assert span.status.description == "boom"
+
+
+def test_otel_tracer_operation_context_records_span(otel_tracer: OtelTracer) -> None:
+    with pytest.raises(RuntimeError):
+        with otel_tracer.operation_context("otel-operation", attributes={"initial": "value"}) as ctx:
+            ctx.record_attributes({"custom": "attr"})
+            raise RuntimeError("otel failure")
+
+    recorded_span = ctx.get_recorded_span()  # type: ignore
+    assert recorded_span.name == "otel-operation"
+    assert recorded_span.attributes["initial"] == "value"
+    assert recorded_span.attributes["custom"] == "attr"
+    assert recorded_span.status.status_code == "ERROR"
+    assert recorded_span.status.description is not None
+    assert "otel failure" in recorded_span.status.description
+    assert recorded_span.start_time is not None
+    assert recorded_span.end_time is not None
+
+
 def test_initialization_and_shutdown():
     """Test processor lifecycle: initialization, loop thread, and shutdown."""
     processor = LightningSpanProcessor()
@@ -167,7 +224,7 @@ def test_span_collection_with_filtering():
     # Only sampled span with context should be collected
     collected = processor.spans()
     assert len(collected) == 1
-    assert collected[0] == sampled_span
+    assert collected[0].name == "sampled"
 
     processor.shutdown()
 
@@ -384,7 +441,7 @@ def test_store_write_timeout(store: MagicMock):
 
     # Create a slow async function that exceeds timeout
     async def slow_write(*args: Any, **kwargs: Any) -> None:
-        await asyncio.sleep(10)
+        await asyncio.sleep(15)
 
     store.add_otel_span = AsyncMock(side_effect=slow_write)
 
@@ -423,8 +480,8 @@ def test_multiple_processors_in_same_process():
 
     assert len(processor1.spans()) == 1
     assert len(processor2.spans()) == 1
-    assert processor1.spans()[0] == span1
-    assert processor2.spans()[0] == span2
+    assert processor1.spans()[0].name == "p1_span"
+    assert processor2.spans()[0].name == "p2_span"
 
     processor1.shutdown()
     processor2.shutdown()
@@ -462,11 +519,6 @@ def _otel_reward_subprocess(mode: str, conn: Connection[tuple[str, Any]]) -> Non
 async def _otel_reward_subprocess_async(mode: str, conn: Connection[tuple[str, Any]]) -> None:
     tracer: OtelTracer | None = None
     try:
-        try:
-            clear_agentops_init()
-        except Exception:
-            # Some environments ship a minimal agentops stub without tracer helpers.
-            pass
         clear_tracer_provider()
 
         tracer = OtelTracer()

tests/tracer/test_weave.py

@@ -3,9 +3,7 @@
 from __future__ import annotations
 
 import datetime
-import multiprocessing
 from types import SimpleNamespace
-from typing import Any, Callable, Coroutine, Sequence
 
 import pytest
 
@@ -21,14 +19,66 @@ class MockLightningStore(LightningStore):
         super().__init__()
         self.spans: list[Span] = []
 
-    async def add_many_spans(self, spans: Sequence[Span]) -> Sequence[Span]:
-        self.spans.extend(spans)
-        return spans
+    async def get_next_span_sequence_id(self, rollout_id: str, attempt_id: str) -> int:
+        return len(self.spans)
+
+    async def add_span(self, span: Span) -> Span:
+        self.spans.append(span)
+        return span
+
+    def clear_spans(self) -> None:
+        self.spans = []
 
     def get_traces(self) -> list[Span]:
         return self.spans
 
 
+class FakeWeaveCall:
+    def __init__(
+        self,
+        op: str,
+        attributes: dict[str, object] | None = None,
+        inputs: dict[str, object] | None = None,
+    ):
+        base_time = datetime.datetime.fromtimestamp(96400, tz=datetime.timezone.utc)
+        self.op_name = op
+        self.attributes = attributes or {}
+        self.inputs = inputs or {}
+        self.output = None
+        self.summary: dict[str, object] | None = {}
+        self.started_at = base_time
+        self.ended_at = base_time
+        self.exception: str | None = None
+
+
+class FakeWeaveClient:
+    def __init__(self, server: object):
+        self.server = server
+        self.project = "fake-project"
+        self.created_calls: list[FakeWeaveCall] = []
+        self.finished_calls: list[FakeWeaveCall] = []
+
+    def create_call(
+        self,
+        *,
+        op: str,
+        attributes: dict[str, object] | None = None,
+        inputs: dict[str, object] | None = None,
+    ) -> FakeWeaveCall:
+        call = FakeWeaveCall(op=op, attributes=attributes or {}, inputs=inputs or {})
+        self.created_calls.append(call)
+        return call
+
+    def finish_call(self, call: FakeWeaveCall, exception: Exception | None = None) -> None:
+        if exception is not None:
+            call.exception = str(exception)
+        call.ended_at = call.started_at + datetime.timedelta(seconds=1)
+        self.finished_calls.append(call)
+
+    def flush(self) -> None:
+        pass
+
+
 def _func_without_exception():
     """Function that always executed successfully to test success tracing."""
     pass
@@ -39,48 +89,54 @@ def _func_with_exception():
     raise ValueError("This is a test exception")
 
 
-@pytest.mark.parametrize("with_exception", [True, False])
-def test_weave_trace_workable_store_valid(with_exception: bool):
+@pytest.mark.weave
+def test_weave_tracer_create_span(monkeypatch: pytest.MonkeyPatch) -> None:
+    import weave  # type: ignore
 
-    if with_exception:
-        func = _test_weave_trace_with_exception
-    else:
-        func = _test_weave_trace_without_exception
+    tracer = WeaveTracer(instrument_managed=False)
+    fake_client = FakeWeaveClient(server=tracer._server)  # pyright: ignore[reportPrivateUsage]
+    monkeypatch.setattr(weave, "get_client", lambda: fake_client)
 
-    ctx = multiprocessing.get_context("spawn")
-    proc = ctx.Process(target=_run_async, args=(func,))
-    proc.start()
-    proc.join(30.0)  # On GPU server, the time is around 10 seconds.
+    span = tracer.create_span("weave-span", attributes={"foo": "bar"})
 
-    if proc.is_alive():
-        proc.terminate()
-        proc.join(5)
-        if proc.is_alive():
-            proc.kill()
-
-        assert False, "Child process hung. Check test output for details."
+    assert span.name == "weave-span"
+    assert span.attributes == {"foo": "bar"}
+    assert len(fake_client.created_calls) == 1
+    assert len(fake_client.finished_calls) == 1
 
 
-def _run_async(coro: Callable[[], Coroutine[Any, Any, Any]]) -> None:
-    """Small wrapper: run async function inside multiprocessing target."""
-    import asyncio
+@pytest.mark.weave
+def test_weave_tracer_operation_context_records_exception(monkeypatch: pytest.MonkeyPatch) -> None:
+    import weave  # type: ignore
 
-    asyncio.run(coro())
+    tracer = WeaveTracer(instrument_managed=False)
+    fake_client = FakeWeaveClient(server=tracer._server)  # pyright: ignore[reportPrivateUsage]
+    monkeypatch.setattr(weave, "get_client", lambda: fake_client)
+
+    with pytest.raises(RuntimeError):
+        with tracer.operation_context("weave-op", attributes={"foo": "bar"}) as ctx:
+            raise RuntimeError("failed")
+
+    recorded_span = ctx.get_recorded_span()  # type: ignore
+    assert recorded_span.name == "weave-op"
+    assert recorded_span.status.status_code == "ERROR"
+    assert recorded_span.status.description == "failed"
+    assert len(fake_client.finished_calls) == 1
 
 
-async def _test_weave_trace_without_exception():
+@pytest.mark.weave
+@pytest.mark.asyncio
+async def test_weave_trace_without_exception():
     tracer = WeaveTracer()
-    tracer.init()
-    tracer.init_worker(0)
-
     store = MockLightningStore()
+    tracer.init()
+    tracer.init_worker(0, store=store)
 
     try:
         # Case where store, rollout_id, and attempt_id are all non-none.
-        async with tracer.trace_context(
-            name="weave_test", store=store, rollout_id="test_rollout_id", attempt_id="test_attempt_id"
-        ):
+        async with tracer.trace_context(name="weave_test", rollout_id="test_rollout_id", attempt_id="test_attempt_id"):
             _func_without_exception()
+        print(tracer.get_last_trace())
         spans = store.get_traces()
         assert len(spans) > 0
 
@@ -94,19 +150,21 @@ async def _test_weave_trace_without_exception():
         tracer.teardown()
 
 
-async def _test_weave_trace_with_exception():
+@pytest.mark.weave
+@pytest.mark.asyncio
+async def test_weave_trace_with_exception():
     tracer = WeaveTracer()
-    tracer.init()
-    tracer.init_worker(0)
-
     store = MockLightningStore()
+    tracer.init()
+    tracer.init_worker(0, store=store)
 
     try:
-        # Case where store, rollout_id, and attempt_id are all non-none.
-        async with tracer.trace_context(
-            name="weave_test", store=store, rollout_id="test_rollout_id", attempt_id="test_attempt_id"
-        ):
-            _func_with_exception()
+        with pytest.raises(ValueError):
+            # Case where store, rollout_id, and attempt_id are all non-none.
+            async with tracer.trace_context(
+                name="weave_test", rollout_id="test_rollout_id", attempt_id="test_attempt_id"
+            ):
+                _func_with_exception()
         spans = store.get_traces()
         assert len(spans) > 0
 
@@ -120,22 +178,9 @@ async def _test_weave_trace_with_exception():
         tracer.teardown()
 
 
-def test_weave_with_op():
-    ctx = multiprocessing.get_context("spawn")
-    proc = ctx.Process(target=_run_async, args=(_test_weave_with_op_imp,))
-    proc.start()
-    proc.join(30.0)  # On GPU server, the time is around 10 seconds.
-
-    if proc.is_alive():
-        proc.terminate()
-        proc.join(5)
-        if proc.is_alive():
-            proc.kill()
-
-        assert False, "Child process hung. Check test output for details."
-
-
-async def _test_weave_with_op_imp():
+@pytest.mark.weave
+@pytest.mark.asyncio
+async def test_weave_with_op():
     import weave  # type: ignore
 
     @weave.op  # type: ignore
@@ -144,24 +189,19 @@ async def _test_weave_with_op_imp():
         pass
 
     tracer = WeaveTracer()
+    store = MockLightningStore()
     tracer.init()
-    tracer.init_worker(0)
+    tracer.init_worker(0, store=store)
 
     try:
-        store = MockLightningStore()
         # Case where store, rollout_id, and attempt_id are all non-none.
-        async with tracer.trace_context(
-            name="weave_test", store=store, rollout_id="test_rollout_id", attempt_id="test_attempt_id"
-        ):
+        async with tracer.trace_context(name="weave_test", rollout_id="test_rollout_id", attempt_id="test_attempt_id"):
             _func_with_op()
         spans = store.get_traces()
         len_spans_with_op = len(spans)
 
-        store = MockLightningStore()
-        # Case where store, rollout_id, and attempt_id are all non-none.
-        async with tracer.trace_context(
-            name="weave_test", store=store, rollout_id="test_rollout_id", attempt_id="test_attempt_id"
-        ):
+        store.clear_spans()
+        async with tracer.trace_context(name="weave_test", rollout_id="test_rollout_id", attempt_id="test_attempt_id"):
             _func_without_exception()
         spans = store.get_traces()
         len_spans_without_op = len(spans)
@@ -175,38 +215,30 @@ async def _test_weave_with_op_imp():
         tracer.teardown()
 
 
-def test_weave_trace_call_to_span():
-    ctx = multiprocessing.get_context("spawn")
-    proc = ctx.Process(target=_test_weave_trace_call_to_span)
-    proc.start()
-    proc.join(30.0)  # On GPU server, the time is around 10 seconds.
-
-    if proc.is_alive():
-        proc.terminate()
-        proc.join(5)
-        if proc.is_alive():
-            proc.kill()
-
-        assert False, "Child process hung. Check test output for details."
-
-
-async def _test_weave_trace_call_to_span():
+@pytest.mark.weave
+@pytest.mark.asyncio
+async def test_weave_trace_call_to_span():
     child = SimpleNamespace(
+        op_name="child_func",
         inputs={"child_input": "x"},
         output={"child_output": 42},
+        attributes={"child_attribute": "z"},
         summary={"status_counts": {"success": 1, "error": 0}},
         _children=[],
-        started_at=None,
+        started_at=datetime.datetime(2025, 12, 1, 0, 0, 1, tzinfo=datetime.timezone.utc),
         ended_at=datetime.datetime(2025, 12, 1, 0, 0, 2, tzinfo=datetime.timezone.utc),
         trace_id="trace-1",
         id="span-2",
         parent_id="span-1",
         func_name="child_func",
+        exception=None,
     )
 
     parent = SimpleNamespace(
+        op_name="parent_func",
         inputs={"parent_input": "y"},
         output={"parent_output": 99},
+        attributes={"parent_attribute": "y"},
         summary={"status_counts": {"success": 1, "error": 0}},
         _children=[child],
         started_at=datetime.datetime(2025, 12, 1, 0, 0, 0, tzinfo=datetime.timezone.utc),
@@ -215,14 +247,15 @@ async def _test_weave_trace_call_to_span():
         id="span-1",
         parent_id=None,
         func_name="parent_func",
+        exception=None,
     )
 
     tracer = WeaveTracer()
-    spans, _ = tracer.convert_call_to_spans(parent)  # type: ignore
+    parent_span = await tracer.convert_call_to_span(parent)  # type: ignore
+    assert parent_span.attributes["agentlightning.operation.input.parent_input"] == "y"
+    assert parent_span.attributes["agentlightning.operation.output.parent_output"] == 99
 
-    assert len(spans) == 2
-    assert spans[0].sequence_id == 0
-    assert spans[1].sequence_id == 1
-    assert spans[1].parent_id == "span-1"
-    assert spans[1].attributes["input.child_input"] == "x"
-    assert spans[1].attributes["output.child_output"] == 42
+    child_span = await tracer.convert_call_to_span(child)  # type: ignore
+    assert child_span.attributes["agentlightning.operation.input.child_input"] == "x"
+    assert child_span.attributes["agentlightning.operation.output.child_output"] == 42
+    assert child_span.parent_id == "span-1"

tests/utils/test_otel.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import json
 import logging
 from types import SimpleNamespace
 from typing import Any, Dict, List
@@ -9,6 +10,7 @@ from typing import Any, Dict, List
 import opentelemetry.trace as trace_api
 import pytest
 from opentelemetry.sdk.trace import ReadableSpan, SynchronousMultiSpanProcessor
+from opentelemetry.semconv.attributes import exception_attributes
 from opentelemetry.trace import TraceFlags
 from pydantic import ValidationError
 
@@ -16,16 +18,22 @@ from agentlightning.semconv import LightningSpanAttributes, LinkPydanticModel
 from agentlightning.types.tracer import Span
 from agentlightning.utils import otel
 from agentlightning.utils.otel import (
+    check_attributes_sanity,
     extract_links_from_attributes,
     extract_tags_from_attributes,
     filter_and_unflatten_attributes,
     filter_attributes,
     flatten_attributes,
+    format_exception_attributes,
+    full_qualified_name,
     get_tracer,
     get_tracer_provider,
     make_link_attributes,
     make_tag_attributes,
     query_linked_spans,
+    sanitize_attribute_value,
+    sanitize_attributes,
+    sanitize_list_attribute_sanity,
     unflatten_attributes,
 )
 
@@ -45,6 +53,15 @@ def _span_context(trace_id_hex: str, span_id_hex: str) -> trace_api.SpanContext:
 def test_flatten_simple_nested_dict_and_list() -> None:
     data = {"a": {"b": 1, "c": [2, 3]}}
     result = flatten_attributes(data)
+    assert result == {
+        "a.b": 1,
+        "a.c": [2, 3],
+    }
+
+
+def test_flatten_simple_nested_dict_and_list_with_leaf_expansion() -> None:
+    data = {"a": {"b": 1, "c": [2, 3]}}
+    result = flatten_attributes(data, expand_leaf_lists=True)
     assert result == {
         "a.b": 1,
         "a.c.0": 2,
@@ -52,6 +69,18 @@ def test_flatten_simple_nested_dict_and_list() -> None:
     }
 
 
+def test_full_qualified_name_handles_builtin_and_custom_classes() -> None:
+    class LocalClass:
+        pass
+
+    builtin_result = full_qualified_name(int)
+    local_result = full_qualified_name(LocalClass)
+
+    assert builtin_result == "int"
+    assert local_result.endswith("LocalClass")
+    assert local_result.startswith("tests.utils.test_otel")
+
+
 def test_flatten_empty_dict() -> None:
     data: Dict[str, Any] = {}
     assert flatten_attributes(data) == {}
@@ -83,13 +112,54 @@ def test_flatten_nested_lists_and_dicts() -> None:
     result = flatten_attributes(data)
     assert result == {
         "users.0.name": "Alice",
-        "users.0.tags.0": "admin",
-        "users.0.tags.1": "staff",
+        "users.0.tags": ["admin", "staff"],
         "users.1.name": "Bob",
-        # Empty list yields no extra keys
+        # Empty leaf lists remain implicit
     }
 
 
+def test_flatten_leaf_lists_can_stay_compact() -> None:
+    data = {"tags": ["fast", "reliable"]}
+    result = flatten_attributes(data, expand_leaf_lists=False)
+    assert result == {"tags": ["fast", "reliable"]}
+
+
+def test_flatten_non_leaf_lists_still_expand_when_leaf_expansion_disabled() -> None:
+    data = {"users": [{"name": "Alice"}, {"name": "Bob"}]}
+    result = flatten_attributes(data, expand_leaf_lists=False)
+    assert result == {
+        "users.0.name": "Alice",
+        "users.1.name": "Bob",
+    }
+
+
+def test_flatten_leaf_lists_with_mixed_types_warns_and_expands(caplog: pytest.LogCaptureFixture) -> None:
+    data = {"values": [1, "two"]}
+    caplog.set_level(logging.WARNING, logger="agentlightning.utils.otel")
+
+    result = flatten_attributes(data, expand_leaf_lists=False)
+
+    assert result == {
+        "values.0": 1,
+        "values.1": "two",
+    }
+    assert "mixed primitive types" in caplog.text
+
+
+def test_flatten_leaf_lists_with_mixed_numbers_warns_and_expands(caplog: pytest.LogCaptureFixture) -> None:
+    data = {"values": [1, 2.0, 3]}
+    caplog.set_level(logging.WARNING, logger="agentlightning.utils.otel")
+
+    result = flatten_attributes(data, expand_leaf_lists=False)
+
+    assert result == {
+        "values.0": 1,
+        "values.1": 2.0,
+        "values.2": 3,
+    }
+    assert "mixed primitive types" in caplog.text
+
+
 def test_flatten_mixed_types_and_none() -> None:
     data = {
         "a": True,
@@ -286,6 +356,106 @@ def test_round_trip_with_empty_list_information_loss_is_expected() -> None:
     assert reconstructed != data  # explicit documentation of the behavior
 
 
+def test_sanitize_attribute_value_handles_primitives_and_lists() -> None:
+    assert sanitize_attribute_value("text") == "text"
+    assert sanitize_attribute_value(42) == 42
+    assert sanitize_attribute_value(3.14) == 3.14
+    assert sanitize_attribute_value(True) is True
+    assert sanitize_attribute_value([True, 2]) == [1, 2]
+
+
+def test_sanitize_attribute_value_falls_back_to_json_for_hybrid_lists(caplog: pytest.LogCaptureFixture) -> None:
+    caplog.set_level(logging.WARNING, logger="agentlightning.utils.otel")
+    result = sanitize_attribute_value([1, "two"])
+    assert result == json.dumps([1, "two"])
+    assert "Failed to sanitize list attribute" in caplog.text
+
+
+def test_sanitize_attribute_value_rejects_non_serializable_objects() -> None:
+    class Unserializable:
+        pass
+
+    with pytest.raises(ValueError, match="Object must be JSON serializable"):
+        sanitize_attribute_value(Unserializable())
+
+
+def test_sanitize_attributes_returns_clean_values() -> None:
+    attributes = {
+        "name": "agent",
+        "enabled": True,
+        "scores": [1, True],
+        "flags": [True, False],
+    }
+
+    result = sanitize_attributes(attributes)
+    assert result["name"] == "agent"
+    assert result["enabled"] is True
+    assert result["scores"] == [1, 1]
+    assert result["flags"] == [True, False]
+
+
+def test_sanitize_attributes_exposes_key_in_error() -> None:
+    class Bad:
+        pass
+
+    attributes = {"ok": "value", "bad": Bad()}
+    with pytest.raises(ValueError, match="Failed to sanitize attribute 'bad'"):
+        sanitize_attributes(attributes)
+
+
+def test_format_exception_attributes_captures_metadata() -> None:
+    try:
+        raise RuntimeError("boom")
+    except RuntimeError as err:
+        attrs = format_exception_attributes(err)
+
+    assert attrs[exception_attributes.EXCEPTION_TYPE] == "RuntimeError"
+    assert attrs[exception_attributes.EXCEPTION_MESSAGE] == "boom"
+    assert attrs[exception_attributes.EXCEPTION_ESCAPED] is True
+    assert exception_attributes.EXCEPTION_STACKTRACE in attrs
+    assert "RuntimeError: boom" in attrs[exception_attributes.EXCEPTION_STACKTRACE]  # type: ignore
+
+
+def test_sanitize_list_attribute_sanity_supports_primitive_lists() -> None:
+    assert sanitize_list_attribute_sanity(["a", "b"]) == ["a", "b"]
+    assert sanitize_list_attribute_sanity([True, False]) == [True, False]
+    assert sanitize_list_attribute_sanity([1, False]) == [1, 0]
+    assert sanitize_list_attribute_sanity([1.0, 2, True]) == [1.0, 2.0, 1.0]
+
+
+def test_sanitize_list_attribute_sanity_rejects_mixed_types() -> None:
+    with pytest.raises(ValueError, match="List must contain only one type of primitive values"):
+        sanitize_list_attribute_sanity([1, "two"])
+
+
+def test_check_attributes_sanity_accepts_valid_payload() -> None:
+    attributes: Dict[str, Any] = {
+        "name": "agent",
+        "count": 3,
+        "ratio": 0.5,
+        "enabled": False,
+        "flags": [True, False],
+        "scores": [1, True],
+    }
+
+    check_attributes_sanity(attributes)
+
+
+def test_check_attributes_sanity_requires_string_keys() -> None:
+    with pytest.raises(ValueError, match="Attribute key must be a string"):
+        check_attributes_sanity({1: "value"})  # type: ignore[arg-type]
+
+
+def test_check_attributes_sanity_wraps_list_errors() -> None:
+    with pytest.raises(ValueError, match="Failed to sanitize list attribute 'mixed'"):
+        check_attributes_sanity({"mixed": [1, "two"]})
+
+
+def test_check_attributes_sanity_rejects_non_primitive_values() -> None:
+    with pytest.raises(ValueError, match="Attribute value must be a string"):
+        check_attributes_sanity({"bad": {"nested": "value"}})
+
+
 def test_make_and_extract_link_attributes_round_trip() -> None:
     flattened = make_link_attributes(
         {