-
Notifications
You must be signed in to change notification settings - Fork 73
Expand file tree
/
Copy pathconfirmation.py
More file actions
155 lines (117 loc) · 4.74 KB
/
confirmation.py
File metadata and controls
155 lines (117 loc) · 4.74 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
#!/usr/bin/env python3
"""
Confirmation mode functionality for OpenHands CLI.
Provides user confirmation prompts before executing commands.
"""
from __future__ import annotations
import asyncio
from typing import Any, cast
from prompt_toolkit import Application, print_formatted_text
from prompt_toolkit.formatted_text import HTML
from prompt_toolkit.key_binding import KeyBindings
from prompt_toolkit.key_binding.key_processor import KeyPressEvent
from prompt_toolkit.layout.containers import HSplit, Window
from prompt_toolkit.layout.controls import FormattedTextControl
from prompt_toolkit.layout.dimension import Dimension
from prompt_toolkit.layout.layout import Layout
from openhands_cli.pt_style import get_cli_style
class ConfirmationMode:
"""Manages confirmation mode settings and behavior."""
def __init__(self) -> None:
self.enabled = True # Whether confirmation is enabled
def should_confirm(self) -> bool:
"""Determine if actions should require confirmation."""
return self.enabled
def set_enabled(self, enabled: bool) -> None:
"""Enable or disable confirmation mode."""
self.enabled = enabled
class UserCancelledError(Exception):
"""Raised when the user cancels an operation."""
pass
def cli_confirm(
question: str = "Are you sure?",
choices: list[str] | None = None,
initial_selection: int = 0,
) -> int:
"""Display a confirmation prompt with the given question and choices.
Returns the index of the selected choice.
"""
if choices is None:
choices = ["Yes", "No"]
selected = [initial_selection] # Using list to allow modification in closure
def get_choice_text() -> list:
return [
("class:question", f"{question}\n\n"),
] + [
(
"class:selected" if i == selected[0] else "class:unselected",
f"{'>' if i == selected[0] else ' '} {choice}\n",
)
for i, choice in enumerate(choices)
]
kb = KeyBindings()
@kb.add("up")
def _handle_up(event: KeyPressEvent) -> None:
selected[0] = (selected[0] - 1) % len(choices)
@kb.add("k") # Vi-style up
def _handle_k(event: KeyPressEvent) -> None:
selected[0] = (selected[0] - 1) % len(choices)
@kb.add("down")
def _handle_down(event: KeyPressEvent) -> None:
selected[0] = (selected[0] + 1) % len(choices)
@kb.add("j") # Vi-style down
def _handle_j(event: KeyPressEvent) -> None:
selected[0] = (selected[0] + 1) % len(choices)
@kb.add("enter")
def _handle_enter(event: KeyPressEvent) -> None:
event.app.exit(result=selected[0])
@kb.add("escape")
def _handle_escape(event: KeyPressEvent) -> None:
event.app.exit(exception=UserCancelledError())
@kb.add("c-c")
def _handle_ctrl_c(event: KeyPressEvent) -> None:
event.app.exit(exception=UserCancelledError())
# Create layout
content_window = Window(
FormattedTextControl(get_choice_text),
always_hide_cursor=True,
height=Dimension(max=8), # Limit height to prevent screen takeover
)
layout = Layout(HSplit([content_window]))
app: Application[int] = Application(
layout=layout,
key_bindings=kb,
style=get_cli_style(),
full_screen=False,
)
return cast(int, app.run())
async def read_confirmation_input() -> str:
"""Read user confirmation input."""
try:
question = "The agent wants to execute a command. Do you want to proceed?"
choices = [
"Yes, proceed",
"No (and allow to enter instructions)",
"Always proceed (don't ask again)",
]
choice_mapping = {0: "yes", 1: "no", 2: "always"}
# Run the confirmation dialog in a thread to keep the event loop responsive
index = await asyncio.to_thread(cli_confirm, question, choices, 0)
return choice_mapping.get(index, "no")
except (KeyboardInterrupt, EOFError, UserCancelledError):
return "no"
def display_action_info(action_type: str, action_data: dict[str, Any]) -> None:
"""Display information about the action to be executed."""
if action_type == "execute_bash":
command = action_data.get("command", "Unknown command")
print_formatted_text(HTML(f"<yellow>Command to execute: {command}</yellow>"))
elif action_type == "str_replace_editor":
command = action_data.get("command", "unknown")
path = action_data.get("path", "unknown file")
print_formatted_text(
HTML(f"<yellow>File operation: {command} on {path}</yellow>")
)
else:
print_formatted_text(HTML(f"<yellow>Action: {action_type}</yellow>"))
# Global confirmation mode instance
confirmation_mode = ConfirmationMode()