band-aider-pub/tests/band_aider/test_cli.py

# SPDX-FileCopyrightText: 2026 Art Sea LLC
#
# SPDX-License-Identifier: AGPL-3.0-only

"""Isolated CLI assembly tests with stationary model and repository boundaries."""

from __future__ import annotations

from argparse import Namespace
from io import StringIO
from multiprocessing import Queue
from pathlib import Path
from unittest.mock import Mock

import pytest

from band_aider import cli
from band_aider.ai_client import llm_logging
from band_aider.ai_client.cached_completion_client import CachedCompletionClient
from band_aider.ai_client.completion_client import CompletionClient, OpenAIConfig
from band_aider.ai_client.cycle_jacker import CycleJacker
from band_aider.ai_client.fence_crashing_client import FenceCrashingClient
from band_aider.ai_client.human_feedback_store import HumanFeedbackStore
from band_aider.ai_client.llm_context_batch import LLMContextBatch
from band_aider.ai_client.llm_context_reducer import LLMContextReducer
from band_aider.ai_client.llm_context_reduction import LLMContextReduction
from band_aider.ai_client.openai_client_kind import OpenAIClientKind
from band_aider.ai_client.openai_compatible_client import OpenAICompatibleClient
from band_aider.ai_client.openai_config_factory import OpenAIConfigFactory
from band_aider.ai_client.openai_proxy_factory import OpenAIProxyFactory
from band_aider.ai_client.openai_responses_client import OpenAIResponsesClient
from band_aider.ai_client.token_jacker import TokenJacker
from band_aider.ai_client.token_budget_estimator import TokenBudgetEstimator
from band_aider.ai_client.token_miser import TokenMiser
from band_aider.ai_client.token_tracker import TokenTracker
from band_aider.cli import (
    _aider_model_name,
    _cleanup_empty_files,
    _configure_runtime_resume_settings,
    _client_model_name,
    _human_io_adapters,
    _max_output_tokens,
    _runtime_client,
    _runtime_clients,
    _skip_startup,
    _terminal_ui_worker_argv,
    build_parser,
    main as cli_main,
)
from band_aider.context.aider_config import AiderConfig
from band_aider.context.resume_state import resume_settings
from band_aider.context.token_budget_state import TokenBudgetSettings
from band_aider.first_aider import _client_from_env
from band_aider.first_aider import build_parser as build_first_aider_parser
from band_aider.first_aider import decompose, main as first_aider_main
from band_aider.judgement.prompt_formatter import MarkdownPromptFormatterBuilder
from band_aider.judgement.prompt_localization import PromptLocalization
from band_aider.judgement.repo_exec import HostRepoExec
from band_aider.judgement.search_replace_token_generator import SearchReplaceTokenGenerator
from band_aider.ui.dto.human_intervention import HumanInterventionRequestDTO
from band_aider.ui.dto.model_escalation import ModelEscalationDTO
from band_aider.ui.queue_status_publisher import (
    QueueHumanCommandReceiver,
    QueueStatusPublisher,
)
from band_aider.ui.status_publisher import (
    ConsoleHumanInterventionAdapter,
    DeferredHumanCommandReceiver,
    NullStatusPublisher,
)
from tests.support.completion_clients import ScriptedCompletionClient, StaticCompletionClient, StaticOpenAICompatibleClient
from tests.support.memory_path import MemoryPath
from tests.support.memory_repo_exec import MemoryRepoExec


class _SingleBatchReducer(LLMContextReducer):
    """Reduce one oversized prompt through a stationary single batch."""

    def __init__(self, *, accepts_prompt: bool = True, final_result: str = "reduced prompt") -> None:
        """Store the explicit reducer behavior used by one test."""

        self._accepts_prompt = accepts_prompt
        self._final_result = final_result

    def can_reduce(self, prompt: str) -> bool:
        """Return the configured eligibility without inspecting external state."""

        del prompt
        return self._accepts_prompt

    def split(
        self,
        prompt: str,
        token_budget: int,
        estimator: TokenBudgetEstimator,
    ) -> list[LLMContextBatch]:
        """Return one deterministic reduction batch."""

        del prompt, token_budget, estimator
        return [LLMContextBatch(batch_index=1, total_batches=1, content="batch")]

    def prompt_for_batch(
        self,
        original_prompt: str,
        batch: LLMContextBatch,
        previous_result: str,
    ) -> str:
        """Return a stationary reduction prompt for the one configured batch."""

        del original_prompt, batch, previous_result
        return "reduction prompt"

    def final_result(self, reductions: list[LLMContextReduction]) -> str:
        """Return the configured final prompt after observing the completed batch."""

        assert len(reductions) == 1
        return self._final_result


