"""Context assembly service — builds ranked, token-bounded context bundles.

Strategy:
- Score every memory and episode by kind priority + recency + relevance.
- When embeddings are available, use semantic similarity for relevance.
- Otherwise, fall back to word-overlap relevance.
- Assemble text incrementally, highest-scored items first.
- Stop adding items when the token budget is exhausted.
- The task header is always included (reserved budget).
- Provenance tracks exactly which items made it into the bundle.
"""

from __future__ import annotations

import uuid
from datetime import datetime, timezone
from typing import Any

import structlog
import tiktoken
from sqlalchemy.ext.asyncio import AsyncSession

from server.core.config import settings
from server.db import repositories as repo
from server.schemas.responses import (
    ContextBundleResponse,
    EpisodeResponse,
    MemoryResponse,
    SessionInfo,
)
from server.services import policy as policy_service
from server.services import receipts as receipts_service
from server.services.compilers.heuristic import extract_payload_text
from server.services.structured import is_structured_episode
from server.services.embeddings import get_provider as get_embedding_provider
from server.services.embeddings.query_cache import cached_embed_query
from server.services.tokenization import EDGE_PUNCT, tokenize

logger = structlog.stdlib.get_logger()

# ---------------------------------------------------------------------------
# Scoring constants
# ---------------------------------------------------------------------------

_KIND_PRIORITY: dict[str, float] = {
    "profile_fact": 10.0,
    "procedure": 8.0,
    "episode_summary": 5.0,
}
_EPISODE_PRIORITY = 3.0
_RECENCY_MAX = 5.0
_RELEVANCE_MAX = 5.0
_SEMANTIC_MAX = 8.0  # Semantic relevance has higher signal than word-overlap
_TEMPORAL_VALID_BONUS = (
    3.0  # Bonus for memories currently valid (valid_to is None or in the future)
)
_TEMPORAL_EXPIRED_PENALTY = -4.0  # Penalty for memories whose valid_to is in the past
_SESSION_BOOST = 6.0  # Bonus for episodes belonging to the current active session
_RESOLVED_SESSION_PENALTY = -5.0  # Penalty for episodes in already-resolved sessions
_BREADCRUMB_MAX = 3.0  # Bonus for memories whose source-doc heading/breadcrumb
# matches the query (docs-grounded subjects only — see _breadcrumb_overlap_bonus).
# Capped well below KIND_PRIORITY (5–10) and SEMANTIC_MAX (8) so it nudges, not dominates.
_LEXICAL_BONUS_MAX = 4.0  # Additive bonus when query terms appear verbatim in
# memory content, applied on top of the semantic score. Without this, a narrow
# factual query like "how do I install with npm" can match the right procedure
# at semantic rank 1 yet sit at /v1/context rank ~127 because cosine spread for
# the candidate pool is ~3 points while KIND_PRIORITY (procedure 8 vs
# profile_fact 10) plus full recency (5) easily exceeds that spread. The
# lexical bonus contributes the missing tiebreaker: when both semantic AND
# literal keyword evidence agree, the memory clears the kind/recency noise
# floor. Capped at 4.0 so it can't override a clearly dominant kind signal —
# only break ties between similarly-ranked candidates.

# Task-relevance gate (issue #116). The ranker always returns *something*
# (kind priority + recency keep memories ranked even when nothing matches the
# task), so an off-topic task — "What is the airline ticket price?" against a
# weather subject — silently renders unrelated facts as if they were relevant.
# We don't drop those facts (recall is still useful and the benchmark relies
# on it), but when NOTHING clears a real task-relevance signal we prepend an
# explicit caveat so the caller knows the context isn't task-specific.
#
# A candidate counts as task-relevant if EITHER:
#   - semantic cosine similarity ≥ _SEMANTIC_RELEVANCE_FLOOR, or
#   - it shares ≥1 meaningful (non-stopword) token with the task.
# semantic_scores[id] = (1 - cosine_distance) * _SEMANTIC_MAX, so the cosine
# similarity is semantic_scores[id] / _SEMANTIC_MAX. 0.25 is deliberately
# conservative: unrelated sentence-embedding pairs typically sit ≤0.15 while
# topically-related ones clear ~0.35+, so this only fires for clearly
# off-topic tasks and never on the (always on-topic) benchmark queries.
_SEMANTIC_RELEVANCE_FLOOR = 0.25
_NO_TASK_CONTEXT_NOTE = (
    "_No stored context is directly associated with this task. "
    "The items below are this subject's general memory, not task-specific._"
)

# Support-agent-specific scoring signals
_OPEN_ISSUE_BOOST = 4.0  # Bonus for episodes in sessions with open/unresolved issues
_ACTION_STEP_BOOST = 2.0  # Bonus for agent/assistant/tool episodes (what was tried)
_URGENCY_BOOST = 2.0  # Bonus for episodes containing urgency keywords
_IDLE_CHATTER_PENALTY = -2.0  # Penalty for very short, low-signal content

# Repeat-issue detection
_REPEAT_ISSUE_BOOST = 4.0  # Boost for resolved-session episodes when issue recurs
_REPEAT_RESOLVED_BOOST = 6.0  # Extra boost when prior session has a useful resolution summary
_REPEAT_OVERLAP_THRESHOLD = 0.3  # Minimum keyword overlap ratio to trigger repeat detection

# Urgency keywords (case-insensitive match)
_URGENCY_KEYWORDS = frozenset(
    [
        "urgent",
        "critical",
        "blocked",
        "deadline",
        "asap",
        "emergency",
        "breaking",
        "outage",
        "downtime",
        "compliance",
        "sla",
        "escalat",
        "immediately",
        "production down",
        "p0",
        "p1",
        "severity",
    ]
)


# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------


