"""In-memory Assistent debug chat sessions for the laptop sidecar.

Multi-turn AssistentChat with a reconstructed per-turn **trace** (patch, Exact
merge hints, compact_context sizes, system_layers). Sessions live only in the
debug process (dict + lock); not part of /snapshot.

Gaps (documented in every trace.gaps):
- compactContext / Exact merge / krea_profile are client-side in swarm-assistent;
  HTTP AssistentChat does not return them — we send a synthetic context_json and
  optionally probe AssistentGetConfig / ListModels.
- Tool hop names are only visible on AssistentChatWS (clear_stream/hop); HTTP
  returns the final reply + system_layers / civitai_results only.
- Park/Warm are Generate-path APIs (AssistentParkLlm / AssistentWarmLlm), not
  invoked by chat turns.
"""

from __future__ import annotations

import json
import threading
import time
import uuid
from dataclasses import dataclass, field
from typing import Any

from gpu_rent.config import Config
from gpu_rent.debug_assistent import (
    DEFAULT_CHAT_EVAL_MESSAGE,
    MAX_CHAT_EVAL_TIMEOUT,
    _clamp_chat_eval_timeout,
    _finish_chat_eval,
    _http_json,
    _resolve_chat_model,
    _session_id,
    _summarize_patch,
    _swarm_base,
    collect_assistent_deep,
    extract_assistent_patch,
)

SESSION_TTL_SEC = 45 * 60
MAX_SESSIONS = 32
MAX_MESSAGE_CHARS = 8_000
MAX_HISTORY_TURNS = 40  # user+assistant pairs roughly

EXACT_GENERATE_PARAM_KEYS = ("steps", "cfg", "sigma_shift")

_TRACE_GAPS = [
    "compactContext is client-side only — AssistentChat never returns it; "
    "debug API sends synthetic context_json and summarizes sizes",
    "Exact merge / recommended_params / krea_profile are reconstructed from "
    "request context + optional AssistentGetConfig.exact (not live Swarm UI)",
    "HTTP AssistentChat does not list tool hops; WS emits hop notices — "
    "infer ask:* from patch.ask only",
    "Park/Warm not run on chat turns (AssistentParkLlm/WarmLlm are Generate-path)",
    "Sqlite SaveChat is client-side; AssistentChat may still return reply when "
    "memory DLL is missing — soft error in trace.errors",
]


@dataclass
class DebugChatSession:
    debug_session_id: str
    swarm_session_id: str | None
    persona: str
    pack: str
    model: str | None
    preferred: str | None
    via: str
    messages: list[dict[str, str]] = field(default_factory=list)
    context: dict[str, Any] = field(default_factory=dict)
    skills: list[str] = field(default_factory=list)
    config_probe: dict[str, Any] | None = None
    created_at: float = field(default_factory=time.time)
    last_used: float = field(default_factory=time.time)
    turn_count: int = 0
    last_trace: dict[str, Any] | None = None
    warnings: list[str] = field(default_factory=list)

    def touch(self) -> None:
        self.last_used = time.time()

    def summary(self) -> dict[str, Any]:
        return {
            "debug_session_id": self.debug_session_id,
            "swarm_session_id": self.swarm_session_id,
            "persona": self.persona,
            "pack": self.pack,
            "model": self.model,
            "preferred": self.preferred,
            "via": self.via,
            "turn_count": self.turn_count,
            "message_count": len(self.messages),
            "skills": list(self.skills),
            "context_keys": sorted(self.context.keys())[:40],
            "created_at": self.created_at,
            "last_used": self.last_used,
            "ttl_sec": SESSION_TTL_SEC,
            "expires_in_sec": max(0, int(SESSION_TTL_SEC - (time.time() - self.last_used))),
            "has_config_probe": self.config_probe is not None,
            "warnings": list(self.warnings),
            "last_trace_ok": (self.last_trace or {}).get("ok"),
        }


class SessionStore:
    """Thread-safe in-memory sessions with TTL eviction."""

    def __init__(
        self,
        *,
        ttl_sec: float = SESSION_TTL_SEC,
        max_sessions: int = MAX_SESSIONS,
    ) -> None:
        self.ttl_sec = ttl_sec
        self.max_sessions = max_sessions
        self._lock = threading.Lock()
        self._sessions: dict[str, DebugChatSession] = {}

    def purge_expired(self) -> int:
        now = time.time()
        with self._lock:
            dead = [
                sid
                for sid, s in self._sessions.items()
                if now - s.last_used > self.ttl_sec
            ]
            for sid in dead:
                del self._sessions[sid]
            return len(dead)

    def put(self, session: DebugChatSession) -> None:
        with self._lock:
            self._evict_unlocked()
            while len(self._sessions) >= self.max_sessions and self._sessions:
                oldest = min(self._sessions.values(), key=lambda s: s.last_used)
                del self._sessions[oldest.debug_session_id]
            self._sessions[session.debug_session_id] = session

    def get(self, session_id: str) -> DebugChatSession | None:
        with self._lock:
            self._evict_unlocked()
            s = self._sessions.get(session_id)
            if s is None:
                return None
            if time.time() - s.last_used > self.ttl_sec:
                del self._sessions[session_id]
                return None
            s.touch()
            return s

    def delete(self, session_id: str) -> bool:
        with self._lock:
            return self._sessions.pop(session_id, None) is not None

    def list_ids(self) -> list[str]:
        with self._lock:
            self._evict_unlocked()
            return list(self._sessions.keys())

    def _evict_unlocked(self) -> None:
        now = time.time()
        dead = [
            sid
            for sid, s in self._sessions.items()
            if now - s.last_used > self.ttl_sec
        ]
        for sid in dead:
            del self._sessions[sid]