def _args(*argv: str) -> Namespace:
    """Parse one stationary command-line sequence."""

    return build_parser().parse_args(list(argv))


def _aider_configs() -> tuple[AiderConfig, ...]:
    """Return an explicit three-tier Aider configuration chain."""

    return (
        AiderConfig("aider", False, "openai/default", None, None),
        AiderConfig("aider", False, "openai/escalated-1", "https://one.invalid", "key"),
        AiderConfig("aider", False, "openai/escalated-2", "https://two.invalid", "key"),
    )


def _openai_configs() -> tuple[OpenAIConfig, ...]:
    """Return the matching stationary OpenAI configuration chain."""

    return (
        OpenAIConfig(model="default", api_base="https://default.invalid", api_key="key"),
        OpenAIConfig(model="escalated-1", api_base="https://one.invalid", api_key="key"),
        OpenAIConfig(model="escalated-2", api_base="https://two.invalid", api_key="key"),
    )


def test_cli_defaults() -> None:
    """The parser exposes the documented project defaults."""

    args = _args("--task", "Fix it")

    assert (args.repo_root, args.language, args.src_root, args.tests_root) == (
        None,
        "python",
        "band_aider",
        "tests",
    )


def test_cli_parser_keeps_explicit_repository_coordinate() -> None:
    """A repository coordinate remains a lexical parser value until execution owns it."""

    assert _args("--task", "Fix it", "--repo-root", "/memory/project").repo_root == "/memory/project"


def test_cli_parser_accepts_resume_flag() -> None:
    """Resume is an explicit parser state independent of any checkpoint storage."""

    assert _args("--resume").resume is True


@pytest.mark.parametrize(
    ("argv", "hydrate"),
    [
        (("--task", "new task"), False),
        (("--resume",), True),
    ],
)
def test_cli_binds_resume_mode_to_the_task_lifecycle(
    argv: tuple[str, ...],
    hydrate: bool,
) -> None:
    """CLI mode is the source of the task-owned provenance resume setting."""

    executor = MemoryRepoExec()
    _configure_runtime_resume_settings(_args(*argv), executor.root, executor)

    assert resume_settings().hydrate is hydrate
    assert resume_settings().repo_exec is executor


def test_cli_parser_accepts_terminal_ui_flag() -> None:
    """Terminal UI selection is represented without creating a worker process."""

    assert _args("--terminal-ui").terminal_ui is True


def test_cli_parser_accepts_container_coordinates() -> None:
    """Container execution coordinates remain explicit parser values."""

    args = _args("--task", "Fix it", "--repo-container", "band-aider-dev", "--repo-python-bin", "python3")

    assert (args.repo_container, args.repo_python_bin) == ("band-aider-dev", "python3")


def test_cli_parser_accepts_explicit_token_controls() -> None:
    """Token controls are parsed before any runtime configuration is consulted."""

    args = _args("--task", "Fix it", "--max-prompt-tokens", "123", "--max-output-tokens", "45", "--tokens-per-word", "2.5")

    assert (args.max_prompt_tokens, args.max_output_tokens, args.tokens_per_word) == (123, 45, 2.5)


def test_cli_parser_keeps_each_extra_aider_argument() -> None:
    """Repeated extra Aider arguments preserve their user-supplied order."""

    args = _args("--task", "Fix it", "--aider-extra-arg=--one", "--aider-extra-arg=--two")

    assert args.aider_extra_arg == ["--one", "--two"]


def test_cli_parser_accepts_confirm_empty_file_cleanup() -> None:
    """Cleanup authorization is explicit rather than inferred from host input."""

    assert _args("--task", "Fix it", "--confirm-empty-file-cleanup").confirm_empty_file_cleanup is True


@pytest.mark.integration
@pytest.mark.parametrize("localize", [False, True])
def test_cli_dry_run_completes_against_a_labeled_local_project(
    tmp_path_factory: pytest.TempPathFactory,
    monkeypatch: pytest.MonkeyPatch,
    localize: bool,
) -> None:
    """The whole tier assembles CLI runtime state without a provider request."""

    root = tmp_path_factory.mktemp("cli-dry-run")
    (root / "src").mkdir()
    (root / "tests").mkdir()
    # Each invocation must set its own policy, including resetting a prior opt-in.
    monkeypatch.setattr(PromptLocalization, "enabled", not localize)

    result = cli_main(
        [
            "--task",
            "CLI_DRY_RUN_TASK_SENTINEL",
            "--repo-root",
            str(root),
            "--src-root",
            "src",
            "--tests-root",
            "tests",
            "--model",
            "openai/stationary-model",
            "--model-api-base",
            "https://stationary-model.invalid",
            "--model-api-key",
            "stationary-key",
            "--dry-run",
            *(["--localize-prompts"] if localize else []),
        ]
    )

    assert result == 0
    assert PromptLocalization.enabled is localize


