169 lines
6.2 KiB
Python
169 lines
6.2 KiB
Python
"""Monitored tests and their failures for one direct-edit episode."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Collection, Iterable, Mapping
|
|
from types import TracebackType
|
|
from typing import TYPE_CHECKING, Callable
|
|
from band_aider.context.validation_receipts import ValidationReceipt
|
|
|
|
# Diagnostics are annotation-only here to avoid a runtime cycle with compiler contracts.
|
|
if TYPE_CHECKING:
|
|
from band_aider.strategies.compiler import TestRunCompileError
|
|
|
|
|
|
class EditedTestInventory:
|
|
"""Track monitored tests and delegate scope cleanup to the owning compiler.
|
|
|
|
A compiler supplies the callback that forgets its inventory handle. Mutable
|
|
sets and mappings represent the episode's accumulated repair work.
|
|
|
|
Example::
|
|
|
|
with compiler.get_blackboard() as inventory:
|
|
inventory.add_monitored({"test_value"})
|
|
inventory.set_candidate_results({"test_value": None})
|
|
inventory.commit_candidate_results()
|
|
"""
|
|
|
|
def __init__(self, forget: Callable[[], None]) -> None:
|
|
"""Start empty and retain the owning compiler's forget operation."""
|
|
|
|
self._monitored: set[str] = set()
|
|
self._failures: dict[str, TestRunCompileError] = {}
|
|
self._candidate_results: dict[str, TestRunCompileError | None] = {}
|
|
self._identified_failures_added = False
|
|
self._forget = forget
|
|
self._receipts: list[ValidationReceipt] = []
|
|
self._candidate_receipts: list[ValidationReceipt] = []
|
|
self._generation = 0
|
|
self._accepted_generation = 0
|
|
|
|
def __enter__(self) -> EditedTestInventory:
|
|
"""Provide the inventory for the caller's edit scope."""
|
|
|
|
return self
|
|
|
|
def __exit__(
|
|
self,
|
|
exc_type: type[BaseException] | None,
|
|
exc: BaseException | None,
|
|
traceback: TracebackType | None,
|
|
) -> None:
|
|
"""Release episode evidence and tell the owning compiler to forget it."""
|
|
|
|
self._monitored.clear()
|
|
self._failures.clear()
|
|
self._candidate_results.clear()
|
|
self._identified_failures_added = False
|
|
self.clear_receipts()
|
|
forget = self._forget
|
|
del self._forget
|
|
forget()
|
|
|
|
def add_identified_failures(
|
|
self,
|
|
failures: Iterable[TestRunCompileError],
|
|
) -> None:
|
|
"""Add identified failures once so cleared tests stay cleared this episode."""
|
|
|
|
# A later compiler must not add a monitored test already cleared in this episode.
|
|
if self._identified_failures_added:
|
|
return
|
|
self._failures.update(map(lambda failure: (failure.test_name, failure), failures))
|
|
self._monitored |= self._failures.keys()
|
|
self._identified_failures_added = True
|
|
|
|
def add_monitored(self, test_names: Collection[str]) -> None:
|
|
"""Add test_names to monitored membership immediately."""
|
|
|
|
self._monitored |= set(test_names)
|
|
|
|
def remove_not_found(self, current_test_names: Collection[str]) -> None:
|
|
"""Remove monitored names absent from current_test_names."""
|
|
|
|
self._monitored &= set(current_test_names)
|
|
self._failures = dict(
|
|
filter(lambda item: item[0] in self._monitored, self._failures.items())
|
|
)
|
|
|
|
def tests_to_run(
|
|
self,
|
|
current_test_names: Collection[str],
|
|
) -> tuple[str, ...]:
|
|
"""Return monitored names included in current_test_names."""
|
|
|
|
return tuple(sorted(self._monitored & set(current_test_names)))
|
|
|
|
def set_candidate_results(
|
|
self,
|
|
results: Mapping[str, TestRunCompileError | None],
|
|
) -> None:
|
|
"""Store test results until the editor decides whether to commit them."""
|
|
|
|
self._candidate_results = dict(results)
|
|
|
|
def begin_candidate(self) -> None:
|
|
"""Advance code identity even if this candidate never reaches test execution."""
|
|
self._generation += 1
|
|
self._candidate_results.clear()
|
|
self._candidate_receipts.clear()
|
|
|
|
def record_receipt(self, receipt: ValidationReceipt) -> None:
|
|
"""Stage an actual observation in the existing candidate transaction."""
|
|
self._candidate_receipts.append(receipt)
|
|
|
|
def clear_receipts(self) -> None:
|
|
"""Invalidate all observations when the outer editor episode rolls back."""
|
|
self._receipts.clear()
|
|
self._candidate_receipts.clear()
|
|
|
|
@property
|
|
def receipts(self) -> tuple[ValidationReceipt, ...]:
|
|
"""Return retained observations only; rejected candidates are never exported."""
|
|
return tuple(self._receipts)
|
|
|
|
@property
|
|
def generation(self) -> int:
|
|
"""Identify the candidate currently being compiled."""
|
|
return self._generation
|
|
|
|
@property
|
|
def accepted_generation(self) -> int:
|
|
"""Identify the last committed code state."""
|
|
return self._accepted_generation
|
|
|
|
def commit_candidate_results(self) -> None:
|
|
"""Apply stored results after the editor commits their candidate source."""
|
|
|
|
# Apply every result against the accepted monitored state.
|
|
for test_name, failure in self._candidate_results.items():
|
|
# A confirmed pass or removal clears membership and prior failure evidence.
|
|
if failure is None:
|
|
self._monitored.discard(test_name)
|
|
self._failures.pop(test_name, None)
|
|
# A confirmed failure replaces evidence for its monitored identity.
|
|
else:
|
|
self._failures[test_name] = failure
|
|
self._candidate_results.clear()
|
|
self._receipts.extend(self._candidate_receipts)
|
|
self._candidate_receipts.clear()
|
|
self._accepted_generation = self._generation
|
|
|
|
def discard_candidate_results(self) -> None:
|
|
"""Forget stored results after the editor rejects their candidate source."""
|
|
|
|
self._candidate_results.clear()
|
|
self._candidate_receipts.clear()
|
|
|
|
@property
|
|
def monitored(self) -> tuple[str, ...]:
|
|
"""Return monitored tests in deterministic execution order."""
|
|
|
|
return tuple(sorted(self._monitored))
|
|
|
|
@property
|
|
def failures(self) -> tuple[TestRunCompileError, ...]:
|
|
"""Return monitored failures in deterministic test-name order."""
|
|
|
|
return tuple(map(self._failures.__getitem__, sorted(self._failures)))
|