From bd7da9df521ccd497fa6c20a2e78d19ac274b58f Mon Sep 17 00:00:00 2001
From: NagaSatish <satish.1290@gmail.com>
Date: Thu, 16 Jul 2026 11:57:24 +0530
Subject: [PATCH 1/2] feat: Add Langfuse governance trace exporter
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Adds tealtiger.integrations.langfuse module that exports governance
decisions as Langfuse spans with appropriate levels (ERROR/WARNING/DEFAULT).

- LangfuseGovernanceExporter class with trace() callback
- Maps GovernanceDecision → Langfuse span with metadata
- DENY → ERROR level, MONITOR → WARNING, ALLOW → DEFAULT
- Includes tests and example
- Session-based trace reuse (same session = same parent trace)

Closes #321
---
 examples/langfuse_governance_traces.py | 116 +++++++++++++++
 src/tealtiger/integrations/__init__.py |   5 +
 src/tealtiger/integrations/langfuse.py | 189 +++++++++++++++++++++++++
 tests/test_langfuse_integration.py     | 138 ++++++++++++++++++
 4 files changed, 448 insertions(+)
 create mode 100644 examples/langfuse_governance_traces.py
 create mode 100644 src/tealtiger/integrations/__init__.py
 create mode 100644 src/tealtiger/integrations/langfuse.py
 create mode 100644 tests/test_langfuse_integration.py

