user_errors.py

Go to the documentation of this file.

# Copyright (c) 2025-present Aquin Labs Private Limited. All Rights Reserved.
"""User-facing CLI errors — no raw HTTP tracebacks unless AQUIN_DEBUG=1."""
from __future__ import annotations
 
import json
import os
import re
import sys
from typing import Any
 
_URL_RE = re.compile(r"https?://[^\s'\"<>]+")
 
 
def debug_enabled() -> bool:
    return os.environ.get("AQUIN_DEBUG", "").strip().lower() in ("1", "true", "yes")
 
 
def log_debug(label: str, detail: Any) -> None:
    if debug_enabled():
        print(f"[debug/{label}] {detail}", file=sys.stderr)
 
 
def _scrub_urls(text: str) -> str:
    return _URL_RE.sub("Aquin cloud", text)
 
 
def sanitize_error_string(text: str) -> str:
    """Turn raw API/library strings into short user-facing messages."""
    if not text:
        return "Something went wrong. Try again."
 
    s = _scrub_urls(str(text).strip())
    low = s.lower()
 
    if s.startswith(("Could not ", "Cannot ", "No API key", "Not logged in", "Run:", "Usage:")):
        return s
    if "run: aquin login" in low or "aquin.app/profile" in low:
        return s
 
    if "401" in s or "unauthorized" in low:
        return "Authentication failed. Set HF_TOKEN / OPENAI_API_KEY if this call needs cloud credentials."
    if "token revoked" in low or "revoked_at" in low:
        return "Cloud token rejected. Use a local path or set the required env token."
    if "403" in s or "forbidden" in low:
        return "Access denied for this resource."
    if "404" in s and ("sae" in low or "not found" in low):
        return (
            "SAE not found in the Aquin library. Try aquin load sae <model-l{n}> "
            "or aquin load sae --path <file.pt>\n"
            "Some models (gpt2-small, pythia-70m) pull SAE weights from Hugging Face instead."
        )
    if "404" in s or "not found" in low:
        return "Not found."
    if "503" in s or "not configured on" in low:
        return "This feature is temporarily unavailable. Try again later."
    if any(x in low for x in ("connection", "timeout", "network", "could not reach", "name resolution")):
        return "Could not reach Aquin. Check your internet connection."
    if isinstance(text, json.JSONDecodeError) or "jsondecode" in low or "extra data" in low:
        return "Could not read the server response. Try again."
    if "hugging face access via aquin proxy" in low:
        return s.split("\n")[0] if "\n" in s else s
    if any(x in low for x in ("zip archive", "central directory", "pytorchstreamreader")):
        return (
            "SAE checkpoint file is corrupted or incomplete. "
            "Delete it under ~/.aquin/sae/ and run: aquin load sae <model>-l<layer>"
        )
    if any(x in low for x in ("out of memory", "cuda out of memory")):
        return s if len(s) <= 500 else s[:500] + "…"
    if any(x in low for x in ("huggingface", "gated repo", "hf hub", "gated")):
        return (
            "Could not download from Hugging Face. "
            "Set HF_TOKEN for gated models, or use a public checkpoint."
        )
    if "client error" in low or "httperror" in low or "for url" in low:
        return "Request failed. Check network / credentials and try aquin status."
    if " failed:" in low:
        _, tail = re.split(r"\bfailed:\s*", s, maxsplit=1, flags=re.I)
        if tail:
            return sanitize_error_string(tail.strip())
 
    if len(s) > 180 and not debug_enabled():
        return "Something went wrong. Try again. (Set AQUIN_DEBUG=1 for details)"
    return s
 
 
def friendly_message(exc: BaseException) -> str:
    if isinstance(exc, KeyboardInterrupt):
        raise exc
 
    try:
        import httpx
 
        if isinstance(exc, httpx.HTTPStatusError):
            code = exc.response.status_code if exc.response is not None else 0
            if code in (401, 403):
                return "Authentication failed. Set the required env token (e.g. HF_TOKEN)."
            if code == 404:
                return "Not found."
            if code >= 500:
                return "Remote service is having trouble. Try again shortly."
            return sanitize_error_string(str(exc))
    except ImportError:
        pass
 
    try:
        import requests
        from requests.exceptions import HTTPError, ConnectionError, Timeout
 
        if isinstance(exc, HTTPError):
            code = exc.response.status_code if exc.response is not None else 0
            if code in (401, 403):
                return "Authentication failed. Set the required env token (e.g. HF_TOKEN)."
            if code == 404:
                return sanitize_error_string(str(exc))
            if code >= 500:
                return "Remote service is having trouble. Try again shortly."
            return sanitize_error_string(str(exc))
        if isinstance(exc, (ConnectionError, Timeout)):
            return "Could not reach remote service. Check your internet connection."
    except ImportError:
        pass
 
    if isinstance(exc, FileNotFoundError):
        msg = str(exc).strip()
        if msg and (
            "sae" in msg.lower()
            or "layer" in msg.lower()
            or "dictionary" in msg.lower()
            or "\n" in msg
        ):
            return msg
        return "File not found. Check the path and try again."
    if isinstance(exc, json.JSONDecodeError):
        return "Could not read JSON input. Check the file format."
    if isinstance(exc, NotImplementedError):
        msg = str(exc).strip()
        if msg and ("not yet ported" in msg.lower() or "not available in" in msg.lower()):
            return msg
        if msg and debug_enabled():
            return f"This command is not available yet. ({msg})"
        if msg and len(msg) < 120:
            return msg
        return "This command is not available yet."
    if isinstance(exc, ValueError):
        msg = str(exc).strip()
        if msg and not msg.startswith("HTTP"):
            if "corrupt" in msg.lower() or "re-download" in msg.lower() or len(msg) < 400:
                return msg
    if isinstance(exc, RuntimeError):
        from aquin.compute.torch_io import is_corrupt_pytorch_zip
 
        msg = str(exc).strip()
        if msg and is_corrupt_pytorch_zip(exc):
            return (
                "SAE checkpoint file is corrupted or incomplete. "
                "Delete it under ~/.aquin/sae/ and run: aquin load sae <model>-l<layer>"
            )
 
    msg = str(exc).strip()
    if msg:
        return sanitize_error_string(msg)
    return "Something went wrong. Try again."
 
 
def die(
    message: str | None = None,
    *,
    exc: BaseException | None = None,
    code: int = 1,
    label: str = "aquin",
) -> None:
    if exc is not None:
        log_debug(label, exc)
        text = message or friendly_message(exc)
    else:
        text = message or "Something went wrong. Try again."
    print(text, file=sys.stderr)
    raise SystemExit(code)
 
 
def exit_on(exc: BaseException, *, label: str = "aquin", code: int = 1) -> None:
    die(exc=exc, label=label, code=code)