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
|
"""Generate ~/.pi/agent/settings.json from the repo and its fragments.
Run by link/pi.sh. It lives in a file rather than a heredoc inside $( ):
macOS ships bash 3.2, which does not skip heredoc content when it scans
for the end of a command substitution, so an odd number of apostrophes in
here would stop the whole script parsing -- which it did, on 2026-09-25.
"""
import glob, json, os, sys
base_path, frag_dir, target, state_path, config_dir = sys.argv[1:6]
LIST_KEYS = ("packages", "extensions", "skills", "prompts", "themes")
PATH_KEYS = ("extensions", "skills", "prompts", "themes")
def load(path):
with open(path) as f:
return json.load(f)
def source(entry):
return entry if isinstance(entry, str) else entry.get("source")
def add_packages(into, entries):
for entry in entries:
key = source(entry)
existing = next((i for i, e in enumerate(into) if source(e) == key), None)
if existing is None:
into.append(entry)
elif not isinstance(entry, str):
into[existing] = entry
# What the repos want now.
desired = {}
for path in [base_path] + sorted(glob.glob(os.path.join(frag_dir, "*.json"))):
try:
layer = load(path)
except (OSError, ValueError) as e:
print("error:cannot read %s: %s" % (path, e))
sys.exit(0)
here = os.path.dirname(os.path.realpath(path))
for key, value in layer.items():
if key in PATH_KEYS and path != base_path:
value = [v if os.path.isabs(os.path.expanduser(v)) else os.path.normpath(os.path.join(here, v))
for v in value]
if key == "packages":
add_packages(desired.setdefault("packages", []), value)
elif key in LIST_KEYS:
merged = desired.setdefault(key, [])
merged.extend(v for v in value if v not in merged)
else:
desired[key] = value
# What is there. A symlink is the pre-generation state: its target is a tracked
# file this repo now owns, so treat it as already matching.
def snapshot():
"""What is on disk now, to notice a write that lands before the lock."""
try:
return (os.path.islink(target), open(target).read())
except OSError:
return (os.path.islink(target), None)
on_disk = snapshot()
first_run = False
if os.path.islink(target):
first_run = True
try:
local = load(target)
except (OSError, ValueError):
local = {}
local = {k: v for k, v in local.items() if k not in desired and k not in LIST_KEYS and k != "packages"}
local.update(json.loads(json.dumps(desired)))
last = desired
else:
try:
local = load(target)
except FileNotFoundError:
local = {}
except ValueError as e:
print("error:%s is not valid JSON, leaving it: %s" % (target, e))
sys.exit(0)
# No record of a previous run -- a new machine where pi made its own file
# first, or lost state: nothing is known to have come from the repos, so
# keys already there are kept and keys missing arrive.
try:
last = load(state_path)
except (OSError, ValueError):
last = {}
MISSING = object()
def three_way(ours, before, theirs):
"""ours: on disk; before: generated last time; theirs: generated now."""
if ours == before:
return theirs
return ours
def merge_list(key, identity):
"""Per entry, three-way, in generated order then local-only extras."""
ours = {identity(e): e for e in local.get(key, [])}
before = {identity(e): e for e in last.get(key, [])}
theirs = {identity(e): e for e in desired.get(key, [])}
merged = []
for ident in [identity(e) for e in desired.get(key, [])] + \
[identity(e) for e in local.get(key, []) if identity(e) not in theirs]:
value = three_way(ours.get(ident, MISSING), before.get(ident, MISSING), theirs.get(ident, MISSING))
if value is not MISSING:
merged.append(value)
return merged
result = {}
for key in list(local) + [k for k in desired if k not in local]:
if key in LIST_KEYS:
merged = merge_list(key, source if key == "packages" else (lambda e: e))
if merged or key in local:
result[key] = merged
continue
value = three_way(local.get(key, MISSING), last.get(key, MISSING), desired.get(key, MISSING))
if value is not MISSING:
result[key] = value
changed = first_run or result != local
lock = target + ".lock"
def take_lock():
"""pi's own lock (proper-lockfile: a directory, stale after 10s), so neither
side can read, then write over the other's change."""
import time
for _ in range(2):
try:
os.mkdir(lock)
return True
except FileExistsError:
try:
if time.time() - os.stat(lock).st_mtime > 10:
os.rmdir(lock)
continue
except FileNotFoundError:
continue
return False
return False
def save_state():
os.makedirs(os.path.dirname(state_path), exist_ok=True)
with open(state_path + ".tmp", "w") as f:
json.dump(desired, f, indent=2)
os.replace(state_path + ".tmp", state_path)
if not changed:
print("unchanged:")
if not os.environ.get("DRY_RUN"):
save_state()
elif os.environ.get("DRY_RUN"):
print("would-write:" + ("replacing the link" if first_run else "merged changes"))
elif not take_lock():
print("locked:")
elif snapshot() != on_disk:
# pi wrote between the read above and taking the lock: merge again next time
# rather than write over it.
os.rmdir(lock)
print("locked:")
else:
try:
tmp = target + ".dotfiles-tmp"
with open(tmp, "w") as f:
json.dump(result, f, indent=2)
f.write("\n")
os.replace(tmp, target)
save_state()
finally:
os.rmdir(lock)
print("wrote:" + ("replaced the link" if first_run else "merged changes"))
|
