reply_formatter.py

#!/usr/bin/env python3
"""
reply_formatter.py -- extract the user-visible answer from a Heartly
Qwen-Code (v3) grammar output.

The v3 model is trained to emit:
    thinking {reasoning}  response<decide>speak|stop</decide><verify>known|unknown</verify> {answer} <stop>

(or `<decide>stop</decide>` for the silence case).

The user should only ever see {answer}, never the scaffolding. In practice the
model -- especially through a GUI like LM Studio -- produces *noisy* variants:
the tags are NOT registered as tokenizer special tokens, so Qwen's BPE shatters
`<decide>` into subwords that can re-decode mangled (e.g. ``<deside>``), the
`<stop>` end-marker can come back truncated (no closing `>`), and verify values
can carry junk (`<verify>known, true</verify>`). Qwen chat-template tokens also
leak as plain text (`<stop> <begin> <sep>`).

This parser is tolerant: it canonicalises the tags, extracts the structured
fields, and has an aggressive fallback so NO grammar token ever reaches the user.

Modes:
    "chat"  -- only the clean answer (default)
    "debug" -- answer + [decide=X verify=Y] metadata per turn
    "raw"   -- the original raw output, untouched

Usage:
    from reply_formatter import format_reply, clean_reply
    shown = format_reply(raw_model_output)
"""
from __future__ import annotations

import re
from dataclasses import dataclass


# ---------------------------------------------------------------------------
# 1. Junk tokens -- decoded-to-plain-text Qwen special tokens that are never
#    answer content. Stripped first so they cannot confuse grammar parsing.
# ---------------------------------------------------------------------------
_JUNK_TOKENS = (
    "<|im_start|>", "<|im_end|>", "im_start", "im_end",
    "<|object_ref_start|>", "<|object_ref_end|>",
    "<|box_start|>", "<|box_end|>",
    "<|quad_start|>", "<|quad_end|>",
    "<|vision_start|>", "<|vision_end|>", "<|vision_pad|>", "<|image_pad|>",
    "<|endoftext|>", "endoftext",
    "<|tool_call_begin|>", "<|tool_call_end|>", "<|tool_call_argument_begin|>",
    "<|tool_call_argument_end|>", "<|tool_call_argument_name|>", "<|tool_call_argument|>",
    "<|tool_calls_section_begin|>", "<|tool_calls_section_end|>",
    "<tool_calls>", "</tool_calls>", "tool_call", "tool_calls",
    "<begin>", "<sep>",
)


# ---------------------------------------------------------------------------
# 2. Meta-commentary patterns -- the model's self-talk about what it knows /
#    doesn't know. If any bleeds into the answer zone, strip it. Sourced from
#    the reasoning / refusal / silence templates in render_code_sft_v3.py.
#    (The think block is removed entirely in the common case, so this is a
#    safety net, not the primary mechanism.)
# ---------------------------------------------------------------------------
_META_PATTERNS = [
    r"\bI (?:do )?know (?:how )?to do this\b[^.]*\.",
    r"\bI (?:do )?know how to write this\b[^.]*\.",
    r"\bI (?:do )?know this function well\b[^.]*\.",
    r"\bI (?:can|will) (?:write|produce|implement|respond|answer|write clean code)\b[^.]*\.",
    r"\bThis is a standard programming task\b[^.]*\.",
    r"\bI recognise this programming problem\b[^.]*\.",
    r"\bI (?:recognise|recognize) this (?:programming|pattern|algorithm)\b[^.]*\.",
    r"\bStandard problem\b[^.]*\.",
    r"\bClear task\b[^.]*\.",
    r"\bI've seen this pattern before\b[^.]*\.",
    r"\bThis is straightforward\b[^.]*\.",
    r"\bI (?:do )?not know this API or library\b[^.]*\.",
    r"\bI should not invent a solution\b[^.]*\.",
    r"\bI have no knowledge of this framework\b[^.]*\.",
    r"\bThe honest response is\b[^.]*\.",
    r"\bI cannot verify the correct implementation\b[^.]*\.",
    r"\bMaking something up would be worse than admitting it\b[^.]*\.",
    r"\bI'll say so rather than produce fake code\b[^.]*\.",
    r"\bThe input is empty or not a real question\b[^.]*\.",
    r"\bNo meaningful request was made\b[^.]*\.",
    r"\bSpeaking would add nothing\b[^.]*\.",
    r"\bSocial turn\b[^.]*\.",
    r"\bGreeting[ —-]\s*respond\b[^.]*\.",
    r"\bNot a factual question\b[^.]*\.",
    r"\bCasual conversation\b[^.]*\.",
    r"\bFollow-up question\b[^.]*\.",
    r"\bThey need motivation\b[^.]*\.",
    r"\bEmotional context\b[^.]*\.",
    r"\bMeta-conversation about how we work together\b[^.]*\.",
    r"\bBe warm\b[^.]*\.",
    r"\bThey (?:want|are|need|'re|'ve|gave) ",
    r"\bThey're (?:just saying|opening up) ",
    r"\bThey gave more detail\b[^.]*\.",
]
_META_RE = re.compile("|".join(_META_PATTERNS), re.IGNORECASE)

