orchestrata_ultimate.core.infra_env_rule

Reference › orchestrata_ultimate › core › infra_env_rule

The source of truth for the no-literal-fallback rule for infrastructure-connection environment keys (RUL-003). Keys that choose which live server a process talks to must be read with no default, and an unset key must raise, so that a misconfigured run cannot quietly write to the production Redis or graph. The user ruling of 2026-08-03 makes the rule strict: no literal default at all, including localhost, 127.0.0.1 or the empty string.

INFRA_ENDPOINT_KEYS is the set of governed keys (Redis host and port, the graph routing and URI keys, the graph database name, and the conversation graph keys). find_infra_env_defaults(tree) walks a parsed syntax tree and returns one InfraDefaultViolation (line number, key, method) for every .get(KEY, default) or .setdefault(KEY, default) call on one of those keys; _is_infra_key recognises the key constants. Two enforcement surfaces import it so they cannot drift: the CI gate scripts/validate_ci.py and the review check check_infra_env_defaults in the verification package. The compliant seams are graph_factory and redis_factory.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

InfraDefaultViolation (dataclass)

Decorators: dataclass(frozen=True) · Line 89

One <mapping>.get("<INFRA_KEY>", <default>) / setdefault site that supplies a forbidden literal fallback.

Functions

Source

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

"""The no-literal-fallback rule for infra-connection env keys (RUL-003).

Infra-connection endpoints must be read from the environment with **no literal
fallback**: an unset key must RAISE, never silently default to some host/URI/
port that could point at the wrong server (the classic failure being a default
of the PRODUCTION Redis or graph, so a mis-configured run quietly writes to
prod). This is the deliberate inverse of the general "always give os.environ.get
a default" rule (``checks_safety.check_env_access``) — that rule prevents a
``KeyError`` for ordinary config; THIS rule forbids a fallback for the specific
keys that select which live infra you talk to.

The canonical seam that satisfies the rule is ``core.graph_factory.resolve_uri``
/ ``resolve_kg_database`` (graph) and ``core.redis_factory`` (redis), each of
which raises when its key is unset. Call those instead of inlining a default.

This module is the single source of truth for the rule, imported by BOTH
enforcement surfaces so they can never drift:
  * the CI gate  — ``scripts/validate_ci.py`` (fails the pipeline)
  * the review   — ``verification.review.checks_safety.check_infra_env_defaults``

User ruling 2026-08-03: STRICT — ban ALL literal defaults for these keys, incl.
``"localhost"`` / ``"127.0.0.1"`` / ``""`` (empty string), not just remote IPs.
"""

from __future__ import annotations

import ast
from dataclasses import dataclass

