Claude Code's usage-limit auto-continue doesn't run in Remote Control sessions. Hooks resume them.

#!/usr/bin/env python3
"""usage-guard: keep a weekly reserve, and resume sessions after the 5-hour limit resets.

Reads ~/.claude/rate-limits.json, which ~/.claude/statusline-command.sh writes from the
statusline's `rate_limits` field (claude.ai plans only). Subcommands:

  usage   print the current numbers, for you or the model
  prompt  UserPromptSubmit hook: add the numbers as context; block the built-in
          auto-continue prompt when weekly usage is at or above the stop line
  gate    PreToolUse hook (Agent|Task|Workflow|SendMessage): ask before dispatching
          or continuing an agent at or above it
  resume  Stop and StopFailure(rate_limit) hook, run with asyncRewake: wait for the
          5-hour reset, then exit 2 to wake the session, unless it already resumed,
          the built-in wait was cancelled, or weekly usage is at or above the stop line

The built-in autoContinueAtUsageLimit never starts its wait while Remote Control is
active (documented; checked in 2.1.288/2.1.289); `resume` covers those sessions. In
other sessions the built-in usually resumes first and `resume` stands down.
"""
import datetime
import json
import os
import re
import shutil
import subprocess
import sys
import time

STOP = float(os.environ.get("USAGE_GUARD_WEEKLY_STOP", "85"))
CACHE = os.path.expanduser(os.environ.get("USAGE_GUARD_CACHE", "~/.claude/rate-limits.json"))
STATE = os.path.expanduser(os.environ.get("USAGE_GUARD_STATE", "~/.claude/usage-guard"))
# Wake this long after the reset. The built-in auto-continue resumes about 1-2 min
# after it, so by then a session it resumed has already written to its transcript.
DELAY = int(os.environ.get("USAGE_GUARD_RESUME_DELAY", "240"))
# Sleep in short steps and re-check the wall clock, so a Mac that slept through the
# reset still resumes within one step of waking.
POLL = int(os.environ.get("USAGE_GUARD_POLL", "30"))
# When the built-in started its own wait, give it this long after the reset before
# waking the session ourselves (its jitter is server-tunable).
BUILTIN_GRACE = int(os.environ.get("USAGE_GUARD_BUILTIN_GRACE", "1200"))
MAX_WAIT = 5.5 * 3600  # how far out a reset may be; the hook's timeout in settings.json is 24 h
MAX_WAKES_PER_DAY = 6
AUTO_CONTINUE = ("Your claude.ai usage limit has reset", "Your claude.ai usage is available again")
CONTINUATION = "Continue the task you were working on when the limit was reached"


def to_epoch(value):
    if isinstance(value, (int, float)):
        return float(value)
    if isinstance(value, str):
        try:
            return float(value)
        except ValueError:
            pass
        try:
            return datetime.datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp()
        except ValueError:
            return None
    return None


def load(now=None):
    """Current usage, or None without a cache. A window whose reset has passed reads as None."""
    now = time.time() if now is None else now
    try:
        with open(CACHE) as f:
            data = json.load(f)
    except (OSError, ValueError):
        return None
    limits = data.get("rate_limits") or {}

    def window(key):
        w = limits.get(key) or {}
        used, reset = w.get("used_percentage"), to_epoch(w.get("resets_at"))
        if not isinstance(used, (int, float)) or (reset is not None and reset <= now):
            used = None
        return used, reset

    h5, h5_reset = window("five_hour")
    wk, wk_reset = window("seven_day")
    return {"h5": h5, "h5_reset": h5_reset, "wk": wk, "wk_reset": wk_reset, "at": to_epoch(data.get("at"))}


def pct(value):
    return "{:g}".format(round(value, 1))


def clock(ts, with_date=False):
    if ts is None:
        return "?"
    return time.strftime("%m/%d %H:%M" if with_date else "%H:%M", time.localtime(ts))


