unsloth/studio/backend/core/tool_healing.py

# SPDX-License-Identifier: AGPL-3.0-only
# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0

"""Lightweight tool-call XML parsing and stripping helpers.

External inference servers import this module without pulling in the inference
orchestrator, structlog, httpx, or the rest of the studio backend.
"""

import json
import re

# Strip patterns. The name-class hyphen matches dashed MCP names. Closed pairs
# strip first so a closed call goes as a unit before any to-EOF sweep reaches
# nested markup; only the final list adds the .*$ EOF sweeps.
_TC_JSON_CLOSED_PAT = re.compile(r"<tool_call>.*?</tool_call>", re.DOTALL)
_TC_GEMMA_CLOSED_PAT = re.compile(r"<\|tool_call>.*?<tool_call\|>", re.DOTALL)
_TC_FUNC_CLOSED_PAT = re.compile(r"<function=[\w-]+>.*?</function>", re.DOTALL)
_TC_GEMMA_END_PAT = re.compile(r"<tool_call\|>")
_TOOL_CLOSED_PATS = [
    _TC_JSON_CLOSED_PAT,
    _TC_GEMMA_CLOSED_PAT,
    _TC_FUNC_CLOSED_PAT,
    _TC_GEMMA_END_PAT,
]
_TOOL_ALL_PATS = _TOOL_CLOSED_PATS + [
    re.compile(r"<\|tool_call>.*$", re.DOTALL),
    re.compile(r"<tool_call>.*$", re.DOTALL),
    re.compile(r"<function=[\w-]+>.*$", re.DOTALL),
]
# Stripped before the quote-aware Gemma helper so a Gemma opener quoted in
# their argument data cannot make the helper truncate the block and its tail.
_TOOL_CLOSED_BLOCK_PATS = [_TC_JSON_CLOSED_PAT, _TC_FUNC_CLOSED_PAT]
# A lazy closed-pair pattern whose close token is absent rescans to EOF from
# every opener (quadratic, re-run per streamed token); skip that doomed pass.
_PAT_REQUIRED_TOKEN = {
    _TC_JSON_CLOSED_PAT: "</tool_call>",
    _TC_GEMMA_CLOSED_PAT: "<tool_call|>",
    _TC_FUNC_CLOSED_PAT: "</function>",
}


def strip_tool_patterns(text: str, patterns) -> str:
    """Apply ``patterns`` in order, skipping closed-pair passes with no close token."""
    for pat in patterns:
        token = _PAT_REQUIRED_TOKEN.get(pat)
        if token is not None and token not in text:
            continue
        text = pat.sub("", text)
    return text


# Pre-compiled patterns for tool-call XML parsing.
_TC_JSON_START_RE = re.compile(r"<tool_call>\s*\{")
# Name class allows dots/hyphens for dotted Gemma names; whitespace-tolerant around
# ``call`` / ``:`` since drift emits ``call: name{`` and ``call : name{``.
_TC_GEMMA_START_RE = re.compile(r"<\|tool_call>\s*call\s*:\s*([\w.\-]+)\s*\{")
_TC_FUNC_START_RE = re.compile(r"<function=([\w-]+)>\s*")
_TC_END_TAG_RE = re.compile(r"</tool_call>")
_TC_GEMMA_END_TAG_RE = re.compile(r"<tool_call\|>")
_TC_FUNC_CLOSE_RE = re.compile(r"\s*</function>\s*$")
# Horizontal whitespace only so the newline + value indentation survive (_trim_param_value trims one newline).
_TC_PARAM_START_RE = re.compile(r"<parameter=([\w-]+)>[^\S\n]*")
_TC_PARAM_CLOSE_RE = re.compile(r"\s*</parameter>\s*$")
_GEMMA_QUOTE = '<|"|>'
_PARAM_CLOSE_TAG = "</parameter>"
_FUNC_CLOSE_TAG = "</function>"
# A bare (unquoted) Gemma value ends at `}` or at a comma beginning the next
# identifier-shaped `key:` pair; a comma before a non-key (`New York, NY`,
# `10:00, 11:00`) stays in the value. Dots let a dotted key end the value.
_GEMMA_NEXT_KEY_RE = re.compile(r"\s*[A-Za-z_][\w.\-]*\s*:")


