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) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RBYGqJHC2TMC6fonFziuuh
This commit is contained in:
Joao Moura
2026-07-08 01:18:15 -07:00
parent 289686ab49
commit 260f85e1b4
4 changed files with 563 additions and 2 deletions

View File

@@ -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:

View File

@@ -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.

View File

@@ -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"

View File

@@ -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()) == {}