feat: Add inline python steering handler and enhance dangerous command protection for project root integrity

Author: Mateusz

config/schemas/tool_call_reactor_config.schema.yaml

@@ -20,6 +20,12 @@ properties:
     type:
       - string
       - "null"
+  inline_python_steering_enabled:
+    type: boolean
+  inline_python_steering_message:
+    type:
+      - string
+      - "null"
   test_execution_reminder_enabled:
     type: boolean
   test_execution_reminder_message:

data/gemini_oauth_request_count.json

@@ -0,0 +1,5 @@
+{
+  "count": 66,
+  "last_reset_date": "2025-12-07",
+  "logged_thresholds": []
+}

docs/user_guide/features/dangerous-command-protection.md

@@ -126,5 +126,6 @@ Only disable this protection if you:
 ## Related Features
 
 - [Tool Access Control](tool-access-control.md) - Fine-grained control over tool execution
-- [File Access Sandboxing](file-access-sandboxing.md) - Restrict file operations to project directory
+- [File Access Sandboxing](file-sandboxing.md) - Restrict file operations to project directory
+- [Inline Python Steering](inline-python-steering.md) - Prevent unstable inline Python execution
 - [Angel Verification System](angel-verification.md) - Real-time response verification

docs/user_guide/features/file-sandboxing.md

@@ -353,6 +353,8 @@ Priority level: **80** (runs before most handlers, after authentication)
 ## Related Documentation
 
 - [Tool Call Reactor System](./tool-call-reactor.md)
+- [Dangerous Command Protection](dangerous-command-protection.md)
+- [Inline Python Steering](inline-python-steering.md)
 - [Security Best Practices](../security/best-practices.md)
 - [Configuration Guide](../configuration.md)
 - [CLI Parameters](../cli-parameters.md)

docs/user_guide/features/inline-python-steering.md

