# memory_consolidator.py — Background memory consolidation (inspired by Google's
# Always-On Memory Agent). Periodically reads MEMORY.md, recent summaries, and
# SESSION-STATE.md, then uses an LLM to find patterns, connections, and insights
# across entries. Results are written to memory/insights.md and optionally merged
# back into MEMORY.md as a condensed "consolidated" block.
from __future__ import annotations

import os
import re
import threading
from datetime import datetime, timedelta
from typing import Callable

from utils.llm_output_sanitize import strip_markdown_fences, strip_think_tags, strip_untagged_cot
from utils.assistant_visibility_placeholders import is_visibility_placeholder
from agent_loop_config import get_memory_policy_prefs

_MAX_INPUT_CHARS = 30_000
_MAX_INSIGHT_CHARS = 4000
_INSIGHTS_WRITE_LOCK = threading.Lock()


def _notify_memory_fts_paths(workspace: str, *paths: str) -> None:
    try:
        from core.config import get_memory_config
        from utils.memory_fts_index import notify_memory_paths_changed

        uniq = [p for p in dict.fromkeys(paths) if p]
        if not uniq:
            return
        notify_memory_paths_changed(workspace, uniq, mem=get_memory_config())
    except Exception:
        pass


# Defense-in-depth: even though chat_storage, session_search_db, active_memory,
# and agent_plane already drop thread IDs prefixed with ``btw_``, the
# consolidator scrubs its aggregated context one more time before handing it
# to the LLM. Any line mentioning a ``btw_<hex>`` token is removed and a
# counter is logged so upstream leaks become visible instead of silently
# training long-term memory.
_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]:
    """Drop any line referencing a ``btw_<hex>`` thread id. Returns (text, n)."""
    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

_CONSOLIDATION_SYSTEM = (
    "You are a memory consolidation engine for a personal AI assistant. "
    "Your job is to analyze memory entries, daily summaries, and session state, "
    "then produce structured insights about the user.\n\n"
    "Curation stance (important):\n"
    "- Prefer **sparse, load-bearing facts**: stable preferences, constraints, goals, relationships, and recurring patterns.\n"
    "- Default **omit** ephemeral chat filler, one-off moods without a pattern, and details that will not matter in a future session.\n"
    "- When something matters, ask implicitly: *would omitting this mislead the assistant later?* If no, leave it out.\n"
    "- Prefer **one line per bullet**; compress rambling source text into a single durable claim.\n"
    "- Put uncertain inferences only under **Suggested facts (not auto-merged)**.\n\n"
    "Rules:\n"
    "- Output facts about the user, their projects, preferences, and decisions.\n"
    "- Include observations about shared interests or patterns in how the user works.\n"
    "- Identify recurring patterns (topics the user returns to, repeated preferences, "
    "  habitual workflows).\n"
    "- Surface connections between entries that the user may not have explicitly linked.\n"
    "- When a **Vault activity** section is present, treat vault note titles and graph links as "
    "high-trust user intent; weave vault themes into **Connections & Links** alongside chat/memory "
    "using **note titles only** (never UUIDs or raw note ids).\n"
    "- Write connection and pattern bullets in plain language suitable for the assistant to cite "
    "later (e.g. 'mesh testing and laptop setup often come up together').\n"
    "- Note any contradictions or superseded information.\n"
    "- Distill actionable takeaways the assistant should remember going forward.\n"
    "- For facts that change often (employer, OS, main branch, default model), add a short "
    "**## Current facts (high-churn)** section with bullets like `- (as of YYYY-MM-DD) ...` "
    "so downstream tools can prefer the latest dated line.\n"
    "- Keep speculative items in a separate suggested-facts section for human review.\n"
    "- Be concise. Output structured markdown with clear headers.\n"
    "- Do NOT invent information. Only synthesize from what is provided.\n"
    "- Do NOT include meta-commentary about the memory system, file statuses, "
    "  consolidation process, or heartbeat operations.\n"
    "- Do NOT include pleasantries or preamble. Start with the insights."
)

_CONSOLIDATION_USER_TEMPLATE = (
    "Below are memory entries and context from the past {days} days. "
    "Analyze them and produce consolidated insights.\n\n"
    "---\n\n{content}\n\n---\n\n"
    "Keep bullets short and durable; skip play-by-play and transient small talk.\n"
    "Produce your insights in this format:\n\n"
    "## Recurring Patterns\n"
    "- ...\n\n"
    "## Connections & Links\n"
    "- ...\n\n"
    "## Superseded / Contradicted Info\n"
    "- ...\n\n"
    "## Actionable Takeaways\n"
    "- ...\n\n"
    "## Current facts (high-churn)\n"
    "- (as of YYYY-MM-DD) ...  # optional; short-lived facts only\n\n"
    "## Suggested facts (not auto-merged)\n"
    "- ..."
)

