From 01b937c42ee042a8026fa39e3acb0abd08a6d84d Mon Sep 17 00:00:00 2001 From: Glenn Jocher Date: Sat, 15 Aug 2026 12:55:44 +0200 Subject: [PATCH] Repair malformed structured output in the review agent (#872) --- .gitignore | 3 + actions/__init__.py | 2 +- actions/review_pr.py | 15 ++-- actions/utils/github_utils.py | 12 +++ actions/utils/openai_utils.py | 126 ++++++++++++++++++-------------- tests/test_first_interaction.py | 1 + tests/test_openai_utils.py | 86 ++++++++++++++++++---- 7 files changed, 171 insertions(+), 74 deletions(-) diff --git a/.gitignore b/.gitignore index b00e58d..3e50bc4 100644 --- a/.gitignore +++ b/.gitignore @@ -109,6 +109,9 @@ ipython_config.py # https://pdm.fming.dev/#use-with-ide .pdm.toml +# uv +uv.lock + # PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm __pypackages__/ diff --git a/actions/__init__.py b/actions/__init__.py index d874548..b66fe7f 100644 --- a/actions/__init__.py +++ b/actions/__init__.py @@ -28,4 +28,4 @@ # ├── test_summarize_pr.py # └── ... -__version__ = "0.3.6" +__version__ = "0.3.7" diff --git a/actions/review_pr.py b/actions/review_pr.py index ca52710..85fc095 100644 --- a/actions/review_pr.py +++ b/actions/review_pr.py @@ -852,8 +852,9 @@ def clear_previous_review(event: Action) -> dict: hard=True, ) for comment in comments: - if comment.get("pull_request_review_id") in owned_reviews: - event.delete(f"{GITHUB_API_URL}/repos/{event.repository}/pulls/comments/{comment['id']}", hard=True) + if comment.get("pull_request_review_id") in owned_reviews: # 404: already deleted by a person or another run + url = f"{GITHUB_API_URL}/repos/{event.repository}/pulls/comments/{comment['id']}" + event.delete(url, expected_status=[200, 204, 404], hard=True) return history @@ -933,11 +934,15 @@ def post_review_summary(event: Action, review_data: dict, review_number: int = 1 review_comments.append(review_comment) # Submit review with inline comments + url = f"{GITHUB_API_URL}/repos/{event.repository}/pulls/{pr_number}/reviews" payload = {"commit_id": commit_sha, "body": body.strip(), "event": event_type} if review_comments: - payload["comments"] = review_comments - - event.post(f"{GITHUB_API_URL}/repos/{event.repository}/pulls/{pr_number}/reviews", json=payload, hard=True) + # One anchor GitHub cannot resolve rejects the whole review; the body logs every finding, so post it alone + response = event.post(url, json={**payload, "comments": review_comments}, expected_status=[200, 422], hard=True) + if response.status_code != 422: + return + print("GitHub rejected the inline comments; posting the review body alone") + event.post(url, json=payload, hard=True) def run_review(event: Action, pr_title: str, pr_description: str) -> None: diff --git a/actions/utils/github_utils.py b/actions/utils/github_utils.py index 6ad5f58..f4f95df 100644 --- a/actions/utils/github_utils.py +++ b/actions/utils/github_utils.py @@ -7,6 +7,8 @@ import os from pathlib import Path import requests +from requests.adapters import HTTPAdapter +from urllib3.util.retry import Retry from actions import __version__ from actions.utils.common_utils import filter_diff_text @@ -126,6 +128,16 @@ class Action: self.headers = {"Authorization": f"Bearer {self.token}", "Accept": "application/vnd.github+json"} self.headers_diff = {"Authorization": f"Bearer {self.token}", "Accept": "application/vnd.github.v3.diff"} self.session = requests.Session() + # Retry transient GitHub failures: connection errors for every method, 5xx only for idempotent methods so a + # retried POST can never double-post; the final response still flows through the status checks below + retry = Retry( + total=3, + backoff_factor=1, + status_forcelist=[500, 502, 503, 504], + allowed_methods=["GET", "PUT", "DELETE"], + raise_on_status=False, + ) + self.session.mount("https://", HTTPAdapter(max_retries=retry)) self.verbose = verbose self.eyes_reaction_id = None self._pr_diff_cache = None diff --git a/actions/utils/openai_utils.py b/actions/utils/openai_utils.py index 2060654..bb9e287 100644 --- a/actions/utils/openai_utils.py +++ b/actions/utils/openai_utils.py @@ -360,11 +360,16 @@ def _handle_function_call(call: dict, tool_handlers: dict[str, Callable]) -> dic """Execute one model-requested function call and return a Responses API output item.""" name = call.get("name") call_id = call.get("call_id") - if name not in tool_handlers: - raise KeyError(f"Unknown tool: {name}") - output = tool_handlers[name](**_parse_tool_arguments(call)) - if not isinstance(output, str): - output = json.dumps(output) + try: + if name not in tool_handlers: + raise KeyError(f"Unknown tool: {name}") + output = tool_handlers[name](**_parse_tool_arguments(call)) + if not isinstance(output, str): + output = json.dumps(output) + except Exception as e: + # The model owns a failed call: it can correct the arguments, use another tool, or report the evidence gap + print(f"Tool {name} failed: {e!r}") + output = f"Tool {name} failed: {type(e).__name__}: {e}" return {"type": "function_call_output", "call_id": call_id, "output": output} @@ -417,7 +422,6 @@ def get_agent_response( "store": True, "temperature": temperature, "tools": tools, - "tool_choice": "auto", "parallel_tool_calls": True, # batched tool calls share one turn, so the history is re-billed fewer times "prompt_cache_key": f"agent-run:{uuid4().hex}", "context_management": [{"type": "compaction", "compact_threshold": 200_000}], @@ -433,12 +437,17 @@ def get_agent_response( total_usage = None previous_response_id = None next_input = conversation - turn = -1 - for turn in range(max_turns): - data = {**base_data, "input": next_input} + tool_choice = "auto" if max_turns > 0 else "none" + repaired = False + # Up to two tool-free turns follow the tool turns: a forced synthesis once max_turns is spent, then one repair + # of a malformed structured reply. Every path returns or raises before the range is exhausted. + for turn in range(max_turns + 2): + data = {**base_data, "input": next_input, "tool_choice": tool_choice} if previous_response_id: data["previous_response_id"] = previous_response_id - response_json, elapsed = _post_openai_response(data, headers, retries, request_timeout) + response_json, elapsed = _post_openai_response( + data, headers, retries if tool_choice == "auto" else max(retries, 2), request_timeout + ) total_elapsed += elapsed total_usage = _add_openai_usage(total_usage, response_json) previous_response_id = response_json.get("id") @@ -454,55 +463,64 @@ def get_agent_response( response_json, model, elapsed, - f"turn {turn + 1}/{max_turns}, {_format_tool_calls(turn_calls)}", + f"turn {turn + 1 if tool_choice == 'auto' else 'final'}/{max_turns}, {_format_tool_calls(turn_calls)}", turn_cost, ) function_calls = [item for item in output_items if item.get("type") == "function_call"] - if not function_calls: - _print_openai_usage( - {"usage": total_usage}, - model, - total_elapsed, - f"agent total, {turn + 1} turns, {_format_tool_calls(tool_calls)}", - total_cost, - ) - return _finalize_response_content(response_json, text_format) + if function_calls: + if not previous_response_id: + raise RuntimeError("OpenAI response did not include an id for server-managed continuation") + if max_cost and total_cost >= max_cost: + raise RuntimeError(f"Agent cost budget ${max_cost:.2f} reached before requested tools could run") + if parallel_tools and len(function_calls) > 1: # opt-in contract: handlers must be thread-safe + with ThreadPoolExecutor(max_workers=min(8, len(function_calls))) as pool: + next_input = list(pool.map(lambda call: _handle_function_call(call, tool_handlers), function_calls)) + else: + next_input = [_handle_function_call(call, tool_handlers) for call in function_calls] + if turn + 1 >= max_turns: # tool turns are spent: the next turn must synthesize from what was gathered + next_input.append( + { + "role": "user", + "content": ( + "You have used all available tool-calling steps. Do not call tools. Synthesize the " + "gathered tool results and return the best final answer now in the required response " + "format. If the gathered context is incomplete, say so in the final answer instead of " + "dumping raw tool output." + ), + } + ) + tool_choice = "none" + continue - if not previous_response_id: - raise RuntimeError("OpenAI response did not include an id for server-managed continuation") - if max_cost and total_cost >= max_cost: - raise RuntimeError(f"Agent cost budget ${max_cost:.2f} reached before requested tools could run") - if parallel_tools and len(function_calls) > 1: # opt-in contract: handlers must be thread-safe - with ThreadPoolExecutor(max_workers=min(8, len(function_calls))) as pool: - next_input = list(pool.map(lambda call: _handle_function_call(call, tool_handlers), function_calls)) - else: - next_input = [_handle_function_call(call, tool_handlers) for call in function_calls] - - final_instruction = { - "role": "user", - "content": ( - "You have used all available tool-calling steps. Do not call tools. Synthesize the gathered tool results " - "and return the best final answer now in the required response format. If the gathered context is " - "incomplete, say so in the final answer instead of dumping raw tool output." - ), - } - data = {**base_data, "input": [*next_input, final_instruction], "tool_choice": "none"} - if previous_response_id: - data["previous_response_id"] = previous_response_id - response_json, elapsed = _post_openai_response(data, headers, max(retries, 2), request_timeout) - total_elapsed += elapsed - total_usage = _add_openai_usage(total_usage, response_json) - total_cost += _openai_usage_cost(response_json.get("usage") or {}, model) - _print_openai_usage(response_json, model, elapsed, f"turn final/{max_turns}, 0 tools") - _print_openai_usage( - {"usage": total_usage}, - model, - total_elapsed, - f"agent total, {turn + 2} turns, {_format_tool_calls(tool_calls)}", - total_cost, - ) - return _finalize_response_content(response_json, text_format) + try: + content = _finalize_response_content(response_json, text_format) + except json.JSONDecodeError as e: + # A completed reply can still carry malformed or cut-off JSON; ask once for the full object before failing + text = _openai_response_text(response_json) + print(f"Malformed structured output ({e}) in {len(text)} chars ending {text[-200:]!r}") + if repaired or not previous_response_id: + raise + repaired = True + next_input = [ + { + "role": "user", + "content": ( + f"Your previous reply was not valid JSON for the required response format ({e}). " + "Return the complete JSON object now, and nothing else." + ), + } + ] + tool_choice = "none" + continue + _print_openai_usage( + {"usage": total_usage}, + model, + total_elapsed, + f"agent total, {turn + 1} turns, {_format_tool_calls(tool_calls)}", + total_cost, + ) + return content def get_response( diff --git a/tests/test_first_interaction.py b/tests/test_first_interaction.py index 176da63..5f4f568 100644 --- a/tests/test_first_interaction.py +++ b/tests/test_first_interaction.py @@ -352,6 +352,7 @@ def test_clear_previous_review_captures_history_and_deletes_inline_comments(): ) event.delete.assert_called_once_with( "https://api.github.com/repos/org/repo/pulls/comments/4", + expected_status=[200, 204, 404], hard=True, ) diff --git a/tests/test_openai_utils.py b/tests/test_openai_utils.py index 46dd15b..166a4ee 100644 --- a/tests/test_openai_utils.py +++ b/tests/test_openai_utils.py @@ -1,5 +1,6 @@ # Ultralytics 🚀 AGPL-3.0 License - https://ultralytics.com/license +import json from unittest.mock import MagicMock, patch import pytest @@ -398,6 +399,51 @@ def test_get_agent_response_summarizes_after_max_turns(mock_post): assert "Synthesize the gathered tool results" in final_payload["input"][-1]["content"] +@patch("requests.post") +def test_get_agent_response_repairs_malformed_json(mock_post): + """Test a completed reply with cut-off structured JSON gets one tool-free repair turn before failing.""" + truncated = MagicMock(status_code=200) + truncated.elapsed.total_seconds.return_value = 1.0 + truncated.json.return_value = { + "id": "resp_truncated", + "status": "completed", + "output": [ + {"type": "message", "content": [{"type": "output_text", "text": '{"comments": [], "summary": "cut'}]} + ], + "usage": {"input_tokens": 10, "output_tokens": 5}, + } + repaired = MagicMock(status_code=200) + repaired.elapsed.total_seconds.return_value = 1.0 + repaired.json.return_value = { + "id": "resp_repaired", + "status": "completed", + "output": [ + {"type": "message", "content": [{"type": "output_text", "text": '{"comments": [], "summary": "ok"}'}]} + ], + "usage": {"input_tokens": 10, "output_tokens": 5}, + } + mock_post.side_effect = [truncated, repaired] + text_format = {"format": {"type": "json_schema", "name": "review", "strict": True, "schema": {"type": "object"}}} + + with patch("actions.utils.openai_utils.OPENAI_API_KEY", "test-key"): + result = get_agent_response( + [{"role": "user", "content": "review"}], tools=[], tool_handlers={}, text_format=text_format, retries=0 + ) + + assert result == {"comments": [], "summary": "ok"} + assert mock_post.call_count == 2 + repair_payload = mock_post.call_args_list[1].kwargs["json"] + assert repair_payload["tool_choice"] == "none" + assert repair_payload["previous_response_id"] == "resp_truncated" + assert "not valid JSON" in repair_payload["input"][0]["content"] + + mock_post.side_effect = [truncated, truncated] + with patch("actions.utils.openai_utils.OPENAI_API_KEY", "test-key"), pytest.raises(json.JSONDecodeError): + get_agent_response( + [{"role": "user", "content": "review"}], tools=[], tool_handlers={}, text_format=text_format, retries=0 + ) + + @patch("requests.post") def test_get_response_anthropic(mock_post): """Test Anthropic Messages API completion function with mocked response.""" @@ -523,27 +569,39 @@ def test_get_response_rejects_incomplete_response(mock_post): @patch("requests.post") -def test_get_agent_response_rejects_failed_tool(mock_post): - """Test failed evidence tools abort the agent run.""" - response = MagicMock(status_code=200) - response.elapsed.total_seconds.return_value = 1.0 - response.json.return_value = { +def test_get_agent_response_feeds_back_failed_tool(mock_post): + """Test a failed tool call is returned to the model as tool output instead of aborting the agent run.""" + tool_response = MagicMock(status_code=200) + tool_response.elapsed.total_seconds.return_value = 1.0 + tool_response.json.return_value = { "id": "resp_tool", "status": "completed", "output": [{"type": "function_call", "call_id": "call_123", "name": "read_file", "arguments": "{}"}], "usage": {"input_tokens": 10, "output_tokens": 5}, } - mock_post.return_value = response + final_response = MagicMock(status_code=200) + final_response.elapsed.total_seconds.return_value = 1.0 + final_response.json.return_value = { + "id": "resp_final", + "status": "completed", + "output": [{"type": "message", "content": [{"type": "output_text", "text": "could not read the file"}]}], + "usage": {"input_tokens": 10, "output_tokens": 5}, + } + mock_post.side_effect = [tool_response, final_response] def failed_read(): raise RuntimeError("read failed") - with patch("actions.utils.openai_utils.OPENAI_API_KEY", "test-key"), pytest.raises( - RuntimeError, match="read failed" - ): - get_agent_response( - [{"role": "user", "content": "review"}], - tools=[], - tool_handlers={"read_file": failed_read}, - retries=0, + with patch("actions.utils.openai_utils.OPENAI_API_KEY", "test-key"): + result = get_agent_response( + [{"role": "user", "content": "review"}], tools=[], tool_handlers={"read_file": failed_read}, retries=0 ) + + assert result == "could not read the file" + assert mock_post.call_args_list[1].kwargs["json"]["input"] == [ + { + "type": "function_call_output", + "call_id": "call_123", + "output": "Tool read_file failed: RuntimeError: read failed", + } + ]