Get Token Insights

#!/usr/bin/env python3
"""
UserPromptSubmit hook: warn once when prompt cache has likely expired.

Claude Code uses the 5-minute prompt cache TTL (ephemeral_5m). When the user
has been idle >5 minutes since Claude last responded, the cached context
will expire on this turn — costing full re-creation tokens. This hook fires
before the prompt is sent and blocks once per idle gap so the user can
compact before the expensive re-creation happens.

State: ~/.claude-memory/cache-warn/<session_id>.json (written by cache-warn-stop.py)
  - last_stop_time: ISO timestamp of last Claude response
  - warned_gaps: list of gap buckets already warned — prevents double-warn

Gap bucket = floor(last_stop_time_unix / 60) — stable per idle gap even if
the hook fires multiple times after the same idle period.
"""

from __future__ import annotations

import json
import math
import sys
from datetime import datetime, timezone
from pathlib import Path

CACHE_TTL_SECONDS = 300       # Claude Code uses 5-minute prompt cache TTL (ephemeral_5m, changed ~Apr 3 2026)
WARN_THRESHOLD_SECONDS = 300  # Warn at exactly 5 min (TTL expiry)
CACHE_WARN_DIR = Path.home() / ".claude-memory" / "cache-warn"


def _gap_bucket(last_stop_time_iso: str) -> str:
    """Stable ID for this idle gap — floor of last_stop_time in whole minutes."""
    dt = datetime.fromisoformat(last_stop_time_iso.replace("Z", "+00:00"))
    return str(math.floor(dt.timestamp() / 60))


def _format_tokens(n: int) -> str:
    if n >= 1000:
        return f"{n / 1000:.0f}k"
    return str(n)


def _safe_state_path(cache_dir: Path, prefix: str, session_id: str) -> Path | None:
    """Return resolved path only if it stays within cache_dir; else None."""
    try:
        candidate = (cache_dir / f"{prefix}{session_id}.json").resolve()
        candidate.relative_to(cache_dir.resolve())
        return candidate
    except (ValueError, OSError, RuntimeError):
        return None


def check_resume_warn(session_id: str) -> str | None:
    """If a resume-pending flag exists for this session, consume it and return a warning."""
    flag_path = _safe_state_path(CACHE_WARN_DIR, "resume-pending-", session_id)
    if flag_path is None or not flag_path.exists():
        return None

    try:
        flag = json.loads(flag_path.read_text())
        cached_tokens = flag.get("cached_tokens", 0)
    except (json.JSONDecodeError, OSError):
        cached_tokens = 0
    finally:
        try:
            flag_path.unlink()
        except OSError:
            pass

    token_str = f"~{_format_tokens(cached_tokens)} cached tokens" if cached_tokens else "cached context"
    return (
        f"Resumed session: this turn will re-process {token_str} from scratch "
        f"(5-minute cache TTL likely expired while session was inactive).\n"
        f"Run /compact to reduce re-creation cost, then re-send — or re-send now to proceed."
    )


def main() -> None:
    try:
        _main()
    except Exception:
        # Never hard-block a UserPromptSubmit hook — always let the prompt through
        print(json.dumps({"continue": True}))


