From 4f30b22640f7aa0a0c1927f2a2d45ee5c9131aad Mon Sep 17 00:00:00 2001 From: Mika Senghaas Date: Tue, 15 Sep 2026 02:37:34 +0000 Subject: [PATCH 1/2] fix streamed reasoning replay The OpenAI stream accumulator concatenates repeated string metadata in provider-specific reasoning details. Preserve scalar fields while joining reasoning text so tool requests replay the message that the interceptor recorded. --- tests/v1/test_e2e.py | 72 ++++++++++++++++++++++++++++ verifiers/v1/harnesses/utils/core.py | 52 ++++++++++++++++++++ 2 files changed, 124 insertions(+) diff --git a/tests/v1/test_e2e.py b/tests/v1/test_e2e.py index ee2dadd742..7ac1e7a619 100644 --- a/tests/v1/test_e2e.py +++ b/tests/v1/test_e2e.py @@ -18,6 +18,78 @@ 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": { + "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["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..485d6ebc16 100644 --- a/verifiers/v1/harnesses/utils/core.py +++ b/verifiers/v1/harnesses/utils/core.py @@ -181,6 +181,49 @@ def run_edit(path: str, old_str: str, new_str: str) -> str: return f"Edited {path}" +_REASONING_FIELDS = ("reasoning", "reasoning_content", "reasoning_details") + + +def _accumulate_reasoning(accumulated: dict, delta: dict) -> None: + """Merge reasoning text and structured blocks from chat completion deltas.""" + 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 +244,16 @@ async def chat( raw_stream=raw_stream, response_format=omit, input_tools=[] ) as response: completion = None + reasoning: 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 _REASONING_FIELDS): + _accumulate_reasoning( + reasoning.setdefault(choice.index, {}), delta + ) if ( completion is None or not completion.choices @@ -214,6 +264,8 @@ async def chat( # 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" + for field_name, value in reasoning.get(choice.index, {}).items(): + setattr(choice.message, field_name, value) return completion From b83c24f8a7fe80f471a2b74eb1c378a334482080 Mon Sep 17 00:00:00 2001 From: Mika Senghaas Date: Tue, 15 Sep 2026 04:44:39 +0000 Subject: [PATCH 2/2] unify streamed field accumulation --- tests/v1/test_e2e.py | 2 ++ verifiers/v1/harnesses/utils/core.py | 31 ++++++++++++++++++---------- 2 files changed, 22 insertions(+), 11 deletions(-) diff --git a/tests/v1/test_e2e.py b/tests/v1/test_e2e.py index 7ac1e7a619..3737831cef 100644 --- a/tests/v1/test_e2e.py +++ b/tests/v1/test_e2e.py @@ -37,6 +37,7 @@ def chunk(text: str, finish_reason: str | None = None) -> dict: { "index": 0, "delta": { + "role": "assistant", "reasoning": text, "reasoning_content": text, "reasoning_details": [ @@ -76,6 +77,7 @@ async def respond(request: httpx.Request) -> httpx.Response: 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"] == [ diff --git a/verifiers/v1/harnesses/utils/core.py b/verifiers/v1/harnesses/utils/core.py index 485d6ebc16..bdd2793ff3 100644 --- a/verifiers/v1/harnesses/utils/core.py +++ b/verifiers/v1/harnesses/utils/core.py @@ -181,11 +181,19 @@ def run_edit(path: str, old_str: str, new_str: str) -> str: return f"Edited {path}" -_REASONING_FIELDS = ("reasoning", "reasoning_content", "reasoning_details") +_STREAMED_MESSAGE_FIELDS = ( + "role", + "reasoning", + "reasoning_content", + "reasoning_details", +) -def _accumulate_reasoning(accumulated: dict, delta: dict) -> None: - """Merge reasoning text and structured blocks from chat completion deltas.""" +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 @@ -244,15 +252,17 @@ async def chat( raw_stream=raw_stream, response_format=omit, input_tools=[] ) as response: completion = None - reasoning: dict[int, dict] = {} + 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 _REASONING_FIELDS): - _accumulate_reasoning( - reasoning.setdefault(choice.index, {}), delta + 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 @@ -261,10 +271,9 @@ 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" - for field_name, value in reasoning.get(choice.index, {}).items(): + 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