from __future__ import annotations

import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Iterator
from uuid import uuid4

from niannian.chat.ingest_gate import IngestDecision, ingest_gate
from niannian.chat.ingest_gate import trace_line as ingest_gate_trace
from niannian.chat.intents import IntentResult
from niannian.brain.thalamus.normalize import ThalamusGate
from niannian.cognitive.judgment import TurnPolicy
from niannian.cognitive.pathway import resolve_turn_policy, update_pathway_sticky
from niannian.cognitive.plane_pack_gate import PlanePackGate
from niannian.cognitive.persona_registry import merge_planes_bias
from niannian.cognitive.persona_resolver import PersonaResolver
from niannian.chat.batch_reject_gate import batch_reject_gate
from niannian.chat.chat_dedup import chat_dedup_key
from niannian.chat.consolidation_gate import consolidation_gate
from niannian.chat.memory_eligibility import is_memory_eligible_turn_pair
from niannian.chat.remember_gate import recall_pending_fields, remember_gate
from niannian.chat.enrichment_actions import (
    build_vault_md,
    enrichment_summary_text,
    find_turn,
    link_url_for_key,
    part_for_enrichment_key,
    resolve_enrichment,
    update_enrichment_meta,
)
from niannian.chat.mentions import format_mention_context, parse_mentions, resolve_mentions
from niannian.chat.trace_tree import TraceTreeBuilder
from niannian.chat.session import (
    AlreadyRememberedError,
    ConsolidationNotBatchModeError,
    EnrichmentAlreadyPersistedError,
    EnrichmentKindNotSupportedError,
    EnrichmentNotActionableError,
    EnrichmentNotFoundError,
    NotRememberableError,
    PairNotFoundError,
    RecallWindowActiveError,
    Session,
    SessionNotIdleError,
    SessionStore,
    Turn,
    TurnInProgressError,
    TurnNotFoundError,
    TurnRecalledError,
    TurnAlreadySupersededError,
    NotAssistantTurnError,
    _now_iso,
    collect_recent_user_texts,
    resolve_turn_pair,
    text_from_parts,
)
from niannian.chat.input_bus import (
    InputBusContext,
    build_llm_context_block,
    is_sticker_only_parts,
    process_parts,
)
from niannian.chat.parts import normalize_text_to_parts, parts_to_content, resolve_message_input
from niannian.chat.sticker_proactive import (
    build_sticker_learn_text,
    pick_assistant_sticker,
    sticker_learn_allowed,
)
from niannian.chat.stickers_catalog import resolve_sticker_id
from niannian.chat.stickers import STICKER_PACK_ID, sticker_label, validate_sticker_id
from niannian.chat.user_stickers import (
    USER_STICKER_PACK_ID,
    is_user_sticker_id,
    user_sticker_label,
    validate_user_sticker_id,
)
from niannian.token.budget import (
    DEFAULT_BOOL_BUDGET,
    DEFAULT_LEVEL_BUDGET,
    budget_allows,
    chat_llm_allows,
    chat_sticker_allows,
    load_token_budget,
    resolve_chat_memory_mode,
    resolve_chat_sticker_level,
)
from niannian.token.context import token_usage_operation, token_usage_user_id
from niannian.device.gateway import MacRequiredError, NeedConfirmationError
from niannian.device.parse import parse_device_command
from niannian.llm.types import LLMUnavailable, Message, RouterTrace
from niannian.memory.adult_redaction import AdultRedactor
from niannian.memory.disclosure import RecallContext, build_ingest_meta
from niannian.soul.assembly import build_object_assembly_block, build_persona_assembly_block, build_subject_assembly_blocks
from niannian.virtual_persona.addressing import append_user_address_line, resolve_user_address_for_vp
from niannian.virtual_persona.equip import (
    append_vp_persona_skills,
    build_equipped_vp_character_block,
    resolve_vp_character_max_chars,
)
from niannian.virtual_persona.incremental_distill import maybe_distill_turn
from niannian.virtual_persona.store import VirtualPersonaStore
from niannian.virtual_persona.trial import skip_vp_l2_ingest
from niannian.soul.firewall import firewall_trace_line, should_update_subject_emotion, validate_atom_meta
from niannian.personas import BRAINSTEM, EXECUTIVE, JUDGMENT, PUBLIC, RIGHT
from niannian.schemas.models import AssistResult, QueryResult
from niannian.skills.context import SkillInjectContext
from niannian.skills.injector import inject_skills
from niannian.storage.sync.push_after_ingest import schedule_push_after_ingest

logger = logging.getLogger(__name__)


def _build_external_tool_block(
    home,
    config: dict[str, Any],
    tool_scope: dict[str, Any],
    query: str,
) -> str:
    project = str(tool_scope.get("project") or "").strip()
    if not project:
        return ""
    function = str(tool_scope.get("function") or "").strip()
    label = str(tool_scope.get("label") or project).strip()
    from niannian.external_data.service import ExternalDataService

    recall = ExternalDataService(home, config=config).recall_external(
        project,
        function,
        "",
        limit=5,
    )
    learned = str(recall.get("learned") or "").strip()
    if query.strip():
        matched = ExternalDataService(home, config=config).recall_external(
            project,
            function,
            query,
            limit=5,
        )
        records = list(matched.get("records") or [])
    else:
        records = list(recall.get("records") or [])
    if not learned and not records:
        return ""
    lines = [f"## 工具岛记忆（{label}）"]
    if learned:
        lines.append(learned)
    for row in records[:3]:
        blob = str(row.get("payload") or row)
        lines.append(f"- {blob[:240]}")
    return "\n".join(lines).strip()


def _trace_sse_payload(lines: list[str]) -> dict[str, Any]:
    tree = TraceTreeBuilder()
    tree.extend_lines(lines)
    return {"lines": list(lines), "tree": tree.to_list()}


@dataclass
class TurnResult:
    turn_id: str
    reply: str
    intent: str
    trace: list[str] = field(default_factory=list)
    ingested: bool = False
    atom_id: str | None = None
    citations: list[dict[str, Any]] = field(default_factory=list)
    usage: dict[str, Any] | None = None
    user_turn_id: str | None = None
    user_turn_created_at: str | None = None
    rememberable: bool = False
    remember_block_reason: str | None = None
    recall_pending: bool = False
    recall_window_remaining_seconds: int | None = None

    def to_dict(self) -> dict[str, Any]:
        out: dict[str, Any] = {
            "turn_id": self.turn_id,
            "reply": self.reply,
            "intent": self.intent,
            "trace": self.trace,
            "ingested": self.ingested,
            "atom_id": self.atom_id,
            "citations": self.citations,
            "usage": self.usage,
            "user_turn_id": self.user_turn_id,
            "user_turn_created_at": self.user_turn_created_at,
            "rememberable": self.rememberable,
            "recall_pending": self.recall_pending,
        }
        if self.remember_block_reason:
            out["remember_block_reason"] = self.remember_block_reason
        if self.recall_window_remaining_seconds is not None:
            out["recall_window_remaining_seconds"] = self.recall_window_remaining_seconds
        return out


