Migrate GPT-5.5 to Claude Opus 4.8

%%capture
%pip install --quiet \
    "openai>=1.50.0" \
    "anthropic>=0.55.0" \
    "azure-identity>=1.19.0" \
    "httpx>=0.27.0" \
    "jsonschema>=4.22.0" \
    "python-dotenv>=1.0.1"

import os
from urllib.parse import urlparse
from dotenv import load_dotenv

load_dotenv(override=True)

OPENAI_ENDPOINT       = os.environ["OPENAI_ENDPOINT"].rstrip("/")
OPENAI_DEPLOYMENT     = os.environ.get("OPENAI_DEPLOYMENT", "gpt-5.5")
OPENAI_API_KEY        = os.environ.get("OPENAI_API_KEY")        # None → use Entra
OPENAI_API_VERSION    = os.environ.get("OPENAI_API_VERSION", "2024-10-21")

ANTH_BASE_URL         = os.environ["ANTH_BASE_URL"].rstrip("/")
ANTH_DEPLOYMENT       = os.environ.get("ANTH_DEPLOYMENT", "claude-opus-4-8")
ANTH_API_KEY          = os.environ.get("ANTH_API_KEY")          # None → use Entra

def _redact_ai_url(url: str) -> str:
    parsed = urlparse(url)
    host = parsed.netloc.lower()
    if host.endswith(".services.ai.azure.com"):
        path = "/anthropic" if parsed.path.rstrip("/") == "/anthropic" else ""
        return f"{parsed.scheme}://<your-foundry-resource>.services.ai.azure.com{path}"
    return "<configured endpoint>"

print(f"OpenAI endpoint   : {_redact_ai_url(OPENAI_ENDPOINT)}")
print(f"OpenAI deployment : {OPENAI_DEPLOYMENT}  (api-version {OPENAI_API_VERSION})")
print(f"Anthropic base    : {_redact_ai_url(ANTH_BASE_URL)}")
print(f"Anthropic model   : {ANTH_DEPLOYMENT}")

from azure.identity import DefaultAzureCredential, get_bearer_token_provider
from openai import AzureOpenAI
from anthropic import AnthropicFoundry

# Two audiences, one credential. Build separate token providers so the right scope
# travels with the right client.
_cred = DefaultAzureCredential()
openai_token_provider = get_bearer_token_provider(
    _cred, "https://cognitiveservices.azure.com/.default"
)
anth_token_provider = get_bearer_token_provider(
    _cred, "https://ai.azure.com/.default"
)

# Baseline client: OpenAI on Foundry (Chat Completions path).
openai_client = AzureOpenAI(
    azure_endpoint=OPENAI_ENDPOINT,
    api_version=OPENAI_API_VERSION,
    **({"api_key": OPENAI_API_KEY} if OPENAI_API_KEY is not None else {"azure_ad_token_provider": openai_token_provider}),
)

# Candidate client: Anthropic on Foundry. base_url ends in /anthropic;
# the SDK appends /v1/messages itself.
anth_client = AnthropicFoundry(
    base_url=ANTH_BASE_URL,
    **({"api_key": ANTH_API_KEY} if ANTH_API_KEY
       else {"azure_ad_token_provider": anth_token_provider}),
)

print("clients built. openai:", type(openai_client).__name__,
      "| anthropic:", type(anth_client).__name__)

import httpx
from urllib.parse import urlparse

# Reachability probe only. The Foundry host root doesn't serve a meaningful endpoint
# for HEAD, so a 404 or 405 here is normal; it confirms DNS, TLS, and routing work.
# We're not exercising the OpenAI or Anthropic API path; that happens in later cells.
# Any 2xx/3xx/4xx response = host is reachable. Only ERR (timeout / DNS) is a real failure.
def _host(url: str) -> str:
    p = urlparse(url)
    return f"{p.scheme}://{p.netloc}"

for label, url in [("OpenAI", OPENAI_ENDPOINT),
                   ("Anthropic-on-Foundry", _host(ANTH_BASE_URL))]:
    display_url = _redact_ai_url(url)
    try:
        r = httpx.head(url, timeout=5.0)
        print(f"  {label:<22} {display_url:<60}  {r.status_code}")
    except Exception as exc:
        print(f"  {label:<22} {display_url:<60}  ERR {type(exc).__name__}: {exc}")

