"""
utils/user_model.py

First-class, versioned, user-correctable model of the human.

This is the canonical source of truth for identity, goals, relationships,
preferences, current focus, values, and open commitments.

Design reference: docs/user_model_design.md
Implementation plan: .cursor/plans/user_model_first_class.plan.md (Phase 0)

Key invariants (non-negotiable):
- All mutation goes through apply_patch (or the proposal path).
- Background, idle, and scheduled contexts are blocked from direct mutation
  (reuses/extends the existing personality write guard philosophy).
- The workspace git (when evolution mode is enabled) provides the primary
  versioning history. We add lightweight provenance + explicit checkpoints.
- This module must remain narrow and high-trust.

This file is intentionally minimal in Phase 0. It introduces the narrow API
and storage layout with no behavior change to existing USER.md, drives,
open_loops, or agent_plane flows.
"""

from __future__ import annotations

import json
import logging
import os
import re
import uuid
from contextlib import contextmanager
from contextvars import ContextVar
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Callable, Iterator

from persistence import secure_atomic_write
from utils.path_safety import safe_resolve_path

_log = logging.getLogger(__name__)

_PROPOSAL_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
_CHECKPOINT_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,191}$")
_patch_user_model_active: ContextVar[bool] = ContextVar("_patch_user_model_active", default=False)
_proposal_write_active: ContextVar[bool] = ContextVar("_proposal_write_active", default=False)
_checkpoint_write_active: ContextVar[bool] = ContextVar("_checkpoint_write_active", default=False)
_audit_write_active: ContextVar[bool] = ContextVar("_audit_write_active", default=False)

_DICT_SECTIONS = frozenset({"identity", "preferences", "current_focus"})
_LIST_SECTIONS = frozenset({"goals", "relationships", "connections", "open_commitments"})

# ---------------------------------------------------------------------------
# Storage layout (under workspace)
# ---------------------------------------------------------------------------

USER_MODEL_DIR = "memory/user_model"
MODEL_JSON = f"{USER_MODEL_DIR}/model.json"
USER_MODEL_MD = f"{USER_MODEL_DIR}/USER_MODEL.md"
PROPOSALS_DIR = f"{USER_MODEL_DIR}/proposals"
CHECKPOINTS_DIR = f"{USER_MODEL_DIR}/checkpoints"
AUDIT_LOG = f"{USER_MODEL_DIR}/audit.jsonl"

# ---------------------------------------------------------------------------
# Schema v1 (pragmatic starting point)
# ---------------------------------------------------------------------------

@dataclass
class UserModelProvenance:
    source: str  # "user", "agent", "ui", "migration", "proposal_accepted", etc.
    thread_id: str | None = None
    timestamp: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
    note: str | None = None


@dataclass
class UserModel:
    schema_version: int = 1
    last_updated: str = ""
    updated_by: str = ""
    provenance: UserModelProvenance | None = None

    identity: dict[str, Any] = field(default_factory=dict)
    goals: list[dict[str, Any]] = field(default_factory=list)
    relationships: list[dict[str, Any]] = field(default_factory=list)
    connections: list[dict[str, Any]] = field(default_factory=list)
    preferences: dict[str, Any] = field(default_factory=dict)
    current_focus: dict[str, Any] = field(default_factory=dict)
    values_north_stars: list[str] = field(default_factory=list)
    open_commitments: list[dict[str, Any]] = field(default_factory=list)

    def to_dict(self) -> dict[str, Any]:
        data = {
            "schema_version": self.schema_version,
            "last_updated": self.last_updated,
            "updated_by": self.updated_by,
            "identity": self.identity,
            "goals": self.goals,
            "relationships": self.relationships,
            "connections": self.connections,
            "preferences": self.preferences,
            "current_focus": self.current_focus,
            "values_north_stars": self.values_north_stars,
            "open_commitments": self.open_commitments,
        }
        if self.provenance:
            data["provenance"] = {
                "source": self.provenance.source,
                "thread_id": self.provenance.thread_id,
                "timestamp": self.provenance.timestamp,
                "note": self.provenance.note,
            }
        return data

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "UserModel":
        prov = None
        if isinstance(data.get("provenance"), dict):
            p = data["provenance"]
            prov = UserModelProvenance(
                source=p.get("source", "unknown"),
                thread_id=p.get("thread_id"),
                timestamp=p.get("timestamp", ""),
                note=p.get("note"),
            )

        def _as_dict(value: Any) -> dict[str, Any]:
            return dict(value) if isinstance(value, dict) else {}

        def _as_list(value: Any) -> list[Any]:
            return list(value) if isinstance(value, list) else []

        def _as_str_list(value: Any) -> list[str]:
            if not isinstance(value, list):
                return []
            return [str(v) for v in value if v is not None and str(v).strip()]

        return cls(
            schema_version=data.get("schema_version", 1),
            last_updated=data.get("last_updated", ""),
            updated_by=data.get("updated_by", ""),
            provenance=prov,
            identity=_as_dict(data.get("identity")),
            goals=[g for g in _as_list(data.get("goals")) if isinstance(g, dict)],
            relationships=[r for r in _as_list(data.get("relationships")) if isinstance(r, dict)],
            connections=[c for c in _as_list(data.get("connections")) if isinstance(c, dict)],
            preferences=_as_dict(data.get("preferences")),
            current_focus=_as_dict(data.get("current_focus")),
            values_north_stars=_as_str_list(data.get("values_north_stars")),
            open_commitments=[c for c in _as_list(data.get("open_commitments")) if isinstance(c, dict)],
        )


# ---------------------------------------------------------------------------
# Core API (narrow surface)
# ---------------------------------------------------------------------------

def _resolve_user_model_path(workspace: str, rel_path: str, *, allow_write: bool = False) -> str | None:
    abs_path, err = safe_resolve_path(workspace, rel_path, allow_write=allow_write, create_parents=allow_write)
    if err:
        return None
    return abs_path


@contextmanager
def patch_user_model_write_context() -> Iterator[None]:
    """Allow apply_patch to write model.json / USER_MODEL.md while global file guards apply."""
    token = _patch_user_model_active.set(True)
    try:
        yield
    finally:
        _patch_user_model_active.reset(token)


@contextmanager
def proposal_write_context() -> Iterator[None]:
    """Allow create_proposal / accept / reject to write under proposals/."""
    token = _proposal_write_active.set(True)
    try:
        yield
    finally:
        _proposal_write_active.reset(token)


@contextmanager
def checkpoint_write_context() -> Iterator[None]:
    """Allow create_checkpoint to write under checkpoints/."""
    token = _checkpoint_write_active.set(True)
    try:
        yield
    finally:
        _checkpoint_write_active.reset(token)


@contextmanager
def audit_write_context() -> Iterator[None]:
    """Allow _append_audit to append audit.jsonl while global file guards apply."""
    token = _audit_write_active.set(True)
    try:
        yield
    finally:
        _audit_write_active.reset(token)


def user_model_storage_root(workspace: str) -> str | None:
    """Absolute path to memory/user_model when resolvable."""
    return _resolve_user_model_path(workspace, USER_MODEL_DIR, allow_write=False)


def is_user_model_storage_path(workspace: str, abs_path: str) -> bool:
    """True when *abs_path* is anywhere under memory/user_model/."""
    root = user_model_storage_root(workspace)
    if not root:
        return False
    norm = os.path.normpath(abs_path)
    root_norm = os.path.normpath(root)
    return norm == root_norm or norm.startswith(root_norm + os.sep)


def is_user_model_managed_path(workspace: str, abs_path: str) -> bool:
    """True when *abs_path* is canonical User Model storage (not proposals/audit)."""
    model_path = _resolve_user_model_path(workspace, MODEL_JSON, allow_write=False)
    md_path = _resolve_user_model_path(workspace, USER_MODEL_MD, allow_write=False)
    if not model_path and not md_path:
        base = os.path.basename(os.path.normpath(abs_path)).lower()
        return base in ("model.json", "user_model.md")
    norm = os.path.normpath(abs_path)
    for protected in (model_path, md_path):
        if protected and norm == os.path.normpath(protected):
            return True
    base = os.path.basename(norm).lower()
    return base in ("model.json", "user_model.md")


def _path_looks_like_user_model_storage(abs_path: str) -> bool:
    norm = os.path.normpath(abs_path).replace("\\", "/").lower()
    return "/memory/user_model/" in norm or norm.endswith("/memory/user_model")


def _user_model_rel_from_abs(abs_path: str) -> str | None:
    norm = os.path.normpath(abs_path).replace("\\", "/")
    lower = norm.lower()
    marker = "/memory/user_model/"
    idx = lower.find(marker)
    if idx == -1:
        if lower.endswith("/memory/user_model"):
            return ""
        return None
    return norm[idx + len(marker) :]