@@ -0,0 +1,133 @@
+# Inline Python Steering
+
+Prevent unstable inline Python execution by steering agents toward script-based execution.
+
+## Overview
+
+The Inline Python Steering feature intercepts attempts to run inline Python code via shell commands (e.g., `python -c "..."`) and guides the agent to use temporary scripts instead. Inline Python code execution in terminals is often unstable, prone to quoting issues, and difficult to debug. By enforcing script usage, this feature improves reliability and maintainability of agent-generated code execution.
+
+When an agent attempts to execute inline Python, the proxy blocks the call and returns a steering message explaining the issue and suggesting the creation of a temporary script.
+
+## Key Features
+
+- **Automatic Detection**: Recognizes various inline Python patterns (`python -c`, `python3 -c`, etc.)
+- **Immediate Blocking**: Prevents the command from executing on the host
+- **Helpful Steering**: Explains why the command was blocked and what to do instead
+- **Robust Parsing**: Handles various python executables and flags
+- **Tool Awareness**: Monitors known shell execution tools
+
+## Configuration
+
+The feature is disabled by default and can be enabled via environment variable or YAML configuration.
+
+### Environment Variable
+
+```bash
+export INLINE_PYTHON_STEERING_ENABLED=true
+```
+
+### YAML Configuration
+
+```yaml
+session:
+  inline_python_steering_enabled: true  # Default: false
+  
+  # Optional: Custom steering message
+  inline_python_steering_message: |
+    Inline Python execution is blocked for stability reasons.
+    Please write your code to a temporary file (e.g., script.py) and execute that file instead.
+```
+
+## Usage Examples
+
+### Enable with Default Message
+
+Set the environment variable before running the proxy:
+
+```bash
+export INLINE_PYTHON_STEERING_ENABLED=true
+.venv/Scripts/python.exe -m src.core.cli --default-backend openai
+```
+
+### Enable with Custom Message
+
+Create `config/my_config.yaml`:
+
+```yaml
+session:
+  inline_python_steering_enabled: true
+  inline_python_steering_message: |
+    [Security/Stability Notice]
+    You are attempting to run inline Python code (python -c).
+    This pattern is unreliable in this environment.
+    
+    Please:
+    1. Create a file named 'temp_script.py' with your code
+    2. Run 'python temp_script.py'
+```
+
+Then run:
+
+```bash
+.venv/Scripts/python.exe -m src.core.cli --config config/my_config.yaml
+```
+
+## Detection Logic
+
+The handler detects commands matching the following patterns:
+
+- `python -c "..."`
+- `python3 -c '...'`
+- `python.exe -c "..."`
+- `python -u -c "..."` (with flags)
+
+It specifically looks for the `-c` flag combined with a Python executable. Normal Python file execution (e.g., `python script.py` or `python -m pytest`) is **allowed**.
+
+## Recognized Shell Tools
+
+The handler monitors the following shell execution tools (from `ShellExecutionTools` constant):
+
+- `bash`
+- `Execute`
+- `ShellTool`
+- `exec_command`
+- `execute_command`
+- `run_shell_command`
+- `run_terminal_command`
+- `shell`
+- `local_shell`
+- `container.exec`
+
+## Rationale
+
+Why block inline Python?
+
+1.  **Quoting Hell**: Passing complex Python code inside shell strings often leads to escaping issues, especially on Windows (cmd.exe vs PowerShell) and with nested quotes.
+2.  **Terminal Stability**: Long or complex inline one-liners can behave unpredictably depending on the underlying shell.
+3.  **Debuggability**: Code in a file is easier to review, debug, and log than a ephemeral one-liner.
+4.  **Agent Behavior**: Encouraging agents to write scripts fosters better coding habits and more robust solutions.
+
+## Behavior Flow
+
+1.  **Agent Action**: Agent calls a shell tool with `python -c "print('hello')"`
+2.  **Interception**: The `InlinePythonSteeringHandler` detects the pattern.
+3.  **Blocking**: The tool call is swallow (not executed).
+4.  **Response**: The agent receives the steering message (default or custom).
+5.  **Correction**: The agent should then create a file and execute it, which is allowed.
+
+## Troubleshooting
+
+**Feature not working:**
+- Verify `INLINE_PYTHON_STEERING_ENABLED` is set to `true`.
+- Ensure the tool being used is one of the recognized shell tools.
+
+**False Positives:**
+- The regex is designed to be specific to `-c`. If you find valid commands being blocked, please report them.
+- Normal `python filename.py` should never be blocked.
+
+## Implementation References
+
+- **Handler**: `src/core/services/tool_call_handlers/inline_python_steering_handler.py`
+- **Tests**: `tests/unit/core/services/tool_call_handlers/test_inline_python_steering_handler.py`
+- **Configuration**: `src/core/config/app_config.py`
+- **DI Registration**: `src/core/di/services.py`

scripts/analyze_stream_field.py

