diff --git a/python/sglang/srt/parser/chat_parsing/content_parsers.py b/python/sglang/srt/parser/chat_parsing/content_parsers.py
index d857b61d228b..f00a87b6e2c1 100644
--- a/python/sglang/srt/parser/chat_parsing/content_parsers.py
+++ b/python/sglang/srt/parser/chat_parsing/content_parsers.py
@@ -79,7 +79,7 @@ def _capture(m: Any) -> str:
 
     try:
         return json.loads(working)
-    except json.JSONDecodeError as e:
+    except (json.JSONDecodeError, RecursionError) as e:
         if args.get("allow_non_json"):
             return _text(text, args)
         if working == text:
@@ -149,12 +149,12 @@ def _kv_lines(text: str, args: dict) -> dict:
     "kv-lines": _kv_lines,
 }
 
-# Parsers whose output is the verbatim body text (modulo whitespace) — chunks
-# from these fields stream with `dirty=False` because each chunk is part of
-# the final value. Structured parsers (`json`, `xml-inline`, `kv-lines`) only
-# produce a meaningful value on close, so their chunks stream raw bytes
-# flagged `dirty=True` while the parsed value is delivered in `region_close`.
-STREAMABLE_PARSERS = frozenset({"text", "int", "float", "bool"})
+# Parsers whose output is the verbatim body text, apart from outer whitespace that
+# `strip` trims at close. Their chunks stream with `dirty=False` unless a `transform`
+# reshapes the value. Every other parser only produces its value on close, so its
+# chunks stream raw bytes flagged `dirty=True` while the parsed value is delivered
+# in `region_close`.
+STREAMABLE_PARSERS = frozenset({"text"})
 
 
 def parse_content(text: str, name: str, args: dict) -> Any:
diff --git a/python/sglang/srt/parser/chat_parsing/response_parser.py b/python/sglang/srt/parser/chat_parsing/response_parser.py
index 24fdc19f7e19..721ab934903b 100644
--- a/python/sglang/srt/parser/chat_parsing/response_parser.py
+++ b/python/sglang/srt/parser/chat_parsing/response_parser.py
@@ -62,7 +62,7 @@ def _coerce(raw: str, types: tuple[str, ...]) -> Any:
                 decoded = json.loads(raw)
                 if isinstance(decoded, dict if type_name == "object" else list):
                     return decoded
-        except ValueError:
+        except (ValueError, RecursionError):
             continue
     return raw  # `string` params, unknown types and failed casts all keep the original text
 
