fix(lens): resolve canonical span content in bounded batches

This commit is contained in:
moe-berri 2026-10-06 13:18:43 -07:00
parent b6d587d23e
commit 2b001ef0bb
30 changed files with 829 additions and 125 deletions

View file

@ -318,6 +318,29 @@ impl NativeTraceStorage {
)
}
fn get_spans<'py>(
&self,
py: Python<'py>,
trace_id: String,
span_ids: Vec<String>,
#[pyo3(from_py_with = litellm_host_python::from_py_argument)] scope: ReadAccessParams,
trace_ref: String,
) -> PyResult<Bound<'py, PyAny>> {
let client = crate::http::host_client(py, ClientVariant::NoRedirect)?;
let connection = self.config.storage().reader().clone();
let reader = Arc::clone(&self.reader);
crate::execution::run_async(
py,
async move {
let store = ClickHouseTraces::new(client, connection);
reader
.get_spans(&store, &scope, &trace_id, &span_ids, &trace_ref)
.await
},
map_read_error,
)
}
#[pyo3(signature = (trace_id, span_id, scope, trace_ref, cursor))]
fn get_span_error<'py>(
&self,

View file

@ -12,10 +12,10 @@ use crate::{
use litellm_traces::{
SpanDetail, SpanErrorPage, Trace, TracePage,
query::named::{
ListTracesParams, ReadAccessParams, SpanDetailParams, SpanErrorParams, TraceIdentityParams,
TraceSpansParams,
ListTracesParams, ReadAccessParams, SpanDetailParams, SpanDetailsParams, SpanErrorParams,
TraceIdentityParams, TraceSpansParams,
},
request::{TRACE_PAGE_SIZE_MAX, TRACE_PAGE_SIZE_MIN},
request::{SPAN_DETAILS_MAX, TRACE_PAGE_SIZE_MAX, TRACE_PAGE_SIZE_MIN},
resolve_trace, to_ui_content,
};
@ -242,14 +242,35 @@ impl TraceReader {
span_id: span_id.to_owned(),
};
let row = store.span_detail(&params).await.map_err(map_store_error)?;
Ok(row.map(|row| SpanDetail {
input_ui: to_ui_content(&row.input),
output_ui: to_ui_content(&row.output),
span_id: row.span_id,
input: row.input,
output: row.output,
attributes: row.attributes,
}))
let detail = row.map(span_detail);
response_fits(&detail, self.response_bytes)?;
Ok(detail)
}
pub async fn get_spans<S: TraceStore>(
&self,
store: &S,
access: &ReadAccessParams,
trace_id: &str,
span_ids: &[String],
trace_ref: &str,
) -> Result<Option<Vec<SpanDetail>>, ReadError<S::Error>> {
if span_ids.is_empty() || span_ids.len() > SPAN_DETAILS_MAX {
return Err(ReadError::InvalidParameters);
}
let Some(trace_ref) = reference(store, access, trace_id, trace_ref).await? else {
return Ok(None);
};
let params = SpanDetailsParams {
access: access.clone(),
trace_id: trace_id.to_owned(),
trace_ref,
span_ids: span_ids.to_vec(),
};
let rows = store.span_details(&params).await.map_err(map_store_error)?;
let details: Vec<_> = rows.into_iter().map(span_detail).collect();
response_fits(&details, self.response_bytes)?;
Ok(Some(details))
}
pub async fn get_span_error<S: TraceStore>(
@ -354,6 +375,25 @@ fn page<E>(
}
}
fn span_detail(row: litellm_traces::query::named::SpanDetailRow) -> SpanDetail {
SpanDetail {
input_ui: to_ui_content(&row.input),
output_ui: to_ui_content(&row.output),
span_id: row.span_id,
input: row.input,
output: row.output,
attributes: row.attributes,
}
}
fn response_fits<T: serde::Serialize, E>(value: &T, limit: usize) -> Result<(), ReadError<E>> {
let bytes = serde_json::to_vec(value).map_err(|error| ReadError::Encode(Arc::new(error)))?;
if bytes.len() > limit {
return Err(ReadError::TooLarge);
}
Ok(())
}
pub(super) fn now_ms() -> u64 {
(time::OffsetDateTime::now_utc().unix_timestamp_nanos() / 1_000_000) as u64
}

View file

