@@ -308,7 +308,8 @@ def cache_inputs(self) -> list[str]:
308308 def _files (self ) -> list [str ]:
309309 files = glob .glob (os .path .join (self .root_dir , "**" , "*.jsonl" ), recursive = True )
310310 # Add the env-var export file, but only if it isn't already one of the globbed
311- # ones -- dedup is per file, so reading the same file twice would double-count.
311+ # ones -- the fallback dedup keys are index-based, so reading the same file
312+ # twice would double-count records that lack trace/span ids.
312313 if self ._extra_file and os .path .isfile (self ._extra_file ):
313314 seen = {os .path .realpath (f ) for f in files }
314315 if os .path .realpath (self ._extra_file ) not in seen :
@@ -319,40 +320,37 @@ def _parse(self) -> dict[str, dict]:
319320 if self ._sessions is not None :
320321 return self ._sessions
321322 sessions : dict [str , dict ] = {}
322- for path , text in read_files_parallel (self ._files ()):
323- self ._parse_file (path , text .split ("\n " ), sessions )
324- for sid , s in sessions .items ():
325- self ._finalize (sid , s )
326- # Drop sessions with no recorded usage (non-LLM-only files): they would only add
327- # $0 / 0-token rows to a spend browser.
328- self ._sessions = {sid : s for sid , s in sessions .items () if s ["model_rows" ]}
329- return self ._sessions
330-
331- def _parse_file (self , path : str , lines : list [str ], sessions : dict [str , dict ]) -> None :
332- records = [
333- o
334- for line in lines
335- if '"attributes"' in line
336- for o in (self ._loads (line ),)
337- if isinstance (o , dict ) and isinstance (o .get ("attributes" ), dict )
323+ # OTEL exporters write spans and logs to DIFFERENT files, so trace context, the
324+ # dedup coverage sets, and the seen keys must span all files: one call logged as
325+ # a chat span in file A and an inference log in file B is still one call. Two
326+ # passes -- collect every file's candidates first, then emit.
327+ per_file = [
328+ (path , self ._file_records (text .split ("\n " )))
329+ for path , text in read_files_parallel (self ._files ())
338330 ]
339- # Per -file trace context: a model / session id seen anywhere on a trace fills in
340- # for records on that trace that omit it.
331+ # Cross -file trace context: a model / session id seen anywhere on a trace fills
332+ # in for records on that trace that omit it.
341333 trace_ctx : dict [str , dict ] = {}
342- for rec in records :
343- tid = self ._trace_id (rec )
344- if not tid :
345- continue
346- attrs = rec ["attributes" ]
347- ctx = trace_ctx .setdefault (tid , {"model" : None , "session" : None , "prio" : - 1 })
348- if ctx ["model" ] is None :
349- ctx ["model" ] = self ._attr_str (attrs , * self ._MODEL_ATTRS )
350- sid , prio = self ._session_attr (attrs )
351- if sid is not None and prio > ctx ["prio" ]:
352- ctx ["session" ], ctx ["prio" ] = sid , prio
353- candidates = [
354- c for i , rec in enumerate (records ) if (c := self ._to_candidate (rec , i , trace_ctx , path ))
355- ]
334+ for _path , records in per_file :
335+ for rec in records :
336+ tid = self ._trace_id (rec )
337+ if not tid :
338+ continue
339+ attrs = rec ["attributes" ]
340+ ctx = trace_ctx .setdefault (tid , {"model" : None , "session" : None , "prio" : - 1 })
341+ if ctx ["model" ] is None :
342+ ctx ["model" ] = self ._attr_str (attrs , * self ._MODEL_ATTRS )
343+ sid , prio = self ._session_attr (attrs )
344+ if sid is not None and prio > ctx ["prio" ]:
345+ ctx ["session" ], ctx ["prio" ] = sid , prio
346+ candidates : list [dict ] = []
347+ idx = 0 # global running index keeps the fallback dedup keys unique across files
348+ for path , records in per_file :
349+ for rec in records :
350+ c = self ._to_candidate (rec , idx , trace_ctx , path )
351+ idx += 1
352+ if c is not None :
353+ candidates .append (c )
356354 # Dedup: a chat span is the source of truth; an inference log is dropped when a
357355 # chat covers its trace/response, an agent-turn log when a chat or inference does,
358356 # an agent-summary span when any of the three do.
@@ -371,6 +369,22 @@ def _parse_file(self, path: str, lines: list[str], sessions: dict[str, dict]) ->
371369 continue
372370 seen .add (c ["dedup_key" ])
373371 self ._fold (sessions , c )
372+ for sid , s in sessions .items ():
373+ self ._finalize (sid , s )
374+ # Drop sessions with no recorded usage (non-LLM-only files): they would only add
375+ # $0 / 0-token rows to a spend browser.
376+ self ._sessions = {sid : s for sid , s in sessions .items () if s ["model_rows" ]}
377+ return self ._sessions
378+
379+ @classmethod
380+ def _file_records (cls , lines : list [str ]) -> list [dict ]:
381+ return [
382+ o
383+ for line in lines
384+ if '"attributes"' in line
385+ for o in (cls ._loads (line ),)
386+ if isinstance (o , dict ) and isinstance (o .get ("attributes" ), dict )
387+ ]
374388
375389 @staticmethod
376390 def _loads (line : str ):
0 commit comments