"""Automatic redemption of Claude Code limit resets (the ``cedar_ember`` program).

Claude Code 2.1.280 shipped ``/limit-reset``: a subscription account can hold
"grants", each worth a number of resets, and a reset refills the limits the grant
names (``clears``: five_hour, seven_day, seven_day_overage_included, ...) while the
weekly reset day stays where it was. The first grant arrived with the Opus 5.5
launch on 2026-09-22 — one reset per Pro/Max account, usable until 2026-10-22.

This is the Claude half of lib/codex_reset.py, and it follows the operator's rule:
an account the pool has PARKED on a window the grant refills gets its reset, so it goes
back to work instead of sitting out the wall. Parked means exactly what the shim means
by it — a bucket the grant clears at or above the pool's own exclusion threshold (90%
by default, the manifest's `threshold`), a limit the server reports exhausted, or an
active `.limited` park (a 429 the client received, or the app-robot panel's park) on
such a window. The one hold: when every such park lifts on its own within
``CLAUDE_MULTIACC_RESET_MIN_HORIZON`` seconds (default 1h), the reset is kept — a
reset is one irreversible shot (support.claude.com "What is a limit reset?").

What differs from codex beyond that is only the transport:

* The allowance is NOT a second endpoint. The OAuth usage endpoint the limits pass
  already reads carries the program's status block when asked for it
  (``?cedar_ember=1&skip_spend=1`` — the exact read Claude Code's own ``/limit-reset``
  makes). That endpoint's budget is roughly one call per account per hour, so a
  second GET per pass would buy 429s that blind telemetry; piggybacking costs none.
* The server decides eligibility from WHO is asking. A request that does not identify
  itself as the Claude Code CLI is answered ``eligible: false, ineligible_reason:
  "surface"``. The writer therefore sends Claude Code's own User-Agent shape with the
  installed CLI version and names this wrapper through the client's documented
  ``client-app/`` slot: ``claude-cli/<version> (external, cli, client-app/claude-multiacc)``.
* A claim is ``POST /api/organizations/<org>/reset_rate_limits`` with
  ``{"program": "cedar_ember", "grant_id": ..., "request_id": ...}``; the organization
  is the account's own (``<acct>/.claude.json`` ``oauthAccount.organizationUuid``).

Idempotency mirrors codex: a pending request id is persisted BEFORE the POST, and the
id is derived deterministically from the account's organization, the grant, the grant's
remaining count and the weekly window, so every machine in the fleet that polls the same
account asks with the same id and a lost response is retried without spending twice. A
pending claim whose response was lost is settled by provider evidence — the grant's
``resets_left`` dropping below what it was when the claim was sent — never by guessing.
"""

from __future__ import annotations

import datetime
import json
import os
import re
import time
import urllib.error
import urllib.parse
import urllib.request
import uuid

PROGRAM = "cedar_ember"
RESET_AT_USED_PERCENT = 95
REDEEM_COOLDOWN_SECONDS = 900
# Limit types that refill on their own within hours; never a reason to spend a reset.
SESSION_LIMITS = frozenset({"five_hour"})
DEFAULT_MIN_HORIZON_SECONDS = 3600
# A claim whose response was lost is retried with the same request id while it is
# younger than this, and dropped afterwards unless the status block proved it landed.
PENDING_TTL_SECONDS = 3600
STATE_FILE = ".usage-reset.json"
# How long limits.json carries the record of a confirmed reset. Every marker it can
# supersede was written before the reset and names a window at most a week long, so
# eight days outlives all of them on every machine the document is pushed to.
RESET_RECORD_TTL_SECONDS = 8 * 86400
CLIENT_APP = "claude-multiacc"
FALLBACK_USER_AGENT = "claude-multiacc/1.0"