@ -1,9 +1,9 @@
use std::future::Future;
use litellm_traces::query::named::{
ListTracesParams, ListTracesRow, SpanDetailParams, SpanDetailRow, SpanErrorParams,
SpanErrorRow, SpendByResponseIdsParams, SpendByResponseIdsRow, TraceIdentityParams,
TracePageSpansParams, TraceSpansParams, TraceSpansRow,
ListTracesParams, ListTracesRow, SpanDetailParams, SpanDetailRow, SpanDetailsParams,
SpanErrorParams, SpanErrorRow, SpendByResponseIdsParams, SpendByResponseIdsRow,
TraceIdentityParams, TracePageSpansParams, TraceSpansParams, TraceSpansRow,
};
#[derive(Debug, thiserror::Error)]
@ -55,6 +55,11 @@ pub trait TraceStore: Sync {
params: &SpanDetailParams,
) -> impl Future<Output = Result<Option<SpanDetailRow>, StoreError<Self::Error>>> + Send;
fn span_details(
&self,
params: &SpanDetailsParams,
) -> impl Future<Output = Result<Vec<SpanDetailRow>, StoreError<Self::Error>>> + Send;
fn span_error(
&self,
params: &SpanErrorParams,

View file

@ -238,6 +238,16 @@ impl TraceStore for FakeStore {
Ok(state.span_detail.clone())
}
async fn span_details(
&self,
_: &litellm_traces::query::named::SpanDetailsParams,
) -> Result<Vec<SpanDetailRow>, StoreError<Self::Error>> {
self.calls.span_detail.fetch_add(1, Ordering::SeqCst);
let state = self.state.lock().unwrap();
Self::failure(&state, Operation::SpanDetail)?;
Ok(state.span_detail.clone().into_iter().collect())
}
async fn span_error(
&self,
_: &SpanErrorParams,
@ -766,3 +776,55 @@ async fn concurrent_pages_of_an_evicted_snapshot_share_one_storage_read() {
assert_eq!(right.unwrap().unwrap().spans[0].span_id, "span-1");
assert_eq!(store.calls(Operation::TraceSpans), 1);
}
#[rstest]
#[case::empty(0)]
#[case::over_limit(litellm_traces::request::SPAN_DETAILS_MAX + 1)]
#[tokio::test]
async fn bulk_content_rejects_invalid_counts_before_storage(#[case] count: usize) {
let store = FakeStore::default();
let reader = TraceReader::new(1_000_000);
let result = reader
.get_spans(
&store,
&access(),
"trace",
&vec!["span".to_owned(); count],
"run",
)
.await;
assert!(matches!(result, Err(ReadError::InvalidParameters)));
assert_eq!(store.calls(Operation::SpanDetail), 0);
assert_eq!(store.calls(Operation::TraceRefs), 0);
}
#[rstest]
#[case::single(false)]
#[case::bulk(true)]
#[tokio::test]
async fn content_reads_enforce_response_budget(#[case] bulk: bool) {
let store = FakeStore::default();
store.state.lock().unwrap().span_detail = Some(SpanDetailRow {
span_id: "span".into(),
input: String::new(),
output: "x".repeat(1_024),
attributes: Default::default(),
});
let reader = TraceReader::new(100);
let oversized = if bulk {
matches!(
reader
.get_spans(&store, &access(), "trace", &["span".into()], "run")
.await,
Err(ReadError::TooLarge)
)
} else {
matches!(
reader
.get_span(&store, &access(), "trace", "span", "run")
.await,
Err(ReadError::TooLarge)
)
};
assert!(oversized);
}

View file

@ -1,24 +1,72 @@
SELECT o.SpanId AS span_id, o.Input AS input,
if(o.Output = '' AND o.ObservationType = 'agent', answer.output, o.Output) AS output,
o.SpanAttributes AS attributes
FROM otel_traces AS o
LEFT JOIN (
SELECT TeamId, ApiKeyHash, ParentSpanId AS parent_span_id, argMax(Output, Timestamp) AS output
FROM otel_traces
WHERE TraceId = {trace_id:String} AND ParentSpanId = {span_id:String}
AND ObservationType = 'llm' AND Output != ''
scoped AS (
SELECT * FROM otel_traces
WHERE TraceId = {trace_id:String}
AND ({all_teams:UInt8} = 1
OR ({user_id:String} != '' AND UserId = {user_id:String})
OR has({team_ids:Array(String)}, TeamId))
AND ({trace_ref:String} = '' OR
hex(SHA256(concat(TeamId, char(0), ApiKeyHash, char(0), TraceId))) = {trace_ref:String})
),
selected AS (
SELECT *, coalesce(nullIf(SpanAttributes['gen_ai.tool.call.id'], ''), SpanAttributes['tool_use_id']) AS call_id,
SpanAttributes['session.id'] AS session_id,
(Framework IN ('claude-code', 'claude-agent-sdk') AND ObservationType = 'tool' AND call_id != '') AS native_tool
FROM scoped WHERE has(requested_ids, SpanId)
ORDER BY Timestamp DESC LIMIT 1 BY TeamId, ApiKeyHash, SpanId
),
inputs AS (
SELECT TeamId, ApiKeyHash, SpanAttributes['session.id'] AS session_id,
SpanAttributes['tool_use_id'] AS call_id,
groupUniqArray(2)(Input) AS values, min(SpanId) AS source_span
FROM scoped
WHERE SpanName = 'claude_code.tool_result' AND Framework IN ('claude-code', 'claude-agent-sdk')
AND Input != '' AND call_id != '' AND call_id IN (SELECT call_id FROM selected WHERE native_tool)
GROUP BY TeamId, ApiKeyHash, session_id, call_id
),
outputs AS (
SELECT TeamId, ApiKeyHash, SpanAttributes['session.id'] AS session_id,
JSONExtractString(result, 'id') AS call_id,
groupUniqArray(2)(tuple(JSONExtractString(result, 'content'), JSONExtractBool(result, 'is_error'))) AS values,
min(SpanId) AS source_span
FROM scoped ARRAY JOIN JSONExtractArrayRaw(Output, 'tool_results') AS result
WHERE SpanName = 'claude_code.api_request_body' AND Framework IN ('claude-code', 'claude-agent-sdk')
AND call_id != '' AND call_id IN (SELECT call_id FROM selected WHERE native_tool)
AND JSONType(result, 'content') = 'String'
GROUP BY TeamId, ApiKeyHash, session_id, call_id
),
executions AS (
SELECT TeamId, ApiKeyHash, SpanAttributes['session.id'] AS session_id,
groupUniqArray(coalesce(nullIf(SpanAttributes['gen_ai.tool.call.id'], ''), SpanAttributes['tool_use_id'])) AS call_ids
FROM scoped
WHERE ObservationType = 'tool' AND Framework IN ('claude-code', 'claude-agent-sdk')
GROUP BY TeamId, ApiKeyHash, session_id
),
answers AS (
SELECT TeamId, ApiKeyHash, ParentSpanId AS parent_span_id, argMax(Output, Timestamp) AS output,
argMax(SpanId, Timestamp) AS source_span
FROM scoped
WHERE has(requested_ids, ParentSpanId) AND ObservationType = 'llm' AND Output != ''
GROUP BY TeamId, ApiKeyHash, ParentSpanId
) AS answer ON answer.parent_span_id = o.SpanId
AND answer.TeamId = o.TeamId AND answer.ApiKeyHash = o.ApiKeyHash
WHERE o.TraceId = {trace_id:String} AND o.SpanId = {span_id:String}
AND ({all_teams:UInt8} = 1
OR ({user_id:String} != '' AND o.UserId = {user_id:String})
OR has({team_ids:Array(String)}, o.TeamId))
AND ({trace_ref:String} = '' OR
hex(SHA256(concat(o.TeamId, char(0), o.ApiKeyHash, char(0), o.TraceId))) = {trace_ref:String})
LIMIT 1
)
SELECT o.SpanId AS span_id,
if(o.native_tool AND length(i.values) = 1, i.values[1], o.Input) AS input,
multiIf(o.Output != '', o.Output,
o.native_tool AND length(r.values) = 1, r.values[1].1,
o.ObservationType = 'agent', a.output, '') AS output,
mapUpdate(o.SpanAttributes, map(
'lens.content.unexecuted_tool_results', if(o.SpanName = 'claude_code.api_request_body',
concat('[', arrayStringConcat(arrayFilter(result -> JSONExtractString(result, 'id') != '' AND NOT has(e.call_ids, JSONExtractString(result, 'id')),
JSONExtractArrayRaw(o.Output, 'tool_results')), ','), ']'), '[]'),
'lens.content.input_source', multiIf(o.native_tool AND length(i.values) = 1, i.source_span, o.Input != '', o.SpanId, ''),
'lens.content.output_source', multiIf(o.Output != '', o.SpanId, o.native_tool AND length(r.values) = 1, r.source_span, o.ObservationType = 'agent' AND a.output != '', a.source_span, ''),
'lens.content.input_status', multiIf(o.native_tool AND length(i.values) > 1, 'conflicting', input != '', 'recorded', 'not_recorded'),
'lens.content.output_status', multiIf(o.Output != '', 'recorded', o.native_tool AND length(r.values) > 1, 'conflicting', o.native_tool AND length(r.values) = 1, 'recorded', output != '', 'recorded', 'not_recorded'),
'lens.content.tool_result_is_error', if(o.native_tool AND length(r.values) = 1, toString(r.values[1].2), '')
)) AS attributes
FROM selected AS o
LEFT JOIN inputs AS i ON o.native_tool AND o.TeamId = i.TeamId AND o.ApiKeyHash = i.ApiKeyHash AND o.session_id = i.session_id AND o.call_id = i.call_id
LEFT JOIN outputs AS r ON o.native_tool AND o.TeamId = r.TeamId AND o.ApiKeyHash = r.ApiKeyHash AND o.session_id = r.session_id AND o.call_id = r.call_id
LEFT JOIN executions AS e ON e.TeamId = o.TeamId AND e.ApiKeyHash = o.ApiKeyHash AND e.session_id = o.session_id
LEFT JOIN answers AS a ON a.parent_span_id = o.SpanId AND a.TeamId = o.TeamId AND a.ApiKeyHash = o.ApiKeyHash
ORDER BY indexOf(requested_ids, o.SpanId)
LIMIT length(requested_ids)

View file

@ -279,7 +279,22 @@ impl Query for SpanDetail {
type Params = SpanDetailParams;
type Row = SpanDetailRow;
const SQL: &'static str = include_str!("../../query/span_detail.sql");
const SQL: &'static str = concat!(
"WITH [{span_id:String}] AS requested_ids, ",
include_str!("../../query/span_detail.sql")
);
}
pub struct SpanDetails;
impl Query for SpanDetails {
type Params = contracts::SpanDetailsParams;
type Row = SpanDetailRow;
const SQL: &'static str = concat!(
"WITH {span_ids:Array(String)} AS requested_ids, ",
include_str!("../../query/span_detail.sql")
);
}
pub struct SpanError;

View file

