Coverage for sources/agentsmgr/population.py: 19%
152 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-10-26 02:00 +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 Only creates symlinks for coders whose default mode is per-project.
206 Coders with per-user default mode are skipped since they do not
207 use per-project directories.
209 Returns tuple of (attempted, created, symlink_names) where
210 symlink_names contains names of all symlinks (both newly created
211 and pre-existing).
212 '''
213 # TODO: Move symlink rendering to coders and call common code from each
214 # one. Should have not have coder-specific logic in this general
215 # function.
216 attempted = 0
217 created = 0
218 symlink_names: list[ str ] = [ ]
219 for coder_name in coders:
220 try: renderer = renderers[ coder_name ]
221 except KeyError as exception:
222 raise _exceptions.CoderAbsence( coder_name ) from exception
223 if renderer.mode_default != 'per-project':
224 _scribe.debug(
225 f"Skipping directory symlink for {coder_name}: "
226 f"default mode is {renderer.mode_default}" )
227 continue
228 source = (
229 target / '.auxiliary' / 'configuration' / 'coders' / coder_name )
230 link_path = target / f'.{coder_name}'
231 attempted += 1
232 was_created, symlink_name = _memorylinks.create_memory_symlink(
233 source, link_path, simulate )
234 if was_created: created += 1
235 symlink_names.append( symlink_name )
236 if coder_name == 'claude':
237 mcp_source = (
238 target / '.auxiliary' / 'configuration' / 'mcp-servers.json' )
239 mcp_link = target / '.mcp.json'
240 attempted += 1
241 was_created, symlink_name = _memorylinks.create_memory_symlink(
242 mcp_source, mcp_link, simulate )
243 if was_created: created += 1
244 symlink_names.append( symlink_name )
245 return ( attempted, created, tuple( symlink_names ) )
248class PopulateProjectCommand( __.appcore_cli.Command ):
249 ''' Generates project-scoped agent content from data sources. '''
251 source: SourceArgument = '.'
252 target: TargetArgument = __.dcls.field( default_factory = __.Path.cwd )
253 profile: __.typx.Annotated[
254 __.typx.Optional[ __.Path ],
255 __.tyro.conf.arg(
256 help = (
257 "Alternative Copier answers file (defaults to "
258 "auto-detected)" ),
259 prefix_name = False ),
260 ] = None
261 simulate: __.typx.Annotated[
262 bool,
263 __.tyro.conf.arg(
264 help = "Dry run mode - show generated content",
265 prefix_name = False ),
266 ] = False
267 tag_prefix: __.typx.Annotated[
268 __.typx.Optional[ str ],
269 __.tyro.conf.arg(
270 help = (
271 "Prefix for version tags (e.g., 'v', 'stable-', 'prod-'); "
272 "only tags with this prefix are considered and the prefix "
273 "is stripped before version parsing" ),
274 prefix_name = False ),
275 ] = None
277 @_cmdbase.intercept_errors( )
278 async def execute( self, auxdata: __.appcore.state.Globals ) -> None: # pyright: ignore[reportIncompatibleMethodOverride]
279 ''' Generates project content from data sources. '''
280 if not isinstance( auxdata, _core.Globals ): # pragma: no cover
281 raise _exceptions.ContextInvalidity
282 _scribe.info(
283 f"Populating project content from {self.source} to {self.target}" )
284 configuration = await _cmdbase.retrieve_configuration(
285 self.target, self.profile )
286 per_project_coders = _filter_coders_by_mode(
287 configuration[ 'coders' ], 'per-project', _renderers.RENDERERS )
288 if not per_project_coders:
289 _scribe.warning(
290 "No per-project default coders found in configuration" )
291 return
292 filtered_configuration = dict( configuration )
293 filtered_configuration[ 'coders' ] = per_project_coders
294 prefix = __.absent if self.tag_prefix is None else self.tag_prefix
295 location = _cmdbase.retrieve_data_location( self.source, prefix )
296 _cmdbase.validate_data_source_structure(
297 location,
298 ( 'configurations', 'contents', 'templates' ) )
299 generator = _generator.ContentGenerator(
300 location = location,
301 configuration = filtered_configuration,
302 application_configuration = auxdata.configuration,
303 mode = 'per-project',
304 )
305 items_attempted, items_generated = _operations.populate_directory(
306 generator, self.target, self.simulate )
307 _scribe.info( f"Generated {items_generated}/{items_attempted} items" )
308 instructions_populated, instructions_target = (
309 _populate_instructions_if_configured(
310 filtered_configuration, self.target, prefix, self.simulate ) )
311 all_symlink_names = _create_all_symlinks(
312 filtered_configuration, self.target, 'per-project', self.simulate )
313 git_exclude_entries: list[ str ] = [ ]
314 if instructions_populated:
315 git_exclude_entries.append( instructions_target )
316 git_exclude_entries.extend( all_symlink_names )
317 if git_exclude_entries:
318 entries_count = _operations.update_git_exclude(
319 self.target, git_exclude_entries, self.simulate )
320 if entries_count > 0:
321 _scribe.info(
322 f"Managing {entries_count} entries in .git/info/exclude" )
323 result = _results.ContentGenerationResult(
324 source_location = location,
325 target_location = self.target,
326 coders = tuple( configuration[ 'coders' ] ),
327 simulated = self.simulate,
328 items_generated = items_generated,
329 )
330 await _core.render_and_print_result(
331 result, auxdata.display, auxdata.exits )
334class PopulateUserCommand( __.appcore_cli.Command ):
335 ''' Populates per-user global settings and executables. '''
337 source: SourceArgument = '.'
338 profile: __.typx.Annotated[
339 __.typx.Optional[ __.Path ],
340 __.tyro.conf.arg(
341 help = (
342 "Alternative Copier answers file (defaults to "
343 "auto-detected)" ),
344 prefix_name = False ),
345 ] = None
346 simulate: __.typx.Annotated[
347 bool,
348 __.tyro.conf.arg(
349 help = "Dry run mode - show what would be installed",
350 prefix_name = False ),
351 ] = False
352 tag_prefix: __.typx.Annotated[
353 __.typx.Optional[ str ],
354 __.tyro.conf.arg(
355 help = (
356 "Prefix for version tags (e.g., 'v', 'stable-', 'prod-'); "
357 "only tags with this prefix are considered and the prefix "
358 "is stripped before version parsing" ),
359 prefix_name = False ),
360 ] = None
362 @_cmdbase.intercept_errors( )
363 async def execute( self, auxdata: __.appcore.state.Globals ) -> None: # pyright: ignore[reportIncompatibleMethodOverride]
364 ''' Populates user-scoped settings and executables. '''
365 if not isinstance( auxdata, _core.Globals ): # pragma: no cover
366 raise _exceptions.ContextInvalidity
367 _scribe.info( f"Populating user configuration from {self.source}" )
368 configuration = await _cmdbase.retrieve_configuration(
369 __.Path.cwd( ), self.profile )
370 per_user_coders = _filter_coders_by_mode(
371 configuration[ 'coders' ], 'per-user', _renderers.RENDERERS )
372 if not per_user_coders:
373 _scribe.warning(
374 "No per-user default coders found in configuration" )
375 return
376 prefix = __.absent if self.tag_prefix is None else self.tag_prefix
377 location = _cmdbase.retrieve_data_location( self.source, prefix )
378 _cmdbase.validate_data_source_structure(
379 location,
380 ( 'configurations', 'contents', 'templates',
381 'user/configurations', 'user/executables' ) )
382 content_attempted, content_generated = _populate_per_user_content(
383 location,
384 per_user_coders,
385 configuration,
386 auxdata.configuration,
387 self.simulate,
388 )
389 if content_attempted > 0:
390 _scribe.info(
391 f"Generated {content_generated}/{content_attempted} items" )
392 globals_attempted, globals_updated = _userdata.populate_globals(
393 location,
394 per_user_coders,
395 auxdata.configuration,
396 self.simulate,
397 )
398 _scribe.info(
399 f"Updated {globals_updated}/{globals_attempted} global files" )
400 wrappers_attempted, wrappers_installed = (
401 _userdata.populate_user_wrappers( location, self.simulate ) )
402 if wrappers_attempted > 0:
403 _scribe.info(
404 f"Installed {wrappers_installed}/{wrappers_attempted} "
405 "wrapper scripts" )
406 total_items = content_generated + globals_updated + wrappers_installed
407 result = _results.ContentGenerationResult(
408 source_location = location,
409 target_location = __.Path.home( ),
410 coders = per_user_coders,
411 simulated = self.simulate,
412 items_generated = total_items,
413 )
414 await _core.render_and_print_result(
415 result, auxdata.display, auxdata.exits )
418class PopulateCommand( __.appcore_cli.Command ):
419 ''' Populates agent content and configuration. '''
421 command: __.typx.Union[
422 __.typx.Annotated[
423 PopulateProjectCommand,
424 __.tyro.conf.subcommand( 'project', prefix_name = False ),
425 ],
426 __.typx.Annotated[
427 PopulateUserCommand,
428 __.tyro.conf.subcommand( 'user', prefix_name = False ),
429 ],
430 ] = __.dcls.field( default_factory = PopulateProjectCommand )
432 async def execute( self, auxdata: __.appcore.state.Globals ) -> None: # pyright: ignore[reportIncompatibleMethodOverride]
433 await self.command( auxdata )