"""Gradio adapter layer — wires :mod:`core.chat` into Gradio events. All pure conversation logic lives in :mod:`core.chat`. This module only deals with Gradio types (``gr.update``, ``gr.Warning``) and yields the tuples that the Blocks event handlers expect. Architecture ------------ The chat surface is a custom HTML container (``
``) owned and mutated by ``static/chat.js``. The server ships **deltas**, never full snapshots, via two hidden bridge components: * ``chat_delta`` (``gr.HTML``, ``elem_id="hy-chat-delta"``): a single ``
`` whose textContent is the latest JSON payload of the form ``{"seq": N, "ops": [...]}``. The browser observes mutations on this element, parses, and applies each op to the chat container. * ``busy_marker`` (``gr.HTML``, ``elem_id="hy-busy-marker"``): drives the Send button's enabled state. ``HY_BUSY_HTML`` while the model is generating, ``HY_IDLE_HTML`` when it's done. Delta wire payload is just the new characters, and the client only re-renders the trailing in-progress paragraph; everything above it is frozen DOM that never changes again. Send-button busy-state ownership -------------------------------- The Send button's busy state is **owned entirely by the browser** — there is NO server-side ``gr.update(interactive=...)`` toggling. See ``static/app.js`` for the MutationObserver-based gate. The terminal IDLE-marker frame is shipped *separately* from any chat-content frame so that even when WebSocket is back-pressured the tiny marker frame still slips through and the UI unblocks promptly. """ from __future__ import annotations import itertools import json import logging from typing import Iterator import gradio as gr from core.chat import ( ChatState, build_api_kwargs, finalize_response, flush_tool_results, init_state, record_tool_result, reset_state_in_place, stream_response, ) from display import ( format_tool_call_prompt, format_tool_calls_for_display, preprocess_latex, ) from i18n import t logger = logging.getLogger(__name__) # Sentinel values driven into the dedicated #hy-busy-marker gr.HTML # component on the server side, observed by the browser to drive the Send # button's enabled state. HY_BUSY_HTML = '' HY_IDLE_HTML = '' # Initial value for the chat container (the inner
that the renderer # owns). Wrapped in another
so Gradio's gr.HTML wrapper has a stable # child to host the chat surface. HY_CHAT_INITIAL_HTML = '
' # Thread-safe monotonically-increasing sequence numbers for the delta # channel. Each delta payload carries one; the client uses it to dedup # (Gradio occasionally re-fires the same value on reconnect, and we never # want to re-apply the same ops twice). _seq_counter = itertools.count(1) def _next_seq() -> int: return next(_seq_counter) # Bubble IDs are pre-allocated per assistant turn so all ops for the same # bubble (assistant_begin → reasoning_delta → content_delta → assistant_end → # tool_call) can be routed to the right DOM node on the client. _bubble_counter = itertools.count(1) def _next_bubble_id() -> str: return f"a-{next(_bubble_counter)}" def _delta(ops: list[dict], epoch: int | None = None) -> str: """Encode a list of ops as a delta payload string. The payload is shipped through ``gr.HTML``'s value, which the front end renders into the DOM via ``innerHTML``. Plain ``json.dumps`` output is valid HTML text 99 % of the time, but any user- or model-supplied ``<`` would be parsed as the opening of an HTML tag and silently truncate the payload (also a stored-XSS sink). We pre-escape the three HTML-meta characters into their JSON ``\\uNNNN`` form — the browser's ``JSON.parse`` reverses the escapes so semantics are preserved end-to-end. ``epoch`` tags the payload with the session's chat-epoch id so the client can drop stale deltas that arrive after a "+" / new-chat reset. The reset payload itself carries the NEW epoch — the client adopts it, then ignores any later payload that still carries the old one. """ payload: dict = {"seq": _next_seq(), "ops": ops} if epoch is not None: payload["epoch"] = epoch raw = json.dumps(payload, ensure_ascii=False) return ( raw.replace("<", "\\u003c") .replace(">", "\\u003e") .replace("&", "\\u0026") ) # Re-export so ``from chat import init_state`` keeps working in app.py. __all__ = [ "init_state", "send_message", "new_chat", "submit_tool_result", "api_chat", "HY_BUSY_HTML", "HY_IDLE_HTML", "HY_CHAT_INITIAL_HTML", ] def _ensure_state(state) -> ChatState: """Normalize ``state`` into a fresh ``ChatState`` dict. ``gr.State(init_state)`` treats ``init_state`` as a per-session factory in the browser, but stateless ``gradio_client`` calls hand the factory itself to the handler. Accept callable/None and fall back to a fresh state so the API works without breaking the in-browser per-session semantics. """ if state is None or callable(state): return init_state() return state def _validate_user_message(message: str, functions_json_str: str) -> bool: if not message or not message.strip(): gr.Warning(t("warn.empty_msg")) return False if functions_json_str and functions_json_str.strip(): try: json.loads(functions_json_str) except json.JSONDecodeError as e: gr.Warning(t("warn.invalid_fn_json", err=str(e))) return False return True def _run_streaming_turn( state: ChatState, bubble_id: str, epoch: int, system_prompt: str, think_level: str, temperature: float, temperature_use_default: bool, max_tokens: int, top_p: float, preserved_thinking: bool, preserved_thinking_use_default: bool, functions_json_str: str, ) -> Iterator[tuple[list[dict], bool, str]]: """Stream a single turn. Yields ``(ops, show_tool, tool_md)``. ``ops`` is the list of bubble-scoped ops (each carrying ``id=bubble_id``) to ship on this frame. ``show_tool`` / ``tool_md`` carry the tool-area UI state for the caller to translate into ``gr.update(...)`` outputs. ``epoch`` is the chat-epoch captured by the caller at turn start. We compare it against the live ``state["epoch"]`` between yields; if they diverge the user clicked "+", so we stop yielding and let the caller exit without finalising the assistant turn. """ # Collapse the (slider value, "Use model default" checkbox) pair into a # single tristate that ``build_api_kwargs`` understands: ``None`` means # "omit the field, let the server apply its own default"; any float # (including 0, i.e. greedy decoding) is sent literally. resolved_temperature: float | None = ( None if temperature_use_default else float(temperature) ) # Same tristate collapse for ``preserved_thinking``: ``None`` omits the # field (server default), otherwise the toggle's boolean is sent literally. resolved_preserved_thinking: bool | None = ( None if preserved_thinking_use_default else bool(preserved_thinking) ) kwargs = build_api_kwargs( state, system_prompt, functions_json_str, think_level, resolved_temperature, max_tokens, top_p, resolved_preserved_thinking, ) if logger.isEnabledFor(logging.DEBUG): # Dump the actual request body (minus the bulky messages list) so we # can verify which sampling knobs reached the wire. loggable = {k: v for k, v in kwargs.items() if k != "messages"} logger.debug("streaming turn kwargs: %r", loggable) def _cancelled() -> bool: return state.get("epoch") != epoch last_ac, last_rc, last_tca = "", "", [] seen_first_content = False for ops, ac, rc, tca, _rid in stream_response(kwargs, is_cancelled=_cancelled): if _cancelled(): return last_ac, last_rc, last_tca = ac, rc, tca if not ops: # Heartbeat — no new content. We still surface an empty-ops # frame so the wire ticks (Gradio dedups equal values; the # heartbeat carries the new seq via the delta payload below). yield [], False, "" continue # Apply per-op LaTeX preprocessing on content_delta. The model may # emit bare \begin{equation}...\end{equation} blocks etc.; KaTeX # only renders math inside $$ delimiters. out_ops: list[dict] = [] for op in ops: if op["type"] == "content_delta": # First content delta implicitly closes the thinking block on # the client. We tag the FIRST content op with a flag so the # client can switch the summary text and auto-collapse. delta_text = preprocess_latex(op["delta"]) wrapped = { "type": "content_delta", "id": bubble_id, "delta": delta_text, } if not seen_first_content: wrapped["thinking_done"] = True seen_first_content = True out_ops.append(wrapped) elif op["type"] == "reasoning_delta": out_ops.append({ "type": "reasoning_delta", "id": bubble_id, "delta": op["delta"], }) elif op["type"] == "tool_calls": # Carry the snapshot through; we materialise it as a UI # block only at end-of-turn (see the finalize section # below). Skipping here keeps mid-stream wire payloads # tiny — the same partial tool-call gets re-shipped on # every frame otherwise. continue if out_ops: yield out_ops, False, "" # ── Terminal phase ───────────────────────────────────────────────── if _cancelled(): return has_tools, pending = finalize_response( state, last_ac, last_rc, last_tca, ) end_ops: list[dict] = [{"type": "assistant_end", "id": bubble_id}] if has_tools: # Render the tool-call display block as Markdown; the client will # run marked over it. Append AFTER the assistant_end op so the DOM # ordering stays correct (content → tool calls). tc_markdown = format_tool_calls_for_display(pending) end_ops.append({ "type": "tool_call", "id": bubble_id, "markdown": tc_markdown, }) tool_md = format_tool_call_prompt(pending[0], 1, len(pending)) yield end_ops, True, tool_md else: yield end_ops, False, "" # ─── send_message ──────────────────────────────────────────────────────────── # Output slot order (must match app.py's ``send_outputs``): # 0 chat_delta (gr.HTML, elem_id="hy-chat-delta") # 1 state # 2 msg_input # 3 tool_area # 4 tool_call_info # 5 tool_result_input # 6 busy_marker (gr.HTML, elem_id="hy-busy-marker") def send_message( message, state, system_prompt, think_level, temperature, temperature_use_default, max_tokens, top_p, preserved_thinking, preserved_thinking_use_default, functions_json_str, ): state = _ensure_state(state) if not _validate_user_message(message, functions_json_str): # Validation failure — never started streaming, marker stays idle, # no chat ops emitted. yield ( gr.update(), state, gr.update(), gr.update(), gr.update(), gr.update(), HY_IDLE_HTML, ) return message = message.strip() state["messages"].append({"role": "user", "content": message}) bubble_id = _next_bubble_id() # Capture the chat epoch at the start of the turn. ``new_chat`` bumps # ``state["epoch"]`` in place; once that happens we stop yielding so # the new fresh chat surface stays clean. epoch = state["epoch"] def _cancelled() -> bool: return state.get("epoch") != epoch # First frame: append the user bubble, open a new assistant bubble # (with typing indicator), set busy marker, clear input. yield ( _delta([ {"type": "user", "text": message}, {"type": "assistant_begin", "id": bubble_id}, ], epoch=epoch), state, "", gr.update(visible=False), gr.update(), gr.update(), HY_BUSY_HTML, ) try: for ops, show_tool, tool_md in _run_streaming_turn( state, bubble_id, epoch, system_prompt, think_level, temperature, temperature_use_default, max_tokens, top_p, preserved_thinking, preserved_thinking_use_default, functions_json_str, ): if _cancelled(): return # Heartbeat / no-op frame: still ship an empty-ops payload so # the wire ticks and the proxy doesn't drop the connection. payload = _delta(ops, epoch=epoch) if ops else _delta([], epoch=epoch) if show_tool: yield ( payload, state, gr.update(), gr.update(visible=True), gr.update(value=tool_md), gr.update(value=""), gr.update(), ) else: yield ( payload, state, gr.update(), gr.update(visible=False), gr.update(), gr.update(), gr.update(), ) except Exception as e: if _cancelled(): # Reset already cleaned the UI; swallow late-arriving errors # from the abandoned upstream so we don't bounce a stale # warning at the user. return logger.exception("send_message failed: %s", e) if state["messages"] and state["messages"][-1].get("role") == "user": state["messages"].pop() gr.Warning(t("warn.request_failed")) # Drop the typing-indicator placeholder bubble client-side, restore # the user's message into the input box, release the busy marker. yield ( _delta([{"type": "remove_bubble", "id": bubble_id}], epoch=epoch), state, gr.update(value=message), gr.update(visible=False), gr.update(), gr.update(), HY_IDLE_HTML, ) return if _cancelled(): return # ── Terminal IDLE-marker frame ────────────────────────────────────── # Only changes the busy marker; every other slot is gr.update(). Tens # of bytes on the wire — slips past any WebSocket back-pressure so the # browser MutationObserver fires the moment the stream genuinely ends, # releasing the Send button. yield ( gr.update(), state, gr.update(), gr.update(), gr.update(), gr.update(), HY_IDLE_HTML, ) # ─── Programmatic API endpoint ─────────────────────────────────────────────── # The UI handlers above (``send_message`` / ``submit_tool_result``) ship chat # content as DOM-mutation deltas through a hidden ``chat_delta`` HTML # component, which is unusable from ``gradio_client``: a remote caller would # only ever see the final ``gr.update()`` placeholders and the busy marker. # # ``api_chat`` is the headless counterpart: a stateless, generator function # whose YIELDED tuple is the entire reply so far. ``gradio_client.predict`` # returns the last yield, so callers get the full ``(content, reasoning, # tool_calls, history)`` snapshot once the stream finishes; users that want # token-by-token streaming can iterate via ``client.submit(...)``. def api_chat( message: str, system_prompt: str = "", history: list | None = None, think_level: str = "high", temperature: float | None = None, max_tokens: int = 0, top_p: float = 0, preserved_thinking: bool | None = None, functions_json_str: str = "", ) -> Iterator[tuple[str, str, list, list]]: """Stateless chat endpoint for ``gradio_client`` callers. Args: message: The new user turn. system_prompt: Optional system prompt prepended to every request. history: Prior conversation as a list of OpenAI-style messages (``[{"role": "user"|"assistant"|"tool", "content": ...}, ...]``). Pass ``None`` or ``[]`` to start a fresh conversation. The caller owns the history; pass the returned ``updated_history`` back on the next call to continue multi-turn. think_level: One of ``"no_think" | "low" | "high"``. temperature: Sampling temperature. Pass ``None`` (the default) to omit the field from the API request and let the server apply its own default; pass any float (including ``0`` for greedy decoding) to send it literally. max_tokens, top_p: Sampling knobs. Pass ``0`` (the default) to omit the field from the API request and let the server apply its own default. preserved_thinking: Whether to preserve the model's chain-of-thought. Pass ``None`` (the default) to omit the field and let the server apply its own default; pass ``True`` / ``False`` to send it literally (via ``extra_body``). functions_json_str: JSON string of tool definitions (OpenAI tools schema). Empty string disables function calling. Yields: ``(content, reasoning_content, tool_calls, updated_history)`` cumulative on every yield. The final yield is the complete reply. * ``content``: assistant visible text. * ``reasoning_content``: chain-of-thought (when ``think_level`` requests it). May be empty for ``no_think``. * ``tool_calls``: list of OpenAI-style tool-call dicts the model requested. Empty when no function was called. The caller is responsible for executing each call and appending the corresponding ``{"role": "tool", "tool_call_id": ..., "content": ...}`` messages to ``history`` on the next ``api_chat`` call. * ``updated_history``: full message list including the new user turn and the assistant reply (with ``tool_calls`` attached when present). Suitable for round-tripping into the next call. Raises: ValueError: when ``message`` is empty or ``functions_json_str`` is not valid JSON. """ if not message or not message.strip(): raise ValueError("message must be a non-empty string") if functions_json_str and functions_json_str.strip(): try: json.loads(functions_json_str) except json.JSONDecodeError as e: raise ValueError(f"functions_json_str is not valid JSON: {e}") from e state = init_state() if history: # Defensive copy — never mutate the caller's list. state["messages"] = [dict(m) for m in history] state["messages"].append({"role": "user", "content": message.strip()}) kwargs = build_api_kwargs( state, system_prompt, functions_json_str, think_level, temperature, max_tokens, top_p, preserved_thinking, ) last_ac, last_rc = "", "" last_tca: list[dict] = [] yielded = False for _ops, ac, rc, tca, _rid in stream_response(kwargs): last_ac, last_rc, last_tca = ac, rc, tca # Snapshot the in-flight assistant turn into a transient history # view so streaming consumers see the assistant text growing. in_flight = list(state["messages"]) in_flight.append({ "role": "assistant", "content": ac or "", **({"reasoning_content": rc} if rc else {}), **({"tool_calls": list(tca)} if tca else {}), }) yield ac or "", rc or "", list(tca), in_flight yielded = True finalize_response(state, last_ac, last_rc, last_tca) final_history = list(state["messages"]) # Always emit a terminal frame so callers consuming only the last yield # observe the persisted history (the in-flight snapshots above attach a # provisional assistant message; this one is the canonical record). if not yielded: yield "", "", [], final_history else: yield last_ac or "", last_rc or "", list(last_tca), final_history def new_chat(state): """Reset the conversation. Critical: when there's an in-flight stream we MUTATE the existing state dict in place (bumping its epoch) instead of returning a brand new dict. The streaming generator holds a reference to this same dict and polls ``state["epoch"]`` between yields — the in-place bump is what tells it to abandon the rest of its turn. Returning a fresh dict would leave the generator pointed at a stale dict it would happily keep streaming into. We also push ``HY_IDLE_HTML`` to the busy marker so the Send button unlocks immediately. Without this the cancelled generator never yields its terminal IDLE frame, so the button would stay stuck until the soft watchdog in static/app.js force-clears it (~4s). """ if not state or not isinstance(state, dict) or not state.get("messages"): gr.Info(t("info.new_chat")) return gr.update(), state, gr.update() new_epoch = reset_state_in_place(state) return ( _delta([{"type": "reset"}], epoch=new_epoch), state, HY_IDLE_HTML, ) # ─── submit_tool_result ────────────────────────────────────────────────────── # Output slot order (must match app.py's ``tool_submit_outputs``): # 0 chat_delta # 1 state # 2 tool_area # 3 tool_call_info # 4 tool_result_input # 5 busy_marker def submit_tool_result( result_text, state, system_prompt, think_level, temperature, temperature_use_default, max_tokens, top_p, preserved_thinking, preserved_thinking_use_default, functions_json_str, ): state = _ensure_state(state) pending = state.get("pending_tool_calls", []) if not pending: yield ( gr.update(), state, gr.update(visible=False), gr.update(), gr.update(), HY_IDLE_HTML, ) return record_tool_result(state, pending[0], result_text) state["pending_tool_calls"] = pending[1:] if state["pending_tool_calls"]: # More tool results still pending — keep the tool area visible, # marker stays idle (we're not streaming yet). remaining = state["pending_tool_calls"] submitted = state.get("submitted_tool_results", []) total = len(remaining) + len(submitted) current_idx = len(submitted) + 1 tool_info_md = format_tool_call_prompt(remaining[0], current_idx, total) yield ( gr.update(), state, gr.update(visible=True), gr.update(value=tool_info_md), gr.update(value=""), HY_IDLE_HTML, ) return flush_tool_results(state) bubble_id = _next_bubble_id() epoch = state["epoch"] def _cancelled() -> bool: return state.get("epoch") != epoch # First streaming frame: open a new assistant bubble for the tool # follow-up response, hide the tool area, set the busy marker. yield ( _delta([{"type": "assistant_begin", "id": bubble_id}], epoch=epoch), state, gr.update(visible=False), gr.update(), gr.update(), HY_BUSY_HTML, ) try: for ops, show_tool, tool_md in _run_streaming_turn( state, bubble_id, epoch, system_prompt, think_level, temperature, temperature_use_default, max_tokens, top_p, preserved_thinking, preserved_thinking_use_default, functions_json_str, ): if _cancelled(): return payload = _delta(ops, epoch=epoch) if ops else _delta([], epoch=epoch) if show_tool: yield ( payload, state, gr.update(visible=True), gr.update(value=tool_md), gr.update(value=""), gr.update(), ) else: yield ( payload, state, gr.update(visible=False), gr.update(), gr.update(), gr.update(), ) except Exception as e: if _cancelled(): return logger.exception("submit_tool_result failed: %s", e) gr.Warning(t("warn.request_failed")) yield ( _delta([{"type": "remove_bubble", "id": bubble_id}], epoch=epoch), state, gr.update(visible=False), gr.update(), gr.update(), HY_IDLE_HTML, ) return if _cancelled(): return yield ( gr.update(), state, gr.update(), gr.update(), gr.update(), HY_IDLE_HTML, )