from __future__ import annotations

from typing import TYPE_CHECKING, Any

from fastapi import APIRouter, HTTPException, Request
from fastapi.responses import Response, StreamingResponse
from pydantic import BaseModel, Field

from niannian.chat.session import (
    RecallWindowExpiredError,
    SessionNotFoundError,
    TurnInProgressError,
    TurnNotFoundError,
    TurnNotRecallableError,
    _now_iso,
)
from niannian.chat.session import (
    AlreadyRememberedError,
    ConsolidationNotBatchModeError,
    EnrichmentAlreadyPersistedError,
    EnrichmentKindNotSupportedError,
    EnrichmentNotActionableError,
    EnrichmentNotFoundError,
    NotRememberableError,
    PairNotFoundError,
    RecallWindowActiveError,
    SessionNotIdleError,
    TurnRecalledError,
    TurnAlreadySupersededError,
    NotAssistantTurnError,
)
from niannian.chat.parts import resolve_message_input
from niannian.chat.stickers import STICKER_PACK_ID, list_stickers, sticker_pack_for_chat_mode, validate_sticker_id
from niannian.chat.stickers_catalog import PACK_LABELS, resolve_sticker_id
from niannian.chat.stream import format_sse_event

from niannian.token.budget import budget_allows, chat_sticker_allows, load_token_budget
from niannian.virtual_persona.chat_mode_gate import (
    VpChatModeForbiddenError,
    assert_session_chat_mode_allowed_for_vp,
    assert_session_ready_for_vp_equip,
)
from niannian.virtual_persona.store import VirtualPersonaStore

if TYPE_CHECKING:
    from niannian.client import NianNian


class CreateSessionBody(BaseModel):
    project: str = "work"


class SendMessageBody(BaseModel):
    text: str = ""
    parts: list[dict] | None = None
    stream: bool = True
    show_trace: bool = True
    project: str | None = None
    plan_mode: bool | None = None


class SendStickerBody(BaseModel):
    sticker_id: str


class RegenerateBody(BaseModel):
    stream: bool = True
    show_trace: bool = True
    plan_mode: bool | None = None


class SessionMetaPatchBody(BaseModel):
    chat_mode: str | None = None
    regime_lock: str | None = None
    current_object_id: str | None = None
    clear_pathway_sticky: bool = False
    persona_override: dict | None = None
    memory_space: str | None = None
    plan_mode: bool | None = None
    tool_scope: dict | None = None


class EquipVirtualPersonaBody(BaseModel):
    vp_id: str


class ProactiveBody(BaseModel):
    scenario: str | None = None
    local_hour: int | None = None
    client_delay_ms: int | None = None
    stream: bool = True


class ProactiveFeedbackBody(BaseModel):
    rating: str = Field(pattern="^(quiet|ok|more)$")
    reason: str | None = None


