"""Per-card feature flags and the capabilities record clients negotiate with. Flags come from the ``HUX_FLAGS`` comma list. A card counts as enabled only when it and every card it depends on are enabled, so a half-configured deployment fails closed. Route ownership per card is declared here so the capabilities record can tell a client exactly what it may call. """ from __future__ import annotations import os from collections.abc import Mapping from hux.errors import FlagOff from hux.identity import Identity from hux.rules import flag_enabled, flag_registry CONTRACT_VERSION = "1.0.0" CARD_ROUTES: dict[str, list[str]] = { "HUX-11": ["/hux/v1/capabilities", "/hux/v1/manifest"], "HUX-01": ["/hux/v1/conversations/{id}/events", "/hux/v1/conversations/{id}/events/stream"], "HUX-02": ["/hux/v1/memory", "/hux/v1/memory/{id}", "/hux/v1/memory/{id}/{action}", "/hux/v1/memory/export"], "HUX-03": ["/hux/v1/projects", "/hux/v1/projects/{id}", "/hux/v1/conversations", "/hux/v1/conversations/{id}", "/hux/v1/conversations/{id}/branch", "/hux/v1/conversations/{id}/lineage", "/hux/v1/search"], "HUX-04": ["/hux/v1/artifacts", "/hux/v1/artifacts/{id}", "/hux/v1/artifacts/{id}/versions", "/hux/v1/artifacts/{id}/versions/{n}", "/hux/v1/artifacts/{id}/versions/{n}/diff", "/hux/v1/artifacts/{id}/promote"], "HUX-05": ["/hux/v1/policy", "/hux/v1/approvals", "/hux/v1/approvals/{id}", "/hux/v1/runs/{id}/stop", "/hux/v1/runs/{id}/budget", "/hux/v1/runs/{id}/gate"], "HUX-06": ["/hux/v1/modes", "/hux/v1/conversations/{id}/mode"], "HUX-07": [], "HUX-08": ["/hux/v1/sources", "/hux/v1/sources/{id}", "/hux/v1/passages", "/hux/v1/messages/{id}/citations", "/hux/v1/notebooks", "/hux/v1/notebooks/{id}"], "HUX-09": ["/hux/v1/suggestions", "/hux/v1/suggestions/{id}/{action}"], "HUX-10": ["/hux/v1/privacy/policy", "/hux/v1/privacy/notices", "/hux/v1/conversations/{id}/forget", "/hux/v1/privacy/audit"], "HUX-12": ["/hux/v1/releases"], } class Flags: """Snapshot of which cards are on for this process.""" def __init__(self, environ: Mapping[str, str] | None = None) -> None: self._environ = dict(os.environ if environ is None else environ) self._registry = flag_registry() def enabled(self, card: str) -> bool: """True when the card and its whole dependency chain are on.""" entry = self._registry.get(card) return bool(entry) and flag_enabled(entry["flag"], self._environ) def require(self, card: str) -> None: """Raise FlagOff unless the card is enabled.""" if not self.enabled(card): raise FlagOff(f"{card} is not enabled") def capabilities(self, identity: Identity, build: Mapping[str, str] | None = None) -> dict: """Serialise ``hux.capabilities.v1`` for one caller.""" cards = [ {"card": card, "flag": entry["flag"], "enabled": self.enabled(card), "routes": CARD_ROUTES.get(card, [])} for card, entry in sorted(self._registry.items()) ] server = {k: v for k, v in (build or {}).items() if k in {"commit", "image_digest"} and v} return { "schema": "hux.capabilities.v1", "contract_version": CONTRACT_VERSION, "identity": identity.record(), "cards": cards, "server": server, } def build_from_environ(environ: Mapping[str, str] | None = None) -> dict[str, str]: """Commit and image digest the pod was started with, when the operator set them.""" environ = os.environ if environ is None else environ return {"commit": environ.get("HUX_BUILD_COMMIT", ""), "image_digest": environ.get("HUX_IMAGE_DIGEST", "")}