Coverage for sources / agentsmgr / population.py: 20%
156 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-11-30 00:03 +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 ) )
240def _copy_coder_resources(
241 location: __.Path,
242 target: __.Path,
243 coders: __.cabc.Sequence[ str ],
244 simulate: bool
245) -> tuple[ int, int ]:
246 ''' Copies static resources for coders.
248 Copies resources from defaults/per-project/resources/<coder>
249 to target directory. Returns tuple of (coders_attempted,
250 coders_processed).
251 '''
252 resources_source = location / 'per-project' / 'resources'
253 if not resources_source.exists( ):
254 _scribe.debug(
255 f"No per-project resources found at {resources_source}" )
256 return ( 0, 0 )
257 coders_target = target / '.auxiliary' / 'configuration' / 'coders'
258 return _operations.copy_coder_resources(
259 resources_source, coders_target, coders, simulate )
262def _manage_project_auxiliaries(
263 configuration: __.cabc.Mapping[ str, __.typx.Any ],
264 target: __.Path,
265 tag_prefix: __.Absential[ str ],
266 simulate: bool
267) -> None:
268 ''' Manages auxiliary project files (instructions, symlinks, excludes). '''
269 instructions_populated, instructions_target = (
270 _populate_instructions_if_configured(
271 configuration, target, tag_prefix, simulate ) )
272 all_symlink_names = _create_all_symlinks(
273 configuration, target, 'per-project', simulate )
274 git_exclude_entries: list[ str ] = [ ]
275 if instructions_populated:
276 git_exclude_entries.append( instructions_target )
277 git_exclude_entries.extend( all_symlink_names )
278 if git_exclude_entries:
279 entries_count = _operations.update_git_exclude(
280 target, git_exclude_entries, simulate )
281 if entries_count > 0:
282 _scribe.info(
283 f"Managing {entries_count} entries in .git/info/exclude" )
286class PopulateProjectCommand( __.appcore_cli.Command ):
287 ''' Generates project-scoped agent content from data sources.
289 Populates agent commands, definitions, and static resources
290 from the specified data source. Copies static resources from
291 defaults/per-project/resources/ to the project configuration.
292 '''
294 source: SourceArgument = '.'
295 target: TargetArgument = __.dcls.field( default_factory = __.Path.cwd )
296 profile: __.typx.Annotated[
297 __.typx.Optional[ __.Path ],
298 __.tyro.conf.arg(
299 help = (
300 "Alternative Copier answers file (defaults to "
301 "auto-detected)" ),
302 prefix_name = False ),
303 ] = None
304 simulate: __.typx.Annotated[
305 bool,
306 __.tyro.conf.arg(
307 help = "Dry run mode - show generated content",
308 prefix_name = False ),
309 ] = False
310 tag_prefix: __.typx.Annotated[
311 __.typx.Optional[ str ],
312 __.tyro.conf.arg(
313 help = (
314 "Prefix for version tags (e.g., 'v', 'stable-', 'prod-'); "
315 "only tags with this prefix are considered and the prefix "
316 "is stripped before version parsing" ),
317 prefix_name = False ),
318 ] = None
320 @_cmdbase.intercept_errors( )
321 async def execute( self, auxdata: __.appcore.state.Globals ) -> None: # pyright: ignore[reportIncompatibleMethodOverride]
322 ''' Generates project content from data sources. '''
323 if not isinstance( auxdata, _core.Globals ): # pragma: no cover
324 raise _exceptions.ContextInvalidity
325 _scribe.info(
326 f"Populating project content from {self.source} to {self.target}" )
327 configuration = await _cmdbase.retrieve_configuration(
328 self.target, self.profile )
329 per_project_coders = _filter_coders_by_mode(
330 configuration[ 'coders' ], 'per-project', _renderers.RENDERERS )
331 if not per_project_coders:
332 _scribe.warning(
333 "No per-project default coders found in configuration" )
334 return
335 filtered_configuration = dict( configuration )
336 filtered_configuration[ 'coders' ] = per_project_coders
337 prefix = __.absent if self.tag_prefix is None else self.tag_prefix
338 location = _cmdbase.retrieve_data_location( self.source, prefix )
339 _cmdbase.validate_data_source_structure(
340 location,
341 ( 'configurations', 'contents', 'templates' ) )
342 generator = _generator.ContentGenerator(
343 location = location,
344 configuration = filtered_configuration,
345 application_configuration = auxdata.configuration,
346 mode = 'per-project',
347 )
348 items_attempted, items_generated = _operations.populate_directory(
349 generator, self.target, self.simulate )
350 _scribe.info( f"Generated {items_generated}/{items_attempted} items" )
351 resources_attempted, resources_copied = _copy_coder_resources(
352 location,
353 self.target,
354 filtered_configuration[ 'coders' ],
355 self.simulate )
356 if resources_copied > 0:
357 _scribe.info(
358 f"Copied resources for "
359 f"{resources_copied}/{resources_attempted} coders" )
360 _manage_project_auxiliaries(
361 filtered_configuration, self.target, prefix, self.simulate )
362 result = _results.ContentGenerationResult(
363 source_location = location,
364 target_location = self.target,
365 coders = tuple( configuration[ 'coders' ] ),
366 simulated = self.simulate,
367 items_generated = items_generated,
368 )
369 await _core.render_and_print_result(
370 result, auxdata.display, auxdata.exits )
373class PopulateUserCommand( __.appcore_cli.Command ):
374 ''' Populates per-user global settings and executables. '''
376 source: SourceArgument = '.'
377 profile: __.typx.Annotated[
378 __.typx.Optional[ __.Path ],
379 __.tyro.conf.arg(
380 help = (
381 "Alternative Copier answers file (defaults to "
382 "auto-detected)" ),
383 prefix_name = False ),
384 ] = None
385 simulate: __.typx.Annotated[
386 bool,
387 __.tyro.conf.arg(
388 help = "Dry run mode - show what would be installed",
389 prefix_name = False ),
390 ] = False
391 tag_prefix: __.typx.Annotated[
392 __.typx.Optional[ str ],
393 __.tyro.conf.arg(
394 help = (
395 "Prefix for version tags (e.g., 'v', 'stable-', 'prod-'); "
396 "only tags with this prefix are considered and the prefix "
397 "is stripped before version parsing" ),
398 prefix_name = False ),
399 ] = None
401 @_cmdbase.intercept_errors( )
402 async def execute( self, auxdata: __.appcore.state.Globals ) -> None: # pyright: ignore[reportIncompatibleMethodOverride]
403 ''' Populates user-scoped settings and executables. '''
404 if not isinstance( auxdata, _core.Globals ): # pragma: no cover
405 raise _exceptions.ContextInvalidity
406 _scribe.info( f"Populating user configuration from {self.source}" )
407 configuration = await _cmdbase.retrieve_configuration(
408 __.Path.cwd( ), self.profile )
409 per_user_coders = _filter_coders_by_mode(
410 configuration[ 'coders' ], 'per-user', _renderers.RENDERERS )
411 if not per_user_coders:
412 _scribe.warning(
413 "No per-user default coders found in configuration" )
414 return
415 prefix = __.absent if self.tag_prefix is None else self.tag_prefix
416 location = _cmdbase.retrieve_data_location( self.source, prefix )
417 _cmdbase.validate_data_source_structure(
418 location,
419 ( 'configurations', 'contents', 'templates',
420 'user/configurations', 'user/executables' ) )
421 content_attempted, content_generated = _populate_per_user_content(
422 location,
423 per_user_coders,
424 configuration,
425 auxdata.configuration,
426 self.simulate,
427 )
428 if content_attempted > 0:
429 _scribe.info(
430 f"Generated {content_generated}/{content_attempted} items" )
431 globals_attempted, globals_updated = _userdata.populate_globals(
432 location,
433 per_user_coders,
434 auxdata.configuration,
435 self.simulate,
436 )
437 _scribe.info(
438 f"Updated {globals_updated}/{globals_attempted} global files" )
439 wrappers_attempted, wrappers_installed = (
440 _userdata.populate_user_wrappers( location, self.simulate ) )
441 if wrappers_attempted > 0:
442 _scribe.info(
443 f"Installed {wrappers_installed}/{wrappers_attempted} "
444 "wrapper scripts" )
445 total_items = content_generated + globals_updated + wrappers_installed
446 result = _results.ContentGenerationResult(
447 source_location = location,
448 target_location = __.Path.home( ),
449 coders = per_user_coders,
450 simulated = self.simulate,
451 items_generated = total_items,
452 )
453 await _core.render_and_print_result(
454 result, auxdata.display, auxdata.exits )
457class PopulateCommand( __.appcore_cli.Command ):
458 ''' Populates agent content and configuration. '''
460 command: __.typx.Union[
461 __.typx.Annotated[
462 PopulateProjectCommand,
463 __.tyro.conf.subcommand( 'project', prefix_name = False ),
464 ],
465 __.typx.Annotated[
466 PopulateUserCommand,
467 __.tyro.conf.subcommand( 'user', prefix_name = False ),
468 ],
469 ] = __.dcls.field( default_factory = PopulateProjectCommand )
471 async def execute( self, auxdata: __.appcore.state.Globals ) -> None: # pyright: ignore[reportIncompatibleMethodOverride]
472 await self.command( auxdata )