"""Session engine for the MOSS-VL-Realtime Space.

One realtime session == one GPU call. The Gradio side (app.py) starts a session
bound to the staged media; prompts typed while the session is live reach the
GPU worker through an on-disk mailbox (ZeroGPU runs @spaces.GPU functions in a
forked worker on the same container, so /tmp is shared between the main
process and the worker).

MOCK mode (MOSS_DEMO_MOCK=1): no torch / spaces / torchcodec imports; a
scripted session drives the exact same event protocol so the full UI can be
exercised on a CPU-only box.
"""

import json
import os
import re
import shutil
import time
import traceback
from collections import deque

MOCK = os.getenv("MOSS_DEMO_MOCK") == "1"

MODEL_ID = os.getenv("MOSS_MODEL_ID", "OpenMOSS-Team/MOSS-VL-Realtime")  # hub id or local path
MAILBOX_ROOT = "/tmp/moss_sessions"

CONTROL_ROUND_START = "<|round_start|>"
CONTROL_ROUND_END = "<|round_end|>"
CONTROL_RESPONSE = "<|response|>"  # real model's round-start marker
CONTROL_SILENCE = "<|silence|>"

# Session budgets (seconds). The paced stream is capped so a session always
# closes gracefully before the ZeroGPU duration kill.
SESSION_VIDEO_CAP_S = 120.0
LIVE_CAP_S = float(os.getenv("MOSS_LIVE_CAP_S", "180"))  # live-camera session length
POSTROLL_IDLE_S = 45.0
HARD_MARGIN_S = 15.0
CLOSE_GRACE_S = 8.0

IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".webp", ".bmp", ".gif", ".tif", ".tiff"}
VIDEO_EXTS = {".mp4", ".mov", ".webm", ".avi", ".mkv", ".ogg", ".m4v"}

FRAME_MAX_SIDE = 1280  # downscale before pickling frames into the GPU worker


if not MOCK:
    import ctypes
    import site

    # nvidia-npp-cu12 installs libnppicc.so.12 inside site-packages/nvidia/npp/lib/,
    # which is not on LD_LIBRARY_PATH. Load it globally before torchcodec is imported
    # so the dynamic linker can resolve it when torchcodec dlopen's its shared libs.
    def _preload_npp():
        for _sp in site.getsitepackages():
            _p = os.path.join(_sp, "nvidia", "npp", "lib", "libnppicc.so.12")
            if os.path.exists(_p):
                ctypes.CDLL(_p, mode=ctypes.RTLD_GLOBAL)
                return

    _preload_npp()

    try:
        import spaces  # MUST come before torch / any CUDA-touching import (ZeroGPU)
    except ImportError:
        spaces = None  # bare GPU box: decorator no-ops, model runs on real CUDA
    import torch
    from transformers import AutoModelForCausalLM, AutoProcessor

    print("Loading processor...")
    processor = AutoProcessor.from_pretrained(
        MODEL_ID, trust_remote_code=True, frame_extract_num_threads=1
    )

    print("Loading model...")
    model = AutoModelForCausalLM.from_pretrained(
        MODEL_ID,
        trust_remote_code=True,
        torch_dtype=torch.bfloat16,
        attn_implementation="sdpa",
    ).to("cuda")
    model.eval()
    print("Model ready.")

    if spaces is not None:
        GPU = spaces.GPU
    else:

        def GPU(*d_args, **d_kwargs):
            if d_args and callable(d_args[0]):
                return d_args[0]

            def _wrap(fn):
                return fn

            return _wrap
else:

    def GPU(*d_args, **d_kwargs):
        """Effect-free stand-in for spaces.GPU in MOCK mode."""
        if d_args and callable(d_args[0]):
            return d_args[0]

        def _wrap(fn):
            return fn

        return _wrap

    processor = None
    model = None  # replaced by _MockModel via get_model()


# --- Media normalization ---


def classify_media(path):
    ext = os.path.splitext(path)[1].lower()
    if ext in IMAGE_EXTS:
        return "image"
    if ext in VIDEO_EXTS:
        return "video"
    return "video"  # gr.Video/gr.Image constrain uploads; default to video


