band-aider-pub/tests/judgement/test_source_preflight_decision.py

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

"""Request-scoped source-preflight decision contracts."""

from __future__ import annotations

import json
from pathlib import Path
from unittest.mock import Mock, PropertyMock

import pytest

from band_aider.context.class_meta import ClassMeta
from band_aider.judgement.edit_request import AiderEditRequest
from band_aider.judgement.contextual_file_renderer import ContextualFileRenderer
from band_aider.judgement.file_manager import FileManager
from band_aider.judgement.file_parser import FileParser
from band_aider.judgement.prompt_formatter import MarkdownPromptFormatterBuilder
from band_aider.judgement.source_preflight_decision import (
    AlphaSourcePreflightDecision,
    BootstrapSourcePreflightDecision,
)
from band_aider.judgement.stabilization_decision import StabilizationDecision
from tests.support.completion_clients import (
    ScriptedCompletionClient,
    StaticOpenAICompatibleClient,
)
from tests.support.memory_repo_exec import MemoryRepoExec


def _file_manager(sources: dict[str, str]) -> FileManager:
    """Build the real renderer manager over stationary in-memory source exhibits."""

    repo_root = Path("/testbed")
    repo_exec = MemoryRepoExec(repo_root)
    for location, content in sources.items():
        repo_exec.write_text(repo_root / location, content)
    prompt_formatter_builder = MarkdownPromptFormatterBuilder()
    parser_client = StaticOpenAICompatibleClient(json.dumps({
        "location": "",
        "summary": "fixture source",
        "relevance": ["relevant"],
        "irrelevance": [],
        "relevant_symbols": {},
    }))
    return FileManager(
        parser=FileParser(parser_client, prompt_formatter_builder, repo_exec),
        client=parser_client,
        repo_root=repo_root,
        src_root=repo_root / "src",
        test_root=repo_root / "tests",
        actor="test",
        prompt_formatter_builder=prompt_formatter_builder,
        repo_exec=repo_exec,
    )


def _source(manager: FileManager, location: str) -> ClassMeta:
    """Return one parsed source fixture with an explicit non-null assertion boundary."""

    source = manager.current_src(location)
    assert source is not None
    return source


def test_alpha_source_preflight_is_a_retrying_stabilization_decision() -> None:
    """Semantically empty preflight output retries without moving Alpha authority."""

    manager = _file_manager({"src/focused.py": "pass\n"})
    source = _source(manager, "src/focused.py")
    request = AiderEditRequest(
        authoritative_prompt="ALPHA_AUTHORITATIVE_TASK_SENTINEL",
        editable_files=[Path(source.location)],
        readonly_files=[],
    )
    client = ScriptedCompletionClient((
        json.dumps([""]),
        json.dumps(["src/new_boundary.py"]),
    ))
    decision = AlphaSourcePreflightDecision(
        client=client,
        file_manager=manager,
        request=request,
        source=source,
        work_needed="ALPHA_PROVISIONAL_WORK_SENTINEL",
    )

    assert isinstance(decision, StabilizationDecision)
    assert decision.request is request
    first_prompt = decision.prompt
    assert decision() is True
    assert decision.additional["requested_locations"] == ["src/new_boundary.py"]
    assert decision.prompt == first_prompt
    assert client.prompts[0] == first_prompt
    assert client.prompts[1].startswith(f"{first_prompt}\n\n== BEGIN RESPONSE VALIDATION ERROR ==\n")
    assert "[0]: must be a nonempty repo-relative source path." in client.prompts[1]
    assert decision.prompt.index("AUTHORITATIVE ALPHA TASK") < decision.prompt.index(
        "PROVISIONAL WORK GUIDANCE"
    )
    assert "pass" in decision.prompt


def test_bootstrap_source_preflight_is_a_retrying_stabilization_decision() -> None:
    """Bootstrap preflight retries semantically empty requests at its validation seam."""

    manager = _file_manager({
        "src/focused.py": "pass\n",
        "src/readonly.py": "pass\n",
    })
    focused = _source(manager, "src/focused.py")
    readonly = _source(manager, "src/readonly.py")
    request = AiderEditRequest(
        authoritative_prompt="BOOTSTRAP_AUTHORITATIVE_TASK_SENTINEL",
        editable_files=[Path(focused.location)],
        readonly_files=[Path(readonly.location)],
    )
    client = ScriptedCompletionClient((
        json.dumps([{"location": "", "reason": "required split"}]),
        json.dumps([{
            "location": "src/new_boundary.py",
            "reason": "required split",
        }]),
    ))
    decision = BootstrapSourcePreflightDecision(
        client=client,
        file_manager=manager,
        request=request,
        original_request="BOOTSTRAP_ORIGINAL_REQUEST_SENTINEL",
        implementation_prompt="BOOTSTRAP_IMPLEMENTATION_PROMPT_SENTINEL",
        design_plan={"new_files": []},
        editable_sources=[focused],
        readonly_sources=[readonly],
        work_needed_by_location={
            focused.location: "FOCUSED_WORK",
            readonly.location: "READONLY_WORK",
        },
    )

    assert isinstance(decision, StabilizationDecision)
    assert decision.request is request
    first_prompt = decision.prompt
    assert decision() is True
    assert decision.additional["requested_files"] == [
        {"location": "src/new_boundary.py", "reason": "required split"},
    ]
    assert decision.prompt == first_prompt
    assert client.prompts[0] == first_prompt
    assert client.prompts[1].startswith(f"{first_prompt}\n\n== BEGIN RESPONSE VALIDATION ERROR ==\n")
    assert "[0].location: must be a nonempty repo-relative source path." in client.prompts[1]
    assert decision.prompt.index("ORIGINAL BOOTSTRAP REQUEST") < decision.prompt.index(
        "IMPLEMENTATION AUTHORITY"
    )
    assert "pass" in decision.prompt


@pytest.mark.parametrize("segmented", [False, True])
def test_bootstrap_preflight_explains_omissions_after_source_rendering(
    monkeypatch: pytest.MonkeyPatch, segmented: bool,
) -> None:
    """Deferred source rendering still controls whether the prompt explains omitted code."""

    manager = _file_manager({"src/focused.py": "pass\n"})
    source = _source(manager, "src/focused.py")
    # Exercise the renderer's two possible source representations at the same boundary.
    evidence = "FILE: src/focused.py\n⋮\n" if segmented else "FILE: src/focused.py\npass\n"
    read = Mock(return_value=evidence)
    segmentation = PropertyMock(side_effect=lambda: segmented and read.called)
    monkeypatch.setattr(ContextualFileRenderer, "read", read)
    monkeypatch.setattr(ContextualFileRenderer, "used_segmentation", segmentation)
    client = ScriptedCompletionClient(("[]",))
    decision = BootstrapSourcePreflightDecision(
        client=client,
        file_manager=manager,
        request=AiderEditRequest("Implement the requested source.", [Path(source.location)], []),
        original_request="Implement the requested source.",
        implementation_prompt="Implement the requested source.",
        design_plan={"new_files": []},
        editable_sources=[source],
        readonly_sources=[],
        work_needed_by_location={source.location: "Implement the source."},
    )

    assert decision() is True
    assert evidence in decision.prompt
    assert ("A standalone line containing only `⋮`" in decision.prompt) is segmented