"""video_vision — analyze a video by sampling frames + transcribing audio.

Pipeline (v1):
  1. ffprobe the duration; pick a frame interval (2-10s) so we take ≤20 frames.
  2. ffmpeg extracts the frames.
  3. ffmpeg extracts the audio; Soniox transcribes it WITH timestamps.
  4. each frame is described by the agent's vision (image_vision), given the
     transcript around that timestamp as extra context.
  5. all frame descriptions + the transcript are synthesized into one
     coherent video description by the agent's own model.

Frame vision is run with limited concurrency. Requires system ``ffmpeg`` +
``ffprobe`` (same assumption the audio pipeline already makes). Soniox
(``SONIOX_API_KEY``) is optional — without it, analysis is frames-only.
"""

from __future__ import annotations

import asyncio
import shutil
import uuid
from pathlib import Path
from typing import Any

from captain_claw.logging import get_logger
from captain_claw.tools.registry import Tool, ToolResult

log = get_logger(__name__)

_MAX_FRAMES = 20
_DEFAULT_INTERVAL = 6.0      # default: one frame every 6 seconds…
_FIRST_FRAME_OFFSET = 1.0    # …starting ~1 second in (skip the black/intro frame)
_FRAME_CONCURRENCY = 4
_PEER_FRAME_CONCURRENCY = 1  # a peer agent serves one request at a time → serialize
_VISION_PEER_HINTS = ("vision", "image", "multimodal", "minimax", "llava", "qwen2.5vl", "vl")
_VIDEO_EXTS = {".mp4", ".mov", ".webm", ".mkv", ".avi", ".m4v"}
_SONIOX_API_BASE = "https://api.soniox.com"
_SONIOX_STT_MODEL = "stt-async-v4"


# ── subprocess helpers ────────────────────────────────────────────────


