"""AI-generated one-line summaries for Matrix threads."""

from __future__ import annotations

import asyncio
import hashlib
import re
from collections import defaultdict
from dataclasses import dataclass
from datetime import UTC, datetime
from html import escape
from typing import TYPE_CHECKING, Protocol, runtime_checkable

from agno.agent import Agent
from pydantic import BaseModel, Field

from mindroom import model_loading
from mindroom.ai_runtime import cached_agent_run
from mindroom.entity_resolution import current_internal_sender_ids, resolve_room_scoped_model_override
from mindroom.logging_config import get_logger
from mindroom.matrix.client_delivery import send_message_result
from mindroom.matrix.message_builder import build_message_content
from mindroom.model_defaults import (
    CLAUDE_PROVIDER_DEFAULT_SAMPLING_MODEL_SUFFIXES,
    GOOGLE_PROVIDER_DEFAULT_SAMPLING_MODEL_SUFFIXES,
)
from mindroom.model_instance_checks import isinstance_of_loaded
from mindroom.thread_tag_vocabulary import (
    claim_vocabulary_check,
    format_tag_vocabulary_with_counts,
    load_tag_vocabulary_snapshot,
    maybe_rebuild_tag_vocabulary,
)
from mindroom.thread_tags import (
    AUTOMATIC_THREAD_TAG_EXCLUSIONS,
    RESOLVED_THREAD_TAG,
    coerce_tag_name,
    get_thread_tags,
    set_thread_tags_if_empty,
)
from mindroom.timing import timed

if TYPE_CHECKING:
    from collections.abc import Collection, Sequence

    import nio

    from mindroom.config.main import Config
    from mindroom.constants import RuntimePaths
    from mindroom.matrix.client_visible_messages import ResolvedVisibleMessage
    from mindroom.matrix.conversation_cache import ConversationCacheProtocol

logger = get_logger(__name__)
_VERTEXAI_CLAUDE_CLASS = ("agno.models.vertexai.claude", "Claude")
_GOOGLE_GEMINI_CLASS = ("mindroom.google_gemini", "MindRoomGoogleGemini")
THREAD_SUMMARY_MAX_LENGTH = 300
_MAX_INITIAL_TAGS = 3
_MARKDOWN_LINK_RE = re.compile(r"!\[([^\]]*)\]\([^)]+\)|\[([^\]]+)\]\([^)]+\)")
_MARKDOWN_CODE_BLOCK_RE = re.compile(r"```(?:[^\n`]*)\n?(.*?)```", re.DOTALL)
_MARKDOWN_DOUBLE_EMPHASIS_RE = re.compile(r"(\*\*|__)(.*?)\1", re.DOTALL)
_MARKDOWN_SINGLE_ASTERISK_RE = re.compile(r"(?<!\*)\*(?!\*)(.+?)(?<!\*)\*(?!\*)", re.DOTALL)
_MARKDOWN_STRIKETHROUGH_RE = re.compile(r"~~(.*?)~~", re.DOTALL)
_MARKDOWN_INLINE_CODE_RE = re.compile(r"`([^`]*)`")
_MARKDOWN_HEADING_RE = re.compile(r"(?m)^\s{0,3}#{1,6}\s+")
_MARKDOWN_BLOCKQUOTE_RE = re.compile(r"(?m)^\s{0,3}>\s?")
_MARKDOWN_LIST_ITEM_RE = re.compile(r"(?m)^\s*(?:[-+*]|\d+\.)\s+")
_PREQUEUE_CONCURRENCY_MARGIN = 2

# In-memory tracking of last summarized message count per thread.
# Key: "{room_id}:{thread_id}", value: message count at last summary.
_last_summary_counts: dict[str, int] = {}
_thread_locks: dict[str, asyncio.Lock] = defaultdict(asyncio.Lock)


class ThreadSummaryWriteError(RuntimeError):
    """Raised when a manual thread summary cannot be written."""


@dataclass(frozen=True)
class _ThreadSummaryWriteResult:
    """Successful manual thread summary write details."""

    event_id: str
    message_count: int
    summary: str


class _ThreadSummary(BaseModel):
    """Structured thread summary response."""

    summary: str = Field(
        max_length=THREAD_SUMMARY_MAX_LENGTH,
        description="One-line summary of the thread conversation",
    )


