aac74be2f1
When an approval prompt expired without a response, every CLI-side path
collapsed the timeout into the same 'deny' choice as an explicit user
refusal, so the agent was told the user denied the action when the user
simply never answered. The gateway wait already distinguished the two
('timed out without user response... Silence is not consent.'); this
brings the CLI/TUI/ACP surfaces to parity.
- prompt_dangerous_approval(): input()-path expiry now returns a distinct
'timeout' choice (still fail-closed).
- cli.py _approval_callback + hermes_cli/callbacks.py approval_callback:
deadline expiry returns 'timeout' instead of 'deny'.
- check_all_command_guards / _run_approval_gate CLI tails: 'timeout' maps
to outcome='timeout' with a 'timed out without user response... Silence
is not consent.' BLOCKED message (matching the gateway wording);
explicit deny keeps outcome='denied' and gains user_consent=False for
shape parity.
- computer_use: 'timeout' verdict threads through the CLI adapter and
yields a 'prompt timed out — the user did not respond' error instead of
'denied by user'.
- ACP permissions bridge: FutureTimeout returns 'timeout' (other failures
still 'deny'); elicitation maps 'timeout' to 'cancel' like the gateway's
unresolved outcome; codex wire mapping documents deny/timeout→decline.
- write_approval already treats unknown choices as 'stage, not drop', so
a timeout now stages the memory write instead of silently refusing it.
Every timeout path remains fail-closed — the action never runs; only the
classification reported to the agent changes.
254 lines
8.8 KiB
Python
254 lines
8.8 KiB
Python
"""Interactive prompt callbacks for terminal_tool integration.
|
|
|
|
These bridge terminal_tool's interactive prompts (clarify, sudo, approval)
|
|
into prompt_toolkit's event loop. Each function takes the HermesCLI instance
|
|
as its first argument and uses its state (queues, app reference) to coordinate
|
|
with the TUI.
|
|
"""
|
|
|
|
import queue
|
|
import time as _time
|
|
|
|
from hermes_cli.banner import cprint, _DIM, _RST
|
|
from hermes_cli.config import save_env_value_secure
|
|
from hermes_cli.secret_prompt import masked_secret_prompt
|
|
from hermes_constants import display_hermes_home
|
|
|
|
|
|
def clarify_callback(cli, question, choices, multi_select=False):
|
|
"""Prompt for clarifying question through the TUI.
|
|
|
|
Sets up the interactive selection UI, then blocks until the user
|
|
responds. Returns the user's choice or a timeout message.
|
|
|
|
When ``multi_select`` is True, shows checkboxes and the user can
|
|
select multiple options with Space, confirming with Enter.
|
|
"""
|
|
from cli import CLI_CONFIG
|
|
from tools.clarify_gateway import resolve_clarify_timeout
|
|
|
|
# Canonical clarify timeout, shared with the gateway/TUI path. `<= 0`
|
|
# means unlimited (never auto-skip mid-think) → a null deadline.
|
|
timeout = resolve_clarify_timeout(CLI_CONFIG)
|
|
response_queue = queue.Queue()
|
|
is_open_ended = not choices
|
|
effective_multi = multi_select and not is_open_ended
|
|
|
|
cli._clarify_state = {
|
|
"question": question,
|
|
"choices": choices if not is_open_ended else [],
|
|
"selected": 0,
|
|
"multi_select": effective_multi,
|
|
"selected_indices": set() if effective_multi else None,
|
|
"response_queue": response_queue,
|
|
}
|
|
cli._clarify_deadline = None if timeout <= 0 else _time.monotonic() + timeout
|
|
cli._clarify_freetext = is_open_ended
|
|
|
|
if hasattr(cli, "_app") and cli._app:
|
|
cli._app.invalidate()
|
|
|
|
while True:
|
|
try:
|
|
result = response_queue.get(timeout=1)
|
|
cli._clarify_deadline = None
|
|
return result
|
|
except queue.Empty:
|
|
# None deadline = unlimited: never auto-skip, just keep polling.
|
|
if cli._clarify_deadline is not None:
|
|
remaining = cli._clarify_deadline - _time.monotonic()
|
|
if remaining <= 0:
|
|
break
|
|
if hasattr(cli, "_app") and cli._app:
|
|
cli._app.invalidate()
|
|
|
|
cli._clarify_state = None
|
|
cli._clarify_freetext = False
|
|
cli._clarify_deadline = None
|
|
if hasattr(cli, "_app") and cli._app:
|
|
cli._app.invalidate()
|
|
cprint(f"\n{_DIM}(clarify timed out after {timeout}s — agent will decide){_RST}")
|
|
return (
|
|
"The user did not provide a response within the time limit. "
|
|
"Use your best judgement to make the choice and proceed."
|
|
)
|
|
|
|
|
|
def prompt_for_secret(cli, var_name: str, prompt: str, metadata=None) -> dict:
|
|
"""Prompt for a secret value through the TUI (e.g. API keys for skills).
|
|
|
|
Returns a dict with keys: success, stored_as, validated, skipped, message.
|
|
The secret is stored in ~/.hermes/.env and never exposed to the model.
|
|
"""
|
|
if not getattr(cli, "_app", None):
|
|
if not hasattr(cli, "_secret_state"):
|
|
cli._secret_state = None
|
|
if not hasattr(cli, "_secret_deadline"):
|
|
cli._secret_deadline = 0
|
|
try:
|
|
value = masked_secret_prompt(f"{prompt} (hidden, ESC or empty Enter to skip): ")
|
|
except (EOFError, KeyboardInterrupt):
|
|
value = ""
|
|
|
|
if not value:
|
|
cprint(f"\n{_DIM} ⏭ Secret entry skipped{_RST}")
|
|
return {
|
|
"success": True,
|
|
"reason": "cancelled",
|
|
"stored_as": var_name,
|
|
"validated": False,
|
|
"skipped": True,
|
|
"message": "Secret setup was skipped.",
|
|
}
|
|
|
|
stored = save_env_value_secure(var_name, value)
|
|
_dhh = display_hermes_home()
|
|
cprint(f"\n{_DIM} ✓ Stored secret in {_dhh}/.env as {var_name}{_RST}")
|
|
return {
|
|
**stored,
|
|
"skipped": False,
|
|
"message": "Secret stored securely. The secret value was not exposed to the model.",
|
|
}
|
|
|
|
timeout = 120
|
|
response_queue = queue.Queue()
|
|
|
|
cli._secret_state = {
|
|
"var_name": var_name,
|
|
"prompt": prompt,
|
|
"metadata": metadata or {},
|
|
"response_queue": response_queue,
|
|
}
|
|
cli._secret_deadline = _time.monotonic() + timeout
|
|
# Avoid storing stale draft input as the secret when Enter is pressed.
|
|
if hasattr(cli, "_clear_secret_input_buffer"):
|
|
try:
|
|
cli._clear_secret_input_buffer()
|
|
except Exception:
|
|
pass
|
|
elif hasattr(cli, "_app") and cli._app:
|
|
try:
|
|
cli._app.current_buffer.reset()
|
|
except Exception:
|
|
pass
|
|
|
|
if hasattr(cli, "_app") and cli._app:
|
|
cli._app.invalidate()
|
|
|
|
while True:
|
|
try:
|
|
value = response_queue.get(timeout=1)
|
|
cli._secret_state = None
|
|
cli._secret_deadline = 0
|
|
if hasattr(cli, "_app") and cli._app:
|
|
cli._app.invalidate()
|
|
|
|
if not value:
|
|
cprint(f"\n{_DIM} ⏭ Secret entry skipped{_RST}")
|
|
return {
|
|
"success": True,
|
|
"reason": "cancelled",
|
|
"stored_as": var_name,
|
|
"validated": False,
|
|
"skipped": True,
|
|
"message": "Secret setup was skipped.",
|
|
}
|
|
|
|
stored = save_env_value_secure(var_name, value)
|
|
_dhh = display_hermes_home()
|
|
cprint(f"\n{_DIM} ✓ Stored secret in {_dhh}/.env as {var_name}{_RST}")
|
|
return {
|
|
**stored,
|
|
"skipped": False,
|
|
"message": "Secret stored securely. The secret value was not exposed to the model.",
|
|
}
|
|
except queue.Empty:
|
|
remaining = cli._secret_deadline - _time.monotonic()
|
|
if remaining <= 0:
|
|
break
|
|
if hasattr(cli, "_app") and cli._app:
|
|
cli._app.invalidate()
|
|
|
|
cli._secret_state = None
|
|
cli._secret_deadline = 0
|
|
if hasattr(cli, "_clear_secret_input_buffer"):
|
|
try:
|
|
cli._clear_secret_input_buffer()
|
|
except Exception:
|
|
pass
|
|
elif hasattr(cli, "_app") and cli._app:
|
|
try:
|
|
cli._app.current_buffer.reset()
|
|
except Exception:
|
|
pass
|
|
if hasattr(cli, "_app") and cli._app:
|
|
cli._app.invalidate()
|
|
cprint(f"\n{_DIM} ⏱ Timeout — secret capture cancelled{_RST}")
|
|
return {
|
|
"success": True,
|
|
"reason": "timeout",
|
|
"stored_as": var_name,
|
|
"validated": False,
|
|
"skipped": True,
|
|
"message": "Secret setup timed out and was skipped.",
|
|
}
|
|
|
|
|
|
def approval_callback(cli, command: str, description: str) -> str:
|
|
"""Prompt for dangerous command approval through the TUI.
|
|
|
|
Shows a selection UI with choices: once / session / always / deny.
|
|
When the command is longer than 70 characters, a "view" option is
|
|
included so the user can reveal the full text before deciding.
|
|
|
|
Uses cli._approval_lock to serialize concurrent requests (e.g. from
|
|
parallel delegation subtasks) so each prompt gets its own turn.
|
|
"""
|
|
lock = getattr(cli, "_approval_lock", None)
|
|
if lock is None:
|
|
import threading
|
|
cli._approval_lock = threading.Lock()
|
|
lock = cli._approval_lock
|
|
|
|
with lock:
|
|
from cli import CLI_CONFIG
|
|
timeout = CLI_CONFIG.get("approvals", {}).get("timeout", 300)
|
|
response_queue = queue.Queue()
|
|
choices = ["once", "session", "always", "deny"]
|
|
if len(command) > 70:
|
|
choices.append("view")
|
|
|
|
cli._approval_state = {
|
|
"command": command,
|
|
"description": description,
|
|
"choices": choices,
|
|
"selected": 0,
|
|
"response_queue": response_queue,
|
|
}
|
|
cli._approval_deadline = _time.monotonic() + timeout
|
|
|
|
if hasattr(cli, "_app") and cli._app:
|
|
cli._app.invalidate()
|
|
|
|
while True:
|
|
try:
|
|
result = response_queue.get(timeout=1)
|
|
cli._approval_state = None
|
|
cli._approval_deadline = 0
|
|
if hasattr(cli, "_app") and cli._app:
|
|
cli._app.invalidate()
|
|
return result
|
|
except queue.Empty:
|
|
remaining = cli._approval_deadline - _time.monotonic()
|
|
if remaining <= 0:
|
|
break
|
|
if hasattr(cli, "_app") and cli._app:
|
|
cli._app.invalidate()
|
|
|
|
cli._approval_state = None
|
|
cli._approval_deadline = 0
|
|
if hasattr(cli, "_app") and cli._app:
|
|
cli._app.invalidate()
|
|
cprint(f"\n{_DIM} ⏱ Timeout — denying command{_RST}")
|
|
return "timeout"
|