"""Dreaming: scored memory promotion pipeline (Light / Deep / REM) + DREAMS.md.

This module replaces the legacy "always merge insights" step in
``memory_consolidator`` with a three-phase pipeline:

- **Light:** stage candidate entries from the consolidator's insights output
  (the "Suggested facts" and "Actionable Takeaways" sections), tagged with
  source labels and first-seen-at timestamps.
- **Deep:** score each candidate using weighted signals (frequency,
  recency decay, corroboration across sources, an optional utility-model tag,
  and a contradiction penalty against the current long-term memory corpus); only promote
  candidates that clear ``[dreaming].promote_threshold`` AND
  ``utils.memory_promotion.classify_memory_entry``. Respects
  ``memory_semantic_strictness`` from prefs.
- **REM:** extract a compact "patterns" block from the accepted dreams and
  append it to ``memory/insights.md`` (same format as the legacy consolidator).

Every cycle also appends a human-readable audit block to ``DREAMS.md`` in the
workspace root (rotated to ``[dreaming].dreams_file_max_kb``).

Dreaming is opt-in via ``[dreaming].enabled``. When disabled, callers should
keep using the legacy consolidator flow — this module is not auto-invoked.
"""

from __future__ import annotations

import logging
import math
import os
import re
import threading
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Callable

from utils.memory_promotion import (
    PromotionClass,
    PromotionStrictness,
    classify_memory_entry,
    strictness_from_semantic_pref,
)
from utils.path_safety import safe_resolve_path, secure_write_file

_log = logging.getLogger(__name__)

DREAMS_FILE = "DREAMS.md"
MEMORY_FILE = "MEMORY.md"
INSIGHTS_REL = "memory/insights.md"

_DEFAULT_MAX_CANDIDATES = 25
_RECENCY_HALF_LIFE_DAYS = 7.0
_WEIGHT_FREQUENCY = 2.0
_WEIGHT_RECENCY = 1.0
_WEIGHT_CORROBORATION = 1.5
_WEIGHT_UTILITY = 2.0
_WEIGHT_CONTRADICTION = -3.0

_DREAMS_LOCK = threading.Lock()

# Defense-in-depth: scrub any ``btw_<hex>`` thread-id references from the
# insights text before scoring. Upstream layers (chat_storage,
# session_search_db, active_memory, agent_plane, memory_consolidator) already
# filter these, but dreaming promotes into long-term memory so a belt-and-suspenders
# check here prevents a single upstream regression from training long-term
# memory on ephemeral /btw content.
_BTW_THREAD_TOKEN_RE = re.compile(r"\bbtw_[0-9a-f]{6,}\b", re.IGNORECASE)


def _scrub_btw_thread_refs(text: str) -> tuple[str, int]:
    if not text or "btw_" not in text:
        return text or "", 0
    dropped = 0
    out: list[str] = []
    for line in text.splitlines():
        if _BTW_THREAD_TOKEN_RE.search(line):
            dropped += 1
            continue
        out.append(line)
    return ("\n".join(out) + ("\n" if text.endswith("\n") else "")), dropped

_SECTION_LABEL_WEIGHT = {
    "recurring patterns": 2,
    "connections & links": 1,
    "superseded / contradicted info": 1,
    "actionable takeaways": 3,
    "suggested facts (not auto-merged)": 1,
}


@dataclass
class DreamCandidate:
    kind: str
    text: str
    sources: list[str] = field(default_factory=list)
    first_seen_at: str = ""
    score: float = 0.0
    tag: str = ""
    rationale: list[str] = field(default_factory=list)
    promoted: bool = False
    rejected_reason: str = ""


@dataclass
class DreamCycleResult:
    candidates: list[DreamCandidate]
    promoted: list[DreamCandidate]
    rejected: list[DreamCandidate]
    timestamp: str
    phase_errors: list[str] = field(default_factory=list)


# ---------------------------------------------------------------------------
# Light phase: stage candidates from the insights markdown.
# ---------------------------------------------------------------------------

_HEADING_RE = re.compile(r"^##\s+(.+?)\s*$", re.MULTILINE)


