orchestrata_ultimate.gateway.runtime_config

Reference › orchestrata_ultimate › gateway › runtime_config

runtime_config is the Redis-backed runtime configuration service behind the config.get and config.watch RPCs (SPEC-014 US-014-005). It is used by the config handler, rpc and server. See gateway.

• RuntimeConfigStore keeps values in a Redis hash, with a second hash of update times and a capped events stream. get reads the whole namespace or a dotted-path subtree and returns a ConfigSnapshot whose last_updated is the latest update time among the returned keys. set writes a key, records a ConfigChange on the events stream with the actor, and returns the new snapshot; it is meant for in-process producers only, and the gateway never routes an RPC to it, so clients can read configuration but not write it. watch_events reads a window from the stream, advancing an internal cursor when called with the default id. close releases resources. Each call runs inside a tracing span.
• SecretRedactor replaces the value of any key matching a secret-name pattern (token, secret, password, api key, credential, private key) with a fixed mask, in every outbound snapshot and every config.changed broadcast, so secret values do not cross the gateway.
• ConfigSnapshot and ConfigChange are the result dataclasses, each with to_dict.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

ConfigSnapshot (dataclass)

Decorators: traces(spec='014', story='US-014-005', reqs=['REQ-014-081', 'REQ-014-082', 'REQ-014-084']), dataclass(slots=True) · Line 55

A point-in-time view of the runtime-config namespace (or a subtree).

ConfigChange (dataclass)

Decorators: traces(spec='014', story='US-014-005', reqs=['REQ-014-081']), dataclass(slots=True) · Line 76

One runtime-config mutation, as delivered on the config:events stream / the config.changed broadcast topic.

SecretRedactor

Line 96

Key-pattern secret redaction (REQ-014-083).

RuntimeConfigStore

Line 134

Redis-backed runtime settings: one flat hash of dotted keys (JSON-encoded values) + a meta hash of per-key last_updated ISO timestamps + an XADD event stream for watch.

Functions

Source

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

"""Runtime configuration service -- Redis-backed get/set/watch (SPEC-014
US-014-005, REQ-014-081..085).

A Redis-backed RUNTIME-settings layer, explicitly DISTINCT from env config:
the env-at-call-time doctrine (``os.environ.get(...)`` at call sites, no
import-time constants) is UNTOUCHED. ``RuntimeConfigStore`` never reads or
mirrors ``os.environ`` -- it owns its own Redis namespace
(``gateway:runtime_config*``); ``config.get`` exposes only that namespace.

Mutation via the gateway is forbidden (REQ-014-085): :meth:`RuntimeConfigStore.set`
is for in-process producers only (kernel boot, ops CLI, DI wiring) --
``ConfigHandler.handle_set`` exists solely to return a structured denial,
never a mutation path (D6).
"""

from __future__ import annotations

import json
import logging
import re
from collections.abc import Sequence
from dataclasses import dataclass
from datetime import UTC, datetime
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.tracing.tracer import Tracer

logger = logging.getLogger(__name__)

# Key-segment patterns whose values are always redacted (case-insensitive).
_DEFAULT_SECRET_PATTERNS: tuple[str, ...] = (
    "token",
    "secret",
    "password",
    "api_key",
    "apikey",
    "credential",
    "private_key",
)


@traces(spec="014", story="US-014-005", reqs=["REQ-014-084"])
def _now_iso() -> str:
    """UTC ISO-8601 timestamp for config timestamps/events."""
    return datetime.now(UTC).isoformat()


@traces(
    spec="014",
    story="US-014-005",
    reqs=["REQ-014-081", "REQ-014-082", "REQ-014-084"],
)
@dataclass(slots=True)
class ConfigSnapshot:
    """A point-in-time view of the runtime-config namespace (or a subtree)."""

    values: dict[str, Any]
    path: str | None
    last_updated: str | None
    redacted: bool

    @traces(spec="014", story="US-014-005", reqs=["REQ-014-084"])
    def to_dict(self) -> dict[str, Any]:
        """Serialize for the ``config.get`` RPC response payload."""
        return {
            "values": dict(self.values),
            "path": self.path,
            "last_updated": self.last_updated,
            "redacted": self.redacted,
        }