# Process-global store for the debug sidecar.
_STORE = SessionStore()


def get_session_store() -> SessionStore:
    return _STORE


def reset_session_store_for_tests() -> SessionStore:
    """Replace the global store (tests only)."""
    global _STORE
    _STORE = SessionStore()
    return _STORE


def clamp_message(text: str | None, *, max_chars: int = MAX_MESSAGE_CHARS) -> tuple[str, bool]:
    raw = (text or "").strip()
    if len(raw) <= max_chars:
        return raw, False
    return raw[:max_chars], True


def summarize_compact_context(context: dict[str, Any] | None) -> dict[str, Any]:
    """Best-effort compactContext summary (keys + sizes + truncated preview)."""
    ctx = context if isinstance(context, dict) else {}
    sizes: dict[str, int] = {}
    for k, v in ctx.items():
        try:
            sizes[k] = len(json.dumps(v, ensure_ascii=False))
        except (TypeError, ValueError):
            sizes[k] = len(str(v))
    blob = json.dumps(ctx, ensure_ascii=False)
    preview = blob if len(blob) <= 600 else blob[:600] + "…"
    return {
        "note": (
            "Synthetic / client-built context_json — Assistent HTTP API does not "
            "return compactContext"
        ),
        "keys": sorted(ctx.keys()),
        "sizes": sizes,
        "chars": len(blob),
        "token_ish": round(len(blob) / 4),  # rough chars/4
        "preview": preview,
    }


def resolve_exact_profile_defaults(
    exact: dict[str, Any] | None,
    *,
    profile_name: str | None = None,
) -> dict[str, Any]:
    """Mirror SA.session.resolveExactProfileDefaults."""
    exact = exact if isinstance(exact, dict) else {}
    gen = exact.get("generation") if isinstance(exact.get("generation"), dict) else {}
    profiles = exact.get("profiles") if isinstance(exact.get("profiles"), dict) else {}
    profile = profile_name or gen.get("profile") or "turbo"
    from_profile = profiles.get(profile) if isinstance(profiles.get(profile), dict) else {}
    return {
        "profile": profile,
        "steps": from_profile.get("steps", gen.get("steps")),
        "cfg": from_profile.get("cfg", gen.get("cfg")),
        "sigma_shift": from_profile.get("sigma_shift", gen.get("sigma_shift")),
    }