def _balanced_brace_end(
    content: str,
    brace_start: int,
    *,
    gemma_quotes: bool = False,
) -> int:
    depth = 0
    i = brace_start
    in_string = False
    in_gemma_string = False
    while i < len(content):
        if gemma_quotes and not in_string and content.startswith(_GEMMA_QUOTE, i):
            in_gemma_string = not in_gemma_string
            i += len(_GEMMA_QUOTE)
            continue
        ch = content[i]
        if in_gemma_string:
            i += 1
            continue
        if in_string:
            if ch == "\\" and i + 1 < len(content):
                i += 2
                continue
            if ch == '"':
                in_string = False
        elif ch == '"':
            in_string = True
        elif ch == "{":
            depth += 1
        elif ch == "}":
            depth -= 1
            if depth == 0:
                return i
        i += 1
    return -1


def _balanced_bracket_end(src: str, start: int) -> int:
    """Index of the ``]`` matching the ``[`` at ``start``, or -1. Tracks nested
    ``[]``/``{}`` and double-quoted strings."""
    depth = 0
    i = start
    in_string = False
    while i < len(src):
        ch = src[i]
        if in_string:
            if ch == "\\" and i + 1 < len(src):
                i += 2
                continue
            if ch == '"':
                in_string = False
        elif ch == '"':
            in_string = True
        elif ch in "[{":
            depth += 1
        elif ch in "]}":
            depth -= 1
            if depth == 0:
                return i
        i += 1
    return -1


def _split_top_level_commas(src: str) -> list:
    """Split on commas that are not inside a nested ``[]``/``{}`` or a string."""
    parts: list[str] = []
    depth = 0
    in_string = False
    start = 0
    i = 0
    while i < len(src):
        ch = src[i]
        if in_string:
            if ch == "\\" and i + 1 < len(src):
                i += 2
                continue
            if ch == '"':
                in_string = False
        elif ch == '"':
            in_string = True
        elif ch in "[{":
            depth += 1
        elif ch in "]}":
            depth -= 1
        elif ch == "," and depth == 0:
            parts.append(src[start:i])
            start = i + 1
        i += 1
    parts.append(src[start:])
    return parts


def _quote_gemma_array_elements(body: str) -> str:
    """Normalise a Gemma array value (``labels:[bug,ui]``) so json.loads succeeds:
    quote bare strings, recurse into objects/arrays, keep quoted/JSON literals."""
    out: list[str] = []
    for element in _split_top_level_commas(body):
        stripped = element.strip()
        if not stripped or stripped[0] == '"':
            out.append(element)
            continue
        if stripped[0] == "{":
            out.append(_quote_gemma_object_keys(stripped))
            continue
        if stripped[0] == "[":
            inner_end = _balanced_bracket_end(stripped, 0)
            if inner_end == len(stripped) - 1:
                out.append("[" + _quote_gemma_array_elements(stripped[1:inner_end]) + "]")
            else:
                out.append(element)
            continue
        try:
            json.loads(stripped)
            out.append(element)
        except (json.JSONDecodeError, ValueError):
            out.append(json.dumps(stripped))
    return ",".join(out)


def _normalise_gemma_quoted_strings(src: str) -> str:
    parts: list[str] = []
    i = 0
    while i < len(src):
        if not src.startswith(_GEMMA_QUOTE, i):
            parts.append(src[i])
            i += 1
            continue
        end = src.find(_GEMMA_QUOTE, i + len(_GEMMA_QUOTE))
        if end < 0:
            parts.append(src[i:])
            break
        raw_value = src[i + len(_GEMMA_QUOTE) : end]
        parts.append(json.dumps(raw_value))
        i = end + len(_GEMMA_QUOTE)
    return "".join(parts)


