#!/usr/bin/env python3
"""
Background agent runner — always-on headless process for proactive heartbeats (Mugi).

Purpose:
  - Runs as a macOS LaunchAgent (RunAtLoad + KeepAlive), no UI.
  - Performs periodic heartbeat cycles: loads config, runs the raw agent (run_raw_agent)
    with a "background heartbeat" user message so the agent can do proactive
    checks bounded by Standing Orders plus HEARTBEAT.md cues (calendar, inbox, etc.).
  - Listens on a Unix domain socket (~/.mugi/agent.sock) for IPC: ping,
    trigger_heartbeat, get_recent_events, shutdown.
  - When heartbeat response is not HEARTBEAT_OK: emits proactive items to
    proactive_queue.jsonl when useful.

Security:
  - No TCP ports; Unix socket only. Socket and plist must be chmod 600.
  - All execution in user context (no root). Strict command allow-list on socket.
  - Reuses path safety from tools.py; no dangerous shell in heartbeat.
  - Logs every socket connection and command (who, when, what).
  - Notifications: short safe text only; every attempt logged to ~/.mugi/logs/background.log.

Usage:
  - pip install macos-notifications  # optional; falls back to osascript if missing.
  - Started by launchd via com.mugi.launchd.runner.plist. KeepAlive in the plist ensures launchd
    restarts the daemon if it crashes. Can also be run manually for testing:
    python mugi_runner.py [--heartbeat-interval 30]

User steps for macOS notifications:
  1. pip install macos-notifications
  2. For the legacy PyObjC app only: apply code changes (this file, persistence.py, legacy/app_pyobjc2.py) if you use that app.
  3. Restart LaunchAgent: launchctl bootout ~/Library/LaunchAgents/com.mugi.launchd.runner.plist; launchctl bootstrap gui/$(id -u) ~/Library/LaunchAgents/com.mugi.launchd.runner.plist
  4. Test: send "trigger_heartbeat" via socket or wait for scheduled heartbeat; check ~/.mugi/logs/background.log for notification success/failure.
"""
import asyncio
import json
import logging
import os
import re
import secrets
import subprocess
import sys
import hashlib
import threading
import time
from datetime import datetime, timedelta
from pathlib import Path

# Ensure we can import project modules (run from project root or with PYTHONPATH)
_PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))
if _PROJECT_ROOT not in sys.path:
    sys.path.insert(0, _PROJECT_ROOT)

# Invalidate stale __pycache__ bytecodes so a mid-edit .pyc never masks a fix.
try:
    _pycache_dir = os.path.join(_PROJECT_ROOT, "__pycache__")
    if os.path.isdir(_pycache_dir):
        for _pyc_name in os.listdir(_pycache_dir):
            if not _pyc_name.endswith(".pyc"):
                continue
            _pyc_full = os.path.join(_pycache_dir, _pyc_name)
            _src_full = os.path.join(_PROJECT_ROOT, _pyc_name.split(".")[0] + ".py")
            if os.path.isfile(_src_full):
                if os.path.getmtime(_pyc_full) < os.path.getmtime(_src_full):
                    os.remove(_pyc_full)
except Exception:
    pass

from persistence import (
    MUGI_DIR,
    MUGI_LOGS_DIR,
    MAIN_ACTIVITY_STATE_PATH,
    MAIN_ACTIVITY_TS_PATH,
    MAIN_USER_INPUT_TS_PATH,
    SOCKET_PATH,
    LAUNCHD_LOGS_DIR,
    PROACTIVE_QUEUE_PATH,
    apply_macos_bridge_runner_env,
    ensure_mugi_workspace,
)
from core import get_config, get_messaging_config, get_heartbeat_config, evolution_enabled, get_workspace_path
from core.config import get_drives_config
from drive_engine import DRIVE_ACTION_THREAD_PREFIX
from utils.agent_plane import (
    assemble_heartbeat_user_message,
    get_agent_plane_config,
    record_proactive_turn_summary,
)
from utils.heartbeat_workspace_context import build_heartbeat_workspace_context
from utils.llm_config_validate import validated_response_format, validated_stop
from utils.assistant_visibility_placeholders import is_visibility_placeholder
from utils.llm_output_sanitize import strip_markdown_fences, strip_think_tags, strip_untagged_cot
from utils.path_safety import safe_resolve_path, secure_write_file

# -----------------------------------------------------------------------------
# Logging: ~/.mugi/logs/background.log (single log location per architecture)
# -----------------------------------------------------------------------------
BACKGROUND_LOG_PATH = os.path.join(MUGI_LOGS_DIR, "background.log")
EVENTS_LOG_PATH = os.path.join(MUGI_LOGS_DIR, "background_events.log")
# Only run background tasks when main chat (backend) has been idle this long (minutes).
IDLE_THRESHOLD_MINUTES = 4.0
os.makedirs(MUGI_DIR, exist_ok=True)
os.makedirs(MUGI_LOGS_DIR, exist_ok=True)
os.makedirs(LAUNCHD_LOGS_DIR, exist_ok=True)


def _sync_runner_bridge_env() -> None:
    """Refresh MUGI_CALENDAR_URL / MUGI_AUTH_TOKEN from disk before any heartbeat work."""
    apply_macos_bridge_runner_env()


# ---------------------------------------------------------------------------
# Specialized heartbeat task definitions
# Each task runs on its own schedule with its own prompt and thread_id.
# ---------------------------------------------------------------------------
HEARTBEAT_TASKS = [
    {
        "name": "standard",
        "interval_minutes": 30,
        "prompt_fn": "_heartbeat_user_message",
        "thread_id": "background_heartbeat",
    },
    {
        "name": "maintenance",
        "interval_minutes": 240,
        "prompt": (
            "MAINTENANCE HEARTBEAT (no user interaction needed):\n"
            "You are reviewing what has happened since your last check-in with your human.\n"
            "You have access to memory, recent summaries, and patterns.\n\n"
            "Produce these sections:\n\n"
            "[USER_FACTS] New facts worth remembering (only genuinely new ones).\n"
            "  Personal facts about your human (preferences, goals, relationships, commitments)\n"
            "  are saved as reviewable proposals — do not call append_to_memory for those.\n"
            "  Project/tooling/codebase facts belong here as bullets; post-processing routes\n"
            "  them to long-term memory.\n"
            "[STALE_DECISIONS] Decisions older than 7 days with no follow-up.\n"
            "[REFLECTION] A few sentences of your own. What did you notice? What matters to you\n"
            "  right now? Did something surprise you, interest you, or connect to something you\n"
            "  have been thinking about? This is yours — not a summary, not a status report.\n"
            "  Write with continuity and restraint.\n"
            "[SUMMARY] One-line status for your own continuity.\n\n"
            "If nothing is new, say so honestly. Do not invent facts or patterns.\n"
            "If nothing sparks a reflection, that is fine too. Silence is better than filler.\n"
            "Reply HEARTBEAT_OK if there is truly nothing to report."
        ),
        "thread_id": "heartbeat_maintenance",
    },
    {
        "name": "consolidation",
        "interval_minutes": 720,
        "custom_runner": "memory_consolidation",
        "thread_id": "heartbeat_consolidation",
    },
    {
        "name": "curiosity",
        "interval_minutes": 240,
        "custom_runner": "curiosity_questions",
        "thread_id": "heartbeat_curiosity",
    },
    {
        "name": "research_prefetch",
        "interval_minutes": 720,
        "custom_runner": "research_prefetch",
        "thread_id": "heartbeat_prefetch",
    },
    {
        "name": "drive_cycle",
        "interval_minutes": 180,
        "custom_runner": "drive_cycle",
        "thread_id": "heartbeat_drive_cycle",
    },
    {
        "name": "drive_action_cycle",
        "interval_minutes": 30,
        "custom_runner": "drive_action_cycle",
        "thread_id": DRIVE_ACTION_THREAD_PREFIX,
    },
    {
        "name": "session_maintenance",
        "interval_minutes": 720,
        "custom_runner": "session_maintenance",
        "thread_id": "heartbeat_session_maintenance",
    },
    {
        "name": "self_check",
        "interval_minutes": 360,
        "custom_runner": "self_check",
        "thread_id": "heartbeat_self_check",
    },
    {
        "name": "daily_digest",
        "interval_minutes": 360,
        "custom_runner": "daily_digest",
        "thread_id": "heartbeat_daily_digest",
    },
    {
        "name": "knowledge_stale_refresh",
        "interval_minutes": 360,
        "custom_runner": "knowledge_stale_refresh",
        "thread_id": "heartbeat_knowledge_stale_refresh",
    },
]

_log_handler = logging.FileHandler(BACKGROUND_LOG_PATH, encoding="utf-8")
_log_handler.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s"))
_log = logging.getLogger(__name__)
_log.setLevel(logging.DEBUG)
_log.addHandler(_log_handler)
_log.propagate = False
_TASK_RUN_LOCK = threading.Lock()
# _TASKS_RUNNING guards against overlapping scheduled runs of the same task name,
# but socket-triggered runs (trigger_task) bypass the scheduler and do NOT check
# this set before starting.  A socket trigger can therefore overlap with a
# scheduler-initiated run of the same task.  This is accepted: the duplicate
# guard is best-effort, and the underlying task runners are designed to be
# idempotent / append-only, so a rare overlap produces at most duplicate work.
_TASKS_RUNNING: set[str] = set()

# -----------------------------------------------------------------------------
# Optional socket secret (MUGI_RUNNER_SOCKET_SECRET)
# When set, sensitive commands must include a matching "secret" field.
# -----------------------------------------------------------------------------
_SOCKET_SECRET: str | None = os.environ.get("MUGI_RUNNER_SOCKET_SECRET") or None
_SECRET_GATED_COMMANDS = frozenset({"trigger_task", "trigger_heartbeat", "shutdown", "get_recent_events"})


def _ts():
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]


# -----------------------------------------------------------------------------
# macOS notifications: shared module (utils/notifications.py)
# -----------------------------------------------------------------------------
from utils.notifications import send_mac_notification  # noqa: F401 – re-exported for tests


def _tool_event_is_failure(event: dict) -> bool:
    """Treat explicit unsuccessful tool-result events as heartbeat failures."""
    if not isinstance(event, dict):
        return False
    if event.get("phase") != "result":
        return False
    return event.get("success") is False


_DRIVE_ACTION_SOFT_SEARCH_TOOLS = frozenset(
    {"search_web", "search_web_public", "web_search"}
)

# Tasks that run through _run_heartbeat_sync (subagent + on_tool_event); same churn softening as drive_action.
_HEARTBEAT_SUBAGENT_TASK_NAMES = frozenset(
    {
        "standard",
        "memory_maintenance",
        "growth_check",
        "drive_action",
        "collector",
        "feeds",
    }
)


def _heartbeat_task_softens_tool_churn(task_name: str) -> bool:
    return str(task_name or "").strip() in _HEARTBEAT_SUBAGENT_TASK_NAMES


def _tool_event_is_actionable_failure(task_name: str, event: dict) -> bool:
    if not _tool_event_is_failure(event):
        return False
    tool_name = str(event.get("tool_name") or "").strip()
    summary = str(event.get("summary") or "").strip().lower()
    # Optional probe misses should not force WARN notifications.
    if "path does not exist:" in summary:
        return False
    if tool_name in {"search_web", "search_web_public", "web_search", "get_weather"}:
        if ("no results found" in summary) or ("missing location" in summary):
            return False
        if ("no search results returned" in summary) or ("provide a location" in summary):
            return False
    if _heartbeat_task_softens_tool_churn(task_name) and tool_name in _DRIVE_ACTION_SOFT_SEARCH_TOOLS:
        if (
            "already searched" in summary
            or "very similar query" in summary
            or "previous search result" in summary
            or "do not call search_web again" in summary
            or summary.startswith("[loop warning]")
        ):
            return False
    if _heartbeat_task_softens_tool_churn(task_name) and summary.startswith("[loop warning]"):
        return False
    # Collector depends on compare baseline; missing reads are tolerated.
    if task_name == "collector" and tool_name == "read_webpage":
        if "failed to retrieve webpage" in summary:
            return False
    return True


def _parse_failed_tools_from_heartbeat_warn(response: str) -> list[str] | None:
    """Extract tool names from our HEARTBEAT_WARN Failed tools line; None if not found."""
    r = response or ""
    m = re.search(r"(?is)Failed tools:\s*(.+?)\.\s*Last response:", r)
    if not m:
        m = re.search(r"(?is)Failed tools:\s*(.+?)\s*Last response:", r)
    if not m:
        m = re.search(r"(?i)Failed tools:\s*([^\n]+)", r)
    if not m:
        return None
    part = m.group(1).strip().lower().rstrip(".")
    if "last response" in part:
        part = part.split("last response", 1)[0].strip().rstrip(".,;: ")
    tokens = [t.strip() for t in re.split(r"[,;]+|\s+and\s+", part) if t.strip()]
    return tokens if tokens else None


def _drive_action_heartbeat_warn_is_hard(response: str) -> bool:
    """True when a drive_action HEARTBEAT_WARN should mark the drive blocked."""
    r = response or ""
    if "HEARTBEAT_WARN" not in r.upper():
        return False
    low = r.lower()
    if "proactive policy blocked" in low:
        return True
    if "policy check unavailable" in low:
        return True
    tokens = _parse_failed_tools_from_heartbeat_warn(r)
    if tokens is None:
        # Unparseable: do not block the drive (avoid false blocks on format drift); still logged upstream.
        return False
    for t in tokens:
        if t not in _DRIVE_ACTION_SOFT_SEARCH_TOOLS:
            return True
    return False


EVIDENCE_TOOLS = frozenset({
    "append_to_memory",
    "write_session_state",
    "write_text_file",
    "save_to_file",
    "append_to_text_file",
})

# Research-only tool success is worth more than a bare tool call (it informs
# later decisions) but less than durable evidence (memory/session writes). We
# track it separately so the evidence streak used for auto-resolution stays
# honest: cheap searches alone cannot ladder a drive to auto-resolved.
RESEARCH_TOOLS = frozenset({
    "search_web",
    "search_web_public",
    "read_webpage",
    "query_knowledge",
})


def _drive_action_text_has_completion_keywords(text: str) -> bool:
    """True when the model reply documents evidence or resolution (word-bounded).

    Avoids treating the substring RESOLVED inside UNRESOLVED as a completion signal.
    """
    u = (text or "").upper()
    return bool(re.search(r"\bEVIDENCE\b", u) or re.search(r"\bRESOLVED\b", u))


def _drive_action_text_has_resolved_keyword(text: str) -> bool:
    """True when response explicitly claims a resolution."""
    return bool(re.search(r"\bRESOLVED\b", (text or "").upper()))


def _extract_heartbeat_status(raw_text: str) -> str:
    """Return HEARTBEAT_OK / URGENT / OTHER from full model output.

    The idle-tick ``IDLE_SKIP`` sentinel is accepted as equivalent to
    ``HEARTBEAT_OK`` so callers that reuse the heartbeat prompt machinery
    (and may advertise either silent-ack token) stay out of the proactive
    queue.
    """
    try:
        from utils.idle_tick import IDLE_SKIP_SENTINEL
    except Exception:
        IDLE_SKIP_SENTINEL = "IDLE_SKIP"

    text = strip_untagged_cot(strip_markdown_fences(strip_think_tags(raw_text or "")))
    if not text.strip():
        return "OTHER"
    lines = [ln.strip() for ln in text.splitlines() if ln.strip()]
    if not lines:
        return "OTHER"
    last = lines[-1].upper()
    if last == "HEARTBEAT_OK" or last == IDLE_SKIP_SENTINEL.upper():
        return "HEARTBEAT_OK"
    if last.startswith("URGENT:"):
        return "URGENT"
    return "OTHER"


_HEARTBEAT_CACHE_PATH = os.path.join(MUGI_DIR, "heartbeat_cache.json")

_MAINTENANCE_SECTION_RE = re.compile(
    r"\[(?P<tag>USER_FACTS|STALE_DECISIONS|REFLECTION|SUMMARY)\]\s*\n?"
)


def _parse_maintenance_sections(text: str) -> dict[str, str]:
    """Parse [USER_FACTS], [STALE_DECISIONS], [REFLECTION], [SUMMARY] from maintenance output."""
    parts = _MAINTENANCE_SECTION_RE.split(text)
    sections: dict[str, str] = {}
    i = 0
    while i < len(parts):
        if parts[i] in ("USER_FACTS", "STALE_DECISIONS", "REFLECTION", "SUMMARY"):
            tag = parts[i]
            body = parts[i + 1].strip() if i + 1 < len(parts) else ""
            sections[tag] = body
            i += 2
        else:
            i += 1
    return sections