_REQUIRED_INSIGHT_HEADINGS = (
    "Recurring Patterns",
    "Connections & Links",
    "Superseded / Contradicted Info",
    "Actionable Takeaways",
)
_STRICT_ALLOWED_MERGE_HEADINGS = (
    "Recurring Patterns",
    "Connections & Links",
    "Actionable Takeaways",
    "Current facts (high-churn)",
)
_PERSONAL_PROPOSAL_SECTIONS = (
    "Recurring Patterns",
    "Actionable Takeaways",
    "Current facts (high-churn)",
)
_SUGGESTED_FACTS_HEADING = "Suggested facts (not auto-merged)"
_BULLET_LINE_RE = re.compile(r"^\s*[-*]\s+(.+)$", re.MULTILINE)
_MIN_SECTION_BULLET_CHARS = 14  # align with background fact routing minimum


def _section_bullets(sections: dict[str, str], heading: str) -> list[str]:
    target = (heading or "").strip().lower()
    body = ""
    for key, section_body in sections.items():
        if (key or "").strip().lower() == target:
            body = (section_body or "").strip()
            break
    if not body:
        return []
    return [
        m.group(1).strip()
        for m in _BULLET_LINE_RE.finditer(body)
        if len(m.group(1).strip()) >= _MIN_SECTION_BULLET_CHARS
    ]


_CONNECTIONS_HEADING = "Connections & Links"
_CONNECTIONS_POINTER = (
    "- Cross-topic links recorded in memory/connections.md (see Connections Journal preload)."
)


def _personal_noticing_enabled(cfg: dict) -> bool:
    return bool(cfg.get("personal_noticing_enabled", True))


def _replace_connections_section_with_pointer(insights: str) -> str:
    """Swap Connections & Links body for a pointer to the journal file."""
    pattern = re.compile(
        r"(^## Connections & Links\s*\n)(.*?)(?=^## |\Z)",
        re.MULTILINE | re.DOTALL,
    )
    if not pattern.search(insights):
        return insights
    return pattern.sub(rf"\1{_CONNECTIONS_POINTER}\n", insights, count=1)


def _process_connections_and_links(
    workspace: str,
    insights: str,
    *,
    budget,
    cfg: dict,
    on_log=None,
) -> tuple[str, set[str]]:
    """Append journal narrative + propose connection facts; return insights + excluded norms."""
    if not _personal_noticing_enabled(cfg):
        return insights, set()

    bullets = _section_bullets(_extract_consolidation_sections(insights), _CONNECTIONS_HEADING)
    if not bullets:
        return insights, set()

    max_bytes = int(cfg.get("connections_journal_max_bytes", 65536))
    try:
        from utils.connections_journal import append_connections_journal

        appended = append_connections_journal(
            workspace,
            bullets,
            source="memory_consolidator",
            max_bytes=max_bytes,
        )
    except Exception:
        appended = 0
        if on_log:
            on_log("DEBUG", "[consolidation] connections journal append failed")

    excluded_norm: set[str] = set()
    if appended and on_log:
        on_log("INFO", f"[consolidation] appended {appended} line(s) to connections journal")

    counts = {"proposed": 0, "memory": 0, "skipped": 0}
    for raw in bullets:
        outcome = None
        try:
            from utils.user_model import propose_background_fact

            outcome = propose_background_fact(
                workspace,
                raw,
                source="memory_consolidator",
                rationale_prefix="Consolidation — Connections & Links",
                budget=budget,
                section_hint="connections",
            )
        except Exception:
            if on_log:
                on_log("DEBUG", "[consolidation] connection proposal failed")
            continue
        outcome_key = "proposed" if outcome == "proposal" else outcome
        if outcome_key in counts:
            counts[outcome_key] += 1
        norm = raw.strip().lower()
        if len(norm) >= _MIN_SECTION_BULLET_CHARS:
            excluded_norm.add(norm)

    if any(counts.values()) and on_log:
        on_log(
            "INFO",
            "[consolidation] connections routed "
            f"proposed={counts['proposed']} memory={counts['memory']} skipped={counts['skipped']}",
        )

    updated_insights = insights
    if appended or counts["proposed"] > 0:
        updated_insights = _replace_connections_section_with_pointer(insights)

    return updated_insights, excluded_norm


