194 lines
7.5 KiB
Python
194 lines
7.5 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Dict, List, Tuple
|
|
|
|
from band_aider.context.class_meta import ClassMeta
|
|
from band_aider.context.contex_bag import ContextBag
|
|
from band_aider.context.topological_liveliness import TopologicalLiveliness
|
|
from band_aider.context.work_completion import WorkCompletion
|
|
from band_aider.judgement.code_dependency_graph_maintainer import (
|
|
CodeDependencyGraphMaintainer,
|
|
)
|
|
from band_aider.context.working_backlog import (
|
|
BackLog,
|
|
DeferredRemediationHandoff,
|
|
FailedQAItem,
|
|
)
|
|
from band_aider.context.traversed_frontier_history import TraversedFrontierHistory
|
|
from band_aider.judgement.file_manager import FileManager
|
|
|
|
@dataclass
|
|
class StabilizationContext(ContextBag):
|
|
"""StabilizationContext tracks source frontiers and story-local backlog state.
|
|
|
|
Example: after topology admits a layer, ``context.append_frontier(frontier)``
|
|
records it while ``frontier_open_hashes`` preserves that depth's pre-Alpha baseline.
|
|
"""
|
|
|
|
upstream_file_manager: FileManager = field()
|
|
root_change: ClassMeta = field()
|
|
file_manager: FileManager | None = field(default=None)
|
|
cot_index: int = field(default=0)
|
|
total_cot: int = field(default=1)
|
|
surrogate_roots: List[ClassMeta] = field(default_factory=list)
|
|
completion: Dict[str, WorkCompletion] = field(default_factory=dict)
|
|
backlog: BackLog | None = field(default=None)
|
|
topology: TopologicalLiveliness = field(default_factory=TopologicalLiveliness)
|
|
# Durable opening content hashes captured once for each admitted BFS depth.
|
|
frontier_open_hashes: Dict[int, Dict[str, str]] = field(default_factory=dict)
|
|
fully_explored: bool = field(default=False)
|
|
final_exam_started: bool = field(default=False)
|
|
final_exam_finished: bool = field(default=False)
|
|
dependency_graph_maintainer: CodeDependencyGraphMaintainer | None = field(default=None)
|
|
active_topology: int | None = field(default=None)
|
|
inherited_remediation_backlog: Dict[str, List[FailedQAItem]] = field(
|
|
default_factory=dict
|
|
)
|
|
deferred_remediation_backlog: Dict[str, List[FailedQAItem]] = field(
|
|
default_factory=dict
|
|
)
|
|
prior_traversed_frontiers: TraversedFrontierHistory | None = field(default=None)
|
|
alpha_inbound_queue_depth: int = field(default=0)
|
|
alpha_outbound_queue_depth: int = field(default=0)
|
|
beta_inbound_queue_depth: int = field(default=0)
|
|
beta_outbound_queue_depth: int = field(default=0)
|
|
squashed: bool = field(default=False)
|
|
accepted: bool = field(default=False)
|
|
blocked_summary: str = field(default="")
|
|
|
|
def __getstate__(self) -> dict[str, object]:
|
|
"""Return durable stabilization state while stripping runtime parent links."""
|
|
|
|
state = super().__getstate__()
|
|
state.pop("upstream_file_manager", None)
|
|
state.pop("dependency_graph_maintainer", None)
|
|
state.pop("runtime_queue_depth_callback", None)
|
|
state.pop("new_files_callback", None)
|
|
state.pop("topology_summary", None)
|
|
state.pop("test_runner_strategy", None)
|
|
return state
|
|
|
|
def __post_init__(self) -> None:
|
|
"""Initialize durable backlog state."""
|
|
|
|
if self.backlog is None:
|
|
self.backlog = BackLog(completion=self.completion)
|
|
elif self.backlog.completion is not self.completion:
|
|
self.completion = self.backlog.completion
|
|
@property
|
|
def t0(self) -> List[ClassMeta]:
|
|
"""Return the root source topology including surrogate roots."""
|
|
|
|
return [self.root_change, *self.surrogate_roots]
|
|
|
|
@property
|
|
def full_topologies(self) -> List[List[ClassMeta]]:
|
|
"""Return every source topology from root through downstream frontiers."""
|
|
|
|
return [self.t0, *self.topology.frontiers]
|
|
|
|
@property
|
|
def remaining_deferred_remediation(self) -> DeferredRemediationHandoff:
|
|
"""Return the live handoff view without changing either staging lane."""
|
|
|
|
return DeferredRemediationHandoff(
|
|
inherited_items=[
|
|
item
|
|
for items in self.inherited_remediation_backlog.values()
|
|
for item in items
|
|
],
|
|
deferred_items=[
|
|
item
|
|
for items in self.deferred_remediation_backlog.values()
|
|
for item in items
|
|
],
|
|
)
|
|
|
|
@property
|
|
def frontier(self) -> Tuple[List[ClassMeta], int]:
|
|
"""Return the active source frontier and its BFS depth."""
|
|
|
|
return self.topology.current_frontier(self.t0)
|
|
|
|
@property
|
|
def visited_frontier_locations(self) -> set[str]:
|
|
"""Expose the liveliness-owned visited set to existing status consumers."""
|
|
|
|
return self.topology.visited
|
|
|
|
def story_source_files(self) -> list[ClassMeta]:
|
|
"""Return deduplicated source metadata currently assigned to this stabilization story."""
|
|
|
|
files: list[ClassMeta] = []
|
|
seen: set[str] = set()
|
|
for source_file in [
|
|
source_file
|
|
for frontier in self.full_topologies
|
|
for source_file in frontier
|
|
]:
|
|
if source_file.location in seen:
|
|
continue
|
|
seen.add(source_file.location)
|
|
current = None
|
|
if self.file_manager is not None:
|
|
current = self.file_manager.current_src(source_file.location)
|
|
files.append(current or source_file)
|
|
return files
|
|
|
|
def traversed_frontier_snapshot(self) -> tuple[tuple[str, ...], ...]:
|
|
"""Return this COT's actually traversed locations in BFS-depth order."""
|
|
|
|
return self.topology.traversed_snapshot(self.t0)
|
|
|
|
def append_frontier(self, frontier: list[ClassMeta]) -> None:
|
|
"""Append one admitted topological frontier."""
|
|
|
|
self.topology.append_frontier(frontier)
|
|
|
|
def extend_roots(self, *additionals: ClassMeta) -> None:
|
|
"""Add surrogate roots to the root topology."""
|
|
|
|
# FileManager synchronization owns metadata updates for newly created surrogate files.
|
|
self.surrogate_roots.extend(additionals)
|
|
# Keep the context-owned COT beam aligned with legitimately added surrogate roots.
|
|
if self.file_manager is not None:
|
|
self.topology.activate(
|
|
cot_index=self.cot_index,
|
|
file_manager=self.file_manager,
|
|
upstream_file_manager=self.upstream_file_manager,
|
|
roots=self.t0,
|
|
)
|
|
self.persist()
|
|
|
|
## NOT NEEDED, already handled by file manager hierarchies.
|
|
# def update_alpha_log(self, summary: str, files: List[ClassMeta]):
|
|
# # TODO:
|
|
# pass
|
|
#
|
|
# def update_beta_log(self, summary: str, files: List[ClassMeta]):
|
|
# # TODO:
|
|
# pass
|
|
|
|
# def update_alpha_log(self):
|
|
#
|
|
# def record_stabilization_result(
|
|
# self,
|
|
# files: list[ClassMeta],
|
|
# summary: str,
|
|
# ) -> None:
|
|
# """Record stabilization output and attach it to current COT task history."""
|
|
#
|
|
# self.marginal_result.append(summary)
|
|
#
|
|
# task = self.current_cot_task or self.original_prompt
|
|
# self.stabilization_summaries_by_cot.setdefault(task, []).append(summary)
|
|
#
|
|
# for class_meta in files:
|
|
# current = self.file_manager.current_src(class_meta.location)
|
|
# if current is not None:
|
|
# self.file_manager.upsert_src(current, summary)
|