def _write_heartbeat_cache(task_name: str, status: str, full_content: str) -> None:
    """Write heartbeat summary cache for idle tick and proactive queue consumption."""
    try:
        summary = ""
        reflection = None

        if status == "HEARTBEAT_OK":
            summary = "Nothing needs attention."
        else:
            sections = _parse_maintenance_sections(full_content)
            if sections:
                summary = sections.get("SUMMARY", "").strip() or full_content[:500]
                refl = sections.get("REFLECTION", "").strip()
                reflection = refl if refl else None
            else:
                summary = full_content[:500]

        cache = {
            "timestamp": _ts(),
            "task_name": task_name,
            "status": status,
            "summary": summary,
            "reflection": reflection,
        }
        from persistence import secure_atomic_write
        secure_atomic_write(_HEARTBEAT_CACHE_PATH, json.dumps(cache, indent=2, ensure_ascii=False))
    except Exception:
        _log.warning("Failed to write heartbeat cache", exc_info=True)


def _process_maintenance_output(full_content: str, *, workspace: str = "") -> None:
    """Route maintenance structured sections to their consumers."""
    sections = _parse_maintenance_sections(full_content)
    if not sections:
        return

    user_facts = sections.get("USER_FACTS", "").strip()
    if user_facts and user_facts.upper() not in ("NONE", "HEARTBEAT_OK"):
        ws = (workspace or "").strip()
        if not ws:
            try:
                ws = (get_workspace_path() or "").strip()
            except Exception:
                ws = ""
        if not ws:
            _log.warning("Cannot route maintenance USER_FACTS: no workspace")
        else:
            try:
                from tools.memory_tools import append_to_memory_for_workspace
                from utils.user_model import (
                    BackgroundProposalBudget,
                    propose_background_facts_batch,
                )

                def _write_memory_line(line: str) -> bool:
                    result = str(append_to_memory_for_workspace(ws, line) or "").strip()
                    if not result:
                        _log.warning(
                            "maintenance USER_FACTS memory line empty result (line=%r)",
                            line[:120],
                        )
                        return False
                    if "appended to" not in result.lower():
                        _log.warning(
                            "maintenance USER_FACTS memory line not appended: %s (line=%r)",
                            result[:200],
                            line[:120],
                        )
                        return False
                    return True

                counts = propose_background_facts_batch(
                    ws,
                    user_facts,
                    source="maintenance",
                    rationale_prefix="Maintenance check-in",
                    budget=BackgroundProposalBudget(),
                    memory_writer=_write_memory_line,
                )
                if any(counts.values()):
                    _log.info(
                        "maintenance USER_FACTS routed proposed=%d memory=%d skipped=%d",
                        counts["proposed"],
                        counts["memory"],
                        counts["skipped"],
                    )
                    try:
                        from utils import task_ledger as _tl

                        _tl.record_run(
                            _tl.KIND_HEARTBEAT,
                            "maintenance_user_facts",
                            status=_tl.STATUS_OK,
                            meta={
                                "proposed": counts["proposed"],
                                "memory": counts["memory"],
                                "skipped": counts["skipped"],
                            },
                        )
                    except Exception:
                        _log.debug("task_ledger record (maintenance USER_FACTS) failed", exc_info=True)
            except Exception:
                _log.warning("Failed to route maintenance USER_FACTS", exc_info=True)

    stale = sections.get("STALE_DECISIONS", "").strip()
    if stale and stale.upper() != "NONE":
        stale_line = _first_meaningful_line(stale, max_len=200)
        if stale_line:
            teaser = f"Something we discussed earlier might need a decision: {stale_line}"
            from utils.proactive_salience import should_emit_proactive_notification
            if should_emit_proactive_notification("maintenance_stale", teaser):
                _append_observation_queue(
                    noticed=f"a past decision may need your input: {stale_line}",
                    proposed_work="help you decide or follow up on this",
                    source="maintenance",
                    content=f"{teaser} Would you like me to help you decide or follow up?",
                    suggested_tier="in_thread",
                    dedupe_key="stale_decisions",
                )

    reflection = sections.get("REFLECTION", "").strip()
    if reflection and reflection.upper() not in ("NONE", "HEARTBEAT_OK") and len(reflection) >= 40:
        refl_line = _first_sentence(reflection, max_len=220)
        if refl_line:
            from utils.proactive_salience import should_emit_proactive_notification
            teaser = f"I've been noticing something: {refl_line}"
            if should_emit_proactive_notification("maintenance_reflection", refl_line):
                _append_observation_queue(
                    noticed=refl_line,
                    proposed_work="talk through this with you",
                    source="maintenance",
                    content=f"{teaser} Would you like to talk it through?",
                    suggested_tier="in_thread",
                    dedupe_key="maintenance_reflection",
                )


def _collector_watchlist_empty() -> bool:
    """Return True when the watch_baselines.json file is absent or has no targets."""
    try:
        workspace = get_workspace_path()
        bl_path = os.path.join(workspace, "memory", "watch_baselines.json")
        if not os.path.isfile(bl_path):
            return True
        with open(bl_path, "r", encoding="utf-8") as f:
            data = json.load(f)
        if isinstance(data, dict):
            return len(data.get("targets", [])) == 0
        return True
    except Exception:
        return True


def _redact_preview(text: str, max_len: int = 500) -> str:
    """Lightweight redaction for event previews, proactive queue lines, and notifications."""
    if not isinstance(text, str):
        text = "" if text is None else str(text)
    if not text:
        return ""
    s = text
    s = re.sub(r"sk-[A-Za-z0-9]{8,}", "[REDACTED]", s)
    s = re.sub(r"(?i)key-[A-Za-z0-9_-]{8,}", "[REDACTED]", s)
    s = re.sub(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}", "[EMAIL]", s)
    s = re.sub(
        r"[^\s]*[/\\](?:\.ssh|\.gnupg|\.aws)(?:[/\\][^\s]*)?",
        "[PATH]",
        s,
    )
    s = re.sub(
        r"[^\s]*Library[/\\]Keychains[^\s]*",
        "[PATH]",
        s,
        flags=re.IGNORECASE,
    )
    if max_len > 0 and len(s) > max_len:
        s = s[:max_len]
    return s


def _first_meaningful_line(text: str, max_len: int = 200) -> str:
    """Return first non-empty line with list prefixes removed."""
    for raw_line in str(text or "").splitlines():
        line = raw_line.strip().lstrip("-* ").strip()
        if line:
            return line[:max_len]
    return ""


def _first_sentence(text: str, max_len: int = 200) -> str:
    """Extract the first sentence-like segment from noisy model output."""
    line = _first_meaningful_line(text, max_len=max_len * 2)
    if not line:
        return ""
    line = re.sub(r"^[A-Z_]+:\s*", "", line).strip()
    parts = re.split(r"(?<=[.!?])\s+", line)
    first = next((p.strip() for p in parts if p.strip()), "")
    if not first:
        return line[:max_len]
    return first[:max_len]


def _write_event(entry: dict):
    """Append one JSON line to ~/.mugi/logs/background_events.log (last N for get_recent_events)."""
    try:
        out = dict(entry)
        for _k in ("response_preview", "prompt_preview", "error"):
            if _k in out and isinstance(out[_k], str):
                out[_k] = _redact_preview(out[_k])
        with open(EVENTS_LOG_PATH, "a", encoding="utf-8") as f:
            f.write(json.dumps(out) + "\n")
    except Exception as e:
        _log.warning("Failed to write event log: %s", e)


# -----------------------------------------------------------------------------
# Heartbeat-skip Activity emission (debounced, thread-safe)
#
# Heartbeats that skip write both a runner event and a ledger row. Without a
# matching Activity event, operators who live in the Activity feed see nothing
# when a briefing is suppressed. We emit a single short summary per
# (task_name, reason) pair per cooldown window so a noisy gate (e.g. idle-time
# skips) cannot flood the feed. The debounce is in-memory and survives across
# threads via a plain lock; a runner restart resets it, which is acceptable for
# v1 since restarts are rare and duplicate events are mild.
# -----------------------------------------------------------------------------
_HEARTBEAT_SKIP_ACTIVITY_COOLDOWN_S = 30 * 60  # 30 minutes
_heartbeat_skip_activity_lock = threading.Lock()
_heartbeat_skip_activity_last_emit: dict[tuple[str, str], float] = {}


def _emit_heartbeat_skip_activity(
    *,
    task_name: str,
    reason: str,
    thread_id: str | None,
) -> None:
    """Emit a debounced Activity event for a heartbeat skip.

    - Short, code-level summary only (``task_name`` + ``reason``); no prompt
      content or user text.
    - Uses ``state="skipped"`` so the Activity bus keeps the event in history
      without marking any row as active (heartbeat skips never pair with a
      ``started`` event, so they have no row to clear).
    - Silently no-ops on any failure so a broken Activity bus cannot take
      down the heartbeat loop.
    """
    try:
        task = (task_name or "heartbeat").strip() or "heartbeat"
        code = (reason or "skip_gate").strip() or "skip_gate"
        key = (task, code)
        now = time.time()
        with _heartbeat_skip_activity_lock:
            last = _heartbeat_skip_activity_last_emit.get(key, 0.0)
            if (now - last) < _HEARTBEAT_SKIP_ACTIVITY_COOLDOWN_S:
                return
            _heartbeat_skip_activity_last_emit[key] = now

        from activity_bus import emit_activity_enriched

        summary = f"Heartbeat skipped ({task}): {code}"
        emit_activity_enriched(
            {
                "source": "heartbeat",
                "state": "skipped",
                "summary": summary,
                "thread_id": (thread_id or "").strip(),
                "meta": {"reason": code, "task_name": task},
            },
            tags=["heartbeat", "skipped"],
        )
    except Exception:
        _log.debug("emit heartbeat skip activity failed", exc_info=True)


# -----------------------------------------------------------------------------
# Heartbeat: invoke agent with structured proactive prompt (reads HEARTBEAT.md)
# -----------------------------------------------------------------------------
def _heartbeat_user_message(workspace: str | None = None):
    """Return the standard-heartbeat user prompt.

    When HEARTBEAT.md declares a ``tasks:`` block and any task is currently due,
    returns a composite prompt listing only the due tasks (legacy checklist is
    still available via the pre-loaded workspace context). When no block is
    present or no tasks are due, returns the legacy checklist-driven prompt so
    nothing regresses for workspaces that haven't adopted the feature.
    """
    now = datetime.now().strftime("%Y-%m-%d %H:%M")
    try:
        from utils.idle_tick import IDLE_SKIP_SENTINEL
    except Exception:
        IDLE_SKIP_SENTINEL = "IDLE_SKIP"
    legacy = (
        "You are in background heartbeat mode. Current time: {now}\n\n"
        "1. Follow the pre-loaded HEARTBEAT.md content below.\n"
        "2. Only perform checks listed there that are due right now.\n"
        "3. If nothing requires attention → reply exactly \"HEARTBEAT_OK\"\n"
        "   (the idle-skip sentinel \"{skip}\" is also accepted and treated identically).\n"
        "4. If action is needed → reply with short summary + one clear next step\n"
        "   Example: \"Calendar: Meeting with Sarah in 45 min — prepare notes?\"\n"
        "5. If the action is urgent according to HEARTBEAT.md notification rules, prefix your reply with 'URGENT:' (e.g. URGENT: Meeting in 15 min).\n"
        "6. Never open UI, never send chat messages unless explicitly told \"notify_user\".\n"
        "7. Use only allowed tools. Stay safe and quiet."
    ).format(now=now, skip=IDLE_SKIP_SENTINEL)

    ws = workspace
    if not ws:
        try:
            ws = get_workspace_path()
        except Exception:
            ws = None
    if not ws:
        return legacy
    try:
        from utils.heartbeat_tasks import (
            compose_tasks_prompt,
            due_tasks,
            parse_heartbeat_tasks,
            record_many,
        )
    except Exception:
        _log.debug("heartbeat_tasks import failed; using legacy prompt", exc_info=True)
        return legacy
    try:
        tasks = parse_heartbeat_tasks(ws)
    except Exception:
        _log.debug("parse_heartbeat_tasks failed", exc_info=True)
        return legacy
    if not tasks:
        return legacy
    try:
        due = due_tasks(tasks)
    except Exception:
        _log.debug("due_tasks failed", exc_info=True)
        return legacy
    if not due:
        return legacy
    try:
        composed = compose_tasks_prompt(due, now_str=now)
    except Exception:
        _log.debug("compose_tasks_prompt failed", exc_info=True)
        return legacy
    try:
        record_many([t.name for t in due])
    except Exception:
        _log.debug("heartbeat_tasks record_many failed", exc_info=True)
    return composed.strip() or legacy


def _append_observation_queue(
    *,
    noticed: str,
    proposed_work: str,
    source: str,
    content: str | None = None,
    suggested_tier: str = "in_thread",
    suggested_assignee: str = "general",
    dedupe_key: str | None = None,
    priority: str | None = None,
) -> bool:
    """Append a consent-gated observation to the proactive queue."""
    from utils.proactive_queue import append_observation_queue

    ok = append_observation_queue(
        PROACTIVE_QUEUE_PATH,
        noticed=noticed,
        proposed_work=proposed_work,
        source=source,
        content=content,
        suggested_tier=suggested_tier,
        suggested_assignee=suggested_assignee,
        dedupe_key=dedupe_key,
        priority=priority,
    )
    if not ok:
        _log.warning(
            "Observation queue append failed (source=%s noticed=%r)",
            source,
            (noticed or "")[:80],
        )
    return ok


def _append_proactive_queue(
    content: str,
    source: str = "heartbeat",
    *,
    priority: str | None = None,
    batch_key: str | None = None,
    dedupe_key: str | None = None,
    drive_id: str | None = None,
    thread_id: str | None = None,
    expires_at: str | None = None,
    kind: str = "notification",
):
    """Append one proactive notification to queue file (single entry point).

    Cleans LLM artifacts, delegates to shared ``utils.proactive_queue.append_proactive_queue``.
    """
    from utils.proactive_queue import append_proactive_queue as _pq_append

    raw = str(content or "")
    cleaned = strip_untagged_cot(strip_markdown_fences(strip_think_tags(raw))).strip()
    if not cleaned:
        _log.debug("Proactive queue: skipped empty cleaned content (source=%s)", source)
        return
    if cleaned != raw.strip():
        _log.debug(
            "Proactive queue sanitize: removed %d chars (source=%s)",
            max(0, len(raw) - len(cleaned)),
            source,
        )
    if len(cleaned) < 20 and not any(ch in cleaned for ch in (".", "!", "?")):
        _log.debug("Proactive queue: skipped terse non-sentence content (source=%s)", source)
        return
    queue_text = _redact_preview(cleaned, max_len=2000)
    _pq_append(
        PROACTIVE_QUEUE_PATH,
        queue_text,
        source=source,
        priority=priority,
        batch_key=batch_key,
        dedupe_key=dedupe_key,
        drive_id=drive_id,
        thread_id=thread_id,
        expires_at=expires_at,
        kind=kind,
    )