async def assemble_context(
    session: AsyncSession,
    subject_id: str,
    task: str,
    max_tokens: int | None = None,
    tenant_id: str | None = None,
    session_id: str | None = None,
    *,
    emit_receipt: bool | None = None,
    query_id: str | None = None,
    task_id: str | None = None,
    parent_receipt_id: str | None = None,
    caller_id: str | None = None,
    caller_type: str | None = None,
    # v0.9 (#159) — replay knobs. Both are private contract between
    # `services.replay` and this function; the public `/v1/context`
    # route never sets them. The bundle override forces evaluation
    # against a specific PolicyBundle (loaded from a receipt's
    # snapshot) instead of resolving the live `policy_bundles` row.
    # The mode override stamps the emitted receipt as
    # ``as_of_replay`` rather than ``retrieval``.
    _policy_bundle_override: "policy_service.PolicyBundle | None" = None,
    _use_policy_bundle_override: bool = False,
    _mode_override: str | None = None,
) -> ContextBundleResponse:
    """Build a deterministic, token-bounded, ranked context bundle.

    When emission is decided (see `server.services.receipts.decide_emission`),
    a state-assembly receipt is written and its ULID is attached to the
    response. Receipt write failures are non-fatal — the bundle is still
    returned and a structured log records the failure."""
    budget = max_tokens or settings.default_max_context_tokens
    enc = tiktoken.get_encoding(settings.tiktoken_model)
    # Captured up front so receipt's `as_of` records the wall-clock moment
    # the assembly resolved against, not the moment the receipt was
    # written (which can drift if downstream work is slow).
    as_of = datetime.now(timezone.utc)

    # -- Fetch candidates ---------------------------------------------------
    fact_rows = await repo.search_memories(
        session, subject_id, tenant_id=tenant_id, kind="profile_fact", limit=50
    )
    procedure_rows = await repo.search_memories(
        session, subject_id, tenant_id=tenant_id, kind="procedure", limit=20
    )
    summary_rows = await repo.search_memories(
        session, subject_id, tenant_id=tenant_id, kind="episode_summary", limit=30
    )
    # `newest_first` so a subject with >30 lifetime episodes contributes its
    # most-recent 30 to the candidate pool — recency boosts, the "Recent
    # interactions" rendering, and repeat-issue detection all assume recent
    # episodes. Without it the oldest 30 would be scored under a "recent"
    # heading.
    episode_rows = await repo.list_episodes_by_subject(
        session, subject_id, tenant_id=tenant_id, limit=30, newest_first=True
    )

    # -- Prepare semantic scoring -------------------------------------------
    # Only treat the provider as a semantic relevance signal if it actually
    # produces vectors with semantic meaning. The hash-based stub provider
    # produces deterministic-but-meaningless vectors; using its scores as
    # ranking input silently corrupts retrieval (verified against production
    # statewave-support-docs: the same garbage scores were dominating
    # KIND_PRIORITY + recency, producing repetitive citations across
    # unrelated queries). When the active provider is the stub (or any
    # provider that flags itself as not-really-semantic), we skip the
    # semantic path entirely and fall back to query-aware word-overlap.
    semantic_scores: dict[uuid.UUID, float] = {}
    provider = get_embedding_provider()
    use_semantic_provider = bool(
        provider and getattr(provider, "provides_semantic_similarity", True)
    )
    semantic_results: list[tuple[Any, float]] = []
    if use_semantic_provider:
        try:
            # Cross-machine query embedding cache: hits the Postgres-backed
            # query_embedding_cache before the provider's in-process L1, so
            # repeated demo queries asked across both Fly machines pay the
            # provider round-trip exactly once cluster-wide. Falls through
            # to the provider transparently on miss / DB error.
            from server.db.engine import get_session_factory

            task_embedding = await cached_embed_query(get_session_factory(), provider, task)
            # Top 100 memories by cosine distance — this set is the SEMANTIC
            # contribution to the candidate pool below, in addition to being
            # used as a score lookup during ranking.
            semantic_results = await repo.search_memories_by_embedding(
                session,
                subject_id,
                task_embedding,
                tenant_id=tenant_id,
                limit=100,
            )
            for row, distance in semantic_results:
                # Convert cosine distance [0, 2] to similarity score [0, SEMANTIC_MAX]
                # distance 0 = identical → max score, distance 2 = opposite → 0
                similarity = max(0.0, 1.0 - distance)
                semantic_scores[row.id] = similarity * _SEMANTIC_MAX
            logger.debug("semantic_scores_computed", count=len(semantic_scores))
        except Exception:
            logger.warning("semantic_scoring_failed_using_word_overlap", exc_info=True)
    else:
        logger.debug(
            "semantic_scoring_skipped_non_semantic_provider",
            provider=type(provider).__name__ if provider else None,
        )

    use_semantic = len(semantic_scores) > 0

    # -- Union semantic candidates into the per-kind pools ------------------
    #
    # The recency-bounded fetches above (top-50 profile_facts, top-20
    # procedures, top-30 episode_summaries by created_at DESC) were the
    # candidate pool for ranking. Memories outside those windows could
    # never be ranked, even when their cosine distance to the query was
    # zero — verified live: docs-pack memories like "API process is
    # CPU-only, no GPU required" sit at position ~130 of 160 profile_facts
    # by recency in the docs subject and were locked out, producing the
    # 25%↔75% eval variability across re-bootstraps depending on which
    # memories happened to land in the top-50 window.
    #
    # Fix: when a real semantic provider is available, also feed every
    # memory returned by search_memories_by_embedding into the candidate
    # pool — bucketed by kind, deduped by id with the recency-fetched
    # rows. The semantic call is already happening for score lookup; we
    # just stop throwing away the rows it returned. Final ranking is
    # unchanged.
    #
    # Stub-provider / no-provider deployments are unaffected:
    # `semantic_results` is empty, the union below is a no-op, behavior
    # is byte-identical to the previous recency-only pool.
    if semantic_results:
        sem_by_id: dict[uuid.UUID, Any] = {row.id: row for row, _ in semantic_results}
        existing_fact_ids = {r.id for r in fact_rows}
        existing_proc_ids = {r.id for r in procedure_rows}
        existing_summary_ids = {r.id for r in summary_rows}
        added_facts = []
        added_procs = []
        added_summaries = []
        for row in sem_by_id.values():
            if row.kind == "profile_fact" and row.id not in existing_fact_ids:
                added_facts.append(row)
            elif row.kind == "procedure" and row.id not in existing_proc_ids:
                added_procs.append(row)
            elif row.kind == "episode_summary" and row.id not in existing_summary_ids:
                added_summaries.append(row)
        if added_facts or added_procs or added_summaries:
            logger.debug(
                "semantic_candidates_unioned",
                added_facts=len(added_facts),
                added_procs=len(added_procs),
                added_summaries=len(added_summaries),
            )
        # SQLAlchemy returns ScalarResult-backed sequences from search_memories;
        # convert to lists so we can extend safely.
        fact_rows = list(fact_rows) + added_facts
        procedure_rows = list(procedure_rows) + added_procs
        summary_rows = list(summary_rows) + added_summaries

    # -- Apply sensitivity-label policy (#50) -------------------------------
    # Evaluate each candidate memory against the active policy bundle BEFORE
    # scoring, so denied rows never enter the ranker and the receipt's
    # `policy.filters_applied` records every fired rule. In log_only mode
    # (the default) no row is removed — the decisions are still recorded
    # into the receipt so tenants can audit what *would* be filtered before
    # flipping to enforce.
    tenant_config = await receipts_service.load_tenant_receipt_config(session, tenant_id)
    policy_mode = (tenant_config or {}).get("policy_mode", "log_only")
    policy_enforce = policy_mode == "enforce"
    # Replay path injects the historical snapshot bundle so the
    # decision matrix matches the original assembly's, even if the
    # live `policy_bundles` row has since been overwritten or
    # removed. The override flag is intentionally separate from the
    # value being non-None: a snapshot that captured "no bundle was
    # active" replays correctly by passing
    # `_use_policy_bundle_override=True, _policy_bundle_override=None`.
    if _use_policy_bundle_override:
        active_bundle = _policy_bundle_override
    else:
        active_bundle = await policy_service.resolve_active_bundle(session, tenant_id)
    policy_context = policy_service.PolicyContext(
        caller_id=caller_id,
        caller_type=caller_type,
        tenant_id=tenant_id,
    )

    all_memory_rows_pre_policy = list(fact_rows) + list(procedure_rows) + list(summary_rows)
    policy_decisions: dict[Any, policy_service.MemoryPolicyDecision] = {}
    for row in all_memory_rows_pre_policy:
        decision = policy_service.evaluate_memory(
            memory_labels=list(getattr(row, "sensitivity_labels", None) or []),
            bundle=active_bundle,
            context=policy_context,
        )
        if decision.action != "allow":
            policy_decisions[row.id] = decision

    if policy_enforce and policy_decisions:
        # Filter the per-kind buckets so the rest of the assembly path
        # never sees a denied row. Redact is applied in place by
        # apply_decisions and the row is kept.
        kept_facts, _ = policy_service.apply_decisions(fact_rows, policy_decisions, enforce=True)
        kept_procs, _ = policy_service.apply_decisions(
            procedure_rows, policy_decisions, enforce=True
        )
        kept_summaries, _ = policy_service.apply_decisions(
            summary_rows, policy_decisions, enforce=True
        )
        fact_rows = kept_facts
        procedure_rows = kept_procs
        summary_rows = kept_summaries
        denied_count = sum(1 for d in policy_decisions.values() if d.action == "deny")
        redacted_count = sum(1 for d in policy_decisions.values() if d.action == "redact")
        logger.info(
            "policy_enforced",
            tenant_id=tenant_id,
            bundle_hash=active_bundle.bundle_hash if active_bundle else None,
            denied=denied_count,
            redacted=redacted_count,
        )

    # -- Score all candidates ------------------------------------------------
    task_tokens = _tokenize_for_relevance(task)

    # Issue #116: track whether ANY candidate is genuinely task-relevant
    # (semantic similarity above the floor, or a literal token overlap).
    # Drives the no-task-context caveat below without altering ranking.
    task_relevant = False

    scored: list[_ScoredItem] = []
    all_memory_rows = list(fact_rows) + list(procedure_rows) + list(summary_rows)

    # -- Resolve source-episode breadcrumbs for docs-grounded ranking --------
    # Memories from the docs pack carry source_episode_ids pointing at
    # episodes whose payload includes a `breadcrumb` (e.g. "Compiler Modes
    # › Heuristic Compilation"). Bringing that signal into ranking is what
    # surfaces `architecture/compiler-modes.md` for a "heuristic vs LLM"
    # query instead of letting `why-statewave.md` win on body-content
    # overlap. We bulk-fetch the union of source episode ids in one query,
    # build a memory_id → [breadcrumbs] map, then apply a small additive
    # bonus during scoring. Memories with no breadcrumb-bearing source
    # (i.e. visitor-memory subjects) get no bonus — the new behavior is
    # auto-scoped to docs-pack data by the data shape itself.
    memory_breadcrumbs: dict[uuid.UUID, list[str]] = {}
    needed_episode_ids: set[uuid.UUID] = set()
    for row in all_memory_rows:
        for sid in row.source_episode_ids or []:
            needed_episode_ids.add(sid)
    if needed_episode_ids:
        try:
            src_episodes = await repo.get_episodes_by_ids(session, list(needed_episode_ids))
            ep_breadcrumb_by_id: dict[uuid.UUID, str] = {}
            for ep in src_episodes:
                payload = ep.payload or {}
                bc = payload.get("breadcrumb") or payload.get("title")
                if isinstance(bc, str) and bc:
                    ep_breadcrumb_by_id[ep.id] = bc
            for row in all_memory_rows:
                bcs = [
                    ep_breadcrumb_by_id[sid]
                    for sid in (row.source_episode_ids or [])
                    if sid in ep_breadcrumb_by_id
                ]
                if bcs:
                    memory_breadcrumbs[row.id] = bcs
        except Exception:
            logger.warning("breadcrumb_lookup_failed", exc_info=True)

    if all_memory_rows:
        ts_range = _timestamp_range([r.created_at for r in all_memory_rows])
        for row in all_memory_rows:
            # Use semantic score when available, else word-overlap. When
            # semantic is in play, also add a lexical-overlap bonus so a
            # memory that BOTH embeds close to the query AND literally
            # contains the query keywords clears the kind/recency noise
            # floor. The fallback word-overlap path already factors in
            # literal matches, so the lexical bonus is only applied on
            # top of the semantic path to avoid double-counting.
            if use_semantic and row.id in semantic_scores:
                relevance = semantic_scores[row.id]
                lexical_bonus = _lexical_overlap_bonus(row.content, task_tokens)
            else:
                relevance = _relevance_score(row.content, task_tokens)
                lexical_bonus = 0.0

            # Issue #116: does THIS memory actually relate to the task?
            # Use the stopword-aware meaningful-overlap predicate (same one
            # the lexical bonus uses) — raw word overlap would treat a shared
            # "the"/"is" as relevance and never fire the caveat.
            if not task_relevant and task_tokens:
                if use_semantic and row.id in semantic_scores:
                    cosine_sim = semantic_scores[row.id] / _SEMANTIC_MAX
                    if cosine_sim >= _SEMANTIC_RELEVANCE_FLOOR or lexical_bonus > 0.0:
                        task_relevant = True
                elif _lexical_overlap_bonus(row.content, task_tokens) > 0.0:
                    task_relevant = True

            breadcrumb_bonus = _breadcrumb_overlap_bonus(
                memory_breadcrumbs.get(row.id, []),
                task_tokens,
            )

            score = (
                _KIND_PRIORITY.get(row.kind, 1.0)
                + _recency_score(row.created_at, ts_range)
                + relevance
                + lexical_bonus
                + _temporal_score(row.valid_from, row.valid_to)
                + breadcrumb_bonus
            )
            scored.append(
                _ScoredItem(
                    score=score,
                    kind="memory",
                    memory_row=row,
                    text=_render_memory_line(row),
                    section=_section_for_kind(row.kind),
                )
            )

    if episode_rows:
        # Collect episode IDs already covered by included summaries
        covered_episode_ids: set[str] = set()
        for row in summary_rows:
            if row.source_episode_ids:
                covered_episode_ids.update(str(eid) for eid in row.source_episode_ids)

        # Suppress episodes whose backing memories are ALL superseded. Without
        # this, a correctly-superseded fact still resurfaces verbatim here
        # ("Recent interactions") and leaks the stale value back into the
        # prompt. Episodes that still back an active memory — or back no memory
        # at all — are preserved. Deterministic set logic; no embeddings/LLM.
        obsolete_episode_ids: set[str] = {
            str(eid)
            for eid in await repo.superseded_only_episode_ids(
                session, subject_id, tenant_id=tenant_id
            )
        }

        # Fetch resolved sessions to penalize their episodes
        resolved_sessions = await repo.get_resolved_session_ids(
            session, subject_id, tenant_id=tenant_id
        )
        open_sessions = await repo.get_open_session_ids(session, subject_id, tenant_id=tenant_id)

        # -- Repeat-issue detection -----------------------------------------
        # Build keyword set for current session episodes
        current_session_texts: list[str] = []
        resolved_session_texts: dict[str, list[str]] = {}
        for row in episode_rows:
            ep_session = getattr(row, "session_id", None)
            text = extract_payload_text(row.payload)
            if not text:
                continue
            if session_id and ep_session == session_id:
                current_session_texts.append(text)
            elif ep_session and ep_session in resolved_sessions:
                resolved_session_texts.setdefault(ep_session, []).append(text)

        current_keywords = _extract_issue_keywords(" ".join(current_session_texts))

        # Also include task text in current keywords (the task describes the issue)
        current_keywords |= _extract_issue_keywords(task)

        # Compute overlap for each resolved session + check resolution summaries
        repeat_issue_sessions: dict[str, float] = {}  # session_id → boost score
        if current_keywords:
            # Fetch resolution summaries for resolved sessions
            resolution_rows = await repo.list_resolutions(
                session, subject_id, tenant_id=tenant_id, status="resolved", limit=20
            )
            resolution_summaries: dict[str, str] = {
                r.session_id: r.resolution_summary or ""
                for r in resolution_rows
                if r.resolution_summary
            }

            for sid, texts in resolved_session_texts.items():
                prior_keywords = _extract_issue_keywords(" ".join(texts))
                # Include resolution summary keywords
                if sid in resolution_summaries:
                    prior_keywords |= _extract_issue_keywords(resolution_summaries[sid])
                overlap = _session_keyword_overlap(current_keywords, prior_keywords)
                if overlap >= _REPEAT_OVERLAP_THRESHOLD:
                    # Higher boost if there's a useful resolution summary
                    if sid in resolution_summaries:
                        repeat_issue_sessions[sid] = _REPEAT_RESOLVED_BOOST
                    else:
                        repeat_issue_sessions[sid] = _REPEAT_ISSUE_BOOST

        ep_ts_range = _timestamp_range([r.created_at for r in episode_rows])
        for row in episode_rows:
            if str(row.id) in covered_episode_ids:
                continue  # already represented by a summary
            if str(row.id) in obsolete_episode_ids:
                continue  # backing memories all superseded (Phase-1 leak fix)
            if is_structured_episode(row.payload):
                # Compiled from structured candidates: its active atomic memories
                # ARE its representation. Never inject the raw mixed body, so a
                # superseded atomic fact cannot leak back through it. The raw
                # episode stays in timeline/admin/history.
                continue
            ep_text = _short_episode_text(row.payload, row.source, row.type)
            content_text = extract_payload_text(row.payload)
            ep_relevance = _relevance_score(content_text, task_tokens)
            # Issue #116: a meaningful (non-stopword) token overlap with an
            # episode also counts as the task being on-topic for this subject.
            if (
                not task_relevant
                and task_tokens
                and _lexical_overlap_bonus(content_text, task_tokens) > 0.0
            ):
                task_relevant = True
            score = _EPISODE_PRIORITY + _recency_score(row.created_at, ep_ts_range) + ep_relevance
            # Boost episodes belonging to the active session
            if session_id and getattr(row, "session_id", None) == session_id:
                score += _SESSION_BOOST
            # Penalize episodes from resolved sessions
            ep_session = getattr(row, "session_id", None)
            if ep_session and ep_session in resolved_sessions:
                score += _RESOLVED_SESSION_PENALTY
            # Repeat-issue boost (counteracts resolved penalty for recurring issues)
            if ep_session and ep_session in repeat_issue_sessions:
                score += repeat_issue_sessions[ep_session]
            # Boost episodes from sessions with open issues
            if ep_session and ep_session in open_sessions:
                score += _OPEN_ISSUE_BOOST
            # Boost action/step episodes (agent, assistant, tool, system)
            if row.source in ("assistant", "agent", "tool", "system") or row.type in (
                "action",
                "tool_call",
                "resolution_attempt",
            ):
                score += _ACTION_STEP_BOOST
            # Urgency keyword detection
            if content_text and _has_urgency(content_text):
                score += _URGENCY_BOOST
            # Idle chatter penalty (very short, low-signal content)
            if content_text and len(content_text.strip()) < 20:
                score += _IDLE_CHATTER_PENALTY
            scored.append(
                _ScoredItem(
                    score=score,
                    kind="episode",
                    episode_row=row,
                    text=f"- {ep_text}",
                    section="episodes",
                )
            )

    # Sort descending by score (stable sort preserves insertion order for ties)
    scored.sort(key=lambda s: s.score, reverse=True)

    # -- Assemble within budget ---------------------------------------------
    task_header = f"## Task\n{task}\n"
    budget_used = len(enc.encode(task_header))

    # Issue #116: if nothing cleared a real task-relevance signal, we still
    # return the subject's memory (recall preserved) but prepend an explicit
    # caveat. Reserve its tokens up front so the budget stays honored.
    show_no_task_context = bool(task.strip()) and not task_relevant
    if show_no_task_context:
        budget_used += len(enc.encode(_NO_TASK_CONTEXT_NOTE + "\n"))

    included_facts: list[MemoryResponse] = []
    included_summaries: list[MemoryResponse] = []
    included_procedures: list[MemoryResponse] = []
    included_episodes: list[EpisodeResponse] = []
    section_lines: dict[str, list[str]] = {
        "facts": [],
        "procedures": [],
        "history": [],
        "episodes": [],
    }

    # Receipt provenance: collect the rows + score that actually made it
    # into the bundle, in rank order. Recorded alongside the response so
    # the receipt-write path doesn't have to walk `scored` again.
    receipt_memory_rows: list[tuple[Any, float, int]] = []
    receipt_episode_rows: list[tuple[Any, float, int]] = []
    rank_counter = 0

    for item in scored:
        item_tokens = len(enc.encode(item.text))
        # Reserve ~10 tokens for section headers
        if budget_used + item_tokens + 10 > budget:
            continue  # skip this item, try smaller ones

        budget_used += item_tokens
        section_lines[item.section].append(item.text)
        rank_counter += 1

        if item.kind == "memory":
            resp = _memory_response(item.memory_row)
            if item.memory_row.kind == "profile_fact":
                included_facts.append(resp)
            elif item.memory_row.kind == "procedure":
                included_procedures.append(resp)
            elif item.memory_row.kind == "episode_summary":
                included_summaries.append(resp)
            receipt_memory_rows.append((item.memory_row, item.score, rank_counter))
        elif item.kind == "episode":
            included_episodes.append(_episode_response(item.episode_row))
            receipt_episode_rows.append((item.episode_row, item.score, rank_counter))

    # -- Render final text --------------------------------------------------
    # Order: task → facts → procedures → history → episodes
    # This mirrors how a human would brief someone: who is this person,
    # what do they need, what happened recently, raw detail if room.
    parts: list[str] = [task_header]
    if show_no_task_context:
        parts.append(_NO_TASK_CONTEXT_NOTE)
        parts.append("")
    if section_lines["facts"]:
        parts.append("## About this user")
        parts.extend(section_lines["facts"])
        parts.append("")
    if section_lines["procedures"]:
        parts.append("## Procedures")
        parts.extend(section_lines["procedures"])
        parts.append("")
    if section_lines["history"]:
        parts.append("## Recent history")
        parts.extend(section_lines["history"])
        parts.append("")
    if section_lines["episodes"]:
        parts.append("## Recent interactions")
        # Group by session when a current session is specified
        if session_id:
            current_lines: list[str] = []
            other_lines: list[str] = []
            for ep in included_episodes:
                line = f"- {_short_episode_text(ep.payload, ep.source, ep.type)}"
                if ep.session_id == session_id:
                    current_lines.append(line)
                else:
                    other_lines.append(line)
            if current_lines:
                parts.append(f"### Current session ({session_id})")
                parts.extend(current_lines)
            if other_lines:
                parts.append("### Previous interactions")
                parts.extend(other_lines)
        else:
            parts.extend(section_lines["episodes"])
        parts.append("")

    assembled = "\n".join(parts)
    token_estimate = len(enc.encode(assembled))

    provenance = {
        "fact_ids": [str(f.id) for f in included_facts],
        "summary_ids": [str(s.id) for s in included_summaries],
        "procedure_ids": [str(p.id) for p in included_procedures],
        "episode_ids": [str(e.id) for e in included_episodes],
    }

    # Build session info from included episodes
    session_map: dict[str, list[EpisodeResponse]] = {}
    for ep in included_episodes:
        sid = ep.session_id or "__none__"
        session_map.setdefault(sid, []).append(ep)
    sessions: list[SessionInfo] = []
    for sid, eps in session_map.items():
        if sid == "__none__":
            continue
        times = [e.created_at for e in eps if e.created_at]
        sessions.append(
            SessionInfo(
                session_id=sid,
                episode_count=len(eps),
                first_at=min(times) if times else None,
                last_at=max(times) if times else None,
            )
        )

    # Include summaries in the facts response list for backward compatibility
    all_facts = included_facts + included_summaries

    # -- Receipt emission ---------------------------------------------------
    # Decide-and-write happens after the bundle is fully built so the
    # `output.context_hash` covers the exact bytes we're about to return.
    receipt_id, receipt_emitted = await _maybe_emit_receipt(
        session=session,
        tenant_id=tenant_id,
        subject_id=subject_id,
        task=task,
        as_of=as_of,
        assembled=assembled,
        token_estimate=token_estimate,
        memory_rows=receipt_memory_rows,
        episode_rows=receipt_episode_rows,
        emit_receipt=emit_receipt,
        query_id=query_id,
        task_id=task_id,
        parent_receipt_id=parent_receipt_id,
        mode=_mode_override or "retrieval",
        caller_id=caller_id,
        caller_type=caller_type,
        tenant_config=tenant_config,
        policy_bundle=active_bundle,
        policy_mode=policy_mode,
        policy_decisions=policy_decisions,
    )

    return ContextBundleResponse(
        subject_id=subject_id,
        task=task,
        facts=all_facts,
        episodes=included_episodes,
        procedures=included_procedures,
        provenance=provenance,
        assembled_context=assembled,
        token_estimate=token_estimate,
        sessions=sessions,
        receipt_id=receipt_id,
        receipt_emitted=receipt_emitted,
    )