def create_chat_router(nn: NianNian) -> APIRouter:
    router = APIRouter(prefix="/v1/chat", tags=["chat"])
    pipeline = nn.chat
    from niannian.chat.proactive_pipeline import (
        ProactiveForbidden,
        ProactivePipeline,
        ProactiveRateLimited,
        ProactiveRequest,
    )

    proactive = ProactivePipeline(nn)

    from niannian.soul.reflect_loop import ReflectLoop
    from niannian.soul.search_providers import build_search_provider_from_yaml
    from niannian.soul.web_learning import WebLearningService
    from niannian.soul.web_learning_runner import WebLearningRunner
    from niannian.vault.import_service import DuplicateImportError, VaultImportService
    from niannian.vault.store import VaultStore
    from niannian.vault.taxonomy import TaxonomyService

    vault_store = VaultStore(nn._home, manifest=getattr(nn._store, "manifest", None), config=nn._config)
    vault_taxonomy = TaxonomyService(vault_store, home=nn._home)
    vault_importer = VaultImportService(vault_store, ingest_fn=nn.ingest, taxonomy=vault_taxonomy)
    web_svc = WebLearningService(nn._subject_store, nn._config, home=nn._home)

    def _run_web_batch(*, topic_id: str | None, query: str | None, manual: bool) -> dict[str, Any]:
        yaml_cfg = web_svc.load_yaml_config()
        search = build_search_provider_from_yaml(yaml_cfg)
        loop = ReflectLoop(nn._subject_store, config=nn._config)
        runner = WebLearningRunner(
            web_svc,
            reflect_loop=loop,
            ingest_fn=nn.ingest,
            config=nn._config,
        )
        return runner.run_batch(
            search=search,
            topic_id=topic_id,
            query=query,
            project=nn.project,
            manual=manual,
        )

    @router.get("/personas")
    def list_chat_personas(request: Request) -> dict[str, Any]:
        from niannian.cognitive.persona_registry import persona_registry_for_user
        from niannian.tenant.context import resolve_request_context

        ctx = resolve_request_context(request, nn._config, home=nn._home)
        reg = persona_registry_for_user(nn._home, ctx.user_id, nn._config)
        presets_out: dict[str, dict[str, str]] = {}
        for preset_id, row in reg.list_presets().items():
            if isinstance(row, dict):
                presets_out[str(preset_id)] = {"label": str(row.get("label") or preset_id)}
            else:
                presets_out[str(preset_id)] = {"label": str(preset_id)}
        return {"user_id": ctx.user_id, "presets": presets_out}

    @router.get("/stickers/niannian")
    def list_niannian_stickers(request: Request, pack: str | None = None) -> dict[str, Any]:
        from niannian.tenant.context import resolve_request_context

        ctx = resolve_request_context(request, nn._config, home=nn._home)
        budget = load_token_budget(nn._home, ctx.user_id)
        if not chat_sticker_allows(budget, "user"):
            raise HTTPException(
                status_code=403,
                detail={"code": "token_budget_sticker_disabled", "message": "聊天表情包已在 Token 页关闭"},
            )
        return {
            "pack": STICKER_PACK_ID,
            "packs": [{"id": pid, "label": PACK_LABELS[pid]} for pid in PACK_LABELS],
            "stickers": list_stickers(pack=pack),
        }

    @router.post("/sessions", status_code=201)
    def create_session(payload: CreateSessionBody):
        from niannian.memory.space import ensure_memory_space

        space = ensure_memory_space(nn._home, payload.project)
        session = pipeline.create_session(space)
        return {"id": session.id, "project": session.project, "created_at": session.created_at}

    @router.get("/sessions/{session_id}")
    def get_session(session_id: str):
        try:
            session = pipeline.sessions.require(session_id)
        except SessionNotFoundError:
            raise HTTPException(status_code=404, detail={"code": "session_not_found"})
        return session.to_dict()

    @router.get("/sessions/{session_id}/messages")
    def list_messages(session_id: str):
        try:
            turns = pipeline.sessions.list_turns(session_id)
        except SessionNotFoundError:
            raise HTTPException(status_code=404, detail={"code": "session_not_found"})
        return {"session_id": session_id, "messages": [t.to_dict() for t in turns]}

    @router.post("/sessions/{session_id}/stickers")
    def send_sticker(session_id: str, payload: SendStickerBody):
        try:
            turn = pipeline.send_sticker(session_id, payload.sticker_id)
        except SessionNotFoundError:
            raise HTTPException(status_code=404, detail={"code": "session_not_found"})
        except ValueError as exc:
            msg = str(exc)
            if "disabled by token budget" in msg:
                raise HTTPException(
                    status_code=403,
                    detail={"code": "token_budget_sticker_disabled", "message": "聊天表情包已在 Token 页关闭"},
                ) from exc
            raise HTTPException(
                status_code=400,
                detail={"code": "invalid_sticker_id", "message": msg},
            ) from exc
        return {"session_id": session_id, "turn": turn.to_dict()}

    @router.delete("/sessions/{session_id}", status_code=204)
    def delete_session(session_id: str):
        if not pipeline.sessions.delete(session_id):
            raise HTTPException(status_code=404, detail={"code": "session_not_found"})
        return None

    @router.patch("/sessions/{session_id}/meta")
    def patch_session_meta(session_id: str, payload: SessionMetaPatchBody):
        try:
            updates = payload.model_dump(exclude_unset=True)
            memory_space = updates.pop("memory_space", None)
            session = pipeline.sessions.patch_meta(session_id, updates)
            if memory_space is not None:
                from niannian.memory.space import ensure_memory_space

                session.project = ensure_memory_space(nn._home, str(memory_space))
                session.updated_at = _now_iso()
        except SessionNotFoundError:
            raise HTTPException(status_code=404, detail={"code": "session_not_found"})
        except VpChatModeForbiddenError as exc:
            raise HTTPException(
                status_code=403,
                detail={"code": "vp_chat_mode_forbidden", "message": str(exc)},
            ) from exc
        except ValueError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "invalid_meta", "message": str(exc)},
            ) from exc
        return {
            "id": session.id,
            "meta": session.meta,
            "project": session.project,
            "tool_scope": session.tool_scope,
        }

    @router.post("/sessions/{session_id}/equip")
    def equip_virtual_persona(session_id: str, payload: EquipVirtualPersonaBody):
        try:
            session = pipeline.sessions.require(session_id)
        except SessionNotFoundError:
            raise HTTPException(status_code=404, detail={"code": "session_not_found"})
        user_id = str(session.meta.get("user_id") or "pinn")
        budget = load_token_budget(nn._home, user_id)
        if not budget_allows(budget, "virtual_persona_enabled"):
            raise HTTPException(
                status_code=403,
                detail={"code": "token_budget_vp_disabled", "message": "虚拟人物已在 Token 页关闭"},
            )
        store = VirtualPersonaStore(nn._home, user_id)
        try:
            vp = store.get(payload.vp_id)
        except FileNotFoundError as exc:
            raise HTTPException(
                status_code=404,
                detail={"code": "vp_not_found", "message": str(exc)},
            ) from exc
        except ValueError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "invalid_vp_id", "message": str(exc)},
            ) from exc
        try:
            assert_session_ready_for_vp_equip(session.meta)
        except VpChatModeForbiddenError as exc:
            raise HTTPException(
                status_code=403,
                detail={"code": "vp_chat_mode_forbidden", "message": str(exc)},
            ) from exc
        session.meta["equipped_vp_id"] = vp["id"]
        session.meta["vp_equipped_at"] = _now_iso()
        session.updated_at = _now_iso()
        return {
            "session_id": session_id,
            "equipped_vp_id": vp["id"],
            "label": vp.get("label"),
            "meta": session.meta,
        }

    @router.post("/sessions/{session_id}/unequip")
    def unequip_virtual_persona(session_id: str):
        try:
            session = pipeline.sessions.require(session_id)
        except SessionNotFoundError:
            raise HTTPException(status_code=404, detail={"code": "session_not_found"})
        session.meta.pop("equipped_vp_id", None)
        session.meta.pop("vp_equipped_at", None)
        session.updated_at = _now_iso()
        return {"session_id": session_id, "meta": session.meta}

    @router.post("/sessions/{session_id}/messages")
    def send_message(session_id: str, payload: SendMessageBody):
        try:
            pipeline.sessions.require(session_id)
        except SessionNotFoundError:
            raise HTTPException(status_code=404, detail={"code": "session_not_found"})

        try:
            resolve_message_input(payload.text, payload.parts)
        except ValueError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "invalid_message", "message": str(exc)},
            ) from exc

        if payload.stream:
            return StreamingResponse(
                _sse_generator(pipeline, session_id, payload),
                media_type="text/event-stream",
                headers={
                    "Cache-Control": "no-cache",
                    "Connection": "keep-alive",
                    "X-Accel-Buffering": "no",
                },
            )

        try:
            result = pipeline.run_turn(
                session_id,
                payload.text,
                parts=payload.parts,
                stream=False,
                show_trace=payload.show_trace,
                project=payload.project,
            )
        except TurnInProgressError:
            raise HTTPException(status_code=409, detail={"code": "turn_in_progress"})
        except Exception as exc:
            raise HTTPException(
                status_code=500,
                detail={"code": "turn_failed", "message": str(exc) or "chat turn failed"},
            )
        return result.to_dict()

    @router.post("/sessions/{session_id}/turns/{assistant_turn_id}/regenerate")
    def regenerate_assistant_turn(session_id: str, assistant_turn_id: str, payload: RegenerateBody):
        try:
            pipeline.sessions.require(session_id)
        except SessionNotFoundError:
            raise HTTPException(status_code=404, detail={"code": "session_not_found"})

        if payload.stream:
            return StreamingResponse(
                _sse_regenerate_generator(pipeline, session_id, assistant_turn_id, payload),
                media_type="text/event-stream",
                headers={
                    "Cache-Control": "no-cache",
                    "Connection": "keep-alive",
                    "X-Accel-Buffering": "no",
                },
            )

        try:
            result = pipeline.regenerate_turn(
                session_id,
                assistant_turn_id,
                stream=False,
                show_trace=payload.show_trace,
                plan_mode=payload.plan_mode,
            )
        except TurnNotFoundError:
            raise HTTPException(status_code=404, detail={"code": "turn_not_found"})
        except PairNotFoundError:
            raise HTTPException(status_code=404, detail={"code": "pair_not_found"})
        except NotAssistantTurnError:
            raise HTTPException(status_code=400, detail={"code": "not_assistant"})
        except TurnRecalledError:
            raise HTTPException(status_code=403, detail={"code": "turn_recalled"})
        except TurnAlreadySupersededError:
            raise HTTPException(status_code=409, detail={"code": "turn_superseded"})
        except TurnInProgressError:
            raise HTTPException(status_code=409, detail={"code": "turn_in_progress"})
        except Exception as exc:
            raise HTTPException(
                status_code=500,
                detail={"code": "turn_failed", "message": str(exc) or "regenerate failed"},
            ) from exc
        return result.to_dict()

    @router.post("/sessions/{session_id}/messages/{turn_id}/recall")
    def recall_message(session_id: str, turn_id: str, request: Request):
        from niannian.tenant.context import resolve_request_context

        chat_cfg = nn._config.get("chat") or {}
        recall_window = float(chat_cfg.get("recall_window_seconds", 120))
        ctx = resolve_request_context(request, nn._config, home=nn._home)
        try:
            recalled, ingested_atom_id = pipeline.sessions.recall_turn(
                session_id,
                turn_id,
                by_user_id=ctx.user_id,
                recall_window_seconds=recall_window,
            )
        except SessionNotFoundError:
            raise HTTPException(status_code=404, detail={"code": "session_not_found"})
        except TurnNotFoundError:
            raise HTTPException(status_code=404, detail={"code": "turn_not_found"})
        except TurnNotRecallableError:
            raise HTTPException(status_code=400, detail={"code": "turn_not_recallable"})
        except RecallWindowExpiredError:
            raise HTTPException(status_code=400, detail={"code": "recall_window_expired"})
        except TurnInProgressError:
            raise HTTPException(status_code=409, detail={"code": "turn_in_progress"})

        payload: dict[str, Any] = {"recalled": recalled}
        if ingested_atom_id:
            payload["ingested_atom_id"] = ingested_atom_id
        return payload

    @router.post("/sessions/{session_id}/turns/{assistant_turn_id}/remember")
    def remember_turn(session_id: str, assistant_turn_id: str):
        chat_cfg = nn._config.get("chat") or {}
        recall_window = float(chat_cfg.get("recall_window_seconds", 120))
        try:
            result = pipeline.remember_turn(
                session_id,
                assistant_turn_id,
                recall_window_seconds=recall_window,
            )
        except SessionNotFoundError:
            raise HTTPException(status_code=404, detail={"code": "session_not_found"})
        except TurnNotFoundError:
            raise HTTPException(status_code=404, detail={"code": "turn_not_found"})
        except PairNotFoundError:
            raise HTTPException(status_code=404, detail={"code": "pair_not_found"})
        except TurnRecalledError:
            raise HTTPException(status_code=403, detail={"code": "turn_recalled"})
        except RecallWindowActiveError:
            raise HTTPException(status_code=403, detail={"code": "recall_window_active"})
        except AlreadyRememberedError:
            raise HTTPException(status_code=409, detail={"code": "already_remembered"})
        except NotRememberableError:
            raise HTTPException(status_code=400, detail={"code": "not_rememberable"})
        except TurnInProgressError:
            raise HTTPException(status_code=409, detail={"code": "turn_in_progress"})
        return result

    @router.post("/sessions/{session_id}/messages/{turn_id}/enrichments/{key}/import-vault")
    def import_enrichment_vault(session_id: str, turn_id: str, key: str):
        try:
            return pipeline.import_enrichment_vault(
                session_id,
                turn_id,
                key,
                importer=vault_importer,
            )
        except SessionNotFoundError:
            raise HTTPException(status_code=404, detail={"code": "session_not_found"})
        except TurnNotFoundError:
            raise HTTPException(status_code=404, detail={"code": "turn_not_found"})
        except EnrichmentNotFoundError:
            raise HTTPException(status_code=404, detail={"code": "enrichment_not_found"})
        except EnrichmentNotActionableError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "enrichment_not_actionable", "reason": exc.reason},
            )
        except EnrichmentAlreadyPersistedError as exc:
            raise HTTPException(
                status_code=409,
                detail={
                    "code": "enrichment_already_persisted",
                    "vault_id": exc.vault_id,
                    "atom_id": exc.atom_id,
                },
            )
        except EnrichmentKindNotSupportedError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "enrichment_kind_not_supported", "kind": exc.kind, "action": exc.action},
            )
        except TurnRecalledError:
            raise HTTPException(status_code=403, detail={"code": "turn_recalled"})
        except DuplicateImportError as exc:
            raise HTTPException(
                status_code=409,
                detail={
                    "code": "duplicate_import",
                    "vault_id": exc.vault_id,
                    "source_url": exc.source_url,
                },
            )
        except TurnInProgressError:
            raise HTTPException(status_code=409, detail={"code": "turn_in_progress"})

    @router.post("/sessions/{session_id}/messages/{turn_id}/enrichments/{key}/learn")
    def learn_enrichment(session_id: str, turn_id: str, key: str):
        try:
            web_svc.patch_settings(enabled=True)
            return pipeline.learn_enrichment(
                session_id,
                turn_id,
                key,
                web_svc=web_svc,
                run_batch_fn=lambda topic_id=None, query=None, manual=True: _run_web_batch(
                    topic_id=topic_id,
                    query=query,
                    manual=manual,
                ),
            )
        except SessionNotFoundError:
            raise HTTPException(status_code=404, detail={"code": "session_not_found"})
        except TurnNotFoundError:
            raise HTTPException(status_code=404, detail={"code": "turn_not_found"})
        except EnrichmentNotFoundError:
            raise HTTPException(status_code=404, detail={"code": "enrichment_not_found"})
        except EnrichmentNotActionableError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "enrichment_not_actionable", "reason": exc.reason},
            )
        except EnrichmentAlreadyPersistedError as exc:
            raise HTTPException(
                status_code=409,
                detail={"code": "enrichment_already_persisted", "vault_id": exc.vault_id},
            )
        except EnrichmentKindNotSupportedError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "enrichment_kind_not_supported", "kind": exc.kind, "action": exc.action},
            )
        except TurnRecalledError:
            raise HTTPException(status_code=403, detail={"code": "turn_recalled"})
        except TurnInProgressError:
            raise HTTPException(status_code=409, detail={"code": "turn_in_progress"})

    @router.post("/sessions/{session_id}/messages/{turn_id}/enrichments/{key}/remember")
    def remember_enrichment(session_id: str, turn_id: str, key: str):
        try:
            return pipeline.remember_enrichment(session_id, turn_id, key)
        except SessionNotFoundError:
            raise HTTPException(status_code=404, detail={"code": "session_not_found"})
        except TurnNotFoundError:
            raise HTTPException(status_code=404, detail={"code": "turn_not_found"})
        except EnrichmentNotFoundError:
            raise HTTPException(status_code=404, detail={"code": "enrichment_not_found"})
        except EnrichmentNotActionableError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "enrichment_not_actionable", "reason": exc.reason},
            )
        except EnrichmentAlreadyPersistedError as exc:
            raise HTTPException(
                status_code=409,
                detail={"code": "enrichment_already_persisted", "atom_id": exc.atom_id},
            )
        except EnrichmentKindNotSupportedError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "enrichment_kind_not_supported", "kind": exc.kind, "action": exc.action},
            )
        except TurnRecalledError:
            raise HTTPException(status_code=403, detail={"code": "turn_recalled"})
        except NotRememberableError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "not_rememberable", "reason": exc.reason},
            )
        except TurnInProgressError:
            raise HTTPException(status_code=409, detail={"code": "turn_in_progress"})

    @router.post("/sessions/{session_id}/consolidate")
    def consolidate_session(session_id: str, request: Request):
        from niannian.tenant.context import resolve_request_context

        try:
            session = pipeline.sessions.require(session_id)
        except SessionNotFoundError:
            raise HTTPException(status_code=404, detail={"code": "session_not_found"})
        ctx = resolve_request_context(request, nn._config, home=nn._home)
        user_id = str(session.meta.get("user_id") or ctx.user_id or "pinn")
        try:
            result = pipeline.consolidate_session(session_id, user_id=user_id)
        except ConsolidationNotBatchModeError:
            raise HTTPException(status_code=403, detail={"code": "consolidation_not_batch_mode"})
        except SessionNotIdleError:
            raise HTTPException(status_code=409, detail={"code": "session_not_idle"})
        except TurnInProgressError:
            raise HTTPException(status_code=409, detail={"code": "turn_in_progress"})
        return result

    @router.get("/proactive-profile")
    def get_proactive_profile(request: Request, session_id: str):
        from niannian.tenant.context import resolve_request_context

        try:
            session = pipeline.sessions.require(session_id)
        except SessionNotFoundError:
            raise HTTPException(status_code=404, detail={"code": "session_not_found"})
        ctx = resolve_request_context(request, nn._config, home=nn._home)
        user_id = str(session.meta.get("user_id") or ctx.user_id or "pinn")
        return proactive.get_profile_bundle(session, user_id=user_id)

    @router.delete("/proactive-profile")
    def reset_proactive_profile(request: Request, session_id: str):
        from niannian.tenant.context import resolve_request_context

        try:
            session = pipeline.sessions.require(session_id)
        except SessionNotFoundError:
            raise HTTPException(status_code=404, detail={"code": "session_not_found"})
        ctx = resolve_request_context(request, nn._config, home=nn._home)
        user_id = str(session.meta.get("user_id") or ctx.user_id or "pinn")
        proactive.reset_profile(session, user_id=user_id)
        return {"ok": True}

    @router.post("/sessions/{session_id}/proactive")
    def proactive_message(session_id: str, payload: ProactiveBody, request: Request):
        from niannian.tenant.context import resolve_request_context

        try:
            session = pipeline.sessions.require(session_id)
        except SessionNotFoundError:
            raise HTTPException(status_code=404, detail={"code": "session_not_found"})
        ctx = resolve_request_context(request, nn._config, home=nn._home)
        user_id = str(session.meta.get("user_id") or ctx.user_id or "pinn")
        body = ProactiveRequest(
            scenario=payload.scenario,
            local_hour=payload.local_hour,
            client_delay_ms=payload.client_delay_ms,
            stream=payload.stream,
        )
        try:
            prep = proactive.prepare(session_id, user_id=user_id, body=body)
        except ProactiveForbidden as exc:
            raise HTTPException(status_code=403, detail={"code": exc.code}) from exc
        except ProactiveRateLimited as exc:
            raise HTTPException(status_code=429, detail={"code": exc.code}) from exc
        except TurnInProgressError:
            raise HTTPException(status_code=409, detail={"code": "turn_in_progress"})

        if prep.hold_back:
            return Response(status_code=204)

        if payload.stream:
            return StreamingResponse(
                _sse_proactive_generator(proactive, session_id, user_id, body),
                media_type="text/event-stream",
                headers={
                    "Cache-Control": "no-cache",
                    "Connection": "keep-alive",
                    "X-Accel-Buffering": "no",
                },
            )

        events = list(proactive.run_stream(session_id, user_id=user_id, body=body))
        done = next((d for ev, d in events if ev == "done"), {})
        return done

    @router.post("/sessions/{session_id}/turns/{turn_id}/proactive-feedback")
    def proactive_feedback(
        session_id: str,
        turn_id: str,
        payload: ProactiveFeedbackBody,
        request: Request,
    ):
        from niannian.chat.session import TurnNotFoundError
        from niannian.tenant.context import resolve_request_context

        try:
            session = pipeline.sessions.require(session_id)
        except SessionNotFoundError:
            raise HTTPException(status_code=404, detail={"code": "session_not_found"})
        ctx = resolve_request_context(request, nn._config, home=nn._home)
        user_id = str(session.meta.get("user_id") or ctx.user_id or "pinn")
        try:
            return proactive.apply_feedback(
                session_id,
                turn_id,
                user_id=user_id,
                rating=payload.rating,
                reason=payload.reason,
            )
        except TurnNotFoundError:
            raise HTTPException(status_code=404, detail={"code": "turn_not_found"})
        except ValueError as exc:
            if str(exc) == "not_proactive_turn":
                raise HTTPException(status_code=400, detail={"code": "not_proactive_turn"})
            raise

    return router