class _ThreadEnrichment(_ThreadSummary):
    """Structured second-summary response with one-shot topic tags."""

    tags: list[str] = Field(
        min_length=1,
        max_length=_MAX_INITIAL_TAGS,
        description="1-3 durable topic tags, most relevant first",
    )


@runtime_checkable
class _SupportsTemperature(Protocol):
    """Protocol for model instances that accept a temperature override."""

    temperature: float | None


@runtime_checkable
class _IdentifiedModel(Protocol):
    """Protocol for model instances exposing their provider request ID."""

    id: str


def _summary_model_requires_provider_temperature(model: object) -> bool:
    """Return whether a summary model requires its provider sampling default."""
    return isinstance_of_loaded(model, _VERTEXAI_CLAUDE_CLASS) or (
        isinstance(model, _IdentifiedModel)
        and (
            model.id.casefold().endswith(CLAUDE_PROVIDER_DEFAULT_SAMPLING_MODEL_SUFFIXES)
            or (
                isinstance_of_loaded(model, _GOOGLE_GEMINI_CLASS)
                and model.id.casefold().endswith(GOOGLE_PROVIDER_DEFAULT_SAMPLING_MODEL_SUFFIXES)
            )
        )
    )


def _configure_summary_model_temperature(
    model: object,
    *,
    summary_temperature: float | None,
    model_name: str,
) -> None:
    """Prepare the summary model's temperature setting for one request."""
    if isinstance(model, _SupportsTemperature):
        if _summary_model_requires_provider_temperature(model):
            model.temperature = None
        else:
            model.temperature = summary_temperature
        return
    if summary_temperature is None:
        return

    model_class = type(model).__name__
    logger.warning(
        f"Thread summary model class {model_class} does not support a runtime temperature override; continuing with provider defaults",
        model_class=model_class,
        model_name=model_name,
    )


def normalize_thread_summary_text(raw_text: str) -> str:
    """Strip common markdown formatting and collapse the result to one plain-text line."""
    normalized = raw_text.strip()
    if not normalized:
        return ""

    normalized = _MARKDOWN_CODE_BLOCK_RE.sub(r"\1", normalized)
    normalized = _MARKDOWN_LINK_RE.sub(lambda match: match.group(1) or match.group(2) or "", normalized)
    normalized = _MARKDOWN_HEADING_RE.sub("", normalized)
    normalized = _MARKDOWN_BLOCKQUOTE_RE.sub("", normalized)
    normalized = _MARKDOWN_LIST_ITEM_RE.sub("", normalized)
    normalized = _MARKDOWN_DOUBLE_EMPHASIS_RE.sub(r"\2", normalized)
    normalized = _MARKDOWN_SINGLE_ASTERISK_RE.sub(r"\1", normalized)
    normalized = _MARKDOWN_STRIKETHROUGH_RE.sub(r"\1", normalized)
    normalized = _MARKDOWN_INLINE_CODE_RE.sub(r"\1", normalized)
    return " ".join(normalized.split())


def _thread_summary_cache_key(room_id: str, thread_id: str) -> str:
    """Return the in-memory cache key for one room/thread pair."""
    return f"{room_id}:{thread_id}"


def _thread_summary_lock(room_id: str, thread_id: str) -> asyncio.Lock:
    """Return the shared per-thread lock for summary writes."""
    return _thread_locks[_thread_summary_cache_key(room_id, thread_id)]


def update_last_summary_count(room_id: str, thread_id: str, message_count: int) -> None:
    """Record the latest summarized message count for one thread monotonically."""
    cache_key = _thread_summary_cache_key(room_id, thread_id)
    existing_count = _last_summary_counts.get(cache_key, 0)
    if message_count > existing_count:
        _last_summary_counts[cache_key] = message_count


def _next_threshold(
    last_summarized_count: int,
    *,
    first_threshold: int,
    subsequent_interval: int,
) -> int:
    """Return the next message count at which a summary should be generated."""
    if last_summarized_count <= 0:
        return first_threshold
    return last_summarized_count + subsequent_interval


def _thread_summary_metadata(
    message: ResolvedVisibleMessage,
    *,
    trusted_sender_ids: Collection[str],
) -> dict[str, object] | None:
    """Return trusted summary metadata emitted by a current runtime-owned sender."""
    if message.sender not in trusted_sender_ids:
        return None
    metadata = message.content.get("io.mindroom.thread_summary")
    return metadata if isinstance(metadata, dict) else None


