mirror of
https://github.com/langchain-ai/deepagents.git
synced 2026-07-22 01:35:28 -04:00
feat(code): in-app Debug Console (#4564)
Deep Agents Code now includes an in-app Debug Console. Press `Ctrl+\` to inspect session details and filter, view, or copy recent application logs without enabling file-based debug logging. --- Debugging client-side TUI problems currently requires restarting with file logging enabled and tailing a log in another terminal. This adds an in-app Debug Console, modeled on Mistral Vibe's console, so users can inspect session state and recent application logs without leaving the TUI. Open the console with `Ctrl+\` or the hidden `/debug` command. It shows a point-in-time snapshot of the version, model, thread, working directory, auto-approve state, sandbox, MCP servers, token usage, and debug-log path. Below that, it tails recent `deepagents_code.*` records from an always-on bounded in-memory buffer and supports filtering by severity. The buffer captures `INFO` and above by default, or the level selected with `DEEPAGENTS_CODE_LOG_LEVEL`. `DEEPAGENTS_CODE_DEBUG` continues to enable append-only file logging and defaults capture to `DEBUG`. `Ctrl+L` clears the current console view without clearing the underlying buffer; `c` copies the visible filtered records retained since the last clear, and clicking a record copies only that record. --------- Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>
This commit is contained in:
Mason Daugherty
+2
-1
@@ -91,7 +91,8 @@ Debug logging is configured **once**, on the `deepagents_code` package logger, b
|
||||
|
||||
- Do **not** add per-module `configure_debug_logging(logger)` calls. They are redundant now that the package logger is configured at import, and they reintroduce the duplicate-handler problem the single-config approach solves.
|
||||
- Every module should create its logger with `logging.getLogger(__name__)` so it stays inside the `deepagents_code.*` hierarchy and inherits the package handler. Don't set `logger.propagate = False` or attach your own handlers.
|
||||
- The handler only attaches when `DEEPAGENTS_CODE_DEBUG` is truthy; the no-op path is a single env-var read, so it's safe on the startup hot path. See `DEVELOPMENT.md` for the `DEEPAGENTS_CODE_DEBUG` / `DEEPAGENTS_CODE_DEBUG_FILE` env vars.
|
||||
- The **file** handler only attaches when `DEEPAGENTS_CODE_DEBUG` is truthy; that path is a single env-var read, so it's safe on the startup hot path. See `DEVELOPMENT.md` for the `DEEPAGENTS_CODE_DEBUG` / `DEEPAGENTS_CODE_DEBUG_FILE` / `DEEPAGENTS_CODE_LOG_LEVEL` env vars.
|
||||
- Separately, an **always-on in-memory ring buffer** (`_debug_buffer.install_log_buffer`, called from `__init__.py` right before `configure_debug_logging`) attaches unconditionally at import so the in-app Debug Console (`Ctrl+\`) can tail recent `deepagents_code.*` records without file logging. It captures `INFO` and above by default (or `DEEPAGENTS_CODE_LOG_LEVEL`), and may lower the package logger to `INFO` for the process lifetime. It never spills to the terminal (a handler is always found, so `lastResort` is skipped) and is bounded (a `deque` of 1000 records), so it's cheap enough to keep on. Because it installs *before* `configure_debug_logging`, warnings that helper emits at startup are captured and visible in the console.
|
||||
|
||||
## CLI help screen
|
||||
|
||||
|
||||
@@ -45,8 +45,9 @@ aliases, descriptions, visibility, or hidden-command metadata.
|
||||
| `/update` | | Check for and install updates |
|
||||
| `/version` | `/about` | Show version information |
|
||||
|
||||
## Hidden (1)
|
||||
## Hidden (2)
|
||||
|
||||
These commands are intentionally omitted from autocomplete and help. See the `HIDDEN_COMMANDS` docstring in the registry for context.
|
||||
|
||||
- `/debug`
|
||||
- `/debug-error`
|
||||
|
||||
@@ -117,6 +117,12 @@ tail -f /tmp/deepagents_debug.log
|
||||
|
||||
To send it elsewhere, also `export DEEPAGENTS_CODE_DEBUG_FILE=<path>`. The handler appends across runs, so a single file accumulates every session.
|
||||
|
||||
### In-app Debug Console (`Ctrl+\`)
|
||||
|
||||
Press `Ctrl+\` (or run the hidden `/debug` command) inside a session to toggle a read-only Debug Console overlay. It shows a point-in-time session/runtime snapshot (version, model, thread, cwd, auto-approve, sandbox, MCP servers, token usage, debug-log path) plus a live tail of recent `deepagents_code.*` log records.
|
||||
|
||||
The tail is fed by an always-on in-memory ring buffer (`_debug_buffer.install_log_buffer`, installed on the package logger in `__init__.py`), so it works **without** `DEEPAGENTS_CODE_DEBUG` — though enabling that switch raises the captured level to `DEBUG` and adds the file handler above. In the console, `Ctrl+L` clears the on-screen view (the buffer keeps accruing) and `c` copies the visible lines.
|
||||
|
||||
## Local dev installs
|
||||
|
||||
A *local dev install* gives you a persistent `dcode-dev` command that launches your checkout directly. It lives in a dedicated editable venv under `~/.local/share/dcode-dev`, symlinked into `~/.local/bin/dcode-dev`. It can sit alongside a released `dcode` without interfering:
|
||||
|
||||
@@ -6,12 +6,14 @@ import logging
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from deepagents_code._debug import configure_debug_logging
|
||||
from deepagents_code._debug_buffer import install_log_buffer
|
||||
from deepagents_code._version import __version__
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Callable
|
||||
|
||||
configure_debug_logging(logging.getLogger(__name__)) # noqa: RUF067 # package logger must be configured before child modules emit logs
|
||||
install_log_buffer(logging.getLogger(__name__)) # noqa: RUF067 # attach the always-on tail first so warnings from configure_debug_logging are captured
|
||||
configure_debug_logging(logging.getLogger(__name__)) # noqa: RUF067 # package logger must be configured before child modules emit logs; sets the final level over the buffer's INFO floor
|
||||
|
||||
__all__ = [
|
||||
"__version__",
|
||||
|
||||
@@ -1,9 +1,9 @@
|
||||
"""Shared debug-logging configuration for verbose file-based tracing.
|
||||
"""Shared debug-logging configuration for runtime and file-based tracing.
|
||||
|
||||
When the `DEEPAGENTS_CODE_DEBUG` environment variable is set, modules that handle
|
||||
streaming or remote communication can enable detailed file-based logging. This
|
||||
helper centralizes the setup so the env-var name, file path, and format are
|
||||
defined in one place.
|
||||
helper centralizes the setup so the env-var names, file path, log level, and
|
||||
format are defined in one place.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -17,32 +17,85 @@ from deepagents_code._env_vars import (
|
||||
DEBUG,
|
||||
DEBUG_FILE,
|
||||
DEFAULT_DEBUG_FILE,
|
||||
LOG_LEVEL,
|
||||
is_env_truthy,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_DEBUG_HANDLER_ATTR = "_deepagents_code_debug_handler"
|
||||
LOG_LEVELS = {
|
||||
"DEBUG": logging.DEBUG,
|
||||
"INFO": logging.INFO,
|
||||
"WARNING": logging.WARNING,
|
||||
"ERROR": logging.ERROR,
|
||||
"CRITICAL": logging.CRITICAL,
|
||||
}
|
||||
"""Canonical level-name to `logging` level mapping.
|
||||
|
||||
The single source of truth for level names and their numeric values, shared with
|
||||
the Debug Console's level filter so severity ordering is never re-derived from
|
||||
hardcoded integers.
|
||||
"""
|
||||
|
||||
|
||||
def resolve_log_level(*, debug_enabled: bool | None = None) -> int:
|
||||
"""Resolve the configured runtime logging level.
|
||||
|
||||
Args:
|
||||
debug_enabled: Whether `DEEPAGENTS_CODE_DEBUG` is truthy. When omitted,
|
||||
the current environment is checked.
|
||||
|
||||
Returns:
|
||||
A standard `logging` level integer. Defaults to `DEBUG` when debug file
|
||||
logging is enabled and `INFO` otherwise.
|
||||
"""
|
||||
if debug_enabled is None:
|
||||
debug_enabled = is_env_truthy(DEBUG)
|
||||
fallback = logging.DEBUG if debug_enabled else logging.INFO
|
||||
raw = os.environ.get(LOG_LEVEL)
|
||||
if raw is None or not raw.strip():
|
||||
return fallback
|
||||
level = LOG_LEVELS.get(raw.strip().upper())
|
||||
if level is not None:
|
||||
return level
|
||||
valid = ", ".join(LOG_LEVELS)
|
||||
message = f"ignoring invalid {LOG_LEVEL}={raw!r}; expected one of {valid}"
|
||||
# stderr for headless / pre-TUI visibility; the logger so it also lands in
|
||||
# the always-on in-memory buffer and surfaces in the Debug Console (the
|
||||
# buffer is installed before this runs; see __init__.py).
|
||||
print(f"Warning: {message}", file=sys.stderr) # noqa: T201
|
||||
logger.warning("%s", message)
|
||||
return fallback
|
||||
|
||||
|
||||
def configure_debug_logging(target: logging.Logger) -> None:
|
||||
"""Attach a file handler to *target* when `DEEPAGENTS_CODE_DEBUG` is set.
|
||||
"""Configure runtime log level and optional file logging for *target*.
|
||||
|
||||
Intended to be called once on the `deepagents_code` package logger; child
|
||||
module loggers reach the same file via propagation, so individual modules do
|
||||
not configure logging themselves.
|
||||
module loggers reach the same handlers via propagation, so individual modules
|
||||
do not configure logging themselves.
|
||||
|
||||
The log file defaults to `DEFAULT_DEBUG_FILE` but can be overridden with
|
||||
`DEEPAGENTS_CODE_DEBUG_FILE`. The handler appends (`mode='a'`) so logs
|
||||
are preserved across separate process runs. Calling this again with the same
|
||||
resolved path is a no-op: the existing tagged handler is reused rather than
|
||||
stacking duplicates. If the resolved path changes, the stale handler is
|
||||
closed and replaced.
|
||||
`DEEPAGENTS_CODE_LOG_LEVEL` controls the package logger level independently of
|
||||
file logging. If it is unset, `DEEPAGENTS_CODE_DEBUG=1` defaults to `DEBUG`;
|
||||
otherwise the runtime level defaults to `INFO`.
|
||||
|
||||
Does nothing when `DEEPAGENTS_CODE_DEBUG` is not truthy (see `is_env_truthy`).
|
||||
When `DEEPAGENTS_CODE_DEBUG` is truthy, a file handler is attached. The log
|
||||
file defaults to `DEFAULT_DEBUG_FILE` but can be overridden with
|
||||
`DEEPAGENTS_CODE_DEBUG_FILE`. The handler appends (`mode='a'`) so logs are
|
||||
preserved across separate process runs. Calling this again with the same
|
||||
resolved path does not stack duplicate handlers: the existing tagged handler
|
||||
is reused and its level re-applied. If the resolved path changes, the stale
|
||||
handler is closed and replaced.
|
||||
|
||||
Args:
|
||||
target: Logger to configure.
|
||||
"""
|
||||
if not is_env_truthy(DEBUG):
|
||||
debug_enabled = is_env_truthy(DEBUG)
|
||||
level = resolve_log_level(debug_enabled=debug_enabled)
|
||||
target.setLevel(level)
|
||||
|
||||
if not debug_enabled:
|
||||
return
|
||||
|
||||
debug_path = Path(os.environ.get(DEBUG_FILE, DEFAULT_DEBUG_FILE))
|
||||
@@ -53,8 +106,7 @@ def configure_debug_logging(target: logging.Logger) -> None:
|
||||
):
|
||||
continue
|
||||
if Path(existing.baseFilename) == debug_path:
|
||||
# Already configured for this path; reuse rather than duplicate.
|
||||
target.setLevel(logging.DEBUG)
|
||||
existing.setLevel(level)
|
||||
return
|
||||
# The debug path changed; drop the stale handler before re-attaching so
|
||||
# we don't leak its file descriptor or fan logs out to two files.
|
||||
@@ -64,16 +116,16 @@ def configure_debug_logging(target: logging.Logger) -> None:
|
||||
try:
|
||||
handler = logging.FileHandler(str(debug_path), mode="a")
|
||||
except OSError as exc:
|
||||
print( # noqa: T201
|
||||
f"Warning: could not open debug log file {debug_path}: {exc}",
|
||||
file=sys.stderr,
|
||||
)
|
||||
message = f"could not open debug log file {debug_path}: {exc}"
|
||||
# stderr for headless / pre-TUI visibility; the logger so it also lands
|
||||
# in the always-on in-memory buffer and surfaces in the Debug Console.
|
||||
print(f"Warning: {message}", file=sys.stderr) # noqa: T201
|
||||
logger.warning("%s", message)
|
||||
return
|
||||
setattr(handler, _DEBUG_HANDLER_ATTR, True)
|
||||
handler.setLevel(logging.DEBUG)
|
||||
handler.setLevel(level)
|
||||
handler.setFormatter(logging.Formatter("%(asctime)s %(name)s %(message)s"))
|
||||
target.addHandler(handler)
|
||||
target.setLevel(logging.DEBUG)
|
||||
|
||||
|
||||
def installed_debug_log_path() -> Path | None:
|
||||
|
||||
@@ -0,0 +1,204 @@
|
||||
r"""In-memory ring buffer of recent log records for the in-app Debug Console.
|
||||
|
||||
A lightweight `logging.Handler` keeps the most recent structured log records in a
|
||||
bounded `deque` so the Debug Console (`Ctrl+\`) can show a live tail without
|
||||
requiring the opt-in file logging from `_debug.configure_debug_logging`. The
|
||||
handler is installed once on the `deepagents_code` package logger (see
|
||||
`__init__.py`); child module loggers reach it via propagation.
|
||||
|
||||
Installation is negligible, and each emitted record only appends a structured
|
||||
record to a bounded `deque`, so the buffer is cheap enough to keep always on.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import itertools
|
||||
import logging
|
||||
import os
|
||||
from collections import deque
|
||||
from dataclasses import dataclass
|
||||
|
||||
from deepagents_code._env_vars import LOG_LEVEL
|
||||
|
||||
DEFAULT_CAPACITY = 1000
|
||||
"""Maximum number of records retained in the ring buffer."""
|
||||
|
||||
_DATE_FORMAT = "%H:%M:%S"
|
||||
_FORMATTER = logging.Formatter(datefmt=_DATE_FORMAT)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class InMemoryLogRecord:
|
||||
"""Structured log record retained by the in-memory debug buffer."""
|
||||
|
||||
timestamp: str
|
||||
level: str
|
||||
levelno: int
|
||||
logger: str
|
||||
message: str
|
||||
|
||||
@property
|
||||
def plain_line(self) -> str:
|
||||
"""The record in the legacy plain-text debug-console format."""
|
||||
return f"{self.timestamp} {self.level} {self.logger} {self.message}"
|
||||
|
||||
|
||||
class InMemoryLogBuffer(logging.Handler):
|
||||
"""Logging handler retaining the most recent structured records in memory."""
|
||||
|
||||
def __init__(self, capacity: int = DEFAULT_CAPACITY) -> None:
|
||||
"""Create the handler with a bounded backing `deque`.
|
||||
|
||||
Args:
|
||||
capacity: Maximum records to retain; oldest are dropped first.
|
||||
|
||||
Raises:
|
||||
ValueError: If *capacity* is less than 1. A zero-capacity buffer
|
||||
would silently discard every record while `total_emitted` kept
|
||||
climbing, so it is rejected at the boundary.
|
||||
"""
|
||||
if capacity < 1:
|
||||
msg = f"capacity must be >= 1, got {capacity}"
|
||||
raise ValueError(msg)
|
||||
super().__init__()
|
||||
self._records: deque[InMemoryLogRecord] = deque(maxlen=capacity)
|
||||
self._total = 0
|
||||
|
||||
def emit(self, record: logging.LogRecord) -> None:
|
||||
"""Append the structured *record* to the ring buffer."""
|
||||
self.acquire()
|
||||
try:
|
||||
self._records.append(self._make_record(record))
|
||||
self._total += 1
|
||||
except Exception: # noqa: BLE001 # never let logging crash the app
|
||||
self.handleError(record)
|
||||
finally:
|
||||
self.release()
|
||||
|
||||
@property
|
||||
def total_emitted(self) -> int:
|
||||
"""Total records ever emitted (monotonic; survives buffer eviction)."""
|
||||
self.acquire()
|
||||
try:
|
||||
return self._total
|
||||
finally:
|
||||
self.release()
|
||||
|
||||
def snapshot_records_since(self, index: int) -> tuple[list[InMemoryLogRecord], int]:
|
||||
"""Return retained structured records and the next absolute index.
|
||||
|
||||
Absolute indices are stable even as old records are evicted, so callers
|
||||
can poll incrementally without re-reading records they already consumed.
|
||||
Returning the records and the resume index together under the handler
|
||||
lock prevents a concurrent append from being skipped between separate
|
||||
reads.
|
||||
|
||||
Args:
|
||||
index: Absolute emission index to start from.
|
||||
|
||||
Returns:
|
||||
A tuple of the structured records from *index* onward that are still
|
||||
retained, and the current total emitted count to resume from.
|
||||
"""
|
||||
self.acquire()
|
||||
try:
|
||||
return self._records_since_unlocked(index), self._total
|
||||
finally:
|
||||
self.release()
|
||||
|
||||
def _records_since_unlocked(self, index: int) -> list[InMemoryLogRecord]:
|
||||
"""Return retained records from *index* while the handler lock is held."""
|
||||
start_abs = self._total - len(self._records)
|
||||
offset = index - start_abs
|
||||
if offset <= 0:
|
||||
return list(self._records)
|
||||
if offset >= len(self._records):
|
||||
return []
|
||||
# islice avoids materializing the full deque just to slice off the tail.
|
||||
return list(itertools.islice(self._records, offset, None))
|
||||
|
||||
@staticmethod
|
||||
def _make_record(record: logging.LogRecord) -> InMemoryLogRecord:
|
||||
"""Convert a standard logging record into a structured debug record.
|
||||
|
||||
Returns:
|
||||
Structured record for display and filtering.
|
||||
"""
|
||||
message = record.getMessage()
|
||||
if record.exc_info:
|
||||
if not record.exc_text:
|
||||
record.exc_text = _FORMATTER.formatException(record.exc_info)
|
||||
if record.exc_text:
|
||||
message = f"{message}\n{record.exc_text}"
|
||||
if record.stack_info:
|
||||
message = f"{message}\n{_FORMATTER.formatStack(record.stack_info)}"
|
||||
return InMemoryLogRecord(
|
||||
timestamp=_FORMATTER.formatTime(record, _DATE_FORMAT),
|
||||
level=record.levelname,
|
||||
levelno=record.levelno,
|
||||
logger=record.name,
|
||||
message=message,
|
||||
)
|
||||
|
||||
|
||||
_buffer: InMemoryLogBuffer | None = None
|
||||
|
||||
|
||||
def install_log_buffer(
|
||||
target: logging.Logger, capacity: int = DEFAULT_CAPACITY
|
||||
) -> InMemoryLogBuffer:
|
||||
"""Attach the in-memory buffer handler to *target* (idempotent).
|
||||
|
||||
Lowers *target*'s level to at most `INFO` so the console shows a useful tail
|
||||
even when `DEEPAGENTS_CODE_DEBUG` is off; never raises the level. In
|
||||
`__init__.py` this runs *before* `configure_debug_logging`, which then sets
|
||||
the final level (honoring `DEEPAGENTS_CODE_DEBUG` and
|
||||
`DEEPAGENTS_CODE_LOG_LEVEL`) over this `INFO` floor — so any startup warnings
|
||||
`configure_debug_logging` emits are captured by the already-installed buffer.
|
||||
On a fresh `NOTSET` logger the `NOTSET` branch forces `INFO` without
|
||||
consulting `DEEPAGENTS_CODE_LOG_LEVEL`; the `> INFO and no env` branch only
|
||||
matters on reconfiguration (e.g. an `importlib.reload`), where it preserves
|
||||
an explicit `DEEPAGENTS_CODE_LOG_LEVEL` rather than clobbering it with `INFO`.
|
||||
|
||||
Lowering the level does not spill log output onto the terminal: because this
|
||||
handler is present in the propagation chain, `Logger.callHandlers` finds a
|
||||
handler (`found > 0`) and Python's `lastResort` stderr handler is never
|
||||
consulted. The exception is an embedding process that attaches its own
|
||||
`INFO`-or-lower handler to the root logger, which would then see the
|
||||
propagated records.
|
||||
|
||||
Note: this runs as an import-time side effect (see `__init__.py`), so every
|
||||
`import deepagents_code` attaches the handler and may lower the package
|
||||
logger's level to `INFO` for the lifetime of the process.
|
||||
|
||||
Args:
|
||||
target: Logger to attach the buffer to (the package logger).
|
||||
capacity: Maximum records to retain.
|
||||
|
||||
Returns:
|
||||
The installed (or already-installed) buffer handler.
|
||||
"""
|
||||
global _buffer # noqa: PLW0603 # module-level singleton accessor
|
||||
for existing in target.handlers:
|
||||
if isinstance(existing, InMemoryLogBuffer):
|
||||
_buffer = existing
|
||||
return existing
|
||||
|
||||
handler = InMemoryLogBuffer(capacity)
|
||||
handler.setLevel(logging.DEBUG)
|
||||
target.addHandler(handler)
|
||||
if target.level == logging.NOTSET or (
|
||||
target.level > logging.INFO and not os.environ.get(LOG_LEVEL)
|
||||
):
|
||||
target.setLevel(logging.INFO)
|
||||
_buffer = handler
|
||||
return handler
|
||||
|
||||
|
||||
def get_log_buffer() -> InMemoryLogBuffer | None:
|
||||
"""Return the installed buffer handler.
|
||||
|
||||
Returns:
|
||||
The installed buffer handler, or `None` if not yet installed.
|
||||
"""
|
||||
return _buffer
|
||||
@@ -183,6 +183,12 @@ The value is comma-separated for forward-compatibility, not because multiple
|
||||
destinations are written today.
|
||||
"""
|
||||
|
||||
LOG_LEVEL = "DEEPAGENTS_CODE_LOG_LEVEL"
|
||||
"""Minimum level for `deepagents_code` runtime logging.
|
||||
|
||||
Accepted values are DEBUG, INFO, WARNING, ERROR, and CRITICAL.
|
||||
"""
|
||||
|
||||
NO_TERMINAL_ESCAPE = "DEEPAGENTS_CODE_NO_TERMINAL_ESCAPE"
|
||||
"""Disable all terminal escape/control sequence output when enabled."""
|
||||
|
||||
|
||||
@@ -366,6 +366,7 @@ if TYPE_CHECKING:
|
||||
from deepagents_code.tui.widgets.ask_user import AskUserMenu
|
||||
from deepagents_code.tui.widgets.auth import AuthManagerScreen
|
||||
from deepagents_code.tui.widgets.cwd_switch import CwdSwitchAbortMode
|
||||
from deepagents_code.tui.widgets.debug_console import SnapshotField
|
||||
from deepagents_code.tui.widgets.goal_review import GoalReviewMenu, GoalReviewResult
|
||||
from deepagents_code.tui.widgets.model_selector import ModelSelectorScreen
|
||||
from deepagents_code.tui.widgets.notification_center import (
|
||||
@@ -1967,6 +1968,16 @@ class DeepAgentsApp(App):
|
||||
show=False,
|
||||
priority=True,
|
||||
),
|
||||
Binding(
|
||||
# Mirrors DEBUG_TOGGLE_KEY in tui.widgets.debug_console; the literal
|
||||
# is repeated here to avoid an eager import at class-body scope (see
|
||||
# the startup-performance rules in AGENTS.md).
|
||||
"ctrl+backslash",
|
||||
"toggle_debug_console",
|
||||
"Debug Console",
|
||||
show=False,
|
||||
priority=True,
|
||||
),
|
||||
# Approval menu keys (handled at App level for reliability)
|
||||
Binding("up", "approval_up", "Up", show=False),
|
||||
Binding("k", "approval_up", "Up", show=False),
|
||||
@@ -9681,6 +9692,7 @@ class DeepAgentsApp(App):
|
||||
f" {newline_shortcut():<15} Insert newline\n"
|
||||
" Ctrl+X Open prompt in external editor\n"
|
||||
" Ctrl+N Review pending notifications\n"
|
||||
" Ctrl+\\ Toggle the debug console\n"
|
||||
" Shift+Tab Toggle auto-approve mode\n"
|
||||
" @filename Auto-complete files and inject content\n"
|
||||
" /command Slash commands (/help, /clear, /quit)\n"
|
||||
@@ -10102,6 +10114,8 @@ class DeepAgentsApp(App):
|
||||
elif cmd.startswith("/skill:"):
|
||||
await self._handle_skill_command(command)
|
||||
# -- Debug commands (not in COMMANDS / autocomplete) ------------------
|
||||
elif cmd == "/debug":
|
||||
self._open_debug_console()
|
||||
elif cmd == "/debug-error":
|
||||
await self._mount_message(
|
||||
ErrorMessage(
|
||||
@@ -14418,6 +14432,93 @@ class DeepAgentsApp(App):
|
||||
"""Open the notification center via the `ctrl+n` keybind."""
|
||||
self._open_notification_center()
|
||||
|
||||
def action_toggle_debug_console(self) -> None:
|
||||
"""Toggle the Debug Console overlay via keybind or the `/debug` command."""
|
||||
from deepagents_code.tui.widgets.debug_console import DebugConsoleScreen
|
||||
|
||||
if isinstance(self.screen, DebugConsoleScreen):
|
||||
self.pop_screen()
|
||||
if self._chat_input:
|
||||
self._chat_input.focus_input()
|
||||
return
|
||||
self._open_debug_console()
|
||||
|
||||
def _open_debug_console(self) -> None:
|
||||
"""Push the read-only Debug Console modal."""
|
||||
from deepagents_code.tui.widgets.debug_console import DebugConsoleScreen
|
||||
|
||||
def handle_result(_: None) -> None:
|
||||
if self._chat_input:
|
||||
self._chat_input.focus_input()
|
||||
|
||||
self.push_screen(
|
||||
DebugConsoleScreen(self._build_debug_snapshot()), handle_result
|
||||
)
|
||||
|
||||
def _build_debug_snapshot(self) -> list[SnapshotField]:
|
||||
"""Capture a point-in-time session/runtime snapshot for the console.
|
||||
|
||||
Each field is captured defensively: a subsystem that raises degrades to
|
||||
an ``(unavailable: ...)`` value rather than aborting the whole overlay,
|
||||
because a diagnostic tool must still open when the app is misbehaving.
|
||||
|
||||
Returns:
|
||||
Ordered ``(label, value)`` fields for the console header.
|
||||
"""
|
||||
from deepagents_code._debug import installed_debug_log_path
|
||||
from deepagents_code._env_vars import DEBUG, is_env_truthy
|
||||
from deepagents_code._version import __version__
|
||||
from deepagents_code.tui.widgets.debug_console import SnapshotField
|
||||
|
||||
def _safe(label: str, fn: Callable[[], str]) -> SnapshotField:
|
||||
try:
|
||||
return SnapshotField(label=label, value=fn())
|
||||
except Exception as exc: # a diagnostic must still open on a bad field
|
||||
# WARNING (not DEBUG) so the traceback lands in the always-on
|
||||
# in-memory buffer and is visible in the console itself; the
|
||||
# package logger sits at INFO by default, which drops DEBUG.
|
||||
logger.warning("Debug snapshot field %r failed", label, exc_info=True)
|
||||
return SnapshotField(
|
||||
label=label, value=f"(unavailable: {type(exc).__name__})"
|
||||
)
|
||||
|
||||
def _mcp() -> str:
|
||||
servers = self._mcp_server_info or []
|
||||
if not servers:
|
||||
return "none"
|
||||
return ", ".join(f"{s.name} ({s.status})" for s in servers)
|
||||
|
||||
def _tokens() -> str:
|
||||
stats = self._session_stats
|
||||
return (
|
||||
f"{stats.input_tokens} in / {stats.output_tokens} out "
|
||||
f"/ {stats.request_count} req"
|
||||
)
|
||||
|
||||
def _log_path() -> str:
|
||||
path = installed_debug_log_path()
|
||||
if path:
|
||||
return str(path)
|
||||
# DEEPAGENTS_CODE_DEBUG can read truthy with no handler installed —
|
||||
# e.g. a bad path, or the var set after import via .env. Distinguish
|
||||
# that from the plain no-file-logging case so the console does not
|
||||
# imply a file exists (and hint that a request went unfulfilled).
|
||||
if is_env_truthy(DEBUG):
|
||||
return "in-memory only (file logging requested but unavailable)"
|
||||
return "in-memory only"
|
||||
|
||||
return [
|
||||
_safe("Version", lambda: __version__),
|
||||
_safe("Model", lambda: self._effective_model_spec() or "(not configured)"),
|
||||
_safe("Thread", lambda: self._lc_thread_id or "(none)"),
|
||||
_safe("CWD", lambda: self._cwd),
|
||||
_safe("Auto-approve", lambda: "on" if self._auto_approve else "off"),
|
||||
_safe("Sandbox", lambda: self._sandbox_type or "local"),
|
||||
_safe("MCP servers", _mcp),
|
||||
_safe("Tokens", _tokens),
|
||||
_safe("Debug log", _log_path),
|
||||
]
|
||||
|
||||
def _open_notification_center(self) -> None:
|
||||
"""Push the notification center modal, or toast when empty."""
|
||||
from deepagents_code.tui.widgets.notification_center import (
|
||||
|
||||
@@ -311,7 +311,7 @@ SIDE_EFFECT_FREE: frozenset[str] = _build_bypass_set(BypassTier.SIDE_EFFECT_FREE
|
||||
QUEUE_BOUND: frozenset[str] = _build_bypass_set(BypassTier.QUEUED)
|
||||
"""Commands that must wait in the queue when the app is busy."""
|
||||
|
||||
HIDDEN_COMMANDS: frozenset[str] = frozenset({"/debug-error"})
|
||||
HIDDEN_COMMANDS: frozenset[str] = frozenset({"/debug", "/debug-error"})
|
||||
"""Power-user commands kept out of autocomplete and help."""
|
||||
|
||||
STARTUP_RECOVERY_COMMANDS: frozenset[str] = frozenset(
|
||||
|
||||
@@ -519,10 +519,13 @@ def _quiet_sdk_tracing_logging() -> None:
|
||||
stay off the terminal.
|
||||
"""
|
||||
from deepagents_code._debug import configure_debug_logging
|
||||
from deepagents_code._env_vars import DEBUG, is_env_truthy
|
||||
|
||||
debug_enabled = is_env_truthy(DEBUG)
|
||||
for name in ("langsmith", "langchain"):
|
||||
sdk_logger = logging.getLogger(name)
|
||||
configure_debug_logging(sdk_logger)
|
||||
if debug_enabled:
|
||||
configure_debug_logging(sdk_logger)
|
||||
if not sdk_logger.handlers:
|
||||
sdk_logger.addHandler(logging.NullHandler())
|
||||
|
||||
|
||||
@@ -70,14 +70,14 @@ class OptionKind(Enum):
|
||||
|
||||
All kinds flow through `resolve_scalar`. The scalar kinds (`BOOL`,
|
||||
`BOOL_PRESENCE`, `INT`, `FLOAT`, `STR`) are coerced inline by
|
||||
`_coerce_env`/`_coerce_toml`. `SHELL_LIST_DELEGATE`, `SKILLS_DIRS_DELEGATE`,
|
||||
`PTC_DELEGATE`, and `STARTUP_MODE_DELEGATE` defer to bespoke parsers (their
|
||||
semantics — colon-split Path resolution, comma + `recommended`/`all`
|
||||
sentinels, the PTC/startup-mode allowlists — do not compress into a generic
|
||||
coercion). `THEME_DELEGATE` is resolved separately at the top of
|
||||
`resolve_scalar` and never reaches the inline coercers. `STRUCTURED` marks
|
||||
user-defined tables that the scalar resolver only passes through for
|
||||
display.
|
||||
`_coerce_env`/`_coerce_toml`. `LOG_LEVEL_DELEGATE`, `SHELL_LIST_DELEGATE`,
|
||||
`SKILLS_DIRS_DELEGATE`, `PTC_DELEGATE`, and `STARTUP_MODE_DELEGATE` defer to
|
||||
bespoke parsers (their semantics — dynamic debug fallback, colon-split Path
|
||||
resolution, comma + `recommended`/`all` sentinels, and the PTC/startup-mode
|
||||
allowlists — do not compress into a generic coercion). `THEME_DELEGATE` is
|
||||
resolved separately at the top of `resolve_scalar` and never reaches the
|
||||
inline coercers. `STRUCTURED` marks user-defined tables that the scalar
|
||||
resolver only passes through for display.
|
||||
"""
|
||||
|
||||
BOOL = "bool"
|
||||
@@ -93,6 +93,9 @@ class OptionKind(Enum):
|
||||
|
||||
STR = "str"
|
||||
|
||||
LOG_LEVEL_DELEGATE = "log_level"
|
||||
"""Validates log levels and resolves the default from debug mode."""
|
||||
|
||||
SHELL_LIST_DELEGATE = "shell_list"
|
||||
"""Delegates to `config.parse_shell_allow_list`."""
|
||||
|
||||
@@ -118,6 +121,7 @@ _KIND_TYPE_LABEL: dict[OptionKind, str] = {
|
||||
OptionKind.INT: "int",
|
||||
OptionKind.FLOAT: "float",
|
||||
OptionKind.STR: "str",
|
||||
OptionKind.LOG_LEVEL_DELEGATE: "str",
|
||||
OptionKind.SHELL_LIST_DELEGATE: "list[str]",
|
||||
OptionKind.SKILLS_DIRS_DELEGATE: "list[path]",
|
||||
OptionKind.PTC_DELEGATE: "str | list[str]",
|
||||
@@ -383,6 +387,15 @@ def _coerce_env(option: ConfigOption, raw: str, name: str) -> object:
|
||||
return bool(raw)
|
||||
if kind is OptionKind.STR:
|
||||
return raw
|
||||
if kind is OptionKind.LOG_LEVEL_DELEGATE:
|
||||
from deepagents_code._debug import LOG_LEVELS
|
||||
|
||||
level = raw.strip().upper()
|
||||
if level in LOG_LEVELS:
|
||||
return level
|
||||
valid = ", ".join(LOG_LEVELS)
|
||||
logger.warning("Ignoring %s=%r (expected one of %s)", name, raw, valid)
|
||||
return _INVALID
|
||||
if kind is OptionKind.INT:
|
||||
try:
|
||||
return int(raw.strip())
|
||||
@@ -610,6 +623,11 @@ def resolve_scalar(
|
||||
if value is not _INVALID:
|
||||
return value, "config.toml"
|
||||
|
||||
if option.kind is OptionKind.LOG_LEVEL_DELEGATE:
|
||||
from deepagents_code._env_vars import DEBUG, is_env_truthy
|
||||
|
||||
return ("DEBUG" if is_env_truthy(DEBUG) else "INFO"), "default"
|
||||
|
||||
return option.default, "default"
|
||||
|
||||
|
||||
@@ -1244,6 +1262,16 @@ _STATIC_OPTIONS: tuple[ConfigOption, ...] = (
|
||||
default="/tmp/deepagents_debug.log", # noqa: S108 # documents the app default, not a write target
|
||||
env_var=_env_vars.DEBUG_FILE,
|
||||
),
|
||||
ConfigOption(
|
||||
key="debug.log_level",
|
||||
group="Debug",
|
||||
summary=(
|
||||
"Minimum runtime log level (DEBUG, INFO, WARNING, ERROR, or CRITICAL); "
|
||||
"defaults to DEBUG in debug mode and INFO otherwise."
|
||||
),
|
||||
kind=OptionKind.LOG_LEVEL_DELEGATE,
|
||||
env_var=_env_vars.LOG_LEVEL,
|
||||
),
|
||||
ConfigOption(
|
||||
key="debug.onboarding",
|
||||
group="Debug",
|
||||
|
||||
@@ -0,0 +1,868 @@
|
||||
r"""Read-only in-app Debug Console modal.
|
||||
|
||||
Toggled with `Ctrl+\` (or the hidden `/debug` command), this overlay shows a
|
||||
point-in-time session/runtime snapshot plus a live tail of recent
|
||||
`deepagents_code.*` log records sourced from the in-memory ring buffer in
|
||||
`_debug_buffer`. It never mutates session state.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import bisect
|
||||
import logging
|
||||
from typing import TYPE_CHECKING, ClassVar, Literal, NamedTuple, cast, get_args
|
||||
|
||||
from rich.segment import Segment
|
||||
from rich.style import Style as RichStyle
|
||||
from textual.binding import Binding, BindingType
|
||||
from textual.cache import LRUCache
|
||||
from textual.containers import Horizontal, Vertical
|
||||
from textual.content import Content
|
||||
from textual.geometry import Size
|
||||
from textual.screen import ModalScreen
|
||||
from textual.scroll_view import ScrollView
|
||||
from textual.strip import Strip
|
||||
from textual.widgets import Select, Static
|
||||
from textual.widgets._select import ( # noqa: PLC2701 # needed to keep Tab navigation inside the open Select overlay
|
||||
SelectCurrent,
|
||||
SelectOverlay,
|
||||
)
|
||||
|
||||
from deepagents_code import theme
|
||||
from deepagents_code._debug import LOG_LEVELS
|
||||
from deepagents_code._debug_buffer import (
|
||||
DEFAULT_CAPACITY,
|
||||
InMemoryLogRecord,
|
||||
get_log_buffer,
|
||||
)
|
||||
from deepagents_code.clipboard import copy_text_to_clipboard
|
||||
from deepagents_code.unicode_security import sanitize_control_chars
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Callable, Sequence
|
||||
|
||||
from textual import events
|
||||
from textual.app import ComposeResult
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
DEBUG_TOGGLE_KEY = "ctrl+backslash"
|
||||
r"""Textual key name for the `Ctrl+\` chord that toggles the console."""
|
||||
|
||||
|
||||
class SnapshotField(NamedTuple):
|
||||
"""A single `label`/`value` row in the console's session snapshot.
|
||||
|
||||
A named pair (rather than a bare ``tuple[str, str]``) so the two display-only
|
||||
string slots cannot be silently transposed at the construction site.
|
||||
"""
|
||||
|
||||
label: str
|
||||
value: str
|
||||
|
||||
|
||||
_REFRESH_INTERVAL = 0.5
|
||||
"""Seconds between log-tail refresh ticks."""
|
||||
|
||||
_RECORD_LIMIT = DEFAULT_CAPACITY
|
||||
"""Maximum records retained by an open debug console view."""
|
||||
|
||||
_FILTER_SELECT_ID = "debug-level-filter"
|
||||
FilterValue = Literal[
|
||||
"all",
|
||||
"min:DEBUG",
|
||||
"min:INFO",
|
||||
"min:WARNING",
|
||||
"min:ERROR",
|
||||
"min:CRITICAL",
|
||||
"only:DEBUG",
|
||||
"only:INFO",
|
||||
"only:WARNING",
|
||||
"only:ERROR",
|
||||
"only:CRITICAL",
|
||||
]
|
||||
_BASE_FILTER_OPTIONS: tuple[tuple[str, FilterValue], ...] = (
|
||||
("All", "all"),
|
||||
("INFO", "min:INFO"),
|
||||
("WARNING", "min:WARNING"),
|
||||
("ERROR", "min:ERROR"),
|
||||
("CRITICAL", "min:CRITICAL"),
|
||||
("Only INFO", "only:INFO"),
|
||||
("Only WARNING", "only:WARNING"),
|
||||
("Only ERROR", "only:ERROR"),
|
||||
("Only CRITICAL", "only:CRITICAL"),
|
||||
)
|
||||
_DEBUG_FILTER_OPTIONS: tuple[tuple[str, FilterValue], ...] = (
|
||||
("DEBUG", "min:DEBUG"),
|
||||
("Only DEBUG", "only:DEBUG"),
|
||||
)
|
||||
_VALID_FILTER_VALUES: frozenset[str] = frozenset(get_args(FilterValue))
|
||||
"""Every legal `FilterValue`, used to validate values crossing the Select
|
||||
boundary before they are trusted as a `FilterValue`."""
|
||||
_LEVEL_STYLES = {
|
||||
"DEBUG": "dim",
|
||||
"INFO": "cyan",
|
||||
"WARNING": "yellow",
|
||||
"ERROR": "red",
|
||||
"CRITICAL": "bold red",
|
||||
}
|
||||
_EMPTY_STYLE = RichStyle()
|
||||
|
||||
|
||||
def _sanitize_display_text(text: str, *, keep_newlines: bool = False) -> str:
|
||||
"""Return text safe to render in the debug console."""
|
||||
return sanitize_control_chars(
|
||||
text,
|
||||
keep_newlines=keep_newlines,
|
||||
collapse_whitespace=False,
|
||||
)
|
||||
|
||||
|
||||
def _debug_records_enabled() -> bool:
|
||||
"""Return whether the package logger can emit DEBUG records."""
|
||||
return logging.getLogger("deepagents_code").isEnabledFor(logging.DEBUG)
|
||||
|
||||
|
||||
def _filter_options() -> tuple[tuple[str, FilterValue], ...]:
|
||||
"""Return level filter options valid for the current logging configuration."""
|
||||
if not _debug_records_enabled():
|
||||
return _BASE_FILTER_OPTIONS
|
||||
all_option = _BASE_FILTER_OPTIONS[:1]
|
||||
rest = _BASE_FILTER_OPTIONS[1:]
|
||||
return (*all_option, *_DEBUG_FILTER_OPTIONS, *rest)
|
||||
|
||||
|
||||
def _record_matches_filter(
|
||||
record: InMemoryLogRecord, level_filter: FilterValue
|
||||
) -> bool:
|
||||
"""Return whether *record* should be visible for *level_filter*."""
|
||||
if level_filter == "all":
|
||||
return True
|
||||
mode, selected_level = level_filter.split(":", maxsplit=1)
|
||||
if mode == "only":
|
||||
return record.level == selected_level
|
||||
threshold = LOG_LEVELS.get(selected_level)
|
||||
if threshold is None:
|
||||
# An unrecognized level should never reach here (FilterValue enumerates
|
||||
# only LOG_LEVELS keys), but a diagnostic must not hide records on a bad
|
||||
# filter: show everything rather than raise on the poll timer.
|
||||
return True
|
||||
return record.levelno >= threshold
|
||||
|
||||
|
||||
def _record_to_content(record: InMemoryLogRecord) -> Content:
|
||||
"""Render a structured log record as styled Textual content.
|
||||
|
||||
Returns:
|
||||
Styled content for the log view.
|
||||
"""
|
||||
timestamp = _sanitize_display_text(record.timestamp)
|
||||
level = _sanitize_display_text(record.level)
|
||||
logger = _sanitize_display_text(record.logger)
|
||||
message = _sanitize_display_text(record.message, keep_newlines=True)
|
||||
level_style = _LEVEL_STYLES.get(record.level, "dim")
|
||||
return Content.assemble(
|
||||
(timestamp, "dim"),
|
||||
" ",
|
||||
(f"{level:<8}", level_style),
|
||||
" ",
|
||||
(logger, "dim"),
|
||||
" ",
|
||||
message,
|
||||
)
|
||||
|
||||
|
||||
class _LogLevelOverlay(SelectOverlay):
|
||||
"""Select overlay that treats Tab and Shift+Tab like down and up arrows."""
|
||||
|
||||
def key_tab(self, event: events.Key) -> None:
|
||||
"""Move the highlighted option down while the menu is open."""
|
||||
event.prevent_default()
|
||||
event.stop()
|
||||
self.action_cursor_down()
|
||||
|
||||
def key_shift_tab(self, event: events.Key) -> None:
|
||||
"""Move the highlighted option up while the menu is open."""
|
||||
event.prevent_default()
|
||||
event.stop()
|
||||
self.action_cursor_up()
|
||||
|
||||
def key_escape(self, event: events.Key) -> None:
|
||||
"""Close the dropdown without dismissing the debug console."""
|
||||
event.prevent_default()
|
||||
event.stop()
|
||||
self.action_dismiss()
|
||||
|
||||
def check_consume_key(self, key: str, character: str | None = None) -> bool:
|
||||
"""Prevent screen-level focus traversal while the menu is open.
|
||||
|
||||
Returns:
|
||||
`True` when this overlay should handle the key itself.
|
||||
"""
|
||||
return key in {"escape", "tab", "shift+tab"} or super().check_consume_key(
|
||||
key, character
|
||||
)
|
||||
|
||||
|
||||
class _LogLevelSelect(Select[FilterValue]):
|
||||
"""Level dropdown whose open menu treats Tab like arrow navigation."""
|
||||
|
||||
def compose(self) -> ComposeResult:
|
||||
"""Compose the select with a Tab-aware overlay.
|
||||
|
||||
Yields:
|
||||
Current value display and dropdown overlay widgets.
|
||||
"""
|
||||
yield SelectCurrent(self.prompt)
|
||||
yield _LogLevelOverlay(type_to_search=self._type_to_search).data_bind(
|
||||
compact=Select.compact
|
||||
)
|
||||
|
||||
|
||||
class _DebugLogView(ScrollView, can_focus=True):
|
||||
"""Scrollable styled log view with logical-record hover and click handling."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
on_copy_record: Callable[[InMemoryLogRecord], None],
|
||||
*,
|
||||
widget_id: str | None = None,
|
||||
classes: str | None = None,
|
||||
) -> None:
|
||||
super().__init__(id=widget_id, classes=classes)
|
||||
self._on_copy_record = on_copy_record
|
||||
self._records: list[InMemoryLogRecord] = []
|
||||
self._notice: Content | None = None
|
||||
self._contents: list[Content] = []
|
||||
self._wrap_counts: list[int] = []
|
||||
self._wrap_prefix: list[int] = [0]
|
||||
self._total_visual = 0
|
||||
self._cached_width = 0
|
||||
self._hover_index: int | None = None
|
||||
self._selected_index: int | None = None
|
||||
self._render_line_cache: LRUCache[
|
||||
tuple[int, int, int, int | None, int | None], Strip
|
||||
] = LRUCache(1024)
|
||||
|
||||
@property
|
||||
def line_count(self) -> int:
|
||||
"""The current visual line count."""
|
||||
return self._total_visual
|
||||
|
||||
@property
|
||||
def records(self) -> Sequence[InMemoryLogRecord]:
|
||||
"""The currently visible logical records."""
|
||||
return self._records
|
||||
|
||||
def set_records(
|
||||
self, records: Sequence[InMemoryLogRecord], *, scroll_end: bool = True
|
||||
) -> None:
|
||||
"""Replace the visible records and optionally scroll to the bottom."""
|
||||
self._notice = None
|
||||
self._records = list(records)
|
||||
self._hover_index = None
|
||||
self._selected_index = self._coerce_selected_index(self._selected_index)
|
||||
self._rebuild_contents()
|
||||
self._reflow()
|
||||
if scroll_end:
|
||||
self.scroll_end(animate=False, immediate=True, x_axis=False)
|
||||
|
||||
def append_records(self, records: Sequence[InMemoryLogRecord]) -> None:
|
||||
"""Append records to the visible view."""
|
||||
if not records:
|
||||
return
|
||||
if self._notice is not None:
|
||||
self.set_records(records)
|
||||
return
|
||||
at_bottom = self.is_vertical_scroll_end
|
||||
start = len(self._contents)
|
||||
self._records.extend(records)
|
||||
self._contents.extend(_record_to_content(record) for record in records)
|
||||
width = self._cached_width or self.size.width
|
||||
if width <= 0:
|
||||
# Not yet sized (e.g. first poll before layout). Assume one visual
|
||||
# line per new content so `_wrap_counts` stays 1:1 with `_contents`;
|
||||
# the first `on_resize` reflow recomputes real counts. Skipping this
|
||||
# would leave the new records out of `_wrap_prefix`/`_total_visual`
|
||||
# and invisible until that resize.
|
||||
self._wrap_counts.extend(1 for _ in self._contents[start:])
|
||||
self._recompute_prefix()
|
||||
self.refresh()
|
||||
return
|
||||
self._cached_width = width
|
||||
counts = [
|
||||
self._wrap_count(content, width) for content in self._contents[start:]
|
||||
]
|
||||
self._wrap_counts.extend(counts)
|
||||
self._recompute_prefix()
|
||||
self.virtual_size = Size(width, self._total_visual)
|
||||
self._render_line_cache.clear()
|
||||
self.refresh()
|
||||
if at_bottom:
|
||||
self.scroll_end(animate=False, immediate=True, x_axis=False)
|
||||
|
||||
def clear_records(self) -> None:
|
||||
"""Clear the visible records."""
|
||||
self._notice = None
|
||||
self._records.clear()
|
||||
self._contents.clear()
|
||||
self._wrap_counts.clear()
|
||||
self._wrap_prefix = [0]
|
||||
self._total_visual = 0
|
||||
self._hover_index = None
|
||||
self._selected_index = None
|
||||
self._render_line_cache.clear()
|
||||
self.virtual_size = Size(self.size.width, 0)
|
||||
self.refresh()
|
||||
|
||||
def show_notice(self, message: str) -> None:
|
||||
"""Render a one-line notice in place of log records."""
|
||||
self._records.clear()
|
||||
self._notice = Content.styled(_sanitize_display_text(message), "dim italic")
|
||||
self._hover_index = None
|
||||
self._selected_index = None
|
||||
self._rebuild_contents()
|
||||
self._reflow()
|
||||
|
||||
def _rebuild_contents(self) -> None:
|
||||
if self._notice is not None:
|
||||
self._contents = [self._notice]
|
||||
return
|
||||
self._contents = [_record_to_content(record) for record in self._records]
|
||||
|
||||
@staticmethod
|
||||
def _wrap_count(content: Content, width: int) -> int:
|
||||
if width <= 0:
|
||||
return 1
|
||||
return max(1, len(content.wrap(width)))
|
||||
|
||||
def _recompute_prefix(self) -> None:
|
||||
self._wrap_prefix = [0]
|
||||
for count in self._wrap_counts:
|
||||
self._wrap_prefix.append(self._wrap_prefix[-1] + count)
|
||||
self._total_visual = self._wrap_prefix[-1]
|
||||
|
||||
def _reflow(self) -> None:
|
||||
width = self.size.width
|
||||
if width <= 0:
|
||||
width = self._cached_width
|
||||
if width <= 0:
|
||||
self._wrap_counts = [1 for _content in self._contents]
|
||||
self._recompute_prefix()
|
||||
self.refresh()
|
||||
return
|
||||
self._cached_width = width
|
||||
self._render_line_cache.clear()
|
||||
self._wrap_counts = [
|
||||
self._wrap_count(content, width) for content in self._contents
|
||||
]
|
||||
self._recompute_prefix()
|
||||
self.virtual_size = Size(width, self._total_visual)
|
||||
self.refresh()
|
||||
|
||||
def _content_index_at_visual_y(self, visual_y: int) -> int | None:
|
||||
if visual_y < 0 or visual_y >= self._total_visual:
|
||||
return None
|
||||
index = bisect.bisect_right(self._wrap_prefix, visual_y) - 1
|
||||
if 0 <= index < len(self._contents):
|
||||
return index
|
||||
return None
|
||||
|
||||
def _record_at_visual_y(self, visual_y: int) -> InMemoryLogRecord | None:
|
||||
if self._notice is not None:
|
||||
return None
|
||||
index = self._content_index_at_visual_y(visual_y)
|
||||
if index is None or index >= len(self._records):
|
||||
return None
|
||||
return self._records[index]
|
||||
|
||||
def _coerce_selected_index(self, index: int | None) -> int | None:
|
||||
if not self._records:
|
||||
return None
|
||||
if index is None:
|
||||
return None
|
||||
return min(max(index, 0), len(self._records) - 1)
|
||||
|
||||
def _select_record(self, index: int) -> None:
|
||||
if not self._records:
|
||||
return
|
||||
self._selected_index = min(max(index, 0), len(self._records) - 1)
|
||||
self._hover_index = None
|
||||
self._render_line_cache.clear()
|
||||
self._scroll_selected_visible()
|
||||
self.refresh()
|
||||
|
||||
def _scroll_selected_visible(self) -> None:
|
||||
if self._selected_index is None or not self._wrap_prefix:
|
||||
return
|
||||
start = self._wrap_prefix[self._selected_index]
|
||||
end = self._wrap_prefix[self._selected_index + 1] - 1
|
||||
_scroll_x, scroll_y = self.scroll_offset
|
||||
height = max(self.size.height, 1)
|
||||
if start < scroll_y:
|
||||
self.scroll_to(y=start, animate=False, immediate=True)
|
||||
elif end >= scroll_y + height:
|
||||
self.scroll_to(y=end - height + 1, animate=False, immediate=True)
|
||||
|
||||
def _copy_selected_record(self) -> None:
|
||||
if self._selected_index is None:
|
||||
if not self._records:
|
||||
return
|
||||
self._selected_index = len(self._records) - 1
|
||||
record = self._records[self._selected_index]
|
||||
self._on_copy_record(record)
|
||||
|
||||
def render_line(self, y: int) -> Strip:
|
||||
_scroll_x, scroll_y = self.scroll_offset
|
||||
abs_y = scroll_y + y
|
||||
width = self.size.width
|
||||
key = (
|
||||
abs_y,
|
||||
width,
|
||||
self._cached_width,
|
||||
self._hover_index,
|
||||
self._selected_index,
|
||||
)
|
||||
cached = self._render_line_cache.get(key)
|
||||
if cached is not None:
|
||||
return cached
|
||||
if abs_y >= self._total_visual:
|
||||
return Strip.blank(width, self.rich_style)
|
||||
|
||||
content_index = self._content_index_at_visual_y(abs_y)
|
||||
if content_index is None:
|
||||
return Strip.blank(width, self.rich_style)
|
||||
content = self._contents[content_index]
|
||||
row_style: RichStyle | None = None
|
||||
if self._selected_index == content_index and self._notice is None:
|
||||
colors = theme.get_theme_colors(self)
|
||||
row_style = RichStyle(
|
||||
color=colors.background,
|
||||
bgcolor=colors.primary,
|
||||
bold=True,
|
||||
)
|
||||
elif self._hover_index == content_index and self._notice is None:
|
||||
colors = theme.get_theme_colors(self)
|
||||
row_style = RichStyle(bgcolor=colors.panel)
|
||||
wrapped = content.wrap(self._cached_width or width)
|
||||
base = self._wrap_prefix[content_index]
|
||||
line = wrapped[abs_y - base] if abs_y - base < len(wrapped) else Content()
|
||||
segments = [
|
||||
segment
|
||||
if segment.style is not None
|
||||
else Segment(segment.text, _EMPTY_STYLE)
|
||||
for segment in line.render_segments(end="")
|
||||
]
|
||||
strip = Strip(segments, line.cell_length).crop_extend(0, width, self.rich_style)
|
||||
if row_style is not None:
|
||||
strip = Strip(
|
||||
Segment.apply_style(strip, None, row_style),
|
||||
strip.cell_length,
|
||||
)
|
||||
self._render_line_cache[key] = strip
|
||||
return strip
|
||||
|
||||
def notify_style_update(self) -> None:
|
||||
"""Clear cached render lines after a style update."""
|
||||
super().notify_style_update()
|
||||
self._render_line_cache.clear()
|
||||
|
||||
def on_resize(self, event: events.Resize) -> None:
|
||||
"""Re-wrap log entries when the view width changes."""
|
||||
if event.size.width != self._cached_width:
|
||||
self._reflow()
|
||||
|
||||
def on_mouse_move(self, event: events.MouseMove) -> None:
|
||||
"""Highlight the logical log record under the pointer."""
|
||||
_scroll_x, scroll_y = self.scroll_offset
|
||||
hover_index = self._content_index_at_visual_y(scroll_y + event.y)
|
||||
if self._notice is not None:
|
||||
hover_index = None
|
||||
self.styles.pointer = "pointer" if hover_index is not None else "default"
|
||||
if hover_index == self._hover_index:
|
||||
return
|
||||
self._hover_index = hover_index
|
||||
self._render_line_cache.clear()
|
||||
self.refresh()
|
||||
|
||||
def on_leave(self) -> None:
|
||||
"""Clear hover highlighting when the pointer leaves the log."""
|
||||
self.styles.pointer = "default"
|
||||
if self._hover_index is None:
|
||||
return
|
||||
self._hover_index = None
|
||||
self._render_line_cache.clear()
|
||||
self.refresh()
|
||||
|
||||
def on_focus(self) -> None:
|
||||
"""Select the latest log record when keyboard focus enters the log."""
|
||||
if self._selected_index is None and self._records:
|
||||
self._select_record(len(self._records) - 1)
|
||||
|
||||
def key_up(self, event: events.Key) -> None:
|
||||
"""Move keyboard selection to the previous logical log record."""
|
||||
event.prevent_default()
|
||||
event.stop()
|
||||
if not self._records:
|
||||
return
|
||||
index = (
|
||||
len(self._records) if self._selected_index is None else self._selected_index
|
||||
)
|
||||
self._select_record(index - 1)
|
||||
|
||||
def key_down(self, event: events.Key) -> None:
|
||||
"""Move keyboard selection to the next logical log record."""
|
||||
event.prevent_default()
|
||||
event.stop()
|
||||
if not self._records:
|
||||
return
|
||||
index = -1 if self._selected_index is None else self._selected_index
|
||||
self._select_record(index + 1)
|
||||
|
||||
def key_enter(self, event: events.Key) -> None:
|
||||
"""Copy the selected logical log record."""
|
||||
event.prevent_default()
|
||||
event.stop()
|
||||
self._copy_selected_record()
|
||||
|
||||
def key_tab(self, event: events.Key) -> None:
|
||||
"""Move focus from the log to the level filter."""
|
||||
event.prevent_default()
|
||||
event.stop()
|
||||
self.screen.focus_next("#debug-level-filter, #debug-log")
|
||||
|
||||
def key_shift_tab(self, event: events.Key) -> None:
|
||||
"""Move focus from the log to the level filter."""
|
||||
event.prevent_default()
|
||||
event.stop()
|
||||
self.screen.focus_previous("#debug-level-filter, #debug-log")
|
||||
|
||||
def on_click(self, event: events.Click) -> None:
|
||||
"""Copy the clicked logical log record."""
|
||||
_scroll_x, scroll_y = self.scroll_offset
|
||||
record = self._record_at_visual_y(scroll_y + event.y)
|
||||
if record is None:
|
||||
return
|
||||
index = self._content_index_at_visual_y(scroll_y + event.y)
|
||||
if index is not None:
|
||||
self._select_record(index)
|
||||
event.stop()
|
||||
self._on_copy_record(record)
|
||||
|
||||
|
||||
class DebugConsoleScreen(ModalScreen[None]):
|
||||
"""Modal showing a session snapshot and a live tail of recent log records."""
|
||||
|
||||
BINDINGS: ClassVar[list[BindingType]] = [
|
||||
Binding("escape", "close", "Close", show=False),
|
||||
Binding(DEBUG_TOGGLE_KEY, "close", "Close", show=False, priority=True),
|
||||
Binding("ctrl+l", "clear_view", "Clear view", show=False, priority=True),
|
||||
# Not `priority`: a priority `c` would pre-empt type-to-search in the
|
||||
# open level dropdown (e.g. typing "c" to reach CRITICAL). The log view
|
||||
# has no `c` binding, so it still bubbles up to this copy action.
|
||||
Binding("c", "copy", "Copy", show=False),
|
||||
]
|
||||
"""The toggle-key close (`ctrl+backslash`) and `ctrl+l` clear-view are
|
||||
`priority`. Escape close and `c` copy are deliberately *not* `priority`:
|
||||
Escape must reach the open level dropdown's overlay first so it closes only
|
||||
the menu (a priority Escape would tear down the whole console instead), and
|
||||
`c` must not pre-empt the dropdown's type-to-search."""
|
||||
|
||||
CSS = """
|
||||
DebugConsoleScreen {
|
||||
align: center middle;
|
||||
}
|
||||
|
||||
DebugConsoleScreen > Vertical {
|
||||
width: 100;
|
||||
max-width: 95%;
|
||||
height: 85%;
|
||||
background: $surface;
|
||||
border: solid $primary;
|
||||
padding: 1 2;
|
||||
}
|
||||
|
||||
DebugConsoleScreen .debug-console-title {
|
||||
text-style: bold;
|
||||
color: $primary;
|
||||
text-align: center;
|
||||
margin-bottom: 1;
|
||||
}
|
||||
|
||||
DebugConsoleScreen .debug-console-snapshot {
|
||||
margin-bottom: 1;
|
||||
}
|
||||
|
||||
DebugConsoleScreen .debug-console-toolbar {
|
||||
height: auto;
|
||||
margin-bottom: 1;
|
||||
}
|
||||
|
||||
DebugConsoleScreen .debug-console-filter-label {
|
||||
width: auto;
|
||||
content-align: center middle;
|
||||
color: $text-muted;
|
||||
margin-right: 1;
|
||||
}
|
||||
|
||||
DebugConsoleScreen #debug-level-filter {
|
||||
width: 18;
|
||||
}
|
||||
|
||||
DebugConsoleScreen .debug-console-log {
|
||||
height: 1fr;
|
||||
min-height: 5;
|
||||
scrollbar-gutter: stable;
|
||||
background: $background;
|
||||
border: solid $primary;
|
||||
overflow-x: hidden;
|
||||
overflow-y: scroll;
|
||||
}
|
||||
|
||||
DebugConsoleScreen .debug-console-log:focus {
|
||||
border: solid $primary-lighten-2;
|
||||
}
|
||||
|
||||
DebugConsoleScreen .debug-console-help {
|
||||
height: 1;
|
||||
color: $text-muted;
|
||||
text-style: italic;
|
||||
margin-top: 1;
|
||||
text-align: center;
|
||||
}
|
||||
"""
|
||||
|
||||
def __init__(self, snapshot: Sequence[SnapshotField]) -> None:
|
||||
"""Initialize with a captured *snapshot* of session/runtime fields.
|
||||
|
||||
Args:
|
||||
snapshot: Ordered `(label, value)` fields rendered in the header.
|
||||
"""
|
||||
super().__init__()
|
||||
self._snapshot = list(snapshot)
|
||||
self._records: list[InMemoryLogRecord] = []
|
||||
# Absolute index of the next unrendered log record (incremental writes).
|
||||
self._rendered_upto = 0
|
||||
# One-shot guard so the "buffer unavailable" notice is written only once.
|
||||
self._missing_notice_shown = False
|
||||
self._level_filter: FilterValue = "all"
|
||||
|
||||
def compose(self) -> ComposeResult:
|
||||
"""Lay out the title, snapshot, filter, log tail, and key-hint footer.
|
||||
|
||||
Yields:
|
||||
The child widgets composing the console.
|
||||
"""
|
||||
with Vertical():
|
||||
yield Static("Debug Console", classes="debug-console-title")
|
||||
yield Static(self._render_snapshot(), classes="debug-console-snapshot")
|
||||
with Horizontal(classes="debug-console-toolbar"):
|
||||
yield Static("Level", classes="debug-console-filter-label")
|
||||
yield _LogLevelSelect(
|
||||
_filter_options(),
|
||||
value="all",
|
||||
allow_blank=False,
|
||||
id=_FILTER_SELECT_ID,
|
||||
compact=True,
|
||||
)
|
||||
yield _DebugLogView(
|
||||
self._copy_record,
|
||||
widget_id="debug-log",
|
||||
classes="debug-console-log",
|
||||
)
|
||||
yield Static(self._render_help(), classes="debug-console-help")
|
||||
|
||||
def on_mount(self) -> None:
|
||||
"""Start the refresh timer and render the current buffer contents."""
|
||||
self.set_interval(_REFRESH_INTERVAL, self._poll_logs)
|
||||
self._poll_logs()
|
||||
self.call_after_refresh(self.query_one("#debug-log", _DebugLogView).focus)
|
||||
|
||||
def key_tab(self, event: events.Key) -> None:
|
||||
"""Cycle focus between the level filter and log lines."""
|
||||
if self._level_select().expanded:
|
||||
return
|
||||
event.prevent_default()
|
||||
event.stop()
|
||||
self.focus_next("#debug-level-filter, #debug-log")
|
||||
|
||||
def key_shift_tab(self, event: events.Key) -> None:
|
||||
"""Cycle focus between the log lines and level filter."""
|
||||
if self._level_select().expanded:
|
||||
return
|
||||
event.prevent_default()
|
||||
event.stop()
|
||||
self.focus_previous("#debug-level-filter, #debug-log")
|
||||
|
||||
def on_select_changed(self, event: Select.Changed) -> None:
|
||||
"""Refresh visible records when the log-level filter changes."""
|
||||
if event.select.id != _FILTER_SELECT_ID:
|
||||
return
|
||||
value = str(event.value)
|
||||
if value == self._level_filter:
|
||||
return
|
||||
if value not in _VALID_FILTER_VALUES:
|
||||
# The Select only offers known options, so this is unreachable in
|
||||
# practice; validate anyway so an unexpected value degrades to the
|
||||
# current filter instead of being trusted as a FilterValue.
|
||||
logger.warning("Ignoring unknown debug level filter %r", value)
|
||||
return
|
||||
self._level_filter = cast("FilterValue", value)
|
||||
self._refresh_log_view(scroll_end=True)
|
||||
|
||||
def _render_snapshot(self) -> Content:
|
||||
"""Build the right-aligned `label: value` snapshot block.
|
||||
|
||||
Returns:
|
||||
The formatted snapshot block.
|
||||
"""
|
||||
if not self._snapshot:
|
||||
return Content.styled("(no session data)", "dim italic")
|
||||
width = max(len(label) for label, _ in self._snapshot)
|
||||
lines = [
|
||||
Content.assemble((f"{label:>{width}} ", "bold"), value)
|
||||
for label, value in self._snapshot
|
||||
]
|
||||
return Content("\n").join(lines)
|
||||
|
||||
@staticmethod
|
||||
def _render_help() -> Content:
|
||||
"""Build the footer key-hint line.
|
||||
|
||||
Returns:
|
||||
The formatted key-hint line.
|
||||
"""
|
||||
return Content.styled(
|
||||
"Esc close · Ctrl+L clear view · c copy visible logs · click copy line",
|
||||
"dim italic",
|
||||
)
|
||||
|
||||
def _poll_logs(self) -> None:
|
||||
"""Append log records emitted since the last tick, guarding the timer.
|
||||
|
||||
Runs on a repeating `set_interval` timer, so an unhandled exception here
|
||||
would propagate out of the callback and tear down the whole host app.
|
||||
A diagnostic overlay must degrade instead: a tick that races teardown
|
||||
(`NoMatches`) is logged at DEBUG and skipped, and any other failure
|
||||
degrades the tail to a notice rather than crashing the app it exists to
|
||||
inspect.
|
||||
"""
|
||||
from textual.css.query import NoMatches
|
||||
|
||||
try:
|
||||
self._poll_logs_once()
|
||||
except NoMatches:
|
||||
# Expected when a queued tick races console teardown: the log widget
|
||||
# is already gone. Logged at DEBUG (not swallowed outright) so a
|
||||
# genuine missing/mis-typed-widget bug still leaves a breadcrumb in
|
||||
# the buffer instead of silently rendering nothing forever.
|
||||
logger.debug("Debug console poll skipped (widget unavailable)")
|
||||
return
|
||||
except Exception: # a diagnostic must never crash the app it inspects
|
||||
logger.warning("Debug console log poll failed", exc_info=True)
|
||||
try:
|
||||
self.query_one("#debug-log", _DebugLogView).show_notice(
|
||||
"(log tail unavailable)"
|
||||
)
|
||||
except Exception: # best-effort notice; never re-raise from here
|
||||
logger.debug("Debug console poll-error notice failed", exc_info=True)
|
||||
|
||||
def _poll_logs_once(self) -> None:
|
||||
"""Append any log records emitted since the last tick to the log view."""
|
||||
log = self.query_one("#debug-log", _DebugLogView)
|
||||
buffer = get_log_buffer()
|
||||
if buffer is None:
|
||||
if not self._missing_notice_shown:
|
||||
log.show_notice("(log buffer unavailable)")
|
||||
self._missing_notice_shown = True
|
||||
return
|
||||
records, total = buffer.snapshot_records_since(self._rendered_upto)
|
||||
self._records.extend(records)
|
||||
pruned = self._prune_records()
|
||||
self._rendered_upto = total
|
||||
if pruned:
|
||||
self._refresh_log_view(scroll_end=log.is_vertical_scroll_end)
|
||||
return
|
||||
log.append_records(self._visible_records(records))
|
||||
|
||||
def _prune_records(self) -> bool:
|
||||
"""Trim retained records to the in-memory ring buffer capacity.
|
||||
|
||||
Returns:
|
||||
`True` when records were pruned.
|
||||
"""
|
||||
overflow = len(self._records) - _RECORD_LIMIT
|
||||
if overflow <= 0:
|
||||
return False
|
||||
del self._records[:overflow]
|
||||
return True
|
||||
|
||||
def _visible_records(
|
||||
self, records: Sequence[InMemoryLogRecord]
|
||||
) -> list[InMemoryLogRecord]:
|
||||
"""Return the subset of *records* matching the current level filter."""
|
||||
return [
|
||||
record
|
||||
for record in records
|
||||
if _record_matches_filter(record, self._level_filter)
|
||||
]
|
||||
|
||||
def _refresh_log_view(self, *, scroll_end: bool) -> None:
|
||||
"""Rebuild the log view using the current filter."""
|
||||
self.query_one("#debug-log", _DebugLogView).set_records(
|
||||
self._visible_records(self._records), scroll_end=scroll_end
|
||||
)
|
||||
|
||||
def action_clear_view(self) -> None:
|
||||
"""Clear the on-screen log view; the in-memory buffer keeps accruing."""
|
||||
self.query_one("#debug-log", _DebugLogView).clear_records()
|
||||
self._records.clear()
|
||||
buffer = get_log_buffer()
|
||||
if buffer is not None:
|
||||
self._rendered_upto = buffer.total_emitted
|
||||
|
||||
def action_copy(self) -> None:
|
||||
"""Copy visible retained log records since the last clear to the clipboard."""
|
||||
lines = [record.plain_line for record in self._visible_records(self._records)]
|
||||
self._copy_lines(lines, empty_message="No visible log lines to copy")
|
||||
|
||||
def _copy_record(self, record: InMemoryLogRecord) -> None:
|
||||
"""Copy a clicked logical log record to the clipboard."""
|
||||
self._copy_lines([record.plain_line], empty_message="No log line to copy")
|
||||
|
||||
def _level_select(self) -> Select[FilterValue]:
|
||||
"""Return the level-filter dropdown."""
|
||||
return cast(
|
||||
"Select[FilterValue]", self.query_one("#debug-level-filter", Select)
|
||||
)
|
||||
|
||||
def _copy_lines(self, lines: Sequence[str], *, empty_message: str) -> None:
|
||||
"""Copy lines to clipboard with user-visible feedback."""
|
||||
text = "\n".join(lines)
|
||||
if not text:
|
||||
self.app.notify(
|
||||
empty_message, severity="information", timeout=2, markup=False
|
||||
)
|
||||
return
|
||||
success, error = copy_text_to_clipboard(self.app, text)
|
||||
if success:
|
||||
self.app.notify(
|
||||
"Debug log copied", severity="information", timeout=2, markup=False
|
||||
)
|
||||
return
|
||||
suffix = f": {error}" if error else ""
|
||||
self.app.notify(
|
||||
f"Failed to copy debug log{suffix}",
|
||||
severity="warning",
|
||||
timeout=3,
|
||||
markup=False,
|
||||
)
|
||||
|
||||
def action_close(self) -> None:
|
||||
"""Close the open level dropdown, or close the debug console."""
|
||||
level_select = self._level_select()
|
||||
if level_select.expanded:
|
||||
level_select.expanded = False
|
||||
level_select.focus()
|
||||
return
|
||||
self.dismiss(None)
|
||||
@@ -24,6 +24,7 @@ _TIPS: dict[str, int] = {
|
||||
"Use /effort high to change the current model's reasoning effort": 1,
|
||||
"Press ctrl+x to compose prompts in your external editor": 1,
|
||||
"Press ctrl+u to delete to the start of the line in the chat input": 1,
|
||||
r"Press Ctrl+\ to open the Debug Console and inspect session state and logs": 1,
|
||||
"Use /skill:<name> to invoke a skill directly": 1,
|
||||
"Type /update to check for and install updates": 1,
|
||||
"Use /install <extra> to add optional dependencies (e.g. /install daytona)": 1,
|
||||
|
||||
@@ -48,7 +48,7 @@ class TestCopyTextToClipboard:
|
||||
|
||||
with (
|
||||
patch("pyperclip.copy", side_effect=RuntimeError("no pyperclip")) as copy,
|
||||
caplog.at_level(logging.DEBUG),
|
||||
caplog.at_level(logging.DEBUG, logger="deepagents_code"),
|
||||
):
|
||||
success, error = copy_text_to_clipboard(mock_app, "hello")
|
||||
|
||||
@@ -66,7 +66,7 @@ class TestCopyTextToClipboard:
|
||||
with (
|
||||
patch("pyperclip.copy", side_effect=RuntimeError("no pyperclip")),
|
||||
patch("deepagents_code.clipboard._copy_osc52") as osc52,
|
||||
caplog.at_level(logging.DEBUG),
|
||||
caplog.at_level(logging.DEBUG, logger="deepagents_code"),
|
||||
):
|
||||
success, error = copy_text_to_clipboard(mock_app, "hello")
|
||||
|
||||
@@ -87,7 +87,7 @@ class TestCopyTextToClipboard:
|
||||
"deepagents_code.clipboard._copy_osc52",
|
||||
side_effect=OSError("no tty"),
|
||||
),
|
||||
caplog.at_level(logging.DEBUG),
|
||||
caplog.at_level(logging.DEBUG, logger="deepagents_code"),
|
||||
):
|
||||
success, error = copy_text_to_clipboard(mock_app, "hello")
|
||||
|
||||
@@ -112,7 +112,7 @@ class TestCopyTextToClipboard:
|
||||
with (
|
||||
patch("pyperclip.copy", side_effect=RuntimeError("no pyperclip")),
|
||||
patch("deepagents_code.clipboard._copy_osc52", side_effect=boom),
|
||||
caplog.at_level(logging.DEBUG),
|
||||
caplog.at_level(logging.DEBUG, logger="deepagents_code"),
|
||||
):
|
||||
success, error = copy_text_to_clipboard(mock_app, "\ud800")
|
||||
|
||||
@@ -385,7 +385,7 @@ class TestCopySelectionToClipboard:
|
||||
mock_widget.get_selection.side_effect = AttributeError("No selection")
|
||||
mock_app.query.return_value = [mock_widget]
|
||||
|
||||
with caplog.at_level(logging.DEBUG):
|
||||
with caplog.at_level(logging.DEBUG, logger="deepagents_code"):
|
||||
copy_selection_to_clipboard(mock_app)
|
||||
|
||||
assert "Failed to get selection from widget" in caplog.text
|
||||
@@ -439,7 +439,7 @@ class TestCopySelectionToClipboard:
|
||||
mock_app.query.return_value = [racy, sibling]
|
||||
|
||||
with (
|
||||
caplog.at_level(logging.DEBUG),
|
||||
caplog.at_level(logging.DEBUG, logger="deepagents_code"),
|
||||
patch(
|
||||
"deepagents_code.clipboard.copy_text_to_clipboard",
|
||||
return_value=(True, None),
|
||||
|
||||
@@ -3229,13 +3229,17 @@ class TestQuietSdkTracingLogging:
|
||||
|
||||
monkeypatch.delenv(DEBUG, raising=False)
|
||||
for name in ("langsmith", "langchain"):
|
||||
logging.getLogger(name).handlers.clear()
|
||||
logger = logging.getLogger(name)
|
||||
logger.handlers.clear()
|
||||
logger.setLevel(logging.NOTSET)
|
||||
|
||||
_quiet_sdk_tracing_logging()
|
||||
|
||||
for name in ("langsmith", "langchain"):
|
||||
handlers = logging.getLogger(name).handlers
|
||||
logger = logging.getLogger(name)
|
||||
handlers = logger.handlers
|
||||
assert any(isinstance(h, logging.NullHandler) for h in handlers)
|
||||
assert logger.level == logging.NOTSET
|
||||
|
||||
def test_idempotent(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Repeated calls do not stack duplicate handlers."""
|
||||
|
||||
@@ -81,6 +81,66 @@ def test_option_keys_unique() -> None:
|
||||
assert len(keys) == len(set(keys))
|
||||
|
||||
|
||||
def test_debug_log_level_resolves_dynamic_default(monkeypatch) -> None:
|
||||
"""The effective log level follows debug mode when no level is explicit."""
|
||||
option = get_option("debug.log_level")
|
||||
assert option is not None
|
||||
monkeypatch.delenv(_env_vars.LOG_LEVEL, raising=False)
|
||||
|
||||
monkeypatch.delenv(_env_vars.DEBUG, raising=False)
|
||||
assert resolve_scalar(option, toml_data={}) == ("INFO", "default")
|
||||
|
||||
monkeypatch.setenv(_env_vars.DEBUG, "true")
|
||||
assert resolve_scalar(option, toml_data={}) == ("DEBUG", "default")
|
||||
|
||||
|
||||
def test_debug_log_level_validates_explicit_value(monkeypatch) -> None:
|
||||
"""Explicit log levels are normalized and invalid values use the runtime default."""
|
||||
option = get_option("debug.log_level")
|
||||
assert option is not None
|
||||
monkeypatch.setenv(_env_vars.DEBUG, "true")
|
||||
|
||||
monkeypatch.setenv(_env_vars.LOG_LEVEL, " warning ")
|
||||
assert resolve_scalar(option, toml_data={}) == (
|
||||
"WARNING",
|
||||
f"env ({_env_vars.LOG_LEVEL})",
|
||||
)
|
||||
|
||||
monkeypatch.setenv(_env_vars.LOG_LEVEL, "TRACE")
|
||||
assert resolve_scalar(option, toml_data={}) == ("DEBUG", "default")
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("debug", "expected"), [(None, "INFO"), ("1", "DEBUG")])
|
||||
def test_config_get_and_show_report_dynamic_log_level(
|
||||
monkeypatch, capsys, debug: str | None, expected: str
|
||||
) -> None:
|
||||
"""Both config command paths report the runtime's effective default."""
|
||||
import json
|
||||
|
||||
monkeypatch.delenv(_env_vars.LOG_LEVEL, raising=False)
|
||||
if debug is None:
|
||||
monkeypatch.delenv(_env_vars.DEBUG, raising=False)
|
||||
else:
|
||||
monkeypatch.setenv(_env_vars.DEBUG, debug)
|
||||
|
||||
get_args = argparse.Namespace(
|
||||
config_command="get",
|
||||
key="debug.log_level",
|
||||
output_format="json",
|
||||
)
|
||||
assert run_config_command(get_args) == 0
|
||||
get_payload = json.loads(capsys.readouterr().out)
|
||||
assert get_payload["data"]["value"] == expected
|
||||
|
||||
show_args = argparse.Namespace(config_command="show", output_format="json")
|
||||
assert run_config_command(show_args) == 0
|
||||
show_payload = json.loads(capsys.readouterr().out)
|
||||
row = next(
|
||||
item for item in show_payload["data"] if item["key"] == "debug.log_level"
|
||||
)
|
||||
assert row["value"] == expected
|
||||
|
||||
|
||||
# --- Provider install helpers ----------------------------------------------
|
||||
|
||||
|
||||
|
||||
@@ -11,9 +11,46 @@ import deepagents_code
|
||||
from deepagents_code._debug import (
|
||||
configure_debug_logging,
|
||||
installed_debug_log_path,
|
||||
resolve_log_level,
|
||||
)
|
||||
|
||||
|
||||
class TestResolveLogLevel:
|
||||
def test_defaults_to_debug_when_debug_enabled(self) -> None:
|
||||
with patch.dict(os.environ, {}, clear=True):
|
||||
assert resolve_log_level(debug_enabled=True) == logging.DEBUG
|
||||
|
||||
def test_defaults_to_info_when_debug_disabled(self) -> None:
|
||||
with patch.dict(os.environ, {}, clear=True):
|
||||
assert resolve_log_level(debug_enabled=False) == logging.INFO
|
||||
|
||||
def test_empty_value_falls_back(self) -> None:
|
||||
with patch.dict(os.environ, {"DEEPAGENTS_CODE_LOG_LEVEL": ""}, clear=True):
|
||||
assert resolve_log_level(debug_enabled=False) == logging.INFO
|
||||
|
||||
def test_whitespace_value_falls_back(self) -> None:
|
||||
with patch.dict(os.environ, {"DEEPAGENTS_CODE_LOG_LEVEL": " "}, clear=True):
|
||||
assert resolve_log_level(debug_enabled=True) == logging.DEBUG
|
||||
|
||||
def test_value_is_case_insensitive(self) -> None:
|
||||
with patch.dict(
|
||||
os.environ, {"DEEPAGENTS_CODE_LOG_LEVEL": "warning"}, clear=True
|
||||
):
|
||||
assert resolve_log_level(debug_enabled=False) == logging.WARNING
|
||||
|
||||
def test_explicit_level_overrides_debug_fallback(self) -> None:
|
||||
"""An explicit level wins over the debug-derived default."""
|
||||
with patch.dict(os.environ, {"DEEPAGENTS_CODE_LOG_LEVEL": "ERROR"}, clear=True):
|
||||
assert resolve_log_level(debug_enabled=True) == logging.ERROR
|
||||
|
||||
def test_reads_debug_env_when_flag_omitted(self) -> None:
|
||||
"""With no explicit flag, the truthiness of the env var decides."""
|
||||
with patch.dict(os.environ, {"DEEPAGENTS_CODE_DEBUG": "1"}, clear=True):
|
||||
assert resolve_log_level() == logging.DEBUG
|
||||
with patch.dict(os.environ, {}, clear=True):
|
||||
assert resolve_log_level() == logging.INFO
|
||||
|
||||
|
||||
class TestConfigureDebugLogging:
|
||||
def test_noop_when_env_unset(self) -> None:
|
||||
"""No handlers should be added when DEEPAGENTS_CODE_DEBUG is unset."""
|
||||
@@ -39,6 +76,46 @@ class TestConfigureDebugLogging:
|
||||
h.close()
|
||||
logger.removeHandler(h)
|
||||
|
||||
def test_log_level_debug_enables_debug_without_file_handler(self) -> None:
|
||||
logger = logging.getLogger("test.debug.level_only")
|
||||
logger.handlers = []
|
||||
with patch.dict(os.environ, {"DEEPAGENTS_CODE_LOG_LEVEL": "DEBUG"}, clear=True):
|
||||
configure_debug_logging(logger)
|
||||
assert logger.level == logging.DEBUG
|
||||
assert not any(isinstance(h, logging.FileHandler) for h in logger.handlers)
|
||||
|
||||
def test_debug_file_can_use_info_runtime_level(self, tmp_path) -> None:
|
||||
logger = logging.getLogger("test.debug.file_info_level")
|
||||
log_file = tmp_path / "debug.log"
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
"DEEPAGENTS_CODE_DEBUG": "1",
|
||||
"DEEPAGENTS_CODE_DEBUG_FILE": str(log_file),
|
||||
"DEEPAGENTS_CODE_LOG_LEVEL": "INFO",
|
||||
},
|
||||
):
|
||||
configure_debug_logging(logger)
|
||||
file_handlers = [
|
||||
h for h in logger.handlers if isinstance(h, logging.FileHandler)
|
||||
]
|
||||
try:
|
||||
assert logger.level == logging.INFO
|
||||
assert file_handlers
|
||||
assert file_handlers[-1].level == logging.INFO
|
||||
finally:
|
||||
for h in file_handlers:
|
||||
h.close()
|
||||
logger.removeHandler(h)
|
||||
|
||||
def test_invalid_log_level_warns_and_uses_default(self, capsys) -> None:
|
||||
logger = logging.getLogger("test.debug.bad_level")
|
||||
with patch.dict(os.environ, {"DEEPAGENTS_CODE_LOG_LEVEL": "TRACE"}, clear=True):
|
||||
configure_debug_logging(logger)
|
||||
assert logger.level == logging.INFO
|
||||
captured = capsys.readouterr()
|
||||
assert "DEEPAGENTS_CODE_LOG_LEVEL" in captured.err
|
||||
|
||||
def test_custom_path_used(self, tmp_path) -> None:
|
||||
logger = logging.getLogger("test.debug.custom_path")
|
||||
log_file = tmp_path / "custom.log"
|
||||
|
||||
@@ -0,0 +1,269 @@
|
||||
"""Tests for the in-memory log ring buffer backing the Debug Console."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
import threading
|
||||
from typing import TYPE_CHECKING
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
import deepagents_code._debug_buffer as debug_buffer
|
||||
from deepagents_code._debug_buffer import (
|
||||
InMemoryLogBuffer,
|
||||
get_log_buffer,
|
||||
install_log_buffer,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Generator
|
||||
|
||||
|
||||
def _record(name: str, message: str, level: int = logging.INFO) -> logging.LogRecord:
|
||||
return logging.LogRecord(
|
||||
name=name,
|
||||
level=level,
|
||||
pathname=__file__,
|
||||
lineno=1,
|
||||
msg=message,
|
||||
args=(),
|
||||
exc_info=None,
|
||||
)
|
||||
|
||||
|
||||
def _lines(buffer: InMemoryLogBuffer, index: int) -> list[str]:
|
||||
"""Return the retained plain-text lines from *index* onward."""
|
||||
records, _total = buffer.snapshot_records_since(index)
|
||||
return [record.plain_line for record in records]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def _restore_global_buffer() -> Generator[None]:
|
||||
"""Preserve the module-level singleton across install-mutating tests.
|
||||
|
||||
Yields:
|
||||
None; the original singleton is restored on teardown.
|
||||
"""
|
||||
original = debug_buffer._buffer
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
debug_buffer._buffer = original
|
||||
|
||||
|
||||
class TestInMemoryLogBuffer:
|
||||
def test_captures_and_formats_records(self) -> None:
|
||||
buffer = InMemoryLogBuffer()
|
||||
buffer.emit(_record("deepagents_code.x", "hello"))
|
||||
lines = _lines(buffer, 0)
|
||||
assert len(lines) == 1
|
||||
assert "hello" in lines[0]
|
||||
assert "INFO" in lines[0]
|
||||
assert "deepagents_code.x" in lines[0]
|
||||
|
||||
def test_rejects_non_positive_capacity(self) -> None:
|
||||
with pytest.raises(ValueError, match="capacity must be >= 1"):
|
||||
InMemoryLogBuffer(capacity=0)
|
||||
with pytest.raises(ValueError, match="capacity must be >= 1"):
|
||||
InMemoryLogBuffer(capacity=-1)
|
||||
|
||||
def test_records_carry_numeric_level(self) -> None:
|
||||
buffer = InMemoryLogBuffer()
|
||||
buffer.emit(_record("deepagents_code.x", "hello", level=logging.WARNING))
|
||||
records, _total = buffer.snapshot_records_since(0)
|
||||
assert records[0].level == "WARNING"
|
||||
assert records[0].levelno == logging.WARNING
|
||||
|
||||
def test_captures_exception_traceback_in_message(self) -> None:
|
||||
buffer = InMemoryLogBuffer()
|
||||
try:
|
||||
msg = "boom-detail"
|
||||
raise ValueError(msg) # noqa: TRY301 # deliberately raised to capture a traceback
|
||||
except ValueError:
|
||||
record = logging.LogRecord(
|
||||
name="deepagents_code.x",
|
||||
level=logging.ERROR,
|
||||
pathname=__file__,
|
||||
lineno=1,
|
||||
msg="handler failed",
|
||||
args=(),
|
||||
exc_info=sys.exc_info(),
|
||||
)
|
||||
buffer.emit(record)
|
||||
|
||||
records, _total = buffer.snapshot_records_since(0)
|
||||
message = records[0].message
|
||||
# The exception text is appended to the message as multiple lines.
|
||||
assert "handler failed" in message
|
||||
assert "Traceback" in message
|
||||
assert "boom-detail" in message
|
||||
assert "\n" in message
|
||||
|
||||
def test_emit_never_raises_on_malformed_record(self) -> None:
|
||||
"""A record that can't be formatted is dropped, not propagated.
|
||||
|
||||
`emit` upholds logging's "never crash the caller" contract: a bad
|
||||
printf-style record routes to `handleError` instead of raising, and the
|
||||
dropped record must not inflate `total_emitted` (the `+= 1` sits after
|
||||
the append that raised).
|
||||
"""
|
||||
buffer = InMemoryLogBuffer()
|
||||
bad = logging.LogRecord(
|
||||
name="deepagents_code.x",
|
||||
level=logging.INFO,
|
||||
pathname=__file__,
|
||||
lineno=1,
|
||||
msg="%d", # %-format expects an int; the str arg raises in getMessage
|
||||
args=("not-an-int",),
|
||||
exc_info=None,
|
||||
)
|
||||
with patch.object(buffer, "handleError") as handle_error:
|
||||
buffer.emit(bad) # must not raise
|
||||
handle_error.assert_called_once_with(bad)
|
||||
records, total = buffer.snapshot_records_since(0)
|
||||
assert records == []
|
||||
assert total == 0
|
||||
assert buffer.total_emitted == 0
|
||||
|
||||
def test_is_bounded_dropping_oldest(self) -> None:
|
||||
buffer = InMemoryLogBuffer(capacity=3)
|
||||
for i in range(5):
|
||||
buffer.emit(_record("deepagents_code", f"msg{i}"))
|
||||
lines = _lines(buffer, 0)
|
||||
assert len(lines) == 3
|
||||
assert "msg2" in lines[0]
|
||||
assert "msg4" in lines[-1]
|
||||
assert buffer.total_emitted == 5
|
||||
|
||||
def test_snapshot_uses_absolute_index(self) -> None:
|
||||
buffer = InMemoryLogBuffer(capacity=10)
|
||||
for i in range(4):
|
||||
buffer.emit(_record("deepagents_code", f"msg{i}"))
|
||||
tail = _lines(buffer, 2)
|
||||
assert len(tail) == 2
|
||||
assert "msg2" in tail[0]
|
||||
assert _lines(buffer, 4) == []
|
||||
assert len(_lines(buffer, -5)) == 4
|
||||
|
||||
def test_snapshot_after_eviction(self) -> None:
|
||||
buffer = InMemoryLogBuffer(capacity=2)
|
||||
for i in range(5):
|
||||
buffer.emit(_record("deepagents_code", f"msg{i}"))
|
||||
assert len(_lines(buffer, 0)) == 2
|
||||
assert _lines(buffer, 4) == [_lines(buffer, 0)[-1]]
|
||||
|
||||
def test_snapshot_since_returns_records_and_next_index(self) -> None:
|
||||
buffer = InMemoryLogBuffer(capacity=10)
|
||||
for i in range(3):
|
||||
buffer.emit(_record("deepagents_code", f"msg{i}"))
|
||||
|
||||
records, total = buffer.snapshot_records_since(1)
|
||||
|
||||
assert total == 3
|
||||
assert len(records) == 2
|
||||
assert "msg1" in records[0].message
|
||||
assert "msg2" in records[1].message
|
||||
|
||||
def test_snapshot_records_since_returns_structured_records(self) -> None:
|
||||
buffer = InMemoryLogBuffer(capacity=10)
|
||||
buffer.emit(_record("deepagents_code.x", "hello", level=logging.WARNING))
|
||||
|
||||
records, total = buffer.snapshot_records_since(0)
|
||||
|
||||
assert total == 1
|
||||
assert len(records) == 1
|
||||
assert records[0].level == "WARNING"
|
||||
assert records[0].logger == "deepagents_code.x"
|
||||
assert records[0].message == "hello"
|
||||
assert records[0].plain_line in _lines(buffer, 0)
|
||||
|
||||
def test_snapshot_is_safe_during_concurrent_emit(self) -> None:
|
||||
buffer = InMemoryLogBuffer(capacity=50)
|
||||
stop = threading.Event()
|
||||
errors: list[BaseException] = []
|
||||
|
||||
def emit_records() -> None:
|
||||
i = 0
|
||||
while not stop.is_set():
|
||||
try:
|
||||
buffer.emit(_record("deepagents_code", f"msg{i}"))
|
||||
except BaseException as exc: # noqa: BLE001 # report thread failures
|
||||
errors.append(exc)
|
||||
stop.set()
|
||||
i += 1
|
||||
|
||||
thread = threading.Thread(target=emit_records)
|
||||
thread.start()
|
||||
try:
|
||||
for _ in range(500):
|
||||
records, total = buffer.snapshot_records_since(0)
|
||||
# No torn read: the resume index is never behind the records.
|
||||
assert total >= len(records)
|
||||
finally:
|
||||
stop.set()
|
||||
thread.join(timeout=1)
|
||||
|
||||
assert not errors
|
||||
|
||||
|
||||
class TestInstallLogBuffer:
|
||||
@pytest.mark.usefixtures("_restore_global_buffer")
|
||||
def test_install_is_idempotent(self) -> None:
|
||||
logger = logging.getLogger("deepagents_code._test_idempotent")
|
||||
logger.handlers = []
|
||||
first = install_log_buffer(logger)
|
||||
second = install_log_buffer(logger)
|
||||
assert first is second
|
||||
installed = [h for h in logger.handlers if isinstance(h, InMemoryLogBuffer)]
|
||||
assert len(installed) == 1
|
||||
assert get_log_buffer() is first
|
||||
|
||||
@pytest.mark.usefixtures("_restore_global_buffer")
|
||||
def test_install_lowers_level_to_info_only(self) -> None:
|
||||
logger = logging.getLogger("deepagents_code._test_level_notset")
|
||||
logger.handlers = []
|
||||
logger.setLevel(logging.NOTSET)
|
||||
install_log_buffer(logger)
|
||||
assert logger.level == logging.INFO
|
||||
|
||||
@pytest.mark.usefixtures("_restore_global_buffer")
|
||||
def test_install_preserves_debug_level(self) -> None:
|
||||
logger = logging.getLogger("deepagents_code._test_level_debug")
|
||||
logger.handlers = []
|
||||
logger.setLevel(logging.DEBUG)
|
||||
install_log_buffer(logger)
|
||||
assert logger.level == logging.DEBUG
|
||||
|
||||
@pytest.mark.usefixtures("_restore_global_buffer")
|
||||
def test_install_lowers_high_level_without_env(self) -> None:
|
||||
# A logger above INFO with no explicit env override drops to INFO so the
|
||||
# always-on tail stays useful even when DEEPAGENTS_CODE_DEBUG is off.
|
||||
logger = logging.getLogger("deepagents_code._test_level_high_no_env")
|
||||
logger.handlers = []
|
||||
logger.setLevel(logging.WARNING)
|
||||
with patch.dict(os.environ, {}, clear=True):
|
||||
install_log_buffer(logger)
|
||||
assert logger.level == logging.INFO
|
||||
|
||||
@pytest.mark.usefixtures("_restore_global_buffer")
|
||||
def test_install_preserves_explicit_env_log_level(self) -> None:
|
||||
logger = logging.getLogger("deepagents_code._test_level_warning")
|
||||
logger.handlers = []
|
||||
logger.setLevel(logging.WARNING)
|
||||
with patch.dict(os.environ, {"DEEPAGENTS_CODE_LOG_LEVEL": "WARNING"}):
|
||||
install_log_buffer(logger)
|
||||
assert logger.level == logging.WARNING
|
||||
|
||||
@pytest.mark.usefixtures("_restore_global_buffer")
|
||||
def test_captures_propagated_records(self) -> None:
|
||||
logger = logging.getLogger("deepagents_code._test_capture")
|
||||
logger.handlers = []
|
||||
logger.setLevel(logging.NOTSET)
|
||||
buffer = install_log_buffer(logger)
|
||||
before = buffer.total_emitted
|
||||
logger.info("captured-line")
|
||||
assert buffer.total_emitted == before + 1
|
||||
assert any("captured-line" in line for line in _lines(buffer, before))
|
||||
@@ -0,0 +1,806 @@
|
||||
r"""Tests for the Debug Console modal and its `Ctrl+\` / `/debug` toggle."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from types import SimpleNamespace
|
||||
from typing import TYPE_CHECKING, cast
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from textual.app import App, ComposeResult
|
||||
from textual.screen import ModalScreen
|
||||
from textual.widgets import Select, Static
|
||||
from textual.widgets._select import SelectOverlay
|
||||
|
||||
import deepagents_code.tui.widgets.debug_console as debug_console_mod
|
||||
from deepagents_code._debug_buffer import InMemoryLogRecord, get_log_buffer
|
||||
from deepagents_code.app import DeepAgentsApp
|
||||
from deepagents_code.tui.widgets.debug_console import (
|
||||
DebugConsoleScreen,
|
||||
SnapshotField,
|
||||
_DebugLogView,
|
||||
_record_matches_filter,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
import pytest
|
||||
|
||||
from deepagents_code.tui.widgets.debug_console import FilterValue
|
||||
|
||||
logger = logging.getLogger("deepagents_code._test_console")
|
||||
|
||||
|
||||
def _widget_text(widget: Static) -> str:
|
||||
return str(widget.render())
|
||||
|
||||
|
||||
def _log_record(
|
||||
message: str, *, level: str = "INFO", levelno: int = logging.INFO
|
||||
) -> InMemoryLogRecord:
|
||||
return InMemoryLogRecord(
|
||||
timestamp="12:00:00",
|
||||
level=level,
|
||||
levelno=levelno,
|
||||
logger="deepagents_code._test_console",
|
||||
message=message,
|
||||
)
|
||||
|
||||
|
||||
class _Harness(App[None]):
|
||||
"""Minimal app wrapper for testing `DebugConsoleScreen` in isolation."""
|
||||
|
||||
def compose(self) -> ComposeResult:
|
||||
yield Static("base")
|
||||
|
||||
|
||||
def _snapshot() -> list[SnapshotField]:
|
||||
return [
|
||||
SnapshotField("Version", "9.9.9"),
|
||||
SnapshotField("Model", "openai:gpt-test"),
|
||||
SnapshotField("CWD", "/tmp/[brackets]/work"),
|
||||
]
|
||||
|
||||
|
||||
class TestDebugConsoleScreen:
|
||||
async def test_renders_snapshot_fields(self) -> None:
|
||||
app = _Harness()
|
||||
async with app.run_test() as pilot:
|
||||
screen = DebugConsoleScreen(_snapshot())
|
||||
app.push_screen(screen)
|
||||
await pilot.pause()
|
||||
|
||||
text = _widget_text(screen.query_one(".debug-console-snapshot", Static))
|
||||
assert "Version" in text
|
||||
assert "9.9.9" in text
|
||||
assert "openai:gpt-test" in text
|
||||
assert "/tmp/[brackets]/work" in text
|
||||
|
||||
async def test_live_tail_writes_buffered_records(self) -> None:
|
||||
logger.info("debug-console-tail-marker")
|
||||
app = _Harness()
|
||||
async with app.run_test() as pilot:
|
||||
screen = DebugConsoleScreen(_snapshot())
|
||||
app.push_screen(screen)
|
||||
await pilot.pause()
|
||||
log = screen.query_one("#debug-log", _DebugLogView)
|
||||
assert log.line_count > 0
|
||||
|
||||
async def test_no_color_renders_every_segment_with_a_style(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""Monochrome filtering must never receive an unstyled Rich segment."""
|
||||
monkeypatch.setenv("NO_COLOR", "1")
|
||||
app = _Harness()
|
||||
async with app.run_test() as pilot:
|
||||
screen = DebugConsoleScreen(_snapshot())
|
||||
app.push_screen(screen)
|
||||
await pilot.pause()
|
||||
log = screen.query_one("#debug-log", _DebugLogView)
|
||||
log.set_records([_log_record("unstyled message")])
|
||||
await pilot.pause()
|
||||
|
||||
strip = log.render_line(0)
|
||||
|
||||
assert strip._segments
|
||||
assert all(segment.style is not None for segment in strip._segments)
|
||||
|
||||
async def test_live_tail_appends_new_records_incrementally(self) -> None:
|
||||
app = _Harness()
|
||||
async with app.run_test() as pilot:
|
||||
screen = DebugConsoleScreen(_snapshot())
|
||||
app.push_screen(screen)
|
||||
await pilot.pause()
|
||||
log = screen.query_one("#debug-log", _DebugLogView)
|
||||
before = len(log.records)
|
||||
|
||||
logger.info("debug-console-incremental-marker")
|
||||
screen._poll_logs()
|
||||
await pilot.pause()
|
||||
|
||||
# Exactly one new record appended; already-consumed records not re-written.
|
||||
assert len(log.records) == before + 1
|
||||
|
||||
async def test_live_tail_stays_bounded(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
records = [_log_record(f"debug-console-bounded-{index}") for index in range(5)]
|
||||
|
||||
class FakeBuffer:
|
||||
total_emitted = len(records)
|
||||
|
||||
def snapshot_records_since(
|
||||
self, index: int
|
||||
) -> tuple[list[InMemoryLogRecord], int]:
|
||||
if index >= self.total_emitted:
|
||||
return [], self.total_emitted
|
||||
return records[index:], self.total_emitted
|
||||
|
||||
monkeypatch.setattr(debug_console_mod, "_RECORD_LIMIT", 3)
|
||||
monkeypatch.setattr(debug_console_mod, "get_log_buffer", lambda: FakeBuffer())
|
||||
|
||||
class FakeLog:
|
||||
is_vertical_scroll_end = True
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.records: list[InMemoryLogRecord] = []
|
||||
|
||||
def append_records(self, records: list[InMemoryLogRecord]) -> None:
|
||||
self.records.extend(records)
|
||||
|
||||
def set_records(
|
||||
self,
|
||||
records: list[InMemoryLogRecord],
|
||||
*,
|
||||
scroll_end: bool,
|
||||
) -> None:
|
||||
_ = scroll_end
|
||||
self.records = list(records)
|
||||
|
||||
def fake_query_one(*args: object) -> FakeLog:
|
||||
_ = args
|
||||
return log
|
||||
|
||||
log = FakeLog()
|
||||
screen = DebugConsoleScreen(_snapshot())
|
||||
monkeypatch.setattr(screen, "query_one", fake_query_one)
|
||||
|
||||
screen._poll_logs()
|
||||
|
||||
messages = [record.message for record in screen._records]
|
||||
visible_messages = [record.message for record in log.records]
|
||||
|
||||
assert messages == [
|
||||
"debug-console-bounded-2",
|
||||
"debug-console-bounded-3",
|
||||
"debug-console-bounded-4",
|
||||
]
|
||||
assert visible_messages == messages
|
||||
|
||||
async def test_poll_degrades_on_buffer_failure_without_crashing(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
app = _Harness()
|
||||
async with app.run_test() as pilot:
|
||||
screen = DebugConsoleScreen(_snapshot())
|
||||
app.push_screen(screen)
|
||||
await pilot.pause()
|
||||
log = screen.query_one("#debug-log", _DebugLogView)
|
||||
|
||||
class BoomBuffer:
|
||||
total_emitted = 0
|
||||
|
||||
def snapshot_records_since(
|
||||
self, _index: int
|
||||
) -> tuple[list[InMemoryLogRecord], int]:
|
||||
msg = "poll boom"
|
||||
raise RuntimeError(msg)
|
||||
|
||||
monkeypatch.setattr(
|
||||
debug_console_mod, "get_log_buffer", lambda: BoomBuffer()
|
||||
)
|
||||
|
||||
# Must not raise: the repeating timer would otherwise crash the app.
|
||||
screen._poll_logs()
|
||||
await pilot.pause()
|
||||
|
||||
assert log._notice is not None
|
||||
assert isinstance(app.screen, DebugConsoleScreen) # app still alive
|
||||
|
||||
async def test_notice_replaced_by_incoming_records(self) -> None:
|
||||
app = _Harness()
|
||||
async with app.run_test() as pilot:
|
||||
screen = DebugConsoleScreen(_snapshot())
|
||||
app.push_screen(screen)
|
||||
await pilot.pause()
|
||||
log = screen.query_one("#debug-log", _DebugLogView)
|
||||
|
||||
log.show_notice("(log buffer unavailable)")
|
||||
await pilot.pause()
|
||||
assert log._notice is not None
|
||||
|
||||
log.append_records([_log_record("debug-console-recovery-marker")])
|
||||
await pilot.pause()
|
||||
|
||||
assert log._notice is None
|
||||
assert any(
|
||||
"debug-console-recovery-marker" in record.message
|
||||
for record in log.records
|
||||
)
|
||||
|
||||
async def test_poll_applies_active_filter_to_new_records(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
monkeypatch.setattr(debug_console_mod, "_debug_records_enabled", lambda: True)
|
||||
app = _Harness()
|
||||
async with app.run_test() as pilot:
|
||||
screen = DebugConsoleScreen(_snapshot())
|
||||
app.push_screen(screen)
|
||||
await pilot.pause()
|
||||
log = screen.query_one("#debug-log", _DebugLogView)
|
||||
select = screen.query_one("#debug-level-filter", Select)
|
||||
select.value = "min:WARNING"
|
||||
await pilot.pause()
|
||||
before = len(log.records)
|
||||
|
||||
logger.info("debug-console-poll-filter-info")
|
||||
logger.error("debug-console-poll-filter-error")
|
||||
screen._poll_logs()
|
||||
await pilot.pause()
|
||||
|
||||
new_messages = [record.message for record in log.records[before:]]
|
||||
assert any("poll-filter-error" in message for message in new_messages)
|
||||
assert not any("poll-filter-info" in message for message in new_messages)
|
||||
|
||||
async def test_empty_filtered_copy_notifies_information(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
def fail_copy(_app: App, _text: str) -> tuple[bool, str | None]:
|
||||
msg = "copy should not be attempted for an empty selection"
|
||||
raise AssertionError(msg)
|
||||
|
||||
monkeypatch.setattr(debug_console_mod, "copy_text_to_clipboard", fail_copy)
|
||||
|
||||
logger.info("debug-console-empty-copy-marker")
|
||||
app = _Harness()
|
||||
async with app.run_test() as pilot:
|
||||
screen = DebugConsoleScreen(_snapshot())
|
||||
app.push_screen(screen)
|
||||
await pilot.pause()
|
||||
# No CRITICAL records exist, so the filtered view is empty.
|
||||
screen.query_one("#debug-level-filter", Select).value = "only:CRITICAL"
|
||||
await pilot.pause()
|
||||
|
||||
await pilot.press("c")
|
||||
await pilot.pause()
|
||||
|
||||
latest = list(app._notifications)[-1]
|
||||
assert latest.severity == "information"
|
||||
assert "No visible log lines" in latest.message
|
||||
|
||||
async def test_wrapped_record_maps_clicks_and_arrows_as_one_unit(self) -> None:
|
||||
logger.info("W" * 240) # long enough to wrap to multiple visual lines
|
||||
logger.info("debug-console-after-wrap-marker")
|
||||
app = _Harness()
|
||||
async with app.run_test(size=(80, 24)) as pilot:
|
||||
screen = DebugConsoleScreen(_snapshot())
|
||||
app.push_screen(screen)
|
||||
await pilot.pause()
|
||||
log = screen.query_one("#debug-log", _DebugLogView)
|
||||
wrap_index = next(
|
||||
index
|
||||
for index, record in enumerate(log.records)
|
||||
if record.message.startswith("WWW")
|
||||
)
|
||||
start = log._wrap_prefix[wrap_index]
|
||||
span = log._wrap_prefix[wrap_index + 1] - start
|
||||
|
||||
# The record occupies more visual lines than it is logical records.
|
||||
assert span > 1
|
||||
assert log.line_count > len(log.records)
|
||||
|
||||
# A click on the continuation line maps back to the same record.
|
||||
assert log._record_at_visual_y(start + 1) is log.records[wrap_index]
|
||||
|
||||
# Selecting it highlights every visual row it spans, not just the first.
|
||||
log._select_record(wrap_index)
|
||||
await pilot.pause()
|
||||
scroll_y = log.scroll_offset.y
|
||||
for visual_y in (start, start + 1):
|
||||
strip = log.render_line(visual_y - scroll_y)
|
||||
assert strip._segments[0].style is not None
|
||||
assert strip._segments[0].style.bgcolor is not None
|
||||
|
||||
# One arrow press moves a whole logical record across the wrap.
|
||||
log.focus()
|
||||
log._select_record(wrap_index + 1)
|
||||
await pilot.press("up")
|
||||
await pilot.pause()
|
||||
assert log._selected_index == wrap_index
|
||||
|
||||
async def test_multiline_record_spans_multiple_visual_lines(self) -> None:
|
||||
record = _log_record("first-line\nsecond-line\nthird-line")
|
||||
app = _Harness()
|
||||
async with app.run_test() as pilot:
|
||||
screen = DebugConsoleScreen(_snapshot())
|
||||
app.push_screen(screen)
|
||||
await pilot.pause()
|
||||
log = screen.query_one("#debug-log", _DebugLogView)
|
||||
log.set_records([record])
|
||||
await pilot.pause()
|
||||
|
||||
# One logical record, three visual lines from the embedded newlines.
|
||||
assert len(log.records) == 1
|
||||
assert log.line_count >= 3
|
||||
|
||||
async def test_selecting_offscreen_record_scrolls_into_view(self) -> None:
|
||||
for index in range(120):
|
||||
logger.info("debug-console-scroll-marker-%s", index)
|
||||
app = _Harness()
|
||||
async with app.run_test(size=(80, 24)) as pilot:
|
||||
screen = DebugConsoleScreen(_snapshot())
|
||||
app.push_screen(screen)
|
||||
await pilot.pause()
|
||||
log = screen.query_one("#debug-log", _DebugLogView)
|
||||
assert log.line_count > log.size.height # content overflows the viewport
|
||||
log.focus()
|
||||
|
||||
log._select_record(0) # jump to the top
|
||||
await pilot.pause()
|
||||
assert log.scroll_offset.y == 0
|
||||
|
||||
log._select_record(len(log.records) - 1) # jump to the bottom
|
||||
await pilot.pause()
|
||||
assert log.scroll_offset.y > 0
|
||||
|
||||
async def test_empty_snapshot_renders_placeholder(self) -> None:
|
||||
app = _Harness()
|
||||
async with app.run_test() as pilot:
|
||||
screen = DebugConsoleScreen([])
|
||||
app.push_screen(screen)
|
||||
await pilot.pause()
|
||||
text = _widget_text(screen.query_one(".debug-console-snapshot", Static))
|
||||
assert "no session data" in text
|
||||
|
||||
async def test_poll_logs_handles_missing_buffer_once(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
monkeypatch.setattr(debug_console_mod, "get_log_buffer", lambda: None)
|
||||
app = _Harness()
|
||||
async with app.run_test() as pilot:
|
||||
screen = DebugConsoleScreen(_snapshot())
|
||||
app.push_screen(screen)
|
||||
await pilot.pause()
|
||||
log = screen.query_one("#debug-log", _DebugLogView)
|
||||
assert log.line_count == 1 # on_mount poll writes the notice
|
||||
|
||||
screen._poll_logs()
|
||||
await pilot.pause()
|
||||
assert log.line_count == 1 # one-shot guard: notice not repeated
|
||||
|
||||
async def test_clear_view_key_empties_log_and_advances_pointer(self) -> None:
|
||||
logger.info("debug-console-clear-marker")
|
||||
app = _Harness()
|
||||
async with app.run_test() as pilot:
|
||||
screen = DebugConsoleScreen(_snapshot())
|
||||
app.push_screen(screen)
|
||||
await pilot.pause()
|
||||
buffer = get_log_buffer()
|
||||
assert buffer is not None
|
||||
|
||||
await pilot.press("ctrl+l")
|
||||
await pilot.pause()
|
||||
log = screen.query_one("#debug-log", _DebugLogView)
|
||||
assert log.line_count == 0
|
||||
assert screen._rendered_upto == buffer.total_emitted
|
||||
|
||||
async def test_copy_key_invokes_clipboard_with_retained_lines(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
captured: dict[str, str] = {}
|
||||
|
||||
def fake_copy(_app: App, text: str) -> tuple[bool, str | None]:
|
||||
captured["text"] = text
|
||||
return True, None
|
||||
|
||||
monkeypatch.setattr(debug_console_mod, "copy_text_to_clipboard", fake_copy)
|
||||
|
||||
logger.info("debug-console-copy-marker")
|
||||
app = _Harness()
|
||||
async with app.run_test() as pilot:
|
||||
screen = DebugConsoleScreen(_snapshot())
|
||||
app.push_screen(screen)
|
||||
await pilot.pause()
|
||||
await pilot.press("c")
|
||||
await pilot.pause()
|
||||
|
||||
assert "debug-console-copy-marker" in captured["text"]
|
||||
|
||||
async def test_level_filter_supports_threshold_and_exact_modes(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
monkeypatch.setattr(debug_console_mod, "_debug_records_enabled", lambda: True)
|
||||
logger.info("debug-console-filter-info-marker")
|
||||
logger.warning("debug-console-filter-warning-marker")
|
||||
logger.error("debug-console-filter-error-marker")
|
||||
app = _Harness()
|
||||
async with app.run_test() as pilot:
|
||||
screen = DebugConsoleScreen(_snapshot())
|
||||
app.push_screen(screen)
|
||||
await pilot.pause()
|
||||
log = screen.query_one("#debug-log", _DebugLogView)
|
||||
select = screen.query_one("#debug-level-filter", Select)
|
||||
|
||||
select.value = "min:WARNING"
|
||||
await pilot.pause()
|
||||
|
||||
assert not any(
|
||||
"debug-console-filter-info-marker" in record.message
|
||||
for record in log.records
|
||||
)
|
||||
assert any(
|
||||
"debug-console-filter-warning-marker" in record.message
|
||||
for record in log.records
|
||||
)
|
||||
assert any(
|
||||
"debug-console-filter-error-marker" in record.message
|
||||
for record in log.records
|
||||
)
|
||||
|
||||
select.value = "min:DEBUG"
|
||||
await pilot.pause()
|
||||
|
||||
assert any(
|
||||
"debug-console-filter-info-marker" in record.message
|
||||
for record in log.records
|
||||
)
|
||||
assert any(
|
||||
"debug-console-filter-error-marker" in record.message
|
||||
for record in log.records
|
||||
)
|
||||
|
||||
select.value = "only:WARNING"
|
||||
await pilot.pause()
|
||||
|
||||
assert any(
|
||||
"debug-console-filter-warning-marker" in record.message
|
||||
for record in log.records
|
||||
)
|
||||
assert not any(
|
||||
"debug-console-filter-error-marker" in record.message
|
||||
for record in log.records
|
||||
)
|
||||
|
||||
async def test_level_filter_hides_debug_when_runtime_level_excludes_it(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
monkeypatch.setattr(debug_console_mod, "_debug_records_enabled", lambda: False)
|
||||
app = _Harness()
|
||||
async with app.run_test() as pilot:
|
||||
screen = DebugConsoleScreen(_snapshot())
|
||||
app.push_screen(screen)
|
||||
await pilot.pause()
|
||||
select = screen.query_one("#debug-level-filter", Select)
|
||||
values = {value for _prompt, value in select._options}
|
||||
|
||||
assert "min:DEBUG" not in values
|
||||
assert "only:DEBUG" not in values
|
||||
assert "min:INFO" in values
|
||||
|
||||
async def test_arrow_keys_move_between_logical_records(self) -> None:
|
||||
logger.info("debug-console-arrow-first-marker")
|
||||
logger.info("debug-console-arrow-second-marker")
|
||||
app = _Harness()
|
||||
async with app.run_test() as pilot:
|
||||
screen = DebugConsoleScreen(_snapshot())
|
||||
app.push_screen(screen)
|
||||
await pilot.pause()
|
||||
log = screen.query_one("#debug-log", _DebugLogView)
|
||||
log.focus()
|
||||
second_index = next(
|
||||
index
|
||||
for index, record in enumerate(log.records)
|
||||
if "debug-console-arrow-second-marker" in record.message
|
||||
)
|
||||
log._select_record(second_index)
|
||||
|
||||
await pilot.press("up")
|
||||
await pilot.pause()
|
||||
assert log._selected_index == second_index - 1
|
||||
|
||||
await pilot.press("down")
|
||||
await pilot.pause()
|
||||
assert log._selected_index == second_index
|
||||
|
||||
async def test_selected_row_highlight_extends_to_full_width(self) -> None:
|
||||
logger.info("debug-console-full-width-marker")
|
||||
app = _Harness()
|
||||
async with app.run_test() as pilot:
|
||||
screen = DebugConsoleScreen(_snapshot())
|
||||
app.push_screen(screen)
|
||||
await pilot.pause()
|
||||
log = screen.query_one("#debug-log", _DebugLogView)
|
||||
index = next(
|
||||
index
|
||||
for index, record in enumerate(log.records)
|
||||
if "debug-console-full-width-marker" in record.message
|
||||
)
|
||||
log._select_record(index)
|
||||
strip = log.render_line(int(log._wrap_prefix[index] - log.scroll_y))
|
||||
|
||||
first_segment = strip._segments[0]
|
||||
trailing_segment = strip._segments[-1]
|
||||
assert trailing_segment.text
|
||||
assert first_segment.style is not None
|
||||
assert trailing_segment.style is not None
|
||||
assert first_segment.style.bgcolor is not None
|
||||
assert trailing_segment.style.bgcolor == first_segment.style.bgcolor
|
||||
|
||||
async def test_enter_copies_selected_logical_record(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
captured: dict[str, str] = {}
|
||||
|
||||
def fake_copy(_app: App, text: str) -> tuple[bool, str | None]:
|
||||
captured["text"] = text
|
||||
return True, None
|
||||
|
||||
monkeypatch.setattr(debug_console_mod, "copy_text_to_clipboard", fake_copy)
|
||||
|
||||
logger.info("debug-console-enter-copy-marker")
|
||||
app = _Harness()
|
||||
async with app.run_test() as pilot:
|
||||
screen = DebugConsoleScreen(_snapshot())
|
||||
app.push_screen(screen)
|
||||
await pilot.pause()
|
||||
log = screen.query_one("#debug-log", _DebugLogView)
|
||||
log.focus()
|
||||
index = next(
|
||||
index
|
||||
for index, record in enumerate(log.records)
|
||||
if "debug-console-enter-copy-marker" in record.message
|
||||
)
|
||||
log._select_record(index)
|
||||
|
||||
await pilot.press("enter")
|
||||
await pilot.pause()
|
||||
|
||||
assert "debug-console-enter-copy-marker" in captured["text"]
|
||||
|
||||
async def test_tab_cycles_between_level_filter_and_log_lines(self) -> None:
|
||||
app = _Harness()
|
||||
async with app.run_test() as pilot:
|
||||
screen = DebugConsoleScreen(_snapshot())
|
||||
app.push_screen(screen)
|
||||
await pilot.pause()
|
||||
log = screen.query_one("#debug-log", _DebugLogView)
|
||||
select = screen.query_one("#debug-level-filter", Select)
|
||||
log.focus()
|
||||
|
||||
await pilot.press("tab")
|
||||
await pilot.pause()
|
||||
assert screen.focused is select
|
||||
|
||||
await pilot.press("shift+tab")
|
||||
await pilot.pause()
|
||||
assert screen.focused is log
|
||||
|
||||
async def test_tab_moves_open_level_dropdown_highlight(self) -> None:
|
||||
app = _Harness()
|
||||
async with app.run_test() as pilot:
|
||||
screen = DebugConsoleScreen(_snapshot())
|
||||
app.push_screen(screen)
|
||||
await pilot.pause()
|
||||
select = screen.query_one("#debug-level-filter", Select)
|
||||
select.action_show_overlay()
|
||||
await pilot.pause()
|
||||
overlay = select.query_one(SelectOverlay)
|
||||
assert overlay.highlighted is not None
|
||||
before = overlay.highlighted
|
||||
|
||||
await pilot.press("tab")
|
||||
await pilot.pause()
|
||||
assert overlay.highlighted == before + 1
|
||||
|
||||
await pilot.press("shift+tab")
|
||||
await pilot.pause()
|
||||
assert overlay.highlighted == before
|
||||
assert select.expanded
|
||||
|
||||
async def test_escape_collapses_level_dropdown_before_dismissing(self) -> None:
|
||||
app = _Harness()
|
||||
async with app.run_test() as pilot:
|
||||
screen = DebugConsoleScreen(_snapshot())
|
||||
app.push_screen(screen)
|
||||
await pilot.pause()
|
||||
select = screen.query_one("#debug-level-filter", Select)
|
||||
select.action_show_overlay()
|
||||
await pilot.pause()
|
||||
assert select.expanded
|
||||
|
||||
await pilot.press("escape")
|
||||
await pilot.pause()
|
||||
assert not select.expanded
|
||||
assert isinstance(app.screen, DebugConsoleScreen)
|
||||
|
||||
await pilot.press("escape")
|
||||
await pilot.pause()
|
||||
assert not isinstance(app.screen, DebugConsoleScreen)
|
||||
|
||||
async def test_click_copy_invokes_clipboard_with_logical_record(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
captured: dict[str, str] = {}
|
||||
|
||||
def fake_copy(_app: App, text: str) -> tuple[bool, str | None]:
|
||||
captured["text"] = text
|
||||
return True, None
|
||||
|
||||
monkeypatch.setattr(debug_console_mod, "copy_text_to_clipboard", fake_copy)
|
||||
|
||||
logger.info("debug-console-click-copy-marker")
|
||||
app = _Harness()
|
||||
async with app.run_test() as pilot:
|
||||
screen = DebugConsoleScreen(_snapshot())
|
||||
app.push_screen(screen)
|
||||
await pilot.pause()
|
||||
log = screen.query_one("#debug-log", _DebugLogView)
|
||||
record = log._record_at_visual_y(log.line_count - 1)
|
||||
assert record is not None
|
||||
assert "debug-console-click-copy-marker" in record.message
|
||||
screen._copy_record(record)
|
||||
await pilot.pause()
|
||||
|
||||
assert "debug-console-click-copy-marker" in captured["text"]
|
||||
|
||||
async def test_copy_failure_notifies_warning(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
def fake_copy(_app: App, _text: str) -> tuple[bool, str | None]:
|
||||
return False, "clipboard unavailable"
|
||||
|
||||
monkeypatch.setattr(debug_console_mod, "copy_text_to_clipboard", fake_copy)
|
||||
|
||||
logger.info("debug-console-copy-fail-marker")
|
||||
app = _Harness()
|
||||
async with app.run_test() as pilot:
|
||||
screen = DebugConsoleScreen(_snapshot())
|
||||
app.push_screen(screen)
|
||||
await pilot.pause()
|
||||
await pilot.press("c")
|
||||
await pilot.pause()
|
||||
|
||||
latest = list(app._notifications)[-1]
|
||||
assert latest.severity == "warning"
|
||||
assert "clipboard unavailable" in latest.message
|
||||
|
||||
async def test_escape_dismisses(self) -> None:
|
||||
app = _Harness()
|
||||
async with app.run_test() as pilot:
|
||||
app.push_screen(DebugConsoleScreen(_snapshot()))
|
||||
await pilot.pause()
|
||||
assert isinstance(app.screen, DebugConsoleScreen)
|
||||
await pilot.press("escape")
|
||||
await pilot.pause()
|
||||
assert not isinstance(app.screen, DebugConsoleScreen)
|
||||
|
||||
|
||||
class TestRecordMatchesFilter:
|
||||
def test_all_matches_everything(self) -> None:
|
||||
record = _log_record("x", level="DEBUG", levelno=logging.DEBUG)
|
||||
assert _record_matches_filter(record, "all")
|
||||
|
||||
def test_min_and_only_modes(self) -> None:
|
||||
info = _log_record("i", level="INFO", levelno=logging.INFO)
|
||||
warning = _log_record("w", level="WARNING", levelno=logging.WARNING)
|
||||
assert not _record_matches_filter(info, "min:WARNING")
|
||||
assert _record_matches_filter(warning, "min:WARNING")
|
||||
assert _record_matches_filter(warning, "only:WARNING")
|
||||
assert not _record_matches_filter(info, "only:WARNING")
|
||||
|
||||
def test_custom_numeric_level_sorts_by_levelno(self) -> None:
|
||||
# A custom level between INFO and WARNING. The old name-based lookup
|
||||
# mapped unknown names to 0 and mis-sorted them below DEBUG.
|
||||
notice = _log_record("n", level="NOTICE", levelno=25)
|
||||
assert _record_matches_filter(notice, "min:INFO")
|
||||
assert not _record_matches_filter(notice, "min:WARNING")
|
||||
|
||||
def test_unknown_threshold_level_shows_record(self) -> None:
|
||||
# Unreachable via FilterValue, but a diagnostic must not hide records
|
||||
# (or raise KeyError on the poll timer) if a bad filter ever slips
|
||||
# through: an unrecognized threshold level matches everything.
|
||||
record = _log_record("x", level="INFO", levelno=logging.INFO)
|
||||
assert _record_matches_filter(record, cast("FilterValue", "min:BOGUS"))
|
||||
|
||||
|
||||
class TestDebugConsoleToggle:
|
||||
async def test_ctrl_backslash_opens_and_closes(self) -> None:
|
||||
app = DeepAgentsApp(agent=MagicMock(), thread_id="thread-123")
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
await pilot.press("ctrl+backslash")
|
||||
await pilot.pause()
|
||||
assert isinstance(app.screen, DebugConsoleScreen)
|
||||
await pilot.press("ctrl+backslash")
|
||||
await pilot.pause()
|
||||
assert not isinstance(app.screen, DebugConsoleScreen)
|
||||
|
||||
async def test_toggle_action_closes_open_console(self) -> None:
|
||||
app = DeepAgentsApp(agent=MagicMock(), thread_id="thread-123")
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
app.action_toggle_debug_console()
|
||||
await pilot.pause()
|
||||
assert isinstance(app.screen, DebugConsoleScreen)
|
||||
|
||||
# Call the app-level toggle directly to exercise its pop branch: a
|
||||
# `ctrl+backslash` keypress would be intercepted by the modal's own
|
||||
# priority binding and closed via the screen action instead.
|
||||
app.action_toggle_debug_console()
|
||||
await pilot.pause()
|
||||
assert not isinstance(app.screen, DebugConsoleScreen)
|
||||
|
||||
async def test_debug_command_opens_console(self) -> None:
|
||||
app = DeepAgentsApp(agent=MagicMock(), thread_id="thread-123")
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
await app._handle_command("/debug")
|
||||
await pilot.pause()
|
||||
assert isinstance(app.screen, DebugConsoleScreen)
|
||||
|
||||
async def test_opens_over_existing_modal(self) -> None:
|
||||
class _OtherModal(ModalScreen[None]):
|
||||
def compose(self) -> ComposeResult:
|
||||
yield Static("other")
|
||||
|
||||
app = DeepAgentsApp(agent=MagicMock(), thread_id="thread-123")
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
app.push_screen(_OtherModal())
|
||||
await pilot.pause()
|
||||
modal = app.screen
|
||||
|
||||
await pilot.press("ctrl+backslash")
|
||||
await pilot.pause()
|
||||
|
||||
assert isinstance(app.screen, DebugConsoleScreen)
|
||||
await pilot.press("escape")
|
||||
await pilot.pause()
|
||||
assert app.screen is modal
|
||||
|
||||
async def test_build_snapshot_contains_core_fields(self) -> None:
|
||||
app = DeepAgentsApp(agent=MagicMock(), thread_id="thread-xyz", cwd="/tmp/work")
|
||||
async with app.run_test():
|
||||
snapshot = dict(app._build_debug_snapshot())
|
||||
assert snapshot["Thread"] == "thread-xyz"
|
||||
assert snapshot["CWD"] == "/tmp/work"
|
||||
assert "Version" in snapshot
|
||||
assert "Auto-approve" in snapshot
|
||||
assert snapshot["MCP servers"] == "none"
|
||||
|
||||
async def test_build_snapshot_formats_mcp_servers(self) -> None:
|
||||
app = DeepAgentsApp(agent=MagicMock(), thread_id="t")
|
||||
async with app.run_test():
|
||||
servers = [
|
||||
SimpleNamespace(name="fs", status="connected"),
|
||||
SimpleNamespace(name="web", status="error"),
|
||||
]
|
||||
app._mcp_server_info = servers # ty: ignore[invalid-assignment] # stub servers expose .name/.status
|
||||
snapshot = dict(app._build_debug_snapshot())
|
||||
assert snapshot["MCP servers"] == "fs (connected), web (error)"
|
||||
|
||||
async def test_build_snapshot_degrades_on_field_error(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
app = DeepAgentsApp(agent=MagicMock(), thread_id="t")
|
||||
async with app.run_test():
|
||||
|
||||
def boom() -> str:
|
||||
msg = "kaboom"
|
||||
raise RuntimeError(msg)
|
||||
|
||||
monkeypatch.setattr(app, "_effective_model_spec", boom)
|
||||
snapshot = dict(app._build_debug_snapshot())
|
||||
# The failing field degrades; the rest of the snapshot still builds.
|
||||
assert snapshot["Model"].startswith("(unavailable")
|
||||
assert "Version" in snapshot
|
||||
assert snapshot["Thread"] == "t"
|
||||
@@ -90,7 +90,7 @@ class TestFileOpsExceptionHandling:
|
||||
|
||||
tracker = FileOpTracker(assistant_id=None, backend=mock_backend)
|
||||
|
||||
with caplog.at_level(logging.DEBUG):
|
||||
with caplog.at_level(logging.DEBUG, logger="deepagents_code"):
|
||||
tracker.start_operation(
|
||||
"write_file",
|
||||
{"file_path": "/test.txt", "content": "test"},
|
||||
@@ -114,7 +114,7 @@ class TestFileOpsExceptionHandling:
|
||||
|
||||
tracker = FileOpTracker(assistant_id=None, backend=mock_backend)
|
||||
|
||||
with caplog.at_level(logging.DEBUG):
|
||||
with caplog.at_level(logging.DEBUG, logger="deepagents_code"):
|
||||
tracker.start_operation(
|
||||
"edit_file",
|
||||
{"file_path": "/test.txt", "old_string": "a", "new_string": "b"},
|
||||
@@ -141,7 +141,7 @@ class TestFileOpsExceptionHandling:
|
||||
|
||||
tracker = FileOpTracker(assistant_id=None, backend=mock_backend)
|
||||
|
||||
with caplog.at_level(logging.DEBUG):
|
||||
with caplog.at_level(logging.DEBUG, logger="deepagents_code"):
|
||||
tracker.start_operation(
|
||||
"write_file",
|
||||
{"file_path": "/test.bin", "content": "test"},
|
||||
@@ -161,7 +161,7 @@ class TestFileOpsExceptionHandling:
|
||||
# Test with non-existent file
|
||||
nonexistent = tmp_path / "does_not_exist.txt"
|
||||
|
||||
with caplog.at_level(logging.DEBUG):
|
||||
with caplog.at_level(logging.DEBUG, logger="deepagents_code"):
|
||||
result = _safe_read(nonexistent)
|
||||
|
||||
assert result is None
|
||||
@@ -208,7 +208,7 @@ class TestMediaUtilsExceptionHandling:
|
||||
mock_run.side_effect = subprocess.TimeoutExpired(cmd="pngpaste", timeout=2)
|
||||
mock_osascript.return_value = None
|
||||
|
||||
with caplog.at_level(logging.DEBUG):
|
||||
with caplog.at_level(logging.DEBUG, logger="deepagents_code"):
|
||||
result = _get_macos_clipboard_image()
|
||||
|
||||
assert result is None
|
||||
@@ -227,7 +227,7 @@ class TestMediaUtilsExceptionHandling:
|
||||
mock_run.side_effect = FileNotFoundError("pngpaste")
|
||||
mock_osascript.return_value = None
|
||||
|
||||
with caplog.at_level(logging.DEBUG):
|
||||
with caplog.at_level(logging.DEBUG, logger="deepagents_code"):
|
||||
result = _get_macos_clipboard_image()
|
||||
|
||||
assert result is None
|
||||
@@ -245,7 +245,7 @@ class TestMediaUtilsExceptionHandling:
|
||||
mock_mkstemp.return_value = (5, "/tmp/test.png")
|
||||
mock_run.side_effect = subprocess.TimeoutExpired(cmd="osascript", timeout=2)
|
||||
|
||||
with caplog.at_level(logging.DEBUG):
|
||||
with caplog.at_level(logging.DEBUG, logger="deepagents_code"):
|
||||
result = _get_clipboard_via_osascript()
|
||||
|
||||
assert result is None
|
||||
|
||||
@@ -3010,7 +3010,7 @@ class TestFetchOllamaInstalledModelProfiles:
|
||||
"""Unexpected capability shape does not produce false flags."""
|
||||
payload = {"model_info": {}, "capabilities": "tools"}
|
||||
|
||||
with caplog.at_level(logging.DEBUG):
|
||||
with caplog.at_level(logging.DEBUG, logger="deepagents_code"):
|
||||
profile = model_config._profile_from_ollama_show_payload(payload)
|
||||
|
||||
assert profile == {}
|
||||
|
||||
Reference in New Issue
Block a user