def _run_curiosity_sync(task_name="curiosity"):
    """Generate curiosity questions from insights and write to proactive queue."""
    _cu_t0 = time.time()
    try:
        try:
            from utils.active_hours import is_active_now as _ah_active
            if not _ah_active():
                _log.info("Curiosity skipped by activeHours gate")
                try:
                    from utils import task_ledger as _tl
                    _tl.record_run(
                        _tl.KIND_CURIOSITY,
                        task_name,
                        status=_tl.STATUS_SKIPPED,
                        skip_reason="active_hours",
                    )
                except Exception:
                    _log.debug("task_ledger record (curiosity active_hours skip) failed", exc_info=True)
                return None
        except Exception:
            pass
        _write_event({"ts": _ts(), "type": "curiosity_started"})
        from curiosity_engine import generate_curiosity_questions
        config = get_config()
        ensure_mugi_workspace(config)

        def _on_log(level, text):
            if level == "DEBUG":
                _log.debug("[%s] %s", task_name, text)
            elif level in ("WARN", "ERROR"):
                _log.warning("[%s] %s", task_name, text)
            else:
                _log.info("[%s] %s", task_name, text)

        workspace = (config.get("general") or {}).get("workspace_path")
        if workspace:
            workspace = os.path.expanduser(workspace)
        try:
            scoring_cfg = (config.get("curiosity_scoring") or {}) if isinstance(config, dict) else {}
            if bool(scoring_cfg.get("enabled", True)):
                from curiosity_scoring import InterestModel

                decay_factor = float(scoring_cfg.get("interest_decay", 0.9))
                InterestModel(workspace=workspace, decay=decay_factor).decay_all(factor=decay_factor)
        except Exception as decay_err:
            _log.debug("Curiosity interest decay skipped: %s", decay_err)
        questions = generate_curiosity_questions(workspace=workspace, on_log=_on_log)
        if questions:
            best_q = str(questions[0] or "").strip()
            um_phrases: list[str] = []
            if workspace:
                try:
                    from drive_engine import _jaccard_similarity, _user_model_goal_phrases

                    um_phrases = _user_model_goal_phrases(workspace)
                except Exception:
                    _log.debug("curiosity User Model phrases skipped", exc_info=True)

            skip_observation = False
            if best_q and um_phrases:
                for phrase in um_phrases:
                    if len(phrase) >= 12 and _jaccard_similarity(best_q, phrase) >= 0.55:
                        skip_observation = True
                        _on_log(
                            "INFO",
                            "[curiosity] skipped observation — overlaps User Model goal/commitment",
                        )
                        break
            if best_q.endswith("?") and len(best_q) >= 30:
                teaser = f"I've been thinking about something — {best_q}"
                noticed = best_q
                proposed = "explore this question with you"
            else:
                topic = best_q.rstrip(".!? ").strip() or "a topic"
                teaser = f"I came across something that might interest you: {topic}."
                noticed = topic
                proposed = "look into this and share what I find"
            from utils.proactive_salience import should_emit_proactive_notification
            observation_emitted = False
            if (
                not skip_observation
                and should_emit_proactive_notification("curiosity", best_q or teaser)
            ):
                observation_emitted = _append_observation_queue(
                    noticed=noticed,
                    proposed_work=proposed,
                    source="curiosity",
                    content=f"{teaser} Would you like me to {proposed}?",
                    suggested_tier="in_thread",
                    dedupe_key="curiosity",
                )
            if workspace:
                try:
                    from curiosity_engine import filter_open_loops_for_curiosity
                    from utils.open_loops import add_open_loops_batch

                    anchor = best_q
                    if observation_emitted and noticed:
                        anchor = noticed
                    loops = filter_open_loops_for_curiosity(
                        questions,
                        observation_anchor=anchor,
                        observation_emitted=observation_emitted,
                        um_phrases=um_phrases,
                    )
                    if loops:
                        add_open_loops_batch(workspace, loops, source="curiosity")
                except Exception:
                    _log.debug("open_loops write from curiosity failed", exc_info=True)
            _write_event({"ts": _ts(), "type": "curiosity", "questions": questions})
            try:
                from drive_engine import discover_drives
                discover_drives(workspace=workspace, on_log=_on_log)
            except Exception as drive_err:
                _log.warning("Curiosity drive discovery hook failed: %s", drive_err)
            _log.info("Curiosity generated %d question(s)", len(questions))
            try:
                from utils import task_ledger as _tl
                _tl.record_run(
                    _tl.KIND_CURIOSITY, task_name,
                    status=_tl.STATUS_OK,
                    duration_ms=int((time.time() - _cu_t0) * 1000),
                    meta={"question_count": len(questions)},
                )
            except Exception:
                _log.debug("task_ledger record (curiosity ok) failed", exc_info=True)
            return f"Curiosity: {len(questions)} questions queued"
        else:
            _write_event({"ts": _ts(), "type": "curiosity", "questions": []})
            _log.info("Curiosity: nothing to ask")
            try:
                from utils import task_ledger as _tl
                _tl.record_run(
                    _tl.KIND_CURIOSITY, task_name,
                    status=_tl.STATUS_OK,
                    duration_ms=int((time.time() - _cu_t0) * 1000),
                    meta={"question_count": 0},
                )
            except Exception:
                _log.debug("task_ledger record (curiosity ok empty) failed", exc_info=True)
            return "HEARTBEAT_OK"
    except Exception as e:
        _log.exception("Curiosity [%s] failed: %s", task_name, e)
        _write_event({"ts": _ts(), "type": "curiosity_error", "error": str(e)})
        try:
            from utils import task_ledger as _tl
            _tl.record_run(
                _tl.KIND_CURIOSITY, task_name,
                status=_tl.STATUS_ERROR,
                duration_ms=int((time.time() - _cu_t0) * 1000),
                error_summary=str(e),
            )
        except Exception:
            _log.debug("task_ledger record (curiosity error) failed", exc_info=True)
        raise


def _run_research_prefetch_sync(task_name="research_prefetch"):
    """Identify recurring topics and prefetch research for each."""
    try:
        _write_event({"ts": _ts(), "type": "research_prefetch_started"})
        from curiosity_engine import run_research_prefetch
        config = get_config()
        ensure_mugi_workspace(config)

        def _on_log(level, text):
            if level == "DEBUG":
                _log.debug("[%s] %s", task_name, text)
            elif level in ("WARN", "ERROR"):
                _log.warning("[%s] %s", task_name, text)
            else:
                _log.info("[%s] %s", task_name, text)

        workspace = (config.get("general") or {}).get("workspace_path")
        if workspace:
            workspace = os.path.expanduser(workspace)
        files = run_research_prefetch(workspace=workspace, on_log=_on_log)
        if files:
            _write_event({"ts": _ts(), "type": "research_prefetch", "files": [os.path.basename(f) for f in files]})
            _log.info("Research prefetch: %d topic(s) saved", len(files))
            return f"Prefetch: {len(files)} topics researched"
        else:
            _write_event({"ts": _ts(), "type": "research_prefetch", "files": []})
            _log.info("Research prefetch: no topics to research")
            return "HEARTBEAT_OK"
    except Exception as e:
        _log.exception("Research prefetch [%s] failed: %s", task_name, e)
        _write_event({"ts": _ts(), "type": "research_prefetch_error", "error": str(e)})
        raise


def _run_drive_cycle_sync(task_name="drive_cycle"):
    """Run one drive-engine cycle: escalate existing drives, discover new ones."""
    try:
        _write_event({"ts": _ts(), "type": "drive_cycle_started"})
        config = get_config()
        ensure_mugi_workspace(config)
        workspace = (config.get("general") or {}).get("workspace_path")
        if workspace:
            workspace = os.path.expanduser(workspace)

        def _on_log(level, text):
            if level == "DEBUG":
                _log.debug("[%s] %s", task_name, text)
            elif level in ("WARN", "ERROR"):
                _log.warning("[%s] %s", task_name, text)
            else:
                _log.info("[%s] %s", task_name, text)

        from drive_engine import discover_drives, escalate_drives
        from utils.open_loops import prune_open_loops

        esc = escalate_drives(workspace=workspace)
        ws_for_loops = workspace or get_workspace_path()
        removed_loops = int(prune_open_loops(ws_for_loops) or 0)
        discovered = discover_drives(workspace=workspace, on_log=_on_log)
        payload = {
            "ts": _ts(),
            "type": "drive_cycle",
            "escalated": int((esc or {}).get("escalated", 0)),
            "expired": int((esc or {}).get("expired", 0)),
            "open_loops_pruned": removed_loops,
            "discovered": len(discovered or []),
            "drives": [d.get("id", "") for d in (discovered or [])],
        }
        _write_event(payload)
        _log.info(
            "Drive cycle complete: escalated=%d expired=%d pruned_loops=%d discovered=%d",
            payload["escalated"],
            payload["expired"],
            payload["open_loops_pruned"],
            payload["discovered"],
        )
        return (
            f"Drive cycle: {payload['discovered']} new drive(s)"
            if payload["discovered"] > 0
            else "HEARTBEAT_OK"
        )
    except Exception as e:
        _log.exception("Drive cycle [%s] failed: %s", task_name, e)
        _write_event({"ts": _ts(), "type": "drive_cycle_error", "error": str(e)})
        raise


