Files
wehub-resource-sync 4b6817381b
Benchmark image — build + push to ECR (any adapter) / build + push (push) Waiting to run
CI / quality (ubuntu-latest) (push) Waiting to run
CI / test (tools-runtime) (push) Waiting to run
CI / test (e2e-general) (push) Waiting to run
CI / test (cli-runtime) (push) Waiting to run
CI / test (e2e-provider-and-openclaw) (push) Waiting to run
CI / test (integrations-and-misc) (push) Waiting to run
CI / coverage-report (push) Blocked by required conditions
CI / test-kubernetes (push) Waiting to run
CI / should-run-thorough (push) Waiting to run
CI / test-thorough (cloudwatch-demo) (push) Blocked by required conditions
CI / test-thorough (flink-ecs) (push) Blocked by required conditions
CI / test-thorough (upstream-lambda) (push) Blocked by required conditions
CI / test-thorough (prefect-ecs-fargate) (push) Blocked by required conditions
CodeQL / Analyze (python) (push) Waiting to run
Release / build-binaries (zip, opensre.exe, onefile, windows-latest, windows-x64) (push) Blocked by required conditions
Release / publish-release (push) Blocked by required conditions
Release / publish-main-release (push) Blocked by required conditions
Release / prepare (push) Waiting to run
Release / verify (push) Blocked by required conditions
Release / build-python-dist (push) Blocked by required conditions
Release / build-binaries (tar.gz, opensre, onedir, macos-15-intel, darwin-x64) (push) Blocked by required conditions
Release / build-binaries (tar.gz, opensre, onedir, macos-latest, darwin-arm64) (push) Blocked by required conditions
Release / build-binaries (tar.gz, opensre, onedir, ubuntu-22.04, linux-x64) (push) Blocked by required conditions
Release / build-binaries (tar.gz, opensre, onedir, ubuntu-22.04-arm, linux-arm64) (push) Blocked by required conditions
Synthetic Deterministic Tests / Synthetic offline (deterministic) (push) Waiting to run
Interactive Shell Live (PR + post-merge) / turn-checks (no-LLM) (push) Waiting to run
Interactive Shell Live (PR + post-merge) / turn-live shard ${{ matrix.shard_index }} (push) Waiting to run
CI (OpenClaw E2E) / openclaw test (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 13:10:45 +08:00

270 lines
9.8 KiB
Python

"""REPL configuration — three-tier resolution: file → env var → CLI flag."""
from __future__ import annotations
import logging
import os
from dataclasses import dataclass
from typing import Any
_VALID_LAYOUTS = ("classic", "pinned")
_FALSE_VALUES = ("", "0", "false", "off", "no")
log = logging.getLogger(__name__)
# ── Release notes ─────────────────────────────────────────────────────────────
# Shown in the "What's new" panel on startup. Update this each release with
# exactly 2 user-visible changes. Keep each entry under ~50 chars so it fits
# the right column without truncation. The banner reads this at import time.
WHATS_NEW: tuple[str, ...] = (
"Confidence scoring now shown during diagnosis",
"New /save command exports investigation reports",
)
def _read_config_file() -> dict[str, Any]:
"""Read the interactive section from ~/.opensre/config.yml.
Returns an empty dict if the file is missing, unreadable, or malformed.
Failures are always silent — a bad config file must never crash the CLI.
"""
try:
import yaml # type: ignore[import-untyped]
from config.constants import OPENSRE_HOME_DIR
config_path = OPENSRE_HOME_DIR / "config.yml"
if not config_path.exists():
return {}
data = yaml.safe_load(config_path.read_text(encoding="utf-8"))
if not isinstance(data, dict):
return {}
interactive = data.get("interactive", {})
if not isinstance(interactive, dict):
return {}
return interactive
except Exception:
return {}
@dataclass(frozen=True)
class ReplConfig:
"""REPL configuration.
Axes
----
enabled : bool
When False the REPL is skipped and ``opensre`` falls back to
``render_landing()``. Controlled by ``--no-interactive`` CLI flag,
the ``OPENSRE_INTERACTIVE`` env var, or ``interactive.enabled`` in
``~/.opensre/config.yml``.
layout : str ("classic" | "pinned")
Which renderer to use. Only ``classic`` is wired today; ``pinned``
is accepted and stored so the flag round-trips cleanly once P3 lands.
Controlled by ``--layout`` CLI option, ``OPENSRE_LAYOUT`` env var, or
``interactive.layout`` in ``~/.opensre/config.yml``.
theme : str
Interactive shell color palette. Controlled by ``--theme`` CLI option,
``OPENSRE_THEME`` env var, or ``interactive.theme`` in
``~/.opensre/config.yml``.
"""
enabled: bool = True
layout: str = "classic"
theme: str = "blue"
alert_listener_enabled: bool = False
alert_listener_host: str = "127.0.0.1"
alert_listener_port: int = 0
alert_listener_token: str | None = None
@staticmethod
def _coerce_bool(value: Any, *, default: bool) -> bool:
if isinstance(value, bool):
return value
if value is None:
return default
return str(value).strip().lower() not in _FALSE_VALUES
@classmethod
def load(
cls,
*,
cli_enabled: bool | None = None,
cli_layout: str | None = None,
cli_theme: str | None = None,
apply_active_theme: bool = True,
) -> ReplConfig:
"""Resolve config from all three tiers.
Priority (highest wins):
1. CLI flag — ``cli_enabled`` / ``cli_layout`` / ``cli_theme`` params
2. Env var — ``OPENSRE_INTERACTIVE`` / ``OPENSRE_LAYOUT`` / ``OPENSRE_THEME``
3. Config file — ``~/.opensre/config.yml`` ``interactive`` section
4. Built-in defaults (enabled=True, layout="classic", theme="blue")
When ``apply_active_theme`` is False, resolve the theme string without
calling :func:`set_active_theme` (for passive config reads during a live
REPL session, e.g. banner rendering).
"""
file_conf = _read_config_file()
# --- enabled ---
if cli_enabled is not None:
enabled = cli_enabled
elif (env_val := os.getenv("OPENSRE_INTERACTIVE")) is not None:
enabled = cls._coerce_bool(env_val, default=True)
else:
enabled = cls._coerce_bool(file_conf.get("enabled"), default=True)
# --- layout ---
if cli_layout is not None:
layout = cli_layout.lower()
elif (env_val := os.getenv("OPENSRE_LAYOUT")) is not None:
layout = env_val.lower()
else:
layout = str(file_conf.get("layout", "classic")).lower()
if layout not in _VALID_LAYOUTS:
layout = "classic"
# --- theme ---
from platform.terminal.theme import (
DEFAULT_THEME_NAME,
get_theme,
list_theme_names,
set_active_theme,
)
if cli_theme is not None:
theme = cli_theme.strip().lower()
elif (env_val := os.getenv("OPENSRE_THEME")) is not None:
theme = env_val.strip().lower()
if theme not in list_theme_names():
log.warning(
"OPENSRE_THEME=%r is not a valid theme; defaulting to %r.",
env_val,
DEFAULT_THEME_NAME,
)
theme = DEFAULT_THEME_NAME
else:
raw_theme = file_conf.get("theme", DEFAULT_THEME_NAME)
theme = str(raw_theme).strip().lower()
if theme not in list_theme_names():
log.warning(
"config.yml interactive.theme=%r is not a valid theme; defaulting to %r.",
raw_theme,
DEFAULT_THEME_NAME,
)
theme = DEFAULT_THEME_NAME
if apply_active_theme:
active_theme = set_active_theme(theme)
theme = active_theme.name
else:
theme = get_theme(theme).name
# --- alert_listener_enabled ---
if (env_val := os.getenv("OPENSRE_ALERT_LISTENER_ENABLED")) is not None:
alert_listener_enabled = cls._coerce_bool(env_val, default=False)
else:
alert_listener_enabled = cls._coerce_bool(
file_conf.get("alert_listener_enabled"), default=False
)
# --- alert_listener_host ---
if (env_val := os.getenv("OPENSRE_ALERT_LISTENER_HOST")) is not None:
alert_listener_host = env_val.strip()
else:
alert_listener_host = str(file_conf.get("alert_listener_host", "127.0.0.1"))
# --- alert_listener_port ---
if (env_val := os.getenv("OPENSRE_ALERT_LISTENER_PORT")) is not None:
try:
alert_listener_port = int(env_val.strip())
except ValueError:
log.warning(
"OPENSRE_ALERT_LISTENER_PORT=%r is not a valid port number; defaulting to 0 (random).",
env_val,
)
alert_listener_port = 0
else:
try:
alert_listener_port = int(file_conf.get("alert_listener_port", 0))
except ValueError:
log.warning(
"config.yml interactive.alert_listener_port=%r is not a valid port number; defaulting to 0 (random).",
file_conf.get("alert_listener_port"),
)
alert_listener_port = 0
# --- alert_listener_token ---
if (env_val := os.getenv("OPENSRE_ALERT_LISTENER_TOKEN")) is not None:
alert_listener_token = env_val.strip() or None
else:
alert_listener_token = file_conf.get("alert_listener_token") or None
return cls(
enabled=enabled,
layout=layout,
theme=theme,
alert_listener_enabled=alert_listener_enabled,
alert_listener_host=alert_listener_host,
alert_listener_port=alert_listener_port,
alert_listener_token=alert_listener_token,
)
@classmethod
def from_env(cls) -> ReplConfig:
"""Convenience alias — loads from env + file, no CLI override."""
return cls.load()
def read_history_settings() -> dict[str, Any]:
"""Return the ``interactive.history`` config block, or empty dict."""
interactive = _read_config_file()
raw = interactive.get("history", {})
return raw if isinstance(raw, dict) else {}
def read_prompt_log_settings() -> dict[str, Any]:
"""Return the ``interactive.prompt_log`` config block, or empty dict."""
interactive = _read_config_file()
raw = interactive.get("prompt_log", {})
return raw if isinstance(raw, dict) else {}
def read_github_login_deferred() -> bool:
"""Return True when the user skipped the first-launch GitHub sign-in prompt."""
interactive = _read_config_file()
return ReplConfig._coerce_bool(interactive.get("github_login_deferred"), default=False)
def write_github_login_deferred(deferred: bool) -> None:
"""Persist or clear the first-launch GitHub sign-in deferral flag."""
try:
import yaml # type: ignore[import-untyped]
from config.constants import OPENSRE_HOME_DIR
config_path = OPENSRE_HOME_DIR / "config.yml"
data: dict[str, Any] = {}
if config_path.exists():
loaded = yaml.safe_load(config_path.read_text(encoding="utf-8"))
if isinstance(loaded, dict):
data = loaded
interactive = data.get("interactive")
if not isinstance(interactive, dict):
interactive = {}
interactive["github_login_deferred"] = deferred
data["interactive"] = interactive
config_path.parent.mkdir(parents=True, exist_ok=True)
config_path.write_text(yaml.safe_dump(data, sort_keys=False), encoding="utf-8")
except Exception:
log.debug("Could not persist github_login_deferred", exc_info=True)