# The client's own validation (Claude Code 2.1.280): it refuses to claim with an id
# outside these shapes, and so do we — a malformed id is a payload we do not understand.
_GRANT_ID = re.compile(r"^[a-z0-9_-]{1,40}$")
_REQUEST_ID = re.compile(r"^[A-Za-z0-9_-]{1,64}$")
_LIMIT_TYPE = re.compile(r"^[a-z0-9_]{1,64}$")
_ORG_UUID = re.compile(r"^[0-9A-Fa-f-]{8,64}$")
_VERSION = re.compile(r"^\d+\.\d+\.\d+$")
_RESULTS = {"reset", "already_used", "not_limited", "cooldown", "ineligible", "unavailable"}
# Ineligibility that describes the REQUEST (which client asked, from where) rather than
# the account. The allowance behind such an answer is unknown, not zero.
_REQUEST_SIDE_REASONS = {"surface", "cli_version", "mobile", "unavailable", "unknown"}


def auto_reset_enabled() -> bool:
    """Return whether automatic redemption is enabled (on by default)."""
    return os.environ.get("CLAUDE_MULTIACC_AUTO_RESET", "1").strip().lower() \
        not in {"0", "false", "no", "off"}


def min_horizon() -> int:
    """Seconds before the natural weekly reset inside which a reset is not spent."""
    try:
        value = int(os.environ.get("CLAUDE_MULTIACC_RESET_MIN_HORIZON",
                                   DEFAULT_MIN_HORIZON_SECONDS))
    except ValueError:
        return DEFAULT_MIN_HORIZON_SECONDS
    return max(0, min(7 * 86400, value))


def cli_version(real_path: str | None) -> str:
    """The installed Claude Code version ("2.1.280"), '' when it cannot be learned.

    Never invented: the server gates on it, and a made-up version is a lie about which
    client is asking. CLAUDE_MULTIACC_CLI_VERSION overrides (tests, pinned hosts). The
    native installer's ``versions/<x.y.z>`` symlink target answers without a fork;
    anything else (an npm cli.js) is asked ``--version`` under a hard timeout, because
    this runs while the limits lock is held.
    """
    override = os.environ.get("CLAUDE_MULTIACC_CLI_VERSION", "").strip()
    if override:
        return override if _VERSION.match(override) else ""
    if not real_path:
        return ""
    name = os.path.basename(os.path.realpath(real_path))
    if _VERSION.match(name):
        return name
    try:
        import subprocess
        env = dict(os.environ, CLAUDE_SHIM_ACTIVE="1")
        done = subprocess.run([real_path, "--version"], capture_output=True, text=True,
                              timeout=10, stdin=subprocess.DEVNULL, env=env, check=False)
        first = (done.stdout or "").strip().split()
        return first[0] if first and _VERSION.match(first[0]) else ""
    except Exception:
        return ""


def user_agent(cli_version: str | None) -> str:
    """Claude Code's User-Agent shape for the installed CLI, naming this wrapper."""
    version = (cli_version or "").strip()
    if _VERSION.match(version):
        return f"claude-cli/{version} (external, cli, client-app/{CLIENT_APP})"
    return FALLBACK_USER_AGENT


def claim_url(usage_url: str, org_uuid: str) -> str | None:
    """The claim endpoint on the usage endpoint's own origin (None for file:// fixtures)."""
    explicit = os.environ.get("CLAUDE_MULTIACC_RESET_URL", "").strip()
    if explicit:
        return explicit.replace("{org}", org_uuid)
    parsed = urllib.parse.urlsplit(usage_url)
    if parsed.scheme not in {"http", "https"} or not parsed.netloc:
        return None
    path = f"/api/organizations/{org_uuid}/reset_rate_limits"
    return urllib.parse.urlunsplit((parsed.scheme, parsed.netloc, path, "", ""))


def _oauth_account(account_dir: str) -> dict:
    try:
        with open(os.path.join(account_dir, ".claude.json"), encoding="utf-8") as handle:
            doc = json.load(handle)
        value = doc.get("oauthAccount") if isinstance(doc, dict) else None
        return value if isinstance(value, dict) else {}
    except Exception:
        return {}


def organization_uuid(account_dir: str) -> str | None:
    """The account's own organization, as Claude Code recorded it for this config dir."""
    org = str(_oauth_account(account_dir).get("organizationUuid") or "")
    return org if _ORG_UUID.match(org) else None


