Fix safety event emission + add cassette tests for LangChain, LlamaIndex, LiteLLM

packages/opentelemetry-instrumentation-fortifyroot/opentelemetry/instrumentation/fortifyroot/__init__.py

@@ -8,6 +8,10 @@
     SafetyResult,
     clear_safety_handlers,
     clone_value,
+    discard_deferred_findings,
+    drain_deferred_findings,
+    emit_deferred_findings,
+    inject_deferred_findings,
     get_completion_safety_handler,
     get_object_value,
     get_prompt_safety_handler,
@@ -61,5 +65,9 @@
     "run_prompt_safety",
     "run_prompt_safety_async",
     "SafetyStreamContext",
+    "discard_deferred_findings",
+    "drain_deferred_findings",
+    "emit_deferred_findings",
+    "inject_deferred_findings",
     "set_object_value",
 ]

packages/opentelemetry-instrumentation-fortifyroot/opentelemetry/instrumentation/fortifyroot/safety.py

@@ -327,14 +327,101 @@ def _normalize_decision(value: str) -> str:
     return SafetyDecision.ALLOW.value
 
 
+# ---------------------------------------------------------------------------
+# FR: Deferred finding buffer — thread-local queue for safety findings that
+# are produced when no valid OTel span is current (e.g. LangChain/LlamaIndex
+# safety pre-wrappers run before the callback handler creates the span).
+#
+# Thread-local so concurrent requests on different threads don't interfere.
+#
+# Four operations on the buffer:
+#
+#   discard   — throw away all findings (data lost). Use at wrapper entry
+#               to prevent stale findings from a prior request leaking.
+#
+#   emit      — write all findings as span events on a given span, then
+#               clear the buffer. Use after the real span is created.
+#
+#   drain     — remove and return all findings from THIS thread's buffer.
+#               Use on a worker thread (asyncio.to_thread) to extract
+#               findings before returning to the calling thread.
+#
+#   inject    — append findings INTO this thread's buffer. Use on the
+#               calling thread to re-insert findings drained from a worker.
+#
+# Async pattern (drain + inject):
+#   Worker threads created by asyncio.to_thread get their own thread-local
+#   buffer. Callers must drain on the worker and inject on the event-loop
+#   thread so that the callback handler's emit() sees the findings.
+# ---------------------------------------------------------------------------
+
+_deferred = threading.local()
+
+
+def discard_deferred_findings() -> None:
+    """Throw away all buffered findings on this thread — data is lost.
+
+    Call at the START of a new safety wrapper invocation to prevent stale
+    findings from a prior request on the same thread from leaking into the
+    current request's span.
+    """
+    _deferred.items = []
+
+
+def emit_deferred_findings(span: Span | None) -> None:
+    """Write all buffered findings as span events on *span*, then clear.
+
+    Each finding becomes a ``fortifyroot.safety.violation`` span event.
+    Safe to call when the buffer is empty or *span* is None/ended (no-op).
+    """
+    items: list[dict[str, Any]] = getattr(_deferred, "items", [])
+    _deferred.items = []
+    if not span or not span.is_recording() or not items:
+        return
+    for attrs in items:
+        span.add_event(SAFETY_EVENT_NAME, attributes=attrs)
+
+
+def drain_deferred_findings() -> list[dict[str, Any]]:
+    """Remove and return all findings from this thread's buffer.
+
+    Use on a worker thread after safety runs, then pass the returned list
+    to ``inject_deferred_findings()`` on the calling thread::
+
+        def _on_worker():
+            apply_safety(...)
+            return drain_deferred_findings()
+
+        findings = await asyncio.to_thread(_on_worker)
+        inject_deferred_findings(findings)
+    """
+    items: list[dict[str, Any]] = getattr(_deferred, "items", [])
+    _deferred.items = []
+    return items
+
+
+def inject_deferred_findings(items: list[dict[str, Any]]) -> None:
+    """Append findings (from another thread) into this thread's buffer.
+
+    Counterpart to ``drain_deferred_findings()``.  No-op if *items* is
+    empty.
+    """
+    if not items:
+        return
+    if not hasattr(_deferred, "items"):
+        _deferred.items = []
+    _deferred.items.extend(items)
+
+
 def _emit_findings(
     span: Span | None,
     context: SafetyContext,
     result: SafetyResult,
 ) -> None:
-    if span is None or not span.is_recording():
+    if not result.findings:
         return
 
