"""Topology-driven fallback for URL-typed ``Settings`` fields. Phase 0 of the AlphaSwarm infra-expansion plan. Centralization rule (plan section C.1): Resolution order: hardcoded default -> env (legacy ``ALPHASWARM_*``) -> topology.yaml. Pydantic-settings already implements the first two layers. This module adds the third: when a URL field has not been set via ``ALPHASWARM_*`` env (i.e. it's still at its hardcoded default) AND the deployment topology declares an endpoint for the matching service, the topology value wins. The fallback is intentionally narrow: - Only fields with ``URL_FALLBACK_FIELDS`` mapping entries are considered. - The mapping is explicit per (Settings field, topology service id, endpoint name) so we never accidentally repoint settings to an unrelated topology entry. - Setting any ``ALPHASWARM_*`` env var on a covered field disables the fallback for that field (env always wins). - A topology load failure logs a warning and leaves Settings untouched. Misbehaving topology files never break boot. The frontend / control-plane side reads service URLs through the ``/manage/topology/services/{id}/endpoint`` route (the canonical admin/control surface). AlphaSwarm-side processes (Celery workers, FastAPI app, CLI) call this fallback once at boot via :func:`apply_topology_fallback`. """ from __future__ import annotations import logging from typing import Any, Iterable, NamedTuple from pydantic_settings import BaseSettings logger = logging.getLogger(__name__) class _Mapping(NamedTuple): """Map a Settings field to a topology endpoint.""" settings_field: str service_id: str endpoint_name: str # Explicit mapping table. Each row says: when topology declares # ``endpoints[]`` on the service whose ``id`` is # ````, use that URL as the fallback for the matching # ``Settings`` field. Anything not in this table is unaffected. URL_FALLBACK_FIELDS: tuple[_Mapping, ...] = ( # --- AlphaSwarm-owned shared services (`alphaswarm-*` namespaces in # `alphaswarm/deployments/kubernetes/`). _Mapping("postgres_dsn", "postgres", "dsn"), _Mapping("postgres_async_dsn", "postgres", "async_dsn"), _Mapping("redis_url", "redis", "url"), _Mapping("redis_pubsub_url", "redis", "pubsub_url"), _Mapping("cache_redis_url", "redis", "cache_url"), _Mapping("minio_endpoint_url", "minio", "endpoint"), _Mapping("s3_endpoint_url", "minio", "endpoint"), _Mapping("kafka_bootstrap", "kafka", "bootstrap"), _Mapping("kafka_admin_bootstrap", "kafka", "admin_bootstrap"), _Mapping("schema_registry_url", "schema-registry", "ccompat"), _Mapping("kafka_admin_schema_registry_url", "schema-registry", "ccompat"), _Mapping("polaris_base_url", "polaris", "rest"), _Mapping("iceberg_rest_uri", "polaris", "iceberg_rest"), _Mapping("mlflow_tracking_uri", "mlflow", "tracking"), _Mapping("mlflow_registry_uri", "mlflow", "tracking"), _Mapping("dagster_webserver_url", "dagster", "webserver"), _Mapping("dagster_graphql_url", "dagster", "graphql"), _Mapping("airbyte_base_url", "airbyte", "ui"), _Mapping("airbyte_api_url", "airbyte", "api"), _Mapping("chroma_host", "chromadb", "host"), _Mapping("datahub_gms_url", "datahub", "gms"), _Mapping("flink_rest_url", "flink", "rest"), _Mapping("trino_uri", "trino", "uri"), _Mapping("trino_http_url", "trino", "http"), _Mapping("otel_endpoint", "otel-collector", "otlp_grpc"), _Mapping("cluster_mgmt_url", "alphaswarm-cp", "manage"), _Mapping("alphaswarm_api_url_internal", "alphaswarm-core", "internal_api"), _Mapping("alphaswarm_admin_url", "alphaswarm-admin", "url"), # --- Controller-as-auth-provider refactor (Phase 1) --- # ``controller_url`` is the cluster-internal base URL of # ``alphaswarm_controller``; the ``controller_m2m`` SecretStore # composes ``{controller_url}/auth/m2m/token`` from this. The # legacy ``cluster_mgmt_url`` (already mapped above) keeps pointing # at the ``/manage/*`` slice for backwards compatibility. _Mapping("controller_url", "alphaswarm-cp", "url"), # --- Additive infra services. Added once their entries land in # topology.yaml; absent today, fallback is a no-op for these. _Mapping("redpanda_bootstrap", "redpanda", "bootstrap"), _Mapping("redpanda_admin_url", "redpanda", "admin"), _Mapping("redpanda_schema_registry_url", "redpanda", "schema_registry"), _Mapping("redpanda_connect_url", "redpanda-connect", "ui"), _Mapping("questdb_pg_url", "questdb", "pgwire"), _Mapping("questdb_ilp_url", "questdb", "ilp_tcp"), _Mapping("questdb_http_url", "questdb", "http"), _Mapping("phoenix_endpoint", "phoenix", "otlp_http"), _Mapping("phoenix_grpc_endpoint", "phoenix", "otlp_grpc"), _Mapping("phoenix_ui_url", "phoenix", "ui"), _Mapping("prometheus_url", "prometheus", "query"), _Mapping("prometheus_remote_write_url", "prometheus", "remote_write"), _Mapping("grafana_url", "grafana", "ui"), _Mapping("loki_url", "loki", "push"), _Mapping("tempo_otlp_url", "tempo", "otlp_grpc"), _Mapping("hudi_warehouse_url", "hudi", "warehouse"), _Mapping("hudi_metastore_url", "hudi", "metastore"), # --- OpenLineage / Marquez relay (Workstream B). --- # When unset, defaults to the cluster-local Marquez endpoint # declared in topology.yaml. Setting ALPHASWARM_LINEAGE_OPENLINEAGE_MARQUEZ_URL # explicitly overrides. _Mapping("lineage_openlineage_marquez_url", "marquez", "http"), # --- MLOps service (Hard Rule 47). --- # ``alphaswarm-ml-mcp`` runs as a sidecar in the GPU pod; its URL is # declared once in topology.yaml so the FastAPI client + the # frontend resolver share a single source of truth. _Mapping("mcp_ml_url", "alphaswarm-ml-mcp", "http"), # --- alphaswarm_kb boundary (Hard Rules 56-60). --- # OpenFGA + OPA + the federation gateway are operated as cluster # services; their endpoints land in topology.yaml so the KB # adapters resolve them via the same fallback chain everything # else uses. _Mapping("openfga_url", "openfga", "http"), _Mapping("opa_url", "opa", "http"), _Mapping("kb_federation_gateway_url", "alphaswarm-kb-federation", "http"), ) def apply_topology_fallback( settings: BaseSettings, *, topology_path: str | None = None, ) -> dict[str, str]: """Mutate ``settings`` in place with topology fallback values. Returns a dict of ``{field_name: applied_url}`` for the fields that were updated. Empty dict means no changes. Safe to call repeatedly; the second call is a no-op (it skips any field that is no longer at its default). Phase 2 of the launcher refactor wires in an optional HTTP path: when ``settings.conn_mgr_through_controller`` is True the fallback fetches the topology snapshot from ``GET {controller_url}/manage/topology`` instead of reading the YAML file. The HTTP path falls through to the file path on any failure so chicken-and-egg startup sequences (monolith booting before the controller) keep working. """ if _conn_mgr_flag_enabled(settings): try: applied = _apply_via_controller(settings) except Exception: # noqa: BLE001 logger.warning( "Topology fallback via controller failed; falling back to file path", exc_info=True, ) applied = {} if applied: return applied try: from alphaswarm_core.topology import ( TopologyLoadError, load_topology, ) except Exception: # noqa: BLE001 logger.debug("alphaswarm_core.topology unavailable; skipping fallback") return {} try: topology = load_topology(topology_path) except TopologyLoadError as exc: logger.warning( "Topology fallback skipped: %s (path=%s)", exc, exc.path ) return {} except Exception: # noqa: BLE001 logger.warning("Topology fallback skipped (unexpected error)", exc_info=True) return {} services = topology.service_map applied: dict[str, str] = {} fields_set = settings.model_fields_set for mapping in URL_FALLBACK_FIELDS: if not hasattr(settings, mapping.settings_field): continue # Env wins: if the operator set ``ALPHASWARM_``, leave it alone. if mapping.settings_field in fields_set: continue service = services.get(mapping.service_id) if service is None: continue url = service.endpoint(mapping.endpoint_name) if not url: continue try: object.__setattr__(settings, mapping.settings_field, url) except Exception: # noqa: BLE001 logger.debug( "topology fallback could not assign %s", mapping.settings_field, exc_info=True, ) continue applied[mapping.settings_field] = url if applied: logger.info( "Topology fallback applied to %d Settings fields", len(applied) ) return applied def _conn_mgr_flag_enabled(settings: BaseSettings) -> bool: """Read the Phase 2 flag from ``settings`` defensively. Returns False when the field is missing (legacy Settings instances) or when the value coerces to falsy. Never raises. """ try: return bool(getattr(settings, "conn_mgr_through_controller", False)) except Exception: # noqa: BLE001 return False def _apply_via_controller(settings: BaseSettings) -> dict[str, str]: """Resolve URL fields by calling ``GET /manage/topology`` on the CP. Used only when ``settings.conn_mgr_through_controller`` is True. The controller is itself a topology consumer (it reads the same YAML file at boot), so this path is purely a centralisation affordance: every alphaswarm_* repo asks the same source for "where is service X" and audit + caching live in one place. Returns ``{}`` on any failure so the caller can fall through to the file-based path. Never raises. """ import httpx # local import — keeps import cycles + cold-boot deps small controller_url = _resolve_controller_url(settings) if not controller_url: logger.debug( "conn_mgr_through_controller=true but no controller_url configured" ) return {} try: with httpx.Client(timeout=5.0) as client: response = client.get( f"{controller_url.rstrip('/')}/manage/topology", params={"include_targets": False}, ) except httpx.HTTPError as exc: logger.warning("topology fetch from %s failed: %s", controller_url, exc) return {} if response.status_code >= 400: logger.warning( "topology fetch from %s returned HTTP %s", controller_url, response.status_code, ) return {} try: envelope = response.json() or {} except Exception: # noqa: BLE001 logger.warning("topology fetch returned non-JSON body") return {} body = envelope.get("data") if isinstance(envelope, dict) else None if not isinstance(body, dict): logger.debug("topology fetch envelope missing 'data'") return {} services_list = body.get("services") or [] services_map: dict[str, dict[str, Any]] = {} for entry in services_list: if not isinstance(entry, dict): continue sid = entry.get("id") if isinstance(sid, str): services_map[sid] = entry applied: dict[str, str] = {} fields_set = settings.model_fields_set for mapping in URL_FALLBACK_FIELDS: if not hasattr(settings, mapping.settings_field): continue if mapping.settings_field in fields_set: continue service_dict = services_map.get(mapping.service_id) if not service_dict: continue endpoints = service_dict.get("endpoints") or {} if not isinstance(endpoints, dict): continue url = endpoints.get(mapping.endpoint_name) if not isinstance(url, str) or not url: continue try: object.__setattr__(settings, mapping.settings_field, url) except Exception: # noqa: BLE001 logger.debug( "controller-mediated topology fallback could not assign %s", mapping.settings_field, exc_info=True, ) continue applied[mapping.settings_field] = url if applied: logger.info( "Topology fallback (controller-mediated) applied to %d Settings fields", len(applied), ) return applied def _resolve_controller_url(settings: BaseSettings) -> str: """Pick the controller base URL with sensible fallback chain. Priority: ``settings.controller_url`` -> ``settings.cluster_mgmt_url`` (legacy slot already mapped to ``alphaswarm-cp.manage`` — trim ``/manage`` so we get the bare root). """ explicit = str(getattr(settings, "controller_url", "") or "").strip() if explicit: return explicit.rstrip("/") legacy = str(getattr(settings, "cluster_mgmt_url", "") or "").strip() if legacy.endswith("/manage"): return legacy[: -len("/manage")] return legacy.rstrip("/") def topology_fallback_mappings() -> Iterable[_Mapping]: """Read-only iteration of the mapping table. Test helper.""" return URL_FALLBACK_FIELDS __all__ = [ "apply_topology_fallback", "topology_fallback_mappings", ]