def snapshot_stage(stage_video, stage_image):
    """Resolve the staged media at session start.

    Returns (kind, path, warning) where warning is a user-facing note or None.
    Video wins when both stages are populated (surfaced, not silent).
    """
    if stage_video and stage_image:
        return (
            "video",
            stage_video,
            "Both video and image are staged — the session runs on the video. "
            "两者都已上传，本次会话使用视频。",
        )
    if stage_video:
        return "video", stage_video, None
    if stage_image:
        return "image", stage_image, None
    return None, None, None


def _downscale(img, max_side=FRAME_MAX_SIDE):
    w, h = img.size
    scale = max(w, h) / float(max_side)
    if scale > 1.0:
        img = img.resize((int(w / scale), int(h / scale)))
    return img


def extract_frames(video_path, video_fps, max_frames):
    """Decode a video into [(PIL.Image, timestamp_seconds)] sampled at video_fps.

    Runs on CPU in the main process (outside the GPU lease).
    """
    if MOCK:
        return _mock_frames(video_fps, max_frames)

    from torchcodec.decoders import VideoDecoder
    from torchvision.transforms.functional import to_pil_image

    decoder = VideoDecoder(video_path)
    duration = float(decoder.metadata.duration_seconds or 0.0)
    if duration <= 0:
        frame = decoder[0]
        return [(_downscale(to_pil_image(frame)), 0.0)]

    step = 1.0 / float(video_fps) if float(video_fps) > 0 else 1.0
    timestamps = []
    t = 0.0
    # keep a small epsilon away from the very end (no frame plays exactly at duration)
    end = max(duration - 1e-3, 0.0)
    while t <= end and len(timestamps) < int(max_frames):
        timestamps.append(round(t, 3))
        t += step
    if not timestamps:
        timestamps = [0.0]

    batch = decoder.get_frames_played_at(seconds=timestamps)
    frames = []
    for i in range(batch.data.shape[0]):
        img = _downscale(to_pil_image(batch.data[i]))
        ts = float(batch.pts_seconds[i])
        frames.append((img, ts))
    return frames


def load_image_frame(image_path):
    from PIL import Image

    img = Image.open(image_path).convert("RGB")
    return [(_downscale(img), 0.0)]


def _mock_frames(video_fps, max_frames):
    from PIL import Image

    step = 1.0 / float(video_fps) if float(video_fps) > 0 else 1.0
    n = min(int(max_frames), 12)
    frames = []
    for i in range(n):
        shade = 40 + (i * 160) // max(n - 1, 1)
        frames.append((Image.new("RGB", (64, 36), (shade, shade, 96)), round(i * step, 3)))
    return frames


# --- Mailbox: main process -> GPU worker channel ---


class Mailbox:
    """Per-session directory under /tmp shared with the forked GPU worker.

    prompts.jsonl : appended by the UI process, tailed by the worker
    stop          : flag file — graceful session shutdown
    """

    @staticmethod
    def _dir(sid):
        return os.path.join(MAILBOX_ROOT, sid)

    @staticmethod
    def create(sid):
        os.makedirs(Mailbox._dir(sid), exist_ok=True)

    @staticmethod
    def is_live(sid):
        return bool(sid) and os.path.isdir(Mailbox._dir(sid))

    @staticmethod
    def write_prompt(sid, text):
        path = os.path.join(Mailbox._dir(sid), "prompts.jsonl")
        with open(path, "a", encoding="utf-8") as f:
            f.write(json.dumps({"text": text, "wall_ts": time.time()}) + "\n")

    @staticmethod
    def read_new_prompts(sid, offset):
        """Return (prompts, new_offset) for lines appended past byte offset."""
        path = os.path.join(Mailbox._dir(sid), "prompts.jsonl")
        if not os.path.exists(path):
            return [], offset
        prompts = []
        with open(path, "r", encoding="utf-8") as f:
            f.seek(offset)
            for line in f:
                if not line.endswith("\n"):
                    break  # partial write; re-read next tick
                offset += len(line.encode("utf-8"))
                try:
                    prompts.append(json.loads(line)["text"])
                except (ValueError, KeyError):
                    continue
        return prompts, offset

    @staticmethod
    def mark_live_camera(sid):
        open(os.path.join(Mailbox._dir(sid), "live_camera"), "w").close()

    @staticmethod
    def is_live_camera(sid):
        return bool(sid) and os.path.exists(os.path.join(Mailbox._dir(sid), "live_camera"))

    @staticmethod
    def write_frame(sid, pil_image):
        """Store a live-camera frame for the GPU worker (name = capture time in ns)."""
        d = os.path.join(Mailbox._dir(sid), "frames")
        os.makedirs(d, exist_ok=True)
        name = f"{time.time_ns():020d}.jpg"
        tmp = os.path.join(d, "." + name)
        pil_image.save(tmp, "JPEG", quality=85)
        os.replace(tmp, os.path.join(d, name))

    @staticmethod
    def read_new_frames(sid, after_name):
        """Return ([(path, name)], last_name) for frames newer than after_name."""
        d = os.path.join(Mailbox._dir(sid), "frames")
        if not os.path.isdir(d):
            return [], after_name
        names = sorted(n for n in os.listdir(d) if not n.startswith(".") and n > (after_name or ""))
        return [(os.path.join(d, n), n) for n in names], (names[-1] if names else after_name)

    @staticmethod
    def signal_stop(sid):
        if Mailbox.is_live(sid):
            open(os.path.join(Mailbox._dir(sid), "stop"), "w").close()

    @staticmethod
    def should_stop(sid):
        return os.path.exists(os.path.join(Mailbox._dir(sid), "stop"))

    @staticmethod
    def cleanup(sid):
        shutil.rmtree(Mailbox._dir(sid), ignore_errors=True)

    @staticmethod
    def cleanup_stale(max_age_s=3600):
        if not os.path.isdir(MAILBOX_ROOT):
            return
        now = time.time()
        for name in os.listdir(MAILBOX_ROOT):
            path = os.path.join(MAILBOX_ROOT, name)
            try:
                if now - os.path.getmtime(path) > max_age_s:
                    shutil.rmtree(path, ignore_errors=True)
            except OSError:
                continue


