596 lines
21 KiB
Python
596 lines
21 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
"""Stationary DTO-to-text contracts for the terminal's non-interactive projection."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import curses
|
|
from multiprocessing.queues import Queue
|
|
from queue import Empty
|
|
|
|
from band_aider.ui.dto.backlog_status import BacklogStatusDTO
|
|
from band_aider.ui.dto.bootstrap_status import BootstrapStatusDTO
|
|
from band_aider.ui.dto.cot_status import CotStatusDTO
|
|
from band_aider.ui.dto.direct_code_editor_status import DirectCodeEditorStatusDTO
|
|
from band_aider.ui.dto.human_intervention import HumanInterventionRequestDTO
|
|
from band_aider.ui.dto.log_message import LogMessageDTO
|
|
from band_aider.ui.dto.logger_namespaces import LoggerNamespacesDTO
|
|
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.dto.stabilization_status import StabilizationStatusDTO
|
|
from band_aider.ui.dto.story_status import StoryStatusDTO
|
|
from band_aider.ui.dto.supreme_status import SupremeStatusDTO
|
|
from band_aider.ui.dto.token_budget_stats import TokenBudgetBucketDTO, TokenBudgetStatsDTO
|
|
from band_aider.ui.terminal_ui import (
|
|
TerminalProjection,
|
|
_completion_stat_rows,
|
|
_format_duration,
|
|
_format_percent,
|
|
_format_token_count,
|
|
_resolve_cot_status,
|
|
_token_bar_segments,
|
|
_token_split,
|
|
_wrap,
|
|
_wrap_panel_lines,
|
|
TerminalUI,
|
|
render_direct_code_editor_lines,
|
|
render_escalation_lines,
|
|
render_human_intervention,
|
|
render_projection,
|
|
render_wrapped_cot_lines,
|
|
)
|
|
|
|
|
|
class _MemoryQueue(Queue[object]):
|
|
"""Use the nominal queue boundary without allocating an operating-system queue."""
|
|
|
|
def __init__(self, values: tuple[object, ...] = ()) -> None:
|
|
"""Store a finite, stationary message sequence in process-local memory."""
|
|
|
|
self._values = list(values)
|
|
self.entries: list[object] = []
|
|
|
|
def get_nowait(self) -> object:
|
|
"""Return the next stationary event or honor Queue's nonblocking empty contract."""
|
|
|
|
if not self._values:
|
|
raise Empty
|
|
return self._values.pop(0)
|
|
|
|
def put(
|
|
self,
|
|
obj: object,
|
|
block: bool = True,
|
|
timeout: float | None = None,
|
|
) -> None:
|
|
"""Record commands in memory without communicating with a worker."""
|
|
|
|
del block, timeout
|
|
self.entries.append(obj)
|
|
|
|
|
|
def _escalation() -> ModelEscalationDTO:
|
|
"""Return one complete, stationary escalation snapshot."""
|
|
|
|
return ModelEscalationDTO(
|
|
trigger_name="StationaryRetry",
|
|
occurred_at=0,
|
|
prompt_preview="first evidence\nsecond evidence\nthird evidence\nfourth evidence",
|
|
reason="The exact fixed response repeated.",
|
|
request_md5="a" * 32,
|
|
response_md5="b" * 32,
|
|
local_model="local",
|
|
active_model="tier-two",
|
|
config_index=1,
|
|
config_count=2,
|
|
state="human_required",
|
|
repeat_count=3,
|
|
tokens_escalated=123,
|
|
active_endpoint="http://stationary.invalid",
|
|
)
|
|
|
|
|
|
def _completion(report_path: str | None = None) -> RunCompletionDTO:
|
|
"""Return final run facts without opening a report file."""
|
|
|
|
return RunCompletionDTO(
|
|
final_summary="SUCCESS stationary terminal contract.",
|
|
report_path=report_path,
|
|
total_tokens_spent=100,
|
|
largest_query_tokens=40,
|
|
total_seconds=61,
|
|
average_seconds_per_cot=30,
|
|
cot_count=2,
|
|
files_modified=1,
|
|
files_added=0,
|
|
llm_call_count=3,
|
|
llm_wait_seconds=2,
|
|
baseline_tokens_spent=75,
|
|
escalated_tokens_spent=25,
|
|
total_cache_hit_tokens=20,
|
|
)
|
|
|
|
|
|
def test_terminal_projection_applies_each_stationary_event_kind() -> None:
|
|
"""Projection state accepts DTO events and resets a stale COT selection per story."""
|
|
|
|
projection = TerminalProjection()
|
|
first_story = StoryStatusDTO(0, 2, ("first", "second"), ())
|
|
first_status = SupremeStatusDTO(
|
|
story=first_story,
|
|
cot=CotStatusDTO(1, 2, ("first task", "second task"), ()),
|
|
loop=LoopStatusDTO("SupremeLoop", "StabilizationLoop", None, "second task"),
|
|
)
|
|
intervention = HumanInterventionRequestDTO(
|
|
request_id="request-1",
|
|
check_name="StationaryRetry",
|
|
original_prompt="Keep the contract local.",
|
|
repeated_response_tail="same stationary response",
|
|
request_md5="a" * 32,
|
|
response_md5="b" * 32,
|
|
local_model="local",
|
|
escalated_model="tier-two",
|
|
)
|
|
|
|
projection.apply(first_status)
|
|
projection.apply(_escalation())
|
|
projection.apply(intervention)
|
|
projection.apply(_completion())
|
|
projection.apply(LogMessageDTO(0, "INFO", "terminal", "fixed event"))
|
|
projection.apply(SupremeStatusDTO(story=StoryStatusDTO(1, 2, ("first", "second"), (0,))))
|
|
|
|
assert projection.status.story == StoryStatusDTO(1, 2, ("first", "second"), (0,))
|
|
assert projection.current_cot_item is None
|
|
assert projection.escalation == _escalation()
|
|
assert projection.intervention == intervention
|
|
assert projection.completion == _completion()
|
|
assert [entry.message for entry in projection.logs] == ["fixed event"]
|
|
|
|
|
|
def test_terminal_projection_renders_full_stabilization_evidence_without_terminal_io() -> None:
|
|
"""A completed DTO graph projects its selected COT, queue, token, and editor facts."""
|
|
|
|
editor = DirectCodeEditorStatusDTO(
|
|
editable_path="tests/test_widget.py",
|
|
repair_turn=3,
|
|
phase="targeted test",
|
|
raw_compiler_error_count=4,
|
|
blocking_compiler_error_count=2,
|
|
model_visible_compiler_error_count=1,
|
|
static_type_check_error_count=3,
|
|
unresolved_targeted_test_failure_count=21,
|
|
named_targeted_failures=("FAIL Widget.test_a",),
|
|
unattributed_targeted_execution_failure_count=21,
|
|
patch_outcome="partial",
|
|
unresolved_patch_issue_count=2,
|
|
decision_branch="retry",
|
|
decision_reason="a target remains",
|
|
)
|
|
projection = TerminalProjection(
|
|
status=SupremeStatusDTO(
|
|
story=StoryStatusDTO(2, 2, ("first", "second"), (0, 1)),
|
|
backlog=BacklogStatusDTO(
|
|
cmp=1,
|
|
ins=2,
|
|
rem=3,
|
|
test=4,
|
|
alpha_inbound_queue=1,
|
|
alpha_outbound_queue=2,
|
|
beta_inbound_queue=3,
|
|
beta_outbound_queue=4,
|
|
final_exam_active=True,
|
|
),
|
|
cot=CotStatusDTO(1, 2, ("first task", "second task"), ()),
|
|
stabilization=StabilizationStatusDTO(
|
|
root="src/widget.py",
|
|
frontier_count=2,
|
|
visited_count=1,
|
|
files_in_play_count=3,
|
|
current="stale current",
|
|
next="next file",
|
|
),
|
|
loop=LoopStatusDTO(
|
|
"SupremeLoop",
|
|
"CustomLoop",
|
|
"BetaLoop",
|
|
"second task",
|
|
"Widget.validate",
|
|
),
|
|
token_budget=TokenBudgetStatsDTO(
|
|
max_prompt_tokens=100,
|
|
last_ai_token_used=10,
|
|
total_tokens=100,
|
|
total_cache_hit_tokens=20,
|
|
baseline_tokens=75,
|
|
escalated_tokens=25,
|
|
bucket_window_seconds=300,
|
|
buckets=(TokenBudgetBucketDTO(0, 100),),
|
|
),
|
|
direct_code_editor=editor,
|
|
),
|
|
escalation=_escalation(),
|
|
completion=_completion(),
|
|
current_cot_item="second task",
|
|
)
|
|
|
|
rendered = render_projection(projection)
|
|
|
|
assert "Story 2/2: completed" in rendered
|
|
assert "LAST EXAM DO NOT INTERRUPT" in rendered
|
|
assert "COT 2/2: second task" in rendered
|
|
assert "STABILIZATION (CUSTOMLOOP)" in rendered
|
|
assert "Focus Widget.validate" in rendered
|
|
assert "Baseline 75 75.0%" in rendered
|
|
assert "MODEL ESCALATION" in rendered
|
|
assert "DIRECT CODE EDITOR" in rendered
|
|
assert "UNKNOWN 1 additional targeted results" in rendered
|
|
assert "RUN COMPLETE" in rendered
|
|
|
|
|
|
def test_terminal_projection_renders_bootstrap_instead_of_stabilization_and_logs_without_editor() -> None:
|
|
"""Bootstrap is the active presentation owner and ordinary logs return when no editor is live."""
|
|
|
|
projection = TerminalProjection(
|
|
status=SupremeStatusDTO(
|
|
task="stationary task",
|
|
bootstrap=BootstrapStatusDTO(
|
|
stage="mapping",
|
|
stage_label="mapping",
|
|
iteration=1,
|
|
max_iterations=2,
|
|
visible_count=3,
|
|
beat_count=4,
|
|
concern_count=5,
|
|
current=None,
|
|
next=None,
|
|
),
|
|
),
|
|
)
|
|
projection.apply(LogMessageDTO(0, "WARN", "bootstrap", "fixed warning"))
|
|
|
|
rendered = render_projection(projection)
|
|
|
|
assert "Task stationary task" in rendered
|
|
assert "BOOTSTRAP (mapping)" in rendered
|
|
assert "STABILIZATION" not in rendered
|
|
assert "LOGS" in rendered
|
|
assert "fixed warning" in rendered
|
|
|
|
|
|
def test_terminal_render_helpers_cover_empty_and_optional_evidence_paths() -> None:
|
|
"""Formatting helpers fail closed for empty input and expose every supplied optional datum."""
|
|
|
|
cot = CotStatusDTO(1, 2, ("first task", "second task"), ())
|
|
status = SupremeStatusDTO(cot=cot)
|
|
direct = DirectCodeEditorStatusDTO(
|
|
editable_path="src/widget.py",
|
|
repair_turn=1,
|
|
phase="start",
|
|
raw_compiler_error_count=0,
|
|
blocking_compiler_error_count=0,
|
|
model_visible_compiler_error_count=0,
|
|
static_type_check_error_count=0,
|
|
unresolved_targeted_test_failure_count=0,
|
|
)
|
|
|
|
assert _resolve_cot_status(status, "unknown") == cot
|
|
assert _resolve_cot_status(status, "second task") == CotStatusDTO(
|
|
2,
|
|
2,
|
|
("first task", "second task"),
|
|
(),
|
|
)
|
|
assert render_wrapped_cot_lines(cot, 0) == ([], 0)
|
|
assert _completion_stat_rows(None)[0] == ("Tokens spent", "0")
|
|
assert "Report:" not in "\n".join(render_projection(TerminalProjection(completion=_completion())))
|
|
assert render_direct_code_editor_lines(direct)[-1] == "Targeted failures 0"
|
|
assert "Endpoint http://stationary.invalid" in render_escalation_lines(_escalation())
|
|
assert "HELP!" in render_human_intervention(
|
|
HumanInterventionRequestDTO(
|
|
request_id="request-2",
|
|
check_name="StationaryRetry",
|
|
original_prompt="Local prompt.",
|
|
repeated_response_tail="local response",
|
|
request_md5="a" * 32,
|
|
response_md5="b" * 32,
|
|
local_model="local",
|
|
escalated_model="tier-two",
|
|
),
|
|
)
|
|
assert _token_split(100, 0, 0) == (100, 0)
|
|
assert _token_split(100, 1, 25) == (75, 25)
|
|
assert _format_token_count(-1).strip() == "0"
|
|
assert _format_percent(-1, 0) == "0.0%"
|
|
assert _token_bar_segments(0, 0, 0) == []
|
|
assert _wrap("", 20) == [""]
|
|
assert _wrap("content", 0) == []
|
|
assert _wrap_panel_lines([], 20) == [""]
|
|
assert _wrap_panel_lines(["content"], 0) == []
|
|
assert (_format_duration(0), _format_duration(61), _format_duration(3_661)) == (
|
|
"0s",
|
|
"1m 1s",
|
|
"1h 1m 1s",
|
|
)
|
|
|
|
|
|
def test_terminal_ui_drains_nominal_memory_events_and_restores_known_logger_choices() -> None:
|
|
"""DTO delivery updates only UI-local projection state and never starts a queue worker."""
|
|
|
|
editor = DirectCodeEditorStatusDTO(
|
|
editable_path="src/widget.py",
|
|
repair_turn=1,
|
|
phase="start",
|
|
raw_compiler_error_count=0,
|
|
blocking_compiler_error_count=0,
|
|
model_visible_compiler_error_count=0,
|
|
static_type_check_error_count=0,
|
|
unresolved_targeted_test_failure_count=0,
|
|
)
|
|
intervention = HumanInterventionRequestDTO(
|
|
request_id="request-3",
|
|
check_name="StationaryRetry",
|
|
original_prompt="Keep local state.",
|
|
repeated_response_tail="same response",
|
|
request_md5="a" * 32,
|
|
response_md5="b" * 32,
|
|
local_model="local",
|
|
escalated_model="tier-two",
|
|
)
|
|
events = _MemoryQueue(
|
|
(
|
|
LoggerNamespacesDTO(("zeta", "alpha", "zeta")),
|
|
SupremeStatusDTO(direct_code_editor=editor),
|
|
intervention,
|
|
),
|
|
)
|
|
ui = TerminalUI(events, _MemoryQueue())
|
|
ui._logger_filter = "stale"
|
|
ui._set_focus("filter")
|
|
|
|
ui._drain_events()
|
|
|
|
assert ui._known_loggers == ["ALL", "alpha", "zeta"]
|
|
assert ui._logger_filter == "ALL"
|
|
assert ui._direct_editor_progress_visible() is True
|
|
assert ui._focused_field() == "editor"
|
|
assert ui.projection.intervention == intervention
|
|
assert ui._escalation_state.focused_field() == "hint"
|
|
|
|
|
|
def test_terminal_ui_edits_each_log_control_through_its_nominal_focus() -> None:
|
|
"""Focused log controls normalize text, lists, levels, and bounded numeric windows."""
|
|
|
|
ui = TerminalUI(_MemoryQueue(), _MemoryQueue())
|
|
ui._known_loggers = ["ALL", "alpha"]
|
|
|
|
ui._set_focus("filter")
|
|
assert ui._handle_log_control_key(ord("x")) is True
|
|
assert ui._filter_text == "x"
|
|
assert ui._handle_log_control_key(curses.KEY_BACKSPACE) is True
|
|
assert ui._filter_text == ""
|
|
|
|
ui._set_focus("logger")
|
|
assert ui._handle_log_control_key(ord("a")) is True
|
|
assert ui._logger_filter == "a"
|
|
assert ui._handle_log_control_key(curses.KEY_ENTER) is True
|
|
assert ui._logger_filter == "alpha"
|
|
assert ui._handle_log_control_key(curses.KEY_ENTER) is True
|
|
assert ui._logger_filter == "ALL"
|
|
|
|
ui._set_focus("level")
|
|
assert ui._handle_log_control_key(curses.KEY_ENTER) is True
|
|
assert ui._level_filter == "INFO"
|
|
assert ui._handle_log_control_key(27) is True
|
|
assert ui._level_filter == "ALL"
|
|
|
|
ui._set_focus("before")
|
|
ui._context_before = "99"
|
|
assert ui._handle_log_control_key(ord("9")) is True
|
|
assert ui._context_before == "999"
|
|
assert ui._handle_log_control_key(ord("x")) is True
|
|
assert ui._context_before == "999"
|
|
|
|
ui._set_focus("after")
|
|
assert ui._handle_log_control_key(curses.KEY_BACKSPACE) is True
|
|
assert ui._context_after == "1"
|
|
assert ui._handle_log_control_key(27) is True
|
|
assert ui._context_after == "0"
|
|
|
|
|
|
def test_terminal_ui_preserves_focus_and_scroll_ownership_across_view_transitions() -> None:
|
|
"""Manual scroll, direct-editor toggles, and changed COT/story keys reset only owned view state."""
|
|
|
|
ui = TerminalUI(_MemoryQueue(), _MemoryQueue())
|
|
cot = CotStatusDTO(2, 3, ("one", "two", "three"), (1,))
|
|
ui._set_focus("logs")
|
|
ui._scroll_limits["logs"] = 2
|
|
|
|
assert ui._handle_scroll_key(curses.KEY_UP) is True
|
|
assert ui._follow_tail["logs"] is False
|
|
ui._set_focus("filter")
|
|
assert ui._handle_scroll_key(curses.KEY_DOWN) is False
|
|
|
|
ui._sync_cot_follow_state(SupremeStatusDTO(cot=cot))
|
|
assert ui._follow_current["cot"] is True
|
|
ui._set_focus("cot")
|
|
assert ui._handle_scroll_key(curses.KEY_DOWN) is True
|
|
assert ui._follow_current["cot"] is False
|
|
ui._sync_cot_follow_state(SupremeStatusDTO())
|
|
assert ui._last_cot_key is None
|
|
assert ui._follow_current["cot"] is True
|
|
|
|
ui.projection.status = SupremeStatusDTO(story=StoryStatusDTO(0, 1, ("only",), ()))
|
|
ui._scroll_offsets["header"] = 3
|
|
ui._scroll_offsets["cot"] = 2
|
|
ui._sync_story_view_state()
|
|
assert (ui._scroll_offsets["header"], ui._scroll_offsets["cot"]) == (0, 0)
|
|
|
|
direct = DirectCodeEditorStatusDTO(
|
|
editable_path="src/widget.py",
|
|
repair_turn=2,
|
|
phase="repair",
|
|
raw_compiler_error_count=1,
|
|
blocking_compiler_error_count=1,
|
|
model_visible_compiler_error_count=1,
|
|
static_type_check_error_count=0,
|
|
unresolved_targeted_test_failure_count=0,
|
|
)
|
|
ui._set_focus("logs")
|
|
ui._sync_direct_editor_mode(SupremeStatusDTO(direct_code_editor=direct))
|
|
assert ui._focused_field() == "editor"
|
|
ui._toggle_direct_editor_view()
|
|
assert ui._focused_field() == "logs"
|
|
ui._toggle_direct_editor_view()
|
|
assert ui._focused_field() == "editor"
|
|
ui._sync_direct_editor_mode(SupremeStatusDTO())
|
|
assert ui._focused_field() == "logs"
|
|
|
|
|
|
def test_terminal_ui_completion_keys_require_a_published_report_for_report_actions() -> None:
|
|
"""Completion navigation remains local and refuses report actions without a report path."""
|
|
|
|
ui = TerminalUI(_MemoryQueue(), _MemoryQueue())
|
|
|
|
assert ui._handle_completion_key(9) is False
|
|
assert ui._completion_focus == 1
|
|
assert ui._handle_completion_key(ord("v")) is False
|
|
assert ui._message == "no report path was published"
|
|
ui._completion_focus = 2
|
|
assert ui._handle_completion_key(curses.KEY_ENTER) is False
|
|
assert ui._completion_action == ""
|
|
assert ui._handle_completion_key(ord("q")) is True
|
|
assert ui._completion_action == "quit"
|
|
|
|
ui._completion_action = ""
|
|
ui.projection.completion = _completion("reference-only-report.md")
|
|
assert ui._handle_completion_key(ord("v")) is True
|
|
assert ui._completion_action == "vim"
|
|
ui._completion_action = ""
|
|
ui._completion_focus = 2
|
|
assert ui._handle_completion_key(curses.KEY_ENTER) is False
|
|
assert ui._completion_action == "report"
|
|
|
|
|
|
def test_terminal_ui_filters_log_context_and_phase_without_terminal_or_worker_io() -> None:
|
|
"""Log windows merge relevant evidence while phase labels remain deterministic DTO projections."""
|
|
|
|
ui = TerminalUI(_MemoryQueue(), _MemoryQueue())
|
|
ui.projection.logs.extend(
|
|
(
|
|
LogMessageDTO(0, "INFO", "alpha.owner", "first evidence"),
|
|
LogMessageDTO(1, "WARN", "beta.owner", "target evidence"),
|
|
LogMessageDTO(2, "ERROR", "beta.owner", "third evidence"),
|
|
LogMessageDTO(3, "WARN", "beta.owner", "target follow-up"),
|
|
),
|
|
)
|
|
|
|
assert [item.message for item in ui._visible_logs(2)] == [
|
|
"third evidence",
|
|
"target follow-up",
|
|
]
|
|
|
|
ui._filter_text = "target"
|
|
ui._context_before = "0"
|
|
ui._context_after = "0"
|
|
assert [item.message for item in ui._visible_logs(10)] == [
|
|
"target evidence",
|
|
"target follow-up",
|
|
]
|
|
|
|
ui._filter_text = "not present"
|
|
assert [item.message for item in ui._visible_logs(1)] == ["target follow-up"]
|
|
|
|
ui._logger_filter = "beta"
|
|
ui._level_filter = "WARN"
|
|
ui._filter_text = "target"
|
|
assert [item.message for item in ui._visible_logs(10)] == [
|
|
"target evidence",
|
|
"target follow-up",
|
|
]
|
|
assert ui._filter_match("anything") is False
|
|
ui._filter_text = ""
|
|
assert ui._filter_match("anything") is True
|
|
|
|
ui.projection.status = SupremeStatusDTO()
|
|
assert ui._current_phase_label() is None
|
|
ui.projection.status = SupremeStatusDTO(
|
|
loop=LoopStatusDTO(None, "AlphaLoop", None, None),
|
|
)
|
|
assert ui._current_phase_label() == "ALPHA"
|
|
ui.projection.status = SupremeStatusDTO(
|
|
loop=LoopStatusDTO(None, "BetaLoop", None, None),
|
|
)
|
|
assert ui._current_phase_label() == "BETA"
|
|
ui.projection.status = SupremeStatusDTO(
|
|
loop=LoopStatusDTO(None, "CustomLoop", None, None),
|
|
)
|
|
assert ui._current_phase_label() == "CUSTOMLOOP"
|
|
|
|
|
|
def test_terminal_ui_clamps_zero_row_views_and_narrow_panel_widths() -> None:
|
|
"""Unavailable rows and narrow layouts retain bounded, in-memory view state."""
|
|
|
|
ui = TerminalUI(_MemoryQueue(), _MemoryQueue())
|
|
cot = CotStatusDTO(1, 1, ("only task",), ())
|
|
|
|
assert ui._visible_scrolled_lines("header", ["one"], 0) == []
|
|
assert ui._visible_cot_lines(cot, 20, 0) == []
|
|
assert ui._stabilization_column_widths(20) == (26, -5)
|
|
assert ui._top_status_panel_widths(20) == (18, 3)
|
|
|
|
|
|
def test_terminal_ui_completion_report_and_intervention_helpers_stay_in_memory(capsys) -> None:
|
|
"""Report notices and intervention responses use only DTO state and the memory queue."""
|
|
|
|
commands = _MemoryQueue()
|
|
ui = TerminalUI(_MemoryQueue(), commands)
|
|
ui._open_report_in_editor()
|
|
assert capsys.readouterr().err == ""
|
|
|
|
ui.projection.completion = _completion("reference-only-report.md")
|
|
ui._completion_focus = 1
|
|
assert ui._handle_completion_key(curses.KEY_ENTER) is True
|
|
assert ui._completion_action == "vim"
|
|
assert ui._handle_completion_key(ord("x")) is False
|
|
ui._open_report_in_editor()
|
|
assert "reference-only-report.md" in capsys.readouterr().err
|
|
|
|
ui._handle_intervention_key(ord("x"))
|
|
request = HumanInterventionRequestDTO(
|
|
request_id="request-4",
|
|
check_name="StationaryRetry",
|
|
original_prompt="Keep it in memory.",
|
|
repeated_response_tail="same response",
|
|
request_md5="a" * 32,
|
|
response_md5="b" * 32,
|
|
local_model="local",
|
|
escalated_model="tier-two",
|
|
)
|
|
ui._enter_intervention_mode(request)
|
|
ui._handle_intervention_key(9)
|
|
assert ui._message == "selected request"
|
|
ui._handle_intervention_key(27)
|
|
assert ui.projection.intervention is None
|
|
assert len(commands.entries) == 1
|
|
|
|
|
|
def test_terminal_ui_merges_overlapping_log_context_ranges() -> None:
|
|
"""Adjacent matched log windows are merged before presentation and retain source order."""
|
|
|
|
ui = TerminalUI(_MemoryQueue(), _MemoryQueue())
|
|
ui.projection.logs.extend(
|
|
LogMessageDTO(index, "INFO", "owner", f"target {index}")
|
|
for index in range(3)
|
|
)
|
|
ui._filter_text = "target"
|
|
ui._context_before = "1"
|
|
ui._context_after = "1"
|
|
|
|
assert [item.message for item in ui._visible_logs(10)] == [
|
|
"target 0",
|
|
"target 1",
|
|
"target 2",
|
|
]
|