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

"""The Claude Agent SDK model client (honesty rule §1, §11).

This module is the ONE place the programme touches a real model (S10, D6).
The network path (``query()``) is run-path-only; the offline suite proves the
loop with the scripted stand-in, imports this module WITHOUT touching a key
or the network, and pins the call options (``tests/test_sdk_isolation.py``).
Each ``complete()`` is one bounded, ISOLATED ``query()`` call: no tools, one
turn, a first-class USD cap (``ClaudeAgentOptions.max_budget_usd``) ON TOP of
the §8 token/round meter that the loop already charges, and NO filesystem
settings (``setting_sources=[]``).

Verified against claude-agent-sdk 0.2.110: ``query()`` yields
``AssistantMessage`` (text blocks + real model id) and a closing
``ResultMessage`` (provider-reported ``usage`` + ``total_cost_usd``). A reply
without usage is passed through as ``None`` so the meter fails CLOSED (§8) —
this client never invents a count.
"""

from __future__ import annotations

import asyncio
from typing import Any

from claude_agent_sdk import (
    AssistantMessage,
    ClaudeAgentOptions,
    ResultMessage,
    TextBlock,
    query,
)

from portfolio_optimiser_claude.contracts import ModelMapContract, resolve_model
from portfolio_optimiser_claude.loop import ModelReply

# The provider-reported usage fields that make up the TOTAL token count (§8).
_USAGE_TOKEN_FIELDS = (
    "input_tokens",
    "output_tokens",
    "cache_creation_input_tokens",
    "cache_read_input_tokens",
)


def build_call_options(model_id: str, *, max_budget_usd: float) -> ClaudeAgentOptions:
    """One bounded, ISOLATED completion call (§8 + S10 post-mortem).

    ``setting_sources=[]`` is the SDK's documented isolation mode (verified
    against 0.2.110): the spawned CLI loads NO filesystem settings — no
    session hooks, no CLAUDE.md, no operator instructions. The default
    (``None``) loads ALL sources: in the S10 live run that injected the
    operator's config into every call (~10-15k uncached tokens each) and
    mandated a confirmation preamble that made pure-JSON replies impossible.
    ``system_prompt=None`` serializes to an EMPTY system prompt, not the
    Claude Code preset.
    """
    return ClaudeAgentOptions(
        model=model_id,
        max_turns=1,  # a single completion — the agentic loop lives in loop.py, not here
        tools=[],  # pure text completion: no tool surface, no silent egress
        max_budget_usd=max_budget_usd,
        setting_sources=[],
        system_prompt=None,
    )


def _total_tokens(usage: dict[str, Any] | None) -> int | None:
    """Sum the provider-reported token fields; no usage stays ``None`` (§8)."""
    if usage is None:
        return None
    counts = [usage[field] for field in _USAGE_TOKEN_FIELDS if isinstance(usage.get(field), int)]
    return sum(counts) if counts else None


class SdkModelClient:
    """``ModelClient`` over ``claude_agent_sdk.query()`` — one bounded call each.

    ``total_cost_usd`` accumulates the provider-reported cost across calls so
    the run can log it (D6); ``last_model`` carries the REAL model id from the
    latest reply for the §9 provenance stamp.
    """

    def __init__(
        self,
        model_map: ModelMapContract,
        *,
        profile: str = "anthropic",
        max_budget_usd_per_call: float = 0.25,
    ) -> None:
        if max_budget_usd_per_call <= 0:
            raise ValueError(
                f"max_budget_usd_per_call must be positive, got {max_budget_usd_per_call}"
            )
        self._model_map = model_map
        self._profile = profile
        self._max_budget_usd_per_call = max_budget_usd_per_call
        self.total_cost_usd = 0.0
        self.last_model: str | None = None

    def complete(self, prompt: str, *, role: str) -> ModelReply:
        model_id = resolve_model(self._model_map, role, profile=self._profile)
        return asyncio.run(self._complete_async(prompt, model_id))

    async def _complete_async(self, prompt: str, model_id: str) -> ModelReply:
        options = build_call_options(model_id, max_budget_usd=self._max_budget_usd_per_call)
        text_parts: list[str] = []
        reply_model: str | None = None
        usage_tokens: int | None = None
        async for message in query(prompt=prompt, options=options):
            if isinstance(message, AssistantMessage):
                if message.error is not None:
                    raise RuntimeError(f"model call failed: {message.error}")
                reply_model = message.model
                text_parts.extend(
                    block.text for block in message.content if isinstance(block, TextBlock)
                )
            elif isinstance(message, ResultMessage):
                if message.is_error:
                    raise RuntimeError(f"model call failed: {message.subtype}: {message.errors}")
                usage_tokens = _total_tokens(message.usage)
                if message.total_cost_usd is not None:
                    self.total_cost_usd += message.total_cost_usd
        if reply_model is not None:
            self.last_model = reply_model
        return ModelReply(text="".join(text_parts), usage_tokens=usage_tokens, model=reply_model)