fix(copilot): dedup otel records across export files

The trace context, coverage sets, and seen keys were local to
_parse_file, but OTEL exporters write spans and logs to different files:
one call logged as a chat span in file A and an inference log in file B
was counted twice. Collect every file's candidates first, then dedup and
emit at _parse scope, preserving the chat span > inference log >
agent-turn log > agent-summary span priority. The trace/span/response
ids in the dedup keys are globally unique, so widening the scope never
merges distinct calls.

Author: Gheat1

CLAUDE.md

@@ -200,7 +200,8 @@ Three logical layers (the class names below live in the files above — `Store`
   `--copilot-dir`); with it off the source never appears. Export carries tokens but **no
   cost** → token-only backend (**`records_cost=False`**, $0/list-price estimate, same
   nudges). OTEL follows the **GenAI semantic conventions** where one call is logged up to
-  four ways, so `_parse_file` **dedups per file** by trace/response id, keeping the
+  four ways — spans and logs land in *different* files — so `_parse` **dedups across all
+  files** by trace/response id, keeping the
   highest-fidelity record (chat span > inference log > agent-turn log > agent-summary span;
   `_classify`/`_emit`). OpenAI-style tokens (input includes cache read; `cache_write` from
   `cache_creation`; reasoning folded into output; a `total_tokens`-only record back-fills).

src/opentab/stores/copilot.py

@@ -308,7 +308,8 @@ def cache_inputs(self) -> list[str]:
     def _files(self) -> list[str]:
         files = glob.glob(os.path.join(self.root_dir, "**", "*.jsonl"), recursive=True)
         # Add the env-var export file, but only if it isn't already one of the globbed
-        # ones -- dedup is per file, so reading the same file twice would double-count.
+        # ones -- the fallback dedup keys are index-based, so reading the same file
+        # twice would double-count records that lack trace/span ids.
         if self._extra_file and os.path.isfile(self._extra_file):
             seen = {os.path.realpath(f) for f in files}
             if os.path.realpath(self._extra_file) not in seen:
@@ -319,40 +320,37 @@ def _parse(self) -> dict[str, dict]:
         if self._sessions is not None:
             return self._sessions
         sessions: dict[str, dict] = {}
