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_helpers.py

@@ -0,0 +1,39 @@
+"""Shared TUI step-definition helpers."""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from typing import Any
+from unittest.mock import patch
+
+from cleveragents.tui.input.shell_exec import ShellResult
+
+
+def _submit_text(context, text: str) -> None:
+    from cleveragents.tui.widgets.prompt import PromptInput
+
+    prompt = context._tui_app.query_one("#prompt", PromptInput)
+    prompt.value = text
+    event = SimpleNamespace()
+    context._tui_app.on_input_submitted(event)
+
+
+def _submit_text_with_mocked_shell(context, text: str, stdout: str = "mocked") -> None:
+    """Submit *text* while faking shell execution.
+
+    The patch target is the use site in ``modes`` rather than the
+    definition site in ``shell_exec``: ``modes`` did
+    ``from cleveragents.tui.input.shell_exec import run_shell_command``,
+    binding the symbol into its own namespace; patching the source
+    module would leave that binding (and therefore the real
+    ``subprocess.run`` call) untouched.
+    """
+
+    def fake_run(command: str, **_: Any) -> ShellResult:
+        return ShellResult(command=command, exit_code=0, stdout=stdout, stderr="")
+
+    with patch(
+        "cleveragents.tui.input.modes.run_shell_command",
+        side_effect=fake_run,
+    ):
+        _submit_text(context, text)

features/steps/_tui_mock_helpers.py