def _is_thread_summary_message(
    message: ResolvedVisibleMessage,
    *,
    trusted_sender_ids: Collection[str],
) -> bool:
    """Return whether a visible thread message is a trusted summary notice."""
    return _thread_summary_metadata(message, trusted_sender_ids=trusted_sender_ids) is not None


def _count_non_summary_thread_messages(
    thread_history: Sequence[ResolvedVisibleMessage],
    *,
    trusted_sender_ids: Collection[str],
) -> int:
    """Count visible thread messages while excluding trusted summary notices."""
    return sum(
        1
        for message in thread_history
        if not _is_thread_summary_message(message, trusted_sender_ids=trusted_sender_ids)
    )


def thread_summary_message_count_hint(
    thread_history: Sequence[ResolvedVisibleMessage],
    *,
    trusted_sender_ids: Collection[str],
) -> int:
    """Return a lower-bound post-response thread size without refetching history."""
    return (
        _count_non_summary_thread_messages(
            thread_history,
            trusted_sender_ids=trusted_sender_ids,
        )
        + 1
    )


def _next_thread_summary_threshold(
    room_id: str,
    thread_id: str,
    config: Config,
) -> int:
    """Return the next summary threshold using the current in-memory baseline."""
    return _next_threshold(
        _last_summary_counts.get(_thread_summary_cache_key(room_id, thread_id), 0),
        first_threshold=config.defaults.thread_summary_first_threshold,
        subsequent_interval=config.defaults.thread_summary_subsequent_interval,
    )


def should_queue_thread_summary(
    room_id: str,
    thread_id: str,
    config: Config,
    runtime_paths: RuntimePaths,
    *,
    message_count_hint: int | None,
) -> bool:
    """Return whether summary generation or room-vocabulary upkeep is due."""
    if message_count_hint is None:
        return True
    threshold = _next_thread_summary_threshold(room_id, thread_id, config)
    return message_count_hint >= threshold - _PREQUEUE_CONCURRENCY_MARGIN or claim_vocabulary_check(
        room_id,
        config,
        runtime_paths,
        now=datetime.now(UTC),
    )


async def _load_thread_history(
    conversation_cache: ConversationCacheProtocol,
    room_id: str,
    thread_id: str,
) -> list[ResolvedVisibleMessage]:
    """Load fresh authoritative history without inherited turn memoization."""
    return list(
        await conversation_cache.get_strict_thread_history(
            room_id,
            thread_id,
            caller_label="thread_summary_background",
        ),
    )


def _recover_last_summary_count(
    thread_history: Sequence[ResolvedVisibleMessage],
    *,
    trusted_sender_ids: Collection[str],
) -> int:
    """Recover the highest durable summary count from authoritative thread history."""
    best_count = 0
    for message in thread_history:
        metadata = _thread_summary_metadata(message, trusted_sender_ids=trusted_sender_ids)
        if metadata is None:
            continue
        count = metadata.get("message_count")
        if not isinstance(count, int) or isinstance(count, bool):
            continue
        best_count = max(best_count, count)
    return best_count


def _parse_summary_generated_at(metadata: dict[str, object]) -> datetime | None:
    """Return the ISO-8601 ``generated_at`` recorded on one summary, when usable."""
    raw = metadata.get("generated_at")
    if not isinstance(raw, str):
        return None
    try:
        parsed = datetime.fromisoformat(raw)
    except ValueError:
        return None
    return parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=UTC)


def _recover_pin_state(
    thread_history: Sequence[ResolvedVisibleMessage],
    *,
    trusted_sender_ids: Collection[str],
) -> bool:
    """Return the newest durable pin decision recorded in summary metadata.

    A later ``pinned: false`` summary releases a thread that an earlier summary
    pinned, so the newest decision has to win. Summaries that omit the key state
    no intent and are ignored entirely.

    Ordering comes from ``generated_at`` rather than the position of the message
    in the history, because history position does not always reflect Matrix
    order: ``_sort_thread_items_root_first`` breaks equal ``origin_server_ts``
    ties with backward-scan input order, which is newest-first. ``generated_at``
    is also what the client uses to choose the summary it displays, so the pin
    decision and the visible title are resolved by the same clock.
    """
    newest_decision: tuple[datetime, int] | None = None
    pinned = False
    for position, message in enumerate(thread_history):
        metadata = _thread_summary_metadata(message, trusted_sender_ids=trusted_sender_ids)
        if metadata is None:
            continue
        recorded = metadata.get("pinned")
        if not isinstance(recorded, bool):
            continue
        generated_at = _parse_summary_generated_at(metadata)
        if generated_at is None:
            continue
        decision = (generated_at, position)
        if newest_decision is None or decision > newest_decision:
            newest_decision = decision
            pinned = recorded
    return pinned