def _quote_gemma_object_keys(src: str) -> str:
    parts: list[str] = []
    i = 0
    in_string = False
    while i < len(src):
        ch = src[i]
        if in_string:
            parts.append(ch)
            if ch == "\\" and i + 1 < len(src):
                parts.append(src[i + 1])
                i += 2
                continue
            if ch == '"':
                in_string = False
            i += 1
            continue
        if ch == '"':
            in_string = True
            parts.append(ch)
            i += 1
            continue
        if ch not in "{,":
            parts.append(ch)
            i += 1
            continue

        parts.append(ch)
        i += 1
        key_start = i
        while i < len(src) and src[i].isspace():
            i += 1
        key_name_start = i
        # Dots match the parser's key/name charset: Gemma emits dotted argument keys
        # (user.name:...) for namespaced schemas.
        while i < len(src) and (src[i].isalnum() or src[i] in "_-."):
            i += 1
        key_name = src[key_name_start:i]
        colon_pos = i
        while colon_pos < len(src) and src[colon_pos].isspace():
            colon_pos += 1
        if key_name and colon_pos < len(src) and src[colon_pos] == ":":
            parts.append(src[key_start:key_name_start])
            parts.append(json.dumps(key_name))
            parts.append(src[i:colon_pos])
            parts.append(":")
            i = colon_pos + 1
            # Quote bare string values ({unit:celsius}); JSON stays as-is.
            ws = i
            while i < len(src) and src[i].isspace():
                i += 1
            parts.append(src[ws:i])
            if i < len(src) and src[i] == "[":
                arr_end = _balanced_bracket_end(src, i)
                if arr_end < 0:
                    parts.append(src[i:])
                    i = len(src)
                else:
                    parts.append("[" + _quote_gemma_array_elements(src[i + 1 : arr_end]) + "]")
                    i = arr_end + 1
            elif i < len(src) and src[i] not in '"{':
                v_start = i
                # Bare value: up to `}` or a comma that starts the next key:pair.
                while i < len(src):
                    if src[i] == "}":
                        break
                    if src[i] == "," and _GEMMA_NEXT_KEY_RE.match(src, i + 1):
                        break
                    i += 1
                raw = src[v_start:i]
                try:
                    json.loads(raw.strip())
                    parts.append(raw)
                except (json.JSONDecodeError, ValueError):
                    # Quote bare value; empty ({k:}) becomes "" so json.loads sees {"k":""} not invalid {"k":}.
                    parts.append(json.dumps(raw.strip()))
        else:
            parts.append(src[key_start:i])
    return "".join(parts)


def _gemma_arguments_to_json(args_src: str) -> dict:
    """Parse Gemma 4's native call:name{key:value} argument object."""
    args_src = args_src.strip()
    if not args_src:
        return {}
    src = _normalise_gemma_quoted_strings(args_src)
    src = "{" + src + "}"
    src = _quote_gemma_object_keys(src)
    return json.loads(src)


def _inside_open_parameter(content: str, pos: int) -> bool:
    """Return True when ``pos`` falls inside an unclosed parameter value."""
    last_param_start = -1
    for match in _TC_PARAM_START_RE.finditer(content, 0, pos):
        last_param_start = match.start()
    if last_param_start < 0:
        return False
    # The parameter's OWN close tag decides: if it closes after ``pos`` the position is
    # argument data (even across literal function closes); an unclosed one falls back to func close.
    own_close = content.find(_PARAM_CLOSE_TAG, last_param_start)
    if own_close >= 0:
        return own_close > pos
    func_close = content.find(_FUNC_CLOSE_TAG, last_param_start)
    return func_close < 0 or pos < func_close


def _func_close_index(content: str, body_start: int, body: str) -> int:
    """Index in ``body`` of the first ``</function>`` that is not argument
    data (not inside an open parameter value); -1 when every close is data.
    Taking the LAST close swallowed prose between the real close and a
    literal ``</function>`` mentioned later in the answer."""
    idx = body.find(_FUNC_CLOSE_TAG)
    while idx >= 0:
        if not _inside_open_parameter(content, body_start + idx):
            return idx
        idx = body.find(_FUNC_CLOSE_TAG, idx + 1)
    return -1


def _trim_param_value(val: str) -> str:
    """Trim only the wrapping newline (not str.strip) so code/diff argument indentation survives."""
    if val.startswith("\n"):
        val = val[1:]
    if val.endswith("\n"):
        val = val[:-1]
    return val


