"""AgentLoopTask actions — conversation"""

import json
import logging
import shutil
import time
import threading
from pathlib import Path
from typing import Dict, Any, List, Optional

from core import FlowFile
from core.llm_client import LLMMessage, LLMClient
from core.tool_registry import ToolRegistry

logger = logging.getLogger(__name__)


def _zip_rel_path(name: str, prefix: str = "") -> Optional[Path]:
    """Return a safe archive-relative path, optionally stripping prefix."""
    if prefix:
        if name == prefix.rstrip("/"):
            return None
        if not name.startswith(prefix):
            return None
        name = name[len(prefix):]
    if not name or name.endswith("/"):
        return None
    rel = Path(name)
    if rel.is_absolute() or ".." in rel.parts:
        raise ValueError(f"Unsafe archive path: {name}")
    if rel.parts and rel.parts[0] in (".git", "filestore"):
        return None
    if str(rel) == "manifest.json":
        return None
    return rel


def _archive_manifest(zf) -> Dict[str, Any]:
    if "manifest.json" not in zf.namelist():
        return {}
    try:
        return json.loads(zf.read("manifest.json").decode("utf-8"))
    except Exception:
        return {}


def _extract_conversation_members(zf, conv_dir: Path) -> None:
    """Extract conversation files while excluding manifest/FileStore payloads."""
    from core.segmented_jsonl import SegmentedJsonl

    names = zf.namelist()
    prefix = "conversation/" if any(n.startswith("conversation/") for n in names) else ""
    for name in names:
        rel = _zip_rel_path(name, prefix=prefix)
        if rel is None:
            continue
        dest = conv_dir / rel
        dest.parent.mkdir(parents=True, exist_ok=True)
        if rel.suffix == ".jsonl":
            text = zf.read(name).decode("utf-8", errors="replace")
            SegmentedJsonl(dest).replace_lines(
                line + "\n" for line in text.splitlines() if line.strip())
            continue
        with zf.open(name) as src, dest.open("wb") as out:
            shutil.copyfileobj(src, out, length=1024 * 1024)


def _patch_json_identity(obj: Any, cid: str, user_id: str,
                         file_id_map: Dict[str, str]) -> Any:
    if isinstance(obj, dict):
        out = {}
        for key, value in obj.items():
            if key == "conversation_id":
                out[key] = cid
            elif key in ("user_id", "_meta_user_id"):
                out[key] = user_id
            else:
                out[key] = _patch_json_identity(value, cid, user_id, file_id_map)
        return out
    if isinstance(obj, list):
        return [_patch_json_identity(v, cid, user_id, file_id_map) for v in obj]
    if isinstance(obj, str) and file_id_map:
        text = obj
        for old_id, new_id in file_id_map.items():
            text = text.replace(old_id, new_id)
        return text
    return obj


def _patch_conversation_files(conv_dir: Path, cid: str, user_id: str,
                              file_id_map: Dict[str, str]) -> None:
    """Patch imported JSON/JSONL identity fields without rebuilding caches."""
    for path in sorted(conv_dir.rglob("*")):
        if not path.is_file() or ".git" in path.parts:
            continue
        if path.suffix == ".json":
            try:
                data = json.loads(path.read_text(encoding="utf-8"))
            except (OSError, json.JSONDecodeError) as exc:
                logger.debug("Skipped non-JSON archive file %s: %s", path, exc)
            else:
                patched = _patch_json_identity(data, cid, user_id, file_id_map)
                path.write_text(json.dumps(patched, ensure_ascii=False, indent=2), encoding="utf-8")
        elif path.suffix == ".jsonl":
            lines = []
            changed = False
            for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
                if not line.strip():
                    continue
                try:
                    row = json.loads(line)
                except json.JSONDecodeError:
                    lines.append(line)
                    continue
                patched = _patch_json_identity(row, cid, user_id, file_id_map)
                lines.append(json.dumps(patched, ensure_ascii=False))
                changed = True
            if changed:
                path.write_text("\n".join(lines) + "\n", encoding="utf-8")


def _ensure_import_summarizer_binding(cid: str, user_id: str) -> Dict[str, str]:
    """Bind an available summarizer on imported conversations.

    Archives can carry a summarizer_binding from another install. If that
    explicit binding is unavailable locally, summarizer resolution stops there
    instead of falling back to user/global services. Normalize the imported
    conversation to the first locally available summarizer.
    """
    try:
        from core.summarizer_bindings import list_available, set_binding, summary
        current = summary(user_id, cid)
        effective = current.get("effective") or {}
        if current.get("explicit") and effective.get("service_id"):
            return {
                "scope": effective.get("scope", ""),
                "service_id": effective.get("service_id", ""),
            }
        available = list_available(user_id, cid)
        if not available:
            return {}
        chosen = available[0]
        scope = chosen.get("scope", "")
        service_id = chosen.get("service_id", "")
        if scope and service_id:
            set_binding(cid, scope, service_id)
            return {"scope": scope, "service_id": service_id}
    except Exception:
        logger.debug("failed to bind summarizer for imported conversation %s", cid[:8], exc_info=True)
    return {}


def _write_filestore_archive(zf, conv_id: str, user_id: str) -> Dict[str, Any]:
    """Write FileStore files for a conversation into an archive."""
    from core.file_store import FileStore
    fs = FileStore.instance()
    objects = []
    total_size = 0
    with fs._store_lock:  # FileStore has no export API yet.
        fs._ensure_loaded()
        entries = [(fid, dict(entry)) for fid, entry in fs._entries.items()
                   if entry.get("conversation_id") == conv_id
                   and entry.get("user_id") == user_id]
    for fid, entry in entries:
        src = Path(entry.get("path", ""))
        if not src.is_file():
            continue
        filename = Path(entry.get("filename") or src.name).name or "file"
        arc = f"filestore/objects/{fid}_{filename}"
        size = src.stat().st_size
        zf.write(src, arc)
        total_size += size
        objects.append({
            "file_id": fid,
            "filename": filename,
            "content_type": entry.get("content_type", "application/octet-stream"),
            "size": size,
            "created_at": entry.get("created_at", 0),
            "access": entry.get("access", "private"),
            "shared_with": entry.get("shared_with", []),
            "ttl": entry.get("ttl", 0),
            "agent_name": entry.get("agent_name", ""),
            "category": entry.get("category", ""),
            "path": f"objects/{fid}_{filename}",
        })
    zf.writestr("filestore/index.json", json.dumps(objects, ensure_ascii=False, indent=2))
    return {"included": True, "count": len(objects), "bytes": total_size}