@ -6,8 +6,8 @@ use litellm_traces_cache::{StoreError, TraceStore};
use crate::{
Connection, Error,
query::named::{
ListTracesParams, ListTracesRow, SpanDetail as SpanDetailQuery, SpanError, SpanErrorParams,
SpendByResponseIdsParams, TraceIdentity, TracePageSpansParams,
ListTracesParams, ListTracesRow, SpanDetail as SpanDetailQuery, SpanDetails, SpanError,
SpanErrorParams, SpendByResponseIdsParams, TraceIdentity, TracePageSpansParams,
},
};
@ -104,6 +104,18 @@ impl TraceStore for ClickHouseTraces {
) -> Result<Option<contracts::SpanDetailRow>, StoreError<Self::Error>> {
match fetch::<SpanDetailQuery>(&self.client, &self.connection, params).await {
Ok(rows) => Ok(rows.into_iter().next()),
Err(StorageError::ResponseTooLarge) => Err(StoreError::TooLarge),
Err(error) => Err(failed(error)),
}
}
async fn span_details(
&self,
params: &contracts::SpanDetailsParams,
) -> Result<Vec<contracts::SpanDetailRow>, StoreError<Self::Error>> {
match fetch::<SpanDetails>(&self.client, &self.connection, params).await {
Ok(rows) => Ok(rows),
Err(StorageError::ResponseTooLarge) => Err(StoreError::TooLarge),
Err(error) => Err(failed(error)),
}
}

View file