# Stray control words left behind after a mangled tag is stripped.
# Any leftover angle-bracket control construct, tolerant of truncation/mangles
# (e.g. `</decide` with no closing `>`, or `<deside speak>`).
_TAG_RE = re.compile(r"</?\s*[a-z_][^>]*>?", re.IGNORECASE)

# Stray control words left behind after a mangled tag is stripped (used inline
# in _clean_text as leading/trailing guards).
_CONTROL_WORDS = r"(?:speak|stop|silent|done|response|known|unknown|precise)"

# Collapse runs of whitespace.
_MULTI_SPACE_RE = re.compile(r"\s{2,}")


# ---------------------------------------------------------------------------
# 3. Normalization -- turn messy real-world tokens into canonical tags so the
#    structured parser can run cleanly.
# ---------------------------------------------------------------------------
def _normalize(raw: str) -> str:
    t = raw or ""

    # 3a. Drop Qwen / decoded-token litter.
    for junk in _JUNK_TOKENS:
        t = t.replace(junk, "")

    # 3b. Decide tags. Canonical is ``<decide>`` / ``</decide>`` but tokenization
    #     can drop a letter (<deside>) or add spaces, and the close can be
    #     truncated (``</decide``). ``de[cs]ide`` catches both spellings.
    t = re.sub(
        r"<\s*/?\s*de[cs]ide\b[^>]*>?>?",
        lambda m: "</decide>" if "/" in m.group(0) else "<decide>",
        t,
        flags=re.IGNORECASE,
    )

    # 3c. Verify tags -- same tolerance.
    t = re.sub(
        r"<\s*/?\s*verify\b[^>]*>?>?",
        lambda m: "</verify>" if "/" in m.group(0) else "<verify>",
        t,
        flags=re.IGNORECASE,
    )

    # 3d. Stop tags -- ``<stop>``, ``<stop (truncated`` (no close), and a bare
    #     ``<stop`` at end-of-output. The optional ``>?>`` handles all of these.
    t = re.sub(r"<\s*stop\b[^>]*>?>?", "<stop>", t, flags=re.IGNORECASE)

    # 3e. Strip a leading noise run (``<stop> <begin> <sep>`` ...) that some
    #     outputs carry before any thinking/decide content. This <stop> is junk,
    #     not the Heartly end-of-turn marker (which is mid/trailing).
    t = re.sub(
        r"^\s*((?:<stop>|<begin>|<sep>)\s*)*", "", t, flags=re.IGNORECASE
    ).lstrip()

    return t


# ---------------------------------------------------------------------------
# 4. Value normalization helpers.
# ---------------------------------------------------------------------------
_DECIDE_VAL_RE = re.compile(r"\b(speak|stop|silent)\b", re.IGNORECASE)


def _decide_value(raw_val: str) -> str:
    """Reduce a decide tag body to canonical ``speak`` / ``stop``."""
    m = _DECIDE_VAL_RE.search(raw_val or "")
    if m:
        v = m.group(1).lower()
        return "stop" if v == "silent" else v
    return ""  # unparseable


_VERIFY_VAL_RE = re.compile(r"\b(known|unknown)\b", re.IGNORECASE)


def _verify_value(raw_val: str) -> str:
    m = _VERIFY_VAL_RE.search(raw_val or "")
    return m.group(1).lower() if m else ""


# ---------------------------------------------------------------------------
# 5. Parsed turn + segment parser.
# ---------------------------------------------------------------------------
@dataclass
class ParsedTurn:
    reasoning: str = ""
    decide: str = ""   # speak | stop | ""
    verify: str = ""   # known | unknown | ""
    answer: str = ""