def _route_consolidation_bullets_to_user_model(
    workspace: str,
    bullets: list[str],
    *,
    rationale_prefix: str,
    budget,
) -> tuple[dict[str, int], set[str]]:
    """Route consolidation bullets via shared helper; return counts + merge-excluded norms."""
    from utils.user_model import (
        build_proposal_changes_from_fact,
        is_personal_fact_line,
        propose_background_fact,
    )

    counts = {"proposed": 0, "memory": 0, "skipped": 0}
    excluded_norm: set[str] = set()
    for raw in bullets:
        outcome = propose_background_fact(
            workspace,
            raw,
            source="memory_consolidator",
            rationale_prefix=rationale_prefix,
            budget=budget,
        )
        outcome_key = "proposed" if outcome == "proposal" else outcome
        if outcome_key in counts:
            counts[outcome_key] += 1
        norm = raw.strip().lower()
        personal_patch = build_proposal_changes_from_fact(raw)
        if personal_patch is not None or (
            outcome in ("proposal", "skipped") and is_personal_fact_line(raw)
        ):
            if len(norm) >= _MIN_SECTION_BULLET_CHARS:
                excluded_norm.add(norm)
    return counts, excluded_norm


def _propose_personal_sections_to_user_model(
    workspace: str,
    insights: str,
    *,
    budget=None,
    on_log=None,
) -> tuple[dict[str, int], set[str]]:
    """Route personal-signal bullets from merge-eligible sections to User Model proposals."""
    sections = _extract_consolidation_sections(insights)
    if not sections:
        return {"proposed": 0, "memory": 0, "skipped": 0}, set()

    from utils.user_model import BackgroundProposalBudget

    if budget is None:
        budget = BackgroundProposalBudget()

    totals = {"proposed": 0, "memory": 0, "skipped": 0}
    excluded_norm: set[str] = set()
    for heading in _PERSONAL_PROPOSAL_SECTIONS:
        bullets = _section_bullets(sections, heading)
        if not bullets:
            continue
        counts, excluded = _route_consolidation_bullets_to_user_model(
            workspace,
            bullets,
            rationale_prefix=f"Consolidation — {heading}",
            budget=budget,
        )
        for key in totals:
            totals[key] += counts[key]
        excluded_norm |= excluded

    if any(totals.values()) and on_log:
        on_log(
            "INFO",
            "[consolidation] personal sections routed "
            f"proposed={totals['proposed']} memory={totals['memory']} skipped={totals['skipped']}",
        )
    return totals, excluded_norm


def _propose_suggested_facts_to_user_model(
    workspace: str,
    insights: str,
    *,
    on_log=None,
    max_proposals: int = 3,
    budget=None,
) -> tuple[int, set[str]]:
    """Turn consolidation 'Suggested facts' bullets into User Model proposals."""
    del max_proposals  # per-cycle cap enforced by shared BackgroundProposalBudget
    sections = _extract_consolidation_sections(insights)
    bullets = _section_bullets(sections, _SUGGESTED_FACTS_HEADING)
    if not bullets:
        return 0, set()

    from utils.user_model import BackgroundProposalBudget

    if budget is None:
        budget = BackgroundProposalBudget()

    try:
        counts, excluded = _route_consolidation_bullets_to_user_model(
            workspace,
            bullets,
            rationale_prefix="Consolidation suggested fact",
            budget=budget,
        )
    except Exception:
        return 0, set()

    created = counts["proposed"]
    if created and on_log:
        on_log("INFO", f"[consolidation] created {created} User Model proposal(s) from suggested facts")
    return created, excluded


def _strip_personal_bullets_from_merge_text(text: str, excluded_norm: set[str] | None = None) -> str:
    """Remove personal-signal bullets before merge_to_memory (proposals own those facts)."""
    if not (text or "").strip():
        return ""
    try:
        from utils.user_model import is_personal_fact_line
    except Exception:
        is_personal_fact_line = lambda _line: False  # type: ignore[assignment,misc]

    excluded = {str(x).strip().lower() for x in (excluded_norm or set()) if str(x).strip()}
    kept: list[str] = []
    for ln in text.splitlines():
        match = _BULLET_LINE_RE.match(ln)
        if match:
            bullet = match.group(1).strip()
            norm = bullet.lower()
            if norm in excluded or is_personal_fact_line(bullet):
                continue
        kept.append(ln)
    merged = "\n".join(kept).strip()
    # Drop section headers whose body became empty.
    blocks: list[str] = []
    current: list[str] = []
    for ln in merged.splitlines():
        if ln.startswith("## "):
            if current:
                body_lines = [x for x in current[1:] if x.strip()]
                if body_lines:
                    blocks.append("\n".join(current))
            current = [ln]
        else:
            if not current:
                current = [""]
            current.append(ln)
    if current:
        body_lines = [x for x in current[1:] if x.strip()]
        if body_lines:
            blocks.append("\n".join(current))
    return "\n\n".join(blocks).strip()


def _extract_consolidation_sections(insights: str) -> dict[str, str]:
    text = (insights or "").strip()
    if not text:
        return {}
    sections: dict[str, str] = {}
    heading_re = re.compile(r"^##\s+(.+?)\s*$", re.MULTILINE)
    matches = list(heading_re.finditer(text))
    if not matches:
        return {}
    for idx, match in enumerate(matches):
        heading = (match.group(1) or "").strip()
        start = match.end()
        end = matches[idx + 1].start() if idx + 1 < len(matches) else len(text)
        body = text[start:end].strip()
        sections[heading] = body
    return sections


