.

.github/workflows/examples-calc-x.yml

@@ -171,12 +171,12 @@ jobs:
       - name: Sync dependencies (latest)
         run: |
           uv sync --frozen --no-default-groups --extra verl \
-            --group dev --group experiment --group agents --group torch-gpu-stable
+            --group dev --group experiment --group agents --extra weave --group torch-gpu-stable
         if: matrix.setup-script == 'latest'
       - name: Sync dependencies (stable & legacy)
         run: |
           uv sync --frozen --no-default-groups --extra verl \
-            --group dev --group experiment --group agents --group torch-gpu-${{ matrix.setup-script }}
+            --group dev --group experiment --group agents --extra weave --group torch-gpu-${{ matrix.setup-script }}
         if: matrix.setup-script != 'latest'
       - name: Freeze dependencies
         run: |
@@ -295,6 +295,29 @@ jobs:
           WANDB_API_KEY: ${{ secrets.MSR_WANDB_API_KEY }}
         if: matrix.setup-script != 'legacy'
 
+      - name: Training with Weave
+        run: |
+          set -ex
+          source .venv/bin/activate
+          cd examples/calc_x
+          ../../scripts/restart_ray.sh
+          sleep 5
+          PYTHONUNBUFFERED=1 python train_calc_agent.py --val-file data/test_mini.parquet --ci-fast --weave
+          sleep 10
+        shell: bash
+        env:
+          WANDB_BASE_URL: ${{ secrets.MSR_WANDB_BASE_URL }}
+          WANDB_API_KEY: ${{ secrets.MSR_WANDB_API_KEY }}
+        id: calc_x_train_weave
+
+      - name: Validate training with Weave
+        run: |
+          set -ex
+          uv run scripts/validate_example_wandb.py ${{ steps.calc_x_train_weave.outputs.project_name }} ${{ steps.calc_x_train_weave.outputs.run_name }}
+        env:
+          WANDB_BASE_URL: ${{ secrets.MSR_WANDB_BASE_URL }}
+          WANDB_API_KEY: ${{ secrets.MSR_WANDB_API_KEY }}
+
       - name: Training with external store
         run: |
           set -euo pipefail

.github/workflows/tests-full.yml

@@ -187,7 +187,7 @@ jobs:
       # mongo, openai, gpu, all enabled by default
       - name: Run tests
         run: |
-          uv run pytest -v --durations=0 tests -m "${{ matrix.mark.pytest-mark }}"
+          uv run pytest -v --durations=0 tests -m "${{ matrix.mark.pytest-mark }}${{ matrix.env.setup-script == 'legacy' && ' and not langchain' || '' }}"
         env:
           PYTEST_ADDOPTS: "--color=yes"
           OPENAI_BASE_URL: http://localhost:12306/

.gitignore

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

agentlightning/adapter/triplet.py

@@ -12,6 +12,7 @@ from opentelemetry.sdk.trace import ReadableSpan
 from pydantic import BaseModel
 
 from agentlightning.emitter.reward import get_reward_value
+from agentlightning.semconv import AGL_OPERATION, AGL_REWARD, LightningSpanAttributes
 from agentlightning.types import Span, Triplet
 from agentlightning.utils.otel import filter_and_unflatten_attributes
 
@@ -20,6 +21,47 @@ from .base import TraceAdapter
 logger = logging.getLogger(__name__)
 
 
+def _attributes_get_multiple(attributes: Dict[str, Any], keys: List[str]) -> Optional[str]:
+    """Get a string from the attributes, if present.
+    If there are multiple matches, the first one is returned.
+    """
+    for key in keys:
+        if key in attributes:
+            if isinstance(attributes[key], str):
+                return attributes[key]
+            else:
+                logger.warning(f"Attribute {key} is found but is not a string: {attributes[key]}")
+    return None
+
+
+def _attributes_get_ids_multiple(attributes: Dict[str, Any], keys: List[str]) -> Optional[List[int]]:
+    """Get a list of integers from the attributes, if present.
+    If there are multiple matches, the first one is returned.
+    """
+    for key in keys:
+        if key in attributes:
+            if (isinstance(attributes[key], list) or isinstance(attributes[key], tuple)) and all(
+                isinstance(x, int) for x in attributes[key]
+            ):
+                return list(attributes[key])
+            else:
+                logger.warning(f"Attribute {key} is found but is not a list of integers: {attributes[key]}")
+    return None
+
+
+def _attributes_unflatten_multiple(
+    attributes: Dict[str, Any], keys: List[str]
+) -> Union[Dict[str, Any], List[Any], None]:
+    """Unflatten the attributes, if present.
+    If there are multiple matches, the first one is returned.
+    """
+    for key in keys:
+        result = filter_and_unflatten_attributes(attributes, key)
+        if result:
+            return result
+    return None
+
+
 class Transition(BaseModel):
     """A single transition within a reinforcement learning trajectory.
 
@@ -132,7 +174,7 @@ class TraceTree:
             if not should_visit(node):
                 return False
             agent_name = node.agent_name()
-            vis_name = node.id[:8] + " (" + node.span.name + ")"
+            vis_name = node.id[-8:] + " (" + node.span.name + ")"
             if agent_name is not None:
                 vis_name += " [" + agent_name + "]"
             dot.node(node.id, vis_name)  # type: ignore
@@ -309,6 +351,19 @@ class TraceTree:
         if agent_name is not None:
             return agent_name
 
+        # Case 6: Weave
+        is_agent_type = attributes.get("type") == "agent"
+        if is_agent_type:
+            agent_name = cast(Optional[str], attributes.get("agentlightning.operation.input.name"))
+            if agent_name is not None:
+                return agent_name
+
+        # Case 7: Weave + LangChain
+        if self.span.name.startswith("langchain.Chain."):
+            attributes_lc_name = cast(Optional[str], attributes.get("lc_name"))
+            if attributes_lc_name is not None:
+                return attributes_lc_name
+
     def maybe_reward_dict(self) -> dict[str, Any]:
         """Return a reward payload if the span encodes one.
 
