"""Project ownership, knowledge validation and server-controlled run policy."""

from __future__ import annotations

import asyncio
import json
import re
from copy import deepcopy

from fastapi import HTTPException, Request

from app.gateway.auth.disabled_mode import is_auth_disabled
from app.gateway.deps import get_agent_store, get_checkpointer, get_optional_user_from_request
from deerflow.agents.analysis_project import (
    SOLUTION_INTENT_BLOCK_LABEL,
    SOLUTION_REQUIREMENTS_CONFIRMATION_TOKEN,
    SOLUTION_REQUIREMENTS_FORCE_TOKEN,
    SOLUTION_REQUIREMENTS_READY_MARKER,
    SOLUTION_TEMPLATE_ID,
)
from deerflow.config.agents_config import load_agent_config, validate_agent_id
from deerflow.runtime import serialize_channel_values
from deerflow.runtime.user_context import get_effective_user_id

_SOLUTION_INTENT_BLOCK_PATTERN = re.compile(
    rf"<!--\s*{re.escape(SOLUTION_INTENT_BLOCK_LABEL)}\s*(?P<payload>[\s\S]*?)-->",
)
_SOLUTION_INTENT_BLOCK_OPEN_PATTERN = re.compile(rf"<!--\s*{re.escape(SOLUTION_INTENT_BLOCK_LABEL)}")
_SOLUTION_INTENT_SCALAR_FIELDS = (
    "goal",
    "audience",
    "success_outcome",
    "scope",
    "constraints",
    "deliverable",
    "acceptance_criteria",
)
_SOLUTION_INTENT_MAX_PAYLOAD_CHARS = 20_000
_SOLUTION_INTENT_MAX_STRING_CHARS = 2_000
_SOLUTION_INTENT_MAX_LIST_ITEMS = 20
_SOLUTION_INTENT_MAX_LIST_ITEM_CHARS = 1_000


async def validate_project_agent(request: Request, agent_id: str | None):
    if not agent_id:
        return None
    try:
        validate_agent_id(agent_id)
    except ValueError as exc:
        raise HTTPException(422, "智能体标识无效") from exc
    user_id, is_admin = await actor(request)
    store = get_agent_store(request)
    row = await store.get_visible(agent_id, user_id)
    if row is None and is_admin:
        row = await store.get_any(agent_id)
    if row is None:
        raise HTTPException(422, "所选智能体不存在或无访问权限，请重新选择")
    return row


def project_store(request: Request):
    store = getattr(request.app.state, "analysis_project_store", None)
    if store is None:
        raise HTTPException(503, "项目存储尚未就绪")
    return store


async def actor(request: Request) -> tuple[str, bool]:
    user = await get_optional_user_from_request(request)
    if user:
        return str(user.id), getattr(user, "system_role", None) == "admin"
    if is_auth_disabled():
        return get_effective_user_id(), True
    raise HTTPException(401, "请先登录")


async def owned_project(request: Request, project_id: str) -> dict:
    user_id, _ = await actor(request)
    project = await project_store(request).get(project_id, user_id)
    if project is None:
        raise HTTPException(404, "项目不存在或无权访问")
    return project


async def validate_knowledge(request: Request, project: dict) -> None:
    user_id, is_admin = await actor(request)
    for store_name, field in (("llmwiki_store", "knowledge_base_ids"), ("assistant_knowledge_store", "assistant_knowledge_base_ids")):
        ids = project.get(field) or []
        if not ids:
            continue
        store = getattr(request.app.state, store_name, None)
        if store is None:
            raise HTTPException(503, "知识库服务暂不可用")
        for base_id in ids:
            row = await store.get_authorized(base_id, user_id, write=False, is_admin=is_admin) if field == "knowledge_base_ids" else await store.get_base(base_id)
            if row is None:
                raise HTTPException(422, "所选知识库不存在或已无访问权限，请重新配置项目背景")


def _message_content_text(value) -> str:
    """Flatten LangGraph's string or content-block message payloads."""
    if isinstance(value, str):
        return value
    if isinstance(value, list):
        return "\n".join(_message_content_text(item) for item in value)
    if isinstance(value, dict):
        block_type = value.get("type")
        if block_type in {None, "text", "input_text", "output_text"} and isinstance(value.get("text"), str):
            return value["text"]
        if block_type in {None, "text", "input_text", "output_text"} and "content" in value:
            return _message_content_text(value["content"])
    return ""


def solution_requirements_confirmed_by_input(payload: dict | None) -> bool:
    """Accept confirmation only from a user message, never caller context."""
    messages = (payload or {}).get("messages")
    if not isinstance(messages, list) or not messages:
        return False
    message = messages[-1]
    if not isinstance(message, dict):
        return False
    role = message.get("role") or message.get("type")
    return role in {"human", "user"} and SOLUTION_REQUIREMENTS_CONFIRMATION_TOKEN in _message_content_text(message.get("content"))


def solution_force_generate_by_input(payload: dict | None) -> bool:
    """Accept force-generation only from the current final user message."""
    messages = (payload or {}).get("messages")
    if not isinstance(messages, list) or not messages:
        return False
    message = messages[-1]
    if not isinstance(message, dict):
        return False
    role = message.get("role") or message.get("type")
    return role in {"human", "user"} and SOLUTION_REQUIREMENTS_FORCE_TOKEN in _message_content_text(message.get("content"))


def _json_object_without_duplicate_keys(pairs: list[tuple[str, object]]) -> dict[str, object]:
    value: dict[str, object] = {}
    for key, item in pairs:
        if key in value:
            raise ValueError(f"duplicate JSON key: {key}")
        value[key] = item
    return value


def _normalized_intent_string(value: object, *, max_chars: int) -> str | None:
    if not isinstance(value, str):
        return None
    normalized = value.strip()
    if not normalized or len(normalized) > max_chars:
        return None
    return normalized


def _normalized_intent_string_list(value: object, *, allow_empty: bool) -> list[str] | None:
    if not isinstance(value, list) or len(value) > _SOLUTION_INTENT_MAX_LIST_ITEMS:
        return None
    normalized: list[str] = []
    for item in value:
        text = _normalized_intent_string(item, max_chars=_SOLUTION_INTENT_MAX_LIST_ITEM_CHARS)
        if text is None:
            return None
        normalized.append(text)
    if not allow_empty and not normalized:
        return None
    return normalized


def _validate_solution_intent_snapshot(value: object) -> dict[str, object] | None:
    if not isinstance(value, dict):
        return None
    snapshot: dict[str, object] = {}
    for field in _SOLUTION_INTENT_SCALAR_FIELDS:
        normalized = _normalized_intent_string(value.get(field), max_chars=_SOLUTION_INTENT_MAX_STRING_CHARS)
        if normalized is None:
            return None
        snapshot[field] = normalized
    open_questions = _normalized_intent_string_list(value.get("open_questions"), allow_empty=True)
    if open_questions is None:
        return None
    snapshot["open_questions"] = open_questions
    return snapshot


def _solution_intent_block(text: str) -> tuple[dict[str, object], int, int] | None:
    if len(_SOLUTION_INTENT_BLOCK_OPEN_PATTERN.findall(text)) != 1:
        return None
    matches = list(_SOLUTION_INTENT_BLOCK_PATTERN.finditer(text))
    if len(matches) != 1:
        return None
    match = matches[0]
    payload = match.group("payload").strip()
    if not payload or len(payload) > _SOLUTION_INTENT_MAX_PAYLOAD_CHARS:
        return None
    try:
        value = json.loads(payload, object_pairs_hook=_json_object_without_duplicate_keys)
    except (TypeError, ValueError):
        return None
    snapshot = _validate_solution_intent_snapshot(value)
    if snapshot is None:
        return None
    return snapshot, match.start(), match.end()


def parse_solution_intent_snapshot(content: object) -> dict[str, object] | None:
    """Parse one bounded, allowlisted intent snapshot from assistant content."""
    parsed = _solution_intent_block(_message_content_text(content))
    return parsed[0] if parsed else None


def latest_solution_intent_snapshot_in_messages(messages: object) -> dict[str, object] | None:
    """Return the newest valid assistant snapshot, including incomplete ones.

    Force-generation may reuse an earlier best-effort snapshot whose
    ``open_questions`` are still non-empty.  It remains only a hint: the force
    prompt requires the model to reconcile it with the complete conversation.
    At least one preceding user turn is required so an orphan assistant message
    cannot manufacture project intent.
    """
    if not isinstance(messages, list):
        return None
    saw_user = False
    latest: dict[str, object] | None = None
    for message in messages:
        if not isinstance(message, dict):
            continue
        role = message.get("role") or message.get("type")
        if role in {"human", "user"}:
            saw_user = True
            continue
        if role not in {"ai", "assistant"} or not saw_user:
            continue
        parsed = parse_solution_intent_snapshot(message.get("content"))
        if parsed is not None:
            latest = parsed
    return latest


def solution_intent_state_in_messages(messages: object) -> tuple[dict[str, object] | None, bool]:
    """Return the latest assistant snapshot and whether it is safe to generate."""
    if not isinstance(messages, list):
        return None, False
    latest_user_index = -1
    latest_assistant_index = -1
    latest_assistant_text = ""
    for index, message in enumerate(messages):
        if not isinstance(message, dict):
            continue
        role = message.get("role") or message.get("type")
        if role in {"human", "user"}:
            latest_user_index = index
        elif role in {"ai", "assistant"}:
            latest_assistant_index = index
            latest_assistant_text = _message_content_text(message.get("content"))
    if latest_user_index < 0 or latest_assistant_index <= latest_user_index:
        return None, False
    parsed = _solution_intent_block(latest_assistant_text)
    if parsed is None:
        return None, False
    snapshot, block_start, block_end = parsed
    heading = "## 意图理解"
    heading_index = latest_assistant_text.rfind(heading, 0, block_start)
    ready_index = latest_assistant_text.find(SOLUTION_REQUIREMENTS_READY_MARKER, block_end)
    visible_summary = latest_assistant_text[heading_index + len(heading) : block_start].strip() if heading_index >= 0 else ""
    ready_tail = latest_assistant_text[ready_index + len(SOLUTION_REQUIREMENTS_READY_MARKER) :] if ready_index >= 0 else ""
    ready = bool(heading_index >= 0 and visible_summary and ready_index >= 0 and latest_assistant_text.count(SOLUTION_REQUIREMENTS_READY_MARKER) == 1 and not ready_tail.strip() and not snapshot["open_questions"])
    return snapshot, ready


def solution_requirements_ready_in_messages(messages: object) -> bool:
    """Require a fresh validated intent snapshot before accepting consent."""
    return solution_intent_state_in_messages(messages)[1]


async def prepare_project_run(body, thread_id: str, request: Request):
    """Resolve policy from the immutable binding, never from caller metadata.

    Applies to all Gateway run endpoints, including direct SDK calls. Strip
    forged policy even on ordinary threads; all other chat behavior is preserved.
    """
    body = body.model_copy(deep=True) if hasattr(body, "model_copy") else deepcopy(body)
    body.context = dict(body.context or {})
    body.context.pop("analysis_project", None)
    body.context.pop("project_memory_scope", None)
    body.context.pop("project_memory", None)
    body.context.pop("solution_requirements_confirmed", None)
    body.context.pop("solution_force_generate", None)
    body.context.pop("solution_intent", None)
    body.config = dict(body.config or {})
    body.config.pop("solution_requirements_confirmed", None)
    body.config.pop("solution_force_generate", None)
    body.config.pop("solution_intent", None)
    for key in ("context", "configurable"):
        if isinstance(body.config.get(key), dict):
            body.config[key].pop("analysis_project", None)
            body.config[key].pop("project_memory_scope", None)
            body.config[key].pop("project_memory", None)
            body.config[key].pop("solution_requirements_confirmed", None)
            body.config[key].pop("solution_force_generate", None)
            body.config[key].pop("solution_intent", None)
    body.metadata = dict(body.metadata or {})
    body.metadata.pop("project_memory_scope", None)
    body.metadata.pop("project_memory", None)
    body.metadata.pop("solution_requirements_confirmed", None)
    body.metadata.pop("solution_force_generate", None)
    body.metadata.pop("solution_intent", None)
    store = getattr(request.app.state, "analysis_project_store", None)
    if store is None:
        return body, None
    project = await store.project_for_thread(thread_id)
    if project is None:
        return body, None
    user_id, _ = await actor(request)
    if project["user_id"] != user_id:
        raise HTTPException(404, "项目不存在或无权访问")
    thread_config = await store.config_for(f"thread:{thread_id}")
    agent_id = thread_config.get("agent_id")
    await validate_project_agent(request, agent_id)
    agent_config = await asyncio.to_thread(load_agent_config, agent_id) if agent_id else None
    knowledge_ids = list(
        dict.fromkeys(
            [
                *(project.get("knowledge_base_ids") or []),
                *(getattr(agent_config, "llmwiki_knowledge_base_ids", None) or []),
            ]
        )
    )
    assistant_knowledge_ids = list(
        dict.fromkeys(project.get("assistant_knowledge_base_ids") or [])
    )
    await validate_knowledge(
        request,
        {
            "knowledge_base_ids": knowledge_ids,
            "assistant_knowledge_base_ids": assistant_knowledge_ids,
        },
    )
    for message in (body.input or {}).get("messages", []):
        if isinstance(message, dict) and isinstance(message.get("additional_kwargs"), dict):
            # Source snapshots must come from this run's server retrieval.
            for key in (
                "llmwiki_reference_sources",
                "llmwiki_reference_query",
                "prompt_prefix",
                "solution_intent",
                "solution_requirements_confirmed",
                "solution_force_generate",
            ):
                message["additional_kwargs"].pop(key, None)
    artifact_path = None
    solution_intent = None
    solution_requirements_ready = False
    solution_force_generate = False
    if project["mode"] == "solution":
        checkpoint = await get_checkpointer(request).aget_tuple({"configurable": {"thread_id": thread_id}})
        values = checkpoint.checkpoint.get("channel_values", {}) if checkpoint else {}
        if "/mnt/user-data/outputs/report.md" in (values.get("artifacts") or []):
            artifact_path = "/mnt/user-data/outputs/report.md"
        serialized_values = serialize_channel_values(values)
        checkpoint_messages = serialized_values.get("messages")
        solution_intent, solution_requirements_ready = solution_intent_state_in_messages(checkpoint_messages)
        solution_force_generate = solution_force_generate_by_input(body.input)
        if solution_force_generate:
            solution_intent = latest_solution_intent_snapshot_in_messages(checkpoint_messages)
    solution_requirements_confirmed = bool(artifact_path) or (
        project["mode"] == "solution"
        and (
            solution_force_generate
            or (solution_requirements_ready and solution_requirements_confirmed_by_input(body.input))
        )
    )
    project_memory_scope = None
    project_memory = None
    if project["mode"] == "qa":
        from app.gateway.analysis_project_memory import (
            latest_user_query,
            prepare_project_memory,
        )

        project_memory_scope, project_memory = await prepare_project_memory(
            request=request,
            project=project,
            thread_id=thread_id,
            query=latest_user_query(body),
        )
    policy = {
        "analysis_project": {
            "id": project["id"],
            "name": project["name"],
            "mode": project["mode"],
            "background": project["background"],
            "template_id": SOLUTION_TEMPLATE_ID,
            "report_config": project.get("report_config") or {},
            "strategy_config": project.get("strategy_config") or {},
        },
        "llmwiki_knowledge_base_ids": knowledge_ids,
        "assistant_knowledge_base_ids": assistant_knowledge_ids,
        "writing_mode": project["mode"] == "solution",
        # The artifact path is resolved by the existing agent/middleware, not a
        # client-supplied path that could skip the first-report pipeline.
        "writing_artifact_path": artifact_path,
        "solution_requirements_confirmed": solution_requirements_confirmed,
        "solution_force_generate": solution_force_generate,
        "solution_intent": solution_intent,
        "agent_id": agent_id,
        "agent_name": None,
        "is_plan_mode": False,
        "subagent_enabled": False,
        "is_bootstrap": False,
        "is_writing_setup": False,
        "is_report_structure_setup": False,
        "knowledge_rag_enabled": False,
    }
    if project_memory_scope is not None:
        policy.update(
            {
                "project_memory_scope": project_memory_scope,
                "project_memory": project_memory,
                "memory_recall_disabled": True,
                "memory_injection_enabled": False,
            }
        )
    body.assistant_id = agent_id or "lead_agent"
    body.context.update(policy)
    body.metadata = {
        **(body.metadata or {}),
        "agent_id": agent_id,
        "agent_name": None,
        "analysis_project_id": project["id"],
        "analysis_project_mode": project["mode"],
        "analysis_project_name": project["name"],
    }
    for key in ("context", "configurable"):
        if isinstance(body.config.get(key), dict):
            body.config[key].update(policy)
    return body, project