Skip to content
Open
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
17 changes: 11 additions & 6 deletions src/claude_statusbar/scoped_usage.py
Original file line number Diff line number Diff line change
Expand Up @@ -69,10 +69,13 @@ def refresh(account):
if 0 <= time.time() - old['ts'] < 300:
return
except (OSError, ValueError, KeyError, TypeError):
pass
old = {}
if account_id() != account:
return
limits = []
# A failed fetch keeps the last good limits (still bounded by ok_ts in
# cached_limits) instead of blanking the segment until the next retry.
limits = old.get('limits', [])
ok_ts = old.get('ok_ts', old.get('ts'))
try:
token = _token()
if not isinstance(token, str) or not token:
Expand All @@ -86,10 +89,11 @@ def redirect_request(self, *args, **kwargs):
'anthropic-beta': 'oauth-2025-04-20'})
with build_opener(NoRedirect).open(req, timeout=3) as response:
limits = parse_limits(json.loads(response.read(1024 * 1024)))
ok_ts = time.time()
except Exception:
pass # Negative-cache failures; never expose credentials/errors.
pass # Keep last good limits; ts still backs off retries. Never expose credentials/errors.
if account_id() == account:
atomic_write_text(path, json.dumps(dict(ts=time.time(), limits=limits)))
atomic_write_text(path, json.dumps(dict(ts=time.time(), ok_ts=ok_ts, limits=limits)))
finally:
lock.close()

Expand All @@ -102,12 +106,13 @@ def cached_limits(*, spawn=True):
try:
data = json.loads(_path(account).read_text())
age = time.time() - data['ts']
ok_age = time.time() - (data.get('ok_ts') or data['ts'])
except (OSError, ValueError, KeyError, TypeError):
data, age = {}, float('inf')
data, age, ok_age = {}, float('inf'), float('inf')
if spawn and not 0 <= age < 300:
from .refresh_pool import submit
submit(('scoped', account), refresh, account)
return data.get('limits', []) if 0 <= age < 600 else []
return data.get('limits', []) if 0 <= ok_age < 600 else []


def render_limits(limits, theme, use_color, warning, critical, projection=True):
Expand Down
37 changes: 37 additions & 0 deletions tests/test_performance_pipeline.py
Original file line number Diff line number Diff line change
Expand Up @@ -114,6 +114,43 @@ def test_scoped_auth_failure_is_cached(tmp_path, monkeypatch):
assert 'token' not in scoped_usage._path('fixture-account').read_text()


def test_scoped_failed_refresh_keeps_last_good_limits(tmp_path, monkeypatch):
from claude_statusbar import scoped_usage, predict
monkeypatch.setenv('HOME', str(tmp_path))
monkeypatch.setattr(predict, 'account_id', lambda: 'fixture-account')
monkeypatch.setattr(scoped_usage, '_token', lambda: None) # every fetch fails
good = [dict(label='Fable', percent=40.0, resets_at=time.time() + 86400)]
path = scoped_usage._path('fixture-account')
path.parent.mkdir(parents=True)
path.write_text(json.dumps(dict(ts=time.time() - 301, limits=good)))
scoped_usage.refresh('fixture-account')
assert scoped_usage.cached_limits(spawn=False) == good # one blip: still shown
data = json.loads(path.read_text())
path.write_text(json.dumps(dict(data, ok_ts=time.time() - 601)))
assert scoped_usage.cached_limits(spawn=False) == [] # long outage: hidden


def test_scoped_successful_refresh_stamps_ok_ts(tmp_path, monkeypatch):
import io
import urllib.request
from claude_statusbar import scoped_usage, predict
monkeypatch.setenv('HOME', str(tmp_path))
monkeypatch.setattr(predict, 'account_id', lambda: 'fixture-account')
monkeypatch.setattr(scoped_usage, '_token', lambda: 'tok')
reset = (datetime.now(timezone.utc) + timedelta(days=1)).isoformat()
body = json.dumps({'limits': [{'kind': 'weekly_scoped', 'percent': 40,
'resets_at': reset, 'scope': {'model': {'display_name': 'Fable'}}}]}).encode()

class Opener:
def open(self, req, timeout):
return io.BytesIO(body)
monkeypatch.setattr(urllib.request, 'build_opener', lambda *a: Opener())
scoped_usage.refresh('fixture-account')
data = json.loads(scoped_usage._path('fixture-account').read_text())
assert abs(data['ok_ts'] - time.time()) < 5
assert [r['label'] for r in scoped_usage.cached_limits(spawn=False)] == ['Fable']


def test_scoped_account_switch_never_reuses_previous_account(tmp_path, monkeypatch):
from claude_statusbar import scoped_usage, predict
monkeypatch.setenv('HOME', str(tmp_path))
Expand Down
Loading