226 lines
7.6 KiB
Python
226 lines
7.6 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
|
|
from pathlib import Path
|
|
from collections.abc import Iterator
|
|
|
|
import pytest
|
|
|
|
from band_aider.ai_client.llm_logging import (
|
|
attach_queue_logging_handler,
|
|
initialize_llm_logging,
|
|
log_llm_request,
|
|
log_llm_response,
|
|
log_markdown_document,
|
|
reset_llm_logging_for_tests,
|
|
safe_log_segment,
|
|
)
|
|
from band_aider.ui.dto.log_message import LogMessageDTO
|
|
from band_aider.ui.queue_logging_handler import QueueLoggingHandler
|
|
from band_aider.context.contex_bag import ContextBag
|
|
from band_aider.loop.loop_gate import LoopGate
|
|
from band_aider.loop.proto_loop import ProtoLoop
|
|
from band_aider.ui.status_publisher import StatusPublisher
|
|
from tests.support.completion_clients import StaticOpenAICompatibleClient
|
|
from tests.support.memory_repo_exec import MemoryRepoExec
|
|
|
|
|
|
class _Publisher(StatusPublisher):
|
|
"""Collect queue projections through the nominal publisher boundary."""
|
|
|
|
def __init__(self) -> None:
|
|
self.events: list[object] = []
|
|
|
|
def publish(self, event: object) -> None:
|
|
self.events.append(event)
|
|
|
|
|
|
@dataclass
|
|
class _TinyContext(ContextBag):
|
|
pass
|
|
|
|
|
|
class _OneShotGate(LoopGate):
|
|
def __init__(self, context: ContextBag) -> None:
|
|
self.context = context
|
|
def may_enter(self) -> bool: return True
|
|
def should_retry(self) -> bool: return False
|
|
def should_continue(self) -> bool: return False
|
|
|
|
|
|
class _ChildLoop(ProtoLoop[_OneShotGate, _TinyContext]):
|
|
def _prepare_work(self) -> ProtoLoop | None: return None
|
|
def _post_run(self) -> str:
|
|
self.logger.info("child-local")
|
|
return "child-done"
|
|
|
|
|
|
class _ParentLoop(ProtoLoop[_OneShotGate, _TinyContext]):
|
|
def __init__(self, gate: _OneShotGate, context: _TinyContext) -> None:
|
|
super().__init__(gate=gate, context=context); self._used = False
|
|
def _prepare_work(self) -> ProtoLoop | None:
|
|
if self._used: return None
|
|
self._used = True
|
|
return _ChildLoop(gate=_OneShotGate(_TinyContext(goal="child", classes=[])), context=_TinyContext(goal="child", classes=[]), logger=self.logger.getChild("child"))
|
|
def _post_run(self) -> str:
|
|
self.logger.info("parent-local")
|
|
return "parent-done"
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_logging_state() -> Iterator[None]:
|
|
"""Keep each logging example independent of process-wide logger state."""
|
|
|
|
reset_llm_logging_for_tests()
|
|
yield
|
|
reset_llm_logging_for_tests()
|
|
|
|
|
|
def test_llm_request_logging_uses_color_and_yaml(capsys) -> None:
|
|
"""Verify console LLM request logging uses color and YAML-style formatting."""
|
|
|
|
adapter = MemoryRepoExec()
|
|
initialize_llm_logging(Path("/testbed"), adapter, log_to_console=True)
|
|
|
|
log_llm_request(
|
|
source="TestClient",
|
|
exchange_id=7,
|
|
endpoint="http://model.test/chat/completions",
|
|
model="test-model",
|
|
payload={"messages": [{"role": "user", "content": "hello"}]},
|
|
)
|
|
|
|
stderr = capsys.readouterr().err
|
|
assert "\x1b[" in stderr
|
|
assert "messages:" in stderr
|
|
assert '{"messages"' not in stderr
|
|
|
|
|
|
def test_llm_markdown_logging_writes_rolling_markdown_file() -> None:
|
|
"""Verify markdown prompt logging writes markdown through the rolling file handler."""
|
|
|
|
adapter = MemoryRepoExec()
|
|
root = Path("/testbed")
|
|
initialize_llm_logging(root, adapter, log_to_console=True)
|
|
|
|
log_markdown_document("unit prompt", "# Prompt\n\nDo work.")
|
|
|
|
text = "\n".join(
|
|
adapter.read_text(path)
|
|
for path in adapter.list_files(root / ".band-aider-logs", ["*_INFO.md"], recursive=True)
|
|
)
|
|
assert "Markdown Prompt - unit prompt" in text
|
|
assert "```markdown" in text
|
|
assert "# Prompt" in text
|
|
|
|
|
|
def test_stationary_completion_response_has_logged_request_and_response(capsys) -> None:
|
|
"""Verify isolated completion data can be logged without invoking an endpoint."""
|
|
|
|
adapter = MemoryRepoExec()
|
|
root = Path("/testbed")
|
|
initialize_llm_logging(root, adapter, log_to_console=True)
|
|
client = StaticOpenAICompatibleClient("logged answer")
|
|
|
|
assert client.complete("hello current client") == "logged answer"
|
|
log_llm_request(
|
|
source="StaticOpenAICompatibleClient",
|
|
exchange_id=1,
|
|
endpoint="http://model.invalid/chat/completions",
|
|
model="test-model",
|
|
payload={"messages": [{"role": "user", "content": "hello current client"}]},
|
|
)
|
|
log_llm_response(
|
|
source="StaticOpenAICompatibleClient",
|
|
exchange_id=1,
|
|
endpoint="http://model.invalid/chat/completions",
|
|
model="test-model",
|
|
status=200,
|
|
content="logged answer",
|
|
)
|
|
|
|
stderr = capsys.readouterr().err
|
|
assert "LLM REQUEST #1 StaticOpenAICompatibleClient" in stderr
|
|
assert "content: hello current client" in stderr
|
|
assert "LLM RESPONSE #1 StaticOpenAICompatibleClient" in stderr
|
|
assert "logged answer" in stderr
|
|
|
|
rolling = "\n".join(
|
|
adapter.read_text(path)
|
|
for path in adapter.list_files(root / ".band-aider-logs", ["*_INFO.md"], recursive=True)
|
|
)
|
|
assert "LLM Request #1 - StaticOpenAICompatibleClient" in rolling
|
|
assert "LLM Response #1 - StaticOpenAICompatibleClient" in rolling
|
|
assert "logged answer" in rolling
|
|
|
|
|
|
def test_nested_loop_loggers_bubble_child_records_to_parent() -> None:
|
|
adapter = MemoryRepoExec()
|
|
root = Path("/testbed")
|
|
initialize_llm_logging(root, adapter)
|
|
context = _TinyContext(goal="parent", classes=[])
|
|
_ParentLoop(gate=_OneShotGate(context), context=context).run()
|
|
parent_dir = root / ".band-aider-logs" / "cli" / safe_log_segment("parentloop")
|
|
parent_logs = adapter.list_files(parent_dir, ["*_INFO.md"], recursive=False)
|
|
child_logs = adapter.list_files(
|
|
parent_dir / safe_log_segment("child"),
|
|
["*_INFO.md"],
|
|
recursive=False,
|
|
)
|
|
assert parent_logs and child_logs
|
|
assert "parent-local" in adapter.read_text(parent_logs[0])
|
|
assert "child-local" in adapter.read_text(parent_logs[0])
|
|
assert "child-local" in adapter.read_text(child_logs[0])
|
|
|
|
|
|
def test_attach_queue_logging_handler_routes_console_logger_without_stderr_duplication(
|
|
capsys,
|
|
) -> None:
|
|
publisher = _Publisher()
|
|
handler = QueueLoggingHandler(publisher)
|
|
|
|
attach_queue_logging_handler(handler)
|
|
log_llm_request(
|
|
source="TestClient",
|
|
exchange_id=9,
|
|
endpoint="http://model.test/chat/completions",
|
|
model="test-model",
|
|
payload={"messages": [{"role": "user", "content": "hello queue"}]},
|
|
)
|
|
|
|
stderr = capsys.readouterr().err
|
|
assert stderr == ""
|
|
console_events = [
|
|
event
|
|
for event in publisher.events
|
|
if isinstance(event, LogMessageDTO) and event.logger_name == "band_aider.llm.console"
|
|
]
|
|
assert console_events
|
|
assert any("LLM REQUEST #9 TestClient" in event.message for event in console_events)
|
|
assert any("content: hello queue" in event.message for event in console_events)
|
|
assert all(event.level == "INFO" for event in console_events)
|
|
|
|
|
|
def test_queue_logging_handler_splits_multiline_messages() -> None:
|
|
publisher = _Publisher()
|
|
handler = QueueLoggingHandler(publisher)
|
|
logger = logging.getLogger("test.queue.multiline")
|
|
record = logger.makeRecord(
|
|
logger.name,
|
|
logging.WARNING,
|
|
"test_llm_logging.py",
|
|
1,
|
|
"first line\nsecond line",
|
|
args=(),
|
|
exc_info=None,
|
|
)
|
|
|
|
handler.emit(record)
|
|
|
|
messages = [event.message for event in publisher.events if isinstance(event, LogMessageDTO)]
|
|
assert messages == ["first line", "second line"]
|