Coverage for sources/agentsmgr/population.py: 20%
144 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-10-28 17:44 +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''' Command for populating agent content from data sources. '''
24from . import __
25from . import cmdbase as _cmdbase
26from . import core as _core
27from . import exceptions as _exceptions
28from . import generator as _generator
29from . import instructions as _instructions
30from . import memorylinks as _memorylinks
31from . import operations as _operations
32from . import renderers as _renderers
33from . import results as _results
34from . import userdata as _userdata
37_scribe = __.provide_scribe( __name__ )
40SourceArgument: __.typx.TypeAlias = __.typx.Annotated[
41 __.tyro.conf.Positional[ str ],
42 __.tyro.conf.arg( help = "Data source (local path or git URL)" ),
43]
44TargetArgument: __.typx.TypeAlias = __.typx.Annotated[
45 __.tyro.conf.Positional[ __.Path ],
46 __.tyro.conf.arg( help = "Target directory for content generation" ),
47]
50def _filter_coders_by_mode(
51 coders: __.cabc.Sequence[ str ],
52 target_mode: _renderers.ExplicitTargetMode,
53 renderers: __.cabc.Mapping[ str, __.typx.Any ],
54) -> tuple[ str, ... ]:
55 ''' Filters coders by their default targeting mode.
57 Returns coders whose mode_default matches the target mode.
58 This ensures populate project only handles per-project coders
59 and populate user only handles per-user coders, respecting each
60 renderer's designed usage pattern.
61 '''
62 filtered: list[ str ] = [ ]
63 for coder_name in coders:
64 try: renderer = renderers[ coder_name ]
65 except KeyError as exception:
66 raise _exceptions.CoderAbsence( coder_name ) from exception
67 if renderer.mode_default == target_mode:
68 filtered.append( coder_name )
69 else:
70 _scribe.debug(
71 f"Skipping {coder_name} for {target_mode} mode: "
72 f"default mode is {renderer.mode_default}" )
73 return tuple( filtered )
76def _create_all_symlinks(
77 configuration: __.cabc.Mapping[ str, __.typx.Any ],
78 target: __.Path,
79 mode: str,
80 simulate: bool,
81) -> tuple[ str, ... ]:
82 ''' Creates all symlinks and returns their names for git exclude.
84 Creates memory symlinks for all coders and coder directory
85 symlinks for per-project mode. Returns tuple of all symlink
86 names (both newly created and pre-existing) for git exclude
87 update.
88 '''
89 all_symlink_names: list[ str ] = [ ]
90 if mode == 'nowhere': return tuple( all_symlink_names )
91 links_attempted, links_created, symlink_names = (
92 _memorylinks.create_memory_symlinks_for_coders(
93 coders = configuration[ 'coders' ],
94 target = target,
95 renderers = _renderers.RENDERERS,
96 simulate = simulate,
97 ) )
98 all_symlink_names.extend( symlink_names )
99 if links_created > 0:
100 _scribe.info(
101 f"Created {links_created}/{links_attempted} memory symlinks" )
102 needs_coder_symlinks = (
103 mode == 'per-project'
104 or ( mode == 'default' and any(
105 _renderers.RENDERERS[ coder ].mode_default == 'per-project'
106 for coder in configuration[ 'coders' ] ) ) )
107 if needs_coder_symlinks:
108 ( coder_symlinks_attempted,
109 coder_symlinks_created,
110 coder_symlink_names ) = (
111 _create_coder_directory_symlinks(
112 coders = configuration[ 'coders' ],
113 target = target,
114 renderers = _renderers.RENDERERS,
115 simulate = simulate,
116 ) )
117 all_symlink_names.extend( coder_symlink_names )
118 if coder_symlinks_created > 0:
119 _scribe.info(
120 f"Created {coder_symlinks_created}/"
121 f"{coder_symlinks_attempted} coder directory symlinks" )
122 return tuple( all_symlink_names )
125def _populate_instructions_if_configured(
126 configuration: __.cabc.Mapping[ str, __.typx.Any ],
127 target: __.Path,
128 tag_prefix: __.Absential[ str ],
129 simulate: bool,
130) -> tuple[ bool, str ]:
131 ''' Populates instructions if configured and returns status.
133 Returns tuple of (sources_present, instructions_target_path).
134 sources_present indicates whether instruction sources were
135 configured and processed.
136 '''
137 if not configuration.get( 'provide_instructions', False ):
138 return ( False, '' )
139 instructions_sources = configuration.get( 'instructions_sources', [ ] )
140 instructions_target = configuration.get(
141 'instructions_target', '.auxiliary/instructions' )
142 if not instructions_sources:
143 return ( False, instructions_target )
144 instructions_attempted, instructions_updated = (
145 _instructions.populate_instructions(
146 instructions_sources,
147 target / instructions_target,
148 tag_prefix,
149 simulate,
150 ) )
151 _scribe.info(
152 f"Updated {instructions_updated}/"
153 f"{instructions_attempted} instruction files" )
154 return ( True, instructions_target )
157def _populate_per_user_content(
158 location: __.Path,
159 coders: __.cabc.Sequence[ str ],
160 configuration: __.cabc.Mapping[ str, __.typx.Any ],
161 application_configuration: __.cabc.Mapping[ str, __.typx.Any ],
162 simulate: bool,
163) -> tuple[ int, int ]:
164 ''' Populates commands and agents for per-user coders.
166 Generates content to each coder's per-user directory using
167 renderer's resolve_base_directory() with per-user mode.
168 Returns tuple of (items_attempted, items_generated).
169 '''
170 items_attempted = 0
171 items_generated = 0
172 for coder_name in coders:
173 try: renderer = _renderers.RENDERERS[ coder_name ]
174 except KeyError as exception:
175 raise _exceptions.CoderAbsence( coder_name ) from exception
176 coder_configuration = { 'coders': [ coder_name ] }
177 generator = _generator.ContentGenerator(
178 location = location,
179 configuration = coder_configuration,
180 application_configuration = application_configuration,
181 mode = 'per-user',
182 )
183 target = renderer.resolve_base_directory(
184 'per-user', __.Path.cwd( ), configuration, dict( __.os.environ ) )
185 attempted, generated = _operations.populate_directory(
186 generator, target, simulate )
187 items_attempted += attempted
188 items_generated += generated
189 return ( items_attempted, items_generated )
192def _create_coder_directory_symlinks(
193 coders: __.cabc.Sequence[ str ],
194 target: __.Path,
195 renderers: __.cabc.Mapping[ str, __.typx.Any ],
196 simulate: bool = False,
197) -> tuple[ int, int, tuple[ str, ... ] ]:
198 ''' Creates symlinks from .{coder} to .auxiliary/configuration/coders/.
200 For per-project mode, creates symlinks that make coder directories
201 accessible at their expected locations (.claude, .opencode, etc.)
202 while keeping actual files organized under
203 .auxiliary/configuration/coders/.
205 Each renderer is responsible for specifying its symlink requirements
206 via provide_project_symlinks(). Population logic simply iterates
207 coders and asks renderers for their symlinks.
209 Only creates symlinks for coders whose default mode is per-project.
210 Coders with per-user default mode are skipped since they do not
211 use per-project directories.
213 Returns tuple of (attempted, created, symlink_names) where
214 symlink_names contains names of all symlinks (both newly created
215 and pre-existing).
216 '''
217 attempted = 0
218 created = 0
219 symlink_names: list[ str ] = [ ]
221 for coder_name in coders:
222 try: renderer = renderers[ coder_name ]
223 except KeyError as exception:
224 raise _exceptions.CoderAbsence( coder_name ) from exception
225 if renderer.mode_default != 'per-project':
226 _scribe.debug(
227 f"Skipping directory symlink for {coder_name}: "
228 f"default mode is {renderer.mode_default}" )
229 continue
230 for source, link_path in renderer.provide_project_symlinks( target ):
231 attempted += 1
232 was_created, symlink_name = (
233 _memorylinks.create_memory_symlink(
234 source, link_path, simulate ) )
235 if was_created: created += 1
236 symlink_names.append( symlink_name )
237 return ( attempted, created, tuple( symlink_names ) )
240class PopulateProjectCommand( __.appcore_cli.Command ):
241 ''' Generates project-scoped agent content from data sources. '''
243 source: SourceArgument = '.'
244 target: TargetArgument = __.dcls.field( default_factory = __.Path.cwd )
245 profile: __.typx.Annotated[
246 __.typx.Optional[ __.Path ],
247 __.tyro.conf.arg(
248 help = (
249 "Alternative Copier answers file (defaults to "
250 "auto-detected)" ),
251 prefix_name = False ),
252 ] = None
253 simulate: __.typx.Annotated[
254 bool,
255 __.tyro.conf.arg(
256 help = "Dry run mode - show generated content",
257 prefix_name = False ),
258 ] = False
259 tag_prefix: __.typx.Annotated[
260 __.typx.Optional[ str ],
261 __.tyro.conf.arg(
262 help = (
263 "Prefix for version tags (e.g., 'v', 'stable-', 'prod-'); "
264 "only tags with this prefix are considered and the prefix "
265 "is stripped before version parsing" ),
266 prefix_name = False ),
267 ] = None
269 @_cmdbase.intercept_errors( )
270 async def execute( self, auxdata: __.appcore.state.Globals ) -> None: # pyright: ignore[reportIncompatibleMethodOverride]
271 ''' Generates project content from data sources. '''
272 if not isinstance( auxdata, _core.Globals ): # pragma: no cover
273 raise _exceptions.ContextInvalidity
274 _scribe.info(
275 f"Populating project content from {self.source} to {self.target}" )
276 configuration = await _cmdbase.retrieve_configuration(
277 self.target, self.profile )
278 per_project_coders = _filter_coders_by_mode(
279 configuration[ 'coders' ], 'per-project', _renderers.RENDERERS )
280 if not per_project_coders:
281 _scribe.warning(
282 "No per-project default coders found in configuration" )
283 return
284 filtered_configuration = dict( configuration )
285 filtered_configuration[ 'coders' ] = per_project_coders
286 prefix = __.absent if self.tag_prefix is None else self.tag_prefix
287 location = _cmdbase.retrieve_data_location( self.source, prefix )
288 _cmdbase.validate_data_source_structure(
289 location,
290 ( 'configurations', 'contents', 'templates' ) )
291 generator = _generator.ContentGenerator(
292 location = location,
293 configuration = filtered_configuration,
294 application_configuration = auxdata.configuration,
295 mode = 'per-project',
296 )
297 items_attempted, items_generated = _operations.populate_directory(
298 generator, self.target, self.simulate )
299 _scribe.info( f"Generated {items_generated}/{items_attempted} items" )
300 instructions_populated, instructions_target = (
301 _populate_instructions_if_configured(
302 filtered_configuration, self.target, prefix, self.simulate ) )
303 all_symlink_names = _create_all_symlinks(
304 filtered_configuration, self.target, 'per-project', self.simulate )
305 git_exclude_entries: list[ str ] = [ ]
306 if instructions_populated:
307 git_exclude_entries.append( instructions_target )
308 git_exclude_entries.extend( all_symlink_names )
309 if git_exclude_entries:
310 entries_count = _operations.update_git_exclude(
311 self.target, git_exclude_entries, self.simulate )
312 if entries_count > 0:
313 _scribe.info(
314 f"Managing {entries_count} entries in .git/info/exclude" )
315 result = _results.ContentGenerationResult(
316 source_location = location,
317 target_location = self.target,
318 coders = tuple( configuration[ 'coders' ] ),
319 simulated = self.simulate,
320 items_generated = items_generated,
321 )
322 await _core.render_and_print_result(
323 result, auxdata.display, auxdata.exits )
326class PopulateUserCommand( __.appcore_cli.Command ):
327 ''' Populates per-user global settings and executables. '''
329 source: SourceArgument = '.'
330 profile: __.typx.Annotated[
331 __.typx.Optional[ __.Path ],
332 __.tyro.conf.arg(
333 help = (
334 "Alternative Copier answers file (defaults to "
335 "auto-detected)" ),
336 prefix_name = False ),
337 ] = None
338 simulate: __.typx.Annotated[
339 bool,
340 __.tyro.conf.arg(
341 help = "Dry run mode - show what would be installed",
342 prefix_name = False ),
343 ] = False
344 tag_prefix: __.typx.Annotated[
345 __.typx.Optional[ str ],
346 __.tyro.conf.arg(
347 help = (
348 "Prefix for version tags (e.g., 'v', 'stable-', 'prod-'); "
349 "only tags with this prefix are considered and the prefix "
350 "is stripped before version parsing" ),
351 prefix_name = False ),
352 ] = None
354 @_cmdbase.intercept_errors( )
355 async def execute( self, auxdata: __.appcore.state.Globals ) -> None: # pyright: ignore[reportIncompatibleMethodOverride]
356 ''' Populates user-scoped settings and executables. '''
357 if not isinstance( auxdata, _core.Globals ): # pragma: no cover
358 raise _exceptions.ContextInvalidity
359 _scribe.info( f"Populating user configuration from {self.source}" )
360 configuration = await _cmdbase.retrieve_configuration(
361 __.Path.cwd( ), self.profile )
362 per_user_coders = _filter_coders_by_mode(
363 configuration[ 'coders' ], 'per-user', _renderers.RENDERERS )
364 if not per_user_coders:
365 _scribe.warning(
366 "No per-user default coders found in configuration" )
367 return
368 prefix = __.absent if self.tag_prefix is None else self.tag_prefix
369 location = _cmdbase.retrieve_data_location( self.source, prefix )
370 _cmdbase.validate_data_source_structure(
371 location,
372 ( 'configurations', 'contents', 'templates',
373 'user/configurations', 'user/executables' ) )
374 content_attempted, content_generated = _populate_per_user_content(
375 location,
376 per_user_coders,
377 configuration,
378 auxdata.configuration,
379 self.simulate,
380 )
381 if content_attempted > 0:
382 _scribe.info(
383 f"Generated {content_generated}/{content_attempted} items" )
384 globals_attempted, globals_updated = _userdata.populate_globals(
385 location,
386 per_user_coders,
387 auxdata.configuration,
388 self.simulate,
389 )
390 _scribe.info(
391 f"Updated {globals_updated}/{globals_attempted} global files" )
392 wrappers_attempted, wrappers_installed = (
393 _userdata.populate_user_wrappers( location, self.simulate ) )
394 if wrappers_attempted > 0:
395 _scribe.info(
396 f"Installed {wrappers_installed}/{wrappers_attempted} "
397 "wrapper scripts" )
398 total_items = content_generated + globals_updated + wrappers_installed
399 result = _results.ContentGenerationResult(
400 source_location = location,
401 target_location = __.Path.home( ),
402 coders = per_user_coders,
403 simulated = self.simulate,
404 items_generated = total_items,
405 )
406 await _core.render_and_print_result(
407 result, auxdata.display, auxdata.exits )
410class PopulateCommand( __.appcore_cli.Command ):
411 ''' Populates agent content and configuration. '''
413 command: __.typx.Union[
414 __.typx.Annotated[
415 PopulateProjectCommand,
416 __.tyro.conf.subcommand( 'project', prefix_name = False ),
417 ],
418 __.typx.Annotated[
419 PopulateUserCommand,
420 __.tyro.conf.subcommand( 'user', prefix_name = False ),
421 ],
422 ] = __.dcls.field( default_factory = PopulateProjectCommand )
424 async def execute( self, auxdata: __.appcore.state.Globals ) -> None: # pyright: ignore[reportIncompatibleMethodOverride]
425 await self.command( auxdata )