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