async def _maybe_emit_receipt(
    *,
    session: AsyncSession,
    tenant_id: str | None,
    subject_id: str,
    task: str,
    as_of: datetime,
    assembled: str,
    token_estimate: int,
    memory_rows: list[tuple[Any, float, int]],
    episode_rows: list[tuple[Any, float, int]],
    emit_receipt: bool | None,
    query_id: str | None,
    task_id: str | None,
    parent_receipt_id: str | None,
    mode: str,
    caller_id: str | None = None,
    caller_type: str | None = None,
    tenant_config: dict[str, Any] | None = None,
    policy_bundle: policy_service.PolicyBundle | None = None,
    policy_mode: str = "log_only",
    policy_decisions: dict[Any, policy_service.MemoryPolicyDecision] | None = None,
) -> tuple[str | None, bool]:
    """Decide, build, and persist a receipt. Returns (receipt_id, emitted).

    Both surfaces that emit receipts (`/v1/context` and `/v1/handoff`)
    share this routine — keeping the decision logic and the write path
    in a single place avoids divergence as the schema grows."""
    if tenant_config is None:
        tenant_config = await receipts_service.load_tenant_receipt_config(session, tenant_id)
    decision = receipts_service.decide_emission(
        request_flag=emit_receipt,
        tenant_config=tenant_config,
        policy_decision=None,  # reserved for force-on (#50 v2)
    )
    if not decision.emit:
        logger.debug(
            "receipt_skipped",
            reason=decision.reason,
            subject_id=subject_id,
            tenant_id=tenant_id,
        )
        return None, False

    context_hash, context_size_bytes = receipts_service.canonicalize_context(assembled)
    policy_decisions = policy_decisions or {}
    filters_applied = policy_service.build_filters_applied(policy_decisions)
    filters_skipped = policy_service.build_filters_skipped(policy_decisions, policy_bundle)

    selected_memories = [
        receipts_service.SelectedMemory(
            memory_id=row.id,
            kind=row.kind,
            valid_from=row.valid_from,
            valid_to=row.valid_to,
            supersession_status=receipts_service.supersession_status_from_row(row),
            source_episode_ids=list(row.source_episode_ids or []),
            rank=rank,
            score=score,
        )
        for row, score, rank in memory_rows
    ]
    selected_episodes = [
        receipts_service.SelectedEpisode(
            episode_id=row.id,
            source=row.source,
            type=row.type,
            occurred_at=getattr(row, "occurred_at", None) or row.created_at,
            rank=rank,
        )
        for row, score, rank in episode_rows
    ]

    receipt_id = receipts_service.new_ulid()
    # v0.9 #159 — embed the active bundle's YAML on every receipt so
    # the replay engine can re-evaluate against the same rule set the
    # original assembly saw, even if the live `policy_bundles` row
    # was later deleted or overwritten. When no bundle is active we
    # still stamp a snapshot envelope with null fields so replay can
    # distinguish "no policy was active" (replayable) from
    # "pre-v0.9 receipt, no snapshot exists" (NOT replayable).
    policy_snapshot = receipts_service.build_policy_snapshot(
        bundle_hash=policy_bundle.bundle_hash if policy_bundle else None,
        bundle_yaml=policy_bundle.yaml_content if policy_bundle else None,
    )
    body = receipts_service.build_receipt_body(
        receipt_id=receipt_id,
        mode=mode,
        tenant_id=tenant_id,
        subject_id=subject_id,
        task=task,
        as_of=as_of,
        selected_memories=selected_memories,
        selected_episodes=selected_episodes,
        context_hash=context_hash,
        context_size_bytes=context_size_bytes,
        token_estimate=token_estimate,
        query_id=query_id,
        task_id=task_id,
        parent_receipt_id=parent_receipt_id,
        policy_bundle_hash=policy_bundle.bundle_hash if policy_bundle else None,
        policy_mode=policy_mode,
        filters_applied=filters_applied,
        filters_skipped=filters_skipped,
        caller_id=caller_id,
        caller_type=caller_type,
        policy_snapshot=policy_snapshot,
        # v0.9 #161 — stamp the local region on every emitted receipt
        # so the audit trail records *where* the decision was made,
        # not just *what* it was. None in single-region deployments.
        region=settings.region,
    )
    written_id = await receipts_service.write_receipt(
        session, receipt_body=body, as_of=as_of, tenant_config=tenant_config
    )
    if written_id is None:
        # Failure path: bundle still authoritative, log already emitted.
        return None, False
    logger.info(
        "receipt_emitted",
        receipt_id=written_id,
        reason=decision.reason,
        subject_id=subject_id,
        tenant_id=tenant_id,
        memory_count=len(selected_memories),
        episode_count=len(selected_episodes),
        context_size_bytes=context_size_bytes,
        policy_filters_applied=len(filters_applied),
        policy_bundle_hash=(policy_bundle.bundle_hash if policy_bundle else None),
    )
    return written_id, True