def _user_model_rel_path(workspace: str, abs_path: str) -> str | None:
    norm = os.path.normpath(abs_path)
    ws = (workspace or "").strip()
    if ws:
        root = user_model_storage_root(ws)
        if root:
            root_norm = os.path.normpath(root)
            if norm == root_norm or norm.startswith(root_norm + os.sep):
                return os.path.relpath(norm, root_norm).replace("\\", "/")
    if not _path_looks_like_user_model_storage(abs_path):
        return None
    return _user_model_rel_from_abs(abs_path)


def check_user_model_direct_write_allowed(
    workspace: str,
    abs_path: str,
    *,
    mode: str = "overwrite",
) -> str | None:
    """Refuse direct tool writes/deletes under memory/user_model/ outside blessed APIs."""
    rel = _user_model_rel_path(workspace, abs_path)
    if rel is None:
        return None

    norm = os.path.normpath(abs_path)
    base = os.path.basename(norm).lower()
    is_managed = base in ("model.json", "user_model.md", "user_projection.md")

    if is_managed:
        if _patch_user_model_active.get():
            return None
        target = "model.json" if base == "model.json" else "USER_MODEL.md"
        if mode == "delete":
            return (
                f"Refused: cannot delete {target}. "
                "The User Model is user-owned; update it via patch_user_model or Settings → You."
            )
        if mode == "append":
            return (
                f"Refused: cannot append to {target}. "
                "Use patch_user_model in chat or accept a proposal in Settings → You."
            )
        return (
            f"Refused: direct writes to {target} are blocked. "
            "Use patch_user_model in chat or accept a proposal in Settings → You."
        )

    if rel == AUDIT_LOG.split("/")[-1] or rel.endswith("/audit.jsonl") or rel == "audit.jsonl":
        if _audit_write_active.get() and mode == "append":
            return None
        return "Refused: User Model audit log is system-managed."

    if rel.startswith("proposals/") or rel == "proposals":
        if _proposal_write_active.get():
            return None
        return (
            "Refused: direct writes to User Model proposals are blocked. "
            "Use propose_user_model_update in chat or Settings → You."
        )

    if rel.startswith("checkpoints/") or rel == "checkpoints":
        if _checkpoint_write_active.get():
            return None
        return (
            "Refused: direct writes to User Model checkpoints are blocked. "
            "Create checkpoints in Settings → You."
        )

    if mode == "delete":
        return (
            "Refused: cannot delete User Model storage files directly. "
            "Use patch_user_model or Settings → You."
        )
    return (
        "Refused: direct writes under memory/user_model/ are blocked. "
        "Use patch_user_model, propose_user_model_update, or Settings → You."
    )


def guard_direct_tool_write(
    workspace: str | None,
    abs_path: str,
    *,
    mode: str = "overwrite",
) -> str | None:
    """Return a refusal message when a tool must not write/delete *abs_path*, else None."""
    return check_user_model_direct_write_allowed((workspace or "").strip(), abs_path, mode=mode)


def refuse_low_level_user_model_write(abs_path: str, *, mode: str = "overwrite") -> None:
    """Raise PermissionError when *abs_path* is User Model storage outside blessed APIs."""
    err = enforce_low_level_user_model_write_guard(abs_path, mode=mode)
    if err:
        raise PermissionError(err)


def enforce_low_level_user_model_write_guard(
    abs_path: str,
    *,
    mode: str = "overwrite",
) -> str | None:
    """Return a refusal message for low-level writes, else None. Fail-closed on guard errors."""
    path = os.path.abspath(os.path.expanduser(abs_path))
    try:
        from core.config import get_workspace_path

        ws = get_workspace_path() or ""
    except Exception:
        ws = ""
    try:
        return guard_direct_tool_write(ws, path, mode=mode)
    except Exception:
        if _path_looks_like_user_model_storage(path):
            return (
                "Refused: direct writes under memory/user_model/ are blocked "
                "(write guard error)."
            )
        return None


def _validate_proposal_id(proposal_id: str) -> str | None:
    raw = (proposal_id or "").strip()
    if not raw or not _PROPOSAL_ID_RE.match(raw):
        return "Invalid proposal id"
    if ".." in raw or "/" in raw or "\\" in raw:
        return "Invalid proposal id"
    return None


def _deep_merge_dict(base: dict[str, Any], patch: dict[str, Any]) -> dict[str, Any]:
    out = dict(base)
    for key, value in patch.items():
        if isinstance(value, dict) and isinstance(out.get(key), dict):
            out[key] = {**out[key], **value}
        else:
            out[key] = value
    return out


def _item_key(item: dict[str, Any]) -> str | None:
    item_id = item.get("id")
    if item_id is not None and str(item_id).strip():
        return str(item_id).strip()
    title = item.get("title") or item.get("text")
    if title is not None and str(title).strip():
        return f"title:{str(title).strip().lower()}"
    return None


def _merge_list_items(existing: list[dict[str, Any]], incoming: list[dict[str, Any]]) -> list[dict[str, Any]]:
    if not incoming:
        return list(existing)
    if not existing:
        return list(incoming)

    merged = [dict(item) for item in existing]
    index_by_key: dict[str, int] = {}
    for idx, item in enumerate(merged):
        key = _item_key(item)
        if key:
            index_by_key[key] = idx

    for item in incoming:
        if not isinstance(item, dict):
            continue
        key = _item_key(item)
        if key and key in index_by_key:
            merged[index_by_key[key]] = {**merged[index_by_key[key]], **item}
        else:
            merged.append(dict(item))
            if key:
                index_by_key[key] = len(merged) - 1
    return merged


def _merge_string_list(existing: list[str], incoming: list[str]) -> list[str]:
    seen: set[str] = set()
    out: list[str] = []
    for value in list(existing) + list(incoming):
        text = str(value).strip()
        if not text or text in seen:
            continue
        seen.add(text)
        out.append(text)
    return out


def _apply_section_patch(model: UserModel, key: str, value: Any, *, replace: bool = False) -> bool:
    """Apply one section; return False when *value* has the wrong shape."""
    if key in _DICT_SECTIONS:
        if not isinstance(value, dict):
            return False
        if replace:
            cleaned = {
                str(k).strip(): v
                for k, v in value.items()
                if str(k).strip() and v is not None and str(v).strip()
            }
            setattr(model, key, cleaned)
        else:
            current = getattr(model, key)
            setattr(model, key, _deep_merge_dict(current if isinstance(current, dict) else {}, value))
        return True
    if key in _LIST_SECTIONS:
        if not isinstance(value, list):
            return False
        current = getattr(model, key)
        items = [x for x in value if isinstance(x, dict)]
        if replace:
            setattr(model, key, items)
        else:
            setattr(model, key, _merge_list_items(current if isinstance(current, list) else [], items))
        return True
    if key == "values_north_stars":
        if not isinstance(value, list):
            return False
        current = getattr(model, key)
        setattr(model, key, _merge_string_list(current if isinstance(current, list) else [], value))
        return True
    return False


def patch_has_applicable_sections(patch: dict[str, Any]) -> bool:
    """True when *patch* contains at least one well-typed, non-empty section."""
    allowed_top_level = {
        "identity", "goals", "relationships", "connections", "preferences",
        "current_focus", "values_north_stars", "open_commitments",
    }
    for key, value in patch.items():
        if key not in allowed_top_level:
            continue
        if key in _DICT_SECTIONS and isinstance(value, dict) and value:
            return True
        if key in _LIST_SECTIONS and isinstance(value, list) and value:
            return True
        if key == "values_north_stars" and isinstance(value, list) and value:
            return True
    return False


def _model_has_injectable_content(model: UserModel) -> bool:
    """True when the model has at least one section worth injecting into a prompt."""
    if any(str(v).strip() for v in model.identity.values()):
        return True
    if str(model.current_focus.get("description") or "").strip():
        return True
    if any(
        g.get("status") in (None, "active", "in_progress")
        and str(g.get("title") or "").strip()
        for g in model.goals
    ):
        return True
    if any(
        isinstance(rel, dict)
        and (
            str(rel.get("name") or rel.get("who") or "").strip()
            or str(rel.get("role") or rel.get("relationship") or "").strip()
            or str(rel.get("note") or rel.get("context") or "").strip()
        )
        for rel in model.relationships
    ):
        return True
    if any(
        isinstance(conn, dict) and str(conn.get("text") or "").strip()
        for conn in model.connections
    ):
        return True
    if any(str(v).strip() for v in model.preferences.values()):
        return True
    if any(
        str(c.get("text") or "").strip()
        and c.get("status", "open") not in ("done", "closed", "cancelled")
        for c in model.open_commitments
    ):
        return True
    if model.values_north_stars:
        return True
    return False


