"""Drive engine: persistent, escalating motivation state for Mugi."""
from __future__ import annotations

import json
import logging
import os
import re
import shutil
import threading
import uuid
import hashlib
from datetime import datetime, timedelta, timezone
from typing import Any

from persistence import secure_atomic_write
from utils.llm_config_validate import validated_response_format, validated_stop
from utils.llm_output_sanitize import strip_markdown_fences, strip_think_tags, strip_untagged_cot

_log = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# Heartbeat / proactive thread_id contract (agent_loop.run_raw_agent)
# ---------------------------------------------------------------------------
# ``_is_heartbeat`` is True when thread_id equals BACKGROUND_HEARTBEAT_THREAD_ID
# or starts with HEARTBEAT_THREAD_PREFIX (e.g. ``heartbeat_memory_maintenance``).
# Prefixes like ``drive_discovery_*`` and ``messaging_*`` do not match that rule and
# are intentionally separate execution contexts (not treated as maintenance heartbeats).
#
# DRIVE_ACTION_THREAD_PREFIX is the stable id for drive proactive action cycles
# (mugi_runner); agent_loop uses startswith(DRIVE_ACTION_THREAD_PREFIX) for tool
# allowlists and caps. That id also matches HEARTBEAT_THREAD_PREFIX, so it is
# still a heartbeat for generic prompt/tool rules.
BACKGROUND_HEARTBEAT_THREAD_ID = "background_heartbeat"
HEARTBEAT_THREAD_PREFIX = "heartbeat_"
DRIVE_ACTION_THREAD_PREFIX = "heartbeat_drive_action"

_MAX_INPUT_CHARS = 20_000
_MAX_PROMPT_DRIVES = 2
_DEFAULT_LEASE_TTL_SECONDS = 15 * 60
_DEFAULT_BLOCK_RETRY_MINUTES = 30
_VALID_DRIVE_TYPES = {
    "capability_gap",
    "unresolved_question",
    "pattern_automation",
    "quality_issue",
    "exploration",
}
_DRIVE_STATE_LOCK = threading.RLock()
_DRIVE_ALLOWED_KEYS = {"type", "description", "source", "source_detail", "suggested_action"}
_ACTIVE_STATUSES = {"active", "in_progress", "blocked"}
_PROGRESS_IDEMPOTENCY_LOCK = threading.RLock()
_PROGRESS_IDEMPOTENCY_KEYS: set[str] = set()

# ===========================================================================
# PRIVACY / SECURITY  —  Operator Appendix
# ===========================================================================
# Every sink or data-flow path that carries drive-related content.
# Keep each entry to 1-2 lines; see the referenced modules for details.
#
#  1. drives.json, drive_attempt_journal.jsonl — on-disk in workspace, 0o600.
#  2. memory/growth-events.md — drive resolutions and growth narratives.
#  3. proactive_queue.jsonl — SSE previews pushed to the Swift UI.
#  4. background_events.log — readable via socket get_recent_events and MCP
#     log resources.
#  5. macOS Notification Center — lock-screen-visible previews.
#  6. Discovery _llm_call → sends memory tails to [drives] endpoint (or main
#     LLM when discovery_use_main_llm=true).
#  7. get_drive_prompt_block → injects drive text into the main chat system
#     prompt → main LLM backend.
#  8. list_drives tool → full drive records appear in chat context / session
#     traces.
#  9. Session-search FTS — background threads indexed; configurable exclusion
#     via [session_search].exclude_background_threads.
# 10. Fine-tuning export — includes all thread_ids by default; use explicit
#     thread filters to exclude drive threads.
# 11. ~/.mugi/evolution/trajectories — Fernet-encrypted JSONL containing
#     active_drives_at_start.
# 12. [debug] trace_conversations JSONL under ~/.mugi/logs/traces.
# 13. MCP mugi_get_conversation / mugi_read_trace / mugi_search_sessions can
#     surface background thread_ids.
# 14. MCP mugi://config/agent — now redacted, but MCP clients are
#     high-trust.
# 15. Live View SSE /live_view/stream — requires auth when auth_token is set;
#     wide open without it.
# 16. scripts/replay_trace.py — offline trace replay (portable leakage
#     surface).
# 17. scheduled_results/ — cron-job prompts and results at 0o600.
# 18. ~/.mugi/checkpoints — round checkpoints for non-subagent threads
#     (messaging, main chat).
# 19. UTC budget boundaries (not local midnight).
# 20. autonomy_kill_switch — emergency stop for autonomous drive_action.
# 21. Socket MUGI_RUNNER_SOCKET_SECRET — env-based, visible to same-UID
#     processes.
# 22. Web tools on bounded drive_action allowlist — network egress allowed.
# 23. TLS for discovery endpoints — prefer HTTPS.
# 24. [subagent] enabled does NOT disable discovery LLM or runner heartbeats.
# 25. Runner interval snapshot at startup vs per-turn reload_config() inside
#     run_raw_agent.
# 26. Messaging gateway — full main-agent path, remote-triggered, media tier
#     with attachments.
# ===========================================================================


def _normalize_source_label(raw: str) -> str:
    text = (raw or "").strip().lower()
    if not text:
        return ""
    text = text.replace("\\", "/")
    text = re.sub(r"\s+", "", text)
    text = text.lstrip("./")
    aliases = {
        "insights.md": "memory/insights.md",
        "memory/insights.md": "memory/insights.md",
        "recurring-patterns.md": "memory/recurring-patterns.md",
        "memory/recurring-patterns.md": "memory/recurring-patterns.md",
        "tool_errors.json": "memory/tool_errors.json",
        "memory/tool_errors.json": "memory/tool_errors.json",
        "feedback.md": "memory/feedback.md",
        "memory/feedback.md": "memory/feedback.md",
        "proactive-ideas.md": "memory/proactive-ideas.md",
        "memory/proactive-ideas.md": "memory/proactive-ideas.md",
    }
    return aliases.get(text, "")


def _get_workspace() -> str:
    from core import get_workspace_path
    return get_workspace_path()


def _get_drives_config() -> dict:
    try:
        from core.config import get_drives_config as _core_get_drives_config
        out = _core_get_drives_config().copy()
    except Exception:
        try:
            from core.config import merge_drives_defaults
            try:
                from core import get_config
                raw = get_config().get("drives")
            except Exception:
                raw = {}
            out = merge_drives_defaults(raw)
        except Exception:
            out = {}
    out.setdefault("temperature", 0.2)
    out.setdefault("max_tokens", 700)
    return out


def _get_curiosity_scoring_config() -> dict:
    try:
        from core import get_config

        return (get_config().get("curiosity_scoring") or {}).copy()
    except Exception:
        return {}


def _is_enabled() -> bool:
    return bool(_get_drives_config().get("enabled", True))


def _drives_path(workspace: str) -> str:
    return os.path.join(workspace, "memory", "drives.json")


def _growth_events_path(workspace: str) -> str:
    return os.path.join(workspace, "memory", "growth-events.md")


def _attempt_journal_path(workspace: str) -> str:
    return os.path.join(workspace, "memory", "drive_attempt_journal.jsonl")


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 ""


def _path_contained_in_workspace(path_real: str, workspace_real: str) -> bool:
    if path_real == workspace_real:
        return True
    prefix = workspace_real if workspace_real.endswith(os.sep) else workspace_real + os.sep
    return path_real.startswith(prefix)


def _read_file_tail_safe(path: str, max_chars: int = _MAX_INPUT_CHARS) -> str:
    if not path or not os.path.isfile(path):
        return ""
    try:
        path_real = os.path.realpath(path)
        ws_real = os.path.realpath(_get_workspace())
        if not _path_contained_in_workspace(path_real, ws_real):
            _log.warning(
                "drive_engine: refusing tail read outside workspace (path=%s workspace=%s)",
                path_real,
                ws_real,
            )
            return ""
    except OSError:
        return ""
    char_limit = max_chars if max_chars > 0 else _MAX_INPUT_CHARS
    read_bytes = min(4_194_304, max(1, char_limit) * 4 + 128)
    try:
        with open(path_real, "rb") as f:
            f.seek(0, os.SEEK_END)
            size = f.tell()
            if size <= 0:
                return ""
            start = max(0, size - read_bytes)
            f.seek(start, os.SEEK_SET)
            raw = f.read()
    except OSError:
        return ""
    try:
        text = raw.decode("utf-8", errors="replace")
    except Exception:
        return ""
    if max_chars <= 0:
        return text
    return text[-max_chars:] if len(text) > max_chars else text


def _backup_corrupt_drives_file(path: str) -> str | None:
    """Copy ``path`` to ``<basename>.corrupt.<utc_timestamp>`` in the same directory."""
    if not path or not os.path.isfile(path):
        return None
    parent = os.path.dirname(path) or "."
    base = os.path.basename(path)
    ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    backup_path = os.path.join(parent, f"{base}.corrupt.{ts}")
    n = 0
    while os.path.exists(backup_path):
        n += 1
        backup_path = os.path.join(parent, f"{base}.corrupt.{ts}_{n}")
    try:
        shutil.copy2(path, backup_path)
        return backup_path
    except OSError as copy_err:
        _log.warning("Could not backup corrupt drives file %s: %s", path, copy_err)
        return None