@traces(spec="014", story="US-014-005", reqs=["REQ-014-081"])
@dataclass(slots=True)
class ConfigChange:
    """One runtime-config mutation, as delivered on the ``config:events``
    stream / the ``config.changed`` broadcast topic."""

    key: str
    value: Any
    actor: str
    timestamp: str

    @traces(spec="014", story="US-014-005", reqs=["REQ-014-081"])
    def to_dict(self) -> dict[str, Any]:
        """Serialize for the ``config.changed`` broadcast payload."""
        return {
            "key": self.key,
            "value": self.value,
            "actor": self.actor,
            "timestamp": self.timestamp,
        }


class SecretRedactor:
    """Key-pattern secret redaction (REQ-014-083).

    Any key segment matching ``token|secret|password|api_key|apikey|
    credential|private_key`` (case-insensitive) has its value replaced with
    ``"***"``. Applied by :class:`ConfigHandler` on every outbound snapshot
    AND on every ``config.changed`` broadcast payload -- secret values never
    cross the gateway in either direction.
    """

    def __init__(self, extra_patterns: Sequence[str] | None = None) -> None:
        """Build the redactor's key-matching pattern.

        Args:
            extra_patterns: Additional literal substrings to treat as
                secret-key indicators, beyond the built-in defaults.
        """
        patterns = list(_DEFAULT_SECRET_PATTERNS)
        if extra_patterns:
            patterns.extend(extra_patterns)
        self._pattern = re.compile(
            "|".join(re.escape(p) for p in patterns), re.IGNORECASE
        )

    @traces(spec="014", story="US-014-005", reqs=["REQ-014-083"])
    def redact(self, values: dict[str, Any]) -> dict[str, Any]:
        """Return a copy of *values* with secret-keyed entries replaced by
        ``"***"``."""
        return {
            key: ("***" if self.is_secret_key(key) else value)
            for key, value in values.items()
        }

    def is_secret_key(self, key: str) -> bool:
        """True if *key* looks like it holds a secret value."""
        return bool(self._pattern.search(key))