def load_user_model(workspace: str) -> UserModel:
    """
    Load the current User Model. Returns a default/empty model if none exists.
    Never raises on missing file — returns a fresh model.
    """
    model_path = _resolve_user_model_path(workspace, MODEL_JSON, allow_write=False)
    if not model_path or not os.path.isfile(model_path):
        return UserModel()

    try:
        with open(model_path, "r", encoding="utf-8") as f:
            data = json.load(f)
        if not isinstance(data, dict):
            _log.warning("User Model at %s is not a JSON object; returning empty model", model_path)
            return UserModel()
        return UserModel.from_dict(data)
    except Exception:
        # Fail closed: return empty model rather than corrupting caller state.
        _log.warning("Failed to load User Model from %s; returning empty model", model_path, exc_info=True)
        return UserModel()


def render_for_prompt(workspace: str, max_chars: int = 6000) -> str:
    """
    Compact, high-signal rendering suitable for system prompt injection.
    Phase 0 version is deliberately simple and safe.
    """
    model = load_user_model(workspace)
    if not _model_has_injectable_content(model):
        return ""  # Nothing meaningful yet — do not inject noise

    sections: list[str] = []

    if model.identity:
        sections.append("**Identity**")
        for k, v in model.identity.items():
            if v:
                sections.append(f"- {k}: {v}")

    if model.current_focus and model.current_focus.get("description"):
        sections.append("\n**Current Focus**")
        sections.append(f"- {model.current_focus['description']}")
        if model.current_focus.get("since"):
            sections.append(f"  (since {model.current_focus['since']})")

    if model.goals:
        active_goals = [g for g in model.goals if g.get("status") in (None, "active", "in_progress")]
        if active_goals:
            sections.append("\n**Active Goals**")
            for g in active_goals[:5]:  # Cap for prompt sanity
                title = g.get("title", "Untitled")
                horizon = g.get("horizon", "")
                why = g.get("why_it_matters", "")
                line = f"- {title}"
                if horizon:
                    line += f" ({horizon})"
                if why:
                    line += f" — {why[:120]}"
                sections.append(line)

    if model.relationships:
        sections.append("\n**Relationships**")
        for rel in model.relationships[:5]:
            if not isinstance(rel, dict):
                continue
            name = str(rel.get("name") or rel.get("who") or "").strip()
            role = str(rel.get("role") or rel.get("relationship") or "").strip()
            note = str(rel.get("note") or rel.get("context") or "").strip()
            if not name and not role:
                continue
            line = f"- {name}" if name else "- (unnamed)"
            if role:
                line += f" ({role})"
            if note:
                line += f" — {note[:100]}"
            sections.append(line)

    if model.connections:
        sections.append("\n**Connections & Threads**")
        for conn in model.connections[:5]:
            if not isinstance(conn, dict):
                continue
            text = str(conn.get("text") or "").strip()
            if not text:
                continue
            related = conn.get("related_to")
            if isinstance(related, list):
                topics = [str(t).strip() for t in related if str(t).strip()]
            else:
                topics = []
            line = f"- {text[:160]}"
            if topics:
                line += f" (related: {', '.join(topics[:4])})"
            sections.append(line)

    if model.preferences:
        sections.append("\n**Key Preferences**")
        for k, v in list(model.preferences.items())[:6]:
            sections.append(f"- {k}: {v}")

    if model.open_commitments:
        open_items = [
            c for c in model.open_commitments
            if str(c.get("text") or "").strip() and c.get("status", "open") not in ("done", "closed", "cancelled")
        ]
        if open_items:
            sections.append("\n**Open Commitments**")
            for item in open_items[:5]:
                text = str(item.get("text", "")).strip()
                if text:
                    sections.append(f"- {text[:160]}")

    if model.values_north_stars:
        sections.append("\n**Values & North Stars**")
        for v in model.values_north_stars[:5]:
            if v:
                sections.append(f"- {v}")

    rendered = "\n".join(sections).strip()
    if len(rendered) > max_chars:
        rendered = rendered[: max_chars - 20] + "\n... [truncated]"
    return rendered


_GUARDED_SECTIONS: tuple[str, ...] = (
    "identity",
    "preferences",
    "current_focus",
    "goals",
    "relationships",
    "connections",
    "open_commitments",
    "values_north_stars",
)


def _section_is_populated(model: "UserModel", key: str) -> bool:
    """True when a User Model section currently holds meaningful content."""
    val = getattr(model, key, None)
    if isinstance(val, dict):
        return any(str(v).strip() for v in val.values())
    if isinstance(val, list):
        for item in val:
            if isinstance(item, dict):
                if any(str(v).strip() for v in item.values()):
                    return True
            elif str(item).strip():
                return True
        return False
    return bool(str(val or "").strip())


def _populated_sections(model: "UserModel") -> set[str]:
    return {k for k in _GUARDED_SECTIONS if _section_is_populated(model, k)}


def _is_bulk_clear(populated_before: set[str], populated_after: set[str]) -> bool:
    """A single patch that clears most/all of the previously-populated sections.

    Conservative on purpose (low false-positive): clearing 3+ sections at once, or
    clearing 2+ while leaving at most one populated, counts as a mass-clear. A normal
    edit that blanks a single section (or a few fields within one) never trips this.
    """
    cleared = populated_before - populated_after
    return len(cleared) >= 3 or (len(cleared) >= 2 and len(populated_after) <= 1)


def apply_patch(
    workspace: str,
    patch: dict[str, Any],
    *,
    actor: str,
    provenance: dict[str, Any] | None = None,
    allow_bulk_clear: bool = False,
) -> dict[str, Any]:
    """
    The single blessed mutation path for the User Model.

    Loads existing model, merges known sections (dict deep-merge; list merge-by-id/title),
    writes atomically with 0o600, and records basic audit.

    Callers (chat tool, UI, migration) are responsible for policy checks
    (recent user text for sensitive fields, etc.).
    """
    if not (workspace or "").strip():
        return {"ok": False, "error": "Workspace not available"}

    if not isinstance(patch, dict) or not patch:
        return {"ok": False, "error": "No valid User Model sections in patch"}

    # Settings → You identity editor sends only ``identity``; replace the section so
    # removed fields are not left behind by deep-merge (including clearing all fields).
    replace_identity = actor == "user" and set(patch.keys()) == {"identity"}
    replace_connections = actor == "user" and set(patch.keys()) == {"connections"}
    if (
        not replace_identity
        and not replace_connections
        and not patch_has_applicable_sections(patch)
    ):
        return {"ok": False, "error": "No valid User Model sections in patch"}
    if replace_identity and not isinstance(patch.get("identity"), dict):
        return {"ok": False, "error": "No valid User Model sections in patch"}
    if replace_connections and not isinstance(patch.get("connections"), list):
        return {"ok": False, "error": "No valid User Model sections in patch"}

    model = load_user_model(workspace)
    populated_before = _populated_sections(model)

    allowed_top_level = {
        "identity", "goals", "relationships", "connections", "preferences",
        "current_focus", "values_north_stars", "open_commitments",
    }
    applied_keys: list[str] = []
    for key, value in patch.items():
        if key not in allowed_top_level:
            continue
        section_replace = (replace_identity and key == "identity") or (
            replace_connections and key == "connections"
        )
        if _apply_section_patch(
            model,
            key,
            value,
            replace=section_replace,
        ):
            applied_keys.append(key)

    if not applied_keys:
        return {"ok": False, "error": "No valid User Model sections in patch"}

    # Trust guard: refuse a single-turn mass-clear of the durable model. A direct
    # user edit (actor="user", e.g. the You pane) is inherently consented, and an
    # explicit allow_bulk_clear=True confirms intent — everything else (notably the
    # agent patch_user_model tool) must not blank the user's profile behind their
    # back. This closes the side door where an agent, blocked from a raw file write
    # to the store, wipes it via the blessed patch API instead.
    if not allow_bulk_clear and actor != "user":
        if _is_bulk_clear(populated_before, _populated_sections(model)):
            cleared = sorted(populated_before - _populated_sections(model))
            return {
                "ok": False,
                "error": (
                    "Refused: this update would clear multiple User Model sections "
                    f"({', '.join(cleared)}) at once. Mass-clearing the user's profile "
                    "needs explicit user confirmation — confirm with the user first, "
                    "then retry with allow_bulk_clear=True."
                ),
            }

    now = datetime.now(timezone.utc).isoformat()
    model.last_updated = now
    model.updated_by = actor

    if provenance:
        model.provenance = UserModelProvenance(
            source=provenance.get("source", actor),
            thread_id=provenance.get("thread_id"),
            note=provenance.get("note"),
        )

    model_dict = model.to_dict()

    model_path = _resolve_user_model_path(workspace, MODEL_JSON, allow_write=True)
    if not model_path:
        return {"ok": False, "error": "Failed to resolve safe path for model.json"}

    os.makedirs(os.path.dirname(model_path), exist_ok=True)

    with patch_user_model_write_context():
        try:
            secure_atomic_write(model_path, json.dumps(model_dict, indent=2, ensure_ascii=False))
        except Exception as exc:
            _log.warning("Failed to write User Model JSON", exc_info=True)
            return {"ok": False, "error": f"Failed to write model.json: {exc}"}

        md_path = _resolve_user_model_path(workspace, USER_MODEL_MD, allow_write=True)
        if md_path:
            try:
                secure_atomic_write(md_path, render_markdown(model))
            except Exception:
                _log.warning("User Model markdown projection failed (non-fatal)", exc_info=True)

        try:
            _sync_user_md_projection(workspace, model)
        except Exception:
            _log.warning("USER.md projection failed (non-fatal)", exc_info=True)

    _append_audit(workspace, {
        "timestamp": now,
        "actor": actor,
        "action": "apply_patch",
        "keys_updated": applied_keys,
    })

    return {"ok": True, "last_updated": now}


