1  """Interactive prompt callbacks for terminal_tool integration.
  2  
  3  These bridge terminal_tool's interactive prompts (clarify, sudo, approval)
  4  into prompt_toolkit's event loop. Each function takes the HermesCLI instance
  5  as its first argument and uses its state (queues, app reference) to coordinate
  6  with the TUI.
  7  """
  8  
  9  import queue
 10  import time as _time
 11  import getpass
 12  
 13  from hermes_cli.banner import cprint, _DIM, _RST
 14  from hermes_cli.config import save_env_value_secure
 15  from hermes_constants import display_hermes_home
 16  
 17  
 18  def clarify_callback(cli, question, choices):
 19      """Prompt for clarifying question through the TUI.
 20  
 21      Sets up the interactive selection UI, then blocks until the user
 22      responds. Returns the user's choice or a timeout message.
 23      """
 24      from cli import CLI_CONFIG
 25  
 26      timeout = CLI_CONFIG.get("clarify", {}).get("timeout", 120)
 27      response_queue = queue.Queue()
 28      is_open_ended = not choices
 29  
 30      cli._clarify_state = {
 31          "question": question,
 32          "choices": choices if not is_open_ended else [],
 33          "selected": 0,
 34          "response_queue": response_queue,
 35      }
 36      cli._clarify_deadline = _time.monotonic() + timeout
 37      cli._clarify_freetext = is_open_ended
 38  
 39      if hasattr(cli, "_app") and cli._app:
 40          cli._app.invalidate()
 41  
 42      while True:
 43          try:
 44              result = response_queue.get(timeout=1)
 45              cli._clarify_deadline = 0
 46              return result
 47          except queue.Empty:
 48              remaining = cli._clarify_deadline - _time.monotonic()
 49              if remaining <= 0:
 50                  break
 51              if hasattr(cli, "_app") and cli._app:
 52                  cli._app.invalidate()
 53  
 54      cli._clarify_state = None
 55      cli._clarify_freetext = False
 56      cli._clarify_deadline = 0
 57      if hasattr(cli, "_app") and cli._app:
 58          cli._app.invalidate()
 59      cprint(f"\n{_DIM}(clarify timed out after {timeout}s — agent will decide){_RST}")
 60      return (
 61          "The user did not provide a response within the time limit. "
 62          "Use your best judgement to make the choice and proceed."
 63      )
 64  
 65  
 66  def prompt_for_secret(cli, var_name: str, prompt: str, metadata=None) -> dict:
 67      """Prompt for a secret value through the TUI (e.g. API keys for skills).
 68  
 69      Returns a dict with keys: success, stored_as, validated, skipped, message.
 70      The secret is stored in ~/.hermes/.env and never exposed to the model.
 71      """
 72      if not getattr(cli, "_app", None):
 73          if not hasattr(cli, "_secret_state"):
 74              cli._secret_state = None
 75          if not hasattr(cli, "_secret_deadline"):
 76              cli._secret_deadline = 0
 77          try:
 78              value = getpass.getpass(f"{prompt} (hidden, ESC or empty Enter to skip): ")
 79          except (EOFError, KeyboardInterrupt):
 80              value = ""
 81  
 82          if not value:
 83              cprint(f"\n{_DIM}  ⏭ Secret entry skipped{_RST}")
 84              return {
 85                  "success": True,
 86                  "reason": "cancelled",
 87                  "stored_as": var_name,
 88                  "validated": False,
 89                  "skipped": True,
 90                  "message": "Secret setup was skipped.",
 91              }
 92  
 93          stored = save_env_value_secure(var_name, value)
 94          _dhh = display_hermes_home()
 95          cprint(f"\n{_DIM}  ✓ Stored secret in {_dhh}/.env as {var_name}{_RST}")
 96          return {
 97              **stored,
 98              "skipped": False,
 99              "message": "Secret stored securely. The secret value was not exposed to the model.",
100          }
101  
102      timeout = 120
103      response_queue = queue.Queue()
104  
105      cli._secret_state = {
106          "var_name": var_name,
107          "prompt": prompt,
108          "metadata": metadata or {},
109          "response_queue": response_queue,
110      }
111      cli._secret_deadline = _time.monotonic() + timeout
112      # Avoid storing stale draft input as the secret when Enter is pressed.
113      if hasattr(cli, "_clear_secret_input_buffer"):
114          try:
115              cli._clear_secret_input_buffer()
116          except Exception:
117              pass
118      elif hasattr(cli, "_app") and cli._app:
119          try:
120              cli._app.current_buffer.reset()
121          except Exception:
122              pass
123  
124      if hasattr(cli, "_app") and cli._app:
125          cli._app.invalidate()
126  
127      while True:
128          try:
129              value = response_queue.get(timeout=1)
130              cli._secret_state = None
131              cli._secret_deadline = 0
132              if hasattr(cli, "_app") and cli._app:
133                  cli._app.invalidate()
134  
135              if not value:
136                  cprint(f"\n{_DIM}  ⏭ Secret entry skipped{_RST}")
137                  return {
138                      "success": True,
139                      "reason": "cancelled",
140                      "stored_as": var_name,
141                      "validated": False,
142                      "skipped": True,
143                      "message": "Secret setup was skipped.",
144                  }
145  
146              stored = save_env_value_secure(var_name, value)
147              _dhh = display_hermes_home()
148              cprint(f"\n{_DIM}  ✓ Stored secret in {_dhh}/.env as {var_name}{_RST}")
149              return {
150                  **stored,
151                  "skipped": False,
152                  "message": "Secret stored securely. The secret value was not exposed to the model.",
153              }
154          except queue.Empty:
155              remaining = cli._secret_deadline - _time.monotonic()
156              if remaining <= 0:
157                  break
158              if hasattr(cli, "_app") and cli._app:
159                  cli._app.invalidate()
160  
161      cli._secret_state = None
162      cli._secret_deadline = 0
163      if hasattr(cli, "_clear_secret_input_buffer"):
164          try:
165              cli._clear_secret_input_buffer()
166          except Exception:
167              pass
168      elif hasattr(cli, "_app") and cli._app:
169          try:
170              cli._app.current_buffer.reset()
171          except Exception:
172              pass
173      if hasattr(cli, "_app") and cli._app:
174          cli._app.invalidate()
175      cprint(f"\n{_DIM}  ⏱ Timeout — secret capture cancelled{_RST}")
176      return {
177          "success": True,
178          "reason": "timeout",
179          "stored_as": var_name,
180          "validated": False,
181          "skipped": True,
182          "message": "Secret setup timed out and was skipped.",
183      }
184  
185  
186  def approval_callback(cli, command: str, description: str) -> str:
187      """Prompt for dangerous command approval through the TUI.
188  
189      Shows a selection UI with choices: once / session / always / deny.
190      When the command is longer than 70 characters, a "view" option is
191      included so the user can reveal the full text before deciding.
192  
193      Uses cli._approval_lock to serialize concurrent requests (e.g. from
194      parallel delegation subtasks) so each prompt gets its own turn.
195      """
196      lock = getattr(cli, "_approval_lock", None)
197      if lock is None:
198          import threading
199          cli._approval_lock = threading.Lock()
200          lock = cli._approval_lock
201  
202      with lock:
203          from cli import CLI_CONFIG
204          timeout = CLI_CONFIG.get("approvals", {}).get("timeout", 60)
205          response_queue = queue.Queue()
206          choices = ["once", "session", "always", "deny"]
207          if len(command) > 70:
208              choices.append("view")
209  
210          cli._approval_state = {
211              "command": command,
212              "description": description,
213              "choices": choices,
214              "selected": 0,
215              "response_queue": response_queue,
216          }
217          cli._approval_deadline = _time.monotonic() + timeout
218  
219          if hasattr(cli, "_app") and cli._app:
220              cli._app.invalidate()
221  
222          while True:
223              try:
224                  result = response_queue.get(timeout=1)
225                  cli._approval_state = None
226                  cli._approval_deadline = 0
227                  if hasattr(cli, "_app") and cli._app:
228                      cli._app.invalidate()
229                  return result
230              except queue.Empty:
231                  remaining = cli._approval_deadline - _time.monotonic()
232                  if remaining <= 0:
233                      break
234                  if hasattr(cli, "_app") and cli._app:
235                      cli._app.invalidate()
236  
237          cli._approval_state = None
238          cli._approval_deadline = 0
239          if hasattr(cli, "_app") and cli._app:
240              cli._app.invalidate()
241          cprint(f"\n{_DIM}  ⏱ Timeout — denying command{_RST}")
242          return "deny"