app/backend/app/main.py

import asyncio
import contextlib
from collections.abc import AsyncIterator
from contextlib import AbstractAsyncContextManager, asynccontextmanager

import sentry_sdk
from fastapi import FastAPI
from fastapi.concurrency import run_in_threadpool
from fastapi.responses import JSONResponse
from fastapi.routing import APIRoute
from starlette.middleware.cors import CORSMiddleware

from app.api.main import api_router
from app.api.routes.alerts import read_config as read_alerts_config
from app.cloud import config as cloud_config
from app.core import security
from app.core.config import settings
from app.flow import logs, modules
from app.flow.alerts import AlertManager
from app.flow.artifacts import ArtifactStore
from app.flow.controller import FlowController
from app.flow.dashboards import DashboardStore
from app.flow.events import event_bus
from app.flow.executor import ExecutionService
from app.flow.metrics import MetricsCollector
from app.flow.nodes.http import close_shared_client
from app.flow.pipeline import ValueSource
from app.flow.plugins import load_plugins
from app.flow.queue import MemoryWorkQueue, RedisWorkQueue, WorkQueue
from app.flow.remote import RemoteWorkerHub
from app.flow.runs import RUN_STATE_TTL, RunService
from app.flow.secrets import init_secrets
from app.flow.state import MemoryState, RedisState, StateBackend
from app.flow.store import FlowStore
from app.flow.watchdog import LoopWatchdog
from app.flow.worker_main import ARTIFACT_DIR_ENV
from app.flow.workers import PythonWorkerPool


def custom_generate_unique_id(route: APIRoute) -> str:
    return f"{route.tags[0]}-{route.name}"


if settings.SENTRY_DSN and settings.ENVIRONMENT != "local":
    sentry_sdk.init(dsn=str(settings.SENTRY_DSN), enable_tracing=True)


def _state_backend() -> StateBackend:
    if settings.REDIS_HOST:
        return RedisState(host=settings.REDIS_HOST, port=settings.REDIS_PORT)
    return MemoryState()


def _work_queue(namespace: str = "pipeline") -> WorkQueue:
    """Redis makes queued work survive the process; memory does not pretend to."""
    if settings.REDIS_HOST:
        return RedisWorkQueue(
            host=settings.REDIS_HOST, port=settings.REDIS_PORT, namespace=namespace
        )
    return MemoryWorkQueue()


def _run_state(namespace: str) -> StateBackend:
    """A state backend of a run's own, which is what isolates it.

    It expires: a finished run's messages are read out into its result, and
    what is left is only worth keeping while someone might look at it.
    """
    if settings.REDIS_HOST:
        return RedisState(
            host=settings.REDIS_HOST,
            port=settings.REDIS_PORT,
            namespace=namespace,
            ttl=RUN_STATE_TTL,
        )
    return MemoryState()