def account_uuid(account_dir: str) -> str | None:
    """The signed-in user within that organization (several Team seats share one org)."""
    value = str(_oauth_account(account_dir).get("accountUuid") or "")
    return value if _ORG_UUID.match(value) else None


def _epoch(value) -> float | None:
    if not isinstance(value, str) or not value:
        return None
    try:
        return datetime.datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp()
    except ValueError:
        return None


def _types(value) -> list[str]:
    if not isinstance(value, list):
        return []
    return sorted({item for item in value if isinstance(item, str) and _LIMIT_TYPE.match(item)})


def _count(value) -> int | None:
    return value if type(value) is int and 0 <= value <= 1_000_000 else None


def _grant(raw) -> dict | None:
    """One grant, or None when it is not a shape the client itself would accept."""
    if not isinstance(raw, dict):
        return None
    grant_id = raw.get("id")
    resets_left = _count(raw.get("resets_left"))
    if not isinstance(grant_id, str) or not _GRANT_ID.match(grant_id) or resets_left is None:
        return None
    percent = {}
    for key, value in (raw.get("percent_used") or {}).items() if isinstance(
            raw.get("percent_used"), dict) else []:
        if isinstance(key, str) and _LIMIT_TYPE.match(key) and type(value) is int \
                and 0 <= value <= 100:
            percent[key] = value
    return {
        "id": grant_id,
        "resets_left": resets_left,
        "resets_total": _count(raw.get("resets_total")) or 0,
        "starts_at": raw.get("starts_at") if isinstance(raw.get("starts_at"), str) else None,
        "ends_at": raw.get("ends_at") if isinstance(raw.get("ends_at"), str) else None,
        "clears": _types(raw.get("clears")),
        "blocking": _types(raw.get("blocking")),
        "paused": raw.get("paused") is True,
        "usable_now": raw.get("usable_now") is True,
        # The client defaults this to TRUE when absent: a grant that says nothing may
        # only be used at a limit, which is the conservative reading.
        "use_requires_limit": raw.get("use_requires_limit") is not False,
        "percent_used": percent,
    }


def parse_status(usage) -> dict | None:
    """The normalized ``cedar_ember`` block of a usage response, or None when absent/unreadable."""
    block = usage.get(PROGRAM) if isinstance(usage, dict) else None
    if not isinstance(block, dict) or not isinstance(block.get("eligible"), bool):
        return None
    grants = [grant for grant in map(_grant, block.get("grants") or []
                                     if isinstance(block.get("grants"), list) else [])
              if grant is not None]
    next_id = block.get("next_grant_id")
    if not any(grant["id"] == next_id for grant in grants):
        next_id = None
    reason = block.get("ineligible_reason")
    return {
        "eligible": block["eligible"],
        "ineligible_reason": reason if isinstance(reason, str) else None,
        "at_limit": block.get("at_limit") is True,
        "exhausted": _types(block.get("exhausted")),
        "grants": grants,
        "next_grant_id": next_id,
        "weekly_resets_at": block.get("weekly_resets_at")
        if isinstance(block.get("weekly_resets_at"), str) else None,
        "cooldown_until": block.get("cooldown_until")
        if isinstance(block.get("cooldown_until"), str) else None,
    }


def _live(grant: dict, now: float) -> bool:
    ends = _epoch(grant["ends_at"])
    return ends is None or ends > now


def credits_view(status: dict | None, now: int) -> dict:
    """Measured reset allowance for limits.json, or {} when the read proves nothing.

    Same contract as the codex report: zero is a measurement, a missing field is an
    unknown. A block that is ineligible because of who ASKED (surface, CLI version) says
    nothing about the account, so it reports nothing rather than a false zero.
    """
    if status is None:
        return {}
    if not status["eligible"] and (status["ineligible_reason"] or "unknown") in _REQUEST_SIDE_REASONS:
        return {}
    count = sum(grant["resets_left"] for grant in status["grants"] if _live(grant, now))
    if count > 1_000_000:
        return {}
    return {"reset_credits_available": count, "reset_credits_fetched_at": int(now)}