def _sse_proactive_generator(proactive, session_id: str, user_id: str, body):
    from niannian.chat.proactive_pipeline import ProactiveForbidden, ProactiveRateLimited
    from niannian.chat.session import SessionNotFoundError, TurnInProgressError

    try:
        for event, data in proactive.run_stream(session_id, user_id=user_id, body=body):
            yield format_sse_event(event, data)
    except ProactiveForbidden as exc:
        yield format_sse_event(
            "error",
            {"code": exc.code, "message": exc.code, "recoverable": False},
        )
        yield format_sse_event("done", {"turn_id": "", "proactive": False})
    except ProactiveRateLimited as exc:
        yield format_sse_event(
            "error",
            {"code": exc.code, "message": exc.code, "recoverable": True},
        )
        yield format_sse_event("done", {"turn_id": "", "proactive": False})
    except TurnInProgressError:
        yield format_sse_event(
            "error",
            {"code": "turn_in_progress", "message": "turn in progress", "recoverable": True},
        )
        yield format_sse_event("done", {"turn_id": "", "proactive": False})
    except SessionNotFoundError:
        yield format_sse_event(
            "error",
            {"code": "session_not_found", "message": "session not found", "recoverable": False},
        )
        yield format_sse_event("done", {"turn_id": "", "proactive": False})


