项目文件夹

文件
wehub-resource-sync 4b6817381b
CI (OpenClaw E2E) / openclaw test (push) Has been cancelled
CI / coverage-report (push) Has been cancelled
CI / test-kubernetes (push) Has been cancelled
CI / should-run-thorough (push) Has been cancelled
CI / test-thorough (cloudwatch-demo) (push) Has been cancelled
CI / test-thorough (flink-ecs) (push) Has been cancelled
CI / test-thorough (upstream-lambda) (push) Has been cancelled
CI / test-thorough (prefect-ecs-fargate) (push) Has been cancelled
Release / build-binaries (zip, opensre.exe, onefile, windows-latest, windows-x64) (push) Has been cancelled
Benchmark image — build + push to ECR (any adapter) / build + push (push) Has been cancelled
CI / quality (ubuntu-latest) (push) Has been cancelled
CI / test (tools-runtime) (push) Has been cancelled
CI / test (e2e-general) (push) Has been cancelled
CI / test (cli-runtime) (push) Has been cancelled
CI / test (e2e-provider-and-openclaw) (push) Has been cancelled
CI / test (integrations-and-misc) (push) Has been cancelled
Release / verify (push) Has been cancelled
Release / build-python-dist (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, macos-15-intel, darwin-x64) (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, macos-latest, darwin-arm64) (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, ubuntu-22.04, linux-x64) (push) Has been cancelled
Release / publish-release (push) Has been cancelled
Release / publish-main-release (push) Has been cancelled
Interactive Shell Live (PR + post-merge) / turn-checks (no-LLM) (push) Has been cancelled
CodeQL / Analyze (python) (push) Has been cancelled
Interactive Shell Live (PR + post-merge) / turn-live shard ${{ matrix.shard_index }} (push) Has been cancelled
Release / prepare (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, ubuntu-22.04-arm, linux-arm64) (push) Has been cancelled
Synthetic Deterministic Tests / Synthetic offline (deterministic) (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 13:10:45 +08:00

1781 行
69 KiB
Python

此文件含有模棱两可的 Unicode 字符
此文件含有可能会与其他字符混淆的 Unicode 字符。 如果您是想特意这样的,可以安全地忽略该警告。 使用 Escape 按钮显示他们。
"""Tests for the interactive shell terminal runtime helpers."""
from __future__ import annotations
import asyncio
import io
import re
import subprocess
import sys
import threading
import time
from pathlib import Path
import pytest
from prompt_toolkit.application import create_app_session
from prompt_toolkit.buffer import Buffer
from prompt_toolkit.completion import CompleteEvent
from prompt_toolkit.document import Document
from prompt_toolkit.history import FileHistory, InMemoryHistory
from prompt_toolkit.input import DummyInput
from prompt_toolkit.input.defaults import create_pipe_input
from prompt_toolkit.keys import Keys
from prompt_toolkit.output import DummyOutput
from platform.terminal import theme as ui_theme
from platform.terminal.theme import (
ANSI_RESET,
get_active_theme_name,
set_active_theme,
)
from surfaces.interactive_shell.command_registry import SLASH_COMMANDS, dispatch_slash
from surfaces.interactive_shell.runtime.core import confirmation as controller_runtime
from surfaces.interactive_shell.runtime.core import state as loop_state
from surfaces.interactive_shell.runtime.core import turn_detection as loop_turn_detection
from surfaces.interactive_shell.runtime.startup import initial_input as startup_initial_input
from surfaces.interactive_shell.session import Session
from surfaces.interactive_shell.ui import input_prompt
from surfaces.interactive_shell.ui.components.cpr_stdin import (
strip_cpr_escape_sequences,
strip_cpr_sequences,
)
from surfaces.interactive_shell.ui.input_prompt import completion as prompt_completion
from surfaces.interactive_shell.ui.input_prompt.completion import ShellCompleter
from surfaces.interactive_shell.ui.input_prompt.key_bindings import (
_SHIFT_ENTER_SEQUENCE,
_build_prompt_key_bindings,
_tab_expand_or_menu,
build_cancel_key_bindings,
)
from surfaces.interactive_shell.ui.input_prompt.lexer import ReplInputLexer
from surfaces.interactive_shell.ui.input_prompt.rendering import _prompt_message
from surfaces.interactive_shell.ui.input_prompt.style import _build_prompt_style
from surfaces.interactive_shell.ui.streaming import _CHARS_PER_TOKEN
from surfaces.interactive_shell.ui.streaming.console import StreamingConsole
def test_agent_presentation_import_does_not_load_shell_turn_execution() -> None:
result = subprocess.run(
[
sys.executable,
"-c",
(
"import sys; "
"import surfaces.interactive_shell.runtime.agent_presentation; "
"print('surfaces.interactive_shell.runtime.shell_turn_execution' in sys.modules)"
),
],
check=True,
capture_output=True,
text=True,
)
assert result.stdout.strip() == "False"
def test_streaming_console_status_does_not_recurse(monkeypatch) -> None:
"""Regression: overriding Console.print broke Rich's status spinner."""
spinner = loop_state.SpinnerState()
console = StreamingConsole(
spinner,
threading.Event(),
file=io.StringIO(),
force_terminal=False,
width=80,
)
with console.status("working", spinner="dots"):
pass
@pytest.mark.parametrize(
("text", "expected"),
[
("\x1b[32;1R", ""),
("[32;1R", ""),
("\x9b32;1R", ""),
("what is our current model?[32;1R", "what is our current model?"),
("before \x1b[12;80R after", "before after"),
("7R[25;57R23;57R", ""),
("25;57R", ""),
# Regression (issue #2942): CPR-shaped substrings embedded in ordinary
# input must NOT be stripped. A bare ``rowR``/``row;colR`` followed by a
# lone digit, or trailed by a word, is real user text, not a leak — only a
# full ``row;colR`` continuation counts as a CPR fragment.
("retry 5R3 times", "retry 5R3 times"),
("ranges 12;34R okay", "ranges 12;34R okay"),
("use r5.xlarge", "use r5.xlarge"),
("scale to 16RAM nodes", "scale to 16RAM nodes"),
("scale 12;34R5 nodes", "scale 12;34R5 nodes"),
# Consecutive bare fragments (no introducer) are a real leaked burst and
# must still strip fully.
("25;57R12;80R", ""),
],
)
def test_strip_cpr_sequences_removes_terminal_cursor_replies(
text: str,
expected: str,
) -> None:
assert strip_cpr_sequences(text) == expected
@pytest.mark.parametrize(
("text", "expected"),
[
("before \x1b[12;80R after", "before after"),
("what is 12;80R?", "what is 12;80R?"),
("[12;80R", "[12;80R"),
],
)
def test_strip_cpr_escape_sequences_only_removes_escaped_variants(
text: str,
expected: str,
) -> None:
assert strip_cpr_escape_sequences(text) == expected
def test_repl_input_lexer_highlights_first_slash_token() -> None:
lexer = ReplInputLexer()
get_line = lexer.lex_document(Document("/model show", len("/model")))
fragments = get_line(0)
cmd_frags = [(frag[0], frag[1]) for frag in fragments if frag[0] == "class:repl-slash-command"]
assert cmd_frags == [("class:repl-slash-command", "/model")]
rest = "".join(frag[1] for frag in fragments if frag[0] == "")
assert " show" in rest or rest.endswith(" show")
def test_build_prompt_session_uses_persistent_history(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
import config.constants as const_module
monkeypatch.setattr(const_module, "OPENSRE_HOME_DIR", tmp_path)
with create_app_session(input=DummyInput(), output=DummyOutput()):
prompt = input_prompt.build_prompt_session()
assert isinstance(prompt.history, FileHistory)
assert prompt.history.filename == str(tmp_path / "interactive_history")
assert tmp_path.exists()
assert isinstance(prompt.completer, ShellCompleter)
assert prompt.multiline is True
assert prompt.reserve_space_for_menu == 8
assert prompt.app.key_bindings is not None
def test_build_prompt_session_falls_back_to_memory_history(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
import config.constants as const_module
blocked_home = tmp_path / "not-a-directory"
blocked_home.write_text("", encoding="utf-8")
monkeypatch.setattr(const_module, "OPENSRE_HOME_DIR", blocked_home)
with create_app_session(input=DummyInput(), output=DummyOutput()):
prompt = input_prompt.build_prompt_session()
assert isinstance(prompt.history, InMemoryHistory)
def test_repl_session_prompt_history_backend_matches_prompt_toolkit_history(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
import config.constants as const_module
monkeypatch.setattr(const_module, "OPENSRE_HOME_DIR", tmp_path)
with create_app_session(input=DummyInput(), output=DummyOutput()):
session = Session()
prompt = input_prompt.build_prompt_session()
session.terminal.prompt_history_backend = prompt.history
assert session.terminal.prompt_history_backend is prompt.history
def test_prompt_message_uses_accent_glyph() -> None:
set_active_theme("blue")
rendered = _prompt_message(Session()).value
assert ui_theme.PROMPT_ACCENT_ANSI in rendered
assert "" in rendered
assert ANSI_RESET in rendered
def test_shift_enter_inserts_newline_before_submit(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
import config.constants as const_module
monkeypatch.setattr(const_module, "OPENSRE_HOME_DIR", tmp_path)
async def _collect() -> str:
with (
create_pipe_input() as pipe_input,
create_app_session(input=pipe_input, output=DummyOutput()),
):
prompt = input_prompt.build_prompt_session()
task = asyncio.create_task(prompt.prompt_async(""))
pipe_input.send_bytes(b"first line")
pipe_input.send_bytes(_SHIFT_ENTER_SEQUENCE.encode())
pipe_input.send_bytes(b"second line\r")
return await asyncio.wait_for(task, timeout=1)
assert asyncio.run(_collect()) == "first line\nsecond line"
def test_shell_completer_previews_all_commands() -> None:
completions = list(
ShellCompleter().get_completions(
Document("/"),
CompleteEvent(text_inserted=True),
)
)
names = [completion.text for completion in completions]
assert "/help" in names
assert "/effort" in names
assert "/integrations" in names
assert "/tools" in names
assert "/model" in names
assert all(name.startswith("/") for name in names)
def test_shell_completer_filters_by_prefix() -> None:
completions = list(
ShellCompleter().get_completions(
Document("/to"),
CompleteEvent(text_inserted=True),
)
)
assert [completion.text for completion in completions] == ["/tools"]
def test_shell_completer_suggests_subcommands_for_tools() -> None:
completions = list(
ShellCompleter().get_completions(
Document("/tools "),
CompleteEvent(text_inserted=True),
)
)
names = sorted({c.text for c in completions})
assert names == ["list", "ls", "tool", "tools"]
def test_shell_completer_hides_inline_picker_autocomplete_in_tty(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(prompt_completion, "repl_tty_interactive", lambda: True)
completions = list(
ShellCompleter().get_completions(
Document("/model "),
CompleteEvent(text_inserted=True),
)
)
assert completions == []
def test_shell_completer_keeps_inline_picker_autocomplete_when_arg_started(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(prompt_completion, "repl_tty_interactive", lambda: True)
completions = list(
ShellCompleter().get_completions(
Document("/model s"),
CompleteEvent(text_inserted=True),
)
)
assert sorted({c.text for c in completions}) == ["set", "show"]
def test_shell_completer_suggests_effort_levels() -> None:
completions = list(
ShellCompleter().get_completions(
Document("/effort "),
CompleteEvent(text_inserted=True),
)
)
names = sorted({c.text for c in completions})
assert names == ["high", "low", "max", "medium", "xhigh"]
def test_tab_applies_unique_slash_command_completion() -> None:
buff = Buffer(completer=ShellCompleter())
buff.insert_text("/mod")
_tab_expand_or_menu(buff)
assert buff.text == "/model"
def test_tab_with_open_completion_menu_applies_current_item() -> None:
from prompt_toolkit.buffer import CompletionState
from prompt_toolkit.completion import Completion
buff = Buffer()
buff.insert_text("/mo")
orig_doc = buff.document
c_model = Completion("/model", start_position=-3)
c_mcp = Completion("/mcp", start_position=-3)
# Assign directly — updating ``buff.document`` afterward clears ``complete_state``.
buff.complete_state = CompletionState(orig_doc, [c_model, c_mcp], 0)
_tab_expand_or_menu(buff)
assert buff.complete_state is None
assert buff.text == "/model"
def test_tab_with_menu_and_no_index_applies_first_choice() -> None:
from prompt_toolkit.buffer import CompletionState
from prompt_toolkit.completion import Completion
buff = Buffer()
buff.insert_text("/mo")
orig_doc = buff.document
c_model = Completion("/model", start_position=-3)
c_mcp = Completion("/mcp", start_position=-3)
buff.complete_state = CompletionState(orig_doc, [c_model, c_mcp], None)
_tab_expand_or_menu(buff)
assert buff.complete_state is None
assert buff.text == "/model"
def test_completion_includes_tab_navigation() -> None:
key_bindings = _build_prompt_key_bindings()
keys = {binding.keys for binding in key_bindings.bindings}
assert (Keys.ControlM,) in keys
assert (Keys.Down,) in keys
assert (Keys.Up,) in keys
assert (Keys.Tab,) in keys
assert (Keys.BackTab,) in keys
def test_build_prompt_style_tracks_active_theme() -> None:
set_active_theme("amber")
amber_attrs = _build_prompt_style().get_attrs_for_style_str("class:prompt-frame-line")
set_active_theme("teal")
teal_attrs = _build_prompt_style().get_attrs_for_style_str("class:prompt-frame-line")
assert amber_attrs.color and amber_attrs.color.lower() == "f2d48a"
assert teal_attrs.color and teal_attrs.color.lower() == "8ae2d6"
assert amber_attrs.color != teal_attrs.color
def test_completion_menu_current_item_uses_highlight_style() -> None:
from platform.terminal.theme import BG, HIGHLIGHT
set_active_theme("green")
style = _build_prompt_style()
attrs = style.get_attrs_for_style_str("class:repl-slash-command")
assert attrs.color == HIGHLIGHT.lstrip("#")
assert attrs.bgcolor == BG.lstrip("#")
assert attrs.bold is True
attrs_menu = style.get_attrs_for_style_str("class:completion-menu.completion.current")
assert attrs_menu.color == HIGHLIGHT.lstrip("#")
assert attrs_menu.bgcolor == BG.lstrip("#")
assert attrs_menu.reverse is False
assert attrs_menu.bold is True
def test_lazy_rich_style_split_tracks_active_theme() -> None:
from platform.terminal import theme as ui_theme
set_active_theme("green")
assert bool(ui_theme.DIM) is True
assert ui_theme.DIM.split() == [ui_theme.get_active_theme().DIM]
assert ui_theme.BOLD_BRAND.split() == ["bold", ui_theme.get_active_theme().BRAND]
set_active_theme("blue")
assert bool(ui_theme.DIM) is True
assert ui_theme.DIM.split() == [ui_theme.get_active_theme().DIM]
assert ui_theme.BOLD_BRAND.split() == ["bold", ui_theme.get_active_theme().BRAND]
def test_lazy_rich_style_parses_as_real_rich_style() -> None:
from rich.style import Style
from platform.terminal import theme as ui_theme
# Rich resolves a _LazyRichStyle via its underlying ``str`` value, so the
# contract at call sites is to pass ``str(...)``. Verify that contract
# produces a non-null Style across themes.
set_active_theme("amber")
assert Style.parse(str(ui_theme.DIM)) != Style.null()
assert Style.parse(str(ui_theme.BOLD_BRAND)) != Style.null()
set_active_theme("blue")
assert Style.parse(str(ui_theme.DIM)) != Style.null()
assert Style.parse(str(ui_theme.BOLD_BRAND)) != Style.null()
def test_shell_completer_path_completion_honors_mixed_case_prefix(tmp_path: Path) -> None:
"""Regression: path fragments must not be lowercased before PathCompleter.
On case-sensitive filesystems, a lowered prefix can stop matching real directory
names (e.g. ``RePoRtS`` no longer matches prefix ``re``).
"""
mixed_dir = tmp_path / "RePoRtS"
mixed_dir.mkdir()
(mixed_dir / "x.txt").write_text("x", encoding="utf-8")
partial = str(tmp_path / "Re")
line = f"/investigate {partial}"
completions = list(
ShellCompleter().get_completions(
Document(line, len(line)),
CompleteEvent(text_inserted=True),
)
)
assert completions
joined = " ".join(str(c.display) for c in completions)
assert "RePoRtS" in joined
def test_shell_completer_investigate_includes_template_hints() -> None:
completions = list(
ShellCompleter().get_completions(
Document("/investigate ", len("/investigate ")),
CompleteEvent(text_inserted=True),
)
)
assert any(c.text == "generic" for c in completions)
assert any(c.text == "splunk" for c in completions)
def test_run_text_investigation_uses_background_launcher_when_mode_enabled(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from rich.console import Console
from tools.interactive_shell.actions.investigation import (
run_text_investigation,
)
launches: list[tuple[str, str]] = []
def _fake_start_background_text_investigation(
*,
alert_text: str,
session: Session,
console: Console,
display_command: str,
) -> str:
_ = (session, console)
launches.append((alert_text, display_command))
return "bg123"
monkeypatch.setattr(
"surfaces.interactive_shell.runtime.background.runner.start_background_text_investigation",
_fake_start_background_text_investigation,
)
session = Session()
session.terminal.background_mode_enabled = True
console = Console(file=io.StringIO(), force_terminal=False, highlight=False)
run_text_investigation("High CPU alert", session, console)
assert launches == [("High CPU alert", "background free-text investigation")]
assert session.task_registry.list_recent(10) == []
def test_run_initial_input_dispatches_as_non_tty(monkeypatch: pytest.MonkeyPatch) -> None:
calls: list[dict[str, object]] = []
def _fake_handle_message(*args: object, **kwargs: object) -> None:
_ = args
calls.append(kwargs)
monkeypatch.setattr(
"surfaces.interactive_shell.runtime.shell_turn_execution.execute_shell_turn",
_fake_handle_message,
)
assert startup_initial_input.run_initial_input("/remote", Session()) == 0
assert len(calls) == 1
assert calls[0]["is_tty"] is False
class TestLooksLikeCorrection:
"""Unit tests for the ``_looks_like_correction`` heuristic.
Pins the v1 catches, false-positive guards, and limitations so future
iteration on ``_INTERVENTION_CORRECTION_RE`` has a regression baseline.
"""
@pytest.mark.parametrize(
"text",
[
"no, do that instead",
"nope",
"nvm",
"never mind",
"actually, let's check Datadog first",
"scratch that, run the synthetic test",
"wait, wrong dashboard",
"let's do an EKS health check instead",
"try a token refresh instead",
"wrong dashboard, fix it",
"instead, log a warning",
"Wait!", # case-insensitive
"NO.",
],
)
def test_correction_cues_match(self, text: str) -> None:
assert loop_turn_detection.looks_like_correction(text) is True
@pytest.mark.parametrize(
"text",
[
# Punctuation-lookahead guards reject content uses of cue words.
"stop the server before redeploying",
"instead of returning null, log a warning",
"no problem, I'll handle it",
"wait for the result",
# v1 limitation: cue must be at start of message.
"hmm, scratch that",
"doesn't work, try X instead",
# Edge cases.
"",
" ",
"```\nstop the server\n```",
],
)
def test_non_correction_text_does_not_match(self, text: str) -> None:
assert loop_turn_detection.looks_like_correction(text) is False
class TestLooksLikeConfirmationAnswer:
"""Unit tests for the y/n token recognizer.
Type-ahead text submitted while a ``Proceed? [Y/n]`` worker is
parked used to be silently delivered to the confirmation handler
and declined the pending action. The recognizer is the
gate that keeps that from happening — only deliberate y/n tokens
(and empty Enter, which the upstream ``[Y/n]`` prompt accepts as
"yes") are treated as answers.
"""
@pytest.mark.parametrize(
"text",
[
"y",
"Y",
"yes",
"YES",
"n",
"N",
"no",
"No",
" y ",
" yes\n",
"",
" ",
None,
],
)
def test_recognised_tokens_match(self, text: str | None) -> None:
assert loop_turn_detection.looks_like_confirmation_answer(text) is True
@pytest.mark.parametrize(
"text",
[
"what is opensre?",
"yeah do it", # extra words even after "yeah" — ambiguous, not delivered
"yep",
"yup",
"nope",
"/help",
"show me logs",
"y'all should run this",
],
)
def test_unrecognised_text_does_not_match(self, text: str) -> None:
assert loop_turn_detection.looks_like_confirmation_answer(text) is False
@pytest.mark.parametrize(
"text",
[
# Multi-line type-ahead whose first token reads as ``y``/``yes``
# must NOT be classified as a confirmation. The gate compares the
# whole stripped/lowered string against the token set, so any
# trailing words or internal newlines disqualify it.
"yes please run that against staging instead",
"yes\nbut do X first",
"y\nrun it now",
"no\nactually wait, let me check the logs first",
"Y but also rotate the keys after",
# Embedded newlines preserved through ``strip()`` so the join
# still fails the membership check.
"yes\nplease",
],
)
def test_multiline_type_ahead_starting_with_y_or_n_does_not_match(self, text: str) -> None:
"""Regression for the type-ahead-as-confirmation footgun: a pasted
or typed sentence beginning with ``y``/``yes`` (or ``n``/``no``)
must be treated as a new turn, not silently delivered as the
Proceed? answer. ``str.strip()`` only trims outer whitespace, so
any inner non-whitespace content keeps the lowered string out
of the token set. (Pure outer-whitespace cases like
``" yes\\n "`` correctly DO match — that's just ``yes`` with
stray whitespace, not a multi-line message.)
"""
assert loop_turn_detection.looks_like_confirmation_answer(text) is False
class TestLooksLikeCancelRequest:
"""Unit tests for the bare-cancel slash recognizer.
The recognizer is the gate that lets the prompt loop intercept
``/cancel``-style slashes typed while a dispatch is parked
(e.g. on a ``Proceed? [y/N]`` confirmation) and send them through
``state.cancel_current_dispatch()`` instead of queueing them
behind the dispatch they're trying to interrupt.
"""
@pytest.mark.parametrize(
"text",
[
"/cancel",
"/CANCEL",
"/Cancel",
"/stop",
"/STOP",
"/abort",
" /cancel ",
"/cancel\n",
"\t/stop\t",
],
)
def test_recognised_cancel_slashes_match(self, text: str) -> None:
assert loop_turn_detection.looks_like_cancel_request(text) is True
@pytest.mark.parametrize(
"text",
[
# Targeted background-task cancel — must keep flowing
# through the normal slash dispatch path so the existing
# ``_cmd_cancel`` handler resolves the task id.
"/cancel 8f5fe574",
"/cancel abc123",
"/stop now",
# Other slashes — unrelated to interrupt.
"/help",
"/tasks",
# Natural-language uses of the same words must NOT be
# intercepted; the user might be talking about cancelling
# a deploy or stopping a service in their environment.
"cancel this please",
"cancel",
"stop the deploy",
"stop",
"abort",
# Empty / whitespace / None — nothing to intercept.
"",
" ",
None,
],
)
def test_unrecognised_text_does_not_match(self, text: str | None) -> None:
assert loop_turn_detection.looks_like_cancel_request(text) is False
# ── Spinner state tests ──────────────────────────────────────────────────────
def _strip_ansi(text: str) -> str:
return re.sub(r"\x1b\[[0-9;?]*[A-Za-z]", "", text)
class TestSpinnerState:
"""``_SpinnerState`` holds the live-stream indicator state and renders
two ANSI views: the inline spinner above the input frame
(``inline_spinner_ansi``) and the bottom toolbar hint
(``toolbar_ansi``).
"""
def test_idle_state_emits_no_inline_spinner(self) -> None:
spinner = loop_state.SpinnerState()
assert spinner.streaming is False
assert spinner.inline_spinner_ansi() == ""
def test_inline_spinner_uses_active_theme_highlight(self) -> None:
from platform.terminal.theme import set_active_theme
set_active_theme("blue")
spinner = loop_state.SpinnerState()
spinner.start()
raw = spinner.inline_spinner_ansi()
assert "168;212;255" in raw
assert "185;237;175" not in raw
def test_streaming_inline_spinner_includes_glyph_and_token_count(self) -> None:
spinner = loop_state.SpinnerState()
spinner.start()
spinner.bytes_in = 1234 * _CHARS_PER_TOKEN # = 1234 tokens
rendered = _strip_ansi(spinner.inline_spinner_ansi())
# The verb is randomly picked from ``_THINKING_VERBS`` per turn —
# any of them followed by ``…`` is acceptable.
assert any(f"{verb}…" in rendered for verb in spinner._THINKING_VERBS)
# 1234 tokens → "1.2k" via format_token_count_short.
assert "1.2k tokens" in rendered
# Spinner glyph from the brail palette.
assert any(g in rendered for g in spinner._SPINNER_FRAMES)
def test_streaming_inline_spinner_hides_token_count_when_zero(self) -> None:
"""When no tokens have been received yet, the spinner should not
display ``↓ 0 tokens`` — the count is meaningless until streaming
starts. Only the elapsed time is shown in the parenthetical.
"""
spinner = loop_state.SpinnerState()
spinner.start()
# bytes_in defaults to 0 → token_count = 0
rendered = _strip_ansi(spinner.inline_spinner_ansi())
assert "tokens" not in rendered, (
"expected no token count in spinner before any tokens arrive"
)
assert "0" not in rendered or "0s" in rendered or "0." in rendered, (
"zero-token count leaked into spinner display"
)
def test_streaming_inline_spinner_shows_token_count_once_tokens_arrive(self) -> None:
"""Once bytes arrive, the token count reappears in the spinner."""
spinner = loop_state.SpinnerState()
spinner.start()
spinner.bytes_in = 100 * _CHARS_PER_TOKEN # = 100 tokens
rendered = _strip_ansi(spinner.inline_spinner_ansi())
assert "tokens" in rendered
def test_streaming_inline_spinner_verb_stays_constant_across_calls(self) -> None:
"""A turn's verb is fixed at ``start()`` so the indicator
doesn't flicker between words mid-stream."""
spinner = loop_state.SpinnerState()
spinner.start()
verbs_seen: set[str] = set()
for _ in range(20):
rendered = _strip_ansi(spinner.inline_spinner_ansi())
for verb in spinner._THINKING_VERBS:
if f"{verb}…" in rendered:
verbs_seen.add(verb)
break
assert len(verbs_seen) == 1, f"verb changed mid-turn — saw {verbs_seen}"
def test_inline_spinner_glyph_animates_with_elapsed_time(self) -> None:
"""The frame is a function of elapsed time, not of render-call count.
prompt_toolkit evaluates the prompt message several times per render
pass, so a per-call counter freezes the on-screen glyph (it advances a
whole number of cycles between visible renders). Repeated calls at one
instant must render one frame; advancing the clock must animate.
"""
spinner = loop_state.SpinnerState()
spinner.start()
same_instant = {
_extract_glyph(spinner.inline_spinner_ansi(), spinner._SPINNER_FRAMES)
for _ in range(len(spinner._SPINNER_FRAMES) * 2)
}
assert len(same_instant) == 1
seen = set()
for step in range(len(spinner._SPINNER_FRAMES)):
spinner.started_at = time.monotonic() - step * spinner._FRAME_INTERVAL_S * 1.001
seen.add(_extract_glyph(spinner.inline_spinner_ansi(), spinner._SPINNER_FRAMES))
assert seen == set(spinner._SPINNER_FRAMES)
def test_stop_returns_to_idle_state(self) -> None:
spinner = loop_state.SpinnerState()
spinner.start()
assert spinner.streaming is True
spinner.stop()
assert spinner.streaming is False
assert spinner.inline_spinner_ansi() == ""
def test_toolbar_always_returns_empty_string(self) -> None:
"""toolbar_ansi() must always return '' in both idle and streaming states.
A visible toolbar causes prompt_toolkit to emit \\033[6n (CPR)
cursor-position queries on every refresh_interval; those responses
corrupt the vt100 input parser with stray keystrokes. Hiding the
toolbar unconditionally also keeps its height at zero in *both* states,
which prevents the one-row height delta that causes prompt_toolkit to
misplace the cursor and leave stale spinner lines on screen. Idle hints
are surfaced through idle_hint_ansi() rendered in the prompt message's
reserved first line instead.
"""
spinner = loop_state.SpinnerState()
assert spinner.toolbar_ansi() == ""
spinner.start()
assert spinner.toolbar_ansi() == ""
def test_idle_hint_lists_shortcut_keys_when_buffer_empty(self) -> None:
"""When idle and the input buffer is empty (no prompt-toolkit app
running in this test → ``get_app_or_none()`` returns None →
treated as empty), the idle hint advertises the always-useful keys
but hides ``esc to clear`` since Esc is a no-op on empty buffer.
"""
spinner = loop_state.SpinnerState()
rendered = _strip_ansi(spinner.idle_hint_ansi())
assert "/ for commands" in rendered
assert "history" in rendered
# Hidden — buffer is empty, Esc would be a no-op, so the hint
# would mislead the user.
assert "esc to clear" not in rendered
def test_idle_hint_includes_esc_to_clear_when_buffer_has_text(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
"""When idle and the input buffer has text, the idle hint appends
``esc to clear`` so the user knows the shortcut exists.
``get_app_or_none`` is monkeypatched to return a fake app whose
``current_buffer.text`` is non-empty.
"""
class _FakeBuffer:
text = "partially typed"
class _FakeApp:
current_buffer = _FakeBuffer()
monkeypatch.setattr(loop_state, "get_app_or_none", lambda: _FakeApp())
spinner = loop_state.SpinnerState()
rendered = _strip_ansi(spinner.idle_hint_ansi())
assert "esc to clear" in rendered
assert "/ for commands" in rendered
def test_inline_spinner_contains_esc_to_cancel_when_streaming(self) -> None:
"""During streaming the inline spinner (shown in the prompt's first
reserved line) carries the ``esc to cancel`` hint so the user can
interrupt the dispatch.
"""
spinner = loop_state.SpinnerState()
spinner.start()
rendered = _strip_ansi(spinner.inline_spinner_ansi())
assert "esc to cancel" in rendered
# Idle hint text should NOT appear in the spinner row.
assert "/ for commands" not in rendered
def test_prompt_prefix_is_consistent_height(self) -> None:
"""The first line of the prompt message is always exactly one row —
it shows the idle hint when not streaming and the spinner when
streaming. Using ``idle_hint_ansi()`` or ``inline_spinner_ansi()``
as the prefix means height never changes between states, so
prompt_toolkit's cursor management stays accurate and stale spinner
lines do not accumulate on screen.
"""
spinner = loop_state.SpinnerState()
idle_prefix = _strip_ansi(spinner.idle_hint_ansi())
assert "\n" not in idle_prefix, f"idle prefix must be 1 row, got: {idle_prefix!r}"
spinner.start()
streaming_prefix = _strip_ansi(spinner.inline_spinner_ansi())
assert "\n" not in streaming_prefix, (
f"streaming prefix must be 1 row, got: {streaming_prefix!r}"
)
def _extract_glyph(ansi_text: str, frames: tuple[str, ...]) -> str:
plain = _strip_ansi(ansi_text)
for g in frames:
if g in plain:
return g
return ""
# ── Streaming-console adapter tests ──────────────────────────────────────────
class TestStreamingConsole:
"""``_StreamingConsole`` is the bridge between the streaming layer and
the prompt-toolkit spinner. It is the only way the dispatch worker
thread can signal back to the prompt: progress updates and
cancellation polling go through this object's optional methods.
"""
def test_update_progress_writes_to_spinner_state(self) -> None:
import threading as _threading
spinner = loop_state.SpinnerState()
spinner.start()
cancel = _threading.Event()
console = StreamingConsole(
spinner,
cancel,
highlight=False,
force_terminal=True,
color_system=None,
)
console.update_streaming_progress(4096)
assert spinner.bytes_in == 4096
def test_cancel_requested_reflects_event_state(self) -> None:
import threading as _threading
spinner = loop_state.SpinnerState()
cancel = _threading.Event()
console = StreamingConsole(
spinner,
cancel,
highlight=False,
force_terminal=True,
color_system=None,
)
assert console.cancel_requested is False
cancel.set()
assert console.cancel_requested is True
cancel.clear()
assert console.cancel_requested is False
def test_blank_print_resets_column_without_preparing_new_line(
self,
monkeypatch: pytest.MonkeyPatch,
) -> None:
import threading as _threading
spinner = loop_state.SpinnerState()
console = StreamingConsole(
spinner,
_threading.Event(),
file=io.StringIO(),
force_terminal=False,
width=80,
)
calls: list[str] = []
monkeypatch.setattr(
"surfaces.interactive_shell.ui.components.choice_menu.ensure_tty_column_zero",
lambda: calls.append("ensure"),
)
monkeypatch.setattr(
"surfaces.interactive_shell.ui.components.choice_menu.prepare_repl_output_line",
lambda: calls.append("prepare"),
)
console.print()
assert calls == ["ensure"]
# ── ReplState dataclass tests ────────────────────────────────────────────────
class TestReplState:
"""``_ReplState`` is the single owner of the cancellation primitives
shared between the prompt loop, the queue processor, and the
Esc/Ctrl+L key bindings. Its methods exist so callers don't poke
raw fields and re-derive ``is_running`` everywhere.
"""
def test_default_state_is_idle(self) -> None:
state = loop_state.ReplState()
assert state.is_dispatch_running() is False
assert state.exit_requested is False
# No active dispatch → no cancel event parked.
assert state.current_cancel_event is None
assert state.queue.empty()
assert state.phase is loop_state.TurnPhase.IDLE
def test_phase_transitions_dispatch_to_idle(self) -> None:
async def _scenario() -> None:
state = loop_state.ReplState()
async def _noop() -> None:
return None
task = asyncio.create_task(_noop())
cancel = threading.Event()
state.start_dispatch(task=task, cancel_event=cancel)
assert state.is_dispatch_running() is True
assert await task is None
state.finish_dispatch(cancel)
assert state.phase is loop_state.TurnPhase.IDLE
state.clear_current_task()
assert state.phase is loop_state.TurnPhase.IDLE
asyncio.run(_scenario())
def test_confirmation_phase_round_trips_to_dispatching(self) -> None:
async def _scenario() -> None:
state = loop_state.ReplState()
async def _waits() -> None:
await asyncio.sleep(1.0)
task = asyncio.create_task(_waits())
state.start_dispatch(task=task, cancel_event=threading.Event())
state.begin_confirmation(threading.Event(), "Proceed? ")
assert state.is_awaiting_confirmation() is True
assert state.phase is loop_state.TurnPhase.AWAITING_CONFIRMATION
# A normal confirmation completion returns to dispatching (task alive).
state.clear_confirmation()
assert state.is_awaiting_confirmation() is False
assert state.phase is loop_state.TurnPhase.DISPATCHING
task.cancel()
asyncio.run(_scenario())
def test_clear_confirmation_returns_to_idle_without_active_task(self) -> None:
state = loop_state.ReplState()
state.begin_confirmation(threading.Event(), "Proceed? ")
assert state.phase is loop_state.TurnPhase.AWAITING_CONFIRMATION
state.clear_confirmation()
assert state.phase is loop_state.TurnPhase.IDLE
def test_cancel_sets_cancelling_phase(self) -> None:
state = loop_state.ReplState()
state.current_cancel_event = threading.Event()
state.cancel_current_dispatch()
assert state.phase is loop_state.TurnPhase.CANCELLING
def test_cancel_during_confirmation_keeps_cancelling_phase(self) -> None:
state = loop_state.ReplState()
state.current_cancel_event = threading.Event()
state.begin_confirmation(threading.Event(), "Proceed? ")
state.cancel_current_dispatch()
assert state.phase is loop_state.TurnPhase.CANCELLING
# The confirmation cleanup must NOT downgrade an in-progress cancel.
state.clear_confirmation()
assert state.phase is loop_state.TurnPhase.CANCELLING
def test_idle_cancel_does_not_set_cancelling_phase(self) -> None:
state = loop_state.ReplState()
state.cancel_current_dispatch()
assert state.phase is loop_state.TurnPhase.IDLE
def test_is_dispatch_running_tracks_task_lifecycle(self) -> None:
async def _scenario() -> None:
state = loop_state.ReplState()
async def _slow() -> None:
await asyncio.sleep(0.05)
state.current_task = asyncio.create_task(_slow())
assert state.is_dispatch_running() is True
await state.current_task
assert state.is_dispatch_running() is False
asyncio.run(_scenario())
def test_cancel_current_dispatch_signals_event_and_task(self) -> None:
async def _scenario() -> None:
import threading as _threading
state = loop_state.ReplState()
dispatch_cancel = _threading.Event()
state.current_cancel_event = dispatch_cancel
async def _waits_forever() -> None:
# Long sleep — only the cancel can interrupt this.
await asyncio.sleep(1.0)
state.current_task = asyncio.create_task(_waits_forever())
# Snapshot to a local — re-reading ``state.current_task``
# across the cancel + await would let CodeQL / code-quality
# bots flag the bare ``await`` as ineffectual and would also
# leave the assertion racey if anything reassigned the field.
task = state.current_task
state.cancel_current_dispatch()
# Both signals must fire — per-dispatch event flipped AND
# the asyncio task cancelled.
assert dispatch_cancel.is_set() is True
try: # noqa: SIM105
await task
except asyncio.CancelledError:
# Expected — that's the whole point of the cancel.
pass
assert task.cancelled() is True
asyncio.run(_scenario())
def test_cancel_from_worker_thread_uses_call_soon_threadsafe(self) -> None:
"""``Task.cancel`` is not thread-safe; ``cancel_current_dispatch``
must send the cancel via ``loop.call_soon_threadsafe`` when
invoked from a worker thread (the ``/exit`` slash handler runs
in ``asyncio.to_thread`` and reaches us through
:func:`_request_exit`).
"""
async def _scenario() -> None:
import threading as _threading
state = loop_state.ReplState()
state.loop = asyncio.get_running_loop()
async def _waits_forever() -> None:
await asyncio.sleep(10.0)
state.current_task = asyncio.create_task(_waits_forever())
task = state.current_task
worker_done = _threading.Event()
def _cancel_in_worker() -> None:
state.cancel_current_dispatch()
worker_done.set()
worker = _threading.Thread(target=_cancel_in_worker)
worker.start()
worker.join(timeout=1.0)
assert worker_done.is_set(), "worker thread did not return"
try: # noqa: SIM105
await task
except asyncio.CancelledError:
# Expected: the worker-triggered cancellation surfaces
# here once the scheduled ``task.cancel`` callback runs
# on the loop. Swallow so the assertion below can verify
# cancellation state without the exception unwinding the
# test.
pass
assert task.cancelled() is True
asyncio.run(_scenario())
def test_cancel_when_no_task_is_a_no_op(self) -> None:
"""``cancel_current_dispatch`` is idempotent — safe to call when
nothing is running. With no active dispatch parked,
``current_cancel_event`` is ``None`` and there's nothing to flip."""
state = loop_state.ReplState()
state.cancel_current_dispatch()
assert state.is_dispatch_running() is False
assert state.current_cancel_event is None
def test_per_dispatch_cancel_events_are_isolated(self) -> None:
"""Regression guard for the shared-event race that used to let
a previous turn's worker-thread observation get clobbered by
the next turn's ``Event.clear()``.
The fix: each dispatch processor turn allocates a fresh
``threading.Event`` and parks it at ``state.current_cancel_event``.
The previous turn's worker keeps a strong reference to its OWN
event; a new turn replacing the parked one never resets the
old worker's signal.
"""
import threading as _threading
state = loop_state.ReplState()
# Turn 1: park its event, fire cancel — that event is now set.
old_event = _threading.Event()
state.current_cancel_event = old_event
state.cancel_current_dispatch()
assert old_event.is_set() is True
# Turn 2 starts: a fresh event is parked. The OLD event must
# still be set (its worker is still polling it from the prior
# turn); the new event must not be — turn 2 hasn't been
# cancelled yet.
new_event = _threading.Event()
state.current_cancel_event = new_event
assert old_event.is_set() is True, (
"old turn's event must not be cleared by a new turn parking"
)
assert new_event.is_set() is False
# Cancelling the new turn flips ONLY the new event.
state.cancel_current_dispatch()
assert new_event.is_set() is True
# Old event still set independently.
assert old_event.is_set() is True
# ── Cancel key bindings ──────────────────────────────────────────────────────
class TestBuildCancelKeyBindings:
"""``_build_cancel_key_bindings`` returns prompt-level control
handlers. The handlers are extracted out of the
prompt loop so they can be exercised without the full async
machinery; this test instantiates the bindings and verifies they
were registered for the right keys."""
def test_returns_bindings_for_escape_and_ctrl_l(self) -> None:
state = loop_state.ReplState()
kb = build_cancel_key_bindings(state)
# Flatten each binding's keys tuple. ``Keys`` enum members have
# ``.value`` strings like ``"escape"``/``"c-l"`` matching the
# decorator argument; plain string keys are themselves.
registered = {getattr(k, "value", k) for b in kb.bindings for k in b.keys}
assert "escape" in registered, f"escape binding missing — registered: {registered}"
assert "c-l" in registered, f"Ctrl+L binding missing — registered: {registered}"
# ── Prompt confirmation bridge (worker-thread handoff) ──────────────────────────────
class TestRequestConfirmationViaPrompt:
"""``request_confirmation_via_prompt`` runs on the worker thread that
runs a queued turn. It parks on a ``threading.Event`` while the main
asyncio loop collects the next ``prompt_async`` return and hands the
text back via ``state.deliver_confirmation``. ``Esc`` (or any other
cancel path) flips ``state.current_cancel_event`` and the polling
loop raises ``DispatchCancelled`` within one prompt refresh tick.
These tests pin both paths so a stuck event can never leave a worker
parked forever. Each runs the function in a real background thread
and asserts it returns within a short timeout.
"""
# Generous join timeout — one poll tick is
# ``loop_state.PROMPT_REFRESH_INTERVAL_S`` (~100ms); every return path
# must complete well inside this, even on slow CI hardware. A
# regression that leaves the worker parked surfaces as a test
# failure (``t.is_alive()``) rather than a hang.
_JOIN_TIMEOUT_S = 2.0
# How long ``_wait_until_parked`` polls for the worker thread to
# reach the parked state. Worker parks in microseconds (a few
# Python statements after thread start), so 1s is a wide safety
# margin while still failing fast if the parking never happens.
# Must be < ``_JOIN_TIMEOUT_S`` so the parking check fails before
# the join would.
_PARK_TIMEOUT_S = 1.0
# Spin granularity for the parking poll. 5ms is fine-grained enough
# that the test sees the parked state within one or two ticks of it
# happening (~10ms upper bound on added test latency), while
# avoiding a tight loop that hogs the GIL from the worker thread
# we're waiting on.
_PARK_POLL_INTERVAL_S = 0.005
def _wait_until_parked(self, state: loop_state.ReplState) -> None:
"""Spin until the worker has assigned ``state.confirm_event``.
The function does this before its first ``response_event.wait``,
so once we see it, the worker is definitively in the poll loop
and ready to receive a delivery or cancel signal.
"""
deadline = time.monotonic() + self._PARK_TIMEOUT_S
while time.monotonic() < deadline:
if state.is_awaiting_confirmation():
return
time.sleep(self._PARK_POLL_INTERVAL_S)
raise AssertionError("worker never parked on confirm_event")
def _run_in_thread(
self, state: loop_state.ReplState, prompt_text: str
) -> tuple[threading.Thread, list[str], list[Exception]]:
"""Run the worker in a background thread and capture both its
return value (``result``) and any raised exception (``exc``).
Cancellation now raises :class:`controller_runtime.DispatchCancelled` instead
of returning ``""``, so tests need access to both channels —
the happy path checks ``result``, the cancel paths check
``exc``.
"""
result: list[str] = []
exc: list[Exception] = []
def target() -> None:
try:
result.append(
controller_runtime.request_confirmation_via_prompt(state, prompt_text)
)
except Exception as e:
exc.append(e)
t = threading.Thread(target=target, daemon=True)
t.start()
return t, result, exc
def test_returns_delivered_response_and_clears_state(self) -> None:
"""Happy path: main loop calls ``state.deliver_confirmation('y')``
→ worker wakes from poll → returns ``"y"``; ``confirm_event`` and
``confirm_response`` are cleared so the next confirmation starts
from a fresh slate.
"""
state = loop_state.ReplState()
# Active dispatch must have a cancel event parked; in production
# ``interactive_shell.runtime.turn_host.run_agent_turn`` allocates this
# before invoking the confirm_fn. Never set in this test.
state.current_cancel_event = threading.Event()
t, result, exc = self._run_in_thread(state, "Proceed? [y/N] ")
self._wait_until_parked(state)
# Prompt text is stored in state so prompt-toolkit owns the render.
assert state.confirm_prompt_text == "Proceed? [y/N] "
state.deliver_confirmation("y")
t.join(timeout=self._JOIN_TIMEOUT_S)
assert not t.is_alive(), "worker did not return within timeout"
assert exc == [], f"happy path raised unexpectedly: {exc}"
assert result == ["y"]
# Finally-block invariant: state cleared for the next prompt.
assert state.confirm_event is None
assert state.confirm_response == []
assert state.confirm_prompt_text == ""
def test_raises_dispatch_cancelled_when_cancel_event_fires(self) -> None:
"""Esc / ``/cancel`` **user-facing** path: goes through
``state.cancel_current_dispatch()``, which sets BOTH
``current_cancel_event`` AND ``confirm_event``. The worker
wakes from ``response_event.wait`` because ``confirm_event``
is the same object as ``response_event``; the polling loop
exits via the natural ``while not response_event.is_set()``
condition.
With ``confirm_response`` never populated, the function MUST
raise :class:`controller_runtime.DispatchCancelled` rather than returning
the empty string. Returning ``""`` would be silently confirmed
by ``execution_policy`` because ``[Y/n]`` treats empty as YES,
so the in-flight action would run despite the user cancelling.
Raising propagates out of ``execution_allowed`` and the
surrounding action loop instead, matching what the user
expects from ``Esc``.
"""
state = loop_state.ReplState()
state.current_cancel_event = threading.Event()
t, result, exc = self._run_in_thread(state, "Proceed? [y/N] ")
self._wait_until_parked(state)
state.cancel_current_dispatch()
t.join(timeout=self._JOIN_TIMEOUT_S)
assert not t.is_alive(), "worker did not return within timeout"
assert result == [], f"cancel path returned a value: {result}"
assert len(exc) == 1, f"expected exactly one exception, got {exc}"
assert isinstance(exc[0], controller_runtime.DispatchCancelled), (
f"expected DispatchCancelled, got {type(exc[0]).__name__}: {exc[0]}"
)
assert state.confirm_event is None
assert state.confirm_response == []
def test_raises_dispatch_cancelled_when_cancel_already_set_before_park(self) -> None:
"""Race-safety: if the user pressed Esc *before* the worker
reaches the poll loop, the first iteration must observe
``current_cancel_event.is_set()`` and raise immediately rather
than waiting forever for a confirmation that won't come.
Isolates the **in-loop cancel-check** on the FIRST iteration:
``current_cancel_event`` is pre-set; ``confirm_event`` is NOT
set; the worker enters the loop, reaches the cancel-check
BEFORE the first ``response_event.wait``, and raises
:class:`controller_runtime.DispatchCancelled`. Deleting the cancel-check
would make this test hang to ``_JOIN_TIMEOUT_S`` and fail.
"""
state = loop_state.ReplState()
cancel = threading.Event()
cancel.set() # already cancelled
state.current_cancel_event = cancel
t, result, exc = self._run_in_thread(state, "Proceed? ")
t.join(timeout=self._JOIN_TIMEOUT_S)
assert not t.is_alive(), "pre-set cancel did not unblock worker"
assert result == []
assert len(exc) == 1
assert isinstance(exc[0], controller_runtime.DispatchCancelled)
assert state.confirm_event is None
def test_in_loop_cancel_check_raises_after_wait_timeout(self) -> None:
"""In-loop cancel-check on a SUBSEQUENT iteration: the worker
is already parked in ``response_event.wait(timeout=0.1s)`` when
cancel fires. The wait times out (because we don't touch
``confirm_event``), the next iteration's cancel-check sees
``current_cancel_event.is_set()`` and raises
:class:`controller_runtime.DispatchCancelled`.
Why this is needed: the user-facing test
(``test_raises_dispatch_cancelled_when_cancel_event_fires``)
calls ``state.cancel_current_dispatch()``, which sets BOTH the
cancel and confirm events — the worker exits via the
confirm_event-set path (the post-wait empty-``confirm_response``
check), NOT via the cancel-check. Deleting the cancel-check
would still pass that test. This test sets ONLY the cancel
event so the only way out is through the check.
"""
state = loop_state.ReplState()
cancel = threading.Event()
state.current_cancel_event = cancel
t, result, exc = self._run_in_thread(state, "Proceed? ")
self._wait_until_parked(state)
# Set cancel directly — do NOT set confirm_event. The worker
# is inside ``response_event.wait(timeout=0.1s)``; that wait
# will time out, the next iteration's cancel-check then fires.
cancel.set()
t.join(timeout=self._JOIN_TIMEOUT_S)
assert not t.is_alive(), (
"in-loop cancel-check did not return within timeout — "
"the function ignored a cancel signal that arrived mid-wait"
)
assert result == []
assert len(exc) == 1
assert isinstance(exc[0], controller_runtime.DispatchCancelled)
assert state.confirm_event is None
def test_confirm_response_reset_before_confirm_event_published(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Race-safety: the response list MUST be reset before
``confirm_event`` is published. Otherwise a concurrent
``deliver_confirmation`` (running between publish and reset)
appends to the current list, the next statement rebinds
``confirm_response`` to ``[]``, and the user's answer is
silently dropped.
``deliver_confirmation`` early-exits when ``confirm_event is
None``, so resetting the list first is invisible to the main
thread; only the event publish makes the parking observable.
This test instruments ``__setattr__`` to verify the ordering
deterministically — timing-based tests can't reliably hit the
sub-microsecond race window even when the bug is present.
"""
state = loop_state.ReplState()
state.current_cancel_event = threading.Event()
# Track every ``confirm_event`` / ``confirm_response`` write
# made by ``request_confirmation_via_prompt``. Monkeypatching
# AFTER state construction so the dataclass ``__init__`` field
# writes don't pollute the recorded order.
assignments: list[str] = []
real_setattr = loop_state.ReplState.__setattr__
def tracking_setattr(obj: object, name: str, value: object) -> None:
if name in ("confirm_event", "confirm_response"):
assignments.append(name)
real_setattr(obj, name, value) # type: ignore[arg-type]
monkeypatch.setattr(loop_state.ReplState, "__setattr__", tracking_setattr)
t, result, exc = self._run_in_thread(state, "Proceed? ")
self._wait_until_parked(state)
state.deliver_confirmation("answer")
t.join(timeout=self._JOIN_TIMEOUT_S)
assert exc == [], f"happy path raised unexpectedly: {exc}"
assert result == ["answer"]
# During setup ``request_confirmation_via_prompt`` writes both
# attributes. The first write of each is the setup phase
# (later writes are the ``finally`` cleanup which clears
# both — order there doesn't matter). Pull out just the
# setup-phase assignment order.
setup_order: list[str] = []
for name in assignments:
setup_order.append(name)
if name == "confirm_event":
break # confirm_event publish is the last setup write
assert "confirm_response" in setup_order, (
f"confirm_response never reset during setup — saw {setup_order}"
)
response_idx = setup_order.index("confirm_response")
event_idx = setup_order.index("confirm_event")
assert response_idx < event_idx, (
f"race window: confirm_event published before "
f"confirm_response was reset — order was {setup_order}"
)
def test_empty_string_delivery_returns_empty_string(self) -> None:
"""User presses Enter on the confirmation prompt without typing
anything → ``state.deliver_confirmation("")`` is called →
function returns ``""``. Real production case: ``[Y/n]`` prompts
treat plain Enter as "yes" (capital Y default).
This is the ONLY path that legitimately yields an empty-string
return: ``deliver_confirmation`` populated ``confirm_response``
with ``""`` BEFORE setting the event, so the post-wait check
sees a non-empty list and returns the user's actual answer.
Cancellation, by contrast, sets the event WITHOUT delivering
an answer and is now distinguishable — that path raises
:class:`controller_runtime.DispatchCancelled` (see the cancel-path tests
above).
"""
state = loop_state.ReplState()
state.current_cancel_event = threading.Event()
t, result, exc = self._run_in_thread(state, "Proceed? [y/N] ")
self._wait_until_parked(state)
state.deliver_confirmation("")
t.join(timeout=self._JOIN_TIMEOUT_S)
assert not t.is_alive(), "empty delivery did not unblock worker"
assert exc == [], f"explicit empty delivery should not raise: {exc}"
assert result == [""]
assert state.confirm_event is None
assert state.confirm_response == []
class TestExecutionAllowedRespectsDispatchCancelled:
"""End-to-end contract: cancelling during ``Proceed? [Y/n]`` must
actually STOP the in-flight action — not just stop the spinner.
Pre-fix, the cancel handler returned ``""``; ``execution_allowed``
treated empty as YES (since ``[Y/n]`` defaults to Y) and the worker
happily ran the action it was supposed to interrupt — only the
spinner stopped, the agent kept going. This test class pins the
new contract: the confirm callable raises ``DispatchCancelled`` and
the exception propagates out of ``execution_allowed`` *without*
silently confirming the action. The action loop in
``run_action_tool_turn`` therefore exits via the exception, the
in-flight action never runs, and any further actions in the same
plan are skipped.
"""
def test_dispatch_cancelled_propagates_through_execution_allowed(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
from rich.console import Console
from surfaces.interactive_shell.ui.execution_confirm import execution_allowed
from tools.interactive_shell.shared import (
ExecutionPolicyResult,
)
monkeypatch.setattr("sys.stdin.isatty", lambda: True)
session = Session()
console = Console(file=io.StringIO(), force_terminal=False, highlight=False)
def _cancel_confirm(_prompt: str) -> str:
raise controller_runtime.DispatchCancelled("cancelled while awaiting confirmation")
policy = ExecutionPolicyResult(
verdict="ask",
tool_type="opensre_cli",
reason="this opensre subcommand may change local config or infrastructure",
)
with pytest.raises(controller_runtime.DispatchCancelled):
execution_allowed(
policy,
session=session,
console=console,
action_summary="opensre remote health --help",
confirm_fn=_cancel_confirm,
is_tty=True,
)
def test_empty_confirm_response_would_silently_allow_without_raise(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Regression guard documenting WHY the raise is required.
If the confirm callable returned ``""`` instead of raising,
``execution_allowed`` would treat the empty answer as YES
(``[Y/n]`` defaults to Y) and the action would run. This test
pins that footgun by demonstrating the bad outcome with an
empty-string return — the cancel handler MUST raise, not
return, to actually stop the in-flight action.
"""
from rich.console import Console
from surfaces.interactive_shell.ui.execution_confirm import execution_allowed
from tools.interactive_shell.shared import (
ExecutionPolicyResult,
)
monkeypatch.setattr("sys.stdin.isatty", lambda: True)
session = Session()
console = Console(file=io.StringIO(), force_terminal=False, highlight=False)
policy = ExecutionPolicyResult(
verdict="ask",
tool_type="opensre_cli",
reason="this opensre subcommand may change local config or infrastructure",
)
# Empty string answer (the pre-fix cancel return value) is
# silently confirmed — proving the bug existed and that we
# cannot rely on a sentinel string for cancellation.
assert (
execution_allowed(
policy,
session=session,
console=console,
action_summary="opensre remote health --help",
confirm_fn=lambda _prompt: "",
is_tty=True,
)
is True
)
class TestThemeCommand:
@staticmethod
def _capture() -> tuple:
from rich.console import Console
buf = io.StringIO()
return Console(file=buf, force_terminal=False, highlight=False), buf
def test_theme_command_is_registered(self) -> None:
assert "/theme" in SLASH_COMMANDS
def test_theme_command_updates_active_theme_and_persists(self, monkeypatch) -> None:
from surfaces.interactive_shell.command_registry import theme as theme_cmd
saved_payloads: list[dict[str, object]] = []
monkeypatch.setattr(theme_cmd, "repl_tty_interactive", lambda: True)
monkeypatch.setattr(theme_cmd, "repl_choose_one", lambda **_kwargs: "blue")
monkeypatch.setattr(theme_cmd, "_refresh_prompt_style", lambda _session: None)
monkeypatch.setattr("surfaces.cli.commands.config._load_config", lambda: {})
monkeypatch.setattr(
"surfaces.cli.commands.config._save_config",
lambda data: saved_payloads.append(dict(data)),
)
set_active_theme("green")
session = Session()
console, _buf = self._capture()
assert dispatch_slash("/theme", session, console) is True
assert get_active_theme_name() == "blue"
assert saved_payloads
interactive = saved_payloads[-1].get("interactive")
assert isinstance(interactive, dict)
assert interactive.get("theme") == "blue"
def test_theme_picker_uses_session_theme_as_current(self, monkeypatch) -> None:
from surfaces.interactive_shell.command_registry import theme as theme_cmd
monkeypatch.setattr(theme_cmd, "repl_tty_interactive", lambda: True)
captured: dict[str, object] = {}
def _fake_choose_one(**kwargs: object) -> None:
captured.update(kwargs)
return None
monkeypatch.setattr(theme_cmd, "repl_choose_one", _fake_choose_one)
session = Session()
session.terminal.active_theme_name = "pink"
set_active_theme("pink")
console, _buf = self._capture()
dispatch_slash("/theme", session, console)
assert captured.get("initial_value") == "pink"
choices = captured.get("choices")
assert isinstance(choices, list)
assert any("pink (current)" in label for _value, label in choices)
def test_theme_command_direct_arg_sets_theme(self, monkeypatch) -> None:
from surfaces.interactive_shell.command_registry import theme as theme_cmd
monkeypatch.setattr(theme_cmd, "repl_tty_interactive", lambda: True)
monkeypatch.setattr(theme_cmd, "_refresh_prompt_style", lambda _session: None)
monkeypatch.setattr("surfaces.cli.commands.config._load_config", lambda: {})
monkeypatch.setattr("surfaces.cli.commands.config._save_config", lambda _data: None)
set_active_theme("green")
session = Session()
console, _buf = self._capture()
assert dispatch_slash("/theme amber", session, console) is True
assert get_active_theme_name() == "amber"
def test_theme_change_refreshes_welcome_poster(self, monkeypatch) -> None:
from surfaces.interactive_shell.command_registry import theme as theme_cmd
monkeypatch.setattr(theme_cmd, "repl_tty_interactive", lambda: True)
monkeypatch.setattr(theme_cmd, "repl_choose_one", lambda **_kwargs: "blue")
monkeypatch.setattr(theme_cmd, "_refresh_prompt_style", lambda _session: None)
monkeypatch.setattr("surfaces.cli.commands.config._load_config", lambda: {})
monkeypatch.setattr("surfaces.cli.commands.config._save_config", lambda _data: None)
refreshed: list[dict[str, object | None]] = []
def _refresh(
console: object,
*,
session: object = None,
theme_notice: str | None = None,
) -> None:
refreshed.append({"console": console, "session": session, "theme_notice": theme_notice})
monkeypatch.setattr(
"surfaces.interactive_shell.ui.components.rendering.refresh_welcome_poster",
_refresh,
)
session = Session()
console, _buf = self._capture()
assert dispatch_slash("/theme", session, console) is True
assert len(refreshed) == 1
assert refreshed[0]["session"] is session
assert refreshed[0]["theme_notice"] == "blue"
def test_theme_picker_lists_all_registered_themes(self, monkeypatch) -> None:
from platform.terminal.theme import list_theme_names
from surfaces.interactive_shell.command_registry import theme as theme_cmd
monkeypatch.setattr(theme_cmd, "repl_tty_interactive", lambda: True)
captured: dict[str, object] = {}
def _fake_choose_one(**kwargs: object) -> None:
captured.update(kwargs)
return None
monkeypatch.setattr(theme_cmd, "repl_choose_one", _fake_choose_one)
session = Session()
console, _buf = self._capture()
dispatch_slash("/theme", session, console)
choices = captured.get("choices")
assert isinstance(choices, list)
assert [value for value, _label in choices] == list(list_theme_names())
def test_theme_persist_drains_cpr_around_poster_and_defers_prompt_refresh(
self, monkeypatch
) -> None:
from surfaces.interactive_shell.command_registry import theme as theme_cmd
drains: list[str] = []
monkeypatch.setattr(
theme_cmd,
"_settle_and_drain_cpr",
lambda: drains.append("drain"),
)
monkeypatch.setattr(
"surfaces.interactive_shell.ui.components.rendering.refresh_welcome_poster",
lambda *_args, **_kwargs: drains.append("poster"),
)
monkeypatch.setattr("surfaces.cli.commands.config._load_config", lambda: {})
monkeypatch.setattr("surfaces.cli.commands.config._save_config", lambda _data: None)
session = Session()
console, _buf = self._capture()
theme_cmd._persist_and_report_theme(session, console, "pink")
assert drains == ["drain", "poster", "drain"]
assert session.terminal.pending_theme_refresh is True
def test_refresh_prompt_theme_skips_invalidate_when_app_not_running() -> None:
from surfaces.interactive_shell.ui.input_prompt.style import refresh_prompt_theme
invalidated: list[bool] = []
class _App:
is_running = False
style = None
placeholder = None
renderer = None
def invalidate(self) -> None:
invalidated.append(True)
session = Session()
session.terminal.prompt_app = _App()
refresh_prompt_theme(session)
assert invalidated == []
assert session.terminal.prompt_app.style is not None