# ---------------------------------------------------------------------------
# Scoring helpers
# ---------------------------------------------------------------------------

# Common stopwords to exclude from issue keyword extraction
_STOPWORDS = frozenset(
    "i me my we our you your he she it they them their this that "
    "is am are was were be been being have has had do does did will would "
    "shall should can could may might must a an the and but or nor for "
    "so yet at by from in into of on to up with as if then else when "
    "where how what which who whom why all any each every no not only "
    "very just also too hi hello thanks thank okay ok yes please hey "
    "sure got get let know think want need help".split()
)


def _extract_issue_keywords(text: str) -> set[str]:
    """Extract meaningful issue keywords (lowercase, punctuation-stripped,
    stopwords removed).

    Shares `_tokenize_for_relevance`'s punctuation handling so a
    sentence-final keyword like "outage!" still matches a clean "outage"
    during repeat-issue overlap detection — without it the trailing
    punctuation silently zeros the overlap (same failure that helper's
    docstring documents for the lexical signal)."""
    if not text:
        return set()
    return _tokenize_for_relevance(text) - _STOPWORDS


def _session_keyword_overlap(current_keywords: set[str], prior_keywords: set[str]) -> float:
    """Compute overlap ratio of current keywords found in prior keywords."""
    if not current_keywords or not prior_keywords:
        return 0.0
    overlap = len(current_keywords & prior_keywords)
    return overlap / len(current_keywords)


