228 lines
10 KiB
Python
228 lines
10 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
"""Shared judgement construction for Alpha postflight and Beta validation flows."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any, TYPE_CHECKING, Sequence, cast
|
|
|
|
from band_aider.context.beta_workflow_story import (
|
|
TargetedFailureStory,
|
|
)
|
|
from band_aider.context.targeted_failure_owner_check import TargetedAlphaOrBetaRoutingDecision
|
|
from band_aider.context.targeted_test_collection_failure_owner_check import (
|
|
TargetedCollectionAlphaOrBetaRoutingDecision,
|
|
)
|
|
from band_aider.judgement.beta_contract_guidance_check import BetaContractGuidanceDecision
|
|
from band_aider.judgement.edit_request import AiderEditRequest
|
|
from band_aider.judgement.code_editor import EditorRecord
|
|
|
|
if TYPE_CHECKING:
|
|
from band_aider.context.class_meta import ClassMeta
|
|
from band_aider.context.cot_task_generator import CotTaskRecord
|
|
from band_aider.context.failing_test import FailingTest
|
|
from band_aider.context.working_backlog import TestingBacklog
|
|
from band_aider.judgement.file_manager import FileManager
|
|
from band_aider.judgement.vibe_check import ReasonableVibeCheck
|
|
|
|
|
|
class BetaWorkflowJudgementService:
|
|
"""BetaWorkflowJudgementService owns shared routing and contract-guidance judgement construction."""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
targeted_routing_cls: type[TargetedAlphaOrBetaRoutingDecision] = TargetedAlphaOrBetaRoutingDecision,
|
|
collection_routing_cls: type[TargetedCollectionAlphaOrBetaRoutingDecision] = TargetedCollectionAlphaOrBetaRoutingDecision,
|
|
guidance_cls: type[BetaContractGuidanceDecision] = BetaContractGuidanceDecision,
|
|
) -> None:
|
|
"""Initialize the shared judgement service with injectable judgement implementations."""
|
|
|
|
self.targeted_routing_cls = targeted_routing_cls
|
|
self.collection_routing_cls = collection_routing_cls
|
|
self.guidance_cls = guidance_cls
|
|
self.targeted_routing_calls: list[ReasonableVibeCheck] = []
|
|
self.collection_routing_calls: list[ReasonableVibeCheck] = []
|
|
self.last_targeted_evidence: dict[str, Any] = {}
|
|
self.last_collection_evidence: dict[str, Any] = {}
|
|
self.logger_name = "band_aider.context.beta_workflow_judgement"
|
|
|
|
def refine_beta_instruction(
|
|
self,
|
|
*,
|
|
file_manager: FileManager,
|
|
focused_test: ClassMeta,
|
|
original_cot: str,
|
|
accepted_source_diff: str,
|
|
assignment_type: str,
|
|
test_status: str,
|
|
previous_beta_guidance: str = "",
|
|
failure_error_message: str = "",
|
|
failure_stack_trace: Sequence[str] = (),
|
|
failure_run_info: str = "",
|
|
readonly_sources: Sequence[ClassMeta] = (),
|
|
previous_cot_records: tuple[CotTaskRecord, ...] = (),
|
|
editor_records: Sequence[EditorRecord] = (),
|
|
logger: logging.Logger | None = None,
|
|
) -> str:
|
|
"""Return the sole refined Beta instruction for an already-selected ticket."""
|
|
|
|
current_readonly_sources = list(
|
|
{
|
|
source.location: source
|
|
for source in readonly_sources
|
|
if source.location != focused_test.location
|
|
}.values()
|
|
)
|
|
request = AiderEditRequest(
|
|
authoritative_prompt="",
|
|
editable_files=[file_manager.test_path(focused_test.location)],
|
|
readonly_files=[
|
|
source.resolve_path(file_manager)
|
|
for source in current_readonly_sources
|
|
],
|
|
previous_cot_records=previous_cot_records,
|
|
editor_records=list(editor_records),
|
|
failure_error_message=failure_error_message,
|
|
failure_stack_trace=list(failure_stack_trace),
|
|
failure_run_info=failure_run_info,
|
|
)
|
|
try:
|
|
check = self.guidance_cls(
|
|
client=file_manager.client,
|
|
file_manager=file_manager,
|
|
request=request,
|
|
focused_test=focused_test,
|
|
original_cot=original_cot,
|
|
accepted_source_diff=accepted_source_diff,
|
|
assignment_type=assignment_type,
|
|
test_status=test_status,
|
|
previous_beta_guidance=previous_beta_guidance,
|
|
readonly_sources=current_readonly_sources,
|
|
logger=logger,
|
|
)
|
|
if not check():
|
|
raise ValueError("Beta prompt refinement returned no valid instruction.")
|
|
return check.beta_instruction
|
|
except Exception as error:
|
|
raise ValueError("Beta prompt refinement failed.") from error
|
|
|
|
def route_targeted_failure(
|
|
self,
|
|
*,
|
|
file_manager: FileManager,
|
|
story: TargetedFailureStory,
|
|
previous_cot_records: tuple[CotTaskRecord, ...] = (),
|
|
logger: logging.Logger | None = None,
|
|
) -> TargetedAlphaOrBetaRoutingDecision:
|
|
"""Run the ordinary targeted Alpha-or-Beta routing check with shared normalized evidence."""
|
|
|
|
evidence = {
|
|
"original_cot": story.original_cot,
|
|
"accepted_alpha_summary": story.accepted_alpha_summary,
|
|
"accepted_source_diff": "\n".join(
|
|
source.get_diff(file_manager)
|
|
for source in story.candidate_sources
|
|
if source.get_diff(file_manager).strip()
|
|
),
|
|
"structural_evidence": dict(story.structural_evidence or {}),
|
|
"semantic_evidence": dict(story.semantic_evidence or {}),
|
|
"failure_phase": story.failure_phase,
|
|
"previous_beta_guidance": story.previous_beta_guidance,
|
|
"beta_edit_diff": story.beta_edit_diff,
|
|
"beta_editor_summary": story.beta_editor_summary,
|
|
}
|
|
source_locations = [source.location for source in story.candidate_sources]
|
|
test_location = story.focused_test.location
|
|
evidence["candidate_source_locations"] = source_locations
|
|
evidence["candidate_test_location"] = test_location
|
|
evidence["failure_error_message"] = story.failing.error_message
|
|
evidence["failure_stack_trace"] = list(story.failing.stack_trace)
|
|
evidence["raw_output_present"] = bool(story.raw_output.strip())
|
|
evidence["run_info"] = story.failing.run_info
|
|
self.last_targeted_evidence = evidence
|
|
request = AiderEditRequest(
|
|
authoritative_prompt="",
|
|
editable_files=[file_manager.test_path(story.focused_test.location)],
|
|
readonly_files=[
|
|
source.resolve_path(file_manager)
|
|
for source in story.candidate_sources
|
|
if source.location != test_location
|
|
],
|
|
previous_cot_records=previous_cot_records,
|
|
future_guidance=story.future_guidance,
|
|
editor_records=list(story.editor_records),
|
|
)
|
|
check = self.targeted_routing_cls(
|
|
client=file_manager.client,
|
|
file_manager=file_manager,
|
|
request=request,
|
|
story=story,
|
|
)
|
|
check()
|
|
self.targeted_routing_calls.append(check)
|
|
active_logger = logger or logging.getLogger(self.logger_name)
|
|
log_location = test_location or "[missing test]"
|
|
active_logger.info("Shared targeted routing chose %s: %s", check.outcome, check.reason)
|
|
active_logger.debug("Shared targeted routing test=%s sources=%s.", log_location, ",".join(source_locations))
|
|
return check
|
|
|
|
def route_collection_failure(
|
|
self,
|
|
*,
|
|
file_manager: FileManager,
|
|
story: TargetedFailureStory,
|
|
previous_cot_records: tuple[CotTaskRecord, ...] = (),
|
|
logger: logging.Logger | None = None,
|
|
) -> TargetedCollectionAlphaOrBetaRoutingDecision:
|
|
"""Run the collection/setup Alpha-or-Beta routing check with shared normalized evidence."""
|
|
|
|
focused_test = cast("ClassMeta", story.focused_test)
|
|
evidence = {
|
|
"original_cot": story.original_cot,
|
|
"accepted_alpha_summary": story.accepted_alpha_summary,
|
|
"work_needed": story.work_needed,
|
|
"structural_evidence": dict(story.structural_evidence or {}),
|
|
"semantic_evidence": dict(story.semantic_evidence or {}),
|
|
"failure_phase": story.failure_phase,
|
|
"previous_beta_guidance": story.previous_beta_guidance,
|
|
"beta_edit_diff": story.beta_edit_diff,
|
|
"beta_editor_summary": story.beta_editor_summary,
|
|
}
|
|
source_locations = [source.location for source in story.candidate_sources]
|
|
test_location = focused_test.location
|
|
evidence["candidate_source_locations"] = source_locations
|
|
evidence["candidate_test_location"] = test_location
|
|
evidence["failure_error_message"] = story.failing.error_message
|
|
evidence["failure_stack_trace"] = list(story.failing.stack_trace)
|
|
evidence["raw_output_present"] = bool(story.raw_output.strip())
|
|
evidence["run_info"] = story.failing.run_info
|
|
self.last_collection_evidence = evidence
|
|
request = AiderEditRequest(
|
|
authoritative_prompt="",
|
|
editable_files=[file_manager.test_path(focused_test.location)],
|
|
readonly_files=[
|
|
source.resolve_path(file_manager)
|
|
for source in story.candidate_sources
|
|
if source.location != focused_test.location
|
|
],
|
|
previous_cot_records=previous_cot_records,
|
|
future_guidance=story.future_guidance,
|
|
editor_records=list(story.editor_records),
|
|
)
|
|
check = self.collection_routing_cls(
|
|
client=file_manager.client,
|
|
file_manager=file_manager,
|
|
request=request,
|
|
story=story,
|
|
)
|
|
check()
|
|
self.collection_routing_calls.append(check)
|
|
active_logger = logger or logging.getLogger(self.logger_name)
|
|
log_location = test_location or "[missing test]"
|
|
active_logger.info("Shared collection routing chose %s: %s", check.outcome, check.reason)
|
|
active_logger.debug("Shared collection routing test=%s sources=%s.", log_location, ",".join(source_locations))
|
|
return check
|