def _mcp_sessions() -> AbstractAsyncContextManager[None]:
    """The MCP session manager's run scope, or nothing when MCP is off."""
    if not settings.MCP_ENABLED:
        return contextlib.nullcontext()
    from app.mcp.server import mcp as mcp_server

    return mcp_server.session_manager.run()


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    """Start the flow engine alongside the API."""
    event_bus.bind(asyncio.get_running_loop())
    # Node code is user code, and `print` is how it says things.
    logs.install()
    init_secrets(settings.SECRETS_FILE, settings.SECRET_KEY)
    # Connectors register their node types before any flow is built with them.
    load_plugins()

    alerts = AlertManager(event_bus, config=read_alerts_config())
    execution = ExecutionService(
        queue=_work_queue(),
        max_workers=settings.FLOW_MAX_WORKERS,
        events=event_bus,
    )
    store = FlowStore(settings.FLOWS_DIR)
    # The packages node code imports, before anything tries to import them.
    await run_in_threadpool(modules.reconcile, store)
    # Beside the flows rather than in them: an artifact is what a run produced,
    # not something anyone wrote, so it has no business in the git repository.
    artifacts = ArtifactStore(settings.FLOWS_DIR.parent / "artifacts")
    app.state.artifact_store = artifacts
    pool = PythonWorkerPool(
        python=modules.venv_python(),
        size=settings.FLOW_MAX_WORKERS,
        events=event_bus,
        # A worker in this container writes to the store directly; a remote one
        # is given a URL instead. Node code calls the same two functions.
        env={ARTIFACT_DIR_ENV: str(artifacts.root)},
    )
    pool.start()
    app.state.worker_pool = pool
    worker_hub = RemoteWorkerHub()
    app.state.worker_hub = worker_hub
    controller = FlowController(
        store=store,
        state=_state_backend(),
        events=event_bus,
        max_workers=settings.FLOW_MAX_WORKERS,
        fastapi_app=app,
        execution=execution,
        alerts=alerts,
        workers=pool,
        remote=worker_hub,
    )
    app.state.flow_controller = controller
    # A "dashboard" alert channel puts its alert into the graph. Bound here
    # rather than passed in: the manager is built before the controller is.
    alerts.publish = lambda name, value: controller.publish_message(
        name, value, ValueSource(kind="api", id="alerts", label="Alerts")
    )
    dashboards = DashboardStore(controller.store)
    app.state.dashboard_store = dashboards
    controller.dashboards = dashboards
    # Charts need a deeper series than the default; tell the engine
    # before it starts recording.
    controller.set_history_limits(dashboards.history_requirements())
    # Runs read from a stream of their own: a burst of sweep runs must not
    # stand between the automations and their work, and a run that takes an
    # hour must not be judged by the cascade reaper's timings.
    run_service = RunService(
        controller=controller,
        queue=_work_queue("run"),
        state_factory=_run_state,
    )
    app.state.run_service = run_service
    watchdog = LoopWatchdog(event_bus)
    app.state.watchdog = watchdog
    watchdog_task = asyncio.create_task(watchdog.run(), name="loop-watchdog")
    alerts_task = asyncio.create_task(alerts.run(), name="alert-manager")
    metrics_task = asyncio.create_task(
        MetricsCollector(event_bus).run(), name="metrics-collector"
    )
    await controller.start()
    run_service.start()
    # Optional, and off unless someone enrolled this installation: the
    # connector dials the portal, nothing dials in.
    cloud_task: asyncio.Task[None] | None = None
    app.state.cloud_connector = None
    app.state.cloud_task = None
    if cloud_config.exists():
        from app.cloud.connector import CloudConnector

        connector = CloudConnector(app)
        app.state.cloud_connector = connector
        cloud_task = asyncio.create_task(
            connector.serve_forever(), name="cloud-connector"
        )
        app.state.cloud_task = cloud_task
    try:
        # A mounted sub-app gets no lifespan of its own, so the MCP session
        # manager is entered here; without it every /mcp request fails.
        async with _mcp_sessions():
            yield
    finally:
        watchdog_task.cancel()
        alerts_task.cancel()
        metrics_task.cancel()
        # Re-read from app.state: enrolling at runtime replaces this.
        running_cloud = getattr(app.state, "cloud_task", None) or cloud_task
        if running_cloud is not None:
            running_cloud.cancel()
        await run_in_threadpool(run_service.stop)
        await controller.stop()
        pool.stop()
        close_shared_client()
        if settings.MCP_ENABLED:
            from app.mcp.http import aclose

            await aclose()


app = FastAPI(
    title=settings.PROJECT_NAME,
    openapi_url=f"{settings.API_V1_STR}/openapi.json",
    generate_unique_id_function=custom_generate_unique_id,
    lifespan=lifespan,
)

# Set all CORS enabled origins
if settings.all_cors_origins:
    app.add_middleware(
        CORSMiddleware,
        allow_origins=settings.all_cors_origins,
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )

app.include_router(api_router, prefix=settings.API_V1_STR)


# Tagged because the operation-id builder reads the first tag; the route
# itself stays out of the schema.
@app.get(
    "/.well-known/oauth-authorization-server",
    include_in_schema=False,
    tags=["oauth"],
)
def oauth_authorization_server() -> JSONResponse:
    """RFC 8414 metadata, so an agent can find its way in unaided.

    The authorization endpoint is the dashboard rather than the API: approving
    a client needs a signed-in human, and the browser session lives there.
    """
    issuer = settings.oauth_issuer
    return JSONResponse(
        content={
            "issuer": issuer,
            "authorization_endpoint": (
                f"{settings.FRONTEND_HOST.rstrip('/')}/oauth/authorize"
            ),
            "token_endpoint": f"{issuer}{settings.API_V1_STR}/oauth/token",
            "registration_endpoint": f"{issuer}{settings.API_V1_STR}/oauth/register",
            "jwks_uri": f"{issuer}/.well-known/jwks.json",
            "response_types_supported": ["code"],
            "grant_types_supported": ["authorization_code", "refresh_token"],
            "code_challenge_methods_supported": ["S256"],
            "token_endpoint_auth_methods_supported": ["none"],
            "scopes_supported": [security.MCP_SCOPE],
        },
        headers={"Cache-Control": "public, max-age=3600"},
    )


@app.get("/.well-known/jwks.json", include_in_schema=False, tags=["oauth"])
def jwks() -> JSONResponse:
    """The public half of the MCP signing key."""
    return JSONResponse(
        content=security.public_jwks(),
        headers={"Cache-Control": "public, max-age=3600"},
    )


# Mounted last, and at the root: the SDK serves both /mcp and the protected
# resource metadata that has to sit beside it, so mounting under /mcp would put
# that metadata somewhere no client looks for it.
if settings.MCP_ENABLED:
    from app.mcp.http import build_http_app

    app.mount("/", build_http_app(app))