def _marker_coverage(content: str, markers) -> list[tuple[int, int]]:
    """Coverage ``[start, end]`` per marker, used to skip markers that are another
    call's data. Closes pair to markers via a per-format stack so an inner close
    is not mistaken for the outer's. Unbalanced braces cover to EOF; balanced with
    a paired close cover through it (markers before the close are data); balanced
    without one cover only the braces, so a later sibling is still recovered."""
    n = len(content)
    brace_regions = [(s, be) for (s, be, _k, _m) in markers if be >= 0]
    events = []  # (position, order) with order 0 = braces-done, 1 = close marker
    for idx, (_start, brace_end, _kind, _m) in enumerate(markers):
        if brace_end >= 0:
            events.append((brace_end, 0, _kind, idx))
    for kind, close_re in (("json", _TC_END_TAG_RE), ("gemma", _TC_GEMMA_END_TAG_RE)):
        for cm in close_re.finditer(content):
            # A close inside another call's balanced braces is quoted data; it
            # must not pop an earlier close-less marker and swallow a sibling.
            if any(s < cm.start() < be for s, be in brace_regions):
                continue
            events.append((cm.start(), 1, kind, cm.end()))
    events.sort(key = lambda e: (e[0], e[1]))
    waiting = {"json": [], "gemma": []}
    close_end_for: dict[int, int] = {}
    for _pos, order, kind, payload in events:
        if order == 0:
            waiting[kind].append(payload)  # marker index, now awaiting its close
        elif waiting[kind]:
            close_end_for[waiting[kind].pop()] = payload  # innermost open marker closes here
    coverage = []
    for idx, (start, brace_end, _kind, _m) in enumerate(markers):
        if brace_end < 0:
            coverage.append((start, n))
        elif idx in close_end_for:
            coverage.append((start, close_end_for[idx]))
        else:
            coverage.append((start, brace_end))
    return coverage


def _build_markers(content: str):
    """JSON/Gemma tool markers as ``(start, brace_end, kind, match)`` in document
    order; ``brace_end < 0`` marks an unbalanced (to-EOF) open."""
    markers = []
    for start_re, gemma, kind in (
        (_TC_JSON_START_RE, False, "json"),
        (_TC_GEMMA_START_RE, True, "gemma"),
    ):
        for m in start_re.finditer(content):
            if _inside_open_parameter(content, m.start()):
                continue
            brace_end = _balanced_brace_end(content, m.end() - 1, gemma_quotes = gemma)
            markers.append((m.start(), brace_end, kind, m))
    markers.sort(key = lambda c: c[0])
    return markers


def marker_coverage(content: str) -> list[tuple[int, int]]:
    """Coverage spans of JSON/Gemma tool markers so other parsers can treat markup
    inside a marker's coverage (even a marker that failed to parse) as that call's
    data rather than a sibling call."""
    return _marker_coverage(content, _build_markers(content))