# Lexical tokenization is centralized in server.services.tokenization so the
# relevance scorer and the conflict detector can never drift apart (#198/#199).
_RELEVANCE_PUNCT = EDGE_PUNCT
_tokenize_for_relevance = tokenize


def _has_urgency(text: str) -> bool:
    """Check if text contains urgency/severity keywords."""
    lower = text.lower()
    return any(kw in lower for kw in _URGENCY_KEYWORDS)


def _relevance_score(content: str, task_tokens: set[str]) -> float:
    """Word-overlap relevance: 0 to _RELEVANCE_MAX."""
    if not task_tokens or not content:
        return 0.0
    content_tokens = _tokenize_for_relevance(content)
    overlap = len(task_tokens & content_tokens)
    # Normalize by task token count
    ratio = overlap / len(task_tokens)
    return min(ratio * _RELEVANCE_MAX, _RELEVANCE_MAX)


def _lexical_overlap_bonus(content: str, task_tokens: set[str]) -> float:
    """Additive bonus when query terms appear verbatim in memory content.

    Applied on top of the semantic score. Stopwords ("how", "do", "the",
    "with", ...) are excluded so trivial overlap on connectives can't
    inflate the bonus — only meaningful keyword agreement counts.

    Normalizes by the count of meaningful task tokens, so a 2/2 hit on a
    short query like "install npm" scores the same as a 4/4 hit on a
    longer one. Capped at _LEXICAL_BONUS_MAX.
    """
    if not task_tokens or not content:
        return 0.0
    meaningful = {t for t in task_tokens if t and t not in _STOPWORDS}
    if not meaningful:
        return 0.0
    content_tokens = _tokenize_for_relevance(content)
    overlap = len(meaningful & content_tokens)
    if overlap == 0:
        return 0.0
    ratio = overlap / len(meaningful)
    return min(ratio * _LEXICAL_BONUS_MAX, _LEXICAL_BONUS_MAX)


