323 lines
12 KiB
Python
323 lines
12 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from dataclasses import dataclass, field
|
|
from typing import Callable, List
|
|
from uuid import uuid4
|
|
|
|
from band_aider.context import provenance
|
|
from band_aider.context.contex_bag import ContextBag
|
|
from band_aider.context.class_meta import ClassMeta
|
|
from band_aider.context.provenance import Phase
|
|
from band_aider.context.working_backlog import (
|
|
BacklogLane,
|
|
BacklogScope,
|
|
ContributorBacklog,
|
|
FailedQAItem,
|
|
InspectionBacklog,
|
|
OutboundBacklogItem,
|
|
StagedInboundItem,
|
|
TestingBacklog,
|
|
)
|
|
from band_aider.judgement.file_manager import FileManager
|
|
|
|
|
|
@dataclass
|
|
class AlphaContext(ContextBag):
|
|
"""AlphaContext owns one source-only alpha pass and its contributor backlog view."""
|
|
|
|
file_manager: FileManager = field()
|
|
backlog: ContributorBacklog = field()
|
|
inbound_queue: List[StagedInboundItem[InspectionBacklog]] = field(default_factory=list)
|
|
outbound_queue: List[OutboundBacklogItem] = field(default_factory=list)
|
|
active_inbound: StagedInboundItem[InspectionBacklog] | None = None
|
|
active_inbound_id: str | None = None
|
|
pinned_context: List[str] = field(default_factory=list)
|
|
# handled_sources is the accepted-change SOT, including editable preflight files.
|
|
alpha_postflight_callback: Callable[[tuple[ClassMeta, ...]], None] | None = None
|
|
|
|
def move_inbound(
|
|
self,
|
|
item: StagedInboundItem[InspectionBacklog] | None = None,
|
|
) -> StagedInboundItem[InspectionBacklog] | None:
|
|
"""Move one source backlog item into Alpha-owned inbound state."""
|
|
|
|
if self.inbound_queue and (item is None or self.active_inbound is not item):
|
|
return None
|
|
if item is None:
|
|
peeked = self.backlog.peek_next_source_work()
|
|
if peeked is None:
|
|
return None
|
|
lane, value = peeked
|
|
inbound = StagedInboundItem(
|
|
lane=lane,
|
|
value=value,
|
|
inbound_id=uuid4().hex,
|
|
)
|
|
else:
|
|
inbound = item
|
|
if not inbound.inbound_id:
|
|
inbound.inbound_id = uuid4().hex
|
|
if inbound.lane is BacklogLane.Remediation:
|
|
if not isinstance(inbound.value, FailedQAItem):
|
|
raise ValueError("Alpha remediation inbound requires FailedQAItem.")
|
|
elif inbound.lane is BacklogLane.Inspection:
|
|
if type(inbound.value) is not InspectionBacklog:
|
|
raise ValueError("Alpha inspection inbound requires InspectionBacklog.")
|
|
else:
|
|
raise ValueError("Alpha inbound accepts only remediation or inspection work.")
|
|
self.inbound_queue = [inbound]
|
|
self.active_inbound = inbound
|
|
self.active_inbound_id = inbound.inbound_id
|
|
return inbound
|
|
|
|
def move_inbound_from_backlog(
|
|
self,
|
|
logger: logging.Logger | None = None,
|
|
) -> BacklogScope[FailedQAItem | InspectionBacklog] | None:
|
|
"""Claim one source backlog item into Alpha-owned inbound state."""
|
|
|
|
if self.inbound_queue:
|
|
return None
|
|
peeked = self.backlog.peek_next_source_work(logger=logger)
|
|
if peeked is None:
|
|
return None
|
|
lane, _ = peeked
|
|
claimed = self.backlog.pop_next_source_work(logger=logger)
|
|
if claimed is None:
|
|
return None
|
|
inbound = self.move_inbound(
|
|
StagedInboundItem[InspectionBacklog](
|
|
lane=lane,
|
|
value=claimed.value,
|
|
inbound_id="",
|
|
)
|
|
)
|
|
if inbound is None:
|
|
raise RuntimeError("Alpha claimed source work could not enter its inbound state.")
|
|
inbound.popped_from_backlog = True
|
|
provenance.start_phase_item(Phase.ALPHA, inbound.value)
|
|
return claimed
|
|
|
|
def move_outbound(self) -> None:
|
|
"""Mark the active Alpha inbound item as moved to outbound."""
|
|
|
|
if self.inbound_queue:
|
|
self.inbound_queue.clear()
|
|
|
|
def move_outbound_to_backlog(self, logger: logging.Logger | None = None) -> None:
|
|
"""Move all staged Alpha outbound items into the shared backlog."""
|
|
|
|
if not self.outbound_queue:
|
|
return
|
|
self.backlog.flush_outbound(self.outbound_queue, logger=logger)
|
|
self.outbound_queue.clear()
|
|
|
|
def stage_testing_outbound(
|
|
self,
|
|
item: TestingBacklog,
|
|
) -> None:
|
|
"""Stage one Alpha testing emission with context-owned dedupe and lineage."""
|
|
|
|
active = self.active_inbound
|
|
if active is None:
|
|
raise RuntimeError("Alpha outbound staging requires an active inbound.")
|
|
# Cross-phase output cannot inherit Alpha's revived guidance.
|
|
# Provenance.revive retains legitimate restart guidance and bypasses this staging method.
|
|
item.future_guidance = ()
|
|
# Only localized guidance crosses into Beta; raw source edit history stays in Alpha.
|
|
item.editor_records = []
|
|
if item.focused_spotlight is None:
|
|
raise ValueError("Alpha testing outbound requires focused_spotlight.")
|
|
key = item.focused_spotlight.location
|
|
# Named test obligations must not merge merely because they share a file.
|
|
if item.failure_test_name:
|
|
key += "\0" + item.failure_test_name
|
|
for staged in self.outbound_queue:
|
|
if staged.lane is not BacklogLane.Testing or staged.dedupe_key != key:
|
|
continue
|
|
staged.lineage[active.inbound_id] = item
|
|
staged.value = [
|
|
value
|
|
for value in staged.lineage.values()
|
|
if isinstance(value, TestingBacklog)
|
|
][0]
|
|
provenance.track_provenance(
|
|
active.value.message_id,
|
|
Phase.BETA,
|
|
staged.value,
|
|
)
|
|
self.move_outbound()
|
|
return
|
|
self.outbound_queue.append(
|
|
OutboundBacklogItem(
|
|
lane=BacklogLane.Testing,
|
|
dedupe_key=key,
|
|
value=item,
|
|
lineage={active.inbound_id: item},
|
|
)
|
|
)
|
|
provenance.track_provenance(active.value.message_id, Phase.BETA, item)
|
|
self.move_outbound()
|
|
|
|
def stage_remediation_outbound(
|
|
self,
|
|
item: FailedQAItem,
|
|
) -> None:
|
|
"""Stage one Alpha postflight remediation item with structural dedupe and lineage."""
|
|
|
|
active = self.active_inbound
|
|
if active is None:
|
|
raise RuntimeError("Alpha remediation outbound staging requires an active inbound.")
|
|
if item.focused_spotlight is None:
|
|
raise ValueError("Alpha remediation outbound requires focused_spotlight.")
|
|
# Require the observed failure identity before establishing remediation lineage.
|
|
if item.failed_test is None:
|
|
raise ValueError("Alpha remediation outbound requires failed_test.")
|
|
key = f"{item.focused_spotlight.location}\0{item.failed_test.test.location}"
|
|
# Preserve the originating method through another Alpha postflight handoff.
|
|
if item.failed_test.test_name:
|
|
key += "\0" + item.failed_test.test_name
|
|
lineage_value: dict[
|
|
str,
|
|
InspectionBacklog | FailedQAItem | TestingBacklog,
|
|
] = {active.inbound_id: item}
|
|
staged_lane = BacklogLane.Remediation
|
|
staged_key = key
|
|
for staged in self.outbound_queue:
|
|
if staged.lane is not staged_lane or staged.dedupe_key != staged_key:
|
|
continue
|
|
staged.lineage[active.inbound_id] = item
|
|
staged.value = [
|
|
value
|
|
for value in staged.lineage.values()
|
|
if isinstance(value, FailedQAItem)
|
|
][0]
|
|
provenance.track_provenance(
|
|
active.value.message_id,
|
|
Phase.ALPHA,
|
|
staged.value,
|
|
)
|
|
self.move_outbound()
|
|
return
|
|
self.outbound_queue.append(
|
|
OutboundBacklogItem(
|
|
lane=staged_lane,
|
|
dedupe_key=staged_key,
|
|
value=item,
|
|
lineage=lineage_value,
|
|
)
|
|
)
|
|
provenance.track_provenance(active.value.message_id, Phase.ALPHA, item)
|
|
self.move_outbound()
|
|
|
|
def recover_active_inbound(
|
|
self,
|
|
logger: logging.Logger | None = None,
|
|
) -> None:
|
|
"""Repair interrupted Alpha pass state before resuming backlog drain."""
|
|
|
|
active = self.active_inbound
|
|
if active is None:
|
|
return
|
|
if not active.popped_from_backlog:
|
|
self.clear_moved_inbound()
|
|
return
|
|
if active.clean_cursor is not None:
|
|
self.backlog.rollback_to_cursor(active.clean_cursor, logger=logger)
|
|
active.clean_cursor = None
|
|
self.backlog.release_claim_for(active.value)
|
|
kept: list[OutboundBacklogItem] = []
|
|
for staged in self.outbound_queue:
|
|
if active.inbound_id not in staged.lineage:
|
|
kept.append(staged)
|
|
continue
|
|
staged.lineage.pop(active.inbound_id, None)
|
|
if not staged.lineage:
|
|
continue
|
|
if staged.lane is BacklogLane.Testing:
|
|
staged.value = [
|
|
value
|
|
for value in staged.lineage.values()
|
|
if isinstance(value, TestingBacklog)
|
|
][0]
|
|
else:
|
|
staged.value = [
|
|
value
|
|
for value in staged.lineage.values()
|
|
if isinstance(value, FailedQAItem)
|
|
][0]
|
|
kept.append(staged)
|
|
self.outbound_queue = kept
|
|
self.move_inbound(active)
|
|
provenance.start_phase_item(Phase.ALPHA, active.value)
|
|
|
|
def discard_active_inbound(self) -> None:
|
|
"""Discard the active item's staged outputs after a phase-level restart request."""
|
|
|
|
active = self.active_inbound
|
|
if active is None:
|
|
return
|
|
self.backlog.release_claim_for(active.value)
|
|
retained: list[OutboundBacklogItem] = []
|
|
# Keep only work emitted by other durable inbound items.
|
|
for staged in self.outbound_queue:
|
|
staged.lineage.pop(active.inbound_id, None)
|
|
if not staged.lineage:
|
|
continue
|
|
if staged.lane is BacklogLane.Testing:
|
|
staged.value = next(
|
|
value
|
|
for value in staged.lineage.values()
|
|
if isinstance(value, TestingBacklog)
|
|
)
|
|
else:
|
|
staged.value = next(
|
|
value
|
|
for value in staged.lineage.values()
|
|
if isinstance(value, FailedQAItem)
|
|
)
|
|
retained.append(staged)
|
|
self.outbound_queue = retained
|
|
self.clear_moved_inbound()
|
|
|
|
def flush_outbound_with_wal(
|
|
self,
|
|
logger: logging.Logger | None = None,
|
|
) -> None:
|
|
"""Flush one staged Alpha outbound batch with WAL cursor bookkeeping."""
|
|
|
|
if not self.outbound_queue:
|
|
self.clear_moved_inbound()
|
|
return
|
|
active = self.active_inbound
|
|
if active is None:
|
|
raise RuntimeError("Alpha outbound flush requires an active inbound anchor.")
|
|
active.clean_cursor = self.backlog.flush_cursor()
|
|
self.move_outbound_to_backlog(logger=logger)
|
|
active.clean_cursor = None
|
|
self.clear_moved_inbound()
|
|
|
|
def clear_moved_inbound(self) -> None:
|
|
"""Clear active Alpha inbound bookkeeping after outbound flush/no-op."""
|
|
|
|
self.active_inbound = None
|
|
self.active_inbound_id = None
|
|
self.inbound_queue = []
|
|
|
|
def __getstate__(self) -> dict[str, object]:
|
|
"""Return durable alpha state while stripping runtime-owned collaborators."""
|
|
|
|
state = super().__getstate__()
|
|
state.pop("file_manager", None)
|
|
state.pop("backlog", None)
|
|
state.pop("new_files_callback", None)
|
|
state.pop("runtime_queue_depth_callback", None)
|
|
state.pop("topology_summary", None)
|
|
state.pop("test_runner_strategy", None)
|
|
state.pop("alpha_postflight_callback", None)
|
|
return state
|