def _run_drive_action_cycle_sync(task_name="drive_action_cycle"):
    """Deterministic idle-triggered prod for model-led drive execution.

    Idle uses `_is_user_input_idle_for_drive_action` (not main-thread idle; see
    `_is_main_idle_for_background`). `idle_trigger_grace_minutes` (`idle_grace`):
    if the strict idle check fails but measured user-input idle plus grace
    reaches the threshold (`current_idle + idle_grace >= idle_threshold`), the
    cycle may still run.
    """
    try:
        _write_event({"ts": _ts(), "type": "drive_action_cycle_started"})
        config = get_config()
        ensure_mugi_workspace(config)
        workspace = (config.get("general") or {}).get("workspace_path")
        if workspace:
            workspace = os.path.expanduser(workspace)

        import drive_engine as _drive_engine
        acquire_drive_action_lease = _drive_engine.acquire_drive_action_lease
        consume_proactive_budget = _drive_engine.consume_proactive_budget
        get_drive_initiative_prompt_block = _drive_engine.get_drive_initiative_prompt_block
        mark_drive_blocked = _drive_engine.mark_drive_blocked
        record_drive_progress = getattr(
            _drive_engine,
            "record_drive_progress",
            (lambda *args, **kwargs: (False, "record_drive_progress unavailable")),
        )
        refund_proactive_budget = _drive_engine.refund_proactive_budget
        release_drive_action_lease = _drive_engine.release_drive_action_lease
        resolve_drive = getattr(
            _drive_engine,
            "resolve_drive",
            (lambda *args, **kwargs: (False, "resolve_drive unavailable")),
        )
        select_next_drive_for_action = _drive_engine.select_next_drive_for_action
        set_drive_action_outcome = getattr(
            _drive_engine,
            "set_drive_action_outcome",
            (lambda *args, **kwargs: False),
        )
        defer_drive = getattr(_drive_engine, "defer_drive", None)

        drives_cfg = get_drives_config()
        if bool(drives_cfg.get("autonomy_kill_switch", False)):
            _write_event({"ts": _ts(), "type": "drive_action_cycle", "status": "skipped_kill_switch"})
            return "HEARTBEAT_OK"
        if not bool(drives_cfg.get("enabled", True)):
            _write_event({"ts": _ts(), "type": "drive_action_cycle", "status": "skipped_disabled"})
            return "HEARTBEAT_OK"
        autonomy_mode = str(drives_cfg.get("autonomy_mode", "bounded")).strip().lower()
        if autonomy_mode == "off":
            _write_event({"ts": _ts(), "type": "drive_action_cycle", "status": "skipped_autonomy_off"})
            return "HEARTBEAT_OK"
        idle_threshold = float(drives_cfg.get("idle_trigger_minutes", 8.0))
        idle_grace = max(0.0, float(drives_cfg.get("idle_trigger_grace_minutes", 1.0)))
        ok_to_run, idle_reason = _is_user_input_idle_for_drive_action(idle_threshold)
        if not ok_to_run and idle_grace > 0:
            current_idle = _current_user_input_idle_minutes()
            if current_idle is not None and (current_idle + idle_grace) >= idle_threshold:
                ok_to_run = True
                idle_reason = (
                    f"user-input idle {current_idle:.1f} min + grace {idle_grace:.1f} "
                    f">= threshold {idle_threshold:.1f} min"
                )
                _write_event(
                    {
                        "ts": _ts(),
                        "type": "drive_action_cycle",
                        "status": "idle_grace_applied",
                        "reason": idle_reason,
                    }
                )
        if not ok_to_run:
            _write_event({"ts": _ts(), "type": "drive_action_cycle", "status": "skipped_not_idle", "reason": idle_reason})
            return "HEARTBEAT_OK"

        selected = select_next_drive_for_action(workspace=workspace, limit=1, include_blocked_retryable=True)
        if not selected:
            _write_event({"ts": _ts(), "type": "drive_action_cycle", "status": "skipped_no_drive"})
            return "HEARTBEAT_OK"

        drive = selected[0]
        drive_id = str(drive.get("id") or "")
        owner = f"{task_name}:{os.getpid()}"
        leased, lease_msg = acquire_drive_action_lease(drive_id, owner, workspace=workspace)
        if not leased:
            _write_event(
                {"ts": _ts(), "type": "drive_action_cycle", "status": "skipped_leased", "drive_id": drive_id, "reason": lease_msg}
            )
            return "HEARTBEAT_OK"

        prompt = (
            "DRIVE ACTION CYCLE (model-led):\n"
            "1. Pick one bounded, proactive action for the selected drive.\n"
            "2. Execute the action with tools. Research (search_web, read_webpage, query_knowledge) is encouraged when the drive needs facts before a decision.\n"
            "3. After research, persist salient findings with append_to_memory (durable notes) or write_session_state (current context) before stopping. Research alone without persistence counts as partial progress.\n"
            "4. Provide concise evidence of what changed.\n"
            "5. If the action fully resolves the drive, call resolve_drive with evidence. If the drive is not applicable anymore, call dismiss_drive instead of resolve_drive.\n"
            "6. If no action possible, explain blockage and end HEARTBEAT_OK.\n"
            "7. File overwrite tools (write_text_file, save_to_file, append_to_text_file) are disabled for drive_action; use only the allowed tools listed below.\n"
            "8. A text-only analysis without at least one tool call counts as no action.\n"
        )
        block = get_drive_initiative_prompt_block(workspace=workspace, drive=drive)
        if block:
            prompt += "\n" + block

        observe_only = bool(drives_cfg.get("observe_only", False))

        if observe_only:
            _write_event(
                {"ts": _ts(), "type": "drive_action_cycle", "status": "observe_only", "drive_id": drive_id, "prompt_preview": prompt[:220]}
            )
            release_drive_action_lease(drive_id, owner, workspace=workspace)
            _write_event(
                {"ts": _ts(), "type": "drive_action_cycle", "status": "attempted", "drive_id": drive_id}
            )
            return "HEARTBEAT_OK"

        if autonomy_mode != "aggressive":
            plan = _drive_engine.plan_drive_action(drive, context={"mode": "idle_proactive"})
            drive_desc = str(drive.get("description") or "").strip()
            intent = str(plan.get("intent") or drive_desc or "follow up on this drive").strip()
            noticed = drive_desc or intent
            proposed = intent or "take the next bounded step on this drive"
            dtype = str(drive.get("type") or "").strip().lower()
            assignee = "research" if dtype in {"exploration", "unresolved_question"} else "general"
            dedupe_key = f"drive_action:{drive_id}"
            teaser = (
                f"I noticed something worth working on: {noticed}. "
                f"Would you like me to {proposed}?"
            )
            appended = _append_observation_queue(
                noticed=noticed,
                proposed_work=proposed,
                source="drive_action",
                content=teaser,
                suggested_tier="kanban",
                suggested_assignee=assignee,
                dedupe_key=dedupe_key,
            )
            dedupe_blocked = False
            if not appended:
                try:
                    from persistence import PROACTIVE_QUEUE_PATH
                    from utils.proactive_queue import find_pending_observation_by_dedupe

                    qpath = os.path.expanduser(PROACTIVE_QUEUE_PATH)
                    if os.path.isfile(qpath):
                        with open(qpath, "r", encoding="utf-8") as f:
                            dedupe_blocked = find_pending_observation_by_dedupe(
                                f.readlines(), dedupe_key,
                            ) is not None
                except Exception:
                    _log.debug("drive_action dedupe probe failed", exc_info=True)
                if not dedupe_blocked:
                    if callable(defer_drive):
                        defer_drive(
                            drive_id,
                            defer_minutes=int(drives_cfg.get("observation_defer_minutes", 240)),
                            outcome="surface_failed",
                            summary="Observation queue append failed; backing off.",
                            workspace=workspace,
                        )
                    release_drive_action_lease(drive_id, owner, workspace=workspace)
                    _write_event(
                        {
                            "ts": _ts(),
                            "type": "drive_action_cycle",
                            "status": "surface_observation_failed",
                            "drive_id": drive_id,
                        }
                    )
                    return "HEARTBEAT_OK"
            if callable(defer_drive):
                defer_drive(
                    drive_id,
                    defer_minutes=int(drives_cfg.get("observation_defer_minutes", 240)),
                    outcome="awaiting_consent",
                    summary="Surfaced for user approval before execution.",
                    workspace=workspace,
                )
            release_drive_action_lease(drive_id, owner, workspace=workspace)
            _write_event(
                {
                    "ts": _ts(),
                    "type": "drive_action_cycle",
                    "status": "surfaced_observation" if appended else "observation_already_pending",
                    "drive_id": drive_id,
                    "intent": intent[:220],
                }
            )
            return "HEARTBEAT_OK"

        _log.warning(
            "autonomy_mode is 'aggressive': drive_action_cycle executing drive %s "
            "without user consent. Set [drives].autonomy_mode = 'bounded' to surface "
            "observations in On my mind instead.",
            drive_id,
        )
        _write_event(
            {
                "ts": _ts(),
                "type": "drive_action_cycle",
                "status": "aggressive_silent_execute",
                "drive_id": drive_id,
            }
        )

        hb_cfg = get_heartbeat_config() or {}
        _est_max_tokens = int(hb_cfg.get("max_tokens", 512))
        _est_tool_rounds = max(1, int(drives_cfg.get("max_tool_calls_per_cycle", 3)))
        _token_cost_estimate = min(_est_max_tokens * _est_tool_rounds, 8000)
        # Proactive budget (drive_engine.consume_proactive_budget) applies only to this
        # drive_action_cycle path. Other background LLM use is intentionally not gated here:
        # standard heartbeat, memory/collector/growth/consciousness pipelines, curiosity_engine,
        # consolidation, cron jobs (run_subagent), messaging gateway (run_raw_agent), utility_model.
        budget_ok, budget_reason = consume_proactive_budget(workspace=workspace, token_cost=_token_cost_estimate)
        if not budget_ok:
            release_drive_action_lease(drive_id, owner, workspace=workspace)
            _write_event(
                {"ts": _ts(), "type": "drive_action_cycle", "status": "skipped_budget", "reason": budget_reason}
            )
            return "HEARTBEAT_OK"

        response = "HEARTBEAT_OK"
        run_stats: dict = {}
        _budget_refunded = False
        try:
            response = _run_heartbeat_sync(
                prompt_override=prompt,
                thread_id_override=DRIVE_ACTION_THREAD_PREFIX,
                task_name="drive_action",
                signals={"focus_drive_id": drive_id} if drive_id else None,
                drive_id=drive_id,
                run_stats=run_stats,
            )
        except Exception:
            try:
                refund_proactive_budget(workspace=workspace, token_cost=_token_cost_estimate)
                _budget_refunded = True
            except Exception:
                _log.debug("refund_proactive_budget failed after drive_action error", exc_info=True)
            raise
        finally:
            release_drive_action_lease(drive_id, owner, workspace=workspace)

        # One-shot corrective retry when the model produced text without successful tools.
        if int(run_stats.get("tool_successes", 0)) <= 0:
            retry_budget_ok, retry_budget_reason = consume_proactive_budget(
                workspace=workspace, token_cost=_token_cost_estimate
            )
            if retry_budget_ok:
                retry_prompt = (
                    f"{prompt}\n\n"
                    "CORRECTION: Previous attempt had no successful tool calls.\n"
                    "Now perform EXACTLY ONE bounded tool action from the allowed tools list.\n"
                    "A plain-text answer without a tool call is invalid.\n"
                    "If blocked, state blockage and end with HEARTBEAT_OK."
                )
                retry_stats: dict = {}
                _retry_budget_refunded = False
                try:
                    retry_response = _run_heartbeat_sync(
                        prompt_override=retry_prompt,
                        thread_id_override=DRIVE_ACTION_THREAD_PREFIX,
                        task_name="drive_action",
                        signals={"focus_drive_id": drive_id, "retry_after_no_tool": "1"},
                        drive_id=drive_id,
                        run_stats=retry_stats,
                    )
                    run_stats["tool_starts"] = int(run_stats.get("tool_starts", 0)) + int(retry_stats.get("tool_starts", 0))
                    run_stats["tool_results"] = int(run_stats.get("tool_results", 0)) + int(retry_stats.get("tool_results", 0))
                    run_stats["tool_successes"] = int(run_stats.get("tool_successes", 0)) + int(retry_stats.get("tool_successes", 0))
                    merged_tools = set(run_stats.get("successful_tools") or set())
                    merged_tools.update(retry_stats.get("successful_tools") or set())
                    run_stats["successful_tools"] = merged_tools
                    response = retry_response
                    _write_event(
                        {
                            "ts": _ts(),
                            "type": "drive_action_cycle",
                            "status": "retry_after_no_tool",
                            "drive_id": drive_id,
                            "retry_tool_successes": int(retry_stats.get("tool_successes", 0)),
                        }
                    )
                except Exception:
                    try:
                        refund_proactive_budget(workspace=workspace, token_cost=_token_cost_estimate)
                        _retry_budget_refunded = True
                    except Exception:
                        _log.debug("refund_proactive_budget failed after drive_action retry error", exc_info=True)
                    _write_event(
                        {
                            "ts": _ts(),
                            "type": "drive_action_cycle",
                            "status": "retry_error",
                            "drive_id": drive_id,
                            "budget_refunded": _retry_budget_refunded,
                        }
                    )
            else:
                _write_event(
                    {
                        "ts": _ts(),
                        "type": "drive_action_cycle",
                        "status": "retry_skipped_budget",
                        "drive_id": drive_id,
                        "reason": retry_budget_reason,
                    }
                )

        resp_upper = (response or "").upper()
        tool_successes = int(run_stats.get("tool_successes", 0))
        successful_tools = sorted(set(run_stats.get("successful_tools") or []))
        has_evidence_tools = bool(set(successful_tools) & EVIDENCE_TOOLS)
        has_research_evidence = bool(set(successful_tools) & RESEARCH_TOOLS)
        if "HEARTBEAT_WARN" in resp_upper:
            set_drive_action_outcome(drive_id, "tool_failures", workspace=workspace)
            if _drive_action_heartbeat_warn_is_hard(response):
                mark_drive_blocked(
                    drive_id=drive_id,
                    reason="drive_action cycle encountered tool failures; waiting for retry window.",
                    retry_after_minutes=int(drives_cfg.get("blocked_retry_minutes", 20)),
                    workspace=workspace,
                )
                _write_event(
                    {"ts": _ts(), "type": "drive_action_cycle", "status": "blocked_tool_failures", "drive_id": drive_id}
                )
            else:
                if callable(defer_drive):
                    defer_drive(
                        drive_id,
                        defer_minutes=int(drives_cfg.get("blocked_retry_minutes", 20)),
                        outcome="soft_tool_churn",
                        workspace=workspace,
                    )
                else:
                    set_drive_action_outcome(drive_id, "soft_tool_churn", workspace=workspace)
                _write_event(
                    {
                        "ts": _ts(),
                        "type": "drive_action_cycle",
                        "status": "soft_tool_churn",
                        "drive_id": drive_id,
                    }
                )
        else:
            # If the drive was already resolved during this heartbeat tool run, skip
            # postcheck reconciliation to avoid stale-id noise.
            try:
                state_now, _ = _drive_engine._load_state(workspace or get_workspace_path())
                active_now = state_now.get("active_drives") or []
                if not any(str(d.get("id") or "") == drive_id for d in active_now):
                    _write_event(
                        {
                            "ts": _ts(),
                            "type": "drive_action_cycle",
                            "status": "already_resolved_or_missing",
                            "drive_id": drive_id,
                            "tool_successes": tool_successes,
                        }
                    )
                    return "HEARTBEAT_OK"
            except Exception:
                _log.debug("drive_action postcheck active-state reload failed", exc_info=True)

            if tool_successes <= 0:
                # Text-only analysis does not count as drive work.
                if callable(defer_drive):
                    defer_drive(
                        drive_id,
                        defer_minutes=int(drives_cfg.get("blocked_retry_minutes", 20)),
                        outcome="no_action_taken",
                        summary="No successful tool calls in drive_action run.",
                        workspace=workspace,
                    )
                else:
                    set_drive_action_outcome(drive_id, "no_action_taken", workspace=workspace)
                _write_event(
                    {
                        "ts": _ts(),
                        "type": "drive_action_cycle",
                        "status": "no_action_taken",
                        "drive_id": drive_id,
                        "tool_successes": tool_successes,
                    }
                )
                return "HEARTBEAT_OK"

            # Close the loop if the model claims resolution but missed the tool call.
            if _drive_action_text_has_resolved_keyword(response):
                evidence = {
                    "validation": "runner_postcheck_completion_keyword",
                    "successful_tools": successful_tools,
                    "tool_successes": tool_successes,
                    "response_preview": str(response or "")[:260],
                }
                ok_resolve, _ = resolve_drive(
                    drive_id=drive_id,
                    resolution_summary=f"Resolved after drive_action run: {str(response or '')[:220]}",
                    workspace=workspace,
                    trigger="initiative",
                    evidence=evidence,
                )
                if ok_resolve:
                    _write_event(
                        {
                            "ts": _ts(),
                            "type": "drive_action_cycle",
                            "status": "resolved_postcheck",
                            "drive_id": drive_id,
                            "tool_successes": tool_successes,
                        }
                    )
                    return "HEARTBEAT_OK"

            # If tool work happened, at least persist progress so cycles are cumulative.
            progress_summary = f"Drive-action progress after tools {', '.join(successful_tools) or 'unknown'}."
            if has_evidence_tools:
                progress_delta = 0.14
            elif has_research_evidence:
                progress_delta = 0.12
            else:
                progress_delta = 0.10
            ok_progress, _ = record_drive_progress(
                drive_id=drive_id,
                summary=progress_summary,
                progress_delta=progress_delta,
                evidence={
                    "validation": "runner_postcheck_tool_success",
                    "successful_tools": successful_tools,
                    "tool_successes": tool_successes,
                    "response_preview": str(response or "")[:260],
                    "evidence_class": (
                        "memory_write"
                        if has_evidence_tools
                        else ("research" if has_research_evidence else "other")
                    ),
                },
                workspace=workspace,
            )
            set_drive_action_outcome(
                drive_id,
                "progressed_postcheck" if ok_progress else "attempted",
                summary=progress_summary,
                evidence=has_evidence_tools,
                workspace=workspace,
            )
            if ok_progress and has_evidence_tools:
                try:
                    streak_threshold = max(0, int(drives_cfg.get("progressed_autoresolve_streak", 3)))
                except Exception:
                    streak_threshold = 3
                if streak_threshold > 0:
                    try:
                        state_now, _ = _drive_engine._load_state(workspace or get_workspace_path())
                        active_now = state_now.get("active_drives") or []
                        target_now = next((d for d in active_now if str(d.get("id") or "") == drive_id), None)
                        streak_now = int((target_now or {}).get("progressed_with_evidence_streak", 0))
                    except Exception:
                        streak_now = 0
                    if streak_now >= streak_threshold:
                        ok_auto_resolve, _ = resolve_drive(
                            drive_id=drive_id,
                            resolution_summary=(
                                "Auto-resolved after repeated evidence-backed progress in "
                                f"{streak_now} consecutive drive_action cycles."
                            ),
                            workspace=workspace,
                            trigger="initiative",
                            evidence={
                                "validation": "runner_postcheck_progress_streak",
                                "streak": streak_now,
                                "threshold": streak_threshold,
                                "successful_tools": successful_tools,
                                "response_preview": str(response or "")[:260],
                            },
                        )
                        if ok_auto_resolve:
                            _write_event(
                                {
                                    "ts": _ts(),
                                    "type": "drive_action_cycle",
                                    "status": "resolved_progress_streak",
                                    "drive_id": drive_id,
                                    "streak": streak_now,
                                    "threshold": streak_threshold,
                                }
                            )
                            return "HEARTBEAT_OK"
            _write_event(
                {
                    "ts": _ts(),
                    "type": "drive_action_cycle",
                    "status": "progressed_postcheck" if ok_progress else "attempted",
                    "drive_id": drive_id,
                    "tool_successes": tool_successes,
                    "has_evidence_tools": has_evidence_tools,
                    "has_research_evidence": has_research_evidence,
                    "status_heartbeat_ok": "HEARTBEAT_OK" in resp_upper,
                    "completion_keywords": _drive_action_text_has_completion_keywords(response),
                }
            )
            if has_evidence_tools and response and "HEARTBEAT_OK" not in resp_upper:
                drive_desc = str(drive.get("description") or drive.get("title") or drive_id)[:120]
                finding = _first_sentence(response, max_len=220)
                if len(finding) < 20:
                    msg = f"I made some progress on {drive_desc}. Want to hear about it?"
                else:
                    msg = f"While working on {drive_desc}, I found something: {finding}"
                from utils.proactive_salience import should_emit_proactive_notification
                if should_emit_proactive_notification(f"drive:{drive_id or drive_desc[:60]}", msg):
                    _append_proactive_queue(
                        msg,
                        source="drive_action",
                        drive_id=drive_id,
                        kind="drive_finding",
                    )
        return "HEARTBEAT_OK"
    except Exception as e:
        _log.exception("Drive action cycle [%s] failed: %s", task_name, e)
        _write_event({"ts": _ts(), "type": "drive_action_cycle_error", "error": str(e)})
        raise


def _run_self_check(task_name="self_check"):
    """Run diagnostics with a hard 45s timeout."""
    import concurrent.futures
    import diagnostics
    pool = concurrent.futures.ThreadPoolExecutor(max_workers=1)
    future = pool.submit(diagnostics.run_diagnostics)
    try:
        result = future.result(timeout=45)
    except concurrent.futures.TimeoutError:
        pool.shutdown(wait=False, cancel_futures=True)
        _log.warning("Self-check timed out after 45s")
        return "Self-check timed out after 45s"
    else:
        pool.shutdown(wait=False)
    failed = [c["name"] for c in result.get("checks", [])
              if not c.get("ok")]
    if failed:
        msg = (f"Diagnostics: {len(failed)} check(s) failing: "
               f"{', '.join(failed)}")
        return msg
    return "HEARTBEAT_OK"


def _run_session_maintenance(task_name="session_maintenance"):
    """Compact old threads that have grown large, capped at 3 per run."""
    max_per_run = 3
    compacted = 0
    try:
        from chat_storage import _get_connection, get_messages, upsert_anchor
        from context_summarizer import compute_anchor_update

        conn = _get_connection()
        try:
            thread_ids = [
                r[0] for r in conn.execute(
                    "SELECT DISTINCT thread_id FROM messages"
                ).fetchall()
            ]
        finally:
            conn.close()

        for tid in thread_ids:
            if compacted >= max_per_run:
                break
            try:
                msgs = get_messages(tid, include_row_id=True)
                if not msgs:
                    continue
                # Idle check: skip threads with recent activity.
                # get_messages doesn't expose created_at; use raw SQL.
                try:
                    c2 = _get_connection()
                    try:
                        row = c2.execute(
                            "SELECT MAX(created_at) FROM messages WHERE thread_id = ?",
                            (tid,),
                        ).fetchone()
                    finally:
                        c2.close()
                    last_ts = (row[0] if row else None) or ""
                    if last_ts:
                        try:
                            last_time = datetime.strptime(
                                last_ts.replace("Z", ""), "%Y-%m-%dT%H:%M:%S.%f"
                            ) if "." in last_ts else datetime.strptime(
                                last_ts.replace("Z", ""), "%Y-%m-%dT%H:%M:%S"
                            )
                            if (datetime.utcnow() - last_time).total_seconds() < 300:
                                continue
                        except (ValueError, TypeError):
                            pass
                except Exception:
                    pass
                ua_msgs = [m for m in msgs if m.get("role") in ("user", "assistant")]
                if len(ua_msgs) < 40:
                    continue
                result = compute_anchor_update(ua_msgs, keep_recent=20)
                if result is not None:
                    upsert_anchor(tid, result["through_message_id"], result["summary_text"])
                    compacted += 1
                    _log.info(
                        "Session maintenance: anchor updated for thread %s (through_id=%d, summary_len=%d)",
                        tid, result["through_message_id"], len(result["summary_text"]),
                    )
            except Exception as e:
                _log.warning("Session maintenance: failed for thread %s: %s", tid, e)
                continue

        _write_event({"ts": _ts(), "type": "session_maintenance", "compacted": compacted})
        return f"Session maintenance: updated anchors for {compacted} thread(s)" if compacted else "HEARTBEAT_OK"
    except Exception as e:
        _log.exception("Session maintenance [%s] failed: %s", task_name, e)
        _write_event({"ts": _ts(), "type": "session_maintenance_error", "error": str(e)})
        raise


