"""STRATEGY_ID and external-token presence helpers. Phase1 (product plan): missing STRATEGY_ID → process must fail to start (exit non-zero). Values are never logged or returned in responses. """ from __future__ import annotations import os import sys from typing import Optional # Community alias order (level C); we only check presence, never echo values. _TOKEN_ENV_KEYS = ( "EXTERNAL_SERVICE_TOKEN", "XC_TOKEN_FILE", "XC_TOKEN", "MODELHUB_XC_TOKEN", ) def get_strategy_id() -> Optional[str]: """Return STRATEGY_ID if set and non-empty after strip; else None.""" raw = os.environ.get("STRATEGY_ID") if raw is None: return None value = raw.strip() return value if value else None def require_strategy_id() -> str: """Return STRATEGY_ID or exit the process with a non-zero status. Does not print the value. Safe for startup path. """ sid = get_strategy_id() if sid is None: print( "ERROR: STRATEGY_ID is required and must be non-empty. " "Refusing to start (Phase1 contract).", file=sys.stderr, ) sys.exit(1) return sid def strategy_id_present() -> bool: return get_strategy_id() is not None def _token_from_file(path: str) -> bool: """True if path exists and has non-empty content (value never read into logs).""" try: if not path or not os.path.isfile(path): return False with open(path, "r", encoding="utf-8") as fh: content = fh.read().strip() return bool(content) except OSError: return False def external_service_token_present() -> bool: """Whether any known token env / file alias is present (boolean only).""" from .task_add_client import resolve_token_with_source try: resolve_token_with_source() return True except (ValueError, OSError, UnicodeError): return False