def summary(u, now=None):
    now = time.time() if now is None else now
    weekly = "weekly {}% (resets {})".format(pct(u["wk"]), clock(u["wk_reset"], True)) if u["wk"] is not None else "weekly ?"
    if u["h5"] is not None:
        five = "5-hour {}% (resets {})".format(pct(u["h5"]), clock(u["h5_reset"]))
    else:
        five = "5-hour: new window, no reading yet"
    age = " · read {} min ago".format(int((now - u["at"]) // 60)) if u["at"] else ""
    return "{} · {}{}".format(weekly, five, age)


def over_stop(u):
    return bool(u) and u["wk"] is not None and u["wk"] >= STOP


def emit(obj):
    sys.stdout.write(json.dumps(obj, ensure_ascii=False) + "\n")


def log(sid, text):
    try:
        os.makedirs(STATE, exist_ok=True)
        path = os.path.join(STATE, "log")
        if os.path.exists(path) and os.path.getsize(path) > 512 * 1024:
            os.replace(path, path + ".1")
        with open(path, "a") as f:
            f.write("{} {} {}\n".format(time.strftime("%Y-%m-%d %H:%M:%S"), (sid or "-")[:8], text))
    except OSError:
        pass


def notify(text):
    if os.environ.get("USAGE_GUARD_NOTIFY", "1") == "0":
        return
    try:
        if shutil.which("terminal-notifier"):
            cmd = ["terminal-notifier", "-title", "Claude Code — usage-guard", "-message", text]
        else:
            cmd = ["osascript", "-e", "display notification {} with title \"Claude Code — usage-guard\"".format(
                json.dumps(text, ensure_ascii=False))]
        subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    except OSError:
        pass


def cmd_usage():
    u = load()
    if u is None:
        print("usage-guard: no reading yet ({} is written by the statusline)".format(CACHE))
        return 1
    verdict = "STOP autonomous work" if over_stop(u) else "ok"
    print("{} · stop line {}% → {}".format(summary(u), pct(STOP), verdict))
    return 0


def is_continuation(prompt):
    """The built-in's continuation prompt, also if the server rewords its opening. A longer
    message that only quotes the sentence (a report, a question about it) is not one."""
    if prompt.startswith(AUTO_CONTINUE):
        return True
    at = prompt.find(CONTINUATION)
    return 0 <= at < 150 and len(prompt) < 400 and not prompt.startswith("<")


def cmd_prompt(inp):
    u = load()
    if u is None:
        return 0
    prompt = (inp.get("prompt") or "").lstrip()
    if over_stop(u) and is_continuation(prompt):
        log(inp.get("session_id"), "blocked the built-in auto-continue: weekly {}%".format(pct(u["wk"])))
        notify("Weekly usage {}% ≥ {}%: auto-continue skipped, session stays paused.".format(pct(u["wk"]), pct(STOP)))
        emit({"decision": "block", "reason": "usage-guard: weekly usage is {}%, at or above the {}% reserve line, "
              "so the automatic continue was stopped. Send a prompt to continue anyway.".format(pct(u["wk"]), pct(STOP))})
        return 0
    context = "usage-guard: {}. Autonomous work stops at weekly {}%.".format(summary(u), pct(STOP))
    if over_stop(u):
        context += (" Weekly usage is at or above that line: answer this prompt, but start no new"
                    " autonomous or fan-out work without asking.")
    emit({"hookSpecificOutput": {"hookEventName": "UserPromptSubmit", "additionalContext": context}})
    return 0


def cmd_gate(inp):
    u = load()
    if not over_stop(u):
        return 0
    target = str((inp.get("tool_input") or {}).get("to") or "").strip()
    if inp.get("tool_name") == "SendMessage" and target == "main":
        return 0  # a background agent reporting to the main session
    emit({"hookSpecificOutput": {
        "hookEventName": "PreToolUse",
        "permissionDecision": "ask",
        "permissionDecisionReason": "usage-guard: weekly usage is {}% (reserve line {}%). Approve to run {} anyway, "
                                    "or deny to keep the reserve.".format(pct(u["wk"]), pct(STOP), inp.get("tool_name") or "it"),
    }})
    return 0


def classify(message, u):
    m = message.lower()
    if "session limit" in m:
        return "five_hour"
    if "weekly" in m:
        return "weekly"
    if u and u["h5"] is not None and u["h5"] >= 100:
        return "five_hour"
    if u and u["wk"] is not None and u["wk"] >= 100:
        return "weekly"
    return "other"  # spend limits, per-model limits, unknown


def parse_reset(message, now):
    """'You've hit your session limit · resets 4:40am (Asia/Tokyo)' -> the next 4:40 in that zone."""
    m = re.search(r"resets\s+(?:at\s+)?(\d{1,2})(?::(\d{2}))?\s*([ap]m)\b(?:\s*\(([^)]+)\))?", message, re.I)
    if not m:
        return None
    hour = int(m.group(1)) % 12 + (12 if m.group(3).lower() == "pm" else 0)
    minute = int(m.group(2) or 0)
    zone = None
    if m.group(4):
        try:
            from zoneinfo import ZoneInfo
            zone = ZoneInfo(m.group(4))
        except Exception:
            zone = None
    base = datetime.datetime.fromtimestamp(now, zone) if zone else datetime.datetime.fromtimestamp(now)
    at = base.replace(hour=hour, minute=minute, second=0, microsecond=0)
    if at.timestamp() <= now:
        at += datetime.timedelta(days=1)
    return at.timestamp()


def recent_wakes(sid, now):
    """Count this session's armed resets in the last day, and drop markers older than three days."""
    armed = os.path.join(STATE, "armed")
    count = 0
    try:
        for name in os.listdir(armed):
            path = os.path.join(armed, name)
            age = now - os.path.getmtime(path)
            if age > 3 * 86400:
                shutil.rmtree(path, ignore_errors=True)
            elif name.startswith(sid + "-") and age < 86400:
                count += 1
    except OSError:
        pass
    return count


def sleep_until(ts):
    while True:
        left = ts - time.time()
        if left <= 0:
            return
        time.sleep(min(POLL, left))


def transcript_state(path, reset, armed_at):
    """What the session did since the limit hit, read from the transcript's tail.

    resumed: a turn ran after the reset (a typed prompt, the built-in's continuation, a
    task notification). Only timestamped turn entries count: Claude Code also appends
    untimestamped lines, e.g. when a Remote Control client connects.
    cancelled: the built-in wait was cancelled (Esc, or the session moved elsewhere).
    builtin_armed: the built-in started its own wait for this limit.
    """
    state = {"resumed": False, "cancelled": False, "builtin_armed": False}
    try:
        with open(path, "rb") as f:
            f.seek(0, 2)
            f.seek(max(0, f.tell() - 4 * 1024 * 1024))
            lines = f.read().splitlines()
    except (OSError, TypeError):
        return state
    for raw in lines:
        try:
            entry = json.loads(raw)
        except ValueError:
            continue
        ts = to_epoch(entry.get("timestamp")) if isinstance(entry, dict) else None
        if ts is None:
            continue
        kind = entry.get("type")
        if kind in ("user", "assistant", "queue-operation") and ts > reset:
            state["resumed"] = True
        elif kind == "system" and ts >= armed_at - 120:
            text = str(entry.get("content") or "")
            if text.startswith("Automatic continue cancelled"):
                state["cancelled"] = True
            elif "continuing automatically at" in text:
                state["builtin_armed"] = True
    return state


def cmd_resume(inp):
    # Hold only interactive CLI sessions. In `claude -p` an asyncRewake hook runs in the
    # foreground, so waiting there would hang the run for hours.
    if os.environ.get("CLAUDE_CODE_ENTRYPOINT") != "cli":
        return 0
    event = inp.get("hook_event_name")
    sid = inp.get("session_id") or "unknown"
    message = inp.get("last_assistant_message") or ""
    now = time.time()
    u = load(now)
    if event == "StopFailure":
        if inp.get("error") != "rate_limit":
            return 0
        kind = classify(message, u)
    elif event == "Stop":
        if not u or u["h5"] is None or u["h5"] < 100:
            return 0  # the usual case: a turn ended below the limit
        kind = "five_hour"
    else:
        return 0

    if kind == "weekly":
        log(sid, "weekly limit hit: no auto-resume")
        notify("Weekly limit reached: this session will not resume on its own.")
        return 0
    if kind != "five_hour":
        log(sid, "rate limit is not the 5-hour window ({}): no auto-resume".format(message.splitlines()[0][:80] if message else "?"))
        return 0
    reset = u["h5_reset"] if u and u["h5_reset"] and u["h5_reset"] > now else parse_reset(message, now)
    if reset is None or reset - now > MAX_WAIT:
        log(sid, "no usable 5-hour reset time: no auto-resume")
        return 0
    if over_stop(u):
        log(sid, "weekly {}% >= {}%: staying paused after the reset".format(pct(u["wk"]), pct(STOP)))
        notify("Weekly usage {}% ≥ {}%: session paused, no auto-resume.".format(pct(u["wk"]), pct(STOP)))
        return 0
    if recent_wakes(sid, now) >= MAX_WAKES_PER_DAY:
        log(sid, "{} resumes in the last day: giving up".format(MAX_WAKES_PER_DAY))
        return 0
    try:
        os.makedirs(os.path.join(STATE, "armed", "{}-{}".format(sid, int(reset))))
    except FileExistsError:
        return 0  # another Stop/StopFailure already armed this reset
    except OSError:
        return 0
    armed_at = time.time()
    target = reset + DELAY
    log(sid, "armed via {}: wake at {}".format(event, clock(target, True)))

    transcript = inp.get("transcript_path")
    while True:
        sleep_until(target)
        state = transcript_state(transcript, reset, armed_at)
        if state["resumed"]:
            log(sid, "already resumed (built-in auto-continue or a prompt): not waking")
            return 0
        if state["cancelled"]:
            log(sid, "the built-in wait was cancelled (Esc, or the session moved): not waking")
            return 0
        if state["builtin_armed"] and time.time() < reset + BUILTIN_GRACE:
            target = min(time.time() + 2 * POLL, reset + BUILTIN_GRACE)  # its continuation may still come
            continue
        break
    u = load()
    if over_stop(u):
        log(sid, "weekly {}% >= {}% after the reset: staying paused".format(pct(u["wk"]), pct(STOP)))
        notify("Weekly usage {}% ≥ {}%: session paused, no auto-resume.".format(pct(u["wk"]), pct(STOP)))
        return 0
    weekly = "{}%".format(pct(u["wk"])) if u and u["wk"] is not None else "unknown"
    sys.stderr.write(
        "usage-guard: the 5-hour usage limit reset at {}. Weekly usage is {}; autonomous work stops at {}%. "
        "Continue the task you were working on when the limit was reached, without redoing finished work. "
        "If that task was already done, reply with one short line and stop.\n".format(clock(reset), weekly, pct(STOP)))
    log(sid, "woke the session")
    return 2


def main():
    cmd = sys.argv[1] if len(sys.argv) > 1 else "usage"
    if cmd == "usage":
        return cmd_usage()
    handler = {"prompt": cmd_prompt, "gate": cmd_gate, "resume": cmd_resume}.get(cmd)
    if handler is None:
        sys.stderr.write("usage: usage_guard.py [usage|prompt|gate|resume]\n")
        return 1
    try:
        inp = json.load(sys.stdin)
        return handler(inp if isinstance(inp, dict) else {})
    except Exception as exc:  # a broken guard must never block a prompt, a tool, or a stop
        log("-", "{} failed: {!r}".format(cmd, exc))
        return 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env bash
# Tests for usage_guard.py. Uses a fake cache and state dir; never touches the real ones.
set -u
G="$(cd "$(dirname "$0")" && pwd)/usage_guard.py"
T=$(mktemp -d); trap 'rm -rf "$T"' EXIT
export USAGE_GUARD_CACHE="$T/rate-limits.json" USAGE_GUARD_STATE="$T/state" USAGE_GUARD_NOTIFY=0
export USAGE_GUARD_RESUME_DELAY=1 USAGE_GUARD_POLL=1
pass=0; fail=0
ok()   { pass=$((pass+1)); }
bad()  { fail=$((fail+1)); echo "FAIL: $1"; }
check() { if eval "$2"; then ok; else bad "$1"; fi; }

cache() { # h5 h5_reset_offset wk [wk_reset_offset]
  local now; now=$(date +%s)
  printf '{"rate_limits":{"five_hour":{"used_percentage":%s,"resets_at":%s},"seven_day":{"used_percentage":%s,"resets_at":%s}},"at":%s}' \
    "$1" $((now + $2)) "$3" $((now + ${4:-300000})) "$now" > "$USAGE_GUARD_CACHE"
}
hook() { # subcommand json -> sets out, rc
  out=$(printf '%s' "$2" | python3 "$G" "$1" 2>"$T/err"); rc=$?; err=$(cat "$T/err")
}

# usage
cache 40 3600 47
out=$(python3 "$G" usage); check "usage prints numbers" '[[ "$out" == *"weekly 47%"* && "$out" == *"5-hour 40%"* && "$out" == *"→ ok"* ]]'
rm -f "$USAGE_GUARD_CACHE"
python3 "$G" usage >/dev/null; check "usage without cache exits 1" '[ $? -eq 1 ]'

# prompt
hook prompt '{"prompt":"hi"}'; check "prompt without cache is silent" '[ $rc -eq 0 ] && [ -z "$out" ]'
cache 40 3600 47
hook prompt '{"prompt":"hi"}'
check "prompt adds context" '[ $rc -eq 0 ] && [[ "$out" == *additionalContext* && "$out" == *"weekly 47%"* ]]'
check "prompt below line has no stop note" '[[ "$out" != *"at or above"* ]]'
hook prompt '{"prompt":"Your claude.ai usage limit has reset. Continue the task you were working on"}'
check "auto-continue passes below the line" '[[ "$out" == *additionalContext* && "$out" != *"\"block\""* ]]'
cache 40 3600 90
hook prompt '{"prompt":"Your claude.ai usage limit has reset. Continue the task you were working on"}'
check "auto-continue blocked at 90%" '[ $rc -eq 0 ] && [[ "$out" == *"\"decision\": \"block\""* ]]'
hook prompt '{"prompt":"Your claude.ai usage is available again before the usage-limit reset."}'
check "early auto-continue blocked at 90%" '[[ "$out" == *"\"block\""* ]]'
hook prompt '{"prompt":"what is 2+2"}'
check "user prompt at 90% passes with a note" '[[ "$out" == *additionalContext* && "$out" == *"at or above"* ]]'
cache 40 3600 90 -10
hook prompt '{"prompt":"Your claude.ai usage limit has reset."}'
check "stale weekly (window over) does not block" '[[ "$out" != *"\"block\""* && "$out" == *"weekly ?"* ]]'

# gate
cache 40 3600 84.9
hook gate '{"tool_name":"Agent"}'; check "gate silent below line" '[ $rc -eq 0 ] && [ -z "$out" ]'
cache 40 3600 85
hook gate '{"tool_name":"Agent"}'; check "gate asks at 85%" '[[ "$out" == *"\"permissionDecision\": \"ask\""* && "$out" == *Agent* ]]'
hook gate '{"tool_name":"SendMessage","tool_input":{"to":"coder-7","message":"fix round 2"}}'
check "gate asks before continuing an agent" '[[ "$out" == *"\"ask\""* ]]'
hook gate '{"tool_name":"SendMessage","tool_input":{"to":"main","message":"done"}}'
check "gate lets an agent report to main" '[ $rc -eq 0 ] && [ -z "$out" ]'
cache 40 3600 90
hook prompt '{"prompt":"[usage] Continue the task you were working on when the limit was reached; do not repeat work."}'
check "reworded continuation still blocked" '[[ "$out" == *"\"block\""* ]]'
long=$(python3 -c 'print("Report: the hook matches \"Continue the task you were working on when the limit was reached\" and " + "x" * 400)')
hook prompt "$(python3 -c 'import json,sys;print(json.dumps({"prompt":sys.argv[1]}))' "$long")"
check "a long report quoting the sentence passes" '[[ "$out" != *"\"block\""* && "$out" == *additionalContext* ]]'
hook prompt '{"prompt":"<task-notification> Continue the task you were working on when the limit was reached </task-notification>"}'
check "a task notification quoting it passes" '[[ "$out" != *"\"block\""* ]]'

# resume: guards
cache 101 3 50
export CLAUDE_CODE_ENTRYPOINT=sdk-cli
hook resume '{"hook_event_name":"Stop","session_id":"s1"}'
check "resume skips non-cli entrypoints" '[ $rc -eq 0 ] && [ ! -d "$USAGE_GUARD_STATE/armed" ]'
export CLAUDE_CODE_ENTRYPOINT=cli
cache 60 3 50
start=$(date +%s); hook resume '{"hook_event_name":"Stop","session_id":"s1"}'
check "Stop below the limit exits at once" '[ $rc -eq 0 ] && [ $(( $(date +%s) - start )) -le 1 ]'
hook resume '{"hook_event_name":"StopFailure","error":"overloaded","session_id":"s1"}'
check "StopFailure other error exits" '[ $rc -eq 0 ] && [ ! -d "$USAGE_GUARD_STATE/armed" ]'
hook resume '{"hook_event_name":"StopFailure","error":"rate_limit","session_id":"s1","last_assistant_message":"You have hit your weekly limit · resets Oct 8, 5am"}'
check "weekly limit is not resumed" '[ $rc -eq 0 ] && grep -q "weekly limit hit" "$USAGE_GUARD_STATE/log"'
cache 60 3 50
hook resume '{"hook_event_name":"StopFailure","error":"rate_limit","session_id":"s1","last_assistant_message":"You have reached your Fable limit."}'
check "per-model limit is not resumed" '[ $rc -eq 0 ] && grep -q "not the 5-hour window" "$USAGE_GUARD_STATE/log"'
cache 101 3 90
hook resume '{"hook_event_name":"Stop","session_id":"s1"}'
check "weekly over the line stays paused" '[ $rc -eq 0 ] && grep -q "staying paused" "$USAGE_GUARD_STATE/log"'

# resume: wakes after the reset
touch "$T/transcript.jsonl"; sleep 1
cache 101 3 50
start=$(date +%s)
hook resume "{\"hook_event_name\":\"Stop\",\"session_id\":\"s2\",\"transcript_path\":\"$T/transcript.jsonl\"}"
took=$(( $(date +%s) - start ))
check "Stop at the limit wakes with exit 2" '[ $rc -eq 2 ] && [[ "$err" == *"5-hour usage limit reset"* && "$err" == *"Weekly usage is 50%"* ]]'
check "it waited for reset + delay" '[ $took -ge 3 ] && [ $took -le 7 ]'

# resume: what the transcript says after the limit hit
entry() { # type [content] -> one timestamped transcript line, like Claude Code writes
  python3 -c 'import json,sys,datetime;print(json.dumps({"type":sys.argv[1],"timestamp":datetime.datetime.utcnow().isoformat(timespec="milliseconds")+"Z","content":sys.argv[2]}))' "$1" "${2:-}"
}
sf() { # session_id transcript -> StopFailure input for a session limit
  printf '{"hook_event_name":"StopFailure","error":"rate_limit","session_id":"%s","transcript_path":"%s","last_assistant_message":"You'"'"'ve hit your session limit · resets 4:40am (Asia/Tokyo)"}' "$1" "$2"
}
: > "$T/t3.jsonl"; cache 101 2 50
( sleep 2.5; entry user "Your claude.ai usage limit has reset. Continue the task" >> "$T/t3.jsonl" ) &
USAGE_GUARD_RESUME_DELAY=2 hook resume "$(sf s3 "$T/t3.jsonl")"
check "no wake when a turn ran after the reset" '[ $rc -eq 0 ] && grep -q "s3 already resumed" "$USAGE_GUARD_STATE/log"'

: > "$T/t6.jsonl"; cache 101 2 50
( sleep 2.5; echo '{"type":"bridge-session","sessionId":"x"}' >> "$T/t6.jsonl" ) &
USAGE_GUARD_RESUME_DELAY=2 hook resume "$(sf s6 "$T/t6.jsonl")"
check "an untimestamped bridge line does not count as resumed" '[ $rc -eq 2 ]'

: > "$T/t7.jsonl"; cache 101 2 50
( sleep 0.5; entry system "Automatic continue cancelled" >> "$T/t7.jsonl" ) &
USAGE_GUARD_RESUME_DELAY=2 hook resume "$(sf s7 "$T/t7.jsonl")"
check "Esc on the built-in countdown is respected" '[ $rc -eq 0 ] && grep -q "s7 the built-in wait was cancelled" "$USAGE_GUARD_STATE/log"'

: > "$T/t8.jsonl"; cache 101 2 50
( sleep 0.5; entry system "Usage limit reached · continuing automatically at 4:40am · esc to cancel" >> "$T/t8.jsonl" ) &
start=$(date +%s)
USAGE_GUARD_RESUME_DELAY=1 USAGE_GUARD_BUILTIN_GRACE=6 hook resume "$(sf s8 "$T/t8.jsonl")"
took=$(( $(date +%s) - start ))
check "a silent built-in gets its grace, then the guard wakes" '[ $rc -eq 2 ] && [ $took -ge 5 ] && [ $took -le 10 ]'

: > "$T/t9.jsonl"; cache 101 2 50
( sleep 0.5; entry system "Usage limit reached · continuing automatically at 4:40am · esc to cancel" >> "$T/t9.jsonl"; sleep 4; entry user "Your claude.ai usage limit has reset." >> "$T/t9.jsonl" ) &
USAGE_GUARD_RESUME_DELAY=1 USAGE_GUARD_BUILTIN_GRACE=30 hook resume "$(sf s9 "$T/t9.jsonl")"
check "a late built-in continuation makes the guard stand down" '[ $rc -eq 0 ] && grep -q "s9 already resumed" "$USAGE_GUARD_STATE/log"'

# resume: Stop and StopFailure for the same reset arm once
cache 101 3 50
printf '%s' "{\"hook_event_name\":\"StopFailure\",\"error\":\"rate_limit\",\"session_id\":\"s4\",\"last_assistant_message\":\"You've hit your session limit\"}" | python3 "$G" resume 2>/dev/null & p1=$!
sleep 0.5
start=$(date +%s); hook resume '{"hook_event_name":"Stop","session_id":"s4"}'
check "second hook for the same reset exits at once" '[ $rc -eq 0 ] && [ $(( $(date +%s) - start )) -le 1 ]'
wait $p1; check "first hook still wakes" '[ $? -eq 2 ]'

# resume: weekly crossed the line during the wait -> no wake
cache 101 2 50
( sleep 1; cache 101 -60 88 ) &
hook resume '{"hook_event_name":"Stop","session_id":"s5"}'
check "weekly crossing during the wait stays paused" '[ $rc -eq 0 ] && grep -q "after the reset: staying paused" "$USAGE_GUARD_STATE/log"'

# reset time parsed from the message when the cache has none
out=$(USAGE_GUARD_CACHE=/nonexistent python3 - "$G" <<'EOF'
import importlib.util, sys, datetime
spec = importlib.util.spec_from_file_location("g", sys.argv[1]); g = importlib.util.module_from_spec(spec); spec.loader.exec_module(g)
from zoneinfo import ZoneInfo
tz = ZoneInfo("Asia/Tokyo")
now = datetime.datetime(2026, 10, 4, 1, 2, tzinfo=tz).timestamp()
r = g.parse_reset("You've hit your session limit · resets 4:40am (Asia/Tokyo)", now)
print(datetime.datetime.fromtimestamp(r, tz).strftime("%m/%d %H:%M"))
r = g.parse_reset("You've hit your session limit · resets 11:40pm (Asia/Tokyo)", datetime.datetime(2026, 10, 3, 22, 43, tzinfo=tz).timestamp())
print(datetime.datetime.fromtimestamp(r, tz).strftime("%m/%d %H:%M"))
r = g.parse_reset("resets 12:10am (Asia/Tokyo)", datetime.datetime(2026, 10, 3, 23, 0, tzinfo=tz).timestamp())
print(datetime.datetime.fromtimestamp(r, tz).strftime("%m/%d %H:%M"))
print(g.parse_reset("no time here", now))
EOF
)
check "parse_reset reads the limit message" '[ "$out" = "$(printf "10/04 04:40\n10/03 23:40\n10/04 00:10\nNone")" ]'

# malformed input never blocks
out=$(echo 'not json' | python3 "$G" prompt); check "malformed input exits 0" '[ $? -eq 0 ]'

echo "passed $pass, failed $fail"
[ $fail -eq 0 ]