baseline_resp = openai_client.chat.completions.create(
    model=OPENAI_DEPLOYMENT,
    messages=[
        {"role": "system", "content": "You are a careful research assistant."},
        {"role": "user",   "content": "In one sentence: what does Microsoft Foundry do?"},
    ],
    max_completion_tokens=200,
)

print("=== gpt-5.5 (OpenAI on Foundry) ===")
print(baseline_resp.choices[0].message.content)
print()
print("usage:", baseline_resp.usage)

candidate_resp = anth_client.messages.create(
    model=ANTH_DEPLOYMENT,
    system="You are a careful research assistant.",
    messages=[
        {"role": "user", "content": "In one sentence: what does Microsoft Foundry do?"},
    ],
    max_tokens=200,
)

# Anthropic returns content as a list of typed blocks. The assistant's text lives in
# `text` blocks; any `thinking` blocks (when extended thinking is enabled) should not
# be rendered to end users.
text_blocks = [b.text for b in candidate_resp.content if b.type == "text"]
print("=== claude-opus-4-8 (Anthropic on Foundry) ===")
print("\n".join(text_blocks))
print()
print("usage:", candidate_resp.usage)

import json
from dataclasses import dataclass, field
from typing import Any, Optional
from jsonschema import Draft202012Validator, ValidationError

# ---------- Canonical types ----------

@dataclass
class Message:
    role: str                            # "system" | "user" | "assistant" | "tool"
    content: Any                         # str or list of typed blocks
    tool_call_id: Optional[str] = None   # only when role == "tool"

@dataclass
class ToolSpec:
    name: str
    description: str
    parameters: dict                     # JSON Schema (OpenAI strict shape)

@dataclass
class ToolCall:
    id: str
    name: str
    arguments: dict                      # parsed JSON

@dataclass
class Conversation:
    system: str
    messages: list[Message] = field(default_factory=list)
    tools: list[ToolSpec] = field(default_factory=list)
    thinking: bool = False               # turn on adaptive thinking (Anthropic only)
    effort: str = "medium"               # low | medium | high (confirm any higher tiers in the Foundry Claude docs for your model version)
    max_tokens: int = 1024
    json_schema: Optional[dict] = None   # when set, output is validated against this schema

@dataclass
class Response:
    text: str
    tool_calls: list[ToolCall]
    usage: dict
    raw: Any
    retries: int = 0                     # JSON-schema retry count (0 for OpenAI strict; 0 or 1 for Anthropic)

# ---------- Helpers ----------

def openai_tools_to_anthropic(tools: list[ToolSpec]) -> list[dict]:
    """OpenAI function tools -> Anthropic tool blocks. The JSON Schema passes through;
    only the wrapper shape changes."""
    return [
        {"name": t.name, "description": t.description, "input_schema": t.parameters}
        for t in tools
    ]

def _strip_json_fences(raw: str) -> str:
    """Remove a ```json ... ``` wrapper if the model insisted on one."""
    raw = raw.strip()
    if raw.startswith("```"):
        raw = raw.removeprefix("```json").removeprefix("```").removesuffix("```").strip()
    return raw

def _anthropic_content(content: Any) -> Any:
    """Convert SDK content blocks to plain dictionaries when a prior response is reused."""
    if not isinstance(content, list):
        return content
    return [block.model_dump() if hasattr(block, "model_dump") else block for block in content]

def anthropic_messages(messages: list[Message]) -> list[dict]:
    """Translate canonical history, including tool results, to Anthropic message blocks."""
    translated = []
    for message in messages:
        if message.role in ("user", "assistant"): 
            translated.append({"role": message.role,
                               "content": _anthropic_content(message.content)})
        elif message.role == "tool":
            if not message.tool_call_id:
                raise ValueError("tool messages require tool_call_id")
            translated.append({
                "role": "user",
                "content": [{
                    "type": "tool_result",
                    "tool_use_id": message.tool_call_id,
                    "content": _anthropic_content(message.content),
                }],
            })
        else:
            raise ValueError(f"Unsupported canonical role: {message.role}")
    return translated

