feat: add unused LLM output detector (#142)

services/pipeline/src/sentinel_pipeline/detectors/__init__.py

@@ -5,6 +5,7 @@
 appends additional detectors to DETECTOR_REGISTRY at import time when installed.
 """
 from sentinel_pipeline.detectors.base import Detector
+from sentinel_pipeline.detectors.unused_llm_output import UnusedLlmOutputDetector
 from sentinel_pipeline.detectors.agent_loop import AgentLoopDetector
 from sentinel_pipeline.detectors.sequential_tools import SequentialToolsDetector
 from sentinel_pipeline.detectors.missing_termination_condition import MissingTerminationConditionDetector
@@ -23,6 +24,7 @@
     LatencySpikeDetector(),
     ContextCacheOpportunityDetector(),
     RetrievalWithoutGroundingDetector(),
+    UnusedLlmOutputDetector(),
 ]
 
 # Commercial engine hook: try to load additional detectors from sentinel_engine if installed.

services/pipeline/src/sentinel_pipeline/detectors/unused_llm_output.py

@@ -0,0 +1,89 @@
+from __future__ import annotations
+
+from typing import Any
+
+from sentinel_pipeline.models.graph import FlowGraph
+from sentinel_pipeline.models.insight import Insight, Severity, Tier
+from sentinel_pipeline.models.span import SpanKind
+from sentinel_pipeline.signals.extractor import Signals
+from .base import Detector
+
+
+def _contains_string(value: Any, target: str) -> bool:
+    """Recursively checks if the target string is present in any nested attribute values."""
+    if isinstance(value, str):
+        return target in value
+    if isinstance(value, dict):
+        return any(_contains_string(v, target) for v in value.values())
+    if isinstance(value, (list, tuple, set)):
+        return any(_contains_string(v, target) for v in value)
+    return False
+
+
+class UnusedLlmOutputDetector(Detector):
+    id       = "unused_llm_output"
+    name     = "Unused LLM Output"
+    severity = Severity.WARNING
+    tier     = Tier.FREE
+
+    def evaluate(self, graph: FlowGraph, signals: Signals) -> list[Insight] | None:
+        llm_spans = [s for s in graph.nodes.values() if s.kind == SpanKind.LLM_CALL]
+        if not llm_spans:
+            return None
+
+        insights: list[Insight] = []
+        for llm_span in llm_spans:
+            # 1. Retrieve the output text generated by the LLM
+            output_raw = llm_span.attributes.get("gen_ai.output")
+            if not output_raw:
+                continue
+            output_text = str(output_raw).strip()
+
+            # Skip empty or trivially short outputs to avoid false positives
+            if len(output_text) < 10:
+                continue
+
+            # 2. Check if the workflow continued after this LLM call
+            subsequent_spans = [
+                s for s in graph.nodes.values()
+                if s.span_id != llm_span.span_id and s.start_time >= llm_span.end_time
+            ]
+            if not subsequent_spans:
+                continue
+
+            # 3. Check if the output was referenced in any other span's attributes
+            referenced = False
+            for other_span in graph.nodes.values():
+                if other_span.span_id == llm_span.span_id:
+                    continue
+                if _contains_string(other_span.attributes, output_text):
+                    referenced = True
+                    break
+
+            # 4. If there were subsequent steps but the LLM output was never used, fire
+            if not referenced:
+                insights.append(Insight(
+                    workspace_id=graph.workspace_id,
+                    trace_id=graph.trace_id,
+                    detector_id=self.id,
+                    severity=self.severity,
+                    title="Unused LLM output",
+                    detail=(
+                        f"LLM call '{llm_span.name}' generated an output, but this output "
+                        f"was never referenced by any other downstream or parent span in the trace. "
+                        f"The generation cost ({llm_span.output_tokens or 0} output tokens) was wasted."
+                    ),
+                    recommendation=(
+                        "Review your agent's control flow. If the output of this LLM call is "
+                        "not needed for subsequent decisions or tool inputs, consider removing "
+                        "the call entirely or refining the prompt to omit this output."
+                    ),
+                    affected_span_ids=[llm_span.span_id],
+                    evidence={
+                        "llm_span_id": llm_span.span_id,
+                        "output_length": len(output_text),
+                        "output_tokens": llm_span.output_tokens or 0,
+                    },
+                ))
+
+        return insights or None

tests/unit/rules/test_unused_llm_output.py

@@ -0,0 +1,82 @@
+from datetime import datetime, timedelta, timezone
+
+from sentinel_pipeline.models.span import NormalizedSpan, SpanKind, SpanStatus
+from sentinel_pipeline.graph.builder import build_graph
+from sentinel_pipeline.signals.extractor import extract_signals
+from sentinel_pipeline.detectors.unused_llm_output import UnusedLlmOutputDetector
+
+rule = UnusedLlmOutputDetector()
+
+
+def _llm_span(span_id, output_text, offset_ms=0, duration_ms=1000):
+    t0 = datetime(2026, 1, 1, tzinfo=timezone.utc) + timedelta(milliseconds=offset_ms)
+    return NormalizedSpan(
+        span_id=span_id, trace_id="t1", parent_span_id="root",
+        name=span_id, kind=SpanKind.LLM_CALL, status=SpanStatus.OK,
+        start_time=t0, end_time=t0 + timedelta(milliseconds=duration_ms),
+        workspace_id="ws1",
+        output_tokens=150,
+        attributes={"gen_ai.output": output_text}
+    )
+
+
+def _tool_span(span_id, input_text, offset_ms=1050, duration_ms=500):
+    t0 = datetime(2026, 1, 1, tzinfo=timezone.utc) + timedelta(milliseconds=offset_ms)
+    return NormalizedSpan(
+        span_id=span_id, trace_id="t1", parent_span_id="root",
+        name=span_id, kind=SpanKind.TOOL_INVOKE, status=SpanStatus.OK,
+        start_time=t0, end_time=t0 + timedelta(milliseconds=duration_ms),
+        workspace_id="ws1",
+        attributes={"gen_ai.input": input_text}
+    )
+
+
+def _chain_span(span_id):
+    t0 = datetime(2026, 1, 1, tzinfo=timezone.utc)
+    return NormalizedSpan(
+        span_id=span_id, trace_id="t1", parent_span_id=None,
+        name=span_id, kind=SpanKind.CHAIN, status=SpanStatus.OK,
+        start_time=t0, end_time=t0 + timedelta(milliseconds=3000),
+        workspace_id="ws1",
+    )
+
+
+def test_fires_when_output_unused():
+    """An LLM output is generated and followed by a tool call, but never referenced."""
+    root = _chain_span("root")
+    llm = _llm_span("writer_llm", "This is the generated story output.", offset_ms=0)
+    tool = _tool_span("log_tool", "unrelated tool input contents", offset_ms=1100)
+
+    graph    = build_graph([root, llm, tool])
+    signals  = extract_signals(graph)
+    insights = rule.evaluate(graph, signals)
+
+    assert insights
+    assert len(insights) == 1
+    assert insights[0].detector_id == "unused_llm_output"
+    assert insights[0].affected_span_ids == ["writer_llm"]
+
+
+def test_no_fire_when_output_referenced():
+    """An LLM output is generated and referenced by a subsequent tool call."""
+    root = _chain_span("root")
+    llm = _llm_span("writer_llm", "This is the generated story output.", offset_ms=0)
+    tool = _tool_span("save_story_tool", "Saving: This is the generated story output.", offset_ms=1100)
+
+    graph    = build_graph([root, llm, tool])
+    signals  = extract_signals(graph)
+    insights = rule.evaluate(graph, signals)
+
+    assert not insights
+
+
+def test_no_fire_when_no_subsequent_spans():
+    """The LLM output is generated but is the last action in the trace (returned to user)."""
+    root = _chain_span("root")
+    llm = _llm_span("writer_llm", "This is the final response to the user.", offset_ms=0)
+
+    graph    = build_graph([root, llm])
+    signals  = extract_signals(graph)
+    insights = rule.evaluate(graph, signals)
+
+    assert not insights