wehub-resource-sync
213 lines
8.1 KiB
Python
213 lines
8.1 KiB
Python
"""Client-side tool result threading across turns.
|
|
|
|
Turn 1 forces a tunneled client-tool call, posts a marker payload as
|
|
the result, and the model must surface it. Turn 2 asks for the value
|
|
with no tool attached: the answer can only come from the turn-1
|
|
transcript, so this fails if tool outputs don't thread into the next
|
|
dispatch's context on the harness under test.
|
|
|
|
Turn 1 is driven over ``GET /v1/sessions/{id}/stream``: on the harness
|
|
path, ``action_required`` function_calls are published to the live
|
|
stream only and never persisted as conversation items (see
|
|
``tests/e2e/test_openai_coder_client_tools.py``), so snapshot polling
|
|
would hang waiting for an item that never appears.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import threading
|
|
import uuid
|
|
from collections.abc import Iterator
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from tests.e2e.conftest import configure_mock_llm, send_user_message_to_session
|
|
from tests.integration.conftest import JourneySession
|
|
from tests.integration.helpers import all_message_text, failure_detail, run_turn
|
|
|
|
_LOOKUP_WIDGET_TOOL: dict[str, Any] = {
|
|
"type": "function",
|
|
"function": {
|
|
"name": "lookup_widget",
|
|
"description": "Look up a widget by id and return its color.",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"widget_id": {"type": "integer", "description": "Widget id, e.g. 42."},
|
|
},
|
|
"required": ["widget_id"],
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
def _iter_sse(response: httpx.Response) -> Iterator[dict[str, Any]]:
|
|
"""Yield decoded SSE event dicts; stops at ``[DONE]``.
|
|
|
|
:param response: Open streaming response from
|
|
``GET /v1/sessions/{id}/stream``.
|
|
"""
|
|
buffer = ""
|
|
for chunk in response.iter_text():
|
|
buffer += chunk
|
|
while "\n\n" in buffer:
|
|
frame, _, buffer = buffer.partition("\n\n")
|
|
data_line = next(
|
|
(line for line in frame.splitlines() if line.startswith("data:")), None
|
|
)
|
|
if data_line is None:
|
|
continue
|
|
payload = data_line[len("data:") :].strip()
|
|
if payload == "[DONE]":
|
|
return
|
|
try:
|
|
yield json.loads(payload)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
|
|
|
|
def _turn_with_tool(
|
|
base_url: str, session_id: str, prompt: str, marker: str, timeout_s: float = 60
|
|
) -> str:
|
|
"""Run one turn, answering the first ``lookup_widget`` call with *marker*.
|
|
|
|
:param base_url: Live server base URL.
|
|
:param session_id: Runner-bound session id.
|
|
:param prompt: User message for turn 1.
|
|
:param marker: Payload to return as the tool result.
|
|
:param timeout_s: Stream budget; with turn 2's 50s poll this
|
|
keeps each rerun attempt under the CI 180s cap.
|
|
:returns: The assistant text assembled from stream deltas.
|
|
:raises AssertionError: If the model never calls the tool or the
|
|
turn does not complete.
|
|
"""
|
|
called = threading.Event()
|
|
text_chunks: list[str] = []
|
|
status: str | None = None
|
|
# Poster threads park failures here; raised after the stream loop.
|
|
# A daemon thread's exception is otherwise swallowed and the test
|
|
# would time out with a misleading "model never called" error.
|
|
errors: list[Exception] = []
|
|
|
|
def _post_message() -> None:
|
|
try:
|
|
with httpx.Client(base_url=base_url, timeout=30) as poster:
|
|
send_user_message_to_session(
|
|
poster, session_id=session_id, content=prompt, tools=[_LOOKUP_WIDGET_TOOL]
|
|
)
|
|
except Exception as exc: # thread boundary; re-raised below
|
|
errors.append(exc)
|
|
|
|
def _post_output(call_id: str) -> None:
|
|
try:
|
|
with httpx.Client(base_url=base_url, timeout=30) as poster:
|
|
resp = poster.post(
|
|
f"/v1/sessions/{session_id}/events",
|
|
json={
|
|
"type": "function_call_output",
|
|
"data": {"call_id": call_id, "output": json.dumps({"color": marker})},
|
|
},
|
|
)
|
|
assert resp.status_code in (200, 202), (
|
|
f"function_call_output POST failed: {resp.status_code} {resp.text[:300]}"
|
|
)
|
|
except Exception as exc: # thread boundary; re-raised below
|
|
errors.append(exc)
|
|
|
|
with httpx.Client(base_url=base_url, timeout=timeout_s) as streamer:
|
|
with streamer.stream("GET", f"/v1/sessions/{session_id}/stream") as response:
|
|
response.raise_for_status()
|
|
posted = False
|
|
for event in _iter_sse(response):
|
|
# Post only after the first frame: the stream does not
|
|
# replay history, so posting earlier can drop events.
|
|
if not posted:
|
|
threading.Thread(target=_post_message, daemon=True).start()
|
|
posted = True
|
|
etype = event.get("type")
|
|
if etype == "response.output_item.done":
|
|
item = event.get("item") or {}
|
|
if (
|
|
item.get("type") == "function_call"
|
|
and item.get("name") == "lookup_widget"
|
|
and item.get("status") == "action_required"
|
|
and not called.is_set()
|
|
):
|
|
called.set()
|
|
threading.Thread(
|
|
target=_post_output, args=(item["call_id"],), daemon=True
|
|
).start()
|
|
elif etype == "response.output_text.delta":
|
|
delta = event.get("delta")
|
|
if isinstance(delta, str):
|
|
text_chunks.append(delta)
|
|
elif etype in ("response.completed", "response.failed"):
|
|
status = etype
|
|
break
|
|
|
|
if errors:
|
|
raise errors[0]
|
|
assert called.is_set(), (
|
|
f"model never called lookup_widget; turn ended {status!r} "
|
|
f"with text: {''.join(text_chunks)[:500]}"
|
|
)
|
|
assert status == "response.completed", f"turn 1 ended {status!r}"
|
|
return "".join(text_chunks)
|
|
|
|
|
|
def test_client_tool_result_recalled_across_turns(
|
|
live_server: str,
|
|
journey_session: JourneySession,
|
|
mock_llm_server_url: str | None,
|
|
) -> None:
|
|
marker = f"widget-color-{uuid.uuid4().hex[:8]}"
|
|
call_id = f"call_{uuid.uuid4().hex[:8]}"
|
|
sid = journey_session.session_id
|
|
|
|
# In mock mode, queue: tool call → text with marker → recall text.
|
|
# The first response is a tool call; after the test posts the tool
|
|
# result, the second response includes the marker; the third
|
|
# response (turn 2) recalls from transcript.
|
|
configure_mock_llm(
|
|
mock_llm_server_url,
|
|
[
|
|
{
|
|
"tool_calls": [
|
|
{
|
|
"call_id": call_id,
|
|
"name": "lookup_widget",
|
|
"arguments": json.dumps({"widget_id": 42}),
|
|
},
|
|
],
|
|
},
|
|
{"text": f"The widget color is {marker}."},
|
|
{"text": f"The color was {marker}."},
|
|
],
|
|
)
|
|
|
|
text_1 = _turn_with_tool(
|
|
live_server,
|
|
sid,
|
|
prompt=(
|
|
"You MUST call the lookup_widget tool for widget id 42 "
|
|
"before answering. Then reply with the color it returns."
|
|
),
|
|
marker=marker,
|
|
)
|
|
# The marker only exists in the tool result we posted; seeing it in
|
|
# the reply proves the result reached the model.
|
|
assert marker in text_1, f"tool result never reached the model; turn 1 said: {text_1[:500]}"
|
|
|
|
with httpx.Client(base_url=live_server, timeout=300) as client:
|
|
body_2 = run_turn(
|
|
client,
|
|
session_id=sid,
|
|
content="What color was the widget? Reply with the exact color value only.",
|
|
)
|
|
assert body_2["status"] == "completed", f"turn 2 failed: {failure_detail(body_2)}"
|
|
# No tool is attached on turn 2: the only source of the marker is
|
|
# the turn-1 transcript (tool call + result) replayed as context.
|
|
assert marker in all_message_text(body_2), failure_detail(body_2)
|