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
45 changes: 39 additions & 6 deletions src/ccbot/bot/_usage_window.py
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,30 @@ async def _capture_with_scrollback(wid: str) -> str | None:
return None


async def _clear_pane_history(wid: str) -> None:
"""Drop the pane's scrollback buffer.

``_capture_with_scrollback`` reads ~100 lines back, which can still
contain a PREVIOUS ``/usage`` render. After a weekly reset that stale
render shows last week's high percentage (e.g. 78%) while the live
week is ~2% — and the parser, walking both, can surface the stale
value and fire a phantom quota alert. Clearing history before each
read guarantees the parse only sees the fresh modal.
"""
try:
proc = await asyncio.create_subprocess_exec(
"tmux",
"clear-history",
"-t",
wid,
stdout=asyncio.subprocess.DEVNULL,
stderr=asyncio.subprocess.DEVNULL,
)
await proc.communicate()
except Exception as e:
logger.debug("clear_pane_history failed: %s", e)


async def _poll_usage_modal(wid: str) -> object | None:
"""Send /usage, poll the pane for quota rows, dismiss with Escape.

Expand All @@ -83,14 +107,22 @@ async def _poll_usage_modal(wid: str) -> object | None:
value that later updated under us (observed: bot showed 43%
while the modal had since stabilised at 45%). Now we require
TWO consecutive captures with identical session / week / week-
Sonnet percentages before we trust the read.
Sonnet percentages before we trust the read. If the modal never
settles within the budget we return ``None`` (caller treats it as
"unavailable") rather than publishing an unsettled / stale frame —
that was firing phantom quota alerts.
"""
from ..terminal_parser import extract_usage_breakdown, parse_usage_output

info = None
resolved = False
last_triple: tuple[int | None, int | None, int | None] | None = None
try:
# Dismiss any leftover modal / feedback prompt so /usage opens
# fresh, and drop stale scrollback so a previous (pre-reset)
# render can't be parsed as the current week.
await tmux_manager.send_keys(wid, "Escape", enter=False, literal=False)
await _clear_pane_history(wid)
await tmux_manager.send_keys(wid, "/usage")
for _ in range(60): # 60 × 200 ms = 12 s
await asyncio.sleep(0.2)
Expand Down Expand Up @@ -121,11 +153,12 @@ async def _poll_usage_modal(wid: str) -> object | None:
except Exception as e:
logger.debug("fetch_claude_usage: tmux failed: %s", e)
return None
# If we ran out of polls before two-agreement, still return the
# last seen result rather than nothing — better a 1-step-stale
# value than the "unavailable" empty state. The 12-s budget should
# normally be plenty for the modal to settle.
return info if (resolved or info is not None) else None
# Only publish a SETTLED read (two consecutive agreeing captures). An
# unsettled frame — a transitional value, or a stale pre-reset render
# lingering in scrollback — was being published and tripping phantom
# quota-threshold alerts. A missed poll (None) is harmless; the loop
# retries on the next cadence.
return info if resolved else None


async def fetch_claude_usage() -> object | None:
Expand Down
81 changes: 81 additions & 0 deletions tests/ccbot/bot/test_usage_window.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,81 @@
"""``_poll_usage_modal`` must only publish a SETTLED /usage read.

An unsettled frame — a transitional value mid-load, or a stale pre-reset
render still in scrollback — used to be returned via a fallback, and the
quota-alerts loop turned it into a phantom threshold crossing (observed:
``week: 78%`` pushed while the live week was 2%). The poller now returns
``None`` unless two consecutive captures agree.
"""

from __future__ import annotations

import itertools
from collections.abc import Iterator

import pytest

from ccbot.bot import _usage_window
from ccbot.terminal_parser import extract_usage_breakdown


def _frame(session: int, week: int, sonnet: int) -> str:
"""A minimal /usage modal body the real parser understands."""
return (
f"Current session\n█ {session}% used\nResets 4pm (UTC)\n\n"
f"Current week (all models)\n█ {week}% used\nResets Jun 21 at 1pm (UTC)\n\n"
f"Current week (Sonnet)\n█ {sonnet}% used\nResets Jun 21 at 1pm (UTC)\n"
)


@pytest.fixture(autouse=True)
def _fast_and_isolated(monkeypatch: pytest.MonkeyPatch) -> None:
"""No real tmux, no real 200 ms sleeps."""

async def _noop_sleep(*_a: object, **_k: object) -> None:
return None

async def _noop_send(*_a: object, **_k: object) -> bool:
return True

async def _noop_clear(*_a: object, **_k: object) -> None:
return None

monkeypatch.setattr(_usage_window.asyncio, "sleep", _noop_sleep)
monkeypatch.setattr(_usage_window.tmux_manager, "send_keys", _noop_send)
monkeypatch.setattr(_usage_window, "_clear_pane_history", _noop_clear)


def _capture_returning(frames: Iterator[str]):
async def _cap(_wid: str) -> str:
return next(frames)

return _cap


@pytest.mark.asyncio
async def test_settled_read_is_published(monkeypatch: pytest.MonkeyPatch):
# Two consecutive identical captures → settled → value returned.
frames = iter([_frame(7, 2, 0), _frame(7, 2, 0)])
monkeypatch.setattr(
_usage_window, "_capture_with_scrollback", _capture_returning(frames)
)

info = await _usage_window._poll_usage_modal("@1")

assert info is not None
assert extract_usage_breakdown(info).week_pct == 2


@pytest.mark.asyncio
async def test_unsettled_read_returns_none(monkeypatch: pytest.MonkeyPatch):
# Week oscillates forever; no two consecutive captures ever agree, so
# the modal never settles. The OLD code returned the last (78%) frame;
# the fix returns None so the quota loop can't fire a phantom alert.
frames = itertools.cycle([_frame(7, 78, 0), _frame(7, 2, 0)])
monkeypatch.setattr(
_usage_window, "_capture_with_scrollback", _capture_returning(frames)
)

info = await _usage_window._poll_usage_modal("@1")

assert info is None
Loading