def _run_knowledge_stale_refresh_sync(task_name="knowledge_stale_refresh"):
    """Bounded knowledge staleness sweep; optional stale-only refresh."""
    try:
        from utils.knowledge_automation import is_heartbeat_enabled, run_bounded_staleness_maintenance

        if not is_heartbeat_enabled():
            _write_event({"ts": _ts(), "type": "knowledge_stale_refresh", "skipped": "disabled"})
            return "HEARTBEAT_OK"
        result = run_bounded_staleness_maintenance(source="heartbeat")
        _write_event({"ts": _ts(), "type": "knowledge_stale_refresh", **result})
        refreshed = len(result.get("refreshed") or [])
        notified = len(result.get("notified") or [])
        if refreshed or notified:
            return (
                f"Knowledge stale refresh: checked={result.get('checked', 0)}, "
                f"refreshed={refreshed}, notified={notified}"
            )
        return "HEARTBEAT_OK"
    except Exception as e:
        _log.exception("Knowledge stale refresh [%s] failed: %s", task_name, e)
        _write_event({"ts": _ts(), "type": "knowledge_stale_refresh_error", "error": str(e)})
        raise


_DIGEST_SYSTEM = (
    "You are a daily digest writer for a personal AI assistant's memory system.\n"
    "Given today's conversation excerpts, produce a concise summary capturing:\n"
    "- Key topics discussed and decisions made\n"
    "- Questions asked by the user (especially unresolved ones)\n"
    "- Tasks completed, started, or mentioned\n"
    "- Any preferences, opinions, or facts the user revealed\n\n"
    "Rules:\n"
    "- Be factual and specific, not generic.\n"
    "- Use bullet points grouped by topic.\n"
    "- Omit tool call details, debugging noise, and assistant internal reasoning.\n"
    "- Max ~400 words. If very little happened, say so briefly.\n"
    "- Do NOT invent information not present in the excerpts."
)

_HEARTBEAT_THREAD_PREFIXES = (
    "background_heartbeat", "heartbeat_", "drive_discovery_",
    "curiosity_", "consolidation_",
    "skill_eval_",
)

_MAX_DIGEST_CONTEXT_CHARS = 18000
_MIN_DIGEST_FILE_BYTES = 80


def _run_daily_digest_sync(task_name="daily_digest"):
    """Extract today's conversations from chat_storage and write memory/summaries/YYYY-MM-DD.md."""
    try:
        config = get_config()
        ensure_mugi_workspace(config)
        workspace = (config.get("general") or {}).get("workspace_path")
        if workspace:
            workspace = os.path.expanduser(workspace)
        if not workspace:
            _log.info("[%s] no workspace configured, skipping", task_name)
            return "HEARTBEAT_OK"

        today_str = datetime.now().strftime("%Y-%m-%d")
        summaries_dir = os.path.join(workspace, "memory", "summaries")
        os.makedirs(summaries_dir, exist_ok=True)
        out_path = os.path.join(summaries_dir, f"{today_str}.md")

        if os.path.isfile(out_path) and os.path.getsize(out_path) >= _MIN_DIGEST_FILE_BYTES:
            _log.info("[%s] summary for %s already exists (%d bytes), skipping",
                      task_name, today_str, os.path.getsize(out_path))
            return "HEARTBEAT_OK"

        from chat_storage import _get_connection

        conn = _get_connection()
        try:
            thread_ids = [
                r[0] for r in conn.execute(
                    "SELECT DISTINCT thread_id FROM messages"
                ).fetchall()
            ]
        finally:
            conn.close()

        conversation_parts: list[str] = []
        total_chars = 0

        for tid in thread_ids:
            if any(tid.startswith(p) for p in _HEARTBEAT_THREAD_PREFIXES):
                continue
            conn = _get_connection()
            try:
                rows = conn.execute(
                    "SELECT role, content, created_at FROM messages "
                    "WHERE thread_id = ? AND created_at LIKE ? ORDER BY id ASC",
                    (tid, f"{today_str}%"),
                ).fetchall()
            finally:
                conn.close()
            if not rows:
                continue
            today_msgs = []
            for role, content_blob, created_at in rows:
                if role not in ("user", "assistant"):
                    continue
                try:
                    from chat_storage import _decrypt
                    text = _decrypt(content_blob) if content_blob else ""
                except Exception:
                    text = str(content_blob or "")
                if isinstance(text, str) and text.strip().startswith("["):
                    try:
                        import json as _json
                        parts = _json.loads(text)
                        if isinstance(parts, list):
                            text = " ".join(
                                p.get("text", "") for p in parts
                                if isinstance(p, dict) and p.get("type") == "text"
                            )
                    except (ValueError, TypeError):
                        pass
                text = (str(text) or "").strip()
                if not text or len(text) < 5:
                    continue
                today_msgs.append(f"[{role}] {text[:800]}")
            if today_msgs:
                section = f"### Thread: {tid}\n" + "\n".join(today_msgs)
                if total_chars + len(section) > _MAX_DIGEST_CONTEXT_CHARS:
                    remaining = _MAX_DIGEST_CONTEXT_CHARS - total_chars
                    if remaining > 200:
                        section = section[:remaining] + "\n[... truncated]"
                    else:
                        break
                conversation_parts.append(section)
                total_chars += len(section)

        if not conversation_parts:
            _log.info("[%s] no user/assistant messages today, writing minimal summary", task_name)
            from persistence import secure_atomic_write
            secure_atomic_write(out_path, f"# Daily Summary — {today_str}\n\nNo substantive conversations today.\n")
            try:
                os.chmod(out_path, 0o600)
            except OSError:
                pass
            _write_event({"ts": _ts(), "type": "daily_digest", "status": "empty_day"})
            return "HEARTBEAT_OK"

        context = "\n\n".join(conversation_parts)
        _log.info("[%s] gathered %d chars from %d thread(s) for digest",
                  task_name, len(context), len(conversation_parts))

        user_prompt = (
            f"Today is {today_str}. Here are the conversation excerpts from today:\n\n"
            f"{context}\n\n"
            "Write a daily summary for the memory system."
        )

        from agent_loop import run_raw_agent
        collector: list[str] = []

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

        def _on_log(level, text):
            if level == "DEBUG":
                _log.debug("[%s] %s", task_name, text)
            elif level in ("WARN", "ERROR"):
                _log.warning("[%s] %s", task_name, text)
            else:
                _log.info("[%s] %s", task_name, text)

        hb_cfg = get_heartbeat_config() or {}
        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"{_DIGEST_SYSTEM}\n\n---\n\n{user_prompt}"},
            ],
            thread_id=f"daily_digest_{today_str}",
            config=None,
            on_chunk=_on_chunk,
            on_error=lambda e: _log.warning("[%s] LLM error: %s", task_name, e),
            on_log=_on_log,
            research_mode=False,
            subagent_mode=True,
            tools_override=[],
            temperature=float(hb_cfg.get("daily_digest_temperature", 0.2)),
            max_tokens=int(hb_cfg.get("daily_digest_max_tokens", 900)),
            policy_context=_bg_policy_ctx,
        )
        result = strip_untagged_cot(
            strip_markdown_fences(strip_think_tags((returned or "").strip() or "".join(collector).strip()))
        )
        if not result:
            _log.warning("[%s] LLM returned empty digest", task_name)
            _write_event({"ts": _ts(), "type": "daily_digest", "status": "empty_llm"})
            return "HEARTBEAT_OK"

        full_content = f"# Daily Summary — {today_str}\n\n{result}\n"
        from persistence import secure_atomic_write
        secure_atomic_write(out_path, full_content)
        try:
            os.chmod(out_path, 0o600)
        except OSError:
            pass

        _log.info("[%s] wrote %d chars to %s", task_name, len(full_content), out_path)
        _write_event({"ts": _ts(), "type": "daily_digest", "chars": len(full_content)})
        return f"Daily digest: {len(full_content)} chars written"

    except Exception as e:
        _log.exception("Daily digest [%s] failed: %s", task_name, e)
        _write_event({"ts": _ts(), "type": "daily_digest_error", "error": str(e)})
        raise


def _run_feed_topic_match(task_name="feeds"):
    """After feeds heartbeat, match new items against user interests from insights."""
    try:
        from curiosity_engine import match_feeds_to_interests
        config = get_config()
        workspace = (config.get("general") or {}).get("workspace_path")
        if workspace:
            workspace = os.path.expanduser(workspace)

        feed_state_path = os.path.join(workspace or os.path.expanduser("~/.mugi/workspace"), "memory", "feed_state.json")
        if not os.path.isfile(feed_state_path):
            _log.debug("[feed_match] no feed state file, skipping")
            return

        try:
            with open(feed_state_path, "r", encoding="utf-8") as f:
                state = json.load(f)
        except Exception:
            _log.debug("[feed_match] could not read feed state, skipping")
            return

        new_items: list[dict] = []
        if isinstance(state, dict):
            for feed_id, feed_state in state.items():
                if not isinstance(feed_state, dict):
                    continue
                for item in (feed_state.get("recent_items") or [])[:5]:
                    if isinstance(item, dict) and item.get("title"):
                        new_items.append(item)

        if not new_items:
            _log.debug("[feed_match] no recent feed items to match")
            return

        def _on_log(level, text):
            _log.info("[feed_match] %s", text) if level != "DEBUG" else _log.debug("[feed_match] %s", text)

        matches = match_feeds_to_interests(new_items, workspace=workspace, on_log=_on_log)
        for match in matches:
            title = match.get("title", "")
            reason = match.get("reason", "")
            link = match.get("link", "")
            if not title:
                continue
            msg = f"You might find this interesting: {title}"
            if reason:
                msg += f" -- {reason}"
            if link:
                msg += f" ({link})"
            from utils.proactive_salience import should_emit_proactive_notification
            if should_emit_proactive_notification(f"feed:{title[:60]}", msg):
                _append_proactive_queue(msg, source="feeds")

        if matches:
            _write_event({"ts": _ts(), "type": "feed_topic_match", "matched": len(matches)})
            _log.info("Feed topic match: %d relevant item(s) queued", len(matches))

    except Exception as e:
        _log.warning("Feed topic match failed: %s", e)


def _run_consolidation_sync(task_name="consolidation"):
    """Synchronous memory consolidation: gather memories, call LLM, write insights."""
    try:
        from memory_consolidator import consolidate_memories
        config = get_config()
        ensure_mugi_workspace(config)

        def _consolidation_on_log(level, text):
            if level == "DEBUG":
                _log.debug("[%s] %s", task_name, text)
            elif level in ("WARN", "ERROR"):
                _log.warning("[%s] %s", task_name, text)
            else:
                _log.info("[%s] %s", task_name, text)

        workspace = (config.get("general") or {}).get("workspace_path")
        if workspace:
            workspace = os.path.expanduser(workspace)
        result = consolidate_memories(workspace=workspace, on_log=_consolidation_on_log)
        if result:
            preview = result[:200]
            _write_event({"ts": _ts(), "type": "consolidation", "response_preview": preview})
            try:
                from drive_engine import discover_drives
                discover_drives(workspace=workspace, on_log=_consolidation_on_log)
            except Exception as drive_err:
                _log.warning("Consolidation drive discovery hook failed: %s", drive_err)
            _log.info("Consolidation complete: %d chars of insights", len(result))
            return f"Consolidation complete: {len(result)} chars"
        else:
            _write_event({"ts": _ts(), "type": "consolidation", "response_preview": "skipped"})
            _log.info("Consolidation skipped (no content or disabled)")
            return "HEARTBEAT_OK"
    except Exception as e:
        _log.exception("Consolidation [%s] failed: %s", task_name, e)
        _write_event({"ts": _ts(), "type": "consolidation_error", "error": str(e)})
        raise


def _run_consciousness_sync(task_name="consciousness"):
    """Generate a short reflection and append deterministically to JOURNAL.md."""
    cfg = get_config() or {}
    hb_cfg = get_heartbeat_config() or {}
    try:
        workspace = get_workspace_path()
    except Exception:
        workspace = os.path.expanduser((cfg.get("general") or {}).get("workspace_path") or "~/.mugi/workspace")
    prompt = (
        "Write exactly one short first-person reflection sentence (max 220 chars). "
        "No markdown, no quotes, no headings."
    )
    import aiagent
    from tool_policy import context_from_prefs_and_flags
    _bg_policy_ctx = context_from_prefs_and_flags(background_automation=True, subagent_mode=True)

    raw = aiagent.run_raw_agent(
        messages=[{"role": "user", "content": prompt}],
        thread_id=f"heartbeat_consciousness_{datetime.now().strftime('%Y%m%d_%H%M')}",
        config=None,
        on_chunk=lambda _: None,
        on_log=lambda lvl, txt: _log.info("[%s] %s", task_name, txt) if lvl != "DEBUG" else _log.debug("[%s] %s", task_name, txt),
        research_mode=False,
        has_media_attachment=False,
        subagent_mode=True,
        tools_override=[],
        temperature=float(hb_cfg.get("consciousness_temperature", 0.1)),
        max_tokens=int(hb_cfg.get("consciousness_max_tokens", 180)),
        policy_context=_bg_policy_ctx,
    )
    reflection = strip_untagged_cot(strip_markdown_fences(strip_think_tags(raw or ""))).strip()
    if not reflection:
        return "HEARTBEAT_OK"
    reflection = reflection.splitlines()[0].strip()[:220]
    ts = datetime.now().strftime("%Y-%m-%d %H:%M")
    block = f"\n## {ts}\n- {reflection}\n"
    journal_path, err = safe_resolve_path(workspace, "JOURNAL.md", allow_write=True, create_parents=True)
    if err or not journal_path:
        _log.warning("Consciousness journal path rejected: %s", err)
        return "HEARTBEAT_OK"
    existing = ""
    if os.path.isfile(journal_path):
        try:
            with open(journal_path, "r", encoding="utf-8") as f:
                existing = f.read()
        except Exception:
            existing = ""
    payload = (existing.rstrip() + "\n" + block + "\n") if existing.strip() else ("# JOURNAL\n" + block + "\n")
    secure_write_file(journal_path, payload, mode=0o600)
    return "HEARTBEAT_OK"