def _strict_sections_required(semantic_strictness: str) -> bool:
    return semantic_strictness in {"insights_sections", "enforce_promotion", "warn_promotion"}


def _build_mergeable_insights_text(
    insights: str,
    *,
    semantic_strictness: str,
    on_log: Callable[[str, str], None] | None = None,
) -> str | None:
    text = (insights or "").strip()
    if not text:
        return None
    if not _strict_sections_required(semantic_strictness):
        return text

    sections = _extract_consolidation_sections(text)
    missing = [h for h in _REQUIRED_INSIGHT_HEADINGS if h not in sections]
    if missing:
        if on_log:
            on_log("WARN", f"[consolidation] strict section validation failed; missing headings: {', '.join(missing)}")
        return None

    merge_blocks: list[str] = []
    for heading in _STRICT_ALLOWED_MERGE_HEADINGS:
        body = (sections.get(heading) or "").strip()
        if not body:
            continue
        merge_blocks.append(f"## {heading}\n{body}")
    merged = "\n\n".join(merge_blocks).strip()
    return merged or None


def _read_file_safe(path: str, max_chars: int = _MAX_INPUT_CHARS) -> str:
    if not path or not os.path.isfile(path):
        return ""
    try:
        with open(path, "r", encoding="utf-8") as f:
            return f.read(max_chars)
    except Exception:
        return ""


_INSIGHTS_BLOCK_RE = re.compile(r"(?=\n---\n\n## Consolidation — \d{4}-\d{2}-\d{2})")


def _cap_insights_file(insights_path: str, max_blocks: int = 15, on_log=None) -> None:
    """Keep only the most recent *max_blocks* consolidation blocks in insights.md."""
    try:
        content = _read_file_safe(insights_path, 500_000)
        if not content.strip():
            return
        parts = _INSIGHTS_BLOCK_RE.split(content)
        if len(parts) <= max_blocks + 1:
            return
        header = parts[0]
        kept = parts[-(max_blocks):]
        new_content = header.rstrip() + "\n" + "".join(kept)
        from utils.path_safety import secure_write_file
        with _INSIGHTS_WRITE_LOCK:
            secure_write_file(insights_path, new_content, mode=0o600)
        if on_log:
            on_log("INFO", f"[consolidation] capped insights.md to {max_blocks} blocks (removed {len(parts) - 1 - max_blocks})")
    except Exception as e:
        if on_log:
            on_log("WARN", f"[consolidation] insights cap failed: {e}")


def _append_vault_digest(combined: str, *, lookback_days: int, cfg: dict) -> str:
    if not _personal_noticing_enabled(cfg) or not cfg.get("vault_noticing_enabled", True):
        return combined
    try:
        from utils.vault_noticing_context import build_vault_noticing_digest, shrink_vault_digest

        vault_block = build_vault_noticing_digest(
            lookback_days=lookback_days,
            max_chars=int(cfg.get("vault_noticing_max_chars", 4096)),
            max_notes=int(cfg.get("vault_noticing_max_notes", 8)),
        )
        if not vault_block:
            return combined
        merged = f"{combined}\n\n---\n\n{vault_block}" if combined.strip() else vault_block
        if len(merged) <= _MAX_INPUT_CHARS:
            return merged
        vault_only = shrink_vault_digest(
            vault_block,
            max_chars=max(512, _MAX_INPUT_CHARS - len(combined) - 8),
        )
        if vault_only:
            merged = f"{combined}\n\n---\n\n{vault_only}" if combined.strip() else vault_only
        if len(merged) > _MAX_INPUT_CHARS:
            merged = merged[:_MAX_INPUT_CHARS] + "\n\n[... truncated for length]"
        return merged
    except Exception as exc:
        import logging

        logging.getLogger(__name__).warning(
            "[consolidation] vault digest skipped (fail-open): %s", exc
        )
        return combined


def _gather_memory_content(workspace: str, lookback_days: int = 7) -> str:
    """Collect consolidation context via agent_plane."""
    combined = ""
    try:
        from utils.agent_plane import build_wake_context_snippet

        combined = build_wake_context_snippet(
            workspace=workspace,
            wake_kind="consolidation",
            task_name="memory_maintenance",
            signals={"lookback_days": lookback_days},
        )
    except Exception:
        combined = ""

    cfg = _get_consolidation_config()
    combined = _append_vault_digest(combined, lookback_days=lookback_days, cfg=cfg)

    if len(combined) > _MAX_INPUT_CHARS:
        combined = combined[:_MAX_INPUT_CHARS] + "\n\n[... truncated for length]"
    return combined


