from __future__ import annotations

import json
import os
import re
from dataclasses import dataclass
from datetime import date, datetime, timedelta, timezone
from typing import Any

from core.config import get_agent_plane_config as _core_get_agent_plane_config
from persistence import secure_atomic_write
from utils.path_safety import safe_resolve_path

_GLOBAL_CONTEXT_CHAR_CAP = 64_000
_MISSING_PLACEHOLDER = "<missing>"
_READ_ERROR_PREFIX = "<read error:"
_SIGNAL_FIELD_MAX_CHARS = 800
_SIGNALS_BLOCK_MAX_CHARS = 2_000
_AGENT_TURNS_REL_PATH = "memory/agent_turns.log"


@dataclass(frozen=True)
class _SectionSpec:
    label: str
    rel_path: str
    max_chars: int
    tail: bool = False


def get_agent_plane_config() -> dict[str, Any]:
    return _core_get_agent_plane_config()


def _read_text_head(path: str, max_chars: int) -> str:
    if max_chars <= 0:
        return ""
    max_bytes = max_chars * 4 + 1024
    with open(path, "rb") as f:
        data = f.read(max_bytes)
    if not data:
        return ""
    return data.decode("utf-8", errors="replace")[:max_chars]


def _read_text_tail(path: str, max_chars: int) -> str:
    if max_chars <= 0:
        return ""
    max_bytes = max_chars * 4 + 1024
    with open(path, "rb") as f:
        size = f.seek(0, os.SEEK_END)
        if size <= 0:
            return ""
        f.seek(max(0, size - max_bytes), os.SEEK_SET)
        data = f.read(max_bytes)
    if not data:
        return ""
    return data.decode("utf-8", errors="replace")[-max_chars:]


def _read_workspace_file(workspace: str, rel_path: str, max_chars: int, *, tail: bool = False) -> str:
    resolved, err = safe_resolve_path(workspace, rel_path, allow_write=False)
    if err or not resolved:
        return _MISSING_PLACEHOLDER
    if not os.path.isfile(resolved):
        return _MISSING_PLACEHOLDER
    try:
        return _read_text_tail(resolved, max_chars) if tail else _read_text_head(resolved, max_chars)
    except Exception as e:
        return f"{_READ_ERROR_PREFIX} {e}>"


def _resolve_heartbeat_path_content(workspace: str, max_chars: int) -> tuple[str, str]:
    for candidate in ("HEARTBEAT.md", "HEARTBEAT.MD"):
        content = _read_workspace_file(workspace, candidate, max_chars, tail=False)
        if content != _MISSING_PLACEHOLDER:
            return candidate, content
    return "HEARTBEAT.md", _MISSING_PLACEHOLDER


def _append_section(parts: list[str], label: str, rel_path: str, content: str, remaining: int) -> int:
    if remaining <= 0:
        return 0
    snippet = content if isinstance(content, str) else ""
    section = (
        f"### {label}\n"
        f"- path: {rel_path}\n"
        "```text\n"
        f"{snippet}\n"
        "```\n"
    )
    if len(section) > remaining:
        truncated = section[: max(0, remaining - 14)] + "\n[truncated]\n"
        parts.append(truncated)
        return len(truncated)
    parts.append(section)
    return len(section)


def _today_backwards(days: int) -> list[str]:
    out: list[str] = []
    today = date.today()
    for i in range(max(0, days)):
        d = today - timedelta(days=i)
        out.append(d.isoformat())
    return out


def _summary_limits_for_task(task_name: str) -> tuple[int, int]:
    if task_name == "maintenance":
        return 3, 20_000
    if task_name in {"memory_maintenance", "consciousness"}:
        return 3, 20_000
    if task_name in {"growth_check"}:
        return 7, 40_000
    return 0, 0


