Spaces:
Running
Running
| """Wire-level contract of the shared Claude client: what each request carries and how replies are read.""" | |
| from __future__ import annotations | |
| import json | |
| import os | |
| import sys | |
| from types import SimpleNamespace | |
| from typing import Callable | |
| sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) | |
| import anthropic | |
| import httpx2 | |
| import pytest | |
| from services import ai_client | |
| USER_HI = [{"role": "user", "content": "hi"}] | |
| CODE_FENCED_JSON = '```json\n{"score": 3}\n```' | |
| REFUSAL_DETAILS = {"type": "refusal", "category": "cyber", "explanation": None} | |
| SCORE_SCHEMA = { | |
| "type": "object", | |
| "properties": {"score": {"type": "integer"}}, | |
| "required": ["score"], | |
| "additionalProperties": False, | |
| } | |
| def _text_block(text: str) -> dict: | |
| return {"type": "text", "text": text} | |
| def _message_response(*content: dict, stop_reason: str = "end_turn", stop_details: dict | None = None) -> httpx2.Response: | |
| return httpx2.Response( | |
| 200, | |
| json={ | |
| "id": "msg_test", | |
| "type": "message", | |
| "role": "assistant", | |
| "model": ai_client.CHAT_MODEL, | |
| "content": list(content), | |
| "stop_reason": stop_reason, | |
| "stop_sequence": None, | |
| "stop_details": stop_details, | |
| "usage": {"input_tokens": 11, "output_tokens": 7}, | |
| }, | |
| ) | |
| def _stream_response(*, thinking: str = "", stop_reason: str = "end_turn", stop_details: dict | None = None) -> httpx2.Response: | |
| """A server-sent-event reply that streams an optional thinking block, then the text "Hel" + "lo".""" | |
| events = [ | |
| { | |
| "type": "message_start", | |
| "message": { | |
| "id": "msg_test", | |
| "type": "message", | |
| "role": "assistant", | |
| "model": ai_client.REASONER_MODEL, | |
| "content": [], | |
| "stop_reason": None, | |
| "stop_sequence": None, | |
| "usage": {"input_tokens": 9, "output_tokens": 1}, | |
| }, | |
| } | |
| ] | |
| index = 0 | |
| if thinking: | |
| events += [ | |
| {"type": "content_block_start", "index": index, "content_block": {"type": "thinking", "thinking": "", "signature": ""}}, | |
| {"type": "content_block_delta", "index": index, "delta": {"type": "thinking_delta", "thinking": thinking}}, | |
| {"type": "content_block_delta", "index": index, "delta": {"type": "signature_delta", "signature": "sig"}}, | |
| {"type": "content_block_stop", "index": index}, | |
| ] | |
| index += 1 | |
| events += [ | |
| {"type": "content_block_start", "index": index, "content_block": {"type": "text", "text": ""}}, | |
| {"type": "content_block_delta", "index": index, "delta": {"type": "text_delta", "text": "Hel"}}, | |
| {"type": "content_block_delta", "index": index, "delta": {"type": "text_delta", "text": "lo"}}, | |
| {"type": "content_block_stop", "index": index}, | |
| { | |
| "type": "message_delta", | |
| "delta": {"stop_reason": stop_reason, "stop_sequence": None, "stop_details": stop_details}, | |
| "usage": {"output_tokens": 5}, | |
| }, | |
| {"type": "message_stop"}, | |
| ] | |
| body = "".join(f"event: {event['type']}\ndata: {json.dumps(event)}\n\n" for event in events) | |
| return httpx2.Response(200, headers={"content-type": "text/event-stream"}, content=body.encode()) | |
| class MockApi: | |
| """Stands in for api.anthropic.com: records each request and answers with `reply()`.""" | |
| def __init__(self) -> None: | |
| self.sent: list[SimpleNamespace] = [] | |
| self.reply: Callable[[], httpx2.Response] = lambda: _message_response(_text_block("ok")) | |
| def handle(self, request: httpx2.Request) -> httpx2.Response: | |
| self.sent.append(SimpleNamespace(headers=request.headers, body=json.loads(request.content))) | |
| return self.reply() | |
| def api(monkeypatch: pytest.MonkeyPatch): | |
| """Build the shared client for real, but over a mock transport so nothing reaches the network.""" | |
| mock_api = MockApi() | |
| sdk_client = anthropic.Anthropic | |
| monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-test-key") | |
| monkeypatch.delenv("ANTHROPIC_WORKSPACE_ID", raising=False) | |
| monkeypatch.setattr( | |
| ai_client, | |
| "Anthropic", | |
| lambda **options: sdk_client(**options, http_client=httpx2.Client(transport=httpx2.MockTransport(mock_api.handle))), | |
| ) | |
| ai_client.get_anthropic_client.cache_clear() | |
| yield mock_api | |
| ai_client.get_anthropic_client.cache_clear() | |
| def test_default_model_ids_are_current_api_ids() -> None: | |
| assert ai_client.CHAT_MODEL == "claude-haiku-5-5" | |
| assert ai_client.REASONER_MODEL == "claude-opus-5-5" | |
| def test_current_model_id_keeps_claude_ids_and_blanks_the_rest() -> None: | |
| assert ai_client.current_model_id(" claude-opus-5-5 ") == "claude-opus-5-5" | |
| assert ai_client.current_model_id("retired-chat-model") == "" | |
| assert ai_client.current_model_id("") == "" | |
| def test_client_requires_an_api_key(monkeypatch: pytest.MonkeyPatch) -> None: | |
| monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False) | |
| ai_client.get_anthropic_client.cache_clear() | |
| with pytest.raises(ValueError, match="ANTHROPIC_API_KEY"): | |
| ai_client.get_anthropic_client() | |
| def test_chat_model_request_disables_thinking(api: MockApi) -> None: | |
| ai_client.create_message(USER_HI, model=ai_client.CHAT_MODEL, max_tokens=300) | |
| request = api.sent[0] | |
| assert request.body["model"] == ai_client.CHAT_MODEL | |
| assert request.body["thinking"] == {"type": "disabled"} | |
| assert request.body["max_tokens"] == 300 | |
| assert "output_config" not in request.body | |
| assert "fallbacks" not in request.body | |
| assert "anthropic-beta" not in request.headers | |
| def test_reasoner_request_sets_effort_token_floor_and_refusal_fallback(api: MockApi) -> None: | |
| ai_client.create_message(USER_HI, model=ai_client.REASONER_MODEL, max_tokens=500) | |
| request = api.sent[0] | |
| assert request.body["model"] == ai_client.REASONER_MODEL | |
| assert request.body.get("thinking", {}).get("type") != "disabled" | |
| assert request.body["output_config"] == {"effort": "medium"} | |
| assert request.body["max_tokens"] >= 16384 | |
| assert request.headers["anthropic-beta"] == "server-side-fallback-2026-07-01" | |
| assert request.body["fallbacks"] == "default" | |
| def test_effort_argument_reaches_the_request(api: MockApi) -> None: | |
| ai_client.create_message(USER_HI, model=ai_client.REASONER_MODEL, effort="high") | |
| ai_client.create_message(USER_HI, model=ai_client.CHAT_MODEL, effort="low") | |
| assert [request.body["output_config"] for request in api.sent] == [{"effort": "high"}, {"effort": "low"}] | |
| assert api.sent[1].body["thinking"] == {"type": "disabled"} | |
| def test_sampling_parameters_are_not_accepted(api: MockApi) -> None: | |
| with pytest.raises(TypeError): | |
| ai_client.create_message(USER_HI, temperature=0.2) # type: ignore[call-arg] | |
| for model in (ai_client.CHAT_MODEL, ai_client.REASONER_MODEL): | |
| ai_client.create_message(USER_HI, model=model) | |
| for request in api.sent: | |
| assert not {"temperature", "top_p", "top_k"} & request.body.keys() | |
| def test_workspace_id_is_sent_as_a_header_when_configured(api: MockApi, monkeypatch: pytest.MonkeyPatch) -> None: | |
| monkeypatch.setenv("ANTHROPIC_WORKSPACE_ID", "wrkspc_test") | |
| ai_client.get_anthropic_client.cache_clear() | |
| ai_client.create_message(USER_HI) | |
| assert api.sent[0].headers["anthropic-workspace-id"] == "wrkspc_test" | |
| def test_workspace_header_is_omitted_by_default(api: MockApi) -> None: | |
| ai_client.create_message(USER_HI) | |
| assert "anthropic-workspace-id" not in api.sent[0].headers | |
| def test_system_messages_are_hoisted_to_the_system_parameter(api: MockApi) -> None: | |
| ai_client.create_message( | |
| [ | |
| {"role": "system", "content": "Be brief."}, | |
| {"role": "user", "content": "Hi"}, | |
| {"role": "system", "content": "Use plain words."}, | |
| {"role": "assistant", "content": "Hello"}, | |
| {"role": "user", "content": "Add 2 and 2"}, | |
| ] | |
| ) | |
| body = api.sent[0].body | |
| assert body["system"] == "Be brief.\n\nUse plain words." | |
| assert body["messages"] == [ | |
| {"role": "user", "content": "Hi"}, | |
| {"role": "assistant", "content": "Hello"}, | |
| {"role": "user", "content": "Add 2 and 2"}, | |
| ] | |
| def test_leading_assistant_and_blank_turns_are_dropped(api: MockApi) -> None: | |
| ai_client.create_message( | |
| [ | |
| {"role": "assistant", "content": "Welcome back"}, | |
| {"role": "user", "content": " "}, | |
| {"role": "user", "content": "Hi"}, | |
| ] | |
| ) | |
| body = api.sent[0].body | |
| assert body["messages"] == [{"role": "user", "content": "Hi"}] | |
| assert "system" not in body | |
| def test_json_output_adds_an_instruction_and_strips_the_code_fence(api: MockApi) -> None: | |
| api.reply = lambda: _message_response(_text_block(CODE_FENCED_JSON)) | |
| reply = ai_client.complete_text( | |
| [{"role": "system", "content": "Grade it."}, {"role": "user", "content": "2 + 2"}], json_output=True | |
| ) | |
| assert json.loads(reply) == {"score": 3} | |
| system = api.sent[0].body["system"] | |
| assert system.startswith("Grade it.") | |
| assert "JSON" in system | |
| def test_json_output_recovers_json_wrapped_in_prose(api: MockApi) -> None: | |
| api.reply = lambda: _message_response( | |
| _text_block('Here is the classification:\n{"risk_label": "high risk of failing", "confidence": 0.91}\nHope it helps.') | |
| ) | |
| reply = ai_client.complete_text(USER_HI, json_output=True) | |
| assert json.loads(reply) == {"risk_label": "high risk of failing", "confidence": 0.91} | |
| def test_json_output_returns_the_text_when_no_json_can_be_recovered(api: MockApi) -> None: | |
| api.reply = lambda: _message_response(_text_block("I cannot classify this.")) | |
| assert ai_client.complete_text(USER_HI, json_output=True) == "I cannot classify this." | |
| def test_json_output_keeps_a_truncated_reply_whole_for_the_callers_salvage_parser(api: MockApi) -> None: | |
| truncated = '[{"question": "Q1", "answer": "A"}, {"question": "Q2", "answer": "B"}, {"question": "Q3", "ans' | |
| api.reply = lambda: _message_response(_text_block(truncated), stop_reason="max_tokens") | |
| assert ai_client.complete_text(USER_HI, json_output=True) == truncated | |
| def test_json_output_never_returns_a_nested_fragment_of_malformed_json(api: MockApi) -> None: | |
| malformed = '{"summary": "Review the "hole" rule", "practice": {"problem": "Find the domain", "solution": "x != 2"}}' | |
| api.reply = lambda: _message_response(_text_block(malformed)) | |
| assert ai_client.complete_text(USER_HI, json_output=True) == malformed | |
| def test_reply_keeps_the_code_fence_unless_json_output_is_requested(api: MockApi) -> None: | |
| api.reply = lambda: _message_response(_text_block(CODE_FENCED_JSON)) | |
| assert ai_client.complete_text(USER_HI) == CODE_FENCED_JSON | |
| assert "system" not in api.sent[0].body | |
| def test_schema_turns_on_structured_outputs_without_a_json_instruction(api: MockApi) -> None: | |
| ai_client.create_message( | |
| [{"role": "system", "content": "Grade it."}, {"role": "user", "content": "2 + 2"}], schema=SCORE_SCHEMA | |
| ) | |
| body = api.sent[0].body | |
| assert body["output_config"] == {"format": {"type": "json_schema", "schema": SCORE_SCHEMA}} | |
| assert body["system"] == "Grade it." | |
| def test_structured_outputs_keep_the_effort_setting(api: MockApi) -> None: | |
| ai_client.create_message(USER_HI, model=ai_client.REASONER_MODEL, schema=SCORE_SCHEMA) | |
| assert api.sent[0].body["output_config"] == { | |
| "effort": "medium", | |
| "format": {"type": "json_schema", "schema": SCORE_SCHEMA}, | |
| } | |
| def test_complete_json_returns_the_parsed_object(api: MockApi) -> None: | |
| api.reply = lambda: _message_response(_text_block('{"score": 3}')) | |
| assert ai_client.complete_json(USER_HI, schema=SCORE_SCHEMA) == {"score": 3} | |
| assert api.sent[0].body["output_config"]["format"]["schema"] == SCORE_SCHEMA | |
| def test_complete_json_raises_when_claude_declines(api: MockApi) -> None: | |
| api.reply = lambda: _message_response(stop_reason="refusal", stop_details=REFUSAL_DETAILS) | |
| with pytest.raises(ai_client.AIRefusalError): | |
| ai_client.complete_json(USER_HI, schema=SCORE_SCHEMA) | |
| def test_non_claude_model_id_is_replaced_by_the_chat_model(api: MockApi, stale_id: str) -> None: | |
| ai_client.create_message(USER_HI, model=stale_id) | |
| assert api.sent[0].body["model"] == ai_client.CHAT_MODEL | |
| def test_reply_text_joins_text_blocks_and_skips_thinking(api: MockApi) -> None: | |
| api.reply = lambda: _message_response( | |
| {"type": "thinking", "thinking": "Considering.", "signature": "sig"}, | |
| _text_block("Hello "), | |
| _text_block("world"), | |
| ) | |
| assert ai_client.complete_text(USER_HI, model=ai_client.REASONER_MODEL) == "Hello world" | |
| def test_refusal_raises_ai_refusal_error(api: MockApi) -> None: | |
| api.reply = lambda: _message_response(stop_reason="refusal", stop_details=REFUSAL_DETAILS) | |
| message = ai_client.create_message(USER_HI) | |
| assert message.stop_reason == "refusal" | |
| with pytest.raises(ai_client.AIRefusalError, match="cyber"): | |
| ai_client.reply_text(message) | |
| with pytest.raises(ai_client.AIRefusalError): | |
| ai_client.complete_text(USER_HI) | |
| def test_rate_limit_surfaces_without_sdk_retries(api: MockApi) -> None: | |
| api.reply = lambda: httpx2.Response( | |
| 429, json={"type": "error", "error": {"type": "rate_limit_error", "message": "slow down"}} | |
| ) | |
| with pytest.raises(ai_client.RateLimitError): | |
| ai_client.create_message(USER_HI) | |
| assert len(api.sent) == 1, "retries belong to InferenceClient; SDK retries would stack under them" | |
| def test_stream_text_yields_text_deltas_and_streams_summarized_thinking(api: MockApi) -> None: | |
| api.reply = lambda: _stream_response(thinking="Weighing options.") | |
| assert list(ai_client.stream_text(USER_HI, model=ai_client.REASONER_MODEL)) == ["Hel", "lo"] | |
| body = api.sent[0].body | |
| assert body["stream"] is True | |
| assert body["thinking"] == {"type": "adaptive", "display": "summarized"} | |
| def test_chat_model_stream_keeps_thinking_disabled(api: MockApi) -> None: | |
| api.reply = lambda: _stream_response() | |
| assert "".join(ai_client.stream_text(USER_HI, model=ai_client.CHAT_MODEL)) == "Hello" | |
| assert api.sent[0].body["thinking"] == {"type": "disabled"} | |
| def _drain_open_stream(model: str) -> None: | |
| with ai_client.open_stream(USER_HI, model=model) as stream: | |
| stream.get_final_message() | |
| def _drain_stream_text(model: str) -> None: | |
| list(ai_client.stream_text(USER_HI, model=model)) | |
| STREAM_ENTRY_POINTS = pytest.mark.parametrize( | |
| "drain", [_drain_open_stream, _drain_stream_text], ids=["open_stream", "stream_text"] | |
| ) | |
| def test_reasoner_stream_carries_the_refusal_fallback(api: MockApi, drain: Callable[[str], None]) -> None: | |
| api.reply = lambda: _stream_response(thinking="Weighing options.") | |
| drain(ai_client.REASONER_MODEL) | |
| request = api.sent[0] | |
| assert request.body["stream"] is True | |
| assert request.headers["anthropic-beta"] == "server-side-fallback-2026-07-01" | |
| assert request.body["fallbacks"] == "default" | |
| def test_chat_model_stream_carries_no_refusal_fallback(api: MockApi, drain: Callable[[str], None]) -> None: | |
| api.reply = lambda: _stream_response() | |
| drain(ai_client.CHAT_MODEL) | |
| request = api.sent[0] | |
| assert request.body["stream"] is True | |
| assert "anthropic-beta" not in request.headers | |
| assert "fallbacks" not in request.body | |
| def test_open_stream_exposes_thinking_deltas_and_final_usage(api: MockApi) -> None: | |
| api.reply = lambda: _stream_response(thinking="Weighing options.") | |
| with ai_client.open_stream(USER_HI, model=ai_client.REASONER_MODEL) as stream: | |
| deltas = [event.delta for event in stream if event.type == "content_block_delta"] | |
| final = stream.get_final_message() | |
| assert "".join(delta.thinking for delta in deltas if delta.type == "thinking_delta") == "Weighing options." | |
| assert "".join(delta.text for delta in deltas if delta.type == "text_delta") == "Hello" | |
| assert final.stop_reason == "end_turn" | |
| assert (final.usage.input_tokens, final.usage.output_tokens) == (9, 5) | |
| def test_stream_text_raises_after_the_stream_when_claude_declines(api: MockApi) -> None: | |
| api.reply = lambda: _stream_response(stop_reason="refusal", stop_details=REFUSAL_DETAILS) | |
| received: list[str] = [] | |
| with pytest.raises(ai_client.AIRefusalError): | |
| for chunk in ai_client.stream_text(USER_HI, model=ai_client.CHAT_MODEL): | |
| received.append(chunk) | |
| assert received == ["Hel", "lo"] | |