wehub-resource-sync
809 lines
28 KiB
Python
809 lines
28 KiB
Python
"""Tests for TurnRunner._maybe_compact_on_t3_upgrade()."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from dataclasses import dataclass, field
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from opensquilla.engine import runtime as runtime_module
|
|
from opensquilla.engine.pipeline import TurnContext
|
|
from opensquilla.engine.runtime import TurnRunner
|
|
from opensquilla.session.compaction import CompactionConfig
|
|
from opensquilla.session.models import TranscriptEntry
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fakes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class _FakeSessionManager:
|
|
def __init__(self, transcript: list[TranscriptEntry] | None = None) -> None:
|
|
self._transcript = transcript or []
|
|
self.compact_calls: list[tuple[str, int]] = []
|
|
self.compact_configs: list[object | None] = []
|
|
|
|
async def get_transcript(self, session_key: str, **kwargs: Any) -> list[TranscriptEntry]:
|
|
return list(self._transcript)
|
|
|
|
async def compact(self, session_key: str, context_window_tokens: int, **kwargs: Any) -> str:
|
|
self.compact_calls.append((session_key, context_window_tokens))
|
|
return "summary"
|
|
|
|
|
|
class _ResultCompactionSessionManager(_FakeSessionManager):
|
|
async def compact_with_result(
|
|
self,
|
|
session_key: str,
|
|
context_window_tokens: int,
|
|
config: object | None = None,
|
|
) -> SimpleNamespace:
|
|
self.compact_calls.append((session_key, context_window_tokens))
|
|
self.compact_configs.append(config)
|
|
return SimpleNamespace(
|
|
summary="summary",
|
|
kept_entries=[{"role": "assistant", "content": "tail"}],
|
|
removed_count=2,
|
|
chunks_processed=1,
|
|
summary_source="llm",
|
|
tokens_before=300,
|
|
tokens_after=100,
|
|
remaining_budget_tokens=context_window_tokens - 100,
|
|
)
|
|
|
|
|
|
class _StaleResultCompactionSessionManager(_FakeSessionManager):
|
|
async def compact_with_result(
|
|
self,
|
|
session_key: str,
|
|
context_window_tokens: int,
|
|
config: object | None = None,
|
|
**kwargs: Any,
|
|
) -> SimpleNamespace:
|
|
self.compact_calls.append((session_key, context_window_tokens))
|
|
self.compact_configs.append(config)
|
|
return SimpleNamespace(
|
|
summary="",
|
|
kept_entries=list(self._transcript),
|
|
removed_count=0,
|
|
chunks_processed=0,
|
|
summary_source="skipped",
|
|
skip_reason="stale_preimage",
|
|
tokens_before=300,
|
|
tokens_after=300,
|
|
remaining_budget_tokens=context_window_tokens - 300,
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class _FakeFlushReceipt:
|
|
mode: str = "llm"
|
|
flushed_paths: list[str] = field(default_factory=list)
|
|
slug: str | None = None
|
|
message_count: int = 1
|
|
duration_ms: int = 10
|
|
raw_reason: str | None = None
|
|
error: str | None = None
|
|
integrity_status: str = "ok"
|
|
indexed_chunk_count: int = 1
|
|
output_coverage_status: str = "ok"
|
|
invalid_candidate_count: int = 0
|
|
candidate_missing_ids: list[str] = field(default_factory=list)
|
|
obligation_status: str = "ok"
|
|
obligation_missing_ids: list[str] = field(default_factory=list)
|
|
|
|
|
|
class _FakeFlushService:
|
|
def __init__(
|
|
self,
|
|
receipt: _FakeFlushReceipt | None = None,
|
|
raise_exc: Exception | None = None,
|
|
delay_seconds: float = 0.0,
|
|
) -> None:
|
|
self._receipt = receipt or _FakeFlushReceipt()
|
|
self._raise_exc = raise_exc
|
|
self._delay_seconds = delay_seconds
|
|
self.execute_calls: list[dict[str, Any]] = []
|
|
|
|
async def execute(self, transcript: Any, session_key: str, **kwargs: Any) -> _FakeFlushReceipt:
|
|
self.execute_calls.append({"session_key": session_key, **kwargs})
|
|
if self._delay_seconds:
|
|
import asyncio
|
|
|
|
await asyncio.sleep(self._delay_seconds)
|
|
if self._raise_exc is not None:
|
|
raise self._raise_exc
|
|
return self._receipt
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _sample_transcript() -> list[TranscriptEntry]:
|
|
return [
|
|
TranscriptEntry(
|
|
session_id="s1",
|
|
session_key="agent:main:webchat:default",
|
|
role="user",
|
|
content="hello",
|
|
token_count=60_000,
|
|
),
|
|
TranscriptEntry(
|
|
session_id="s1",
|
|
session_key="agent:main:webchat:default",
|
|
role="assistant",
|
|
content="hi there",
|
|
token_count=60_000,
|
|
),
|
|
]
|
|
|
|
|
|
def _within_budget_transcript() -> list[TranscriptEntry]:
|
|
return [
|
|
TranscriptEntry(
|
|
session_id="s1",
|
|
session_key="agent:main:webchat:default",
|
|
role="user",
|
|
content="hello",
|
|
token_count=10,
|
|
),
|
|
TranscriptEntry(
|
|
session_id="s1",
|
|
session_key="agent:main:webchat:default",
|
|
role="assistant",
|
|
content="hi there",
|
|
token_count=10,
|
|
),
|
|
]
|
|
|
|
|
|
def _tool_heavy_transcript() -> list[TranscriptEntry]:
|
|
line = "drwxr-xr-x staff 4096 synthetic/file.txt "
|
|
result_text = (line * 50)[:2000]
|
|
entries: list[TranscriptEntry] = []
|
|
for turn in range(10):
|
|
tool_calls: list[dict[str, Any]] = []
|
|
for pair in range(4):
|
|
tool_id = f"tool-{turn}-{pair}"
|
|
tool_calls.append(
|
|
{
|
|
"type": "tool_use",
|
|
"tool_use_id": tool_id,
|
|
"name": "exec_shell",
|
|
"input": {"command": f"ls batch_{turn}/{pair}"},
|
|
}
|
|
)
|
|
tool_calls.append(
|
|
{
|
|
"type": "tool_result",
|
|
"tool_use_id": tool_id,
|
|
"name": "exec_shell",
|
|
"result": result_text,
|
|
"is_error": False,
|
|
}
|
|
)
|
|
entries.append(
|
|
TranscriptEntry(
|
|
session_id="s1",
|
|
session_key="agent:main:webchat:default",
|
|
role="assistant",
|
|
content=f"inspected batch {turn}",
|
|
token_count=120,
|
|
tool_calls=tool_calls,
|
|
)
|
|
)
|
|
return entries
|
|
|
|
|
|
def _make_turn(
|
|
routed_tier: str = "c3",
|
|
previous_tier: str | None = "c2",
|
|
base_tier: str | None = None,
|
|
final_tier: str | None = None,
|
|
routing_applied: bool = True,
|
|
) -> TurnContext:
|
|
routing_extra: dict[str, Any] = {}
|
|
if previous_tier is not None:
|
|
routing_extra["previous_tier"] = previous_tier
|
|
if base_tier is not None:
|
|
routing_extra["base_tier"] = base_tier
|
|
if final_tier is not None:
|
|
routing_extra["final_tier"] = final_tier
|
|
|
|
return TurnContext(
|
|
message="test",
|
|
session_key="agent:main:webchat:default",
|
|
config=None,
|
|
provider=None,
|
|
model="anthropic/claude-opus-4.8",
|
|
tool_defs=[],
|
|
system_prompt="you are helpful",
|
|
metadata={
|
|
"routed_tier": routed_tier,
|
|
"routing_applied": routing_applied,
|
|
"routing_extra": routing_extra,
|
|
},
|
|
)
|
|
|
|
|
|
def _make_runner(
|
|
session_manager: Any = None,
|
|
flush_service: Any = None,
|
|
enabled: bool = True,
|
|
*,
|
|
flush_enabled: bool = True,
|
|
flush_timeout_seconds: float = 15.0,
|
|
flush_background_timeout_seconds: float = 120.0,
|
|
flush_pre_compaction: bool = True,
|
|
flush_compaction_requires_safe_receipt: bool = False,
|
|
compaction_config: Any = None,
|
|
) -> TurnRunner:
|
|
config = SimpleNamespace(
|
|
squilla_router=SimpleNamespace(upgrade_to_c3_compaction_enabled=enabled),
|
|
compaction=compaction_config,
|
|
memory=SimpleNamespace(
|
|
flush_enabled=flush_enabled,
|
|
flush_pre_compaction=flush_pre_compaction,
|
|
flush_timeout_seconds=flush_timeout_seconds,
|
|
flush_background_timeout_seconds=flush_background_timeout_seconds,
|
|
flush_compaction_requires_safe_receipt=flush_compaction_requires_safe_receipt,
|
|
),
|
|
)
|
|
return TurnRunner(
|
|
provider_selector=SimpleNamespace(clone=lambda: SimpleNamespace()),
|
|
session_manager=session_manager,
|
|
config=config,
|
|
session_flush_service=flush_service,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_t2_to_t3_triggers_flush_then_compact() -> None:
|
|
sm = _FakeSessionManager(_sample_transcript())
|
|
fs = _FakeFlushService()
|
|
runner = _make_runner(session_manager=sm, flush_service=fs)
|
|
|
|
turn = _make_turn(routed_tier="c3", previous_tier="c2")
|
|
result = await runner._maybe_compact_on_t3_upgrade("agent:main:webchat:default", turn, 100_000)
|
|
|
|
assert result == "handled"
|
|
await asyncio.sleep(0)
|
|
assert len(fs.execute_calls) == 1
|
|
assert len(sm.compact_calls) == 1
|
|
assert sm.compact_calls[0] == ("agent:main:webchat:default", 100_000)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_t3_passes_profile_config_without_provider_or_model() -> None:
|
|
sm = _ResultCompactionSessionManager(_sample_transcript())
|
|
fs = _FakeFlushService()
|
|
runner = _make_runner(
|
|
session_manager=sm,
|
|
flush_service=fs,
|
|
compaction_config=SimpleNamespace(
|
|
enabled=False,
|
|
compaction_profile="coding",
|
|
protected_recent_messages=5,
|
|
timeout_seconds=11.0,
|
|
),
|
|
)
|
|
|
|
turn = _make_turn(routed_tier="t3", previous_tier="t2")
|
|
result = await runner._maybe_compact_on_t3_upgrade(
|
|
"agent:main:webchat:default",
|
|
turn,
|
|
100_000,
|
|
)
|
|
|
|
assert result == "handled"
|
|
config = sm.compact_configs[0]
|
|
assert isinstance(config, CompactionConfig)
|
|
assert config.model is None
|
|
assert config.api_key == ""
|
|
assert config.compaction_profile == "coding"
|
|
assert config.protected_recent_messages == 5
|
|
assert config.timeout_seconds == 11.0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_t3_within_budget_skips_flush_and_compact() -> None:
|
|
sm = _FakeSessionManager(_within_budget_transcript())
|
|
fs = _FakeFlushService()
|
|
runner = _make_runner(session_manager=sm, flush_service=fs)
|
|
|
|
turn = _make_turn(routed_tier="c3", previous_tier="c2")
|
|
result = await runner._maybe_compact_on_t3_upgrade("agent:main:webchat:default", turn, 100_000)
|
|
|
|
assert result == "handled"
|
|
await asyncio.sleep(0)
|
|
assert fs.execute_calls == []
|
|
assert sm.compact_calls == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_t3_budget_check_counts_full_tool_call_replay() -> None:
|
|
from opensquilla.session.compaction import (
|
|
estimate_entry_model_replay_tokens,
|
|
estimate_entry_replay_tokens,
|
|
)
|
|
|
|
transcript = _tool_heavy_transcript()
|
|
window = 30_000
|
|
summarized = sum(estimate_entry_replay_tokens(e) for e in transcript)
|
|
model_replay = sum(estimate_entry_model_replay_tokens(e) for e in transcript)
|
|
# The summarized estimate looks within budget while the model replay overflows.
|
|
assert summarized * 1.2 <= window < model_replay * 1.2
|
|
|
|
sm = _FakeSessionManager(transcript)
|
|
fs = _FakeFlushService()
|
|
runner = _make_runner(session_manager=sm, flush_service=fs)
|
|
|
|
turn = _make_turn(routed_tier="c3", previous_tier="c2")
|
|
result = await runner._maybe_compact_on_t3_upgrade("agent:main:webchat:default", turn, window)
|
|
|
|
assert result == "handled"
|
|
await asyncio.sleep(0)
|
|
assert sm.compact_calls == [("agent:main:webchat:default", window)]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_t3_completed_event_reports_compaction_metadata(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
sm = _ResultCompactionSessionManager(_sample_transcript())
|
|
fs = _FakeFlushService()
|
|
events: list[tuple[str, dict[str, Any]]] = []
|
|
monkeypatch.setattr(
|
|
runtime_module,
|
|
"notify_compaction",
|
|
lambda session_key, **payload: events.append((session_key, payload)),
|
|
)
|
|
runner = _make_runner(session_manager=sm, flush_service=fs)
|
|
|
|
turn = _make_turn(routed_tier="c3", previous_tier="c2")
|
|
result = await runner._maybe_compact_on_t3_upgrade("agent:main:webchat:default", turn, 100_000)
|
|
|
|
assert result == "handled"
|
|
assert [(key, payload["status"]) for key, payload in events] == [
|
|
("agent:main:webchat:default", "started"),
|
|
("agent:main:webchat:default", "observed"),
|
|
("agent:main:webchat:default", "observed"),
|
|
("agent:main:webchat:default", "completed"),
|
|
]
|
|
compaction_ids = {payload.get("compaction_id") for _, payload in events}
|
|
assert len(compaction_ids) == 1
|
|
assert None not in compaction_ids
|
|
assert events[0][1]["event"] == "compaction.triggered"
|
|
assert events[1][1]["event"] == "compaction.chunk_summarized"
|
|
assert events[2][1]["event"] == "compaction.summary_verified"
|
|
completed = events[-1][1]
|
|
assert completed["applied"] is True
|
|
assert completed["durability"] == "durable"
|
|
assert completed["user_visible"] is True
|
|
assert completed["event"] == "compaction.persisted"
|
|
assert completed["event_chain"] == [
|
|
"compaction.triggered",
|
|
"compaction.chunk_summarized",
|
|
"compaction.summary_verified",
|
|
"compaction.persisted",
|
|
]
|
|
assert completed["coverage_status"] == "unknown"
|
|
assert completed["chunk_count"] == 1
|
|
assert completed["summary_source"] == "llm"
|
|
assert completed["removed_count"] == 2
|
|
assert completed["kept_count"] == 1
|
|
assert completed["tokens_after"] == 100
|
|
assert completed["remaining_budget_tokens"] == 99_900
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_t3_stale_preimage_skip_does_not_mark_compacted(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
sm = _StaleResultCompactionSessionManager(_sample_transcript())
|
|
fs = _FakeFlushService()
|
|
events: list[tuple[str, dict[str, Any]]] = []
|
|
monkeypatch.setattr(
|
|
runtime_module,
|
|
"notify_compaction",
|
|
lambda session_key, **payload: events.append((session_key, payload)),
|
|
)
|
|
runner = _make_runner(session_manager=sm, flush_service=fs)
|
|
session_key = "agent:main:webchat:default"
|
|
|
|
turn = _make_turn(routed_tier="c3", previous_tier="c2")
|
|
result = await runner._maybe_compact_on_t3_upgrade(session_key, turn, 100_000)
|
|
|
|
assert result == "handled"
|
|
assert sm.compact_calls == [(session_key, 100_000)]
|
|
assert runner.has_compacted_this_turn(session_key) is False
|
|
skipped = [payload for _, payload in events if payload.get("status") == "skipped"]
|
|
assert skipped[-1]["reason"] == "stale_preimage"
|
|
assert skipped[-1]["applied"] is False
|
|
assert skipped[-1]["durability"] == "none"
|
|
assert skipped[-1]["user_visible"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_t0_t1_to_t3_triggers() -> None:
|
|
for prev in ("c0", "c1"):
|
|
sm = _FakeSessionManager(_sample_transcript())
|
|
fs = _FakeFlushService()
|
|
runner = _make_runner(session_manager=sm, flush_service=fs)
|
|
|
|
turn = _make_turn(routed_tier="c3", previous_tier=prev)
|
|
result = await runner._maybe_compact_on_t3_upgrade(
|
|
"agent:main:webchat:default", turn, 100_000
|
|
)
|
|
|
|
assert result == "handled", f"failed for previous_tier={prev}"
|
|
await asyncio.sleep(0)
|
|
assert len(fs.execute_calls) == 1
|
|
assert len(sm.compact_calls) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_t3_to_t3_skips() -> None:
|
|
sm = _FakeSessionManager(_sample_transcript())
|
|
fs = _FakeFlushService()
|
|
runner = _make_runner(session_manager=sm, flush_service=fs)
|
|
|
|
turn = _make_turn(routed_tier="c3", previous_tier="c3")
|
|
result = await runner._maybe_compact_on_t3_upgrade("agent:main:webchat:default", turn, 100_000)
|
|
|
|
assert result == "not_applicable"
|
|
assert len(fs.execute_calls) == 0
|
|
assert len(sm.compact_calls) == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_non_t3_route_skips() -> None:
|
|
sm = _FakeSessionManager(_sample_transcript())
|
|
fs = _FakeFlushService()
|
|
runner = _make_runner(session_manager=sm, flush_service=fs)
|
|
|
|
turn = _make_turn(routed_tier="c1", previous_tier="c0")
|
|
result = await runner._maybe_compact_on_t3_upgrade("agent:main:webchat:default", turn, 100_000)
|
|
|
|
assert result == "not_applicable"
|
|
assert len(fs.execute_calls) == 0
|
|
assert len(sm.compact_calls) == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_config_disabled_skips() -> None:
|
|
sm = _FakeSessionManager(_sample_transcript())
|
|
fs = _FakeFlushService()
|
|
runner = _make_runner(session_manager=sm, flush_service=fs, enabled=False)
|
|
|
|
turn = _make_turn(routed_tier="c3", previous_tier="c2")
|
|
result = await runner._maybe_compact_on_t3_upgrade("agent:main:webchat:default", turn, 100_000)
|
|
|
|
assert result == "not_applicable"
|
|
assert len(fs.execute_calls) == 0
|
|
assert len(sm.compact_calls) == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_observe_mode_skips() -> None:
|
|
sm = _FakeSessionManager(_sample_transcript())
|
|
fs = _FakeFlushService()
|
|
runner = _make_runner(session_manager=sm, flush_service=fs)
|
|
|
|
turn = _make_turn(routed_tier="c3", previous_tier="c2", routing_applied=False)
|
|
result = await runner._maybe_compact_on_t3_upgrade("agent:main:webchat:default", turn, 100_000)
|
|
|
|
assert result == "not_applicable"
|
|
assert len(fs.execute_calls) == 0
|
|
assert len(sm.compact_calls) == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_flush_raises_does_not_block_compaction() -> None:
|
|
sm = _FakeSessionManager(_sample_transcript())
|
|
fs = _FakeFlushService(raise_exc=RuntimeError("flush boom"))
|
|
runner = _make_runner(session_manager=sm, flush_service=fs)
|
|
|
|
turn = _make_turn(routed_tier="c3", previous_tier="c2")
|
|
result = await runner._maybe_compact_on_t3_upgrade("agent:main:webchat:default", turn, 100_000)
|
|
|
|
assert result == "handled"
|
|
await asyncio.sleep(0)
|
|
assert len(fs.execute_calls) == 1
|
|
assert len(sm.compact_calls) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_flush_error_receipt_does_not_block_compaction() -> None:
|
|
sm = _FakeSessionManager(_sample_transcript())
|
|
fs = _FakeFlushService(receipt=_FakeFlushReceipt(mode="error", error="provider down"))
|
|
runner = _make_runner(session_manager=sm, flush_service=fs)
|
|
|
|
turn = _make_turn(routed_tier="c3", previous_tier="c2")
|
|
result = await runner._maybe_compact_on_t3_upgrade("agent:main:webchat:default", turn, 100_000)
|
|
|
|
assert result == "handled"
|
|
await asyncio.sleep(0)
|
|
assert len(fs.execute_calls) == 1
|
|
assert len(sm.compact_calls) == 1
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"receipt",
|
|
[
|
|
_FakeFlushReceipt(mode="raw", raw_reason="no_provider"),
|
|
_FakeFlushReceipt(integrity_status="missing_chunks"),
|
|
_FakeFlushReceipt(output_coverage_status="coverage_warning"),
|
|
_FakeFlushReceipt(invalid_candidate_count=1),
|
|
_FakeFlushReceipt(candidate_missing_ids=["candidate-1"]),
|
|
_FakeFlushReceipt(obligation_missing_ids=["obligation-1"]),
|
|
],
|
|
)
|
|
@pytest.mark.asyncio
|
|
async def test_degraded_flush_receipts_do_not_block_compaction(
|
|
receipt: _FakeFlushReceipt,
|
|
) -> None:
|
|
sm = _FakeSessionManager(_sample_transcript())
|
|
fs = _FakeFlushService(receipt=receipt)
|
|
runner = _make_runner(session_manager=sm, flush_service=fs)
|
|
|
|
turn = _make_turn(routed_tier="c3", previous_tier="c2")
|
|
result = await runner._maybe_compact_on_t3_upgrade("agent:main:webchat:default", turn, 100_000)
|
|
|
|
assert result == "handled"
|
|
await asyncio.sleep(0)
|
|
assert len(fs.execute_calls) == 1
|
|
assert len(sm.compact_calls) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_t3_strict_flush_receipt_skips_destructive_compaction() -> None:
|
|
sm = _FakeSessionManager(_sample_transcript())
|
|
fs = _FakeFlushService(receipt=_FakeFlushReceipt(integrity_status="missing_chunks"))
|
|
runner = _make_runner(
|
|
session_manager=sm,
|
|
flush_service=fs,
|
|
flush_compaction_requires_safe_receipt=True,
|
|
)
|
|
|
|
turn = _make_turn(routed_tier="c3", previous_tier="c2")
|
|
result = await runner._maybe_compact_on_t3_upgrade("agent:main:webchat:default", turn, 100_000)
|
|
|
|
assert result == "handled"
|
|
assert len(fs.execute_calls) == 1
|
|
assert sm.compact_calls == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_backfilled_flush_receipt_allows_compact() -> None:
|
|
sm = _FakeSessionManager(_sample_transcript())
|
|
fs = _FakeFlushService(receipt=_FakeFlushReceipt(obligation_status="backfilled"))
|
|
runner = _make_runner(session_manager=sm, flush_service=fs)
|
|
|
|
turn = _make_turn(routed_tier="c3", previous_tier="c2")
|
|
result = await runner._maybe_compact_on_t3_upgrade("agent:main:webchat:default", turn, 100_000)
|
|
|
|
assert result == "handled"
|
|
await asyncio.sleep(0)
|
|
assert len(fs.execute_calls) == 1
|
|
assert len(sm.compact_calls) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_t3_flush_uses_background_timeout_for_service_call() -> None:
|
|
sm = _FakeSessionManager(_sample_transcript())
|
|
fs = _FakeFlushService()
|
|
runner = _make_runner(
|
|
session_manager=sm,
|
|
flush_service=fs,
|
|
flush_timeout_seconds=0.25,
|
|
flush_background_timeout_seconds=42.0,
|
|
)
|
|
|
|
turn = _make_turn(routed_tier="c3", previous_tier="c2")
|
|
result = await runner._maybe_compact_on_t3_upgrade("agent:main:webchat:default", turn, 100_000)
|
|
|
|
assert result == "handled"
|
|
await asyncio.sleep(0)
|
|
assert fs.execute_calls[0]["timeout"] == 42.0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_t3_flush_uses_longer_default_background_timeout() -> None:
|
|
sm = _FakeSessionManager(_sample_transcript())
|
|
fs = _FakeFlushService()
|
|
runner = _make_runner(session_manager=sm, flush_service=fs)
|
|
|
|
turn = _make_turn(routed_tier="c3", previous_tier="c2")
|
|
result = await runner._maybe_compact_on_t3_upgrade("agent:main:webchat:default", turn, 100_000)
|
|
|
|
assert result == "handled"
|
|
await asyncio.sleep(0)
|
|
assert fs.execute_calls[0]["timeout"] == 120.0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_t3_flush_grace_timeout_does_not_block_compaction() -> None:
|
|
sm = _FakeSessionManager(_sample_transcript())
|
|
fs = _FakeFlushService(delay_seconds=0.05)
|
|
runner = _make_runner(
|
|
session_manager=sm,
|
|
flush_service=fs,
|
|
flush_timeout_seconds=0.001,
|
|
flush_background_timeout_seconds=42.0,
|
|
)
|
|
|
|
turn = _make_turn(routed_tier="c3", previous_tier="c2")
|
|
result = await runner._maybe_compact_on_t3_upgrade("agent:main:webchat:default", turn, 100_000)
|
|
|
|
assert result == "handled"
|
|
await asyncio.sleep(0)
|
|
assert fs.execute_calls[0]["timeout"] == 42.0
|
|
assert len(sm.compact_calls) == 1
|
|
await asyncio.sleep(0.06)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_memory_flush_disabled_compacts_without_flush_service() -> None:
|
|
sm = _FakeSessionManager(_sample_transcript())
|
|
runner = _make_runner(
|
|
session_manager=sm,
|
|
flush_service=None,
|
|
flush_enabled=False,
|
|
)
|
|
|
|
turn = _make_turn(routed_tier="c3", previous_tier="c2")
|
|
result = await runner._maybe_compact_on_t3_upgrade("agent:main:webchat:default", turn, 100_000)
|
|
|
|
assert result == "handled"
|
|
assert len(sm.compact_calls) == 1
|
|
|
|
|
|
@pytest.mark.parametrize("value", ["0", "false", "no", "off"])
|
|
@pytest.mark.asyncio
|
|
async def test_env_flush_disabled_compacts_without_flush_service(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
value: str,
|
|
) -> None:
|
|
monkeypatch.setenv("OPENSQUILLA_SESSION_FLUSH", value)
|
|
sm = _FakeSessionManager(_sample_transcript())
|
|
runner = _make_runner(session_manager=sm, flush_service=None)
|
|
|
|
turn = _make_turn(routed_tier="c3", previous_tier="c2")
|
|
result = await runner._maybe_compact_on_t3_upgrade("agent:main:webchat:default", turn, 100_000)
|
|
|
|
assert result == "handled"
|
|
assert len(sm.compact_calls) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_compact_raises_continues() -> None:
|
|
sm = _FakeSessionManager(_sample_transcript())
|
|
|
|
async def _boom(session_key: str, context_window_tokens: int, **kw: Any) -> str:
|
|
raise RuntimeError("compact boom")
|
|
|
|
sm.compact = _boom # type: ignore[assignment]
|
|
fs = _FakeFlushService()
|
|
runner = _make_runner(session_manager=sm, flush_service=fs)
|
|
|
|
turn = _make_turn(routed_tier="c3", previous_tier="c2")
|
|
result = await runner._maybe_compact_on_t3_upgrade("agent:main:webchat:default", turn, 100_000)
|
|
|
|
assert result == "compact_failed"
|
|
await asyncio.sleep(0)
|
|
assert len(fs.execute_calls) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_t3_compact_failure_uses_emergency_ephemeral_history_trim(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
session_key = "agent:main:webchat:t3-emergency"
|
|
transcript = [
|
|
TranscriptEntry(
|
|
session_id="s1",
|
|
session_key=session_key,
|
|
role="user" if index % 2 == 0 else "assistant",
|
|
content=f"historic t3 message {index} " + ("x" * 500),
|
|
token_count=300,
|
|
)
|
|
for index in range(8)
|
|
]
|
|
sm = _FakeSessionManager(transcript)
|
|
|
|
async def _boom(session_key: str, context_window_tokens: int, **kw: Any) -> str:
|
|
raise RuntimeError("compact boom")
|
|
|
|
sm.compact = _boom # type: ignore[assignment]
|
|
events: list[tuple[str, dict[str, Any]]] = []
|
|
monkeypatch.setattr(
|
|
runtime_module,
|
|
"notify_compaction",
|
|
lambda session_key, **payload: events.append((session_key, payload)),
|
|
)
|
|
runner = _make_runner(session_manager=sm, flush_service=_FakeFlushService())
|
|
|
|
result = await runner._maybe_compact_on_t3_upgrade(
|
|
session_key,
|
|
_make_turn(routed_tier="c3", previous_tier="c2"),
|
|
1000,
|
|
)
|
|
|
|
class _HistoryCapture:
|
|
provider = SimpleNamespace(provider_name="test")
|
|
|
|
def __init__(self) -> None:
|
|
self.history: list[Any] = []
|
|
|
|
def set_history(self, history: list[Any]) -> None:
|
|
self.history = history
|
|
|
|
agent = _HistoryCapture()
|
|
summary_context = await runner._load_history(agent, session_key, trim_last_user=False)
|
|
|
|
assert result == "compact_failed"
|
|
assert len(await sm.get_transcript(session_key)) == len(transcript)
|
|
assert 0 < len(agent.history) < len(transcript)
|
|
assert summary_context is not None
|
|
assert "emergency request-scoped compaction" in summary_context.lower()
|
|
statuses = [payload["status"] for _, payload in events]
|
|
assert statuses[:2] == ["started", "emergency_ephemeral"]
|
|
assert "failed" not in statuses
|
|
emergency = next(payload for _, payload in events if payload["status"] == "emergency_ephemeral")
|
|
assert emergency["durability"] == "request_scoped"
|
|
assert runner._compaction_failures[session_key].count == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_t3_open_circuit_still_uses_request_scoped_emergency_trim(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
session_key = "agent:main:webchat:t3-open-circuit"
|
|
transcript = [
|
|
TranscriptEntry(
|
|
session_id="s1",
|
|
session_key=session_key,
|
|
role="user" if index % 2 == 0 else "assistant",
|
|
content=f"historic t3 message {index} " + ("x" * 500),
|
|
token_count=300,
|
|
)
|
|
for index in range(8)
|
|
]
|
|
sm = _FakeSessionManager(transcript)
|
|
events: list[tuple[str, dict[str, Any]]] = []
|
|
monkeypatch.setattr(
|
|
runtime_module,
|
|
"notify_compaction",
|
|
lambda session_key, **payload: events.append((session_key, payload)),
|
|
)
|
|
runner = _make_runner(session_manager=sm, flush_service=_FakeFlushService())
|
|
runner._compaction_failures[session_key] = runtime_module._CompactionFailureState(
|
|
count=3,
|
|
opened_at=runtime_module.time.monotonic(),
|
|
)
|
|
|
|
result = await runner._maybe_compact_on_t3_upgrade(
|
|
session_key,
|
|
_make_turn(routed_tier="c3", previous_tier="c2"),
|
|
1000,
|
|
)
|
|
|
|
assert result == "handled"
|
|
assert sm.compact_calls == []
|
|
assert [payload["status"] for _, payload in events] == ["emergency_ephemeral"]
|
|
emergency = events[-1][1]
|
|
assert emergency["reason"] == "durable_compaction_circuit_open"
|
|
assert emergency["durability"] == "request_scoped"
|
|
assert runner._compaction_failures[session_key].count == 3
|