mirror of
https://github.com/usestrix/strix.git
synced 2026-08-22 11:02:08 +02:00
Orphaned files/dirs: - ``strix/agents/StrixAgent/`` — empty, only ``__pycache__``. - ``strix/tools/browser/litellm/`` — empty, only ``__pycache__``. - ``strix/strix_runs/`` — runtime output left in the working tree. - ``strix/prompts/`` — single Jinja template that nothing renders. Dead streaming pipeline (was never wired in the SDK migration): - Delete ``strix/interface/streaming_parser.py`` (XML tool-call parser for an output format the SDK doesn't produce). - Strip ``streaming_content`` / ``interrupted_content`` dicts and five unused methods from ``Tracer``. - Strip the streaming-render path + ``interrupted`` branch from TUI. - Trim ``strix/llm/utils.py``: drop ``normalize_tool_format``, ``parse_tool_invocations``, ``format_tool_call``, ``fix_incomplete_tool_call`` and the XML-stripping in ``clean_content``. Keep only the inter-agent-XML scrub. Unwired session compression: - Delete ``strix/llm/strix_session.py`` and ``strix/llm/memory_compressor.py``. ``Runner.run`` was never called with a ``session=``, so the compressor never ran. Drop the matching test file and the ``strix_memory_compressor_timeout`` config knob. Tracer cleanup: - Remove ``log_agent_creation``, ``log_tool_execution_start``, ``update_tool_execution``, ``update_agent_status``, ``get_agent_tools`` — none had production callers. - Rewrite the redaction + correlation tests against ``log_chat_message`` (which still emits events).
215 lines
6.5 KiB
Python
215 lines
6.5 KiB
Python
import contextlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Any, ClassVar
|
|
|
|
|
|
class Config:
|
|
"""Configuration Manager for Strix."""
|
|
|
|
# LLM Configuration
|
|
strix_llm = None
|
|
llm_api_key = None
|
|
llm_api_base = None
|
|
openai_api_base = None
|
|
litellm_base_url = None
|
|
ollama_api_base = None
|
|
strix_reasoning_effort = "high"
|
|
strix_llm_max_retries = "5"
|
|
llm_timeout = "300"
|
|
_LLM_CANONICAL_NAMES = (
|
|
"strix_llm",
|
|
"llm_api_key",
|
|
"llm_api_base",
|
|
"openai_api_base",
|
|
"litellm_base_url",
|
|
"ollama_api_base",
|
|
"strix_reasoning_effort",
|
|
"strix_llm_max_retries",
|
|
"llm_timeout",
|
|
)
|
|
|
|
# Tool & Feature Configuration
|
|
perplexity_api_key = None
|
|
strix_disable_browser = "false"
|
|
|
|
# Runtime Configuration
|
|
strix_image = "ghcr.io/usestrix/strix-sandbox:0.1.13"
|
|
strix_runtime_backend = "docker"
|
|
strix_sandbox_execution_timeout = "120"
|
|
strix_sandbox_connect_timeout = "10"
|
|
|
|
# Telemetry
|
|
strix_telemetry = "1"
|
|
strix_otel_telemetry = None
|
|
strix_posthog_telemetry = None
|
|
traceloop_base_url = None
|
|
traceloop_api_key = None
|
|
traceloop_headers = None
|
|
|
|
# Config file override (set via --config CLI arg)
|
|
_config_file_override: Path | None = None
|
|
|
|
# Tracks env vars set by the initial default-config load so they can be
|
|
# cleared when a --config override is later applied (avoids leakage).
|
|
_applied_from_default: ClassVar[dict[str, str]] = {}
|
|
|
|
@classmethod
|
|
def _tracked_names(cls) -> list[str]:
|
|
return [
|
|
k
|
|
for k, v in vars(cls).items()
|
|
if not k.startswith("_") and k[0].islower() and (v is None or isinstance(v, str))
|
|
]
|
|
|
|
@classmethod
|
|
def tracked_vars(cls) -> list[str]:
|
|
return [name.upper() for name in cls._tracked_names()]
|
|
|
|
@classmethod
|
|
def _llm_env_vars(cls) -> set[str]:
|
|
return {name.upper() for name in cls._LLM_CANONICAL_NAMES}
|
|
|
|
@classmethod
|
|
def _llm_env_changed(cls, saved_env: dict[str, Any]) -> bool:
|
|
for var_name in cls._llm_env_vars():
|
|
current = os.getenv(var_name)
|
|
if current is None:
|
|
continue
|
|
if saved_env.get(var_name) != current:
|
|
return True
|
|
return False
|
|
|
|
@classmethod
|
|
def get(cls, name: str) -> str | None:
|
|
env_name = name.upper()
|
|
default = getattr(cls, name, None)
|
|
return os.getenv(env_name, default)
|
|
|
|
@classmethod
|
|
def config_dir(cls) -> Path:
|
|
return Path.home() / ".strix"
|
|
|
|
@classmethod
|
|
def config_file(cls) -> Path:
|
|
if cls._config_file_override is not None:
|
|
return cls._config_file_override
|
|
return cls.config_dir() / "cli-config.json"
|
|
|
|
@classmethod
|
|
def load(cls) -> dict[str, Any]:
|
|
path = cls.config_file()
|
|
if not path.exists():
|
|
return {}
|
|
try:
|
|
with path.open("r", encoding="utf-8") as f:
|
|
data: dict[str, Any] = json.load(f)
|
|
return data
|
|
except (json.JSONDecodeError, OSError):
|
|
return {}
|
|
|
|
@classmethod
|
|
def save(cls, config: dict[str, Any]) -> bool:
|
|
try:
|
|
cls.config_dir().mkdir(parents=True, exist_ok=True)
|
|
config_path = cls.config_dir() / "cli-config.json"
|
|
with config_path.open("w", encoding="utf-8") as f:
|
|
json.dump(config, f, indent=2)
|
|
except OSError:
|
|
return False
|
|
with contextlib.suppress(OSError):
|
|
config_path.chmod(0o600) # may fail on Windows
|
|
return True
|
|
|
|
@classmethod
|
|
def apply_saved(cls, force: bool = False) -> dict[str, str]:
|
|
saved = cls.load()
|
|
env_vars = saved.get("env", {})
|
|
if not isinstance(env_vars, dict):
|
|
env_vars = {}
|
|
cleared_vars = {
|
|
var_name
|
|
for var_name in cls.tracked_vars()
|
|
if var_name in os.environ and os.environ.get(var_name) == ""
|
|
}
|
|
if cleared_vars:
|
|
for var_name in cleared_vars:
|
|
env_vars.pop(var_name, None)
|
|
if cls._config_file_override is None:
|
|
cls.save({"env": env_vars})
|
|
if cls._llm_env_changed(env_vars):
|
|
for var_name in cls._llm_env_vars():
|
|
env_vars.pop(var_name, None)
|
|
if cls._config_file_override is None:
|
|
cls.save({"env": env_vars})
|
|
applied = {}
|
|
|
|
for var_name, var_value in env_vars.items():
|
|
if var_name in cls.tracked_vars() and (force or var_name not in os.environ):
|
|
os.environ[var_name] = var_value
|
|
applied[var_name] = var_value
|
|
|
|
# Record what was applied from the default config so it can be cleared
|
|
# if a --config override is later provided (prevents leakage).
|
|
if cls._config_file_override is None and not force:
|
|
cls._applied_from_default = applied
|
|
|
|
return applied
|
|
|
|
@classmethod
|
|
def capture_current(cls) -> dict[str, Any]:
|
|
env_vars = {}
|
|
for var_name in cls.tracked_vars():
|
|
value = os.getenv(var_name)
|
|
if value:
|
|
env_vars[var_name] = value
|
|
return {"env": env_vars}
|
|
|
|
@classmethod
|
|
def save_current(cls) -> bool:
|
|
existing = cls.load().get("env", {})
|
|
merged = dict(existing)
|
|
|
|
for var_name in cls.tracked_vars():
|
|
value = os.getenv(var_name)
|
|
if value is None:
|
|
pass
|
|
elif value == "":
|
|
merged.pop(var_name, None)
|
|
else:
|
|
merged[var_name] = value
|
|
|
|
return cls.save({"env": merged})
|
|
|
|
|
|
def apply_saved_config(force: bool = False) -> dict[str, str]:
|
|
return Config.apply_saved(force=force)
|
|
|
|
|
|
def save_current_config() -> bool:
|
|
return Config.save_current()
|
|
|
|
|
|
def resolve_llm_config() -> tuple[str | None, str | None, str | None]:
|
|
"""Resolve LLM model, api_key, and api_base.
|
|
|
|
Returns ``(model_name, api_key, api_base)``. ``api_base`` falls back
|
|
through the ``LLM_API_BASE`` / ``OPENAI_API_BASE`` /
|
|
``LITELLM_BASE_URL`` / ``OLLAMA_API_BASE`` env chain so the user can
|
|
point at any OpenAI-compatible endpoint without changing the code.
|
|
"""
|
|
model = Config.get("strix_llm")
|
|
if not model:
|
|
return None, None, None
|
|
|
|
api_key = Config.get("llm_api_key")
|
|
api_base: str | None = (
|
|
Config.get("llm_api_base")
|
|
or Config.get("openai_api_base")
|
|
or Config.get("litellm_base_url")
|
|
or Config.get("ollama_api_base")
|
|
)
|
|
|
|
return model, api_key, api_base
|