46 lines
1.9 KiB
Python
46 lines
1.9 KiB
Python
"""Todo control messages stay available to the model without becoming user turns."""
|
|
|
|
import pytest
|
|
from langchain_core.messages import AIMessage, HumanMessage
|
|
|
|
from deerflow.agents.middlewares.todo_middleware import TodoMiddleware
|
|
from deerflow.runtime.serialization import serialize_channel_values, serialize_messages_tuple
|
|
|
|
|
|
@pytest.mark.parametrize("phase", ["before", "after"])
|
|
def test_todo_reminders_keep_control_semantics_and_hide_after_serialization(phase):
|
|
middleware = TodoMiddleware()
|
|
state = {
|
|
"messages": [HumanMessage(content="Draw a map"), AIMessage(content="The map is ready")],
|
|
"todos": [{"content": "Finish the map", "status": "in_progress"}],
|
|
}
|
|
result = getattr(middleware, f"{phase}_model")(state, None)
|
|
assert result is not None
|
|
reminder = result["messages"][0]
|
|
assert isinstance(reminder, HumanMessage)
|
|
assert "Finish the map" in reminder.content
|
|
assert reminder.name == ("todo_reminder" if phase == "before" else "todo_completion_reminder")
|
|
if phase == "after":
|
|
assert result["jump_to"] == "model"
|
|
|
|
for serialized in (
|
|
serialize_channel_values({"messages": [reminder]})["messages"][0],
|
|
serialize_messages_tuple((reminder, {}))[0],
|
|
):
|
|
assert serialized["additional_kwargs"]["hide_from_ui"] is True
|
|
assert serialized["additional_kwargs"]["lc_source"] == "todo"
|
|
|
|
# Hiding a reminder must not break the existing deduplication/retry bound.
|
|
if phase == "before":
|
|
state["messages"].append(reminder)
|
|
else:
|
|
state["messages"].extend([reminder, reminder])
|
|
assert getattr(middleware, f"{phase}_model")(state, None) is None
|
|
|
|
|
|
def test_completed_todos_do_not_create_a_completion_reminder():
|
|
assert TodoMiddleware().after_model(
|
|
{"messages": [AIMessage(content="Done")], "todos": [{"content": "Draw a map", "status": "completed"}]},
|
|
None,
|
|
) is None
|