portfolio-optimiser-claude/src/portfolio_optimiser_claude/preflight.py

"""SDK/API preflight (method-spec §4.1/§10 analog; S4.1; paritetsrad 20; K7).

Everything that CAN be validated WITHOUT a model call is validated HERE, before
the operator pays for one. The preflight is the BOUNDARY: it never calls
``query()``, never validates the credential VALUE online (only that one is SET
and is not an obvious placeholder form), never touches the network. It returns a
list of structured ``Refusal``s — one actionable message per deficiency; an
empty list means "clear to run". The CLI exits non-zero on any refusal so a
broken config stops here, cheaply, instead of on the first billed call.

Four offline checks (for a given backend profile, ``anthropic`` today):

* **credential** — a credential is available (``ANTHROPIC_API_KEY``, or the
  bundled CLI's own ``CLAUDE_CODE_OAUTH_TOKEN``) and is not a placeholder form.
  The value itself is NEVER checked online — a bad key surfaces on the first
  call, never here (§1 honesty: the preflight claims no more than it proves).
* **model_map** — the requested profile exists, and every id it resolves to
  (THROUGH ``resolve_model``, so the ``default`` fall-through is covered) is a
  real id, not a placeholder left in the config.
* **sdk** — ``claude_agent_sdk`` imports (a missing install is a structured
  refusal naming ``uv sync``, never an ImportError), the run-path symbols exist,
  and the bundled Claude Code CLI is present on disk — located offline via the
  SDK package's own files, mirroring the SDK's ``_find_cli`` order (bundled
  first, then a ``claude`` on PATH). Verified against installed 0.2.139 by
  reading its ``_find_cli`` (bundled at line 250, PATH at 256) — the order is
  unchanged from 0.2.120.
* **termination** — the §8 stop contract constructs with positive round/token
  caps and the per-call USD cap is positive (the budget belt is SET).

Importing the SDK to prove it imports is offline-safe (no key, no socket) — the
whole suite already imports the client. The no-network grep-guard
(``tests/test_preflight.py``) proves this module carries no HTTP path of its own.

Run:  uv run python -m portfolio_optimiser_claude.preflight [--profile anthropic]
      [--max-rounds N] [--max-tokens N] [--max-budget-usd-per-call F]
"""

from __future__ import annotations

import argparse
import importlib
import os
import platform
import shutil
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Mapping

from pydantic import ValidationError

from portfolio_optimiser_claude.contracts import (
    ModelMapContract,
    TerminationContract,
    _bundled_model_map,
    resolve_model,
)

# Tokens that never appear in a real Anthropic key or Claude model id, but do in
# the placeholders operators leave behind. Matched case-insensitively as a
# SUBSTRING so ``sk-ant-YOUR-KEY`` and ``<model>`` are both caught. Kept
# conservative to avoid ever refusing a genuine value (§1: no false alarms).
_PLACEHOLDER_TOKENS: tuple[str, ...] = (
    "placeholder",
    "your-",
    "your_",
    "yourkey",
    "changeme",
    "change-me",
    "change_me",
    "replace",
    "example",
    "todo",
    "fixme",
    "dummy",
    "xxxx",
    "<",
    ">",
    "...",
)

# A probe role guaranteed absent from any mapping, used to exercise the
# resolve_model default fall-through (the id an UNMAPPED role would receive).
_FALLTHROUGH_PROBE = "__preflight_fallthrough_probe__"

# The run-path symbols sdk_client.py depends on; their presence is what
# "the SDK imports" must actually mean (a bare module is not enough).
_REQUIRED_SDK_SYMBOLS: tuple[str, ...] = ("query", "ClaudeAgentOptions")


@dataclass(frozen=True)
class Refusal:
    """One preflight deficiency: which check tripped + an actionable message."""

    check: str
    detail: str


def _looks_like_placeholder(value: str) -> bool:
    """True for an empty/whitespace value or one carrying a placeholder token.

    Shared by the credential and model-id checks so a stub left in either place
    is caught by the same conservative rule (never validates a value online)."""
    low = value.strip().lower()
    if not low:
        return True
    return any(token in low for token in _PLACEHOLDER_TOKENS)


