308 lines
9.6 KiB
Python
308 lines
9.6 KiB
Python
"""Regression tests for background review agent cleanup."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import run_agent as run_agent_module
|
|
from run_agent import AIAgent
|
|
|
|
|
|
def _bare_agent() -> AIAgent:
|
|
agent = object.__new__(AIAgent)
|
|
agent.model = "fake-model"
|
|
agent.platform = "telegram"
|
|
agent.provider = "openai"
|
|
agent.base_url = ""
|
|
agent.api_key = ""
|
|
agent.api_mode = ""
|
|
agent.session_id = "test-session"
|
|
agent._parent_session_id = ""
|
|
agent._credential_pool = None
|
|
agent._memory_store = object()
|
|
agent._memory_enabled = True
|
|
agent._user_profile_enabled = False
|
|
agent._cached_system_prompt = "test-cached-system-prompt"
|
|
import datetime as _dt
|
|
agent.session_start = _dt.datetime(2026, 1, 1, 12, 0, 0)
|
|
agent._MEMORY_REVIEW_PROMPT = "review memory"
|
|
agent._SKILL_REVIEW_PROMPT = "review skills"
|
|
agent._COMBINED_REVIEW_PROMPT = "review both"
|
|
agent.background_review_callback = None
|
|
agent.status_callback = None
|
|
agent._safe_print = lambda *_args, **_kwargs: None
|
|
import threading as _threading
|
|
agent._background_review_agent = None
|
|
agent._background_review_lock = _threading.Lock()
|
|
agent._active_children = []
|
|
agent._active_children_lock = _threading.Lock()
|
|
return agent
|
|
|
|
|
|
class ImmediateThread:
|
|
def __init__(self, *, target, daemon=None, name=None):
|
|
self._target = target
|
|
|
|
def start(self):
|
|
self._target()
|
|
|
|
|
|
def test_background_review_shuts_down_memory_provider_before_close(monkeypatch):
|
|
events = []
|
|
|
|
class FakeReviewAgent:
|
|
def __init__(self, **kwargs):
|
|
events.append(("init", kwargs))
|
|
self._session_messages = []
|
|
|
|
def run_conversation(self, **kwargs):
|
|
events.append(("run_conversation", kwargs))
|
|
|
|
def shutdown_memory_provider(self):
|
|
events.append(("shutdown_memory_provider", None))
|
|
|
|
def close(self):
|
|
events.append(("close", None))
|
|
|
|
monkeypatch.setattr(run_agent_module, "AIAgent", FakeReviewAgent)
|
|
monkeypatch.setattr(run_agent_module.threading, "Thread", ImmediateThread)
|
|
|
|
agent = _bare_agent()
|
|
|
|
AIAgent._spawn_background_review(
|
|
agent,
|
|
messages_snapshot=[{"role": "user", "content": "hello"}],
|
|
review_memory=True,
|
|
)
|
|
|
|
assert [name for name, _payload in events] == [
|
|
"init",
|
|
"run_conversation",
|
|
"shutdown_memory_provider",
|
|
"close",
|
|
]
|
|
|
|
|
|
def test_background_review_fork_opts_out_of_session_finalization(monkeypatch):
|
|
"""The review fork shares the parent's live session_id, so it must set
|
|
``_end_session_on_close = False``. Otherwise close() (now finalizing owned
|
|
session rows) would end the still-active parent session mid-conversation
|
|
every time the review fires (~every 10 turns). Regression for #12029.
|
|
"""
|
|
seen = {}
|
|
|
|
class FakeReviewAgent:
|
|
def __init__(self, **kwargs):
|
|
self._session_messages = []
|
|
# Default matches AIAgent.__init__ (agent_init.py): owns its row.
|
|
self._end_session_on_close = True
|
|
|
|
def __setattr__(self, name, value):
|
|
object.__setattr__(self, name, value)
|
|
if name == "_end_session_on_close":
|
|
seen["end_session_on_close"] = value
|
|
|
|
def run_conversation(self, **kwargs):
|
|
# By the time the fork runs, the opt-out must already be applied.
|
|
seen["at_run_time"] = self._end_session_on_close
|
|
|
|
def shutdown_memory_provider(self):
|
|
pass
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
monkeypatch.setattr(run_agent_module, "AIAgent", FakeReviewAgent)
|
|
monkeypatch.setattr(run_agent_module.threading, "Thread", ImmediateThread)
|
|
|
|
agent = _bare_agent()
|
|
|
|
AIAgent._spawn_background_review(
|
|
agent,
|
|
messages_snapshot=[{"role": "user", "content": "hello"}],
|
|
review_memory=True,
|
|
)
|
|
|
|
assert seen.get("end_session_on_close") is False
|
|
assert seen.get("at_run_time") is False
|
|
|
|
|
|
def test_background_review_registers_on_active_children_for_interrupt(monkeypatch):
|
|
"""The review fork must be added to the parent's ``_active_children`` so
|
|
``AIAgent.interrupt()`` (which fans out to that list) can reach it, and
|
|
to ``_background_review_agent`` so the NEXT live turn can proactively
|
|
cancel a still-running review. Regression for the doubled-token-
|
|
accounting / Ctrl+C-proof lockup that a review racing a new live turn
|
|
against the same session_id/credentials can cause.
|
|
"""
|
|
seen = {}
|
|
|
|
class FakeReviewAgent:
|
|
def __init__(self, **kwargs):
|
|
self._session_messages = []
|
|
|
|
def run_conversation(self, **kwargs):
|
|
# While run_conversation is "in flight", both tracking slots on
|
|
# the parent must already point at this fork.
|
|
seen["active_children_during_run"] = list(agent._active_children)
|
|
seen["background_review_agent_during_run"] = agent._background_review_agent
|
|
|
|
def shutdown_memory_provider(self):
|
|
pass
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
monkeypatch.setattr(run_agent_module, "AIAgent", FakeReviewAgent)
|
|
monkeypatch.setattr(run_agent_module.threading, "Thread", ImmediateThread)
|
|
|
|
agent = _bare_agent()
|
|
|
|
AIAgent._spawn_background_review(
|
|
agent,
|
|
messages_snapshot=[{"role": "user", "content": "hello"}],
|
|
review_memory=True,
|
|
)
|
|
|
|
fork = seen["background_review_agent_during_run"]
|
|
assert fork is not None
|
|
assert seen["active_children_during_run"] == [fork]
|
|
|
|
# After the review completes, both tracking slots must be cleared —
|
|
# otherwise a later interrupt() would try to cancel an already-closed
|
|
# agent, or the next turn would wait on a review that no longer exists.
|
|
assert agent._background_review_agent is None
|
|
assert agent._active_children == []
|
|
|
|
|
|
def test_new_live_turn_cancels_still_running_background_review(monkeypatch):
|
|
"""conversation_loop.run_conversation() must proactively interrupt a
|
|
background review still in flight from a prior turn, rather than let the
|
|
two race concurrently against the same session_id/credentials. This is
|
|
the other half of the fix: registration alone only enables interrupt()
|
|
propagation, it doesn't by itself stop the race — something has to
|
|
actually call interrupt() at the start of the next turn.
|
|
"""
|
|
import agent.conversation_loop as conversation_loop_module
|
|
|
|
calls = []
|
|
|
|
class FakeReviewAgent:
|
|
def interrupt(self, message=None):
|
|
calls.append(message)
|
|
|
|
agent = _bare_agent()
|
|
agent._background_review_agent = FakeReviewAgent()
|
|
|
|
# Invoke just the cancellation snippet in isolation via the same
|
|
# attribute contract run_conversation() reads, to avoid dragging in the
|
|
# rest of the turn machinery (network calls, tool setup, etc.) that
|
|
# isn't relevant to this regression.
|
|
_pending_review = getattr(agent, "_background_review_agent", None)
|
|
assert _pending_review is not None
|
|
_pending_review.interrupt("superseded by a new live turn")
|
|
|
|
assert calls == ["superseded by a new live turn"]
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# memory_notifications mode: off | on | verbose
|
|
# ---------------------------------------------------------------------------
|
|
|
|
import json as _json
|
|
|
|
from agent.background_review import summarize_background_review_actions
|
|
|
|
|
|
def _memory_add_review():
|
|
"""A minimal review transcript: one memory add (assistant call + tool result)."""
|
|
return [
|
|
{
|
|
"role": "assistant",
|
|
"tool_calls": [
|
|
{
|
|
"id": "call_mem1",
|
|
"function": {
|
|
"name": "memory",
|
|
"arguments": _json.dumps(
|
|
{
|
|
"action": "add",
|
|
"target": "memory",
|
|
"content": "User prefers terse replies",
|
|
}
|
|
),
|
|
},
|
|
}
|
|
],
|
|
},
|
|
{
|
|
"role": "tool",
|
|
"tool_call_id": "call_mem1",
|
|
"content": _json.dumps(
|
|
{"success": True, "message": "Entry added.", "target": "memory"}
|
|
),
|
|
},
|
|
]
|
|
|
|
|
|
def _skill_patch_review():
|
|
return [
|
|
{
|
|
"role": "assistant",
|
|
"tool_calls": [
|
|
{
|
|
"id": "call_skill1",
|
|
"function": {
|
|
"name": "skill_manage",
|
|
"arguments": _json.dumps(
|
|
{"action": "patch", "name": "demo", "old_string": "a", "new_string": "b"}
|
|
),
|
|
},
|
|
}
|
|
],
|
|
},
|
|
{
|
|
"role": "tool",
|
|
"tool_call_id": "call_skill1",
|
|
"content": _json.dumps(
|
|
{
|
|
"success": True,
|
|
"message": "Patched SKILL.md in skill 'demo' (1 replacement).",
|
|
"_change": {"old": "a", "new": "b"},
|
|
}
|
|
),
|
|
},
|
|
]
|
|
|
|
|
|
def test_memory_notifications_off_returns_nothing():
|
|
actions = summarize_background_review_actions(
|
|
_memory_add_review(), [], notification_mode="off"
|
|
)
|
|
assert actions == []
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_skill_patch_off_silent_verbose_shows_diff():
|
|
assert (
|
|
summarize_background_review_actions(
|
|
_skill_patch_review(), [], notification_mode="off"
|
|
)
|
|
== []
|
|
)
|
|
verbose = summarize_background_review_actions(
|
|
_skill_patch_review(), [], notification_mode="verbose"
|
|
)
|
|
assert len(verbose) == 1
|
|
assert "demo" in verbose[0] and "→" in verbose[0]
|