# Generic words that drag breadcrumb overlap toward noise without informing
# topic match. Excluded so e.g. "Statewave Documentation" and the user's
# "What is Statewave?" don't trivially co-overlap on "statewave".
_BREADCRUMB_STOPWORDS = frozenset(
    "statewave documentation guide overview the a an of for to on in with "
    "is are be use using how what why which when where do does i my we "
    "& > › -".split()
)


def _breadcrumb_overlap_bonus(
    breadcrumbs: list[str],
    task_tokens: set[str],
) -> float:
    """Lexical token overlap between query and a memory's source breadcrumbs.

    Used only for memories whose source episode is a docs-pack section (the
    only episodes that carry `payload.breadcrumb`); visitor-memory subjects
    don't carry this shape so the bonus is automatically inert there. The
    breadcrumb is short ("Compiler Modes › Heuristic Compilation"), so a
    handful of token matches is a strong topical signal — this is the lever
    that gets `architecture/compiler-modes.md` to surface for "heuristic
    vs LLM" instead of `why-statewave.md` winning by general semantic match
    on the body content.

    Returns 0.0 when there are no breadcrumbs or no task tokens. Returns at
    most _BREADCRUMB_MAX so this can never dominate KIND_PRIORITY or
    SEMANTIC_MAX — it nudges, doesn't override.
    """
    if not breadcrumbs or not task_tokens:
        return 0.0
    # Strip punctuation that _tokenize_for_relevance leaves attached
    # (e.g. "GPU?" → "gpu") so task tokens match cleanly against the
    # breadcrumb tokens we compute below.
    task_meaningful = {t.strip("?.,:;()[]{}'\"") for t in task_tokens}
    task_meaningful = {t for t in task_meaningful if t and t not in _BREADCRUMB_STOPWORDS}
    if not task_meaningful:
        return 0.0
    best = 0.0
    for bc in breadcrumbs:
        if not bc:
            continue
        bc_tokens = {
            t.lower().strip("?.,:;()[]{}'\"")
            for t in bc.replace("›", " ").replace(">", " ").split()
        }
        bc_tokens -= _BREADCRUMB_STOPWORDS
        bc_tokens.discard("")
        if not bc_tokens:
            continue
        overlap = len(task_meaningful & bc_tokens)
        if overlap == 0:
            continue
        # Normalize by min(task_meaningful, bc_tokens) so a short breadcrumb
        # with a perfect 2/2 match scores as well as a long one with 2/4.
        denom = min(len(task_meaningful), len(bc_tokens))
        ratio = overlap / denom
        score = ratio * _BREADCRUMB_MAX
        if score > best:
            best = score
    return min(best, _BREADCRUMB_MAX)