def _usable_grant(status: dict, now: float) -> dict | None:
    """The grant the client would offer now (its ``next_grant_id``), if it can be used."""
    grant = next((g for g in status["grants"] if g["id"] == status["next_grant_id"]), None)
    if grant is None or not grant["usable_now"] or grant["paused"] or grant["resets_left"] <= 0:
        return None
    starts = _epoch(grant["starts_at"])
    if not _live(grant, now) or (starts is not None and starts > now):
        return None
    return grant


def _percent(grant: dict, usage: dict, limit_type: str) -> int | None:
    value = grant["percent_used"].get(limit_type)
    if value is not None:
        return value
    # The payload also names most limit types at the top level; use that reading when
    # the grant did not carry one for this type.
    bucket = usage.get(limit_type) if isinstance(usage, dict) else None
    if isinstance(bucket, dict):
        try:
            return max(0, min(100, int(round(float(bucket.get("utilization"))))))
        except (TypeError, ValueError):
            return None
    return None


def _lift_epoch(limit_type: str, status: dict, usage: dict) -> float | None:
    """When a parked window refills on its own: the session window's reset, or the week's."""
    if limit_type in SESSION_LIMITS:
        for limit in (usage.get("limits") or []) if isinstance(usage, dict) else []:
            if isinstance(limit, dict) and str(limit.get("kind") or "").startswith("session"):
                found = _epoch(limit.get("resets_at"))
                if found is not None:
                    return found
        bucket = usage.get("five_hour") if isinstance(usage, dict) else None
        return _epoch(bucket.get("resets_at")) if isinstance(bucket, dict) else None
    return _epoch(status.get("weekly_resets_at"))


def _active_park(account_dir: str, now: float) -> tuple[str | None, float] | None:
    """(limit type, lift epoch) of the account's `.limited` park, when it is a USAGE park
    still in force and not already lifted by a reset on record; else None."""
    try:
        with open(os.path.join(account_dir, ".limited"), errors="replace") as handle:
            text = handle.read(4096)
        until = int(text.splitlines()[0])
    except (OSError, ValueError, IndexError):
        return None
    if until <= now or not ("reason=limits" in text or "reason=client-rate-limit" in text):
        return None
    record = latest(record_from(_load(os.path.join(account_dir, "limits.json")), now),
                    local_record(account_dir, now))
    if marker_superseded(text, record):
        return None
    bucket = next((part[len("bucket="):] for part in text.split()
                   if part.startswith("bucket=")), "")
    return bucket_limit_type(bucket), float(until)


def parked_on(status: dict, grant: dict, usage: dict, account_dir: str, threshold: int,
              now: float) -> list[tuple[str, float | None]]:
    """Every (limit type, when it lifts on its own) the pool parks this account on,
    among the windows this grant refills."""
    clears = set(grant["clears"])
    parks = [(t, _lift_epoch(t, status, usage)) for t in sorted(clears & set(status["exhausted"]))]
    parks += [(t, _lift_epoch(t, status, usage)) for t in sorted(clears)
              if (_percent(grant, usage, t) or 0) >= threshold]
    park = _active_park(account_dir, now)
    if park and (park[0] == "*" or park[0] in clears):
        parks.append(park)
    return parks


# ---- the reset record the whole fleet honours -------------------------------------
# A confirmed reset has to lift every park that predates it, on EVERY machine: peers
# only ever receive `.limited` pushes (never deletions), and the app-robot panel keeps
# its own verdicts. So the writer records `reset_redeemed_at` (epoch) and
# `reset_cleared` (comma-separated limit types) in limits.json, which IS pushed, and a
# marker is superseded when it was written at or before that moment for a window the
# reset refilled. bin/claude's reset_supersedes_marker is the same rule in bash.