def render_markdown(model: UserModel) -> str:
    """Human-readable markdown projection of the User Model."""
    return _render_markdown_projection(model)


USER_PROJECTION_MD = f"{USER_MODEL_DIR}/USER_PROJECTION.md"


def _sync_user_md_projection(workspace: str, model: UserModel) -> None:
    """Write a lightweight USER_PROJECTION.md inside user_model/ storage.

    This is a convenience projection — NOT the workspace USER.md (which may
    have hand-edited content we must never clobber).
    """
    ws = os.path.expanduser((workspace or "").strip())
    if not ws:
        return
    path = _resolve_user_model_path(ws, USER_PROJECTION_MD, allow_write=True)
    if not path:
        return
    lines = [
        "# User Model Projection",
        "",
        "_Auto-generated from Settings → You. Edit identity there._",
        "",
    ]
    if model.identity:
        for key in sorted(model.identity.keys()):
            val = str(model.identity.get(key) or "").strip()
            if not val:
                continue
            label = key.replace("_", " ").strip().title()
            lines.append(f"- **{label}:** {val}")
    focus = model.current_focus or {}
    focus_text = str(focus.get("description") or "").strip()
    if focus_text:
        lines.extend(["", "## Current focus", focus_text])
    content = "\n".join(lines).strip() + "\n"
    os.makedirs(os.path.dirname(path), exist_ok=True)
    secure_atomic_write(path, content)


def _render_markdown_projection(model: UserModel) -> str:
    """Human-readable markdown projection kept in sync with render_for_prompt sections."""
    lines = ["# USER MODEL (generated from model.json)\n"]
    if model.last_updated:
        lines.append(f"Last updated: {model.last_updated} by {model.updated_by}\n")

    if model.identity:
        lines.append("## Identity\n")
        for k, v in model.identity.items():
            if v is not None and str(v).strip():
                lines.append(f"- **{k}**: {v}")
        lines.append("")

    if str(model.current_focus.get("description") or "").strip():
        lines.append("## Current Focus\n")
        lines.append(str(model.current_focus.get("description", "")))
        if model.current_focus.get("since"):
            lines.append(f"\n_Since {model.current_focus['since']}_")
        lines.append("")

    active_goals = [g for g in model.goals if g.get("status") in (None, "active", "in_progress")]
    if active_goals:
        lines.append("## Goals\n")
        for g in active_goals[:10]:
            title = g.get("title", "Untitled")
            horizon = g.get("horizon", "")
            why = (g.get("why_it_matters") or "")[:160]
            line = f"- {title}"
            if horizon:
                line += f" ({horizon})"
            if why:
                line += f" — {why}"
            lines.append(line)
        lines.append("")

    if model.relationships:
        lines.append("## Relationships\n")
        for rel in model.relationships[:10]:
            if not isinstance(rel, dict):
                continue
            name = str(rel.get("name") or rel.get("who") or "").strip()
            role = str(rel.get("role") or rel.get("relationship") or "").strip()
            note = str(rel.get("note") or rel.get("context") or "").strip()
            if not name and not role:
                continue
            line = f"- {name}" if name else "- (unnamed)"
            if role:
                line += f" ({role})"
            if note:
                line += f" — {note}"
            lines.append(line)
        lines.append("")

    if model.connections:
        lines.append("## Connections & Threads\n")
        for conn in model.connections[:10]:
            if not isinstance(conn, dict):
                continue
            text = str(conn.get("text") or "").strip()
            if not text:
                continue
            related = conn.get("related_to")
            if isinstance(related, list):
                topics = [str(t).strip() for t in related if str(t).strip()]
            else:
                topics = []
            line = f"- {text}"
            if topics:
                line += f" _(related: {', '.join(topics[:6])})_"
            lines.append(line)
        lines.append("")

    if model.preferences:
        lines.append("## Preferences\n")
        for k, v in list(model.preferences.items())[:12]:
            if v is not None and str(v).strip():
                lines.append(f"- **{k}**: {v}")
        lines.append("")

    open_items = [
        c for c in model.open_commitments
        if str(c.get("text") or "").strip() and c.get("status", "open") not in ("done", "closed", "cancelled")
    ]
    if open_items:
        lines.append("## Open Commitments\n")
        for item in open_items[:10]:
            text = str(item.get("text", "")).strip()
            status = item.get("status", "open")
            lines.append(f"- [{status}] {text[:200]}")
        lines.append("")

    if model.values_north_stars:
        lines.append("## Values & North Stars\n")
        for v in model.values_north_stars[:10]:
            if v:
                lines.append(f"- {v}")
        lines.append("")

    return "\n".join(lines).strip() + "\n"


def _append_audit(workspace: str, entry: dict[str, Any]) -> None:
    audit_path = _resolve_user_model_path(workspace, AUDIT_LOG, allow_write=True)
    if not audit_path:
        return
    os.makedirs(os.path.dirname(audit_path), exist_ok=True)
    line = json.dumps(entry, ensure_ascii=False) + "\n"
    try:
        with audit_write_context():
            with open(audit_path, "a", encoding="utf-8") as f:
                f.write(line)
        try:
            os.chmod(audit_path, 0o600)
        except OSError:
            pass
    except Exception:
        _log.warning("Failed to append User Model audit entry", exc_info=True)


def create_checkpoint(workspace: str, label: str, *, actor: str = "ui") -> dict[str, Any]:
    """Persist a labeled snapshot of the current User Model under checkpoints/."""
    if not (workspace or "").strip():
        return {"ok": False, "error": "Workspace not available"}

    clean_label = (label or "").strip()
    if not clean_label:
        return {"ok": False, "error": "Checkpoint label is required"}

    model = load_user_model(workspace)
    safe_label = re.sub(r"[^A-Za-z0-9._-]+", "-", clean_label)[:80] or "checkpoint"
    checkpoint_id = f"{datetime.now(timezone.utc).strftime('%Y%m%d-%H%M%S')}-{safe_label}"
    ck_path = _resolve_user_model_path(
        workspace,
        f"{CHECKPOINTS_DIR}/{checkpoint_id}.json",
        allow_write=True,
    )
    if not ck_path:
        return {"ok": False, "error": "Failed to resolve checkpoint path"}

    payload = {
        "id": checkpoint_id,
        "label": clean_label,
        "created_at": datetime.now(timezone.utc).isoformat(),
        "actor": actor,
        "model": model.to_dict(),
    }
    os.makedirs(os.path.dirname(ck_path), exist_ok=True)
    try:
        with checkpoint_write_context():
            secure_atomic_write(ck_path, json.dumps(payload, indent=2, ensure_ascii=False))
    except Exception as exc:
        _log.warning("Failed to write User Model checkpoint", exc_info=True)
        return {"ok": False, "error": f"Failed to write checkpoint: {exc}"}

    _append_audit(workspace, {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "actor": actor,
        "action": "create_checkpoint",
        "label": clean_label,
        "checkpoint_id": checkpoint_id,
    })
    return {"ok": True, "label": clean_label, "checkpoint_id": checkpoint_id}


def list_checkpoints(workspace: str) -> list[dict[str, Any]]:
    """Return lightweight metadata for all checkpoints (newest first). Does not include full model snapshots."""
    if not (workspace or "").strip():
        return []

    ck_dir = _resolve_user_model_path(workspace, CHECKPOINTS_DIR, allow_write=False)
    if not ck_dir or not os.path.isdir(ck_dir):
        return []

    results: list[dict[str, Any]] = []
    try:
        for name in sorted(os.listdir(ck_dir), reverse=True):
            if not name.endswith(".json"):
                continue
            try:
                with open(os.path.join(ck_dir, name), "r", encoding="utf-8") as f:
                    data = json.load(f)
                results.append({
                    "id": data.get("id", name.replace(".json", "")),
                    "label": data.get("label", ""),
                    "created_at": data.get("created_at", ""),
                    "actor": data.get("actor", "ui"),
                })
            except Exception:
                continue
    except Exception:
        return []

    return results