def tool_calls_meet_contract(tools: list[ToolSpec], expected_name: str, calls: list,
                            error: Optional[str] = None) -> bool:
    """Fail closed unless every tool call has the expected name and valid arguments."""
    tools_by_name = {tool.name: tool for tool in tools}
    if error or not calls:
        return False
    for name, arguments in calls:
        if name != expected_name or name not in tools_by_name:
            return False
        try:
            Draft202012Validator(tools_by_name[name].parameters).validate(arguments)
        except (ValidationError, TypeError):
            return False
    return True

# ---------- Adapters ----------

class OpenAIAdapter:
    def __init__(self, client): self.client = client

    def run(self, conv: Conversation, deployment: str) -> Response:
        msgs = [{"role": "system", "content": conv.system}]
        for m in conv.messages:
            entry = {"role": m.role, "content": m.content}
            if m.tool_call_id is not None:
                entry["tool_call_id"] = m.tool_call_id
            msgs.append(entry)
        kwargs: dict = {"model": deployment, "messages": msgs,
                        "max_completion_tokens": conv.max_tokens}
        if conv.tools:
            kwargs["tools"] = [
                {"type": "function",
                 "function": {"name": t.name, "description": t.description,
                              "parameters": t.parameters}}
                for t in conv.tools
            ]
        if conv.json_schema is not None:
            # strict:true is the hard guarantee; retries are always 0 on this path.
            kwargs["response_format"] = {
                "type": "json_schema",
                "json_schema": {"name": "output", "strict": True, "schema": conv.json_schema},
            }
        resp = self.client.chat.completions.create(**kwargs)
        msg = resp.choices[0].message
        calls = [
            ToolCall(id=tc.id, name=tc.function.name,
                     arguments=json.loads(tc.function.arguments or "{}"))
            for tc in (msg.tool_calls or [])
        ]
        return Response(
            text=msg.content or "",
            tool_calls=calls,
            usage={"input":  resp.usage.prompt_tokens,
                   "output": resp.usage.completion_tokens},
            raw=resp,
            retries=0,
        )


class AnthropicAdapter:
    def __init__(self, client): self.client = client

    def run(self, conv: Conversation, deployment: str) -> Response:
        if conv.json_schema is not None:
            return self._run_with_json_schema(conv, deployment)

        msgs = anthropic_messages(conv.messages)
        kwargs: dict = {"model": deployment, "system": conv.system,
                        "messages": msgs, "max_tokens": conv.max_tokens}
        if conv.tools:
            kwargs["tools"] = openai_tools_to_anthropic(conv.tools)
        if conv.thinking:
            kwargs["thinking"] = {"type": "adaptive"}
            kwargs["output_config"] = {"effort": conv.effort}
        resp = self.client.messages.create(**kwargs)
        return self._build_response(resp, retries=0)

    def _run_with_json_schema(self, conv: Conversation, deployment: str) -> Response:
        """Validate-and-retry once. Anthropic has no strict JSON-schema mode, so we
        instruct, parse, validate, and retry exactly once on failure."""
        schema_text = json.dumps(conv.json_schema, indent=2)
        msgs = anthropic_messages(conv.messages)
        # Augment the final user turn with the schema. Preserves the caller's prompt.
        if msgs and msgs[-1]["role"] == "user" and isinstance(msgs[-1]["content"], str):
            msgs[-1] = {"role": "user",
                        "content": f"{msgs[-1]['content']}\n\n"
                                   f"Return JSON only matching this schema:\n{schema_text}"}
        system = conv.system + "\n\nReturn JSON only. No prose. No markdown fences."

        last_exc: Exception = RuntimeError("no attempts ran")
        for attempt in range(2):
            resp = self.client.messages.create(
                model=deployment, system=system, messages=msgs,
                max_tokens=conv.max_tokens,
            )
            raw = "".join(b.text for b in resp.content if b.type == "text")
            raw = _strip_json_fences(raw)
            try:
                parsed = json.loads(raw)
                Draft202012Validator(conv.json_schema).validate(parsed)
                return Response(
                    text=json.dumps(parsed),
                    tool_calls=[],
                    usage=self._usage(resp),
                    raw=resp,
                    retries=attempt,
                )
            except (json.JSONDecodeError, ValidationError) as exc:
                last_exc = exc
                msgs.append({"role": "assistant", "content": raw})
                msgs.append({"role": "user",
                             "content": f"That did not match the schema ({exc}). "
                                        "Re-emit valid JSON only."})
        raise RuntimeError(f"Anthropic structured output failed twice: {last_exc}")

    def _build_response(self, resp, retries: int) -> Response:
        text = "".join(b.text for b in resp.content if b.type == "text")
        calls = [
            ToolCall(id=b.id, name=b.name, arguments=b.input or {})
            for b in resp.content if b.type == "tool_use"
        ]
        return Response(text=text, tool_calls=calls,
                        usage=self._usage(resp), raw=resp, retries=retries)

    @staticmethod
    def _usage(resp) -> dict:
        u = resp.usage
        return {
            "input":        u.input_tokens,
            "output":       u.output_tokens,
            "cache_read":   getattr(u, "cache_read_input_tokens", 0),
            "cache_create": getattr(u, "cache_creation_input_tokens", 0),
        }