def _get_insights_path(workspace: str) -> str:
    insights_dir = os.path.join(workspace, "memory")
    os.makedirs(insights_dir, exist_ok=True)
    return os.path.join(insights_dir, "insights.md")


def _get_consolidation_config() -> dict:
    try:
        from core.config import get_consolidation_config
        return get_consolidation_config()
    except Exception:
        return {
            "personal_noticing_enabled": True,
            "personal_pattern_observations_enabled": True,
        }


def _knowledge_hint_block(context_text: str = "") -> str:
    try:
        from tools.knowledge_tools import list_knowledge_corpora
    except Exception:
        return ""
    corpora = list_knowledge_corpora()
    if not corpora:
        return ""
    token_re = re.compile(r"[a-zA-Z0-9_]{3,}")
    context_tokens = {t.lower() for t in token_re.findall(context_text or "")}
    lines = ["## Knowledge corpus hints (not auto-merged)"]
    overlap_count = 0
    for row in corpora[:8]:
        label = str(row.get("label") or row.get("slug") or "").strip()
        topics = [str(t).strip() for t in (row.get("topics") or []) if str(t).strip()]
        topic_tokens = {t.lower() for t in token_re.findall(" ".join(topics))}
        overlap = sorted(topic_tokens & context_tokens)
        if context_tokens and not overlap:
            continue
        overlap_count += 1
        topic_text = ", ".join(topics[:5]) if topics else "no topics yet"
        if overlap:
            lines.append(f"- {label}: overlaps with recent themes ({', '.join(overlap[:4])}); topics: {topic_text}")
        else:
            lines.append(f"- {label}: {topic_text}")
    if overlap_count == 0:
        return ""
    return "\n".join(lines)


