diff --git a/src/claude_statusbar/scoped_usage.py b/src/claude_statusbar/scoped_usage.py index 93baa8b..71235d7 100644 --- a/src/claude_statusbar/scoped_usage.py +++ b/src/claude_statusbar/scoped_usage.py @@ -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: @@ -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() @@ -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): diff --git a/tests/test_performance_pipeline.py b/tests/test_performance_pipeline.py index 182140f..9fcc896 100644 --- a/tests/test_performance_pipeline.py +++ b/tests/test_performance_pipeline.py @@ -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))