fix(conversational): persist the assistant reply from custom @listen routes (#6766)

[Anai-Guo]

Summary

Fixes #6766. Replies produced by a custom @listen route never reached the persistence backend, so across turns run on fresh Flow instances (the normal web-server pattern) the assistant side of the conversation silently disappeared and the agent could no longer recall its own prior statements.

Root cause

@persist snapshots state per method inside kickoff(). handle_turn() appends the assistant reply after kickoff() returns:

result = self.kickoff(inputs={"id": sid}, **kickoff_kwargs)
if result is not None and not self._assistant_reply_appended and self._is_public_turn_result(result):
    self.append_assistant_message(self._stringify_result(result))

That fallback append is the only thing recording the reply of a custom route that just returns a string, and it lands after the final snapshot. The built-in converse_turn is unaffected because it calls append_assistant_message() inside the method body, so its reply is part of that method's snapshot.

Fix

Take one more snapshot right after the fallback append. Flow._persist_method_completion() is split so the snapshot half is reusable as _persist_state_snapshot(label, persist_definition=None), defaulting to the flow-level @persist config; handle_turn() and stream_turn() call it after appending. When no @persist is configured it is a no-op, and handlers that append their own reply are untouched (the existing _assistant_reply_appended guard already skips the fallback).

Verification

Ran against crewai==1.15.9, whose conversational_mixin.py and flow/runtime/__init__.py are byte-identical to main.

Issue repro (LLM router replaced with an overridden route_turn() so no API call is needed), before:

turn 2 in-memory : ['user', 'user', 'assistant']

--- what actually persisted ---
  route_conversation     ['user']
  do_greet               ['user']
  route_conversation     ['user', 'user']
  do_greet               ['user', 'user']
final persisted roles: ['user', 'user']   FAIL

after:

turn 2 in-memory : ['user', 'assistant', 'user', 'assistant']

--- what actually persisted ---
  route_conversation     ['user']
  do_greet               ['user']
  handle_turn            ['user', 'assistant']
  route_conversation     ['user', 'assistant', 'user']
  do_greet               ['user', 'assistant', 'user']
  handle_turn            ['user', 'assistant', 'user', 'assistant']
final persisted roles: ['user', 'assistant', 'user', 'assistant']   PASS

Test suites (test_flow_conversation.py, test_flow_persistence.py, test_flow_persistence_factory.py, test_flow_from_definition.py): 207 passed, 21 skipped, with the single pre-existing failure TestDeferredFlowLifecycleEvents::test_finalize_batch_is_idempotent also failing unpatched on main.

The new TestPersistedReplyFromCustomRoute::test_custom_route_reply_survives_a_fresh_instance fails without the fix and passes with it; the companion test asserts a handler that appends its own reply still persists exactly one assistant message.

ruff==0.15.1 check/format with the repo pyproject.toml: no new findings (the 4 I001 reported are pre-existing and reproduce unpatched when the files are linted outside the full source tree).

---

AI-assisted, human reviewed.

Generated with Claude Code

[coderabbitai]

📝 Walkthrough

Walkthrough

The runtime adds a labeled snapshot hook and routes method completion persistence through it. Conversational and streaming turns persist fallback assistant replies after appending them. Tests cover persistence across fresh flow instances and prevent duplicate assistant messages.

Changes

Conversational persistence

Layer / File(s)	Summary
Snapshot persistence helper lib/crewai/src/crewai/flow/runtime/__init__.py, lib/crewai/src/crewai/experimental/conversational_mixin.py	The runtime adds _persist_state_snapshot(label, persistence_definition). Method completion persistence delegates to this helper and uses the supplied snapshot label.
Fallback reply persistence and regression coverage lib/crewai/src/crewai/experimental/conversational_mixin.py, lib/crewai/tests/test_flow_conversation.py	handle_turn() and streaming turns persist fallback assistant replies after appending them. Regression tests verify restoration across fresh flow instances and prevent duplicate assistant messages.

Suggested reviewers: vinibrsl

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title clearly identifies the fix for persistence of assistant replies from custom @listen routes.
Description check	✅ Passed	The description explains the persistence bug, root cause, fix, verification, and expected behavior for custom conversational routes.
Linked Issues check	✅ Passed	The changes satisfy issue #6766 by persisting fallback replies across fresh Flow instances and preventing duplicate assistant messages.
Out of Scope Changes check	✅ Passed	All code and test changes directly support the persistence fix and its regression coverage; no unrelated changes are present.
Docstring Coverage	✅ Passed	No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check.

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[coderabbitai]

🧹 Nitpick comments (1)

lib/crewai/src/crewai/experimental/conversational_mixin.py (1)

430-430: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick win

Add persistence coverage for stream_turn().

No test runs stream_turn() to completion, creates a fresh flow instance, and checks that the fallback assistant reply is restored from persistence. Add that behavior test to cover the same web-server restart pattern that TestPersistedReplyFromCustomRoute uses for handle_turn().

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@lib/crewai/src/crewai/experimental/conversational_mixin.py` at line 430, Add
a persistence test for the conversational mixin’s stream_turn() flow, following
the restart pattern used by TestPersistedReplyFromCustomRoute: run stream_turn()
to completion, create a fresh flow instance, and verify the persisted fallback
assistant reply is restored. Keep the existing handle_turn coverage unchanged
and target the test around _persist_state_snapshot("stream_turn").

Source: Coding guidelines

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Nitpick comments:
In `@lib/crewai/src/crewai/experimental/conversational_mixin.py`:
- Line 430: Add a persistence test for the conversational mixin’s stream_turn()
flow, following the restart pattern used by TestPersistedReplyFromCustomRoute:
run stream_turn() to completion, create a fresh flow instance, and verify the
persisted fallback assistant reply is restored. Keep the existing handle_turn
coverage unchanged and target the test around
_persist_state_snapshot("stream_turn").

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: 59283000-1265-44ae-87bc-c35790b5a72c

📥 Commits

Reviewing files that changed from the base of the PR and between c8f441c and b6d832f.

📒 Files selected for processing (3)

• lib/crewai/src/crewai/experimental/conversational_mixin.py
• lib/crewai/src/crewai/flow/runtime/__init__.py
• lib/crewai/tests/test_flow_conversation.py