Advertisement

← refacto-feature-scout

scout/llm.py

8.7 KB Download file
"""Whose Claude account this run uses — and the guarantee that it is never someone else's.

Feature Scout has no AI account of its own. Every scan and every build is billed to the person
running it. The one rule that makes that true is in `claude_env()`: the environment handed to
the `claude` command is built from an allowed list, never copied wholesale, so an operator's
own settings — mail credentials, deploy tokens, a hosting key — cannot reach a model session
even when they happen to be sitting in the environment.
"""
from __future__ import annotations

import json
import os
import subprocess
import tempfile
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable

from .config import CLOUD, DATA_DIR

# Harmless things a program needs to run at all. Anything not named here (or matched by the
# rules in claude_env) does not reach the model.
BASE_ENV = (
    "PATH", "HOME", "USER", "LOGNAME", "SHELL", "LANG", "LC_ALL", "LC_CTYPE", "TERM", "TMPDIR",
    "TZ",
    # Windows equivalents
    "USERPROFILE", "SystemRoot", "SystemDrive", "ComSpec", "PATHEXT", "APPDATA", "LOCALAPPDATA",
    "TEMP", "TMP", "NUMBER_OF_PROCESSORS", "PROCESSOR_ARCHITECTURE",
    # networks that need a proxy or their own certificates
    "HTTP_PROXY", "HTTPS_PROXY", "NO_PROXY", "http_proxy", "https_proxy", "no_proxy",
    "SSL_CERT_FILE", "SSL_CERT_DIR", "NODE_EXTRA_CA_CERTS", "REQUESTS_CA_BUNDLE",
)

# A nested session must not inherit the parent's markers.
MARKERS = ("CLAUDECODE", "CLAUDE_CODE_ENTRYPOINT")

STATE_PATH = DATA_DIR / "state.json"
CHECK_GOOD_FOR_S = 24 * 60 * 60


class NoLLMAccess(Exception):
    """The person running this hasn't set up their own Claude access yet."""


HOW_TO_FIX = (
    "Feature Scout doesn't come with AI included — it runs on your own Claude account.\n"
    "Either:\n"
    "  1. Open a terminal, type `claude`, and sign in. That uses your Claude subscription.\n"
    "  2. Or get a key from console.anthropic.com and run `feature-scout doctor --set-key`.\n"
    "Nothing will run until one of those works."
)


@dataclass(frozen=True)
class Access:
    kind: str        # "subscription" | "api_key" | "unknown" | "none"
    detail: str      # a sentence a person can read; never the secret itself

    @property
    def usable(self) -> bool:
        return self.kind in ("subscription", "api_key")


def _mask(key: str) -> str:
    return f"ending {key[-4:]}" if len(key) >= 4 else "a short one"


def credentials(env: dict[str, str] | None = None) -> Access:
    """What this computer's environment says about Claude access, without spending anything.

    "unknown" means there may be a signed-in Claude on this machine; only asking it will say.
    """
    env = os.environ if env is None else env
    if env.get("CLAUDE_CODE_OAUTH_TOKEN"):
        return Access("subscription", "your Claude subscription (a sign-in token)")
    remembered = _remembered_check()
    if remembered and remembered.get("kind") == "subscription":
        return Access("subscription", "your Claude login on this computer")
    if env.get("ANTHROPIC_API_KEY"):
        return Access("api_key", f"your Claude API key ({_mask(env['ANTHROPIC_API_KEY'])})")
    if remembered and remembered.get("kind") == "none":
        return Access("none", "no Claude account is set up on this computer")
    return Access("unknown", "a Claude login on this computer, if there is one")


def claude_env(env: dict[str, str] | None = None) -> dict[str, str]:
    """The environment the `claude` command is given. Built up, never copied down."""
    src = dict(os.environ if env is None else env)
    out = {k: src[k] for k in BASE_ENV if k in src}

    # The person's own Claude settings travel; the two nesting markers never do.
    for k, v in src.items():
        if k.startswith("CLAUDE_") and k not in MARKERS:
            out[k] = v

    access = credentials(src)
    keep_key = bool(src.get("ANTHROPIC_API_KEY")) and (
        src.get("SCOUT_USE_API_KEY") == "1"
        or access.kind == "api_key"                     # it is the only access there is
        or (CLOUD and not src.get("CLAUDE_CODE_OAUTH_TOKEN"))
    )
    if keep_key:
        out["ANTHROPIC_API_KEY"] = src["ANTHROPIC_API_KEY"]
    if os.name != "nt":
        out.setdefault("PATH", "/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin")
    return out