-        for path, text in read_files_parallel(self._files()):
-            self._parse_file(path, text.split("\n"), sessions)
-        for sid, s in sessions.items():
-            self._finalize(sid, s)
-        # Drop sessions with no recorded usage (non-LLM-only files): they would only add
-        # $0 / 0-token rows to a spend browser.
-        self._sessions = {sid: s for sid, s in sessions.items() if s["model_rows"]}
-        return self._sessions
-
-    def _parse_file(self, path: str, lines: list[str], sessions: dict[str, dict]) -> None:
-        records = [
-            o
-            for line in lines
-            if '"attributes"' in line
-            for o in (self._loads(line),)
-            if isinstance(o, dict) and isinstance(o.get("attributes"), dict)
+        # OTEL exporters write spans and logs to DIFFERENT files, so trace context, the
+        # dedup coverage sets, and the seen keys must span all files: one call logged as
+        # a chat span in file A and an inference log in file B is still one call. Two
+        # passes -- collect every file's candidates first, then emit.
+        per_file = [
+            (path, self._file_records(text.split("\n")))
+            for path, text in read_files_parallel(self._files())
         ]
-        # Per-file trace context: a model / session id seen anywhere on a trace fills in
-        # for records on that trace that omit it.
+        # Cross-file trace context: a model / session id seen anywhere on a trace fills
+        # in for records on that trace that omit it.
         trace_ctx: dict[str, dict] = {}
-        for rec in records:
-            tid = self._trace_id(rec)
-            if not tid:
-                continue
-            attrs = rec["attributes"]
-            ctx = trace_ctx.setdefault(tid, {"model": None, "session": None, "prio": -1})
-            if ctx["model"] is None:
-                ctx["model"] = self._attr_str(attrs, *self._MODEL_ATTRS)
-            sid, prio = self._session_attr(attrs)
-            if sid is not None and prio > ctx["prio"]:
-                ctx["session"], ctx["prio"] = sid, prio
-        candidates = [
-            c for i, rec in enumerate(records) if (c := self._to_candidate(rec, i, trace_ctx, path))
-        ]
+        for _path, records in per_file:
+            for rec in records:
+                tid = self._trace_id(rec)
+                if not tid:
+                    continue
+                attrs = rec["attributes"]
+                ctx = trace_ctx.setdefault(tid, {"model": None, "session": None, "prio": -1})
+                if ctx["model"] is None:
+                    ctx["model"] = self._attr_str(attrs, *self._MODEL_ATTRS)
+                sid, prio = self._session_attr(attrs)
+                if sid is not None and prio > ctx["prio"]:
+                    ctx["session"], ctx["prio"] = sid, prio
+        candidates: list[dict] = []
+        idx = 0  # global running index keeps the fallback dedup keys unique across files
+        for path, records in per_file:
+            for rec in records:
+                c = self._to_candidate(rec, idx, trace_ctx, path)
+                idx += 1
+                if c is not None:
+                    candidates.append(c)
         # Dedup: a chat span is the source of truth; an inference log is dropped when a
         # chat covers its trace/response, an agent-turn log when a chat or inference does,
         # 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]) ->
                 continue
             seen.add(c["dedup_key"])
             self._fold(sessions, c)
+        for sid, s in sessions.items():
+            self._finalize(sid, s)
+        # Drop sessions with no recorded usage (non-LLM-only files): they would only add
+        # $0 / 0-token rows to a spend browser.
+        self._sessions = {sid: s for sid, s in sessions.items() if s["model_rows"]}
+        return self._sessions
+
+    @classmethod
+    def _file_records(cls, lines: list[str]) -> list[dict]:
+        return [
+            o
+            for line in lines
+            if '"attributes"' in line
+            for o in (cls._loads(line),)
+            if isinstance(o, dict) and isinstance(o.get("attributes"), dict)
+        ]
 
     @staticmethod
     def _loads(line: str):

test_opentab.py

@@ -6891,6 +6891,38 @@ def test_copilot_store_dedupes_redundant_records_keeping_chat_span():
         assert rows[0]["tokens_total"] == 70
 
 
+def test_copilot_store_dedupes_span_and_log_split_across_files():
+    with tempfile.TemporaryDirectory() as tmp:
+        otel = os.path.join(tmp, ".copilot", "otel")
+        # OTEL exporters write spans and logs to DIFFERENT files: the same call appears
+        # as a chat span in spans.jsonl and an inference log in logs.jsonl. It must
+        # count once (the chat span's 60/10), never twice.
+        chat = _otel_chat(
+            "conv-x", "gpt-5.4", 60, 10, trace="trace-x", span="chat-1", resp="resp-x"
+        )
+        inference = {
+            "traceId": "trace-x",
+            "hrTime": [1775934263, 0],
+            "_body": "GenAI inference: gpt-5.4",
+            "attributes": {
+                "event.name": "gen_ai.client.inference.operation.details",
+                "gen_ai.response.model": "gpt-5.4",
+                "gen_ai.conversation.id": "conv-x",
+                "gen_ai.response.id": "resp-x",
+                "gen_ai.usage.input_tokens": 80,
+                "gen_ai.usage.output_tokens": 20,
+            },
+        }
+        _write_otel(otel, [chat], name="spans.jsonl")
+        _write_otel(otel, [inference], name="logs.jsonl")
+        store = ot.CopilotStore(otel, _copilot_args(otel))
+        rows = [r for r in store.model_breakdown() if r["root_id"] == "conv-x"]
+        assert len(rows) == 1
+        assert rows[0]["runs"] == 1  # the log in the other file was suppressed
+        assert rows[0]["unpriced_input"] == 60 and rows[0]["unpriced_output"] == 10
+        assert rows[0]["tokens_total"] == 70
+
+
 def test_copilot_store_enriches_cwd_and_title_from_session_store_db():
     with tempfile.TemporaryDirectory() as tmp:
         copilot = os.path.join(tmp, ".copilot")