diff --git a/.release-intents/20260816-python-beeai-braintrust-burr-tracing.json b/.release-intents/20260816-python-beeai-braintrust-burr-tracing.json new file mode 100644 index 00000000..21cde5d9 --- /dev/null +++ b/.release-intents/20260816-python-beeai-braintrust-burr-tracing.json @@ -0,0 +1,8 @@ +{ + "summary": "Repair BeeAI failure spans, Braintrust evaluation metadata, and Burr root errors", + "packages": { + "respan-instrumentation-beeai": "patch", + "respan-instrumentation-braintrust": "patch", + "respan-instrumentation-burr": "patch" + } +} diff --git a/python-sdks/instrumentations/respan-instrumentation-beeai/src/respan_instrumentation_beeai/_instrumentation.py b/python-sdks/instrumentations/respan-instrumentation-beeai/src/respan_instrumentation_beeai/_instrumentation.py index aa807f29..063897ba 100644 --- a/python-sdks/instrumentations/respan-instrumentation-beeai/src/respan_instrumentation_beeai/_instrumentation.py +++ b/python-sdks/instrumentations/respan-instrumentation-beeai/src/respan_instrumentation_beeai/_instrumentation.py @@ -3,11 +3,15 @@ import importlib import json import logging +from threading import Lock from typing import Any +from openinference.semconv.trace import OpenInferenceSpanKindValues +from openinference.semconv.trace import SpanAttributes as OISpanAttributes from opentelemetry import trace from opentelemetry.sdk.trace import ReadableSpan, SpanProcessor from opentelemetry.semconv_ai import SpanAttributes as TLSpanAttributes +from opentelemetry.trace import Status, StatusCode from respan_instrumentation_openinference import OpenInferenceInstrumentor from respan_sdk.constants.span_attributes import ( RESPAN_SPAN_TOOL_CALLS, @@ -19,6 +23,10 @@ BEEAI_INSTRUMENTATION_NAME = "beeai" OPENINFERENCE_BEEAI_MODULE = "openinference.instrumentation.beeai" +OPENINFERENCE_BEEAI_PROCESSOR_MODULE = ( + "openinference.instrumentation.beeai.processors.base" +) +OPENINFERENCE_BEEAI_SPAN_MODULE = "openinference.instrumentation.beeai._span" _OFF_CONTRACT_ALIAS_KEYS = ( RESPAN_SPAN_TOOLS, RESPAN_SPAN_TOOL_CALLS, @@ -35,6 +43,178 @@ ) _TOOL_CALLS_SUFFIX = ".tool_calls" +_BEEAI_PATCH_LOCK = Lock() +_BEEAI_PATCH_REFCOUNT = 0 +_BEEAI_PATCHES: list[tuple[type, str, Any, Any]] = [] + + +def _exception_chain(error: BaseException) -> list[BaseException]: + chain: list[BaseException] = [] + seen: set[int] = set() + current: BaseException | None = error + while current is not None and id(current) not in seen: + seen.add(id(current)) + chain.append(current) + next_error = getattr(current, "__cause__", None) + if not isinstance(next_error, BaseException): + next_error = getattr(current, "__context__", None) + if not isinstance(next_error, BaseException): + next_error = getattr(current, "_predecessor", None) + current = next_error if isinstance(next_error, BaseException) else None + return chain + + +def _exception_status_code(error: BaseException) -> int: + for item in _exception_chain(error): + for value in ( + getattr(item, "status_code", None), + getattr(getattr(item, "response", None), "status_code", None), + getattr(getattr(item, "response", None), "status", None), + ): + if isinstance(value, int) and value >= 400: + return value + return 500 + + +def _exception_message(error: BaseException) -> str: + explain = getattr(error, "explain", None) + if callable(explain): + try: + message = explain() + except Exception: + message = None + if isinstance(message, str) and message: + return message + return str(error) or type(error).__name__ + + +def _event_error(value: Any) -> BaseException | None: + if isinstance(value, BaseException): + return value + error = getattr(value, "error", None) + return error if isinstance(error, BaseException) else None + + +def _record_active_parent_exception(error: BaseException) -> None: + active_span = trace.get_current_span() + is_recording = getattr(active_span, "is_recording", None) + if not callable(is_recording) or not is_recording(): + return + + message = _exception_message(error) + current_status = getattr(getattr(active_span, "status", None), "status_code", None) + if current_status != StatusCode.ERROR: + active_span.record_exception(error) + active_span.set_status(Status(StatusCode.ERROR, message)) + active_span.set_attribute("status_code", _exception_status_code(error)) + active_span.set_attribute("error.message", message) + + +def _record_exception_wrapper(original: Any) -> Any: + def record_exception(span: Any, error: BaseException) -> None: + original(span, error) + _record_active_parent_exception(error) + attrs = getattr(span, "attributes", None) + if not isinstance(attrs, dict): + return + # Error text is diagnostic data, not an assistant completion. Keeping + # it in output causes platform token/cost estimation for failed calls. + attrs.pop(OISpanAttributes.OUTPUT_VALUE, None) + attrs.pop(OISpanAttributes.OUTPUT_MIME_TYPE, None) + attrs["status_code"] = _exception_status_code(error) + attrs["error.message"] = _exception_message(error) + + return record_exception + + +def _child_wrapper(original: Any) -> Any: + def child( + span: Any, + name: str | None = None, + event: tuple[Any, Any] | None = None, + ) -> Any: + error = _event_error(event[0]) if event is not None else None + if error is None: + return original(span, name=name, event=event) + + meta = event[1] + span.add_event( + name or getattr(meta, "name", None) or "error", + {"error.message": _exception_message(error)}, + getattr(meta, "created_at", None), + ) + span.record_exception(error) + # Error events describe the owning operation; they are not another + # model invocation. Returning the owner also supports upstream callers + # that immediately call record_exception() on the child result. + return span + + return child + + +def _end_wrapper(original: Any) -> Any: + async def end(processor: Any, event: Any, meta: Any) -> None: + await original(processor, event, meta) + output = getattr(event, "output", None) + get_text_content = getattr(output, "get_text_content", None) + if ( + getattr(processor.span, "kind", None) == OpenInferenceSpanKindValues.LLM + and callable(get_text_content) + ): + content = get_text_content() + if content: + processor.span.set_attribute( + f"{OISpanAttributes.LLM_OUTPUT_MESSAGES}.0.message.content", + content, + ) + error = _event_error(event) + if error is not None: + # Upstream currently resets ERROR to OK when output is also set. + processor.span.record_exception(error) + + return end + + +def _patch_beeai_processors() -> None: + global _BEEAI_PATCH_REFCOUNT + + with _BEEAI_PATCH_LOCK: + if _BEEAI_PATCH_REFCOUNT == 0: + processor_module = importlib.import_module( + OPENINFERENCE_BEEAI_PROCESSOR_MODULE + ) + span_module = importlib.import_module(OPENINFERENCE_BEEAI_SPAN_MODULE) + patch_specs = ( + ( + span_module.SpanWrapper, + "record_exception", + _record_exception_wrapper, + ), + (span_module.SpanWrapper, "child", _child_wrapper), + (processor_module.Processor, "end", _end_wrapper), + ) + for owner, attribute, wrapper_factory in patch_specs: + original = getattr(owner, attribute) + patched = wrapper_factory(original) + setattr(owner, attribute, patched) + _BEEAI_PATCHES.append((owner, attribute, original, patched)) + _BEEAI_PATCH_REFCOUNT += 1 + + +def _unpatch_beeai_processors() -> None: + global _BEEAI_PATCH_REFCOUNT + + with _BEEAI_PATCH_LOCK: + if _BEEAI_PATCH_REFCOUNT == 0: + return + _BEEAI_PATCH_REFCOUNT -= 1 + if _BEEAI_PATCH_REFCOUNT != 0: + return + for owner, attribute, original, patched in reversed(_BEEAI_PATCHES): + if getattr(owner, attribute) is patched: + setattr(owner, attribute, original) + _BEEAI_PATCHES.clear() + def _load_openinference_beeai_class() -> type: beeai_module = importlib.import_module(OPENINFERENCE_BEEAI_MODULE) @@ -113,6 +293,7 @@ def __init__(self, **instrumentor_kwargs: Any) -> None: self._instrumentor_kwargs = instrumentor_kwargs self._delegate = None self._cleanup_processor = None + self._processors_patched = False self._is_instrumented = False @staticmethod @@ -143,6 +324,8 @@ def activate(self) -> None: return try: + _patch_beeai_processors() + self._processors_patched = True self._delegate = OpenInferenceInstrumentor( beeai_instrumentor_class, **self._instrumentor_kwargs, @@ -159,6 +342,9 @@ def activate(self) -> None: logger.exception("Failed to clean up BeeAI instrumentation") self._delegate = None self._cleanup_processor = None + if self._processors_patched: + _unpatch_beeai_processors() + self._processors_patched = False self._is_instrumented = False logger.exception("Failed to activate BeeAI instrumentation") @@ -209,6 +395,9 @@ def deactivate(self) -> None: self._delegate.deactivate() except Exception: logger.exception("Failed to deactivate BeeAI instrumentation") + if self._processors_patched: + _unpatch_beeai_processors() + self._processors_patched = False self._delegate = None self._is_instrumented = False logger.info("BeeAI instrumentation deactivated") diff --git a/python-sdks/instrumentations/respan-instrumentation-beeai/tests/test_instrumentation.py b/python-sdks/instrumentations/respan-instrumentation-beeai/tests/test_instrumentation.py index 9d663201..1347860e 100644 --- a/python-sdks/instrumentations/respan-instrumentation-beeai/tests/test_instrumentation.py +++ b/python-sdks/instrumentations/respan-instrumentation-beeai/tests/test_instrumentation.py @@ -1,6 +1,7 @@ import json import logging import sys +from datetime import datetime, timezone from types import ModuleType, SimpleNamespace import pytest @@ -58,6 +59,8 @@ def deactivate(self): "OpenInferenceInstrumentor", FakeOpenInferenceInstrumentor, ) + monkeypatch.setattr(_instrumentation, "_patch_beeai_processors", lambda: None) + monkeypatch.setattr(_instrumentation, "_unpatch_beeai_processors", lambda: None) return SimpleNamespace( beeai_instrumentor_class=FakeBeeAIInstrumentor, @@ -257,3 +260,138 @@ def _get_translator(cls): instrumentor.deactivate() assert active_span_processor._span_processors == (translator, exporter) + + +def test_beeai_error_patch_preserves_status_and_drops_duplicate_child() -> None: + from openinference.instrumentation.beeai._span import SpanWrapper + from openinference.semconv.trace import OpenInferenceSpanKindValues, SpanAttributes + from opentelemetry.trace import StatusCode + + class ProviderError(RuntimeError): + status_code = 404 + + provider_error = ProviderError("requested model was not found") + error = RuntimeError("Chat Model error") + error.__cause__ = provider_error + span = SpanWrapper(name="ChatModel", kind=OpenInferenceSpanKindValues.LLM) + meta = SimpleNamespace(name="error", created_at=datetime.now(timezone.utc)) + + _instrumentation._patch_beeai_processors() + try: + child = span.child( + "error", + event=(SimpleNamespace(error=error), meta), + ) + finally: + _instrumentation._unpatch_beeai_processors() + + assert child is span + assert span.children == [] + assert span.status == StatusCode.ERROR + assert span.attributes["status_code"] == 404 + assert span.attributes["error.message"] == "Chat Model error" + assert SpanAttributes.OUTPUT_VALUE not in span.attributes + assert [event.name for event in span.events] == ["error"] + + +def test_beeai_finish_error_cannot_be_reset_to_success() -> None: + import asyncio + + from openinference.instrumentation.beeai._span import SpanWrapper + from openinference.instrumentation.beeai.processors.base import Processor + from openinference.semconv.trace import OpenInferenceSpanKindValues, SpanAttributes + from opentelemetry.trace import StatusCode + + error = RuntimeError("deterministic failure") + processor = object.__new__(Processor) + processor.span = SpanWrapper( + name="ChatModel", + kind=OpenInferenceSpanKindValues.LLM, + ) + event = SimpleNamespace(error=error, output={"partial": "ignored"}) + meta = SimpleNamespace(created_at=datetime.now(timezone.utc)) + + _instrumentation._patch_beeai_processors() + try: + asyncio.run(processor.end(event, meta)) + finally: + _instrumentation._unpatch_beeai_processors() + + assert processor.span.status == StatusCode.ERROR + assert processor.span.attributes["status_code"] == 500 + assert processor.span.attributes["error.message"] == "deterministic failure" + assert SpanAttributes.OUTPUT_VALUE not in processor.span.attributes + + +def test_beeai_error_marks_active_parent_span(monkeypatch) -> None: + from openinference.instrumentation.beeai._span import SpanWrapper + from openinference.semconv.trace import OpenInferenceSpanKindValues + from opentelemetry.trace import StatusCode + + class ActiveSpan: + def __init__(self) -> None: + self.attributes = {} + self.exceptions = [] + self.status = SimpleNamespace(status_code=StatusCode.UNSET) + + def is_recording(self) -> bool: + return True + + def record_exception(self, error) -> None: + self.exceptions.append(error) + + def set_status(self, status) -> None: + self.status = status + + def set_attribute(self, key, value) -> None: + self.attributes[key] = value + + class ProviderError(RuntimeError): + status_code = 404 + + active_span = ActiveSpan() + monkeypatch.setattr(_instrumentation.trace, "get_current_span", lambda: active_span) + wrapped_span = SpanWrapper(name="ChatModel", kind=OpenInferenceSpanKindValues.LLM) + error = RuntimeError("Chat Model error") + error.__cause__ = ProviderError("requested model was not found") + + _instrumentation._patch_beeai_processors() + try: + wrapped_span.record_exception(error) + finally: + _instrumentation._unpatch_beeai_processors() + + assert active_span.status.status_code == StatusCode.ERROR + assert active_span.attributes["status_code"] == 404 + assert active_span.attributes["error.message"] == "Chat Model error" + assert active_span.exceptions == [error] + + +def test_beeai_finish_preserves_direct_chat_text_content() -> None: + import asyncio + + from openinference.instrumentation.beeai._span import SpanWrapper + from openinference.instrumentation.beeai.processors.base import Processor + from openinference.semconv.trace import OpenInferenceSpanKindValues, SpanAttributes + + processor = object.__new__(Processor) + processor.span = SpanWrapper( + name="ChatModel", + kind=OpenInferenceSpanKindValues.LLM, + ) + output = SimpleNamespace(get_text_content=lambda: "A complete assistant answer.") + event = SimpleNamespace(error=None, output=output) + meta = SimpleNamespace(created_at=datetime.now(timezone.utc)) + + _instrumentation._patch_beeai_processors() + try: + asyncio.run(processor.end(event, meta)) + finally: + _instrumentation._unpatch_beeai_processors() + + assert ( + processor.span.attributes[ + f"{SpanAttributes.LLM_OUTPUT_MESSAGES}.0.message.content" + ] + == "A complete assistant answer." + ) diff --git a/python-sdks/instrumentations/respan-instrumentation-braintrust/src/respan_instrumentation_braintrust/_instrumentation.py b/python-sdks/instrumentations/respan-instrumentation-braintrust/src/respan_instrumentation_braintrust/_instrumentation.py index f5cd8aed..c6bb966b 100644 --- a/python-sdks/instrumentations/respan-instrumentation-braintrust/src/respan_instrumentation_braintrust/_instrumentation.py +++ b/python-sdks/instrumentations/respan-instrumentation-braintrust/src/respan_instrumentation_braintrust/_instrumentation.py @@ -361,6 +361,62 @@ def _build_metadata(record: Mapping[str, Any]) -> dict[str, Any] | None: return _sanitize_json(metadata) +def _metadata_mapping(value: Any) -> dict[str, Any]: + if isinstance(value, Mapping): + return dict(value) + if not isinstance(value, str): + return {} + try: + parsed = json.loads(value) + except json.JSONDecodeError: + return {} + return dict(parsed) if isinstance(parsed, Mapping) else {} + + +def _merge_metadata( + braintrust_metadata: Mapping[str, Any] | None, + propagated_metadata: Any, +) -> dict[str, Any] | None: + """Merge propagated metadata without replacing Braintrust evaluation data. + + Propagated fields win generic-key collisions because they describe the + active Respan run. Braintrust-owned scores, metrics, tags, and identifiers + are namespaced under ``braintrust_*`` and therefore remain intact. + """ + + merged = dict(braintrust_metadata or {}) + merged.update(_metadata_mapping(propagated_metadata)) + return _sanitize_json(merged) if merged else None + + +def _flatten_metadata(metadata: Mapping[str, Any]) -> dict[str, Any]: + """Expose metadata through canonical per-key attributes used by OTLP.""" + + flattened: dict[str, Any] = {} + for key, value in metadata.items(): + attribute = f"{RESPAN_METADATA}.{key}" + if value is None or isinstance(value, str | int | float | bool): + flattened[attribute] = value + else: + flattened[attribute] = json.dumps(_sanitize_json(value), default=str) + return flattened + + +def _propagated_metadata(extra_attributes: Mapping[str, Any] | None) -> dict[str, Any]: + if not extra_attributes: + return {} + metadata = _metadata_mapping(extra_attributes.get(RESPAN_METADATA)) + prefix = f"{RESPAN_METADATA}." + metadata.update( + { + key[len(prefix) :]: value + for key, value in extra_attributes.items() + if key.startswith(prefix) + } + ) + return metadata + + def _record_mapping(item: Any) -> Mapping[str, Any] | None: if isinstance(item, Mapping): return item @@ -435,8 +491,10 @@ def _build_span_from_record( attrs[TLSpanAttributes.TRACELOOP_ENTITY_INPUT] = input_string if output_string is not None: attrs[TLSpanAttributes.TRACELOOP_ENTITY_OUTPUT] = output_string + metadata = _merge_metadata(metadata, _propagated_metadata(extra_attributes)) if metadata is not None: attrs[RESPAN_METADATA] = json.dumps(metadata, default=str) + attrs.update(_flatten_metadata(metadata)) if log_type == LOG_TYPE_CHAT: attrs[LLM_REQUEST_TYPE] = LLMRequestTypeValues.CHAT.value @@ -455,7 +513,13 @@ def _build_span_from_record( attrs[_LLM_USAGE_TOTAL_TOKENS] = total_tokens if extra_attributes: - attrs.update(extra_attributes) + attrs.update( + { + key: value + for key, value in extra_attributes.items() + if key != RESPAN_METADATA and not key.startswith(f"{RESPAN_METADATA}.") + } + ) if workflow_name is not None: attrs[TLSpanAttributes.TRACELOOP_WORKFLOW_NAME] = workflow_name diff --git a/python-sdks/instrumentations/respan-instrumentation-braintrust/tests/test_instrumentation.py b/python-sdks/instrumentations/respan-instrumentation-braintrust/tests/test_instrumentation.py index e31c63d0..97b7931c 100644 --- a/python-sdks/instrumentations/respan-instrumentation-braintrust/tests/test_instrumentation.py +++ b/python-sdks/instrumentations/respan-instrumentation-braintrust/tests/test_instrumentation.py @@ -144,3 +144,57 @@ def get(self): exported[0].attributes[TLSpanAttributes.TRACELOOP_WORKFLOW_NAME] == "Braintrust Basic Workflow" ) + + +def test_braintrust_and_propagated_metadata_are_merged() -> None: + record = _llm_record() + record["tags"] = ["evaluation", "release"] + record["scores"] = { + "overall": 0.94, + "completeness": 0.96, + "brevity": 0.9, + } + record["metrics"] = { + **record["metrics"], + "latency_ms": 1250, + } + + span = _build_span_from_record( + record, + extra_attributes={ + RESPAN_TRACE_GROUP_ID: "Braintrust Scored Evaluation Workflow", + f"{RESPAN_METADATA}.example_set": "braintrust", + f"{RESPAN_METADATA}.example_name": "03_scored_evaluation_workflow", + f"{RESPAN_METADATA}.workflow_name": "Braintrust Scored Evaluation Workflow", + f"{RESPAN_METADATA}.run_id": "otel2-fix-py-group-10", + f"{RESPAN_METADATA}.request_id": "propagated-request", + }, + ) + + metadata = json.loads(span.attributes[RESPAN_METADATA]) + assert metadata["example_set"] == "braintrust" + assert metadata["example_name"] == "03_scored_evaluation_workflow" + assert metadata["workflow_name"] == "Braintrust Scored Evaluation Workflow" + assert metadata["run_id"] == "otel2-fix-py-group-10" + assert metadata["request_id"] == "propagated-request" + assert metadata["braintrust_tags"] == ["evaluation", "release"] + assert metadata["braintrust_scores"] == { + "overall": 0.94, + "completeness": 0.96, + "brevity": 0.9, + } + assert metadata["braintrust_metrics"]["latency_ms"] == 1250 + assert metadata["braintrust_log_id"] == "00000000000000000000000000000099" + assert json.loads( + span.attributes[f"{RESPAN_METADATA}.braintrust_scores"] + ) == metadata["braintrust_scores"] + assert json.loads( + span.attributes[f"{RESPAN_METADATA}.braintrust_metrics"] + ) == metadata["braintrust_metrics"] + assert json.loads( + span.attributes[f"{RESPAN_METADATA}.braintrust_tags"] + ) == metadata["braintrust_tags"] + assert ( + span.attributes[f"{RESPAN_METADATA}.braintrust_log_id"] + == "00000000000000000000000000000099" + ) diff --git a/python-sdks/instrumentations/respan-instrumentation-burr/src/respan_instrumentation_burr/_adapter.py b/python-sdks/instrumentations/respan-instrumentation-burr/src/respan_instrumentation_burr/_adapter.py index 248fef5f..a592f7a4 100644 --- a/python-sdks/instrumentations/respan-instrumentation-burr/src/respan_instrumentation_burr/_adapter.py +++ b/python-sdks/instrumentations/respan-instrumentation-burr/src/respan_instrumentation_burr/_adapter.py @@ -39,6 +39,8 @@ from respan_sdk.utils.serialization import serialize_value logger = logging.getLogger(__name__) +_BURR_METADATA_ATTRIBUTE = f"{RESPAN_METADATA}.burr" +_MAX_CAPTURED_STREAM_ITEMS = 32 @dataclasses.dataclass @@ -47,6 +49,7 @@ class _ActiveSpan: span: Any token: Any metadata: dict[str, Any] + application_exception: BaseException | None = None _ACTIVE_SPANS: ContextVar[tuple[_ActiveSpan, ...]] = ContextVar( @@ -151,6 +154,7 @@ def _attributes( RESPAN_LOG_TYPE: log_type, RESPAN_TRACE_GROUP_ID: str(app_id), RESPAN_METADATA: _json_string({"burr": metadata}), + _BURR_METADATA_ATTRIBUTE: _json_string(metadata), SpanAttributes.TRACELOOP_ENTITY_NAME: entity_name, SpanAttributes.TRACELOOP_ENTITY_PATH: entity_name, } @@ -217,6 +221,43 @@ def _current(self) -> _ActiveSpan | None: stack = _ACTIVE_SPANS.get() return stack[-1] if stack else None + @staticmethod + def _sync_metadata(active: _ActiveSpan) -> None: + active.span.set_attribute( + RESPAN_METADATA, + _json_string({"burr": active.metadata}), + ) + active.span.set_attribute( + _BURR_METADATA_ATTRIBUTE, + _json_string(active.metadata), + ) + + def _remember_application_exception(self, exception: BaseException) -> None: + """Retain a failed child for Burr versions that drop the root error. + + Burr 0.42 invokes the application post hook with ``exception=None`` + after a failing action. Replace the nearest application entry in this + context-local immutable stack so nested or concurrent runs cannot + consume one another's failures. + """ + + stack = list(_ACTIVE_SPANS.get()) + for index in range(len(stack) - 1, -1, -1): + active = stack[index] + if active.scope == "application": + stack[index] = dataclasses.replace( + active, + application_exception=exception, + ) + _ACTIVE_SPANS.set(tuple(stack)) + return + + def _application_exception(self) -> BaseException | None: + for active in reversed(_ACTIVE_SPANS.get()): + if active.scope == "application": + return active.application_exception + return None + def pre_run_execute_call( self, *, @@ -259,6 +300,8 @@ def post_run_execute_call( **future_kwargs: Any, ) -> None: del future_kwargs + if exception is None: + exception = self._application_exception() output = ( {"status": "completed", "state": _jsonable(state)} if exception is None @@ -319,6 +362,12 @@ def post_run_step( **future_kwargs: Any, ) -> None: del future_kwargs + if exception is not None: + self._remember_application_exception(exception) + active = self._current() + if active is not None and "stream" in active.metadata: + active.metadata["stream"]["completed"] = exception is None + self._sync_metadata(active) output = { "status": "completed" if exception is None else "error", "result": _jsonable(result), @@ -392,10 +441,7 @@ def do_log_attributes( active.metadata["logged_attributes"] = _jsonable(attributes) if tags: active.metadata["tags"] = _jsonable(tags) - active.span.set_attribute( - RESPAN_METADATA, - _json_string({"burr": active.metadata}), - ) + self._sync_metadata(active) def pre_start_stream( self, @@ -410,6 +456,18 @@ def pre_start_stream( active = self._current() if not self.enabled or active is None: return + active.metadata["stream"] = { + "action": action, + "sequence_id": sequence_id, + "app_id": app_id, + "partition_key": partition_key, + "started": True, + "completed": False, + "item_count": 0, + } + if self.capture_content: + active.metadata["stream"]["items"] = [] + self._sync_metadata(active) active.span.add_event( "burr.stream.start", { @@ -433,6 +491,18 @@ def post_stream_item( active = self._current() if not self.enabled or active is None: return + stream = active.metadata.setdefault("stream", {}) + if stream.get("action") != action: + return + stream["item_count"] = int(stream.get("item_count", 0)) + 1 + items = stream.get("items") + if ( + self.capture_content + and isinstance(items, list) + and len(items) < _MAX_CAPTURED_STREAM_ITEMS + ): + items.append({"index": item_index, "value": _jsonable(item)}) + self._sync_metadata(active) attributes = { "burr.action": action, "burr.sequence_id": sequence_id, @@ -453,6 +523,17 @@ def post_end_stream( active = self._current() if not self.enabled or active is None: return + stream = active.metadata.setdefault("stream", {}) + if stream.get("action") != action: + return + stream.update( + { + "action": action, + "sequence_id": sequence_id, + "completed": True, + } + ) + self._sync_metadata(active) active.span.add_event( "burr.stream.end", { diff --git a/python-sdks/instrumentations/respan-instrumentation-burr/tests/test_adapter.py b/python-sdks/instrumentations/respan-instrumentation-burr/tests/test_adapter.py index 67b691e5..16bc2d07 100644 --- a/python-sdks/instrumentations/respan-instrumentation-burr/tests/test_adapter.py +++ b/python-sdks/instrumentations/respan-instrumentation-burr/tests/test_adapter.py @@ -1,6 +1,12 @@ +import json +from contextvars import copy_context from types import SimpleNamespace -from respan_instrumentation_burr._adapter import BurrLifecycleAdapter, _ACTIVE_SPANS +from respan_instrumentation_burr._adapter import ( + _ACTIVE_SPANS, + _BURR_METADATA_ATTRIBUTE, + BurrLifecycleAdapter, +) from respan_sdk.constants.span_attributes import ( RESPAN_LOG_TYPE, RESPAN_THREADS_ID, @@ -135,6 +141,136 @@ def test_action_failure_uses_backend_error_contract() -> None: assert span.exceptions == [error] +def test_action_failure_fails_application_when_burr_drops_root_exception() -> None: + _ACTIVE_SPANS.set(()) + tracer = FakeTracer() + adapter = BurrLifecycleAdapter(tracer=tracer) + method = SimpleNamespace(value="run") + state = FakeState({"count": 0}) + + adapter.pre_run_execute_call( + app_id="app-failure", + partition_key="customer-42", + state=state, + method=method, + ) + adapter.pre_run_step( + app_id="app-failure", + partition_key="customer-42", + sequence_id=1, + state=state, + action=_action(), + inputs={}, + ) + error = RuntimeError("deterministic Burr failure") + adapter.post_run_step( + state=state, + result=None, + exception=error, + ) + adapter.post_run_execute_call(state=state, exception=None) + + workflow, task = tracer.spans + assert workflow.attributes["status_code"] == 500 + assert workflow.attributes["error.message"] == "deterministic Burr failure" + assert '"status": "error"' in workflow.attributes[ + SpanAttributes.TRACELOOP_ENTITY_OUTPUT + ] + assert task.attributes["status_code"] == 500 + assert workflow.exceptions == [error] + assert _ACTIVE_SPANS.get() == () + + +def test_nested_application_failure_does_not_leak_to_outer_application() -> None: + _ACTIVE_SPANS.set(()) + tracer = FakeTracer() + adapter = BurrLifecycleAdapter(tracer=tracer) + method = SimpleNamespace(value="run") + state = FakeState({"count": 0}) + + adapter.pre_run_execute_call( + app_id="outer", + partition_key="outer-thread", + state=state, + method=method, + ) + adapter.pre_run_execute_call( + app_id="inner", + partition_key="inner-thread", + state=state, + method=method, + ) + adapter.pre_run_step( + app_id="inner", + partition_key="inner-thread", + sequence_id=1, + state=state, + action=_action(), + inputs={}, + ) + error = RuntimeError("inner failure") + adapter.post_run_step(state=state, result=None, exception=error) + adapter.post_run_execute_call(state=state, exception=None) + adapter.post_run_execute_call(state=state, exception=None) + + outer, inner, task = tracer.spans + assert inner.attributes["status_code"] == 500 + assert task.attributes["status_code"] == 500 + assert outer.attributes["status_code"] == 200 + assert _ACTIVE_SPANS.get() == () + + +def test_concurrent_contexts_keep_application_failures_isolated() -> None: + _ACTIVE_SPANS.set(()) + tracer = FakeTracer() + adapter = BurrLifecycleAdapter(tracer=tracer) + method = SimpleNamespace(value="arun") + state = FakeState({"count": 0}) + failed_context = copy_context() + successful_context = copy_context() + + def start(app_id: str) -> None: + adapter.pre_run_execute_call( + app_id=app_id, + partition_key=app_id, + state=state, + method=method, + ) + + failed_context.run(start, "failed-app") + successful_context.run(start, "successful-app") + + def fail() -> None: + adapter.pre_run_step( + app_id="failed-app", + partition_key="failed-app", + sequence_id=1, + state=state, + action=_action(), + inputs={}, + ) + adapter.post_run_step( + state=state, + result=None, + exception=RuntimeError("context failure"), + ) + adapter.post_run_execute_call(state=state, exception=None) + + failed_context.run(fail) + successful_context.run( + adapter.post_run_execute_call, + state=state, + exception=None, + ) + + failed_workflow, successful_workflow, failed_task = tracer.spans + assert failed_workflow.attributes["status_code"] == 500 + assert failed_task.attributes["status_code"] == 500 + assert successful_workflow.attributes["status_code"] == 200 + assert failed_context.run(_ACTIVE_SPANS.get) == () + assert successful_context.run(_ACTIVE_SPANS.get) == () + + def test_custom_span_attributes_and_stream_events_follow_burr_hooks() -> None: _ACTIVE_SPANS.set(()) tracer = FakeTracer() @@ -173,3 +309,11 @@ def test_custom_span_attributes_and_stream_events_follow_burr_hooks() -> None: "burr.stream.end", ] assert '"logged_attributes": {"documents": 2}' in span.attributes["respan.metadata"] + assert json.loads(span.attributes[_BURR_METADATA_ATTRIBUTE])[ + "logged_attributes" + ] == {"documents": 2} + stream = json.loads(span.attributes[_BURR_METADATA_ATTRIBUTE])["stream"] + assert stream["started"] is True + assert stream["completed"] is True + assert stream["item_count"] == 1 + assert stream["items"] == [{"index": 0, "value": {"text": "hello"}}]