openclaw-mission-control/backend/app/main.py

"""FastAPI application entrypoint and router wiring for the backend."""

from __future__ import annotations

from contextlib import asynccontextmanager
from typing import TYPE_CHECKING, Any

from fastapi import APIRouter, FastAPI, status
from fastapi.middleware.cors import CORSMiddleware
from fastapi.openapi.utils import get_openapi
from fastapi_pagination import add_pagination

from app.api.activity import router as activity_router
from app.api.agent import router as agent_router
from app.api.agents import router as agents_router
from app.api.approvals import router as approvals_router
from app.api.auth import router as auth_router
from app.api.board_group_memory import router as board_group_memory_router
from app.api.board_groups import router as board_groups_router
from app.api.board_memory import router as board_memory_router
from app.api.board_onboarding import router as board_onboarding_router
from app.api.board_webhooks import router as board_webhooks_router
from app.api.boards import router as boards_router
from app.api.gateway import router as gateway_router
from app.api.gateways import router as gateways_router
from app.api.metrics import router as metrics_router
from app.api.organizations import router as organizations_router
from app.api.skills_marketplace import router as skills_marketplace_router
from app.api.souls_directory import router as souls_directory_router
from app.api.tags import router as tags_router
from app.api.task_custom_fields import router as task_custom_fields_router
from app.api.tasks import router as tasks_router
from app.api.users import router as users_router
from app.core.config import settings
from app.core.error_handling import install_error_handling
from app.core.logging import configure_logging, get_logger
from app.core.rate_limit import validate_rate_limit_redis
from app.core.rate_limit_backend import RateLimitBackend
from app.core.security_headers import SecurityHeadersMiddleware
from app.db.session import init_db
from app.schemas.health import HealthStatusResponse

if TYPE_CHECKING:
    from collections.abc import AsyncIterator

configure_logging()
logger = get_logger(__name__)
OPENAPI_TAGS = [
    {
        "name": "auth",
        "description": (
            "Authentication bootstrap endpoints for resolving caller identity and session context."
        ),
    },
    {
        "name": "health",
        "description": (
            "Service liveness/readiness probes used by infrastructure and runtime checks."
        ),
    },
    {
        "name": "agents",
        "description": "Organization-level agent directory, lifecycle, and management operations.",
    },
    {
        "name": "activity",
        "description": "Activity feed and audit timeline endpoints across boards and operations.",
    },
    {
        "name": "gateways",
        "description": "Gateway management, synchronization, and runtime control operations.",
    },
    {
        "name": "metrics",
        "description": "Aggregated operational and board analytics metrics endpoints.",
    },
    {
        "name": "organizations",
        "description": "Organization profile, membership, and governance management endpoints.",
    },
    {
        "name": "souls-directory",
        "description": "Directory and lookup endpoints for agent soul templates and variants.",
    },
    {
        "name": "skills",
        "description": "Skills marketplace, install, uninstall, and synchronization endpoints.",
    },
    {
        "name": "board-groups",
        "description": "Board group CRUD, assignment, and grouping workflow endpoints.",
    },
    {
        "name": "board-group-memory",
        "description": "Shared memory endpoints scoped to board groups and grouped boards.",
    },
    {
        "name": "boards",
        "description": "Board lifecycle, configuration, and board-level management endpoints.",
    },
    {
        "name": "board-memory",
        "description": "Board-scoped memory read/write endpoints for persistent context.",
    },
    {
        "name": "board-webhooks",
        "description": "Board webhook registration, delivery config, and lifecycle endpoints.",
    },
    {
        "name": "board-onboarding",
        "description": "Board onboarding state, setup actions, and onboarding workflow endpoints.",
    },
    {
        "name": "approvals",
        "description": "Approval request, review, and status-tracking operations for board tasks.",
    },
    {
        "name": "tasks",
        "description": "Task CRUD, dependency management, and task workflow operations.",
    },
    {
        "name": "custom-fields",
        "description": "Organization custom-field definitions and board assignment endpoints.",
    },
    {
        "name": "tags",
        "description": "Tag catalog and task-tag association management endpoints.",
    },
    {
        "name": "users",
        "description": "User profile read/update operations and user-centric settings endpoints.",
    },
    {
        "name": "agent",
        "description": (
            "Agent-scoped API surface. All endpoints require `X-Agent-Token` and are "
            "constrained by agent board access policies."
        ),
    },
    {
        "name": "agent-lead",
        "description": (
            "Lead workflows: delegation, review orchestration, approvals, and "
            "coordination actions."
        ),
    },
    {
        "name": "agent-worker",
        "description": (
            "Worker workflows: task execution, task comments, and board/group context "
            "reads/writes used during heartbeat loops."
        ),
    },
    {
        "name": "agent-main",
        "description": (
            "Gateway-main control workflows that message board leads or broadcast "
            "coordination requests."
        ),
    },
]