@@ -328,7 +383,17 @@ class TraceTree:
             `True` when the span payload describes a reward, otherwise `False`.
         """
         maybe_reward = self.maybe_reward_dict()
-        return maybe_reward and maybe_reward.get("type") == "reward"  # type: ignore
+        if maybe_reward and maybe_reward.get("type") == "reward":  # type: ignore
+            return True
+
+        # Agent-lightning 0.3+
+        if (
+            self.span.name == AGL_OPERATION
+            and self.span.attributes.get(LightningSpanAttributes.OPERATION_NAME.value) == AGL_REWARD
+        ):
+            return True
+
+        return False
 
     def find_llm_calls(
         self,
@@ -365,7 +430,9 @@ class TraceTree:
             is_llm_call = False
         if is_llm_call:
             # Check the response id
-            response_id: Optional[str] = self.span.attributes.get("gen_ai.response.id")  # type: ignore
+            response_id = _attributes_get_multiple(
+                self.span.attributes, ["gen_ai.response.id", "agentlightning.operation.output.id"]
+            )
             if response_id is None and within_llm_call is True:
                 is_llm_call = False
             if (
@@ -547,7 +614,7 @@ class TraceTree:
                 try:
                     content = json.loads(content)  # This content should now be a list
                 except json.JSONDecodeError:
-                    logger.warning(f"Failed to parse message content as JSON: {content}")
+                    logger.debug(f"Failed to parse message content as JSON: {content}")
                     continue
             if isinstance(content, list):
                 for content_part in cast(List[Dict[str, Any]], content):
@@ -567,18 +634,57 @@ class TraceTree:
         Subclass can override this method to add more fields to the triplet,
         such as chat messages and tool calls.
         """
-        prompt_token_ids = span.attributes.get("prompt_token_ids", [])  # type: ignore
-        response_token_ids = span.attributes.get("response_token_ids", [])  # type: ignore
-        response_id = span.attributes.get("gen_ai.response.id", None)  # type: ignore
-        request_metadata = filter_and_unflatten_attributes(span.attributes, "gen_ai.request")
-        response_metadata = filter_and_unflatten_attributes(span.attributes, "gen_ai.response")
-        prompt_raw_content = filter_and_unflatten_attributes(span.attributes, "gen_ai.prompt")
-        completion_raw_content = filter_and_unflatten_attributes(span.attributes, "gen_ai.completion")
-        image_urls = self.extract_prompt_image_urls(prompt_raw_content)
+        prompt_token_ids = (
+            _attributes_get_ids_multiple(
+                span.attributes,
+                [
+                    "prompt_token_ids",
+                    "agentlightning.operation.output.prompt_token_ids",  # Weave tracer
+                ],
+            )
+            or []
+        )
+        response_token_ids = (
+            _attributes_get_ids_multiple(
+                span.attributes,
+                [
+                    "response_token_ids",
+                    "agentlightning.operation.output.response_token_ids.0",  # Weave tracer
+                    "agentlightning.operation.output.choices.0.token_ids",  # Weave tracer with newer vLLM
+                    "agentlightning.operation.output.choices.0.provider_specific_fields.token_ids",  # new vLLM + new OpenAI client SDK
+                ],
+            )
+            or []
+        )
 
+        response_id = _attributes_get_multiple(
+            span.attributes, ["gen_ai.response.id", "agentlightning.operation.output.id"]
+        )
+        request_metadata = _attributes_unflatten_multiple(
+            span.attributes, ["gen_ai.request", "agentlightning.operation.input"]
+        )
+        response_metadata = _attributes_unflatten_multiple(
+            span.attributes, ["gen_ai.response", "agentlightning.operation.output"]
+        )
+        # Special handling for Weave tracer: messages are handled separately
+        if isinstance(request_metadata, dict):
+            request_metadata.pop("messages", None)
+        if isinstance(response_metadata, dict):
+            response_metadata.pop("choices", None)
+            response_metadata.pop("prompt_token_ids", None)
+            response_metadata.pop("response_token_ids", None)
+
+        prompt_raw_content = _attributes_unflatten_multiple(
+            span.attributes, ["gen_ai.prompt", "agentlightning.operation.input.messages"]
+        )
+        completion_raw_content = _attributes_unflatten_multiple(
+            span.attributes, ["gen_ai.completion", "agentlightning.operation.output.choices"]
+        )
+        image_urls = self.extract_prompt_image_urls(prompt_raw_content)
         prompt_payload = {"token_ids": prompt_token_ids, "raw_content": prompt_raw_content, "image_urls": image_urls}
         response_payload = {"token_ids": response_token_ids, "raw_content": completion_raw_content}
 
+        # FIXME: logprob doesn't support Weave tracer yet.
         logprobs_content = span.attributes.get("logprobs.content", None)  # type: ignore
         if isinstance(logprobs_content, str):
             logprobs_content = json.loads(logprobs_content)

agentlightning/emitter/annotation.py

@@ -274,17 +274,38 @@ class OperationContext:
 
 
 @overload
