70 lines
2.3 KiB
Python
70 lines
2.3 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
from __future__ import annotations
|
|
|
|
from contextlib import AbstractContextManager
|
|
from typing import Generic, TypeVar
|
|
|
|
from band_aider.context.work_completion import WorkCompletion
|
|
|
|
T = TypeVar("T")
|
|
|
|
|
|
class BacklogScope(AbstractContextManager, Generic[T]):
|
|
"""BacklogScope owns claim/commit rollback for one claimed backlog item."""
|
|
|
|
def __init__(
|
|
self,
|
|
value: T,
|
|
*,
|
|
ledger: dict[str, WorkCompletion] | None = None,
|
|
) -> None:
|
|
"""Store one claimed backlog value plus explicit completion references."""
|
|
|
|
self.value = value
|
|
self._ledger = ledger
|
|
self._committed = False
|
|
|
|
def commit(self) -> None:
|
|
"""Commit one claimed backlog item exactly once."""
|
|
|
|
if self._committed:
|
|
return
|
|
|
|
self._committed = True
|
|
|
|
def __enter__(self) -> "BacklogScope[T]":
|
|
"""Return the claimed scope."""
|
|
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, exc_tb) -> None:
|
|
"""Rollback uncommitted or exceptional completion claims."""
|
|
|
|
if exc_type is not None or not self._committed:
|
|
if self._ledger is None:
|
|
return
|
|
|
|
from band_aider.context.working_backlog import FailedQAItem, InspectionBacklog, TestingBacklog
|
|
|
|
if isinstance(self.value, FailedQAItem):
|
|
if self.value.focused_spotlight is not None:
|
|
self._ledger.pop(self.value.focused_spotlight.location, None)
|
|
return
|
|
|
|
if isinstance(self.value, TestingBacklog):
|
|
if self.value.focused_spotlight is None:
|
|
return
|
|
for downstream in self.value.downstream_classes:
|
|
source_completion = self._ledger.get(downstream.location)
|
|
if source_completion is None or source_completion.tests is None:
|
|
continue
|
|
source_completion.tests.pop(self.value.focused_spotlight.location, None)
|
|
return
|
|
|
|
if isinstance(self.value, InspectionBacklog):
|
|
if self.value.focused_spotlight is not None:
|
|
self._ledger.pop(self.value.focused_spotlight.location, None)
|
|
return
|