mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-21 16:18:55 +00:00
661 lines
25 KiB
Python
661 lines
25 KiB
Python
"""Regression tests for the CLI interrupt-acknowledgement race.
|
|
|
|
Symptom (user report, July 2026): interrupting an active turn is
|
|
unreliable — the interrupt message is sometimes "vacuumed into the void".
|
|
|
|
Root cause: ``HermesCLI.chat()`` fires ``agent.interrupt(msg)`` from its
|
|
monitor loop, but only re-queued the message when the turn RESULT carried
|
|
``interrupted=True``. Two races defeat that:
|
|
|
|
1. The agent thread passes its last ``_interrupt_requested`` check (or
|
|
finishes entirely) just before the interrupt lands — the turn
|
|
completes "normally", ``finalize_turn()`` never acknowledges the
|
|
interrupt, and the user's message was silently dropped.
|
|
2. Worse, when the interrupt lands *after* ``finalize_turn()``'s
|
|
``clear_interrupt()``, the stale ``_interrupt_requested`` flag
|
|
survives on the agent and instantly aborts the NEXT turn at its
|
|
first loop check.
|
|
|
|
The fix: when ``chat()`` consumed an ``interrupt_msg`` but the result
|
|
doesn't acknowledge the interrupt, re-queue the message as the next turn
|
|
and clear the stale agent flag (only when the agent thread has exited).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib
|
|
import queue
|
|
import sys
|
|
import threading
|
|
import time
|
|
import types
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
|
|
def _make_cli():
|
|
"""Build a HermesCLI with prompt_toolkit stubbed (same pattern as
|
|
test_cli_interrupt_drain_regression.py)."""
|
|
_clean_config = {
|
|
"model": {
|
|
"default": "anthropic/claude-opus-4.6",
|
|
"base_url": "https://openrouter.ai/api/v1",
|
|
"provider": "auto",
|
|
},
|
|
"display": {"compact": False, "tool_progress": "all"},
|
|
"agent": {},
|
|
"terminal": {"env_type": "local"},
|
|
}
|
|
clean_env = {"LLM_MODEL": "", "HERMES_MAX_ITERATIONS": ""}
|
|
prompt_toolkit_stubs = {
|
|
"prompt_toolkit": MagicMock(),
|
|
"prompt_toolkit.history": MagicMock(),
|
|
"prompt_toolkit.styles": MagicMock(),
|
|
"prompt_toolkit.patch_stdout": MagicMock(),
|
|
"prompt_toolkit.application": MagicMock(),
|
|
"prompt_toolkit.layout": MagicMock(),
|
|
"prompt_toolkit.layout.processors": MagicMock(),
|
|
"prompt_toolkit.filters": MagicMock(),
|
|
"prompt_toolkit.layout.dimension": MagicMock(),
|
|
"prompt_toolkit.layout.menus": MagicMock(),
|
|
"prompt_toolkit.widgets": MagicMock(),
|
|
"prompt_toolkit.key_binding": MagicMock(),
|
|
"prompt_toolkit.completion": MagicMock(),
|
|
"prompt_toolkit.formatted_text": MagicMock(),
|
|
"prompt_toolkit.auto_suggest": MagicMock(),
|
|
}
|
|
with patch.dict(sys.modules, prompt_toolkit_stubs), patch.dict(
|
|
"os.environ", clean_env, clear=False
|
|
):
|
|
import cli as _cli_mod
|
|
|
|
_cli_mod = importlib.reload(_cli_mod)
|
|
with patch.object(_cli_mod, "get_tool_definitions", return_value=[]), patch.dict(
|
|
_cli_mod.__dict__, {"CLI_CONFIG": _clean_config}
|
|
):
|
|
return _cli_mod.HermesCLI()
|
|
|
|
|
|
class _StubAgent:
|
|
"""Agent whose turn completes WITHOUT acknowledging the interrupt."""
|
|
|
|
def __init__(self, session_id, turn_seconds=0.5):
|
|
self.session_id = session_id
|
|
self.turn_seconds = turn_seconds
|
|
self._interrupt_requested = False
|
|
self._interrupt_message = None
|
|
self._active_children = []
|
|
self.interrupt_calls = []
|
|
self.clear_calls = 0
|
|
self.max_iterations = 90
|
|
self.model = "test/model"
|
|
self.platform = "cli"
|
|
|
|
def run_conversation(self, **kwargs):
|
|
# Simulate a turn that finishes normally — it never observed the
|
|
# interrupt flag (raced past its last check).
|
|
time.sleep(self.turn_seconds)
|
|
return {
|
|
"final_response": "turn finished normally",
|
|
"messages": [
|
|
{"role": "user", "content": "original"},
|
|
{"role": "assistant", "content": "turn finished normally"},
|
|
],
|
|
"api_calls": 1,
|
|
"completed": True,
|
|
# NOTE: no "interrupted" key — the race means finalize_turn
|
|
# never saw the flag (or cleared it before it was re-set).
|
|
"partial": True, # skip auto-title thread in the test
|
|
# Skip the Rich Panel rendering path (crashes under the
|
|
# prompt_toolkit/skin mocks; irrelevant to this regression).
|
|
"response_previewed": True,
|
|
}
|
|
|
|
def interrupt(self, message=None):
|
|
self.interrupt_calls.append(message)
|
|
self._interrupt_requested = True
|
|
self._interrupt_message = message
|
|
|
|
def clear_interrupt(self):
|
|
self.clear_calls += 1
|
|
self._interrupt_requested = False
|
|
self._interrupt_message = None
|
|
|
|
|
|
def test_unacknowledged_interrupt_message_is_requeued_not_dropped():
|
|
cli = _make_cli()
|
|
agent = _StubAgent(cli.session_id)
|
|
cli.agent = agent
|
|
|
|
cli._interrupt_queue = queue.Queue()
|
|
cli._pending_input = queue.Queue()
|
|
cli._interrupt_queue.put("urgent new message")
|
|
|
|
with patch.object(cli, "_ensure_runtime_credentials", return_value=True), \
|
|
patch.object(cli, "_resolve_turn_agent_config", return_value={
|
|
"signature": cli._active_agent_route_signature,
|
|
"model": None, "runtime": None, "request_overrides": None,
|
|
}), \
|
|
patch.object(cli, "_init_agent", return_value=True):
|
|
cli.chat("original")
|
|
|
|
# The interrupt fired against the agent...
|
|
assert agent.interrupt_calls == ["urgent new message"]
|
|
# ...the turn result never acknowledged it, so the message must be
|
|
# re-queued as the next turn instead of dropped.
|
|
queued = []
|
|
while not cli._pending_input.empty():
|
|
queued.append(cli._pending_input.get_nowait())
|
|
assert any("urgent new message" in str(q) for q in queued), (
|
|
f"interrupt message was dropped; pending_input={queued!r}"
|
|
)
|
|
# ...and the stale flag must be cleared so the NEXT turn doesn't
|
|
# instantly self-abort at its first _interrupt_requested check.
|
|
assert agent._interrupt_requested is False
|
|
assert agent.clear_calls >= 1
|
|
|
|
|
|
def test_acknowledged_interrupt_still_requeues_message():
|
|
"""The pre-existing path (result carries interrupted=True) still works."""
|
|
cli = _make_cli()
|
|
|
|
class _AckAgent(_StubAgent):
|
|
def run_conversation(self, **kwargs):
|
|
# Wait until the monitor loop delivers the interrupt.
|
|
for _ in range(100):
|
|
if self._interrupt_requested:
|
|
break
|
|
time.sleep(0.05)
|
|
return {
|
|
"final_response": "partial work",
|
|
"messages": [{"role": "assistant", "content": "partial work"}],
|
|
"api_calls": 1,
|
|
"completed": False,
|
|
"interrupted": True,
|
|
"interrupt_message": self._interrupt_message,
|
|
"partial": True,
|
|
}
|
|
|
|
agent = _AckAgent(cli.session_id)
|
|
cli.agent = agent
|
|
cli._interrupt_queue = queue.Queue()
|
|
cli._pending_input = queue.Queue()
|
|
cli._interrupt_queue.put("redirect please")
|
|
|
|
with patch.object(cli, "_ensure_runtime_credentials", return_value=True), \
|
|
patch.object(cli, "_resolve_turn_agent_config", return_value={
|
|
"signature": cli._active_agent_route_signature,
|
|
"model": None, "runtime": None, "request_overrides": None,
|
|
}), \
|
|
patch.object(cli, "_init_agent", return_value=True):
|
|
cli.chat("original")
|
|
|
|
queued = []
|
|
while not cli._pending_input.empty():
|
|
queued.append(cli._pending_input.get_nowait())
|
|
assert any("redirect please" in str(q) for q in queued)
|
|
assert cli._last_turn_interrupted is True
|
|
|
|
|
|
def test_chat_persists_clean_input_when_a_queued_note_changes_api_message():
|
|
"""Queued notes remain API-local and preserve close-handoff marker identity."""
|
|
cli = _make_cli()
|
|
|
|
class _NoteAgent(_StubAgent):
|
|
def __init__(self, session_id):
|
|
super().__init__(session_id, turn_seconds=0)
|
|
self.captured = None
|
|
|
|
def run_conversation(self, **kwargs):
|
|
self.captured = kwargs
|
|
return {
|
|
"final_response": "done",
|
|
"messages": [{"role": "assistant", "content": "done"}],
|
|
"api_calls": 1,
|
|
"completed": True,
|
|
"partial": True,
|
|
"response_previewed": True,
|
|
}
|
|
|
|
agent = _NoteAgent(cli.session_id)
|
|
cli.agent = agent
|
|
cli._interrupt_queue = queue.Queue()
|
|
cli._pending_input = queue.Queue()
|
|
cli._pending_model_switch_note = "[MODEL SWITCH NOTE]"
|
|
|
|
with patch.object(cli, "_ensure_runtime_credentials", return_value=True), \
|
|
patch.object(cli, "_resolve_turn_agent_config", return_value={
|
|
"signature": cli._active_agent_route_signature,
|
|
"model": None, "runtime": None, "request_overrides": None,
|
|
}), \
|
|
patch.object(cli, "_init_agent", return_value=True):
|
|
cli.chat("clean prompt")
|
|
|
|
assert agent.captured is not None
|
|
assert agent.captured["user_message"] == "[MODEL SWITCH NOTE]\n\nclean prompt"
|
|
assert agent.captured["persist_user_message"] == "clean prompt"
|
|
|
|
|
|
def test_chat_preserves_clean_multimodal_input_when_note_changes_api_message():
|
|
"""A queued note forwards original native parts as the persistence override."""
|
|
cli = _make_cli()
|
|
|
|
class _NoteAgent(_StubAgent):
|
|
def __init__(self, session_id):
|
|
super().__init__(session_id, turn_seconds=0)
|
|
self.captured = None
|
|
|
|
def run_conversation(self, **kwargs):
|
|
self.captured = kwargs
|
|
return {
|
|
"final_response": "done",
|
|
"messages": [{"role": "assistant", "content": "done"}],
|
|
"api_calls": 1,
|
|
"completed": True,
|
|
"partial": True,
|
|
"response_previewed": True,
|
|
}
|
|
|
|
clean_parts = [
|
|
{"type": "text", "text": "Describe this screenshot"},
|
|
{"type": "image_url", "image_url": {"url": "data:image/png;base64,AAAA"}},
|
|
]
|
|
agent = _NoteAgent(cli.session_id)
|
|
cli.agent = agent
|
|
cli._interrupt_queue = queue.Queue()
|
|
cli._pending_input = queue.Queue()
|
|
cli._pending_model_switch_note = "[MODEL SWITCH NOTE]"
|
|
|
|
with patch.object(cli, "_ensure_runtime_credentials", return_value=True), \
|
|
patch.object(cli, "_resolve_turn_agent_config", return_value={
|
|
"signature": cli._active_agent_route_signature,
|
|
"model": None, "runtime": None, "request_overrides": None,
|
|
}), \
|
|
patch.object(cli, "_init_agent", return_value=True):
|
|
cli.chat(clean_parts)
|
|
|
|
assert agent.captured is not None
|
|
assert agent.captured["persist_user_message"] == clean_parts
|
|
assert agent.captured["persist_user_message"] is not agent.captured["user_message"]
|
|
api_parts = agent.captured["user_message"]
|
|
assert api_parts[0]["text"] == "[MODEL SWITCH NOTE]\n\nDescribe this screenshot"
|
|
assert api_parts[1] == clean_parts[1]
|
|
|
|
|
|
def test_chat_multimodal_note_persists_clean_input_once(tmp_path, monkeypatch):
|
|
"""The real CLI-to-agent path stores clean image parts, never the queued note."""
|
|
from hermes_state import SessionDB
|
|
from run_agent import AIAgent
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
|
|
cli = _make_cli()
|
|
session_id = cli.session_id
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
db.create_session(session_id=session_id, source="cli")
|
|
|
|
agent = object.__new__(AIAgent)
|
|
agent._session_db = db
|
|
agent._session_db_created = True
|
|
agent.session_id = session_id
|
|
agent.platform = "cli"
|
|
agent.model = "test-model"
|
|
agent.provider = "test"
|
|
agent.base_url = ""
|
|
agent.api_key = ""
|
|
agent.api_mode = "chat_completions"
|
|
agent._session_messages = []
|
|
agent._last_flushed_db_idx = 0
|
|
agent._flushed_db_message_ids = set()
|
|
agent._flushed_db_message_session_id = None
|
|
agent._persist_disabled = False
|
|
agent._cached_system_prompt = "test system prompt"
|
|
agent._session_init_model_config = None
|
|
agent._parent_session_id = None
|
|
agent._session_json_enabled = False
|
|
agent._pending_cli_user_message = None
|
|
agent._session_persist_lock = threading.RLock()
|
|
agent._persist_user_message_idx = None
|
|
agent._persist_user_message_override = None
|
|
agent._persist_user_message_timestamp = None
|
|
agent._active_children = []
|
|
agent._interrupt_requested = False
|
|
|
|
clean_parts = [
|
|
{"type": "text", "text": "Describe this screenshot"},
|
|
{"type": "image_url", "image_url": {"url": "data:image/png;base64,AAAA"}},
|
|
]
|
|
captured = {}
|
|
|
|
def _realish_run(**kwargs):
|
|
captured.update(kwargs)
|
|
# Drive production turn setup and the real SQLite persistence seam,
|
|
# then return a normal CLI result without starting a provider loop.
|
|
from agent.turn_context import build_turn_context
|
|
|
|
agent.quiet_mode = True
|
|
agent.max_iterations = 1
|
|
agent.tools = []
|
|
agent.valid_tool_names = set()
|
|
agent.enabled_toolsets = None
|
|
agent.disabled_toolsets = None
|
|
agent._skip_mcp_refresh = True
|
|
agent.compression_enabled = False
|
|
agent.context_compressor = types.SimpleNamespace(protect_first_n=2, protect_last_n=2)
|
|
agent._memory_store = None
|
|
agent._memory_manager = None
|
|
agent._memory_nudge_interval = 0
|
|
agent._turns_since_memory = 0
|
|
agent._user_turn_count = 0
|
|
agent._todo_store = types.SimpleNamespace(has_items=lambda: True)
|
|
agent._tool_guardrails = types.SimpleNamespace(reset_for_turn=lambda: None)
|
|
agent._compression_warning = None
|
|
agent._memory_write_origin = "assistant_tool"
|
|
agent._stream_context_scrubber = None
|
|
agent._stream_think_scrubber = None
|
|
agent._restore_primary_runtime = lambda: None
|
|
agent._cleanup_dead_connections = lambda: False
|
|
agent._emit_status = lambda _message: None
|
|
agent._replay_compression_warning = lambda: None
|
|
agent._hydrate_todo_store = lambda *_args: None
|
|
agent._safe_print = lambda *_args: None
|
|
|
|
context = build_turn_context(
|
|
agent,
|
|
kwargs["user_message"],
|
|
None,
|
|
kwargs["conversation_history"],
|
|
kwargs["task_id"],
|
|
None,
|
|
kwargs["persist_user_message"],
|
|
None,
|
|
restore_or_build_system_prompt=lambda *_args: None,
|
|
install_safe_stdio=lambda: None,
|
|
sanitize_surrogates=lambda value: value,
|
|
summarize_user_message_for_log=lambda value: (
|
|
value if isinstance(value, str) else "[multimodal test message]"
|
|
),
|
|
set_session_context=lambda _session_id: None,
|
|
set_current_write_origin=lambda _origin: None,
|
|
ra=lambda: types.SimpleNamespace(_set_interrupt=lambda *_args: None),
|
|
)
|
|
agent._apply_persist_user_message_override(context.messages)
|
|
agent._persist_session(context.messages, kwargs["conversation_history"])
|
|
return {
|
|
"final_response": "done",
|
|
"messages": context.messages + [{"role": "assistant", "content": "done"}],
|
|
"api_calls": 1,
|
|
"completed": True,
|
|
"partial": True,
|
|
"response_previewed": True,
|
|
}
|
|
|
|
agent.run_conversation = _realish_run
|
|
cli.agent = agent
|
|
cli._interrupt_queue = queue.Queue()
|
|
cli._pending_input = queue.Queue()
|
|
cli._pending_model_switch_note = "[MODEL SWITCH NOTE]"
|
|
|
|
with patch.object(cli, "_ensure_runtime_credentials", return_value=True), \
|
|
patch.object(cli, "_resolve_turn_agent_config", return_value={
|
|
"signature": cli._active_agent_route_signature,
|
|
"model": None, "runtime": None, "request_overrides": None,
|
|
}), \
|
|
patch.object(cli, "_init_agent", return_value=True):
|
|
cli.chat(clean_parts)
|
|
|
|
assert captured["persist_user_message"] == clean_parts
|
|
assert captured["user_message"][0]["text"] == "[MODEL SWITCH NOTE]\n\nDescribe this screenshot"
|
|
assert [m["content"] for m in db.get_messages_as_conversation(session_id)] == [
|
|
"Describe this screenshot\n[screenshot]"
|
|
]
|
|
|
|
|
|
def test_chat_clears_previous_turn_persistence_override_before_staging():
|
|
"""A close before the next worker starts cannot reuse a stale override."""
|
|
cli = _make_cli()
|
|
|
|
class _StagingAgent(_StubAgent):
|
|
def __init__(self, session_id):
|
|
super().__init__(session_id, turn_seconds=0)
|
|
self.staged_override = None
|
|
self.staged_message = None
|
|
self._session_messages = []
|
|
self._persist_user_message_idx = 7
|
|
self._persist_user_message_override = "previous clean prompt"
|
|
self._persist_user_message_timestamp = 123.0
|
|
|
|
def run_conversation(self, **kwargs):
|
|
self.staged_override = self._persist_user_message_override
|
|
self.staged_message = self._pending_cli_user_message
|
|
return {
|
|
"final_response": "done",
|
|
"messages": [{"role": "assistant", "content": "done"}],
|
|
"api_calls": 1,
|
|
"completed": True,
|
|
"partial": True,
|
|
"response_previewed": True,
|
|
}
|
|
|
|
agent = _StagingAgent(cli.session_id)
|
|
cli.agent = agent
|
|
cli._interrupt_queue = queue.Queue()
|
|
cli._pending_input = queue.Queue()
|
|
|
|
with patch.object(cli, "_ensure_runtime_credentials", return_value=True), \
|
|
patch.object(cli, "_resolve_turn_agent_config", return_value={
|
|
"signature": cli._active_agent_route_signature,
|
|
"model": None, "runtime": None, "request_overrides": None,
|
|
}), \
|
|
patch.object(cli, "_init_agent", return_value=True):
|
|
cli.chat("new prompt")
|
|
|
|
assert agent.staged_override is None
|
|
assert agent._persist_user_message_idx is None
|
|
assert agent._persist_user_message_timestamp is None
|
|
assert agent.staged_message == {"role": "user", "content": "new prompt"}
|
|
|
|
|
|
def test_chat_close_does_not_persist_previous_turn_override(tmp_path, monkeypatch):
|
|
"""A close after input staging writes the new prompt, not old API-only text."""
|
|
from hermes_state import SessionDB
|
|
from run_agent import AIAgent
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
|
|
cli = _make_cli()
|
|
session_id = cli.session_id
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
db.create_session(session_id=session_id, source="cli")
|
|
prefix = [
|
|
{"role": "user", "content": "old prompt"},
|
|
{"role": "assistant", "content": "old answer"},
|
|
]
|
|
for message in prefix:
|
|
db.append_message(
|
|
session_id=session_id,
|
|
role=message["role"],
|
|
content=message["content"],
|
|
)
|
|
|
|
agent = object.__new__(AIAgent)
|
|
agent._session_db = db
|
|
agent._session_db_created = True
|
|
agent.session_id = session_id
|
|
agent.platform = "cli"
|
|
agent.model = "test-model"
|
|
agent._session_messages = []
|
|
agent._last_flushed_db_idx = 0
|
|
agent._flushed_db_message_ids = set()
|
|
agent._flushed_db_message_session_id = None
|
|
agent._persist_disabled = False
|
|
agent._cached_system_prompt = "test system prompt"
|
|
agent._session_init_model_config = None
|
|
agent._parent_session_id = None
|
|
agent._session_json_enabled = False
|
|
agent._pending_cli_user_message = None
|
|
agent._session_persist_lock = threading.RLock()
|
|
agent._persist_user_message_idx = len(prefix)
|
|
agent._persist_user_message_override = "previous clean prompt"
|
|
agent._persist_user_message_timestamp = 123.0
|
|
agent._active_children = []
|
|
agent._interrupt_requested = False
|
|
entered = threading.Event()
|
|
release = threading.Event()
|
|
|
|
def _block_run(**_kwargs):
|
|
entered.set()
|
|
assert release.wait(timeout=5)
|
|
return {
|
|
"final_response": "done",
|
|
"messages": prefix + [{"role": "assistant", "content": "done"}],
|
|
"api_calls": 1,
|
|
"completed": True,
|
|
"partial": True,
|
|
"response_previewed": True,
|
|
}
|
|
|
|
agent.run_conversation = _block_run
|
|
cli.agent = agent
|
|
cli.conversation_history = list(prefix)
|
|
cli._interrupt_queue = queue.Queue()
|
|
cli._pending_input = queue.Queue()
|
|
|
|
with patch.object(cli, "_ensure_runtime_credentials", return_value=True), \
|
|
patch.object(cli, "_resolve_turn_agent_config", return_value={
|
|
"signature": cli._active_agent_route_signature,
|
|
"model": None, "runtime": None, "request_overrides": None,
|
|
}), \
|
|
patch.object(cli, "_init_agent", return_value=True):
|
|
chat_thread = threading.Thread(target=lambda: cli.chat("new prompt"))
|
|
chat_thread.start()
|
|
assert entered.wait(timeout=5)
|
|
cli._persist_active_session_before_close()
|
|
release.set()
|
|
chat_thread.join(timeout=10)
|
|
|
|
assert not chat_thread.is_alive()
|
|
assert [m["content"] for m in db.get_messages_as_conversation(session_id)] == [
|
|
"old prompt",
|
|
"old answer",
|
|
"new prompt",
|
|
]
|
|
|
|
|
|
def test_close_waits_for_atomic_cli_staging_before_snapshot(tmp_path, monkeypatch):
|
|
"""Close cannot retain the mutable pre-append history as its DB baseline."""
|
|
from hermes_state import SessionDB
|
|
from run_agent import AIAgent
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
|
|
cli = _make_cli()
|
|
session_id = cli.session_id
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
db.create_session(session_id=session_id, source="cli")
|
|
prefix = [
|
|
{"role": "user", "content": "old prompt"},
|
|
{"role": "assistant", "content": "old answer"},
|
|
]
|
|
for message in prefix:
|
|
db.append_message(
|
|
session_id=session_id,
|
|
role=message["role"],
|
|
content=message["content"],
|
|
)
|
|
|
|
agent = object.__new__(AIAgent)
|
|
agent._session_db = db
|
|
agent._session_db_created = True
|
|
agent.session_id = session_id
|
|
agent.platform = "cli"
|
|
agent.model = "test-model"
|
|
# Deliberately distinct from CLI history: this is the normal pre-worker
|
|
# state that used to let close retain the wrong mutable baseline.
|
|
agent._session_messages = list(prefix)
|
|
agent._last_flushed_db_idx = 0
|
|
agent._flushed_db_message_ids = set()
|
|
agent._flushed_db_message_session_id = None
|
|
agent._persist_disabled = False
|
|
agent._cached_system_prompt = "test system prompt"
|
|
agent._session_init_model_config = None
|
|
agent._parent_session_id = None
|
|
agent._session_json_enabled = False
|
|
agent._pending_cli_user_message = None
|
|
agent._session_persist_lock = threading.RLock()
|
|
agent._persist_user_message_idx = None
|
|
agent._persist_user_message_override = None
|
|
agent._persist_user_message_timestamp = None
|
|
agent._active_children = []
|
|
agent._interrupt_requested = False
|
|
|
|
staging_entered = threading.Event()
|
|
release_staging = threading.Event()
|
|
run_entered = threading.Event()
|
|
release_run = threading.Event()
|
|
|
|
class _BlockingHistory(list):
|
|
def __init__(self, values):
|
|
super().__init__(values)
|
|
self._block_next_append = True
|
|
|
|
def append(self, value):
|
|
if self._block_next_append:
|
|
self._block_next_append = False
|
|
staging_entered.set()
|
|
assert release_staging.wait(timeout=5)
|
|
return super().append(value)
|
|
|
|
def _block_run(**_kwargs):
|
|
run_entered.set()
|
|
assert release_run.wait(timeout=5)
|
|
return {
|
|
"final_response": "done",
|
|
"messages": prefix + [{"role": "assistant", "content": "done"}],
|
|
"api_calls": 1,
|
|
"completed": True,
|
|
"partial": True,
|
|
"response_previewed": True,
|
|
}
|
|
|
|
agent.run_conversation = _block_run
|
|
cli.agent = agent
|
|
cli.conversation_history = _BlockingHistory(prefix)
|
|
cli._interrupt_queue = queue.Queue()
|
|
cli._pending_input = queue.Queue()
|
|
|
|
with patch.object(cli, "_ensure_runtime_credentials", return_value=True), \
|
|
patch.object(cli, "_resolve_turn_agent_config", return_value={
|
|
"signature": cli._active_agent_route_signature,
|
|
"model": None, "runtime": None, "request_overrides": None,
|
|
}), \
|
|
patch.object(cli, "_init_agent", return_value=True):
|
|
chat_thread = threading.Thread(target=lambda: cli.chat("new prompt"))
|
|
chat_thread.start()
|
|
assert staging_entered.wait(timeout=5)
|
|
|
|
close_started = threading.Event()
|
|
close_finished = threading.Event()
|
|
|
|
def _close():
|
|
close_started.set()
|
|
cli._persist_active_session_before_close()
|
|
close_finished.set()
|
|
|
|
close_thread = threading.Thread(target=_close)
|
|
close_thread.start()
|
|
assert close_started.wait(timeout=5)
|
|
# The close snapshot must wait for the locked pending-pointer/history
|
|
# handoff; otherwise the subsequent append poisons its DB baseline.
|
|
assert not close_finished.wait(timeout=0.1)
|
|
|
|
release_staging.set()
|
|
assert run_entered.wait(timeout=5)
|
|
assert close_finished.wait(timeout=5)
|
|
release_run.set()
|
|
chat_thread.join(timeout=10)
|
|
close_thread.join(timeout=10)
|
|
|
|
assert not chat_thread.is_alive()
|
|
assert not close_thread.is_alive()
|
|
assert [m["content"] for m in db.get_messages_as_conversation(session_id)] == [
|
|
"old prompt",
|
|
"old answer",
|
|
"new prompt",
|
|
]
|