fix(utils): preserve outer boundaries when extracting JSON

Author: ulleo

backend/common/utils/utils.py

@@ -53,25 +53,49 @@ def deepcopy_ignore_extra(src, dest):
     return dest
 
 
-def extract_nested_json(text: str) -> Optional[str]:
+def extract_nested_json(text: str) -> str | None:
+    """Return the first valid outer JSON object/array, never a failed one's child."""
     if not text or not isinstance(text, str):
         return None
 
-    decoder = json.JSONDecoder(strict=False)
+    depth = 0
+    start = 0
+    in_string = False
+    escaped = False
     for i, char in enumerate(text):
-        if char in '{[':
-            try:
-                _, end = decoder.raw_decode(text, i)
-                candidate = text[i:end]
+        if depth == 0:
+            if char in "{[":
+                start = i
+                depth = 1
+            continue
+
+        if in_string:
+            if escaped:
+                escaped = False
+            elif char == "\\":
+                escaped = True
+            elif char == '"':
+                in_string = False
+            continue
+
+        if char == '"':
+            in_string = True
+        elif char in "{[":
+            depth += 1
+        elif char in "}]":
+            depth -= 1
+            if depth == 0:
+                # Count boundaries only; orjson checks bracket types and syntax.
+                # Skip an invalid candidate as a whole instead of retrying inside it.
+                candidate = text[start : i + 1]
                 try:
                     orjson.loads(candidate)
-                    return candidate
-                except Exception:
+                except orjson.JSONDecodeError:
                     continue
-            except Exception:
-                continue
+                return candidate
     return None
 
+
 def string_to_numeric_hash(text: str, bits: Optional[int] = 64) -> int:
     hash_bytes = hashlib.sha256(text.encode()).digest()
     hash_num = int.from_bytes(hash_bytes, byteorder='big')
@@ -287,4 +311,4 @@ def equals_ignore_case(str1: str, *args: str) -> bool:
             continue
         if str1.casefold() == arg.casefold():
             return True
-    return False
+    return False

backend/tests/test_extract_nested_json.py

@@ -1,14 +1,15 @@
 import os
 import sys
-import pytest
+
 import orjson
+import pytest
 
 # Ensure backend root is in sys.path
 backend_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 if backend_dir not in sys.path:
     sys.path.insert(0, backend_dir)
 
-from common.utils.utils import extract_nested_json
+from common.utils.utils import extract_nested_json  # noqa: E402
 
 
 def test_standard_json_extraction():
@@ -116,6 +117,58 @@ def test_invalid_candidate_before_valid_json():
     assert data["sql"] == "SELECT 1"
 
 
-@pytest.mark.parametrize("invalid_input", [None, "", "   ", 123, [], {}, "No json at all here"])
+@pytest.mark.parametrize(
+    "invalid_input", [None, "", "   ", 123, [], {}, "No json at all here"]
+)
 def test_invalid_or_empty_inputs(invalid_input):
     assert extract_nested_json(invalid_input) is None
+
+
+@pytest.mark.parametrize(
+    "invalid_candidate",
+    [
+        '{"tables": [],}',
+        '{"outer": {"success": true, "sql": "SELECT 0"}, bad}',
+        '[{"year": 2026, "value": 10},]',
+        '{"sql": "SELECT \'{}\'", bad}',
+        '{"tables": [}, "other": {"success": true}}',
+        '{"value": NaN, "tables": []}',
+        '{"sql": "SELECT\n1", "tables": []}',
+    ],
+)
+def test_invalid_outer_candidate_does_not_expose_nested_values(invalid_candidate):
+    assert extract_nested_json(invalid_candidate) is None
+    answer = '{"success": true, "sql": "SELECT 1", "tables": ["t"]}'
+    assert extract_nested_json(invalid_candidate + " then " + answer) == answer
+
+
+@pytest.mark.parametrize(
+    "truncated",
+    [
+        '{"sql": "SELECT 1", "tables": ["t"]',
+        '[{"year": 2026, "value": 10},',
+        '[{"x": 1}',
+        '{"outer": {"success": true, "sql": "SELECT 1"}',
+        '{"text": "unterminated [] {}',
+    ],
+)
+def test_truncated_outer_candidate_is_rejected(truncated):
+    assert extract_nested_json(truncated) is None
+
+
+@pytest.mark.parametrize(
+    "answer",
+    [
+        '{"outer": [{"inner": [1, 2]}, []], "empty": {}}',
+        "{\"sql\": \"SELECT '[', ']', '{', '}'\", \"tables\": []}",
+        r'{"text": "escaped quote: \" { [", "tail": "ends in slash \\"}',
+        "{}",
+        "[]",
+    ],
+)
+def test_complete_outer_candidate_is_preserved(answer):
+    assert (
+        extract_nested_json('Explanation: "answer" ```json\n' + answer + "\n```")
+        == answer
+    )
+    assert extract_nested_json(answer + ' {"next": true}') == answer