openai_adapter = OpenAIAdapter(openai_client)
anth_adapter   = AnthropicAdapter(anth_client)
print("adapters ready")

# Credential-free assertion: retain the assistant tool_use block and translate the
 # canonical tool result into the Anthropic user-message shape.
history = [
    Message(role="user", content="What is the weather in Seattle?"),
    Message(
        role="assistant",
        content=[{
            "type": "tool_use",
            "id": "call_weather",
            "name": "get_weather",
            "input": {"city": "Seattle", "units": "f"},
        }],
    ),
    Message(
        role="tool",
        tool_call_id="call_weather",
        content="{\"temperature\": 62, \"units\": \"f\"}",
    ),
]
translated = anthropic_messages(history)
assert translated[1]["role"] == "assistant"
assert translated[1]["content"][0]["type"] == "tool_use"
assert translated[2] == {
    "role": "user",
    "content": [{
        "type": "tool_result",
        "tool_use_id": "call_weather",
        "content": "{\"temperature\": 62, \"units\": \"f\"}",
    }],
}
weather_schema = ToolSpec(
    name="get_weather",
    description="Get weather.",
    parameters={"type": "object", "properties": {"city": {"type": "string"}},
                "required": ["city"], "additionalProperties": False},
)
assert tool_calls_meet_contract([weather_schema], "get_weather",
                                [("get_weather", {"city": "Seattle"})])
assert not tool_calls_meet_contract([weather_schema], "get_weather",
                                    [("get_weather", {"city": 42})])
assert not tool_calls_meet_contract([weather_schema], "get_weather", [],
                                    error="JSONDecodeError: malformed arguments")
print("Anthropic translation and tool-contract checks passed.")

GET_WEATHER = ToolSpec(
    name="get_weather",
    description="Get the current weather for a city.",
    parameters={
        "type": "object",
        "properties": {
            "city":  {"type": "string", "description": "City name, e.g. 'Seattle'"},
            "units": {"type": "string", "enum": ["c", "f"]},
        },
        "required": ["city"],
        "additionalProperties": False,
    },
)

conv = Conversation(
    system="You can call tools when you need real-world data.",
    messages=[Message(role="user", content="What's the weather in Seattle in Fahrenheit?")],
    tools=[GET_WEATHER],
)

for label, adapter, deployment in [
    ("gpt-5.5",          openai_adapter, OPENAI_DEPLOYMENT),
    ("claude-opus-4-8",  anth_adapter,   ANTH_DEPLOYMENT),
]:
    r = adapter.run(conv, deployment)
    print(f"--- {label} ---")
    print(f"  text       : {r.text[:80]!r}")
    print(f"  tool_calls : {[(c.name, c.arguments) for c in r.tool_calls]}")
    print()

# Complete the Claude tool round trip. Preserve the assistant's tool_use block, then
 # append the canonical tool result; AnthropicAdapter turns it into a user tool_result block.
first_turn = anth_adapter.run(conv, ANTH_DEPLOYMENT)
if not first_turn.tool_calls:
    raise RuntimeError("Claude did not request get_weather; retry with a more explicit tool instruction.")
weather_call = first_turn.tool_calls[0]
continuation = Conversation(
    system=conv.system,
    messages=[
        *conv.messages,
        Message(role="assistant", content=_anthropic_content(first_turn.raw.content)),
        Message(
            role="tool",
            tool_call_id=weather_call.id,
            content=json.dumps({"city": "Seattle", "temperature": 62, "units": "f"}),
        ),
    ],
    tools=[GET_WEATHER],
)
completed_turn = anth_adapter.run(continuation, ANTH_DEPLOYMENT)
print(completed_turn.text)