# --- Round parsing (CPU side) ---

_CTRL_RE = re.compile(r"(<\|[a-zA-Z_]+\|>)")


class RoundParser:
    """Turn raw session chunks into UI ops.

    Ops: ("round_open", ts) | ("text", delta) | ("round_break", ts)
        | ("round_close", ts) | ("silence", ts) | ("control", token) for any
        other unknown <|...|> control token, which must stay out of the chat
        text but is worth logging in the raw view.
    Rounds open on <|round_start|> or the real model's <|response|>; they close
    on <|round_end|> or when silence resumes (the real model has no end marker).
    The real model RE-EMITS <|response|> every frame while narrating one
    continuous utterance — that yields ("round_break", ts): a raw-view round
    boundary that must NOT break the flowing chat text.
    Control tokens normally arrive as standalone chunks; the regex split is a
    defensive path for tokens embedded inside a larger chunk.
    """

    def __init__(self):
        self.in_round = False
        self._round_has_text = False

    def _close(self, ops, ts):
        if self.in_round:
            self.in_round = False
            self._round_has_text = False
            ops.append(("round_close", ts))

    def feed(self, chunk, ts):
        ops = []
        pieces = [chunk] if _CTRL_RE.fullmatch(chunk) else [p for p in _CTRL_RE.split(chunk) if p]
        for piece in pieces:
            if piece in (CONTROL_ROUND_START, CONTROL_RESPONSE):
                if self.in_round and self._round_has_text:
                    # per-frame re-emitted marker mid-narration: raw-view
                    # boundary only — the utterance keeps flowing in chat
                    self._round_has_text = False
                    ops.append(("round_break", ts))
                elif not self.in_round:
                    self.in_round = True
                    self._round_has_text = False
                    ops.append(("round_open", ts))
                # else: duplicate marker in a still-empty round — ignore
            elif piece == CONTROL_ROUND_END:
                self._close(ops, ts)
            elif piece == CONTROL_SILENCE:
                # the real model has no explicit round end — silence resuming
                # after a response marks the round as finished
                self._close(ops, ts)
                ops.append(("silence", ts))
            elif _CTRL_RE.fullmatch(piece):
                ops.append(("control", piece))
            else:
                if not self.in_round:
                    # text without an explicit round marker — open one implicitly
                    self.in_round = True
                    ops.append(("round_open", ts))
                self._round_has_text = True
                ops.append(("text", piece))
        return ops


# --- The session generator (runs in the GPU worker) ---


def _speed_factor(playback_speed):
    return {"1×": 1.0, "2×": 2.0, "Fast-forward": 0.0}.get(playback_speed, 1.0)


