"""Shared proactive queue utilities: entry identity, schema constants, and file bounds.

Single source of truth for proactive_entry_id so all consumers (backend HTTP routes,
SSE poller, headless poller, tests) produce identical identifiers for the same JSONL row.
"""

from __future__ import annotations

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

try:
    import fcntl
    _HAS_FLOCK = True
except ImportError:
    _HAS_FLOCK = False

_log = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# Schema constants
# ---------------------------------------------------------------------------

PROACTIVE_SCHEMA_VERSION = 2

KIND_NOTIFICATION = "notification"
KIND_OPEN_LOOP = "open_loop"
KIND_STALE_DECISION = "stale_decision"
KIND_OBSERVATION = "observation"
_VALID_KINDS = frozenset({KIND_NOTIFICATION, KIND_OPEN_LOOP, KIND_STALE_DECISION, KIND_OBSERVATION})

APPROVAL_PENDING = "pending"
APPROVAL_STARTING = "starting"
APPROVAL_APPROVED = "approved"
APPROVAL_DISMISSED = "dismissed"
APPROVAL_ENGAGED = "engaged"
_VALID_APPROVAL_STATUSES = frozenset({
    APPROVAL_PENDING,
    APPROVAL_STARTING,
    APPROVAL_APPROVED,
    APPROVAL_DISMISSED,
    APPROVAL_ENGAGED,
})

TIER_IN_THREAD = "in_thread"
TIER_KANBAN = "kanban"
_VALID_SUGGESTED_TIERS = frozenset({TIER_IN_THREAD, TIER_KANBAN})

PRIORITY_HIGH = "high"
PRIORITY_NORMAL = "normal"
PRIORITY_LOW = "low"
_VALID_PRIORITIES = frozenset({PRIORITY_HIGH, PRIORITY_NORMAL, PRIORITY_LOW})
_PRIORITY_SORT_KEY = {PRIORITY_HIGH: 0, PRIORITY_NORMAL: 1, PRIORITY_LOW: 2}

# File bounds: refuse append when queue exceeds these limits.
MAX_QUEUE_LINES = 5_000
MAX_QUEUE_BYTES = 10 * 1024 * 1024  # 10 MB

# Field length caps for metadata strings written by internal callers.
MAX_BATCH_KEY_LEN = 120
MAX_DEDUPE_KEY_LEN = 120
MAX_DRIVE_ID_LEN = 80
MAX_THREAD_ID_LEN = 120
MAX_SOURCE_LEN = 50
MAX_CONTENT_LEN = 2_000

# Pending API response cap.
MAX_PENDING_ENTRIES = 200

# mark_processed request cap.
MAX_MARK_PROCESSED_IDS = 500

# ---------------------------------------------------------------------------
# Entry identity (single implementation for all consumers)
# ---------------------------------------------------------------------------

_PROACTIVE_QUEUE_LOCK = threading.Lock()


def _flock_append(queue_path: str, line: str) -> None:
    """Append a single line with cross-process flock coordination."""
    if _HAS_FLOCK:
        lock_path = queue_path + ".lock"
        fd = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o600)
        try:
            fcntl.flock(fd, fcntl.LOCK_EX)
            with open(queue_path, "a", encoding="utf-8") as f:
                f.write(line)
        finally:
            fcntl.flock(fd, fcntl.LOCK_UN)
            os.close(fd)
    else:
        with open(queue_path, "a", encoding="utf-8") as f:
            f.write(line)


def proactive_entry_id(item: dict) -> str:
    """Stable id from raw ts|source|content (UTF-8, SHA-256 prefix).

    All consumers MUST use this function — never hash sanitized/capped content.
    """
    ts = str(item.get("ts") or "")
    source = str(item.get("source") or "heartbeat")
    content = str(item.get("content") or "").strip()
    return hashlib.sha256(f"{ts}|{source}|{content}".encode("utf-8")).hexdigest()[:16]


# ---------------------------------------------------------------------------
# Sanitization helpers
# ---------------------------------------------------------------------------

def sanitize_system_event_text(text: str, max_chars: int = 500) -> str:
    """Strip control chars and cap length for SSE / UI display."""
    raw = str(text or "")
    raw = raw.replace("\r", " ").replace("\x00", " ")
    raw = "".join(ch for ch in raw if ch.isprintable() or ch in ("\n", "\t"))
    return raw.strip()[:max_chars]


