99 lines
3.6 KiB
Python
99 lines
3.6 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from pathlib import Path
|
|
|
|
from band_aider.context.band_aider_facts import (
|
|
FACTS_FILENAME,
|
|
get_band_aider_facts,
|
|
reset_band_aider_facts_for_tests,
|
|
)
|
|
from tests.support.memory_repo_exec import MemoryRepoExec
|
|
|
|
|
|
def test_band_aider_facts_persist_and_restore_run_metrics() -> None:
|
|
"""Verify facts own durable run metrics across hydration boundaries."""
|
|
|
|
adapter = MemoryRepoExec()
|
|
checkpoint_dir = Path("/testbed/facts")
|
|
reset_band_aider_facts_for_tests()
|
|
facts = get_band_aider_facts()
|
|
facts.begin_operation(10.0)
|
|
facts.record_llm_call(tokens_spent=100, wait_seconds=2.5, wall_time=300)
|
|
facts.record_llm_call(
|
|
tokens_spent=250,
|
|
cache_hit_tokens=40,
|
|
wait_seconds=4.0,
|
|
wall_time=300,
|
|
model_name="gpt-5.5",
|
|
endpoint="https://cloud.example",
|
|
level="escalated",
|
|
)
|
|
facts.finish_cot(facts.begin_cot(20.0), when=35.0, wall_time=300)
|
|
facts.finish_operation(40.0)
|
|
facts.record_report(report_path="/tmp/report.md", files_modified=3, files_added=1)
|
|
|
|
facts.persist(checkpoint_dir, repo_exec=adapter)
|
|
assert adapter.exists(checkpoint_dir / FACTS_FILENAME)
|
|
|
|
reset_band_aider_facts_for_tests()
|
|
restored = get_band_aider_facts()
|
|
restored.restore(checkpoint_dir, repo_exec=adapter)
|
|
snapshot = restored.snapshot()
|
|
|
|
assert snapshot.total_tokens_spent == 350
|
|
assert snapshot.total_cache_hit_tokens == 40
|
|
assert snapshot.largest_query_tokens == 250
|
|
assert snapshot.llm_call_count == 2
|
|
assert snapshot.llm_wait_seconds == 6.5
|
|
assert snapshot.cot_count == 1
|
|
assert snapshot.cot_seconds == 15.0
|
|
assert snapshot.total_seconds == 30.0
|
|
assert snapshot.files_modified == 3
|
|
assert snapshot.files_added == 1
|
|
assert snapshot.report_path == "/tmp/report.md"
|
|
assert snapshot.buckets[-1].tokens_spent == 350
|
|
assert snapshot.buckets[-1].cache_hit_tokens == 40
|
|
assert snapshot.buckets[-1].llm_calls == 2
|
|
assert snapshot.baseline_tokens_spent == 100
|
|
assert snapshot.escalated_tokens_spent == 250
|
|
assert [
|
|
(item.level, item.model_name, item.endpoint, item.tokens_spent, item.cache_hit_tokens)
|
|
for item in snapshot.token_breakdown
|
|
] == [
|
|
("baseline", "unknown", "unknown", 100, 0),
|
|
("escalated", "gpt-5.5", "https://cloud.example", 250, 40),
|
|
]
|
|
|
|
|
|
def test_restore_corrupted_facts_checkpoint_resets_without_raising(caplog) -> None:
|
|
adapter = MemoryRepoExec()
|
|
checkpoint_dir = Path("/testbed/facts")
|
|
reset_band_aider_facts_for_tests()
|
|
facts = get_band_aider_facts()
|
|
facts.record_llm_call(tokens_spent=321, wait_seconds=0, wall_time=300)
|
|
adapter.write_text(checkpoint_dir / FACTS_FILENAME, '{"total_tokens_spent":')
|
|
|
|
with caplog.at_level(logging.WARNING):
|
|
facts.restore(checkpoint_dir, repo_exec=adapter)
|
|
|
|
assert facts.snapshot().total_tokens_spent == 0
|
|
assert "Ignoring corrupted token statistics checkpoint" in caplog.text
|
|
|
|
|
|
def test_restore_corrupted_legacy_checkpoint_resets_without_raising(caplog) -> None:
|
|
adapter = MemoryRepoExec()
|
|
checkpoint_dir = Path("/testbed/facts")
|
|
reset_band_aider_facts_for_tests()
|
|
facts = get_band_aider_facts()
|
|
adapter.write_text(checkpoint_dir / "TokenBudgetStats.json", "[]")
|
|
|
|
with caplog.at_level(logging.WARNING):
|
|
facts.restore(checkpoint_dir, repo_exec=adapter)
|
|
|
|
assert facts.snapshot().total_tokens_spent == 0
|
|
assert "Ignoring corrupted token statistics checkpoint" in caplog.text
|