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

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}"
+    )

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"]
 
 

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")

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
 

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
 
-            ### [<sequence>] <ROLE> — <timestamp>
+            ### [<sequence>] <ROLE> - <timestamp>
 
             <content>
 
@@ -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("")

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}
             )

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]:

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())

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