# ``thinking {reasoning}  response<decide>...``  -- the word "response" is
# OPTIONAL (the model drops it sometimes; seen in the 2026-08-01 log). The
# lookahead ``(?=<decide>|\\Z)`` bounds the reasoning block at the decide tag
# (or end of string) so the (non-greedy) body doesn't greedily swallow the
# answer, and it does NOT consume the decide opener (leaving it for the decide
# parser to read the decide value).
_THINK_RE = re.compile(
    r"\bthinking\b\s*(.*?)(\s*\bresponse\b)?\s*(?=<decide>|\Z)",
    re.DOTALL | re.IGNORECASE,
)
# After normalization, tags are canonical.
_DECIDE_CLOSE_RE = re.compile(r"<decide>(.*?)</decide>", re.DOTALL | re.IGNORECASE)
_VERIFY_CLOSE_RE = re.compile(r"<verify>(.*?)</verify>", re.DOTALL | re.IGNORECASE)
# Tolerant open-tag grabs (for mangled/missing close).
_DECIDE_OPEN_RE = re.compile(r"<decide\b([^<]*)", re.IGNORECASE)
_VERIFY_OPEN_RE = re.compile(r"<verify\b([^<]*)", re.IGNORECASE)


def _parse_segment(segment: str) -> ParsedTurn:
    """Parse one turn (already split on ``<stop>``)."""
    seg = segment.strip()
    if not seg:
        return ParsedTurn()

    # --- thinking block (word "response" optional) ---
    reasoning = ""
    m = _THINK_RE.search(seg)
    if m:
        reasoning = (m.group(1) or "").strip()
        seg = seg[m.end():]

    # --- decide ---
    decide = ""
    m = _DECIDE_CLOSE_RE.search(seg)
    if m:
        decide = _decide_value(m.group(1))
        seg = seg[m.end():]
    else:
        mo = _DECIDE_OPEN_RE.search(seg)
        if mo:
            decide = _decide_value(mo.group(1))
            seg = seg[mo.end():]

    # --- verify ---
    verify = ""
    m = _VERIFY_CLOSE_RE.search(seg)
    if m:
        verify = _verify_value(m.group(1))
        seg = seg[m.end():]
    else:
        mo = _VERIFY_OPEN_RE.search(seg)
        if mo:
            verify = _verify_value(mo.group(1))
            seg = seg[mo.end():]

    # --- answer zone: everything left, up to any stray stop ---
    answer = seg
    sm = re.search(r"<stop", answer, re.IGNORECASE)
    if sm:
        answer = answer[: sm.start()]

    # Default an unparseable decide to "speak" (the model is answering).
    if not decide:
        decide = "speak"

    return ParsedTurn(
        reasoning=reasoning,
        decide=decide,
        verify=verify,
        answer=answer,
    )


def parse_reply(raw: str) -> list[ParsedTurn]:
    """Parse raw model output into one or more turns (multi-turn aware)."""
    text = _normalize(raw)
    turns: list[ParsedTurn] = []
    for segment in re.split(r"<stop>", text):
        seg = segment.strip()
        if not seg:
            continue
        # A real turn must carry a decide marker; stray fence/backtick noise
        # emitted between repeated <stop> markers is skipped.
        if not _DECIDE_OPEN_RE.search(seg):
            continue
        turns.append(_parse_segment(seg))
    if not turns:
        # No structured grammar found at all -- treat the whole output as one
        # best-effort turn.
        turns.append(_parse_segment(text))
    return turns


# ---------------------------------------------------------------------------
# 6. Answer sanitizers.
# ---------------------------------------------------------------------------
def _strip_meta(text: str) -> str:
    """Remove the model's reasoning self-talk from a piece of text."""
    return _META_RE.sub("", text).strip()


def _dedupe(text: str) -> str:
    """Drop verbatim-repeated sentences, keep first occurrence."""
    out, seen = [], set()
    for sent in re.split(r"(?<=[.!?])\s+", text):
        s = sent.strip()
        if not s:
            continue
        key = s.lower()
        if key not in seen:
            seen.add(key)
            out.append(s)
    return " ".join(out)


def _clean_text(text: str) -> str:
    """Final pass on an answer zone: strip residual tags/junk, meta, dedupe.

    Whitespace collapse and sentence de-duplication run on the PROSE regions
    only -- fenced code blocks (``` ``` ```) are preserved verbatim, because
    their indentation is significant and must not be collapsed to a single
    space.
    """
    t = text or ""
    # Qwen litter first (never answer content).
    for junk in _JUNK_TOKENS:
        t = t.replace(junk, "")
    # Fence out code blocks so their whitespace survives intact.  re.split with
    # a capturing group yields [prose, code, prose, code, ...] -- code blocks
    # land at odd indices and are emitted untouched.
    pieces = re.split(r"(```[^\n]*\n.*?```)", t, flags=re.DOTALL)
    out = []
    for i, chunk in enumerate(pieces):
        if not chunk:
            continue
        if i % 2 == 1:  # captured fenced block -> keep verbatim
            # The model emits literal backslash-n / backslash-t as plain text in
            # code (training artifact). Restore real newlines/tabs so multi-line
            # code renders instead of collapsing to a single line.
            chunk = chunk.replace(chr(92) + "n", chr(10)).replace(chr(92) + "t", chr(9))
            out.append(chunk)
            continue
        # prose region
        chunk = _TAG_RE.sub(" ", chunk)
        chunk = re.sub(rf"^\s*{_CONTROL_WORDS}\b\s*", "", chunk, flags=re.IGNORECASE)
        chunk = re.sub(rf"\s*{_CONTROL_WORDS}\s*$", "", chunk, flags=re.IGNORECASE)
        chunk = _strip_meta(chunk)
        chunk = _MULTI_SPACE_RE.sub(" ", chunk).strip()
        chunk = _dedupe(chunk)
        if chunk:
            out.append(chunk)
    return " ".join(out).strip()