def estimate_duration(sid, frames, initial_prompt, gen_kwargs, playback_speed, postroll_idle_s=POSTROLL_IDLE_S, live=False):
    """Dynamic @spaces.GPU duration: paced stream span + post-roll + margin."""
    if live:
        return int(LIVE_CAP_S + HARD_MARGIN_S)
    speed = _speed_factor(playback_speed)
    span = frames[-1][1] if frames else 0.0
    if speed > 0:
        paced = min(span / speed, SESSION_VIDEO_CAP_S)
    else:
        paced = min(len(frames) * 0.35 + 10.0, 90.0)
    return int(paced + postroll_idle_s + HARD_MARGIN_S)


def _drain(session):
    chunks = []
    while True:
        chunk = session.poll_output(timeout=0.0)
        if chunk is None:
            break
        chunks.append(chunk)
    return chunks


def _poll_mailbox(session, sid, offset, video_ts, events):
    """Push any newly mailed prompts into the session; emit ack events."""
    if sid is None:
        return offset
    prompts, offset = Mailbox.read_new_prompts(sid, offset)
    for text in prompts:
        session.push_prompt(text)
        events.append({"type": "prompt", "text": text, "video_ts": video_ts})
    return offset


@GPU(duration=estimate_duration)
def gpu_session(sid, frames, initial_prompt, gen_kwargs, playback_speed, postroll_idle_s=POSTROLL_IDLE_S, live=False):
    """Run one realtime session; yields typed event dicts.

    Uploaded media: frames are paced against wall clock (speed factor from
    playback_speed; fast-forward pushes as fast as the model consumes), then a
    post-roll keeps the session open for Q&A. Live camera (live=True): frames
    arrive through the mailbox from the browser's webcam stream and are pushed
    with capture-time timestamps until stop flag / budget.
    """
    speed = _speed_factor(playback_speed)
    budget = estimate_duration(sid, frames, initial_prompt, gen_kwargs, playback_speed, postroll_idle_s, live)
    total = len(frames) if frames else 0
    mail_offset = 0
    dropped_frames = 0

    session = get_model().create_realtime_session(
        get_processor(), initial_prompt="", **gen_kwargs
    )
    try:
        # The model runs a single realtime loop at a time; a just-closed session
        # can take a few seconds to release it. Wait it out instead of failing.
        for _ in range(20):
            try:
                session.start()
                break
            except RuntimeError as exc:
                if "active realtime generation" not in str(exc):
                    raise
                time.sleep(0.5)
        else:
            yield {
                "type": "error",
                "message": "The model is busy with another session — try again in a moment. 模型正忙，请稍后重试。",
            }
            return

        t0 = time.monotonic()
        deadline = t0 + budget - CLOSE_GRACE_S
        yield {"type": "session_start", "frames_total": total, "budget_s": budget, "live": live}

        if initial_prompt:
            session.push_prompt(initial_prompt)
            yield {"type": "prompt", "text": initial_prompt, "video_ts": 0.0}

        if live:
            yield from _live_loop(session, sid, deadline)
            return

        end_reason = "stream ended"
        last_ts = 0.0
        for i, (img, ts) in enumerate(frames):
            if Mailbox.should_stop(sid) if sid else False:
                end_reason = "stopped"
                break
            if time.monotonic() > deadline:
                end_reason = "session budget reached"
                break

            # pace: wait until this frame's wall-clock slot, staying responsive
            target = t0 + (ts / speed) if speed > 0 else 0.0
            while time.monotonic() < target:
                events = []
                mail_offset = _poll_mailbox(session, sid, mail_offset, last_ts, events)
                chunks = _drain(session)
                if chunks:
                    events.append(
                        {"type": "chunk_batch", "video_ts": last_ts, "chunks": chunks}
                    )
                for ev in events:
                    yield ev
                if (sid and Mailbox.should_stop(sid)) or time.monotonic() > deadline:
                    break
                time.sleep(0.05)
            if sid and Mailbox.should_stop(sid):
                end_reason = "stopped"
                break

            dropped_frames += 1 if session.push_frame(img, timestamp=ts) else 0
            last_ts = ts
            events = []
            mail_offset = _poll_mailbox(session, sid, mail_offset, ts, events)
            chunks = _drain(session)
            for ev in events:
                yield ev
            yield {
                "type": "frame",
                "frame": i + 1,
                "total": total,
                "video_ts": ts,
                "chunks": chunks,
                "dropped_frames": dropped_frames,
            }
        else:
            end_reason = "stream ended"

        # Post-roll Q&A: the session stays open on the observed stream.
        if end_reason == "stream ended":
            yield {"type": "postroll", "video_ts": last_ts}
            idle_deadline = time.monotonic() + postroll_idle_s
            last_emit = time.monotonic()
            while time.monotonic() < min(idle_deadline, deadline):
                if sid and Mailbox.should_stop(sid):
                    end_reason = "stopped"
                    break
                events = []
                prev_offset = mail_offset
                mail_offset = _poll_mailbox(session, sid, mail_offset, last_ts, events)
                if mail_offset != prev_offset:
                    idle_deadline = time.monotonic() + postroll_idle_s
                    # A prompt spliced without a frame parks the model: the
                    # assistant turn opens with <|silence|> (training format)
                    # and the loop waits for new input, so the question is
                    # never answered. Re-anchor postroll questions on the last
                    # frame — prompt+frame in one drain cycle is the path that
                    # actually generates a response (same as `analyze`).
                    if frames:
                        session.push_frame(frames[-1][0], timestamp=last_ts)
                chunk = session.poll_output(timeout=0.2)
                if chunk is not None:
                    events.append(
                        {"type": "chunk_batch", "video_ts": last_ts, "chunks": [chunk]}
                    )
                    idle_deadline = time.monotonic() + postroll_idle_s
                if not events and time.monotonic() - last_emit > 0.3:
                    events.append({"type": "tick", "video_ts": last_ts})
                if events:
                    last_emit = time.monotonic()
                for ev in events:
                    yield ev
            else:
                if time.monotonic() >= deadline:
                    end_reason = "session budget reached"
                elif end_reason == "stream ended":
                    end_reason = "idle timeout"

        yield {"type": "session_end", "reason": end_reason, "video_ts": last_ts}
    except Exception as exc:
        traceback.print_exc()
        yield {"type": "error", "message": f"{type(exc).__name__}: {exc}"}
    finally:
        # close() can raise (join timeout / late worker error) — never let that
        # skip the mailbox cleanup, or the sid stays "live" and blocks every
        # new session until the stale sweep an hour later.
        try:
            session.close()
        except Exception:
            traceback.print_exc()
        if sid:
            Mailbox.cleanup(sid)