def _temporal_score(
    valid_from: datetime | None,
    valid_to: datetime | None,
) -> float:
    """Score boost/penalty based on temporal validity window.

    - Memory with no valid_to (still current): gets a bonus.
    - Memory whose valid_to is in the future: gets a bonus.
    - Memory whose valid_to is in the past (expired/superseded): gets a penalty.
    """
    now = datetime.now(timezone.utc)
    if valid_to is None:
        return _TEMPORAL_VALID_BONUS  # still active / no expiry
    if valid_to.tzinfo is None:
        valid_to = valid_to.replace(tzinfo=timezone.utc)
    if valid_to > now:
        return _TEMPORAL_VALID_BONUS  # still within validity window
    return _TEMPORAL_EXPIRED_PENALTY  # expired


def _timestamp_range(
    timestamps: list[datetime],
) -> tuple[float, float]:
    """Return (min_ts, max_ts) as unix floats."""
    ts_floats = [t.timestamp() for t in timestamps if t]
    if not ts_floats:
        return (0.0, 0.0)
    return (min(ts_floats), max(ts_floats))


def _recency_score(
    created_at: datetime | None,
    ts_range: tuple[float, float],
) -> float:
    """Linear recency: most recent = _RECENCY_MAX, oldest = 0."""
    if not created_at:
        return 0.0
    ts_min, ts_max = ts_range
    if ts_max == ts_min:
        return _RECENCY_MAX  # single item
    t = created_at.timestamp()
    ratio = (t - ts_min) / (ts_max - ts_min)
    return ratio * _RECENCY_MAX


