settled-reach/tooling/domains/assets/endpoints.py

"""Where the generators live, and how a call to one fails.

Two jobs, both shared by every connector in the domain:

1. **Endpoints and keys.** Base URLs come from `config.json` beside this file
   (moved deliberately from tooling/db/config.json in T-1290). Keys come from the
   environment only: config.json is tracked, so a config fallback for a paid
   API key is how one gets committed.

2. **Failure classification.** Network failure is the NORMAL case here, not an
   exception. Stable Audio and Trellis are kept stopped because VRAM on
   tower-of-joy is scarce (system-admin-danoontje docs/topology.md, D-17), so
   "connection refused" means *switched off*, and the remedy is to get it
   turned on — never to restart it blindly, which takes VRAM from whatever is
   running. That is a different answer from "the service rejected what you
   sent", and a caller that cannot tell the two apart wastes a round trip
   either way. `call()` maps every urllib failure to one of the two.

`ensure_venv()` (formerly in tooling/db/common.py) is gone: it re-exec'd the
script under .venv/bin/python via os.execv. Under reach the dependencies are
declared by the package itself, and an execv carrying reach's argv into another
interpreter would relaunch something that is not the command at all.
"""

from __future__ import annotations

import json
import os
import urllib.error
import urllib.request
from pathlib import Path
from typing import Any

from tooling.core.errors import ReachError

CONFIG_PATH = Path(__file__).resolve().parent / "config.json"

TOPOLOGY = "/var/mnt/data/projects/system-admin-danoontje/docs/topology.md"

# service key -> (display name, config key, fallback URL)
SERVICES: dict[str, tuple[str, str, str]] = {
    "audio": ("Stable Audio Open", "stable_audio_url", "http://tower-of-joy:11500"),
    "trellis": ("Trellis", "trellis_url", "http://tower-of-joy:11510"),
}


def load_config() -> dict:
    """The tracked endpoint configuration (URLs only, never secrets)."""
    with open(CONFIG_PATH) as f:
        return json.load(f)


def get_base_url(key: str, default: str) -> str:
    """Resolve a service base URL from config.json, with a fallback default."""
    return load_config().get(key, default)


def base_url(service: str) -> str:
    """The configured base URL for one of SERVICES."""
    _name, config_key, default = SERVICES[service]
    return get_base_url(config_key, default)


def get_api_key(env_var: str) -> str:
    """An API key from the environment — environment-only, by design."""
    key = os.environ.get(env_var)
    if key:
        return key
    raise ReachError(
        f"{env_var} is not set",
        fix=(
            f"export {env_var}, or add it to the machine-local "
            ".claude/settings.local.json env block (untracked). Never put keys in "
            "tooling/domains/assets/config.json — it is tracked."
        ),
    )


def unreachable(service: str, url: str, detail: str) -> ReachError:
    """The service did not answer at all.

    For the tower-of-joy generators that is their normal resting state. For
    anything else (the Gemini API) it means the network, not the service.
    """
    if service not in SERVICES:
        return ReachError(
            f"{service} is not reachable at {url} ({detail})",
            fix="check this machine's network connection, then re-run",
        )
    name = SERVICES[service][0]
    return ReachError(
        f"{name} is not reachable at {url} ({detail}). It is kept switched off to "
        "save VRAM on tower-of-joy, so this usually means OFF, not broken.",
        fix=(
            f"ask for {name} to be turned on (something else may need to stop first "
            f"to free VRAM — {TOPOLOGY}); do not restart it blindly. "
            f"Then: reach assets {service} health"
        ),
    )


def rejected(name: str, what: str, code: int, body: str) -> ReachError:
    """The service answered, and said no — the request is what needs changing."""
    snippet = body.strip()[:500]
    if code in (401, 403):
        fix = "the service is up but refused the credentials — check the API key"
    elif code == 429:
        fix = "rate-limited or out of quota — wait, or check the account's quota"
    else:
        fix = "the service is up — check the arguments and inputs, then re-run"
    return ReachError(
        f"{name} rejected {what} (HTTP {code})" + (f": {snippet}" if snippet else ""),
        fix=fix,
    )


def call(
    request: urllib.request.Request,
    *,
    service: str,
    what: str,
    timeout: float,
) -> bytes:
    """Perform one HTTP request, classifying any failure. Returns the body."""
    name = SERVICES[service][0] if service in SERVICES else service
    try:
        with urllib.request.urlopen(request, timeout=timeout) as resp:
            return resp.read()
    except urllib.error.HTTPError as exc:
        raise rejected(name, what, exc.code, exc.read().decode("utf-8", errors="replace")) from exc
    except (urllib.error.URLError, TimeoutError, ConnectionError) as exc:
        reason = getattr(exc, "reason", exc)
        raise unreachable(service, _origin(request), str(reason)) from exc


def call_json(request: urllib.request.Request, *, service: str, what: str, timeout: float) -> Any:
    """`call()`, then parse JSON — an unparseable body is the service's fault, named as such."""
    body = call(request, service=service, what=what, timeout=timeout)
    try:
        return json.loads(body)
    except json.JSONDecodeError as exc:
        raise ReachError(
            f"{what}: the service answered with something that is not JSON: {body[:200]!r}",
            fix="the service may be starting up or misconfigured — check its logs on tower-of-joy",
        ) from exc


def _origin(request: urllib.request.Request) -> str:
    """scheme://host:port of a request — the part that identifies the box."""
    parts = request.full_url.split("/")
    return "/".join(parts[:3])