From 85c467dfe2bd4d021ab9760263e57adfe4775dd8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Moura?= Date: Fri, 10 Jul 2026 11:42:12 -0300 Subject: [PATCH] feat(cli): run declarative flows on the TUI (headless terminal fallback) (#6484) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(cli): run declarative flows on the TUI with a headless terminal fallback Declarative flows now run on the CrewRunApp TUI when interactive, matching declarative crews and conversational flows. Headless contexts — CREWAI_DMN (deploy), piped output, CI, any non-TTY — fall back to the direct-terminal kickoff, gated by is_interactive() (folds in the CREWAI_DMN check and requires a real TTY). The TUI shows per-method progress: a new STEPS panel driven by flow method events (FlowStarted / MethodExecutionStarted/Finished/Failed), each labeled with its declarative call type (crew/agent/expression/…) read from the flow definition. Crews/agents inside a method keep streaming in the main panel via the existing crew/task/LLM handlers. - crew_run_tui.py: _run_flow_worker (flow.kickoff in a thread worker; reuses _on_crew_done/_on_crew_failed + _stringify_output), _is_flow_run gate so crew rendering is byte-identical, flow-event subscriptions building _flow_steps, and the STEPS sidebar + flow-aware header. - run_declarative_flow.py: is_interactive() branch → _run_declarative_flow_tui (EventListener, method-type map from flow._definition, crew-parity exit codes and deploy chaining) or the existing terminal path. Deviation from the approved plan: gate on is_interactive() rather than is_dmn_mode_enabled() alone, so non-TTY runs (CI/pipes/CliRunner) never launch a TUI — this also keeps existing headless flow tests green. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01RBYGqJHC2TMC6fonFziuuh * fix(cli): force flow events on for the TUI so STEPS renders under suppress_flow_events Review follow-up: the STEPS panel and header are driven by flow method events (FlowStarted / MethodExecution*), but the declarative runtime skips emitting those when the flow declared config.suppress_flow_events. Interactive TUI runs would then keep STEPS on "waiting…" and the header on "Starting flow…" while nested crews still execute. _run_declarative_flow_tui now forces flow.suppress_flow_events = False for the interactive run (mirroring how the conversational path mutates the flow for the TUI). The headless/terminal path never reaches this and keeps the flow's declared setting. Regression test: test_run_declarative_flow_tui_enables_flow_events. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01RBYGqJHC2TMC6fonFziuuh * fix(cli): clear flow header's current method when a method ends Review follow-up: the flow header keys off _current_method, which was set on MethodExecutionStarted but never cleared on Finished/Failed. Between steps (or after a failed method before kickoff exits) the header kept spinning the old method name while the STEPS sidebar already showed it done/failed. _clear_current_method now drops the header's active method when it ends, falling back to another still-active step (methods can overlap) or none. The header's idle fallback shows "Working…" once a step has run and "Starting flow…" only before the first method. Tests: test_current_method_clears_and_falls_back_across_overlap, plus a _current_method assertion in test_flow_method_events_build_steps. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01RBYGqJHC2TMC6fonFziuuh * fix: suppress flow console panels in TUI mode; clear header agent on method change Two review follow-ups: 1) Method panels break Textual TUI (Cursor): forcing suppress_flow_events off so the STEPS panel receives events also un-gated the EventListener's Rich flow/method panels (ConsoleFormatter.print_panel prints is_flow=True panels regardless of verbose), which interleave with Textual and corrupt the TUI. print_panel now skips is_flow panels when is_tui_mode() is set (the same context the TUI worker already establishes and the tracing listeners already honor). Non-TUI/headless flow runs are unaffected. Test: test_console_formatter_tui_mode. 2) Flow header showed a stale agent (CodeRabbit): _current_agent persisted across methods. It's now cleared when a method starts and when the active method changes, so the header never shows the previous method's agent until a new agent event arrives. Test: test_flow_method_transitions_clear_current_agent. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01RBYGqJHC2TMC6fonFziuuh * fix(cli): keep flow name over nested crews; show paused flow methods Two review follow-ups on the flow TUI: 1) Crew kickoff renamed the flow (Cursor): CrewKickoffStartedEvent overwrote _crew_name / the app title with a nested `call: crew` step's crew name, so the post-run summary could be labeled with a child crew. The rename is now gated on `not _is_flow_run`, preserving the flow's name; crew runs still adopt the crew name. Tests: test_crew_kickoff_does_not_rename_flow_run, test_crew_kickoff_renames_in_crew_mode. 2) Paused methods showed active (Cursor): the TUI didn't handle MethodExecutionPausedEvent, so a @human_feedback pause left the STEPS spinner running (flow status panels are suppressed in TUI mode). It now marks the step "paused" (⏸, teal) and the header shows "waiting for feedback" instead of a spinner. Test: test_method_paused_marks_step_paused. Note: interactively *providing* human feedback from the flow TUI is a separate follow-up; this only makes the pause visible instead of a silent stuck spinner. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01RBYGqJHC2TMC6fonFziuuh * fix(cli): run human-feedback declarative flows on the terminal, not the TUI Two review follow-ups, both rooted in @human_feedback methods: - Paused flow marked complete (Cursor): async human feedback makes kickoff RETURN a HumanFeedbackPending marker (not raise), which _run_flow_worker would stringify and report as a successful completion with exit 0. - Sync feedback breaks TUI (Cursor): default (sync) @human_feedback collects input via the flow runtime's Rich console.print + blocking input(), which interleaves with Textual and leaves the user unable to review output or submit feedback. run_declarative_flow now routes any flow whose declarative definition declares human feedback (_flow_uses_human_feedback) to the terminal path, where blocking input and Rich prompts work natively — regardless of interactivity. Non-feedback flows still get the TUI. Tests: test_flow_uses_human_feedback_detection, test_human_feedback_flow_uses_terminal_even_when_interactive. Fully interactive human feedback inside the TUI remains a separate follow-up. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01RBYGqJHC2TMC6fonFziuuh * refactor(cli): address review — Flow typing, debug logging, flow-vs-crew naming Review follow-ups from @lucasgomide: - Type flow helpers as Flow[Any] (via TYPE_CHECKING import) instead of Any and drop the defensive getattr chains — _definition is a typed PrivateAttr and name/suppress_flow_events are typed fields, so attribute access is safe. - Replace the silent `except Exception: pass` blocks with logger.debug(..., exc_info=True) so unexpected failures are diagnosable in the field (_flow_method_types, _flow_uses_human_feedback, suppress_flow_events toggle). - Flow-vs-crew naming: the flow worker now uses group="flow" (was the misleading "crew"), and the shared completion/failure handlers report the run with an entity-aware noun ("flow" vs "crew") via _run_noun. Deferred (separate PR): the os._exit(130) hard-kill on user quit is kept as-is to match the existing crew convention (run_crew._run_json_crew). Tests: test_flow_done_uses_flow_wording_for_unfinished_tool; existing crew wording tests unchanged. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01RBYGqJHC2TMC6fonFziuuh --------- Co-authored-by: Claude Opus 4.8 (1M context) --- lib/cli/src/crewai_cli/crew_run_tui.py | 237 ++++++++++++++++- .../src/crewai_cli/run_declarative_flow.py | 179 ++++++++++++- lib/cli/tests/test_crew_run_tui.py | 240 ++++++++++++++++++ lib/cli/tests/test_run_declarative_flow.py | 211 +++++++++++++++ .../crewai/events/utils/console_formatter.py | 7 + .../test_console_formatter_tui_mode.py | 46 ++++ 6 files changed, 914 insertions(+), 6 deletions(-) create mode 100644 lib/crewai/tests/utilities/test_console_formatter_tui_mode.py diff --git a/lib/cli/src/crewai_cli/crew_run_tui.py b/lib/cli/src/crewai_cli/crew_run_tui.py index cefb7d142..e6bdeef0d 100644 --- a/lib/cli/src/crewai_cli/crew_run_tui.py +++ b/lib/cli/src/crewai_cli/crew_run_tui.py @@ -568,12 +568,32 @@ FooterKey .footer-key--key { self._default_inputs: dict[str, Any] | None = None self._crew_result: Any = None self._crew_json_path: Any = None + # Declarative-flow execution state. A flow renders per-method "STEPS" + # (built from flow method events) instead of the crew task list. + self._flow_inputs: dict[str, Any] | None = None + self._flow_method_types: dict[str, str] = {} + self._flow_steps: list[dict[str, Any]] = [] + self._current_method: str | None = None self._elapsed_frozen: float | None = None self._want_deploy: bool = False self._trace_url: str | None = None self._consent_screen: TraceConsentScreen | None = None self._telemetry: Telemetry | None = None + @property + def _is_flow_run(self) -> bool: + """True for a non-conversational declarative flow (the STEPS view). + + Gates every flow-specific rendering branch so crew and conversational + paths stay byte-identical. + """ + 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: @@ -602,6 +622,8 @@ FooterKey .footer-key--key { self._tick_timer = self.set_interval(1 / 8, self._tick) if self._is_conversational and self._flow: self._start_conversational_session() + elif self._flow: + self._run_flow_worker() elif self._crew: self._run_crew_worker() elif self._crew_json_path: @@ -681,6 +703,49 @@ FooterKey .footer-key--key { except Exception as e: self.call_from_thread(self._on_crew_failed, str(e)) + @work(thread=True, exclusive=True, group="flow") + def _run_flow_worker(self) -> None: + from crewai.events.listeners.tracing.utils import ( + set_suppress_tracing_messages, + set_tui_mode, + ) + + set_tui_mode(True) + set_suppress_tracing_messages(True) + try: + # A declarative flow returns either a CrewOutput (has ``.raw``) or a + # bare value (str/dict/pydantic); _stringify_output handles both. + result = self._flow.kickoff(inputs=self._flow_inputs) + output = self._stringify_output(result) + with self._lock: + self._crew_result = result + self.call_from_thread(self._on_crew_done, output) + except Exception as e: + self.call_from_thread(self._on_crew_failed, str(e)) + + def _set_flow_step_status(self, name: str, status: str) -> None: + """Update a flow method step's status. Caller must hold ``self._lock``.""" + for step in self._flow_steps: + if step["name"] == name: + step["status"] = status + return + + def _clear_current_method(self, finished_name: str) -> None: + """Drop the header's active method once it ends. Caller holds the lock. + + Falls back to another still-active step (methods can overlap) so the + header never keeps spinning a method the STEPS list already shows as + done or failed. + """ + if self._current_method != finished_name: + return + self._current_method = next( + (s["name"] for s in self._flow_steps if s["status"] == "active"), None + ) + # The active method changed; drop its agent so the header doesn't show a + # stale agent until the next method's agent event arrives. + self._current_agent = "" + def _on_crew_done(self, output: str | None) -> None: with self._lock: self._status = "completed" @@ -694,13 +759,18 @@ FooterKey .footer-key--key { for k in self._task_statuses: if self._task_statuses[k] == "active": self._task_statuses[k] = "done" + for step in self._flow_steps: + if step["status"] == "active": + step["status"] = "done" now = time.time() for entry in self._log_entries: if entry["status"] == "running": 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 ( @@ -739,13 +809,18 @@ FooterKey .footer-key--key { self._is_streaming = False self._current_step = None self._elapsed_frozen = time.time() - self._start_time + for step in self._flow_steps: + if step["status"] == "active": + step["status"] = "failed" now = time.time() for entry in self._log_entries: if entry["status"] == "running": 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) @@ -1156,6 +1231,45 @@ FooterKey .footer-key--key { widget.update(t) return + if self._is_flow_run: + t.append(" STEPS\n", style=f"bold {_C_PRIMARY}") + t.append("\n") + if not self._flow_steps: + t.append(" ○ waiting…\n", style=_C_DIM) + for step in self._flow_steps: + name = step["name"] + max_name = sidebar_width - 6 + if len(name) > max_name: + name = name[: max_name - 1] + "…" + status = step.get("status", "pending") + if status == "done": + t.append(" ✔ ", style=_C_GREEN) + t.append(name, style=_C_DIM) + elif status == "active": + t.append(f" {self._spinner()} ", style=_C_PRIMARY) + t.append(name, style=f"bold {_C_TEXT}") + elif status == "failed": + t.append(" ✘ ", style=_C_RED) + t.append(name, style=_C_RED) + elif status == "paused": + t.append(" ⏸ ", style=_C_TEAL) + t.append(name, style=_C_TEAL) + else: + t.append(" ○ ", style=_C_DIM) + t.append(name, style=_C_DIM) + if step.get("call_type"): + t.append(f" ({step['call_type']})", style=_C_DIM) + t.append("\n") + + t.append("\n") + t.append(" TOKENS\n", style=f"bold {_C_PRIMARY}") + t.append("\n") + out = self._output_tokens + self._live_out_tokens + t.append(f" ↑ {self._input_tokens:,}\n", style=_C_DIM) + t.append(f" ↓ {out:,}\n", style=_C_DIM) + widget.update(t) + return + t.append(" TASKS\n", style=f"bold {_C_PRIMARY}") t.append("\n") @@ -1225,6 +1339,55 @@ FooterKey .footer-key--key { widget.update(t) return + if self._is_flow_run: + if self._status == "completed": + elapsed = self._elapsed_frozen or (time.time() - self._start_time) + t.append("✔ ", style=f"bold {_C_GREEN}") + t.append("Flow complete", style=f"bold {_C_GREEN}") + t.append(f" {elapsed:.1f}s", style=_C_DIM) + out = self._output_tokens + self._live_out_tokens + parts = [] + if self._input_tokens: + parts.append(f"↑{self._input_tokens:,}") + if out: + parts.append(f"↓{out:,}") + if parts: + t.append(f" {' '.join(parts)} tokens", style=_C_DIM) + elif self._status == "failed": + t.append("✘ ", style=f"bold {_C_RED}") + t.append("Failed", style=f"bold {_C_RED}") + if self._error: + t.append(f"\n{self._error[:120]}", style=_C_RED) + elif self._current_method: + paused = any( + s["name"] == self._current_method and s["status"] == "paused" + for s in self._flow_steps + ) + if paused: + t.append("⏸ ", style=_C_TEAL) + t.append(self._current_method, style=f"bold {_C_TEAL}") + else: + t.append(f"{self._spinner()} ", style=_C_PRIMARY) + t.append(self._current_method, style=f"bold {_C_PRIMARY}") + call_type = self._flow_method_types.get(self._current_method) + if call_type: + t.append(f" ({call_type})", style=_C_DIM) + if paused: + t.append(" waiting for feedback", style=_C_DIM) + elif self._current_agent: + t.append("\nAgent: ", style=_C_DIM) + t.append(self._current_agent, style=f"bold {_C_TEXT}") + else: + t.append(f"{self._spinner()} ", style=_C_PRIMARY) + # "Working…" once a step has run (between/after methods); + # "Starting flow…" only before the first method. + t.append( + "Working…" if self._flow_steps else "Starting flow…", + style=_C_DIM, + ) + widget.update(t) + return + if self._status == "completed": elapsed = self._elapsed_frozen or (time.time() - self._start_time) t.append("✔ ", style=f"bold {_C_GREEN}") @@ -1839,6 +2002,13 @@ FooterKey .footer-key--key { def _subscribe(self) -> None: from crewai.events.event_bus import crewai_event_bus from crewai.events.types.crew_events import CrewKickoffStartedEvent + from crewai.events.types.flow_events import ( + FlowStartedEvent, + MethodExecutionFailedEvent, + MethodExecutionFinishedEvent, + MethodExecutionPausedEvent, + MethodExecutionStartedEvent, + ) from crewai.events.types.llm_events import ( LLMCallCompletedEvent, LLMCallStartedEvent, @@ -1872,13 +2042,74 @@ FooterKey .footer-key--key { @crewai_event_bus.on(CrewKickoffStartedEvent) def on_crew_started(source: Any, event: CrewKickoffStartedEvent) -> None: with self._lock: - if event.crew_name: + # In flow mode the app is named for the flow; a nested crew's + # kickoff (a `call: crew` step) must not rename it. + if event.crew_name and not self._is_flow_run: self._crew_name = event.crew_name self.title = f"CrewAI — {event.crew_name}" self._status = "working" self._register_handler(CrewKickoffStartedEvent, on_crew_started) + # ── Declarative-flow method events → STEPS panel ──────── + @crewai_event_bus.on(FlowStartedEvent) + def on_flow_started(source: Any, event: FlowStartedEvent) -> None: + with self._lock: + self._status = "working" + + self._register_handler(FlowStartedEvent, on_flow_started) + + @crewai_event_bus.on(MethodExecutionStartedEvent) + def on_method_started(source: Any, event: MethodExecutionStartedEvent) -> None: + with self._lock: + name = event.method_name + self._current_method = name + # Agent is per-method; clear it so the header doesn't show the + # previous method's agent until a new agent event arrives. + self._current_agent = "" + for step in self._flow_steps: + if step["name"] == name: + step["status"] = "active" + break + else: + self._flow_steps.append( + { + "name": name, + "call_type": self._flow_method_types.get(name), + "status": "active", + } + ) + + self._register_handler(MethodExecutionStartedEvent, on_method_started) + + @crewai_event_bus.on(MethodExecutionFinishedEvent) + def on_method_finished( + source: Any, event: MethodExecutionFinishedEvent + ) -> None: + with self._lock: + self._set_flow_step_status(event.method_name, "done") + self._clear_current_method(event.method_name) + + self._register_handler(MethodExecutionFinishedEvent, on_method_finished) + + @crewai_event_bus.on(MethodExecutionFailedEvent) + def on_method_failed(source: Any, event: MethodExecutionFailedEvent) -> None: + with self._lock: + self._set_flow_step_status(event.method_name, "failed") + self._clear_current_method(event.method_name) + + self._register_handler(MethodExecutionFailedEvent, on_method_failed) + + @crewai_event_bus.on(MethodExecutionPausedEvent) + def on_method_paused(source: Any, event: MethodExecutionPausedEvent) -> None: + # A @human_feedback method paused; flow status panels are suppressed + # in TUI mode, so surface the wait in STEPS/header instead of leaving + # a spinner. _current_method stays pointed at it. + with self._lock: + self._set_flow_step_status(event.method_name, "paused") + + self._register_handler(MethodExecutionPausedEvent, on_method_paused) + @crewai_event_bus.on(TaskStartedEvent) def on_task_started(source: Any, event: TaskStartedEvent) -> None: with self._lock: diff --git a/lib/cli/src/crewai_cli/run_declarative_flow.py b/lib/cli/src/crewai_cli/run_declarative_flow.py index 6061d369e..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: @@ -66,17 +74,182 @@ def run_declarative_flow(definition: str | Path, inputs: str | None = None) -> N flow = load_declarative_flow(definition) resolved_inputs = _resolve_flow_inputs(flow, provided) + # The TUI is the interactive default. Headless contexts run directly on the + # terminal: deploy/CREWAI_DMN, piped output, CI — anything without an + # interactive TTY. is_interactive() already folds in the CREWAI_DMN check. + # Human-feedback flows also run on the terminal: their methods collect input + # via the flow runtime's blocking input()/Rich prompts (and async feedback + # returns a pending marker rather than completing), neither of which the + # Textual TUI can handle correctly. + if is_interactive() and not _flow_uses_human_feedback(flow): + _run_declarative_flow_tui(flow, resolved_inputs or None) + return + try: result = flow.kickoff(inputs=resolved_inputs or None) except Exception as exc: click.echo( - f"An error occurred while running the declarative flow: {exc}", err=True + f"An error occurred while running the declarative flow: {exc}", + err=True, ) raise SystemExit(1) from exc - click.echo(_format_result(result)) +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``): + a failed flow exits non-zero, a user quit ends the process so in-flight LLM + work stops, and choosing Deploy chains into the deploy command. + """ + import os + import sys + + from crewai.events.event_listener import EventListener + + from crewai_cli.crew_run_tui import CrewRunApp + + # The flow runtime (unlike a Crew constructor) doesn't create the event + # listener, and the TUI's trace/telemetry features depend on it. + EventListener() + + # The STEPS panel and header are driven by flow method events. A flow may + # declare ``config.suppress_flow_events`` (a headless/production + # optimization) which would leave STEPS stuck on "waiting…" here — so force + # emission on for the interactive TUI run. The headless path never reaches + # this and keeps the flow's declared setting. + try: + flow.suppress_flow_events = False + except Exception: + logger.debug( + "Could not disable suppress_flow_events for the flow TUI", exc_info=True + ) + + 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) + + app.run() + + _print_flow_post_tui_summary(app) + + if app._status == "failed": + raise SystemExit(1) + + if app._status not in ("completed", "failed"): + # User quit mid-run. kickoff runs in a thread worker that cannot be + # force-cancelled, so end the process to stop in-flight LLM and tool + # work instead of letting it burn tokens in the background. + click.secho("\n Run cancelled.", fg="yellow") + sys.stdout.flush() + os._exit(130) + + if getattr(app, "_want_deploy", False): + from crewai_cli.run_crew import _chain_deploy + + _chain_deploy() + + return app._crew_result + + +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: + return any( + 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: 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 + carry the call type, so it's read from the flow definition up front. + """ + method_types: dict[str, str] = {} + try: + 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 + + +def _print_flow_post_tui_summary(app: Any) -> None: + """Print a compact result panel after the flow TUI exits.""" + import time + + from rich.console import Console + from rich.markdown import Markdown + from rich.padding import Padding + from rich.panel import Panel + from rich.text import Text + + console = Console() + elapsed = (app._elapsed_frozen or (time.time() - app._start_time)) or 0.0 + + out_tokens = app._output_tokens + app._live_out_tokens + token_parts = [] + if app._input_tokens: + token_parts.append(f"↑{app._input_tokens:,}") + if out_tokens: + token_parts.append(f"↓{out_tokens:,}") + token_str = " ".join(token_parts) + if token_str: + token_str += " tokens" + + crewai_red = "#FF5A50" + crewai_teal = "#1F7982" + + if app._status == "completed": + summary = Text() + summary.append(" ✔ Flow complete", style=f"bold {crewai_teal}") + summary.append(f" in {elapsed:.1f}s", style="dim") + if token_str: + summary.append(f" {token_str}", style="dim") + console.print( + Panel( + summary, + title=f" {app._crew_name} ", + title_align="left", + border_style=crewai_teal, + padding=(0, 1), + ) + ) + if app._final_output: + console.print() + console.print(Text(" Final Result", style=f"bold {crewai_teal}")) + console.print() + console.print(Padding(Markdown(app._final_output), (0, 2))) + elif app._status == "failed": + content = Text() + content.append(" ✘ Failed", style=f"bold {crewai_red}") + content.append(f" after {elapsed:.1f}s\n", style="dim") + if app._error: + content.append(f"\n {app._error}\n", style=crewai_red) + console.print( + Panel( + content, + title=f" {app._crew_name} ", + title_align="left", + border_style=crewai_red, + padding=(0, 1), + ) + ) + + def _resolve_flow_inputs(flow: Any, provided: dict[str, Any]) -> dict[str, Any]: """Resolve kickoff inputs from the flow's state schema. diff --git a/lib/cli/tests/test_crew_run_tui.py b/lib/cli/tests/test_crew_run_tui.py index e41b4ec9b..c95a9d877 100644 --- a/lib/cli/tests/test_crew_run_tui.py +++ b/lib/cli/tests/test_crew_run_tui.py @@ -6,6 +6,14 @@ from unittest.mock import Mock import pytest from crewai.events.event_bus import crewai_event_bus +from crewai.events.types.crew_events import CrewKickoffStartedEvent +from crewai.events.types.flow_events import ( + FlowStartedEvent, + MethodExecutionFailedEvent, + MethodExecutionFinishedEvent, + MethodExecutionPausedEvent, + MethodExecutionStartedEvent, +) from crewai.events.types.memory_events import ( MemorySaveCompletedEvent, MemorySaveFailedEvent, @@ -959,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() @@ -1481,3 +1514,210 @@ def test_overlapping_task_logs_keep_their_own_state() -> None: assert any(step.get("summary") == "thinking" for step in entry2["steps"]) finally: app._unsubscribe() + + +# ── Declarative-flow (non-conversational) TUI support ─────── + + +def test_is_flow_run_gating() -> None: + """The flow-render gate must be true only for a non-conversational flow.""" + crew_app = CrewRunApp(total_tasks=1) + crew_app._crew = SimpleNamespace() + assert crew_app._is_flow_run is False + + conv_app = CrewRunApp(conversational=True) + conv_app._flow = SimpleNamespace() + assert conv_app._is_flow_run is False + + flow_app = CrewRunApp() + flow_app._flow = SimpleNamespace() + assert flow_app._is_flow_run is True + + +def test_flow_method_events_build_steps() -> None: + app = CrewRunApp(crew_name="Demo") + app._flow = SimpleNamespace() + app._flow_method_types = {"research": "crew", "summarize": "agent"} + app._subscribe() + try: + _emit_event(FlowStartedEvent(flow_name="Demo")) + assert app._status == "working" + + _emit_event( + MethodExecutionStartedEvent( + flow_name="Demo", method_name="research", state={} + ) + ) + assert app._flow_steps == [ + {"name": "research", "call_type": "crew", "status": "active"} + ] + assert app._current_method == "research" + + _emit_event( + MethodExecutionFinishedEvent( + flow_name="Demo", method_name="research", result="ok", state={} + ) + ) + _emit_event( + MethodExecutionStartedEvent( + flow_name="Demo", method_name="summarize", state={} + ) + ) + _emit_event( + MethodExecutionFailedEvent( + flow_name="Demo", + method_name="summarize", + error=RuntimeError("boom"), + ) + ) + finally: + app._unsubscribe() + + assert app._flow_steps == [ + {"name": "research", "call_type": "crew", "status": "done"}, + {"name": "summarize", "call_type": "agent", "status": "failed"}, + ] + # The header must not keep spinning a method that already ended. + assert app._current_method is None + + +def test_current_method_clears_and_falls_back_across_overlap() -> None: + app = CrewRunApp(crew_name="Demo") + app._flow = SimpleNamespace() + app._subscribe() + try: + _emit_event( + MethodExecutionStartedEvent(flow_name="Demo", method_name="a", state={}) + ) + _emit_event( + MethodExecutionStartedEvent(flow_name="Demo", method_name="b", state={}) + ) + assert app._current_method == "b" + + # 'a' finishes while 'b' is still active → header stays on 'b'. + _emit_event( + MethodExecutionFinishedEvent( + flow_name="Demo", method_name="a", result=None, state={} + ) + ) + assert app._current_method == "b" + + # 'b' finishes → nothing active left → header clears. + _emit_event( + MethodExecutionFinishedEvent( + flow_name="Demo", method_name="b", result=None, state={} + ) + ) + assert app._current_method is None + finally: + app._unsubscribe() + + +def test_flow_method_transitions_clear_current_agent() -> None: + app = CrewRunApp(crew_name="Demo") + app._flow = SimpleNamespace() + app._subscribe() + try: + _emit_event( + MethodExecutionStartedEvent(flow_name="Demo", method_name="a", state={}) + ) + app._current_agent = "Researcher" # an agent ran during method 'a' + + # Starting a new method clears the previous method's agent. + _emit_event( + MethodExecutionStartedEvent(flow_name="Demo", method_name="b", state={}) + ) + assert app._current_agent == "" + + app._current_agent = "Writer" + # 'b' ending switches the active method ('a' still active) → agent clears. + _emit_event( + MethodExecutionFinishedEvent( + flow_name="Demo", method_name="b", result=None, state={} + ) + ) + assert app._current_agent == "" + finally: + app._unsubscribe() + + +def test_crew_kickoff_does_not_rename_flow_run() -> None: + # A `call: crew` step must not relabel the flow with the nested crew's name. + app = CrewRunApp(crew_name="My Flow") + app._flow = SimpleNamespace() + app._subscribe() + try: + _emit_event(CrewKickoffStartedEvent(crew_name="Nested Crew", inputs=None)) + assert app._crew_name == "My Flow" + assert app._status == "working" + finally: + app._unsubscribe() + + +def test_crew_kickoff_renames_in_crew_mode() -> None: + # Regression: crew runs still adopt the crew name from the event. + app = CrewRunApp(crew_name="Crew") + app._crew = SimpleNamespace() + app._subscribe() + try: + _emit_event(CrewKickoffStartedEvent(crew_name="Real Crew", inputs=None)) + assert app._crew_name == "Real Crew" + finally: + app._unsubscribe() + + +def test_method_paused_marks_step_paused() -> None: + app = CrewRunApp(crew_name="Demo") + app._flow = SimpleNamespace() + app._subscribe() + try: + _emit_event( + MethodExecutionStartedEvent(flow_name="Demo", method_name="ask", state={}) + ) + _emit_event( + MethodExecutionPausedEvent( + flow_name="Demo", + method_name="ask", + state={}, + flow_id="flow-1", + message="Need your input", + ) + ) + assert app._flow_steps == [ + {"name": "ask", "call_type": None, "status": "paused"} + ] + assert app._current_method == "ask" + finally: + app._unsubscribe() + + +@pytest.mark.asyncio +async def test_declarative_flow_runs_on_tui() -> None: + """End-to-end: on_mount dispatches _run_flow_worker → flow.kickoff → + _on_crew_done, and any still-active step is swept to done on completion.""" + kicked: dict[str, object] = {} + + class FakeFlow: + name = "Demo Flow" + + def kickoff(self, inputs=None): + kicked["inputs"] = inputs + return "flow result" + + app = CrewRunApp(crew_name="Demo Flow") + app._flow = FakeFlow() + app._flow_inputs = {"topic": "AI"} + # A step left active (no Finished event) must be swept to done by _on_crew_done. + app._flow_steps = [{"name": "compute", "call_type": "expression", "status": "active"}] + + async with app.run_test() as pilot: + for _ in range(100): + await pilot.pause(0.05) + if app._status == "completed": + break + + assert kicked["inputs"] == {"topic": "AI"} + assert app._status == "completed" + assert app._final_output == "flow result" + assert app._crew_result == "flow result" + assert app._flow_steps[0]["status"] == "done" diff --git a/lib/cli/tests/test_run_declarative_flow.py b/lib/cli/tests/test_run_declarative_flow.py index abc49396b..378156725 100644 --- a/lib/cli/tests/test_run_declarative_flow.py +++ b/lib/cli/tests/test_run_declarative_flow.py @@ -2,6 +2,7 @@ from __future__ import annotations import os from pathlib import Path +from types import SimpleNamespace import pytest @@ -9,6 +10,17 @@ import crewai_cli.input_prompt as input_prompt_module import crewai_cli.run_declarative_flow as run_declarative_flow_module +@pytest.fixture(autouse=True) +def _headless_by_default(monkeypatch: pytest.MonkeyPatch) -> None: + """Default these tests to the headless/terminal path. + + ``run_declarative_flow`` now launches the TUI when interactive, which can't + run under pytest; tests here assert the terminal/headless contract. Tests + that exercise TUI routing override ``is_dmn_mode_enabled`` explicitly. + """ + monkeypatch.setenv("CREWAI_DMN", "true") + + FLOW_YAML = """\ schema: crewai.flow/v1 name: TestFlow @@ -400,3 +412,202 @@ def test_id_restore_still_drops_unknown_keys( assert resolved == {"id": "run-123"} # id kept, typo dropped assert "Ignoring unknown input 'prospect_emai'" in captured.err assert "Ignoring unknown input 'id'" not in captured.err + + +# ── TUI vs terminal (headless/deploy) routing ────────────────────── + + +def _install_fake_flow_app(monkeypatch, *, status, want_deploy=False): + """Replace CrewRunApp/EventListener/summary so _run_declarative_flow_tui is + driven by a controllable fake app.""" + + class FakeEventListener: + pass + + class FakeApp: + def __init__(self, crew_name=""): + self._crew_name = crew_name + self._status = status + self._want_deploy = want_deploy + self._crew_result = "result" + + def run(self): + pass + + monkeypatch.setattr( + "crewai.events.event_listener.EventListener", FakeEventListener + ) + monkeypatch.setattr("crewai_cli.crew_run_tui.CrewRunApp", FakeApp) + monkeypatch.setattr( + run_declarative_flow_module, "_print_flow_post_tui_summary", lambda app: None + ) + + +def test_run_declarative_flow_dmn_uses_terminal( + tmp_path: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setenv("CREWAI_DMN", "true") + monkeypatch.setattr( + run_declarative_flow_module, + "_run_declarative_flow_tui", + lambda *a, **k: pytest.fail("DMN/headless mode must not launch the TUI"), + ) + path = _write(tmp_path, REQUIRED_FLOW_YAML) + + run_declarative_flow_module.run_declarative_flow( + str(path), '{"prospect_email":"a@b.com"}' + ) + + assert capsys.readouterr().out == "a@b.com\n" + + +def test_run_declarative_flow_interactive_uses_tui( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(run_declarative_flow_module, "is_interactive", lambda: True) + captured: dict[str, object] = {} + monkeypatch.setattr( + run_declarative_flow_module, + "_run_declarative_flow_tui", + lambda flow, resolved: captured.update(flow=flow, inputs=resolved), + ) + path = _write(tmp_path, REQUIRED_FLOW_YAML) + + run_declarative_flow_module.run_declarative_flow( + str(path), '{"prospect_email":"a@b.com"}' + ) + + assert captured["inputs"] == {"prospect_email": "a@b.com"} + assert captured["flow"] is not None + + +def test_run_declarative_flow_tui_failed_exits_nonzero( + monkeypatch: pytest.MonkeyPatch, +) -> None: + _install_fake_flow_app(monkeypatch, status="failed") + + with pytest.raises(SystemExit) as exc_info: + run_declarative_flow_module._run_declarative_flow_tui( + SimpleNamespace(name="Flow"), None + ) + + assert exc_info.value.code == 1 + + +def test_run_declarative_flow_tui_user_quit_exits_130( + monkeypatch: pytest.MonkeyPatch, +) -> None: + _install_fake_flow_app(monkeypatch, status="chatting") + exit_calls: list[int] = [] + monkeypatch.setattr(os, "_exit", lambda code: exit_calls.append(code)) + + run_declarative_flow_module._run_declarative_flow_tui( + SimpleNamespace(name="Flow"), None + ) + + assert exit_calls == [130] + + +def test_run_declarative_flow_tui_chains_deploy( + monkeypatch: pytest.MonkeyPatch, +) -> None: + _install_fake_flow_app(monkeypatch, status="completed", want_deploy=True) + deploy_calls: list[bool] = [] + monkeypatch.setattr( + "crewai_cli.run_crew._chain_deploy", lambda: deploy_calls.append(True) + ) + + run_declarative_flow_module._run_declarative_flow_tui( + SimpleNamespace(name="Flow"), None + ) + + assert deploy_calls == [True] + + +def test_run_declarative_flow_tui_no_deploy_when_not_requested( + monkeypatch: pytest.MonkeyPatch, +) -> None: + _install_fake_flow_app(monkeypatch, status="completed", want_deploy=False) + deploy_calls: list[bool] = [] + monkeypatch.setattr( + "crewai_cli.run_crew._chain_deploy", lambda: deploy_calls.append(True) + ) + + run_declarative_flow_module._run_declarative_flow_tui( + SimpleNamespace(name="Flow"), None + ) + + assert deploy_calls == [] + + +def test_run_declarative_flow_tui_enables_flow_events( + monkeypatch: pytest.MonkeyPatch, +) -> None: + # The STEPS panel depends on flow method events; a flow that declared + # suppress_flow_events must have it forced off for the interactive TUI run. + _install_fake_flow_app(monkeypatch, status="completed") + flow = SimpleNamespace(name="Flow", suppress_flow_events=True) + + run_declarative_flow_module._run_declarative_flow_tui(flow, None) + + assert flow.suppress_flow_events is False + + +def test_flow_uses_human_feedback_detection() -> None: + hf_flow = SimpleNamespace( + _definition=SimpleNamespace( + methods={ + "ask": SimpleNamespace(human_feedback=SimpleNamespace(emit=None)), + "plain": SimpleNamespace(human_feedback=None), + } + ) + ) + assert run_declarative_flow_module._flow_uses_human_feedback(hf_flow) is True + + no_hf = SimpleNamespace( + _definition=SimpleNamespace( + methods={"a": SimpleNamespace(human_feedback=None)} + ) + ) + assert run_declarative_flow_module._flow_uses_human_feedback(no_hf) is False + # No definition → False, no error. + assert run_declarative_flow_module._flow_uses_human_feedback(SimpleNamespace()) is False + + +def test_human_feedback_flow_uses_terminal_even_when_interactive( + tmp_path: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch +) -> None: + # A human-feedback flow must run on the terminal (blocking input / Rich + # prompts) even in an interactive session, never on the TUI. + monkeypatch.setattr(run_declarative_flow_module, "is_interactive", lambda: True) + monkeypatch.setattr( + run_declarative_flow_module, "_flow_uses_human_feedback", lambda flow: True + ) + monkeypatch.setattr( + run_declarative_flow_module, + "_run_declarative_flow_tui", + lambda *a, **k: pytest.fail("human-feedback flow must run on the terminal"), + ) + path = _write(tmp_path, FLOW_YAML) + + run_declarative_flow_module.run_declarative_flow(str(path), '{"topic":"AI"}') + + assert capsys.readouterr().out == "AI\n" + + +def test_flow_method_types_from_definition() -> None: + flow = SimpleNamespace( + _definition=SimpleNamespace( + methods={ + "fetch": SimpleNamespace(do=SimpleNamespace(call="expression")), + "research": SimpleNamespace(do=SimpleNamespace(call="crew")), + } + ) + ) + + assert run_declarative_flow_module._flow_method_types(flow) == { + "fetch": "expression", + "research": "crew", + } + # No definition → empty map, no error. + assert run_declarative_flow_module._flow_method_types(SimpleNamespace()) == {} diff --git a/lib/crewai/src/crewai/events/utils/console_formatter.py b/lib/crewai/src/crewai/events/utils/console_formatter.py index 604c7b051..858dde0ac 100644 --- a/lib/crewai/src/crewai/events/utils/console_formatter.py +++ b/lib/crewai/src/crewai/events/utils/console_formatter.py @@ -211,6 +211,13 @@ To enable tracing, do any one of these: """Print a panel with consistent formatting if verbose is enabled.""" panel = self.create_panel(content, title, style) if is_flow: + # A TUI (e.g. the CLI's CrewRunApp) owns the screen and renders flow + # progress in its own STEPS panel; emitting Rich panels here would + # interleave with and corrupt the TUI, so suppress them in TUI mode. + from crewai.events.listeners.tracing.utils import is_tui_mode + + if is_tui_mode(): + return self.print(panel) self.print() else: diff --git a/lib/crewai/tests/utilities/test_console_formatter_tui_mode.py b/lib/crewai/tests/utilities/test_console_formatter_tui_mode.py new file mode 100644 index 000000000..cee544748 --- /dev/null +++ b/lib/crewai/tests/utilities/test_console_formatter_tui_mode.py @@ -0,0 +1,46 @@ +"""Flow panels must be suppressed while a TUI owns the screen.""" + +from rich.text import Text + +from crewai.events.listeners.tracing.utils import set_tui_mode +from crewai.events.utils.console_formatter import ConsoleFormatter + + +def _make_formatter(monkeypatch): + fmt = ConsoleFormatter(verbose=True) + calls: list[object] = [] + monkeypatch.setattr(fmt, "print", lambda *a, **k: calls.append(a)) + return fmt, calls + + +def test_flow_panel_suppressed_in_tui_mode(monkeypatch): + fmt, calls = _make_formatter(monkeypatch) + set_tui_mode(True) + try: + fmt.print_panel(Text("x"), "🌊 Flow Started", "blue", is_flow=True) + finally: + set_tui_mode(False) + + assert calls == [] + + +def test_flow_panel_prints_when_not_tui_mode(monkeypatch): + fmt, calls = _make_formatter(monkeypatch) + set_tui_mode(False) + + fmt.print_panel(Text("x"), "🌊 Flow Started", "blue", is_flow=True) + + # Panel + trailing blank line. + assert len(calls) == 2 + + +def test_non_flow_panel_unaffected_by_tui_mode(monkeypatch): + # tui_mode only gates flow panels; regular panels still follow verbose. + fmt, calls = _make_formatter(monkeypatch) + set_tui_mode(True) + try: + fmt.print_panel(Text("x"), "Task", "blue", is_flow=False) + finally: + set_tui_mode(False) + + assert len(calls) == 2