"""HTTP router for the MAF roundtable experiment at ``/api/maf-roundtable``.

Phase 1 ships:

- ``GET  /health`` — feature-flag probe (open read; the frontend reads
  ``enabled`` to decide whether to show the entry).
- Session CRUD: ``POST /sessions``, ``GET /sessions``, ``GET /sessions/{id}``,
  ``PUT /sessions/{id}``, ``DELETE /sessions/{id}``.
- Run lifecycle: ``POST /sessions/{id}/runs``, ``GET /runs/{id}``,
  ``GET /sessions/{id}/runs``, ``POST /runs/{id}/cancel``,
  ``POST /runs/{id}/resume``, ``POST /runs/{id}/recover``.
- Event stream: ``GET /runs/{id}/events?after_seq=N`` (断线补拉) — **read-only**;
  there is no public ``POST /runs/{id}/events``. Real workflow events are
  written only by the backend executor via ``service.append_event``.
- Actions: ``GET /runs/{id}/actions``, ``GET /actions/{id}``.
- Approvals: ``GET /runs/{id}/approvals``, ``POST /approvals/{id}/decision``.

User isolation (MAF-01-07): every read / write is scoped to the caller's
``user_id``; admins may *read* cross-user when ``admin_diagnostic_access`` is
on, but **writes** remain owner-only (admin write of another user's resource
→ 403). A non-admin user cannot read or mutate another user's session / run /
action / approval — the store returns ``None`` for out-of-scope rows and the
router surfaces that as ``404`` (not ``403``, so the existence of another
user's session is not leaked).

Feature flag (MAF-01-04): when ``maf_roundtable.enabled`` is off, write paths
return ``503 Service Disabled``; reads stay open so admins can review history.
The MAF SDK is **not** imported by this router — Phase 2+ imports it lazily
inside ``app.maf_roundtable.service`` only when a run is actually started.
"""

from __future__ import annotations

from typing import Any

from fastapi import APIRouter, HTTPException, Query, Request, status

from app.maf_roundtable import (
    ActionListResponse,
    ActionResponse,
    ApprovalDecisionRequest,
    ApprovalListResponse,
    ApprovalResponse,
    CreateRunRequest,
    CreateSessionRequest,
    EventListResponse,
    EventResponse,
    HealthResponse,
    MafConflictError,
    MafDisabledError,
    MafForbiddenError,
    MafNotFoundError,
    RunListResponse,
    RunResponse,
    SessionListResponse,
    SessionResponse,
    UpdateSessionRequest,
)

router = APIRouter(prefix="/api/maf-roundtable", tags=["maf-roundtable"])


# ── helpers ───────────────────────────────────────────────────────────────


def _get_service(request: Request):
    """Resolve the MAF service from app.state. Returns None when the feature
    is fully off (no store wired) — the router then 503s on writes.
    """
    return getattr(request.app.state, "maf_roundtable_service", None)


def _user_ctx(request: Request) -> tuple[str | None, bool]:
    """Return ``(user_id, is_admin)`` for the current request.

    When auth is disabled (no user stamped on request.state), the caller is
    treated as the ``"default"`` user (same convention as
    ``get_effective_user_id``) and non-admin — so a no-auth deployment stays
    owner-scoped to ``"default"`` rather than accidentally bypassing the
    owner filter.
    """
    user = getattr(request.state, "user", None)
    if user is None:
        return "default", False
    return str(user.id), getattr(user, "system_role", None) == "admin"


def _map_service_error(exc: Exception) -> HTTPException:
    if isinstance(exc, MafDisabledError):
        return HTTPException(status_code=503, detail="MAF roundtable feature is disabled")
    if isinstance(exc, MafForbiddenError):
        return HTTPException(status_code=403, detail=f"not the owner of {exc.what}")
    if isinstance(exc, MafNotFoundError):
        return HTTPException(status_code=404, detail=f"{exc.what} not found")
    if isinstance(exc, MafConflictError):
        return HTTPException(status_code=409, detail=str(exc))
    return HTTPException(status_code=500, detail=str(exc))


def _require_service(request: Request):
    svc = _get_service(request)
    if svc is None:
        raise HTTPException(status_code=503, detail="MAF roundtable store not available")
    return svc


# ── health ─────────────────────────────────────────────────────────────────