def _section_for_kind(kind: str) -> str:
    if kind == "profile_fact":
        return "facts"
    if kind == "procedure":
        return "procedures"
    if kind == "episode_summary":
        return "history"
    return "history"


def _short_episode_text(payload: dict, source: str, type_: str) -> str:
    """Render an episode as a single line for context text."""
    text = extract_payload_text(payload)
    if text:
        return f"[{source}/{type_}] {text[:150]}"
    return f"[{source}/{type_}] (no text content)"


def _render_memory_line(row: Any) -> str:
    """Render a memory as a clean context line, appropriate for its kind.

    Date prefix: every memory carries a `valid_from` (set by the compiler
    from `payload.event_time` / `payload.messages[0].timestamp` /
    `ep.created_at`). If the memory's content text doesn't already
    mention an absolute date, we prefix the line with `[YYYY-MM-DD]` so
    downstream answer-model prompts can reason about timing. Without
    this, questions like "when did X happen?" become unanswerable from
    the bundle even though the temporal info is on the row.
    """
    if row.kind == "episode_summary":
        # Summaries contain raw "role: content" lines — condense into a readable note
        body = _clean_summary(row.content)
    else:
        body = row.content
    date_prefix = _date_prefix_for(row, body)
    return f"- {date_prefix}{body}" if date_prefix else f"- {body}"


def _date_prefix_for(row: Any, body: str) -> str:
    """Return a `[YYYY-MM-DD] ` prefix if `row.valid_from` looks real
    (not the bench-incident default of "compile time") AND the body
    doesn't already mention an absolute date. Returns empty string
    otherwise.
    """
    import re

    valid_from = getattr(row, "valid_from", None)
    if valid_from is None:
        return ""
    # If the body already contains a YYYY year token (2019-2099 range),
    # the compiler already surfaced a date — don't double-stamp.
    if re.search(r"\b(20[1-9]\d)\b", body):
        return ""
    try:
        stamp = valid_from.strftime("%Y-%m-%d")
    except (AttributeError, ValueError):
        return ""
    return f"[{stamp}] "


def _clean_summary(text: str) -> str:
    """Turn raw 'user: X\\nassistant: Y' into a readable history note."""
    import re

    lines = text.strip().split("\n")
    parts: list[str] = []
    for line in lines:
        # Strip role prefix
        cleaned = re.sub(r"^(user|assistant|system):\s*", "", line.strip())
        if cleaned:
            parts.append(cleaned)
    if len(parts) <= 2:
        return " → ".join(parts)
    return parts[0] + " → " + parts[-1]  # first user msg → last assistant response


# ---------------------------------------------------------------------------
# Row → response converters
# ---------------------------------------------------------------------------


def _memory_response(row: Any) -> MemoryResponse:
    return MemoryResponse(
        id=row.id,
        subject_id=row.subject_id,
        kind=row.kind,
        content=row.content,
        summary=row.summary,
        confidence=row.confidence,
        valid_from=row.valid_from,
        valid_to=row.valid_to,
        source_episode_ids=row.source_episode_ids or [],
        metadata=row.metadata_,
        status=row.status,
        sensitivity_labels=list(getattr(row, "sensitivity_labels", None) or []),
        created_at=row.created_at,
        updated_at=row.updated_at,
    )


def _episode_response(row: Any) -> EpisodeResponse:
    return EpisodeResponse(
        id=row.id,
        subject_id=row.subject_id,
        source=row.source,
        type=row.type,
        payload=row.payload,
        metadata=row.metadata_,
        provenance=row.provenance,
        occurred_at=row.occurred_at,
        created_at=row.created_at,
        session_id=getattr(row, "session_id", None),
    )


# ---------------------------------------------------------------------------
# Internal types
# ---------------------------------------------------------------------------


class _ScoredItem:
    """An item (memory or episode) with its computed relevance score."""

    __slots__ = ("score", "kind", "memory_row", "episode_row", "text", "section")

    def __init__(
        self,
        score: float,
        kind: str,
        text: str,
        section: str,
        memory_row: Any = None,
        episode_row: Any = None,
    ) -> None:
        self.score = score
        self.kind = kind
        self.memory_row = memory_row
        self.episode_row = episode_row
        self.text = text
        self.section = section