def _live_loop(session, sid, deadline):
    """Consume live-camera frames from the mailbox until stop / budget."""
    from PIL import Image

    mail_offset = 0
    last_frame_name = None
    first_frame_ns = None
    pushed = 0
    last_ts = 0.0
    end_reason = "stopped"
    t0 = time.monotonic()
    last_emit = t0  # heartbeat so the UI can flush pending updates while idle

    while True:
        if sid and Mailbox.should_stop(sid):
            end_reason = "stopped"
            break
        if time.monotonic() > deadline:
            end_reason = "session budget reached"
            break

        frame_files, last_frame_name = Mailbox.read_new_frames(sid, last_frame_name)
        for path, name in frame_files:
            ns = int(name.split(".")[0])
            if first_frame_ns is None:
                first_frame_ns = ns
            ts = max((ns - first_frame_ns) / 1e9, last_ts)
            try:
                img = Image.open(path).convert("RGB")
            except OSError:
                continue
            finally:
                try:
                    os.remove(path)
                except OSError:
                    pass
            session.push_frame(_downscale(img), timestamp=ts)
            last_ts = ts
            pushed += 1

        events = []
        mail_offset = _poll_mailbox(session, sid, mail_offset, last_ts, events)
        chunks = _drain(session)
        for ev in events:
            yield ev
        if frame_files or chunks:
            yield {
                "type": "frame",
                "frame": pushed,
                "total": 0,  # unbounded live stream
                "video_ts": last_ts,
                "chunks": chunks,
                "dropped_frames": 0,
            }
            last_emit = time.monotonic()
        elif time.monotonic() - last_emit > 0.3:
            yield {"type": "tick", "video_ts": time.monotonic() - t0, "frames": pushed}
            last_emit = time.monotonic()
        time.sleep(0.1)

    # flush any final output briefly before closing
    flush_deadline = time.monotonic() + 2.0
    while time.monotonic() < flush_deadline:
        chunk = session.poll_output(timeout=0.2)
        if chunk is None:
            continue
        yield {"type": "chunk_batch", "video_ts": last_ts, "chunks": [chunk]}
    yield {"type": "session_end", "reason": end_reason, "video_ts": last_ts}