@router.get("/health", response_model=HealthResponse)
async def get_health(request: Request) -> HealthResponse:
    """Feature-flag probe. Open read — the frontend reads ``enabled`` to
    decide whether to show the MAF entry. Does not require the store.
    """
    from app.maf_roundtable.service import get_maf_write_guard_readiness
    from deerflow.config import get_app_config

    app_config = get_app_config()
    cfg = app_config.maf_roundtable
    svc = _get_service(request)
    store_available = svc is not None and svc.store is not None
    from app.maf_roundtable.dispatch_contracts import MafDispatchCatalog

    dispatch_ready, dispatch_status = MafDispatchCatalog(cfg).readiness()
    readiness = get_maf_write_guard_readiness(app_config=app_config)
    stream_ready = bool(getattr(request.app.state, "maf_conversation_stream_ready", False))
    stream_status = str(getattr(request.app.state, "maf_conversation_stream_status", "unknown"))
    conv_backend = str(getattr(request.app.state, "maf_conversation_event_backend", "memory"))
    return HealthResponse(
        enabled=bool(cfg.enabled and dispatch_ready and svc is not None and svc.enabled),
        conversation_dispatch_ready=bool(cfg.enabled and dispatch_ready and svc is not None and svc.enabled),
        conversation_dispatch_status=dispatch_status,
        conversation_stream_ready=stream_ready,
        conversation_stream_status=stream_status,
        conversation_event_backend=conv_backend,
        store_available=store_available,
        default_mode=cfg.default_mode,
        default_write_policy=cfg.default_write_policy,
        default_max_rounds=cfg.default_max_rounds,
        admin_diagnostic_access=cfg.admin_diagnostic_access,
        write_guard_enabled=readiness["enabled"],
        write_guard_ready=readiness["ready"],
        action_secret_configured=readiness["secret_configured"],
        write_guard_status=readiness["status"],
        sse_hub_type="in_memory",
        sse_multi_worker=False,
        sse_warning="Single-worker only. Multi-worker requires Redis Pub/Sub.",
    )


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


@router.post("/sessions", response_model=SessionResponse)
async def create_session(request: Request, body: CreateSessionRequest) -> SessionResponse:
    svc = _require_service(request)
    user_id, is_admin = _user_ctx(request)
    try:
        session = await svc.create_session(user_id=user_id, body=body.model_dump(), is_admin=is_admin)
    except (MafDisabledError, MafForbiddenError, MafNotFoundError, MafConflictError) as exc:
        raise _map_service_error(exc) from exc
    return SessionResponse(**session)


@router.get("/sessions", response_model=SessionListResponse)
async def list_sessions(
    request: Request,
    status: str | None = Query(default=None),
    mode: str | None = Query(default=None),
    limit: int = Query(default=100, ge=1, le=500),
    offset: int = Query(default=0, ge=0),
) -> SessionListResponse:
    svc = _require_service(request)
    user_id, is_admin = _user_ctx(request)
    rows, total = await svc.list_sessions(
        user_id=user_id, is_admin=is_admin, status=status, mode=mode, limit=limit, offset=offset
    )
    return SessionListResponse(sessions=[SessionResponse(**r) for r in rows], total=total)


@router.get("/sessions/{session_id}", response_model=SessionResponse)
async def get_session(request: Request, session_id: str) -> SessionResponse:
    svc = _require_service(request)
    user_id, is_admin = _user_ctx(request)
    try:
        session = await svc.get_session(session_id=session_id, user_id=user_id, is_admin=is_admin)
    except MafNotFoundError as exc:
        raise _map_service_error(exc) from exc
    return SessionResponse(**session)


@router.put("/sessions/{session_id}", response_model=SessionResponse)
async def update_session(request: Request, session_id: str, body: UpdateSessionRequest) -> SessionResponse:
    svc = _require_service(request)
    user_id, is_admin = _user_ctx(request)
    try:
        session = await svc.update_session(
            session_id=session_id, user_id=user_id, fields=body.model_dump(exclude_unset=True), is_admin=is_admin
        )
    except (MafDisabledError, MafForbiddenError, MafNotFoundError, MafConflictError) as exc:
        raise _map_service_error(exc) from exc
    return SessionResponse(**session)


@router.delete("/sessions/{session_id}")
async def delete_session(request: Request, session_id: str) -> dict[str, Any]:
    svc = _require_service(request)
    user_id, is_admin = _user_ctx(request)
    try:
        await svc.delete_session(session_id=session_id, user_id=user_id, is_admin=is_admin)
    except (MafDisabledError, MafForbiddenError, MafNotFoundError) as exc:
        raise _map_service_error(exc) from exc
    return {"deleted": True, "id": session_id}


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