@@ -0,0 +1,191 @@
+"""Shared TUI mock-Textual infrastructure helpers for step-definition files."""
+
+from __future__ import annotations
+
+import importlib
+import shutil
+import sys
+import tempfile
+from pathlib import Path
+from types import ModuleType
+from typing import Any, cast
+from unittest.mock import MagicMock
+
+
+_MOCK_TEXTUAL_KEYS = [
+    "textual",
+    "textual.app",
+    "textual.containers",
+    "textual.widgets",
+]
+
+
+def _build_mock_textual():
+    """Build mock textual modules that satisfy the app import gate."""
+    mock_textual = ModuleType("textual")
+    mock_textual_app = ModuleType("textual.app")
+    mock_textual_containers = ModuleType("textual.containers")
+    mock_textual_widgets = ModuleType("textual.widgets")
+
+    class MockApp:
+        """Minimal App stand-in for the Textual base class."""
+
+        def __init__(self, *args, **kwargs):
+            self._widgets = {}
+
+        def query_one(self, selector, widget_type=None):
+            if selector in self._widgets:
+                return self._widgets[selector]
+            if widget_type is not None:
+                widget = widget_type(id=selector.lstrip("#"))
+                self._widgets[selector] = widget
+                return widget
+            return MagicMock()
+
+    class MockVertical:
+        def __init__(self, *args, **kwargs):
+            pass
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *args):
+            pass
+
+    class MockHeader:
+        def __init__(self, *args, **kwargs):
+            pass
+
+    class MockFooter:
+        def __init__(self, *args, **kwargs):
+            pass
+
+    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."""
+
+        value = ""
+
+        def __init__(self, *args, **kwargs):
+            self.value = ""
+            self._classes: set[str] = set()
+
+        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,
+        "textual.app": mock_textual_app,
+        "textual.containers": mock_textual_containers,
+        "textual.widgets": mock_textual_widgets,
+    }
+
+
+def _install_mock_textual(context):
+    """Inject mock textual into sys.modules and reload the app module."""
+    mocks = _build_mock_textual()
+    context._tui_saved_modules = {}
+    for key in _MOCK_TEXTUAL_KEYS:
+        context._tui_saved_modules[key] = sys.modules.pop(key, None)
+    for key, mod in mocks.items():
+        sys.modules[key] = mod
+
+    # Reload widget modules so they pick up the mock Static/Input base class
+    import cleveragents.tui.widgets.help_panel_overlay as hp_mod
+    import cleveragents.tui.widgets.persona_bar as pb_mod
+    import cleveragents.tui.widgets.prompt as prompt_mod
+    import cleveragents.tui.widgets.reference_picker as rp_mod
+    import cleveragents.tui.widgets.slash_command_overlay as sco_mod
+
+    importlib.reload(hp_mod)
+    importlib.reload(pb_mod)
+    importlib.reload(prompt_mod)
+    importlib.reload(rp_mod)
+    importlib.reload(sco_mod)
+
+    import cleveragents.tui.app as app_mod
+
+    importlib.reload(app_mod)
+    context._tui_app_mod = app_mod
+    context._tui_mock_static = mocks["textual.widgets"].Static
+
+
+def _restore_modules(context):
+    """Restore original sys.modules and reload the app module."""
+    for key, val in getattr(context, "_tui_saved_modules", {}).items():
+        if val is None:
+            sys.modules.pop(key, None)
+        else:
+            sys.modules[key] = val
+
+    # Reload widget modules so they pick up the real Static/Input base class again
+    import cleveragents.tui.widgets.help_panel_overlay as hp_mod
+    import cleveragents.tui.widgets.persona_bar as pb_mod
+    import cleveragents.tui.widgets.prompt as prompt_mod
+    import cleveragents.tui.widgets.reference_picker as rp_mod
+    import cleveragents.tui.widgets.slash_command_overlay as sco_mod
+
+    importlib.reload(hp_mod)
+    importlib.reload(pb_mod)
+    importlib.reload(prompt_mod)
+    importlib.reload(rp_mod)
+    importlib.reload(sco_mod)
+
+    import cleveragents.tui.app as app_mod
+
+    importlib.reload(app_mod)
+
+
+def _make_persona_state(context):
+    """Create a real PersonaState backed by a temp directory."""
+    from cleveragents.tui.persona.registry import PersonaRegistry
+    from cleveragents.tui.persona.state import PersonaState
+
+    tmp = tempfile.mkdtemp()
+    context._tui_tmpdir = tmp
+    registry = PersonaRegistry(config_dir=Path(tmp))
+    registry.ensure_default()
+    return PersonaState(registry=registry)
+
+
+def _cleanup_tmpdir(context):
+    tmp = getattr(context, "_tui_tmpdir", None)
+    if tmp:
+        shutil.rmtree(tmp, ignore_errors=True)
+
+
+class _FakeCommandRouter:
+    """Test command router that returns predictable responses."""
+
+    def handle(self, raw, *, session_id):
+        return f"handled:{raw}"

features/steps/tui_app_coverage_steps.py

@@ -1,181 +1,17 @@
-"""Step definitions for tui_app_coverage.feature.
+"""Step definitions for tui_app_coverage.feature targeting cleveragents.tui.app."""
 