@@ -83,7 +83,13 @@ def parse_response(
     response_template = load_response_template(response_template)
     stream = ResponseParser(response_template, prefix=prefix, tools=tools)
     stream.feed(text)
-    message, _ = stream.finalize()
+    try:
+        message, _ = stream.finalize()
+    except ValueError:
+        if not stream._malformed:
+            raise
+    if stream._malformed:
+        raise next(iter(stream._malformed.values()))
     return message
 
 
@@ -107,7 +113,14 @@ class ResponseParser:
     Pass OpenAI-style `tools=` dictionaries to cast tool-call arguments
     using the calling tool's JSON schema as each region closes.
 
-    Events can be either "region_open", "region_chunk", or "region_close".
+    Events can be "region_open", "region_chunk", "region_close", or "region_malformed".
+    Open, close and malformed events carry `start` and `end`, the span of `input_text` the
+    event consumed: its opening or closing delimiter, empty for implicit boundaries.
+    Explicit opens also carry the opener's named `captures`. A region whose value fails to
+    parse ends with "region_malformed" instead of "region_close", carrying the `error`, and
+    parsing continues after it. Only a plain "text" field without a `transform` streams
+    `dirty=False` chunks: verbatim pieces of the region's value, whose outer whitespace
+    `strip` may still trim at close. Other chunks are raw input parsed on close.
 
     ResponseParser requires the chat `prefix` (i.e. the chat history, the prefill before the current generation).
     This is because chat templates or assistant prefills can sometimes write part of the message, and if we
@@ -152,10 +165,18 @@ def __init__(
         self._body: str = ""
         self._opened: bool = False
         self._finalized: bool = False
+        self._prefix_len: int = 0
+        self._malformed: dict[str, Exception] = {}
         self.initial_events: list[dict] = []
         if prefix:
             self._consume_prefix(prefix)
 
+    @property
+    def input_text(self) -> str:
+        """Raw parser input after start-anchor truncation. Before the first `feed()`, this is
+        exactly the truncated prefix, so its length marks where generated text begins."""
+        return self._buffer
+
     def _consume_prefix(self, prefix: str) -> None:
         """Loads the prefix (the chat prefill sent to the model), right-truncates it to the start of the
         assistant message (as determined by start_anchor) and then runs the remainder through the parser.
@@ -169,6 +190,7 @@ def _consume_prefix(self, prefix: str) -> None:
             return
         self._buffer = truncated
         self._process(self.initial_events, eos=False)
+        self._prefix_len = len(truncated)
 
     def feed(self, text: str) -> list[dict]:
         """Feeds more text/tokens from the model output into the tokenizer, and returns any events that result
@@ -194,7 +216,11 @@ def _is_empty(v: Any) -> bool:
             raise RuntimeError("ResponseParser already finalized")
         events: list[dict] = []
         self._process(events, eos=True)
-        missing = [n for n, f in self._spec.fields.items() if not f.optional and n not in self._output]
+        missing = [
+            n
+            for n, f in self._spec.fields.items()
+            if not f.optional and n not in self._output and n not in self._malformed
+        ]
         if missing:
             raise ValueError(f"Required response_template fields missing from parsed output: {missing}")
         defaults = self._spec.defaults
@@ -213,12 +239,12 @@ def _process(self, events: list[dict], eos: bool) -> None:
                     self._accumulate(events, self._buffer[self._pos : m.start()])
                 self._pos = m.end()
                 if kind == "open":
-                    self._close_current(events)
+                    self._close_current(events, m.start(), m.start())
                     self._open_explicit(events, field, m)
                 else:  # "close" (always the implicit region's close here,
                     #   since explicit regions only expose their own close)
                     had_content = self._opened
-                    self._close_current(events)
+                    self._close_current(events, m.start(), m.end())
                     # Zero-width close on an already-empty region would just
                     # re-fire next iteration -- bail out to make progress.
                     if not had_content and m.start() == m.end():
@@ -230,7 +256,7 @@ def _process(self, events: list[dict], eos: bool) -> None:
                 if self._pos < len(self._buffer):
                     self._accumulate(events, self._buffer[self._pos :])
                     self._pos = len(self._buffer)
-                self._close_current(events)
+                self._close_current(events, self._pos, self._pos)
                 break
             # Stream everything up to the earliest still-pending delimiter. When
             # nothing is pending `hold_start == len(self._buffer)`, so this flushes
@@ -331,27 +357,31 @@ def _accumulate(self, events: list[dict], text: str) -> None:
         """Route `text` into the currently active region. When the current
         region is the null sink (no implicit declared, no explicit open), we
         silently discard. Every routed chunk emits a `region_chunk` event so
-        consumers can render live; `dirty=True` flags chunks from structured
-        parsers (json, xml-inline, kv-lines) whose raw bytes will only be
-        parsed into the final value on close."""
+        consumers can render live; `dirty=True` flags chunks whose raw bytes
+        only become the final value once parsed or transformed on close."""
         if not text or self._current is None:
             return
         field = self._spec.fields[self._current]
         if not self._opened:
-            events.append({"type": "region_open", "field": self._current})
+            events.append({"type": "region_open", "field": self._current, "start": self._pos, "end": self._pos})
             self._opened = True
         self._body += text
-        dirty = field.content not in STREAMABLE_PARSERS
-        events.append({"type": "region_chunk", "field": self._current, "text": text, "dirty": dirty})
+        # Prefix bytes held back as a possible delimiter belong to the prompt, not the chunk.
+        text = text[max(0, self._prefix_len - self._pos) :]
+        if text:
+            dirty = field.content not in STREAMABLE_PARSERS or field.transform is not None
+            events.append({"type": "region_chunk", "field": self._current, "text": text, "dirty": dirty})
 
     def _open_explicit(self, events: list[dict], field: ResponseTemplateField, m: Any) -> None:
         self._current = field.name
         self._captures = {k: v for k, v in m.groupdict().items() if v is not None}
         self._body = ""
         self._opened = True
-        events.append({"type": "region_open", "field": field.name})
+        events.append(
+            {"type": "region_open", "field": field.name, "start": m.start(), "end": m.end(), "captures": dict(self._captures)}
+        )
 
-    def _close_current(self, events: list[dict]) -> None:
+    def _close_current(self, events: list[dict], close_start: int, close_end: int) -> None:
         """Close the current region and reset to the implicit/null region.
         Skipped (aside from the reset) when the current region never opened --
         avoids vacuous open/close pairs at every explicit boundary."""
@@ -359,22 +389,30 @@ def _close_current(self, events: list[dict]) -> None:
             self._reset_to_implicit()
             return
         field = self._spec.fields[self._current]
-        value = process_field(self._body, field, self._captures)
-        if self._tool_params:
-            value = self._coerce_tool_calls(value)
-        if field.join is not None:
-            if not isinstance(value, str):
+        try:
+            value = process_field(self._body, field, self._captures)
+            if self._tool_params:
+                value = self._coerce_tool_calls(value)
+            if field.join is not None and not isinstance(value, str):
                 raise ValueError(
                     f"Field '{field.name}': 'join' requires each match to parse to a string, "
                     f"got {type(value).__name__}."
                 )
+        except (KeyError, TypeError, ValueError) as error:
+            self._malformed.setdefault(self._current, error)
+            events.append(
+                {"type": "region_malformed", "field": self._current, "start": close_start, "end": close_end, "error": error}
+            )
+            self._reset_to_implicit()
+            return
+        if field.join is not None:
             previous = self._output.get(self._current)
             self._output[self._current] = value if previous is None else previous + field.join + value
         elif field.repeats:
             self._output.setdefault(self._current, []).append(value)
         else:
             self._output[self._current] = value
-        events.append({"type": "region_close", "field": self._current, "value": value})
+        events.append({"type": "region_close", "field": self._current, "start": close_start, "end": close_end, "value": value})
         self._reset_to_implicit()
 
     def _coerce_tool_calls(self, value: Any) -> Any:
diff --git a/test/registered/unit/parser/test_chat_parsing.py b/test/registered/unit/parser/test_chat_parsing.py
index d3d6e6d505e6..bc7f2cd4d531 100644
--- a/test/registered/unit/parser/test_chat_parsing.py
+++ b/test/registered/unit/parser/test_chat_parsing.py
@@ -27,6 +27,7 @@
 from sglang.srt.parser.chat_parsing.response_parser import _coerce, _schema_types
 from sglang.srt.parser.chat_parsing.response_templates import load_response_template
 from sglang.test.ci.ci_register import register_cpu_ci
+from sglang.test.test_utils import CustomTestCase
 
 register_cpu_ci(est_time=8, suite="base-a-test-cpu")
 
@@ -953,7 +954,7 @@ def _chunk_random(text: str, rng: random.Random):
     yield text[prev:]
 
 
-class ResponseEventStreamTest(unittest.TestCase):
+class ResponseEventStreamTest(CustomTestCase):
     def test_stream_matches_whole_string_all_templates_fixed_chunking(self):
         """For every fixed chunking step we try, the streamed finalize()
         output must equal the whole-string parse. Regression coverage for
@@ -1066,16 +1067,17 @@ def test_region_chunks_reconstruct_text_regions(self):
         self.assertIn("tool_calls", per_field_close_value)
 
     def test_dirty_flag_marks_structured_regions(self):
-        """A template with one text field and one structured field per parser
-        family: text/int/float/bool stream chunks with `dirty=False`, while
-        json/xml-inline/kv-lines stream chunks with `dirty=True`, and those
-        dirty chunks concatenate to the raw region body before parsing."""
+        """Only plain text without a `transform` streams `dirty=False` chunks; int/float/bool,
+        structured parsers and transforms stream dirty chunks of the raw region body."""
         spec = {
             "defaults": {"role": "assistant"},
             "start_anchor": "<|assistant|>",
             "fields": {
                 "thinking": {"open": "<t>", "close": "</t>", "content": "text"},
                 "score": {"open": "<n>", "close": "</n>", "content": "int"},
+                "ratio": {"open": "<f>", "close": "</f>", "content": "float"},
+                "flag": {"open": "<b>", "close": "</b>", "content": "bool"},
+                "labeled": {"open": "<l>", "close": "</l>", "transform": {"label": "{content}"}},
                 "json_call": {"open": "<j>", "close": "</j>", "content": "json"},
                 "xml_call": {
                     "open": "<x>",
@@ -1090,7 +1092,10 @@ def test_dirty_flag_marks_structured_regions(self):
                 },
             },
         }
-        text = '<t>hello world</t><n>42</n><j>{"a": 1, "b": 2}</j><x><name=foo><age=10></x><kv>k1: v1\nk2: v2</kv>'
+        text = (
+            "<t> hello world </t><n>42</n><f>0.5</f><b>true</b><l>tag</l>"
+            '<j>{"a": 1, "b": 2}</j><x><name=foo><age=10></x><kv>k1: v1\nk2: v2</kv>'
+        )
         # Drive byte-by-byte to maximise chunk count.
         streamer = ResponseParser(spec, prefix="")
         events: list[dict] = []
@@ -1101,22 +1106,24 @@ def test_dirty_flag_marks_structured_regions(self):
 
         per_field_chunks: dict[str, list[str]] = {}
         per_field_dirty: dict[str, set[bool]] = {}
+        per_field_close_value: dict[str, object] = {}
         for ev in events:
             if ev["type"] == "region_chunk":
                 per_field_chunks.setdefault(ev["field"], []).append(ev["text"])
                 per_field_dirty.setdefault(ev["field"], set()).add(ev["dirty"])
+            elif ev["type"] == "region_close":
+                per_field_close_value[ev["field"]] = ev["value"]
 
-        # Clean (streamable) regions.
-        for field in ("thinking", "score"):
-            self.assertEqual(per_field_dirty[field], {False}, f"{field} should be clean")
-        # Dirty (structured) regions.
-        for field in ("json_call", "xml_call", "kv_call"):
+        self.assertEqual(per_field_dirty["thinking"], {False}, "thinking should be clean")
+        for field in ("score", "ratio", "flag", "labeled", "json_call", "xml_call", "kv_call"):
             self.assertEqual(per_field_dirty[field], {True}, f"{field} should be dirty")
 
         # Dirty chunks reconstruct the raw region body (un-parsed). Clean
         # chunks reconstruct the verbatim body too: stripping happens at close.
-        self.assertEqual("".join(per_field_chunks["thinking"]), "hello world")
+        self.assertEqual("".join(per_field_chunks["thinking"]), " hello world ")
+        self.assertEqual(per_field_close_value["thinking"], "hello world")
         self.assertEqual("".join(per_field_chunks["score"]), "42")
+        self.assertEqual(per_field_close_value["labeled"], {"label": "tag"})
         self.assertEqual("".join(per_field_chunks["json_call"]), '{"a": 1, "b": 2}')
         self.assertEqual("".join(per_field_chunks["xml_call"]), "<name=foo><age=10>")
         self.assertEqual("".join(per_field_chunks["kv_call"]), "k1: v1\nk2: v2")
@@ -1159,6 +1166,137 @@ def test_empty_input_streams_cleanly(self):
         self.assertEqual(result, {"role": "assistant"})
         self.assertEqual(final_events, [])
 
+    def test_region_events_expose_delimiter_offsets(self):
+        spec = {
+            "start_anchor": "<assistant>",
+            "fields": {
+                "content": {"content_args": {"strip": False}},
+                "tag": {"open_pattern": r"<tag id=\d+>", "close": "</tag>"},
+            },
+        }
+        text = "plain<tag id=7>body</tag>tail"
+        expected = [
+            {"type": "region_open", "field": "content", "start": 0, "end": 0},
+            {"type": "region_close", "field": "content", "start": 5, "end": 5, "value": "plain"},
+            {"type": "region_open", "field": "tag", "start": 5, "end": 15, "captures": {}},
+            {"type": "region_close", "field": "tag", "start": 19, "end": 25, "value": "body"},
+            {"type": "region_open", "field": "content", "start": 25, "end": 25},
+            {"type": "region_close", "field": "content", "start": 29, "end": 29, "value": "tail"},
+        ]
+        for prefix_len in (0, len("plain<tag")):
+            generated = text[prefix_len:]
+            chunkings = [[generated], list(generated)]
+            chunkings += [[generated[:i], generated[i:]] for i in range(1, len(generated))]
+            for chunks in chunkings:
+                with self.subTest(prefix=text[:prefix_len], chunks=chunks):
+                    parser = ResponseParser(spec, prefix="<assistant>" + text[:prefix_len])
+                    events = parser.initial_events + [event for chunk in chunks for event in parser.feed(chunk)]
+                    events += parser.finalize()[1]
+                    self.assertEqual(parser.input_text, text)
+                    self.assertEqual([event for event in events if event["type"] != "region_chunk"], expected)
+
+    def test_open_captures_and_prefix_boundary(self):
+        spec = {
+            "start_anchor": "[BEGIN]",
+            "fields": {
+                "tool_calls": {
+                    "open_pattern": r"<call:(?P<name>\w+)>",
+                    "close": "</call>",
+                    "content": "json",
+                    "transform": {"name": "{name}", "arguments": "{content}"},
+                }
+            },
+        }
+        parser = ResponseParser(spec, prefix="history[BEGIN]<call:get_")
+        prefix_end = len(parser.input_text)
+        opening = parser.feed('weather>{"city":"Paris"}')[0]
+
+        self.assertEqual(prefix_end, len("<call:get_"))
+        self.assertEqual((opening["type"], opening["start"], opening["end"]), ("region_open", 0, len("<call:get_weather>")))
+        self.assertEqual(opening["captures"], {"name": "get_weather"})
+        opening["captures"]["name"] = "edited"
+        self.assertEqual(parser.feed("</call>")[-1]["value"]["name"], "get_weather")
+
+    def test_held_prefix_bytes_are_not_rechunked(self):
+        spec = {"start_anchor": "[BEGIN]", "fields": {"content": {}, "tag": {"open": "<tag>", "close": "</tag>"}}}
+        parser = ResponseParser(spec, prefix="[BEGIN]Existing <")
+        chunks = [event["text"] for event in parser.feed("b> tail") if event["type"] == "region_chunk"]
+        message, _ = parser.finalize()
+
+        self.assertEqual(chunks, ["b> tail"])
+        self.assertEqual(message["content"], "Existing <b> tail")
+
+    def test_malformed_region_events_recover_and_continue(self):
+        spec = {
+            "start_anchor": "[BEGIN]",
+            "fields": {
+                "tool_calls": {
+                    "open_pattern": r"<call:(?P<name>\w+)>",
+                    "close": "</call>",
+                    "content": "json",
+                    "repeats": True,
+                    "transform": {
+                        "type": "function",
+                        "function": {"name": "{name}", "arguments": "{content}"},
+                    },
+                }
+            },
+        }
+        parser = ResponseParser(spec, prefix="")
+        events = parser.feed('<call:bad>{"x":</call><call:good>{"x":1}</call>')
+        message, _ = parser.finalize()
+        opening, malformed = events[0], events[2]
+
+        self.assertEqual(malformed["type"], "region_malformed")
+        self.assertIsInstance(malformed["error"], ValueError)
+        self.assertEqual(parser.input_text[opening["end"] : malformed["start"]], '{"x":')
+        self.assertEqual(parser.input_text[malformed["start"] : malformed["end"]], "</call>")
+        self.assertEqual(
+            message["tool_calls"],
+            [{"type": "function", "function": {"name": "good", "arguments": {"x": 1}}}],
+        )
+
+        parser = ResponseParser(spec, prefix="")
+        parser.feed('<call:bad>{"x":')
+        _, events = parser.finalize()
+        self.assertEqual(events[-1]["type"], "region_malformed")
+        self.assertEqual(events[-1]["start"], len(parser.input_text))
+
+        deep = "[" * 20000
+        tools = [{"type": "function", "function": {"name": "good", "parameters": {"properties": {"x": {"type": "array"}}}}}]
+        parser = ResponseParser(spec, prefix="", tools=tools)
+        events = parser.feed(f'<call:bad>{deep}</call><call:good>{{"x": "{deep}"}}</call>')
+        message, _ = parser.finalize()
+        self.assertEqual(events[2]["type"], "region_malformed")
+        self.assertIsInstance(events[2]["error"], ValueError)
+        self.assertEqual(message["tool_calls"], [{"type": "function", "function": {"name": "good", "arguments": {"x": deep}}}])
+
+    def test_parse_response_raises_for_malformed_region(self):
+        spec = {
+            "start_anchor": "<|assistant|>",
+            "fields": {"x": {"open": "<x>", "close": "</x>", "content": "json"}},
+        }
+        with self.assertRaisesRegex(ValueError, "could not parse region as JSON"):
+            parse_response('<x>{"name":', spec, prefix="")
+        with self.assertRaisesRegex(ValueError, "could not parse region as JSON"):
+            parse_response("", spec, prefix='<|assistant|><x>{"name":</x>')
+
+        required = {
+            "start_anchor": "<|assistant|>",
+            "fields": {
+                "x": {"open": "<x>", "close": "</x>", "content": "json", "optional": False},
+                "y": {"open": "<y>", "close": "</y>", "optional": False},
+                "n": {"open": "<n>", "close": "</n>", "content": "int"},
+            },
+        }
+        for prefix, text in [("", "<x>{</x>"), ("<|assistant|><x>{</x>", ""), ("", "<x>{"), ("", "<x>{</x><n>a</n>")]:
+            with self.subTest(prefix=prefix, text=text), self.assertRaisesRegex(ValueError, "could not parse region as JSON"):
+                parse_response(text, required, prefix=prefix)
+        parser = ResponseParser(required, prefix="")
+        parser.feed("<x>{</x>")
+        with self.assertRaisesRegex(ValueError, r"missing from parsed output: \['y'\]"):
+            parser.finalize()
+
 
 class PrefixAndTruncationTest(unittest.TestCase):
     def test_prefix_lands_inside_explicit_region(self):
@@ -1323,7 +1461,7 @@ def test_prefix_partial_pattern_at_boundary(self):
         """Post-truncation prefix ends mid-delimiter. The first feed completes
         the match; initial_events is empty (no region opened within the
         prefix yet) and the open fires from `feed()`."""
-        prefix = "<|im_start|>assistant\n<thi"  # incomplete `<think>`
+        prefix = "<|im_start|>assistant\n<thi"  # incomplete `<think>`  # codespell:ignore
         stream = ResponseParser(qwen3_template, prefix=prefix)
         self.assertEqual(stream.initial_events, [])
         events = stream.feed("nk>real body</think>")