def _split_sections(text: str) -> dict[str, str]:
    sections: dict[str, str] = {}
    matches = list(_HEADING_RE.finditer(text or ""))
    for idx, m in enumerate(matches):
        heading = (m.group(1) or "").strip()
        start = m.end()
        end = matches[idx + 1].start() if idx + 1 < len(matches) else len(text)
        body = (text[start:end] or "").strip()
        sections[heading.lower()] = body
    return sections


def _iter_bullets(body: str) -> list[str]:
    items: list[str] = []
    current: list[str] = []
    for raw_line in (body or "").splitlines():
        line = raw_line.rstrip()
        if re.match(r"^\s*[-*]\s+", line):
            if current:
                items.append(" ".join(current).strip())
                current = []
            current.append(re.sub(r"^\s*[-*]\s+", "", line).strip())
        elif re.match(r"^\s+", line) and current:
            current.append(line.strip())
        else:
            if current:
                items.append(" ".join(current).strip())
                current = []
    if current:
        items.append(" ".join(current).strip())
    cleaned: list[str] = []
    for item in items:
        stripped = item.strip("- *").strip()
        if stripped and len(stripped) >= 8:
            cleaned.append(stripped)
    return cleaned


class LightPhase:
    """Stage dream candidates from consolidator output."""

    @staticmethod
    def stage_candidates(
        insights_text: str,
        *,
        max_candidates: int = _DEFAULT_MAX_CANDIDATES,
        now: datetime | None = None,
    ) -> list[DreamCandidate]:
        ts = (now or datetime.now(timezone.utc)).isoformat().replace("+00:00", "Z")
        sections = _split_sections(insights_text)
        candidates: list[DreamCandidate] = []
        seen_texts: dict[str, DreamCandidate] = {}
        for heading_lc, body in sections.items():
            if heading_lc.startswith("knowledge corpus hints"):
                continue
            kind = heading_lc
            for bullet in _iter_bullets(body):
                key = re.sub(r"\W+", "", bullet.lower())[:120]
                if not key:
                    continue
                if key in seen_texts:
                    existing = seen_texts[key]
                    if kind not in existing.sources:
                        existing.sources.append(kind)
                    continue
                cand = DreamCandidate(
                    kind=kind,
                    text=bullet,
                    sources=[kind],
                    first_seen_at=ts,
                )
                seen_texts[key] = cand
                candidates.append(cand)
                if len(candidates) >= max_candidates:
                    return candidates
        return candidates


# ---------------------------------------------------------------------------
# Deep phase: score and gate promotion.
# ---------------------------------------------------------------------------

def _recency_weight(first_seen_at: str, *, now: datetime) -> float:
    if not first_seen_at:
        return 0.5
    try:
        seen = datetime.fromisoformat(first_seen_at.replace("Z", "+00:00"))
        if seen.tzinfo is None:
            seen = seen.replace(tzinfo=timezone.utc)
    except Exception:
        return 0.5
    age_days = max(0.0, (now - seen).total_seconds() / 86_400.0)
    return math.pow(0.5, age_days / _RECENCY_HALF_LIFE_DAYS)


def _frequency_weight(candidate: DreamCandidate) -> float:
    weight = 1.0
    for src in candidate.sources:
        weight += _SECTION_LABEL_WEIGHT.get(src.lower(), 1) * 0.35
    return min(weight, 4.0)


def _corroboration_weight(candidate: DreamCandidate) -> float:
    distinct = {s.lower() for s in candidate.sources}
    if len(distinct) >= 3:
        return 1.0
    if len(distinct) == 2:
        return 0.6
    return 0.2


def _contradicts_memory(text: str, memory_text: str) -> bool:
    if not memory_text or not text:
        return False
    norm = re.sub(r"\W+", " ", text.lower())
    tokens = [t for t in norm.split() if len(t) > 3]
    if not tokens:
        return False
    memory_lower = memory_text.lower()
    negation_phrases = (" not ", " never ", " no longer ", " stopped ", " was ", " used to ")
    for phrase in negation_phrases:
        if phrase.strip() in norm:
            anchor = next((t for t in tokens if t in memory_lower), None)
            if anchor:
                return True
    return False


