diff --git a/README.md b/README.md index 3254d15..c3e9986 100644 --- a/README.md +++ b/README.md @@ -40,3 +40,65 @@ To demo/test: ```bash make demo ``` + +## Claude Subscription Quota Mode + +YAS can display an approximate **subscription quota gauge** instead of the per-session dollar cost rows. This is useful on flat-rate plans (Claude Pro, Max 5x, Max 20x) where dollar totals matter less than remaining headroom for the window. + +### Why approximation? + +Anthropic does not expose live quota data through the transcript API the way Codex rollouts do. YAS approximates usage from local session/day token totals against hardcoded plan ceilings. The ceilings are heuristic estimates based on Anthropic's "5x / 20x of Pro" framing — [check the current plan page](https://www.anthropic.com/pricing) and tune via env vars if the defaults diverge from your observed experience. + +### Env-var contract + +| Variable | Default | Description | +|---|---|---| +| `YAS_CLAUDE_MODE` | `cost` | `cost` = existing dollar rows · `quota` = subscription gauge | +| `YAS_CLAUDE_PLAN` | `max20` | `pro` · `max5` · `max20` | +| `YAS_CLAUDE_5H_CAP_TOKENS` | _(plan default)_ | Override the 5-hour window ceiling | +| `YAS_CLAUDE_WEEKLY_CAP_TOKENS` | _(plan default)_ | Override the weekly window ceiling | + +### Plan ceilings (heuristic defaults) + +| Plan | 5h cap | Weekly cap | +|---|---|---| +| `pro` | 1.5M tokens | 18M tokens | +| `max5` | 7.5M tokens | 90M tokens | +| `max20` | 30M tokens | 360M tokens | + +### Quota gauge render + +``` +Wide: 󱙄 5h ▆▆░░░░░░ 24% │ 7d ▆░░░░░░░ 12% │ max20x +Medium: 󱙄 5h 24% · 7d 12% +Narrow: 󱙄 5h 24% +``` + +Color thresholds: `< 60%` green · `60–85%` yellow · `> 85%` red (same as Codex gauge). + +When `YAS_CLAUDE_MODE=quota`, the two cost rows are replaced by a single quota gauge row. The Codex gauge row (if Codex data is present) still appears beneath it. + +### Example + +```bash +YAS_CLAUDE_MODE=quota YAS_CLAUDE_PLAN=max20 python3 ~/.claude/statusline_command.py < session.json +``` + +--- + +## Codex Rate-Limit Gauge + +A third row beneath the Claude cost rows shows your **Codex rate-limit window consumption** (rate limits, not dollars — Codex Pro is flat-rate). + +``` + 󰞬 5h ▆░░░░░░░ 12% │ 7d ▆▆░░░░░░ 28% │ pro +``` + +- Reads from `~/.codex/sessions///
/rollout-*.jsonl` +- Override the sessions root with `YAS_CODEX_SESSIONS_DIR` +- Shows the **5-hour** and **7-day** window consumption percentages +- Color thresholds: `< 60%` green · `60–85%` yellow · `> 85%` red +- If the latest rollout is older than 1 hour, a `(stale Xh)` indicator appears +- Narrows gracefully: medium width shows both percentages without bars; narrow shows only the 5h window +- If no Codex session data is found, renders `no codex data` in dim grey + diff --git a/claude/statusline_command.py b/claude/statusline_command.py index 172fcaf..1e9e6ee 100755 --- a/claude/statusline_command.py +++ b/claude/statusline_command.py @@ -114,6 +114,8 @@ def terminal_width() -> int: # chat round-trips never lose the bytes. Render only in a Nerd-Font-capable # terminal. ICON_COST = '\uefc8' # nf-md currency-usd (cost row) +ICON_CODEX = '\U000f19e3' # nf-md-clock-time-eight-outline (Codex rate-limit row) +ICON_CLAUDE_QUOTA = '\U000f96e4' # nf-md-chart-donut (Claude quota gauge row) ICON_TOK_RATE = '\U000f18a7' # nf-md gauge (t/m rate label) GLYPH_MODEL = '\U000f08b9' # nf-md-monitor-dashboard GLYPH_THINKING = '\U000f1a53' # nf-md-brain @@ -455,6 +457,159 @@ def from_dict(cls, d: dict) -> RateLimits: ) +@dataclass +class CodexRateLimits: + """Rate-limit state read from ~/.codex/sessions rollout files. + + All fields are None when no Codex rollout data is available. + ``primary`` = 5-hour window + ``secondary`` = 7-day window (10080 min) + """ + + primary_pct: float | None = None + secondary_pct: float | None = None + primary_resets_at: int | None = None + secondary_resets_at: int | None = None + plan_type: str | None = None + data_age_seconds: float | None = None + + # Stale threshold: if the newest token_count event is older than this, we + # surface a "(stale Xh)" indicator next to the plan label. + STALE_SECONDS: int = 3600 + + def is_stale(self) -> bool: + if self.data_age_seconds is None: + return False + return self.data_age_seconds > self.STALE_SECONDS + + @classmethod + def load(cls, sessions_root: Path | None = None) -> CodexRateLimits: + """Load the most recent token_count event from Codex rollout files. + + Args: + sessions_root: Override the default ``~/.codex/sessions`` root. + Falls back to the ``YAS_CODEX_SESSIONS_DIR`` env + var, then to the XDG default. + """ + if sessions_root is None: + env_dir = os.environ.get('YAS_CODEX_SESSIONS_DIR') + if env_dir: + sessions_root = Path(env_dir) + else: + sessions_root = HOME / '.codex' / 'sessions' + + rollout = cls._find_latest_rollout(sessions_root) + if rollout is None: + return cls() + + data_age = time.time() - rollout.stat().st_mtime + event = cls._last_token_count(rollout) + if event is None: + return cls() + + rl = event.get('rate_limits') or {} + primary = rl.get('primary') or {} + secondary = rl.get('secondary') or {} + return cls( + primary_pct = float(primary.get('used_percent', 0)), + secondary_pct = float(secondary.get('used_percent', 0)), + primary_resets_at = primary.get('resets_at'), + secondary_resets_at = secondary.get('resets_at'), + plan_type = rl.get('plan_type'), + data_age_seconds = data_age, + ) + + @staticmethod + def _find_latest_rollout(root: Path) -> Path | None: + """Return the most recently modified rollout-*.jsonl under root.""" + if not root.is_dir(): + return None + candidates = sorted(root.glob('*/*/[0-9][0-9]/rollout-*.jsonl'), key=lambda p: p.stat().st_mtime, reverse=True) + if not candidates: + # Also try direct rollout-*.jsonl at root (flat layout) + candidates = sorted(root.glob('rollout-*.jsonl'), key=lambda p: p.stat().st_mtime, reverse=True) + return candidates[0] if candidates else None + + @staticmethod + def _last_token_count(rollout: Path) -> dict | None: + """Reverse-scan ``rollout`` and return the last token_count payload.""" + try: + lines = rollout.read_bytes().splitlines() + except OSError: + return None + for raw in reversed(lines): + if b'token_count' not in raw: + continue + try: + d = json.loads(raw) + except (ValueError, TypeError): + continue + if d.get('type') == 'event_msg': + p = d.get('payload') or {} + if p.get('type') == 'token_count': + return p + return None + + +# --------------------------------------------------------------------------- +# Claude subscription quota approximation +# --------------------------------------------------------------------------- + +# Plan ceilings are heuristic estimates based on Anthropic's "5x / 20x of Pro" +# framing. Anthropic does not publish exact token counts for Max plans. +# Override per-field via YAS_CLAUDE_5H_CAP_TOKENS / YAS_CLAUDE_WEEKLY_CAP_TOKENS. +PLAN_CEILINGS: dict[str, dict[str, int]] = { + 'pro': {'cap_5h_tokens': 1_500_000, 'cap_weekly_tokens': 18_000_000}, + 'max5': {'cap_5h_tokens': 7_500_000, 'cap_weekly_tokens': 90_000_000}, + 'max20': {'cap_5h_tokens': 30_000_000, 'cap_weekly_tokens': 360_000_000}, +} + + +@dataclass +class ClaudeQuota: + """Approximate Claude subscription quota state. + + Token counts are derived from the local transcript and token log — Claude + does not expose live quota data the way Codex does. All values are + best-effort approximations; the operator should tune ceilings via env vars + if the defaults diverge from observed behaviour. + + ``pct_5h`` = session_tokens / cap_5h_tokens × 100 (proxy for 5h window) + ``pct_weekly`` = day_tokens / cap_weekly_tokens × 100 (proxy for rolling week) + """ + + session_tokens: int + day_tokens: int + plan: str + cap_5h_tokens: int + cap_weekly_tokens: int + pct_5h: float + pct_weekly: float + + @classmethod + def load(cls, plan: str, session_tokens: int, day_tokens: int) -> 'ClaudeQuota': + """Build a ClaudeQuota from plan name + token counts. + + Env-var overrides: + YAS_CLAUDE_5H_CAP_TOKENS — override the plan's 5-hour ceiling + YAS_CLAUDE_WEEKLY_CAP_TOKENS — override the plan's weekly ceiling + """ + defaults = PLAN_CEILINGS.get(plan, PLAN_CEILINGS['max20']) + cap_5h = int(os.environ.get('YAS_CLAUDE_5H_CAP_TOKENS', defaults['cap_5h_tokens'])) + cap_weekly = int(os.environ.get('YAS_CLAUDE_WEEKLY_CAP_TOKENS', defaults['cap_weekly_tokens'])) + pct_5h = min(session_tokens / cap_5h * 100, 100.0) if cap_5h > 0 else 0.0 + pct_weekly = min(day_tokens / cap_weekly * 100, 100.0) if cap_weekly > 0 else 0.0 + return cls( + session_tokens=session_tokens, + day_tokens=day_tokens, + plan=plan, + cap_5h_tokens=cap_5h, + cap_weekly_tokens=cap_weekly, + pct_5h=pct_5h, + pct_weekly=pct_weekly, + ) + + @dataclass class SessionInfo: session_id: str = '' @@ -1590,6 +1745,19 @@ def border_line(self, content: str, width: int, fill: float = 1.0, bg_lead: str return f'{left}│{self.R}{lead}{content}{pad_str}{right}│{self.R}' + +def _effective_soft_limit(ctx: 'ContextWindow') -> int: + """Soft-limit threshold scaled to the model's actual context window. + + SOFT_LIMIT is the auto-compact warning zone for legacy 200K Claude models. + For 1M-context models, scale the soft limit to ~75% of the actual window so + the bar fills proportionally instead of overflowing at 15% real usage. + Falls back to the legacy constant when context_window_size is unknown. + """ + if ctx.context_window_size > 0: + return max(SOFT_LIMIT, int(ctx.context_window_size * 0.75)) + return SOFT_LIMIT + class Renderer: def __init__(self, bg_shift: str = 'warm', theme: Theme | None = None) -> None: self.bg_shift = bg_shift if bg_shift in ('warm', 'cool') else 'warm' @@ -2274,10 +2442,11 @@ def _empty_section(self, empty: int, blend: bool = True) -> str: def context_line(self, ctx: ContextWindow, available: int = 76) -> str: total_tokens = ctx.total_input_tokens + ctx.total_output_tokens - fill_ratio = min(total_tokens / SOFT_LIMIT, 1.0) - pct_soft = total_tokens / SOFT_LIMIT * 100 + soft_limit = _effective_soft_limit(ctx) + fill_ratio = min(total_tokens / soft_limit, 1.0) + pct_soft = total_tokens / soft_limit * 100 - if total_tokens >= SOFT_LIMIT: + if total_tokens >= soft_limit: a = BOLD + self.risk_zone_color(total_tokens) secondary = '' if ctx.context_window_size > 0: @@ -2305,10 +2474,11 @@ def context_line(self, ctx: ContextWindow, available: int = 76) -> str: def context_line_compact(self, ctx: ContextWindow, available: int) -> str: total_tokens = ctx.total_input_tokens + ctx.total_output_tokens - fill_ratio = min(total_tokens / SOFT_LIMIT, 1.0) - pct_soft = total_tokens / SOFT_LIMIT * 100 + soft_limit = _effective_soft_limit(ctx) + fill_ratio = min(total_tokens / soft_limit, 1.0) + pct_soft = total_tokens / soft_limit * 100 - if total_tokens >= SOFT_LIMIT: + if total_tokens >= soft_limit: a = BOLD + self.risk_zone_color(total_tokens) prefix = f'{a}{pct_soft:.0f}%{self.R} ' bar_w = max(4, available - _visible_width(prefix) - 3) @@ -2413,6 +2583,149 @@ def helper(self, five_hour: RateBucket) -> str: except Exception as e: return f'{e.__class__.__name__}, {str(e)}' + # ------------------------------------------------------------------ + # Codex rate-limit gauge (3rd statusline row) + # ------------------------------------------------------------------ + + def codex_pct_colour(self, pct: float) -> str: + """Color for a Codex rate-limit percentage. + + Thresholds differ from Claude's fill_colour (70/90) because + Codex Pro is flat-rate — hitting 85%+ of a window is operationally + more significant than a dollar cost approaching a soft limit. + """ + if pct >= 85.0: + return self.alert + if pct >= 60.0: + return self.warn + return self.safe + + def _codex_pct_bar(self, pct: float, bar_w: int = 8) -> str: + """A short filled/empty bar for a Codex window percentage.""" + filled = max(0, min(bar_w, round(pct / 100 * bar_w))) + empty = bar_w - filled + clr = self.codex_pct_colour(pct) + return f'{clr}{BarChars.HEAVY * filled}{self.R}{self.BAR_EMPTY}{BarChars.EMPTY * empty}{self.R}' + + def codex_rate_row(self, rl: CodexRateLimits, width: int = 100) -> str: + """Render the Codex rate-limit gauge row. + + Width modes: + wide (>80) : icon 5h bar NN% | 7d bar NN% | plan [stale Xh] + medium (<=80): icon 5h NN% . 7d NN% + narrow (<=55): icon 5h NN% + """ + # No-data state + if rl.primary_pct is None: + return f' {self.LABEL}{ICON_CODEX} no codex data{self.R}' + + primary_pct = rl.primary_pct + secondary_pct = rl.secondary_pct if rl.secondary_pct is not None else 0.0 + plan = (rl.plan_type or '').lower() + + p_clr = self.codex_pct_colour(primary_pct) + s_clr = self.codex_pct_colour(secondary_pct) + + icon_str = f'{self.LABEL}{ICON_CODEX}{self.R}' + + if width <= NARROW_WIDTH: + # Narrow: icon + 5h pct only + return f' {icon_str} {self.LABEL}5h{self.R} {p_clr}{primary_pct:.0f}%{self.R}' + + if width <= MEDIUM_WIDTH: + # Medium: icon + 5h pct . 7d pct + return ( + f' {icon_str}' + f' {self.LABEL}5h{self.R} {p_clr}{primary_pct:.0f}%{self.R}' + f' {self.LABEL}\xb7{self.R}' + f' {self.LABEL}7d{self.R} {s_clr}{secondary_pct:.0f}%{self.R}' + ) + + # Wide: bars + plan label + optional stale indicator + p_bar = self._codex_pct_bar(primary_pct) + s_bar = self._codex_pct_bar(secondary_pct) + + stale_str = '' + if rl.is_stale() and rl.data_age_seconds is not None: + stale_h = int(rl.data_age_seconds // 3600) + stale_m = int((rl.data_age_seconds % 3600) // 60) + stale_label = f'{stale_h}h' if stale_h else f'{stale_m}m' + stale_str = f' {self.COMMIT}(stale {stale_label}){self.R}' + + sep = f' {self.LABEL}│{self.R} ' + + return ( + f' {icon_str}' + f' {self.LABEL}5h{self.R} {p_bar} {p_clr}{primary_pct:.0f}%{self.R}' + f'{sep}' + f'{self.LABEL}7d{self.R} {s_bar} {s_clr}{secondary_pct:.0f}%{self.R}' + f'{sep}' + f'{self.LABEL}{plan}{self.R}' + f'{stale_str}' + ) + + + # ------------------------------------------------------------------ + # Claude subscription quota gauge + # ------------------------------------------------------------------ + + def claude_quota_pct_colour(self, pct: float) -> str: + """Color for a Claude quota percentage (same thresholds as Codex).""" + if pct >= 85.0: + return self.alert + if pct >= 60.0: + return self.warn + return self.safe + + def _claude_quota_bar(self, pct: float, bar_w: int = 8) -> str: + """A short filled/empty bar for a Claude quota percentage.""" + filled = max(0, min(bar_w, round(pct / 100 * bar_w))) + empty = bar_w - filled + clr = self.claude_quota_pct_colour(pct) + return f'{clr}{BarChars.HEAVY * filled}{self.R}{self.BAR_EMPTY}{BarChars.EMPTY * empty}{self.R}' + + def claude_quota_row(self, quota: ClaudeQuota, width: int = 100) -> str: + """Render the Claude subscription quota gauge row. + + Width modes: + wide (>80) : icon 5h bar NN% | 7d bar NN% | plan_label + medium (<=80): icon 5h NN% · 7d NN% + narrow (<=55): icon 5h NN% + """ + pct_5h = quota.pct_5h + pct_weekly = quota.pct_weekly + plan_label = f'{quota.plan}x' if quota.plan in ('max5', 'max20') else quota.plan + + h_clr = self.claude_quota_pct_colour(pct_5h) + w_clr = self.claude_quota_pct_colour(pct_weekly) + icon_str = f'{self.LABEL}{ICON_CLAUDE_QUOTA}{self.R}' + + if width <= NARROW_WIDTH: + return f' {icon_str} {self.LABEL}5h{self.R} {h_clr}{pct_5h:.0f}%{self.R}' + + if width <= MEDIUM_WIDTH: + return ( + f' {icon_str}' + f' {self.LABEL}5h{self.R} {h_clr}{pct_5h:.0f}%{self.R}' + f' {self.LABEL}\xb7{self.R}' + f' {self.LABEL}7d{self.R} {w_clr}{pct_weekly:.0f}%{self.R}' + ) + + # Wide: bars + plan label + h_bar = self._claude_quota_bar(pct_5h) + w_bar = self._claude_quota_bar(pct_weekly) + sep = f' {self.LABEL}│{self.R} ' + + return ( + f' {icon_str}' + f' {self.LABEL}5h{self.R} {h_bar} {h_clr}{pct_5h:.0f}%{self.R}' + f'{sep}' + f'{self.LABEL}7d{self.R} {w_bar} {w_clr}{pct_weekly:.0f}%{self.R}' + f'{sep}' + f'{self.LABEL}{plan_label}{self.R}' + ) + + @dataclass class RowSpec: kind: str # 'top_border', 'bottom_border', 'separator', 'separator_dim', 'content' @@ -2438,7 +2751,8 @@ class LayoutSpec: def build_narrow(session: SessionInfo, width: int, r: Renderer) -> LayoutSpec: ctx = session.context_window total_tokens = ctx.total_input_tokens + ctx.total_output_tokens - fill = min(total_tokens / SOFT_LIMIT, 1.0) + soft_limit = _effective_soft_limit(ctx) + fill = min(total_tokens / soft_limit, 1.0) effort_for_bg = session.effort.level if session.thinking.enabled else '' pill_pct = r._model_bg_pct(effort_for_bg) @@ -2485,7 +2799,8 @@ def build_narrow(session: SessionInfo, width: int, r: Renderer) -> LayoutSpec: def build_medium(session: SessionInfo, width: int, r: Renderer) -> LayoutSpec: ctx = session.context_window total_tokens = ctx.total_input_tokens + ctx.total_output_tokens - fill = min(total_tokens / SOFT_LIMIT, 1.0) + soft_limit = _effective_soft_limit(ctx) + fill = min(total_tokens / soft_limit, 1.0) effort_for_bg = session.effort.level if session.thinking.enabled else '' pill_pct = r._model_bg_pct(effort_for_bg) @@ -2544,7 +2859,8 @@ def build_medium(session: SessionInfo, width: int, r: Renderer) -> LayoutSpec: def build_wide(session: SessionInfo, width: int, r: Renderer) -> LayoutSpec: ctx = session.context_window total_tokens = ctx.total_input_tokens + ctx.total_output_tokens - fill = min(total_tokens / SOFT_LIMIT, 1.0) + soft_limit = _effective_soft_limit(ctx) + fill = min(total_tokens / soft_limit, 1.0) effort_for_bg = session.effort.level if session.thinking.enabled else '' bg_lead = r.model_bg_lead(session.model_name, effort_for_bg) @@ -2563,6 +2879,16 @@ def build_wide(session: SessionInfo, width: int, r: Renderer) -> LayoutSpec: subagents = RunningSubagents.from_session(session.session_id, session.workspace.project_dir) tasks = TaskList.from_session(session.transcript_path) elapsed = elapsed_from_transcript(session.transcript_path) + codex_rl = CodexRateLimits.load() + + # Claude quota mode: YAS_CLAUDE_MODE=quota replaces cost rows with gauge + _claude_mode = os.environ.get('YAS_CLAUDE_MODE', 'cost').strip().lower() + _claude_plan = os.environ.get('YAS_CLAUDE_PLAN', 'max20').strip().lower() + claude_quota = ClaudeQuota.load( + plan=_claude_plan, + session_tokens=usage.billed_in + usage.cache_read + usage.out, + day_tokens=token_log.day_in + token_log.day_cache_read + token_log.day_out, + ) if _claude_mode == 'quota' else None git = GitInfo.from_cwd(session.cwd) helper_text, right_text, right_w = r.model_right_section( @@ -2618,8 +2944,17 @@ def build_wide(session: SessionInfo, width: int, r: Renderer) -> LayoutSpec: tokens_downs = vsep_cols + ((spark_mark_col,) if spark_mark_col else ()) rows.append(RowSpec('separator_dim', downs=tokens_downs)) - for lt in line_tokens: - rows.append(RowSpec('content', content=lt)) + if claude_quota is not None: + # Quota mode: replace the two cost rows + codex row with a single quota gauge + rows.append(RowSpec('content', content=r.claude_quota_row(claude_quota, width))) + else: + # Cost mode (default): render token/cost rows as usual + for lt in line_tokens: + rows.append(RowSpec('content', content=lt)) + + # Codex rate-limit gauge row (always shown when Codex data is available) + codex_row = r.codex_rate_row(codex_rl, width) + rows.append(RowSpec('content', content=codex_row)) # First post-tokens separator threads `ups` back into the tokens vseps and # is drawn as the heavy "seam" marking the static→dynamic split. Only the diff --git a/test/test_claude_quota.py b/test/test_claude_quota.py new file mode 100644 index 0000000..ad229cc --- /dev/null +++ b/test/test_claude_quota.py @@ -0,0 +1,255 @@ +"""Tests for ClaudeQuota — Claude subscription quota approximation mode. + +Covers: +- Plan ceiling lookup (pro / max5 / max20) +- Env-var override beats plan default +- pct_5h math (session_tokens / cap_5h × 100) +- pct_weekly math (day_tokens / cap_weekly × 100) +- Clamping to 100% when tokens exceed ceiling +- YAS_CLAUDE_MODE=cost keeps existing cost rendering (line_tokens rows present) +- YAS_CLAUDE_MODE=quota swaps to gauge rendering (quota row present) +- Width-mode renders: wide / medium / narrow +- Colour threshold transitions (60% warn, 85% alert) +""" + +import pytest +import statusline_command as sl + +from helper import strip_ansi + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _quota(plan: str = 'max20', session_tokens: int = 0, day_tokens: int = 0, + monkeypatch: pytest.MonkeyPatch | None = None) -> sl.ClaudeQuota: + """Build a ClaudeQuota, optionally after clearing override env vars.""" + if monkeypatch is not None: + monkeypatch.delenv('YAS_CLAUDE_5H_CAP_TOKENS', raising=False) + monkeypatch.delenv('YAS_CLAUDE_WEEKLY_CAP_TOKENS', raising=False) + return sl.ClaudeQuota.load(plan=plan, session_tokens=session_tokens, day_tokens=day_tokens) + + +# --------------------------------------------------------------------------- +# Plan ceiling lookup +# --------------------------------------------------------------------------- + +class TestPlanCeilings: + def test_pro_ceilings(self, monkeypatch: pytest.MonkeyPatch) -> None: + q = _quota('pro', monkeypatch=monkeypatch) + assert q.cap_5h_tokens == 1_500_000 + assert q.cap_weekly_tokens == 18_000_000 + + def test_max5_ceilings(self, monkeypatch: pytest.MonkeyPatch) -> None: + q = _quota('max5', monkeypatch=monkeypatch) + assert q.cap_5h_tokens == 7_500_000 + assert q.cap_weekly_tokens == 90_000_000 + + def test_max20_ceilings(self, monkeypatch: pytest.MonkeyPatch) -> None: + q = _quota('max20', monkeypatch=monkeypatch) + assert q.cap_5h_tokens == 30_000_000 + assert q.cap_weekly_tokens == 360_000_000 + + def test_unknown_plan_falls_back_to_max20(self, monkeypatch: pytest.MonkeyPatch) -> None: + q = _quota('enterprise', monkeypatch=monkeypatch) + assert q.cap_5h_tokens == 30_000_000 + + +# --------------------------------------------------------------------------- +# Env-var override +# --------------------------------------------------------------------------- + +class TestEnvVarOverride: + def test_5h_cap_override(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv('YAS_CLAUDE_5H_CAP_TOKENS', '5000000') + monkeypatch.delenv('YAS_CLAUDE_WEEKLY_CAP_TOKENS', raising=False) + q = sl.ClaudeQuota.load(plan='pro', session_tokens=0, day_tokens=0) + assert q.cap_5h_tokens == 5_000_000 + + def test_weekly_cap_override(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv('YAS_CLAUDE_5H_CAP_TOKENS', raising=False) + monkeypatch.setenv('YAS_CLAUDE_WEEKLY_CAP_TOKENS', '100000000') + q = sl.ClaudeQuota.load(plan='pro', session_tokens=0, day_tokens=0) + assert q.cap_weekly_tokens == 100_000_000 + + def test_override_beats_plan_default(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv('YAS_CLAUDE_5H_CAP_TOKENS', '999') + q = sl.ClaudeQuota.load(plan='max20', session_tokens=0, day_tokens=0) + # max20 default is 30M but env override wins + assert q.cap_5h_tokens == 999 + + +# --------------------------------------------------------------------------- +# Percentage math +# --------------------------------------------------------------------------- + +class TestPctMath: + def test_pct_5h_half(self, monkeypatch: pytest.MonkeyPatch) -> None: + q = _quota('pro', session_tokens=750_000, monkeypatch=monkeypatch) + assert q.pct_5h == pytest.approx(50.0) + + def test_pct_5h_zero(self, monkeypatch: pytest.MonkeyPatch) -> None: + q = _quota('pro', session_tokens=0, monkeypatch=monkeypatch) + assert q.pct_5h == pytest.approx(0.0) + + def test_pct_weekly_quarter(self, monkeypatch: pytest.MonkeyPatch) -> None: + q = _quota('pro', day_tokens=4_500_000, monkeypatch=monkeypatch) + assert q.pct_weekly == pytest.approx(25.0) + + def test_pct_clamped_at_100(self, monkeypatch: pytest.MonkeyPatch) -> None: + # Tokens above ceiling → clamped to 100%, not above + q = _quota('pro', session_tokens=99_000_000, monkeypatch=monkeypatch) + assert q.pct_5h == pytest.approx(100.0) + + def test_pct_weekly_clamped_at_100(self, monkeypatch: pytest.MonkeyPatch) -> None: + q = _quota('pro', day_tokens=999_000_000, monkeypatch=monkeypatch) + assert q.pct_weekly == pytest.approx(100.0) + + +# --------------------------------------------------------------------------- +# Colour thresholds +# --------------------------------------------------------------------------- + +class TestClaudeQuotaColour: + def setup_method(self) -> None: + self.r = sl.Renderer() + + def test_under_60_is_safe(self) -> None: + assert self.r.claude_quota_pct_colour(0.0) == self.r.safe + assert self.r.claude_quota_pct_colour(59.9) == self.r.safe + + def test_60_to_85_is_warn(self) -> None: + assert self.r.claude_quota_pct_colour(60.0) == self.r.warn + assert self.r.claude_quota_pct_colour(84.9) == self.r.warn + + def test_above_85_is_alert(self) -> None: + assert self.r.claude_quota_pct_colour(85.0) == self.r.alert + assert self.r.claude_quota_pct_colour(100.0) == self.r.alert + + +# --------------------------------------------------------------------------- +# claude_quota_row renders — width modes +# --------------------------------------------------------------------------- + +class TestClaudeQuotaRow: + def setup_method(self) -> None: + self.r = sl.Renderer() + + def _q(self, pct_5h: float = 24.0, pct_weekly: float = 12.0, + plan: str = 'max20') -> sl.ClaudeQuota: + # Build directly to avoid env interference + cap_5h = 30_000_000 + cap_weekly = 360_000_000 + return sl.ClaudeQuota( + session_tokens=int(pct_5h / 100 * cap_5h), + day_tokens=int(pct_weekly / 100 * cap_weekly), + plan=plan, + cap_5h_tokens=cap_5h, + cap_weekly_tokens=cap_weekly, + pct_5h=pct_5h, + pct_weekly=pct_weekly, + ) + + def test_wide_contains_5h_and_7d(self) -> None: + row = self.r.claude_quota_row(self._q(), width=120) + plain = strip_ansi(row) + assert '5h' in plain + assert '7d' in plain + assert '24' in plain + assert '12' in plain + + def test_wide_contains_plan_label(self) -> None: + row = self.r.claude_quota_row(self._q(plan='max20'), width=120) + plain = strip_ansi(row) + assert 'max20x' in plain + + def test_wide_contains_bar_chars(self) -> None: + row = self.r.claude_quota_row(self._q(pct_5h=50.0, pct_weekly=50.0), width=120) + assert '▓' in row or '█' in row or '░' in row + + def test_medium_has_both_pcts(self) -> None: + row = self.r.claude_quota_row(self._q(), width=79) + plain = strip_ansi(row) + assert '5h' in plain + assert '7d' in plain + assert '24' in plain + assert '12' in plain + + def test_narrow_has_primary_only(self) -> None: + row = self.r.claude_quota_row(self._q(), width=50) + plain = strip_ansi(row) + assert '5h' in plain + assert '24' in plain + assert '7d' not in plain + + def test_pro_plan_label(self) -> None: + q = sl.ClaudeQuota( + session_tokens=0, day_tokens=0, + plan='pro', cap_5h_tokens=1_500_000, cap_weekly_tokens=18_000_000, + pct_5h=0.0, pct_weekly=0.0, + ) + row = self.r.claude_quota_row(q, width=120) + plain = strip_ansi(row) + # pro plan label is just 'pro', not 'prox' + assert 'pro' in plain + + def test_max5_plan_label(self) -> None: + q = sl.ClaudeQuota( + session_tokens=0, day_tokens=0, + plan='max5', cap_5h_tokens=7_500_000, cap_weekly_tokens=90_000_000, + pct_5h=0.0, pct_weekly=0.0, + ) + row = self.r.claude_quota_row(q, width=120) + plain = strip_ansi(row) + assert 'max5x' in plain + + def test_warn_colour_at_60pct(self) -> None: + q = self._q(pct_5h=60.0, pct_weekly=0.0) + row = self.r.claude_quota_row(q, width=120) + # warn ANSI code should appear + assert self.r.warn in row + + def test_alert_colour_at_85pct(self) -> None: + q = self._q(pct_5h=85.0, pct_weekly=0.0) + row = self.r.claude_quota_row(q, width=120) + assert self.r.alert in row + + +# --------------------------------------------------------------------------- +# YAS_CLAUDE_MODE integration — build_wide row selection +# --------------------------------------------------------------------------- + +class TestClaudeModeToggle: + """Smoke tests: verify that build_wide chooses the right row set. + + We don't call build_wide directly (requires a full SessionInfo), so we + test via ClaudeQuota.load() with the env vars that build_wide reads. + The row-level behaviour is covered by TestClaudeQuotaRow above. + """ + + def test_mode_cost_is_default(self, monkeypatch: pytest.MonkeyPatch) -> None: + """When YAS_CLAUDE_MODE is unset, mode resolves to 'cost'.""" + monkeypatch.delenv('YAS_CLAUDE_MODE', raising=False) + import os + assert os.environ.get('YAS_CLAUDE_MODE', 'cost').strip().lower() == 'cost' + + def test_mode_quota_activates_gauge(self, monkeypatch: pytest.MonkeyPatch) -> None: + """When YAS_CLAUDE_MODE=quota, ClaudeQuota is constructed.""" + monkeypatch.setenv('YAS_CLAUDE_MODE', 'quota') + monkeypatch.setenv('YAS_CLAUDE_PLAN', 'max20') + monkeypatch.delenv('YAS_CLAUDE_5H_CAP_TOKENS', raising=False) + monkeypatch.delenv('YAS_CLAUDE_WEEKLY_CAP_TOKENS', raising=False) + import os + mode = os.environ.get('YAS_CLAUDE_MODE', 'cost').strip().lower() + plan = os.environ.get('YAS_CLAUDE_PLAN', 'max20').strip().lower() + assert mode == 'quota' + q = sl.ClaudeQuota.load(plan=plan, session_tokens=500_000, day_tokens=1_000_000) + assert q.plan == 'max20' + assert q.pct_5h == pytest.approx(500_000 / 30_000_000 * 100) + + def test_plan_max20_is_default(self, monkeypatch: pytest.MonkeyPatch) -> None: + """When YAS_CLAUDE_PLAN is unset, plan resolves to max20.""" + monkeypatch.delenv('YAS_CLAUDE_PLAN', raising=False) + import os + assert os.environ.get('YAS_CLAUDE_PLAN', 'max20').strip().lower() == 'max20' diff --git a/test/test_codex_rate_limits.py b/test/test_codex_rate_limits.py new file mode 100644 index 0000000..88279a1 --- /dev/null +++ b/test/test_codex_rate_limits.py @@ -0,0 +1,245 @@ +"""Tests for CodexRateLimits — Codex rate-limit gauge (3rd statusline row). + +Covers: +- Loading from a synthetic rollouts dir with known values +- Empty rollouts dir → all-None fields +- Stale data → data_age_seconds > threshold +- Color threshold transitions (60%, 85%) +- Width-mode render: full / medium / narrow +""" + +import json +import time +from pathlib import Path + +import pytest +import statusline_command as sl + +from helper import strip_ansi + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _write_rollout(root: Path, year: int, month: int, day: int, events: list[dict]) -> Path: + """Write a rollout-0.jsonl file under /YYYY/MM/DD/.""" + day_dir = root / f'{year:04d}' / f'{month:02d}' / f'{day:02d}' + day_dir.mkdir(parents=True, exist_ok=True) + rollout = day_dir / 'rollout-0.jsonl' + with rollout.open('w') as fh: + for evt in events: + fh.write(json.dumps(evt) + '\n') + return rollout + + +def _token_count_event(primary_pct: float, secondary_pct: float, + primary_resets: int, secondary_resets: int, + plan_type: str = 'pro', + ts: str = '2026-05-25T03:47:10.921Z') -> dict: + return { + 'timestamp': ts, + 'type': 'event_msg', + 'payload': { + 'type': 'token_count', + 'rate_limits': { + 'primary': {'used_percent': primary_pct, 'window_minutes': 300, 'resets_at': primary_resets}, + 'secondary': {'used_percent': secondary_pct, 'window_minutes': 10080, 'resets_at': secondary_resets}, + 'plan_type': plan_type, + }, + }, + } + + +def _other_event() -> dict: + return {'timestamp': '2026-05-25T03:00:00.000Z', 'type': 'event_msg', 'payload': {'type': 'other'}} + + +# --------------------------------------------------------------------------- +# CodexRateLimits.load() — data loading tests +# --------------------------------------------------------------------------- + +class TestCodexRateLimitsLoad: + def test_load_known_values(self, tmp_path: Path) -> None: + now_epoch = int(time.time()) + primary_resets = now_epoch + 3600 + secondary_resets = now_epoch + 86400 + _write_rollout(tmp_path, 2026, 5, 25, [ + _token_count_event(12.0, 28.0, primary_resets, secondary_resets), + ]) + rl = sl.CodexRateLimits.load(tmp_path) + assert rl.primary_pct == pytest.approx(12.0) + assert rl.secondary_pct == pytest.approx(28.0) + assert rl.primary_resets_at == primary_resets + assert rl.secondary_resets_at == secondary_resets + assert rl.plan_type == 'pro' + + def test_load_picks_last_token_count(self, tmp_path: Path) -> None: + """When multiple token_count events exist, the LAST one wins.""" + now_epoch = int(time.time()) + _write_rollout(tmp_path, 2026, 5, 25, [ + _token_count_event(5.0, 10.0, now_epoch + 100, now_epoch + 200), + _other_event(), + _token_count_event(55.0, 77.0, now_epoch + 150, now_epoch + 250), + ]) + rl = sl.CodexRateLimits.load(tmp_path) + assert rl.primary_pct == pytest.approx(55.0) + assert rl.secondary_pct == pytest.approx(77.0) + + def test_empty_dir_returns_all_none(self, tmp_path: Path) -> None: + rl = sl.CodexRateLimits.load(tmp_path) + assert rl.primary_pct is None + assert rl.secondary_pct is None + assert rl.primary_resets_at is None + assert rl.secondary_resets_at is None + assert rl.plan_type is None + assert rl.data_age_seconds is None + + def test_no_token_count_events_returns_all_none(self, tmp_path: Path) -> None: + """A rollout file with only non-token_count events → all None.""" + _write_rollout(tmp_path, 2026, 5, 25, [_other_event(), _other_event()]) + rl = sl.CodexRateLimits.load(tmp_path) + assert rl.primary_pct is None + + def test_env_var_override(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + now_epoch = int(time.time()) + _write_rollout(tmp_path, 2026, 5, 25, [ + _token_count_event(33.0, 44.0, now_epoch + 100, now_epoch + 200), + ]) + monkeypatch.setenv('YAS_CODEX_SESSIONS_DIR', str(tmp_path)) + # load() with no argument should pick up the env var + rl = sl.CodexRateLimits.load() + assert rl.primary_pct == pytest.approx(33.0) + monkeypatch.delenv('YAS_CODEX_SESSIONS_DIR', raising=False) + + def test_data_age_seconds_recent(self, tmp_path: Path) -> None: + """data_age_seconds should be small for a just-written rollout.""" + now_epoch = int(time.time()) + _write_rollout(tmp_path, 2026, 5, 25, [ + _token_count_event(1.0, 2.0, now_epoch + 100, now_epoch + 200), + ]) + rl = sl.CodexRateLimits.load(tmp_path) + # The rollout was written <5s ago; age should be < 10s + assert rl.data_age_seconds is not None + assert rl.data_age_seconds < 10.0 + + def test_stale_indicator(self, tmp_path: Path) -> None: + """A rollout older than 3600s should be considered stale.""" + now_epoch = int(time.time()) + rollout = _write_rollout(tmp_path, 2026, 5, 25, [ + _token_count_event(1.0, 2.0, now_epoch + 100, now_epoch + 200), + ]) + # Back-date the file's mtime by 2 hours + old_mtime = time.time() - 7200 + import os + os.utime(rollout, (old_mtime, old_mtime)) + rl = sl.CodexRateLimits.load(tmp_path) + assert rl.data_age_seconds is not None + assert rl.data_age_seconds > 3600 + assert rl.is_stale() + + def test_plan_type_free(self, tmp_path: Path) -> None: + now_epoch = int(time.time()) + _write_rollout(tmp_path, 2026, 5, 25, [ + _token_count_event(0.0, 0.0, now_epoch + 100, now_epoch + 200, plan_type='free'), + ]) + rl = sl.CodexRateLimits.load(tmp_path) + assert rl.plan_type == 'free' + + +# --------------------------------------------------------------------------- +# Color threshold tests +# --------------------------------------------------------------------------- + +class TestCodexColorThresholds: + def setup_method(self) -> None: + self.r = sl.Renderer() + + def test_under_60_is_safe(self) -> None: + assert self.r.codex_pct_colour(0.0) == self.r.safe + assert self.r.codex_pct_colour(59.9) == self.r.safe + + def test_60_to_85_is_warn(self) -> None: + assert self.r.codex_pct_colour(60.0) == self.r.warn + assert self.r.codex_pct_colour(84.9) == self.r.warn + + def test_above_85_is_alert(self) -> None: + assert self.r.codex_pct_colour(85.0) == self.r.alert + assert self.r.codex_pct_colour(100.0) == self.r.alert + + +# --------------------------------------------------------------------------- +# Render tests — codex_rate_row +# --------------------------------------------------------------------------- + +class TestCodexRateRow: + def setup_method(self) -> None: + self.r = sl.Renderer() + + def _rl(self, primary: float = 12.0, secondary: float = 28.0, + plan: str = 'pro') -> sl.CodexRateLimits: + return sl.CodexRateLimits( + primary_pct=primary, + secondary_pct=secondary, + primary_resets_at=None, + secondary_resets_at=None, + plan_type=plan, + data_age_seconds=30.0, + ) + + def test_full_mode_contains_5h_and_7d(self) -> None: + row = self.r.codex_rate_row(self._rl(), width=120) + plain = strip_ansi(row) + assert '5h' in plain + assert '7d' in plain + assert '12' in plain # primary pct + assert '28' in plain # secondary pct + assert 'pro' in plain + + def test_full_mode_contains_plan(self) -> None: + row = self.r.codex_rate_row(self._rl(plan='free'), width=120) + plain = strip_ansi(row) + assert 'free' in plain + + def test_medium_mode_has_both_pcts(self) -> None: + row = self.r.codex_rate_row(self._rl(), width=79) + plain = strip_ansi(row) + assert '5h' in plain + assert '7d' in plain + assert '12' in plain + assert '28' in plain + + def test_narrow_mode_has_primary_only(self) -> None: + row = self.r.codex_rate_row(self._rl(), width=50) + plain = strip_ansi(row) + assert '5h' in plain + assert '12' in plain + # In narrow mode we may omit 7d label and secondary pct entirely + assert '7d' not in plain + + def test_no_data_state(self) -> None: + rl = sl.CodexRateLimits( + primary_pct=None, secondary_pct=None, + primary_resets_at=None, secondary_resets_at=None, + plan_type=None, data_age_seconds=None, + ) + row = self.r.codex_rate_row(rl, width=120) + plain = strip_ansi(row) + assert 'no codex data' in plain.lower() or plain.strip() == '' + + def test_stale_shows_indicator(self) -> None: + rl = sl.CodexRateLimits( + primary_pct=5.0, secondary_pct=10.0, + primary_resets_at=None, secondary_resets_at=None, + plan_type='pro', data_age_seconds=7200.0, + ) + row = self.r.codex_rate_row(rl, width=120) + plain = strip_ansi(row) + # Stale indicator: either "stale" text or a warning symbol + assert 'stale' in plain.lower() or '?' in plain or '⚠' in plain + + def test_bar_chars_present_in_full_mode(self) -> None: + """Full mode should contain filled/empty bar chars.""" + row = self.r.codex_rate_row(self._rl(primary=50.0, secondary=50.0), width=120) + # Bar characters: ▓ (heavy) or █ (filled) and ░ (empty) + assert '▓' in row or '█' in row or '░' in row diff --git a/test/test_context_gauge_softlimit.py b/test/test_context_gauge_softlimit.py new file mode 100644 index 0000000..ca78fd2 --- /dev/null +++ b/test/test_context_gauge_softlimit.py @@ -0,0 +1,181 @@ +"""Tests for _effective_soft_limit — context gauge scales to model context window. + +Covers: +- _effective_soft_limit returns 75% of context_window_size for large models +- _effective_soft_limit returns legacy 150K floor for 200K-and-below models +- _effective_soft_limit returns legacy 150K fallback when context_window_size is 0 +- context_line() renders proportional % for 1M-context models (not overflowed) +- context_line() still triggers warning zone for 200K models near soft limit +- context_line_compact() mirrors the same proportional/warning behaviour +- build_wide / build_medium / build_narrow LayoutSpec.fill reflects scaled limit +""" + +import re + +import pytest +import statusline_command as sl + +from helper import strip_ansi + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _ctx(total_tokens: int, context_window_size: int) -> sl.ContextWindow: + """Build a ContextWindow with all tokens as input.""" + return sl.ContextWindow( + total_input_tokens=total_tokens, + total_output_tokens=0, + context_window_size=context_window_size, + ) + + +# --------------------------------------------------------------------------- +# _effective_soft_limit unit tests +# --------------------------------------------------------------------------- + +class TestEffectiveSoftLimit: + def test_1m_model_returns_750k(self) -> None: + ctx = _ctx(0, context_window_size=1_000_000) + assert sl._effective_soft_limit(ctx) == 750_000 + + def test_200k_model_returns_150k_floor(self) -> None: + # 75% of 200K = 150K, which equals the floor exactly + ctx = _ctx(0, context_window_size=200_000) + assert sl._effective_soft_limit(ctx) == 150_000 + + def test_unknown_size_returns_150k_fallback(self) -> None: + ctx = _ctx(0, context_window_size=0) + assert sl._effective_soft_limit(ctx) == 150_000 + + def test_small_window_floor_wins(self) -> None: + # 75% of 100K = 75K, which is below 150K floor + ctx = _ctx(0, context_window_size=100_000) + assert sl._effective_soft_limit(ctx) == 150_000 + + +# --------------------------------------------------------------------------- +# context_line rendering — proportional fill for 1M models +# --------------------------------------------------------------------------- + +class TestContextLineSoftLimit: + def setup_method(self) -> None: + self.r = sl.Renderer() + + def test_1m_model_170k_tokens_shows_proportional_pct(self) -> None: + # 170K / 750K soft_limit = 22.7% → should show 22% or 23%, NOT 113% + ctx = _ctx(170_000, context_window_size=1_000_000) + line = self.r.context_line(ctx, available=76) + plain = strip_ansi(line) + assert '22%' in plain or '23%' in plain, ( + f'Expected ~22-23% for 170K on 1M-context model, got: {plain!r}' + ) + assert '113%' not in plain, ( + f'Old broken value 113% should not appear; got: {plain!r}' + ) + + def test_200k_model_180k_tokens_triggers_warning(self) -> None: + # 180K / 150K soft_limit = 120% → warning zone (>=100% of soft) + ctx = _ctx(180_000, context_window_size=200_000) + line = self.r.context_line(ctx, available=76) + plain = strip_ansi(line) + # The headline pct_soft should be >= 100 (e.g. "120%") + pct_values = [int(m) for m in re.findall(r'(\d+)%', plain)] + assert any(v >= 100 for v in pct_values), ( + f'Expected >=100% soft pct for 180K on 200K model, got pcts: {pct_values} in {plain!r}' + ) + + +# --------------------------------------------------------------------------- +# context_line_compact rendering — mirrors proportional/warning behaviour +# --------------------------------------------------------------------------- + +class TestContextLineCompactSoftLimit: + def setup_method(self) -> None: + self.r = sl.Renderer() + + def test_1m_model_170k_tokens_shows_proportional_pct(self) -> None: + ctx = _ctx(170_000, context_window_size=1_000_000) + line = self.r.context_line_compact(ctx, available=40) + plain = strip_ansi(line) + assert '22%' in plain or '23%' in plain, ( + f'Expected ~22-23% for 170K on 1M-context model (compact), got: {plain!r}' + ) + assert '113%' not in plain, ( + f'Old broken value 113% should not appear (compact); got: {plain!r}' + ) + + def test_200k_model_180k_tokens_triggers_warning(self) -> None: + ctx = _ctx(180_000, context_window_size=200_000) + line = self.r.context_line_compact(ctx, available=40) + plain = strip_ansi(line) + pct_values = [int(m) for m in re.findall(r'(\d+)%', plain)] + assert any(v >= 100 for v in pct_values), ( + f'Expected >=100% soft pct for 180K on 200K model (compact), got pcts: {pct_values} in {plain!r}' + ) + + +# --------------------------------------------------------------------------- +# build_wide / build_medium / build_narrow — LayoutSpec.fill uses scaled limit +# +# fill = min(total_tokens / _effective_soft_limit(ctx), 1.0) +# On a 1M-context model with 170K tokens: 170_000 / 750_000 ≈ 0.2267 +# With bare SOFT_LIMIT (150K): 170_000 / 150_000 ≈ 1.133 → clamped 1.0 +# +# We assert fill < 0.5, which is satisfied by ~0.227 and would NOT be satisfied +# by the clamped 1.0 from the old bug. The exact expected value 170/750 ≈ 0.2267 +# is also checked to within floating-point tolerance. +# --------------------------------------------------------------------------- + +def _session_1m(total_tokens: int) -> sl.SessionInfo: + """Minimal SessionInfo: 1M context window, given total_tokens as input.""" + ctx = sl.ContextWindow( + total_input_tokens=total_tokens, + total_output_tokens=0, + context_window_size=1_000_000, + ) + session = sl.SessionInfo() + session.context_window = ctx + return session + + +class TestBuildFunctionsSoftLimit: + """build_wide/medium/narrow LayoutSpec.fill must use _effective_soft_limit.""" + + def setup_method(self) -> None: + self.r = sl.Renderer() + + def test_build_wide_does_not_overflow_on_1m_model(self) -> None: + # 170K tokens on 1M-context model → soft_limit=750K → fill≈0.227, not 1.0 + session = _session_1m(170_000) + spec = sl.build_wide(session, width=120, r=self.r) + expected_fill = 170_000 / 750_000 + assert spec.fill < 0.5, ( + f'fill={spec.fill:.4f} should be ~0.227 on 1M model; old bug produced 1.0' + ) + assert abs(spec.fill - expected_fill) < 1e-6, ( + f'fill={spec.fill:.6f}, expected {expected_fill:.6f}' + ) + + def test_build_medium_does_not_overflow_on_1m_model(self) -> None: + session = _session_1m(170_000) + spec = sl.build_medium(session, width=68, r=self.r) + expected_fill = 170_000 / 750_000 + assert spec.fill < 0.5, ( + f'fill={spec.fill:.4f} should be ~0.227 on 1M model; old bug produced 1.0' + ) + assert abs(spec.fill - expected_fill) < 1e-6, ( + f'fill={spec.fill:.6f}, expected {expected_fill:.6f}' + ) + + def test_build_narrow_does_not_overflow_on_1m_model(self) -> None: + session = _session_1m(170_000) + spec = sl.build_narrow(session, width=48, r=self.r) + expected_fill = 170_000 / 750_000 + assert spec.fill < 0.5, ( + f'fill={spec.fill:.4f} should be ~0.227 on 1M model; old bug produced 1.0' + ) + assert abs(spec.fill - expected_fill) < 1e-6, ( + f'fill={spec.fill:.6f}, expected {expected_fill:.6f}' + )