def _check_credentials(env: Mapping[str, str]) -> list[Refusal]:
    """A usable credential is present (§10) — the VALUE is never checked online.

    ``ANTHROPIC_API_KEY`` is the named contract; the bundled CLI's own
    ``CLAUDE_CODE_OAUTH_TOKEN`` also satisfies it (run_s10 relies on the CLI's
    credentials when no key is exported — refusing that would be a false alarm).
    A credential set to a placeholder form is refused; NEITHER set is refused.
    """
    key = env.get("ANTHROPIC_API_KEY", "").strip()
    oauth = env.get("CLAUDE_CODE_OAUTH_TOKEN", "").strip()
    if key:
        if _looks_like_placeholder(key):
            return [
                Refusal(
                    "credential",
                    "ANTHROPIC_API_KEY is set to a placeholder-form value — export a "
                    "real key (its value is never validated online, only that one is set).",
                )
            ]
        return []
    if oauth and not _looks_like_placeholder(oauth):
        return []  # the bundled CLI authenticates via its OAuth token
    return [
        Refusal(
            "credential",
            "no API credential: export ANTHROPIC_API_KEY (or authenticate the bundled "
            "CLI, which sets CLAUDE_CODE_OAUTH_TOKEN). The value is never checked online "
            "— only that a credential is present.",
        )
    ]


def _check_model_map(profile: str, *, model_map: ModelMapContract | None = None) -> list[Refusal]:
    """The requested profile exists and every id it RESOLVES to is real (§10).

    Placeholder refusal runs THROUGH ``resolve_model`` (not a raw dict read), so
    the ``default`` fall-through an unmapped role receives is checked too. An
    unknown profile — which ``resolve_model`` would raise on — becomes a
    structured refusal, never a stack trace before spend.
    """
    resolved = ModelMapContract(**_bundled_model_map()) if model_map is None else model_map
    mapping = resolved.profiles.get(profile)
    if mapping is None:
        have = ", ".join(sorted(resolved.profiles)) or "(none)"
        return [
            Refusal(
                "model_map",
                f"backend profile '{profile}' is not in model_map (have: {have}).",
            )
        ]
    refusals: list[Refusal] = []
    # Every configured role PLUS the fall-through the default serves — all via
    # resolve_model, deduped so one placeholder id is reported once.
    for role in sorted(mapping) + [_FALLTHROUGH_PROBE]:
        model_id = resolve_model(resolved, role, profile=profile)
        if _looks_like_placeholder(model_id):
            refusals.append(
                Refusal(
                    "model_map",
                    f"model id '{model_id}' (profile '{profile}') is a placeholder form — "
                    "set a real Claude model id (verified against platform.claude.com/docs).",
                )
            )
            break  # a placeholder id is a single config fault; one message suffices
    return refusals


def _locate_cli(sdk: Any) -> Path | None:
    """The CLI the SDK would spawn, resolved OFFLINE via the package's files.

    Mirrors the SDK's ``_find_cli`` order: the bundled binary shipped in the
    wheel first, then a ``claude`` on PATH. No network, no subprocess — just a
    file-existence probe, so the preflight can prove there IS a CLI to run.
    """
    cli_name = "claude.exe" if platform.system() == "Windows" else "claude"
    bundled = Path(sdk.__file__).parent / "_bundled" / cli_name
    if bundled.is_file():
        return bundled
    found = shutil.which("claude")
    return Path(found) if found else None


