Coverage for sources / agentsmgr / population.py: 19%
163 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-12-01 15:37 +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 list 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_memory = (
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_memory )
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 openspec_link_path = target / 'openspec'
123 openspec_source_path = (
124 target / 'documentation' / 'architecture' / 'openspec' )
125 if not simulate:
126 openspec_source_path.mkdir( parents = True, exist_ok = True )
127 _, symlink_name_openspec = _memorylinks.create_memory_symlink(
128 openspec_source_path, openspec_link_path, simulate )
129 all_symlink_names.append( symlink_name_openspec )
130 _scribe.info( "Created 1/1 openspec symlink" )
131 return tuple( all_symlink_names )
134def _populate_instructions_if_configured(
135 configuration: __.cabc.Mapping[ str, __.typx.Any ],
136 target: __.Path,
137 tag_prefix: __.Absential[ str ],
138 simulate: bool,
139) -> tuple[ bool, str ]:
140 ''' Populates instructions if configured and returns status.
142 Returns tuple of (sources_present, instructions_target_path).
143 sources_present indicates whether instruction sources were
144 configured and processed.
145 '''
146 if not configuration.get( 'provide_instructions', False ):
147 return ( False, '' )
148 instructions_sources = configuration.get( 'instructions_sources', [ ] )
149 instructions_target = configuration.get(
150 'instructions_target', '.auxiliary/instructions' )
151 if not instructions_sources:
152 return ( False, instructions_target )
153 instructions_attempted, instructions_updated = (
154 _instructions.populate_instructions(
155 instructions_sources,
156 target / instructions_target,
157 tag_prefix,
158 simulate,
159 ) )
160 _scribe.info(
161 f"Updated {instructions_updated}/{instructions_attempted} "
162 "instruction files" )
163 return ( True, instructions_target )
166def _populate_per_user_content(
167 location: __.Path,
168 coders: __.cabc.Sequence[ str ],
169 configuration: __.cabc.Mapping[ str, __.typx.Any ],
170 application_configuration: __.cabc.Mapping[ str, __.typx.Any ],
171 simulate: bool,
172) -> tuple[ int, int ]:
173 ''' Populates commands and agents for per-user coders.
175 Generates content to each coder's per-user directory using
176 renderer's resolve_base_directory() with per-user mode.
177 Returns tuple of (items_attempted, items_generated).
178 '''
179 items_attempted = 0
180 items_generated = 0
181 for coder_name in coders:
182 try: renderer = _renderers.RENDERERS[ coder_name ]
183 except KeyError as exception:
184 raise _exceptions.CoderAbsence( coder_name ) from exception
185 coder_configuration = { 'coders': [ coder_name ] }
186 generator = _generator.ContentGenerator(
187 location = location,
188 configuration = coder_configuration,
189 application_configuration = application_configuration,
190 mode = 'per-user',
191 )
192 target = renderer.resolve_base_directory(
193 'per-user', __.Path.cwd( ), configuration, dict( __.os.environ ) )
194 attempted, generated = _operations.populate_directory(
195 generator, target, simulate )
196 items_attempted += attempted
197 items_generated += generated
198 return ( items_attempted, items_generated )
201def _create_coder_directory_symlinks(
202 coders: __.cabc.Sequence[ str ],
203 target: __.Path,
204 renderers: __.cabc.Mapping[ str, __.typx.Any ],
205 simulate: bool = False,
206) -> tuple[ int, int, tuple[ str, ... ] ]:
207 ''' Creates symlinks from .{coder} to .auxiliary/configuration/coders/.
209 For per-project mode, creates symlinks that make coder directories
210 accessible at their expected locations (.claude, .opencode, etc.)
211 while keeping actual files organized under
212 .auxiliary/configuration/coders/.
214 Each renderer is responsible for specifying its symlink requirements
215 via provide_project_symlinks(). Population logic simply iterates
216 coders and asks renderers for their symlinks.
218 Only creates symlinks for coders whose default mode is per-project.
219 Coders with per-user default mode are skipped since they do not
220 use per-project directories.
222 Returns tuple of (attempted, created, symlink_names) where
223 symlink_names contains names of all symlinks (both newly created
224 and pre-existing).
225 '''
226 attempted = 0
227 created = 0
228 symlink_names: list[ str ] = [ ]
230 for coder_name in coders:
231 try: renderer = renderers[ coder_name ]
232 except KeyError as exception:
233 raise _exceptions.CoderAbsence( coder_name ) from exception
234 if renderer.mode_default != 'per-project':
235 _scribe.debug(
236 f"Skipping directory symlink for {coder_name}: "
237 f"default mode is {renderer.mode_default}" )
238 continue
239 for source, link_path in renderer.provide_project_symlinks( target ):
240 attempted += 1
241 was_created, symlink_name = (
242 _memorylinks.create_memory_symlink(
243 source, link_path, simulate ) )
244 if was_created: created += 1
245 symlink_names.append( symlink_name )
246 return ( attempted, created, tuple( symlink_names ) )
249def _copy_coder_resources(
250 location: __.Path,
251 target: __.Path,
252 coders: __.cabc.Sequence[ str ],
253 simulate: bool
254) -> tuple[ int, int ]:
255 ''' Copies static resources for coders.
257 Copies resources from defaults/per-project/resources/<coder>
258 to target directory. Returns tuple of (coders_attempted,
259 coders_processed).
260 '''
261 resources_source = location / 'per-project' / 'resources'
262 if not resources_source.exists( ):
263 _scribe.debug(
264 f"No per-project resources found at {resources_source}" )
265 return ( 0, 0 )
266 coders_target = target / '.auxiliary' / 'configuration' / 'coders'
267 return _operations.copy_coder_resources(
268 resources_source, coders_target, coders, simulate )
271def _manage_project_auxiliaries(
272 configuration: __.cabc.Mapping[ str, __.typx.Any ],
273 target: __.Path,
274 tag_prefix: __.Absential[ str ],
275 simulate: bool
276) -> None:
277 ''' Manages auxiliary project files (instructions, symlinks, excludes). '''
278 instructions_populated, instructions_target = (
279 _populate_instructions_if_configured(
280 configuration, target, tag_prefix, simulate ) )
281 all_symlink_names: list[ str ] = list( _create_all_symlinks(
282 configuration, target, 'per-project', simulate ) )
283 git_exclude_entries: list[ str ] = [ ]
284 if instructions_populated:
285 git_exclude_entries.append( instructions_target )
286 git_exclude_entries.extend( all_symlink_names )
287 if git_exclude_entries:
288 entries_count = _operations.update_git_exclude(
289 target, git_exclude_entries, simulate )
290 if entries_count > 0:
291 _scribe.info(
292 f"Managing {entries_count} entries in .git/info/exclude" )
295class PopulateProjectCommand( __.appcore_cli.Command ):
296 ''' Generates project-scoped agent content from data sources.
298 Populates agent commands, definitions, and static resources
299 from the specified data source. Copies static resources from
300 defaults/per-project/resources/ to the project configuration.
301 '''
303 source: SourceArgument = '.'
304 target: TargetArgument = __.dcls.field( default_factory = __.Path.cwd )
305 profile: __.typx.Annotated[
306 __.typx.Optional[ __.Path ],
307 __.tyro.conf.arg(
308 help = (
309 "Alternative Copier answers file (defaults to "
310 "auto-detected)" ),
311 prefix_name = False ),
312 ] = None
313 simulate: __.typx.Annotated[
314 bool,
315 __.tyro.conf.arg(
316 help = "Dry run mode - show generated content",
317 prefix_name = False ),
318 ] = False
319 tag_prefix: __.typx.Annotated[
320 __.typx.Optional[ str ],
321 __.tyro.conf.arg(
322 help = (
323 "Prefix for version tags (e.g., 'v', 'stable-', 'prod-'); "
324 "only tags with this prefix are considered and the prefix "
325 "is stripped before version parsing" ),
326 prefix_name = False ),
327 ] = None
329 @_cmdbase.intercept_errors( )
330 async def execute( self, auxdata: __.appcore.state.Globals ) -> None: # pyright: ignore[reportIncompatibleMethodOverride]
331 ''' Generates project content from data sources. '''
332 if not isinstance( auxdata, _core.Globals ): # pragma: no cover
333 raise _exceptions.ContextInvalidity
334 _scribe.info(
335 f"Populating project content from {self.source} to {self.target}" )
336 configuration = await _cmdbase.retrieve_configuration(
337 self.target, self.profile )
338 per_project_coders = _filter_coders_by_mode(
339 configuration[ 'coders' ], 'per-project', _renderers.RENDERERS )
340 if not per_project_coders:
341 _scribe.warning(
342 "No per-project default coders found in configuration" )
343 return
344 filtered_configuration = dict( configuration )
345 filtered_configuration[ 'coders' ] = per_project_coders
346 prefix = __.absent if self.tag_prefix is None else self.tag_prefix
347 location = _cmdbase.retrieve_data_location( self.source, prefix )
348 _cmdbase.validate_data_source_structure(
349 location,
350 ( 'configurations', 'contents', 'templates' ) )
351 generator = _generator.ContentGenerator(
352 location = location,
353 configuration = filtered_configuration,
354 application_configuration = auxdata.configuration,
355 mode = 'per-project',
356 )
357 items_attempted, items_generated = _operations.populate_directory(
358 generator, self.target, self.simulate )
359 _scribe.info( f"Generated {items_generated}/{items_attempted} items" )
360 resources_attempted, resources_copied = _copy_coder_resources(
361 location,
362 self.target,
363 filtered_configuration[ 'coders' ],
364 self.simulate )
365 if resources_copied > 0:
366 _scribe.info(
367 f"Copied resources for "
368 f"{resources_copied}/{resources_attempted} coders" )
369 _manage_project_auxiliaries(
370 filtered_configuration, self.target, prefix, self.simulate )
371 result = _results.ContentGenerationResult(
372 source_location = location,
373 target_location = self.target,
374 coders = tuple( configuration[ 'coders' ] ),
375 simulated = self.simulate,
376 items_generated = items_generated,
377 )
378 await _core.render_and_print_result(
379 result, auxdata.display, auxdata.exits )
382class PopulateUserCommand( __.appcore_cli.Command ):
383 ''' Populates per-user global settings and executables. '''
385 source: SourceArgument = '.'
386 profile: __.typx.Annotated[
387 __.typx.Optional[ __.Path ],
388 __.tyro.conf.arg(
389 help = (
390 "Alternative Copier answers file (defaults to "
391 "auto-detected)" ),
392 prefix_name = False ),
393 ] = None
394 simulate: __.typx.Annotated[
395 bool,
396 __.tyro.conf.arg(
397 help = "Dry run mode - show what would be installed",
398 prefix_name = False ),
399 ] = False
400 tag_prefix: __.typx.Annotated[
401 __.typx.Optional[ str ],
402 __.tyro.conf.arg(
403 help = (
404 "Prefix for version tags (e.g., 'v', 'stable-', 'prod-'); "
405 "only tags with this prefix are considered and the prefix "
406 "is stripped before version parsing" ),
407 prefix_name = False ),
408 ] = None
410 @_cmdbase.intercept_errors( )
411 async def execute( self, auxdata: __.appcore.state.Globals ) -> None: # pyright: ignore[reportIncompatibleMethodOverride]
412 ''' Populates user-scoped settings and executables. '''
413 if not isinstance( auxdata, _core.Globals ): # pragma: no cover
414 raise _exceptions.ContextInvalidity
415 _scribe.info( f"Populating user configuration from {self.source}" )
416 configuration = await _cmdbase.retrieve_configuration(
417 __.Path.cwd( ), self.profile )
418 per_user_coders = _filter_coders_by_mode(
419 configuration[ 'coders' ], 'per-user', _renderers.RENDERERS )
420 if not per_user_coders:
421 _scribe.warning(
422 "No per-user default coders found in configuration" )
423 return
424 prefix = __.absent if self.tag_prefix is None else self.tag_prefix
425 location = _cmdbase.retrieve_data_location( self.source, prefix )
426 _cmdbase.validate_data_source_structure(
427 location,
428 ( 'configurations', 'contents', 'templates',
429 'user/configurations', 'user/executables' ) )
430 content_attempted, content_generated = _populate_per_user_content(
431 location,
432 per_user_coders,
433 configuration,
434 auxdata.configuration,
435 self.simulate,
436 )
437 if content_attempted > 0:
438 _scribe.info(
439 f"Generated {content_generated}/{content_attempted} items" )
440 globals_attempted, globals_updated = _userdata.populate_globals(
441 location,
442 per_user_coders,
443 auxdata.configuration,
444 self.simulate,
445 )
446 _scribe.info(
447 f"Updated {globals_updated}/{globals_attempted} global files" )
448 wrappers_attempted, wrappers_installed = (
449 _userdata.populate_user_wrappers( location, self.simulate ) )
450 if wrappers_attempted > 0:
451 _scribe.info(
452 f"Installed {wrappers_installed}/{wrappers_attempted} "
453 "wrapper scripts" )
454 total_items = content_generated + globals_updated + wrappers_installed
455 result = _results.ContentGenerationResult(
456 source_location = location,
457 target_location = __.Path.home( ),
458 coders = per_user_coders,
459 simulated = self.simulate,
460 items_generated = total_items,
461 )
462 await _core.render_and_print_result(
463 result, auxdata.display, auxdata.exits )
466class PopulateCommand( __.appcore_cli.Command ):
467 ''' Populates agent content and configuration. '''
469 command: __.typx.Union[
470 __.typx.Annotated[
471 PopulateProjectCommand,
472 __.tyro.conf.subcommand( 'project', prefix_name = False ),
473 ],
474 __.typx.Annotated[
475 PopulateUserCommand,
476 __.tyro.conf.subcommand( 'user', prefix_name = False ),
477 ],
478 ] = __.dcls.field( default_factory = PopulateProjectCommand )
480 async def execute( self, auxdata: __.appcore.state.Globals ) -> None: # pyright: ignore[reportIncompatibleMethodOverride]
481 await self.command( auxdata )