-def operation(fn: _FnType, *, propagate: bool = True, **additional_attributes: Any) -> _FnType: ...
+def operation(
+    fn: _FnType, *, propagate: bool = True, name: Optional[str] = None, **additional_attributes: Any
+) -> _FnType: ...
 
 
 @overload
-def operation(*, propagate: bool = True, **additional_attributes: Any) -> OperationContext: ...
+def operation(
+    *, propagate: bool = True, name: Optional[str] = None, **additional_attributes: Any
+) -> OperationContext: ...
+
+
+@overload
+def operation(fn: _FnType, *, name: Optional[str] = None, **additional_attributes: Any) -> _FnType: ...
+
+
+@overload
+def operation(*, name: Optional[str] = None, **additional_attributes: Any) -> OperationContext: ...
+
+
+@overload
+def operation(fn: _FnType, **additional_attributes: Any) -> _FnType: ...
+
+
+@overload
+def operation(**additional_attributes: Any) -> OperationContext: ...
 
 
 def operation(
     fn: Optional[_FnType] = None,
     *,
     propagate: bool = True,
+    name: Optional[str] = None,
     **additional_attributes: Any,
 ) -> Union[_FnType, OperationContext]:
     """Entry point for tracking operations.
@@ -320,6 +341,9 @@ def operation(
             left as `None`) and only keyword attributes are provided.
         propagate: Whether spans should use the active span processor. When False,
             spans will stay local and not be exported.
+        name: Optional alias that populates
+            [`LightningSpanAttributes.OPERATION_NAME`][agentlightning.semconv.LightningSpanAttributes.OPERATION_NAME]
+            when `additional_attributes` does not already define it.
         **additional_attributes: Additional span attributes to attach at
             creation time.
 
@@ -328,6 +352,12 @@ def operation(
         [`OperationContext`][agentlightning.emitter.annotation.OperationContext]
         (when used as a context manager factory).
     """
+
+    if name is not None:
+        if LightningSpanAttributes.OPERATION_NAME.value in additional_attributes:
+            raise ValueError("Cannot specify both `name` and `additional_attributes.operation_name`.")
+        additional_attributes[LightningSpanAttributes.OPERATION_NAME.value] = name
+
     # Case 1: Used as @operation (bare decorator or with attributes)
     if callable(fn):
         # Create context with fixed name, then immediately wrap the function

agentlightning/instrumentation/weave.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import logging
 import threading
+import warnings
 from datetime import datetime, timezone
 from typing import Any, Callable, Dict, Iterator, List
 