TICKET_SCHEMA = {
    "type": "object",
    "properties": {
        "category": {"type": "string",
                     "enum": ["hardware", "software", "billing", "other"]},
        "priority": {"type": "string", "enum": ["low", "med", "high"]},
    },
    "required": ["category", "priority"],
    "additionalProperties": False,
}

# Same Conversation -> both adapters. No side-channel helpers, no provider branching
# in user code. The adapter handles strict-mode (OpenAI) vs. validate-and-retry (Anthropic).
conv = Conversation(
    system="Classify support tickets.",
    messages=[Message(role="user",
                      content="Classify this support ticket: 'My laptop won't charge.'")],
    json_schema=TICKET_SCHEMA,
)

r_oai  = openai_adapter.run(conv, OPENAI_DEPLOYMENT)
r_anth = anth_adapter.run(conv,   ANTH_DEPLOYMENT)

print(f"gpt-5.5         -> {r_oai.text}   (retries: {r_oai.retries})")
print(f"claude-opus-4-8 -> {r_anth.text}  (retries: {r_anth.retries})")

from typing import Iterator, Literal

@dataclass
class StreamEvent:
    kind: Literal["text", "tool_call_start", "tool_call_args", "thinking", "done"]
    data: dict

def normalize_openai_stream(resp) -> Iterator[StreamEvent]:
    for chunk in resp:
        if not chunk.choices:
            continue
        delta = chunk.choices[0].delta
        if delta.content:
            yield StreamEvent("text", {"text": delta.content})
        for tc in (delta.tool_calls or []):
            if tc.function and tc.function.name:
                yield StreamEvent("tool_call_start",
                                  {"id": tc.id, "name": tc.function.name})
            if tc.function and tc.function.arguments:
                yield StreamEvent("tool_call_args", {"args": tc.function.arguments})
    yield StreamEvent("done", {})

def normalize_anthropic_stream(resp) -> Iterator[StreamEvent]:
    for event in resp:
        t = getattr(event, "type", None)
        if t == "content_block_start":
            block = event.content_block
            if block.type == "tool_use":
                yield StreamEvent("tool_call_start",
                                  {"id": block.id, "name": block.name})
        elif t == "content_block_delta":
            d = event.delta
            if d.type == "text_delta":
                yield StreamEvent("text", {"text": d.text})
            elif d.type == "input_json_delta":
                yield StreamEvent("tool_call_args", {"args": d.partial_json})
            elif d.type == "thinking_delta":
                yield StreamEvent("thinking", {"text": d.thinking})
    yield StreamEvent("done", {})

# Same prompt AND same system message on both sides. What we're showing is the
# difference in event shape, so anything that changes output length (like only one
# side getting a "be concise" instruction) would muddy the comparison.
SYSTEM = "Be concise."
prompt = "List three things to do in Seattle in summer."

print("--- gpt-5.5 stream ---")
oai_stream = openai_client.chat.completions.create(
    model=OPENAI_DEPLOYMENT,
    messages=[{"role": "system", "content": SYSTEM},
              {"role": "user",   "content": prompt}],
    max_completion_tokens=200, stream=True,
)
for i, ev in enumerate(normalize_openai_stream(oai_stream)):
    if i >= 25: break
    print(f"  {ev.kind:<16} {str(ev.data)[:60]}")

print("\n--- claude-opus-4-8 stream ---")
anth_stream = anth_client.messages.create(
    model=ANTH_DEPLOYMENT, system=SYSTEM,
    messages=[{"role": "user", "content": prompt}],
    max_tokens=200, stream=True,
)
for i, ev in enumerate(normalize_anthropic_stream(anth_stream)):
    if i &gt;= 25: break
    print(f"  {ev.kind:<16} {str(ev.data)[:60]}")

import time, re