# --- Model access (real or mock) ---


def get_model():
    if MOCK:
        global model
        if model is None:
            model = _MockModel()
        return model
    return model


def get_processor():
    return processor


class _MockSession:
    """Scripted realtime session mirroring the wire protocol.

    Emits silences while 'observing', two scripted rounds during the stream,
    and an echo round for every pushed prompt (proves the mailbox path).
    """

    _ROUND_A = ["The stream opens on ", "a synthetic test pattern ", "fading in."]
    _ROUND_B = ["Brightness keeps increasing — ", "the pattern is nearly white now."]

    def __init__(self):
        self._out = deque()
        self._frames = 0

    def start(self):
        return self

    def _queue_round(self, chunks):
        self._out.append(CONTROL_ROUND_START)
        self._out.append("<|response|>")  # real model emits this inside rounds
        self._out.extend(chunks)
        self._out.append(CONTROL_ROUND_END)

    def push_frame(self, img, timestamp=None, drop_oldest=True):
        self._frames += 1
        if self._frames == 4:
            self._queue_round(self._ROUND_A)
        elif self._frames == 9:
            self._queue_round(self._ROUND_B)
        elif self._frames % 3 == 0:
            self._out.append(CONTROL_SILENCE)
        return False

    def push_prompt(self, prompt):
        self._queue_round(
            ["(mock) You asked: ", f"“{prompt}” — ", f"I have seen {self._frames} frames so far."]
        )

    def poll_output(self, timeout=0.0):
        if self._out:
            return self._out.popleft()
        if timeout > 0:
            time.sleep(min(timeout, 0.05))
            if self._out:
                return self._out.popleft()
        return None

    def close(self, timeout=None):
        pass


class _MockModel:
    def create_realtime_session(self, processor_, initial_prompt="", **kwargs):
        return _MockSession()


# --- Stateless one-shot for MCP ---


def analyze(
    media: str,
    prompt: str,
    max_new_tokens: int = 512,
    temperature: float = 0.0,
    video_fps: float = 1.0,
    max_frames: int = 64,
) -> str:
    """Analyze a video or image with MOSS-VL-Realtime and return the answer text.

    The media is streamed through a realtime session frame by frame (images are
    a single frame) and all model responses are collected and returned.

    Args:
        media: Path or http(s) URL of a video (.mp4/.mov/.webm) or image (.png/.jpg/...).
        prompt: The question or instruction about the media.
        max_new_tokens: Maximum number of tokens to generate per response round.
        temperature: Sampling temperature (0 = deterministic).
        video_fps: Frames per second sampled from a video.
        max_frames: Maximum number of frames sampled from a video.
    """
    if media.startswith(("http://", "https://")):
        import tempfile
        import urllib.request

        suffix = os.path.splitext(media.split("?")[0])[1] or ".mp4"
        with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp:
            with urllib.request.urlopen(media, timeout=60) as resp:
                shutil.copyfileobj(resp, tmp)
            media = tmp.name

    kind = classify_media(media)
    if kind == "image":
        frames = load_image_frame(media)
    else:
        frames = extract_frames(media, video_fps, max_frames)

    gen_kwargs = {
        "max_new_tokens": int(max_new_tokens),
        "temperature": float(temperature),
        "do_sample": float(temperature) > 0.0,
    }

    parser = RoundParser()
    rounds, current = [], []
    # short post-roll: a one-shot call should not idle out the GPU lease
    for event in gpu_session(None, frames, prompt, gen_kwargs, "Fast-forward", postroll_idle_s=8.0):
        if event["type"] == "error":
            raise RuntimeError(event["message"])
        for chunk in event.get("chunks", []):
            for op, payload in parser.feed(chunk, event.get("video_ts", 0.0)):
                if op == "text":
                    current.append(payload)
                elif op == "round_close" and current:
                    rounds.append("".join(current).strip())
                    current = []
    if current:
        rounds.append("".join(current).strip())
    return "\n\n".join(r for r in rounds if r) or "(the model stayed silent)"