def _specs_for_task(task_name: str) -> list[_SectionSpec]:
    if task_name in {"standard", "drive_action"}:
        return [_SectionSpec("Heartbeat Instructions", "HEARTBEAT.md", 24_000)]
    if task_name in {"maintenance", "memory_maintenance"}:
        return [
            _SectionSpec("Long-Term Memory", "MEMORY.md", 16_000),
            _SectionSpec("Recurring Patterns", "memory/recurring-patterns.md", 8_000),
        ]
    if task_name == "growth_check":
        return [_SectionSpec("Recurring Patterns", "memory/recurring-patterns.md", 8_000)]
    if task_name == "consciousness":
        return [
            _SectionSpec("Long-Term Memory", "MEMORY.md", 16_000),
            _SectionSpec("Journal Tail", "JOURNAL.md", 8_000, tail=True),
        ]
    if task_name in {"collector", "feeds"}:
        return [_SectionSpec("Heartbeat Instructions", "HEARTBEAT.md", 6_000)]
    return []


def _shared_specs_for_wake(wake_kind: str, cfg: dict[str, Any]) -> list[_SectionSpec]:
    wake = str(wake_kind or "").strip().lower()
    if wake == "cron":
        specs = [
            _SectionSpec("Session State Tail", "SESSION-STATE.md", 4_000, tail=True),
            _SectionSpec("Long-Term Memory", "MEMORY.md", 4_000),
        ]
        if bool(cfg.get("cron_include_insights", False)):
            specs.append(_SectionSpec("Consolidated Insights", "memory/insights.md", 2_000))
        if bool(cfg.get("cron_include_user_md", False)):
            specs.append(_SectionSpec("User Profile", "USER.md", 2_000))
        return specs

    specs = [
        _SectionSpec("Session State", "SESSION-STATE.md", 8_000),
        _SectionSpec("Long-Term Memory", "MEMORY.md", 12_000),
        _SectionSpec("Consolidated Insights", "memory/insights.md", 6_000),
        _SectionSpec("Connections Journal", "memory/connections.md", 4_096, tail=True),
        _SectionSpec("User Profile", "USER.md", 4_000),
        _SectionSpec("Recent Growth Events", "memory/growth-events.md", 2_000, tail=True),
    ]
    if bool(cfg.get("include_agent_turns_tail", True)) and wake in {"drive_action", "standard", "memory_maintenance", "growth_check", "consciousness"}:
        specs.append(_SectionSpec("Recent Autonomous Turns", _AGENT_TURNS_REL_PATH, 2_500, tail=True))
    return specs


def _normalized_include_key(rel_path: str, tail: bool) -> tuple[str, bool]:
    return (str(rel_path or "").strip().lower(), bool(tail))


def _iter_specs_with_dedupe(task_name: str, wake_kind: str, cfg: dict[str, Any]) -> list[_SectionSpec]:
    out: list[_SectionSpec] = []
    seen: set[tuple[str, bool]] = set()
    for spec in _specs_for_task(task_name) + _shared_specs_for_wake(wake_kind, cfg):
        key = _normalized_include_key(spec.rel_path, spec.tail)
        if key in seen:
            continue
        seen.add(key)
        out.append(spec)
    return out


def _read_agent_turns_tail(workspace: str, max_chars: int) -> str:
    raw = _read_workspace_file(workspace, _AGENT_TURNS_REL_PATH, max_chars, tail=True)
    if raw in {_MISSING_PLACEHOLDER, ""}:
        return raw
    lines: list[str] = []
    for line in raw.splitlines():
        txt = line.strip()
        if not txt:
            continue
        try:
            obj = json.loads(txt)
            lines.append(json.dumps(obj, ensure_ascii=False))
        except Exception:
            continue
    return "\n".join(lines) if lines else _MISSING_PLACEHOLDER


def _sanitize_signal_value(value: Any) -> str:
    if value is None:
        return ""
    text = str(value)
    text = re.sub(r"[\x00-\x08\x0b-\x1f\x7f]", " ", text)
    text = text.replace("```", "` ` `")
    if len(text) > _SIGNAL_FIELD_MAX_CHARS:
        return text[:_SIGNAL_FIELD_MAX_CHARS] + " [truncated]"
    return text