def normalize_priority(value: Any) -> str:
    """Coerce to valid priority string; unknown values become 'normal'."""
    v = str(value or "").strip().lower()
    return v if v in _VALID_PRIORITIES else PRIORITY_NORMAL


def _cap_str(value: Any, max_len: int) -> str:
    return str(value or "")[:max_len]


# ---------------------------------------------------------------------------
# Queue file bounds check
# ---------------------------------------------------------------------------

def _queue_file_within_bounds(queue_path: str) -> bool:
    """Return True if the queue file is below size/line limits."""
    try:
        if not os.path.isfile(queue_path):
            return True
        size = os.path.getsize(queue_path)
        if size > MAX_QUEUE_BYTES:
            _log.warning("Proactive queue exceeds max bytes (%d > %d); refusing append", size, MAX_QUEUE_BYTES)
            return False
        with open(queue_path, "r", encoding="utf-8") as f:
            line_count = sum(1 for _ in f)
        if line_count > MAX_QUEUE_LINES:
            _log.warning("Proactive queue exceeds max lines (%d > %d); refusing append", line_count, MAX_QUEUE_LINES)
            return False
    except OSError as e:
        _log.debug("Queue bounds check failed: %s", e)
    return True


# ---------------------------------------------------------------------------
# Append (single entry point — replaces mugi_runner._append_proactive_queue)
# ---------------------------------------------------------------------------

def append_proactive_queue(
    queue_path: str,
    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",
) -> bool:
    """Append one proactive notification to queue JSONL.

    Returns True on success, False if skipped (empty content or file bounds exceeded).
    Caller is responsible for content cleaning (strip_think_tags etc.) before calling.

    ``kind`` distinguishes entry types: "notification" (default), "open_loop",
    "stale_decision", "observation".
    """
    cleaned = str(content or "").strip()
    if not cleaned:
        _log.debug("Proactive queue: skipped empty content (source=%s)", source)
        return False

    kind_norm = str(kind or KIND_NOTIFICATION).strip().lower()
    if kind_norm not in _VALID_KINDS:
        kind_norm = KIND_NOTIFICATION

    queue_text = cleaned[:MAX_CONTENT_LEN]
    entry: dict[str, Any] = {
        "schema_version": PROACTIVE_SCHEMA_VERSION,
        "ts": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
        "content": queue_text,
        "source": _cap_str(source, MAX_SOURCE_LEN),
        "kind": kind_norm,
        "processed": False,
    }
    if priority is not None:
        entry["priority"] = normalize_priority(priority)
    if batch_key is not None:
        entry["batch_key"] = _cap_str(batch_key, MAX_BATCH_KEY_LEN)
    if dedupe_key is not None:
        entry["dedupe_key"] = _cap_str(dedupe_key, MAX_DEDUPE_KEY_LEN)
    if drive_id is not None:
        entry["drive_id"] = _cap_str(drive_id, MAX_DRIVE_ID_LEN)
    if thread_id is not None:
        entry["thread_id"] = _cap_str(thread_id, MAX_THREAD_ID_LEN)
    if expires_at is not None:
        entry["expires_at"] = str(expires_at)[:40]

    try:
        with _PROACTIVE_QUEUE_LOCK:
            if not _queue_file_within_bounds(queue_path):
                return False
            _flock_append(queue_path, json.dumps(entry, ensure_ascii=True) + "\n")
        try:
            os.chmod(queue_path, 0o600)
        except OSError:
            pass
        _log.info("Proactive queue: appended %d chars (source=%s)", len(queue_text), source)
        return True
    except Exception as e:
        _log.warning("Failed to write proactive queue: %s", e)
        return False