EVAL_ROWS = [
    {"id": "text-1",
     "prompt": "In one sentence, what does Microsoft Foundry do?",
     "expected": "Microsoft Foundry is a platform for building, deploying, and managing AI agents and models on Azure."},
    {"id": "text-2",
     "prompt": "Name two benefits of multi-provider model access.",
     "expected": "Avoiding vendor lock-in and matching the best model to each task."},
    {"id": "text-3",
     "prompt": "What is prompt caching used for?",
     "expected": "Reusing long static prompt prefixes to cut input cost and latency."},
    {"id": "tool-1",
     "prompt": "What's the weather in Seattle?",
     "tools": [GET_WEATHER],
     "expected_tool": "get_weather"},
    {"id": "tool-2",
     "prompt": "Tell me the weather in Tokyo in Celsius.",
     "tools": [GET_WEATHER],
     "expected_tool": "get_weather"},
    {"id": "json-1",
     "prompt": "Classify this support ticket: 'My laptop won't charge.'",
     "json_schema": TICKET_SCHEMA},
    {"id": "json-2",
     "prompt": "Classify this support ticket: 'I was double-billed last month.'",
     "json_schema": TICKET_SCHEMA},
    {"id": "text-4",
     "prompt": "What does the Claude `tool_result` content block do?",
     "expected": "Returns the result of a tool call back to the model inside a user message."},
]

def _tokens(s: str) -> set[str]:
    return set(re.findall(r"[a-z0-9]+", (s or "").lower()))

def jaccard(a: str, b: str) -> float:
    A, B = _tokens(a), _tokens(b)
    return len(A & B) / max(1, len(A | B))

def run_row(adapter, deployment, row) -> dict:
    """Every row builds a Conversation and calls adapter.run(). No provider branching."""
    t0 = time.perf_counter()
    if "tools" in row:
        conv = Conversation(
            system="You can call tools when you need real-world data.",
            messages=[Message(role="user", content=row["prompt"])],
            tools=row["tools"],
        )
        try:
            r = adapter.run(conv, deployment)
            out = {"tool_calls": [(c.name, c.arguments) for c in r.tool_calls],
                   "usage": r.usage}
        except Exception as exc:
            # A malformed tool-call payload is a failed contract, not a crashed gate.
            out = {"tool_calls": [], "usage": {},
                   "error": f"{type(exc).__name__}: {exc}"}
    elif "json_schema" in row:
        conv = Conversation(
            system="Be concise and accurate.",
            messages=[Message(role="user", content=row["prompt"])],
            json_schema=row["json_schema"],
        )
        r = adapter.run(conv, deployment)
        out = {"json": json.loads(r.text), "retries": r.retries, "usage": r.usage}
    else:
        conv = Conversation(
            system="Be concise and accurate.",
            messages=[Message(role="user", content=row["prompt"])],
        )
        r = adapter.run(conv, deployment)
        out = {"text": r.text, "usage": r.usage}
    out["latency_s"] = round(time.perf_counter() - t0, 2)
    return out

results = {"openai": {}, "anthropic": {}}
for row in EVAL_ROWS:
    results["openai"][row["id"]]    = run_row(openai_adapter, OPENAI_DEPLOYMENT, row)
    results["anthropic"][row["id"]] = run_row(anth_adapter,   ANTH_DEPLOYMENT,    row)

print(f"{'id':<8} {'kind':<6} {'gpt-5.5 lat':<13} {'opus-4-8 lat':<14} preview (candidate)")
print("-" * 80)
for row in EVAL_ROWS:
    o = results["openai"][row["id"]]
    a = results["anthropic"][row["id"]]
    kind = "tool" if "tools" in row else "json" if "json_schema" in row else "text"
    preview = a.get("text") or a.get("json") or a.get("tool_calls", "")
    print(f"{row['id']:<8} {kind:<6} {o['latency_s']:<13} {a['latency_s']:<14} {str(preview)[:55]}")

import statistics

# 1. Schema parity (hard): both providers must emit the expected tool with
# arguments that validate against its declared JSON Schema.
tool_rows = [r for r in EVAL_ROWS if "tools" in r]
schema_parity_hits = []