def _recover_initial_enrichment_complete(
    thread_history: Sequence[ResolvedVisibleMessage],
    *,
    trusted_sender_ids: Collection[str],
) -> bool:
    """Return whether durable summary metadata records completed initial tags."""
    for message in thread_history:
        metadata = _thread_summary_metadata(message, trusted_sender_ids=trusted_sender_ids)
        if metadata is not None and metadata.get("initial_enrichment_complete") is True:
            return True
    return False


_MAX_MESSAGES_BEFORE_TRUNCATION = 50
_TRUNCATION_SAMPLE_SIZE = 3


def _build_conversation_text(
    thread_history: Sequence[ResolvedVisibleMessage],
    *,
    trusted_sender_ids: Collection[str],
) -> str:
    """Build conversation text from thread history.

    Prior thread summary notices (``io.mindroom.thread_summary``) are excluded
    so they don't pollute the conversation.

    For threads exceeding ``_MAX_MESSAGES_BEFORE_TRUNCATION`` messages, samples
    the first and last few messages with an omission note in between.
    """
    lines: list[str] = []
    for msg in thread_history:
        if _is_thread_summary_message(msg, trusted_sender_ids=trusted_sender_ids):
            continue
        sender = msg.sender or "unknown"
        body = msg.body or ""
        if body:
            lines.append(f"{sender}: {body}")

    if len(lines) > _MAX_MESSAGES_BEFORE_TRUNCATION:
        n = _TRUNCATION_SAMPLE_SIZE
        omitted = len(lines) - 2 * n
        lines = [*lines[:n], f"[... {omitted} messages omitted ...]", *lines[-n:]]

    return "\n".join(lines)


def _resolve_thread_summary_model_name(
    config: Config,
    runtime_paths: RuntimePaths,
    room_id: str | None,
    *,
    entity_name: str | None = None,
) -> str:
    """Return the model name for automatic thread summaries in one room.

    Precedence: room-scoped override (alias or raw room ID) > responding
    entity's name as a ``room_thread_summary_models`` key (covers ad-hoc
    rooms with no managed alias) > ``defaults.thread_summary_model``.
    """
    if override := resolve_room_scoped_model_override(
        config.room_thread_summary_models,
        room_id,
        runtime_paths,
        allow_raw_room_id=True,
    ):
        return override
    if entity_name and entity_name in config.room_thread_summary_models:
        return config.room_thread_summary_models[entity_name]
    return config.defaults.thread_summary_model or "default"


async def _generate_summary(
    thread_history: Sequence[ResolvedVisibleMessage],
    config: Config,
    runtime_paths: RuntimePaths,
    *,
    model_name: str | None = None,
    tag_vocabulary: str | None = None,
    trusted_sender_ids: Collection[str] | None = None,
) -> str | _ThreadEnrichment | None:
    """Generate a summary and optional one-shot tags via one LLM run."""
    resolved_model_name = model_name or config.defaults.thread_summary_model or "default"
    model = model_loading.get_model_instance(config, runtime_paths, resolved_model_name)
    _configure_summary_model_temperature(
        model,
        summary_temperature=config.defaults.thread_summary_temperature,
        model_name=resolved_model_name,
    )

    if trusted_sender_ids is None:
        trusted_sender_ids = current_internal_sender_ids(config, runtime_paths)
    conversation = escape(
        _build_conversation_text(
            thread_history,
            trusted_sender_ids=trusted_sender_ids,
        ),
    )
    prompt = config.render_prompt(
        "THREAD_SUMMARY_USER_PROMPT_TEMPLATE",
        conversation=conversation,
        tag_vocabulary=(
            escape(tag_vocabulary)
            if tag_vocabulary is not None
            else "(tags are not requested for this summary refresh)"
        ),
    )
    session_hash = hashlib.sha256(prompt.encode()).hexdigest()[:16]
    agent = Agent(
        name="ThreadSummarizer",
        instructions=config.get_prompt("THREAD_SUMMARY_INSTRUCTIONS").splitlines(),
        model=model,
        output_schema=_ThreadEnrichment if tag_vocabulary is not None else _ThreadSummary,
        telemetry=False,
    )
    response = await cached_agent_run(
        agent=agent,
        run_input=prompt,
        session_id=f"thread_summary_{session_hash}",
    )
    content = response.content
    if tag_vocabulary is not None:
        if not isinstance(content, _ThreadEnrichment):
            return None
        normalized_tags: list[str] = []
        for raw_tag in content.tags:
            normalized_tag = coerce_tag_name(raw_tag)
            if (
                normalized_tag is not None
                and normalized_tag not in AUTOMATIC_THREAD_TAG_EXCLUSIONS
                and normalized_tag not in normalized_tags
            ):
                normalized_tags.append(normalized_tag)
        if not normalized_tags:
            return content.summary
        return _ThreadEnrichment(
            summary=content.summary,
            tags=normalized_tags[:_MAX_INITIAL_TAGS],
        )
    if isinstance(content, _ThreadEnrichment):
        return None
    if isinstance(content, _ThreadSummary):
        return content.summary
    return None