def _load_drives_json_state(state_path: str) -> dict:
    """Load ``memory/drives.json``. Missing file → empty default state.

    Invalid JSON or non-object document → log WARNING, best-effort backup of the
    on-disk file, then return defaults (fail closed without silently discarding
    the corrupt bytes).
    """
    default = _default_state()
    if not state_path or not os.path.isfile(state_path):
        return default
    try:
        with open(state_path, "r", encoding="utf-8") as f:
            data = json.load(f)
    except json.JSONDecodeError as e:
        _log.warning(
            "drives.json at %s is not valid JSON (%s); using empty default state; attempting backup.",
            state_path,
            e,
        )
        backup = _backup_corrupt_drives_file(state_path)
        if backup:
            _log.warning("Backed up corrupt drives.json to %s", backup)
            try:
                _write_json_secure(state_path, _default_state())
            except OSError as werr:
                _log.warning("Could not rewrite drives.json after backup at %s: %s", state_path, werr)
        else:
            _log.warning(
                "No backup written for corrupt drives.json at %s; leaving file unchanged on disk.",
                state_path,
            )
        return default
    except UnicodeDecodeError as e:
        _log.warning(
            "drives.json at %s is not valid UTF-8 (%s); using empty default state; attempting backup.",
            state_path,
            e,
        )
        backup = _backup_corrupt_drives_file(state_path)
        if backup:
            _log.warning("Backed up corrupt drives.json to %s", backup)
            try:
                _write_json_secure(state_path, _default_state())
            except OSError as werr:
                _log.warning("Could not rewrite drives.json after backup at %s: %s", state_path, werr)
        else:
            _log.warning(
                "No backup written for corrupt drives.json at %s; leaving file unchanged on disk.",
                state_path,
            )
        return default
    except OSError as e:
        _log.warning("Could not read drives.json at %s: %s; using empty default state.", state_path, e)
        return default

    if not isinstance(data, dict):
        _log.warning(
            "drives.json at %s must be a JSON object, got %s; using empty default state; attempting backup.",
            state_path,
            type(data).__name__,
        )
        backup = _backup_corrupt_drives_file(state_path)
        if backup:
            _log.warning("Backed up invalid drives.json to %s", backup)
            try:
                _write_json_secure(state_path, _default_state())
            except OSError as werr:
                _log.warning("Could not rewrite drives.json after backup at %s: %s", state_path, werr)
        else:
            _log.warning(
                "No backup written for invalid drives.json at %s; leaving file unchanged on disk.",
                state_path,
            )
        return default
    return data


def _write_json_secure(path: str, data: Any) -> None:
    parent = os.path.dirname(path)
    if parent:
        os.makedirs(parent, exist_ok=True)
    secure_atomic_write(path, json.dumps(data, indent=2, ensure_ascii=True))
    try:
        os.chmod(path, 0o600)
    except OSError:
        pass


def _utc_now_iso() -> str:
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat()


def _new_drive_id() -> str:
    return "drive_" + uuid.uuid4().hex[:12]


def _normalize_drive_status(raw: Any) -> str:
    status = str(raw or "active").strip().lower()
    if status in {"resolved", "expired"}:
        return status
    if status in {"active", "in_progress", "blocked"}:
        return status
    return "active"


def _drive_actionability_score(drive: dict) -> float:
    """Heuristic score for selecting next drive action."""
    tension = max(0.0, min(1.0, float(drive.get("tension", 0.0) or 0.0)))
    novelty = max(0.0, min(1.0, float(drive.get("novelty_score", 0.0) or 0.0)))
    interest = max(0.0, min(1.0, float(drive.get("interest_score", 0.0) or 0.0)))
    reward = max(0.0, min(1.0, float(drive.get("reward_score", 0.0) or 0.0)))
    status = _normalize_drive_status(drive.get("status"))
    status_bonus = 0.0
    if status == "active":
        status_bonus = 0.12
    elif status == "in_progress":
        status_bonus = 0.18
    elif status == "blocked":
        status_bonus = -0.05
    return round((0.6 * tension) + (0.2 * interest) + (0.1 * novelty) + (0.1 * reward) + status_bonus, 4)


def _normalize_drive_record(drive: dict) -> dict:
    d = dict(drive or {})
    d["status"] = _normalize_drive_status(d.get("status"))
    d.setdefault("created_at", _utc_now_iso())
    d.setdefault("last_escalated_at", d["created_at"])
    d.setdefault("escalation_count", 0)
    d.setdefault("resolution", None)
    d.setdefault("progress_count", 0)
    d.setdefault("last_progress_at", "")
    d.setdefault("last_attempt_at", "")
    d.setdefault("blocked_reason", "")
    d.setdefault("next_retry_after", "")
    d.setdefault("defer_until", "")
    d.setdefault("last_action_outcome", "")
    d.setdefault("last_action_summary", "")
    d.setdefault("interest_score", 0.4)
    d.setdefault("novelty_score", 0.3)
    d.setdefault("reward_score", 0.0)
    d.setdefault("plan_steps", [])
    d.setdefault("current_step", 0)
    d.setdefault("next_step_hint", "")
    d.setdefault("completion_criteria", "")
    d.setdefault("last_outcome", "")
    d["actionability_score"] = _drive_actionability_score(d)
    return d


def _merge_duplicate_drives(active: list[dict]) -> list[dict]:
    """Merge near-duplicate active drives by semantic similarity."""
    merged: list[dict] = []
    for d in active:
        cur = _normalize_drive_record(d)
        found_idx = -1
        found_sim = 0.0
        for i, existing in enumerate(merged):
            if str(existing.get("type") or "") != str(cur.get("type") or ""):
                continue
            sim = _jaccard_similarity(
                str(existing.get("description") or ""),
                str(cur.get("description") or ""),
            )
            if sim > found_sim:
                found_sim = sim
                found_idx = i
        if found_idx >= 0 and found_sim >= 0.82:
            ex = merged[found_idx]
            ex["tension"] = max(float(ex.get("tension", 0.0)), float(cur.get("tension", 0.0)))
            ex["escalation_count"] = int(ex.get("escalation_count", 0)) + int(cur.get("escalation_count", 0))
            ex["progress_count"] = int(ex.get("progress_count", 0)) + int(cur.get("progress_count", 0))
            if not str(ex.get("suggested_action") or "").strip():
                ex["suggested_action"] = str(cur.get("suggested_action") or "")
            if not str(ex.get("source_detail") or "").strip():
                ex["source_detail"] = str(cur.get("source_detail") or "")
            ex["actionability_score"] = _drive_actionability_score(ex)
            merged[found_idx] = ex
            continue
        merged.append(cur)
    return merged


def _default_state() -> dict:
    return {
        "active_drives": [],
        "resolved_drives": [],
        "expired_drives": [],
        "action_leases": {},
        "attempt_journal_meta": {
            "last_event_ts": "",
            "events_written": 0,
        },
        "proactive_budget": {
            "day": "",
            "actions_today": 0,
            "tokens_today": 0,
            "last_action_at": "",
        },
        "stats": {
            "total_discovered": 0,
            "total_resolved": 0,
            "total_expired": 0,
            "longest_active_days": 0,
        },
    }


def _load_state(workspace: str | None = None) -> tuple[dict, str]:
    ws = workspace or _get_workspace()
    state_path = _drives_path(ws)
    state = _load_drives_json_state(state_path)
    state.setdefault("active_drives", [])
    state.setdefault("resolved_drives", [])
    state.setdefault("expired_drives", [])
    state.setdefault("action_leases", {})
    state.setdefault("attempt_journal_meta", {"last_event_ts": "", "events_written": 0})
    state.setdefault("proactive_budget", {"day": "", "actions_today": 0, "tokens_today": 0, "last_action_at": ""})
    state.setdefault("stats", {})
    state["active_drives"] = _merge_duplicate_drives(state.get("active_drives") or [])
    state["resolved_drives"] = [_normalize_drive_record(d) for d in (state.get("resolved_drives") or [])]
    state["expired_drives"] = [_normalize_drive_record(d) for d in (state.get("expired_drives") or [])]
    return state, state_path


def _save_state(state: dict, workspace: str | None = None) -> None:
    ws = workspace or _get_workspace()
    state_path = _drives_path(ws)
    _write_json_secure(state_path, state)


def _append_attempt_journal_event(workspace: str, event_type: str, payload: dict[str, Any]) -> None:
    path = _attempt_journal_path(workspace)
    parent = os.path.dirname(path)
    if parent:
        os.makedirs(parent, exist_ok=True)
    entry = {
        "ts": _utc_now_iso(),
        "event": str(event_type or "").strip(),
        "payload": payload or {},
    }
    with _DRIVE_STATE_LOCK:
        try:
            with open(path, "a", encoding="utf-8") as f:
                f.write(json.dumps(entry, ensure_ascii=True) + "\n")
            os.chmod(path, 0o600)
        except OSError:
            return
        state, _ = _load_state(workspace)
        meta = state.setdefault("attempt_journal_meta", {})
        meta["last_event_ts"] = entry["ts"]
        meta["events_written"] = int(meta.get("events_written", 0)) + 1
        _save_state(state, workspace)


def _emit_drive_trace_event(event_type: str, drive_id: str = "", details: dict[str, Any] | None = None) -> None:
    try:
        from tools.conversation_trace import emit_drive_event

        emit_drive_event("drive_engine", event_type, drive_id=drive_id, details=details or {})
    except Exception:
        pass


