From 11ba4ef51db31f994f7012ca9c02be080213c2af Mon Sep 17 00:00:00 2001 From: Jeremy Schoemaker Date: Wed, 5 Aug 2026 11:39:04 -0500 Subject: [PATCH] fix: let streaming failures reach the retry loop `send_request` implements retry/backoff by catching exceptions around the provider call. `_handle_stream_request` wraps its own body in a second try/except and converts any failure into `return "failed", {...}`. A return is not an exception, so it walks straight past the enclosing retry loop: the `for retry_attempt in range(self.max_retries)` exits on its first iteration no matter what `--retries` is set to. Non-stream requests, whose exceptions propagate normally, get the full budget. That is the same class of failure the retry loop exists for -- a 429 or 5xx on connect, or a provider that starts the SSE stream and dies mid-iteration. Removing the inner handler lets it propagate. The error is still logged and still returns "failed" once the attempts are exhausted, by the existing code in send_request. Counted provider calls for a mid-stream failure with max_retries=3: before: [stream] 1 attempt [nonstream] 3 attempts after: [stream] 3 attempts [nonstream] 3 attempts The diff is mostly the dedent; `git diff -w` shows the four lines that actually changed. --- verify.py | 152 ++++++++++++++++++++++++++---------------------------- 1 file changed, 74 insertions(+), 78 deletions(-) diff --git a/verify.py b/verify.py index 50813bf..e8ca548 100644 --- a/verify.py +++ b/verify.py @@ -277,84 +277,80 @@ async def send_request(self, request: dict) -> tuple[str, dict]: async def _handle_stream_request(self, request: dict) -> tuple[str, dict]: """Handle streaming request.""" - try: - stream = await self.client.chat.completions.create(**request, extra_body=self.extra_body) - - request_id = None - created = None - full_content = [] - tool_calls: dict[int, dict] = {} - finish_reason = None - usage = None - provider = None - - async for event in stream: - if hasattr(event, 'id') and event.id: - request_id = event.id - if hasattr(event, 'created') and event.created: - created = event.created - - if hasattr(event, 'provider') and event.provider: - provider = event.provider - - if not hasattr(event, 'choices') or not event.choices: - logger.warning("Empty choices in stream event") - continue - - choice = event.choices[0] - - if hasattr(choice, 'delta') and choice.delta: - if hasattr(choice.delta, 'content') and choice.delta.content: - full_content.append(choice.delta.content) - - if hasattr(choice.delta, 'tool_calls') and choice.delta.tool_calls: - for tc in choice.delta.tool_calls: - idx = tc.index if tc.index is not None else 0 - - if idx not in tool_calls: - tool_calls[idx] = { - "id": tc.id, - "type": tc.type, - "function": {"name": "", "arguments": ""}, - } - - if hasattr(tc, 'function') and tc.function: - if hasattr(tc.function, 'name') and tc.function.name: - tool_calls[idx]["function"]["name"] = tc.function.name - if hasattr(tc.function, 'arguments') and tc.function.arguments: - tool_calls[idx]["function"]["arguments"] += tc.function.arguments - - if hasattr(choice, 'finish_reason') and choice.finish_reason: - finish_reason = choice.finish_reason - - if hasattr(choice, 'usage') and choice.usage: - usage = choice.usage - - response = { - "id": request_id, - "object": "chat.completion", - "created": created, - "model": request.get("model", ""), - "choices": [ - { - "index": 0, - "message": { - "role": "assistant", - "content": "".join(full_content), - "tool_calls": ( - list(tool_calls.values()) if tool_calls else None - ), - }, - "finish_reason": finish_reason or "stop", - } - ], - "usage": usage, - "provider": provider - } - return "success", response - except Exception as e: - logger.error(f"Stream request failed: {e}") - return "failed", {"error": str(e)} + stream = await self.client.chat.completions.create(**request, extra_body=self.extra_body) + + request_id = None + created = None + full_content = [] + tool_calls: dict[int, dict] = {} + finish_reason = None + usage = None + provider = None + + async for event in stream: + if hasattr(event, 'id') and event.id: + request_id = event.id + if hasattr(event, 'created') and event.created: + created = event.created + + if hasattr(event, 'provider') and event.provider: + provider = event.provider + + if not hasattr(event, 'choices') or not event.choices: + logger.warning("Empty choices in stream event") + continue + + choice = event.choices[0] + + if hasattr(choice, 'delta') and choice.delta: + if hasattr(choice.delta, 'content') and choice.delta.content: + full_content.append(choice.delta.content) + + if hasattr(choice.delta, 'tool_calls') and choice.delta.tool_calls: + for tc in choice.delta.tool_calls: + idx = tc.index if tc.index is not None else 0 + + if idx not in tool_calls: + tool_calls[idx] = { + "id": tc.id, + "type": tc.type, + "function": {"name": "", "arguments": ""}, + } + + if hasattr(tc, 'function') and tc.function: + if hasattr(tc.function, 'name') and tc.function.name: + tool_calls[idx]["function"]["name"] = tc.function.name + if hasattr(tc.function, 'arguments') and tc.function.arguments: + tool_calls[idx]["function"]["arguments"] += tc.function.arguments + + if hasattr(choice, 'finish_reason') and choice.finish_reason: + finish_reason = choice.finish_reason + + if hasattr(choice, 'usage') and choice.usage: + usage = choice.usage + + response = { + "id": request_id, + "object": "chat.completion", + "created": created, + "model": request.get("model", ""), + "choices": [ + { + "index": 0, + "message": { + "role": "assistant", + "content": "".join(full_content), + "tool_calls": ( + list(tool_calls.values()) if tool_calls else None + ), + }, + "finish_reason": finish_reason or "stop", + } + ], + "usage": usage, + "provider": provider + } + return "success", response async def process_request(self, prepared_req: dict, data_index: int) -> dict: """Process a single request and run all validators."""