diff --git a/features/conversation_state_management.feature b/features/conversation_state_management.feature new file mode 100644 index 000000000..00ffa8135 --- /dev/null +++ b/features/conversation_state_management.feature @@ -0,0 +1,82 @@ +Feature: Conversation state management + The ConversationStateManager separates persistent conversation history from + ephemeral execution state. Conversation history survives across runs while + execution state (current_node, execution_count, error) resets each time. + + Scenario: Conversation history persists across execution resets + Given a ConversationStateManager with no initial state + When I append messages "hello" and "world" to the conversation history + And I reset the execution state + Then the conversation history should still contain 2 messages + And the execution count should be 0 after reset + + Scenario: reset_execution_state clears only execution fields + Given a ConversationStateManager with no initial state + When I update the state with current_node "agent_a" and execution_count 5 + And I append a message "keep me" to the conversation history + And I reset the execution state + Then the conversation history should contain the message "keep me" + And the current_node should be None after reset + + Scenario: get_full_history returns all persistent messages + Given a ConversationStateManager with no initial state + When I append messages "first" and "second" and "third" to the conversation history + Then get_full_history should return 3 messages + + Scenario: history property returns persistent conversation messages + Given a ConversationStateManager with no initial state + When I append a message "persistent" to the conversation history + Then the history property should contain the message "persistent" + + Scenario: update_state with messages key appends to conversation history + Given a ConversationStateManager with no initial state + When I call update_state with messages containing "new message" + Then the conversation history should contain the message "new message" + + Scenario: GraphState.to_graph_state returns conversation_history and execution_state keys + Given a GraphState with messages and metadata + When I call to_graph_state on the GraphState + Then the result should contain a conversation_history key + And the result should contain an execution_state key with current_node + + Scenario: ExecutionState model holds execution-level fields + Given an ExecutionState with current_node "node_x" and execution_count 3 + Then the ExecutionState current_node should be "node_x" + And the ExecutionState execution_count should be 3 + + Scenario: StateManager backward-compat alias works as ConversationStateManager + Given a StateManager created via the backward-compat alias + When I append a message "compat" to the conversation history via StateManager + Then the StateManager history property should contain the message "compat" + + Scenario: append_messages evicts oldest entries when MAX_HISTORY_SIZE is exceeded + Given a ConversationStateManager with no initial state + When I append 55 messages to the conversation history + Then the conversation history should contain at most 50 messages + And the most recent messages should be retained + + Scenario: Full reset clears both conversation history and execution state + Given a ConversationStateManager with no initial state + When I append a message "before reset" to the conversation history + And I update the state with current_node "some_node" and execution_count 5 + And I perform a full reset + Then the conversation history should be empty after full reset + And the execution count should be 0 after full reset + + Scenario: Bridge default update mode is APPEND + Given a RxPyLangGraphBridge instance + Then the default update mode should be APPEND + + Scenario: Session full_history property returns all messages + Given a Session with two messages appended + Then the full_history property should return 2 messages + + Scenario: Session get_messages with no limit returns full thread + Given a Session with three messages appended + When I call get_messages with no limit + Then all 3 messages should be returned + + Scenario: Session append_message preserves full history + Given a Session with one existing message + When I append a second message to the session + Then the session should have 2 messages in total diff --git a/features/steps/conversation_state_management_steps.py b/features/steps/conversation_state_management_steps.py new file mode 100644 index 000000000..91c985763 --- /dev/null +++ b/features/steps/conversation_state_management_steps.py @@ -0,0 +1,407 @@ +"""BDD step definitions for conversation state management feature.""" + +from __future__ import annotations + +from behave import given, then, when +from behave.runner import Context + +from unittest.mock import MagicMock + +from cleveragents.domain.models.core.session import MessageRole, Session +from cleveragents.langgraph.bridge import RxPyLangGraphBridge +from cleveragents.langgraph.state import ( + ConversationStateManager, + ExecutionState, + GraphState, + StateManager, + StateUpdateMode, +) +from cleveragents.reactive.stream_router import ReactiveStreamRouter + + +def _ensure(context: Context) -> None: + context.results = getattr(context, "results", {}) or {} + context.managers = getattr(context, "managers", {}) or {} + + +# --------------------------------------------------------------------------- +# ConversationStateManager scenarios +# --------------------------------------------------------------------------- + + +@given("a ConversationStateManager with no initial state") +def step_create_conversation_state_manager(context: Context) -> None: + _ensure(context) + context.managers["csm"] = ConversationStateManager() + + +@when('I append messages "hello" and "world" to the conversation history') +def step_append_hello_world(context: Context) -> None: + mgr: ConversationStateManager = context.managers["csm"] + mgr.append_messages([{"role": "user", "content": "hello"}]) + mgr.append_messages([{"role": "user", "content": "world"}]) + + +@when("I reset the execution state") +def step_reset_execution_state(context: Context) -> None: + mgr: ConversationStateManager = context.managers["csm"] + mgr.reset_execution_state() + + +@then("the conversation history should still contain 2 messages") +def step_assert_history_has_2_messages(context: Context) -> None: + mgr: ConversationStateManager = context.managers["csm"] + assert len(mgr.get_full_history()) == 2, ( + f"Expected 2 messages, got {len(mgr.get_full_history())}" + ) + + +@then("the execution count should be 0 after reset") +def step_assert_execution_count_zero(context: Context) -> None: + mgr: ConversationStateManager = context.managers["csm"] + state = mgr.get_state() + assert state.execution_count == 0, ( + f"Expected execution_count=0, got {state.execution_count}" + ) + + +@when('I update the state with current_node "agent_a" and execution_count 5') +def step_update_state_agent_a(context: Context) -> None: + mgr: ConversationStateManager = context.managers["csm"] + mgr.update_state( + {"current_node": "agent_a"}, + mode=StateUpdateMode.REPLACE, + ) + # Perform 5 updates to increment execution_count + for _ in range(5): + mgr.update_state({}, mode=StateUpdateMode.REPLACE) + + +@when('I append a message "keep me" to the conversation history') +def step_append_keep_me(context: Context) -> None: + mgr: ConversationStateManager = context.managers["csm"] + mgr.append_messages([{"role": "user", "content": "keep me"}]) + + +@then('the conversation history should contain the message "keep me"') +def step_assert_history_contains_keep_me(context: Context) -> None: + mgr: ConversationStateManager = context.managers["csm"] + history = mgr.get_full_history() + contents = [m.get("content") for m in history] + assert "keep me" in contents, f"'keep me' not found in history: {contents}" + + +@then("the current_node should be None after reset") +def step_assert_current_node_none(context: Context) -> None: + mgr: ConversationStateManager = context.managers["csm"] + state = mgr.get_state() + assert state.current_node is None, ( + f"Expected current_node=None, got {state.current_node!r}" + ) + + +@when('I append messages "first" and "second" and "third" to the conversation history') +def step_append_three_messages(context: Context) -> None: + mgr: ConversationStateManager = context.managers["csm"] + for content in ("first", "second", "third"): + mgr.append_messages([{"role": "user", "content": content}]) + + +@then("get_full_history should return 3 messages") +def step_assert_full_history_3(context: Context) -> None: + mgr: ConversationStateManager = context.managers["csm"] + history = mgr.get_full_history() + assert len(history) == 3, f"Expected 3 messages, got {len(history)}" + + +@when('I append a message "persistent" to the conversation history') +def step_append_persistent(context: Context) -> None: + mgr: ConversationStateManager = context.managers["csm"] + mgr.append_messages([{"role": "user", "content": "persistent"}]) + + +@then('the history property should contain the message "persistent"') +def step_assert_history_property_persistent(context: Context) -> None: + mgr: ConversationStateManager = context.managers["csm"] + history = mgr.history + contents = [m.get("content") for m in history] + assert "persistent" in contents, ( + f"'persistent' not found in history property: {contents}" + ) + + +@when('I call update_state with messages containing "new message"') +def step_update_state_with_new_message(context: Context) -> None: + mgr: ConversationStateManager = context.managers["csm"] + mgr.update_state( + {"messages": [{"role": "user", "content": "new message"}]}, + mode=StateUpdateMode.APPEND, + ) + + +@then('the conversation history should contain the message "new message"') +def step_assert_history_contains_new_message(context: Context) -> None: + mgr: ConversationStateManager = context.managers["csm"] + history = mgr.get_full_history() + contents = [m.get("content") for m in history] + assert "new message" in contents, f"'new message' not found in history: {contents}" + + +# --------------------------------------------------------------------------- +# GraphState.to_graph_state scenario +# --------------------------------------------------------------------------- + + +@given("a GraphState with messages and metadata") +def step_create_graph_state_with_data(context: Context) -> None: + _ensure(context) + context.results["graph_state"] = GraphState( + messages=[{"role": "user", "content": "hi"}], + metadata={"key": "value"}, + current_node="test_node", + execution_count=2, + ) + + +@when("I call to_graph_state on the GraphState") +def step_call_to_graph_state(context: Context) -> None: + gs: GraphState = context.results["graph_state"] + context.results["graph_state_dict"] = gs.to_graph_state() + + +@then("the result should contain a conversation_history key") +def step_assert_conversation_history_key(context: Context) -> None: + result: dict = context.results["graph_state_dict"] + assert "conversation_history" in result, ( + f"'conversation_history' key missing from: {list(result.keys())}" + ) + assert isinstance(result["conversation_history"], list) + assert len(result["conversation_history"]) == 1 + + +@then("the result should contain an execution_state key with current_node") +def step_assert_execution_state_key(context: Context) -> None: + result: dict = context.results["graph_state_dict"] + assert "execution_state" in result, ( + f"'execution_state' key missing from: {list(result.keys())}" + ) + exec_state = result["execution_state"] + assert exec_state["current_node"] == "test_node" + assert exec_state["execution_count"] == 2 + + +# --------------------------------------------------------------------------- +# ExecutionState scenario +# --------------------------------------------------------------------------- + + +@given('an ExecutionState with current_node "node_x" and execution_count 3') +def step_create_execution_state(context: Context) -> None: + _ensure(context) + context.results["exec_state"] = ExecutionState( + current_node="node_x", + execution_count=3, + ) + + +@then('the ExecutionState current_node should be "node_x"') +def step_assert_exec_state_current_node(context: Context) -> None: + es: ExecutionState = context.results["exec_state"] + assert es.current_node == "node_x", f"Expected 'node_x', got {es.current_node!r}" + + +@then("the ExecutionState execution_count should be 3") +def step_assert_exec_state_execution_count(context: Context) -> None: + es: ExecutionState = context.results["exec_state"] + assert es.execution_count == 3, f"Expected 3, got {es.execution_count}" + + +# --------------------------------------------------------------------------- +# StateManager backward-compat alias scenario +# --------------------------------------------------------------------------- + + +@given("a StateManager created via the backward-compat alias") +def step_create_state_manager_alias(context: Context) -> None: + _ensure(context) + context.managers["sm"] = StateManager() + + +@when('I append a message "compat" to the conversation history via StateManager') +def step_append_compat_via_state_manager(context: Context) -> None: + mgr: StateManager = context.managers["sm"] + mgr.append_messages([{"role": "user", "content": "compat"}]) + + +@then('the StateManager history property should contain the message "compat"') +def step_assert_state_manager_history_compat(context: Context) -> None: + mgr: StateManager = context.managers["sm"] + history = mgr.history + contents = [m.get("content") for m in history] + assert "compat" in contents, ( + f"'compat' not found in StateManager history: {contents}" + ) + + +# --------------------------------------------------------------------------- +# MAX_HISTORY_SIZE eviction scenario +# --------------------------------------------------------------------------- + + +@when("I append 55 messages to the conversation history") +def step_append_55_messages(context: Context) -> None: + mgr: ConversationStateManager = context.managers["csm"] + for i in range(55): + mgr.append_messages([{"role": "user", "content": f"msg_{i}"}]) + + +@then("the conversation history should contain at most 50 messages") +def step_assert_history_at_most_50(context: Context) -> None: + mgr: ConversationStateManager = context.managers["csm"] + history = mgr.get_full_history() + assert len(history) <= 50, f"Expected at most 50 messages, got {len(history)}" + + +@then("the most recent messages should be retained") +def step_assert_most_recent_retained(context: Context) -> None: + mgr: ConversationStateManager = context.managers["csm"] + history = mgr.get_full_history() + # The last message appended was msg_54 + assert history[-1]["content"] == "msg_54", ( + f"Expected last message to be 'msg_54', got {history[-1]['content']!r}" + ) + + +# --------------------------------------------------------------------------- +# Full reset scenario +# --------------------------------------------------------------------------- + + +@when('I append a message "before reset" to the conversation history') +def step_append_before_reset(context: Context) -> None: + mgr: ConversationStateManager = context.managers["csm"] + mgr.append_messages([{"role": "user", "content": "before reset"}]) + + +@when('I update the state with current_node "some_node" and execution_count 5') +def step_update_state_some_node(context: Context) -> None: + mgr: ConversationStateManager = context.managers["csm"] + mgr.update_state({"current_node": "some_node"}, mode=StateUpdateMode.REPLACE) + for _ in range(5): + mgr.update_state({}, mode=StateUpdateMode.REPLACE) + + +@when("I perform a full reset") +def step_perform_full_reset(context: Context) -> None: + mgr: ConversationStateManager = context.managers["csm"] + mgr.reset() + + +@then("the conversation history should be empty after full reset") +def step_assert_history_empty_after_full_reset(context: Context) -> None: + mgr: ConversationStateManager = context.managers["csm"] + history = mgr.get_full_history() + assert len(history) == 0, f"Expected empty history, got {len(history)} messages" + + +@then("the execution count should be 0 after full reset") +def step_assert_execution_count_zero_after_full_reset(context: Context) -> None: + mgr: ConversationStateManager = context.managers["csm"] + state = mgr.get_state() + assert state.execution_count == 0, ( + f"Expected execution_count=0, got {state.execution_count}" + ) + + +# --------------------------------------------------------------------------- +# Bridge default update mode scenario +# --------------------------------------------------------------------------- + + +@given("a RxPyLangGraphBridge instance") +def step_create_bridge(context: Context) -> None: + _ensure(context) + scheduler = MagicMock() + router = ReactiveStreamRouter(scheduler=scheduler) + context.results["bridge"] = RxPyLangGraphBridge(stream_router=router) + + +@then("the default update mode should be APPEND") +def step_assert_bridge_default_mode_append(context: Context) -> None: + bridge: RxPyLangGraphBridge = context.results["bridge"] + assert bridge._DEFAULT_UPDATE_MODE == StateUpdateMode.APPEND, ( + f"Expected APPEND, got {bridge._DEFAULT_UPDATE_MODE}" + ) + + +# --------------------------------------------------------------------------- +# Session scenarios +# --------------------------------------------------------------------------- + + +def _make_session() -> Session: + from ulid import ULID + + return Session(session_id=str(ULID())) + + +@given("a Session with two messages appended") +def step_session_with_two_messages(context: Context) -> None: + _ensure(context) + session = _make_session() + session.append_message(MessageRole.USER, "first message") + session.append_message(MessageRole.ASSISTANT, "first response") + context.results["session"] = session + + +@then("the full_history property should return 2 messages") +def step_assert_full_history_2(context: Context) -> None: + session: Session = context.results["session"] + full_history = session.full_history + assert len(full_history) == 2, ( + f"Expected 2 messages in full_history, got {len(full_history)}" + ) + + +@given("a Session with three messages appended") +def step_session_with_three_messages(context: Context) -> None: + _ensure(context) + session = _make_session() + session.append_message(MessageRole.USER, "msg one") + session.append_message(MessageRole.ASSISTANT, "msg two") + session.append_message(MessageRole.USER, "msg three") + context.results["session"] = session + + +@when("I call get_messages with no limit") +def step_call_get_messages_no_limit(context: Context) -> None: + session: Session = context.results["session"] + context.results["messages"] = session.get_messages(limit=None) + + +@then("all 3 messages should be returned") +def step_assert_all_3_messages(context: Context) -> None: + messages = context.results["messages"] + assert len(messages) == 3, f"Expected 3 messages, got {len(messages)}" + + +@given("a Session with one existing message") +def step_session_with_one_message(context: Context) -> None: + _ensure(context) + session = _make_session() + session.append_message(MessageRole.USER, "first") + context.results["session"] = session + + +@when("I append a second message to the session") +def step_append_second_message(context: Context) -> None: + session: Session = context.results["session"] + session.append_message(MessageRole.ASSISTANT, "second") + + +@then("the session should have 2 messages in total") +def step_assert_session_has_2_messages(context: Context) -> None: + session: Session = context.results["session"] + assert session.message_count == 2, ( + f"Expected 2 messages, got {session.message_count}" + ) diff --git a/features/steps/langgraph_state_additional_coverage_steps.py b/features/steps/langgraph_state_additional_coverage_steps.py index 5ced80b5e..a52aaedb6 100644 --- a/features/steps/langgraph_state_additional_coverage_steps.py +++ b/features/steps/langgraph_state_additional_coverage_steps.py @@ -116,13 +116,13 @@ def step_perform_4_updates_time_travel(context: Context): @then("only the two most recent history snapshots should remain") def step_verify_two_history_snapshots(context: Context): manager = context.state_managers["history_trim"] - assert len(manager.history) == 2 + assert len(manager.snapshots) == 2 @then("the earliest snapshots should be discarded") def step_verify_earliest_discarded(context: Context): manager = context.state_managers["history_trim"] - ids = [snap.node_id for snap in manager.history] + ids = [snap.node_id for snap in manager.snapshots] assert ids == ["node_2", "node_3"] diff --git a/features/steps/langgraph_state_branch_coverage_steps.py b/features/steps/langgraph_state_branch_coverage_steps.py index f023f704f..83d5438d6 100644 --- a/features/steps/langgraph_state_branch_coverage_steps.py +++ b/features/steps/langgraph_state_branch_coverage_steps.py @@ -188,12 +188,12 @@ def step_manager_with_two_checkpoints(context: Context): manager = StateManager(checkpoint_dir=checkpoint_dir) # Write first checkpoint - manager.state.metadata = {"first": True} + manager.update_state({"metadata": {"first": True}}, mode=StateUpdateMode.REPLACE) manager.update_count = 1 manager._save_checkpoint() # Write second checkpoint with newer mtime - manager.state.metadata = {"second": True} + manager.update_state({"metadata": {"second": True}}, mode=StateUpdateMode.REPLACE) manager.update_count = 2 manager._save_checkpoint() @@ -246,7 +246,7 @@ def step_clear_and_reset(context: Context, metadata_json: str): @then("the history should be empty after reset") def step_verify_history_empty(context: Context): manager: StateManager = context.results["reset_manager"] - assert len(manager.history) == 0 + assert len(manager.snapshots) == 0 @then("the execution count should be zero after reset") diff --git a/src/cleveragents/application/services/session_service.py b/src/cleveragents/application/services/session_service.py index 24c74f303..c00672995 100644 --- a/src/cleveragents/application/services/session_service.py +++ b/src/cleveragents/application/services/session_service.py @@ -47,6 +47,10 @@ class PersistentSessionService(SessionService): ``SessionMessageRepository`` for persistence. All operations delegate to the repositories which flush but do not commit; callers must manage transactions via the UnitOfWork pattern. + + Message appending uses APPEND-mode semantics: new messages are + appended to the existing session history without discarding prior + content. Session exports include the full conversation history. """ def __init__( @@ -157,7 +161,10 @@ class PersistentSessionService(SessionService): content: str, metadata: dict[str, Any] | None = None, ) -> SessionMessage: - """Append a message to a session. + """Append a message to a session (APPEND mode). + + New messages are **appended** to the existing conversation + history; no prior messages are discarded or replaced. Args: session_id: The ULID of the target session. @@ -180,7 +187,7 @@ class PersistentSessionService(SessionService): result = self._sanitizer.sanitize_user_input(content) content = result.sanitized - # Determine next sequence number + # Determine next sequence number (APPEND: always after the last) count = self._message_repo.count_for_session(session_id) message = SessionMessage( @@ -220,6 +227,9 @@ class PersistentSessionService(SessionService): def export_session(self, session_id: str) -> dict[str, Any]: """Export a session as a JSON-serializable dict. + Includes the **full conversation history** (all messages in the + session thread) as part of the export. + Args: session_id: The ULID of the session to export. @@ -233,7 +243,7 @@ class PersistentSessionService(SessionService): if session is None: raise SessionNotFoundError(f"Session '{session_id}' not found") - # Load all messages + # Load all messages to include the full history in export messages = self._message_repo.get_for_session(session_id) session.messages = messages diff --git a/src/cleveragents/domain/models/core/session.py b/src/cleveragents/domain/models/core/session.py index ba9716449..bf8f0c7f8 100644 --- a/src/cleveragents/domain/models/core/session.py +++ b/src/cleveragents/domain/models/core/session.py @@ -293,6 +293,15 @@ class Session(BaseModel): """Return True if the session has no messages.""" return len(self.messages) == 0 + @property + def full_history(self) -> list[SessionMessage]: + """Return the complete conversation history (full thread). + + Messages are always stored in sequence-order so this property + provides read-only access to the entire persistent thread. + """ + return list(self.messages) + # -- Methods ------------------------------------------------------------ def append_message( @@ -307,6 +316,10 @@ class Session(BaseModel): Auto-generates a ULID for the message, sets the sequence number based on existing messages, and updates the session timestamp. + **Full message history is preserved**: every call to + ``append_message`` adds to the existing list; no prior messages + are discarded or replaced. + Args: role: The message role. content: The message content. @@ -335,6 +348,10 @@ class Session(BaseModel): ) -> list[SessionMessage]: """Return messages with optional pagination. + When *limit* is ``None`` the method returns the **full thread** + starting from *offset*. This makes the entire conversation + queryable in a single call. + Args: limit: Maximum number of messages to return (None = all). offset: Number of messages to skip from the beginning. @@ -470,7 +487,7 @@ class Session(BaseModel): This is a **lossy** export intended for sharing and documentation. It does not contain enough information to fully restore a session via - ``import_session`` — use :meth:`as_export_dict` for that. + ``import_session`` -- use :meth:`as_export_dict` for that. The output format is:: @@ -485,7 +502,7 @@ class Session(BaseModel): ## Messages - ### [] — + ### [] - @@ -508,7 +525,7 @@ class Session(BaseModel): lines.append("") for msg in self.messages: ts = msg.timestamp.strftime("%Y-%m-%d %H:%M:%S") - lines.append(f"### [{msg.sequence}] {msg.role.value.upper()} — {ts}") + lines.append(f"### [{msg.sequence}] {msg.role.value.upper()} - {ts}") lines.append("") lines.append(msg.content) lines.append("") diff --git a/src/cleveragents/langgraph/bridge.py b/src/cleveragents/langgraph/bridge.py index 0e171e583..7d1a5f9b8 100644 --- a/src/cleveragents/langgraph/bridge.py +++ b/src/cleveragents/langgraph/bridge.py @@ -14,7 +14,7 @@ from rx import operators as ops # type: ignore[attr-defined] from cleveragents.langgraph.graph import GraphConfig, LangGraph from cleveragents.langgraph.nodes import Edge, NodeConfig, NodeType -from cleveragents.langgraph.state import GraphState +from cleveragents.langgraph.state import GraphState, StateUpdateMode from cleveragents.reactive.stream_router import ( ReactiveStreamRouter, StreamConfig, @@ -24,7 +24,13 @@ from cleveragents.reactive.stream_router import ( class RxPyLangGraphBridge: - """Allows LangGraph nodes as RxPy operators and streams triggering graphs.""" + """Allows LangGraph nodes as RxPy operators and streams triggering graphs. + + Uses ``StateUpdateMode.APPEND`` as the default update mode so that + conversation history is preserved across operator invocations. + """ + + _DEFAULT_UPDATE_MODE = StateUpdateMode.APPEND def __init__(self, stream_router: ReactiveStreamRouter): self.stream_router = stream_router @@ -237,16 +243,24 @@ class RxPyLangGraphBridge: return ops.flat_map(lambda msg: rx.from_future(create_future_task(msg))) def _create_state_updater(self, params: dict[str, Any]) -> Any: + """Create a state-update operator. + + Uses ``StateUpdateMode.APPEND`` by default so that incoming + messages are appended to the existing conversation history rather + than replacing it. An explicit ``mode`` key in *params* can + override this default. + """ graph_name = params.get("graph") if not graph_name or graph_name not in self.graphs: raise ValueError(f"Invalid graph name: {graph_name}") graph = self.graphs[graph_name] + mode = StateUpdateMode(params.get("mode", self._DEFAULT_UPDATE_MODE.value)) def update_state(msg: StreamMessage) -> StreamMessage: updates = ( msg.content if isinstance(msg.content, dict) else {"data": msg.content} ) - graph.state_manager.update_state(updates) + graph.state_manager.update_state(updates, mode=mode) return msg.copy_with( metadata={**msg.metadata, "state_updated": True, "graph": graph_name} ) diff --git a/src/cleveragents/langgraph/graph.py b/src/cleveragents/langgraph/graph.py index 78b4f2f49..f3e136265 100644 --- a/src/cleveragents/langgraph/graph.py +++ b/src/cleveragents/langgraph/graph.py @@ -19,7 +19,11 @@ from rx.subject import Subject # type: ignore[attr-defined] from cleveragents.agents.base import Agent from cleveragents.langgraph.nodes import Edge, Node, NodeConfig, NodeType -from cleveragents.langgraph.state import GraphState, StateManager +from cleveragents.langgraph.state import ( + GraphState, + StateManager, + StateUpdateMode, +) from cleveragents.reactive.stream_router import ( ReactiveStreamRouter, StreamConfig, @@ -83,7 +87,7 @@ class LangGraph: # pylint: disable=too-many-instance-attributes ) self._node_executors: dict[str, Callable[[StreamMessage], StreamMessage]] = {} - # Default: min(32, cpu_count+4) — suitable for I/O-bound node executors + # Default: min(32, cpu_count+4) -- suitable for I/O-bound node executors self._executor_pool: concurrent.futures.ThreadPoolExecutor = ( concurrent.futures.ThreadPoolExecutor() ) @@ -101,6 +105,14 @@ class LangGraph: # pylint: disable=too-many-instance-attributes self._executor_pool.shutdown(wait=False) async def execute(self, input_data: GraphState | dict[str, Any]) -> GraphState: + """Execute the graph for one run. + + At the start of each run the execution state is reset so that + ``current_node``, ``execution_count`` and ``error`` start fresh, + while the full conversation history (persistent across runs) is + made available to every node via the graph state and the node + config. + """ if not self.is_running: raise RuntimeError( f"Cannot execute graph {self.name!r}: graph is not running" @@ -110,15 +122,29 @@ class LangGraph: # pylint: disable=too-many-instance-attributes if isinstance(input_data, GraphState) else GraphState.from_dict(input_data) ) - # Use replace_state() so the is_closed guard is enforced and state stream - # subscribers are notified through the proper StateManager API, preventing - # silent state corruption after StateManager.close() has been called. - self.state_manager.replace_state(state) + + # --- reset execution state, preserve conversation history --- + self.state_manager.reset_execution_state() + + # --- inject user message using APPEND mode --- + if isinstance(input_data, dict) and "messages" in input_data: + self.state_manager.update_state( + {"messages": input_data["messages"]}, + mode=StateUpdateMode.APPEND, + ) + else: + self.state_manager.update_state( + {"messages": list(state.messages)}, + mode=StateUpdateMode.APPEND, + ) + + # Start stream signalling. start_stream = f"__{self.name}_node_start__" if start_stream in self.stream_router.streams: self.stream_router.send_message(start_stream, state.model_copy(deep=True)) else: raise ValueError("Start stream not initialized") + return self.state_manager.get_state() def get_execution_history(self) -> list[str]: diff --git a/src/cleveragents/langgraph/nodes.py b/src/cleveragents/langgraph/nodes.py index c2cca9f24..fc2c508e9 100644 --- a/src/cleveragents/langgraph/nodes.py +++ b/src/cleveragents/langgraph/nodes.py @@ -59,6 +59,11 @@ class Edge(BaseModel): metadata: dict[str, Any] = Field(default_factory=dict) +def _is_tool_agent(agent: Any) -> bool: + """Return True if *agent* is an instance of :class:`ToolAgent`.""" + return isinstance(agent, ToolAgent) + + class Node: # pylint: disable=too-many-instance-attributes MAX_HISTORY_MESSAGES = 20 MAX_HISTORY_CHARS = 12000 @@ -76,8 +81,18 @@ class Node: # pylint: disable=too-many-instance-attributes def _prepare_conversation_history( self, messages: list[dict[str, Any]] ) -> tuple[list[dict[str, Any]], bool]: + """Prepare the conversation history that will be passed to an agent. + + If ``config.metadata["full_history"]`` is ``True`` the entire message + list is returned without truncation -- this is the path used by nodes + that need full thread context. + """ if not messages: return [], False + + if self.config.metadata.get("full_history", False): + return list(messages), False + raw_max_messages = self.config.metadata.get( "max_history_messages", self.MAX_HISTORY_MESSAGES ) @@ -150,11 +165,17 @@ class Node: # pylint: disable=too-many-instance-attributes self.last_execution_time = loop.time() - start_time async def _execute_agent(self, state: GraphState) -> dict[str, Any]: + """Execute an agent node, passing full conversation history to the context. + + When the node config has ``full_history: true`` the untruncated + message list is placed into :attr:`NodeConfig.metadata` under the + key ``conversation_history`` so that the executor can forward it + to the agent via the context dict. + """ if not self.config.agent: raise ValueError(f"Agent node {self.name} has no agent specified") agent = self.agents.get(self.config.agent) if not agent: - # Fallback: synthesize a response when agent instance is unavailable return { "messages": [ { @@ -167,8 +188,9 @@ class Node: # pylint: disable=too-many-instance-attributes "current_node": self.name, } + # Extract actionable input. if state.messages: - if isinstance(agent, ToolAgent): + if _is_tool_agent(agent): current_msg = state.metadata.get("current_message", "") agent_input = ( str(current_msg) @@ -191,9 +213,37 @@ class Node: # pylint: disable=too-many-instance-attributes else: agent_input = "" - trimmed_history, history_truncated = self._prepare_conversation_history( - state.messages + # Prepare conversation history for the agent. + # When full_history is enabled the entire thread is used. + full_history_enabled = self.config.metadata.get("full_history", False) + if full_history_enabled: + trimmed_history = list(state.messages) + history_truncated = False + else: + trimmed_history, history_truncated = self._prepare_conversation_history( + state.messages + ) + + # Build a node config that explicitly contains conversation history + # so the executor layer can pass it to the agent process_message call. + config_with_history = NodeConfig( + name=self.config.name, + type=self.config.type, + agent=self.config.agent, + function=self.config.function, + tools=list(self.config.tools), + retry_policy=self.config.retry_policy, + timeout=self.config.timeout, + parallel=self.config.parallel, + condition=self.config.condition, + subgraph=self.config.subgraph, + metadata={ + **self.config.metadata, + "conversation_history": trimmed_history, + "full_thread": full_history_enabled, + }, ) + graph_state_dict = state.to_dict() graph_state_dict["messages"] = trimmed_history @@ -201,6 +251,7 @@ class Node: # pylint: disable=too-many-instance-attributes "graph_state": graph_state_dict, "conversation_history": trimmed_history, "full_context": True, + "node_config": config_with_history.model_dump(), } if history_truncated: context["_history_truncated"] = True @@ -238,8 +289,8 @@ class Node: # pylint: disable=too-many-instance-attributes "graph_state", "conversation_history", "full_context", - "_history_truncated", - "_history_original_length", + "node_config", + "full_thread", } snapshot_keys = set(context_snapshot.keys()) if context_snapshot else set() current_keys = set(context.keys()) diff --git a/src/cleveragents/langgraph/state.py b/src/cleveragents/langgraph/state.py index ecc7a7605..5bdac3da1 100644 --- a/src/cleveragents/langgraph/state.py +++ b/src/cleveragents/langgraph/state.py @@ -41,7 +41,7 @@ class GraphState(BaseModel): model_config = ConfigDict(validate_assignment=True) def update( - self, updates: dict[str, Any], mode: StateUpdateMode = StateUpdateMode.MERGE + self, updates: dict[str, Any], mode: StateUpdateMode = StateUpdateMode.APPEND ) -> None: if mode == StateUpdateMode.REPLACE: for key, value in updates.items(): @@ -80,110 +80,360 @@ class GraphState(BaseModel): "error": self.error, } + def to_graph_state(self) -> dict[str, Any]: + """Return both conversation_history and execution_state keys. + + This method is the canonical interface for serialising the state + into a graph-compatible dictionary, producing keys + ``conversation_history`` and ``execution_state`` as required by + the LangGraph wiring layer. + """ + return { + "conversation_history": list(self.messages), + "execution_state": { + "current_node": self.current_node, + "execution_count": self.execution_count, + "error": self.error, + "metadata": dict(self.metadata), + }, + } + @classmethod def from_dict(cls: type[T], data: dict[str, Any]) -> T: return cls(**data) -class StateManager: # pylint: disable=too-many-instance-attributes +class ExecutionState(BaseModel): + """Execution-level state that resets between runs. + + This intentionally does NOT include messages—that belongs to the + conversation manager and must persist across runs. + """ + + current_node: str | None = None + execution_count: int = 0 + error: str | None = None + metadata: dict[str, Any] = Field(default_factory=dict) + + +class ConversationStateManager: # pylint: disable=too-many-instance-attributes + """High-level state manager separating conversation history from execution state. + + Conversation history (messages) *persists* across executions. Execution-level + state (``current_node``, ``execution_count``, ``error``) is reset at the start + of each run via :meth:`reset_execution_state`. + + Default update mode is ``StateUpdateMode.APPEND``. + """ + + MAX_HISTORY_SIZE: int = 50 + def __init__( self, initial_state: GraphState | None = None, checkpoint_dir: Path | None = None, enable_time_travel: bool = False, - ): + ) -> None: self.logger = logging.getLogger(__name__) self._lock = threading.RLock() - self.state = initial_state or GraphState() - self.checkpoint_dir = checkpoint_dir - self.enable_time_travel = enable_time_travel - self.state_stream = BehaviorSubject(self.state) - self.history: list[StateSnapshot] = [] + self._mode: StateUpdateMode = StateUpdateMode.APPEND + # Persistent conversation messages. + self._conversation_history: list[dict[str, Any]] = [] + if initial_state is not None and initial_state.messages: + self._conversation_history = list(initial_state.messages) + # Execution bookkeeping only. + self._execution_state = ExecutionState() + self.state_stream = BehaviorSubject( + initial_state or GraphState(messages=list(self._conversation_history)) + ) + self._snapshots: list[StateSnapshot] = [] self.max_history_size = 100 self.checkpoint_interval = 10 self.update_count = 0 self.is_closed = False + self.checkpoint_dir = checkpoint_dir + self.enable_time_travel = enable_time_travel if self.checkpoint_dir: self.checkpoint_dir.mkdir(parents=True, exist_ok=True) - def get_state(self) -> GraphState: + # -- Public API ---------------------------------------------------------- + + @property + def state(self) -> GraphState: + """Return the current full graph state (live reference for backward compat). + + .. note:: + This property returns a *live* ``GraphState`` built from the + internal conversation history and execution state. Mutations + to the returned object's ``metadata`` dict are reflected in + ``_execution_state.metadata`` for backward compatibility with + code that does ``manager.state.metadata[key] = value``. + """ with self._lock: - return self.state.model_copy(deep=True) + return self._full_state() + + def get_state(self) -> GraphState: + """Return a deep-copy of the full graph state.""" + with self._lock: + return self._full_state() + + @property + def history(self) -> list[dict[str, Any]]: + """Return persistent conversation messages (the full history). + + This property provides read-only access to the running conversation + transcript stored within this manager. + """ + with self._lock: + return list(self._conversation_history) + + @history.setter + def history(self, value: list[dict[str, Any]]) -> None: + """Replace the persistent conversation history.""" + with self._lock: + self._conversation_history = list(value) + + def get_full_history(self) -> list[dict[str, Any]]: + """Return the full persistent conversation history (deep copy).""" + with self._lock: + return list(self._conversation_history) + + def reset_execution_state(self) -> None: + """Reset only execution-level state, NOT conversation history. + + Clears ``current_node``, ``execution_count``, and ``error`` while + leaving the stored message history intact. + """ + with self._lock: + self._execution_state = ExecutionState() + self.update_count = 0 + state_copy = self._full_state() + self.state_stream.on_next(state_copy) + self.logger.debug("Execution state reset; conversation history preserved") + + def append_messages(self, messages: list[dict[str, Any]]) -> None: + """Append new messages to the persistent conversation history. + + If the accumulated history exceeds :attr:`MAX_HISTORY_SIZE` the + oldest entries are evicted, keeping the most recent messages. + """ + if not messages: + return + with self._lock: + self._conversation_history.extend(messages) + if len(self._conversation_history) > self.MAX_HISTORY_SIZE: + self._conversation_history = self._conversation_history[ + -self.MAX_HISTORY_SIZE : + ] + state_copy = self._full_state() + self.state_stream.on_next(state_copy) def update_state( self, updates: dict[str, Any], - mode: StateUpdateMode = StateUpdateMode.MERGE, + mode: StateUpdateMode | None = None, node_id: str | None = None, ) -> GraphState: """Apply *updates* to the internal state and notify subscribers. - .. note:: - - Emission order is **not** guaranteed to match mutation order - under concurrent access. The internal state is always - consistent (mutations are serialized by ``_lock``), but - ``state_stream.on_next()`` is called outside the lock, so - Thread B's emission may be observed before Thread A's if the - OS schedules B first after both release the lock. + Uses ``APPEND`` mode by default. Message lists are *never* + clobbered—appending or merging is always used so that + conversation history is preserved across invocations. """ + effective_mode = mode if mode is not None else self._mode with self._lock: if self.is_closed: raise RuntimeError("StateManager is closed") + if self.enable_time_travel: snapshot = StateSnapshot( - state=self.state.to_dict(), + state=self._full_state().to_dict(), timestamp=datetime.now(), node_id=node_id, ) - self.history.append(snapshot) - if len(self.history) > self.max_history_size: - self.history = self.history[-self.max_history_size :] - - self.state.update(updates, mode) - self.state.execution_count += 1 + self._snapshots.append(snapshot) + if len(self._snapshots) > self.max_history_size: + self._snapshots = self._snapshots[-self.max_history_size :] + self._execution_state.execution_count += 1 self.update_count += 1 + + for key, value in updates.items(): + if key == "messages": + if isinstance(value, list): + self._conversation_history.extend(value) + if len(self._conversation_history) > self.MAX_HISTORY_SIZE: + self._conversation_history = self._conversation_history[ + -self.MAX_HISTORY_SIZE : + ] + elif key == "metadata": + if isinstance(value, dict): + self._execution_state.metadata.update(value) + else: + exec_field = key if key in ExecutionState.model_fields else None + if exec_field is not None: + if effective_mode == StateUpdateMode.REPLACE: + setattr(self._execution_state, exec_field, value) + elif effective_mode == StateUpdateMode.MERGE: + cur = getattr(self._execution_state, exec_field) + if isinstance(cur, dict) and isinstance(value, dict): + cur.update(value) + else: + setattr(self._execution_state, exec_field, value) + else: # APPEND + setattr(self._execution_state, exec_field, value) + checkpoint_data_to_save: dict[str, Any] | None = None if ( self.checkpoint_dir and self.update_count % self.checkpoint_interval == 0 ): checkpoint_data_to_save = { - "state": self.state.to_dict(), + "state": self._full_state().to_dict(), "timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"), "update_count": self.update_count, } - # Capture a deep copy while still under the lock so subscribers - # receive an immutable point-in-time snapshot. - state_copy = self.state.model_copy(deep=True) + state_copy = self._full_state() - # Perform file I/O outside the lock to avoid blocking other threads. if checkpoint_data_to_save is not None: self._save_checkpoint(checkpoint_data_to_save) - # Emit outside the lock to avoid re-entrant deadlock and priority - # inversion when subscribers call back into StateManager. self.state_stream.on_next(state_copy) return state_copy - def _save_checkpoint(self, checkpoint_data: dict[str, Any] | None = None) -> None: - """Persist a checkpoint to disk. + def get_state_observable(self) -> Observable: + return self.state_stream - When *checkpoint_data* is ``None`` the method acquires ``_lock`` - internally to capture a consistent snapshot. The lock is - reentrant (``threading.RLock``), so calling this method while - ``_lock`` is already held by the same thread is safe. + @property + def snapshots(self) -> list[StateSnapshot]: + """Return the list of time-travel state snapshots (read-only copy).""" + with self._lock: + return list(self._snapshots) + + def clear_history(self) -> None: + """Clear the time-travel snapshot history.""" + with self._lock: + self._snapshots.clear() + + def load_checkpoint(self, checkpoint_file: Path) -> None: + """Restore state from a checkpoint file on disk.""" + checkpoint_data = json.loads(checkpoint_file.read_text(encoding="utf-8")) + with self._lock: + if self.is_closed: + raise RuntimeError("StateManager is closed") + loaded = GraphState.from_dict(checkpoint_data["state"]) + self._conversation_history = list(loaded.messages) + self._execution_state = ExecutionState( + current_node=loaded.current_node, + execution_count=loaded.execution_count, + error=loaded.error, + metadata=dict(loaded.metadata), + ) + self.update_count = checkpoint_data.get("update_count", 0) + state_copy = self._full_state() + self.state_stream.on_next(state_copy) + self.logger.info("Loaded checkpoint: %s", checkpoint_file) + + def get_latest_checkpoint(self) -> Path | None: + """Return the path to the most recently written checkpoint, or None.""" + if not self.checkpoint_dir: + return None + checkpoints = list(self.checkpoint_dir.glob("checkpoint_*.json")) + if not checkpoints: + return None + return max(checkpoints, key=lambda p: p.stat().st_mtime) + + def time_travel(self, steps_back: int = 1) -> GraphState | None: + """Restore state to a previous snapshot and return it. + + Returns ``None`` when time travel is disabled or no snapshots exist. """ + with self._lock: + if self.is_closed: + raise RuntimeError("StateManager is closed") + if not self.enable_time_travel or not self._snapshots: + return None + if steps_back >= len(self._snapshots): + steps_back = len(self._snapshots) - 1 + snapshot = self._snapshots[-(steps_back + 1)] + restored = GraphState.from_dict(snapshot.state) + self._conversation_history = list(restored.messages) + self._execution_state = ExecutionState( + current_node=restored.current_node, + execution_count=restored.execution_count, + error=restored.error, + metadata=dict(restored.metadata), + ) + state_copy = self._full_state() + self.state_stream.on_next(state_copy) + return state_copy + + def replace_state(self, new_state: GraphState) -> None: + """Atomically replace the internal state and notify subscribers.""" + with self._lock: + if self.is_closed: + raise RuntimeError("StateManager is closed") + self._conversation_history = list(new_state.messages) + self._execution_state = ExecutionState( + current_node=new_state.current_node, + execution_count=new_state.execution_count, + error=new_state.error, + metadata=dict(new_state.metadata), + ) + state_copy = self._full_state() + self.state_stream.on_next(state_copy) + + def reset(self, initial_state: GraphState | None = None) -> None: + """Full reset: both conversation history AND execution state.""" + with self._lock: + if self.is_closed: + raise RuntimeError("StateManager is closed") + self._conversation_history = [] + if initial_state is not None: + if initial_state.messages: + self._conversation_history = list(initial_state.messages) + self._execution_state = ExecutionState( + current_node=initial_state.current_node, + execution_count=initial_state.execution_count, + error=initial_state.error, + metadata=dict(initial_state.metadata), + ) + else: + self._execution_state = ExecutionState() + self.update_count = 0 + self._snapshots.clear() + state_copy = self._full_state() + self.state_stream.on_next(state_copy) + + def close(self) -> None: + """Mark this manager as closed and complete the state stream.""" + if self.is_closed: + return + self.is_closed = True + self.state_stream.on_completed() + self.logger.info("StateManager closed") + + # -- Private helpers ----------------------------------------------------- + + def _full_state(self) -> GraphState: + """Build a GraphState from persistent + execution state.""" + return GraphState( + messages=list(self._conversation_history), + metadata=dict(self._execution_state.metadata), + current_node=self._execution_state.current_node, + execution_count=self._execution_state.execution_count, + error=self._execution_state.error, + ) + + def _save_checkpoint(self, checkpoint_data: dict[str, Any] | None = None) -> None: + """Persist a checkpoint to disk.""" if not self.checkpoint_dir: return if checkpoint_data is None: - # External callers (e.g. bridge) may invoke without pre-built - # data. Acquire the lock to capture a consistent snapshot. with self._lock: checkpoint_data = { - "state": self.state.to_dict(), + "state": self._full_state().to_dict(), "timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"), "update_count": self.update_count, } @@ -200,84 +450,15 @@ class StateManager: # pylint: disable=too-many-instance-attributes ) self.logger.debug("Saved checkpoint: %s", checkpoint_file) - def load_checkpoint(self, checkpoint_file: Path) -> None: - # Read file I/O outside the lock to avoid blocking other threads. - checkpoint_data = json.loads(checkpoint_file.read_text(encoding="utf-8")) - with self._lock: - if self.is_closed: - raise RuntimeError("StateManager is closed") - self.state = GraphState.from_dict(checkpoint_data["state"]) - self.update_count = checkpoint_data.get("update_count", 0) - state_copy = self.state.model_copy(deep=True) - self.state_stream.on_next(state_copy) - self.logger.info("Loaded checkpoint: %s", checkpoint_file) - def get_latest_checkpoint(self) -> Path | None: - if not self.checkpoint_dir: - return None - checkpoints = list(self.checkpoint_dir.glob("checkpoint_*.json")) - if not checkpoints: - return None - return max(checkpoints, key=lambda p: p.stat().st_mtime) +# -- Backward-compatible alias ------------------------------------------------- - def time_travel(self, steps_back: int = 1) -> GraphState | None: - with self._lock: - if self.is_closed: - raise RuntimeError("StateManager is closed") - if not self.enable_time_travel or not self.history: - return None - if steps_back >= len(self.history): - steps_back = len(self.history) - 1 - snapshot = self.history[-(steps_back + 1)] - self.state = GraphState.from_dict(snapshot.state) - state_copy = self.state.model_copy(deep=True) - self.state_stream.on_next(state_copy) - return state_copy - def replace_state(self, new_state: GraphState) -> None: - """Atomically replace the internal state and notify subscribers. +class StateManager(ConversationStateManager): + """Backward-compatible alias for ``ConversationStateManager``. - Raises :class:`RuntimeError` if the manager has been closed. + Provides the same interface as the original ``StateManager`` class while + inheriting all new conversation-state separation logic. + """ - .. note:: - - See :meth:`update_state` for a note on emission ordering - under concurrent access. - """ - with self._lock: - if self.is_closed: - raise RuntimeError("StateManager is closed") - self.state = new_state.model_copy(deep=True) - state_copy = new_state.model_copy(deep=True) - self.state_stream.on_next(state_copy) - - def get_state_observable(self) -> Observable: - return self.state_stream - - def clear_history(self) -> None: - with self._lock: - self.history.clear() - - def reset(self, initial_state: GraphState | None = None) -> None: - with self._lock: - if self.is_closed: - raise RuntimeError("StateManager is closed") - self.state = ( - initial_state.model_copy(deep=True) if initial_state else GraphState() - ) - self.update_count = 0 - self.history.clear() - state_copy = self.state.model_copy(deep=True) - self.state_stream.on_next(state_copy) - - def close(self) -> None: - """Mark this manager as closed and complete the state stream. - - After calling ``close()``, no further state updates should be - performed. Any underlying checkpoint resources are released. - """ - if self.is_closed: - return - self.is_closed = True - self.state_stream.on_completed() - self.logger.info("StateManager closed") + pass