Support conversational flows in the CLI TUI (#6293)

* Add conversational flow TUI support

* properly support tui

lib/cli/tests/test_crew_run_tui.py

@@ -126,6 +126,52 @@ def test_chain_deploy_does_not_login_for_deploy_exit(monkeypatch, capsys) -> Non
     assert "Deploy failed with exit code 42" in capsys.readouterr().out
 
 
+def test_conversation_turn_done_records_assistant_message() -> None:
+    class RawResult:
+        raw = "hello from the flow"
+
+    app = CrewRunApp(conversational=True)
+    app._conversation_turn_in_progress = True
+    app._enable_conversation_input = lambda: None  # type: ignore[method-assign]
+    app._tick = lambda: None  # type: ignore[method-assign]
+    app._scroll_to_result = lambda: None  # type: ignore[method-assign]
+
+    app._on_conversation_turn_done(RawResult())
+
+    assert app._conversation_messages == [("assistant", "hello from the flow")]
+    assert app._conversation_turn_in_progress is False
+    assert app._status == "chatting"
+    assert isinstance(app._crew_result, RawResult)
+
+
+@pytest.mark.asyncio
+async def test_conversation_input_submits_turn() -> None:
+    class FakeFlow:
+        defer_trace_finalization = False
+
+        def handle_turn(self, message: str) -> str:
+            return f"reply: {message}"
+
+        def finalize_session_traces(self) -> None:
+            pass
+
+    app = CrewRunApp(crew_name="Demo", conversational=True)
+    app._flow = FakeFlow()
+
+    async with app.run_test() as pilot:
+        await pilot.click("#conversation-input")
+        await pilot.press("h", "i", "enter")
+        for _ in range(50):
+            await pilot.pause(0.05)
+            if app._conversation_messages[-1:] == [("assistant", "reply: hi")]:
+                break
+
+    assert app._conversation_messages == [
+        ("user", "hi"),
+        ("assistant", "reply: hi"),
+    ]
+
+
 def test_plan_step_status_updates_only_the_explicit_step() -> None:
     app = _app_with_plan()
 

lib/cli/tests/test_kickoff_flow.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+import sys
+
+from crewai_cli import kickoff_flow
+
+
+def test_loads_conversational_flow_from_kickoff_script(tmp_path, monkeypatch) -> None:
+    package_dir = tmp_path / "src" / "demo_chat"
+    package_dir.mkdir(parents=True)
+    (package_dir / "__init__.py").write_text("")
+    (package_dir / "main.py").write_text(
+        "\n".join(
+            [
+                "from crewai.flow import Flow",
+                "",
+                "class DemoChatFlow(Flow):",
+                "    conversational = True",
+            ]
+        )
+    )
+    (tmp_path / "pyproject.toml").write_text(
+        "\n".join(
+            [
+                "[project]",
+                'name = "demo-chat"',
+                "[project.scripts]",
+                'kickoff = "demo_chat.main:kickoff"',
+            ]
+        )
+    )
+    monkeypatch.chdir(tmp_path)
+    sys.modules.pop("demo_chat.main", None)
+    sys.modules.pop("demo_chat", None)
+
+    flow = kickoff_flow._load_conversational_flow_from_kickoff_script()
+
+    assert flow is not None
+    assert type(flow).__name__ == "DemoChatFlow"
+    assert flow.conversational is True
+
+
+def test_kickoff_flow_falls_back_to_uv_when_no_conversational_flow(
+    monkeypatch,
+) -> None:
+    calls: list[list[str]] = []
+
+    def fake_run(command, capture_output, text, check):
+        calls.append(command)
+
+        class Result:
+            stderr = ""
+
+        return Result()
+
+    monkeypatch.setattr(
+        kickoff_flow, "_load_conversational_flow_from_kickoff_script", lambda: None
+    )
+    monkeypatch.setattr(kickoff_flow.subprocess, "run", fake_run)
+
+    kickoff_flow.kickoff_flow()
+
+    assert calls == [["uv", "run", "kickoff"]]

lib/cli/tests/test_run_crew.py

@@ -645,6 +645,10 @@ def test_run_crew_runs_python_flow_project(monkeypatch, capsys):
         "_execute_uv_script",
         lambda script_name, **kwargs: calls.append((script_name, kwargs)),
     )
+    monkeypatch.setattr(
+        "crewai_cli.kickoff_flow._load_conversational_flow_from_kickoff_script",
+        lambda: None,
+    )
 
     run_crew_module.run_crew()
 
@@ -652,6 +656,41 @@ def test_run_crew_runs_python_flow_project(monkeypatch, capsys):
     assert calls == [("kickoff", {"entity_type": "flow"})]
 
 
+def test_run_crew_runs_conversational_flow_tui(monkeypatch, capsys):
+    class Flow:
+        pass
+
+    flow = Flow()
+    calls = []
+
+    monkeypatch.setattr(run_crew_module, "_has_json_crew", lambda: False)
+    monkeypatch.setattr(
+        run_crew_module,
+        "read_toml",
+        lambda: {"tool": {"crewai": {"type": "flow"}}},
+    )
+    monkeypatch.setattr(
+        "crewai_cli.kickoff_flow._load_conversational_flow_from_kickoff_script",
+        lambda: flow,
+    )
+    monkeypatch.setattr(
+        "crewai_cli.kickoff_flow._run_conversational_flow_tui",
+        lambda loaded_flow: calls.append(loaded_flow),
+    )
+    monkeypatch.setattr(
+        run_crew_module,
+        "_execute_uv_script",
+        lambda *_args, **_kwargs: pytest.fail(
+            "conversational flows must use the TUI"
+        ),
+    )
+
+    run_crew_module.run_crew()
+
+    assert capsys.readouterr().out == ""
+    assert calls == [flow]
+
+
 def test_run_crew_rejects_filename_for_flow_project(monkeypatch):
     monkeypatch.setattr(run_crew_module, "_has_json_crew", lambda: False)
     monkeypatch.setattr(