@router.post("/sessions/{session_id}/runs", response_model=RunResponse)
async def create_run(request: Request, session_id: str, body: CreateRunRequest) -> RunResponse:
    svc = _require_service(request)
    user_id, is_admin = _user_ctx(request)

    # Check MAF SDK availability BEFORE creating the run — so the user gets a
    # clean 503 rather than a run that will immediately fail.
    from app.maf_roundtable.dispatcher import MafSdkNotAvailableError, get_dispatcher

    dispatcher = get_dispatcher(request.app)
    try:
        dispatcher.check_sdk_available()
    except MafSdkNotAvailableError as exc:
        raise HTTPException(status_code=503, detail=str(exc)) from exc

    try:
        run = await svc.create_run(
            session_id=session_id, user_id=user_id, workflow_version=body.workflow_version, is_admin=is_admin
        )
    except (MafDisabledError, MafForbiddenError, MafNotFoundError, MafConflictError) as exc:
        raise _map_service_error(exc) from exc

    # Launch the workflow in the background. The run is already persisted as
    # "pending"; the dispatcher transitions it to "running" then completes/fails it.
    # Auto-dispatch can be disabled (e.g. in tests) via app.state.maf_auto_dispatch=False.
    if getattr(request.app.state, "maf_auto_dispatch", True):
        session = await svc.get_session(session_id=session_id, user_id=user_id, is_admin=is_admin)
        await dispatcher.dispatch(session, run, user_id=user_id or "default")

    return RunResponse(**run)


@router.get("/sessions/{session_id}/runs", response_model=RunListResponse)
async def list_runs(request: Request, session_id: str) -> RunListResponse:
    svc = _require_service(request)
    user_id, is_admin = _user_ctx(request)
    try:
        runs = await svc.list_runs(session_id=session_id, user_id=user_id, is_admin=is_admin)
    except MafNotFoundError as exc:
        raise _map_service_error(exc) from exc
    return RunListResponse(runs=[RunResponse(**r) for r in runs])


@router.get("/runs/{run_id}", response_model=RunResponse)
async def get_run(request: Request, run_id: str) -> RunResponse:
    svc = _require_service(request)
    user_id, is_admin = _user_ctx(request)
    try:
        run = await svc.get_run(run_id=run_id, user_id=user_id, is_admin=is_admin)
    except MafNotFoundError as exc:
        raise _map_service_error(exc) from exc
    return RunResponse(**run)


@router.post("/runs/{run_id}/cancel", response_model=RunResponse)
async def cancel_run(request: Request, run_id: str) -> RunResponse:
    svc = _require_service(request)
    user_id, is_admin = _user_ctx(request)

    # Phase 1: atomically set run+session to cancelling (not yet cancelled).
    try:
        await svc.start_cancel(run_id=run_id, user_id=user_id, is_admin=is_admin)
    except (MafDisabledError, MafForbiddenError, MafNotFoundError) as exc:
        raise _map_service_error(exc) from exc

    # Phase 2: cascade cancel to the background task + DeerFlow sub-runs.
    from app.maf_roundtable.dispatcher import get_dispatcher

    dispatcher = get_dispatcher(request.app)
    result = await dispatcher.cancel(run_id)

    # Phase 3: finalize based on the structured cancel result.
    # - all_cancelled → finalize to ``cancelled`` (terminal).
    # - some sub-runs failed/still active → mark ``cancel_failed`` (non-terminal)
    #   so the failure is visible and retryable. A failed cancel must NEVER
    #   masquerade as cancelled.
    try:
        if result.get("all_cancelled"):
            final_run = await svc.finalize_cancel(run_id=run_id, user_id=user_id, is_admin=is_admin)
        else:
            failed = result.get("failed_run_ids") or []
            still_active = result.get("still_active_run_ids") or []
            detail = []
            if failed:
                detail.append(f"{len(failed)} sub-run(s) failed to cancel")
            if still_active:
                detail.append(f"{len(still_active)} sub-run(s) had no DeerFlow run id")
            reason = "; ".join(detail) or "cancel cascade incomplete"
            final_run = await svc.mark_cancel_failed(
                run_id=run_id, user_id=user_id, is_admin=is_admin, reason=reason
            )
    except (MafDisabledError, MafForbiddenError, MafNotFoundError) as exc:
        raise _map_service_error(exc) from exc

    return RunResponse(**final_run)


@router.post("/runs/{run_id}/resume", response_model=RunResponse)
async def resume_run(request: Request, run_id: str) -> RunResponse:
    """Resume after approvals are decided (owner-only).

    If the run is not ``waiting_approval`` or still has pending approvals,
    returns 409.
    """
    svc = _require_service(request)
    user_id, is_admin = _user_ctx(request)
    try:
        run = await svc.get_run(run_id=run_id, user_id=user_id, is_admin=False)
    except (MafDisabledError, MafForbiddenError, MafNotFoundError) as exc:
        raise _map_service_error(exc) from exc

    if run.get("status") != "waiting_approval":
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=f"run {run_id} is not waiting_approval (status={run.get('status')})",
        )

    from app.maf_roundtable.dispatcher import get_dispatcher

    dispatcher = get_dispatcher(request.app)
    try:
        updated = await dispatcher.resume(run_id, user_id=user_id or "default")
    except Exception as exc:  # noqa: BLE001
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail=f"resume failed: {type(exc).__name__}",
        ) from exc
    if updated is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="run not found")
    return RunResponse(**updated)