@ -710,3 +710,203 @@ async fn gateway_ids_resolve_through_detail_and_batch_reads_with_legacy_fallback
}
Ok(())
}
#[rstest]
#[case::recorded(false)]
#[case::conflicting(true)]
#[tokio::test]
async fn native_tool_content_agrees_between_single_and_bulk_reads(
#[future(awt)] migrated_database: TestResult<SeededDatabase>,
#[case] conflict: bool,
) -> TestResult {
use litellm_storage_clickhouse::fetch;
use litellm_traces::query::named::{SpanDetailParams, SpanDetailsParams};
use litellm_traces_clickhouse::query::named::{SpanDetail, SpanDetails};
let fixture = migrated_database?;
let client = &fixture.database.client;
let writer = Connection::writer(&fixture.database.url)?;
let rows = [
(
"tool",
"team-a",
"key-a",
"s",
"Bash",
"tool",
"",
r#"{"tool":"Bash"}"#,
"",
),
(
"input",
"team-a",
"key-a",
"s",
"claude_code.tool_result",
"framework",
"",
r#"{"command":"exit 7"}"#,
"",
),
(
"output",
"team-a",
"key-a",
"s",
"claude_code.api_request_body",
"framework",
"",
"",
r#"{"tool_results":[{"id":"call","content":"Exit code 7","is_error":true},{"id":"attempt","content":"Tool permission denied","is_error":true}]}"#,
),
(
"duplicate",
"team-a",
"key-a",
"s",
"claude_code.api_request_body",
"framework",
"",
"",
if conflict {
r#"{"tool_results":[{"id":"call","content":"different","is_error":false}]}"#
} else {
r#"{"tool_results":[{"id":"call","content":"Exit code 7","is_error":true},{"id":"attempt","content":"Tool permission denied","is_error":true}]}"#
},
),
(
"other-key",
"team-a",
"key-b",
"s",
"claude_code.tool_result",
"framework",
"",
"private-key-input",
"",
),
(
"other-team",
"team-b",
"key-a",
"s",
"claude_code.api_request_body",
"framework",
"",
"",
r#"{"tool_results":[{"id":"call","content":"private-team-output"}]}"#,
),
(
"other-session",
"team-a",
"key-a",
"other",
"claude_code.tool_result",
"framework",
"",
"other-session-input",
"",
),
(
"agent", "team-a", "key-a", "s", "agent", "agent", "", "", "",
),
(
"reply",
"team-a",
"key-a",
"s",
"model",
"llm",
"agent",
"",
"final answer",
),
];
insert_rows(
client,
&writer,
DATABASE,
InsertTable::OtelTraces,
rows.into_iter()
.map(
|(id, team, key, session, name, kind, parent, input, output)| {
BTreeMap::from([
("Timestamp".into(), json!(1_790_000_000_000_000_000_i64)),
("TraceId".into(), json!("content-trace")),
("SpanId".into(), json!(id)),
("ParentSpanId".into(), json!(parent)),
("SpanName".into(), json!(name)),
("ObservationType".into(), json!(kind)),
("Framework".into(), json!("claude-code")),
("Input".into(), json!(input)),
("Output".into(), json!(output)),
(
"SpanAttributes".into(),
json!({"session.id":session,"tool_use_id":"call"}),
),
("TeamId".into(), json!(team)),
("ApiKeyHash".into(), json!(key)),
])
},
)
.collect(),
)
.await?;
let connection = fixture
.readers
.connection(client, &QueryScope::All, "fixture-secret")
.await?;
let access = ReadAccessParams {
all_teams: false,
user_id: String::new(),
team_ids: vec!["team-a".into()],
};
let single = fetch::<SpanDetail>(
client,
&connection,
&SpanDetailParams {
access: access.clone(),
trace_id: "content-trace".into(),
trace_ref: String::new(),
span_id: "tool".into(),
},
)
.await?;
let bulk = fetch::<SpanDetails>(
client,
&connection,
&SpanDetailsParams {
access,
trace_id: "content-trace".into(),
trace_ref: String::new(),
span_ids: vec![
"tool".into(),
"agent".into(),
"output".into(),
"missing".into(),
],
},
)
.await?;
assert_eq!(
serde_json::to_value(&single[0])?,
serde_json::to_value(&bulk[0])?
);
assert_eq!(bulk.len(), 3);
assert_eq!(bulk[0].input, r#"{"command":"exit 7"}"#);
assert_eq!(bulk[0].attributes["lens.content.input_source"], "input");
assert_eq!(bulk[0].output, if conflict { "" } else { "Exit code 7" });
assert_eq!(
bulk[0].attributes["lens.content.output_status"],
if conflict { "conflicting" } else { "recorded" }
);
let attempts: serde_json::Value =
serde_json::from_str(&bulk[2].attributes["lens.content.unexecuted_tool_results"])?;
assert_eq!(
attempts,
json!([{"id":"attempt","content":"Tool permission denied","is_error":true}])
);
assert_eq!(bulk[1].output, "final answer");
assert_eq!(bulk[1].attributes["lens.content.output_source"], "reply");
Ok(())
}

View file

@ -131,6 +131,15 @@ pub struct SpanDetailParams {
pub span_id: String,
}
#[derive(Debug, Deserialize, Serialize)]
pub struct SpanDetailsParams {
#[serde(flatten)]
pub access: ReadAccessParams,
pub trace_id: String,
pub trace_ref: String,
pub span_ids: Vec<String>,
}
#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct SpanDetailRow {
pub span_id: String,

View file

@ -1,5 +1,6 @@
pub const TRACE_PAGE_SIZE_MIN: u16 = 1;
pub const TRACE_PAGE_SIZE_MAX: u16 = 500;
pub const SPAN_DETAILS_MAX: usize = 100;
#[macro_rules_attribute::apply(request_type)]
#[derive(Clone, Debug)]
@ -38,6 +39,16 @@ pub struct TraceSpanRequest {
pub trace_ref: String,
}
#[macro_rules_attribute::apply(request_type)]
#[derive(Clone, Debug)]
#[serde(deny_unknown_fields)]
pub struct TraceSpanDetailsRequest {
#[serde(default)]
pub trace_ref: String,
#[cfg_attr(feature = "schema", schemars(length(min = 1, max = SPAN_DETAILS_MAX)))]
pub span_ids: Vec<String>,
}
#[macro_rules_attribute::apply(request_type)]
#[derive(Clone, Debug)]
pub struct TraceErrorPageRequest {

View file

@ -80,6 +80,10 @@ pub fn request_schemas() -> BTreeMap<&'static str, Schema> {
"TraceSpanRequest",
requested::<crate::request::TraceSpanRequest>(),
),
(
"TraceSpanDetailsRequest",
requested::<crate::request::TraceSpanDetailsRequest>(),
),
(
"TraceErrorPageRequest",
requested::<crate::request::TraceErrorPageRequest>(),

View file

@ -34,6 +34,7 @@ from litellm.rust_bridge.trace.generated.requests import (
TraceErrorPageRequest,
TraceListRequest,
TraceQueryRequest,
TraceSpanDetailsRequest,
TraceSpanRequest,
)
from litellm.rust_bridge.trace.generated.responses import TraceSQLResponse
@ -291,6 +292,22 @@ async def get_agent_trace(
return trace
@router.post("/v1/traces/{trace_id}/spans", response_model=list[SpanDetail])
async def get_agent_trace_spans(
trace_id: str,
body: TraceSpanDetailsRequest,
context: Annotated[TraceAccessContext, Depends(provide_trace_access)],
) -> list[SpanDetail]:
tracing, scope = context.reader()
try:
spans: Final = await tracing.get_spans(trace_id, body.span_ids, scope, body.trace_ref)
except (TraceChanged, ValueError, OverflowError, RuntimeError) as error:
raise read_failure(error) from error
if spans is None:
raise HTTPException(status_code=404, detail=f"Trace {trace_id} not found")
return spans
@router.get("/v1/traces/{trace_id}/spans/{span_id}", response_model=SpanDetail)
async def get_agent_trace_span(
trace_id: str,

View file

@ -49,6 +49,9 @@ class NativeTraceStorage:
self, trace_id: str, scope: TraceScope, trace_ref: str, cursor: str | None = None, page_size: int | None = None
) -> Future[JsonValue]: ...
def get_span(self, trace_id: str, span_id: str, scope: TraceScope, trace_ref: str) -> Future[JsonValue]: ...
def get_spans(
self, trace_id: str, span_ids: Sequence[str], scope: TraceScope, trace_ref: str
) -> Future[JsonValue]: ...
def get_span_error(
self, trace_id: str, span_id: str, scope: TraceScope, trace_ref: str, cursor: str | None
) -> Future[JsonValue]: ...

View file

@ -47,6 +47,16 @@ class TraceQueryRequest(LiteLLMBaseModel):
sql: str
class TraceSpanDetailsRequest(LiteLLMBaseModel):
model_config = ConfigDict(
extra="forbid",
frozen=True,
)
trace_ref: str = ""
span_ids: tuple[str, ...] = Field(..., max_length=100, min_length=1)
class TraceSpanRequest(LiteLLMBaseModel):
model_config = ConfigDict(
frozen=True,
@ -56,6 +66,11 @@ class TraceSpanRequest(LiteLLMBaseModel):
TraceWireRequests: TypeAlias = Annotated[
TraceDetailRequest | TraceErrorPageRequest | TraceListRequest | TraceQueryRequest | TraceSpanRequest,
TraceDetailRequest
| TraceErrorPageRequest
| TraceListRequest
| TraceQueryRequest
| TraceSpanDetailsRequest
| TraceSpanRequest,
Field(..., title="TraceWireRequests"),
]

View file

@ -76,6 +76,10 @@ class NativeStore(Protocol):
def get_span(self, trace_id: str, span_id: str, scope: TraceScope, trace_ref: str) -> Awaitable[JsonValue]: ...
def get_spans(
self, trace_id: str, span_ids: Sequence[str], scope: TraceScope, trace_ref: str
) -> Awaitable[JsonValue]: ...
def get_span_error(
self, trace_id: str, span_id: str, scope: TraceScope, trace_ref: str, cursor: str | None
) -> Awaitable[JsonValue]: ...
@ -107,6 +111,7 @@ _HELP_RESPONSE: Final = TypeAdapter(TraceQueryHelp)
_TRACE_PAGE: Final = TypeAdapter(TracePage)
_TRACE: Final[TypeAdapter[Trace | None]] = TypeAdapter(Trace | None)
_SPAN_DETAIL: Final[TypeAdapter[SpanDetail | None]] = TypeAdapter(SpanDetail | None)
_SPAN_DETAILS: Final[TypeAdapter[list[SpanDetail] | None]] = TypeAdapter(list[SpanDetail] | None)
_SPAN_ERROR_PAGE: Final[TypeAdapter[SpanErrorPage | None]] = TypeAdapter(SpanErrorPage | None)
_ResponseT: Final = TypeVar("_ResponseT")
_NATIVE_ADAPTER: Final[TypeAdapter[NativeTraces]] = TypeAdapter(
@ -209,6 +214,12 @@ class ClickHouseStorage:
result: Final = await self._native.get_span(trace_id, span_id, scope, trace_ref)
return _validate_query_response(_SPAN_DETAIL, result)
async def get_spans(
self, trace_id: str, span_ids: Sequence[str], scope: TraceScope, trace_ref: str = ""
) -> list[SpanDetail] | None:
result: Final = await self._native.get_spans(trace_id, span_ids, scope, trace_ref)
return _validate_query_response(_SPAN_DETAILS, result)
async def get_span_error(
self, trace_id: str, span_id: str, scope: TraceScope, trace_ref: str = "", cursor: str | None = None
) -> SpanErrorPage | None:

View file

@ -13,7 +13,7 @@ The proxy endpoints are thin wrappers: auth -> build tenant/scope -> call one me
"""
import asyncio
from collections.abc import AsyncIterable, Callable, Mapping
from collections.abc import AsyncIterable, Callable, Mapping, Sequence
from io import BytesIO
from threading import BoundedSemaphore
from typing import Final
@ -114,6 +114,11 @@ class TraceReceiver:
async def get_span(self, trace_id: str, span_id: str, scope: TraceScope, trace_ref: str = "") -> SpanDetail | None:
return await self.storage.get_span(trace_id, span_id, scope, trace_ref)
async def get_spans(
self, trace_id: str, span_ids: Sequence[str], scope: TraceScope, trace_ref: str = ""
) -> list[SpanDetail] | None:
return await self.storage.get_spans(trace_id, span_ids, scope, trace_ref)
async def get_span_error(
self, trace_id: str, span_id: str, scope: TraceScope, trace_ref: str = "", cursor: str | None = None
) -> SpanErrorPage | None:

View file

@ -0,0 +1,23 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"additionalProperties": false,
"properties": {
"span_ids": {
"items": {
"type": "string"
},
"maxItems": 100,
"minItems": 1,
"type": "array"
},
"trace_ref": {
"default": "",
"type": "string"
}
},
"required": [
"span_ids"
],
"title": "TraceSpanDetailsRequest",
"type": "object"
}

View file

@ -1056,3 +1056,23 @@ async def test_storage_rejects_native_query_help_that_drifts_from_the_contract(
storage: Final = ClickHouseStorage(TraceStorageConfig("http://clickhouse:8123"))
with pytest.raises(RuntimeError, match="invalid response"):
await storage.query_help({"kind": "all"}, "secret")
@pytest.mark.parametrize("count", (0, 101))
def test_bulk_span_details_rejects_invalid_count(client: TestClient, receiver: MagicMock, count: int) -> None:
response: Final = client.post("/v1/traces/t1/spans", json={"span_ids": ["s1"] * count})
assert response.status_code == 422
receiver.get_spans.assert_not_called()
def test_bulk_span_details_preserves_scope_and_canonical_content(client: TestClient, receiver: MagicMock) -> None:
receiver.get_spans = AsyncMock(return_value=[SPAN_DETAIL_RESPONSE])
response: Final = client.post("/v1/traces/t1/spans", json={"trace_ref": "run-one", "span_ids": ["s1"]})
assert response.status_code == 200, response.text
assert response.json() == [SPAN_DETAIL_RESPONSE]
receiver.get_spans.assert_awaited_once_with(
"t1", ("s1",), {"all_teams": 0, "user_id": "user", "team_ids": ()}, "run-one"
)
receiver.get_spans.return_value = None
assert client.post("/v1/traces/missing/spans", json={"span_ids": ["s1"]}).status_code == 404
assert client.post("/v1/traces/t1/spans", json={"span_ids": ["s1"], "unexpected": True}).status_code == 422

View file

@ -90,6 +90,7 @@ function demoTracesApi(data: LensDemoData): TracesApi {
}),
anyRecorded: async () => data.runs.length > 0,
trace: (traceId) => found(run(traceId)?.trace),
spans: (traceId, spanIds) => found(run(traceId)?.details.filter((span) => spanIds.includes(span.span_id))),
span: (traceId, spanId) => found(run(traceId)?.details.find((span) => span.span_id === spanId)),
spanError: async (traceId, spanId) => {
const span = run(traceId)?.trace.spans.find((item) => item.span_id === spanId);

View file

@ -5,6 +5,7 @@ import {
agentTraceCall,
agentTraceListCall,
agentTraceSpanCall,
agentTraceSpansCall,
agentTraceSpanErrorCall,
apiClient,
getProxyBaseUrl,
@ -39,6 +40,7 @@ export interface TracesApi {
anyRecorded(): Promise<boolean>;
trace(traceId: string, traceRef?: string, cursor?: string | null): Promise<Trace>;
span(traceId: string, spanId: string, traceRef?: string): Promise<SpanDetail>;
spans(traceId: string, spanIds: string[], traceRef?: string): Promise<SpanDetail[]>;
spanError(
traceId: string,
spanId: string,
@ -75,6 +77,7 @@ export function liveTracesApi(accessToken: string): TracesApi {
},
trace: (traceId, traceRef, cursor) => agentTraceCall(accessToken, traceId, traceRef, cursor),
span: (traceId, spanId, traceRef) => agentTraceSpanCall(accessToken, traceId, spanId, traceRef),
spans: (traceId, spanIds, traceRef) => agentTraceSpansCall(accessToken, traceId, spanIds, traceRef),
spanError: (traceId, spanId, options) => agentTraceSpanErrorCall(accessToken, traceId, spanId, options),
};
}

View file

@ -21,7 +21,7 @@ export function useSpanDetail(accessToken: string, traceId: string, spanId: stri
queryKey: ["agentTraceSpan", traceId, traceRef, spanId, accessToken],
queryFn: () => traces.span(traceId, spanId as string, traceRef),
enabled: spanId !== null,
staleTime: Infinity,
staleTime: 30_000,
};
return useQuery(queryOptions);
}
@ -95,13 +95,24 @@ export function ContentTab({ accessToken, traceId, traceRef, span }: ContentTabP
)}
{detailQuery.isLoading && <div className={STATUS_TEXT}>Loading span…</div>}
{detailQuery.isError && <div className={STATUS_TEXT}>Could not load span: {detailQuery.error.message}</div>}
{detail &&
["input", "output"].map((field) =>
detail.attributes[`lens.content.${field}_status`] === "conflicting" ? (
<p key={field} role="status" className={STATUS_TEXT}>
Conflicting {field} content was recorded. Inspect the capture source spans.
</p>
) : null,
)}
{detail?.input ? (
<PayloadSection title="Input" raw={detail.input} content={detail.input_ui} span={span} role="input" />
) : null}
{detail?.output ? (
<PayloadSection title="Output" raw={detail.output} content={detail.output_ui} span={span} role="output" />
) : null}
{empty && span.status !== "error" && (
{detail?.attributes["lens.content.output_status"] === "recorded" && !detail.output && (
<p className={STATUS_TEXT}>The recorded output is empty.</p>
)}
{empty && span.status !== "error" && detail.attributes["lens.content.output_status"] !== "recorded" && (
<div className="py-12 text-center text-sm text-muted-foreground">No content recorded for this span.</div>
)}
</div>

View file

@ -1,3 +1,4 @@
import { ApiError } from "@/lib/http/client";
import { act, screen, waitFor, within } from "@testing-library/react";
import userEvent from "@testing-library/user-event";
import { beforeEach, describe, expect, it, vi } from "vitest";
@ -12,9 +13,10 @@ import research from "../../__fixtures__/research_trace.json";
vi.mock("../../../../networking", () => ({
agentTraceCall: vi.fn(),
agentTraceSpanCall: vi.fn(),
agentTraceSpansCall: vi.fn(),
getProxyBaseUrl: () => "http://proxy.test",
}));
import { agentTraceCall, agentTraceSpanCall } from "../../../../networking";
import { agentTraceCall, agentTraceSpanCall, agentTraceSpansCall } from "../../../../networking";
function RoutedRunView(props: Omit<ComponentProps<typeof RunView>, "selection">) {
const { selection } = useOpenTraceRouting();
@ -41,6 +43,13 @@ describe("TraceConversation", () => {
beforeEach(() => {
testQueryClient.clear();
vi.mocked(agentTraceCall).mockReset().mockResolvedValue(trace);
vi.mocked(agentTraceSpansCall)
.mockReset()
.mockImplementation(async (token, traceId, ids, traceRef) =>
Promise.all(
ids.map(async (id) => ({ ...(await agentTraceSpanCall(token, traceId, id, traceRef)), span_id: id })),
),
);
vi.mocked(agentTraceSpanCall)
.mockReset()
.mockImplementation(async (_token, _trace, id) => (id === "root" ? rootDetail : { ...toolDetail, span_id: id }));
@ -130,11 +139,11 @@ describe("TraceConversation", () => {
renderWithProviders(<TraceConversation trace={long} accessToken="test" onOpenStep={vi.fn()} />);
const more = await screen.findByRole("button", { name: "Load next 20 entries" });
await waitFor(() => expect(more).toBeEnabled());
expect(agentTraceSpanCall).toHaveBeenCalledTimes(20);
expect(agentTraceSpansCall).toHaveBeenCalledTimes(1);
expect(screen.queryByText("The release is ready")).not.toBeInTheDocument();
await user.click(more);
expect(await screen.findByText("The release is ready")).toBeVisible();
expect(agentTraceSpanCall).toHaveBeenCalledTimes(31);
expect(agentTraceSpansCall).toHaveBeenCalledTimes(2);
expect(screen.queryByRole("button", { name: /Load next/ })).not.toBeInTheDocument();
});
@ -191,7 +200,7 @@ describe("TraceConversation", () => {
);
await user.click(await screen.findByRole("tab", { name: "Conversation" }));
expect(await screen.findByText("2 entries shown")).toBeVisible();
expect(agentTraceSpanCall).toHaveBeenCalledTimes(20);
expect(agentTraceSpansCall).toHaveBeenCalledTimes(1);
await user.click(screen.getByRole("button", { name: "Load next 20 entries", exact: true }));
expect(await screen.findByText("22 entries shown")).toBeVisible();
expect(screen.getAllByRole("region", { name: /Conversation step parent tool/ })).toHaveLength(20);
@ -252,7 +261,7 @@ describe("TraceConversation", () => {
await user.click(screen.getByText("Subagent: Reviewer", { exact: true }));
expect(screen.getByText("Review finished")).toBeVisible();
expect(screen.queryByRole("button", { name: /entries in Reviewer/ })).not.toBeInTheDocument();
expect(agentTraceSpanCall).toHaveBeenCalledTimes(20);
expect(agentTraceSpansCall).toHaveBeenCalledTimes(1);
expect(loadMore).not.toHaveBeenCalled();
});
@ -310,7 +319,7 @@ describe("TraceConversation", () => {
expect(screen.getAllByRole("region", { name: /Conversation step child tool/ })).toHaveLength(25);
expect(screen.getByRole("button", { name: "Load next 20 entries", exact: true })).toBeEnabled();
expect(screen.queryByText("Loading conversation…")).not.toBeInTheDocument();
expect(agentTraceSpanCall).toHaveBeenCalledTimes(40);
expect(agentTraceSpansCall).toHaveBeenCalledTimes(2);
expect(agentTraceCall).toHaveBeenCalledTimes(paged ? 2 : 1);
expect(agentTraceCall).not.toHaveBeenCalledWith("test", trace.summary.trace_id, undefined, "unrelated-page");
});
@ -446,12 +455,34 @@ describe("TraceConversation", () => {
},
);
it("splits oversized content batches and still renders canonical results", async () => {
vi.mocked(agentTraceSpansCall).mockImplementation(async (_token, _trace, ids) => {
if (ids.length > 1) throw new ApiError("Content too large", 413, {});
return [ids[0] === "root" ? rootDetail : toolDetail];
});
renderWithProviders(<TraceConversation trace={trace} accessToken="test" onOpenStep={vi.fn()} />);
expect(await screen.findByText("The release is ready")).toBeVisible();
expect(vi.mocked(agentTraceSpansCall).mock.calls.map((call) => call[2])).toEqual([
["root", "tool"],
["root"],
["tool"],
]);
expect(agentTraceSpanCall).not.toHaveBeenCalled();
});
it("reports a missing bulk result instead of silently ending the conversation", async () => {
vi.mocked(agentTraceSpansCall).mockResolvedValue([rootDetail]);
renderWithProviders(<TraceConversation trace={trace} accessToken="test" onOpenStep={vi.fn()} />);
expect(await screen.findByRole("alert")).toHaveTextContent("Retry this batch");
expect(screen.queryByText("End of conversation")).not.toBeInTheDocument();
});
it("shows a missing step explicitly and lets the user retry it", async () => {
const user = userEvent.setup();
vi.mocked(agentTraceSpanCall).mockRejectedValueOnce(new Error("temporarily unavailable"));
renderWithProviders(<TraceConversation trace={trace} accessToken="test" onOpenStep={vi.fn()} />);
expect(await screen.findByRole("alert")).toHaveTextContent("Retry this step to continue the conversation");
await user.click(screen.getByRole("button", { name: "Retry step" }));
expect(await screen.findByRole("alert")).toHaveTextContent("Retry this batch to continue");
await user.click(screen.getByRole("button", { name: "Retry batch" }));
expect(await screen.findByText("Read the release notes")).toBeVisible();
expect(screen.queryByRole("alert")).not.toBeInTheDocument();
});
@ -476,10 +507,12 @@ describe("TraceConversation", () => {
const more = screen.getByRole("button", { name: "Load next 20 entries", exact: true });
await waitFor(() => expect(more).toBeEnabled());
await user.click(more);
expect(await screen.findByRole("alert")).toHaveTextContent("Could not load check 21");
expect(await screen.findByRole("alert")).toHaveTextContent(
"Could not load conversation content starting at check 19",
);
expect(retry).toHaveBeenCalledTimes(1);
expect(screen.queryByRole("button", { name: "Expand check 22 tool call" })).not.toBeInTheDocument();
await user.click(screen.getByRole("button", { name: "Retry step" }));
await user.click(screen.getByRole("button", { name: "Retry batch" }));
expect(await screen.findByText("40 entries shown")).toBeVisible();
expect(screen.getAllByRole("region", { name: /Conversation step check/ })).toHaveLength(39);
expect(screen.queryByRole("alert")).not.toBeInTheDocument();
@ -512,13 +545,13 @@ describe("TraceConversation", () => {
});
renderWithProviders(<TraceConversation trace={traced} accessToken="test" onOpenStep={vi.fn()} />);
expect(await screen.findByRole("alert")).toHaveTextContent("read_file");
expect(await screen.findByText("Checking the release")).toBeVisible();
expect(await screen.findByRole("alert")).toHaveTextContent("Retry this batch");
expect(screen.queryByText("Checking the release")).not.toBeInTheDocument();
expect(screen.queryByText("The release is ready")).not.toBeInTheDocument();
expect(screen.queryByText("End of conversation")).not.toBeInTheDocument();
expect(screen.getByText("2 entries shown")).toBeVisible();
expect(screen.queryByText("2 entries shown")).not.toBeInTheDocument();
await user.click(screen.getByRole("button", { name: "Retry step" }));
await user.click(screen.getByRole("button", { name: "Retry batch" }));
expect(await screen.findByText("The release is ready")).toBeVisible();
expect(screen.getAllByText("Checking the release")).toHaveLength(1);
expect(screen.getAllByText("Read the release notes")).toHaveLength(1);

View file

@ -141,9 +141,9 @@ function ConversationFeedback({
({ query, span }) =>
query.isError && (
<div key={span.span_id} role="alert" className="rounded-md border p-3 text-sm">
Could not load {span.name}. Retry this step to continue the conversation.
Could not load conversation content starting at {span.name}. Retry this batch to continue.
<Button variant="outline" size="sm" className="mt-2" onClick={() => query.refetch()}>
Retry step
Retry batch
</Button>
</div>
),
@ -300,8 +300,10 @@ function ConversationStep({
}
function ConversationTool({ item, onOpenStep }: { item: ConversationItem; onOpenStep: (id: string) => void }) {
const [open, setOpen] = useState(item.span.status === "error");
const failed = item.span.status === "error";
const failed = item.toolAttempt?.isError ?? item.span.status === "error";
const executionStatus = failed ? "Failed" : "Completed";
const [open, setOpen] = useState(failed);
const name = item.toolAttempt ? "Tool attempt" : item.span.name;
const summary = toolSummary(item.toolCall?.args);
return (
<div className={cn("min-w-0 overflow-hidden rounded-md border", failed && "border-destructive/40")}>
@ -309,7 +311,7 @@ function ConversationTool({ item, onOpenStep }: { item: ConversationItem; onOpen
<button
type="button"
aria-expanded={open}
aria-label={`${open ? "Collapse" : "Expand"} ${item.span.name} tool call`}
aria-label={`${open ? "Collapse" : "Expand"} ${name} tool call`}
onClick={() => setOpen((value) => !value)}
className="flex min-w-0 flex-1 items-center gap-2 px-3 py-2.5 text-left text-sm hover:bg-muted/40"
>
@ -318,7 +320,8 @@ function ConversationTool({ item, onOpenStep }: { item: ConversationItem; onOpen
/>
<Wrench className="size-3.5 shrink-0 text-muted-foreground" />
<span className="min-w-0 flex-1">
<span className="block truncate font-medium">{item.span.name}</span>
<span className="block truncate font-medium">{name}</span>
{item.toolAttempt && <span className="block text-xs text-muted-foreground">Execution not recorded</span>}
{summary && !open && (
<span className="mt-1 block truncate font-mono text-xs text-muted-foreground" title={summary}>
{summary}
@ -326,7 +329,7 @@ function ConversationTool({ item, onOpenStep }: { item: ConversationItem; onOpen
)}
</span>
<span className={cn("ml-auto shrink-0 text-xs", failed ? "text-destructive" : "text-muted-foreground")}>
{failed ? "Failed" : "Completed"}
{item.toolAttempt ? "Result recorded" : executionStatus}
</span>
</button>
<Button
@ -357,7 +360,7 @@ function ConversationTool({ item, onOpenStep }: { item: ConversationItem; onOpen
{item.toolResult ? (
<ToolOutput result={item.toolResult} failed={failed} />
) : (
<p className="text-sm text-muted-foreground">No result content recorded.</p>
<p className="text-sm text-muted-foreground">{item.contentWarning ?? "No result content recorded."}</p>
)}
</div>
)}

View file

@ -564,7 +564,37 @@ describe("coding sessions", () => {
expect(items.find((item) => item.span.span_id === response.span_id)?.agentId).toBe("agent-tool");
});
it("fills native tool arguments and missing results from logs without duplicating calls or replacing recorded output", () => {
it("shows unexecuted tool results once as attempts without inventing executed spans", () => {
const body = {
...root,
span_id: "body",
framework: "claude-code",
type: "framework" as const,
name: "claude_code.api_request_body",
};
const duplicate = { ...body, span_id: "retry" };
const output = {
...detail("body", "", ""),
attributes: {
"lens.content.unexecuted_tool_results": JSON.stringify([
{ id: "denied", content: "Tool permission denied", is_error: true },
]),
},
};
const details = new Map([
[body.span_id, output],
[duplicate.span_id, { ...output, span_id: duplicate.span_id }],
]);
const items = buildConversation([body, duplicate], details, true);
expect(items).toHaveLength(1);
expect(items[0]).toMatchObject({
toolAttempt: { isError: true },
toolResult: "Tool permission denied",
span: { type: "framework" },
});
});
it("uses canonical native tool content without browser supplements overriding it", () => {
const tool = {
...root,
span_id: "tool",
@ -583,8 +613,8 @@ describe("coding sessions", () => {
};
const body = { ...log, span_id: "body", name: "claude_code.api_request_body", start_offset_ms: 3 };
const toolDetail = {
...detail(tool.span_id, { command: "exit 3" }, ""),
output: "",
...detail(tool.span_id, { command: "exit 3", description: "Expected failure" }, ""),
output: "Canonical stdout",
attributes: { "gen_ai.tool.call.id": "call-1" },
};
const details = new Map([
@ -612,7 +642,7 @@ describe("coding sessions", () => {
expect(items.filter((item) => item.toolCall)).toHaveLength(1);
expect(items.find((item) => item.toolCall)).toMatchObject({
toolCall: { args: { command: "exit 3", description: "Expected failure" } },
toolResult: "Expected stdout",
toolResult: "Canonical stdout",
});
details.set(tool.span_id, { ...toolDetail, output: "Recorded output" });
expect(buildConversation(spans, details, true).find((item) => item.toolCall)?.toolResult).toBe("Recorded output");

View file

@ -2,7 +2,7 @@ import { groupBy, orderBy } from "es-toolkit";
import type { Span, SpanDetail, TraceMessage, TraceToolCall, UIContent } from "../../types";
import { isFrameworkSpan, parseAssistantSummary, parseMessages, prettyPayload } from "../../utils";
import { toTraceMessage, toolInput } from "../content/payload";
import { claudeCaptureWarnings, claudeToolDetails, isClaudeSupplement } from "./nativeClaude";
import { claudeCaptureWarnings, isClaudeSupplement, unexecutedToolResults } from "./nativeClaude";
export const CONVERSATION_PAGE_SIZE = 20;
@ -121,6 +121,8 @@ export interface ConversationItem {
messages: TraceMessage[];
toolCall?: TraceToolCall;
toolResult?: string;
toolAttempt?: { isError: boolean };
contentWarning?: string;
agentId?: string;
agentName?: string;
branchId?: string;
@ -156,7 +158,13 @@ function toolItem(
detail.output_ui?.kind === "text"
? detail.output_ui.text
: prettyPayload(detail.output) || contentText(detail.output, detail.output_ui);
return { id: span.span_id, span, messages: [], toolCall: call, toolResult: result };
const status = detail.attributes["lens.content.output_status"];
const contentWarning = (() => {
if (status === "conflicting") return "Conflicting tool outputs were recorded. Inspect the capture source spans.";
if (status === "recorded" && !result) return "The recorded result is empty.";
return undefined;
})();
return { id: span.span_id, span, messages: [], toolCall: call, toolResult: result, contentWarning };
}
interface ConversationEvent {
@ -297,17 +305,32 @@ export function buildConversation(
complete: boolean,
pendingBranches?: ReadonlySet<string>,
): ConversationItem[] {
const details = claudeToolDetails(spans, recordedDetails);
const details = recordedDetails;
const byId = new Map(spans.map((span) => [span.span_id, span]));
const histories = new Map<string, TraceMessage[]>();
const completedOutputs = new Map<string, TraceMessage[]>();
const pendingCalls = new Map<string, TraceToolCall[]>();
const items: ConversationItem[] = [];
const attempts = new Set<string>();
const events = conversationEvents(conversationSteps(spans), byId, details, { complete, pendingBranches });
const branch = (span: Span): string => conversationBranch(span, byId);
for (const event of events) {
const { span } = event;
if (isClaudeSupplement(span)) continue;
if (isClaudeSupplement(span)) {
for (const result of unexecutedToolResults(details.get(span.span_id)!)) {
if (attempts.has(result.id)) continue;
attempts.add(result.id);
items.push({
id: `${span.span_id}-${result.id}`,
span,
messages: [],
toolCall: { name: "Tool attempt", args: { tool_call_id: result.id } },
toolAttempt: { isError: result.isError },
toolResult: result.content,
});
}
continue;
}
const detail = details.get(span.span_id)!;
if (["generate_session_title", "prompt_suggestion"].includes(detail.attributes["query_source_safe"])) continue;
const key = branch(span);

View file

@ -20,45 +20,6 @@ function payload(detail: SpanDetail): Record<string, unknown> | undefined {
}
}
export function claudeToolDetails(
spans: readonly Span[],
details: ReadonlyMap<string, SpanDetail>,
): ReadonlyMap<string, SpanDetail> {
const inputs = new Map<string, SpanDetail>();
const outputs = new Map<string, string>();
for (const span of spans.filter(isClaudeSupplement)) {
const detail = details.get(span.span_id);
if (!detail) continue;
const id = detail.attributes["tool_use_id"];
if (span.name === "claude_code.tool_result" && id && detail.input) inputs.set(id, detail);
const results = payload(detail)?.tool_results;
if (!Array.isArray(results)) continue;
for (const value of results) {
const result = object(value);
if (typeof result?.id === "string" && typeof result.content === "string") {
outputs.set(result.id, result.content);
}
}
}
return new Map(
[...details].map(([id, detail]) => {
const callId = detail.attributes["gen_ai.tool.call.id"] || detail.attributes["tool_use_id"];
const input = inputs.get(callId);
const output = outputs.get(callId);
return [
id,
{
...detail,
...(input ? { input: input.input, input_ui: input.input_ui } : {}),
...(!detail.output && output !== undefined
? { output, output_ui: { kind: "text" as const, text: output } }
: {}),
},
];
}),
);
}
export function claudeCaptureWarnings(details: ReadonlyMap<string, SpanDetail>): string[] {
return [...details.values()].flatMap((detail) => {
if (detail.attributes["event.name"] !== "api_request_body") return [];
@ -66,3 +27,18 @@ export function claudeCaptureWarnings(details: ReadonlyMap<string, SpanDetail>):
return typeof warning === "string" ? [warning] : [];
});
}
export function unexecutedToolResults(detail: SpanDetail): { id: string; content: string; isError: boolean }[] {
try {
const results: unknown = JSON.parse(detail.attributes["lens.content.unexecuted_tool_results"] ?? "[]");
if (!Array.isArray(results)) return [];
return results.flatMap((value) => {
const result = object(value);
return typeof result?.id === "string" && typeof result.content === "string"
? [{ id: result.id, content: result.content, isError: result.is_error === true }]
: [];
});
} catch {
return [];
}
}

View file

@ -1,45 +1,71 @@
import { useQueries, useQueryClient } from "@tanstack/react-query";
import { useCallback, useMemo, useState } from "react";
import { useTracesApi } from "../../api";
import { ApiError } from "@/lib/http/client";
import { useTracesApi, type TracesApi } from "../../api";
import type { Span, SpanDetail, Trace } from "../../types";
import { CONVERSATION_PAGE_SIZE, conversationSteps } from "./conversation";
async function readBatch(traces: TracesApi, traceId: string, ids: string[], traceRef: string): Promise<SpanDetail[]> {
try {
return await traces.spans(traceId, ids, traceRef);
} catch (error) {
if (!(error instanceof ApiError) || error.status !== 413 || ids.length <= 1) throw error;
const middle = Math.ceil(ids.length / 2);
const first = await readBatch(traces, traceId, ids.slice(0, middle), traceRef);
return [...first, ...(await readBatch(traces, traceId, ids.slice(middle), traceRef))];
}
}
export function useConversationDetails(trace: Trace, accessToken: string) {
const traces = useTracesApi(accessToken);
const queryClient = useQueryClient();
const [limit, setLimit] = useState(CONVERSATION_PAGE_SIZE);
const steps = useMemo(() => conversationSteps(trace.spans), [trace.spans]);
const visible = steps.slice(0, limit);
const { trace_id: traceId, trace_ref: traceRef } = trace.summary;
const spanQuery = useCallback(
(span: Span) => ({
queryKey: ["agentTraceSpan", traceId, traceRef, span.span_id, accessToken],
queryFn: (): Promise<SpanDetail> => traces.span(traceId, span.span_id, traceRef),
staleTime: Infinity,
retry: false as const,
retryOnMount: false,
}),
[traceId, traceRef, accessToken, traces],
const { trace_id: traceId, trace_ref: traceRef, span_count: spanCount } = trace.summary;
const batchQuery = useCallback(
(batch: Span[]) => {
const ids = batch.map((span) => span.span_id);
return {
queryKey: ["agentTraceContents", traceId, traceRef, ids, spanCount, accessToken],
queryFn: async (): Promise<SpanDetail[]> => {
const details = await readBatch(traces, traceId, ids, traceRef);
const returned = new Set(details.map((detail) => detail.span_id));
if (ids.some((id) => !returned.has(id))) throw new Error("Some requested spans are unavailable");
for (const detail of details) {
queryClient.setQueryData(["agentTraceSpan", traceId, traceRef, detail.span_id, accessToken], detail);
}
return details;
},
staleTime: 30_000,
retry: false as const,
retryOnMount: false,
};
},
[traceId, traceRef, spanCount, accessToken, traces, queryClient],
);
const queries = useQueries({ queries: visible.map(spanQuery) });
const batches = Array.from({ length: Math.ceil(visible.length / CONVERSATION_PAGE_SIZE) }, (_, index) =>
visible.slice(index * CONVERSATION_PAGE_SIZE, (index + 1) * CONVERSATION_PAGE_SIZE),
);
const queries = useQueries({ queries: batches.map(batchQuery) });
const unresolvedIndex = queries.findIndex((query) => !query.isSuccess);
const loadedCount = unresolvedIndex < 0 ? queries.length : unresolvedIndex;
const details = new Map(
queries.slice(0, loadedCount).map((query, index) => [visible[index].span_id, query.data!] as const),
queries.slice(0, loadedCount).flatMap((query) => query.data!.map((detail) => [detail.span_id, detail] as const)),
);
const loadMore = useCallback(
async (signal: AbortSignal) => {
const nextLimit = limit + CONVERSATION_PAGE_SIZE;
const batch = steps.slice(limit, nextLimit);
await Promise.all(batch.map((span) => queryClient.prefetchQuery(spanQuery(span))));
await queryClient.prefetchQuery(batchQuery(batch));
if (!signal.aborted) setLimit(nextLimit);
},
[limit, steps, queryClient, spanQuery],
[limit, steps, queryClient, batchQuery],
);
return {
details,
entries: visible.map((span, index) => ({ span, query: queries[index] })),
complete: loadedCount === steps.length,
entries: batches.map((batch, index) => ({ span: batch[0], query: queries[index] })),
complete: loadedCount === batches.length && visible.length === steps.length,
loading: queries.some((query) => query.isPending),
failed: queries.some((query) => query.isError),
hasMore: visible.length < steps.length,

View file

@ -10,6 +10,7 @@ export type SpanQuery = NonNullable<paths["/v1/traces/{trace_id}/spans/{span_id}
export type SpanErrorQuery = NonNullable<
paths["/v1/traces/{trace_id}/spans/{span_id}/error"]["get"]["parameters"]["query"]
>;
export type SpanDetailsBody = components["schemas"]["TraceSpanDetailsRequest"];
export type TraceQueryBody = components["schemas"]["TraceQueryRequest"];
export type TraceFindingsRequest = components["schemas"]["TraceFindingsRequest"];
export type TraceFindingCount = components["schemas"]["TraceFindingCount"];

View file

@ -119,6 +119,7 @@ import type { VectorStoreIndex } from "@/app/(dashboard)/vector-stores/_componen
import type { RoutingDecision } from "./logs/detail/RoutingDecisionCard";
import type {
SpanDetail,
SpanDetailsBody,
SpanErrorPage,
SpanErrorQuery,
SpanQuery,
@ -2006,6 +2007,17 @@ export const agentTraceSpanCall = async (
query: { trace_ref: traceRef || undefined } satisfies SpanQuery,
});
export const agentTraceSpansCall = async (
accessToken: string,
traceId: string,
spanIds: string[],
traceRef?: string,
): Promise<SpanDetail[]> =>
apiClient.post<SpanDetail[]>(`/v1/traces/${encodeURIComponent(traceId)}/spans`, {
accessToken,
body: { trace_ref: traceRef || "", span_ids: spanIds } satisfies SpanDetailsBody,
});
export const agentTraceSpanErrorCall = async (
accessToken: string,
traceId: string,

View file

@ -22500,6 +22500,23 @@ export interface paths {
patch?: never;
trace?: never;
};
"/v1/traces/{trace_id}/spans": {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
get?: never;
put?: never;
/** Get Agent Trace Spans */
post: operations["get_agent_trace_spans_v1_traces__trace_id__spans_post"];
delete?: never;
options?: never;
head?: never;
patch?: never;
trace?: never;
};
"/v1/traces/{trace_id}/spans/{span_id}": {
parameters: {
query?: never;
@ -47376,6 +47393,16 @@ export interface components {
[key: string]: components["schemas"]["JsonValue"];
}[];
};
/** TraceSpanDetailsRequest */
TraceSpanDetailsRequest: {
/** Span Ids */
span_ids: string[];
/**
* Trace Ref
* @default
*/
trace_ref: string;
};
/** TraceSummary */
TraceSummary: {
/** Agent Count */
@ -80445,6 +80472,41 @@ export interface operations {
};
};
};
get_agent_trace_spans_v1_traces__trace_id__spans_post: {
parameters: {
query?: never;
header?: never;
path: {
trace_id: string;
};
cookie?: never;
};
requestBody: {
content: {
"application/json": components["schemas"]["TraceSpanDetailsRequest"];
};
};
responses: {
/** @description Successful Response */
200: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["SpanDetail"][];
};
};
/** @description Validation Error */
422: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["HTTPValidationError"];
};
};
};
};
get_agent_trace_span_v1_traces__trace_id__spans__span_id__get: {
parameters: {
query?: {