feat(cli): incognito shell command mode (#3252)

Closes #2091

---

Adds an incognito local shell mode for `!!` in the interactive CLI.
Existing `!` commands still follow the normal shell path, while `!!`
commands run locally and render their command and output as app-level
transcript entries without adding user or assistant records to the
model-visible conversation.

The input parser now detects the longest mode prefix so `!!` survives
alongside `!`, history recall restores the right mode, queued and sent
messages render the correct shell glyph, and shell-mode typing can be
upgraded to incognito by entering a second `!`.

The UI distinguishes incognito shell mode with a dedicated theme color,
input border title, message border, and `SHELL` status badge styling.
Help text and startup tips document the new prefix, and focused unit
coverage exercises prefix parsing, input and history behavior, queue
draining, shell routing, local-only output, rendering, status, theme
variables, help text, and tips.

---------

Co-authored-by: Mason Daugherty <mason@langchain.dev>
Co-authored-by: Mason Daugherty <github@mdrxy.com>
This commit is contained in:
Miguel Mongiardim
2026-05-10 22:44:48 +01:00
committed by GitHub
parent d7e229fa53
commit 56aee504b5
17 changed files with 736 additions and 62 deletions
+112 -15
View File
@@ -340,7 +340,7 @@ def _format_model_params(extra_kwargs: dict[str, Any] | None) -> str:
return f" with model params {json.dumps(extra_kwargs, sort_keys=True)}"
InputMode = Literal["normal", "shell", "command"]
InputMode = Literal["normal", "shell", "shell_incognito", "command"]
_TYPING_IDLE_THRESHOLD_SECONDS: float = 2.0
"""Seconds since the last keystroke after which the user is considered idle and
@@ -1224,8 +1224,9 @@ class DeepAgentsApp(App):
Most styling uses Textual's built-in variables (`$primary`,
`$text-muted`, `$error-muted`, etc.). This override injects the
app-specific variables (`$mode-bash`, `$mode-command`, `$skill`,
`$skill-hover`, `$tool`, `$tool-hover`) that have no Textual equivalent.
app-specific variables (`$mode-bash`, `$mode-command`,
`$mode-incognito`, `$skill`, `$skill-hover`, `$tool`, `$tool-hover`)
that have no Textual equivalent.
Returns:
Dict of CSS variable names to hex color values.
@@ -3300,15 +3301,82 @@ class DeepAgentsApp(App):
value: The message text to process.
mode: The input mode that determines message routing.
"""
if mode == "shell":
await self._handle_shell_command(value.removeprefix("!"))
if mode == "shell_incognito":
await self._handle_shell_command(
self._strip_mode_value(value, "!!", "!", mode), incognito=True
)
elif mode == "shell":
await self._handle_shell_command(
self._strip_mode_value(value, "!", "!!", mode)
)
elif mode == "command":
await self._handle_command(value)
elif mode == "normal":
await self._handle_user_message(value)
else:
logger.warning("Unrecognized input mode %r, treating as normal", mode)
await self._handle_user_message(value)
# Fail safe: never default to the agent dispatch path on an
# unrecognized mode, since that would silently leak `!!`/`!`
# prefixed text to the LLM if the mode literal is ever wrong.
logger.error(
"Unrecognized input mode %r; refusing to forward to agent", mode
)
await self._mount_message(
ErrorMessage(
f"Internal error: unknown input mode {mode!r}. "
"Message was not sent."
)
)
@staticmethod
def _strip_mode_value(
value: str, prefix: str, conflicting_prefix: str, mode: InputMode
) -> str:
"""Strip `prefix` from `value`, logging if a wrong prefix was supplied.
Three submission paths feed `_process_message`: (1) typed input, where
the chat input has already stripped the prefix, so `value` does not
start with `prefix`; (2) re-submission via the queue, where the value
was re-prepended with `prefix`; and (3) external/programmatic callers,
which may send either form. `removeprefix` is a no-op for path (1) and
does the work for paths (2) and (3).
A leading `conflicting_prefix` (the sibling shell mode's trigger)
indicates state-machine drift between the declared `mode` and the
actual text for example, mode `"shell_incognito"` paired with a
value starting with a single `!`. We log for diagnostics but still
strip `prefix` so the user is not surprised by a sudden refusal; the
sibling prefix becomes part of the command body and the shell will
report any resulting error locally.
Examples:
shell_incognito + `"!!ls"` -> `"ls"` (queued submission)
shell_incognito + `"ls"` -> `"ls"` (typed submission, prefix
already stripped)
shell_incognito + `"!ls"` -> `"!ls"` (drift; logs a warning,
shell sees `!ls`)
shell + `"!ls"` -> `"ls"`
shell + `"!!ls"` -> `"!ls"` (drift; logs a warning)
Args:
value: Submitted text expected to match `mode`.
prefix: Trigger prefix associated with `mode` (e.g. `"!!"` for
`shell_incognito`, `"!"` for `shell`).
conflicting_prefix: Sibling-mode prefix whose presence at the
start of `value` signals drift (e.g. pass `"!"` when
`prefix="!!"`).
mode: Input mode for diagnostic messages.
Returns:
`value` with a leading `prefix` removed if present, otherwise
`value` unchanged.
"""
if value.startswith(conflicting_prefix) and not value.startswith(prefix):
logger.warning(
"Mode %r received value with conflicting prefix %r",
mode,
conflicting_prefix,
)
return value.removeprefix(prefix)
def _has_initial_submission(self) -> bool:
"""Return whether startup should auto-submit a prompt or skill."""
@@ -3975,7 +4043,12 @@ class DeepAgentsApp(App):
if self._chat_input:
self.call_after_refresh(self._chat_input.focus_input)
async def _handle_shell_command(self, command: str) -> None:
async def _handle_shell_command(
self,
command: str,
*,
incognito: bool = False,
) -> None:
"""Handle a shell command (! prefix).
Thin dispatcher that mounts the user message and spawns a worker
@@ -3983,19 +4056,21 @@ class DeepAgentsApp(App):
Args:
command: The shell command to execute.
incognito: Whether the command/output should remain local-only.
"""
await self._mount_message(UserMessage(f"!{command}"))
if not incognito:
await self._mount_message(UserMessage(f"!{command}"))
self._shell_running = True
if self._chat_input:
self._chat_input.set_cursor_active(active=False)
self._shell_worker = self.run_worker(
self._run_shell_task(command),
self._run_shell_task(command, incognito=incognito),
exclusive=False,
)
async def _run_shell_task(self, command: str) -> None:
async def _run_shell_task(self, command: str, *, incognito: bool = False) -> None:
"""Run a shell command in a background worker.
This mirrors `_run_agent_task`: running in a worker keeps the event
@@ -4004,6 +4079,7 @@ class DeepAgentsApp(App):
Args:
command: The shell command to execute.
incognito: Whether the command/output should remain local-only.
Raises:
CancelledError: If the command is interrupted by the user.
@@ -4042,9 +4118,14 @@ class DeepAgentsApp(App):
output += f"\n[stderr]\n{stderr_text}"
if output:
msg = AssistantMessage(f"```\n{output}\n```")
await self._mount_message(msg)
await msg.write_initial_content()
if incognito:
await self._mount_message(
AppMessage(f"```\n{output}\n```", markdown=True)
)
else:
msg = AssistantMessage(f"```\n{output}\n```")
await self._mount_message(msg)
await msg.write_initial_content()
else:
await self._mount_message(AppMessage("Command completed (no output)"))
@@ -4059,6 +4140,20 @@ class DeepAgentsApp(App):
logger.exception("Failed to execute shell command: %s", command)
err_msg = f"Failed to run command: {e}"
await self._mount_message(ErrorMessage(err_msg))
except Exception:
# Defense in depth: a crash between subprocess read and
# `_mount_message` could leave the user with no signal that the
# command ran at all (privacy-sensitive in the incognito path).
# Surface a local-only error and re-raise so the worker layer
# records the failure.
logger.exception(
"Shell task crashed (incognito=%s): %s", incognito, command
)
with suppress(Exception):
await self._mount_message(
ErrorMessage("Shell command crashed; see logs.")
)
raise
finally:
await self._cleanup_shell_task(refresh_git_branch=not refresh_started)
@@ -4313,7 +4408,9 @@ class DeepAgentsApp(App):
" Shift+Tab Toggle auto-approve mode\n"
" @filename Auto-complete files and inject content\n"
" /command Slash commands (/help, /clear, /quit)\n"
" !command Run shell commands directly\n\n"
" !command Run shell commands directly\n"
" !!command Run shell commands without adding "
"command/output to model context\n\n"
"Docs: "
)
help_text = Content.assemble(
+8
View File
@@ -152,6 +152,10 @@ UserMessage.-mode-shell {
border-left: wide $mode-bash;
}
UserMessage.-mode-shell-incognito {
border-left: wide $mode-incognito;
}
UserMessage.-mode-command {
border-left: wide $mode-command;
}
@@ -165,6 +169,10 @@ UserMessage.-ascii.-mode-shell {
border-left: ascii $mode-bash;
}
UserMessage.-ascii.-mode-shell-incognito {
border-left: ascii $mode-incognito;
}
UserMessage.-ascii.-mode-command {
border-left: ascii $mode-command;
}
+29 -2
View File
@@ -245,12 +245,14 @@ if TYPE_CHECKING:
console: Console
MODE_PREFIXES: dict[str, str] = {
"shell_incognito": "!!",
"shell": "!",
"command": "/",
}
"""Maps each non-normal mode to its trigger character."""
MODE_DISPLAY_GLYPHS: dict[str, str] = {
"shell_incognito": "$",
"shell": "$",
"command": "/",
}
@@ -265,8 +267,33 @@ if MODE_PREFIXES.keys() != MODE_DISPLAY_GLYPHS.keys():
)
raise ValueError(msg)
PREFIX_TO_MODE: dict[str, str] = {v: k for k, v in MODE_PREFIXES.items()}
"""Reverse lookup: trigger character -> mode name."""
_MODE_PREFIXES_BY_LENGTH: tuple[tuple[str, str], ...] = tuple(
sorted(MODE_PREFIXES.items(), key=lambda item: len(item[1]), reverse=True)
)
"""Mode entries ordered longest-prefix-first.
Pre-sorted at import so `detect_mode_prefix` runs in constant time per
keystroke without re-sorting.
"""
def detect_mode_prefix(text: str) -> tuple[str, str] | None:
"""Return the longest mode prefix and mode for `text`, if any.
Longer prefixes win so multi-character triggers like `!!` are matched
before their single-character prefixes (`!`).
Args:
text: Input text that may start with a mode trigger.
Returns:
Tuple of `(prefix, mode)` for the longest matching trigger, otherwise
`None`.
"""
for mode, prefix in _MODE_PREFIXES_BY_LENGTH:
if text.startswith(prefix):
return prefix, mode
return None
class CharsetMode(StrEnum):
+25 -11
View File
@@ -5,8 +5,9 @@ Single source of truth for color values used in Python code (Rich markup,
Textual CSS variables: built-in variables
(`$primary`, `$background`, `$text-muted`, `$error-muted`, etc.) are set via
`register_theme()` in `DeepAgentsApp.__init__`, while the few app-specific
variables (`$mode-bash`, `$mode-command`, `$skill`, `$skill-hover`, `$tool`,
`$tool-hover`) are backed by these constants via `App.get_theme_variable_defaults()`.
variables (`$mode-bash`, `$mode-command`, `$mode-incognito`, `$skill`,
`$skill-hover`, `$tool`, `$tool-hover`) are backed by these constants via
`App.get_theme_variable_defaults()`.
Code that needs custom CSS variable values should call
`get_css_variable_defaults(dark=...)`. For the full semantic color palette, look
@@ -94,6 +95,10 @@ LC_TOOL = LC_AMBER
LC_TOOL_HOVER = "#FFCB91"
"""Tool call hover — lighter variant for interactive feedback."""
LC_INCOGNITO = "#2DD4BF"
"""Incognito shell accent — teal, must read distinctly against `LC_PINK`
shell and `error` warning border colors."""
# ---------------------------------------------------------------------------
# Brand palette — light
@@ -152,6 +157,9 @@ LC_LIGHT_TOOL = LC_LIGHT_AMBER
LC_LIGHT_TOOL_HOVER = "#78350F"
"""Tool call hover (darkened for light bg contrast)."""
LC_LIGHT_INCOGNITO = "#0F766E"
"""Incognito shell accent (darkened for light bg contrast)."""
# ---------------------------------------------------------------------------
# Semantic constants (ANSI color names for Rich console output)
@@ -273,6 +281,9 @@ class ThemeColors:
mode_command: str
"""Command mode indicator — borders, prompts, and message prefixes."""
mode_incognito: str
"""Incognito shell indicator — borders, prompts, and message prefixes."""
skill: str
"""Skill invocation accent — border and header text."""
@@ -346,6 +357,7 @@ DARK_COLORS = ThemeColors(
muted=LC_MUTED,
mode_bash=LC_PINK,
mode_command=LC_PURPLE,
mode_incognito=LC_INCOGNITO,
skill=LC_SKILL,
skill_hover=LC_SKILL_HOVER,
tool=LC_TOOL,
@@ -367,6 +379,7 @@ LIGHT_COLORS = ThemeColors(
muted=LC_LIGHT_MUTED,
mode_bash=LC_LIGHT_PINK,
mode_command=LC_LIGHT_PURPLE,
mode_incognito=LC_LIGHT_INCOGNITO,
skill=LC_LIGHT_SKILL,
skill_hover=LC_LIGHT_SKILL_HOVER,
tool=LC_LIGHT_TOOL,
@@ -439,8 +452,9 @@ def _builtin_themes() -> dict[str, ThemeEntry]:
newly shipped Textual themes appear automatically. They are not registered
via `register_theme()` — Textual's own `$primary`, `$background`, etc.
apply. The `colors` field provides fallback values for app-specific CSS
vars (`$mode-bash`, `$mode-command`) and Python-side styling. For standard
properties (primary, secondary, etc.), `get_theme_colors()` dynamically
vars (`$mode-bash`, `$mode-command`, `$mode-incognito`) and Python-side
styling. For standard properties (primary, secondary, etc.),
`get_theme_colors()` dynamically
resolves from the actual Textual theme at runtime so the Python and CSS
color systems stay in sync.
@@ -726,6 +740,7 @@ def get_css_variable_defaults(
return {
"mode-bash": c.mode_bash,
"mode-command": c.mode_command,
"mode-incognito": c.mode_incognito,
"skill": c.skill,
"skill-hover": c.skill_hover,
"tool": c.tool,
@@ -753,13 +768,11 @@ def _colors_from_textual_theme(app: object) -> ThemeColors:
"""Construct `ThemeColors` from the app's active Textual theme.
Reads standard properties (primary, secondary, etc.) from the resolved
theme so Python-side styling matches CSS. `muted` falls back to the
dark/light base unconditionally (no Textual equivalent).
`mode_bash` is derived from the theme's `error` color, and `mode_command`
from `secondary`, falling back to the base palette when non-hex.
Non-hex values (e.g. `ansi_blue` in the ANSI theme) are detected and fall
back to the base palette automatically.
theme so Python-side styling matches CSS. `muted` and `mode_incognito`
have no Textual equivalent and always source from the dark/light base
palette. `mode_bash` is derived from the theme's `error` color and
`mode_command` from `secondary`, both falling back to the base palette
when non-hex values (e.g. `ansi_blue` in the ANSI theme) are detected.
Args:
app: The Textual App instance.
@@ -797,6 +810,7 @@ def _colors_from_textual_theme(app: object) -> ThemeColors:
muted=base.muted,
mode_bash=_hex_or(ct.error, base.mode_bash),
mode_command=_hex_or(ct.secondary, base.mode_command),
mode_incognito=base.mode_incognito,
# No Textual equivalent — always use base palette.
skill=base.skill,
skill_hover=base.skill_hover,
+65 -16
View File
@@ -26,7 +26,7 @@ from deepagents_cli.command_registry import SLASH_COMMANDS, CommandEntry
from deepagents_cli.config import (
MODE_DISPLAY_GLYPHS,
MODE_PREFIXES,
PREFIX_TO_MODE,
detect_mode_prefix,
is_ascii_mode,
)
from deepagents_cli.input import IMAGE_PLACEHOLDER_PATTERN, VIDEO_PLACEHOLDER_PATTERN
@@ -1003,6 +1003,12 @@ class ChatInput(Vertical):
border: solid $mode-command;
}
ChatInput.mode-shell-incognito {
border: solid $mode-incognito;
border-title-color: $mode-incognito;
border-title-style: bold;
}
ChatInput .input-row {
height: auto;
width: 100%;
@@ -1024,6 +1030,10 @@ class ChatInput(Vertical):
color: $mode-command;
}
ChatInput.mode-shell-incognito .input-prompt {
color: $mode-incognito;
}
ChatInput ChatTextArea {
width: 1fr;
height: auto;
@@ -1257,8 +1267,21 @@ class ChatInput(Vertical):
# a prefix character.
if self._stripping_prefix:
self._stripping_prefix = False
elif text and text[0] in PREFIX_TO_MODE:
if text[0] == "/" and is_path_payload:
elif detected_prefix := detect_mode_prefix(text):
prefix, detected = detected_prefix
strip_length = len(prefix)
if self.mode == "shell" and detected == "shell":
# First `!` was stripped on entry to shell mode, so the
# currently-visible `!` is the second bang of `!!`. Promote to
# incognito and consume it.
detected = "shell_incognito"
elif self.mode == "shell_incognito" and detected == "shell":
# Already in incognito; an extra `!` is part of the command
# body. Skip the strip-and-demote path that would otherwise
# drop us back to plain shell mode.
detected = "shell_incognito"
strip_length = 0
if prefix == "/" and is_path_payload:
# Absolute dropped paths stay normal input, not slash-command mode.
if self.mode != "normal":
self.mode = "normal"
@@ -1269,10 +1292,10 @@ class ChatInput(Vertical):
# text that re-includes the trigger character. The
# _stripping_prefix guard prevents the resulting change event
# from looping back here.
detected = PREFIX_TO_MODE[text[0]]
if self.mode != detected:
self.mode = detected
self._strip_mode_prefix()
if strip_length:
self._strip_mode_prefix(strip_length)
# Fall through to update completion suggestions in the same
# refresh cycle as the mode/glyph change rather than waiting
# for the next text-change event caused by the prefix strip.
@@ -1397,11 +1420,15 @@ class ChatInput(Vertical):
return self._is_existing_path_payload(candidate)
return False
def _strip_mode_prefix(self) -> None:
"""Remove the first character (mode trigger) from the text area.
def _strip_mode_prefix(self, length: int = 1) -> None:
"""Remove the mode trigger from the text area.
Sets the `_stripping_prefix` guard so the resulting text-change event is
not misinterpreted as new input.
Args:
length: Number of leading characters to strip (matches the trigger
length detected by `detect_mode_prefix`).
"""
if not self._text_area:
return
@@ -1415,9 +1442,9 @@ class ChatInput(Vertical):
return
row, col = self._text_area.cursor_location
self._stripping_prefix = True
self._text_area.text = text[1:]
self._text_area.text = text[length:]
if row == 0 and col > 0:
col -= 1
col = max(0, col - length)
self._text_area.move_cursor((row, col))
def _completion_text_and_cursor(self) -> tuple[str, int]:
@@ -1783,10 +1810,9 @@ class ChatInput(Vertical):
Tuple of `(mode, display_text)` where mode-trigger prefixes are
removed from `display_text`.
"""
for prefix, mode in PREFIX_TO_MODE.items():
# Small dict; loop is fine. No need to over-engineer right now
if entry.startswith(prefix):
return mode, entry[len(prefix) :]
if mode_match := detect_mode_prefix(entry):
prefix, mode = mode_match
return mode, entry[len(prefix) :]
return "normal", entry
async def on_key(self, event: events.Key) -> None:
@@ -1881,15 +1907,38 @@ class ChatInput(Vertical):
)
def _apply() -> None:
self.remove_class("mode-shell", "mode-command")
self.remove_class("mode-shell", "mode-command", "mode-shell-incognito")
if glyph:
self.add_class(f"mode-{mode}")
class_name = (
"mode-shell-incognito"
if mode == "shell_incognito"
else f"mode-{mode}"
)
self.add_class(class_name)
try:
prompt = self.query_one("#prompt", Static)
except NoMatches:
logger.warning("watch_mode._apply: #prompt widget not found")
logger.warning("watch_mode._apply: prompt widget not found")
if mode == "shell_incognito":
# Privacy-sensitive: surface a visible warning so the user
# never types an incognito command without confirmation
# that the mode is active.
app = getattr(self, "app", None)
if app is not None:
with contextlib.suppress(Exception):
app.notify(
"Incognito mode UI failed to render; "
"switching back to normal input.",
severity="warning",
markup=False,
)
self.mode = "normal"
return
prompt.update(glyph or ">")
if mode == "shell_incognito":
self.border_title = "incognito"
else:
self.border_title = None
self.call_after_refresh(_apply)
self.post_message(self.ModeChanged(mode))
+17 -12
View File
@@ -21,7 +21,7 @@ from textual.widgets import Static
from deepagents_cli import theme
from deepagents_cli.config import (
MODE_DISPLAY_GLYPHS,
PREFIX_TO_MODE,
detect_mode_prefix,
get_glyphs,
is_ascii_mode,
)
@@ -96,6 +96,8 @@ def _mode_color(mode: str | None, widget_or_app: object | None = None) -> str:
colors = theme.get_theme_colors(widget_or_app)
if not mode:
return colors.primary
if mode == "shell_incognito":
return colors.mode_incognito
if mode == "shell":
return colors.mode_bash
if mode == "command":
@@ -188,9 +190,10 @@ class UserMessage(_TimestampClickMixin, Static):
def on_mount(self) -> None:
"""Add CSS classes for mode-specific border and ASCII border type."""
mode = PREFIX_TO_MODE.get(self._content[:1]) if self._content else None
if mode:
self.add_class(f"-mode-{mode}")
mode_match = detect_mode_prefix(self._content)
if mode_match:
_prefix, mode = mode_match
self.add_class(f"-mode-{mode.replace('_', '-')}")
if is_ascii_mode():
self.add_class("-ascii")
@@ -207,11 +210,12 @@ class UserMessage(_TimestampClickMixin, Static):
# Use mode-specific prefix indicator when content starts with a
# mode trigger character (e.g. "!" for shell, "/" for commands).
# The display glyph may differ from the trigger (e.g. "$" for shell).
mode = PREFIX_TO_MODE.get(content[:1]) if content else None
if mode:
glyph = MODE_DISPLAY_GLYPHS.get(mode, content[0])
mode_match = detect_mode_prefix(content)
if mode_match:
prefix_text, mode = mode_match
glyph = MODE_DISPLAY_GLYPHS.get(mode, prefix_text[0])
parts.append((f"{glyph} ", f"bold {_mode_color(mode, self)}"))
content = content[1:]
content = content[len(prefix_text) :]
else:
parts.append(("> ", f"bold {colors.primary}"))
@@ -288,11 +292,12 @@ class QueuedUserMessage(Static):
"""
colors = theme.get_theme_colors(self)
content = self._content
mode = PREFIX_TO_MODE.get(content[:1]) if content else None
if mode:
glyph = MODE_DISPLAY_GLYPHS.get(mode, content[0])
mode_match = detect_mode_prefix(content)
if mode_match:
prefix_text, mode = mode_match
glyph = MODE_DISPLAY_GLYPHS.get(mode, prefix_text[0])
prefix = (f"{glyph} ", f"bold {colors.muted}")
content = content[1:]
content = content[len(prefix_text) :]
else:
prefix = ("> ", f"bold {colors.muted}")
return Content.assemble(prefix, (content, colors.muted))
+13 -4
View File
@@ -124,6 +124,12 @@ class StatusBar(Horizontal):
color: white;
}
StatusBar .status-mode.shell-incognito {
background: $mode-incognito;
color: $background;
text-style: bold;
}
StatusBar .status-auto-approve {
width: auto;
padding: 0 1;
@@ -212,7 +218,7 @@ class StatusBar(Horizontal):
"""
yield Static("", classes="status-mode normal", id="mode-indicator")
yield Static(
"manual | shift+tab to cycle",
"manual",
classes="status-auto-approve off",
id="auto-approve-indicator",
)
@@ -270,11 +276,14 @@ class StatusBar(Horizontal):
indicator = self.query_one("#mode-indicator", Static)
except NoMatches:
return
indicator.remove_class("normal", "shell", "command")
indicator.remove_class("normal", "shell", "command", "shell-incognito")
if mode == "shell":
indicator.update("SHELL")
indicator.add_class("shell")
elif mode == "shell_incognito":
indicator.update("SHELL")
indicator.add_class("shell-incognito")
elif mode == "command":
indicator.update("CMD")
indicator.add_class("command")
@@ -291,10 +300,10 @@ class StatusBar(Horizontal):
indicator.remove_class("on", "off")
if new_value:
indicator.update("auto | shift+tab to cycle")
indicator.update("auto")
indicator.add_class("on")
else:
indicator.update("manual | shift+tab to cycle")
indicator.update("manual")
indicator.add_class("off")
def watch_cwd(self, new_value: str) -> None:
@@ -53,7 +53,9 @@ _TIPS: dict[str, int] = {
"Use /auto-update to toggle automatic CLI updates": 1,
"Use /agents to browse and switch between your available agents": 2,
"In /agents, press Ctrl+S to set the highlighted agent as your default": 1,
"Press Shift+Tab to toggle auto-approve mode": 2,
"Use --startup-cmd to run a shell command before the first prompt": 1,
"Use !! for incognito shell commands that stay out of model context": 1,
"Run `deepagents mcp login <server>` to authorize a remote MCP server": 1,
"Deep Agents can explain its own features and look up its docs. Ask it how to use.": 3, # noqa: E501
}
+2 -2
View File
@@ -78,8 +78,8 @@ def _register_theme_variables(monkeypatch: pytest.MonkeyPatch) -> None:
Production code defines these in `DeepAgentsApp.get_theme_variable_defaults`
but many tests use lightweight `App[None]` subclasses that lack the override.
Patching the base class ensures `$mode-bash`, `$mode-command` resolve
everywhere without requiring each test app to opt in.
Patching the base class ensures custom mode variables resolve everywhere
without requiring each test app to opt in.
"""
from textual.app import App
+187
View File
@@ -1721,6 +1721,28 @@ class TestMessageQueue:
assert len(widgets) == 1
assert len(app._queued_widgets) == 1
async def test_queued_incognito_shell_preserves_mode_on_drain(self) -> None:
"""Queued incognito shell commands should drain as incognito shell."""
app = DeepAgentsApp()
async with app.run_test() as pilot:
await pilot.pause()
app._agent_running = True
app.post_message(ChatInput.Submitted("!!echo hidden", "shell_incognito"))
await pilot.pause()
assert len(app._pending_messages) == 1
assert app._pending_messages[0].text == "!!echo hidden"
assert app._pending_messages[0].mode == "shell_incognito"
handler = AsyncMock()
app._handle_shell_command = handler # type: ignore[method-assign]
app._agent_running = False
await app._process_next_from_queue()
handler.assert_awaited_once_with("echo hidden", incognito=True)
async def test_immediate_processing_when_agent_idle(self) -> None:
"""Messages should process immediately when agent is not running."""
app = DeepAgentsApp()
@@ -3266,6 +3288,40 @@ class TestShellCommandInterrupt:
error_msgs = app.query(ErrorMessage)
assert any("timed out" in w._content for w in error_msgs)
async def test_incognito_timeout_feedback_is_not_model_visible(self) -> None:
"""Incognito timeout feedback should stay out of user/assistant records."""
from deepagents_cli.widgets.message_store import MessageType
app = DeepAgentsApp()
mock_proc = AsyncMock()
mock_proc.communicate = AsyncMock(side_effect=asyncio.TimeoutError)
mock_proc.returncode = None
mock_proc.pid = 12345
mock_proc.wait = AsyncMock()
mock_proc.terminate = MagicMock()
mock_proc.kill = MagicMock()
async with app.run_test() as pilot:
await pilot.pause()
with patch(
"asyncio.create_subprocess_shell",
return_value=mock_proc,
):
await app._run_shell_task("echo secret", incognito=True)
await pilot.pause()
messages = app._message_store.get_all_messages()
assert any(
msg.type == MessageType.ERROR and "timed out" in msg.content
for msg in messages
)
assert not any(
msg.type in {MessageType.USER, MessageType.ASSISTANT}
and "secret" in msg.content
for msg in messages
)
async def test_posix_killpg_called(self) -> None:
"""On POSIX, _kill_shell_process should use os.killpg with SIGTERM."""
app = DeepAgentsApp()
@@ -3360,6 +3416,137 @@ class TestShellCommandInterrupt:
coro = mock_rw.call_args[0][0]
coro.close()
async def test_process_message_routes_incognito_shell_command(self) -> None:
"""`shell_incognito` mode should strip `!!` and mark the shell run."""
app = DeepAgentsApp()
handler = AsyncMock()
app._handle_shell_command = handler # type: ignore[method-assign]
await app._process_message("!!echo secret", "shell_incognito")
handler.assert_awaited_once_with("echo secret", incognito=True)
async def test_incognito_shell_command_does_not_mount_header(self) -> None:
"""Incognito shell commands should not echo the command before output."""
app = DeepAgentsApp()
async with app.run_test() as pilot:
await pilot.pause()
with patch.object(app, "run_worker") as mock_rw:
mock_rw.return_value = MagicMock()
await app._handle_shell_command("echo secret", incognito=True)
messages = app._message_store.get_all_messages()
assert not any(
"incognito shell command" in msg.content or "echo secret" in msg.content
for msg in messages
)
# Close the unawaited coroutine to suppress RuntimeWarning.
coro = mock_rw.call_args[0][0]
coro.close()
async def test_incognito_shell_output_is_app_message(self) -> None:
"""Incognito shell output should avoid assistant transcript records."""
from deepagents_cli.widgets.message_store import MessageType
app = DeepAgentsApp()
mock_proc = AsyncMock()
mock_proc.communicate = AsyncMock(return_value=(b"secret\n", b""))
mock_proc.returncode = 0
mock_proc.pid = 12345
async with app.run_test() as pilot:
await pilot.pause()
app._schedule_git_branch_refresh = MagicMock() # type: ignore[assignment]
app._maybe_drain_deferred = AsyncMock() # type: ignore[assignment]
app._process_next_from_queue = AsyncMock() # type: ignore[assignment]
with (
patch(
"asyncio.create_subprocess_shell",
return_value=mock_proc,
),
patch(
"deepagents_cli.app.AssistantMessage.write_initial_content",
new=AsyncMock(),
) as write_mock,
):
await app._run_shell_task("echo secret", incognito=True)
await pilot.pause()
messages = app._message_store.get_all_messages()
assert any(
msg.type == MessageType.APP and "secret" in msg.content for msg in messages
)
assert not any(
msg.type in {MessageType.USER, MessageType.ASSISTANT}
and "secret" in msg.content
for msg in messages
)
write_mock.assert_not_awaited()
async def test_incognito_nonzero_exit_keeps_stderr_out_of_model(self) -> None:
"""A failing incognito command must not leak stderr to model records."""
from deepagents_cli.widgets.message_store import MessageType
app = DeepAgentsApp()
mock_proc = AsyncMock()
mock_proc.communicate = AsyncMock(return_value=(b"", b"secret leak"))
mock_proc.returncode = 1
mock_proc.pid = 12345
async with app.run_test() as pilot:
await pilot.pause()
app._schedule_git_branch_refresh = MagicMock() # type: ignore[assignment]
app._maybe_drain_deferred = AsyncMock() # type: ignore[assignment]
app._process_next_from_queue = AsyncMock() # type: ignore[assignment]
with patch(
"asyncio.create_subprocess_shell",
return_value=mock_proc,
):
await app._run_shell_task("falsey", incognito=True)
await pilot.pause()
messages = app._message_store.get_all_messages()
assert not any(
msg.type in {MessageType.USER, MessageType.ASSISTANT}
and "secret leak" in msg.content
for msg in messages
)
async def test_unknown_input_mode_does_not_dispatch_to_agent(self) -> None:
"""An unrecognized mode must surface an error rather than reach the LLM.
Regression guard for the privacy invariant: a typo or stale mode
literal must never silently fall through to `_handle_user_message`.
"""
from deepagents_cli.widgets.message_store import MessageType
app = DeepAgentsApp()
user_handler = AsyncMock()
shell_handler = AsyncMock()
app._handle_user_message = user_handler # type: ignore[method-assign]
app._handle_shell_command = shell_handler # type: ignore[method-assign]
async with app.run_test() as pilot:
await pilot.pause()
await app._process_message("!!echo secret", "shell_incognto") # type: ignore[arg-type]
await pilot.pause()
user_handler.assert_not_awaited()
shell_handler.assert_not_awaited()
messages = app._message_store.get_all_messages()
assert any(
msg.type == MessageType.ERROR and "unknown input mode" in msg.content
for msg in messages
)
async def test_kill_noop_when_already_exited(self) -> None:
"""_kill_shell_process should no-op if process already exited."""
app = DeepAgentsApp()
@@ -312,6 +312,41 @@ class TestPromptIndicator:
assert _prompt_text(prompt) == "$"
assert chat_input.has_class("mode-shell")
async def test_prompt_shows_shell_style_in_incognito_shell_mode(self) -> None:
"""Incognito shell mode sets the `$` prompt, border title, and class."""
app = _ChatInputTestApp()
async with app.run_test() as pilot:
chat_input = app.query_one(ChatInput)
prompt = chat_input.query_one("#prompt", Static)
chat_input.mode = "shell_incognito"
await pilot.pause()
assert _prompt_text(prompt) == "$"
assert chat_input.border_title == "incognito"
assert chat_input.has_class("mode-shell-incognito")
async def test_incognito_shell_to_shell_clears_incognito_styling(self) -> None:
"""Transitioning out of incognito must clear the incognito styling.
Regression guard: a future change forgetting to drop the incognito
title or CSS class would leave stale styling on the input.
"""
app = _ChatInputTestApp()
async with app.run_test() as pilot:
chat_input = app.query_one(ChatInput)
chat_input.mode = "shell_incognito"
await pilot.pause()
assert chat_input.border_title == "incognito"
assert chat_input.has_class("mode-shell-incognito")
chat_input.mode = "shell"
await pilot.pause()
assert chat_input.border_title is None
assert not chat_input.has_class("mode-shell-incognito")
assert chat_input.has_class("mode-shell")
async def test_prompt_shows_slash_in_command_mode(self) -> None:
"""Setting mode to 'command' should change prompt and styling."""
app = _ChatInputTestApp()
@@ -339,6 +374,7 @@ class TestPromptIndicator:
chat_input.mode = "normal"
await pilot.pause()
assert _prompt_text(prompt) == ">"
assert chat_input.border_title is None
assert not chat_input.has_class("mode-shell")
assert not chat_input.has_class("mode-command")
@@ -1015,6 +1051,103 @@ class TestModePrefixStripping:
assert app.submitted[0].value == "!ls"
assert app.submitted[0].mode == "shell"
async def test_submission_prepends_incognito_shell_prefix(self) -> None:
"""Submitting in incognito shell mode should preserve the `'!!'` prefix."""
app = _RecordingApp()
async with app.run_test() as pilot:
chat = app.query_one(ChatInput)
assert chat._text_area is not None
chat._text_area.text = "!!pwd"
await _pause_for_strip(pilot)
assert chat.mode == "shell_incognito"
assert chat._text_area.text == "pwd"
await pilot.press("enter")
await pilot.pause()
assert len(app.submitted) == 1
assert app.submitted[0].value == "!!pwd"
assert app.submitted[0].mode == "shell_incognito"
async def test_typing_second_bang_enters_incognito_shell_mode(self) -> None:
"""Typing `!!pwd` as separate keypresses should submit incognito shell."""
app = _RecordingApp()
async with app.run_test() as pilot:
chat = app.query_one(ChatInput)
assert chat._text_area is not None
await pilot.press("!")
await _pause_for_strip(pilot)
assert chat.mode == "shell"
assert chat._text_area.text == ""
await pilot.press("!")
await _pause_for_strip(pilot)
assert chat.mode == "shell_incognito"
assert chat._text_area.text == ""
chat._text_area.insert("pwd")
await pilot.pause()
await pilot.press("enter")
await pilot.pause()
assert len(app.submitted) == 1
assert app.submitted[0].value == "!!pwd"
assert app.submitted[0].mode == "shell_incognito"
async def test_third_bang_stays_in_incognito_shell_mode(self) -> None:
"""Typing `!`+`!`+`!` must not demote `shell_incognito` back to `shell`.
Regression guard for the privacy-sensitive parser path: a stray third
bang should be treated as command-body content, not as a mode change
out of incognito.
"""
app = _RecordingApp()
async with app.run_test() as pilot:
chat = app.query_one(ChatInput)
assert chat._text_area is not None
await pilot.press("!")
await _pause_for_strip(pilot)
await pilot.press("!")
await _pause_for_strip(pilot)
assert chat.mode == "shell_incognito"
await pilot.press("!")
await _pause_for_strip(pilot)
assert chat.mode == "shell_incognito"
assert chat._text_area.text == "!"
chat._text_area.insert("ls")
await pilot.pause()
await pilot.press("enter")
await pilot.pause()
assert len(app.submitted) == 1
assert app.submitted[0].mode == "shell_incognito"
assert app.submitted[0].value == "!!!ls"
async def test_pasted_three_bangs_routes_to_incognito(self) -> None:
"""Pasting `!!!ls` must enter `shell_incognito` with body `!ls`."""
app = _RecordingApp()
async with app.run_test() as pilot:
chat = app.query_one(ChatInput)
assert chat._text_area is not None
chat._text_area.text = "!!!ls"
await _pause_for_strip(pilot)
assert chat.mode == "shell_incognito"
assert chat._text_area.text == "!ls"
await pilot.press("enter")
await pilot.pause()
assert len(app.submitted) == 1
assert app.submitted[0].mode == "shell_incognito"
assert app.submitted[0].value == "!!!ls"
async def test_submission_prepends_command_prefix(self) -> None:
"""Submitting in command mode should prepend `'/'` to the value."""
app = _RecordingApp()
@@ -1382,6 +1515,28 @@ class TestHistorySlashPrefixRecall:
assert app.submitted[0].value == "/help"
assert app.submitted[0].mode == "command"
async def test_history_incognito_shell_entry_enters_incognito_mode(self) -> None:
"""Recalling a `!!` history entry should enter incognito shell mode."""
app = _RecordingApp()
async with app.run_test() as pilot:
chat = app.query_one(ChatInput)
assert chat._text_area is not None
chat._history._entries.append("!!pwd")
await pilot.press("up")
await _pause_for_strip(pilot)
assert chat.mode == "shell_incognito"
assert chat._text_area.text == "pwd"
await pilot.press("enter")
await pilot.pause()
assert len(app.submitted) == 1
assert app.submitted[0].value == "!!pwd"
assert app.submitted[0].mode == "shell_incognito"
class TestCompletionIndexToTextIndex:
"""Edge-case tests for _completion_index_to_text_index clamping."""
@@ -196,3 +196,26 @@ class TestHelpBodyDrift:
f"Commands in /help body but missing from COMMANDS: {extra}\n"
"Remove them from help_body or add to COMMANDS in command_registry.py."
)
def test_help_body_describes_incognito_shell_prefix(self) -> None:
"""The `/help` body should document local-only incognito shell mode."""
app_src = (
Path(__file__).resolve().parents[2] / "deepagents_cli" / "app.py"
).read_text()
# Locate the Interactive Features block where the `!!` row lives.
match = re.search(
r'"Interactive Features:\\n"(.*?)"\s*Docs:',
app_src,
re.DOTALL,
)
assert match, "Could not locate Interactive Features section in help_body"
section = match.group(1)
assert "!!command" in section, "Help body must show `!!command` literal"
# Concept-level checks rather than exact wording — independent of
# whether the sentence reads "command/output to model context" or
# "output and command to model context".
assert "model context" in section
assert "command" in section
assert "output" in section
+37
View File
@@ -20,6 +20,7 @@ from deepagents_cli.config import (
_get_provider_kwargs,
build_langsmith_thread_url,
create_model,
detect_mode_prefix,
detect_provider,
fetch_langsmith_project_url,
get_langsmith_project_name,
@@ -2965,3 +2966,39 @@ class TestFindDotenvFromStartPath:
# Should continue past the OSError and find .env in parent
assert result == env_file
class TestDetectModePrefix:
"""Tests for `detect_mode_prefix`.
This helper is the linchpin for routing typed prefixes to the correct
mode. The longest-prefix-first invariant is critical: if `!!` ever loses
to `!`, every `!!` command would silently route as a single-bang shell
command and leak content to the model.
"""
@pytest.mark.parametrize(
("text", "expected"),
[
("!!ls", ("!!", "shell_incognito")),
("!!", ("!!", "shell_incognito")),
("!!!ls", ("!!", "shell_incognito")),
("!ls", ("!", "shell")),
("!", ("!", "shell")),
("/help", ("/", "command")),
("/", ("/", "command")),
],
)
def test_matches_known_prefixes(self, text: str, expected: tuple[str, str]) -> None:
assert detect_mode_prefix(text) == expected
@pytest.mark.parametrize(
"text",
["", "ls", "echo hi", " !!ls", "\t!ls", "hello !!", "x!"],
)
def test_no_match_for_non_prefixed(self, text: str) -> None:
assert detect_mode_prefix(text) is None
def test_double_bang_wins_over_single_bang(self) -> None:
"""Regression guard: `!!` must beat `!` even if iteration order changes."""
assert detect_mode_prefix("!!whoami") == ("!!", "shell_incognito")
@@ -636,6 +636,13 @@ class TestUserMessageModeRendering:
first_span = content._spans[0]
assert theme.DARK_COLORS.mode_bash in str(first_span.style)
def test_incognito_shell_prefix_renders_dollar_indicator(self) -> None:
"""`UserMessage('!!ls')` should strip the full incognito prefix."""
content = _render_content(UserMessage("!!ls"))
assert content.plain == "$ ls"
first_span = content._spans[0]
assert theme.DARK_COLORS.mode_incognito in str(first_span.style)
def test_command_prefix_renders_slash_indicator(self) -> None:
"""`UserMessage('/help')` should render with `'/ '` prefix and body."""
content = _render_content(UserMessage("/help"))
@@ -679,6 +686,11 @@ class TestQueuedUserMessageModeRendering:
content = _render_content(QueuedUserMessage("!ls"))
assert content.plain == "$ ls"
def test_incognito_shell_prefix_renders_dimmed_dollar(self) -> None:
"""`QueuedUserMessage('!!ls')` should strip the full incognito prefix."""
content = _render_content(QueuedUserMessage("!!ls"))
assert content.plain == "$ ls"
def test_command_prefix_renders_dimmed_slash(self) -> None:
"""`QueuedUserMessage('/help')` should render dimmed `'/ '` prefix."""
content = _render_content(QueuedUserMessage("/help"))
+41
View File
@@ -290,6 +290,47 @@ class TestTokenDisplay:
assert "+" not in rendered
class TestModeIndicator:
"""Tests for the input-mode indicator in the status bar."""
async def test_incognito_shell_mode_shows_indicator(self) -> None:
"""Incognito shell mode renders the SHELL indicator with its own class."""
async with StatusBarApp().run_test() as pilot:
bar = pilot.app.query_one("#status-bar", StatusBar)
indicator = pilot.app.query_one("#mode-indicator")
bar.set_mode("shell_incognito")
await pilot.pause()
assert str(indicator.render()) == "SHELL"
assert indicator.has_class("shell-incognito")
async def test_mode_transition_clears_incognito_class(self) -> None:
"""Leaving `shell_incognito` must remove the badge class.
Regression guard: a future change forgetting to clear
`shell-incognito` on transition would leak the badge across modes.
"""
async with StatusBarApp().run_test() as pilot:
bar = pilot.app.query_one("#status-bar", StatusBar)
indicator = pilot.app.query_one("#mode-indicator")
bar.set_mode("shell_incognito")
await pilot.pause()
assert indicator.has_class("shell-incognito")
bar.set_mode("normal")
await pilot.pause()
assert not indicator.has_class("shell-incognito")
bar.set_mode("shell_incognito")
await pilot.pause()
bar.set_mode("shell")
await pilot.pause()
assert not indicator.has_class("shell-incognito")
assert indicator.has_class("shell")
class TestModelLabelPrefixStripping:
"""Tests for provider-specific model prefix stripping in ModelLabel."""
+4
View File
@@ -170,6 +170,7 @@ EXPECTED_CSS_KEYS = frozenset(
{
"mode-bash",
"mode-command",
"mode-incognito",
"skill",
"skill-hover",
"tool",
@@ -188,14 +189,17 @@ class TestGetCssVariableDefaults:
def test_dark_mode_uses_dark_colors(self) -> None:
result = get_css_variable_defaults(dark=True)
assert result["mode-bash"] == DARK_COLORS.mode_bash
assert result["mode-incognito"] == DARK_COLORS.mode_incognito
def test_light_mode_uses_light_colors(self) -> None:
result = get_css_variable_defaults(dark=False)
assert result["mode-bash"] == LIGHT_COLORS.mode_bash
assert result["mode-incognito"] == LIGHT_COLORS.mode_incognito
def test_explicit_colors_take_precedence(self) -> None:
result = get_css_variable_defaults(dark=True, colors=LIGHT_COLORS)
assert result["mode-bash"] == LIGHT_COLORS.mode_bash
assert result["mode-incognito"] == LIGHT_COLORS.mode_incognito
def test_all_values_are_hex_colors(self) -> None:
import re
@@ -426,6 +426,10 @@ class TestBuildWelcomeFooter:
"""New `--startup-cmd` flag must have a discoverability tip."""
assert any("--startup-cmd" in tip for tip in _TIPS)
def test_incognito_shell_tip_registered(self) -> None:
"""New `!!` shell mode must have a discoverability tip."""
assert any("!!" in tip and "incognito" in tip.lower() for tip in _TIPS)
def test_copy_command_tip_registered(self) -> None:
"""The `/copy` command must have a discoverability tip."""
assert "Use /copy to copy the latest assistant message" in _TIPS