import gradio as gr
import subprocess
import re
import os
import uuid
import json
import random
import html
import threading
from queue import Queue, Empty
from typing import Optional, List, Any, Dict, Tuple
from run_question import answer_question_recall
# --------------------------------------------------------------------
# Required part – start the serper host server (required for fathom search)
# (left UNCHANGED as requested)
# --------------------------------------------------------------------
port = os.environ["PORT_SERPER_HOST"]
# lsof = subprocess.run(["lsof", "-i", f":{port}"], capture_output=True)
# if len(lsof.stdout) == 0:
# print("Starting serper host server")
# _ = subprocess.Popen(["nohup", "sh", "./web_agents_5/host_serper2.sh"])
try:
print("Starting serper host server")
_ = subprocess.Popen(["nohup", "sh", "./web_agents_5/host_serper2.sh"])
except Exception as e:
print(f"Failed to start serper host server: {e}")
pass
# --------------------------------------------------------------------
# Rendering options
# --------------------------------------------------------------------
ORDER_TOOL_CALLS_FIRST = False # keep assistant text before calls
SPACER = '' # small vertical gap
MIN_TOOLCALLOUT_CHARS = 160 # hide calls in *content* if text is too short
PREVIEW_LIMIT = 135 # condensed preview length
BOXED_WRAP_WIDTH = int(os.getenv("BOXED_WRAP_WIDTH", "130")) # boxed soft-wrap width (chars)
# --------------------------------------------------------------------
# Helpers
# --------------------------------------------------------------------
INVISIBLES = ('\ufeff', '\u200b', '\u200d') # BOM/ZWSP/ZWNJ
def _strip_leading_garbage(md: str) -> str:
if not isinstance(md, str):
md = str(md)
# remove invisible chars
for ch in INVISIBLES:
md = md.replace(ch, '')
# normalise newlines
md = md.replace('\r\n', '\n').replace('\r', '\n')
# drop ALL leading blank/space-only lines
md = re.sub(r'^\s*\n', '', md, count=0, flags=re.M)
return md
def _ensure_md_blocks(md: str) -> str:
# ensure a blank line after any Markdown heading
md = re.sub(r'(^|\n)(#{1,6}[^\n]+)\n(?!\n)', r'\1\2\n\n', md)
return md
def format_math(text: str) -> str:
"""
No-op math formatter.
Previously this function attempted to:
- Convert [ ... ] into $$ ... $$, and
- Convert \( ... \) into $ ... $
That interfered with already-sanitized LaTeX (ensure_display_math), producing
nested $$ inside $$ blocks and breaking KaTeX rendering.
We now leave the string unchanged to avoid introducing invalid math delimiters
post-sanitization.
"""
if not isinstance(text, str):
return str(text)
return text
def generate_conversation_id():
return str(uuid.uuid4())[:8]
def _maybe_json_load(s: Any) -> Any:
if isinstance(s, (dict, list, int, float, bool)) or s is None:
return s
if not isinstance(s, str):
return s
for _ in range(2):
try:
s = json.loads(s)
except Exception:
break
return s
def _pretty_json(anything: Any) -> str:
obj = _maybe_json_load(anything)
if isinstance(obj, (dict, list, int, float, bool)) or obj is None:
try:
return json.dumps(obj, ensure_ascii=False, indent=2)
except Exception:
return str(obj)
if isinstance(obj, str):
if "\\n" in obj or "\\t" in obj:
obj = obj.replace("\\n", "\n").replace("\\t", "\t")
return obj
return str(obj)
def _code_block(lang: str, text: Any) -> str:
s = str(text) if not isinstance(text, str) else text
s = s.replace("```", "``\u200b`") # avoid fence confusion
return f"```{lang}\n{s}\n```"
def _normalize_tool_call(tc: Any) -> Dict[str, Any]:
if isinstance(tc, dict):
return {
"name": tc.get("name") or "tool",
"arguments": tc.get("arguments") or tc.get("args") or {}
}
parsed = _maybe_json_load(tc)
if isinstance(parsed, dict):
return {
"name": parsed.get("name") or "tool",
"arguments": parsed.get("arguments") or parsed.get("args") or {}
}
return {"name": "tool", "arguments": {"raw": str(tc)}}
def _letter(i: int) -> str:
return f"{chr(96 + i)})" # 1->a), 2->b), ...
def _summarize_call(tc: Any) -> Tuple[str, str]:
norm = _normalize_tool_call(tc)
name = (norm.get("name") or "tool").lower()
args = norm.get("arguments") or {}
if name == "search_urls":
return "Search", str(args.get("query", "")).strip() or "(no query)"
if name == "query_url":
return "Query URL", str(args.get("url", "")).strip() or "(no url)"
label = norm.get("name") or "tool"
val = json.dumps(args, ensure_ascii=False) if isinstance(args, (dict, list)) else str(args)
if len(val) > PREVIEW_LIMIT:
val = val[:PREVIEW_LIMIT] + "…"
return label, val
def _render_tool_calls_details(tool_calls: List[Any], tool_results: Optional[List[Any]] = None) -> str:
if not tool_calls:
return ""
tool_results = tool_results or []
parts = []
parts.append("Tool calls:")
parts.append(SPACER)
for i, tc in enumerate(tool_calls, 1):
label, value = _summarize_call(tc)
summary_line = f"{_letter(i)} {label}: {value}"
call_json = _pretty_json(tc)
result_json = _pretty_json(tool_results[i - 1]) if (i - 1) < len(tool_results) else None
parts.append(f"{summary_line}
\n")
parts.append("Tool Call
\n\n")
parts.append(_code_block("json", call_json))
parts.append("\n")
parts.append("Tool Result
\n\n")
if result_json is None:
parts.append("_(pending)_\n")
else:
parts.append(_code_block("json", result_json))
parts.append("\n")
parts.append(""
f"{format_math(text_md).strip()}"
f"
\n\n"
f"{SPACER}"
)
def _wrap_turn(inner_html: str) -> str:
return (
f""
f"{inner_html}
"
)
def _wrap_condensed_step(inner_html: str) -> str:
"""Yellow sub-bubble wrapper for each condensed step inside the blue cluster."""
return (
f""
f"{inner_html}
"
)
def _condensed_step_block(assistant_text: str,
tool_calls: Optional[List[Any]] = None,
tool_results: Optional[List[Any]] = None,
last_open: bool = False) -> str:
"""A single condensed step: details with assistant preview + (full assistant + tool calls) inside, wrapped yellow.
last_open: when True, render this step expanded by default.
"""
preview = _plain_preview(assistant_text, PREVIEW_LIMIT)
inner_parts = []
inner_parts.append(_assistant_section(assistant_text))
if tool_calls:
inner_parts.append(_render_tool_calls_details(tool_calls, tool_results))
inner_html = "".join(inner_parts).strip()
open_attr = " open" if last_open else ""
details_html = (
f"Thinking (preview): {preview}\n\n"
f"{inner_html}\n\n"
)
return _wrap_condensed_step(details_html)
def _cluster_wrapper(inner_html: str) -> str:
"""Blue cluster bubble that holds all condensed steps. Expanded by default."""
return (
f""
f"Initial thinking, steps, tool calls
\n\n"
f"{inner_html}\n\n"
f"
"
)
# import re
# LATEX_SPECIALS = {'%': r'\%', '#': r'\#', '&': r'\&', '_': r'\_'}
LATEX_ESC_MAP_IN_MATH = {
'%': r'\%',
'#': r'\#',
# NOTE: do NOT escape & or _ in math; they’re valid there
}
def _escape_unescaped_dollars(s: str) -> str:
# $ -> \$ unless already escaped
return re.sub(r'(? str:
return re.sub(r'\*\*(.+?)\*\*', r'\\textbf{\1}', s)
def _escape_other_specials_in_math(s: str) -> str:
return re.sub(r'(? str:
"""
Remove any standalone $$ tokens that appear inside math content (not proper $$…$$ pairs).
We do this BEFORE escaping $, so they don't turn into literal '\$\$'.
"""
# Collapse isolated $$ on their own lines or with spaces around
s = re.sub(r'(^|\s)\$\$(\s|$)', lambda m: (m.group(1) or '') + (m.group(2) or ''), s)
# Also remove any leftover pairs of $$ that aren't part of a proper $$…$$ block
# (very conservative: delete $$ that are not followed by content then $$ soon)
s = re.sub(r'\$\$(?=\s*(?:\\end|\)|}|$))', '', s)
s = re.sub(r'(?:(?<=\\begin\{aligned\})|\n)\s*\$\$\s*', '', s)
return s
def _mask_text_blocks(s: str):
"""
Temporarily mask \text{...} spans (balanced) so we don’t insert \allowbreak inside them.
Returns masked string and list of (token, original).
"""
out, tokens = [], []
i, n = 0, len(s)
tcount = 0
while i < n:
j = s.find(r'\text{', i)
if j == -1:
out.append(s[i:]); break
out.append(s[i:j])
# balance braces
k = j + len(r'\text{'); depth = 1
while k < n and depth > 0:
c = s[k]
if c == '\\' and k + 1 < n:
k += 2; continue
if c == '{': depth += 1
elif c == '}': depth -= 1
k += 1
block = s[j:k] if depth == 0 else s[j:]
token = f'__TEXTBLOCK_{tcount}__'
tokens.append((token, block))
out.append(token)
i = k if depth == 0 else n
tcount += 1
return ''.join(out), tokens
def _unmask_text_blocks(s: str, tokens) -> str:
for token, block in tokens:
s = s.replace(token, block)
return s
def _mask_command_blocks(s: str, commands: List[str]):
"""
Mask balanced \cmd{...} spans for specific single-argument commands so we can
safely wrap surrounding plain text into \text{...} without splitting inside these blocks.
IMPORTANT:
- Only mask single-argument commands (e.g., \textbf{...}, \emph{...}).
- Do NOT mask multi-argument commands (e.g., \href{..}{..}) to avoid brace imbalance.
Returns (masked_string, tokens) where tokens is a list of (token, original_block).
"""
try:
tokenized = s or ""
tokens: List[Tuple[str, str]] = []
tcount = 0
for cmd in (commands or []):
pattern = r'\\' + cmd + r'{'
i = 0
n = len(tokenized)
while i < n:
j = tokenized.find(pattern, i)
if j == -1:
break
# scan balanced braces for this single-arg command
k = j + len(pattern)
depth = 1
while k < n and depth > 0:
c = tokenized[k]
if c == '\\' and k + 1 < n:
k += 2
continue
if c == '{':
depth += 1
elif c == '}':
depth -= 1
k += 1
if depth == 0:
block = tokenized[j:k]
token = f'__CMDBLOCK_{tcount}__'
tokens.append((token, block))
tokenized = tokenized[:j] + token + tokenized[k:]
n = len(tokenized)
i = j + len(token)
tcount += 1
else:
# Unbalanced command; do not mask to avoid dropping trailing content
i = j + len(pattern)
return tokenized, tokens
except Exception:
return s, []
def _add_break_opportunities_outside_text(s: str) -> str:
"""
Add \allowbreak only outside \text{...} spans.
"""
masked, toks = _mask_text_blocks(s)
# Soft breaks after slashes
masked = re.sub(r'(? str:
r"""
Inside math, handle model patterns like: $$Label$$(https://example.com)
Be robust to \allowbreak{} sprinkled by earlier passes.
1) $$Label$$(url) -> \href{url}{Label}
2) any remaining $$...$$ -> \text{...}
"""
# allow break markers inside URLs
AB = r'(?:\\allowbreak\{\})*'
# url = protocol://... (allow \allowbreak{} anywhere)
url_pat = rf'(https?{AB}:{AB}//{AB}[^)]+?)'
# 1) $$Label$$(url) -> \href{url}{Label}
s = re.sub(
rf'\$\$(.+?)\$\$\({url_pat}\)',
r'\\href{\2}{\1}',
s,
flags=re.S
)
# 2) Any remaining $$...$$ (malformed or not a link) -> \text{...}
s = re.sub(r'\$\$(.+?)\$\$', r'\\text{\1}', s, flags=re.S)
return s
# --- add this final safety net (remove ANY nested $$ left in math) ---
def _strip_all_nested_dollars(s: str) -> str:
"""
After all transforms, if any '$$' survive inside math content,
convert the smallest $$...$$ spans to \text{...}. Repeat until gone.
"""
# keep replacing until no $$ remain
while '$$' in s:
new_s = re.sub(r'\$\$(.+?)\$\$', r'\\text{\1}', s, flags=re.S)
if new_s == s:
# if we couldn't match a pair, drop lone $$ defensively
new_s = s.replace('$$', '')
s = new_s
return s
def _cleanup_tex_command_braces_and_percent(s: str) -> str:
"""
Final cleanup for LaTeX text:
- Fix escaped brace after single-argument commands: \textbf\{ -> \textbf{, \emph\{ -> \emph{
- Collapse double-backslash percent: \\% -> \% (avoids '%' commenting the rest of the line after a linebreak)
Patterns are conservative and won't affect '\\\\' row breaks since they aren't followed by '%'.
"""
try:
# 1) Repair malformed single-arg command braces
s = re.sub(r'\\(text|textbf|emph)\\\{', r'\\\1{', s)
# 2) Ensure percent is escaped with exactly one backslash
s = re.sub(r'\\{2}%', r'\\%', s)
return s
except Exception:
return s
def _normalize_escaped_closing_braces_for_single_arg_commands(s: str, commands: List[str]) -> str:
"""
Regex-normalize cases like '\textbf{Batters\}:' -> '\textbf{Batters}:'
Works for simple single-arg commands where the content has no nested braces.
This is a fast-path fix; more complex cases are handled by the scanner-based repair.
"""
try:
out = s or ""
for cmd in (commands or []):
# \cmd{ ... \}} -> \cmd{ ... }}
out = re.sub(rf'\\{cmd}\s*\{{([^{{}}]*?)\\\}}', rf'\\{cmd}{{\1}}', out)
return out
except Exception:
return s
def _repair_unclosed_single_arg_commands(s: str, commands: List[str]) -> str:
"""
Heuristically repair single-argument commands like \textbf{...}, \emph{...}, \text{...}
that were incorrectly closed with a literal '\}' instead of '}' (leaving the group
unbalanced and causing 'File ended while scanning ...').
Supports optional whitespace between the command and '{'.
Only affects the specified single-argument commands and only when their top-level
argument fails to close.
"""
try:
out = s or ""
for cmd in (commands or []):
pat = re.compile(r'\\' + cmd + r'\s*{')
i = 0
while True:
m = pat.search(out, i)
if not m:
break
j = m.start()
k = m.end()
n = len(out)
depth = 1
last_esc_close_top = -1
while k < n:
c = out[k]
if c == '\\' and k + 1 < n:
# record top-level literal '\}' so we can convert it if needed
if out[k + 1] == '}' and depth == 1:
last_esc_close_top = k
k += 2
continue
if c == '{':
depth += 1
elif c == '}':
depth -= 1
if depth == 0:
break
k += 1
if depth == 0:
# Properly closed, continue after this block
i = k + 1
continue
# Unclosed: promote the last top-level '\}' to '}' and rescan
if last_esc_close_top != -1:
out = out[:last_esc_close_top] + out[last_esc_close_top + 1:]
# Rescan from the same command start (structure changed)
i = j
continue
# Nothing to fix; skip past this occurrence to avoid infinite loop
i = m.end()
return out
except Exception:
return s
def _repair_unclosed_single_arg_commands_in_text_groups(s: str, commands: List[str]) -> str:
"""
Within each \text{...} group, repair unclosed single-arg commands like \textbf{...}, \emph{...}
where the closing brace was emitted as '\}' and thus remained unbalanced inside the text group.
This confines the scan to the text group's inner content so outer braces aren't misinterpreted.
"""
try:
out: List[str] = []
i, n = 0, len(s or "")
text_pat = re.compile(r'\\text\s*{')
while i < n:
m = text_pat.search(s, i)
if not m:
out.append(s[i:])
break
# emit up to and including '\text{'
out.append(s[i:m.end()])
# find matching closing '}' for this \text{
k = m.end()
depth = 1
while k < n and depth > 0:
c = s[k]
if c == '\\' and k + 1 < n:
k += 2
continue
if c == '{':
depth += 1
elif c == '}':
depth -= 1
k += 1
if depth != 0:
# unmatched -> emit rest and stop
out.append(s[m.end():])
break
inner = s[m.end():k-1]
fixed_inner = _repair_unclosed_single_arg_commands(inner, commands)
out.append(fixed_inner)
out.append('}')
i = k
return ''.join(out) or s
except Exception:
return s
def _unwrap_envs_from_text(s: str) -> str:
"""
If the model emits \text{ \begin{env} ... \end{env} }, unwrap the environment
from the \text{...} wrapper so it's in math mode again.
Run repeatedly until nothing to unwrap.
"""
# generic LaTeX environments, e.g., aligned, align, cases, bmatrix, etc.
env_pat = re.compile(r'\\text\s*\{\s*(\\begin\{[a-zA-Z*]+\}.*?\\end\{[a-zA-Z*]+\})\s*\}', re.S)
# also handle \text{\boxed{...}} -> \boxed{...}
boxed_pat = re.compile(r'\\text\s*\{\s*(\\boxed\{.*?\})\s*\}', re.S)
while True:
new_s = env_pat.sub(r'\1', s)
new_s = boxed_pat.sub(r'\1', new_s)
if new_s == s:
break
s = new_s
return s
def _sanitize_inside_math(s: str) -> str:
# 0) FIX: get environments out of \text{...}
s = _unwrap_envs_from_text(s)
# 1) markdown -> tex
s = _markdown_bold_to_tex(s)
# 2) convert $$Label$$(url) and $$...$$ early (handles \allowbreak in URLs)
s = _fix_inline_links_in_math(s)
# 3) escape raw $ (prices) AFTER dealing with $$…$$
s = _escape_unescaped_dollars(s)
# 4) escape only % and # (leave & and _ alone for math)
s = _escape_other_specials_in_math(s)
# 5) add wrap hints (avoid inside \text{...})
# s = _add_break_opportunities_outside_text(s)
# 6) FINAL guard: if any $$ survived, neutralize them
s = _strip_all_nested_dollars(s)
return s
def _force_multiline_box_content(s: str) -> str:
"""
If a \boxed{...} inner content appears to be plain text (no multiline constructs),
render it as multiple lines with left alignment.
Heuristics (applied in order):
- If inner already has \begin{...}, \end{...}, \aligned, '\\\\' or '&', leave as-is.
- If it contains explicit '\\' breaks without an environment, split on them.
- If content looks like a JSON array [ {...}, {...}, ... ], render one object per row.
- Split on ' > ' (or any whitespace '>' whitespace) to break comparative/ranking summaries.
- Prefer splitting at numbered list markers like '1. ', '2. '.
- Otherwise honor explicit newlines.
- As fallback for very long text, split at ' - ' bullets.
Formatting:
- Use \\begin{array}{l} ... \\end{array} so rows are left-aligned.
- Wrap a row into \\text{...} when it has no TeX commands already.
- Escape { } _ & inside \\text{...}.
"""
def _fmt_lines_array(lines: List[str]) -> str:
out = []
for l in lines:
l = l.strip()
if not l:
continue
if re.search(r'\\[a-zA-Z]+', l):
# Mixed content: keep existing TeX commands, but wrap plain text runs in \text{...}
try:
masked_text, toks_text = _mask_text_blocks(l)
masked_cmd, toks_cmd = _mask_command_blocks(masked_text, ["textbf", "emph"])
# Split around tokens and wrap non-token chunks
parts = re.split(r'(__TEXTBLOCK_\d+__|__CMDBLOCK_\d+__)', masked_cmd)
rebuilt = []
for part in parts:
if not part:
continue
if part.startswith('__TEXTBLOCK_') or part.startswith('__CMDBLOCK_'):
rebuilt.append(part)
else:
# wrap plain chunk in \text{...} with text-mode escaping
esc = (
part.replace('{', r'\{')
.replace('}', r'\}')
.replace('_', r'\_')
.replace('&', r'\&')
.replace('%', r'\%')
.replace('#', r'\#')
)
rebuilt.append(r'\text{' + esc + '}')
mixed = "".join(rebuilt)
# Unmask in reverse order: commands first, then existing \text blocks
mixed = _unmask_text_blocks(mixed, toks_cmd)
mixed = _unmask_text_blocks(mixed, toks_text)
out.append(mixed)
except Exception:
out.append(l)
else:
esc = (
l.replace('{', r'\{')
.replace('}', r'\}')
.replace('_', r'\_')
.replace('&', r'\&')
.replace('%', r'\%')
.replace('#', r'\#')
)
out.append(r'\text{' + esc + '}')
return r'\begin{array}{l}' + r' \\\\ '.join(out) + r'\end{array}'
try:
txt = s or ""
# Already structured -> leave unchanged
# NOTE: Do not treat a bare '&' as "structured"; allow multiline formatting even if '&' exists in plain text.
if re.search(r'(\\begin\{|\\end\{|\\aligned|\\\\)', txt):
return txt
# SAFETY: Avoid inserting breaks inside \text{...} groups by masking them first
masked_txt, toks = _mask_text_blocks(txt)
# Force a line break after every ':' except URL schemes like 'http://' or 'https://'
masked_txt = re.sub(r':(?!//)', ':\n', masked_txt)
# Force a line break before every ' - ' bullet if not already at a new line.
# Conservative rule: only when surrounded by spaces, so hyphens in words/URLs aren't affected.
masked_txt = re.sub(r'\s-\s', '\n- ', masked_txt)
# Enforce max line length: 130 chars (BOXED_WRAP_WIDTH) by wrapping at the NEXT whitespace >= width
# 1) Split on existing/inserted newlines first
parts = [ln.strip() for ln in re.split(r'\s*\n+\s*', masked_txt) if ln.strip()]
if not parts:
parts = [masked_txt.strip()]
# 2) Soft-wrap each part by spaces without breaking masked \text{...} tokens.
# - Prefer the first whitespace at or after the width threshold.
# - If a single token exceeds width and has no whitespace, keep it on one line (do not split tokens).
W = BOXED_WRAP_WIDTH
def _wrap_masked_by_space_next(s: str, width: int) -> List[str]:
out: List[str] = []
s = s.strip()
while len(s) > width:
# find first whitespace at or after width
m = re.search(r'\s', s[width:])
if m:
cut = width + m.start()
out.append(s[:cut].rstrip())
s = s[cut+1:].lstrip() # drop the breaking whitespace
continue
# no whitespace ahead; if there's whitespace before, break there
prev = s.rfind(' ')
if prev != -1:
out.append(s[:prev].rstrip())
s = s[prev+1:].lstrip()
else:
# single long token; emit as a line without splitting
out.append(s)
s = ""
if s:
out.append(s)
return out
wrapped_masked: List[str] = []
for p in parts:
wrapped_masked.extend(_wrap_masked_by_space_next(p, W))
# 3) If we produced multiple rows, unmask and return as left-aligned array
if len(wrapped_masked) > 1:
unmasked_lines = [_unmask_text_blocks(line, toks) for line in wrapped_masked]
return _fmt_lines_array(unmasked_lines)
# Single line after wrapping; unmask back and continue with other heuristics
txt = _unmask_text_blocks(wrapped_masked[0], toks)
# 0) If explicit math line breaks (\\) are present without an environment, split on them
if re.search(r'\\\\', txt):
lines = [ln.strip() for ln in re.split(r'\s*\\\\\s*', txt) if ln.strip()]
if lines:
return _fmt_lines_array(lines)
# 1) JSON array lines (e.g., [ {..}, {..}, ... ])
m = re.match(r'^\s*\[(.*)\]\s*$', txt, flags=re.S)
if m:
inner = m.group(1)
parts = re.split(r'\},\s*', inner)
lines = []
for p in parts:
p = p.strip()
if not p:
continue
if not p.endswith('}'):
p = p + '}'
p = p.lstrip(',')
lines.append(p)
if lines:
return _fmt_lines_array(lines)
# 2) Split on '>' separators commonly used in rankings
gt_parts = [p.strip() for p in re.split(r'\s*>\s*', txt) if p and p.strip()]
if len(gt_parts) > 1:
lines = []
for idx, part in enumerate(gt_parts):
lines.append(part if idx == 0 else ('> ' + part))
return _fmt_lines_array(lines)
# 3) Assumptions block: split heading and bullets/newlines into separate rows
# Supports '**Assumptions**:' or '\textbf{Assumptions}:' or plain 'Assumptions:'
m_assume = re.search(r'(?im)^\s*(?:\*{0,2}\s*)?(?:\\textbf\{)?Assumptions(?:\})?\s*:?', txt)
if m_assume:
head = txt[:m_assume.start()].strip()
tail = txt[m_assume.start():].strip()
# Split head as numbered list or by newlines
head_tokens = re.split(r'(?=(?:^|\n|\s)\d+\.\s)', head)
head_lines = [t.strip() for t in head_tokens if t and t.strip()]
if len(head_lines) <= 1:
head_lines = [ln.strip() for ln in re.split(r'\s*\n+\s*', head) if ln.strip()]
# Split tail by newlines first; if it stays one line, split by '- ' bullets
tail_lines = [ln.strip() for ln in re.split(r'\s*\n+\s*', tail) if ln.strip()]
if len(tail_lines) <= 1:
tail_lines = [p.strip() for p in re.split(r'\s+-\s+', tail) if p.strip()]
lines = head_lines + tail_lines
if lines:
return _fmt_lines_array(lines)
# 4) Prefer split at numbered list markers like '1. ', '2. '
tokens = re.split(r'(?=(?:^|\n|\s)\d+\.\s)', txt)
lines = [t.strip() for t in tokens if t and t.strip()]
# 4) If not split, honor explicit newlines if present
if len(lines) <= 1:
nl_lines = [ln.strip() for ln in re.split(r'\s*\n+\s*', txt) if ln.strip()]
if len(nl_lines) > 1:
lines = nl_lines
# 5) Fallback for very long text: split on bullet-like dashes
if len(lines) <= 1 and len(txt.strip()) > 120:
bullet_lines = [b.strip() for b in re.split(r'\s+-\s+', txt) if b.strip()]
if len(bullet_lines) > 1:
lines = bullet_lines
if len(lines) > 1:
return _fmt_lines_array(lines)
# 8) Width-based soft wrap at 80 chars (split at nearest space)
W = BOXED_WRAP_WIDTH
txt_stripped = txt.strip()
if len(txt_stripped) > W:
def _soft_wrap_by_space(t: str, width: int = 80) -> List[str]:
rows: List[str] = []
s2 = t.strip()
while len(s2) > width:
cut = s2.rfind(' ', 0, width + 1)
if cut == -1:
cut = width
rows.append(s2[:cut].strip())
s2 = s2[cut:].lstrip()
if s2:
rows.append(s2)
return rows
soft_lines = _soft_wrap_by_space(txt_stripped, W)
if len(soft_lines) > 1:
return _fmt_lines_array(soft_lines)
# Final safeguard for single-line boxed content:
# Render upright (non-italic). If TeX commands are present, wrap only plain fragments
# in \text{...}; otherwise wrap the whole line in \text{...}.
def _escape_text_mode_local(s2: str) -> str:
return (
s2.replace('{', r'\{')
.replace('}', r'\}')
.replace('_', r'\_')
.replace('&', r'\&')
.replace('%', r'\%')
.replace('#', r'\#')
)
def _wrap_plain_outside_commands_local(s2: str) -> str:
n2 = len(s2)
i2 = 0
parts2: List[str] = []
plain2: List[str] = []
def flush_plain2():
if not plain2:
return
chunk2 = ''.join(plain2)
if chunk2:
parts2.append(r'\text{' + _escape_text_mode_local(chunk2) + '}')
plain2.clear()
while i2 < n2:
ch2 = s2[i2]
if ch2 == '\\':
# finish accumulated plain text before emitting the command
flush_plain2()
j2 = i2 + 1
# read command name (letters)
while j2 < n2 and s2[j2].isalpha():
j2 += 1
# handle escapes like \%, \_, \{, \}
if j2 == i2 + 1:
if i2 + 1 < n2:
parts2.append(s2[i2:i2+2])
i2 += 2
else:
parts2.append('\\')
i2 += 1
continue
# capture following balanced brace groups (arguments), if any
k2 = j2
while k2 < n2 and s2[k2] == '{':
depth2 = 1
t = k2 + 1
while t < n2 and depth2 > 0:
c3 = s2[t]
if c3 == '\\' and t + 1 < n2:
t += 2
continue
if c3 == '{':
depth2 += 1
elif c3 == '}':
depth2 -= 1
t += 1
k2 = t
if depth2 != 0:
break
parts2.append(s2[i2:(k2 if k2 > j2 else j2)])
i2 = (k2 if k2 > j2 else j2)
else:
plain2.append(ch2)
i2 += 1
flush_plain2()
return ''.join(parts2)
if not re.search(r'\\[a-zA-Z]+', txt):
# No TeX commands at all → wrap whole line
esc = _escape_text_mode_local(txt_stripped)
return r'\text{' + esc + '}'
else:
# Mixed content on a single line → wrap only the plain fragments
return _wrap_plain_outside_commands_local(txt)
except Exception:
return s
def ensure_display_math(s: str) -> str:
"""
Sanitize math content robustly WITHOUT double-wrapping:
- Sanitize content of EXISTING $$…$$ blocks in place.
- For \begin{aligned}…\end{aligned}: if outside $$, wrap; if inside, just sanitize.
- For \boxed{…}: if outside $$, wrap; if inside, just sanitize.
- ALSO: unwrap envs accidentally wrapped in \text{...} at the whole-string level.
"""
if not isinstance(s, str) or not s:
return s
# 🔧 unwrap envs/boxed placed incorrectly inside \text{...} (top level)
s = _unwrap_envs_from_text(s)
display_blocks = []
def _sanitize_display(m):
start, end = m.span()
content = m.group(1)
content = _sanitize_inside_math(content)
display_blocks.append((start, start + 2, end - 2, end))
return "$$\n" + content + "\n$$"
s = re.sub(r'\$\$(.*?)\$\$', _sanitize_display, s, flags=re.S)
# Recompute display '$$...$$' spans in the NEW string so inside checks are accurate
def _collect_display_spans_in_text(text: str):
spans = []
i3, n3 = 0, len(text)
while i3 < n3:
a = text.find('$$', i3)
if a == -1:
break
b = text.find('$$', a + 2)
if b == -1:
break
spans.append((a, b + 2))
i3 = b + 2
return spans
display_spans = _collect_display_spans_in_text(s)
def _inside_display(idx: int) -> bool:
for (a, b) in display_spans:
if a <= idx <= b:
return True
return False
# Collect boxed spans to prevent wrapping aligned/array inside \boxed{...} with $$
def _collect_boxed_spans(text: str):
spans = []
i2, n2 = 0, len(text)
while i2 < n2:
j2 = text.find(r'\boxed{', i2)
if j2 == -1:
break
k2 = j2 + len(r'\boxed{'); depth2 = 1
while k2 < n2 and depth2 > 0:
c2 = text[k2]
if c2 == '\\' and k2 + 1 < n2:
k2 += 2; continue
if c2 == '{':
depth2 += 1
elif c2 == '}':
depth2 -= 1
if depth2 == 0:
spans.append((j2, k2))
break
k2 += 1
i2 = k2 if k2 > j2 else j2 + 1
return spans
boxed_spans = _collect_boxed_spans(s)
def _inside_boxed(idx: int) -> bool:
for (a, b) in boxed_spans:
if a <= idx <= b:
return True
return False
# aligned (only wrap with $$ if not already inside display or boxed)
def _proc_aligned(m):
start, _ = m.span(1)
blk = _sanitize_inside_math(m.group(1))
return blk if (_inside_display(start) or _inside_boxed(start)) else "$$\n" + blk + "\n$$"
s = re.sub(r'(\\begin\{aligned\}.*?\\end\{aligned\})', _proc_aligned, s, flags=re.S)
# Pre-normalize escaped closing braces like '\}' inside single-arg commands to real '}' so subsequent scans see balanced braces
s = _normalize_escaped_closing_braces_for_single_arg_commands(s, ["textbf", "emph"])
# boxed (balanced-scan)
out = []; i, n = 0, len(s)
while i < n:
j = s.find(r'\boxed{', i)
if j == -1:
out.append(s[i:]); break
out.append(s[i:j])
k = j + len(r'\boxed{'); depth = 1
while k < n and depth > 0:
c = s[k]
if c == '\\' and k + 1 < n:
k += 2; continue
if c == '{': depth += 1
elif c == '}': depth -= 1
k += 1
if depth != 0:
out.append(s[j:]); i = n; break
inner = s[j+len(r'\boxed{'):k-1]
inner = _sanitize_inside_math(inner)
inner = _force_multiline_box_content(inner)
fixed = r'\boxed{' + inner + '}'
out.append(fixed if _inside_display(j) else "$$\n" + fixed + "\n$$")
i = k
s = "".join(out)
# 🔧 unwrap again in case earlier steps exposed a fresh \text{ \begin...\end... }
s = _unwrap_envs_from_text(s)
# 🔧 final cleanup for malformed commands and percent escapes
s = _cleanup_tex_command_braces_and_percent(s)
# 🔧 normalize simple '\}' -> '}' within single-arg commands (fast path)
s = _normalize_escaped_closing_braces_for_single_arg_commands(s, ["textbf", "emph", "text"])
# 🔧 first, repair unclosed single-arg commands INSIDE \text{...} groups
s = _repair_unclosed_single_arg_commands_in_text_groups(s, ["textbf", "emph"])
# 🔧 then, global repair for any remaining single-arg commands
s = _repair_unclosed_single_arg_commands(s, ["textbf", "emph", "text"])
return s
def _final_response_bubble(text_md: str) -> str:
"""Standalone GREEN bubble for the last/primary narrative."""
# Log exactly final response before/after LaTeX processing (only two prints)
# try:
# # print("FINAL_RESPONSE_RAW_BEFORE_LATEX:\n", text_md)
# pass
# except Exception:
# pass
text_md = ensure_display_math(text_md)
# try:
# # print("FINAL_RESPONSE_STRUCTURED_AFTER_LATEX:\n", format_math(text_md).strip())
# pass
# except Exception:
# pass
nlc = ' \n\n'
return (
f""
f"
Final Response:\n\n"
f"
"
f"{nlc + format_math(text_md).strip()}"
f"
\n\n"
f"
"
f"
{frame}
"
f"
{label}
"
f"
"
f"
"
f"
{_truncate_50(esc_snip)}
"
"