def parse_tool_calls_from_text(
    content: str,
    *,
    id_offset: int = 0,
    allow_incomplete: bool = True,
    with_spans: bool = False,
):
    """Parse OpenAI-format tool calls from model text.

    Handles formats like:
      <tool_call>{"name":"web_search","arguments":{"query":"..."}}</tool_call>
      <|tool_call>call:web_search{query:"..."}<tool_call|>
      <tool_call><function=web_search><parameter=query>...</parameter></function></tool_call>

    With ``with_spans=True`` returns ``(tool_calls, spans)`` where ``spans[i]``
    is the half-open ``(start, end)`` byte range of ``tool_calls[i]``'s markup
    in ``content`` (including its close tag when present), so a caller can
    remove exactly the parsed markup and keep every other byte intact.
    """
    tool_calls: list[dict] = []
    call_spans: list[tuple] = []
    # Collect JSON/Gemma markers; _marker_coverage decides nesting. A marker inside
    # another call's coverage, or an open <parameter=> value, is data not executed.
    markers = _build_markers(content)
    coverage = _marker_coverage(content, markers)
    parsed_items = []  # (start, span_end, name, arguments) in document order
    for idx, (start, brace_end, kind, m) in enumerate(markers):
        # A marker starting inside another's coverage is that call's data. The
        # end is exclusive so a marker at a close's end is an adjacent sibling.
        if any(s <= start < e for j, (s, e) in enumerate(coverage) if j != idx):
            continue
        if brace_end < 0:
            continue  # unclosed: not parseable; the fallback still excludes its XML
        if not allow_incomplete:
            tail = content[brace_end + 1 :].lstrip()
            close_re = _TC_END_TAG_RE if kind == "json" else _TC_GEMMA_END_TAG_RE
            if close_re.match(tail) is None:
                continue
        try:
            if kind == "json":
                obj = json.loads(content[m.end() - 1 : brace_end + 1])
                name = obj.get("name", "")
                # Accept ``parameters`` alias for ``arguments`` (Llama-3.2 drift inside a Hermes <tool_call>).
                arguments = obj.get("arguments")
                if arguments is None:
                    arguments = obj.get("parameters", {})
                if isinstance(arguments, dict):
                    arguments = json.dumps(arguments)
            else:
                name = m.group(1)
                arguments = json.dumps(_gemma_arguments_to_json(content[m.end() : brace_end]))
        except (json.JSONDecodeError, ValueError):
            continue
        # Span reaches through the close tag when present, else just the braces.
        span_end = brace_end + 1
        close_re = _TC_END_TAG_RE if kind == "json" else _TC_GEMMA_END_TAG_RE
        ws = len(content[span_end:]) - len(content[span_end:].lstrip())
        close_m = close_re.match(content, span_end + ws)
        if close_m:
            span_end = close_m.end()
        parsed_items.append((start, span_end, name, arguments))

    # Function-XML calls promote in document order alongside marker calls (the
    # #6801 contract). A <function=> inside any marker's coverage is excluded --
    # even if that marker failed to parse -- so nested XML cannot escape; one
    # after a balanced close-less marker is a sibling, not swallowed to EOF.
    func_starts = [
        fm
        for fm in _TC_FUNC_START_RE.finditer(content)
        if not _inside_open_parameter(content, fm.start())
        and not any(s <= fm.start() < e for s, e in coverage)
    ]
    for idx, fm in enumerate(func_starts):
        func_name = fm.group(1)
        body_start = fm.end()
        next_func = func_starts[idx + 1].start() if idx + 1 < len(func_starts) else len(content)
        end_tag = _TC_END_TAG_RE.search(content[body_start:])
        if end_tag:
            body_end = body_start + end_tag.start()
        else:
            body_end = len(content)
        body_end = min(body_end, next_func)
        body = content[body_start:body_end]
        close_idx = _func_close_index(content, body_start, body)
        if close_idx >= 0:
            span_end = body_start + close_idx + len(_FUNC_CLOSE_TAG)
            body = body[:close_idx]
        elif not allow_incomplete:
            continue
        else:
            body = _TC_FUNC_CLOSE_RE.sub("", body)
            span_end = body_end

        arguments: dict = {}
        param_starts = list(_TC_PARAM_START_RE.finditer(body))
        if len(param_starts) == 1:
            pm = param_starts[0]
            val = body[pm.end() :]
            if not allow_incomplete:
                stripped_val = val.rstrip()
                if not stripped_val.endswith(_PARAM_CLOSE_TAG):
                    continue
                val = stripped_val[: -len(_PARAM_CLOSE_TAG)]
            else:
                val = _TC_PARAM_CLOSE_RE.sub("", val)
            arguments[pm.group(1)] = _trim_param_value(val)
        else:
            valid_params = True
            for pidx, pm in enumerate(param_starts):
                param_name = pm.group(1)
                val_start = pm.end()
                next_param = (
                    param_starts[pidx + 1].start() if pidx + 1 < len(param_starts) else len(body)
                )
                val = body[val_start:next_param]
                if not allow_incomplete:
                    stripped_val = val.rstrip()
                    if not stripped_val.endswith(_PARAM_CLOSE_TAG):
                        valid_params = False
                        break
                    val = stripped_val[: -len(_PARAM_CLOSE_TAG)]
                else:
                    val = _TC_PARAM_CLOSE_RE.sub("", val)
                arguments[param_name] = _trim_param_value(val)
            if not valid_params:
                continue

        span_start = fm.start()
        wrap_open = re.search(r"<tool_call>\s*$", content[:span_start])
        wrap_close = re.match(r"\s*</tool_call>", content[span_end:])
        if wrap_open and wrap_close:
            span_start = wrap_open.start()
            span_end += wrap_close.end()
        parsed_items.append((span_start, span_end, func_name, json.dumps(arguments)))

    parsed_items.sort(key = lambda item: item[0])
    for start, span_end, name, arguments in parsed_items:
        tool_calls.append(
            {
                "id": f"call_{id_offset + len(tool_calls)}",
                "type": "function",
                "function": {"name": name, "arguments": arguments},
            }
        )
        call_spans.append((start, span_end))

    if with_spans:
        return tool_calls, call_spans
    return tool_calls