@@ -442,7 +443,11 @@ def get_entity_project_from_project_name_factory(entity_name: str) -> tuple[str,
     # Bypass the usage of API
     try:
         assert _original_get_entity_project_from_project_name is not None
-        return _original_get_entity_project_from_project_name(entity_name)
+        if _original_get_entity_project_from_project_name is not get_entity_project_from_project_name_factory:
+            return _original_get_entity_project_from_project_name(entity_name)
+        else:
+            warnings.warn("W&B integration might have been repeatedly/recursively instrumented.")
+            return "agl", "weave"
     except weave.trace.weave_init.WeaveWandbAuthenticationException:
         # In case API is not available.
         return "agl", "weave"

agentlightning/semconv.py

@@ -34,6 +34,9 @@ AGL_OPERATION = "agentlightning.operation"
 Wrap function or code-blocks as operations.
 """
 
+AGL_REWARD = "agentlightning.reward"
+"""Agent-lightning's standard span name for reward operations."""
+
 AGL_VIRTUAL = "agentlightning.virtual"
 """Agent-lightning's standard span name for virtual operations.
 

agentlightning/tracer/weave.py

@@ -172,11 +172,16 @@ class WeaveTracerManagedTraceServer(InMemoryWeaveTraceServer):
         super().__init__()
         self.partial_call_callback = partial_call_callback
         self.complete_call_callback = complete_call_callback
+        self._calls_already_invoked: set[str] = set()
 
     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])
+                if call_id not in self._calls_already_invoked:
+                    self._calls_already_invoked.add(call_id)
+                    self.complete_call_callback(self.calls[call_id])
+                else:
+                    logger.info(f"Call {call_id} has callback already invoked. Skipping.")
             elif call_id in self.partial_calls:
                 self.partial_call_callback(self.partial_calls[call_id])
             else:
@@ -200,6 +205,9 @@ class WeaveTracerManagedTraceServer(InMemoryWeaveTraceServer):
             logger.exception(f"Error calling call_end: {req}", exc_info=True)
             raise
 
+    def clear(self) -> None:
+        self._calls_already_invoked.clear()
+
 
 class WeaveTracer(Tracer):
     """Tracer implementation using Weave for telemetry and trace logging.
@@ -380,6 +388,7 @@ class WeaveTracer(Tracer):
                 # Mandatory cleanup
                 self._rollout_id = None
                 self._attempt_id = None
+                self._server.clear()
 
     def create_span(
         self,
@@ -550,7 +559,12 @@ class WeaveTracer(Tracer):
             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.")
+            if call.id in self._calls:
+                logger.warning(
+                    f"Call {call.id} is already in calls. The call is already completed. Overwriting the call."
+                )
+            else:
+                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

agentlightning/utils/otel.py

@@ -454,7 +454,7 @@ def unflatten_attributes(flat_data: Dict[str, Any]) -> Union[Dict[str, Any], Lis
     return convert(root)
 
 
-def sanitize_attribute_value(object: Any) -> AttributeValue:
+def sanitize_attribute_value(object: Any, force: bool = True) -> AttributeValue:
     """Sanitize an attribute value to be a valid OpenTelemetry attribute value."""
     if isinstance(object, (str, int, float, bool)):
         return object
@@ -467,18 +467,23 @@ def sanitize_attribute_value(object: Any) -> AttributeValue:
 
     try:
         # This include null, dict, etc.
-        serialized = json.dumps(object)
+        serialized = json.dumps(object, default=str if force else None)
     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."""
+def sanitize_attributes(attributes: Dict[str, Any], force: bool = True) -> Attributes:
+    """Sanitize a dictionary of attributes to be a valid OpenTelemetry attributes.
+
+    Args:
+        attributes: A dictionary of attributes to sanitize.
+        force: Whether to force sanitization even when the value is not JSON serializable.
+    """
     result: Attributes = {}
     for k, v in attributes.items():
         try:
-            result[k] = sanitize_attribute_value(v)
+            result[k] = sanitize_attribute_value(v, force=force)
         except ValueError as exc:
             raise ValueError(f"Failed to sanitize attribute '{k}': {exc}") from exc
     return result

examples/calc_x/train_calc_agent.py

@@ -120,6 +120,7 @@ def train(
     lora: bool,
     lora_rank: int,
     lora_adapter_path: Optional[str],
+    weave: bool,
 ):
     """The training entrypoint function for Calc-X agent with VERL algorithm.
 
@@ -135,6 +136,7 @@ def train(
         lora: Whether to enable LoRA training.
         lora_rank: LoRA rank to use when LoRA is enabled.
         lora_adapter_path: Optional path to a pre-trained LoRA adapter to load.
+        weave: Whether to enable Weave tracing.
     """
     # Load datasets (respect CLI file paths)
     train_dataset = cast(agl.Dataset[MathProblem], HuggingFaceDataset.from_parquet(train_file).to_list())  # type: ignore
@@ -209,6 +211,11 @@ def train(
         tracer = agl.OtelTracer()  # dummy tracer for LLM Proxy
         adapter = agl.LlmProxyTraceToTriplet()
         trainer = agl.Trainer(algorithm=algorithm, n_runners=n_runners, store=store, tracer=tracer, adapter=adapter)
+    elif weave:
+        from agentlightning.tracer.weave import WeaveTracer
+
+        tracer = WeaveTracer()
+        trainer = agl.Trainer(algorithm=algorithm, n_runners=n_runners, store=store, tracer=tracer)
     else:
         trainer = agl.Trainer(algorithm=algorithm, n_runners=n_runners, store=store)
 
@@ -221,6 +228,7 @@ def main():
     parser.add_argument("--val-file", type=str, default="data/test.parquet", help="Path to val parquet file")
     parser.add_argument("--model", type=str, default=None, help="HF model id or path (optional)")
     parser.add_argument("--llm-proxy", action="store_true", help="Enable LLM Proxy tracing/adapter")
+    parser.add_argument("--weave", action="store_true", help="Enable Weave tracing")
     parser.add_argument("--ci", action="store_true", help="Run a minimal CI-style training loop")
     parser.add_argument(
         "--ci-fast", action="store_true", help="Limit the training loop to a single step (implies --ci)"
@@ -278,6 +286,7 @@ def main():
         lora=args.lora,
         lora_rank=args.lora_rank,
         lora_adapter_path=args.lora_adapter_path,
+        weave=args.weave,
     )
 
 

pyproject.toml

@@ -355,6 +355,7 @@ markers = [
   "store: tests for agentlightning.store module",
   "prometheus: tests that require Prometheus",
   "utils: tests for utility functions",
+  "langchain: tests that require LangChain",
 ]
 
 [tool.black]

tests/emitter/test_operation.py

@@ -7,7 +7,7 @@ import time
 from contextlib import contextmanager
 from dataclasses import dataclass
 from types import TracebackType
-from typing import Any, ContextManager, Dict, Iterator, List, Optional, Tuple, Type
+from typing import Any, ContextManager, Dict, Iterator, List, Optional, Tuple, Type, cast
 
 import opentelemetry.trace as trace_api
 import pytest
@@ -208,17 +208,22 @@ def test_operation_context_set_input_expands_positional_attributes(monkeypatch:
     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)
+def test_operation_context_serializes_non_serializable_output(monkeypatch: pytest.MonkeyPatch) -> None:
+    tracer = _install_recording_tracer(monkeypatch)
 
-    class Unserializable:
-        pass
+    class CustomObject:
+        def __str__(self) -> str:
+            return "custom-output"
 
     ctx = OperationContext("ctx", {})
 
-    with pytest.raises(ValueError, match="Object must be JSON serializable"):
-        with ctx as op:
-            op.set_output(Unserializable())
+    with ctx as op:
+        op.set_output(CustomObject())
+
+    recording = tracer.recordings[-1]
+    assert (
+        json.loads(cast(str, recording.attributes[LightningSpanAttributes.OPERATION_OUTPUT.value])) == "custom-output"
+    )
 
 
 def test_operation_context_records_exceptions(monkeypatch: pytest.MonkeyPatch) -> None:
@@ -252,6 +257,16 @@ def test_operation_factory_context_records_inputs_and_outputs(monkeypatch: pytes
     assert _resolve_attr(recording, LightningSpanAttributes.OPERATION_OUTPUT.value) == ["beta", "gamma"]
 
 
+def test_operation_factory_aliases_name_attribute(monkeypatch: pytest.MonkeyPatch) -> None:
+    tracer = _install_recording_tracer(monkeypatch)
+
+    with operation(name="custom-operation") as ctx:
+        ctx.set_output("done")
+
+    recording = tracer.recordings[-1]
+    assert recording.attributes[LightningSpanAttributes.OPERATION_NAME.value] == "custom-operation"
+
+
 def test_operation_factory_uses_standard_span_name(monkeypatch: pytest.MonkeyPatch) -> None:
     tracer = _install_recording_tracer(monkeypatch)
 
@@ -289,6 +304,18 @@ def test_operation_decorator_sync_records_span_attributes(monkeypatch: pytest.Mo
     assert _resolve_attr(recording, LightningSpanAttributes.OPERATION_OUTPUT.value) == result
 
 
+def test_operation_decorator_aliases_operation_name_attribute(monkeypatch: pytest.MonkeyPatch) -> None:
+    tracer = _install_recording_tracer(monkeypatch)
+
+    @operation(name="explicit-name")
+    def compute(value: int) -> int:
+        return value * 2
+
+    assert compute(3) == 6
+    recording = tracer.recordings[-1]
+    assert recording.attributes[LightningSpanAttributes.OPERATION_NAME.value] == "explicit-name"
+
+
 def test_operation_decorator_handles_complex_signature(monkeypatch: pytest.MonkeyPatch) -> None:
     tracer = _install_recording_tracer(monkeypatch)
 
@@ -305,8 +332,7 @@ def test_operation_decorator_handles_complex_signature(monkeypatch: pytest.Monke
     ) -> ComplexResult:
         return ComplexResult(values=(first, len(extra), len(rest)), marker=kwonly + kwdefault + required)
 
-    with pytest.raises(ValueError):
-        complicated(1, "req", 7, 8, 9, kwonly="x", kwdefault="y", tag="value")
+    result = complicated(1, "req", 7, 8, 9, kwonly="x", kwdefault="y", tag="value")
 
     recording = tracer.recordings[-1]
     input_prefix = LightningSpanAttributes.OPERATION_INPUT.value
@@ -319,7 +345,23 @@ def test_operation_decorator_handles_complex_signature(monkeypatch: pytest.Monke
     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"
+    assert _resolve_attr(recording, LightningSpanAttributes.OPERATION_OUTPUT.value) == (
+        "ComplexResult(values=(1, 2, 1), marker='xyreq')"
+    )
+    assert isinstance(result, ComplexResult)
+    assert recording.status.status_code == "OK"
+
+
+def test_operation_name_alias_does_not_override_explicit_attribute(monkeypatch: pytest.MonkeyPatch) -> None:
+    _install_recording_tracer(monkeypatch)
+
+    attrs = {
+        LightningSpanAttributes.OPERATION_NAME.value: "explicit-name",
+    }
+
+    with pytest.raises(ValueError, match="specify both"):
+        with operation(name="alias-name", **attrs) as ctx:
+            ctx.set_output("done")
 
 
 def test_operation_decorator_records_exceptions(monkeypatch: pytest.MonkeyPatch) -> None:

tests/tracer/test_integration.py

@@ -23,7 +23,6 @@ import os
 import pprint
 import re
 import shutil
-import textwrap
 import time
 import warnings
 from dataclasses import dataclass
@@ -41,16 +40,6 @@ from autogen_agentchat.teams import RoundRobinGroupChat
 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
-from agentlightning.reward import reward
-from agentlightning.tracer import Tracer
-from agentlightning.tracer.agentops import AgentOpsTracer
-from agentlightning.types import Span
-from agentlightning.utils.server_launcher import PythonServerLauncher, PythonServerLauncherArgs
 
 try:
     import langchain  # type: ignore
@@ -71,6 +60,19 @@ if TYPE_CHECKING or LANGCHAIN_INSTALLED:
     from langchain_openai import ChatOpenAI
     from langgraph.graph import END, START, MessagesState, StateGraph
 
+from openai import OpenAI
+from opentelemetry.sdk.trace import ReadableSpan
+from pydantic import BaseModel
+
+from agentlightning.adapter.triplet import TracerTraceToTriplet, TraceTree
+from agentlightning.emitter.annotation import operation
+from agentlightning.emitter.reward import emit_reward
+from agentlightning.semconv import AGL_REWARD
+from agentlightning.tracer import Tracer
+from agentlightning.tracer.agentops import AgentOpsTracer
+from agentlightning.types import Span
+from agentlightning.utils.server_launcher import PythonServerLauncher, PythonServerLauncherArgs
+
 USE_OPENAI = os.environ.get("USE_OPENAI", "false").lower() == "true"
 OPENAI_MODEL = "gpt-4.1-mini"
 OPENAI_BASE_URL = os.getenv("OPENAI_BASE_URL")
@@ -431,14 +433,11 @@ async def openai_agents_sdk_eval_hook_and_guardrail(settings: OpenAISettings, tr
         reasoning: str
 
     class EvalHook(AgentHooks):
-        @reward
-        def evaluate(self, context: Any, agent: Agent, output: Any):
-            # Custom reward logic: reward if the answer contains 'homework'
-            return 1.0 if output and "no" in str(output).lower() else 0.0
-
         async def on_end(self, context: Any, agent: Agent, output: Any):
-            nonlocal final_reward
-            final_reward = final_reward or self.evaluate(context, agent, output)
+            # Custom reward logic: reward if the answer contains 'no'
+            final_reward = 1.0 if output and "no" in str(output).lower() else 0.0
+            emit_reward(final_reward)
+            final_rewards.append(final_reward)
 
     guardrail_agent = Agent(
         name="Guardrail check",
@@ -463,14 +462,16 @@ async def openai_agents_sdk_eval_hook_and_guardrail(settings: OpenAISettings, tr
         input_guardrails=[InputGuardrail(guardrail_function=homework_guardrail)],
         hooks=EvalHook(),
     )
-    final_reward = None
+    final_rewards: List[float] = []
     result = await Runner.run(
         main_agent,
         "The teacher asks to answer whether hummingbirds are mammals.",
         run_config=openai_agents_sdk_run_config(settings),
     )
     # Should trigger the guardrail and reward should be 1.0
-    assert final_reward == 1.0, f"Expected reward to be 1.0, got {final_reward}"
+    assert any(
+        final_reward == 1.0 for final_reward in final_rewards
+    ), f"Expected reward to have 1.0, got {final_rewards}"
     assert hasattr(result, "final_output")
 
 
@@ -489,7 +490,6 @@ async def openai_agents_sdk_mcp_tool_use(settings: OpenAISettings, tracer: Trace
 
 
 async def openai_agents_sdk_handoff_tool_output_type_and_reward(settings: OpenAISettings, tracer: Tracer) -> None:
-
     class MathOutput(BaseModel):
         answer: int
 
@@ -498,8 +498,9 @@ async def openai_agents_sdk_handoff_tool_output_type_and_reward(settings: OpenAI
         return a + b
 
     class RewardHook(AgentHooks):
-        @reward
-        async def evaluate(self, context: Any, agent: Agent, output: Any):
+        @operation(name=AGL_REWARD)
+        async def on_end(self, context: Any, agent: Agent, output: Any):
+            nonlocal final_reward
             # Use another agent to check the answer and compute reward
             checker = Agent(
                 name="Checker",
@@ -509,11 +510,8 @@ async def openai_agents_sdk_handoff_tool_output_type_and_reward(settings: OpenAI
             result = await Runner.run(
                 checker, str(getattr(output, "answer", "")), run_config=openai_agents_sdk_run_config(settings)
             )
-            return float(result.final_output)
-
-        async def on_end(self, context: Any, agent: Agent, output: Any):
-            nonlocal final_reward
-            final_reward = await self.evaluate(context, agent, output)
+            final_reward = float(result.final_output)
+            emit_reward(final_reward)
 
     math_agent = Agent(
         name="MathAgent",
@@ -564,19 +562,19 @@ AgentName = Literal[
 ]
 
 
-AGENTOPS_EXPECTED_TREES: Mapping[AgentName, List[Tuple[str, str]]] = {
+AGENTOPS_EXPECTED_TREES: Mapping[AgentName, List[Tuple[Union[str, re.Pattern[str]], Union[str, re.Pattern[str]]]]] = {
     "agent_pure_openai": [("openai.chat.completion", "openai.chat.completion")],
     "agent_litellm": [("openai.chat.completion", "openai.chat.completion")],
     "agent_langchain": [("openai.chat.completion", "openai.chat.completion")],
     "agent_langchain_tooluse": [
-        ("chat_model.llm", "openai.chat.completion"),
-        ("chat_model.llm", "openai.chat.completion"),
+        (re.compile(r"(chat_model\.llm)|(model)"), "openai.chat.completion"),
+        (re.compile(r"(chat_model\.llm)|(model)"), "openai.chat.completion"),
     ],
     "agent_langgraph": [
         ("call_get_schema", "openai.chat.completion"),
         ("generate_query", "openai.chat.completion"),
         ("check_query", "openai.chat.completion"),
-        ("run_query", "tool.tool"),
+        ("run_query", re.compile(r"(tool.tool)|(sql_db_query)")),
     ],
     "agent_autogen_multiagent": [
         ("primary", "openai.chat.completion"),
@@ -586,9 +584,9 @@ AGENTOPS_EXPECTED_TREES: Mapping[AgentName, List[Tuple[str, str]]] = {
         ("calc_agent", "openai.chat.completion"),
     ],
     "openai_agents_sdk_eval_hook_and_guardrail": [
-        ("homework_guardrail", "openai.chat.completion"),
+        (re.compile(r"(homework_guardrail)|(Guardrail check)"), "openai.chat.completion"),
         ("Main Agent", "openai.chat.completion"),
-        ("Main Agent", "agentops_reward_operation.task"),
+        ("Main Agent", "agentlightning.annotation"),
     ],
     "openai_agents_sdk_mcp_tool_use": [
         ("MCP Tool Agent", "openai.chat.completion"),
@@ -599,7 +597,7 @@ AGENTOPS_EXPECTED_TREES: Mapping[AgentName, List[Tuple[str, str]]] = {
         ("TriageAgent", "openai.chat.completion"),
         ("MathAgent", "openai.chat.completion"),
         ("MathAgent", "openai.chat.completion"),
-        ("MathAgent", "agentops_reward_operation.task"),
+        ("MathAgent", "agentlightning.annotation"),
         ("HistoryAgent", "openai.chat.completion"),
     ],
 }
@@ -637,14 +635,22 @@ AGENT_FUNCTIONS: Mapping[AgentName, Callable[[OpenAISettings, Tracer], Awaitable
 }
 
 
-def assert_expected_pairs_in_tree(root_tuple: Tuple[str, List[Any]], expected_pairs: List[Tuple[str, str]]) -> None:
+def assert_expected_pairs_in_tree(
+    root_tuple: Tuple[str, List[Any]],
+    expected_pairs: List[Tuple[Union[str, re.Pattern[str]], Union[str, re.Pattern[str]]]],
+) -> None:
     """
     Assert that every (ancestor_name, child_name) pair in `expected_pairs`
     occurs somewhere in the tree produced by TraceTree.names_tuple().
     """
 
+    expected_patterns = [
+        (re.compile(re.escape(x)) if isinstance(x, str) else x, re.compile(re.escape(y)) if isinstance(y, str) else y)
+        for x, y in expected_pairs
+    ]
+
     # Collect every node's full path from root → node
-    paths: list[tuple[str, ...]] = []  # e.g. [["root", "A", "B"], …]
+    paths: list[tuple[str, ...]] = []  # e.g. [["root", "A", "B"], ...]
 
     def _collect(node_tuple: tuple[str, Any], prefix: list[str]):
         name, children = node_tuple
@@ -656,57 +662,86 @@ def assert_expected_pairs_in_tree(root_tuple: Tuple[str, List[Any]], expected_pa
     _collect(root_tuple, [])
 
     # Greedy—but safe—matching of each expected pair
-    used_child_paths: set[tuple[str, ...]] = set()
+    paths_used: list[bool] = [False] * len(paths)
 
-    for anc_name, child_name in expected_pairs:
+    for anc_name, child_name in expected_patterns:
         matched = False
-        for p in paths:
-            if child_name not in p[-1] or tuple(p) in used_child_paths:
+        for i, (p, used) in enumerate(zip(paths, paths_used, strict=True)):
+            if child_name.search(p[-1]) is None or used:
                 continue
-            if any(anc_name in pv for pv in p[:-1]):  # ancestor appears anywhere above
-                used_child_paths.add(tuple(p))
+            if any(anc_name.search(pv) is not None for pv in p):  # ancestor appears anywhere above (including itself)
+                paths_used[i] = True
                 matched = True
                 break
         if not matched:
-            raise AssertionError(
+            err_msg = (
                 f"Expected ancestor/child pair ({anc_name!r}, {child_name!r}) "
                 "not found or child already matched.\n"
-                f"Root tuple: {pprint.pformat(root_tuple)}\n",
-                f"Expected pairs: {expected_pairs}",
+                f"Root paths: {pprint.pformat(paths)}\n"
+                f"Expected pairs: {expected_pairs}"
             )
+            print(err_msg)
+            raise AssertionError(err_msg)
+
+
+@pytest.fixture(
+    params=[
+        "agent_pure_openai",
+        "agent_litellm",
+        pytest.param("agent_langchain", marks=pytest.mark.langchain),
+        pytest.param("agent_langchain_tooluse", marks=pytest.mark.langchain),
+        pytest.param("agent_langgraph", marks=pytest.mark.langchain),
+        "agent_autogen_multiagent",
+        "agent_autogen_mcp",
+        "openai_agents_sdk_eval_hook_and_guardrail",
+        "openai_agents_sdk_mcp_tool_use",
+        "openai_agents_sdk_handoff_tool_output_type_and_reward",
+    ]
+)
+def agent_function(
+    request: pytest.FixtureRequest,
+) -> Tuple[AgentName, Callable[[OpenAISettings, Tracer], Awaitable[Any]]]:
+    return request.param, AGENT_FUNCTIONS[request.param]
 
 
 @pytest.mark.agentops
-@pytest.mark.parametrize("agent_name", list(AGENT_FUNCTIONS.keys()), ids=str)
 @pytest.mark.asyncio
-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.")
-
+async def test_tracer_integration_agentops(
+    agent_function: Tuple[AgentName, Callable[[OpenAISettings, Tracer], Awaitable[Any]]],
+):
+    name, func = agent_function
     async with MockOpenAICompatibleServer() as settings:
         tracer = AgentOpsTracer()
-        await _run_tracer_with_agent(settings, tracer, agent_name)
+        await _run_tracer_with_agent(settings, tracer, name, func)
 
 
 @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.")
-
+async def test_tracer_integration_weave(
+    agent_function: Tuple[AgentName, Callable[[OpenAISettings, Tracer], Awaitable[Any]]],
+    monkeypatch: pytest.MonkeyPatch,
+):
     from agentlightning.tracer.weave import WeaveTracer
 
+    name, func = agent_function
+    skip_assert = "autogen" in name
+
+    if name == "openai_agents_sdk_handoff_tool_output_type_and_reward":
+        monkeypatch.setitem(AGENTOPS_EXPECTED_TRIPLETS_NUMBER, name, 6)
+        monkeypatch.setitem(AGENTOPS_EXPECTED_REWARDS, name, [None, None, None, 1.0, None, None])
+
     async with MockOpenAICompatibleServer() as settings:
         tracer = WeaveTracer()
-        await _run_tracer_with_agent(settings, tracer, agent_name, _skip_assert=True)
+        await _run_tracer_with_agent(settings, tracer, name, func, skip_assert)
 
 
 async def _run_tracer_with_agent(
-    settings: OpenAISettings, tracer: Tracer, agent_name: AgentName, _skip_assert: bool = False
+    settings: OpenAISettings,
+    tracer: Tracer,
+    agent_name: AgentName,
+    agent_func: Callable[[OpenAISettings, Tracer], Awaitable[Any]],
+    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)
@@ -715,6 +750,8 @@ async def _run_tracer_with_agent(
             Span.from_opentelemetry(span, "dummy", "dummy", 0) if isinstance(span, ReadableSpan) else span
             for span in tracer.get_last_trace()
         ]
+        assert len(last_trace_normalized) > 0
+
         for span in last_trace_normalized:
             print(">>> rollout_id =", span.rollout_id)
             print("... attempt_id =", span.attempt_id)
@@ -724,39 +761,45 @@ async def _run_tracer_with_agent(
             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(),
-            )
+            print("... attributes =", span.attributes.keys())
+
+        debug_dir = os.path.join(os.path.dirname(__file__), "debug", tracer.__class__.__name__)
+        os.makedirs(debug_dir, exist_ok=True)
+        with open(os.path.join(debug_dir, f"{agent_name}_raw.json"), "w") as f:
+            json.dump([span.model_dump() for span in last_trace_normalized], f, indent=2)
+
         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}"))
+            tree.visualize(filename=os.path.join(debug_dir, agent_name))
         else:
             warnings.warn("dot is not installed. Skipping trace tree visualization.")
 
         tree.repair_hierarchy()
+        triplets = TracerTraceToTriplet().adapt(last_trace_normalized)
 
-        if _skip_assert:
+        with open(os.path.join(debug_dir, f"{agent_name}_triplets.json"), "w") as f:
+            json.dump([triplet.model_dump() for triplet in triplets], f, indent=2)
+
+        if skip_assert:
             return
 
         assert_expected_pairs_in_tree(tree.names_tuple(), AGENTOPS_EXPECTED_TREES[agent_name])
-
-        triplets = TracerTraceToTriplet().adapt(last_trace_normalized)
-        assert (
-            len(triplets) == AGENTOPS_EXPECTED_TRIPLETS_NUMBER[agent_name]
-        ), f"Expected {AGENTOPS_EXPECTED_TRIPLETS_NUMBER[agent_name]} triplets, but got: {triplets}"
+        if len(triplets) != AGENTOPS_EXPECTED_TRIPLETS_NUMBER[agent_name]:
+            triplet_assert_err_msg = f"Expected {AGENTOPS_EXPECTED_TRIPLETS_NUMBER[agent_name]} triplets, but got:\n{pprint.pformat(triplets)}"
+            print(triplet_assert_err_msg)
+            raise AssertionError(triplet_assert_err_msg)
         if agent_name in AGENTOPS_EXPECTED_REWARDS:
             expected_reward = AGENTOPS_EXPECTED_REWARDS[agent_name]
             if isinstance(expected_reward, tuple):
                 # If the expected rewards are a tuple, make sure at least one of them matches
-                assert any([r.reward in expected for r in triplets for expected in expected_reward]), (
-                    f"Expected rewards {expected_reward}, " f"but got: {pprint.pformat(triplets)}"
-                )
+                if not any([r.reward in expected for r in triplets for expected in expected_reward]):
+                    err_msg = f"Expected rewards {expected_reward}, but got: {pprint.pformat(triplets)}"
+                    print(err_msg)
+                    raise AssertionError(err_msg)
             else:
-                assert [r.reward for r in triplets] == expected_reward, (
-                    f"Expected rewards {expected_reward}, " f"but got: {pprint.pformat(triplets)}"
-                )
+                if [r.reward for r in triplets] != expected_reward:
+                    err_msg = f"Expected rewards {expected_reward}, but got: {pprint.pformat(triplets)}"
+                    print(err_msg)
+                    raise AssertionError(err_msg)

tests/utils/test_otel.py

@@ -5,7 +5,7 @@ from __future__ import annotations
 import json
 import logging
 from types import SimpleNamespace
-from typing import Any, Dict, List
+from typing import Any, Dict, List, cast
 
 import opentelemetry.trace as trace_api
 import pytest
@@ -371,12 +371,21 @@ def test_sanitize_attribute_value_falls_back_to_json_for_hybrid_lists(caplog: py
     assert "Failed to sanitize list attribute" in caplog.text
 
 
-def test_sanitize_attribute_value_rejects_non_serializable_objects() -> None:
+def test_sanitize_attribute_value_serializes_non_serializable_objects_when_forced() -> None:
+    class Unserializable:
+        def __str__(self) -> str:
+            return "forced-serialization"
+
+    result = sanitize_attribute_value(Unserializable())
+    assert json.loads(cast(str, result)) == "forced-serialization"
+
+
+def test_sanitize_attribute_value_respects_force_flag() -> None:
     class Unserializable:
         pass
 
     with pytest.raises(ValueError, match="Object must be JSON serializable"):
-        sanitize_attribute_value(Unserializable())
+        sanitize_attribute_value(Unserializable(), force=False)
 
 
 def test_sanitize_attributes_returns_clean_values() -> None:
@@ -394,13 +403,34 @@ def test_sanitize_attributes_returns_clean_values() -> None:
     assert result["flags"] == [True, False]
 
 
+def test_sanitize_attributes_serializes_non_serializable_values_by_default() -> None:
+    class CustomValue:
+        def __str__(self) -> str:
+            return "custom-payload"
+
+    attributes = {"payload": CustomValue()}
+
+    result = sanitize_attributes(attributes)
+    assert json.loads(cast(str, result["payload"])) == "custom-payload"
+
+
+def test_sanitize_attributes_respects_force_flag() -> None:
+    class CustomValue:
+        pass
+
+    attributes = {"payload": CustomValue()}
+
+    with pytest.raises(ValueError, match="Failed to sanitize attribute 'payload'"):
+        sanitize_attributes(attributes, force=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)
+        sanitize_attributes(attributes, force=False)
 
 
 def test_format_exception_attributes_captures_metadata() -> None: