"""Service layer for the MAF roundtable experiment.

Enforces the cross-cutting policies the individual repositories don't:

- **User isolation (MAF-01-07)**: every read / write is scoped to the caller's
  ``user_id``. Read and write scopes are deliberately **separate**:
  ``admin_diagnostic_access`` only widens the *read* scope (cross-user
  diagnostics) — it never grants cross-user *writes*. Every mutating operation
  (update / delete session, start / cancel run, append event, decide approval)
  is owner-checked via ``_write_scope_user`` regardless of the diagnostic flag.
- **Run gating + atomic launch**: a session may have at most one active run
  (``starting`` / ``running`` / ``waiting_*``). The launch is a **single
  database transaction** (``store.claim_and_create_pending_run``) that
  conditionally updates the session to ``running`` and inserts a ``pending``
  run together — not a multi-step claim/create/set_status dance, and not a
  Python lock. Concurrent callers cannot both win; a mid-transaction failure
  rolls everything back so the session never sticks in a half-claimed state.
- **Feature flag**: when ``maf_roundtable.enabled`` is off, write paths raise
  ``MafDisabledError`` (the router maps it to 503); reads stay open so admins
  can review history.
- **Config-driven defaults**: ``mode`` / ``write_policy`` / ``max_rounds``
  default to the deployed ``MafRoundtableConfig`` values when the create
  request omits them; an explicit request value always wins.

The service does **not** load the MAF SDK — that is Phase 2+ (the
``DeerFlowAgentAdapter``). Phase 1 ships the persistence + API skeleton; until
a real executor exists, a launched run is parked in ``pending`` (clearly a
reserved / not-yet-executing state) rather than pretending to be ``running``.
"""

from __future__ import annotations

import logging
from typing import Any

from app.maf_roundtable.sanitize import sanitize, sanitize_payload
from deerflow.config.maf_roundtable_config import MafRoundtableConfig
from deerflow.persistence.maf_roundtable.store import MafRoundtableStore

logger = logging.getLogger(__name__)


class MafDisabledError(RuntimeError):
    """Raised when a write is attempted while the feature flag is off."""


class MafNotFoundError(KeyError):
    """Raised when a session / run / action / approval is missing or out of scope."""

    def __init__(self, what: str, ident: str) -> None:
        super().__init__(f"{what} not found or out of scope: {ident}")
        self.what = what
        self.ident = ident


class MafConflictError(RuntimeError):
    """Raised when a session is in a state that rejects the requested edit."""


class MafForbiddenError(PermissionError):
    """Raised when a caller is authenticated but not the owner of the target
    resource (e.g. an admin trying to *write* another user's session while
    only holding cross-user *read* via the diagnostic flag)."""

    def __init__(self, what: str, ident: str) -> None:
        super().__init__(f"not the owner of {what}: {ident}")
        self.what = what
        self.ident = ident


class MafConfigurationError(RuntimeError):
    """MAF is enabled but a required safety prerequisite is missing.

    Raised by :func:`validate_maf_prerequisites` on the startup path so the
    feature can never come up in a half-safe state ("MAF on, approval guard not
    mounted"). The gateway lifespan catches this and forces the feature OFF
    (clearly logged) rather than aborting the whole (experimental, opt-in)
    service.
    """


def get_maf_write_guard_readiness(*, app_config: Any | None = None) -> dict[str, Any]:
    """Return non-sensitive WriteGuard readiness information for MAF health.

    A missing secret intentionally does not disable read-only MAF work:
    WriteGuard fails closed before any external write can execute. ``ready``
    is therefore the signal that the complete approval + encryption path is
    usable, while ``enabled`` describes whether the middleware is mounted.
    """
    import os

    from deerflow.config.write_guard_config import get_write_guard_config
    from deerflow.reflection import resolve_variable

    wg = getattr(app_config, "write_guard", None) if app_config is not None else None
    if wg is None:
        wg = get_write_guard_config()

    provider = getattr(wg, "provider", None)
    provider_configured = bool(
        provider
        and getattr(provider, "policy_use", "")
        and getattr(provider, "interceptor_use", "")
    )
    bridge_resolvable = False
    if provider_configured:
        try:
            resolve_variable(provider.policy_use)
            resolve_variable(provider.interceptor_use)
            bridge_resolvable = True
        except Exception:  # noqa: BLE001 - health must remain safe and non-fatal
            bridge_resolvable = False

    secret_configured = bool((os.getenv("MAF_ACTION_SECRET_KEY") or "").strip())
    if not getattr(wg, "enabled", False):
        status = "disabled"
    elif not provider_configured or not bridge_resolvable:
        status = "provider_unavailable"
    elif not secret_configured:
        status = "missing_secret"
    else:
        status = "ready"

    return {
        "enabled": bool(getattr(wg, "enabled", False)),
        "ready": status == "ready",
        "secret_configured": secret_configured,
        "status": status,
    }


def validate_maf_prerequisites() -> None:
    """Fail loudly if MAF is enabled without its full safety stack in place.

    Checks, in order:
    1. ``write_guard.enabled`` is true (the approval guard is mounted);
    2. ``write_guard.provider`` is complete (``policy_use`` + ``interceptor_use``);
    3. the bridge factories actually resolve (the app-layer guard is importable);
    4. secret readiness is reported separately; a missing secret must not
       disable read-only MAF work, while writes fail closed before execution.

    Raises ``MafConfigurationError`` on the first unmet condition. Callers on
    the startup path treat that as "force the feature off"; direct unit-test
    construction of :class:`MafRoundtableService` does NOT call this.
    """
    from deerflow.config.write_guard_config import get_write_guard_config
    from deerflow.reflection import resolve_variable

    wg = get_write_guard_config()
    if not wg.enabled:
        raise MafConfigurationError(
            "maf_roundtable.enabled=true requires write_guard.enabled=true "
            "(the approval guard must be mounted)"
        )
    provider = wg.provider
    if provider is None or not provider.policy_use or not provider.interceptor_use:
        raise MafConfigurationError(
            "maf_roundtable.enabled=true requires a complete write_guard.provider "
            "(policy_use + interceptor_use)"
        )
    try:
        resolve_variable(provider.policy_use)
        resolve_variable(provider.interceptor_use)
    except Exception as exc:  # noqa: BLE001
        raise MafConfigurationError(
            f"write_guard bridge factories are not resolvable: {exc}"
        ) from exc


# Session statuses that may be edited (config changes).
_EDITABLE_STATUSES = {"draft", "completed", "failed", "cancelled"}
# Session statuses that mean a run is in flight (no config edits). ``starting``
# is the transient CAS-claim state between "claimed" and "run row created".
# ``cancel_failed`` blocks edits because the cancel cascade is incomplete and
# the may still have active sub-runs.
_RUNNING_STATUSES = {"starting", "running", "waiting_approval", "waiting_input", "cancelling", "cancel_failed"}


class MafRoundtableService:
    def __init__(
        self,
        store: MafRoundtableStore,
        *,
        enabled: bool,
        admin_diagnostic_access: bool,
        config: MafRoundtableConfig | None = None,
        agent_store: Any | None = None,
        app_config: Any | None = None,
    ) -> None:
        self._store = store
        self._admin_diagnostic_access = admin_diagnostic_access
        self._agent_store = agent_store
        self._app_config = app_config
        # Defaults source for create_session; falls back to the pydantic
        # defaults when no config is injected (keeps unit tests lightweight).
        self._config = config or MafRoundtableConfig()

        # A missing MAF_ACTION_SECRET_KEY does not disable read-only MAF
        # operation. The action coordinator raises before creating or
        # executing an external-write action, so writes fail closed until the
        # deployment injects the secret through the environment.
        self._enabled = enabled

    @property
    def store(self) -> MafRoundtableStore:
        return self._store

    @property
    def enabled(self) -> bool:
        return self._enabled

    @property
    def config(self) -> MafRoundtableConfig:
        return self._config

    def _require_enabled(self) -> None:
        if not self._enabled:
            raise MafDisabledError("MAF roundtable feature is disabled")

    # ── permission scopes ──────────────────────────────────────────────────

    def _read_scope_user(self, user_id: str | None, is_admin: bool) -> str | None:
        """User id to scope **read** queries by.

        - Admin with diagnostic access: ``None`` (cross-user read for diagnostics).
        - Anyone else: their own ``user_id`` (enforced).
        """
        if is_admin and self._admin_diagnostic_access:
            return None
        return user_id

    def _write_scope_user(self, user_id: str | None, is_admin: bool) -> str | None:
        """User id to scope **write** operations by.

        The diagnostic flag is read-only — it never widens the write scope.
        Every mutating operation is owner-checked against the caller's own
        ``user_id`` (admin or not). A future explicit operator/admin force-stop
        would be a separate, audited code path, not a reuse of this flag.
        """
        return user_id

    # ── sessions ──────────────────────────────────────────────────────────

    async def create_session(
        self, *, user_id: str, body: dict[str, Any], is_admin: bool = False
    ) -> dict[str, Any]:
        self._require_enabled()
        cfg = self._config
        from app.maf_roundtable.capability_snapshot import (
            AgentNotAccessibleError,
            InvalidExcludedToolsError,
            build_trusted_capability_snapshot,
            seat_snapshot_for_agent,
        )
        from deerflow.config import get_app_config

        app_config = self._app_config or get_app_config()
        raw_participants = list(body.get("participants") or [])
        # Strip any client-supplied snapshot / tool metadata — never trusted.
        participants: list[dict[str, Any]] = []
        for p in raw_participants:
            participants.append(
                {
                    "agent_id": p.get("agent_id"),
                    "display_name": p.get("display_name", ""),
                    "role": p.get("role", "seat"),
                    "sort_order": p.get("sort_order", 0),
                    "thread_id": p.get("thread_id"),
                    "excluded_tools": p.get("excluded_tools") or [],
                }
            )

        try:
            trusted_snapshot = await build_trusted_capability_snapshot(
                user_id=user_id,
                participants=participants,
                app_config=app_config,
                agent_store=self._agent_store,
                session_model_name=body.get("model_name"),
                session_reasoning_effort=body.get("reasoning_effort"),
                session_excluded_tools=body.get("excluded_tools"),
            )
        except AgentNotAccessibleError as exc:
            raise MafForbiddenError("agent", exc.agent_id) from exc
        except InvalidExcludedToolsError as exc:
            raise MafConflictError(
                f"excluded_tools not allowed for agent {exc.agent_id}: {exc.tools}"
            ) from exc

        for p in participants:
            aid = p.get("agent_id")
            if aid:
                p["capability_snapshot"] = seat_snapshot_for_agent(trusted_snapshot, str(aid))

        session_data: dict[str, Any] = {
            "user_id": user_id,
            "title": body.get("title", ""),
            "task_text": body["task_text"],
            "task_payload": body.get("task_payload"),
            "mode": body.get("mode") or cfg.default_mode,
            "status": "draft",
            "model_name": body.get("model_name"),
            "reasoning_effort": body.get("reasoning_effort"),
            "manager_agent_id": body.get("manager_agent_id"),
            "max_rounds": body.get("max_rounds") if body.get("max_rounds") is not None else cfg.default_max_rounds,
            "write_policy": body.get("write_policy") or cfg.default_write_policy,
            "capability_snapshot": trusted_snapshot,
        }
        session = await self._store.sessions.create_session(session_data)
        if participants:
            await self._store.participants.replace_participants(session["id"], participants)
        return await self.get_session(session_id=session["id"], user_id=user_id, is_admin=is_admin)

    async def get_session(self, *, session_id: str, user_id: str | None, is_admin: bool = False) -> dict[str, Any]:
        scope = self._read_scope_user(user_id, is_admin)
        session = await self._store.sessions.get_session(session_id, user_id=scope)
        if session is None:
            raise MafNotFoundError("session", session_id)
        participants = await self._store.participants.list_participants(session_id)
        session["participants"] = participants
        return session

    async def list_sessions(
        self,
        *,
        user_id: str | None,
        is_admin: bool = False,
        status: str | None = None,
        mode: str | None = None,
        limit: int = 100,
        offset: int = 0,
    ) -> tuple[list[dict[str, Any]], int]:
        scope = self._read_scope_user(user_id, is_admin)
        rows = await self._store.sessions.list_sessions(
            user_id=scope, status=status, mode=mode, limit=limit, offset=offset
        )
        total = await self._store.sessions.count_sessions(user_id=scope, status=status, mode=mode)
        # Attach participants to each row (cheap — Phase 1 lists are bounded).
        for row in rows:
            row["participants"] = await self._store.participants.list_participants(row["id"])
        return rows, total

    async def update_session(
        self, *, session_id: str, user_id: str | None, fields: dict[str, Any], is_admin: bool = False
    ) -> dict[str, Any]:
        self._require_enabled()
        # Writes are owner-scoped even for admins (diagnostic access is read-only).
        scope = self._write_scope_user(user_id, is_admin)
        existing = await self._store.sessions.get_session(session_id, user_id=scope)
        if existing is None:
            # Distinguish "missing" from "owned by someone else" so an admin
            # (who can read cross-user) gets a clear 403 rather than a 404.
            if is_admin and await self._store.sessions.get_session(session_id, user_id=None) is not None:
                raise MafForbiddenError("session", session_id)
            raise MafNotFoundError("session", session_id)
        if existing["status"] in _RUNNING_STATUSES:
            raise MafConflictError(
                f"session {session_id} is running ({existing['status']}); config edits not allowed"
            )
        # capability_snapshot is frozen — never accept it on update.
        fields.pop("capability_snapshot", None)
        updated = await self._store.sessions.update_session(session_id, user_id=scope, fields=fields)
        if updated is None:
            raise MafNotFoundError("session", session_id)
        if "participants" in fields and fields["participants"] is not None:
            await self._store.participants.replace_participants(session_id, fields["participants"])
        return await self.get_session(session_id=session_id, user_id=user_id, is_admin=is_admin)

    async def delete_session(self, *, session_id: str, user_id: str | None, is_admin: bool = False) -> None:
        self._require_enabled()
        scope = self._write_scope_user(user_id, is_admin)
        # Pre-check existence cross-user so an admin gets 403, not 404.
        if await self._store.sessions.get_session(session_id, user_id=scope) is None:
            if is_admin and await self._store.sessions.get_session(session_id, user_id=None) is not None:
                raise MafForbiddenError("session", session_id)
            raise MafNotFoundError("session", session_id)
        ok = await self._store.sessions.delete_session(session_id, user_id=scope)
        if not ok:
            raise MafNotFoundError("session", session_id)

    # ── runs ────────────────────────────────────────────────────────────────

    async def create_run(
        self, *, session_id: str, user_id: str | None, workflow_version: str | None = None, is_admin: bool = False
    ) -> dict[str, Any]:
        self._require_enabled()
        # Owner-only write: an admin cannot start another user's run. The
        # owner_user_id also rides in the atomic SQL (not just this pre-check).
        if not user_id:
            raise MafNotFoundError("session", session_id)

        run = await self._store.claim_and_create_pending_run(
            session_id=session_id,
            owner_user_id=user_id,
            workflow_version=workflow_version,
        )
        if run is not None:
            return run

        # Classify the failed claim: missing / wrong-owner / already active.
        any_session = await self._store.sessions.get_session(session_id, user_id=None)
        if any_session is None:
            raise MafNotFoundError("session", session_id)
        if any_session["user_id"] != user_id:
            if is_admin:
                raise MafForbiddenError("session", session_id)
            raise MafNotFoundError("session", session_id)
        raise MafConflictError(
            f"session {session_id} already has an active run; concurrent launch rejected"
        )

    async def get_run(self, *, run_id: str, user_id: str | None, is_admin: bool = False) -> dict[str, Any]:
        run = await self._store.runs.get_run(run_id)
        if run is None:
            raise MafNotFoundError("run", run_id)
        # Enforce session owner scope (read scope — admins may diagnose).
        scope = self._read_scope_user(user_id, is_admin)
        session = await self._store.sessions.get_session(run["session_id"], user_id=scope)
        if session is None:
            raise MafNotFoundError("run", run_id)
        return run

    async def list_runs(self, *, session_id: str, user_id: str | None, is_admin: bool = False) -> list[dict[str, Any]]:
        scope = self._read_scope_user(user_id, is_admin)
        session = await self._store.sessions.get_session(session_id, user_id=scope)
        if session is None:
            raise MafNotFoundError("session", session_id)
        return await self._store.runs.list_runs(session_id)

    async def cancel_run(self, *, run_id: str, user_id: str | None, is_admin: bool = False) -> dict[str, Any]:
        self._require_enabled()
        if not user_id:
            raise MafNotFoundError("run", run_id)

        result = await self._store.cancel_run_atomically(run_id=run_id, owner_user_id=user_id)
        if result is not None:
            return result

        # Classify the miss: missing vs wrong-owner.
        run = await self._store.runs.get_run(run_id)
        if run is None:
            raise MafNotFoundError("run", run_id)
        if is_admin:
            raise MafForbiddenError("run", run_id)
        raise MafNotFoundError("run", run_id)

    async def start_cancel(self, *, run_id: str, user_id: str | None, is_admin: bool = False) -> dict[str, Any]:
        """Phase 1 of two-phase cancel: atomically set run+session to ``cancelling``.

        The caller (router) then calls ``dispatcher.cancel(run_id)`` to
        cascade to DeerFlow, then calls ``cancel_run`` to finalize to
        ``cancelled``.
        """
        self._require_enabled()
        if not user_id:
            raise MafNotFoundError("run", run_id)

        result = await self._store.start_cancel_atomically(run_id=run_id, owner_user_id=user_id)
        if result is not None:
            return result

        # Classify the miss.
        run = await self._store.runs.get_run(run_id)
        if run is None:
            raise MafNotFoundError("run", run_id)
        if is_admin:
            raise MafForbiddenError("run", run_id)
        raise MafNotFoundError("run", run_id)

    async def finalize_cancel(self, *, run_id: str, user_id: str | None, is_admin: bool = False) -> dict[str, Any]:
        """Phase 2 of two-phase cancel: finalize ``cancelling`` → ``cancelled``."""
        return await self.cancel_run(run_id=run_id, user_id=user_id, is_admin=is_admin)

    async def mark_cancel_failed(
        self,
        *,
        run_id: str,
        user_id: str | None,
        is_admin: bool = False,
        reason: str = "",
    ) -> dict[str, Any]:
        """Mark a cancelling run as ``cancel_failed`` (non-terminal).

        Used when the dispatcher's cancel cascade could not confirm all
        DeerFlow sub-runs were cancelled. The run AND session both enter
        ``cancel_failed`` atomically so their statuses never diverge.
        The user can retry cancel later — ``cancel_run_atomically``
        accepts ``cancel_failed`` as a valid input state.
        A failed cancel must NEVER masquerade as ``cancelled``.
        """
        self._require_enabled()
        if not user_id:
            raise MafNotFoundError("run", run_id)

        scope = self._write_scope_user(user_id, is_admin)
        run = await self._store.runs.get_run(run_id)
        if run is None:
            raise MafNotFoundError("run", run_id)
        # Owner (or admin) check.
        session = await self._store.sessions.get_session(run["session_id"], user_id=scope)
        if session is None:
            if is_admin and await self._store.sessions.get_session(run["session_id"], user_id=None) is not None:
                raise MafForbiddenError("run", run_id)
            raise MafNotFoundError("run", run_id)

        safe_reason = str(sanitize(reason))[:500] if reason else "one or more sub-runs could not be cancelled"
        updated = await self._store.mark_cancel_failed_atomically(
            run_id=run_id,
            owner_user_id=user_id,
            error_code="CancelCascadeIncomplete",
            error_message=safe_reason,
        )
        if updated is None:
            raise MafNotFoundError("run", run_id)
        return updated

    # ── events ─────────────────────────────────────────────────────────────

    async def append_event(
        self,
        *,
        run_id: str,
        user_id: str | None,
        event_type: str,
        agent_id: str | None = None,
        turn_index: int | None = None,
        payload: dict[str, Any] | None = None,
        source_event_id: str | None = None,
        is_admin: bool = False,
    ) -> dict[str, Any]:
        """Append an event to a run's stream (internal / executor only).

        There is **no** public HTTP write endpoint for events — Phase-1 audit
        integrity requires that only the backend Workflow executor (Phase 2+)
        call this method. Frontend联调 mocks belong in tests / scripts, not in
        a production-reachable route.
        """
        self._require_enabled()
        # Owner-only write: an admin cannot forge events into another user's run.
        run = await self._store.runs.get_run(run_id)
        if run is None:
            raise MafNotFoundError("run", run_id)
        scope = self._write_scope_user(user_id, is_admin)
        session = await self._store.sessions.get_session(run["session_id"], user_id=scope)
        if session is None:
            if is_admin and await self._store.sessions.get_session(run["session_id"], user_id=None) is not None:
                raise MafForbiddenError("run", run_id)
            raise MafNotFoundError("run", run_id)
        return await self._store.events.append_event(
            session_id=run["session_id"],
            run_id=run_id,
            event_type=event_type,
            agent_id=agent_id,
            turn_index=turn_index,
            payload=sanitize_payload(payload),
            source_event_id=source_event_id,
        )

    async def list_events(
        self, *, run_id: str, user_id: str | None, after_seq: int = 0, limit: int = 1000, is_admin: bool = False
    ) -> tuple[list[dict[str, Any]], int]:
        # get_run enforces owner read-scope (raises MafNotFoundError if out of scope).
        await self.get_run(run_id=run_id, user_id=user_id, is_admin=is_admin)
        events = await self._store.events.list_events(run_id, after_seq=after_seq, limit=limit)
        latest = await self._store.events.latest_seq(run_id)
        return events, latest

    # ── actions ────────────────────────────────────────────────────────────

    async def list_actions(
        self, *, run_id: str, user_id: str | None, is_admin: bool = False
    ) -> list[dict[str, Any]]:
        from deerflow.persistence.maf_roundtable.actions import action_to_public_dict

        run = await self.get_run(run_id=run_id, user_id=user_id, is_admin=is_admin)
        return [
            action_to_public_dict(action)
            for action in await self._store.actions.list_actions(run_id=run["id"])
        ]

    async def get_action(self, *, action_id: str, user_id: str | None, is_admin: bool = False) -> dict[str, Any]:
        from deerflow.persistence.maf_roundtable.actions import action_to_public_dict

        action = await self._store.actions.get_action(action_id)
        if action is None:
            raise MafNotFoundError("action", action_id)
        scope = self._read_scope_user(user_id, is_admin)
        session = await self._store.sessions.get_session(action["session_id"], user_id=scope)
        if session is None:
            raise MafNotFoundError("action", action_id)
        return action_to_public_dict(action)

    # ── approvals ──────────────────────────────────────────────────────────

    async def list_approvals(
        self, *, run_id: str, user_id: str | None, is_admin: bool = False
    ) -> list[dict[str, Any]]:
        run = await self.get_run(run_id=run_id, user_id=user_id, is_admin=is_admin)
        # Reads are observational. Expiry is performed by the dedicated
        # scanner; decision/resume paths retain the fail-closed pre-CAS check.
        return await self._store.approvals.list_approvals(run_id=run["id"])

    async def decide_approval(
        self,
        *,
        approval_id: str,
        decision: str,
        user_id: str | None,
        is_admin: bool = False,
    ) -> dict[str, Any]:
        self._require_enabled()
        approval = await self._store.approvals.get_approval(approval_id)
        if approval is None:
            raise MafNotFoundError("approval", approval_id)
        # Owner-only write: an admin cannot decide another user's approval.
        scope = self._write_scope_user(user_id, is_admin)
        session = await self._store.sessions.get_session(approval["session_id"], user_id=scope)
        if session is None:
            if is_admin and await self._store.sessions.get_session(approval["session_id"], user_id=None) is not None:
                raise MafForbiddenError("approval", approval_id)
            raise MafNotFoundError("approval", approval_id)
        # The timeout transition is server-owned and happens before the
        # decision CAS, so an expired approval can never be revived by a late
        # browser request.
        if approval.get("run_id"):
            await self._store.expire_due_approvals(run_id=approval["run_id"])
        try:
            updated = await self._store.approvals.decide(approval_id, decision=decision, decided_by=user_id)
        except ValueError as exc:
            raise MafConflictError(str(exc)) from exc
        if updated is None:
            raise MafNotFoundError("approval", approval_id)
        return updated