Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
81 changes: 48 additions & 33 deletions src/core/app/stages/command.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,17 +10,64 @@
from __future__ import annotations

import logging
from typing import Any

from src.core.config.app_config import AppConfig
from src.core.di.container import ServiceCollection
from src.core.interfaces.application_state_interface import IApplicationState
from src.core.interfaces.command_settings_interface import ICommandSettingsService
from src.core.interfaces.di_interface import IServiceProvider
from src.core.interfaces.state_provider_interface import (
ISecureStateAccess,
ISecureStateModification,
)

from .base import InitializationStage

logger = logging.getLogger(__name__)


class DefaultCommandStateService(ISecureStateAccess, ISecureStateModification):
"""Lightweight state holder used when auto-registering domain commands."""

def __init__(self, settings_service: ICommandSettingsService) -> None:
self._settings = settings_service
self._routes: list[dict[str, Any]] = []
self._command_prefix_override: str | None = None
self._api_key_redaction_override: bool | None = None
self._disable_interactive_override: bool | None = None

def get_command_prefix(self) -> str:
if self._command_prefix_override is not None:
return self._command_prefix_override
return self._settings.get_command_prefix()

def get_failover_routes(self) -> list[dict[str, Any]] | None:
return self._routes

def update_failover_routes(self, routes: list[dict[str, Any]]) -> None:
self._routes = routes

def get_api_key_redaction_enabled(self) -> bool:
if self._api_key_redaction_override is not None:
return self._api_key_redaction_override
return self._settings.get_api_key_redaction_enabled()

def get_disable_interactive_commands(self) -> bool:
if self._disable_interactive_override is not None:
return self._disable_interactive_override
return self._settings.get_disable_interactive_commands()

def update_command_prefix(self, prefix: str) -> None:
if isinstance(prefix, str) and prefix:
self._command_prefix_override = prefix

Comment on lines +61 to +64
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Preserve command-prefix validation parity

Line 62 now accepts any truthy string, so a value like " " becomes the session prefix and all commands stop matching. The old SecureStateService.update_command_prefix rejected blank/whitespace prefixes; we need the same guard here.

     def update_command_prefix(self, prefix: str) -> None:
-        if isinstance(prefix, str) and prefix:
-            self._command_prefix_override = prefix
+        if not isinstance(prefix, str):
+            return
+
+        trimmed = prefix.strip()
+        if not trimmed:
+            return
+
+        self._command_prefix_override = trimmed
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
def update_command_prefix(self, prefix: str) -> None:
if isinstance(prefix, str) and prefix:
self._command_prefix_override = prefix
def update_command_prefix(self, prefix: str) -> None:
if not isinstance(prefix, str):
return
trimmed = prefix.strip()
if not trimmed:
return
self._command_prefix_override = trimmed
🤖 Prompt for AI Agents
In src/core/app/stages/command.py around lines 61 to 64, the current setter
accepts any truthy string so a whitespace-only value like "   " becomes the
prefix and breaks command matching; change the guard to require a non-empty
string after trimming whitespace (e.g., keep the isinstance(prefix, str) check
but also require prefix.strip() to be truthy) so blank/whitespace-only prefixes
are rejected and the previous validation parity with
SecureStateService.update_command_prefix is preserved.

def update_api_key_redaction(self, enabled: bool) -> None:
self._api_key_redaction_override = bool(enabled)

def update_interactive_commands(self, enabled: bool) -> None:
self._disable_interactive_override = bool(enabled)

class CommandStage(InitializationStage):
"""
Stage for registering command-related services.
Expand Down Expand Up @@ -188,39 +235,7 @@ def populate_commands_factory(provider: IServiceProvider) -> None:
ICommandSettingsService # type: ignore[type-abstract]
)

# Create a simple state service for commands
class DefaultStateService(
ISecureStateAccess, ISecureStateModification
):
def __init__(self, settings_service):
self._settings = settings_service
self._routes = []

def get_command_prefix(self):
return self._settings.get_command_prefix()

def get_failover_routes(self):
return self._routes

def update_failover_routes(self, routes):
self._routes = routes

def get_api_key_redaction_enabled(self):
return self._settings.get_api_key_redaction_enabled()

def get_disable_interactive_commands(self):
return self._settings.get_disable_interactive_commands()

def update_command_prefix(self, prefix: str) -> None:
self._settings.command_prefix = prefix

def update_api_key_redaction(self, enabled: bool) -> None:
self._settings.api_key_redaction_enabled = enabled

def update_interactive_commands(self, enabled: bool) -> None:
pass

state_service = DefaultStateService(settings_service)
state_service = DefaultCommandStateService(settings_service)

# Auto-register all commands from the domain command registry
for (
Expand Down
40 changes: 40 additions & 0 deletions tests/unit/core/app/test_default_command_state_service.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,40 @@
from src.core.app.stages.command import DefaultCommandStateService
from src.core.services.command_settings_service import CommandSettingsService


def _make_settings() -> CommandSettingsService:
return CommandSettingsService(
default_command_prefix="!/",
default_api_key_redaction=True,
default_disable_interactive_commands=False,
)


def test_command_prefix_override_is_session_local() -> None:
settings = _make_settings()
state_service = DefaultCommandStateService(settings)

state_service.update_command_prefix("$/")

assert state_service.get_command_prefix() == "$/"
assert settings.get_command_prefix() == "!/"


def test_api_key_redaction_override_is_session_local() -> None:
settings = _make_settings()
state_service = DefaultCommandStateService(settings)

state_service.update_api_key_redaction(False)

assert state_service.get_api_key_redaction_enabled() is False
assert settings.get_api_key_redaction_enabled() is True


def test_disable_interactive_commands_override_is_session_local() -> None:
settings = _make_settings()
state_service = DefaultCommandStateService(settings)

state_service.update_interactive_commands(True)

assert state_service.get_disable_interactive_commands() is True
assert settings.get_disable_interactive_commands() is False
Loading