def _utility_tag(
    text: str,
    *,
    enabled: bool,
    on_log: Callable[[str, str], None] | None = None,
) -> str:
    if not enabled or not text:
        return ""
    try:
        from utility_model import is_available as util_available, utility_complete
    except Exception:
        return ""
    try:
        if not util_available():
            return ""
    except Exception:
        return ""
    messages = [
        {
            "role": "system",
            "content": (
                "Classify the following candidate memory entry. "
                "Return exactly one word from: durable_fact, ephemeral, speculative. "
                "durable_fact = a stable fact about the user, their relationships, or their preferences. "
                "ephemeral = a transient state (today's weather, current task). "
                "speculative = a guess or inference that lacks direct evidence."
            ),
        },
        {"role": "user", "content": text[:600]},
    ]
    try:
        raw = utility_complete(
            messages,
            max_tokens=8,
            temperature=0.1,
            activity_operation="memory_refresh",
        )
    except Exception:
        return ""
    if not raw:
        return ""
    token = re.split(r"[^a-z_]+", raw.strip().lower())
    for t in token:
        if t in {"durable_fact", "ephemeral", "speculative"}:
            return t
    return ""


def _utility_weight(tag: str) -> float:
    if tag == "durable_fact":
        return 1.0
    if tag == "ephemeral":
        return -0.3
    if tag == "speculative":
        return -0.5
    return 0.0


class DeepPhase:
    @staticmethod
    def score(
        candidate: DreamCandidate,
        *,
        memory_text: str,
        utility_tag_enabled: bool,
        now: datetime | None = None,
        on_log: Callable[[str, str], None] | None = None,
    ) -> DreamCandidate:
        now = now or datetime.now(timezone.utc)
        freq = _frequency_weight(candidate)
        recency = _recency_weight(candidate.first_seen_at, now=now)
        corrob = _corroboration_weight(candidate)
        tag = _utility_tag(candidate.text, enabled=utility_tag_enabled, on_log=on_log)
        candidate.tag = tag
        utility = _utility_weight(tag)
        contradiction = 1.0 if _contradicts_memory(candidate.text, memory_text) else 0.0
        score = (
            _WEIGHT_FREQUENCY * freq / 4.0
            + _WEIGHT_RECENCY * recency
            + _WEIGHT_CORROBORATION * corrob
            + _WEIGHT_UTILITY * utility
            + _WEIGHT_CONTRADICTION * contradiction
        )
        max_possible = (
            _WEIGHT_FREQUENCY + _WEIGHT_RECENCY + _WEIGHT_CORROBORATION + _WEIGHT_UTILITY
        )
        normalized = max(-1.0, min(1.0, score / max_possible))
        scaled = max(0.0, (normalized + 1.0) / 2.0)
        candidate.score = round(scaled, 4)
        candidate.rationale = [
            f"freq={round(freq, 2)}",
            f"recency={round(recency, 2)}",
            f"corroboration={round(corrob, 2)}",
            f"tag={tag or '-'}",
            f"contradiction={int(contradiction)}",
        ]
        return candidate

    @staticmethod
    def promote(
        candidate: DreamCandidate,
        *,
        threshold: float,
        strictness: PromotionStrictness,
    ) -> tuple[bool, str]:
        if candidate.score < threshold:
            return False, f"below_threshold({candidate.score:.2f}<{threshold:.2f})"
        decision = classify_memory_entry(candidate.text)
        if decision.classification == PromotionClass.BLOCKED:
            return False, f"blocked:{decision.reason}"
        if (
            strictness == PromotionStrictness.ENFORCE
            and decision.classification == PromotionClass.DISCOURAGED
        ):
            return False, f"discouraged_enforced:{decision.reason}"
        return True, decision.reason


# ---------------------------------------------------------------------------
# REM phase: pattern extraction.
# ---------------------------------------------------------------------------

