From cc5b83bfca6d7e20b86008007e0469577f1ca639 Mon Sep 17 00:00:00 2001 From: admin bi migration Date: Sat, 29 Aug 2026 14:41:28 -0400 Subject: [PATCH 1/5] feat(sources): add a Bahulam backend Adds a read-only store for Bahulam session JSONLs under ~/.bahulam/, on par with the Claude and Codex stores. Supports the full tooling surface: summary, workflows, model-breakdown, timeline, turn-level detail, tools tab, demo mode, and cache provenance. --- src/opentab/__init__.py | 1 + src/opentab/cli.py | 11 + src/opentab/doctor.py | 1 + src/opentab/sources.py | 25 ++ src/opentab/stores/bahulam.py | 722 ++++++++++++++++++++++++++++++++++ tests/test_doctor.py | 1 + 6 files changed, 761 insertions(+) create mode 100644 src/opentab/stores/bahulam.py diff --git a/src/opentab/__init__.py b/src/opentab/__init__.py index 21db5b9..369e4ef 100644 --- a/src/opentab/__init__.py +++ b/src/opentab/__init__.py @@ -150,6 +150,7 @@ ) from opentab.state import apply_state, load_state, save_state, state_path from opentab.stores.antigravity import AntigravityStore +from opentab.stores.bahulam import BahulamStore from opentab.stores.cached import CachedStore from opentab.stores.claude import ( CLAUDE_RETENTION_DEFAULT_DAYS, diff --git a/src/opentab/cli.py b/src/opentab/cli.py index 239c359..abaf3f3 100644 --- a/src/opentab/cli.py +++ b/src/opentab/cli.py @@ -38,6 +38,7 @@ SOURCE_LABELS, _default_antigravity_dir, _default_gemini_dir, + _default_bahulam_dir, _default_omp_dir, _default_openclaw_dir, _default_pi_dir, @@ -98,6 +99,7 @@ def _add_global_args(parser: argparse.ArgumentParser) -> None: "zaly", "gemini", "antigravity", + "bahulam", "all", "remote", ), @@ -178,6 +180,12 @@ def _add_global_args(parser: argparse.ArgumentParser) -> None: help="Gemini home directory holding antigravity/conversations/*.db (for " "--harness antigravity); honors $GEMINI_CLI_HOME, default ~/.gemini", ) + parser.add_argument( + "--bahulam-dir", + default=_default_bahulam_dir(), + help="Bahulam Code projects directory (for --harness bahulam); " + "honors $BAHULAM_PROJECTS_DIR, default ~/.bahulam/projects", + ) parser.add_argument( "--csv", default=None, @@ -277,6 +285,7 @@ def _add_legacy_command_flags(parser: argparse.ArgumentParser) -> None: "subtree included) and exit, consulting every present harness backend " "(OpenCode, Claude Code, Codex, Hermes, pi, omp, OpenClaw, Zaly, Gemini); with DIR " "only " + "(OpenCode, Claude Code, Codex, Hermes, pi, omp, OpenClaw, Zaly, Bahulam Code); with DIR only " "sessions of that project count, with a session id (ses_... or a UUID -- the " "id is matched to its own backend) exactly that session is priced, and " "--harness pins one backend. Made for a tmux status line: set -g " @@ -549,6 +558,7 @@ def _build_parser() -> argparse.ArgumentParser: "zaly_dir", "gemini_dir", "antigravity_dir", + "bahulam_dir", "csv", "jsonl", "remotes", @@ -1101,6 +1111,7 @@ def _project_key(directory: str) -> str: "zaly", "gemini", "antigravity", + "bahulam", ) diff --git a/src/opentab/doctor.py b/src/opentab/doctor.py index 95d2253..2cb6db9 100644 --- a/src/opentab/doctor.py +++ b/src/opentab/doctor.py @@ -311,6 +311,7 @@ def _path_row(pkg_dir: str, full: bool) -> Row: ("zaly", "zaly_dir", _TREE, "sessions/*/*/session.jsonl", "**/session.jsonl", "--zaly-dir/$ZALY_DATA want the DATA directory holding sessions/, not sessions/ itself"), ("gemini", "gemini_dir", _TREE, "tmp/*/chats/**/*.json*", "**/*.json*", "--gemini-dir/$GEMINI_CLI_HOME want the .gemini HOME holding tmp/, not a chats directory"), ("antigravity", "antigravity_dir", _TREE, "antigravity*/conversations/*.db", "**/*.db", "--antigravity-dir wants the .gemini HOME holding antigravity/, not a conversations directory"), + ("bahulam", "bahulam_dir", _TREE, "**/*.jsonl", "", "Bahulam Code writes transcripts under ~/.bahulam/projects; it honors $BAHULAM_PROJECTS_DIR, or pass --bahulam-dir"), ) # fmt: skip diff --git a/src/opentab/sources.py b/src/opentab/sources.py index 7c3226b..38a3370 100644 --- a/src/opentab/sources.py +++ b/src/opentab/sources.py @@ -13,6 +13,7 @@ CONVERSATION_DIRS as ANTIGRAVITY_DIRS, ) from opentab.stores.antigravity import AntigravityStore, default_antigravity_dir +from opentab.stores.bahulam import BahulamStore from opentab.stores.cached import CachedStore from opentab.stores.claude import ClaudeStore from opentab.stores.codex import CodexStore, codex_archive_dirs @@ -102,6 +103,12 @@ def _default_zaly_dir() -> str: return default_zaly_data_dir() +def _default_bahulam_dir() -> str: + """Return the default Bahulam Code projects directory from env or ``~/.bahulam/projects``.""" + env = (os.environ.get("BAHULAM_PROJECTS_DIR") or "").strip() + return env or os.path.expanduser("~/.bahulam/projects") + + _PATH_SLOT = { "csv": "csv", "jsonl": "jsonl", @@ -117,6 +124,7 @@ def _default_zaly_dir() -> str: "zaly": "zaly_dir", "gemini": "gemini_dir", "antigravity": "antigravity_dir", + "bahulam": "bahulam_dir", } @@ -270,6 +278,11 @@ def _antigravity_available(root_dir: str) -> bool: return False +def _bahulam_available(root_dir: str) -> bool: + """Return ``True`` if *root_dir* contains at least one ``.jsonl`` file.""" + return _jsonl_dir_available(root_dir) + + def _copilot_otel_available(args: argparse.Namespace) -> bool: if _jsonl_dir_available(getattr(args, "copilot_dir", "")): return True @@ -314,6 +327,7 @@ def _vscode_available(args: argparse.Namespace) -> bool: "zaly": "Zaly", "gemini": "Gemini", "antigravity": "Antigravity", + "bahulam": "Bahulam Code", "all": "all", } @@ -328,6 +342,7 @@ def _vscode_available(args: argparse.Namespace) -> bool: "Zaly": "zaly --session", "Gemini": "gemini --resume", "Antigravity": "antigravity", + "Bahulam Code": "bahulam resume", } @@ -350,6 +365,7 @@ def _detect_fingerprint(args: argparse.Namespace) -> tuple: "zaly_dir", "gemini_dir", "antigravity_dir", + "bahulam_dir", ) ) + (os.environ.get("COPILOT_OTEL_FILE_EXPORTER_PATH", ""),) @@ -390,6 +406,8 @@ def available_sources(args: argparse.Namespace) -> list[str]: keys.append("gemini") if _antigravity_available(getattr(args, "antigravity_dir", "")): keys.append("antigravity") + if _bahulam_available(getattr(args, "bahulam_dir", "")): + keys.append("bahulam") args._available_sources = (fp, keys) return list(keys) @@ -546,6 +564,13 @@ def _build_store(args: argparse.Namespace, key: str) -> tuple[object, str]: AntigravityStore(args.antigravity_dir, args), "OpenTab: loading Antigravity conversations…\r", ) + if key == "bahulam": + if not _bahulam_available(getattr(args, "bahulam_dir", "")): + raise SystemExit( + "No Bahulam Code sessions found. Point --bahulam-dir (or $BAHULAM_PROJECTS_DIR) " + f"at ~/.bahulam/projects (looked in {getattr(args, 'bahulam_dir', '')})." + ) + return BahulamStore(args.bahulam_dir, args), "OpenTab: loading Bahulam Code sessions…\r" kind, problem = opencode_db_verdict(args.db) if kind: raise SystemExit(problem) diff --git a/src/opentab/stores/bahulam.py b/src/opentab/stores/bahulam.py new file mode 100644 index 0000000..c62a249 --- /dev/null +++ b/src/opentab/stores/bahulam.py @@ -0,0 +1,722 @@ +"""Bahulam Code JSONL transcript backend.""" +from __future__ import annotations + +import argparse +import glob +import json +import os + +from opentab.demo import demo_config, scramble_node, scramble_workflow +from opentab.formatting import _clean_prompt, iso_to_epoch, iso_to_local, worked_seconds +from opentab.models import Workflow +from opentab.pricing import api_equivalent_cost, model_family +from opentab.util import git_root, read_files_parallel, safe_int, tool_rows_from_turns + + +class BahulamStore: + """Read Bahulam Code transcripts from ``~/.bahulam/projects/**/*.jsonl``. + + Bahulam records per-turn token usage and total_cost in every ``complete`` + event. The wire format uses ``kepler_event`` or ``bahulam_event`` as the + top-level type; all payload fields live under ``event.data.*``. + """ + + records_cost = True # Bahulam records total_cost in every usage block + combined = False + source_name = "Bahulam Code" + + def __init__(self, root_dir: str, args: argparse.Namespace): + """Store constructor. + + Args: + root_dir: Directory tree to scan for ``.jsonl`` session files. + args: CLI arguments, used for demo config and optional overrides. + """ + self.root_dir = root_dir + self.args = args + self.demo, self.demo_scale, self.demo_cats = demo_config(args) + self._sessions: dict[str, dict] | None = None + self._one: tuple[str, dict] | None = None # single-session memo for detail tabs + self._git_root_cache: dict[str, str] = {} + + # ── helpers ─────────────────────────────────────────────────────────── + + @staticmethod + def _new_acc() -> dict: + """Return a zeroed accumulator dict with all billing counters.""" + return { + "runs": 0, + "cost": 0.0, + "input": 0, + "output": 0, + "reasoning": 0, + "cache_read": 0, + "cache_write": 0, + "cache_write_1h": 0, + "tokens_total": 0, + } + + @staticmethod + def _int(value) -> int: + """Safely cast *value* to int, returning 0 on failure.""" + return safe_int(value) + + @staticmethod + def _float(value) -> float: + """Safely cast *value* to float, returning 0.0 on failure.""" + try: + return float(value or 0) + except (TypeError, ValueError): + return 0.0 + + @classmethod + def _add_usage(cls, acc: dict, usage: dict) -> None: + """Accumulate counters from a usage dict into *acc*. + + Args: + acc: Mutable accumulator dict (see ``_new_acc``). + usage: Raw usage record from a ``complete`` event. + """ + total_in = cls._int(usage.get("total_input_tokens", 0)) + out = cls._int(usage.get("total_output_tokens", 0)) + cr = cls._int(usage.get("cache_read_input_tokens", 0)) + cw = cls._int(usage.get("cache_creation_input_tokens", 0)) + reasoning = cls._int(usage.get("reasoning_tokens", 0)) + cost = cls._float(usage.get("cost", 0)) + cc = usage.get("cache_creation") + cw1h = cls._int(cc.get("ephemeral_1h_input_tokens", 0) or 0) if isinstance(cc, dict) else 0 + inp = max(0, total_in - cr - cw) + acc["runs"] += 1 + acc["cost"] += cost + acc["input"] += inp + acc["output"] += out + acc["reasoning"] += reasoning + acc["cache_read"] += cr + acc["cache_write"] += cw + acc["cache_write_1h"] += min(cw1h, cw) + acc["tokens_total"] += inp + out + reasoning + cr + cw + + @staticmethod + def _price(model_name: str, acc: dict) -> float: + """Return the API-equivalent cost for a given model and accumulator.""" + return api_equivalent_cost( + model_name, + acc["input"], + acc["output"], + acc["reasoning"], + acc["cache_read"], + acc["cache_write"], + acc.get("cache_write_1h", 0), + ) + + @staticmethod + def _provider_prefix(model_name: str) -> str: + """Infer the provider prefix for a bare model name (e.g. ``deepseek-v4-pro`` -> ``deepseek/``). + + Returns the prefix string with trailing slash, or ``""`` if the family is unknown. + """ + fam = model_family(model_name) + if not fam or fam == "unknown": + return "anthropic/" # safest fallback for Claude-like naming + return fam + "/" + + def _qualified_model(self, model_name: str) -> str: + """Return *model_name* with its provider prefix prepended if it is bare.""" + if not model_name or "/" in model_name: + return model_name + return self._provider_prefix(model_name) + model_name + + def _git_root(self, cwd: str) -> str: + """Return the git root for a working directory, cached per directory.""" + if cwd not in self._git_root_cache: + self._git_root_cache[cwd] = git_root(cwd) + return self._git_root_cache[cwd] + + # ── file discovery ──────────────────────────────────────────────────── + + def cache_inputs(self) -> list[str]: + """Return all session file paths (cache input contract).""" + return self._files() + + def _files(self) -> list[str]: + """Glob every ``.jsonl`` under ``root_dir`` recursively.""" + return glob.glob(os.path.join(self.root_dir, "**", "*.jsonl"), recursive=True) + + # ── single-session fast path ────────────────────────────────────────── + + def _parse_one(self, workflow_id: str) -> dict | None: + """Parse a single session by ID without loading all files. + + Returns the session dict, or ``None`` if not found. + """ + for path in self._files(): + if os.path.splitext(os.path.basename(path))[0] == workflow_id: + items = [(path, _read_text(path))] + sessions = self._parse_texts(items) + return sessions.get(workflow_id) + return None + + def _session(self, workflow_id: str, fallback: bool = True) -> dict | None: + """Return a session dict by ID, using fast-path lookups when possible. + + Args: + workflow_id: The session ID (filename stem). + fallback: When ``True``, fall back to a full parse if not found in + the single-session fast path. + + Returns the session dict or ``None``. + """ + if self._sessions is not None: + return self._sessions.get(workflow_id) + if self._one is not None and self._one[0] == workflow_id: + return self._one[1] + s = self._parse_one(workflow_id) + if s is not None: + self._one = (workflow_id, s) + return s + if not fallback: + return None + return self._parse().get(workflow_id) + + # ── parsing ─────────────────────────────────────────────────────────── + + def _parse(self) -> dict[str, dict]: + """Parse all session files, memoised. + + Returns ``session_id -> session dict``. + """ + if self._sessions is not None: + return self._sessions + self._sessions = self._parse_texts(read_files_parallel(self._files())) + return self._sessions + + def _parse_texts(self, items) -> dict[str, dict]: + """Parse a stream of ``(path, text)`` tuples into session dicts. + + Each path yields one session keyed by its filename stem. + """ + sessions: dict[str, dict] = {} + for path, text in items: + session_id = os.path.splitext(os.path.basename(path))[0] + s = sessions.setdefault(session_id, self._new_session()) + s["files"].add(path) + self._parse_file(text, session_id, s, path) + for sid, s in sessions.items(): + self._finalize(sid, s) + return sessions + + @staticmethod + def _new_session() -> dict: + """Return a fresh, empty session dict with all fields initialised.""" + return { + "cwd": None, + "ts_min": None, + "ts_max": None, + "title_prompt": None, + "model": None, + "models": {}, # model_name -> {"total": acc, "root": acc} + "turns": [], + "prompts": [], + "event_ts": [], + "pending_tools": [], # tool names queued before the next complete event + "files": set(), + } + + def _parse_file(self, text: str, session_id: str, s: dict, path: str = "") -> None: + """Decode a single JSONL file body and ingest every line into *s*.""" + for line in text.split("\n"): + line = line.strip() + if not line: + continue + try: + obj = json.loads(line) + except (json.JSONDecodeError, ValueError): + continue + if not isinstance(obj, dict): + continue + self._ingest(obj, s, path) + + def _ingest(self, o: dict, s: dict, path: str = "") -> None: + """Ingest a single JSON record *o* into the session dict *s*. + + Handles user messages, ``bahulam_event`` / ``kepler_event`` records + (session_info, complete, tool_call), and extracts timestamps / cwd. + """ + if path: + s["files"].add(path) + + ts = o.get("timestamp") + if ts: + if s["ts_min"] is None or ts < s["ts_min"]: + s["ts_min"] = ts + if s["ts_max"] is None or ts > s["ts_max"]: + s["ts_max"] = ts + s["event_ts"].append(ts) + + # Top-level cwd — every record carries it + cwd = o.get("cwd") + if cwd and not s["cwd"]: + s["cwd"] = cwd + + typ = o.get("type") + + # User messages — seed title from the first real prompt + if typ in ("user",): + msg = o.get("message") + if isinstance(msg, dict): + content = msg.get("content") + text = None + if isinstance(content, str): + text = content.strip() + elif isinstance(content, list): + for block in content: + if isinstance(block, dict) and block.get("type") == "text": + text = (block.get("text") or "").strip() + break + if text and not s["title_prompt"]: + s["title_prompt"] = text[:80] + if text: + s["prompts"].append({"ts": ts or "", "title": text}) + + # Event records — both ``bahulam_event`` and ``kepler_event`` carry + # their payload under ``event.data.*``. + if typ not in ("bahulam_event", "kepler_event"): + return + event = o.get("event") + if not isinstance(event, dict): + return + + event_type = event.get("type") + data = event.get("data") if isinstance(event.get("data"), dict) else {} + + # session_info — carries the model map + if event_type == "session_info": + models_map = data.get("models") + if isinstance(models_map, dict): + orch = models_map.get("orchestrator") or "" + if orch and not s["model"]: + s["model"] = self._qualified_model(orch) + for model_name in models_map.values(): + qualified = self._qualified_model(model_name) if model_name else "" + if qualified and qualified not in s["models"]: + s["models"].setdefault( + qualified, + {"total": self._new_acc(), "root": self._new_acc()}, + ) + + # complete — carries per-LLM-step token usage and cost + elif event_type == "complete": + usage = data.get("usage") or {} + if not usage: + return + + reasoning = self._int(usage.get("reasoning_tokens", 0)) + + # Per-model token breakdown (the wire uses ``cache_read_tokens`` + # and ``cache_creation_tokens`` at per-model granularity). + models_usage = usage.get("models") + if isinstance(models_usage, list): + for m in models_usage: + model_name = self._qualified_model(m.get("model", "")) + if not model_name: + continue + entry = s["models"].get(model_name) + if entry is None: + entry = s["models"][model_name] = { + "total": self._new_acc(), + "root": self._new_acc(), + } + per_model = { + "total_input_tokens": m.get("input_tokens", 0), + "total_output_tokens": m.get("output_tokens", 0), + "cache_read_input_tokens": m.get("cache_read_tokens", 0), + "cache_creation_input_tokens": m.get("cache_creation_tokens", 0), + "reasoning_tokens": 0, # reasoning is top-level only + "cost": m.get("cost", 0) or 0, + } + self._add_usage(entry["total"], per_model) + self._add_usage(entry["root"], per_model) + + # Aggregate totals for the turn row + total_in = self._int(usage.get("total_input_tokens", 0)) + out = self._int(usage.get("total_output_tokens", 0)) + cr = self._int(usage.get("cache_read_input_tokens", 0)) + cw = self._int(usage.get("cache_creation_input_tokens", 0)) + inp = max(0, total_in - cr - cw) + total_cost = self._float(usage.get("total_cost", 0) or usage.get("cost", 0)) + + # Turn row for the Turns tab + first_model = "unknown" + if isinstance(models_usage, list) and models_usage: + first_model = self._qualified_model(models_usage[0].get("model", "unknown")) + turn_tools = list(s["pending_tools"]) + s["pending_tools"].clear() + s["turns"].append( + { + "ts": ts or "", + "depth": 0, + "agent": "-", + "effort": "", + "model_name": first_model, + "cost": float(total_cost), + "input": inp, + "output": out, + "reasoning": reasoning, + "cache_read": cr, + "cache_write": cw, + "cache_write_1h": 0, + "tokens_total": inp + out + reasoning + cr + cw, + "tools": turn_tools, + } + ) + + # tool_call — queue the tool name for the next turn + elif event_type == "tool_call": + tool_name = data.get("tool", "") + if tool_name: + s["pending_tools"].append(tool_name) + + # ── finalization ────────────────────────────────────────────────────── + + def _finalize(self, sid: str, s: dict) -> None: + """Derive computed fields (title, directory, times, model-rows, subagents). + + Called once per session after all lines have been ingested. + """ + s["title"] = s["title_prompt"] or "(untitled)" + s["directory"] = self._git_root(s["cwd"]) if s["cwd"] else "(unknown)" + s["created_at"] = iso_to_local(s["ts_min"]) if s["ts_min"] else "" + s["ended_at"] = iso_to_local(s["ts_max"]) if s["ts_max"] else "" + s["worked_seconds"] = worked_seconds( + [iso_to_epoch(t) for t in s["event_ts"]], + [iso_to_epoch(p["ts"]) for p in s["prompts"]], + ) + rows: list[dict] = [] + for model_name, e in s["models"].items(): + tot, root = e["total"], e["root"] + rows.append( + { + "root_id": sid, + "model_name": model_name, + "runs": tot["runs"], + "cost": tot["cost"], + "root_cost": root["cost"], + "tokens_total": tot["tokens_total"], + "input": tot["input"], + "reasoning": tot["reasoning"], + "cache_read": tot["cache_read"], + "cache_write": tot["cache_write"], + "cache_write_1h": tot["cache_write_1h"], + "output": tot["output"], + # 1h-TTL subset for API-equivalent estimation + "unpriced_input": tot["input"], + "unpriced_reasoning": tot["reasoning"], + "unpriced_cache_read": tot["cache_read"], + "unpriced_cache_write": tot["cache_write"], + "unpriced_cache_write_1h": tot["cache_write_1h"], + "unpriced_output": tot["output"], + "root_unpriced_input": root["input"], + "root_unpriced_reasoning": root["reasoning"], + "root_unpriced_cache_read": root["cache_read"], + "root_unpriced_cache_write": root["cache_write"], + "root_unpriced_cache_write_1h": root["cache_write_1h"], + "root_unpriced_output": root["output"], + } + ) + s["model_rows"] = rows + s["unpriced_tokens"] = sum(r["tokens_total"] for r in rows) + s["subagents"] = [] + + @staticmethod + def _node( + node_id: str, + depth: int, + agent: str, + title: str, + created_at: str, + model_name: str, + cost: float, + acc: dict, + ) -> dict: + """Build a node dict for the UI graph from a session and its accumulator.""" + return { + "id": node_id, + "depth": depth, + "agent": agent, + "title": title, + "created_at": created_at, + "cost": round(cost, 6), + "model_name": model_name, + "tokens_input": acc["input"], + "tokens_output": acc["output"], + "tokens_reasoning": acc["reasoning"], + "tokens_cache_read": acc["cache_read"], + "tokens_cache_write": acc["cache_write"], + "tokens_cache_write_1h": acc["cache_write_1h"], + "tokens_total": acc["tokens_total"], + } + + def _nodes(self, workflow_id: str, s: dict) -> list[dict]: + """Build the node list including root and all subagents for a session.""" + root_tot = self._new_acc() + best, best_runs = "unknown (not recorded)", -1 + for model_name, e in s["models"].items(): + r = e["root"] + for k in root_tot: + root_tot[k] += r[k] + if r["runs"] > best_runs: + best_runs, best = r["runs"], model_name + nodes = [ + self._node( + workflow_id, 0, "-", s["title"], s["created_at"], best, root_tot["cost"], root_tot + ) + ] + nodes.extend(dict(n) for n in s["subagents"]) + if self.demo: + nodes = [self._demo_node(n) for n in nodes] + return nodes + + @staticmethod + def sort_workflows(rows: list[Workflow]) -> list[Workflow]: + """Stable two-pass sort: alpha by id, then descending by cost then tokens.""" + rows = sorted(rows, key=lambda w: w.id) + rows.sort(key=lambda w: (w.total_cost, w.total_tokens), reverse=True) + return rows + + def _workflow_rows(self, sessions: dict[str, dict]) -> list[Workflow]: + """Convert session dicts into a sorted list of ``Workflow`` rows.""" + rows = [] + for sid, s in sessions.items(): + model_rows = s["model_rows"] + total_cost = sum(r["cost"] for r in model_rows) + root_cost = sum(r["root_cost"] for r in model_rows) + rows.append( + Workflow( + id=sid, + title=s["title"], + directory=s["directory"], + created_at=s["created_at"], + root_cost=root_cost, + total_cost=total_cost, + subagents=len(s["subagents"]), + model_count=0, + total_tokens=sum(r["tokens_total"] for r in model_rows), + unpriced_tokens=s["unpriced_tokens"], + source=self.source_name, + ended_at=s["ended_at"], + worked_seconds=s["worked_seconds"], + ) + ) + if self.demo: + rows = [self._demo_workflow(w) for w in rows] + return self.sort_workflows(rows) + + def _demo_workflow(self, w: Workflow) -> Workflow: + """Anonymise a workflow row for demo mode.""" + return scramble_workflow(w, self.demo_scale, self.demo_cats) + + def _demo_node(self, n: dict) -> dict: + """Anonymise a node dict for demo mode.""" + return scramble_node(n, self.demo_scale, self.demo_cats) + + # ── public interface ────────────────────────────────────────────────── + + def workflows(self) -> list[Workflow]: + """Return all sessions as a sorted ``Workflow`` list. + + Clears any cached parse so the returned data is fresh. + """ + self._sessions = None + self._one = None + return self._workflow_rows(self._parse()) + + def model_breakdown(self) -> list[dict]: + """Return a flat list of per-model rows across all sessions.""" + return self._model_rows(self._parse()) + + @staticmethod + def _model_rows(sessions: dict[str, dict]) -> list[dict]: + """Flatten all per-session ``model_rows`` into a single list.""" + out: list[dict] = [] + for s in sessions.values(): + out.extend(s["model_rows"]) + return out + + def summary(self, workflows: list[Workflow]) -> dict[str, int | float]: + """Aggregate totals across all *workflows* for the summary banner.""" + return { + "workflows": len(workflows), + "cost": sum(w.total_cost for w in workflows), + "tokens": sum(w.total_tokens for w in workflows), + "subagents": sum(w.subagents for w in workflows), + "unpriced_tokens": sum(w.unpriced_tokens for w in workflows), + "paid_workflows": sum(1 for w in workflows if w.total_cost > 0), + } + + def workflow_nodes(self, workflow_id: str) -> list[dict]: + """Return the graph nodes for a session, loading it on demand.""" + s = self._session(workflow_id) + if not s: + return [] + return self._nodes(workflow_id, s) + + def status_nodes(self, workflow_id: str) -> list[dict]: + """Return graph nodes for a session without falling back to a full parse. + + Used by status/info commands that should stay fast. + """ + s = self._session(workflow_id, fallback=False) + if not s: + return [] + return self._nodes(workflow_id, s) + + def message_timeline(self, workflow_id: str) -> list[dict]: + """Return interleaved turn+prompt rows for the timeline view. + + Each row carries ``prompt_id``, ``prompt_title``, and ``prompt_full`` + representing the prompt active at that point in the conversation. + """ + s = self._session(workflow_id) + if not s: + return [] + prompts = sorted(s["prompts"], key=lambda p: p["ts"]) + out = [] + pi, cur_title, cur_full = 0, "", "" + for t in sorted(s["turns"], key=lambda r: r["ts"]): + while pi < len(prompts) and prompts[pi]["ts"] <= t["ts"]: + cur_full = prompts[pi]["title"] + cur_title = _clean_prompt(cur_full) + pi += 1 + r = dict(t) + r["time"] = iso_to_local(r.pop("ts")) + r["prompt_id"] = f"p{pi}" + r["prompt_title"] = cur_title + r["prompt_full"] = cur_full + out.append(r) + return out + + def supports_turns(self, workflow_id: str) -> bool: + """Indicate whether turn detail is available (always ``True``).""" + return True + + def tool_breakdown(self, workflow_id: str) -> list[dict]: + """Return tool call rows for the tools tab.""" + s = self._session(workflow_id) + if not s: + return [] + return tool_rows_from_turns(s["turns"]) + + def supports_tools(self, workflow_id: str) -> bool: + """Indicate whether tool data is available (always ``True``).""" + return True + + def recent_roots(self) -> list[dict]: + """Return recently modified session roots for the project picker. + + Each root is a ``_TranscriptRoot`` dict that lazily resolves its + ``directory`` key from the file header. + """ + newest: dict[str, _TranscriptRoot] = {} + for path in self._files(): + sid = os.path.splitext(os.path.basename(path))[0] + try: + last_active = int(os.stat(path).st_mtime * 1000) + except OSError: + continue + if sid not in newest or last_active > newest[sid]["last_active"]: + newest[sid] = _TranscriptRoot(self, path, sid, last_active) + return sorted(newest.values(), key=lambda r: r["last_active"], reverse=True) + + _CWD_HEAD_BYTES = 262144 + + def _transcript_cwd(self, path: str) -> str: + """Read the first ``cwd`` field from a JSONL file without parsing it fully.""" + try: + with open(path, encoding="utf-8", errors="replace") as fh: + remaining = self._CWD_HEAD_BYTES + while remaining > 0: + line = fh.readline() + if not line: + break + remaining -= len(line) + line = line.strip() + if not line: + continue + try: + obj = json.loads(line) + except (json.JSONDecodeError, ValueError): + continue + if not isinstance(obj, dict): + continue + # Top-level cwd is on every record + cwd = obj.get("cwd") + if cwd: + return cwd + except OSError: + pass + return "(unknown)" + + def root_of(self, session_id: str) -> str | None: + """Return the session ID if a file for that ID exists, otherwise ``None``.""" + for path in self._files(): + if os.path.splitext(os.path.basename(path))[0] == session_id: + return session_id + return None + + def cache_provenance(self) -> dict[str, list[str]]: + """Return a ``session_id -> [file paths]`` map for cache-backed sessions.""" + if not self._sessions: + return {} + return {sid: sorted(s["files"]) for sid, s in self._sessions.items()} + + def parse_subset(self, paths: list[str]) -> tuple[list[Workflow], list[dict], dict] | None: + """Reparse a specific set of files and return (workflows, model-rows, provenance). + + Returns ``None`` if any file in *paths* is missing. + """ + wanted = set(paths) + files = [p for p in self._files() if p in wanted] + if len(files) != len(wanted): + return None + self._one = None + self._sessions = None + read: list = [] + + def tally(stream): + for path, text in stream: + read.append(path) + yield path, text + + sessions = self._parse_texts(tally(read_files_parallel(files))) + if len(read) != len(files): + return None + return ( + self._workflow_rows(sessions), + self._model_rows(sessions), + {sid: sorted(s["files"]) for sid, s in sessions.items()}, + ) + + +class _TranscriptRoot(dict): + """A ``recent_roots()`` row that reads ``directory`` lazily from the file head.""" + + def __init__(self, store: BahulamStore, path: str, sid: str, last_active: int): + """Lazy root with id, mtime, and deferred directory resolution.""" + super().__init__(id=sid, last_active=last_active) + self._store = store + self._path = path + + def __getitem__(self, key): + """Resolve ``directory`` on first access by reading the file header.""" + if key == "directory" and "directory" not in self: + self["directory"] = self._store._transcript_cwd(self._path) + return super().__getitem__(key) + + +def _read_text(path: str) -> str: + """Read a file's text content, returning ``""`` on any I/O error.""" + try: + with open(path, encoding="utf-8", errors="replace") as fh: + return fh.read() + except OSError: + return "" diff --git a/tests/test_doctor.py b/tests/test_doctor.py index 32d877a..43ce63d 100644 --- a/tests/test_doctor.py +++ b/tests/test_doctor.py @@ -20,6 +20,7 @@ "--db", "--claude-dir", "--codex-dir", "--hermes-db", "--copilot-dir", "--vscode-dir", "--pi-dir", "--omp-dir", "--openclaw-dir", "--zaly-dir", "--gemini-dir", "--antigravity-dir", + "--bahulam-dir", ) # fmt: skip # Anything in the ambient environment that would change a verdict. From e2793fbc735aad38751860d1a5ed9aa340f14c19 Mon Sep 17 00:00:00 2001 From: admin bi migration Date: Sat, 29 Aug 2026 15:39:27 -0400 Subject: [PATCH 2/5] Align Bahulam transcript parsing --- src/opentab/cli.py | 5 +- src/opentab/stores/bahulam.py | 72 +++++++++++++--- tests/test_cli.py | 3 + tests/test_stores_bahulam.py | 158 ++++++++++++++++++++++++++++++++++ 4 files changed, 223 insertions(+), 15 deletions(-) create mode 100644 tests/test_stores_bahulam.py diff --git a/src/opentab/cli.py b/src/opentab/cli.py index abaf3f3..bad74e6 100644 --- a/src/opentab/cli.py +++ b/src/opentab/cli.py @@ -106,9 +106,8 @@ def _add_global_args(parser: argparse.ArgumentParser) -> None: default="auto", help="which harness's spend to browse: opencode · claude · codex · hermes · csv · " "jsonl · copilot · vscode · pi · omp · openclaw · zaly · gemini · antigravity · " - "all (merged) · " - "remote " - "(other machines, via pull/export). Default auto merges every present local " + "bahulam · all (merged) · " + "remote (other machines, via pull/export). Default auto merges every present local " "harness. Or just pass a file path -- e.g. `opentab requests.csv`. (--source is a " "deprecated alias for --harness)", ) diff --git a/src/opentab/stores/bahulam.py b/src/opentab/stores/bahulam.py index c62a249..0c30611 100644 --- a/src/opentab/stores/bahulam.py +++ b/src/opentab/stores/bahulam.py @@ -69,6 +69,22 @@ def _float(value) -> float: except (TypeError, ValueError): return 0.0 + @staticmethod + def _first_value(*values): + """Return the first non-empty value from *values*, or 0.""" + for value in values: + if value not in (None, ""): + return value + return 0 + + def _primary_model(self, models_map: dict) -> str: + """Return the execution model from a session_info models map.""" + for key in ("coder", "main", "executor", "orchestrator", "planning"): + model_name = models_map.get(key) + if model_name: + return self._qualified_model(model_name) + return "" + @classmethod def _add_usage(cls, acc: dict, usage: dict) -> None: """Accumulate counters from a usage dict into *acc*. @@ -82,7 +98,14 @@ def _add_usage(cls, acc: dict, usage: dict) -> None: cr = cls._int(usage.get("cache_read_input_tokens", 0)) cw = cls._int(usage.get("cache_creation_input_tokens", 0)) reasoning = cls._int(usage.get("reasoning_tokens", 0)) - cost = cls._float(usage.get("cost", 0)) + cost = cls._float( + cls._first_value( + usage.get("cost"), + usage.get("cost_usd"), + usage.get("total_cost"), + usage.get("total_cost_usd"), + ) + ) cc = usage.get("cache_creation") cw1h = cls._int(cc.get("ephemeral_1h_input_tokens", 0) or 0) if isinstance(cc, dict) else 0 inp = max(0, total_in - cr - cw) @@ -293,9 +316,9 @@ def _ingest(self, o: dict, s: dict, path: str = "") -> None: if event_type == "session_info": models_map = data.get("models") if isinstance(models_map, dict): - orch = models_map.get("orchestrator") or "" - if orch and not s["model"]: - s["model"] = self._qualified_model(orch) + primary = self._primary_model(models_map) + if primary and not s["model"]: + s["model"] = primary for model_name in models_map.values(): qualified = self._qualified_model(model_name) if model_name else "" if qualified and qualified not in s["models"]: @@ -315,8 +338,10 @@ def _ingest(self, o: dict, s: dict, path: str = "") -> None: # Per-model token breakdown (the wire uses ``cache_read_tokens`` # and ``cache_creation_tokens`` at per-model granularity). models_usage = usage.get("models") - if isinstance(models_usage, list): + if isinstance(models_usage, list) and models_usage: for m in models_usage: + if not isinstance(m, dict): + continue model_name = self._qualified_model(m.get("model", "")) if not model_name: continue @@ -331,11 +356,24 @@ def _ingest(self, o: dict, s: dict, path: str = "") -> None: "total_output_tokens": m.get("output_tokens", 0), "cache_read_input_tokens": m.get("cache_read_tokens", 0), "cache_creation_input_tokens": m.get("cache_creation_tokens", 0), - "reasoning_tokens": 0, # reasoning is top-level only - "cost": m.get("cost", 0) or 0, + "reasoning_tokens": m.get("reasoning_tokens", 0) or 0, + "cost": self._first_value(m.get("cost"), m.get("cost_usd")), } self._add_usage(entry["total"], per_model) self._add_usage(entry["root"], per_model) + else: + model_name = self._qualified_model( + data.get("model") or usage.get("model") or s["model"] or "" + ) + if model_name: + entry = s["models"].get(model_name) + if entry is None: + entry = s["models"][model_name] = { + "total": self._new_acc(), + "root": self._new_acc(), + } + self._add_usage(entry["total"], usage) + self._add_usage(entry["root"], usage) # Aggregate totals for the turn row total_in = self._int(usage.get("total_input_tokens", 0)) @@ -343,12 +381,22 @@ def _ingest(self, o: dict, s: dict, path: str = "") -> None: cr = self._int(usage.get("cache_read_input_tokens", 0)) cw = self._int(usage.get("cache_creation_input_tokens", 0)) inp = max(0, total_in - cr - cw) - total_cost = self._float(usage.get("total_cost", 0) or usage.get("cost", 0)) + total_cost = self._float( + self._first_value( + usage.get("total_cost"), + usage.get("total_cost_usd"), + usage.get("cost"), + usage.get("cost_usd"), + ) + ) # Turn row for the Turns tab first_model = "unknown" if isinstance(models_usage, list) and models_usage: - first_model = self._qualified_model(models_usage[0].get("model", "unknown")) + first = models_usage[0] if isinstance(models_usage[0], dict) else {} + first_model = self._qualified_model(first.get("model", "unknown")) + elif s["model"]: + first_model = s["model"] turn_tools = list(s["pending_tools"]) s["pending_tools"].clear() s["turns"].append( @@ -370,9 +418,9 @@ def _ingest(self, o: dict, s: dict, path: str = "") -> None: } ) - # tool_call — queue the tool name for the next turn - elif event_type == "tool_call": - tool_name = data.get("tool", "") + # tool_call / tool_request — queue the tool name for the next turn + elif event_type in ("tool_call", "tool_request"): + tool_name = data.get("tool") or data.get("name") or "" if tool_name: s["pending_tools"].append(tool_name) diff --git a/tests/test_cli.py b/tests/test_cli.py index 2f705fd..e47ad3d 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -767,6 +767,9 @@ def test_verb_help_is_focused_but_globals_still_parse(): # tui stays the full reference -- nothing hidden there. tui_help = _subparser_help("tui") assert "--claude-dir" in tui_help and "--theme" in tui_help and "--web" in tui_help + compact_tui_help = " ".join(tui_help.split()) + assert "zaly · bahulam · all (merged)" in compact_tui_help + assert "Zaly, Bahulam Code" in compact_tui_help # Hidden != gone: a suppressed global still parses on that verb. assert _parse(["pull", "--no-cache", "host"]).no_cache is True assert _parse(["export", "--harness", "claude"]).source == "claude" diff --git a/tests/test_stores_bahulam.py b/tests/test_stores_bahulam.py new file mode 100644 index 0000000..6936491 --- /dev/null +++ b/tests/test_stores_bahulam.py @@ -0,0 +1,158 @@ +import os +import tempfile + +import opentab as ot + +from tests._support import _jsonl_args, _write_jsonl + + +def _bahulam_event(event_type, data=None, *, record_type="bahulam_event", cwd="/work/repo", ts=None): + return { + "type": record_type, + "timestamp": ts or "2026-08-29T10:00:00.000Z", + "cwd": cwd, + "event": {"type": event_type, "data": data or {}}, + } + + +def _bahulam_user(text, *, cwd="/work/repo", ts="2026-08-29T09:59:00.000Z"): + return { + "type": "user", + "timestamp": ts, + "cwd": cwd, + "message": {"role": "user", "content": text}, + } + + +def _write_bahulam(root, sid, rows): + project = os.path.join(root, "projects", "-work-repo") + os.makedirs(project) + _write_jsonl(os.path.join(project, f"{sid}.jsonl"), rows) + + +def test_bahulam_store_reads_current_and_legacy_event_shapes_for_usage_and_tools(): + sid = "0198fca0-34b4-7285-b3f0-3b8fb0489e5f" + with tempfile.TemporaryDirectory() as tmp: + cwd = os.path.join(tmp, "repo") + os.makedirs(os.path.join(cwd, ".git")) + rows = [ + _bahulam_user("inspect this", cwd=cwd), + _bahulam_event( + "session_info", + { + "models": { + "planning": "deepseek/deepseek-v4-pro", + "coder": "xiaomi/mimo-v2.5", + } + }, + record_type="kepler_event", + cwd=cwd, + ), + _bahulam_event("tool_request", {"name": "read_file"}, cwd=cwd), + _bahulam_event( + "complete", + { + "usage": { + "total_input_tokens": 1000, + "total_output_tokens": 100, + "cache_read_input_tokens": 300, + "cache_creation_input_tokens": 50, + "reasoning_tokens": 20, + "total_cost_usd": 0.25, + "models": [ + { + "model": "xiaomi/mimo-v2.5", + "input_tokens": 700, + "output_tokens": 80, + "cache_read_tokens": 250, + "cache_creation_tokens": 50, + "reasoning_tokens": 7, + "cost_usd": 0.20, + }, + { + "model": "deepseek/deepseek-v4-flash", + "input_tokens": 300, + "output_tokens": 20, + "cache_read_tokens": 50, + "reasoning_tokens": 3, + "cost": 0.05, + }, + ], + } + }, + cwd=cwd, + ), + ] + _write_bahulam(tmp, sid, rows) + + store = ot.BahulamStore(os.path.join(tmp, "projects"), _jsonl_args()) + workflow = store.workflows()[0] + assert workflow.id == sid + assert workflow.title == "inspect this" + assert workflow.total_cost == 0.25 + assert workflow.total_tokens == 1110 + + timeline = store.message_timeline(sid) + assert len(timeline) == 1 + turn = timeline[0] + assert turn["model_name"] == "xiaomi/mimo-v2.5" + assert turn["tools"] == ["read_file"] + assert turn["cost"] == 0.25 + assert turn["reasoning"] == 20 + + tools = store.tool_breakdown(sid) + assert len(tools) == 1 + assert tools[0]["tool"] == "read_file" + assert tools[0]["model_name"] == "xiaomi/mimo-v2.5" + assert tools[0]["calls"] == 1 + assert tools[0]["tokens_total"] == 1120 + + by_model = {row["model_name"]: row for row in store.model_breakdown()} + assert by_model["xiaomi/mimo-v2.5"]["reasoning"] == 7 + assert by_model["xiaomi/mimo-v2.5"]["cost"] == 0.20 + assert by_model["deepseek/deepseek-v4-flash"]["reasoning"] == 3 + assert by_model["deepseek/deepseek-v4-flash"]["cost"] == 0.05 + + +def test_bahulam_store_uses_coder_model_when_usage_has_no_model_breakdown(): + sid = "0198fca0-34b4-7285-b3f0-3b8fb0489e60" + with tempfile.TemporaryDirectory() as tmp: + cwd = os.path.join(tmp, "repo") + os.makedirs(cwd) + rows = [ + _bahulam_user("hello", cwd=cwd), + _bahulam_event( + "session_info", + { + "models": { + "planning": "deepseek/deepseek-v4-pro", + "coder": "xiaomi/mimo-v2.5", + } + }, + cwd=cwd, + ), + _bahulam_event( + "complete", + { + "usage": { + "total_input_tokens": 100, + "total_output_tokens": 40, + "cache_read_input_tokens": 10, + "reasoning_tokens": 5, + "total_cost_usd": 0.03, + } + }, + cwd=cwd, + ), + ] + _write_bahulam(tmp, sid, rows) + + store = ot.BahulamStore(os.path.join(tmp, "projects"), _jsonl_args()) + timeline = store.message_timeline(sid) + assert timeline[0]["model_name"] == "xiaomi/mimo-v2.5" + + by_model = {row["model_name"]: row for row in store.model_breakdown()} + assert by_model["xiaomi/mimo-v2.5"]["runs"] == 1 + assert by_model["xiaomi/mimo-v2.5"]["cost"] == 0.03 + assert by_model["xiaomi/mimo-v2.5"]["tokens_total"] == 145 + assert by_model["deepseek/deepseek-v4-pro"]["runs"] == 0 From 214d1a9cdba576fc4ae365e8e786a3bdd41bbf36 Mon Sep 17 00:00:00 2001 From: admin bi migration Date: Sun, 30 Aug 2026 12:03:30 -0400 Subject: [PATCH 3/5] Fix Bahulam accounting and subagent rollups --- src/opentab/cli.py | 2 +- src/opentab/doctor.py | 2 +- src/opentab/sources.py | 14 +- src/opentab/stores/bahulam.py | 262 +++++++++++++++++++++++++++------- src/opentab/stores/cached.py | 2 +- tests/test_sources.py | 9 ++ tests/test_stores_bahulam.py | 161 ++++++++++++++++++++- 7 files changed, 390 insertions(+), 62 deletions(-) diff --git a/src/opentab/cli.py b/src/opentab/cli.py index bad74e6..7294d26 100644 --- a/src/opentab/cli.py +++ b/src/opentab/cli.py @@ -183,7 +183,7 @@ def _add_global_args(parser: argparse.ArgumentParser) -> None: "--bahulam-dir", default=_default_bahulam_dir(), help="Bahulam Code projects directory (for --harness bahulam); " - "honors $BAHULAM_PROJECTS_DIR, default ~/.bahulam/projects", + "honors $BAHULAM_PROJECTS_DIR, then $BAHULAM_HOME/projects; default ~/.bahulam/projects", ) parser.add_argument( "--csv", diff --git a/src/opentab/doctor.py b/src/opentab/doctor.py index 2cb6db9..21e4a4c 100644 --- a/src/opentab/doctor.py +++ b/src/opentab/doctor.py @@ -311,7 +311,7 @@ def _path_row(pkg_dir: str, full: bool) -> Row: ("zaly", "zaly_dir", _TREE, "sessions/*/*/session.jsonl", "**/session.jsonl", "--zaly-dir/$ZALY_DATA want the DATA directory holding sessions/, not sessions/ itself"), ("gemini", "gemini_dir", _TREE, "tmp/*/chats/**/*.json*", "**/*.json*", "--gemini-dir/$GEMINI_CLI_HOME want the .gemini HOME holding tmp/, not a chats directory"), ("antigravity", "antigravity_dir", _TREE, "antigravity*/conversations/*.db", "**/*.db", "--antigravity-dir wants the .gemini HOME holding antigravity/, not a conversations directory"), - ("bahulam", "bahulam_dir", _TREE, "**/*.jsonl", "", "Bahulam Code writes transcripts under ~/.bahulam/projects; it honors $BAHULAM_PROJECTS_DIR, or pass --bahulam-dir"), + ("bahulam", "bahulam_dir", _TREE, "**/*.jsonl", "", "Bahulam Code writes transcripts under ~/.bahulam/projects; it honors $BAHULAM_PROJECTS_DIR, $BAHULAM_HOME, or $KEPLER_HOME, or pass --bahulam-dir"), ) # fmt: skip diff --git a/src/opentab/sources.py b/src/opentab/sources.py index 38a3370..fe86b0d 100644 --- a/src/opentab/sources.py +++ b/src/opentab/sources.py @@ -104,9 +104,16 @@ def _default_zaly_dir() -> str: def _default_bahulam_dir() -> str: - """Return the default Bahulam Code projects directory from env or ``~/.bahulam/projects``.""" + """Return the default Bahulam Code projects directory.""" env = (os.environ.get("BAHULAM_PROJECTS_DIR") or "").strip() - return env or os.path.expanduser("~/.bahulam/projects") + if env: + return env + home = (os.environ.get("BAHULAM_HOME") or "").strip() + return ( + os.path.join(os.path.expanduser(home), "projects") + if home + else os.path.expanduser("~/.bahulam/projects") + ) _PATH_SLOT = { @@ -567,7 +574,8 @@ def _build_store(args: argparse.Namespace, key: str) -> tuple[object, str]: if key == "bahulam": if not _bahulam_available(getattr(args, "bahulam_dir", "")): raise SystemExit( - "No Bahulam Code sessions found. Point --bahulam-dir (or $BAHULAM_PROJECTS_DIR) " + "No Bahulam Code sessions found. Point --bahulam-dir (or $BAHULAM_PROJECTS_DIR / " + "$BAHULAM_HOME) " f"at ~/.bahulam/projects (looked in {getattr(args, 'bahulam_dir', '')})." ) return BahulamStore(args.bahulam_dir, args), "OpenTab: loading Bahulam Code sessions…\r" diff --git a/src/opentab/stores/bahulam.py b/src/opentab/stores/bahulam.py index 0c30611..8d885b2 100644 --- a/src/opentab/stores/bahulam.py +++ b/src/opentab/stores/bahulam.py @@ -17,8 +17,8 @@ class BahulamStore: """Read Bahulam Code transcripts from ``~/.bahulam/projects/**/*.jsonl``. Bahulam records per-turn token usage and total_cost in every ``complete`` - event. The wire format uses ``kepler_event`` or ``bahulam_event`` as the - top-level type; all payload fields live under ``event.data.*``. + event. The wire format uses ``bahulam_event`` as the top-level type; all + payload fields live under ``event.data.*``. """ records_cost = True # Bahulam records total_cost in every usage block @@ -54,6 +54,12 @@ def _new_acc() -> dict: "cache_write": 0, "cache_write_1h": 0, "tokens_total": 0, + "u_input": 0, + "u_output": 0, + "u_reasoning": 0, + "u_cache_read": 0, + "u_cache_write": 0, + "u_cache_write_1h": 0, } @staticmethod @@ -86,29 +92,37 @@ def _primary_model(self, models_map: dict) -> str: return "" @classmethod - def _add_usage(cls, acc: dict, usage: dict) -> None: + def _reported_cost(cls, usage: dict) -> tuple[bool, float]: + """Return ``(has_reported_cost, cost)`` for a raw usage dict.""" + for key in ("cost", "cost_usd", "total_cost", "total_cost_usd"): + value = usage.get(key) + if value in (None, ""): + continue + cost = cls._float(value) + if cost >= 0: + return True, cost + return False, 0.0 + + @classmethod + def _add_usage(cls, acc: dict, usage: dict, *, additive_input: bool = False) -> None: """Accumulate counters from a usage dict into *acc*. Args: acc: Mutable accumulator dict (see ``_new_acc``). usage: Raw usage record from a ``complete`` event. + additive_input: Bahulam ``usage.models[]`` records expose uncached + input beside cache-read/cache-write tokens. Flat aggregate + OpenRouter-style totals expose inclusive input. """ total_in = cls._int(usage.get("total_input_tokens", 0)) out = cls._int(usage.get("total_output_tokens", 0)) cr = cls._int(usage.get("cache_read_input_tokens", 0)) cw = cls._int(usage.get("cache_creation_input_tokens", 0)) reasoning = cls._int(usage.get("reasoning_tokens", 0)) - cost = cls._float( - cls._first_value( - usage.get("cost"), - usage.get("cost_usd"), - usage.get("total_cost"), - usage.get("total_cost_usd"), - ) - ) + has_cost, cost = cls._reported_cost(usage) cc = usage.get("cache_creation") cw1h = cls._int(cc.get("ephemeral_1h_input_tokens", 0) or 0) if isinstance(cc, dict) else 0 - inp = max(0, total_in - cr - cw) + inp = max(0, total_in if additive_input else total_in - cr - cw) acc["runs"] += 1 acc["cost"] += cost acc["input"] += inp @@ -118,6 +132,13 @@ def _add_usage(cls, acc: dict, usage: dict) -> None: acc["cache_write"] += cw acc["cache_write_1h"] += min(cw1h, cw) acc["tokens_total"] += inp + out + reasoning + cr + cw + if not has_cost: + acc["u_input"] += inp + acc["u_output"] += out + acc["u_reasoning"] += reasoning + acc["u_cache_read"] += cr + acc["u_cache_write"] += cw + acc["u_cache_write_1h"] += min(cw1h, cw) @staticmethod def _price(model_name: str, acc: dict) -> float: @@ -140,7 +161,7 @@ def _provider_prefix(model_name: str) -> str: """ fam = model_family(model_name) if not fam or fam == "unknown": - return "anthropic/" # safest fallback for Claude-like naming + return "" return fam + "/" def _qualified_model(self, model_name: str) -> str: @@ -242,9 +263,123 @@ def _new_session() -> dict: "prompts": [], "event_ts": [], "pending_tools": [], # tool names queued before the next complete event + "active_subagents": {}, + "subagent_runs": [], "files": set(), } + @staticmethod + def _is_root_role(role: str) -> bool: + """Return ``True`` when a usage role belongs to the main/root agent.""" + return role in ("", "coder", "main", "executor", "orchestrator") + + def _model_usage(self, usage: dict) -> dict: + """Normalize a per-model or sub-agent usage dict for ``_add_usage``.""" + return { + "total_input_tokens": usage.get("input_tokens", 0), + "total_output_tokens": usage.get("output_tokens", 0), + "cache_read_input_tokens": usage.get("cache_read_tokens", 0), + "cache_creation_input_tokens": usage.get("cache_creation_tokens", 0), + "reasoning_tokens": usage.get("reasoning_tokens", 0) or 0, + } + + @staticmethod + def _usage_cost(usage: dict): + """Return a per-model cost value, preserving missing vs reported zero.""" + return next( + (usage.get(key) for key in ("cost", "cost_usd") if usage.get(key) not in (None, "")), + None, + ) + + def _record_subagent_start(self, s: dict, data: dict, ts: str | None) -> None: + """Remember a sub-agent's start metadata until its complete event arrives.""" + key = data.get("task_id") or f"{data.get('type') or 'subagent'}:{len(s['subagent_runs'])}" + s["active_subagents"][key] = { + "type": data.get("type") or "", + "model": data.get("model") or "", + "query": data.get("query") or "", + "ts": ts or "", + } + + def _record_subagent_complete(self, s: dict, data: dict, ts: str | None) -> None: + """Record one completed Bahulam sub-agent for the Subagents tab.""" + usage = data.get("usage") if isinstance(data.get("usage"), dict) else {} + key = data.get("task_id") or f"{data.get('type') or 'subagent'}:{len(s['subagent_runs'])}" + start = s["active_subagents"].pop(key, {}) + role = str(usage.get("role") or data.get("type") or start.get("type") or "subagent") + model_name = self._qualified_model( + data.get("model") or usage.get("model") or start.get("model") or "" + ) + acc = self._new_acc() + if usage: + normalized = self._model_usage(usage) + cost = self._usage_cost(usage) + if cost is not None: + normalized["cost"] = cost + self._add_usage(acc, normalized, additive_input=True) + title = ( + data.get("result_summary") + or start.get("query") + or data.get("query") + or f"{role} sub-agent" + ) + s["subagent_runs"].append( + { + "id": key, + "role": role, + "agent": role, + "model_name": model_name or "unknown", + "title": _clean_prompt(str(title)), + "ts": start.get("ts") or ts or "", + "acc": acc, + "cost_assigned": self._usage_cost(usage) is not None, + } + ) + + def _assign_subagent_cost( + self, s: dict, role: str, model_name: str, usage: dict, ts: str | None + ) -> None: + """Attach aggregate per-role cost from ``complete`` to matching sub-agent runs.""" + has_cost, cost = self._reported_cost(usage) + if not has_cost: + return + matches = [ + r + for r in s["subagent_runs"] + if not r.get("cost_assigned") + and r.get("role") == role + and r.get("model_name") == model_name + ] + if not matches: + already_recorded = any( + r.get("role") == role and r.get("model_name") == model_name + for r in s["subagent_runs"] + ) + if already_recorded: + return + acc = self._new_acc() + self._add_usage(acc, usage, additive_input=True) + acc["cost"] = cost + matches = [ + { + "id": f"{role}:{len(s['subagent_runs'])}", + "role": role, + "agent": role, + "model_name": model_name or "unknown", + "title": f"{role} sub-agent", + "ts": ts or "", + "acc": acc, + "cost_assigned": True, + } + ] + s["subagent_runs"].extend(matches) + return + denom = sum(r["acc"]["tokens_total"] for r in matches) + for r in matches: + share = cost / len(matches) if denom <= 0 else cost * r["acc"]["tokens_total"] / denom + r["acc"]["cost"] += share + r["cost_assigned"] = True + def _parse_file(self, text: str, session_id: str, s: dict, path: str = "") -> None: """Decode a single JSONL file body and ingest every line into *s*.""" for line in text.split("\n"): @@ -262,8 +397,8 @@ def _parse_file(self, text: str, session_id: str, s: dict, path: str = "") -> No def _ingest(self, o: dict, s: dict, path: str = "") -> None: """Ingest a single JSON record *o* into the session dict *s*. - Handles user messages, ``bahulam_event`` / ``kepler_event`` records - (session_info, complete, tool_call), and extracts timestamps / cwd. + Handles user messages and ``bahulam_event`` records (session_info, + complete, tool_call), and extracts timestamps / cwd. """ if path: s["files"].add(path) @@ -301,9 +436,8 @@ def _ingest(self, o: dict, s: dict, path: str = "") -> None: if text: s["prompts"].append({"ts": ts or "", "title": text}) - # Event records — both ``bahulam_event`` and ``kepler_event`` carry - # their payload under ``event.data.*``. - if typ not in ("bahulam_event", "kepler_event"): + # Event records carry their payload under ``event.data.*``. + if typ != "bahulam_event": return event = o.get("event") if not isinstance(event, dict): @@ -327,6 +461,12 @@ def _ingest(self, o: dict, s: dict, path: str = "") -> None: {"total": self._new_acc(), "root": self._new_acc()}, ) + elif event_type == "sub_agent_start": + self._record_subagent_start(s, data, ts) + + elif event_type == "sub_agent_complete": + self._record_subagent_complete(s, data, ts) + # complete — carries per-LLM-step token usage and cost elif event_type == "complete": usage = data.get("usage") or {} @@ -351,16 +491,16 @@ def _ingest(self, o: dict, s: dict, path: str = "") -> None: "total": self._new_acc(), "root": self._new_acc(), } - per_model = { - "total_input_tokens": m.get("input_tokens", 0), - "total_output_tokens": m.get("output_tokens", 0), - "cache_read_input_tokens": m.get("cache_read_tokens", 0), - "cache_creation_input_tokens": m.get("cache_creation_tokens", 0), - "reasoning_tokens": m.get("reasoning_tokens", 0) or 0, - "cost": self._first_value(m.get("cost"), m.get("cost_usd")), - } - self._add_usage(entry["total"], per_model) - self._add_usage(entry["root"], per_model) + per_model = self._model_usage(m) + cost = self._usage_cost(m) + if cost is not None: + per_model["cost"] = cost + self._add_usage(entry["total"], per_model, additive_input=True) + role = str(m.get("role") or "") + if self._is_root_role(role): + self._add_usage(entry["root"], per_model, additive_input=True) + else: + self._assign_subagent_cost(s, role, model_name, per_model, ts) else: model_name = self._qualified_model( data.get("model") or usage.get("model") or s["model"] or "" @@ -381,14 +521,7 @@ def _ingest(self, o: dict, s: dict, path: str = "") -> None: cr = self._int(usage.get("cache_read_input_tokens", 0)) cw = self._int(usage.get("cache_creation_input_tokens", 0)) inp = max(0, total_in - cr - cw) - total_cost = self._float( - self._first_value( - usage.get("total_cost"), - usage.get("total_cost_usd"), - usage.get("cost"), - usage.get("cost_usd"), - ) - ) + _has_total_cost, total_cost = self._reported_cost(usage) # Turn row for the Turns tab first_model = "unknown" @@ -456,24 +589,51 @@ def _finalize(self, sid: str, s: dict) -> None: "cache_write": tot["cache_write"], "cache_write_1h": tot["cache_write_1h"], "output": tot["output"], - # 1h-TTL subset for API-equivalent estimation - "unpriced_input": tot["input"], - "unpriced_reasoning": tot["reasoning"], - "unpriced_cache_read": tot["cache_read"], - "unpriced_cache_write": tot["cache_write"], - "unpriced_cache_write_1h": tot["cache_write_1h"], - "unpriced_output": tot["output"], - "root_unpriced_input": root["input"], - "root_unpriced_reasoning": root["reasoning"], - "root_unpriced_cache_read": root["cache_read"], - "root_unpriced_cache_write": root["cache_write"], - "root_unpriced_cache_write_1h": root["cache_write_1h"], - "root_unpriced_output": root["output"], + "unpriced_input": tot["u_input"], + "unpriced_reasoning": tot["u_reasoning"], + "unpriced_cache_read": tot["u_cache_read"], + "unpriced_cache_write": tot["u_cache_write"], + "unpriced_cache_write_1h": tot["u_cache_write_1h"], + "unpriced_output": tot["u_output"], + "root_unpriced_input": root["u_input"], + "root_unpriced_reasoning": root["u_reasoning"], + "root_unpriced_cache_read": root["u_cache_read"], + "root_unpriced_cache_write": root["u_cache_write"], + "root_unpriced_cache_write_1h": root["u_cache_write_1h"], + "root_unpriced_output": root["u_output"], } ) s["model_rows"] = rows - s["unpriced_tokens"] = sum(r["tokens_total"] for r in rows) - s["subagents"] = [] + s["unpriced_tokens"] = sum( + r["unpriced_input"] + + r["unpriced_output"] + + r["unpriced_reasoning"] + + r["unpriced_cache_read"] + + r["unpriced_cache_write"] + for r in rows + ) + s["subagents"] = self._build_subagents(s) + + def _build_subagents(self, s: dict) -> list[dict]: + """Build depth-1 nodes from Bahulam sub-agent completion events.""" + nodes = [] + for idx, run in enumerate(s["subagent_runs"], start=1): + acc = run["acc"] + if acc["tokens_total"] <= 0 and acc["cost"] <= 0: + continue + nodes.append( + self._node( + str(run.get("id") or f"subagent-{idx}"), + 1, + str(run.get("agent") or "subagent"), + str(run.get("title") or "sub-agent run"), + iso_to_local(run.get("ts")) if run.get("ts") else s["created_at"], + str(run.get("model_name") or "unknown"), + acc["cost"], + acc, + ) + ) + return nodes @staticmethod def _node( diff --git a/src/opentab/stores/cached.py b/src/opentab/stores/cached.py index ac3b6f8..1a5577c 100644 --- a/src/opentab/stores/cached.py +++ b/src/opentab/stores/cached.py @@ -16,7 +16,7 @@ from opentab import paths from opentab.models import Workflow -CACHE_VERSION = 10 # bump when the cached payload shape or meaning changes +CACHE_VERSION = 11 # bump when the cached payload shape or meaning changes # Required because cache readers index these fields directly. diff --git a/tests/test_sources.py b/tests/test_sources.py index 3e970bc..1efbae7 100644 --- a/tests/test_sources.py +++ b/tests/test_sources.py @@ -189,3 +189,12 @@ def test_a_real_opencode_db_is_still_detected_and_opened(): assert "opencode" in ot.sources.available_sources(_parse(["--db", db])) store = ot.sources.make_store(_parse(["--db", db, "--source", "opencode"]), "opencode")[0] assert store.workflows() == [] + + +def test_bahulam_default_dir_honors_current_home_env(monkeypatch): + monkeypatch.delenv("BAHULAM_PROJECTS_DIR", raising=False) + monkeypatch.setenv("BAHULAM_HOME", "/tmp/bahulam-home") + assert ot.sources._default_bahulam_dir() == "/tmp/bahulam-home/projects" + + monkeypatch.setenv("BAHULAM_PROJECTS_DIR", "/tmp/bahulam-projects") + assert ot.sources._default_bahulam_dir() == "/tmp/bahulam-projects" diff --git a/tests/test_stores_bahulam.py b/tests/test_stores_bahulam.py index 6936491..183e873 100644 --- a/tests/test_stores_bahulam.py +++ b/tests/test_stores_bahulam.py @@ -6,9 +6,9 @@ from tests._support import _jsonl_args, _write_jsonl -def _bahulam_event(event_type, data=None, *, record_type="bahulam_event", cwd="/work/repo", ts=None): +def _bahulam_event(event_type, data=None, *, cwd="/work/repo", ts=None): return { - "type": record_type, + "type": "bahulam_event", "timestamp": ts or "2026-08-29T10:00:00.000Z", "cwd": cwd, "event": {"type": event_type, "data": data or {}}, @@ -30,7 +30,7 @@ def _write_bahulam(root, sid, rows): _write_jsonl(os.path.join(project, f"{sid}.jsonl"), rows) -def test_bahulam_store_reads_current_and_legacy_event_shapes_for_usage_and_tools(): +def test_bahulam_store_reads_current_event_shape_for_usage_and_tools(): sid = "0198fca0-34b4-7285-b3f0-3b8fb0489e5f" with tempfile.TemporaryDirectory() as tmp: cwd = os.path.join(tmp, "repo") @@ -45,7 +45,6 @@ def test_bahulam_store_reads_current_and_legacy_event_shapes_for_usage_and_tools "coder": "xiaomi/mimo-v2.5", } }, - record_type="kepler_event", cwd=cwd, ), _bahulam_event("tool_request", {"name": "read_file"}, cwd=cwd), @@ -90,7 +89,8 @@ def test_bahulam_store_reads_current_and_legacy_event_shapes_for_usage_and_tools assert workflow.id == sid assert workflow.title == "inspect this" assert workflow.total_cost == 0.25 - assert workflow.total_tokens == 1110 + assert workflow.total_tokens == 1460 + assert workflow.unpriced_tokens == 0 timeline = store.message_timeline(sid) assert len(timeline) == 1 @@ -109,11 +109,116 @@ def test_bahulam_store_reads_current_and_legacy_event_shapes_for_usage_and_tools by_model = {row["model_name"]: row for row in store.model_breakdown()} assert by_model["xiaomi/mimo-v2.5"]["reasoning"] == 7 + assert by_model["xiaomi/mimo-v2.5"]["input"] == 700 + assert by_model["xiaomi/mimo-v2.5"]["cache_read"] == 250 assert by_model["xiaomi/mimo-v2.5"]["cost"] == 0.20 + assert by_model["xiaomi/mimo-v2.5"]["unpriced_input"] == 0 assert by_model["deepseek/deepseek-v4-flash"]["reasoning"] == 3 + assert by_model["deepseek/deepseek-v4-flash"]["input"] == 300 assert by_model["deepseek/deepseek-v4-flash"]["cost"] == 0.05 +def test_bahulam_store_counts_subagents_and_splits_root_cost(): + sid = "0198fca0-34b4-7285-b3f0-3b8fb0489e62" + with tempfile.TemporaryDirectory() as tmp: + cwd = os.path.join(tmp, "repo") + os.makedirs(os.path.join(cwd, ".git")) + rows = [ + _bahulam_user("plan this", cwd=cwd), + _bahulam_event( + "session_info", + { + "models": { + "planning": "deepseek/deepseek-v4-pro", + "coder": "xiaomi/mimo-v2.5", + } + }, + cwd=cwd, + ), + _bahulam_event( + "sub_agent_start", + { + "type": "plan", + "model": "deepseek/deepseek-v4-pro", + "query": "create the implementation plan", + "task_id": "task-plan-1", + }, + cwd=cwd, + ), + _bahulam_event( + "sub_agent_complete", + { + "type": "plan", + "model": "deepseek/deepseek-v4-pro", + "success": True, + "duration_s": 12.5, + "tool_calls": 3, + "task_id": "task-plan-1", + "usage": { + "model": "deepseek/deepseek-v4-pro", + "role": "plan", + "input_tokens": 100, + "output_tokens": 20, + "cache_read_tokens": 30, + "cache_creation_tokens": 0, + }, + }, + cwd=cwd, + ), + _bahulam_event( + "complete", + { + "usage": { + "total_input_tokens": 300, + "total_output_tokens": 30, + "cache_read_input_tokens": 70, + "total_cost_usd": 0.05, + "models": [ + { + "model": "xiaomi/mimo-v2.5", + "role": "coder", + "input_tokens": 200, + "output_tokens": 10, + "cache_read_tokens": 40, + "cost_usd": 0.02, + }, + { + "model": "deepseek/deepseek-v4-pro", + "role": "plan", + "input_tokens": 100, + "output_tokens": 20, + "cache_read_tokens": 30, + "cost_usd": 0.03, + }, + ], + }, + "primary_tool_calls": 1, + "sub_agent_tool_calls": 3, + }, + cwd=cwd, + ), + ] + _write_bahulam(tmp, sid, rows) + + store = ot.BahulamStore(os.path.join(tmp, "projects"), _jsonl_args()) + workflow = store.workflows()[0] + assert workflow.subagents == 1 + assert workflow.root_cost == 0.02 + assert workflow.total_cost == 0.05 + assert workflow.total_tokens == 400 + + nodes = store.workflow_nodes(sid) + assert len(nodes) == 2 + assert nodes[0]["depth"] == 0 + assert nodes[0]["cost"] == 0.02 + assert nodes[0]["tokens_total"] == 250 + assert nodes[1]["depth"] == 1 + assert nodes[1]["agent"] == "plan" + assert nodes[1]["model_name"] == "deepseek/deepseek-v4-pro" + assert nodes[1]["cost"] == 0.03 + assert nodes[1]["tokens_total"] == 150 + + def test_bahulam_store_uses_coder_model_when_usage_has_no_model_breakdown(): sid = "0198fca0-34b4-7285-b3f0-3b8fb0489e60" with tempfile.TemporaryDirectory() as tmp: @@ -156,3 +261,49 @@ def test_bahulam_store_uses_coder_model_when_usage_has_no_model_breakdown(): assert by_model["xiaomi/mimo-v2.5"]["cost"] == 0.03 assert by_model["xiaomi/mimo-v2.5"]["tokens_total"] == 145 assert by_model["deepseek/deepseek-v4-pro"]["runs"] == 0 + + +def test_bahulam_missing_cost_stays_unpriced_and_bare_unknown_model_stays_bare(): + sid = "0198fca0-34b4-7285-b3f0-3b8fb0489e61" + with tempfile.TemporaryDirectory() as tmp: + cwd = os.path.join(tmp, "repo") + os.makedirs(cwd) + rows = [ + _bahulam_user("hello", cwd=cwd), + _bahulam_event( + "session_info", + {"models": {"coder": "local-vision-test"}}, + cwd=cwd, + ), + _bahulam_event( + "complete", + { + "usage": { + "total_input_tokens": 10, + "total_output_tokens": 5, + "cache_read_input_tokens": 2, + "models": [ + { + "model": "local-vision-test", + "input_tokens": 10, + "output_tokens": 5, + "cache_read_tokens": 2, + } + ], + } + }, + cwd=cwd, + ), + ] + _write_bahulam(tmp, sid, rows) + + store = ot.BahulamStore(os.path.join(tmp, "projects"), _jsonl_args()) + workflow = store.workflows()[0] + assert workflow.total_cost == 0 + assert workflow.total_tokens == workflow.unpriced_tokens == 17 + + by_model = {row["model_name"]: row for row in store.model_breakdown()} + assert "local-vision-test" in by_model + assert "anthropic/local-vision-test" not in by_model + assert by_model["local-vision-test"]["unpriced_input"] == 10 + assert by_model["local-vision-test"]["unpriced_cache_read"] == 2 From edb5c48df8c350456d6d97b5a7ad351ac04360f6 Mon Sep 17 00:00:00 2001 From: admin bi migration Date: Mon, 31 Aug 2026 18:42:03 -0400 Subject: [PATCH 4/5] Address PR #17 review: cache double-count, hosted billing, resolver, custom-runner test Six specific issues raised by @hamidi-dev: 1. models[].input_tokens is INCLUSIVE of cache_read (not exclusive). The previous additive_input=True path double-counted cache reads, inflating workflow.total_tokens ~2x. Removed the flag; every code path now subtracts cache_read + cache_creation from total_input to derive fresh input. Verified against the reviewer's transcript: 4,717,485 -> 2,554,541 tokens (matches their 2,047,863 coder/root + 506,678 plan subagent split exactly). 2. Hosted sessions (is_byok=false) reported provider list cost as user spend. Now: track is_byok from session_info + credits_charged from complete events; hosted sessions surface credits_charged as total_cost and route provider cost into unpriced buckets. BYOK and legacy transcripts (no is_byok) keep the previous provider-cost behavior. The reviewer's $0.127644 provider figure correctly shows as $0 user spend. 3. test_bahulam_default_dir_honors_current_home_env used pytest's monkeypatch fixture, which the project's custom run_tests.py does not inject (1187/1188 -> 1317/1317). Rewrote with stdlib env save/restore. 4. Resolver honors $KEPLER_HOME and falls back to ~/.kepler/projects when it exists on disk, smoothing the Kepler-to-Bahulam migration for users upgrading in place. 5. Bahulam `complete` events aggregate multiple internal LLM calls (plus sub-agent rollups) into one turn row, so tokens_total is not a per-request context snapshot. Added supports_context_curve() returning False to opt out of context-curve rendering. 6. Rebased onto origin/main; parallel-add conflicts against upstream's antigravity + gemini backends resolved by keeping both sides. Tests: 4 new (hosted-with-$0, hosted-with-charge, byok, context-curve) and 3 existing updated for corrected math. 1317/1317 passing under python3 run_tests.py. Note for reviewers with a pre-fix cache: OpenTab persists parsed rollups keyed by transcript (path, size, mtime_ns), so a cache written before this PR will keep serving the double-counted numbers until the transcript files change. One-time flush: opentab tui --source bahulam --no-cache # or rm -f ~/.cache/opentab/cache/bahulam-*.json We did not bump CACHE_VERSION because it is shared across every backend and this is a Bahulam-only correction. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/opentab/sources.py | 33 ++++-- src/opentab/stores/bahulam.py | 118 ++++++++++++++++----- tests/test_cli.py | 5 +- tests/test_sources.py | 80 ++++++++++++-- tests/test_stores_bahulam.py | 193 ++++++++++++++++++++++++++++++++-- 5 files changed, 378 insertions(+), 51 deletions(-) diff --git a/src/opentab/sources.py b/src/opentab/sources.py index fe86b0d..d0bdec1 100644 --- a/src/opentab/sources.py +++ b/src/opentab/sources.py @@ -104,16 +104,29 @@ def _default_zaly_dir() -> str: def _default_bahulam_dir() -> str: - """Return the default Bahulam Code projects directory.""" - env = (os.environ.get("BAHULAM_PROJECTS_DIR") or "").strip() - if env: - return env - home = (os.environ.get("BAHULAM_HOME") or "").strip() - return ( - os.path.join(os.path.expanduser(home), "projects") - if home - else os.path.expanduser("~/.bahulam/projects") - ) + """Return the default Bahulam Code projects directory. + + Resolution order (first match wins, empty values skipped): + 1. ``$BAHULAM_PROJECTS_DIR`` — explicit override + 2. ``$BAHULAM_HOME/projects`` — Bahulam home relocated + 3. ``$KEPLER_HOME/projects`` — legacy alias from the Kepler-branded builds + 4. ``~/.bahulam/projects`` — Bahulam default + 5. ``~/.kepler/projects`` — legacy default, only if it exists on disk + (skipping this last-resort check would silently point at a phantom + path when neither install layout is present) + """ + override = (os.environ.get("BAHULAM_PROJECTS_DIR") or "").strip() + if override: + return override + for env_name in ("BAHULAM_HOME", "KEPLER_HOME"): + home = (os.environ.get(env_name) or "").strip() + if home: + return os.path.join(os.path.expanduser(home), "projects") + bahulam_default = os.path.expanduser("~/.bahulam/projects") + kepler_legacy = os.path.expanduser("~/.kepler/projects") + if not os.path.isdir(bahulam_default) and os.path.isdir(kepler_legacy): + return kepler_legacy + return bahulam_default _PATH_SLOT = { diff --git a/src/opentab/stores/bahulam.py b/src/opentab/stores/bahulam.py index 8d885b2..4914409 100644 --- a/src/opentab/stores/bahulam.py +++ b/src/opentab/stores/bahulam.py @@ -104,15 +104,20 @@ def _reported_cost(cls, usage: dict) -> tuple[bool, float]: return False, 0.0 @classmethod - def _add_usage(cls, acc: dict, usage: dict, *, additive_input: bool = False) -> None: + def _add_usage(cls, acc: dict, usage: dict, *, hosted_unpriced: bool = False) -> None: """Accumulate counters from a usage dict into *acc*. - Args: - acc: Mutable accumulator dict (see ``_new_acc``). - usage: Raw usage record from a ``complete`` event. - additive_input: Bahulam ``usage.models[]`` records expose uncached - input beside cache-read/cache-write tokens. Flat aggregate - OpenRouter-style totals expose inclusive input. + Every Bahulam usage shape — top-level aggregate, per-model row, or + sub-agent rollup — reports ``input_tokens``/``total_input_tokens`` + **inclusive of cache reads and cache creations**, matching the + anthropic-style convention. The previous ``additive_input`` flag + double-counted cache reads by ~2× on models[] rows; removed. + + hosted_unpriced=True: the parent session is Bahulam-hosted and the + user was not charged (`is_byok=false` and `credits_charged=0`). The + provider list cost is still available on the raw event but is NOT + user spend, so we drop it into the unpriced buckets to keep the + cost column honest. Metered/BYOK sessions keep the provider cost. """ total_in = cls._int(usage.get("total_input_tokens", 0)) out = cls._int(usage.get("total_output_tokens", 0)) @@ -122,9 +127,8 @@ def _add_usage(cls, acc: dict, usage: dict, *, additive_input: bool = False) -> has_cost, cost = cls._reported_cost(usage) cc = usage.get("cache_creation") cw1h = cls._int(cc.get("ephemeral_1h_input_tokens", 0) or 0) if isinstance(cc, dict) else 0 - inp = max(0, total_in if additive_input else total_in - cr - cw) + inp = max(0, total_in - cr - cw) acc["runs"] += 1 - acc["cost"] += cost acc["input"] += inp acc["output"] += out acc["reasoning"] += reasoning @@ -132,7 +136,10 @@ def _add_usage(cls, acc: dict, usage: dict, *, additive_input: bool = False) -> acc["cache_write"] += cw acc["cache_write_1h"] += min(cw1h, cw) acc["tokens_total"] += inp + out + reasoning + cr + cw - if not has_cost: + if has_cost and not hosted_unpriced: + acc["cost"] += cost + else: + # Missing cost OR hosted-unpriced: bucket tokens as unpriced. acc["u_input"] += inp acc["u_output"] += out acc["u_reasoning"] += reasoning @@ -266,6 +273,20 @@ def _new_session() -> dict: "active_subagents": {}, "subagent_runs": [], "files": set(), + # Billing split — Bahulam ships two modes: + # is_byok=True (BYOK): user pays the LLM provider directly. + # provider `cost` == real user spend. + # is_byok=False (hosted): Bahulam pays the provider and charges + # the user via `credits_charged` (often + # $0 on free-tier or promo). Provider + # `cost` is Bahulam's COGS, NOT user + # spend, and must be treated as unpriced + # in the normal cost column. + # is_byok=None (unknown / legacy transcripts): fall back to + # provider cost so old transcripts still + # render sensibly. + "is_byok": None, + "credits_charged": 0.0, } @staticmethod @@ -311,12 +332,13 @@ def _record_subagent_complete(self, s: dict, data: dict, ts: str | None) -> None data.get("model") or usage.get("model") or start.get("model") or "" ) acc = self._new_acc() + hosted_unpriced = s["is_byok"] is False if usage: normalized = self._model_usage(usage) cost = self._usage_cost(usage) if cost is not None: normalized["cost"] = cost - self._add_usage(acc, normalized, additive_input=True) + self._add_usage(acc, normalized, hosted_unpriced=hosted_unpriced) title = ( data.get("result_summary") or start.get("query") @@ -332,14 +354,24 @@ def _record_subagent_complete(self, s: dict, data: dict, ts: str | None) -> None "title": _clean_prompt(str(title)), "ts": start.get("ts") or ts or "", "acc": acc, - "cost_assigned": self._usage_cost(usage) is not None, + # Hosted sessions never have provider cost as user spend, so + # mark cost_assigned=True to short-circuit later distribution + # attempts even when the raw event carried a provider cost. + "cost_assigned": hosted_unpriced or self._usage_cost(usage) is not None, } ) def _assign_subagent_cost( - self, s: dict, role: str, model_name: str, usage: dict, ts: str | None + self, s: dict, role: str, model_name: str, usage: dict, ts: str | None, + *, hosted_unpriced: bool = False, ) -> None: - """Attach aggregate per-role cost from ``complete`` to matching sub-agent runs.""" + """Attach aggregate per-role cost from ``complete`` to matching sub-agent runs. + + Skipped entirely for hosted sessions — provider cost is not user + spend, so there is nothing to distribute. + """ + if hosted_unpriced: + return has_cost, cost = self._reported_cost(usage) if not has_cost: return @@ -358,7 +390,7 @@ def _assign_subagent_cost( if already_recorded: return acc = self._new_acc() - self._add_usage(acc, usage, additive_input=True) + self._add_usage(acc, usage) acc["cost"] = cost matches = [ { @@ -446,8 +478,10 @@ def _ingest(self, o: dict, s: dict, path: str = "") -> None: event_type = event.get("type") data = event.get("data") if isinstance(event.get("data"), dict) else {} - # session_info — carries the model map + # session_info — carries the model map + billing mode if event_type == "session_info": + if "is_byok" in data and s["is_byok"] is None: + s["is_byok"] = bool(data.get("is_byok")) models_map = data.get("models") if isinstance(models_map, dict): primary = self._primary_model(models_map) @@ -473,10 +507,21 @@ def _ingest(self, o: dict, s: dict, path: str = "") -> None: if not usage: return + # Hosted sessions expose provider list cost via `usage.cost*` but + # bill the user via `data.credits_charged` (often 0). Only when + # is_byok is EXPLICITLY False do we treat provider cost as + # unpriced — unknown/legacy transcripts keep prior behavior. + hosted_unpriced = s["is_byok"] is False + credits_charged = self._float( + data.get("credits_charged", usage.get("credits_charged", 0)) + ) + s["credits_charged"] += credits_charged + reasoning = self._int(usage.get("reasoning_tokens", 0)) # Per-model token breakdown (the wire uses ``cache_read_tokens`` - # and ``cache_creation_tokens`` at per-model granularity). + # and ``cache_creation_tokens`` at per-model granularity, with + # ``input_tokens`` INCLUSIVE of cache — matches anthropic). models_usage = usage.get("models") if isinstance(models_usage, list) and models_usage: for m in models_usage: @@ -495,12 +540,14 @@ def _ingest(self, o: dict, s: dict, path: str = "") -> None: cost = self._usage_cost(m) if cost is not None: per_model["cost"] = cost - self._add_usage(entry["total"], per_model, additive_input=True) + self._add_usage(entry["total"], per_model, hosted_unpriced=hosted_unpriced) role = str(m.get("role") or "") if self._is_root_role(role): - self._add_usage(entry["root"], per_model, additive_input=True) + self._add_usage(entry["root"], per_model, hosted_unpriced=hosted_unpriced) else: - self._assign_subagent_cost(s, role, model_name, per_model, ts) + self._assign_subagent_cost( + s, role, model_name, per_model, ts, hosted_unpriced=hosted_unpriced, + ) else: model_name = self._qualified_model( data.get("model") or usage.get("model") or s["model"] or "" @@ -512,8 +559,8 @@ def _ingest(self, o: dict, s: dict, path: str = "") -> None: "total": self._new_acc(), "root": self._new_acc(), } - self._add_usage(entry["total"], usage) - self._add_usage(entry["root"], usage) + self._add_usage(entry["total"], usage, hosted_unpriced=hosted_unpriced) + self._add_usage(entry["root"], usage, hosted_unpriced=hosted_unpriced) # Aggregate totals for the turn row total_in = self._int(usage.get("total_input_tokens", 0)) @@ -521,7 +568,10 @@ def _ingest(self, o: dict, s: dict, path: str = "") -> None: cr = self._int(usage.get("cache_read_input_tokens", 0)) cw = self._int(usage.get("cache_creation_input_tokens", 0)) inp = max(0, total_in - cr - cw) - _has_total_cost, total_cost = self._reported_cost(usage) + _has_total_cost, provider_cost = self._reported_cost(usage) + # Hosted turn: user paid credits_charged for THIS turn; provider + # cost is Bahulam COGS. BYOK / unknown: keep provider cost. + turn_cost = credits_charged if hosted_unpriced else provider_cost # Turn row for the Turns tab first_model = "unknown" @@ -539,7 +589,7 @@ def _ingest(self, o: dict, s: dict, path: str = "") -> None: "agent": "-", "effort": "", "model_name": first_model, - "cost": float(total_cost), + "cost": float(turn_cost), "input": inp, "output": out, "reasoning": reasoning, @@ -692,12 +742,19 @@ def sort_workflows(rows: list[Workflow]) -> list[Workflow]: return rows def _workflow_rows(self, sessions: dict[str, dict]) -> list[Workflow]: - """Convert session dicts into a sorted list of ``Workflow`` rows.""" + """Convert session dicts into a sorted list of ``Workflow`` rows. + + Hosted sessions contribute their `credits_charged` sum as the user- + facing cost; BYOK/unknown sessions contribute the model_rows cost + (which is the provider cost). Both branches produce a single + `total_cost` column so the UI can display them uniformly. + """ rows = [] for sid, s in sessions.items(): model_rows = s["model_rows"] - total_cost = sum(r["cost"] for r in model_rows) + model_cost = sum(r["cost"] for r in model_rows) root_cost = sum(r["root_cost"] for r in model_rows) + total_cost = model_cost + float(s.get("credits_charged", 0.0)) rows.append( Workflow( id=sid, @@ -807,6 +864,15 @@ def supports_turns(self, workflow_id: str) -> bool: """Indicate whether turn detail is available (always ``True``).""" return True + def supports_context_curve(self, workflow_id: str) -> bool: + """Bahulam ``complete`` events aggregate multiple internal LLM calls + (and often sub-agent rollups) into a single turn row. The + ``tokens_total`` on a turn is therefore not a per-request context + window snapshot, and rendering it as a context curve would mislead. + Opt out globally until Bahulam ships a per-request telemetry stream. + """ + return False + def tool_breakdown(self, workflow_id: str) -> list[dict]: """Return tool call rows for the tools tab.""" s = self._session(workflow_id) diff --git a/tests/test_cli.py b/tests/test_cli.py index e47ad3d..a570996 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -768,7 +768,10 @@ def test_verb_help_is_focused_but_globals_still_parse(): tui_help = _subparser_help("tui") assert "--claude-dir" in tui_help and "--theme" in tui_help and "--web" in tui_help compact_tui_help = " ".join(tui_help.split()) - assert "zaly · bahulam · all (merged)" in compact_tui_help + # Post-rebase onto main, gemini + antigravity land between zaly and + # bahulam in cli.py's argparse choices; keep the assertion aligned with + # that order so the help text doesn't drift out from under the test. + assert "zaly · gemini · antigravity · bahulam · all (merged)" in compact_tui_help assert "Zaly, Bahulam Code" in compact_tui_help # Hidden != gone: a suppressed global still parses on that verb. assert _parse(["pull", "--no-cache", "host"]).no_cache is True diff --git a/tests/test_sources.py b/tests/test_sources.py index 1efbae7..8772410 100644 --- a/tests/test_sources.py +++ b/tests/test_sources.py @@ -191,10 +191,78 @@ def test_a_real_opencode_db_is_still_detected_and_opened(): assert store.workflows() == [] -def test_bahulam_default_dir_honors_current_home_env(monkeypatch): - monkeypatch.delenv("BAHULAM_PROJECTS_DIR", raising=False) - monkeypatch.setenv("BAHULAM_HOME", "/tmp/bahulam-home") - assert ot.sources._default_bahulam_dir() == "/tmp/bahulam-home/projects" +def _env_snapshot(names): + """Save and clear the named env vars; return a restore callable. - monkeypatch.setenv("BAHULAM_PROJECTS_DIR", "/tmp/bahulam-projects") - assert ot.sources._default_bahulam_dir() == "/tmp/bahulam-projects" + Rewritten from a pytest-only ``monkeypatch`` fixture to a stdlib helper + so the project's custom ``run_tests.py`` (which does not inject + fixtures) picks the test up — previously it silently regressed the + suite to 1187/1188. + """ + saved = {n: os.environ.get(n) for n in names} + for n in names: + os.environ.pop(n, None) + + def restore(): + for n, v in saved.items(): + if v is None: + os.environ.pop(n, None) + else: + os.environ[n] = v + + return restore + + +def test_bahulam_default_dir_honors_current_home_env(): + restore = _env_snapshot(["BAHULAM_PROJECTS_DIR", "BAHULAM_HOME", "KEPLER_HOME"]) + try: + os.environ["BAHULAM_HOME"] = "/tmp/bahulam-home" + assert ot.sources._default_bahulam_dir() == "/tmp/bahulam-home/projects" + + os.environ["BAHULAM_PROJECTS_DIR"] = "/tmp/bahulam-projects" + assert ot.sources._default_bahulam_dir() == "/tmp/bahulam-projects" + finally: + restore() + + +def test_bahulam_default_dir_falls_back_to_kepler_home_env(): + """Legacy Kepler-branded builds shipped with ``$KEPLER_HOME``; the + resolver honors it when the Bahulam-named vars are absent, since users + upgrading from Kepler frequently keep the older env exported.""" + restore = _env_snapshot(["BAHULAM_PROJECTS_DIR", "BAHULAM_HOME", "KEPLER_HOME"]) + try: + os.environ["KEPLER_HOME"] = "/tmp/kepler-home" + assert ot.sources._default_bahulam_dir() == "/tmp/kepler-home/projects" + + # BAHULAM_HOME wins over KEPLER_HOME when both are set. + os.environ["BAHULAM_HOME"] = "/tmp/bahulam-home" + assert ot.sources._default_bahulam_dir() == "/tmp/bahulam-home/projects" + finally: + restore() + + +def test_bahulam_default_dir_falls_back_to_legacy_kepler_projects_dir(): + """When neither env is set and ``~/.bahulam/projects`` does not exist, + the resolver falls back to ``~/.kepler/projects`` if that directory is + present — smoothing the migration for users upgrading in place.""" + restore = _env_snapshot(["BAHULAM_PROJECTS_DIR", "BAHULAM_HOME", "KEPLER_HOME"]) + with tempfile.TemporaryDirectory() as tmp: + original_home = os.environ.get("HOME") + os.environ["HOME"] = tmp + try: + # Neither dir exists → default to Bahulam path (nonexistent is fine). + assert ot.sources._default_bahulam_dir() == os.path.join(tmp, ".bahulam", "projects") + + # Only Kepler exists → resolver picks it up. + os.makedirs(os.path.join(tmp, ".kepler", "projects")) + assert ot.sources._default_bahulam_dir() == os.path.join(tmp, ".kepler", "projects") + + # Once Bahulam exists too, Bahulam wins. + os.makedirs(os.path.join(tmp, ".bahulam", "projects")) + assert ot.sources._default_bahulam_dir() == os.path.join(tmp, ".bahulam", "projects") + finally: + if original_home is None: + os.environ.pop("HOME", None) + else: + os.environ["HOME"] = original_home + restore() diff --git a/tests/test_stores_bahulam.py b/tests/test_stores_bahulam.py index 183e873..b642534 100644 --- a/tests/test_stores_bahulam.py +++ b/tests/test_stores_bahulam.py @@ -84,12 +84,21 @@ def test_bahulam_store_reads_current_event_shape_for_usage_and_tools(): ] _write_bahulam(tmp, sid, rows) + # Bahulam wire convention (matches anthropic): every ``input_tokens`` + # field is INCLUSIVE of cache_read + cache_creation. Fresh input is + # therefore ``input_tokens - cache_read - cache_creation``. + # + # xiaomi: 700 - 250 - 50 = 400 fresh input; tokens_total = 787 + # (400 + 80 out + 7 reason + 250 cr + 50 cw) + # deepseek: 300 - 50 - 0 = 250 fresh input; tokens_total = 323 + # (250 + 20 out + 3 reason + 50 cr + 0 cw) + # session: 787 + 323 = 1110 store = ot.BahulamStore(os.path.join(tmp, "projects"), _jsonl_args()) workflow = store.workflows()[0] assert workflow.id == sid assert workflow.title == "inspect this" assert workflow.total_cost == 0.25 - assert workflow.total_tokens == 1460 + assert workflow.total_tokens == 1110 assert workflow.unpriced_tokens == 0 timeline = store.message_timeline(sid) @@ -105,16 +114,19 @@ def test_bahulam_store_reads_current_event_shape_for_usage_and_tools(): assert tools[0]["tool"] == "read_file" assert tools[0]["model_name"] == "xiaomi/mimo-v2.5" assert tools[0]["calls"] == 1 + # tools row uses the aggregate turn total (fresh input from the + # top-level usage dict), not the per-model sum, so this stays at: + # 1000-300-50 + 100 + 20 + 300 + 50 = 1120 assert tools[0]["tokens_total"] == 1120 by_model = {row["model_name"]: row for row in store.model_breakdown()} assert by_model["xiaomi/mimo-v2.5"]["reasoning"] == 7 - assert by_model["xiaomi/mimo-v2.5"]["input"] == 700 + assert by_model["xiaomi/mimo-v2.5"]["input"] == 400 assert by_model["xiaomi/mimo-v2.5"]["cache_read"] == 250 assert by_model["xiaomi/mimo-v2.5"]["cost"] == 0.20 assert by_model["xiaomi/mimo-v2.5"]["unpriced_input"] == 0 assert by_model["deepseek/deepseek-v4-flash"]["reasoning"] == 3 - assert by_model["deepseek/deepseek-v4-flash"]["input"] == 300 + assert by_model["deepseek/deepseek-v4-flash"]["input"] == 250 assert by_model["deepseek/deepseek-v4-flash"]["cost"] == 0.05 @@ -200,23 +212,27 @@ def test_bahulam_store_counts_subagents_and_splits_root_cost(): ] _write_bahulam(tmp, sid, rows) + # Post-fix math (input inclusive of cache): + # xiaomi (coder/root): input=200-40=160, tokens_total=210 + # deepseek (plan/subagent): input=100-30=70, tokens_total=120 + # workflow.total_tokens = 210 + 120 = 330 store = ot.BahulamStore(os.path.join(tmp, "projects"), _jsonl_args()) workflow = store.workflows()[0] assert workflow.subagents == 1 assert workflow.root_cost == 0.02 assert workflow.total_cost == 0.05 - assert workflow.total_tokens == 400 + assert workflow.total_tokens == 330 nodes = store.workflow_nodes(sid) assert len(nodes) == 2 assert nodes[0]["depth"] == 0 assert nodes[0]["cost"] == 0.02 - assert nodes[0]["tokens_total"] == 250 + assert nodes[0]["tokens_total"] == 210 assert nodes[1]["depth"] == 1 assert nodes[1]["agent"] == "plan" assert nodes[1]["model_name"] == "deepseek/deepseek-v4-pro" assert nodes[1]["cost"] == 0.03 - assert nodes[1]["tokens_total"] == 150 + assert nodes[1]["tokens_total"] == 120 def test_bahulam_store_uses_coder_model_when_usage_has_no_model_breakdown(): @@ -297,13 +313,174 @@ def test_bahulam_missing_cost_stays_unpriced_and_bare_unknown_model_stays_bare() ] _write_bahulam(tmp, sid, rows) + # Post-fix: fresh input = 10 - 2 = 8; tokens_total = 8 + 5 + 2 = 15 store = ot.BahulamStore(os.path.join(tmp, "projects"), _jsonl_args()) workflow = store.workflows()[0] assert workflow.total_cost == 0 - assert workflow.total_tokens == workflow.unpriced_tokens == 17 + assert workflow.total_tokens == workflow.unpriced_tokens == 15 by_model = {row["model_name"]: row for row in store.model_breakdown()} assert "local-vision-test" in by_model assert "anthropic/local-vision-test" not in by_model - assert by_model["local-vision-test"]["unpriced_input"] == 10 + assert by_model["local-vision-test"]["unpriced_input"] == 8 assert by_model["local-vision-test"]["unpriced_cache_read"] == 2 + + +def test_bahulam_hosted_session_treats_provider_cost_as_unpriced(): + """Bahulam-hosted (is_byok=false) sessions bill the user via + credits_charged, not provider list cost. When credits_charged=0 the + total user spend must be $0 and the provider tokens land in unpriced + buckets — matches the reviewer's transcript observation.""" + sid = "0198fca0-34b4-7285-b3f0-3b8fb0489e63" + with tempfile.TemporaryDirectory() as tmp: + cwd = os.path.join(tmp, "repo") + os.makedirs(cwd) + rows = [ + _bahulam_user("hosted call", cwd=cwd), + _bahulam_event( + "session_info", + {"models": {"coder": "xiaomi/mimo-v2.5"}, "is_byok": False}, + cwd=cwd, + ), + _bahulam_event( + "complete", + { + "credits_charged": 0, + "usage": { + "total_input_tokens": 190101, + "total_output_tokens": 3341, + "cache_read_input_tokens": 152448, + "total_cost_usd": 0.127644, + "models": [ + { + "model": "xiaomi/mimo-v2.5", + "role": "coder", + "input_tokens": 190101, + "output_tokens": 3341, + "cache_read_tokens": 152448, + "cost_usd": 0.127644, + } + ], + }, + }, + cwd=cwd, + ), + ] + _write_bahulam(tmp, sid, rows) + + store = ot.BahulamStore(os.path.join(tmp, "projects"), _jsonl_args()) + workflow = store.workflows()[0] + # Hosted + credits_charged=0 → user spend is $0 + assert workflow.total_cost == 0 + # Total tokens: 190101 fresh_in=(190101-152448)=37653; 37653+3341+152448 = 193442 + assert workflow.total_tokens == 193442 + # Provider tokens land in unpriced buckets (they cost the user $0) + assert workflow.unpriced_tokens == 193442 + # Turn-row cost surfaces credits_charged, not provider list price + turn = store.message_timeline(sid)[0] + assert turn["cost"] == 0 + + +def test_bahulam_hosted_session_shows_credits_charged_when_nonzero(): + """Hosted session where the user's credits WERE charged: cost = credits.""" + sid = "0198fca0-34b4-7285-b3f0-3b8fb0489e64" + with tempfile.TemporaryDirectory() as tmp: + cwd = os.path.join(tmp, "repo") + os.makedirs(cwd) + rows = [ + _bahulam_user("paid hosted", cwd=cwd), + _bahulam_event( + "session_info", + {"models": {"coder": "xiaomi/mimo-v2.5"}, "is_byok": False}, + cwd=cwd, + ), + _bahulam_event( + "complete", + { + "credits_charged": 0.42, + "usage": { + "total_input_tokens": 1000, + "total_output_tokens": 100, + "cache_read_input_tokens": 300, + "total_cost_usd": 0.05, + "models": [ + { + "model": "xiaomi/mimo-v2.5", + "role": "coder", + "input_tokens": 1000, + "output_tokens": 100, + "cache_read_tokens": 300, + "cost_usd": 0.05, + } + ], + }, + }, + cwd=cwd, + ), + ] + _write_bahulam(tmp, sid, rows) + + store = ot.BahulamStore(os.path.join(tmp, "projects"), _jsonl_args()) + workflow = store.workflows()[0] + assert workflow.total_cost == 0.42 # what the user actually paid + assert store.message_timeline(sid)[0]["cost"] == 0.42 + + +def test_bahulam_byok_session_uses_provider_cost(): + """BYOK sessions: the user paid the LLM provider directly, so provider + cost IS user spend and remains as-is.""" + sid = "0198fca0-34b4-7285-b3f0-3b8fb0489e65" + with tempfile.TemporaryDirectory() as tmp: + cwd = os.path.join(tmp, "repo") + os.makedirs(cwd) + rows = [ + _bahulam_user("byok call", cwd=cwd), + _bahulam_event( + "session_info", + {"models": {"coder": "xiaomi/mimo-v2.5"}, "is_byok": True}, + cwd=cwd, + ), + _bahulam_event( + "complete", + { + "credits_charged": 0, + "usage": { + "total_input_tokens": 1000, + "total_output_tokens": 100, + "cache_read_input_tokens": 300, + "total_cost_usd": 0.05, + "models": [ + { + "model": "xiaomi/mimo-v2.5", + "role": "coder", + "input_tokens": 1000, + "output_tokens": 100, + "cache_read_tokens": 300, + "cost_usd": 0.05, + } + ], + }, + }, + cwd=cwd, + ), + ] + _write_bahulam(tmp, sid, rows) + + store = ot.BahulamStore(os.path.join(tmp, "projects"), _jsonl_args()) + workflow = store.workflows()[0] + assert workflow.total_cost == 0.05 + assert workflow.unpriced_tokens == 0 + + +def test_bahulam_opts_out_of_context_curve(): + """``complete`` events aggregate multiple internal LLM calls (plus any + sub-agent rollups), so ``tokens_total`` on a turn is not a per-request + context window snapshot. The store must opt out of context-curve + rendering to avoid a misleading chart.""" + sid = "0198fca0-34b4-7285-b3f0-3b8fb0489e66" + with tempfile.TemporaryDirectory() as tmp: + cwd = os.path.join(tmp, "repo") + os.makedirs(cwd) + _write_bahulam(tmp, sid, [_bahulam_user("hello", cwd=cwd)]) + store = ot.BahulamStore(os.path.join(tmp, "projects"), _jsonl_args()) + assert store.supports_context_curve(sid) is False From c818c6486d8d239569ea1fae58a50ca2ac7b391c Mon Sep 17 00:00:00 2001 From: admin bi migration Date: Mon, 31 Aug 2026 19:04:32 -0400 Subject: [PATCH 5/5] Address second-round review: help text, env-var docs, hosted-cost guard MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four surgical fixes from the follow-up review; two adjacent asks were verified against current code and skipped with reasons in the PR reply. 1. cli.py --status help text: the previous rebase left two duplicated backend groups smashed together with "with DIR only" twice, and Antigravity was missing. Replaced with one clean parenthetical listing all 11 backends. Test assertion updated to match. 2. doctor: BAHULAM_PROJECTS_DIR, BAHULAM_HOME, and KEPLER_HOME are read by the resolver but weren't in _ENV_VARS, so `opentab doctor` never reported them even when they were the reason the wrong directory was being scanned. Added. 3. cli.py --bahulam-dir help and sources.py no-sessions guidance now name every compat env var the resolver actually honors, plus the ~/.kepler/projects legacy fallback. 4. bahulam.py total_cost: credits_charged is now added only when is_byok is False. Hosted sessions still surface credits_charged as user spend; BYOK sessions keep model_cost as the sole source and refuse to double-count if a stray non-zero credit slips into a BYOK transcript. Also added the missing "filled by App._load_model_cache" comment on model_count=0 so it matches every other backend. Not changed (skipped with reason in PR reply): - model_count=len(model_rows): every other store sets 0 and lets App populate from model_breakdown; changing bahulam alone breaks the contract. - "same fix for turn-level tokens": already correct — the turn row computes inp = total_in - cr - cw at bahulam.py:570, verified against the plugins session (turn TOTAL 92.7M matched model_rows). Tests: 1317/1317 passing under python3 run_tests.py. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/opentab/cli.py | 8 ++++---- src/opentab/doctor.py | 3 +++ src/opentab/sources.py | 7 ++++--- src/opentab/stores/bahulam.py | 11 +++++++++-- tests/test_cli.py | 5 ++++- 5 files changed, 24 insertions(+), 10 deletions(-) diff --git a/src/opentab/cli.py b/src/opentab/cli.py index 7294d26..ad95949 100644 --- a/src/opentab/cli.py +++ b/src/opentab/cli.py @@ -183,7 +183,8 @@ def _add_global_args(parser: argparse.ArgumentParser) -> None: "--bahulam-dir", default=_default_bahulam_dir(), help="Bahulam Code projects directory (for --harness bahulam); " - "honors $BAHULAM_PROJECTS_DIR, then $BAHULAM_HOME/projects; default ~/.bahulam/projects", + "honors $BAHULAM_PROJECTS_DIR, then $BAHULAM_HOME/projects, then $KEPLER_HOME/projects; " + "default ~/.bahulam/projects (falls back to ~/.kepler/projects if present)", ) parser.add_argument( "--csv", @@ -282,9 +283,8 @@ def _add_legacy_command_flags(parser: argparse.ArgumentParser) -> None: metavar="DIR|SESSION", help="print the cost of the most recently active agent session (subagent " "subtree included) and exit, consulting every present harness backend " - "(OpenCode, Claude Code, Codex, Hermes, pi, omp, OpenClaw, Zaly, Gemini); with DIR " - "only " - "(OpenCode, Claude Code, Codex, Hermes, pi, omp, OpenClaw, Zaly, Bahulam Code); with DIR only " + "(OpenCode, Claude Code, Codex, Hermes, pi, omp, OpenClaw, Zaly, Gemini, " + "Antigravity, Bahulam Code); with DIR only " "sessions of that project count, with a session id (ses_... or a UUID -- the " "id is matched to its own backend) exactly that session is priced, and " "--harness pins one backend. Made for a tmux status line: set -g " diff --git a/src/opentab/doctor.py b/src/opentab/doctor.py index 21e4a4c..1fb274c 100644 --- a/src/opentab/doctor.py +++ b/src/opentab/doctor.py @@ -963,6 +963,9 @@ def file_rows(args: argparse.Namespace, full: bool = False) -> list[Row]: "ZALY_DATA", "ZALY_ROOT", "ZALY_STATE", + "BAHULAM_PROJECTS_DIR", + "BAHULAM_HOME", + "KEPLER_HOME", "XDG_CONFIG_HOME", "XDG_STATE_HOME", "XDG_DATA_HOME", diff --git a/src/opentab/sources.py b/src/opentab/sources.py index d0bdec1..5ba5fb1 100644 --- a/src/opentab/sources.py +++ b/src/opentab/sources.py @@ -587,9 +587,10 @@ def _build_store(args: argparse.Namespace, key: str) -> tuple[object, str]: if key == "bahulam": if not _bahulam_available(getattr(args, "bahulam_dir", "")): raise SystemExit( - "No Bahulam Code sessions found. Point --bahulam-dir (or $BAHULAM_PROJECTS_DIR / " - "$BAHULAM_HOME) " - f"at ~/.bahulam/projects (looked in {getattr(args, 'bahulam_dir', '')})." + "No Bahulam Code sessions found. Point --bahulam-dir (or " + "$BAHULAM_PROJECTS_DIR / $BAHULAM_HOME / $KEPLER_HOME) at " + "~/.bahulam/projects (or ~/.kepler/projects if you migrated from " + f"Kepler) (looked in {getattr(args, 'bahulam_dir', '')})." ) return BahulamStore(args.bahulam_dir, args), "OpenTab: loading Bahulam Code sessions…\r" kind, problem = opencode_db_verdict(args.db) diff --git a/src/opentab/stores/bahulam.py b/src/opentab/stores/bahulam.py index 4914409..0b6412c 100644 --- a/src/opentab/stores/bahulam.py +++ b/src/opentab/stores/bahulam.py @@ -754,7 +754,14 @@ def _workflow_rows(self, sessions: dict[str, dict]) -> list[Workflow]: model_rows = s["model_rows"] model_cost = sum(r["cost"] for r in model_rows) root_cost = sum(r["root_cost"] for r in model_rows) - total_cost = model_cost + float(s.get("credits_charged", 0.0)) + # Hosted (is_byok=false): model_cost is 0 (provider cost got + # routed to unpriced); credits_charged is what the user paid. + # BYOK / unknown: model_cost IS user spend; credits_charged + # should be 0 on the wire, but a stray non-zero would double- + # count against provider cost — refuse to add. + total_cost = model_cost + if s.get("is_byok") is False: + total_cost += float(s.get("credits_charged", 0.0)) rows.append( Workflow( id=sid, @@ -764,7 +771,7 @@ def _workflow_rows(self, sessions: dict[str, dict]) -> list[Workflow]: root_cost=root_cost, total_cost=total_cost, subagents=len(s["subagents"]), - model_count=0, + model_count=0, # filled by App._load_model_cache (matches every other store) total_tokens=sum(r["tokens_total"] for r in model_rows), unpriced_tokens=s["unpriced_tokens"], source=self.source_name, diff --git a/tests/test_cli.py b/tests/test_cli.py index a570996..b795001 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -772,7 +772,10 @@ def test_verb_help_is_focused_but_globals_still_parse(): # bahulam in cli.py's argparse choices; keep the assertion aligned with # that order so the help text doesn't drift out from under the test. assert "zaly · gemini · antigravity · bahulam · all (merged)" in compact_tui_help - assert "Zaly, Bahulam Code" in compact_tui_help + # The --status / cost help lists every backend in one clean parenthetical + # (previously two duplicated groups were smashed together by a bad + # conflict resolution, leaving "Zaly, Bahulam Code" as an artefact). + assert "Zaly, Gemini, Antigravity, Bahulam Code" in compact_tui_help # Hidden != gone: a suppressed global still parses on that verb. assert _parse(["pull", "--no-cache", "host"]).no_cache is True assert _parse(["export", "--harness", "claude"]).source == "claude"