415 lines
16 KiB
Python
415 lines
16 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from itertools import groupby
|
|
from pathlib import Path
|
|
import time
|
|
from typing import Iterable
|
|
|
|
from band_aider.context.band_aider_facts import (
|
|
BandAiderFactBucket,
|
|
BandAiderFactsSnapshot,
|
|
BandAiderTokenBreakdown,
|
|
get_band_aider_facts,
|
|
)
|
|
from band_aider.context.class_meta import ClassMeta
|
|
from band_aider.judgement.file_manager import FileManager
|
|
|
|
|
|
@dataclass
|
|
class ChangeReportWriter:
|
|
"""Render one SupremeLoop completion report as markdown in the repo root."""
|
|
|
|
repo_root: Path
|
|
reports_dir_name: str = "band-aider reports"
|
|
|
|
def write(
|
|
self,
|
|
*,
|
|
original_job: str,
|
|
cot_tasks: Iterable[str],
|
|
file_manager: FileManager,
|
|
final_summary: str,
|
|
) -> Path:
|
|
"""Write one timestamped markdown report and return its path."""
|
|
|
|
now = datetime.now().astimezone().replace(microsecond=0)
|
|
generated_at = now.isoformat()
|
|
filename = f"run-report-ending-{now.strftime('%Y%m%d-%H%M%S-%z')}.md"
|
|
reports_dir = self.repo_root / self.reports_dir_name
|
|
path = reports_dir / filename
|
|
counts = self.changed_file_counts(file_manager)
|
|
body = self.render(
|
|
generated_at=generated_at,
|
|
original_job=original_job,
|
|
cot_tasks=self._report_cot_tasks(cot_tasks, file_manager),
|
|
file_manager=file_manager,
|
|
final_summary=final_summary,
|
|
facts_snapshot=get_band_aider_facts().snapshot(
|
|
files_modified=int(counts.get("changed", 0) or 0),
|
|
files_added=int(counts.get("created", 0) or 0),
|
|
report_path=path,
|
|
),
|
|
)
|
|
repo_exec = file_manager.repo_exec
|
|
# Let RepoExec ensure parent directories exist inside the execution environment
|
|
repo_exec.write_text(path, body, encoding="utf-8")
|
|
return path
|
|
|
|
def changed_file_counts(self, file_manager: FileManager) -> dict[str, int]:
|
|
"""Return changed-file counts for final UI/report statistics."""
|
|
|
|
counts = {
|
|
"total": 0,
|
|
"changed": 0,
|
|
"created": 0,
|
|
"deleted": 0,
|
|
"unchanged": 0,
|
|
}
|
|
for record in self._changed_records(file_manager):
|
|
effect = str(record["effect"])
|
|
if effect == "unchanged":
|
|
effect = "changed"
|
|
counts["total"] += 1
|
|
counts[effect] = counts.get(effect, 0) + 1
|
|
return counts
|
|
|
|
def render(
|
|
self,
|
|
*,
|
|
generated_at: str,
|
|
original_job: str,
|
|
cot_tasks: list[str],
|
|
file_manager: FileManager,
|
|
final_summary: str,
|
|
facts_snapshot: BandAiderFactsSnapshot | None = None,
|
|
) -> str:
|
|
"""Return the markdown body for one run report."""
|
|
|
|
facts = facts_snapshot or get_band_aider_facts().snapshot()
|
|
sections = [
|
|
"# Band-Aider Change Report",
|
|
"",
|
|
f"- Generated at: `{generated_at}`",
|
|
f"- Final summary: {final_summary}",
|
|
"",
|
|
"## Run Statistics",
|
|
"",
|
|
*self._render_facts_summary(facts),
|
|
"",
|
|
"## Run Metrics by Time Bucket",
|
|
"",
|
|
*self._render_facts_buckets(facts),
|
|
"",
|
|
"## Token Breakdown",
|
|
"",
|
|
*self._render_token_breakdown(facts),
|
|
"",
|
|
"## Original Job",
|
|
"",
|
|
original_job.strip() or "_No original job recorded._",
|
|
"",
|
|
"## COT Tasks",
|
|
"",
|
|
*self._render_cot_tasks(cot_tasks),
|
|
"",
|
|
"## Changes by Directory",
|
|
"",
|
|
*self._render_directory_groups(file_manager),
|
|
]
|
|
return "\n".join(sections).rstrip() + "\n"
|
|
|
|
def _render_cot_tasks(self, cot_tasks: list[str]) -> list[str]:
|
|
"""Return markdown for the persisted task breakdown."""
|
|
|
|
if not cot_tasks:
|
|
return ["_No COT tasks were recorded._"]
|
|
return [f"{index}. {task}" for index, task in enumerate(cot_tasks, start=1)]
|
|
|
|
def _render_facts_summary(self, facts: BandAiderFactsSnapshot) -> list[str]:
|
|
"""Return the top-level run facts table."""
|
|
|
|
return [
|
|
"| Metric | Value |",
|
|
"| --- | ---: |",
|
|
f"| Total tokens spent | {facts.total_tokens_spent} |",
|
|
f"| Cache-hit tokens | {facts.total_cache_hit_tokens} |",
|
|
f"| Baseline tokens | {facts.baseline_tokens_spent} ({self._format_percent(facts.baseline_tokens_spent, facts.total_tokens_spent)}) |",
|
|
f"| Escalated tokens | {facts.escalated_tokens_spent} ({self._format_percent(facts.escalated_tokens_spent, facts.total_tokens_spent)}) |",
|
|
f"| Largest query tokens | {facts.largest_query_tokens} |",
|
|
f"| LLM calls | {facts.llm_call_count} |",
|
|
f"| Time waiting on LLM | {self._format_duration(facts.llm_wait_seconds)} |",
|
|
f"| Total operation time | {self._format_duration(facts.total_seconds)} |",
|
|
f"| COT count | {facts.cot_count} |",
|
|
f"| Average time per COT | {self._format_duration(facts.average_seconds_per_cot)} |",
|
|
f"| Files modified | {facts.files_modified} |",
|
|
f"| Files added | {facts.files_added} |",
|
|
]
|
|
|
|
def _render_facts_buckets(self, facts: BandAiderFactsSnapshot) -> list[str]:
|
|
"""Return bucketed metric rows with a final TOTAL row."""
|
|
|
|
lines = [
|
|
"| Window | LLM calls | Tokens spent | Cache hits | LLM wait | COT count | COT time |",
|
|
"| --- | ---: | ---: | ---: | ---: | ---: | ---: |",
|
|
]
|
|
for bucket in facts.buckets:
|
|
lines.append(self._render_bucket_row(bucket))
|
|
total_bucket = BandAiderFactBucket(
|
|
started_at=0,
|
|
tokens_spent=sum(bucket.tokens_spent for bucket in facts.buckets),
|
|
cache_hit_tokens=sum(bucket.cache_hit_tokens for bucket in facts.buckets),
|
|
llm_calls=sum(bucket.llm_calls for bucket in facts.buckets),
|
|
llm_wait_seconds=sum(bucket.llm_wait_seconds for bucket in facts.buckets),
|
|
cot_seconds=sum(bucket.cot_seconds for bucket in facts.buckets),
|
|
cot_count=sum(bucket.cot_count for bucket in facts.buckets),
|
|
)
|
|
lines.append(self._render_bucket_row(total_bucket, label="TOTAL"))
|
|
return lines
|
|
|
|
def _render_bucket_row(
|
|
self,
|
|
bucket: BandAiderFactBucket,
|
|
*,
|
|
label: str | None = None,
|
|
) -> str:
|
|
window = label or time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(bucket.started_at))
|
|
return (
|
|
f"| {window} | {bucket.llm_calls} | {bucket.tokens_spent} | {bucket.cache_hit_tokens} | "
|
|
f"{self._format_duration(bucket.llm_wait_seconds)} | {bucket.cot_count} | "
|
|
f"{self._format_duration(bucket.cot_seconds)} |"
|
|
)
|
|
|
|
def _render_token_breakdown(self, facts: BandAiderFactsSnapshot) -> list[str]:
|
|
"""Return markdown token breakdown rows grouped by time, tier, model, and endpoint."""
|
|
|
|
if not facts.token_breakdown:
|
|
return ["_No token breakdown was recorded._"]
|
|
lines = [
|
|
"| Time bin | Level | Model name | Endpoint | Tokens spent | Cache hits |",
|
|
"| --- | --- | --- | --- | ---: | ---: |",
|
|
]
|
|
for item in self._aggregate_token_breakdown(facts.token_breakdown):
|
|
lines.append(self._render_token_breakdown_row(item))
|
|
lines.append(f"| TOTAL | all | all | all | {facts.total_tokens_spent} | {facts.total_cache_hit_tokens} |")
|
|
lines.extend(["", "### Subtotal by Model", ""])
|
|
lines.extend(
|
|
[
|
|
"| Level | Model name | Endpoint | Tokens spent | Cache hits |",
|
|
"| --- | --- | --- | ---: | ---: |",
|
|
]
|
|
)
|
|
subtotals: dict[tuple[str, str, str], tuple[int, int]] = {}
|
|
for item in facts.token_breakdown:
|
|
key = (item.level, item.model_name, item.endpoint)
|
|
spent, cache_hits = subtotals.get(key, (0, 0))
|
|
subtotals[key] = (
|
|
spent + item.tokens_spent,
|
|
cache_hits + item.cache_hit_tokens,
|
|
)
|
|
for level, model_name, endpoint in sorted(subtotals):
|
|
spent, cache_hits = subtotals[(level, model_name, endpoint)]
|
|
lines.append(f"| {level} | {model_name} | {endpoint} | {spent} | {cache_hits} |")
|
|
lines.append(f"| TOTAL | all | all | {facts.total_tokens_spent} | {facts.total_cache_hit_tokens} |")
|
|
return lines
|
|
|
|
def _aggregate_token_breakdown(
|
|
self,
|
|
items: tuple[BandAiderTokenBreakdown, ...],
|
|
) -> list[BandAiderTokenBreakdown]:
|
|
totals: dict[tuple[int, str, str, str], tuple[int, int]] = {}
|
|
for item in items:
|
|
key = (item.started_at, item.level, item.model_name, item.endpoint)
|
|
spent, cache_hits = totals.get(key, (0, 0))
|
|
totals[key] = (
|
|
spent + item.tokens_spent,
|
|
cache_hits + item.cache_hit_tokens,
|
|
)
|
|
return [
|
|
BandAiderTokenBreakdown(
|
|
started_at=started_at,
|
|
level=level,
|
|
model_name=model_name,
|
|
endpoint=endpoint,
|
|
tokens_spent=usage[0],
|
|
cache_hit_tokens=usage[1],
|
|
)
|
|
for (started_at, level, model_name, endpoint), usage in sorted(totals.items())
|
|
]
|
|
|
|
def _render_token_breakdown_row(self, item: BandAiderTokenBreakdown) -> str:
|
|
window = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(item.started_at))
|
|
return (
|
|
f"| {window} | {item.level} | {item.model_name} | {item.endpoint} | "
|
|
f"{item.tokens_spent} | {item.cache_hit_tokens} |"
|
|
)
|
|
|
|
def _format_percent(self, value: int, total: int) -> str:
|
|
if total <= 0:
|
|
return "0.0%"
|
|
return f"{(max(0, value) / total) * 100:.1f}%"
|
|
|
|
def _format_duration(self, seconds: float) -> str:
|
|
seconds = max(0, int(round(seconds)))
|
|
minutes, remaining = divmod(seconds, 60)
|
|
hours, minutes = divmod(minutes, 60)
|
|
if hours:
|
|
return f"{hours}h {minutes}m {remaining}s"
|
|
if minutes:
|
|
return f"{minutes}m {remaining}s"
|
|
return f"{remaining}s"
|
|
|
|
def _report_cot_tasks(
|
|
self,
|
|
cot_tasks: Iterable[str],
|
|
file_manager: FileManager,
|
|
) -> list[str]:
|
|
"""Return COT tasks, falling back to stabilization history when available."""
|
|
|
|
tasks = [task for task in cot_tasks if str(task).strip()]
|
|
if tasks:
|
|
return tasks
|
|
history = getattr(file_manager, "stabilization_summaries_by_cot", None)
|
|
if isinstance(history, dict):
|
|
return [str(task) for task in history if str(task).strip()]
|
|
return []
|
|
|
|
def _render_directory_groups(self, file_manager: FileManager) -> list[str]:
|
|
"""Return markdown grouped by parent directory in location order."""
|
|
|
|
records = self._changed_records(file_manager)
|
|
if not records:
|
|
return ["_No file changes were recorded._"]
|
|
|
|
lines: list[str] = []
|
|
for directory, group in groupby(records, key=lambda record: record["directory"]):
|
|
lines.extend([f"### `{directory}`", ""])
|
|
for record in group:
|
|
lines.extend(self._render_file_record(record))
|
|
return lines[:-1] if lines and lines[-1] == "" else lines
|
|
|
|
def _render_file_record(self, record: dict[str, object]) -> list[str]:
|
|
"""Return markdown for one changed file."""
|
|
|
|
class_meta: ClassMeta = record["meta"] # type: ignore[assignment]
|
|
lines = [
|
|
f"#### `{record['location']}`",
|
|
"",
|
|
f"- Kind: `{record['scope']}`",
|
|
f"- Effect: `{record['effect']}`",
|
|
f"- Summary: {class_meta.summary or '_No summary recorded._'}",
|
|
]
|
|
|
|
symbols = class_meta.relevant_symbols
|
|
if symbols:
|
|
lines.append("- Relevant symbols:")
|
|
lines.extend(
|
|
f" - `{symbol}`: {reason}"
|
|
for symbol, reason in sorted(symbols.items())
|
|
)
|
|
|
|
if class_meta.cdc:
|
|
lines.append("- CDC:")
|
|
lines.extend(
|
|
f" - `{changed_at.isoformat()}` `{actor}` {description}"
|
|
for changed_at, actor, description in class_meta.cdc
|
|
)
|
|
else:
|
|
lines.append("- CDC: _No CDC entries recorded._")
|
|
|
|
lines.append("")
|
|
return lines
|
|
|
|
def _changed_records(self, file_manager: FileManager) -> list[dict[str, object]]:
|
|
"""Return dirty-file report records sorted by repo-relative location."""
|
|
|
|
records: list[dict[str, object]] = []
|
|
src_effects = file_manager.src_net_effects()
|
|
test_effects = file_manager.test_net_effects()
|
|
|
|
for location in self._changed_locations(file_manager, kind="src", effects=src_effects):
|
|
class_meta = file_manager.current_src(location)
|
|
if class_meta is None:
|
|
continue
|
|
records.append(
|
|
self._record_for(
|
|
location=location,
|
|
effect=src_effects.get(location, "unchanged"),
|
|
scope="source",
|
|
class_meta=class_meta,
|
|
)
|
|
)
|
|
|
|
for location in self._changed_locations(file_manager, kind="test", effects=test_effects):
|
|
class_meta = file_manager.current_test(location)
|
|
if class_meta is None:
|
|
continue
|
|
records.append(
|
|
self._record_for(
|
|
location=location,
|
|
effect=test_effects.get(location, "unchanged"),
|
|
scope="test",
|
|
class_meta=class_meta,
|
|
)
|
|
)
|
|
|
|
return sorted(records, key=lambda record: str(record["location"]))
|
|
|
|
def _changed_locations(
|
|
self,
|
|
file_manager: FileManager,
|
|
*,
|
|
kind: str,
|
|
effects: dict[str, str],
|
|
) -> list[str]:
|
|
"""Return reportable locations from dirty flags, net effects, or new CDC."""
|
|
|
|
dirty = getattr(file_manager, f"{kind}_dirty", {}) or {}
|
|
current = getattr(file_manager, f"{kind}_classes", []) or []
|
|
original = getattr(file_manager, f"original_{kind}_classes", []) or []
|
|
original_by_location = {
|
|
class_meta.location: class_meta
|
|
for class_meta in original
|
|
}
|
|
locations: set[str] = {
|
|
location
|
|
for location, flag in dirty.items()
|
|
if flag
|
|
}
|
|
locations.update(
|
|
location
|
|
for location, effect in effects.items()
|
|
if effect != "unchanged"
|
|
)
|
|
for class_meta in current:
|
|
original_meta = original_by_location.get(class_meta.location)
|
|
original_cdc_count = len(original_meta.cdc) if original_meta is not None else 0
|
|
if len(class_meta.cdc) > original_cdc_count:
|
|
locations.add(class_meta.location)
|
|
return sorted(locations)
|
|
|
|
def _record_for(
|
|
self,
|
|
*,
|
|
location: str,
|
|
effect: str,
|
|
scope: str,
|
|
class_meta: ClassMeta,
|
|
) -> dict[str, object]:
|
|
"""Return one normalized report record."""
|
|
|
|
directory = str(Path(location).parent)
|
|
if directory == ".":
|
|
directory = "repo root"
|
|
return {
|
|
"directory": directory,
|
|
"location": location,
|
|
"effect": effect,
|
|
"scope": scope,
|
|
"meta": class_meta,
|
|
}
|