def _render_sanitized_signals(signals: dict[str, Any] | None, remaining: int) -> tuple[str, int]:
    if not signals or remaining <= 0:
        return "", 0
    rows: list[str] = []
    for key in sorted(signals.keys()):
        safe_key = _sanitize_signal_value(key).replace("\n", " ").strip()[:80]
        safe_val = _sanitize_signal_value(signals.get(key))
        if not safe_key or not safe_val:
            continue
        rows.append(f"- {safe_key}: {safe_val}")
    if not rows:
        return "", 0
    block = "### Untrusted signal data\n" + "\n".join(rows) + "\n"
    if len(block) > _SIGNALS_BLOCK_MAX_CHARS:
        block = block[:_SIGNALS_BLOCK_MAX_CHARS] + "\n[truncated]\n"
    if len(block) > remaining:
        block = block[: max(0, remaining - 14)] + "\n[truncated]\n"
    return block, len(block)


def _build_drives_dashboard(workspace: str, limit: int, remaining: int) -> tuple[str, int]:
    if remaining <= 0 or limit <= 0:
        return "", 0
    try:
        from drive_engine import get_active_drives
    except Exception:
        return "", 0
    try:
        drives = list(get_active_drives(workspace=workspace, limit=max(1, limit)) or [])
    except Exception:
        return "", 0
    if not drives:
        return "", 0
    lines = ["### Active drives snapshot"]
    for d in drives[:limit]:
        did = str(d.get("id") or "").strip() or "unknown"
        title = str(d.get("title") or d.get("description") or "").strip()
        urg = d.get("urgency")
        stat = str(d.get("status") or "").strip()
        parts = [f"- {did}"]
        if title:
            parts.append(f"title={title[:80]}")
        if urg is not None:
            parts.append(f"urgency={urg}")
        if stat:
            parts.append(f"status={stat}")
        lines.append(", ".join(parts))
    block = "\n".join(lines) + "\n"
    if len(block) > remaining:
        block = block[: max(0, remaining - 14)] + "\n[truncated]\n"
    return block, len(block)


