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