def _main() -> None:
    try:
        hook_input = json.load(sys.stdin)
    except (json.JSONDecodeError, EOFError):
        hook_input = {}

    session_id = hook_input.get("session_id", "")
    if not session_id:
        print(json.dumps({"continue": True}))
        return

    # Skip cache warning for system-generated messages — these are not human prompts
    # and blocking them breaks background task result delivery.
    _prompt = hook_input.get("prompt", "").lstrip()
    _SYSTEM_PREFIXES = (
        "<task-notification>",
        "<local-command-caveat>",
        "<command-name>",
        "<command-message>",
    )
    if any(_prompt.startswith(p) for p in _SYSTEM_PREFIXES):
        print(json.dumps({"continue": True}))
        return

    # Check resume warning first — takes priority over idle-gap warning
    resume_warning = check_resume_warn(session_id)
    if resume_warning:
        # Also mark the current gap as warned so the idle-gap check doesn't
        # double-fire on the re-send after this block.
        state_path = _safe_state_path(CACHE_WARN_DIR, "", session_id)
        if state_path is not None and state_path.exists():
            try:
                state = json.loads(state_path.read_text())
                last_stop_iso = state.get("last_stop_time", "")
                if last_stop_iso:
                    warned_gaps = state.get("warned_gaps", [])
                    bucket = _gap_bucket(last_stop_iso)
                    if bucket not in warned_gaps:
                        warned_gaps.append(bucket)
                        state_path.write_text(json.dumps({
                            "session_id": session_id,
                            "last_stop_time": last_stop_iso,
                            "warned_gaps": warned_gaps,
                        }))
            except (json.JSONDecodeError, OSError, ValueError):
                pass
        print(json.dumps({"decision": "block", "reason": resume_warning}))
        return

    state_path = _safe_state_path(CACHE_WARN_DIR, "", session_id)
    if state_path is None or not state_path.exists():
        # Stop hook hasn't fired yet this session — no baseline to compare
        print(json.dumps({"continue": True}))
        return

    try:
        state = json.loads(state_path.read_text())
    except (json.JSONDecodeError, OSError):
        print(json.dumps({"continue": True}))
        return

    last_stop_iso = state.get("last_stop_time", "")
    warned_gaps: list = state.get("warned_gaps", [])

    if not last_stop_iso:
        print(json.dumps({"continue": True}))
        return

    try:
        last_stop_dt = datetime.fromisoformat(last_stop_iso.replace("Z", "+00:00"))
    except ValueError:
        print(json.dumps({"continue": True}))
        return

    now = datetime.now(timezone.utc)
    gap_seconds = (now - last_stop_dt).total_seconds()

    if gap_seconds < WARN_THRESHOLD_SECONDS:
        print(json.dumps({"continue": True}))
        return

    bucket = _gap_bucket(last_stop_iso)
    if bucket in warned_gaps:
        # Already warned for this idle gap — let this prompt through
        print(json.dumps({"continue": True}))
        return

    # First warning for this gap — record and block
    warned_gaps.append(bucket)
    try:
        state_path.write_text(json.dumps({
            "session_id": session_id,
            "last_stop_time": last_stop_iso,
            "warned_gaps": warned_gaps,
        }))
    except OSError:
        pass

    idle_minutes = gap_seconds / 60

    overrun = gap_seconds - CACHE_TTL_SECONDS
    status = f"expired ~{overrun / 60:.0f} min ago" if overrun >= 60 else "just expired"

    warning = (
        f"Cache warning: prompt cache {status} (idle: {idle_minutes:.1f} min). "
        f"This turn will re-process full context from scratch.\n"
        f"Re-send to proceed, or run /compact first to reduce cost."
    )

    print(json.dumps({
        "decision": "block",
        "reason": warning,
    }))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
SessionStart hook: detect resume sessions and write a pending-warn flag.

When source == "resume", reads the transcript JSONL to find the last
cache_creation_input_tokens value, then writes a flag file so
cache-expiry-warn.py can block the first UserPromptSubmit with a cost warning.

Flag file: ~/.claude-memory/cache-warn/resume-pending-<session_id>.json
  - cached_tokens: last known cache_creation_input_tokens from transcript
"""

from __future__ import annotations

import json
import sys
from pathlib import Path

CACHE_WARN_DIR = Path.home() / ".claude-memory" / "cache-warn"
_CLAUDE_DIR = Path.home() / ".claude"


def _safe_state_path(cache_dir: Path, prefix: str, session_id: str) -> Path | None:
    """Return resolved path only if it stays within cache_dir; else None."""
    try:
        candidate = (cache_dir / f"{prefix}{session_id}.json").resolve()
        candidate.relative_to(cache_dir.resolve())
        return candidate
    except (ValueError, OSError, RuntimeError):
        return None


def get_cached_tokens(transcript_path: str) -> int:
    """Read last assistant message usage from transcript JSONL."""
    path = Path(str(transcript_path or ""))
    try:
        resolved = path.resolve()
        resolved.relative_to(_CLAUDE_DIR.resolve())
    except (ValueError, OSError, RuntimeError):
        return 0
    if not path.exists():
        return 0

    cached_tokens = 0
    try:
        for line in path.read_text(encoding="utf-8").splitlines():
            line = line.strip()
            if not line:
                continue
            try:
                entry = json.loads(line)
            except json.JSONDecodeError:
                continue
            # Look for assistant turns with usage stats
            msg = entry.get("message", {})
            if msg.get("role") == "assistant":
                usage = msg.get("usage", {})
                ct = usage.get("cache_creation_input_tokens", 0) + usage.get("cache_read_input_tokens", 0)
                if ct:
                    cached_tokens = ct
    except OSError:
        pass

    return cached_tokens


def main() -> None:
    try:
        raw = sys.stdin.read()
        data = json.loads(raw) if raw.strip() else {}
    except Exception:
        data = {}

    if data.get("source") != "resume":
        return

    session_id = data.get("session_id", "")
    transcript_path = data.get("transcript_path", "")

    if not session_id:
        return

    cached_tokens = get_cached_tokens(transcript_path)

    CACHE_WARN_DIR.mkdir(parents=True, exist_ok=True)
    flag_path = _safe_state_path(CACHE_WARN_DIR, "resume-pending-", session_id)
    if flag_path is None:
        return
    flag_path.write_text(json.dumps({
        "session_id": session_id,
        "cached_tokens": cached_tokens,
    }))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Stop hook: record the time Claude last finished responding, for cache expiry warning.

Called after every Claude turn. Writes ~/.claude-memory/cache-warn/<session_id>.json
with the current timestamp so cache-expiry-warn.py (UserPromptSubmit hook) can
compute the idle gap on the next user prompt.

Preserves any existing warned_gaps list so the warn-once-per-gap logic survives
across multiple turns in the same session.
"""

