Repository navigation
Expand file tree
/
Copy pathonboarding.py
More file actions
300 lines (236 loc) · 9.54 KB
/
Copy pathonboarding.py
File metadata and controls
300 lines (236 loc) · 9.54 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
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
"""First-run onboarding state for the interactive TUI."""
from __future__ import annotations
import json
import logging
import os
from typing import TYPE_CHECKING
from deepagents_code._env_vars import ONBOARDING, classify_env_bool
from deepagents_code.model_config import DEFAULT_STATE_DIR
if TYPE_CHECKING:
from pathlib import Path
logger = logging.getLogger(__name__)
ONBOARDING_MARKER_FILENAME = "onboarding_complete"
"""Marker filename under `~/.deepagents/.state` after onboarding has completed."""
GOAL_AUTO_ACCEPT_PROMPT_MARKER_FILENAME = "goal_auto_accept_criteria_prompted_v1"
"""Marker written after the first goal-criteria preference prompt is answered."""
ONBOARDING_NAME_MEMORY_START = "<!-- deepagents:onboarding-name:start -->"
"""Start marker for the managed onboarding name memory block."""
ONBOARDING_NAME_MEMORY_END = "<!-- deepagents:onboarding-name:end -->"
"""End marker for the managed onboarding name memory block."""
def onboarding_marker_path(state_dir: Path | None = None) -> Path:
"""Return the first-run onboarding marker path.
Args:
state_dir: Optional state directory override for tests.
Returns:
Path to the onboarding completion marker.
"""
return (state_dir or DEFAULT_STATE_DIR) / ONBOARDING_MARKER_FILENAME
def has_completed_onboarding(state_dir: Path | None = None) -> bool:
"""Return whether the user has completed onboarding.
Args:
state_dir: Optional state directory override for tests.
Returns:
`True` when the onboarding marker exists, otherwise `False`.
"""
try:
return onboarding_marker_path(state_dir).exists()
except OSError:
logger.warning("Could not inspect onboarding marker", exc_info=True)
return False
def mark_onboarding_complete(state_dir: Path | None = None) -> bool:
"""Persist that onboarding has completed.
Args:
state_dir: Optional state directory override for tests.
Returns:
`True` when the marker was written, otherwise `False`.
"""
path = onboarding_marker_path(state_dir)
try:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("1\n", encoding="utf-8")
except OSError:
logger.warning("Could not write onboarding marker at %s", path, exc_info=True)
return False
return True
def goal_auto_accept_prompt_marker_path(state_dir: Path | None = None) -> Path:
"""Return the goal criteria preference prompt marker path.
Args:
state_dir: Optional state directory override for tests.
Returns:
Path to the versioned one-time prompt marker.
"""
return (state_dir or DEFAULT_STATE_DIR) / GOAL_AUTO_ACCEPT_PROMPT_MARKER_FILENAME
def has_shown_goal_auto_accept_prompt(state_dir: Path | None = None) -> bool:
"""Return whether the goal criteria preference prompt was answered.
Args:
state_dir: Optional state directory override for tests.
Returns:
`True` when the prompt marker exists, otherwise `False`.
"""
try:
return goal_auto_accept_prompt_marker_path(state_dir).exists()
except OSError:
logger.warning("Could not inspect goal preference prompt marker", exc_info=True)
return False
def mark_goal_auto_accept_prompt_shown(state_dir: Path | None = None) -> bool:
"""Persist that the goal criteria preference prompt was answered.
Args:
state_dir: Optional state directory override for tests.
Returns:
`True` when the marker was written, otherwise `False`.
"""
path = goal_auto_accept_prompt_marker_path(state_dir)
try:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("1\n", encoding="utf-8")
except OSError:
logger.warning(
"Could not write goal preference prompt marker at %s",
path,
exc_info=True,
)
return False
return True
def write_onboarding_name_memory(
name: str,
assistant_id: str,
*,
memory_path: Path | None = None,
) -> bool:
"""Persist the optional onboarding name into user agent memory.
Empty or whitespace-only names are skipped (no file is written).
Args:
name: Submitted user name.
assistant_id: Agent identifier whose user memory should be updated.
memory_path: Optional memory file override for tests.
Returns:
`True` when memory was written, otherwise `False`.
"""
clean = _normalize_memory_name(name)
if not clean:
return False
if memory_path is None:
from deepagents_code._paths import get_user_agent_md_path
try:
path = get_user_agent_md_path(assistant_id)
except ValueError:
# An invalid or app-owned agent name. Writing the marker anyway is
# what would stamp `AGENTS.md` into app state, so skip it.
logger.warning(
"Skipping onboarding name memory for agent %r",
assistant_id,
exc_info=True,
)
return False
else:
path = memory_path
block = _onboarding_name_memory_block(clean)
try:
path.parent.mkdir(parents=True, exist_ok=True)
try:
existing = path.read_text(encoding="utf-8")
except FileNotFoundError:
existing = ""
except UnicodeDecodeError:
# Existing memory file is not valid UTF-8. Overwriting would clobber
# whatever the user has there, so abort and let them resolve it.
logger.warning(
"Existing memory file %s is not valid UTF-8; skipping onboarding "
"name memory write to avoid clobbering user content",
path,
exc_info=True,
)
return False
path.write_text(
_upsert_onboarding_name_memory(existing, block),
encoding="utf-8",
)
except OSError:
logger.warning(
"Could not write onboarding name memory at %s",
path,
exc_info=True,
)
return False
return True
def _normalize_memory_name(name: str) -> str:
"""Normalize whitespace in a name before writing it to memory.
Returns:
Name with leading/trailing whitespace stripped and internal runs
collapsed to single spaces.
"""
return " ".join(name.split())
def _onboarding_name_memory_block(name: str) -> str:
"""Return the managed memory block for an onboarding name."""
quoted = json.dumps(name)
return (
f"{ONBOARDING_NAME_MEMORY_START}\n"
f"- The user's preferred name is {quoted}.\n"
f"{ONBOARDING_NAME_MEMORY_END}"
)
def _upsert_onboarding_name_memory(existing: str, block: str) -> str:
"""Insert or replace the managed onboarding name memory block.
Returns:
Updated memory file content.
"""
start = existing.find(ONBOARDING_NAME_MEMORY_START)
end = existing.find(ONBOARDING_NAME_MEMORY_END)
if start != -1 and end != -1 and start < end:
end += len(ONBOARDING_NAME_MEMORY_END)
prefix = existing[:start].rstrip()
suffix = existing[end:].strip()
parts = [part for part in (prefix, block, suffix) if part]
return "\n\n".join(parts).rstrip() + "\n"
base = existing.rstrip()
if not base:
return f"## User Preferences\n\n{block}\n"
if "## User Preferences" in base:
return f"{base}\n\n{block}\n"
return f"{base}\n\n## User Preferences\n\n{block}\n"
def extract_onboarding_name_block(text: str) -> str | None:
"""Return the managed onboarding name block (markers included) if present.
Args:
text: Memory file content to inspect.
Returns:
The substring from the start marker through the end marker, or `None`
when a well-formed block is absent.
"""
start = text.find(ONBOARDING_NAME_MEMORY_START)
end = text.find(ONBOARDING_NAME_MEMORY_END)
if start == -1 or end == -1 or start >= end:
return None
return text[start : end + len(ONBOARDING_NAME_MEMORY_END)]
def strip_onboarding_name_markers(text: str) -> str:
"""Remove every onboarding-name marker occurrence from `text`.
A partial edit can leave a lone start or end marker behind. Stripping all
marker strings before re-inserting the managed block keeps re-insertion from
producing orphaned markers that would confuse `extract_onboarding_name_block`.
Args:
text: Memory file content to sanitize.
Returns:
`text` with all start and end marker strings removed.
"""
return text.replace(ONBOARDING_NAME_MEMORY_START, "").replace(
ONBOARDING_NAME_MEMORY_END, ""
)
def should_run_onboarding(state_dir: Path | None = None) -> bool:
"""Return whether onboarding should open at interactive startup.
`DEEPAGENTS_CODE_ONBOARDING` overrides the marker in both directions: a
truthy value forces the flow open on every startup, and a falsy value keeps
it closed even on a fresh install. An unset or unrecognized value leaves the
marker in charge, so the override never has to be unset to get first-run
behavior back.
Args:
state_dir: Optional state directory override for tests.
Returns:
`True` when the env override is truthy, or when it does not apply and no
completion marker exists.
"""
raw = os.environ.get(ONBOARDING)
if raw is not None:
override = classify_env_bool(raw)
if override is None:
logger.warning("Ignoring %s=%r (expected bool)", ONBOARDING, raw)
else:
return override
return not has_completed_onboarding(state_dir)