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

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

from __future__ import annotations

from band_aider.judgement.test_runner_strategy import TestRunOutcome
from collections.abc import Mapping

from collections.abc import Callable
from band_aider.judgement.source_discovery_strategy import CallTraceNode

import json
import logging
import random
from collections.abc import Callable, Iterator, Mapping, Sequence
from io import StringIO
from pathlib import Path
from typing import TextIO, cast
from unittest.mock import Mock, patch

import pytest

from band_aider.ai_client.openai_compatible_client import OpenAICompatibleClient
from band_aider.ai_client.token_budget_estimator import TokenBudgetEstimator
from band_aider.context.class_meta import ClassMeta, ContentType
from band_aider.context.cot_task_generator import CotTaskRecord
from band_aider.context.failing_test import FailingTest
from band_aider.context.edited_test_inventory import EditedTestInventory
from band_aider.context.memento import Memento
from band_aider.context.token_budget_state import TokenBudgetSettings
from band_aider.context import provenance
from band_aider.context.provenance import DeadEndError, Direction, Phase, Provenance
from band_aider.context.working_backlog import FailedQAItem, InspectionBacklog, TestingBacklog
from band_aider.judgement.direct_code_editor import DirectCodeEditor
from band_aider.judgement.dead_end_decision import DeadEndDecisionBuilder
from band_aider.judgement.edit_request import (
    AiderEditRequest,
    AiderEditRequestMask,
    AiderFailureMode,
)
from band_aider.judgement.file_patcher import FilePatcher, QuickRetryError
from band_aider.judgement.file_manager import FileManager
from band_aider.judgement.code_parser import CodeParser
from band_aider.judgement.dependency_discovery_strategy import DependencyDiscoveryStrategy
from band_aider.judgement.contextual_file_renderer import ContextualFileRenderer
from band_aider.judgement.prompt_localization import PromptLocalization
from band_aider.judgement.search_replace_token_generator import SearchReplaceTokenGenerator
from band_aider.judgement.source_discovery_strategy import (
    SearchableBeatValue,
    SourceDiscoveryStrategy,
)
from band_aider.judgement.prompt_formatter import MarkdownPromptFormatterBuilder
from band_aider.strategies.compiler import (
    BlackboardCompiler,
    CompileAttempt,
    Compiler,
    CompileError,
    CompilerFactory,
    ReferenceError,
    StaticCompileError,
    TestRunCompileError,
    TypeCheckError,
)
from band_aider.strategies.language_strategy_set import LanguageStrategySet
from band_aider.strategies.directives import CustomDirectives
from band_aider.strategies.python.python_compiler import PythonCompiler, PythonCompilerFactory
from band_aider.strategies.python.python_bom_editor import PythonBomEditor
from band_aider.strategies.python.python_source_discovery import PythonSourceDiscovery
from band_aider.judgement.test_runner_strategy import TestRunResult, TestRunnerStrategy as _TestRunnerStrategy
from band_aider.loop.beta_loop import BetaLoop
from band_aider.loop.alpha_loop import AlphaLoop
from band_aider.ui.dto.supreme_status import SupremeStatusDTO
from band_aider.ui.status import status_publisher_scope
from band_aider.ui.status_publisher import StatusPublisher
from band_aider.strategies.oop_signature_extract import (
    Declaration,
    DeclarationType,
    OOpSignature,
    OOpSignatureExtract,
    OOpSignatureRender,
)
from tests.support.completion_clients import (
    ScriptedCompletionClient,
    StaticOpenAICompatibleClient,
)
from tests.support.memory_path import MemoryPath
from tests.support.memory_repo_exec import MemoryRepoExec
from band_aider.judgement.code_editor import CodeEditorResult, EditorRecord


class _AlphaContext:
    """Test workflow context with Alpha's frontier requirement."""


class _ReferenceCardExtract(OOpSignatureExtract):
    """Return current-source-derived cards for renderer-owned prompt tests."""

    def extract(self, module: str, name: str) -> OOpSignature | None:
        del module, name
        return None

    def extract_references(
        self,
        sources_by_location: Mapping[str, str],
    ) -> tuple[OOpSignature, ...]:
        source = "\n".join(sources_by_location.values())
        if "internal_mismatch" in source:
            argument = Declaration(DeclarationType.ARGUMENT, "location")
            return (
                OOpSignature(
                    "tests.support.file_manager",
                    "_FileManager",
                    Declaration(
                        DeclarationType.CLASS,
                        "_FileManager",
                        declarations=(
                            Declaration(
                                DeclarationType.METHOD,
                                "source_path",
                                arguments=(Declaration(DeclarationType.ARGUMENT, "self"), argument),
                            ),
                        ),
                    ),
                ),
                OOpSignature(
                    "pkg.file_manager",
                    "FileManager",
                    Declaration(DeclarationType.CLASS, "FileManager"),
                ),
            )
        if "rejected_dependency" in source:
            return (
                OOpSignature(
                    "rejected.types",
                    "RejectedType",
                    Declaration(DeclarationType.CLASS, "RejectedType"),
                ),
            )
        if "accepted_dependency" in source:
            return (
                OOpSignature(
                    "accepted.types",
                    "AcceptedType",
                    Declaration(DeclarationType.CLASS, "AcceptedType"),
                ),
            )
        return ()

    def extract_source(
        self,
        location: str,
        source: str,
    ) -> tuple[OOpSignature, ...]:
        """Return no current-file cards outside the focused renderer tests."""

        del location, source
        return ()


class _ReferenceCardRender(OOpSignatureRender):
    """Render test cards without coupling editor tests to Python source syntax."""

    def render(self, signature: OOpSignature, output: StringIO) -> None:
        """Write a stable declaration label for one test reference card."""

        output.write(f"{signature.declaration.decl_type.value} {signature.name}\n")


class _ReferenceSelectionClient(StaticOpenAICompatibleClient):
    """Return every listed test card through the production selection response contract."""

    def __init__(self) -> None:
        """Initialize a nominal client without an external endpoint."""

        super().__init__("")

    def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
        """Select only module/name pairs explicitly present in one candidate catalogue."""

        del logger
        # Source chunks use a different response contract from dependency cards.
        if "BEGIN CONTEXTUAL FILE SELECTION" in prompt:
            return "NO_KEEP\nREASON: no additional chunks"
        candidates = [
            (module, name)
            for module, name in (
                ("tests.support.file_manager", "_FileManager"),
                ("pkg.file_manager", "FileManager"),
                ("accepted.types", "AcceptedType"),
                ("rejected.types", "RejectedType"),
            )
            if f"REFERENCE: {module}.{name}" in prompt
        ]
        return json.dumps({
            "references": [
                {"module": module, "name": name}
                for module, name in candidates
            ],
        })


class _CotSelectionClient(_ReferenceSelectionClient):
    """Select one prior COT identity and retain its exact relevance prompt.

    Example: ``_CotSelectionClient().complete(prompt)`` selects COT-0000.
    """

    def __init__(self) -> None:
        """Initialize the reference selector and an empty COT prompt history."""

        super().__init__()
        self.cot_prompts: list[str] = []

    def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
        """Select COT-0000 or delegate dependency-card selection to the parent fake."""

        # Keep prior-COT selection separate from dependency-reference selection.
        if "CONTEXTUAL COT SELECTION" in prompt:
            self.cot_prompts.append(prompt)
            return "KEEP: COT-0000"
        return super().complete(prompt, logger=logger)


@pytest.fixture(autouse=True)
def _isolate_editor_globals() -> Iterator[None]:
    """Restore process-wide strategy selection and a deterministic random stream."""

    original_strategy = LanguageStrategySet.instance
    random.seed(0)
    yield
    LanguageStrategySet.set_instance(original_strategy)


@pytest.fixture
def memory_root() -> MemoryPath:
    """Return a fresh lexical repository that never touches the host disk."""

    return MemoryPath.from_repo(MemoryRepoExec())


class _StubGenerator(SearchReplaceTokenGenerator):
    """Return configured model output while recording repair feedback."""

    def __init__(self, responses: str | list[str]) -> None:
        """Initialize a production-bound generator with stationary responses."""

        formatter = MarkdownPromptFormatterBuilder()
        super().__init__(
            clients=(_ReferenceSelectionClient(),),
            repo_root=Path("/testbed"),
            repo_exec=MemoryRepoExec(),
            prompt_formatter_builder=formatter,
        )
        self.responses = [responses] if isinstance(responses, str) else list(responses)
        self.calls: list[dict[str, object]] = []
        self.render_calls: list[dict[str, object]] = []

    def render_section(
        self,
        name: str,
        *,
        depth: int = 1,
        escape: bool = False,
    ):
        """Generate retry feedback through the same formatter contract as production."""

        return self.prompt_formatter_builder.generate(
            name,
            depth=depth,
            escape=escape,
        )

    def render_prompt(
        self,
        request: AiderEditRequest | AiderEditRequestMask,
        *,
        editable_files: list[Path],
        readonly_files: list[Path],
        renderer: ContextualFileRenderer,
        repair_feedback: str = "",
    ) -> str:
        """Record one baseline render for a Direct edit attempt."""

        self.render_calls.append(
            {
                "request": request,
                "editable_files": list(editable_files),
                "readonly_files": list(readonly_files),
                "renderer": renderer,
                "repair_feedback": repair_feedback,
            }
        )
        return "baseline prompt"

    def generate(
        self,
        request: AiderEditRequest,
        *,
        base_prompt: str,
        editable_files: list[Path],
        logger: logging.Logger | None = None,
        unapplied_attempts: int = 0,
    ) -> str:
        self.calls.append(
            {
                "request": request,
                "base_prompt": base_prompt,
                "editable_files": list(editable_files),
                "logger": logger,
                "unapplied_attempts": unapplied_attempts,
            }
        )
        if len(self.responses) > 1:
            return self.responses.pop(0)
        return self.responses[0]


class _RenderingStubGenerator(_StubGenerator):
    """Record prompts after rendering through the production evidence boundary."""

    def render_prompt(
        self,
        request: AiderEditRequest | AiderEditRequestMask,
        *,
        editable_files: list[Path],
        readonly_files: list[Path],
        renderer: ContextualFileRenderer,
        repair_feedback: str = "",
    ) -> str:
        """Render with the real generator while retaining the scenario inputs."""

        prompt = SearchReplaceTokenGenerator.render_prompt(
            self,
            request,
            editable_files=editable_files,
            readonly_files=readonly_files,
            renderer=renderer,
            repair_feedback=repair_feedback,
        )
        self.render_calls.append(
            {
                "request": request,
                "editable_files": list(editable_files),
                "readonly_files": list(readonly_files),
                "renderer": renderer,
                "repair_feedback": repair_feedback,
            }
        )
        return prompt


class _EditorParser(CodeParser):
    """Provide FileManager's parser contract without opening host files."""

    def parse_file(
        self,
        file_name: str,
        prompt: str,
        content_type: ContentType = ContentType.Source,
    ) -> ClassMeta:
        """Return one minimal source record for an in-memory path."""

        del prompt
        return ClassMeta(file_name, "", [], {}, content_type=content_type)

    def parse_stream(
        self,
        file_name: str,
        stream: TextIO,
        prompt: str,
        content_type: ContentType = ContentType.Source,
    ) -> ClassMeta:
        """Return one minimal source record without consuming host-backed data."""

        del stream
        return self.parse_file(file_name, prompt, content_type)

    def parse_files(
        self,
        file_names: list[str],
        prompt: str,
        content_type: ContentType = ContentType.Source,
    ) -> list[ClassMeta]:
        """Return one minimal source record per requested in-memory path."""

        return [self.parse_file(file_name, prompt, content_type) for file_name in file_names]


class _ManagedFileManager(FileManager):
    """Provide the FileManager/ClassMeta subset needed by the direct editor."""

    def __init__(
        self,
        root: MemoryPath,
        locations: list[str],
        test_locations: Sequence[str] = (),
    ) -> None:
        """Seed one nominal FileManager with explicit in-memory metadata."""

        self._source_locations = list(locations)
        self._test_locations = list(test_locations)
        self.metadata_requests: list[tuple[str, ContentType | None]] = []
        self.refresh_requests: list[tuple[ClassMeta, str, str]] = []
        super().__init__(
            parser=_EditorParser(),
            client=StaticOpenAICompatibleClient('{"cdc": ""}'),
            repo_root=root,
            src_root=root / "src",
            test_root=root / "tests",
            actor="test",
            prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
            repo_exec=root.repository,
            prompt="Original stabilization goal.",
        )

    def resync(self) -> None:
        """Build FileManager indexes from fixed scenario metadata only."""

        self.src_classes = [
            ClassMeta(
                location=location,
                summary="",
                relevance=[],
                relevant_symbols={},
                content_type=ContentType.Source,
            )
            for location in self._source_locations
        ]
        self.test_classes = [
            ClassMeta(
                location=location,
                summary="",
                relevance=[],
                relevant_symbols={},
                content_type=ContentType.Test,
            )
            for location in self._test_locations
        ]
        self.src_index = {item.location: index for index, item in enumerate(self.src_classes)}
        self.test_index = {item.location: index for index, item in enumerate(self.test_classes)}
        self.src_dirty.clear()
        self.test_dirty.clear()

    def git_diff_for(self, location: str | Path) -> str:
        """Direct editor tests do not need repository diff evidence."""

        del location
        return ""

    def current_metadata(
        self,
        location: str | Path,
        *,
        content_type: ContentType | None = None,
    ) -> ClassMeta | None:
        """Return this scenario's canonical metadata within the requested content kind."""

        path = Path(str(location))
        relative = path.relative_to(self.repo_root).as_posix() if path.is_absolute() else path.as_posix()
        self.metadata_requests.append((relative, content_type))
        # Match the requested metadata kind at the production factory boundary.
        if content_type is ContentType.Test:
            metadata = self.current_test(relative)
        # A source-only lookup cannot accidentally select a test record.
        elif content_type is ContentType.Source:
            metadata = self.current_src(relative)
        # Unqualified editor captures may select either managed content kind.
        else:
            metadata = self.current_src(relative) or self.current_test(relative)
        return metadata

    def refresh_class_meta(
        self,
        class_meta: ClassMeta,
        prompt: str,
        tailing_logs: str = "",
    ) -> ClassMeta:
        """Return the test fixture's canonical metadata for a stale repair file."""

        self.refresh_requests.append((class_meta, prompt, tailing_logs))
        return self.current_metadata(
            class_meta.location,
            content_type=class_meta.content_type,
        ) or class_meta


class _RecordingCompiler(BlackboardCompiler):
    """Compile one recorded attempt from a stationary diagnostics queue."""

    def __init__(self, factory: "_RecordingCompilerFactory", attempt: CompileAttempt) -> None:
        """Bind one compiler invocation to its factory-owned attempt evidence."""

        self._factory = factory
        self._attempt = attempt
        self.blackboard = attempt.blackboard or EditedTestInventory(
            self._clean_blackboard
        )

    def get_blackboard(self) -> EditedTestInventory:
        """Return the inventory owned or borrowed by this recorded compiler."""

        return self.blackboard

    def _clean_blackboard(self) -> None:
        """Record compiler-owned inventory release by forgetting its handle."""

        del self.blackboard

    def compile(self, paths: Sequence[Path]) -> list[CompileError]:
        """Record current Memento state and return the next configured diagnostics."""

        call = {
            "paths": list(paths),
            "pre": [memento.pre for memento in self._attempt.mementos],
            "post": [memento.post for memento in self._attempt.mementos],
            "layers": [memento.layer for memento in self._attempt.mementos],
        }
        self._factory.compile_calls.append(call)
        self._factory.candidate_calls.append(
            {
                "pre": [memento.pre for memento in self._attempt.candidate_mementos],
                "post": [memento.post for memento in self._attempt.candidate_mementos],
                "previous_errors": list(self._attempt.previous_errors),
            }
        )
        pre = call["pre"]
        post = call["post"]
        if all(before == after for before, after in zip(pre, post)):
            return self._factory.baseline_results.pop(0) if self._factory.baseline_results else []
        return self._factory.results.pop(0) if self._factory.results else []


class _RecordingCompilerFactory(CompilerFactory):
    """Record CompileAttempt Mementos and return configured diagnostics."""

    def __init__(self, results: Sequence[Sequence[CompileError]] | None = None) -> None:
        self.results = [list(result) for result in results or [[]]]
        self.baseline_results: list[list[CompileError]] = []
        self.build_calls: list[CompileAttempt] = []
        self.compile_calls: list[dict[str, object]] = []
        self.candidate_calls: list[dict[str, object]] = []
        self.compilers: list[_RecordingCompiler] = []

    def build(self, attempt: CompileAttempt) -> Compiler:
        self.build_calls.append(attempt)
        compiler = _RecordingCompiler(self, attempt)
        self.compilers.append(compiler)
        return compiler