def _answer_from_reasoning(reasoning: str) -> str:
    """Last-resort: if the answer zone is empty, try to harvest a real answer
    from the reasoning block (e.g. 'The answer is X' phrasing).  Returns "" when
    the reasoning only holds placeholder / meta chatter -- in that case the
    caller maps the empty answer to the silence sentinel ("...")."""
    if not reasoning:
        return ""
    m = re.search(
        r"(?:the answer is|answer is|it's|it is)\s*:?\s*(.+?)(?:\.\s*$|\.\s*<|$)",
        reasoning,
        re.IGNORECASE | re.DOTALL,
    )
    if m:
        return _clean_text(m.group(1))
    return ""


# ---------------------------------------------------------------------------
# 7. Resolution -- decide what the user actually sees.
# ---------------------------------------------------------------------------
def _resolve_chat(turns: list[ParsedTurn]) -> str:
    parts = []
    for t in turns:
        if t.decide == "stop":
            parts.append("...")
            continue
        ans = _clean_text(t.answer)
        if not ans and t.verify == "unknown":
            parts.append(ans or "I don't have that information.")
            continue
        if not ans:
            ans = _answer_from_reasoning(t.reasoning)
        parts.append(ans if ans else "...")
    return "\n".join(parts)


def _resolve_debug(turns):
    parts = []
    for t in turns:
        ans = _clean_text(t.answer)
        if not ans:
            ans = _answer_from_reasoning(t.reasoning)
        meta = " ".join(
            f"[{k}={v}]" for k, v in (("decide", t.decide), ("verify", t.verify)) if v
        )
        shown = f"{ans}  {meta}".strip() if ans else (meta or "(empty)")
        parts.append(shown)
    return "\n".join(parts)


# 6b. Last-resort legacy cleaner -- used only if structured parsing leaves a
# grammar token in the chat output. Aggressively strips every angle-bracket
# construct + control word, guaranteeing a clean result.
def _legacy_clean(text):
    t = _normalize(text)
    t = re.split(r"<stop", t, maxsplit=1)[0]
    dm = _DECIDE_OPEN_RE.search(t)
    if dm:
        t = t[dm.start():]
    else:
        m = _THINK_RE.search(t)
        if m:
            t = t[m.end():] if m.end() <= len(t) else ""
    t = _DECIDE_CLOSE_RE.sub(" ", t)
    t = _VERIFY_CLOSE_RE.sub(" ", t)
    t = _TAG_RE.sub(" ", t)
    t = re.sub(rf"^\s*{_CONTROL_WORDS}\b\s*", "", t, flags=re.IGNORECASE)
    t = re.sub(rf"\s*{_CONTROL_WORDS}\b\s*$", "", t, flags=re.IGNORECASE)
    return _clean_text(t)


# 8. Public API
def format_reply(raw, mode="chat"):
    """Format a raw Heartly model output for display.

    mode: "chat" (default), "debug", or "raw".
    """
    if mode == "raw":
        return raw if raw else ""
    if mode not in ("chat", "debug"):
        mode = "chat"
    if not raw or not raw.strip():
        return ""

    turns = parse_reply(raw)
    if mode == "debug":
        return _resolve_debug(turns)
    out = _resolve_chat(turns)
    # Leak guarantee: if any grammar token survived, fall back hard.
    if out == "" and raw.strip():
        out = _legacy_clean(raw)
    elif out != "" and _TAG_RE.search(out):
        out = _legacy_clean(raw) or out
    # Safety net: the model intermittently emits code (especially un-fenced) as
    # literal backslash-n / backslash-t text. Render those as real newlines/tabs
    # so answers never collapse to a single line in the browser or CLI.
    out = out.replace(chr(92) + "n", chr(10)).replace(chr(92) + "t", chr(9))
    return out


def clean_reply(text):
    """Legacy-compatible drop-in replacement for the old clean_reply()."""
    return format_reply(text, mode="chat")


if __name__ == "__main__":
    import sys
    sample = sys.stdin.read() if not sys.argv[1:] else open(sys.argv[1]).read()
    print(format_reply(sample, mode="debug"))