53e0065e3e
配置 SSOT（TOML + .env） 统一错误契约 Auth 与事务边界 Redis / Celery 可靠性:业务 Redis（DB/0）与 Celery broker/backend（DB/1）显式拆分；连接池、sync client 可观测性（OpenTelemetry + LGTM）
75 lines
3.1 KiB
Python
75 lines
3.1 KiB
Python
"""agent_logging: DEBUG 下载荷、hash_only、去重。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import app.core.agent_logging as agent_logging
|
|
from app.core.runtime_constants import agent_log_defaults
|
|
|
|
|
|
class _StubLogger:
|
|
def __init__(self) -> None:
|
|
self.debug_calls: list[tuple[str, tuple[object, ...]]] = []
|
|
|
|
def debug(self, msg: str, *args: object, **kwargs: object) -> None:
|
|
self.debug_calls.append((msg, args))
|
|
|
|
|
|
def _clear_dedup() -> None:
|
|
with agent_logging._dedup_lock:
|
|
agent_logging._last_prompt_sha256_by_label.clear()
|
|
|
|
|
|
def test_log_agent_payload_skips_when_not_verbose(monkeypatch: object) -> None:
|
|
monkeypatch.setattr("app.core.config.settings.log_level", "INFO")
|
|
log = _StubLogger()
|
|
agent_logging.log_agent_payload(log, "x.prompt", "hello")
|
|
assert log.debug_calls == []
|
|
|
|
|
|
def test_log_agent_payload_preview_includes_sha12(monkeypatch: object) -> None:
|
|
monkeypatch.setattr("app.core.config.settings.log_level", "DEBUG")
|
|
monkeypatch.setattr("app.core.runtime_constants.agent_log_defaults.prompt_mode", "preview")
|
|
monkeypatch.setattr("app.core.runtime_constants.agent_log_defaults.prompt_dedup", False)
|
|
monkeypatch.setattr("app.core.runtime_constants.agent_log_defaults.max_chars", 100)
|
|
_clear_dedup()
|
|
log = _StubLogger()
|
|
agent_logging.log_agent_payload(log, "Unit.prompt", "hello world")
|
|
assert len(log.debug_calls) == 1
|
|
msg, args = log.debug_calls[0]
|
|
assert "agent_payload" in msg
|
|
assert "sha12=" in msg
|
|
assert args[0] == "Unit.prompt"
|
|
assert args[4] == "hello world"
|
|
|
|
|
|
def test_log_agent_payload_hash_only_no_preview(monkeypatch: object) -> None:
|
|
monkeypatch.setattr("app.core.config.settings.log_level", "DEBUG")
|
|
monkeypatch.setattr("app.core.runtime_constants.agent_log_defaults.prompt_mode", "hash_only")
|
|
monkeypatch.setattr("app.core.runtime_constants.agent_log_defaults.prompt_dedup", False)
|
|
_clear_dedup()
|
|
log = _StubLogger()
|
|
body = "x" * 500
|
|
agent_logging.log_agent_payload(log, "Unit.prompt", body)
|
|
assert len(log.debug_calls) == 1
|
|
msg, args = log.debug_calls[0]
|
|
assert "mode=hash_only" in msg
|
|
assert args[0] == "Unit.prompt"
|
|
assert args[1] == 500
|
|
assert isinstance(args[2], str) and len(args[2]) == 12
|
|
|
|
|
|
def test_log_agent_payload_dedup_second_call_skipped(monkeypatch: object) -> None:
|
|
monkeypatch.setattr("app.core.config.settings.log_level", "DEBUG")
|
|
monkeypatch.setattr("app.core.runtime_constants.agent_log_defaults.prompt_mode", "preview")
|
|
monkeypatch.setattr("app.core.runtime_constants.agent_log_defaults.prompt_dedup", True)
|
|
monkeypatch.setattr("app.core.runtime_constants.agent_log_defaults.max_chars", 200)
|
|
_clear_dedup()
|
|
log = _StubLogger()
|
|
agent_logging.log_agent_payload(log, "DedupLabel.prompt", "same text")
|
|
agent_logging.log_agent_payload(log, "DedupLabel.prompt", "same text")
|
|
assert len(log.debug_calls) == 2
|
|
assert "agent_payload_skipped" in log.debug_calls[1][0]
|
|
skip_args = log.debug_calls[1][1]
|
|
assert skip_args[0] == "DedupLabel.prompt"
|
|
assert skip_args[2] == len("same text")
|