@pytest.mark.integration
@pytest.mark.parametrize("options", [[], ["--localize-prompts"]])
def test_terminal_worker_arguments_preserve_an_explicit_task_for_local_project(
    tmp_path_factory: pytest.TempPathFactory,
    options: list[str],
) -> None:
    """The labeled adapter tier constructs worker argv without prompting or Docker."""

    root = tmp_path_factory.mktemp("cli-worker-task")
    argv = [
        *options,
        "--terminal-ui",
        "--repo-root",
        str(root),
        "--task",
        "WORKER_TASK_SENTINEL",
    ]

    worker_argv = _terminal_ui_worker_argv(_args(*argv), argv)

    assert worker_argv == [
        *options,
        "--repo-root",
        str(root),
        "--task",
        "WORKER_TASK_SENTINEL",
        "--language",
        "python",
    ]


@pytest.mark.integration
def test_terminal_worker_arguments_resume_from_a_labeled_local_checkpoint(
    tmp_path_factory: pytest.TempPathFactory,
) -> None:
    """A real local checkpoint selects resume without consulting a host prompt."""

    root = tmp_path_factory.mktemp("cli-worker-resume")
    checkpoint = root / ".band-aider-chkpts"
    checkpoint.mkdir()
    (checkpoint / "state.json").write_text("{}", encoding="utf-8")
    argv = ["--terminal-ui", "--repo-root", str(root), "--resume"]

    worker_argv = _terminal_ui_worker_argv(_args(*argv), argv)

    assert worker_argv == ["--repo-root", str(root), "--resume", "--language", "python"]


@pytest.mark.integration
def test_terminal_worker_arguments_reject_missing_local_resume_checkpoint(
    tmp_path_factory: pytest.TempPathFactory,
) -> None:
    """A missing local checkpoint fails before a worker process can start."""

    root = tmp_path_factory.mktemp("cli-worker-no-resume")
    argv = ["--terminal-ui", "--repo-root", str(root), "--resume"]

    with pytest.raises(ValueError, match="No saved Band-Aider state found"):
        _terminal_ui_worker_argv(_args(*argv), argv)


@pytest.mark.integration
def test_empty_file_cleanup_uses_the_labeled_local_adapter(
    tmp_path_factory: pytest.TempPathFactory,
) -> None:
    """Explicit cleanup deletes only zero-byte local files through HostRepoExec."""

    root = tmp_path_factory.mktemp("cli-empty-cleanup")
    src_root = root / "src"
    src_root.mkdir()
    empty = src_root / "empty.py"
    retained = src_root / "retained.py"
    empty.write_text("", encoding="utf-8")
    retained.write_text("VALUE = 1\n", encoding="utf-8")
    args = _args(
        "--task",
        "CLEANUP_TASK_SENTINEL",
        "--confirm-empty-file-cleanup",
    )

    result = _cleanup_empty_files(args, root, src_root, HostRepoExec(root))

    assert result == 0
    assert not empty.exists()
    assert retained.read_text(encoding="utf-8") == "VALUE = 1\n"


@pytest.mark.integration
def test_cli_returns_a_usage_error_before_repository_access_without_task(capsys: pytest.CaptureFixture[str]) -> None:
    """An omitted task exits at parser ownership before a repository adapter exists."""

    assert cli_main([]) == 2
    assert "usage: band-aider" in capsys.readouterr().err