class ChatPipeline:
    """单轮 Chat 编排：意图 → 召回 → 小思 LLM → ingest_gate。"""

    def __init__(
        self,
        orchestrator,
        right,
        router,
        config: dict[str, Any],
        session_store: SessionStore,
        *,
        recall_engine=None,
        skills_loader=None,
        device_gateway=None,
        memory_store=None,
        task_state=None,
        executive=None,
        track_registry=None,
        subject_store=None,
        object_store=None,
    ) -> None:
        self._orchestrator = orchestrator
        self._right = right
        self._router = router
        self._config = config
        self._sessions = session_store
        self._chat_cfg = config.get("chat", {})
        self._recall = recall_engine
        self._skills = skills_loader
        self._device = device_gateway
        self._memory_store = memory_store
        self._task_state = task_state
        self._executive = executive
        self._track_registry = track_registry
        self._subject_store = subject_store
        self._object_store = object_store
        self._redactor = AdultRedactor.from_app_config(config)
        self._enrichment_job_store: Any | None = None

    def _enrichment_jobs(self):
        from niannian.chat.enrichment_jobs import EnrichmentJobStore

        if self._enrichment_job_store is None:
            self._enrichment_job_store = EnrichmentJobStore(self._data_home())
        return self._enrichment_job_store

    def _data_home(self):
        store = self._subject_store
        if store is not None:
            home = getattr(store, "_home", None)
            if home is not None:
                return home
            inner = getattr(store, "_subject", None)
            if inner is not None:
                return getattr(inner, "_home", None)
        if self._object_store is not None:
            return self._object_store._root.parent
        return None

    def _mention_searchers(self, home):
        vault_search = None
        wiki_search = None
        if home is not None:
            from niannian.vault.search import VaultSearchService
            from niannian.vault.store import VaultStore

            manifest = getattr(self._memory_store, "manifest", None) if self._memory_store else None
            vault_store = VaultStore(home, manifest=manifest, config=self._config)
            vault_search = VaultSearchService(vault_store, config=self._config)
        if self._memory_store is not None:
            from niannian.brain.cerebellum.wiki_search import WikiSearch

            wiki_search = WikiSearch(self._memory_store)
        return vault_search, wiki_search

    def _decode_on_export(self) -> bool:
        cfg = self._config.get("adult_redaction") or {}
        return bool(cfg.get("decode_on_export", True))

    def _export_text(self, text: str, *, project: str) -> str:
        if not self._decode_on_export():
            return text
        return self._redactor.decode_for_display(text, project=project)

    @property
    def sessions(self) -> SessionStore:
        return self._sessions

    def create_session(self, project: str = "global") -> Session:
        return self._sessions.create(project)

    def send_sticker(self, session_id: str, sticker_id: str) -> Turn:
        """用户发送表情包 — 仅写入 user turn · 不触发 LLM · 不 ingest。"""
        session = self._sessions.require(session_id)
        home = self._data_home()
        user_id = str(session.meta.get("user_id") or "pinn")
        if home is not None:
            budget = load_token_budget(home, user_id)
            if not chat_sticker_allows(budget, "user"):
                raise ValueError("chat stickers disabled by token budget")
        canonical = resolve_sticker_id(sticker_id)
        if is_user_sticker_id(canonical):
            if home is None:
                raise ValueError(f"unknown sticker_id: {sticker_id!r}")
            validate_user_sticker_id(home, user_id, canonical)
            pack = USER_STICKER_PACK_ID
            label = user_sticker_label(home, user_id, canonical)
        else:
            validate_sticker_id(sticker_id)
            pack = STICKER_PACK_ID
            label = sticker_label(canonical)
        turn = Turn(
            id=uuid4().hex[:12],
            role="user",
            content="",
            created_at=_now_iso(),
            meta={
                "sticker_id": canonical,
                "sticker_pack": pack,
                "label": label,
            },
        )
        self._sessions.append_turn(session_id, turn)
        return turn

    def regenerate_turn(
        self,
        session_id: str,
        assistant_turn_id: str,
        *,
        stream: bool = False,
        show_trace: bool = False,
        plan_mode: bool | None = None,
    ) -> TurnResult | Iterator[tuple[str, dict[str, Any]]]:
        session = self._sessions.require(session_id)
        assistant_turn: Turn | None = None
        for turn in session.turns:
            if turn.id == assistant_turn_id:
                assistant_turn = turn
                break
        if assistant_turn is None:
            raise TurnNotFoundError(assistant_turn_id)
        if assistant_turn.role != "assistant":
            raise NotAssistantTurnError(assistant_turn_id)
        user_turn, assistant_turn = resolve_turn_pair(session, assistant_turn_id)
        if user_turn.meta.get("recalled_at") or assistant_turn.meta.get("recalled_at"):
            raise TurnRecalledError(assistant_turn_id)
        if assistant_turn.meta.get("superseded_by"):
            raise TurnAlreadySupersededError(assistant_turn_id)

        normalized_parts = list(user_turn.parts or normalize_text_to_parts(user_turn.content))
        content = (user_turn.content or parts_to_content(normalized_parts)).strip()
        if not content and not normalized_parts:
            raise ValueError("empty message")

        kwargs: dict[str, Any] = {
            "normalized_parts": normalized_parts,
            "show_trace": show_trace,
            "project": None,
            "plan_mode": plan_mode,
            "regenerate_assistant_id": assistant_turn_id,
        }
        if stream:
            return self._run_turn_stream(session_id, content, **kwargs)
        return self._run_turn_sync(session_id, content, **kwargs)

    def remember_turn(
        self,
        session_id: str,
        assistant_turn_id: str,
        *,
        recall_window_seconds: float = 120.0,
    ) -> dict[str, Any]:
        """手动记忆 user+assistant turn 对；写入单个 atom 并回写双方 meta。"""
        if session_id in self._sessions._turn_locks:
            raise TurnInProgressError(session_id)

        session = self._sessions.require(session_id)
        user_turn, assistant_turn = resolve_turn_pair(session, assistant_turn_id)

        if user_turn.meta.get("recalled_at") or assistant_turn.meta.get("recalled_at"):
            raise TurnRecalledError(assistant_turn_id)

        if user_turn.meta.get("ingested_atom_id") or assistant_turn.meta.get("ingested_atom_id"):
            raise AlreadyRememberedError(assistant_turn_id)

        try:
            created = datetime.fromisoformat(assistant_turn.created_at)
        except ValueError as exc:
            raise RecallWindowActiveError(assistant_turn_id) from exc
        if created.tzinfo is None:
            created = created.replace(tzinfo=timezone.utc)
        elapsed = (datetime.now(timezone.utc) - created).total_seconds()
        if elapsed < recall_window_seconds:
            raise RecallWindowActiveError(assistant_turn_id)

        block_reason = remember_gate(
            user_turn,
            assistant_turn,
            config=self._config,
            check_recall_window=False,
        )
        if block_reason:
            raise NotRememberableError(assistant_turn_id, reason=block_reason)

        gate = PlanePackGate.from_session_meta(session.meta)
        if not gate.allows("P10"):
            raise NotRememberableError(assistant_turn_id, reason="P10_skipped")

        max_chars = int(self._chat_cfg.get("max_ingest_chars", 2000))
        text = f"{user_turn.content}\n{assistant_turn.content}".strip()[:max_chars]

        ingest_meta = build_ingest_meta(session.meta)
        ingest_meta["remembered_via"] = "manual"
        ingest_meta["assistant_turn_id"] = assistant_turn.id
        ingest_meta["user_turn_id"] = user_turn.id
        meta_ok, violations = validate_atom_meta(ingest_meta)
        if not meta_ok:
            raise NotRememberableError(
                assistant_turn_id,
                reason=f"meta_blocked:{','.join(violations)}",
            )

        atom, _orch_trace = self._orchestrator.ingest(
            text,
            tags=["chat"],
            source="chat",
            project=session.project,
            meta=ingest_meta,
        )
        atom_id = atom.id

        user_meta = dict(user_turn.meta)
        user_meta["ingested_atom_id"] = atom_id
        user_turn.meta = user_meta

        assistant_meta = dict(assistant_turn.meta)
        assistant_meta["ingested_atom_id"] = atom_id
        assistant_turn.meta = assistant_meta

        oid = session.meta.get("current_object_id")
        if self._object_store is not None and oid:
            self._object_store.register_disclosed_atom(str(oid), atom_id, ingest_meta)

        ids = session.meta.setdefault("ingested_atom_ids", [])
        ids.append(atom_id)
        schedule_push_after_ingest(self._memory_store, self._config)
        session.updated_at = _now_iso()

        return {
            "ingested_atom_id": atom_id,
            "user_turn_id": user_turn.id,
            "assistant_turn_id": assistant_turn.id,
        }

    def import_enrichment_vault(
        self,
        session_id: str,
        turn_id: str,
        key: str,
        *,
        importer,
    ) -> dict[str, Any]:
        if session_id in self._sessions._turn_locks:
            raise TurnInProgressError(session_id)
        session = self._sessions.require(session_id)
        turn = find_turn(session, turn_id)
        enrichment = resolve_enrichment(turn, key)
        kind = str(enrichment.get("kind") or "")
        if kind == "sticker":
            raise EnrichmentKindNotSupportedError(key, kind, "import-vault")
        part = part_for_enrichment_key(turn, key)
        md_body, hints = build_vault_md(kind, enrichment, part)
        result = importer.import_md(md_body, namespace="user", hints=hints)
        vault_id = str(result.get("vault_id") or "")
        update_enrichment_meta(
            turn,
            key,
            {"persisted": True, "vault_id": vault_id, "imported_at": _now_iso()},
        )
        session.updated_at = _now_iso()
        return {"vault_id": vault_id, "enrichment_key": key}

    def learn_enrichment(
        self,
        session_id: str,
        turn_id: str,
        key: str,
        *,
        web_svc,
        run_batch_fn,
    ) -> dict[str, Any]:
        if session_id in self._sessions._turn_locks:
            raise TurnInProgressError(session_id)
        session = self._sessions.require(session_id)
        turn = find_turn(session, turn_id)
        enrichment = resolve_enrichment(turn, key)
        kind = str(enrichment.get("kind") or "")
        if kind != "link":
            raise EnrichmentKindNotSupportedError(key, kind, "learn")
        url = link_url_for_key(turn, key, enrichment)
        title = str(enrichment.get("title") or url)
        state = web_svc.patch_settings(
            append_query={"topic_id": "chat_link", "query": url, "label": title},
        )
        queries = list(state.get("custom_queries") or [])
        job_id = str(queries[-1].get("id") if queries else f"q_{key[:12]}")
        batch_result = run_batch_fn(topic_id="chat_link", query=title, manual=True)
        if batch_result.get("ok"):
            batch = batch_result.get("batch") or {}
            job_id = str(batch.get("id") or job_id)
        update_enrichment_meta(turn, key, {"learn_job_id": job_id, "learn_queued_at": _now_iso()})
        session.updated_at = _now_iso()
        return {"job_id": job_id, "batch": batch_result}

    def remember_enrichment(
        self,
        session_id: str,
        turn_id: str,
        key: str,
    ) -> dict[str, Any]:
        if session_id in self._sessions._turn_locks:
            raise TurnInProgressError(session_id)
        session = self._sessions.require(session_id)
        turn = find_turn(session, turn_id)
        enrichment = resolve_enrichment(turn, key)
        kind = str(enrichment.get("kind") or "")
        if kind == "sticker":
            raise EnrichmentKindNotSupportedError(key, kind, "remember")
        part = part_for_enrichment_key(turn, key)
        text = enrichment_summary_text(kind, enrichment, part)
        if not text.strip():
            raise EnrichmentNotActionableError(key, "empty_summary")

        gate = PlanePackGate.from_session_meta(session.meta)
        if not gate.allows("P10"):
            raise NotRememberableError(turn_id, reason="P10_skipped")

        max_chars = int(self._chat_cfg.get("max_ingest_chars", 2000))
        ingest_meta = build_ingest_meta(session.meta)
        ingest_meta["remembered_via"] = "enrichment"
        ingest_meta["enrichment_key"] = key
        ingest_meta["user_turn_id"] = turn.id
        meta_ok, violations = validate_atom_meta(ingest_meta)
        if not meta_ok:
            raise NotRememberableError(turn_id, reason=f"meta_blocked:{','.join(violations)}")

        atom, _orch_trace = self._orchestrator.ingest(
            text[:max_chars],
            tags=["chat", "enrichment"],
            source="chat",
            project=session.project,
            meta=ingest_meta,
        )
        atom_id = atom.id
        update_enrichment_meta(
            turn,
            key,
            {"persisted": True, "atom_id": atom_id, "remembered_at": _now_iso()},
        )
        oid = session.meta.get("current_object_id")
        if self._object_store is not None and oid:
            self._object_store.register_disclosed_atom(str(oid), atom_id, ingest_meta)
        ids = session.meta.setdefault("ingested_atom_ids", [])
        ids.append(atom_id)
        schedule_push_after_ingest(self._memory_store, self._config)
        session.updated_at = _now_iso()
        return {"atom_id": atom_id, "enrichment_key": key}

    def consolidate_session(
        self,
        session_id: str,
        *,
        user_id: str = "pinn",
    ) -> dict[str, Any]:
        """Idle batch consolidation — only when chat_memory_mode=batch."""
        if session_id in self._sessions._turn_locks:
            raise TurnInProgressError(session_id)

        session = self._sessions.require(session_id)
        home = self._data_home()
        if home is not None:
            budget = load_token_budget(home, user_id)
        else:
            budget = {**DEFAULT_BOOL_BUDGET, **DEFAULT_LEVEL_BUDGET}

        if resolve_chat_memory_mode(budget) != "batch":
            raise ConsolidationNotBatchModeError()

        consolidation_cfg = self._chat_cfg.get("consolidation") or {}
        idle_minutes = float(consolidation_cfg.get("idle_minutes", 30))
        required_idle = idle_minutes * 60.0
        now = datetime.now(timezone.utc)
        try:
            updated = datetime.fromisoformat(session.updated_at)
        except ValueError as exc:
            raise SessionNotIdleError(session_id, idle_seconds=0, required_seconds=required_idle) from exc
        if updated.tzinfo is None:
            updated = updated.replace(tzinfo=timezone.utc)
        idle_seconds = (now - updated).total_seconds()
        if idle_seconds < required_idle:
            raise SessionNotIdleError(
                session_id,
                idle_seconds=idle_seconds,
                required_seconds=required_idle,
            )

        left_store = getattr(self._orchestrator, "_left", None)
        dedup_hours = int(self._config.get("memory", {}).get("dedup_window_hours", 24))
        batch_id = uuid4().hex[:12]
        scanned = 0
        eligible_pairs: list[tuple[Turn, Turn]] = []

        for turn in session.turns:
            if turn.role != "assistant":
                continue
            try:
                user_turn, assistant_turn = resolve_turn_pair(session, turn.id)
            except (PairNotFoundError, TurnNotFoundError):
                continue
            scanned += 1
            if not is_memory_eligible_turn_pair(
                user_turn,
                assistant_turn,
                now=now,
                config=self._config,
                for_batch=True,
            ):
                assistant_meta = dict(assistant_turn.meta)
                assistant_meta["consolidation_scanned_at"] = _now_iso()
                assistant_turn.meta = assistant_meta
                continue
            if batch_reject_gate(
                user_turn,
                assistant_turn,
                config=self._config,
                left_store=left_store,
                project=session.project,
                dedup_hours=dedup_hours,
            ):
                assistant_meta = dict(assistant_turn.meta)
                assistant_meta["consolidation_scanned_at"] = _now_iso()
                assistant_turn.meta = assistant_meta
                continue
            eligible_pairs.append((user_turn, assistant_turn))

        candidates = consolidation_gate(
            eligible_pairs,
            config=self._config,
            budget=budget,
            llm=self._router.llm,
            left_store=left_store,
            project=session.project,
        )

        max_candidates = int(consolidation_cfg.get("max_candidates_per_batch", 3))
        ingested_atom_ids: list[str] = []
        gate = PlanePackGate.from_session_meta(session.meta)

        for candidate in candidates[:max_candidates]:
            user_turn = candidate.user_turn
            assistant_turn = candidate.assistant_turn
            if user_turn.meta.get("ingested_atom_id") or assistant_turn.meta.get("ingested_atom_id"):
                continue
            if not gate.allows("P10"):
                continue

            ingest_meta = build_ingest_meta(session.meta)
            ingest_meta["remembered_via"] = "batch"
            ingest_meta["consolidation_batch_id"] = batch_id
            ingest_meta["assistant_turn_id"] = assistant_turn.id
            ingest_meta["user_turn_id"] = user_turn.id
            ingest_meta["dedup_key"] = chat_dedup_key(candidate.text)
            meta_ok, violations = validate_atom_meta(ingest_meta)
            if not meta_ok:
                logger.debug("consolidate meta blocked: %s", violations)
                continue

            atom, _orch_trace = self._orchestrator.ingest(
                candidate.text,
                tags=candidate.tags or ["chat"],
                source="chat",
                project=session.project,
                meta=ingest_meta,
            )
            atom_id = atom.id
            ingested_atom_ids.append(atom_id)

            user_meta = dict(user_turn.meta)
            user_meta["ingested_atom_id"] = atom_id
            user_meta["consolidation_scanned_at"] = _now_iso()
            user_turn.meta = user_meta

            assistant_meta = dict(assistant_turn.meta)
            assistant_meta["ingested_atom_id"] = atom_id
            assistant_meta["consolidation_scanned_at"] = _now_iso()
            assistant_turn.meta = assistant_meta

            oid = session.meta.get("current_object_id")
            if self._object_store is not None and oid:
                self._object_store.register_disclosed_atom(str(oid), atom_id, ingest_meta)

            ids = session.meta.setdefault("ingested_atom_ids", [])
            ids.append(atom_id)

        for user_turn, assistant_turn in eligible_pairs:
            if assistant_turn.meta.get("consolidation_scanned_at"):
                continue
            assistant_meta = dict(assistant_turn.meta)
            assistant_meta["consolidation_scanned_at"] = _now_iso()
            assistant_turn.meta = assistant_meta
            user_meta = dict(user_turn.meta)
            if not user_meta.get("consolidation_scanned_at"):
                user_meta["consolidation_scanned_at"] = _now_iso()
                user_turn.meta = user_meta

        if ingested_atom_ids:
            schedule_push_after_ingest(self._memory_store, self._config)
        session.updated_at = _now_iso()

        return {
            "consolidation_batch_id": batch_id,
            "scanned_pairs": scanned,
            "eligible_pairs": len(eligible_pairs),
            "ingested_count": len(ingested_atom_ids),
            "ingested_atom_ids": ingested_atom_ids,
        }

    def run_turn(
        self,
        session_id: str,
        text: str = "",
        *,
        parts: list[dict[str, Any]] | None = None,
        stream: bool = False,
        show_trace: bool = False,
        project: str | None = None,
        plan_mode: bool | None = None,
    ) -> TurnResult | Iterator[tuple[str, dict[str, Any]]]:
        if parts is not None:
            normalized_parts, content = resolve_message_input("", parts)
        elif text.strip():
            normalized_parts = normalize_text_to_parts(text)
            content = text.strip()
        else:
            raise ValueError("empty message")
        if stream:
            return self._run_turn_stream(
                session_id,
                content,
                normalized_parts=normalized_parts,
                show_trace=show_trace,
                project=project,
                plan_mode=plan_mode,
            )
        return self._run_turn_sync(
            session_id,
            content,
            normalized_parts=normalized_parts,
            show_trace=show_trace,
            project=project,
            plan_mode=plan_mode,
        )

    def _run_turn_sync(
        self,
        session_id: str,
        text: str,
        *,
        normalized_parts: list[dict[str, Any]],
        show_trace: bool,
        project: str | None,
        plan_mode: bool | None = None,
        regenerate_assistant_id: str | None = None,
    ) -> TurnResult:
        events = list(
            self._run_turn_stream(
                session_id,
                text,
                normalized_parts=normalized_parts,
                show_trace=show_trace,
                project=project,
                plan_mode=plan_mode,
                regenerate_assistant_id=regenerate_assistant_id,
            )
        )
        reply_parts: list[str] = []
        trace: list[str] = []
        turn_id = uuid4().hex[:12]
        ingested = False
        atom_id = None
        citations: list[dict[str, Any]] = []
        usage = None
        intent = "chat"
        user_turn_id = None
        user_turn_created_at = None
        rememberable = False
        remember_block_reason = None
        recall_pending = False
        recall_window_remaining_seconds = None

        for event, data in events:
            if event == "_return":
                continue
            if event == "error":
                code = data.get("code")
                if code == "turn_in_progress":
                    raise TurnInProgressError(session_id)
                if code not in ("llm_unavailable",):
                    raise RuntimeError(str(data.get("message") or code or "chat turn failed"))
            if event == "trace":
                trace.extend(data.get("lines", []))
            elif event == "delta":
                reply_parts.append(data.get("text", ""))
            elif event == "done":
                turn_id = data.get("turn_id", turn_id)
                ingested = data.get("ingested", False)
                atom_id = data.get("atom_id")
                citations = data.get("citations", [])
                usage = data.get("usage")
                intent = data.get("intent", intent)
                user_turn_id = data.get("user_turn_id")
                user_turn_created_at = data.get("user_turn_created_at")
                rememberable = bool(data.get("rememberable", False))
                remember_block_reason = data.get("remember_block_reason")
                recall_pending = bool(data.get("recall_pending", False))
                recall_window_remaining_seconds = data.get("recall_window_remaining_seconds")
                trace.extend(data.get("trace", []))
                if data.get("reply"):
                    reply_parts = [data["reply"]]

        return TurnResult(
            turn_id=turn_id,
            reply="".join(reply_parts),
            intent=intent,
            trace=trace,
            ingested=ingested,
            atom_id=atom_id,
            citations=citations,
            usage=usage,
            user_turn_id=user_turn_id,
            user_turn_created_at=user_turn_created_at,
            rememberable=rememberable,
            remember_block_reason=remember_block_reason,
            recall_pending=recall_pending,
            recall_window_remaining_seconds=recall_window_remaining_seconds,
        )

    def _run_turn_stream(
        self,
        session_id: str,
        text: str,
        *,
        normalized_parts: list[dict[str, Any]],
        show_trace: bool,
        project: str | None,
        plan_mode: bool | None = None,
        regenerate_assistant_id: str | None = None,
    ) -> Iterator[tuple[str, dict[str, Any]]]:
        session = self._sessions.require(session_id)
        active_plan = bool(plan_mode) if plan_mode is not None else bool(session.meta.get("plan_mode"))
        scope = project or session.project
        turn_id = uuid4().hex[:12]
        trace: list[str] = []

        try:
            self._sessions.acquire_turn_lock(session_id)
        except TurnInProgressError as exc:
            yield ("error", {"code": "turn_in_progress", "message": str(exc), "recoverable": True})
            yield ("done", {"turn_id": turn_id, "ingested": False, "trace": trace})
            return

        if self._memory_store is not None:
            from niannian.storage.sync.hydrate import ensure_fresh_for_read

            fresh = ensure_fresh_for_read(self._memory_store, config=self._config)
            if fresh is not None and not fresh.ok:
                trace.append(f"[{BRAINSTEM}] 读前同步未完成（继续 Chat）")

        try:
            trace.append(f"[{BRAINSTEM}] {PUBLIC} Chat Turn {turn_id} 开始")
            home = self._data_home()
            chat_user_id = str(session.meta.get("user_id") or "pinn")
            async_enrichment = bool(self._chat_cfg.get("async_enrichment"))
            bus_ctx = InputBusContext(
                user_id=chat_user_id,
                config=self._config,
                home=home,
                llm=self._router.llm,
                async_enrichment=async_enrichment,
                job_store=self._enrichment_jobs() if async_enrichment else None,
                sessions=self._sessions,
            )
            mention_context = ""
            if regenerate_assistant_id:
                user_turn, old_assistant = resolve_turn_pair(session, regenerate_assistant_id)
                if old_assistant.role != "assistant":
                    yield (
                        "error",
                        {"code": "not_assistant", "message": "turn is not assistant", "recoverable": False},
                    )
                    yield ("done", {"turn_id": turn_id, "ingested": False, "trace": trace if show_trace else []})
                    return
                if user_turn.meta.get("recalled_at") or old_assistant.meta.get("recalled_at"):
                    yield (
                        "error",
                        {"code": "turn_recalled", "message": "turn recalled", "recoverable": False},
                    )
                    yield ("done", {"turn_id": turn_id, "ingested": False, "trace": trace if show_trace else []})
                    return
                if old_assistant.meta.get("superseded_by"):
                    yield (
                        "error",
                        {"code": "turn_superseded", "message": "turn already superseded", "recoverable": False},
                    )
                    yield ("done", {"turn_id": turn_id, "ingested": False, "trace": trace if show_trace else []})
                    return
                old_meta = dict(old_assistant.meta)
                old_meta["superseded_by"] = turn_id
                old_assistant.meta = old_meta
                trace.append(f"[{BRAINSTEM}] regenerate {regenerate_assistant_id} -> {turn_id}")
                normalized_parts = list(user_turn.parts or normalize_text_to_parts(user_turn.content))
                text = (user_turn.content or parts_to_content(normalized_parts)).strip()
                enrichments = dict(user_turn.meta.get("enrichments") or {})
                mention_results = user_turn.meta.get("mentions") or {}
                if mention_results:
                    mention_context = format_mention_context(mention_results)
            else:
                user_turn = Turn(
                    id=uuid4().hex[:12],
                    role="user",
                    content=parts_to_content(normalized_parts),
                    created_at=_now_iso(),
                    parts=normalized_parts,
                    meta={},
                )
                self._sessions.append_turn(session_id, user_turn)
                bus_ctx.session_id = session_id
                bus_ctx.turn_id = user_turn.id
                enrichments = process_parts(bus_ctx, normalized_parts)
                user_turn.meta["enrichments"] = enrichments

                mention_refs = parse_mentions(text_from_parts(normalized_parts))
                if mention_refs:
                    vault_search, wiki_search = self._mention_searchers(home)
                    mention_results = resolve_mentions(
                        mention_refs,
                        vault_search=vault_search,
                        wiki_search=wiki_search,
                    )
                    user_turn.meta["mentions"] = mention_results
                    mention_context = format_mention_context(mention_results)
                    vault_n = len(mention_results.get("vault") or [])
                    wiki_n = len(mention_results.get("wiki") or [])
                    if vault_n or wiki_n:
                        trace.append(f"[{BRAINSTEM}] mentions vault={vault_n} wiki={wiki_n}")

            if is_sticker_only_parts(normalized_parts):
                trace.append(f"[{BRAINSTEM}] sticker-only · 跳过 LLM")
                yield (
                    "done",
                    {
                        "turn_id": turn_id,
                        "reply": "",
                        "ingested": False,
                        "trace": trace if show_trace else [],
                        "intent": "chat",
                        "citations": [],
                        "usage": None,
                        "user_turn_id": user_turn.id,
                        "user_turn_created_at": user_turn.created_at,
                        "rememberable": False,
                        "recall_pending": False,
                    },
                )
                return

            history = self._history_messages(session)
            thalamus = ThalamusGate(self._config)
            normalized = thalamus.normalize(text, source="chat", session_meta=session.meta)
            text = normalized.text
            trace.append(thalamus.trace_line(normalized))
            recent_user_texts = collect_recent_user_texts(session.turns)
            token_usage_user_id.set(chat_user_id)
            token_usage_operation.set("chat")
            if home is not None:
                chat_budget = load_token_budget(home, chat_user_id)
            else:
                chat_budget = {**DEFAULT_BOOL_BUDGET, **DEFAULT_LEVEL_BUDGET}
            regime_llm = self._router.llm if chat_llm_allows(chat_budget, "auxiliary") else None
            resolved = resolve_turn_policy(
                text,
                history_len=len(history) // 2,
                session_meta=session.meta,
                config=self._config,
                recent_user_texts=recent_user_texts,
                chat_budget=chat_budget,
                llm=regime_llm,
            )
            policy = resolved.policy
            persona_block = ""
            character_block = ""
            if home is not None:
                recall_ctx_early = RecallContext.from_session_meta(session.meta)
                object_id = recall_ctx_early.object_id
                equipped_vp_id = session.meta.get("equipped_vp_id")
                if equipped_vp_id is not None and str(equipped_vp_id).strip():
                    if not budget_allows(chat_budget, "virtual_persona_enabled"):
                        trace.append(f"[{JUDGMENT}] vp equip blocked (token budget)")
                    else:
                        try:
                            vp_id = str(equipped_vp_id).strip()
                            vp_store = VirtualPersonaStore(home, chat_user_id)
                            vp_max_chars = resolve_vp_character_max_chars(self._chat_cfg)
                            character_block, equip_trace = build_equipped_vp_character_block(
                                vp_store,
                                vp_id,
                                max_chars=vp_max_chars,
                            )
                            if self._skills is not None:
                                character_block, skill_trace = append_vp_persona_skills(
                                    character_block,
                                    vp_store,
                                    vp_id,
                                    self._skills,
                                    max_total_chars=vp_max_chars,
                                )
                                equip_trace.extend(skill_trace)
                            if character_block:
                                user_address = resolve_user_address_for_vp(
                                    vp_store,
                                    vp_id,
                                    chat_user_id,
                                    data_home=home,
                                )
                                character_block = append_user_address_line(character_block, user_address)
                                equip_trace.append(f"vp user address {user_address}")
                            trace.append(f"[{JUDGMENT}] vp equip {vp_id}")
                            for line in equip_trace:
                                trace.append(f"[{JUDGMENT}] {line}")
                        except FileNotFoundError:
                            trace.append(f"[{JUDGMENT}] vp missing {equipped_vp_id}")
                persona_pack = PersonaResolver(
                    self._config,
                    home=home,
                    user_id=chat_user_id,
                ).resolve(
                    session.meta,
                    object_id=object_id,
                )
                if not persona_pack.detached:
                    merged_planes = merge_planes_bias(
                        policy.planes,
                        persona_pack.planes_bias,
                        required=frozenset({"P1", "P2", "P3"}),
                    )
                    policy = TurnPolicy(
                        regime=policy.regime,
                        intent=policy.intent,
                        planes=merged_planes,
                        force_ingest=policy.force_ingest,
                        confidence=policy.confidence,
                        values_gate=policy.values_gate,
                        regime_source=policy.regime_source,
                    )
                    persona_block = build_persona_assembly_block(persona_pack)
                    session.meta["persona_preset"] = persona_pack.preset_id
                    trace.append(f"[{JUDGMENT}] persona:{persona_pack.preset_id}")
                else:
                    session.meta["persona_preset"] = None
                    trace.append(f"[{JUDGMENT}] persona:detach")
            session.meta = update_pathway_sticky(
                session.meta,
                policy,
                config=self._config,
                persona_preset=session.meta.get("persona_preset"),
            )
            session.meta["pathway_tier"] = resolved.tier
            if resolved.route_id:
                session.meta["pathway_route"] = resolved.route_id
            session.meta["intent"] = policy.intent
            session.meta["regime"] = policy.regime
            session.meta["turn_policy"] = policy.to_dict()
            track_id = f"chat:{session_id}"
            if self._track_registry is not None:
                self._track_registry.register(track_id, label="Chat", priority="P0")
            regime_suffix = f" ({policy.regime_source})" if policy.regime_source else ""
            trace.append(
                f"[{JUDGMENT}] regime={policy.regime}{regime_suffix} "
                f"intent={policy.intent} pathway={resolved.tier}"
            )
            intent_result = IntentResult(
                intent=policy.intent,
                force_ingest=policy.force_ingest,
                confidence=policy.confidence,
            )

            if self._executive is not None and self._task_state is not None:
                track_id = f"chat:{session_id}"
                graph = None
                existing = self._task_state.get_active_graph(track_id)
                if existing is not None and not existing.is_complete():
                    graph = existing
                    trace.append(self._executive.trace_line(f"reuse TaskGraph {graph.task_id}"))
                else:
                    graph = self._executive.plan_from_message(text, policy, plan_mode=active_plan)
                    if graph is not None:
                        self._task_state.register_track(track_id, label="Chat 工作任务", regime=policy.regime)
                        self._task_state.set_active_task(track_id, graph)
                        trace.append(self._executive.trace_line(f"TaskGraph {graph.task_id} steps={len(graph.steps)}"))
                    elif policy.regime == "companion" and not active_plan:
                        self._task_state.set_active_task(track_id, None)
                if graph is not None:
                    yield ("plan", graph.to_dict())

            if intent_result.intent == "device":
                gate = PlanePackGate(policy.planes)
                action = parse_device_command(text)
                if not gate.allows("P12"):
                    trace.append(gate.trace_skip("P12"))
                    reply = "（P12 设备控制已跳过）"
                elif action is None or self._device is None:
                    reply = "用法：/device open_app <应用名> 或 /device run_shortcut <快捷指令名>"
                else:
                    try:
                        result = self._device.execute(action, confirmed=False)
                        reply = result.message
                        trace.append(f"[{BRAINSTEM}] device {result.action_type} ok={result.ok}")
                    except MacRequiredError:
                        reply = "设备控制需在 Mac 念念上执行；Docker 环境请通过 Bridge 转发。"
                    except NeedConfirmationError as exc:
                        reply = f"需要确认后执行：{exc.action}（请 POST /v1/device/execute confirmed=true）"
                yield from self._finalize(
                    session_id,
                    session,
                    turn_id,
                    text,
                    reply,
                    intent_result,
                    trace,
                    show_trace,
                    chat_budget=chat_budget,
                )
                return

            gate = PlanePackGate(policy.planes)

            citations: list[dict[str, Any]] = []
            usage: dict[str, Any] | None = None
            reply = ""

            if intent_result.intent == "memory_query":
                question = text.removeprefix("/query").strip() or text
                recall_ctx = RecallContext.from_session_meta(session.meta)
                if gate.allows("P9") and self._recall is not None:
                    query_llm = self._router.llm if chat_llm_allows(chat_budget, "auxiliary") else None
                    if query_llm is None:
                        trace.append(f"[{JUDGMENT}] memory_query LLM 跳过（Token 页 · 聊天 LLM 非重度）")
                    query_result = self._recall.query(
                        question, project=scope, llm=query_llm, for_llm=False,
                        recall_context=recall_ctx,
                    )
                    trace.extend(query_result.trace)
                    citations = list(query_result.citations)
                    reply = query_result.answer
                else:
                    trace.append(gate.trace_skip("P9"))
                    reply = "（P9 召回已跳过）"
                    citations = []
                if show_trace:
                    yield ("trace", _trace_sse_payload(trace))
                for chunk in self._chunk_text(reply):
                    yield ("delta", {"text": chunk})
            else:
                recall_ctx = RecallContext.from_session_meta(session.meta)
                if gate.allows("P9") and self._recall is not None:
                    assist_result = self._recall.assist(
                        text,
                        context=self._history_summary(session),
                        project=scope,
                        for_llm=True,
                        recall_context=recall_ctx,
                    )
                    trace.extend(assist_result.trace)
                    citations = list(assist_result.citations)
                    for concept in assist_result.wiki_links:
                        if concept:
                            citations.append(
                                {
                                    "kind": "wiki",
                                    "snippet": str(concept),
                                    "reason": "wiki_hit",
                                }
                            )
                else:
                    trace.append(gate.trace_skip("P9"))
                    assist_result = AssistResult(
                        context_block="", fragments=[], wiki_links=[], citations=[], trace=[],
                    )
                if show_trace:
                    yield ("trace", _trace_sse_payload(trace))

                skill_block = ""
                if self._skills is not None and gate.allows("P12"):
                    chat_mode = str(session.meta.get("chat_mode") or "auto")
                    skill_block = inject_skills(
                        self._skills,
                        "chat",
                        context=SkillInjectContext(mode=chat_mode),
                    )
                    if skill_block:
                        trace.append(f"[{BRAINSTEM}] skills 注入 {len(skill_block)} 字符")
                task_block = ""
                if self._task_state is not None:
                    task_block = self._task_state.format_task_block()
                subject_block = ""
                if self._subject_store is not None:
                    subject_block = build_subject_assembly_blocks(self._subject_store.load_subject_pack())
                object_block = ""
                if recall_ctx.mode == "object" and recall_ctx.object_id:
                    snippets = list(assist_result.fragments) if assist_result.fragments else []
                    object_block = build_object_assembly_block(recall_ctx.object_id, snippets=snippets)
                non_text_parts = [p for p in normalized_parts if p.get("type") != "text"]
                enrichment_block = build_llm_context_block(non_text_parts, enrichments)
                llm_user_text = text
                if enrichment_block:
                    llm_user_text = (
                        f"{text}\n{enrichment_block}".strip()
                        if text.strip()
                        else enrichment_block
                    )
                if mention_context:
                    llm_user_text = (
                        f"{llm_user_text}\n{mention_context}".strip()
                        if llm_user_text.strip()
                        else mention_context
                    )
                memory_block = assist_result.context_block
                if session.tool_scope and home is not None:
                    external_block = _build_external_tool_block(
                        home,
                        self._config,
                        session.tool_scope,
                        text,
                    )
                    if external_block:
                        trace.append(
                            f"[{BRAINSTEM}] tool_scope external 注入 {len(external_block)} 字符"
                        )
                        memory_block = (
                            f"{memory_block}\n\n{external_block}".strip()
                            if memory_block.strip()
                            else external_block
                        )
                messages = self._right.build_chat_messages(
                    llm_user_text,
                    memory_block=memory_block,
                    history=self._history_messages(session),
                    skill_block=skill_block,
                    task_block=task_block,
                    subject_block=subject_block,
                    persona_block=persona_block,
                    object_block=object_block,
                    character_block=character_block,
                )
                for event, data in self._emit_chat_deltas(messages, trace, chat_budget):
                    if event == "_return":
                        reply = data["reply"]
                        usage = data.get("usage")
                    else:
                        yield (event, data)

            yield from self._finalize(
                session_id,
                session,
                turn_id,
                text,
                reply,
                intent_result,
                trace,
                show_trace,
                citations=citations,
                usage=usage,
                chat_budget=chat_budget,
                regenerated_from=regenerate_assistant_id,
            )
        except Exception as exc:
            logger.exception("chat turn failed session=%s", session_id)
            yield (
                "error",
                {
                    "code": "turn_failed",
                    "message": str(exc) or "chat turn failed",
                    "recoverable": True,
                },
            )
            yield ("done", {"turn_id": turn_id, "ingested": False, "trace": trace if show_trace else []})
        finally:
            self._sessions.release_turn_lock(session_id)

    def _emit_chat_deltas(
        self,
        messages: list[Message],
        trace: list[str],
        chat_budget: dict[str, Any],
    ) -> Iterator[tuple[str, dict[str, Any]]]:
        llm = self._router.llm
        reply_parts: list[str] = []
        usage: dict[str, Any] | None = None

        if not chat_llm_allows(chat_budget, "reply"):
            fallback = self._right.fallback_chat_reply(messages)
            trace.append(self._right.trace_line("聊天 LLM 已关闭（Token 页）· 规则回复"))
            for chunk in self._chunk_text(fallback):
                yield ("delta", {"text": chunk})
            yield ("_return", {"reply": fallback, "usage": None})
            return

        if not llm.is_module_configured("right"):
            fallback = self._right.fallback_chat_reply(messages)
            trace.append(self._right.trace_line("未配置 LLM，规则降级"))
            for chunk in self._chunk_text(fallback):
                yield ("delta", {"text": chunk})
            yield ("_return", {"reply": fallback, "usage": None})
            return

        router_trace = RouterTrace()
        try:
            for chunk in llm.stream_complete("right", messages, trace=router_trace):
                if chunk.text:
                    reply_parts.append(chunk.text)
                    yield ("delta", {"text": chunk.text})
                if chunk.done and chunk.usage:
                    usage = {
                        "prompt_tokens": chunk.usage.prompt_tokens,
                        "completion_tokens": chunk.usage.completion_tokens,
                    }
            trace.extend(router_trace.lines)
            yield ("_return", {"reply": "".join(reply_parts), "usage": usage})
        except LLMUnavailable:
            memory = next((m.content for m in messages if "记忆包" in m.content), "")
            fallback = self._right.fallback_chat_reply(messages, memory_block=memory)
            trace.append(self._right.trace_line("云端暂时不可用，降级规则回复"))
            yield ("error", {"code": "llm_unavailable", "message": "云端暂时不可用", "recoverable": True})
            for chunk in self._chunk_text(fallback):
                yield ("delta", {"text": chunk})
            yield ("_return", {"reply": fallback, "usage": None})

    def _finalize(
        self,
        session_id: str,
        session: Session,
        turn_id: str,
        user_text: str,
        reply: str,
        intent_result: IntentResult,
        trace: list[str],
        show_trace: bool,
        *,
        citations: list[dict[str, Any]] | None = None,
        usage: dict[str, Any] | None = None,
        chat_budget: dict[str, Any] | None = None,
        regenerated_from: str | None = None,
    ) -> Iterator[tuple[str, dict[str, Any]]]:
        project = session.project
        reply = self._export_text(reply, project=project)
        if citations and self._decode_on_export():
            citations = self._redactor.decode_citations(citations, project=project)

        user_turn: Turn | None = None
        for turn in reversed(session.turns):
            if turn.role == "user" and not turn.meta.get("recalled_at"):
                user_turn = turn
                break

        budget = chat_budget or {**DEFAULT_BOOL_BUDGET, **DEFAULT_LEVEL_BUDGET}
        assistant_meta: dict[str, Any] = {"intent": intent_result.intent}
        if regenerated_from:
            assistant_meta["regenerated_from"] = regenerated_from
        if user_turn is not None:
            assistant_meta["user_turn_id"] = user_turn.id

        effective_user_text = user_text
        if user_turn is not None and not effective_user_text.strip():
            effective_user_text = parts_to_content(user_turn.parts or []) or user_turn.content

        chat_mode = str(session.meta.get("chat_mode") or "auto")
        sticker_pick = None
        if chat_sticker_allows(budget, "assistant") and user_turn is not None:
            home = self._data_home()
            chat_user_id = str(session.meta.get("user_id") or "pinn")
            equipped_vp_id = session.meta.get("equipped_vp_id")
            vp_id = str(equipped_vp_id).strip() if equipped_vp_id is not None else None
            sticker_pick = pick_assistant_sticker(
                user_text=effective_user_text,
                reply=reply,
                chat_mode=chat_mode,
                level=resolve_chat_sticker_level(budget),
                regime=str(session.meta.get("regime") or ""),
                persona_preset=str(session.meta.get("persona_preset") or "") or None,
                vp_id=vp_id or None,
                home=home,
                user_id=chat_user_id,
            )
            if sticker_pick:
                assistant_meta["sticker_id"] = sticker_pick.sticker_id
                pack = str(sticker_pick.pack or "")
                assistant_meta["sticker_pack"] = pack if pack.startswith("vp:") else STICKER_PACK_ID
                assistant_meta["label"] = sticker_pick.label
                assistant_meta["sticker_proactive"] = True
                assistant_meta["sticker_rule"] = sticker_pick.rule
                assistant_meta["sticker_trigger"] = sticker_pick.trigger
                if sticker_pick.sticker_only:
                    reply = ""

        assistant_turn = Turn(
            id=turn_id,
            role="assistant",
            content=reply,
            created_at=_now_iso(),
            meta=assistant_meta,
        )
        if user_turn is not None:
            user_meta = dict(user_turn.meta)
            user_meta["assistant_turn_id"] = turn_id
            user_turn.meta = user_meta
        self._sessions.append_turn(session_id, assistant_turn)

        if sticker_pick and assistant_meta.get("sticker_id"):
            yield (
                "sticker",
                {
                    "sticker_id": assistant_meta["sticker_id"],
                    "label": assistant_meta.get("label"),
                    "sticker_only": sticker_pick.sticker_only,
                },
            )

        chat_cfg = self._config.get("chat", {})
        auto_ingest = bool(chat_cfg.get("auto_ingest", False))
        vp_trial = skip_vp_l2_ingest(session.meta)
        if vp_trial:
            decision = IngestDecision(should_ingest=False, skip_reason="vp_trial_no_l2")
        elif not auto_ingest and not intent_result.force_ingest:
            decision = IngestDecision(should_ingest=False, skip_reason="manual_only_policy")
        else:
            ingest_llm = self._router.llm if chat_llm_allows(budget, "auxiliary") else None
            decision = ingest_gate(
                user_text,
                reply,
                config=self._config,
                llm=ingest_llm,
                project=session.project,
                force=intent_result.force_ingest,
            )
        gate = PlanePackGate.from_session_meta(session.meta)
        if not gate.allows("P10") and decision.should_ingest:
            trace.append(gate.trace_skip("P10"))
            decision = IngestDecision(
                should_ingest=False,
                skip_reason="P10_skipped",
                confidence=decision.confidence,
            )
        ingested = False
        atom_id = None
        if decision.should_ingest and decision.text:
            ingest_meta = build_ingest_meta(session.meta)
            meta_ok, violations = validate_atom_meta(ingest_meta)
            if not meta_ok:
                trace.append(firewall_trace_line(f"ingest blocked: {','.join(violations)}"))
            else:
                atom, orch_trace = self._orchestrator.ingest(
                    decision.text,
                    tags=decision.tags or ["chat"],
                    source="chat",
                    project=session.project,
                    meta=ingest_meta,
                )
                ingested = True
                atom_id = atom.id
                if user_turn is not None:
                    user_meta = dict(user_turn.meta)
                    user_meta["ingested_atom_id"] = atom_id
                    user_turn.meta = user_meta
                trace.extend(orch_trace)
                oid = session.meta.get("current_object_id")
                if self._object_store is not None and oid:
                    registered = self._object_store.register_disclosed_atom(str(oid), atom.id, ingest_meta)
                    if registered:
                        trace.append(ingest_gate_trace(f"OT1 索引 object={oid} atom #{atom.id}"))
                    elif ingest_meta.get("memory_scope") == "object":
                        trace.append(ingest_gate_trace("OT1 跳过：无 disclosure.told_to"))
                ids = session.meta.setdefault("ingested_atom_ids", [])
                ids.append(atom.id)
                trace.append(
                    ingest_gate_trace(
                        f"ingest_gate confidence={decision.confidence:.2f} → atom #{atom.id}"
                    )
                )
                schedule_push_after_ingest(self._memory_store, self._config)
        elif decision.skip_reason:
            trace.append(ingest_gate_trace(f"ingest_gate 跳过: {decision.skip_reason}"))

        equipped_vp_id = session.meta.get("equipped_vp_id")
        vp_id_for_distill = str(equipped_vp_id).strip() if equipped_vp_id is not None else ""
        if vp_id_for_distill and not vp_trial and reply.strip():
            try:
                home = self._data_home()
                chat_user_id = str(session.meta.get("user_id") or "pinn")
                vp_store = VirtualPersonaStore(home, chat_user_id)
                if maybe_distill_turn(
                    vp_store,
                    vp_id_for_distill,
                    user_text=effective_user_text,
                    reply=reply,
                ):
                    trace.append(f"[{JUDGMENT}] vp incremental distill → Daily notes")
            except FileNotFoundError:
                pass

        if not vp_trial and sticker_pick and sticker_learn_allowed(budget, trigger=sticker_pick.trigger):
            learn_text = build_sticker_learn_text(
                effective_user_text,
                sticker_pick.label,
                pack=sticker_pick.pack,
                trigger=sticker_pick.trigger,
            )
            learn_meta = build_ingest_meta(session.meta)
            learn_meta["sticker_learn"] = True
            learn_meta["sticker_label"] = sticker_pick.label
            meta_ok, violations = validate_atom_meta(learn_meta)
            if meta_ok and gate.allows("P10"):
                learn_atom, learn_trace = self._orchestrator.ingest(
                    learn_text,
                    tags=["sticker_learn", "proactive", "chat"],
                    source="chat",
                    project=session.project,
                    meta=learn_meta,
                )
                trace.extend(learn_trace)
                trace.append(ingest_gate_trace(f"sticker_learn → atom #{learn_atom.id} label={sticker_pick.label}"))
                schedule_push_after_ingest(self._memory_store, self._config)
            elif not meta_ok:
                trace.append(firewall_trace_line(f"sticker_learn blocked: {','.join(violations)}"))
            else:
                trace.append(gate.trace_skip("P10"))

        if self._executive is not None and self._task_state is not None and reply.strip():
            track_id = f"chat:{session_id}"
            graph = self._task_state.get_active_graph(track_id)
            if graph is not None and not graph.is_complete():
                updated = self._executive.observe_turn(graph, reply)
                self._task_state.set_active_task(track_id, updated)
                done, total = updated.progress()
                trace.append(self._executive.trace_line(f"observe {done}/{total}"))

        if self._subject_store is not None and should_update_subject_emotion(session.meta):
            self._subject_store.save_emotion({"updated_at": _now_iso()})

        recall_window = float(chat_cfg.get("recall_window_seconds", 120))
        block_reason = (
            remember_gate(user_turn, assistant_turn, config=self._config)
            if user_turn is not None
            else "pair_not_found"
        )
        recall_pending, recall_remaining = recall_pending_fields(
            assistant_turn,
            recall_window_seconds=recall_window,
        )
        rememberable = block_reason is None and not recall_pending

        done_payload: dict[str, Any] = {
            "turn_id": turn_id,
            "user_turn_id": user_turn.id if user_turn else None,
            "user_turn_created_at": user_turn.created_at if user_turn else None,
            "reply": reply,
            "ingested": ingested,
            "atom_id": atom_id,
            "trace": trace if show_trace else [],
            "intent": intent_result.intent,
            "citations": citations or [],
            "usage": usage,
            "rememberable": rememberable,
            "recall_pending": recall_pending,
        }
        if assistant_meta.get("sticker_id"):
            done_payload["sticker_id"] = assistant_meta["sticker_id"]
            done_payload["sticker_label"] = assistant_meta.get("label")
            done_payload["sticker_only"] = bool(sticker_pick and sticker_pick.sticker_only)
        if block_reason:
            done_payload["remember_block_reason"] = block_reason
        if recall_pending and recall_remaining is not None:
            done_payload["recall_window_remaining_seconds"] = recall_remaining
        yield ("done", done_payload)

    def _history_messages(self, session: Session) -> list[Message]:
        max_turns = int(self._chat_cfg.get("max_history_turns", 20))
        turns = session.turns[-max_turns * 2 :]
        out: list[Message] = []
        for t in turns:
            if t.role not in ("user", "assistant"):
                continue
            if t.meta.get("recalled_at") or t.meta.get("superseded_by"):
                continue
            text = self._turn_text_for_history(t)
            out.append(Message(role=t.role, content=text))
        return out

    @staticmethod
    def _turn_text_for_history(turn: Turn) -> str:
        sticker_id = turn.meta.get("sticker_id") if turn.meta else None
        if sticker_id:
            label = (turn.meta or {}).get("label") or sticker_id
            return f"[表情包:{label}]"
        return turn.content

    def _history_summary(self, session: Session) -> str:
        lines: list[str] = []
        for turn in session.turns[-6:]:
            if turn.role in ("user", "assistant") and not turn.meta.get("recalled_at") and not turn.meta.get("superseded_by"):
                prefix = "用户" if turn.role == "user" else "念念"
                lines.append(f"{prefix}: {self._turn_text_for_history(turn)[:200]}")
        return "\n".join(lines)

    @staticmethod
    def _chunk_text(text: str, size: int = 12) -> list[str]:
        return [text[i : i + size] for i in range(0, len(text), size)] or [""]