from __future__ import annotations

import json
import sys
from datetime import datetime, timezone
from pathlib import Path

CACHE_WARN_DIR = Path.home() / ".claude-memory" / "cache-warn"


def _safe_state_path(cache_dir: Path, prefix: str, session_id: str) -> Path | None:
    """Return resolved path only if it stays within cache_dir; else None."""
    try:
        candidate = (cache_dir / f"{prefix}{session_id}.json").resolve()
        candidate.relative_to(cache_dir.resolve())
        return candidate
    except (ValueError, OSError, RuntimeError):
        return None


def main() -> None:
    try:
        raw = sys.stdin.read()
        data = json.loads(raw) if raw.strip() else {}
        session_id = data.get("session_id", "")
        if not session_id:
            return

        CACHE_WARN_DIR.mkdir(parents=True, exist_ok=True)
        state_path = _safe_state_path(CACHE_WARN_DIR, "", session_id)
        if state_path is None:
            return

        warned_gaps: list = []
        if state_path.exists():
            try:
                existing = json.loads(state_path.read_text())
                warned_gaps = existing.get("warned_gaps", [])
            except (json.JSONDecodeError, OSError):
                pass

        state_path.write_text(json.dumps({
            "session_id": session_id,
            "last_stop_time": datetime.now(timezone.utc).isoformat(),
            "warned_gaps": warned_gaps,
        }))
    except Exception:
        pass  # Never block the Stop hook


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Cache-bust hook installer for get-token-insights.

Copies 3 validated hook scripts into ~/.claude/hooks/ and merges the
corresponding hook matcher blocks into ~/.claude/settings.json.

Hooks installed:
  SessionStart  → cache-resume-detect.py  (flags resumed sessions for warn)
  Stop          → cache-warn-stop.py      (stamps last-idle timestamp)
  UserPromptSubmit → cache-expiry-warn.py (blocks prompt when cache expired)

Usage:
  python3 install_cache_hooks.py [--dry-run]

Flags:
  --dry-run   Print what would change without writing anything.
  --status    Show which hooks are installed, skip install.
