Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
@@ -1,5 +1,11 @@
# Changelog

## 0.5.79-dev (unreleased)

### Features

- **Per-session model badge on overview row** (#153) — `board overview` now annotates each tongxue's row with a compact `[opus→sonnet]` badge when their session has downgraded. Builds on the alert path from PR #221 — the alert block still shows the full picture below; the badge is the instant-glance signal that matches a specific row. Same noise filters apply: cross-provider switches and `<synthetic>` placeholders don't trigger a badge. Stacks on #221.

## 0.5.78-dev (unreleased)

### Features
Expand Down
2 changes: 1 addition & 1 deletion VERSION
Original file line number Diff line number Diff line change
@@ -1 +1 @@
0.5.78-dev
0.5.79-dev
10 changes: 8 additions & 2 deletions lib/board_view.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@
from lib.common import validate_identity
from lib.fmt import error, heading, ok, warn
from lib.tmux_utils import capture_pane, has_session, pane_command
from lib.token_usage import collect_runtime_alerts, load_budget_defaults
from lib.token_usage import collect_runtime_alerts, load_budget_defaults, session_model_badges

SHELL_COMMANDS = {"zsh", "bash", "sh", "-zsh", "-bash", ""}
SPINNER_RE = re.compile(r"^\s*(⠋|⠙|⠹|⠸|⠼|⠴|⠦|⠧|⠇|⠏|●)", re.MULTILINE)
Expand Down Expand Up @@ -149,6 +149,9 @@ def cmd_overview(db: BoardDB) -> None:
print(heading(f"=== {db.env.project_root.name} {now} ==="))
print()

# Compute once so the per-session loop is just a dict lookup.
badges = session_model_badges(db.env.project_root)

# ── sessions ──
for row in db.query("SELECT name, status, last_heartbeat FROM sessions WHERE name != 'all' ORDER BY name"):
name, task, last_hb = row[0], row[1], row[2]
Expand All @@ -161,7 +164,10 @@ def cmd_overview(db: BoardDB) -> None:
else:
task = "(no status)"

line = f" {status:12s} {name:<10s} {task}"
badge = badges.get(name.lower(), "")
badge_str = f" {warn(f'[{badge}]')}" if badge else ""

line = f" {status:12s} {name:<10s}{badge_str} {task}"
if ago:
line += f" {ago}"
if inbox:
Expand Down
55 changes: 55 additions & 0 deletions lib/token_usage.py
Original file line number Diff line number Diff line change
Expand Up @@ -190,6 +190,61 @@ def tongxue_token_summary(
return agg[0] if agg else None


def _short_model_label(model: str) -> str:
"""Pick a one-word label suitable for a row badge: opus/sonnet/haiku/gpt-5.x/mini/other."""
lowered = model.lower()
if "mini" in lowered:
return "mini"
if "haiku" in lowered:
return "haiku"
if "opus" in lowered:
return "opus"
if "sonnet" in lowered:
return "sonnet"
if lowered.startswith("gpt-"):
# gpt-5.4-mini already handled above; for plain gpt-5.x return the family marker.
# Guard the empty-second-segment case ("gpt-" alone, or "gpt--5.4") so the badge
# never renders as just an arrow with nothing on one side.
parts = lowered.split("-", 2)
candidate = parts[1] if len(parts) > 1 else ""
return candidate or model[:8]
return model[:8]


def session_model_badges(
project_root: Path,
*,
recent_hours: float | None = DEFAULT_RECENT_HOURS,
) -> dict[str, str]:
"""Return `{tongxue_name_lower: "opus→sonnet"}` for downgraded sessions only.

Uses the same tier-aware filtering as `model_state_alerts` so cross-provider
switches do not produce badges. Names are lowercased to match how the board DB
stores session names. Sessions without a downgrade are omitted; consumers can
treat a missing key as "all clear".
"""
sessions = _load_project_sessions(project_root, recent_hours=recent_hours)
if not sessions:
return {}
badges: dict[str, str] = {}
for s in aggregate_by_name(sessions):
models = [m for m in s.get("models", []) if m]
if len(models) < 2:
continue
first, latest = models[0], models[-1]
first_tier = _model_tier(first)
latest_tier = _model_tier(latest)
if first_tier == 0 or latest_tier == 0:
continue
if latest_tier >= first_tier:
continue
name = (s.get("name") or "").lower()
if not name:
continue
badges[name] = f"{_short_model_label(first)}→{_short_model_label(latest)}"
return badges


def _parse_usage_args(args: list[str]) -> dict[str, Any]:
parsed: dict[str, Any] = {"detail": False, "budget": 0.0, "warn_pct": DEFAULT_BUDGET_WARN_PCT}
i = 0
Expand Down
2 changes: 1 addition & 1 deletion package.json
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
{
"name": "claude-nb",
"version": "0.5.78-dev",
"version": "0.5.79-dev",
"description": "Multi-agent coordination framework for Claude Code sessions",
"engines": {
"node": ">=18"
Expand Down
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"

[project]
name = "claude-nb"
version = "0.5.78.dev0"
version = "0.5.79.dev0"
description = "Multi-agent coordination framework for Claude Code sessions"
requires-python = ">=3.11"
license = "MIT"
Expand Down
118 changes: 118 additions & 0 deletions tests/test_token_usage.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,13 +8,15 @@

from lib.token_usage import (
_project_slug,
_short_model_label,
aggregate_by_name,
cmd_usage,
collect_runtime_alerts,
estimate_cost,
load_budget_defaults,
model_state_alerts,
parse_session_usage,
session_model_badges,
tongxue_token_summary,
)

Expand Down Expand Up @@ -636,3 +638,119 @@ def test_merges_multiple_jsonls(self, tmp_path):
assert result["input"] == 30
assert result["output"] == 150
assert result["messages"] == 2


class TestShortModelLabel:
def test_opus(self):
assert _short_model_label("claude-opus-4-7") == "opus"

def test_sonnet(self):
assert _short_model_label("claude-sonnet-4-7") == "sonnet"

def test_haiku(self):
assert _short_model_label("claude-haiku-4-5-20251001") == "haiku"

def test_mini_takes_precedence(self):
# gpt-5.4-mini should label as "mini" (the downgrade tier signal), not the family.
assert _short_model_label("gpt-5.4-mini") == "mini"

def test_gpt_family(self):
assert _short_model_label("gpt-5.4") == "5.4"

def test_gpt_bare_falls_back_to_truncated(self):
"""`gpt-` alone has an empty second segment — must not render as empty string."""
assert _short_model_label("gpt-") == "gpt-"

def test_gpt_double_dash_falls_back(self):
"""`gpt--5.4` splits to ('gpt', '', '5.4') — empty segment must fall back."""
# split("-", 2) gives ['gpt', '', '5.4'], parts[1] is "", must fall back.
assert _short_model_label("gpt--5.4") == "gpt--5.4"[:8]


class TestSessionModelBadges:
def test_empty_when_no_jsonls(self, tmp_path):
project_dir = tmp_path / "empty"
project_dir.mkdir()
with patch("lib.token_usage._find_project_dir", return_value=project_dir):
assert session_model_badges(tmp_path / "project") == {}

def test_no_badge_when_all_clear(self, tmp_path):
project_dir = tmp_path / "jsonls"
project_dir.mkdir()
_write_jsonl(
project_dir / "s1.jsonl",
[
{"type": "custom-title", "customTitle": "alice"},
_make_assistant_msg(model="claude-opus-4-7"),
],
)
with patch("lib.token_usage._find_project_dir", return_value=project_dir):
assert session_model_badges(tmp_path / "project") == {}

def test_emits_badge_for_downgrade(self, tmp_path):
project_dir = tmp_path / "jsonls"
project_dir.mkdir()
_write_jsonl(
project_dir / "s1.jsonl",
[
{"type": "custom-title", "customTitle": "alice"},
_make_assistant_msg(model="claude-opus-4-7"),
_make_assistant_msg(model="claude-sonnet-4-7"),
],
)
with patch("lib.token_usage._find_project_dir", return_value=project_dir):
badges = session_model_badges(tmp_path / "project")
assert badges == {"alice": "opus→sonnet"}

def test_skips_cross_provider_switch(self, tmp_path):
"""User-initiated cnb model use shouldn't surface as a badge either."""
project_dir = tmp_path / "jsonls"
project_dir.mkdir()
_write_jsonl(
project_dir / "s1.jsonl",
[
{"type": "custom-title", "customTitle": "alice"},
_make_assistant_msg(model="claude-opus-4-7"),
_make_assistant_msg(model="deepseek-v4-pro"),
],
)
with patch("lib.token_usage._find_project_dir", return_value=project_dir):
assert session_model_badges(tmp_path / "project") == {}

def test_name_is_lowercased(self, tmp_path):
project_dir = tmp_path / "jsonls"
project_dir.mkdir()
_write_jsonl(
project_dir / "s1.jsonl",
[
{"type": "custom-title", "customTitle": "Alice"},
_make_assistant_msg(model="claude-opus-4-7"),
_make_assistant_msg(model="claude-haiku-4-5-20251001"),
],
)
with patch("lib.token_usage._find_project_dir", return_value=project_dir):
badges = session_model_badges(tmp_path / "project")
assert "alice" in badges
assert badges["alice"] == "opus→haiku"

def test_merges_multiple_jsonls_per_session(self, tmp_path):
"""Two JSONLs for the same tongxue should aggregate via aggregate_by_name first."""
project_dir = tmp_path / "jsonls"
project_dir.mkdir()
_write_jsonl(
project_dir / "s1.jsonl",
[
{"type": "custom-title", "customTitle": "bob"},
_make_assistant_msg(model="claude-opus-4-7"),
],
)
_write_jsonl(
project_dir / "s2.jsonl",
[
{"type": "custom-title", "customTitle": "bob"},
_make_assistant_msg(model="claude-sonnet-4-7"),
],
)
with patch("lib.token_usage._find_project_dir", return_value=project_dir):
badges = session_model_badges(tmp_path / "project")
assert badges == {"bob": "opus→sonnet"}