def bucket_limit_type(name: str) -> str | None:
    """The provider limit type a bucket or marker name describes; '*' for a park that
    names no window (it came from a refused session, so any account-level refill lifts
    it); None for a model window this code does not know."""
    token = (name or "").strip()
    lower = token.lower()
    if lower.startswith("client:"):
        rest = lower[len("client:"):]
        return rest if _LIMIT_TYPE.match(rest) else None
    if lower.startswith("weekly_scoped:"):
        model = lower[len("weekly_scoped:"):]
        if "fable" in model:
            return "seven_day_overage_included"
        for family in ("opus", "sonnet"):
            if family in model:
                return f"seven_day_{family}"
        return None
    if lower in {"session", "five_hour", "5h"} or lower.startswith("session"):
        return "five_hour"
    if lower in {"weekly_all", "seven_day", "weekly", "7d"}:
        return "seven_day"
    if lower in {"", "panel:observed", "limit_reached", "unknown"}:
        return "*"
    return lower if _LIMIT_TYPE.match(lower) else None


def record_from(doc: dict, now: float) -> tuple[int, list[str]] | None:
    """(redeemed_at, cleared types) from a limits document, or None when absent/expired."""
    if not isinstance(doc, dict):
        return None
    at = doc.get("reset_redeemed_at")
    raw = doc.get("reset_cleared")
    if type(at) is not int or at <= 0 or now - at > RESET_RECORD_TTL_SECONDS:
        return None
    cleared = [t for t in str(raw or "").split(",") if _LIMIT_TYPE.match(t)]
    return (at, cleared) if cleared else None


def grants_seen(status: dict | None, now: float, redeemed: dict | None = None) -> dict:
    """{grant id: resets_left} for the live grants this pass read — the baseline the next
    pass compares against. A claim this pass made is recorded at its post-claim count, so
    the pass after it never mistakes its own reset for someone else's."""
    if status is None:
        return {}
    seen = {g["id"]: g["resets_left"] for g in status["grants"] if _live(g, now)}
    if redeemed and redeemed.get("grant_id") in seen:
        left = redeemed.get("resets_left")
        seen[redeemed["grant_id"]] = left if type(left) is int \
            else max(0, seen[redeemed["grant_id"]] - 1)
    return seen


def observed_reset(prev: dict, status: dict | None, now: float) -> tuple[int, list[str]] | None:
    """A reset this machine did NOT claim — a fleet peer's, or a human's /limit-reset —
    seen as one grant's OWN remaining count dropping since the last pass. Never inferred
    from the total: that also drops when an unused grant simply expires, and reading an
    expiry as a refill would lift truthful weekly parks (the 2026-09-04 failure class)."""
    seen = prev.get("reset_grants_seen") if isinstance(prev, dict) else None
    if not isinstance(seen, dict) or status is None:
        return None
    hit = [g for g in status["grants"] if _live(g, now)
           and type(seen.get(g["id"])) is int and g["resets_left"] < seen[g["id"]]]
    cleared = sorted({t for g in hit for t in g["clears"]})
    return (int(now), cleared) if cleared else None


def local_record(account_dir: str, now: float) -> tuple[int, list[str]] | None:
    """The redeeming machine's own receipt, for a pass that died between the claim and
    the limits.json write."""
    state = _load(os.path.join(account_dir, STATE_FILE))
    at = state.get("redeemed_at")
    cleared = _types(state.get("cleared"))
    if state.get("state") != "complete" or type(at) is not int or not cleared \
            or now - at > RESET_RECORD_TTL_SECONDS:
        return None
    return at, cleared


def latest(*records):
    """The most recent of several (epoch, cleared) records, ignoring absent ones."""
    present = [r for r in records if r]
    return max(present, key=lambda r: r[0]) if present else None


def _marked_epoch(text: str) -> float | None:
    for part in text.split():
        if part.startswith("marked_at="):
            return _epoch(part[len("marked_at="):])
    return None


def marker_superseded(text: str, record: tuple[int, list[str]] | None) -> bool:
    """Was this `.limited` marker written before a confirmed reset that refilled its window?"""
    if not record or not isinstance(text, str):
        return False
    at, cleared = record
    marked = _marked_epoch(text)
    if marked is None or marked > at:
        return False
    bucket = next((part[len("bucket="):] for part in text.split()
                   if part.startswith("bucket=")), "")
    limit_type = bucket_limit_type(bucket)
    if limit_type == "*":
        return bool(cleared)
    return limit_type in cleared


