test(anthropic): trim DeepSeek proxy replay tests to two invariants + local wire probe
Fold the adapter-level unsigned-preservation case and the name-detector matrices into one parametrized invariant (thinking family keeps unsigned only; non-thinking DeepSeek and unrelated models still take the generic third-party strip) plus the native-endpoint immutability guard. Add evals/anthropic_proxy_thinking_replay.py, a real-AIAgent probe against a synthetic local Anthropic-compatible relay used for the before/after wire check.
This commit is contained in:
Teknium
103
evals/anthropic_proxy_thinking_replay.py
Normal file
103
evals/anthropic_proxy_thinking_replay.py
Normal file
@@ -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 <out.json>
|
||||
"""
|
||||
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()
|
||||
@@ -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 = [
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user