class REMPhase:
    @staticmethod
    def extract_patterns(accepted: list[DreamCandidate]) -> str:
        if not accepted:
            return ""
        by_section: dict[str, list[str]] = {}
        for cand in accepted:
            src = (cand.sources[0] if cand.sources else cand.kind or "").strip().lower()
            label = "Actionable Takeaways" if "action" in src else "Recurring Patterns"
            by_section.setdefault(label, []).append(cand.text)
        blocks: list[str] = []
        for label in ("Recurring Patterns", "Actionable Takeaways"):
            items = by_section.get(label)
            if not items:
                continue
            blocks.append(f"## {label}")
            for item in items:
                blocks.append(f"- {item}")
            blocks.append("")
        return "\n".join(blocks).rstrip() + "\n"


# ---------------------------------------------------------------------------
# DREAMS.md audit log
# ---------------------------------------------------------------------------


def _read_text(path: str, max_chars: int = 200_000) -> str:
    if not os.path.isfile(path):
        return ""
    try:
        with open(path, "r", encoding="utf-8") as f:
            return f.read(max_chars)
    except OSError:
        return ""


def write_dreams_md(
    workspace: str,
    result: DreamCycleResult,
    *,
    max_bytes: int = 256 * 1024,
) -> str | None:
    """Append a human-readable audit block for this cycle to <workspace>/DREAMS.md."""
    resolved, err = safe_resolve_path(workspace, DREAMS_FILE, allow_write=True, create_parents=True)
    if err or not resolved:
        _log.debug("dreaming: cannot resolve DREAMS.md: %s", err)
        return None
    lines: list[str] = []
    lines.append(f"\n---\n\n## Dream Cycle — {result.timestamp}\n")
    if result.phase_errors:
        lines.append("### Errors")
        for e in result.phase_errors:
            lines.append(f"- {e}")
        lines.append("")
    lines.append(
        f"### Summary\n- candidates={len(result.candidates)} promoted={len(result.promoted)} "
        f"rejected={len(result.rejected)}\n"
    )
    if result.promoted:
        lines.append("### Promoted to long-term memory")
        for cand in result.promoted:
            lines.append(
                f"- score={cand.score:.2f} tag={cand.tag or '-'}  {cand.text[:220]}"
            )
            lines.append(f"    rationale: {', '.join(cand.rationale)}")
        lines.append("")
    if result.rejected:
        lines.append("### Rejected")
        for cand in result.rejected:
            lines.append(
                f"- score={cand.score:.2f} reason={cand.rejected_reason or 'unspecified'} "
                f"tag={cand.tag or '-'}  {cand.text[:180]}"
            )
        lines.append("")
    block = "\n".join(lines)
    with _DREAMS_LOCK:
        existing = _read_text(resolved, max_bytes * 4) or "# DREAMS.md — Dream cycle audit log\n"
        combined = existing.rstrip() + "\n" + block
        encoded = combined.encode("utf-8", errors="ignore")
        if len(encoded) > max_bytes:
            trimmed = encoded[-max_bytes:]
            cut_at = trimmed.find(b"\n---\n")
            if cut_at > 0:
                trimmed = trimmed[cut_at:]
            combined = "# DREAMS.md — Dream cycle audit log (rotated)\n" + trimmed.decode(
                "utf-8", errors="ignore"
            )
        try:
            secure_write_file(resolved, combined, mode=0o600)
        except Exception as e:
            _log.debug("dreaming: failed to write DREAMS.md: %s", e)
            return None
    return resolved


# ---------------------------------------------------------------------------
# Top-level cycle entry point
# ---------------------------------------------------------------------------


