fix(tui): integrate ShellSafetyService properly in TUI app (#6361)

- route shell submissions through ShellSafetyService and surface warnings in the UI
- add shell warning banner, prompt styling, and configurable shell.warn_dangerous flag
- extend TUI coverage scenarios for shell safety and document the fix

ISSUES CLOSED: #6361

CHANGELOG.md

@@ -1255,6 +1255,14 @@ iteration` and data corruption under concurrent plan execution. All public
   `ResourceEdgeModel`, so the child-link check correctly blocks deletion.
 
 
+### Fixed
+
+- **TUI — Shell safety integration** (#6361): The Textual prompt now uses
+  `ShellSafetyService` to analyse shell commands, highlight dangerous input
+  with `$error` styling, and display an advisory warning banner. Removed the
+  inline environment-variable gate, added `shell.warn_dangerous` configuration,
+  and wired the warning indicator into the TUI layout.
+
 ---
 
 ### Fixed

features/steps/tui_app_coverage_steps.py

@@ -19,6 +19,7 @@ import sys
 import tempfile
 from pathlib import Path
 from types import ModuleType, SimpleNamespace
+from typing import Any, cast
 from unittest.mock import MagicMock, patch
 
 from behave import given, then, when
@@ -78,10 +79,21 @@ def _build_mock_textual():
     class MockStatic:
         def __init__(self, *args, **kwargs):
             self._text = ""
+            self.display = False
+            self._classes: set[str] = set()
 
         def update(self, text):
             self._text = text
 
+        def add_class(self, name):
+            self._classes.add(name)
+
+        def remove_class(self, name):
+            self._classes.discard(name)
+
+        def has_class(self, name):
+            return name in self._classes
+
     class MockInput:
         """Minimal Input stand-in for the Textual base class."""
 
@@ -89,13 +101,23 @@ def _build_mock_textual():
 
         def __init__(self, *args, **kwargs):
             self.value = ""
+            self._classes: set[str] = set()
 
-    mock_textual_app.App = MockApp
-    mock_textual_containers.Vertical = MockVertical
-    mock_textual_widgets.Header = MockHeader
-    mock_textual_widgets.Footer = MockFooter
-    mock_textual_widgets.Static = MockStatic
-    mock_textual_widgets.Input = MockInput
+        def add_class(self, name):
+            self._classes.add(name)
+
+        def remove_class(self, name):
+            self._classes.discard(name)
+
+        def has_class(self, name):
+            return name in self._classes
+
+    cast(Any, mock_textual_app).App = MockApp
+    cast(Any, mock_textual_containers).Vertical = MockVertical
+    cast(Any, mock_textual_widgets).Header = MockHeader
+    cast(Any, mock_textual_widgets).Footer = MockFooter
+    cast(Any, mock_textual_widgets).Static = MockStatic
+    cast(Any, mock_textual_widgets).Input = MockInput
 
     return {
         "textual": mock_textual,
@@ -189,6 +211,20 @@ def step_import_with_mock_textual(context):
     context.add_cleanup(lambda: _cleanup_tmpdir(context))
 
 
+@given("shell danger warnings are disabled in settings")
+def step_disable_shell_warnings(context):
+    stub = SimpleNamespace(shell_warn_dangerous=False)
+    patcher = patch(
+        "cleveragents.config.settings.get_settings",
+        return_value=stub,
+    )
+    patcher.start()
+    context.add_cleanup(patcher.stop)
+    app_patcher = patch("cleveragents.tui.app.get_settings", return_value=stub)
+    app_patcher.start()
+    context.add_cleanup(app_patcher.stop)
+
+
 # ---------------------------------------------------------------------------
 # Module-level import gate (lines 31-38)
 # ---------------------------------------------------------------------------
@@ -434,6 +470,19 @@ def _submit_text(context, text):
     context._tui_app.on_input_submitted(event)
 
 
+def _submit_text_with_mocked_shell(context, text, stdout: str = "mocked") -> None:
+    from cleveragents.tui.input.shell_exec import ShellResult
+
+    def fake_run(command: str, **_: Any) -> ShellResult:
+        return ShellResult(command=command, exit_code=0, stdout=stdout, stderr="")
+
+    with patch(
+        "cleveragents.tui.input.shell_exec.run_shell_command",
+        side_effect=fake_run,
+    ):
+        _submit_text(context, text)
+
+
 # ---------------------------------------------------------------------------
 # on_input_submitted: empty text (lines 144-150)
 # ---------------------------------------------------------------------------
@@ -466,6 +515,17 @@ def step_submit_text(context, text):
     _submit_text(context, text)
 
 
+@when('I submit "{text}" to the app with shell execution mocked')
+def step_submit_text_mocked_shell(context, text):
+    os.environ["CLEVERAGENTS_ALLOW_DANGEROUS_SHELL"] = "1"
+
+    def restore_env():
+        os.environ.pop("CLEVERAGENTS_ALLOW_DANGEROUS_SHELL", None)
+
+    context.add_cleanup(restore_env)
+    _submit_text_with_mocked_shell(context, text)
+
+
 # ---------------------------------------------------------------------------
 # on_input_submitted: shell returning None (lines 170-171)
 # ---------------------------------------------------------------------------
@@ -482,11 +542,47 @@ def step_submit_shell_none(context):
             references=[],
             shell_result=None,
             command_result=None,
+            shell_warning=None,
         ),
     ):
         _submit_text(context, "!nothing")
 
 
+# ---------------------------------------------------------------------------
+# Shell warning indicator assertions
+# ---------------------------------------------------------------------------
+@then("the shell warning indicator should be visible")
+def step_shell_warning_visible(context):
+    MockStatic = context._tui_mock_static
+    banner = context._tui_app.query_one("#shell-warning", MockStatic)
+    assert banner.display is True, "Expected shell warning indicator to be visible"
+    assert banner._text, "Expected warning indicator text to be populated"
+
+
+@then("the shell warning indicator should not be visible")
+def step_shell_warning_hidden(context):
+    MockStatic = context._tui_mock_static
+    banner = context._tui_app.query_one("#shell-warning", MockStatic)
+    assert banner.display is False, "Expected shell warning indicator to be hidden"
+    assert banner._text == "", "Expected warning indicator text to be cleared"
+
+
+@then("the prompt should be marked as dangerous")
+def step_prompt_marked_dangerous(context):
+    from cleveragents.tui.widgets.prompt import PromptInput
+
+    prompt = context._tui_app.query_one("#prompt", PromptInput)
+    assert prompt.has_class("dangerous"), "Expected prompt to have dangerous class"
+
+
+@then("the prompt should not be marked as dangerous")
+def step_prompt_not_dangerous(context):
+    from cleveragents.tui.widgets.prompt import PromptInput
+
+    prompt = context._tui_app.query_one("#prompt", PromptInput)
+    assert not prompt.has_class("dangerous"), "Expected prompt to be safe"
+
+
 # ---------------------------------------------------------------------------
 # on_input_submitted: normal text with @ (lines 179-185)
 # ---------------------------------------------------------------------------

features/tui_app_coverage.feature

@@ -146,6 +146,32 @@ Feature: TUI App Coverage
     And I submit "!echo tui_shell_test" to the app
     Then the conversation widget should contain "tui_shell_test"
 
+  Scenario: on_input_submitted surfaces shell safety warnings
+    Given a mock command router and persona state
+    When I instantiate the Textual TUI app
+    And I call on_mount on the app
+    And I submit "!rm -rf /tmp" to the app with shell execution mocked
+    Then the shell warning indicator should be visible
+    And the prompt should be marked as dangerous
+
+  Scenario: shell warning indicator is cleared after safe command
+    Given a mock command router and persona state
+    When I instantiate the Textual TUI app
+    And I call on_mount on the app
+    And I submit "!rm -rf /tmp" to the app with shell execution mocked
+    And I submit "!echo cleared" to the app with shell execution mocked
+    Then the shell warning indicator should not be visible
+    And the prompt should not be marked as dangerous
+
+  Scenario: shell danger warnings can be disabled via settings
+    Given shell danger warnings are disabled in settings
+    And a mock command router and persona state
+    When I instantiate the Textual TUI app
+    And I call on_mount on the app
+    And I submit "!rm -rf /tmp" to the app with shell execution mocked
+    Then the shell warning indicator should not be visible
+    And the prompt should not be marked as dangerous
+
   # --- on_input_submitted with shell returning None (lines 170-171) ---
 
   Scenario: on_input_submitted handles None shell result

src/cleveragents/config/settings.py

@@ -180,6 +180,15 @@ class Settings(BaseSettings):
         validation_alias=AliasChoices("CLEVERAGENTS_DEBUG_ENABLED"),
     )
 
+    shell_warn_dangerous: bool = Field(
+        default=True,
+        validation_alias=AliasChoices("CLEVERAGENTS_SHELL_WARN_DANGEROUS"),
+        description=(
+            "When true, highlight potentially destructive shell commands in the TUI "
+            "and surface advisory warnings."
+        ),
+    )
+
     # Core output and estimation configuration
     format: str | None = Field(
         default=None,

src/cleveragents/tui/app.py

@@ -12,6 +12,7 @@ import structlog
 from rich.markup import escape as _escape
 
 from cleveragents.a2a.models import A2aRequest
+from cleveragents.config.settings import get_settings
 from cleveragents.core.exceptions import DatabaseError
 from cleveragents.domain.models.core.session import (
     SessionActorNotConfiguredError,
@@ -21,6 +22,7 @@ from cleveragents.tui.first_run import create_default_persona_for_actor, is_firs
 from cleveragents.tui.input.modes import InputMode, InputModeRouter
 from cleveragents.tui.input.reference_parser import suggestions
 from cleveragents.tui.persona.state import PersonaState
+from cleveragents.tui.shell_safety import DangerousCommandWarning, ShellSafetyService
 from cleveragents.tui.slash_catalog import slash_command_specs
 from cleveragents.tui.widgets.actor_selection_overlay import ActorSelectionOverlay
 from cleveragents.tui.widgets.help_panel_overlay import (
@@ -296,6 +298,18 @@ if _TEXTUAL_AVAILABLE:
             self._dispatch_gen: int = 0
             # Cached after mount to avoid repeated query_one() on every submit.
             self._conversation: Any = None
+            self._settings = get_settings()
+            self._shell_warn_enabled = getattr(
+                self._settings, "shell_warn_dangerous", True
+            )
+            self._allow_dangerous_shell = self._resolve_allow_dangerous_shell()
+            self._shell_warning_active = False
+            self._last_shell_warning: DangerousCommandWarning | None = None
+            self._shell_safety: ShellSafetyService | None = (
+                ShellSafetyService(warn_callback=self._handle_shell_warning)
+                if self._shell_warn_enabled
+                else None
+            )
 
         def compose(self) -> Any:
             yield _Header(show_clock=True)
@@ -308,6 +322,7 @@ if _TEXTUAL_AVAILABLE:
                 yield PromptInput(
                     placeholder="Type message, /command, or !shell ...", id="prompt"
                 )
+                yield _Static("", id="shell-warning")
                 yield PersonaBar(id="persona-bar")
             yield _Footer()
 
@@ -335,6 +350,7 @@ if _TEXTUAL_AVAILABLE:
             # the inner Input never gets focus and typing appears to do nothing.
             prompt = self.query_one("#prompt", PromptInput)
             prompt.focus()
+            self._clear_shell_warning()
 
         def _complete_first_run(self, actor: str) -> None:
             """Persist the chosen actor as the default persona and refresh the bar."""
@@ -472,14 +488,15 @@ if _TEXTUAL_AVAILABLE:
             if not text:
                 return
 
+            self._allow_dangerous_shell = self._resolve_allow_dangerous_shell()
+            self._clear_shell_warning()
+
             mode_router = InputModeRouter(
                 command_handler=lambda raw: self._command_router.handle(
                     raw, session_id=self._session.session_id
                 ),
-                shell_confirm=lambda _cmd: (
-                    os.environ.get("CLEVERAGENTS_ALLOW_DANGEROUS_SHELL", "").strip()
-                    in {"1", "true"}
-                ),
+                shell_confirm=self._confirm_dangerous_shell,
+                shell_safety=self._shell_safety,
             )
             result = mode_router.process(text)
             # Use the cached widget (set in on_mount) to avoid repeated
@@ -580,6 +597,55 @@ if _TEXTUAL_AVAILABLE:
             )
             worker.done_callback = _on_llm_done
 
+        def _confirm_dangerous_shell(self, command: str) -> bool:
+            if self._shell_safety is not None:
+                return self._allow_dangerous_shell
+            return self._allow_dangerous_shell
+
+        def _handle_shell_warning(self, warning: DangerousCommandWarning) -> bool:
+            if not self._shell_warn_enabled:
+                return True
+            self._show_shell_warning(warning)
+            return self._allow_dangerous_shell
+
+        def _show_shell_warning(self, warning: DangerousCommandWarning) -> None:
+            try:
+                shell_warning = self.query_one("#shell-warning", _Static)
+                prompt = self.query_one("#prompt", PromptInput)
+            except Exception:  # pragma: no cover - defensive
+                return
+
+            level = warning.danger_level.name.capitalize()
+            shell_warning.update(
+                f"⚠ Potentially destructive command detected ({level})"
+            )
+            shell_warning.display = True
+            if hasattr(prompt, "add_class"):
+                prompt.add_class("dangerous")
+            self._last_shell_warning = warning
+            self._shell_warning_active = True
+
+        def _clear_shell_warning(self) -> None:
+            try:
+                shell_warning = self.query_one("#shell-warning", _Static)
+                prompt = self.query_one("#prompt", PromptInput)
+            except Exception:  # pragma: no cover - defensive
+                return
+
+            shell_warning.update("")
+            shell_warning.display = False
+            if hasattr(prompt, "remove_class"):
+                prompt.remove_class("dangerous")
+            self._last_shell_warning = None
+            self._shell_warning_active = False
+
+        @staticmethod
+        def _resolve_allow_dangerous_shell() -> bool:
+            raw = os.environ.get("CLEVERAGENTS_ALLOW_DANGEROUS_SHELL", "").strip()
+            if not raw:
+                return True
+            return raw.lower() in {"1", "true", "yes", "on"}
+
     _ResolvedTuiApp = _TextualCleverAgentsTuiApp
 
 CleverAgentsTuiApp = _ResolvedTuiApp

src/cleveragents/tui/cleveragents.tcss

@@ -59,6 +59,21 @@ Screen {
   height: 1;
 }
 
+#prompt.dangerous {
+  border: round $error;
+  color: $error;
+}
+
+#shell-warning {
+  height: auto;
+  margin: 0 0 1 0;
+  padding: 0 1;
+  border: round $warning;
+  color: $warning;
+  background: $warning 12%;
+  display: none;
+}
+
 #persona-bar {
   height: auto;
   padding: 0 1;

src/cleveragents/tui/input/modes.py

@@ -11,6 +11,7 @@ from cleveragents.tui.input.reference_parser import (
     parse_references,
 )
 from cleveragents.tui.input.shell_exec import ShellResult, run_shell_command
+from cleveragents.tui.shell_safety import DangerousCommandWarning, ShellSafetyService
 
 
 class InputMode(StrEnum):
@@ -31,6 +32,7 @@ class ModeResult:
     references: list[str]
     shell_result: ShellResult | None
     command_result: str | None
+    shell_warning: DangerousCommandWarning | None
 
 
 class InputModeRouter:
@@ -42,10 +44,12 @@ class InputModeRouter:
         *,
         shell_confirm: Callable[[str], bool] | None = None,
         shell_timeout_seconds: int = 30,
+        shell_safety: ShellSafetyService | None = None,
     ) -> None:
         self._command_handler = command_handler
         self._shell_confirm = shell_confirm
         self._shell_timeout_seconds = shell_timeout_seconds
+        self._shell_safety = shell_safety
 
     @staticmethod
     def detect_mode(text: str) -> InputMode:
@@ -69,12 +73,21 @@ class InputModeRouter:
                 references=[],
                 shell_result=None,
                 command_result=result,
+                shell_warning=None,
             )
         if mode == InputMode.SHELL:
             command = text.lstrip()[1:].strip()
+            warning: DangerousCommandWarning | None = None
+            allowed = True
+            confirm = self._shell_confirm
+            if self._shell_safety is not None:
+                safety_result = self._shell_safety.check_command(command)
+                warning = safety_result.warning
+                allowed = safety_result.allowed
+                confirm = lambda _cmd, allow=allowed: allow
             shell_result = run_shell_command(
                 command,
-                confirm_dangerous=self._shell_confirm,
+                confirm_dangerous=confirm,
                 timeout_seconds=self._shell_timeout_seconds,
             )
             return ModeResult(
@@ -83,6 +96,7 @@ class InputModeRouter:
                 references=[],
                 shell_result=shell_result,
                 command_result=None,
+                shell_warning=warning,
             )
 
         parse_result: ReferenceParseResult = parse_references(text)
@@ -92,4 +106,5 @@ class InputModeRouter:
             references=[match.canonical for match in parse_result.matches],
             shell_result=None,
             command_result=None,
+            shell_warning=None,
         )