def get_checkpoint(workspace: str, checkpoint_id: str) -> dict[str, Any] | None:
    """Return the full stored payload for a specific checkpoint (including the model snapshot at that time)."""
    if not (workspace or "").strip() or not checkpoint_id:
        return None

    ck_dir = _resolve_user_model_path(workspace, CHECKPOINTS_DIR, allow_write=False)
    if not ck_dir or not os.path.isdir(ck_dir):
        return None

    if not _CHECKPOINT_ID_RE.match(checkpoint_id):
        return None

    ck_path = os.path.join(ck_dir, f"{checkpoint_id}.json")
    if not os.path.isfile(ck_path):
        return None

    try:
        with open(ck_path, "r", encoding="utf-8") as f:
            data = json.load(f)
        return data  # contains id, label, created_at, actor, and the full "model" snapshot
    except Exception:
        return None


# ---------------------------------------------------------------------------
# Background fact routing (Slice 0 — personal → proposal, project → MEMORY)
# ---------------------------------------------------------------------------

BackgroundFactRoute = str  # "proposal" | "memory" | "skipped"

_MIN_BACKGROUND_FACT_CHARS = 14  # align with utils.memory_promotion._MIN_ENTRY_CHARS
_MAX_BACKGROUND_FACT_CHARS = 800

_PERSONAL_SIGNAL = re.compile(
    r"\b("
    r"user|human|they|their|them|he|she|prefers?|preference|preferences|"
    r"partner|spouse|family|parent|child|employer|timezone|lives in|based in|"
    r"committed|commitment|goal|goals|wants to|trying to|working toward|"
    r"relationship|values?|north star|allerg|schedule"
    r")\b",
    re.IGNORECASE,
)
_PROJECT_TECHNICAL = re.compile(
    r"\b("
    r"kanban|sqlite|codebase|refactor|implements|module|backend|swiftui|"
    r"dispatcher|storage|pytest|mugi|xcode|fernet|sqlalchemy|docker|"
    r"implementation|architecture|endpoint|api route"
    r")\b",
    re.IGNORECASE,
)
_USER_ANCHORED = re.compile(
    r"\b(user|prefers?|wants to|their|goal|family|partner|employer|timezone|lives in)\b",
    re.IGNORECASE,
)
_GOAL_CUES = re.compile(
    r"\b(goal|goals|wants to|trying to|working toward|aims to|hopes to)\b",
    re.IGNORECASE,
)
_COMMITMENT_CUES = re.compile(
    r"\b(commitment|committed to|promised to|waiting on|follow up on|open loop)\b",
    re.IGNORECASE,
)
_RELATIONSHIP_CUES = re.compile(
    r"\b(partner|spouse|wife|husband|colleague|friend|family|parent|child|sibling)\b",
    re.IGNORECASE,
)
_CONNECTION_THREAD_CUES = re.compile(
    r"\b("
    r"link|links|connect|connection|connections|ties to|tied to|relates to|related to|"
    r"together with|alongside|cross[- ]topic|cross[- ]session|often coincides|"
    r"mentioned together|recurring theme|top of mind|parallel|intersect"
    r")\b",
    re.IGNORECASE,
)
_FOCUS_CUES = re.compile(
    r"\b(current focus|focused on|priority|main thing|primary focus)\b",
    re.IGNORECASE,
)
_PREFERENCE_CUES = re.compile(
    r"\b(prefers?|preference|likes to|likes|avoids|doesn't like|dislikes|usually)\b",
    re.IGNORECASE,
)
_IDENTITY_CUES = re.compile(
    r"\b(timezone|employer|works at|role at|lives in|based in|located in|"
    r"primary editor|uses .+ for|name is)\b",
    re.IGNORECASE,
)
_VALUES_CUES = re.compile(
    r"\b(values?|north star|cares deeply|matters most)\b",
    re.IGNORECASE,
)
_SENSITIVE_PROPOSAL_KEY = re.compile(
    r"(password|secret|token|api[_-]?key|credential|private[_-]?key)",
    re.IGNORECASE,
)


def get_user_model_routing_config() -> dict[str, Any]:
    """Load [user_model] routing knobs (lazy import avoids core cycle at import)."""
    try:
        from core.config import get_user_model_config

        return get_user_model_config()
    except Exception:
        _log.debug("get_user_model_routing_config failed; using defaults", exc_info=True)
        return {
            "background_proposals_enabled": True,
            "max_proposals_per_cycle": 3,
            "dedupe_against_observations": True,
        }


class BackgroundProposalBudget:
    """Per-runner-invocation cap on background User Model proposals."""

    def __init__(self, max_proposals: int | None = None) -> None:
        cfg = get_user_model_routing_config()
        cap = max_proposals if max_proposals is not None else int(cfg.get("max_proposals_per_cycle", 3))
        self.max_proposals = max(0, int(cap))
        self.proposed = 0

    def can_propose(self) -> bool:
        return self.proposed < self.max_proposals

    def consume(self, source: str = "") -> bool:
        del source  # reserved for per-source caps later
        if not self.can_propose():
            return False
        self.proposed += 1
        return True


def _normalize_background_fact_line(line: str) -> str:
    from utils.noticing_plaintext import sanitize_noticing_plaintext

    text = sanitize_noticing_plaintext(str(line or "").strip(), max_len=800)
    text = re.sub(r"^\s*[-*]\s+", "", text)
    text = re.sub(r"^\s*\(\s*as of [^)]+\)\s*", "", text, flags=re.IGNORECASE).strip()
    return text.strip()


def _slugify_field_key(text: str, *, max_len: int = 40) -> str:
    slug = re.sub(r"[^a-z0-9]+", "_", str(text or "").lower()).strip("_")
    return (slug or "general")[:max_len]


def is_personal_fact_line(line: str) -> bool:
    """True when *line* reads as a personal-signal fact (vs project/work MEMORY)."""
    text = _normalize_background_fact_line(line)
    if len(text) < _MIN_BACKGROUND_FACT_CHARS:
        return False
    if _PROJECT_TECHNICAL.search(text) and not _USER_ANCHORED.search(text):
        return False
    if _PERSONAL_SIGNAL.search(text):
        return True
    try:
        from utils.memory_promotion import _STABILITY_HINTS  # noqa: PLC2701

        if _STABILITY_HINTS.search(text) and _USER_ANCHORED.search(text):
            return True
    except Exception:
        pass
    return False


def build_proposal_changes_from_fact(
    line: str,
    *,
    section_hint: str | None = None,
) -> dict[str, Any] | None:
    """Map a personal-fact bullet to a typed User Model proposal patch, or None if project-only."""
    text = _normalize_background_fact_line(line)[:_MAX_BACKGROUND_FACT_CHARS]
    if len(text) < _MIN_BACKGROUND_FACT_CHARS:
        return None

    if section_hint == "connections":
        from utils.connections_journal import extract_related_topics

        return {
            "connections": [
                {
                    "text": text[:400],
                    "related_to": extract_related_topics(text),
                    "noted_at": datetime.now(timezone.utc).isoformat(),
                    "source": "background_noticer",
                }
            ]
        }

    if not is_personal_fact_line(text):
        return None

    if _COMMITMENT_CUES.search(text):
        return {"open_commitments": [{"text": text, "status": "open"}]}

    if _GOAL_CUES.search(text):
        title = text
        m = re.search(r"(?:goal(?:s)?(?:\s+to)?|wants to|trying to|working toward)\s+(.+)", text, re.IGNORECASE)
        if m and len(m.group(1).strip()) >= 8:
            title = m.group(1).strip().rstrip(".")
        return {
            "goals": [
                {
                    "title": title[:300],
                    "status": "active",
                    "why_it_matters": text[:400],
                }
            ]
        }

    if _RELATIONSHIP_CUES.search(text):
        return {
            "relationships": [
                {
                    "note": text[:400],
                }
            ]
        }

    if _CONNECTION_THREAD_CUES.search(text):
        from utils.connections_journal import extract_related_topics

        return {
            "connections": [
                {
                    "text": text[:400],
                    "related_to": extract_related_topics(text),
                    "noted_at": datetime.now(timezone.utc).isoformat(),
                    "source": "background_noticer",
                }
            ]
        }

    if _FOCUS_CUES.search(text):
        return {"current_focus": {"description": text[:500]}}

    if _PREFERENCE_CUES.search(text):
        key = "general"
        m = re.search(r"prefers?\s+(.+)", text, re.IGNORECASE)
        if m:
            key = _slugify_field_key(m.group(1).split(".")[0][:60])
        elif re.search(r"\bavoids?\b", text, re.IGNORECASE):
            key = "avoid"
        return {"preferences": {key: text[:400]}}

    if _VALUES_CUES.search(text):
        return {"values_north_stars": [text[:400]]}

    if _IDENTITY_CUES.search(text):
        key = "note"
        if re.search(r"\btimezone\b", text, re.IGNORECASE):
            key = "timezone"
        elif re.search(r"\b(employer|works at|role at)\b", text, re.IGNORECASE):
            key = "employer"
        elif re.search(r"\b(lives in|based in|located in)\b", text, re.IGNORECASE):
            key = "location"
        elif re.search(r"\bprimary editor\b", text, re.IGNORECASE):
            key = "primary_editor"
        return {"identity": {key: text[:400]}}

    # Personal signal but no finer bucket — narrative thread is safer than a
    # single catch-all preferences slot (background_noticed starved the model).
    from utils.connections_journal import extract_related_topics

    return {
        "connections": [
            {
                "text": text[:400],
                "related_to": extract_related_topics(text),
                "noted_at": datetime.now(timezone.utc).isoformat(),
                "source": "background_noticer",
            }
        ]
    }


