2361 lines
91 KiB
Python
2361 lines
91 KiB
Python
"""Regression: prevent transcript fork when two paths compress the same session_id.
|
|
|
|
Damien's incident (Discord, 2026-05-28): a long Hermes session in a Discord
|
|
gateway hit the compression threshold at the end of a turn. The parent agent
|
|
finished delivering the response and ``conversation_loop.py`` fired
|
|
``_spawn_background_review(...)`` — which builds a forked ``AIAgent`` that
|
|
inherits ``agent.session_id`` (see ``agent/background_review.py``::
|
|
``review_agent.session_id = agent.session_id``). Roughly two seconds later
|
|
a synthetic ``Background process proc_… completed`` event arrived and
|
|
started a fresh turn on the same parent ``session_id`` (still cached in the
|
|
gateway's ``SessionEntry``). Both paths hit preflight compression on the
|
|
same parent transcript and called ``_compress_context`` concurrently. Each
|
|
ended the parent and created its own CHILD session in ``state.db``, both
|
|
parented to the same old id. The gateway's ``SessionEntry`` only caught one
|
|
rotation; the other child became an orphan that silently accumulated writes.
|
|
|
|
Repro shape on Damien's machine:
|
|
|
|
parent 20260527_234659_e65f0e ended_at=set end_reason='compression'
|
|
child 20260528_113619_fc80e1 parent=20260527_234659_e65f0e (in SessionEntry)
|
|
child <orphan> parent=20260527_234659_e65f0e (silent writes)
|
|
|
|
This regression simulates the two concurrent ``compress_context`` calls
|
|
against a shared ``state.db`` and asserts that the per-session compression
|
|
lock added in this PR prevents the orphan child. Without the lock the
|
|
fixture deterministically produces 2 children; with the lock, exactly 1.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import inspect
|
|
import json
|
|
import logging
|
|
import os
|
|
import sqlite3
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from hermes_state import SessionDB
|
|
|
|
|
|
def _build_agent_with_db(
|
|
db: SessionDB,
|
|
session_id: str,
|
|
*,
|
|
stub_compressor: bool = True,
|
|
):
|
|
"""Build an AIAgent that's wired to ``db`` and pinned to ``session_id``."""
|
|
with patch.dict(os.environ, {"OPENROUTER_API_KEY": "test-key"}):
|
|
from run_agent import AIAgent
|
|
|
|
agent = AIAgent(
|
|
api_key="test-key",
|
|
base_url="https://openrouter.ai/api/v1",
|
|
model="test/model",
|
|
quiet_mode=True,
|
|
session_db=db,
|
|
session_id=session_id,
|
|
skip_context_files=True,
|
|
skip_memory=True,
|
|
)
|
|
|
|
# Stub the compressor so it returns deterministic output and DOESN'T make
|
|
# an LLM call. Sleep inside compress() so the two threads' rotations
|
|
# actually overlap — without that the OS could happen to serialize them
|
|
# and hide the bug.
|
|
if not stub_compressor:
|
|
return agent
|
|
|
|
compressor = MagicMock()
|
|
|
|
def _compress_with_overlap(*_a, **_kw):
|
|
time.sleep(0.15)
|
|
return [
|
|
{"role": "user", "content": "[CONTEXT COMPACTION] summary"},
|
|
{"role": "user", "content": "tail"},
|
|
]
|
|
|
|
compressor.compress.side_effect = _compress_with_overlap
|
|
compressor.compression_count = 1
|
|
compressor.last_prompt_tokens = 0
|
|
compressor.last_completion_tokens = 0
|
|
compressor._last_summary_error = None
|
|
compressor._last_compress_aborted = False
|
|
compressor._last_aux_model_failure_model = None
|
|
compressor._last_aux_model_failure_error = None
|
|
agent.context_compressor = compressor
|
|
# The compressor is a stub — the one-time compression-model feasibility
|
|
# probe would resolve a REAL auxiliary provider (credential pools, live
|
|
# token exchanges over the network on some dev machines). That makes the
|
|
# first _compress_context call in a test nondeterministically slow (>2s)
|
|
# and flakes event-based timing assertions. Mark it done: these tests
|
|
# exercise locking/fencing/rotation, never aux-model feasibility.
|
|
agent._compression_feasibility_checked = True
|
|
# These tests cover the ROTATION fallback path (forking, child sessions,
|
|
# lock contention) — pin in_place=False so they keep exercising it
|
|
# regardless of the global default (which flipped to True in #38763).
|
|
agent.compression_in_place = False
|
|
return agent
|
|
|
|
|
|
def _count_children(db: SessionDB, parent_sid: str) -> int:
|
|
"""Count rows in state.db whose parent_session_id == parent_sid."""
|
|
rows = db._conn.execute(
|
|
"SELECT id FROM sessions WHERE parent_session_id = ?",
|
|
(parent_sid,),
|
|
).fetchall()
|
|
return len(rows)
|
|
|
|
|
|
def _live_child_id(db: SessionDB, parent_sid: str) -> str | None:
|
|
"""The single child id of ``parent_sid``, or None when there is none.
|
|
|
|
Fails loudly on more than one child: callers use this to prove the agents
|
|
converged on the winner's session, so a multi-child state is a fork and
|
|
must not be silently reduced to 'the first row'.
|
|
"""
|
|
rows = db._conn.execute(
|
|
"SELECT id FROM sessions WHERE parent_session_id = ?",
|
|
(parent_sid,),
|
|
).fetchall()
|
|
assert len(rows) <= 1, f"expected at most one child of {parent_sid}, got {rows!r}"
|
|
return rows[0][0] if rows else None
|
|
|
|
|
|
def _wait_for_touch(touch_calls: list[str], value: str, timeout: float = 1.0) -> None:
|
|
deadline = time.monotonic() + timeout
|
|
while time.monotonic() < deadline:
|
|
if value in touch_calls:
|
|
return
|
|
time.sleep(0.01)
|
|
pytest.fail(f"Timed out waiting for touch activity {value!r}; calls={touch_calls!r}")
|
|
|
|
|
|
def test_compression_activity_heartbeat_touches_agent_during_long_compress(tmp_path: Path) -> None:
|
|
"""Long compression must refresh agent activity so gateway watchdogs do not fire."""
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
session_id = "HEARTBEAT_TEST"
|
|
db.create_session(session_id, source="test")
|
|
|
|
agent = _build_agent_with_db(db, session_id)
|
|
agent._compression_activity_heartbeat_interval = 0.1
|
|
touch_calls: list[str] = []
|
|
agent._touch_activity = lambda desc, **_kw: touch_calls.append(desc)
|
|
|
|
def _slow_compress(*_a, **_kw):
|
|
_wait_for_touch(touch_calls, "context compression in progress")
|
|
return [
|
|
{"role": "user", "content": "[CONTEXT COMPACTION] summary"},
|
|
{"role": "user", "content": "tail"},
|
|
]
|
|
|
|
agent.context_compressor.compress.side_effect = _slow_compress
|
|
messages = [{"role": "user", "content": f"m{i}"} for i in range(20)]
|
|
|
|
agent._compress_context(messages, "sys", approx_tokens=120_000)
|
|
|
|
assert touch_calls[0] == "context compression started"
|
|
assert "context compression in progress" in touch_calls
|
|
assert touch_calls[-1] == "context compression completed"
|
|
assert db.get_compression_lock_holder(session_id) is None
|
|
|
|
|
|
def test_compression_activity_heartbeat_emits_client_status_events(tmp_path: Path) -> None:
|
|
"""The heartbeat must re-emit the compacting status, not just DB touches.
|
|
|
|
Remote transports (e.g. the Android relay app) run idle-progress turn
|
|
watchdogs that ``session.interrupt`` a turn after ~180s with no gateway
|
|
events. Compression is silent on the event stream, so without periodic
|
|
status heartbeats a long compression is killed mid-flight and retriggers
|
|
forever on sessions near the context ceiling.
|
|
"""
|
|
from agent.conversation_compression import (
|
|
COMPACTION_HEARTBEAT_STATUS,
|
|
is_compaction_progress_status,
|
|
)
|
|
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
session_id = "HEARTBEAT_STATUS_TEST"
|
|
db.create_session(session_id, source="test")
|
|
|
|
agent = _build_agent_with_db(db, session_id)
|
|
agent._compression_activity_heartbeat_interval = 0.1
|
|
touch_calls: list[str] = []
|
|
agent._touch_activity = lambda desc, **_kw: touch_calls.append(desc)
|
|
status_events: list[tuple[str, str]] = []
|
|
setattr(
|
|
agent,
|
|
"status_callback",
|
|
lambda event, message: status_events.append((event, message)),
|
|
)
|
|
|
|
def _slow_compress(*_a, **_kw):
|
|
_wait_for_touch(touch_calls, "context compression in progress")
|
|
return [
|
|
{"role": "user", "content": "[CONTEXT COMPACTION] summary"},
|
|
{"role": "user", "content": "tail"},
|
|
]
|
|
|
|
agent.context_compressor.compress.side_effect = _slow_compress
|
|
messages = [{"role": "user", "content": f"m{i}"} for i in range(20)]
|
|
|
|
agent._compress_context(messages, "sys", approx_tokens=120_000)
|
|
|
|
heartbeats = [e for e in status_events if e[1] == COMPACTION_HEARTBEAT_STATUS]
|
|
assert heartbeats, "no heartbeat status reached the client"
|
|
# Same "lifecycle" key as the other compaction statuses so the TUI gateway
|
|
# re-tags it to kind="compacting" and Telegram edits one bubble in place.
|
|
assert {event for event, _ in heartbeats} == {"lifecycle"}
|
|
assert is_compaction_progress_status(COMPACTION_HEARTBEAT_STATUS)
|
|
# Exactly one routine start line precedes the first heartbeat; the
|
|
# heartbeat no longer re-emits a start of its own (adapters without
|
|
# send_or_update_status would otherwise post two messages).
|
|
assert status_events[0][1] != COMPACTION_HEARTBEAT_STATUS
|
|
# Every heartbeat is a periodic tick: none may precede the first
|
|
# "in progress" DB touch, which is what start() would have produced.
|
|
first_tick_touch = touch_calls.index("context compression in progress")
|
|
assert first_tick_touch >= 1 # "started" touch came first
|
|
assert len(heartbeats) <= touch_calls.count("context compression in progress")
|
|
|
|
|
|
def test_compression_heartbeat_is_silent_for_quiet_context_engines(tmp_path: Path) -> None:
|
|
"""A context engine that suppresses the routine start status opens no
|
|
visible compaction phase; the heartbeat must not open one either (there
|
|
would be no terminal edge to close it)."""
|
|
from agent.conversation_compression import COMPACTION_HEARTBEAT_STATUS
|
|
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
session_id = "HEARTBEAT_QUIET_TEST"
|
|
db.create_session(session_id, source="test")
|
|
|
|
agent = _build_agent_with_db(db, session_id)
|
|
agent._compression_activity_heartbeat_interval = 0.1
|
|
touch_calls: list[str] = []
|
|
agent._touch_activity = lambda desc, **_kw: touch_calls.append(desc)
|
|
status_events: list[tuple[str, str]] = []
|
|
setattr(
|
|
agent,
|
|
"status_callback",
|
|
lambda event, message: status_events.append((event, message)),
|
|
)
|
|
|
|
def _slow_compress(*_a, **_kw):
|
|
_wait_for_touch(touch_calls, "context compression in progress")
|
|
return [
|
|
{"role": "user", "content": "[CONTEXT COMPACTION] summary"},
|
|
{"role": "user", "content": "tail"},
|
|
]
|
|
|
|
agent.context_compressor.compress.side_effect = _slow_compress
|
|
messages = [{"role": "user", "content": f"m{i}"} for i in range(20)]
|
|
|
|
with patch(
|
|
"agent.conversation_compression.automatic_compaction_status_message",
|
|
return_value="",
|
|
):
|
|
agent._compress_context(messages, "sys", approx_tokens=120_000)
|
|
|
|
assert all(m != COMPACTION_HEARTBEAT_STATUS for _, m in status_events)
|
|
assert "context compression in progress" in touch_calls # DB touches still ran
|
|
|
|
|
|
def test_lock_contender_preserves_terminal_compaction_lifecycle(tmp_path: Path) -> None:
|
|
"""A lock loser still closes the structured compaction lifecycle.
|
|
|
|
The gateway independently filters this routine notice for chat surfaces
|
|
unless ``compression.progress_notices`` is enabled. The low-level event
|
|
must remain available so the desktop can retire its compaction phase.
|
|
"""
|
|
from agent.conversation_compression import COMPACTION_DONE_STATUS
|
|
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
session_id = "LOCK_CONTENDER_STATUS_TEST"
|
|
db.create_session(session_id, source="discord")
|
|
assert db.try_acquire_compression_lock(session_id, "winner", ttl_seconds=60)
|
|
|
|
agent = _build_agent_with_db(db, session_id)
|
|
status_events: list[tuple[str, str]] = []
|
|
setattr(
|
|
agent,
|
|
"status_callback",
|
|
lambda event, message: status_events.append((event, message)),
|
|
)
|
|
messages = [{"role": "user", "content": f"m{i}"} for i in range(20)]
|
|
|
|
returned, _system_prompt = agent._compress_context(
|
|
messages,
|
|
"sys",
|
|
approx_tokens=120_000,
|
|
)
|
|
|
|
assert returned is messages
|
|
assert getattr(agent, "_compression_skipped_due_to_lock", None) == "winner"
|
|
assert status_events.count(("compacted", COMPACTION_DONE_STATUS)) == 1
|
|
|
|
|
|
def test_failed_session_split_does_not_announce_compaction_complete(tmp_path: Path) -> None:
|
|
"""A failed durable split must not emit a successful completion edge."""
|
|
from agent.conversation_compression import COMPACTION_DONE_STATUS
|
|
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
session_id = "FAILED_SPLIT_STATUS_TEST"
|
|
db.create_session(session_id, source="discord")
|
|
agent = _build_agent_with_db(db, session_id)
|
|
setattr(agent, "compression_in_place", False)
|
|
db.publish_compression_child = MagicMock(side_effect=RuntimeError("split boom"))
|
|
status_events: list[tuple[str, str]] = []
|
|
setattr(
|
|
agent,
|
|
"status_callback",
|
|
lambda event, message: status_events.append((event, message)),
|
|
)
|
|
messages = [{"role": "user", "content": f"m{i}"} for i in range(20)]
|
|
|
|
agent._compress_context(
|
|
messages,
|
|
"sys",
|
|
approx_tokens=120_000,
|
|
force=True,
|
|
)
|
|
|
|
db.publish_compression_child.assert_called_once()
|
|
assert ("compacted", COMPACTION_DONE_STATUS) not in status_events
|
|
assert db.get_compression_lock_holder(session_id) is None
|
|
|
|
|
|
def test_failed_in_place_split_does_not_announce_compaction_complete(tmp_path: Path) -> None:
|
|
"""An in-place persistence failure must not emit a completion edge."""
|
|
from agent.conversation_compression import COMPACTION_DONE_STATUS
|
|
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
session_id = "FAILED_IN_PLACE_STATUS_TEST"
|
|
db.create_session(session_id, source="discord")
|
|
agent = _build_agent_with_db(db, session_id)
|
|
setattr(agent, "compression_in_place", True)
|
|
db.archive_and_compact = MagicMock(side_effect=RuntimeError("archive boom"))
|
|
status_events: list[tuple[str, str]] = []
|
|
setattr(
|
|
agent,
|
|
"status_callback",
|
|
lambda event, message: status_events.append((event, message)),
|
|
)
|
|
messages = [{"role": "user", "content": f"m{i}"} for i in range(20)]
|
|
|
|
agent._compress_context(
|
|
messages,
|
|
"sys",
|
|
approx_tokens=120_000,
|
|
force=True,
|
|
)
|
|
|
|
db.archive_and_compact.assert_called_once()
|
|
assert ("compacted", COMPACTION_DONE_STATUS) not in status_events
|
|
assert getattr(agent, "session_id", None) == session_id
|
|
assert db.get_compression_lock_holder(session_id) is None
|
|
|
|
|
|
def test_compression_activity_heartbeat_stops_on_compress_exception(tmp_path: Path) -> None:
|
|
"""Exception paths must stop the heartbeat and release the compression lock."""
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
session_id = "HEARTBEAT_FAIL_TEST"
|
|
db.create_session(session_id, source="test")
|
|
|
|
agent = _build_agent_with_db(db, session_id)
|
|
agent._compression_activity_heartbeat_interval = 0.1
|
|
touch_calls: list[str] = []
|
|
agent._touch_activity = lambda desc, **_kw: touch_calls.append(desc)
|
|
|
|
def _failing_compress(*_a, **_kw):
|
|
_wait_for_touch(touch_calls, "context compression in progress")
|
|
raise RuntimeError("compress boom")
|
|
|
|
agent.context_compressor.compress.side_effect = _failing_compress
|
|
messages = [{"role": "user", "content": f"m{i}"} for i in range(20)]
|
|
|
|
with pytest.raises(RuntimeError, match="compress boom"):
|
|
agent._compress_context(messages, "sys", approx_tokens=120_000)
|
|
|
|
assert touch_calls[0] == "context compression started"
|
|
assert "context compression in progress" in touch_calls
|
|
assert touch_calls[-1] == "context compression failed"
|
|
assert db.get_compression_lock_holder(session_id) is None
|
|
|
|
|
|
def test_compression_activity_heartbeat_ignores_touch_errors(tmp_path: Path) -> None:
|
|
"""Activity touch failures must not affect compression success semantics."""
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
session_id = "HEARTBEAT_TOUCH_ERROR_TEST"
|
|
db.create_session(session_id, source="test")
|
|
|
|
agent = _build_agent_with_db(db, session_id)
|
|
agent._compression_activity_heartbeat_interval = 0.1
|
|
agent._touch_activity = lambda _desc, **_kw: (_ for _ in ()).throw(RuntimeError("touch boom"))
|
|
messages = [{"role": "user", "content": f"m{i}"} for i in range(20)]
|
|
|
|
compressed, _sp = agent._compress_context(messages, "sys", approx_tokens=120_000)
|
|
|
|
assert compressed[0]["content"] == "[CONTEXT COMPACTION] summary"
|
|
assert db.get_compression_lock_holder(session_id) is None
|
|
|
|
|
|
def test_compression_activity_heartbeat_strict_signature_fallback_releases_lock(tmp_path: Path) -> None:
|
|
"""Strict compressor signatures still compress while heartbeat cleanup runs.
|
|
|
|
Main inspects the engine signature up front (_supported_compression_kwargs)
|
|
instead of catching TypeError, so a strict-signature engine is invoked
|
|
exactly once with only the kwargs it accepts. The heartbeat (with a
|
|
non-numeric configured interval falling back to the default) must still
|
|
wrap the call and stop cleanly.
|
|
"""
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
session_id = "HEARTBEAT_TYPEERROR_TEST"
|
|
db.create_session(session_id, source="test")
|
|
|
|
agent = _build_agent_with_db(db, session_id)
|
|
agent._compression_activity_heartbeat_interval = "not-a-number"
|
|
touch_calls: list[str] = []
|
|
agent._touch_activity = lambda desc, **_kw: touch_calls.append(desc)
|
|
messages = [{"role": "user", "content": f"m{i}"} for i in range(20)]
|
|
|
|
strict_calls: list[int | None] = []
|
|
|
|
def _strict_compress(messages, current_tokens=None):
|
|
strict_calls.append(current_tokens)
|
|
return [
|
|
{"role": "user", "content": "[CONTEXT COMPACTION] strict summary"},
|
|
{"role": "user", "content": "tail"},
|
|
]
|
|
|
|
agent.context_compressor.compress = _strict_compress
|
|
|
|
compressed, _sp = agent._compress_context(messages, "sys", approx_tokens=120_000)
|
|
|
|
assert compressed[0]["content"] == "[CONTEXT COMPACTION] strict summary"
|
|
assert touch_calls[0] == "context compression started"
|
|
assert touch_calls[-1] == "context compression completed"
|
|
assert db.get_compression_lock_holder(session_id) is None
|
|
assert strict_calls == [120_000]
|
|
|
|
|
|
def test_compression_activity_heartbeat_nonfinite_interval_falls_back(tmp_path: Path) -> None:
|
|
"""Non-finite heartbeat intervals must not reach Event.wait()."""
|
|
from agent.conversation_compression import _CompressionActivityHeartbeat
|
|
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
session_id = "HEARTBEAT_NONFINITE_INTERVAL_TEST"
|
|
db.create_session(session_id, source="test")
|
|
|
|
agent = _build_agent_with_db(db, session_id)
|
|
touch_calls: list[str] = []
|
|
touch_provenances: list = []
|
|
|
|
def _capture(desc, *, provenance=None, force_persist=False):
|
|
touch_calls.append(desc)
|
|
touch_provenances.append(provenance)
|
|
|
|
agent._touch_activity = _capture
|
|
|
|
heartbeat = _CompressionActivityHeartbeat(agent, interval_seconds=float("inf"))
|
|
|
|
assert heartbeat._interval_seconds == 60.0
|
|
heartbeat.start()
|
|
heartbeat.stop()
|
|
assert touch_calls == ["context compression started", "context compression completed"]
|
|
from agent.session_activity import ActivityProvenance
|
|
|
|
assert touch_provenances == [
|
|
ActivityProvenance.AGENT_COMPRESSION,
|
|
ActivityProvenance.AGENT_COMPRESSION,
|
|
]
|
|
|
|
|
|
def test_compression_heartbeat_stop_persists_completed_over_in_progress(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""/compress is outside run_conversation, so turn-end clear never runs.
|
|
|
|
Heartbeat progress stamps persist to SessionDB; completion is often
|
|
rate-limited out. stop() must force-persist the terminal label so idle
|
|
sessions show "context compression completed", not a stale "in progress".
|
|
"""
|
|
from agent.conversation_compression import _CompressionActivityHeartbeat
|
|
from agent.session_activity import ActivityProvenance
|
|
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
session_id = "HEARTBEAT_PERSIST_COMPLETED_TEST"
|
|
db.create_session(session_id, source="test")
|
|
|
|
agent = _build_agent_with_db(db, session_id)
|
|
# Long interval: only start/stop touch; we inject the progress stamp.
|
|
hb = _CompressionActivityHeartbeat(agent, interval_seconds=3600.0)
|
|
hb.start()
|
|
|
|
agent._session_activity_last_persist_mono = 0.0
|
|
agent._touch_activity(
|
|
"context compression in progress",
|
|
provenance=ActivityProvenance.AGENT_COMPRESSION,
|
|
)
|
|
row = db.get_session(session_id)
|
|
assert row["last_activity_description"] == "context compression in progress"
|
|
assert row["last_activity_provenance"] == ActivityProvenance.AGENT_COMPRESSION.value
|
|
|
|
# Mimic the common case: completion falls inside the 60s persist window.
|
|
agent._session_activity_last_persist_mono = time.monotonic()
|
|
hb.stop("context compression completed")
|
|
|
|
row = db.get_session(session_id)
|
|
assert row["last_activity_description"] == "context compression completed"
|
|
assert row["last_activity_provenance"] == ActivityProvenance.AGENT_COMPRESSION.value
|
|
assert agent._last_activity_desc == "context compression completed"
|
|
assert agent._last_activity_provenance is ActivityProvenance.AGENT_COMPRESSION
|
|
|
|
|
|
def test_compression_heartbeat_does_not_clobber_timeout_provenance() -> None:
|
|
"""Detached heartbeat/stop must not overwrite a host timeout stamp."""
|
|
from types import SimpleNamespace
|
|
|
|
from agent.conversation_compression import _CompressionActivityHeartbeat
|
|
from agent.session_activity import ActivityProvenance
|
|
|
|
agent = SimpleNamespace(
|
|
_last_activity_provenance=ActivityProvenance.AGENT_COMPRESSION_TIMEOUT,
|
|
_last_activity_desc="context compression timed out",
|
|
touches=[],
|
|
)
|
|
|
|
def _touch(desc, *, provenance=None, force_persist=False):
|
|
agent.touches.append((desc, provenance))
|
|
agent._last_activity_provenance = provenance
|
|
agent._last_activity_desc = desc
|
|
|
|
agent._touch_activity = _touch
|
|
|
|
hb = _CompressionActivityHeartbeat(agent, interval_seconds=60.0)
|
|
hb._touch("context compression in progress")
|
|
hb.stop("context compression completed")
|
|
|
|
assert agent.touches == []
|
|
assert agent._last_activity_provenance is ActivityProvenance.AGENT_COMPRESSION_TIMEOUT
|
|
assert agent._last_activity_desc == "context compression timed out"
|
|
|
|
|
|
def test_compression_heartbeat_does_not_clobber_cooldown_provenance() -> None:
|
|
"""Cooldown/abort stamps must also survive a late heartbeat stop."""
|
|
from types import SimpleNamespace
|
|
|
|
from agent.conversation_compression import _CompressionActivityHeartbeat
|
|
from agent.session_activity import ActivityProvenance
|
|
|
|
agent = SimpleNamespace(
|
|
_last_activity_provenance=ActivityProvenance.AGENT_COMPRESSION_COOLDOWN,
|
|
_last_activity_desc="compression blocked (cooldown: 30s remaining)",
|
|
touches=[],
|
|
)
|
|
|
|
def _touch(desc, *, provenance=None, force_persist=False):
|
|
agent.touches.append((desc, provenance))
|
|
agent._last_activity_provenance = provenance
|
|
agent._last_activity_desc = desc
|
|
|
|
agent._touch_activity = _touch
|
|
|
|
hb = _CompressionActivityHeartbeat(agent, interval_seconds=60.0)
|
|
hb._touch("context compression in progress")
|
|
hb.stop("context compression failed")
|
|
|
|
assert agent.touches == []
|
|
assert agent._last_activity_provenance is ActivityProvenance.AGENT_COMPRESSION_COOLDOWN
|
|
|
|
|
|
def test_compression_heartbeat_start_republishes_after_terminal_provenance() -> None:
|
|
"""A new compression episode may overwrite a prior timeout/cooldown stamp."""
|
|
from types import SimpleNamespace
|
|
|
|
from agent.conversation_compression import _CompressionActivityHeartbeat
|
|
from agent.session_activity import ActivityProvenance
|
|
|
|
agent = SimpleNamespace(
|
|
_last_activity_provenance=ActivityProvenance.AGENT_COMPRESSION_TIMEOUT,
|
|
_last_activity_desc="context compression timed out",
|
|
touches=[],
|
|
)
|
|
|
|
def _touch(desc, *, provenance=None, force_persist=False):
|
|
agent.touches.append((desc, provenance))
|
|
agent._last_activity_provenance = provenance
|
|
agent._last_activity_desc = desc
|
|
|
|
agent._touch_activity = _touch
|
|
|
|
hb = _CompressionActivityHeartbeat(agent, interval_seconds=60.0)
|
|
hb.start()
|
|
hb.stop()
|
|
|
|
assert agent.touches[0] == (
|
|
"context compression started",
|
|
ActivityProvenance.AGENT_COMPRESSION,
|
|
)
|
|
assert agent.touches[-1] == (
|
|
"context compression completed",
|
|
ActivityProvenance.AGENT_COMPRESSION,
|
|
)
|
|
assert agent._last_activity_provenance is ActivityProvenance.AGENT_COMPRESSION
|
|
|
|
|
|
def test_compression_heartbeat_does_not_rearm_after_unknown_provenance() -> None:
|
|
"""After a terminal stamp, UNKNOWN must not re-arm a detached heartbeat."""
|
|
from types import SimpleNamespace
|
|
|
|
from agent.conversation_compression import _CompressionActivityHeartbeat
|
|
from agent.session_activity import ActivityProvenance
|
|
|
|
agent = SimpleNamespace(
|
|
_last_activity_provenance=ActivityProvenance.AGENT_COMPRESSION_TIMEOUT,
|
|
_last_activity_desc="context compression timed out",
|
|
touches=[],
|
|
)
|
|
|
|
def _touch(desc, *, provenance=None, force_persist=False):
|
|
agent.touches.append((desc, provenance))
|
|
agent._last_activity_provenance = provenance
|
|
agent._last_activity_desc = desc
|
|
|
|
agent._touch_activity = _touch
|
|
|
|
hb = _CompressionActivityHeartbeat(agent, interval_seconds=60.0)
|
|
# First tick observes TIMEOUT and latches silent.
|
|
hb._touch("context compression in progress")
|
|
assert hb._suppressed is True
|
|
# Turn continues / ends and clears labels to UNKNOWN — must stay silent.
|
|
agent._last_activity_provenance = ActivityProvenance.UNKNOWN
|
|
agent._last_activity_desc = "calling model"
|
|
hb._touch("context compression in progress")
|
|
hb.stop("context compression completed")
|
|
|
|
assert agent.touches == []
|
|
assert agent._last_activity_provenance is ActivityProvenance.UNKNOWN
|
|
assert agent._last_activity_desc == "calling model"
|
|
|
|
|
|
def test_compression_heartbeat_stops_when_commit_fence_cancelled() -> None:
|
|
"""Host fence cancel must silence detached heartbeat refresh and late stop."""
|
|
from types import SimpleNamespace
|
|
|
|
from agent.conversation_compression import (
|
|
CompressionCommitFence,
|
|
_CompressionActivityHeartbeat,
|
|
)
|
|
from agent.session_activity import ActivityProvenance
|
|
|
|
agent = SimpleNamespace(
|
|
_last_activity_provenance=ActivityProvenance.AGENT_COMPRESSION,
|
|
_last_activity_desc="context compression started",
|
|
touches=[],
|
|
)
|
|
|
|
def _touch(desc, *, provenance=None, force_persist=False):
|
|
agent.touches.append((desc, provenance))
|
|
agent._last_activity_provenance = provenance
|
|
agent._last_activity_desc = desc
|
|
|
|
agent._touch_activity = _touch
|
|
|
|
fence = CompressionCommitFence()
|
|
assert fence.cancel_before_commit() is True
|
|
|
|
hb = _CompressionActivityHeartbeat(
|
|
agent, interval_seconds=60.0, commit_fence=fence
|
|
)
|
|
hb._touch("context compression in progress")
|
|
hb.stop("context compression completed")
|
|
|
|
assert agent.touches == []
|
|
assert hb._suppressed is True
|
|
assert agent._last_activity_provenance is ActivityProvenance.AGENT_COMPRESSION
|
|
assert agent._last_activity_desc == "context compression started"
|
|
|
|
|
|
def test_concurrent_compression_does_not_fork_session(tmp_path: Path) -> None:
|
|
"""Two AIAgents that share a session_id MUST NOT both rotate it.
|
|
|
|
Without the per-session compression lock this fixture deterministically
|
|
produces 2 child sessions (transcript fork). With the lock at most one
|
|
path rotates: normally exactly 1 canonical child, or — under heavy DB
|
|
write contention that makes the winner's child create_session exhaust its
|
|
retries — 0, because _compress_context safely rolls back to the parent
|
|
instead of orphaning a child. The forbidden outcome is 2+ (the fork).
|
|
"""
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
|
|
parent_sid = "PARENT_TEST_SESSION"
|
|
db.create_session(parent_sid, source="discord")
|
|
|
|
# Two agents on the same session_id, both wired to the same db —
|
|
# mirrors the parent-turn agent + the background-review fork right
|
|
# after a turn ends.
|
|
agent_a = _build_agent_with_db(db, parent_sid)
|
|
agent_b = _build_agent_with_db(db, parent_sid)
|
|
messages = [{"role": "user", "content": f"m{i}"} for i in range(20)]
|
|
|
|
def run(agent):
|
|
try:
|
|
agent._compress_context(messages, "sys", approx_tokens=120_000)
|
|
except Exception:
|
|
# Surface to the test if either raises — should not happen.
|
|
raise
|
|
|
|
t_a = threading.Thread(target=run, args=(agent_a,), name="main_turn")
|
|
t_b = threading.Thread(target=run, args=(agent_b,), name="review_fork")
|
|
t_a.start()
|
|
t_b.start()
|
|
t_a.join(timeout=10)
|
|
t_b.join(timeout=10)
|
|
|
|
# The invariant Damien's incident is about: the parent must NEVER end up
|
|
# with two (or more) children — that is the transcript fork. The lock
|
|
# guarantees only one path rotates.
|
|
#
|
|
# Zero children is also a valid, non-forking outcome: under heavy DB write
|
|
# contention the winner's child ``create_session`` can exhaust its retry
|
|
# budget, and ``_compress_context`` deliberately rolls the live id back to
|
|
# the (still-indexed) parent rather than stranding an orphan child — see
|
|
# the create-failure rollback in agent/conversation_compression.py. That
|
|
# safe rollback leaves 0 children and is correct. So the contract is
|
|
# ``children <= 1``; only ``>= 2`` is the bug. Asserting an exact ``== 1``
|
|
# made this test flaky under the concurrent CI load that triggers the
|
|
# contention rollback (#54465 churn surfaced it).
|
|
n_children = _count_children(db, parent_sid)
|
|
assert n_children <= 1, (
|
|
f"Compression lock failed: parent session has {n_children} children in "
|
|
"state.db (transcript fork). This is Damien's incident shape — see the "
|
|
"test docstring. Two or more children means the lock did not serialize "
|
|
"the concurrent rotations."
|
|
)
|
|
|
|
# Every agent that moved off the parent must have landed on the SAME id.
|
|
# Counting movers is the wrong contract: the loser can legitimately end up
|
|
# on the child too, without rotating anything itself — it takes the lock
|
|
# after the winner released it, sees the parent was already rotated, and
|
|
# _adopt_live_compression_child() points it at the winner's single child
|
|
# (the "compression recovery: stale session=... adopted live child=..."
|
|
# path). That convergence is the fix working, not a fork; the fork is two
|
|
# DIFFERENT live ids. Asserting ``movers <= 1`` failed on that healthy
|
|
# outcome under concurrent load.
|
|
moved = {a.session_id for a in (agent_a, agent_b) if a.session_id != parent_sid}
|
|
assert len(moved) <= 1, (
|
|
f"Expected at most one post-compression session id, got {sorted(moved)}. "
|
|
"Two distinct ids means the lock didn't serialize them (transcript fork)."
|
|
)
|
|
assert len(moved) == n_children, (
|
|
f"Inconsistent state: agents live on {sorted(moved)} but {n_children} "
|
|
"child session(s) exist — rotation and child creation diverged."
|
|
)
|
|
if moved:
|
|
child = _live_child_id(db, parent_sid)
|
|
assert moved == {child}, (
|
|
f"Agents live on {sorted(moved)} but the parent's only child is "
|
|
f"{child} — an agent is writing to a session outside the lineage."
|
|
)
|
|
|
|
# The lock must be released after both paths finished, regardless of
|
|
# whether the winner committed a child or rolled back.
|
|
assert db.get_compression_lock_holder(parent_sid) is None, (
|
|
"Compression lock leaked: still held after both paths completed."
|
|
)
|
|
|
|
|
|
def test_durable_message_committed_before_lease_is_adopted(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""A durable row absent from the caller snapshot must still be compressed.
|
|
|
|
Previously this path aborted and returned the stale snapshot unchanged,
|
|
which permanently wedged busy sessions: every compress attempt saw the
|
|
DB ahead of the in-memory list, logged "changed before lease
|
|
acquisition", and never called the compressor. Adopting the durable
|
|
transcript keeps the late-committed turn and lets compression proceed.
|
|
"""
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
parent_sid = "PRE_LEASE_DURABLE_RACE"
|
|
db.create_session(parent_sid, source="webui")
|
|
db.append_message(parent_sid, "user", "old durable")
|
|
|
|
# Frontend takes its snapshot, then another producer commits before this
|
|
# compressor acquires the lease.
|
|
stale_snapshot = [{"role": "user", "content": "old durable"}]
|
|
db.append_message(parent_sid, "assistant", "late committed before lease")
|
|
agent = _build_agent_with_db(db, parent_sid)
|
|
|
|
returned, _system_prompt = agent._compress_context(
|
|
stale_snapshot, "sys", approx_tokens=120_000
|
|
)
|
|
|
|
agent.context_compressor.compress.assert_called_once()
|
|
compressed_arg = agent.context_compressor.compress.call_args.args[0]
|
|
assert [m["content"] for m in compressed_arg] == [
|
|
"old durable",
|
|
"late committed before lease",
|
|
]
|
|
# Must not echo the stale snapshot — compression proceeded on the
|
|
# adopted durable transcript (rotation publishes a child session).
|
|
assert returned is not stale_snapshot
|
|
assert returned[0]["content"] == "[CONTEXT COMPACTION] summary"
|
|
assert agent.session_id != parent_sid
|
|
child_id = _live_child_id(db, parent_sid)
|
|
assert child_id is not None
|
|
assert child_id == agent.session_id
|
|
|
|
|
|
|
|
|
|
|
|
def test_fence_cancelled_compression_leaves_lock_reacquirable(tmp_path: Path) -> None:
|
|
"""A fence-cancelled attempt must not poison the per-session lock.
|
|
|
|
Lock-release verification for the hygiene-timeout path: after the gateway
|
|
times out and cancels a hygiene compression at the commit fence, the very
|
|
next attempt on the same session (e.g. the user running ``/compress``)
|
|
must acquire the compression lock and commit normally. A leaked lock here
|
|
would silently block every future compaction for the session until TTL
|
|
expiry.
|
|
"""
|
|
from agent.conversation_compression import CompressionCommitFence
|
|
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
session_id = "HYGIENE_LOCK_REACQUIRE"
|
|
db.create_session(session_id, source="telegram")
|
|
|
|
agent = _build_agent_with_db(db, session_id)
|
|
agent.compression_in_place = True
|
|
agent._cached_system_prompt = "sys"
|
|
summary_started = threading.Event()
|
|
release_summary = threading.Event()
|
|
|
|
def _slow_summary(*_args, **_kwargs):
|
|
summary_started.set()
|
|
assert release_summary.wait(timeout=5)
|
|
agent.context_compressor._proactive_prune_rearm_tokens = 0
|
|
return [
|
|
{"role": "user", "content": "[CONTEXT COMPACTION] summary"},
|
|
{"role": "user", "content": "tail"},
|
|
]
|
|
|
|
agent.context_compressor.compress.side_effect = _slow_summary
|
|
agent.context_compressor._proactive_prune_rearm_tokens = 120_000
|
|
messages = [{"role": "user", "content": f"m{i}"} for i in range(20)]
|
|
fence = CompressionCommitFence()
|
|
result = {}
|
|
|
|
def _run_compression() -> None:
|
|
result["value"] = agent._compress_context(
|
|
messages,
|
|
"sys",
|
|
approx_tokens=120_000,
|
|
commit_fence=fence,
|
|
)
|
|
|
|
worker = threading.Thread(target=_run_compression, name="fenced-hygiene")
|
|
worker.start()
|
|
assert summary_started.wait(timeout=2)
|
|
assert fence.cancel_before_commit() is True
|
|
release_summary.set()
|
|
worker.join(timeout=5)
|
|
assert not worker.is_alive()
|
|
|
|
# Cancelled attempt: no mutation, and — the invariant under test — the
|
|
# per-session compression lock is fully released.
|
|
assert result["value"][0] is messages
|
|
assert agent.context_compressor._proactive_prune_rearm_tokens == 120_000
|
|
assert db.get_compression_lock_holder(session_id) is None
|
|
|
|
# The NEXT attempt (no fence — a manual /compress retry) must be able to
|
|
# acquire the lock and commit an in-place compaction normally.
|
|
agent.context_compressor.compress.side_effect = lambda *_a, **_kw: [
|
|
{"role": "user", "content": "[CONTEXT COMPACTION] retry summary"},
|
|
{"role": "user", "content": "tail"},
|
|
]
|
|
retried, _sp = agent._compress_context(messages, "sys", approx_tokens=120_000)
|
|
|
|
assert retried is not messages
|
|
assert len(retried) < len(messages)
|
|
assert agent.session_id == session_id # in-place: same session id
|
|
assert agent._last_compaction_in_place is True
|
|
assert db.get_compression_lock_holder(session_id) is None
|
|
|
|
|
|
def test_commit_fence_waits_for_an_active_commit() -> None:
|
|
"""A timeout that loses the fence race cannot overlap the live turn."""
|
|
from agent.conversation_compression import CompressionCommitFence
|
|
|
|
fence = CompressionCommitFence()
|
|
assert fence.begin_commit() is True
|
|
assert fence.try_cancel_before_commit() is None
|
|
cancel_started = threading.Event()
|
|
cancel_finished = threading.Event()
|
|
result = {}
|
|
|
|
def _cancel() -> None:
|
|
cancel_started.set()
|
|
result["cancelled"] = fence.cancel_before_commit()
|
|
cancel_finished.set()
|
|
|
|
waiter = threading.Thread(target=_cancel, name="hygiene-timeout-fence")
|
|
waiter.start()
|
|
try:
|
|
assert cancel_started.wait(timeout=2)
|
|
assert not cancel_finished.is_set()
|
|
finally:
|
|
fence.finish_commit()
|
|
waiter.join(timeout=2)
|
|
|
|
assert not waiter.is_alive()
|
|
assert result["cancelled"] is False
|
|
|
|
|
|
def test_total_deadline_cancellation_retains_lock_until_worker_cleanup() -> None:
|
|
"""A total-ceiling timeout must exclude retries while its worker is alive."""
|
|
from agent.conversation_compression import CompressionCommitFence
|
|
|
|
released = threading.Event()
|
|
fence = CompressionCommitFence(total_ceiling_seconds=1.0)
|
|
fence.register_cancelled_lock_release(released.set)
|
|
fence.retain_compression_lock_until_worker_done()
|
|
|
|
now = time.monotonic()
|
|
with patch(
|
|
"agent.conversation_compression.time.monotonic",
|
|
return_value=now + 2.0,
|
|
):
|
|
assert fence.is_cancelled
|
|
assert fence.try_cancel_before_commit() is True
|
|
fence.release_cancelled_compression_lock()
|
|
|
|
assert not released.is_set()
|
|
|
|
|
|
def test_delayed_contender_adopts_unique_rotated_child(tmp_path: Path) -> None:
|
|
"""A stale agent must continue on the winner's compacted child transcript."""
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
parent_sid = "STALE_PARENT"
|
|
child_sid = "CANONICAL_CHILD"
|
|
db.create_session(parent_sid, source="webui")
|
|
db.end_session(parent_sid, "compression")
|
|
db.create_session(child_sid, source="webui", parent_session_id=parent_sid)
|
|
compacted = [
|
|
{"role": "user", "content": "[CONTEXT COMPACTION] summary"},
|
|
{"role": "assistant", "content": "compacted tail"},
|
|
]
|
|
db.replace_messages(child_sid, compacted)
|
|
|
|
agent = _build_agent_with_db(db, parent_sid)
|
|
stale_messages = [
|
|
{"role": "user", "content": "stale"},
|
|
{"role": "assistant", "content": "x" * 1000},
|
|
]
|
|
recovered, _system_prompt = agent._compress_context(
|
|
stale_messages, "sys", approx_tokens=120_000
|
|
)
|
|
|
|
assert agent.session_id == child_sid
|
|
assert [(m["role"], m["content"]) for m in recovered] == [
|
|
("user", "[CONTEXT COMPACTION] summary"),
|
|
("assistant", "compacted tail"),
|
|
]
|
|
assert agent._session_db_created is True
|
|
assert agent._flushed_db_message_session_id == child_sid
|
|
assert agent._last_flushed_db_idx == len(recovered)
|
|
agent.context_compressor.compress.assert_not_called()
|
|
lifecycle_args, lifecycle_kwargs = agent.context_compressor.on_session_start.call_args
|
|
assert lifecycle_args == (child_sid,)
|
|
assert lifecycle_kwargs["boundary_reason"] == "compression"
|
|
assert lifecycle_kwargs["old_session_id"] == parent_sid
|
|
assert lifecycle_kwargs["session_db"] is db
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _no_consecutive_user_roles(messages: list) -> bool:
|
|
roles = [m.get("role") for m in messages if isinstance(m, dict)]
|
|
return all(
|
|
not (roles[i] == roles[i + 1] == "user") for i in range(len(roles) - 1)
|
|
)
|
|
|
|
|
|
def test_restored_anchor_never_creates_consecutive_user_roles() -> None:
|
|
"""Anchor restoration must preserve strict role alternation (#55677).
|
|
|
|
The original insertion helper could land the human anchor directly next
|
|
to user-role scaffolding (index-0 insert before a leading synthetic user
|
|
turn, or a bare scaffolding-only transcript), producing user/user
|
|
adjacency that strict chat templates reject.
|
|
"""
|
|
from agent.conversation_compression import _insert_real_user_anchor
|
|
|
|
anchor = {"role": "user", "content": "REAL HUMAN ASK"}
|
|
|
|
# Leading synthetic user turn before the assistant summary.
|
|
compressed = [
|
|
{
|
|
"role": "user",
|
|
"content": "[System: Your previous response was truncated ...]",
|
|
"_empty_recovery_synthetic": True,
|
|
},
|
|
{"role": "assistant", "content": "summary"},
|
|
{
|
|
"role": "user",
|
|
"content": "[Your active task list was preserved across context compression]",
|
|
"_todo_snapshot_synthetic": True,
|
|
},
|
|
]
|
|
_insert_real_user_anchor(compressed, dict(anchor))
|
|
assert _no_consecutive_user_roles(compressed)
|
|
assert any(m.get("content", "").startswith("REAL HUMAN ASK") for m in compressed)
|
|
|
|
# Scaffolding-only transcript: the anchor is merged, not inserted
|
|
# adjacent, and the merged turn leads with the human ask.
|
|
compressed = [
|
|
{
|
|
"role": "user",
|
|
"content": "[Your active task list was preserved across context compression]",
|
|
"_todo_snapshot_synthetic": True,
|
|
},
|
|
]
|
|
_insert_real_user_anchor(compressed, dict(anchor))
|
|
assert _no_consecutive_user_roles(compressed)
|
|
assert len(compressed) == 1
|
|
assert compressed[0]["content"].startswith("REAL HUMAN ASK")
|
|
assert not compressed[0].get("_todo_snapshot_synthetic")
|
|
|
|
|
|
|
|
|
|
def test_compression_persists_child_handoff_immediately(tmp_path: Path) -> None:
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
parent_sid = "HEADLESS_PREFLIGHT_PARENT"
|
|
db.create_session(parent_sid, source="cli")
|
|
|
|
agent = _build_agent_with_db(db, parent_sid)
|
|
messages = [{"role": "user", "content": f"m{i}"} for i in range(20)]
|
|
|
|
compressed, _sp = agent._compress_context(messages, "sys", approx_tokens=120_000)
|
|
child_sid = agent.session_id
|
|
|
|
assert child_sid != parent_sid
|
|
assert db.get_session(parent_sid)["end_reason"] == "compression"
|
|
assert len(db.get_messages(child_sid)) == len(compressed)
|
|
|
|
agent._flush_messages_to_session_db(compressed, None)
|
|
assert len(db.get_messages(child_sid)) == len(compressed)
|
|
|
|
|
|
|
|
|
|
def test_rotation_publish_failure_restores_proactive_prune_runway(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
parent_sid = "PRUNE_RUNWAY_ROLLBACK_PARENT"
|
|
db.create_session(
|
|
parent_sid,
|
|
source="cli",
|
|
model_config={"keep": "value", "_proactive_prune_rearm_tokens": 120_000},
|
|
)
|
|
messages = [{"role": "user", "content": f"m{i}"} for i in range(20)]
|
|
db.append_messages_batch(parent_sid, messages)
|
|
for message in messages:
|
|
message["_db_persisted"] = True
|
|
agent = _build_agent_with_db(db, parent_sid)
|
|
agent.context_compressor._proactive_prune_rearm_tokens = 120_000
|
|
|
|
def _compress(*_args, **_kwargs):
|
|
agent.context_compressor._proactive_prune_rearm_tokens = 0
|
|
return [
|
|
{"role": "user", "content": "[CONTEXT COMPACTION] summary"},
|
|
{"role": "user", "content": "tail"},
|
|
]
|
|
|
|
agent.context_compressor.compress.side_effect = _compress
|
|
durable_before = db.get_messages_as_conversation(parent_sid)
|
|
with patch.object(
|
|
db,
|
|
"publish_compression_child",
|
|
side_effect=RuntimeError("publish failed"),
|
|
):
|
|
returned, _sp = agent._compress_context(
|
|
messages, "sys", approx_tokens=120_000,
|
|
)
|
|
|
|
assert returned is messages
|
|
assert agent.session_id == parent_sid
|
|
assert agent.context_compressor._proactive_prune_rearm_tokens == 120_000
|
|
assert db.get_messages_as_conversation(parent_sid) == durable_before
|
|
assert json.loads(db.get_session(parent_sid)["model_config"]) == {
|
|
"keep": "value",
|
|
"_proactive_prune_rearm_tokens": 120_000,
|
|
}
|
|
|
|
|
|
def test_full_in_place_compression_atomically_clears_durable_prune_runway(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
session_id = "IN_PLACE_CLEARS_PRUNE_RUNWAY"
|
|
db.create_session(
|
|
session_id,
|
|
source="cli",
|
|
model_config={"keep": "value", "_proactive_prune_rearm_tokens": 120_000},
|
|
)
|
|
messages = [{"role": "user", "content": f"m{i}"} for i in range(20)]
|
|
db.append_messages_batch(session_id, messages)
|
|
agent = _build_agent_with_db(db, session_id)
|
|
agent.compression_in_place = True
|
|
agent.context_compressor._proactive_prune_rearm_tokens = 120_000
|
|
|
|
compressed, _sp = agent._compress_context(
|
|
messages, "sys", approx_tokens=120_000,
|
|
)
|
|
|
|
assert agent.session_id == session_id
|
|
assert [message["content"] for message in db.get_messages_as_conversation(session_id)] == [
|
|
message["content"] for message in compressed
|
|
]
|
|
assert json.loads(db.get_session(session_id)["model_config"]) == {"keep": "value"}
|
|
|
|
|
|
def test_rotation_child_starts_without_durable_prune_runway(tmp_path: Path) -> None:
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
parent_sid = "ROTATION_CLEARS_PRUNE_RUNWAY"
|
|
db.create_session(
|
|
parent_sid,
|
|
source="cli",
|
|
model_config={"keep": "parent", "_proactive_prune_rearm_tokens": 120_000},
|
|
)
|
|
messages = [{"role": "user", "content": f"m{i}"} for i in range(20)]
|
|
db.append_messages_batch(parent_sid, messages)
|
|
agent = _build_agent_with_db(db, parent_sid)
|
|
|
|
agent._compress_context(messages, "sys", approx_tokens=120_000)
|
|
|
|
assert agent.session_id != parent_sid
|
|
child_config = json.loads(db.get_session(agent.session_id)["model_config"])
|
|
assert "_proactive_prune_rearm_tokens" not in child_config
|
|
assert json.loads(db.get_session(parent_sid)["model_config"])[
|
|
"_proactive_prune_rearm_tokens"
|
|
] == 120_000
|
|
|
|
|
|
@pytest.mark.parametrize("in_place", [False, True])
|
|
def test_equal_copy_compression_result_does_not_rewrite_session(
|
|
tmp_path: Path,
|
|
in_place: bool,
|
|
) -> None:
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
parent_sid = f"EQUAL_COPY_NOOP_{in_place}"
|
|
db.create_session(parent_sid, source="cli")
|
|
|
|
agent = _build_agent_with_db(db, parent_sid)
|
|
setattr(agent, "compression_in_place", in_place)
|
|
messages = [{"role": "user", "content": f"m{i}"} for i in range(20)]
|
|
compressor = getattr(agent, "context_compressor")
|
|
compressor.compress.side_effect = lambda incoming, **_kw: list(incoming)
|
|
|
|
with patch.object(
|
|
db,
|
|
"archive_and_compact",
|
|
wraps=db.archive_and_compact,
|
|
) as archive_and_compact:
|
|
returned, _sp = agent._compress_context(
|
|
messages,
|
|
"sys",
|
|
approx_tokens=120_000,
|
|
)
|
|
|
|
assert returned is messages
|
|
assert getattr(agent, "session_id") == parent_sid
|
|
assert _count_children(db, parent_sid) == 0
|
|
parent = db.get_session(parent_sid)
|
|
assert parent is not None
|
|
assert parent["end_reason"] is None
|
|
assert db.get_compression_lock_holder(parent_sid) is None
|
|
archive_and_compact.assert_not_called()
|
|
|
|
|
|
|
|
|
|
def test_post_compress_exception_stops_lock_refresher(tmp_path: Path, monkeypatch) -> None:
|
|
"""A warning-path exception after compress() returns must still release the lock."""
|
|
real_try_acquire = SessionDB.try_acquire_compression_lock
|
|
|
|
def _short_ttl(self, session_id: str, holder: str, ttl_seconds: float = 300.0) -> bool:
|
|
return real_try_acquire(self, session_id, holder, ttl_seconds=0.15)
|
|
|
|
monkeypatch.setattr(SessionDB, "try_acquire_compression_lock", _short_ttl)
|
|
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
parent_sid = "REFRESH_EXCEPTION_TEST"
|
|
db.create_session(parent_sid, source="discord")
|
|
|
|
agent = _build_agent_with_db(db, parent_sid)
|
|
agent._compression_lock_ttl_seconds = 0.15
|
|
agent._compression_lock_refresh_interval = 0.05
|
|
agent.context_compressor._last_summary_error = "summary failed"
|
|
agent._emit_warning = lambda *_a, **_k: (_ for _ in ()).throw(RuntimeError("warn boom"))
|
|
|
|
messages = [{"role": "user", "content": f"m{i}"} for i in range(20)]
|
|
|
|
with pytest.raises(RuntimeError, match="warn boom"):
|
|
agent._compress_context(messages, "sys", approx_tokens=120_000)
|
|
|
|
time.sleep(0.25)
|
|
assert db.try_acquire_compression_lock(parent_sid, "probe", ttl_seconds=1.0) is True
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_signature_introspection_exception_releases_lock_and_refresher(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
"""Capability inspection failures must not leak the acquired lock lease."""
|
|
from agent.conversation_compression import (
|
|
_CompressionLockLeaseRefresher as RealLeaseRefresher,
|
|
)
|
|
|
|
refreshers = []
|
|
|
|
class RecordingLeaseRefresher(RealLeaseRefresher):
|
|
def start(self):
|
|
refreshers.append(self)
|
|
return super().start()
|
|
|
|
monkeypatch.setattr(
|
|
"agent.conversation_compression._CompressionLockLeaseRefresher",
|
|
RecordingLeaseRefresher,
|
|
)
|
|
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
parent_sid = "SIGNATURE_EXCEPTION_TEST"
|
|
db.create_session(parent_sid, source="discord")
|
|
|
|
agent = _build_agent_with_db(db, parent_sid)
|
|
agent._compression_lock_refresh_interval = 0.1
|
|
|
|
class SignatureBomb:
|
|
calls = 0
|
|
|
|
@property
|
|
def __signature__(self):
|
|
raise RuntimeError("signature boom")
|
|
|
|
def __call__(self, *_args, **_kwargs):
|
|
self.calls += 1
|
|
raise AssertionError("engine must not run after signature failure")
|
|
|
|
bomb = SignatureBomb()
|
|
agent.context_compressor.compress = bomb
|
|
messages = [{"role": "user", "content": f"m{i}"} for i in range(20)]
|
|
|
|
with pytest.raises(RuntimeError, match="signature boom"):
|
|
agent._compress_context(messages, "sys", approx_tokens=120_000)
|
|
|
|
assert bomb.calls == 0
|
|
assert db.get_compression_lock_holder(parent_sid) is None
|
|
assert len(refreshers) == 1
|
|
assert not refreshers[0]._thread.is_alive()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _make_legacy_session_db_class() -> type:
|
|
"""Model the class retained in ``sys.modules`` before the lock API existed.
|
|
|
|
During the real version-skew incident, a re-imported compression module
|
|
imports the same still-loaded ``hermes_state`` module, whose ``SessionDB``
|
|
class is old. The test replaces that module attribute with this lockless
|
|
class and forwards all persistence operations to a current real database.
|
|
"""
|
|
source_path = inspect.getfile(SessionDB)
|
|
namespace = {"__name__": "hermes_state"}
|
|
source = '''
|
|
class SessionDB:
|
|
def __init__(self, real_db):
|
|
self._real = real_db
|
|
|
|
def __getattribute__(self, name):
|
|
if name in {"_real", "__class__"}:
|
|
return object.__getattribute__(self, name)
|
|
return getattr(object.__getattribute__(self, "_real"), name)
|
|
'''
|
|
exec(compile(source, source_path, "exec"), namespace)
|
|
return namespace["SessionDB"]
|
|
|
|
|
|
class _NominalSessionDBImpostor:
|
|
"""A proxy that spoofs names but lacks the real SessionDB source contract."""
|
|
|
|
def __init__(self, real_db: SessionDB) -> None:
|
|
self._real = real_db
|
|
|
|
def create_session(self, *args, **kwargs):
|
|
return self._real.create_session(*args, **kwargs)
|
|
|
|
def __getattr__(self, name):
|
|
if name == "try_acquire_compression_lock":
|
|
raise AttributeError(name)
|
|
return getattr(self._real, name)
|
|
|
|
|
|
_NominalSessionDBImpostor.__module__ = "hermes_state"
|
|
_NominalSessionDBImpostor.__name__ = "SessionDB"
|
|
|
|
|
|
class _BrokenLockLookupDB:
|
|
"""A present lock API whose instance lookup fails unexpectedly."""
|
|
|
|
def __init__(self, real_db: SessionDB, error: Exception) -> None:
|
|
self._real = real_db
|
|
self._error = error
|
|
|
|
def try_acquire_compression_lock(self, *_args, **_kwargs):
|
|
raise AssertionError("the broken lookup must not resolve to a callable")
|
|
|
|
def __getattribute__(self, name):
|
|
if name == "try_acquire_compression_lock":
|
|
raise object.__getattribute__(self, "_error")
|
|
if name in {"_real", "_error", "__class__"}:
|
|
return object.__getattribute__(self, name)
|
|
return getattr(object.__getattribute__(self, "_real"), name)
|
|
|
|
|
|
class _NonCallableLockAPI:
|
|
"""A present lock API descriptor that resolves to a non-callable value."""
|
|
|
|
def __init__(self, real_db: SessionDB) -> None:
|
|
self._real = real_db
|
|
|
|
try_acquire_compression_lock = None
|
|
|
|
def __getattr__(self, name):
|
|
return getattr(self._real, name)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"error",
|
|
[
|
|
RuntimeError("simulated lock-table corruption"),
|
|
AttributeError("simulated internal lock attribute error"),
|
|
TypeError("simulated internal lock type error"),
|
|
],
|
|
)
|
|
def test_real_lock_api_internal_errors_fail_closed_skips_compression(
|
|
tmp_path: Path, monkeypatch, error: Exception
|
|
) -> None:
|
|
"""Errors after a real lock API resolves must preserve session lineage.
|
|
|
|
``AttributeError`` only means version skew while resolving the method. This
|
|
test injects failures beneath the real ``SessionDB.try_acquire...`` body,
|
|
proving that an internal AttributeError or TypeError cannot take the
|
|
structural-absence compatibility path.
|
|
"""
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
parent_sid = "ERRORING_LOCK_TEST"
|
|
db.create_session(parent_sid, source="discord")
|
|
|
|
def _fail_lock_write(_fn):
|
|
raise error
|
|
|
|
monkeypatch.setattr(db, "_execute_write", _fail_lock_write)
|
|
agent = _build_agent_with_db(db, parent_sid)
|
|
messages = [{"role": "user", "content": f"m{i}"} for i in range(20)]
|
|
|
|
compressed, _sp = agent._compress_context(messages, "sys", approx_tokens=120_000)
|
|
|
|
# Skipped: messages returned verbatim, no rotation, compressor never ran.
|
|
assert compressed is messages or compressed == messages
|
|
assert agent.session_id == parent_sid
|
|
assert _count_children(db, parent_sid) == 0
|
|
agent.context_compressor.compress.assert_not_called()
|
|
|
|
|
|
|
|
|
|
def test_review_fork_compacts_oversized_snapshot_in_memory(tmp_path: Path) -> None:
|
|
"""An oversized review snapshot replays warm on the first request, then
|
|
compacts in memory before further requests — without mutating the parent.
|
|
|
|
Regression for #93057: the fork historically pinned ``compression_enabled =
|
|
False`` because it shares the parent's session_id (issue #38727). That
|
|
left the review's private snapshot unbounded — every follow-up request in
|
|
the review tool loop replayed the whole snapshot (350k-384k input tokens
|
|
per request in production). The fix detaches the fork's compressor from
|
|
the parent's SessionDB/session_id and enables in-memory-only compaction.
|
|
|
|
This test drives the REAL ``_run_review_in_thread`` + ``run_conversation``
|
|
with a threshold-crossing snapshot across two provider requests and
|
|
asserts:
|
|
• the FIRST request replays the full snapshot untouched (warm
|
|
prompt-cache parity) — no compaction summary, middle turns present;
|
|
• compression actually fired before the SECOND request (a real
|
|
threshold crossing, not just setup-time binding state), and that
|
|
request carries the compaction summary and none of the middle
|
|
snapshot turns;
|
|
• the fork keeps the parent's session_id (prompt-cache parity) but its
|
|
agent-level AND compressor-level session bindings are detached;
|
|
• the parent's durable transcript, session row, and child-session graph
|
|
are byte-for-byte unchanged after the compaction ran.
|
|
"""
|
|
import agent.background_review as br
|
|
|
|
parent_sid = "REVIEW_FORK_IN_MEMORY_COMPACTION_93057"
|
|
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
db.create_session(parent_sid, source="discord")
|
|
db.append_message(parent_sid, role="user", content="durable parent turn")
|
|
durable_before = db.get_messages(parent_sid)
|
|
session_row_before = tuple(
|
|
db._conn.execute(
|
|
"SELECT id, parent_session_id, ended_at, end_reason FROM sessions WHERE id = ?",
|
|
(parent_sid,),
|
|
).fetchone()
|
|
)
|
|
parent = _build_agent_with_db(db, parent_sid)
|
|
parent._cached_system_prompt = "stable parent prompt"
|
|
|
|
snapshot = [
|
|
{
|
|
"role": "user" if i % 2 == 0 else "assistant",
|
|
"content": f"review turn {i} " + "x" * 200,
|
|
}
|
|
for i in range(24)
|
|
]
|
|
|
|
captured = {}
|
|
|
|
def _tool_response(prompt_tokens: int) -> SimpleNamespace:
|
|
message = SimpleNamespace(
|
|
content=None,
|
|
reasoning_content=None,
|
|
reasoning=None,
|
|
tool_calls=[
|
|
SimpleNamespace(
|
|
id="call_1",
|
|
type="function",
|
|
function=SimpleNamespace(
|
|
name="web_search", arguments='{"query": "x"}'
|
|
),
|
|
)
|
|
],
|
|
)
|
|
return SimpleNamespace(
|
|
choices=[SimpleNamespace(message=message, finish_reason="tool_calls")],
|
|
model="test/model",
|
|
usage=SimpleNamespace(
|
|
prompt_tokens=prompt_tokens,
|
|
completion_tokens=1,
|
|
total_tokens=prompt_tokens + 1,
|
|
),
|
|
)
|
|
|
|
def _final_response():
|
|
return SimpleNamespace(
|
|
choices=[
|
|
SimpleNamespace(
|
|
index=0,
|
|
message=SimpleNamespace(
|
|
role="assistant",
|
|
content="review complete",
|
|
tool_calls=None,
|
|
reasoning_content=None,
|
|
),
|
|
finish_reason="stop",
|
|
)
|
|
],
|
|
usage=SimpleNamespace(
|
|
prompt_tokens=100, completion_tokens=10, total_tokens=110
|
|
),
|
|
model="test/model",
|
|
)
|
|
|
|
# The worker does a local ``from run_agent import AIAgent``; patching the
|
|
# class method covers that import path.
|
|
from run_agent import AIAgent
|
|
|
|
real_run_conversation = AIAgent.run_conversation
|
|
|
|
def _run_threshold_crossing_review(self, *args, **kwargs):
|
|
captured["compression_enabled"] = self.compression_enabled
|
|
captured["compression_in_place"] = self.compression_in_place
|
|
captured["session_id"] = self.session_id
|
|
captured["session_db"] = self._session_db
|
|
captured["input_budget"] = getattr(
|
|
self, "_review_input_token_budget", "missing"
|
|
)
|
|
captured["defer_first_request"] = getattr(
|
|
self, "_review_defer_compaction_before_first_response", "missing"
|
|
)
|
|
captured["compressor_session_db"] = getattr(
|
|
self.context_compressor, "_session_db", "missing"
|
|
)
|
|
captured["compressor_session_id"] = getattr(
|
|
self.context_compressor, "_session_id", "missing"
|
|
)
|
|
# Stub the fork's compressor so compaction output is deterministic
|
|
# and no aux-LLM call happens; the trigger/commit paths stay real.
|
|
self.context_compressor.threshold_tokens = 1
|
|
self.context_compressor.protect_first_n = 1
|
|
self.context_compressor.protect_last_n = 1
|
|
self.context_compressor.compress = MagicMock(
|
|
return_value=[
|
|
{"role": "user", "content": "[CONTEXT COMPACTION] review summary"},
|
|
{"role": "assistant", "content": "summary acknowledged"},
|
|
]
|
|
)
|
|
# Compress on the first pressure check after the first response, then
|
|
# stand down so the compacted request proceeds instead of looping.
|
|
_should_compress_calls = {"count": 0}
|
|
|
|
def _should_compress(_tokens):
|
|
_should_compress_calls["count"] += 1
|
|
return _should_compress_calls["count"] == 1
|
|
|
|
self.context_compressor.should_compress = MagicMock(
|
|
side_effect=_should_compress
|
|
)
|
|
self.context_compressor.should_compress_info = MagicMock(
|
|
return_value=(True, "over threshold")
|
|
)
|
|
self.context_compressor.should_compress_preflight = MagicMock(
|
|
return_value=True
|
|
)
|
|
self.context_compressor.should_defer_preflight_to_real_usage = MagicMock(
|
|
return_value=False
|
|
)
|
|
self.context_compressor.get_active_compression_failure_cooldown = MagicMock(
|
|
return_value=None
|
|
)
|
|
self.context_compressor.select_context = MagicMock(return_value=None)
|
|
self._compression_feasibility_checked = True
|
|
self.client = MagicMock()
|
|
self.client.chat.completions.create.side_effect = [
|
|
_tool_response(100),
|
|
_final_response(),
|
|
]
|
|
self._disable_streaming = True
|
|
self._use_prompt_caching = False
|
|
|
|
def _fake_execute_tool_calls(assistant_message, messages, *_args):
|
|
tool_call = assistant_message.tool_calls[0]
|
|
messages.append(
|
|
{
|
|
"role": "tool",
|
|
"name": tool_call.function.name,
|
|
"tool_call_id": tool_call.id,
|
|
"content": "ok",
|
|
}
|
|
)
|
|
|
|
self._execute_tool_calls = _fake_execute_tool_calls
|
|
|
|
result = real_run_conversation(self, *args, **kwargs)
|
|
captured["compression_calls"] = self.context_compressor.compress.call_count
|
|
create = self.client.chat.completions.create
|
|
captured["create_calls"] = create.call_count
|
|
captured["outbound"] = [
|
|
call.kwargs.get("messages") for call in create.call_args_list
|
|
]
|
|
return result
|
|
|
|
try:
|
|
with patch.object(AIAgent, "run_conversation", _run_threshold_crossing_review):
|
|
br._run_review_in_thread(parent, snapshot, "review this conversation")
|
|
|
|
assert captured["compression_calls"] >= 1, (
|
|
"FIX REGRESSION: the review fork did not compact its oversized "
|
|
"snapshot. compression_enabled was historically set False on the "
|
|
"fork, which removed the only bound on the replayed snapshot "
|
|
"(issue #93057)."
|
|
)
|
|
assert captured["create_calls"] == 2, (
|
|
f"expected a 2-request review (tool call + final), "
|
|
f"got {captured['create_calls']}"
|
|
)
|
|
first_outbound, second_outbound = captured["outbound"]
|
|
first_contents = [str(m.get("content", "")) for m in first_outbound]
|
|
second_contents = [str(m.get("content", "")) for m in second_outbound]
|
|
# Warm-cache parity: the first request replays the full snapshot
|
|
# untouched — middle turns present, no compaction summary yet.
|
|
assert any("review turn 12" in text for text in first_contents), (
|
|
"the review fork's FIRST request must replay the full snapshot "
|
|
"(warm prompt-cache read) — compaction must not rewrite it before "
|
|
"the first provider call"
|
|
)
|
|
assert not any(
|
|
"[CONTEXT COMPACTION]" in text for text in first_contents
|
|
), f"first request was compacted prematurely: {first_contents!r}"
|
|
# The SECOND request carries the compaction summary and none of the
|
|
# middle snapshot turns.
|
|
assert any(
|
|
"[CONTEXT COMPACTION] review summary" in text
|
|
for text in second_contents
|
|
), f"outbound request did not contain the compaction summary: {second_contents!r}"
|
|
assert not any("review turn 12" in text for text in second_contents), (
|
|
"outbound request still replays the middle of the snapshot — "
|
|
"the review replayed an unbounded transcript despite compaction"
|
|
)
|
|
assert captured["session_id"] == parent_sid, (
|
|
"Review fork should inherit the parent's session_id for "
|
|
"prompt-cache parity."
|
|
)
|
|
assert captured["session_db"] is None, (
|
|
"Review fork must keep its agent-level SessionDB detached "
|
|
"(_session_db=None) so compaction can never persist."
|
|
)
|
|
assert captured["compressor_session_db"] is None, (
|
|
"Review fork's compressor must be detached from the parent "
|
|
"SessionDB, otherwise durable cooldown/streak writes land on "
|
|
"the parent's row (issue #93057)."
|
|
)
|
|
assert captured["compressor_session_id"] == ""
|
|
assert captured["compression_enabled"] is True
|
|
assert captured["compression_in_place"] is True
|
|
assert captured["defer_first_request"] is True
|
|
assert isinstance(captured["input_budget"], int) and captured["input_budget"] > 0
|
|
|
|
# Parent session must be byte-for-byte unchanged after the review
|
|
# compacted its private snapshot.
|
|
assert parent.session_id == parent_sid
|
|
assert db.get_messages(parent_sid) == durable_before
|
|
session_row_after = tuple(
|
|
db._conn.execute(
|
|
"SELECT id, parent_session_id, ended_at, end_reason FROM sessions WHERE id = ?",
|
|
(parent_sid,),
|
|
).fetchone()
|
|
)
|
|
assert session_row_after == session_row_before
|
|
assert _count_children(db, parent_sid) == 0
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_review_fork_fails_closed_when_compressor_rebind_raises(
|
|
tmp_path: Path, caplog
|
|
) -> None:
|
|
"""A failed compressor detachment must keep the fork's compression OFF.
|
|
|
|
Regression for the #93057 adversarial review: if ``bind_session_state``
|
|
cannot sever the engine's binding to the parent's SessionDB/session_id,
|
|
enabling compression would run it against the parent's live session
|
|
binding — durable cooldown/streak/ineffective-count writes on the
|
|
parent's row and the sibling-session race behind #38727 re-opened. The
|
|
fork must fail CLOSED: keep the historical ``compression_enabled = False``
|
|
behavior and warn. The review still runs (the iteration cap and the
|
|
aggregate input budget still bound it).
|
|
"""
|
|
import agent.background_review as br
|
|
from agent.context_compressor import ContextCompressor
|
|
|
|
parent_sid = "REVIEW_FORK_REBIND_FAIL_CLOSED_93057"
|
|
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
db.create_session(parent_sid, source="discord")
|
|
parent = _build_agent_with_db(db, parent_sid)
|
|
parent._cached_system_prompt = "stable parent prompt"
|
|
|
|
snapshot = [
|
|
{
|
|
"role": "user" if i % 2 == 0 else "assistant",
|
|
"content": f"review turn {i}",
|
|
}
|
|
for i in range(8)
|
|
]
|
|
|
|
captured = {}
|
|
|
|
def _capture_fork_flags(self, *args, **kwargs):
|
|
captured["compression_enabled"] = self.compression_enabled
|
|
captured["compression_in_place"] = self.compression_in_place
|
|
captured["input_budget"] = getattr(
|
|
self, "_review_input_token_budget", "missing"
|
|
)
|
|
return {
|
|
"completed": True,
|
|
"final_response": "review complete",
|
|
"api_call_count": 0,
|
|
}
|
|
|
|
# The worker does a local ``from run_agent import AIAgent``; patching the
|
|
# class method covers that import path.
|
|
from run_agent import AIAgent
|
|
|
|
_real_bind = ContextCompressor.bind_session_state
|
|
|
|
def _failing_bind(self, session_db=None, session_id=""):
|
|
# Only the detachment rebind may fail; any other binding passes
|
|
# through so the fork's construction path stays intact.
|
|
if session_db is None:
|
|
raise RuntimeError("detachment boom")
|
|
return _real_bind(self, session_db, session_id)
|
|
|
|
try:
|
|
with (
|
|
patch.object(AIAgent, "run_conversation", _capture_fork_flags),
|
|
patch.object(
|
|
ContextCompressor, "bind_session_state", _failing_bind
|
|
),
|
|
):
|
|
with caplog.at_level(logging.WARNING, logger="agent.background_review"):
|
|
br._run_review_in_thread(parent, snapshot, "review this conversation")
|
|
|
|
assert captured["compression_enabled"] is False, (
|
|
"FIX REGRESSION: a failed compressor rebind must leave "
|
|
"compression_enabled False on the review fork (fail-closed). "
|
|
"Enabling compression with the engine still bound to the "
|
|
"parent's session re-opens the #38727 race (issue #93057)."
|
|
)
|
|
assert any(
|
|
"detachment failed" in record.message for record in caplog.records
|
|
), (
|
|
"the failed rebind must log a warning so operators can see the "
|
|
"fork fell back to the pre-fix behavior"
|
|
)
|
|
assert (
|
|
isinstance(captured["input_budget"], int) and captured["input_budget"] > 0
|
|
), (
|
|
"the aggregate input budget must still be armed on the fail-closed "
|
|
"path — it bounds the review even when compaction stays off"
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
# ── Lease-refresher bounded-failure tolerance (salvage follow-up, #54465) ────
|
|
# A single falsy refresh (transient DB blip) must NOT permanently kill the
|
|
# lease — only a *persistent* failure (genuine lost-ownership) should stop the
|
|
# refresher after a bounded number of consecutive failures. Without this, one
|
|
# escaped lock-contention error silently reintroduces the TTL-expiry wedge the
|
|
# PR set out to fix.
|
|
|
|
|
|
class _FlakyRefreshDB:
|
|
"""A db whose refresh_compression_lock returns a scripted sequence."""
|
|
|
|
def __init__(self, results):
|
|
self._results = list(results)
|
|
self.calls = 0
|
|
|
|
def refresh_compression_lock(self, session_id, holder, ttl_seconds=300.0):
|
|
self.calls += 1
|
|
if self._results:
|
|
return self._results.pop(0)
|
|
return True # steady-state success after the scripted prefix
|
|
|
|
|
|
def _no_sleep(refresher) -> None:
|
|
"""Make the refresher loop iterate without real wall-clock sleeps.
|
|
|
|
``_stop.wait(interval)`` returns False (keep looping) instantly instead of
|
|
blocking for the (clamped) interval, so count-based tests stay fast and
|
|
deterministic — the loop's termination is driven by the failure cap / the
|
|
scripted db, not by timing.
|
|
"""
|
|
refresher._stop.wait = lambda _interval: False # type: ignore[assignment]
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_lease_refresher_failure_window_is_bounded_by_ttl() -> None:
|
|
"""Persistent failure stops within one lease's worth of time, not forever.
|
|
|
|
The contract (not a magic count): the give-up window
|
|
``cap * refresh_interval`` must be <= the TTL, so a stuck refresher can
|
|
never hold the lock past its TTL. We assert that relationship directly
|
|
rather than freezing a literal cap (behavior contract over snapshot).
|
|
"""
|
|
from agent.conversation_compression import _CompressionLockLeaseRefresher
|
|
|
|
ttl, interval = 10.0, 2.0 # cap should be int(10/2) = 5
|
|
db = _FlakyRefreshDB([False] * 50) # never recovers (lost ownership)
|
|
refresher = _CompressionLockLeaseRefresher(
|
|
db, "sess", "holder", ttl_seconds=ttl, refresh_interval_seconds=interval
|
|
)
|
|
_no_sleep(refresher)
|
|
refresher._run()
|
|
|
|
cap = refresher._max_consecutive_failures
|
|
assert cap == int(ttl / interval), "cap must derive from ttl/interval"
|
|
# Stops at the cap — not on the first failure, not forever.
|
|
assert db.calls == cap
|
|
# The invariant that makes the cap honest: total tolerance <= one TTL.
|
|
assert cap * interval <= ttl, (
|
|
f"give-up window {cap * interval}s must not exceed the lease TTL {ttl}s"
|
|
)
|
|
|
|
|
|
def test_hard_interrupt_aborts_compression_and_unblocks_session_writes(tmp_path: Path) -> None:
|
|
"""Ctrl+C must abort an interrupt-protected summary without leaving the
|
|
session write-blocked behind its compression lease."""
|
|
from agent import auxiliary_client as aux
|
|
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
session_id = "HARD_INTERRUPT_COMPRESSION_TEST"
|
|
db.create_session(session_id, source="cli")
|
|
|
|
agent = _build_agent_with_db(db, session_id)
|
|
messages = [{"role": "user", "content": f"m{i}"} for i in range(20)]
|
|
original_messages = copy.deepcopy(messages)
|
|
|
|
def _cancelled_compress(current, *_args, **_kwargs):
|
|
agent._hard_interrupt_requested.set()
|
|
assert aux._aux_interrupt_cancel_requested() is True
|
|
# F3 isolation: the engine only ever sees the pooled snapshot, so an
|
|
# in-place mutation lands on ``current`` (the worker's copy), never on
|
|
# the caller's live list. The rollback contract is that the RETURNED
|
|
# transcript equals the pre-compression one.
|
|
current[0]["content"] = "must be rolled back"
|
|
raise aux.AuxiliaryExplicitCancellation()
|
|
|
|
agent.context_compressor.compress.side_effect = _cancelled_compress
|
|
|
|
compressed, _prompt = agent._compress_context(
|
|
messages, "sys", approx_tokens=120_000
|
|
)
|
|
|
|
assert compressed == original_messages
|
|
assert messages == original_messages
|
|
assert db.get_compression_lock_holder(session_id) is None
|
|
db.append_message(session_id, "assistant", "writes recovered")
|
|
|
|
|
|
def test_late_hard_interrupt_restores_full_compressor_attempt_state_and_retry(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""A stop after provider success but before compress() returns is a true no-op."""
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
session_id = "LATE_HARD_INTERRUPT_STATE_TEST"
|
|
db.create_session(session_id, source="cli")
|
|
agent = _build_agent_with_db(db, session_id)
|
|
agent.compression_in_place = True
|
|
agent._cached_system_prompt = "sys"
|
|
messages = [{"role": "user", "content": f"m{i}"} for i in range(20)]
|
|
provider_returned = threading.Event()
|
|
allow_compress_return = threading.Event()
|
|
shared_telemetry = {"shared": [1, 2, 3]}
|
|
state_fields = {
|
|
"_previous_summary": "old-summary",
|
|
"_summary_has_user_turn": False,
|
|
"compression_count": 4,
|
|
"_last_compression_savings_pct": 37.5,
|
|
"_ineffective_compression_count": 1,
|
|
"_anti_thrash_recovery_deadline": 123.0,
|
|
"_fallback_compression_streak": 1,
|
|
"_verify_compaction_cleared_threshold": False,
|
|
"_last_compression_made_progress": False,
|
|
"_summary_failure_cooldown_until": 456.0,
|
|
"_cooldown_persist_failed": True,
|
|
"_last_summary_error": "old-error",
|
|
"_consecutive_timeout_failures": 2,
|
|
"_last_summary_dropped_count": 3,
|
|
"_last_summary_fallback_used": True,
|
|
"_last_compress_aborted": False,
|
|
"_last_summary_auth_failure": True,
|
|
"_last_summary_network_failure": True,
|
|
"_last_aux_model_failure_error": "old-aux-error",
|
|
"_last_aux_model_failure_model": "old-aux-model",
|
|
"_summary_model_fallen_back": True,
|
|
"summary_model": "old-summary-model",
|
|
"_last_compression_telemetry": shared_telemetry,
|
|
"_active_compression_telemetry": shared_telemetry,
|
|
"_compression_telemetry_seed": {"seed": [3]},
|
|
}
|
|
for name, value in state_fields.items():
|
|
setattr(agent.context_compressor, name, copy.deepcopy(value))
|
|
restored_shared_telemetry = copy.deepcopy(shared_telemetry)
|
|
agent.context_compressor._last_compression_telemetry = restored_shared_telemetry
|
|
agent.context_compressor._active_compression_telemetry = restored_shared_telemetry
|
|
|
|
def _provider_succeeded_then_waits(*_args, **_kwargs):
|
|
for name in state_fields:
|
|
setattr(agent.context_compressor, name, f"mutated-{name}")
|
|
provider_returned.set()
|
|
assert allow_compress_return.wait(timeout=5)
|
|
return [
|
|
{"role": "user", "content": "[CONTEXT COMPACTION] cancelled summary"},
|
|
{"role": "user", "content": "tail"},
|
|
]
|
|
|
|
agent.context_compressor.compress.side_effect = _provider_succeeded_then_waits
|
|
result: dict[str, tuple] = {}
|
|
worker = threading.Thread(
|
|
target=lambda: result.setdefault(
|
|
"value", agent._compress_context(messages, "sys", approx_tokens=120_000)
|
|
),
|
|
daemon=True,
|
|
)
|
|
worker.start()
|
|
assert provider_returned.wait(timeout=2)
|
|
agent.hard_interrupt("cancel after provider return")
|
|
allow_compress_return.set()
|
|
worker.join(timeout=5)
|
|
|
|
assert not worker.is_alive()
|
|
assert result["value"][0] is messages
|
|
assert {
|
|
name: copy.deepcopy(getattr(agent.context_compressor, name))
|
|
for name in state_fields
|
|
} == state_fields
|
|
assert (
|
|
agent.context_compressor._active_compression_telemetry
|
|
is agent.context_compressor._last_compression_telemetry
|
|
)
|
|
assert db.get_compression_lock_holder(session_id) is None
|
|
|
|
agent.clear_interrupt()
|
|
agent.context_compressor.compress.side_effect = lambda *_a, **_kw: [
|
|
{"role": "user", "content": "[CONTEXT COMPACTION] retry summary"},
|
|
{"role": "user", "content": "tail"},
|
|
]
|
|
retried, _prompt = agent._compress_context(
|
|
messages, "sys", approx_tokens=120_000
|
|
)
|
|
assert retried is not messages
|
|
assert retried[0]["content"] == "[CONTEXT COMPACTION] retry summary"
|
|
|
|
|
|
def test_force_cancel_restores_newer_durable_cooldown_captured_under_lease(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""A stale forced attempt rolls back to the lease-protected durable row."""
|
|
from agent.auxiliary_client import AuxiliaryExplicitCancellation
|
|
from agent.context_compressor import ContextCompressor
|
|
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
session_id = "FORCE_CANCEL_DURABLE_COOLDOWN"
|
|
db.create_session(session_id, source="cli")
|
|
|
|
# B binds first and therefore has no local cooldown. A then persists a
|
|
# newer cooldown for the same durable session before B acquires its lease.
|
|
stale_agent = _build_agent_with_db(
|
|
db, session_id, stub_compressor=False
|
|
)
|
|
writer_agent = _build_agent_with_db(
|
|
db, session_id, stub_compressor=False
|
|
)
|
|
stale = stale_agent.context_compressor
|
|
writer = writer_agent.context_compressor
|
|
assert isinstance(stale, ContextCompressor)
|
|
assert isinstance(writer, ContextCompressor)
|
|
assert stale._summary_failure_cooldown_until == 0.0
|
|
|
|
writer._record_compression_failure_cooldown(120.0, "newer durable failure")
|
|
durable_before = tuple(
|
|
db._conn.execute(
|
|
"SELECT compression_failure_cooldown_until, compression_failure_error "
|
|
"FROM sessions WHERE id = ?",
|
|
(session_id,),
|
|
).fetchone()
|
|
)
|
|
assert durable_before[0] is not None
|
|
|
|
stale_seed = {"seed": ["truly-pre-attempt"]}
|
|
stale._compression_telemetry_seed = copy.deepcopy(stale_seed)
|
|
stale._previous_summary = "pre-attempt-summary"
|
|
stale_agent._compression_feasibility_checked = True
|
|
stale_agent.compression_in_place = True
|
|
messages = [{"role": "user", "content": f"m{i}"} for i in range(20)]
|
|
|
|
real_clear = ContextCompressor._clear_compression_failure_cooldown
|
|
|
|
def _clear_then_hard_cancel() -> None:
|
|
real_clear(stale)
|
|
stale_agent._hard_interrupt_requested.set()
|
|
raise AuxiliaryExplicitCancellation()
|
|
|
|
# Exercise the built-in force=True mutation point deterministically: force
|
|
# clears the durable cooldown, then the frozen host cancellation unwinds it.
|
|
stale._clear_compression_failure_cooldown = _clear_then_hard_cancel
|
|
|
|
compressed, _prompt = stale_agent._compress_context(
|
|
messages,
|
|
"sys",
|
|
approx_tokens=120_000,
|
|
force=True,
|
|
)
|
|
|
|
assert compressed is messages
|
|
durable_after = tuple(
|
|
db._conn.execute(
|
|
"SELECT compression_failure_cooldown_until, compression_failure_error "
|
|
"FROM sessions WHERE id = ?",
|
|
(session_id,),
|
|
).fetchone()
|
|
)
|
|
assert durable_after == durable_before
|
|
assert stale._summary_failure_cooldown_until > time.monotonic()
|
|
assert stale._last_summary_error == "newer durable failure"
|
|
assert stale._cooldown_persist_failed is False
|
|
assert stale._compression_telemetry_seed == stale_seed
|
|
assert stale._previous_summary == "pre-attempt-summary"
|
|
assert db.get_compression_lock_holder(session_id) is None
|
|
|
|
# A future compressor refresh must still observe the exact row rather than
|
|
# the cancelled force attempt having permanently cleared it.
|
|
future_agent = _build_agent_with_db(
|
|
db, session_id, stub_compressor=False
|
|
)
|
|
future = future_agent.context_compressor.get_active_compression_failure_cooldown(
|
|
refresh=True
|
|
)
|
|
assert future is not None
|
|
assert future["error"] == "newer durable failure"
|
|
|
|
|
|
def test_unrelated_interrupted_error_propagates_and_releases_compression_lease(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""A plugin/OS InterruptedError is a failure, not an explicit transaction abort."""
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
session_id = "UNRELATED_INTERRUPT_COMPRESSION_TEST"
|
|
db.create_session(session_id, source="cli")
|
|
|
|
agent = _build_agent_with_db(db, session_id)
|
|
messages = [{"role": "user", "content": f"m{i}"} for i in range(20)]
|
|
|
|
def _provider_interrupted(*_args, **_kwargs):
|
|
messages[0]["content"] = "must be rolled back"
|
|
raise InterruptedError("provider syscall interrupted")
|
|
|
|
agent.context_compressor.compress.side_effect = _provider_interrupted
|
|
|
|
with pytest.raises(InterruptedError, match="provider syscall interrupted"):
|
|
agent._compress_context(messages, "sys", approx_tokens=120_000)
|
|
|
|
assert db.get_compression_lock_holder(session_id) is None
|
|
db.append_message(session_id, "assistant", "writes recovered")
|
|
|
|
|
|
def test_redirect_interrupt_remains_protected_during_compression(tmp_path: Path) -> None:
|
|
"""Redirects use interrupt_requested=True/message=None; only the atomic
|
|
hard-cancel event may override summary protection."""
|
|
from agent import auxiliary_client as aux
|
|
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
session_id = "REDIRECT_COMPRESSION_TEST"
|
|
db.create_session(session_id, source="cli")
|
|
agent = _build_agent_with_db(db, session_id)
|
|
agent._interrupt_requested = True
|
|
agent._interrupt_message = None
|
|
agent._pending_redirect = "new correction"
|
|
messages = [{"role": "user", "content": f"m{i}"} for i in range(20)]
|
|
|
|
def _protected_noop(current, **_kwargs):
|
|
assert aux._aux_interrupt_cancel_requested() is False
|
|
return copy.deepcopy(current)
|
|
|
|
agent.context_compressor.compress.side_effect = _protected_noop
|
|
|
|
compressed, _prompt = agent._compress_context(
|
|
messages, "sys", approx_tokens=120_000
|
|
)
|
|
|
|
assert compressed == messages
|
|
assert db.get_compression_lock_holder(session_id) is None
|
|
|
|
|
|
def test_hard_cancel_between_compress_return_and_commit_begin_wins_atomically(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""The hard-stop admission and commit admission share one fence lock."""
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
session_id = "HARD_CANCEL_COMMIT_RACE"
|
|
db.create_session(session_id, source="tui")
|
|
agent = _build_agent_with_db(db, session_id)
|
|
agent.compression_in_place = True
|
|
agent._cached_system_prompt = "sys"
|
|
messages = [{"role": "user", "content": f"m{i}"} for i in range(20)]
|
|
before_commit = threading.Event()
|
|
allow_commit_check = threading.Event()
|
|
|
|
class _CommitBarrierList(list):
|
|
def __eq__(self, other):
|
|
before_commit.set()
|
|
assert allow_commit_check.wait(timeout=5)
|
|
return super().__eq__(other)
|
|
|
|
agent.context_compressor.compress.side_effect = lambda *_a, **_kw: _CommitBarrierList(
|
|
[
|
|
{"role": "user", "content": "[CONTEXT COMPACTION] summary"},
|
|
{"role": "user", "content": "tail"},
|
|
]
|
|
)
|
|
archive_spy = MagicMock(wraps=db.archive_and_compact)
|
|
db.archive_and_compact = archive_spy
|
|
result: dict[str, tuple] = {}
|
|
worker = threading.Thread(
|
|
target=lambda: result.setdefault(
|
|
"value", agent._compress_context(messages, "sys", approx_tokens=120_000)
|
|
),
|
|
daemon=True,
|
|
)
|
|
worker.start()
|
|
assert before_commit.wait(timeout=2)
|
|
|
|
agent.hard_interrupt("cancel before commit admission")
|
|
allow_commit_check.set()
|
|
worker.join(timeout=5)
|
|
|
|
assert not worker.is_alive()
|
|
assert result["value"][0] is messages
|
|
archive_spy.assert_not_called()
|
|
assert db.get_compression_lock_holder(session_id) is None
|
|
|
|
|
|
def test_hard_stop_waits_for_commit_already_admitted(tmp_path: Path) -> None:
|
|
"""A surfaced stop never races an untracked post-return transcript commit."""
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
session_id = "HARD_CANCEL_AFTER_COMMIT_ADMISSION"
|
|
db.create_session(session_id, source="tui")
|
|
agent = _build_agent_with_db(db, session_id)
|
|
agent.compression_in_place = True
|
|
agent._cached_system_prompt = "sys"
|
|
messages = [{"role": "user", "content": f"m{i}"} for i in range(20)]
|
|
commit_started = threading.Event()
|
|
allow_commit = threading.Event()
|
|
stop_returned = threading.Event()
|
|
original_archive = db.archive_and_compact
|
|
|
|
def _blocked_archive(*args, **kwargs):
|
|
commit_started.set()
|
|
assert allow_commit.wait(timeout=5)
|
|
return original_archive(*args, **kwargs)
|
|
|
|
db.archive_and_compact = _blocked_archive
|
|
agent.context_compressor.compress.side_effect = lambda *_a, **_kw: [
|
|
{"role": "user", "content": "[CONTEXT COMPACTION] summary"},
|
|
{"role": "user", "content": "tail"},
|
|
]
|
|
compression_result: dict[str, tuple] = {}
|
|
compression = threading.Thread(
|
|
target=lambda: compression_result.setdefault(
|
|
"value", agent._compress_context(messages, "sys", approx_tokens=120_000)
|
|
),
|
|
daemon=True,
|
|
)
|
|
compression.start()
|
|
assert commit_started.wait(timeout=2)
|
|
|
|
stop = threading.Thread(
|
|
target=lambda: (
|
|
agent.hard_interrupt("stop after commit admission"),
|
|
stop_returned.set(),
|
|
),
|
|
daemon=True,
|
|
)
|
|
stop.start()
|
|
assert not stop_returned.wait(timeout=0.1)
|
|
allow_commit.set()
|
|
compression.join(timeout=5)
|
|
stop.join(timeout=5)
|
|
|
|
assert not compression.is_alive()
|
|
assert not stop.is_alive()
|
|
assert stop_returned.is_set()
|
|
assert compression_result["value"][0][0]["content"] == (
|
|
"[CONTEXT COMPACTION] summary"
|
|
)
|
|
assert agent._hard_interrupt_requested.is_set()
|
|
assert db.get_compression_lock_holder(session_id) is None
|
|
|
|
|
|
@pytest.mark.parametrize("deadline_offset", [-10.0, 0.05, None])
|
|
def test_force_cancel_restores_exact_expired_or_expiring_cooldown_row(
|
|
tmp_path: Path,
|
|
deadline_offset: float | None,
|
|
) -> None:
|
|
"""Cancellation preserves raw cooldown columns even after their deadline."""
|
|
from agent.auxiliary_client import AuxiliaryExplicitCancellation
|
|
from agent.context_compressor import ContextCompressor
|
|
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
session_id = f"RAW_COOLDOWN_{deadline_offset}"
|
|
db.create_session(session_id, source="cli")
|
|
deadline = time.time() + deadline_offset if deadline_offset is not None else None
|
|
db.restore_compression_failure_cooldown_row(
|
|
session_id,
|
|
{
|
|
"session_exists": True,
|
|
"cooldown_until": deadline,
|
|
"error": "expired-but-exact",
|
|
},
|
|
)
|
|
before = db.get_compression_failure_cooldown_row(session_id)
|
|
|
|
agent = _build_agent_with_db(db, session_id, stub_compressor=False)
|
|
compressor = agent.context_compressor
|
|
assert isinstance(compressor, ContextCompressor)
|
|
# A stale local persistence-failure marker must not suppress restoration
|
|
# once the raw durable row was captured authoritatively under the lease.
|
|
compressor._cooldown_persist_failed = True
|
|
agent._compression_feasibility_checked = True
|
|
agent.compression_in_place = True
|
|
messages = [{"role": "user", "content": f"m{i}"} for i in range(20)]
|
|
real_clear = ContextCompressor._clear_compression_failure_cooldown
|
|
|
|
def _mutate_then_cancel() -> None:
|
|
real_clear(compressor)
|
|
if deadline_offset is not None and deadline_offset > 0:
|
|
assert deadline is not None
|
|
while time.time() <= deadline:
|
|
time.sleep(0.005)
|
|
agent._hard_interrupt_requested.set()
|
|
raise AuxiliaryExplicitCancellation()
|
|
|
|
compressor._clear_compression_failure_cooldown = _mutate_then_cancel
|
|
|
|
compressed, _prompt = agent._compress_context(
|
|
messages,
|
|
"sys",
|
|
approx_tokens=120_000,
|
|
force=True,
|
|
)
|
|
|
|
assert compressed is messages
|
|
assert db.get_compression_failure_cooldown_row(session_id) == before
|
|
assert db.get_compression_lock_holder(session_id) is None
|
|
|
|
|
|
def test_cooldown_rollback_failure_surfaces_and_releases_lease(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""A failed compensating write cannot masquerade as a mutation-free cancel."""
|
|
from agent.auxiliary_client import AuxiliaryExplicitCancellation
|
|
from agent.context_compressor import ContextCompressor
|
|
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
session_id = "COOLDOWN_ROLLBACK_WRITE_FAILURE"
|
|
db.create_session(session_id, source="cli")
|
|
db.record_compression_failure_cooldown(
|
|
session_id,
|
|
time.time() + 120.0,
|
|
"must-restore",
|
|
)
|
|
agent = _build_agent_with_db(db, session_id, stub_compressor=False)
|
|
compressor = agent.context_compressor
|
|
assert isinstance(compressor, ContextCompressor)
|
|
agent._compression_feasibility_checked = True
|
|
agent.compression_in_place = True
|
|
messages = [{"role": "user", "content": f"m{i}"} for i in range(20)]
|
|
real_clear = ContextCompressor._clear_compression_failure_cooldown
|
|
|
|
def _mutate_then_cancel() -> None:
|
|
real_clear(compressor)
|
|
agent._hard_interrupt_requested.set()
|
|
raise AuxiliaryExplicitCancellation()
|
|
|
|
compressor._clear_compression_failure_cooldown = _mutate_then_cancel
|
|
|
|
def _rollback_write_fails(_self, _session_id, _snapshot) -> None:
|
|
raise sqlite3.OperationalError("forced rollback write failure")
|
|
|
|
monkeypatch.setattr(
|
|
SessionDB,
|
|
"restore_compression_failure_cooldown_row",
|
|
_rollback_write_fails,
|
|
)
|
|
|
|
with pytest.raises(sqlite3.OperationalError, match="forced rollback write failure"):
|
|
agent._compress_context(
|
|
messages,
|
|
"sys",
|
|
approx_tokens=120_000,
|
|
force=True,
|
|
)
|
|
|
|
assert db.get_compression_lock_holder(session_id) is None
|
|
|
|
|
|
def test_exact_cooldown_restore_api_propagates_sqlite_write_failure(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
session_id = "RAW_COOLDOWN_WRITE_FAILURE"
|
|
db.create_session(session_id, source="test")
|
|
|
|
def _write_fails(_callback) -> None:
|
|
raise sqlite3.OperationalError("forced low-level write failure")
|
|
|
|
monkeypatch.setattr(db, "_execute_write", _write_fails)
|
|
|
|
with pytest.raises(sqlite3.OperationalError, match="forced low-level write failure"):
|
|
db.restore_compression_failure_cooldown_row(
|
|
session_id,
|
|
{
|
|
"session_exists": True,
|
|
"cooldown_until": time.time() + 10.0,
|
|
"error": "must propagate",
|
|
},
|
|
)
|
|
|
|
|
|
def test_failed_split_arms_failure_cooldown(tmp_path: Path) -> None:
|
|
"""Regression #97948 symptom B: a failed split/archive must arm the
|
|
compression failure cooldown so the next automatic turn cannot
|
|
immediately re-run the identical doomed compression."""
|
|
from agent.conversation_compression import _SPLIT_FAILURE_COOLDOWN_SECONDS
|
|
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
session_id = "SPLIT_FAIL_COOLDOWN_TEST"
|
|
db.create_session(session_id, source="test")
|
|
agent = _build_agent_with_db(db, session_id)
|
|
setattr(agent, "compression_in_place", True)
|
|
db.archive_and_compact = MagicMock(side_effect=RuntimeError("archive boom"))
|
|
messages = [{"role": "user", "content": f"m{i}"} for i in range(20)]
|
|
|
|
agent._compress_context(
|
|
messages,
|
|
"sys",
|
|
approx_tokens=120_000,
|
|
force=True,
|
|
)
|
|
|
|
cooldown_calls = (
|
|
agent.context_compressor._record_compression_failure_cooldown.call_args_list
|
|
)
|
|
assert len(cooldown_calls) == 1, (
|
|
"split failure must arm the failure cooldown (#97948 symptom B)"
|
|
)
|
|
seconds, error = cooldown_calls[0].args
|
|
assert seconds == _SPLIT_FAILURE_COOLDOWN_SECONDS
|
|
assert "session_split_failed" in str(error)
|