# ---- asking Claude whether it is there -------------------------------------

def _read_state() -> dict[str, Any]:
    try:
        return json.loads(STATE_PATH.read_text())
    except (OSError, json.JSONDecodeError):
        return {}


def _write_state(data: dict[str, Any]) -> None:
    try:
        STATE_PATH.parent.mkdir(parents=True, exist_ok=True)
        STATE_PATH.write_text(json.dumps(data, indent=2))
    except OSError:
        pass


def _remembered_check() -> dict[str, Any] | None:
    got = _read_state().get("llm")
    if got and time.time() - float(got.get("checked_at") or 0) < CHECK_GOOD_FOR_S:
        return got
    return None


def _ask_claude(timeout_s: int = 90) -> tuple[bool, str]:
    """One cheap turn, only to find out whether this computer can talk to Claude at all."""
    from .claude_runner import claude_bin

    cmd = [claude_bin(), "-p", "reply with the word ready", "--output-format", "json",
           "--max-turns", "1"]
    try:
        with tempfile.TemporaryDirectory() as tmp:
            proc = subprocess.run(cmd, cwd=tmp, env=claude_env(), capture_output=True,
                                  text=True, timeout=timeout_s)
    except FileNotFoundError:
        return False, ("I can't find the `claude` command on this computer. Install Claude Code "
                       "first: https://claude.com/claude-code")
    except subprocess.TimeoutExpired:
        return False, "Claude took too long to answer, so I couldn't check your account."
    if proc.returncode == 0 and proc.stdout.strip():
        try:
            data = json.loads(proc.stdout)
        except json.JSONDecodeError:
            return True, "Claude answered."
        if not data.get("is_error"):
            return True, "Claude answered."
    said = (proc.stderr + proc.stdout)[-400:].strip().lower()
    if "credit" in said or "billing" in said or "quota" in said:
        return False, "Your Claude account answered, but it's out of credit."
    if "login" in said or "auth" in said or "api key" in said or "unauthorized" in said:
        return False, "Claude is installed but not signed in. Type `claude` and sign in."
    return False, f"Claude couldn't be reached: {said[:200]}" if said else "Claude couldn't be reached."


def check_access(prober: Callable[[], tuple[bool, str]] = _ask_claude,
                 force: bool = False, env: dict[str, str] | None = None) -> Access:
    """What access there really is, asking Claude if the environment doesn't already say.

    The answer is remembered for a day so that every command doesn't pay to ask again.
    """
    known = credentials(env)
    if known.usable and known.kind == "subscription" and not force:
        return known
    if not force:
        remembered = _remembered_check()
        if remembered:
            return Access(remembered["kind"], remembered["detail"])

    ok, detail = prober()
    env_now = os.environ if env is None else env
    if ok:
        kind = "api_key" if (known.kind == "api_key" and not env_now.get("CLAUDE_CODE_OAUTH_TOKEN")) \
            else "subscription"
        access = Access(kind, known.detail if known.usable else "your Claude login on this computer")
    else:
        access = Access("api_key" if known.kind == "api_key" else "none", detail)
    state = _read_state()
    state["llm"] = {"kind": access.kind, "detail": access.detail, "checked_at": time.time()}
    _write_state(state)
    return access


def require_llm_access(prober: Callable[[], tuple[bool, str]] = _ask_claude,
                       env: dict[str, str] | None = None) -> Access:
    """Stop before spending anything if this person has no Claude account of their own."""
    access = check_access(prober, env=env)
    if not access.usable:
        raise NoLLMAccess(f"{access.detail}\n\n{HOW_TO_FIX}")
    return access


def save_api_key(key: str) -> Path:
    """Keep a pasted key on this computer only, readable by nobody else."""
    path = DATA_DIR / "settings.env"
    path.parent.mkdir(parents=True, exist_ok=True)
    lines = [ln for ln in (path.read_text().splitlines() if path.exists() else [])
             if not ln.startswith("ANTHROPIC_API_KEY=")]
    lines.append(f"ANTHROPIC_API_KEY={key.strip()}")
    path.write_text("\n".join(lines) + "\n")
    os.chmod(path, 0o600)
    return path