def _proposal_payload_safe(changes: dict[str, Any]) -> bool:
    """Reject proposal patches that carry secret-like keys or values."""
    from utils.memory_promotion import _SECRET_PATTERNS  # noqa: PLC2701

    def _check_value(val: Any) -> bool:
        if isinstance(val, str):
            text = val.strip()
            if not text:
                return True
            for pattern in _SECRET_PATTERNS:
                if pattern.search(text):
                    return False
            return True
        if isinstance(val, dict):
            for k, v in val.items():
                if _SENSITIVE_PROPOSAL_KEY.search(str(k)):
                    return False
                if not _check_value(v):
                    return False
            return True
        if isinstance(val, list):
            return all(_check_value(item) for item in val)
        return True

    if not isinstance(changes, dict):
        return False
    for key, value in changes.items():
        if _SENSITIVE_PROPOSAL_KEY.search(str(key)):
            return False
        if not _check_value(value):
            return False
    return True



def _normalize_connection_headline(text: str) -> str:
    from utils.noticing_dedupe import normalize_connection_headline

    return normalize_connection_headline(text)


def connection_topic_pair(text: str) -> tuple[str, str] | None:
    from utils.noticing_dedupe import connection_topic_pair as _pair

    return _pair(text)


def _connection_pair_signatures_from_texts(texts: list[str]) -> set[tuple[str, str]]:
    from utils.noticing_dedupe import connection_pair_signatures_from_texts

    return connection_pair_signatures_from_texts(texts)


def _collect_model_text_phrases(model: UserModel) -> set[str]:
    phrases: set[str] = set()

    def _add(text: Any) -> None:
        norm = str(text or "").strip().lower()
        if len(norm) >= _MIN_BACKGROUND_FACT_CHARS:
            phrases.add(norm)

    for val in model.identity.values():
        _add(val)
    for val in model.preferences.values():
        _add(val)
    for g in model.goals:
        if isinstance(g, dict):
            _add(g.get("title"))
            _add(g.get("why_it_matters"))
    for c in model.open_commitments:
        if isinstance(c, dict):
            _add(c.get("text"))
    for r in model.relationships:
        if isinstance(r, dict):
            _add(r.get("note"))
            _add(r.get("name"))
    for conn in model.connections:
        if isinstance(conn, dict):
            _add(conn.get("text"))
    if isinstance(model.current_focus, dict):
        _add(model.current_focus.get("description"))
    for v in model.values_north_stars:
        _add(v)
    return phrases


def _collect_archived_proposal_phrases(workspace: str, *, status: str = "rejected") -> set[str]:
    """Collect text from non-pending proposal files (e.g. rejected) for dedupe."""
    props_dir = _resolve_user_model_path(workspace, PROPOSALS_DIR, allow_write=False)
    if not props_dir or not os.path.isdir(props_dir):
        return set()
    phrases: set[str] = set()
    for name in os.listdir(props_dir):
        if not name.endswith(".proposal.json"):
            continue
        try:
            with open(os.path.join(props_dir, name), "r", encoding="utf-8") as f:
                data = json.load(f)
            if not isinstance(data, dict):
                continue
            if str(data.get("status") or "") != status:
                continue
            phrases |= _collect_proposal_text_phrases([data])
        except Exception:
            _log.debug("Skipping unreadable archived proposal %s", name, exc_info=True)
            continue
    return phrases


def _collect_proposal_text_phrases(proposals: list[dict[str, Any]]) -> set[str]:
    phrases: set[str] = set()

    def _walk(val: Any) -> None:
        if isinstance(val, str):
            norm = val.strip().lower()
            if len(norm) >= _MIN_BACKGROUND_FACT_CHARS:
                phrases.add(norm)
        elif isinstance(val, dict):
            for v in val.values():
                _walk(v)
        elif isinstance(val, list):
            for item in val:
                _walk(item)

    for prop in proposals:
        if not isinstance(prop, dict):
            continue
        _walk(prop.get("changes"))
        rationale = str(prop.get("rationale") or "").strip().lower()
        if len(rationale) >= _MIN_BACKGROUND_FACT_CHARS:
            phrases.add(rationale)
    return phrases


def _proposal_write_dedupe_pool(workspace: str) -> list[str] | None:
    """Candidate texts for blocking duplicate proposal writes. None = fail-closed."""
    try:
        model = load_user_model(workspace)
        pool = _collect_model_text_phrases(model)
        pool |= _collect_proposal_text_phrases(list_proposals(workspace))
        pool |= _collect_archived_proposal_phrases(workspace, status="rejected")
        obs_phrases = _pending_observation_noticed_phrases(workspace)
        if obs_phrases is None:
            return None
        pool |= obs_phrases
        return list(pool)
    except Exception:
        _log.debug("proposal write dedupe pool failed", exc_info=True)
        return None


def filter_proposals_for_you_pane(
    workspace: str,
    proposals: list[dict[str, Any]] | None = None,
) -> list[dict[str, Any]]:
    """Return pending proposals collapsed for You pane display and badge counts."""
    from utils.noticing_dedupe import (
        PROPOSAL_NEAR_DUPLICATE_JACCARD,
        dedupe_proposals_for_display,
    )

    pending = list_proposals(workspace) if proposals is None else proposals
    model = load_user_model(workspace)
    baseline = list(_collect_model_text_phrases(model))
    return dedupe_proposals_for_display(
        pending,
        baseline_texts=baseline,
        min_jaccard=PROPOSAL_NEAR_DUPLICATE_JACCARD,
    )


def _pending_observation_noticed_phrases(workspace: str) -> set[str] | None:
    del workspace  # queue is global under MUGI_DIR
    cfg = get_user_model_routing_config()
    if not bool(cfg.get("dedupe_against_observations", True)):
        return set()
    try:
        from persistence import PROACTIVE_QUEUE_PATH
        from utils.proactive_queue import pending_entries_from_lines

        queue_path = os.path.expanduser(PROACTIVE_QUEUE_PATH)
        if not os.path.isfile(queue_path):
            return set()
        with open(queue_path, "r", encoding="utf-8") as f:
            lines = f.readlines()
        phrases: set[str] = set()
        for entry in pending_entries_from_lines(lines, include_notified=True):
            obs = entry.get("observation") if isinstance(entry.get("observation"), dict) else {}
            noticed = str(obs.get("noticed") or "").strip().lower()
            if len(noticed) >= _MIN_BACKGROUND_FACT_CHARS:
                phrases.add(noticed)
            content = str(entry.get("content") or "").strip().lower()
            if len(content) >= _MIN_BACKGROUND_FACT_CHARS:
                phrases.add(content)
        return phrases
    except Exception:
        _log.warning("observation dedupe scan failed; skipping proposal (fail-closed)", exc_info=True)
        return None