def append_observation_queue(
    queue_path: str,
    *,
    noticed: str,
    proposed_work: str,
    source: str,
    content: str | None = None,
    suggested_tier: str = TIER_IN_THREAD,
    suggested_assignee: str = "general",
    dedupe_key: str | None = None,
    priority: str | None = None,
    thread_id: str | None = None,
) -> bool:
    """Append a consent-gated observation (notice → approve → work).

    ``content`` is the human-facing inbox line; built from noticed/proposed_work when omitted.
    """
    noticed_clean = str(noticed or "").strip()
    work_clean = str(proposed_work or "").strip()
    if not noticed_clean or not work_clean:
        _log.debug("Observation queue: skipped empty noticed/proposed_work (source=%s)", source)
        return False

    tier = str(suggested_tier or TIER_IN_THREAD).strip().lower()
    if tier not in _VALID_SUGGESTED_TIERS:
        tier = TIER_IN_THREAD

    display = str(content or "").strip()
    if not display:
        display = f"I noticed {noticed_clean}. Would you like me to {work_clean}?"

    observation_payload: dict[str, Any] = {
        "noticed": noticed_clean[:800],
        "proposed_work": work_clean[:800],
        "suggested_tier": tier,
        "suggested_assignee": _cap_str(suggested_assignee, MAX_SOURCE_LEN),
    }
    if dedupe_key is not None:
        observation_payload["dedupe_key"] = _cap_str(dedupe_key, MAX_DEDUPE_KEY_LEN)

    try:
        from utils.noticing_dedupe import (
            is_duplicate_observation_noticed,
            semantic_fingerprint,
        )

        semantic_key = semantic_fingerprint(noticed_clean, extra=work_clean)
        if semantic_key and dedupe_key is None:
            dedupe_key = f"obs_sem:{semantic_key}"
            observation_payload["dedupe_key"] = _cap_str(dedupe_key, MAX_DEDUPE_KEY_LEN)
            entry_dedupe_key = dedupe_key
        else:
            entry_dedupe_key = dedupe_key

        with _PROACTIVE_QUEUE_LOCK:
            if not _queue_file_within_bounds(queue_path):
                return False

            existing_lines: list[str] = []
            if os.path.isfile(queue_path):
                try:
                    with open(queue_path, "r", encoding="utf-8") as f:
                        existing_lines = f.readlines()
                except OSError as e:
                    _log.debug("Observation dedupe scan failed: %s", e)

            if entry_dedupe_key and existing_lines:
                if find_pending_observation_by_dedupe(existing_lines, entry_dedupe_key):
                    _log.debug(
                        "Observation queue: skipped duplicate pending dedupe_key=%s (source=%s)",
                        entry_dedupe_key,
                        source,
                    )
                    return False

            if existing_lines:
                pending_obs = pending_entries_from_lines(existing_lines, include_notified=False)
                obs_rows = [
                    e for e in pending_obs if str(e.get("kind") or "") == KIND_OBSERVATION
                ]
                from utils.noticing_dedupe import collect_pending_observation_notices

                prior_noticed, prior_work = collect_pending_observation_notices(obs_rows)
                if is_duplicate_observation_noticed(
                    noticed_clean,
                    work_clean,
                    existing_noticed=prior_noticed,
                    existing_work=prior_work,
                ):
                    _log.debug(
                        "Observation queue: skipped semantic duplicate (source=%s noticed=%r)",
                        source,
                        noticed_clean[:80],
                    )
                    return False

            queue_text = display[:MAX_CONTENT_LEN]
            entry: dict[str, Any] = {
                "schema_version": PROACTIVE_SCHEMA_VERSION,
                "ts": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
                "content": queue_text,
                "source": _cap_str(source, MAX_SOURCE_LEN),
                "kind": KIND_OBSERVATION,
                "processed": False,
                "observation": observation_payload,
                "approval": {
                    "status": APPROVAL_PENDING,
                    "approved_at": None,
                    "kanban_task_id": None,
                    "thread_id": None,
                },
            }
            if priority is not None:
                entry["priority"] = normalize_priority(priority)
            if entry_dedupe_key is not None:
                entry["dedupe_key"] = _cap_str(entry_dedupe_key, MAX_DEDUPE_KEY_LEN)
            if thread_id is not None:
                entry["thread_id"] = _cap_str(thread_id, MAX_THREAD_ID_LEN)

            _flock_append(queue_path, json.dumps(entry, ensure_ascii=True) + "\n")
        try:
            os.chmod(queue_path, 0o600)
        except OSError:
            pass
        _log.info("Observation queue: appended (source=%s tier=%s)", source, tier)
        return True
    except Exception as e:
        _log.warning("Failed to write observation queue: %s", e)
        return False


