vibe.assistant.services.turn_orchestrator

TurnOrchestrator - pure Python logic for turn validation and auto-reply.

Responsibilities: - Turn validation (finalize preconditions, followup requirements) - History analysis (checking for previous questions) - Auto-reply decision logic - Question summary composition

Note: This is pure Python logic without Flask dependencies. SSE streaming and template rendering remain in AssistantService.

TurnValidationResult

Result of turn validation.

HistoryAnalyzer

Analyzes conversation history for turn validation.

history_contains_questions 

history_contains_questions(history: list[Message] | None) -> bool

Return True if conversation history already includes ask_question tool calls.

Parameters:
  • history (list[Message] | None) –

    List of Message objects from session state
Returns:
  • bool –

    True if history contains LLM-generated ask_question calls

TurnResultBuilder

Builds summaries and extracts information from tool calls.

__init__ 

__init__(question_definitions: dict[str, Any] | None = None) -> None

Initialize the builder.

Parameters:
  • question_definitions (dict[str, Any] | None, default: None ) –

    Dict of predefined question definitions

tool_calls_include 

tool_calls_include(tool_calls: list[dict[str, Any]] | None, tool_name: str) -> bool

Return True if the provided tool name appears in the tool call list.

Parameters:
  • tool_calls (list[dict[str, Any]] | None) –

    List of tool call dicts

  • tool_name (str) –

    Name of tool to search for
Returns:
  • bool –

    True if tool_name found in tool_calls

compose_question_summary 

compose_question_summary(tool_calls: list[dict[str, Any]]) -> list[str] | None

Generate a short summary of questions asked this turn for fallback chat updates.

Parameters:
  • tool_calls (list[dict[str, Any]]) –

    List of tool call dicts from this turn
Returns:
  • list[str] | None –

    List of question strings, or None if no questions

TurnValidator

Validates turn results and enforces business rules.

__init__ 

__init__(auto_reply_enabled: bool = True, history: list[Message] | None = None) -> None

Initialize the validator.

Parameters:
  • auto_reply_enabled (bool, default: True ) –

    Whether auto-reply is enabled

  • history (list[Message] | None, default: None ) –

    Conversation history for validation

validate_turn_result 

validate_turn_result(tool_calls: list[dict[str, Any]], turn_has_question: bool, draft_blocks: list[dict[str, Any]]) -> TurnValidationResult

Validate the turn result and enforce business rules.

Parameters:
  • tool_calls (list[dict[str, Any]]) –

    Tool calls from this turn

  • turn_has_question (bool) –

    Whether this turn asked a question

  • draft_blocks (list[dict[str, Any]]) –

    Current draft blocks
Returns:
Raises:

AutoReplyManager

Manages auto-reply decisions and message generation.

__init__ 

__init__(max_auto_replies: int = 3) -> None

Initialize the manager.

Parameters:
  • max_auto_replies (int, default: 3 ) –

    Maximum number of auto-replies allowed

should_auto_reply 

should_auto_reply(exception: Exception, current_count: int) -> bool

Determine if we should auto-reply for this exception.

Parameters:
  • exception (Exception) –

    The exception that was raised

  • current_count (int) –

    Current auto-reply count
Returns:
  • bool –

    True if should auto-reply, False otherwise

get_auto_reply_message 

get_auto_reply_message(exception: PrematureFinalizeException | NoFollowUpException) -> str

Get the auto-reply message from an exception.

Parameters:
Returns:
  • str –

    Auto-reply message string