diff --git a/lib/cli/src/crewai_cli/crew_run_tui.py b/lib/cli/src/crewai_cli/crew_run_tui.py index cefb7d142..35767cdb6 100644 --- a/lib/cli/src/crewai_cli/crew_run_tui.py +++ b/lib/cli/src/crewai_cli/crew_run_tui.py @@ -568,12 +568,27 @@ 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 + # ── Layout ────────────────────────────────────────────── def compose(self) -> ComposeResult: @@ -602,6 +617,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 +698,33 @@ 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") + 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 _on_crew_done(self, output: str | None) -> None: with self._lock: self._status = "completed" @@ -694,6 +738,9 @@ 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": @@ -739,6 +786,9 @@ 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": @@ -1156,6 +1206,42 @@ 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) + 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 +1311,40 @@ 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: + 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 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) + t.append("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 +1959,12 @@ 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, + MethodExecutionStartedEvent, + ) from crewai.events.types.llm_events import ( LLMCallCompletedEvent, LLMCallStartedEvent, @@ -1879,6 +2005,50 @@ FooterKey .footer-key--key { 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 + 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._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._register_handler(MethodExecutionFailedEvent, on_method_failed) + @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..7b6ff8347 100644 --- a/lib/cli/src/crewai_cli/run_declarative_flow.py +++ b/lib/cli/src/crewai_cli/run_declarative_flow.py @@ -66,17 +66,151 @@ 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. + if is_interactive(): + _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: 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() + + app = CrewRunApp(crew_name=getattr(flow, "name", None) 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_method_types(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: + 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 + 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..489ea1358 100644 --- a/lib/cli/tests/test_crew_run_tui.py +++ b/lib/cli/tests/test_crew_run_tui.py @@ -6,6 +6,12 @@ from unittest.mock import Mock import pytest from crewai.events.event_bus import crewai_event_bus +from crewai.events.types.flow_events import ( + FlowStartedEvent, + MethodExecutionFailedEvent, + MethodExecutionFinishedEvent, + MethodExecutionStartedEvent, +) from crewai.events.types.memory_events import ( MemorySaveCompletedEvent, MemorySaveFailedEvent, @@ -1481,3 +1487,98 @@ 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"}, + ] + + +@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..c0b86fc42 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,147 @@ 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_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()) == {}