def _emit_observation_subscribe_warning(*, task_id: str, channel: str, exc: Exception) -> None:
    _log.warning(
        "observation approve %s subscribe failed for %s: %s",
        channel,
        task_id,
        exc,
    )
    try:
        from activity_bus import emit_activity_event

        emit_activity_event(
            {
                "source": "proactive",
                "kind": "observation_subscribe_failed",
                "title": "Kanban subscription missed",
                "summary": f"{channel} subscribe failed for task {task_id}",
                "tags": ["proactive", "kanban"],
                "meta": {"task_id": task_id, "channel": channel, "error": str(exc)[:200]},
            }
        )
    except Exception:
        _log.debug("activity emit for subscribe failure failed", exc_info=True)


def observation_starting_mutator(
    item: dict,
    *,
    prior_approval: dict[str, Any],
    started_at: str,
    thread_id: str | None = None,
    matched_notified: bool = False,
) -> dict:
    """Claim an observation row for in-flight approve/engage (crash-safe)."""
    approval = {
        **prior_approval,
        "status": APPROVAL_STARTING,
        "starting_at": started_at,
    }
    if thread_id:
        approval["thread_id"] = thread_id
    item["approval"] = approval
    item["processed"] = False
    item["notified"] = matched_notified
    return item


def revert_observation_to_pending_mutator(
    item: dict,
    *,
    prior_approval: dict[str, Any],
    matched_notified: bool = False,
) -> dict:
    item["approval"] = dict(prior_approval)
    item["processed"] = False
    item["notified"] = matched_notified
    return item


def reconcile_stale_proactive_starts(queue_path: str) -> int:
    """Revert orphaned ``starting`` observations to pending after crash/restart."""
    if not (queue_path or "").strip() or not os.path.isfile(queue_path):
        return 0

    reverted = 0

    def _maybe_revert(item: dict) -> dict:
        nonlocal reverted
        if str(item.get("kind") or "") != KIND_OBSERVATION:
            return item
        appr = approval_payload_from_entry(item)
        if appr.get("status") != APPROVAL_STARTING:
            return item
        item["approval"] = {
            "status": APPROVAL_PENDING,
            "approved_at": None,
            "kanban_task_id": None,
            "thread_id": None,
        }
        item["processed"] = False
        reverted += 1
        return item

    try:
        with _PROACTIVE_QUEUE_LOCK:
            with open(queue_path, "r", encoding="utf-8") as f:
                lines = f.readlines()
            changed = False
            new_lines: list[str] = []
            for line in lines:
                raw = (line or "").strip()
                if not raw:
                    new_lines.append(line)
                    continue
                try:
                    item = json.loads(raw)
                except Exception:
                    new_lines.append(line)
                    continue
                if not isinstance(item, dict):
                    new_lines.append(line)
                    continue
                before = json.dumps(item, sort_keys=True)
                updated_item = _maybe_revert(item)
                after = json.dumps(updated_item, sort_keys=True)
                if before != after:
                    changed = True
                    new_lines.append(json.dumps(updated_item, ensure_ascii=True) + "\n")
                else:
                    new_lines.append(line)
            if changed:
                flock_rewrite(queue_path, new_lines)
                try:
                    os.chmod(queue_path, 0o600)
                except OSError:
                    pass
        if reverted:
            _log.warning(
                "Reverted %d orphaned proactive observation(s) from starting → pending",
                reverted,
            )
    except Exception:
        _log.warning("proactive start reconcile failed", exc_info=True)
        return 0
    return reverted