diff --git a/examples/langfuse_governance_traces.py b/examples/langfuse_governance_traces.py
new file mode 100644
index 0000000..cd31e0c
--- /dev/null
+++ b/examples/langfuse_governance_traces.py
@@ -0,0 +1,116 @@
+"""Example: Export TealTiger governance decisions to Langfuse.
+
+This example shows how governance decisions appear as spans in the Langfuse
+trace viewer — inline with your LLM traces.
+
+Requirements:
+    pip install tealtiger langfuse
+
+Set environment variables:
+    LANGFUSE_PUBLIC_KEY=pk-...
+    LANGFUSE_SECRET_KEY=sk-...
+    LANGFUSE_HOST=https://cloud.langfuse.com  (or your self-hosted URL)
+    OPENAI_API_KEY=sk-...
+"""
+
+import os
+from langfuse import Langfuse
+from tealtiger.integrations.langfuse import LangfuseGovernanceExporter
+
+# --- Setup ---
+
+langfuse = Langfuse()
+exporter = LangfuseGovernanceExporter(langfuse)
+
+
+# --- Example 1: Manual export of governance decisions ---
+
+# Simulate an ALLOW decision
+exporter.trace({
+    "action": "ALLOW",
+    "correlation_id": "550e8400-e29b-41d4-a716-446655440000",
+    "agent_id": "research-bot",
+    "session_id": "session-001",
+    "tool_slug": "GITHUB_GET_REPOS",
+    "toolkit_slug": "github",
+    "reason": "Policy allows: tool in allowlist",
+    "reason_codes": ["POLICY_ALLOW"],
+    "risk_score": 0,
+    "evaluation_time_ms": 0.38,
+    "mode": "ENFORCE",
+    "pii_detected": [],
+    "cost_tracked": 0.001,
+    "cumulative_cost": 0.015,
+    "timestamp_ms": 1720000000000,
+})
+
+# Simulate a DENY decision
+exporter.trace({
+    "action": "DENY",
+    "correlation_id": "660e8400-e29b-41d4-a716-446655440001",
+    "agent_id": "research-bot",
+    "session_id": "session-001",
+    "tool_slug": "GMAIL_SEND_EMAIL",
+    "toolkit_slug": "gmail",
+    "reason": "Tool 'GMAIL_SEND_EMAIL' not in allowlist for agent 'research-bot'",
+    "reason_codes": ["TOOL_NOT_ALLOWED"],
+    "risk_score": 0.9,
+    "evaluation_time_ms": 0.52,
+    "mode": "ENFORCE",
+    "pii_detected": [],
+    "cost_tracked": 0.0,
+    "cumulative_cost": 0.015,
+    "timestamp_ms": 1720000001000,
+})
+
+# Simulate a PII detection (monitor mode)
+exporter.trace({
+    "action": "MONITOR",
+    "correlation_id": "770e8400-e29b-41d4-a716-446655440002",
+    "agent_id": "research-bot",
+    "session_id": "session-001",
+    "tool_slug": "SLACK_SEND_MESSAGE",
+    "toolkit_slug": "slack",
+    "reason": "PII detected in arguments (monitor mode - not blocked)",
+    "reason_codes": ["PII_DETECTED"],
+    "risk_score": 0.6,
+    "evaluation_time_ms": 1.1,
+    "mode": "MONITOR",
+    "pii_detected": [{"type": "email", "start": 12, "end": 30}],
+    "cost_tracked": 0.002,
+    "cumulative_cost": 0.017,
+    "timestamp_ms": 1720000002000,
+})
+
+# Flush to ensure all events are sent
+exporter.flush()
+
+print("Governance decisions exported to Langfuse!")
+print("Check your Langfuse dashboard to see the traces.")
+print()
+print("In the Langfuse UI you'll see:")
+print("  - ALLOW decisions: DEFAULT level (grey)")
+print("  - DENY decisions: ERROR level (red)")
+print("  - MONITOR decisions: WARNING level (amber)")
+print()
+print("Each span shows: tool name, action, reason codes, risk score,")
+print("evaluation time, PII findings, and cost tracking.")
+
+
+# --- Example 2: Using with TealTiger observe() ---
+# (Uncomment when running with a real OpenAI key)
+
+# from tealtiger import observe
+# from openai import OpenAI
+#
+# client = observe(
+#     OpenAI(),
+#     agent_id="my-agent",
+#     on_decision=exporter.trace,  # Each decision → Langfuse span
+# )
+#
+# # All governance decisions now appear in Langfuse traces
+# response = client.chat.completions.create(
+#     model="gpt-4o-mini",
+#     messages=[{"role": "user", "content": "Hello!"}]
+# )
diff --git a/src/tealtiger/integrations/__init__.py b/src/tealtiger/integrations/__init__.py
new file mode 100644
index 0000000..7eeac76
--- /dev/null
+++ b/src/tealtiger/integrations/__init__.py
@@ -0,0 +1,5 @@
+"""TealTiger integrations with external observability and monitoring platforms."""
+
+from tealtiger.integrations.langfuse import LangfuseGovernanceExporter
+
+__all__ = ["LangfuseGovernanceExporter"]
diff --git a/src/tealtiger/integrations/langfuse.py b/src/tealtiger/integrations/langfuse.py
new file mode 100644
index 0000000..126f0da
--- /dev/null
+++ b/src/tealtiger/integrations/langfuse.py
@@ -0,0 +1,189 @@
+"""TealTiger → Langfuse governance trace exporter.
+
+Exports TealTiger governance decisions as Langfuse spans, enabling teams to see
+governance enforcement inline with their LLM traces in the Langfuse UI.
+
+Usage:
+    from langfuse import Langfuse
+    from tealtiger.integrations.langfuse import LangfuseGovernanceExporter
+
+    langfuse = Langfuse()
+    exporter = LangfuseGovernanceExporter(langfuse)
+
+    # Use as on_decision callback
+    client = observe(OpenAI(), on_decision=exporter.trace)
+
+    # Or manually export a decision
+    exporter.trace(decision)
+"""
+
+from __future__ import annotations
+
+import time
+from typing import Any, Dict, Optional
+
+try:
+    from langfuse import Langfuse
+    from langfuse.client import StatefulSpanClient, StatefulTraceClient
+except ImportError:
+    raise ImportError(
+        "langfuse is required for this integration. "
+        "Install it with: pip install langfuse"
+    )
+
+
+class LangfuseGovernanceExporter:
+    """Export TealTiger governance decisions as Langfuse spans.
+
+    Each governance decision becomes a Langfuse span with:
+    - name: "tealtiger.governance"
+    - metadata: {action, reason_codes, risk_score, evaluation_time_ms, policy_digest, ...}
+    - level: ERROR (deny), WARNING (monitor), DEFAULT (allow)
+    - input: tool/action being governed
+    - output: governance decision result
+
+    Args:
+        langfuse: An initialized Langfuse client instance.
+        trace_name: Name for the parent trace (default: "tealtiger-governance").
+        span_name: Name for individual governance spans (default: "tealtiger.governance").
+        flush_on_trace: Whether to flush after each trace call (default: False).
+    """
+
+    def __init__(
+        self,
+        langfuse: "Langfuse",
+        trace_name: str = "tealtiger-governance",
+        span_name: str = "tealtiger.governance",
+        flush_on_trace: bool = False,
+    ):
+        self._langfuse = langfuse
+        self._trace_name = trace_name
+        self._span_name = span_name
+        self._flush_on_trace = flush_on_trace
+        self._active_traces: Dict[str, Any] = {}
+
+    def _action_to_level(self, action: str) -> str:
+        """Map governance action to Langfuse span level."""
+        action_upper = action.upper() if action else "ALLOW"
+        if action_upper == "DENY":
+            return "ERROR"
+        elif action_upper in ("MONITOR", "REFER"):
+            return "WARNING"
+        else:
+            return "DEFAULT"
+
+    def _get_or_create_trace(
+        self, session_id: Optional[str] = None, agent_id: Optional[str] = None
+    ) -> Any:
+        """Get existing trace for a session or create a new one."""
+        trace_key = session_id or agent_id or "default"
+
+        if trace_key not in self._active_traces:
+            trace = self._langfuse.trace(
+                name=self._trace_name,
+                session_id=session_id,
+                user_id=agent_id,
+                metadata={
+                    "source": "tealtiger",
+                    "agent_id": agent_id,
+                },
+            )
+            self._active_traces[trace_key] = trace
+
+        return self._active_traces[trace_key]
+
+    def trace(self, decision: Dict[str, Any], **kwargs) -> None:
+        """Export a governance decision as a Langfuse span.
+
+        This method is designed to be used as the `on_decision` callback
+        for TealTiger's observe() or TealEngine.
+
+        Args:
+            decision: A TealTiger GovernanceDecision dict containing at minimum:
+                - action: "ALLOW", "DENY", "MONITOR", or "REFER"
+                - correlation_id: UUID v4 for the decision
+                - Optional: reason, reason_codes, risk_score, evaluation_time_ms,
+                  agent_id, session_id, tool_slug, pii_detected, cost_tracked, etc.
+        """
+        action = decision.get("action", "ALLOW")
+        correlation_id = decision.get("correlation_id", "")
+        agent_id = decision.get("agent_id")
+        session_id = decision.get("session_id")
+
+        # Get or create parent trace
+        trace = self._get_or_create_trace(
+            session_id=session_id, agent_id=agent_id
+        )
+
+        # Build span metadata
+        metadata = {
+            "action": action,
+            "reason_codes": decision.get("reason_codes", []),
+            "risk_score": decision.get("risk_score", 0),
+            "evaluation_time_ms": decision.get("evaluation_time_ms", 0),
+            "mode": decision.get("mode", "OBSERVE"),
+            "pii_detected": decision.get("pii_detected", []),
+            "cost_tracked": decision.get("cost_tracked", 0),
+            "cumulative_cost": decision.get("cumulative_cost", 0),
+        }
+
+        # Add policy digest if present
+        if "policy_digest" in decision:
+            metadata["policy_digest"] = decision["policy_digest"]
+        if "policy_ref" in decision:
+            metadata["policy_ref"] = decision["policy_ref"]
+
+        # Build input context
+        input_data = {}
+        if "tool_slug" in decision:
+            input_data["tool"] = decision["tool_slug"]
+        if "toolkit_slug" in decision:
+            input_data["toolkit"] = decision["toolkit_slug"]
+        if "intent_ref" in decision:
+            input_data["intent"] = decision["intent_ref"]
+
+        # Build output
+        output_data = {
+            "action": action,
+            "reason": decision.get("reason", ""),
+            "reason_codes": decision.get("reason_codes", []),
+        }
+
+        # Determine span level
+        level = self._action_to_level(action)
+
+        # Calculate timestamps
+        timestamp_ms = decision.get("timestamp_ms")
+        start_time = None
+        end_time = None
+        if timestamp_ms:
+            eval_time_ms = decision.get("evaluation_time_ms", 0)
+            # Start time is timestamp minus evaluation time
+            start_time = (timestamp_ms - eval_time_ms) / 1000.0
+            end_time = timestamp_ms / 1000.0
+
+        # Create the span
+        span = trace.span(
+            name=self._span_name,
+            span_id=correlation_id or None,
+            input=input_data if input_data else None,
+            output=output_data,
+            level=level,
+            metadata=metadata,
+            status_message=decision.get("reason", ""),
+        )
+
+        # End the span
+        span.end()
+
+        if self._flush_on_trace:
+            self._langfuse.flush()
+
+    def flush(self) -> None:
+        """Flush all pending Langfuse events."""
+        self._langfuse.flush()
+
+    def shutdown(self) -> None:
+        """Flush and shutdown the Langfuse client."""
+        self._langfuse.flush()
+        self._langfuse.shutdown()
diff --git a/tests/test_langfuse_integration.py b/tests/test_langfuse_integration.py
new file mode 100644
index 0000000..d3a2a53
--- /dev/null
+++ b/tests/test_langfuse_integration.py
@@ -0,0 +1,138 @@
+"""Tests for TealTiger → Langfuse governance trace export."""
+
+import pytest
+from unittest.mock import MagicMock, patch, call
+
+
+@pytest.fixture
+def mock_langfuse():
+    """Create a mock Langfuse client."""
+    langfuse = MagicMock()
+    mock_trace = MagicMock()
+    mock_span = MagicMock()
+    langfuse.trace.return_value = mock_trace
+    mock_trace.span.return_value = mock_span
+    return langfuse
+
+
+@pytest.fixture
+def exporter(mock_langfuse):
+    """Create a LangfuseGovernanceExporter with mocked client."""
+    with patch.dict("sys.modules", {"langfuse": MagicMock(), "langfuse.client": MagicMock()}):
+        from tealtiger.integrations.langfuse import LangfuseGovernanceExporter
+        return LangfuseGovernanceExporter(mock_langfuse)
+
+
+def test_allow_decision_creates_default_level_span(exporter, mock_langfuse):
+    """ALLOW decisions should create spans with DEFAULT level."""
+    decision = {
+        "action": "ALLOW",
+        "correlation_id": "test-123",
+        "agent_id": "coder",
+        "session_id": "session-1",
+        "reason": "Policy allows",
+        "reason_codes": ["POLICY_ALLOW"],
+        "risk_score": 0,
+        "evaluation_time_ms": 0.42,
+        "mode": "ENFORCE",
+    }
+
+    exporter.trace(decision)
+
+    mock_langfuse.trace.assert_called_once()
+    trace = mock_langfuse.trace.return_value
+    trace.span.assert_called_once()
+
+    span_kwargs = trace.span.call_args[1]
+    assert span_kwargs["level"] == "DEFAULT"
+    assert span_kwargs["name"] == "tealtiger.governance"
+
+
+def test_deny_decision_creates_error_level_span(exporter, mock_langfuse):
+    """DENY decisions should create spans with ERROR level."""
+    decision = {
+        "action": "DENY",
+        "correlation_id": "test-456",
+        "agent_id": "coder",
+        "reason": "Tool not in allowlist",
+        "reason_codes": ["TOOL_NOT_ALLOWED"],
+        "risk_score": 0.8,
+        "evaluation_time_ms": 1.2,
+        "tool_slug": "GMAIL_SEND_EMAIL",
+        "toolkit_slug": "gmail",
+    }
+
+    exporter.trace(decision)
+
+    trace = mock_langfuse.trace.return_value
+    span_kwargs = trace.span.call_args[1]
+    assert span_kwargs["level"] == "ERROR"
+    assert span_kwargs["input"]["tool"] == "GMAIL_SEND_EMAIL"
+    assert span_kwargs["output"]["action"] == "DENY"
+
+
+def test_monitor_decision_creates_warning_level_span(exporter, mock_langfuse):
+    """MONITOR decisions should create spans with WARNING level."""
+    decision = {
+        "action": "MONITOR",
+        "correlation_id": "test-789",
+        "reason": "PII detected but not blocked",
+        "reason_codes": ["PII_DETECTED"],
+        "pii_detected": [{"type": "email", "start": 5, "end": 20}],
+    }
+
+    exporter.trace(decision)
+
+    trace = mock_langfuse.trace.return_value
+    span_kwargs = trace.span.call_args[1]
+    assert span_kwargs["level"] == "WARNING"
+    assert span_kwargs["metadata"]["pii_detected"] == [{"type": "email", "start": 5, "end": 20}]
+
+
+def test_metadata_includes_governance_fields(exporter, mock_langfuse):
+    """Span metadata should include all governance-relevant fields."""
+    decision = {
+        "action": "ALLOW",
+        "correlation_id": "test-meta",
+        "risk_score": 0.3,
+        "evaluation_time_ms": 2.1,
+        "mode": "ENFORCE",
+        "cost_tracked": 0.005,
+        "cumulative_cost": 1.23,
+        "policy_digest": "sha256:abc123",
+    }
+
+    exporter.trace(decision)
+
+    trace = mock_langfuse.trace.return_value
+    metadata = trace.span.call_args[1]["metadata"]
+    assert metadata["risk_score"] == 0.3
+    assert metadata["evaluation_time_ms"] == 2.1
+    assert metadata["mode"] == "ENFORCE"
+    assert metadata["cost_tracked"] == 0.005
+    assert metadata["cumulative_cost"] == 1.23
+    assert metadata["policy_digest"] == "sha256:abc123"
+
+
+def test_same_session_reuses_trace(exporter, mock_langfuse):
+    """Multiple decisions in the same session should reuse the same trace."""
+    decision1 = {"action": "ALLOW", "session_id": "session-x", "correlation_id": "1"}
+    decision2 = {"action": "DENY", "session_id": "session-x", "correlation_id": "2"}
+
+    exporter.trace(decision1)
+    exporter.trace(decision2)
+
+    # Should only create one trace for the session
+    assert mock_langfuse.trace.call_count == 1
+
+
+def test_flush_on_trace_option(mock_langfuse):
+    """When flush_on_trace=True, flush after each trace call."""
+    with patch.dict("sys.modules", {"langfuse": MagicMock(), "langfuse.client": MagicMock()}):
+        from tealtiger.integrations.langfuse import LangfuseGovernanceExporter
+        exporter = LangfuseGovernanceExporter(mock_langfuse, flush_on_trace=True)
+
+    decision = {"action": "ALLOW", "correlation_id": "flush-test"}
+    exporter.trace(decision)
+
+    mock_langfuse.flush.assert_called_once()