def _dreaming_memory_corpus(workspace: str, max_chars: int = 60_000) -> str:
    """Hub + shard text for contradiction scoring (matches sharded layout when active)."""
    try:
        from utils.memory_shard_layout import layout_is_sharded, list_shard_rel_paths
    except Exception:
        memory_path, _err = safe_resolve_path(workspace, MEMORY_FILE, allow_write=False)
        return _read_text(memory_path or "", max_chars)

    if not layout_is_sharded(workspace):
        memory_path, _err = safe_resolve_path(workspace, MEMORY_FILE, allow_write=False)
        return _read_text(memory_path or "", max_chars)

    parts: list[str] = []
    hub_path, _ = safe_resolve_path(workspace, MEMORY_FILE, allow_write=False)
    hub = _read_text(hub_path or "", min(8000, max_chars))
    parts.append(hub)
    used = len(hub)
    for rel in list_shard_rel_paths(workspace):
        if used >= max_chars - 20:
            break
        p, _e = safe_resolve_path(workspace, rel, allow_write=False)
        take = min(12_000, max_chars - used)
        chunk = _read_text(p or "", take)
        parts.append(f"\n--- {rel} ---\n{chunk}")
        used += len(parts[-1])
    return "\n".join(parts)[:max_chars]


def _append_to_memory_md(workspace: str, entries: list[DreamCandidate]) -> int:
    """Promote dream candidates: personal signal → User Model proposal; project → MEMORY."""
    if not entries:
        return 0
    from utils.user_model import BackgroundProposalBudget, propose_background_fact

    budget = BackgroundProposalBudget()
    memory_entries: list[DreamCandidate] = []
    routed = 0
    for cand in entries:
        text = (cand.text or "").strip()
        if not text:
            continue
        outcome = propose_background_fact(
            workspace,
            text,
            source="dreaming",
            rationale_prefix="Dream cycle promoted candidate",
            budget=budget,
        )
        if outcome == "memory":
            memory_entries.append(cand)
        elif outcome == "proposal":
            routed += 1
        else:
            _log.info("dreaming: skipped personal candidate (dedupe/cap/safety): %r", text[:120])

    if not memory_entries:
        return routed

    now = datetime.now()
    blocks: list[str] = []
    for cand in memory_entries:
        ts = now.strftime("%Y-%m-%d %H:%M")
        blocks.append(f"## {ts}  (dream-promoted, score={cand.score:.2f})\n{cand.text.strip()}\n")
    block = "\n".join(blocks)
    try:
        from utils.memory_shard_layout import ensure_shard_file, layout_is_sharded, regenerate_memory_hub_toc
    except Exception:
        layout_is_sharded = lambda _w: False  # type: ignore[misc]

    if layout_is_sharded(workspace):
        abs_path, serr = ensure_shard_file(workspace, "general")
        if serr or not abs_path:
            _log.debug("dreaming: could not open general shard: %s", serr)
            return routed
        existing = _read_text(abs_path, 200_000)
        if existing.strip():
            new_content = existing.rstrip() + "\n\n" + block.strip() + "\n"
        else:
            new_content = block.strip() + "\n"
        try:
            secure_write_file(abs_path, new_content, mode=0o600)
        except Exception as e:
            _log.debug("dreaming: append_to general shard failed: %s", e)
            return routed
        toc_err = regenerate_memory_hub_toc(workspace)
        if toc_err:
            _log.debug("dreaming: TOC refresh: %s", toc_err)
        return routed + len(memory_entries)

    resolved, err = safe_resolve_path(workspace, MEMORY_FILE, allow_write=True, create_parents=True)
    if err or not resolved:
        return routed
    existing = _read_text(resolved, 200_000)
    if existing.strip():
        new_content = existing.rstrip() + "\n\n" + block.strip() + "\n"
    else:
        new_content = block.strip() + "\n"
    try:
        secure_write_file(resolved, new_content, mode=0o600)
    except Exception as e:
        _log.debug("dreaming: append_to_memory failed: %s", e)
        return routed
    return routed + len(memory_entries)