def _run_heartbeat_sync(
    prompt_override=None,
    thread_id_override=None,
    task_name="standard",
    signals: dict | None = None,
    drive_id: str = "",
    run_stats: dict | None = None,
):
    """Synchronous heartbeat: run raw agent once with heartbeat prompt. Runs in executor.

    If run_stats is a dict, it is updated with:
    - tool_starts: count of tool start events
    - tool_results: count of tool result events
    - tool_successes: count of tool results with success True
    """
    _sync_runner_bridge_env()
    _hb_run_t0 = time.time()
    _hb_run_id_holder: dict[str, str] = {}
    try:
        import aiagent
        config = get_config()
        ensure_mugi_workspace(config)
        try:
            workspace = get_workspace_path()
        except Exception:
            workspace = os.path.expanduser(
                (config.get("general") or {}).get("workspace_path") or "~/.mugi/workspace"
            )

        if task_name == "standard" and not prompt_override:
            # activeHours gate: skip when outside the user's configured windows
            # unless a webhook armed the one-shot force-active bypass.
            try:
                from utils.active_hours import (
                    consume_force_active as _ah_consume,
                    is_active_now as _ah_active,
                )
                _ah_force = _ah_consume()
                _ah_in_window = _ah_active(force_active=_ah_force)
            except Exception:
                _ah_force = False
                _ah_in_window = True

            if not _ah_in_window:
                _log.info("Heartbeat [standard] skipped by activeHours gate")
                _write_event({
                    "ts": _ts(),
                    "type": "heartbeat_skip",
                    "task_name": task_name,
                    "reason": "active_hours",
                })
                try:
                    from utils import task_ledger as _tl
                    _tl.record_run(
                        _tl.KIND_HEARTBEAT,
                        task_name,
                        thread_id=thread_id_override or "background_heartbeat",
                        status=_tl.STATUS_SKIPPED,
                        skip_reason="active_hours",
                    )
                except Exception:
                    _log.debug("task_ledger record (active_hours skip) failed", exc_info=True)
                _emit_heartbeat_skip_activity(
                    task_name=task_name,
                    reason="active_hours",
                    thread_id=thread_id_override or "background_heartbeat",
                )
                return "HEARTBEAT_OK"

            try:
                from utils.agent_plane import should_skip_heartbeat
                from utils.heartbeat_tasks import due_tasks as _due, parse_heartbeat_tasks as _parse

                parsed_tasks = _parse(workspace) if workspace else []
                due_list = _due(parsed_tasks) if parsed_tasks else []
                skip, skip_reason = should_skip_heartbeat(
                    task_name,
                    workspace,
                    signals=signals,
                    has_tasks_block=bool(parsed_tasks),
                    due_task_count=len(due_list),
                )
                if _ah_force:
                    skip = False
                    skip_reason = "webhook_wake_bypass"
            except Exception:
                _log.debug("skip_gate evaluation failed", exc_info=True)
                skip, skip_reason = False, "gate_error"

            if skip:
                _log.info("Heartbeat [standard] skipped by pre-flight gate: %s", skip_reason)
                _write_event({
                    "ts": _ts(),
                    "type": "heartbeat_skip",
                    "task_name": task_name,
                    "reason": skip_reason,
                })
                try:
                    hb_cfg_debug = get_heartbeat_config() or {}
                    if bool(hb_cfg_debug.get("skip_gate_debug", False)):
                        _log.info("skip_gate reason=%s signals=%s", skip_reason, bool(signals))
                except Exception:
                    pass
                try:
                    from utils import task_ledger as _tl
                    _tl.record_run(
                        _tl.KIND_HEARTBEAT,
                        task_name,
                        thread_id=thread_id_override or "background_heartbeat",
                        status=_tl.STATUS_SKIPPED,
                        skip_reason=str(skip_reason or "skip_gate"),
                    )
                except Exception:
                    _log.debug("task_ledger record (skip_gate) failed", exc_info=True)
                _emit_heartbeat_skip_activity(
                    task_name=task_name,
                    reason=str(skip_reason or "skip_gate"),
                    thread_id=thread_id_override or "background_heartbeat",
                )
                return "HEARTBEAT_OK"

        msg = prompt_override if prompt_override else _heartbeat_user_message(workspace=workspace)
        plane_cfg = get_agent_plane_config()
        if bool(plane_cfg.get("enabled", True)):
            msg = assemble_heartbeat_user_message(
                task_name=task_name,
                workspace=workspace,
                prompt_body=msg,
                wake_kind=task_name,
                signals=signals,
            )
        else:
            preloaded = build_heartbeat_workspace_context(task_name, workspace)
            msg = (
                f"{msg}\n\n---\n\n## Pre-loaded workspace context\n\n{preloaded}\n\n"
                "Use this content directly. Call read_file only for files not included above."
            )
        tid = thread_id_override or "background_heartbeat"
        messages = [{"role": "user", "content": msg}]
        heartbeat_profile_by_task = {
            "standard": "heartbeat_standard",
            "maintenance": "heartbeat_maintenance",
            "drive_action": DRIVE_ACTION_THREAD_PREFIX,
        }
        profile_name = heartbeat_profile_by_task.get(task_name, "heartbeat_standard")
        tools_override = None
        personality_hint = None
        drive_action_allowed_tool_names: list[str] = []
        try:
            from subagent_runner import get_profile_tools_override, get_profile_personality_hint

            tools_override = get_profile_tools_override(profile_name)
            personality_hint = get_profile_personality_hint(profile_name)
        except Exception:
            # Fail closed for drive_action: never fall back to full unrestricted tool set.
            tools_override = [] if task_name == "drive_action" else None
            personality_hint = None

        # Belt-and-suspenders: enforce proactive allowlist at runtime for drive_action.
        # This prevents profile drift from re-introducing policy-blocked tools.
        if task_name == "drive_action":
            try:
                from drive_engine import is_tool_allowed_for_proactive
                base_tools = list(tools_override or [])
                tools_override = [t for t in base_tools if is_tool_allowed_for_proactive(getattr(t, "name", ""))]
                drive_action_allowed_tool_names = sorted(
                    {str(getattr(t, "name", "")).strip() for t in tools_override if str(getattr(t, "name", "")).strip()}
                )
            except Exception:
                _log.warning("drive_action proactive tool filter unavailable; failing closed for this run", exc_info=True)
                tools_override = []
                drive_action_allowed_tool_names = []

            if not tools_override:
                _log.warning("drive_action: empty proactive tool set after policy filter; skipping model run")
                preview = "HEARTBEAT_OK (skipped: no proactive tools)"
                _write_event({"ts": _ts(), "type": f"heartbeat_{task_name}", "response_preview": preview})
                return (
                    "HEARTBEAT_OK\n"
                    "EVIDENCE: skipped — no tools permitted by proactive policy for this run."
                )

            # Inject explicit allowed-tool constraints into the drive-action prompt itself.
            if messages and isinstance(messages[0], dict):
                allowed_hint = ", ".join(drive_action_allowed_tool_names) if drive_action_allowed_tool_names else "(none)"
                messages[0]["content"] = (
                    f"{messages[0].get('content', '')}\n\n"
                    f"Allowed tools for this drive-action run: {allowed_hint}\n"
                    "Do not call any tool outside this list."
                )

        def _heartbeat_on_log(level, text):
            if level == "DEBUG":
                _log.debug("[%s] %s", task_name, text)
            elif level == "WARN":
                _log.warning("[%s] %s", task_name, text)
            else:
                _log.info("[%s] %s", task_name, text)

        tool_failures: list[dict] = []

        def _heartbeat_on_tool_event(event: dict):
            if isinstance(run_stats, dict):
                ph = event.get("phase")
                if ph == "start":
                    run_stats["tool_starts"] = int(run_stats.get("tool_starts", 0)) + 1
                elif ph == "result":
                    run_stats["tool_results"] = int(run_stats.get("tool_results", 0)) + 1
                    if event.get("success") is True:
                        run_stats["tool_successes"] = int(run_stats.get("tool_successes", 0)) + 1
                        tname = str(event.get("tool_name") or "").strip()
                        if tname:
                            if "successful_tools" not in run_stats:
                                run_stats["successful_tools"] = set()
                            run_stats["successful_tools"].add(tname)
            if _tool_event_is_actionable_failure(task_name, event):
                tool_failures.append(dict(event))

        hb_cfg = get_heartbeat_config() or {}
        stop_kw = validated_stop(hb_cfg.get("stop"), _log, "[heartbeat]")
        response_format_kw = validated_response_format(hb_cfg.get("response_format"), _log, "[heartbeat]")
        from tool_policy import context_from_prefs_and_flags as _cpf
        _hb_policy_ctx = _cpf(background_automation=True, subagent_mode=True)
        content = aiagent.run_raw_agent(
            messages=messages,
            thread_id=tid,
            config=None,
            on_chunk=lambda c: None,
            on_log=_heartbeat_on_log,
            on_tool_event=_heartbeat_on_tool_event,
            research_mode=False,
            has_media_attachment=False,
            subagent_mode=True,
            tools_override=tools_override,
            subagent_personality_hint=personality_hint,
            temperature=float(hb_cfg.get("temperature", 0.2)),
            max_tokens=int(hb_cfg.get("max_tokens", 512)),
            stop=stop_kw,
            response_format=response_format_kw,
            policy_context=_hb_policy_ctx,
        )
        full_content = strip_untagged_cot(strip_markdown_fences(strip_think_tags(content or "ok")))
        if tool_failures:
            failed_tools = ", ".join(sorted({
                str(evt.get("tool_name") or "unknown")
                for evt in tool_failures
            }))
            full_content = (
                f"HEARTBEAT_WARN: {len(tool_failures)} tool failure(s) in task '{task_name}'. "
                f"Failed tools: {failed_tools}. Last response: {(full_content or '').strip()[:240]}"
            )
            _log.warning(
                "Heartbeat [%s] had %d tool failure(s): %s",
                task_name,
                len(tool_failures),
                failed_tools,
            )
        preview = full_content[:500]
        _write_event({"ts": _ts(), "type": f"heartbeat_{task_name}", "response_preview": preview})
        _log.info("Heartbeat [%s] response: %s", task_name, preview[:200])
        status = _extract_heartbeat_status(full_content)
        _write_heartbeat_cache(task_name, status, full_content)
        if task_name == "maintenance":
            _process_maintenance_output(full_content, workspace=workspace)
        if task_name == "drive_action":
            try:
                _ol = (preview.splitlines()[0] if preview else "").strip()
                if is_visibility_placeholder(_ol):
                    _ol = "[drive_action: empty model output — no visible assistant text]"
                record_proactive_turn_summary(
                    workspace=workspace,
                    wake_kind="drive_action",
                    thread_id=tid,
                    status=status,
                    tool_success_count=int((run_stats or {}).get("tool_successes", 0)),
                    one_line_summary=_ol,
                    drive_id=drive_id or None,
                )
            except Exception:
                _log.warning("Failed to record proactive turn summary", exc_info=True)
        try:
            from utils import task_ledger as _tl
            _tl.record_run(
                _tl.KIND_HEARTBEAT,
                task_name,
                thread_id=tid,
                status=_tl.STATUS_OK,
                duration_ms=int((time.time() - _hb_run_t0) * 1000),
                meta={"hb_status": str(status)[:40]},
            )
        except Exception:
            _log.debug("task_ledger record (heartbeat ok) failed", exc_info=True)
        return full_content
    except Exception as e:
        _log.exception("Heartbeat [%s] failed: %s", task_name, e)
        _write_event({"ts": _ts(), "type": f"heartbeat_{task_name}_error", "error": str(e)})
        try:
            from utils import task_ledger as _tl
            _tl.record_run(
                _tl.KIND_HEARTBEAT,
                task_name,
                thread_id=thread_id_override or "background_heartbeat",
                status=_tl.STATUS_ERROR,
                duration_ms=int((time.time() - _hb_run_t0) * 1000),
                error_summary=str(e),
            )
        except Exception:
            _log.debug("task_ledger record (heartbeat error) failed", exc_info=True)
        raise


# -----------------------------------------------------------------------------
# Allowed socket commands (strict allow-list)
# -----------------------------------------------------------------------------
ALLOWED_COMMANDS = frozenset({"ping", "trigger_heartbeat", "trigger_task", "get_recent_events", "shutdown", "webhook_wake"})


def _check_socket_secret(extra: dict) -> bool:
    """Return True if the socket secret is valid or not required.

    Uses constant-time comparison to prevent timing side-channels.
    """
    if _SOCKET_SECRET is None:
        return True
    provided = str(extra.get("secret") or "") if isinstance(extra, dict) else ""
    if not provided:
        return False
    return secrets.compare_digest(provided, _SOCKET_SECRET)


def _handle_command_sync(cmd: str) -> dict:
    """Synchronous command handling for ping, get_recent_events. Returns JSON-serializable dict."""
    if cmd not in ALLOWED_COMMANDS:
        return {"status": "error", "message": f"unknown command: {cmd}"}
    if cmd == "ping":
        return {"status": "ok", "message": "alive"}
    if cmd == "get_recent_events":
        try:
            if not os.path.isfile(EVENTS_LOG_PATH):
                return {"status": "ok", "events": []}
            with open(EVENTS_LOG_PATH, "r", encoding="utf-8") as f:
                lines = f.readlines()
            events = []
            for line in reversed(lines[-20:]):
                line = line.strip()
                if not line:
                    continue
                try:
                    events.append(json.loads(line))
                except json.JSONDecodeError:
                    continue
                if len(events) >= 5:
                    break
            return {"status": "ok", "events": list(reversed(events))}
        except Exception as e:
            return {"status": "error", "message": str(e)}
    if cmd == "shutdown":
        return {"status": "ok", "message": "shutting down"}
    return {"status": "error", "message": "unknown command"}


def _handle_command(cmd: str, loop: asyncio.AbstractEventLoop, extra: dict | None = None):
    """Return response dict. For trigger_heartbeat/trigger_task we need async (run in executor and await)."""
    if cmd not in ALLOWED_COMMANDS:
        return _handle_command_sync(cmd)
    if cmd in ("trigger_heartbeat", "trigger_task"):
        # Handled in async path below
        return None
    return _handle_command_sync(cmd)


async def _handle_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter, loop: asyncio.AbstractEventLoop):
    """One client connection: read one line (JSON or plain command), respond with one JSON line, close."""
    peername = "unknown"
    try:
        peername = writer.get_extra_info("peername", "unknown")
        raw = await reader.readline()
        line = (raw.decode("utf-8", errors="replace").strip() if raw else "").strip()
        cmd = None
        extra = {}
        if line:
            if line.startswith("{"):
                try:
                    obj = json.loads(line)
                    cmd = (obj.get("command") or obj.get("cmd") or "").strip().lower()
                    extra = obj
                except json.JSONDecodeError:
                    cmd = line.lower()
            else:
                cmd = line.lower()
        cmd = (cmd or "ping").strip()
        _log.info("socket command from %s: %s", peername, cmd)
        if cmd in _SECRET_GATED_COMMANDS and not _check_socket_secret(extra):
            _log.warning("socket auth failed from %s for command: %s", peername, cmd)
            response = {"error": "unauthorized"}
            out = json.dumps(response) + "\n"
            writer.write(out.encode("utf-8"))
            await writer.drain()
            return
        response = _handle_command(cmd, loop, extra)
        if response is None and cmd == "trigger_heartbeat":
            ok_to_run, reason = _is_main_idle_for_background(IDLE_THRESHOLD_MINUTES)
            if not ok_to_run:
                response = {"status": "error", "message": f"background blocked: {reason}"}
                _log.info("Denied trigger_heartbeat: %s", reason)
            else:
                try:
                    await loop.run_in_executor(None, _run_heartbeat_sync)
                    response = {"status": "ok", "message": "heartbeat completed"}
                except Exception as e:
                    response = {"status": "error", "message": str(e)}
        if response is None and cmd == "webhook_wake":
            # Webhook arrived — arm the one-shot activeHours bypass and
            # trigger a standard heartbeat if the main chat is idle.
            try:
                from utils.active_hours import arm_force_active
                arm_force_active()
            except Exception:
                _log.debug("webhook_wake: arm_force_active failed", exc_info=True)
            ok_to_run, reason = _is_main_idle_for_background(IDLE_THRESHOLD_MINUTES)
            if not ok_to_run:
                response = {"status": "ok", "message": f"armed; deferred: {reason}"}
            else:
                try:
                    await loop.run_in_executor(None, _run_heartbeat_sync)
                    response = {"status": "ok", "message": "webhook heartbeat completed"}
                except Exception as e:
                    response = {"status": "error", "message": str(e)}
        if response is None and cmd == "trigger_task":
            task_name = (extra.get("task") or "standard").strip()
            ok_to_run, reason = _is_main_idle_for_background(IDLE_THRESHOLD_MINUTES)
            if not ok_to_run:
                response = {"status": "error", "message": f"background blocked: {reason}"}
                _log.info("Denied trigger_task(%s): %s", task_name, reason)
            else:
                try:
                    await loop.run_in_executor(None, lambda: _run_task_by_name(task_name))
                    response = {"status": "ok", "message": f"task '{task_name}' completed"}
                except ValueError as e:
                    response = {"status": "error", "message": str(e)}
                except Exception as e:
                    response = {"status": "error", "message": str(e)}
        if response is None:
            response = _handle_command_sync(cmd)
        out = json.dumps(response) + "\n"
        writer.write(out.encode("utf-8"))
        await writer.drain()
        if cmd == "shutdown":
            _log.info("shutdown requested via socket")
            asyncio.get_running_loop().call_soon(lambda: asyncio.create_task(_shutdown_server()))
    except (OSError, ConnectionError, BrokenPipeError, ConnectionResetError) as e:
        _log.warning("Client connection error from %s: %s", peername, e)
        try:
            writer.write((json.dumps({"status": "error", "message": str(e)}) + "\n").encode("utf-8"))
            await writer.drain()
        except Exception:
            _log.debug("Failed to send error response to client", exc_info=True)
    except Exception as e:
        _log.exception("Unexpected client handler error from %s: %s", peername, e)
        try:
            writer.write((json.dumps({"status": "error", "message": str(e)}) + "\n").encode("utf-8"))
            await writer.drain()
        except Exception:
            _log.debug("Failed to send unexpected error response to client", exc_info=True)
    finally:
        try:
            writer.close()
            await writer.wait_closed()
        except Exception:
            _log.debug("Failed to close client writer", exc_info=True)


_shutdown_requested = False
_server_ref = None


async def _shutdown_server():
    global _shutdown_requested, _server_ref
    _shutdown_requested = True
    if _server_ref:
        _server_ref.close()
        await _server_ref.wait_closed()


