890 lines
41 KiB
Python
890 lines
41 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
from __future__ import annotations
|
|
|
|
from contextlib import nullcontext
|
|
from dataclasses import dataclass, field
|
|
from itertools import filterfalse
|
|
from pathlib import Path
|
|
|
|
from band_aider.ai_client import llm_logging
|
|
from band_aider.context import admitted_sources, provenance
|
|
from band_aider.context.memento import Memento
|
|
from band_aider.context.alpha_context import AlphaContext
|
|
from band_aider.context.class_meta import ClassMeta, ContentType
|
|
from band_aider.context.cot_task_generator import CotTaskRecord
|
|
from band_aider.context.provenance import DeadEndError, Phase, Provenance
|
|
from band_aider.context.failing_test import FailingTest
|
|
from band_aider.context.working_backlog import (
|
|
BacklogLane,
|
|
ContributorBacklog,
|
|
FailedQAItem,
|
|
InspectionBacklog,
|
|
OutboundBacklogItem,
|
|
StagedInboundItem,
|
|
TestingBacklog,
|
|
)
|
|
from band_aider.judgement.alpha_edit_acceptance_check import (
|
|
AlphaEditAcceptanceDecision,
|
|
)
|
|
from band_aider.judgement.alpha_blank_source_correction_check import (
|
|
AlphaBlankSourceCorrectionDecision,
|
|
)
|
|
from band_aider.judgement.edit_request import AiderEditRequest
|
|
from band_aider.context.validation_receipts import ValidationReceipts
|
|
from band_aider.judgement.code_edit_or_leave_it import (
|
|
CodeEditLeaveSummary,
|
|
consume_code_edit_leave_summary,
|
|
is_code_edit_leave_summary,
|
|
)
|
|
from band_aider.judgement.code_editor import CodeEditor, CodeEditorResult
|
|
from band_aider.judgement.search_replace_token_generator import RepairEvidenceBuilder
|
|
from band_aider.judgement.source_preflight_decision import AlphaSourcePreflightDecision
|
|
from band_aider.judgement.test_runner_strategy import TestRunnerStrategy
|
|
from band_aider.judgement.vibe_check import ReasonableVibeCheck
|
|
from band_aider.loop.loop_gate import LoopGate
|
|
from band_aider.loop.proto_loop import ProtoLoop
|
|
from band_aider.strategies.compiler import FrontierProvider
|
|
from band_aider.ui.status import publish_loop_status
|
|
|
|
|
|
@dataclass(init=False)
|
|
class AlphaLoop(ProtoLoop[LoopGate, AlphaContext]):
|
|
"""AlphaLoop coordinates source-only work and communicates test needs through ContributorBacklog."""
|
|
|
|
aider_worker: CodeEditor | None = None
|
|
test_runner_strategy: TestRunnerStrategy
|
|
frontier_provider: FrontierProvider | None = None
|
|
previous_cot_root_readonly_file: Path | None = field(default=None, init=False)
|
|
previous_cot_records: tuple[CotTaskRecord, ...] = field(default=(), init=False, repr=False)
|
|
new_files_callback = None
|
|
_handled_sources: list[ClassMeta] = field(default_factory=list, init=False, repr=False)
|
|
_source_summary: str = field(default="", init=False, repr=False)
|
|
provenance: Provenance | None = field(default=None, init=False, repr=False)
|
|
|
|
def __init__(
|
|
self,
|
|
gate: LoopGate,
|
|
context: AlphaContext,
|
|
aider_worker: CodeEditor | None = None,
|
|
frontier_provider: FrontierProvider | None = None,
|
|
previous_cot_root_readonly_file: Path | None = None,
|
|
previous_cot_records: tuple[CotTaskRecord, ...] = (),
|
|
provenance: Provenance | None = None,
|
|
logger=None,
|
|
) -> None:
|
|
super().__init__(gate=gate, context=context, logger=logger)
|
|
self.aider_worker = aider_worker
|
|
self.frontier_provider = frontier_provider
|
|
self.previous_cot_root_readonly_file = previous_cot_root_readonly_file
|
|
self.previous_cot_records = previous_cot_records
|
|
self.test_runner_strategy = context.test_runner_strategy # pyright: ignore[reportAttributeAccessIssue]
|
|
self.new_files_callback = context.new_files_callback # pyright: ignore[reportAttributeAccessIssue]
|
|
self._handled_sources = []
|
|
self._source_summary = ""
|
|
self.provenance = provenance
|
|
self.validation_receipts = ValidationReceipts()
|
|
|
|
def _source_path(self, source_file: ClassMeta) -> Path:
|
|
"""Return the absolute path for one source metadata record."""
|
|
|
|
return self.context.file_manager.source_path(source_file.location)
|
|
|
|
def _readonly_paths(
|
|
self,
|
|
source_files: list[ClassMeta],
|
|
readonly_sources: list[ClassMeta],
|
|
readonly_tests: list[ClassMeta],
|
|
) -> list[Path]:
|
|
"""Return Alpha readonly source/test paths without editable sources."""
|
|
|
|
editable = {
|
|
self._source_path(source_file)
|
|
for source_file in source_files
|
|
}
|
|
paths: list[Path] = []
|
|
for source_file in sorted(readonly_sources, key=lambda item: item.location):
|
|
path = self.context.file_manager.source_path(source_file.location)
|
|
if path in editable:
|
|
continue
|
|
paths.append(path)
|
|
for test_file in sorted(readonly_tests, key=lambda item: item.location):
|
|
path = self.context.file_manager.test_path(test_file.location)
|
|
if path in editable:
|
|
continue
|
|
paths.append(path)
|
|
# The immediately preceding root is readonly evidence unless Alpha already owns it.
|
|
if (
|
|
self.previous_cot_root_readonly_file is not None
|
|
and self.previous_cot_root_readonly_file not in editable
|
|
and self.previous_cot_root_readonly_file not in paths
|
|
):
|
|
paths.append(self.previous_cot_root_readonly_file)
|
|
return paths
|
|
|
|
def _authoritative_prompt(self) -> str:
|
|
"""Return Alpha's bounded source-edit instructions."""
|
|
|
|
return "\n".join(
|
|
[
|
|
"You are Alpha.",
|
|
"Edit source files only.",
|
|
"You may edit only listed editable source files.",
|
|
"You may not edit tests.",
|
|
"You may not edit readonly files.",
|
|
"Do not create new files during this edit. Any required new source files must be added by Alpha preflight before editing begins.",
|
|
# "Do not delete, move, or remove code from the original file until the target file is available as editable.",
|
|
"When splitting code out of an original file, use a two-phase move: populate each editable target file with the exact implementation being moved, plus required imports; then update the original file to import or re-export the moved symbols. Never remove code from the original file before its target file is editable and populated with real implementation.",
|
|
"Keep the change scoped to the current topology.",
|
|
self.context.topology_summary, # pyright: ignore[reportAttributeAccessIssue]
|
|
]
|
|
)
|
|
|
|
def _preflight_new_source_files(
|
|
self,
|
|
source_files: list[ClassMeta],
|
|
work_needed_by_location: dict[str, str],
|
|
) -> list[ClassMeta]:
|
|
"""Create and register preflight-requested source files before Aider mutates."""
|
|
|
|
requested_locations: list[str] = []
|
|
reasons_by_location: dict[str, str] = {}
|
|
seen: set[str] = set()
|
|
for source_file in source_files:
|
|
work_needed = work_needed_by_location[source_file.location]
|
|
preflight_request = AiderEditRequest(
|
|
authoritative_prompt=self._authoritative_prompt(),
|
|
editable_files=[self._source_path(source_file)],
|
|
readonly_files=[],
|
|
backlog_guidance=[work_needed],
|
|
previous_cot_records=self.previous_cot_records,
|
|
future_guidance=(
|
|
self.context.active_inbound.value.future_guidance
|
|
if self.context.active_inbound is not None
|
|
else ()
|
|
),
|
|
)
|
|
decision = AlphaSourcePreflightDecision(
|
|
client=self.context.file_manager.client,
|
|
file_manager=self.context.file_manager,
|
|
request=preflight_request,
|
|
source=source_file,
|
|
work_needed=work_needed,
|
|
logger=self.logger.getChild("source-preflight"),
|
|
)
|
|
decision()
|
|
requested = decision.additional["requested_locations"]
|
|
|
|
for location in requested:
|
|
normalized = self.context.file_manager.source_location(location)
|
|
if normalized in seen:
|
|
continue
|
|
|
|
seen.add(normalized)
|
|
requested_locations.append(normalized)
|
|
reasons_by_location[normalized] = (
|
|
f"Required by Alpha source preflight for {source_file.location}."
|
|
)
|
|
|
|
if not requested_locations:
|
|
return []
|
|
|
|
payload = {
|
|
Path(location): reasons_by_location[location]
|
|
for location in requested_locations
|
|
}
|
|
if self.new_files_callback is None:
|
|
raise RuntimeError("Alpha new-files callback seam is unavailable.")
|
|
self.new_files_callback(payload)
|
|
created_sources: list[ClassMeta] = []
|
|
for location in requested_locations:
|
|
current_source = self.context.file_manager.current_src(location)
|
|
if current_source is None:
|
|
raise ValueError(
|
|
f"Alpha preflight created {location}, but FileManager did not register current source metadata."
|
|
)
|
|
created_sources.append(current_source)
|
|
return created_sources
|
|
|
|
def _refresh_changed_sources(
|
|
self,
|
|
changed_paths: list[Path],
|
|
prompt: str,
|
|
tailing_logs: str = "",
|
|
) -> list[ClassMeta]:
|
|
"""Refresh changed source files through FileManager and return current metadata."""
|
|
|
|
handled: list[ClassMeta] = []
|
|
for path in changed_paths:
|
|
try:
|
|
location = self.context.file_manager.source_location(path)
|
|
except ValueError:
|
|
continue
|
|
|
|
handled.append(self.context.file_manager.refresh_src(location, prompt, tailing_logs))
|
|
|
|
return handled
|
|
|
|
def _blank_sources(self, source_files: list[ClassMeta]) -> list[ClassMeta]:
|
|
"""Return source files whose resolved disk content is missing or blank."""
|
|
|
|
blank_sources: list[ClassMeta] = []
|
|
for source_file in source_files:
|
|
path = self._source_path(source_file)
|
|
try:
|
|
repo_exec = self.context.file_manager.repo_exec
|
|
content = repo_exec.read_text(path) if repo_exec.exists(path) else None
|
|
except OSError:
|
|
content = None
|
|
if content is not None and content.strip():
|
|
continue
|
|
blank_sources.append(source_file)
|
|
return blank_sources
|
|
|
|
def _blank_source_correction_check(
|
|
self,
|
|
*,
|
|
editable_sources: list[ClassMeta],
|
|
readonly_sources: list[ClassMeta],
|
|
readonly_tests: list[ClassMeta],
|
|
blank_sources: list[ClassMeta],
|
|
work_needed_by_location: dict[str, str],
|
|
previous_summary: str,
|
|
request: AiderEditRequest,
|
|
) -> AlphaBlankSourceCorrectionDecision:
|
|
"""Return a correction check for blank Alpha-preflight-created source files."""
|
|
|
|
return AlphaBlankSourceCorrectionDecision(
|
|
client=self.context.file_manager.client,
|
|
file_manager=self.context.file_manager,
|
|
request=request,
|
|
work_needed_by_location=work_needed_by_location,
|
|
editable_sources=editable_sources,
|
|
readonly_sources=readonly_sources,
|
|
readonly_tests=readonly_tests,
|
|
blank_sources=blank_sources,
|
|
previous_summary=previous_summary,
|
|
)
|
|
|
|
def _run_source_edit(
|
|
self,
|
|
request: AiderEditRequest,
|
|
) -> CodeEditorResult:
|
|
"""Run one Alpha Aider edit with the supplied editable and readonly scope."""
|
|
|
|
if self.aider_worker is None:
|
|
raise RuntimeError("Alpha source worker seam is unavailable.")
|
|
|
|
self.aider_worker.new_files_callback = self.new_files_callback # pyright: ignore[reportAttributeAccessIssue]
|
|
return self.aider_worker.edit(request, logger=self.logger.getChild("aider"))
|
|
|
|
def _run_failed_source_item(
|
|
self,
|
|
item: FailedQAItem,
|
|
) -> list[ClassMeta]:
|
|
"""Run one failed source remediation item and return the handled source files."""
|
|
|
|
if item.focused_spotlight is None:
|
|
raise ValueError("FailedQAItem requires focused_spotlight.")
|
|
source_file = self.context.file_manager.current_src(item.focused_spotlight.location)
|
|
if source_file is None:
|
|
raise ValueError(f"Alpha remediation source {item.focused_spotlight.location} is missing from current scope.")
|
|
readonly_by_location: dict[str, ClassMeta] = {}
|
|
missing: list[str] = []
|
|
failure = item.failed_test
|
|
# Preserve topology and current failure diagnosis as readonly support for this source-only repair.
|
|
for related in [
|
|
*item.upstream_classes,
|
|
*item.downstream_classes,
|
|
*item.failure_context_sources,
|
|
*admitted_sources.current_sources(self.context.file_manager, item),
|
|
]:
|
|
# The editable source and separately retained originating test already have owners.
|
|
if related.location in {source_file.location, failure.test.location if failure is not None else ""}:
|
|
continue
|
|
current_related = self.context.file_manager.current_metadata(
|
|
related.location, content_type=related.content_type,
|
|
)
|
|
if current_related is None:
|
|
missing.append(related.location)
|
|
continue
|
|
readonly_by_location.setdefault(current_related.location, current_related)
|
|
if missing:
|
|
raise ValueError(f"Alpha remediation readonly sources are missing from current scope: {', '.join(missing)}")
|
|
# Keep every supporting test readonly while preserving one originating execution target.
|
|
if item.failed_test is not None:
|
|
readonly_by_location[item.failed_test.test.location] = item.failed_test.test
|
|
readonly_tests = list(filter(lambda related: related.content_type is ContentType.Test, readonly_by_location.values()))
|
|
readonly_sources = list(filter(lambda related: related.content_type is ContentType.Source, readonly_by_location.values()))
|
|
remediation_request = AiderEditRequest(
|
|
editor_records=list(item.editor_records),
|
|
authoritative_prompt=self._authoritative_prompt(),
|
|
editable_files=[self._source_path(source_file)],
|
|
readonly_files=self._readonly_paths(
|
|
[source_file],
|
|
readonly_sources,
|
|
readonly_tests,
|
|
),
|
|
is_remediation=True,
|
|
remediation_originator_test=failure.test if failure is not None else None,
|
|
failure_test_location=failure.test.location if failure is not None else "",
|
|
failure_test_name=failure.test_name if failure is not None else "",
|
|
failure_outcome=failure.outcome if failure is not None else None,
|
|
failure_run_info=failure.run_info if failure is not None else "",
|
|
failure_error_message=failure.error_message if failure is not None else "",
|
|
failure_stack_trace=list(failure.stack_trace) if failure is not None else [],
|
|
failure_raw_output=failure.raw_output if failure is not None else "",
|
|
previous_cot_records=self.previous_cot_records,
|
|
future_guidance=item.future_guidance,
|
|
)
|
|
work_needed_by_location = {
|
|
source_file.location: self.context.backlog.remediation_work_needed_for(
|
|
source_file,
|
|
[item],
|
|
request=remediation_request,
|
|
)
|
|
}
|
|
handled_sources, self._source_summary = self._run_source_work(
|
|
[source_file],
|
|
work_needed_by_location,
|
|
readonly_sources=readonly_sources,
|
|
readonly_tests=readonly_tests,
|
|
failing_test=item.failed_test,
|
|
is_remediation=True,
|
|
)
|
|
self._handled_sources.extend(handled_sources)
|
|
return handled_sources
|
|
|
|
def _run_inspection_source_item(
|
|
self,
|
|
item: InspectionBacklog,
|
|
) -> list[ClassMeta]:
|
|
"""Run one source inspection item and return the handled source files."""
|
|
|
|
if item.focused_spotlight is None:
|
|
raise ValueError("InspectionBacklog requires focused_spotlight.")
|
|
source_file = self.context.file_manager.current_src(item.focused_spotlight.location)
|
|
if source_file is None:
|
|
raise ValueError(f"Alpha inspection source {item.focused_spotlight.location} is missing from current scope.")
|
|
readonly_sources: list[ClassMeta] = []
|
|
missing: list[str] = []
|
|
for related in [*item.upstream_classes, *item.downstream_classes]:
|
|
if related.location == source_file.location:
|
|
continue
|
|
current_related = self.context.file_manager.current_src(related.location)
|
|
if current_related is None:
|
|
missing.append(related.location)
|
|
continue
|
|
readonly_sources.append(current_related)
|
|
if missing:
|
|
raise ValueError(f"Alpha inspection readonly sources are missing from current scope: {', '.join(missing)}")
|
|
inspection_request = AiderEditRequest(
|
|
editor_records=list(item.editor_records),
|
|
authoritative_prompt="",
|
|
editable_files=[self._source_path(source_file)],
|
|
readonly_files=self._readonly_paths(
|
|
[source_file],
|
|
readonly_sources,
|
|
[],
|
|
),
|
|
previous_cot_records=self.previous_cot_records,
|
|
future_guidance=item.future_guidance,
|
|
)
|
|
work_needed = self.context.backlog.inspection_work_needed_for(
|
|
source_file,
|
|
self.context.file_manager.prompt,
|
|
[item],
|
|
request=inspection_request,
|
|
readonly_sources=readonly_sources,
|
|
)
|
|
# A neutral inspection result closes this frontier item before it can manufacture an edit request.
|
|
if not work_needed.strip():
|
|
self._source_summary = "Inspection established no source work for this file."
|
|
return []
|
|
handled_sources, self._source_summary = self._run_source_work(
|
|
[source_file],
|
|
{source_file.location: work_needed},
|
|
readonly_sources,
|
|
[],
|
|
failing_test=None,
|
|
is_remediation=False,
|
|
)
|
|
self._handled_sources.extend(handled_sources)
|
|
return handled_sources
|
|
|
|
def _stage_postflight_alpha_results(
|
|
self,
|
|
item: FailedQAItem | InspectionBacklog,
|
|
handled_sources: list[ClassMeta],
|
|
*,
|
|
verify_unchanged: bool = False,
|
|
) -> None:
|
|
"""Stage postflight work for changed sources or the recovered no-op's current source."""
|
|
|
|
postflight_sources = handled_sources
|
|
# An admitted remediation LEAVE already resolved the obligation, including on resume.
|
|
if (
|
|
verify_unchanged
|
|
and isinstance(item, FailedQAItem)
|
|
and is_code_edit_leave_summary(self._source_summary)
|
|
):
|
|
verify_unchanged = False
|
|
# A recovered no-op must verify its current source without claiming another edit.
|
|
if verify_unchanged:
|
|
assert item.focused_spotlight is not None
|
|
postflight_sources = [item.focused_spotlight]
|
|
self.logger.info(
|
|
"Alpha verifying resumed no-op for %s before closing source work.",
|
|
item.focused_spotlight.location,
|
|
)
|
|
# Ordinary no-change work may close without producing a testing assignment.
|
|
if not postflight_sources:
|
|
if self.context.active_inbound is not None and not self.context.outbound_queue:
|
|
self.context.clear_moved_inbound()
|
|
self.context.persist()
|
|
return
|
|
created = self.context.backlog.create_outbound_for(
|
|
*postflight_sources,
|
|
upstream_classes=list(item.upstream_classes),
|
|
downstream_classes=list(item.downstream_classes),
|
|
queue=False,
|
|
original_cot=item.original_cot if isinstance(item, FailedQAItem) else self.context.file_manager.prompt,
|
|
future_guidance=item.future_guidance,
|
|
verify_unchanged=verify_unchanged,
|
|
editor_records=list(item.editor_records),
|
|
logger=self.logger.getChild("backlog"),
|
|
)
|
|
# Stage each routed observation through the transaction lane selected by ownership.
|
|
for outbound_item in created:
|
|
# Send source-owned work to remediation and all admitted test work to testing.
|
|
if isinstance(outbound_item, FailedQAItem):
|
|
self.context.stage_remediation_outbound(outbound_item)
|
|
else:
|
|
self.context.stage_testing_outbound(outbound_item)
|
|
self.context.persist()
|
|
callback = getattr(self.context, "alpha_postflight_callback", None)
|
|
# Accepted-change observers receive only sources actually changed by this attempt.
|
|
if handled_sources and callable(callback):
|
|
callback(tuple(handled_sources))
|
|
|
|
def _process_inbound_item(
|
|
self,
|
|
item: FailedQAItem | InspectionBacklog,
|
|
*,
|
|
resume: bool = False,
|
|
) -> None:
|
|
"""Run one Alpha inbound item and stage any resulting testing work."""
|
|
|
|
self.validation_receipts = ValidationReceipts()
|
|
handled_sources = (
|
|
self._run_failed_source_item(item)
|
|
if isinstance(item, FailedQAItem)
|
|
else self._run_inspection_source_item(item)
|
|
)
|
|
self._stage_postflight_alpha_results(
|
|
item,
|
|
handled_sources,
|
|
verify_unchanged=resume and not handled_sources,
|
|
)
|
|
|
|
def _source_edit_request(
|
|
self,
|
|
*,
|
|
editable_sources: list[ClassMeta],
|
|
readonly_sources: list[ClassMeta],
|
|
readonly_tests: list[ClassMeta],
|
|
work_needed_by_location: dict[str, str],
|
|
failing_test: FailingTest | None,
|
|
is_remediation: bool,
|
|
extra_guidance: list[str] | None,
|
|
) -> AiderEditRequest:
|
|
"""Build the exact Aider source-edit request for one Alpha attempt."""
|
|
|
|
backlog_guidance = [
|
|
f"{location}: {work_needed}"
|
|
for location, work_needed in work_needed_by_location.items()
|
|
]
|
|
if extra_guidance:
|
|
backlog_guidance.extend(extra_guidance)
|
|
return AiderEditRequest(
|
|
authoritative_prompt=self._authoritative_prompt(),
|
|
editable_files=[
|
|
self._source_path(source_file)
|
|
for source_file in editable_sources
|
|
],
|
|
readonly_files=self._readonly_paths(
|
|
editable_sources,
|
|
readonly_sources,
|
|
readonly_tests,
|
|
),
|
|
pinned_context=self.context.pinned_context,
|
|
backlog_guidance=backlog_guidance,
|
|
editor_records=(list(self.context.active_inbound.value.editor_records)
|
|
if self.context.active_inbound is not None else []),
|
|
previous_cot_records=self.previous_cot_records,
|
|
future_guidance=(
|
|
self.context.active_inbound.value.future_guidance
|
|
if self.context.active_inbound is not None
|
|
else ()
|
|
),
|
|
failure_error_message=(
|
|
failing_test.error_message
|
|
if failing_test is not None
|
|
else ""
|
|
),
|
|
failure_stack_trace=(
|
|
list(failing_test.stack_trace)
|
|
if failing_test is not None
|
|
else []
|
|
),
|
|
failure_raw_output=(
|
|
failing_test.raw_output
|
|
if failing_test is not None
|
|
else ""
|
|
),
|
|
is_remediation=is_remediation,
|
|
failure_test_location=failing_test.test.location if failing_test is not None else "",
|
|
failure_test_name=failing_test.test_name if failing_test is not None else "",
|
|
failure_outcome=failing_test.outcome if failing_test is not None else None,
|
|
failure_run_info=failing_test.run_info if failing_test is not None else "",
|
|
remediation_originator_test=(
|
|
failing_test.test if is_remediation and failing_test is not None else None
|
|
),
|
|
context=AlphaLoop,
|
|
frontier_provider=self.frontier_provider,
|
|
)
|
|
|
|
def build_correction_guide(
|
|
self,
|
|
failed_checks: tuple[ReasonableVibeCheck, ...],
|
|
rejected_diff: str,
|
|
) -> str:
|
|
"""Pair the rejected Alpha patch with acceptance reasons for a restored retry."""
|
|
|
|
section = self.context.file_manager.prompt_formatter_builder.generate
|
|
failure_lines = [
|
|
"The previous source edit was rejected and rolled back. The editable source",
|
|
"files have been restored to their exact pre-attempt state.",
|
|
"",
|
|
"Preflight-created source files remain present. Any such blank source must be",
|
|
"populated with real implementation during this attempt.",
|
|
"",
|
|
"This patch was rejected by semantic acceptance. It is no longer applied:",
|
|
section("REJECTED PATCH", depth=2, escape=True)(
|
|
rejected_diff or "(No source changes in this attempt.)"
|
|
),
|
|
"",
|
|
"Semantic acceptance rejected this patch because:",
|
|
]
|
|
# Keep each rejection reason beside its corresponding repair instruction.
|
|
for index, failed_check in enumerate(failed_checks, start=1):
|
|
failure_lines.extend(
|
|
[
|
|
"",
|
|
section(f"FAILURE {index}", depth=2, escape=True)(failed_check.reason),
|
|
"",
|
|
section(f"REQUIRED CORRECTION {index}", depth=2, escape=True)(
|
|
str(failed_check.additional.get("instruction", "")).strip()
|
|
),
|
|
]
|
|
)
|
|
failure_lines.append(
|
|
section("RETRY REFINEMENT STEPS", depth=2)("\n".join((
|
|
"1. Re-check the established concrete divergence against the restored current source and the rejected patch; prior inspection or retry guidance is provisional.",
|
|
"2. Use the rejected patch to reconstruct the intended repair, address the rejection, and apply the corrected change to the restored source. SEARCH text must match the restored source, not the rejected patch.",
|
|
"3. If the premise is disproven and no COT or failure-grounded obligation remains, leave the source unchanged rather than inventing work.",
|
|
"4. Verify the focused obligation without changing tests, unrelated compiler debt, or files outside the current editable scope.",
|
|
"Do not delete preflight-created source files.",
|
|
)))
|
|
)
|
|
return section("ALPHA EDIT ACCEPTANCE FAILED")("\n".join(failure_lines))
|
|
|
|
def _run_source_work(
|
|
self,
|
|
source_files: list[ClassMeta],
|
|
work_needed_by_location: dict[str, str],
|
|
readonly_sources: list[ClassMeta],
|
|
readonly_tests: list[ClassMeta],
|
|
failing_test: FailingTest | None = None,
|
|
*,
|
|
is_remediation: bool,
|
|
) -> tuple[list[ClassMeta], str]:
|
|
"""Run the source-edit seam with backlog guidance and return accepted changed source files."""
|
|
|
|
if not source_files:
|
|
return [], "Alpha had no source files assigned."
|
|
|
|
if self.aider_worker is None:
|
|
raise RuntimeError("Alpha source worker seam is unavailable.")
|
|
|
|
historical = admitted_sources.current_sources(
|
|
self.context.file_manager,
|
|
self.context.active_inbound.value if self.context.active_inbound is not None else None,
|
|
)
|
|
editable_locations = set(map(lambda source: source.location, source_files))
|
|
# Historical files supply acceptance and editor context without granting edit authority.
|
|
readonly_sources = list(dict(map(lambda source: (source.location, source), filter(
|
|
lambda source: source.content_type is ContentType.Source and source.location not in editable_locations,
|
|
[*readonly_sources, *historical],
|
|
))).values())
|
|
readonly_tests = list(dict(map(lambda source: (source.location, source), filter(
|
|
lambda source: source.content_type is ContentType.Test,
|
|
[*readonly_tests, *historical],
|
|
))).values())
|
|
|
|
restored_diffs: list[str] = []
|
|
# Inbound records describe earlier committed work; this attempt publishes records only after commit.
|
|
# Preserve leftover attempt evidence before clearing source snapshots.
|
|
for source_file in source_files:
|
|
memento = source_file.set_or_get_memento(self.context.file_manager)
|
|
# clear() restores layer zero, including edits retained by nested snapshots.
|
|
restored_diffs.append(Memento(memento.file_path, memento.repo_exec).diff)
|
|
memento.clear()
|
|
|
|
# Only ordinary inspection work may expand its source assignment before editing.
|
|
preflight_sources = self._preflight_new_source_files(
|
|
source_files,
|
|
work_needed_by_location,
|
|
) if not is_remediation else []
|
|
editable_sources = [*source_files, *preflight_sources]
|
|
# Preflight may have introduced another candidate before the coordinated attempt.
|
|
for source_file in editable_sources:
|
|
memento = source_file.set_or_get_memento(self.context.file_manager)
|
|
# Preserve the same complete stack boundary that clear() will restore.
|
|
restored_diffs.append(Memento(memento.file_path, memento.repo_exec).diff)
|
|
memento.clear()
|
|
|
|
restored_diff = "\n".join(filter(None, restored_diffs))
|
|
extra_guidance = [RepairEvidenceBuilder.compiler_rejection(
|
|
"Alpha restored earlier source snapshots before this attempt.", restored_diff,
|
|
)] if restored_diff else []
|
|
while True:
|
|
self.validation_receipts = ValidationReceipts()
|
|
mementos_by_location = {}
|
|
for source_file in editable_sources:
|
|
mementos_by_location[source_file.location] = source_file.set_or_get_memento(
|
|
self.context.file_manager,
|
|
)
|
|
|
|
request = self._source_edit_request(
|
|
editable_sources=editable_sources,
|
|
readonly_sources=readonly_sources,
|
|
readonly_tests=readonly_tests,
|
|
work_needed_by_location=work_needed_by_location,
|
|
failing_test=failing_test,
|
|
is_remediation=is_remediation,
|
|
extra_guidance=extra_guidance,
|
|
)
|
|
try:
|
|
result = self._run_source_edit(request)
|
|
changed_paths, summary = result.changed_paths, result.summary
|
|
request.editor_records = list(result.records)
|
|
self.validation_receipts = request.validation_receipts
|
|
leave_it_judged = is_code_edit_leave_summary(summary)
|
|
normalized_summary = consume_code_edit_leave_summary(summary)
|
|
summary_parts = [normalized_summary] if normalized_summary else []
|
|
|
|
if leave_it_judged:
|
|
passed = True
|
|
failed_checks: tuple[ReasonableVibeCheck, ...] = ()
|
|
else:
|
|
blank_sources = self._blank_sources(editable_sources)
|
|
blank_check = self._blank_source_correction_check(
|
|
editable_sources=editable_sources,
|
|
readonly_sources=readonly_sources,
|
|
readonly_tests=readonly_tests,
|
|
blank_sources=blank_sources,
|
|
work_needed_by_location=work_needed_by_location,
|
|
previous_summary=summary,
|
|
request=request,
|
|
)
|
|
check = AlphaEditAcceptanceDecision(
|
|
client=self.context.file_manager.client,
|
|
file_manager=self.context.file_manager,
|
|
request=request,
|
|
editable_sources=editable_sources,
|
|
readonly_sources=readonly_sources,
|
|
readonly_tests=readonly_tests,
|
|
previous_summary=summary,
|
|
)
|
|
blank_passed = blank_check()
|
|
acceptance_passed = check()
|
|
self.validation_receipts = request.validation_receipts
|
|
check_results = ((blank_check, blank_passed), (check, acceptance_passed))
|
|
failed_checks = tuple(
|
|
map(
|
|
lambda check_result: check_result[0],
|
|
filterfalse(lambda check_result: check_result[1], check_results),
|
|
)
|
|
)
|
|
passed = not any(map(lambda check_result: not check_result[1], check_results))
|
|
|
|
except DeadEndError as error:
|
|
self.validation_receipts = ValidationReceipts()
|
|
rejected_diff = "\n".join(filter(None, map(lambda memento: memento.diff, mementos_by_location.values())))
|
|
# Carry the abandoned Alpha candidate into the historical revival's future guidance.
|
|
if rejected_diff:
|
|
error.message += "\n\n" + RepairEvidenceBuilder.compiler_rejection(str(error), rejected_diff)
|
|
for memento in mementos_by_location.values():
|
|
memento.rollback()
|
|
request.editor_records = []
|
|
raise
|
|
|
|
if passed:
|
|
changed_paths = [memento.file_path for memento in mementos_by_location.values() if memento.diff]
|
|
handled = self._refresh_changed_sources(
|
|
changed_paths,
|
|
self._authoritative_prompt(),
|
|
"\n".join(
|
|
part
|
|
for part in summary_parts
|
|
if part
|
|
),
|
|
)
|
|
for memento in mementos_by_location.values():
|
|
memento.commit()
|
|
# Postflight decisions still belong to Alpha and may read its retained edit history.
|
|
if self.context.active_inbound is not None:
|
|
self.context.active_inbound.value.editor_records = list(request.editor_records)
|
|
# Preserve the admitted LEAVE for Alpha's resumed postflight decision.
|
|
return handled, CodeEditLeaveSummary(normalized_summary) if leave_it_judged else "\n".join(
|
|
part
|
|
for part in summary_parts
|
|
if part
|
|
)
|
|
|
|
# Preserve the entire Alpha attempt before rollback discards the editor's changes.
|
|
rejected_diff = "\n".join(filter(None, map(lambda memento: memento.diff, mementos_by_location.values())))
|
|
extra_guidance = [self.build_correction_guide(failed_checks, rejected_diff)]
|
|
for memento in mementos_by_location.values():
|
|
memento.rollback()
|
|
request.editor_records = []
|
|
|
|
def _prepare_work(self) -> ProtoLoop | None:
|
|
"""Return no child loop because Alpha drains source backlog directly."""
|
|
|
|
return None
|
|
|
|
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: FailedQAItem | InspectionBacklog) -> None:
|
|
"""Publish the concrete source item claimed by Alpha."""
|
|
|
|
spotlight = item.focused_spotlight
|
|
location = spotlight.location if spotlight is not None else None
|
|
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__,
|
|
next_loop="BetaLoop",
|
|
current_item=location,
|
|
work_focus=focus,
|
|
)
|
|
|
|
def _restart_item(self, error: DeadEndError) -> None:
|
|
"""Revive the selected phase start after abandoning Alpha'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.ALPHA:
|
|
if 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.")
|
|
elif isinstance(revived, TestingBacklog):
|
|
self.context.backlog.queue_testing(revived, prepend=True, logger=self.logger)
|
|
else:
|
|
raise TypeError("Beta restart requires testing backlog work.")
|
|
self.logger.info("Alpha %s revived %s phase work.", error.direction.value, phase.value)
|
|
|
|
def _drain_source_backlog(self) -> None:
|
|
"""Run source work once per claimed backlog item and create downstream testing backlog."""
|
|
|
|
if not isinstance(self.context.backlog, ContributorBacklog):
|
|
return
|
|
|
|
self.context.recover_active_inbound(logger=self.logger.getChild("backlog"))
|
|
# Like Supreme's resume branch, only the retained active item owns this local flag.
|
|
resume_inbound = self.context.active_inbound is not None
|
|
self.context.persist()
|
|
self._publish_queue_depth()
|
|
while True:
|
|
if self.context.inbound_queue and self.context.active_inbound is not None:
|
|
self._publish_work_status(self.context.active_inbound.value)
|
|
try:
|
|
with provenance.item_scope(Phase.ALPHA, self.context.active_inbound.value):
|
|
self._process_inbound_item(
|
|
self.context.active_inbound.value,
|
|
resume=resume_inbound,
|
|
)
|
|
except DeadEndError as error:
|
|
self._restart_item(error)
|
|
resume_inbound = False
|
|
self.context.persist()
|
|
self._publish_queue_depth()
|
|
continue
|
|
self.context.move_outbound()
|
|
self.context.persist()
|
|
resume_inbound = False
|
|
self._publish_queue_depth()
|
|
continue
|
|
claimed = self.context.move_inbound_from_backlog(logger=self.logger.getChild("backlog"))
|
|
self.context.persist()
|
|
self._publish_queue_depth()
|
|
if claimed is None:
|
|
break
|
|
try:
|
|
with claimed as scope:
|
|
self._publish_work_status(scope.value)
|
|
with provenance.item_scope(Phase.ALPHA, scope.value):
|
|
self._process_inbound_item(scope.value)
|
|
scope.commit()
|
|
except DeadEndError as error:
|
|
self._restart_item(error)
|
|
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()
|
|
|
|
def run(self, logger_name: str | None = None) -> str:
|
|
"""Drain source backlog work directly and return the alpha 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._drain_source_backlog()
|
|
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)
|
|
|
|
def _post_run(self) -> str:
|
|
"""Return a concise Alpha summary after source work and testing backlog creation."""
|
|
|
|
if not self._handled_sources:
|
|
return self._source_summary or "Alpha completed no source work."
|
|
|
|
summary = f"Alpha completed source work for {len(self._handled_sources)} file(s)."
|
|
if self.context.marginal_result:
|
|
summary = f"{summary} {' '.join(self.context.marginal_result)}"
|
|
|
|
return summary
|