def _restore_filestore_archive(zf, cid: str, user_id: str,
                               restore: bool,
                               file_id_policy: str = "preserve_or_remap"
                               ) -> Dict[str, Any]:
    """Restore archived FileStore entries and return an old->new id map."""
    if "filestore/index.json" not in zf.namelist():
        return {"restored": 0, "bytes": 0, "file_id_map": {}}
    if not restore:
        return {"restored": 0, "bytes": 0, "file_id_map": {}}
    import uuid as _uuid
    from core.file_store import FileStore
    try:
        entries = json.loads(zf.read("filestore/index.json").decode("utf-8"))
    except Exception as e:
        raise ValueError(f"Invalid FileStore index: {e}")
    if file_id_policy not in ("preserve", "remap", "preserve_or_remap"):
        raise ValueError("Invalid file_id_policy")
    fs = FileStore.instance()
    id_map: Dict[str, str] = {}
    total_size = 0
    restored = 0
    with fs._store_lock:
        fs._ensure_loaded()
        used = set(fs._entries.keys())
        for meta in entries:
            old_id = str(meta.get("file_id", ""))
            if not old_id:
                continue
            collision = old_id in used
            if file_id_policy == "preserve" and collision:
                raise ValueError(f"FileStore file_id collision: {old_id}")
            if file_id_policy == "remap" or collision:
                new_id = _uuid.uuid4().hex[:12]
                while new_id in used:
                    new_id = _uuid.uuid4().hex[:12]
            else:
                new_id = old_id
            used.add(new_id)
            if new_id != old_id:
                id_map[old_id] = new_id

            rel_obj = str(meta.get("path", ""))
            if not rel_obj or rel_obj.startswith("/") or ".." in Path(rel_obj).parts:
                raise ValueError(f"Unsafe FileStore object path: {rel_obj}")
            arc = "filestore/" + rel_obj
            if arc not in zf.namelist():
                raise ValueError(f"Missing FileStore object: {rel_obj}")
            filename = Path(meta.get("filename") or "file").name or "file"
            scope_dir = fs._scope_dir(user_id, cid)
            bucket = fs._pick_bucket(scope_dir)
            bucket_dir = scope_dir / bucket
            bucket_dir.mkdir(parents=True, exist_ok=True)
            disk_path = bucket_dir / f"{new_id}_{filename}"
            with zf.open(arc) as src, disk_path.open("wb") as out:
                shutil.copyfileobj(src, out, length=1024 * 1024)
            size = disk_path.stat().st_size
            total_size += size
            restored += 1
            fs._entries[new_id] = {
                "filename": filename,
                "path": str(disk_path),
                "content_type": meta.get("content_type", "application/octet-stream"),
                "size": size,
                "created_at": meta.get("created_at", time.time()),
                "conversation_id": cid,
                "user_id": user_id,
                "access": meta.get("access", "private"),
                "shared_with": meta.get("shared_with", []),
                "ttl": meta.get("ttl", 0),
                "agent_name": meta.get("agent_name", ""),
                "category": meta.get("category", ""),
            }
        fs._save_index()
    return {"restored": restored, "bytes": total_size, "file_id_map": id_map}