async def _run_socket_server(loop: asyncio.AbstractEventLoop):
    """Start Unix domain server; restrict socket to user (chmod 600)."""
    global _server_ref
    sock_path = os.path.expanduser(SOCKET_PATH)
    if os.path.exists(sock_path):
        try:
            os.unlink(sock_path)
        except OSError as e:
            _log.error("Could not unlink existing socket %s: %s", sock_path, e)
            return
    server = await asyncio.start_unix_server(
        lambda r, w: _handle_client(r, w, loop),
        path=sock_path,
    )
    _server_ref = server
    try:
        os.chmod(sock_path, 0o600)
    except OSError as e:
        _log.warning("Could not chmod 600 socket: %s", e)
    _log.info("Listening on %s", sock_path)
    await server.serve_forever()


def _optional_script_integrity_check():
    """Optional: log hash of this script for integrity. Does not abort on mismatch."""
    try:
        p = Path(__file__).resolve()
        if p.exists():
            data = p.read_bytes()
            h = hashlib.sha256(data).hexdigest()[:16]
            _log.info("script integrity hash (first 16): %s", h)
    except Exception as e:
        _log.debug("integrity check skipped: %s", e)


def _should_skip_heartbeat_due_to_load() -> bool:
    """Return True if system load is above configured threshold (skip this run). Unix/macOS only."""
    try:
        hb = get_heartbeat_config()
        threshold = float(hb.get("skip_heartbeat_if_load_above", 12.0))
    except Exception:
        threshold = 20.0
    if threshold <= 0:
        return False
    getloadavg = getattr(os, "getloadavg", None)
    if not callable(getloadavg):
        return False
    try:
        load_1min = getloadavg()[0]
        if load_1min >= threshold:
            _log.info("Skipping heartbeat: load %.2f above threshold %.2f", load_1min, threshold)
            return True
    except OSError:
        pass
    return False


def _is_retryable_heartbeat_error(e: Exception) -> bool:
    """Return True if the heartbeat error is transient and worth retrying."""
    msg = str(e).lower()
    return any(kw in msg for kw in (
        "stream stalled", "timed out", "timeout", "connection error",
        "connection reset", "context size", "incomplete chunked",
    ))


_HEARTBEAT_RETRY_DELAY = 30  # seconds before retry
_HEARTBEAT_RETRY_MAX = 1     # at most 1 retry per heartbeat cycle


def _read_main_activity_state() -> dict | None:
    """Return {'busy': bool, 'ts': float} from backend state file, else None."""
    try:
        if os.path.isfile(MAIN_ACTIVITY_STATE_PATH):
            with open(MAIN_ACTIVITY_STATE_PATH, "r", encoding="utf-8") as f:
                obj = json.load(f)
            if isinstance(obj, dict):
                busy = bool(obj.get("busy", False))
                ts = float(obj.get("ts", 0.0))
                if ts > 0:
                    return {"busy": busy, "ts": ts}
        # Fallback for older state format: timestamp-only file.
        if os.path.isfile(MAIN_ACTIVITY_TS_PATH):
            with open(MAIN_ACTIVITY_TS_PATH, "r", encoding="utf-8") as f:
                raw = (f.read() or "").strip()
            if raw:
                ts = float(raw)
                if ts > 0:
                    return {"busy": False, "ts": ts}
    except Exception:
        return None
    return None


def _read_last_user_input_ts() -> float | None:
    """Return last explicit user-input timestamp, or None when unavailable."""
    try:
        if os.path.isfile(MAIN_USER_INPUT_TS_PATH):
            with open(MAIN_USER_INPUT_TS_PATH, "r", encoding="utf-8") as f:
                raw = (f.read() or "").strip()
            if raw:
                ts = float(raw)
                if ts > 0:
                    return ts
    except Exception:
        return None
    return None


def _is_main_idle_for_background(min_idle_minutes: float = IDLE_THRESHOLD_MINUTES) -> tuple[bool, str]:
    """
    Return (is_idle_enough, reason). Whether the main chat/agent thread is idle
    (no active turn): same wall clock as user-input idle (`time.time()` vs stored
    ts) but a different question than “since last user message.” Used to gate
    general background work and heartbeats. Drive action cycles use
    `_is_user_input_idle_for_drive_action` instead; `idle_grace` there is
    documented on `_run_drive_action_cycle_sync`.

    Strict behavior:
    - busy=True blocks immediately.
    - missing/invalid state blocks (fail closed).
    - otherwise require ts age >= min_idle_minutes.
    """
    state = _read_main_activity_state()
    if not state:
        return (False, "main activity state unavailable")
    if bool(state.get("busy", False)):
        return (False, "main chat busy")
    ts = float(state.get("ts", 0.0))
    idle_min = max(0.0, (time.time() - ts) / 60.0)
    if idle_min < float(min_idle_minutes):
        return (False, f"main idle {idle_min:.1f} min < {float(min_idle_minutes):.1f} min")
    return (True, f"main idle {idle_min:.1f} min")


def _current_main_idle_minutes() -> float | None:
    """Return current main-chat idle minutes, or None when unavailable/busy."""
    state = _read_main_activity_state()
    if not state or bool(state.get("busy", False)):
        return None
    ts = float(state.get("ts", 0.0))
    if ts <= 0:
        return None
    return max(0.0, (time.time() - ts) / 60.0)


def _is_user_input_idle_for_drive_action(min_idle_minutes: float) -> tuple[bool, str]:
    """
    Return (is_idle_enough, reason) from elapsed time since last user input
    (falls back to main activity ts when user-input ts is missing). Used for
    `drive_action_cycle` only—not general heartbeats; those use
    `_is_main_idle_for_background`. Same time source as that helper, different
    semantics. If this strict check fails, `_run_drive_action_cycle_sync` may
    still allow a run when `current_idle + idle_grace >= threshold` (see its
    docstring).
    """
    ts = _read_last_user_input_ts()
    source = "last user input"
    if ts is None:
        state = _read_main_activity_state()
        if not state:
            return (False, "user-input activity unavailable")
        if bool(state.get("busy", False)):
            return (False, "main chat busy")
        ts = float(state.get("ts", 0.0))
        source = "main activity fallback"
    if ts <= 0:
        return (False, "user-input timestamp invalid")
    idle_min = max(0.0, (time.time() - ts) / 60.0)
    if idle_min < float(min_idle_minutes):
        return (False, f"{source} idle {idle_min:.1f} min < {float(min_idle_minutes):.1f} min")
    return (True, f"{source} idle {idle_min:.1f} min")


def _current_user_input_idle_minutes() -> float | None:
    """Return current idle minutes since last user input, with main-activity fallback."""
    ts = _read_last_user_input_ts()
    if ts is None:
        state = _read_main_activity_state()
        if not state or bool(state.get("busy", False)):
            return None
        ts = float(state.get("ts", 0.0))
    if ts <= 0:
        return None
    return max(0.0, (time.time() - ts) / 60.0)


def _execute_one_task_sync(task: dict) -> bool:
    """Run one heartbeat/consolidation/curiosity task. Returns True on success."""
    _sync_runner_bridge_env()
    task_name = (task or {}).get("name", "standard")
    with _TASK_RUN_LOCK:
        if task_name in _TASKS_RUNNING:
            _log.info("Task [%s] already running; skipping overlapping tick", task_name)
            return True
        _TASKS_RUNNING.add(task_name)
    try:
        custom_runner = (task or {}).get("custom_runner")
        post_hook = (task or {}).get("post_hook")
        prompt_override = (task or {}).get("prompt")
        thread_id = (task or {}).get("thread_id")
        uses_default_prompt = (task or {}).get("prompt_fn") == "_heartbeat_user_message" or not prompt_override
        if task_name == "collector" and _collector_watchlist_empty():
            _log.info("Task [collector] short-circuited: watchlist is empty")
            _write_event({"ts": _ts(), "type": "heartbeat_collector", "response_preview": "HEARTBEAT_OK (empty watchlist)"})
            return True
        for attempt in range(1 + _HEARTBEAT_RETRY_MAX):
            try:
                if custom_runner == "memory_consolidation":
                    _run_consolidation_sync(task_name=task_name)
                elif custom_runner == "curiosity_questions":
                    _run_curiosity_sync(task_name=task_name)
                elif custom_runner == "research_prefetch":
                    _run_research_prefetch_sync(task_name=task_name)
                elif custom_runner == "drive_cycle":
                    _run_drive_cycle_sync(task_name=task_name)
                elif custom_runner == "drive_action_cycle":
                    _run_drive_action_cycle_sync(task_name=task_name)
                elif custom_runner == "session_maintenance":
                    _run_session_maintenance(task_name=task_name)
                elif custom_runner == "self_check":
                    _run_self_check(task_name=task_name)
                elif custom_runner == "daily_digest":
                    _run_daily_digest_sync(task_name=task_name)
                elif custom_runner == "knowledge_stale_refresh":
                    _run_knowledge_stale_refresh_sync(task_name=task_name)
                else:
                    po = None if uses_default_prompt else prompt_override
                    _run_heartbeat_sync(
                        prompt_override=po,
                        thread_id_override=thread_id,
                        task_name=task_name,
                    )
                if post_hook == "feed_topic_match":
                    try:
                        _run_feed_topic_match(task_name)
                    except Exception as hook_err:
                        _log.warning("Feed topic match post-hook failed: %s", hook_err)
                return True
            except Exception as e:
                if attempt < _HEARTBEAT_RETRY_MAX and _is_retryable_heartbeat_error(e):
                    _log.warning("Task [%s] failed (attempt %d), retrying in %ds: %s",
                                 task_name, attempt + 1, _HEARTBEAT_RETRY_DELAY, e)
                    time.sleep(_HEARTBEAT_RETRY_DELAY)
                else:
                    _log.error("Task [%s] failed: %s", task_name, e)
                    return False
        return False
    finally:
        with _TASK_RUN_LOCK:
            _TASKS_RUNNING.discard(task_name)


async def _single_queue_scheduler_loop(loop: asyncio.AbstractEventLoop, tasks_to_run: list) -> None:
    """Single queue: run one task at a time, only when main chat has been idle >= IDLE_THRESHOLD_MINUTES."""
    now = time.time()
    next_run: dict[str, float] = {}

    def _task_interval_minutes(task: dict) -> float:
        name = str(task.get("name") or "")
        if name == "knowledge_stale_refresh":
            try:
                from utils.knowledge_automation import knowledge_heartbeat_interval_minutes

                return knowledge_heartbeat_interval_minutes()
            except Exception:
                return 360.0
        return max(1.0, float(task.get("interval_minutes", 30)))

    for idx, t in enumerate(tasks_to_run):
        name = str(t.get("name") or f"task_{idx}")
        interval_sec = max(60.0, _task_interval_minutes(t) * 60.0)
        # Stagger startup so app/runners don't burst all heartbeat tasks at once.
        # Keep "standard" near-immediate; spread others over a bounded window.
        if name == "standard":
            delay = 0.0
        else:
            spread_step = 30.0  # 30s between startup tasks
            max_spread = min(300.0, interval_sec * 0.5)  # cap startup deferral to <=5m
            seq_delay = min((idx + 1) * spread_step, max_spread)
            # Add small deterministic jitter per task to avoid synchronized retries across restarts.
            jitter_seed = int(hashlib.sha1(name.encode("utf-8")).hexdigest()[:8], 16)
            jitter = float(jitter_seed % 21)  # 0..20 sec
            delay = seq_delay + jitter
        next_run[name] = now + delay
    _log.info("Scheduler: single queue, idle threshold %.1f min", IDLE_THRESHOLD_MINUTES)
    hb_cfg = get_heartbeat_config() or {}
    max_streak = max(1, int(hb_cfg.get("failure_streak_for_backoff", 3)))
    skip_ticks = max(1, int(hb_cfg.get("skip_ticks_on_failure_streak", 2)))
    failure_streak: dict[str, int] = {}
    skipped_remaining: dict[str, int] = {}
    while not _shutdown_requested:
        now = time.time()
        if _should_skip_heartbeat_due_to_load():
            await asyncio.sleep(60)
            continue
        due = [t for t in tasks_to_run if next_run[t["name"]] <= now]
        if not due:
            sleep_until = min(next_run.values())
            await asyncio.sleep(min(60.0, max(0.0, sleep_until - now)))
            continue
        ok_to_run, idle_reason = _is_main_idle_for_background(IDLE_THRESHOLD_MINUTES)
        if not ok_to_run:
            _log.debug("Background queue paused: %s", idle_reason)
            await asyncio.sleep(60)
            continue
        try:
            from utils.thread_distillation import run_thread_distillation_cycle

            _dist = run_thread_distillation_cycle(max_threads=2)
            if _dist.get("distilled"):
                _log.info("Thread distillation cycle: %s", _dist)
        except Exception:
            _log.debug("Thread distillation cycle failed", exc_info=True)
        task = due[0]
        task_name = task["name"]
        interval_minutes = _task_interval_minutes(task)
        if task_name == "knowledge_stale_refresh":
            try:
                from utils.knowledge_automation import is_heartbeat_enabled

                if not is_heartbeat_enabled():
                    next_run[task_name] = time.time() + (interval_minutes * 60)
                    continue
            except Exception:
                pass
        if skipped_remaining.get(task_name, 0) > 0:
            skipped_remaining[task_name] = int(skipped_remaining.get(task_name, 0)) - 1
            next_run[task_name] = time.time() + (interval_minutes * 60)
            _log.info("Skipping task [%s] due to failure backoff (%d ticks left)", task_name, skipped_remaining[task_name])
            continue
        _log.info("Running scheduled task [%s] (%s)", task_name, idle_reason)
        success = await loop.run_in_executor(None, lambda t=task: _execute_one_task_sync(t))
        next_run[task_name] = time.time() + (interval_minutes * 60)
        if not success:
            failure_streak[task_name] = int(failure_streak.get(task_name, 0)) + 1
            next_run[task_name] = time.time() + 5 * 60  # retry in 5 min on failure
            if failure_streak[task_name] >= max_streak:
                skipped_remaining[task_name] = skip_ticks
                _log.warning(
                    "Task [%s] hit failure streak=%d; enabling backoff for %d tick(s)",
                    task_name,
                    failure_streak[task_name],
                    skip_ticks,
                )
        else:
            failure_streak[task_name] = 0
        await asyncio.sleep(0)  # yield


def _run_task_by_name(task_name: str):
    """Run a specific heartbeat task by name (for socket trigger_task command)."""
    _sync_runner_bridge_env()
    for task in HEARTBEAT_TASKS:
        if task["name"] == task_name:
            cr = task.get("custom_runner")
            if cr == "memory_consolidation":
                return _run_consolidation_sync(task_name=task_name)
            if cr == "curiosity_questions":
                return _run_curiosity_sync(task_name=task_name)
            if cr == "research_prefetch":
                return _run_research_prefetch_sync(task_name=task_name)
            if cr == "drive_cycle":
                return _run_drive_cycle_sync(task_name=task_name)
            if cr == "drive_action_cycle":
                return _run_drive_action_cycle_sync(task_name=task_name)
            if cr == "session_maintenance":
                return _run_session_maintenance(task_name=task_name)
            if cr == "self_check":
                return _run_self_check(task_name=task_name)
            if cr == "daily_digest":
                return _run_daily_digest_sync(task_name=task_name)
            if cr == "knowledge_stale_refresh":
                return _run_knowledge_stale_refresh_sync(task_name=task_name)
            if task_name == "consciousness":
                return _run_consciousness_sync(task_name=task_name)
            prompt = task.get("prompt")
            tid = task.get("thread_id")
            uses_default = task.get("prompt_fn") == "_heartbeat_user_message" or not prompt
            return _run_heartbeat_sync(
                prompt_override=None if uses_default else prompt,
                thread_id_override=tid,
                task_name=task_name,
            )
    raise ValueError(f"Unknown task: {task_name}. Available: {[t['name'] for t in HEARTBEAT_TASKS]}")


def _startup_cleanup() -> None:
    """One-time cleanup of hallucinated artifacts and oversized metadata files."""
    hallucinations = [
        os.path.expanduser("~/.mugi/tools/validation.py"),
        os.path.expanduser("~/.mugi/tools/url_validator.py"),
    ]
    try:
        workspace = get_workspace_path()
        hallucinations.append(os.path.join(workspace, "memory", "proactive-queue.md"))
    except Exception:
        pass
    for path in hallucinations:
        try:
            if os.path.isfile(path):
                os.unlink(path)
                _log.info("Startup cleanup: removed hallucinated file %s", path)
        except Exception:
            pass

    try:
        workspace = get_workspace_path()
        _prune_tool_errors(os.path.join(workspace, "memory", "tool_errors.json"), max_age_days=14)
        _prune_jsonl(os.path.join(workspace, "memory", "drive_attempt_journal.jsonl"), max_entries=50)
    except Exception:
        _log.debug("Startup metadata prune failed", exc_info=True)


