tokentab provider pipeline

45def collect() -> list[UsageRecord]:
46    root = _config_dir() / "projects"
47    if not root.is_dir():
48        return []
49
50    records: list[UsageRecord] = []
51    seen: set[str] = set()
52
53    for file in walk_files(root, lambda n: n.endswith(".jsonl")):
54        project = pretty_project(_derive_project(file))
55
56        try:
57            text = file.read_text(encoding="utf-8")
58        except OSError:
59            continue
60
61        lines = text.split("\n")
62
63        # first pass: first user message (for activity) + tools used this session
64        first_user_text = ""
65        tool_names: list[str] = []
66        for line in lines:
67            if not line.strip():
68                continue
69            obj = try_json(line)
70            msg = obj.get("message") if isinstance(obj, dict) else None
71            if not msg:
72                continue
73            if msg.get("role") == "user" and not first_user_text:
74                first_user_text = _extract_text(msg.get("content"))
75            for block in msg.get("content") or []:
76                if block.get("type") == "tool_use" and block.get("name"):
77                    tool_names.append(block["name"])
78        activity = classify(tool_names, first_user_text)
79
80        # second pass: token usage from assistant turns
81        for line in lines:
82            if not line.strip():
83                continue
84            obj = try_json(line)
85            msg = obj.get("message") if isinstance(obj, dict) else None
86            usage = msg.get("usage") if msg else None
87            if not usage:
88                continue
89
90            rec_id = msg.get("id") or f"{file}:{len(records)}"
91            if rec_id in seen:
92                continue
93            seen.add(rec_id)
94
95            records.append(
96                UsageRecord(
97                    provider=NAME,
98                    model=msg.get("model") or "unknown",
99                    project=project,
100                    timestamp=to_date(obj.get("timestamp")) or EPOCH,
101                    activity=activity,
102                    id=rec_id,
103                    tokens=TokenCounts(
104                        input=usage.get("input_tokens", 0),
105                        output=usage.get("output_tokens", 0),
106                        cache_read=usage.get("cache_read_input_tokens", 0),
107                        cache_write=usage.get("cache_creation_input_tokens", 0),
108                    ),
109                )
110            )

38def collect() -> list[UsageRecord]:
39    roots = [_config_dir() / "sessions", _config_dir() / "archived_sessions"]
40    records: list[UsageRecord] = []
41
42    for root in roots:
43        if not root.is_dir():
44            continue
45
46        for file in walk_files(root, lambda n: n.startswith("rollout-") and n.endswith(".jsonl")):
47            try:
48                text = file.read_text(encoding="utf-8")
49            except OSError:
50                continue
51
52            lines = text.split("\n")
53            project = "unknown"
54            model = "unknown"
55            first_user_text = ""
56            tool_names: list[str] = []
57
58            # context pass
59            for line in lines:
60                if not line.strip():
61                    continue
62                ev = try_json(line)
63                if not isinstance(ev, dict):
64                    continue
65                if ev.get("cwd"):
66                    project = pretty_project(ev["cwd"])
67                if ev.get("model"):
68                    model = ev["model"]
69                payload = ev.get("payload") or {}
70                if ev.get("type") == "function_call" or payload.get("type") == "function_call":
71                    name = payload.get("name") or (ev.get("info") or {}).get("name")
72                    if name:
73                        tool_names.append(str(name))
74                if (ev.get("type") == "message" or payload.get("role") == "user") and not first_user_text:
75                    content = payload.get("content") or (ev.get("info") or {}).get("content")
76                    if isinstance(content, str):
77                        first_user_text = content[:500]
78
79            activity = classify(tool_names, first_user_text)
80
81            # usage pass — cumulative totals diffed into per-call deltas
82            prev_in = prev_out = prev_cache = 0
83            i = 0
84            for line in lines:
85                if not line.strip():
86                    continue
87                ev = try_json(line)
88                if not isinstance(ev, dict):
89                    continue
90                tc = _find_token_count(ev)
91                if not tc:
92                    continue
93
94                tot_in = tc.get("input_tokens", tc.get("input", 0)) or 0
95                tot_out = tc.get("output_tokens", tc.get("output", 0)) or 0
96                tot_cache = tc.get("cached_input_tokens", tc.get("cache_read", 0)) or 0
97
98                d_in = max(0, tot_in - prev_in)
99                d_out = max(0, tot_out - prev_out)
100                d_cache = max(0, tot_cache - prev_cache)
101                prev_in, prev_out, prev_cache = tot_in, tot_out, tot_cache
102
103                if d_in + d_out + d_cache == 0:
104                    continue
105
106                records.append(
107                    UsageRecord(
108                        provider=NAME,
109                        model=model,
110                        project=project,
111                        timestamp=to_date(ev.get("timestamp")) or EPOCH,
112                        activity=activity,
113                        id=f"{file}:{i}",
114                        tokens=TokenCounts(input=d_in, output=d_out, cache_read=d_cache, cache_write=0),
115                    )
116                )
117                i += 1

