portfolio-optimiser-claude/tests/test_sdk_isolation.py

"""Prompt-isolation proof for the SDK call options (§11, S10 post-mortem).

The S10 live run leaked the operator's Claude Code configuration into every
spawned CLI session: ``ClaudeAgentOptions.setting_sources`` defaults to
``None``, which loads ALL filesystem settings (verified against SDK 0.2.110)
— session-start hooks injected STATE.md into the model's context, every reply
opened with a mandated confirmation line (so a reply was NEVER pure JSON),
and each call paid ~10-15k uncached context tokens. ``[]`` is the SDK's
documented isolation mode: no filesystem settings, no hooks, no CLAUDE.md.

Importing the client here is offline-safe: constructing options touches no
network and needs no API key; ``query()`` is replaced with a recording fake.
"""

from __future__ import annotations

from typing import Any, AsyncIterator

import pytest
from claude_agent_sdk import (
    AssistantMessage,
    ClaudeAgentOptions,
    ResultMessage,
    TextBlock,
    ThinkingBlock,
)

from portfolio_optimiser_claude import sdk_client
from portfolio_optimiser_claude.contracts import ModelMapContract
from portfolio_optimiser_claude.sdk_client import SdkModelClient, _total_tokens, build_call_options


class TestBuildCallOptions:
    def test_all_filesystem_settings_are_disabled(self) -> None:
        # LOAD-BEARING (§11): [] is SDK isolation mode. The default (None)
        # loads user+project+local settings — hooks and CLAUDE.md leak into
        # the model's context, exactly the observed S10 failure.
        #
        # Anchor-degeneration control: [] is only a CHOICE for as long as the
        # SDK's own default differs from it. Should a future SDK ship [] as
        # the default, the assertion below would keep passing while saying
        # nothing about our code — so pin the default we are choosing AGAINST.
        assert ClaudeAgentOptions().setting_sources is None, (
            "the SDK default changed — `setting_sources=[]` no longer proves an active choice"
        )
        options = build_call_options("model-x", max_budget_usd=0.25)
        assert options.setting_sources == []

    def test_the_system_prompt_is_not_the_claude_code_preset(self) -> None:
        # Pins the OPTION value against the Claude Code preset. That None
        # reaches the spawned CLI as --system-prompt "" was verified by
        # READING subprocess_cli.py (0.2.110–0.2.120) — this test does NOT
        # bind that transport serialization; doing so would couple the suite
        # to SDK-private API (the F11 fragility this repo retired).
        #
        # HONEST LIMIT (measured, 0.2.120): `system_prompt=None` is NOT
        # distinguishable from leaving the field untouched — the SDK default
        # is None too, so deleting `system_prompt=None` from build_call_options
        # left the old `is None` assertion GREEN. It pinned the SDK's default,
        # not our code. What IS load-bearing is the preset the S10 post-mortem
        # retired, so that is what this guards.
        assert ClaudeAgentOptions().system_prompt is None  # the untouched-field baseline

        # Positive control: the assertion below must be ABLE to fail. The
        # preset form is a real, accepted, distinguishable value.
        preset = ClaudeAgentOptions(
            model="model-x", system_prompt={"type": "preset", "preset": "claude_code"}
        )
        assert preset.system_prompt is not None

        options = build_call_options("model-x", max_budget_usd=0.25)
        assert options.system_prompt != preset.system_prompt
        assert options.system_prompt is None

    def test_the_call_stays_bounded_single_turn_no_tools(self) -> None:
        options = build_call_options("model-x", max_budget_usd=0.25)
        assert options.max_turns == 1
        assert options.tools == []
        assert options.max_budget_usd == 0.25
        assert options.model == "model-x"


