"""Background distillation of cold chat threads into durable MEMORY.md entries."""
from __future__ import annotations

import hashlib
import json
import logging
import os
import time

_log = logging.getLogger(__name__)

_LAST_CYCLE_TS = 0.0
_MIN_CYCLE_INTERVAL_SEC = 300.0
_MARKER_REL = "memory/thread_distillations.jsonl"
_IDLE_MINUTES_DEFAULT = 30.0
_MIN_USER_MESSAGES_DEFAULT = 6
_MARKER_TTL_SEC = 86400.0

_BLOCKED_PREFIXES = (
    "mesh",
    "group",
    "peer",
    "btw",
    "synthetic",
    "heartbeat",
    "background_heartbeat",
    "drive_action",
    "drive_discovery_",
    "skill_eval_",
)


def _thread_eligible_for_distillation(thread_id: str) -> bool:
    tid = (thread_id or "").strip().lower()
    if not tid or tid in ("default", "__quick__"):
        return False
    if any(tid.startswith(p) for p in _BLOCKED_PREFIXES):
        return False
    try:
        from session_search_db import _thread_id_is_background

        if _thread_id_is_background(tid):
            return False
    except Exception:
        pass
    return True


def _marker_path(workspace: str) -> str:
    return os.path.join(os.path.abspath(os.path.expanduser(workspace)), _MARKER_REL)


def _recent_marker_ts(workspace: str, thread_id: str) -> float | None:
    path = _marker_path(workspace)
    if not os.path.isfile(path):
        return None
    latest: float | None = None
    try:
        with open(path, encoding="utf-8", errors="replace") as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                try:
                    row = json.loads(line)
                except json.JSONDecodeError:
                    continue
                if str(row.get("thread_id") or "") != thread_id:
                    continue
                ts = row.get("ts")
                if isinstance(ts, (int, float)):
                    latest = max(latest or 0.0, float(ts))
    except OSError:
        return None
    return latest


def _marker_is_fresh(ts: float | None) -> bool:
    if ts is None:
        return False
    return (time.time() - float(ts)) < _MARKER_TTL_SEC


def _append_marker(workspace: str, thread_id: str, entry_hash: str, *, outcome: str = "distilled") -> None:
    path = _marker_path(workspace)
    os.makedirs(os.path.dirname(path), exist_ok=True)
    row = {
        "thread_id": thread_id,
        "ts": time.time(),
        "entry_hash": entry_hash,
        "outcome": outcome,
    }
    with open(path, "a", encoding="utf-8") as f:
        f.write(json.dumps(row, separators=(",", ":")) + "\n")


def _record_distillation_attempt(workspace: str, thread_id: str, *, outcome: str) -> None:
    """Record a terminal distillation attempt so utility re-runs are not hammered."""
    digest = hashlib.sha256(f"{thread_id}:{outcome}".encode("utf-8")).hexdigest()[:16]
    _append_marker(workspace, thread_id, digest, outcome=outcome)


def _count_user_messages(thread_id: str) -> int:
    from chat_storage import get_messages

    msgs = get_messages(thread_id) or []
    return sum(1 for m in msgs if (m.get("role") or "").strip().lower() == "user")


def _recent_thread_candidates(limit: int = 40) -> list[str]:
    from chat_storage import _get_connection

    conn = _get_connection()
    try:
        rows = conn.execute(
            """
            SELECT thread_id, MAX(id) AS mx
            FROM messages
            WHERE LOWER(role) = 'user'
            GROUP BY thread_id
            ORDER BY mx DESC
            LIMIT ?
            """,
            (max(1, int(limit)),),
        ).fetchall()
    finally:
        conn.close()
    return [str(r[0]) for r in rows if r and r[0]]


def _append_memory_entry(workspace: str, entry: str) -> bool:
    from datetime import datetime as dt

    from agent_loop_config import get_memory_policy_prefs
    from memory_consolidator import _scrub_btw_thread_refs
    from tools.memory_tools import _read_memory_file, _touch_memory_fts, _workspace
    from utils.memory_promotion import (
        PromotionClass,
        PromotionStrictness,
        classify_memory_entry,
        strictness_from_semantic_pref,
    )
    from utils.path_safety import secure_write_file

    cleaned, _ = _scrub_btw_thread_refs(entry.strip())
    if not cleaned.strip():
        return False
    _policy_prefs = get_memory_policy_prefs()
    _strictness = strictness_from_semantic_pref(_policy_prefs.get("memory_semantic_strictness"))
    _decision = classify_memory_entry(cleaned)
    if _strictness == PromotionStrictness.ENFORCE and _decision.classification == PromotionClass.BLOCKED:
        _log.info("thread distillation blocked by promotion gate")
        return False
    if _strictness == PromotionStrictness.ENFORCE and _decision.classification == PromotionClass.DISCOURAGED:
        _log.info("thread distillation discouraged by promotion gate")
        return False

    ws = workspace or _workspace()
    if not ws:
        return False
    from utils.memory_shard_layout import ensure_shard_file, layout_is_sharded, regenerate_memory_hub_toc

    sharded = layout_is_sharded(ws)
    if sharded:
        abs_path, serr = ensure_shard_file(ws, "distillation")
        if serr or not abs_path:
            _log.warning("thread distillation shard unavailable: %s", serr)
            return False
        rel_label = "memory/shards/distillation.md"
    else:
        abs_path = os.path.join(os.path.abspath(os.path.expanduser(ws)), "MEMORY.md")
        rel_label = "MEMORY.md"
    now = dt.now()
    block = f"## {now.strftime('%Y-%m-%d %H:%M')}\n{cleaned.strip()}\n"
    existing = _read_memory_file(abs_path) if os.path.isfile(abs_path) else ""
    new_content = (existing.rstrip() + "\n\n" + block).strip() + "\n"
    try:
        secure_write_file(abs_path, new_content, mode=0o600)
    except OSError as exc:
        _log.warning("thread distillation write failed (%s): %s", rel_label, exc)
        return False
    notify_paths = [abs_path]
    if sharded:
        toc_err = regenerate_memory_hub_toc(ws)
        hub_p = os.path.join(os.path.abspath(os.path.expanduser(ws)), "MEMORY.md")
        if os.path.isfile(hub_p):
            notify_paths.append(hub_p)
        if toc_err:
            _log.warning("thread distillation TOC refresh failed: %s", toc_err)
    _touch_memory_fts(ws, *notify_paths)
    return True


