diff --git a/evals/anthropic_proxy_thinking_replay.py b/evals/anthropic_proxy_thinking_replay.py new file mode 100644 index 0000000000..4072e4b04a --- /dev/null +++ b/evals/anthropic_proxy_thinking_replay.py @@ -0,0 +1,103 @@ +"""Local wire-contract probe: which thinking blocks a real AIAgent replays to an +Anthropic-compatible relay. Synthetic SSE fixtures; NOT live-provider proof. + +Run with the repo interpreter in an isolated HOME/HERMES_HOME: + python evals/anthropic_proxy_thinking_replay.py +""" +import json +import os +import sys +import tempfile +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +HISTORY = [ + {"role": "user", "content": "inspect the repo"}, + {"role": "assistant", "content": "checking", "reasoning_details": [ + {"type": "thinking", "thinking": "unsigned proxy reasoning"}, + {"type": "thinking", "thinking": "foreign signed", "signature": "sig-from-elsewhere"}, + ], "tool_calls": [{"id": "call_1", "type": "function", + "function": {"name": "terminal", "arguments": "{\"command\": \"ls\"}"}}]}, + {"role": "tool", "tool_call_id": "call_1", "content": "README.md"}, + {"role": "assistant", "content": "done"}, +] + + +def _sse(events): + return "".join(f"event: {e['type']}\ndata: {json.dumps(e)}\n\n" for e in events).encode() + + +def scenario(model, path="/anthropic"): + import run_agent + requests = [] + + class Handler(BaseHTTPRequestHandler): + def log_message(self, *_a): + pass + + def do_POST(self): + payload = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) + if self.path.endswith("/messages"): # skip the Ollama /api/show capability probe + requests.append(payload) + msg = {"id": "msg_local", "type": "message", "role": "assistant", "model": payload.get("model"), + "content": [], "stop_reason": None, "stop_sequence": None, + "usage": {"input_tokens": 10, "output_tokens": 0}} + events = [ + {"type": "message_start", "message": msg}, + {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}, + {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "local fixture reply"}}, + {"type": "content_block_stop", "index": 0}, + {"type": "message_delta", "delta": {"stop_reason": "end_turn", "stop_sequence": None}, + "usage": {"output_tokens": 3}}, + {"type": "message_stop"}, + ] + raw = _sse(events) + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.send_header("Content-Length", str(len(raw))) + self.end_headers() + self.wfile.write(raw) + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + threading.Thread(target=server.serve_forever, daemon=True).start() + agent = run_agent.AIAgent( + model=model, provider="custom", base_url=f"http://127.0.0.1:{server.server_port}{path}", + api_key="local-fixture-key", enabled_toolsets=["terminal"], max_iterations=1, quiet_mode=True, + reasoning_config={"enabled": True, "effort": "high"}, skip_context_files=True, skip_memory=True, + skip_background_review=True, save_trajectories=False, session_id=f"proxy-thinking-{model.replace('/', '-')}", + ) + agent._api_max_retries = 1 + try: + result = agent.run_conversation("reply briefly", system_message="Local wire probe.", + conversation_history=[dict(m) for m in HISTORY]) + finally: + agent.close() + server.shutdown() + server.server_close() + wire = requests[0] if requests else {} + assistant_turns = [m for m in wire.get("messages", []) if m.get("role") == "assistant"] + thinking = [ + {k: b.get(k) for k in ("type", "thinking", "signature") if k in b} + for m in assistant_turns for b in (m.get("content") if isinstance(m.get("content"), list) else []) + if b.get("type") in ("thinking", "redacted_thinking") + ] + return {"model": model, "api_mode": agent.api_mode, "completed": result.get("completed"), + "requests": len(requests), "wire_thinking_blocks": thinking, + "thinking_param": wire.get("thinking"), "tool_count": len(wire.get("tools") or [])} + + +def main(): + out = {"surface": "real AIAgent + Anthropic SDK against a synthetic local relay; not provider proof", + "scenarios": [scenario("deepseek-ai/deepseek-v4-pro"), scenario("some-vendor/other-model")]} + Path(sys.argv[1]).write_text(json.dumps(out, indent=2) + "\n", encoding="utf-8") + print(json.dumps(out, indent=2)) + + +if __name__ == "__main__": + with tempfile.TemporaryDirectory(prefix="proxy-thinking-") as home: + os.environ.update(HOME=home, HERMES_HOME=home, HERMES_DISABLE_PLUGINS="1") + main() diff --git a/tests/agent/test_anthropic_adapter.py b/tests/agent/test_anthropic_adapter.py index bbc9a822e1..3ded294a6c 100644 --- a/tests/agent/test_anthropic_adapter.py +++ b/tests/agent/test_anthropic_adapter.py @@ -1225,31 +1225,6 @@ class TestThinkingBlockSignatureManagement: - def test_third_party_deepseek_preserves_unsigned_thinking(self): - messages = [ - { - "role": "assistant", - "content": "Response text.", - "reasoning_details": [ - {"type": "thinking", "thinking": "Proxy reasoning."}, - ], - }, - ] - - _, result = convert_messages_to_anthropic( - messages, - base_url="https://proxy.example.com/anthropic", - model="deepseek-ai/deepseek-v4", - ) - - assistant = next(m for m in result if m["role"] == "assistant") - thinking = [ - block - for block in assistant["content"] - if block.get("type") == "thinking" - ] - assert thinking == [{"type": "thinking", "thinking": "Proxy reasoning."}] - def test_redacted_thinking_with_data_preserved(self): """Redacted thinking with 'data' field is kept on last turn.""" messages = [ diff --git a/tests/agent/test_deepseek_anthropic_thinking.py b/tests/agent/test_deepseek_anthropic_thinking.py index 543ad1952c..48cce7d8bd 100644 --- a/tests/agent/test_deepseek_anthropic_thinking.py +++ b/tests/agent/test_deepseek_anthropic_thinking.py @@ -105,21 +105,6 @@ class TestDeepSeekAnthropicPreservesThinking: assert "cache_control" not in b -@pytest.mark.parametrize("model", [ - "deepseek-r1", "deepseek-v4", "vendor/deepseek-v4-pro", - "gateway/deepseek-ai/deepseek_flash", " DeepSeek-Pro ", "deepseek_v4_flash", -]) -def test_thinking_family_name_detection(model): - from agent.anthropic_endpoints import _model_name_is_deepseek_thinking - assert _model_name_is_deepseek_thinking(model) - - -@pytest.mark.parametrize("model", [None, "", " ", 42, "deepseek-chat", "deepseek-v3", "vendor/", "not-deepseek-v4", "qwen-thinking"]) -def test_thinking_family_name_detection_rejects_unknown_models(model): - from agent.anthropic_endpoints import _model_name_is_deepseek_thinking - assert not _model_name_is_deepseek_thinking(model) - - @pytest.mark.parametrize("url", [None, "https://api.anthropic.com", "https://inference-api.nousresearch.com/anthropic"]) def test_deepseek_model_name_does_not_override_native_signature_contract(url): from agent.anthropic_message_convert import _manage_thinking_signatures @@ -129,7 +114,13 @@ def test_deepseek_model_name_does_not_override_native_signature_contract(url): assert messages[0]["content"][0] == block -def test_deepseek_proxy_keeps_unsigned_thinking_in_older_tool_turns_only(): +@pytest.mark.parametrize(("model", "kept"), [ + ("vendor/deepseek-v4", [{"type": "thinking", "thinking": "unsigned"}]), # thinking family: keep unsigned only + (" DeepSeek-Pro ", [{"type": "thinking", "thinking": "unsigned"}]), + ("deepseek-chat", []), # non-thinking DeepSeek and unrelated models: generic third-party strip + ("vendor/other-model", []), +]) +def test_deepseek_proxy_keeps_unsigned_thinking_in_older_tool_turns_only(model, kept): import copy from agent.anthropic_message_convert import convert_messages_to_anthropic history = [ @@ -143,10 +134,7 @@ def test_deepseek_proxy_keeps_unsigned_thinking_in_older_tool_turns_only(): {"role": "assistant", "content": "done"}, ] snapshot = copy.deepcopy(history) - _, result = convert_messages_to_anthropic(history, base_url="https://proxy.example/anthropic", model="vendor/deepseek-v4") + _, result = convert_messages_to_anthropic(history, base_url="https://proxy.example/anthropic", model=model) assistant = next(m for m in result if m["role"] == "assistant") - assert [b for b in assistant["content"] if b.get("type") in {"thinking", "redacted_thinking"}] == [ - {"type": "thinking", "thinking": "unsigned"}, - ] + assert [b for b in assistant["content"] if b.get("type") in {"thinking", "redacted_thinking"}] == kept assert history == snapshot -