Implement user input handling in Flows

[joaomdmoura]

Add self.ask() for conversational flows

Flows can now request user input mid-execution via self.ask(). This enables conversational patterns like gathering requirements, interactive troubleshooting, multi-turn dialogs, without breaking the flow lifecycle and requiring HITL.

What it does

class ResearchFlow(Flow):
    @start()
    def gather_and_research(self):
        topic = self.ask("What topic should we research?")
        depth = self.ask("How deep? (brief/detailed/exhaustive)")
        return crew.kickoff(inputs={"topic": topic, "depth": depth})

• self.ask(message, timeout=None, metadata=None) -- pauses the method, prompts the user, returns their response as a string. Works in both sync and async flow methods.
• Timeout -- returns None when expired, enabling while (msg := self.ask("You:", timeout=300)) is not None: loop patterns that always terminate.
• Metadata -- bidirectional. Send context to the provider (metadata={"user_id": "u123"}), receive context back from the provider via InputResponse(text="answer", metadata={"responded_by": "u456"}). Thread-safe by design (no shared mutable state).
• Auto-checkpoint -- persists self.state before each ask() call when persistence is configured, so previously gathered data survives crashes.
• Non-blocking -- the flow framework runs sync methods in a thread pool (asyncio.to_thread), so ask() never blocks the event loop. Parallel listeners execute concurrently even when one is waiting for input.

How it differs from @human_feedback

	self.ask()	@human_feedback
When	Mid-method, inline	Post-method, wraps return value
Purpose	Gather input	Review/approve output
Can be conditional, looped, called N times	Yes	No (once per decorated method)
Routes to listeners via emit	No	Yes
Persist and reload flow execution	No	Yes

They coexist -- same flow can use both.