async def _thread_is_resolved(
    client: nio.AsyncClient,
    room_id: str,
    thread_id: str,
) -> bool:
    """Return whether a thread carries the resolved lifecycle tag.

    Fails open, matching ``_refresh_tag_vocabulary`` and the history load in this
    module: this is a best-effort background read, and the cost of being wrong is
    one summary on a resolved thread. Letting a transport error escape instead
    would abort the pass before it advances the baseline, so a persistently
    failing room-state read would re-run the full state and history reads on
    every turn.
    """
    try:
        tags_state = await get_thread_tags(client, room_id, thread_id)
    except Exception as exc:
        # Keep the type and traceback: this catch is broad, and intermittent
        # Matrix failures are otherwise indistinguishable from each other.
        logger.exception(
            "Thread tag read failed; continuing with automatic summary",
            room_id=room_id,
            thread_id=thread_id,
            error_type=type(exc).__name__,
        )
        return False
    return tags_state is not None and RESOLVED_THREAD_TAG in tags_state.tags


async def _pinned_since_generation_started(
    conversation_cache: ConversationCacheProtocol,
    room_id: str,
    thread_id: str,
    *,
    trusted_sender_ids: Collection[str],
) -> bool:
    """Return whether a pin landed while this pass was generating a summary.

    The gate before generation cannot cover the generation window itself. The
    per-thread lock is process-local, so another runtime can write a manual pin
    while this process is waiting on the model, and the finished automatic
    summary would then land on top of a title the user just fixed.

    Reads from source rather than through ``get_strict_thread_history``. That
    read is strict about staleness but still accepts a valid local cache hit, so
    it cannot observe a pin another runtime just wrote — which is the only case
    this guard exists for. Costs one homeserver read per generated summary, so
    once per interval rather than per turn.

    Fails open, like the other background reads here: if the re-read fails the
    pass delivers, which is the same exposure the pre-generation gate already
    has. An automatic summary carries no ``pinned`` key, so the worst case is a
    single superseded title and the next pass bails at the gate.
    """
    try:
        thread_history = list(
            await conversation_cache.refresh_strict_thread_history_from_source(
                room_id,
                thread_id,
                caller_label="thread_summary_pin_recheck",
            ),
        )
    except Exception:
        logger.exception(
            "Pin re-check before summary delivery failed; delivering anyway",
            room_id=room_id,
            thread_id=thread_id,
        )
        return False
    return _recover_pin_state(thread_history, trusted_sender_ids=trusted_sender_ids)


@timed("maybe_generate_thread_summary")
async def _timed_generate_summary(
    thread_history: Sequence[ResolvedVisibleMessage],
    config: Config,
    runtime_paths: RuntimePaths,
    *,
    model_name: str | None = None,
    tag_vocabulary: str | None = None,
    trusted_sender_ids: Collection[str],
) -> str | _ThreadEnrichment | None:
    """Run the summary generation attempt with timing instrumentation."""
    return await _generate_summary(
        thread_history,
        config,
        runtime_paths,
        model_name=model_name,
        tag_vocabulary=tag_vocabulary,
        trusted_sender_ids=trusted_sender_ids,
    )


