diff --git a/simpleaudit/tracing/otlp.py b/simpleaudit/tracing/otlp.py index 5bcf7fd..bb59d90 100644 --- a/simpleaudit/tracing/otlp.py +++ b/simpleaudit/tracing/otlp.py @@ -113,11 +113,41 @@ def parse_otlp_json(payload: Any) -> List[Dict[str, Any]]: "end_time": _proto_ts_to_unix(span.get("endTimeUnixNano")), "status": _status_code(span.get("status")), "attributes": attrs, + **_status_message_and_events( + span.get("status"), span.get("events"), _decode_attrs, "timeUnixNano" + ), } ) return spans +def _status_message_and_events( + status: Optional[Dict[str, Any]], events: Any, decode_attrs: Any, time_key: str +) -> Dict[str, Any]: + """A span's status message and events, when it has them (exceptions live in events). + + OpenTelemetry records an exception as an ``exception`` event whose attributes + carry ``exception.type`` and ``exception.message``; the status message holds + the error description. Both are what a reader needs to see *why* a span failed. + """ + out: Dict[str, Any] = {} + message = (status or {}).get("message") if isinstance(status, dict) else None + if isinstance(message, str) and message: + out["status_message"] = message + decoded = [ + { + "name": event.get("name") or "event", + "time": _proto_ts_to_unix(event.get(time_key)), + "attributes": decode_attrs(event.get("attributes")), + } + for event in events or [] + if isinstance(event, dict) + ] + if decoded: + out["events"] = decoded + return out + + def _decode_attrs(attrs: Optional[List[Dict[str, Any]]]) -> Dict[str, Any]: out: Dict[str, Any] = {} for a in attrs or []: @@ -672,6 +702,9 @@ def _spans_from_otlp_dict(data: Dict[str, Any]) -> List[Dict[str, Any]]: "end_time": _proto_ts_to_unix(span.get("end_time_unix_nano")), "status": _status_from_dict(span.get("status")), "attributes": attrs, + **_status_message_and_events( + span.get("status"), span.get("events"), _attrs_from_list, "time_unix_nano" + ), } ) return spans diff --git a/simpleaudit/tracing/selection.py b/simpleaudit/tracing/selection.py index c6332f0..c8c0ba8 100644 --- a/simpleaudit/tracing/selection.py +++ b/simpleaudit/tracing/selection.py @@ -569,8 +569,61 @@ def evidence_spans_for_turn( return result.selected -def summarize_for_judge(result: SelectionResult, *, max_chars_per_span: int = 2000) -> str: - """Render selected spans into a compact text block for the judge prompt.""" +def _clip_values(value: Any, limit: int) -> Any: + """``value`` with every string longer than ``limit`` shortened (structure kept).""" + if isinstance(value, str): + return value if len(value) <= limit else value[:limit] + "…(truncated)" + if isinstance(value, dict): + return {key: _clip_values(item, limit) for key, item in value.items()} + if isinstance(value, list): + return [_clip_values(item, limit) for item in value] + return value + + +def _span_body(attrs: Dict[str, Any], max_chars_per_span: int, max_chars_per_value: int) -> str: + """The span's attributes as JSON within ``max_chars_per_span``, every key kept. + + Long values are shortened first (halving the per-value limit until the span + fits, down to 80 chars), so a long input history no longer pushes later + attributes such as the model's output out of view. Only if the keys alone + exceed the budget is the JSON cut at the end. + """ + limit = max(80, max_chars_per_value) + body = json.dumps(_clip_values(attrs, limit), ensure_ascii=False, default=str) + while len(body) > max_chars_per_span and limit > 80: + limit = max(80, limit // 2) + body = json.dumps(_clip_values(attrs, limit), ensure_ascii=False, default=str) + if len(body) > max_chars_per_span: + body = body[:max_chars_per_span] + "…(truncated)" + return body + + +def _failure_lines(span: Dict[str, Any], limit: int) -> List[str]: + """Why a span failed: its status message and exception events.""" + lines: List[str] = [] + if str(span.get("status") or "").upper() == "ERROR": + message = span.get("status_message") + lines.append("status: ERROR" + (f" — {str(message)[:limit]}" if message else "")) + for event in span.get("events") or []: + if not isinstance(event, dict) or event.get("name") != "exception": + continue + attrs = event.get("attributes") or {} + kind = attrs.get("exception.type") or "exception" + message = str(attrs.get("exception.message") or "")[:limit] + lines.append(f"exception: {kind}" + (f": {message}" if message else "")) + return lines + + +def summarize_for_judge( + result: SelectionResult, *, max_chars_per_span: int = 2000, max_chars_per_value: int = 600 +) -> str: + """Render selected spans into a compact text block for the judge prompt. + + Each span keeps all its attributes within ``max_chars_per_span``: long values + are shortened (``max_chars_per_value``, reduced further when needed) instead of + cutting the span's JSON from the front. A failed span also shows its status + message and exception. + """ if not result.selected: note = f"\n(elided {result.elided_count} span(s))" if result.elided else "" return f"(no evidence spans){note}" @@ -579,10 +632,9 @@ def summarize_for_judge(result: SelectionResult, *, max_chars_per_span: int = 20 prov = span.get("provenance", {}) header = f"[{prov.get('kind', '?')} {prov.get('name', '?')}] trace={prov.get('trace_id', '?')[:8]} span={prov.get('span_id', '?')[:8]}" attrs = _redacted(span.get("attributes") or {}) - body = json.dumps(attrs, ensure_ascii=False, default=str) - if len(body) > max_chars_per_span: - body = body[:max_chars_per_span] + "…(truncated)" - blocks.append(f"{header}\n{body}") + body = _span_body(attrs, max_chars_per_span, max_chars_per_value) + lines = [header, *(_redacted(line) for line in _failure_lines(span, max_chars_per_value)), body] + blocks.append("\n".join(lines)) text = "\n\n".join(blocks) if result.elided: text += f"\n\n(elided {result.elided_count} additional span(s): " + ", ".join( diff --git a/simpleaudit/tracing/store.py b/simpleaudit/tracing/store.py index cbfe48a..bc85514 100644 --- a/simpleaudit/tracing/store.py +++ b/simpleaudit/tracing/store.py @@ -69,6 +69,8 @@ def _attr(*keys: str) -> Any: "start_time": raw.get("start_time"), "end_time": raw.get("end_time"), "status": raw.get("status") or "OK", + # Why a span failed: the status message and the exception event (if any). + **{key: raw[key] for key in ("status_message", "events") if raw.get(key)}, } diff --git a/tests/test_tracing.py b/tests/test_tracing.py index e47f352..274a2fa 100644 --- a/tests/test_tracing.py +++ b/tests/test_tracing.py @@ -832,3 +832,52 @@ async def _run(): shared.stop() asyncio.run(_run()) + + +from simpleaudit.tracing.store import normalize_span # noqa: E402 + + +def test_parse_otlp_keeps_status_message_and_exception_events(): + """Why a span failed lives in its status message and its exception event.""" + payload = {"resourceSpans": [{"scopeSpans": [{"spans": [{ + "traceId": "t" * 32, "spanId": "s" * 16, "name": "execute_tool acme_get_order", + "startTimeUnixNano": "1000000000", "endTimeUnixNano": "2000000000", + "status": {"code": 2, "message": "ValueError: No order found"}, + "events": [{"name": "exception", "timeUnixNano": "1500000000", "attributes": [ + {"key": "exception.type", "value": {"stringValue": "ValueError"}}, + {"key": "exception.message", "value": {"stringValue": "No order found for 'ACME-9999'."}}, + ]}], + }]}]}]} + (span,) = parse_otlp_json(payload) + assert span["status"] == "ERROR" + assert span["status_message"] == "ValueError: No order found" + assert span["events"] == [{"name": "exception", "time": 1.5, "attributes": { + "exception.type": "ValueError", "exception.message": "No order found for 'ACME-9999'."}}] + normalized = normalize_span(span) + assert normalized["status_message"] == span["status_message"] and normalized["events"] == span["events"] + + +def test_parse_otlp_without_events_adds_no_keys(): + (span,) = parse_otlp_json(_otlp_payload())[:1] + assert "events" not in span and "status_message" not in span + + +def test_summarize_for_judge_shortens_values_and_keeps_every_attribute(): + """A long input history no longer pushes the model's output out of the span.""" + span = {"span_id": "a" * 16, "trace_id": "b" * 32, "name": "chat", "kind": "LLM", "status": "OK", + "attributes": {"gen_ai.input.messages": "x" * 5000, "gen_ai.output.messages": "the answer"}} + text = summarize_for_judge(select_spans([span])) + assert '"gen_ai.output.messages": "the answer"' in text + assert "…(truncated)" in text + assert len(text) < 2300 + + +def test_summarize_for_judge_shows_why_a_span_failed(): + span = normalize_span({"span_id": "a" * 16, "trace_id": "b" * 32, "name": "execute_tool", "kind": "TOOL", + "status": "ERROR", "status_message": "lookup failed", + "events": [{"name": "exception", "attributes": { + "exception.type": "ValueError", "exception.message": "No order found"}}], + "attributes": {"gen_ai.tool.name": "acme_get_order"}}) + text = summarize_for_judge(select_spans([span])) + assert "status: ERROR — lookup failed" in text + assert "exception: ValueError: No order found" in text