fix(e2e): derive mock stage from request body, not a call counter

Review feedback: the global _STATE["calls"] counter was incremented
unlocked and before the response was emitted. Two failure modes follow —
ThreadingHTTPServer can serve requests concurrently, and an aborted or
retried call consumed the tool stage, leaving the retry to receive the
final message instead. Reproduced the latter: aborting call 1 mid-stream
made the retry return output_text rather than re-emitting function_call,
which would surface as a bare 'no tool spans' failure.

The conversation already carries the stage, so read it instead of
counting: turn 1 has only the user message, turn 2 carries the tool
result (function_call_output on the Responses wire, role: "tool" on
Chat Completions). That is idempotent — a retry resends the same body
and gets the same stage — so the mutable global goes away entirely
rather than being guarded by a lock.

Verified: the retry now re-emits function_call; both wires stage
correctly on turns 1 and 2; mock E2E passes on latest, v2026.7.30 and
v2026.7.20; wheel and real-Opik E2E pass; 113 unit tests pass.

Author: JetoPistola

e2e/mock_llm_server.py

@@ -17,9 +17,11 @@
 
 It also probes GET /v1/models and POST /api/show (Ollama-style).
 
-Deterministic two-step interaction, identical on either protocol:
-  call 1  -> a tool call      (Hermes runs a tool  -> tool span)
-  call 2+ -> assistant text   (turn completes)
+Deterministic two-step interaction, identical on either protocol. The stage is
+derived from the request body (has the tool result come back yet?), not from a
+call counter, so it is both concurrency-safe and stable across retries:
+  no tool result yet -> a tool call     (Hermes runs a tool -> tool span)
+  tool result present -> assistant text (turn completes)
 
 That yields one LLM -> tool -> LLM cycle: a root trace with two LLM spans and
 one tool span. No real model, no key, no external network.
@@ -36,10 +38,36 @@
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 
 PORT = int(os.environ.get("MOCK_LLM_PORT", "18790"))
-_STATE = {"calls": 0}
 _USAGE = {"prompt_tokens": 20, "completion_tokens": 8, "total_tokens": 28}
 
 
+def _already_ran_tool(payload: bytes) -> bool:
+    """Has the tool already run in this conversation?
+
+    The stage is derived from the request body rather than a call counter.
+    A counter is wrong on two axes: `ThreadingHTTPServer` can serve requests
+    concurrently, and a retried/aborted call would consume the tool stage and
+    leave the retry to get the final message — silently producing a run with
+    no tool span. Reading the conversation is idempotent, so a retry of the
+    same request always gets the same stage back.
+
+    Turn 1 carries only the user message; turn 2 carries the tool result
+    (Responses: a `function_call_output` input item; Chat Completions: a
+    `role: "tool"` message).
+    """
+    try:
+        data = json.loads(payload or b"{}")
+    except ValueError:
+        return False
+
+    for item in data.get("input") or data.get("messages") or []:
+        if not isinstance(item, dict):
+            continue
+        if item.get("type") == "function_call_output" or item.get("role") == "tool":
+            return True
+    return False
+
+
 def _tool_call_chunks():
     """Stream a single tool_call, then finish_reason=tool_calls."""
     base = {"id": "chatcmpl-mock1", "object": "chat.completion.chunk", "model": "gpt-5"}
@@ -186,11 +214,10 @@ def do_POST(self):
 
         # Responses API (codex_responses api_mode).
         if self.path.rstrip("/").endswith("/responses"):
-            _STATE["calls"] += 1
             events = (
-                _responses_tool_events()
-                if _STATE["calls"] == 1
-                else _responses_final_events()
+                _responses_final_events()
+                if _already_ran_tool(body)
+                else _responses_tool_events()
             )
             wants_stream = b'"stream": true' in body or b'"stream":true' in body
             if wants_stream:
@@ -204,8 +231,7 @@ def do_POST(self):
 
         # Chat Completions (the other wire Hermes may pick).
         if "chat/completions" in self.path:
-            _STATE["calls"] += 1
-            chunks = _tool_call_chunks() if _STATE["calls"] == 1 else _final_chunks()
+            chunks = _final_chunks() if _already_ran_tool(body) else _tool_call_chunks()
             wants_stream = b'"stream": true' in body or b'"stream":true' in body
             if wants_stream:
                 self._sse(chunks)