diff --git a/lib/cli/src/crewai_cli/crew_run_tui.py b/lib/cli/src/crewai_cli/crew_run_tui.py index e07ec6912..e6bdeef0d 100644 --- a/lib/cli/src/crewai_cli/crew_run_tui.py +++ b/lib/cli/src/crewai_cli/crew_run_tui.py @@ -589,6 +589,11 @@ FooterKey .footer-key--key { """ return self._flow is not None and not self._is_conversational + @property + def _run_noun(self) -> str: + """User-facing noun for the run — 'flow' for a declarative flow, else 'crew'.""" + return "flow" if self._is_flow_run else "crew" + # ── Layout ────────────────────────────────────────────── def compose(self) -> ComposeResult: @@ -698,7 +703,7 @@ FooterKey .footer-key--key { except Exception as e: self.call_from_thread(self._on_crew_failed, str(e)) - @work(thread=True, exclusive=True, group="crew") + @work(thread=True, exclusive=True, group="flow") def _run_flow_worker(self) -> None: from crewai.events.listeners.tracing.utils import ( set_suppress_tracing_messages, @@ -763,7 +768,9 @@ FooterKey .footer-key--key { if entry["tool_name"] == "memory_save": continue entry["status"] = "timeout" - entry["error"] = "No result received before crew completed" + entry["error"] = ( + f"No result received before {self._run_noun} completed" + ) entry["duration"] = now - entry["start_time"] try: from crewai.events.listeners.tracing.trace_listener import ( @@ -811,7 +818,9 @@ FooterKey .footer-key--key { if entry["tool_name"] == "memory_save": continue entry["status"] = "error" - entry["error"] = "No result received before crew failed" + entry["error"] = ( + f"No result received before {self._run_noun} failed" + ) entry["duration"] = now - entry["start_time"] self._tick() self.call_later(self._focus_activity_log) diff --git a/lib/cli/src/crewai_cli/run_declarative_flow.py b/lib/cli/src/crewai_cli/run_declarative_flow.py index 58f7ad307..927965f59 100644 --- a/lib/cli/src/crewai_cli/run_declarative_flow.py +++ b/lib/cli/src/crewai_cli/run_declarative_flow.py @@ -1,9 +1,10 @@ from __future__ import annotations import json +import logging from pathlib import Path import subprocess -from typing import Any +from typing import TYPE_CHECKING, Any import click from crewai_core.project import ProjectDefinitionError, configured_project_definition @@ -18,6 +19,13 @@ from crewai_cli.input_prompt import ( from crewai_cli.utils import build_env_with_all_tool_credentials +if TYPE_CHECKING: + from crewai.flow.flow import Flow + + +logger = logging.getLogger(__name__) + + def run_declarative_flow_in_project_env( definition: str | Path, inputs: str | None = None ) -> None: @@ -88,7 +96,9 @@ def run_declarative_flow(definition: str | Path, inputs: str | None = None) -> N click.echo(_format_result(result)) -def _run_declarative_flow_tui(flow: Any, resolved_inputs: dict[str, Any] | None) -> Any: +def _run_declarative_flow_tui( + flow: Flow[Any], resolved_inputs: dict[str, Any] | None +) -> Any: """Run a declarative flow on the CrewAI TUI (the interactive default). Mirrors the declarative-crew TUI contract (``run_crew._run_json_crew``): @@ -113,10 +123,12 @@ def _run_declarative_flow_tui(flow: Any, resolved_inputs: dict[str, Any] | None) # this and keeps the flow's declared setting. try: flow.suppress_flow_events = False - except Exception: # noqa: S110 - pass + except Exception: + logger.debug( + "Could not disable suppress_flow_events for the flow TUI", exc_info=True + ) - app = CrewRunApp(crew_name=getattr(flow, "name", None) or type(flow).__name__) + app = CrewRunApp(crew_name=flow.name or type(flow).__name__) app._flow = flow app._flow_inputs = resolved_inputs app._flow_method_types = _flow_method_types(flow) @@ -144,22 +156,23 @@ def _run_declarative_flow_tui(flow: Any, resolved_inputs: dict[str, Any] | None) return app._crew_result -def _flow_uses_human_feedback(flow: Any) -> bool: +def _flow_uses_human_feedback(flow: Flow[Any]) -> bool: """True if any declarative method declares ``@human_feedback``. Such flows need the flow runtime's interactive stdin / Rich prompts, which don't compose with Textual — so they run on the terminal, not the TUI. """ try: - methods = getattr(getattr(flow, "_definition", None), "methods", None) or {} return any( - getattr(m, "human_feedback", None) is not None for m in methods.values() + method.human_feedback is not None + for method in flow._definition.methods.values() ) except Exception: + logger.debug("Could not inspect flow for human feedback", exc_info=True) return False -def _flow_method_types(flow: Any) -> dict[str, str]: +def _flow_method_types(flow: Flow[Any]) -> dict[str, str]: """Map each declarative method name to its ``call`` type (crew/agent/…). Best-effort: the STEPS panel shows this as a dim label. Method events don't @@ -167,13 +180,10 @@ def _flow_method_types(flow: Any) -> dict[str, str]: """ method_types: dict[str, str] = {} try: - methods = getattr(getattr(flow, "_definition", None), "methods", None) or {} - for name, method_definition in methods.items(): - call_type = getattr(getattr(method_definition, "do", None), "call", None) - if isinstance(call_type, str): - method_types[name] = call_type - except Exception: # noqa: S110 - pass + for name, method_definition in flow._definition.methods.items(): + method_types[name] = method_definition.do.call + except Exception: + logger.debug("Could not derive flow method types", exc_info=True) return method_types diff --git a/lib/cli/tests/test_crew_run_tui.py b/lib/cli/tests/test_crew_run_tui.py index 0172052d9..c95a9d877 100644 --- a/lib/cli/tests/test_crew_run_tui.py +++ b/lib/cli/tests/test_crew_run_tui.py @@ -967,6 +967,31 @@ async def test_crew_done_does_not_mark_unfinished_tool_successful() -> None: assert app._plan_step_status == {1: "failed", 2: "done", 3: "done"} +@pytest.mark.asyncio +async def test_flow_done_uses_flow_wording_for_unfinished_tool() -> None: + # The shared completion handler reports "flow" (not "crew") in flow mode. + app = CrewRunApp(crew_name="Demo Flow") + app._flow = SimpleNamespace() + + async with app.run_test(size=(100, 40)) as pilot: + app._log_entries = [ + { + "tool_name": "search", + "status": "running", + "args": None, + "result": None, + "error": None, + "start_time": time.time() - 2, + "duration": None, + "task_idx": 1, + } + ] + app._on_crew_done("final output") + await pilot.pause() + + assert app._log_entries[0]["error"] == "No result received before flow completed" + + @pytest.mark.asyncio async def test_crew_done_does_not_timeout_memory_save() -> None: app = _app_with_plan()