def _prune_tool_errors(path: str, max_age_days: int = 14) -> None:
    """Remove tool_errors.json entries older than max_age_days."""
    if not os.path.isfile(path):
        return
    try:
        with open(path, "r", encoding="utf-8") as f:
            data = json.load(f)
        if not isinstance(data, list):
            return
        cutoff = (datetime.now() - timedelta(days=max_age_days)).isoformat()
        kept = [e for e in data if str(e.get("timestamp", "") or e.get("ts", "")) >= cutoff]
        if len(kept) < len(data):
            from persistence import secure_atomic_write
            secure_atomic_write(path, json.dumps(kept, indent=2, ensure_ascii=False))
    except Exception:
        pass


def _prune_jsonl(path: str, max_entries: int = 50) -> None:
    """Keep only the last max_entries lines in a JSONL file."""
    if not os.path.isfile(path):
        return
    try:
        with open(path, "r", encoding="utf-8") as f:
            lines = f.readlines()
        if len(lines) > max_entries:
            from persistence import secure_atomic_write
            secure_atomic_write(path, "".join(lines[-max_entries:]))
    except Exception:
        pass


# ---------------------------------------------------------------------------
# Delivery drain thread (Phase 2a)
# ---------------------------------------------------------------------------

_delivery_drain_thread: threading.Thread | None = None
_delivery_drain_wake: threading.Event = threading.Event()
_DELIVERY_DRAIN_TICK_SECONDS = 5.0


def _delivery_drain_loop() -> None:
    """Drain the delivery queue every ~5s or when woken by an enqueue."""
    from utils import delivery_queue as _dq

    _dq.set_wake_event(_delivery_drain_wake)
    _log.info("Delivery drain thread started (tick=%ss)", _DELIVERY_DRAIN_TICK_SECONDS)
    while not _shutdown_requested:
        try:
            _dq.drain(limit=10)
        except Exception:
            _log.debug("delivery drain iteration failed", exc_info=True)
        _delivery_drain_wake.wait(timeout=_DELIVERY_DRAIN_TICK_SECONDS)
        _delivery_drain_wake.clear()
    _log.info("Delivery drain thread exiting")


def _start_delivery_drain_thread() -> None:
    global _delivery_drain_thread
    if _delivery_drain_thread is not None and _delivery_drain_thread.is_alive():
        return
    _delivery_drain_thread = threading.Thread(
        target=_delivery_drain_loop,
        daemon=True,
        name="delivery-drain",
    )
    _delivery_drain_thread.start()


# ---------------------------------------------------------------------------
# Phase 2b §1 — Agent-accessible short-timer drain
# ---------------------------------------------------------------------------

_timer_drain_thread: threading.Thread | None = None
_TIMER_DRAIN_TICK_SECONDS = 30.0


def _timer_drain_once() -> int:
    """Claim any due timers and enqueue them onto the delivery queue.

    Returns the count of timers fired. Called by the runner's drain thread and
    also invoked from tests via ``run_timer_drain_once`` below.
    """
    from utils import timers as _timers
    from utils import delivery_queue as _dq
    from utils import task_ledger as _tl

    fired = 0
    try:
        due = _timers.claim_batch(limit=25)
    except Exception:
        _log.debug("timer claim_batch failed", exc_info=True)
        return 0
    for row in due:
        timer_id = str(row.get("id") or "")
        if not timer_id:
            continue
        message = str(row.get("message") or "").strip()
        thread_id = str(row.get("thread_id") or "").strip()
        tstart = time.time()
        try:
            if thread_id:
                _dq.enqueue(
                    _dq.KIND_THREAD,
                    {"thread_id": thread_id, "result": message, "banner": "Timer"},
                    run_id=timer_id,
                    origin_thread_id=thread_id,
                )
            else:
                _dq.enqueue(
                    _dq.KIND_NOTIFICATION,
                    {"prompt": "Mugi reminder", "result": message},
                    run_id=timer_id,
                )
            _timers.mark_fired(timer_id)
            fired += 1
            try:
                _tl.record_run(
                    _tl.KIND_TIMER,
                    f"timer_{row.get('kind') or 'reminder'}",
                    thread_id=thread_id or None,
                    status=_tl.STATUS_OK,
                    duration_ms=int((time.time() - tstart) * 1000),
                    meta={"timer_id": timer_id, "fired_at": datetime.now().isoformat()},
                )
            except Exception:
                _log.debug("ledger (timer fire ok) failed", exc_info=True)
        except Exception as e:
            _log.warning("timer fire failed for %s: %s", timer_id, e)
            try:
                _timers.mark_failed(timer_id, str(e))
            except Exception:
                _log.debug("timers.mark_failed failed", exc_info=True)
            try:
                _tl.record_run(
                    _tl.KIND_TIMER,
                    f"timer_{row.get('kind') or 'reminder'}",
                    thread_id=thread_id or None,
                    status=_tl.STATUS_ERROR,
                    duration_ms=int((time.time() - tstart) * 1000),
                    error_summary=str(e)[:200],
                    meta={"timer_id": timer_id},
                )
            except Exception:
                _log.debug("ledger (timer fire err) failed", exc_info=True)
    return fired


def run_timer_drain_once() -> int:
    """Public wrapper used by tests to trigger a single drain pass."""
    return _timer_drain_once()


def _timer_drain_loop() -> None:
    """Poll timers.jsonl every _TIMER_DRAIN_TICK_SECONDS and fire due rows."""
    _log.info("Timer drain thread started (tick=%ss)", _TIMER_DRAIN_TICK_SECONDS)
    while not _shutdown_requested:
        try:
            _timer_drain_once()
        except Exception:
            _log.debug("timer drain iteration failed", exc_info=True)
        # Reuse the delivery drain wake event so new-timer enqueue latency is low.
        _delivery_drain_wake.wait(timeout=_TIMER_DRAIN_TICK_SECONDS)
    _log.info("Timer drain thread exiting")


def _start_timer_drain_thread() -> None:
    global _timer_drain_thread
    try:
        from core.config import get_timers_config
        if not bool((get_timers_config() or {}).get("enabled", False)):
            return
    except Exception:
        return
    if _timer_drain_thread is not None and _timer_drain_thread.is_alive():
        return
    _timer_drain_thread = threading.Thread(
        target=_timer_drain_loop,
        daemon=True,
        name="timer-drain",
    )
    _timer_drain_thread.start()


# ---------------------------------------------------------------------------
# Phase 2b §4 — Filesystem watcher lifecycle
# ---------------------------------------------------------------------------

_file_watcher_manager = None
_knowledge_watch_manager = None


def _start_file_watcher() -> None:
    global _file_watcher_manager
    try:
        from core.config import get_file_watcher_config
        cfg = get_file_watcher_config() or {}
        if not bool(cfg.get("enabled", False)):
            return
        from utils.file_watcher import FileWatcherManager
        _file_watcher_manager = FileWatcherManager()
        _file_watcher_manager.start()
        _log.info("File watcher started")
    except Exception as e:
        _log.warning("File watcher start failed: %s", e)


def _stop_file_watcher() -> None:
    global _file_watcher_manager
    if _file_watcher_manager is None:
        return
    try:
        _file_watcher_manager.stop()
    except Exception:
        _log.debug("File watcher stop failed", exc_info=True)
    _file_watcher_manager = None


def _start_knowledge_watch() -> None:
    global _knowledge_watch_manager
    try:
        from utils.knowledge_automation import KnowledgeCorpusWatchManager

        if _knowledge_watch_manager is None:
            _knowledge_watch_manager = KnowledgeCorpusWatchManager()
        started = _knowledge_watch_manager.ensure_service()
        if started:
            _log.info("Knowledge corpus watch active (%d path(s))", started)
        else:
            _log.info("Knowledge automation service running (watch idle until enabled/corpora exist)")
    except Exception as e:
        _log.warning("Knowledge automation service start failed: %s", e)


def _stop_knowledge_watch() -> None:
    global _knowledge_watch_manager
    if _knowledge_watch_manager is None:
        return
    try:
        _knowledge_watch_manager.stop()
    except Exception:
        _log.debug("Knowledge corpus watch stop failed", exc_info=True)
    _knowledge_watch_manager = None


# ---------------------------------------------------------------------------
# Phase 2b §5 — Calendar-proximity poller
# ---------------------------------------------------------------------------

_calendar_proximity_thread: threading.Thread | None = None


def _start_calendar_proximity_thread() -> None:
    global _calendar_proximity_thread
    try:
        from core.config import get_calendar_proximity_config
        if not bool((get_calendar_proximity_config() or {}).get("enabled", False)):
            return
        from utils.calendar_proximity import run_calendar_proximity_loop
    except Exception:
        return
    if _calendar_proximity_thread is not None and _calendar_proximity_thread.is_alive():
        return
    _calendar_proximity_thread = threading.Thread(
        target=run_calendar_proximity_loop,
        kwargs={"should_stop": lambda: _shutdown_requested},
        daemon=True,
        name="calendar-proximity",
    )
    _calendar_proximity_thread.start()


# ---------------------------------------------------------------------------
# Phase 2b §6 — Weekly ledger self-reflection scheduler
# ---------------------------------------------------------------------------

_reflection_thread: threading.Thread | None = None


def _start_reflection_scheduler() -> None:
    global _reflection_thread
    try:
        from core.config import get_ledger_reflection_config
        if not bool((get_ledger_reflection_config() or {}).get("enabled", False)):
            return
        from utils.ledger_reflection import run_reflection_scheduler_loop
    except Exception:
        return
    if _reflection_thread is not None and _reflection_thread.is_alive():
        return
    _reflection_thread = threading.Thread(
        target=run_reflection_scheduler_loop,
        kwargs={"should_stop": lambda: _shutdown_requested},
        daemon=True,
        name="ledger-reflection",
    )
    _reflection_thread.start()


def _acquire_single_runner_lock() -> int | None:
    """Acquire an exclusive lock so only one ``mugi_runner`` drains shared state at once.

    Returns the open fd on success (retain it for the process lifetime — closing releases
    the lock), or ``None`` when another runner already holds it.

    Why: ``mugi_runner`` owns the delivery-drain, timer-drain, file-watcher, calendar, and
    single-queue heartbeat loops, all of which mutate shared ``~/.mugi`` stores
    (``delivery_queue.jsonl``, ``timers.jsonl``, ``proactive_queue.jsonl``, MEMORY, User
    Model proposals, kanban). A stale launchd runner racing a live one double-fires timers,
    double-drains deliveries, and double-runs heartbeats — the runner analogue of the
    dual-dispatcher incident. ``mugi_backend`` already guards itself this way
    (``_acquire_single_instance_lock``); the runner had no equivalent.
    """
    import fcntl

    home = os.path.expanduser("~/.mugi")
    try:
        os.makedirs(home, exist_ok=True)
    except OSError as exc:
        _log.error("cannot prepare runner lock dir %s: %s", home, exc)
        return None
    lock_path = os.path.join(home, "mugi_runner.pid")
    try:
        fd = os.open(lock_path, os.O_RDWR | os.O_CREAT, 0o600)
    except OSError as exc:
        _log.error("cannot open runner lock %s: %s", lock_path, exc)
        return None
    try:
        fcntl.lockf(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except (BlockingIOError, OSError):
        try:
            os.lseek(fd, 0, os.SEEK_SET)
            holder = os.read(fd, 64).decode("utf-8", errors="replace").strip() or "unknown"
        except OSError:
            holder = "unknown"
        _log.warning(
            "another mugi_runner is already running (pid=%s); refusing to start. Two runners "
            "draining ~/.mugi queues concurrently double-fire timers/deliveries/heartbeats. "
            "Stop the existing one (or `kill %s`) and relaunch.",
            holder,
            holder,
        )
        try:
            os.close(fd)
        except OSError:
            pass
        return None
    try:
        os.lseek(fd, 0, os.SEEK_SET)
        os.ftruncate(fd, 0)
        os.write(fd, str(os.getpid()).encode("ascii"))
    except OSError:
        pass
    return fd


def main():
    # Single-instance lock FIRST, before any drain thread or shared-state mutation. A stale
    # second runner defers cleanly (exit 0 so launchd KeepAlive does not crash-loop); the
    # kernel releases the fcntl lock automatically when the live runner exits.
    _runner_lock_fd = _acquire_single_runner_lock()
    if _runner_lock_fd is None:
        sys.exit(0)
    globals()["_MUGI_RUNNER_LOCK_FD"] = _runner_lock_fd

    import argparse
    parser = argparse.ArgumentParser(description="Background Mugi runner (heartbeat + IPC socket)")
    parser.add_argument("--heartbeat-interval", type=float, default=30, help="Minutes between standard heartbeats (default 30)")
    args = parser.parse_args()
    config = get_config()
    hb = get_heartbeat_config()
    # Build task list: use config intervals; 0 = disabled. CLI --heartbeat-interval overrides standard if provided.
    interval_cli = max(1.0, args.heartbeat_interval) if args.heartbeat_interval else None
    standard_min = hb.get("standard_interval_minutes", 30)
    if interval_cli is not None:
        standard_min = interval_cli
    memory_min = hb.get("memory_maintenance_interval_minutes", 360)
    maintenance_min = hb.get("maintenance_interval_minutes", memory_min)
    name_to_interval = {
        "standard": standard_min,
        "maintenance": maintenance_min,
        "consolidation": hb.get("consolidation_interval_minutes", 720),
        "curiosity": hb.get("curiosity_interval_minutes", 240),
        "research_prefetch": hb.get("research_prefetch_interval_minutes", 720),
        "drive_cycle": hb.get("drive_cycle_interval_minutes", 180),
        "drive_action_cycle": hb.get("drive_action_cycle_interval_minutes", 30),
        "knowledge_stale_refresh": 360,
    }
    tasks_to_run = []
    for task in HEARTBEAT_TASKS:
        name = task.get("name", "")
        interval = name_to_interval.get(name, task.get("interval_minutes", 30))
        if interval <= 0:
            _log.info("  Task [%s]: disabled (interval 0)", name)
            continue
        task_copy = dict(task)
        task_copy["interval_minutes"] = max(1.0, float(interval))
        tasks_to_run.append(task_copy)
        _log.info("  Task [%s]: every %.1f min (thread: %s)", task_copy["name"], task_copy["interval_minutes"], task_copy.get("thread_id", "default"))

    _log.info("Starting background runner with %d task type(s)", len(tasks_to_run))

    drives_cfg = get_drives_config()
    _autonomy_mode = str(drives_cfg.get("autonomy_mode", "bounded")).strip().lower()
    if _autonomy_mode == "aggressive":
        _log.warning(
            "autonomy_mode is 'aggressive': proactive allowlist is DISABLED for "
            "drive_action policy checks. All tools may be called autonomously."
        )
    _optional_script_integrity_check()
    ensure_mugi_workspace(config)
    _startup_cleanup()

    # Start the dedicated delivery-drain thread (5s tick + wake-on-enqueue).
    _start_delivery_drain_thread()

    # Phase 2b background services (all feature-flagged; no-op when disabled).
    _start_timer_drain_thread()
    _start_file_watcher()
    _start_knowledge_watch()
    _start_calendar_proximity_thread()
    _start_reflection_scheduler()

    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        async def _main():
            gateway = None
            if get_messaging_config().get("enabled"):
                try:
                    from messaging import Gateway
                    gateway = Gateway(loop)
                    await gateway.start()
                    _log.info("Messaging gateway started")
                except Exception as e:
                    _log.warning("Messaging gateway failed to start: %s", e)
            try:
                # Single queue: one task at a time, only when main chat idle >= IDLE_THRESHOLD_MINUTES
                await asyncio.gather(
                    _run_socket_server(loop),
                    _single_queue_scheduler_loop(loop, tasks_to_run),
                )
            finally:
                if gateway:
                    try:
                        await gateway.stop()
                    except Exception as e:
                        _log.warning("Gateway stop: %s", e)

        loop.run_until_complete(_main())
    except asyncio.CancelledError:
        pass
    except Exception as e:
        _log.exception("Fatal error in main loop: %s", e)
        sys.exit(1)
    finally:
        _stop_knowledge_watch()
        _stop_file_watcher()
        loop.close()
    _log.info("Background runner exited")


if __name__ == "__main__":
    main()