async def _apply_initial_tags(
    client: nio.AsyncClient,
    room_id: str,
    thread_id: str,
    tags: Sequence[str],
) -> bool:
    """Apply generated tags only when no manual or concurrent tags exist."""
    set_by = client.user_id
    if not tags or not set_by:
        return False
    try:
        result = await set_thread_tags_if_empty(
            client,
            room_id,
            thread_id,
            tags,
            set_by=set_by,
        )
    except Exception:
        logger.exception(
            "Failed to apply automatic thread tags",
            room_id=room_id,
            thread_id=thread_id,
        )
        return False
    if result.skipped_due_to_prior_mutation:
        logger.info(
            "Skipping automatic tags because another tag mutation completed first",
            room_id=room_id,
            thread_id=thread_id,
        )
        return True
    if result.had_existing_tags:
        logger.info(
            "Skipping automatic tags because the thread already has tags",
            room_id=room_id,
            thread_id=thread_id,
        )
        return True
    if result.failed_tags:
        logger.warning(
            "Failed to write some automatic thread tags",
            room_id=room_id,
            thread_id=thread_id,
            tags=result.failed_tags,
        )
    if result.applied_tags:
        logger.info(
            "Automatically tagged thread",
            room_id=room_id,
            thread_id=thread_id,
            tags=result.applied_tags,
        )
    return bool(result.applied_tags)


async def _refresh_tag_vocabulary(
    client: nio.AsyncClient,
    room_id: str,
    config: Config,
    runtime_paths: RuntimePaths,
) -> str | None:
    """Refresh vocabulary and return text when the check already loaded a snapshot."""
    try:
        snapshot = await maybe_rebuild_tag_vocabulary(
            client,
            room_id,
            config,
            runtime_paths,
            now=datetime.now(UTC),
        )
    except Exception:
        logger.exception("Tag vocabulary rebuild failed", room_id=room_id)
        return None
    if snapshot is None:
        return None
    return format_tag_vocabulary_with_counts(snapshot)


async def _deliver_generated_summary(
    client: nio.AsyncClient,
    room_id: str,
    thread_id: str,
    generated: str | _ThreadEnrichment,
    normalized_summary: str,
    message_count: int,
    model_name: str,
    conversation_cache: ConversationCacheProtocol,
    *,
    trusted_sender_ids: Collection[str],
) -> None:
    """Apply initial tags, then independently deliver the generated summary.

    Re-reads pin state first. The gate before generation cannot cover the
    generation window, and the per-thread lock is process-local, so another
    runtime can write a manual pin while this process waits on the model. That
    check belongs here rather than in the caller because it is a delivery-time
    question: a superseded summary should apply neither its tags nor its title.
    """
    if await _pinned_since_generation_started(
        conversation_cache,
        room_id,
        thread_id,
        trusted_sender_ids=trusted_sender_ids,
    ):
        logger.info(
            "Discarding an automatic thread summary that a pin superseded during generation",
            room_id=room_id,
            thread_id=thread_id,
            message_count=message_count,
        )
        return

    initial_enrichment_complete: bool | None = None
    if isinstance(generated, _ThreadEnrichment):
        initial_enrichment_complete = await _apply_initial_tags(
            client,
            room_id,
            thread_id,
            generated.tags,
        )

    try:
        await send_thread_summary_event(
            client,
            room_id,
            thread_id,
            normalized_summary,
            message_count,
            model_name,
            conversation_cache,
            initial_enrichment_complete=initial_enrichment_complete,
        )
    except Exception:
        logger.exception("Thread summary send failed", room_id=room_id, thread_id=thread_id)