From 28cc75b67d936125d3b10443839ee33d924f88df Mon Sep 17 00:00:00 2001
From: NagaSatish <satish.1290@gmail.com>
Date: Thu, 16 Jul 2026 12:06:56 +0530
Subject: [PATCH 2/2] fix: Remove unused imports and dead code (CodeQL
 findings)

---
 examples/langfuse_governance_traces.py |  1 -
 src/tealtiger/integrations/langfuse.py | 12 ------------
 tests/test_langfuse_integration.py     |  2 +-
 3 files changed, 1 insertion(+), 14 deletions(-)

diff --git a/examples/langfuse_governance_traces.py b/examples/langfuse_governance_traces.py
index cd31e0c..01e3dea 100644
--- a/examples/langfuse_governance_traces.py
+++ b/examples/langfuse_governance_traces.py
@@ -13,7 +13,6 @@
     OPENAI_API_KEY=sk-...
 """
 
-import os
 from langfuse import Langfuse
 from tealtiger.integrations.langfuse import LangfuseGovernanceExporter
 
diff --git a/src/tealtiger/integrations/langfuse.py b/src/tealtiger/integrations/langfuse.py
index 126f0da..2a1dac7 100644
--- a/src/tealtiger/integrations/langfuse.py
+++ b/src/tealtiger/integrations/langfuse.py
@@ -19,12 +19,10 @@
 
 from __future__ import annotations
 
-import time
 from typing import Any, Dict, Optional
 
 try:
     from langfuse import Langfuse
-    from langfuse.client import StatefulSpanClient, StatefulTraceClient
 except ImportError:
     raise ImportError(
         "langfuse is required for this integration. "
@@ -152,16 +150,6 @@ def trace(self, decision: Dict[str, Any], **kwargs) -> None:
         # Determine span level
         level = self._action_to_level(action)
 
-        # Calculate timestamps
-        timestamp_ms = decision.get("timestamp_ms")
-        start_time = None
-        end_time = None
-        if timestamp_ms:
-            eval_time_ms = decision.get("evaluation_time_ms", 0)
-            # Start time is timestamp minus evaluation time
-            start_time = (timestamp_ms - eval_time_ms) / 1000.0
-            end_time = timestamp_ms / 1000.0
-
         # Create the span
         span = trace.span(
             name=self._span_name,
diff --git a/tests/test_langfuse_integration.py b/tests/test_langfuse_integration.py
index d3a2a53..2ef3bd5 100644
--- a/tests/test_langfuse_integration.py
+++ b/tests/test_langfuse_integration.py
@@ -1,7 +1,7 @@
 """Tests for TealTiger → Langfuse governance trace export."""
 
 import pytest
-from unittest.mock import MagicMock, patch, call
+from unittest.mock import MagicMock, patch
 
 
 @pytest.fixture