orchestrata_ultimate.core.env_bootstrap

Reference › orchestrata_ultimate › core › env_bootstrap

The canonical loader for the RUL-019 environment contract, called by service and CLI entrypoints (the agent runtime, gateway CLI and MCP transport, harness CLI, kernel client, knowledge-graph server and many scripts) before anything reads an endpoint. It is deliberately limited to os and pathlib plus the @traces decorator. It cannot live in infra_config, because importing that module builds InfraConfig() and so freezes the environment as it stood before the overlay.

load_env_contract finds the checkout root with _find_repo_root (it looks for .env and env/), parses the files with _parse_env_file (split on the first =, blanks and comments skipped, quotes stripped) and loads them into os.environ in two phases, the second overriding the first so that a stale shell export cannot outrank the profile. It returns the list of keys it set. A process that has computed a value for itself, such as a per-worker graph database name in the test suite, claims the key with declare_owned; the claim is stored in the ORCHESTRATA_ENV_OWNED_KEYS variable, owned_keys reads it, and the overlay leaves those keys alone. The script scripts/check_env_contract.py allows this function as the one registered loader.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/core/env_bootstrap.py

"""The RUL-019 environment contract — THE canonical two-phase loader.

Deliberately dependency-light: ``os`` and ``pathlib`` only. It must be
importable from any entrypoint BEFORE anything reads an endpoint, so it
cannot live in ``core.infra_config`` — importing that module constructs
``InfraConfig()`` at import time (``_cfg = InfraConfig()``), which captures
the environment as it stands at import and would therefore freeze the
PRE-overlay values this loader exists to correct.

Enforced by ``scripts/check_env_contract.py``, whose ``REGISTERED_LOADERS``
allowlist names this function as the one loader an entrypoint may reach.
"""

import os
from collections.abc import MutableMapping
from pathlib import Path

from orchestrata_ultimate.core.base.traces import traces


@traces(spec="020", reqs=["REQ-020-235"])
def _parse_env_file(path: Path) -> list[tuple[str, str]]:
    """Parse ``KEY=value`` lines, skipping blanks, comments and non-pairs.

    Only the FIRST ``=`` separates key from value: connection URIs and tokens
    routinely contain more.
    """
    out: list[tuple[str, str]] = []
    for raw in path.read_text(encoding="utf-8").splitlines():
        line = raw.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, _, value = line.partition("=")
        key = key.strip()
        if key:
            out.append((key, value.strip().strip('"').strip("'")))
    return out


#: Env var naming the keys THIS PROCESS has deliberately specialized and which
#: the profile overlay must therefore not overwrite. Comma-separated.
#:
#: Phase 2 overrides on purpose (see ``load_env_contract``): a stale shell
#: export must never outrank the profile. But a value a process COMPUTED for
#: itself — the test suite's per-worker ``KG_DATABASE=orchkg_w<N>`` and
#: ``KG_GRPC_PORT``, or the ``MEMGRAPH_URIS`` blanking that keeps the
#: standalone test node from HA-discovering its way to the prod coordinators —
#: is not a stale export, and outranks both files.
#:
#: A MARKER rather than a hardcoded key list, because the two cases are
#: indistinguishable by key name alone: the same ``KG_DATABASE`` may arrive
#: from conftest (authoritative) or from a leftover shell export (must lose).
#: Only the process that computed the value can say which, so it says so.
#: This machine really does carry stale exports — ``tests/config.py`` aliases
#: GRAPH_USERNAME/GRAPH_PASSWORD to keys nothing sets precisely because the
#: shell still exports the NEO4J_* credentials retired on 2026-07-27.
ENV_OWNED_KEYS_VAR = "ORCHESTRATA_ENV_OWNED_KEYS"


@traces(spec="020", reqs=["REQ-020-235"])
def owned_keys(environ: MutableMapping[str, str] | None = None) -> frozenset[str]:
    """Keys the current process has claimed via :data:`ENV_OWNED_KEYS_VAR`."""
    target = os.environ if environ is None else environ
    raw = target.get(ENV_OWNED_KEYS_VAR, "")
    return frozenset(part.strip() for part in raw.split(",") if part.strip())


def declare_owned(*keys: str, environ: MutableMapping[str, str] | None = None) -> None:
    """Claim *keys* as this process's own, additively.

    Call AFTER writing the specialized values. Any loader that overlays an env
    file — ``load_env_contract`` here and ``start_test_services._load_dotenv``
    — then leaves them alone.
    """
    target = os.environ if environ is None else environ
    target[ENV_OWNED_KEYS_VAR] = ",".join(sorted(owned_keys(target) | set(keys)))


@traces(spec="020", reqs=["REQ-020-235"])
def _find_repo_root(start: Path) -> Path | None:
    """Walk upward for the checkout holding ``.env`` and/or ``env/``.

    Bounded at 8 levels: entrypoints live a few packages deep, and an unbounded
    walk on a container path would climb to ``/`` and adopt a stray file there.
    """
    candidate = start.resolve()
    for _ in range(8):
        if (candidate / ".env").exists() or (candidate / "env").is_dir():
            return candidate
        if candidate.parent == candidate:
            break
        candidate = candidate.parent
    return None