async def _run(cmd: list[str], *, input_bytes: bytes | None = None, timeout: float = 120.0) -> tuple[int, bytes, bytes]:
    proc = await asyncio.create_subprocess_exec(
        *cmd,
        stdin=asyncio.subprocess.PIPE if input_bytes is not None else None,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    try:
        out, err = await asyncio.wait_for(proc.communicate(input=input_bytes), timeout=timeout)
    except asyncio.TimeoutError:
        proc.kill()
        return 1, b"", b"timeout"
    return proc.returncode or 0, out or b"", err or b""


async def _ffprobe_duration(ffprobe: str, path: str) -> float:
    rc, out, _ = await _run([
        ffprobe, "-v", "error", "-show_entries", "format=duration",
        "-of", "default=noprint_wrappers=1:nokey=1", path,
    ], timeout=30.0)
    if rc != 0:
        return 0.0
    try:
        return float(out.decode().strip())
    except (ValueError, AttributeError):
        return 0.0


def _seek_args(ss: float, t: float | None) -> tuple[list[str], list[str]]:
    """Return (pre_input_args, post_input_args) for an optional [ss, ss+t] window."""
    pre = ["-ss", f"{ss:.3f}"] if ss and ss > 0 else []
    post = ["-t", f"{t:.3f}"] if t and t > 0 else []
    return pre, post


async def _extract_frames_at(
    ffmpeg: str, path: str, times: list[float], out_dir: Path,
) -> list[tuple[float, Path]]:
    """Extract exactly one frame at each absolute timestamp in *times*.

    One precise ``-ss <t> -i`` seek per frame. This is deterministic — you get
    exactly the scheduled in-range frames — unlike a single ``fps=1/N`` pass,
    whose sampler can silently drop the final frame near the clip's end.
    """
    out_dir.mkdir(parents=True, exist_ok=True)
    sem = asyncio.Semaphore(_FRAME_CONCURRENCY)

    async def _one(idx: int, ts: float) -> tuple[float, Path] | None:
        outp = out_dir / f"frame_{idx:03d}.jpg"
        pre, _ = _seek_args(ts, None)  # accurate enough; fast input seek
        async with sem:
            rc, _, err = await _run([
                ffmpeg, "-hide_banner", "-loglevel", "error", *pre, "-i", path,
                "-frames:v", "1", "-vf", "scale='min(1280,iw)':-2", "-y", str(outp),
            ], timeout=60.0)
        if rc == 0 and outp.is_file() and outp.stat().st_size > 0:
            return (round(ts, 1), outp)
        log.warning("ffmpeg frame extraction at %.1fs failed: %s", ts, err.decode(errors="replace")[:160])
        return None

    results = await asyncio.gather(*(_one(i, ts) for i, ts in enumerate(times)))
    return [r for r in results if r is not None]


async def _extract_audio(ffmpeg: str, path: str, *, ss: float = 0.0, t: float | None = None) -> tuple[bytes, str, str]:
    """Extract speech audio as compressed Opus (16kHz mono).

    Returns (bytes, filename, mime). Opus keeps long videos tiny for the Soniox
    upload (~1-2 KB/s vs ~32 KB/s for WAV → ~115 MB/hour). Falls back to WAV if
    the ffmpeg build lacks libopus.
    """
    pre, post = _seek_args(ss, t)
    rc, out, _ = await _run([
        ffmpeg, "-hide_banner", "-loglevel", "error", *pre, "-i", path, *post,
        "-vn", "-ar", "16000", "-ac", "1", "-c:a", "libopus", "-b:a", "16k",
        "-f", "ogg", "pipe:1",
    ], timeout=180.0)
    if rc == 0 and out:
        return out, "audio.ogg", "audio/ogg"
    # Fallback: uncompressed WAV (still works, just larger upload).
    rc, out, _ = await _run([
        ffmpeg, "-hide_banner", "-loglevel", "error", *pre, "-i", path, *post,
        "-vn", "-ar", "16000", "-ac", "1", "-f", "wav", "pipe:1",
    ], timeout=180.0)
    return (out, "audio.wav", "audio/wav") if rc == 0 else (b"", "audio.wav", "audio/wav")


def _parse_time(val: Any) -> float | None:
    """Parse a time spec: seconds (number/'90') or 'MM:SS' / 'HH:MM:SS'. None if blank/invalid."""
    if val is None:
        return None
    if isinstance(val, (int, float)):
        return float(val) if val >= 0 else None
    s = str(val).strip()
    if not s:
        return None
    try:
        if ":" in s:
            parts = [float(p) for p in s.split(":")]
            secs = 0.0
            for p in parts:
                secs = secs * 60 + p
            return secs if secs >= 0 else None
        return max(0.0, float(s))
    except ValueError:
        return None


# ── Soniox transcription WITH timestamps ──────────────────────────────


async def _transcribe_timestamped(
    audio_bytes: bytes, filename: str = "audio.ogg", mime: str = "audio/ogg",
) -> tuple[str, list[tuple[float, str]]]:
    """Return (full_text, [(start_seconds, token_text), ...]) or ('', [])."""
    import os

    import httpx

    api_key = os.environ.get("SONIOX_API_KEY", "").strip()
    if not api_key or not audio_bytes:
        return "", []
    headers = {"Authorization": f"Bearer {api_key}"}
    file_id = transcription_id = ""
    try:
        async with httpx.AsyncClient(timeout=180.0) as client:
            up = await client.post(
                f"{_SONIOX_API_BASE}/v1/files", headers=headers,
                files={"file": (filename, audio_bytes, mime)},
            )
            up.raise_for_status()
            file_id = str((up.json() or {}).get("id") or "")
            if not file_id:
                return "", []
            cr = await client.post(
                f"{_SONIOX_API_BASE}/v1/transcriptions",
                headers={**headers, "Content-Type": "application/json"},
                json={"model": _SONIOX_STT_MODEL, "file_id": file_id,
                      "enable_language_identification": True},
            )
            cr.raise_for_status()
            transcription_id = str((cr.json() or {}).get("id") or "")
            if not transcription_id:
                return "", []
            for _ in range(120):
                p = await client.get(
                    f"{_SONIOX_API_BASE}/v1/transcriptions/{transcription_id}", headers=headers)
                p.raise_for_status()
                st = str((p.json() or {}).get("status") or "")
                if st in ("completed", "error"):
                    break
                await asyncio.sleep(1)
            text, tokens = "", []
            tr = await client.get(
                f"{_SONIOX_API_BASE}/v1/transcriptions/{transcription_id}/transcript", headers=headers)
            if tr.status_code == 200:
                data = tr.json() or {}
                text = str(data.get("text") or "").strip()
                for tok in (data.get("tokens") or []):
                    t = str(tok.get("text") or "")
                    start_ms = tok.get("start_ms")
                    if t and start_ms is not None:
                        tokens.append((float(start_ms) / 1000.0, t))
            # cleanup
            for pth in (f"/v1/transcriptions/{transcription_id}", f"/v1/files/{file_id}"):
                try:
                    await client.delete(f"{_SONIOX_API_BASE}{pth}", headers=headers)
                except Exception:
                    pass
            return text, tokens
    except Exception as exc:
        log.warning("soniox timestamped transcription failed: %s", exc)
        return "", []


def _transcript_window(tokens: list[tuple[float, str]], center: float, half: float) -> str:
    """Join tokens whose start is within [center-half, center+half]."""
    parts = [t for (ts, t) in tokens if center - half <= ts <= center + half]
    return "".join(parts).strip()


# ── vision-peer delegation (for agents without a local vision model) ───


def _session_metadata(kwargs: dict[str, Any]) -> dict[str, Any]:
    session = kwargs.get("_session")
    meta = getattr(session, "metadata", {}) or {} if session is not None else {}
    if not meta:
        agent = kwargs.get("_agent")
        sess = getattr(agent, "session", None) if agent is not None else None
        meta = getattr(sess, "metadata", {}) or {} if sess is not None else {}
    return meta if isinstance(meta, dict) else {}


def _find_vision_peer(kwargs: dict[str, Any]) -> str | None:
    """Pick a multimodal peer (by name/description) to describe frames for us."""
    meta = _session_metadata(kwargs)
    peers = meta.get("peer_agents") or []
    my_name = str(meta.get("session_display_name") or "").lower()
    for p in peers:
        if not isinstance(p, dict):
            continue
        name = str(p.get("name") or "")
        if not name or name.lower() == my_name:
            continue
        blob = f"{name} {p.get('description') or ''}".lower()
        if any(h in blob for h in _VISION_PEER_HINTS):
            return name
    return None


_TRANSCRIPT_FULL_LIMIT = 700  # chars; longer → show summary + excerpts instead of the full transcript


def _emit_status(agent: Any, text: str) -> None:
    """Transient status line (keeps the user company during the long run)."""
    bc = getattr(agent, "ws_broadcast", None)
    if callable(bc):
        try:
            bc({"type": "status", "status": text})
        except Exception:
            pass


def _emit_message(agent: Any, content: str) -> None:
    """Persistent interim message (renders as a centered system bubble, no completion side-effects)."""
    bc = getattr(agent, "ws_broadcast", None)
    if callable(bc):
        try:
            bc({"type": "chat_message", "role": "system", "content": content})
        except Exception:
            pass


def _whatsapp_waid(kwargs: dict[str, Any]) -> str:
    """The WhatsApp chat this turn arrived from, if any (set by chat_handler)."""
    return str(_session_metadata(kwargs).get("whatsapp_waid") or "").strip()


async def _send_whatsapp(wa_id: str, text: str) -> None:
    """Best-effort interim WhatsApp message (mirrors the voice-note path's UX)."""
    if not wa_id:
        return
    try:
        from captain_claw.tools.whatsapp_send_file import send_whatsapp_text
        ok, err = await send_whatsapp_text(wa_id, text)
        if not ok:
            log.warning("video_vision whatsapp interim send skipped: %s", err)
    except Exception as exc:
        log.warning("video_vision whatsapp interim send failed: %s", exc)


async def _summarize_transcript(agent: Any, text: str, *, with_excerpts: bool) -> str:
    """1-3 sentence summary of the transcript; for long ones, also a few key quotes."""
    if agent is None or not getattr(agent, "provider", None) or not text.strip():
        return ""
    from captain_claw.llm import Message
    sys = (
        "Summarize this audio transcript in 1-3 sentences. Be concise and concrete; "
        "do not invent anything. Keep the original language of the transcript."
    )
    if with_excerpts:
        sys += " Then add 2-4 short, verbatim key excerpts as a bulleted list of quotes."
    try:
        resp = await agent.provider.complete(
            [Message(role="system", content=sys), Message(role="user", content=text)],
        )
        return str(getattr(resp, "content", "") or "").strip()
    except Exception as exc:
        log.warning("video_vision transcript summary failed: %s", exc)
        return ""


def _clean_frame_desc(text: str) -> str:
    """Strip internal-context / memory leakage a peer may parrot into a frame desc."""
    try:
        from captain_claw.llm import _strip_internal_context
        text = _strip_internal_context(text)
    except Exception:
        pass
    # Drop a "Persistent insights from memory:" block (and its bullet list) if
    # the peer echoed its memory context instead of describing the frame.
    out_lines: list[str] = []
    skipping = False
    for line in text.splitlines():
        low = line.strip().lower()
        if low.startswith("persistent insights from memory"):
            skipping = True
            continue
        if skipping:
            # keep skipping bullet/insight lines; stop at a blank or normal line
            if not line.strip() or line.lstrip().startswith(("-", "•", "*", "[")):
                continue
            skipping = False
        out_lines.append(line)
    return "\n".join(out_lines).strip()


async def _describe_frame_via_peer(fdt: Any, fd_url: str, peer: str, frame: Path, prompt: str, kwargs: dict[str, Any]) -> str:
    """Ask a multimodal peer to describe one frame (Flight Deck uploads the image)."""
    call_kwargs = {k: v for k, v in kwargs.items() if k != "_attach_file"}
    call_kwargs["_attach_file"] = str(frame)
    msg = prompt + " Reply with ONLY the description — no preamble, no questions."
    try:
        res = await fdt._consult(fd_url, peer, msg, **call_kwargs)
    except Exception as exc:
        return f"(frame not described: peer error: {exc})"
    if not res.success:
        return f"(frame not described: {res.error})"
    text = str(res.content or "").strip()
    # _consult wraps the reply as "Response from <peer>:\n\n<text>".
    if text.startswith("Response from") and "\n\n" in text:
        text = text.split("\n\n", 1)[1].strip()
    text = _clean_frame_desc(text)
    return text or "(frame not described: empty peer reply)"


class VideoVisionTool(Tool):
    """Analyze a video: sample frames + transcribe audio, then describe it."""

    name = "video_vision"
    timeout_seconds = 600.0
    description = (
        "Analyze and describe a VIDEO file. Samples up to 20 frames across the "
        "video, transcribes the audio (with timestamps), describes each frame "
        "with vision, and synthesizes a coherent description of the whole video "
        "(visuals + spoken content over time). Use for .mp4/.mov/.webm/etc. "
        "Pass the video 'path'. Slow (many vision calls) — call once per video."
    )
    parameters = {
        "type": "object",
        "properties": {
            "path": {"type": "string", "description": "Path to the video file (saved/ workspace or absolute)."},
            "prompt": {
                "type": "string",
                "description": "Optional focus for the description (e.g. 'what is the person doing?').",
            },
            "start": {
                "type": "string",
                "description": "Optional start of the segment to analyze: seconds ('90') or 'MM:SS'/'HH:MM:SS'. Omit for the whole video.",
            },
            "end": {
                "type": "string",
                "description": "Optional end of the segment to analyze (same format as start). Omit to go to the end of the video.",
            },
            "interval": {
                "type": "number",
                "description": "Optional seconds between sampled frames. Default 6 (first frame ~1s in, then every 6s), capped at 20 frames total.",
            },
        },
        "required": ["path"],
    }

    async def execute(self, path: str, prompt: str = "", start: Any = None, end: Any = None, interval: Any = None, **kwargs: Any) -> ToolResult:
        path_str = str(path or "").strip()
        if not path_str:
            return ToolResult(success=False, error="Missing required argument: path")

        ffmpeg = shutil.which("ffmpeg")
        ffprobe = shutil.which("ffprobe")
        if not ffmpeg or not ffprobe:
            return ToolResult(
                success=False,
                error=(
                    "video_vision requires system 'ffmpeg' and 'ffprobe' (install with "
                    "'apt install ffmpeg'). Do NOT substitute a custom script (cv2/etc.) — "
                    "that skips audio transcription. Report this so ffmpeg can be installed."
                ),
            )

        # Resolve the video file.
        from captain_claw.tools.image_ocr import _require_existing_file
        runtime_base = kwargs.get("_runtime_base_path")
        file_path, error = _require_existing_file(path_str, runtime_base_path=runtime_base)
        if error:
            return ToolResult(success=False, error=error)
        if file_path.suffix.lower() not in _VIDEO_EXTS:
            return ToolResult(
                success=False,
                error=f"Not a video ('{file_path.suffix}'). Supported: {', '.join(sorted(_VIDEO_EXTS))}.",
            )

        duration = await _ffprobe_duration(ffprobe, str(file_path))
        if duration <= 0:
            return ToolResult(success=False, error="Could not read the video's duration (corrupt or unsupported codec?).")

        # v2: optional [start, end] segment. Clamp to the video; default = whole.
        seg_start = _parse_time(start) or 0.0
        seg_start = max(0.0, min(seg_start, duration))
        parsed_end = _parse_time(end)
        seg_end = duration if parsed_end is None else max(0.0, min(parsed_end, duration))
        if seg_end <= seg_start:
            seg_end = duration  # invalid/zero window → fall back to the rest of the video
        seg_dur = seg_end - seg_start
        is_segment = seg_start > 0.0 or seg_end < duration

        # Default cadence: first frame ~1s in, then every 6s, capped at 20.
        # If the segment is long enough that 6s would exceed the cap, widen the
        # interval so 20 frames still span the whole segment (no silent tail-drop).
        try:
            interval = float(interval) if interval is not None else _DEFAULT_INTERVAL
        except (TypeError, ValueError):
            interval = _DEFAULT_INTERVAL
        interval = max(0.5, interval)
        first_off = _FIRST_FRAME_OFFSET if seg_dur > _FIRST_FRAME_OFFSET else 0.0
        usable = max(0.0, seg_dur - first_off)
        n_frames = max(1, int(usable // interval) + 1)
        widened = False
        if n_frames > _MAX_FRAMES:
            n_frames = _MAX_FRAMES
            interval = usable / (_MAX_FRAMES - 1) if _MAX_FRAMES > 1 else interval
            widened = True
        # Explicit absolute timestamps: first_off into the segment, then every
        # `interval`s. Keep only those strictly inside the segment (small epsilon
        # so a frame isn't requested past EOF). Extracted one-seek-each below.
        frame_times = [seg_start + first_off + i * interval for i in range(n_frames)]
        frame_times = [ts for ts in frame_times if ts < seg_end - 0.02] or [seg_start + first_off]
        n_frames = len(frame_times)
        # Audio covers the whole segment (frames start ~1s in, audio doesn't).
        audio_ss = seg_start if is_segment else 0.0
        audio_t = seg_dur if is_segment else None

        # Frame output dir under saved/.
        saved_base = kwargs.get("_saved_base_path")
        base = Path(saved_base) if saved_base else (Path(runtime_base) / "saved" if runtime_base else file_path.parent)
        frame_dir = Path(base) / "video_frames" / uuid.uuid4().hex[:8]

        log.info("video_vision: analyzing", path=str(file_path), duration=round(duration, 1),
                 segment=f"{round(seg_start,1)}-{round(seg_end,1)}s" if is_segment else "full",
                 first_at=round(frame_times[0], 1), interval=round(interval, 1),
                 frames=n_frames, widened=widened)

        # Extract frames (one seek each, exact timestamps) + audio (whole segment).
        frame_pairs, (audio, audio_name, audio_mime) = await asyncio.gather(
            _extract_frames_at(ffmpeg, str(file_path), frame_times, frame_dir),
            _extract_audio(ffmpeg, str(file_path), ss=audio_ss, t=audio_t),
        )
        if not frame_pairs:
            return ToolResult(success=False, error="Frame extraction produced no frames (ffmpeg failed?).")
        frames = [fp for _, fp in frame_pairs]

        agent_obj = kwargs.get("_agent")
        wa_id = _whatsapp_waid(kwargs)  # if set, mirror key updates to WhatsApp

        # ── Phase 1: audio → transcript → summary (fast; shown first) ──────
        import os as _os
        has_audio_track = bool(audio)
        has_soniox = bool(_os.environ.get("SONIOX_API_KEY", "").strip())
        _emit_status(agent_obj, "🎙️ Transcribing audio…")
        await _send_whatsapp(wa_id, "🎙 Transcribing the video's audio…")
        full_text, tokens = await _transcribe_timestamped(audio, audio_name, audio_mime)
        # Disambiguate the empty-transcript reasons so "no audio" isn't misleading.
        if not full_text:
            if not has_audio_track:
                audio_note = "🎙 The video has no audio track."
            elif not has_soniox:
                audio_note = "🎙 Audio is present, but transcription isn't configured (no SONIOX_API_KEY)."
            else:
                audio_note = "🎙 No speech detected in the audio."
        if full_text:
            is_long = len(full_text) > _TRANSCRIPT_FULL_LIMIT
            if not is_long:
                _emit_message(agent_obj, f"🎙️ **Transcript**\n\n{full_text}")
            _emit_status(agent_obj, "📝 Summarizing transcript…")
            tsummary = await _summarize_transcript(agent_obj, full_text, with_excerpts=is_long)
            if tsummary:
                label = "📝 **Transcript summary**" if is_long else "📝 **Summary**"
                _emit_message(agent_obj, f"{label}\n\n{tsummary}")
            # WhatsApp: short → full transcript (like the voice-note path);
            # long → the summary+excerpts so we don't flood the chat.
            if is_long and tsummary:
                await _send_whatsapp(wa_id, f"🎙 Transcription (summary):\n\n{tsummary}")
            else:
                await _send_whatsapp(wa_id, f"🎙 Transcription:\n\n\"{full_text}\"")
        else:
            _emit_message(agent_obj, audio_note)
            await _send_whatsapp(wa_id, audio_note)

        # ── Phase 2: per-frame vision (the slow part) ──────────────────────
        # Describe each frame (with nearby transcript) — limited concurrency.
        # If THIS agent has no vision model, image_vision would fall back to the
        # text chat model and reject the image. In that case, delegate frame
        # description to a multimodal peer (same path as the WhatsApp image flow).
        from captain_claw.tools.image_ocr import ImageVisionTool
        ivt = ImageVisionTool()
        has_local_vision = ivt._find_model() is not None
        vision_peer = None if has_local_vision else _find_vision_peer(kwargs)
        delegating = vision_peer is not None
        fdt = fd_url = None
        if delegating:
            from captain_claw.tools.flight_deck import FlightDeckTool
            fdt = FlightDeckTool()
            fd_url = fdt._get_fd_url(**kwargs)
            if not fd_url:
                delegating = False  # no Flight Deck URL → fall back to local (will report the error)

        sem = asyncio.Semaphore(_PEER_FRAME_CONCURRENCY if delegating else _FRAME_CONCURRENCY)
        half = max(interval / 2.0, 1.5)
        log.info("video_vision: frame vision path",
                 mode=("peer:" + vision_peer if delegating else ("local" if has_local_vision else "local-fallback")))

        total = len(frames)
        frame_label = f" via {vision_peer}" if delegating else ""
        prog = {"n": 0}
        _emit_status(agent_obj, f"🎞️ Describing {total} frames{frame_label}… (0/{total})")
        # One WhatsApp heads-up (the live per-frame counter stays web/glasses-only).
        await _send_whatsapp(wa_id, f"🎞 Analyzing {total} frames now — this takes a couple of minutes…")

        async def describe(abs_ts: float, fp: Path) -> dict[str, Any]:
            rel = abs_ts - seg_start                  # offset within the segment
            # Transcript tokens are segment-relative (audio extracted from seg_start).
            audio = _transcript_window(tokens, rel, half) if tokens else ""
            fprompt = "Describe this single video frame concisely: people, objects, on-screen text, action."
            if audio:
                fprompt += f" Audio spoken around this moment: \"{audio}\"."
            async with sem:
                try:
                    if delegating:
                        desc = await _describe_frame_via_peer(fdt, fd_url, vision_peer, fp, fprompt, kwargs)
                    else:
                        res = await ivt.execute(path=str(fp), prompt=fprompt, **kwargs)
                        desc = res.content if res.success else f"(frame not described: {res.error})"
                except Exception as exc:
                    desc = f"(frame error: {exc})"
            prog["n"] += 1
            _emit_status(agent_obj, f"🎞️ Describing frames{frame_label}… ({prog['n']}/{total})")
            return {"t": round(abs_ts, 1), "desc": desc.strip(), "audio": audio}

        analyzed = await asyncio.gather(*(describe(ts, fp) for ts, fp in frame_pairs))

        # Build the per-frame breakdown.
        lines = []
        for a in analyzed:
            seg = f"[{a['t']}s] {a['desc']}"
            if a["audio"]:
                seg += f"  (audio: \"{a['audio']}\")"
            lines.append(seg)
        breakdown = "\n".join(lines)

        _emit_status(agent_obj, "🧩 Synthesizing the final description…")
        # Step 5 — synthesize a coherent description via the agent's own model.
        scope = (f"segment {round(seg_start)}-{round(seg_end)}s of a {round(duration)}s video"
                 if is_segment else f"a {round(duration)}s video")
        summary = await self._synthesize(kwargs.get("_agent"), scope, full_text, breakdown, prompt)

        seg_label = f" of {round(seg_start)}-{round(seg_end)}s" if is_segment else ""
        vision_note = (f" (frames described by peer '{vision_peer}')" if delegating
                       else ("" if has_local_vision else " (no vision model/peer — frames not described)"))
        out = summary or "(synthesis unavailable)"
        out += f"\n\n— Frame-by-frame{seg_label} ({len(frames)} frames @ ~{round(interval,1)}s{vision_note}) —\n{breakdown}"
        if full_text:
            out += f"\n\n— Transcript —\n{full_text}"
        return ToolResult(success=True, content=out)

    @staticmethod
    async def _synthesize(agent: Any, scope: str, transcript: str, breakdown: str, focus: str) -> str:
        if agent is None or not getattr(agent, "provider", None):
            return ""
        from captain_claw.llm import Message
        sys = (
            "You are given a frame-by-frame visual analysis of a video (or a "
            "segment of it) plus its audio transcript. Write a clear, coherent "
            "description: what it shows, what happens over time, and the key "
            "spoken content. Be concise and concrete; do not invent details."
        )
        user = f"This is {scope}.\n\n"
        if focus.strip():
            user += f"Focus on: {focus.strip()}\n\n"
        user += f"Frame-by-frame analysis:\n{breakdown}\n\n"
        user += f"Full transcript:\n{transcript or '(no speech detected)'}"
        try:
            resp = await agent.provider.complete(
                [Message(role="system", content=sys), Message(role="user", content=user)],
            )
            return str(getattr(resp, "content", "") or "").strip()
        except Exception as exc:
            log.warning("video_vision synthesis failed: %s", exc)
            return ""