class _EmptySourceDiscovery(SourceDiscoveryStrategy):
    """Provide the unrelated discovery contract for direct-editor scenarios."""

    def visit_call_sites(self, target: CallTraceNode, already_seen: set[str] | None = None, predicate: Callable[[CallTraceNode], bool] | None = None, *, target_args: dict[str, CallTraceNode] | None = None, pristine: Mapping[str, str] | None = None) -> None:
        """Supply no imported calls in this stationary discovery double."""
        del target, already_seen, predicate

    def get_call_tree(self, location: str, name: str | None = None, *, line: int | None = None, depth: int = 1, already_seen: set[str] | None = None, predicate: Callable[[CallTraceNode], bool] | None = None, pristine: Mapping[str, str] | None = None) -> CallTraceNode:
        """Reject origins because this double contains no callable definitions."""
        raise LookupError(location)

    def get_sources(self, recursive: bool = True, type: object | None = None) -> list[Path]:
        """Return no discovery-owned source paths."""

        del recursive, type
        return []

    def get_tests(self, recursive: bool = True, type: object | None = None) -> list[Path]:
        """Return no discovery-owned test paths."""

        del recursive, type
        return []

    def searchable_beats(self, file_name: str, source: str) -> list[SearchableBeatValue]:
        """Return no searchable beats for the fixed direct-editor fixture."""

        del file_name, source
        return []


class _FixedStrategySet:
    """Return one configured language strategy set without mutation helpers."""

    def __init__(self, strategy_set: LanguageStrategySet) -> None:
        """Store the one scenario-local strategy set."""

        self._strategy_set = strategy_set

    def __call__(self, file_manager: FileManager, **overrides: object) -> LanguageStrategySet:
        """Return this scenario's strategy set for any bound file manager."""

        del file_manager, overrides
        return self._strategy_set


class _RecordingPublisher(StatusPublisher):
    """Retain status events through the nominal UI publication boundary."""

    def __init__(self, events: list[object] | None = None) -> None:
        """Use supplied event storage or create an isolated scenario-local list."""

        self.events = events if events is not None else []

    def publish(self, event: object) -> None:
        """Store one immutable event for behavior assertions."""

        self.events.append(event)


def _install_strategy(
    compiler_factory: CompilerFactory,
    signature_extract: OOpSignatureExtract | None = None,
) -> None:
    """Install minimal neutral contracts for one focused editor test."""

    strategy_set = LanguageStrategySet(
        bindings={
            CompilerFactory: compiler_factory,
            SourceDiscoveryStrategy: _EmptySourceDiscovery(),
            OOpSignatureExtract: signature_extract or _ReferenceCardExtract(),
            OOpSignatureRender: _ReferenceCardRender(),
        }
    )
    LanguageStrategySet.set_instance(_FixedStrategySet(strategy_set))


@pytest.fixture
def compiler_factory() -> _RecordingCompilerFactory:
    """Register a deterministic successful compiler."""

    factory = _RecordingCompilerFactory()
    _install_strategy(factory)
    return factory


def _replace(path: str, before: str, after: str) -> str:
    """Render one minimal SEARCH/REPLACE block."""

    return "\n".join(
        [
            f"FILE: {path}",
            "<<<<<<< SEARCH",
            before,
            "=======",
            after,
            ">>>>>>> REPLACE",
        ]
    )


def _runtime_editor(
    memory_root: MemoryPath,
    generator: _StubGenerator,
) -> tuple[DirectCodeEditor, Path]:
    """Create an editor backed by one discovered source file."""

    source = memory_root / "src" / "mod.py"
    source.parent.mkdir(parents=True)
    source.write_text("value = 'old'\n", encoding="utf-8")
    editor = DirectCodeEditor(
        token_generator=generator,
        prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
        file_patcher=FilePatcher(memory_root, repo_exec=memory_root.repository),
        repo_root=memory_root,
    )
    editor.file_manager = _ManagedFileManager(memory_root, ["src/mod.py"])
    return editor, source


def _generator(editor: DirectCodeEditor) -> _StubGenerator:
    """Return the nominal stationary generator bound by this test harness."""

    generator = editor.token_generator
    if not isinstance(generator, _StubGenerator):
        raise AssertionError("Direct editor did not retain the stationary test generator.")
    return generator


def _file_manager(editor: DirectCodeEditor) -> _ManagedFileManager:
    """Return the nominal in-memory FileManager bound by this test harness."""

    manager = editor.file_manager
    if not isinstance(manager, _ManagedFileManager):
        raise AssertionError("Direct editor did not retain the in-memory test FileManager.")
    return manager


def _source_meta(editor: DirectCodeEditor, location: str) -> ClassMeta:
    """Return one required source record from the nominal in-memory manager."""

    class_meta = _file_manager(editor).current_src(location)
    if class_meta is None:
        raise AssertionError(f"Scenario did not seed source metadata for {location}.")
    return class_meta


def _rendered_masks(editor: DirectCodeEditor) -> list[AiderEditRequestMask]:
    """Return the explicitly projected request masks recorded by the generator."""

    masks: list[AiderEditRequestMask] = []
    for call in _generator(editor).render_calls:
        request = call["request"]
        if not isinstance(request, AiderEditRequestMask):
            raise AssertionError("Direct editor rendered an unmasked request.")
        masks.append(request)
    return masks


def _rendered_renderer(editor: DirectCodeEditor, index: int) -> ContextualFileRenderer:
    """Return one nominal contextual renderer captured for a direct-edit turn."""

    renderer = _generator(editor).render_calls[index]["renderer"]
    if not isinstance(renderer, ContextualFileRenderer):
        raise AssertionError("Direct editor did not render through ContextualFileRenderer.")
    return renderer


def _request(source: Path, context: type = _AlphaContext) -> AiderEditRequest:
    """Create one workflow-aware request with the appropriate frontier scope."""

    return AiderEditRequest(
        authoritative_prompt="Update the value.",
        editable_files=[source],
        readonly_files=[],
        context=context,
        frontier_provider=lambda: [[]],
    )