class RuntimeConfigStore:
    """Redis-backed runtime settings: one flat hash of dotted keys
    (JSON-encoded values) + a meta hash of per-key ``last_updated`` ISO
    timestamps + an ``XADD`` event stream for watch.

    NOT env config: never touches ``os.environ``; the env-at-call-time
    doctrine is untouched. :meth:`set` is the in-process producer API
    (kernel boot, ops CLI) -- the gateway NEVER routes an RPC to it
    (REQ-014-085, D6).
    """

    HASH_KEY = "gateway:runtime_config"
    META_KEY = "gateway:runtime_config:meta"
    EVENTS_STREAM = "config:events"

    def __init__(self, redis_client: Any, tracer: Tracer | None = None) -> None:
        """Wire the store to a connected async Redis client.

        Args:
            redis_client: An async Redis client (typically built via
                ``core.redis_factory.create_async_redis``).
            tracer: Injected debug tracer; defaults to the process Singleton.
        """
        self._redis = redis_client
        self._tracer = tracer or Tracer.default()
        # Remembered XREAD cursor for watch_events()'s default ("$") calls --
        # lets the gateway's continuous forwarder loop call watch_events()
        # with no arguments each iteration while still advancing correctly;
        # callers that pass an explicit last_id (e.g. tests isolating a run
        # to a captured tail position) always override this.
        self._watch_cursor = "$"

    @traces(
        spec="014",
        story="US-014-005",
        reqs=["REQ-014-081", "REQ-014-082", "REQ-014-084"],
    )
    async def get(self, path: str | None = None) -> ConfigSnapshot:
        """Read the runtime-config namespace, optionally scoped to *path*.

        Args:
            path: Optional dotted-path leaf/subtree selector
                (REQ-014-082); ``None`` returns the full namespace.

        Returns:
            A :class:`ConfigSnapshot` with ``last_updated`` set to the most
            recent ``_touch`` across the returned keys (REQ-014-084), or
            ``None`` when no keys are selected.
        """
        with self._tracer.span("RuntimeConfigStore.get"):
            raw = await self._redis.hgetall(self.HASH_KEY)
            values: dict[str, Any] = {
                key: self._decode(raw_value) for key, raw_value in raw.items()
            }
            if path:
                values = self._select_path(values, path)
            last_updated = None
            if values:
                meta = await self._redis.hgetall(self.META_KEY)
                timestamps = [meta[key] for key in values if key in meta]
                if timestamps:
                    last_updated = max(timestamps)
            return ConfigSnapshot(
                values=values, path=path, last_updated=last_updated, redacted=False
            )

    @traces(spec="014", story="US-014-005", reqs=["REQ-014-085"])
    async def set(
        self, key: str, value: Any, actor: str = "in-process"
    ) -> ConfigSnapshot:
        """Write *key* = *value* -- IN-PROCESS PRODUCERS ONLY (REQ-014-085).

        The gateway never routes an RPC to this method; it exists for
        kernel boot / ops CLI / DI wiring to publish runtime settings that
        ``config.get``/``config.watch`` then expose read-only.

        Args:
            key: Dotted runtime-config key (e.g. ``"pipeline.max_retries"``).
            value: JSON-serializable value.
            actor: Who made the change, for the ``config.changed`` audit
                trail.

        Returns:
            The ``ConfigSnapshot`` for *key* after the write.
        """
        with self._tracer.span("RuntimeConfigStore.set"):
            await self._redis.hset(self.HASH_KEY, key, json.dumps(value))
            timestamp = await self._touch(key)
            change = ConfigChange(
                key=key, value=value, actor=actor, timestamp=timestamp
            )
            await self._redis.xadd(
                self.EVENTS_STREAM,
                {
                    "key": change.key,
                    "value": json.dumps(change.value),
                    "actor": change.actor,
                    "timestamp": change.timestamp,
                },
                maxlen=1000,
            )
            return await self.get(path=key)

    @traces(spec="014", story="US-014-005", reqs=["REQ-014-081"])
    async def watch_events(self, last_id: str = "$") -> list[ConfigChange]:
        """Tail one window of ``config:events`` (REQ-014-081).

        Args:
            last_id: Stream id to read after. The default ``"$"`` uses this
                store's own remembered cursor (advanced as a side effect of
                every call), so a continuous forwarder loop can call this
                with no arguments each iteration. Pass an explicit id (a
                captured tail position, or ``"0"``) to bypass the
                remembered cursor -- the shape tests use for isolation.

        Returns:
            The ``ConfigChange``s read in this window (possibly empty).
        """
        with self._tracer.span("RuntimeConfigStore.watch_events"):
            cursor = self._watch_cursor if last_id == "$" else last_id
            response = await self._redis.xread(
                {self.EVENTS_STREAM: cursor}, count=100, block=1000
            )
            changes: list[ConfigChange] = []
            for _, entries in response:
                for entry_id, fields in entries:
                    self._watch_cursor = entry_id
                    changes.append(
                        ConfigChange(
                            key=fields.get("key", ""),
                            value=self._decode(fields.get("value", "null")),
                            actor=fields.get("actor", ""),
                            timestamp=fields.get("timestamp", ""),
                        )
                    )
            return changes

    async def close(self) -> None:
        """Close the underlying Redis client."""
        await self._redis.aclose()

    @traces(spec="014", story="US-014-005", reqs=["REQ-014-082"])
    def _select_path(self, values: dict[str, Any], path: str) -> dict[str, Any]:
        """Select a dotted-path leaf or subtree from the flat *values* dict.

        A path with no matching keys returns an empty dict, not an error
        (REQ-014-082).
        """
        if path in values:
            return {path: values[path]}
        prefix = f"{path}."
        return {key: value for key, value in values.items() if key.startswith(prefix)}

    @traces(spec="014", story="US-014-005", reqs=["REQ-014-084"])
    async def _touch(self, key: str) -> str:
        """Record *key*'s ``last_updated`` timestamp; returns it."""
        timestamp = _now_iso()
        await self._redis.hset(self.META_KEY, key, timestamp)
        return timestamp

    @staticmethod
    def _decode(raw_value: Any) -> Any:
        """Best-effort JSON-decode a stored value; falls back to the raw
        string for values that predate JSON encoding or aren't valid JSON."""
        try:
            return json.loads(raw_value)
        except json.JSONDecodeError, TypeError:
            return raw_value