+    attrs_list = []
     for finding in result.findings:
         attributes: dict[str, Any] = {
             "fortifyroot.safety.category": finding.category,
@@ -349,4 +436,15 @@ def _emit_findings(
             attributes["fortifyroot.safety.segment_index"] = context.segment_index
         if context.segment_role:
             attributes["fortifyroot.safety.segment_role"] = context.segment_role
-        span.add_event(SAFETY_EVENT_NAME, attributes=attributes)
+        attrs_list.append(attributes)
+
+    # If the span is valid and recording, emit immediately
+    if span is not None and span.is_recording():
+        for attributes in attrs_list:
+            span.add_event(SAFETY_EVENT_NAME, attributes=attributes)
+        return
+
+    # Otherwise buffer for deferred emission
+    if not hasattr(_deferred, "items"):
+        _deferred.items = []
+    _deferred.items.extend(attrs_list)

packages/opentelemetry-instrumentation-langchain/opentelemetry/instrumentation/langchain/callback_handler.py

@@ -496,6 +496,9 @@ def on_chat_model_start(
             serialized=serialized,
         )
         set_request_params(span, kwargs, self.spans[run_id])
+        # FR: emit deferred prompt safety findings now that the span exists
+        from opentelemetry.instrumentation.fortifyroot import emit_deferred_findings
+        emit_deferred_findings(span)
         if should_emit_events():
             self._emit_chat_input_events(messages)
         else:
@@ -526,6 +529,9 @@ def on_llm_start(
             serialized=serialized,
         )
         set_request_params(span, kwargs, self.spans[run_id])
+        # FR: emit deferred prompt safety findings now that the span exists
+        from opentelemetry.instrumentation.fortifyroot import emit_deferred_findings
+        emit_deferred_findings(span)
         if should_emit_events():
             for prompt in prompts:
                 emit_event(MessageEvent(content=prompt, role="user"))
@@ -642,6 +648,10 @@ def on_llm_end(
             },
         )
 
+        # FR: emit deferred completion safety findings before span ends
+        from opentelemetry.instrumentation.fortifyroot import emit_deferred_findings
+        emit_deferred_findings(span)
+
         self._end_span(span, run_id)
 
     @dont_throw

packages/opentelemetry-instrumentation-langchain/opentelemetry/instrumentation/langchain/safety.py

@@ -26,26 +26,46 @@
 
 
 def base_chat_model_generate_wrapper(wrapped, instance, args, kwargs):
+    from opentelemetry.instrumentation.fortifyroot import discard_deferred_findings
+    discard_deferred_findings()  # FR: prevent stale findings from prior request on same thread
     updated_args, updated_kwargs = _apply_chat_prompt_safety(instance, args, kwargs)
     return wrapped(*updated_args, **updated_kwargs)
 
 
 async def base_chat_model_agenerate_wrapper(wrapped, instance, args, kwargs):
-    updated_args, updated_kwargs = await asyncio.to_thread(  # FR: async safety
-        _apply_chat_prompt_safety, instance, args, kwargs
+    from opentelemetry.instrumentation.fortifyroot import (
+        discard_deferred_findings, drain_deferred_findings, inject_deferred_findings,
     )
+    discard_deferred_findings()
+
+    def _safety_on_worker():
+        result = _apply_chat_prompt_safety(instance, args, kwargs)
+        return result, drain_deferred_findings()
+
+    (updated_args, updated_kwargs), worker_findings = await asyncio.to_thread(_safety_on_worker)
+    inject_deferred_findings(worker_findings)
     return await wrapped(*updated_args, **updated_kwargs)
 
 
 def base_llm_generate_wrapper(wrapped, instance, args, kwargs):
+    from opentelemetry.instrumentation.fortifyroot import discard_deferred_findings
+    discard_deferred_findings()
     updated_args, updated_kwargs = _apply_llm_prompt_safety(instance, args, kwargs)
     return wrapped(*updated_args, **updated_kwargs)
 
 
 async def base_llm_agenerate_wrapper(wrapped, instance, args, kwargs):
-    updated_args, updated_kwargs = await asyncio.to_thread(  # FR: async safety
-        _apply_llm_prompt_safety, instance, args, kwargs
+    from opentelemetry.instrumentation.fortifyroot import (
+        discard_deferred_findings, drain_deferred_findings, inject_deferred_findings,
     )
+    discard_deferred_findings()
+
+    def _safety_on_worker():
+        result = _apply_llm_prompt_safety(instance, args, kwargs)
+        return result, drain_deferred_findings()
+
+    (updated_args, updated_kwargs), worker_findings = await asyncio.to_thread(_safety_on_worker)
+    inject_deferred_findings(worker_findings)
     return await wrapped(*updated_args, **updated_kwargs)
 
 
@@ -58,8 +78,15 @@ def base_chat_model_generate_with_cache_wrapper(wrapped, instance, args, kwargs)
 async def base_chat_model_agenerate_with_cache_wrapper(
     wrapped, instance, args, kwargs
 ):
+    from opentelemetry.instrumentation.fortifyroot import drain_deferred_findings, inject_deferred_findings
     response = await wrapped(*args, **kwargs)
-    await asyncio.to_thread(_apply_chat_result_completion_safety, instance, response)  # FR: async safety
+
+    def _completion_safety_on_worker():
+        _apply_chat_result_completion_safety(instance, response)
+        return drain_deferred_findings()
+
+    worker_findings = await asyncio.to_thread(_completion_safety_on_worker)  # FR: async safety
+    inject_deferred_findings(worker_findings)
     return response
 
 
@@ -70,8 +97,15 @@ def base_llm_generate_helper_wrapper(wrapped, instance, args, kwargs):
 
 
 async def base_llm_agenerate_helper_wrapper(wrapped, instance, args, kwargs):
+    from opentelemetry.instrumentation.fortifyroot import drain_deferred_findings, inject_deferred_findings
     response = await wrapped(*args, **kwargs)
-    await asyncio.to_thread(_apply_llm_result_completion_safety, instance, response)  # FR: async safety
+
+    def _completion_safety_on_worker():
+        _apply_llm_result_completion_safety(instance, response)
+        return drain_deferred_findings()
+
+    worker_findings = await asyncio.to_thread(_completion_safety_on_worker)  # FR: async safety
+    inject_deferred_findings(worker_findings)
     return response