feat(cli): run declarative flows on the TUI (headless terminal fallback) (#6484)

* 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

* 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) <noreply@anthropic.com>
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) <noreply@anthropic.com>
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) <noreply@anthropic.com>
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) <noreply@anthropic.com>
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) <noreply@anthropic.com>
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) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RBYGqJHC2TMC6fonFziuuh

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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"

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