Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 72 additions & 0 deletions tests/v1/test_e2e.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
52 changes: 52 additions & 0 deletions verifiers/v1/harnesses/utils/core.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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
Expand All @@ -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


Expand Down
Loading