calute.operators.user_prompt#

Interactive user-question manager for operator tooling.

Provides UserPromptManager, which manages one live user clarification question at a time. The ask_user operator tool creates a PendingUserPrompt via UserPromptManager.request() and awaits a future that the TUI resolves by calling UserPromptManager.answer().

class calute.operators.user_prompt.UserPromptManager[source]#

Bases: object

Manage one live user clarification question at a time.

At most one question can be pending. The ask_user tool calls request(), which creates a PendingUserPrompt and returns a future. The TUI polls get_pending() to discover the question and calls answer() to resolve the future.

_pending#

The currently active PendingUserPrompt, or None when no question is outstanding.

_pending_future#

The asyncio.Future that will be resolved with the answer dictionary when the user responds.

answer(raw_input: str) dict[str, Any][source]#

Resolve the pending question from typed UI input.

Matches the user’s raw input against the pending question’s options (by index or label/value text) and resolves the internal future so that the awaiting ask_user tool call receives the result.

Parameters

raw_input – The raw string entered by the user in the TUI. May be a numeric index (1-based) referring to a listed option, the full option label or value text, or arbitrary freeform text when allowed.

Returns

  • request_id: The prompt request identifier.

  • question: The original question text.

  • answer: The normalised answer value.

  • raw_input: The cleaned user input.

  • selected_option: The matched option dictionary, or None if no listed option was selected.

  • used_freeform: True when the answer did not match any listed option.

Return type

A dictionary containing

Raises

ValueError – If the input is empty, or if freeform is disallowed and the input does not match any option.

get_pending() dict[str, Any] | None[source]#

Return the current pending question, if any.

Returns

A serialised dictionary of the pending prompt (via PendingUserPrompt.to_dict()), or None when no question is outstanding.

has_pending() bool[source]#

Report whether the runtime is currently waiting on the user.

Returns

True if a question is pending, False otherwise.

async request(question: str, *, options: list[str] | None = None, allow_freeform: bool = True, placeholder: str | None = None) dict[str, Any][source]#

Create a pending question and wait until the UI submits an answer.

Builds a PendingUserPrompt, stores it as the current pending prompt, creates an asyncio.Future, and awaits it. The future is resolved when answer() is called by the TUI layer.

Parameters
  • question – The question text to display to the user.

  • options – Optional list of string choices presented as numbered options. Each string is converted into a UserPromptOption.

  • allow_freeform – When True, the user may type a custom answer in addition to (or instead of) the listed options.

  • placeholder – Optional hint text shown in the input field while waiting for a response.

Returns

The resolved answer dictionary containing fields such as request_id, question, answer, raw_input, selected_option, and used_freeform.

Raises

RuntimeError – If another question is already pending.