1240 lines
43 KiB
Python
1240 lines
43 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
from __future__ import annotations
|
|
|
|
import curses
|
|
from dataclasses import fields
|
|
from multiprocessing.queues import Queue
|
|
from queue import Empty
|
|
from tests.band_aider_loop_impl import _idle_language_strategy_set
|
|
|
|
from band_aider.context.class_meta import ClassMeta, ContentType
|
|
from band_aider.context.global_context import StoryCursor, TaskCursor
|
|
from band_aider.context.cot_task_generator import CotTaskRecord
|
|
from band_aider.context.working_backlog import BackLog, ChangeType, InspectionBacklog, TestingBacklog
|
|
from band_aider.ui.dto.cot_status import CotStatusDTO
|
|
from band_aider.ui.dto.human_intervention import (
|
|
CycleInterruptCommandDTO,
|
|
InterventionSuppressionCommandDTO,
|
|
HumanInterventionRequestDTO,
|
|
HumanInterventionResponseDTO,
|
|
)
|
|
from band_aider.ui.dto.log_message import LogMessageDTO
|
|
from band_aider.ui.dto.loop_status import LoopStatusDTO
|
|
from band_aider.ui.dto.model_escalation import ModelEscalationDTO
|
|
from band_aider.ui.dto.run_completion import RunCompletionDTO
|
|
from band_aider.ui.escalation_ui import (
|
|
EscalationUIState,
|
|
HELP_SUBTITLE,
|
|
SEND_HINT_KEY,
|
|
_draw_header_content,
|
|
draw_escalation_screen,
|
|
_escalation_pane_layout,
|
|
handle_escalation_key,
|
|
render_plain_escalation_prompt,
|
|
)
|
|
from band_aider.ui.dto.supreme_status import SupremeStatusDTO
|
|
from band_aider.ui.dto.stabilization_status import StabilizationStatusDTO
|
|
from band_aider.ui.dto.token_budget_stats import TokenBudgetBucketDTO, TokenBudgetStatsDTO
|
|
from band_aider.ui.status import (
|
|
publish_backlog_status,
|
|
publish_bootstrap_status,
|
|
publish_cot_status,
|
|
publish_loop_status,
|
|
status_publisher_scope,
|
|
)
|
|
from band_aider.loop.alpha_loop import AlphaLoop
|
|
from band_aider.loop.beta_loop import BetaLoop
|
|
from band_aider.ui.terminal_ui import (
|
|
DEFAULT_PANEL_TITLE,
|
|
TerminalUI,
|
|
_format_token_count,
|
|
_safe_color_pair,
|
|
_token_bar,
|
|
_token_bar_segments,
|
|
_wrap_panel_lines,
|
|
render_completion_lines,
|
|
render_cot_lines,
|
|
render_projection,
|
|
)
|
|
|
|
|
|
class _Publisher:
|
|
def __init__(self) -> None:
|
|
self.events: list[object] = []
|
|
|
|
def publish(self, event: object) -> None:
|
|
self.events.append(event)
|
|
|
|
|
|
class _Queue(Queue):
|
|
"""Queue-backed terminal boundary with stationary preloaded events."""
|
|
|
|
def __init__(self, events: list[object] | None = None) -> None:
|
|
self._events = list(events or [])
|
|
self.items: list[object] = []
|
|
|
|
def get_nowait(self) -> object:
|
|
"""Return the next predefined terminal event without blocking."""
|
|
|
|
if not self._events:
|
|
raise Empty
|
|
return self._events.pop(0)
|
|
|
|
def put(self, obj: object, block: bool = True, timeout: float | None = None) -> None:
|
|
"""Record one command without allocating an operating-system queue."""
|
|
|
|
self.items.append(obj)
|
|
|
|
|
|
class _MemoryWindow:
|
|
"""In-memory curses subwindow that records visible text through its screen."""
|
|
|
|
def __init__(self, screen: "_MemoryScreen", y: int, x: int) -> None:
|
|
self._screen = screen
|
|
self._y = y
|
|
self._x = x
|
|
|
|
def erase(self) -> None:
|
|
"""Model an in-memory redraw boundary."""
|
|
|
|
def border(self) -> None:
|
|
"""Model a border without terminal I/O."""
|
|
|
|
def addstr(self, y: int, x: int, text: str, attr: int = curses.A_NORMAL) -> None:
|
|
"""Record text at the window-relative coordinate."""
|
|
|
|
self._screen.addstr(self._y + y, self._x + x, text, attr)
|
|
|
|
|
|
class _MemoryScreen:
|
|
"""In-memory terminal boundary with fixed dimensions and optional key input."""
|
|
|
|
def __init__(
|
|
self,
|
|
height: int = 24,
|
|
width: int = 120,
|
|
keys: list[int] | None = None,
|
|
) -> None:
|
|
self._height = height
|
|
self._width = width
|
|
self._keys = list(keys or [])
|
|
self.writes: list[tuple[int, int, str]] = []
|
|
|
|
@property
|
|
def text(self) -> str:
|
|
"""Return all visible text in draw order."""
|
|
|
|
return "\n".join(text for _, _, text in self.writes)
|
|
|
|
def erase(self) -> None:
|
|
"""Clear prior paint records for one frame."""
|
|
|
|
self.writes.clear()
|
|
|
|
def refresh(self) -> None:
|
|
"""Finish an in-memory frame without flushing a terminal."""
|
|
|
|
def getmaxyx(self) -> tuple[int, int]:
|
|
"""Return the stationary terminal dimensions."""
|
|
|
|
return self._height, self._width
|
|
|
|
def addstr(self, y: int, x: int, text: str, attr: int = curses.A_NORMAL) -> None:
|
|
"""Record one clipped terminal write."""
|
|
|
|
self.writes.append((y, x, text))
|
|
|
|
def derwin(self, height: int, width: int, y: int, x: int) -> _MemoryWindow:
|
|
"""Return an in-memory child window for a drawn panel."""
|
|
|
|
return _MemoryWindow(self, y, x)
|
|
|
|
def getch(self) -> int:
|
|
"""Return the next preloaded keystroke without blocking."""
|
|
|
|
return self._keys.pop(0) if self._keys else -1
|
|
|
|
|
|
def _intervention_request() -> HumanInterventionRequestDTO:
|
|
return HumanInterventionRequestDTO(
|
|
request_id="req-1",
|
|
check_name="CycleJacker",
|
|
original_prompt="Update terminal_ui to isolate escalation mode.",
|
|
repeated_response_tail="No edits required.",
|
|
request_md5="a" * 32,
|
|
response_md5="b" * 32,
|
|
local_model="local",
|
|
escalated_model="cloud-2",
|
|
)
|
|
|
|
|
|
def _escalation_event() -> ModelEscalationDTO:
|
|
return ModelEscalationDTO(
|
|
trigger_name="CycleJacker",
|
|
occurred_at=1784056538,
|
|
prompt_preview="prompt",
|
|
reason="Equivalent response repeated 4 times.",
|
|
request_md5="a" * 32,
|
|
local_model="PLEModels default",
|
|
active_model="gpt-5.5",
|
|
config_index=2,
|
|
config_count=3,
|
|
state="human_required",
|
|
response_md5="b" * 32,
|
|
response_tail="tail",
|
|
repeat_count=4,
|
|
)
|
|
|
|
|
|
def test_backlog_mutation_publishes_backlog_status() -> None:
|
|
publisher = _Publisher()
|
|
backlog = BackLog(completion={})
|
|
item = InspectionBacklog(
|
|
focused_spotlight=ClassMeta(
|
|
location="src/foo.py",
|
|
summary="Inspect source.",
|
|
relevance=[],
|
|
relevant_symbols={},
|
|
),
|
|
summary="Inspect source.",
|
|
)
|
|
|
|
with status_publisher_scope(publisher):
|
|
backlog.queue_inspection(item)
|
|
|
|
status_events = [event for event in publisher.events if isinstance(event, SupremeStatusDTO)]
|
|
assert status_events
|
|
assert status_events[-1].backlog is not None
|
|
assert status_events[-1].backlog.ins == 1
|
|
assert status_events[-1].backlog.alpha_inbound_queue == 0
|
|
assert status_events[-1].backlog.alpha_outbound_queue == 0
|
|
|
|
|
|
def test_stabilization_status_publishes_alpha_beta_queue_depths() -> None:
|
|
publisher = _Publisher()
|
|
backlog = BackLog(completion={})
|
|
context = type(
|
|
"StabilizationStatusContext",
|
|
(),
|
|
{
|
|
"backlog": backlog,
|
|
"deferred_remediation_backlog": {},
|
|
"alpha_inbound_queue_depth": 1,
|
|
"alpha_outbound_queue_depth": 2,
|
|
"beta_inbound_queue_depth": 3,
|
|
"beta_outbound_queue_depth": 4,
|
|
"final_exam_started": False,
|
|
"final_exam_finished": False,
|
|
"root_change": None,
|
|
"frontier": ([], 0),
|
|
"full_topologies": [],
|
|
"visited_frontier_locations": set(),
|
|
"active_topology": None,
|
|
},
|
|
)()
|
|
|
|
with status_publisher_scope(publisher):
|
|
from band_aider.ui.status import publish_stabilization_status
|
|
|
|
publish_stabilization_status(context)
|
|
|
|
status_events = [event for event in publisher.events if isinstance(event, SupremeStatusDTO)]
|
|
assert status_events
|
|
assert status_events[-1].backlog is not None
|
|
assert status_events[-1].backlog.alpha_inbound_queue == 1
|
|
assert status_events[-1].backlog.alpha_outbound_queue == 2
|
|
assert status_events[-1].backlog.beta_inbound_queue == 3
|
|
assert status_events[-1].backlog.beta_outbound_queue == 4
|
|
|
|
|
|
def test_beta_claim_status_replaces_stale_alpha_item_and_tracks_each_test() -> None:
|
|
"""Verify Beta owns live status once it claims each concrete test artifact."""
|
|
|
|
publisher = _Publisher()
|
|
beta = object.__new__(BetaLoop)
|
|
first = TestingBacklog(
|
|
focused_spotlight=ClassMeta(
|
|
location="tests/test_first.ext",
|
|
summary="First test.",
|
|
relevance=[],
|
|
relevant_symbols={"first_case": "current callable"},
|
|
content_type=ContentType.Test,
|
|
),
|
|
summary="Run first test.",
|
|
)
|
|
second = TestingBacklog(
|
|
focused_spotlight=ClassMeta(
|
|
location="tests/test_second.ext",
|
|
summary="Second test.",
|
|
relevance=[],
|
|
relevant_symbols={"second_case": "current callable"},
|
|
content_type=ContentType.Test,
|
|
),
|
|
summary="Run second test.",
|
|
)
|
|
|
|
with status_publisher_scope(publisher):
|
|
publish_loop_status("AlphaLoop", current_item="src/stale.ext")
|
|
beta._publish_work_status(first)
|
|
beta._publish_work_status(second)
|
|
|
|
events = [event for event in publisher.events if isinstance(event, SupremeStatusDTO)]
|
|
assert events[-2].loop is not None
|
|
assert events[-2].loop.current_loop == "BetaLoop"
|
|
assert events[-2].loop.current_item == "tests/test_first.ext"
|
|
assert events[-2].loop.work_focus == "first_case"
|
|
assert events[-1].loop is not None
|
|
assert events[-1].loop.current_loop == "BetaLoop"
|
|
assert events[-1].loop.current_item == "tests/test_second.ext"
|
|
assert events[-1].loop.work_focus == "second_case"
|
|
|
|
|
|
def test_alpha_claim_status_publishes_concrete_source_with_generic_focus() -> None:
|
|
"""Verify Alpha owns live status for its claimed source artifact."""
|
|
|
|
publisher = _Publisher()
|
|
alpha = object.__new__(AlphaLoop)
|
|
item = InspectionBacklog(
|
|
focused_spotlight=ClassMeta(
|
|
location="src/component.ext",
|
|
summary="Update component.",
|
|
relevance=[],
|
|
relevant_symbols={"render_component": "current callable"},
|
|
),
|
|
summary="Update component.",
|
|
)
|
|
|
|
with status_publisher_scope(publisher):
|
|
alpha._publish_work_status(item)
|
|
|
|
status = [event for event in publisher.events if isinstance(event, SupremeStatusDTO)][-1]
|
|
assert status.loop is not None
|
|
assert status.loop.current_loop == "AlphaLoop"
|
|
assert status.loop.current_item == "src/component.ext"
|
|
assert status.loop.work_focus == "render_component"
|
|
|
|
|
|
def test_loop_status_contract_is_language_neutral_and_ui_renders_published_focus() -> None:
|
|
"""Verify the generic status DTO and renderer contain no language-specific discovery contract."""
|
|
|
|
names = {item.name for item in fields(LoopStatusDTO)}
|
|
assert names == {"previous_loop", "current_loop", "next_loop", "current_item", "work_focus"}
|
|
projection = TerminalUI(_Queue(), _Queue()).projection
|
|
projection.status = SupremeStatusDTO(
|
|
stabilization=StabilizationStatusDTO(root="src/root.ext", current="src/stale.ext"),
|
|
loop=LoopStatusDTO(
|
|
previous_loop="StabilizationLoop",
|
|
current_loop="BetaLoop",
|
|
next_loop=None,
|
|
current_item="tests/component_spec.ext",
|
|
work_focus="component_contract",
|
|
),
|
|
)
|
|
|
|
rendered = render_projection(projection)
|
|
|
|
assert "STABILIZATION (BETA)" in rendered
|
|
assert "Current tests/component_spec.ext" in rendered
|
|
assert "Focus component_contract" in rendered
|
|
assert "python" not in " ".join(names).lower()
|
|
assert "pytest" not in " ".join(names).lower()
|
|
|
|
|
|
def test_bootstrap_status_publishes_panel_and_temporary_cot_task() -> None:
|
|
publisher = _Publisher()
|
|
context = type(
|
|
"BootstrapStatusContext",
|
|
(),
|
|
{
|
|
"goal": "Add a terminal UI progress display.",
|
|
"stage": "beats_extracted",
|
|
"iteration": 0,
|
|
"max_iterations": 3,
|
|
"visible_locations": ["band_aider/ui/terminal_ui.py"],
|
|
"beats_by_location": {"band_aider/ui/terminal_ui.py": ["draw status", "render COT"]},
|
|
"cross_cutting_concerns": [{"name": "terminal status"}],
|
|
"selected_concern": {"name": "terminal status"},
|
|
"idiom_catalog": [],
|
|
"pending_idiom_pairs": [],
|
|
"design_plan": {},
|
|
"created_locations": [],
|
|
"involved_locations": [],
|
|
"refreshed_locations": [],
|
|
},
|
|
)()
|
|
|
|
with status_publisher_scope(publisher):
|
|
publish_bootstrap_status(context)
|
|
|
|
status_events = [event for event in publisher.events if isinstance(event, SupremeStatusDTO)]
|
|
assert status_events
|
|
status = status_events[-1]
|
|
assert status.task == "Add a terminal UI progress display."
|
|
assert status.bootstrap is not None
|
|
assert status.bootstrap.stage_label == "concern selection"
|
|
assert status.bootstrap.beat_count == 2
|
|
assert status.bootstrap.current == "beats band_aider/ui/terminal_ui.py: 2"
|
|
assert status.stabilization is None
|
|
assert status.cot is not None
|
|
assert status.cot.all_tasks == ("bootstrap: concern selection",)
|
|
|
|
|
|
def test_render_projection_uses_bootstrap_panel_in_stabilization_slot() -> None:
|
|
publisher = _Publisher()
|
|
context = type(
|
|
"BootstrapStatusContext",
|
|
(),
|
|
{
|
|
"goal": "Add bootstrap progress.",
|
|
"stage": "concern_selected",
|
|
"iteration": 1,
|
|
"max_iterations": 3,
|
|
"visible_locations": ["src/a.py"],
|
|
"beats_by_location": {"src/a.py": ["beat"]},
|
|
"cross_cutting_concerns": [{"name": "shared progress"}],
|
|
"selected_concern": {"name": "shared progress"},
|
|
"idiom_catalog": [],
|
|
"pending_idiom_pairs": [("src/a.py",)],
|
|
"design_plan": {},
|
|
"created_locations": [],
|
|
"involved_locations": [],
|
|
"refreshed_locations": [],
|
|
},
|
|
)()
|
|
|
|
with status_publisher_scope(publisher):
|
|
publish_bootstrap_status(context)
|
|
|
|
status = [event for event in publisher.events if isinstance(event, SupremeStatusDTO)][-1]
|
|
projection = TerminalUI(_Queue(), _Queue()).projection
|
|
projection.status = status
|
|
rendered = render_projection(projection)
|
|
|
|
assert "BOOTSTRAP (idiom samples)" in rendered
|
|
assert "STABILIZATION" not in rendered
|
|
assert "1 > bootstrap: idiom samples" in rendered
|
|
|
|
|
|
def test_backlog_status_holds_peak_for_five_minutes() -> None:
|
|
"""Verify the decay calculation retains a peak until its explicit deadline."""
|
|
|
|
from band_aider.ui.status import _held_backlog_value
|
|
|
|
publisher = _Publisher()
|
|
|
|
with status_publisher_scope(publisher):
|
|
assert _held_backlog_value("cmp", 1, 1_000.0) == 1
|
|
assert _held_backlog_value("cmp", 0, 1_001.0) == 1
|
|
assert _held_backlog_value("cmp", 0, 1_301.0) == 0
|
|
|
|
|
|
def test_alpha_loop_publishes_live_queue_depth_snapshots() -> None:
|
|
"""Verify Alpha queue depth reaches the shared status projection unchanged."""
|
|
|
|
publisher = _Publisher()
|
|
|
|
with status_publisher_scope(publisher):
|
|
publish_backlog_status(
|
|
BackLog(completion={}),
|
|
alpha_inbound_queue_depth=1,
|
|
alpha_outbound_queue_depth=0,
|
|
)
|
|
|
|
status = [event for event in publisher.events if isinstance(event, SupremeStatusDTO)][-1]
|
|
assert status.backlog is not None
|
|
assert (status.backlog.alpha_inbound_queue, status.backlog.alpha_outbound_queue) == (1, 0)
|
|
|
|
|
|
def test_beta_loop_publishes_live_queue_depth_snapshots() -> None:
|
|
"""Verify Beta queue depth reaches the shared status projection unchanged."""
|
|
|
|
publisher = _Publisher()
|
|
|
|
with status_publisher_scope(publisher):
|
|
publish_backlog_status(
|
|
BackLog(completion={}),
|
|
beta_inbound_queue_depth=1,
|
|
beta_outbound_queue_depth=0,
|
|
)
|
|
|
|
status = [event for event in publisher.events if isinstance(event, SupremeStatusDTO)][-1]
|
|
assert status.backlog is not None
|
|
assert (status.backlog.beta_inbound_queue, status.backlog.beta_outbound_queue) == (1, 0)
|
|
|
|
|
|
def test_backlog_status_renders_final_exam_warning() -> None:
|
|
publisher = _Publisher()
|
|
backlog = BackLog(completion={})
|
|
marker = ClassMeta(
|
|
location="__last_exam__",
|
|
summary="last exam",
|
|
relevance=[],
|
|
relevant_symbols={},
|
|
)
|
|
|
|
with status_publisher_scope(publisher):
|
|
backlog.queue_testing(
|
|
TestingBacklog(
|
|
focused_spotlight=marker,
|
|
summary="last exam",
|
|
downstream_classes=[marker],
|
|
change_type=ChangeType.Update,
|
|
last_exam=True,
|
|
)
|
|
)
|
|
|
|
status_events = [event for event in publisher.events if isinstance(event, SupremeStatusDTO)]
|
|
assert status_events[-1].backlog is not None
|
|
assert status_events[-1].backlog.final_exam_pending is True
|
|
projection = TerminalUI(_Queue(), _Queue()).projection
|
|
projection.status = status_events[-1]
|
|
rendered = render_projection(projection)
|
|
assert "LAST EXAM PENDING" in rendered
|
|
|
|
|
|
def test_terminal_ui_log_controls_filter_visible_logs() -> None:
|
|
ui = TerminalUI(_Queue(), _Queue())
|
|
ui.projection.logs.append(
|
|
LogMessageDTO(created=1.0, level="INFO", logger_name="beta-loop.search", message="all good")
|
|
)
|
|
ui.projection.logs.append(
|
|
LogMessageDTO(created=2.0, level="WARN", logger_name="beta-loop.search", message="ImportError found")
|
|
)
|
|
ui.projection.logs.append(
|
|
LogMessageDTO(created=3.0, level="WARN", logger_name="alpha-loop.edit", message="ImportError ignored")
|
|
)
|
|
ui._filter_text = "ImportError"
|
|
ui._logger_filter = "beta-loop"
|
|
ui._level_filter = "WARN"
|
|
ui._context_before = "0"
|
|
ui._context_after = "0"
|
|
|
|
visible = ui._visible_logs(10)
|
|
|
|
assert [item.message for item in visible] == ["ImportError found"]
|
|
|
|
|
|
def test_visible_log_lines_wrap_long_log_rows() -> None:
|
|
ui = TerminalUI(_Queue(), _Queue())
|
|
ui.projection.logs.append(
|
|
LogMessageDTO(
|
|
created=1.0,
|
|
level="INFO",
|
|
logger_name="restored.window.viewport.logger",
|
|
message="this message should wrap when the viewport gets narrow",
|
|
)
|
|
)
|
|
|
|
lines = ui._visible_log_lines(24)
|
|
|
|
assert len(lines) > 1
|
|
assert lines[0].startswith("INFO ")
|
|
assert all(len(line) <= 24 for line in lines)
|
|
|
|
|
|
def test_render_projection_includes_rich_escalation_details() -> None:
|
|
ui = TerminalUI(_Queue(), _Queue())
|
|
ui.projection.escalation = ModelEscalationDTO(
|
|
trigger_name="CycleJacker",
|
|
occurred_at=0,
|
|
prompt_preview="line one\nline two",
|
|
reason="Equivalent response repeated 2 times.",
|
|
request_md5="a" * 32,
|
|
local_model="local",
|
|
active_model="cloud-1",
|
|
config_index=1,
|
|
config_count=3,
|
|
state="escalating",
|
|
response_md5="b" * 32,
|
|
response_tail="tail",
|
|
repeat_count=2,
|
|
)
|
|
|
|
rendered = render_projection(ui.projection)
|
|
|
|
assert "Trigger CycleJacker" in rendered
|
|
assert "When unknown" in rendered
|
|
assert "Prompt" in rendered
|
|
assert " line one" in rendered
|
|
|
|
|
|
def test_render_projection_includes_token_budget_graph() -> None:
|
|
ui = TerminalUI(_Queue(), _Queue())
|
|
ui.projection.status = SupremeStatusDTO(
|
|
token_budget=TokenBudgetStatsDTO(
|
|
max_prompt_tokens=100,
|
|
last_ai_token_used=25,
|
|
total_tokens=225,
|
|
total_cache_hit_tokens=75,
|
|
bucket_window_seconds=300,
|
|
buckets=(
|
|
TokenBudgetBucketDTO(started_at=0, tokens_used=10),
|
|
TokenBudgetBucketDTO(started_at=300, tokens_used=60),
|
|
TokenBudgetBucketDTO(started_at=600, tokens_used=120),
|
|
),
|
|
)
|
|
)
|
|
|
|
rendered = render_projection(ui.projection)
|
|
|
|
assert "TOKEN BUDGET" in rendered
|
|
assert f"Tokens spent {_format_token_count(225)}" in rendered
|
|
assert f"Cache hits {_format_token_count(75)}" in rendered
|
|
assert "Last used" not in rendered
|
|
assert f"Baseline {_format_token_count(225)} 100.0%" in rendered
|
|
assert f"Escalated {_format_token_count(0)} 0.0%" in rendered
|
|
assert f"t-0 {_token_bar(120, 100)} {_format_token_count(120)}" in rendered
|
|
|
|
|
|
def test_render_completion_lines_include_summary_report_and_statistics() -> None:
|
|
completion = RunCompletionDTO(
|
|
final_summary="SUCCESS updated reporting.",
|
|
report_path="/repo/band-aider reports/run-report-ending-20260717-141500-0700.md",
|
|
total_tokens_spent=12345,
|
|
largest_query_tokens=6789,
|
|
total_seconds=125,
|
|
average_seconds_per_cot=42,
|
|
cot_count=3,
|
|
files_modified=2,
|
|
files_added=1,
|
|
llm_call_count=4,
|
|
llm_wait_seconds=91,
|
|
baseline_tokens_spent=10000,
|
|
escalated_tokens_spent=2345,
|
|
total_cache_hit_tokens=456,
|
|
)
|
|
|
|
rendered = "\n".join(render_completion_lines(completion))
|
|
|
|
assert "RUN COMPLETE" in rendered
|
|
assert "Final summary: SUCCESS updated reporting." in rendered
|
|
assert "Report: /repo/band-aider reports/run-report-ending-20260717-141500-0700.md" in rendered
|
|
assert "Tokens spent" in rendered
|
|
assert "12345" in rendered
|
|
assert "Cache hits" in rendered
|
|
assert "456" in rendered
|
|
assert "Baseline tokens" in rendered
|
|
assert "10000 (81.0%)" in rendered
|
|
assert "Escalated tokens" in rendered
|
|
assert "2345 (19.0%)" in rendered
|
|
assert "Largest query" in rendered
|
|
assert "6789" in rendered
|
|
assert "LLM calls" in rendered
|
|
assert "4" in rendered
|
|
assert "LLM wait" in rendered
|
|
assert "1m 31s" in rendered
|
|
assert "Total time" in rendered
|
|
assert "2m 5s" in rendered
|
|
assert "Files modified" in rendered
|
|
assert "Files added" in rendered
|
|
|
|
|
|
def test_format_token_count_switches_to_padded_scientific_notation() -> None:
|
|
assert _format_token_count(14026) == " 14026"
|
|
assert _format_token_count(27000000) == "2.70000E7 "
|
|
|
|
|
|
def test_token_bar_scales_to_two_x_budget() -> None:
|
|
assert _token_bar(0, 100) == "[▏▏▏▏▏▏▏▏▏▏▏▏▏▏▏▏▏▏▏▏]"
|
|
assert _token_bar(100, 100) == "[░░░░▒▒▒▒▒▒▏▏▏▏▏▏▏▏▏▏]"
|
|
assert _token_bar(120, 100) == "[░░░░▒▒▒▒▒▒▓▓▏▏▏▏▏▏▏▏]"
|
|
assert _token_bar(200, 100) == "[░░░░▒▒▒▒▒▒▓▓▓▓▓▓▓▓▓▓]"
|
|
|
|
|
|
def test_token_bar_colors_over_budget_region() -> None:
|
|
"""Verify threshold segments share one computed color attribute per region."""
|
|
|
|
yellow_segments = _token_bar_segments(120, 100, 20)
|
|
red_segments = _token_bar_segments(200, 100, 20)
|
|
yellow_attr = yellow_segments[9][1]
|
|
red_attr = red_segments[9][1]
|
|
|
|
assert all(attr == curses.A_NORMAL for _, attr in yellow_segments[:9])
|
|
assert yellow_segments[9] == ("▒", yellow_attr)
|
|
assert yellow_segments[10] == ("▓", yellow_attr)
|
|
assert yellow_segments[11] == ("▓", yellow_attr)
|
|
assert all(attr == curses.A_DIM for _, attr in yellow_segments[12:])
|
|
|
|
assert all(attr == curses.A_NORMAL for _, attr in red_segments[:9])
|
|
assert all(attr == red_attr for _, attr in red_segments[9:])
|
|
|
|
|
|
def test_safe_color_pair_returns_a_curses_attribute_without_terminal_setup() -> None:
|
|
"""Verify the safe color boundary always returns a usable integer attribute."""
|
|
|
|
assert isinstance(_safe_color_pair(1), int)
|
|
|
|
|
|
def test_publish_cot_status_uses_original_prompt_for_header_task() -> None:
|
|
publisher = _Publisher()
|
|
|
|
class _Context:
|
|
story_cursor = StoryCursor(
|
|
cot_task_records=[CotTaskRecord("split completion client", "a.py", ClassMeta("a.py", "", [], {})), CotTaskRecord("split cycle jacker", "b.py", ClassMeta("b.py", "", [], {}))],
|
|
active_story="",
|
|
task_cursor=TaskCursor(current_cot_index=1),
|
|
)
|
|
original_prompt = "Refactor the ai_client package into one class per file."
|
|
goal = "Refined recursive pass prompt"
|
|
|
|
with status_publisher_scope(publisher):
|
|
publish_cot_status(_Context())
|
|
|
|
status_events = [event for event in publisher.events if isinstance(event, SupremeStatusDTO)]
|
|
assert status_events
|
|
assert status_events[-1].project is None
|
|
assert status_events[-1].task == "Refactor the ai_client package into one class per file."
|
|
assert status_events[-1].cot is not None
|
|
assert status_events[-1].cot.all_tasks == ("split completion client", "split cycle jacker")
|
|
assert status_events[-1].cot.all_tasks[status_events[-1].cot.current_index - 1] == "split cycle jacker"
|
|
|
|
|
|
def test_render_cot_lines_includes_entire_work_tree() -> None:
|
|
cot = CotStatusDTO(
|
|
current_index=3,
|
|
total_count=5,
|
|
all_tasks=("one", "two", "three", "four", "five"),
|
|
completed_indices=(1, 2),
|
|
)
|
|
|
|
assert render_cot_lines(cot) == [
|
|
"1 x one",
|
|
"2 x two",
|
|
"3 > three",
|
|
"4 . four",
|
|
"5 . five",
|
|
]
|
|
|
|
|
|
def test_wrap_panel_lines_wraps_long_lines() -> None:
|
|
lines = _wrap_panel_lines(
|
|
["Refactor source files under band_aider toward one class per file for easier review and maintenance."],
|
|
width=24,
|
|
)
|
|
|
|
assert len(lines) > 2
|
|
assert lines[0] == "Refactor source files"
|
|
assert all(len(line) <= 24 for line in lines)
|
|
|
|
|
|
def test_visible_scrolled_lines_clamps_to_available_range() -> None:
|
|
ui = TerminalUI(_Queue(), _Queue())
|
|
ui._scroll_offsets["header"] = 10
|
|
|
|
visible = ui._visible_scrolled_lines("header", ["one", "two", "three", "four"], visible_rows=2)
|
|
|
|
assert visible == ["three", "four"]
|
|
assert ui._scroll_offsets["header"] == 2
|
|
assert ui._scroll_limits["header"] == 2
|
|
|
|
|
|
def test_log_scroll_key_updates_offset_within_range() -> None:
|
|
ui = TerminalUI(_Queue(), _Queue())
|
|
ui._set_focus("logs")
|
|
ui._scroll_limits["logs"] = 3
|
|
|
|
handled = ui._handle_scroll_key(258)
|
|
|
|
assert handled is True
|
|
assert ui._scroll_offsets["logs"] == 1
|
|
|
|
|
|
def test_logs_follow_tail_by_default() -> None:
|
|
ui = TerminalUI(_Queue(), _Queue())
|
|
|
|
visible = ui._visible_scrolled_lines("logs", ["one", "two", "three", "four"], visible_rows=2)
|
|
|
|
assert visible == ["three", "four"]
|
|
assert ui._scroll_offsets["logs"] == 2
|
|
assert ui._scroll_limits["logs"] == 2
|
|
|
|
|
|
def test_logs_stop_following_tail_when_scrolled_up() -> None:
|
|
ui = TerminalUI(_Queue(), _Queue())
|
|
ui._set_focus("logs")
|
|
ui._scroll_limits["logs"] = 3
|
|
ui._scroll_offsets["logs"] = 2
|
|
ui._follow_tail["logs"] = True
|
|
|
|
ui._handle_scroll_key(259)
|
|
visible = ui._visible_scrolled_lines("logs", ["one", "two", "three", "four", "five"], visible_rows=2)
|
|
|
|
assert ui._follow_tail["logs"] is False
|
|
assert visible == ["two", "three"]
|
|
assert ui._scroll_offsets["logs"] == 1
|
|
|
|
|
|
def test_cot_view_centers_active_task_by_default() -> None:
|
|
ui = TerminalUI(_Queue(), _Queue())
|
|
cot = CotStatusDTO(
|
|
current_index=6,
|
|
total_count=9,
|
|
all_tasks=tuple(f"task {index}" for index in range(1, 10)),
|
|
completed_indices=(1, 2, 3, 4, 5),
|
|
)
|
|
|
|
visible = ui._visible_cot_lines(cot, width=40, visible_rows=3)
|
|
|
|
assert visible == ["5 x task 5", "6 > task 6", "7 . task 7"]
|
|
assert ui._scroll_offsets["cot"] == 4
|
|
|
|
|
|
def test_cot_manual_scroll_disables_follow_until_task_changes() -> None:
|
|
ui = TerminalUI(_Queue(), _Queue())
|
|
ui._set_focus("cot")
|
|
ui._scroll_limits["cot"] = 8
|
|
ui._scroll_offsets["cot"] = 4
|
|
ui._handle_scroll_key(curses.KEY_UP)
|
|
cot = CotStatusDTO(
|
|
current_index=6,
|
|
total_count=9,
|
|
all_tasks=tuple(f"task {index}" for index in range(1, 10)),
|
|
completed_indices=(1, 2, 3, 4, 5),
|
|
)
|
|
|
|
visible = ui._visible_cot_lines(cot, width=40, visible_rows=3)
|
|
assert visible == ["4 x task 4", "5 x task 5", "6 > task 6"]
|
|
|
|
ui._sync_cot_follow_state(
|
|
CotStatusDTO(
|
|
current_index=7,
|
|
total_count=9,
|
|
all_tasks=tuple(f"task {index}" for index in range(1, 10)),
|
|
completed_indices=(1, 2, 3, 4, 5, 6),
|
|
)
|
|
)
|
|
visible = ui._visible_cot_lines(
|
|
CotStatusDTO(
|
|
current_index=7,
|
|
total_count=9,
|
|
all_tasks=tuple(f"task {index}" for index in range(1, 10)),
|
|
completed_indices=(1, 2, 3, 4, 5, 6),
|
|
),
|
|
width=40,
|
|
visible_rows=3,
|
|
)
|
|
|
|
assert visible == ["6 x task 6", "7 > task 7", "8 . task 8"]
|
|
|
|
|
|
def test_terminal_projection_tracks_loop_task_signal_for_cot_cursor() -> None:
|
|
ui = TerminalUI(_Queue(), _Queue())
|
|
ui.projection.apply(
|
|
SupremeStatusDTO(
|
|
cot=CotStatusDTO(
|
|
current_index=1,
|
|
total_count=3,
|
|
all_tasks=("task 1", "task 2", "task 3"),
|
|
completed_indices=(1,),
|
|
),
|
|
loop=LoopStatusDTO(
|
|
previous_loop="SupremeLoop",
|
|
current_loop="StabilizationLoop",
|
|
next_loop="BetaLoop",
|
|
current_item="task 2",
|
|
),
|
|
)
|
|
)
|
|
|
|
rendered = render_projection(ui.projection)
|
|
|
|
assert ui.projection.current_cot_item == "task 2"
|
|
assert "2 > task 2" in rendered
|
|
|
|
|
|
def test_render_projection_uses_panel_title() -> None:
|
|
rendered = render_projection(TerminalUI(_Queue(), _Queue()).projection)
|
|
|
|
assert rendered.splitlines()[0] == DEFAULT_PANEL_TITLE
|
|
|
|
|
|
def test_terminal_ui_enters_intervention_mode_with_hint_focus() -> None:
|
|
request = _intervention_request()
|
|
ui = TerminalUI(_Queue([request]), _Queue())
|
|
|
|
ui._drain_events()
|
|
|
|
assert ui.projection.intervention == request
|
|
assert ui._escalation_state.focused_field() == "hint"
|
|
assert "human guidance required" in ui._message
|
|
|
|
|
|
def test_terminal_ui_exits_intervention_mode_after_retry_command() -> None:
|
|
command_queue = _Queue()
|
|
ui = TerminalUI(_Queue(), command_queue)
|
|
ui._enter_intervention_mode(_intervention_request())
|
|
|
|
ui._handle_intervention_key(27)
|
|
|
|
assert ui.projection.intervention is None
|
|
assert ui._escalation_state.focused_field() == "request"
|
|
assert len(command_queue.items) == 1
|
|
response = command_queue.items[0]
|
|
assert isinstance(response, HumanInterventionResponseDTO)
|
|
assert response.action == "retry"
|
|
assert ui._message == "skipping hint"
|
|
|
|
|
|
def test_terminal_f9_toggles_intervention_and_dismisses_an_open_request() -> None:
|
|
"""F9 exposes the same suppression choice from the dashboard and a blocked human screen."""
|
|
|
|
commands = _Queue()
|
|
ui = TerminalUI(_Queue(), commands)
|
|
ui._handle_key(_MemoryScreen(keys=[curses.KEY_F9]), worker=None)
|
|
ui._handle_key(_MemoryScreen(keys=[curses.KEY_F9]), worker=None)
|
|
|
|
assert commands.items == [InterventionSuppressionCommandDTO(True), InterventionSuppressionCommandDTO(False)]
|
|
assert "enabled" in ui._message
|
|
|
|
request = _intervention_request()
|
|
ui._enter_intervention_mode(request)
|
|
ui._handle_intervention_key(curses.KEY_F9)
|
|
|
|
assert ui.projection.intervention is None
|
|
response = commands.items[-1]
|
|
assert isinstance(response, HumanInterventionResponseDTO)
|
|
assert response.action == "suppress"
|
|
assert response.request_id == request.request_id
|
|
assert "suppressed" in ui._message
|
|
|
|
|
|
def test_terminal_ui_f5_requests_cycle_interrupt() -> None:
|
|
"""Blank selectors send the existing next-interaction request and await acknowledgement."""
|
|
command_queue = _Queue()
|
|
ui = TerminalUI(_Queue(), command_queue)
|
|
|
|
ui._handle_key(_MemoryScreen(keys=[curses.KEY_F5]), worker=None)
|
|
|
|
assert command_queue.items == [CycleInterruptCommandDTO()]
|
|
assert "waiting for worker acknowledgement" in ui._message
|
|
|
|
ui.projection.completion = RunCompletionDTO(
|
|
final_summary="No report.",
|
|
report_path=None,
|
|
total_tokens_spent=0,
|
|
largest_query_tokens=0,
|
|
total_seconds=0,
|
|
average_seconds_per_cot=0,
|
|
cot_count=0,
|
|
files_modified=0,
|
|
files_added=0,
|
|
)
|
|
ui._completion_focus = 2
|
|
ui._handle_completion_key(10)
|
|
|
|
assert ui._message == "no report path was published"
|
|
assert ui._completion_action != "report"
|
|
|
|
ui.projection.completion = RunCompletionDTO(
|
|
final_summary="Report exists.",
|
|
report_path="reference-only-report.md",
|
|
total_tokens_spent=0,
|
|
largest_query_tokens=0,
|
|
total_seconds=0,
|
|
average_seconds_per_cot=0,
|
|
cot_count=0,
|
|
files_modified=0,
|
|
files_added=0,
|
|
)
|
|
ui._completion_focus = 2
|
|
ui._handle_completion_key(10)
|
|
|
|
assert ui._completion_action == "report"
|
|
|
|
|
|
def test_terminal_ui_draw_switches_to_escalation_screen() -> None:
|
|
"""Verify escalation mode draws the actual review panels into memory."""
|
|
|
|
request = _intervention_request()
|
|
ui = TerminalUI(_Queue(), _Queue())
|
|
ui._enter_intervention_mode(request)
|
|
screen = _MemoryScreen(width=100)
|
|
|
|
ui._draw(screen, worker=None)
|
|
|
|
assert " REQUEST " in screen.text
|
|
assert " RESPONSE " in screen.text
|
|
assert request.original_prompt in screen.text
|
|
|
|
|
|
def test_terminal_ui_draw_returns_to_main_screen_after_intervention() -> None:
|
|
"""Verify leaving intervention mode restores the ordinary terminal panels."""
|
|
|
|
ui = TerminalUI(_Queue(), _Queue())
|
|
ui._enter_intervention_mode(_intervention_request())
|
|
ui._exit_intervention_mode("done")
|
|
screen = _MemoryScreen(width=100)
|
|
|
|
ui._draw(screen, worker=None)
|
|
|
|
assert " BAND-AIDER " in screen.text
|
|
assert " REQUEST " not in screen.text
|
|
|
|
|
|
def test_terminal_ui_draw_splits_backlog_and_queue_depth_boxes() -> None:
|
|
"""Verify the real draw path creates independent backlog and queue panels."""
|
|
|
|
ui = TerminalUI(_Queue(), _Queue())
|
|
screen = _MemoryScreen()
|
|
|
|
ui._draw(screen, worker=None)
|
|
|
|
assert " BACKLOG " in screen.text
|
|
assert " QUEUE DEPTH " in screen.text
|
|
|
|
|
|
def test_terminal_ui_top_right_panels_align_with_token_budget_width() -> None:
|
|
"""Verify adjacent top panels span the same width as the token panel."""
|
|
|
|
ui = TerminalUI(_Queue(), _Queue())
|
|
_, token_width = ui._stabilization_column_widths(120)
|
|
backlog_width, queue_depth_width = ui._top_status_panel_widths(token_width)
|
|
|
|
assert backlog_width + queue_depth_width - 1 == token_width
|
|
|
|
|
|
def test_escalation_ui_tab_cycles_focus() -> None:
|
|
state = EscalationUIState()
|
|
request = _intervention_request()
|
|
|
|
_, message = handle_escalation_key(9, request, state)
|
|
assert state.focused_field() == "response"
|
|
assert message == "selected response"
|
|
|
|
_, message = handle_escalation_key(9, request, state)
|
|
assert state.focused_field() == "hint"
|
|
assert message == "selected hint"
|
|
|
|
|
|
def test_escalation_ui_scrolls_focused_panel() -> None:
|
|
state = EscalationUIState()
|
|
request = _intervention_request()
|
|
state.scroll_limits["request"] = 3
|
|
|
|
_, message = handle_escalation_key(curses.KEY_DOWN, request, state)
|
|
|
|
assert state.scroll_offsets["request"] == 1
|
|
assert message == "request scroll 1/3"
|
|
|
|
|
|
def test_escalation_ui_stacks_review_panes_on_narrow_terminals() -> None:
|
|
request_layout, response_layout = _escalation_pane_layout(pane_y=7, body_height=10, width=100)
|
|
|
|
assert request_layout.x == 0
|
|
assert request_layout.width == 100
|
|
assert response_layout.x == 0
|
|
assert response_layout.width == 100
|
|
assert response_layout.y == request_layout.y + request_layout.height - 1
|
|
|
|
|
|
def test_escalation_ui_keeps_review_panes_as_columns_on_wide_terminals() -> None:
|
|
request_layout, response_layout = _escalation_pane_layout(pane_y=7, body_height=10, width=140)
|
|
|
|
assert request_layout.x == 0
|
|
assert response_layout.x == request_layout.width - 1
|
|
assert request_layout.width > 60
|
|
assert response_layout.width > 60
|
|
|
|
|
|
def test_escalation_ui_header_text_does_not_overlap_when_wrapped() -> None:
|
|
class _Screen:
|
|
def __init__(self) -> None:
|
|
self.writes: list[tuple[int, int, str]] = []
|
|
|
|
def getmaxyx(self) -> tuple[int, int]:
|
|
return 7, 100
|
|
|
|
def addstr(self, y, x, text, attr=0) -> None:
|
|
self.writes.append((y, x, text))
|
|
|
|
screen = _Screen()
|
|
|
|
_draw_header_content(screen, header_height=7, width=100, escalation=_escalation_event())
|
|
|
|
text_writes = [(y, x, text) for y, x, text in screen.writes if text.strip()]
|
|
for index, (left_y, left_x, left_text) in enumerate(text_writes):
|
|
left_range = range(left_x, left_x + len(left_text))
|
|
for right_y, right_x, right_text in text_writes[index + 1 :]:
|
|
right_range = range(right_x, right_x + len(right_text))
|
|
assert (
|
|
left_y != right_y
|
|
or left_range.stop <= right_range.start
|
|
or right_range.stop <= left_range.start
|
|
)
|
|
|
|
|
|
def test_escalation_ui_accepts_multiline_hint_and_sends_with_f8() -> None:
|
|
state = EscalationUIState(guidance="first line", focus_index=2)
|
|
request = _intervention_request()
|
|
|
|
response, message = handle_escalation_key(10, request, state)
|
|
assert response is None
|
|
assert message is None
|
|
assert state.guidance == "first line\n"
|
|
|
|
response, message = handle_escalation_key(ord("x"), request, state)
|
|
assert response is None
|
|
assert state.guidance == "first line\nx"
|
|
|
|
response, message = handle_escalation_key(SEND_HINT_KEY, request, state)
|
|
assert response is not None
|
|
assert response.action == "resend"
|
|
assert response.guidance == "first line\nx"
|
|
assert message == "human response sent"
|
|
|
|
|
|
def test_escalation_ui_escape_retries_without_hint_and_ctrl_c_aborts() -> None:
|
|
request = _intervention_request()
|
|
|
|
response, message = handle_escalation_key(27, request, EscalationUIState())
|
|
assert response is not None
|
|
assert response.action == "retry"
|
|
assert message == "skipping hint"
|
|
|
|
response, message = handle_escalation_key(3, request, EscalationUIState())
|
|
assert response is not None
|
|
assert response.action == "abort"
|
|
assert message == "intervention aborted"
|
|
|
|
|
|
def test_render_plain_escalation_prompt_uses_help_copy() -> None:
|
|
rendered = render_plain_escalation_prompt(_intervention_request(), None)
|
|
|
|
assert rendered.splitlines()[0] == "HELP!"
|
|
assert HELP_SUBTITLE in rendered
|
|
assert "REQUEST" in rendered
|
|
assert "RESPONSE" in rendered
|
|
|
|
|
|
def test_escalation_curses_panes_receive_formatted_json_text() -> None:
|
|
"""Verify the real in-memory curses path renders structured request evidence."""
|
|
|
|
request = HumanInterventionRequestDTO(
|
|
request_id="req-1",
|
|
check_name="CycleJacker",
|
|
original_prompt='Prompt prose {"prompt": true}',
|
|
repeated_response_tail='before {"root":{"value":42}} after',
|
|
request_md5="a" * 32,
|
|
response_md5="b" * 32,
|
|
local_model="local",
|
|
escalated_model="cloud-2",
|
|
)
|
|
screen = _MemoryScreen()
|
|
|
|
draw_escalation_screen(screen, 24, 120, request, None, EscalationUIState())
|
|
|
|
response_text = screen.text
|
|
request_text = screen.text
|
|
assert "# root" in response_text
|
|
assert "## root value" in response_text
|
|
assert '{"root"' not in response_text
|
|
assert "Prompt prose" in request_text
|
|
assert "# prompt" in request_text
|
|
assert '{"prompt": true}' not in request_text
|
|
assert request.original_prompt == 'Prompt prose {"prompt": true}'
|
|
assert request.repeated_response_tail == 'before {"root":{"value":42}} after'
|
|
|
|
|
|
def test_plain_escalation_prompt_receives_formatted_request_and_response_text() -> None:
|
|
request = HumanInterventionRequestDTO(
|
|
request_id="req-1",
|
|
check_name="CycleJacker",
|
|
original_prompt='Prompt prose {"prompt": true}',
|
|
repeated_response_tail='before {"root":{"value":42}} after',
|
|
request_md5="a" * 32,
|
|
response_md5="b" * 32,
|
|
local_model="local",
|
|
escalated_model="cloud-2",
|
|
)
|
|
|
|
rendered = render_plain_escalation_prompt(request, None)
|
|
|
|
assert "Prompt prose" in rendered
|
|
assert "# prompt" in rendered
|
|
assert '{"prompt": true}' not in rendered
|
|
assert "# root" in rendered
|
|
assert "## root value" in rendered
|
|
assert '{"root"' not in rendered
|
|
assert request.original_prompt == 'Prompt prose {"prompt": true}'
|
|
assert request.repeated_response_tail == 'before {"root":{"value":42}} after'
|
|
|
|
|
|
def test_both_escalation_render_paths_use_json2md_escalator() -> None:
|
|
"""Verify plain and curses rendering expose the same formatted JSON fragments."""
|
|
|
|
request = HumanInterventionRequestDTO(
|
|
request_id="req-1",
|
|
check_name="CycleJacker",
|
|
original_prompt='Prompt prose {"prompt": true}',
|
|
repeated_response_tail='before {"root":{"value":42}} after',
|
|
request_md5="a" * 32,
|
|
response_md5="b" * 32,
|
|
local_model="local",
|
|
escalated_model="cloud-2",
|
|
)
|
|
screen = _MemoryScreen()
|
|
|
|
draw_escalation_screen(screen, 24, 120, request, None, EscalationUIState())
|
|
plain = render_plain_escalation_prompt(request, None)
|
|
|
|
assert "# prompt" in screen.text
|
|
assert "# root" in screen.text
|
|
assert "# prompt" in plain
|
|
assert "# root" in plain
|
|
|
|
|
|
def test_escalation_interaction_behavior_remains_unchanged() -> None:
|
|
request = _intervention_request()
|
|
response, message = handle_escalation_key(27, request, EscalationUIState())
|
|
|
|
assert response is not None
|
|
assert response.action == "retry"
|
|
assert response.guidance == ""
|
|
assert message == "skipping hint"
|
|
|
|
|
|
def test_terminal_selectors_send_exact_target_and_keep_log_shortcuts() -> None:
|
|
"""Keyboard selection sends a resolved one-shot target without taking over log controls."""
|
|
commands = _Queue()
|
|
ui = TerminalUI(_Queue(), commands)
|
|
ui._intervention_selection.files = ("src/widget.py", "tests/widget.py")
|
|
ui._handle_key(_MemoryScreen(keys=[curses.KEY_F2]), worker=None)
|
|
assert ui._focused_field() == "decision"
|
|
# A short prefix selects the concrete owner rather than a transport wrapper.
|
|
for key in map(ord, "Intent"):
|
|
ui._handle_key(_MemoryScreen(keys=[key]), worker=None)
|
|
ui._handle_key(_MemoryScreen(keys=[curses.KEY_ENTER]), worker=None)
|
|
ui._handle_key(_MemoryScreen(keys=[9]), worker=None)
|
|
for key in map(ord, "src/"):
|
|
ui._handle_key(_MemoryScreen(keys=[key]), worker=None)
|
|
ui._handle_key(_MemoryScreen(keys=[curses.KEY_ENTER]), worker=None)
|
|
ui._handle_key(_MemoryScreen(keys=[9]), worker=None)
|
|
for key in map(ord, "rep"):
|
|
ui._handle_key(_MemoryScreen(keys=[key]), worker=None)
|
|
ui._handle_key(_MemoryScreen(keys=[curses.KEY_ENTER]), worker=None)
|
|
ui._handle_key(_MemoryScreen(keys=[curses.KEY_F5]), worker=None)
|
|
assert commands.items == [CycleInterruptCommandDTO("IntentCheck", "src/widget.py", "replace")]
|
|
ui._handle_key(_MemoryScreen(keys=[curses.KEY_F3]), worker=None)
|
|
assert ui._focused_field() == "filter"
|
|
|
|
|
|
def test_target_controls_are_visible_in_small_terminal_and_preserve_active_draft() -> None:
|
|
"""A narrow dashboard exposes selectors and logs; intervention preserves the draft."""
|
|
ui = TerminalUI(_Queue(), _Queue())
|
|
ui.projection.logs.append(LogMessageDTO(1.0, "INFO", "editor", "Editing widget"))
|
|
screen = _MemoryScreen(height=24, width=80)
|
|
ui._draw(screen, worker=None)
|
|
assert "Decision/check: [Any]" in screen.text
|
|
assert "File: [Any]" in screen.text
|
|
assert "Response: [guide]" in screen.text
|
|
assert "Manual stop: not armed" in screen.text
|
|
assert "Editing widget" in screen.text
|
|
request = _intervention_request()
|
|
ui._enter_intervention_mode(request)
|
|
ui._escalation_state.guidance = "keep this unsent draft"
|
|
ui._intervention_selection.values["file"] = "src/another.py"
|
|
assert ui.projection.intervention is request
|
|
assert ui._escalation_state.guidance == "keep this unsent draft"
|