def _check_sdk() -> list[Refusal]:
    """The SDK imports, exposes the run-path symbols, and its CLI is present.

    A missing install is a structured refusal naming the fix (never an
    ImportError raised out of the preflight itself). Importing is offline-safe.
    """
    try:
        sdk = importlib.import_module("claude_agent_sdk")
    except ImportError as exc:
        return [
            Refusal(
                "sdk",
                f"claude-agent-sdk is not importable ({exc}); run 'uv sync' to install the "
                "pinned SDK (it bundles the Claude Code CLI — no separate install).",
            )
        ]
    missing = [name for name in _REQUIRED_SDK_SYMBOLS if not hasattr(sdk, name)]
    if missing:
        return [
            Refusal(
                "sdk",
                f"claude-agent-sdk is installed but missing run-path symbols {missing} — "
                "the pin may be out of the verified range (see test_sdk_version_guard).",
            )
        ]
    if _locate_cli(sdk) is None:
        pkg_file = sdk.__file__
        bundled = f"{Path(pkg_file).parent / '_bundled'}" if pkg_file else "the SDK package"
        return [
            Refusal(
                "sdk",
                f"no Claude Code CLI found (neither the SDK's bundled binary under {bundled} "
                "nor a 'claude' on PATH); reinstall claude-agent-sdk or pass "
                "ClaudeAgentOptions(cli_path=...).",
            )
        ]
    return []


def _check_termination(
    max_rounds: int, max_tokens: int, max_budget_usd_per_call: float
) -> list[Refusal]:
    """The §8 stop contract constructs and the per-call USD belt is positive."""
    refusals: list[Refusal] = []
    try:
        TerminationContract(max_rounds=max_rounds, max_tokens=max_tokens)
    except ValidationError as exc:
        refusals.append(
            Refusal(
                "termination",
                f"stop contract invalid — max_rounds/max_tokens must be positive (§8): {exc}",
            )
        )
    if max_budget_usd_per_call <= 0:
        refusals.append(
            Refusal(
                "termination",
                f"max_budget_usd_per_call must be positive, got {max_budget_usd_per_call} "
                "(the per-call USD belt must be SET before any spend).",
            )
        )
    return refusals


def run_preflight(
    *,
    profile: str = "anthropic",
    env: Mapping[str, str] | None = None,
    max_rounds: int = 12,
    max_tokens: int = 150_000,
    max_budget_usd_per_call: float = 0.25,
) -> list[Refusal]:
    """Run every offline check; the refusals are the union (empty = clear to run).

    No check makes a model call or a network request — this is the boundary the
    operator crosses BEFORE spending. ``env`` defaults to ``os.environ``.
    """
    environ = os.environ if env is None else env
    return [
        *_check_credentials(environ),
        *_check_model_map(profile),
        *_check_sdk(),
        *_check_termination(max_rounds, max_tokens, max_budget_usd_per_call),
    ]


def main(argv: list[str] | None = None) -> int:
    """The thin CLI: run the offline preflight → print refusals → exit code.

    Exit 0: clear to run (no refusal). Exit 1: one or more refusals — printed
    with their fix, and a line stating no model call was made (the preflight IS
    the boundary; a green preflight never implies a validated key or a run).
    """
    parser = argparse.ArgumentParser(
        description=(
            "Validate everything that CAN be checked WITHOUT a model call, BEFORE any "
            "spend — credential presence, model_map, SDK + bundled CLI, stop/budget "
            "contract (offline; the preflight IS the boundary, it never calls the API)."
        )
    )
    parser.add_argument("--profile", type=str, default="anthropic")
    parser.add_argument("--max-rounds", type=int, default=12)
    parser.add_argument("--max-tokens", type=int, default=150_000)
    parser.add_argument("--max-budget-usd-per-call", type=float, default=0.25)
    args = parser.parse_args(argv)

    refusals = run_preflight(
        profile=args.profile,
        max_rounds=args.max_rounds,
        max_tokens=args.max_tokens,
        max_budget_usd_per_call=args.max_budget_usd_per_call,
    )
    if not refusals:
        print(
            f"PREFLIGHT OK — clear to run (profile={args.profile}); no model call was "
            "made (the preflight is the boundary, not a validated key or a run)."
        )
        return 0
    print(f"PREFLIGHT FAILED — {len(refusals)} refusal(s), no spend (no model call was made):")
    for refusal in refusals:
        print(f"  [{refusal.check}] {refusal.detail}")
    return 1


if __name__ == "__main__":
    raise SystemExit(main())