def spawn_kanban_for_approved_observation(
    *,
    item_id: str,
    noticed: str,
    proposed: str,
    source: str,
    assignee: str,
    origin_thread_id: str,
    approved_at: str,
    drive_id: str = "",
) -> dict[str, Any]:
    """Create a kanban worker task and terminal subscriptions for an approved observation."""
    import kanban_storage as ks
    from utils.kanban_notifications import kanban_notify_subscribe_if_absent

    title_seed = proposed[:40] or noticed[:40] or "Approved work"
    title_raw = title_seed[:40]
    title = title_raw.rsplit(" ", 1)[0] if len(title_raw) > 40 and " " in title_raw else title_raw
    body = (
        "Approved proactive observation.\n\n"
        f"Source: {source}\n"
        f"You noticed: {noticed}\n"
        f"Approved work: {proposed}\n\n"
        "Execute using your tools. Post progress with kanban_post_progress if this takes a while. "
        "End with a kanban-result block on completion."
    )
    metadata = {
        "proactive_observation_id": item_id,
        "noticed": noticed[:400],
        "source": source,
        "approved_at": approved_at,
    }
    if drive_id:
        metadata["drive_id"] = drive_id
    task = ks.create_task(
        title=title,
        body=body,
        assignee=(assignee or "general").strip() or "general",
        origin_thread_id=origin_thread_id,
        tags=["observation-approved"],
        metadata=metadata,
        idempotency_key=f"observation:{item_id}",
        priority=3,
    )
    task_id = str(task.get("id") or "").strip()
    if task_id and origin_thread_id:
        try:
            kanban_notify_subscribe_if_absent(
                task_id,
                channel="thread",
                target=origin_thread_id,
                fire_on=("done", "gave_up"),
            )
        except Exception as exc:
            _emit_observation_subscribe_warning(task_id=task_id, channel="thread", exc=exc)
        try:
            kanban_notify_subscribe_if_absent(
                task_id,
                channel="notification",
                target=f"Approved: {title}",
                fire_on=("done", "gave_up"),
            )
        except Exception as exc:
            _emit_observation_subscribe_warning(task_id=task_id, channel="notification", exc=exc)
    try:
        from utils.kanban_dispatcher import nudge_dispatcher

        nudge_dispatcher()
    except Exception as exc:
        _log.warning("observation approve: dispatcher nudge failed: %s", exc)
    return task


def observation_payload_from_entry(item: dict) -> dict[str, Any]:
    """Return normalized observation sub-object (may be empty)."""
    raw = item.get("observation")
    if not isinstance(raw, dict):
        return {}
    tier = str(raw.get("suggested_tier") or TIER_IN_THREAD).strip().lower()
    if tier not in _VALID_SUGGESTED_TIERS:
        tier = TIER_IN_THREAD
    out: dict[str, Any] = {
        "noticed": str(raw.get("noticed") or "")[:800],
        "proposed_work": str(raw.get("proposed_work") or "")[:800],
        "suggested_tier": tier,
        "suggested_assignee": str(raw.get("suggested_assignee") or "general")[:MAX_SOURCE_LEN],
    }
    dk = raw.get("dedupe_key")
    if dk is not None:
        out["dedupe_key"] = str(dk)[:MAX_DEDUPE_KEY_LEN]
    return out


def approval_payload_from_entry(item: dict) -> dict[str, Any]:
    """Return normalized approval sub-object (defaults to pending)."""
    raw = item.get("approval")
    if not isinstance(raw, dict):
        return {"status": APPROVAL_PENDING, "approved_at": None, "kanban_task_id": None, "thread_id": None}
    status = str(raw.get("status") or APPROVAL_PENDING).strip().lower()
    if status not in _VALID_APPROVAL_STATUSES:
        status = APPROVAL_PENDING
    return {
        "status": status,
        "approved_at": raw.get("approved_at"),
        "kanban_task_id": raw.get("kanban_task_id"),
        "thread_id": raw.get("thread_id"),
    }


def find_pending_observation_by_dedupe(lines: list[str], dedupe_key: str) -> dict | None:
    """Return a pending observation with the same ``dedupe_key``, if any."""
    target = str(dedupe_key or "").strip()
    if not target:
        return None
    for line in lines:
        raw = (line or "").strip()
        if not raw:
            continue
        try:
            item = json.loads(raw)
        except Exception:
            continue
        if not isinstance(item, dict) or item.get("processed"):
            continue
        if str(item.get("kind") or "") != KIND_OBSERVATION:
            continue
        if approval_payload_from_entry(item).get("status") != APPROVAL_PENDING:
            continue
        obs = observation_payload_from_entry(item)
        row_key = str(obs.get("dedupe_key") or item.get("dedupe_key") or "").strip()
        if row_key == target:
            return item
    return None


def find_unprocessed_entry_by_id(lines: list[str], item_id: str) -> dict | None:
    """Return the first unprocessed queue row matching ``item_id``, or None."""
    target = str(item_id or "").strip()
    if not target:
        return None
    for line in lines:
        raw = (line or "").strip()
        if not raw:
            continue
        try:
            item = json.loads(raw)
        except Exception:
            continue
        if not isinstance(item, dict) or item.get("processed"):
            continue
        if proactive_entry_id(item) == target:
            return item
    return None


