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..ad95949 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,15 +99,15 @@ def _add_global_args(parser: argparse.ArgumentParser) -> None: "zaly", "gemini", "antigravity", + "bahulam", "all", "remote", ), 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)", ) @@ -178,6 +179,13 @@ 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, then $BAHULAM_HOME/projects, then $KEPLER_HOME/projects; " + "default ~/.bahulam/projects (falls back to ~/.kepler/projects if present)", + ) parser.add_argument( "--csv", default=None, @@ -275,8 +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, 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 " @@ -549,6 +557,7 @@ def _build_parser() -> argparse.ArgumentParser: "zaly_dir", "gemini_dir", "antigravity_dir", + "bahulam_dir", "csv", "jsonl", "remotes", @@ -1101,6 +1110,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..1fb274c 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, $BAHULAM_HOME, or $KEPLER_HOME, or pass --bahulam-dir"), ) # fmt: skip @@ -962,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 7c3226b..5ba5fb1 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,32 @@ def _default_zaly_dir() -> str: return default_zaly_data_dir() +def _default_bahulam_dir() -> str: + """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 = { "csv": "csv", "jsonl": "jsonl", @@ -117,6 +144,7 @@ def _default_zaly_dir() -> str: "zaly": "zaly_dir", "gemini": "gemini_dir", "antigravity": "antigravity_dir", + "bahulam": "bahulam_dir", } @@ -270,6 +298,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 +347,7 @@ def _vscode_available(args: argparse.Namespace) -> bool: "zaly": "Zaly", "gemini": "Gemini", "antigravity": "Antigravity", + "bahulam": "Bahulam Code", "all": "all", } @@ -328,6 +362,7 @@ def _vscode_available(args: argparse.Namespace) -> bool: "Zaly": "zaly --session", "Gemini": "gemini --resume", "Antigravity": "antigravity", + "Bahulam Code": "bahulam resume", } @@ -350,6 +385,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 +426,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 +584,15 @@ 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 / $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) 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..0b6412c --- /dev/null +++ b/src/opentab/stores/bahulam.py @@ -0,0 +1,1003 @@ +"""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 ``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, + "u_input": 0, + "u_output": 0, + "u_reasoning": 0, + "u_cache_read": 0, + "u_cache_write": 0, + "u_cache_write_1h": 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 + + @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 _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, *, hosted_unpriced: bool = False) -> None: + """Accumulate counters from a usage dict into *acc*. + + 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)) + 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)) + 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) + acc["runs"] += 1 + 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 + 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 + 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: + """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 "" + 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 + "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 + 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() + 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, hosted_unpriced=hosted_unpriced) + 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, + # 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, + *, hosted_unpriced: bool = False, + ) -> None: + """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 + 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) + 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"): + 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 and ``bahulam_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 carry their payload under ``event.data.*``. + if typ != "bahulam_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 + 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) + 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"]: + s["models"].setdefault( + qualified, + {"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 {} + 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, 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: + if not isinstance(m, dict): + continue + 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 = 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, hosted_unpriced=hosted_unpriced) + role = str(m.get("role") or "") + if self._is_root_role(role): + self._add_usage(entry["root"], per_model, hosted_unpriced=hosted_unpriced) + else: + 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 "" + ) + 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, 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)) + 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) + _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" + if isinstance(models_usage, list) and models_usage: + 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( + { + "ts": ts or "", + "depth": 0, + "agent": "-", + "effort": "", + "model_name": first_model, + "cost": float(turn_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 / 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) + + # ── 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"], + "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["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( + 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. + + 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"] + model_cost = sum(r["cost"] for r in model_rows) + root_cost = sum(r["root_cost"] for r in model_rows) + # 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, + 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, # 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, + 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 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) + 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/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_cli.py b/tests/test_cli.py index 2f705fd..b795001 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -767,6 +767,15 @@ 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()) + # 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 + # 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" 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. diff --git a/tests/test_sources.py b/tests/test_sources.py index 3e970bc..8772410 100644 --- a/tests/test_sources.py +++ b/tests/test_sources.py @@ -189,3 +189,80 @@ 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 _env_snapshot(names): + """Save and clear the named env vars; return a restore callable. + + 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 new file mode 100644 index 0000000..b642534 --- /dev/null +++ b/tests/test_stores_bahulam.py @@ -0,0 +1,486 @@ +import os +import tempfile + +import opentab as ot + +from tests._support import _jsonl_args, _write_jsonl + + +def _bahulam_event(event_type, data=None, *, cwd="/work/repo", ts=None): + return { + "type": "bahulam_event", + "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_event_shape_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", + } + }, + 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) + + # 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 == 1110 + assert workflow.unpriced_tokens == 0 + + 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 + # 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"] == 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"] == 250 + 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) + + # 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 == 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"] == 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"] == 120 + + +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 + + +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) + + # 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 == 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"] == 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