def analyze_exact_merge(
    patch: dict[str, Any] | None,
    *,
    recommended: dict[str, Any] | None = None,
    krea_profile: str | None = None,
    session_exact: dict[str, Any] | None = None,
    user_param_intent: bool = False,
    exact: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Simulate client mergeExactParamsForGenerate for diagnostics."""
    patch = patch if isinstance(patch, dict) else None
    acts = patch.get("actions") if patch else None
    wants_gen = bool(
        patch
        and (
            patch.get("generate") is True
            or (isinstance(acts, list) and "generate" in [str(a) for a in acts])
        )
    )
    defaults = resolve_exact_profile_defaults(exact, profile_name=krea_profile)
    if recommended and isinstance(recommended, dict):
        for k in EXACT_GENERATE_PARAM_KEYS:
            if defaults.get(k) is None and recommended.get(k) is not None:
                defaults[k] = recommended[k]
        if not krea_profile and recommended.get("profile"):
            defaults["profile"] = recommended.get("profile")

    present = {
        k: patch.get(k)
        for k in EXACT_GENERATE_PARAM_KEYS
        if patch and patch.get(k) is not None
    }
    missing = [k for k in EXACT_GENERATE_PARAM_KEYS if k not in present]
    would_fill: dict[str, Any] = {}
    would_force: dict[str, Any] = {}
    hints: list[str] = []

    if not wants_gen:
        return {
            "wants_generate": False,
            "patch_params": present,
            "missing_exact_keys": missing,
            "recommended": {
                k: defaults.get(k) for k in EXACT_GENERATE_PARAM_KEYS + ("profile",)
            },
            "would_fill": {},
            "would_force_to_exact": {},
            "hints": ["no generate in patch — Exact merge not applied"],
        }

    merged = dict(patch or {})
    for key in EXACT_GENERATE_PARAM_KEYS:
        if merged.get(key) is not None:
            if (
                not user_param_intent
                and defaults.get(key) is not None
                and str(merged[key]) != str(defaults[key])
            ):
                would_force[key] = {"from": merged[key], "to": defaults[key]}
                merged[key] = defaults[key]
                hints.append(
                    f"{key}={would_force[key]['from']} ≠ Exact {defaults['profile']} "
                    f"{defaults[key]} — client would force Exact (0.14.1)"
                )
            continue
        if user_param_intent and isinstance(session_exact, dict) and session_exact.get(key) is not None:
            would_fill[key] = session_exact[key]
            merged[key] = session_exact[key]
            hints.append(f"{key} omitted — would fill from session_exact")
            continue
        if defaults.get(key) is not None:
            would_fill[key] = defaults[key]
            merged[key] = defaults[key]
            hints.append(
                f"{key} omitted — client would fill Exact {defaults.get('profile')}={defaults[key]}"
            )

    return {
        "wants_generate": True,
        "patch_params": present,
        "missing_exact_keys": missing,
        "recommended": {
            k: defaults.get(k) for k in EXACT_GENERATE_PARAM_KEYS + ("profile",)
        },
        "would_fill": would_fill,
        "would_force_to_exact": would_force,
        "merged_preview": {
            k: merged.get(k) for k in EXACT_GENERATE_PARAM_KEYS if merged.get(k) is not None
        },
        "hints": hints,
    }


def _default_context(
    *,
    persona: str,
    pack: str,
    exact: dict[str, Any] | None = None,
    krea_profile: str | None = None,
    extra: dict[str, Any] | None = None,
) -> dict[str, Any]:
    profile = krea_profile or "turbo"
    defaults = resolve_exact_profile_defaults(exact, profile_name=profile)
    ctx: dict[str, Any] = {
        "session": True,
        "persona": persona,
        "pack": pack,
        "debug_eval": True,
        "has_vision_image": False,
        "images_in_request": False,
        "krea_profile": defaults.get("profile") or profile,
        "recommended_params": {
            "steps": defaults.get("steps"),
            "cfg": defaults.get("cfg"),
            "sigma_shift": defaults.get("sigma_shift"),
        },
        "architecture_ok": True,
    }
    if extra:
        ctx.update(extra)
    return ctx


def _compact_get_config(data: dict[str, Any]) -> dict[str, Any]:
    exact = data.get("exact") if isinstance(data.get("exact"), dict) else {}
    skills = data.get("skills") if isinstance(data.get("skills"), list) else []
    enabled = data.get("enabled_skills") if isinstance(data.get("enabled_skills"), list) else []
    profiles = exact.get("profiles") if isinstance(exact.get("profiles"), dict) else {}
    return {
        "persona": data.get("persona"),
        "default_persona": data.get("default_persona"),
        "skills": [
            (s.get("id") if isinstance(s, dict) else s) for s in skills[:30]
        ],
        "enabled_skills": [str(x) for x in enabled[:30]],
        "exact_generation": exact.get("generation")
        if isinstance(exact.get("generation"), dict)
        else None,
        "exact_profiles": {
            name: {
                k: (profiles[name] or {}).get(k)
                for k in EXACT_GENERATE_PARAM_KEYS
            }
            for name in list(profiles.keys())[:8]
            if isinstance(profiles.get(name), dict)
        },
        "keys": sorted(data.keys())[:40],
    }


def _probe_config(base: str, sid: str, persona: str) -> dict[str, Any]:
    ok, data, ms = _http_json(
        f"{base}/API/AssistentGetConfig",
        method="POST",
        body={"session_id": sid, "persona": persona},
        timeout=25.0,
    )
    out: dict[str, Any] = {"ok": False, "ms": round(ms, 1), "error": None}
    if not ok:
        out["error"] = str(data)[:300]
        return out
    if isinstance(data, dict) and data.get("error"):
        out["error"] = str(data.get("error"))[:300]
        return out
    if isinstance(data, dict):
        out["ok"] = True
        out["data"] = _compact_get_config(data)
        out["_exact"] = data.get("exact") if isinstance(data.get("exact"), dict) else {}
        return out
    out["error"] = f"unexpected GetConfig body: {str(data)[:200]}"
    return out


def build_chat_trace(
    *,
    ok: bool,
    timings_ms: dict[str, float],
    model: str | None,
    preferred: str | None,
    persona: str,
    pack: str,
    context: dict[str, Any],
    skills: list[str],
    config_probe: dict[str, Any] | None,
    reply: str | None,
    response: dict[str, Any] | None,
    errors: list[str],
    warnings: list[str],
    hints: list[str],
) -> dict[str, Any]:
    extracted = extract_assistent_patch(reply)
    patch = extracted.get("patch")
    exact = None
    enabled_skills: list[str] = []
    if isinstance(config_probe, dict):
        exact = config_probe.get("_exact")
        data = config_probe.get("data") or {}
        if isinstance(data.get("enabled_skills"), list):
            enabled_skills = [str(x) for x in data["enabled_skills"]]

    recommended = context.get("recommended_params")
    if not isinstance(recommended, dict):
        recommended = None
    exact_merge = analyze_exact_merge(
        patch,
        recommended=recommended,
        krea_profile=context.get("krea_profile")
        if isinstance(context.get("krea_profile"), str)
        else None,
        session_exact=context.get("session_exact")
        if isinstance(context.get("session_exact"), dict)
        else None,
        exact=exact if isinstance(exact, dict) else None,
    )

    ask = None
    if isinstance(patch, dict) and patch.get("ask") is not None:
        ask = patch.get("ask")

    soft_errors = list(errors)
    if isinstance(response, dict):
        blob = json.dumps(response, ensure_ascii=False)
        if "sqlite" in blob.lower() or "SaveChat" in blob:
            soft_errors.append("response mentions Sqlite/SaveChat — chat text may still be OK")

    system_layers = None
    if isinstance(response, dict) and isinstance(response.get("system_layers"), dict):
        system_layers = response.get("system_layers")

    raw_preview = None
    if isinstance(response, dict):
        raw = response.get("raw")
        if isinstance(raw, dict):
            raw_preview = {k: raw[k] for k in list(raw.keys())[:12]}
        elif raw is not None:
            raw_preview = str(raw)[:400]

    out_hints = list(hints) + list(exact_merge.get("hints") or [])
    out: dict[str, Any] = {
        "ok": ok,
        "timings_ms": timings_ms,
        "model": model,
        "preferred": preferred,
        "persona": persona,
        "pack": pack,
        "krea_profile": context.get("krea_profile"),
        "checkpoint": context.get("checkpoint"),
        "recommended_params": recommended
        or exact_merge.get("recommended"),
        "compact_context": summarize_compact_context(context),
        "skills": {
            "requested": list(skills),
            "enabled_from_config": enabled_skills,
            "note": "skills[] on AssistentChat selects overlays; catalog from GetConfig",
        },
        "hops": {
            "ask_from_patch": ask,
            "note": (
                "HTTP AssistentChat has no hop list; ask:settings/inventory run "
                "server-side when patch.ask requests them (final reply only)"
            ),
        },
        "system_chars": response.get("system_chars") if isinstance(response, dict) else None,
        "system_layers": system_layers,
        "prompt_eval_count": response.get("prompt_eval_count")
        if isinstance(response, dict)
        else None,
        "reply": reply,
        "reply_prose": extracted.get("prose"),
        "patch": _summarize_patch(patch),
        "exact_merge": exact_merge,
        "park_warm": {
            "invoked": False,
            "note": "AssistentParkLlm / AssistentWarmLlm are Generate-path; not called here",
        },
        "raw": raw_preview,
        "config_probe_ok": bool(config_probe and config_probe.get("ok")),
        "errors": soft_errors,
        "warnings": list(warnings),
        "hints": out_hints,
        "gaps": list(_TRACE_GAPS),
    }
    if isinstance(response, dict) and response.get("civitai_results"):
        out["civitai_results"] = response.get("civitai_results")
    return out


def create_debug_session(
    cfg: Config,
    *,
    persona: str | None = None,
    pack: str | None = None,
    model: str | None = None,
    context: dict[str, Any] | None = None,
    skills: list[str] | None = None,
    probe_config: bool = True,
) -> dict[str, Any]:
    """Start a debug session: GetNewSession + optional GetConfig probe."""
    store = get_session_store()
    store.purge_expired()
    persona_id = (persona or "").strip() or "neutral"
    pack_name = (pack or "").strip() or "ordinary"
    skill_ids = [str(s) for s in (skills or []) if s]
    warnings: list[str] = [
        "opt-in Assistent session — may load VRAM; may write sqlite chat if UI/API persists",
        "not included in /snapshot",
        f"TTL {SESSION_TTL_SEC // 60} min idle; max {MAX_SESSIONS} sessions; "
        f"message cap {MAX_MESSAGE_CHARS} chars",
    ]
    t0 = time.perf_counter()
    base, via = _swarm_base(cfg)
    ms_sess = 0.0
    swarm_sid: str | None = None
    err: str | None = None
    config_probe: dict[str, Any] | None = None
    exact: dict[str, Any] | None = None

    if base:
        swarm_sid, err, ms_sess = _session_id(base)
        if not swarm_sid:
            return {
                "ok": False,
                "error": f"GetNewSession failed: {err}",
                "via": via,
                "ms": round((time.perf_counter() - t0) * 1000, 1),
                "warnings": warnings,
            }
        chosen, preferred = _resolve_chat_model(cfg, base, swarm_sid, model=model)
        if probe_config:
            config_probe = _probe_config(base, swarm_sid, persona_id)
            if config_probe.get("_exact"):
                exact = config_probe["_exact"]
            if not config_probe.get("ok"):
                warnings.append(
                    f"AssistentGetConfig probe failed: {config_probe.get('error')}"
                )
    else:
        # No local tunnel — session still usable for message bookkeeping; chat uses SSH.
        via = "ssh"
        chosen, preferred = _resolve_chat_model(cfg, None, None, model=model)
        warnings.append(
            "no local Swarm tunnel — chat turns will use SSH :7801; "
            "GetNewSession deferred to first chat"
        )

    if not chosen and model:
        chosen = model.strip() or None

    ctx = _default_context(
        persona=persona_id,
        pack=pack_name,
        exact=exact,
        krea_profile=(context or {}).get("krea_profile") if context else None,
        extra=context if isinstance(context, dict) else None,
    )

    sid = str(uuid.uuid4())
    session = DebugChatSession(
        debug_session_id=sid,
        swarm_session_id=swarm_sid,
        persona=persona_id,
        pack=pack_name,
        model=chosen,
        preferred=preferred,
        via=via,
        context=ctx,
        skills=skill_ids,
        config_probe=config_probe,
        warnings=warnings,
    )
    store.put(session)
    return {
        "ok": True,
        "debug_session_id": sid,
        "swarm_session_id": swarm_sid,
        "persona": persona_id,
        "pack": pack_name,
        "model": chosen,
        "preferred": preferred,
        "via": via,
        "session_ms": round(ms_sess, 1),
        "ms": round((time.perf_counter() - t0) * 1000, 1),
        "context_keys": sorted(ctx.keys()),
        "skills": skill_ids,
        "config_probe": {
            "ok": bool(config_probe and config_probe.get("ok")),
            "ms": (config_probe or {}).get("ms"),
            "data": (config_probe or {}).get("data"),
            "error": (config_probe or {}).get("error"),
        }
        if config_probe
        else None,
        "warnings": warnings,
        "ttl_sec": SESSION_TTL_SEC,
        "gaps": list(_TRACE_GAPS),
        "session": session.summary(),
    }


def get_debug_session(session_id: str) -> dict[str, Any]:
    store = get_session_store()
    s = store.get(session_id)
    if s is None:
        return {"ok": False, "error": "session not found or expired"}
    return {
        "ok": True,
        "session": s.summary(),
        "messages": [
            {"role": m["role"], "content_chars": len(m.get("content") or "")}
            for m in s.messages
        ],
        "context": summarize_compact_context(s.context),
        "last_trace": s.last_trace,
        "gaps": list(_TRACE_GAPS),
    }


def delete_debug_session(session_id: str) -> dict[str, Any]:
    ok = get_session_store().delete(session_id)
    return {"ok": ok, "deleted": ok, "debug_session_id": session_id}


def chat_debug_session(
    cfg: Config,
    session_id: str,
    *,
    message: str | None = None,
    timeout: float | int | str | None = None,
    context: dict[str, Any] | None = None,
    pack: str | None = None,
    model: str | None = None,
    skills: list[str] | None = None,
    persona: str | None = None,
) -> dict[str, Any]:
    """Send one user turn; return reply + trace."""
    store = get_session_store()
    session = store.get(session_id)
    if session is None:
        return {"ok": False, "error": "session not found or expired"}

    text, truncated = clamp_message(message)
    if not text:
        return {"ok": False, "error": "message required", "debug_session_id": session_id}

    warnings = list(session.warnings)
    if truncated:
        warnings.append(f"message truncated to {MAX_MESSAGE_CHARS} chars")

    if pack and pack.strip():
        session.pack = pack.strip()
    if persona and persona.strip():
        session.persona = persona.strip()
    if model and str(model).strip():
        session.model = str(model).strip()
    if skills is not None:
        session.skills = [str(s) for s in skills if s]
    if isinstance(context, dict) and context:
        session.context.update(context)
        session.context["persona"] = session.persona
        session.context["pack"] = session.pack

    t_chat = _clamp_chat_eval_timeout(timeout)
    t0 = time.perf_counter()
    timings: dict[str, float] = {}

    # Trim history
    session.messages.append({"role": "user", "content": text})
    if len(session.messages) > MAX_HISTORY_TURNS * 2:
        session.messages = session.messages[-(MAX_HISTORY_TURNS * 2) :]

    base, via = _swarm_base(cfg)
    if via == "local":
        session.via = "local"

    errors: list[str] = []
    reply: str | None = None
    response: dict[str, Any] | None = None
    ms_call = 0.0
    ms_sess = 0.0
    ok = False

    if base:
        if not session.swarm_session_id:
            sid, err, ms_sess = _session_id(base)
            timings["session"] = round(ms_sess, 1)
            if not sid:
                session.messages.pop()  # rollback user msg
                return {
                    "ok": False,
                    "debug_session_id": session_id,
                    "error": f"GetNewSession failed: {err}",
                    "warnings": warnings,
                }
            session.swarm_session_id = sid
        else:
            timings["session"] = 0.0

        if not session.model:
            chosen, preferred = _resolve_chat_model(
                cfg, base, session.swarm_session_id, model=None
            )
            session.model = chosen
            session.preferred = preferred or session.preferred
        if not session.model:
            session.messages.pop()
            return {
                "ok": False,
                "debug_session_id": session_id,
                "error": "no model (AssistentListModels.preferred / default_chat)",
                "hints": ["GET /assistent/roles + /ollama — нет preferred chat model"],
                "warnings": warnings,
            }

        if session.config_probe is None:
            probe = _probe_config(base, session.swarm_session_id, session.persona)
            session.config_probe = probe
            timings["config_probe"] = float(probe.get("ms") or 0)
            if probe.get("_exact") and not session.context.get("recommended_params"):
                defaults = resolve_exact_profile_defaults(
                    probe["_exact"],
                    profile_name=session.context.get("krea_profile"),
                )
                session.context["recommended_params"] = {
                    "steps": defaults.get("steps"),
                    "cfg": defaults.get("cfg"),
                    "sigma_shift": defaults.get("sigma_shift"),
                }
                session.context.setdefault("krea_profile", defaults.get("profile"))

        payload = {
            "session_id": session.swarm_session_id,
            "baseUrl": "http://127.0.0.1:11434",
            "model": session.model,
            "pack": session.pack,
            "persona": session.persona,
            "includeBase": True,
            "messages": list(session.messages),
            "context_json": json.dumps(session.context, ensure_ascii=False),
            "skills": list(session.skills),
        }
        ok_http, data, ms_call = _http_json(
            f"{base}/API/AssistentChat",
            method="POST",
            body=payload,
            timeout=t_chat,
        )
        timings["chat"] = round(ms_call, 1)
        if not ok_http:
            errors.append(str(data)[:400])
            response = None
            ok = False
        elif isinstance(data, dict):
            response = data
            if data.get("error"):
                err_s = str(data.get("error"))[:400]
                # Soft-fail Sqlite: keep reply if present
                reply = data.get("reply")
                if reply is not None:
                    reply = str(reply)
                if reply:
                    errors.append(err_s)
                    warnings.append(
                        "AssistentChat returned error but also reply — treating as soft fail"
                    )
                    ok = True
                else:
                    errors.append(err_s)
                    ok = False
            else:
                reply = data.get("reply")
                if reply is not None:
                    reply = str(reply)
                ok = reply is not None
                if data.get("model"):
                    session.model = str(data.get("model"))
        else:
            errors.append(f"unexpected AssistentChat body: {str(data)[:200]}")
            ok = False
    else:
        # Full message history over SSH :7801 when local tunnel is down.
        session.via = "ssh"
        finish = _assistent_chat_via_ssh_messages(
            cfg,
            messages=list(session.messages),
            persona=session.persona,
            pack=session.pack,
            model=session.model,
            context=session.context,
            skills=session.skills,
            timeout=t_chat,
            swarm_session_id=session.swarm_session_id,
            t0=t0,
        )
        if finish.get("swarm_session_id"):
            session.swarm_session_id = finish["swarm_session_id"]
        if finish.get("model"):
            session.model = finish["model"]
        if finish.get("preferred"):
            session.preferred = finish["preferred"]
        timings["session"] = float(finish.get("session_ms") or 0)
        timings["chat"] = float(finish.get("chat_ms") or 0)
        reply = finish.get("reply")
        ok = bool(finish.get("ok") and reply is not None)
        errors.extend(finish.get("errors") or [])
        if finish.get("error") and finish["error"] not in errors:
            errors.append(str(finish["error"]))
        response = {
            "reply": reply,
            "system_chars": finish.get("system_chars"),
            "system_layers": finish.get("system_layers"),
            "prompt_eval_count": finish.get("prompt_eval_count"),
            "raw": finish.get("raw"),
            "model": finish.get("model"),
        }

    timings["total"] = round((time.perf_counter() - t0) * 1000, 1)

    if reply:
        session.messages.append({"role": "assistant", "content": reply})
        session.turn_count += 1
    else:
        # rollback user message on hard failure
        if session.messages and session.messages[-1].get("role") == "user":
            session.messages.pop()

    hints: list[str] = []
    if any("sqlite" in e.lower() for e in errors):
        hints.append(
            "Sqlite — Microsoft.Data.Sqlite рядом с extension DLL (seed-extensions ≥0.13.1)"
        )
    if any("timeout" in e.lower() or "timed out" in e.lower() for e in errors):
        hints.append(f"timeout {t_chat:.0f}s — увеличь timeout (cap {MAX_CHAT_EVAL_TIMEOUT:.0f})")

    trace = build_chat_trace(
        ok=ok,
        timings_ms=timings,
        model=session.model,
        preferred=session.preferred,
        persona=session.persona,
        pack=session.pack,
        context=session.context,
        skills=session.skills,
        config_probe=session.config_probe,
        reply=reply,
        response=response if isinstance(response, dict) else None,
        errors=errors,
        warnings=warnings,
        hints=hints,
    )
    session.last_trace = trace
    session.touch()

    return {
        "ok": ok,
        "debug_session_id": session_id,
        "swarm_session_id": session.swarm_session_id,
        "turn_count": session.turn_count,
        "message": text,
        "reply": reply,
        "trace": trace,
        "session": session.summary(),
        "timeout_sec": t_chat,
        "via": session.via,
        "error": errors[0] if errors and not ok else None,
        "errors": errors,
        "warnings": warnings,
        "hints": hints + list(trace.get("hints") or []),
    }


def _assistent_chat_via_ssh_messages(
    cfg: Config,
    *,
    messages: list[dict[str, str]],
    persona: str,
    pack: str,
    model: str | None,
    context: dict[str, Any],
    skills: list[str],
    timeout: float,
    swarm_session_id: str | None,
    t0: float,
) -> dict[str, Any]:
    """Multi-turn AssistentChat over SSH when local tunnel is down."""
    from gpu_rent import debug_checks
    from gpu_rent.state import load_state

    state = load_state()
    if not debug_checks.ssh_ready(cfg, state):
        return {
            "ok": False,
            "error": debug_checks.SSH_UNAVAILABLE,
            "errors": [debug_checks.SSH_UNAVAILABLE],
            "ms": round((time.perf_counter() - t0) * 1000, 1),
        }
    host = debug_checks.ssh_host(state)
    assert host is not None
    chosen = model
    preferred = None
    if not chosen:
        from gpu_rent.debug_assistent import collect_assistent_roles

        roles = collect_assistent_roles(cfg)
        preferred = (roles.get("roles") or {}).get("default_chat")
        if isinstance(preferred, str) and preferred.strip():
            chosen = preferred.strip()

    script = f'''
import json, urllib.request, time
MESSAGES = {json.dumps(messages, ensure_ascii=False)}
PERSONA = {json.dumps(persona, ensure_ascii=False)}
PACK = {json.dumps(pack, ensure_ascii=False)}
MODEL = {json.dumps(chosen or "", ensure_ascii=False)}
CONTEXT = {json.dumps(context, ensure_ascii=False)}
SKILLS = {json.dumps(skills, ensure_ascii=False)}
SID = {json.dumps(swarm_session_id or "", ensure_ascii=False)}
TIMEOUT = {float(timeout)}

def post(path, payload, timeout=TIMEOUT):
    t0 = time.time()
    req = urllib.request.Request(
        "http://127.0.0.1:7801" + path,
        data=json.dumps(payload).encode(),
        headers={{"Content-Type": "application/json"}},
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            raw = resp.read().decode("utf-8", "replace")
            ms = (time.time() - t0) * 1000
            try:
                data = json.loads(raw)
            except Exception:
                data = raw[:500]
            return True, data, ms
    except Exception as e:
        return False, str(e), (time.time() - t0) * 1000

ms_sess = 0
sid = SID.strip()
if not sid:
    ok, sess, ms_sess = post("/API/GetNewSession", {{}}, timeout=20)
    if not ok or not isinstance(sess, dict) or not sess.get("session_id"):
        print(json.dumps({{"ok": False, "error": sess, "session_ms": ms_sess, "via": "ssh"}}))
        raise SystemExit(0)
    sid = sess["session_id"]
preferred = None
model = MODEL.strip()
if not model:
    cok, mdata, _ = post("/API/AssistentListModels", {{
        "session_id": sid, "baseUrl": "http://127.0.0.1:11434"
    }}, timeout=25)
    if cok and isinstance(mdata, dict):
        preferred = mdata.get("preferred")
        model = (preferred or "").strip()
if not model:
    print(json.dumps({{
        "ok": False, "error": "no model", "session_ms": ms_sess, "via": "ssh",
        "preferred": preferred, "swarm_session_id": sid
    }}))
    raise SystemExit(0)
payload = {{
    "session_id": sid,
    "baseUrl": "http://127.0.0.1:11434",
    "model": model,
    "pack": PACK,
    "persona": PERSONA,
    "includeBase": True,
    "messages": MESSAGES,
    "context_json": json.dumps(CONTEXT, ensure_ascii=False),
    "skills": SKILLS,
}}
cok, data, cms = post("/API/AssistentChat", payload, timeout=TIMEOUT)
print(json.dumps({{
    "ok": cok, "data": data, "chat_ms": cms, "session_ms": ms_sess,
    "via": "ssh", "model": model, "preferred": preferred or model,
    "swarm_session_id": sid
}}, ensure_ascii=False))
'''
    try:
        from gpu_rent.ssh_ops import run_ssh

        out = run_ssh(
            cfg,
            host,
            "python3 - <<'PY'\n" + script + "\nPY",
            check=False,
            timeout=int(timeout) + 40,
        ).strip()
        remote = json.loads(out.splitlines()[-1])
    except Exception as exc:
        return {
            "ok": False,
            "error": str(exc)[:240],
            "errors": [str(exc)[:240]],
            "ms": round((time.perf_counter() - t0) * 1000, 1),
        }

    data = remote.get("data")
    ok = bool(remote.get("ok"))
    finished = _finish_chat_eval(
        ok=ok,
        data=data if ok or isinstance(data, dict) else remote.get("error") or data,
        ms_call=float(remote.get("chat_ms") or 0),
        ms_total=(time.perf_counter() - t0) * 1000,
        via="ssh",
        session_ms=float(remote.get("session_ms") or 0),
        message=messages[-1]["content"] if messages else "",
        persona=persona,
        pack=pack,
        model=remote.get("model") or chosen,
        preferred=remote.get("preferred") or preferred,
        hints=[],
        errors=[],
        timeout=timeout,
    )
    finished["swarm_session_id"] = remote.get("swarm_session_id")
    finished["chat_ms"] = remote.get("chat_ms")
    finished["session_ms"] = remote.get("session_ms")
    if isinstance(data, dict):
        for k in ("system_chars", "system_layers", "prompt_eval_count"):
            if k in data:
                finished[k] = data[k]
    return finished


def run_assistent_chat_eval_via_session(
    cfg: Config,
    *,
    message: str | None = None,
    persona: str | None = None,
    pack: str | None = None,
    model: str | None = None,
    timeout: float | int | str | None = None,
    context: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """One-shot convenience: create session → one chat → delete session."""
    text = (message or DEFAULT_CHAT_EVAL_MESSAGE).strip() or DEFAULT_CHAT_EVAL_MESSAGE
    created = create_debug_session(
        cfg,
        persona=persona,
        pack=pack,
        model=model,
        context=context,
        probe_config=True,
    )
    if not created.get("ok"):
        return {
            **created,
            "message": text,
            "persona": (persona or "").strip() or "neutral",
            "pack": (pack or "").strip() or "ordinary",
        }
    sid = created["debug_session_id"]
    try:
        result = chat_debug_session(
            cfg,
            sid,
            message=text,
            timeout=timeout,
            model=model,
            pack=pack,
            persona=persona,
        )
    finally:
        delete_debug_session(sid)

    # Flatten for backward-compatible chat-eval consumers + nested trace.
    trace = result.get("trace") or {}
    return {
        "ok": result.get("ok"),
        "via": result.get("via") or created.get("via"),
        "ms": (trace.get("timings_ms") or {}).get("total"),
        "chat_ms": (trace.get("timings_ms") or {}).get("chat"),
        "session_ms": created.get("session_ms")
        or (trace.get("timings_ms") or {}).get("session"),
        "timeout_sec": result.get("timeout_sec"),
        "message": text,
        "persona": result.get("session", {}).get("persona")
        or created.get("persona"),
        "pack": result.get("session", {}).get("pack") or created.get("pack"),
        "model": trace.get("model") or created.get("model"),
        "preferred": trace.get("preferred") or created.get("preferred"),
        "reply": result.get("reply"),
        "reply_prose": trace.get("reply_prose"),
        "patch": trace.get("patch"),
        "raw": trace.get("raw"),
        "system_chars": trace.get("system_chars"),
        "system_layers": trace.get("system_layers"),
        "prompt_eval_count": trace.get("prompt_eval_count"),
        "trace": trace,
        "errors": result.get("errors") or [],
        "hints": result.get("hints") or [],
        "warnings": result.get("warnings") or created.get("warnings") or [],
        "error": result.get("error"),
        "gaps": list(_TRACE_GAPS),
        "note": "chat-eval wraps POST /assistent/session + one /chat + DELETE",
    }


def collect_assistent_diagnose(
    cfg: Config,
    *,
    chat_smoke: bool = False,
    include_logs: bool = True,
) -> dict[str, Any]:
    """Full static+live Assistent health bundle (alias of deep collect)."""
    deep = collect_assistent_deep(
        cfg, chat_smoke=chat_smoke, include_logs=include_logs
    )
    deep["playbook"] = {
        **(deep.get("playbook") or {}),
        "session": (
            "POST /assistent/session → POST /assistent/session/{id}/chat "
            "(multi-turn + trace) → DELETE when done"
        ),
        "chat_eval": "POST /assistent/chat-eval (one-shot session wrapper)",
        "diagnose": "GET /assistent/diagnose (this bundle; also GET /assistent?logs=1)",
    }
    deep["session_store"] = {
        "active": len(get_session_store().list_ids()),
        "ttl_sec": SESSION_TTL_SEC,
        "max_sessions": MAX_SESSIONS,
        "max_message_chars": MAX_MESSAGE_CHARS,
    }
    deep["gaps"] = list(_TRACE_GAPS)
    return deep