def update_entry_by_id_in_lines(
    lines: list[str],
    item_id: str,
    mutator,
    *,
    require_unprocessed: bool = False,
) -> tuple[list[str], bool]:
    """Apply ``mutator(item_dict) -> item_dict`` to the matching row. Returns (new_lines, updated).

    ``ValueError`` from ``mutator`` is re-raised so callers can map validation failures to HTTP codes.
    When *require_unprocessed* is True, processed rows are skipped (``updated`` stays False).
    """
    target = str(item_id or "").strip()
    if not target:
        return lines, False
    updated = False
    new_lines: list[str] = []
    for line in lines:
        raw = (line or "").strip()
        if not raw:
            continue
        try:
            item = json.loads(raw)
        except Exception:
            new_lines.append(raw)
            continue
        if not isinstance(item, dict):
            new_lines.append(raw)
            continue
        if proactive_entry_id(item) == target:
            if require_unprocessed and item.get("processed"):
                new_lines.append(json.dumps(item, ensure_ascii=True))
                continue
            try:
                item = mutator(dict(item))
            except ValueError:
                raise
            except Exception:
                new_lines.append(json.dumps(item, ensure_ascii=True))
                continue
            updated = True
        new_lines.append(json.dumps(item, ensure_ascii=True))
    return new_lines, updated


# ---------------------------------------------------------------------------
# Read helpers (for backend routes and pollers)
# ---------------------------------------------------------------------------

def is_entry_expired(item: dict) -> bool:
    """Check if queue entry has passed its expires_at timestamp."""
    exp = str(item.get("expires_at") or "").strip()
    if not exp:
        return False
    try:
        exp_dt = datetime.fromisoformat(exp)
        if exp_dt.tzinfo is None:
            exp_dt = exp_dt.replace(tzinfo=timezone.utc)
        return datetime.now(timezone.utc) > exp_dt
    except Exception:
        return False


def pending_entries_from_lines(
    lines: list[str],
    *,
    max_entries: int = MAX_PENDING_ENTRIES,
    content_max_chars: int = MAX_CONTENT_LEN,
    include_notified: bool = False,
) -> list[dict]:
    """Parse JSONL lines into pending entry dicts for the API, sorted by priority then ts.

    When *include_notified* is True, items that have been shown in the Activity
    feed (``notified=True``) but not yet explicitly handled (``processed=False``)
    are included.  This powers the inbox badge count.
    """
    entries: list[dict] = []
    for line in lines:
        raw = (line or "").strip()
        if not raw:
            continue
        try:
            item = json.loads(raw)
        except Exception:
            continue
        if not isinstance(item, dict):
            continue
        if item.get("processed"):
            continue
        if item.get("notified") and not include_notified:
            continue
        if is_entry_expired(item):
            continue
        if str(item.get("kind") or "") == KIND_OBSERVATION:
            raw_appr = item.get("approval")
            if isinstance(raw_appr, dict):
                raw_status = str(raw_appr.get("status") or "").strip().lower()
                if raw_status and raw_status not in _VALID_APPROVAL_STATUSES:
                    continue
            appr = approval_payload_from_entry(item)
            if appr.get("status") != APPROVAL_PENDING:
                continue
        content = sanitize_system_event_text(item.get("content") or "", max_chars=content_max_chars)
        if not content:
            continue
        eid = proactive_entry_id(item)
        entry: dict[str, Any] = {
            "id": eid,
            "ts": str(item.get("ts") or ""),
            "content": content,
            "source": str(item.get("source") or "heartbeat"),
            "kind": str(item.get("kind") or "notification"),
        }
        for optional_key in ("priority", "batch_key", "dedupe_key", "drive_id", "thread_id", "expires_at"):
            val = item.get(optional_key)
            if val is not None:
                entry[optional_key] = str(val)
        if str(item.get("kind") or "") == KIND_OBSERVATION:
            obs = observation_payload_from_entry(item)
            if obs:
                entry["observation"] = obs
            appr = approval_payload_from_entry(item)
            entry["approval"] = appr
        entries.append(entry)

    entries.sort(key=lambda e: (
        _PRIORITY_SORT_KEY.get(e.get("priority", PRIORITY_NORMAL), 1),
        e.get("ts", ""),
    ))
    return entries[:max_entries]