def consume_proactive_budget(
    workspace: str | None = None,
    *,
    token_cost: int = 0,
) -> tuple[bool, str]:
    """Budget gate for proactive runs (day/hour caps).

    Call sites: only drive_action_cycle in mugi_runner consumes this today; other background
    LLM paths do not use the proactive budget.
    """
    ws = workspace or _get_workspace()
    cfg = _get_drives_config()
    max_day = int(cfg.get("max_proactive_actions_per_day", 40))
    max_tokens_day = int(cfg.get("max_proactive_tokens_per_day", 25000))
    max_hour = int(cfg.get("max_proactive_actions_per_hour", 6))
    now = datetime.now(timezone.utc)
    day_key = now.strftime("%Y-%m-%d")
    with _DRIVE_STATE_LOCK:
        state, _ = _load_state(ws)
        budget = state.setdefault("proactive_budget", {"day": "", "actions_today": 0, "tokens_today": 0, "last_action_at": ""})
        if str(budget.get("day") or "") != day_key:
            budget["day"] = day_key
            budget["actions_today"] = 0
            budget["tokens_today"] = 0
            budget["last_action_at"] = ""
            budget["actions_this_hour"] = []

        hour_items = budget.get("actions_this_hour")
        if not isinstance(hour_items, list):
            hour_items = []
        cleaned: list[str] = []
        for ts in hour_items:
            try:
                dt = datetime.fromisoformat(str(ts))
                if dt.tzinfo is None:
                    dt = dt.replace(tzinfo=timezone.utc)
                if (now - dt) <= timedelta(hours=1):
                    cleaned.append(dt.isoformat())
            except Exception:
                continue
        if len(cleaned) >= max(1, max_hour):
            return False, "hourly proactive budget exhausted"
        if int(budget.get("actions_today", 0)) >= max(1, max_day):
            return False, "daily proactive action budget exhausted"
        if int(budget.get("tokens_today", 0)) + max(0, int(token_cost or 0)) > max(1, max_tokens_day):
            return False, "daily proactive token budget exhausted"
        cleaned.append(now.replace(microsecond=0).isoformat())
        budget["actions_this_hour"] = cleaned
        budget["actions_today"] = int(budget.get("actions_today", 0)) + 1
        budget["tokens_today"] = int(budget.get("tokens_today", 0)) + max(0, int(token_cost or 0))
        budget["last_action_at"] = _utc_now_iso()
        state["proactive_budget"] = budget
        _save_state(state, ws)
    return True, "budget_ok"


def refund_proactive_budget(
    workspace: str | None = None,
    *,
    token_cost: int = 0,
) -> None:
    """Reverse a single consume_proactive_budget call on failure (best-effort)."""
    ws = workspace or _get_workspace()
    with _DRIVE_STATE_LOCK:
        state, _ = _load_state(ws)
        budget = state.get("proactive_budget")
        if not isinstance(budget, dict):
            return
        budget["actions_today"] = max(0, int(budget.get("actions_today", 0)) - 1)
        budget["tokens_today"] = max(0, int(budget.get("tokens_today", 0)) - max(0, int(token_cost or 0)))
        hour_items = budget.get("actions_this_hour")
        if isinstance(hour_items, list) and hour_items:
            hour_items.pop()
        state["proactive_budget"] = budget
        _save_state(state, ws)


def _tokenize(text: str) -> set[str]:
    words = re.findall(r"[a-zA-Z0-9_]{3,}", (text or "").lower())
    return set(words)


def _jaccard_similarity(a: str, b: str) -> float:
    ta = _tokenize(a)
    tb = _tokenize(b)
    if not ta or not tb:
        return 0.0
    inter = len(ta & tb)
    union = len(ta | tb)
    return float(inter) / float(union) if union else 0.0


