bootstrapper/start.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""
Atlas - Start Script
Python implementation of start.sh with full feature parity.
Cross-platform startup script for Atlas — the self-hosted engineering platform.
"""
import importlib
import re
import sys
import os
import json
import shlex
import subprocess
from datetime import date
from pathlib import Path
import click
from typing import Dict, Optional
from services.migrations.migration_v1 import (
apply as _apply_v1,
needs_migration as _needs_v1,
stamp_version as _stamp_v1,
)
from services.migrations.migration_v2 import (
URL_VAR_TO_PORT_VAR as _V2_URL_TO_PORT,
apply as _apply_v2,
needs_migration as _needs_v2,
stamp_version as _stamp_v2,
)
from services.migrations.migration_v3 import (
apply as _apply_v3,
needs_migration as _needs_v3,
stamp_version as _stamp_v3,
)
from services.migrations.migration_v4 import (
apply as _apply_v4,
needs_migration as _needs_v4,
stamp_version as _stamp_v4,
)
from services.migrations.migration_v5 import (
MigrationV5Error,
apply as _apply_v5,
needs_migration as _needs_v5,
stamp_version as _stamp_v5,
)
from services.migrations.build_pin_refresh import refresh as _refresh_build_pins
def _format_today() -> str:
"""Return today's date as ``YYYY-MM-DD`` for env-backfill markers.
Factored to a tiny helper so it's trivial to monkey-patch in tests
without freezing the system clock globally.
"""
return date.today().isoformat()
def _write_private_text(path: Path, text: str) -> None:
"""Atomically replace *path* with an owner-readable text file."""
atomic_write_text(path, text, mode=0o600)
def _read_env_backfill_text(path: Path) -> str:
with path.open(encoding="utf-8", newline="") as handle:
return handle.read()
def _backfill_env_assignment(line: str, origin: str) -> tuple[str, str, str, bool] | None:
"""Keep the decoded value and its validated encoded RHS separate."""
assignment = decode_env_assignment(line, origin=origin)
if assignment is None:
return None
key, value = assignment
encoded = line.partition("=")[2].rstrip("\r\n")
exported = re.match(r"^export[ \t]+", line.strip()) is not None
return key, value, encoded, exported
def _backfill_default_values(text: str, origin: str) -> dict[str, tuple[str, str]]:
values = {}
for index, line in enumerate(env_lines(text.removeprefix("\ufeff")), 1):
item = _backfill_env_assignment(line, f"{origin}:{index}")
if item is not None:
key, value, encoded, _exported = item
values[key] = value, encoded
return values
def _fill_env_backfill_blanks(text: str, replacements: dict[str, str], origin: str) -> str:
out = []
for index, original in enumerate(env_lines(text, keepends=True)):
bom = "\ufeff" if index == 0 and original.startswith("\ufeff") else ""
line = original.removeprefix(bom) if bom else original
item = _backfill_env_assignment(line, f"{origin}:{index + 1}")
if item is not None:
key, value, _encoded, exported = item
if key in replacements and value == "" and not exported:
ending = line[len(line.rstrip("\r\n")):]
out.append(f"{bom}{key}={replacements[key]}{ending}")
continue
out.append(original)
return "".join(out)
_ENV_MIGRATION_VERSION = 5
_MIGRATION_MARKER = "BOOTSTRAPPER_PORT_LAYOUT_VERSION"
def _is_migration_marker_assignment(line: str) -> bool:
"""Use the reader's key normalization without decoding unrelated values."""
key, separator, _ = line.strip().partition("=")
key = re.sub(r"^export[ \t]+", "", key).strip()
return bool(separator) and key == _MIGRATION_MARKER
def _duplicate_migration_markers(lines: list[str], env_path: Path) -> list[tuple[int, str]]:
"""Decode duplicate sentinel assignments with the canonical env grammar."""
indices = [index for index, line in enumerate(lines) if _is_migration_marker_assignment(line)]
if len(indices) < 2:
return []
return [
(index, decode_env_assignment(lines[index], origin=f"{env_path}:{index + 1}")[1])
for index in indices
]
def _consolidated_migration_marker(lines: list[str], indices: set[int], version: int) -> str:
"""Retain original marker text as comments and append one active marker."""
text = "".join("# " + line if index in indices else line for index, line in enumerate(lines))
separator = "" if not text or text.endswith(("\r", "\n")) else "\n"
return f"{text}{separator}{_MIGRATION_MARKER}={version}\n"
def _prepare_migration_marker(env_path: Path) -> bool:
"""Reconcile duplicate last-wins metadata before the frozen migration chain.
Return True only when an effective current/future duplicate marker makes
the entire chain a no-op. A legacy repair keeps the effective old version.
"""
if not env_path.is_file():
return False
try:
original = env_path.read_bytes().decode("utf-8")
bom = "\ufeff" if original.startswith("\ufeff") else ""
lines = env_lines(original.removeprefix("\ufeff"), keepends=True)
markers = _duplicate_migration_markers(lines, env_path)
if not markers:
return False
effective = markers[-1][1].strip() or "0"
if not re.fullmatch(r"[0-9]+", effective):
raise ValueError("invalid duplicate migration version")
version = int(effective)
if version >= _ENV_MIGRATION_VERSION:
return True
create_private_backup(env_path, version="migration-marker")
text = bom + _consolidated_migration_marker(lines, {index for index, _ in markers}, version)
atomic_write_text(env_path, text, mode=0o600)
except (OSError, ValueError):
raise click.ClickException(
f"{env_path}: {_MIGRATION_MARKER}: could not safely resolve duplicate migration metadata; "
"check the marker assignments and file permissions."
) from None
return False
def _other_profile_binds(active: str, bundles: dict) -> set[str]:
return {
candidate.host_bind_ip
for name, candidate in bundles.items()
if name != active and candidate.host_bind_ip
}
def _profile_owned_bind(
current_bind: str,
switching: bool,
other_binds: set[str],
) -> bool:
return switching and current_bind in other_binds
def _profile_host_bind_overrides(
active: str,
bundle,
bundles: dict,
env_vars: dict[str, str],
) -> tuple[dict[str, str], str, bool]:
"""Resolve profile-owned host binding without clobbering operator input."""
prior_applied = (env_vars.get("ATLAS_PROFILE_APPLIED", "") or "").strip()
switching = prior_applied not in ("", active)
other_binds = _other_profile_binds(active, bundles)
declared_bind = bundle.host_bind_ip
current_bind = env_vars.get("HOST_BIND_IP", "")
if declared_bind:
if not current_bind:
return {"HOST_BIND_IP": declared_bind}, prior_applied, switching
if _profile_owned_bind(current_bind, switching, other_binds):
if current_bind != declared_bind:
return {"HOST_BIND_IP": declared_bind}, prior_applied, switching
elif _profile_owned_bind(current_bind, switching, other_binds):
print(
f"profile={active}: cleared HOST_BIND_IP (ports return to all-interfaces)"
)
return {"HOST_BIND_IP": ""}, prior_applied, switching
return {}, prior_applied, switching
def _ollama_is_engine(env: Dict[str, str]) -> bool:
return (env.get("LLM_PROVIDER_SOURCE") or "").strip().lower().startswith("ollama-")
_DEFAULT_MODEL_KEYS = (
"LITELLM_DEFAULT_MODEL", "LITELLM_VISION_MODEL",
"LITELLM_EMBEDDING_MODEL", "LANGMEM_EMBEDDING_MODEL",
)
def _cloud_provider_of(model: str):
"""The cloud provider a model id belongs to (a ``<key>/`` prefix or a
catalog name), or None for anything else, e.g. a consumer model."""
from utils import llm_catalog # noqa: PLC0415
from utils.cloud_providers import CLOUD_PROVIDERS # noqa: PLC0415
for provider in CLOUD_PROVIDERS:
names = {entry.name for entry in llm_catalog.cloud_entries(provider.key)}
if model.startswith(f"{provider.key}/") or model in names:
return provider
return None
def _unrouted_ollama_reason(env: Dict[str, str]) -> Optional[str]:
"""Why an unregistered ``ollama/*`` default will not be routed, or None
when host auto-import may register tags this run cannot see."""
if not _ollama_is_engine(env):
return "Ollama is not enabled"
# The same test litellm-init uses before importing the host's tags.
host_imports = (env.get("LLM_PROVIDER_SOURCE") or "").strip().lower().startswith(
"ollama-localhost"
) and (env.get("OLLAMA_AUTO_IMPORT_LOCAL_MODELS") or "").strip().lower() in (
"true", "1", "yes", "on", "enabled")
return None if host_imports else "not among the Ollama models this launch registers"
def _unrouted_reason(env: Dict[str, str], model: str, routed: set) -> Optional[str]:
"""Why ``model`` will not be routed this launch, or None when it will be,
or when that cannot be known here (a host auto-import or a consumer model):
a default that may still be routed is never overwritten."""
if not model or model in routed:
return None
if model.startswith("ollama/"):
return _unrouted_ollama_reason(env)
provider = _cloud_provider_of(model)
if provider is None:
return None
if (env.get(provider.source_var) or "").strip() != "enabled":
return f"{provider.name} is not enabled"
if not (env.get(provider.api_key_var) or "").strip():
# A missing key is a misconfiguration to fix, not a reason to move the
# embedding pair (which would re-embed memory) for good.
return None
return f"not among the {provider.name} models this launch registers"
def _routed_model_names(env: Dict[str, str]) -> set:
"""Model names litellm-init registers this launch: every active model
(an Ollama one under both names) and the managed vLLM Metal model, often
an `openai/...` Hugging Face id."""
from utils.model_resolver import active_models # noqa: PLC0415
active = active_models(env)
routed = {entry.name for entry in active}
routed |= {f"ollama/{entry.name}" for entry in active if entry.provider == "ollama"}
if (env.get("VLLM_METAL_SOURCE") or "").strip().lower() == "managed-localhost":
routed.add((env.get("VLLM_METAL_MODEL") or "").strip())
return routed
def _preserve_keyless_cloud_embedding(env: Dict[str, str], routed: set) -> bool:
"""Keep an unrouted cloud memory contract until credentials are restored.
Source normalization persists disabled providers, so this must also hold
on later launches. Routed aliases can belong to a different upstream.
"""
for key in ("LITELLM_EMBEDDING_MODEL", "LANGMEM_EMBEDDING_MODEL"):
model = (env.get(key) or "").strip()
if not model or model in routed:
continue
provider = _cloud_provider_of(model)
if provider is not None and not (env.get(provider.api_key_var) or "").strip():
return True
return False
def _stale_default_keys(env: Dict[str, str]) -> Dict[str, str]:
"""Default-model keys naming a model this launch will not route, with why (#1359)."""
routed = _routed_model_names(env)
reasons = {key: _unrouted_reason(env, (env.get(key) or "").strip(), routed) for key in _DEFAULT_MODEL_KEYS}
if _preserve_keyless_cloud_embedding(env, routed):
for key in ("LITELLM_EMBEDDING_MODEL", "LANGMEM_EMBEDDING_MODEL"):
reasons.pop(key, None)
return {key: reason for key, reason in reasons.items() if reason}
def _embedding_replacement(env: Dict[str, str], stale: set) -> str:
"""The embedding model the stale ``ollama/*`` embedding pair moves to.
A deliberate non-Ollama LangMem override wins, so the pair stays aligned
(``apply_user_model_selections`` refuses a misaligned pair); otherwise
the best active embedding model, or the already non-Ollama LiteLLM one.
"""
from utils.model_resolver import best # noqa: PLC0415
langmem = (env.get("LANGMEM_EMBEDDING_MODEL") or "").strip()
if langmem and "LANGMEM_EMBEDDING_MODEL" not in stale:
return langmem
if "LITELLM_EMBEDDING_MODEL" not in stale:
return (env.get("LITELLM_EMBEDDING_MODEL") or "").strip()
return best("embeddings", env) or ""
def _launch_model_env(env: Dict[str, str]) -> Dict[str, str]:
"""Project this launch's cloud sources into resolver activation flags."""
from utils.cloud_providers import CLOUD_PROVIDERS # noqa: PLC0415
# The derived flags are written later, during service configuration.
# First launches and provider toggles must use this run's source values.
return {
**env,
**{
p.enabled_flag_var: "true" if env.get(p.source_var) == "enabled" else "false"
for p in CLOUD_PROVIDERS
},
}
def _default_model_replacements(env: Dict[str, str]) -> tuple[Dict[str, str], Dict[str, str]]:
"""(replacements, reason per stale key) for default models this launch
will not route (see ``AtlasStarter.reconcile_default_models``). Blank
resolutions are dropped: the end-of-run backfill would only restore the
template's value in the same launch."""
from utils.model_resolver import resolved_defaults # noqa: PLC0415
# The LITELLM_<PROVIDER>_ENABLED flags the resolver reads are written
# later (generate_service_configuration); derive them from this run's
# CLOUD_*_SOURCE values so a first launch, or a provider just toggled,
# is judged by what this launch will route.
env = _launch_model_env(env)
reasons = _stale_default_keys(env)
stale = set(reasons)
updates = {
key: value for key, value in resolved_defaults(env).items()
if key in stale and value
}
if not stale & {"LITELLM_EMBEDDING_MODEL", "LANGMEM_EMBEDDING_MODEL"}:
return updates, reasons
embedding = _embedding_replacement(env, stale)
if not embedding:
print(
"WARNING: the default embedding model will not be routed this "
"launch, and no active provider offers an embedding model; memory "
"and RAG embedding calls will fail until one is selected."
)
return updates, reasons
# Both keys are always passed so apply_user_model_selections recomputes
# LANGMEM_EMBEDDING_DIM for the pair.
updates["LITELLM_EMBEDDING_MODEL"] = embedding
if (env.get("LANGMEM_EMBEDDING_MODEL") or "").strip():
updates["LANGMEM_EMBEDDING_MODEL"] = embedding
return updates, reasons
def _run_privileged_hosts_setup(non_interactive: bool = False) -> bool:
"""Run only the hosts-file mutation in a sudo child process.
The shell wrapper intentionally refuses to run the whole startup flow as
root because that creates root-owned repo artifacts. For `--setup-hosts`,
ask for elevation only around the one operation that needs it.
``non_interactive`` uses ``sudo -n`` with captured output, for callers
that cannot hand the terminal to a password prompt (the Textual wizard):
it succeeds only when sudo needs no password.
"""
from utils.system import is_elevated
if is_elevated():
return HostsManager().setup_hosts_entries()
if os.name == "nt":
print(" • ❌ Please run from an Administrator shell to modify hosts file")
return False
bootstrapper_dir = Path(__file__).resolve().parent
repo_root = bootstrapper_dir.parent
env = os.environ.copy()
existing_pythonpath = env.get("PYTHONPATH")
env["PYTHONPATH"] = (
str(bootstrapper_dir)
if not existing_pythonpath
else f"{bootstrapper_dir}{os.pathsep}{existing_pythonpath}"
)
env["PYTHONDONTWRITEBYTECODE"] = "1"
helper = (
"from utils.hosts_manager import HostsManager; "
"raise SystemExit(0 if HostsManager().setup_hosts_entries() else 1)"
)
if not non_interactive:
print(" • --setup-hosts needs to edit your hosts file; requesting sudo for that write only.")
try:
result = subprocess.run(
[
"sudo",
*(["-n"] if non_interactive else []),
"env",
f"PYTHONPATH={env['PYTHONPATH']}",
"PYTHONDONTWRITEBYTECODE=1",
sys.executable,
"-c",
helper,
],
cwd=repo_root,
env=env,
check=False,
capture_output=non_interactive,
)
except OSError as exc: # no sudo binary: hosts setup is never fatal
print(f" • Could not launch sudo for hosts setup: {exc}")
return False
return result.returncode == 0
# Add the current directory to the path so we can import our modules
sys.path.insert(0, str(Path(__file__).parent))
from utils.atomic_write import (
atomic_replace_text, atomic_write_text, create_private_backup, decode_env_assignment,
env_lines, render_env_assignment,
set_env_assignment,
)
from utils.banner import BannerDisplay
from utils.hosts_manager import HostsManager
from utils.key_generator import KeyGenerator
from core.env_snapshot import EnvSnapshot
from core.linear_startup import LinearStartupOptions, json_cli_guard, run_linear_startup
from core.launch_outcome import (
DETACH,
ProbeSkipped,
build_skip_rows,
cancel_notice,
launch_cancelled_notice,
started_without,
)
from utils.localhost_validator import LocalhostValidator
from core.config_parser import (
ConfigParser, DEFAULT_BASE_PORT, DEFAULT_PROJECT_NAME,
external_env_user_overlay_path, parse_env_overlay_file,
)
from core.docker_manager import DockerManager
from core.port_manager import PortManager
from core.process_runner import (
_one_shot_row_verdict, run_with_deadline, compose_container_identity,
validate_compose_status_rows,
)
from utils.system import service_replica_targets
from services.source_validator import SourceValidator
from services.service_config import ServiceConfig
from services.dependency_manager import DependencyManager
from utils.source_override_manager import SourceOverrideManager
from utils.source_availability import source_has_endpoint
#: Lists (mostly wizard multi-selects) where BLANK is a deliberate answer —
#: "I selected none" — not a missing value. `.env.example` ships a non-empty
#: default for each, so the blank-backfill in backfill_missing_env_vars would
#: otherwise re-seed them on the very run that cleared them: deselecting every
#: Ollama model writes `OLLAMA_USER_MODELS=`, backfill restored the default
#: trio, and `ollama-pull` then downloaded several GB the user had just
#: declined. Backfill exists to repair values that were never set, so it must
#: not overwrite an intentional empty.
_USER_OWNED_BLANKABLE: frozenset = frozenset({
"OLLAMA_USER_MODELS",
"OLLAMA_CUSTOM_MODELS",
"OPENAI_USER_MODELS",
"ANTHROPIC_USER_MODELS",
"OPENROUTER_USER_MODELS",
"COMFYUI_USER_MODELS",
# NOT WEAVIATE_ENABLE_MODULES: nothing in the wizard writes it, so it can
# only be blank by hand-edit or a legacy `.env` — precisely what backfill
# exists to repair. Exempting it meant the blank survived into
# `ServiceConfig`, where a present-but-empty key made `.get(key, default)`
# return `''` and collapse the module list to nothing, durably.
#
# N8N_INIT_NODES stays for symmetry with the other multi-select lists, but
# is belt-and-braces: `services/n8n/compose.yml` substitutes its defaults
# with `${N8N_INIT_NODES:-...}`, so a blank there is already harmless.
"N8N_INIT_NODES",
# Blanked on purpose by ServiceConfig when MULTI2VEC_CLIP_SOURCE is
# disabled; refilling it here contradicted that decision every launch.
# ServiceConfig restores the URL itself when CLIP is enabled again (its
# `... or 'http://multi2vec-clip:8080'` treats a blank as unset).
"CLIP_INFERENCE_API",
# Documented as "blank = no Atlas-created demo topics"; refilling it from
# .env.example re-created atlas_stream_events on every launch.
"REDPANDA_DEMO_TOPICS",
})
def _env_example_values(env_example_path) -> dict[str, str]:
"""Shipped ``.env.example`` scalar defaults (``KEY=value`` lines)."""
try:
text = env_example_path.read_text(encoding="utf-8")
except OSError:
return {}
values: dict[str, str] = {}
for raw_line in text.splitlines():
line = raw_line.strip()
if line and not line.startswith("#") and "=" in line:
key, _, val = line.partition("=")
values[key.strip()] = val.strip()
return values
# Consumer-manifest keys derived from plugins/sidecars (#451): the only ones
# preflight refreshes once a start has persisted the stack's own values.
_PREFLIGHT_DERIVED_KEYS = frozenset({
"BACKEND_PLUGINS_DIR", "COMFYUI_CUSTOM_MODELS_FILE",
"COMFYUI_CUSTOM_NODES_FILE", "OLLAMA_CUSTOM_MODELS", "ATLAS_DERIVED_KEYS",
})
def _with_stale_derived_keys_cleared(
overrides: dict, env_vars: dict, consumer_config=None, protected=frozenset()
) -> dict:
"""Blank the derived keys a previous start wrote (per ATLAS_DERIVED_KEYS)
whose plugin or sidecar is gone, e.g. a removed model_sidecars.ollama left
OLLAMA_CUSTOM_MODELS pinning models no manifest declares (#1368).
Only while a consumer manifest is configured: the manifest is passed on
each run, so a plain `doctor` or start without it must not strip a running
consumer stack. Keys in ``protected`` (just set by a .env.user overlay)
are never blanked."""
from core.consumer_manifest import DERIVED_ENV_KEYS, DERIVED_KEYS_MARKER
def keys(raw) -> set:
return {item.strip() for item in str(raw or "").split(",") if item.strip()}
if consumer_config is not None and not getattr(consumer_config, "consumers", None):
return overrides
previous = keys(env_vars.get(DERIVED_KEYS_MARKER)) & set(DERIVED_ENV_KEYS)
current = keys(overrides.get(DERIVED_KEYS_MARKER))
if not previous - current:
return overrides
cleared = dict(overrides)
for key in previous - current - set(overrides) - set(protected):
cleared[key] = ""
cleared.setdefault(DERIVED_KEYS_MARKER, "")
return cleared
def _should_record_profile(overrides, auto_vars, switching: bool, env_vars: dict) -> bool:
"""Record ATLAS_PROFILE_APPLIED when the profile changed anything, and
always on the first start (even a no-op default profile): the marker is
what tells doctor's preflight that this stack's values, including CLI
overrides, are persisted and must not be re-merged."""
return bool(overrides or auto_vars or switching) or not _known_applied_profile(env_vars)
def _preflight_declared(consumer_config, env_vars: dict) -> dict:
"""Manifest env a preflight may write. Once a start has persisted this
stack's values, including CLI overrides (--base-port, -p,
--<svc>-source) that must beat the manifest, re-merging the manifest
silently undid them (doctor then pointed .env, ports and PROJECT_NAME at
a different stack), so only the derived overlay paths (#451) and keys
.env does not hold yet (newly added to the manifest) remain."""
declared = dict(consumer_config.env_overrides or {})
if not _known_applied_profile(env_vars):
return declared
return {
key: value for key, value in declared.items()
if key in _PREFLIGHT_DERIVED_KEYS or key not in env_vars
}
def _validate_consumer_manifests_early(starter, from_cli: bool) -> None:
"""Fail on a bad --consumer / ATLAS_CONSUMER_MANIFEST manifest as a usage
error before anything mutates the checkout (not a traceback mid-setup)."""
if not from_cli and not os.environ.get("ATLAS_CONSUMER_MANIFEST", "").strip():
return
try:
starter.config_parser.load_consumer_config()
except Exception as exc: # ConsumerManifestError + I/O failures
source = "--consumer" if from_cli else "ATLAS_CONSUMER_MANIFEST"
raise click.UsageError(f"invalid {source} manifest: {exc}") from exc
def _published_port_label(port_var: str, env_vars: dict) -> str:
"""":<port>", or "-" for a reserved slot no compose file publishes
(pg-meta, Supabase Studio): advertising it read as reachable."""
from services.topology import unpublished_port_vars
if port_var in unpublished_port_vars():
return "-"
return f":{env_vars.get(port_var, '?')}"
def _ascii_int(raw: str):
"""ASCII digits as an int, else None (`"²".isdigit()` is true)."""
raw = (raw or "").strip()
return int(raw) if raw.isascii() and raw.isdigit() else None
def _doctor_active_profile(starter, env: dict, declared) -> str:
"""The profile the stack runs: this run's --profile, else the last
applied one, else the consumer `profile:`. Reading only `profile:`
reported `default` after `./start.sh --profile prod`."""
from services.profiles import canonical_profile
return canonical_profile(getattr(starter, "profile", None) or _known_applied_profile(env) or declared)
def _known_applied_profile(env_vars: dict) -> str:
"""``ATLAS_PROFILE_APPLIED`` canonicalized, or "" when blank/unknown."""
from services.profiles import canonical_profile, is_known_profile
applied = (env_vars.get("ATLAS_PROFILE_APPLIED") or "").strip()
return canonical_profile(applied) if applied and is_known_profile(applied) else ""
def _manifest_source_vars(consumer_config) -> set[str]:
"""Non-empty ``*_SOURCE`` keys a consumer manifest declares (env.values
or env.file)."""
declared = getattr(consumer_config, "env_overrides", None) or {}
return {var for var, value in declared.items() if var.endswith("_SOURCE") and value}
def _env_user_source_vars(starter) -> set[str]:
"""``*_SOURCE`` keys this run's ``.env.user`` / ATLAS_ENV_USER_FILE set."""
return {key for key in getattr(starter, "_env_user_keys", None) or () if key.endswith("_SOURCE")}
def _declared_sources(values: dict) -> dict:
"""The ``*_SOURCE`` entries of a consumer manifest's env, which the auto
port probe must see before they reach .env (#1391)."""
return {key: value for key, value in values.items() if key.endswith("_SOURCE") and value}
#: The ConsumerRecord field each plugin/sidecar-derived env key comes from.
_DERIVED_KEY_FIELDS = (
("BACKEND_PLUGINS_DIR", "backend_plugins"),
("COMFYUI_CUSTOM_MODELS_FILE", "comfyui_sidecars"),
("COMFYUI_CUSTOM_NODES_FILE", "comfyui_custom_node_files"),
("OLLAMA_CUSTOM_MODELS", "ollama_models"),
)
def _derived_env_origins(consumer_config) -> Dict[str, str]:
"""The manifest that derives each plugin/sidecar key (#1057 origins).
``env_origins`` covers only ``env.values`` and ``env.file`` keys, so a
derived key looked like a ``.env`` value. The last manifest that declares
the field wins."""
return {
key: str(record.manifest_path)
for record in consumer_config.consumers
for key, attribute in _DERIVED_KEY_FIELDS
if getattr(record, attribute)
}
def _cli_source_overrides(params: dict) -> dict:
"""``--<svc>-source`` values given this run, keyed by env var."""
return {name.upper(): value for name, value in params.items()
if name.endswith("_source") and value is not None}
def _operator_env_keys(starter, consumer_config) -> set[str]:
"""Keys an operator pinned this run: ``.env.user`` overlay + manifest env."""
keys = set(getattr(starter, "_env_user_keys", None) or ())
return keys | set(getattr(consumer_config, "env_overrides", None) or {})
def _replaceable_env(maps: list[dict], protected: set[str]) -> list[dict]:
"""Drop operator-pinned keys so a profile never treats them as defaults."""
return [{k: v for k, v in m.items() if k not in protected} for m in maps]
def _prior_profile_env(bundles: dict, prior_applied: str, switching: bool) -> dict:
"""The env the prior profile wrote, when this run switches profiles."""
prior = bundles.get(prior_applied) if switching else None
return prior.env if prior is not None else {}
def _profile_env_overrides(
active: str, declared: dict, env_vars: dict, replaceable: list[dict],
) -> dict[str, str]:
"""A profile's ``env`` values for .env, keeping operator-set ones.
A value is operator-set only when it differs from every ``replaceable``
map (the shipped ``.env.example`` defaults, the prior profile's env): every
profile-managed knob ships a non-empty default, so "unset or empty" alone
never let a profile (or a consumer ``profile_overrides`` env) apply.
"""
overrides: dict[str, str] = {}
for env_key, declared_value in declared.items():
current = env_vars.get(env_key)
if current == declared_value:
continue
if not current or any(current == m.get(env_key) for m in replaceable):
overrides[env_key] = declared_value
if current:
print(
f"profile={active}: set {env_key}={declared_value!r} "
f"(was the shipped or prior-profile default {current!r})"
)
else:
print(
f"profile={active}: keeping operator-set {env_key}={current!r} "
f"(profile default is {declared_value!r})"
)
return overrides
def _catalog_embedding_dim_repair(env: dict) -> dict[str, str]:
"""LANGMEM_EMBEDDING_DIM when a catalog embedding model's declared dim
differs from .env; the contract rejects that pair, so it only arises from
a corrected catalog entry. Custom (uncatalogued) models are left alone."""
from utils.model_resolver import ( # noqa: PLC0415
dim_for_model_id, embedding_dimension_contract,
)
active_env = _launch_model_env(env)
if _preserve_keyless_cloud_embedding(active_env, _routed_model_names(active_env)):
return {}
model = (env.get("LANGMEM_EMBEDDING_MODEL") or env.get("LITELLM_EMBEDDING_MODEL") or "").strip()
dim = dim_for_model_id(model, env=active_env)
if dim is None or str(dim) == (env.get("LANGMEM_EMBEDDING_DIM") or "").strip():
return {}
return {"LANGMEM_EMBEDDING_DIM": str(embedding_dimension_contract(model, None, env=active_env))}
def _detect_env_image_drift(
existing_env: dict, env_example_path,
) -> list[tuple[str, str, str]]:
"""Return [(key, user_value, example_value), ...] for every ``*_IMAGE``
key whose value in the user's ``.env`` differs from ``.env.example``.
Why this matters: CI tests `docker compose ... --env-file .env.example
config -q`, so divergence in the user's `.env` is invisible to CI but
breaks `docker build` at user-side launch. Example incident: PR #35
migrated SPARK_IMAGE bitnami/spark:4.1.2 → apache/spark:4.1.2 (Bitnami
went paywalled), but a user's pre-migration `.env` retained the stale
Bitnami pin → `docker.io/bitnami/spark:4.1.2: not found` at the
spark-history image-pull step. See PR #35 docs.
Scope: ONLY ``*_IMAGE`` keys (image pins control what gets pulled /
built and are the only class with this CI-blind failure mode). Other
env divergence (ports, secrets, source toggles) is often intentional
and would produce noisy false-positives.
Empty user values are skipped — placeholder lines in `.env` and
auto-managed keys correctly defer to compose `:-` fallbacks.
Kept as a module-level free function so it can be unit-tested without
spinning up the full Starter.
"""
if not env_example_path.exists():
return []
example_pins: dict[str, str] = {}
for raw_line in env_example_path.read_text(encoding="utf-8").splitlines():
line = raw_line.strip()
if not line or line.startswith('#'):
continue
if '_IMAGE=' not in line:
continue
key, _, val = line.partition('=')
if key.endswith('_IMAGE') and val:
example_pins[key] = val.split('#', 1)[0].strip()
drift: list[tuple[str, str, str]] = []
for key, example_val in example_pins.items():
user_val = (existing_env.get(key, '') or '').strip()
if not user_val:
continue
if user_val != example_val:
drift.append((key, user_val, example_val))
return drift
def _detect_port_collisions(rows) -> list[str]:
"""Return human-readable warning strings, one per colliding host port.
A *collision* is two or more rows whose port value (the ":<num>"
suffix or just the bare number) is equal AND nonempty. Disabled
rows (port = "-" / "" / None) don't participate.
`rows` is an iterable of ``(name, port_val)`` tuples — the same
shape the pre-launch summary builder already accumulates as it
iterates services. Kept as a module-level free function so it can
be unit-tested without instantiating ``AtlasStarter``.
The warnings are purely informational — launch still proceeds.
Compose-up would otherwise fail with an opaque "address already in
use" error from Docker, so this gives the user a chance to ack and
continue or step back and pick another port.
"""
by_port: dict[str, list[str]] = {}
for name, port_val in rows:
port = (port_val or "").lstrip(":").strip()
if not port or port == "-":
continue
# Only digits count as a host port. Skip anything that doesn't
# look like a numeric port (e.g. an external URL).
if not port.isdigit():
continue
by_port.setdefault(port, []).append(name or "<unknown>")
warnings: list[str] = []
for port, names in by_port.items():
if len(names) >= 2:
warnings.append(
f"⚠ port {port} used by {' + '.join(names)} — "
f"compose-up may fail to bind."
)
return warnings
class AtlasStarter:
"""Main class for starting Atlas."""
def __init__(self):
# Set root directory first
self.root_dir = Path(__file__).resolve().parent.parent
# Initialize all managers with correct root_dir
self.banner = BannerDisplay()
self.hosts_manager = HostsManager()
self.key_generator = KeyGenerator(str(self.root_dir))
self.config_parser = ConfigParser(str(self.root_dir))
self.localhost_validator = LocalhostValidator(self.config_parser)
self.docker_manager = DockerManager(str(self.root_dir))
self.port_manager = PortManager(str(self.root_dir))
self.source_validator = SourceValidator(self.config_parser)
self.service_config = ServiceConfig(self.config_parser)
self.dependency_manager = DependencyManager(self.config_parser)
self.source_override_manager = SourceOverrideManager(self.config_parser)
self.active_track = None
self.active_track_overrides = frozenset()
# Managers recorded here were started by this AtlasStarter invocation.
# Pre-existing native hosts are deliberately never rollback-owned.
self._managed_hosts_started_this_run: list[tuple[str, object]] = []
# True once a nonzero `up --wait` was reclassified as converged only
# after re-polling still-`starting` rows within the grace window
# (#677/#681) — surfaced in the --detach --json summary so automation
# can see the health race happened.
self._up_converged_after_grace: bool = False
# Services the image build left out of `up` because their image failed
# and the launch could go without them (#989). Both front ends read it
# to name them; one-shot verification skips them.
self.skipped_builds: list[str] = []
# --support-bundle PATH (#1057): where a failed start writes its
# redacted bundle. Both the linear flow and the Textual launch screen
# read it from here.
self.support_bundle_path: Optional[Path] = None
# Set when the port check stopped this project's running stack, so a
# later decline/failure can say it is down (the stop is not undone).
self.stopped_previous_instance: bool = False
# True when this run chose BASE_PORT with `auto` (CLI flag or a fresh
# manifest resolution) rather than reusing a block.
self.base_port_auto_chosen: bool = False
# True once this run's cold cleanup has taken the project down; its
# ports may still be held for a moment afterwards (#1438).
self.project_stopped_this_run: bool = False
def show_banner(self):
"""Display the startup banner (hero + wordmark/credits)."""
self.banner.show_hero(no_splash=getattr(self, "no_splash", False))
self.banner.show_banner()
def ensure_dependencies_available(self) -> bool:
"""Ensure all required dependencies are available."""
self.banner.show_section_header("Checking Dependencies", "🔍")
# Check Docker availability
if not self.docker_manager.check_docker_available():
self.banner.show_status_message(
"Docker is not available. Please install Docker and ensure it's running.",
"error"
)
return False
# Show detected Docker compose command
compose_cmd = self.docker_manager.get_compose_command_display()
self.banner.show_status_message(f"Using Docker Compose command: {compose_cmd}", "info")
compose_ok, compose_message = self.docker_manager.check_compose_version()
self.banner.show_status_message(
compose_message,
"success" if compose_ok else "error",
)
if not compose_ok:
return False
# Check docker-compose.yml exists
compose_file = self.root_dir / "docker-compose.yml"
if not compose_file.exists():
self.banner.show_status_message(
f"Docker Compose file not found: {compose_file}",
"error"
)
return False
self.banner.show_status_message(f"Docker Compose file found: {compose_file}", "success")
# Python YAML parsing replaces yq dependency
self.banner.show_status_message("Using native Python YAML parsing (replaces yq dependency)", "info")
return True
def apply_source_overrides(self, **kwargs) -> bool:
"""
Apply SOURCE overrides from command-line arguments.
Args:
**kwargs: Command-line SOURCE override arguments
Returns:
bool: True if successful
"""
overrides = self.source_override_manager.collect_overrides(**kwargs)
# Remember which SOURCE vars the operator set explicitly THIS run, so
# the profile applier can honor operator-wins for any profile-managed
# source (generalizes the historical prometheus/grafana-only guards).
self._explicit_source_vars = set(overrides.keys())
if overrides:
self._warn_on_consumer_pin_overrides(overrides)
return self.source_override_manager.apply_overrides(overrides)
return True
def _warn_on_consumer_pin_overrides(self, overrides: Dict[str, str]) -> None:
consumer = self.config_parser.load_consumer_config()
for key, reason in consumer.env_reasons.items():
if key not in overrides or _consumer_pin_is_auto(key, consumer.env_overrides[key]):
continue
if overrides[key] != consumer.env_overrides[key]:
flag = "--" + key.lower().replace("_", "-")
self.banner.show_status_message(
f"{key} consumer pin overridden by {flag}: "
f"{reason} ({consumer.env_origins[key]}). The next start reapplies the pin.",
"warning",
)
def apply_profile_overrides(self, profile: str) -> bool:
"""Apply the active deployment profile's declarative bundle to .env (#755).
Bundles live in ``bootstrapper/profiles.yml`` (platform-defined
``default``/``prod``; ``dev`` aliases ``default``) with consumer
``profile_overrides:`` merged on top. Field semantics (preserving the
behaviors this method used to hard-code):
- ``host_bind_ip``: non-empty → fills a blank ``HOST_BIND_IP`` and
replaces a bind the prior profile asserted during a profile switch;
an operator's non-empty bind (e.g. ``0.0.0.0:``) is kept, as
``profiles.yml`` documents. Empty/undeclared → cleared only when
the current value equals another profile's non-empty bind.
- ``sources``: asserted on every start of this profile, EXCEPT when
that service's source was set by an explicit CLI flag this run
(tracked via ``_explicit_source_vars``, which
``apply_source_overrides`` fills before this runs) or is pinned in
the consumer manifest's ``env.values``, ``.env.user`` or
``ATLAS_ENV_USER_FILE`` (CLI flag > manifest/.env.user > profile).
``auto`` delegates to the durable ``<SVC>_SOURCE: auto`` resolver
(#753). On a profile SWITCH (tracked via ``ATLAS_PROFILE_APPLIED``),
sources the prior profile asserted are reset to their service
default first, so transitions leave no residue; a same-profile
restart never resets (a wizard/operator selection that happens to
equal a bundle value is safe).
- ``env``: applied when the var is unset/empty, the shipped
``.env.example`` default, or (on a switch) the prior profile's value
for the same key, unless ``.env.user`` or the consumer manifest pins
it; any other operator-set value is kept with a notice (LOG_MAX_*).
Called from both the linear (--no-tui) path and the TUI wizard
pipeline so profile configuration applies regardless of how Atlas is
started.
"""
from services.host_capabilities import probe_host_capabilities # noqa: F401
from services.manifests import load_manifests, option_in_profile
from services.profiles import (
ProfileConfigError,
canonical_profile,
load_profile_bundles,
merge_consumer_profile_overrides,
)
active = canonical_profile(profile)
try:
bundles = load_profile_bundles()
consumer_config = self.config_parser.load_consumer_config()
bundles = merge_consumer_profile_overrides(
bundles, getattr(consumer_config, "profile_overrides", None) or {}
)
except ProfileConfigError as exc:
print(f"profile={active}: invalid profile configuration: {exc}")
return False
bundle = bundles.get(active)
if bundle is None:
print(f"profile={active}: unknown profile")
return False
manifests = load_manifests(self.root_dir / "services")
by_name = {m.name: m for m in manifests if m.sources is not None}
env_vars = self.config_parser.parse_env_file()
overrides, prior_applied, switching = _profile_host_bind_overrides(
active,
bundle,
bundles,
env_vars,
)
explicit_vars = set(getattr(self, "_explicit_source_vars", set()) or set())
# Consumer manifest env.values beat the profile, as they beat the
# track (#783): CLI flag > manifest > profile.
explicit_vars.update(_manifest_source_vars(consumer_config))
# A *_SOURCE pinned in .env.user / ATLAS_ENV_USER_FILE wins too; the
# overlay only guarded the profile's env, not its sources (#1391).
explicit_vars.update(_env_user_source_vars(self))
# ── sources: undo the prior profile's asserts on a SWITCH ────
auto_vars: list[str] = []
if switching:
prior_bundle = bundles.get(prior_applied)
if prior_bundle is not None:
for svc, prior_id in prior_bundle.sources.items():
if svc not in by_name:
continue
var = by_name[svc].sources.var
if var in explicit_vars or svc in bundle.sources:
continue # operator wins / new profile re-declares it
if prior_id == "auto":
# A prior bundle-`auto` resolution: reset only when the
# resolved value is not offered under the NEW profile
# (dev-only residue that would fail validation).
current_value = (env_vars.get(var, "") or "").strip()
if current_value and not option_in_profile(
manifests, svc, current_value, active
):
overrides[var] = by_name[svc].sources.default
print(
f"profile={active}: reset {var} to "
f"'{by_name[svc].sources.default}' (prior profile's "
f"auto-resolved '{current_value}' is not offered here)"
)
continue
if (env_vars.get(var, "") or "").strip() == prior_id:
overrides[var] = by_name[svc].sources.default
print(
f"profile={active}: reset {var} to "
f"'{by_name[svc].sources.default}' (was asserted by "
f"profile '{prior_applied}')"
)
# ── sources: assert this profile's selections ────────────────
for svc, source_id in bundle.sources.items():
manifest = by_name.get(svc)
if manifest is None:
print(
f"profile={active}: sources.{svc} names a service without a "
f"sources block; ignoring"
)
continue
var = manifest.sources.var
if var in explicit_vars:
continue # explicit CLI flag this run wins
if source_id == "auto":
current = (env_vars.get(var, "") or "").strip()
if switching:
prior_bundle = bundles.get(prior_applied)
prior_id = (prior_bundle.sources.get(svc) if prior_bundle else None)
if prior_id and prior_id != "auto" and current == prior_id:
# Clear the prior profile's assert so `auto` re-resolves
# instead of durably keeping the residue.
self._merge_env_file_overrides({var: manifest.sources.default})
auto_vars.append(var)
continue
option_ids = {opt.id for opt in manifest.sources.options}
if source_id not in option_ids:
print(
f"profile={active}: sources.{svc}='{source_id}' is not one of "
f"{sorted(option_ids)}"
)
return False
if not option_in_profile(manifests, svc, source_id, active):
print(
f"profile={active}: sources.{svc}='{source_id}' is not offered "
f"under this profile"
)
return False
overrides[var] = source_id
# ── env: defaults unless operator-set ────────────────────────
overrides.update(_profile_env_overrides(
active, bundle.env, env_vars,
_replaceable_env(
[
_env_example_values(self.config_parser.env_example_path),
_prior_profile_env(bundles, prior_applied, switching),
],
_operator_env_keys(self, consumer_config),
),
))
# Write the marker when this run changes something or completes a
# switch, and on the first start; a later no-op run leaves .env
# byte-identical.
if _should_record_profile(overrides, auto_vars, switching, env_vars):
overrides["ATLAS_PROFILE_APPLIED"] = active
if overrides and not self.source_override_manager.update_env_file(overrides):
return False
# `auto` selections route through the durable #753 resolver (which
# reads the just-updated .env for its keep/resolve decision).
if auto_vars:
resolved = self._resolve_auto_source_overrides(
{var: "auto" for var in auto_vars}
)
concrete = {k: v for k, v in resolved.items() if v != "auto"}
if concrete:
self._merge_env_file_overrides(concrete)
return True
def apply_cloud_api_keys(self, keys: Dict[str, str]) -> bool:
"""
Persist cloud LLM provider API keys (OPENAI_API_KEY,
ANTHROPIC_API_KEY, OPENROUTER_API_KEY) into ``.env``.
Reuses the in-place .env writer the source-override manager
already employs, so format and comment lines are preserved.
Empty values are written verbatim — used to clear a key.
Args:
keys: Mapping of env-var name to value (e.g.
{'OPENAI_API_KEY': 'sk-...'}). Empty dict is a no-op.
Returns:
bool: True on success (or no-op).
"""
if not keys:
return True
return self.source_override_manager.update_env_file(keys)
def apply_user_model_selections(self, selections: Dict[str, str]) -> bool:
"""
Persist user-selected model lists (OPENAI_USER_MODELS,
ANTHROPIC_USER_MODELS, OPENROUTER_USER_MODELS, OLLAMA_USER_MODELS,
OLLAMA_CUSTOM_MODELS) into ``.env``.
Values are comma-separated model names. ``litellm-init``
consumes them on the next ``docker compose up`` via
``model_resolver`` (YAML catalogs + env) to build the active
model set.
Args:
selections: Mapping of env-var name to comma-separated
model names. Empty dict is a no-op.
Returns:
bool: True on success (or no-op).
"""
if not selections:
return True
# Persist one model/dimension contract. Curated models derive their
# dimension from catalog metadata; custom models must carry a selected
# dimension or preserve an explicit value already present in .env.
embed = (selections.get("LITELLM_EMBEDDING_MODEL", "") or "").strip()
if embed:
from utils.model_resolver import ( # noqa: PLC0415
dim_for_model_id,
embedding_dimension_contract,
)
selections = dict(selections)
parser = getattr(self, "config_parser", None)
current = parser.parse_env_file() if parser is not None else {}
active_env = _launch_model_env({**current, **selections})
current_override = (
current.get("LANGMEM_EMBEDDING_MODEL", "") or ""
).strip()
explicit_override = (
selections.get("LANGMEM_EMBEDDING_MODEL", "") or ""
).strip()
if explicit_override and explicit_override != embed:
raise ValueError(
"LITELLM_EMBEDDING_MODEL and LANGMEM_EMBEDDING_MODEL "
"must select the same embedding contract"
)
if current_override and not explicit_override and current_override != embed:
raise ValueError(
"LANGMEM_EMBEDDING_MODEL override must be explicitly aligned "
"with the selected LITELLM_EMBEDDING_MODEL"
)
effective_embed = explicit_override or embed
configured_dimension = selections.get("LANGMEM_EMBEDDING_DIM")
if (
configured_dimension is None
and dim_for_model_id(effective_embed, env=active_env) is None
):
current_model = (
current.get("LANGMEM_EMBEDDING_MODEL")
or current.get("LITELLM_EMBEDDING_MODEL")
or "ollama/nomic-embed-text"
)
# Preserve an existing explicit custom contract only when it
# belongs to the same effective model. A generic fresh 768
# must never silently declare a newly selected custom model.
if str(current_model).strip() == effective_embed:
configured_dimension = current.get("LANGMEM_EMBEDDING_DIM")
selections["LANGMEM_EMBEDDING_DIM"] = str(
embedding_dimension_contract(effective_embed, configured_dimension, env=active_env)
)
return self.source_override_manager.update_env_file(selections)
def reconcile_default_models(self) -> bool:
"""Repoint default models this launch will not route.
LiteLLM registers only active models, so a default naming a disabled
engine or cloud provider, or a model no longer selected, made every
default chat or embedding call fail with "invalid model" (#1359).
`.env.example`'s ``ollama/*`` defaults under a cloud-only launch were
the first case fixed. A default is replaced only when it is known not
to be routed; a routed model, a consumer model or a host-imported
Ollama tag is never overwritten.
"""
env = self.config_parser.parse_env_file()
replacements, reasons = _default_model_replacements(env)
changed = {
key: value for key, value in replacements.items() if value != env.get(key)
}
for key, value in sorted(changed.items()):
print(f"Default model {key}: {env.get(key)} -> {value} ({reasons.get(key, 'its pair moved')})")
for key in sorted(set(reasons) - set(replacements)):
print(f"WARNING: default model {key}={env.get(key)} is not routed ({reasons[key]}) "
"and no active provider offers a replacement")
# The embedding pair goes through the normal path, unchanged half
# included, so its dimension contract (LANGMEM_EMBEDDING_DIM)
# follows the new model.
if changed.keys() & {"LITELLM_EMBEDDING_MODEL", "LANGMEM_EMBEDDING_MODEL"}:
for key in ("LITELLM_EMBEDDING_MODEL", "LANGMEM_EMBEDDING_MODEL"):
if key in replacements:
changed[key] = replacements[key]
if not changed:
return self._repair_catalog_embedding_dim(env)
return self._apply_default_model_changes(env, changed)
def _apply_default_model_changes(self, env: dict, changed: dict) -> bool:
"""Write the repointed defaults, then run the catalog dimension repair
when no embedding key moved: repointing only a chat or vision default
skipped it, so a corrected catalog dimension waited a launch."""
try:
if not self.apply_user_model_selections(changed):
return False
except ValueError as exc:
# An embedding contract that was already inconsistent (e.g. a
# custom model with no declared dimension) must not abort a launch
# this best-effort repair was never needed for.
print(f"WARNING: default models left unchanged: {exc}")
return True
if changed.keys() & {"LITELLM_EMBEDDING_MODEL", "LANGMEM_EMBEDDING_MODEL"}:
return True # the selection path already rewrote LANGMEM_EMBEDDING_DIM
return self._repair_catalog_embedding_dim({**env, **changed})
def _repair_catalog_embedding_dim(self, env: dict) -> bool:
"""Follow a corrected catalog dimension (qwen3-embedding:0.6b was
declared 1536 but emits 1024) without waiting for a wizard re-run."""
repair = _catalog_embedding_dim_repair(env)
if not repair:
return True
print(
f"Embedding dimension LANGMEM_EMBEDDING_DIM: {env.get('LANGMEM_EMBEDDING_DIM')} "
f"-> {repair['LANGMEM_EMBEDDING_DIM']} (catalog value for the embedding model)"
)
return self.source_override_manager.update_env_file(repair)
def validate_source_configurations(self) -> bool:
"""Validate all SOURCE configurations and scale values against YAML.
Calls the validator's repair pass first (auto-disables cloud
providers with missing keys, etc.), then runs the read-only
validation. Splitting the two lets pure-tooling callers
(linters, CI dry-runs) run validation alone without mutating
.env — see SourceValidator.enforce_runtime_invariants.
"""
# Repair pass — mutates .env when necessary. A failed write
# (disk full, permissions, etc.) is not silently recoverable:
# halt before the read-only validate pass, which would
# otherwise see stale .env.
if not self.source_validator.enforce_runtime_invariants():
self.source_validator.print_validation_results()
return False
# Read-only validation pass.
sources_valid = self.source_validator.validate_all_sources()
if not sources_valid:
self.source_validator.print_validation_results()
return False
scales_valid = self.source_validator.validate_scale_values()
if not scales_valid:
self.banner.console.print("[bright_red]❌ Scale validation failed:[/bright_red]")
for error in self.source_validator.get_validation_errors():
self.banner.console.print(f" {error}")
return False
return True
def _persist_project_name(self, project_name: Optional[str]) -> bool:
"""Persist ``PROJECT_NAME`` to .env so start AND stop (and every
``docker compose -p``) target the same container family.
Sticky by design: a later bare ``./stop.sh`` reads PROJECT_NAME back
from .env and tears down exactly what ``./start.sh`` launched. Consumer
stacks reusing Atlas as a submodule just set this once (via --project or
by editing .env) to isolate their namespace from a base Atlas stack.
"""
if not project_name:
return True
if self.config_parser.stored_project_name_matches(project_name):
self.docker_manager.project_name_override = project_name
return True
if self.source_override_manager.update_env_file({"PROJECT_NAME": project_name}):
self.docker_manager.project_name_override = project_name
self.banner.show_status_message(
f"Project name set to '{project_name}' (persisted to .env "
f"PROJECT_NAME; container family: {project_name}-*)",
"info",
)
return True
self.banner.show_status_message(
f"Failed to persist PROJECT_NAME={project_name!r}; aborting so "
"docker compose does not target a stale project.",
"error",
)
return False
def _env_user_overlay_path(self) -> Path:
"""Return the optional user-owned env overlay next to the active .env."""
return self.config_parser.env_file_path.parent / ".env.user"
def _apply_single_env_user_overlay(
self,
overlay_path: Path,
*,
label: str,
warn_if_missing: bool,
) -> Dict[str, str]:
if not warn_if_missing and not overlay_path.exists():
return {}
overrides = self.config_parser.read_env_user_overlay(
overlay_path, label=label,
warn=lambda message: self.banner.show_status_message(message, "warning"),
)
if overrides is None:
return {}
if not overrides:
self.banner.show_status_message(f" • {overlay_path} has no env overrides", "info")
return {}
self._merge_env_file_overrides(overrides)
self.banner.show_status_message(
f" • Applied {overlay_path} ({len(overrides)} override{'s' if len(overrides) != 1 else ''})",
"info",
)
return overrides
def _apply_env_user_overlay(self, project_context=None) -> Dict[str, str]:
"""Merge optional user-owned env overlays into the active .env file.
Precedence is deterministic: the sibling .env.user is applied first,
then ATLAS_ENV_USER_FILE, then any consumer manifest env values.
"""
applied_overrides: Dict[str, str] = {}
context = dict(project_context or {})
if not context.get("previous_project"):
context["previous_project"] = (self.config_parser.parse_env_file().get("PROJECT_NAME") or "").strip().lower()
sibling_overrides = self._apply_single_env_user_overlay(
self._env_user_overlay_path(),
label=".env.user",
warn_if_missing=False,
)
applied_overrides.update(sibling_overrides)
external_overlay_path = external_env_user_overlay_path()
if external_overlay_path is not None:
external_overrides = self._apply_single_env_user_overlay(
external_overlay_path,
label="ATLAS_ENV_USER_FILE",
warn_if_missing=True,
)
applied_overrides.update(external_overrides)
context["overlay_project"] = applied_overrides.get("PROJECT_NAME")
applied_overrides.update(self._apply_consumer_manifest_env(set(applied_overrides), context))
return applied_overrides
def _apply_consumer_manifest_env(self, overlay_keys: set, project_context=None) -> Dict[str, str]:
"""Merge the consumer manifest's env into .env, blanking derived keys
whose source is gone (never the overlay's ``overlay_keys``), and return
the values an operator pinned this way."""
consumer_config = self.config_parser.load_consumer_config()
declared = dict(consumer_config.env_overrides or {})
manifest_overrides = _with_stale_derived_keys_cleared(
declared, self.config_parser.parse_env_file(),
consumer_config, protected=overlay_keys,
)
if not manifest_overrides:
return {}
resolved_overrides = self._resolve_auto_source_overrides(
self._resolve_auto_base_port_override(manifest_overrides, project_context)
)
self._merge_env_file_overrides(resolved_overrides)
# The bookkeeping marker is not a consumer setting.
count = len([key for key in resolved_overrides if key != "ATLAS_DERIVED_KEYS"])
names = ", ".join(consumer.name for consumer in consumer_config.consumers)
self.banner.show_status_message(
f" • Applied consumer manifest env ({count} override{'s' if count != 1 else ''})"
f" for {names}",
"info",
)
# Blanked stale keys are Atlas housekeeping, not values an operator
# pinned, so later profile handling must not treat them as pinned.
return {
key: value for key, value in resolved_overrides.items()
if key in declared or key not in manifest_overrides
}
def build_support_bundle(self, options, checks=None):
"""Assemble the redacted support bundle (#1057); writes nothing.
``checks`` reuses an earlier ``(results, notes)`` from
``support_bundle.run_checks``; otherwise the doctor checks run here,
offline and inside the bundle's wall-time budget.
"""
from core import support_bundle as sb
results, notes = checks or sb.run_checks(DOCTOR_CHECKS, self, options.limits)
notes = list(notes)
logs = list(options.logs)
try:
env = self.config_parser.parse_env_file()
except Exception as exc: # noqa: BLE001 - recorded in the bundle
# Fail closed: without .env the known secret values cannot be
# scrubbed, so no log text goes in.
env, logs = {}, []
notes.append(f"could not read .env ({exc}); log excerpts left out")
return sb.build_bundle(sb.BundleRequest(
env=env,
env_origins=self._env_origins(),
env_file=Path(self.config_parser.env_file_path),
checks=results,
logs=logs,
context=dict(options.context),
include_unlisted=options.include_unlisted,
limits=options.limits,
notes=notes,
))
def _env_origins(self) -> Dict[str, str]:
"""The file that sets each overridden env key, in the order a start
applies them: .env.user, ATLAS_ENV_USER_FILE, then consumer
manifests (or their env files). The last one wins."""
origins: Dict[str, str] = {}
for overlay in (self._env_user_overlay_path(), external_env_user_overlay_path()):
if overlay is None or not overlay.is_file():
continue
try:
keys = parse_env_overlay_file(overlay)
except Exception: # noqa: BLE001 - an unreadable overlay sets nothing
continue
origins.update(dict.fromkeys(keys, str(overlay)))
try:
consumer = self.config_parser.load_consumer_config()
origins.update(consumer.env_origins)
origins.update(_derived_env_origins(consumer))
except Exception: # noqa: BLE001 - a broken manifest has its own check
pass
return origins
def materialize_consumer_env_for_preflight(self) -> Dict[str, str]:
"""Persist the consumer manifest's derived ``env_overrides`` into ``.env``
before a standalone ``doctor`` / ``compose validate``.
The full ``start`` flow materializes these (via ``_apply_env_user_overlay``),
but the preflight subcommands validate the assembled compose *without*
applying them — so a consumer overlay that interpolates ``${BACKEND_PLUGINS_DIR}``
(etc.) fails on a fresh checkout that has never started (#451). This applies
the same derived, deterministic values a real start would write, so preflight
validates the same assembled config a launch produces.
Quiet by construction (no banner) so ``doctor --format json`` stays
machine-clean. No-op when there is no ``.env`` yet or no overrides.
Returns the overrides applied (empty dict when none).
"""
if not self.config_parser.env_file_path.exists():
return {}
try:
consumer_config = self.config_parser.load_consumer_config()
except Exception: # noqa: BLE001 — best-effort: a malformed manifest
# must surface through the doctor's own consumer-manifest check
# (which reports it as a structured `fail`), not crash preflight
# before any check has run.
return {}
if getattr(self, "profile", None) in (None, ""):
# Standalone doctor/endpoints runs never pass --profile; resolve
# against the consumer manifest's declared default so `auto`
# cannot poison a prod deployment's .env with dev-only sources.
# A running stack's applied profile (a launch-time --profile)
# outranks the manifest default.
self.profile = (
_known_applied_profile(self.config_parser.parse_env_file())
or getattr(consumer_config, "profile", None) or "default"
)
env_vars = self.config_parser.parse_env_file()
declared = _with_stale_derived_keys_cleared(
_preflight_declared(consumer_config, env_vars), env_vars, consumer_config
)
overrides = self._resolve_auto_source_overrides(
self._resolve_auto_base_port_override(declared),
quiet=True,
)
if overrides:
self._merge_env_file_overrides(overrides)
# Writing BASE_PORT alone left every *_PORT on the old block, so
# `endpoints export` after `doctor` passed its drift check while
# emitting ports from the wrong stack. Recompute them, as a start does.
base_port = _parsed_base_port(str(overrides.get("BASE_PORT", "")))
# Validate first: update_env_ports prints to stdout on a bad value,
# which would corrupt `doctor --format json`; the doctor base-port
# check fails on it instead.
if base_port is not None and self.port_manager.validate_base_port(base_port):
# Same env file as the merge above (ATLAS_ENV_FILE / test roots).
self.port_manager.config_parser = self.config_parser
self.port_manager.update_env_ports(base_port, create_backup=False)
return overrides
def _merge_env_file_overrides(self, overrides: Dict[str, str]) -> None:
"""Replace or append env keys without duplicate-key ambiguity."""
env_file_path = self.config_parser.env_file_path
content = env_file_path.read_text(encoding="utf-8")
updated_content = content
for var_name, raw_value in overrides.items():
# Belt-and-braces against `.env` line injection. `_set_scalar`
# rejects these at the parse boundary, but several paths build
# overrides without going through it — the env-user overlay,
# storage exports, auto-source and auto-BASE_PORT resolution, and
# the two profile-apply sites — so the writer must be safe whatever
# produced the entry.
#
# Both halves matter, because the emitted line is `KEY=VALUE`: a
# newline in the VALUE appends further assignments, and a newline
# or `=` in the KEY does exactly the same. `parse_env_file`
# resolves last-wins, so an injected line beats the real one — and
# the `^KEY=.*$` rewrite below is MULTILINE, not DOTALL, so a later
# run rewrites only the first line and leaves the remainder in
# place permanently.
var_value = render_env_assignment(var_name, raw_value)
# `export KEY=` is KEY to Compose and parse_env_file: rewrite it
# in place too (keeping `export`) or it would win over this value.
updated_content = set_env_assignment(updated_content, var_name, var_value)
atomic_write_text(env_file_path, updated_content, mode=0o600)
def _resolve_auto_base_port_override(self, overrides: Dict[str, str], project_context=None) -> Dict[str, str]:
"""Resolve a manifest ``BASE_PORT: auto`` to a concrete, **durable** free
block (returns a new dict; non-auto overrides are returned unchanged).
Manifest ``auto`` means "give this consumer a stable free BASE_PORT
block". It resolves to the first wholly-free ``BASE_PORT+0..N`` block —
which skips the default 63000 AND any block whose ports are in use by
another running stack, so several consumers started in turn each land on
a **distinct** block. The resolved value is persisted to ``.env`` and
**kept** on later starts (a previously-resolved non-default block is
reused as-is, so a warm restart never moves and never mistakes its own
running containers for another stack's). A cold start (regenerated
``.env`` → default) re-resolves, still skipping occupied blocks.
Unlike the one-off ``--base-port auto`` CLI flag (which resolves fresh
every time it is passed), the manifest form is committed and must be
stable — that is the whole point of pinning identity in the manifest.
"""
if str(overrides.get("BASE_PORT", "")).strip().lower() != "auto":
return overrides
self._durable_auto_manifest = True
if (project_context or {}).get("explicit_base") is not None:
return {**overrides, "BASE_PORT": str(project_context["explicit_base"])}
current = (self.config_parser.parse_env_file().get("BASE_PORT", "") or "").strip()
in_use = False
try:
current_int: Optional[int] = int(current)
except ValueError:
current_int = None
if (
current_int is not None
and current_int != DEFAULT_BASE_PORT
and self.port_manager.validate_base_port(current_int)
):
# Durable keep — but only when the block is actually OURS to keep.
# Several consumers resolving `auto` at different times (while the
# others are down) can each persist the SAME "first free" block;
# the first to start binds it and every later stack must re-resolve
# or refuse to boot (#727 composed-run finding). Distinguish:
# block free → keep (durable, incl. cold)
# occupied + our containers up → keep (warm restart)
# occupied + no containers of ours → FOREIGN stack: re-resolve
in_use = self.port_manager.check_port_range_availability(
current_int, _declared_sources(overrides)
)
if not in_use or self._can_keep_auto_block(overrides, project_context):
resolved = current_int # durable: keep this consumer's block
else:
self.banner.show_status_message(
f"BASE_PORT=auto: persisted block {current_int} is bound by "
f"another stack on this host — re-resolving to a free block "
f"(the prior block was not ours to keep).",
"warning",
)
resolved = self.port_manager.auto_base_port(source_overrides=_declared_sources(overrides))
self.base_port_auto_chosen = resolved is not None
if resolved is None:
self.banner.show_status_message(
"BASE_PORT=auto could not find a free port block; keeping "
f"{current_int} (expect bind conflicts).",
"warning",
)
resolved = current_int
else:
resolved = self.port_manager.auto_base_port(source_overrides=_declared_sources(overrides))
self.base_port_auto_chosen = resolved is not None
if resolved is None:
self.banner.show_status_message(
"BASE_PORT=auto could not find a free port block; using the "
f"default {DEFAULT_BASE_PORT}. Set an explicit BASE_PORT.",
"warning",
)
resolved = DEFAULT_BASE_PORT
project = self._auto_port_project(overrides, project_context)
previous = (project_context or {}).get("previous_project")
self._durable_auto_project = previous if resolved == current_int and in_use and previous else project
merged = dict(overrides)
merged["BASE_PORT"] = str(resolved)
return merged
def _auto_port_project(self, overrides, context=None) -> str:
"""Resolve identity without writing it or introducing a consumer pin."""
from core.config_parser import normalize_project_name
context = context or {}
project = (context.get("requested_project") or overrides.get("PROJECT_NAME")
or context.get("overlay_project") or context.get("fallback_project")
or self.config_parser.parse_env_file().get("PROJECT_NAME")
or DEFAULT_PROJECT_NAME)
return normalize_project_name(project)
def _can_keep_auto_block(self, overrides, context=None) -> bool:
if context is None:
return self._project_has_running_containers(overrides)
project = self._auto_port_project(overrides, context)
running = self._project_running_state(project)
if running is None:
self.base_port_auto_chosen = False
previous = context.get("previous_project")
return running is None or bool(running and (not previous or previous == project))
def resolve_wizard_base_port(self, sources, options) -> None:
"""Finalize only port intent; never replay consumer sources, keys or models."""
intent = options.get("base_port_intent")
if intent == "numeric":
self.base_port_auto_chosen = False
elif options.get("base_port_auto"):
self.base_port_auto_chosen = True
elif getattr(self, "_durable_auto_manifest", False):
context = {"requested_project": options.get("project_name"),
"previous_project": getattr(self, "_durable_auto_project", None)}
declared = {key.upper(): value for key, value in sources.items()}
resolved = self._resolve_auto_base_port_override({**declared, "BASE_PORT": "auto"}, context)
options["base_port"] = int(resolved["BASE_PORT"])
def _project_has_running_containers(self, overrides: Dict[str, str]) -> bool:
"""True when this consumer's compose project has containers up — the
warm-restart signal that its persisted BASE_PORT block is genuinely
its own. Conservative on any docker error: returns True (keep the
block; never surprise-move ports when ownership can't be verified)."""
state = self._project_running_state(self._auto_port_project(overrides))
return state is not False
def _project_running_state(self, project: str) -> Optional[bool]:
"""None means Docker could not verify ownership; retain conservatively."""
try:
probe = subprocess.run(
[
"docker", "ps", "-q",
"--filter", f"label=com.docker.compose.project={project}",
],
capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=10, check=False,
)
if probe.returncode != 0:
return None # docker unreachable → conservative keep
return bool(probe.stdout.strip())
except (OSError, subprocess.SubprocessError):
return None # conservative keep
def _resolve_auto_source_overrides(
self, overrides: Dict[str, str], *, quiet: bool = False
) -> Dict[str, str]:
"""Resolve manifest ``<SVC>_SOURCE: auto`` sentinels to concrete,
**durable**, host-correct option ids (#753) — the source-selection
analog of ``_resolve_auto_base_port_override`` above, sharing its
durability contract:
1. **Durable keep.** A concrete, valid, **non-default** value already
in ``.env`` (a prior ``auto`` resolution or an operator's explicit
``--<svc>-source`` override) is kept — provided the active profile
offers it (a dev-only value under ``--profile prod`` re-resolves,
or prod validation would fail every start). KNOWN LIMIT: an
operator override *to the service default id* is indistinguishable
from a cold regen and is re-resolved on the next start — to durably
force the default, commit the concrete id instead of ``auto``.
2. **Platform-adaptive resolve.** Otherwise pick the first
``sources.auto_prefer`` entry whose host capability holds
(services/host_capabilities.py) and whose option is offered under
the active deployment profile (``option_in_profile``). Entries are
declarative, ordered, and end in an unconditional terminal
fallback (lint-enforced), so resolution never dead-ends.
3. **Persist & keep.** The concrete id is merged into ``.env``; the
next start hits step 1. A cold regen (``.env`` back to defaults)
re-resolves — still host-correct.
Runs before source validation, so the validator only ever sees
concrete ids. ``*_SOURCE`` vars set to ``auto`` that match no manifest
sources block are left untouched — the validator then rejects the literal
``auto`` with its normal valid-options error, keeping the mistake
loud rather than silently dropped. Other keys (``DOCLING_USE_OCR``)
may hold ``auto`` as a value and pass through.
"""
auto_keys = [
key
for key, value in overrides.items()
# Only a *_SOURCE key is a sentinel: DOCLING_USE_OCR=auto is a value.
if key.endswith("_SOURCE") and str(value).strip().lower() == "auto"
]
if not auto_keys:
return overrides
from services.host_capabilities import probe_host_capabilities
from services.manifests import load_manifests, option_in_profile
manifests = load_manifests(self.root_dir / "services")
by_source_var = {m.sources.var: m for m in manifests if m.sources is not None}
env = self.config_parser.parse_env_file()
profile = getattr(self, "profile", "default") or "default"
capabilities = probe_host_capabilities()
merged = dict(overrides)
for key in auto_keys:
manifest = by_source_var.get(key)
if manifest is None or manifest.sources is None:
if not quiet:
self.banner.show_status_message(
f"{key}=auto has no manifest sources block to resolve against; "
"leaving as-is (source validation will reject it).",
"warning",
)
continue
sources = manifest.sources
option_ids = {opt.id for opt in sources.options}
current = (env.get(key, "") or "").strip()
if (
current
and current.lower() != "auto"
and current != sources.default
and current in option_ids
):
if option_in_profile(manifests, manifest.name, current, profile):
merged[key] = current # durable: keep prior resolution/override
continue
# The kept value is not offered under the ACTIVE profile (e.g.
# a dev-only managed-localhost-mps under --profile prod): keeping
# it would fail prod source validation on every start until a
# hand-edit. Re-resolve instead — self-healing beats durability
# when the durable value is invalid for this run.
if not quiet:
self.banner.show_status_message(
f" • [auto] {key}: prior value '{current}' is not offered "
f"under profile '{profile}' — re-resolving.",
"info",
)
resolved: Optional[str] = None
matched = ""
for pref in sources.auto_prefer:
if pref.id not in option_ids:
continue # lint catches this; stay safe at runtime too
if not option_in_profile(manifests, manifest.name, pref.id, profile):
continue
if pref.requires_capability is not None and not capabilities.has(
pref.requires_capability
):
continue
resolved = pref.id
matched = pref.requires_capability or "terminal fallback"
break
if resolved is None:
reason = (
"no auto_prefer entries declared"
if not sources.auto_prefer
else "no auto_prefer entry eligible on this host/profile"
)
if not quiet:
self.banner.show_status_message(
f"{key}=auto could not resolve ({reason}); using the service "
f"default '{sources.default}'.",
"warning",
)
resolved = sources.default
matched = "service default"
merged[key] = resolved
if not quiet:
self.banner.show_status_message(
f" • [auto] {key}=auto → {resolved} (matched capability: {matched})",
"info",
)
return merged
def _remove_env_keys_by_prefix(self, prefix: str) -> None:
"""Drop every ``.env`` line whose KEY starts with ``prefix``. Used to
clear stale storage export vars before re-emitting the current set, so
a removed/renamed store leaves no dangling export."""
env_file_path = self.config_parser.env_file_path
if not env_file_path.exists():
return
content = env_file_path.read_text(encoding="utf-8")
kept = [
line
for line in env_lines(content, keepends=True)
if not line.lstrip().startswith(prefix)
]
updated = "".join(kept)
if updated == content:
return
atomic_write_text(env_file_path, updated, mode=0o600)
def validate_persisted_project_name(self) -> bool:
"""Fail before mutating .env when its stored PROJECT_NAME is invalid."""
try:
self.config_parser.get_project_name()
except ValueError as e:
self.banner.show_status_message(
f"Invalid PROJECT_NAME in {self.config_parser.env_file_path}: {e}",
"error",
)
return False
return True
def _strip_env_byte_order_mark(self) -> None:
"""Drop a UTF-8 byte-order mark from .env before any writer runs: the
reader ignores it, but the writers' line patterns did not match the
first key behind it (#1391)."""
path = getattr(self.config_parser, "env_file_path", None)
try:
data = path.read_bytes() if path is not None else b""
except OSError:
return
if data.startswith(b"\xef\xbb\xbf"):
from utils.atomic_write import atomic_write_text
atomic_write_text(path, data[3:].decode("utf-8"), mode=path.stat().st_mode & 0o777)
def prepare_environment(
self,
cold_start: bool,
base_port: Optional[int] = None,
project_name: Optional[str] = None,
) -> bool:
"""Clean a cold project before replacing its credential-bearing env file."""
self._strip_env_byte_order_mark()
if cold_start:
# A fresh clone has no env file for Compose interpolation, and no
# existing credentials to preserve. Materialize it first. Existing
# installs must clean first so a failed `down --volumes` leaves the
# credential file matching any surviving volumes.
if not self.config_parser.env_file_exists():
if not self.setup_env_file(
cold_start=True,
base_port=base_port,
project_name=project_name,
):
return False
return self.perform_cold_start_cleanup(project_name=project_name)
# Backfill is additive and preserves every existing credential. It
# gives upgraded Compose fragments all required interpolation keys
# without replacing the env file before volume deletion succeeds.
if not self.backfill_missing_env_vars():
return False
if not self.perform_cold_start_cleanup(project_name=project_name):
return False
return self.setup_env_file(
cold_start=cold_start,
base_port=base_port,
project_name=project_name,
)
def _setup_port_context(self, base_port, project_name) -> dict:
"""Capture the saved block's identity before replacement or overlays."""
prior_env = self.config_parser.parse_env_file()
if base_port is not None and not getattr(self, "_cli_base_port_auto", False):
self.base_port_auto_chosen = False
return {"requested_project": project_name,
"previous_project": (prior_env.get("PROJECT_NAME") or "").strip().lower(),
"explicit_base": base_port}
def setup_env_file(self, cold_start: bool, base_port: Optional[int] = None,
project_name: Optional[str] = None) -> bool:
"""
Setup .env file from .env.example if needed.
Supports custom .env file paths via ATLAS_ENV_FILE environment variable
(and the deprecated GENAI_ENV_FILE alias).
Replicates the .env setup logic from the original start.sh.
Args:
cold_start: Whether this is a cold start
base_port: Optional custom base port
project_name: Optional normalized PROJECT_NAME to persist (from
--project / -p). Written AFTER the .env is created/preserved so
a cold-start copy of .env.example doesn't reset it to the default.
Returns:
bool: True if successful
"""
# Use config_parser paths which respect ATLAS_ENV_FILE
env_file_path = self.config_parser.env_file_path
env_example_path = self.config_parser.env_example_path
# Show which env file we're using if custom
if self.config_parser.is_using_custom_env_file():
self.banner.show_status_message(
f"Using custom env file: {env_file_path}",
"info"
)
if project_name is None and env_file_path.exists() and not cold_start:
if not self.validate_persisted_project_name():
return False
project_context = self._setup_port_context(base_port, project_name)
existing_project_name: Optional[str] = None
if cold_start and env_file_path.exists():
try:
existing_project_name = self.config_parser.get_project_name()
except ValueError:
existing_project_name = None
project_context["fallback_project"] = existing_project_name
# A cold start replaces .env with .env.example; keep a private copy
# first. It holds generated secrets plus operator-only values such
# as BACKUP_MANIFEST_HMAC_KEY, without which backups cannot be
# restored.
# Versioned: routine start-up backups share the plain slot and
# would rotate this copy out within a couple of starts; the cold
# slot keeps its own five most recent copies.
try:
saved_env = create_private_backup(env_file_path, version="cold")
except OSError as exc:
self.banner.show_status_message(
f"Could not back up {env_file_path} before the cold start: {exc}",
"error",
)
return False
self.banner.show_status_message(f"Saved the previous env file to {saved_env}", "info")
# Check if .env exists, if not or if cold start is requested, create from .env.example
if not env_file_path.exists() or cold_start:
if not env_example_path.exists():
self.banner.show_status_message(
f".env.example file not found: {env_example_path}",
"error"
)
return False
self.banner.show_section_header("Setting Up Environment", "📋")
if cold_start:
self.banner.show_status_message("Creating new .env file from .env.example (cold start)...", "info")
else:
self.banner.show_status_message("Creating new .env file from .env.example", "info")
try:
# Ensure parent directory exists (important for custom paths)
env_file_path.parent.mkdir(parents=True, exist_ok=True)
# Materialize atomically with owner-only permissions. A direct
# copy can leave a partial or briefly world-readable secret file.
_write_private_text(
env_file_path,
env_example_path.read_text(encoding="utf-8"),
)
self.banner.show_status_message(f" • Copied {env_example_path}", "info")
self.banner.show_status_message(f" • to {env_file_path}", "info")
overlay_overrides = self._apply_env_user_overlay(project_context)
self._env_user_keys = set(overlay_overrides)
# Unset potentially lingering port environment variables if cold start and custom base port are used
effective_base_port = base_port if base_port is not None else DEFAULT_BASE_PORT
if cold_start and effective_base_port != DEFAULT_BASE_PORT:
self.unset_port_environment_variables()
self.banner.show_status_message("Environment file setup completed", "success")
project_name_to_persist = project_name
if (
project_name_to_persist is None
and existing_project_name
and "PROJECT_NAME" not in overlay_overrides
):
project_name_to_persist = existing_project_name
if not self._persist_project_name(project_name_to_persist):
return False
return self.validate_persisted_project_name()
except Exception as e:
self.banner.show_status_message(f"Failed to create .env file: {e}", "error")
return False
os.chmod(env_file_path, 0o600)
overlay_overrides = self._apply_env_user_overlay(project_context)
self._env_user_keys = set(overlay_overrides)
if not self._persist_project_name(project_name): # .env already exists and not cold start
return False
if project_name is None and overlay_overrides:
return self.validate_persisted_project_name()
return True
def _refresh_retired_build_pins(self) -> None:
"""Move a build pin that still holds a retired shipped default to the
``.env.example`` value (``services/migrations/build_pin_refresh.py``).
An operator's own value stays; a failure warns and the start goes on."""
try:
changes, backup = _refresh_build_pins(
Path(self.config_parser.env_file_path),
Path(self.config_parser.env_example_path),
# A pin an overlay or consumer manifest set this run is theirs.
frozenset(getattr(self, "_env_user_keys", None) or ()),
)
except (OSError, UnicodeError) as exc:
self.banner.show_status_message(f"Could not refresh retired build pins: {exc}", "warning")
return
for key, old, new in changes:
self.banner.show_status_message(
f"Refreshed {key} from the retired default {old} to {new} (backup: {backup.name})",
"info",
)
def backfill_missing_env_vars(self) -> bool:
"""Append variables present in ``.env.example`` but missing from
the user's ``.env`` (preserving every existing nonblank value).
Catches the merge-from-upstream case: a worktree adds new
services to ``.env.example`` (e.g. MinIO's ``MINIO_IMAGE``,
``MINIO_PORT``, bucket names) but the user's pre-existing
``.env`` doesn't carry those keys. ``docker compose config``
then warns ``variable X not set, defaulting to blank`` and
fails with ``service has neither an image nor a build context``
when a critical key like ``${MINIO_IMAGE}`` is empty.
Preserves the source file's organisation: missing vars are
emitted under their original section heading (the
``# === FOO ===`` banner above them in ``.env.example``),
with the immediate context comment block kept intact. The
result reads as if the new entries had been there from the
start — no flat "everything dumped at the bottom" lump.
Idempotent — running again with no missing keys is a no-op.
Adds missing keys and fills eligible bare empty assignments;
other assignments retain their values and order (so user-edited
nonblank values and pre-existing comments stay put). Auto-managed
keys with empty defaults in the example (passwords, access keys)
are appended blank; the
cold-start secret-generation step fills them later. The one
exception is a build pin still on a retired shipped default
(``_refresh_retired_build_pins``).
"""
env_file_path = self.config_parser.env_file_path
env_example_path = self.config_parser.env_example_path
if not env_file_path.exists() or not env_example_path.exists():
return True # Nothing to backfill against.
# The one non-blank value backfill changes: a build pin still on a
# default an earlier Atlas shipped, which the new image rejects.
self._refresh_retired_build_pins()
try:
example_text = _read_env_backfill_text(env_example_path)
env_text = _read_env_backfill_text(env_file_path)
except UnicodeError:
self.banner.show_status_message(
f"Cannot decode {env_file_path} or {env_example_path} as UTF-8 for backfill.", "error",
)
return False
except OSError as e:
self.banner.show_status_message(
f"Could not read env files for backfill: {e}", "warning",
)
return True # Non-fatal — surface compose's own error later.
try:
existing_keys, blank_keys = _scan_env_keys(env_text, str(env_file_path))
example_values = _backfill_default_values(example_text, str(env_example_path))
except ValueError as exc:
self.banner.show_status_message(f"Cannot backfill env files: {exc}", "error")
return False
# Keys the migration chain (services/migrations) is about to
# write: backfill must NOT seed them from .env.example, or
# run_port_migration() — which runs AFTER backfill — finds them
# already present and skips the user's legacy values. Concretely:
# seeding the sentinel stamps a legacy .env as already-migrated
# (every migration silently skips); seeding a *_LOCALHOST_PORT
# while the legacy *_LOCALHOST_URL is still in the file makes v2
# discard the URL's custom port; seeding the COMFYUI model vars
# while COMFYUI_MODEL_SET exists pre-empts v3's translation.
migration_owned: set[str] = {"BOOTSTRAPPER_PORT_LAYOUT_VERSION"}
for _url_var, _port_var in _V2_URL_TO_PORT.items():
if _url_var in existing_keys:
migration_owned.add(_port_var)
if "COMFYUI_MODEL_SET" in existing_keys:
migration_owned.update(
{"COMFYUI_USER_MODELS", "COMFYUI_CUSTOM_MODELS_FILE"}
)
# Build a lookup of .env.example values so we can fill in BLANK
# entries (a key exists in .env but with no value) using a
# non-blank manifest default. This handles the case where the
# user's .env was created when a secret's manifest `default` was
# different from the current one — e.g., the supabase DB
# password placeholder got reintroduced to the example after a
# secret-emission policy change. Intentional autogen blanks
# (LITELLM_MASTER_KEY etc.) have `default: ""` in the manifest
# and therefore stay blank in .env.example, so this branch is a
# no-op for them.
blank_fills = {
key: example_values[key][1]
for key in blank_keys
if example_values.get(key, ("", ""))[0]
and key not in migration_owned
and key not in _USER_OWNED_BLANKABLE
}
if blank_fills:
filled_text = _fill_env_backfill_blanks(env_text, blank_fills, str(env_file_path))
try:
atomic_write_text(
env_file_path, filled_text, mode=0o600,
)
except OSError as e:
self.banner.show_status_message(
f"Failed to fill blank values in {env_file_path}: {e}", "error",
)
return False
env_text = _read_env_backfill_text(env_file_path)
self.banner.show_status_message(
f"Filled {len(blank_fills)} blank value(s) from .env.example: "
f"{', '.join(sorted(blank_fills)[:4])}"
f"{' …' if len(blank_fills) > 4 else ''}",
"info",
)
groups = self._parse_env_example_sections(
example_text, existing_keys | migration_owned,
)
if not groups:
return True
groups = self._rebase_backfilled_ports(groups)
# Insert each group AT THE END of its matching section in the
# user's .env. If the section doesn't exist in .env (older
# layout, e.g. a brand-new service family), the section banner
# plus its entries land in an "Auto-backfilled" trailer at the
# bottom. This preserves the source-of-truth grouping the
# docstring promised — historically the regex below was
# `[=]{3,}` which never matched .env.example's `─` (U+2500)
# bars, so EVERY key fell into "(unsectioned)" and got dumped
# at the bottom. The fix is to (a) match both bar chars and
# (b) actually splice in-place.
new_env_text, total, in_place_sections, trailer_sections = (
self._splice_backfill_in_place(env_text, groups)
)
try:
atomic_write_text(env_file_path, new_env_text, mode=0o600)
except OSError as e:
self.banner.show_status_message(
f"Failed to backfill {env_file_path}: {e}", "error",
)
return False
msg_bits = []
if in_place_sections:
msg_bits.append(
f"into {len(in_place_sections)} existing section"
f"{'s' if len(in_place_sections) != 1 else ''} "
f"({', '.join(in_place_sections[:3])}"
f"{' …' if len(in_place_sections) > 3 else ''})"
)
if trailer_sections:
msg_bits.append(
f"with {len(trailer_sections)} new section"
f"{'s' if len(trailer_sections) != 1 else ''} appended "
f"({', '.join(trailer_sections[:3])}"
f"{' …' if len(trailer_sections) > 3 else ''})"
)
self.banner.show_status_message(
f"Backfilled {total} missing env var(s) from .env.example, "
+ ("; ".join(msg_bits) or "no placement available"),
"info",
)
return True
@staticmethod
def _splice_backfill_in_place(
env_text: str,
groups: "list[tuple[str, list[tuple[list[str], str, str]]]]",
) -> "tuple[str, int, list[str], list[str]]":
"""Insert each backfill group at the END of its matching section
in ``env_text``. Sections that don't exist in ``env_text`` get
an auto-backfilled trailer at the bottom.
Returns ``(new_env_text, total_keys_added, in_place_section_names,
trailer_section_names)``.
Section identity matches the banner-title text emitted by
``env_assembler`` (e.g. ``"data: Apache Spark (standalone
cluster) (services/spark/service.yml)"``). We tolerate both
``─`` and ``=`` bar chars for back-compat with hand-edited
``.env`` files.
"""
bar_re = re.compile(r"^#\s*[=─]{3,}\s*$")
lines = env_lines(env_text, keepends=True)
n = len(lines)
# Walk env_text and record [(section_name, start_idx, end_idx)]
# — start_idx is the line AFTER the closing bar of the banner
# block; end_idx is exclusive of the next banner's opening bar.
sections: list[tuple[str, int, int]] = []
current_name = "(preamble)"
current_start = 0
i = 0
while i < n:
line = lines[i].removeprefix("\ufeff") if i == 0 else lines[i]
# Banner = bar / # TITLE / bar (3 lines).
if (
bar_re.match(line.rstrip("\r\n"))
and i + 2 < n
and lines[i + 1].lstrip().startswith("#")
and bar_re.match(lines[i + 2].rstrip("\r\n"))
):
title = lines[i + 1].lstrip("#").strip()
# Close out the prior section at the line that holds
# this banner's opening bar.
sections.append((current_name, current_start, i))
current_name = title or "(unnamed)"
current_start = i + 3
i += 3
continue
i += 1
sections.append((current_name, current_start, n))
# For each group, find an in-place insertion point or queue for
# the trailer.
section_lookup = {name: idx for idx, (name, _, _) in enumerate(sections)}
# Map section index → list of additional lines to splice in
# right BEFORE the section's end (so they land at the bottom of
# the section, before any blank-line gap to the next banner).
per_section_splice: dict[int, list[str]] = {}
trailer_groups: list[tuple[str, list[tuple[list[str], str, str]]]] = []
in_place_names: list[str] = []
trailer_names: list[str] = []
total = 0
for section_name, entries in groups:
if section_name in section_lookup:
in_place_names.append(section_name)
idx = section_lookup[section_name]
bucket = per_section_splice.setdefault(idx, [])
for context, key, value in entries:
for ctx_line in context:
bucket.append(ctx_line + "\n")
bucket.append(f"{key}={value}\n")
total += 1
else:
trailer_names.append(section_name)
trailer_groups.append((section_name, entries))
total += len(entries)
# Reassemble env_text with in-place splices applied. Walk from
# the end backwards so prior splices don't shift later indices.
out_lines = list(lines)
for idx in sorted(per_section_splice.keys(), reverse=True):
name, start, end = sections[idx]
# Trim trailing blank lines from the section body so the
# spliced entries sit flush with the prior content; the
# blank is reinserted between sections.
insertion_point = end
while (
insertion_point > start
and out_lines[insertion_point - 1].strip() == ""
):
insertion_point -= 1
splice = per_section_splice[idx]
if insertion_point and not out_lines[insertion_point - 1].endswith(("\r", "\n")):
splice = ["\n", *splice]
out_lines[insertion_point:insertion_point] = splice
# Append the trailer for any groups whose section didn't exist.
if trailer_groups:
trailer: list[str] = []
joined = "".join(out_lines)
if joined and not joined.endswith("\n"):
trailer.append("\n")
trailer.extend([
"\n",
"# ────────────────────────────────────────────────────────\n",
f"# Auto-backfilled from .env.example on {_format_today()}\n",
"# Sections new in .env.example since this .env was written.\n",
"# ────────────────────────────────────────────────────────\n",
])
for section_name, entries in trailer_groups:
trailer.append("\n")
trailer.append(f"# === {section_name} ===\n")
for context, key, value in entries:
for ctx_line in context:
trailer.append(ctx_line + "\n")
trailer.append(f"{key}={value}\n")
out_lines.extend(trailer)
return "".join(out_lines), total, in_place_names, trailer_names
def _rebase_backfilled_ports(self, groups):
"""Seed newly backfilled *_PORT keys on this stack's BASE_PORT block:
.env.example carries the default 63000 layout, which a headless
`env backfill` + compose/doctor/endpoints run would otherwise use."""
# Decoded like every reader (quotes, `export`, last assignment wins).
base_port = _parsed_base_port(self.config_parser.parse_env_file().get("BASE_PORT", "") or "")
if base_port is None or not self.port_manager.validate_base_port(base_port):
return groups
assignments = self.port_manager.calculate_port_assignments(base_port)
return [
(section, [
(context, key, str(assignments[key]) if key in assignments else value)
for context, key, value in entries
])
for section, entries in groups
]
@staticmethod
def _parse_env_example_sections(
example_text: str, existing_keys: set[str],
) -> "list[tuple[str, list[tuple[list[str], str, str]]]]":
"""Walk ``example_text`` and group missing variables by section.
Returns a list of ``(section_name, entries)`` where entries is
a list of ``(context_comments, key, value)``. Section name
comes from the most recent ``# ────`` banner (legacy ``# ====`` too).
Context comments are the contiguous comment lines immediately
preceding the variable (an inline description like
``# Required when COMFYUI_SOURCE=localhost:``),
capped to the previous variable or section banner so the
backfill doesn't drag unrelated commentary along.
"""
# Match the 3-line section banner pattern in .env.example.
# env_assembler emits box-drawing `─` (U+2500), the canonical
# form. Legacy `=` bars are also tolerated for
# backwards-compat with hand-edited `.env.example` files.
#
# Example match:
# # ──────────────────────────────────────────────────
# # data: Apache Spark (standalone cluster) (services/spark/service.yml)
# # ──────────────────────────────────────────────────
bar_re = re.compile(r"^#\s*[=─]{3,}\s*$")
lines = env_lines(example_text.removeprefix("\ufeff"))
current_section = "(unsectioned)"
# Buffer of comment lines accumulated since the last variable
# line or section banner — used as the immediate context for
# the next variable we encounter.
comment_buf: list[str] = []
# Section name → list of (context, key, value).
per_section: dict[str, list[tuple[list[str], str, str]]] = {}
ordered_sections: list[str] = []
seen_keys: set[str] = set()
i = 0
n = len(lines)
while i < n:
line = lines[i]
# Section banner detection: bar, title, bar (3 lines).
if (
bar_re.match(line)
and i + 2 < n
and lines[i + 1].startswith("#")
and bar_re.match(lines[i + 2])
):
title = lines[i + 1].lstrip("#").strip()
if title:
current_section = title
comment_buf = []
i += 3
continue
stripped = line.strip()
if not stripped:
# Blank line resets the running comment buffer so we
# don't paste unrelated text above a variable two
# sections later.
comment_buf = []
i += 1
continue
if stripped.startswith("#"):
comment_buf.append(line)
i += 1
continue
if "=" not in stripped:
comment_buf = []
i += 1
continue
key, _decoded, value, _exported = _backfill_env_assignment(stripped, "env example")
if key and key not in existing_keys and key not in seen_keys:
seen_keys.add(key)
if current_section not in per_section:
per_section[current_section] = []
ordered_sections.append(current_section)
per_section[current_section].append(
(list(comment_buf), key, value),
)
comment_buf = []
i += 1
return [(s, per_section[s]) for s in ordered_sections]
def unset_port_environment_variables(self) -> None:
"""
Unset potentially lingering port environment variables.
Replicates the unset logic from the original start.sh.
"""
# Every container-PORT slot the bootstrapper allocates. A stale
# shell-exported value would shadow the freshly-computed value
# on cold-start with a custom --base-port, so unset before we
# re-allocate. Only *_LOCALHOST_PORT vars (host-side override
# ports) stay out of the slot allocator and out of this list;
# every slot-allocated *_PORT must appear here — including the
# cAdvisor / node / postgres / redis exporter ports, which DO
# recompute from BASE_PORT (test_port_unset_covers_topology guards
# this list against topology.port_defaults).
port_variables = [
'SUPABASE_DB_PORT',
'REDIS_PORT',
'KONG_HTTP_PORT',
'KONG_HTTPS_PORT',
'SUPABASE_META_PORT',
'SUPABASE_STORAGE_PORT',
'SUPABASE_AUTH_PORT',
'SUPABASE_API_PORT',
'SUPABASE_REALTIME_PORT',
'SUPABASE_STUDIO_PORT',
'GRAPH_DB_PORT',
'GRAPH_DB_DASHBOARD_PORT',
'LITELLM_PORT',
'LOCAL_DEEP_RESEARCHER_PORT',
'ASSET_WORKER_PORT',
'ASSET_BAKER_PORT',
'SEARXNG_PORT',
'CRAWL4AI_PORT',
'TIKA_PORT',
'LLM_GRAPH_BUILDER_PORT',
'FLOWER_PORT',
'OPEN_WEB_UI_PORT',
'BACKEND_PORT',
'N8N_PORT',
'COMFYUI_PORT',
'WEAVIATE_PORT',
'WEAVIATE_GRPC_PORT',
'DOC_PROCESSOR_PORT',
'STT_PROVIDER_PORT',
'TTS_PROVIDER_PORT',
'SPEACHES_PORT',
'CHATTERBOX_PORT',
'OPENCLAW_GATEWAY_PORT',
'OPENCLAW_BRIDGE_PORT',
'HERMES_API_PORT',
'HERMES_DASHBOARD_PORT',
'MCP_SERVERS_PORT',
'LANGFUSE_PORT',
'TRUEFORGE_PORT',
'TEI_RERANKER_PORT',
'LIGHTRAG_API_PORT',
'ICEBERG_REST_PORT',
'TRINO_PORT',
'REDPANDA_KAFKA_PORT',
'REDPANDA_CONSOLE_PORT',
'MINIO_PORT',
'MINIO_CONSOLE_PORT',
'JUPYTERHUB_PORT',
'JENKINS_PORT',
'MLFLOW_PORT',
'LABEL_STUDIO_PORT',
'VERBA_PORT',
# PR #29 / PR #35 additions: ray + spark + airflow + zeppelin
# + prometheus + grafana. Without these the previous-run
# exports silently shadow the freshly-computed slots.
'RAY_DASHBOARD_PORT',
'RAY_CLIENT_PORT',
'RAY_GCS_PORT',
'SPARK_MASTER_UI_PORT',
'SPARK_HISTORY_PORT',
'AIRFLOW_PORT',
'ZEPPELIN_PORT',
'PROMETHEUS_PORT',
'GRAFANA_PORT',
# Observability exporters: all four are slot-allocated (recompute
# from BASE_PORT) and host-published, so a stale shell export
# shadows the fresh slot on a --base-port cold start just like any
# other port. They were missing here for several releases.
'CADVISOR_PORT',
'NODE_EXPORTER_PORT',
'POSTGRES_EXPORTER_PORT',
'REDIS_EXPORTER_PORT',
]
self.banner.show_status_message(" • Unsetting potentially lingering port environment variables...", "info")
for var in port_variables:
if var in os.environ:
del os.environ[var]
def validate_supabase_keys(self, cold_start: bool = False) -> bool:
"""
Ensure required Supabase JWT keys are present.
Three outcomes:
- All three keys present → no-op.
- All three keys blank (fresh clone, or post-cold-reset of .env) →
auto-generate. No --cold flag required.
- Mixed state (some present, some blank) → refuse and direct the
user to ./bootstrapper/generate_supabase_keys.sh, since the anon
and service keys are HMAC-signed by SUPABASE_JWT_SECRET and the
generator always rewrites all three. Auto-regenerating here would
silently clobber whatever values the user hand-pasted.
Args:
cold_start: Phrases the status message only. Auto-generation
triggers on the all-blank case regardless of this flag.
Returns:
bool: True if all keys are present or successfully generated.
"""
env_vars = self.config_parser.parse_env_file()
keys = {
'SUPABASE_JWT_SECRET': env_vars.get('SUPABASE_JWT_SECRET', '').strip(),
'SUPABASE_ANON_KEY': env_vars.get('SUPABASE_ANON_KEY', '').strip(),
'SUPABASE_SERVICE_KEY': env_vars.get('SUPABASE_SERVICE_KEY', '').strip(),
}
missing_keys = [name for name, value in keys.items() if not value]
if not missing_keys:
return True
# Mixed state: some keys set, others blank. Don't auto-regenerate —
# the SupabaseKeyGenerator rewrites all three together, which would
# destroy whatever the user pasted into the present ones.
if len(missing_keys) < len(keys):
present_keys = [name for name in keys if name not in missing_keys]
self.banner.show_section_header("Inconsistent Supabase Keys", "⚠️")
self.banner.show_status_message(
"Some Supabase keys are set and others are blank — refusing to "
"auto-regenerate to avoid clobbering the values you've already set.",
"warning",
)
self.banner.show_status_message(
f" Present: {', '.join(present_keys)}", "warning",
)
self.banner.show_status_message(
f" Missing: {', '.join(missing_keys)}", "warning",
)
self.banner.show_status_message(" To resolve:", "info")
self.banner.show_status_message(
" • Run ./bootstrapper/generate_supabase_keys.sh to regenerate "
"ALL three (overwrites existing), or",
"info",
)
self.banner.show_status_message(
" • Manually fill in the missing keys in .env so all three "
"are populated.",
"info",
)
return False
self.banner.show_section_header("Generating Supabase Keys", "🔐")
if cold_start:
self.banner.show_status_message(
"Cold start detected - generating fresh Supabase JWT keys...",
"info",
)
else:
self.banner.show_status_message(
"No Supabase keys found in .env; generating fresh JWT keys...",
"info",
)
from utils.supabase_keys import SupabaseKeyGenerator
key_generator = SupabaseKeyGenerator(str(self.root_dir))
if key_generator.generate_and_update_env():
self.banner.show_status_message(
"Supabase keys generated and applied successfully!", "success"
)
return True
self.banner.show_status_message("Failed to generate Supabase keys", "error")
return False
def handle_port_configuration(self, base_port: Optional[int]) -> bool:
"""Handle port configuration and updates."""
# No --base-port flag: preserve the BASE_PORT already configured in
# .env (e.g. from an earlier --base-port 64000 run) instead of
# silently rewriting every *_PORT back to the default layout. This
# mirrors the TUI path's fallback in ui/textual/integration.py.
if base_port is None:
current = (self.config_parser.parse_env_file()
.get('BASE_PORT', '') or '').strip()
if current.lower() == "auto":
# Defensive: a manifest BASE_PORT=auto is normally resolved to a
# concrete value by _resolve_auto_base_port_override during the
# env-overlay merge; if it ever reaches here unresolved, pick a
# free block rather than int()-failing to the default 63000.
base_port = self.port_manager.auto_base_port() or DEFAULT_BASE_PORT
else:
try:
base_port = int(current)
except ValueError:
base_port = DEFAULT_BASE_PORT
base_port = self._reresolve_auto_block(base_port)
# Validate base port
if not self.port_manager.validate_base_port(base_port):
offsets = self.port_manager.port_offsets()
max_offset = max(offsets.values()) if offsets else 0
self.banner.show_status_message(
f"Invalid base port: {base_port}. Must be between 1024 and "
f"{65535 - max_offset}",
"error"
)
return False
# Check for port conflicts
conflicts = self.port_manager.get_port_conflicts(base_port)
if conflicts or self._port_block_moves(base_port):
# Our own containers hold the ports, or the block moves: a moved
# block re-publishes every container, and a targeted warm start
# would leave a now-disabled service running on its old ports.
if self.docker_manager.are_project_containers_running():
self.banner.show_status_message(
"Previous instance detected — stopping existing containers...",
"info"
)
stop_result = self.docker_manager.stop_services(
remove_volumes=False, remove_orphans=True
)
if stop_result != 0:
self.banner.show_status_message(
"Failed to stop previous instance", "error"
)
return False
self.banner.show_status_message(
"Previous instance stopped successfully", "success"
)
self.stopped_previous_instance = True
# Re-check ports after cleanup, waiting for Docker to
# release the ports it just unpublished (#1438).
conflicts = self.port_manager.conflicts_after_release(base_port)
elif conflicts and getattr(self, "project_stopped_this_run", False):
# A cold start stopped the stack before this check; wait for
# Docker to release its ports as the warm path does.
conflicts = self.port_manager.conflicts_after_release(base_port)
# If conflicts remain, show the original error
if conflicts:
self.banner.show_status_message("Port conflicts detected:", "warning")
if self.stopped_previous_instance:
self.banner.show_status_message(
" (the previous instance was stopped above and stays down)",
"warning",
)
for port_var, port in conflicts.items():
self.banner.show_status_message(
f" • {port_var}: Port {port} is already in use", "warning"
)
# Suggest alternative base port
suggested_port = self.port_manager.suggest_available_base_port()
if suggested_port:
self.banner.show_status_message(
f"Suggested available base port: {suggested_port}",
"info"
)
return False
# Update ports in .env file
if not self.port_manager.update_env_ports(base_port):
return False
return True
def _reresolve_auto_block(self, base_port: int) -> int:
"""An `auto` block was probed before the profile, the consumer
manifest and the CLI flags had all written their *_SOURCE values. If a
service they enable now finds its port taken, choose again with the
sources in .env, before a running stack is stopped for a block this
launch would then refuse (2026-10-08 run, cycle 75)."""
if not self.base_port_auto_chosen:
return base_port
if not self.port_manager.check_port_range_availability(base_port):
return base_port
fresh = self.port_manager.auto_base_port()
if fresh is None or fresh == base_port:
return base_port
self.banner.show_status_message(
f"BASE_PORT=auto: block {base_port} has a port in use by a service "
f"this launch enables; using {fresh} instead.",
"warning",
)
return fresh
def _port_block_moves(self, base_port: int) -> bool:
"""Whether `.env`'s ports differ from ``base_port``'s block: a
--base-port change, or a hand-edited BASE_PORT whose *_PORT values
still name the old block the running containers publish."""
env = self.config_parser.parse_env_file()
current = (env.get('BASE_PORT', '') or '').strip()
# ASCII only: "²".isdigit() is true but int("²") raises, which
# crashed the start where handle_port_configuration falls back.
if _ascii_int(current) not in (None, base_port):
return True
target = self.port_manager.calculate_port_assignments(base_port)
# A *_PORT pinned in .env.user / a consumer manifest is merged into
# .env on every start and reset by update_env_ports; it is not a
# move, and counting it tore the stack down on every warm start.
pinned = set(getattr(self, "_env_user_keys", None) or ())
return any(
_ascii_int(str(env.get(var, ''))) not in (None, port)
for var, port in target.items() if var not in pinned
)
def run_port_migration(self, no_port_migrate: bool) -> None:
"""Chained .env migrations: v0 → v1 (port-layout), v1 → v2 (URL→PORT),
v2 → v3 (COMFYUI_MODEL_SET → COMFYUI_USER_MODELS schema), v3 → v4
(stale curated Ollama model reference cleanup), v4 → v5
(enable Weaviate's native filesystem backup module).
Idempotent. Each step is gated by its own ``_needs_*`` predicate
so re-running is safe and stamping is independent. Reads
``BOOTSTRAPPER_PORT_LAYOUT_VERSION`` from the active env file
(honors ``ATLAS_ENV_FILE``).
v1: rewrites every port var whose current value matches the v0
default to the topology-derived v1 default. User-customized
values are left alone.
v2: rewrites legacy ``<SVC>_LOCALHOST_URL`` lines into
``<SVC>_LOCALHOST_PORT`` lines, commenting the old URLs for
audit. Drives both compose runtime and Kong routes off the new
PORT schema.
v3: translates the old ``COMFYUI_MODEL_SET`` enum to the new
``COMFYUI_USER_MODELS`` CSV + sidecar/cache vars introduced in
the model-picker feature. Removes the old enum var and any
preceding comment block.
v4: rewrites LITELLM_DEFAULT_MODEL / LITELLM_VISION_MODEL /
OLLAMA_USER_MODELS entries that reference a curated Ollama
model retired from ``services/ollama/models.yaml`` (e.g.
``qwen3.6`` → ``qwen3.8``) to the current catalog equivalent.
v5: appends ``backup-filesystem`` to a persisted
``WEAVIATE_ENABLE_MODULES`` list without changing operator-selected
modules, so upgrades receive the executable backup contract.
When ``no_port_migrate`` is True we skip all five migrations AND skip
the sentinel stamps so the next run re-prompts — matches the
user intent "skip this run, ask next time."
Must be called AFTER setup_env_file + backfill so the file
exists and is fully populated, and BEFORE any caller that
relies on the v2 port values.
"""
env_path = self.config_parser.env_file_path
if not no_port_migrate and _prepare_migration_marker(env_path):
return
# v0 → v1: port-layout rewrite.
if _needs_v1(env_path):
if no_port_migrate:
self.banner.console.print(
"[dim]Skipping port-layout v1 migration (--no-port-migrate); "
"will re-prompt next run.[/dim]"
)
else:
from services.topology import get_topology as _get_topology
services_root = self.root_dir / "services"
env_vars = self.config_parser.parse_env_file()
# ``.get(key, default)`` returns the empty string when the key is
# present-but-blank — only missing keys hit the default. A blank
# BASE_PORT (auto-managed quirk) would crash ``int("")``.
_raw_base = (env_vars.get("BASE_PORT") or "").strip()
try:
base_port = int(_raw_base) if _raw_base else DEFAULT_BASE_PORT
except ValueError:
base_port = DEFAULT_BASE_PORT
topology = _get_topology(services_root, base_port=base_port)
result = _apply_v1(env_path, topology.port_defaults, base_port=base_port)
if result.backup_path:
self.banner.console.print(
f"[green]• Backed up .env to {result.backup_path}[/green]"
)
self.banner.console.print(
f"[green]• Port layout updated (v0 → v1)[/green]: "
f"rewrote {len(result.rewritten)} ports; "
f"preserved {len(result.preserved)} customizations."
)
if result.rewritten:
self.banner.console.print("[dim] Changes:[/dim]")
for var, (old, new) in sorted(result.rewritten.items()):
self.banner.console.print(
f"[dim] {var}: {old} → {new}[/dim]"
)
_stamp_v1(env_path, 1)
# v1 → v2: URL → PORT schema rewrite. Idempotent on re-run.
# Runs after v1 so the sentinel transitions cleanly 0/none → 1 → 2
# rather than skipping intermediate state on a v0 .env (the
# combined behavior is what we want for users on older checkouts
# who haven't run any bootstrapper since the topology refactor).
if _needs_v2(env_path):
if no_port_migrate:
self.banner.console.print(
"[dim]Skipping LOCALHOST schema migration "
"(--no-port-migrate); will re-prompt next run.[/dim]"
)
else:
self.banner.show_status_message(
"Migrating .env to LOCALHOST_PORT schema (v2) ...",
"info",
)
_apply_v2(env_path)
_stamp_v2(env_path)
self.banner.show_status_message(
"LOCALHOST schema migration complete (v2). "
"Old <SVC>_LOCALHOST_URL lines are commented out for "
"audit; new <SVC>_LOCALHOST_PORT lines drive both "
"compose runtime and Kong routes.",
"success",
)
# v2 → v3: COMFYUI_MODEL_SET → COMFYUI_USER_MODELS schema rewrite.
# Runs after v2 so the sentinel transitions cleanly: … → 2 → 3.
if _needs_v3(env_path):
if no_port_migrate:
self.banner.console.print(
"[dim]Skipping model-set schema migration "
"(--no-port-migrate); will re-prompt next run.[/dim]"
)
else:
self.banner.show_status_message(
"Migrating .env to COMFYUI_USER_MODELS schema (v3) ...",
"info",
)
_apply_v3(env_path)
_stamp_v3(env_path)
self.banner.show_status_message(
"Model-set migration complete (v3). "
"COMFYUI_MODEL_SET translated to COMFYUI_USER_MODELS.",
"success",
)
# v3 → v4: stale curated Ollama model reference cleanup.
# Runs after v3 so the sentinel transitions cleanly: … → 3 → 4.
if _needs_v4(env_path):
if no_port_migrate:
self.banner.console.print(
"[dim]Skipping stale-model-reference migration "
"(--no-port-migrate); will re-prompt next run.[/dim]"
)
else:
self.banner.show_status_message(
"Checking for retired curated Ollama model references (v4) ...",
"info",
)
if _apply_v4(env_path):
_stamp_v4(env_path)
self.banner.show_status_message(
"Stale-model-reference migration complete (v4).",
"success",
)
else:
self.banner.show_status_message(
"Could not load the curated Ollama catalog; migration v4 "
"was not stamped and will retry on the next start.",
"warning",
)
# v4 → v5: retain custom modules and add the native backup provider.
# Never leapfrog a retryable v4 failure by stamping the shared
# migration sentinel directly to 5.
if not _needs_v4(env_path) and _needs_v5(env_path):
if no_port_migrate:
self.banner.console.print(
"[dim]Skipping Weaviate backup-module migration "
"(--no-port-migrate); will re-prompt next run.[/dim]"
)
else:
try:
_apply_v5(env_path)
except MigrationV5Error as exc:
# A hand-edited .env (duplicate or malformed
# WEAVIATE_ENABLE_MODULES line) must not crash every start
# with a traceback. Like v4: report, leave unstamped, retry.
self.banner.show_status_message(
f"Weaviate backup-module migration (v5) skipped: {exc} "
f"in {env_path}. Fix the line; the migration retries on "
"the next start.",
"warning",
)
else:
_stamp_v5(env_path)
self.banner.show_status_message(
"Weaviate backup-module migration complete (v5).",
"success",
)
def generate_service_configuration(self) -> bool:
"""Generate and update service configuration."""
try:
generated = self.service_config.generate_and_update_env()
except ValueError as exc:
# The per-service gates (e.g. Spark needs MinIO) raise ValueError
# with an actionable message; the --no-tui flow has no handler
# above this step, so report it instead of a raw traceback.
# print, not the banner: the TUI swaps in a no-op banner and
# captures stdout into its log pane.
print(f"ERROR: {exc}")
return False
if not generated:
return False
# Finalize consumer object-storage (#404) AFTER endpoints are resolved
# into .env by generate_and_update_env, and before compose up. Covers
# both the linear flow and the TUI/launch pipeline (single call site).
if not self._finalize_consumer_storage():
return False
# Finalize consumer LiteLLM model rows (#411): write the generated
# consumer-models.yaml (merged by litellm-init) + the api-key overlay,
# or remove them when no consumer declares litellm_models. Same call
# site so both the linear and TUI/launch flows are covered.
if not self._finalize_consumer_litellm_models():
return False
# Finalize consumer n8n workflow seeding (#412): write the normalized
# workflow JSONs + plan + seed overlay, or remove stale artifacts when no
# consumer declares n8n_workflows.
if not self._finalize_consumer_n8n_workflows():
return False
# Finalize consumer RAG ingestion profiles (#413): write the compiled
# profiles JSON + backend-mount overlay, or remove them when no consumer
# declares rag_ingestion_profiles.
if not self._finalize_consumer_rag_ingestion_profiles():
return False
# Finalize consumer LightRAG query profiles (#414): write the compiled
# query-profile registry JSON + backend-mount overlay, or remove them
# when no consumer declares lightrag_query_profiles.
if not self._finalize_consumer_lightrag_query_profiles():
return False
return True
def _finalize_consumer_storage(self) -> bool:
"""Provision manifest-declared object stores (#404): generate scoped
credentials (blank-only), export stable endpoint/credential-reference
fields, and write the Atlas-owned minio-init overlay so no consumer
compose override is needed. Idempotent; a no-storage config removes any
stale generated overlay.
"""
from core.consumer_manifest import (
MINIO_STORAGE_OVERLAY_PATH,
_validate_storage_collisions,
compile_storage_exports,
storage_credential_tokens,
)
try:
config = self.config_parser.load_consumer_config()
if config.storage:
env = self.config_parser.parse_env_file()
if str(env.get("MINIO_SOURCE", "container")).strip() != "disabled":
_validate_storage_collisions(
config.storage, self.docker_manager.resolved_minio_environment(),
)
except Exception as exc: # ConsumerManifestError etc. — surface + fail
self.banner.show_status_message(
f"Consumer storage manifest error: {exc}", "error"
)
return False
overlay_path = self.root_dir / MINIO_STORAGE_OVERLAY_PATH
# Always clear stale export vars first — a removed or renamed store must
# not leave dangling ATLAS_STORE_* fields; the current set is re-emitted
# below. Provisioning (buckets / MINIO_EXTRA_CONSUMERS) lives only in the
# regenerated overlay, so there is nothing stale to strip from .env there.
self._remove_env_keys_by_prefix("ATLAS_STORE_")
if not config.storage:
if overlay_path.exists():
overlay_path.unlink()
return True
if str(env.get("MINIO_SOURCE", "container")).strip() == "disabled":
# A storage declaration against disabled MinIO can't be provisioned;
# skip loudly rather than emit an unusable (empty-endpoint) contract.
if overlay_path.exists():
overlay_path.unlink()
self.banner.show_status_message(
" • MINIO_SOURCE=disabled — skipping consumer storage "
f"provisioning for {len(config.storage)} declared store(s)",
"warning",
)
return True
# 1. Scoped credentials — generated once, persist across restarts.
tokens = storage_credential_tokens(config.storage)
self.key_generator.generate_and_update_extra_minio_consumer_keys(tokens)
# 2. Stable export fields (resolved endpoints track BASE_PORT/host).
exports = compile_storage_exports(
config.storage,
minio_endpoint=env.get("MINIO_ENDPOINT", ""),
minio_public_endpoint=env.get("MINIO_PUBLIC_ENDPOINT", ""),
minio_region=env.get("MINIO_REGION", "us-east-1"),
)
if exports:
self._merge_env_file_overrides(exports)
# 3. Atlas-owned minio-init overlay (routed via consumer overlays).
overlay = config.storage_overlay
if overlay is not None:
overlay.path.parent.mkdir(parents=True, exist_ok=True)
overlay.path.write_text(overlay.content, encoding="utf-8")
self.banner.show_status_message(
f" • Provisioned {len(config.storage)} consumer object store(s)",
"info",
)
return True
def _finalize_consumer_litellm_models(self) -> bool:
"""Materialize consumer-owned LiteLLM model rows (#411): write the
generated ``consumer-models.yaml`` (merged by litellm-init into the
LiteLLM config before startup) and the companion api-key compose overlay
(injects consumer api-key references into the litellm container).
Idempotent; a no-models config removes any stale generated artifacts so a
removed manifest drops only its own rows on a warm restart.
"""
from core.consumer_manifest import (
LITELLM_CONSUMER_MODELS_PATH,
LITELLM_CONSUMER_OVERLAY_PATH,
)
try:
config = self.config_parser.load_consumer_config()
except Exception as exc: # ConsumerManifestError etc. — surface + fail
self.banner.show_status_message(
f"Consumer LiteLLM manifest error: {exc}", "error"
)
return False
models_path = self.root_dir / LITELLM_CONSUMER_MODELS_PATH
overlay_path = self.root_dir / LITELLM_CONSUMER_OVERLAY_PATH
# No consumer declares litellm_models → remove any stale generated
# artifacts so nothing leaks into a later, model-less start.
if not config.litellm_models:
for stale in (models_path, overlay_path):
if stale.exists():
stale.unlink()
return True
artifact = config.litellm_models_file
if artifact is not None:
artifact.path.parent.mkdir(parents=True, exist_ok=True)
artifact.path.write_text(artifact.content, encoding="utf-8")
# The overlay only exists when at least one model declares an api_key_var.
# Remove a stale overlay when the current config no longer needs one.
if config.litellm_overlay is not None:
config.litellm_overlay.path.parent.mkdir(parents=True, exist_ok=True)
config.litellm_overlay.path.write_text(
config.litellm_overlay.content, encoding="utf-8"
)
elif overlay_path.exists():
overlay_path.unlink()
owners = sorted({model.consumer for model in config.litellm_models})
self.banner.show_status_message(
f" • Merged {len(config.litellm_models)} consumer LiteLLM model(s) "
f"from {', '.join(owners)}",
"info",
)
return True
def _finalize_consumer_n8n_workflows(self) -> bool:
"""Materialize consumer-owned n8n workflow seeding (#412): write the
normalized workflow JSONs + plan.json into the gitignored seed dir and the
seed compose overlay, or remove stale generated artifacts when no consumer
declares n8n_workflows. Idempotent; a removed manifest or workflow drops
exactly its own artifacts on the next start."""
from core.consumer_manifest import (
N8N_CONSUMER_OVERLAY_PATH,
N8N_CONSUMER_WORKFLOWS_DIR,
)
try:
config = self.config_parser.load_consumer_config()
except Exception as exc: # ConsumerManifestError etc. — surface + fail
self.banner.show_status_message(
f"Consumer n8n workflow manifest error: {exc}", "error"
)
return False
seed_dir = self.root_dir / N8N_CONSUMER_WORKFLOWS_DIR
overlay_path = self.root_dir / N8N_CONSUMER_OVERLAY_PATH
if self._skip_n8n_seed_when_disabled(config, seed_dir, overlay_path):
return True
# No consumer declares n8n_workflows → remove any stale generated
# artifacts so a warm restart doesn't re-seed removed workflows.
if not config.n8n_workflows:
if (seed_dir / "plan.json").exists():
return self._reconcile_last_n8n_workflows(seed_dir, overlay_path)
if seed_dir.exists():
for stale in seed_dir.glob("*.json"):
stale.unlink()
if overlay_path.exists():
overlay_path.unlink()
return True
seed_dir.mkdir(parents=True, exist_ok=True)
# Drop stale *.json (workflows removed since the last start) first, or a
# removed workflow's file lingers and the seed would re-import it.
current_files = {artifact.path.name for artifact in config.n8n_artifacts}
for stale in seed_dir.glob("*.json"):
if stale.name not in current_files:
stale.unlink()
for artifact in config.n8n_artifacts:
artifact.path.parent.mkdir(parents=True, exist_ok=True)
artifact.path.write_text(artifact.content, encoding="utf-8")
if config.n8n_overlay is not None:
config.n8n_overlay.path.parent.mkdir(parents=True, exist_ok=True)
config.n8n_overlay.path.write_text(
config.n8n_overlay.content, encoding="utf-8"
)
owners = sorted({wf.consumer for wf in config.n8n_workflows})
self.banner.show_status_message(
f" • Seeding {len(config.n8n_workflows)} consumer n8n workflow(s) "
f"from {', '.join(owners)}",
"info",
)
return True
def _skip_n8n_seed_when_disabled(self, config, seed_dir: Path, overlay_path: Path) -> bool:
"""With N8N_SOURCE=disabled there is no n8n for the seed to wait on:
drop the overlay and warn, as consumer storage does for disabled MinIO.
The seed dir and plan.json stay, so a pending empty-plan reconcile
resumes once n8n is enabled again."""
if not _n8n_disabled(self.config_parser.parse_env_file()):
return False
if overlay_path.exists():
overlay_path.unlink()
if config.n8n_workflows:
detail = f" for {len(config.n8n_workflows)} declared workflow(s)"
elif (seed_dir / "plan.json").exists():
detail = "; previously seeded workflows are reconciled when n8n is enabled again"
else:
return True
self.banner.show_status_message(
f" • N8N_SOURCE=disabled — skipping the consumer n8n seed{detail}",
"warning",
)
return True
def _reconcile_last_n8n_workflows(self, seed_dir: Path, overlay_path: Path) -> bool:
"""Workflows were seeded before and none is declared now: run the seed
with an empty plan on every start, so its reconcile deactivates and
deletes every ``atlas-consumer-*`` workflow. Dropping the overlay
skipped the seed and left their webhooks live (cycle 31); dropping it
after one run lost the cleanup whenever that run could not reconcile
(no N8N_API_KEY, n8n not healthy) (2026-10-08 run, cycle 34). Delete
volumes/n8n/consumer-workflows/ to stop it."""
from core.consumer_manifest import compile_n8n_plan, render_n8n_seed_overlay
for stale in seed_dir.glob("*.json"):
if stale.name != "plan.json":
stale.unlink()
(seed_dir / "plan.json").write_text(compile_n8n_plan([]), encoding="utf-8")
overlay_path.parent.mkdir(parents=True, exist_ok=True)
overlay_path.write_text(render_n8n_seed_overlay([]), encoding="utf-8")
self.banner.show_status_message(
" • No consumer n8n workflows declared: the n8n seed removes any previously seeded ones",
"info",
)
if not (self.config_parser.parse_env_file().get("N8N_API_KEY") or "").strip():
self.banner.show_status_message(
" • N8N_API_KEY is not set, so previously seeded consumer workflows stay active "
"until it is set",
"warning",
)
return True
def _finalize_consumer_rag_ingestion_profiles(self) -> bool:
"""Materialize consumer-owned RAG ingestion profiles (#413): write the
compiled profiles JSON + the backend-mount compose overlay, or remove any
stale generated artifacts when no consumer declares rag_ingestion_profiles.
Idempotent; a removed manifest drops exactly its own artifacts next start.
"""
from core.consumer_manifest import (
RAG_INGESTION_OVERLAY_PATH,
RAG_INGESTION_PROFILES_PATH,
)
try:
config = self.config_parser.load_consumer_config()
except Exception as exc: # ConsumerManifestError etc. — surface + fail
self.banner.show_status_message(
f"Consumer RAG ingestion manifest error: {exc}", "error"
)
return False
profiles_path = self.root_dir / RAG_INGESTION_PROFILES_PATH
overlay_path = self.root_dir / RAG_INGESTION_OVERLAY_PATH
if not config.rag_ingestion_profiles:
# Remove stale artifacts so a warm restart doesn't mount removed
# profiles into the backend.
if profiles_path.exists():
profiles_path.unlink()
if overlay_path.exists():
overlay_path.unlink()
return True
if config.rag_ingestion_file is not None:
config.rag_ingestion_file.path.parent.mkdir(parents=True, exist_ok=True)
config.rag_ingestion_file.path.write_text(
config.rag_ingestion_file.content, encoding="utf-8"
)
if config.rag_ingestion_overlay is not None:
config.rag_ingestion_overlay.path.parent.mkdir(parents=True, exist_ok=True)
config.rag_ingestion_overlay.path.write_text(
config.rag_ingestion_overlay.content, encoding="utf-8"
)
owners = sorted({p.consumer for p in config.rag_ingestion_profiles})
self.banner.show_status_message(
f" • Registering {len(config.rag_ingestion_profiles)} consumer RAG "
f"ingestion profile(s) from {', '.join(owners)}",
"info",
)
return True
def _finalize_consumer_lightrag_query_profiles(self) -> bool:
"""Materialize consumer-owned LightRAG query profiles (#414): write the
compiled query-profile registry JSON + the backend-mount compose overlay,
or remove any stale generated artifacts when no consumer declares
lightrag_query_profiles. Idempotent; a removed manifest drops exactly its
own artifacts next start, so a no-profile deployment stays compatible.
"""
from core.consumer_manifest import (
LIGHTRAG_QUERY_PROFILES_OVERLAY_PATH,
LIGHTRAG_QUERY_PROFILES_PATH,
)
try:
config = self.config_parser.load_consumer_config()
except Exception as exc: # ConsumerManifestError etc. — surface + fail
self.banner.show_status_message(
f"Consumer LightRAG query profile manifest error: {exc}", "error"
)
return False
profiles_path = self.root_dir / LIGHTRAG_QUERY_PROFILES_PATH
overlay_path = self.root_dir / LIGHTRAG_QUERY_PROFILES_OVERLAY_PATH
if not config.lightrag_query_profiles:
# Remove stale artifacts so a warm restart doesn't mount removed
# profiles into the backend.
if profiles_path.exists():
profiles_path.unlink()
if overlay_path.exists():
overlay_path.unlink()
return True
if config.lightrag_query_profiles_file is not None:
config.lightrag_query_profiles_file.path.parent.mkdir(
parents=True, exist_ok=True
)
config.lightrag_query_profiles_file.path.write_text(
config.lightrag_query_profiles_file.content, encoding="utf-8"
)
if config.lightrag_query_profiles_overlay is not None:
config.lightrag_query_profiles_overlay.path.parent.mkdir(
parents=True, exist_ok=True
)
config.lightrag_query_profiles_overlay.path.write_text(
config.lightrag_query_profiles_overlay.content, encoding="utf-8"
)
owners = sorted({p.consumer for p in config.lightrag_query_profiles})
self.banner.show_status_message(
f" • Registering {len(config.lightrag_query_profiles)} consumer LightRAG "
f"query profile(s) from {', '.join(owners)}",
"info",
)
return True
def preflight_managed_host_processes(self) -> bool:
"""Refuse a launch whose managed host would refuse to start (#1342).
Runs before port configuration, which stops a running stack on a warm
start. The checks are the ones the host start treats as fatal (host
preflight, untrusted pid record, a foreign listener on the port), all
read-only, so a failure leaves the previous containers running.
"""
env = self.config_parser.parse_env_file()
for var, value, label, module in _MANAGED_HOST_SOURCES:
if (env.get(var) or "").strip() != value:
continue
problem = _managed_host_pool_blocker(importlib.import_module(module), env, label)
if problem:
self.banner.show_status_message(
f"Managed {label} host cannot start: {problem}. The launch "
"stops here, before a warm start would stop the running "
"containers.",
"error",
)
return False
return True
def start_managed_host_processes(self) -> bool:
"""Start selected native hosts immediately before Compose startup.
Configuration generation remains side-effect free, and a later managed
host failure rolls back only processes created by this invocation.
"""
try:
if not self._finalize_managed_comfyui_mps():
self.rollback_managed_host_processes()
return False
if not self._finalize_managed_vllm_metal():
self.rollback_managed_host_processes()
return False
if not self._finalize_managed_blender_mcp():
self.rollback_managed_host_processes()
return False
# #757: non-fatal — a host daemon the user hasn't started (or a
# typo'd tag) must never abort the stack; it only warns.
self._finalize_ollama_localhost_models()
except BaseException:
self.rollback_managed_host_processes()
raise
return True
def _leave_legacy_managed_host(self, label: str, exc: BaseException) -> bool:
"""Warn and carry on when ``exc`` is the stamp-less pid refusal (#990).
A pid file from an Atlas pin older than the managed-host framework has
no identity stamp, so the guard cannot prove the live process is
Atlas's and will not signal it. That refusal stands. What changes is
its cost: it no longer aborts the whole bring-up before Compose runs.
The process is left exactly as found (typically the previous pin's
host, still serving) and nothing is added to rollback ownership,
because this run created nothing. Every other ownership refusal stays
fatal. A stamped record whose live pid is provably a younger process
never reaches here: the guard treats that recycled pid as a stale
record and the start replaces it (#1341).
"""
from services import legacy_pid_refusal_file
if legacy_pid_refusal_file(exc) is None:
return False
# The remediation commands end the message, so nothing is appended
# after them that could be pasted into a shell with the path.
self.banner.show_status_message(
f"Managed {label} host left as found; the rest of the stack starts, "
f"and once the old process is stopped the next ./start.sh launches "
f"a fresh, identity-stamped {label} process. {exc}",
"warning",
)
return True
def rollback_managed_host_processes(self) -> bool:
"""Stop native hosts started by this invocation, in reverse order."""
from services import tracked_process_may_survive
remaining: list[tuple[str, object]] = []
all_stopped = True
for label, manager in reversed(self._managed_hosts_started_this_run):
try:
manager.stop()
pid, may_survive = tracked_process_may_survive(manager)
except Exception as exc: # noqa: BLE001 - preserve other cleanup
self.banner.show_status_message(
f"Could not roll back managed {label} host: {exc}", "warning"
)
remaining.append((label, manager))
all_stopped = False
continue
if not may_survive:
self.banner.show_status_message(
f"Rolled back managed {label} host started by this launch.",
"info",
)
else:
detail = f" (pid {pid})" if pid is not None else ""
self.banner.show_status_message(
f"Managed {label} host may still be running after rollback"
f"{detail}; tracking evidence was preserved.",
"warning",
)
remaining.append((label, manager))
all_stopped = False
self._managed_hosts_started_this_run = list(reversed(remaining))
return all_stopped
def commit_managed_host_processes(self) -> None:
"""Release rollback ownership after the stack has converged."""
for _label, manager in self._managed_hosts_started_this_run:
commit = getattr(manager, "commit_started_process", None)
if callable(commit):
commit()
self._managed_hosts_started_this_run.clear()
def _finalize_managed_blender_mcp(self) -> bool:
"""Bring up the Atlas-managed headless Blender + MCP bridge (#759).
``BLENDER_MCP_SOURCE=managed-localhost`` provisions the pinned add-on
+ launcher and runs `blender --background` with the main-thread queue
shim, so composition automation needs no GUI and no manual Connect
click. Fatal on failure (the user explicitly chose this source);
no-op for localhost (user-run GUI) and disabled. With
BLENDER_MCP_INSTANCES=N (#851) every pool instance is started and
health-gated on its own allocated port, and instances left above N by
a smaller pool are stopped first.
"""
from services.blender_mcp_manager import pool_from_env, pool_moves, share_verified_addon
env = self.config_parser.parse_env_file()
if (env.get("BLENDER_MCP_SOURCE", "") or "").strip() != "managed-localhost":
return True
pool = pool_from_env(env)
count = f" ({len(pool)} instances)" if len(pool) > 1 else ""
self.banner.show_status_message(
" • BLENDER_MCP_SOURCE=managed-localhost — provisioning the pinned "
f"blender-mcp add-on and launching headless Blender{count}…",
"info",
)
self._reap_stray_blender_mcp(env)
if len(pool) > 1 and pool_moves(pool):
# A shifted base port can make one instance's new port another's
# old one, so the pool restarts together rather than one by one.
for manager in pool:
manager.stop()
for manager in pool:
if getattr(manager, "pool_index", 0) == 1:
share_verified_addon(pool)
if not self._start_blender_mcp_instance(manager):
return False
return True
def _reap_stray_blender_mcp(self, env: dict) -> None:
"""Stop pool instances above the configured size (#851)."""
from services.blender_mcp_manager import manager_from_env
reap_stray_blender_mcp(manager_from_env(env), self.banner.show_status_message)
def _start_blender_mcp_instance(self, manager) -> bool:
from services.blender_mcp_manager import BlenderMcpError
index = getattr(manager, "pool_index", 0)
label = f"Blender MCP #{index}" if index else "Blender MCP"
try:
status, created = manager.ensure_running()
except BlenderMcpError as exc:
if self._leave_legacy_managed_host(label, exc):
return True
if exc.surviving_process:
self._managed_hosts_started_this_run.append((label, manager))
self.banner.show_status_message(
f"Managed {label} bridge could not start: {exc}", "error"
)
return False
if created:
self._managed_hosts_started_this_run.append((label, manager))
health = manager.health()
if health.get("reachable"):
self.banner.show_status_message(
f" • {label} bridge healthy on tcp://{manager.bind}:{manager.port} "
f"(scene objects: {health.get('objects')})",
"info",
)
else:
self.banner.show_status_message(
f" • {label} bridge started but not yet answering commands — "
"first load can lag; check `./start.sh blender-mcp health`.",
"warning",
)
return True
def _finalize_ollama_localhost_models(self) -> None:
"""Pull declared models onto the host Ollama for ``ollama-localhost``
(#757) — the host analog of the ``ollama-pull`` init container, so a
consumer's ``model_sidecars.ollama`` declaration provisions identically
across sources. Idempotent (present tags skip; Ollama verifies layers
natively) and strictly non-fatal: an unreachable daemon or a failed tag
warns and moves on. Pulled models surface in LiteLLM ``/v1/models``
without a restart via OLLAMA_AUTO_IMPORT_LOCAL_MODELS (default true).
"""
from services.ollama_localhost import declared_models, pull_declared_models
env = self.config_parser.parse_env_file()
if (env.get("LLM_PROVIDER_SOURCE", "") or "").strip() != "ollama-localhost":
return
declared = declared_models(env)
if not declared:
return
self.banner.show_status_message(
f" • LLM_PROVIDER_SOURCE=ollama-localhost — ensuring {len(declared)} "
f"declared model tag(s) on the host Ollama (present tags skip)…",
"info",
)
result = pull_declared_models(
env, log=lambda m: self.banner.show_status_message(f" {m}", "info")
)
if not result.reachable:
self.banner.show_status_message(
" Host Ollama is not reachable — start it (`ollama serve`) and "
"re-run ./start.sh to provision the declared models.",
"warning",
)
return
if result.failed:
for failure in result.failed:
self.banner.show_status_message(f" ✗ {failure}", "warning")
self.banner.show_status_message(
" • Some Ollama models could not be pulled — the stack starts "
"anyway; fix the tag(s) and re-run ./start.sh.",
"warning",
)
def _finalize_managed_comfyui_mps(self) -> bool:
"""Bring up the Atlas-managed Apple-Silicon/Metal ComfyUI host (#335).
Docker Desktop on macOS can't pass Metal into a Linux container, so the
``managed-localhost-mps`` source runs a native ComfyUI process on the HOST
and containers reach it via ``host.docker.internal`` (identical wiring to
the unmanaged ``localhost`` source). When that source is selected, preflight
+ install (idempotent — only the first run downloads Torch) + start the
host process here, before ``docker compose up``, so the endpoint is live
when downstream containers come up. A no-op for every other source, so CI
(whose ``.env.example`` default is ``container-cpu``) never touches it.
Fatal on failure: the user explicitly chose this source, and downstream
image generation is broken without the host process — surface it loudly
rather than boot a half-configured stack.
"""
from services.comfyui_mps_manager import ComfyUiMpsError, manager_from_env
env = self.config_parser.parse_env_file()
if str(env.get("COMFYUI_SOURCE", "")).strip() != "managed-localhost-mps":
return True
manager = manager_from_env(env)
self.banner.show_status_message(
" • COMFYUI_SOURCE=managed-localhost-mps — preparing the native "
"Apple-Silicon/Metal ComfyUI host (first run downloads Torch; this "
"can take several minutes)…",
"info",
)
# #754: provision the resolved model set into the host tree BEFORE the
# process serves requests — same catalog selection the container init
# would pull, same non-fatal philosophy (a failed download must not
# abort the stack; generation just 404s for that model until re-run).
rows = _resolved_comfyui_model_rows(env)
if rows and not manager.preflight().ok:
self.banner.show_status_message(
" • Skipping model provisioning — host preflight fails (the "
"start below will report why); nothing multi-GB is downloaded "
"onto an unsupported host.",
"warning",
)
rows = []
if rows:
self.banner.show_status_message(
f" • Provisioning {len(rows)} declared ComfyUI model file(s) "
f"into {manager.models_path} (idempotent; present files skip)…",
"info",
)
provision = manager.provision_models(
rows, log=lambda m: self.banner.show_status_message(f" {m}", "info")
)
for warning in provision.warnings:
self.banner.show_status_message(f" {warning}", "warning")
if not provision.ok:
for failure in provision.failed:
self.banner.show_status_message(f" ✗ {failure}", "warning")
self.banner.show_status_message(
" • Some ComfyUI models could not be provisioned — the host "
"starts anyway; re-run `./start.sh comfyui-mps provision` "
"after fixing the issue.",
"warning",
)
# #905: custom nodes live inside the ComfyUI checkout's custom_nodes/ and
# are scanned only at process STARTUP (unlike models, which load lazily),
# so they must be on disk before the process launches below.
node_rows = _resolved_comfyui_custom_nodes(env)
if node_rows and not manager.preflight().ok:
self.banner.show_status_message(
" • Skipping custom-node provisioning — host preflight fails.",
"warning",
)
node_rows = []
if node_rows and not manager.repo_dir.exists():
# The checkout is cloned inside ensure_running_with_ownership on first
# run; clone it now so custom nodes have a home before provisioning.
try:
manager.install()
except ComfyUiMpsError as exc:
self.banner.show_status_message(
f" • Could not install ComfyUI for custom-node provisioning "
f"({exc}); nodes skipped — the start below retries the install.",
"warning",
)
node_rows = []
if node_rows:
already_running = manager.status().running
self.banner.show_status_message(
f" • Provisioning {len(node_rows)} ComfyUI custom node(s) into "
f"{manager.repo_dir}/custom_nodes (idempotent; present skip)…",
"info",
)
node_provision = manager.provision_custom_nodes(
node_rows, log=lambda m: self.banner.show_status_message(f" {m}", "info")
)
for warning in node_provision.warnings:
self.banner.show_status_message(f" {warning}", "warning")
if not node_provision.ok:
for failure in node_provision.failed:
self.banner.show_status_message(f" ✗ {failure}", "warning")
self.banner.show_status_message(
" • Some ComfyUI custom nodes could not be provisioned — the "
"host starts anyway; re-run `./start.sh comfyui-mps "
"provision-nodes` after fixing the issue.",
"warning",
)
elif node_provision.provisioned and already_running:
# Custom nodes load only at process startup; a process that was
# already running before this start does not rescan custom_nodes/.
self.banner.show_status_message(
" • Newly installed ComfyUI custom node(s) take effect only "
"after a ComfyUI (MPS) restart — run `./start.sh comfyui-mps "
"stop` then `./start.sh` (the nodes are cloned; a warm process "
"does not reload them).",
"warning",
)
try:
status, created = manager.ensure_running_with_ownership()
except ComfyUiMpsError as exc:
if self._leave_legacy_managed_host("ComfyUI (MPS)", exc):
return True
if exc.surviving_process:
self._managed_hosts_started_this_run.append(
("ComfyUI (MPS)", manager)
)
self.banner.show_status_message(
f"Managed ComfyUI (MPS) host could not start: {exc}", "error"
)
return False
if created:
self._managed_hosts_started_this_run.append(("ComfyUI (MPS)", manager))
health = manager.wait_healthy(timeout=60.0)
if not manager.confirm_started_process(status.pid):
self.banner.show_status_message(
f"Managed ComfyUI (MPS) process {status.pid} exited before "
"startup verification completed.",
"error",
)
return False
if health.get("reachable"):
self.banner.show_status_message(
f" • Managed ComfyUI (MPS) is up on port {status.port} "
f"(device={health.get('device')}, pid={status.pid})",
"info",
)
else:
# Not fatal: ComfyUI loads lazily and downstream containers retry.
self.banner.show_status_message(
f" • Managed ComfyUI (MPS) launched (pid={status.pid}, port "
f"{status.port}); still warming up — the first request loads the "
f"model. Logs: {status.log_file}",
"warning",
)
return True
def _finalize_managed_vllm_metal(self) -> bool:
"""Bring up the Atlas-managed vLLM Metal host (#379).
Docker Desktop on macOS can't pass Metal into a Linux container, so the
``managed-localhost`` source runs a native vLLM process on the HOST (via
the ``vllm-metal`` plugin) and LiteLLM reaches it as an OpenAI-compatible
upstream at ``host.docker.internal:<port>/v1``. When that source is
selected, preflight + install (idempotent — only the first run installs
the wheel + downloads weights) + start the host process here, before
``docker compose up``, so the endpoint is live when litellm-init renders
the model_list. A no-op for every other source, so CI (whose
``.env.example`` default is ``disabled``) never touches it.
Fatal on failure: the user explicitly chose this source, so surface a
broken host process loudly rather than boot a stack whose LiteLLM
upstream is dead.
"""
from services.vllm_metal_manager import VllmMetalError, manager_from_env
env = self.config_parser.parse_env_file()
if str(env.get("VLLM_METAL_SOURCE", "")).strip() != "managed-localhost":
return True
manager = manager_from_env(env)
self.banner.show_status_message(
" • VLLM_METAL_SOURCE=managed-localhost — preparing the native "
"Apple-Silicon/Metal vLLM host (first run installs the wheel and "
"downloads model weights; this can take several minutes)…",
"info",
)
try:
status, created = manager.ensure_running_with_ownership()
except VllmMetalError as exc:
if self._leave_legacy_managed_host("vLLM (Metal)", exc):
return True
if exc.surviving_process:
self._managed_hosts_started_this_run.append(
("vLLM (Metal)", manager)
)
self.banner.show_status_message(
f"Managed vLLM (Metal) host could not start: {exc}", "error"
)
return False
if created:
self._managed_hosts_started_this_run.append(("vLLM (Metal)", manager))
health = manager.wait_healthy(timeout=120.0)
if not manager.confirm_started_process(status.pid):
self.banner.show_status_message(
f"Managed vLLM (Metal) process {status.pid} exited before "
"startup verification completed.",
"error",
)
return False
if health.get("reachable"):
self.banner.show_status_message(
f" • Managed vLLM (Metal) is up on port {status.port} "
f"(pid={status.pid}); LiteLLM will register model "
f"'{env.get('VLLM_METAL_MODEL', '').strip()}'.",
"info",
)
else:
# Not fatal: vLLM's OpenAI server answers HTTP only after it has
# loaded the weights (downloading them on a first run), which can
# outlast the 120 s wait. litellm-init registers the model without
# probing it, so calls through LiteLLM fail to connect until the
# server answers.
self.banner.show_status_message(
f" • Managed vLLM (Metal) launched (pid={status.pid}, port "
f"{status.port}); still loading weights — calls through LiteLLM "
f"fail to connect until it answers. Logs: {status.log_file}",
"warning",
)
return True
def generate_litellm_configuration(self) -> bool:
"""Write a STUB volumes/litellm/config.yaml so the bind mount has
a file to attach to. The real model_list is rendered by
``litellm-init`` from the YAML catalogs + env on every ``docker compose up``.
``force=True`` here, but the writer is NOT unconditionally
destructive — ``LiteLLMConfigGenerator.write_config`` checks
for the litellm-init sentinel header + non-empty model_list
and preserves that file even with force=True. This protects
the previous run's real config across re-runs that haven't
yet completed a docker compose up. Stub / corrupt / missing
files DO get overwritten. See ``_is_litellm_init_managed``.
"""
try:
from utils.litellm_config_generator import LiteLLMConfigGenerator
generator = LiteLLMConfigGenerator(self.config_parser)
config_path = self.root_dir / "volumes/litellm/config.yaml"
# Pre-flight: ensure the bind-mount target directory is
# writable. Earlier docker compose runs (litellm-init runs
# as root inside the container) can leave the host directory
# root-owned, blocking subsequent container writes —
# symptom is ``PermissionError: '/litellm-config/config.yaml.tmp'``.
self._ensure_volume_dir_writable(config_path.parent)
generator.write_config(config_path, force=True)
return True
except Exception as e:
self.banner.show_status_message(f"Failed to generate LiteLLM configuration: {e}", "error")
return False
def generate_comfyui_manifest(self) -> bool:
"""Write ``volumes/comfyui/selected-models.yaml`` and
``volumes/comfyui/active-models.tsv`` so ``comfyui-init`` can
download the active model set without querying the DB.
Resolves the active set via ``comfyui_resolver.active_comfyui_models``
(C2 — DB-free, pure env + catalog computation). Both files are
written atomically; the TSV uses the same column order as the former
``psql SELECT`` so ``download_models.sh``'s existing loop is unchanged.
Skipped cleanly when ``COMFYUI_SOURCE == "disabled"``; that case
returns True (not an error) after ensuring ``volumes/comfyui/``
exists — the always-on backend bind-mounts the directory, and a
Docker-auto-created one would be root-owned on rootful Linux
daemons (#568).
Mirrors ``generate_litellm_configuration`` — see that method for the
general pattern.
"""
try:
from utils.comfyui_manifest_generator import ComfyUIManifestGenerator
env = self.config_parser.parse_env_file()
generator = ComfyUIManifestGenerator(env)
manifest_dir = self.root_dir / "volumes/comfyui"
if not generator.is_enabled():
# Disabled — nothing to write, but keep the directory
# present: the always-on backend bind-mounts it, and a
# Docker-auto-created host dir is root-owned on rootful
# Linux daemons (#568). Plain mkdir only — no permission
# repair, so the tracked marker files are never wiped.
manifest_dir.mkdir(parents=True, exist_ok=True)
return True
self._ensure_volume_dir_writable(manifest_dir)
generator.write(manifest_dir)
if getattr(generator, "unresolved", None):
self.banner.show_status_message(
"ComfyUI models in COMFYUI_USER_MODELS not resolvable from the "
"catalog, the custom sidecar or an earlier start, so not staged: "
+ ", ".join(generator.unresolved),
"warning",
)
return True
except Exception as e:
self.banner.show_status_message(
f"Failed to generate ComfyUI manifest: {e}", "error"
)
return False
def _ensure_volume_dir_writable(self, path: "Path") -> None:
"""Make sure a bootstrapper-managed host directory is writable
by both the current host user and any future container that
bind-mounts it. Earlier container runs can leave the directory
root-owned with 755 mode, blocking subsequent re-writes.
Strategy: if the directory exists and is not writable, add owner
rwx. Only the owner can chmod, and no container writes these
directories as a non-root user, so other-write is never added
(#1387). If chmod fails (another user owns it), preserve the
directory and report the ownership repair command.
Never raises — falls back to letting the original write fail with its
native error when ownership prevents the permission repair.
"""
if not path.exists():
path.mkdir(parents=True, exist_ok=True)
return
if not path.is_dir():
return # caller's problem; not our job to second-guess
if os.access(path, os.W_OK):
return # already writable, nothing to do
# Owner rwx is the cheapest, least destructive repair.
try:
path.chmod(path.stat().st_mode & 0o7777 | 0o700)
if os.access(path, os.W_OK):
self.banner.show_status_message(
f" • Relaxed permissions on {path} (was root-owned from a prior container run)",
"info",
)
return
except OSError:
pass
# Never delete a bind-mounted directory to recover permissions: it may
# contain user models, generated configuration, or application state.
self.banner.show_status_message(
f" • Could not make {path} writable; existing contents were preserved. "
f"Repair ownership with `sudo chown -R $(id -u):$(id -g) "
f"{shlex.quote(str(path))}` "
f"and re-run ./start.sh",
"warning",
)
def _derive_plugin_route_policies(self) -> tuple[list, list]:
"""Derive Kong auth and timeout policy from plugin.yml manifests.
Returns ``(route_auth, route_timeouts)`` from one discovery snapshot.
Auth entries are ``(route_prefix, mode)`` for non-inherit auth; timeout
entries are ``(plugin_name, route_prefix, explicit_timeout_mapping)``.
A malformed/conflicting *plugin* manifest is handled inside
``discover_plugin_manifests`` (collected as an error, its plugin dropped)
and never raises — the consumer doctor surfaces those. This deliberately
does NOT swallow other exceptions: an Atlas-internal failure (missing
schema, import error) must **fail closed** through the caller's handler,
which aborts Kong generation with a visible message, rather than silently
downgrading a ``key-auth`` prefix to the (possibly open) default
(#402 review M2).
"""
from core.plugin_manifest import (
derive_route_auth,
derive_route_timeouts,
discover_plugin_manifests,
)
plugin_dirs = _resolve_plugin_dirs(self)
if not plugin_dirs:
return [], []
discovery = discover_plugin_manifests(plugin_dirs)
return (
derive_route_auth(discovery.manifests),
derive_route_timeouts(discovery.manifests),
)
def generate_kong_configuration(self) -> bool:
"""Generate dynamic Kong configuration based on SOURCE values."""
try:
from utils.kong_config_generator import KongConfigGenerator
generator = KongConfigGenerator(self.config_parser)
generator.track_key = getattr(self, "active_track", None)
generator.overridden_services = getattr(
self, "active_track_overrides", frozenset()
)
# Derive auth and service-level timeout policy from one plugin.yml
# discovery snapshot. Empty for base Atlas (no plugins) preserves
# the historical backend route shape.
(
generator.plugin_route_auth,
generator.plugin_route_timeouts,
) = self._derive_plugin_route_policies()
# Pre-flight: same root-owned-from-prior-container guard as
# the litellm bind-mount uses (kong-api-gateway also writes
# nothing into volumes/api but the bootstrapper drops the
# dynamic config there for the container to read).
self._ensure_volume_dir_writable(self.root_dir / "volumes/api")
kong_config = generator.generate_kong_config()
errors = generator.validate_config(kong_config)
if errors:
self.banner.show_status_message("Kong configuration validation failed:", "error")
for error in errors:
self.banner.console.print(f" • {error}")
return False
config_path = self.root_dir / "volumes/api/kong-dynamic.yml"
if not generator.write_config(kong_config, config_path):
return False
return True
except Exception as e:
self.banner.show_status_message(f"Failed to generate Kong configuration: {e}", "error")
return False
def check_service_dependencies(self) -> bool:
"""Check and enforce service dependencies. Silent on success."""
dependencies_satisfied = self.dependency_manager.check_service_dependencies()
if not dependencies_satisfied:
violations = self.dependency_manager.get_dependency_violations()
self.banner.show_status_message("Service dependency violations found:", "warning")
for violation in violations:
self.banner.console.print(f" ⚠️ {violation['error_message']}")
disabled_services = self.dependency_manager.auto_resolve_dependency_violations()
if not disabled_services:
self.banner.show_status_message("Could not auto-resolve dependency violations", "error")
return False
for service in disabled_services:
self.banner.show_status_message(f"Auto-disabled {service} due to missing dependencies", "warning")
# RE-CHECK. A non-empty list only says SOMETHING was disabled — not
# that every violation is resolved. If four of five writes failed,
# startup proceeded with the rest unresolved; and because the
# original check ran against the pre-write env, a service requiring
# one just auto-disabled was never re-evaluated (no such 2-hop
# chain exists in today's graph, but nothing prevents one).
if self.dependency_manager.check_service_dependencies():
return True
self.banner.show_status_message(
"Dependency violations remain after auto-resolution", "error"
)
return False
return True
def handle_hosts_configuration(self, setup_hosts: bool, skip_hosts: bool) -> bool:
"""Handle hosts file configuration. Never fatal: missing entries only
break the friendly ``*.localhost`` URLs, not the stack.
skip: no check. setup: add missing entries (see below). default:
warn once when entries are missing, as the wizard option and
``--skip-hosts`` help promise.
"""
if skip_hosts:
return True
if setup_hosts:
# The wizard runs this inside the Textual app, which the
# non-root wrapper never elevates and where sudo cannot prompt.
# Nothing to write needs no privilege; a passwordless sudo is
# used; otherwise the missing entries are reported, not fatal.
if not self.hosts_manager.check_missing_hosts():
return True
if _run_privileged_hosts_setup(non_interactive=True):
return True
# print: the TUI's log pane captures stdout as well as its banner.
print(
"WARNING: hosts entries not added: sudo needs a password, which "
"the wizard cannot prompt for. Run ./start.sh --setup-hosts in a "
"terminal to add them."
)
return True
missing = self.hosts_manager.check_missing_hosts()
if missing:
print(
f"WARNING: {len(missing)} *.localhost hosts entries are missing; "
"friendly URLs need them. Run ./start.sh --setup-hosts to add "
"them, or --skip-hosts to silence this check."
)
return True
def perform_cold_start_cleanup(self, project_name: Optional[str] = None) -> bool:
"""Perform cold start cleanup if requested."""
self.banner.show_section_header("Cold Start Cleanup", "🧹")
self.banner.show_status_message("Performing cold start cleanup...", "info")
# Use the enhanced cold start cleanup. Forward the CLI project name so a
# `--cold --project foo` run tears down `foo` rather than the stale
# project still recorded in .env (the override is not persisted until
# setup_env_file runs later).
success = self.docker_manager.perform_cold_start_cleanup(project_name=project_name)
self.project_stopped_this_run = getattr(self, "project_stopped_this_run", False) or bool(success)
if not success:
self.banner.show_status_message("Cold cleanup failed; secrets were not rotated", "error")
else:
self.banner.show_status_message("Cold cleanup completed successfully", "success")
return success
def generate_encryption_keys(self, cold_start: bool = False) -> bool:
"""
Generate missing encryption keys for services.
Always generates missing keys; regenerates ALL keys on cold start
(``cold_start=True``).
Args:
cold_start: If True, regenerate all keys. If False, only generate missing ones.
Returns:
bool: True if successful
"""
force_regenerate = cold_start
try:
results = self.key_generator.generate_missing_keys(force_regenerate=force_regenerate)
if all(results.values()):
return True
else:
failed_keys = [key for key, success in results.items() if not success]
self.banner.show_status_message(
f"Failed to generate encryption keys: {', '.join(failed_keys)}",
"error"
)
return False
except Exception as e:
self.banner.show_status_message(f"Error generating encryption keys: {e}", "error")
return False
def validate_localhost_services(self) -> bool:
"""Validate localhost services are accessible before starting."""
# Check if any services are configured for localhost
if not self.localhost_validator.has_localhost_services():
return True # No localhost services to validate
self.banner.show_section_header("Validating Localhost Services", "🔍")
try:
results = self.localhost_validator.validate_all_localhost_services()
if not results:
return True # No localhost services found
# Display results
all_valid = True
# STT_PROVIDER_SOURCE / TTS_PROVIDER_SOURCE use the per_source
# SERVICE_CHECKS shape (service_name lives inside each variant, not
# at the top level), so index via the per_source-aware resolver —
# raw SERVICE_CHECKS[source_var]['service_name'] KeyErrors for them.
service_sources = self.localhost_validator.config_parser.parse_service_sources()
for source_var, (is_valid, messages) in results.items():
config = self.localhost_validator._resolve_source_config(
source_var, service_sources.get(source_var, "")
) or {}
service_name = config.get('service_name', source_var)
level = "info" if is_valid else "warning"
self.banner.show_status_message(f" • {service_name}:", level)
for message in messages:
self.banner.show_status_message(f" {message}", level)
if not is_valid:
all_valid = False
if all_valid:
self.banner.show_status_message("All localhost services are accessible", "success")
else:
self.banner.show_status_message(
"Some localhost services are not accessible (warnings above)",
"warning"
)
self.banner.show_status_message(
" • The stack will still start, but affected services may not work correctly",
"warning",
)
self.banner.show_status_message(
" • Please ensure localhost services are running as indicated",
"warning",
)
return True # Always continue, just show warnings
except Exception as e:
self.banner.show_status_message(f"Error validating localhost services: {e}", "error")
return True # Continue anyway
def start_docker_services(self, cold_start: bool = False, wait: bool = False) -> bool:
"""Start Docker services, building local images first when needed.
A cold start builds every enabled image without cache; a warm start
builds only when its local images are stale (#506). Either way, an
image that fails to build stops the launch only when another enabled
service depends on it or it belongs to the always-running core;
otherwise it is named and left out of ``up`` (#989, see
``isolate_failed_build``).
"""
self.banner.show_section_header("Starting Services", "🚀")
self.skipped_builds = []
self._readiness_snapshot = None
if cold_start:
self.banner.show_status_message("Starting containers with fresh build (cold start)...", "info")
result = self._cold_compose_up(wait)
else:
self.banner.show_status_message("Starting Atlas services...", "info")
result = self._warm_compose_up(wait)
if result is None:
self.banner.show_status_message("Failed to build some services", "error")
self.rollback_managed_host_processes()
return False
if result != 0:
# Known benign race (#508): `up -d --wait` can return nonzero when
# an enabled one-shot init exits 0 during the wait window (Compose
# reports e.g. "container …-n8n-init exited (0)"), even though the
# whole stack converged. Before failing, inspect the resolved
# state — if every expected service/replica is running/healthy or
# a declared one-shot exited 0, continue to the normal one-shot
# verification instead of false-failing automation consumers. Genuine failures (nonzero
# exits, unhealthy/non-running services) still fail, with names.
if not (wait and self._up_wait_race_converged()):
self.banner.show_status_message("Failed to start some services", "error")
self.rollback_managed_host_processes()
return False
if not self.verify_one_shot_init_containers():
self.rollback_managed_host_processes()
return False
if not self._reactivate_n8n_if_needed():
self.rollback_managed_host_processes()
return False
self.commit_managed_host_processes()
self._show_started_status()
return True
def _show_started_status(self) -> None:
"""Say all services started, or which ones the image build left out."""
if self.skipped_builds:
self.banner.show_status_message(
started_without(self.skipped_builds), "warning"
)
return
self.banner.show_status_message("All services started successfully", "success")
def _cold_compose_up(self, wait: bool):
"""Build every enabled image without cache, then ``up``.
Returns the ``up`` exit code, or None when a build stopped the launch.
"""
# Enabled-service target set from the rendered projection (#504) —
# a broken build for a disabled/out-of-track service must not
# abort the cold start. None (projection failure) falls back to
# the historical full-graph build/up.
targets = self.docker_manager.enabled_service_targets()
self.docker_manager.capture_build_state(targets)
# Build images without cache (matching original Bash script behavior)
print(" - Building images without cache...")
built, targets = self._build_launch_images(targets, no_cache=True)
if not built:
return None
# Cold start just built every enabled local image fresh — record the
# source commit so the next warm start (#506) doesn't rebuild them
# again until the source actually changes. When a failed image was
# left out, `targets` no longer matches the captured build state,
# so nothing is recorded and the next start rebuilds.
self.docker_manager.mark_source_built(targets)
print(" - Starting containers...")
return self._compose_up(targets, wait)
def _warm_compose_up(self, wait: bool):
"""``up`` the enabled services, building stale images first (#506).
Returns the ``up`` exit code, or None when a build stopped the launch.
Current images: one ``up`` through ``start_services``, as before.
Stale images with a known target set: an explicit build of the
targets, so a failed optional image can be left out (#989), then
``up`` without ``--build``; the build state is recorded only when
``up`` succeeded with every target. Without a target set,
``start_services`` keeps the historical full-graph ``up --build``.
"""
targets = self.docker_manager.enabled_service_targets()
if not (targets and self.docker_manager.prepare_build_args(False, targets)):
self._capture_readiness_targets(targets)
return self.docker_manager.start_services(
detached=True, wait=wait, services=targets
)
print(" - Building stale images...")
built, targets = self._build_launch_images(targets, no_cache=False)
if not built:
return None
print(" - Starting containers...")
result = self._compose_up(targets, wait)
if result == 0:
self.docker_manager.mark_source_built(targets)
return result
def _compose_up(self, targets, wait: bool) -> int:
self._capture_readiness_targets(targets)
# Always force-recreate, matching start_services.
up_args = ['up', '-d', '--force-recreate']
if wait:
up_args.extend(['--wait', '--wait-timeout', '900'])
if targets:
up_args.extend(targets)
return self.docker_manager.execute_compose_command(up_args)
def _build_launch_images(self, targets, no_cache: bool):
"""Build the targets' local images before ``up`` (#989).
Returns ``(built, targets)``: the build passed, or every failed image
was one the launch can go without, in which case those services are
named here and dropped from the returned targets.
"""
build_result = self.docker_manager.build_services(
no_cache=no_cache, pull=False, services=targets
)
if build_result == 0:
return True, targets
built, targets, skipped = self.isolate_failed_build(targets, no_cache)
for text, tone in build_skip_rows(skipped):
self.banner.show_status_message(
text, {"warn": "warning"}.get(tone, tone)
)
return built, targets
def isolate_failed_build(self, targets, no_cache: bool = True):
"""Find out which images failed after the whole-target build did (#989).
Returns ``(built, targets_to_start, skipped)``. The normal build is
untouched: this runs only after one ``docker compose build`` of every
enabled target has failed. The images another enabled service lists
in ``depends_on``, and those of the always-running core, are rebuilt
together first and still stop the launch if they fail. Each
remaining image is then rebuilt on its own;
one that fails again is left out of the targets and named in
``skipped`` (and ``self.skipped_builds``) so the rest of the stack
starts. Without a known target set, or with no image that could be
left out, the launch stops exactly as before, with no rebuild.
"""
self.skipped_builds = []
plan = self._local_build_plan(targets)
if plan is None:
return False, targets, ()
required, optional = plan
print(
" - Build failed; rebuilding to find which image failed "
"(images other services depend on and core images first, then "
"each other image on its own)..."
)
if required and self._build_images(required, no_cache) != 0:
print(
" - An image other enabled services depend on, or a core "
"image, did not build."
)
return False, targets, ()
skipped = tuple(
name
for group in optional
if self._build_images(group, no_cache) != 0
for name in group
)
self.skipped_builds = list(skipped)
return True, [name for name in targets if name not in skipped], skipped
def _build_images(self, services: list[str], no_cache: bool) -> int:
return self.docker_manager.build_services(
no_cache=no_cache, pull=False, services=services
)
def _local_build_plan(self, targets):
"""``(required, optional_groups)`` for ``isolate_failed_build``, or None.
None when the target set, the rendered configuration or the core
services are unknown, or when every local image is required, because
then nothing could be left out and the launch stops as before.
"""
if not targets:
return None
services = self._rendered_compose_services()
core = self._always_running_services()
if services is None or core is None:
return None
required, optional = _split_local_builds(services, targets, core)
return (required, optional) if optional else None
def _always_running_services(self) -> Optional[set[str]]:
"""Compose services of the always-running core, or None if unknown.
A manifest the topology marks ``locked`` has no source choice, so its
containers run in every configuration: Supabase, Kong, Redis, LiteLLM
and Backend. (``chatterbox`` and ``speaches`` are also locked, as TTS
engines activated through TTS_PROVIDER_SOURCE; neither builds a local
image, and a locked image is only ever kept required.)
"""
try:
from services.manifests import load_manifests
from services.topology import get_topology
root = self.root_dir / "services"
locked = {row.manifest for row in get_topology(root).rows if row.locked}
return {
str(container)
for manifest in load_manifests(root)
if manifest.name in locked
for container in manifest.containers
}
except Exception: # noqa: BLE001 — unknown means stop as before
return None
def _rendered_compose_services(self) -> Optional[dict]:
"""The resolved ``docker compose config`` services, or None."""
try:
cmd = self.docker_manager._build_compose_command(
['config', '--format', 'json']
)
result = run_with_deadline(
cmd, cwd=str(self.docker_manager.root_dir), timeout_seconds=60.0
)
if result.returncode != 0:
return None
return (json.loads(result.stdout) or {}).get("services") or None
except Exception: # noqa: BLE001 — unknown means stop as before
return None
def _capture_readiness_targets(self, targets) -> None:
"""Freeze the effective post-build launch model for recovery and status."""
self._readiness_snapshot = self._resolve_readiness_targets(targets)
def _resolve_readiness_targets(self, targets=None) -> tuple[dict, set, str | None, str | None]:
"""Resolve requested services and dependency companions from Compose."""
services = self._rendered_compose_services()
try:
replicas = service_replica_targets(services)
skipped = set(getattr(self, "skipped_builds", ()))
selected = set(targets) if targets else {
name for name, count in replicas.items() if count > 0
} - skipped
expected = _desired_compose_targets(services, replicas, selected, skipped)
project = (
self.docker_manager.project_name_override
or self.docker_manager.config_parser.get_project_name()
)
return expected, _one_shot_service_names(services), project, None
except ValueError as exc:
return {}, set(), None, f"could not determine desired Compose services: {exc}"
def _reactivate_n8n_if_needed(self) -> bool:
"""Restart n8n once after seeding so a consumer's production webhook
registers when the workflow was activated **without** an ``N8N_API_KEY``.
n8n CE registers a workflow's production webhook only when the n8n
*server* (re)starts. With a key, the seed activates over the public API
and the webhook registers immediately (no restart). Without a key, the
seed persists ``active=true`` via ``n8n publish:workflow`` — but the
running server ignores that until it restarts (empirically verified on
n8nio/n8n:2.28.2; the 2.36.7 image retains the same
``publish:workflow --id`` CLI boundary). So Atlas performs the one
restart the consumer would otherwise do by hand. No-op with a key, with
n8n disabled, or when no active consumer workflow is declared.
"""
try:
consumer = self.config_parser.load_consumer_config()
env = self.config_parser.parse_env_file()
except Exception as exc:
self.banner.show_status_message(
f"Could not determine whether n8n webhook reactivation is "
f"required: {exc}",
"error",
)
return False
if not _n8n_needs_reactivation_restart(env, consumer):
return True
self.banner.show_status_message(
"Restarting n8n to register consumer webhook(s) (no N8N_API_KEY)...",
"info",
)
try:
result = self.docker_manager.execute_compose_command(["restart", "n8n"])
except Exception as exc: # noqa: BLE001 — launch must convert to a verdict
self.banner.show_status_message(
f"Failed to restart n8n for webhook registration: {exc}", "error"
)
return False
if result != 0:
self.banner.show_status_message(
"Failed to restart n8n for consumer webhook registration; "
f"compose returned exit code {result}.",
"error",
)
return False
return self._n8n_restart_converged()
def _n8n_restart_converged(self) -> bool:
services, converged, _waited, error = self._poll_until_converged(
grace_seconds=120.0,
poll_interval_seconds=2.0,
poll_rows=self._strict_n8n_health_rows,
desired_scope={"n8n"},
)
if error is not None:
detail = error
elif not services:
detail = "n8n is absent from compose status"
elif not converged:
detail = "; ".join(
f"{entry.get('service', 'n8n')}: {entry.get('reason', 'not healthy')}"
for entry in services
if not entry.get("ok")
) or "n8n did not become healthy"
else:
return True
self.banner.show_status_message(
f"n8n restart did not converge to healthy: {detail}", "error"
)
return False
def _strict_n8n_health_rows(self) -> tuple[list[dict], str | None]:
rows, error = self.docker_manager.compose_service_ps_json("n8n")
return [self._strict_n8n_health_row(row) for row in rows], error
@staticmethod
def _strict_n8n_health_row(row: dict) -> dict:
state = str(row.get("State") or "").strip().lower()
health = str(row.get("Health") or "").strip().lower()
if state == "running" and health == "healthy":
return row
strict = dict(row)
if state == "running" and health in {"", "starting"}:
strict["Health"] = "starting"
elif state != "running":
strict["ExitCode"] = "1"
strict["Status"] = "n8n is not running"
return strict
def _poll_until_converged(
self,
*,
grace_seconds: float = 120.0,
poll_interval_seconds: float = 2.0,
poll_rows=None,
sleep=None,
monotonic=None,
desired_scope: set[str] | None = None,
) -> tuple[list[dict], bool, bool, str | None]:
"""Inspect ``compose ps`` and re-poll convergent-pending rows within a
bounded grace window (#677/#681).
A ``state=running, health=starting`` row is *convergent-pending*, not a
failure — its healthcheck is simply still in its start period. This
re-polls while any row is pending, up to ``grace_seconds``, so a stack
that is merely mid-probe at the first snapshot is given time to settle
before being classified.
Returns ``(services, converged, waited, error)``:
- ``services`` — classified target rows, including missing replicas.
- ``converged`` — True iff every desired service/replica is observed
and ok, with at least one target and nothing still pending.
- ``waited`` — True iff a pending row was observed and re-polled, so a
caller can flag "converged after grace".
- ``error`` — a ``compose ps`` inspection error, or None.
A genuine failure (``not ok and not pending``) short-circuits
immediately — there is no point waiting on a doomed start. Dependencies
(``poll_rows`` / ``sleep`` / ``monotonic``) are injectable so the
grace loop is unit-testable with fake ps snapshots and a fake clock.
"""
import time as _time
poll_rows = poll_rows or self.docker_manager.compose_ps_json
sleep = sleep or _time.sleep
monotonic = monotonic or _time.monotonic
waited = False
snapshot = getattr(self, "_readiness_snapshot", None)
expected, finite_services, project, error = (
snapshot if snapshot is not None else self._resolve_readiness_targets()
)
if error is not None:
return [], False, waited, error
if desired_scope is not None:
if not desired_scope <= expected.keys():
return [], False, waited, "requested status scope is absent from desired Compose services"
expected = {name: count for name, count in expected.items() if name in desired_scope}
# Config inspection has its own deadline; it must not spend health grace.
deadline = monotonic() + grace_seconds
while True:
rows, error = poll_rows()
if error is not None:
return [], False, waited, error
try:
name_ids = validate_compose_status_rows(rows, project)
except ValueError as exc:
return [], False, waited, f"invalid docker compose ps inventory: {exc}"
services = []
identities = {name: set() for name in expected}
for row in rows:
name = row["Service"]
if name not in expected:
continue
identity = compose_container_identity(row, name_ids)
identities[name].add(identity)
services.append(self._compose_row_status(row, name in finite_services))
for name, count in expected.items():
observed = len(identities[name])
if observed < count:
services.append({
"service": name, "state": "missing", "health": None,
"status": None, "exit_code": None, "ok": False, "pending": False,
"reason": f"missing containers: expected {count}, observed {observed} (missing {count - observed})",
"expected_replicas": count, "observed_replicas": observed,
"missing_replicas": count - observed,
})
genuine_failure = any(
not entry["ok"] and not entry["pending"] for entry in services
)
pending = [entry for entry in services if entry["pending"]]
# Classify now on a genuine failure (don't wait on a doomed start),
# an empty stack, or once nothing is still pending.
if genuine_failure or not services or not pending:
converged = bool(services) and all(
entry["ok"] for entry in services
)
return services, converged, waited, None
# Only ``starting`` rows remain and nothing has failed — grace-wait.
if monotonic() >= deadline:
# Grace exhausted; still-pending rows now count as failures.
return services, False, waited, None
waited = True
sleep(min(poll_interval_seconds, max(0.0, deadline - monotonic())))
def _up_wait_race_converged(
self,
*,
grace_seconds: float = 120.0,
poll_interval_seconds: float = 2.0,
poll_rows=None,
sleep=None,
monotonic=None,
) -> bool:
"""Classify a nonzero ``up -d --wait`` as a benign, converged start.
Two benign races are covered: the #508 one-shot-init race (a one-shot
exits 0 during the wait window) and the #677/#681 long-lived-healthcheck
race (a service is still ``health=starting`` at the snapshot). Returns
True only when every desired service/replica is accounted for and ok
per ``_compose_row_status`` — after re-polling any still-starting
rows within ``grace_seconds``. Any other outcome — inspection error,
empty stack, unhealthy/non-running service, or a row still starting
after the grace window — returns False and names the offenders, so a
genuine failure keeps failing loudly. Injectable deps mirror
``_poll_until_converged`` for testing.
"""
services, converged, waited, error = self._poll_until_converged(
grace_seconds=grace_seconds,
poll_interval_seconds=poll_interval_seconds,
poll_rows=poll_rows,
sleep=sleep,
monotonic=monotonic,
)
if error is not None:
self.banner.show_status_message(
f"Could not inspect services after nonzero `up --wait`: {error}",
"error",
)
return False
if not services:
return False
if not converged:
for entry in services:
if entry["ok"]:
continue
exit_note = (
f", exit code {entry['exit_code']}" if entry["exit_code"] else ""
)
reason = entry["reason"]
if entry["pending"]:
# Still starting after the full grace window — a genuine
# timeout now, not a benign mid-probe.
reason = f"{reason} (still not healthy after grace window)"
self.banner.show_status_message(
f"{entry['service']}: {reason}{exit_note}",
"error",
)
return False
self._up_converged_after_grace = waited
detail = (
"services converged within the health-check grace window"
if waited
else "known successful one-shot init race"
)
self.banner.show_status_message(
"Compose `up --wait` returned nonzero, but every service is "
f"running/healthy and every one-shot exited 0 — continuing ({detail}).",
"warning",
)
return True
@staticmethod
def _compose_row_status(row: dict, one_shot: bool = False) -> dict:
service = str(row.get("Service") or row.get("Name") or "unknown")
state = str(row.get("State") or "").strip().lower()
health = str(row.get("Health") or "").strip().lower()
status = str(row.get("Status") or "").strip()
exit_code = str(row.get("ExitCode") if row.get("ExitCode") is not None else "").strip()
ok = False
pending = False
reason = status or state or health or "unknown"
if state == "running":
# A running container's exit code is meaningless (Compose reports 0
# for a live container); never surface it as a failure signal — it
# is the source of the misleading "starting, exit code 0" line.
exit_code = ""
if health == "starting":
# Convergent-pending, NOT failed: the healthcheck is still in
# its start period. Callers re-poll pending rows within a grace
# window before classifying (#677/#681).
pending = True
reason = "starting"
else:
ok = health in {"", "healthy"}
if health and health != "healthy":
reason = health
elif state == "exited":
ok = one_shot and _one_shot_row_verdict(row)[0] == "done"
if not one_shot:
reason = f"{reason}; successful one-shot lifecycle not declared"
elif state in {"created", "restarting", "paused", "dead"}:
ok = False
return {
"service": service,
"state": state or None,
"health": health or None,
"status": status or None,
"exit_code": exit_code or None,
"ok": ok,
"pending": pending,
"reason": reason,
}
def show_detached_status_summary(
self,
*,
json_output: bool = False,
grace_seconds: float = 120.0,
poll_interval_seconds: float = 2.0,
poll_rows=None,
sleep=None,
monotonic=None,
) -> bool:
"""Print final compose status for automation-friendly detached starts.
Grace-aware (#677/#681): a service still ``health=starting`` at the
summary moment is re-polled within a bounded window before classifying,
so the final summary never false-fails on a mid-probe healthcheck. The
``--json`` payload carries ``converged_after_grace`` so automation can
tell a health race apart from a first-pass-healthy start.
"""
services, converged, waited, error = self._poll_until_converged(
grace_seconds=grace_seconds,
poll_interval_seconds=poll_interval_seconds,
poll_rows=poll_rows,
sleep=sleep,
monotonic=monotonic,
)
if error is not None:
payload = {
"ok": False,
"error": error,
"services": [],
"converged_after_grace": False,
"not_started": list(self.skipped_builds),
}
if json_output:
print(json.dumps(payload, indent=2, sort_keys=True))
else:
self.banner.show_status_message(f"Could not inspect services: {error}", "error")
return False
ok = converged
# The race happened if this summary had to wait, or the earlier
# `up --wait` reclassification did.
converged_after_grace = bool(self._up_converged_after_grace or waited)
payload = {
"ok": ok,
"services": services,
"converged_after_grace": converged_after_grace,
# Images that failed to build were left out of `up` (#989); the
# stack still counts as started, but automation must see the gap.
"not_started": list(self.skipped_builds),
}
if json_output:
print(json.dumps(payload, indent=2, sort_keys=True))
return ok
self.banner.show_section_header("Detached Status", "📊")
if not services:
self.banner.show_status_message("No compose services were reported", "error")
return False
for entry in services:
level = "success" if entry["ok"] else "error"
health = f" / {entry['health']}" if entry["health"] else ""
self.banner.show_status_message(
f"{entry['service']}: {entry['state'] or 'unknown'}{health} — {entry['reason']}",
level,
)
return ok
def verify_one_shot_init_containers(self, on_line=None) -> bool:
"""Fail startup if enabled post-start one-shot init containers failed.
The TUI's cancel ends the up-to-900 s wait early through
``docker_manager.should_stop`` (#1357)."""
env_vars = self.config_parser.parse_env_file()
services: list[str] = []
if env_vars.get("N8N_INIT_SCALE", "0") != "0":
services.append("n8n-init")
if env_vars.get("OPEN_WEB_UI_INIT_SCALE", "0") != "0":
services.append("open-webui-init")
if env_vars.get("COMFYUI_INIT_SCALE", "0") != "0":
services.append("comfyui-init")
if env_vars.get("REDPANDA_INIT_SCALE", "0") != "0":
services.append("redpanda-init")
if env_vars.get("ZEPPELIN_INIT_SCALE", "0") != "0":
services.append("zeppelin-init")
try:
consumer_config = self.config_parser.load_consumer_config()
except Exception as exc:
message = (
"Could not inspect consumer one-shot services "
f"(error_type={type(exc).__name__})"
)
if on_line is None:
self.banner.show_status_message(message, "error")
else:
on_line(message, "error")
return False
if _n8n_seed_expected(consumer_config, env_vars):
services.append("n8n-seed")
# An image build that left a failed optional image out never started
# it, so there is no one-shot run to wait for (#989).
services = [name for name in services if name not in self.skipped_builds]
if not services:
return True
failures = self.docker_manager.failed_one_shot_services(
services,
timeout_seconds=900.0,
poll_interval_seconds=5.0,
)
if not failures:
return True
for service, reason in failures:
msg = f"{service} failed after compose up ({reason})"
if reason != "cancelled" and not (
self.docker_manager.should_stop is not None and self.docker_manager.should_stop()
):
# The exit status alone does not say why (#1357).
tail = self.docker_manager.one_shot_log_tail(service)
if tail:
msg += "\n last log lines:\n" + "\n".join(f" {line}" for line in tail)
if on_line is None:
self.banner.show_status_message(msg, "error")
else:
on_line(f"❌ {msg}", "error")
return False
def show_pre_launch_summary(
self,
*,
track: str | None = None,
assume_yes: bool = False,
) -> bool:
"""
Display the combined configuration summary table with access URLs
and hosted endpoints, then prompt for confirmation.
``track`` — forwarded to ``build_pre_launch_summary_table`` so the
``Track: <display_name>`` banner line is emitted when a track is active.
Returns:
bool: True if user confirms, False to cancel.
"""
table = self.build_pre_launch_summary_table(track=track)
self.banner.console.print(table)
self.banner.console.print()
from rich.text import Text as _Text
from ui.textual.palette import style_for_category as _style_for_category
from services.topology import CATEGORY_LABELS, CATEGORY_ORDER
_legend = _Text()
_first = True
for _slug in CATEGORY_ORDER:
if not _first:
_legend.append(" ")
_first = False
_legend.append("▰", style=_style_for_category(_slug))
_legend.append(f" {CATEGORY_LABELS[_slug]}")
self.banner.console.print(_legend)
self.banner.console.print()
# Confirmation prompt — legacy linear flow only. TUI mode runs the
# launch confirmation as the wizard's last step; this branch is
# reached only when --no-tui or non-TTY.
if assume_yes:
return True
if sys.stdin.isatty():
response = self.banner.console.input(
" [color(245)]Launch the stack? (Y/n):[/color(245)] "
).strip().lower()
return response in ('', 'y', 'yes')
return True # non-TTY: auto-confirm
def build_pre_launch_summary_table(self, *, track: str | None = None):
"""
Build the configuration summary as a Rich Table renderable —
used by the --no-tui / non-TTY linear flow (`show_pre_launch_summary`).
The Textual wizard renders its own info-box and never reaches this
table.
``track`` — the active track key (e.g. ``"gen-ai-rag"``), or None
when no track was selected. When set, a ``Track: <display_name>``
line is prepended above the services table per spec §5.2 #7.
"""
from rich.table import Table
from rich.text import Text
from rich.box import HEAVY_HEAD
from wizard.model.state_builder import all_services, all_cloud_apis, alias_for, cloud_api_status_text, kong_listener_port
from services.topology import get_topology
from ui.textual.palette import style_for_category
_topology = get_topology()
_category_by_name = {r.display_name: r.category for r in _topology.rows}
env_vars = self.config_parser.parse_env_file()
service_sources = self.config_parser.parse_service_sources()
kong_port = kong_listener_port(env_vars)
# Check if hosts entries are configured (yields the set of hostnames
# that are PRESENT in /etc/hosts).
hosts_present = set()
try:
existing_missing = self.hosts_manager.check_missing_hosts()
all_hosts = self.hosts_manager.get_atlas_hosts()
hosts_present = set(all_hosts) - set(existing_missing)
except Exception:
pass
table = Table(
title="Stack Services Overview",
title_style="bold bright_white",
box=HEAVY_HEAD,
border_style="color(240)",
header_style="bold bright_white",
show_lines=True,
padding=(0, 1),
expand=True,
)
table.add_column("PORT", style="color(248)", justify="left", ratio=1, no_wrap=True)
table.add_column("SERVICE", style="color(252)", justify="left", ratio=3, no_wrap=True)
# Category marker — between SERVICE and SOURCE, mirroring the
# TUI box layout so both surfaces speak the same visual language.
table.add_column("", justify="left", width=2, no_wrap=True)
table.add_column("SOURCE", justify="left", ratio=3, no_wrap=True)
table.add_column("ALIAS", justify="left", ratio=4, no_wrap=True)
table.add_column("STATUS", justify="left", ratio=2, no_wrap=True)
# Service definitions come from state_builder.all_services() — single
# source of truth shared with the TUI info-box (no more inline list
# to drift out of sync).
services = list(all_services())
# Sort by port number ascending; services with no port go to the end.
def _sort_key(svc):
name, source_var, port_var, _scale_var = svc
source = service_sources.get(source_var, env_vars.get(source_var, 'container'))
if source == 'disabled' or not port_var:
return (2, 99999)
if 'localhost' in source:
lp = self._get_localhost_port(name, env_vars)
match = re.search(r':(\d+)', lp)
return (1, int(match.group(1)) if match else 99999)
try:
return (0, int(env_vars.get(port_var, '99999')))
except ValueError:
return (2, 99999)
services.sort(key=_sort_key)
from ui.textual.palette import style_for_source_choice as _style_for_source
# Collected `(name, port_val)` for post-loop collision detection.
# Disabled / portless rows still flow through here as ("-",); the
# detector filters them out.
collision_rows: list[tuple[str, str]] = []
for name, source_var, port_var, scale_var in services:
source = service_sources.get(source_var, env_vars.get(source_var, 'container'))
scale = env_vars.get(scale_var, '0') if scale_var else '1'
status_text, status_style = self._get_service_status(source, scale)
# Color the SOURCE cell with the same helper the TUI uses:
# container → green, localhost / external / api → blue,
# disabled → muted grey. Previously hardcoded grey, which
# made localhost variants visually indistinguishable from
# containerised ones.
source_style = _style_for_source(source)
# PORT column
if source == 'disabled':
port_val = "-"
elif 'localhost' in source:
port_val = self._get_localhost_port(name, env_vars)
elif port_var:
port_val = _published_port_label(port_var, env_vars)
else:
port_val = "-"
# ALIAS column — alias map from state_builder (single source).
hostname = alias_for(name)
if hostname and hostname in hosts_present and source_has_endpoint(source):
alias_text = Text(f"{hostname}:{kong_port}", style="color(75)")
else:
alias_text = Text("-", style="color(243)")
category = _category_by_name.get(name, "")
bar = Text("▰", style=style_for_category(category))
table.add_row(
port_val,
name,
bar,
Text(source, style=source_style),
alias_text,
Text(status_text, style=status_style),
)
collision_rows.append((name, port_val))
# Cloud APIs panel — renders below the services table. Cloud
# providers don't run as containers (scale: 0) so they don't
# belong as rows in the services grid; this keeps them visible
# without misleading the user about what's getting started.
from rich.console import Group
from rich.panel import Panel
cloud_lines = []
for name, source_var, api_key_var in all_cloud_apis():
source = (
service_sources.get(source_var, env_vars.get(source_var, 'disabled'))
or ''
).strip().lower()
key_set = bool((env_vars.get(api_key_var, '') or '').strip())
enabled = source == 'enabled'
line = Text()
line.append(f" {name:<11}", style="bright_white")
# Status string is shared with the Textual CloudApisRow via
# state_builder.cloud_api_status_text — only the Rich style
# is local to this renderer.
if enabled and key_set:
style = "bright_green"
elif enabled and not key_set:
style = "bright_yellow"
else:
style = "color(243)"
line.append(cloud_api_status_text(enabled, key_set), style=style)
cloud_lines.append(line)
cloud_panel = Panel(
Group(*cloud_lines) if cloud_lines else Text("(none)", style="color(243)"),
title="[bold bright_white]Cloud APIs[/bold bright_white] "
"[color(243)](LiteLLM-routed, no containers)[/color(243)]",
border_style="color(240)",
padding=(0, 1),
expand=True,
)
consumer_config = self.config_parser.load_consumer_config()
consumer_panel = None
if consumer_config.consumers:
consumer_lines = []
for consumer in consumer_config.consumers:
line = Text()
line.append(f" {consumer.name:<16}", style="bright_white")
details = []
if consumer.compose_overlays:
details.append(f"{len(consumer.compose_overlays)} overlay")
if consumer.backend_plugins:
details.append(f"{len(consumer.backend_plugins)} plugin dir")
if consumer.comfyui_sidecars:
details.append(f"{len(consumer.comfyui_sidecars)} ComfyUI sidecar")
if consumer.ollama_models:
details.append(f"{len(consumer.ollama_models)} Ollama model")
line.append(", ".join(details) if details else "registered", style="color(75)")
consumer_lines.append(line)
consumer_panel = Panel(
Group(*consumer_lines),
title="[bold bright_white]Consumers[/bold bright_white] "
"[color(243)](atlas.consumer.yml)[/color(243)]",
border_style="color(240)",
padding=(0, 1),
expand=True,
)
# Track banner line (spec §5.2 #7): when a track was active,
# prepend a "Track: <display_name>" line above the services table.
track_line: list = []
if track:
_track_label = track
try:
from tracks import load_tracks as _lt_sum
_reg_sum = _lt_sum()
_t_sum = _reg_sum.by_key.get(track)
if _t_sum:
_track_label = _t_sum.display_name
except Exception: # noqa: BLE001
pass
track_line = [Text.from_markup(
f"[bold bright_white]Track:[/bold bright_white] "
f"[color(75)]{_track_label}[/color(75)]"
)]
# Port-collision warnings — informational only (warn-don't-block).
# When two rows resolve to the same host port (e.g. the user
# picked ollama-localhost on Kong's port), surface that here so
# the user can step back and adjust before Docker barfs with an
# opaque "address already in use" error.
warning_lines = _detect_port_collisions(collision_rows)
panels = [table, cloud_panel]
if consumer_panel is not None:
panels.append(consumer_panel)
if warning_lines:
warning_texts = [
Text.from_markup(f"[yellow]{msg}[/yellow]")
for msg in warning_lines
]
return Group(*track_line, *panels, *warning_texts)
return Group(*track_line, *panels)
@staticmethod
def _get_localhost_port(service_name: str, env_vars: dict) -> str:
"""Extract the localhost port for the pre-launch summary. Prefers the
service's endpoint env var (which encodes the active source variant's
port — e.g. ComfyUI's MPS port), then falls back to the dedicated
localhost_port_var for services that declare only a port number and no
endpoint URL (e.g. OpenClaw, Neo4j). Without the fallback those rows
rendered '-' even when a localhost port was configured."""
from services.topology import get_topology
_topology = get_topology()
row = None
for r in _topology.rows:
if r.display_name == service_name:
row = r
break
if row is None:
return "-"
# ONE definition, shared with the Textual ServiceTable. The local copy
# parsed the endpoint with a bare `:(\d+)`, which cannot match the
# stored literal `http://host:${COMFYUI_MPS_LOCALHOST_PORT:-8188}` —
# the `:` is followed by `$` — so it fell through to the wrong var and
# reported ComfyUI-MPS on 8000 instead of 8188, then raised a phantom
# collision against vLLM Metal.
from wizard.model.state_builder import resolve_localhost_port
port = resolve_localhost_port(row, env_vars)
return f":{port}" if port else "-"
@staticmethod
def _get_service_status(source: str, scale: str) -> tuple:
"""Get a status label with ● indicator and style for a service."""
if not source_has_endpoint(source):
return "● off", "color(245)"
if 'localhost' in source:
return "● local", "bright_cyan"
if 'external' in source:
return "● external", "bright_yellow"
if source == 'api':
return "● API", "bright_yellow"
if 'gpu' in source:
return "● GPU", "bright_green"
if scale == '0':
return "● off", "color(245)"
return "● on", "bright_green"
def check_comfyui_models(self, on_line=None):
"""Check ComfyUI local models.
Returns the probe result: ``True``/``False`` for a localhost ComfyUI,
or a ``ProbeSkipped`` naming why the check does not apply, so neither
front end reports a check that never ran as a pass (#1032).
"""
found = self.service_config.check_comfyui_local_models(on_line=on_line)
if found is not None:
return found
source = self.service_config.service_sources.get("COMFYUI_SOURCE", "container-cpu")
return ProbeSkipped(
f"applies only to COMFYUI_SOURCE=localhost; this run uses {source}"
)
def show_container_status_and_verify_ports(self, on_line=None):
"""
Show container status and verify actual vs expected ports.
Replicates the verification logic from original start.sh.
When `on_line` is provided (TUI mode), the redundant `docker compose ps`
text dump is dropped and per-service results route through `on_line`
with a level keyword ("ok"/"warn"/"error"). When `on_line` is None
(legacy mode), the printed output is unchanged from the original
implementation. Both branches return the number of ports that could
not be determined or did not match, which the launch result reports
as a failed check rather than a pass (#1032).
"""
# Get expected ports from .env (used by both branches)
env_vars = self.config_parser.parse_env_file()
def include_port_check(
source_var: str | None = None,
scale_var: str | None = None,
) -> bool:
if source_var:
source = env_vars.get(source_var, "").strip()
if source == "disabled" or "localhost" in source:
return False
if scale_var and env_vars.get(scale_var, "1").strip() == "0":
return False
return True
# Service definitions matching original Bash script
services_to_check = [
("supabase-db", "5432", env_vars.get("SUPABASE_DB_PORT", ""), None, None),
("redis", "6379", env_vars.get("REDIS_PORT", ""), None, None),
("supabase-storage", "5000", env_vars.get("SUPABASE_STORAGE_PORT", ""), "SUPABASE_STORAGE_SOURCE", "SUPABASE_STORAGE_SCALE"),
# supabase-auth is not published on the host (2026-10-08 run, cycle 56).
("supabase-api", "3000", env_vars.get("SUPABASE_API_PORT", ""), "SUPABASE_API_SOURCE", "SUPABASE_API_SCALE"),
("supabase-realtime", "4000", env_vars.get("SUPABASE_REALTIME_PORT", ""), "SUPABASE_REALTIME_SOURCE", "SUPABASE_REALTIME_SCALE"),
("neo4j-graph-db", "7687", env_vars.get("GRAPH_DB_PORT", ""), "NEO4J_GRAPH_DB_SOURCE", "NEO4J_SCALE"),
("weaviate", "8080", env_vars.get("WEAVIATE_PORT", ""), "WEAVIATE_SOURCE", "WEAVIATE_SCALE"),
("local-deep-researcher", "2024", env_vars.get("LOCAL_DEEP_RESEARCHER_PORT", ""), "LOCAL_DEEP_RESEARCHER_SOURCE", "LOCAL_DEEP_RESEARCHER_SCALE"),
("open-web-ui", "8080", env_vars.get("OPEN_WEB_UI_PORT", ""), "OPEN_WEB_UI_SOURCE", "OPEN_WEB_UI_SCALE"),
("backend", "8000", env_vars.get("BACKEND_PORT", ""), "BACKEND_SOURCE", "BACKEND_SCALE"),
("kong-api-gateway", "8000", env_vars.get("KONG_HTTP_PORT", ""), None, None),
("kong-api-gateway", "8443", env_vars.get("KONG_HTTPS_PORT", ""), None, None),
("n8n", "5678", env_vars.get("N8N_PORT", ""), "N8N_SOURCE", "N8N_SCALE"),
("searxng", "8080", env_vars.get("SEARXNG_PORT", ""), "SEARXNG_SOURCE", "SEARXNG_SCALE"),
("crawl4ai", "11235", env_vars.get("CRAWL4AI_PORT", ""), "CRAWL4AI_SOURCE", "CRAWL4AI_SCALE"),
("flower", "5555", env_vars.get("FLOWER_PORT", ""), "CELERY_SOURCE", "FLOWER_SCALE"),
("jupyterhub", "8888", env_vars.get("JUPYTERHUB_PORT", ""), "JUPYTERHUB_SOURCE", "JUPYTERHUB_SCALE"),
]
services_to_check = [
(service_name, internal_port, expected_port)
for service_name, internal_port, expected_port, source_var, scale_var in services_to_check
if include_port_check(source_var, scale_var)
]
# LiteLLM is the always-on LLM front door — its host port is the
# canonical LLM-stack port now.
services_to_check.append(("litellm", "4000", env_vars.get("LITELLM_PORT", "")))
# Add conditional services based on their scales
ollama_scale = env_vars.get("OLLAMA_SCALE", "0")
if ollama_scale != "0":
# Ollama upstream is internal-only; no host port mapping to verify.
pass
comfyui_scale = env_vars.get("COMFYUI_SCALE", "0")
if comfyui_scale != "0":
services_to_check.append(("comfyui", "18188", env_vars.get("COMFYUI_PORT", "")))
if on_line is None:
# Legacy linear flow — preserve today's exact behavior including
# the `docker compose ps` text dump.
print()
self.docker_manager.show_container_status()
print()
print("🔍 Checking if Docker assigned the expected ports...")
# Counted the same way as the TUI branch below, so both front
# ends classify the same mismatches as the same outcome (#1032).
mismatches = 0
for service_name, internal_port, expected_port in services_to_check:
if not expected_port:
continue
actual_port = self.docker_manager.get_service_port(service_name, internal_port)
if not actual_port:
print(f" • ❌ {service_name}: Could not determine port mapping")
mismatches += 1
elif actual_port == expected_port:
print(f" • ✅ {service_name}: Using expected port {expected_port}")
else:
print(f" • ⚠️ {service_name}: Expected port {expected_port} but got {actual_port}")
mismatches += 1
return mismatches
# TUI mode — route per-service lines through on_line, skip the ps dump.
# The dots in the anchored box already convey "is the container up";
# this verification is specifically about port-mapping correctness.
mismatches = 0
for service_name, internal_port, expected_port in services_to_check:
if not expected_port:
continue
actual_port = self.docker_manager.get_service_port(service_name, internal_port)
if not actual_port:
on_line(f"❌ {service_name}: could not determine port mapping", "error")
mismatches += 1
elif actual_port == expected_port:
on_line(f"✅ {service_name}: port {expected_port} ok", "ok")
else:
on_line(f"⚠️ {service_name}: expected :{expected_port}, got :{actual_port}", "warn")
mismatches += 1
return mismatches
def show_container_logs(self) -> int:
"""
Show container logs with follow option.
Replicates the logs display from original start.sh.
"""
try:
return self.docker_manager.show_container_logs(follow=True)
except KeyboardInterrupt:
print("\n🔄 Log viewing interrupted by user")
print(" Use 'docker compose logs -f' to view logs again")
# Stopping the log stream is a detach, not a teardown (#1032).
print(f" Detached — {DETACH.consequences}; ./stop.sh stops them")
return 130
def _parse_env_values(path: Path) -> dict[str, str]:
if not path.exists():
return {}
values = _backfill_default_values(_read_env_backfill_text(path), str(path))
return {key: value for key, (value, _encoded) in values.items()}
def _env_example_section_by_key(path: Path) -> dict[str, str]:
if not path.exists():
return {}
bar_re = re.compile(r"^#\s*[=─]{3,}\s*$")
lines = env_lines(_read_env_backfill_text(path).removeprefix("\ufeff"))
current_section = "(unsectioned)"
by_key: dict[str, str] = {}
i = 0
while i < len(lines):
line = lines[i]
if (
bar_re.match(line)
and i + 2 < len(lines)
and lines[i + 1].startswith("#")
and bar_re.match(lines[i + 2])
):
current_section = lines[i + 1].lstrip("#").strip() or "(unnamed)"
i += 3
continue
stripped = line.strip()
if stripped and not stripped.startswith("#") and "=" in stripped:
key, _decoded, _encoded, _exported = _backfill_env_assignment(stripped, str(path))
by_key[key] = current_section
i += 1
return by_key
def _print_env_backfill_summary(
*,
env_path: Path,
env_example_path: Path,
before: dict[str, str],
after: dict[str, str],
) -> None:
added = sorted(key for key in after if key not in before)
filled = sorted(
key
for key, prior in before.items()
if key in after and not prior and bool(after[key])
)
# Backfill changes a non-blank value only to refresh a retired build pin.
refreshed = sorted(key for key, prior in before.items() if prior and after.get(key, prior) != prior)
if not added and not filled and not refreshed:
click.echo(f"No env changes needed: {env_path}")
return
section_by_key = _env_example_section_by_key(env_example_path)
grouped: dict[str, list[str]] = {}
for keys, label in ((added, "added"), (filled, "filled blank"),
(refreshed, "refreshed retired default")):
for key in keys:
section = section_by_key.get(key, "(unsectioned)")
grouped.setdefault(section, []).append(f"{key} ({label})")
click.echo(f"Env backfill updated {env_path}:")
for section in sorted(grouped):
click.echo(f"{section}:")
for item in grouped[section]:
click.echo(f" - {item}")
def _compose_validation_summary(output: str) -> str | None:
variable: str | None = None
service: str | None = None
variable_match = re.search(
r'["\\]*([A-Z0-9_]+)["\\]* variable is not set',
output,
)
if variable_match:
variable = variable_match.group(1)
service_match = re.search(
r'service ["\\]*([^"\\]+)["\\]* has neither an image nor a build context',
output,
)
if service_match:
service = service_match.group(1)
if variable and service:
return f"Service {service} is missing compose variable {variable}."
if variable:
return f"Compose variable {variable} is not set."
if service:
return f"Service {service} has neither an image nor a build context."
return None
def _n8n_workflow_effective_active(wf) -> bool:
"""Mirror seed-workflows.js ``effectiveActive``: ``'true'`` → active,
``'false'`` → not, ``'fromJson'`` → read the workflow file's own ``active``
flag (defaulting to inactive on any read/parse error)."""
if wf.active == "true":
return True
if wf.active == "false":
return False
try:
data = json.loads(Path(wf.source_path).read_text(encoding="utf-8"))
return bool(data.get("active"))
except Exception:
return False
def _n8n_disabled(env: "dict[str, str]") -> bool:
"""True only for an explicit ``N8N_SOURCE=disabled``. A missing value means
the manifest default (container), as for ``MINIO_SOURCE`` in consumer storage."""
return str(env.get("N8N_SOURCE") or "container").strip().lower() == "disabled"
def _n8n_seed_expected(consumer_config, env: "dict[str, str]") -> bool:
"""Whether the one-shot ``n8n-seed`` runs: workflows are declared and n8n is
not disabled (the overlay is dropped then)."""
return bool(consumer_config.n8n_workflows) and not _n8n_disabled(env)
def _n8n_needs_reactivation_restart(env: "dict[str, str]", consumer) -> bool:
"""True when Atlas must restart n8n after seeding to register a consumer's
production webhook: an active n8n workflow is declared, n8n is enabled, and
**no** ``N8N_API_KEY`` is configured. With a key the seed activates over the
API and the webhook registers immediately (no restart needed)."""
if (env.get("N8N_API_KEY", "") or "").strip():
return False
if (env.get("N8N_SOURCE", "") or "").strip() in ("", "disabled"):
return False
return any(
_n8n_workflow_effective_active(wf)
for wf in getattr(consumer, "n8n_workflows", ())
)
# Exact Atlas-owned finite identities, including post-start jobs and the
# generated consumer seed. Effective Compose metadata still controls admission.
_COMPOSE_FINITE_SERVICES = frozenset({
"airflow-init", "backup", "comfyui-init", "hermes-init", "iceberg-rest-init",
"label-studio-init", "langfuse-init", "lightrag-init", "litellm-init", "minio-init",
"mlflow-init", "n8n-init", "n8n-seed", "ollama-pull", "open-webui-init",
"openclaw-init", "redpanda-init", "spark-init", "supabase-db-init", "trueforge-init",
"weaviate-init", "zeppelin-init",
})
def _readiness_dependencies(name, spec, replicas, skipped):
"""Validate dependencies; optional absent/disabled/skipped services stay out."""
dependencies = spec.get("depends_on", {})
if not isinstance(dependencies, dict):
raise ValueError(f"{name}: invalid rendered dependencies")
required = []
for dependency, settings in dependencies.items():
if not isinstance(settings, dict):
raise ValueError(f"{name}: invalid dependency {dependency}")
unavailable = dependency in skipped or replicas.get(dependency, 0) == 0
if unavailable:
if settings.get("required") is False:
continue
raise ValueError(f"{name}: required dependency {dependency} is absent, disabled or skipped")
required.append(dependency)
return required
def _desired_compose_targets(services, replicas, targets, skipped):
"""Follow the actual up target list, including Compose dependency companions."""
selected = set(targets)
pending = list(selected)
while pending:
name = pending.pop()
if name not in replicas or name in skipped:
raise ValueError(f"{name}: requested service absent or skipped in rendered Compose config")
dependencies = _readiness_dependencies(name, services[name], replicas, skipped)
for dependency in dependencies:
if dependency not in selected:
selected.add(dependency)
pending.append(dependency)
return {name: replicas[name] for name in sorted(selected) if replicas[name] > 0}
def _one_shot_service_names(services) -> frozenset[str]:
"""Positive finite-role evidence from the effective Compose model."""
if not isinstance(services, dict):
return frozenset()
try:
completed = {
name for spec in services.values()
for name, dependency in (spec.get("depends_on") or {}).items()
if dependency.get("condition") == "service_completed_successfully"
}
return frozenset(
name for name, spec in services.items()
if _finite_compose_service(name, spec, completed)
)
except (AttributeError, TypeError):
return frozenset() # malformed metadata cannot prove completion
def _finite_compose_service(name: str, spec: dict, completed: set[str]) -> bool:
restart = str(spec.get("restart") or "no").strip().lower()
if restart.split(":", 1)[0] not in {"no", "on-failure"}:
return False
if name in completed:
return True
health = spec.get("healthcheck") or {}
return (
name in _COMPOSE_FINITE_SERVICES
and restart == "no"
and not _enabled_compose_healthcheck(health)
)
def _enabled_compose_healthcheck(health: dict) -> bool:
return bool(health) and not health.get("disable") and health.get("test") != ["NONE"]
def _split_local_builds(
services: dict, targets: list[str], core: set[str]
) -> tuple[list[str], list[list[str]]]:
"""Split the enabled local builds by whether the launch can go on without
them (#989).
An image is required when an enabled service lists one of the services
that build it in ``depends_on``, or when one of them is in ``core``, the
always-running tier. Services that share an ``image`` name build the
same image, so they are one unit: required if any of them is, otherwise
rebuilt and reported together. Returns the required services as one
list and every other image as its own group of services.
"""
needed, groups = _local_build_groups(services, targets)
needed |= core
required = [
name for group in groups if needed.intersection(group) for name in group
]
optional = [group for group in groups if not needed.intersection(group)]
return required, optional
def _local_build_groups(
services: dict, targets: list[str]
) -> tuple[set[str], list[list[str]]]:
"""Every service an enabled target lists in ``depends_on``, and the
enabled local-build services grouped by the image they build."""
needed: set[str] = set()
groups: dict[str, list[str]] = {}
for name in sorted(set(targets)):
spec = services.get(name) or {}
needed.update(spec.get("depends_on") or ())
if spec.get("build") is not None:
groups.setdefault(spec.get("image") or name, []).append(name)
return needed, list(groups.values())
# Selected-source values that run a managed native host, with the module that
# builds its manager (#1342). Order matches start_managed_host_processes.
_MANAGED_HOST_SOURCES = (
("COMFYUI_SOURCE", "managed-localhost-mps", "ComfyUI (MPS)", "services.comfyui_mps_manager"),
("VLLM_METAL_SOURCE", "managed-localhost", "vLLM (Metal)", "services.vllm_metal_manager"),
("BLENDER_MCP_SOURCE", "managed-localhost", "Blender MCP", "services.blender_mcp_manager"),
)
def _managed_host_pool_blocker(module, env: dict, label: str) -> Optional[str]:
"""The launch blocker for every instance a host module starts: one, or
each of a Blender MCP pool (#851). A port the pool's own running
instances hold is freed when the pool restarts, so it is not foreign."""
pool_from_env = getattr(module, "pool_from_env", None)
if pool_from_env is None:
return _managed_host_launch_blocker(module.manager_from_env, env, label)
pool = pool_from_env(env)
# Strays above the pool size are stopped before the start, so the
# ports they hold are not foreign either.
held = module.pool_held_ports(pool_from_env(env, include_strays=True))
for member in pool:
name = f"{label} #{member.pool_index}" if member.pool_index else label
problem = _managed_host_launch_blocker(lambda _env, m=member: m, env, name, held)
if problem:
return f"instance {member.pool_index}: {problem}" if len(pool) > 1 else problem
return None
def _port_held_by_stranger(manager, held_ports) -> bool:
"""The manager's port is in use, and not by its own pool (#851)."""
return manager.port not in held_ports and manager._port_in_use()
def _managed_host_launch_blocker(factory, env: dict, label: str, held_ports=frozenset()) -> Optional[str]:
"""Why the host start would refuse, or None. Read-only: nothing is
installed, launched or signalled."""
from services import (
launched_with_other_settings,
legacy_pid_refusal_file,
refuse_untrusted_tracked_pid,
)
try:
manager = factory(env)
pre = manager.preflight()
if not pre.ok:
failed = [f"{c['name']}: {c['detail']}" for c in pre.checks if c["status"] == "fail"]
return "preflight failed: " + "; ".join(failed)
refuse_untrusted_tracked_pid(
(manager._read_pid(), manager.pid_file),
manager._managed_process_alive, manager._pid_is_stranger,
(label, RuntimeError),
)
status = manager.status()
# A process on an old port is restarted on the configured one (#1361),
# which must then be free just as for a fresh launch.
record = getattr(manager, "_launch_record", dict)() # only the built-in hosts keep one
moves = status.running and launched_with_other_settings(
record, getattr(status, "pid", None), {"port": manager.port}
)
if (moves or not status.running) and _port_held_by_stranger(manager, held_ports):
return f"port {manager.port} is already in use by an unmanaged process"
except Exception as exc: # noqa: BLE001 - the start would fail the same way
# The stamp-less record from an older pin only warns at start (#990).
return None if legacy_pid_refusal_file(exc) is not None else str(exc)
return None
def _doctor_result(
check_id: str,
status: str,
message: str,
*,
details: dict | None = None,
) -> dict:
return {
"id": check_id,
"status": status,
"message": message,
"details": details or {},
}
def _resolved_comfyui_model_rows(env: dict) -> list[dict]:
"""The per-file ComfyUI model rows the container TSV is built from (#754).
One source of truth for "what models this stack needs": the same
``comfyui_resolver`` output drives the container init download AND the
managed-host provisioner. Returns [] when nothing is selected or the
resolver cannot run (a malformed catalog surfaces through its own lint)."""
try:
from utils.comfyui_resolver import active_comfyui_models, manifest_dict
return list(manifest_dict(active_comfyui_models(env)).get("models", []))
except Exception: # noqa: BLE001 — resolver issues have their own surfaces
return []
def _resolved_comfyui_custom_nodes(env: dict) -> list:
"""The resolved ComfyUI custom-node set (#905).
One source of truth: the same ``active_custom_nodes`` output the container
TSV is built from, delivered to the managed-host provisioner. Returns []
when nothing is declared or the resolver cannot run."""
try:
from utils.comfyui_custom_nodes import active_custom_nodes
from utils.comfyui_resolver import active_comfyui_models
return list(active_custom_nodes(active_comfyui_models(env), env))
except Exception: # noqa: BLE001 — resolver issues have their own surfaces
return []
def _doctor_check_compose(starter: "AtlasStarter") -> dict:
try:
returncode, stdout, stderr, _cmd = starter.docker_manager.validate_compose_config()
except ValueError as exc:
return _doctor_result(
"compose",
"fail",
f"Compose config validation could not load consumer manifests: {exc}",
)
except RuntimeError as exc:
return _doctor_result(
"compose",
"skipped",
f"Docker Compose unavailable: {exc}",
)
if returncode == 0:
return _doctor_result("compose", "pass", "Compose config is valid.")
output = "\n".join(part for part in (stderr.strip(), stdout.strip()) if part)
summary = _compose_validation_summary(output)
return _doctor_result(
"compose",
"fail",
summary or "Compose config validation failed.",
details={"returncode": returncode, "output": output},
)
def _consumer_pin_is_auto(key: str, value: str) -> bool:
return (key == "BASE_PORT" or key.endswith("_SOURCE")) and value.strip().lower() == "auto"
def _consumer_pin_report(consumer_config, env: dict) -> tuple[dict, str]:
pins = {
key: {
"reason": reason, "origin": consumer_config.env_origins[key],
"overridden": key in env and not _consumer_pin_is_auto(key, consumer_config.env_overrides[key])
and env[key] != consumer_config.env_overrides[key],
}
for key, reason in consumer_config.env_reasons.items()
}
pin_summary = "".join(
f" {key}: {pin['reason']} ({pin['origin']})."
+ (" Consumer pin overridden." if pin["overridden"] else "")
for key, pin in pins.items()
)
return pins, pin_summary
def _doctor_check_consumer_manifests(starter: "AtlasStarter") -> dict:
try:
consumer_config = starter.config_parser.load_consumer_config()
except ValueError as exc:
return _doctor_result(
"consumer-manifests",
"fail",
f"Consumer manifest validation failed: {exc}",
)
if consumer_config.is_empty:
return _doctor_result(
"consumer-manifests",
"pass",
"No consumer manifests configured.",
)
names = [consumer.name for consumer in consumer_config.consumers]
overlay_count = sum(len(consumer.compose_overlays) for consumer in consumer_config.consumers)
pins, pin_summary = _consumer_pin_report(consumer_config, starter.config_parser.parse_env_file())
return _doctor_result(
"consumer-manifests",
"warn" if any(pin["overridden"] for pin in pins.values()) else "pass",
f"{len(names)} consumer manifest(s) valid: {', '.join(names)}." + pin_summary,
details={"consumers": names, "compose_overlays": overlay_count, "pins": pins},
)
# `$$` is Compose's escape for a literal `$` (e.g. `$${PORT}` in a shell
# healthcheck), so it is consumed first and never read as a reference.
_COMPOSE_VAR_RE = re.compile(
r"\$\$|\$\{([A-Za-z_][A-Za-z0-9_]*)(?:(:?[-?])([^}]*))?\}"
)
def _doctor_compose_var_refs(text: str) -> list[tuple[str, bool]]:
refs: list[tuple[str, bool]] = []
for match in _COMPOSE_VAR_RE.finditer(text):
var_name = match.group(1)
if var_name is None: # escaped `$$`
continue
operator = match.group(2) or ""
has_default = "-" in operator
refs.append((var_name, has_default))
return refs
def _doctor_check_overlay_env(starter: "AtlasStarter") -> dict:
root = starter.config_parser.root_dir
user_dir = root / "services" / "_user"
overlays = sorted(user_dir.glob("*/compose.yml")) if user_dir.is_dir() else []
if not overlays:
return _doctor_result(
"overlay-env",
"pass",
"No services/_user compose overlays found.",
)
env_values = dict(os.environ)
env_values.update(starter.config_parser.parse_env_file())
missing: list[dict[str, str]] = []
checked_refs = 0
for overlay in overlays:
text = overlay.read_text(encoding="utf-8")
for var_name, has_default in _doctor_compose_var_refs(text):
checked_refs += 1
if has_default or var_name in env_values:
continue
missing.append(
{
"file": str(overlay.relative_to(root)),
"variable": var_name,
}
)
if missing:
first = missing[0]
return _doctor_result(
"overlay-env",
"fail",
f"{first['file']} references unresolved variable {first['variable']}.",
details={"missing": missing, "checked_refs": checked_refs},
)
return _doctor_result(
"overlay-env",
"pass",
f"All {checked_refs} overlay env reference(s) resolve or provide defaults.",
details={"overlays": [str(path.relative_to(root)) for path in overlays]},
)
def _resolve_plugin_dirs(starter: "AtlasStarter", env_values: dict | None = None) -> list[Path]:
"""Resolve BACKEND_PLUGINS_DIR to existing host directories.
Relative parts resolve from the bootstrapper root (matching how the compose
mount is authored). Nonexistent parts are skipped — this is a best-effort
host-side view used by the doctor and the Kong plugin-auth derivation.
"""
if env_values is None:
env_values = starter.config_parser.parse_env_file()
raw_path = env_values.get("BACKEND_PLUGINS_DIR", "").strip()
dirs: list[Path] = []
for raw_part in raw_path.split(os.pathsep):
raw_part = raw_part.strip()
if not raw_part:
continue
plugins_dir = Path(raw_part).expanduser()
if not plugins_dir.is_absolute():
plugins_dir = starter.config_parser.root_dir / plugins_dir
if plugins_dir.exists():
dirs.append(plugins_dir)
return dirs
def _doctor_check_plugins(starter: "AtlasStarter") -> dict:
env_values = starter.config_parser.parse_env_file()
raw_path = env_values.get("BACKEND_PLUGINS_DIR", "").strip()
if not raw_path:
return _doctor_result(
"plugins",
"pass",
"No BACKEND_PLUGINS_DIR configured.",
)
plugin_dirs_to_check: list[Path] = []
for raw_part in raw_path.split(os.pathsep):
raw_part = raw_part.strip()
if not raw_part:
continue
plugins_dir = Path(raw_part).expanduser()
if not plugins_dir.is_absolute():
plugins_dir = starter.config_parser.root_dir / plugins_dir
if not plugins_dir.exists():
return _doctor_result(
"plugins",
"skipped",
f"Plugin directory does not exist: {plugins_dir}",
)
plugin_dirs_to_check.append(plugins_dir)
problems: list[str] = []
requirement_files = 0
requirement_entries = 0
plugin_names: list[str] = []
for plugins_dir in plugin_dirs_to_check:
for req in [plugins_dir / "requirements.txt", *plugins_dir.glob("*/requirements.txt")]:
try:
if not req.exists():
continue
requirement_files += 1
for line in req.read_text(encoding="utf-8").splitlines():
stripped = line.strip()
if not stripped or stripped.startswith("#"):
continue
requirement_entries += 1
except OSError as exc:
problems.append(f"{req} could not be read: {exc}")
plugin_names.extend(path.name for path in plugins_dir.iterdir() if path.is_dir())
if problems:
return _doctor_result(
"plugins",
"fail",
problems[0],
details={"problems": problems},
)
return _doctor_result(
"plugins",
"pass",
f"Plugin directory readable; {len(plugin_names)} plugin dir(s) found.",
details={
"plugins": plugin_names,
"requirement_files": requirement_files,
"requirement_entries": requirement_entries,
},
)
def _doctor_check_plugin_manifests(starter: "AtlasStarter") -> dict:
"""Validate optional plugin.yml manifests + their declared env (#402).
Reports malformed manifests and env problems (required-missing, type/enum
mismatch) as warnings naming the plugin and var, so a bad manifest surfaces
as a startup diagnostic rather than a runtime 500. Secret values are never
echoed. Manifest-less plugins are ignored here (the plugins check covers the
directory itself).
"""
from core.plugin_manifest import (
discover_plugin_manifests,
validate_plugin_env,
)
env_values = starter.config_parser.parse_env_file()
plugin_dirs = _resolve_plugin_dirs(starter, env_values)
if not plugin_dirs:
return _doctor_result(
"plugin-manifests",
"pass",
"No plugin directories to scan for manifests.",
)
discovery = discover_plugin_manifests(plugin_dirs)
warnings: list[str] = list(discovery.errors)
for manifest in discovery.manifests:
warnings.extend(validate_plugin_env(manifest, env_values))
names = [m.name for m in discovery.manifests]
if warnings:
return _doctor_result(
"plugin-manifests",
"warn",
warnings[0],
details={"warnings": warnings, "plugins": names},
)
if not names:
return _doctor_result(
"plugin-manifests",
"pass",
"No plugin.yml manifests found.",
)
return _doctor_result(
"plugin-manifests",
"pass",
f"{len(names)} plugin manifest(s) valid: {', '.join(names)}.",
details={"plugins": names},
)
def _doctor_check_ollama_parallel(starter: "AtlasStarter") -> dict:
"""Fail when a host Ollama daemon is provisioned below the declared need.
Only meaningful for ``ollama-localhost``: on the container sources
Atlas sets ``OLLAMA_NUM_PARALLEL`` itself, but the host-prereq
doctrine means it cannot own a host daemon's environment. Ollama
defaults to ONE parallel slot and silently serializes concurrent
requests rather than rejecting them, so a consumer that needs 8 gets
correct-but-slow behaviour with no signal — which is exactly the
failure this check exists to surface (#849).
Advisory by construction: an unreadable host config reports ``skipped``,
never ``fail``. Warning about a value we could not read would train
people to ignore doctor output.
"""
from services.ollama_localhost import host_parallel_config, parallel_shortfall
env_values = starter.config_parser.parse_env_file()
source = (env_values.get("LLM_PROVIDER_SOURCE", "") or "").strip()
if source != "ollama-localhost":
return _doctor_result(
"ollama-parallel",
"pass",
f"LLM_PROVIDER_SOURCE={source or '(unset)'} — Atlas owns the "
"parallel-serving env for container sources.",
)
raw_min = (env_values.get("OLLAMA_PARALLEL_MIN", "") or "").strip()
if not raw_min:
return _doctor_result(
"ollama-parallel",
"pass",
"No OLLAMA_PARALLEL_MIN declared — nothing to compare against.",
)
try:
declared_min = int(raw_min)
except ValueError:
return _doctor_result(
"ollama-parallel",
"fail",
f"OLLAMA_PARALLEL_MIN={raw_min!r} is not an integer.",
)
if declared_min <= 1:
return _doctor_result(
"ollama-parallel",
"pass",
f"OLLAMA_PARALLEL_MIN={declared_min} is within Ollama's default.",
)
observed = host_parallel_config()
shortfall, explanation = parallel_shortfall(declared_min, observed)
if shortfall:
return _doctor_result("ollama-parallel", "fail", explanation)
if observed.get("OLLAMA_NUM_PARALLEL") is None:
return _doctor_result("ollama-parallel", "skipped", explanation)
return _doctor_result("ollama-parallel", "pass", explanation)
def _doctor_check_ollama_residency(starter: "AtlasStarter") -> dict:
"""Fail when a host Ollama will evict its own working set mid-run.
A multi-model ingest (extract + embed + keyword) touches several models
in sequence. When OLLAMA_MAX_LOADED_MODELS is below that count, Ollama
unloads one to load the next and reloads it moments later — the run
crawls with no error anywhere, which is exactly the failure a lint
should surface (#798).
Advisory like its parallel-serving sibling: an unreadable host config
reports ``skipped``, never ``fail``.
"""
from services.ollama_localhost import (
host_keep_alive,
host_parallel_config,
residency_shortfall,
)
env_values = starter.config_parser.parse_env_file()
source = (env_values.get("LLM_PROVIDER_SOURCE", "") or "").strip()
if source != "ollama-localhost":
return _doctor_result(
"ollama-residency",
"pass",
f"LLM_PROVIDER_SOURCE={source or '(unset)'} — Atlas owns "
"OLLAMA_KEEP_ALIVE / OLLAMA_MAX_LOADED_MODELS for container sources.",
)
raw_min = (env_values.get("OLLAMA_MODELS_RESIDENT_MIN", "") or "").strip()
if not raw_min:
return _doctor_result(
"ollama-residency",
"pass",
"No OLLAMA_MODELS_RESIDENT_MIN declared — nothing to compare against.",
)
try:
declared_min = int(raw_min)
except ValueError:
return _doctor_result(
"ollama-residency",
"fail",
f"OLLAMA_MODELS_RESIDENT_MIN={raw_min!r} is not an integer.",
)
if declared_min <= 1:
return _doctor_result(
"ollama-residency",
"pass",
f"OLLAMA_MODELS_RESIDENT_MIN={declared_min} is within Ollama's default.",
)
observed = host_parallel_config()
shortfall, explanation = residency_shortfall(
declared_min, observed, host_keep_alive()
)
if shortfall:
return _doctor_result("ollama-residency", "fail", explanation)
if observed.get("OLLAMA_MAX_LOADED_MODELS") is None:
return _doctor_result("ollama-residency", "skipped", explanation)
return _doctor_result("ollama-residency", "pass", explanation)
def _missing_sidecar_status(sidecar: Path) -> str:
return "skipped" if str(sidecar) == "/custom-models.yaml" else "fail"
def _doctor_check_model_sidecars(starter: "AtlasStarter") -> dict:
env_values = starter.config_parser.parse_env_file()
raw_path = env_values.get("COMFYUI_CUSTOM_MODELS_FILE", "").strip()
if not raw_path:
return _doctor_result(
"model-sidecars",
"pass",
"No COMFYUI_CUSTOM_MODELS_FILE configured.",
)
sidecars: list[Path] = []
for raw_part in raw_path.split(os.pathsep):
raw_part = raw_part.strip()
if not raw_part:
continue
sidecar = Path(raw_part).expanduser()
if not sidecar.is_absolute():
sidecar = starter.config_parser.root_dir / sidecar
sidecars.append(sidecar)
import yaml
parsed = 0
for sidecar in sidecars:
if not sidecar.exists():
# The shipped default is a container path, absent on the host; a
# path the operator configured that is missing drops its models.
return _doctor_result(
"model-sidecars",
_missing_sidecar_status(sidecar),
f"Model sidecar does not exist: {sidecar}",
)
try:
data = yaml.safe_load(sidecar.read_text(encoding="utf-8"))
except Exception as exc: # noqa: BLE001
return _doctor_result(
"model-sidecars",
"fail",
f"Could not parse model sidecar {sidecar}: {exc}",
)
if data is not None and not isinstance(data, (dict, list)):
return _doctor_result(
"model-sidecars",
"fail",
f"Model sidecar {sidecar} must parse to a mapping or list.",
)
parsed += 1
return _doctor_result(
"model-sidecars",
"pass",
f"{parsed} model sidecar file(s) parse.",
)
def _doctor_check_unpullable_models(starter: "AtlasStarter") -> dict:
"""Presence-check declared models for the selected sources (#718 → #754/#757).
Container sources are provisioned by init containers; managed host sources
are provisioned by Atlas at start (ComfyUI managed-localhost-mps downloads
the resolved catalog, ollama-localhost pulls declared tags onto the host
daemon). This lint passes when the declared set is present, and warns —
naming what's missing and the command that provisions it — when not. Only
an unmanaged ComfyUI ``localhost`` install remains hands-off (Atlas never
writes into a user-owned tree).
"""
env = starter.config_parser.parse_env_file()
warnings: list[str] = []
llm_source = (env.get("LLM_PROVIDER_SOURCE", "") or "").strip()
if llm_source == "ollama-localhost":
# #757: Atlas provisions the host daemon now. Presence of the declared
# tag set (OLLAMA_USER_MODELS ∪ OLLAMA_CUSTOM_MODELS — the same union
# the container pull uses) decides pass vs. actionable warn, naming
# each missing tag (the host analogue of the #718 container lint).
from services.ollama_localhost import (
declared_models,
host_base_url,
list_host_tags,
)
declared = declared_models(env)
if declared:
try:
present = list_host_tags(host_base_url(env))
except Exception as exc: # noqa: BLE001 — a malformed port/env must
# degrade to a warning, never crash the whole doctor run.
present = None
warnings.append(f"Could not query the host Ollama: {exc}")
if present is None:
warnings.append(
f"Host Ollama at {host_base_url(env)} is not reachable — the "
f"{len(declared)} declared model tag(s) will be pulled on the "
f"next ./start.sh once the daemon is running (`ollama serve`)."
)
else:
normalized = {t if ":" in t else f"{t}:latest" for t in declared}
missing = sorted(normalized - present)
if missing:
warnings.append(
f"Declared Ollama model(s) missing from the host daemon: "
f"{', '.join(missing)}. Atlas pulls these on the next "
f"./start.sh (or pull now: `ollama pull "
f"{' && ollama pull '.join(missing)}`)."
)
comfy_models = (env.get("COMFYUI_USER_MODELS", "") or "").strip()
comfy_source = (env.get("COMFYUI_SOURCE", "") or "").strip()
if (
comfy_models
and comfy_source
and not comfy_source.startswith("container")
and comfy_source != "disabled"
):
if comfy_source == "managed-localhost-mps":
# #754: the managed host IS provisioned by Atlas now. Presence of
# the resolved file set decides pass vs. actionable warn.
try:
from services.comfyui_mps_manager import manager_from_env
rows = _resolved_comfyui_model_rows(env)
if not rows:
# A resolver failure yields [] — declared models with an
# unresolvable catalog must warn, not silently pass (#754).
satisfied, missing = False, [
"declared catalog could not be resolved (see catalog lint)"
]
else:
satisfied, missing = manager_from_env(env).models_satisfied(rows)
except Exception as exc: # noqa: BLE001 - defensive
satisfied, missing = False, [f"could not check host tree: {exc}"]
if not satisfied:
warnings.append(
f"COMFYUI_USER_MODELS ({comfy_models}) is declared but the "
f"host tree is missing: {', '.join(missing)}. Atlas "
f"provisions these on the next `./start.sh` (or run "
f"`./start.sh comfyui-mps provision` now)."
)
else:
warnings.append(
f"COMFYUI_USER_MODELS ({comfy_models}) is declared but "
f"COMFYUI_SOURCE={comfy_source} reuses a host weight tree Atlas "
f"does not manage — the catalog is downloaded for container* "
f"sources and provisioned for managed-localhost-mps (#754), but "
f"not for an unmanaged localhost install. Provision manually: "
f"place the weights in your host ComfyUI models directory."
)
# #905: lint consumer-declared custom nodes for the managed host — independent
# of COMFYUI_USER_MODELS (a consumer may declare workflow nodes without any
# catalog models, e.g. an edit-workflow node). Uses load_custom_nodes (not
# active_custom_nodes) so the doctor does NOT resolve the full model catalog
# (which would fetch/print the live registry); model-required Atlas nodes are
# covered by the models doctor above when COMFYUI_USER_MODELS is declared.
if comfy_source == "managed-localhost-mps":
try:
from utils.comfyui_custom_nodes import load_custom_nodes as _load_nodes
declared_nodes = [n for n in _load_nodes(env) if n.from_consumer]
if declared_nodes:
from services.comfyui_mps_manager import manager_from_env as _node_mgr
n_ok, n_missing = _node_mgr(env).nodes_satisfied(declared_nodes)
if not n_ok:
warnings.append(
f"ComfyUI custom nodes are declared but missing on the "
f"host: {', '.join(n_missing)}. Atlas provisions these "
f"on the next `./start.sh` (or run `./start.sh "
f"comfyui-mps provision-nodes` now)."
)
except Exception: # noqa: BLE001 - defensive
pass
if warnings:
return _doctor_result(
"unpullable-models",
"warn",
" ".join(warnings),
details={"warnings": warnings},
)
return _doctor_result(
"unpullable-models",
"pass",
"No declared-but-unpullable model provisioning for the selected sources.",
)
def _doctor_check_litellm_models(starter: "AtlasStarter") -> dict:
"""Validate consumer-declared LiteLLM model rows (#411).
Load-time already resolved api_base against the approved endpoint allowlist
and stamped manifest-derived ownership (a parse failure surfaces as a fail
here). This check adds the cross-surface validation the triage calls for:
each backend-hosted model's first route segment is cross-checked against the
declared #402 backend plugin route_prefixes (when any manifests are present),
so a model pointing at a route no plugin serves surfaces as a startup
diagnostic rather than a dead /v1/models entry. Secrets never appear (rows
carry only ``os.environ/<VAR>`` references).
"""
from core.consumer_manifest import LITELLM_ENDPOINT_TEMPLATES
try:
config = starter.config_parser.load_consumer_config()
except ValueError as exc:
return _doctor_result(
"litellm-models",
"fail",
f"Consumer LiteLLM model validation failed: {exc}",
)
if not config.litellm_models:
return _doctor_result(
"litellm-models",
"pass",
"No consumer LiteLLM models declared.",
)
import urllib.parse
backend_host = urllib.parse.urlparse(
LITELLM_ENDPOINT_TEMPLATES["ATLAS_BACKEND_INTERNAL"]
).netloc
# Declared #402 backend plugin route heads (empty when no manifests present)
# plus the built-in backend route prefixes (research/health/… are legitimate
# targets even with no plugin manifest, so they must never trip the warning).
from core.plugin_manifest import RESERVED_ROUTE_PREFIXES
env_values = starter.config_parser.parse_env_file()
plugin_dirs = _resolve_plugin_dirs(starter, env_values)
plugin_heads: set[str] = set()
have_manifests = False
if plugin_dirs:
from core.plugin_manifest import discover_plugin_manifests
discovery = discover_plugin_manifests(plugin_dirs)
have_manifests = bool(discovery.manifests)
plugin_heads = {m.prefix_head for m in discovery.manifests}
warnings: list[str] = []
for model in config.litellm_models:
parsed = urllib.parse.urlparse(model.api_base)
if parsed.netloc != backend_host:
continue # non-backend endpoints have no plugin route to match
segments = [seg for seg in parsed.path.split("/") if seg]
head = segments[0] if segments else ""
# A bare base cannot be cross-checked against a declared route head.
# Nonempty built-in prefixes remain valid without plugin manifests.
if not head:
warnings.append(
f"model {model.name!r} (owner {model.consumer}) has a backend "
"api_base with no route head; confirm a consumer plugin serves "
"its OpenAI endpoint before selecting this model"
)
elif (
have_manifests
and head not in plugin_heads
and head not in RESERVED_ROUTE_PREFIXES
):
warnings.append(
f"model {model.name!r} (owner {model.consumer}) points at backend "
f"route /{head}, which matches no declared plugin route_prefix "
f"({', '.join('/' + h for h in sorted(plugin_heads)) or 'none'})"
)
owners = sorted({model.consumer for model in config.litellm_models})
names = [model.name for model in config.litellm_models]
if warnings:
return _doctor_result(
"litellm-models",
"warn",
warnings[0],
details={"warnings": warnings, "models": names, "owners": owners},
)
return _doctor_result(
"litellm-models",
"pass",
f"{len(names)} consumer LiteLLM model(s) valid: {', '.join(names)}.",
details={"models": names, "owners": owners},
)
def _doctor_check_n8n_workflows(starter: "AtlasStarter") -> dict:
"""Validate consumer-declared n8n workflows to seed (#412).
Load-time already validated the files (JSON parseable, credential-safe,
checksum, stable/unique id, non-colliding webhook routes); a parse failure
surfaces as a fail here. This check re-confirms each source file still exists
and adds an operational signal: an active workflow with declared webhooks but
no ``N8N_API_KEY`` will only register its production webhook after an n8n
restart (the seed can't activate it live without a key). Secrets never appear.
"""
try:
config = starter.config_parser.load_consumer_config()
except ValueError as exc:
return _doctor_result(
"n8n-workflows",
"fail",
f"Consumer n8n workflow validation failed: {exc}",
)
if not config.n8n_workflows:
return _doctor_result(
"n8n-workflows",
"pass",
"No consumer n8n workflows declared.",
)
env_values = starter.config_parser.parse_env_file()
warnings: list[str] = []
if _n8n_disabled(env_values):
warnings.append(
f"{len(config.n8n_workflows)} consumer n8n workflow(s) declared but "
"N8N_SOURCE is disabled — the seed is skipped until n8n is enabled."
)
for wf in config.n8n_workflows:
if not wf.source_path.is_file():
warnings.append(
f"workflow {wf.id!r} (owner {wf.consumer}) source file is missing: "
f"{wf.source_path}"
)
has_api_key = bool(env_values.get("N8N_API_KEY", "").strip())
if not has_api_key:
# Resolve the imported-active state so the warning fires only for
# workflows that will really be active (``fromJson`` reads the file).
needs_live = [
wf.id
for wf in config.n8n_workflows
if wf.webhooks and _n8n_workflow_effective_active(wf)
]
if needs_live:
warnings.append(
"N8N_API_KEY is unset — active workflows with declared webhooks "
f"({', '.join(needs_live)}) register their production webhook only "
"after an n8n restart (the seed cannot activate them live)."
)
ids = [wf.id for wf in config.n8n_workflows]
owners = sorted({wf.consumer for wf in config.n8n_workflows})
if warnings:
return _doctor_result(
"n8n-workflows",
"warn",
warnings[0],
details={"warnings": warnings, "workflows": ids, "owners": owners},
)
return _doctor_result(
"n8n-workflows",
"pass",
f"{len(ids)} consumer n8n workflow(s) valid: {', '.join(ids)}.",
details={"workflows": ids, "owners": owners},
)
def _doctor_check_rag_ingestion_profiles(starter: "AtlasStarter") -> dict:
"""Validate consumer-declared RAG ingestion profiles to register (#413).
Load-time already validated the profiles (unique names, corpus path safety,
parser/chunker/target schema, collection collisions); a parse failure surfaces
as a fail here. This check adds an operational signal: a profile whose vector
or graph target has ``on_unavailable: fail`` but whose backend endpoint is
disabled by its SOURCE will hard-fail that ingestion at runtime — surface it now.
"""
try:
config = starter.config_parser.load_consumer_config()
except ValueError as exc:
return _doctor_result(
"rag-ingestion-profiles",
"fail",
f"Consumer RAG ingestion profile validation failed: {exc}",
)
if not config.rag_ingestion_profiles:
return _doctor_result(
"rag-ingestion-profiles",
"pass",
"No consumer RAG ingestion profiles declared.",
)
env_values = starter.config_parser.parse_env_file()
# SOURCE var that gates each target backend. Its endpoint (WEAVIATE_URL,
# LIGHTRAG_ENDPOINT) is only written by a start, so a fresh .env with the
# backend enabled read as disabled (#1391).
source_var = {"weaviate": "WEAVIATE_SOURCE", "lightrag": "LIGHTRAG_SOURCE"}
warnings: list[str] = []
for profile in config.rag_ingestion_profiles:
targets = [
(t.backend, t.on_unavailable) for t in profile.vector_targets
] + [(t.backend, t.on_unavailable) for t in profile.graph_targets]
for backend, on_unavailable in targets:
var = source_var.get(backend)
enabled = _source_enabled(env_values, var) if var else True
if on_unavailable == "fail" and not enabled:
warnings.append(
f"profile {profile.name!r} target {backend} is on_unavailable=fail "
f"but {var} is disabled — ingestion will hard-fail until it is enabled."
)
names = [p.name for p in config.rag_ingestion_profiles]
owners = sorted({p.consumer for p in config.rag_ingestion_profiles})
if warnings:
return _doctor_result(
"rag-ingestion-profiles",
"warn",
warnings[0],
details={"warnings": warnings, "profiles": names, "owners": owners},
)
return _doctor_result(
"rag-ingestion-profiles",
"pass",
f"{len(names)} consumer RAG ingestion profile(s) valid: {', '.join(names)}.",
details={"profiles": names, "owners": owners},
)
def _source_enabled(env_values: dict, source_var: str) -> bool:
"""Whether ``source_var`` runs its service (anything but disabled/empty)."""
return (env_values.get(source_var) or "").strip().lower() not in ("", "disabled")
def _doctor_check_lightrag_query_profiles(starter: "AtlasStarter") -> dict:
"""Validate consumer-declared LightRAG query profiles to register (#414).
Load-time already validated each profile (unique namespaced names, supported
mode, bounded positive integers, rerank rejection, alias contract); a parse
failure surfaces as a fail here. This check adds an operational signal: a
profile can only be *served* when LightRAG itself is reachable, so warn when
profiles are declared but ``LIGHTRAG_SOURCE`` is disabled (the registry mounts
fine, but every flavor would 5xx at query time until LightRAG is enabled).
"""
try:
config = starter.config_parser.load_consumer_config()
except ValueError as exc:
return _doctor_result(
"lightrag-query-profiles",
"fail",
f"Consumer LightRAG query profile validation failed: {exc}",
)
if not config.lightrag_query_profiles:
return _doctor_result(
"lightrag-query-profiles",
"pass",
"No consumer LightRAG query profiles declared.",
)
names = [p.name for p in config.lightrag_query_profiles]
owners = sorted({p.consumer for p in config.lightrag_query_profiles})
aliases = sorted(
{p.litellm_alias for p in config.lightrag_query_profiles if p.litellm_alias}
)
env_values = starter.config_parser.parse_env_file()
if not _source_enabled(env_values, "LIGHTRAG_SOURCE"):
return _doctor_result(
"lightrag-query-profiles",
"warn",
f"{len(names)} LightRAG query profile(s) declared but LIGHTRAG_SOURCE is "
f"disabled — the registry mounts but flavors cannot be served until LightRAG "
f"is enabled.",
details={"profiles": names, "owners": owners, "aliases": aliases},
)
return _doctor_result(
"lightrag-query-profiles",
"pass",
f"{len(names)} consumer LightRAG query profile(s) valid: {', '.join(names)}.",
details={"profiles": names, "owners": owners, "aliases": aliases},
)
def _doctor_check_lightrag_rerank_adapter(starter: "AtlasStarter") -> dict:
"""Operational signal for the LightRAG → TEI rerank adapter (#415).
The adapter route always exists on the backend, but it only reranks when
LightRAG is wired to it. Wiring happens only when the operator opts in
(``LIGHTRAG_RERANK_ADAPTER_ENABLED=true``) AND both LightRAG and the TEI
reranker are enabled. Warn on the mismatch — flag on but a prerequisite
service off — because reranking would silently be a no-op (RERANK_BINDING
stays ``null``).
"""
env_values = starter.config_parser.parse_env_file()
flag = (env_values.get("LIGHTRAG_RERANK_ADAPTER_ENABLED", "") or "").strip().lower() == "true"
if not flag:
return _doctor_result(
"lightrag-rerank-adapter",
"pass",
"LightRAG rerank adapter disabled (default) — LightRAG will not rerank through TEI.",
)
lightrag_enabled = (env_values.get("LIGHTRAG_SOURCE", "disabled") or "disabled").strip() != "disabled"
tei_enabled = (env_values.get("TEI_RERANKER_SOURCE", "disabled") or "disabled").strip() != "disabled"
missing = []
if not lightrag_enabled:
missing.append("LIGHTRAG_SOURCE")
if not tei_enabled:
missing.append("TEI_RERANKER_SOURCE")
if missing:
return _doctor_result(
"lightrag-rerank-adapter",
"warn",
"LIGHTRAG_RERANK_ADAPTER_ENABLED=true but "
f"{' and '.join(missing)} is disabled — reranking will be a no-op "
"until both LightRAG and the TEI reranker are enabled.",
details={"missing": missing},
)
if not (env_values.get("LIGHTRAG_RERANK_ADAPTER_TOKEN", "") or "").strip():
return _doctor_result(
"lightrag-rerank-adapter",
"warn",
"LightRAG rerank adapter enabled but LIGHTRAG_RERANK_ADAPTER_TOKEN is empty — "
"the /lightrag/rerank route will 503 until a token is generated (re-run start).",
)
return _doctor_result(
"lightrag-rerank-adapter",
"pass",