724 lines
38 KiB
Python
724 lines
38 KiB
Python
"""Targeted intervention keyframes exercised without a terminal or model service."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import curses
|
|
import logging
|
|
import sys
|
|
from collections.abc import Iterator
|
|
from io import StringIO
|
|
from multiprocessing.queues import Queue
|
|
from queue import Empty
|
|
from typing import cast
|
|
|
|
import pytest
|
|
|
|
from band_aider.ai_client import completion_client, json_completion, escalation_utils
|
|
from band_aider.ai_client.completion_client import CompletionClient, HumanResponse, OpenAIConfig
|
|
from band_aider.ai_client.cycle_jacker import CycleJacker, cycle
|
|
from band_aider.ai_client.human_feedback_store import HumanFeedbackStore
|
|
from band_aider.ai_client.stabilization_escalation import StabilizationEscalation
|
|
from band_aider.judgement.prompt_formatter import MarkdownPromptFormatterBuilder
|
|
from band_aider.ai_client.openai_compatible_client import OpenAICompatibleClient
|
|
from band_aider.context.class_meta import ClassMeta
|
|
from band_aider.judgement.global_relevant_check import GlobalRelevantCheck
|
|
from band_aider.ui import escalation_ui, intervention_selection
|
|
from band_aider.ui.dto.human_intervention import (
|
|
CycleInterruptCommandDTO, HumanInterventionRequestDTO, HumanInterventionResponseDTO,
|
|
InterventionChoicesDTO, InterventionStatusDTO, InterventionClearCommandDTO, InterventionSuppressionCommandDTO,
|
|
)
|
|
from band_aider.ui.queue_status_publisher import QueueHumanCommandReceiver
|
|
from band_aider.ui.status_publisher import HumanCommandReceiver, StatusPublisher
|
|
from band_aider.ui.terminal_ui import TerminalUI
|
|
from tests.support.memory_repo_exec import MemoryRepoExec
|
|
|
|
|
|
class _Provider(CompletionClient):
|
|
"""Record exactly what reaches the provider; _Provider(fail_guidance=True) simulates failure."""
|
|
|
|
def __init__(self, fail_guidance: bool = False) -> None:
|
|
"""Bind an independent call record and optional guided failure."""
|
|
self.prompts: list[str] = []
|
|
self.fail_guidance = fail_guidance
|
|
|
|
@property
|
|
def inner(self) -> CompletionClient | None:
|
|
"""Expose a provider with no decorators."""
|
|
return None
|
|
|
|
@property
|
|
def openai_configs(self) -> tuple[OpenAIConfig, ...]:
|
|
"""Return the single stationary provider identity."""
|
|
return (OpenAIConfig("local", "http://stationary.invalid"),)
|
|
|
|
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
|
|
"""Record each invocation and fail only the guided attempt when requested."""
|
|
self.prompts.append(prompt)
|
|
# Failure occurs after the request is dispatched, when guidance must already be consumed.
|
|
if self.fail_guidance and "single-use advice" in prompt:
|
|
assert cycle.human_interrupt is False
|
|
raise OSError("provider unavailable")
|
|
return '{"outcome": "pass"}'
|
|
|
|
|
|
class _Human(StatusPublisher, HumanCommandReceiver):
|
|
"""Provide one scripted operator action; _Human('replace text') preserves exact text."""
|
|
|
|
def __init__(self, text: str = "single-use advice", action: str = "resend") -> None:
|
|
"""Initialize independent event and consumption observations."""
|
|
self.text, self.action = text, action
|
|
self.events: list[object] = []
|
|
self.requests: list[HumanInterventionRequestDTO] = []
|
|
self.armed_during_input: list[bool] = []
|
|
|
|
def publish(self, event: object) -> None:
|
|
"""Retain UI evidence including the exact targeted request."""
|
|
self.events.append(event)
|
|
# Only human requests require a scripted response.
|
|
if isinstance(event, HumanInterventionRequestDTO):
|
|
self.requests.append(event)
|
|
|
|
def consume_pending(self) -> None:
|
|
"""Supply no out-of-band commands during the stationary exchange."""
|
|
|
|
def wait_for(self, request_id: str) -> HumanInterventionResponseDTO:
|
|
"""Observe that the manual request survives until actual submission."""
|
|
self.armed_during_input.append(cycle.human_interrupt)
|
|
return HumanInterventionResponseDTO(request_id, self.text, self.action)
|
|
|
|
|
|
class _Queue(Queue):
|
|
"""Own in-memory UI transport; _Queue(events) avoids multiprocessing resources."""
|
|
|
|
def __init__(self, events: tuple[object, ...] = ()) -> None:
|
|
"""Seed an independent FIFO."""
|
|
self.entries = list(events)
|
|
|
|
def get_nowait(self) -> object:
|
|
"""Honor the queue's empty boundary."""
|
|
if not self.entries:
|
|
raise Empty
|
|
return self.entries.pop(0)
|
|
|
|
def put(self, obj: object, block: bool = True, timeout: float | None = None) -> None:
|
|
"""Record a UI command in FIFO order."""
|
|
self.entries.append(obj)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def isolated_intervention(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
|
|
"""Restore the run-global interruption state independently for every scenario."""
|
|
monkeypatch.setattr(cycle, "human_interrupt", False)
|
|
monkeypatch.setattr(cycle, "intervention_suppressed", False)
|
|
monkeypatch.setattr(cycle, "target", CycleInterruptCommandDTO())
|
|
monkeypatch.setattr(cycle, "cues", {})
|
|
yield
|
|
|
|
|
|
@pytest.mark.parametrize("name,files", (("OtherCheck", ("src/a.py",)), ("IntentCheck", ("src/b.py",)), ("IntentCheck", ())))
|
|
def test_near_match_retains_request_until_exact_owner(name: str, files: tuple[str, ...]) -> None:
|
|
"""Changing only decision or authoritative file scope cannot consume the stop."""
|
|
provider, human = _Provider(), _Human()
|
|
repo = MemoryRepoExec()
|
|
client = CycleJacker(provider, feedback_store=HumanFeedbackStore(repo.root, repo_exec=repo),
|
|
event_publisher=human, command_receiver=human, prompt_formatter_builder=MarkdownPromptFormatterBuilder())
|
|
target = CycleInterruptCommandDTO("IntentCheck", "src/a.py")
|
|
cycle.arm(target, human)
|
|
with completion_client.scoped_completion(name, files):
|
|
client.complete("The prompt mentions IntentCheck and src/a.py as readonly evidence.")
|
|
assert human.requests == []
|
|
assert cycle.human_interrupt is True
|
|
assert cycle.target is target
|
|
with completion_client.scoped_completion("IntentCheck", ("src/a.py",)):
|
|
client.complete("actual decision")
|
|
client.complete("later decision")
|
|
assert len(human.requests) == 1
|
|
assert human.armed_during_input == [True]
|
|
assert human.requests[0].check_name == "IntentCheck"
|
|
assert human.requests[0].files == ("src/a.py",)
|
|
assert "single-use advice" in provider.prompts[-2]
|
|
assert provider.prompts[-1] == "later decision"
|
|
assert cycle.human_interrupt is False
|
|
|
|
|
|
@pytest.mark.parametrize("target", (CycleInterruptCommandDTO(), CycleInterruptCommandDTO("IntentCheck"), CycleInterruptCommandDTO(file="src/a.py")))
|
|
def test_blank_filters_preserve_next_matching_call(target: CycleInterruptCommandDTO) -> None:
|
|
"""Any, decision-only, and file-only selections preserve their intended scope."""
|
|
provider, human = _Provider(), _Human()
|
|
workflow = StabilizationEscalation((provider,), check_name="IntentCheck", event_publisher=human,
|
|
command_receiver=human, prompt_formatter_builder=MarkdownPromptFormatterBuilder())
|
|
workflow.files = ("src/a.py",)
|
|
cycle.arm(target, human)
|
|
workflow.complete("judge")
|
|
assert len(human.requests) == 1
|
|
assert cycle.human_interrupt is False
|
|
|
|
|
|
@pytest.mark.parametrize("stabilization", (False, True))
|
|
def test_replacement_reaches_validator_without_model(stabilization: bool) -> None:
|
|
"""Human replacement crosses either completion owner verbatim with zero model calls."""
|
|
provider, human = _Provider(), _Human(' {"outcome": "human"} \n')
|
|
repo = MemoryRepoExec()
|
|
client = (StabilizationEscalation((provider,), check_name="IntentCheck", event_publisher=human,
|
|
command_receiver=human, prompt_formatter_builder=MarkdownPromptFormatterBuilder())
|
|
if stabilization else CycleJacker(provider, feedback_store=HumanFeedbackStore(repo.root, repo_exec=repo),
|
|
event_publisher=human, command_receiver=human, prompt_formatter_builder=MarkdownPromptFormatterBuilder()))
|
|
cycle.arm(CycleInterruptCommandDTO("IntentCheck", strength="replace"), human)
|
|
with completion_client.scoped_completion("IntentCheck"):
|
|
response = client.complete("Return strict JSON")
|
|
assert isinstance(response, HumanResponse)
|
|
assert response == human.text
|
|
assert provider.prompts == []
|
|
assert human.armed_during_input == [True]
|
|
assert human.requests[0].mode == "direct_response"
|
|
assert cycle.human_interrupt is False
|
|
|
|
|
|
@pytest.mark.parametrize("stabilization", (False, True))
|
|
def test_failed_guided_call_cannot_replay_guidance(stabilization: bool) -> None:
|
|
"""Provider failure leaves both the stop and active guidance consumed before retry."""
|
|
provider, human = _Provider(fail_guidance=True), _Human()
|
|
repo = MemoryRepoExec()
|
|
client = (StabilizationEscalation((provider,), check_name="IntentCheck", event_publisher=human,
|
|
command_receiver=human, prompt_formatter_builder=MarkdownPromptFormatterBuilder())
|
|
if stabilization else CycleJacker(provider, feedback_store=HumanFeedbackStore(repo.root, repo_exec=repo),
|
|
event_publisher=human, command_receiver=human, prompt_formatter_builder=MarkdownPromptFormatterBuilder()))
|
|
cycle.arm(CycleInterruptCommandDTO("IntentCheck"), human)
|
|
with completion_client.scoped_completion("IntentCheck"):
|
|
with pytest.raises(OSError, match="provider unavailable"):
|
|
client.complete("judge")
|
|
client.complete("judge")
|
|
assert cycle.human_interrupt is False
|
|
assert human.armed_during_input == [True]
|
|
assert len(human.requests) == 1
|
|
assert provider.prompts[-1] == "judge"
|
|
assert sum("single-use advice" in prompt for prompt in provider.prompts) == 1
|
|
|
|
|
|
def test_automatic_escalation_does_not_consume_unrelated_manual_request() -> None:
|
|
"""Automatic exhaustion may ask for help while another decision's target remains armed."""
|
|
provider, human = _Provider(), _Human()
|
|
repo = MemoryRepoExec()
|
|
client = CycleJacker(provider, feedback_store=HumanFeedbackStore(repo.root, repo_exec=repo), repeat_threshold=1,
|
|
event_publisher=human, command_receiver=human, prompt_formatter_builder=MarkdownPromptFormatterBuilder())
|
|
target = CycleInterruptCommandDTO("OtherCheck", "src/other.py")
|
|
cycle.arm(target, human)
|
|
with completion_client.scoped_completion("IntentCheck"):
|
|
client.complete("stationary")
|
|
client.complete("stationary")
|
|
assert len(human.requests) == 1
|
|
assert not human.requests[0].guide_only
|
|
assert cycle.target is target
|
|
assert cycle.human_interrupt is True
|
|
|
|
|
|
def test_nested_json_helper_cannot_borrow_enclosing_decision() -> None:
|
|
"""A helper's explicit completion identity masks its enclosing owner and restores it afterward."""
|
|
provider, human = _Provider(), _Human()
|
|
repo = MemoryRepoExec()
|
|
client = CycleJacker(provider, feedback_store=HumanFeedbackStore(repo.root, repo_exec=repo),
|
|
event_publisher=human, command_receiver=human, prompt_formatter_builder=MarkdownPromptFormatterBuilder())
|
|
cycle.arm(CycleInterruptCommandDTO("IntentCheck", "src/a.py"), human)
|
|
with completion_client.scoped_completion("IntentCheck", ("src/a.py",)):
|
|
json_completion.load_json_completion(client, "helper", check_name="Helper", prompt_formatter_builder=MarkdownPromptFormatterBuilder())
|
|
assert human.requests == []
|
|
client.complete("owner")
|
|
assert len(human.requests) == 1
|
|
assert completion_client.completion_scope.get().check_name == ""
|
|
|
|
|
|
@pytest.mark.parametrize("action", ("defer", "retry", "suppress"))
|
|
@pytest.mark.parametrize("strength", ("guide", "replace"))
|
|
def test_skipping_manual_intervention_resumes_model_without_draft(action: str, strength: str) -> None:
|
|
"""Skipped manual screens resume model work without dictation or discarded draft text."""
|
|
provider, human = _Provider(), _Human("discard this draft", action)
|
|
workflow = StabilizationEscalation((provider,), check_name="IntentCheck", event_publisher=human,
|
|
command_receiver=human, prompt_formatter_builder=MarkdownPromptFormatterBuilder())
|
|
cycle.arm(CycleInterruptCommandDTO("IntentCheck", strength=strength), human)
|
|
workflow.complete("judge")
|
|
assert provider.prompts == ["judge"]
|
|
assert len(human.requests) == 1
|
|
assert human.requests[0].guide_only is (strength == "guide")
|
|
assert cycle.human_interrupt is False
|
|
|
|
|
|
def test_curses_selection_and_acknowledgement_keep_draft_separate() -> None:
|
|
"""Selection resolves canonical identities; edits do not mutate the acknowledged request."""
|
|
commands = _Queue()
|
|
events = _Queue((InterventionChoicesDTO(("src/a.py", "tests/a.py")),))
|
|
ui = TerminalUI(events, commands)
|
|
ui._drain_events()
|
|
ui._intervention_selection.values.update(decision="Intent", file="src/", strength="rep")
|
|
command = ui._intervention_selection.command()
|
|
assert command == CycleInterruptCommandDTO("IntentCheck", "src/a.py", "replace")
|
|
events.put(InterventionStatusDTO(command, True))
|
|
ui._drain_events()
|
|
ui._intervention_selection.handle(27, "file")
|
|
assert ui.projection.manual_stop == InterventionStatusDTO(command, True)
|
|
assert ui._intervention_selection.values["file"] == ""
|
|
events.put(InterventionStatusDTO(command, False))
|
|
ui._drain_events()
|
|
assert ui.projection.manual_stop == InterventionStatusDTO(command, False)
|
|
|
|
|
|
def test_cli_choices_reject_ambiguity_and_preserve_canonical_paths() -> None:
|
|
"""Plain selectors refine ambiguity and share exact resolution with curses."""
|
|
entries = iter(("", "s", "src/b.py", "rep"))
|
|
output = StringIO()
|
|
target = intervention_selection.read_target(("src/a.py", "src/b.py"), lambda _: next(entries), output)
|
|
assert target == CycleInterruptCommandDTO("", "src/b.py", "replace")
|
|
assert "Ambiguous choice" in output.getvalue()
|
|
completer = intervention_selection.ChoiceCompleter(("src/a.py", "src/b.py", "tests/a.py"))
|
|
assert completer.complete("src/", 0) == "src/a.py"
|
|
assert completer.complete("src/", 1) == "src/b.py"
|
|
assert completer.complete("src/", 2) is None
|
|
assert completer.complete("missing", 0) is None
|
|
|
|
|
|
def test_invalid_replacement_is_rejected_without_model_repair() -> None:
|
|
"""Dictation can supply bytes but cannot bypass the existing JSON admission boundary."""
|
|
import json
|
|
|
|
provider, human = _Provider(), _Human("not JSON")
|
|
repo = MemoryRepoExec()
|
|
client = CycleJacker(provider, feedback_store=HumanFeedbackStore(repo.root, repo_exec=repo),
|
|
event_publisher=human, command_receiver=human, prompt_formatter_builder=MarkdownPromptFormatterBuilder())
|
|
cycle.arm(CycleInterruptCommandDTO("IntentCheck", strength="replace"), human)
|
|
with pytest.raises(json.JSONDecodeError):
|
|
json_completion.load_json_completion(client, "Return strict JSON", check_name="IntentCheck", prompt_formatter_builder=MarkdownPromptFormatterBuilder())
|
|
assert provider.prompts == []
|
|
assert cycle.human_interrupt is False
|
|
|
|
|
|
def test_rejected_stabilization_replacement_returns_to_human_with_draft() -> None:
|
|
"""After manual consumption, validation still owns correction of a dictated answer."""
|
|
provider, human = _Provider(), _Human("invalid draft")
|
|
workflow = StabilizationEscalation((provider,), check_name="IntentCheck", event_publisher=human,
|
|
command_receiver=human, prompt_formatter_builder=MarkdownPromptFormatterBuilder())
|
|
cycle.arm(CycleInterruptCommandDTO("IntentCheck", strength="replace"), human)
|
|
assert workflow.complete("judge") == "invalid draft"
|
|
assert cycle.human_interrupt is False
|
|
workflow.validation_errors = "Return the required JSON shape."
|
|
human.text = '{"outcome": "pass"}'
|
|
assert workflow.complete("judge with validation feedback") == human.text
|
|
assert human.requests[-1].mode == "direct_response"
|
|
assert human.requests[-1].draft == "invalid draft"
|
|
assert human.requests[-1].validation_errors == "Return the required JSON shape."
|
|
assert provider.prompts == []
|
|
|
|
|
|
def test_rejected_manual_guidance_does_not_force_human_replacement() -> None:
|
|
"""A guidance attempt leaves the model correction policy intact on the next attempt."""
|
|
provider, human = _Provider(), _Human()
|
|
workflow = StabilizationEscalation((provider,), check_name="IntentCheck", event_publisher=human,
|
|
command_receiver=human, prompt_formatter_builder=MarkdownPromptFormatterBuilder())
|
|
cycle.arm(CycleInterruptCommandDTO("IntentCheck"), human)
|
|
workflow.complete("judge")
|
|
workflow.validation_errors = "The answer needs correction."
|
|
workflow.complete("judge with correction")
|
|
assert len(human.requests) == 1
|
|
assert provider.prompts[-1] == "judge with correction"
|
|
assert "single-use advice" not in provider.prompts[-1]
|
|
|
|
|
|
def test_upstream_relevance_judgment_can_be_replaced_for_one_focused_file() -> None:
|
|
"""Target the actual file-admission judgment before its answer admits source work."""
|
|
class RelevanceProvider(_Provider):
|
|
"""Return a typed negative when no operator replacement targets this file."""
|
|
|
|
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
|
|
"""Retain invocation evidence and decline this source explicitly."""
|
|
self.prompts.append(prompt)
|
|
return '{"pull": false, "reason": "", "relevant_symbols": {}}'
|
|
|
|
provider = RelevanceProvider()
|
|
human = _Human('{"pull": true, "reason": "Operator selected this file", "relevant_symbols": {}}')
|
|
repo = MemoryRepoExec()
|
|
client = CycleJacker(provider, feedback_store=HumanFeedbackStore(repo.root, repo_exec=repo),
|
|
event_publisher=human, command_receiver=human, prompt_formatter_builder=MarkdownPromptFormatterBuilder())
|
|
check = GlobalRelevantCheck(changes=[], prompt="Repair the widget", client=cast(OpenAICompatibleClient, client),
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder())
|
|
other = ClassMeta(location="src/b.py", summary="Other", relevance=[], relevant_symbols={})
|
|
focused = ClassMeta(location="src/a.py", summary="Widget", relevance=[], relevant_symbols={})
|
|
cycle.arm(CycleInterruptCommandDTO("GlobalRelevantCheck", "src/a.py", "replace"), human)
|
|
assert check(other) is False
|
|
assert human.requests == []
|
|
assert check(focused) is True
|
|
assert focused.relevance == ["Operator selected this file"]
|
|
assert len(provider.prompts) == 1
|
|
assert human.requests[0].files == ("src/a.py",)
|
|
assert cycle.human_interrupt is False
|
|
|
|
|
|
def test_cli_intervention_options_and_unattended_setup_do_not_read_stdin(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Explicit selector arguments parse without input; unattended interactive setup is rejected."""
|
|
from band_aider import cli
|
|
|
|
args = cli.build_parser().parse_args(["--intervention", "--intervention-decision", "IntentCheck", "--intervention-file", "src/a.py", "--intervention-response", "replace"])
|
|
assert args.intervention is True
|
|
assert args.intervention_decision == "IntentCheck"
|
|
assert args.intervention_file == "src/a.py"
|
|
assert args.intervention_response == "replace"
|
|
monkeypatch.setattr(sys, "stdin", StringIO())
|
|
with pytest.raises(SystemExit) as error:
|
|
cli.main(["--intervention-setup"])
|
|
assert error.value.code == 2
|
|
|
|
|
|
@pytest.mark.parametrize("stabilization", (False, True))
|
|
@pytest.mark.parametrize("strength", ("guide", "replace"))
|
|
def test_saved_cue_supplies_one_matching_human_submission_without_screen(stabilization: bool, strength: str) -> None:
|
|
"""Ahead-of-time feedback follows the existing guide or verbatim validation path once."""
|
|
provider, human = _Provider(), _Human()
|
|
repo = MemoryRepoExec()
|
|
client = (StabilizationEscalation((provider,), check_name="IntentCheck", event_publisher=human,
|
|
command_receiver=human, prompt_formatter_builder=MarkdownPromptFormatterBuilder())
|
|
if stabilization else CycleJacker(provider, feedback_store=HumanFeedbackStore(repo.root, repo_exec=repo),
|
|
event_publisher=human, command_receiver=human, prompt_formatter_builder=MarkdownPromptFormatterBuilder()))
|
|
text = ' {"outcome": "human"}\nλ ' if strength == "replace" else "single-use advice\n retain the contract "
|
|
target = CycleInterruptCommandDTO("IntentCheck", "src/a.py", strength, text)
|
|
cycle.arm(target, human)
|
|
# Both owners receive authoritative file scope, never a filename guessed from prompt prose.
|
|
if isinstance(client, StabilizationEscalation):
|
|
client.files = ("src/a.py",)
|
|
with completion_client.scoped_completion("IntentCheck", ("src/a.py",)):
|
|
response = client.complete("judge")
|
|
assert human.requests == []
|
|
assert cycle.human_interrupt is False
|
|
assert cycle.status.cues == ()
|
|
# A replacement reaches validation unchanged; guidance uses the existing provider path.
|
|
if strength == "replace":
|
|
assert isinstance(response, HumanResponse)
|
|
assert response == text
|
|
assert provider.prompts == []
|
|
else:
|
|
assert text in provider.prompts[-1]
|
|
# A new judgment key has ordinary model policy; rejected replacement stays in dictation.
|
|
if isinstance(client, StabilizationEscalation):
|
|
client.complete("later ordinary request", disambiguation="later")
|
|
else:
|
|
client.complete("later ordinary request")
|
|
assert human.requests == []
|
|
assert provider.prompts[-1] == "later ordinary request"
|
|
|
|
|
|
@pytest.mark.parametrize("check_name,files", (("OtherCheck", ("src/a.py",)), ("IntentCheck", ("src/b.py",)), ("IntentCheck", ())))
|
|
def test_saved_cue_near_match_cannot_consume_or_inject_feedback(check_name: str, files: tuple[str, ...]) -> None:
|
|
"""Changing either key or omitting file scope leaves the exact owner's cue untouched."""
|
|
provider, human = _Provider(), _Human()
|
|
repo = MemoryRepoExec()
|
|
client = CycleJacker(provider, feedback_store=HumanFeedbackStore(repo.root, repo_exec=repo),
|
|
event_publisher=human, command_receiver=human, prompt_formatter_builder=MarkdownPromptFormatterBuilder())
|
|
target = CycleInterruptCommandDTO("IntentCheck", "src/a.py", "guide", "single-use advice")
|
|
cycle.arm(target, human)
|
|
with completion_client.scoped_completion(check_name, files):
|
|
client.complete("Readonly prose mentions IntentCheck and src/a.py.")
|
|
assert provider.prompts == ["Readonly prose mentions IntentCheck and src/a.py."]
|
|
assert cycle.status.cues == (target,)
|
|
assert human.requests == []
|
|
with completion_client.scoped_completion("IntentCheck", ("src/./a.py",)):
|
|
client.complete("owner")
|
|
assert "single-use advice" in provider.prompts[-1]
|
|
assert cycle.status.cues == ()
|
|
assert human.requests == []
|
|
|
|
|
|
@pytest.mark.parametrize("stabilization", (False, True))
|
|
def test_saved_guidance_is_consumed_even_when_provider_dispatch_fails(stabilization: bool) -> None:
|
|
"""Provider failure cannot turn a popped cue into repeated guidance or a stuck manual stop."""
|
|
provider, human = _Provider(fail_guidance=True), _Human()
|
|
repo = MemoryRepoExec()
|
|
client = (StabilizationEscalation((provider,), check_name="IntentCheck", event_publisher=human,
|
|
command_receiver=human, prompt_formatter_builder=MarkdownPromptFormatterBuilder())
|
|
if stabilization else CycleJacker(provider, feedback_store=HumanFeedbackStore(repo.root, repo_exec=repo),
|
|
event_publisher=human, command_receiver=human, prompt_formatter_builder=MarkdownPromptFormatterBuilder()))
|
|
# Stabilization owns its own file scope rather than the renderer's completion scope.
|
|
if isinstance(client, StabilizationEscalation):
|
|
client.files = ("src/a.py",)
|
|
cycle.arm(CycleInterruptCommandDTO("IntentCheck", "src/a.py", "guide", "single-use advice"), human)
|
|
with completion_client.scoped_completion("IntentCheck", ("src/a.py",)):
|
|
with pytest.raises(OSError, match="provider unavailable"):
|
|
client.complete("judge")
|
|
assert cycle.status.cues == ()
|
|
assert cycle.human_interrupt is False
|
|
client.complete("judge")
|
|
assert human.requests == []
|
|
assert provider.prompts[-1] == "judge"
|
|
assert sum("single-use advice" in prompt for prompt in provider.prompts) == 1
|
|
|
|
|
|
@pytest.mark.parametrize("stabilization", (False, True))
|
|
def test_automatic_human_boundary_uses_cue_without_consuming_another_manual_target(stabilization: bool) -> None:
|
|
"""A saved key remains usable at automatic human help after another target is armed."""
|
|
provider, human = _Provider(), _Human()
|
|
repo = MemoryRepoExec()
|
|
target = CycleInterruptCommandDTO("IntentCheck", "src/a.py", "replace", '{"outcome":"human"}')
|
|
cycle.arm(target, human)
|
|
other = CycleInterruptCommandDTO("OtherCheck", "src/b.py")
|
|
cycle.arm(other, human)
|
|
# The existing cadence decides when to ask for help; saving cues does not advance it.
|
|
if stabilization:
|
|
client = StabilizationEscalation((provider,), check_name="IntentCheck", event_publisher=human,
|
|
command_receiver=human, prompt_formatter_builder=MarkdownPromptFormatterBuilder())
|
|
client.files = ("src/a.py",)
|
|
with completion_client.scoped_completion("IntentCheck", ("src/a.py",)):
|
|
for _ in range(12):
|
|
client.complete("stationary")
|
|
assert cycle.status.cues == (target,)
|
|
response = client.complete("stationary")
|
|
else:
|
|
client = CycleJacker(provider, feedback_store=HumanFeedbackStore(repo.root, repo_exec=repo), repeat_threshold=1,
|
|
event_publisher=human, command_receiver=human, prompt_formatter_builder=MarkdownPromptFormatterBuilder())
|
|
with completion_client.scoped_completion("IntentCheck", ("src/a.py",)):
|
|
client.complete("stationary")
|
|
assert cycle.status.cues == (target,)
|
|
response = client.complete("stationary")
|
|
assert isinstance(response, HumanResponse)
|
|
assert response == target.guidance
|
|
assert cycle.status.cues == ()
|
|
assert cycle.target is other
|
|
assert cycle.human_interrupt is True
|
|
assert human.requests == []
|
|
|
|
|
|
def test_saved_invalid_replacement_still_returns_to_human_validation_correction() -> None:
|
|
"""Stored replacement skips input once, then rejected bytes require the existing human correction."""
|
|
provider, human = _Provider(), _Human('{"outcome":"pass"}')
|
|
client = StabilizationEscalation((provider,), check_name="IntentCheck", event_publisher=human,
|
|
command_receiver=human, prompt_formatter_builder=MarkdownPromptFormatterBuilder())
|
|
client.files = ("src/a.py",)
|
|
cycle.arm(CycleInterruptCommandDTO("IntentCheck", "src/a.py", "replace", "invalid draft"), human)
|
|
assert isinstance(client.complete("judge"), HumanResponse)
|
|
assert human.requests == []
|
|
assert cycle.status.cues == ()
|
|
client.validation_errors = "Return the required JSON shape."
|
|
assert client.complete("judge with validation feedback") == human.text
|
|
assert human.requests[-1].mode == "direct_response"
|
|
assert human.requests[-1].draft == "invalid draft"
|
|
assert human.requests[-1].validation_errors == client.validation_errors
|
|
assert provider.prompts == []
|
|
|
|
|
|
def test_worker_cues_replace_same_key_and_clear_only_requested_entries() -> None:
|
|
"""Worker storage upserts one decision/file key and acknowledges keyed or complete clears."""
|
|
human = _Human()
|
|
first = CycleInterruptCommandDTO("IntentCheck", "src/a.py", "guide", "old")
|
|
replacement = CycleInterruptCommandDTO("IntentCheck", "src/./a.py", "replace", "new")
|
|
other = CycleInterruptCommandDTO("IntentCheck", "src/b.py", "guide", "other")
|
|
commands = _Queue((first, replacement, other))
|
|
receiver = QueueHumanCommandReceiver(commands, human)
|
|
receiver.consume_pending()
|
|
assert cycle.status.cues == (replacement, other)
|
|
commands.put(InterventionClearCommandDTO(first))
|
|
receiver.consume_pending()
|
|
assert cycle.status.cues == (other,)
|
|
assert cycle.target is other
|
|
assert cycle.human_interrupt is True
|
|
commands.put(InterventionClearCommandDTO())
|
|
receiver.consume_pending()
|
|
assert cycle.status.cues == ()
|
|
assert cycle.target is other
|
|
assert cast(InterventionStatusDTO, human.events[-1]).cues == ()
|
|
|
|
|
|
@pytest.mark.parametrize("decision,file", (("", "src/a.py"), ("IntentCheck", "")))
|
|
def test_cue_requires_both_exact_keys_but_regular_manual_stop_keeps_any_filters(decision: str, file: str) -> None:
|
|
"""Ahead-of-time text cannot drift into an unrestricted stop's next unrelated file."""
|
|
with pytest.raises(ValueError, match="both a decision/check and a file"):
|
|
CycleInterruptCommandDTO(decision, file, "guide", "specific advice")
|
|
assert CycleInterruptCommandDTO(decision, file).guidance == ""
|
|
|
|
|
|
class _CueScreen:
|
|
"""Capture dashboard drawing and keyboard input without a terminal or worker."""
|
|
|
|
def __init__(self, keys: tuple[int | str, ...] = ()) -> None:
|
|
"""Allocate independent input and rendered-text evidence."""
|
|
self.keys = iter(keys)
|
|
self.text: list[str] = []
|
|
|
|
def getch(self) -> int:
|
|
"""Honor the ordinary dashboard's numeric keyboard boundary."""
|
|
key = next(self.keys)
|
|
return ord(key) if isinstance(key, str) else key
|
|
|
|
def get_wch(self) -> int | str:
|
|
"""Preserve exact Unicode and function keys for the cue editor."""
|
|
return next(self.keys)
|
|
|
|
def getmaxyx(self) -> tuple[int, int]:
|
|
"""Provide a stationary dashboard viewport."""
|
|
return 36, 160
|
|
|
|
def addstr(self, y: int, x: int, text: str, attr: int = 0) -> None:
|
|
"""Retain rendered labels and tooltip text."""
|
|
self.text.append(text)
|
|
|
|
def derwin(self, height: int, width: int, y: int, x: int) -> _CueScreen:
|
|
"""Let boxed dashboard panels share the drawing capture."""
|
|
return self
|
|
|
|
def erase(self) -> None:
|
|
"""Keep draw evidence until the test inspects it."""
|
|
|
|
def border(self) -> None:
|
|
"""Avoid host curses drawing."""
|
|
|
|
def refresh(self) -> None:
|
|
"""Avoid host terminal refresh."""
|
|
|
|
def timeout(self, delay: int) -> None:
|
|
"""Keep terminal paste framing independent of host timing."""
|
|
|
|
def nodelay(self, enabled: bool) -> None:
|
|
"""Honor the paste reader's return to dashboard polling without host IO."""
|
|
|
|
|
|
def test_dashboard_cue_requires_clear_ack_before_replacement_and_f10_clears_all() -> None:
|
|
"""The dashboard sends complete cue text, locks saved keys, and waits for worker clears."""
|
|
commands, events = _Queue(), _Queue()
|
|
ui = TerminalUI(events, commands)
|
|
ui._intervention_selection.files = ("src/a.py",)
|
|
ui._intervention_selection.values.update(decision="IntentCheck", file="src/a.py", strength="replace")
|
|
ui._set_focus("cue")
|
|
screen = _CueScreen(('λ', '\n', 'x', curses.KEY_F5, 'y', curses.KEY_F5, 10, curses.KEY_F5, curses.KEY_F10))
|
|
ui._handle_key(screen, None)
|
|
ui._handle_key(screen, None)
|
|
ui._handle_key(screen, None)
|
|
assert commands.entries == []
|
|
ui._handle_key(screen, None)
|
|
saved = cast(CycleInterruptCommandDTO, commands.entries[-1])
|
|
assert saved == CycleInterruptCommandDTO("IntentCheck", "src/a.py", "replace", "λ\nx")
|
|
events.put(InterventionStatusDTO(saved, True, (saved,)))
|
|
ui._drain_events()
|
|
ui._handle_key(screen, None)
|
|
ui._handle_key(screen, None)
|
|
assert commands.entries == [saved]
|
|
assert "clear" in ui._message.lower()
|
|
ui._set_focus("clear_cue")
|
|
ui._handle_key(screen, None)
|
|
assert commands.entries[-1] == InterventionClearCommandDTO(CycleInterruptCommandDTO("IntentCheck", "src/a.py", "replace"))
|
|
events.put(InterventionStatusDTO(saved, True, ()))
|
|
ui._drain_events()
|
|
ui._cue_state.guidance = "replacement after clear"
|
|
ui._handle_key(screen, None)
|
|
assert commands.entries[-1] == CycleInterruptCommandDTO("IntentCheck", "src/a.py", "replace", "replacement after clear")
|
|
ui._handle_key(screen, None)
|
|
assert commands.entries[-1] == InterventionClearCommandDTO()
|
|
|
|
|
|
def test_dashboard_clear_buttons_and_tooltip_share_keyed_worker_commands(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Virtual buttons expose both clear actions and display the unused F10 shortcut."""
|
|
commands = _Queue()
|
|
ui = TerminalUI(_Queue(), commands)
|
|
ui._intervention_selection.files = ("src/a.py",)
|
|
ui._intervention_selection.values.update(decision="IntentCheck", file="src/a.py")
|
|
saved = CycleInterruptCommandDTO("IntentCheck", "src/a.py", "guide", "stored advice")
|
|
ui.projection.apply(InterventionStatusDTO(saved, True, (saved,)))
|
|
ui._set_focus("clear_cues")
|
|
screen = _CueScreen()
|
|
ui._draw(screen, None)
|
|
assert "F10" in ui._message
|
|
assert "Clear all cues (F10)" in "\n".join(screen.text)
|
|
assert "clear before replacement" in "\n".join(screen.text)
|
|
top, left, _, _ = ui._focus_regions["clear_cues"]
|
|
monkeypatch.setattr(curses, "getmouse", lambda: (0, left, top, 0, curses.REPORT_MOUSE_POSITION))
|
|
ui._handle_mouse()
|
|
assert "F10" in ui._message
|
|
assert commands.entries == []
|
|
# Clicking each virtual control preserves its distinct keyed or all-cues command.
|
|
for name, expected in (("clear_cue", InterventionClearCommandDTO(CycleInterruptCommandDTO("IntentCheck", "src/a.py"))),
|
|
("clear_cues", InterventionClearCommandDTO())):
|
|
top, left, _, _ = ui._focus_regions[name]
|
|
monkeypatch.setattr(curses, "getmouse", lambda: (0, left, top, 0, curses.BUTTON1_CLICKED))
|
|
ui._handle_mouse()
|
|
assert commands.entries[-1] == expected
|
|
|
|
|
|
@pytest.mark.parametrize("check_name,files", (("OtherCheck", ("src/a.py",)), ("IntentCheck", ("src/b.py",)), ("IntentCheck", ())))
|
|
def test_automatic_human_near_match_preserves_cue_and_opens_existing_screen(check_name: str, files: tuple[str, ...]) -> None:
|
|
"""At an actual human boundary, either unmatched key prevents cue consumption."""
|
|
provider, human = _Provider(), _Human()
|
|
repo = MemoryRepoExec()
|
|
client = CycleJacker(provider, feedback_store=HumanFeedbackStore(repo.root, repo_exec=repo), repeat_threshold=1,
|
|
event_publisher=human, command_receiver=human, prompt_formatter_builder=MarkdownPromptFormatterBuilder())
|
|
cue = CycleInterruptCommandDTO("IntentCheck", "src/a.py", "replace", "specific stored replacement")
|
|
cycle.arm(cue, human)
|
|
with completion_client.scoped_completion(check_name, files):
|
|
client.complete("stationary")
|
|
client.complete("stationary")
|
|
assert len(human.requests) == 1
|
|
assert human.requests[0].check_name == check_name
|
|
assert human.requests[0].files == files
|
|
assert cycle.status.cues == (cue,)
|
|
assert "specific stored replacement" not in "\n".join(provider.prompts)
|
|
assert cycle.human_interrupt is True
|
|
|
|
|
|
def test_dashboard_bracketed_paste_preserves_full_cue_without_submitting_control_characters() -> None:
|
|
"""Multiline Unicode paste is stored exactly and only F5 arms it."""
|
|
commands = _Queue()
|
|
ui = TerminalUI(_Queue(), commands)
|
|
ui._intervention_selection.files = ("src/a.py",)
|
|
ui._intervention_selection.values.update(decision="IntentCheck", file="src/a.py", strength="replace")
|
|
ui._set_focus("cue")
|
|
text = "λ\n indented\ttext\n\x03"
|
|
screen = _CueScreen(tuple("\x1b[200~" + text + "\x1b[201~") + (curses.KEY_F5,))
|
|
ui._handle_key(screen, None)
|
|
# Every pasted byte remains data, including characters normally used to abort or navigate.
|
|
for _ in range(len(text) + len("\x1b[201~")):
|
|
ui._handle_key(screen, None)
|
|
assert commands.entries == []
|
|
assert ui._cue_state.guidance == text
|
|
ui._handle_key(screen, None)
|
|
assert commands.entries == [CycleInterruptCommandDTO("IntentCheck", "src/a.py", "replace", text)]
|
|
|
|
|
|
def test_dashboard_cue_focus_restores_terminal_paste_framing_after_human_screen(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Actual terminal paste framing follows the cue field through an intervening human screen."""
|
|
output = StringIO()
|
|
monkeypatch.setattr(output, "isatty", lambda: True)
|
|
monkeypatch.setattr(sys, "stdout", output)
|
|
ui = TerminalUI(_Queue(), _Queue())
|
|
ui._set_focus("cue")
|
|
assert output.getvalue() == "\x1b[?2004h"
|
|
request = HumanInterventionRequestDTO("req", "IntentCheck", "judge", "previous", "a", "b", "local", "cloud")
|
|
ui._enter_intervention_mode(request)
|
|
ui._exit_intervention_mode()
|
|
assert output.getvalue() == "\x1b[?2004h" * 3
|
|
ui._set_focus("file")
|
|
assert output.getvalue().endswith("\x1b[?2004l")
|
|
|
|
|
|
def test_dashboard_cue_viewport_reviews_all_saved_lines_without_mutating_text() -> None:
|
|
"""Arrow navigation changes only the text viewport while the complete stored cue remains intact."""
|
|
ui = TerminalUI(_Queue(), _Queue())
|
|
ui._intervention_selection.values.update(decision="IntentCheck", file="src/a.py")
|
|
saved = CycleInterruptCommandDTO("IntentCheck", "src/a.py", "guide", "first line\nlast line")
|
|
ui.projection.apply(InterventionStatusDTO(saved, True, (saved,)))
|
|
ui._set_focus("cue")
|
|
first = _CueScreen()
|
|
ui._draw(first, None)
|
|
assert "first line" in first.text
|
|
ui._handle_key(_CueScreen((curses.KEY_DOWN,)), None)
|
|
second = _CueScreen()
|
|
ui._draw(second, None)
|
|
assert "last line" in second.text
|
|
assert ui.projection.manual_stop is not None
|
|
assert ui.projection.manual_stop.cues == (saved,)
|
|
|
|
|
|
def test_suppression_at_human_boundary_preserves_cue_and_skips_guided_provider() -> None:
|
|
"""A newly queued F9 suppression keeps saved feedback unconsumed and resumes model policy."""
|
|
provider, human = _Provider(), _Human()
|
|
cue = CycleInterruptCommandDTO("IntentCheck", "src/a.py", "guide", "specific advice")
|
|
cycle.arm(cue, human)
|
|
receiver = QueueHumanCommandReceiver(_Queue((InterventionSuppressionCommandDTO(True),)), human)
|
|
with pytest.raises(escalation_utils.EscalationDeferred):
|
|
escalation_utils.human_response("judge", clients=(provider,), check_name="IntentCheck",
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(), event_publisher=human,
|
|
command_receiver=receiver, direct=False, previous_response="old",
|
|
pop_cue=lambda: cycle.pop_cue("IntentCheck", ("src/a.py",), human))
|
|
assert cycle.status.cues == (cue,)
|
|
assert cycle.intervention_suppressed is True
|
|
assert provider.prompts == []
|