# The REQUIRED connection keys that select which live infra a process talks to.
# A literal default for any of these is forbidden — read it from env and raise
# when unset. This set is scripts/validate_env.py REQUIRED_KEYS (the keys the
# suite/preflight already refuse to run without) plus ``KG_URI`` (the routing
# alias resolve_uri reads when KG_ROUTING_URI is absent) and the two
# ``CONVERSATION_KG_*`` keys.
#
# The conversation keys are deliberately NOT in validate_env's REQUIRED_KEYS,
# and the two sets are no longer identical. The rules differ in kind: this one
# is a SOURCE rule (no endpoint may carry a literal fallback anywhere in the
# tree), validate_env is a RUNTIME PREFLIGHT (these keys must be present before
# the suite starts). Only processes that read or write the conversation graph
# need its endpoint, so making every suite run refuse to start without it would
# gate unrelated work on an unrelated subsystem. The source rule has no such
# cost and closes the actual hole — a fallback is what would let a writer
# resolve an endpoint its readers do not share. Because these keys are
# already contractually required, a literal fallback is dead code that can only
# mask a misconfiguration by silently connecting somewhere — the exact wrong-
# server risk RUL-003 exists to remove.
#
# Deliberately EXCLUDED: genuinely-optional / feature-toggle keys where an
# absent value is a valid state, not a missing endpoint — ``MEMGRAPH_URIS`` /
# ``MEMGRAPH_HA_CONTROLLER_URL`` (empty = no HA; conftest itself sets
# ``MEMGRAPH_URIS=""``), ``TEST_MEMGRAPH_URI`` (test-only override),
# ``REDIS_DB`` (index 0), ``CXP_GRPC_HOST`` (localhost — the kernel is co-hosted),
# and the retired ``NEO4J_URI``. Those legitimately carry a default.
INFRA_ENDPOINT_KEYS: frozenset[str] = frozenset(
    {
        "REDIS_HOST",
        "REDIS_PORT",
        "KG_ROUTING_URI",
        "KG_URI",
        "KG_DATABASE",
        "MEMGRAPH_URI",
        "CXP_GRPC_PORT",
        "KG_GRPC_PORT",
        # The CONVERSATION graph's own endpoint (SPEC-021). Required for the
        # same reason as the rest, plus one specific to it: this graph is
        # cross-project operational data on the prod cluster and deliberately
        # does NOT follow ORCHESTRATA_ENV_PROFILE, so these keys appear only in
        # the secrets-only .env and in NO env/*.env overlay. A literal fallback
        # here would be the one thing that lets a writer resolve an endpoint
        # the readers do not share — which is precisely the split that
        # create_conversation_graph_repository exists to prevent.
        "CONVERSATION_KG_URI",
        "CONVERSATION_KG_DATABASE",
        # The queue half of the same pipeline: producers and consumers must
        # share one Redis, so a literal fallback here is the same split.
        "CONVERSATION_REDIS_HOST",
        "CONVERSATION_REDIS_PORT",
        "CONVERSATION_REDIS_DB",
    }
)

# Dict-read methods that take an optional default as their 2nd positional arg.
_DEFAULTING_METHODS = frozenset({"get", "setdefault"})


@dataclass(frozen=True)
class InfraDefaultViolation:
    """One ``<mapping>.get("<INFRA_KEY>", <default>)`` / ``setdefault`` site that
    supplies a forbidden literal fallback."""

    lineno: int
    key: str
    method: str  # "get" or "setdefault"


def _is_infra_key(node: ast.expr) -> str | None:
    """Return the infra key name if *node* is a string constant naming one."""
    if isinstance(node, ast.Constant) and isinstance(node.value, str):
        if node.value in INFRA_ENDPOINT_KEYS:
            return node.value
    return None


def find_infra_env_defaults(tree: ast.AST) -> list[InfraDefaultViolation]:
    """Return every ``X.get(KEY, default)`` / ``X.setdefault(KEY, default)`` call
    in *tree* whose KEY is an infra-connection key AND which supplies a default
    (a 2nd positional argument).

    Detection is receiver-agnostic on purpose: the offending reads appear as
    ``os.environ.get(...)``, ``env.get(...)``, ``active_env.get(...)``,
    ``e.get(...)`` and so on across the codebase — all are equally forbidden, so
    the rule keys off the METHOD + the KEY literal + the presence of a default,
    not the receiver's name. A one-arg ``X.get(KEY)`` (no default → ``None`` →
    caller must raise) is allowed and never flagged.
    """
    violations: list[InfraDefaultViolation] = []
    for node in ast.walk(tree):
        if not isinstance(node, ast.Call):
            continue
        func = node.func
        if not isinstance(func, ast.Attribute):
            continue
        if func.attr not in _DEFAULTING_METHODS:
            continue
        if not node.args:
            continue
        key = _is_infra_key(node.args[0])
        if key is None:
            continue
        # A default supplied as the 2nd positional arg is the violation.
        if len(node.args) >= 2:
            violations.append(
                InfraDefaultViolation(lineno=node.lineno, key=key, method=func.attr)
            )
    return violations