- Deduplicated coerce_list/bool/int across platform adapters - Consolidated entry_matches, normalize_entry into helpers.py - Removed duplicate get_project_root from uninstall.py - Recovered stashed changes from subagent 1
108 lines
4.5 KiB
Python
108 lines
4.5 KiB
Python
"""Tests for hard context overflow guard (#296)."""
|
|
|
|
import pytest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
|
|
class TestHardOverflowGuard:
|
|
"""Verify the 85% hard overflow guard catches context overflow."""
|
|
|
|
def test_model_usage_calculation(self):
|
|
"""Verify model usage = real_tokens / context_length."""
|
|
real_tokens = 85_000
|
|
context_length = 100_000
|
|
usage = real_tokens / context_length
|
|
assert usage == 0.85
|
|
|
|
def test_85_percent_is_threshold(self):
|
|
"""85% of model context should trigger the hard guard."""
|
|
context_length = 100_000
|
|
# At 85% exactly
|
|
assert (85_000 / context_length) >= 0.85
|
|
# At 84.9% — should NOT trigger
|
|
assert (84_900 / context_length) < 0.85
|
|
|
|
def test_hard_guard_only_when_voluntary_skipped(self):
|
|
"""Hard guard should use elif — not fire when voluntary compression fires."""
|
|
import inspect
|
|
from run_agent import AIAgent
|
|
# Find the hard guard code in run_conversation
|
|
src = inspect.getsource(AIAgent.run_conversation)
|
|
# It should be an elif, not a separate if
|
|
# The elif ensures it only fires when voluntary compression didn't
|
|
assert "elif" in src.split("Hard overflow guard")[0].split("should_compress")[-1]
|
|
|
|
def test_hard_guard_checks_85_percent(self):
|
|
"""Hard guard threshold should be 0.85 (85%)."""
|
|
import inspect
|
|
from run_agent import AIAgent
|
|
src = inspect.getsource(AIAgent.run_conversation)
|
|
# Find the line with the threshold
|
|
for line in src.split('\n'):
|
|
if 'model_usage >= 0.85' in line or 'model_usage >= 0.85' in line:
|
|
assert True
|
|
return
|
|
# Alternative: check for >= 0.85 anywhere near the hard guard
|
|
assert "0.85" in src.split("Hard overflow guard")[1].split("Save session log")[0]
|
|
|
|
def test_hard_guard_logs_warning(self):
|
|
"""Hard guard should log a warning when triggered."""
|
|
import inspect
|
|
from run_agent import AIAgent
|
|
src = inspect.getsource(AIAgent.run_conversation)
|
|
guard_section = src.split("Hard overflow guard")[1].split("Save session log")[0]
|
|
assert "logger.warning" in guard_section
|
|
assert "forcing compression" in guard_section
|
|
|
|
def test_context_length_zero_skips(self):
|
|
"""Guard should skip when context_length is 0 (unknown model)."""
|
|
context_length = 0
|
|
# The guard checks context_length > 0 before computing usage
|
|
assert context_length > 0 is False
|
|
|
|
def test_usage_scenarios(self):
|
|
"""Test various usage levels against the 85% threshold."""
|
|
context_length = 128_000
|
|
scenarios = [
|
|
(50_000, False, "39% — well under"),
|
|
(80_000, False, "62% — under"),
|
|
(100_000, False, "78% — under but close"),
|
|
(108_800, True, "85% — exactly at threshold"),
|
|
(110_000, True, "86% — just over"),
|
|
(120_000, True, "94% — dangerously high"),
|
|
(128_000, True, "100% — at limit"),
|
|
]
|
|
for tokens, should_trigger, desc in scenarios:
|
|
usage = tokens / context_length
|
|
triggers = usage >= 0.85
|
|
assert triggers == should_trigger, f"{desc}: usage={usage:.1%}, expected trigger={should_trigger}, got={triggers}"
|
|
|
|
|
|
class TestHardGuardIntegration:
|
|
"""Test that the hard guard is present in the right location."""
|
|
|
|
def test_guard_is_in_run_conversation(self):
|
|
import inspect
|
|
from run_agent import AIAgent
|
|
src = inspect.getsource(AIAgent.run_conversation)
|
|
assert "Hard overflow guard" in src
|
|
|
|
def test_guard_uses_elif_chain(self):
|
|
"""Verify the elif structure: voluntary → hard guard → else."""
|
|
import inspect
|
|
from run_agent import AIAgent
|
|
src = inspect.getsource(AIAgent.run_conversation)
|
|
# Find the section
|
|
section = src.split("should_compress(_real_tokens)")[1].split("Save session log")[0]
|
|
# Should contain elif for the hard guard
|
|
assert "elif" in section
|
|
assert "_model_usage" in section
|
|
|
|
def test_compression_disabled_skips_hard_guard(self):
|
|
"""If compression is disabled, hard guard should also be skipped."""
|
|
import inspect
|
|
from run_agent import AIAgent
|
|
src = inspect.getsource(AIAgent.run_conversation)
|
|
section = src.split("Hard overflow guard")[1].split("Save session log")[0]
|
|
assert "self.compression_enabled" in section
|