def _handle_conversation(self, action, body, store, user_id, flowfile):
    """Handle conversation actions. Returns [flowfile] or None."""


    if action == "list_conversations":
        convs = store.list_conversations(user_id=user_id)
        # Override persisted status with real-time active agent state
        try:
            from tasks.ai.agent_loop import AgentLoopTask
            inst = AgentLoopTask._live_instance
            if inst:
                _active_cids = set()
                with inst._active_contexts_lock:
                    for k in inst._active_contexts:
                        _active_cids.add(k.split(":")[0])
                for c in convs:
                    c["status"] = "active" if c["conversation_id"] in _active_cids else "idle"
        except Exception:
            logging.getLogger(__name__).debug("Ignored exception", exc_info=True)
        for c in convs:
            branch = store.git_current_branch(c["conversation_id"])
            c["branch"] = branch or ""
        result = json.dumps({"conversations": convs}, ensure_ascii=False)
        flowfile.set_content(result.encode("utf-8"))
        return [flowfile]

    if action == "load_history":
        conv_id = body.get("conversation_id", "")
        limit = int(body.get("limit", 50))
        offset = int(body.get("offset", 0))
        if not conv_id:
            flowfile.set_content(json.dumps({"error": "Missing conversation_id"}).encode())
            flowfile.set_attribute("http.response.status", "400")
            return [flowfile]

        page = store.load_page(conv_id, limit=limit, offset=offset, user_id=user_id)
        if page is None:
            flowfile.set_content(json.dumps({"error": "Conversation not found"}).encode())
            flowfile.set_attribute("http.response.status", "404")
            return [flowfile]

        raw_messages = page["messages"]
        raw_count = len(raw_messages)
        history = self._classify_messages_for_display(raw_messages)
        extras = store.get_extras_snapshot(conv_id)
        nicknames = extras.get("agent_nicknames") or {}
        active_res = extras.get("active_resources") or {}
        active_res = self._ensure_active_agent(conv_id, active_res, user_id)
        active_agent = active_res.get("agent", "")
        custom_css = extras.get("custom_css") or ""
        def _resolve_chat_flag(key: str, default: str = "true") -> bool:
            try:
                from core.expression import resolve_expression
                raw = resolve_expression(
                    "$" + '{chat.' + key + ':default("' + default + '")}',
                    owner=user_id,
                    conversation_id=conv_id,
                )
            except Exception:
                raw = default
            return str(raw).strip().lower() in ("1", "true", "yes", "on")

        group_technical_messages = _resolve_chat_flag("group_technical_messages")
        group_task_messages = _resolve_chat_flag("group_task_messages")
        group_delegate_messages = _resolve_chat_flag("group_delegate_messages")

        result = json.dumps({
            "conversation_id": conv_id,
            "messages": history,
            "message_count": page["total_count"],
            "has_more": page["has_more"],
            "offset": page["offset"],
            "raw_count": raw_count,
            "nicknames": nicknames,
            "active_agent": active_agent,
            "custom_css": custom_css,
            "group_technical_messages": group_technical_messages,
            "group_task_messages": group_task_messages,
            "group_delegate_messages": group_delegate_messages,
        }, ensure_ascii=False)
        flowfile.set_content(result.encode("utf-8"))
        return [flowfile]

    if action == "set_conv_title":
        conv_id = body.get("conversation_id", "")
        title = body.get("title", "").strip()
        if not conv_id or not title:
            flowfile.set_content(json.dumps({"error": "Missing conversation_id or title"}).encode())
            return [flowfile]
        store.set_extra(conv_id, "title", title, user_id=user_id)
        flowfile.set_content(json.dumps({"ok": True, "title": title}).encode())
        return [flowfile]

    if action == "delete_conversation":
        conv_id = body.get("conversation_id", "")
        if not conv_id:
            flowfile.set_content(json.dumps({"error": "Missing conversation_id"}).encode())
            flowfile.set_attribute("http.response.status", "400")
            return [flowfile]
        # Collect file IDs from conversation before deleting
        history = store.load(conv_id, user_id=user_id)
        if history:
            self._cleanup_conversation_files(history)
        # Cascade cleanup: flows, dynamic tools, secrets
        self._cleanup_conversation_resources(conv_id)
        deleted = store.delete(conv_id, user_id=user_id)
        logger.info(f"[action] delete_conversation {conv_id}: deleted={deleted}, "
                    f"user_id={user_id}")
        result = json.dumps({"deleted": deleted, "conversation_id": conv_id})
        flowfile.set_content(result.encode("utf-8"))
        return [flowfile]

    if action == "resume_conversation":
        conv_id = body.get("conversation_id", "")
        _rs_agent = body.get("agent_name", "")
        max_summary_tokens = int(body.get("max_tokens", 500))
        if not conv_id:
            flowfile.set_content(json.dumps({"error": "Missing conversation_id"}).encode())
            flowfile.set_attribute("http.response.status", "400")
            return [flowfile]
        _rs_msgs = store.load(conv_id, user_id=user_id)
        if not _rs_msgs:
            flowfile.set_content(json.dumps({"error": "Conversation not found"}).encode())
            flowfile.set_attribute("http.response.status", "404")
            return [flowfile]
        # Resolve LLM client
        _summ_client, _, _ = self._get_summarizer_client(user_id, conversation_id=conv_id)
        _rs_client = _summ_client
        if not _rs_client:
            _rs_svc = self._resolve_service_param("llm_service", user_id) or "default"
            _rs_client, _ = self._resolve_client(_rs_svc, user_id)
        if not _rs_client:
            flowfile.set_content(json.dumps({"error": "No LLM service for summarization"}).encode())
            return [flowfile]

        def _do_resume():
            deserialized = self._deserialize_messages(_rs_msgs, conversation_id=conv_id)
            content_msgs = [m for m in deserialized if m.role != "system"]
            context_max = int(self.config.get("max_context_size", 64000))
            # Resolve agent's max_tokens
            if _rs_agent:
                _, _, _sv = self._resolve_agent_client(_rs_agent, user_id, conv_id)
                if _sv:
                    _v = int((getattr(_sv, 'config', {}) or {}).get("max_context_size", 0))
                    if _v:
                        context_max = _v
            summary = self._summarize_messages(
                content_msgs, _rs_client, context_max,
                target_tokens=max_summary_tokens,
                conversation_id=conv_id,
            )
            sys_prompt = self.config.get("system_prompt", "You are a helpful assistant.")
            # NO datetime in system prompt — breaks KV cache
            new_context = [
                LLMMessage(role="system", content=sys_prompt,
                            conversation_id=conv_id),
                LLMMessage(role="user",
                           content=f"[Conversation summary â€” earlier messages compacted]\n\n{summary}",
                           conversation_id=conv_id),
                LLMMessage(role="assistant",
                           content="Understood. I have the context from our earlier conversation. Continuing from where we left off.",
                           source={"type": "context"},
                           conversation_id=conv_id),
            ]
            store.save_agent_context(conv_id, _rs_agent, self._serialize_messages(new_context))
            return {"summary_length": len(summary),
                    "messages_summarized": len(_rs_msgs),
                    "agent": _rs_agent or "shared"}

        _rs_lock_agent = (
            "" if _rs_agent in ("", "ALL", "shared") else _rs_agent)
        return self._run_bg_context_op(
            conv_id, "summary", _do_resume, flowfile,
            agent_name=_rs_lock_agent)

    if action == "ping":
        # Keep-alive: session renewal happens in validateSessionAuth upstream
        flowfile.set_content(json.dumps({"status": "ok"}).encode())
        return [flowfile]

    if action == "poll":
        # Efficient delta check: client sends last known message_count,
        # server returns new messages only if count increased.
        conv_id = body.get("conversation_id", "")
        last_count = int(body.get("last_count", 0))
        if not conv_id:
            flowfile.set_content(json.dumps({"new_messages": []}).encode())
            return [flowfile]
        current_count = int(store.get_extra_snapshot(
            conv_id, "_meta_msg_count", last_count) or last_count)
        if current_count <= last_count:
            flowfile.set_content(json.dumps({
                "new_messages": [], "message_count": current_count,
            }).encode())
            return [flowfile]
        # Load only the new messages (max 50) — never the full conversation
        delta = current_count - last_count
        if delta > 50:
            # Too many missed — client should just update count, not render all
            delta = 50
        page = store.load_page(conv_id, limit=delta, offset=0, user_id=user_id)
        if page is None:
            flowfile.set_content(json.dumps({
                "new_messages": [], "message_count": current_count,
            }).encode())
            return [flowfile]
        new_classified = self._classify_messages_for_display(page["messages"])
        flowfile.set_content(json.dumps({
            "new_messages": new_classified,
            "message_count": current_count,
        }, ensure_ascii=False).encode())
        return [flowfile]

    if action == "export":
        fmt = body.get("format", "markdown")
        conv_id = body.get("conversation_id", "")
        if not conv_id:
            flowfile.set_content(json.dumps({"error": "No conversation to export"}).encode())
            flowfile.set_attribute("http.response.status", "400")
            return [flowfile]
        msgs = store.load(conversation_id=conv_id, user_id=user_id)
        if not msgs:
            flowfile.set_content(json.dumps({"error": "Conversation not found or empty"}).encode())
            flowfile.set_attribute("http.response.status", "404")
            return [flowfile]

        if fmt == "json":
            export = json.dumps([
                {"role": m.get("role", ""), "content": m.get("content", ""),
                 "source": m.get("source", None)}
                if isinstance(m, dict) else
                {"role": m.role, "content": m.content,
                 "source": getattr(m, "source", None)}
                for m in msgs
            ], indent=2, ensure_ascii=False)
            filename = f"conversation_{conv_id[:8]}.json"
        else:
            lines = [f"# Conversation {conv_id[:8]}\n"]
            for m in msgs:
                if isinstance(m, dict):
                    role = (m.get("role") or "").upper()
                    source = m.get("source")
                    content = m.get("content", "")
                else:
                    role = (m.role or "").upper()
                    source = getattr(m, "source", None)
                    content = m.content if isinstance(m.content, str) else str(m.content)
                if source and isinstance(source, dict) and source.get("name"):
                    role = f"{role} ({source['name']})"
                lines.append(f"## {role}\n\n{content}\n")
            export = "\n".join(lines)
            filename = f"conversation_{conv_id[:8]}.md"

        # Store in FileStore for download
        from core.file_store import FileStore
        mime = "application/json" if fmt == "json" else "text/markdown"
        fid = FileStore.instance().store(filename, export.encode("utf-8"), mime,
                                           user_id=user_id, conversation_id=conv_id)
        flowfile.set_content(json.dumps({
            "ok": True,
            "url": f"/files/{fid}/{filename}",
            "filename": filename,
            "format": fmt,
        }).encode())
        return [flowfile]

    # Filesystem explorer actions

    if action == "clear":
        conv_id = body.get("conversation_id", "")
        if not conv_id:
            flowfile.set_content(json.dumps({"error": "Missing conversation_id"}).encode())
            flowfile.set_attribute("http.response.status", "400")
            return [flowfile]
        _clear_msgs = store.load(conv_id, user_id=user_id)
        if not _clear_msgs:
            flowfile.set_content(json.dumps({"error": "Conversation not found"}).encode())
            flowfile.set_attribute("http.response.status", "404")
            return [flowfile]

        def _do_clear():
            deserialized = self._deserialize_messages(_clear_msgs, conversation_id=conv_id)
            system_msgs = [m for m in deserialized if m.role == "system"]
            serialized_ctx = self._serialize_messages(system_msgs)
            # Clear shared context
            store.save_agent_context(conv_id, "", serialized_ctx)
            # Clear all agent-specific contexts
            agent_ctxs = store.list_agent_contexts(conv_id)
            count = 0
            for agent_name in agent_ctxs:
                if agent_name == "*":
                    continue
                store.save_agent_context(conv_id, agent_name, serialized_ctx)
                count += 1
            return {"cleared": True, "agents_reset": count + 1}

        return self._run_bg_context_op(conv_id, "clear", _do_clear, flowfile)

    if action == "clear_store":
        conv_id = body.get("conversation_id", "")
        if not conv_id:
            flowfile.set_content(json.dumps({"error": "Missing conversation_id"}).encode())
            flowfile.set_attribute("http.response.status", "400")
            return [flowfile]
        from core.file_store import FileStore
        fs = FileStore.instance()
        scope = body.get("scope", "")
        agent_name = body.get("agent_name", "")
        if agent_name:
            # Delete tool results for a specific agent
            count = fs.delete_by(category="tool_result",
                                 conversation_id=conv_id, agent_name=agent_name)
            flowfile.set_content(json.dumps({
                "deleted": count, "scope": f"agent:{agent_name}",
            }).encode())
        elif scope == "all_agents":
            # Delete all tool results for all agents in this conversation
            count = fs.delete_by(category="tool_result", conversation_id=conv_id)
            flowfile.set_content(json.dumps({
                "deleted": count, "scope": "all_agents",
            }).encode())
        else:
            # Delete ALL filestore files for this conversation
            count = fs.delete_by(conversation_id=conv_id)
            flowfile.set_content(json.dumps({
                "deleted": count, "scope": "conversation",
            }).encode())
        return [flowfile]

    if action == "loop_start":
        conv_id = body.get("conversation_id", "")
        interval = body.get("interval_seconds", 600)
        prompt = body.get("prompt", "")
        if not conv_id or not prompt:
            flowfile.set_content(json.dumps({"error": "Missing conversation_id or prompt"}).encode())
            flowfile.set_attribute("http.response.status", "400")
            return [flowfile]
        from core.poll_scheduler import PollScheduler
        key = PollScheduler.instance().schedule_loop(
            conv_id, interval, prompt=prompt, user_id=user_id)
        flowfile.set_content(json.dumps({
            "started": True, "key": key, "interval": interval, "prompt": prompt,
        }).encode())
        return [flowfile]

    if action == "loop_stop":
        key = body.get("key", "")
        if not key:
            flowfile.set_content(json.dumps({"error": "Missing key"}).encode())
            flowfile.set_attribute("http.response.status", "400")
            return [flowfile]
        from core.poll_scheduler import PollScheduler
        ok = PollScheduler.instance().cancel(key)
        flowfile.set_content(json.dumps({"stopped": ok, "key": key}).encode())
        return [flowfile]

    if action == "loop_list":
        conv_id = body.get("conversation_id", "")
        from core.poll_scheduler import PollScheduler
        loops = PollScheduler.instance().list_loops(conv_id)
        flowfile.set_content(json.dumps({"loops": loops}).encode())
        return [flowfile]

    # ── Git versioning ─────────────────────────────────────────────

    if action == "conv_git_log":
        conv_id = body.get("conversation_id", "")
        limit = int(body.get("limit", 30))
        if not conv_id:
            flowfile.set_content(json.dumps({"error": "Missing conversation_id"}).encode())
            return [flowfile]
        commits = store.git_log(conv_id, limit=limit)
        tags = store.git_list_tags(conv_id)
        tag_by_commit = {t["commit"]: t["name"] for t in tags}
        for c in commits:
            c["tag"] = tag_by_commit.get(c["hash"][:7], "")
        flowfile.set_content(json.dumps({
            "commits": commits,
            "branch": store.git_current_branch(conv_id),
        }).encode())
        return [flowfile]

    if action == "conv_fork":
        conv_id = body.get("conversation_id", "")
        if not conv_id:
            flowfile.set_content(json.dumps({"error": "Missing conversation_id"}).encode())
            return [flowfile]
        try:
            new_cid = store.fork(conv_id, user_id)
            flowfile.set_content(json.dumps({
                "ok": True, "conversation_id": new_cid, "source": conv_id,
            }).encode())
        except (RuntimeError, ValueError) as e:
            flowfile.set_content(json.dumps({"error": str(e)}).encode())
        return [flowfile]

    if action == "conv_branch":
        conv_id = body.get("conversation_id", "")
        branch_name = body.get("branch_name", "").strip()
        if not conv_id or not branch_name:
            flowfile.set_content(json.dumps({"error": "Missing conversation_id or branch_name"}).encode())
            return [flowfile]
        try:
            ok = store.git_branch(conv_id, branch_name)
            flowfile.set_content(json.dumps({
                "ok": ok, "branch": branch_name,
            }).encode())
        except RuntimeError as e:
            flowfile.set_content(json.dumps({"error": str(e)}).encode())
        return [flowfile]

    if action == "conv_switch_branch":
        conv_id = body.get("conversation_id", "")
        branch_name = body.get("branch_name", "").strip()
        if not conv_id or not branch_name:
            flowfile.set_content(json.dumps({"error": "Missing conversation_id or branch_name"}).encode())
            return [flowfile]
        try:
            ok = store.git_switch(conv_id, branch_name)
            flowfile.set_content(json.dumps({"ok": ok, "branch": branch_name}).encode())
        except RuntimeError as e:
            flowfile.set_content(json.dumps({"error": str(e)}).encode())
        return [flowfile]

    if action == "conv_list_branches":
        conv_id = body.get("conversation_id", "")
        if not conv_id:
            flowfile.set_content(json.dumps({"error": "Missing conversation_id"}).encode())
            return [flowfile]
        branches = store.git_list_branches(conv_id)
        flowfile.set_content(json.dumps({
            "branches": branches,
            "current": store.git_current_branch(conv_id),
        }).encode())
        return [flowfile]

    if action == "conv_delete_branch":
        conv_id = body.get("conversation_id", "")
        branch_name = body.get("branch_name", "").strip()
        if not conv_id or not branch_name:
            flowfile.set_content(json.dumps({"error": "Missing conversation_id or branch_name"}).encode())
            return [flowfile]
        try:
            ok = store.git_delete_branch(conv_id, branch_name)
            flowfile.set_content(json.dumps({"ok": ok}).encode())
        except ValueError as e:
            flowfile.set_content(json.dumps({"error": str(e)}).encode())
        return [flowfile]

    if action == "conv_rollback":
        conv_id = body.get("conversation_id", "")
        commit_hash = body.get("commit_hash", "").strip()
        rewind_files = body.get("rewind_files", False)
        if not conv_id or not commit_hash:
            flowfile.set_content(json.dumps({"error": "Missing conversation_id or commit_hash"}).encode())
            return [flowfile]
        try:
            store._require_idle(conv_id)
        except RuntimeError as e:
            flowfile.set_content(json.dumps({"error": str(e)}).encode())
            return [flowfile]
        ok = store.git_rollback(conv_id, commit_hash)
        result = {"ok": ok, "commit": commit_hash}
        if rewind_files:
            from core.checkpoint import CheckpointManager
            checkpoints = CheckpointManager.list_checkpoints(conv_id)
            # Find closest checkpoint by git commit timestamp
            commit_log = store.git_log(conv_id, limit=100)
            target_ts = 0
            for c in commit_log:
                if c["hash"].startswith(commit_hash):
                    target_ts = c["timestamp"]
                    break
            target_cp = None
            for cp in checkpoints:
                if cp.get("timestamp", 0) <= target_ts:
                    target_cp = cp
            if target_cp:
                def _svc_resolver(svc_id):
                    try:
                        from core.service_registry import ServiceRegistry
                        return ServiceRegistry.get_instance().get_live_instance("global", "", svc_id)
                    except Exception:
                        return self._find_filesystem_service(user_id) if hasattr(self, '_find_filesystem_service') else None
                file_result = CheckpointManager.rewind_files(
                    conv_id, target_cp["id"], service_resolver=_svc_resolver)
                result["files"] = file_result
            else:
                result["files"] = {"error": "No matching checkpoint found for file rewind"}
        flowfile.set_content(json.dumps(result).encode())
        return [flowfile]

    if action == "conv_tag":
        conv_id = body.get("conversation_id", "")
        tag_name = body.get("tag_name", "").strip()
        message = body.get("message", "")
        if not conv_id or not tag_name:
            flowfile.set_content(json.dumps({"error": "Missing conversation_id or tag_name"}).encode())
            return [flowfile]
        ok = store.git_tag(conv_id, tag_name, message)
        flowfile.set_content(json.dumps({"ok": ok, "tag": tag_name}).encode())
        return [flowfile]

    if action == "conv_list_tags":
        conv_id = body.get("conversation_id", "")
        if not conv_id:
            flowfile.set_content(json.dumps({"error": "Missing conversation_id"}).encode())
            return [flowfile]
        tags = store.git_list_tags(conv_id)
        flowfile.set_content(json.dumps({"tags": tags}).encode())
        return [flowfile]

    if action == "conv_delete_tag":
        conv_id = body.get("conversation_id", "")
        tag_name = body.get("tag_name", "").strip()
        if not conv_id or not tag_name:
            flowfile.set_content(json.dumps({"error": "Missing conversation_id or tag_name"}).encode())
            return [flowfile]
        ok = store.git_delete_tag(conv_id, tag_name)
        flowfile.set_content(json.dumps({"ok": ok}).encode())
        return [flowfile]

    if action == "conv_export_pawflow":
        conv_id = body.get("conversation_id", "")
        include_filestore = bool(body.get("include_filestore", False))
        if not conv_id:
            flowfile.set_content(json.dumps({"error": "Missing conversation_id"}).encode())
            return [flowfile]
        import io, zipfile
        from pathlib import Path as _Path
        from core.file_store import FileStore
        from core.segmented_jsonl import SegmentedJsonl
        conv_dir = store._conv_dir(conv_id)
        if not conv_dir.is_dir():
            flowfile.set_content(json.dumps({"error": "Conversation not found"}).encode())
            return [flowfile]
        buf = io.BytesIO()
        manifest = {
            "format": "pawflow-conversation-archive",
            "version": 2,
            "mode": "full",
            "conversation_id": conv_id,
            "exported_at": time.time(),
            "includes": {
                "transcript": True,
                "shared_context": True,
                "agent_contexts": True,
                "buckets": True,
                "extras": True,
                "bindings": (conv_dir / "bindings.json").exists(),
                "filestore": include_filestore,
            },
            "filestore": {"included": False, "count": 0, "bytes": 0},
        }
        with zipfile.ZipFile(buf, 'w', zipfile.ZIP_DEFLATED) as zf:
            written = set()

            def _write_logical_jsonl(rel_path: _Path):
                log = SegmentedJsonl(conv_dir / rel_path)
                if not log.exists():
                    return
                data = "".join(
                    json.dumps(row, ensure_ascii=False) + "\n"
                    for row in log.iter_rows()
                )
                arcname = str(rel_path)
                zf.writestr(arcname, data)
                written.add(arcname)

            _write_logical_jsonl(_Path("transcript.jsonl"))
            _write_logical_jsonl(_Path("shared.jsonl"))
            for entry in sorted(conv_dir.iterdir()):
                if entry.is_dir() and entry.name not in (".git", "transcript", "shared", "summaries"):
                    _write_logical_jsonl(_Path(entry.name) / "context.jsonl")

            for f in sorted(conv_dir.rglob('*')):
                if not f.is_file() or '.git' in f.parts:
                    continue
                rel = f.relative_to(conv_dir)
                arcname = str(rel)
                if arcname in written:
                    continue
                parts = rel.parts
                if parts and parts[0] in ("filestore",):
                    continue
                if parts and parts[0] in ("transcript", "shared"):
                    continue
                if len(parts) >= 2 and parts[1] == "context":
                    continue
                zf.write(f, arcname)
            if include_filestore:
                manifest["filestore"] = _write_filestore_archive(zf, conv_id, user_id)
            zf.writestr("manifest.json", json.dumps(manifest, ensure_ascii=False, indent=2))
        filename = f"conversation_{conv_id[:8]}.pfconv.zip"
        fid = FileStore.instance().store(filename, buf.getvalue(),
            "application/zip", user_id=user_id, conversation_id=conv_id)
        flowfile.set_content(json.dumps({
            "ok": True, "url": f"/files/{fid}/{filename}", "filename": filename,
            "include_filestore": include_filestore,
            "filestore": manifest.get("filestore", {}),
        }).encode())
        return [flowfile]

    if action == "conv_export_claude_code":
        conv_id = body.get("conversation_id", "")
        if not conv_id:
            flowfile.set_content(json.dumps({"error": "Missing conversation_id"}).encode())
            return [flowfile]
        from core.file_store import FileStore
        msgs = store.load(conversation_id=conv_id, user_id=user_id)
        if not msgs:
            flowfile.set_content(json.dumps({"error": "Conversation empty"}).encode())
            return [flowfile]
        lines = []
        i = 0
        while i < len(msgs):
            m = msgs[i]
            role = m.get("role", "") if isinstance(m, dict) else getattr(m, "role", "")
            content = m.get("content", "") if isinstance(m, dict) else getattr(m, "content", "")
            if role == "user":
                lines.append(json.dumps({"type": "human", "message": {"role": "user", "content": content}}, ensure_ascii=False))
            elif role == "assistant":
                blocks = []
                if content:
                    blocks.append({"type": "text", "text": content})
                parent = m.get("msg_id", "") if isinstance(m, dict) else getattr(m, "msg_id", "")
                j = i + 1
                while j < len(msgs):
                    child = msgs[j]
                    child_role = child.get("role", "") if isinstance(child, dict) else getattr(child, "role", "")
                    if child_role not in ("thinking", "tool_call"):
                        break
                    child_parent = child.get("parent_message_id", "") if isinstance(child, dict) else getattr(child, "parent_message_id", "")
                    if child_parent and parent and child_parent != parent:
                        break
                    if child_role == "thinking":
                        blocks.append({"type": "thinking", "thinking": child.get("content", "")})
                    else:
                        blocks.append({
                            "type": "tool_use",
                            "id": child.get("tool_call_id", ""),
                            "name": child.get("tool_name") or child.get("name") or "",
                            "input": child.get("arguments", {}) or {},
                        })
                    j += 1
                lines.append(json.dumps({"type": "assistant", "message": {"role": "assistant", "content": blocks or content}}, ensure_ascii=False))
                i = j
                continue
            elif role == "tool":
                tcid = m.get("tool_call_id", "") if isinstance(m, dict) else getattr(m, "tool_call_id", "")
                lines.append(json.dumps({"type": "tool_result", "tool_use_id": tcid, "message": {"role": "user", "content": content}}, ensure_ascii=False))
            i += 1
        export = "\n".join(lines) + "\n"
        filename = f"conversation_{conv_id[:8]}.cc.jsonl"
        fid = FileStore.instance().store(filename, export.encode("utf-8"),
            "application/jsonl", user_id=user_id, conversation_id=conv_id)
        flowfile.set_content(json.dumps({
            "ok": True, "url": f"/files/{fid}/{filename}", "filename": filename,
        }).encode())
        return [flowfile]

    if action == "conv_compare_branches":
        conv_id = body.get("conversation_id", "")
        branch_a = body.get("branch_a", "").strip()
        branch_b = body.get("branch_b", "").strip()
        if not conv_id or not branch_a or not branch_b:
            flowfile.set_content(json.dumps({"error": "Missing parameters"}).encode())
            return [flowfile]
        result = store.git_compare_branches(conv_id, branch_a, branch_b)
        flowfile.set_content(json.dumps(result).encode())
        return [flowfile]

    if action == "conv_import_cleanup":
        import tempfile, shutil
        temp_id = body.get("temp_id", "")
        if temp_id:
            temp_dir = Path(tempfile.gettempdir()) / f"pf_import_{temp_id}"
            shutil.rmtree(temp_dir, ignore_errors=True)
        flowfile.set_content(json.dumps({"ok": True}).encode())
        return [flowfile]

    if action == "conv_import_analyze":
        import tempfile, uuid
        fmt = body.get("format", "")
        file_id = body.get("file_id", "")
        if not file_id or fmt not in ("pawflow", "claude_code"):
            flowfile.set_content(json.dumps({"error": "Missing file_id or invalid format"}).encode())
            return [flowfile]
        from core.file_store import FileStore
        fs = FileStore.instance()
        result = fs.get(file_id, user_id=user_id)
        if result is None:
            flowfile.set_content(json.dumps({"error": "Upload not found or expired"}).encode())
            return [flowfile]
        _fname, raw, _ct = result
        # Delete the uploaded file from FileStore — we copy raw to temp
        fs.delete(file_id)
        temp_id = uuid.uuid4().hex[:16]
        temp_dir = Path(tempfile.gettempdir()) / f"pf_import_{temp_id}"
        temp_dir.mkdir(parents=True, exist_ok=True)
        (temp_dir / "raw").write_bytes(raw)
        agents_found = []
        message_count = 0
        if fmt == "pawflow":
            import zipfile, io
            try:
                with zipfile.ZipFile(io.BytesIO(raw)) as zf:
                    names = zf.namelist()
                    transcript_name = "transcript.jsonl" if "transcript.jsonl" in names else "conversation/transcript.jsonl"
                    if transcript_name not in names:
                        flowfile.set_content(json.dumps({"error": "Not a valid PawFlow archive (missing transcript.jsonl)"}).encode())
                        return [flowfile]
                    manifest = _archive_manifest(zf)
                    # Count messages
                    for line in zf.read(transcript_name).decode("utf-8", errors="replace").splitlines():
                        if line.strip(): message_count += 1
                    # Extract agents from extras.json
                    extras_name = "extras.json" if "extras.json" in names else "conversation/extras.json"
                    if extras_name in names:
                        extras = json.loads(zf.read(extras_name))
                        conv_agents = extras.get("conv_agents", {})
                        for name, cfg in conv_agents.items():
                            agents_found.append({"name": name, "definition": cfg.get("definition", name)})
                    filestore_entries = []
                    if "filestore/index.json" in names:
                        try:
                            filestore_entries = json.loads(zf.read("filestore/index.json").decode("utf-8"))
                        except Exception:
                            filestore_entries = []
                    bucket_count = len([
                        n for n in names
                        if (n.startswith("summaries/_shared/") or n.startswith("conversation/summaries/_shared/"))
                        and n.endswith(".json") and not n.endswith("meta.json")
                    ])
                    agent_context_count = len([
                        n for n in names
                        if n.endswith("/context.jsonl")
                    ])
            except zipfile.BadZipFile:
                flowfile.set_content(json.dumps({"error": "Invalid zip file"}).encode())
                return [flowfile]
        elif fmt == "claude_code":
            text = raw.decode("utf-8", errors="replace")
            for line in text.splitlines():
                if line.strip(): message_count += 1
            agents_found = [{"name": "claude", "definition": "claude"}]
        flowfile.set_content(json.dumps({
            "ok": True, "temp_id": temp_id, "format": fmt,
            "agents": agents_found, "message_count": message_count,
            "full_archive": bool(fmt == "pawflow"),
            "manifest": manifest if fmt == "pawflow" else {},
            "bucket_count": bucket_count if fmt == "pawflow" else 0,
            "agent_context_count": agent_context_count if fmt == "pawflow" else 0,
            "filestore_count": len(filestore_entries) if fmt == "pawflow" else 0,
            "filestore_bytes": sum(int(x.get("size", 0) or 0) for x in filestore_entries) if fmt == "pawflow" else 0,
        }).encode())
        return [flowfile]

    if action == "conv_import_execute":
        import base64, tempfile, uuid as _uuid
        temp_id = body.get("temp_id", "")
        fmt = body.get("format", "")
        agent_mapping = body.get("agent_mapping", {})  # {import_name: {definition, params, llm_service}}
        title = body.get("title", "Imported conversation")
        relay_ids = body.get("relays", []) or []
        default_relay = body.get("default_relay", "") or ""
        restore_filestore = bool(body.get("restore_filestore", False))
        file_id_policy = body.get("file_id_policy", "preserve_or_remap") or "preserve_or_remap"
        # Resolve llm_service -> {provider, model, base_url, containerized}
        # once per agent mapping so imported assistant/tool messages carry
        # the right `source` metadata (name, llm_service, provider, model).
        # Without this the message meta bar is empty on imported convs.
        _src_by_agent = {}
        try:
            from core.service_registry import ServiceRegistry
            _reg = ServiceRegistry.get_instance()
        except Exception:
            _reg = None
        for _ag_name, _ag_cfg in (agent_mapping or {}).items():
            _svc_id = (_ag_cfg or {}).get("llm_service", "") or ""
            _src = {"type": "agent", "name": _ag_name,
                    "llm_service": _svc_id,
                    "provider": "", "model": "", "base_url": "",
                    "containerized": False}
            if _reg and _svc_id:
                try:
                    _sdef = _reg.resolve_definition(_svc_id, user_id=user_id)
                    if _sdef is not None:
                        _cfg = _sdef.config or {}
                        _src["provider"] = _cfg.get("provider", "") or ""
                        _src["model"] = _cfg.get("model", "") or ""
                        _src["base_url"] = _cfg.get("base_url", "") or _cfg.get("api_base", "") or ""
                        _src["containerized"] = bool(_cfg.get("docker_image", ""))
                except Exception:
                    logging.getLogger(__name__).debug("Ignored exception", exc_info=True)
            _src_by_agent[_ag_name] = _src
        if not temp_id:
            flowfile.set_content(json.dumps({"error": "Missing temp_id"}).encode())
            return [flowfile]
        temp_dir = Path(tempfile.gettempdir()) / f"pf_import_{temp_id}"
        raw_file = temp_dir / "raw"
        if not raw_file.exists():
            flowfile.set_content(json.dumps({"error": "Import data expired"}).encode())
            return [flowfile]
        raw = raw_file.read_bytes()
        cid = _uuid.uuid4().hex[:16] + _uuid.uuid4().hex[:16]
        conv_dir = store._store_dir / store._safe_name(user_id) / store._safe_name(cid)
        conv_dir.mkdir(parents=True, exist_ok=True)
        if fmt == "pawflow":
            import zipfile, io
            filestore_result = {"restored": 0, "bytes": 0, "file_id_map": {}}
            try:
                with zipfile.ZipFile(io.BytesIO(raw)) as zf:
                    _extract_conversation_members(zf, conv_dir)
                    filestore_result = _restore_filestore_archive(
                        zf, cid, user_id, restore_filestore, file_id_policy)
            except (zipfile.BadZipFile, ValueError) as e:
                import shutil as _shutil
                _shutil.rmtree(conv_dir, ignore_errors=True)
                flowfile.set_content(json.dumps({"error": str(e)}).encode())
                return [flowfile]
            store._cid_user[cid] = user_id  # required for segmented log helpers
            now_ts = time.time()
            # Update extras with new cid, user, and agent mapping
            extras_path = conv_dir / "extras.json"
            if extras_path.exists():
                extras = json.loads(extras_path.read_text(encoding="utf-8"))
            else:
                extras = {}
            extras["conversation_id"] = cid
            extras["user_id"] = user_id
            extras["title"] = title or extras.get("title", "Imported")
            extras["_meta_user_id"] = user_id
            extras["_meta_created_at"] = now_ts
            extras["_meta_updated_at"] = now_ts
            extras["_meta_status"] = extras.get("_meta_status", "idle")
            # Remap agents
            if agent_mapping:
                new_conv_agents = {}
                for imp_name, mapping in agent_mapping.items():
                    new_conv_agents[imp_name] = {
                        "definition": mapping.get("definition", imp_name),
                        "params": mapping.get("params", {"name": imp_name}),
                        "llm_service": mapping.get("llm_service", ""),
                    }
                extras["conv_agents"] = new_conv_agents
                if new_conv_agents:
                    extras["selectedAgent"] = list(new_conv_agents.keys())[0]
            extras_path.write_text(json.dumps(extras, ensure_ascii=False, indent=2), encoding="utf-8")
            _patch_conversation_files(
                conv_dir, cid, user_id,
                filestore_result.get("file_id_map", {}) or {})
        elif fmt == "claude_code":
            # Convert CC JSONL to PawFlow transcript.
            # Real Claude Code session files use type="user"/"assistant".
            # PawFlow's own CC export uses type="human"/"tool_result".
            # We handle both. Structured content blocks (text, tool_use,
            # tool_result) are expanded into proper tool_call / tool
            # messages so the UI renders them like native PawFlow traces
            # instead of empty bubbles.
            text = raw.decode("utf-8", errors="replace")
            transcript_lines = []
            import uuid as _u2
            import re as _re
            # CC transcripts have a single agent — pick the (only) entry from
            # the agent mapping so source/name propagate from the dialog.
            _cc_agent_name = next(iter(agent_mapping.keys()), "claude") if agent_mapping else "claude"
            _cc_source = _src_by_agent.get(_cc_agent_name, {
                "type": "agent", "name": _cc_agent_name,
                "llm_service": "", "provider": "", "model": "",
                "base_url": "", "containerized": False,
            })
            # Monotonic seq assigned to every emitted non-system message.
            # _deserialize_messages in agent_serialization.py refuses
            # entries without seq + ts, so missing seq would make the
            # imported conv unusable by any agent.
            _seq_counter = [0]
            # Raw messages collected for later shared.jsonl population.
            # We don't hand-roll the shared filter — ConversationStore
            # owns that logic (skip tools, skip context injections,
            # strip tool_calls, keep source/badges). See
            # ConversationStore.filter_for_shared.
            raw_msgs = []
            def _emit(obj):
                if obj.get("role") != "system" and not obj.get("seq"):
                    _seq_counter[0] += 1
                    obj["seq"] = _seq_counter[0]
                transcript_lines.append(json.dumps(obj, ensure_ascii=False))
                raw_msgs.append(dict(obj))
            # Claude CLI stuffs meta blocks into the user transcript:
            #   <local-command-caveat>...</local-command-caveat>
            #   <command-name>...</command-name><command-message>...</command-message><command-args>...</command-args>
            #   <local-command-stdout>...</local-command-stdout>
            # These are model-facing scaffolding, not user speech. Skip them.
            _CC_META_RE = _re.compile(
                r"^\s*(?:<local-command-(?:caveat|stdout|stderr)>.*?</local-command-(?:caveat|stdout|stderr)>"
                r"|(?:<command-(?:name|message|args)>.*?</command-(?:name|message|args)>\s*)+)\s*$",
                _re.DOTALL,
            )
            def _is_cc_meta(s):
                return isinstance(s, str) and bool(_CC_META_RE.match(s))
            def _stringify(c):
                if isinstance(c, str):
                    return c
                if isinstance(c, list):
                    parts = []
                    for b in c:
                        if not isinstance(b, dict):
                            continue
                        if b.get("type") == "text":
                            parts.append(b.get("text", ""))
                        elif b.get("type") == "tool_result":
                            parts.append(_stringify(b.get("content", "")))
                    return "\n".join(p for p in parts if p)
                return json.dumps(c, ensure_ascii=False)
            base_ts = time.time() - 1.0
            for idx, line in enumerate(text.splitlines()):
                line = line.strip()
                if not line:
                    continue
                try:
                    entry = json.loads(line)
                except json.JSONDecodeError:
                    continue
                msg_type = entry.get("type", "")
                if msg_type == "summary":
                    continue
                message = entry.get("message", {}) or {}
                content = message.get("content", "")
                raw_ts = entry.get("timestamp") or message.get("timestamp") or ""
                try:
                    if isinstance(raw_ts, (int, float)):
                        ts = float(raw_ts)
                    elif isinstance(raw_ts, str) and raw_ts:
                        ts = time.mktime(time.strptime(raw_ts[:19], "%Y-%m-%dT%H:%M:%S"))
                    else:
                        ts = base_ts + idx * 0.001
                except Exception:
                    ts = base_ts + idx * 0.001
                mid = _u2.uuid4().hex[:12]
                if msg_type in ("human", "user"):
                    # User messages may carry tool_results (CC encodes
                    # tool feedback as user role). Emit those as
                    # role=tool so the UI shows them in-line.
                    if isinstance(content, list):
                        tool_results = [b for b in content if isinstance(b, dict) and b.get("type") == "tool_result"]
                        text_parts = [b.get("text", "") for b in content if isinstance(b, dict) and b.get("type") == "text"]
                        user_text = "\n".join(p for p in text_parts if p)
                        if user_text and not _is_cc_meta(user_text):
                            _emit({"role": "user", "content": user_text, "msg_id": mid, "ts": ts})
                        for tr in tool_results:
                            _emit({
                                "role": "tool",
                                "content": _stringify(tr.get("content", "")),
                                "tool_call_id": tr.get("tool_use_id", ""),
                                "msg_id": _u2.uuid4().hex[:12], "ts": ts,
                            })
                        if not user_text and not tool_results:
                            continue
                    else:
                        if not (content or "").strip():
                            continue
                        if _is_cc_meta(content):
                            continue
                        _emit({"role": "user", "content": content, "msg_id": mid, "ts": ts})
                elif msg_type == "assistant":
                    # Split text blocks and tool_use blocks. Preserve
                    # tool calls so the chat UI renders them as proper
                    # tool-call blocks instead of empty assistant bubbles.
                    tool_calls = []
                    assistant_text = ""
                    if isinstance(content, list):
                        text_parts = []
                        for b in content:
                            if not isinstance(b, dict):
                                continue
                            if b.get("type") == "text":
                                text_parts.append(b.get("text", ""))
                            elif b.get("type") == "tool_use":
                                tool_calls.append({
                                    "id": b.get("id", ""),
                                    "name": b.get("name", ""),
                                    "arguments": b.get("input", {}) or {},
                                })
                        assistant_text = "\n".join(p for p in text_parts if p)
                    else:
                        assistant_text = content or ""
                    if not assistant_text and not tool_calls:
                        continue
                    obj = {"role": "assistant", "content": assistant_text,
                           "source": dict(_cc_source),
                           "msg_id": mid, "ts": ts}
                    if tool_calls:
                        obj["tool_calls"] = tool_calls
                    _emit(obj)
                elif msg_type == "tool_result":
                    _emit({
                        "role": "tool",
                        "content": _stringify(content),
                        "source": dict(_cc_source),
                        "tool_call_id": message.get("tool_use_id", "") or entry.get("tool_use_id", ""),
                        "msg_id": mid, "ts": ts,
                    })
            now_ts = time.time()
            store._cid_user[cid] = user_id  # required for _conv_dir lookups
            transcript_rows = []
            _tool_call_parents = {}
            for _msg in raw_msgs:
                for _row in store._canonical_message_rows(cid, _msg, _tool_call_parents):
                    transcript_rows.append(store._stamp_line(cid, _row))
            store._transcript_log(cid).replace_dicts(transcript_rows)
            # Shared context file: agents resuming the imported conv read
            # their LLM context from here (or from their own agent dir, which
            # falls back to shared). Without this file the "Shared" view
            # shows "diverged / no context" and the agent has no memory.
            # Use ConversationStore's filter + shared transform so imported
            # convs match native appends exactly: no tool/detail rows, agent
            # turns prefixed for the agent-neutral shared view.
            store._cid_user[cid] = user_id  # required for _conv_dir lookups
            shared_candidates = store.filter_for_shared(transcript_rows)
            shared_msgs = [store._transform_for_shared(m) for m in shared_candidates]
            if shared_msgs:
                store._append_shared_ctx(cid, shared_msgs)
            # Create minimal extras
            agent_name = list(agent_mapping.keys())[0] if agent_mapping else "claude"
            agent_cfg = agent_mapping.get(agent_name, {"definition": "claude", "params": {"name": agent_name}, "llm_service": ""})
            extras = {
                "conversation_id": cid,
                "user_id": user_id,
                "title": title,
                "selectedAgent": agent_name,
                "_meta_user_id": user_id,
                "_meta_created_at": now_ts,
                "_meta_updated_at": now_ts,
                "_meta_status": "idle",
                "conv_agents": {
                    agent_name: {
                        "definition": agent_cfg.get("definition", "claude"),
                        "params": agent_cfg.get("params", {"name": agent_name}),
                        "llm_service": agent_cfg.get("llm_service", ""),
                    }
                },
            }
            (conv_dir / "extras.json").write_text(json.dumps(extras, ensure_ascii=False, indent=2), encoding="utf-8")
        # Init git and register in the cache so list_conversations
        # picks up the new conversation immediately.
        store._cid_user[cid] = user_id
        store._git_init(cid)
        store._reload_cache(cid)
        summarizer_binding = _ensure_import_summarizer_binding(cid, user_id)
        # Relay bindings (mirrors create_conversation in agent_resource.py).
        if relay_ids:
            from core.relay_bindings import link_relay, set_default_relay
            for rid in relay_ids:
                link_relay(cid, rid)
            if default_relay and default_relay in relay_ids:
                set_default_relay(cid, default_relay)
        # Cleanup temp
        import shutil
        shutil.rmtree(temp_dir, ignore_errors=True)
        result = {"ok": True, "conversation_id": cid}
        if summarizer_binding:
            result["summarizer_binding"] = summarizer_binding
        if fmt == "pawflow":
            result["filestore_restored"] = filestore_result.get("restored", 0)
            result["filestore_bytes"] = filestore_result.get("bytes", 0)
            result["filestore_remapped"] = len(filestore_result.get("file_id_map", {}) or {})
        flowfile.set_content(json.dumps(result).encode())
        return [flowfile]

    return None