def consolidate_memories(
    workspace: str | None = None,
    lookback_days: int = 7,
    on_log=None,
) -> str | None:
    """Run one consolidation cycle: gather memory, call LLM, write insights.

    Returns the insights text on success, or None if skipped/failed.
    Uses subagent_runner infrastructure for the LLM call (isolated, no personality).
    """
    if workspace is None:
        try:
            from core import get_workspace_path
            workspace = get_workspace_path()
        except Exception:
            return None

    cfg = _get_consolidation_config()
    semantic_strictness = str(get_memory_policy_prefs().get("memory_semantic_strictness", "off"))
    _conn_excluded: set[str] = set()
    _conn_hook_failed = False
    if not cfg.get("enabled", True):
        if on_log:
            on_log("INFO", "[consolidation] disabled in config, skipping")
        return None

    lookback = int(cfg.get("lookback_days", lookback_days))
    content = _gather_memory_content(workspace, lookback_days=lookback)
    content, _btw_dropped = _scrub_btw_thread_refs(content)
    if _btw_dropped and on_log:
        on_log(
            "WARNING",
            f"[consolidation] scrubbed {_btw_dropped} btw_* line(s) from "
            "aggregated context (defense-in-depth; upstream filters should "
            "have removed these)",
        )
    if not content.strip():
        if on_log:
            on_log("INFO", "[consolidation] no memory content to consolidate")
        return None

    if on_log:
        on_log("INFO", f"[consolidation] gathered {len(content)} chars of memory content")

    user_prompt = _CONSOLIDATION_USER_TEMPLATE.format(days=lookback, content=content)
    messages = [
        {"role": "user", "content": f"{_CONSOLIDATION_SYSTEM}\n\n---\n\n{user_prompt}"},
    ]

    model = (cfg.get("model") or "").strip() or None
    base_url = (cfg.get("base_url") or "").strip() or None
    api_key = (cfg.get("api_key") or "").strip() or None
    temp = float(cfg.get("temperature", 0.2))
    max_tokens = int(cfg.get("max_tokens", 1200))

    try:
        from agent_loop import run_raw_agent

        collector: list[str] = []
        error_ref: list[str] = []

        def _on_chunk(text):
            if text:
                collector.append(text)

        def _on_error(err):
            error_ref.append(str(err))

        from tool_policy import context_from_prefs_and_flags
        _bg_policy_ctx = context_from_prefs_and_flags(background_automation=True, subagent_mode=True)

        returned = run_raw_agent(
            messages=messages,
            thread_id=f"consolidation_{datetime.now().strftime('%Y%m%d_%H%M')}",
            config=None,
            on_chunk=_on_chunk,
            on_error=_on_error,
            on_log=on_log,
            research_mode=False,
            subagent_mode=True,
            model_override=model,
            base_url_override=base_url,
            api_key_override=api_key,
            tools_override=[],
            temperature=temp,
            max_tokens=max_tokens,
            stop=cfg.get("stop"),
            response_format=cfg.get("response_format") if isinstance(cfg.get("response_format"), dict) else None,
            policy_context=_bg_policy_ctx,
        )

        insights = ((returned or "").strip() or "".join(collector).strip())
        insights = strip_untagged_cot(strip_markdown_fences(strip_think_tags(insights)))
        if is_visibility_placeholder(insights):
            if on_log:
                on_log(
                    "WARN",
                    "[consolidation] LLM returned chat visibility placeholder only; skipping write",
                )
            return None
        if error_ref and not insights:
            if on_log:
                on_log("ERROR", f"[consolidation] LLM error: {error_ref[0]}")
            return None

        if not insights:
            if on_log:
                on_log("WARN", "[consolidation] LLM returned empty response")
            return None

        insights, _btw_llm = _scrub_btw_thread_refs(insights)
        if _btw_llm and on_log:
            on_log(
                "WARNING",
                f"[consolidation] scrubbed {_btw_llm} btw_* line(s) from LLM output before write",
            )

        if len(insights) > _MAX_INSIGHT_CHARS:
            insights = insights[:_MAX_INSIGHT_CHARS] + "\n\n[... truncated]"
        hints = _knowledge_hint_block(content)
        if hints:
            insights = f"{insights.strip()}\n\n{hints}\n"

        _conn_cfg = _get_consolidation_config()
        try:
            from utils.user_model import BackgroundProposalBudget

            _conn_budget = BackgroundProposalBudget()
            insights, _conn_excluded = _process_connections_and_links(
                workspace,
                insights,
                budget=_conn_budget,
                cfg=_conn_cfg,
                on_log=on_log,
            )
        except Exception:
            _conn_hook_failed = True
            if on_log:
                on_log("WARNING", "[consolidation] connections processing failed; merge_to_memory may be skipped")

    except Exception as e:
        if on_log:
            on_log("ERROR", f"[consolidation] failed: {e}")
        return None

    now = datetime.now().strftime("%Y-%m-%d %H:%M")
    insights_path = _get_insights_path(workspace)
    existing = _read_file_safe(insights_path, _MAX_INPUT_CHARS)

    new_block = f"\n---\n\n## Consolidation — {now}\n\n{insights}\n"

    if existing.strip():
        new_content = existing.rstrip() + "\n" + new_block
    else:
        new_content = f"# Memory Insights (auto-consolidated)\n{new_block}"

    try:
        from utils.path_safety import secure_write_file
        with _INSIGHTS_WRITE_LOCK:
            secure_write_file(insights_path, new_content, mode=0o600)
    except Exception:
        try:
            with _INSIGHTS_WRITE_LOCK:
                with open(insights_path, "w", encoding="utf-8") as f:
                    f.write(new_content)
                os.chmod(insights_path, 0o600)
        except Exception as write_err:
            if on_log:
                on_log("ERROR", f"[consolidation] write failed: {write_err}")
            return None

    if on_log:
        on_log("INFO", f"[consolidation] wrote {len(insights)} chars to memory/insights.md")

    _merge_excluded: set[str] = set()
    _um_hook_failed = False
    try:
        from utils.user_model import BackgroundProposalBudget

        _proposal_budget = BackgroundProposalBudget()
        _suggested_created, _suggested_excluded = _propose_suggested_facts_to_user_model(
            workspace,
            insights,
            budget=_proposal_budget,
            on_log=on_log,
        )
        del _suggested_created
        _merge_excluded |= _suggested_excluded
        _personal_counts, _personal_excluded = _propose_personal_sections_to_user_model(
            workspace,
            insights,
            budget=_proposal_budget,
            on_log=on_log,
        )
        del _personal_counts
        _merge_excluded |= _personal_excluded
        _merge_excluded |= _conn_excluded
        if (
            bool(_conn_cfg.get("personal_noticing_enabled", True))
            and bool(_conn_cfg.get("personal_pattern_observations_enabled", True))
        ):
            try:
                from utils.personal_pattern_observations import (
                    maybe_emit_consolidation_pattern_observation,
                )

                _pattern_bullets = _section_bullets(
                    _extract_consolidation_sections(insights),
                    "Recurring Patterns",
                )
                maybe_emit_consolidation_pattern_observation(
                    workspace,
                    _pattern_bullets,
                    on_log=on_log,
                )
            except Exception:
                if on_log:
                    on_log("DEBUG", "[consolidation] personal pattern observation hook failed")
    except Exception:
        _um_hook_failed = True
        if on_log:
            on_log(
                "WARNING",
                "[consolidation] User Model proposal hook failed; merge_to_memory skipped if enabled",
            )

    _cap_insights_file(insights_path, max_blocks=15, on_log=on_log)

    dreaming_enabled = False
    try:
        from core.config import get_dreaming_config
        _dream_cfg = get_dreaming_config() or {}
        dreaming_enabled = bool(_dream_cfg.get("enabled", False))
    except Exception:
        _dream_cfg = {}
        dreaming_enabled = False

    if dreaming_enabled:
        try:
            from utils.dreaming import run_cycle as _run_dream_cycle

            _run_dream_cycle(
                workspace,
                insights,
                promote_threshold=float(_dream_cfg.get("promote_threshold", 0.60)),
                utility_tag_enabled=bool(_dream_cfg.get("utility_tag_enabled", True)),
                max_candidates=int(_dream_cfg.get("max_candidates_per_cycle", 25)),
                semantic_strictness=semantic_strictness,
                on_log=on_log,
            )
        except Exception as dream_err:
            if on_log:
                on_log("WARN", f"[consolidation] dreaming cycle failed: {dream_err}")
    elif cfg.get("merge_to_memory", False):
        if _um_hook_failed or _conn_hook_failed:
            if on_log:
                on_log(
                    "WARNING",
                    "[consolidation] merge_to_memory skipped (personal routing hook failed)",
                )
        else:
            merge_limit = int(cfg.get("merge_max_chars", 1600))
            merge_source = _build_mergeable_insights_text(
                insights,
                semantic_strictness=semantic_strictness,
                on_log=on_log,
            )
            if merge_source is None:
                if on_log:
                    on_log("INFO", "[consolidation] merge_to_memory skipped (strict section validation or empty merge payload)")
            elif len(merge_source) > merge_limit:
                if on_log:
                    on_log("INFO", f"[consolidation] merge_to_memory skipped (insights exceeds merge_max_chars={merge_limit})")
            else:
                merge_source = _strip_personal_bullets_from_merge_text(
                    merge_source,
                    _merge_excluded,
                )
                if not merge_source.strip():
                    if on_log:
                        on_log("INFO", "[consolidation] merge_to_memory skipped (no project bullets after personal routing)")
                else:
                    _merge_insight_to_memory(workspace, merge_source, now, on_log)

    # Auto-prune: if insights mention superseded/contradicted info and auto_prune is on,
    # scan MEMORY.md for stale time-relative entries and remove the most obvious ones.
    if cfg.get("auto_prune", True):
        _auto_prune_stale(workspace, insights, on_log)

    return insights