@@ -0,0 +1,90 @@
+"""
+Analyze CBOR wire capture to examine PROXY_TO_BACKEND request payloads in detail.
+"""
+import cbor2
+import json
+import zlib
+from pathlib import Path
+
+# Direction constants
+DIRECTION_PROXY_TO_BACKEND = 2
+
+
+def main() -> None:
+    path = Path("var/wire_captures_cbor/proxy-20251209_1017.cbor")
+    if not path.exists():
+        print(f"File not found: {path}")
+        return
+
+    # Load using same method as inspect script
+    entries = []
+    with open(path, "rb") as f:
+        header = cbor2.load(f)
+        while True:
+            try:
+                entry = cbor2.load(f)
+                # Handle decompression
+                if entry.get("enc") == "zlib":
+                    entry["data"] = zlib.decompress(entry["data"])
+                    del entry["enc"]
+                entries.append(entry)
+            except (EOFError, cbor2.CBORDecodeEOF):
+                break
+    
+    print(f"Session ID: {header.get('session_id', 'N/A')}")
+    print(f"Total entries: {len(entries)}")
+    print()
+    
+    # Look at all PROXY_TO_BACKEND entries
+    proxy_to_backend = [e for e in entries if e.get("dir") == DIRECTION_PROXY_TO_BACKEND]
+    print(f"Found {len(proxy_to_backend)} PROXY_TO_BACKEND entries")
+    print("=" * 100)
+    
+    for i, entry in enumerate(proxy_to_backend):
+        seq = entry.get("seq", i)
+        meta = entry.get("meta", {})
+        data = entry.get("data", b"")
+        backend = meta.get("be", "N/A")
+        session_id = meta.get("sid", "N/A")[:16] if meta.get("sid") else "N/A"
+        
+        print(f"\n[{seq}] Backend: {backend} | Session: {session_id}")
+        print(f"Data size: {len(data)} bytes")
+        
+        if isinstance(data, (bytes, bytearray)):
+            try:
+                text = data.decode("utf-8", errors="ignore")
+            except Exception:
+                print("  (Could not decode)")
+                continue
+        elif isinstance(data, str):
+            text = data
+        else:
+            print(f"  (Unexpected data type: {type(data)})")
+            continue
+        
+        # Try to parse as JSON
+        text = text.strip()
+        if text.startswith("{"):
+            try:
+                obj = json.loads(text)
+                # Print key fields
+                print(f"  model: {obj.get('model', 'N/A')}")
+                print(f"  stream: {obj.get('stream', '(NOT PRESENT)')}")
+                print(f"  messages count: {len(obj.get('messages', []))}")
+                
+                # Show first 200 chars of the text
+                preview = text[:300]
+                if len(text) > 300:
+                    preview += "..."
+                print(f"  Preview: {preview}")
+            except json.JSONDecodeError as e:
+                print(f"  JSON parse error: {e}")
+                print(f"  First 200 chars: {text[:200]}")
+        else:
+            print(f"  Not JSON. First 200 chars: {text[:200]}")
+        
+        print()
+
+
+if __name__ == "__main__":
+    main()

src/core/di/services.py

@@ -2150,6 +2150,38 @@ def _tool_call_reactor_factory(
                         f"Failed to register PytestFullSuiteHandler: {e}", exc_info=True
                     )
 
+            # Register InlinePythonSteeringHandler if enabled
+            try:
+                if getattr(reactor_config, "inline_python_steering_enabled", False):
+                    from src.core.services.tool_call_handlers.inline_python_steering_handler import (
+                        InlinePythonSteeringHandler,
+                    )
+
+                    steering_message = getattr(
+                        reactor_config, "inline_python_steering_message", None
+                    )
+                    inline_python_handler = InlinePythonSteeringHandler(
+                        message=steering_message,
+                        enabled=True,
+                    )
+                    try:
+                        reactor.register_handler_sync(inline_python_handler)
+                        if logger.isEnabledFor(logging.INFO):
+                            logger.info(
+                                "Registered InlinePythonSteeringHandler with priority 95"
+                            )
+                    except Exception as e:
+                        if logger.isEnabledFor(logging.WARNING):
+                            logger.warning(
+                                f"Failed to register inline python steering handler: {e}",
+                                exc_info=True,
+                            )
+            except Exception as e:
+                if logger.isEnabledFor(logging.WARNING):
+                    logger.warning(
+                        f"Failed to register InlinePythonSteeringHandler: {e}", exc_info=True
+                    )
+
             # Register PytestContextSavingHandler if enabled
             try:
                 if getattr(reactor_config, "pytest_context_saving_enabled", False):

src/core/domain/configuration/dangerous_command_config.py

@@ -2,6 +2,8 @@
 from re import Pattern
 from typing import NamedTuple
 
+from src.core.domain.tool_constants import ShellExecutionTools
+
 
 class DangerousCommandRule(NamedTuple):
     pattern: Pattern[str]
@@ -282,17 +284,6 @@ def get_default_dangerous_command_rules() -> list[DangerousCommandRule]:
 DEFAULT_DANGEROUS_COMMAND_RULES = get_default_dangerous_command_rules()
 
 DEFAULT_DANGEROUS_COMMAND_CONFIG = DangerousCommandConfig(
-    tool_names=[
-        "bash",
-        "Execute",
-        "ShellTool",
-        "exec_command",
-        "execute_command",
-        "run_shell_command",
-        "run_terminal_command",
-        "shell",
-        "local_shell",
-        "container.exec",
-    ],
+    tool_names=ShellExecutionTools.get_all(),
     rules=DEFAULT_DANGEROUS_COMMAND_RULES,
 )