def test_cli_persists_unexpected_workflow_traceback_before_propagating(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A runtime failure survives in the repository error log and never publishes completion."""

    repository = MemoryRepoExec()
    repository.write_text(repository.root / "src/app.py", "VALUE = 1\n")
    repository.mkdir(repository.root / "tests", parents=True, exist_ok=True)
    client = StaticOpenAICompatibleClient("BAND_AIDER_READY")
    client.api_key = "stationary-test-key"
    failure = RuntimeError("Unexpected selector transport failure.")
    loop = Mock(spec=cli.SovereignLoop)
    loop.run.side_effect = failure
    publisher = Mock(spec=NullStatusPublisher)
    monkeypatch.delenv("BAND_AIDER_TERMINAL_UI", raising=False)
    monkeypatch.setattr(cli, "HostRepoExec", Mock(return_value=repository))
    monkeypatch.setattr(cli.AiderConfigFactory, "sequence", Mock(return_value=_aider_configs()))
    monkeypatch.setattr(cli.OpenAIConfigFactory, "sequence", Mock(return_value=_openai_configs()))
    monkeypatch.setattr(cli.OpenAIProxyFactory, "from_config", Mock(return_value=client))
    monkeypatch.setattr(cli.OpenAIProxyFactory, "from_configs", Mock(return_value=(client,)))
    monkeypatch.setattr(cli, "SovereignLoop", Mock(return_value=loop))
    monkeypatch.setattr(cli, "_human_io_adapters", Mock(return_value=(publisher, DeferredHumanCommandReceiver())))
    llm_logging.reset_llm_logging_for_tests()
    try:
        with pytest.raises(RuntimeError) as caught:
            cli_main([
                "--task", "Repair the stale caller.",
                "--repo-root", str(repository.root),
                "--src-root", "src", "--tests-root", "tests",
                "--worker-mode", "--token-overflow-strategy", "jacker",
            ])
        error_logs = repository.list_files(
            repository.root / ".band-aider-logs/cli",
            ("*_ERROR.md",),
            recursive=False,
        )
        assert caught.value is failure
        assert len(error_logs) == 1
        persisted = repository.read_text(error_logs[0])
        assert "Band-Aider workflow failed before completion." in persisted
        assert "Traceback (most recent call last):" in persisted
        assert "summary = loop.run()" in persisted
        assert "RuntimeError: Unexpected selector transport failure." in persisted
        assert all(type(call.args[0]).__name__ != "RunCompletionDTO" for call in publisher.publish.call_args_list)
    finally:
        llm_logging.reset_llm_logging_for_tests()
        cli.set_status_publisher(NullStatusPublisher())


@pytest.mark.integration
def test_cli_rejects_a_missing_labeled_local_repository(
    tmp_path_factory: pytest.TempPathFactory,
    capsys: pytest.CaptureFixture[str],
) -> None:
    """A missing explicit repository reports an input error without provider traffic."""

    missing_root = tmp_path_factory.mktemp("cli-missing-root") / "missing"

    assert cli_main(["--task", "TASK_SENTINEL", "--repo-root", str(missing_root)]) == 2
    assert "Repository root does not exist" in capsys.readouterr().err


@pytest.mark.integration
def test_cli_rejects_unsupported_language_after_local_configuration(
    tmp_path_factory: pytest.TempPathFactory,
    capsys: pytest.CaptureFixture[str],
) -> None:
    """A configured local dry run still rejects an unavailable language explicitly."""

    root = tmp_path_factory.mktemp("cli-unsupported-language")
    (root / "src").mkdir()
    (root / "tests").mkdir()

    result = cli_main(
        [
            "--task",
            "TASK_SENTINEL",
            "--repo-root",
            str(root),
            "--src-root",
            "src",
            "--tests-root",
            "tests",
            "--model",
            "openai/stationary-model",
            "--model-api-base",
            "https://stationary-model.invalid",
            "--model-api-key",
            "stationary-key",
            "--language",
            "unavailable-language",
            "--dry-run",
        ]
    )

    assert result == 2
    assert "language is not available" in capsys.readouterr().err


@pytest.mark.integration
def test_cli_rejects_a_traversing_source_root_before_any_model_configuration(
    tmp_path_factory: pytest.TempPathFactory,
    capsys: pytest.CaptureFixture[str],
) -> None:
    """A traversing source root stops before model configuration or provider setup."""

    root = tmp_path_factory.mktemp("cli-invalid-source-root")
    (root / "src").mkdir()
    (root / "tests").mkdir()

    result = cli_main(
        [
            "--task",
            "TASK_SENTINEL",
            "--repo-root",
            str(root),
            "--src-root",
            "../outside",
            "--tests-root",
            "tests",
        ]
    )

    assert result == 2
    assert "source root must not contain '..'" in capsys.readouterr().err


@pytest.mark.integration
def test_cli_helpers_preserve_noninteractive_and_queue_transport_boundaries() -> None:
    """The CLI selects explicit unattended or queue-backed human transport contracts."""

    assert _skip_startup(_args("--task", "TASK_SENTINEL", "--skip-startup")) is True
    assert _max_output_tokens(_args("--task", "TASK_SENTINEL", "--max-output-tokens", "73")) == 73

    unattended_status, unattended_receiver = _human_io_adapters(None, None)
    assert isinstance(unattended_status, NullStatusPublisher)
    assert isinstance(unattended_receiver, DeferredHumanCommandReceiver)

    event_queue = Queue()
    command_queue = Queue()
    try:
        queue_status, queue_receiver = _human_io_adapters(event_queue, command_queue)
        assert isinstance(queue_status, QueueStatusPublisher)
        assert isinstance(queue_receiver, QueueHumanCommandReceiver)
        with pytest.raises(ValueError, match="requires both event and command queues"):
            _human_io_adapters(event_queue, None)
    finally:
        event_queue.close()
        command_queue.close()


def test_runtime_client_stacks_fence_crasher_outside_token_miser() -> None:
    """The standard runtime stack cleans fences after token budgeting."""

    client = _runtime_client(
        ScriptedCompletionClient(("{}",)),
        _args("--task", "x", "--token-overflow-strategy", "miser"),
        _aider_configs(),
        MarkdownPromptFormatterBuilder(),
        repo_exec=MemoryRepoExec(),
    )

    assert isinstance(client, TokenTracker)
    assert isinstance(client.inner, FenceCrashingClient)
    assert isinstance(client.inner.inner, TokenMiser)


def test_runtime_client_caches_completed_prompts() -> None:
    """A cache decorator returns the first stationary completion for a repeated prompt."""

    raw_client = ScriptedCompletionClient(("{}",))
    client = _runtime_client(
        CachedCompletionClient(raw_client),
        _args("--task", "x", "--token-overflow-strategy", "miser"),
        _aider_configs(),
        MarkdownPromptFormatterBuilder(),
        repo_exec=MemoryRepoExec(),
    )

    assert client.complete("same prompt") == "{}"
    assert client.complete("same prompt") == "{}"
    assert raw_client.prompts == ["same prompt"]


def test_runtime_client_can_select_token_jacker() -> None:
    """An explicit jacker strategy selects the overflow-routing decorator."""

    client = _runtime_client(
        ScriptedCompletionClient(("{}",), _openai_configs()),
        _args("--task", "x", "--token-overflow-strategy", "jacker"),
        _aider_configs(),
        MarkdownPromptFormatterBuilder(),
        repo_exec=MemoryRepoExec(),
    )

    assert isinstance(client, TokenTracker)
    assert isinstance(client.inner, FenceCrashingClient)
    assert isinstance(client.inner.inner, TokenJacker)


def test_runtime_client_jacker_has_no_implicit_overflow_clients() -> None:
    """A single runtime client retains no hidden escalation transport."""

    client = _runtime_client(
        ScriptedCompletionClient(("{}",)),
        _args("--task", "x", "--token-overflow-strategy", "jacker"),
        _aider_configs(),
        MarkdownPromptFormatterBuilder(),
        repo_exec=MemoryRepoExec(),
    )

    assert isinstance(client.inner, FenceCrashingClient)
    assert isinstance(client.inner.inner, TokenJacker)
    assert client.inner.inner.escalation_clients == ()


def test_runtime_clients_build_one_wrapped_client_per_openai_config() -> None:
    """A runtime chain has one wrapped client per explicit model configuration."""

    clients = _runtime_clients(
        _openai_configs(),
        _args("--task", "x", "--token-overflow-strategy", "jacker"),
        _aider_configs(),
        MarkdownPromptFormatterBuilder(),
        repo_exec=MemoryRepoExec(),
    )

    trackers = tuple(client for client in clients if isinstance(client, TokenTracker))

    assert len(trackers) == 3
    for tracker in trackers:
        assert isinstance(tracker.inner, FenceCrashingClient)
        assert isinstance(tracker.inner.inner, TokenJacker)


def test_runtime_clients_use_token_miser_for_every_tier() -> None:
    """An explicit miser strategy applies consistently to every tier."""

    clients = _runtime_clients(
        _openai_configs(),
        _args("--task", "x", "--token-overflow-strategy", "miser"),
        _aider_configs(),
        MarkdownPromptFormatterBuilder(),
        repo_exec=MemoryRepoExec(),
    )

    trackers = tuple(client for client in clients if isinstance(client, TokenTracker))

    assert len(trackers) == 3
    for tracker in trackers:
        assert isinstance(tracker.inner, FenceCrashingClient)
        assert isinstance(tracker.inner.inner, TokenMiser)


def test_runtime_clients_preserve_escalation_openai_config_chain(memory_root: MemoryPath) -> None:
    """Cycle-aware runtime clients expose the trailing configuration chain in order."""

    response_config = OpenAIConfig(
        model="escalated-1",
        api_base="https://one.invalid",
        api_key="key",
        client=OpenAIClientKind.RESPONSE,
    )
    configs = (_openai_configs()[0], response_config, _openai_configs()[2])
    clients = _runtime_clients(
        configs,
        _args("--task", "x", "--token-overflow-strategy", "jacker"),
        _aider_configs(),
        MarkdownPromptFormatterBuilder(),
        repo_root=memory_root,
        repo_exec=memory_root.repository,
        cycle_detection=True,
    )

    assert [type(client.inner) for client in clients] == [CycleJacker, CycleJacker, CycleJacker]
    assert [[config.model for config in client.openai_configs] for client in clients] == [
        ["default", "escalated-1", "escalated-2"],
        ["escalated-1", "escalated-2"],
        ["escalated-2"],
    ]
    assert [config.client for config in clients[0].openai_configs] == [
        OpenAIClientKind.COMPLETION,
        OpenAIClientKind.RESPONSE,
        OpenAIClientKind.COMPLETION,
    ]


def test_runtime_client_injects_memory_repo_feedback_store(memory_root: MemoryPath) -> None:
    """Cycle detection persists through the supplied in-memory repository adapter."""

    client = _runtime_client(
        ScriptedCompletionClient(("{}",)),
        _args("--task", "x", "--token-overflow-strategy", "jacker"),
        _aider_configs(),
        MarkdownPromptFormatterBuilder(),
        repo_root=memory_root,
        repo_exec=memory_root.repository,
        cycle_detection=True,
    )

    assert isinstance(client.inner, CycleJacker)
    assert isinstance(client.inner.feedback_store, HumanFeedbackStore)
    assert client.inner.feedback_store.repo_root == memory_root.resolve()


def test_runtime_clients_feed_search_replace_generator_with_escalation_order(memory_root: MemoryPath) -> None:
    """Search-replace selection receives the runtime chain in explicit escalation order."""

    clients = _runtime_clients(
        _openai_configs(),
        _args("--task", "x", "--token-overflow-strategy", "jacker"),
        _aider_configs(),
        MarkdownPromptFormatterBuilder(),
        repo_exec=memory_root.repository,
    )
    generator = SearchReplaceTokenGenerator(
        clients,
        memory_root,
        repo_exec=memory_root.repository,
        prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
    )
    editable: list[Path] = [memory_root / "pkg" / "module.py"]

    for _ in range(3):
        generator._record_editable_scope(editable)
    selected = generator._client_for_next_edit(editable, prompt="prompt", logger=None)

    assert [client.openai_configs[0].model for client in generator.clients] == [
        "default",
        "escalated-1",
        "escalated-2",
    ]
    assert selected.openai_configs[0].model == "escalated-1"


def test_console_human_intervention_adapter_collects_guidance() -> None:
    """The injected console stream combines multiple stationary guidance lines."""

    output = StringIO()
    responses = iter(("check imports", "remove retry", "/submit"))
    adapter = ConsoleHumanInterventionAdapter(input_func=lambda prompt: next(responses), output_stream=output)
    adapter.publish(
        ModelEscalationDTO(
            trigger_name="CycleJacker",
            occurred_at=0,
            prompt_preview="prompt",
            reason="repeat",
            request_md5="a" * 32,
            local_model="local",
            active_model="cloud",
            config_index=1,
            config_count=2,
            state="human_required",
            response_md5="b" * 32,
            response_tail="tail",
            repeat_count=3,
        )
    )
    adapter.publish(
        HumanInterventionRequestDTO(
            request_id="request-1",
            check_name="CycleJacker",
            original_prompt="Fix parser.",
            repeated_response_tail="No changes.",
            request_md5="a" * 32,
            response_md5="b" * 32,
            local_model="local",
            escalated_model="cloud",
        )
    )

    response = adapter.wait_for("request-1")

    assert (response.action, response.guidance) == ("resend", "check imports\nremove retry")
    assert "REQUEST" in output.getvalue()


def test_console_human_intervention_adapter_explicitly_skips_hint() -> None:
    """An explicit skip requests direct dictation without interpreting blank lines as commands."""

    adapter = ConsoleHumanInterventionAdapter(input_func=lambda prompt: "/skip", output_stream=StringIO())
    adapter.publish(
        HumanInterventionRequestDTO(
            request_id="request-1",
            check_name="CycleJacker",
            original_prompt="Fix parser.",
            repeated_response_tail="No changes.",
            request_md5="a" * 32,
            response_md5="b" * 32,
            local_model="local",
            escalated_model="cloud",
        )
    )

    response = adapter.wait_for("request-1")

    assert (response.action, response.guidance) == ("retry", "")


def test_console_human_intervention_adapter_defers_unknown_request() -> None:
    """A nonmatching request identifier fails closed without reading input."""

    adapter = ConsoleHumanInterventionAdapter(input_func=lambda prompt: "unexpected", output_stream=StringIO())

    response = adapter.wait_for("unknown")

    assert (response.action, response.guidance) == ("defer", "")


def test_model_name_helpers_normalize_across_boundaries() -> None:
    """Aider and provider model names apply their opposite normalization rules."""

    assert _aider_model_name("default") == "openai/default"
    assert _aider_model_name("openai/default") == "openai/default"
    assert _client_model_name("openai/default") == "default"
    assert _client_model_name("default") == "default"


def test_openai_config_factory_stamps_response_client_mode_from_explicit_mapping() -> None:
    """A stationary CLIENT mapping selects the response-client enum."""

    config = OpenAIConfigFactory.from_env(
        _args("--task", "x"),
        {"OPENAI_MODEL": "openai/default", "CLIENT": "response"},
    )

    assert (config.model, config.client) == ("default", OpenAIClientKind.RESPONSE)


def test_openai_config_factory_defaults_unknown_client_mode_to_completion() -> None:
    """Unknown client labels retain the compatible-completions mode."""

    config = OpenAIConfigFactory.from_env(
        _args("--task", "x"),
        {"OPENAI_MODEL": "openai/default", "CLIENT": "unknown"},
    )

    assert config.client is OpenAIClientKind.COMPLETION


def test_cli_factory_builds_response_proxy_from_response_mode_configs() -> None:
    """Response-mode configuration constructs a proxy without performing a request."""

    client = OpenAIProxyFactory.from_config(
        OpenAIConfig(
            model="model",
            api_base="https://model.invalid",
            api_key="key",
            client=OpenAIClientKind.RESPONSE,
        )
    )

    assert isinstance(client, OpenAIResponsesClient)


def test_cli_factory_builds_legacy_proxy_by_default() -> None:
    """Default configuration constructs the compatibility proxy without transport."""

    client = OpenAIProxyFactory.from_config(
        OpenAIConfig(model="model", api_base="https://model.invalid", api_key="key")
    )

    assert isinstance(client, OpenAICompatibleClient)


def test_first_aider_arg_validation() -> None:
    """The first-aider parser rejects mutually exclusive prompt sources before execution."""

    try:
        first_aider_main(["prompt", "--task", "task"])
    except SystemExit as error:
        assert error.code == 1
    else:
        raise AssertionError("first-aider should reject duplicate prompt inputs")


def test_first_aider_parser_accepts_container_coordinate() -> None:
    """First-aider retains container coordinates as parser-only state."""

    args = build_first_aider_parser().parse_args(
        ["prompt", "--repo-root", "/memory/project", "--repo-container", "band-aider-dev"]
    )

    assert (args.repo_root, args.repo_container) == ("/memory/project", "band-aider-dev")


def test_first_aider_parser_accepts_dry_run() -> None:
    """Dry-run intent is parsed without constructing a repository executor."""

    assert build_first_aider_parser().parse_args(["prompt", "--dry-run"]).dry_run is True


def test_first_aider_decomposition_uses_a_stationary_completion_response() -> None:
    """Decomposition owns its JSON envelope and accepts an explicit static task list."""

    tasks = decompose(
        StaticCompletionClient(
            '{"tasks": [{"title": "First", "task": "FIRST_TASK_SENTINEL", "reason": "ordered"}]}'
        ),
        "LARGE_TASK_SENTINEL",
        prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
    )

    assert tasks == [
        {"title": "First", "task": "FIRST_TASK_SENTINEL", "reason": "ordered"}
    ]


def test_first_aider_decomposition_corrects_invalid_json_until_tasks_are_typed(
    capsys: pytest.CaptureFixture[str],
) -> None:
    """Rejected replies never become fallback work while a valid correction is available."""

    rejected = tuple(map(lambda index: f"invalid-decomposition-{index}", range(16)))
    client = ScriptedCompletionClient((*rejected,
        '{"tasks":[{"title":"Repair","task":"LARGE_TASK_SENTINEL","reason":"One coherent task."}]}',
    ))

    tasks = decompose(
        client,
        "LARGE_TASK_SENTINEL",
        prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
    )

    assert tasks == [{"title": "Repair", "task": "LARGE_TASK_SENTINEL", "reason": "One coherent task."}]
    assert len(client.prompts) == 17
    assert client.prompts[-1].startswith(client.prompts[0])
    assert rejected[-1] in client.prompts[-1]
    assert rejected[-2] not in client.prompts[-1]
    assert "Warning: Decomposition failed" not in capsys.readouterr().err


def test_first_aider_decomposition_transport_failure_keeps_fallback(
    monkeypatch: pytest.MonkeyPatch,
    capsys: pytest.CaptureFixture[str],
) -> None:
    """An unavailable provider retains the existing fallback without output correction."""

    client = StaticCompletionClient("unused")
    completion = Mock(side_effect=ConnectionError("provider unavailable"))
    monkeypatch.setattr(client, "complete", completion)

    tasks = decompose(client, "LARGE_TASK_SENTINEL", prompt_formatter_builder=MarkdownPromptFormatterBuilder())

    assert tasks is None
    completion.assert_called_once()
    assert "Warning: Decomposition failed: provider unavailable" in capsys.readouterr().err


@pytest.mark.integration
def test_first_aider_client_uses_explicit_cli_provider_coordinates() -> None:
    """CLI coordinates take precedence without creating a provider request."""

    client = _client_from_env(
        build_first_aider_parser().parse_args(
            [
                "LARGE_TASK_SENTINEL",
                "--model",
                "openai/stationary-model",
                "--model-api-base",
                "https://stationary-model.invalid",
                "--model-api-key",
                "stationary-key",
            ]
        )
    )

    assert (client.model, client.api_base, client.api_key) == (
        "openai/stationary-model",
        "https://stationary-model.invalid",
        "stationary-key",
    )


def test_token_miser_reduction_flow_uses_stationary_responses() -> None:
    """TokenMiser reduces through a nominal reducer and then completes the reduced prompt."""

    inner = ScriptedCompletionClient(("reduced prompt", "final result"))
    miser = TokenMiser(
        inner=inner,
        settings=TokenBudgetSettings(max_prompt_tokens=1, max_output_tokens=0, tokens_per_word=1.0),
        reducers=[_SingleBatchReducer()],
    )

    assert miser.complete("too long") == "final result"
    assert inner.prompts == ["reduction prompt", "reduced prompt"]


def test_token_miser_forwards_when_no_nominal_reducer_accepts() -> None:
    """An oversized prompt remains intact when no reducer accepts it."""

    inner = ScriptedCompletionClient(("result",))
    miser = TokenMiser(
        inner=inner,
        settings=TokenBudgetSettings(max_prompt_tokens=1, max_output_tokens=0, tokens_per_word=1.0),
        reducers=[_SingleBatchReducer(accepts_prompt=False)],
    )

    assert miser.complete("too long") == "result"
    assert inner.prompts == ["too long"]


def test_token_miser_forwards_when_reduction_is_empty() -> None:
    """An empty reduction result preserves the original prompt as the semantic fallback."""

    inner = ScriptedCompletionClient(("intermediate", "result"))
    miser = TokenMiser(
        inner=inner,
        settings=TokenBudgetSettings(max_prompt_tokens=1, max_output_tokens=0, tokens_per_word=1.0),
        reducers=[_SingleBatchReducer(final_result="")],
    )

    assert miser.complete("too long") == "result"
    assert inner.prompts == ["reduction prompt", "too long"]


def test_runtime_client_strips_stationary_json_fences() -> None:
    """Fence cleanup remains visible through the assembled runtime client."""

    client = _runtime_client(
        ScriptedCompletionClient(("```json\n{}\n```",)),
        _args("--task", "x", "--token-overflow-strategy", "miser"),
        _aider_configs(),
        MarkdownPromptFormatterBuilder(),
        repo_exec=MemoryRepoExec(),
    )

    assert client.complete("prompt") == "{}"


def test_runtime_tracker_exposes_stationary_config_chain() -> None:
    """The assembled tracker preserves the direct client's configuration metadata."""

    configs = _openai_configs()
    client = _runtime_client(
        ScriptedCompletionClient(("{}",), configs),
        _args("--task", "x", "--token-overflow-strategy", "jacker"),
        _aider_configs(),
        MarkdownPromptFormatterBuilder(),
        repo_exec=MemoryRepoExec(),
    )

    assert client.openai_configs == configs


def test_runtime_client_returns_completion_client_contract() -> None:
    """The runtime assembly endpoint retains the nominal completion-client contract."""

    client = _runtime_client(
        ScriptedCompletionClient(("{}",)),
        _args("--task", "x", "--token-overflow-strategy", "jacker"),
        _aider_configs(),
        MarkdownPromptFormatterBuilder(),
        repo_exec=MemoryRepoExec(),
    )

    assert isinstance(client, CompletionClient)