1020 lines
49 KiB
Python
1020 lines
49 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
# band_aider/loop/beta_loop.py
|
|
|
|
from __future__ import annotations
|
|
|
|
from band_aider.judgement.work_definition_constants import Work
|
|
|
|
from band_aider.judgement.test_runner_strategy import TestRunOutcome
|
|
|
|
from contextlib import nullcontext
|
|
from dataclasses import dataclass, field, replace
|
|
from pathlib import Path
|
|
from typing import cast
|
|
|
|
from band_aider.ai_client import llm_logging
|
|
from band_aider.context import admitted_sources, provenance, working_backlog
|
|
from band_aider.context.beta_context import BetaContext
|
|
from band_aider.context.beta_workflow_story import TargetedFailureStory
|
|
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.provenance import DeadEndError, Phase, Provenance
|
|
from band_aider.context.working_backlog import (
|
|
BacklogLane,
|
|
ChangeType,
|
|
FailedQAItem,
|
|
InspectionBacklog,
|
|
OutboundBacklogItem,
|
|
StagedInboundItem,
|
|
TestingBacklog,
|
|
curate_failure_context_sources,
|
|
)
|
|
from band_aider.judgement.bi_dependency_discovery_strategy import (
|
|
BiDependencyDiscoveryStrategy,
|
|
)
|
|
from band_aider.judgement.edit_request import AiderEditRequest
|
|
from band_aider.judgement.code_edit_or_leave_it import consume_code_edit_leave_summary
|
|
from band_aider.judgement.cached_code_editor import CachedCodeEditor
|
|
from band_aider.judgement.code_editor import CodeEditor
|
|
from band_aider.judgement.test_runner_strategy import TestRunResult, TestRunnerStrategy
|
|
from band_aider.judgement.vibe_check import ReasonableVibeCheck
|
|
from band_aider.loop.exceptions import BetaRemediationRequested, BetaRequeueRequested
|
|
from band_aider.loop.loop_gate import LoopGate
|
|
from band_aider.loop.proto_loop import ProtoLoop
|
|
from band_aider.strategies import LanguageStrategySet
|
|
from band_aider.strategies.compiler import FrontierProvider
|
|
from band_aider.ui.status import publish_loop_status
|
|
|
|
@dataclass(frozen=True)
|
|
class AppliedTestWork:
|
|
"""AppliedTestWork records the result of one Beta test-edit pass."""
|
|
|
|
testing_item: TestingBacklog
|
|
work_needed: str
|
|
test_path: Path
|
|
changed_paths: tuple[Path, ...]
|
|
summary: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TestingItemResult:
|
|
"""Report one QA obligation's completion and its post-edit execution evidence.
|
|
|
|
``passed`` describes the incoming obligation; ``targeted.passed`` describes
|
|
the whole test-file execution. The workflow inspects a new observation::
|
|
|
|
if result.passed and result.new_failure is not None:
|
|
logger.info("Completed prior work; observed %s", result.new_failure.test_name)
|
|
"""
|
|
|
|
summary: str
|
|
applied: AppliedTestWork
|
|
targeted: TestRunResult
|
|
new_failure: FailingTest | None = None
|
|
|
|
@property
|
|
def passed(self) -> bool:
|
|
"""Return whether the incoming obligation completed, even if a different test failed."""
|
|
|
|
return self.targeted.passed or self.new_failure is not None
|
|
|
|
|
|
@dataclass(init=False)
|
|
class BetaLoop(ProtoLoop[LoopGate, BetaContext]):
|
|
"""BetaLoop drains QA backlog items while only touching test metadata."""
|
|
|
|
test_runner_strategy: TestRunnerStrategy
|
|
aider_worker: CodeEditor | None = None
|
|
frontier_provider: FrontierProvider | None = None
|
|
previous_cot_records: tuple[CotTaskRecord, ...] = field(default=(), init=False, repr=False)
|
|
_blocked_summary: str = field(default="", init=False, repr=False)
|
|
provenance: Provenance | None = field(default=None, init=False, repr=False)
|
|
|
|
def __init__(
|
|
self,
|
|
gate: LoopGate,
|
|
context: BetaContext,
|
|
aider_worker: CodeEditor | None = None,
|
|
frontier_provider: FrontierProvider | None = None,
|
|
previous_cot_records: tuple[CotTaskRecord, ...] = (),
|
|
provenance: Provenance | None = None,
|
|
logger=None,
|
|
) -> None:
|
|
super().__init__(gate=gate, context=context, logger=logger);
|
|
self.test_runner_strategy = context.test_runner_strategy
|
|
self.aider_worker = aider_worker;
|
|
self.frontier_provider = frontier_provider
|
|
self.previous_cot_records = previous_cot_records
|
|
self._blocked_summary = ""
|
|
self.provenance = provenance
|
|
|
|
def _run_one_testing_item(self, testing_item: TestingBacklog) -> TestingItemResult:
|
|
"""Run one ordinary Beta item within its editor cache, returning its validation result."""
|
|
|
|
cache_scope = nullcontext()
|
|
if isinstance(self.aider_worker, CachedCodeEditor):
|
|
cache_sources = testing_item.downstream_classes
|
|
source_hash = ":".join(
|
|
source.get_hash(self.context.file_manager)
|
|
for source in cache_sources
|
|
)
|
|
cache_scope = self.aider_worker.enable_cache(
|
|
f"{','.join(source.location for source in cache_sources)}:{source_hash}"
|
|
)
|
|
with cache_scope as active_cache_scope:
|
|
if active_cache_scope is not None:
|
|
active_cache_scope.infer_from(testing_item.downstream_classes)
|
|
return self._handle_testing_item(testing_item)
|
|
|
|
def _refine_beta_instruction(self, testing_item: TestingBacklog) -> str:
|
|
"""Refine one selected test file into the sole Beta work instruction."""
|
|
focused = testing_item.focused_spotlight
|
|
if focused is None:
|
|
raise ValueError("TestingBacklog requires focused_spotlight before Beta refinement.")
|
|
# Test-owned work remains repairable without an associated production change.
|
|
source_diff = (
|
|
testing_item.downstream_classes[0].get_diff(self.context.file_manager)
|
|
if testing_item.downstream_classes else ""
|
|
)
|
|
if testing_item.change_type is ChangeType.Create:
|
|
status = "NEW / NOT RUN"
|
|
elif testing_item.failure_error_message or testing_item.failure_stack_trace:
|
|
status = "FAILED"
|
|
else:
|
|
status = "PASSING"
|
|
try:
|
|
return self.context.backlog.judgement_service.refine_beta_instruction(
|
|
editor_records=testing_item.editor_records,
|
|
file_manager=self.context.file_manager,
|
|
focused_test=focused,
|
|
original_cot=testing_item.original_cot,
|
|
accepted_source_diff=source_diff,
|
|
assignment_type=testing_item.summary,
|
|
test_status=status,
|
|
previous_beta_guidance=testing_item.previous_beta_guidance,
|
|
failure_error_message=testing_item.failure_error_message,
|
|
failure_stack_trace=list(testing_item.failure_stack_trace),
|
|
failure_run_info=testing_item.failure_run_info,
|
|
readonly_sources=self._readonly_sources(testing_item),
|
|
previous_cot_records=self.previous_cot_records,
|
|
logger=self.logger.getChild("backlog"),
|
|
)
|
|
except ValueError as error:
|
|
raise BetaRequeueRequested(
|
|
f"Beta prompt refinement requires correction; no edit was attempted: {error}",
|
|
testing_item,
|
|
) from error
|
|
|
|
def _readonly_sources(self, testing_item: TestingBacklog) -> list[ClassMeta]:
|
|
"""Return deduped topology and failure-context sources for Beta readonly use."""
|
|
|
|
sources_by_location: dict[str, ClassMeta] = {}
|
|
# Keep topology and current failure diagnosis visible without changing Beta's test-only authority.
|
|
for source in [
|
|
*testing_item.upstream_classes,
|
|
*testing_item.downstream_classes,
|
|
*testing_item.failure_context_sources,
|
|
*admitted_sources.current_sources(self.context.file_manager, testing_item),
|
|
]:
|
|
# The selected test remains editable while supporting tests stay readonly.
|
|
if testing_item.focused_spotlight is None or source.location != testing_item.focused_spotlight.location:
|
|
sources_by_location[source.location] = source
|
|
return list(sources_by_location.values())
|
|
|
|
def _readonly_paths(self, testing_item: TestingBacklog) -> list[Path]:
|
|
"""Return Beta readonly source paths from topology and current failure context."""
|
|
|
|
editable = (
|
|
self.context.file_manager.test_path(testing_item.focused_spotlight.location)
|
|
if testing_item.focused_spotlight is not None
|
|
else None
|
|
)
|
|
readonly: list[Path] = []
|
|
seen: set[Path] = set()
|
|
# Convert the shared readonly source scope into editor paths without widening edit authority.
|
|
for class_meta in self._readonly_sources(testing_item):
|
|
path = class_meta.resolve_path(self.context.file_manager)
|
|
if path == editable or path in seen:
|
|
continue
|
|
seen.add(path)
|
|
readonly.append(path)
|
|
return readonly
|
|
|
|
def _authoritative_prompt(self, editable_tests: list[Path]) -> str:
|
|
"""Return static Beta role and file-scoped assignment instructions."""
|
|
locations = [self.context.file_manager.test_location(path) for path in editable_tests]
|
|
return "\n".join([
|
|
"You are Beta.",
|
|
Work.AUTHORITY,
|
|
Work.CONTRACT,
|
|
"Perform the exact test assignment supplied in backlog guidance.",
|
|
"Edit only the listed editable test file.",
|
|
"The assignment may require edits in multiple functions or regions within that file.",
|
|
"Do not edit source files, readonly files, or another test file.",
|
|
"Address every supplied blocking diagnostic that belongs to the editable test file before commit.",
|
|
"The selected test file must pass the configured compilation gate; resolving its diagnostics is not scope broadening.",
|
|
"Preserve assertions that enforce the required contract.",
|
|
"Do not weaken assertions merely to obtain a pass.",
|
|
"After this edit, the harness will run the focused test and then the broader suite.",
|
|
"EDITABLE TEST FILE",
|
|
*[f"- {location}" for location in locations],
|
|
])
|
|
|
|
def _ensure_test_file(self, testing_item: TestingBacklog) -> ClassMeta:
|
|
"""Return the concrete test target, creating missing Create-ticket files."""
|
|
|
|
if testing_item.focused_spotlight is None:
|
|
raise ValueError("TestingBacklog requires focused_spotlight before Beta processing.")
|
|
file_manager = self.context.file_manager
|
|
repo_exec = file_manager.repo_exec
|
|
spotlight = testing_item.focused_spotlight
|
|
current = file_manager.current_test(spotlight.location)
|
|
test_path = file_manager.test_path(spotlight.location)
|
|
if not repo_exec.exists(test_path) and testing_item.change_type is not ChangeType.Create:
|
|
raise ValueError(f"Beta assigned test file {spotlight.location} is missing in the repository execution environment.")
|
|
if current is not None and repo_exec.exists(test_path):
|
|
testing_item.focused_spotlight = current
|
|
return current
|
|
if current is not None:
|
|
repo_exec.mkdir(test_path.parent, parents=True, exist_ok=True)
|
|
repo_exec.write_text(test_path, "")
|
|
testing_item.focused_spotlight = current
|
|
return current
|
|
if repo_exec.exists(test_path) and testing_item.change_type is not ChangeType.Create:
|
|
refreshed = file_manager.refresh_test(
|
|
spotlight.location,
|
|
"Beta resolved assigned test file metadata before edit.",
|
|
)
|
|
testing_item.focused_spotlight = refreshed
|
|
return refreshed
|
|
|
|
repo_exec.mkdir(test_path.parent, parents=True, exist_ok=True)
|
|
if not repo_exec.exists(test_path):
|
|
repo_exec.write_text(test_path, "")
|
|
temporary = self._bootstrap_test_classmeta(testing_item, spotlight, test_path)
|
|
file_manager._upsert(temporary, file_manager.test_classes, file_manager.test_index)
|
|
file_manager.test_dirty[temporary.location] = True
|
|
current = file_manager.current_test(temporary.location) or temporary
|
|
testing_item.focused_spotlight = current
|
|
return current
|
|
|
|
def _bootstrap_test_classmeta(
|
|
self,
|
|
testing_item: TestingBacklog,
|
|
spotlight: ClassMeta,
|
|
test_path: Path,
|
|
) -> ClassMeta:
|
|
"""Bootstrap temporary ClassMeta for an already-assigned Beta test target."""
|
|
|
|
file_manager = self.context.file_manager
|
|
downstream_locations = [
|
|
source.location
|
|
for source in testing_item.downstream_classes
|
|
]
|
|
tested_target = ", ".join(downstream_locations) or "assigned source"
|
|
summary = (
|
|
spotlight.summary.strip()
|
|
or testing_item.summary.strip()
|
|
or f"Create test coverage for {tested_target}."
|
|
)
|
|
temporary = ClassMeta(
|
|
location=file_manager.test_location(test_path),
|
|
summary=summary,
|
|
relevance=[
|
|
summary,
|
|
(
|
|
"Beta create-ticket placeholder; this new test file is assigned "
|
|
f"to validate {tested_target}."
|
|
),
|
|
],
|
|
relevant_symbols={},
|
|
cdc=list(spotlight.cdc),
|
|
content_type=ContentType.Test,
|
|
)
|
|
return temporary
|
|
|
|
def _refresh_changed_tests(
|
|
self,
|
|
testing_item: TestingBacklog,
|
|
changed_paths: list[Path],
|
|
prompt: str,
|
|
tailing_logs: str = "",
|
|
) -> ClassMeta:
|
|
"""Refresh changed editable test metadata after Aider edits disk."""
|
|
|
|
refreshed: ClassMeta | None = None
|
|
for path in changed_paths:
|
|
try:
|
|
location = self.context.file_manager.test_location(path)
|
|
except ValueError:
|
|
continue
|
|
|
|
refreshed = self.context.file_manager.refresh_test(location, prompt, tailing_logs)
|
|
|
|
if testing_item.focused_spotlight is None:
|
|
raise ValueError("TestingBacklog requires focused_spotlight before test refresh.")
|
|
test_file = testing_item.focused_spotlight
|
|
test_path = self.context.file_manager.test_path(test_file.location)
|
|
if refreshed is None and self.context.file_manager.repo_exec.exists(test_path):
|
|
refreshed = self.context.file_manager.refresh_test(
|
|
test_file.location,
|
|
prompt,
|
|
tailing_logs,
|
|
)
|
|
|
|
if refreshed is not None:
|
|
testing_item.focused_spotlight = refreshed
|
|
return refreshed
|
|
|
|
if testing_item.focused_spotlight is None:
|
|
raise ValueError("TestingBacklog requires focused_spotlight after test refresh.")
|
|
return testing_item.focused_spotlight
|
|
|
|
def _apply_test_work(
|
|
self,
|
|
testing_item: TestingBacklog,
|
|
work_needed: str,
|
|
) -> AppliedTestWork:
|
|
"""Run Aider for one test item and return the bounded edit result."""
|
|
|
|
if testing_item.focused_spotlight is None:
|
|
raise ValueError("TestingBacklog requires focused_spotlight before Beta edit.")
|
|
test_file = testing_item.focused_spotlight
|
|
test_path = self.context.file_manager.test_path(test_file.location)
|
|
if not self.context.file_manager.repo_exec.exists(test_path):
|
|
raise ValueError(
|
|
f"Beta assigned test file {test_file.location} is missing on disk in the repository execution environment; "
|
|
"test files must be created by preflight before Beta edit."
|
|
)
|
|
backlog_guidance = [work_needed]
|
|
backlog_guidance.append(f"Change type: {testing_item.change_type.name}")
|
|
readonly_files = self._readonly_paths(testing_item)
|
|
request = AiderEditRequest(
|
|
editor_records=list(testing_item.editor_records),
|
|
authoritative_prompt=self._authoritative_prompt([test_path]),
|
|
editable_files=[test_path],
|
|
readonly_files=readonly_files,
|
|
pinned_context=self.context.pinned_context,
|
|
backlog_guidance=backlog_guidance,
|
|
future_guidance=testing_item.future_guidance,
|
|
previous_cot_records=self.previous_cot_records,
|
|
failure_error_message=testing_item.failure_error_message,
|
|
failure_stack_trace=list(testing_item.failure_stack_trace),
|
|
failure_raw_output=testing_item.failure_raw_output,
|
|
failure_test_location=test_file.location,
|
|
failure_test_name=testing_item.failure_test_name,
|
|
failure_outcome=testing_item.failure_outcome,
|
|
failure_run_info=testing_item.failure_run_info,
|
|
allowed_additional_roots=[self.context.file_manager.test_root],
|
|
context=BetaLoop,
|
|
frontier_provider=self.frontier_provider,
|
|
)
|
|
# The item handler admits this operation only with a configured editor.
|
|
worker = cast(CodeEditor, self.aider_worker)
|
|
try:
|
|
result = worker.edit(request, logger=self.logger.getChild("aider"))
|
|
except TypeError as error:
|
|
if "logger" not in str(error):
|
|
raise
|
|
result = worker.edit(request)
|
|
changed_paths, summary = result.changed_paths, result.summary
|
|
testing_item.editor_records = list(result.records)
|
|
summary = consume_code_edit_leave_summary(summary)
|
|
self._refresh_changed_tests(
|
|
testing_item,
|
|
changed_paths,
|
|
self._authoritative_prompt([test_path]),
|
|
summary,
|
|
)
|
|
return AppliedTestWork(
|
|
testing_item=testing_item,
|
|
work_needed=work_needed,
|
|
test_path=test_path,
|
|
changed_paths=tuple(changed_paths),
|
|
summary=summary,
|
|
)
|
|
|
|
def _targeted_failure_story(
|
|
self,
|
|
testing_item: TestingBacklog,
|
|
work_needed: str,
|
|
applied: AppliedTestWork,
|
|
result: TestRunResult,
|
|
failing: FailingTest,
|
|
) -> TargetedFailureStory:
|
|
"""Build the shared targeted-failure routing story for one Beta postflight failure."""
|
|
|
|
if testing_item.focused_spotlight is None:
|
|
raise ValueError("TestingBacklog requires focused_spotlight before failure routing.")
|
|
candidate_by_location = {
|
|
source.location: source
|
|
for source in [
|
|
*testing_item.downstream_classes,
|
|
*testing_item.upstream_classes,
|
|
*testing_item.failure_context_sources,
|
|
]
|
|
}
|
|
candidate_sources = list(candidate_by_location.values())
|
|
return TargetedFailureStory(
|
|
testing_item=testing_item,
|
|
raw_output=result.output,
|
|
work_needed=work_needed,
|
|
candidate_sources=candidate_sources,
|
|
failing=failing,
|
|
failure_phase="after_beta",
|
|
structural_evidence={},
|
|
semantic_evidence={"phase": "after_beta"},
|
|
previous_beta_guidance=work_needed,
|
|
beta_edit_diff=testing_item.focused_spotlight.get_diff(self.context.file_manager),
|
|
beta_editor_summary=applied.summary,
|
|
)
|
|
|
|
def _record_targeted_failure(
|
|
self,
|
|
testing_item: TestingBacklog,
|
|
result: TestRunResult,
|
|
) -> FailingTest:
|
|
"""Replace one testing item's failure record from its just-completed targeted run."""
|
|
|
|
# Reject recording until the targeted test provides the failure record's identity.
|
|
if testing_item.focused_spotlight is None:
|
|
raise ValueError("TestingBacklog requires focused_spotlight before failure recording.")
|
|
reports = self.test_runner_strategy.failure_reports(
|
|
[testing_item.focused_spotlight], result,
|
|
)
|
|
# A file error covers the whole known file, including simultaneous call failures.
|
|
failure = next((report for report in reports if (report.outcome is not None and report.outcome.is_error)), reports[0])
|
|
testing_item.record_failure(
|
|
failure,
|
|
curate_failure_context_sources(
|
|
file_manager=self.context.file_manager,
|
|
failure=failure,
|
|
future_guidance=testing_item.future_guidance,
|
|
previous_cot_records=self.previous_cot_records,
|
|
),
|
|
)
|
|
return failure
|
|
|
|
def _run_targeted_tests(self, testing_item: TestingBacklog) -> TestRunResult:
|
|
"""Validate the edited test file, stopping at its first failure."""
|
|
|
|
if testing_item.focused_spotlight is None:
|
|
raise ValueError("TestingBacklog requires focused_spotlight before targeted tests.")
|
|
result = self.test_runner_strategy.run_tests_for(
|
|
[testing_item.focused_spotlight],
|
|
max_failures=1,
|
|
test_names=(),
|
|
)
|
|
if (testing_item.failure_outcome is not None and testing_item.failure_outcome.is_error) and result.returncode == 5:
|
|
return replace(result, outcome=TestRunOutcome.Error)
|
|
return result
|
|
|
|
def _raise_for_targeted_test_failure(
|
|
self,
|
|
testing_item: TestingBacklog,
|
|
work_needed: str,
|
|
result: TestRunResult,
|
|
applied: AppliedTestWork,
|
|
) -> None:
|
|
"""Route one targeted test failure after Beta makes an explicit Alpha/Beta judgement."""
|
|
|
|
if testing_item.focused_spotlight is None:
|
|
raise ValueError("TestingBacklog requires focused_spotlight before failure routing.")
|
|
test_file = testing_item.focused_spotlight
|
|
failing = self._record_targeted_failure(testing_item, result)
|
|
story = self._targeted_failure_story(
|
|
testing_item,
|
|
work_needed,
|
|
applied,
|
|
result,
|
|
failing,
|
|
)
|
|
# Preserve collection-specific ownership semantics at the shared routing boundary.
|
|
if (failing.outcome is not None and failing.outcome.is_error):
|
|
routing = self.context.backlog.targeted_collection_alpha_or_beta_routing_check(
|
|
story=story,
|
|
logger=self.logger.getChild("backlog"),
|
|
)
|
|
assignments = routing.assignments
|
|
else:
|
|
routing = self.context.backlog.targeted_alpha_or_beta_routing_check(
|
|
story=story,
|
|
logger=self.logger.getChild("backlog"),
|
|
)
|
|
assignments = ()
|
|
source_assignments = tuple(
|
|
assignment
|
|
for assignment in assignments
|
|
if assignment.outcome == "source"
|
|
)
|
|
# Only established source ownership may leave the selected test's existing Beta lane.
|
|
if routing.outcome != "source":
|
|
retry_item = TestingBacklog(
|
|
editor_records=list(testing_item.editor_records),
|
|
message_id=testing_item.message_id,
|
|
summary=testing_item.summary,
|
|
focused_spotlight=test_file,
|
|
upstream_classes=list(testing_item.upstream_classes),
|
|
downstream_classes=list(testing_item.downstream_classes),
|
|
original_cot=testing_item.original_cot,
|
|
previous_beta_guidance="\n\n".join(filter(None, (work_needed, applied.summary))),
|
|
change_type=testing_item.change_type,
|
|
failure_error_message=failing.error_message,
|
|
failure_stack_trace=list(failing.stack_trace),
|
|
failure_raw_output=failing.raw_output,
|
|
failure_test_name=failing.test_name,
|
|
failure_outcome=failing.outcome,
|
|
failure_run_info=failing.run_info,
|
|
failure_context_sources=list(testing_item.failure_context_sources),
|
|
future_guidance=testing_item.future_guidance,
|
|
)
|
|
raise BetaRequeueRequested(
|
|
f"Beta requeued {test_file.location} after Beta-owned targeted failure: {routing.reason}",
|
|
retry_item,
|
|
)
|
|
|
|
source_guidance = {
|
|
clue["location"]: clue["reason"]
|
|
for assignment in source_assignments
|
|
for clue in assignment.source_clues
|
|
if clue["location"]
|
|
}
|
|
created = self.context.backlog.report_remediation(
|
|
summary="\n\n".join(
|
|
part
|
|
for part in (
|
|
f"Targeted test command failed for {test_file.location}.",
|
|
*(assignment.reason for assignment in source_assignments),
|
|
)
|
|
if part.strip()
|
|
),
|
|
testing_item=testing_item,
|
|
failed_test=failing,
|
|
candidates=story.candidate_sources,
|
|
source_guidance=source_guidance,
|
|
routing=routing,
|
|
queue=False,
|
|
logger=self.logger.getChild("backlog"),
|
|
)
|
|
# A confirmed source culprit proceeds to Alpha; a confirmed contradiction returns to Beta.
|
|
for remediation_item in created:
|
|
# FailedQAItem inherits TestingBacklog, so production attribution is checked first.
|
|
if isinstance(remediation_item, FailedQAItem):
|
|
self.context.stage_remediation_outbound(remediation_item)
|
|
self.context.persist()
|
|
else:
|
|
raise BetaRequeueRequested(
|
|
f"Beta retained {test_file.location} after higher-tier FileBlame review: {remediation_item.previous_beta_guidance}",
|
|
remediation_item,
|
|
)
|
|
raise BetaRemediationRequested(
|
|
f"Beta routed {test_file.location} to source remediation: {routing.reason}",
|
|
)
|
|
|
|
def _report_all_tests_failure(
|
|
self,
|
|
result: TestRunResult,
|
|
) -> None:
|
|
"""Execute AST-matched methods individually and stage fresh owner-specific work."""
|
|
|
|
file_manager = self.context.file_manager
|
|
test_files = dict(map(lambda test: (test.location, test), file_manager.test_classes))
|
|
reports = self.test_runner_strategy.failure_reports(list(test_files.values()), result)
|
|
bi_dependency_strategy = cast(BiDependencyDiscoveryStrategy, LanguageStrategySet.get(file_manager)[
|
|
BiDependencyDiscoveryStrategy
|
|
])
|
|
queued = 0
|
|
# Discovery selects candidates; only an individual execution establishes repair evidence.
|
|
for report in reports:
|
|
location, name = report.test.location, report.test_name
|
|
test_file = report.test
|
|
sources = list(bi_dependency_strategy.upstream_sources_for(
|
|
[test_file], file_manager.src_classes, file_manager.src_root,
|
|
file_manager.test_root, recursive=True,
|
|
))
|
|
observed = (replace(result, outcome=report.outcome, output=report.raw_output,
|
|
stack_trace=list(report.stack_trace), test_name="")
|
|
if (report.outcome is not None and report.outcome.is_error) else
|
|
self.test_runner_strategy.run_tests_for([test_file], max_failures=1, test_names=(name,)))
|
|
# Empty collection is not execution; collection errors still belong to routing.
|
|
if observed.passed and observed.returncode == 0:
|
|
self.logger.info("Final-exam method passed: %s::%s.", location, name)
|
|
else:
|
|
failure = FailingTest.from_result(test_file, replace(observed, test_name=name if observed.outcome is TestRunOutcome.Failure else ""))
|
|
context_sources = working_backlog.curate_failure_context_sources(
|
|
file_manager=file_manager, failure=failure,
|
|
previous_cot_records=self.previous_cot_records,
|
|
)
|
|
candidates = list(dict(map(lambda source: (source.location, source), (*sources, *context_sources))).values())
|
|
summary = f"Final-exam targeted execution failed for {location}::{name}."
|
|
item = TestingBacklog(
|
|
test_file, summary, downstream_classes=sources,
|
|
original_cot=file_manager.prompt, change_type=ChangeType.Update,
|
|
)
|
|
item.record_failure(failure, context_sources)
|
|
story = TargetedFailureStory(
|
|
testing_item=item, raw_output=observed.output, work_needed=summary,
|
|
candidate_sources=candidates, failing=failure, failure_phase="last_exam",
|
|
)
|
|
service = self.context.backlog.judgement_service
|
|
# Collection has its established ownership rules even for a named invocation.
|
|
if observed.outcome.is_error:
|
|
routing = service.route_collection_failure(
|
|
file_manager=file_manager, story=story,
|
|
previous_cot_records=self.previous_cot_records, logger=self.logger,
|
|
)
|
|
else:
|
|
routing = service.route_targeted_failure(
|
|
file_manager=file_manager, story=story,
|
|
previous_cot_records=self.previous_cot_records, logger=self.logger,
|
|
)
|
|
self.context.backlog.clear_ledger(map(lambda source: (source, test_file), sources), logger=self.logger)
|
|
# Follow Alpha postflight: choose the owner before staging either repair lane.
|
|
if routing.outcome == "source":
|
|
repairs = self.context.backlog.report_remediation(
|
|
testing_item=item, summary=summary + "\n" + routing.reason,
|
|
failed_test=failure, candidates=candidates, queue=False,
|
|
failure_phase="last_exam", logger=self.logger,
|
|
routing=routing,
|
|
)
|
|
# Persist every attributed method before advancing to the next candidate.
|
|
for repair in repairs:
|
|
# Final-exam work follows the same confirmed attribution as targeted work.
|
|
if isinstance(repair, FailedQAItem):
|
|
self.context.stage_remediation_outbound(repair)
|
|
else:
|
|
self.context.stage_testing_outbound(repair)
|
|
self.context.persist()
|
|
queued += 1
|
|
else:
|
|
self.context.stage_testing_outbound(item)
|
|
self.context.persist()
|
|
queued += 1
|
|
# This change routes discovered methods without redefining suite-exhaustion policy.
|
|
if queued:
|
|
self._blocked_summary = f"Full test-suite failure queued {queued} method repair item(s)."
|
|
elif not reports:
|
|
self._blocked_summary = "Running the full test suite failed, but no current test method matched the output."
|
|
|
|
def _route_post_edit(self, story: TargetedFailureStory) -> str:
|
|
"""Assign a new post-edit failure without reopening the completed QA claim."""
|
|
|
|
item = story.testing_item
|
|
# Beta post-edit routing always owns a fresh QA item before choosing its repair lane.
|
|
if item is None:
|
|
raise ValueError("Beta post-edit routing requires a QA item.")
|
|
routing = self.context.backlog.targeted_alpha_or_beta_routing_check(
|
|
story=story, logger=self.logger.getChild("backlog"),
|
|
)
|
|
# Reuse attribution and contradiction review, without unfinished-item exceptions.
|
|
if routing.outcome == "source":
|
|
repairs = self.context.backlog.report_remediation(
|
|
testing_item=item, summary=story.work_needed + "\n" + routing.reason,
|
|
failed_test=story.failing, candidates=story.candidate_sources,
|
|
routing=routing, queue=False, logger=self.logger.getChild("backlog"),
|
|
)
|
|
# A confirmed contradiction stages new Beta work; source ownership stages Alpha.
|
|
for repair in repairs:
|
|
# FailedQAItem specializes TestingBacklog, so test production ownership first.
|
|
if isinstance(repair, FailedQAItem):
|
|
self.context.stage_remediation_outbound(repair)
|
|
else:
|
|
self.context.stage_testing_outbound(repair)
|
|
else:
|
|
self.context.stage_testing_outbound(item)
|
|
return (
|
|
f"Beta post-edit routed new failure {story.failing.test.location}::"
|
|
f"{story.failing.test_name}: {routing.reason}"
|
|
)
|
|
|
|
def _prepare_post_edit_failure(
|
|
self,
|
|
testing_item: TestingBacklog,
|
|
result: TestingItemResult,
|
|
) -> TargetedFailureStory:
|
|
"""Prepare new QA work for ``run`` to register after closing the old claim.
|
|
|
|
The workflow supplies a result with a new failure and binds the old
|
|
provenance item while gathering carried context. Registration and
|
|
ownership routing wait until the workflow has closed the old claim.
|
|
"""
|
|
|
|
test_file = cast(ClassMeta, testing_item.focused_spotlight)
|
|
failing = cast(FailingTest, result.new_failure)
|
|
item = TestingBacklog(
|
|
focused_spotlight=testing_item.focused_spotlight,
|
|
summary="FAILED CANDIDATE",
|
|
upstream_classes=list(testing_item.upstream_classes),
|
|
downstream_classes=list(testing_item.downstream_classes),
|
|
future_guidance=testing_item.future_guidance,
|
|
editor_records=list(testing_item.editor_records),
|
|
original_cot=testing_item.original_cot,
|
|
change_type=ChangeType.Update,
|
|
)
|
|
item.record_failure(failing, [
|
|
*admitted_sources.current_sources(self.context.file_manager, testing_item),
|
|
*working_backlog.curate_failure_context_sources(
|
|
file_manager=self.context.file_manager, failure=failing,
|
|
future_guidance=item.future_guidance,
|
|
previous_cot_records=self.previous_cot_records,
|
|
),
|
|
])
|
|
candidates = list(dict(map(lambda source: (source.location, source), (
|
|
*item.downstream_classes, *item.upstream_classes, *item.failure_context_sources,
|
|
))).values())
|
|
story = TargetedFailureStory(
|
|
testing_item=item, raw_output=result.targeted.output,
|
|
work_needed=(f"Route the new failure {test_file.location}::{failing.test_name} "
|
|
"after the previous Beta obligation completed."),
|
|
candidate_sources=candidates, failing=failing, failure_phase="after_beta",
|
|
beta_edit_diff=test_file.get_diff(self.context.file_manager),
|
|
beta_editor_summary=result.applied.summary,
|
|
)
|
|
# Cleanup only: downstream decisions do not read this old item's failure fields again.
|
|
# The QA claim owns completion; routing uses the new item and story.failing above.
|
|
testing_item.clear_failure()
|
|
return story
|
|
|
|
def _handle_testing_item(
|
|
self,
|
|
testing_item: TestingBacklog,
|
|
) -> TestingItemResult:
|
|
"""Repair and validate one obligation, reporting completion and any new failure."""
|
|
|
|
if self.aider_worker is None:
|
|
raise BetaRequeueRequested(
|
|
"Beta test worker seam is unavailable.",
|
|
testing_item,
|
|
blocked_summary="Beta test worker seam is unavailable.",
|
|
stop=True,
|
|
)
|
|
|
|
test_file = self._ensure_test_file(testing_item)
|
|
work_needed = self._refine_beta_instruction(testing_item)
|
|
self.context.backlog.claim_testing_pair(
|
|
testing_item,
|
|
logger=self.logger.getChild("backlog"),
|
|
)
|
|
applied = self._apply_test_work(
|
|
testing_item,
|
|
work_needed,
|
|
)
|
|
|
|
targeted = self._run_targeted_tests(testing_item)
|
|
new_failure = None
|
|
# Only an ordinary, differently named failure completes the old named obligation.
|
|
if targeted.outcome is TestRunOutcome.Failure and testing_item.failure_test_name:
|
|
reports = self.test_runner_strategy.failure_reports([test_file], targeted)
|
|
failing = next((report for report in reports if report.outcome is not None and report.outcome.is_error), reports[0])
|
|
# File errors and the same identity retain the existing unfinished-obligation path.
|
|
if (failing.outcome is TestRunOutcome.Failure and failing.test_name
|
|
and failing.test_name != testing_item.failure_test_name):
|
|
new_failure = failing
|
|
# A passing current execution invalidates any failure evidence that entered this Beta attempt.
|
|
if targeted.passed:
|
|
testing_item.clear_failure()
|
|
|
|
downstream_locations = [
|
|
source.location
|
|
for source in testing_item.downstream_classes
|
|
]
|
|
# Report obligation completion separately from the whole-file execution outcome.
|
|
judgement = "PASS" if targeted.passed or new_failure is not None else "FAIL"
|
|
summary = (
|
|
f"Beta handled {test_file.location} for "
|
|
f"{', '.join(downstream_locations)} with judgement {judgement}."
|
|
)
|
|
return TestingItemResult(summary, applied, targeted, new_failure)
|
|
|
|
def _handle_last_exam(self) -> str:
|
|
"""Run final full-suite validation and queue any discovered test work."""
|
|
|
|
result = self.test_runner_strategy.run_all_tests_under(
|
|
self.context.file_manager.test_root
|
|
)
|
|
if result.passed:
|
|
return "Beta completed last exam."
|
|
|
|
self._report_all_tests_failure(result)
|
|
return "Beta converted last exam failure into targeted test work."
|
|
|
|
def _prepare_work(self) -> ProtoLoop | None:
|
|
"""Return no child loop because Beta drains QABacklog directly."""
|
|
|
|
return None
|
|
|
|
def _post_run(self) -> str:
|
|
"""Return a concise beta summary after the QA backlog is drained or blocked."""
|
|
|
|
handled = len(self.context.marginal_result)
|
|
if self._blocked_summary:
|
|
return f"Beta incomplete: {self._blocked_summary}"
|
|
|
|
if self.context.backlog.has_pending_work():
|
|
return "Beta incomplete: backlog work remains queued."
|
|
|
|
if handled == 0:
|
|
return "Beta found no testing backlog work."
|
|
|
|
return f"Beta completed {handled} testing backlog item(s)."
|
|
|
|
def _publish_queue_depth(self) -> None:
|
|
"""Publish one live queue-depth snapshot when Stabilization provided a callback."""
|
|
|
|
callback = getattr(self.context, "runtime_queue_depth_callback", None)
|
|
if callable(callback):
|
|
callback()
|
|
|
|
def _publish_work_status(self, item: TestingBacklog) -> None:
|
|
"""Publish the concrete test item claimed by Beta."""
|
|
|
|
spotlight = item.focused_spotlight
|
|
location = spotlight.location if spotlight is not None else "Full test suite"
|
|
symbols = tuple(spotlight.relevant_symbols) if spotlight is not None else ()
|
|
# Publish a narrower focus only when metadata identifies one unambiguous symbol.
|
|
focus = symbols[0] if len(symbols) == 1 else None
|
|
publish_loop_status(
|
|
previous_loop="StabilizationLoop",
|
|
current_loop=type(self).__name__,
|
|
current_item=location,
|
|
work_focus=focus,
|
|
)
|
|
|
|
def _restart_item(self, error: DeadEndError) -> None:
|
|
"""Revive the selected phase start after abandoning Beta's current item."""
|
|
|
|
if self.provenance is None:
|
|
raise error
|
|
self.context.discard_active_inbound()
|
|
phase, revived = self.provenance.revive(error)
|
|
if phase is Phase.BETA:
|
|
if not isinstance(revived, TestingBacklog):
|
|
raise TypeError("Beta restart requires testing backlog work.")
|
|
self.context.backlog.queue_testing(revived, prepend=True, logger=self.logger)
|
|
elif isinstance(revived, FailedQAItem):
|
|
self.context.backlog.prepend_remediation(revived, logger=self.logger)
|
|
elif isinstance(revived, InspectionBacklog):
|
|
self.context.backlog.prepend_inspection(revived, logger=self.logger)
|
|
else:
|
|
raise TypeError("Alpha restart requires source backlog work.")
|
|
self.logger.info("Beta %s revived %s phase work.", error.direction.value, phase.value)
|
|
|
|
def run(self, logger_name: str | None = None) -> str:
|
|
"""Drain QA backlog work through QABacklog and return the beta summary."""
|
|
|
|
if not self._logger_stack: self._logger_stack.append(llm_logging.logger_for_loop(self))
|
|
if logger_name: self._logger_stack.append(self.logger.getChild(llm_logging.safe_log_segment(logger_name)))
|
|
llm_logging.set_current_logger(self.logger)
|
|
try:
|
|
if not self.gate.may_enter(): return self._post_run()
|
|
phase_binding = provenance.bind(self.provenance, self.logger) if self.provenance is not None else nullcontext()
|
|
with phase_binding:
|
|
self.context.recover_active_inbound(logger=self.logger.getChild("backlog"))
|
|
self.context.persist()
|
|
self._publish_queue_depth()
|
|
while True:
|
|
# Resume ownership routing without rerunning the completed item's editor.
|
|
if self.context.post_edit is not None:
|
|
story = self.context.post_edit
|
|
phase_binding = provenance.bind(self.provenance, self.logger) if self.provenance is not None else nullcontext()
|
|
try:
|
|
with phase_binding:
|
|
with provenance.item_scope(Phase.BETA, cast(TestingBacklog, story.testing_item)):
|
|
local_summary = self._route_post_edit(story)
|
|
except DeadEndError as error:
|
|
self._restart_item(error)
|
|
self.context.persist()
|
|
self._publish_queue_depth()
|
|
continue
|
|
except Exception:
|
|
self.logger.exception("Beta retained new post-edit failure for ownership routing.")
|
|
raise
|
|
remediation = any(item.lane is BacklogLane.Remediation for item in self.context.outbound_queue)
|
|
self.context.marginal_result.append(local_summary)
|
|
self.context.persist()
|
|
self.context.flush_outbound_with_wal(logger=self.logger.getChild("backlog"))
|
|
self.context.persist()
|
|
self._publish_queue_depth()
|
|
# A confirmed production owner returns to Alpha before more Beta editing.
|
|
if remediation:
|
|
break
|
|
continue
|
|
if self.context.inbound_queue and self.context.active_inbound is not None:
|
|
testing_item = self.context.active_inbound.value
|
|
claimed = None
|
|
else:
|
|
phase_binding = provenance.bind(self.provenance, self.logger) if self.provenance is not None else nullcontext()
|
|
with phase_binding:
|
|
claimed = self.context.move_inbound_from_backlog(
|
|
logger=self.logger.getChild("backlog")
|
|
)
|
|
self.context.persist()
|
|
self._publish_queue_depth()
|
|
if claimed is None: break
|
|
testing_item = claimed.value
|
|
self._publish_work_status(testing_item)
|
|
pending_story = None
|
|
try:
|
|
# Keep final-exam dispatch and QA claim ownership in the workflow.
|
|
with (claimed if claimed is not None else nullcontext()) as scope:
|
|
phase_binding = provenance.bind(self.provenance, self.logger) if self.provenance is not None else nullcontext()
|
|
with phase_binding:
|
|
with provenance.item_scope(Phase.BETA, testing_item):
|
|
# Suite-wide work is distinct from repairing one ordinary obligation.
|
|
if testing_item.last_exam:
|
|
local_summary = self._handle_last_exam()
|
|
self.context.marginal_result.append(local_summary)
|
|
else:
|
|
handled = self._run_one_testing_item(testing_item)
|
|
local_summary = handled.summary
|
|
# Unfinished work routes within its existing claim, before completion.
|
|
if not handled.passed:
|
|
self._raise_for_targeted_test_failure(
|
|
testing_item, handled.applied.work_needed,
|
|
handled.targeted, handled.applied,
|
|
)
|
|
# Prepare new work while its predecessor's evidence is still bound.
|
|
elif handled.new_failure is not None:
|
|
pending_story = self._prepare_post_edit_failure(testing_item, handled)
|
|
# Only a freshly claimed item has a claim scope to commit here.
|
|
if scope is not None:
|
|
scope.commit()
|
|
except DeadEndError as error:
|
|
self._restart_item(error)
|
|
self.context.persist()
|
|
self._publish_queue_depth()
|
|
continue
|
|
except BetaRequeueRequested as exc:
|
|
self.logger.info("%s", exc.summary);
|
|
phase_binding = provenance.bind(self.provenance, self.logger) if self.provenance is not None else nullcontext()
|
|
with phase_binding:
|
|
self.context.stage_testing_outbound(exc.testing_item, prepend=True)
|
|
self.context.persist()
|
|
if exc.blocked_summary: self._blocked_summary = exc.blocked_summary
|
|
self.context.marginal_result.append(exc.summary);
|
|
if not exc.stop:
|
|
self.context.flush_outbound_with_wal(logger=self.logger.getChild("backlog"))
|
|
self.context.persist()
|
|
self._publish_queue_depth()
|
|
continue
|
|
self.context.flush_outbound_with_wal(logger=self.logger.getChild("backlog"))
|
|
self.context.persist()
|
|
self._publish_queue_depth()
|
|
break
|
|
except BetaRemediationRequested as exc:
|
|
self.logger.info("%s", exc.summary);
|
|
self.context.marginal_result.append(exc.summary);
|
|
break
|
|
except Exception:
|
|
location = (
|
|
testing_item.focused_spotlight.location
|
|
if testing_item.focused_spotlight is not None
|
|
else "[missing test]"
|
|
)
|
|
self.logger.exception("Beta rolled back claimed testing work for %s.",
|
|
location);
|
|
raise
|
|
# Only the workflow registers new QA work, after the old claim has closed.
|
|
if pending_story is not None:
|
|
phase_binding = provenance.bind(self.provenance, self.logger) if self.provenance is not None else nullcontext()
|
|
with phase_binding:
|
|
self.context.start_post_edit(pending_story)
|
|
self.context.marginal_result.append(
|
|
f"Beta completed the prior obligation in {pending_story.failing.test.location} "
|
|
f"with judgement PASS; discovered {pending_story.failing.test_name}."
|
|
)
|
|
self.context.persist()
|
|
self._publish_queue_depth()
|
|
continue
|
|
if testing_item.last_exam:
|
|
if self.context.outbound_queue:
|
|
self.context.flush_outbound_with_wal(logger=self.logger.getChild("backlog"))
|
|
self.context.persist()
|
|
self._publish_queue_depth()
|
|
if self.context.backlog.has_pending_work():
|
|
break
|
|
if self.context.active_inbound is not None and not self.context.outbound_queue:
|
|
self.context.clear_moved_inbound()
|
|
self.context.persist()
|
|
self._publish_queue_depth()
|
|
continue
|
|
self.context.marginal_result.append(local_summary)
|
|
if self.context.active_inbound is not None and not self.context.outbound_queue:
|
|
self.context.clear_moved_inbound()
|
|
self.context.persist()
|
|
self._publish_queue_depth()
|
|
else:
|
|
self.context.move_outbound()
|
|
self.context.persist()
|
|
self._publish_queue_depth()
|
|
self.context.flush_outbound_with_wal(logger=self.logger.getChild("backlog"))
|
|
self.context.persist()
|
|
self._publish_queue_depth()
|
|
return self._post_run()
|
|
finally:
|
|
if logger_name: self._logger_stack.pop()
|
|
llm_logging.set_current_logger(self.logger if self._logger_stack else None)
|