def _build_preload_sections(
    *,
    task_name: str,
    wake_kind: str,
    workspace: str,
    signals: dict[str, Any] | None = None,
    char_budget: int | None = None,
) -> str:
    if not workspace or not isinstance(workspace, str):
        return "No workspace context available."
    cfg = get_agent_plane_config()
    budget = int(char_budget or cfg.get("max_chars") or _GLOBAL_CONTEXT_CHAR_CAP)
    if wake_kind == "cron":
        budget = min(budget, int(cfg.get("cron_char_budget", 12_000)))
    remaining = max(0, budget)
    parts: list[str] = []
    wk_norm = str(wake_kind or "").strip().lower()
    try:
        from utils.memory_shard_layout import layout_is_sharded, list_shard_rel_paths
    except Exception:
        def layout_is_sharded(_w: str) -> bool:  # type: ignore[misc]
            return False

        def list_shard_rel_paths(_w: str) -> list[str]:  # type: ignore[misc]
            return []
    for spec in _iter_specs_with_dedupe(task_name, wake_kind, cfg):
        rel = spec.rel_path
        if rel in {"HEARTBEAT.md", "HEARTBEAT.MD"}:
            rel, content = _resolve_heartbeat_path_content(workspace, spec.max_chars)
        elif rel == _AGENT_TURNS_REL_PATH:
            content = _read_agent_turns_tail(workspace, spec.max_chars)
        elif rel == "MEMORY.md" and layout_is_sharded(workspace):
            hub_cap = max(1500, min(spec.max_chars // 4, 8000))
            hub_content = _read_workspace_file(workspace, "MEMORY.md", hub_cap, tail=spec.tail)
            if wk_norm == "consolidation":
                try:
                    from utils.memory_shard_layout import strip_dated_memory_blocks_from_hub_text

                    hub_content = strip_dated_memory_blocks_from_hub_text(hub_content)
                except Exception:
                    pass
            used = _append_section(parts, f"{spec.label} (hub)", "MEMORY.md", hub_content, remaining)
            remaining -= used
            shard_rels = list_shard_rel_paths(workspace)
            _max_plane_shards = 24
            if shard_rels and remaining > 80:
                shard_budget_total = min(remaining, int(spec.max_chars * 3 // 4), 48_000)
                n = min(len(shard_rels), _max_plane_shards)
                per = max(1200, shard_budget_total // max(1, n))
                for srel in shard_rels[:_max_plane_shards]:
                    if remaining <= 60:
                        break
                    take = min(per, remaining - 40)
                    if take < 200:
                        break
                    scontent = _read_workspace_file(workspace, srel, take, tail=False)
                    used_s = _append_section(
                        parts,
                        f"{spec.label} (shard: {os.path.basename(srel)})",
                        srel,
                        scontent,
                        remaining,
                    )
                    remaining -= used_s
            continue
        else:
            content = _read_workspace_file(workspace, rel, spec.max_chars, tail=spec.tail)
        used = _append_section(parts, spec.label, rel, content, remaining)
        remaining -= used
        if remaining <= 0:
            break

    days, max_chars = _summary_limits_for_task(task_name)
    if days > 0 and max_chars > 0 and remaining > 0:
        summary_budget = min(max_chars, remaining)
        per_file = max(2_000, max_chars // days)
        for day in _today_backwards(days):
            rel = f"memory/summaries/{day}.md"
            content = _read_workspace_file(workspace, rel, per_file, tail=False)
            if content == _MISSING_PLACEHOLDER:
                continue
            used = _append_section(parts, f"Daily Summary {day}", rel, content, summary_budget)
            summary_budget -= used
            remaining -= used
            if summary_budget <= 0 or remaining <= 0:
                break

    if remaining > 0:
        drives_block, used = _build_drives_dashboard(
            workspace=workspace,
            limit=max(1, int(get_agent_plane_config().get("drive_dashboard_limit", 5))),
            remaining=remaining,
        )
        if drives_block:
            parts.append(drives_block)
            remaining -= used

    if (
        remaining > 0
        and wake_kind != "cron"
        and bool(cfg.get("enabled", True))
        and bool(cfg.get("include_open_loops", True))
    ):
        um_phrases_for_dedupe: list[str] = []
        try:
            from drive_engine import _user_model_goal_phrases

            um_phrases_for_dedupe = _user_model_goal_phrases(workspace)
        except Exception:
            um_phrases_for_dedupe = []
        try:
            from utils.user_model import load_user_model

            model = load_user_model(workspace)
            um_lines: list[str] = []

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

            active_goals = [
                g for g in model.goals
                if g.get("status") in (None, "active", "in_progress")
                and str(g.get("title") or "").strip()
            ]
            if active_goals:
                if um_lines:
                    um_lines.append("")
                um_lines.append("### User Model — Active Goals")
                for g in active_goals[:6]:
                    title = g.get("title", "Untitled")
                    horizon = g.get("horizon", "")
                    why = (g.get("why_it_matters") or "")[:120]
                    line = f"- {title}"
                    if horizon:
                        line += f" ({horizon})"
                    if why:
                        line += f" — {why}"
                    um_lines.append(line)

            active_connections = [
                c for c in model.connections
                if isinstance(c, dict) and str(c.get("text") or "").strip()
            ]
            if active_connections:
                if um_lines:
                    um_lines.append("")
                um_lines.append("### User Model — Connections & Threads")
                for conn in active_connections[:6]:
                    text = str(conn.get("text") or "").strip()[:200]
                    related = conn.get("related_to")
                    if isinstance(related, list):
                        topics = [str(t).strip() for t in related if str(t).strip()]
                    else:
                        topics = []
                    line = f"- {text}"
                    if topics:
                        line += f" (related: {', '.join(topics[:4])})"
                    um_lines.append(line)

            if um_lines:
                block = "### Personal Commitments & Goals (from User Model)\n" + "\n".join(um_lines) + "\n"
                if len(block) <= remaining:
                    parts.append(block)
                    remaining -= len(block)
        except Exception:
            pass

        try:
            from utils.open_loops import render_open_loops_section
            ol_block = render_open_loops_section(
                workspace,
                max_chars=min(3_000, remaining),
                exclude_phrases=um_phrases_for_dedupe,
            )
            if ol_block:
                parts.append(ol_block)
                remaining -= len(ol_block)
        except Exception:
            pass

    elif remaining > 0 and wake_kind != "cron":
        # User Model block when open_loops section is disabled but UM preload is wanted.
        try:
            from utils.user_model import load_user_model
            model = load_user_model(workspace)
            um_lines: list[str] = []

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

            active_goals = [
                g for g in model.goals
                if g.get("status") in (None, "active", "in_progress")
                and str(g.get("title") or "").strip()
            ]
            if active_goals:
                if um_lines:
                    um_lines.append("")
                um_lines.append("### User Model — Active Goals")
                for g in active_goals[:6]:
                    title = g.get("title", "Untitled")
                    horizon = g.get("horizon", "")
                    why = (g.get("why_it_matters") or "")[:120]
                    line = f"- {title}"
                    if horizon:
                        line += f" ({horizon})"
                    if why:
                        line += f" — {why}"
                    um_lines.append(line)

            active_connections = [
                c for c in model.connections
                if isinstance(c, dict) and str(c.get("text") or "").strip()
            ]
            if active_connections:
                if um_lines:
                    um_lines.append("")
                um_lines.append("### User Model — Connections & Threads")
                for conn in active_connections[:6]:
                    text = str(conn.get("text") or "").strip()[:200]
                    related = conn.get("related_to")
                    if isinstance(related, list):
                        topics = [str(t).strip() for t in related if str(t).strip()]
                    else:
                        topics = []
                    line = f"- {text}"
                    if topics:
                        line += f" (related: {', '.join(topics[:4])})"
                    um_lines.append(line)

            if um_lines:
                block = "### Personal Commitments & Goals (from User Model)\n" + "\n".join(um_lines) + "\n"
                if len(block) <= remaining:
                    parts.append(block)
                    remaining -= len(block)
        except Exception:
            pass

    # Legacy duplicate UM block removed — consolidated above (Phase 5C).

    if remaining > 0:
        sig_block, used = _render_sanitized_signals(signals, remaining)
        if sig_block:
            parts.append(sig_block)
            remaining -= used

    if not parts:
        return "No pre-loaded workspace context found for this task."
    return "\n".join(parts)


def _format_proactive_envelope(wake_kind: str, local_time_iso: str, inner_markdown: str) -> str:
    return (
        f"Wake: {wake_kind}\n"
        f"Local time: {local_time_iso}\n\n"
        f"{inner_markdown}"
    )


def build_heartbeat_sections_only(
    task_name: str,
    workspace: str,
    *,
    wake_kind: str = "standard",
    signals: dict[str, Any] | None = None,
) -> str:
    return _build_preload_sections(
        task_name=task_name,
        wake_kind=wake_kind,
        workspace=workspace,
        signals=signals,
        char_budget=None,
    )


def assemble_heartbeat_user_message(
    *,
    task_name: str,
    workspace: str,
    prompt_body: str,
    wake_kind: str | None = None,
    signals: dict[str, Any] | None = None,
    due_tasks: list[Any] | None = None,
) -> str:
    wk = str(wake_kind or task_name or "standard").strip() or "standard"
    preload = _build_preload_sections(
        task_name=task_name,
        wake_kind=wk,
        workspace=workspace,
        signals=signals,
        char_budget=None,
    )
    due_block = ""
    if due_tasks:
        names = []
        for t in due_tasks:
            name = getattr(t, "name", None) or (t.get("name") if isinstance(t, dict) else None)
            if name:
                names.append(str(name))
        if names:
            due_block = (
                "\n\n## Due Tasks (from HEARTBEAT.md tasks block)\n"
                + "\n".join(f"- {n}" for n in names)
                + "\nOnly respond to the sections above that are actually due this cycle."
            )
    inner = (
        f"{prompt_body}{due_block}\n\n---\n\n## Pre-loaded workspace context\n\n{preload}\n\n"
        "Use this content directly. Call read_file only for files not included above."
    )
    try:
        local_ts = datetime.now().astimezone().isoformat()
    except Exception:
        local_ts = datetime.now(timezone.utc).isoformat()
    return _format_proactive_envelope(wk, local_ts, inner)


def build_wake_context_snippet(
    *,
    workspace: str,
    wake_kind: str = "cron",
    task_name: str = "cron",
    signals: dict[str, Any] | None = None,
) -> str:
    return _build_preload_sections(
        task_name=task_name,
        wake_kind=wake_kind,
        workspace=workspace,
        signals=signals,
        char_budget=None,
    )


def _redact_turn_summary(text: str, max_chars: int = 220) -> str:
    t = str(text or "").strip().replace("\n", " ")
    t = re.sub(r"sk-[A-Za-z0-9_-]{8,}", "[REDACTED]", t)
    t = re.sub(r"(?i)bearer\s+[a-z0-9._-]{8,}", "Bearer [REDACTED]", t)
    t = re.sub(r"(?i)api[_-]?key\s*[:=]\s*[^\s]+", "api_key=[REDACTED]", t)
    return t[:max_chars]


def record_proactive_turn_summary(
    *,
    workspace: str,
    wake_kind: str,
    thread_id: str,
    status: str,
    tool_success_count: int = 0,
    one_line_summary: str = "",
    drive_id: str | None = None,
) -> None:
    cfg = get_agent_plane_config()
    if not bool(cfg.get("enabled", True)):
        return
    if not bool(cfg.get("enable_turn_log", True)):
        return
    if isinstance(thread_id, str) and thread_id.startswith("btw_"):
        return
    path, err = safe_resolve_path(workspace, _AGENT_TURNS_REL_PATH, allow_write=True, create_parents=True)
    if err or not path:
        return
    ts = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
    entry = {
        "timestamp": ts,
        "wake_kind": str(wake_kind or ""),
        "thread_id": str(thread_id or ""),
        "status": str(status or ""),
        "tool_success_count": int(tool_success_count or 0),
        "one_line_summary": _redact_turn_summary(one_line_summary, 220),
    }
    if drive_id:
        entry["drive_id"] = str(drive_id)

    existing = ""
    if os.path.isfile(path):
        try:
            existing = _read_text_tail(path, int(cfg.get("turn_log_max_kb", 128)) * 1024)
        except Exception:
            existing = ""

    lines = [line for line in existing.splitlines() if line.strip()]
    lines.append(json.dumps(entry, ensure_ascii=False))
    max_lines = max(1, int(cfg.get("turn_log_max_lines", 2_000)))
    if len(lines) > max_lines:
        lines = lines[-max_lines:]
    payload = "\n".join(lines).rstrip() + "\n"
    max_bytes = max(8 * 1024, int(cfg.get("turn_log_max_kb", 128)) * 1024)
    if len(payload.encode("utf-8", errors="ignore")) > max_bytes:
        payload = payload.encode("utf-8", errors="ignore")[-max_bytes:].decode("utf-8", errors="ignore")
        payload = payload[payload.find("\n") + 1 :] if "\n" in payload else payload
        payload = payload.rstrip() + "\n"

    secure_atomic_write(path, payload)
    try:
        os.chmod(path, 0o600)
    except Exception:
        pass


# ---------------------------------------------------------------------------
# Pre-flight skip gate — decide whether to skip the heartbeat LLM call entirely
# ---------------------------------------------------------------------------

def _get_heartbeat_skip_config() -> dict[str, Any]:
    try:
        from core.config import get_heartbeat_config
        cfg = get_heartbeat_config() or {}
    except Exception:
        cfg = {}
    return {
        "skip_gate_enabled": bool(cfg.get("skip_gate_enabled", True)),
        "skip_gate_debug": bool(cfg.get("skip_gate_debug", False)),
        "standard_interval_minutes": int(cfg.get("standard_interval_minutes", 30) or 30),
        "high_tension_threshold": float(cfg.get("skip_gate_tension_threshold", 0.6) or 0.6),
    }


def _heartbeat_cache_is_fresh(task_name: str, interval_minutes: int) -> bool:
    try:
        from persistence import MUGI_DIR
    except Exception:
        return False
    path = os.path.join(MUGI_DIR, "heartbeat_cache.json")
    if not os.path.isfile(path):
        return False
    try:
        mtime = os.path.getmtime(path)
    except OSError:
        return False
    age_seconds = max(0.0, datetime.now(timezone.utc).timestamp() - mtime)
    fresh_window = max(60.0, float(interval_minutes) * 60.0 / 2.0)
    if age_seconds >= fresh_window:
        return False
    try:
        with open(path, "r", encoding="utf-8") as f:
            cache = json.load(f)
    except Exception:
        return False
    if not isinstance(cache, dict):
        return False
    if str(cache.get("task_name") or "").strip() != str(task_name or "").strip():
        return False
    return True


def _has_high_tension_drive(workspace: str, threshold: float) -> bool:
    if not workspace:
        return False
    try:
        from drive_engine import get_active_drives
    except Exception:
        return False
    try:
        drives = list(get_active_drives(workspace=workspace, limit=10) or [])
    except Exception:
        return False
    for d in drives:
        status = str(d.get("status") or "").strip().lower()
        if status in {"resolved", "dismissed", "expired", "blocked"}:
            continue
        try:
            tension = float(d.get("tension") or 0.0)
        except (TypeError, ValueError):
            tension = 0.0
        try:
            urgency = float(d.get("urgency") or 0.0)
        except (TypeError, ValueError):
            urgency = 0.0
        if tension >= threshold or urgency >= threshold:
            return True
    return False


def _signals_all_unchanged(task_name: str, signals: dict[str, Any] | None) -> bool:
    """Return True only if every provided signal has been seen before (same fingerprint)."""
    if not signals:
        return True
    try:
        from utils.proactive_queue import should_emit_proactive_notification
    except Exception:
        return True
    any_new = False
    for sig_key, sig_value in signals.items():
        content = ""
        if isinstance(sig_value, (str, int, float, bool)):
            content = str(sig_value)
        else:
            try:
                content = json.dumps(sig_value, sort_keys=True, ensure_ascii=True, default=str)
            except Exception:
                content = str(sig_value)
        fingerprint_key = f"heartbeat:{task_name}:{sig_key}"
        if should_emit_proactive_notification(fingerprint_key, content):
            any_new = True
    return not any_new


def should_skip_heartbeat(
    task_name: str,
    workspace: str,
    *,
    signals: dict[str, Any] | None = None,
    has_tasks_block: bool | None = None,
    due_task_count: int | None = None,
) -> tuple[bool, str]:
    """Return ``(should_skip, reason)`` for the current heartbeat cycle.

    Skips when **all** conditions hold:
    - skip gate is enabled in config,
    - ``due_task_count`` is 0 (when a tasks block is present in HEARTBEAT.md),
    - no high-tension drive is active,
    - signal fingerprints (if any) are unchanged since last cycle,
    - heartbeat_cache.json is fresh (recent write for the same task).

    Callers may pass ``has_tasks_block`` / ``due_task_count`` to avoid re-parsing
    HEARTBEAT.md. When both are omitted, the helper computes them itself.
    """
    cfg = _get_heartbeat_skip_config()
    if not cfg["skip_gate_enabled"]:
        return False, "skip_gate_disabled"

    task = str(task_name or "standard").strip() or "standard"

    if has_tasks_block is None or due_task_count is None:
        try:
            from utils.heartbeat_tasks import due_tasks as _due, parse_heartbeat_tasks as _parse
            parsed = _parse(workspace) if workspace else []
            if has_tasks_block is None:
                has_tasks_block = bool(parsed)
            if due_task_count is None:
                due_task_count = len(_due(parsed)) if parsed else 0
        except Exception:
            has_tasks_block = bool(has_tasks_block)
            due_task_count = int(due_task_count or 0)

    if has_tasks_block and int(due_task_count or 0) > 0:
        return False, "tasks_due"

    if _has_high_tension_drive(workspace, cfg["high_tension_threshold"]):
        return False, "high_tension_drive"

    if not _signals_all_unchanged(task, signals):
        return False, "new_signals"

    if not _heartbeat_cache_is_fresh(task, cfg["standard_interval_minutes"]):
        return False, "stale_cache"

    reason = "no_work"
    if has_tasks_block and int(due_task_count or 0) == 0:
        reason = "no_tasks_due"
    return True, reason