def _is_duplicate_background_fact(workspace: str, norm_line: str) -> bool:
    if not norm_line:
        return True
    try:
        model = load_user_model(workspace)
        existing = _collect_model_text_phrases(model)
        if norm_line in existing:
            return True
        pending = list_proposals(workspace)
        proposal_phrases = _collect_proposal_text_phrases(pending)
        rejected_phrases = _collect_archived_proposal_phrases(workspace, status="rejected")
        if norm_line in proposal_phrases or norm_line in rejected_phrases:
            return True
        obs_phrases = _pending_observation_noticed_phrases(workspace)
        if obs_phrases is None:
            return True
        if norm_line in obs_phrases:
            return True
        combined = existing | proposal_phrases | rejected_phrases | obs_phrases
        for phrase in combined:
            if len(phrase) >= 20 and (norm_line in phrase or phrase in norm_line):
                return True
        new_pair = connection_topic_pair(norm_line)
        if new_pair:
            existing_pairs = _connection_pair_signatures_from_texts(list(existing))
            for prop in pending:
                if not isinstance(prop, dict):
                    continue
                changes = prop.get("changes")
                if not isinstance(changes, dict):
                    continue
                for conn in changes.get("connections") or []:
                    if isinstance(conn, dict):
                        existing_pairs |= _connection_pair_signatures_from_texts(
                            [str(conn.get("text") or "")]
                        )
            props_dir = _resolve_user_model_path(workspace, PROPOSALS_DIR, allow_write=False)
            if props_dir and os.path.isdir(props_dir):
                for name in os.listdir(props_dir):
                    if not name.endswith(".proposal.json"):
                        continue
                    try:
                        with open(os.path.join(props_dir, name), "r", encoding="utf-8") as f:
                            archived = json.load(f)
                        if not isinstance(archived, dict):
                            continue
                        if str(archived.get("status") or "") != "rejected":
                            continue
                        changes = archived.get("changes")
                        if not isinstance(changes, dict):
                            continue
                        for conn in changes.get("connections") or []:
                            if isinstance(conn, dict):
                                existing_pairs |= _connection_pair_signatures_from_texts(
                                    [str(conn.get("text") or "")]
                                )
                    except Exception:
                        continue
            if new_pair in existing_pairs:
                return True
        from utils.noticing_dedupe import PROPOSAL_NEAR_DUPLICATE_JACCARD, is_near_duplicate_text

        jaccard = PROPOSAL_NEAR_DUPLICATE_JACCARD
        connection_texts = [
            str(conn.get("text") or "")
            for conn in model.connections
            if isinstance(conn, dict)
        ]
        for prop in pending:
            if not isinstance(prop, dict):
                continue
            changes = prop.get("changes")
            if not isinstance(changes, dict):
                continue
            for conn in changes.get("connections") or []:
                if isinstance(conn, dict):
                    connection_texts.append(str(conn.get("text") or ""))
        if connection_texts and is_near_duplicate_text(norm_line, connection_texts, min_jaccard=jaccard):
            return True
        if is_near_duplicate_text(norm_line, list(existing), min_jaccard=jaccard):
            return True
        if is_near_duplicate_text(norm_line, list(rejected_phrases), min_jaccard=jaccard):
            return True
        if is_near_duplicate_text(norm_line, list(proposal_phrases), min_jaccard=jaccard):
            return True
        obs_noticed, obs_work = [], []
        try:
            from persistence import PROACTIVE_QUEUE_PATH
            from utils.noticing_dedupe import collect_pending_observation_notices
            from utils.proactive_queue import pending_entries_from_lines

            queue_path = os.path.expanduser(PROACTIVE_QUEUE_PATH)
            if os.path.isfile(queue_path):
                with open(queue_path, "r", encoding="utf-8") as f:
                    pending_obs = [
                        e
                        for e in pending_entries_from_lines(f.readlines(), include_notified=True)
                        if str(e.get("kind") or "") == "observation"
                    ]
                obs_noticed, obs_work = collect_pending_observation_notices(pending_obs)
        except Exception:
            obs_noticed, obs_work = [], []
        if obs_noticed:
            from utils.noticing_dedupe import is_duplicate_observation_noticed

            if is_duplicate_observation_noticed(norm_line, "", existing_noticed=obs_noticed, existing_work=obs_work):
                return True
    except Exception:
        _log.warning(
            "background fact dedupe check failed for %r; skipping proposal (fail-closed)",
            norm_line[:80],
            exc_info=True,
        )
        return True
    return False


def propose_background_fact(
    workspace: str,
    line: str,
    *,
    source: str,
    rationale_prefix: str,
    budget: BackgroundProposalBudget | None = None,
    section_hint: str | None = None,
) -> BackgroundFactRoute:
    """Route one background bullet to User Model proposal, MEMORY, or skip."""
    if not (workspace or "").strip():
        return "skipped"

    text = _normalize_background_fact_line(line)
    if len(text) < _MIN_BACKGROUND_FACT_CHARS:
        return "skipped"
    if len(text) > _MAX_BACKGROUND_FACT_CHARS:
        text = text[:_MAX_BACKGROUND_FACT_CHARS]

    from utils.memory_promotion import PromotionClass, classify_memory_entry

    decision = classify_memory_entry(text)
    if decision.classification == PromotionClass.BLOCKED:
        return "skipped"
    if section_hint != "connections" and decision.classification == PromotionClass.DISCOURAGED:
        return "skipped"

    norm = text.lower()
    if _is_duplicate_background_fact(workspace, norm):
        return "skipped"

    changes = build_proposal_changes_from_fact(text, section_hint=section_hint)
    if changes is None:
        return "memory"

    cfg = get_user_model_routing_config()
    if not bool(cfg.get("background_proposals_enabled", True)):
        _log.warning(
            "background proposals disabled; skipping personal fact (not routing to MEMORY) source=%s",
            source,
        )
        return "skipped"

    if not _proposal_payload_safe(changes):
        _log.warning(
            "background fact proposal blocked by payload safety source=%s line=%r",
            source,
            text[:120],
        )
        return "skipped"

    if budget is None:
        budget = BackgroundProposalBudget()
    if not budget.consume(source):
        return "skipped"

    if section_hint == "connections":
        rationale = text[:280]
    else:
        rationale = f"{(rationale_prefix or 'Background noticed').strip()}: {text[:220]}"
    prop_id = create_proposal(workspace, changes, rationale, source=source)
    if prop_id:
        _log.info(
            "background fact proposal created source=%s id=%s sections=%s",
            source,
            prop_id,
            list(changes.keys()),
        )
        return "proposal"

    _log.warning("background fact proposal write failed source=%s", source)
    if budget.proposed > 0:
        budget.proposed -= 1
    return "skipped"


def propose_background_facts_batch(
    workspace: str,
    lines: list[str] | str,
    *,
    source: str,
    rationale_prefix: str,
    budget: BackgroundProposalBudget | None = None,
    memory_writer: Callable[[str], bool | None] | None = None,
) -> dict[str, int]:
    """Route multiple bullets; returns counts ``{proposed, memory, skipped}``.

    Note: ``propose_background_fact`` returns ``"proposal"`` (singular); this batch
    aggregator maps that to the ``proposed`` count key. When *memory_writer* is set,
    lines routed to ``"memory"`` are passed to it (e.g. ``append_to_memory``).
    If the writer returns ``False`` or raises, the line is counted as ``skipped``.
    """
    if isinstance(lines, str):
        raw_lines = [ln.strip() for ln in lines.splitlines()]
    else:
        raw_lines = [str(ln or "").strip() for ln in lines]

    if budget is None:
        budget = BackgroundProposalBudget()

    counts = {"proposed": 0, "memory": 0, "skipped": 0}
    for raw in raw_lines:
        line = _normalize_background_fact_line(raw)
        if not line or line.upper() in ("NONE", "HEARTBEAT_OK"):
            continue
        outcome = propose_background_fact(
            workspace,
            line,
            source=source,
            rationale_prefix=rationale_prefix,
            budget=budget,
        )
        outcome_key = "proposed" if outcome == "proposal" else outcome
        if outcome_key in counts:
            counts[outcome_key] += 1
        else:
            counts["skipped"] += 1
        if outcome == "memory" and memory_writer is not None:
            try:
                wrote = memory_writer(line)
                if wrote is False:
                    counts["memory"] -= 1
                    counts["skipped"] += 1
            except Exception:
                counts["memory"] -= 1
                counts["skipped"] += 1
                _log.warning(
                    "background fact memory write failed source=%s line=%r",
                    source,
                    line[:120],
                    exc_info=True,
                )
    if any(counts.values()):
        _log.info(
            "background facts batch source=%s proposed=%d memory=%d skipped=%d",
            source,
            counts["proposed"],
            counts["memory"],
            counts["skipped"],
        )
    return counts


# ---------------------------------------------------------------------------
# Proposal scaffolding (Phase 0 — minimal, not yet wired to agent tools)
# ---------------------------------------------------------------------------

def create_proposal(workspace: str, changes: dict[str, Any], rationale: str, source: str) -> str | None:
    """Create a proposal file. Returns proposal id or None on failure/duplicate.

    Back-compatible thin wrapper over :func:`create_proposal_detailed`. Callers
    that need to distinguish "a near-duplicate is already pending" from a real
    failure (e.g. the chat ``propose_user_model_update`` tool, so the model does
    not retry-loop on an opaque error) should call ``create_proposal_detailed``.
    """
    result = create_proposal_detailed(workspace, changes, rationale, source)
    return result.get("id") if result.get("status") == "created" else None


# Proposal-create outcome statuses (create_proposal_detailed).
PROPOSAL_CREATE_CREATED = "created"
PROPOSAL_CREATE_DUPLICATE = "duplicate"
PROPOSAL_CREATE_REJECTED = "rejected"