def distill_thread(thread_id: str, *, workspace: str | None = None) -> dict[str, object]:
    """Distill one cold thread into MEMORY.md. Returns status dict."""
    from core import get_workspace_path
    from chat_storage import get_last_real_user_message_age_seconds
    from tools.session_search_tools import _thread_excerpt_head_tail

    ws = workspace or get_workspace_path() or ""
    if not ws:
        return {"thread_id": thread_id, "status": "skipped", "reason": "no_workspace"}
    if not _thread_eligible_for_distillation(thread_id):
        return {"thread_id": thread_id, "status": "skipped", "reason": "ineligible_thread"}

    age = get_last_real_user_message_age_seconds(thread_id)
    if age is None or age < _IDLE_MINUTES_DEFAULT * 60:
        return {"thread_id": thread_id, "status": "skipped", "reason": "thread_still_hot"}

    if _count_user_messages(thread_id) < _MIN_USER_MESSAGES_DEFAULT:
        return {"thread_id": thread_id, "status": "skipped", "reason": "too_few_messages"}

    marker_ts = _recent_marker_ts(ws, thread_id)
    if _marker_is_fresh(marker_ts):
        return {"thread_id": thread_id, "status": "skipped", "reason": "recently_distilled"}

    _date, excerpt = _thread_excerpt_head_tail(thread_id, max_chars=4000)
    if not excerpt.strip():
        return {"thread_id": thread_id, "status": "skipped", "reason": "empty_excerpt"}

    from memory_consolidator import _scrub_btw_thread_refs

    excerpt, _ = _scrub_btw_thread_refs(excerpt)
    prompt_messages = [
        {
            "role": "system",
            "content": (
                "Extract 3-5 durable user facts from the conversation excerpt. "
                "Output one bullet per fact. Skip ephemeral filler. No preamble."
            ),
        },
        {"role": "user", "content": f"Conversation excerpt:\n\n{excerpt[:4000]}"},
    ]
    try:
        from utility_model import utility_complete

        facts = (utility_complete(
            prompt_messages,
            max_tokens=400,
            temperature=0.2,
            activity_operation="thread_distillation",
        ) or "").strip()
    except Exception as exc:
        _record_distillation_attempt(ws, thread_id, outcome="utility_error")
        return {"thread_id": thread_id, "status": "error", "reason": str(exc)[:200]}

    facts, _ = _scrub_btw_thread_refs(facts)
    if not facts.strip():
        _record_distillation_attempt(ws, thread_id, outcome="empty_facts")
        return {"thread_id": thread_id, "status": "skipped", "reason": "empty_facts"}

    entry_hash = hashlib.sha256(facts.encode("utf-8")).hexdigest()[:16]
    if not _append_memory_entry(ws, facts):
        _record_distillation_attempt(ws, thread_id, outcome="write_rejected")
        return {"thread_id": thread_id, "status": "skipped", "reason": "write_rejected"}
    _append_marker(ws, thread_id, entry_hash, outcome="distilled")
    return {"thread_id": thread_id, "status": "distilled", "entry_hash": entry_hash}


def run_thread_distillation_cycle(*, max_threads: int = 2) -> dict[str, object]:
    """Scan recent threads and distill up to max_threads cold candidates."""
    global _LAST_CYCLE_TS
    now = time.time()
    if now - _LAST_CYCLE_TS < _MIN_CYCLE_INTERVAL_SEC:
        return {"status": "skipped", "reason": "rate_limited"}
    _LAST_CYCLE_TS = now

    from core import get_workspace_path

    ws = get_workspace_path() or ""
    if not ws:
        return {"status": "skipped", "reason": "no_workspace"}

    results: list[dict[str, object]] = []
    for tid in _recent_thread_candidates(limit=40):
        if len(results) >= max(1, int(max_threads)):
            break
        out = distill_thread(tid, workspace=ws)
        if out.get("status") == "distilled":
            results.append(out)
    return {"status": "ran", "distilled": len(results), "results": results}


def thread_distillation_pending_count(*, workspace: str | None = None) -> int:
    """Rough count of threads that look eligible but not yet distilled (for trace)."""
    from core import get_workspace_path
    from chat_storage import get_last_real_user_message_age_seconds

    ws = workspace or get_workspace_path() or ""
    if not ws:
        return 0
    pending = 0
    for tid in _recent_thread_candidates(limit=20):
        if not _thread_eligible_for_distillation(tid):
            continue
        age = get_last_real_user_message_age_seconds(tid)
        if age is None or age < _IDLE_MINUTES_DEFAULT * 60:
            continue
        if _count_user_messages(tid) < _MIN_USER_MESSAGES_DEFAULT:
            continue
        marker_ts = _recent_marker_ts(ws, tid)
        if _marker_is_fresh(marker_ts):
            continue
        pending += 1
    return pending