def test_direct_editor_compiles_memento_pre_and_post_then_commits(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """A clean patch is compiled from the editor-owned ClassMeta Memento."""

    editor, source = _runtime_editor(
        memory_root,
        _StubGenerator(_replace("src/mod.py", "value = 'old'", "value = 'new'")),
    )

    _edit_result = editor.edit(_request(source))
    changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    assert compiler_factory.compile_calls == [
        {
            "paths": [source],
            "pre": ["value = 'old'\n"],
            "post": ["value = 'old'\n"],
            "layers": [0],
        },
        {
            "paths": [source],
            "pre": ["value = 'old'\n"],
            "post": ["value = 'new'\n"],
            "layers": [0],
        }
    ]
    assert compiler_factory.build_calls[0].mementos[0].file_path == source
    assert compiler_factory.candidate_calls == [
        {"pre": [], "post": [], "previous_errors": []},
        {
            "pre": ["value = 'old'\n"],
            "post": ["value = 'new'\n"],
            "previous_errors": [],
        },
    ]
    assert _file_manager(editor).metadata_requests == [
        ("src/mod.py", None),
        ("src/mod.py", None),
    ]
    assert not (source.parent / ".mod.py~").exists()


def test_direct_editor_repairs_static_failure_without_dead_end_admission(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """A failed static candidate continues repair before any restart admission."""

    generator = _StubGenerator([
        _replace("src/mod.py", "value = 'old'", "value = 'partial'"),
        _replace("src/mod.py", "value = 'partial'", "value = 'done'"),
    ])
    boundary_client = ScriptedCompletionClient((
        json.dumps({
            "dead_end": True,
            "scope_fault": "wrong_editable_file",
            "message": "Reconsider the selected source remediation.",
            "justification": "The failure evidence indicates the selected source scope is wrong.",
        }),
    ))
    generator.clients = (boundary_client,)
    editor, source = _runtime_editor(memory_root, generator)
    compiler_factory.results = [[TypeCheckError(
        "candidate static failure",
        path=source,
        line=7,
        column=3,
    )], []]
    graph = Provenance.open(repo_root=memory_root, repo_exec=memory_root.repository)
    item = InspectionBacklog(
        focused_spotlight=_source_meta(editor, "src/mod.py"),
        summary="repair the focused source",
    )
    parent_id = graph.start(Phase.BETA, TestingBacklog(
        focused_spotlight=_source_meta(editor, "src/mod.py"),
        summary="prior test turn",
    ))
    graph.track_provenance(parent_id, Phase.ALPHA, item)

    with provenance.bind(graph):
        with provenance.item_scope(Phase.ALPHA, item):
            _edit_result = editor.edit(_request(source))
            changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    assert boundary_client.prompts == []
    assert len(generator.calls) == 2
    assert "candidate static failure" in str(generator.render_calls[1]["repair_feedback"])
    assert source.read_text(encoding="utf-8") == "value = 'done'\n"
    assert not any(edge.direction is not None for edge in graph.edges)
    assert not Memento.current(source, memory_root.repository)


def test_direct_editor_repairs_runtime_failure_without_dead_end_admission(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """A failed runtime candidate continues repair before any restart admission."""

    generator = _StubGenerator([
        _replace("src/mod.py", "value = 'old'", "value = 'partial'"),
        _replace("src/mod.py", "value = 'partial'", "value = 'done'"),
    ])
    boundary_client = ScriptedCompletionClient((
        json.dumps({
            "dead_end": True,
            "scope_fault": "wrong_editable_file",
            "message": "Reconsider the selected source remediation.",
            "justification": "The targeted failure indicates the selected source scope is wrong.",
        }),
    ))
    generator.clients = (boundary_client,)
    editor, source = _runtime_editor(memory_root, generator)
    compiler_factory.results = [[TestRunCompileError(
        "targeted test still fails",
        test_name="test_value",
    )], []]
    graph = Provenance.open(repo_root=memory_root, repo_exec=memory_root.repository)
    item = InspectionBacklog(
        focused_spotlight=_source_meta(editor, "src/mod.py"),
        summary="repair the focused source",
    )
    parent_id = graph.start(Phase.BETA, TestingBacklog(
        focused_spotlight=_source_meta(editor, "src/mod.py"),
        summary="prior test turn",
    ))
    graph.track_provenance(parent_id, Phase.ALPHA, item)

    with provenance.bind(graph):
        with provenance.item_scope(Phase.ALPHA, item):
            _edit_result = editor.edit(_request(source))
            changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    assert boundary_client.prompts == []
    assert len(generator.calls) == 2
    assert "targeted test still fails" in str(generator.render_calls[1]["repair_feedback"])
    assert source.read_text(encoding="utf-8") == "value = 'done'\n"
    assert not any(edge.direction is not None for edge in graph.edges)
    assert not Memento.current(source, memory_root.repository)


def test_direct_editor_corrects_invalid_no_patch_admission_without_unwinding(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """Invalid admission after NO_PATCH preserves the accepted repair baseline."""

    generator = _StubGenerator([
        _replace("src/mod.py", "value = 'old'", "value = 'partial'"),
        'NO_PATCH: {"file": "src/caller.py", "reason": "The caller still uses the stale API."}',
        _replace("src/mod.py", "value = 'partial'", "value = 'done'"),
    ])
    boundary_client = ScriptedCompletionClient((
        json.dumps({"dead_end": "yes"}),
        json.dumps({
            "dead_end": False,
            "scope_fault": "none",
            "message": "",
            "justification": "",
        }),
    ))
    generator.clients = (boundary_client,)
    editor, source = _runtime_editor(memory_root, generator)
    compiler_factory.results = [[TestRunCompileError(
        "targeted test still fails",
        test_name="test_value",
    )], []]
    graph = Provenance.open(repo_root=memory_root, repo_exec=memory_root.repository)
    trace = [f'File "{memory_root}/src/caller.py", line 9']
    item = FailedQAItem(
        focused_spotlight=_source_meta(editor, "src/mod.py"),
        summary="repair the focused source",
        failure_stack_trace=trace,
        failed_test=FailingTest(ClassMeta("tests/test_value.py", "", [], {}), trace, "Stale caller API"),
    )
    parent_id = graph.start(Phase.BETA, TestingBacklog(
        focused_spotlight=_source_meta(editor, "src/mod.py"),
        summary="prior test turn",
    ))
    graph.track_provenance(parent_id, Phase.ALPHA, item)
    request = _request(source)
    request.is_remediation = True
    request.failure_error_message = "Stale caller API"
    request.failure_stack_trace = list(trace)

    with provenance.bind(graph):
        with provenance.item_scope(Phase.ALPHA, item):
            _edit_result = editor.edit(request)
            changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    assert source.read_text(encoding="utf-8") == "value = 'done'\n"
    assert len(generator.calls) == 3
    assert len(boundary_client.prompts) == 2
    assert "RESPONSE VALIDATION ERROR" in boundary_client.prompts[1]
    assert not any(edge.direction is not None for edge in graph.edges)
    assert not Memento.current(source, memory_root.repository)


def test_direct_editor_publishes_progress_without_extra_compilation_or_work_queue(
    memory_root: MemoryPath,
    caplog,
) -> None:
    """Compiler facts travel through SupremeStatusDTO while the established repair sequence remains intact."""

    factory = _RecordingCompilerFactory(
        results=[
            [
                TestRunCompileError("render failed", test_name="TestWidget.test_render"),
                TestRunCompileError("runner failed"),
            ],
            [],
        ]
    )
    _install_strategy(factory)
    editor, source = _runtime_editor(
        memory_root,
        _StubGenerator(
            [
                _replace("src/mod.py", "value = 'old'", "value = 'new'"),
                _replace("src/mod.py", "value = 'new'", "value = 'final'"),
            ]
        ),
    )

    class _Publisher(StatusPublisher):
        """Collect only UI events from the existing status publisher path."""

        def __init__(self) -> None:
            self.events: list[object] = []

        def publish(self, event: object) -> None:
            self.events.append(event)

    publisher = _Publisher()
    logger = logging.getLogger("tests.direct-code-editor-progress")
    caplog.set_level(logging.INFO, logger=logger.name)
    with status_publisher_scope(publisher):
        _edit_result = editor.edit(_request(source), logger=logger)
        changed, _ = _edit_result.changed_paths, _edit_result.summary

    status_events = [event for event in publisher.events if isinstance(event, SupremeStatusDTO)]
    targeted = [
        event.direct_code_editor
        for event in status_events
        if event.direct_code_editor is not None
        and event.direct_code_editor.unresolved_targeted_test_failure_count == 2
    ]

    assert changed == [source]
    assert len(factory.compile_calls) == 3
    assert len(_generator(editor).calls) == 2
    assert all(isinstance(event, SupremeStatusDTO) for event in publisher.events)
    assert targeted[-1].model_visible_compiler_error_count == 1
    assert targeted[-1].named_targeted_failures == ("FAIL TestWidget.test_render",)
    assert targeted[-1].unattributed_targeted_execution_failure_count == 1
    assert status_events[-1].direct_code_editor is None
    progress_messages = [
        record.getMessage()
        for record in caplog.records
        if record.getMessage().startswith("DirectCodeEditor progress:")
    ]
    assert any(
        "phase=initial compile" in message and "blocking_remaining=0" in message
        for message in progress_messages
    )
    assert any(
        "named_targeted_failures=('FAIL TestWidget.test_render',)" in message
        for message in progress_messages
    )
    assert progress_messages[-1] == "DirectCodeEditor progress: inactive"


def test_progress_retains_named_failures(memory_root: MemoryPath) -> None:
    """Status retains the 101st named targeted failure instead of losing it at the former limit."""

    test_names = tuple(f"TestWidget.test_{index}" for index in range(101))
    targeted_errors = list(
        map(
            lambda test_name: TestRunCompileError("render failed", test_name=test_name),
            test_names,
        )
    )
    factory = _RecordingCompilerFactory(results=[targeted_errors, []])
    _install_strategy(factory)
    editor, source = _runtime_editor(
        memory_root,
        _StubGenerator(
            [
                _replace("src/mod.py", "value = 'old'", "value = 'new'"),
                _replace("src/mod.py", "value = 'new'", "value = 'final'"),
            ]
        ),
    )

    class _Publisher(StatusPublisher):
        """Collect status events emitted through the existing publisher boundary."""

        def __init__(self) -> None:
            self.events: list[object] = []

        def publish(self, event: object) -> None:
            self.events.append(event)

    publisher = _Publisher()
    with status_publisher_scope(publisher):
        editor.edit(_request(source))

    status_events = [event for event in publisher.events if isinstance(event, SupremeStatusDTO)]
    targeted = [
        event.direct_code_editor
        for event in status_events
        if event.direct_code_editor is not None
        and event.direct_code_editor.unresolved_targeted_test_failure_count == len(test_names)
    ]

    assert targeted[-1].named_targeted_failures == tuple(
        f"FAIL {test_name}" for test_name in test_names
    )


def test_direct_editor_consumes_no_patch_protocol_before_returning_its_summary(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """NO_PATCH is direct-editor control state, never downstream Alpha or Beta evidence."""

    editor, source = _runtime_editor(
        memory_root,
        _StubGenerator("NO_PATCH: no concrete divergence exists in the focused source"),
    )

    _edit_result = editor.edit(_request(source))
    changed, summary = _edit_result.changed_paths, _edit_result.summary

    assert changed == []
    assert "NO_PATCH:" not in summary
    assert "no concrete divergence exists" in summary
    assert compiler_factory.compile_calls


@pytest.mark.parametrize("response", ["NO_PATCH: production caller requires repair", "SUMMARY: unable to finish"])
def test_abandoned_edit_returns_candidate_diff_and_failure_without_changing_identity(
    memory_root: MemoryPath, compiler_factory: _RecordingCompilerFactory, response: str,
) -> None:
    """An abandoned test migration restores bytes while its caller receives the production blocker and patch."""

    generator = _StubGenerator([
        _replace("src/mod.py", "value = 'old'", "value = 'active'"), response,
    ])
    editor, source = _runtime_editor(memory_root, generator)
    failures: list[CompileError] = [TestRunCompileError("Production caller uses removed clear()", test_name="test_resume")]
    compiler_factory.results = [failures]
    graph = Provenance.open(repo_root=memory_root, repo_exec=memory_root.repository)
    item = TestingBacklog(_source_meta(editor, "src/mod.py"), "Migrate test setup")
    identity = graph.start(Phase.BETA, item)
    with provenance.bind(graph), provenance.item_scope(Phase.BETA, item):
        _edit_result = editor.edit(_request(source))
        changed, summary = _edit_result.changed_paths, _edit_result.summary

    assert changed == []
    assert source.read_text() == "value = 'old'\n"
    assert "-value = 'old'" in summary
    assert "+value = 'active'" in summary
    assert "Production caller uses removed clear()" in summary
    assert item.message_id == identity
    assert tuple(graph.messages) == (identity,)
    assert graph.edges == ()
    assert Memento.current(source, memory_root.repository) is None


@pytest.mark.parametrize("response", ["NO_PATCH: looks correct", "", "SUMMARY: nothing to do"])
@pytest.mark.parametrize("remediation", [False, True])
def test_noop_response_keeps_only_remediation_in_the_editor(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
    response: str,
    remediation: bool,
) -> None:
    """The same no-op completes ordinary work but must retry an incoming failure."""

    generator = _StubGenerator([response, _replace("src/mod.py", "value = 'old'", "value = 'fixed'")])
    editor, source = _runtime_editor(memory_root, generator)
    request = _request(source)
    request.is_remediation = remediation
    request.failure_error_message = "Expected fixed value" if remediation else ""

    _edit_result = editor.edit(request)
    changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == ([source] if remediation else [])
    assert source.read_text() == ("value = 'fixed'\n" if remediation else "value = 'old'\n")
    assert len(generator.calls) == (2 if remediation else 1)
    assert not Memento.current(source, memory_root.repository)
    if remediation:
        assert "No restart was admitted" in str(generator.render_calls[1]["repair_feedback"])


def test_remediation_unchanged_patch_requires_another_attempt(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """A mechanically valid unchanged patch cannot consume an incoming failure."""

    generator = _StubGenerator([
        _replace("src/mod.py", "value = 'old'", "value = 'old'"),
        _replace("src/mod.py", "value = 'old'", "value = 'fixed'"),
    ])
    editor, source = _runtime_editor(memory_root, generator)
    request = _request(source)
    request.is_remediation = True
    request.failure_error_message = "Expected fixed value"

    _edit_result = editor.edit(request)
    changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    assert source.read_text() == "value = 'fixed'\n"
    assert len(generator.calls) == 2
    assert "no net change" in str(generator.render_calls[1]["repair_feedback"])
    assert not Memento.current(source, memory_root.repository)


@pytest.mark.parametrize("admitted", [False, True])
def test_remediation_traceback_attribution_requires_dead_end_admission(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
    admitted: bool,
) -> None:
    """NO_PATCH after a runtime failure either revives Alpha or keeps the repair baseline."""

    attribution = {"file": "src/caller.py", "reason": "The traceback locates a stale call in src/caller.py."}
    generator = _StubGenerator([
        _replace("src/mod.py", "value = 'old'", "value = 'partial'"),
        "NO_PATCH: " + json.dumps(attribution),
        _replace("src/mod.py", "value = 'partial'", "value = 'fixed'"),
    ])
    boundary_client = ScriptedCompletionClient((json.dumps({
        "dead_end": admitted,
        "scope_fault": "wrong_editable_file" if admitted else "none",
        "message": "Reconsider the attribution." if admitted else "",
        "justification": "The caller owns the stale call." if admitted else "",
    }),))
    generator.clients = (boundary_client,)
    editor, source = _runtime_editor(memory_root, generator)
    compiler_factory.results = [[TestRunCompileError(
        "The caller still uses the stale API.",
        test_name="test_value",
    )], []]
    graph = Provenance.open(repo_root=memory_root, repo_exec=memory_root.repository)
    parent = graph.start(Phase.ALPHA, InspectionBacklog(_source_meta(editor, "src/mod.py"), "Inspect source."))
    trace = [f'File "{memory_root}/src/caller.py", line 9']
    failure = FailingTest(ClassMeta("tests/test_value.py", "", [], {}), trace, "Stale caller API")
    item = FailedQAItem(
        _source_meta(editor, "src/mod.py"), "Repair source.",
        failure_stack_trace=trace, failed_test=failure,
    )
    graph.track_provenance(parent, Phase.ALPHA, item)
    request = _request(source)
    request.is_remediation = True
    request.failure_error_message = "Stale caller API"
    request.failure_stack_trace = list(item.failure_stack_trace)

    with provenance.bind(graph), provenance.item_scope(Phase.ALPHA, item):
        # Only an admitted attribution may discard the entire repair episode.
        if admitted:
            with pytest.raises(DeadEndError) as raised:
                editor.edit(request)
            phase, revived = graph.revive(raised.value)
            assert phase is Phase.ALPHA
            assert revived.message_id != parent
            assert graph.messages[revived.message_id].parent_id == graph.messages[parent].parent_id
            assert raised.value.direction is Direction.ROLLBACK
            assert attribution["reason"] in revived.future_guidance[-1].message
            assert "-value = 'old'" in revived.future_guidance[-1].message
            assert "+value = 'partial'" in revived.future_guidance[-1].message
            assert source.read_text() == "value = 'old'\n"
        else:
            _edit_result = editor.edit(request)
            changed, _ = _edit_result.changed_paths, _edit_result.summary
            assert changed == [source]
            assert source.read_text() == "value = 'fixed'\n"
            assert not any(edge.direction is not None for edge in graph.edges)
            assert "No restart was admitted" in str(generator.render_calls[2]["repair_feedback"])
    assert len(generator.calls) == (2 if admitted else 3)
    assert len(boundary_client.prompts) == 1
    assert "src/caller.py" in boundary_client.prompts[0]
    assert not Memento.current(source, memory_root.repository)


@pytest.mark.parametrize("response", ["", "SUMMARY: nothing to do"])
def test_remediation_empty_response_does_not_request_dead_end_admission(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
    response: str,
) -> None:
    """Empty output retries remediation without proposing a historical restart."""

    generator = _StubGenerator([
        response,
        _replace("src/mod.py", "value = 'old'", "value = 'fixed'"),
    ])
    editor, source = _runtime_editor(memory_root, generator)
    graph = Provenance.open(repo_root=memory_root, repo_exec=memory_root.repository)
    parent = graph.start(Phase.BETA, TestingBacklog(_source_meta(editor, "src/mod.py"), "Test source."))
    item = FailedQAItem(_source_meta(editor, "src/mod.py"), "Repair source.")
    graph.track_provenance(parent, Phase.ALPHA, item)
    request = _request(source)
    request.is_remediation = True
    request.failure_error_message = "Expected fixed value"

    with provenance.bind(graph), provenance.item_scope(Phase.ALPHA, item):
        with patch.object(
            DeadEndDecisionBuilder,
            "for_direct_code_editor",
            wraps=DeadEndDecisionBuilder.for_direct_code_editor,
        ) as admission:
            _edit_result = editor.edit(request)
            changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    assert source.read_text() == "value = 'fixed'\n"
    assert len(generator.calls) == 2
    assert "No restart was admitted" in str(generator.render_calls[1]["repair_feedback"])
    admission.assert_not_called()
    assert not any(edge.direction is not None for edge in graph.edges)
    assert not Memento.current(source, memory_root.repository)


@pytest.mark.parametrize("file", ["src/missing.py", "src/mod.py", "../outside.py"])
def test_remediation_rejects_attribution_outside_its_provenance(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
    file: str,
) -> None:
    """A named file alone cannot authorize leaving the current remediation."""

    generator = _StubGenerator([
        "NO_PATCH: " + json.dumps({"file": file, "reason": "The fault is in this file."}),
        _replace("src/mod.py", "value = 'old'", "value = 'fixed'"),
    ])
    boundary_client = ScriptedCompletionClient(())
    generator.clients = (boundary_client,)
    editor, source = _runtime_editor(memory_root, generator)
    graph = Provenance.open(repo_root=memory_root, repo_exec=memory_root.repository)
    parent = graph.start(Phase.BETA, TestingBacklog(_source_meta(editor, "src/mod.py"), "Test source."))
    item = FailedQAItem(_source_meta(editor, "src/mod.py"), "Repair source.")
    graph.track_provenance(parent, Phase.ALPHA, item)
    request = _request(source)
    request.is_remediation = True
    request.failure_error_message = "Expected fixed value"

    with provenance.bind(graph), provenance.item_scope(Phase.ALPHA, item):
        _edit_result = editor.edit(request)
        changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    assert len(generator.calls) == 2
    assert boundary_client.prompts == []
    assert not any(edge.direction is not None for edge in graph.edges)


def test_remediation_attribution_without_history_retries_locally(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """Recorded attribution cannot leave remediation when no parent can be revived."""

    generator = _StubGenerator([
        'NO_PATCH: {"file": "src/caller.py", "reason": "The traceback locates the stale call in src/caller.py."}',
        _replace("src/mod.py", "value = 'old'", "value = 'fixed'"),
    ])
    boundary_client = ScriptedCompletionClient(())
    generator.clients = (boundary_client,)
    editor, source = _runtime_editor(memory_root, generator)
    graph = Provenance.open(repo_root=memory_root, repo_exec=memory_root.repository)
    trace = [f'File "{memory_root}/src/caller.py", line 9']
    failure = FailingTest(ClassMeta("tests/test_value.py", "", [], {}), trace, "Stale caller API")
    item = FailedQAItem(
        _source_meta(editor, "src/mod.py"), "Repair source.",
        failure_stack_trace=trace, failed_test=failure,
    )
    graph.start(Phase.ALPHA, item)
    request = _request(source)
    request.is_remediation = True
    request.failure_error_message = "Stale caller API"
    request.failure_stack_trace = list(trace)

    with provenance.bind(graph), provenance.item_scope(Phase.ALPHA, item):
        _edit_result = editor.edit(request)
        changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    assert source.read_text() == "value = 'fixed'\n"
    assert len(generator.calls) == 2
    assert "no retained parent" in str(generator.render_calls[1]["repair_feedback"])
    assert boundary_client.prompts == []
    assert not any(edge.direction is not None for edge in graph.edges)
    assert not Memento.current(source, memory_root.repository)


def test_remediation_provider_failure_is_not_a_successful_empty_edit(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """Provider interruption preserves the failure and rolls back the attempt."""

    editor, source = _runtime_editor(memory_root, _StubGenerator("unused"))
    request = _request(source)
    request.is_remediation = True
    request.failure_error_message = "Expected fixed value"

    with patch.object(editor.token_generator, "generate", side_effect=ConnectionError("unavailable")):
        with pytest.raises(ConnectionError, match="unavailable"):
            editor.edit(request)

    assert source.read_text() == "value = 'old'\n"
    assert request.failure_error_message == "Expected fixed value"
    assert not Memento.current(source, memory_root.repository)


def test_direct_editor_treats_literal_no_patch_source_text_as_inert(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """Only the direct model-result channel, never source evidence, can activate NO_PATCH."""

    source_text = "value = 'old'  # NO_PATCH: literal source evidence\n"
    editor, source = _runtime_editor(
        memory_root,
        _StubGenerator(
            _replace(
                "src/mod.py",
                source_text.rstrip(),
                "value = 'new'",
            )
        ),
    )
    source.write_text(source_text, encoding="utf-8")

    _edit_result = editor.edit(_request(source))
    changed, summary = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    assert "NO_PATCH: literal source evidence" in _edit_result.records[0].search
    assert "NO_PATCH:" not in _edit_result.records[0].render()
    assert source.read_text(encoding="utf-8") == "value = 'new'\n"
    assert compiler_factory.compile_calls


def test_direct_editor_preserves_caller_memento_and_acceptance_diff(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """A nested editor transaction cannot consume the caller's rollback boundary."""

    editor, source = _runtime_editor(
        memory_root,
        _StubGenerator(_replace("src/mod.py", "value = 'old'", "value = 'new'")),
    )
    caller_memento = _source_meta(editor, "src/mod.py").set_or_get_memento(
        _file_manager(editor),
    )

    _edit_result = editor.edit(_request(source))
    changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    assert compiler_factory.compile_calls[0]["layers"] == [1]
    assert compiler_factory.compile_calls[1]["layers"] == [1]
    assert caller_memento.layer == 0
    assert caller_memento.path.exists()
    assert caller_memento.pre == "value = 'old'\n"
    assert caller_memento.post == "value = 'new'\n"
    assert caller_memento.diff

    acceptance_memento = _source_meta(editor, "src/mod.py").set_or_get_memento(
        _file_manager(editor),
    )
    assert acceptance_memento.layer == caller_memento.layer
    assert acceptance_memento.pre == "value = 'old'\n"
    assert acceptance_memento.post == "value = 'new'\n"
    assert acceptance_memento.diff == caller_memento.diff


def test_caller_memento_rollback_remains_authoritative_after_direct_editor(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """Alpha can reject a successful nested edit and restore its original source."""

    editor, source = _runtime_editor(
        memory_root,
        _StubGenerator(_replace("src/mod.py", "value = 'old'", "value = 'new'")),
    )
    caller_memento = _source_meta(editor, "src/mod.py").set_or_get_memento(
        _file_manager(editor),
    )

    editor.edit(_request(source))
    caller_memento.rollback()

    assert source.read_text(encoding="utf-8") == "value = 'old'\n"
    assert not Memento.current(source, _file_manager(editor).repo_exec)
    assert not caller_memento.path.exists()
    assert not caller_memento.state_path.exists()


def test_direct_editor_renders_selected_prior_cot_for_one_patch_turn(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """The patch model sees selected prior authority while the original request stays fixed."""

    generator = _StubGenerator(
        _replace("src/mod.py", "value = 'old'", "value = 'new'")
    )
    selector = _CotSelectionClient()
    generator.clients = (selector,)
    editor, source = _runtime_editor(memory_root, generator)
    prior = CotTaskRecord(
        "Remove Memento.clear.",
        "src/memento.py",
        ClassMeta("src/memento.py", "remove clear", [], {}),
    )
    request = _request(source)
    request.previous_cot_records = (prior,)

    _edit_result = editor.edit(request)
    changed, _summary = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    assert "Remove Memento.clear." in str(generator.calls[0]["base_prompt"])
    assert "CURRENT COT — PRIMARY AUTHORITY AND EDIT SCOPE" in str(
        generator.calls[0]["base_prompt"]
    )
    assert request.authoritative_prompt == "Update the value."
    assert len(selector.cot_prompts) == 1


def test_caller_memento_commit_remains_authoritative_after_direct_editor(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """Alpha can accept a successful nested edit and finalize its own layer."""

    editor, source = _runtime_editor(
        memory_root,
        _StubGenerator(_replace("src/mod.py", "value = 'old'", "value = 'new'")),
    )
    caller_memento = _source_meta(editor, "src/mod.py").set_or_get_memento(
        _file_manager(editor),
    )

    editor.edit(_request(source))
    caller_memento.commit()

    assert source.read_text(encoding="utf-8") == "value = 'new'\n"
    assert not Memento.current(source, _file_manager(editor).repo_exec)
    assert not caller_memento.path.exists()
    assert not caller_memento.state_path.exists()


def test_direct_editor_deduplicates_editable_mementos_before_committing(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """One editable file produces one nested snapshot even when listed twice."""

    editor, source = _runtime_editor(
        memory_root,
        _StubGenerator(_replace("src/mod.py", "value = 'old'", "value = 'new'")),
    )
    request = _request(source)
    request.editable_files.append(source)

    _edit_result = editor.edit(request)
    changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    assert len(compiler_factory.build_calls[0].mementos) == 1
    assert not (source.parent / ".mod.py~").exists()


def test_direct_editor_accepts_first_mechanical_candidate_as_repair_baseline(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """The first mechanical compiler error establishes a repair baseline without annealing."""

    compiler_factory.results = [[StaticCompileError("bad candidate")], []]
    editor, source = _runtime_editor(
        memory_root,
        _StubGenerator(
            [
                _replace("src/mod.py", "value = 'old'", "value = 'bad'"),
                _replace("src/mod.py", "value = 'bad'", "value = 'good'"),
            ]
        ),
    )

    _edit_result = editor.edit(_request(source))
    changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    assert [call["pre"] for call in compiler_factory.compile_calls] == [
        ["value = 'old'\n"],
        ["value = 'old'\n"],
        ["value = 'old'\n"],
    ]
    assert [call["post"] for call in compiler_factory.compile_calls] == [
        ["value = 'old'\n"],
        ["value = 'bad'\n"],
        ["value = 'good'\n"],
    ]
    repair_feedback = str(_generator(editor).render_calls[1]["repair_feedback"])
    assert "==== BEGIN CURRENT COMPILER STATE ====" in repair_feedback
    assert "bad candidate" in repair_feedback
    assert "LAST REJECTED CANDIDATE FEEDBACK" not in repair_feedback
    assert "rejected and rolled back" not in repair_feedback
    assert len(_generator(editor).render_calls) == 2
    assert _generator(editor).calls[0]["base_prompt"] == "baseline prompt"
    assert _generator(editor).calls[1]["base_prompt"] == "baseline prompt"
    assert len(_file_manager(editor).refresh_requests) == 1


def test_direct_editor_sends_equal_static_errors_through_annealing(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """An unchanged static-error count rolls back when annealing rejects it."""

    compiler_factory.baseline_results = [[StaticCompileError("baseline compiler error")]]
    compiler_factory.results = [[StaticCompileError("unchanged compiler error")], []]
    editor, source = _runtime_editor(
        memory_root,
        _StubGenerator(
            [
                _replace("src/mod.py", "value = 'old'", "value = 'unchanged'"),
                _replace("src/mod.py", "value = 'old'", "value = 'good'"),
            ]
        ),
    )

    # Reject the equal-count candidate through the existing annealing boundary.
    with patch("band_aider.judgement.direct_code_editor.random.random", return_value=1.0):
        _edit_result = editor.edit(_request(source))
        changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    assert source.read_text(encoding="utf-8") == "value = 'good'\n"
    assert compiler_factory.compile_calls[2]["pre"] == ["value = 'old'\n"]
    feedback = str(_generator(editor).render_calls[1]["repair_feedback"])
    assert "unchanged compiler error" in feedback
    assert "This candidate was rejected and rolled back." in feedback


def test_direct_editor_commits_a_targeted_test_failure_fix_forward(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """A static-clean runtime semantic failure commits outside annealing."""

    compiler_factory.results = [
        [TestRunCompileError("targeted test failed", test_name="test_value")],
        [],
    ]
    editor, source = _runtime_editor(
        memory_root,
        _StubGenerator(
            [
                _replace("src/mod.py", "value = 'old'", "value = 'test-failing'"),
                _replace("src/mod.py", "value = 'test-failing'", "value = 'passing'"),
            ]
        ),
    )

    _edit_result = editor.edit(_request(source))
    changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    assert source.read_text(encoding="utf-8") == "value = 'passing'\n"
    feedback = str(_generator(editor).render_calls[1]["repair_feedback"])
    assert "targeted test failed" in feedback
    assert "LAST REJECTED CANDIDATE FEEDBACK" not in feedback
    assert compiler_factory.candidate_calls[2]["pre"] == [
        "value = 'test-failing'\n"
    ]


def test_direct_editor_commits_unestablished_execution_with_prior_failure(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """Unestablished execution commits while its prior named failure remains current."""

    prior_error = TestRunCompileError(
        "original test_value failure",
        test_name="test_value",
        error_message="original assertion",
        stack_trace=("original trace",),
        raw_output="original output",
    )
    execution_error = TestRunCompileError(
        "pytest could not establish test_value",
        error_message="pytest exited with return code 5.",
        raw_output="\n",
    )
    compiler_factory.baseline_results = [[prior_error]]
    compiler_factory.results = [[prior_error, execution_error], []]
    editor, source = _runtime_editor(
        memory_root,
        _StubGenerator(
            [
                _replace("src/mod.py", "value = 'old'", "value = 'unestablished'"),
                _replace("src/mod.py", "value = 'unestablished'", "value = 'done'"),
            ]
        ),
    )

    _edit_result = editor.edit(_request(source))
    changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    assert source.read_text(encoding="utf-8") == "value = 'done'\n"
    assert compiler_factory.candidate_calls[2] == {
        "pre": ["value = 'unestablished'\n"],
        "post": ["value = 'done'\n"],
        "previous_errors": [prior_error, execution_error],
    }
    feedback = str(_generator(editor).render_calls[1]["repair_feedback"])
    assert "original test_value failure" in feedback
    assert "pytest could not establish test_value" not in feedback
    assert "LAST REJECTED CANDIDATE FEEDBACK" not in feedback


def test_direct_editor_preserves_outer_failure_until_compiler_tracks_test_state(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """Compiler-owned test state subsumes outer failure evidence after mechanical repair."""

    compiler_factory.baseline_results = [[
        StaticCompileError("baseline compiler error one"),
        StaticCompileError("baseline compiler error two"),
    ]]
    compiler_factory.results = [
        [StaticCompileError("accepted compiler error")],
        [TestRunCompileError("latest targeted test failure", test_name="test_value")],
        [],
    ]
    editor, source = _runtime_editor(
        memory_root,
        _StubGenerator(
            [
                _replace("src/mod.py", "value = 'old'", "value = 'mechanical'"),
                _replace("src/mod.py", "value = 'mechanical'", "value = 'test-failing'"),
                _replace("src/mod.py", "value = 'test-failing'", "value = 'passing'"),
            ]
        ),
    )
    request = _request(source)
    request.failure_error_message = "Beta outer test failure"
    request.failure_stack_trace = ["outer traceback"]
    request.failure_raw_output = "outer output"

    _edit_result = editor.edit(request)
    changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    assert source.read_text(encoding="utf-8") == "value = 'passing'\n"
    prompt_masks = _rendered_masks(editor)
    assert [mask.mode for mask in prompt_masks] == [
        AiderFailureMode.STATIC,
        AiderFailureMode.STATIC,
        AiderFailureMode.SEMANTIC,
    ]
    prompt_requests = [mask.projected() for mask in prompt_masks]
    assert [item.failure_error_message for item in prompt_requests] == [
        "",
        "",
        "Beta outer test failure",
    ]
    assert [item.failure_stack_trace for item in prompt_requests] == [
        [],
        [],
        ["outer traceback"],
    ]
    assert [item.failure_raw_output for item in prompt_requests] == [
        "",
        "",
        "outer output",
    ]
    assert "baseline compiler error" in str(
        _generator(editor).render_calls[0]["repair_feedback"]
    )
    assert "accepted compiler error" in str(
        _generator(editor).render_calls[1]["repair_feedback"]
    )
    assert "latest targeted test failure" in str(
        _generator(editor).render_calls[2]["repair_feedback"]
    )
    # Preserve the evidence-bearing request for every compiler construction.
    for attempt in compiler_factory.build_calls:
        assert attempt.request is request
    assert request.failure_error_message == "Beta outer test failure"
    assert request.failure_stack_trace == ["outer traceback"]
    assert request.failure_raw_output == "outer output"
    assert _rendered_renderer(editor, 0).failure_error_message == ""


def test_direct_editor_commits_partial_test_progress_with_its_catalog(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """A reduced unresolved catalog and its source candidate advance together."""

    alpha_error = TestRunCompileError(
        "alpha still fails",
        test_name="test_alpha",
    )
    beta_error = TestRunCompileError(
        "beta still fails",
        test_name="test_beta",
    )
    compiler_factory.baseline_results = [[alpha_error, beta_error]]
    compiler_factory.results = [[alpha_error], []]
    editor, source = _runtime_editor(
        memory_root,
        _StubGenerator(
            [
                _replace("src/mod.py", "value = 'old'", "value = 'partial'"),
                _replace("src/mod.py", "value = 'partial'", "value = 'done'"),
            ]
        ),
    )

    _edit_result = editor.edit(_request(source))
    changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    assert source.read_text(encoding="utf-8") == "value = 'done'\n"
    assert compiler_factory.candidate_calls[1]["previous_errors"] == [
        alpha_error,
        beta_error,
    ]
    assert compiler_factory.candidate_calls[2] == {
        "pre": ["value = 'partial'\n"],
        "post": ["value = 'done'\n"],
        "previous_errors": [alpha_error],
    }
    feedback = str(_generator(editor).render_calls[1]["repair_feedback"])
    assert "alpha still fails" in feedback
    assert "beta still fails" not in feedback


def test_direct_editor_rolls_back_a_static_regression_from_test_phase(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """A static regression restores accepted source and test evidence deterministically."""

    accepted_error = TestRunCompileError(
        "accepted test failure",
        test_name="test_value",
    )
    compiler_factory.baseline_results = [[accepted_error]]
    compiler_factory.results = [[StaticCompileError("syntax regression")], []]
    editor, source = _runtime_editor(
        memory_root,
        _RenderingStubGenerator(
            [
                _replace("src/mod.py", "value = 'old'", "value = 'broken'"),
                _replace("src/mod.py", "value = 'old'", "value = 'fixed'"),
            ]
        ),
    )

    request = _request(source)
    request.failure_error_message = "outer test failure"
    request.failure_stack_trace = ["outer traceback"]
    request.failure_raw_output = "outer output"
    request.backlog_guidance = ["Repair the superseded original test assignment."]

    _edit_result = editor.edit(request)
    changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    assert source.read_text(encoding="utf-8") == "value = 'fixed'\n"
    assert compiler_factory.candidate_calls[2]["previous_errors"] == [accepted_error]
    feedback = str(_generator(editor).render_calls[1]["repair_feedback"])
    assert "CURRENT COMPILER STATE" in feedback
    assert "accepted test failure" in feedback
    assert "LAST REJECTED CANDIDATE FEEDBACK" in feedback
    assert "syntax regression" in feedback
    assert "-value = 'old'" in feedback
    assert "+value = 'broken'" in feedback
    second_mask = _generator(editor).render_calls[1]["request"]
    assert isinstance(second_mask, AiderEditRequestMask)
    assert second_mask.mode is AiderFailureMode.SEMANTIC
    second_request = second_mask.projected()
    assert second_request.failure_error_message == "outer test failure"
    assert second_request.failure_stack_trace == ["outer traceback"]
    assert second_request.failure_raw_output == "outer output"
    rendered_prompt = str(_generator(editor).calls[1]["base_prompt"])
    assert "outer test failure" in rendered_prompt
    assert "accepted test failure" in rendered_prompt
    assert "syntax regression" in rendered_prompt
    assert "Repair the superseded original test assignment." not in rendered_prompt
    assert "-value = 'old'" in rendered_prompt
    assert "+value = 'broken'" in rendered_prompt
    assert "This candidate was rejected and rolled back." in rendered_prompt
    # Preserve the outer evidence for compiler reconstruction after rollback.
    for attempt in compiler_factory.build_calls:
        assert attempt.request is request


def test_direct_editor_keeps_equal_test_failures_fix_forward(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """An unchanged runtime semantic failure cannot roll back static-clean source."""

    alpha_error = TestRunCompileError("alpha fails", test_name="test_alpha")
    beta_error = TestRunCompileError("beta fails", test_name="test_beta")
    compiler_factory.baseline_results = [[alpha_error, beta_error]]
    compiler_factory.results = [[alpha_error, beta_error], []]
    editor, source = _runtime_editor(
        memory_root,
        _StubGenerator(
            [
                _replace("src/mod.py", "value = 'old'", "value = 'forward'"),
                _replace("src/mod.py", "value = 'forward'", "value = 'done'"),
            ]
        ),
    )

    _edit_result = editor.edit(_request(source))
    changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    assert compiler_factory.candidate_calls[2]["pre"] == ["value = 'forward'\n"]
    assert compiler_factory.candidate_calls[2]["previous_errors"] == [
        alpha_error,
        beta_error,
    ]


def test_direct_editor_shows_initial_compiler_failures(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """Every initial compiler failure is part of the repair assignment."""

    compiler_factory.baseline_results = [[StaticCompileError("existing baseline error")]]
    editor, source = _runtime_editor(
        memory_root,
        _StubGenerator(_replace("src/mod.py", "value = 'old'", "value = 'good'")),
    )

    _edit_result = editor.edit(_request(source))
    changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    feedback = str(_generator(editor).render_calls[0]["repair_feedback"])
    assert "CURRENT COMPILER STATE" in feedback
    assert "existing baseline error" in feedback


def test_direct_editor_preserves_first_pyright_location_in_feedback(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """The repair prompt retains the first Pyright error's structured location."""

    compiler_factory.baseline_results = [[
        TypeCheckError(
            "first type mismatch",
            path=Path("src/mod.py"),
            line=7,
            column=12,
        ),
        TypeCheckError("later mismatch", path=Path("src/other.py"), line=3, column=1),
    ]]
    editor, source = _runtime_editor(
        memory_root,
        _StubGenerator(_replace("src/mod.py", "value = 'old'", "value = 'good'")),
    )

    request = _request(source)
    request.failure_error_message = "first type mismatch"
    _edit_result = editor.edit(request)
    changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    feedback = str(_generator(editor).render_calls[0]["repair_feedback"])
    assert "src/mod.py:7:12: first type mismatch" in feedback
    assert "later mismatch" not in feedback


def test_direct_editor_scores_multiline_type_check_errors_as_diagnostics(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """Two Pyright diagnostics are worse than one multiline baseline diagnostic."""

    compiler_factory.baseline_results = [
        [TypeCheckError("baseline line one\nbaseline line two\nbaseline line three")]
    ]
    compiler_factory.results = [
        [
            TypeCheckError("first candidate"),
            TypeCheckError("second candidate"),
        ],
        [],
    ]
    editor, source = _runtime_editor(
        memory_root,
        _StubGenerator(
            [
                _replace("src/mod.py", "value = 'old'", "value = 'partial'"),
                _replace("src/mod.py", "value = 'partial'", "value = 'good'"),
            ]
        ),
    )

    # Select the accepted mechanical branch deterministically for this scenario.
    with patch("band_aider.judgement.direct_code_editor.random.random", return_value=0.0):
        _edit_result = editor.edit(_request(source))
        changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    assert compiler_factory.compile_calls[2]["pre"] == ["value = 'old'\n"]


def test_direct_editor_limits_compiler_diagnostics_in_feedback(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """Model feedback admits only the current ordered compiler diagnostic prefix."""

    compiler_factory.results = [
        [
            TypeCheckError("type one"),
            TypeCheckError("type two"),
            ReferenceError("reference problem", ["bad"], ["available"], ["closest"]),
            StaticCompileError("other problem"),
            StaticCompileError("another problem"),
        ],
        [],
    ]
    editor, source = _runtime_editor(
        memory_root,
        _StubGenerator(
            [
                _replace("src/mod.py", "value = 'old'", "value = 'bad'"),
                _replace("src/mod.py", "value = 'bad'", "value = 'good'"),
            ]
        ),
    )

    _edit_result = editor.edit(_request(source))
    changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    feedback = str(_generator(editor).render_calls[1]["repair_feedback"])
    assert "type one" in feedback
    assert "type two" not in feedback
    assert "reference problem" not in feedback
    assert "other problem" not in feedback
    assert "another problem" not in feedback


@pytest.mark.parametrize(
    ("word_count", "expected_tokens", "complete_evidence"),
    ((45713, 127997, False), (45714, 128000, False), (45715, 128002, True)),
    ids=("below-128k", "exactly-128k", "over-128k"),
)
def test_direct_editor_exposes_complete_evidence_only_above_128k(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
    word_count: int,
    expected_tokens: int,
    complete_evidence: bool,
) -> None:
    """Final prompt size expands diagnostics while preserving selected source and repair authority."""

    generator = _StubGenerator(_replace("src/mod.py", "value = 'old'", "value = 'good'"))
    editor, source = _runtime_editor(memory_root, generator)
    compiler_factory.baseline_results = [[
        TypeCheckError("first type diagnostic", path=source, line=7, column=3),
        TypeCheckError("second type diagnostic"),
        StaticCompileError("last static diagnostic"),
        TestRunCompileError("current runtime diagnostic\ncurrent runtime output tail"),
    ]]
    request = _request(source)
    request.failure_run_info = "supplied isolated test invocation"
    request.failure_error_message = "supplied test assertion"
    request.failure_stack_trace = ["first supplied frame", "last supplied frame"]
    request.failure_raw_output = "supplied output beginning\nsupplied output tail\n"
    selected_source = "CURRENT FILE EVIDENCE\nvalue = 'old'\n⋮\n"
    source_words = TokenBudgetEstimator(tokens_per_word=1).estimate(selected_source)
    composed_prompt = selected_source + "context " * (word_count - source_words)
    assert TokenBudgetEstimator().estimate(composed_prompt) == expected_tokens
    publisher = _RecordingPublisher()

    with (
        patch.object(PromptLocalization, "localize", return_value=composed_prompt),
        status_publisher_scope(publisher),
    ):
        result = editor.edit(request)

    sent_prompt = str(generator.calls[0]["base_prompt"])
    assert sent_prompt.startswith(composed_prompt)
    assert sent_prompt.count(selected_source) == 1
    assert len(generator.render_calls) == 1
    masked_request = generator.render_calls[0]["request"]
    assert isinstance(masked_request, AiderEditRequestMask)
    assert masked_request.mode is AiderFailureMode.STATIC
    assert masked_request.projected().failure_error_message == ""
    awaiting = tuple(filter(
        lambda event: event is not None and event.phase == "awaiting model",
        map(lambda event: cast(SupremeStatusDTO, event).direct_code_editor, filter(
            lambda event: isinstance(event, SupremeStatusDTO), publisher.events,
        )),
    ))
    assert len(awaiting) == 1
    assert awaiting[0] is not None
    # Above the threshold, the generator receives all diagnostics and the unmasked captured failure.
    if complete_evidence:
        assert f"{source}:7:3: first type diagnostic" in sent_prompt
        assert "second type diagnostic" in sent_prompt
        assert "last static diagnostic" in sent_prompt
        assert "current runtime output tail" in sent_prompt
        assert request.failure_run_info in sent_prompt
        assert request.failure_error_message in sent_prompt
        assert "\n".join(request.failure_stack_trace) in sent_prompt
        assert request.failure_raw_output in sent_prompt
        assert awaiting[0].model_visible_compiler_error_count == 4
    # Small and exactly-at-budget prompts retain their existing evidence projection.
    else:
        assert sent_prompt == composed_prompt
        assert awaiting[0].model_visible_compiler_error_count == 1
    assert request.failure_stack_trace == ["first supplied frame", "last supplied frame"]
    assert request.failure_raw_output == "supplied output beginning\nsupplied output tail\n"
    assert result.changed_paths == [source]
    assert source.read_text(encoding="utf-8") == "value = 'good'\n"


def test_direct_editor_rechecks_evidence_visibility_for_each_prompt(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """A smaller prompt after an accepted baseline restores normal evidence visibility."""

    generator = _StubGenerator([
        _replace("src/mod.py", "value = 'old'", "value = 'partial'"),
        _replace("src/mod.py", "value = 'partial'", "value = 'good'"),
    ])
    editor, source = _runtime_editor(memory_root, generator)
    compiler_factory.baseline_results = [[
        TypeCheckError("initial diagnostic one"),
        TypeCheckError("initial diagnostic two"),
        TypeCheckError("initial diagnostic three"),
    ]]
    compiler_factory.results = [[
        TypeCheckError("remaining diagnostic one"),
        TypeCheckError("remaining diagnostic two"),
    ], []]
    request = _request(source)
    request.failure_error_message = "supplied failure remains captured"
    request.failure_raw_output = "complete supplied output"
    oversized_prompt = "context " * 45715
    smaller_prompt = "selected current source\nvalue = 'partial'\n⋮\n"
    publisher = _RecordingPublisher()

    with (
        patch.object(PromptLocalization, "localize", side_effect=(oversized_prompt, smaller_prompt)),
        status_publisher_scope(publisher),
    ):
        result = editor.edit(request)

    assert len(generator.calls) == 2
    assert "initial diagnostic three" in str(generator.calls[0]["base_prompt"])
    assert request.failure_raw_output in str(generator.calls[0]["base_prompt"])
    assert generator.calls[1]["base_prompt"] == smaller_prompt
    visible_counts = tuple(map(
        lambda event: event.model_visible_compiler_error_count if event is not None else None,
        filter(
            lambda event: event is not None and event.phase == "awaiting model",
            map(lambda event: cast(SupremeStatusDTO, event).direct_code_editor, filter(
                lambda event: isinstance(event, SupremeStatusDTO), publisher.events,
            )),
        ),
    ))
    assert visible_counts == (3, 1)
    assert result.changed_paths == [source]
    assert source.read_text(encoding="utf-8") == "value = 'good'\n"


def test_direct_editor_preserves_already_complete_oversized_prompt(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """Complete evidence already in the composed prompt is not appended a second time."""

    generator = _StubGenerator(_replace("src/mod.py", "value = 'old'", "value = 'good'"))
    editor, source = _runtime_editor(memory_root, generator)
    compiler_factory.baseline_results = [[
        StaticCompileError("first diagnostic"),
        StaticCompileError("last diagnostic"),
    ]]
    request = _request(source)
    request.failure_run_info = "isolated invocation"
    request.failure_error_message = "captured assertion"
    request.failure_stack_trace = ["captured frame"]
    request.failure_raw_output = "captured output\n"
    complete_errors = generator.render_section(
        "CURRENT COMPILER STATE", depth=2, escape=True,
    )("first diagnostic\nlast diagnostic")
    complete_failure = generator.render_section(
        "FAILURE EVIDENCE", depth=2, escape=True,
    )("Run information:\nisolated invocation\nError message: captured assertion\n"
      "Failure evidence:\ncaptured frame\nRaw failure output:\ncaptured output\n")
    composed_prompt = "context " * 45715 + complete_errors + "\n\n" + complete_failure

    with patch.object(PromptLocalization, "localize", return_value=composed_prompt):
        result = editor.edit(request)

    assert generator.calls[0]["base_prompt"] == composed_prompt
    assert result.changed_paths == [source]


def test_direct_editor_widens_visible_compiler_errors_geometrically(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """Every third unaccepted compiler block doubles visible compiler diagnostics."""

    baseline_errors: list[CompileError] = [
        TypeCheckError("baseline type mismatch 1"),
        TypeCheckError("baseline type mismatch 2"),
        TypeCheckError("baseline type mismatch 3"),
        TypeCheckError("baseline type mismatch 4"),
        TypeCheckError("baseline type mismatch 5"),
        TypeCheckError("baseline type mismatch 6"),
        TypeCheckError("baseline type mismatch 7"),
        TypeCheckError("baseline type mismatch 8"),
    ]
    rejected_errors: list[CompileError] = [
        *baseline_errors,
        TypeCheckError("candidate extra mismatch"),
    ]
    compiler_factory.baseline_results = [baseline_errors]
    compiler_factory.results = [rejected_errors] * 8 + [[]]
    responses: list[str] = []
    # Each rejected candidate starts from the unchanged committed baseline.
    for number in range(1, 9):
        responses.append(
            _replace("src/mod.py", "value = 'old'", f"value = 'rejected-{number}'")
        )
    responses.append(_replace("src/mod.py", "value = 'old'", "value = 'good'"))
    generator = _StubGenerator(responses)
    editor, source = _runtime_editor(memory_root, generator)
    request = _request(source)
    request.failure_error_message = "\n".join(error.message for error in baseline_errors)

    _edit_result = editor.edit(request)
    changed, _ = _edit_result.changed_paths, _edit_result.summary

    permits = (1, 1, 2, 2, 2, 4, 4, 4, 8)
    # Each rendered compiler-state section proves the allowance for that repair block.
    for index, permit in enumerate(permits):
        feedback = str(generator.render_calls[index]["repair_feedback"])
        compiler_state = feedback.split("BEGIN CURRENT COMPILER STATE", 1)[1].split(
            "END CURRENT COMPILER STATE", 1
        )[0]
        # The compiler reports diagnostics in order, so each permit exposes its prefix.
        for number in range(1, permit + 1):
            assert f"baseline type mismatch {number}" in compiler_state
        # The next diagnostic remains withheld until a later widening block.
        if permit < len(baseline_errors):
            assert f"baseline type mismatch {permit + 1}" not in compiler_state
    assert changed == [source]
    assert source.read_text(encoding="utf-8") == "value = 'good'\n"


def test_direct_editor_resets_visible_compiler_errors_after_accepted_baseline(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """An accepted intermediate baseline restores one visible compiler diagnostic."""

    baseline_errors: list[CompileError] = [
        TypeCheckError("initial type mismatch one"),
        TypeCheckError("initial type mismatch two"),
    ]
    rejected_errors: list[CompileError] = [
        *baseline_errors,
        TypeCheckError("candidate extra mismatch"),
    ]
    accepted_errors: list[CompileError] = [TypeCheckError("accepted type mismatch one")]
    compiler_factory.baseline_results = [baseline_errors]
    compiler_factory.results = [
        rejected_errors,
        rejected_errors,
        accepted_errors,
        [],
    ]
    generator = _StubGenerator(
        [
            _replace("src/mod.py", "value = 'old'", "value = 'rejected-one'"),
            _replace("src/mod.py", "value = 'old'", "value = 'rejected-two'"),
            _replace("src/mod.py", "value = 'old'", "value = 'accepted'"),
            _replace("src/mod.py", "value = 'accepted'", "value = 'good'"),
        ]
    )
    editor, source = _runtime_editor(memory_root, generator)
    request = _request(source)
    request.failure_error_message = "\n".join(error.message for error in baseline_errors)

    _edit_result = editor.edit(request)
    changed, _ = _edit_result.changed_paths, _edit_result.summary

    widened_state = str(generator.render_calls[2]["repair_feedback"])
    widened_state = widened_state.split("BEGIN CURRENT COMPILER STATE", 1)[1].split(
        "END CURRENT COMPILER STATE", 1
    )[0]
    reset_state = str(generator.render_calls[3]["repair_feedback"])
    reset_state = reset_state.split("BEGIN CURRENT COMPILER STATE", 1)[1].split(
        "END CURRENT COMPILER STATE", 1
    )[0]
    assert "initial type mismatch one" in widened_state
    assert "initial type mismatch two" in widened_state
    assert "accepted type mismatch one" in reset_state
    assert "accepted type mismatch two" not in reset_state
    assert changed == [source]


def test_direct_editor_renders_compile_cards_when_static_gate_has_not_cleared(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """Static gate progress exposes both initial failures and their reference cards."""

    compiler_factory.baseline_results = [
        [TypeCheckError('"_FileManager" is incompatible with "FileManager"')]
    ]
    editor, source = _runtime_editor(
        memory_root,
        _StubGenerator(
            _replace(
                "src/mod.py",
                "import internal_mismatch\nvalue = 'old'",
                "value = 'good'",
            )
        ),
    )
    source.write_text("import internal_mismatch\nvalue = 'old'\n", encoding="utf-8")

    _edit_result = editor.edit(_request(source))
    changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    feedback = str(_generator(editor).render_calls[0]["repair_feedback"])
    assert "CURRENT COMPILER STATE" in feedback
    assert '"_FileManager" is incompatible' in feedback
    assert "COMPILE REFERENCE CARDS" in feedback
    assert "REFERENCE: tests.support.file_manager._FileManager" in feedback


def test_direct_editor_renders_reference_card_section_for_test_run_only_state(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """Test execution state neither creates nor suppresses source-derived cards."""

    compiler_factory.baseline_results = [
        [TestRunCompileError("test failed", test_name="test_value")]
    ]
    editor, source = _runtime_editor(
        memory_root,
        _StubGenerator(_replace("src/mod.py", "value = 'old'", "value = 'good'")),
    )

    _edit_result = editor.edit(_request(source))
    changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    feedback = str(_generator(editor).render_calls[0]["repair_feedback"])
    assert "test failed" in feedback
    assert "COMPILE REFERENCE CARDS" in feedback
    assert "No reference cards resolved." in feedback


def test_direct_editor_reference_cards_follow_candidate_repair_state(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """A candidate-introduced static error retains its current repair-state reference cards."""

    compiler_factory.baseline_results = [
        [TypeCheckError("accepted mismatch")]
    ]
    compiler_factory.results = [
        [
            TypeCheckError("candidate mismatch"),
            StaticCompileError("candidate extra regression"),
        ],
        [],
    ]
    editor, source = _runtime_editor(
        memory_root,
        _StubGenerator(
            [
                _replace(
                    "src/mod.py",
                    "import accepted_dependency\nvalue = 'old'",
                    "import rejected_dependency\nvalue = 'bad'",
                ),
                _replace(
                    "src/mod.py",
                    "import rejected_dependency\nvalue = 'bad'",
                    "value = 'good'",
                ),
            ]
        ),
    )
    source.write_text("import accepted_dependency\nvalue = 'old'\n", encoding="utf-8")

    # Select the accepted mechanical branch deterministically for this scenario.
    with patch("band_aider.judgement.direct_code_editor.random.random", return_value=0.0):
        _edit_result = editor.edit(_request(source))
        changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    feedback = str(_generator(editor).render_calls[1]["repair_feedback"])
    assert "accepted mismatch" not in feedback
    assert "candidate mismatch" in feedback
    assert "REFERENCE: accepted.types.AcceptedType" not in feedback
    assert "REFERENCE: rejected.types.RejectedType" in feedback


def test_partial_mutation_that_compiles_is_accepted(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """A partial patch is accepted when the changed source compiles."""

    class _PartialPatcher(FilePatcher):
        """Apply one partial in-memory candidate before returning retry evidence."""

        def __init__(self) -> None:
            """Bind the nominal patcher to this test's in-memory repository."""

            super().__init__(memory_root, repo_exec=memory_root.repository)

        def apply(
            self,
            raw_output: str,
            *,
            editable_files: list[Path],
            readonly_files: list[Path],
        ) -> CodeEditorResult:
            """Mutate the candidate and return patcher-owned partial evidence."""

            del raw_output
            del readonly_files
            memory_root.repository.write_text(editable_files[0], "value = 'partial'\n")
            raise QuickRetryError(
                "partial",
                changed_paths=editable_files,
                summary="partially applied",
                issues=["SEARCH text was not found exactly once in src/mod.py"],
            )

    source = memory_root / "src" / "mod.py"
    source.parent.mkdir(parents=True)
    source.write_text("value = 'old'\n", encoding="utf-8")
    editor = DirectCodeEditor(
        token_generator=_StubGenerator(["FILE: src/mod.py", "NO_PATCH: done"]),
        prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
        file_patcher=_PartialPatcher(),
        repo_root=memory_root,
    )
    editor.file_manager = _ManagedFileManager(memory_root, ["src/mod.py"])

    _edit_result = editor.edit(_request(source))
    changed, summary = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    assert "PARTIAL PATCH APPLIED" in summary
    assert "SEARCH text was not found exactly once in src/mod.py" in summary
    assert len(_generator(editor).calls) == 1
    assert compiler_factory.compile_calls[-1]["pre"] == ["value = 'old'\n"]
    assert compiler_factory.compile_calls[-1]["post"] == ["value = 'partial'\n"]


def test_partial_patch_with_compiler_errors_establishes_a_repair_baseline(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """A partial first mechanical error is retained as the editor's repair baseline."""

    class _PartialThenNormalPatcher(FilePatcher):
        """Return one partial candidate before production in-memory application."""

        def __init__(self) -> None:
            """Bind the nominal patcher to this test's in-memory repository."""

            super().__init__(memory_root, repo_exec=memory_root.repository)
            self.calls = 0

        def apply(
            self,
            raw_output: str,
            *,
            editable_files: list[Path],
            readonly_files: list[Path],
        ) -> CodeEditorResult:
            """Return partial evidence once, then use normal memory-backed application."""

            self.calls += 1
            if self.calls == 1:
                memory_root.repository.write_text(editable_files[0], "value = 'partial'\n")
                raise QuickRetryError(
                    "partial",
                    changed_paths=editable_files,
                    summary="partially applied",
                    issues=["SEARCH text was not found exactly once in src/mod.py"],
                )
            return super().apply(
                raw_output,
                editable_files=editable_files,
                readonly_files=readonly_files,
            )

    compiler_factory.results = [[StaticCompileError("partial is invalid")], []]
    editor, source = _runtime_editor(
        memory_root,
        _StubGenerator(
            [
                _replace("src/mod.py", "value = 'old'", "value = 'partial'"),
                _replace("src/mod.py", "value = 'partial'", "value = 'good'"),
            ]
        ),
    )
    editor.file_patcher = _PartialThenNormalPatcher()

    _edit_result = editor.edit(_request(source))
    changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    assert source.read_text(encoding="utf-8") == "value = 'good'\n"
    assert [call["pre"] for call in compiler_factory.compile_calls] == [
        ["value = 'old'\n"],
        ["value = 'old'\n"],
        ["value = 'old'\n"],
    ]
    feedback = str(_generator(editor).render_calls[1]["repair_feedback"])
    assert "partial is invalid" in feedback
    assert "LAST REJECTED CANDIDATE FEEDBACK" not in feedback
    assert len(_file_manager(editor).refresh_requests) == 1


def test_partial_candidate_refreshes_its_changed_class_meta_before_next_repair(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """An accepted QuickRetry mutation becomes fresh renderer evidence before retrying."""

    class _PartialThenNormalPatcher(FilePatcher):
        """Return one refreshable partial candidate before normal application."""

        def __init__(self) -> None:
            """Bind the nominal patcher to this test's in-memory repository."""

            super().__init__(memory_root, repo_exec=memory_root.repository)
            self.calls = 0

        def apply(
            self,
            raw_output: str,
            *,
            editable_files: list[Path],
            readonly_files: list[Path],
        ) -> CodeEditorResult:
            """Return partial evidence once, then use normal memory-backed application."""

            self.calls += 1
            if self.calls == 1:
                memory_root.repository.write_text(editable_files[0], "value = 'partial'\n")
                raise QuickRetryError(
                    "partial",
                    changed_paths=editable_files,
                    summary="partially applied",
                    issues=["second SEARCH block did not apply"],
                )
            return super().apply(
                raw_output,
                editable_files=editable_files,
                readonly_files=readonly_files,
            )

    compiler_factory.results = [[StaticCompileError("partial candidate error")], []]
    editor, source = _runtime_editor(
        memory_root,
        _StubGenerator(
            [
                _replace("src/mod.py", "value = 'old'", "value = 'partial'"),
                _replace("src/mod.py", "value = 'partial'", "value = 'good'"),
            ]
        ),
    )
    editor.file_patcher = _PartialThenNormalPatcher()

    _edit_result = editor.edit(_request(source))
    changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    assert source.read_text(encoding="utf-8") == "value = 'good'\n"
    assert _file_manager(editor).refresh_requests == [
        (
            _file_manager(editor).current_src("src/mod.py"),
            "ORIGINAL COT: IMMUTABLE STABILIZATION TARGET\n"
            "====== BEGIN ORIGINAL STABILIZATION GOAL ======\n"
            "Original stabilization goal.\n"
            "====== END ORIGINAL STABILIZATION GOAL ======\n\n"
            "ROLE AND EDIT BOUNDARIES\n"
            "Update the value.",
            "partial candidate error",
        )
    ]
    assert len(_generator(editor).render_calls) == 2


def test_direct_editor_retries_past_the_former_compiler_repair_cap(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """Compiler rejection keeps retrying until a mechanically valid candidate arrives."""

    rejections: list[list[CompileError]] = [[
        StaticCompileError("bad candidate"),
        StaticCompileError("candidate regression"),
    ]] * 6
    compiler_factory.baseline_results = [[StaticCompileError("baseline compiler error")]]
    compiler_factory.results = rejections
    editor, source = _runtime_editor(
        memory_root,
        _StubGenerator(
            [
                _replace("src/mod.py", "value = 'old'", "value = 'bad-1'"),
                _replace("src/mod.py", "value = 'old'", "value = 'bad-2'"),
                _replace("src/mod.py", "value = 'old'", "value = 'bad-3'"),
                _replace("src/mod.py", "value = 'old'", "value = 'bad-4'"),
                _replace("src/mod.py", "value = 'old'", "value = 'bad-5'"),
                _replace("src/mod.py", "value = 'old'", "value = 'bad-6'"),
                _replace("src/mod.py", "value = 'old'", "value = 'good'"),
            ],
        ),
    )

    # Reject every worse candidate through annealing before the clean repair arrives.
    with patch("band_aider.judgement.direct_code_editor.random.random", return_value=1.0):
        _edit_result = editor.edit(_request(source))
        changed, summary = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    assert len(compiler_factory.compile_calls) == 8
    assert source.read_text(encoding="utf-8") == "value = 'good'\n"
    assert _edit_result.records == [EditorRecord("", "value = 'old'", "value = 'good'")]
    assert not (source.parent / ".mod.py~").exists()


def test_direct_editor_renders_current_feedback_for_each_mechanical_candidate(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """Each candidate render receives the repair evidence current for that iteration."""

    compiler_factory.baseline_results = [[StaticCompileError("baseline compiler error")]]
    compiler_factory.results = [
        [StaticCompileError("first compiler rejection"), StaticCompileError("first extra error")],
        [StaticCompileError("second compiler rejection"), StaticCompileError("second extra error")],
        [],
    ]
    editor, source = _runtime_editor(
        memory_root,
        _StubGenerator(
            [
                _replace("src/mod.py", "value = 'old'", "value = 'first-candidate'"),
                _replace("src/mod.py", "value = 'first-candidate'", "value = 'second-candidate'"),
                _replace("src/mod.py", "value = 'second-candidate'", "value = 'good'"),
            ]
        ),
    )

    # Select the accepted mechanical branch deterministically for this scenario.
    with patch("band_aider.judgement.direct_code_editor.random.random", return_value=0.0):
        _edit_result = editor.edit(_request(source))
        changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    initial_feedback = str(_generator(editor).render_calls[0]["repair_feedback"])
    first_feedback = str(_generator(editor).render_calls[1]["repair_feedback"])
    latest_feedback = str(_generator(editor).render_calls[2]["repair_feedback"])
    assert "baseline compiler error" in initial_feedback
    assert "first compiler rejection" not in initial_feedback
    assert "first compiler rejection" in first_feedback
    assert "first-candidate" not in first_feedback
    assert "second compiler rejection" not in first_feedback
    assert "second compiler rejection" in latest_feedback
    assert "second-candidate" not in latest_feedback
    assert "first compiler rejection" not in latest_feedback
    assert "first-candidate" not in latest_feedback
    assert len(_generator(editor).render_calls) == len(_generator(editor).calls) == 3


def test_direct_editor_records_keep_complete_patch_payload(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """Retained records preserve complete source while summaries cannot replay patch control."""

    replacement = "\n".join(f"line-{index}" for index in range(60))
    raw_output = "\n".join(
        [
            "SUMMARY: beginning marker",
            "FILE: src/mod.py",
            "<<<<<<< SEARCH",
            "value = 'old'",
            "=======",
            replacement,
            ">>>>>>> REPLACE",
        ]
    )
    editor, source = _runtime_editor(memory_root, _StubGenerator(raw_output))

    _edit_result = editor.edit(_request(source))
    changed, summary = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    assert _edit_result.records == [EditorRecord("beginning marker", "value = 'old'", replacement)]
    assert "<<<<<<< SEARCH" not in summary
    assert ">>>>>>> REPLACE" not in summary
    assert compiler_factory.compile_calls


def test_editor_keeps_static_work_logs_but_discards_a_rejected_candidate(
    memory_root: MemoryPath, compiler_factory: _RecordingCompilerFactory,
) -> None:
    """The next static repair sees retained history, and rollback removes only the rejected attempt."""

    compiler_factory.baseline_results = [[StaticCompileError("one"), StaticCompileError("two")]]
    compiler_factory.results = [
        [StaticCompileError("one")],
        [StaticCompileError("one"), StaticCompileError("regression")],
        [],
    ]
    generator = _StubGenerator([
        "SUMMARY: first static repair\n" + _replace("src/mod.py", "value = 'old'", "value = 'partial'"),
        "SUMMARY: rejected repair\n" + _replace("src/mod.py", "value = 'partial'", "value = 'bad'"),
        "SUMMARY: final repair\n" + _replace("src/mod.py", "value = 'partial'", "value = 'good'"),
    ])
    editor, source = _runtime_editor(memory_root, generator)
    request = _request(source)
    with patch("band_aider.judgement.direct_code_editor.random.random", return_value=1.0), patch.object(
        _StubGenerator, "render_prompt", SearchReplaceTokenGenerator.render_prompt,
    ):
        result = editor.edit(request)

    assert source.read_text() == "value = 'good'\n"
    assert result.records == [
        EditorRecord("first static repair", "value = 'old'", "value = 'partial'"),
        EditorRecord("final repair", "value = 'partial'", "value = 'good'"),
    ]
    assert request.editor_records == result.records
    assert "Work Logs\n\nSummary:\nfirst static repair" in str(generator.calls[1]["base_prompt"])
    third_prompt = str(generator.calls[2]["base_prompt"])
    work_logs = third_prompt.split("Work Logs", 1)[1].split("END BACKLOG GUIDANCE", 1)[0]
    assert "first static repair" in work_logs
    assert "rejected repair" not in work_logs


@pytest.mark.parametrize("terminal", ["NO_PATCH: production owns the next repair", ""])
def test_editor_whole_rollback_restores_its_incoming_work_logs(
    memory_root: MemoryPath, compiler_factory: _RecordingCompilerFactory, terminal: str,
) -> None:
    """Whole-edit rollback removes its intermediate history while retaining earlier surviving work."""

    remaining: list[CompileError] = [StaticCompileError("still broken")]
    compiler_factory.results = [remaining]
    editor, source = _runtime_editor(memory_root, _StubGenerator([
        _replace("src/mod.py", "value = 'old'", "value = 'partial'"), terminal,
    ]))
    earlier = EditorRecord("earlier retained edit", "value = 0", "value = 'old'")
    request = _request(source)
    request.editor_records = [earlier]
    result = editor.edit(request)

    assert source.read_text() == "value = 'old'\n"
    assert result.changed_paths == []
    assert result.records == [earlier]
    assert request.editor_records == [earlier]


def test_editor_records_only_successful_blocks_of_an_accepted_partial_patch(
    memory_root: MemoryPath, compiler_factory: _RecordingCompilerFactory,
) -> None:
    """A partial response contributes the block that was applied, without its unmatched sibling."""

    editor, source = _runtime_editor(memory_root, _StubGenerator(
        "SUMMARY: repair value\n" + _replace("src/mod.py", "value = 'old'", "value = 'good'")
        + "\n" + _replace("src/mod.py", "absent = 0", "absent = 1")
    ))
    result = editor.edit(_request(source))

    assert source.read_text() == "value = 'good'\n"
    assert result.records == [EditorRecord("repair value", "value = 'old'", "value = 'good'")]


def test_direct_editor_fails_closed_without_compilation_context(memory_root: MemoryPath) -> None:
    """Mutation never begins when the edit has no workflow context."""

    source = memory_root / "src" / "mod.py"
    source.parent.mkdir(parents=True)
    source.write_text("value = 'old'\n", encoding="utf-8")
    generator = _StubGenerator(_replace("src/mod.py", "value = 'old'", "value = 'new'"))
    editor = DirectCodeEditor(
        token_generator=generator,
        prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
        file_patcher=FilePatcher(memory_root, repo_exec=memory_root.repository),
        repo_root=memory_root,
    )
    editor.file_manager = _ManagedFileManager(memory_root, ["src/mod.py"])

    _edit_result = editor.edit(
        AiderEditRequest("Change it.", editable_files=[source], readonly_files=[])
    )
    changed, summary = _edit_result.changed_paths, _edit_result.summary

    assert changed == []
    assert "owning workflow context" in summary
    assert generator.calls == []
    assert source.read_text(encoding="utf-8") == "value = 'old'\n"


def test_direct_editor_preserves_patch_repair_feedback(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """Ordinary patch failures remain in the editor's local repair loop."""

    editor, source = _runtime_editor(
        memory_root,
        _StubGenerator(
            [
                _replace("src/mod.py", "missing", "value = 'bad'"),
                _replace("src/mod.py", "value = 'old'", "value = 'new'"),
            ]
        ),
    )

    _edit_result = editor.edit(_request(source))
    changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    feedback = str(_generator(editor).render_calls[1]["repair_feedback"])
    assert "==== BEGIN PATCH APPLICATION FEEDBACK ====" in feedback
    assert "Patch application stage: PATCH APPLICATION DID NOT APPLY" in feedback
    assert compiler_factory.compile_calls[-1]["post"] == ["value = 'new'\n"]


def test_unexpected_patcher_failure_rolls_back_the_candidate_memento(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """An unexpected patcher failure cannot leave a candidate mutation or snapshot."""

    class _FailingPatcher(FilePatcher):
        """Raise one unexpected patcher failure after a visible in-memory mutation."""

        def __init__(self) -> None:
            """Bind the nominal patcher to this test's in-memory repository."""

            super().__init__(memory_root, repo_exec=memory_root.repository)

        def apply(
            self,
            raw_output: str,
            *,
            editable_files: list[Path],
            readonly_files: list[Path],
        ) -> CodeEditorResult:
            """Raise the non-retry failure after changing candidate state."""

            del raw_output
            del readonly_files
            memory_root.repository.write_text(editable_files[0], "value = 'broken'\n")
            raise RuntimeError("patcher transport failed")

    editor, source = _runtime_editor(
        memory_root,
        _StubGenerator(_replace("src/mod.py", "value = 'old'", "value = 'broken'")),
    )
    editor.file_patcher = _FailingPatcher()

    _edit_result = editor.edit(_request(source))
    changed, summary = _edit_result.changed_paths, _edit_result.summary

    assert changed == []
    assert "patcher transport failed" in summary
    assert "-value = 'old'" in summary
    assert "+value = 'broken'" in summary
    assert source.read_text(encoding="utf-8") == "value = 'old'\n"
    assert not (source.parent / ".mod.py~").exists()
    assert len(compiler_factory.compile_calls) == 1


def test_prepare_edit_scope_still_uses_discovery_membership(memory_root: MemoryPath) -> None:
    """The compile gate does not change language-neutral scope filtering."""

    source = memory_root / "src" / "main.rb"
    test = memory_root / "spec" / "main_spec.rb"
    ignored = memory_root / "notes.py"
    editor = DirectCodeEditor(
        token_generator=_StubGenerator("NO_PATCH: unused"),
        prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
        file_patcher=FilePatcher(memory_root, repo_exec=memory_root.repository),
        repo_root=memory_root,
    )
    editor.file_manager = _ManagedFileManager(
        memory_root,
        ["src/main.rb"],
        ["spec/main_spec.rb"],
    )

    editable, readonly, _ = editor.prepare_edit_scope(
        AiderEditRequest("scope", [source, ignored], [test])
    )

    assert editable == [source]
    assert readonly == [test]


def test_direct_editor_can_commit_a_worse_compiler_state_for_repair(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """Annealing remains available only for a worse mechanical compiler result."""

    compiler_factory.baseline_results = [[StaticCompileError("baseline compiler error")]]
    compiler_factory.results = [
        [StaticCompileError("first"), StaticCompileError("second")],
        [],
    ]
    editor, source = _runtime_editor(
        memory_root,
        _StubGenerator(
            [
                _replace("src/mod.py", "value = 'old'", "value = 'partial'"),
                _replace("src/mod.py", "value = 'partial'", "value = 'good'"),
            ]
        ),
    )

    # Select the accepted mechanical branch deterministically for this scenario.
    with patch("band_aider.judgement.direct_code_editor.random.random", return_value=0.0):
        _edit_result = editor.edit(_request(source))
        changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    assert source.read_text(encoding="utf-8") == "value = 'good'\n"
    assert compiler_factory.compile_calls[2]["pre"] == ["value = 'old'\n"]
    assert "CURRENT COMPILER STATE" in str(
        _generator(editor).render_calls[1]["repair_feedback"]
    )
    assert _file_manager(editor).refresh_requests == [
        (
            _file_manager(editor).current_src("src/mod.py"),
            "ORIGINAL COT: IMMUTABLE STABILIZATION TARGET\n"
            "====== BEGIN ORIGINAL STABILIZATION GOAL ======\n"
            "Original stabilization goal.\n"
            "====== END ORIGINAL STABILIZATION GOAL ======\n\n"
            "ROLE AND EDIT BOUNDARIES\n"
            "Update the value.",
            "first",
        )
    ]


def test_changed_class_metas_uses_only_mementos_with_actual_diffs(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """Stale propagation uses canonical ClassMeta, not editable path strings."""

    source_a = memory_root / "src" / "a.py"
    source_b = memory_root / "src" / "b.py"
    source_a.parent.mkdir(parents=True)
    source_a.write_text("value = 'old-a'\n", encoding="utf-8")
    source_b.write_text("value = 'old-b'\n", encoding="utf-8")
    editor = DirectCodeEditor(
        token_generator=_StubGenerator("NO_PATCH: unused"),
        prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
        file_patcher=FilePatcher(memory_root, repo_exec=memory_root.repository),
        repo_root=memory_root,
    )
    editor.file_manager = _ManagedFileManager(memory_root, ["src/a.py", "src/b.py"])
    del compiler_factory

    mementos = editor._capture_mementos([source_a, source_b])
    source_a.write_text("value = 'new-a'\n", encoding="utf-8")

    assert editor._changed_class_metas(mementos) == [
        _file_manager(editor).current_src("src/a.py")
    ]
    assert all(memento.diff for memento in mementos[:1])
    assert not mementos[1].diff
    for memento in reversed(mementos):
        memento.rollback()


def test_semantic_watermark_survives_a_rolled_back_static_candidate(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """A rejected static candidate cannot lower an attained semantic workflow level."""

    static_error = TypeCheckError("current type mismatch")
    test_error = TestRunCompileError("current test failure", test_name="test_value")
    compiler_factory.results = [[static_error], [test_error], []]
    editor, source = _runtime_editor(
        memory_root,
        _StubGenerator(
            [
                _replace("src/mod.py", "value = 'old'", "value = 'static'"),
                _replace("src/mod.py", "value = 'old'", "value = 'semantic'"),
                _replace("src/mod.py", "value = 'semantic'", "value = 'done'"),
            ]
        ),
    )
    request = _request(source)
    request.backlog_guidance = ["Repair test_value only."]
    request.failure_error_message = "AssertionError: wrong value"
    request.failure_stack_trace = ["tests/test_value.py:7: AssertionError"]
    request.failure_raw_output = "FAILED tests/test_value.py::test_value"

    _edit_result = editor.edit(request)
    changed, _ = _edit_result.changed_paths, _edit_result.summary

    masks = _rendered_masks(editor)
    assert [mask.mode for mask in masks] == [
        AiderFailureMode.SEMANTIC,
        AiderFailureMode.SEMANTIC,
        AiderFailureMode.SEMANTIC,
    ]
    second_feedback = str(_generator(editor).render_calls[1]["repair_feedback"])
    assert "current type mismatch" in second_feedback
    assert "This candidate was rejected and rolled back." in second_feedback
    assert masks[1].projected().failure_error_message == "AssertionError: wrong value"
    assert all(mask.original is request for mask in masks)
    assert request.failure_error_message == "AssertionError: wrong value"
    assert request.failure_stack_trace == ["tests/test_value.py:7: AssertionError"]
    assert request.failure_raw_output == "FAILED tests/test_value.py::test_value"
    assert changed == [source]
    assert source.read_text(encoding="utf-8") == "value = 'done'\n"


def test_ambiguous_response_writes_nothing_and_skips_candidate_compile(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """A duplicate SEARCH rejects every response block before compiler admission."""

    generator = _StubGenerator(
        [
            "\n".join(
                [
                    _replace("src/mod.py", "value = 'old'", "value = 'wrong'"),
                    _replace("src/mod.py", "return self.value", "return 'ambiguous'"),
                ]
            ),
            _replace(
                "src/mod.py",
                "    def first(self):\n        return self.value",
                "    def first(self):\n        return 'fixed'",
            ),
        ]
    )
    boundary_client = ScriptedCompletionClient(())
    generator.clients = (boundary_client,)
    editor, source = _runtime_editor(
        memory_root,
        generator,
    )
    source.write_text(
        "value = 'old'\n\n"
        "class Example:\n"
        "    def first(self):\n"
        "        return self.value\n\n"
        "    def second(self):\n"
        "        return self.value\n",
        encoding="utf-8",
    )

    events: list[object] = []
    publisher = _RecordingPublisher(events)
    graph = Provenance.open(repo_root=memory_root, repo_exec=memory_root.repository)
    item = InspectionBacklog(
        focused_spotlight=_source_meta(editor, "src/mod.py"),
        summary="repair the focused source",
    )
    graph.start(Phase.ALPHA, item)
    with provenance.bind(graph):
        with provenance.item_scope(Phase.ALPHA, item):
            with status_publisher_scope(publisher):
                _edit_result = editor.edit(_request(source))
                changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    assert source.read_text(encoding="utf-8") == (
        "value = 'old'\n\n"
        "class Example:\n"
        "    def first(self):\n"
        "        return 'fixed'\n\n"
        "    def second(self):\n"
        "        return self.value\n"
    )
    assert len(compiler_factory.compile_calls) == 2
    ambiguity = str(_generator(editor).render_calls[1]["repair_feedback"])
    assert "SEARCH AMBIGUITY" in ambiguity
    assert "MATCH COUNT: 2" in ambiguity
    assert "MATCH 1:" in ambiguity
    assert "MATCH 2:" in ambiguity
    assert "Did you mean one of these 2 locations?" in ambiguity
    assert ambiguity.count("<<<<<<< SEARCH") == 2
    assert ambiguity.count(">>>>>>> REPLACE") == 2
    assert "Previous model patch output" not in ambiguity
    assert "Do not answer with only a match number." in ambiguity
    assert "Every line added while expanding a choice appears identically" in ambiguity
    retry = [
        event.direct_code_editor
        for event in events
        if isinstance(event, SupremeStatusDTO)
        and event.direct_code_editor is not None
        and event.direct_code_editor.phase == "applying patch"
        and event.direct_code_editor.patch_outcome == "not applied"
    ]
    assert boundary_client.prompts == []
    assert retry[-1].unresolved_patch_issue_count == 1
    assert retry[-1].decision_branch == "retrying"
    assert retry[-1].decision_reason == (
        "SEARCH payload is ambiguous and no candidate changes exist"
    )


def test_direct_editor_presents_patcher_owned_retry_policy(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """The editor orchestrates a retry without interpreting patcher issue prose."""

    class _FeedbackPatcher(FilePatcher):
        """Expose one patcher-owned retry result before applying a unique patch."""

        def __init__(self) -> None:
            """Retain the real patcher used after the owned retry evidence is presented."""

            super().__init__(memory_root, repo_exec=memory_root.repository)
            self.calls = 0

        def apply(
            self,
            raw_output: str,
            *,
            editable_files: list[Path],
            readonly_files: list[Path],
        ) -> CodeEditorResult:
            """Return patcher policy without requiring the caller to classify its issue."""

            self.calls += 1
            # Supply an issue phrase whose presentation must not control editor behavior.
            if self.calls == 1:
                raise QuickRetryError(
                    "owned retry",
                    issues=["SEARCH text appeared more than once in decoy.py"],
                    retry_feedback="PATCHER-OWNED RETRY FEEDBACK",
                    decision_reason="patcher-owned retry reason",
                )
            return super().apply(
                raw_output,
                editable_files=editable_files,
                readonly_files=readonly_files,
            )

    editor, source = _runtime_editor(
        memory_root,
        _StubGenerator(
            [
                _replace("src/mod.py", "missing", "value = 'wrong'"),
                _replace("src/mod.py", "value = 'old'", "value = 'new'"),
            ]
        ),
    )
    editor.file_patcher = _FeedbackPatcher()
    events: list[object] = []
    publisher = _RecordingPublisher(events)

    with status_publisher_scope(publisher):
        _edit_result = editor.edit(_request(source))
        changed, _ = _edit_result.changed_paths, _edit_result.summary

    feedback = str(_generator(editor).render_calls[1]["repair_feedback"])
    retry = [
        event.direct_code_editor
        for event in events
        if isinstance(event, SupremeStatusDTO)
        and event.direct_code_editor is not None
        and event.direct_code_editor.phase == "applying patch"
        and event.direct_code_editor.patch_outcome == "not applied"
    ]
    assert changed == [source]
    assert "PATCHER-OWNED RETRY FEEDBACK" in feedback
    assert "Previous model patch output" not in feedback
    assert retry[-1].decision_reason == "patcher-owned retry reason"
    assert len(compiler_factory.compile_calls) == 2


def test_direct_editor_widens_visible_compiler_errors_after_patch_retries(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """Patch retries widen one episode's visible compiler diagnostics only."""

    class _RetryingPatcher(FilePatcher):
        """Reject two candidate blocks before normal later application."""

        def __init__(self) -> None:
            """Retain the normal patcher after the owned retry sequence."""

            super().__init__(memory_root, repo_exec=memory_root.repository)
            self.calls = 0

        def apply(
            self,
            raw_output: str,
            *,
            editable_files: list[Path],
            readonly_files: list[Path],
        ) -> CodeEditorResult:
            """Return two editor-validation retries before normal application."""

            self.calls += 1
            # Keep source unchanged until the retry cadence has reached its third block.
            if self.calls <= 2:
                raise QuickRetryError(
                    "retry required",
                    issues=["SEARCH text is stale"],
                    retry_feedback="PATCHER RETRY FEEDBACK",
                    decision_reason="patcher requested retry",
                )
            return super().apply(
                raw_output,
                editable_files=editable_files,
                readonly_files=readonly_files,
            )

    baseline_errors: list[CompileError] = [
        TypeCheckError("patch retry type mismatch one"),
        TypeCheckError("patch retry type mismatch two"),
    ]
    compiler_factory.baseline_results = [baseline_errors, baseline_errors]
    compiler_factory.results = [[]]
    generator = _StubGenerator(
        [
            _replace("src/mod.py", "value = 'old'", "value = 'retry-one'"),
            _replace("src/mod.py", "value = 'old'", "value = 'retry-two'"),
            "NO_PATCH: stop the first repair episode",
            _replace("src/mod.py", "value = 'old'", "value = 'good'"),
        ]
    )
    editor, source = _runtime_editor(memory_root, generator)
    editor.file_patcher = _RetryingPatcher()
    request = _request(source)
    # Compiler-only work can end without a patch; incoming remediation cannot.
    _edit_result = editor.edit(request)
    unchanged, _ = _edit_result.changed_paths, _edit_result.summary
    _edit_result = editor.edit(request)
    changed, _ = _edit_result.changed_paths, _edit_result.summary

    first_state = str(generator.render_calls[0]["repair_feedback"])
    first_state = first_state.split("BEGIN CURRENT COMPILER STATE", 1)[1].split(
        "END CURRENT COMPILER STATE", 1
    )[0]
    second_state = str(generator.render_calls[1]["repair_feedback"])
    second_state = second_state.split("BEGIN CURRENT COMPILER STATE", 1)[1].split(
        "END CURRENT COMPILER STATE", 1
    )[0]
    widened_state = str(generator.render_calls[2]["repair_feedback"])
    widened_state = widened_state.split("BEGIN CURRENT COMPILER STATE", 1)[1].split(
        "END CURRENT COMPILER STATE", 1
    )[0]
    reset_state = str(generator.render_calls[3]["repair_feedback"])
    reset_state = reset_state.split("BEGIN CURRENT COMPILER STATE", 1)[1].split(
        "END CURRENT COMPILER STATE", 1
    )[0]
    assert "patch retry type mismatch one" in first_state
    assert "patch retry type mismatch two" not in first_state
    assert "patch retry type mismatch one" in second_state
    assert "patch retry type mismatch two" not in second_state
    assert "patch retry type mismatch one" in widened_state
    assert "patch retry type mismatch two" in widened_state
    assert "patch retry type mismatch one" in reset_state
    assert "patch retry type mismatch two" not in reset_state
    assert unchanged == []
    assert changed == [source]


def test_direct_editor_retests_alpha_source_until_the_failure_clears(memory_root: MemoryPath) -> None:
    """Clearing static errors cannot finish Alpha while its unchanged test still fails."""
    from band_aider.judgement.stack_trace_order_strategy import StackTraceOrderStrategy
    from band_aider.strategies.python.python_stack_trace_order_strategy import PythonStackTraceOrderStrategy

    target = memory_root / "src/value.py"
    target.parent.mkdir(parents=True)
    target.write_text("value = 'old'\n")
    test_path = memory_root / "tests/test_value.py"
    test_path.parent.mkdir(parents=True)
    test_source = "def test_value():\n    assert False\n\ndef test_unrelated():\n    assert False\n"
    test_path.write_text(test_source)
    manager = _ManagedFileManager(memory_root, ["src/value.py"], ("tests/test_value.py",))
    runner = Mock(spec=_TestRunnerStrategy)
    runner.run_tests_for.side_effect = (
        TestRunResult("pytest test_value", TestRunOutcome.Failure, 1, "fresh lifecycle failure", ["tests/test_value.py:2: in test_value"]),
        TestRunResult("pytest test_value", TestRunOutcome.Success, 0, "passed"),
    )
    dependency_strategy = Mock(spec=DependencyDiscoveryStrategy)
    dependency_strategy.upstream_sources_for.return_value = []
    factory = PythonCompilerFactory(
        file_manager=manager,
        repo_exec=manager.repo_exec,
        bom_editor=PythonBomEditor(memory_root, manager.repo_exec),
        dependency_strategy=dependency_strategy,
        repo_root=memory_root,
        source_discovery=PythonSourceDiscovery(
            src_root=manager.src_root,
            test_root=manager.test_root,
            repo_exec=manager.repo_exec,
            repo_root=memory_root,
        ),
        test_runner=runner,
    )
    _install_strategy(factory)
    generator = _StubGenerator([
        _replace("src/value.py", "value = 'old'", "value = 'typed'"),
        _replace("src/value.py", "value = 'typed'", "value = 'fixed'"),
    ])
    editor = DirectCodeEditor(
        token_generator=generator,
        prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
        file_patcher=FilePatcher(memory_root, repo_exec=manager.repo_exec),
        repo_root=memory_root,
    )
    editor.file_manager = manager
    request = _request(target, AlphaLoop)
    request.readonly_files = [test_path]
    request.is_remediation = True
    request.remediation_originator_test = manager.current_test("tests/test_value.py")
    request.failure_raw_output = "FAILED tests/test_value.py::test_value - incoming lifecycle failure"
    request.failure_stack_trace = ["tests/test_value.py:2: in test_value"]
    LanguageStrategySet.get(manager).bindings[SourceDiscoveryStrategy] = factory.source_discovery
    LanguageStrategySet.get(manager).bindings[StackTraceOrderStrategy] = PythonStackTraceOrderStrategy()

    with patch.object(PythonCompiler, "type_check", side_effect=(
        [TypeCheckError("compiler factory needs a cast")], [], [],
    )):
        _edit_result = editor.edit(request)
        changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == [target]
    assert target.read_text() == "value = 'fixed'\n"
    assert test_path.read_text() == test_source
    assert tuple(call.kwargs["test_names"] for call in runner.run_tests_for.call_args_list) == (
        ("test_value",), ("test_value",),
    )
    assert len(generator.render_calls) == 2
    feedback = str(generator.render_calls[1]["repair_feedback"])
    assert "fresh lifecycle failure" in feedback
    assert "tests/test_value.py:2: in test_value" in feedback
    receipts = request.validation_receipts
    assert len(receipts.receipts) == 1
    assert receipts.receipts[0].test_name == "test_value"
    assert receipts.receipts[0].passed
    assert receipts.receipts[0].generation == receipts.generation
    assert "fresh lifecycle failure" not in receipts.render()
    assert "def test_value():" in receipts.receipts[0].test_definition


def test_direct_editor_clears_helper_failures(memory_root: MemoryPath) -> None:
    """A real Beta compiler carries monitored tests across one editor episode."""

    target = memory_root / "tests/test_widget.py"
    target.parent.mkdir(parents=True)
    original = (
        "def helper():\n    return 1\n\n"
        "def test_alpha():\n    assert helper() == 2\n\n"
        "def test_beta():\n    assert helper() == 3\n"
    )
    target.write_text(original)
    manager = _ManagedFileManager(memory_root, [], test_locations=("tests/test_widget.py",))
    runner = Mock(spec=_TestRunnerStrategy)
    runner.run_tests_for.side_effect = (
        TestRunResult("pytest alpha", TestRunOutcome.Success, 0, ""),
        TestRunResult(
            "pytest beta",
            TestRunOutcome.Failure,
            1,
            "tests/test_widget.py:7: in test_beta\nassert 2 == 3",
        ),
        TestRunResult("pytest beta", TestRunOutcome.Success, 0, ""),
    )
    dependency_strategy = Mock(spec=DependencyDiscoveryStrategy)
    dependency_strategy.upstream_sources_for.return_value = []
    factory = Mock(spec=CompilerFactory, wraps=PythonCompilerFactory(
        file_manager=manager,
        repo_exec=manager.repo_exec,
        bom_editor=PythonBomEditor(memory_root, manager.repo_exec),
        dependency_strategy=dependency_strategy,
        repo_root=memory_root,
        source_discovery=PythonSourceDiscovery(
            src_root=manager.src_root,
            test_root=manager.test_root,
            repo_exec=manager.repo_exec,
            repo_root=memory_root,
        ),
        test_runner=runner,
    ))
    _install_strategy(factory)
    generator = _StubGenerator([
        _replace("tests/test_widget.py", "return 1", "return 2"),
        _replace("tests/test_widget.py", "return 2", "return 3"),
    ])
    editor = DirectCodeEditor(
        token_generator=generator,
        prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
        file_patcher=FilePatcher(memory_root, repo_exec=manager.repo_exec),
        repo_root=memory_root,
    )
    editor.file_manager = manager
    request = _request(target, BetaLoop)
    request.failure_raw_output = (
        "FAILED tests/test_widget.py::test_alpha\n"
        "FAILED tests/test_widget.py::test_beta"
    )

    _edit_result = editor.edit(request)
    changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == [target]
    assert target.read_text() == original.replace("return 1", "return 3")
    assert tuple(call.kwargs["test_names"] for call in runner.run_tests_for.call_args_list) == (
        ("test_alpha",), ("test_beta",), ("test_beta",),
    )
    feedback = str(generator.render_calls[1]["repair_feedback"])
    assert "tests/test_widget.py:7: in test_beta" in feedback
    assert "assert 2 == 3" in feedback
    assert "Known failure for test_alpha" not in feedback
    attempts = tuple(call.args[0] for call in factory.build.call_args_list)
    assert len(attempts) == 3
    assert attempts[0].blackboard is None
    inventory = attempts[1].blackboard
    assert inventory is not None
    assert attempts[2].blackboard is inventory
    assert inventory.monitored == ()
    assert inventory.failures == ()

    receipts = request.validation_receipts
    assert receipts.receipts == ()
    assert inventory.receipts == ()


def test_direct_editor_keeps_admission_when_candidate_is_rejected(
    memory_root: MemoryPath,
) -> None:
    """Rejection discards test results while preserving immediate admission."""

    decisions: list[tuple[str, tuple[str, ...]]] = []

    class _DecisionInventory(EditedTestInventory):
        """Record inventory state at editor-owned candidate decisions."""

        def commit_candidate_results(self) -> None:
            super().commit_candidate_results()
            decisions.append(("commit", self.monitored))

        def discard_candidate_results(self) -> None:
            super().discard_candidate_results()
            decisions.append(("discard", self.monitored))

    class _AdmissionCompiler(_RecordingCompiler):
        """Admit once, then stage a passing result for both candidate checks."""

        def compile(self, paths: Sequence[Path]) -> list[CompileError]:
            errors = super().compile(paths)
            # Candidate checks stage one exact test result for the editor's decision.
            if self._attempt.candidate_mementos:
                # Only the rejected first candidate performs admission.
                if not self.blackboard.monitored:
                    self.blackboard.add_monitored({"test_changed"})
                self.blackboard.set_candidate_results({"test_changed": None})
            return errors

    class _AdmissionCompilerFactory(_RecordingCompilerFactory):
        """Share one decision-recording inventory across compiler instances."""

        def build(self, attempt: CompileAttempt) -> Compiler:
            self.build_calls.append(attempt)
            compiler = _AdmissionCompiler(self, attempt)
            # The initial compiler owns the episode inventory.
            if attempt.blackboard is None:
                compiler.blackboard = _DecisionInventory(compiler._clean_blackboard)
            self.compilers.append(compiler)
            return compiler

    factory = _AdmissionCompilerFactory([
        [StaticCompileError("candidate remains mechanically invalid")],
        [],
    ])
    factory.baseline_results = [[TestRunCompileError(
        "incoming semantic failure",
        test_name="test_changed",
    )]]
    _install_strategy(factory)
    generator = _StubGenerator([
        _replace("src/mod.py", "value = 'old'", "value = 'rejected'"),
        _replace("src/mod.py", "value = 'old'", "value = 'accepted'"),
    ])
    editor, source = _runtime_editor(memory_root, generator)
    request = _request(source)
    request.failure_error_message = "incoming semantic failure"

    _edit_result = editor.edit(request)
    changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    assert source.read_text() == "value = 'accepted'\n"
    assert decisions == [
        ("discard", ("test_changed",)),
        ("commit", ()),
    ]


def test_direct_editor_isolates_blackboards(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """Reusing one editor scopes independent compiler inventories per episode."""

    generator = _StubGenerator(_replace("src/mod.py", "value = 'old'", "value = 'first'"))
    editor, source = _runtime_editor(memory_root, generator)
    assert editor.edit(_request(source)).changed_paths == [source]
    first = compiler_factory.build_calls[1].blackboard
    assert compiler_factory.build_calls[0].blackboard is None
    assert not hasattr(compiler_factory.compilers[0], "blackboard")

    generator.responses = [_replace("src/mod.py", "value = 'first'", "value = 'second'")]
    assert editor.edit(_request(source)).changed_paths == [source]
    second = compiler_factory.build_calls[3].blackboard

    assert compiler_factory.build_calls[2].blackboard is None
    assert not hasattr(compiler_factory.compilers[2], "blackboard")
    assert first is not None
    assert second is not None
    assert second is not first
    assert first.monitored == second.monitored == ()
    assert source.read_text() == "value = 'second'\n"


def test_direct_editor_releases_blackboard_on_failure(memory_root: MemoryPath) -> None:
    """An exception after admission releases the inventory and rolls back episode source."""

    factory = Mock(spec=CompilerFactory)
    compiler = Mock(spec=BlackboardCompiler)
    factory.build.side_effect = (compiler, RuntimeError("compiler construction failed"))
    forgotten: list[bool] = []
    inventory = EditedTestInventory(lambda: forgotten.append(True))
    compiler.get_blackboard.return_value = inventory
    _install_strategy(factory)
    editor, source = _runtime_editor(
        memory_root,
        _StubGenerator(_replace("src/mod.py", "value = 'old'", "value = 'candidate'")),
    )
    # Add identified failures before the next compiler construction fails.
    compiler.compile.side_effect = lambda paths: (
        inventory.add_identified_failures((
            TestRunCompileError("incoming failure", test_name="test_value"),
        )) or []
    )

    _edit_result = editor.edit(_request(source))
    changed, summary = _edit_result.changed_paths, _edit_result.summary

    assert changed == []
    assert "compiler construction failed" in summary
    assert "-value = 'old'" in summary
    assert "+value = 'candidate'" in summary
    assert source.read_text() == "value = 'old'\n"
    assert inventory.monitored == ()
    assert inventory.failures == ()
    assert forgotten == [True]


def test_direct_editor_cleanup_failure_preserves_error_after_rollback(memory_root: MemoryPath) -> None:
    """Cleanup failure after NO_PATCH must not try to read an already deleted snapshot."""

    factory = Mock(spec=CompilerFactory)
    compiler = Mock(spec=BlackboardCompiler)
    factory.build.return_value = compiler
    compiler.compile.return_value = []
    compiler.get_blackboard.return_value = EditedTestInventory(
        Mock(side_effect=RuntimeError("inventory release failed")),
    )
    _install_strategy(factory)
    editor, source = _runtime_editor(memory_root, _StubGenerator("NO_PATCH: no edit"))

    _edit_result = editor.edit(_request(source))
    changed, summary = _edit_result.changed_paths, _edit_result.summary

    assert changed == []
    assert "inventory release failed" in summary
    assert source.read_text() == "value = 'old'\n"
    assert not Memento.current(source, memory_root.repository)


def test_direct_editor_rejects_hidden_static_errors(
    memory_root: MemoryPath,
    compiler_factory: _RecordingCompilerFactory,
) -> None:
    """A returning initial diagnostic blocks completion after semantic validation begins."""

    initial = StaticCompileError("original static error")
    test_failure = TestRunCompileError("test still fails", test_name="test_value")
    compiler_factory.baseline_results = [[initial]]
    compiler_factory.results = [[test_failure], [initial], []]
    editor, source = _runtime_editor(memory_root, _StubGenerator([
        _replace("src/mod.py", "value = 'old'", "value = 'semantic'"),
        _replace("src/mod.py", "value = 'semantic'", "value = 'regression'"),
        _replace("src/mod.py", "value = 'semantic'", "value = 'fixed'"),
    ]))

    _edit_result = editor.edit(_request(source))
    changed, _ = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    assert source.read_text() == "value = 'fixed'\n"
    assert len(compiler_factory.build_calls) == 4
    assert compiler_factory.candidate_calls[3]["pre"] == ["value = 'semantic'\n"]
    initial_feedback = str(_generator(editor).render_calls[0]["repair_feedback"])
    rejected_feedback = str(_generator(editor).render_calls[2]["repair_feedback"])
    assert "original static error" in initial_feedback
    assert "original static error" in rejected_feedback
    assert "This candidate was rejected and rolled back." in rejected_feedback
    assert _rendered_masks(editor)[2].mode is AiderFailureMode.SEMANTIC


@pytest.mark.parametrize("failure_kind", ("unmatched", "ambiguous", "malformed"))
def test_patch_failures_reset_counter_after_widening_and_retry_past_whole_source(
    memory_root: MemoryPath, compiler_factory: _RecordingCompilerFactory, failure_kind: str,
) -> None:
    """The third model request follows refinement; whole context never exhausts retries."""

    editor, source = _runtime_editor(memory_root, _StubGenerator(""))
    baseline = (
        "value = 'old'\n"
        "def first():\n" + "    item = 1\n" * 24 + "    return item\n"
        "def second():\n" + "    item = 2\n" * 24 + "    return item\n"
    )
    source.write_text(baseline)
    bad = {
        "unmatched": _replace("src/mod.py", "missing text", "wrong"),
        "ambiguous": _replace("src/mod.py", "    return item", "    return 0"),
        "malformed": "FILE: src/mod.py\n<<<<<<< SEARCH\nmissing separator\n",
    }[failure_kind]
    responses = iter([bad] * 12 + [_replace("src/mod.py", "value = 'old'", "value = 'fixed'")])
    client = ScriptedCompletionClient(())
    editor.token_generator = SearchReplaceTokenGenerator(
        (client,), memory_root, repo_exec=memory_root.repository,
        prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
    )
    editor.file_manager = _file_manager(editor)
    strategies = LanguageStrategySet.get(_file_manager(editor))
    strategies.bindings[CustomDirectives] = CustomDirectives({})
    strategies.bindings[SourceDiscoveryStrategy] = PythonSourceDiscovery(
        memory_root / "src", memory_root / "tests", repo_exec=memory_root.repository,
    )
    real_refine = ContextualFileRenderer.refine
    refinement_attempts: list[int] = []
    with (
        patch.object(client, "complete", side_effect=lambda prompt, logger=None: (
            next(responses) if "BEGIN CURRENT FILE EVIDENCE" in prompt else "NO_KEEP\nREASON: none\n"
        )) as completion,
        patch("band_aider.judgement.contextual_file_renderer.get_token_budget_settings", return_value=TokenBudgetSettings(max_prompt_tokens=1)),
        patch.object(ContextualFileRenderer, "refine", autospec=True, side_effect=lambda *args, **kwargs: (
            refinement_attempts.append(editor.attempts_since_acceptance) or real_refine(*args, **kwargs)
        )) as refinement,
        patch.object(editor.token_generator, "generate", wraps=editor.token_generator.generate) as generation,
        patch.object(editor, "_render_compiler_errors", wraps=editor._render_compiler_errors) as diagnostics,
    ):
        _edit_result = editor.edit(_request(source))
        changed, _summary = _edit_result.changed_paths, _edit_result.summary

    prompts = tuple(
        call.args[0] for call in completion.call_args_list
        if "BEGIN CURRENT FILE EVIDENCE" in call.args[0]
    )
    assert changed == [source]
    assert source.read_text() == baseline.replace("value = 'old'", "value = 'fixed'")
    assert len(prompts) == 13
    assert refinement.call_count == 6
    assert refinement_attempts == [3, 6, 9, 12, 15, 18]
    assert tuple(call.kwargs["unapplied_attempts"] for call in generation.call_args_list) == (0, 1, 2) * 6 + (0,)
    # Each exhausted group reaches the existing cadence once, without revisiting it on retry.
    assert diagnostics.call_count == 8  # Initial compile, six cadence widenings, final candidate.
    assert "def first():" not in prompts[0]
    assert "def first():" not in prompts[2]  # Progressive NO_KEEP is not a terminal result.
    assert baseline in prompts[8]  # Existing fourth refinement exposes whole source.
    assert baseline in prompts[12]  # Further failures still reach generation.
    assert "PATCH APPLICATION FEEDBACK" in prompts[2]
    assert len(compiler_factory.compile_calls) == 2
    assert compiler_factory.candidate_calls[-1]["pre"] == [baseline]


@pytest.mark.parametrize("static_gate", (False, True))
def test_initial_visible_failure_reaches_context_selection(
    memory_root: MemoryPath, compiler_factory: _RecordingCompilerFactory, static_gate: bool,
) -> None:
    """First-attempt source visibility follows the same failure mask as the repair prompt."""

    generator = _RenderingStubGenerator(_replace("src/mod.py", "value = 'old'", "value = 'fixed'"))
    editor, source = _runtime_editor(memory_root, generator)
    source.write_text(
        "value = 'old'\n"
        "def traced():\n" + "    item = 1\n" * 24 + "    return item\n"
        "def unrelated():\n" + "    item = 2\n" * 24 + "    return item\n"
    )
    failure = TestRunCompileError("traced failed", stack_trace=("traced",))
    compiler_factory.baseline_results = [[StaticCompileError("Unknown type")] if static_gate else [failure]]
    strategies = LanguageStrategySet.get(_file_manager(editor))
    strategies.bindings[SourceDiscoveryStrategy] = PythonSourceDiscovery(
        memory_root / "src", memory_root / "tests", repo_exec=memory_root.repository,
    )
    request = _request(source)
    request.failure_stack_trace = ["traced"]

    with patch("band_aider.judgement.contextual_file_renderer.get_token_budget_settings", return_value=TokenBudgetSettings(max_prompt_tokens=1)):
        _edit_result = editor.edit(request)
        changed, _summary = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    source_evidence = str(generator.calls[0]["base_prompt"]).split("BEGIN CURRENT FILE EVIDENCE", 1)[1]
    assert ("def traced():" in source_evidence) is not static_gate
    assert "def unrelated():" not in source_evidence


@pytest.mark.parametrize("partial", (False, True))
def test_applied_candidate_resets_unapplied_count_before_compiler_repair(
    memory_root: MemoryPath, compiler_factory: _RecordingCompilerFactory, partial: bool,
) -> None:
    """Applied and partial candidates move into compiler tracking rather than patch exhaustion."""

    middle = _replace("src/mod.py", "value = 'old'", "value = 'middle'")
    # A partially applicable response still belongs to the existing compiler transition.
    if partial:
        middle += "\n" + _replace("src/mod.py", "missing second block", "unused")
    generator = _StubGenerator([
        _replace("src/mod.py", "missing first block", "unused"),
        middle,
        _replace("src/mod.py", "value = 'middle'", "value = 'fixed'"),
    ])
    editor, source = _runtime_editor(memory_root, generator)
    compiler_factory.results = [[StaticCompileError("Repair the intermediate candidate")], []]

    _edit_result = editor.edit(_request(source))
    changed, _summary = _edit_result.changed_paths, _edit_result.summary

    assert changed == [source]
    assert tuple(call["unapplied_attempts"] for call in generator.calls) == (0, 1, 0)
    assert source.read_text() == "value = 'fixed'\n"
    assert len(compiler_factory.compile_calls) == 3