async def send_thread_summary_event(
    client: nio.AsyncClient,
    room_id: str,
    thread_id: str,
    summary: str,
    message_count: int,
    model_name: str,
    conversation_cache: ConversationCacheProtocol,
    *,
    initial_enrichment_complete: bool | None = None,
    pinned: bool | None = None,
    known_latest_thread_event_id: str | None = None,
) -> str | None:
    """Send a thread summary as a standard Matrix notice event.

    ``known_latest_thread_event_id`` lets a caller that already knows the newest
    event in the thread (for example the creator of a brand-new thread) supply it
    directly, skipping the history read that would otherwise scan the homeserver
    for a thread with no cache snapshot yet.

    ``pinned`` records an explicit decision about whether this summary should
    stop automatic re-summarization. It defaults to ``None``, which omits the
    key entirely and leaves any existing pin decision untouched. Only callers
    acting on a user's intent should pass a boolean: writers that summarize as a
    side effect, such as the subagent spawn path, must not disturb pin state on
    a thread the user already pinned.
    """
    normalized_summary = normalize_thread_summary_text(summary)
    if not normalized_summary:
        logger.warning(
            "Refusing to send empty normalized thread summary",
            room_id=room_id,
            thread_id=thread_id,
            message_count=message_count,
        )
        return None

    truncated_summary = (
        normalized_summary[: THREAD_SUMMARY_MAX_LENGTH - 3] + "..."
        if len(normalized_summary) > THREAD_SUMMARY_MAX_LENGTH
        else normalized_summary
    )
    latest_thread_event_id = known_latest_thread_event_id
    if latest_thread_event_id is None:
        try:
            latest_thread_event_id = await conversation_cache.get_latest_thread_event_id_if_needed(
                room_id,
                thread_id,
                caller_label="thread_summary_send",
            )
        except Exception as exc:
            logger.warning(
                "Falling back to thread root for summary send after latest-event lookup failure",
                room_id=room_id,
                thread_id=thread_id,
                error=str(exc),
            )
            latest_thread_event_id = None
    summary_metadata: dict[str, object] = {
        "version": 1,
        "summary": truncated_summary,
        "message_count": message_count,
        "generated_at": datetime.now(UTC).isoformat(),
        "model": model_name,
    }
    if initial_enrichment_complete is not None:
        summary_metadata["initial_enrichment_complete"] = initial_enrichment_complete
    if pinned is not None:
        summary_metadata["pinned"] = pinned

    content = build_message_content(
        truncated_summary,
        thread_event_id=thread_id,
        latest_thread_event_id=latest_thread_event_id or thread_id,
        extra_content={
            "msgtype": "m.notice",
            "io.mindroom.thread_summary": summary_metadata,
        },
    )
    delivered = await send_message_result(client, room_id, content)
    if delivered is not None:
        conversation_cache.notify_outbound_message(
            room_id,
            delivered.event_id,
            delivered.content_sent,
        )
        logger.info(
            "Sent thread summary",
            room_id=room_id,
            thread_id=thread_id,
            message_count=message_count,
        )
        return delivered.event_id
    logger.warning("Failed to send thread summary", room_id=room_id, thread_id=thread_id)
    return None


async def set_manual_thread_summary(
    client: nio.AsyncClient,
    room_id: str,
    thread_id: str,
    summary: str,
    *,
    config: Config,
    runtime_paths: RuntimePaths,
    conversation_cache: ConversationCacheProtocol,
    pin: bool = True,
) -> _ThreadSummaryWriteResult:
    """Write one validated manual summary for a canonical thread root.

    Pins the thread by default so the written title survives automatic
    re-summarization. ``pin=False`` writes the summary and releases any pin an
    earlier manual write established.
    """
    if not isinstance(summary, str) or not summary.strip():
        msg = "summary must be a non-empty string."
        raise ThreadSummaryWriteError(msg)

    normalized_summary = normalize_thread_summary_text(summary)
    if not normalized_summary:
        msg = "summary must be a non-empty string."
        raise ThreadSummaryWriteError(msg)
    if len(normalized_summary) > THREAD_SUMMARY_MAX_LENGTH:
        msg = f"summary must be {THREAD_SUMMARY_MAX_LENGTH} characters or fewer after whitespace normalization."
        raise ThreadSummaryWriteError(msg)

    async with _thread_summary_lock(room_id, thread_id):
        try:
            thread_history = await _load_thread_history(
                conversation_cache,
                room_id,
                thread_id,
            )
        except Exception as exc:
            msg = "Failed to fetch thread history for the target thread."
            raise ThreadSummaryWriteError(msg) from exc

        message_count = _count_non_summary_thread_messages(
            thread_history,
            trusted_sender_ids=current_internal_sender_ids(config, runtime_paths),
        )
        try:
            event_id = await send_thread_summary_event(
                client,
                room_id,
                thread_id,
                normalized_summary,
                message_count,
                "manual",
                conversation_cache,
                pinned=pin,
            )
        except Exception as exc:
            msg = "Failed to send thread summary event."
            raise ThreadSummaryWriteError(msg) from exc
        if event_id is None:
            msg = "Failed to send thread summary event."
            raise ThreadSummaryWriteError(msg)

        update_last_summary_count(room_id, thread_id, message_count)
        return _ThreadSummaryWriteResult(
            event_id=event_id,
            message_count=message_count,
            summary=normalized_summary,
        )