"""

from __future__ import annotations

import argparse
import json
import shutil
import sys
from datetime import datetime, timezone
from pathlib import Path

ASSET_DIR = Path(__file__).parent.parent / "assets" / "cache-hooks"
HOOKS_DIR = Path.home() / ".claude" / "hooks"
SETTINGS_PATH = Path.home() / ".claude" / "settings.json"

HOOK_FILES = [
    "cache-resume-detect.py",
    "cache-warn-stop.py",
    "cache-expiry-warn.py",
]

# Exact matcher blocks to inject — keyed by (event, command_substring) for
# idempotency check. The installer skips injection if any existing block's
# command already contains the substring.
HOOK_BLOCKS = {
    "SessionStart": {
        "check_substr": "cache-resume-detect.py",
        "block": {
            "matcher": "*",
            "hooks": [{"type": "command", "command": "python3 ~/.claude/hooks/cache-resume-detect.py"}],
        },
    },
    "Stop": {
        "check_substr": "cache-warn-stop.py",
        "block": {
            "matcher": "*",
            "hooks": [{"type": "command", "command": "python3 ~/.claude/hooks/cache-warn-stop.py"}],
        },
    },
    "UserPromptSubmit": {
        "check_substr": "cache-expiry-warn.py",
        "block": {
            "matcher": "*",
            "hooks": [{"type": "command", "command": "python3 ~/.claude/hooks/cache-expiry-warn.py"}],
        },
    },
}


def _already_in_hooks(event_list: list[dict], substr: str) -> bool:
    """Return True if any existing hook command contains substr."""
    for matcher_block in event_list:
        for hook in matcher_block.get("hooks", []):
            if substr in hook.get("command", ""):
                return True
    return False


def check_status() -> dict[str, dict]:
    """Return install status for each component."""
    result: dict[str, dict] = {}
    for fname in HOOK_FILES:
        dest = HOOKS_DIR / fname
        result[fname] = {"file_exists": dest.exists()}
    if SETTINGS_PATH.exists():
        try:
            cfg = json.loads(SETTINGS_PATH.read_text())
        except json.JSONDecodeError:
            cfg = {}
        hooks_cfg = cfg.get("hooks", {})
        for event, spec in HOOK_BLOCKS.items():
            wired = _already_in_hooks(hooks_cfg.get(event, []), spec["check_substr"])
            result[f"settings:{event}"] = {"wired": wired}
    return result


class InstallError(Exception):
    """Raised by install() on hard failures; caught by main() which calls sys.exit(1)."""


def install(dry_run: bool = False) -> bool:
    """
    Install hook scripts and settings.json entries.
    Returns True if any change was made (or would be made in dry-run).
    Raises InstallError on hard failures (missing assets, bad settings.json).
    """
    changed = False

    # 1. Copy hook scripts (skip if identical)
    HOOKS_DIR.mkdir(parents=True, exist_ok=True)
    for fname in HOOK_FILES:
        src = ASSET_DIR / fname
        dest = HOOKS_DIR / fname
        if not src.exists():
            raise InstallError(f"Asset not found: {src}")
        if dest.exists():
            if dest.read_bytes() == src.read_bytes():
                print(f"  [skip] {fname} — already up to date")
                continue
            else:
                print(f"  [update] {fname} — content differs, will overwrite")
        else:
            print(f"  [install] {fname}")
        if not dry_run:
            shutil.copy2(src, dest)
            dest.chmod(0o644)
        changed = True

    # 2. Merge settings.json
    if not SETTINGS_PATH.exists():
        raise InstallError("~/.claude/settings.json not found — cannot wire hooks")

    try:
        cfg = json.loads(SETTINGS_PATH.read_text())
    except json.JSONDecodeError as exc:
        raise InstallError(f"settings.json is malformed: {exc}") from exc

    hooks_cfg = cfg.setdefault("hooks", {})
    settings_changed = False

    for event, spec in HOOK_BLOCKS.items():
        event_list = hooks_cfg.setdefault(event, [])
        if _already_in_hooks(event_list, spec["check_substr"]):
            print(f"  [skip] settings.json {event} hook — already wired")
            continue
        print(f"  [add] settings.json {event} hook → {spec['check_substr']}")
        if not dry_run:
            event_list.append(spec["block"])
        settings_changed = True
        changed = True

    if settings_changed and not dry_run:
        # Backup before write
        ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
        backup = SETTINGS_PATH.with_suffix(f".json.bak-{ts}")
        shutil.copy2(SETTINGS_PATH, backup)
        print(f"  [backup] settings.json → {backup.name}")
        SETTINGS_PATH.write_text(json.dumps(cfg, indent=2))
        print("  [wrote] ~/.claude/settings.json")

    return changed


def main() -> None:
    parser = argparse.ArgumentParser(description="Install Claude cache-bust warning hooks")
    parser.add_argument("--dry-run", action="store_true", help="Show what would change without writing")
    parser.add_argument("--status", action="store_true", help="Show current install status and exit")
    args = parser.parse_args()

    if args.status:
        status = check_status()
        print("Cache-bust hook install status:")
        for key, val in status.items():
            flags = ", ".join(f"{k}={v}" for k, v in val.items())
            print(f"  {key}: {flags}")
        return

    if args.dry_run:
        print("[dry-run] No files will be written.\n")

    try:
        changed = install(dry_run=args.dry_run)
    except InstallError as exc:
        print(f"[ERROR] {exc}", file=sys.stderr)
        sys.exit(1)

    if not changed:
        print("\nAll hooks already installed — nothing to do.")
    elif args.dry_run:
        print("\n[dry-run complete] Re-run without --dry-run to apply.")
    else:
        print("\nInstall complete. Restart Claude Code for hooks to take effect.")
        print("What changes when you idle >5 minutes:")
        print("  1. Your next prompt will be blocked once with a cost warning.")
        print("  2. Press ↑ to resend as-is (accepts the cache rebuild cost).")
        print("  3. Or run /compact to compress context, then resend.")
        print("  4. Or run /clear to start a fresh session with no rebuild cost.")
        print("One warning per idle gap — not per prompt. No repeat nags.")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Tests for security fixes and the _bust_overhead cost function.

Coverage:
  - _safe_state_path (all three hook files) — path traversal prevention
  - get_cached_tokens (cache-resume-detect) — transcript bounds check
  - _bust_overhead (ingest_token_data) — incremental cache-bust cost vs _turn_cost
"""

from __future__ import annotations

import importlib.util
import json
import sys
from pathlib import Path

import pytest

# ---------------------------------------------------------------------------
# Import helpers — hook files use hyphens, not valid identifiers
# ---------------------------------------------------------------------------

_HOOKS_DIR = (
    Path(__file__).resolve().parent.parent
    / "assets" / "cache-hooks"
)
_SCRIPTS_DIR = (
    Path(__file__).resolve().parent.parent / "scripts"
)