def _sse_generator(pipeline, session_id: str, payload: SendMessageBody):
    try:
        events = pipeline.run_turn(
            session_id,
            payload.text,
            parts=payload.parts,
            stream=True,
            show_trace=payload.show_trace,
            project=payload.project,
            plan_mode=payload.plan_mode,
        )
        for event, data in events:
            if event == "_return":
                continue
            yield format_sse_event(event, data)
    except TurnInProgressError:
        yield format_sse_event(
            "error",
            {"code": "turn_in_progress", "message": "turn in progress", "recoverable": True},
        )
        yield format_sse_event("done", {"turn_id": "", "ingested": False, "trace": []})
    except SessionNotFoundError:
        yield format_sse_event(
            "error",
            {"code": "session_not_found", "message": "session not found", "recoverable": False},
        )
        yield format_sse_event("done", {"turn_id": "", "ingested": False, "trace": []})
    except Exception as exc:
        yield format_sse_event(
            "error",
            {"code": "turn_failed", "message": str(exc) or "chat turn failed", "recoverable": True},
        )
        yield format_sse_event("done", {"turn_id": "", "ingested": False, "trace": []})


def _sse_regenerate_generator(pipeline, session_id: str, assistant_turn_id: str, payload: RegenerateBody):
    try:
        events = pipeline.regenerate_turn(
            session_id,
            assistant_turn_id,
            stream=True,
            show_trace=payload.show_trace,
            plan_mode=payload.plan_mode,
        )
        for event, data in events:
            if event == "_return":
                continue
            yield format_sse_event(event, data)
    except TurnInProgressError:
        yield format_sse_event(
            "error",
            {"code": "turn_in_progress", "message": "turn in progress", "recoverable": True},
        )
        yield format_sse_event("done", {"turn_id": "", "ingested": False, "trace": []})
    except TurnNotFoundError:
        yield format_sse_event(
            "error",
            {"code": "turn_not_found", "message": "turn not found", "recoverable": False},
        )
        yield format_sse_event("done", {"turn_id": "", "ingested": False, "trace": []})
    except PairNotFoundError:
        yield format_sse_event(
            "error",
            {"code": "pair_not_found", "message": "pair not found", "recoverable": False},
        )
        yield format_sse_event("done", {"turn_id": "", "ingested": False, "trace": []})
    except TurnRecalledError:
        yield format_sse_event(
            "error",
            {"code": "turn_recalled", "message": "turn recalled", "recoverable": False},
        )
        yield format_sse_event("done", {"turn_id": "", "ingested": False, "trace": []})
    except TurnAlreadySupersededError:
        yield format_sse_event(
            "error",
            {"code": "turn_superseded", "message": "turn superseded", "recoverable": False},
        )
        yield format_sse_event("done", {"turn_id": "", "ingested": False, "trace": []})
    except NotAssistantTurnError:
        yield format_sse_event(
            "error",
            {"code": "not_assistant", "message": "not assistant turn", "recoverable": False},
        )
        yield format_sse_event("done", {"turn_id": "", "ingested": False, "trace": []})
    except SessionNotFoundError:
        yield format_sse_event(
            "error",
            {"code": "session_not_found", "message": "session not found", "recoverable": False},
        )
        yield format_sse_event("done", {"turn_id": "", "ingested": False, "trace": []})
    except Exception as exc:
        yield format_sse_event(
            "error",
            {"code": "turn_failed", "message": str(exc) or "regenerate failed", "recoverable": True},
        )
        yield format_sse_event("done", {"turn_id": "", "ingested": False, "trace": []})
