feat(tools): TraceIndex — progressive trace disclosure for judge-based evaluators

README.md

@@ -200,6 +200,31 @@ evaluator = TrajectoryEvaluator(
 )
 ```
 
+### Evaluating Large Traces with Progressive Disclosure
+
+When a session is too large to inline into a judge prompt (large tool results,
+many turns), give the judge a compact overview plus discovery tools instead of
+the full trajectory. The judge loads only the spans the rubric requires:
+
+```python
+from strands_evals.evaluators import OutputEvaluator
+from strands_evals.tools.trace_index import TraceIndex
+
+index = TraceIndex(session)  # session: a Session from any provider/mapper
+
+evaluator = OutputEvaluator(
+    rubric="Every factual claim must be supported by tool-result evidence in the trace.",
+    tools=index.tools,  # list_spans, get_span, search_spans
+)
+
+# Compose the prompt with the compact overview instead of the full trajectory
+evaluation_output = f"{agent_answer}\n\n<TraceOverview>\n{index.overview()}\n</TraceOverview>"
+```
+
+The overview is one line per span (index, type, tool name, sizes, preview);
+`get_span` pages through oversized spans so no single tool return can overflow
+the judge's context.
+
 ### Trace-based Helpfulness Evaluation
 
 Evaluate agent helpfulness using OpenTelemetry traces with seven-level scoring:

src/strands_evals/tools/trace_index.py

@@ -0,0 +1,177 @@
+"""Progressive trace disclosure for judge agents.
+
+A large agent trajectory does not fit in a judge's context window. Rather than
+inlining the whole Session into the evaluation prompt (which overflows and gets
+scored as a failure), `TraceIndex` builds a small in-memory index over the trace
+and gives the judge two things:
+
+1. An `overview()` — one line per span (index, type, tool name, sizes, truncated
+   preview) — cheap enough to always fit in context, and
+2. **Lookup tools** the judge calls to load only the spans it needs to verify the
+   rubric: `list_spans`, `get_span`, `search_spans`.
+
+This is the same list / get / search shape used to query any indexed collection,
+and the same progressive-disclosure pattern skills use: the overview is the
+"name + description" line; the tools load the full content on demand.
+
+Example::
+
+    from strands_evals.evaluators import TrajectoryEvaluator
+    from strands_evals.tools.trace_index import TraceIndex
+
+    index = TraceIndex(session)
+    evaluator = TrajectoryEvaluator(
+        rubric="Every claim in the final response must be supported by a tool result.",
+        tools=index.tools,
+    )
+    # Compose the prompt with index.overview() instead of the full trajectory.
+"""
+
+import json
+import re
+
+from strands import tool
+
+from ..types.trace import (
+    AgentInvocationSpan,
+    InferenceSpan,
+    Session,
+    SpanUnion,
+    ToolExecutionSpan,
+)
+
+_PREVIEW_CHARS = 120
+_DEFAULT_MAX_READ_CHARS = 8_000
+
+
+def _flatten_spans(session: Session) -> list[SpanUnion]:
+    """Flatten all spans across traces in start_time order."""
+    spans = [span for trace in session.traces for span in trace.spans]
+    spans.sort(key=lambda s: s.span_info.start_time)
+    return spans
+
+
+def _span_text(span: SpanUnion) -> str:
+    """Full text content of a span, for search and retrieval."""
+    if isinstance(span, ToolExecutionSpan):
+        return json.dumps(
+            {
+                "tool_call": span.tool_call.model_dump(),
+                "tool_result": span.tool_result.model_dump(),
+            },
+            default=str,
+        )
+    if isinstance(span, AgentInvocationSpan):
+        return json.dumps(
+            {"user_prompt": span.user_prompt, "agent_response": span.agent_response},
+            default=str,
+        )
+    if isinstance(span, InferenceSpan):
+        return json.dumps([m.model_dump() for m in span.messages], default=str)
+    return json.dumps(span.model_dump(), default=str)
+
+
+def _preview(text: str, limit: int = _PREVIEW_CHARS) -> str:
+    text = re.sub(r"\s+", " ", text).strip()
+    return text if len(text) <= limit else text[: limit - 3] + "..."
+
+
+def _describe(span: SpanUnion) -> str:
+    """One overview line describing a span without its full payload."""
+    if isinstance(span, ToolExecutionSpan):
+        args = json.dumps(span.tool_call.arguments, default=str)
+        result_size = len(str(span.tool_result.content))
+        return (
+            f"TOOL {span.tool_call.name}({_preview(args, 80)}) "
+            f"-> result: {result_size} chars: {_preview(str(span.tool_result.content))}"
+        )
+    if isinstance(span, AgentInvocationSpan):
+        return (
+            f"AGENT prompt: {_preview(span.user_prompt, 80)} "
+            f"-> response: {len(span.agent_response)} chars: {_preview(span.agent_response)}"
+        )
+    if isinstance(span, InferenceSpan):
+        return f"INFERENCE {len(span.messages)} messages"
+    return f"{type(span).__name__}"
+
+
+class TraceIndex:
+    """Read-only list / get / search index over a Session for judge agents.
+
+    Attributes:
+        session: The Session being evaluated.
+        max_read_chars: Cap on any single tool return, so a huge span can't
+            overflow the judge's context in one call. Oversized content is
+            windowed and the tool reports how to page through it.
+    """
+
+    def __init__(self, session: Session, max_read_chars: int = _DEFAULT_MAX_READ_CHARS):
+        self.session = session
+        self.max_read_chars = max_read_chars
+        self._spans = _flatten_spans(session)
+
+        # Bind instance state into plain functions so @tool sees clean signatures.
+        # `this` (not `index`) so the public get_span(index=...) arg name is free.
+        this = self
+
+        @tool
+        def list_spans() -> str:
+            """List every span in the trace: one line per span with its index, type,
+            tool name, argument preview, and result size. Call this first to decide
+            which spans to inspect."""
+            return this.overview()
+
+        @tool
+        def get_span(index: int, offset: int = 0) -> str:
+            """Get the full content of one span by its index from the span list.
+            Large spans are windowed; the response says how to page with offset.
+
+            Args:
+                index: Span index as shown by list_spans.
+                offset: Character offset for paging through oversized spans.
+            """
+            if not 0 <= index < len(this._spans):
+                return f"ERROR: index {index} out of range (0..{len(this._spans) - 1})"
+            return this._window(_span_text(this._spans[index]), offset)
+
+        @tool
+        def search_spans(pattern: str, max_matches: int = 20) -> str:
+            """Search all span content for a regex or literal string. Returns matching
+            span indices with a short excerpt around each match. Use get_span to load
+            a matching span in full.
+
+            Args:
+                pattern: Regex (or literal text) to search for.
+                max_matches: Maximum matches to return.
+            """
+            try:
+                rx = re.compile(pattern, re.IGNORECASE)
+            except re.error:
+                rx = re.compile(re.escape(pattern), re.IGNORECASE)
+            hits = []
+            for i, span in enumerate(this._spans):
+                text = _span_text(span)
+                m = rx.search(text)
+                if m:
+                    start = max(0, m.start() - 60)
+                    hits.append(f"[{i}] ...{_preview(text[start : m.end() + 60], 160)}...")
+                if len(hits) >= max_matches:
+                    break
+            return "\n".join(hits) if hits else f"No matches for {pattern!r}"
+
+        self.tools = [list_spans, get_span, search_spans]
+
+    def overview(self) -> str:
+        """Compact one-line-per-span overview of the session."""
+        lines = [f"Trace overview: {len(self._spans)} spans (session {self.session.session_id})"]
+        lines += [f"[{i}] {_describe(span)}" for i, span in enumerate(self._spans)]
+        return "\n".join(lines)
+
+    def _window(self, text: str, offset: int) -> str:
+        if offset >= len(text):
+            return f"ERROR: offset {offset} beyond content length {len(text)}"
+        window = text[offset : offset + self.max_read_chars]
+        if offset + len(window) < len(text):
+            remaining = len(text) - offset - len(window)
+            window += f"\n[TRUNCATED: {remaining} chars remain; call again with offset={offset + len(window)}]"
+        return window

tests/strands_evals/tools/test_trace_index.py

@@ -0,0 +1,130 @@
+from datetime import datetime, timezone
+
+import pytest
+
+from strands_evals.tools.trace_index import TraceIndex
+from strands_evals.types.trace import (
+    AgentInvocationSpan,
+    Session,
+    SpanInfo,
+    ToolCall,
+    ToolExecutionSpan,
+    ToolResult,
+    Trace,
+)
+
+
+def _span_info(second: int) -> SpanInfo:
+    return SpanInfo(
+        session_id="s1",
+        span_id=f"sp{second}",
+        start_time=datetime(2026, 1, 1, 0, 0, second, tzinfo=timezone.utc),
+        end_time=datetime(2026, 1, 1, 0, 0, second + 1, tzinfo=timezone.utc),
+    )
+
+
+@pytest.fixture
+def session():
+    spans = [
+        AgentInvocationSpan(
+            span_info=_span_info(0),
+            user_prompt="Look up ticket TKT-1042",
+            agent_response="Ticket TKT-1042 was refunded $150.",
+            available_tools=[],
+        ),
+        ToolExecutionSpan(
+            span_info=_span_info(1),
+            tool_call=ToolCall(name="lookup_ticket", arguments={"id": "TKT-1042"}),
+            tool_result=ToolResult(content="x" * 20_000 + " refund_amount=$150"),
+        ),
+        ToolExecutionSpan(
+            span_info=_span_info(2),
+            tool_call=ToolCall(name="get_customer", arguments={"id": "C-7"}),
+            tool_result=ToolResult(content="customer name: Alex"),
+        ),
+    ]
+    return Session(traces=[Trace(spans=spans, trace_id="t1", session_id="s1")], session_id="s1")
+
+
+def test_overview_is_compact_and_ordered(session):
+    index = TraceIndex(session)
+    overview = index.overview()
+
+    lines = overview.splitlines()
+    assert "3 spans" in lines[0]
+    assert lines[1].startswith("[0] AGENT")
+    assert "lookup_ticket" in lines[2]
+    assert "get_customer" in lines[3]
+    # Manifest must not inline the 20K-char tool result
+    assert len(overview) < 2_000
+
+
+def test_get_span_returns_full_content_for_small_span(session):
+    index = TraceIndex(session)
+    get_span = index.tools[1]
+
+    content = get_span(index=2)
+
+    assert "customer name: Alex" in content
+    assert "TRUNCATED" not in content
+
+
+def test_get_span_windows_oversized_content_and_pages(session):
+    index = TraceIndex(session, max_read_chars=5_000)
+    get_span = index.tools[1]
+
+    first = get_span(index=1)
+    assert "TRUNCATED" in first
+    assert "offset=5000" in first
+
+    second = get_span(index=1, offset=5_000)
+    assert second.startswith("x") or '"' in second  # continuation, not a restart
+    assert first[:100] != second[:100]
+
+
+def test_get_span_index_out_of_range(session):
+    index = TraceIndex(session)
+    get_span = index.tools[1]
+
+    assert "ERROR" in get_span(index=99)
+    assert "ERROR" in get_span(index=-1)
+
+
+def test_search_spans_finds_span_by_content(session):
+    index = TraceIndex(session)
+    search_spans = index.tools[2]
+
+    result = search_spans(pattern=r"refund_amount=\$150")
+
+    assert result.startswith("[1]")
+    assert "refund_amount" in result
+
+
+def test_search_spans_falls_back_to_literal_on_bad_regex(session):
+    index = TraceIndex(session)
+    search_spans = index.tools[2]
+
+    result = search_spans(pattern="refund_amount=$150[")
+
+    assert "No matches" in result or result.startswith("[")
+
+
+def test_search_spans_no_matches(session):
+    index = TraceIndex(session)
+    search_spans = index.tools[2]
+
+    assert "No matches" in search_spans(pattern="nonexistent-zzz")
+
+
+def test_list_spans_tool_matches_overview(session):
+    index = TraceIndex(session)
+    list_spans = index.tools[0]
+
+    assert list_spans() == index.overview()
+
+
+def test_tools_are_strands_tools(session):
+    index = TraceIndex(session)
+
+    for t in index.tools:
+        assert hasattr(t, "tool_spec") or hasattr(t, "TOOL_SPEC") or callable(t)