def _load_hook(filename: str):
    """Load a hyphenated hook script as a module via importlib."""
    module_name = filename.replace("-", "_").removesuffix(".py")
    spec = importlib.util.spec_from_file_location(
        module_name, _HOOKS_DIR / filename
    )
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


# Load once at module import so parametrize can reference the functions
_expiry_warn = _load_hook("cache-expiry-warn.py")
_resume_detect = _load_hook("cache-resume-detect.py")
_warn_stop = _load_hook("cache-warn-stop.py")

# Load ingest script
sys.path.insert(0, str(_SCRIPTS_DIR))
from ingest_token_data import _bust_overhead, _turn_cost, _get_pricing  # noqa: E402

# ---------------------------------------------------------------------------
# Parametrize _safe_state_path across all three hook modules
# ---------------------------------------------------------------------------

_HOOK_MODULES = [
    pytest.param(_expiry_warn, id="cache-expiry-warn"),
    pytest.param(_resume_detect, id="cache-resume-detect"),
    pytest.param(_warn_stop, id="cache-warn-stop"),
]


# ---------------------------------------------------------------------------
# _safe_state_path — path traversal prevention
# ---------------------------------------------------------------------------

class TestSafeStatePath:
    """_safe_state_path must confine state files within cache_dir.

    A crafted session_id that traverses out of cache_dir would let an
    attacker read or overwrite arbitrary files (e.g. ~/.claude/settings.json).
    Returning None on traversal prevents that write/read from ever happening.
    """

    @pytest.mark.parametrize("mod", _HOOK_MODULES)
    def test_normal_session_id_returns_path_inside_cache_dir(self, mod, tmp_path):
        result = mod._safe_state_path(tmp_path, "", "abc123")
        assert result is not None
        assert result.is_relative_to(tmp_path)
        assert result.name == "abc123.json"

    @pytest.mark.parametrize("mod", _HOOK_MODULES)
    def test_normal_session_id_with_prefix(self, mod, tmp_path):
        result = mod._safe_state_path(tmp_path, "resume-pending-", "abc123")
        assert result is not None
        assert result.is_relative_to(tmp_path)
        assert result.name == "resume-pending-abc123.json"

    @pytest.mark.parametrize("mod", _HOOK_MODULES)
    def test_traversal_dotdot_returns_none(self, mod, tmp_path):
        # Prevents overwriting ~/.claude/settings.json via crafted session_id
        result = mod._safe_state_path(tmp_path, "", "../../.claude/settings")
        assert result is None

    @pytest.mark.parametrize("mod", _HOOK_MODULES)
    def test_traversal_dotdot_with_prefix_returns_none(self, mod, tmp_path):
        # "resume-pending-../../evil" resolves to <cache_dir>/evil — still inside,
        # because the prefix contributes a pseudo-directory level that absorbs one "..".
        # A genuine escape requires 3 levels of "../".  Verify the guard catches it.
        result = mod._safe_state_path(tmp_path, "resume-pending-", "../../../evil")
        assert result is None

    @pytest.mark.parametrize("mod", _HOOK_MODULES)
    def test_two_dotdot_with_prefix_stays_inside(self, mod, tmp_path):
        # "resume-pending-../../evil" lands at <cache_dir>/evil — guard correctly
        # allows it because the resolved path is still within cache_dir.
        result = mod._safe_state_path(tmp_path, "resume-pending-", "../../evil")
        assert result is not None
        assert result.is_relative_to(tmp_path)

    @pytest.mark.parametrize("mod", _HOOK_MODULES)
    def test_absolute_component_returns_none(self, mod, tmp_path):
        # On POSIX, joining an absolute string replaces the base path entirely
        result = mod._safe_state_path(tmp_path, "", "/etc/passwd")
        assert result is None

    @pytest.mark.parametrize("mod", _HOOK_MODULES)
    def test_null_byte_in_session_id_returns_none(self, mod, tmp_path):
        # Null bytes in filenames are rejected by the OS; resolve() raises on some
        # platforms and produces an out-of-bounds path on others.  Either way,
        # the result must be None or a valid in-bounds path — never a path outside.
        try:
            result = mod._safe_state_path(tmp_path, "", "abc\x00../../evil")
        except (ValueError, OSError):
            return  # raising is also an acceptable guard
        if result is not None:
            assert result.is_relative_to(tmp_path)

    @pytest.mark.parametrize("mod", _HOOK_MODULES)
    def test_empty_session_id_returns_path_inside_cache_dir(self, mod, tmp_path):
        # Empty session_id is weird but harmless — cache_dir/".json" stays inside.
        result = mod._safe_state_path(tmp_path, "", "")
        assert result is not None
        assert result.is_relative_to(tmp_path)

    @pytest.mark.parametrize("mod", _HOOK_MODULES)
    def test_returned_path_has_json_suffix(self, mod, tmp_path):
        result = mod._safe_state_path(tmp_path, "pfx-", "sess99")
        assert result is not None
        assert result.suffix == ".json"


