397 lines
14 KiB
Python
397 lines
14 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
from __future__ import annotations
|
|
|
|
from contextlib import contextmanager
|
|
import time
|
|
from typing import Iterator
|
|
from typing import TYPE_CHECKING
|
|
from band_aider.ui.dto.human_intervention import InterventionChoicesDTO
|
|
|
|
if TYPE_CHECKING:
|
|
from band_aider.judgement.file_manager import FileManager
|
|
|
|
from band_aider.ui.dto.backlog_status import BacklogStatusDTO
|
|
from band_aider.ui.dto.bootstrap_status import BootstrapStatusDTO
|
|
from band_aider.ui.dto.cot_status import CotStatusDTO
|
|
from band_aider.ui.dto.frontier_status import FrontierStatusDTO
|
|
from band_aider.ui.dto.loop_status import LoopStatusDTO
|
|
from band_aider.ui.dto.stabilization_status import StabilizationStatusDTO
|
|
from band_aider.ui.dto.story_status import StoryStatusDTO
|
|
from band_aider.ui.dto.supreme_status import SupremeStatusDTO
|
|
from band_aider.ui.status_publisher import NullStatusPublisher, StatusPublisher
|
|
|
|
_PUBLISHER: StatusPublisher = NullStatusPublisher()
|
|
_SUPREME_STATUS = SupremeStatusDTO()
|
|
_BACKLOG_DECAY_SECONDS = 5 * 60
|
|
_BACKLOG_DECAY_STATE = {
|
|
"cmp": {"value": 0, "timestamp": 0.0},
|
|
"ins": {"value": 0, "timestamp": 0.0},
|
|
"rem": {"value": 0, "timestamp": 0.0},
|
|
"test": {"value": 0, "timestamp": 0.0},
|
|
}
|
|
|
|
|
|
def _reset_backlog_decay_state() -> None:
|
|
for key in _BACKLOG_DECAY_STATE:
|
|
_BACKLOG_DECAY_STATE[key]["value"] = 0
|
|
_BACKLOG_DECAY_STATE[key]["timestamp"] = 0.0
|
|
|
|
|
|
def _held_backlog_value(name: str, current: int, now: float) -> int:
|
|
state = _BACKLOG_DECAY_STATE[name]
|
|
held = int(state["value"])
|
|
seen_at = float(state["timestamp"])
|
|
if current >= held:
|
|
state["value"] = current
|
|
state["timestamp"] = now
|
|
return current
|
|
if now - seen_at <= _BACKLOG_DECAY_SECONDS:
|
|
return held
|
|
state["value"] = current
|
|
state["timestamp"] = now
|
|
return current
|
|
|
|
|
|
def set_status_publisher(publisher: StatusPublisher | None) -> None:
|
|
global _PUBLISHER
|
|
_PUBLISHER = publisher or NullStatusPublisher()
|
|
|
|
|
|
@contextmanager
|
|
def status_publisher_scope(publisher: StatusPublisher | None) -> Iterator[None]:
|
|
global _PUBLISHER
|
|
previous = _PUBLISHER
|
|
_PUBLISHER = publisher or NullStatusPublisher()
|
|
_reset_backlog_decay_state()
|
|
try:
|
|
yield
|
|
finally:
|
|
_PUBLISHER = previous
|
|
_reset_backlog_decay_state()
|
|
|
|
|
|
def publish(event: object) -> None:
|
|
_PUBLISHER.publish(event)
|
|
|
|
|
|
def publish_intervention_files(manager: FileManager) -> None:
|
|
"""Publish authoritative managed paths across the current manager's ancestry."""
|
|
files: set[str] = set()
|
|
current: FileManager | None = manager
|
|
# Child inventories supplement the root without hiding its selectable files.
|
|
while current is not None:
|
|
files.update(map(current.repo_location, current.src_index))
|
|
files.update(map(current.repo_location, current.test_index))
|
|
current = current.upstream_file_manager
|
|
publish(InterventionChoicesDTO(tuple(sorted(files))))
|
|
|
|
|
|
def publish_supreme_status(**changes: object) -> None:
|
|
global _SUPREME_STATUS
|
|
data = {
|
|
"title": _SUPREME_STATUS.title,
|
|
"project": _SUPREME_STATUS.project,
|
|
"task": _SUPREME_STATUS.task,
|
|
"backlog": _SUPREME_STATUS.backlog,
|
|
"bootstrap": _SUPREME_STATUS.bootstrap,
|
|
"cot": _SUPREME_STATUS.cot,
|
|
"stabilization": _SUPREME_STATUS.stabilization,
|
|
"frontier": _SUPREME_STATUS.frontier,
|
|
"loop": _SUPREME_STATUS.loop,
|
|
"escalation": _SUPREME_STATUS.escalation,
|
|
"token_budget": _SUPREME_STATUS.token_budget,
|
|
"story": _SUPREME_STATUS.story,
|
|
"direct_code_editor": _SUPREME_STATUS.direct_code_editor,
|
|
}
|
|
data.update(changes)
|
|
_SUPREME_STATUS = SupremeStatusDTO(**data)
|
|
publish(_SUPREME_STATUS)
|
|
|
|
|
|
def publish_story_status(context: object) -> None:
|
|
"""Publish persisted Sovereign story progress through the existing status queue."""
|
|
|
|
stories = tuple(str(story) for story in context.stories or ())
|
|
if not stories:
|
|
return
|
|
story_cursor = context.story_cursor
|
|
current_index = max(
|
|
0,
|
|
min(int(story_cursor.current_story_index or 0), len(stories)),
|
|
)
|
|
completed = tuple(range(current_index))
|
|
publish_supreme_status(
|
|
story=StoryStatusDTO(
|
|
current_index=current_index,
|
|
total_count=len(stories),
|
|
all_stories=stories,
|
|
completed_indices=completed,
|
|
),
|
|
cot=None,
|
|
)
|
|
|
|
|
|
def publish_loop_status(
|
|
current_loop: str,
|
|
previous_loop: str | None = None,
|
|
next_loop: str | None = None,
|
|
current_item: str | None = None,
|
|
work_focus: str | None = None,
|
|
) -> None:
|
|
publish_supreme_status(
|
|
loop=LoopStatusDTO(
|
|
previous_loop=previous_loop,
|
|
current_loop=current_loop,
|
|
next_loop=next_loop,
|
|
current_item=current_item,
|
|
work_focus=work_focus,
|
|
)
|
|
)
|
|
|
|
|
|
def publish_backlog_status(
|
|
backlog: object,
|
|
deferred_count: int = 0,
|
|
final_exam_pending: bool = False,
|
|
final_exam_active: bool = False,
|
|
alpha_inbound_queue_depth: int = 0,
|
|
alpha_outbound_queue_depth: int = 0,
|
|
beta_inbound_queue_depth: int = 0,
|
|
beta_outbound_queue_depth: int = 0,
|
|
) -> None:
|
|
# Backlog has four components to display: Completion, Inspection, Remediation, Testing
|
|
now = time.time()
|
|
cmp = _held_backlog_value("cmp", len(getattr(backlog, "completion", {}) or {}), now)
|
|
ins = _held_backlog_value("ins", len(getattr(backlog, "inspection", ()) or ()), now)
|
|
rem = _held_backlog_value("rem", len(getattr(backlog, "remediation", ()) or ()), now)
|
|
test = _held_backlog_value("test", len(getattr(backlog, "testing", ()) or ()), now)
|
|
has_last_exam = any(
|
|
getattr(item, "last_exam", False)
|
|
for item in getattr(backlog, "testing", ()) or ()
|
|
)
|
|
dto = BacklogStatusDTO(
|
|
cmp=cmp,
|
|
ins=ins,
|
|
rem=rem,
|
|
test=test,
|
|
alpha_inbound_queue=alpha_inbound_queue_depth,
|
|
alpha_outbound_queue=alpha_outbound_queue_depth,
|
|
beta_inbound_queue=beta_inbound_queue_depth,
|
|
beta_outbound_queue=beta_outbound_queue_depth,
|
|
final_exam_pending=final_exam_pending or has_last_exam,
|
|
final_exam_active=final_exam_active,
|
|
)
|
|
publish_supreme_status(backlog=dto)
|
|
|
|
|
|
def publish_cot_status(context: object) -> None:
|
|
story_cursor = context.story_cursor
|
|
task_cursor = story_cursor.task_cursor
|
|
tasks = list(story_cursor.cot_task_records or [])
|
|
if not tasks:
|
|
return
|
|
current_index = int(task_cursor.current_cot_index or 0)
|
|
current_index = max(0, min(current_index, len(tasks) - 1))
|
|
completed = tuple(range(1, current_index + 1))
|
|
publish_supreme_status(
|
|
task=(
|
|
story_cursor.active_story
|
|
or context.original_prompt
|
|
or context.goal
|
|
),
|
|
bootstrap=None,
|
|
cot=CotStatusDTO(
|
|
current_index=current_index + 1,
|
|
total_count=len(tasks),
|
|
all_tasks=tuple(record.task for record in tasks),
|
|
completed_indices=completed,
|
|
),
|
|
)
|
|
|
|
|
|
_BOOTSTRAP_STAGE_LABELS = {
|
|
"initialized": "initializing",
|
|
"extract_beats": "beats extraction",
|
|
"beats_extracted": "concern selection",
|
|
"concern_selected": "idiom samples",
|
|
"cataloging_idioms": "idiom catalog",
|
|
"designing": "design",
|
|
"editing": "editing",
|
|
"refreshing": "refresh",
|
|
"completed": "completed",
|
|
"blocked": "blocked",
|
|
}
|
|
_BOOTSTRAP_STAGE_ORDER = [
|
|
"initialized",
|
|
"extract_beats",
|
|
"beats_extracted",
|
|
"concern_selected",
|
|
"cataloging_idioms",
|
|
"designing",
|
|
"editing",
|
|
"refreshing",
|
|
"completed",
|
|
]
|
|
|
|
|
|
def _trim_status_text(value: object, limit: int = 48) -> str:
|
|
"""Return a single-line terminal-safe status fragment."""
|
|
|
|
# TODO: Deliberate with a human whether to remove this arbitrary status-text cap.
|
|
text = " ".join(str(value or "").split())
|
|
if len(text) <= limit:
|
|
return text
|
|
return text[: max(0, limit - 3)].rstrip() + "..."
|
|
|
|
|
|
def _bootstrap_stage_label(stage: str) -> str:
|
|
"""Return a compact human label for one bootstrap stage."""
|
|
|
|
return _BOOTSTRAP_STAGE_LABELS.get(stage, stage.replace("_", " "))
|
|
|
|
|
|
def _bootstrap_next_stage(stage: str) -> str | None:
|
|
"""Return the next broad bootstrap stage label."""
|
|
|
|
try:
|
|
index = _BOOTSTRAP_STAGE_ORDER.index(stage)
|
|
except ValueError:
|
|
return None
|
|
if index + 1 >= len(_BOOTSTRAP_STAGE_ORDER):
|
|
return None
|
|
return _bootstrap_stage_label(_BOOTSTRAP_STAGE_ORDER[index + 1])
|
|
|
|
|
|
def _bootstrap_detail(context: object, stage: str) -> str:
|
|
"""Return one compact sample line for bootstrap status."""
|
|
|
|
visible = list(getattr(context, "visible_locations", []) or [])
|
|
beats_by_location = getattr(context, "beats_by_location", {}) or {}
|
|
concerns = list(getattr(context, "cross_cutting_concerns", []) or [])
|
|
selected = getattr(context, "selected_concern", None)
|
|
idioms = list(getattr(context, "idiom_catalog", []) or [])
|
|
pending_pairs = list(getattr(context, "pending_idiom_pairs", []) or [])
|
|
design_plan = getattr(context, "design_plan", {}) or {}
|
|
refreshed = list(getattr(context, "refreshed_locations", []) or [])
|
|
involved = list(getattr(context, "involved_locations", []) or [])
|
|
|
|
if stage in {"initialized", "extract_beats", "beats_extracted"}:
|
|
for location, beats in beats_by_location.items():
|
|
if beats:
|
|
return _trim_status_text(f"beats {location}: {len(beats)}", 42)
|
|
return f"visible {len(visible)} files"
|
|
if stage == "concern_selected" or stage == "cataloging_idioms":
|
|
if isinstance(selected, dict):
|
|
name = selected.get("name") or selected.get("summary") or "selected"
|
|
return _trim_status_text(f"concern {name}", 42)
|
|
return f"concerns {len(concerns)}"
|
|
if stage == "designing":
|
|
return f"idioms {len(idioms)} pending {len(pending_pairs)}"
|
|
if stage == "editing":
|
|
editable = len(design_plan.get("editable_locations", []) or [])
|
|
new_files = len(design_plan.get("new_files", []) or [])
|
|
return f"editables {editable} new {new_files}"
|
|
if stage == "refreshing":
|
|
return f"refresh {len(refreshed)}/{len(involved)}"
|
|
if stage == "blocked":
|
|
return _trim_status_text(getattr(context, "blocked_summary", "") or "blocked", 42)
|
|
if stage == "completed":
|
|
created = len(getattr(context, "created_locations", []) or [])
|
|
return f"created {created} involved {len(involved)}"
|
|
return _bootstrap_stage_label(stage)
|
|
|
|
|
|
def publish_bootstrap_status(context: object) -> None:
|
|
"""Publish bootstrap progress in the stabilization panel and temporary COT slot."""
|
|
|
|
stage = str(getattr(context, "stage", "") or "initialized")
|
|
stage_label = _bootstrap_stage_label(stage)
|
|
beats_by_location = getattr(context, "beats_by_location", {}) or {}
|
|
beat_count = sum(len(beats) for beats in beats_by_location.values())
|
|
completed = stage == "completed"
|
|
task = getattr(context, "goal", None)
|
|
bootstrap_task = f"bootstrap: {stage_label}"
|
|
publish_supreme_status(
|
|
task=task,
|
|
cot=CotStatusDTO(
|
|
current_index=1,
|
|
total_count=1,
|
|
all_tasks=(bootstrap_task,),
|
|
completed_indices=(1,) if completed else (),
|
|
),
|
|
bootstrap=BootstrapStatusDTO(
|
|
stage=stage,
|
|
stage_label=stage_label,
|
|
iteration=int(getattr(context, "iteration", 0) or 0) + 1,
|
|
max_iterations=int(getattr(context, "max_iterations", 0) or 0),
|
|
visible_count=len(getattr(context, "visible_locations", []) or []),
|
|
beat_count=beat_count,
|
|
concern_count=len(getattr(context, "cross_cutting_concerns", []) or []),
|
|
idiom_count=len(getattr(context, "idiom_catalog", []) or []),
|
|
created_count=len(getattr(context, "created_locations", []) or []),
|
|
involved_count=len(getattr(context, "involved_locations", []) or []),
|
|
current=_bootstrap_detail(context, stage),
|
|
next=_bootstrap_next_stage(stage),
|
|
detail=_bootstrap_detail(context, stage),
|
|
),
|
|
stabilization=None,
|
|
frontier=None,
|
|
)
|
|
|
|
|
|
def publish_stabilization_status(context: object) -> None:
|
|
backlog = getattr(context, "backlog", None)
|
|
deferred = getattr(context, "deferred_remediation_backlog", {}) or {}
|
|
deferred_count = sum(len(items) for items in deferred.values())
|
|
if backlog is not None:
|
|
publish_backlog_status(
|
|
backlog,
|
|
deferred_count=deferred_count,
|
|
alpha_inbound_queue_depth=int(getattr(context, "alpha_inbound_queue_depth", 0) or 0),
|
|
alpha_outbound_queue_depth=int(getattr(context, "alpha_outbound_queue_depth", 0) or 0),
|
|
beta_inbound_queue_depth=int(getattr(context, "beta_inbound_queue_depth", 0) or 0),
|
|
beta_outbound_queue_depth=int(getattr(context, "beta_outbound_queue_depth", 0) or 0),
|
|
final_exam_pending=any(
|
|
getattr(item, "last_exam", False)
|
|
for item in getattr(backlog, "testing", ()) or ()
|
|
),
|
|
final_exam_active=bool(
|
|
getattr(context, "final_exam_started", False)
|
|
and not getattr(context, "final_exam_finished", False)
|
|
),
|
|
)
|
|
|
|
root = getattr(getattr(context, "root_change", None), "location", None)
|
|
frontier_pair = getattr(context, "frontier", ([], 0))
|
|
source_frontier, frontier_index = frontier_pair
|
|
files = tuple(getattr(item, "location", str(item)) for item in source_frontier)
|
|
all_frontiers = getattr(context, "full_topologies", []) or []
|
|
next_frontier = (
|
|
all_frontiers[frontier_index + 1]
|
|
if frontier_index + 1 < len(all_frontiers)
|
|
else []
|
|
)
|
|
next_file = getattr(next_frontier[0], "location", None) if next_frontier else None
|
|
visited = getattr(context, "visited_frontier_locations", set()) or set()
|
|
active = getattr(context, "active_topology", None)
|
|
current = files[0] if files else None
|
|
stabilization = StabilizationStatusDTO(
|
|
root=root,
|
|
frontier_count=len(files),
|
|
visited_count=len(visited),
|
|
files_in_play_count=len(files),
|
|
previous=None,
|
|
current=f"Topology {frontier_index}" if active is not None else current,
|
|
next=next_file,
|
|
)
|
|
frontier = FrontierStatusDTO(
|
|
current=current,
|
|
next_sibling=files[1] if len(files) > 1 else next_file,
|
|
files=files,
|
|
visited_count=len(visited),
|
|
files_in_play_count=len(files),
|
|
)
|
|
publish_supreme_status(bootstrap=None, stabilization=stabilization, frontier=frontier)
|