-
Notifications
You must be signed in to change notification settings - Fork 252
Expand file tree
/
Copy pathonboard_write.py
More file actions
55 lines (46 loc) · 1.92 KB
/
onboard_write.py
File metadata and controls
55 lines (46 loc) · 1.92 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
"""Atomic write of PREFERENCES.md with backup and existing-content detection."""
import os, shutil, tempfile
REL = ".agent/memory/personal/PREFERENCES.md"
_STUB_MARKERS = ("_(e.g.,", "<!-- generated by")
def is_customized(target_dir: str) -> bool:
"""Return True if PREFERENCES.md has real user content (not just stubs)."""
path = os.path.join(target_dir, REL)
try:
text = open(path).read()
except FileNotFoundError:
return False
# Already generated by this wizard → treat as customized
if "<!-- generated by agentic-stack onboard" in text:
return True
# Original stub: every data line starts with _(e.g.,
data_lines = [
ln.strip() for ln in text.splitlines()
if ln.strip() and not ln.strip().startswith(("#", ">", "##", "<!--"))
]
stub_lines = [ln for ln in data_lines if any(m in ln for m in _STUB_MARKERS)]
return bool(data_lines) and len(stub_lines) < len(data_lines)
def write_prefs(target_dir: str, content: str, force: bool = False) -> str:
"""Write content to PREFERENCES.md; return the absolute path written."""
dest = os.path.abspath(os.path.join(target_dir, REL))
dest_dir = os.path.dirname(dest)
if not os.path.isdir(dest_dir):
raise RuntimeError(
f".agent/memory/personal/ not found in {target_dir}\n"
"Run the adapter install first, then re-run the wizard."
)
# Back up an existing customized file (non-destructive by default)
if is_customized(target_dir) and not force:
shutil.copy2(dest, dest + ".bak")
# Atomic write via a temp file in the same directory
fd, tmp = tempfile.mkstemp(dir=dest_dir, prefix=".prefs_tmp_")
try:
with os.fdopen(fd, "w", encoding="utf-8") as f:
f.write(content)
os.replace(tmp, dest)
except Exception:
try:
os.unlink(tmp)
except OSError:
pass
raise
return dest