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 List
|
|
from uuid import uuid4
|
|
|
|
from band_aider.context import provenance
|
|
from band_aider.context.beta_workflow_story import TargetedFailureStory
|
|
from band_aider.context.contex_bag import ContextBag
|
|
from band_aider.context.provenance import Phase
|
|
from band_aider.context.working_backlog import (
|
|
BacklogLane,
|
|
BacklogScope,
|
|
FailedQAItem,
|
|
OutboundBacklogItem,
|
|
QABacklog,
|
|
StagedInboundItem,
|
|
TestingBacklog,
|
|
)
|
|
from band_aider.judgement.file_manager import FileManager
|
|
from band_aider.judgement.test_runner_strategy import TestRunnerStrategy
|
|
|
|
|
|
@dataclass
|
|
class BetaContext(ContextBag):
|
|
"""Own Beta's active QA work, staged emissions and new post-edit failure awaiting routing.
|
|
|
|
After closing the old claim, ``context.start_post_edit(story)`` retains the
|
|
new observation until its ownership decision and outbound flush complete.
|
|
"""
|
|
|
|
file_manager: FileManager = field()
|
|
backlog: QABacklog = field()
|
|
test_runner_strategy: TestRunnerStrategy = field(init=False, repr=False)
|
|
inbound_queue: List[StagedInboundItem[TestingBacklog]] = field(default_factory=list)
|
|
outbound_queue: List[OutboundBacklogItem] = field(default_factory=list)
|
|
active_inbound: StagedInboundItem[TestingBacklog] | None = None
|
|
active_inbound_id: str | None = None
|
|
pinned_context: List[str] = field(default_factory=list)
|
|
post_edit: TargetedFailureStory | None = None
|
|
|
|
def start_post_edit(self, story: TargetedFailureStory) -> None:
|
|
"""Retain a new failure for routing after the preceding QA claim has committed."""
|
|
|
|
active = self.active_inbound
|
|
item = story.testing_item
|
|
# Post-edit routing must link actual new work to its completed predecessor.
|
|
if active is None or item is None:
|
|
raise ValueError("Beta post-edit routing requires the completed inbound and a new QA item.")
|
|
provenance.track_provenance(active.value.message_id, Phase.BETA, item)
|
|
self.clear_moved_inbound()
|
|
self.move_inbound(StagedInboundItem(BacklogLane.Testing, item, ""))
|
|
self.post_edit = story
|
|
|
|
def move_inbound(
|
|
self,
|
|
item: StagedInboundItem[TestingBacklog] | None = None,
|
|
) -> StagedInboundItem[TestingBacklog] | None:
|
|
"""Move one testing backlog item into Beta-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_test_backlog()
|
|
if peeked is None:
|
|
return None
|
|
inbound = StagedInboundItem(
|
|
lane=BacklogLane.Testing,
|
|
value=peeked,
|
|
inbound_id=uuid4().hex,
|
|
)
|
|
else:
|
|
inbound = item
|
|
if not inbound.inbound_id:
|
|
inbound.inbound_id = uuid4().hex
|
|
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[TestingBacklog] | None:
|
|
"""Claim one testing backlog item into Beta-owned inbound state."""
|
|
|
|
if self.inbound_queue:
|
|
return None
|
|
claimed = self.backlog.pop_test_backlog(logger=logger)
|
|
if claimed is None:
|
|
return None
|
|
inbound = self.move_inbound(
|
|
StagedInboundItem[TestingBacklog](
|
|
lane=BacklogLane.Testing,
|
|
value=claimed.value,
|
|
inbound_id="",
|
|
)
|
|
)
|
|
if inbound is None:
|
|
raise RuntimeError("Beta claimed test work could not enter its inbound state.")
|
|
inbound.popped_from_backlog = True
|
|
provenance.start_phase_item(Phase.BETA, inbound.value)
|
|
return claimed
|
|
|
|
def move_outbound(self) -> None:
|
|
"""Mark the active Beta 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 Beta 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,
|
|
*,
|
|
prepend: bool = False,
|
|
) -> None:
|
|
"""Stage one Beta testing emission with context-owned dedupe and lineage."""
|
|
|
|
active = self.active_inbound
|
|
if active is None:
|
|
raise RuntimeError("Beta outbound staging requires an active inbound.")
|
|
if item.focused_spotlight is None:
|
|
raise ValueError("Beta testing outbound requires focused_spotlight.")
|
|
key = item.focused_spotlight.location
|
|
# Separate named observations while preserving historical file-level staging keys.
|
|
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]
|
|
staged.prepend = staged.prepend or prepend
|
|
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},
|
|
prepend=prepend,
|
|
)
|
|
)
|
|
provenance.track_provenance(active.value.message_id, Phase.BETA, item)
|
|
self.move_outbound()
|
|
|
|
def stage_remediation_outbound(
|
|
self,
|
|
item: FailedQAItem,
|
|
) -> None:
|
|
"""Stage one Beta remediation emission with context-owned dedupe and lineage."""
|
|
|
|
active = self.active_inbound
|
|
if active is None:
|
|
raise RuntimeError("Beta outbound staging requires an active inbound.")
|
|
# Cross-phase output cannot inherit Beta's revived guidance.
|
|
# Provenance.revive retains legitimate restart guidance and bypasses this staging method.
|
|
item.future_guidance = ()
|
|
# Only localized guidance crosses into Alpha; raw test edit history stays in Beta.
|
|
item.editor_records = []
|
|
if item.focused_spotlight is None:
|
|
raise ValueError("Beta remediation outbound requires focused_spotlight.")
|
|
key = item.focused_spotlight.location
|
|
# Several methods may independently require repair in the same source file.
|
|
if item.failed_test is not None and item.failed_test.test_name:
|
|
key += "\0" + item.failed_test.test.location + "\0" + item.failed_test.test_name
|
|
for staged in self.outbound_queue:
|
|
if staged.lane is not BacklogLane.Remediation or staged.dedupe_key != 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=BacklogLane.Remediation,
|
|
dedupe_key=key,
|
|
value=item,
|
|
lineage={active.inbound_id: item},
|
|
)
|
|
)
|
|
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 Beta pass state before resuming backlog drain."""
|
|
|
|
active = self.active_inbound
|
|
if active is None:
|
|
return
|
|
# A post-edit observation owns routing, not an unfinished editor/QA claim.
|
|
if not active.popped_from_backlog and self.post_edit is None:
|
|
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
|
|
# Retrying new-failure routing cannot undo the preceding completed obligation.
|
|
if self.post_edit is 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.BETA, 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 Beta 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("Beta 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 Beta inbound bookkeeping after outbound flush/no-op."""
|
|
|
|
self.active_inbound = None
|
|
self.active_inbound_id = None
|
|
self.inbound_queue = []
|
|
self.post_edit = None
|
|
|
|
def __getstate__(self) -> dict[str, object]:
|
|
"""Return durable beta state while stripping runtime-owned collaborators."""
|
|
|
|
state = super().__getstate__()
|
|
state.pop("file_manager", None)
|
|
state.pop("backlog", None)
|
|
state.pop("runtime_queue_depth_callback", None)
|
|
state.pop("test_runner_strategy", None)
|
|
return state
|