diff --git a/tests/v1/test_e2e.py b/tests/v1/test_e2e.py index ee2dadd742..3737831cef 100644 --- a/tests/v1/test_e2e.py +++ b/tests/v1/test_e2e.py @@ -18,6 +18,80 @@ def pair(a: str, b: str, id: str, *extra_marks): return pytest.param(a, b, marks=[*marks, *extra_marks], id=id) +@pytest.mark.asyncio +async def test_chat_harness_preserves_streamed_reasoning(): + import json + + import httpx + from openai import AsyncOpenAI + + from verifiers.v1.harnesses.utils.core import chat + + def chunk(text: str, finish_reason: str | None = None) -> dict: + return { + "id": "chatcmpl-test", + "object": "chat.completion.chunk", + "created": 0, + "model": "test-model", + "choices": [ + { + "index": 0, + "delta": { + "role": "assistant", + "reasoning": text, + "reasoning_content": text, + "reasoning_details": [ + { + "type": "reasoning.text", + "index": 0, + "id": "r1", + "format": "unknown", + "signature": "sig", + "text": text, + } + ], + }, + "finish_reason": finish_reason, + } + ], + } + + events = [chunk("Plan: "), chunk("call ls", "stop")] + content = "".join(f"data: {json.dumps(event)}\n\n" for event in events) + content += "data: [DONE]\n\n" + + async def respond(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 200, + content=content, + headers={"content-type": "text/event-stream"}, + request=request, + ) + + async with httpx.AsyncClient(transport=httpx.MockTransport(respond)) as http_client: + client = AsyncOpenAI( + api_key="test", + base_url="https://example.test/v1", + http_client=http_client, + ) + completion = await chat(client, "test-model", [], []) + + message = completion.choices[0].message.model_dump(exclude_none=True) + assert message["role"] == "assistant" + assert message["reasoning"] == "Plan: call ls" + assert message["reasoning_content"] == "Plan: call ls" + assert message["reasoning_details"] == [ + { + "type": "reasoning.text", + "index": 0, + "id": "r1", + "format": "unknown", + "signature": "sig", + "text": "Plan: call ls", + } + ] + + # harness x harness runtime: every harness once, both local runtimes hit (subprocess # only carries the in-house loops — the rest NEEDS_CONTAINER), one remote row per # provider. codex/claude-code are excluded here (unreliable on a no-op echo chat diff --git a/verifiers/v1/harnesses/utils/core.py b/verifiers/v1/harnesses/utils/core.py index 743131c55c..bdd2793ff3 100644 --- a/verifiers/v1/harnesses/utils/core.py +++ b/verifiers/v1/harnesses/utils/core.py @@ -181,6 +181,57 @@ def run_edit(path: str, old_str: str, new_str: str) -> str: return f"Edited {path}" +_STREAMED_MESSAGE_FIELDS = ( + "role", + "reasoning", + "reasoning_content", + "reasoning_details", +) + + +def _accumulate_streamed_message(accumulated: dict, delta: dict) -> None: + """Accumulate message fields whose stream semantics differ from the SDK defaults.""" + if role := delta.get("role"): + accumulated["role"] = role + + for field_name in ("reasoning", "reasoning_content"): + if value := delta.get(field_name): + accumulated[field_name] = accumulated.get(field_name, "") + value + + delta_details = delta.get("reasoning_details") or [] + if not delta_details: + return + reasoning_details = accumulated.setdefault("reasoning_details", []) + for detail in delta_details: + previous = reasoning_details[-1] if reasoning_details else {} + detail_type = detail.get("type") + content_field = { + "reasoning.summary": "summary", + "reasoning.text": "text", + }.get(detail_type) + if ( + content_field + and detail_type == previous.get("type") + and all( + previous.get(field_name) is None + or detail.get(field_name) is None + or previous[field_name] == detail[field_name] + for field_name in ("id", "index", "format") + ) + ): + previous[content_field] = (previous.get(content_field) or "") + ( + detail.get(content_field) or "" + ) + for field_name in ("id", "index", "signature", "format"): + if ( + previous.get(field_name) is None + and detail.get(field_name) is not None + ): + previous[field_name] = detail[field_name] + else: + reasoning_details.append(dict(detail)) + + async def chat( client: AsyncOpenAI, model: str, @@ -201,9 +252,18 @@ async def chat( raw_stream=raw_stream, response_format=omit, input_tools=[] ) as response: completion = None + message_overrides: dict[int, dict] = {} async for event in response: if event.type == "chunk": completion = event.snapshot + for choice in event.chunk.choices: + delta = choice.delta.model_dump(exclude_none=True) + if any( + delta.get(field_name) for field_name in _STREAMED_MESSAGE_FIELDS + ): + _accumulate_streamed_message( + message_overrides.setdefault(choice.index, {}), delta + ) if ( completion is None or not completion.choices @@ -211,9 +271,10 @@ async def chat( ): raise RuntimeError("model stream ended before a completion finished") for choice in completion.choices: - # Some providers repeat the role in each delta. The SDK concatenates - # these strings, but the role is metadata, not incremental content. - choice.message.role = "assistant" + overrides = message_overrides.setdefault(choice.index, {}) + overrides.setdefault("role", "assistant") + for field_name, value in overrides.items(): + setattr(choice.message, field_name, value) return completion