_JSON_SCHEMA_REF_PREFIX = "#/components/schemas/"
_OPENAPI_EXAMPLE_TAGS = {
    "agents",
    "activity",
    "gateways",
    "metrics",
    "organizations",
    "souls-directory",
    "skills",
    "board-groups",
    "board-group-memory",
    "boards",
    "board-memory",
    "board-webhooks",
    "board-onboarding",
    "approvals",
    "tasks",
    "custom-fields",
    "tags",
    "users",
}
_GENERIC_RESPONSE_DESCRIPTIONS = {"Successful Response", "Validation Error"}
_HTTP_RESPONSE_DESCRIPTIONS = {
    "200": "Request completed successfully.",
    "201": "Resource created successfully.",
    "202": "Request accepted for processing.",
    "204": "Request completed successfully with no response body.",
    "400": "Request validation failed.",
    "401": "Authentication is required or token is invalid.",
    "403": "Caller is authenticated but not authorized for this operation.",
    "404": "Requested resource was not found.",
    "409": "Request conflicts with the current resource state.",
    "422": "Request payload failed schema or field validation.",
    "429": "Request was rate-limited.",
    "500": "Internal server error.",
}
_METHOD_SUMMARY_PREFIX = {
    "get": "List",
    "post": "Create",
    "put": "Replace",
    "patch": "Update",
    "delete": "Delete",
}


def _resolve_schema_ref(
    schema: dict[str, Any],
    *,
    components: dict[str, Any],
    seen_refs: set[str] | None = None,
) -> dict[str, Any]:
    """Resolve local component refs for OpenAPI schema traversal."""
    ref = schema.get("$ref")
    if not isinstance(ref, str):
        return schema
    if not ref.startswith(_JSON_SCHEMA_REF_PREFIX):
        return schema
    if seen_refs is None:
        seen_refs = set()
    if ref in seen_refs:
        return schema
    seen_refs.add(ref)
    schema_name = ref[len(_JSON_SCHEMA_REF_PREFIX) :]
    schemas = components.get("schemas")
    if not isinstance(schemas, dict):
        return schema
    target = schemas.get(schema_name)
    if not isinstance(target, dict):
        return schema
    return _resolve_schema_ref(target, components=components, seen_refs=seen_refs)


def _example_from_schema(schema: dict[str, Any], *, components: dict[str, Any]) -> Any:
    """Generate an OpenAPI example from schema metadata with sensible fallbacks."""
    resolved = _resolve_schema_ref(schema, components=components)

    if "example" in resolved:
        return resolved["example"]
    examples = resolved.get("examples")
    if isinstance(examples, list) and examples:
        return examples[0]

    for composite_key in ("anyOf", "oneOf", "allOf"):
        composite = resolved.get(composite_key)
        if isinstance(composite, list):
            for branch in composite:
                if not isinstance(branch, dict):
                    continue
                branch_example = _example_from_schema(branch, components=components)
                if branch_example is not None:
                    return branch_example

    enum_values = resolved.get("enum")
    if isinstance(enum_values, list) and enum_values:
        return enum_values[0]

    schema_type = resolved.get("type")
    if schema_type == "object":
        output: dict[str, Any] = {}
        properties = resolved.get("properties")
        if isinstance(properties, dict):
            for key, property_schema in properties.items():
                if not isinstance(property_schema, dict):
                    continue
                property_example = _example_from_schema(property_schema, components=components)
                if property_example is not None:
                    output[key] = property_example
        if output:
            return output
        additional_properties = resolved.get("additionalProperties")
        if isinstance(additional_properties, dict):
            value_example = _example_from_schema(additional_properties, components=components)
            if value_example is not None:
                return {"key": value_example}
        return {}

    if schema_type == "array":
        items = resolved.get("items")
        if isinstance(items, dict):
            item_example = _example_from_schema(items, components=components)
            if item_example is not None:
                return [item_example]
        return []

    if schema_type == "string":
        return "string"
    if schema_type == "integer":
        return 0
    if schema_type == "number":
        return 0
    if schema_type == "boolean":
        return False

    return None