def create_proposal_detailed(
    workspace: str, changes: dict[str, Any], rationale: str, source: str
) -> dict[str, Any]:
    """Create a proposal, returning a structured outcome.

    Returns a dict with ``status`` one of ``created`` / ``duplicate`` /
    ``rejected``, plus ``id`` (when created) and ``reason`` (a short, model- and
    operator-readable explanation). The ``duplicate`` status lets the caller tell
    the agent "a similar proposal is already pending; do not retry" instead of
    surfacing an opaque failure that induces a retry loop.
    """
    def _rejected(reason: str) -> dict[str, Any]:
        return {"status": PROPOSAL_CREATE_REJECTED, "id": None, "reason": reason}

    if not (workspace or "").strip():
        return _rejected("workspace unavailable")
    if not isinstance(changes, dict) or not changes:
        return _rejected("changes must be a non-empty dict of User Model sections")
    if not patch_has_applicable_sections(changes):
        return _rejected("no well-formed User Model section in changes")
    if not (rationale or "").strip():
        return _rejected("rationale is required")
    if not _proposal_payload_safe(changes):
        _log.warning(
            "User Model proposal blocked by payload safety source=%s sections=%s",
            source,
            list(changes.keys()),
        )
        return _rejected("payload blocked by safety check (looks secret-like)")

    from utils.noticing_dedupe import (
        PROPOSAL_NEAR_DUPLICATE_JACCARD,
        collect_proposal_texts,
        is_near_duplicate_text,
    )

    dedupe_pool = _proposal_write_dedupe_pool(workspace)
    if dedupe_pool is None:
        _log.warning("proposal dedupe check failed; suppressing write (fail-closed)")
        return _rejected("could not verify proposal uniqueness; not written (fail-closed)")
    probe = {
        "changes": changes,
        "rationale": rationale,
    }
    for text in collect_proposal_texts(probe):
        if is_near_duplicate_text(
            text,
            dedupe_pool,
            min_jaccard=PROPOSAL_NEAR_DUPLICATE_JACCARD,
        ):
            _log.info(
                "User Model proposal suppressed as duplicate source=%s text=%r",
                source,
                text[:120],
            )
            return {
                "status": PROPOSAL_CREATE_DUPLICATE,
                "id": None,
                "reason": (
                    "a near-duplicate proposal is already pending the user's review"
                ),
            }

    safe_source = re.sub(r"[^A-Za-z0-9._-]+", "-", (source or "unknown").strip())[:40] or "unknown"
    prop_id = f"{datetime.now(timezone.utc).strftime('%Y%m%d-%H%M%S')}-{safe_source}-{uuid.uuid4().hex[:8]}"
    prop_path = _resolve_user_model_path(workspace, f"{PROPOSALS_DIR}/{prop_id}.proposal.json", allow_write=True)
    if not prop_path:
        return _rejected("could not resolve proposal storage path")

    os.makedirs(os.path.dirname(prop_path), exist_ok=True)

    payload = {
        "id": prop_id,
        "created_at": datetime.now(timezone.utc).isoformat(),
        "source": source,
        "rationale": rationale,
        "changes": changes,
        "status": "pending",
    }
    try:
        with proposal_write_context():
            secure_atomic_write(prop_path, json.dumps(payload, indent=2))
    except Exception:
        _log.warning("Failed to write User Model proposal %s", prop_id, exc_info=True)
        return _rejected("storage write failed")
    return {"status": PROPOSAL_CREATE_CREATED, "id": prop_id, "reason": ""}


def list_proposals(workspace: str) -> list[dict[str, Any]]:
    """Return list of pending proposals (Phase 0 — simple file scan)."""
    props_dir = _resolve_user_model_path(workspace, PROPOSALS_DIR, allow_write=False)
    if not props_dir or not os.path.isdir(props_dir):
        return []

    results: list[dict[str, Any]] = []
    for name in sorted(os.listdir(props_dir)):
        if not name.endswith(".proposal.json"):
            continue
        try:
            with open(os.path.join(props_dir, name), "r", encoding="utf-8") as f:
                data = json.load(f)
            if not isinstance(data, dict):
                continue
            prop_id = str(data.get("id") or "").strip()
            expected_id = name[: -len(".proposal.json")]
            if not prop_id:
                prop_id = expected_id
                data["id"] = prop_id
            elif prop_id != expected_id:
                _log.warning("Skipping proposal file with id/filename mismatch: %s", name)
                continue
            if data.get("status") != "pending":
                continue
            changes = data.get("changes")
            if not isinstance(changes, dict) or not patch_has_applicable_sections(changes):
                _log.warning("Skipping pending proposal with invalid changes: %s", name)
                continue
            results.append(data)
        except Exception:
            _log.debug("Skipping unreadable proposal file %s", name, exc_info=True)
            continue
    return results


def accept_proposal(workspace: str, proposal_id: str, user_note: str | None = None) -> dict[str, Any]:
    """Accept a pending proposal and apply its changes to the live User Model."""
    id_err = _validate_proposal_id(proposal_id)
    if id_err:
        return {"ok": False, "error": id_err}

    prop_path = _resolve_user_model_path(workspace, f"{PROPOSALS_DIR}/{proposal_id}.proposal.json", allow_write=True)
    if not prop_path or not os.path.isfile(prop_path):
        return {"ok": False, "error": "Proposal not found"}

    try:
        with open(prop_path, "r", encoding="utf-8") as f:
            proposal = json.load(f)

        if not isinstance(proposal, dict):
            return {"ok": False, "error": "Proposal file is corrupt"}

        if proposal.get("id") and proposal.get("id") != proposal_id:
            return {"ok": False, "error": "Proposal id mismatch"}

        if proposal.get("status") != "pending":
            return {"ok": False, "error": f"Proposal is not pending (status: {proposal.get('status')})"}

        changes = proposal.get("changes")
        if not isinstance(changes, dict) or not changes:
            return {"ok": False, "error": "Proposal has no valid changes"}
        if not patch_has_applicable_sections(changes):
            return {"ok": False, "error": "Proposal changes are not well-formed"}

        import copy

        proposal_backup = copy.deepcopy(proposal)

        # Mark accepted before apply_patch so a failed status write cannot leave split-brain.
        proposal["status"] = "accepted"
        proposal["accepted_at"] = datetime.now(timezone.utc).isoformat()
        if user_note:
            proposal["user_note"] = user_note

        try:
            with proposal_write_context():
                secure_atomic_write(prop_path, json.dumps(proposal, indent=2))
        except Exception as exc:
            _log.warning("Failed to mark proposal %s accepted before apply", proposal_id, exc_info=True)
            return {"ok": False, "error": f"Failed to accept proposal: {exc}"}

        result = apply_patch(
            workspace,
            changes,
            actor="proposal_accepted",
            provenance={
                "source": "proposal_accepted",
                "thread_id": None,
                "note": user_note or proposal.get("rationale"),
            },
        )

        if not result.get("ok"):
            try:
                with proposal_write_context():
                    secure_atomic_write(prop_path, json.dumps(proposal_backup, indent=2))
            except Exception:
                _log.error(
                    "Proposal %s marked accepted but apply_patch failed and rollback failed",
                    proposal_id,
                    exc_info=True,
                )
            return {"ok": False, "error": result.get("error", "Failed to apply proposal changes")}

        _append_audit(workspace, {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "actor": "user",
            "action": "accept_proposal",
            "proposal_id": proposal_id,
        })

        return {"ok": True, "proposal_id": proposal_id}

    except Exception as e:
        return {"ok": False, "error": f"Failed to accept proposal: {str(e)}"}


def reject_proposal(workspace: str, proposal_id: str, user_note: str | None = None) -> dict[str, Any]:
    """Reject a pending proposal."""
    id_err = _validate_proposal_id(proposal_id)
    if id_err:
        return {"ok": False, "error": id_err}

    prop_path = _resolve_user_model_path(workspace, f"{PROPOSALS_DIR}/{proposal_id}.proposal.json", allow_write=True)
    if not prop_path or not os.path.isfile(prop_path):
        return {"ok": False, "error": "Proposal not found"}

    try:
        with open(prop_path, "r", encoding="utf-8") as f:
            proposal = json.load(f)

        if not isinstance(proposal, dict):
            return {"ok": False, "error": "Proposal file is corrupt"}

        if proposal.get("id") and proposal.get("id") != proposal_id:
            return {"ok": False, "error": "Proposal id mismatch"}

        if proposal.get("status") != "pending":
            return {"ok": False, "error": f"Proposal is not pending (status: {proposal.get('status')})"}

        proposal["status"] = "rejected"
        proposal["rejected_at"] = datetime.now(timezone.utc).isoformat()
        if user_note:
            proposal["user_note"] = user_note

        try:
            with proposal_write_context():
                secure_atomic_write(prop_path, json.dumps(proposal, indent=2))
        except Exception as exc:
            _log.warning("Failed to mark proposal %s rejected", proposal_id, exc_info=True)
            return {"ok": False, "error": f"Failed to reject proposal: {exc}"}

        _append_audit(workspace, {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "actor": "user",
            "action": "reject_proposal",
            "proposal_id": proposal_id,
        })

        return {"ok": True, "proposal_id": proposal_id}

    except Exception as e:
        return {"ok": False, "error": f"Failed to reject proposal: {str(e)}"}


# Note: Background systems can call create_proposal directly.
# Agent tools for propose_user_model_update / list_pending_proposals are in tools/file_ops.py.