# ---------------------------------------------------------------------------
# get_cached_tokens — transcript bounds check in cache-resume-detect.py
# ---------------------------------------------------------------------------

class TestGetCachedTokens:
    """get_cached_tokens must refuse to read files outside ~/.claude.

    If the bounds check is bypassed, a malicious transcript_path could read
    arbitrary files on disk (e.g. SSH keys) by supplying an attacker-controlled
    path as the transcript location.
    """

    def _write_transcript(self, path: Path, cache_creation: int, cache_read: int) -> None:
        """Write a minimal JSONL transcript with one assistant turn."""
        entry = {
            "message": {
                "role": "assistant",
                "usage": {
                    "cache_creation_input_tokens": cache_creation,
                    "cache_read_input_tokens": cache_read,
                },
            }
        }
        path.write_text(json.dumps(entry) + "\n")

    def test_path_inside_claude_dir_returns_tokens(self, tmp_path, monkeypatch):
        # Redirect _CLAUDE_DIR so we don't depend on the real ~/.claude existing
        claude_dir = tmp_path / ".claude"
        claude_dir.mkdir()
        monkeypatch.setattr(_resume_detect, "_CLAUDE_DIR", claude_dir)

        transcript = claude_dir / "projects" / "session.jsonl"
        transcript.parent.mkdir(parents=True)
        self._write_transcript(transcript, cache_creation=5000, cache_read=1000)

        result = _resume_detect.get_cached_tokens(str(transcript))
        assert result == 6000  # cache_creation + cache_read

    def test_path_outside_claude_dir_returns_zero(self, tmp_path, monkeypatch):
        # Prevents reading /tmp/evil.jsonl (or any attacker-supplied path)
        claude_dir = tmp_path / ".claude"
        claude_dir.mkdir()
        monkeypatch.setattr(_resume_detect, "_CLAUDE_DIR", claude_dir)

        evil = tmp_path / "evil.jsonl"
        self._write_transcript(evil, cache_creation=9999, cache_read=9999)

        result = _resume_detect.get_cached_tokens(str(evil))
        assert result == 0

    def test_symlink_escaping_claude_dir_returns_zero(self, tmp_path, monkeypatch):
        # Symlink inside ~/.claude pointing outside must not be followed
        claude_dir = tmp_path / ".claude"
        claude_dir.mkdir()
        outside = tmp_path / "real_data.jsonl"
        self._write_transcript(outside, cache_creation=8888, cache_read=0)

        link = claude_dir / "escape_link.jsonl"
        link.symlink_to(outside)
        monkeypatch.setattr(_resume_detect, "_CLAUDE_DIR", claude_dir)

        result = _resume_detect.get_cached_tokens(str(link))
        assert result == 0

    def test_nonexistent_path_inside_claude_dir_returns_zero(self, tmp_path, monkeypatch):
        claude_dir = tmp_path / ".claude"
        claude_dir.mkdir()
        monkeypatch.setattr(_resume_detect, "_CLAUDE_DIR", claude_dir)

        missing = claude_dir / "no-such-session.jsonl"
        result = _resume_detect.get_cached_tokens(str(missing))
        assert result == 0

    def test_empty_transcript_returns_zero(self, tmp_path, monkeypatch):
        claude_dir = tmp_path / ".claude"
        claude_dir.mkdir()
        monkeypatch.setattr(_resume_detect, "_CLAUDE_DIR", claude_dir)

        transcript = claude_dir / "empty.jsonl"
        transcript.write_text("")
        result = _resume_detect.get_cached_tokens(str(transcript))
        assert result == 0

    def test_returns_last_assistant_turn_tokens(self, tmp_path, monkeypatch):
        # get_cached_tokens should track the latest value, not the first
        claude_dir = tmp_path / ".claude"
        claude_dir.mkdir()
        monkeypatch.setattr(_resume_detect, "_CLAUDE_DIR", claude_dir)

        transcript = claude_dir / "multi_turn.jsonl"
        lines = [
            json.dumps({"message": {"role": "assistant", "usage": {
                "cache_creation_input_tokens": 100, "cache_read_input_tokens": 0
            }}}),
            json.dumps({"message": {"role": "user", "content": "hello"}}),
            json.dumps({"message": {"role": "assistant", "usage": {
                "cache_creation_input_tokens": 0, "cache_read_input_tokens": 4000
            }}}),
        ]
        transcript.write_text("\n".join(lines) + "\n")

        result = _resume_detect.get_cached_tokens(str(transcript))
        assert result == 4000  # last assistant turn, not first


# ---------------------------------------------------------------------------
# _bust_overhead — incremental cache-bust cost function
# ---------------------------------------------------------------------------

