124 lines
3.9 KiB
Python
124 lines
3.9 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
"""Immutable reporting and reconciliation values used by StabilizationLoop."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from enum import Enum
|
|
from typing import Mapping
|
|
|
|
from band_aider.context.class_meta import ClassMeta
|
|
|
|
|
|
class RemediationClassification(str, Enum):
|
|
"""Name reporting-only relationships between remediation and COT topology."""
|
|
|
|
NORMAL = "normal"
|
|
DEFERRED = "deferred"
|
|
WALK_BACK = "walk_back"
|
|
LATENT_HIDDEN_BUG = "latent_hidden_bug"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RemediationClassificationResult:
|
|
"""Carry deterministic remediation reporting evidence without scheduler authority."""
|
|
|
|
classification: RemediationClassification
|
|
focus_location: str
|
|
admission_locations: tuple[str, ...]
|
|
previous_cot_index: int | None = None
|
|
previous_topology_index: int | None = None
|
|
relation: str = "neither"
|
|
reason: str = ""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DeferredTopologyRequirementLookback:
|
|
"""Carry bounded upstream perimeter evidence for one unmet direct requirement."""
|
|
|
|
perimeter_locations: tuple[str, ...]
|
|
alternate_live_path: bool
|
|
virtual_visited: bool = False
|
|
|
|
@property
|
|
def explained(self) -> bool:
|
|
"""Return whether the requirement has a complete unchanged-perimeter closure."""
|
|
|
|
return (
|
|
bool(self.perimeter_locations) or self.virtual_visited
|
|
) and not self.alternate_live_path
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TopologicalReadiness:
|
|
"""Carry deterministic current-COT admission evidence without frontier authority.
|
|
|
|
Example: ``readiness.closure_ready`` reports the proof result while the
|
|
topology owner retains exclusive authority to admit the candidate.
|
|
"""
|
|
|
|
candidate_location: str
|
|
direct_in_beam_requirements: tuple[str, ...]
|
|
visited_requirements: tuple[str, ...]
|
|
unmet_requirements: tuple[str, ...]
|
|
unchanged_perimeter_by_requirement: tuple[tuple[str, tuple[str, ...]], ...]
|
|
alternate_live_path_by_requirement: tuple[tuple[str, bool], ...]
|
|
closure_ready: bool
|
|
rejection_reason: str = ""
|
|
virtual_visited_requirements: tuple[str, ...] = ()
|
|
|
|
def diagnostic_payload(
|
|
self,
|
|
*,
|
|
decision: str,
|
|
rejection_reason: str,
|
|
) -> dict[str, object]:
|
|
"""Return this readiness proof in the terminal diagnostic schema."""
|
|
|
|
perimeter_by_requirement = dict(map(
|
|
lambda item: (item[0], list(item[1])),
|
|
self.unchanged_perimeter_by_requirement,
|
|
))
|
|
return {
|
|
"decision": decision,
|
|
"direct_in_beam_requirements": list(
|
|
self.direct_in_beam_requirements
|
|
),
|
|
"visited_requirements": list(self.visited_requirements),
|
|
"unmet_requirements": list(self.unmet_requirements),
|
|
"virtual_visited_requirements": list(
|
|
self.virtual_visited_requirements
|
|
),
|
|
"unchanged_perimeter_by_requirement": perimeter_by_requirement,
|
|
"skip_witness_by_requirement": perimeter_by_requirement,
|
|
"alternate_live_path_by_requirement": dict(
|
|
self.alternate_live_path_by_requirement
|
|
),
|
|
"rejection_reason": rejection_reason,
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class FrontierBoundary:
|
|
"""Carry the active frontier's unchanged and live-seed boundary.
|
|
|
|
Example: ``FrontierBoundary(..., live_seed_locations={"a.py"})``
|
|
supplies the ordinary BFS seeds for one active frontier.
|
|
"""
|
|
|
|
current_locations: frozenset[str]
|
|
unchanged_locations: frozenset[str]
|
|
live_seed_locations: frozenset[str]
|
|
source_by_location: Mapping[str, ClassMeta]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TopologicalCandidateDecision:
|
|
"""Carry one candidate's deterministic ``ready`` or ``deferred`` admission result."""
|
|
|
|
decision: str
|
|
rejection_reason: str
|
|
readiness: TopologicalReadiness | None = None
|