def _write_insights_rem(workspace: str, rem_text: str) -> str | None:
    """Append REM pattern block to ``memory/insights.md`` (legacy consolidator format)."""
    if not rem_text:
        return None
    resolved, err = safe_resolve_path(workspace, INSIGHTS_REL, allow_write=True, create_parents=True)
    if err or not resolved:
        return None
    now = datetime.now().strftime("%Y-%m-%d %H:%M")
    block = f"\n---\n\n## Consolidation — {now} (dream REM)\n\n{rem_text}\n"
    existing = _read_text(resolved, 500_000)
    if existing.strip():
        combined = existing.rstrip() + "\n" + block
    else:
        combined = "# Memory Insights (auto-consolidated)\n" + block
    try:
        secure_write_file(resolved, combined, mode=0o600)
    except Exception as e:
        _log.debug("dreaming: write REM insights failed: %s", e)
        return None
    return resolved


def run_cycle(
    workspace: str,
    insights_text: str,
    *,
    promote_threshold: float,
    utility_tag_enabled: bool = True,
    max_candidates: int = _DEFAULT_MAX_CANDIDATES,
    semantic_strictness: str | None = None,
    on_log: Callable[[str, str], None] | None = None,
    now: datetime | None = None,
) -> DreamCycleResult:
    """Run Light → Deep → REM over a consolidator insights text.

    Returns a ``DreamCycleResult`` and always appends an audit block to
    ``DREAMS.md`` (best-effort).
    """
    import time as _time
    _t0 = _time.monotonic()
    stamp = (now or datetime.now(timezone.utc)).isoformat().replace("+00:00", "Z")
    errors: list[str] = []
    insights_text, _btw_dropped = _scrub_btw_thread_refs(insights_text or "")
    if _btw_dropped and on_log:
        on_log(
            "WARNING",
            f"[dreaming] scrubbed {_btw_dropped} btw_* line(s) from insights "
            "input (defense-in-depth; upstream filters should have removed these)",
        )
    try:
        candidates = LightPhase.stage_candidates(
            insights_text, max_candidates=max_candidates, now=now
        )
    except Exception as e:
        errors.append(f"light_phase: {e}")
        candidates = []

    memory_text = _dreaming_memory_corpus(workspace, 60_000)

    strictness = strictness_from_semantic_pref(semantic_strictness)
    promoted: list[DreamCandidate] = []
    rejected: list[DreamCandidate] = []
    for cand in candidates:
        try:
            DeepPhase.score(
                cand,
                memory_text=memory_text,
                utility_tag_enabled=utility_tag_enabled,
                now=now,
                on_log=on_log,
            )
            ok, reason = DeepPhase.promote(
                cand, threshold=promote_threshold, strictness=strictness
            )
            if ok:
                cand.promoted = True
                promoted.append(cand)
            else:
                cand.rejected_reason = reason
                rejected.append(cand)
        except Exception as e:
            errors.append(f"deep_phase: {e}")
            cand.rejected_reason = f"error: {e}"
            rejected.append(cand)

    try:
        if promoted:
            _append_to_memory_md(workspace, promoted)
    except Exception as e:
        errors.append(f"memory_append: {e}")

    try:
        rem_text = REMPhase.extract_patterns(promoted)
        if rem_text:
            _write_insights_rem(workspace, rem_text)
    except Exception as e:
        errors.append(f"rem_phase: {e}")

    result = DreamCycleResult(
        candidates=candidates,
        promoted=promoted,
        rejected=rejected,
        timestamp=stamp,
        phase_errors=errors,
    )
    try:
        write_dreams_md(workspace, result)
    except Exception as e:
        errors.append(f"dreams_md: {e}")

    if on_log:
        on_log(
            "INFO",
            f"[dreaming] cycle candidates={len(candidates)} promoted={len(promoted)} "
            f"rejected={len(rejected)} errors={len(errors)}",
        )

    try:
        from utils.task_ledger import record_run as _ledger_record
        _ledger_record(
            kind="dreaming",
            task_name="dream_cycle",
            thread_id=None,
            status="error" if errors else "ok",
            duration_ms=int((_time.monotonic() - _t0) * 1000),
            error_summary=("; ".join(errors))[:200] if errors else None,
            meta={
                "candidates": len(candidates),
                "promoted": len(promoted),
                "rejected": len(rejected),
                "promote_threshold": float(promote_threshold),
                "semantic_strictness": str(semantic_strictness or ""),
            },
        )
    except Exception:
        pass
    return result