_SONNET_PRICING = _get_pricing("claude-sonnet")
# Rates: input=3.0, output=15.0, cache_write_5m=3.75, cache_write_1h=6.0, cache_read=0.30


class TestBustOverhead:
    """_bust_overhead measures only the extra cost of re-creating a cache miss
    (write rate minus read rate), not the full turn cost.

    If the formula is wrong, cache-bust impact reports will over- or under-count
    the cost savings from compacting — leading users to ignore or over-react to
    cache TTL warnings.
    """

    def test_all_zeros_returns_zero(self):
        result = _bust_overhead(0, 0, 0, _SONNET_PRICING)
        assert result == 0.0

    def test_only_ephem_5m_tokens(self):
        # 1M tokens re-created at 5m tier: delta = (3.75 - 0.30) / 1M per token
        result = _bust_overhead(1_000_000, 1_000_000, 0, _SONNET_PRICING)
        expected = (1_000_000 * (3.75 - 0.30)) / 1_000_000
        assert abs(result - expected) < 1e-9

    def test_only_ephem_1h_tokens(self):
        # 1M tokens re-created at 1h tier: delta = (6.0 - 0.30) / 1M per token
        result = _bust_overhead(1_000_000, 0, 1_000_000, _SONNET_PRICING)
        expected = (1_000_000 * (6.0 - 0.30)) / 1_000_000
        assert abs(result - expected) < 1e-9

    def test_mixed_5m_and_1h_tokens(self):
        ephem_5m = 600_000
        ephem_1h = 400_000
        cache_creation = ephem_5m + ephem_1h
        result = _bust_overhead(cache_creation, ephem_5m, ephem_1h, _SONNET_PRICING)
        expected = (
            ephem_5m * (3.75 - 0.30) + ephem_1h * (6.0 - 0.30)
        ) / 1_000_000
        assert abs(result - expected) < 1e-9

    def test_unclassified_tokens_attributed_to_5m_tier(self):
        # cache_creation > ephem_5m + ephem_1h — remainder goes to 5m (cheaper)
        cache_creation = 1_000_000
        ephem_5m = 400_000
        ephem_1h = 200_000
        unclassified = cache_creation - ephem_5m - ephem_1h  # 400_000
        result = _bust_overhead(cache_creation, ephem_5m, ephem_1h, _SONNET_PRICING)
        expected = (
            (ephem_5m + unclassified) * (3.75 - 0.30)
            + ephem_1h * (6.0 - 0.30)
        ) / 1_000_000
        assert abs(result - expected) < 1e-9

    def test_bust_overhead_never_exceeds_turn_cost(self):
        # _bust_overhead measures only the re-creation premium; _turn_cost
        # includes input/output/read charges too — so bust <= turn for any input.
        for input_tok, output_tok, cache_read, cache_creation, ep5m, ep1h in [
            (1000, 500, 2000, 5000, 3000, 2000),
            (0, 0, 0, 1_000_000, 1_000_000, 0),
            (500_000, 100_000, 800_000, 300_000, 150_000, 150_000),
            (0, 0, 0, 0, 0, 0),
        ]:
            bust = _bust_overhead(cache_creation, ep5m, ep1h, _SONNET_PRICING)
            full = _turn_cost(
                input_tok, output_tok, cache_read, cache_creation,
                ep5m, ep1h, _SONNET_PRICING
            )
            assert bust <= full + 1e-12, (
                f"bust_overhead ({bust}) exceeded turn_cost ({full}) for "
                f"cache_creation={cache_creation} ep5m={ep5m} ep1h={ep1h}"
            )

    def test_bust_overhead_is_zero_when_no_cache_creation(self):
        # If there's nothing to re-create, overhead is zero regardless of other charges
        result = _bust_overhead(0, 0, 0, _SONNET_PRICING)
        assert result == 0.0

    def test_haiku_pricing_produces_lower_cost_than_sonnet(self):
        haiku = _get_pricing("claude-haiku")
        ep5m = 1_000_000
        sonnet_cost = _bust_overhead(ep5m, ep5m, 0, _SONNET_PRICING)
        haiku_cost = _bust_overhead(ep5m, ep5m, 0, haiku)
        assert haiku_cost < sonnet_cost

    def test_per_tier_bucketing_is_independent(self):
        """Bust bucket attribution uses separate if-blocks, not elif.

        When a turn has both 5m and 1h tokens and gap > 1h, both tiers expired.
        The fix changes elif → two independent ifs so both buckets accumulate.
        """
        ep5m = 600_000
        ep1h = 400_000
        gap_over_1h = 4_000_000  # > 3,600,000 ms — both tiers expired
        pricing = _SONNET_PRICING

        cost_5m = _bust_overhead(ep5m, ep5m, 0, pricing)
        cost_1h = _bust_overhead(ep1h, 0, ep1h, pricing)

        # Simulate what the aggregation loop now does (two independent ifs)
        bucket: dict = {"busts_5m": 0, "busts_1h": 0, "cost_5m": 0.0, "cost_1h": 0.0}
        if ep5m and gap_over_1h > 300_000:
            bucket["busts_5m"] += 1
            bucket["cost_5m"] += cost_5m
        if ep1h and gap_over_1h > 3_600_000:
            bucket["busts_1h"] += 1
            bucket["cost_1h"] += cost_1h

        assert bucket["busts_5m"] == 1
        assert bucket["busts_1h"] == 1
        assert bucket["cost_5m"] == pytest.approx(cost_5m)
        assert bucket["cost_1h"] == pytest.approx(cost_1h)
        # Total cost equals sum of per-tier costs
        assert bucket["cost_5m"] + bucket["cost_1h"] == pytest.approx(cost_5m + cost_1h)

    def test_5m_only_bust_does_not_touch_1h_bucket(self):
        """A gap between 5min and 1h only busts the 5m tier."""
        ep5m = 600_000
        ep1h = 400_000
        gap_5m_only = 600_000  # 10 min — only 5m expired

        bucket: dict = {"busts_5m": 0, "busts_1h": 0, "cost_5m": 0.0, "cost_1h": 0.0}
        if ep5m and gap_5m_only > 300_000:
            bucket["busts_5m"] += 1
            bucket["cost_5m"] += _bust_overhead(ep5m, ep5m, 0, _SONNET_PRICING)
        if ep1h and gap_5m_only > 3_600_000:
            bucket["busts_1h"] += 1
            bucket["cost_1h"] += _bust_overhead(ep1h, 0, ep1h, _SONNET_PRICING)

        assert bucket["busts_5m"] == 1
        assert bucket["busts_1h"] == 0
        assert bucket["cost_1h"] == 0.0


