"""Color-coded rewritten bullets and turn-table HTML for Streamlit."""
from __future__ import annotations
import html
import re
from agent.state import ChatTurn, Claim, MessageAudit
_COLORS = {
"supported": ("#d1fae5", "#065f46"),
"uncertain": ("#ffedd5", "#9a3412"),
"contradicted": ("#fee2e2", "#991b1b"),
"non_claim": (None, None),
}
_PHASE_LABELS = {
"queued": "Queued…",
"extracting": "Extracting bullets…",
"searching": "Searching the web…",
"judging": "Judging claims…",
"running": "Auditing…",
}
_TABLE_CSS = """
"""
def claims_to_html(
claims: list[Claim],
*,
colored: bool = True,
include_comment: bool = True,
) -> str:
if not claims:
return ""
items: list[str] = []
for claim in claims:
verdict = claim.get("verdict", "uncertain") if colored else "non_claim"
bg, fg = _COLORS.get(verdict, _COLORS["uncertain"])
text = html.escape(claim.get("text", ""))
comment = claim.get("comment") or ""
reason = claim.get("reason") or comment
tip = html.escape(f"{verdict} · {reason}" if colored else text)
if bg is None:
bullet = f"{text}"
else:
bullet = (
"{text}'
)
if include_comment and comment:
bullet += (
f' '
f"({html.escape(comment)})"
)
bullet += ""
items.append(bullet)
return (
'"
)
def audit_message_html(
audit: MessageAudit,
*,
colored: bool = True,
include_comment: bool = True,
) -> str:
claims = audit.get("claims") or []
if claims:
return claims_to_html(
claims, colored=colored, include_comment=include_comment
)
rewritten = html.escape(audit.get("rewritten") or audit.get("content") or "")
return f'{rewritten}
'
def audit_status_html(
audit: MessageAudit,
*,
include_comment: bool = True,
) -> str:
"""Render in-progress audit with preliminary green/orange colors."""
phase = audit.get("phase") or audit.get("status") or "running"
label = _PHASE_LABELS.get(str(phase), "Auditing…")
claims = audit.get("claims") or []
rewritten = (audit.get("rewritten") or "").strip()
if claims or rewritten:
# Show preliminary extract colors (green = common knowledge, orange = checking).
body = audit_message_html(
audit, colored=True, include_comment=include_comment
)
return (
f"{body}"
f''
f"⏳ {html.escape(label)}
"
)
content = html.escape(audit.get("content") or "")
return (
f'{content}
'
f''
f"⏳ {html.escape(label)}
"
)
def _verdict_color(verdict: str) -> str:
colors = {
"supported": "#065f46",
"uncertain": "#9a3412",
"contradicted": "#991b1b",
"non_claim": "#6b7280",
}
return colors.get(verdict, colors["uncertain"])
def _friendly_audit_error(raw: str) -> str:
text = (raw or "").strip()
low = text.lower()
if not text:
return "Audit unavailable"
if any(
tok in low
for tok in ("503", "unavailable", "429", "resource exhausted", "timeout")
):
return "Audit unavailable"
if len(text) > 120 or text.startswith("{"):
return "Audit unavailable"
return text
def _claim_bullet_html(claim: Claim, *, colored: bool = True) -> str:
verdict = claim.get("verdict", "uncertain") if colored else "non_claim"
bg, fg = _COLORS.get(verdict, _COLORS["uncertain"])
text = html.escape(claim.get("text", ""))
reason = claim.get("reason") or claim.get("comment") or ""
tip = html.escape(f"{verdict} · {reason}" if colored else text)
if bg is None:
return f'{text}'
return (
f'{text}'
)
def _sources_details_html(citations: list) -> str:
if not citations:
return ""
links: list[str] = []
for cite in citations:
title = html.escape(cite.get("title") or "source")
uri = cite.get("uri") or ""
if uri:
safe_uri = html.escape(uri, quote=True)
links.append(
f'{title}'
)
else:
links.append(f"{title}")
n = len(links)
label = "1 source" if n == 1 else f"{n} sources"
return (
f''
f"{label}
"
f""
f" "
)
def _claim_evidence_html(claim: Claim, index: int) -> str:
verdict = claim.get("verdict", "uncertain")
comment = (claim.get("comment") or "").strip()
reason = (claim.get("reason") or "").strip()
detail = reason or comment
color = _verdict_color(str(verdict))
block = (
f''
f"{index}. {html.escape(str(verdict))}
"
)
if detail:
block += f'{html.escape(detail)}
'
else:
block += 'No reasoning yet.
'
block += _sources_details_html(claim.get("citations") or [])
return block
def evidence_cell_html(audit: MessageAudit | None) -> str:
"""Reasoning + sources for one turn (legacy single-cell helper)."""
if audit is None:
return 'Waiting for audit…
'
status = audit.get("status")
if status == "error":
err = html.escape(_friendly_audit_error(audit.get("rewritten") or ""))
return f'{err}
'
claims = audit.get("claims") or []
phase = audit.get("phase") or status or "running"
label = _PHASE_LABELS.get(str(phase), "Auditing…")
if not claims:
if status in ("running", "pending"):
return f'⏳ {html.escape(label)}
'
if status == "done":
return 'No claims extracted.
'
return f'⏳ {html.escape(label)}
'
parts = [
f'{_claim_evidence_html(claim, i)}
'
for i, claim in enumerate(claims, start=1)
]
body = "".join(parts)
if status in ("running", "pending"):
body += f'⏳ {html.escape(label)}
'
return body
def _bullets_cell_html(audit: MessageAudit | None, turn: ChatTurn) -> str:
if audit is None:
return 'Audit queued…
'
status = audit.get("status")
if status == "done":
body = audit_message_html(audit, colored=True, include_comment=False)
return f'{body}
' if body else (
'No bullets.
'
)
if status in ("running", "pending"):
return (
f''
f"{audit_status_html(audit, include_comment=False)}"
f"
"
)
if status == "error":
err = html.escape(_friendly_audit_error(audit.get("rewritten") or ""))
return f'{err}
'
_ = turn
return 'Audit queued…
'
def _turn_audit_rows(
turn: ChatTurn, audit: MessageAudit | None, row_class: str
) -> list[str]:
"""Original rowspan + one thin subrow per claim (cols 2–3 aligned)."""
original = _original_cell_html(turn)
claims = list((audit or {}).get("claims") or []) if audit else []
status = (audit or {}).get("status") if audit else None
phase = (audit or {}).get("phase") or status or "running"
phase_label = _PHASE_LABELS.get(str(phase), "Auditing…")
if not claims:
bullets = _bullets_cell_html(audit, turn)
evidence = evidence_cell_html(audit)
return [
f''
f'| {original} | '
f'{bullets} | '
f'{evidence} | '
f"
"
]
n = len(claims)
rows: list[str] = []
for i, claim in enumerate(claims):
is_last = i == n - 1
sub_class = "ht-subrow-last" if is_last else "ht-subrow"
claim_html = _claim_bullet_html(claim, colored=True)
evidence_html = _claim_evidence_html(claim, i + 1)
if is_last and status in ("running", "pending"):
evidence_html += (
f'⏳ {html.escape(phase_label)}
'
)
if i == 0:
rows.append(
f''
f'| {original} | '
f'{claim_html} | '
f'{evidence_html} | '
f"
"
)
else:
rows.append(
f''
f'| {claim_html} | '
f'{evidence_html} | '
f"
"
)
return rows
def _latex_to_mathml(latex: str, *, display: bool) -> str:
"""Convert TeX to MathML; fall back to monospace on parse errors."""
body = (latex or "").strip()
if not body:
return ""
try:
from latex2mathml.converter import convert
mathml = convert(body)
# latex2mathml defaults to inline; force block when needed.
if display:
mathml = mathml.replace('display="inline"', 'display="block"', 1)
if 'display="' not in mathml:
mathml = mathml.replace("