-These steps target uncovered lines in cleveragents/tui/app.py:
-- Lines 31-38: Textual import success path (mocked)
-- Lines 81-100: _TextualCleverAgentsTuiApp class definition + __init__
-- Lines 102-112: compose method
-- Lines 114-121: on_mount method
-- Lines 123-125: action_help method
-- Lines 127-129: action_cycle_preset method
-- Lines 131-142: _refresh_persona_bar method
-- Lines 144-185: on_input_submitted (all branches)
-- Line 189: CleverAgentsTuiApp alias
-"""
-
-import importlib
-import os
-import shutil
-import sys
-import tempfile
-from pathlib import Path
-from types import ModuleType, SimpleNamespace
-from unittest.mock import MagicMock, patch
+from types import SimpleNamespace
 
 from behave import given, then, when
 
-# ---------------------------------------------------------------------------
-# Mock Textual infrastructure
-# ---------------------------------------------------------------------------
-
-_MOCK_TEXTUAL_KEYS = [
-    "textual",
-    "textual.app",
-    "textual.containers",
-    "textual.widgets",
-]
-
-
-def _build_mock_textual():
-    """Build mock textual modules that satisfy the app import gate."""
-    mock_textual = ModuleType("textual")
-    mock_textual_app = ModuleType("textual.app")
-    mock_textual_containers = ModuleType("textual.containers")
-    mock_textual_widgets = ModuleType("textual.widgets")
-
-    class MockApp:
-        """Minimal App stand-in for the Textual base class."""
-
-        def __init__(self, *args, **kwargs):
-            self._widgets = {}
-
-        def query_one(self, selector, widget_type=None):
-            if selector in self._widgets:
-                return self._widgets[selector]
-            if widget_type is not None:
-                widget = widget_type(id=selector.lstrip("#"))
-                self._widgets[selector] = widget
-                return widget
-            return MagicMock()
-
-    class MockVertical:
-        def __init__(self, *args, **kwargs):
-            pass
-
-        def __enter__(self):
-            return self
-
-        def __exit__(self, *args):
-            pass
-
-    class MockHeader:
-        def __init__(self, *args, **kwargs):
-            pass
-
-    class MockFooter:
-        def __init__(self, *args, **kwargs):
-            pass
-
-    class MockStatic:
-        def __init__(self, *args, **kwargs):
-            self._text = ""
-
-        def update(self, text):
-            self._text = text
-
-    class MockInput:
-        """Minimal Input stand-in for the Textual base class."""
-
-        value = ""
-
-        def __init__(self, *args, **kwargs):
-            self.value = ""
-
-    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
-
-    return {
-        "textual": mock_textual,
-        "textual.app": mock_textual_app,
-        "textual.containers": mock_textual_containers,
-        "textual.widgets": mock_textual_widgets,
-    }
-
-
-def _install_mock_textual(context):
-    """Inject mock textual into sys.modules and reload the app module."""
-    mocks = _build_mock_textual()
-    context._tui_saved_modules = {}
-    for key in _MOCK_TEXTUAL_KEYS:
-        context._tui_saved_modules[key] = sys.modules.pop(key, None)
-    for key, mod in mocks.items():
-        sys.modules[key] = mod
-
-    # Reload widget modules so they pick up the mock Static/Input base class
-    import cleveragents.tui.widgets.help_panel_overlay as hp_mod
-    import cleveragents.tui.widgets.persona_bar as pb_mod
-    import cleveragents.tui.widgets.prompt as prompt_mod
-    import cleveragents.tui.widgets.reference_picker as rp_mod
-    import cleveragents.tui.widgets.slash_command_overlay as sco_mod
-
-    importlib.reload(hp_mod)
-    importlib.reload(pb_mod)
-    importlib.reload(prompt_mod)
-    importlib.reload(rp_mod)
-    importlib.reload(sco_mod)
-
-    import cleveragents.tui.app as app_mod
-
-    importlib.reload(app_mod)
-    context._tui_app_mod = app_mod
-    context._tui_mock_static = mocks["textual.widgets"].Static
-
-
-def _restore_modules(context):
-    """Restore original sys.modules and reload the app module."""
-    for key, val in getattr(context, "_tui_saved_modules", {}).items():
-        if val is None:
-            sys.modules.pop(key, None)
-        else:
-            sys.modules[key] = val
-
-    # Reload widget modules so they pick up the real Static/Input base class again
-    import cleveragents.tui.widgets.help_panel_overlay as hp_mod
-    import cleveragents.tui.widgets.persona_bar as pb_mod
-    import cleveragents.tui.widgets.prompt as prompt_mod
-    import cleveragents.tui.widgets.reference_picker as rp_mod
-    import cleveragents.tui.widgets.slash_command_overlay as sco_mod
-
-    importlib.reload(hp_mod)
-    importlib.reload(pb_mod)
-    importlib.reload(prompt_mod)
-    importlib.reload(rp_mod)
-    importlib.reload(sco_mod)
-
-    import cleveragents.tui.app as app_mod
-
-    importlib.reload(app_mod)
-
-
-def _make_persona_state(context):
-    """Create a real PersonaState backed by a temp directory."""
-    from cleveragents.tui.persona.registry import PersonaRegistry
-    from cleveragents.tui.persona.state import PersonaState
-
-    tmp = tempfile.mkdtemp()
-    context._tui_tmpdir = tmp
-    registry = PersonaRegistry(config_dir=Path(tmp))
-    registry.ensure_default()
-    return PersonaState(registry=registry)
-
-
-def _cleanup_tmpdir(context):
-    tmp = getattr(context, "_tui_tmpdir", None)
-    if tmp:
-        shutil.rmtree(tmp, ignore_errors=True)
+from features.steps._tui_helpers import _submit_text
+from features.steps._tui_mock_helpers import (
+    _FakeCommandRouter,
+    _cleanup_tmpdir,
+    _install_mock_textual,
+    _make_persona_state,
+    _restore_modules,
+)
 
 
 # ---------------------------------------------------------------------------
@@ -421,19 +257,6 @@ def step_persona_bar_shows_name(context):
     assert "scope refs" in bar._text
 
 
-# ---------------------------------------------------------------------------
-# on_input_submitted helpers
-# ---------------------------------------------------------------------------
-def _submit_text(context, text):
-    """Set prompt value and fire on_input_submitted."""
-    from cleveragents.tui.widgets.prompt import PromptInput
-
-    prompt = context._tui_app.query_one("#prompt", PromptInput)
-    prompt.value = text
-    event = SimpleNamespace()
-    context._tui_app.on_input_submitted(event)
-
-
 # ---------------------------------------------------------------------------
 # on_input_submitted: empty text (lines 144-150)
 # ---------------------------------------------------------------------------
@@ -455,38 +278,6 @@ def step_conv_not_updated(context):
 # ---------------------------------------------------------------------------
 # on_input_submitted: command mode (lines 152-167)
 # ---------------------------------------------------------------------------
-@when('I submit "{text}" to the app')
-def step_submit_text(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(context, text)
-
-
-# ---------------------------------------------------------------------------
-# on_input_submitted: shell returning None (lines 170-171)
-# ---------------------------------------------------------------------------
-@when("I submit shell text that produces a None shell result")
-def step_submit_shell_none(context):
-    from cleveragents.tui.input.modes import InputMode, ModeResult
-
-    # We patch the InputModeRouter.process to return a shell result with None
-    with patch(
-        "cleveragents.tui.app.InputModeRouter.process",
-        return_value=ModeResult(
-            mode=InputMode.SHELL,
-            expanded_text="!nothing",
-            references=[],
-            shell_result=None,
-            command_result=None,
-        ),
-    ):
-        _submit_text(context, "!nothing")
-
-
 # ---------------------------------------------------------------------------
 # on_input_submitted: normal text with @ (lines 179-185)
 # ---------------------------------------------------------------------------

features/steps/tui_shell_safety_steps.py

@@ -0,0 +1,130 @@
+"""Shell safety-related step definitions for the TUI app coverage suite."""
+
+from __future__ import annotations
+
+import os
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from behave import given, then, when
+
+from cleveragents.tui.input.modes import InputMode, ModeResult
+from cleveragents.tui.shell_safety import (
+    DangerousCommandWarning,
+    DangerousPattern,
+    ShellDangerLevel,
+)
+
+from features.steps._tui_helpers import _submit_text, _submit_text_with_mocked_shell
+
+
+@when('I submit "{text}" to the app')
+def step_submit_text(context, text):
+    os.environ["CLEVERAGENTS_ALLOW_DANGEROUS_SHELL"] = "1"
+
+    def restore_env() -> None:
+        os.environ.pop("CLEVERAGENTS_ALLOW_DANGEROUS_SHELL", None)
+
+    context.add_cleanup(restore_env)
+    _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() -> None:
+        os.environ.pop("CLEVERAGENTS_ALLOW_DANGEROUS_SHELL", None)
+
+    context.add_cleanup(restore_env)
+    _submit_text_with_mocked_shell(context, text)
+
+
+@when("I submit shell text that produces a None shell result")
+def step_submit_shell_none(context):
+    # Patch the mode router to emulate a shell submission returning None.
+    with patch(
+        "cleveragents.tui.app.InputModeRouter.process",
+        return_value=ModeResult(
+            mode=InputMode.SHELL,
+            expanded_text="!nothing",
+            references=[],
+            shell_result=None,
+            command_result=None,
+            shell_warning=None,
+        ),
+    ):
+        _submit_text(context, "!nothing")
+
+
+@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"
+
+
+@given("shell danger warnings are disabled in settings")
+@when("shell danger warnings are disabled in settings")
+def step_disable_shell_warnings(context):
+    stub = SimpleNamespace(shell_warn_dangerous=False)
+    app_patcher = patch("cleveragents.tui.app.get_settings", return_value=stub)
+    app_patcher.start()
+    context.add_cleanup(app_patcher.stop)
+
+
+@when('I ask the app to confirm shell command "{command}"')
+def step_confirm_shell_command(context, command):
+    context._tui_shell_confirmation = context._tui_app._confirm_dangerous_shell(command)
+
+
+@then("the shell confirmation result should be allowed")
+def step_shell_confirmation_allowed(context):
+    assert context._tui_shell_confirmation is True
+
+
+@then("the shell confirmation result should be blocked")
+def step_shell_confirmation_blocked(context):
+    assert context._tui_shell_confirmation is False
+
+
+@when("I ask the app to handle a shell warning")
+def step_handle_shell_warning(context):
+    pattern = DangerousPattern(
+        name="test_warning",
+        pattern=r"rm -rf",
+        level=ShellDangerLevel.CRITICAL,
+        description="test warning",
+    )
+    warning = DangerousCommandWarning.from_pattern("rm -rf /tmp", pattern)
+    context._tui_shell_warning_result = context._tui_app._handle_shell_warning(warning)
+
+
+@then("the shell warning callback result should be allowed")
+def step_shell_warning_callback_allowed(context):
+    assert context._tui_shell_warning_result is True

features/tui_app_coverage.feature

@@ -146,6 +146,59 @@ 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
+
+  Scenario: ShellSafetyService owns dangerous command confirmation
+    Given a mock command router and persona state
+    When I instantiate the Textual TUI app
+    And I ask the app to confirm shell command "rm -rf /tmp"
+    Then the shell confirmation result should be allowed
+
+  Scenario: shell confirmation allows safe commands without ShellSafetyService
+    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 ask the app to confirm shell command "echo safe"
+    Then the shell confirmation result should be allowed
+
+  Scenario: shell confirmation uses the dangerous shell flag without ShellSafetyService
+    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 ask the app to confirm shell command "rm -rf /tmp"
+    Then the shell confirmation result should be blocked
+
+  Scenario: shell warning callback allows execution when warnings are disabled
+    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 ask the app to handle a shell warning
+    Then the shell warning callback result should be allowed
+
   # --- on_input_submitted with shell returning None (lines 170-171) ---
 
   Scenario: on_input_submitted handles None shell result

features/tui_shell_exec_coverage.feature

@@ -38,7 +38,7 @@ Feature: TUI Shell Exec Coverage
     Given a confirm_dangerous callback that returns False
     When I run a dangerous command "rm -rf /" with the callback
     Then the shell result exit code should be 1
-    And the shell result stderr should be "blocked dangerous shell command"
+    And the shell result stderr should be "blocked by shell safety policy"
 
   Scenario: Command that exceeds timeout returns timeout result
     Given subprocess run is mocked to raise TimeoutExpired

robot/tui_shell_safety.robot

@@ -0,0 +1,62 @@
+*** Settings ***
+Documentation    Integration coverage for TUI shell safety wiring and safeguards.
+Resource         ${CURDIR}/common.resource
+Suite Setup      Setup Test Environment With Database Isolation
+Suite Teardown   Cleanup Test Environment
+
+*** Variables ***
+${TIMEOUT}       45s
+
+*** Test Cases ***
+Shell Safety Service Blocks Denied Command
+    [Documentation]    ShellSafetyService verdict should block commands even when heuristics allow them.
+    [Tags]    tui    shell_safety    regression
+    ${script}=    Catenate    SEPARATOR=\n
+    ...    import os
+    ...    from cleveragents.tui.input.modes import InputModeRouter
+    ...    from cleveragents.tui.shell_safety import ShellSafetyService
+    ...    from cleveragents.tui.shell_safety.warning import DangerousCommandWarning
+    ...
+    ...    warnings: list[DangerousCommandWarning] = []
+    ...
+    ...    def warn_callback(warning: DangerousCommandWarning) -> bool:
+    ...    ${SPACE * 4}warnings.append(warning)
+    ...    ${SPACE * 4}return False
+    ...
+    ...    router = InputModeRouter(lambda cmd: "handled", shell_safety=ShellSafetyService(warn_callback=warn_callback))
+    ...    os.environ.pop("CLEVERAGENTS_ALLOW_DANGEROUS_SHELL", None)
+    ...    result = router.process("!chmod -R 777 /tmp/test-shell-safety")
+    ...    assert result.shell_warning is not None, "Shell safety warning should be surfaced"
+    ...    assert warnings and warnings[0].command == "chmod -R 777 /tmp/test-shell-safety"
+    ...    assert result.shell_result is not None, "Shell result should be populated"
+    ...    assert result.shell_result.exit_code == 1, f"Expected blocked exit code, got {result.shell_result.exit_code}"
+    ...    assert "blocked" in result.shell_result.stderr.lower(), result.shell_result.stderr
+    ...    print("blocked-ok")
+    ${result}=    Run Process    ${PYTHON}    -c    ${script}
+    ...    timeout=${TIMEOUT}    on_timeout=kill
+    ...    env:PYTHONPATH=${CURDIR}/../src
+    Should Be Equal As Integers    ${result.rc}    0    Shell safety blocking script failed: ${result.stderr}
+    Should Contain    ${result.stdout}    blocked-ok
+
+Shell Confirm Callback Gates All Commands
+    [Documentation]    run_shell_command must respect confirm callback regardless of built-in heuristics.
+    [Tags]    tui    shell_safety    regression
+    ${script}=    Catenate    SEPARATOR=\n
+    ...    from cleveragents.tui.input.shell_exec import run_shell_command
+    ...
+    ...    counter = {"count": 0}
+    ...
+    ...    def deny(command: str) -> bool:
+    ...    ${SPACE * 4}counter["count"] += 1
+    ...    ${SPACE * 4}return False
+    ...
+    ...    result = run_shell_command("chmod -R 777 /tmp/test-shell-safety", confirm_dangerous=deny)
+    ...    assert counter["count"] == 1, f"Expected confirm to be invoked once, got {counter['count']}"
+    ...    assert result.exit_code == 1, f"Expected blocked exit code, got {result.exit_code}"
+    ...    assert "blocked" in result.stderr.lower(), result.stderr
+    ...    print("confirm-gate-ok")
+    ${result}=    Run Process    ${PYTHON}    -c    ${script}
+    ...    timeout=${TIMEOUT}    on_timeout=kill
+    ...    env:PYTHONPATH=${CURDIR}/../src
+    Should Be Equal As Integers    ${result.rc}    0    Shell confirm gate script failed: ${result.stderr}
+    Should Contain    ${result.stdout}    confirm-gate-ok

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,
@@ -20,7 +21,9 @@ from cleveragents.domain.models.core.session import (
 from cleveragents.tui.first_run import create_default_persona_for_actor, is_first_run
 from cleveragents.tui.input.modes import InputMode, InputModeRouter
 from cleveragents.tui.input.reference_parser import suggestions
+from cleveragents.tui.input.shell_exec import looks_dangerous
 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 +299,14 @@ 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 = self._settings.shell_warn_dangerous
+            self._allow_dangerous_shell = self._resolve_allow_dangerous_shell()
+            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 +319,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 +347,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 +485,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 +594,50 @@ if _TEXTUAL_AVAILABLE:
             )
             worker.done_callback = _on_llm_done
 
+        def _confirm_dangerous_shell(self, command: str) -> bool:
+            if self._shell_safety is not None:
+                # The ShellSafetyService already provided the execution verdict.
+                return True
+            if not looks_dangerous(command):
+                return True
+            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
+
+            shell_warning.update("⚠ Potentially destructive command detected")
+            shell_warning.display = True
+            if hasattr(prompt, "add_class"):
+                prompt.add_class("dangerous")
+
+        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")
+
+        @staticmethod
+        def _resolve_allow_dangerous_shell() -> bool:
+            raw = os.environ.get("CLEVERAGENTS_ALLOW_DANGEROUS_SHELL", "").strip()
+            # Default to disallowing dangerous commands unless explicitly enabled.
+            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,25 @@ 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
+
+                def _safety_gate(_cmd: str, *, allow: bool = allowed) -> bool:
+                    return allow
+
+                confirm = _safety_gate
             shell_result = run_shell_command(
                 command,
-                confirm_dangerous=self._shell_confirm,
+                confirm_dangerous=confirm,
                 timeout_seconds=self._shell_timeout_seconds,
             )
             return ModeResult(
@@ -83,6 +100,7 @@ class InputModeRouter:
                 references=[],
                 shell_result=shell_result,
                 command_result=None,
+                shell_warning=warning,
             )
 
         parse_result: ReferenceParseResult = parse_references(text)
@@ -92,4 +110,5 @@ class InputModeRouter:
             references=[match.canonical for match in parse_result.matches],
             shell_result=None,
             command_result=None,
+            shell_warning=None,
         )

src/cleveragents/tui/input/shell_exec.py

@@ -55,17 +55,20 @@ def run_shell_command(
             stdout="",
             stderr="shell mode is disabled",
         )
-    if looks_dangerous(command):
-        confirmed = False
-        if confirm_dangerous is not None:
-            confirmed = confirm_dangerous(command)
-        if not confirmed:
-            return ShellResult(
-                command=command,
-                exit_code=1,
-                stdout="",
-                stderr="blocked dangerous shell command",
-            )
+    if confirm_dangerous is not None and not confirm_dangerous(command):
+        return ShellResult(
+            command=command,
+            exit_code=1,
+            stdout="",
+            stderr="blocked by shell safety policy",
+        )
+    if looks_dangerous(command) and confirm_dangerous is None:
+        return ShellResult(
+            command=command,
+            exit_code=1,
+            stdout="",
+            stderr="blocked dangerous shell command",
+        )
     try:
         proc = subprocess.run(
             command,

src/cleveragents/tui/widgets/prompt.py

@@ -248,12 +248,20 @@ class _TextualPromptInput(_PromptSymbolMixin, _HorizontalBase):
 
 
 class _FallbackPromptInput(_PromptSymbolMixin):
-    """Fallback prompt input used when Textual is unavailable."""
+    """Fallback prompt input used when Textual is unavailable.
+
+    Mirrors the Textual Widget CSS-class API (``add_class`` /
+    ``remove_class`` / ``has_class``) so callers that toggle styling
+    classes on the outer prompt (e.g. ``add_class("dangerous")`` from
+    the shell-safety surfacing path) work uniformly across production
+    and fallback/mock paths.
+    """
 
     def __init__(self, placeholder: str = "", **_: object) -> None:
         self.placeholder = placeholder
         self._input = cast(_MutableValueInput, _InputBase())
         self._current_symbol = _PROMPT_SYMBOLS[InputMode.NORMAL]
+        self._classes: set[str] = set()
         self._update_symbol(self._input.value)
 
     @property
@@ -270,6 +278,15 @@ class _FallbackPromptInput(_PromptSymbolMixin):
         if callable(focus):
             focus()
 
+    def add_class(self, name: str) -> None:
+        self._classes.add(name)
+
+    def remove_class(self, name: str) -> None:
+        self._classes.discard(name)
+
+    def has_class(self, name: str) -> bool:
+        return name in self._classes
+
     def _apply_symbol(self, symbol: str) -> None:
         self._current_symbol = symbol