# ---------------------------------------------------------------------------
# _safe_state_path — exception resilience (OSError / RuntimeError)
# ---------------------------------------------------------------------------

class TestSafeStatePathExceptions:
    """_safe_state_path must return None (not crash) for any malformed input."""

    @pytest.mark.parametrize("mod", _HOOK_MODULES)
    def test_returns_none_not_raises_on_oserror(self, mod, tmp_path, monkeypatch):
        """Simulate Path.resolve() raising OSError (e.g., path too long)."""
        import pathlib

        def bad_resolve(self, strict=False):
            raise OSError("simulated OS-level resolve failure")

        monkeypatch.setattr(pathlib.Path, "resolve", bad_resolve)
        result = mod._safe_state_path(tmp_path, "", "abc123")
        assert result is None

    @pytest.mark.parametrize("mod", _HOOK_MODULES)
    def test_returns_none_not_raises_on_runtime_error(self, mod, tmp_path, monkeypatch):
        """Simulate Path.resolve() raising RuntimeError (e.g., infinite symlink loop)."""
        import pathlib

        def bad_resolve(self, strict=False):
            raise RuntimeError("simulated infinite symlink loop")

        monkeypatch.setattr(pathlib.Path, "resolve", bad_resolve)
        result = mod._safe_state_path(tmp_path, "", "abc123")
        assert result is None


# ---------------------------------------------------------------------------
# get_cached_tokens — None and encoding edge cases
# ---------------------------------------------------------------------------

class TestGetCachedTokensEdgeCases:
    def test_none_transcript_path_returns_zero(self, monkeypatch, tmp_path):
        """get_cached_tokens(None) must not raise TypeError."""
        claude_dir = tmp_path / ".claude"
        claude_dir.mkdir()
        monkeypatch.setattr(_resume_detect, "_CLAUDE_DIR", claude_dir)
        # None passed as transcript_path — cast to str("") then bounds check returns 0
        result = _resume_detect.get_cached_tokens(None)  # type: ignore[arg-type]
        assert result == 0

    def test_utf8_transcript_is_read_correctly(self, tmp_path, monkeypatch):
        """Transcripts with non-ASCII content must not crash the reader."""
        claude_dir = tmp_path / ".claude"
        claude_dir.mkdir()
        monkeypatch.setattr(_resume_detect, "_CLAUDE_DIR", claude_dir)

        transcript = claude_dir / "utf8_session.jsonl"
        entry = {
            "message": {
                "role": "assistant",
                "usage": {
                    "cache_creation_input_tokens": 100,
                    "cache_read_input_tokens": 50,
                },
                "content": "こんにちは世界",  # non-ASCII
            }
        }
        transcript.write_text(json.dumps(entry) + "\n", encoding="utf-8")
        result = _resume_detect.get_cached_tokens(str(transcript))
        assert result == 150