30def collect() -> list[UsageRecord]:
31    root = home(".gemini", "tmp")
32    if not root.is_dir():
33        return []
34
35    records: list[UsageRecord] = []
36
37    for file in walk_files(root, lambda n: n.startswith("session-") and n.endswith(".json")):
38        try:
39            raw = file.read_text(encoding="utf-8")
40        except OSError:
41            continue
42        doc = try_json(raw)
43        messages = doc.get("messages") if isinstance(doc, dict) else None
44        if not messages:
45            continue
46
47        project = pretty_project(_derive_project(file))
48
49        first_user_text = ""
50        tool_names: list[str] = []
51        for m in messages:
52            if m.get("role") == "user" and not first_user_text:
53                first_user_text = (m.get("text") or "")[:500]
54            for p in m.get("parts") or []:
55                fc = p.get("functionCall")
56                if fc and fc.get("name"):
57                    tool_names.append(fc["name"])
58        activity = classify(tool_names, first_user_text)
59
60        for i, m in enumerate(messages):
61            t = m.get("tokens")
62            if not t:
63                continue
64            cached = t.get("cached", 0) or 0
65            raw_input = t.get("input", 0) or 0
66            fresh_input = max(0, raw_input - cached)  # input is inclusive of cache
67            output = t.get("output", 0) or 0
68
69            if fresh_input + output + cached == 0:
70                continue
71
72            records.append(
73                UsageRecord(
74                    provider=NAME,
75                    model=m.get("model") or "gemini-2.5-pro",
76                    project=project,
77                    timestamp=to_date(m.get("timestamp")) or EPOCH,
78                    activity=activity,
79                    id=f"{file}:{i}",
80                    tokens=TokenCounts(input=fresh_input, output=output, cache_read=cached, cache_write=0),
81                )
82            )
83

16def try_json(text: str):
17    """Parse one JSON value, returning None instead of raising."""
18    try:
19        return json.loads(text)
20    except (json.JSONDecodeError, ValueError):
21        return None
22
23
24def walk_files(root: Path, test) -> Iterator[Path]:
25    """Yield every file under root whose name passes test(name).
26
27    Skips anything unreadable rather than blowing up the whole run.
28    """
29    if not root.is_dir():
30        return
31    for dirpath, _dirs, files in os.walk(root):
32        for name in files:
33            if test(name):
34                yield Path(dirpath) / name
35
36
37def pretty_project(raw: str) -> str:
38    """Tools encode the project path into a folder name by swapping slashes for
39    dashes. We can't perfectly reverse that, but the last meaningful segment is a
40    good label — "my-app" instead of a giant string.
41    """
42    if not raw:
43        return "unknown"
44    cleaned = raw.lstrip("-")
45    parts = [p for p in cleaned.replace("\\", "/").replace("-", "/").split("/") if p]
46    return parts[-1] if parts else "unknown"
47
48
49def to_date(value) -> datetime | None:
50    if value is None:
51        return None
52    try:
53        if isinstance(value, (int, float)):
54            # heuristic: seconds vs milliseconds
55            secs = value / 1000 if value > 1e12 else value
56            return datetime.fromtimestamp(secs, tz=timezone.utc)
57        s = str(value).replace("Z", "+00:00")
58        return datetime.fromisoformat(s)
59    except (ValueError, OSError, OverflowError):
60        return None
61
62
63EPOCH = datetime.fromtimestamp(0, tz=timezone.utc)