def _auto_prune_stale(workspace: str, insights: str, on_log=None):
    """Remove obviously stale dated entries from long-term memory after consolidation.

    In **sharded** layout, scans each ``memory/shards/*.md`` plus the hub if it
    still contains dated blocks. **Monolith:** ``MEMORY.md`` only.
    """
    from utils.memory_shard_layout import layout_is_sharded, list_shard_abs_paths

    paths: list[str] = []
    if layout_is_sharded(workspace):
        paths.extend(list_shard_abs_paths(workspace))
        mem_hub = os.path.join(workspace, "MEMORY.md")
        if os.path.isfile(mem_hub):
            try:
                hub_raw = _read_file_safe(mem_hub, 500_000)
            except Exception:
                hub_raw = ""
            if hub_raw and re.search(r"##\s+\d{4}-\d{2}-\d{2}", hub_raw):
                paths.append(mem_hub)
    else:
        memory_path = os.path.join(workspace, "MEMORY.md")
        if os.path.isfile(memory_path):
            paths = [memory_path]

    for memory_path in paths:
        _auto_prune_stale_one_file(workspace, memory_path, insights, on_log)


def _auto_prune_stale_one_file(workspace: str, memory_path: str, insights: str, on_log=None) -> None:
    """Remove obviously stale entries from one memory markdown file."""

    if not os.path.isfile(memory_path):
        return

    try:
        from tools.memory_tools import _split_into_blocks, _blocks_to_content, _read_memory_file
        from utils.path_safety import secure_write_file

        content = _read_memory_file(memory_path)
        if not content.strip():
            return

        blocks = _split_into_blocks(content)
        if len(blocks) < 3:
            return

        # Extract superseded/contradicted mentions from insights
        superseded_lower = ""
        sup_match = re.search(
            r"(?:superseded|contradicted|outdated|stale).*?\n(.*?)(?=\n## |\Z)",
            insights, re.DOTALL | re.IGNORECASE
        )
        if sup_match:
            superseded_lower = sup_match.group(1).lower()

        if not superseded_lower:
            return

        _time_relative_re = re.compile(
            r"\b(this week|this month|right now|currently|at the moment|today|yesterday|"
            r"just now|these days|lately|recently|at present|working on)\b",
            re.IGNORECASE,
        )

        now = datetime.now()
        removed_count = 0
        new_blocks = []
        for header, body in blocks:
            date_match = re.search(r"(\d{4}-\d{2}-\d{2})", header)
            if not date_match:
                new_blocks.append((header, body))
                continue
            try:
                entry_date = datetime.strptime(date_match.group(1), "%Y-%m-%d")
            except ValueError:
                new_blocks.append((header, body))
                continue
            age_days = (now - entry_date).days
            if age_days < 14:
                new_blocks.append((header, body))
                continue

            body_lower = body.lower()
            is_time_relative = bool(_time_relative_re.search(body))
            is_superseded = any(
                phrase in superseded_lower
                for phrase in body_lower.split("\n")
                if len(phrase.strip()) > 10
            )

            if is_time_relative and is_superseded:
                removed_count += 1
                continue

            new_blocks.append((header, body))

        if removed_count == 0:
            return

        new_content = _blocks_to_content(new_blocks)
        try:
            secure_write_file(memory_path, new_content, mode=0o600)
        except Exception:
            with open(memory_path, "w", encoding="utf-8") as f:
                f.write(new_content)
            import stat
            os.chmod(memory_path, stat.S_IRUSR | stat.S_IWUSR)

        _notify_memory_fts_paths(workspace, memory_path)

        if on_log:
            try:
                label = os.path.relpath(memory_path, workspace)
            except Exception:
                label = memory_path
            on_log("INFO", f"[consolidation] auto-pruned {removed_count} stale entries from {label}")

    except Exception as e:
        if on_log:
            on_log("WARN", f"[consolidation] auto-prune failed for {memory_path}: {e}")


