latent.chat.controller¶

ChatController — pure-Python REPL logic with zero Textual imports.

Owns message list management, event-to-output processing, slash command dispatch, usage aggregation, cost tracking, and session save orchestration. Communicates with the UI through the ChatOutput protocol.

Classes¶

ChatController¶

ChatController(agent: BaseAgent, output: ChatOutput, session_id: str = new_session_id(), messages: list[Message] = list(), turn_count: int = 0, total_input_tokens: int = 0, total_output_tokens: int = 0, total_cost: float = 0.0, tools_visible: bool = True, pricing_data: list[dict] | None = None, _streaming: bool = False)

Pure-Python REPL controller — no Textual imports.

ChatOutput¶

ChatOutput()

UI-agnostic output sink. Textual App implements this; tests use a stub.

TurnMetrics¶

TurnMetrics(input_tokens: int = 0, output_tokens: int = 0, model_id: str | None = None, response_time_ms: float = 0, ttft_ms: float | None = None, cost: float | None = None)

Per-turn token and timing metrics.

Methods¶

ChatController.cancel_streaming¶

cancel_streaming() -> None

Called by UI on Ctrl+C to stop consuming the stream.

ChatController.handle_input¶

handle_input(text: str) -> None

Entry point for user input. Routes to slash command or agent.

ChatController.session_path¶

ChatOutput.clear_display¶

clear_display() -> None

ChatOutput.dismiss_history_picker¶

dismiss_history_picker() -> None

ChatOutput.hide_loading¶

hide_loading() -> None

ChatOutput.show_history_picker¶

show_history_picker(sessions: list[SessionInfo]) -> None

ChatOutput.show_loading¶

show_loading() -> None

ChatOutput.write_assistant_complete¶

write_assistant_complete(text: str) -> None

ChatOutput.write_assistant_text¶

write_assistant_text(text: str) -> None

ChatOutput.write_error¶

write_error(message: str) -> None

ChatOutput.write_info¶

write_info(text: str) -> None

ChatOutput.write_reasoning¶

write_reasoning(text: str) -> None

ChatOutput.write_status¶

write_status(text: str) -> None

ChatOutput.write_tool_call¶

write_tool_call(name: str, arguments: dict[str, Any], tool_call_id: str) -> None

ChatOutput.write_tool_result¶

write_tool_result(output: Any, error: str | None, tool_call_id: str) -> None

ChatOutput.write_user_message¶

write_user_message(text: str) -> None