for row in tool_rows:
    o_names = {n for n, _ in results["openai"][row["id"]]["tool_calls"]}
    a_names = {n for n, _ in results["anthropic"][row["id"]]["tool_calls"]}
    openai_result = results["openai"][row["id"]]
    anthropic_result = results["anthropic"][row["id"]]
    openai_contract = tool_calls_meet_contract(
        row["tools"], row["expected_tool"], openai_result.get("tool_calls", []),
        openai_result.get("error"),
    )
    anthropic_contract = tool_calls_meet_contract(
        row["tools"], row["expected_tool"], anthropic_result.get("tool_calls", []),
        anthropic_result.get("error"),
    )
    schema_parity_hits.append(openai_contract and anthropic_contract and o_names == a_names)
schema_parity_rate = (sum(schema_parity_hits) / len(schema_parity_hits)
                      if schema_parity_hits else 1.0)

# 2. Structured-output retry rate (hard).
json_rows = [r for r in EVAL_ROWS if "json_schema" in r]
retry_rate = (
    sum(results["anthropic"][r["id"]].get("retries", 0) for r in json_rows)
    / max(1, len(json_rows))
)

# 3. Quality on plain text (soft): Jaccard token overlap vs `expected`.
text_rows = [r for r in EVAL_ROWS if "expected" in r]
baseline_q  = statistics.mean(
    jaccard(results["openai"][r["id"]]["text"], r["expected"]) for r in text_rows
)
candidate_q = statistics.mean(
    jaccard(results["anthropic"][r["id"]]["text"], r["expected"]) for r in text_rows
)

# 4. Latency P90 across all rows.
def p90(xs): return sorted(xs)[max(0, int(0.9 * len(xs)) - 1)]
base_p90 = p90([results["openai"][r["id"]]["latency_s"]    for r in EVAL_ROWS])
cand_p90 = p90([results["anthropic"][r["id"]]["latency_s"] for r in EVAL_ROWS])

gates = [
    ("Schema parity (tool rows)",       schema_parity_rate == 1.0,
     f"{schema_parity_rate*100:.0f}% match"),
    ("Structured retry rate (cap 10%)", retry_rate <= 0.10,
     f"{retry_rate*100:.0f}% retries"),
    ("Quality (>= baseline - 0.05)",    candidate_q >= baseline_q - 0.05,
     f"baseline {baseline_q:.2f} | candidate {candidate_q:.2f}"),
    ("P90 latency (<= 1.3x baseline)",  cand_p90 &lt;= base_p90 * 1.3,
     f"baseline {base_p90:.2f}s | candidate {cand_p90:.2f}s"),
]

print("=" * 72)
print(f"  {'GATE':<36} {'STATUS':<8} DETAIL")
print("-" * 72)
for name, passed, detail in gates:
    print(f"  {name:<36} {'PASS' if passed else 'FAIL':<8} {detail}")
print("-" * 72)
overall = all(p for _, p, _ in gates)
print(f"  OVERALL: {'PASS. Safe to start the side-by-side test on real traffic.' if overall else 'FAIL. Do not promote.'}")
print("=" * 72)

# Latency diagnostic: if the P90 gate failed, find which row dominates.
# Shows per-row latency for both providers plus the P50/P90 spread.

print(f"  {'id':<8} {'kind':<6} {'gpt-5.5':<10} {'opus-4-8':<10} ratio")
print("-" * 50)
for row in EVAL_ROWS:
    rid  = row["id"]
    o    = results["openai"][rid]["latency_s"]
    a    = results["anthropic"][rid]["latency_s"]
    kind = "tool" if "tools" in row else "json" if "json_schema" in row else "text"
    ratio = a / o if o &gt; 0 else float("inf")
    print(f"  {rid:<8} {kind:<6} {o:<10.2f} {a:<10.2f} {ratio:.1f}x")

# Compare medians to means: if P50 is fine but P90 fails, the tail is the problem.
import statistics
oai_lat  = [results["openai"][r["id"]]["latency_s"]    for r in EVAL_ROWS]
anth_lat = [results["anthropic"][r["id"]]["latency_s"] for r in EVAL_ROWS]
print()
print(f"  P50  gpt-5.5 {statistics.median(oai_lat):.2f}s   opus-4-8 {statistics.median(anth_lat):.2f}s")
print(f"  P90  gpt-5.5 {base_p90:.2f}s              opus-4-8 {cand_p90:.2f}s")
print(f"  Ratio P50 = {statistics.median(anth_lat)/statistics.median(oai_lat):.1f}x   "
      f"Ratio P90 = {cand_p90/base_p90:.1f}x")