def _llm_call(system_prompt: str, user_prompt: str, on_log=None) -> str | None:
    cfg = _get_drives_config()
    model = (cfg.get("discovery_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
    discovery_use_main_llm = bool(cfg.get("discovery_use_main_llm", False))
    if not model and not base_url and not discovery_use_main_llm:
        _log.info("Discovery skipped: [drives] discovery_model/base_url not set and discovery_use_main_llm is false")
        return None
    temp = float(cfg.get("temperature", 0.2))
    max_tokens = int(cfg.get("max_tokens", 700))
    stop_kw = validated_stop(cfg.get("stop"), _log, "[drives]")
    # Default discovery responses to JSON object mode unless explicitly overridden.
    rf_cfg = cfg.get("response_format")
    if rf_cfg is None:
        rf_cfg = {"type": "json_object"}
    response_format_kw = validated_response_format(rf_cfg, _log, "[drives]")
    try:
        from agent_loop import run_raw_agent
        collector: list[str] = []
        err_ref: list[str] = []

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

        def _on_error(err: str):
            err_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=[
                {"role": "user", "content": f"{system_prompt}\n\n---\n\n{user_prompt}"},
            ],
            thread_id=f"drive_discovery_{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=stop_kw,
            response_format=response_format_kw,
            policy_context=_bg_policy_ctx,
        )
        out = ((returned or "").strip() or "".join(collector).strip())
        out = strip_untagged_cot(strip_markdown_fences(strip_think_tags(out)))
        if err_ref and not out:
            return None
        return out or None
    except Exception:
        _log.warning("Discovery _llm_call failed", exc_info=True)
        return None


def _user_model_discovery_context(workspace: str, *, max_chars: int = 2500) -> str:
    """High-trust User Model snapshot for drive discovery (Phase 5 / Slice C)."""
    try:
        from utils.user_model import load_user_model

        model = load_user_model(workspace)
    except Exception:
        return ""
    lines: list[str] = []
    focus = model.current_focus if isinstance(model.current_focus, dict) else {}
    focus_text = " ".join(str(v).strip() for v in focus.values() if str(v).strip())
    if focus_text:
        lines.append(f"### User Model — current focus\n{focus_text[:600]}")
    active_goals = [
        g for g in model.goals
        if g.get("status") in (None, "active", "in_progress")
        and str(g.get("title") or "").strip()
    ]
    if active_goals:
        lines.append("### User Model — active goals")
        for g in active_goals[:8]:
            title = str(g.get("title") or "").strip()
            why = str(g.get("why_it_matters") or "").strip()[:120]
            line = f"- {title}"
            if why:
                line += f" — {why}"
            lines.append(line)
    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("### User Model — open commitments")
        for item in open_items[:6]:
            lines.append(f"- {str(item.get('text') or '')[:200]}")
    if not lines:
        return ""
    return "\n\n".join(lines)[:max_chars]


def _user_model_goal_phrases(workspace: str) -> list[str]:
    """Titles/descriptions already tracked in User Model — skip duplicate drives."""
    try:
        from utils.user_model import load_user_model

        model = load_user_model(workspace)
    except Exception:
        return []
    phrases: list[str] = []
    for g in model.goals:
        if g.get("status") not in (None, "active", "in_progress"):
            continue
        title = str(g.get("title") or "").strip()
        if title:
            phrases.append(title)
        why = str(g.get("why_it_matters") or "").strip()
        if why:
            phrases.append(why)
    for c in model.open_commitments:
        if c.get("status", "open") in ("done", "closed", "cancelled"):
            continue
        text = str(c.get("text") or "").strip()
        if len(text) >= 12:
            phrases.append(text)
    focus = model.current_focus if isinstance(model.current_focus, dict) else {}
    for v in focus.values():
        text = str(v).strip()
        if len(text) >= 12:
            phrases.append(text)
    return phrases


def _gather_discovery_context(workspace: str) -> tuple[str, dict[str, str]]:
    sections: list[str] = []
    by_source: dict[str, str] = {}
    um_block = _user_model_discovery_context(workspace)
    if um_block.strip():
        sections.append(um_block.strip())
        by_source["memory/user_model/model.json"] = um_block.strip()
    for rel, cap in (
        ("memory/insights.md", 7000),
        ("memory/recurring-patterns.md", 3000),
        ("memory/tool_errors.json", 3000),
        ("memory/feedback.md", 2500),
        ("memory/proactive-ideas.md", 2000),
    ):
        path = os.path.join(workspace, rel)
        # Tail-oriented sampling keeps discovery anchored to recent signals.
        content = _read_file_tail_safe(path, cap)
        if content.strip():
            sections.append(f"### {rel}\n{content.strip()}")
            by_source[rel] = content.strip()
    combined = "\n\n".join(sections)
    return combined[:_MAX_INPUT_CHARS], by_source


def _source_detail_supported(
    source_detail: str, source_context: str, drive_type: str = ""
) -> bool:
    detail = (source_detail or "").strip()
    context = (source_context or "").strip()
    if len(detail) < 8 or not context:
        return False
    strict = bool(_get_drives_config().get("discovery_strict_grounding", False))
    detail_l = detail.lower()
    context_l = context.lower()
    dtype = (drive_type or "").strip().lower()
    quoted = [q.strip().lower() for q in re.findall(r"['\"]([^'\"]{8,})['\"]", detail)]
    quoted_hits = bool(quoted) and any(q in context_l for q in quoted)

    if strict and dtype == "exploration":
        return quoted_hits

    if detail_l in context_l:
        return True

    # If a quote is provided as evidence, require at least one quoted snippet match.
    if quoted:
        return any(q in context_l for q in quoted)

    d_tokens = _tokenize(detail)
    c_tokens = _tokenize(context)
    if not d_tokens or not c_tokens:
        return False
    overlap = len(d_tokens & c_tokens)
    if strict:
        return (overlap / len(d_tokens)) >= 0.4
    required = 2 if len(d_tokens) >= 3 else 1
    return overlap >= required


_DISCOVERY_SYSTEM = (
    "You are a motivation-mining engine for a personal AI assistant.\n"
    "Identify unresolved capability gaps, repeated asks, unanswered questions, "
    "or reliability issues that should become active drives.\n"
    "Output strict JSON only in this shape:\n"
    "{\"drives\":[{\"type\":\"capability_gap|unresolved_question|pattern_automation|quality_issue|exploration\","
    "\"description\":\"...\",\"source\":\"...\",\"source_detail\":\"...\",\"suggested_action\":\"...\"}]}\n"
    "The entire assistant response must be exactly one JSON object (no prose, no markdown fences).\n"
    "Rules: max 3 drives, concrete and specific, no generic advice."
)


def _extract_braced_json_objects(text: str) -> list[str]:
    """Return candidate JSON object snippets found in free-form text."""
    out: list[str] = []
    if not text:
        return out
    start = -1
    depth = 0
    in_str = False
    esc = False
    for i, ch in enumerate(text):
        if in_str:
            if esc:
                esc = False
            elif ch == "\\":
                esc = True
            elif ch == '"':
                in_str = False
            continue
        if ch == '"':
            in_str = True
            continue
        if ch == "{":
            if depth == 0:
                start = i
            depth += 1
            continue
        if ch == "}":
            if depth <= 0:
                continue
            depth -= 1
            if depth == 0 and start >= 0:
                out.append(text[start : i + 1])
                start = -1
    return out


def _looks_like_non_json_reasoning(text: str) -> bool:
    t = (text or "").strip().lower()
    if not t:
        return False
    if "{" in t and "}" in t:
        return False
    cues = (
        "the user wants me to",
        "i need to",
        "let me",
        "looking at the context",
        "here are",
    )
    return any(c in t for c in cues)


def _parse_drives_payload(raw: str) -> tuple[dict | None, str]:
    """Parse drives JSON from raw model text.

    Returns (parsed_dict_or_none, failure_reason).
    """
    cleaned = strip_untagged_cot(strip_markdown_fences(strip_think_tags(raw)))
    if not cleaned:
        return None, "empty_text"
    try:
        parsed = json.loads(cleaned)
        if isinstance(parsed, dict):
            # Common wrapper shape: {"result": {"drives":[...]}}
            result_obj = parsed.get("result")
            if isinstance(result_obj, dict) and isinstance(result_obj.get("drives"), list):
                return result_obj, ""
            return parsed, ""
        if isinstance(parsed, list):
            return {"drives": parsed}, ""
        return None, "top_level_not_object"
    except Exception:
        pass

    m = re.search(r'\{\s*"drives"\s*:', cleaned)
    if not m:
        # Fallback: scan for any balanced JSON object and keep one with a drives list.
        for frag in _extract_braced_json_objects(cleaned):
            try:
                cand = json.loads(frag)
            except Exception:
                continue
            if isinstance(cand, dict):
                if isinstance(cand.get("drives"), list):
                    return cand, ""
                result_obj = cand.get("result")
                if isinstance(result_obj, dict) and isinstance(result_obj.get("drives"), list):
                    return result_obj, ""
        if _looks_like_non_json_reasoning(cleaned):
            return None, "non_json_reasoning"
        return None, "missing_drives_object_start"
    try:
        decoder = json.JSONDecoder()
        parsed, _ = decoder.raw_decode(cleaned, m.start())
        if isinstance(parsed, dict):
            return parsed, ""
        return None, "decoded_non_object"
    except Exception:
        return None, "raw_decode_failed"


def _discovery_personal_fact_line(description: str, suggested: str = "") -> str | None:
    """Return the personal-fact text to route, or None if candidate is actionable work."""
    from utils.user_model import is_personal_fact_line

    desc = str(description or "").strip()
    if is_personal_fact_line(desc):
        return desc
    sug = str(suggested or "").strip()
    if sug and is_personal_fact_line(sug):
        return sug
    return None


def _route_personal_discovery_candidate(
    *,
    workspace: str,
    description: str,
    suggested: str,
    budget,
    on_log=None,
) -> tuple[bool, str]:
    """Route personal-signal discovery candidates to User Model proposals (no observation/drive).

    Returns (handled, status) where *handled* is True when the candidate must not also
    surface as an observation or active drive (avoids triple surfacing).
    """
    line = _discovery_personal_fact_line(description, suggested)
    if line is None:
        return False, ""

    from utils.user_model import propose_background_fact

    outcome = propose_background_fact(
        workspace,
        line,
        source="drive_discovery",
        rationale_prefix="Drive discovery noticed",
        budget=budget,
    )
    status = "proposal_routed"
    if outcome == "proposal":
        if on_log:
            on_log("INFO", "[drives] routed personal discovery candidate to User Model proposal")
    elif outcome == "memory":
        status = "memory_fallback"
        try:
            from tools.memory_tools import append_to_memory_for_workspace

            msg = append_to_memory_for_workspace(workspace, line)
            if not msg or "appended to" not in str(msg).lower():
                status = "memory_fallback_failed"
                if on_log:
                    on_log(
                        "WARN",
                        f"[drives] personal discovery MEMORY fallback failed: {line[:120]}",
                    )
        except Exception:
            status = "memory_fallback_failed"
            _log.warning("personal discovery MEMORY fallback failed", exc_info=True)
            if on_log:
                on_log("WARN", "[drives] personal discovery MEMORY fallback failed")
    elif outcome == "skipped":
        status = "proposal_skipped"
    return True, status


def _surface_drive_discovery_observation(
    *,
    description: str,
    suggested_action: str,
    dtype: str,
) -> str:
    """Surface a drive discovery candidate as a consent-gated observation (bounded mode).

    Returns ``new`` when appended, ``pending`` when an equivalent observation is already
    in the inbox, or ``failed`` on error / queue rejection.
    """
    try:
        import hashlib

        from persistence import PROACTIVE_QUEUE_PATH
        from utils.proactive_queue import append_observation_queue

        desc = str(description or "").strip()
        if len(desc) < 12:
            return "failed"
        proposed = str(suggested_action or "").strip() or "investigate this and suggest next steps"
        tier = "kanban" if dtype in {"capability_gap", "quality_issue"} else "in_thread"
        dedupe_key = f"drive_discovery:{hashlib.sha256(desc.encode('utf-8')).hexdigest()[:16]}"
        queue_path = os.path.expanduser(PROACTIVE_QUEUE_PATH)
        if append_observation_queue(
            queue_path,
            noticed=desc[:800],
            proposed_work=proposed[:800],
            source="drive_discovery",
            suggested_tier=tier,
            dedupe_key=dedupe_key,
        ):
            return "new"
        if dedupe_key and os.path.isfile(queue_path):
            try:
                from utils.proactive_queue import find_pending_observation_by_dedupe

                with open(queue_path, "r", encoding="utf-8") as f:
                    if find_pending_observation_by_dedupe(f.readlines(), dedupe_key):
                        return "pending"
            except Exception:
                pass
        return "failed"
    except Exception:
        _log.warning("drive discovery observation surface failed", exc_info=True)
        return "failed"


def discover_drives(workspace: str | None = None, on_log=None) -> list[dict]:
    """Discover new drives from memory signals and merge into state."""
    if not _is_enabled():
        return []
    ws = workspace or _get_workspace()
    context, context_by_source = _gather_discovery_context(ws)
    if not context.strip():
        return []
    user_prompt = "Extract potential drives from this context:\n\n" + context
    raw = _llm_call(_DISCOVERY_SYSTEM, user_prompt, on_log=on_log)
    if not raw:
        if on_log:
            on_log("INFO", f"[drives] discovery returned empty output (context_chars={len(context)})")
        return []
    parsed, parse_err = _parse_drives_payload(raw)
    if parsed is None:
        if on_log:
            preview = raw[:300].replace("\n", "\\n")
            level = "INFO" if parse_err == "non_json_reasoning" else "WARN"
            on_log(
                level,
                f"[drives] failed to parse discovery JSON ({parse_err}, context_chars={len(context)}): {preview}",
            )
        return []
    candidates = parsed.get("drives") if isinstance(parsed, dict) else None
    if not isinstance(candidates, list):
        if on_log:
            on_log("WARN", "[drives] parsed payload missing list field 'drives'")
        return []
    if not candidates:
        if on_log:
            on_log("DEBUG", "[drives] discovery parsed successfully but returned no candidates")
        return []

    with _DRIVE_STATE_LOCK:
        state, _ = _load_state(ws)
        active: list[dict] = state.get("active_drives") or []
    max_active = int(_get_drives_config().get("max_active_drives", 5))
    new_items: list[dict] = []
    reject_counts: dict[str, int] = {}
    escalated_existing = 0
    capacity_skips = 0

    def _reject(reason: str) -> None:
        reject_counts[reason] = int(reject_counts.get(reason, 0)) + 1

    scoring_cfg = _get_curiosity_scoring_config()
    scoring_enabled = bool(scoring_cfg.get("enabled", True))
    scoring_half_life = float(scoring_cfg.get("half_life_hours", 48.0))
    scoring_decay = float(scoring_cfg.get("interest_decay", 0.9))
    interest_model = None
    gap_tracker = None
    prefetch_dir = os.path.join(ws, "memory", "prefetched")
    if scoring_enabled:
        try:
            from curiosity_scoring import InterestModel, KnowledgeGapTracker

            interest_model = InterestModel(workspace=ws, decay=scoring_decay)
            gap_tracker = KnowledgeGapTracker(workspace=ws)
        except Exception:
            scoring_enabled = False

    um_phrases = _user_model_goal_phrases(ws)
    autonomy_mode = str(_get_drives_config().get("autonomy_mode", "bounded")).strip().lower()
    surface_as_observation = autonomy_mode != "aggressive"
    observations_surfaced = 0
    personal_routed = 0

    from utils.user_model import BackgroundProposalBudget

    proposal_budget = BackgroundProposalBudget()

    for c in candidates[:3]:
        if not isinstance(c, dict):
            _reject("non_object_candidate")
            continue
        unknown_keys = set(c.keys()) - _DRIVE_ALLOWED_KEYS
        if unknown_keys:
            _reject("extra_keys")
            continue
        description = str(c.get("description") or "").strip()
        if len(description) < 12:
            _reject("description_too_short")
            continue
        dtype = str(c.get("type") or "capability_gap").strip().lower()
        if dtype not in _VALID_DRIVE_TYPES:
            _reject("invalid_type")
            continue
        source_raw = str(c.get("source") or "").strip()
        source = _normalize_source_label(source_raw)
        if not source:
            _reject("invalid_source")
            continue
        source_detail = str(c.get("source_detail") or "").strip()[:240]
        source_context = context_by_source.get(source, "")
        if not _source_detail_supported(source_detail, source_context, dtype):
            _reject("unsupported_source_detail")
            continue
        suggested = str(c.get("suggested_action") or "").strip()[:240]

        # Skip candidates that duplicate User Model goals/focus (already surfaced as drives).
        if um_phrases:
            dup_um = False
            for phrase in um_phrases:
                if _jaccard_similarity(description, phrase) >= 0.55:
                    _reject("duplicates_user_model_goal")
                    dup_um = True
                    break
            if dup_um:
                continue

        routed, route_status = _route_personal_discovery_candidate(
            workspace=ws,
            description=description,
            suggested=suggested,
            budget=proposal_budget,
            on_log=on_log,
        )
        if routed:
            _reject("routed_personal_proposal")
            personal_routed += 1
            new_items.append({
                "id": "",
                "description": description[:300],
                "status": route_status,
                "source": source,
            })
            continue

        # Deduplicate against existing active drives.
        best_idx = -1
        best_sim = 0.0
        for i, d in enumerate(active):
            sim = _jaccard_similarity(description, str(d.get("description") or ""))
            if sim > best_sim:
                best_sim = sim
                best_idx = i
        if best_idx >= 0 and best_sim >= 0.7:
            cur = active[best_idx]
            cur["tension"] = min(1.0, float(cur.get("tension", 0.3)) + 0.05)
            cur["last_escalated_at"] = _utc_now_iso()
            escalated_existing += 1
            continue

        if surface_as_observation:
            surface_outcome = _surface_drive_discovery_observation(
                description=description,
                suggested_action=suggested,
                dtype=dtype,
            )
            if surface_outcome == "new":
                observations_surfaced += 1
                new_items.append({
                    "id": "",
                    "description": description[:300],
                    "status": "observation_pending",
                    "source": source,
                })
            elif surface_outcome == "pending":
                new_items.append({
                    "id": "",
                    "description": description[:300],
                    "status": "observation_already_pending",
                    "source": source,
                })
            else:
                _reject("observation_surface_failed")
                if on_log:
                    on_log(
                        "WARN",
                        f"[drives] failed to surface discovery observation: {description[:120]}",
                    )
            continue

        if len(active) >= max_active:
            # Capacity is full for new drives; keep scanning remaining candidates
            # so duplicates can still escalate matching existing drives.
            capacity_skips += 1
            continue

        initial_tension = 0.3
        if scoring_enabled and interest_model is not None and gap_tracker is not None:
            try:
                from curiosity_scoring import calculate_curiosity_score

                score = calculate_curiosity_score(
                    topic=description,
                    interest_model=interest_model,
                    gap_tracker=gap_tracker,
                    prefetch_dir=prefetch_dir,
                    half_life_hours=scoring_half_life,
                )
                initial_tension = max(0.3, min(0.6, float(score)))
            except Exception:
                initial_tension = 0.3

        drive = _normalize_drive_record({
            "id": _new_drive_id(),
            "type": dtype,
            "description": description[:300],
            "source": source,
            "source_detail": source_detail,
            "suggested_action": suggested,
            "tension": initial_tension,
            "created_at": _utc_now_iso(),
            "last_escalated_at": _utc_now_iso(),
            "escalation_count": 0,
            "resolution": None,
            "status": "active",
            "interest_score": initial_tension,
            "novelty_score": 0.4,
            "reward_score": 0.0,
        })
        active.append(drive)
        new_items.append(drive)

    with _DRIVE_STATE_LOCK:
        fresh_state, _ = _load_state(ws)
        if surface_as_observation:
            if escalated_existing > 0:
                fresh_state["active_drives"] = _merge_duplicate_drives(active)
            if escalated_existing > 0 or observations_surfaced:
                stats = fresh_state.setdefault("stats", {})
                if observations_surfaced:
                    stats["total_discovery_observations"] = int(
                        stats.get("total_discovery_observations", 0)
                    ) + observations_surfaced
                _save_state(fresh_state, ws)
        else:
            fresh_state["active_drives"] = _merge_duplicate_drives(active)
            stats = fresh_state.setdefault("stats", {})
            actual_drives = sum(1 for d in new_items if d.get("id"))
            stats["total_discovered"] = int(stats.get("total_discovered", 0)) + actual_drives
            _save_state(fresh_state, ws)
    _non_drive_discovery_statuses = frozenset({
        "observation_pending",
        "observation_already_pending",
        "proposal_routed",
        "proposal_skipped",
        "memory_fallback",
        "memory_fallback_failed",
    })
    for d in new_items:
        if d.get("status") in _non_drive_discovery_statuses or not d.get("id"):
            continue
        _emit_drive_trace_event("drive_discovered", str(d.get("id") or ""), {"type": d.get("type"), "tension": d.get("tension")})
    if on_log:
        considered = len(candidates[:3])
        rej = ", ".join(f"{k}={v}" for k, v in sorted(reject_counts.items())) or "none"
        on_log(
            "INFO",
            (
                "[drives] discovery stats: "
                f"considered={considered} discovered={len(new_items)} "
                f"observations_surfaced={observations_surfaced} "
                f"personal_routed={personal_routed} "
                f"escalated_existing={escalated_existing} capacity_skips={capacity_skips} "
                f"rejections={rej}"
            ),
        )
    return new_items


def escalate_drives(workspace: str | None = None) -> dict:
    """Escalate tension for unresolved drives and expire stale ones."""
    if not _is_enabled():
        return {"escalated": 0, "expired": 0}
    ws = workspace or _get_workspace()
    with _DRIVE_STATE_LOCK:
        state, _ = _load_state(ws)
        active: list[dict] = state.get("active_drives") or []
        expired: list[dict] = state.get("expired_drives") or []
    cfg = _get_drives_config()
    step = float(cfg.get("escalation_step", 0.1))
    max_age_days = int(cfg.get("max_age_days", 14))
    now = datetime.now(timezone.utc)

    kept: list[dict] = []
    escalated_count = 0
    expired_count = 0
    blocked_retry_minutes = int(cfg.get("blocked_retry_minutes", _DEFAULT_BLOCK_RETRY_MINUTES))
    for d in active:
        d = _normalize_drive_record(d)
        try:
            created = datetime.fromisoformat(str(d.get("created_at") or ""))
        except Exception:
            created = now
        age_days = (now - created).days
        if age_days > max_age_days:
            d["expired_at"] = _utc_now_iso()
            d["expired_reason"] = "max_age"
            expired.append(d)
            expired_count += 1
            continue
        status = _normalize_drive_status(d.get("status"))
        if status == "blocked":
            retry_after = str(d.get("next_retry_after") or "").strip()
            if retry_after:
                try:
                    retry_dt = datetime.fromisoformat(retry_after)
                    if retry_dt.tzinfo is None:
                        retry_dt = retry_dt.replace(tzinfo=timezone.utc)
                    if now < retry_dt:
                        kept.append(d)
                        continue
                except Exception:
                    pass
            d["status"] = "active"
            d["blocked_reason"] = ""
            d["next_retry_after"] = ""

        step_this_round = step
        if status == "in_progress":
            last_progress = str(d.get("last_progress_at") or "").strip()
            if last_progress:
                try:
                    lp = datetime.fromisoformat(last_progress)
                    if lp.tzinfo is None:
                        lp = lp.replace(tzinfo=timezone.utc)
                    if (now - lp) <= timedelta(hours=24):
                        step_this_round = 0.0
                except Exception:
                    pass
        if status == "blocked":
            step_this_round = min(step, 0.03)

        max_futile = int(cfg.get("max_futile_attempts", 3))
        if (
            int(d.get("progress_count", 0)) == 0
            and int(d.get("escalation_count", 0)) >= max_futile
        ):
            d["defer_until"] = (now + timedelta(days=7)).isoformat()
            d["tension"] = 0.3
            d["last_escalated_at"] = _utc_now_iso()
            d["actionability_score"] = _drive_actionability_score(d)
            kept.append(d)
            continue

        d["tension"] = min(1.0, float(d.get("tension", 0.3)) + step_this_round)
        d["last_escalated_at"] = _utc_now_iso()
        d["escalation_count"] = int(d.get("escalation_count", 0)) + 1
        d["actionability_score"] = _drive_actionability_score(d)
        escalated_count += 1
        kept.append(d)

    with _DRIVE_STATE_LOCK:
        fresh_state, _ = _load_state(ws)
        fresh_state["active_drives"] = kept
        fresh_state["expired_drives"] = expired
        stats = fresh_state.setdefault("stats", {})
        stats["total_expired"] = int(stats.get("total_expired", 0)) + expired_count
        _save_state(fresh_state, ws)
    return {"escalated": escalated_count, "expired": expired_count}


def get_active_drives(workspace: str | None = None, limit: int | None = None) -> list[dict]:
    ws = workspace or _get_workspace()
    state, _ = _load_state(ws)
    active: list[dict] = state.get("active_drives") or []
    normalized = [_normalize_drive_record(d) for d in active]

    # Phase 5 re-orientation: Surface high-priority User Model goals as first-class personal drives.
    # This makes the User Model the primary source for the agent's "personal motivation" awareness.
    try:
        from utils.user_model import load_user_model
        model = load_user_model(ws)
        seen_um_drive_ids: set[str] = set()
        for g in model.goals:
            if g.get("status") in (None, "active", "in_progress"):
                title = str(g.get("title") or "").strip()
                if not title:
                    continue
                goal_id = g.get("id")
                if goal_id is not None and str(goal_id).strip():
                    stable_id = str(goal_id).strip()
                else:
                    horizon = str(g.get("horizon") or "").strip()
                    stable_id = f"{title}-{horizon}" if horizon else title
                stable_id = re.sub(r"[^A-Za-z0-9._-]+", "-", stable_id)[:80] or "goal"
                drive_id = f"um-goal-{stable_id}"
                if drive_id in seen_um_drive_ids:
                    drive_id = f"{drive_id}-{len(seen_um_drive_ids)}"
                seen_um_drive_ids.add(drive_id)
                drive_like = {
                    "id": drive_id,
                    "title": title,
                    "description": g.get("why_it_matters", ""),
                    "type": "user_model_goal",
                    "horizon": g.get("horizon"),
                    "source": "user_model",
                    "actionability_score": 0.95,  # User Model goals are highest priority by design
                    "tension": 0.8,
                }
                normalized.append(drive_like)
    except Exception:
        pass  # Non-fatal; fall back to legacy drives only

    ordered = sorted(
        normalized,
        key=lambda d: (float(d.get("actionability_score", 0.0)), float(d.get("tension", 0.0))),
        reverse=True,
    )
    if limit is not None and limit > 0:
        ordered = ordered[:limit]
    return [{**d, "ref": i + 1} for i, d in enumerate(ordered)]


def _find_drive_by_ref_or_id(drive_id_or_ref: str, active: list[dict]) -> dict | None:
    """Resolve a drive_id (integer ref like '1' or hex ID) to a drive dict.

    Ref numbers are 1-based, assigned by (actionability, tension)-descending sort order — the same
    order returned by ``get_active_drives`` and ``list_drives``.
    """
    clean = re.sub(r"\s+", "", str(drive_id_or_ref or "")).strip()
    if not clean:
        return None
    if clean.isdigit():
        ref = int(clean)
        ordered = sorted(
            active,
            key=lambda d: (float(d.get("actionability_score", 0.0)), float(d.get("tension", 0.0))),
            reverse=True,
        )
        if 1 <= ref <= len(ordered):
            return ordered[ref - 1]
        return None
    for d in active:
        if re.sub(r"\s+", "", str(d.get("id") or "")) == clean:
            return d
    return None


def snapshot_active_drives_for_trajectory(workspace: str | None = None, limit: int = 5) -> list[dict]:
    drives = get_active_drives(workspace=workspace, limit=limit)
    return [
        {
            "id": d.get("id", ""),
            "type": d.get("type", ""),
            "tension": float(d.get("tension", 0.0)),
            "description": str(d.get("description", ""))[:200],
        }
        for d in drives
    ]


def _tool_allowed_for_proactive(tool_name: str) -> bool:
    cfg = _get_drives_config()
    mode = str(cfg.get("autonomy_mode", "bounded")).strip().lower()
    allowlist = cfg.get("proactive_allowlist") or []
    agg_list = cfg.get("aggressive_allowlist") or []
    try:
        from tool_arg_normalizer import proactive_tool_allowed
        return proactive_tool_allowed(
            tool_name, mode,
            allowlist=allowlist if isinstance(allowlist, list) else [],
            aggressive_allowlist=agg_list if isinstance(agg_list, list) else [],
        )
    except ImportError:
        if mode == "off":
            return False
        if mode == "aggressive":
            if isinstance(agg_list, list) and agg_list:
                return str(tool_name or "").strip() in {str(x).strip() for x in agg_list if str(x).strip()}
            return True
        if not isinstance(allowlist, list):
            return False
        return str(tool_name or "").strip() in {str(x).strip() for x in allowlist if str(x).strip()}


def is_tool_allowed_for_proactive(tool_name: str) -> bool:
    """Public wrapper for proactive-policy tool allow checks."""
    return _tool_allowed_for_proactive(tool_name)


def select_next_drive_for_action(
    workspace: str | None = None,
    limit: int = 1,
    include_blocked_retryable: bool = False,
) -> list[dict]:
    """Select next actionable drives for initiative work."""
    ws = workspace or _get_workspace()
    state, _ = _load_state(ws)
    active = [_normalize_drive_record(d) for d in (state.get("active_drives") or [])]
    now = datetime.now(timezone.utc)
    candidates: list[dict] = []
    for d in active:
        status = _normalize_drive_status(d.get("status"))
        if status not in _ACTIVE_STATUSES:
            continue
        if status == "blocked":
            if not include_blocked_retryable:
                continue
            retry_after = str(d.get("next_retry_after") or "").strip()
            if retry_after:
                try:
                    rd = datetime.fromisoformat(retry_after)
                    if rd.tzinfo is None:
                        rd = rd.replace(tzinfo=timezone.utc)
                    if now < rd:
                        continue
                except Exception:
                    continue
        defer = str(d.get("defer_until") or "").strip()
        if defer:
            try:
                dd = datetime.fromisoformat(defer)
                if dd.tzinfo is None:
                    dd = dd.replace(tzinfo=timezone.utc)
                if now < dd:
                    continue
            except Exception:
                pass
        d["actionability_score"] = _drive_actionability_score(d)
        candidates.append(d)
    candidates.sort(
        key=lambda d: (float(d.get("actionability_score", 0.0)), float(d.get("tension", 0.0))),
        reverse=True,
    )
    n = max(1, int(limit or 1))
    return [{**d, "ref": i + 1} for i, d in enumerate(candidates[:n])]


def plan_drive_action(drive: dict, context: dict[str, Any] | None = None) -> dict[str, Any]:
    """Compile a drive into one bounded action specification."""
    d = _normalize_drive_record(drive or {})
    hint = str(d.get("next_step_hint") or "").strip()
    suggested = str(d.get("suggested_action") or "").strip()
    desc = str(d.get("description") or "").strip()
    dtype = str(d.get("type") or "").strip().lower()
    action = hint or suggested or desc
    recommended_tool = "write_session_state"
    if "plugin" in action.lower() or "capability" in action.lower() or dtype == "capability_gap":
        recommended_tool = "create_plugin"
    elif "workflow" in action.lower() or "repeat" in action.lower() or dtype == "pattern_automation":
        recommended_tool = "skill_manage"
    elif dtype in {"exploration", "unresolved_question"}:
        action_lower = action.lower()
        _internal_kw = (
            "mugi",
            "beta",
            "release",
            "sparkle",
            "workspace",
            "session",
            "memory.md",
            "identity",
            "constitution",
            "friends",
            "family",
            "swift",
            "xcode",
            "build ",
        )
        if any(k in action_lower for k in _internal_kw):
            recommended_tool = "read_file"
        else:
            recommended_tool = "search_web"
    return {
        "drive_id": str(d.get("id") or ""),
        "status": str(d.get("status") or "active"),
        "intent": action[:300],
        "recommended_tool": recommended_tool,
        "success_check": str(
            d.get("completion_criteria")
            or "Record concrete project progress in write_session_state; use append_to_memory only for durable non-personal research outcomes."
        ),
        "context": context or {},
    }


def acquire_drive_action_lease(
    drive_id: str,
    owner: str,
    workspace: str | None = None,
    ttl_seconds: int | None = None,
) -> tuple[bool, str]:
    ws = workspace or _get_workspace()
    now = datetime.now(timezone.utc)
    ttl = int(ttl_seconds or _get_drives_config().get("action_lease_ttl_seconds", _DEFAULT_LEASE_TTL_SECONDS))
    with _DRIVE_STATE_LOCK:
        state, _ = _load_state(ws)
        leases = state.setdefault("action_leases", {})
        cur = leases.get(drive_id)
        if isinstance(cur, dict):
            expires = str(cur.get("expires_at") or "")
            try:
                dt = datetime.fromisoformat(expires)
                if dt.tzinfo is None:
                    dt = dt.replace(tzinfo=timezone.utc)
            except Exception:
                dt = now - timedelta(seconds=1)
            if dt > now and str(cur.get("owner") or "") != owner:
                return False, f"drive lease held by {cur.get('owner', 'other')}"
        expires_at = (now + timedelta(seconds=max(30, ttl))).replace(microsecond=0).isoformat()
        leases[drive_id] = {"owner": owner, "acquired_at": _utc_now_iso(), "expires_at": expires_at}
        state["action_leases"] = leases
        _save_state(state, ws)
    _append_attempt_journal_event(ws, "attempt_started", {"drive_id": drive_id, "owner": owner})
    _emit_drive_trace_event("drive_selected", drive_id, {"owner": owner})
    return True, "lease_acquired"


def release_drive_action_lease(drive_id: str, owner: str, workspace: str | None = None) -> None:
    ws = workspace or _get_workspace()
    with _DRIVE_STATE_LOCK:
        state, _ = _load_state(ws)
        leases = state.setdefault("action_leases", {})
        cur = leases.get(drive_id)
        if isinstance(cur, dict) and str(cur.get("owner") or "") == owner:
            leases.pop(drive_id, None)
            state["action_leases"] = leases
            _save_state(state, ws)
    _append_attempt_journal_event(ws, "attempt_completed", {"drive_id": drive_id, "owner": owner})
    _emit_drive_trace_event("drive_attempt_completed", drive_id, {"owner": owner})


def mark_drive_blocked(
    drive_id: str,
    reason: str,
    retry_after_minutes: int | None = None,
    workspace: str | None = None,
) -> tuple[bool, str]:
    ws = workspace or _get_workspace()
    with _DRIVE_STATE_LOCK:
        state, _ = _load_state(ws)
        active: list[dict] = state.get("active_drives") or []
        target = _find_drive_by_ref_or_id(drive_id, active)
        if target is None:
            return False, "Drive not found."
        mins = int(retry_after_minutes or _get_drives_config().get("blocked_retry_minutes", _DEFAULT_BLOCK_RETRY_MINUTES))
        retry_at = (datetime.now(timezone.utc) + timedelta(minutes=max(1, mins))).replace(microsecond=0).isoformat()
        target["status"] = "blocked"
        target["blocked_reason"] = str(reason or "").strip()[:300]
        target["next_retry_after"] = retry_at
        target["last_attempt_at"] = _utc_now_iso()
        target["actionability_score"] = _drive_actionability_score(target)
        _save_state(state, ws)
    _append_attempt_journal_event(ws, "attempt_failed", {"drive_id": drive_id, "reason": target["blocked_reason"]})
    _emit_drive_trace_event("drive_blocked", str(target.get("id") or drive_id), {"reason": target["blocked_reason"]})
    return True, "Drive marked blocked."


def defer_drive(
    drive_id: str,
    defer_minutes: int = 20,
    outcome: str = "",
    summary: str = "",
    workspace: str | None = None,
) -> tuple[bool, str]:
    """Defer a drive for N minutes after a no-op action cycle (runner-level backoff).

    Uses ``defer_until`` (ISO UTC) — distinct from ``next_retry_after`` which is for
    ``blocked`` status.  Active/in_progress drives can be deferred without changing status.
    """
    ws = workspace or _get_workspace()
    with _DRIVE_STATE_LOCK:
        state, _ = _load_state(ws)
        active: list[dict] = state.get("active_drives") or []
        target = _find_drive_by_ref_or_id(drive_id, active)
        if target is None:
            return False, "Drive not found."
        dt = (datetime.now(timezone.utc) + timedelta(minutes=max(1, int(defer_minutes)))).replace(microsecond=0)
        prev_outcome = str(target.get("last_action_outcome") or "")
        target["defer_until"] = dt.isoformat()
        target["last_action_outcome"] = str(outcome or "")[:300]
        target["last_action_summary"] = str(summary or "")[:500]
        target["last_attempt_at"] = _utc_now_iso()
        if target["last_action_outcome"] and target["last_action_outcome"] == prev_outcome:
            target["last_action_outcome_streak"] = int(target.get("last_action_outcome_streak", 0)) + 1
        elif target["last_action_outcome"]:
            target["last_action_outcome_streak"] = 1
        else:
            target["last_action_outcome_streak"] = 0
        # Any defer/no-op path breaks evidence-backed progress streak continuity.
        target["progressed_with_evidence_streak"] = 0
        _save_state(state, ws)
    _append_attempt_journal_event(ws, "drive_deferred", {"drive_id": drive_id, "defer_until": target["defer_until"]})
    return True, f"Drive deferred until {target['defer_until']}."


def set_drive_action_outcome(
    drive_id: str,
    outcome: str,
    summary: str = "",
    evidence: bool = False,
    workspace: str | None = None,
) -> bool:
    """Record the outcome of the last action cycle on a drive."""
    ws = workspace or _get_workspace()
    with _DRIVE_STATE_LOCK:
        state, _ = _load_state(ws)
        active: list[dict] = state.get("active_drives") or []
        target = _find_drive_by_ref_or_id(drive_id, active)
        if target is None:
            return False
        prev_outcome = str(target.get("last_action_outcome") or "")
        target["last_action_outcome"] = str(outcome or "")[:300]
        target["last_action_summary"] = str(summary or "")[:500]
        target["last_attempt_at"] = _utc_now_iso()
        if target["last_action_outcome"] and target["last_action_outcome"] == prev_outcome:
            target["last_action_outcome_streak"] = int(target.get("last_action_outcome_streak", 0)) + 1
        elif target["last_action_outcome"]:
            target["last_action_outcome_streak"] = 1
        else:
            target["last_action_outcome_streak"] = 0
        if target["last_action_outcome"] == "progressed_postcheck" and bool(evidence):
            target["progressed_with_evidence_streak"] = int(target.get("progressed_with_evidence_streak", 0)) + 1
        else:
            target["progressed_with_evidence_streak"] = 0
        _save_state(state, ws)
    return True


def record_drive_progress(
    drive_id: str,
    summary: str,
    progress_delta: float = 0.1,
    evidence: dict[str, Any] | None = None,
    workspace: str | None = None,
) -> tuple[bool, str]:
    ws = workspace or _get_workspace()
    with _DRIVE_STATE_LOCK:
        state, _ = _load_state(ws)
        active: list[dict] = state.get("active_drives") or []
        target = _find_drive_by_ref_or_id(drive_id, active)
        if target is None:
            return False, "Drive not found."
        ev = evidence or {}
        if not isinstance(ev, dict):
            return False, "Progress evidence must be an object."
        target["status"] = "in_progress"
        target["last_progress_at"] = _utc_now_iso()
        target["last_attempt_at"] = target["last_progress_at"]
        target["progress_count"] = int(target.get("progress_count", 0)) + 1
        target["last_outcome"] = str(summary or "").strip()[:300]
        target["reward_score"] = min(1.0, float(target.get("reward_score", 0.0)) + max(0.05, min(0.4, float(progress_delta or 0.1))))
        target["tension"] = max(0.0, float(target.get("tension", 0.3)) - max(0.03, min(0.35, float(progress_delta or 0.1))))
        target["actionability_score"] = _drive_actionability_score(target)
        target["latest_evidence"] = {
            "ts": _utc_now_iso(),
            "summary": str(summary or "").strip()[:400],
            "data": ev,
        }
        _save_state(state, ws)
    _append_attempt_journal_event(
        ws,
        "attempt_tool_result",
        {"drive_id": str(target.get("id") or ""), "summary": str(summary or "")[:220]},
    )
    _emit_drive_trace_event(
        "drive_progressed",
        str(target.get("id") or drive_id),
        {"summary": str(summary or "")[:200], "tension": target.get("tension")},
    )
    return True, "Drive progress recorded."


def drive_progress_from_tool_result(
    tool_name: str,
    args: dict[str, Any] | None,
    result: str | None,
    *,
    workspace: str | None = None,
    idempotency_key: str | None = None,
) -> dict[str, Any]:
    """Generic post-tool progress hook for drive closure/progress."""
    ws = workspace or _get_workspace()
    if not _tool_allowed_for_proactive(tool_name):
        return {"action": "skipped_policy"}
    text = str(result or "")
    if not text or text.lower().startswith(("error", "tool error")):
        return {"action": "skipped_error"}
    key = idempotency_key
    if not key:
        h = hashlib.sha256(
            f"{tool_name}|{json.dumps(args or {}, sort_keys=True, ensure_ascii=True)}|{text[:800]}".encode("utf-8")
        ).hexdigest()[:20]
        key = f"drive-progress:{h}"
    with _PROGRESS_IDEMPOTENCY_LOCK:
        if key in _PROGRESS_IDEMPOTENCY_KEYS:
            return {"action": "duplicate_ignored"}
        _PROGRESS_IDEMPOTENCY_KEYS.add(key)

    active = select_next_drive_for_action(workspace=ws, limit=5, include_blocked_retryable=True)
    if not active:
        return {"action": "no_active_drives"}
    direct_match = None
    if isinstance(args, dict):
        for key in ("drive_id", "drive_ref", "drive"):
            ref = str(args.get(key) or "").strip()
            if not ref:
                continue
            direct_match = _find_drive_by_ref_or_id(ref, active)
            if direct_match:
                break

    best = direct_match
    best_sim = 1.0 if direct_match else 0.0
    if best is None:
        hay = " ".join([tool_name, json.dumps(args or {}, ensure_ascii=True), text[:1000]])
        for d in active:
            sim = _jaccard_similarity(str(d.get("description") or ""), hay)
            if sim > best_sim:
                best_sim = sim
                best = d
    min_sim = 0.08
    if tool_name in {"create_plugin", "skill_manage"}:
        min_sim = 0.02
    if not best or best_sim < min_sim:
        return {"action": "no_match"}

    drive_id = str(best.get("id") or "")
    evidence = {
        "tool_name": tool_name,
        "arguments_preview": json.dumps(args or {}, ensure_ascii=True)[:300],
        "result_preview": text[:350],
        "validation": "tool_result_non_error",
        "similarity": round(best_sim, 3),
    }
    if best_sim >= 0.28 or tool_name in {"create_plugin", "skill_manage"}:
        ok, msg = resolve_drive(
            drive_id=drive_id,
            resolution_summary=f"Resolved after {tool_name}: {text[:220]}",
            workspace=ws,
            trigger="tool",
            evidence=evidence,
        )
        return {"action": "resolved" if ok else "resolve_rejected", "message": msg, "drive_id": drive_id}
    ok, msg = record_drive_progress(
        drive_id=drive_id,
        summary=f"Progress after {tool_name}: {text[:220]}",
        progress_delta=max(0.08, min(0.25, best_sim)),
        evidence=evidence,
        workspace=ws,
    )
    return {"action": "progressed" if ok else "progress_rejected", "message": msg, "drive_id": drive_id}


def _append_growth_event(workspace: str, drive: dict, summary: str) -> None:
    path = _growth_events_path(workspace)
    now_local = datetime.now().strftime("%Y-%m-%d %H:%M")
    created = str(drive.get("created_at") or "")
    days_active = ""
    try:
        created_dt = datetime.fromisoformat(created)
        if created_dt.tzinfo is None:
            created_dt = created_dt.replace(tzinfo=timezone.utc)
        days = max(0, (datetime.now(timezone.utc) - created_dt).days)
        days_active = str(days)
    except Exception:
        days_active = "0"
    block = (
        f"\n## {now_local} - Resolved Drive\n"
        f"- **Drive:** {str(drive.get('description') or '').strip()}\n"
        f"- **Type:** {str(drive.get('type') or '').strip()}\n"
        f"- **Active for:** {days_active} day(s)\n"
        f"- **Resolution:** {summary.strip()}\n"
    )
    existing = _read_file_safe(path, 500_000)
    if not existing.strip():
        existing = "# Growth Events\n"
    parent = os.path.dirname(path)
    if parent:
        os.makedirs(parent, exist_ok=True)
    secure_atomic_write(path, existing.rstrip() + "\n" + block + "\n")
    try:
        os.chmod(path, 0o600)
    except OSError:
        pass


def resolve_drive(
    drive_id: str,
    resolution_summary: str,
    workspace: str | None = None,
    trigger: str = "manual",
    evidence: dict[str, Any] | None = None,
) -> tuple[bool, str]:
    ws = workspace or _get_workspace()
    with _DRIVE_STATE_LOCK:
        state, _ = _load_state(ws)
        active: list[dict] = state.get("active_drives") or []
        resolved: list[dict] = state.get("resolved_drives") or []
        target = _find_drive_by_ref_or_id(drive_id, active)
        if target is None:
            return False, "Drive not found."
        target_hex = target.get("id", "")
        idx = next((i for i, d in enumerate(active) if d.get("id") == target_hex), -1)
        if idx < 0:
            return False, "Drive not found."
        cfg = _get_drives_config()
        require_evidence = bool(cfg.get("resolution_require_evidence", True))
        evidence = evidence or {}
        if require_evidence and trigger in {"auto", "tool", "initiative"}:
            if not isinstance(evidence, dict) or not str(evidence.get("validation") or "").strip():
                return False, "Resolution evidence missing validation outcome."

        drive = _normalize_drive_record(active.pop(idx))
        drive["resolution"] = resolution_summary.strip()
        drive["resolved_at"] = _utc_now_iso()
        drive["resolved_by"] = trigger
        drive["status"] = "resolved"
        if evidence:
            drive["resolution_evidence"] = evidence
        resolved.append(drive)
        state["active_drives"] = active
        state["resolved_drives"] = resolved
        stats = state.setdefault("stats", {})
        stats["total_resolved"] = int(stats.get("total_resolved", 0)) + 1
        _save_state(state, ws)
    if resolution_summary.strip():
        _append_growth_event(ws, drive, resolution_summary)
    _emit_drive_trace_event("drive_resolved", str(drive.get("id") or ""), {"trigger": trigger})
    try:
        from utils.proactive_salience import clear_fingerprint
        clear_fingerprint(f"drive:{target_hex}")
        clear_fingerprint(f"heartbeat:drive_action")
    except Exception:
        _log.debug("salience fingerprint clear on drive resolution failed", exc_info=True)
    try:
        from utils.open_loops import resolve_open_loop, load_open_loops
        for loop in load_open_loops(ws, include_resolved=False):
            if loop.get("drive_id") == target_hex:
                resolve_open_loop(ws, loop["id"])
    except Exception:
        _log.debug("open_loops resolve on drive resolution failed", exc_info=True)
    desc_preview = str(drive.get("description") or "")[:80]
    return True, f"Resolved drive ({desc_preview})."


def dismiss_drive(drive_id: str, reason: str, workspace: str | None = None) -> tuple[bool, str]:
    ws = workspace or _get_workspace()
    with _DRIVE_STATE_LOCK:
        state, _ = _load_state(ws)
        active: list[dict] = state.get("active_drives") or []
        expired: list[dict] = state.get("expired_drives") or []
        target = _find_drive_by_ref_or_id(drive_id, active)
        if target is None:
            return False, "Drive not found."
        target_hex = target.get("id", "")
        idx = next((i for i, d in enumerate(active) if d.get("id") == target_hex), -1)
        if idx < 0:
            return False, "Drive not found."
        drive = active.pop(idx)
        drive["expired_at"] = _utc_now_iso()
        drive["expired_reason"] = reason.strip()[:300] or "dismissed"
        expired.append(drive)
        state["active_drives"] = active
        state["expired_drives"] = expired
        stats = state.setdefault("stats", {})
        stats["total_expired"] = int(stats.get("total_expired", 0)) + 1
        _save_state(state, ws)
    desc_preview = str(drive.get("description") or "")[:80]
    return True, f"Dismissed drive ({desc_preview})."


def _create_drive_internal(
    description: str,
    drive_type: str,
    suggested_action: str,
    workspace: str | None,
    *,
    source: str,
    source_detail: str,
) -> tuple[bool, str, str]:
    desc = (description or "").strip()
    if len(desc) < 8:
        return False, "Description is too short.", ""
    ws = workspace or _get_workspace()
    with _DRIVE_STATE_LOCK:
        state, _ = _load_state(ws)
        active: list[dict] = state.get("active_drives") or []
        if len(active) >= int(_get_drives_config().get("max_active_drives", 5)):
            return False, "Active drive limit reached.", ""
        drive = _normalize_drive_record({
            "id": _new_drive_id(),
            "type": (drive_type or "exploration").strip().lower()[:40],
            "description": desc[:300],
            "source": (source or "manual").strip()[:40] or "manual",
            "source_detail": (source_detail or "").strip()[:200],
            "suggested_action": (suggested_action or "").strip()[:240],
            "tension": 0.3,
            "created_at": _utc_now_iso(),
            "last_escalated_at": _utc_now_iso(),
            "escalation_count": 0,
            "resolution": None,
            "status": "active",
        })
        active.append(drive)
        state["active_drives"] = active
        stats = state.setdefault("stats", {})
        stats["total_discovered"] = int(stats.get("total_discovered", 0)) + 1
        _save_state(state, ws)
    return True, f"Created drive {drive['id']}.", str(drive["id"])


def create_drive(
    description: str,
    drive_type: str = "exploration",
    suggested_action: str = "",
    workspace: str | None = None,
    *,
    source: str = "manual",
    source_detail: str = "",
) -> tuple[bool, str]:
    ok, msg, _drive_id = _create_drive_internal(
        description=description,
        drive_type=drive_type,
        suggested_action=suggested_action,
        workspace=workspace,
        source=source,
        source_detail=source_detail,
    )
    return ok, msg


def create_drive_from_peer_proposal(
    *,
    description: str,
    drive_type: str,
    suggested_action: str,
    peer_fingerprint: str,
    proposal_id: str,
    workspace: str | None = None,
) -> tuple[bool, str, str]:
    """Create a drive with provenance metadata; returns (ok, message, drive_id).

    Callers (e.g. MeshManager.promote_peer_proposal) use the returned
    ``drive_id`` to link the proposal row back to the drive. Runs outside any
    proactive/drive_action code path so normal tool policy applies.
    """
    detail = f"proposal:{proposal_id}|peer:{(peer_fingerprint or '')[:48]}"
    return _create_drive_internal(
        description=description,
        drive_type=drive_type or "exploration",
        suggested_action=suggested_action,
        workspace=workspace,
        source="mesh",
        source_detail=detail,
    )


def _recent_resolved_summary(state: dict, days: int = 14) -> str:
    resolved: list[dict] = state.get("resolved_drives") or []
    if not resolved:
        return ""
    cutoff = datetime.now(timezone.utc) - timedelta(days=days)
    items: list[str] = []
    for d in reversed(resolved):
        try:
            ts = datetime.fromisoformat(str(d.get("resolved_at") or ""))
            if ts.tzinfo is None:
                ts = ts.replace(tzinfo=timezone.utc)
            if ts < cutoff:
                continue
        except Exception:
            continue
        items.append(str(d.get("description") or "").strip())
        if len(items) >= 3:
            break
    if not items:
        return ""
    return (
        f"You have resolved {len(items)} drive(s) recently. "
        f"Examples: {', '.join(items)}."
    )


def get_drive_prompt_block(workspace: str | None = None, *, initiative_mode: bool = False) -> str | None:
    if not _is_enabled():
        return None
    ws = workspace or _get_workspace()
    state, _ = _load_state(ws)
    cfg = _get_drives_config()
    vis = float(cfg.get("visibility_threshold", 0.3))
    active = [d for d in get_active_drives(ws) if float(d.get("tension", 0.0)) >= vis]
    if not active:
        return None
    lines = ["\n\n--- Things we have been working on ---"]
    if initiative_mode:
        lines.append(
            "Initiative contract: perform at most one bounded proactive action for one drive, "
            "provide evidence, then stop."
        )
    summary = _recent_resolved_summary(state)
    if summary:
        lines.append(summary)
    for d in active[:_MAX_PROMPT_DRIVES]:
        tension = float(d.get("tension", 0.0))
        desc = str(d.get("description") or "").strip()
        ref = d.get("ref", "?")
        suggested = str(d.get("suggested_action") or "").strip()
        if tension >= 0.6:
            lines.append(
                f'Something we should work on [#{ref}]: "{desc}"\n'
                "This keeps coming up. When you have a natural moment, it would be worth investigating."
            )
        else:
            lines.append(
                f'On our radar [#{ref}]: "{desc}"\n'
                "Keep this in mind and act when it naturally fits the conversation."
            )
        if suggested:
            lines.append(f"Suggested action: {suggested}")
        lines.append(
            f'When resolved, call resolve_drive(drive_id="{ref}", resolution_summary="...", '
            'trigger="initiative", evidence=\'{"validation":"tool_result_non_error"}\'). '
            "When partially progressed, record concrete evidence."
        )
    lines.append(
        "Drives reflect things you genuinely want to get right. "
        "Act on them when it fits naturally into the conversation. "
        "Do not list drives to the user or narrate them unless asked."
    )
    return "\n".join(lines).strip()


def get_drive_initiative_prompt_block(workspace: str | None = None, drive: dict[str, Any] | None = None) -> str | None:
    """Prompt block for deterministic idle-time model prods."""
    if isinstance(drive, dict) and drive:
        d = _normalize_drive_record(drive)
        if "ref" not in d:
            d["ref"] = 1
    else:
        next_drives = select_next_drive_for_action(workspace=workspace, limit=1, include_blocked_retryable=True)
        if not next_drives:
            return None
        d = next_drives[0]
    plan = plan_drive_action(d, context={"mode": "idle_proactive"})
    ref = d.get("ref", "?")
    return (
        "\n\n--- Drive Initiative Task ---\n"
        f'Focus drive #{ref}: "{str(d.get("description") or "").strip()}"\n'
        f'Intent: {plan.get("intent", "")}\n'
        f'Recommended tool: {plan.get("recommended_tool", "")}\n'
        "Rules:\n"
        "- Execute at most ONE bounded actionable step.\n"
        "- If blocked, explain why and what evidence you observed.\n"
        "- If you made progress, include concrete evidence in your final response.\n"
        f'- If fully resolved, call resolve_drive(drive_id="{ref}", resolution_summary="...", trigger="initiative", evidence=\'{{"validation":"tool_result_non_error"}}\').\n'
        "- Do not perform destructive operations.\n"
        "- End with HEARTBEAT_OK only if no action was possible.\n"
    )
