Coverage for sources/agentsmgr/userdata.py: 33%
143 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-08 06:57 +0000
1# vim: set filetype=python fileencoding=utf-8:
2# -*- coding: utf-8 -*-
4#============================================================================#
5# #
6# Licensed under the Apache License, Version 2.0 (the "License"); #
7# you may not use this file except in compliance with the License. #
8# You may obtain a copy of the License at #
9# #
10# http://www.apache.org/licenses/LICENSE-2.0 #
11# #
12# Unless required by applicable law or agreed to in writing, software #
13# distributed under the License is distributed on an "AS IS" BASIS, #
14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. #
15# See the License for the specific language governing permissions and #
16# limitations under the License. #
17# #
18#============================================================================#
21''' Global settings management for coder configurations.
23 Provides functionality for populating per-user global settings files,
24 including direct file copying and JSON/TOML settings merging with user
25 preservation semantics.
26'''
29import json as _json
31import toml as _toml
33from . import __
34from . import exceptions as _exceptions
35from . import resolver as _resolver
38_scribe = __.provide_scribe( __name__ )
41def _is_json_dict(
42 value: __.typx.Any
43) -> __.typx.TypeGuard[ dict[ str, __.typx.Any ] ]:
44 ''' Type guard for JSON dictionary values. '''
45 return isinstance( value, dict )
48def populate_globals(
49 data_location: __.Path,
50 coders: __.cabc.Sequence[ str ],
51 application_configuration: __.cabc.Mapping[ str, __.typx.Any ],
52 simulate: bool = False,
53) -> tuple[ int, int ]:
54 ''' Populates per-user global files for configured coders.
56 Surveys distribution/per-user/coders directory for coder-specific
57 files and populates them to per-user locations. Handles two types
58 of files: direct copy for non-settings files and merge for settings
59 files (preserving user values).
61 Returns tuple of (files_attempted, files_updated) counts.
62 '''
63 globals_directory = data_location / 'per-user' / 'coders'
64 if not globals_directory.exists( ):
65 return ( 0, 0 )
66 files_attempted = 0
67 files_updated = 0
68 for coder, renderer in _resolver.resolve_coders( coders ):
69 coder_globals = globals_directory / coder
70 if not coder_globals.exists( ):
71 continue
72 per_user_directory = renderer.resolve_base_directory(
73 mode = 'per-user',
74 target = __.Path.cwd( ),
75 configuration = application_configuration,
76 environment = __.os.environ,
77 )
78 for global_file in coder_globals.iterdir( ):
79 if not global_file.is_file( ):
80 continue
81 files_attempted += 1
82 target_file = per_user_directory / global_file.name
83 if _is_settings_file( global_file, coder ):
84 updated = _merge_settings_file(
85 global_file, target_file, simulate )
86 else:
87 updated = _copy_file_directly(
88 global_file, target_file, simulate )
89 if updated:
90 files_updated += 1
91 return ( files_attempted, files_updated )
94def _is_settings_file( file: __.Path, coder: str ) -> bool:
95 ''' Determines whether file is a settings file requiring merge logic.
97 Settings files have coder-specific names and contain JSON or TOML
98 configuration that should be merged rather than replaced. Non-settings
99 files are directly copied, replacing any existing version.
100 '''
101 settings_names: dict[ str, tuple[ str, ... ] ] = {
102 'claude': ( 'settings.json', ),
103 'opencode': ( 'opencode.json', 'opencode.jsonc' ),
104 'codex': ( 'config.toml', ),
105 }
106 return file.name in settings_names.get( coder, ( ) )
109def _copy_file_directly(
110 source: __.Path, target: __.Path, simulate: bool
111) -> bool:
112 ''' Copies file directly from source to target location.
114 Creates target directory if needed. Returns True if file was
115 updated (or would be updated in simulation mode).
116 '''
117 if simulate:
118 return True
119 target.parent.mkdir( parents = True, exist_ok = True )
120 try: __.shutil.copy2( source, target )
121 except ( OSError, IOError ) as exception:
122 raise _exceptions.GlobalsPopulationFailure(
123 source, target
124 ) from exception
125 return True
128def _merge_settings_file(
129 source: __.Path, target: __.Path, simulate: bool
130) -> bool:
131 ''' Merges JSON or TOML settings file preserving user values.
133 Loads both source template and target user settings, performs deep
134 merge adding missing keys from template while preserving all user
135 values. Creates backup before writing merged result. Returns True
136 if file was updated (or would be updated in simulation mode).
137 '''
138 if source.suffix == '.toml': 138 ↛ 160line 138 didn't jump to line 160 because the condition on line 138 was always true
139 template = _load_toml_file( source, target )
140 user_settings: dict[ str, __.typx.Any ] = (
141 _load_toml_file( target, target ) if target.exists( )
142 else { } )
143 merged = _deep_merge_settings( user_settings, template )
144 if simulate: 144 ↛ 145line 144 didn't jump to line 145 because the condition on line 144 was never true
145 return True
146 if merged == user_settings: 146 ↛ 147line 146 didn't jump to line 147 because the condition on line 146 was never true
147 return False
148 if target.exists( ) and _toml_content_contains_comments( 148 ↛ 151line 148 didn't jump to line 151 because the condition on line 148 was never true
149 target.read_text( encoding = 'utf-8' )
150 ):
151 backup_path = target.with_suffix( '.toml.backup' )
152 _scribe.warning(
153 "TOML settings merge rewrites '%s' and may drop comments. "
154 "A backup will be written to '%s'.",
155 target,
156 backup_path,
157 )
158 _write_merged_toml_settings( target, merged )
159 else:
160 template = _load_json_file( source, target )
161 user_settings = (
162 _load_json_file( target, target ) if target.exists( )
163 else { } )
164 merged = _deep_merge_settings( user_settings, template )
165 if simulate:
166 return True
167 if merged == user_settings:
168 return False
169 _write_merged_settings( target, merged )
170 return True
173def _load_json_file(
174 filepath: __.Path, target_context: __.Path
175) -> dict[ str, __.typx.Any ]:
176 ''' Loads JSON file with error handling.
178 Raises GlobalsPopulationFailure with source context on any error.
179 '''
180 try: content = filepath.read_text( encoding = 'utf-8' )
181 except ( OSError, IOError ) as exception:
182 raise _exceptions.GlobalsPopulationFailure(
183 filepath, target_context ) from exception
184 try:
185 loaded: __.typx.Any = _json.loads( content )
186 except ValueError as exception:
187 raise _exceptions.GlobalsPopulationFailure(
188 filepath, target_context ) from exception
189 if not _is_json_dict( loaded ):
190 raise _exceptions.GlobalsPopulationFailure( filepath, target_context )
191 return loaded
194def _write_merged_settings(
195 target: __.Path, merged: dict[ str, __.typx.Any ]
196) -> None:
197 ''' Writes merged settings with backup of existing file.
199 Creates target directory if needed. Backs up existing file before
200 writing merged result.
201 '''
202 target.parent.mkdir( parents = True, exist_ok = True )
203 if target.exists( ):
204 backup_path = target.with_suffix( '.json.backup' )
205 try: __.shutil.copy2( target, backup_path )
206 except ( OSError, IOError ) as exception:
207 raise _exceptions.GlobalsPopulationFailure(
208 target, target ) from exception
209 try:
210 target.write_text(
211 _json.dumps( merged, indent = 2 ), encoding = 'utf-8' )
212 except ( OSError, IOError ) as exception:
213 raise _exceptions.GlobalsPopulationFailure(
214 target, target
215 ) from exception
218def _load_toml_file(
219 filepath: __.Path, target_context: __.Path
220) -> dict[ str, __.typx.Any ]:
221 ''' Loads TOML file with error handling.
223 Raises GlobalsPopulationFailure with source context on any error.
224 '''
225 try: content = filepath.read_text( encoding = 'utf-8' )
226 except ( OSError, IOError ) as exception:
227 raise _exceptions.GlobalsPopulationFailure(
228 filepath, target_context ) from exception
229 try:
230 loaded: dict[ str, __.typx.Any ] = __.tomli.loads(
231 content )
232 except __.tomli.TOMLDecodeError as exception:
233 raise _exceptions.GlobalsPopulationFailure(
234 filepath, target_context ) from exception
235 return loaded
238def _write_merged_toml_settings(
239 target: __.Path, merged: dict[ str, __.typx.Any ]
240) -> None:
241 ''' Writes merged TOML settings with backup of existing file.
243 Creates target directory if needed. Backs up existing file before
244 writing merged result.
245 '''
246 target.parent.mkdir( parents = True, exist_ok = True )
247 if target.exists( ): 247 ↛ 253line 247 didn't jump to line 253 because the condition on line 247 was always true
248 backup_path = target.with_suffix( '.toml.backup' )
249 try: __.shutil.copy2( target, backup_path )
250 except ( OSError, IOError ) as exception:
251 raise _exceptions.GlobalsPopulationFailure(
252 target, target ) from exception
253 try:
254 target.write_text( _toml.dumps( merged ), encoding = 'utf-8' )
255 except ( OSError, IOError ) as exception:
256 raise _exceptions.GlobalsPopulationFailure(
257 target, target
258 ) from exception
261def _toml_content_contains_comments( content: str ) -> bool:
262 ''' Heuristic detection for comments in a TOML file.
264 TOML comments begin with '#'. We warn when overwriting TOML settings
265 because our merge process rewrites the file and does not preserve
266 comments or formatting.
267 '''
268 for line in content.splitlines( ):
269 stripped = line.lstrip( )
270 if stripped.startswith( '#' ): 270 ↛ 271line 270 didn't jump to line 271 because the condition on line 270 was never true
271 return True
272 if ' #' in line: 272 ↛ 273line 272 didn't jump to line 273 because the condition on line 272 was never true
273 return True
274 return False
277def populate_user_wrappers(
278 data_location: __.Path,
279 simulate: bool = False,
280) -> tuple[ int, int ]:
281 ''' Installs wrapper scripts to user bin directory.
283 Copies wrapper scripts from data source to ~/.local/bin,
284 making them executable. Returns tuple of (files_attempted,
285 files_installed) counts.
286 '''
287 wrappers_dir = data_location / 'per-user' / 'general'
288 user_bin = __.Path.home( ) / '.local' / 'bin'
289 if not wrappers_dir.exists( ):
290 return ( 0, 0 )
291 files_attempted = 0
292 files_installed = 0
293 for script in wrappers_dir.iterdir( ):
294 if not script.is_file( ):
295 continue
296 files_attempted += 1
297 target = user_bin / script.name
298 if not simulate:
299 user_bin.mkdir( parents = True, exist_ok = True )
300 try: __.shutil.copy2( script, target )
301 except ( OSError, IOError ) as exception:
302 raise _exceptions.GlobalsPopulationFailure(
303 script, target
304 ) from exception
305 try: target.chmod( target.stat( ).st_mode | 0o111 )
306 except ( OSError, IOError ) as exception:
307 raise _exceptions.GlobalsPopulationFailure(
308 script, target
309 ) from exception
310 files_installed += 1
311 return ( files_attempted, files_installed )
314def _deep_merge_settings(
315 target: dict[ str, __.typx.Any ], source: dict[ str, __.typx.Any ]
316) -> dict[ str, __.typx.Any ]:
317 ''' Recursively merges source into target preserving target values.
319 Implements additive merge: adds keys from source that are missing
320 in target. When both contain same key with dict values, recursively
321 merges nested dicts. For conflicting scalar values, target value
322 wins (user preferences preserved).
323 '''
324 result = target.copy( )
325 for key, source_value in source.items( ):
326 if key not in result:
327 result[ key ] = source_value
328 elif (
329 _is_json_dict( result[ key ] )
330 and _is_json_dict( source_value )
331 ):
332 result[ key ] = _deep_merge_settings(
333 result[ key ], source_value )
334 return result