def mark_processed_in_lines(
    lines: list[str],
    ids_set: set[str],
    *,
    allow_observations: bool = False,
) -> tuple[list[str], int, list[str]]:
    """Mark matching entries as processed, preserving ALL lines (including malformed).

    Returns (new_lines_for_rewrite, count_updated, blocked_observation_ids).
    Observations are skipped unless *allow_observations* is True (consent must use approve/dismiss).
    """
    updated = 0
    blocked: list[str] = []
    new_lines: list[str] = []
    for line in lines:
        raw = (line or "").strip()
        if not raw:
            continue
        try:
            item = json.loads(raw)
        except Exception:
            new_lines.append(raw)
            continue
        if not isinstance(item, dict):
            new_lines.append(raw)
            continue
        eid = proactive_entry_id(item)
        if eid in ids_set:
            if (
                not allow_observations
                and str(item.get("kind") or "") == KIND_OBSERVATION
            ):
                blocked.append(eid)
                new_lines.append(json.dumps(item, ensure_ascii=True))
                continue
            item["processed"] = True
            item["notified"] = True
            updated += 1
        new_lines.append(json.dumps(item, ensure_ascii=True))
    return new_lines, updated, blocked


def mark_notified_in_lines(
    lines: list[str],
    ids_set: set[str],
) -> tuple[list[str], int]:
    """Mark matching entries as notified (shown in Activity feed) without setting processed.

    Returns (new_lines_for_rewrite, count_updated).
    """
    updated = 0
    new_lines: list[str] = []
    for line in lines:
        raw = (line or "").strip()
        if not raw:
            continue
        try:
            item = json.loads(raw)
        except Exception:
            new_lines.append(raw)
            continue
        if not isinstance(item, dict):
            new_lines.append(raw)
            continue
        eid = proactive_entry_id(item)
        if eid in ids_set and not item.get("notified"):
            item["notified"] = True
            updated += 1
        new_lines.append(json.dumps(item, ensure_ascii=True))
    return new_lines, updated


def flock_rewrite(queue_path: str, new_lines: list[str]) -> None:
    """Rewrite the queue file with cross-process flock coordination."""
    content = "\n".join(new_lines) + ("\n" if new_lines else "")
    if _HAS_FLOCK:
        lock_path = queue_path + ".lock"
        fd = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o600)
        try:
            fcntl.flock(fd, fcntl.LOCK_EX)
            with open(queue_path, "w", encoding="utf-8") as f:
                f.write(content)
        finally:
            fcntl.flock(fd, fcntl.LOCK_UN)
            os.close(fd)
    else:
        with open(queue_path, "w", encoding="utf-8") as f:
            f.write(content)


# ---------------------------------------------------------------------------
# Proactive salience (migrated from utils/proactive_salience.py)
# ---------------------------------------------------------------------------

from persistence import MUGI_DIR, secure_atomic_write as _secure_atomic_write

_FINGERPRINT_PATH = os.path.join(MUGI_DIR, "proactive_salience_fingerprints.json")


def _load_fingerprints() -> dict[str, str]:
    try:
        if os.path.isfile(_FINGERPRINT_PATH):
            with open(_FINGERPRINT_PATH, "r", encoding="utf-8") as f:
                data = json.load(f)
            if isinstance(data, dict):
                return data
    except Exception as e:
        _log.debug("salience fingerprints load: %s", e)
    return {}


def _save_fingerprints(fp: dict[str, str]) -> None:
    try:
        _secure_atomic_write(_FINGERPRINT_PATH, json.dumps(fp, indent=2, ensure_ascii=True))
        os.chmod(_FINGERPRINT_PATH, 0o600)
    except Exception as e:
        _log.debug("salience fingerprints save: %s", e)


def _content_hash(text: str) -> str:
    return hashlib.sha256(text.strip().encode("utf-8")).hexdigest()[:24]


def should_emit_proactive_notification(
    key: str,
    content: str,
    *,
    force: bool = False,
) -> bool:
    """Return True if ``content`` for ``key`` represents new information worth notifying."""
    if force:
        fp = _load_fingerprints()
        fp[key] = _content_hash(content)
        _save_fingerprints(fp)
        return True

    h = _content_hash(content)
    fp = _load_fingerprints()
    prev = fp.get(key)
    if prev == h:
        _log.debug("salience suppressed: key=%s (same hash)", key)
        return False
    fp[key] = h
    _save_fingerprints(fp)
    return True


def clear_fingerprint(key: str) -> None:
    """Remove a single key from the fingerprint store."""
    fp = _load_fingerprints()
    if key in fp:
        del fp[key]
        _save_fingerprints(fp)