def _strip_gemma_native_spans(text: str, *, final: bool) -> str:
    """Remove complete Gemma-native spans, brace/quote-balanced so a literal
    ``<tool_call|>`` in a quoted argument cannot truncate the span. An incomplete
    span is dropped to EOF when ``final``, else kept (still streaming)."""
    out: list[str] = []
    cursor = 0
    for match in _TC_GEMMA_START_RE.finditer(text):
        start = match.start()
        if start < cursor:
            continue
        brace_end = _balanced_brace_end(text, match.end() - 1, gemma_quotes = True)
        if brace_end < 0:
            # Unbalanced: nothing completes from here on. Drop the rest if final,
            # else keep it; stop either way (rescanning would be quadratic).
            if final:
                out.append(text[cursor:start])
                cursor = len(text)
            break
        # Junk between } and <tool_call|> is malformed-call markup: strip through
        # the close, keep text after it. No close anywhere means stop (linear).
        close = _TC_GEMMA_END_TAG_RE.search(text, brace_end + 1)
        if close is None:
            if final:
                out.append(text[cursor:start])
                cursor = len(text)
            break
        out.append(text[cursor:start])
        cursor = close.end()
    out.append(text[cursor:])
    return "".join(out)


def _gemma_span_ranges(text: str) -> list:
    """``(start, end)`` of each complete Gemma-native span; same walk as
    ``_strip_gemma_native_spans`` without stripping."""
    ranges: list[tuple] = []
    cursor = 0
    for match in _TC_GEMMA_START_RE.finditer(text):
        start = match.start()
        if start < cursor:
            continue
        brace_end = _balanced_brace_end(text, match.end() - 1, gemma_quotes = True)
        if brace_end < 0:
            break
        close = _TC_GEMMA_END_TAG_RE.search(text, brace_end + 1)
        if close is None:
            break
        ranges.append((start, close.end()))
        cursor = close.end()
    return ranges


def _strip_closed_blocks_outside_gemma(text: str) -> str:
    """Closed JSON/function pre-pass that skips matches starting inside a complete
    Gemma span: deleting across the span boundary would mangle the Gemma close and
    truncate the tail. A skipped match resumes at the covering span's end, so a
    real function-XML call after the span is still stripped."""
    ranges = _gemma_span_ranges(text)
    if not ranges:
        return strip_tool_patterns(text, _TOOL_CLOSED_BLOCK_PATS)
    for pat in _TOOL_CLOSED_BLOCK_PATS:
        token = _PAT_REQUIRED_TOKEN.get(pat)
        if token is not None and token not in text:
            continue
        out: list[str] = []
        pos = 0
        while True:
            m = pat.search(text, pos)
            if m is None:
                out.append(text[pos:])
                break
            covering = next((r for r in ranges if r[0] <= m.start() < r[1]), None)
            if covering is not None:
                out.append(text[pos : covering[1]])
                pos = covering[1]
                continue
            out.append(text[pos : m.start()])
            pos = m.end()
        new_text = "".join(out)
        if new_text != text:
            text = new_text
            ranges = _gemma_span_ranges(text)
    return text


def strip_tool_markup_final(text: str) -> str:
    """Final display strip, shared with the streaming wrappers so all paths order
    the passes identically: Gemma-aware closed JSON/function blocks first, then
    well-formed Gemma spans (quote-aware), then the regex sweeps mop up malformed
    spans and drop any unclosed remainder to EOF. Whitespace is kept."""
    text = _strip_closed_blocks_outside_gemma(text)
    text = _strip_gemma_native_spans(text, final = True)
    return strip_tool_patterns(text, _TOOL_ALL_PATS)


def strip_tool_call_markup(text: str, *, final: bool = False) -> str:
    """Strip tool-call XML markup from text.

    When ``final`` is False, only fully closed tool-call blocks are removed.
    When ``final`` is True, trailing incomplete tool-call blocks are removed
    too, and the result is stripped of surrounding whitespace.
    """
    if final:
        return strip_tool_markup_final(text).strip()
    # Non-final: same ordering as the final path, but incomplete blocks are kept.
    text = _strip_closed_blocks_outside_gemma(text)
    text = _strip_gemma_native_spans(text, final = False)
    return strip_tool_patterns(text, _TOOL_CLOSED_PATS)