def _inject_json_content_example(
    *,
    content: dict[str, Any],
    components: dict[str, Any],
) -> None:
    """Attach an example to application/json content when one is missing."""
    app_json = content.get("application/json")
    if not isinstance(app_json, dict):
        return
    if "example" in app_json or "examples" in app_json:
        return
    schema = app_json.get("schema")
    if not isinstance(schema, dict):
        return
    generated_example = _example_from_schema(schema, components=components)
    if generated_example is not None:
        app_json["example"] = generated_example


def _build_operation_summary(*, method: str, path: str) -> str:
    """Build a readable summary when an operation does not define one."""
    prefix = _METHOD_SUMMARY_PREFIX.get(method.lower(), "Handle")
    path_without_prefix = path.removeprefix("/api/v1/")
    parts = [
        part.replace("-", " ")
        for part in path_without_prefix.split("/")
        if part and not (part.startswith("{") and part.endswith("}"))
    ]
    if not parts:
        return prefix
    return f"{prefix} {' '.join(parts)}".strip().title()


def _normalize_operation_docs(
    *,
    operation: dict[str, Any],
    method: str,
    path: str,
) -> None:
    """Normalize summary/description/responses/request-body docs for tagged operations."""
    summary = str(operation.get("summary", "")).strip()
    if not summary:
        summary = _build_operation_summary(method=method, path=path)
        operation["summary"] = summary

    description = str(operation.get("description", "")).strip()
    if not description:
        operation["description"] = f"{summary}."

    request_body = operation.get("requestBody")
    if isinstance(request_body, dict):
        if not str(request_body.get("description", "")).strip():
            request_body["description"] = "JSON request payload."

    responses = operation.get("responses")
    if not isinstance(responses, dict):
        return
    for status_code, response in responses.items():
        if not isinstance(response, dict):
            continue
        existing_description = str(response.get("description", "")).strip()
        if not existing_description or existing_description in _GENERIC_RESPONSE_DESCRIPTIONS:
            response["description"] = _HTTP_RESPONSE_DESCRIPTIONS.get(
                str(status_code),
                "Request processed.",
            )


def _inject_tagged_operation_openapi_docs(openapi_schema: dict[str, Any]) -> None:
    """Ensure targeted-tag operations expose consistent OpenAPI docs and examples."""
    components = openapi_schema.get("components")
    if not isinstance(components, dict):
        return
    paths = openapi_schema.get("paths")
    if not isinstance(paths, dict):
        return

    for path, path_item in paths.items():
        if not isinstance(path_item, dict):
            continue
        for method, operation in path_item.items():
            if not isinstance(operation, dict):
                continue
            tags = operation.get("tags")
            if not isinstance(tags, list):
                continue
            if not _OPENAPI_EXAMPLE_TAGS.intersection(tags):
                continue

            _normalize_operation_docs(operation=operation, method=method, path=path)

            request_body = operation.get("requestBody")
            if isinstance(request_body, dict):
                request_content = request_body.get("content")
                if isinstance(request_content, dict):
                    _inject_json_content_example(content=request_content, components=components)

            responses = operation.get("responses")
            if isinstance(responses, dict):
                for response in responses.values():
                    if not isinstance(response, dict):
                        continue
                    response_content = response.get("content")
                    if isinstance(response_content, dict):
                        _inject_json_content_example(
                            content=response_content, components=components
                        )


def _build_custom_openapi(fastapi_app: FastAPI) -> dict[str, Any]:
    """Generate OpenAPI schema with normalized docs/examples for targeted tags."""
    if fastapi_app.openapi_schema:
        return fastapi_app.openapi_schema
    openapi_schema = get_openapi(
        title=fastapi_app.title,
        version=fastapi_app.version,
        openapi_version=fastapi_app.openapi_version,
        description=fastapi_app.description,
        routes=fastapi_app.routes,
        tags=fastapi_app.openapi_tags,
        servers=fastapi_app.servers,
    )
    _inject_tagged_operation_openapi_docs(openapi_schema)
    fastapi_app.openapi_schema = openapi_schema
    return fastapi_app.openapi_schema


