#!/usr/bin/env python3 """Sandboxed integration tests for automatic Claude limit-reset redemption. The twin of tests/test_codex_reset*.py. A stateful loopback server plays the OAuth usage endpoint (with the ``cedar_ember`` status block Claude Code 2.1.280 reads) and ``POST /api/organizations//reset_rate_limits``, and ``claude-accounts limits`` runs against it with a file-held OAuth login — no network, no real accounts. """ from __future__ import annotations import copy from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import json import os from pathlib import Path import subprocess import tempfile import threading import time import unittest import urllib.parse import uuid REPO = Path(__file__).resolve().parents[1] ORG = "11111111-2222-4333-8444-555555555555" USER = "99999999-8888-4777-8666-555555555555" GRANT = "opus55-launch-promax-20260921" CLI_UA = "claude-cli/2.1.280 (external, cli, client-app/claude-multiacc)" def iso(offset: int) -> str: return time.strftime("%Y-%m-%dT%H:%M:%S+00:00", time.gmtime(time.time() + offset)) def usage_doc(*, session=0, weekly=50, fable=50, exhausted=(), resets_left=1, use_requires_limit=False, eligible=True, reason=None, blocking=(), clears=("five_hour", "seven_day", "seven_day_overage_included"), usable=True, week_resets_in=3 * 86400, session_resets_in=3600) -> dict: """A usage response in the live 2026-09-22 shape, cedar_ember block included.""" grants = [{ "id": GRANT, "label": "Claude Opus 5.5 launch: one usage-limit reset for Pro and Max", "resets_total": 1, "resets_left": resets_left, "starts_at": iso(-3600), "ends_at": iso(30 * 86400), "clears": list(clears), "paused": False, "usable_now": usable and resets_left > 0, "use_requires_limit": use_requires_limit, "percent_used": {"five_hour": session, "seven_day": weekly, "seven_day_overage_included": fable}, "blocking": list(blocking), "arm": None}] if eligible else [] return { "limits": [ {"kind": "session", "group": "session", "percent": session, "resets_at": iso(session_resets_in) if session else None, "scope": None}, {"kind": "weekly_all", "group": "weekly", "percent": weekly, "resets_at": iso(week_resets_in), "scope": None}, {"kind": "weekly_scoped", "group": "weekly", "percent": fable, "resets_at": iso(week_resets_in), "scope": {"model": {"id": None, "display_name": "Fable"}, "surface": None}}], "five_hour": {"utilization": float(session), "resets_at": None}, "seven_day": {"utilization": float(weekly), "resets_at": iso(week_resets_in)}, "cedar_ember": { "eligible": eligible, "ineligible_reason": reason, "at_limit": bool(exhausted), "exhausted": list(exhausted), "grants": grants, "next_grant_id": GRANT if grants else None, "weekly_resets_at": iso(week_resets_in), "cooldown_until": None, "event_props": None}, "juniper_tide": None, } class ResetHandler(BaseHTTPRequestHandler): def _send(self, status: int, body: dict, headers: dict | None = None) -> None: encoded = json.dumps(body).encode() self.send_response(status) for name, value in (headers or {}).items(): self.send_header(name, value) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(encoded))) self.end_headers() self.wfile.write(encoded) def do_GET(self) -> None: # noqa: N802 - BaseHTTPRequestHandler contract state = self.server.state parsed = urllib.parse.urlsplit(self.path) if parsed.path != "/api/oauth/usage": self.send_error(404) return state["usage_gets"].append({"query": parsed.query, "ua": self.headers.get("User-Agent"), "beta": self.headers.get("anthropic-beta")}) body = copy.deepcopy(state["usage"]) if state.get("surface_gate") and "(external, cli" not in str(self.headers.get("User-Agent")): # The live server's answer to a client it does not recognize as the CLI. body["cedar_ember"] = {"eligible": False, "ineligible_reason": "surface", "at_limit": False, "exhausted": [], "grants": [], "next_grant_id": None, "weekly_resets_at": None, "cooldown_until": None, "event_props": None} self._send(200, body) def do_POST(self) -> None: # noqa: N802 - BaseHTTPRequestHandler contract state = self.server.state if self.path != f"/api/organizations/{ORG}/reset_rate_limits": self.send_error(404) return length = int(self.headers.get("Content-Length") or 0) state["posts"].append({"body": json.loads(self.rfile.read(length)), "ua": self.headers.get("User-Agent"), "auth": self.headers.get("Authorization"), "beta": self.headers.get("anthropic-beta")}) status, body, applied, *extra = state["responses"].pop(0) if applied: # The reset landed server-side (whether or not the response makes it back). block = state["usage"]["cedar_ember"] for grant in block["grants"]: grant["resets_left"] = max(0, grant["resets_left"] - 1) grant["usable_now"] = grant["resets_left"] > 0 block["exhausted"], block["at_limit"] = [], False for limit in state["usage"]["limits"]: limit["percent"] = 0 self._send(status, body, extra[0] if extra else None) def log_message(self, _format: str, *_args) -> None: return class ClaudePoolSandbox: """A one-account claude pool whose OAuth login is a plain .credentials.json.""" def setUp(self) -> None: self.temp = tempfile.TemporaryDirectory(prefix="multiacc-claude-reset-") self.pool = Path(self.temp.name) / "pool" self.account = self.pool / "acct-01" self.account.mkdir(parents=True) manifest = {"version": 1, "threshold": 90, "accounts": [ {"id": "acct-01", "email": "test@example.invalid", "home": "mac"}]} (self.pool / "accounts.json").write_text(json.dumps(manifest), encoding="utf-8") creds = {"claudeAiOauth": { "accessToken": "sk-ant-oat01-test-access", "refreshToken": "test-refresh", "expiresAt": int(time.time() + 86400) * 1000, "refreshTokenExpiresAt": int(time.time() + 30 * 86400) * 1000, "scopes": ["user:inference", "user:profile"], "subscriptionType": "max"}} (self.account / ".credentials.json").write_text(json.dumps(creds), encoding="utf-8") (self.account / ".claude.json").write_text(json.dumps({"oauthAccount": { "organizationUuid": ORG, "accountUuid": USER}}), encoding="utf-8") self.server = ThreadingHTTPServer(("127.0.0.1", 0), ResetHandler) self.server.state = {"usage_gets": [], "posts": [], "responses": [], "usage": usage_doc()} self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) self.thread.start() def tearDown(self) -> None: self.server.shutdown() self.server.server_close() self.thread.join(timeout=5) self.temp.cleanup() def env(self, **overrides) -> dict: base = f"http://127.0.0.1:{self.server.server_port}" env = os.environ.copy() for key in ("CLAUDE_ACCOUNTS_ROOT", "CLAUDE_MULTIACC_RESET_URL", "CLAUDE_MULTIACC_TOKEN_URL"): env.pop(key, None) env.update({ "CLAUDE_ACCOUNTS_DIR": str(self.pool), "CLAUDE_MULTIACC_NO_SYNC": "1", "CLAUDE_MULTIACC_NO_DISTRIBUTE": "1", "CLAUDE_MULTIACC_MIN_FETCH": "0", "CLAUDE_MULTIACC_KEYCHAIN": "0", "CLAUDE_MULTIACC_PATH_PROBE": "0", "CLAUDE_MULTIACC_AUTO_RESET": "1", "CLAUDE_MULTIACC_CLI_VERSION": "2.1.280", "PYTHONDONTWRITEBYTECODE": "1", "CLAUDE_MULTIACC_USAGE_URL": f"{base}/api/oauth/usage?cedar_ember=1&skip_spend=1"}) env.update(overrides) return env def run_limits(self, *extra: str, **overrides) -> subprocess.CompletedProcess: result = subprocess.run([REPO / "bin/claude-accounts", "limits", "--force", *extra], capture_output=True, text=True, env=self.env(**overrides), timeout=60, check=False) self.assertEqual(result.returncode, 0, result.stdout + result.stderr) return result def limits_doc(self) -> dict: return json.loads((self.account / "limits.json").read_text(encoding="utf-8")) def reset_state(self) -> dict: return json.loads((self.account / ".usage-reset.json").read_text(encoding="utf-8")) def log(self) -> str: path = self.pool / "limits.log" return path.read_text(encoding="utf-8") if path.exists() else "" class ClaudeResetIntegrationTest(ClaudePoolSandbox, unittest.TestCase): def test_redeems_an_exhausted_limit_once(self) -> None: self.server.state["usage"] = usage_doc(fable=100, exhausted=["seven_day_overage_included"]) self.server.state["responses"] = [(200, { "result": "reset", "resets_left": 0, "cleared": ["five_hour", "seven_day", "seven_day_overage_included"], "weekly_resets_at": "2026-09-26T00:00:00+00:00"}, True)] self.run_limits() posts = self.server.state["posts"] self.assertEqual(len(posts), 1) body = posts[0]["body"] self.assertEqual(body["program"], "cedar_ember") self.assertEqual(body["grant_id"], GRANT) uuid.UUID(body["request_id"]) self.assertEqual(posts[0]["auth"], "Bearer sk-ant-oat01-test-access") self.assertEqual(posts[0]["beta"], "oauth-2025-04-20") self.assertEqual(posts[0]["ua"], CLI_UA) # The status rides on the one usage call, asked for the way the client asks. gets = self.server.state["usage_gets"] self.assertEqual(len(gets), 1) self.assertEqual(gets[0]["query"], "cedar_ember=1&skip_spend=1") self.assertEqual(gets[0]["ua"], CLI_UA) state = self.reset_state() self.assertEqual(state["state"], "complete") self.assertEqual(state["outcome"], "reset") doc = self.limits_doc() self.assertEqual(doc["fetched_at"], 0) # pre-reset numbers cannot exclude self.assertNotIn("weekly_resets_epoch", doc) # ...nor rank as a lower bound self.assertEqual(doc["auto_reset"]["status"], "redeemed") self.assertEqual(doc["reset_credits_available"], 0) # the server's own count self.assertFalse((self.account / ".limited").exists()) self.assertIn("limit reset redeemed automatically", self.log()) # The next pass sees a spent grant and a refilled account: nothing more to do. self.run_limits() self.assertEqual(len(self.server.state["posts"]), 1) self.assertFalse((self.account / ".limited").exists()) def test_early_use_grant_redeems_at_five_percent_remaining(self) -> None: self.server.state["usage"] = usage_doc(fable=95) self.server.state["responses"] = [(200, {"result": "reset", "resets_left": 0, "cleared": ["seven_day_overage_included"]}, True)] self.run_limits() self.assertEqual(len(self.server.state["posts"]), 1) def test_does_not_redeem_below_the_pools_own_threshold(self) -> None: self.server.state["usage"] = usage_doc(session=89, weekly=80, fable=89) self.run_limits() self.assertEqual(self.server.state["posts"], []) self.assertFalse((self.account / ".usage-reset.json").exists()) self.assertEqual(self.limits_doc()["reset_credits_available"], 1) def test_a_week_parked_at_the_pools_threshold_is_reset(self) -> None: # The live acct-15 shape (2026-09-22): weekly_all 93% parks the account for six # days at the pool's 90% line; waiting for 95% left its reset unused. self.server.state["usage"] = usage_doc(session=100, weekly=93, fable=64, session_resets_in=5400) self.server.state["responses"] = [(200, {"result": "reset", "resets_left": 0}, True)] self.run_limits() self.assertEqual(len(self.server.state["posts"]), 1) self.assertFalse((self.account / ".limited").exists()) def park(self, bucket: str, lifts_in: int, reason: str = "client-rate-limit") -> Path: marker = self.account / ".limited" marked = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(time.time() - 600)) marker.write_text(f"{int(time.time()) + lifts_in}\nbucket={bucket} percent=100 " f"marked_at={marked} reason={reason}\n", encoding="utf-8") return marker def test_a_park_the_telemetry_cannot_see_is_reset(self) -> None: # acct-13 (a client five-hour 429 while OAuth read 0% everywhere) and acct-19 (the # panel's weekly park while OAuth read 24%): parked is parked. for bucket, lifts_in in (("client:five_hour", 5400), ("weekly_all", 3 * 86400), ("panel:observed", 86400)): with self.subTest(bucket=bucket): self.server.state["posts"] = [] self.server.state["usage"] = usage_doc(session=0, weekly=19, fable=24) self.server.state["responses"] = [(200, {"result": "reset", "resets_left": 0}, True)] (self.account / ".usage-reset.json").unlink(missing_ok=True) (self.account / "limits.json").unlink(missing_ok=True) marker = self.park(bucket, lifts_in) self.run_limits() self.assertEqual(len(self.server.state["posts"]), 1) self.assertFalse(marker.exists()) def test_an_error_cooldown_is_not_a_usage_park(self) -> None: self.park("session", 600, reason="error-cooldown") self.server.state["usage"] = usage_doc(weekly=19, fable=24) self.run_limits() self.assertEqual(self.server.state["posts"], []) def test_a_five_hour_wall_that_lifts_within_the_hour_keeps_its_reset(self) -> None: self.server.state["usage"] = usage_doc(session=100, weekly=40, fable=40, exhausted=["five_hour"], session_resets_in=1800) self.run_limits() self.assertEqual(self.server.state["posts"], []) self.assertIn("limit reset held", self.log()) self.assertTrue((self.account / ".limited").exists()) # still parked for the 5h # ...but one that holds the account for longer is reset. self.server.state["usage"] = usage_doc(session=100, weekly=40, fable=40, exhausted=["five_hour"], session_resets_in=5400) self.server.state["responses"] = [(200, {"result": "reset", "resets_left": 0}, True)] self.run_limits() self.assertEqual(len(self.server.state["posts"]), 1) def test_a_five_hour_wall_beside_a_spent_week_is_redeemed(self) -> None: # The live acct-17 shape of 2026-09-22: session 100% AND Fable weekly at 95%. self.server.state["usage"] = usage_doc(session=100, weekly=65, fable=95, exhausted=["five_hour"]) self.server.state["responses"] = [(200, {"result": "reset", "resets_left": 0}, True)] self.run_limits() self.assertEqual(len(self.server.state["posts"]), 1) self.assertFalse((self.account / ".limited").exists()) def test_a_week_that_refills_within_the_hour_keeps_its_reset(self) -> None: self.server.state["usage"] = usage_doc(fable=100, exhausted=["seven_day_overage_included"], week_resets_in=1800) self.run_limits() self.assertEqual(self.server.state["posts"], []) self.assertIn("limit reset held", self.log()) self.assertFalse((self.account / ".usage-reset.json").exists()) # The horizon is the operator's to move. self.server.state["responses"] = [(200, {"result": "reset", "resets_left": 0}, True)] self.run_limits(CLAUDE_MULTIACC_RESET_MIN_HORIZON="0") self.assertEqual(len(self.server.state["posts"]), 1) def test_a_week_that_refills_in_eleven_hours_is_reset(self) -> None: # acct-21 (2026-09-22): Fable at 95%, week back at 09:00 — out of service 11h. self.server.state["usage"] = usage_doc(weekly=56, fable=95, week_resets_in=11 * 3600) self.server.state["responses"] = [(200, {"result": "reset", "resets_left": 0}, True)] self.run_limits() self.assertEqual(len(self.server.state["posts"]), 1) def test_a_grant_that_needs_the_limit_waits_for_it(self) -> None: self.server.state["usage"] = usage_doc(fable=97, use_requires_limit=True) self.run_limits() self.assertEqual(self.server.state["posts"], []) self.server.state["usage"] = usage_doc(fable=100, use_requires_limit=True, exhausted=["seven_day_overage_included"]) self.server.state["responses"] = [(200, {"result": "reset", "resets_left": 0}, True)] self.run_limits() self.assertEqual(len(self.server.state["posts"]), 1) def test_a_limit_the_grant_does_not_refill_never_spends_it(self) -> None: # Exhausted, but the grant refills only the five-hour window. self.server.state["usage"] = usage_doc(fable=100, clears=("five_hour",), exhausted=["seven_day_overage_included"]) self.run_limits() self.assertEqual(self.server.state["posts"], []) # ...and a grant whose benefit another exhausted limit would block is kept too. self.server.state["usage"] = usage_doc(fable=100, exhausted=["seven_day_overage_included"], blocking=["seven_day_opus"]) self.run_limits() self.assertEqual(self.server.state["posts"], []) def test_lost_response_is_retried_with_the_same_request_id(self) -> None: self.server.state["usage"] = usage_doc(fable=100, exhausted=["seven_day_overage_included"]) self.server.state["responses"] = [ (500, {"error": "response lost"}, False), (200, {"result": "reset", "resets_left": 0}, True)] first = self.run_limits() self.assertIn("retry is idempotent", self.log(), first.stdout) self.assertEqual(self.reset_state()["state"], "pending") self.run_limits() posts = self.server.state["posts"] self.assertEqual(len(posts), 2) self.assertEqual(posts[0]["body"]["request_id"], posts[1]["body"]["request_id"]) self.assertEqual(self.reset_state()["state"], "complete") def test_a_reset_that_landed_behind_a_lost_response_is_confirmed_not_reclaimed(self) -> None: self.server.state["usage"] = usage_doc(fable=100, exhausted=["seven_day_overage_included"]) (self.account / ".limited").write_text( f"{int(time.time()) + 3 * 86400}\nbucket=client:seven_day percent=100 " f"marked_at={time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime(time.time() - 3600))} " f"reason=client-rate-limit\n", encoding="utf-8") # The server applies the reset, but the answer never arrives. self.server.state["responses"] = [(502, {"error": "gateway"}, True)] self.run_limits() self.assertEqual(self.reset_state()["state"], "pending") self.assertTrue((self.account / ".limited").exists()) self.run_limits() self.assertEqual(len(self.server.state["posts"]), 1) # never a second claim self.assertEqual(self.reset_state()["outcome"], "confirmed") self.assertFalse((self.account / ".limited").exists()) self.assertIn("confirmed by the server's allowance", self.log()) def test_a_client_reported_weekly_limit_is_lifted_by_a_confirmed_reset(self) -> None: # A weekly client marker outlives every telemetry reading until its reset — # except a redemption, which is first-hand proof the window refilled. marked = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(time.time() - 7200)) (self.account / ".limited").write_text( f"{int(time.time()) + 3 * 86400}\nbucket=client:seven_day percent=100 " f"marked_at={marked} reason=client-rate-limit\n", encoding="utf-8") self.server.state["usage"] = usage_doc(fable=96) self.server.state["responses"] = [(200, {"result": "reset", "resets_left": 0}, True)] self.run_limits() self.assertFalse((self.account / ".limited").exists()) # The shim must not re-read the old rejection out of the transcript. watermark = int((self.account / ".client-limit-cleared").read_text().strip()) self.assertGreaterEqual(watermark, int(time.time()) - 60) def test_already_used_is_a_completed_reset(self) -> None: self.server.state["usage"] = usage_doc(fable=100, exhausted=["seven_day_overage_included"]) self.server.state["responses"] = [(200, {"result": "already_used", "resets_left": 0}, False)] self.run_limits() self.assertEqual(self.reset_state()["outcome"], "already_used") self.assertEqual(self.limits_doc()["auto_reset"]["status"], "redeemed") def test_answers_that_spend_nothing_leave_no_claim_behind(self) -> None: for answer in ("not_limited", "ineligible", "cooldown"): with self.subTest(answer=answer): self.server.state["usage"] = usage_doc(fable=100, exhausted=["seven_day_overage_included"]) self.server.state["responses"] = [(200, {"result": answer}, False)] self.run_limits() self.assertFalse((self.account / ".usage-reset.json").exists()) doc = self.limits_doc() self.assertNotIn("auto_reset", doc) self.assertNotEqual(doc["fetched_at"], 0) self.assertTrue((self.account / ".limited").exists()) # still at the limit def test_the_fleet_converges_on_one_request_id(self) -> None: # Two machines polling one account (each with its own sign-in) ask identically, # so the server can deduplicate the claim instead of spending twice. self.server.state["usage"] = usage_doc(fable=100, exhausted=["seven_day_overage_included"]) self.server.state["responses"] = [(500, {}, False), (500, {}, False)] self.run_limits() (self.account / ".usage-reset.json").unlink() # the peer has no local receipt self.run_limits() posts = self.server.state["posts"] self.assertEqual(posts[0]["body"]["request_id"], posts[1]["body"]["request_id"]) def test_disabled_automation_reports_without_spending(self) -> None: self.server.state["usage"] = usage_doc(fable=100, exhausted=["seven_day_overage_included"]) self.run_limits(CLAUDE_MULTIACC_AUTO_RESET="0") self.assertEqual(self.server.state["posts"], []) self.assertEqual(self.limits_doc()["reset_credits_available"], 1) self.assertTrue((self.account / ".limited").exists()) def test_without_the_organization_nothing_is_claimed(self) -> None: (self.account / ".claude.json").unlink() self.server.state["usage"] = usage_doc(fable=100, exhausted=["seven_day_overage_included"]) self.run_limits() self.assertEqual(self.server.state["posts"], []) self.assertIn("organization is unknown", self.log()) def test_an_unknown_cli_version_is_not_disguised(self) -> None: # No invented version: without the real one the writer names itself plainly, # the server answers "surface", and the allowance is unknown rather than zero. self.server.state["surface_gate"] = True self.server.state["usage"] = usage_doc(fable=100, exhausted=["seven_day_overage_included"]) self.run_limits(CLAUDE_MULTIACC_CLI_VERSION="not-a-version") self.assertEqual(self.server.state["usage_gets"][0]["ua"], "claude-multiacc/1.0") self.assertEqual(self.server.state["posts"], []) self.assertNotIn("reset_credits_available", self.limits_doc()) # ...and it is not silent: on the one machine that redeems, this is the whole # feature switched off (review 2026-09-22). self.assertIn("automatic redemption is OFF on this host", self.log()) class ClaudeResetReviewRegressionTest(ClaudePoolSandbox, unittest.TestCase): """Findings of the 2026-09-22 review, each pinned by the scenario that proved it.""" def weekly_park(self, offset: int = -3600) -> Path: marker = self.account / ".limited" marked = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(time.time() + offset)) marker.write_text(f"{int(time.time()) + 3 * 86400}\nbucket=client:seven_day " f"percent=100 marked_at={marked} reason=client-rate-limit\n", encoding="utf-8") return marker def test_a_replayed_claim_that_finds_nothing_to_do_keeps_its_receipt(self) -> None: # The first send lands but its answer is lost; the status read by the next pass # still predates it; the replay is told `not_limited` — BECAUSE it landed. marker = self.weekly_park() self.server.state["usage"] = usage_doc(fable=100, exhausted=["seven_day_overage_included"]) self.server.state["responses"] = [(502, {}, False), (200, {"result": "not_limited"}, False)] self.run_limits() self.run_limits() self.assertEqual(self.reset_state()["state"], "pending") # The server's own count finally shows the spent reset: confirmed, recorded, lifted. self.server.state["usage"] = usage_doc(fable=3, resets_left=0) self.run_limits() self.assertEqual(len(self.server.state["posts"]), 2) self.assertEqual(self.reset_state()["outcome"], "confirmed") self.assertIn("reset_redeemed_at", self.limits_doc()) self.assertFalse(marker.exists()) def test_a_rate_limited_claim_backs_off_and_keeps_the_measured_allowance(self) -> None: self.server.state["usage"] = usage_doc(fable=100, exhausted=["seven_day_overage_included"]) self.server.state["responses"] = [(429, {"error": "slow down"}, False, {"Retry-After": "3600"})] self.run_limits() state = self.reset_state() self.assertEqual(state["state"], "backoff") self.assertGreaterEqual(state["retry_after"] - int(time.time()), 3500) self.assertEqual(self.limits_doc()["reset_credits_available"], 1) self.run_limits() self.assertEqual(len(self.server.state["posts"]), 1) # not asked again self.assertIn("not asking again", self.log()) def test_a_refused_claim_is_not_retried_every_pass(self) -> None: self.server.state["usage"] = usage_doc(fable=100, exhausted=["seven_day_overage_included"]) self.server.state["responses"] = [(403, {"error": "forbidden"}, False)] self.run_limits() self.run_limits() self.assertEqual(len(self.server.state["posts"]), 1) self.assertGreaterEqual(self.reset_state()["backoff"], 3600) def test_a_reset_claimed_elsewhere_is_observed_and_lifts_older_parks(self) -> None: # A fleet peer (or a human's /limit-reset) spent the grant; this machine never # claimed, but it saw the grant's own count drop. self.server.state["usage"] = usage_doc(fable=96) self.run_limits(CLAUDE_MULTIACC_AUTO_RESET="0") self.assertEqual(self.limits_doc()["reset_grants_seen"], {GRANT: 1}) marker = self.weekly_park() self.server.state["usage"] = usage_doc(fable=2, resets_left=0) self.run_limits() doc = self.limits_doc() self.assertEqual(self.server.state["posts"], []) self.assertIn("reset_redeemed_at", doc) self.assertNotEqual(doc["fetched_at"], 0) # a post-reset reading stands self.assertFalse(marker.exists()) self.assertIn("limit reset observed", self.log()) def test_an_observed_reset_lifts_a_5h_park_only_on_a_measured_session(self) -> None: # The reset is only SEEN at this pass, later than it happened: a five-hour park # taken in between may be real post-reset exhaustion, and a silent session bucket # cannot disprove it. A weekly park in the same gap cannot be (review, 2026-09-22). self.server.state["usage"] = usage_doc(fable=96) self.run_limits(CLAUDE_MULTIACC_AUTO_RESET="0") marked = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(time.time() - 30)) park = self.account / ".limited" park.write_text(f"{int(time.time()) + 3600}\nbucket=client:five_hour percent=100 " f"marked_at={marked} reason=client-rate-limit\n", encoding="utf-8") self.server.state["usage"] = usage_doc(fable=2, resets_left=0) # session silent self.run_limits() doc = self.limits_doc() self.assertEqual(doc["reset_cleared"], "seven_day,seven_day_overage_included") self.assertTrue(park.exists()) # ...while a weekly park from the same gap is lifted by the same record. park.write_text(f"{int(time.time()) + 3 * 86400}\nbucket=client:seven_day percent=100 " f"marked_at={marked} reason=client-rate-limit\n", encoding="utf-8") self.run_limits() self.assertFalse(park.exists()) def test_an_expired_grant_is_never_read_as_a_reset(self) -> None: self.server.state["usage"] = usage_doc(fable=96) self.run_limits(CLAUDE_MULTIACC_AUTO_RESET="0") marker = self.weekly_park() expired = usage_doc(fable=96) expired["cedar_ember"]["grants"][0]["ends_at"] = iso(-60) self.server.state["usage"] = expired self.run_limits() self.assertNotIn("reset_redeemed_at", self.limits_doc()) self.assertTrue(marker.exists()) def test_our_own_claim_is_not_observed_again_next_pass(self) -> None: self.server.state["usage"] = usage_doc(fable=100, exhausted=["seven_day_overage_included"]) self.server.state["responses"] = [(200, {"result": "reset", "resets_left": 0}, True)] self.run_limits() stamp = self.limits_doc()["reset_redeemed_at"] self.run_limits() self.assertEqual(self.limits_doc()["reset_redeemed_at"], stamp) self.assertNotIn("limit reset observed", self.log()) class ClaudeCliVersionTest(ClaudePoolSandbox, unittest.TestCase): """The User-Agent names the INSTALLED Claude Code version, learned from the real binary.""" def run_with_fake(self, make) -> str: fakebin = Path(self.temp.name) / "fakebin" fakebin.mkdir() make(fakebin) home = Path(self.temp.name) / "home" home.mkdir() self.run_limits(CLAUDE_MULTIACC_CLI_VERSION="", HOME=str(home), PATH=f"{REPO / 'bin'}:{fakebin}:{os.environ.get('PATH', '')}") return self.server.state["usage_gets"][-1]["ua"] def test_the_native_installers_version_symlink_answers(self) -> None: def make(fakebin: Path) -> None: versions = Path(self.temp.name) / "versions" versions.mkdir() binary = versions / "2.1.281" binary.write_text("#!/usr/bin/env bash\nexit 3\n", encoding="utf-8") binary.chmod(0o755) (fakebin / "claude").symlink_to(binary) self.assertEqual(self.run_with_fake(make), "claude-cli/2.1.281 (external, cli, client-app/claude-multiacc)") def test_an_npm_install_is_asked_for_its_version(self) -> None: def make(fakebin: Path) -> None: script = fakebin / "claude" script.write_text('#!/usr/bin/env bash\n[ "$1" = --version ] && echo "2.1.282 (Claude Code)"\n', encoding="utf-8") script.chmod(0o755) self.assertEqual(self.run_with_fake(make), "claude-cli/2.1.282 (external, cli, client-app/claude-multiacc)") def test_an_unreadable_version_is_never_invented(self) -> None: def make(fakebin: Path) -> None: script = fakebin / "claude" script.write_text('#!/usr/bin/env bash\necho "CFG=none"\n', encoding="utf-8") script.chmod(0o755) self.assertEqual(self.run_with_fake(make), "claude-multiacc/1.0") class ClaudeResetFleetRecordTest(ClaudePoolSandbox, unittest.TestCase): """What the redeeming machine leaves behind for every OTHER reader of the account.""" def redeem(self) -> None: self.server.state["usage"] = usage_doc(session=100, weekly=65, fable=96, exhausted=["five_hour"]) self.server.state["responses"] = [(200, { "result": "reset", "resets_left": 0, "cleared": ["five_hour", "seven_day", "seven_day_overage_included"]}, True)] self.run_limits() def test_the_reset_is_recorded_and_the_pre_reset_numbers_withdrawn(self) -> None: before = int(time.time()) self.redeem() doc = self.limits_doc() self.assertGreaterEqual(doc["reset_redeemed_at"], before) self.assertEqual(doc["reset_cleared"], "five_hour,seven_day,seven_day_overage_included") # Readers that ignore freshness (the shim's fallback-model check, the panel's # window reads) must not find the 100% session or the 96% Fable week. for key in ("max_percent", "weekly_percent", "session_percent", "weekly_resets_epoch", "session_inferred"): self.assertNotIn(key, doc) self.assertEqual(doc["buckets"], []) def test_the_record_rides_along_on_later_passes(self) -> None: self.redeem() stamp = self.limits_doc()["reset_redeemed_at"] self.run_limits() # a normal post-reset pass doc = self.limits_doc() self.assertEqual(doc["reset_redeemed_at"], stamp) self.assertNotEqual(doc["fetched_at"], 0) self.assertIn("weekly_percent", doc) # server truth is back # ...until nothing it could supersede can still exist (nine days on, for the # document AND the machine's own receipt). doc["reset_redeemed_at"] = stamp - 9 * 86400 (self.account / "limits.json").write_text(json.dumps(doc), encoding="utf-8") receipt = self.reset_state() receipt["redeemed_at"] = stamp - 9 * 86400 (self.account / ".usage-reset.json").write_text(json.dumps(receipt), encoding="utf-8") self.run_limits() self.assertNotIn("reset_redeemed_at", self.limits_doc()) def test_the_watermark_is_written_even_without_a_marker(self) -> None: self.assertFalse((self.account / ".limited").exists()) (self.account / ".client-limit-cleared").write_text("5\n", encoding="utf-8") self.server.state["usage"] = usage_doc(fable=96) # 96% is below no marker yet self.server.state["responses"] = [(200, {"result": "reset", "resets_left": 0}, True)] self.run_limits() mark = int((self.account / ".client-limit-cleared").read_text().split()[0]) self.assertEqual(mark, self.limits_doc()["reset_redeemed_at"]) def test_a_panel_park_older_than_the_reset_is_cleared_by_the_next_pass(self) -> None: self.redeem() stamp = self.limits_doc()["reset_redeemed_at"] older = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(stamp - 3600)) # What the panel's heartbeat writes back until app-robot honours the record. (self.account / ".limited").write_text( f"{stamp + 3 * 86400}\nbucket=weekly_all percent=95 marked_at={older} " f"reason=client-rate-limit source=stream\n", encoding="utf-8") self.run_limits() self.assertFalse((self.account / ".limited").exists()) self.assertIn("written before a confirmed limit reset", self.log()) class ClaudeDeadSignInReportTest(ClaudePoolSandbox, unittest.TestCase): """A sign-in that EXISTS but can no longer work must say so (2026-09-22: acct-02 and acct-07 read `oauth: true` with a refresh token expired days before, so the panel never offered the sign-in that restores their telemetry and limit resets).""" def listed(self) -> dict: result = subprocess.run([REPO / "bin/claude-accounts", "list", "--json"], capture_output=True, text=True, env=self.env(), timeout=60, check=False) self.assertEqual(result.returncode, 0, result.stderr) return json.loads(result.stdout)["accounts"][0]["credentials"] def test_a_live_sign_in_is_alive(self) -> None: creds = self.listed() self.assertTrue(creds["oauth"]) self.assertTrue(creds["oauth_alive"]) self.assertNotIn("oauth_problem", creds) def test_an_expired_refresh_token_is_a_dead_sign_in(self) -> None: (self.account / "server.token").write_text("sk-ant-oat01-portable\n", encoding="utf-8") dead = {"claudeAiOauth": {"accessToken": "old", "refreshToken": "r", "expiresAt": int(time.time() - 7 * 86400) * 1000, "refreshTokenExpiresAt": int(time.time() - 3 * 86400) * 1000}} (self.account / ".credentials.json").write_text(json.dumps(dead), encoding="utf-8") creds = self.listed() self.assertTrue(creds["oauth"]) # it exists... self.assertFalse(creds["oauth_alive"]) # ...but cannot work self.assertIn("refresh token expired", creds["oauth_problem"]) class ClaudeResetReportingTest(ClaudePoolSandbox, unittest.TestCase): def report(self, **overrides) -> dict: result = self.run_limits("--json", **overrides) return json.loads(result.stdout)["accounts"][0] def test_idle_account_reports_its_allowance(self) -> None: account = self.report() self.assertEqual(account["usage"]["reset_credits_available"], 1) self.assertTrue(account["usage"]["reset_credits_fetched_at"].endswith("Z")) self.assertEqual(self.server.state["posts"], []) def test_an_account_without_grants_reports_a_measured_zero(self) -> None: self.server.state["usage"] = usage_doc(eligible=False, reason="no_grant") self.assertEqual(self.report()["usage"]["reset_credits_available"], 0) def test_a_payload_without_the_block_reports_nothing(self) -> None: del self.server.state["usage"]["cedar_ember"] usage = self.report()["usage"] self.assertNotIn("reset_credits_available", usage) self.assertEqual(usage["weekly_percent"], 50) def test_an_unconfirmed_claim_reports_no_allowance(self) -> None: self.server.state["usage"] = usage_doc(fable=100, exhausted=["seven_day_overage_included"]) self.server.state["responses"] = [(200, {"result": "unavailable"}, False)] usage = self.report()["usage"] self.assertNotIn("reset_credits_available", usage) self.assertEqual(self.reset_state()["state"], "pending") if __name__ == "__main__": unittest.main()