# REQ-020-235 is the only requirement in the corpus that states this exact
# behaviour ("Env files parsed for key=value pairs (strips quotes, skips
# comments); profile overlays .env with override=True"). It was written for the
# SPEC-020 harness hook and had no implementation traced to it; this loader is
# the general form of it, so the citation closes an untraced requirement rather
# than inventing an id. (An earlier SPEC-023 story-scoped id here was
# fabricated — no spec defined it — and `scripts/check_req_format.py` rejected
# its shape as non-canonical. Do not spell a rejected id out in a comment
# either: the gate scans every REQ- token in the file, comments included.)
@traces(spec="020", reqs=["REQ-020-235"])
def load_env_contract(
    *, environ: dict[str, str] | None = None, start: Path | None = None
) -> list[str]:
    """Load the RUL-019 two-phase environment contract into ``os.environ``.

    THE canonical loader. Every production entrypoint calls this before it
    reads an infrastructure endpoint, so a process started from a bare shell
    resolves the same endpoints as one started by systemd with
    ``EnvironmentFile=``. Enforced by ``scripts/check_env_contract.py``.

    Two phases, in order:

    1. ``.env`` — secrets and the prod-BASE layer — loaded WITHOUT override
       (``setdefault``), so anything already exported wins.
    2. ``env/<profile>.env`` — the endpoint overlay chosen by
       ``ORCHESTRATA_ENV_PROFILE`` (default ``test``) — loaded WITH override,
       because the profile is the authority on where infrastructure lives,
       EXCEPT for keys the process has claimed via :func:`declare_owned`.

    The override in phase 2 is the whole point: ``.env`` carries
    ``CXP_GRPC_HOST=localhost`` for the prod host user-units, and a process
    that stopped after phase 1 talked to the PROD kernel while believing it
    was on the test fleet (measured 2026-08-05 — it cost a day's worth of
    phantom transport bugs). ``.envrc`` layers the same two phases for
    interactive shells.

    The ``declare_owned`` exception exists because that override had no upper
    bound. Any entrypoint calling this mid-process reset the test suite's
    per-worker isolation back to the profile baseline — measured 2026-08-07:
    ``KG_DATABASE`` ``orchkg_w9`` -> ``orchkg``, ``KG_GRPC_PORT`` ``50161`` ->
    ``50052``, and the deliberately-blanked ``MEMGRAPH_URIS`` re-planted with
    the shared cluster. Production code then wrote to ``orchkg`` while each
    test read back from ``orchkg_w<N>``, so writes vanished and the failures
    read as "record is None" / "0 rows" naming nothing. It was diagnosed as
    Memgraph contention across four gate runs before the real mechanism was
    found; the give-away was that a victim module passed ALONE and failed only
    when a triggering module preceded it in the same process.

    ``scripts/start_test_services.py`` already carried a hardcoded list for
    exactly these keys, and this loader carried none — two loaders, two
    override policies. Both now consult :func:`owned_keys`, so the policy has
    one definition.

    Args:
        environ: Mapping to populate; defaults to ``os.environ``. Injected in
            tests so no test mutates the process environment (monkeypatch is
            banned suite-wide).
        start: Directory to search upward from for the repo root; defaults to
            this module's location.

    Returns:
        The paths actually loaded, in order — empty when no ``.env`` was
        found (a deployed container that injects env directly is legitimate).
    """
    target = os.environ if environ is None else environ
    loaded: list[str] = []
    root = _find_repo_root(start or Path(__file__).resolve().parent)
    if root is None:
        return loaded

    base = root / ".env"
    if base.exists():
        # PermissionError, not FileNotFoundError: Path.exists() only needs
        # execute on the parent directory (stat), not read on the file
        # itself, so an 0600 .env owned by a different uid STILL passes this
        # check and then blows up in read_text(). Measured 2026-08-13: the
        # containerized TEST kernel/gateway bind-mount the repo read-only
        # into an UNPRIVILEGED Incus container, which UID-shifts the host
        # owner of .env's SOPS+age-hardened 0600 permissions to "nobody" --
        # container-root cannot open it. That is exactly the scenario this
        # function's own docstring already names as legitimate ("a deployed
        # container that injects env directly"): systemd's EnvironmentFile=
        # already populated os.environ with .env's contents (from a
        # container-local, correctly-owned copy) before this process ever
        # started, so failing to also read it here a second time is not a
        # missing-secret condition -- it is exactly the redundant read the
        # docstring says is fine to skip.
        try:
            pairs = _parse_env_file(base)
        except PermissionError:
            pairs = None
        if pairs is not None:
            for key, value in pairs:
                target.setdefault(key, value)
            loaded.append(str(base))

    profile = target.get("ORCHESTRATA_ENV_PROFILE", "test")
    overlay = root / "env" / f"{profile}.env"
    if overlay.exists():
        owned = owned_keys(target)
        for key, value in _parse_env_file(overlay):
            # Membership, not truthiness: a claimed key may be deliberately
            # BLANK. tests/conftest.py empties MEMGRAPH_URIS so the standalone
            # test node cannot HA-discover its way to the prod coordinators —
            # re-planting the profile value over that blank is the failure
            # this guard exists to prevent, and `if not target.get(key)`
            # would do exactly that.
            if key in owned and key in target:
                continue
            target[key] = value
        target["ORCHESTRATA_ENV_PROFILE"] = profile
        loaded.append(str(overlay))
    return loaded