def _merge_insight_to_memory(workspace: str, insights: str, timestamp: str, on_log=None):
    """Optionally append a condensed version of insights to long-term memory.

    **Sharded layout:** appends to ``memory/shards/consolidation.md`` (keeps the
    hub small). **Monolith:** appends to root ``MEMORY.md``.

    Each line is screened through classify_memory_entry; blocked and discouraged
    merge lines are dropped before the write (secrets, meta-memory, vague, episodic-only, etc.).
    """
    from utils.memory_promotion import classify_memory_entry, PromotionClass
    from utils.memory_shard_layout import ensure_shard_file, layout_is_sharded, regenerate_memory_hub_toc

    from utils.memory_dedup import dedupe_bullet_lines

    lines = [ln for ln in insights.splitlines() if ln.strip()]
    kept: list[str] = []
    for ln in lines:
        decision = classify_memory_entry(ln)
        if decision.classification == PromotionClass.BLOCKED:
            if on_log:
                on_log("DEBUG", f"[consolidation] merge: dropped blocked line ({decision.reason}): {ln[:80]}")
            continue
        if decision.classification == PromotionClass.DISCOURAGED:
            if on_log:
                on_log("DEBUG", f"[consolidation] merge: dropped discouraged line ({decision.reason}): {ln[:80]}")
            continue
        kept.append(ln)
    kept = dedupe_bullet_lines(kept)
    screened = "\n".join(kept).strip()
    screened, btw_merge_drop = _scrub_btw_thread_refs(screened)
    if btw_merge_drop and on_log:
        on_log(
            "WARNING",
            f"[consolidation] scrubbed {btw_merge_drop} btw_* line(s) from merge payload (defense-in-depth)",
        )
    if not screened:
        if on_log:
            on_log("INFO", "[consolidation] merge_to_memory skipped (all content removed by promotion filter)")
        return

    sharded = layout_is_sharded(workspace)
    if sharded:
        abs_path, serr = ensure_shard_file(workspace, "consolidation")
        if serr or not abs_path:
            if on_log:
                on_log("ERROR", f"[consolidation] merge: could not open consolidation shard: {serr}")
            return
        memory_path = abs_path
        dest_label = "memory/shards/consolidation.md"
    else:
        memory_path = os.path.join(workspace, "MEMORY.md")
        dest_label = "MEMORY.md"

    existing = _read_file_safe(memory_path, _MAX_INPUT_CHARS)

    short = screened[:1500]
    if len(screened) > 1500:
        short = short.rstrip() + "\n\n[See memory/insights.md for full consolidation]"

    block = f"\n\n## {timestamp} (consolidated insight)\n{short}\n"
    hub_stub = "# Consolidation merge targets\n\n"
    new_content = (existing.rstrip() + block) if existing.strip() else hub_stub + block.strip() + "\n"

    try:
        from utils.path_safety import secure_write_file

        secure_write_file(memory_path, new_content, mode=0o600)
        notify_paths = [memory_path]
        if sharded:
            toc_err = regenerate_memory_hub_toc(workspace)
            hub_p = os.path.join(workspace, "MEMORY.md")
            if os.path.isfile(hub_p):
                notify_paths.append(hub_p)
            if toc_err and on_log:
                on_log("WARN", f"[consolidation] merge wrote shard but TOC refresh failed: {toc_err}")
        _notify_memory_fts_paths(workspace, *notify_paths)
        if on_log:
            on_log("INFO", f"[consolidation] merged condensed insight to {dest_label}")
    except Exception as e:
        if on_log:
            on_log("ERROR", f"[consolidation] merge to {dest_label} failed: {e}")