def _load(path: str) -> dict:
    try:
        with open(path, encoding="utf-8") as handle:
            value = json.load(handle)
        return value if isinstance(value, dict) else {}
    except Exception:
        return {}


def _write(path: str, value: dict) -> None:
    temp = f"{path}.tmp.{os.getpid()}"
    descriptor = os.open(temp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
        json.dump(value, handle, indent=2, sort_keys=True)
        handle.write("\n")
    os.replace(temp, path)


def _remove(path: str) -> None:
    try:
        os.remove(path)
    except OSError:
        pass


def request_id(org_uuid: str, subject: str, grant: dict, status: dict) -> str:
    """One fleet-stable id per (account, grant, remaining count, week).

    Every machine polling this account sees the same organization, user, grant, count
    and weekly window, so they converge on ONE claim; after a reset lands the count
    drops, so the next episode of a multi-reset grant gets a new id instead of a
    replay. The user is part of the seed because Team seats share an organization.
    """
    seed = json.dumps({"program": PROGRAM, "org": org_uuid, "subject": subject,
                       "grant": grant["id"],
                       "resets_left": grant["resets_left"],
                       "week": status.get("weekly_resets_at") or ""}, sort_keys=True)
    return str(uuid.uuid5(uuid.NAMESPACE_URL, seed))


def _post(url: str, headers: dict, payload: dict) -> tuple[int, dict, int]:
    """(HTTP status, JSON object body or {}, Retry-After seconds or 0)."""
    request = urllib.request.Request(url, data=json.dumps(payload).encode(),
                                     headers=headers, method="POST")
    try:
        response = urllib.request.urlopen(request, timeout=30)
        status, raw = response.status, response.read()
    except urllib.error.HTTPError as error:
        try:
            wait = max(0, int(error.headers.get("Retry-After") or 0))
        except (TypeError, ValueError):
            wait = 0
        return error.code, {}, wait
    try:
        value = json.loads(raw.decode())
    except Exception:
        value = None
    return status, value if isinstance(value, dict) else {}, 0


# A claim the server REFUSED outright (401/403/429/other 4xx) was provably not processed;
# retrying it on every pass is how the usage endpoint once talked itself into a days-long
# 429 (CLAUDE.md, 2026-08-11). It backs off like every other non-2xx here does, honouring
# Retry-After, doubling from 15 minutes to a day. A 5xx or a lost connection may have
# landed, so those stay pending and are retried with the same request id instead.
BACKOFF_FLOOR_SECONDS = 900
BACKOFF_CEILING_SECONDS = 86400


def _backoff(state: dict, code: int, retry_after: int, current: int) -> dict:
    previous = state.get("backoff") if type(state.get("backoff")) is int else 0
    floor = 3600 if code in (401, 403) else BACKOFF_FLOOR_SECONDS
    wait = min(BACKOFF_CEILING_SECONDS, max(floor, 2 * previous))
    wait = max(wait, min(BACKOFF_CEILING_SECONDS, retry_after))
    return {"backoff": wait, "retry_after": current + wait, "last_error": f"HTTP {code}"}


def _usage_peak(usage) -> int:
    peak = 0
    for limit in (usage.get("limits") or []) if isinstance(usage, dict) else []:
        if isinstance(limit, dict) and type(limit.get("percent")) in (int, float):
            peak = max(peak, int(limit["percent"]))
    return peak


def _complete(state_path: str, pending: dict, outcome: str, now: int, cleared=None,
              resets_left=None) -> dict:
    # The wall clock AFTER the claim, not the pass's start: a marker the client wrote
    # while this pass was in flight describes the pre-reset state and must be covered.
    at = max(int(now), int(time.time()))
    cleared = sorted(set(cleared or []) or set(pending.get("clears") or []))
    complete = {"schema": 1, "program": PROGRAM, "state": "complete", "outcome": outcome,
                "grant_id": pending.get("grant_id"), "request_id": pending.get("request_id"),
                "redeemed_at": at, "suppress_until": at + REDEEM_COOLDOWN_SECONDS,
                "cleared": cleared}
    _write(state_path, complete)
    result = {"status": "redeemed", "program": PROGRAM, "outcome": outcome,
              "grant_id": pending.get("grant_id"), "cleared": cleared,
              "redeemed_at": at}
    if resets_left is not None:
        result["resets_left"] = resets_left
    return result


def try_auto_redeem(account_dir: str, account_id: str, usage: dict, status: dict | None,
                    usage_url: str, headers: dict, say, now: int | None = None,
                    threshold: int = RESET_AT_USED_PERCENT) -> dict:
    """Redeem one reset when policy says so and return a non-secret outcome document."""
    current = int(now if now is not None else time.time())
    if not auto_reset_enabled():
        return {"status": "not_eligible", "reason": "disabled"}
    if status is None:
        return {"status": "no_status"}
    state_path = os.path.join(account_dir, STATE_FILE)
    state = _load(state_path)
    backing_off = type(state.get("retry_after")) is int and state["retry_after"] > current
    if state.get("state") == "pending":
        grant = next((g for g in status["grants"] if g["id"] == state.get("grant_id")), None)
        before = state.get("resets_left_before")
        if grant is not None and type(before) is int and grant["resets_left"] < before:
            # The response was lost, but the server's own count moved: it landed.
            say(f"{account_id}: limit reset confirmed by the server's allowance "
                f"(claim {state.get('request_id')})")
            return _complete(state_path, state, "confirmed", current,
                             resets_left=grant["resets_left"])
        started = state.get("started_at")
        if not backing_off and (type(started) is not int
                                or current - started > PENDING_TTL_SECONDS):
            _remove(state_path)
            state = {}
    if backing_off:
        return {"status": "pending" if state.get("state") == "pending" else "backoff",
                "reason": state.get("last_error")}
    if state.get("state") == "backoff":
        state = {}
    if state.get("state") == "complete" and int(state.get("suppress_until") or 0) > current:
        return {"status": "cooldown", "outcome": state.get("outcome")}
    if not status["eligible"]:
        reason = status["ineligible_reason"] or "unknown"
        if reason in _REQUEST_SIDE_REASONS and _usage_peak(usage) >= threshold:
            # Not a fact about the account: the server would not answer THIS client. On the
            # one machine that can redeem, that silently turns redemption off fleet-wide.
            say(f"{account_id}: limit-reset status refused for this client ({reason}; "
                f"User-Agent {headers.get('User-Agent')!r}) — automatic redemption is OFF "
                f"on this host")
        return {"status": "ineligible", "reason": reason}
    cooldown = _epoch(status["cooldown_until"])
    if cooldown is not None and cooldown > current:
        return {"status": "cooldown", "reason": "server"}
    grant = _usable_grant(status, current)
    if grant is None:
        return {"status": "no_credit"}
    parks = parked_on(status, grant, usage, account_dir, threshold, current)
    if not parks:
        return {"status": "not_eligible"}
    parked = sorted({t for t, _ in parks})
    if grant["blocking"]:
        # The client's own rule: a limit this grant does not refill is exhausted too,
        # so refilling the others would not let the account work.
        return {"status": "not_eligible", "reason": "blocked",
                "blocking": grant["blocking"]}
    if grant["use_requires_limit"] and not set(grant["clears"]) & set(status["exhausted"]):
        return {"status": "not_eligible", "reason": "waiting_for_limit"}
    if all(lift is not None and lift - current < min_horizon() for _, lift in parks):
        say(f"{account_id}: limit reset held — parked on {', '.join(parked)}, which "
            f"lifts on its own within {min_horizon() // 60} minutes")
        return {"status": "not_eligible", "reason": "natural_reset_soon"}
    org = organization_uuid(account_dir)
    url = claim_url(usage_url, org) if org else None
    if not url:
        say(f"{account_id}: limit reset available but the account's organization is "
            f"unknown here (no .claude.json oauthAccount); not claiming")
        return {"status": "error", "reason": "no_org"}
    retry = state.get("state") == "pending" and state.get("grant_id") == grant["id"] \
        and _REQUEST_ID.match(str(state.get("request_id") or ""))
    if not retry:
        state = {"schema": 1, "program": PROGRAM, "state": "pending",
                 "grant_id": grant["id"],
                 "request_id": request_id(org, account_uuid(account_dir) or account_id,
                                          grant, status),
                 "resets_left_before": grant["resets_left"], "started_at": current,
                 "clears": grant["clears"], "trigger": parked}
        _write(state_path, state)
    payload = {"program": PROGRAM, "grant_id": state["grant_id"],
               "request_id": state["request_id"]}
    try:
        code, response, retry_after = _post(url, headers, payload)
    except Exception as error:
        say(f"{account_id}: limit reset claim failed ({type(error).__name__}); "
            f"retry is idempotent")
        return {"status": "pending"}
    if 400 <= code < 500:
        backoff = _backoff(state, code, retry_after, current)
        if retry:
            # This send was refused, but an EARLIER one whose answer was lost may have
            # landed: keep the receipt so a count drop can still confirm it.
            state.update(backoff)
            _write(state_path, state)
        else:
            _write(state_path, {"schema": 1, "program": PROGRAM, "state": "backoff",
                                "grant_id": state["grant_id"], **backoff})
        say(f"{account_id}: limit reset claim refused (HTTP {code}); "
            f"not asking again for {backoff['backoff']}s")
        return {"status": "pending" if retry else "refused", "reason": f"http_{code}"}
    if not 200 <= code < 300:
        say(f"{account_id}: limit reset claim failed (HTTP {code}); retry is idempotent")
        return {"status": "pending"}
    outcome = response.get("result") if response.get("result") in _RESULTS else "unavailable"
    cleared = _types(response.get("cleared"))
    left = _count(response.get("resets_left"))
    if outcome in {"reset", "already_used"}:
        said = "redeemed" if outcome == "reset" else "already redeemed"
        say(f"{account_id}: limit reset {said} automatically (parked on {', '.join(parked)})")
        return _complete(state_path, state, outcome, current, cleared, left)
    if outcome == "unavailable":
        say(f"{account_id}: limit reset not confirmed (server unavailable); retry is idempotent")
        return {"status": "pending", "outcome": outcome}
    if retry and outcome in {"cooldown", "not_limited", "ineligible"}:
        # A REPLAY of a claim whose first answer was lost. Claude Code reads these very
        # answers to a retried claim as "your earlier try may have gone through": the
        # first send may have refilled the limits, which is exactly why this one finds
        # nothing to do. Keep the receipt; a later count drop confirms it, the TTL ends it.
        return {"status": "pending", "outcome": outcome}
    # not_limited / ineligible / cooldown on a FIRST send: the server used nothing.
    _remove(state_path)
    return {"status": outcome}


def refresh_reset_credits(account_dir, account_id, usage, usage_url, headers, say, now,
                          threshold=RESET_AT_USED_PERCENT):
    """Read the allowance from this pass's usage response, redeem if due, report both.

    The view reported after a claim is the server's own post-claim count (the claim
    response's ``resets_left`` plus the untouched grants); a claim whose outcome is
    unknown reports no count at all, since a reset may have been spent.
    """
    status = parse_status(usage)
    view = credits_view(status, now)
    result = try_auto_redeem(account_dir, account_id, usage, status, usage_url, headers,
                             say, now, threshold)
    if result.get("status") == "redeemed":
        left = result.get("resets_left")
        if view and type(left) is int:
            others = sum(g["resets_left"] for g in status["grants"]
                         if g["id"] != result.get("grant_id") and _live(g, now))
            view = {"reset_credits_available": others + left,
                    "reset_credits_fetched_at": int(now)}
        else:
            view = {}
    elif result.get("status") == "pending":
        view = {}
    return result, view