async def maybe_generate_thread_summary(  # noqa: PLR0911
    client: nio.AsyncClient,
    room_id: str,
    thread_id: str,
    config: Config,
    runtime_paths: RuntimePaths,
    *,
    conversation_cache: ConversationCacheProtocol,
    entity_name: str | None = None,
) -> None:
    """Generate an early summary, then one-shot initial tags on its first refresh."""
    refreshed_tag_vocabulary = await _refresh_tag_vocabulary(client, room_id, config, runtime_paths)
    async with _thread_summary_lock(room_id, thread_id):
        # This background task inherits the response turn's ContextVars, so it
        # must bypass per-turn memoization to observe the delivered response.
        try:
            thread_history = await _load_thread_history(conversation_cache, room_id, thread_id)
        except Exception:
            logger.exception(
                "Authoritative thread history load failed",
                room_id=room_id,
                thread_id=thread_id,
            )
            return
        trusted_sender_ids = current_internal_sender_ids(config, runtime_paths)
        recovered_summary_count = _recover_last_summary_count(
            thread_history,
            trusted_sender_ids=trusted_sender_ids,
        )
        if recovered_summary_count > 0:
            update_last_summary_count(room_id, thread_id, recovered_summary_count)

        threshold = _next_thread_summary_threshold(room_id, thread_id, config)
        message_count = _count_non_summary_thread_messages(
            thread_history,
            trusted_sender_ids=trusted_sender_ids,
        )
        if _recover_pin_state(thread_history, trusted_sender_ids=trusted_sender_ids):
            logger.debug(
                "Skipping automatic thread summary for a pinned thread",
                room_id=room_id,
                thread_id=thread_id,
                message_count=message_count,
            )
            # Advance the baseline so the cheap pre-check stops firing, and
            # therefore stops spawning a history-loading pass, on every turn.
            update_last_summary_count(room_id, thread_id, message_count)
            return
        if message_count < threshold:
            return
        if await _thread_is_resolved(client, room_id, thread_id):
            logger.debug(
                "Skipping automatic thread summary for a resolved thread",
                room_id=room_id,
                thread_id=thread_id,
                message_count=message_count,
            )
            update_last_summary_count(room_id, thread_id, message_count)
            return

        initial_enrichment = recovered_summary_count > 0 and not _recover_initial_enrichment_complete(
            thread_history,
            trusted_sender_ids=trusted_sender_ids,
        )
        tag_vocabulary = None
        if initial_enrichment:
            tag_vocabulary = refreshed_tag_vocabulary or format_tag_vocabulary_with_counts(
                load_tag_vocabulary_snapshot(runtime_paths, room_id),
            )
        try:
            model_name = _resolve_thread_summary_model_name(
                config,
                runtime_paths,
                room_id,
                entity_name=entity_name,
            )
            generated = await _timed_generate_summary(
                thread_history,
                config,
                runtime_paths,
                model_name=model_name,
                tag_vocabulary=tag_vocabulary,
                trusted_sender_ids=trusted_sender_ids,
            )
        except Exception:
            logger.exception("Thread summary generation failed", room_id=room_id, thread_id=thread_id)
            # Record current count to prevent retry storms until next threshold
            update_last_summary_count(room_id, thread_id, message_count)
            return

        if generated is None:
            logger.warning("Thread summary generation returned None", room_id=room_id, thread_id=thread_id)
            # Record current count to prevent retry storms until next threshold
            update_last_summary_count(room_id, thread_id, message_count)
            return

        summary = generated.summary if isinstance(generated, _ThreadEnrichment) else generated
        normalized_summary = normalize_thread_summary_text(summary)
        if not normalized_summary:
            logger.warning(
                "Thread summary generation returned no plain-text content",
                room_id=room_id,
                thread_id=thread_id,
            )
            update_last_summary_count(room_id, thread_id, message_count)
            return

        await _deliver_generated_summary(
            client,
            room_id,
            thread_id,
            generated,
            normalized_summary,
            message_count,
            model_name,
            conversation_cache,
            trusted_sender_ids=trusted_sender_ids,
        )
        # Record after the delivery attempt so cancellation cannot leave a
        # partially delivered initial enrichment marked complete.
        update_last_summary_count(room_id, thread_id, message_count)
