121 lines
4.1 KiB
Python
121 lines
4.1 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
from __future__ import annotations
|
|
|
|
from multiprocessing.queues import Queue
|
|
from queue import Empty
|
|
|
|
from band_aider.ui.dto.cot_status import CotStatusDTO
|
|
from band_aider.ui.dto.supreme_status import SupremeStatusDTO
|
|
from band_aider.ui.dto.story_status import StoryStatusDTO
|
|
from band_aider.ui.status import publish_cot_status, publish_story_status, status_publisher_scope
|
|
from band_aider.ui.terminal_ui import TerminalProjection, TerminalUI, render_projection
|
|
from band_aider.context.global_context import StoryCursor
|
|
from band_aider.context.class_meta import ClassMeta
|
|
from band_aider.context.cot_task_generator import CotTaskRecord
|
|
|
|
|
|
class _Publisher:
|
|
"""Collect status events sent through the existing status publisher."""
|
|
|
|
def __init__(self) -> None:
|
|
self.events: list[object] = []
|
|
|
|
def publish(self, event: object) -> None:
|
|
self.events.append(event)
|
|
|
|
|
|
class _IdleQueue(Queue[object]):
|
|
"""Supply the terminal's nominal queue type when this test needs no events."""
|
|
|
|
def __init__(self) -> None:
|
|
"""Avoid allocating worker-backed queue resources for idle presentation tests."""
|
|
|
|
def get_nowait(self) -> object:
|
|
"""Report that the stationary test queue has no pending events."""
|
|
|
|
raise Empty
|
|
|
|
def put(
|
|
self,
|
|
obj: object,
|
|
block: bool = True,
|
|
timeout: float | None = None,
|
|
) -> None:
|
|
"""Accept unused terminal commands without communicating with a worker."""
|
|
|
|
del obj, block, timeout
|
|
|
|
|
|
class _Context:
|
|
"""Minimal story/COT status context."""
|
|
|
|
stories = ["first story", "second story"]
|
|
story_cursor = StoryCursor(
|
|
current_story_index=1,
|
|
active_story="second story",
|
|
cot_task_records=[CotTaskRecord("src/example.py: update evidence", "src/example.py", ClassMeta("src/example.py", "", [], {}))],
|
|
)
|
|
original_prompt = "original unsplit prompt"
|
|
|
|
|
|
def test_story_status_travels_through_supreme_status_queue() -> None:
|
|
"""Story state is part of SupremeStatusDTO and precedes restored COT status."""
|
|
|
|
publisher = _Publisher()
|
|
with status_publisher_scope(publisher):
|
|
publish_story_status(_Context())
|
|
publish_cot_status(_Context())
|
|
|
|
assert isinstance(publisher.events[0], SupremeStatusDTO)
|
|
first = publisher.events[0]
|
|
assert first.story is not None
|
|
assert first.story.all_stories == ("first story", "second story")
|
|
assert first.story.current_index == 1
|
|
assert first.story.completed_indices == (0,)
|
|
assert isinstance(publisher.events[1], SupremeStatusDTO)
|
|
second = publisher.events[1]
|
|
assert second.story is not None
|
|
assert second.story.current_index == 1
|
|
|
|
|
|
def test_story_header_precedes_legacy_task_and_plain_output_shows_cot() -> None:
|
|
"""Plain rendering prefers the active numbered story and nests COT progress."""
|
|
|
|
projection = TerminalProjection(
|
|
status=SupremeStatusDTO(
|
|
task="original unsplit prompt",
|
|
story=StoryStatusDTO(1, 2, ("first story", "second story"), (0,)),
|
|
cot=CotStatusDTO(1, 1, ("src/example.py: update evidence",), ()),
|
|
)
|
|
)
|
|
|
|
rendered = render_projection(projection)
|
|
assert "Story 2/2: second story" in rendered
|
|
assert "original unsplit prompt" not in rendered
|
|
assert "COT 1/1" in rendered
|
|
|
|
|
|
def test_story_change_resets_header_and_cot_view_state() -> None:
|
|
"""Changing the active story resets scroll/follow state and stale COT item."""
|
|
|
|
ui = TerminalUI(_IdleQueue(), _IdleQueue())
|
|
ui.projection.status = SupremeStatusDTO(
|
|
story=StoryStatusDTO(0, 2, ("first", "second"), ()),
|
|
)
|
|
ui._scroll_offsets["header"] = 4
|
|
ui._scroll_offsets["cot"] = 3
|
|
ui._follow_current["cot"] = False
|
|
ui.projection.current_cot_item = "old"
|
|
|
|
ui.projection.status = SupremeStatusDTO(
|
|
story=StoryStatusDTO(1, 2, ("first", "second"), (0,)),
|
|
)
|
|
ui._sync_story_view_state()
|
|
|
|
assert ui._scroll_offsets["header"] == 0
|
|
assert ui._scroll_offsets["cot"] == 0
|
|
assert ui._follow_current["cot"] is True
|
|
assert ui.projection.current_cot_item is None
|