class MissionControlFastAPI(FastAPI):
    """FastAPI application with custom OpenAPI normalization."""

    def openapi(self) -> dict[str, Any]:
        return _build_custom_openapi(self)


@asynccontextmanager
async def lifespan(_: FastAPI) -> AsyncIterator[None]:
    """Initialize application resources before serving requests."""
    logger.info(
        "app.lifecycle.starting environment=%s db_auto_migrate=%s",
        settings.environment,
        settings.db_auto_migrate,
    )
    await init_db()
    if settings.rate_limit_backend == RateLimitBackend.REDIS:
        validate_rate_limit_redis(settings.rate_limit_redis_url)
        logger.info("app.lifecycle.rate_limit backend=redis")
    else:
        logger.info("app.lifecycle.rate_limit backend=memory")
    logger.info("app.lifecycle.started")
    try:
        yield
    finally:
        logger.info("app.lifecycle.stopped")


app = MissionControlFastAPI(
    title="Mission Control API",
    version="0.1.0",
    lifespan=lifespan,
    openapi_tags=OPENAPI_TAGS,
)

origins = [o.strip() for o in settings.cors_origins.split(",") if o.strip()]
if origins:
    app.add_middleware(
        CORSMiddleware,
        allow_origins=origins,
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
        expose_headers=["X-Total-Count", "X-Limit", "X-Offset"],
    )
    logger.info("app.cors.enabled origins_count=%s", len(origins))
else:
    logger.info("app.cors.disabled")

app.add_middleware(
    SecurityHeadersMiddleware,
    x_content_type_options=settings.security_header_x_content_type_options,
    x_frame_options=settings.security_header_x_frame_options,
    referrer_policy=settings.security_header_referrer_policy,
    permissions_policy=settings.security_header_permissions_policy,
)
install_error_handling(app)


@app.get(
    "/health",
    tags=["health"],
    response_model=HealthStatusResponse,
    summary="Health Check",
    description="Lightweight liveness probe endpoint.",
    responses={
        status.HTTP_200_OK: {
            "description": "Service is alive.",
            "content": {"application/json": {"example": {"ok": True}}},
        }
    },
)
def health() -> HealthStatusResponse:
    """Lightweight liveness probe endpoint."""
    return HealthStatusResponse(ok=True)


@app.get(
    "/healthz",
    tags=["health"],
    response_model=HealthStatusResponse,
    summary="Health Alias Check",
    description="Alias liveness probe endpoint for platform compatibility.",
    responses={
        status.HTTP_200_OK: {
            "description": "Service is alive.",
            "content": {"application/json": {"example": {"ok": True}}},
        }
    },
)
def healthz() -> HealthStatusResponse:
    """Alias liveness probe endpoint for platform compatibility."""
    return HealthStatusResponse(ok=True)


@app.get(
    "/readyz",
    tags=["health"],
    response_model=HealthStatusResponse,
    summary="Readiness Check",
    description="Readiness probe endpoint for service orchestration checks.",
    responses={
        status.HTTP_200_OK: {
            "description": "Service is ready.",
            "content": {"application/json": {"example": {"ok": True}}},
        }
    },
)
def readyz() -> HealthStatusResponse:
    """Readiness probe endpoint for service orchestration checks."""
    return HealthStatusResponse(ok=True)


api_v1 = APIRouter(prefix="/api/v1")
api_v1.include_router(auth_router)
api_v1.include_router(agent_router)
api_v1.include_router(agents_router)
api_v1.include_router(activity_router)
api_v1.include_router(gateway_router)
api_v1.include_router(gateways_router)
api_v1.include_router(metrics_router)
api_v1.include_router(organizations_router)
api_v1.include_router(souls_directory_router)
api_v1.include_router(skills_marketplace_router)
api_v1.include_router(board_groups_router)
api_v1.include_router(board_group_memory_router)
api_v1.include_router(boards_router)
api_v1.include_router(board_memory_router)
api_v1.include_router(board_webhooks_router)
api_v1.include_router(board_onboarding_router)
api_v1.include_router(approvals_router)
api_v1.include_router(tasks_router)
api_v1.include_router(task_custom_fields_router)
api_v1.include_router(tags_router)
api_v1.include_router(users_router)
app.include_router(api_v1)

add_pagination(app)
logger.debug("app.routes.registered count=%s", len(app.routes))