@router.post("/runs/{run_id}/recover", response_model=RunResponse)
async def recover_run(request: Request, run_id: str) -> RunResponse:
    """Graph-level recover after process restart (owner-only).

    Re-enters the workflow from checkpoint; completed_nodes are skipped.
    Does not silently scan all runs at startup.
    """
    svc = _require_service(request)
    user_id, is_admin = _user_ctx(request)
    try:
        run = await svc.get_run(run_id=run_id, user_id=user_id, is_admin=False)
    except (MafDisabledError, MafForbiddenError, MafNotFoundError) as exc:
        raise _map_service_error(exc) from exc

    if run.get("status") not in ("running", "waiting_approval"):
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=f"run {run_id} cannot be recovered (status={run.get('status')})",
        )

    from app.maf_roundtable.dispatcher import get_dispatcher

    dispatcher = get_dispatcher(request.app)
    try:
        updated = await dispatcher.recover(run_id, user_id=user_id or "default")
    except Exception as exc:  # noqa: BLE001
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail=f"recover failed: {type(exc).__name__}",
        ) from exc
    if updated is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="run not found")
    return RunResponse(**updated)


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


@router.get("/runs/{run_id}/events", response_model=EventListResponse)
async def list_events(
    request: Request,
    run_id: str,
    after_seq: int = Query(default=0, ge=0),
    limit: int = Query(default=1000, ge=1, le=10000),
) -> EventListResponse:
    svc = _require_service(request)
    user_id, is_admin = _user_ctx(request)
    try:
        events, latest = await svc.list_events(
            run_id=run_id, user_id=user_id, after_seq=after_seq, limit=limit, is_admin=is_admin
        )
    except MafNotFoundError as exc:
        raise _map_service_error(exc) from exc
    return EventListResponse(events=[EventResponse(**e) for e in events], latest_seq=latest)


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


@router.get("/runs/{run_id}/actions", response_model=ActionListResponse)
async def list_actions(request: Request, run_id: str) -> ActionListResponse:
    svc = _require_service(request)
    user_id, is_admin = _user_ctx(request)
    try:
        actions = await svc.list_actions(run_id=run_id, user_id=user_id, is_admin=is_admin)
    except MafNotFoundError as exc:
        raise _map_service_error(exc) from exc
    return ActionListResponse(actions=[ActionResponse(**a) for a in actions])


@router.get("/actions/{action_id}", response_model=ActionResponse)
async def get_action(request: Request, action_id: str) -> ActionResponse:
    svc = _require_service(request)
    user_id, is_admin = _user_ctx(request)
    try:
        action = await svc.get_action(action_id=action_id, user_id=user_id, is_admin=is_admin)
    except MafNotFoundError as exc:
        raise _map_service_error(exc) from exc
    return ActionResponse(**action)


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


@router.get("/runs/{run_id}/approvals", response_model=ApprovalListResponse)
async def list_approvals(request: Request, run_id: str) -> ApprovalListResponse:
    svc = _require_service(request)
    user_id, is_admin = _user_ctx(request)
    try:
        approvals = await svc.list_approvals(run_id=run_id, user_id=user_id, is_admin=is_admin)
    except MafNotFoundError as exc:
        raise _map_service_error(exc) from exc
    return ApprovalListResponse(approvals=[ApprovalResponse(**a) for a in approvals])


@router.post("/approvals/{approval_id}/decision", response_model=ApprovalResponse)
async def decide_approval(request: Request, approval_id: str, body: ApprovalDecisionRequest) -> ApprovalResponse:
    svc = _require_service(request)
    user_id, is_admin = _user_ctx(request)
    try:
        approval = await svc.decide_approval(
            approval_id=approval_id, decision=body.decision, user_id=user_id, is_admin=is_admin
        )
    except (MafDisabledError, MafForbiddenError, MafNotFoundError, MafConflictError) as exc:
        raise _map_service_error(exc) from exc

    # P1: after a successful approval decision, resume the workflow from its
    # checkpoint. Only the owner can resume (decide_approval already enforced
    # owner-only write). The resume loads the checkpoint, restores
    # RoundtableState.from_dict(), and executes the approved action (at-most-once).
    run_id = approval.get("run_id")
    if run_id and body.decision in ("approved", "rejected"):
        from app.maf_roundtable.dispatcher import get_dispatcher

        dispatcher = get_dispatcher(request.app)
        try:
            await dispatcher.resume(run_id, user_id=user_id or "default")
        except Exception:
            # Resume failure should not fail the approval response itself —
            # the approval is already persisted. Log for observability.
            import logging

            logging.getLogger(__name__).exception(
                "MAF resume failed for run %s after approval %s", run_id, approval_id
            )

    return ApprovalResponse(**approval)