class TestCompleteThreadsIsolatedOptions:
    def test_complete_passes_the_isolated_options_to_query(
        self, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        # Detach-proof: if complete() ever builds its options inline again
        # (dropping the isolation), this goes RED — grønn-men-død guard.
        captured: dict[str, Any] = {}

        def fake_query(*, prompt: str, options: Any) -> AsyncIterator[Any]:
            captured["prompt"] = prompt
            captured["options"] = options

            async def _empty() -> AsyncIterator[Any]:
                return
                yield

            return _empty()

        monkeypatch.setattr(sdk_client, "query", fake_query)
        client = SdkModelClient(
            ModelMapContract(profiles={"anthropic": {"default": "model-default"}}),
            max_budget_usd_per_call=0.10,
        )
        reply = client.complete("the prompt", role="proposer")
        assert captured["prompt"] == "the prompt"
        assert captured["options"].setting_sources == []
        assert captured["options"].max_budget_usd == 0.10
        assert captured["options"].model == "model-default"
        # No usage surfaced by the fake → the reply fails CLOSED (§8).
        assert reply.usage_tokens is None


def _stream_of(*messages: Any) -> Any:
    """A fake ``query`` yielding a scripted stream of REAL SDK message objects."""

    def fake_query(*, prompt: str, options: Any) -> AsyncIterator[Any]:
        async def _stream() -> AsyncIterator[Any]:
            for message in messages:
                yield message

        return _stream()

    return fake_query


def _assistant(*blocks: Any, model: str = "model-real", error: Any = None) -> AssistantMessage:
    return AssistantMessage(content=list(blocks), model=model, error=error)


def _result(
    usage: dict[str, Any] | None = None,
    total_cost_usd: float | None = None,
    is_error: bool = False,
    subtype: str = "success",
    errors: list[str] | None = None,
) -> ResultMessage:
    return ResultMessage(
        subtype=subtype,
        duration_ms=1,
        duration_api_ms=1,
        is_error=is_error,
        num_turns=1,
        session_id="s",
        usage=usage,
        total_cost_usd=total_cost_usd,
        errors=errors,
    )


def _client() -> SdkModelClient:
    return SdkModelClient(ModelMapContract(profiles={"anthropic": {"default": "model-default"}}))


_FULL_USAGE = {
    "input_tokens": 10,
    "output_tokens": 5,
    "cache_creation_input_tokens": 3,
    "cache_read_input_tokens": 2,
}


class TestCompleteAsyncStreamBinding:
    """C2.5 (R-4/R-5): the read loop is BOUND offline with real SDK message types.

    Before C2.5 nothing in the suite executed sdk_client's aggregation,
    error, usage or cost branches — the fake stream (real ``AssistantMessage``
    / ``ResultMessage`` / ``TextBlock`` objects, so constructor drift also
    goes red) binds every branch without a key or the network.
    """

    def test_text_aggregates_and_non_text_blocks_are_ignored(
        self, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        monkeypatch.setattr(
            sdk_client,
            "query",
            _stream_of(
                _assistant(TextBlock("{"), ThinkingBlock(thinking="hmm", signature="sig")),
                _assistant(TextBlock("}")),
                _result(usage=_FULL_USAGE, total_cost_usd=0.01),
            ),
        )
        client = _client()
        reply = client.complete("p", role="proposer")
        assert reply.text == "{}"
        assert reply.model == "model-real"
        assert client.last_model == "model-real"

    def test_usage_tokens_sum_the_four_provider_fields(
        self, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        monkeypatch.setattr(
            sdk_client,
            "query",
            _stream_of(_assistant(TextBlock("ok")), _result(usage=_FULL_USAGE)),
        )
        assert _client().complete("p", role="proposer").usage_tokens == 20

    def test_cost_accumulates_across_calls(self, monkeypatch: pytest.MonkeyPatch) -> None:
        client = _client()
        for cost in (0.01, 0.02):
            monkeypatch.setattr(
                sdk_client,
                "query",
                _stream_of(_assistant(TextBlock("ok")), _result(total_cost_usd=cost)),
            )
            client.complete("p", role="proposer")
        assert client.total_cost_usd == pytest.approx(0.03)

    def test_an_assistant_error_fails_the_call(self, monkeypatch: pytest.MonkeyPatch) -> None:
        monkeypatch.setattr(
            sdk_client, "query", _stream_of(_assistant(TextBlock("x"), error="rate_limit"))
        )
        with pytest.raises(RuntimeError, match="rate_limit"):
            _client().complete("p", role="proposer")

    def test_a_result_error_fails_the_call_naming_subtype_and_errors(
        self, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        monkeypatch.setattr(
            sdk_client,
            "query",
            _stream_of(
                _assistant(TextBlock("x")),
                _result(is_error=True, subtype="error_during_execution", errors=["boom"]),
            ),
        )
        with pytest.raises(RuntimeError, match="error_during_execution.*boom"):
            _client().complete("p", role="proposer")


class TestTotalTokensFailsClosed:
    """§8: the meter is never fed an invented count — no usage stays ``None``."""

    def test_no_usage_dict_is_none(self) -> None:
        assert _total_tokens(None) is None

    def test_an_empty_usage_dict_is_none(self) -> None:
        assert _total_tokens({}) is None

    def test_non_int_fields_are_ignored_not_coerced(self) -> None:
        assert _total_tokens({"input_tokens": "10"}) is None
        assert _total_tokens({"input_tokens": 10, "output_tokens": "x"}) == 10


class TestBudgetGuard:
    """§8: a non-positive per-call USD cap is refused at construction."""

    @pytest.mark.parametrize("cap", [0.0, -0.5])
    def test_non_positive_caps_are_rejected(self, cap: float) -> None:
        with pytest.raises(ValueError):
            SdkModelClient(
                ModelMapContract(profiles={"anthropic": {"default": "m"}}),
                max_budget_usd_per_call=cap,
            )