Coverage for sources/agentsmgr/population.py: 82%
262 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-15 21:08 +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.
23 Also provides the generate command for maintainer-facing generation
24 from components/ to distribution/.
25'''
28from . import __
29from . import cmdbase as _cmdbase
30from . import core as _core
31from . import exceptions as _exceptions
32from . import generator as _generator
33from . import memorylinks as _memorylinks
34from . import operations as _operations
35from . import renderers as _renderers
36from . import resolver as _resolver
37from . import results as _results
38from . import userdata as _userdata
41_scribe = __.provide_scribe( __name__ )
44def _produce_default_configuration(
45 location: __.Path,
46) -> __.cabc.Mapping[ str, __.typx.Any ]:
47 ''' Produces default configuration for generate command.
49 Uses all known coders from the renderer registry so that
50 fallback content is generated for all coders, not just those
51 with direct component content.
52 '''
53 from . import renderers as _renderers
54 coders = sorted( _renderers.RENDERERS.keys( ) )
55 return { 'coders': coders, 'languages': [ 'python' ] }
58SourceArgument: __.typx.TypeAlias = __.typx.Annotated[
59 __.tyro.conf.Positional[ str ],
60 __.tyro.conf.arg( help = "Data source (local path or git URL)" ),
61]
62TargetArgument: __.typx.TypeAlias = __.typx.Annotated[
63 __.tyro.conf.Positional[ __.Path ],
64 __.tyro.conf.arg( help = "Target directory for content generation" ),
65]
68def _filter_coders_by_mode(
69 coders: __.cabc.Sequence[ str ],
70 target_mode: _renderers.ExplicitTargetMode,
71) -> tuple[ str, ... ]:
72 ''' Filters coders by their default targeting mode.
74 Returns coders whose mode_default matches the target mode.
75 This ensures populate project only handles per-project coders
76 and populate user only handles per-user coders, respecting each
77 renderer's designed usage pattern.
78 '''
79 return tuple(
80 name
81 for name, _renderer in _resolver.resolve_coders(
82 coders, mode = target_mode )
83 )
86def _format_exclude_path( path: __.Path ) -> str:
87 ''' Formats a project-relative path for git exclude syntax. '''
88 return path.as_posix( )
91def _create_all_symlinks(
92 configuration: __.cabc.Mapping[ str, __.typx.Any ],
93 target: __.Path,
94 mode: str,
95 simulate: bool,
96) -> tuple[ str, ... ]:
97 ''' Creates all symlinks and returns their names for git exclude.
99 Creates memory symlinks for all coders and coder directory
100 symlinks for per-project mode. Returns list of all symlink
101 names (both newly created and pre-existing) for git exclude
102 update.
103 '''
104 all_symlink_names: list[ str ] = [ ]
105 if mode == 'nowhere': return tuple( all_symlink_names ) 105 ↛ exitline 105 didn't return from function '_create_all_symlinks' because the return on line 105 wasn't executed
106 links_attempted, links_created, symlink_names_memory = (
107 _memorylinks.create_memory_symlinks_for_coders(
108 coders = configuration[ 'coders' ],
109 target = target,
110 simulate = simulate,
111 ) )
112 all_symlink_names.extend( symlink_names_memory )
113 if links_created > 0:
114 _scribe.info(
115 f"Created {links_created}/{links_attempted} memory symlinks" )
116 needs_coder_symlinks = (
117 mode == 'per-project'
118 or ( mode == 'default' and any(
119 coder in _renderers.RENDERERS
120 and _renderers.RENDERERS[ coder ].mode_default == 'per-project'
121 for coder in configuration[ 'coders' ] ) ) )
122 if needs_coder_symlinks: 122 ↛ 136line 122 didn't jump to line 136 because the condition on line 122 was always true
123 ( coder_symlinks_attempted,
124 coder_symlinks_created,
125 coder_symlink_names ) = (
126 _create_coder_directory_symlinks(
127 coders = configuration[ 'coders' ],
128 target = target,
129 simulate = simulate,
130 ) )
131 all_symlink_names.extend( coder_symlink_names )
132 if coder_symlinks_created > 0:
133 _scribe.info(
134 f"Created {coder_symlinks_created}/"
135 f"{coder_symlinks_attempted} coder directory symlinks" )
136 openspec_link_path = target / 'openspec'
137 openspec_source_path = (
138 target / 'documentation' / 'architecture' / 'openspec' )
139 if not simulate: 139 ↛ 141line 139 didn't jump to line 141 because the condition on line 139 was always true
140 openspec_source_path.mkdir( parents = True, exist_ok = True )
141 _, symlink_name_openspec = _memorylinks.create_memory_symlink(
142 openspec_source_path, openspec_link_path, simulate )
143 all_symlink_names.append( symlink_name_openspec )
144 _scribe.info( "Created 1/1 openspec symlink" )
145 return tuple( all_symlink_names )
148def _copy_instructions_from_distribution(
149 distribution: __.Path,
150 target: __.Path,
151 instructions_target: str,
152 simulate: bool,
153) -> tuple[ int, int, tuple[ str, ... ] ]:
154 ''' Copies instruction files from distribution/ to target.
156 Reads from distribution/per-project/general/instructions/ and
157 copies to the configured instructions target path.
158 Returns tuple of (files_attempted, files_written, exclude_entries).
159 '''
160 import contextlib as _contextlib
161 source_dir = distribution / 'per-project' / 'general' / 'instructions'
162 if not source_dir.exists( ): 162 ↛ 163line 162 didn't jump to line 163 because the condition on line 162 was never true
163 return ( 0, 0, ( ) )
164 target_dir = target / instructions_target
165 files_attempted = 0
166 files_written = 0
167 exclude_entries: list[ str ] = [ ]
168 for source_file in source_dir.glob( '*' ):
169 if not source_file.is_file( ): continue 169 ↛ 168line 169 didn't jump to line 168 because the continue on line 169 wasn't executed
170 files_attempted += 1
171 dest_path = target_dir / source_file.name
172 if _operations.save_content_text( 172 ↛ 178line 172 didn't jump to line 178 because the condition on line 172 was always true
173 source_file.read_text( encoding = 'utf-8' ),
174 dest_path,
175 simulate,
176 ):
177 files_written += 1
178 with _contextlib.suppress( ValueError ):
179 exclude_entries.append(
180 _format_exclude_path( dest_path.relative_to( target ) ) )
181 return ( files_attempted, files_written, tuple( exclude_entries ) )
184def _populate_per_user_content(
185 location: __.Path,
186 coders: __.cabc.Sequence[ str ],
187 configuration: __.cabc.Mapping[ str, __.typx.Any ],
188 simulate: bool,
189) -> tuple[ int, int ]:
190 ''' Populates commands, agents, and skills for per-user coders.
192 Copies distribution items to each coder's per-user directory.
193 Returns tuple of (items_attempted, items_written).
194 '''
195 attempted, written, _ = _copy_distribution_items(
196 location,
197 coders,
198 __.Path.cwd( ),
199 configuration = configuration,
200 mode = 'per-user',
201 simulate = simulate )
202 return ( attempted, written )
205def _create_coder_directory_symlinks(
206 coders: __.cabc.Sequence[ str ],
207 target: __.Path,
208 simulate: bool = False,
209) -> tuple[ int, int, tuple[ str, ... ] ]:
210 ''' Creates symlinks from .{coder} to .auxiliary/configuration/coders/.
212 For per-project mode, creates symlinks that make coder directories
213 accessible at their expected locations (.claude, .opencode, etc.)
214 while keeping actual files organized under
215 .auxiliary/configuration/coders/.
217 Each renderer is responsible for specifying its symlink requirements
218 via provide_project_symlinks(). Population logic simply iterates
219 coders and asks renderers for their symlinks.
221 Only creates symlinks for coders whose default mode is per-project.
222 Coders with per-user default mode are skipped since they do not
223 use per-project directories.
225 Returns tuple of (attempted, created, symlink_names) where
226 symlink_names contains names of all symlinks (both newly created
227 and pre-existing).
228 '''
229 attempted = 0
230 created = 0
231 symlink_names: list[ str ] = [ ]
232 for coder_name, renderer in _resolver.resolve_coders(
233 coders, mode = 'per-project'
234 ):
235 for source, link_path in renderer.provide_project_symlinks( target ):
236 attempted += 1
237 was_created, symlink_name = (
238 _memorylinks.create_memory_symlink(
239 source, link_path, simulate ) )
240 if was_created: created += 1
241 symlink_names.append( symlink_name )
242 return ( attempted, created, tuple( symlink_names ) )
245def _copy_distribution_items( # noqa: PLR0913
246 distribution: __.Path,
247 coders: __.cabc.Sequence[ str ],
248 target: __.Path,
249 *,
250 configuration: __.cabc.Mapping[ str, __.typx.Any ],
251 mode: _renderers.ExplicitTargetMode,
252 simulate: bool,
253) -> tuple[ int, int, tuple[ str, ... ] ]:
254 ''' Copies distribution items to downstream target paths.
256 For each coder, copies the entire
257 distribution/<mode>/coders/<coder>/ tree to the target.
258 Skills are copied separately from
259 distribution/per-project/general/skills/.
261 The distribution tree mirrors downstream layout, so this is
262 a single copy operation per coder.
264 Returns tuple of (items_attempted, items_written, exclude_entries).
265 '''
266 items_attempted = 0
267 items_written = 0
268 exclude_entries: list[ str ] = [ ]
269 for coder_name, manager in _resolver.resolve_coders(
270 coders, mode = mode
271 ):
272 base_directory = manager.resolve_base_directory(
273 mode = mode,
274 target = target,
275 configuration = configuration,
276 environment = __.os.environ,
277 )
278 coder_source = distribution / mode / 'coders' / coder_name
279 # Copy entire coder tree (commands, agents, resources).
280 if coder_source.exists( ): 280 ↛ 287line 280 didn't jump to line 287 because the condition on line 280 was always true
281 attempted, written, entries = _copy_tree(
282 coder_source, base_directory, target, simulate )
283 items_attempted += attempted
284 items_written += written
285 exclude_entries.extend( entries )
286 # Copy skills from general directory.
287 if mode == 'per-project': 287 ↛ 269line 287 didn't jump to line 269 because the condition on line 287 was always true
288 attempted, written, entries = _copy_skills(
289 distribution, base_directory, manager, target, simulate )
290 items_attempted += attempted
291 items_written += written
292 exclude_entries.extend( entries )
293 return ( items_attempted, items_written, tuple( exclude_entries ) )
296def _copy_tree(
297 source: __.Path,
298 target: __.Path,
299 project_root: __.Path,
300 simulate: bool,
301) -> tuple[ int, int, tuple[ str, ... ] ]:
302 ''' Copies directory tree from source to target.
304 Recursively copies all files and subdirectories (binary-safe) and
305 preserves each source file mode (including executable bits).
306 Returns tuple of (files_attempted, files_written, exclude_entries).
307 '''
308 import contextlib as _contextlib
309 files_attempted = 0
310 files_written = 0
311 exclude_entries: list[ str ] = [ ]
312 for source_file in source.rglob( '*' ):
313 if not source_file.is_file( ): continue
314 files_attempted += 1
315 relative = source_file.relative_to( source )
316 dest_path = target / relative
317 if _operations.save_content_bytes( 317 ↛ 324line 317 didn't jump to line 324 because the condition on line 317 was always true
318 source_file.read_bytes( ),
319 dest_path,
320 simulate,
321 ):
322 files_written += 1
323 __.shutil.copymode( source_file, dest_path )
324 with _contextlib.suppress( ValueError ):
325 exclude_entries.append(
326 _format_exclude_path(
327 dest_path.relative_to( project_root ) ) )
328 return ( files_attempted, files_written, tuple( exclude_entries ) )
331def _copy_skills(
332 distribution: __.Path,
333 base_directory: __.Path,
334 manager: _renderers.RendererBase,
335 project_root: __.Path,
336 simulate: bool,
337) -> tuple[ int, int, tuple[ str, ... ] ]:
338 ''' Copies skill packages from distribution/ to target paths.
340 Skills are static artifacts that require no rendering. Layout follows
341 the Agent Skills specification
342 (https://agentskills.io/specification.md). Source layout under
343 ``distribution/per-project/general/skills/`` may be either:
345 - Directory package (Agent Skills): ``<name>/SKILL.md`` plus optional
346 ``scripts/``, ``references/``, ``assets/``, and other supporting
347 files — the entire directory is copied to
348 ``<base>/skills/<name>/``.
349 - Legacy flat file: ``<name>.md`` → ``<base>/skills/<name>/SKILL.md``.
351 When both a directory package and a flat ``<name>.md`` exist for the
352 same name, the directory package wins. Supporting files are not
353 language-filtered; a skill is a portable package. Destination
354 ``SKILL.md`` naming follows the Skills protocol and does not use
355 ``manager.calculate_artifact_pattern``. Returns tuple of
356 (files_attempted, files_written, exclude_entries).
357 '''
358 import contextlib as _contextlib
359 items_attempted = 0
360 items_written = 0
361 exclude_entries: list[ str ] = [ ]
362 skills_dir = (
363 distribution / 'per-project' / 'general' / 'skills' )
364 if not skills_dir.exists( ): 364 ↛ 365line 364 didn't jump to line 365 because the condition on line 364 was never true
365 return ( items_attempted, items_written, tuple( exclude_entries ) )
366 skills_output = manager.calculate_directory_location( 'skills' )
367 directory_skills: dict[ str, __.Path ] = { }
368 flat_skills: dict[ str, __.Path ] = { }
369 for entry in skills_dir.iterdir( ):
370 if entry.is_dir( ) and ( entry / 'SKILL.md' ).is_file( ):
371 directory_skills[ entry.name ] = entry
372 elif entry.is_file( ) and entry.suffix == '.md': 372 ↛ 369line 372 didn't jump to line 369 because the condition on line 372 was always true
373 flat_skills[ entry.stem ] = entry
374 for item_name, source_dir in sorted( directory_skills.items( ) ):
375 dest_root = base_directory / skills_output / item_name
376 attempted, written, entries = _copy_tree(
377 source_dir, dest_root, project_root, simulate )
378 items_attempted += attempted
379 items_written += written
380 exclude_entries.extend( entries )
381 for item_name, skill_file in sorted( flat_skills.items( ) ):
382 if item_name in directory_skills: continue
383 items_attempted += 1
384 dest_path = (
385 base_directory / skills_output / item_name / 'SKILL.md' )
386 if _operations.save_content_bytes( 386 ↛ 393line 386 didn't jump to line 393 because the condition on line 386 was always true
387 skill_file.read_bytes( ),
388 dest_path,
389 simulate,
390 ):
391 items_written += 1
392 __.shutil.copymode( skill_file, dest_path )
393 with _contextlib.suppress( ValueError ):
394 exclude_entries.append(
395 _format_exclude_path(
396 dest_path.relative_to( project_root ) ) )
397 return ( items_attempted, items_written, tuple( exclude_entries ) )
400def _manage_project_auxiliaries(
401 configuration: __.cabc.Mapping[ str, __.typx.Any ],
402 distribution: __.Path,
403 target: __.Path,
404 distribution_entries: __.cabc.Sequence[ str ],
405 simulate: bool
406) -> None:
407 ''' Manages auxiliary project files (instructions, symlinks, excludes). '''
408 instruction_entries: tuple[ str, ... ] = ( )
409 if configuration.get( 'provide_instructions', False ):
410 instructions_target = configuration.get(
411 'instructions_target', '.auxiliary/agents/standards' )
412 instructions_attempted, instructions_written, instruction_entries = (
413 _copy_instructions_from_distribution(
414 distribution, target, instructions_target, simulate ) )
415 if instructions_written > 0: 415 ↛ 419line 415 didn't jump to line 419 because the condition on line 415 was always true
416 _scribe.info(
417 f"Copied {instructions_written}/{instructions_attempted} "
418 "instruction files" )
419 all_symlink_names: list[ str ] = list( _create_all_symlinks(
420 configuration, target, 'per-project', simulate ) )
421 git_exclude_entries: list[ str ] = list( distribution_entries )
422 git_exclude_entries.extend( instruction_entries )
423 git_exclude_entries.extend( all_symlink_names )
424 if git_exclude_entries: 424 ↛ exitline 424 didn't return from function '_manage_project_auxiliaries' because the condition on line 424 was always true
425 entries_count = _operations.update_git_exclude(
426 target, git_exclude_entries, simulate )
427 if entries_count > 0:
428 _scribe.info(
429 f"Managing {entries_count} entries in .git/info/exclude" )
432class PopulateProjectCommand( __.appcore_cli.Command ):
433 ''' Generates project-scoped agent content from data sources.
435 Populates agent commands, definitions, and static resources
436 from the specified data source. Copies pre-rendered commands
437 and agents from distribution/, generates skills, and copies
438 static resources.
439 '''
441 source: SourceArgument = '.'
442 target: TargetArgument = __.dcls.field( default_factory = __.Path.cwd )
443 profile: __.typx.Annotated[
444 __.typx.Optional[ __.Path ],
445 __.tyro.conf.arg(
446 help = (
447 "Alternative Copier answers file (defaults to "
448 "auto-detected)" ),
449 prefix_name = False ),
450 ] = None
451 simulate: __.typx.Annotated[
452 bool,
453 __.tyro.conf.arg(
454 help = "Dry run mode - show generated content",
455 prefix_name = False ),
456 ] = False
457 tag_prefix: __.typx.Annotated[
458 __.typx.Optional[ str ],
459 __.tyro.conf.arg(
460 help = (
461 "Prefix for version tags (e.g., 'v', 'stable-', 'prod-'); "
462 "only tags with this prefix are considered and the prefix "
463 "is stripped before version parsing" ),
464 prefix_name = False ),
465 ] = None
467 @_cmdbase.intercept_errors( )
468 async def execute( self, auxdata: __.appcore.state.Globals ) -> None: # pyright: ignore[reportIncompatibleMethodOverride]
469 ''' Generates project content from data sources. '''
470 if not isinstance( auxdata, _core.Globals ): # pragma: no cover
471 raise _exceptions.ContextInvalidity
472 _scribe.info(
473 f"Populating project content from {self.source} to {self.target}" )
474 configuration = await _cmdbase.retrieve_configuration(
475 self.target, self.profile )
476 per_project_coders = _filter_coders_by_mode(
477 configuration[ 'coders' ], 'per-project' )
478 if not per_project_coders: 478 ↛ 479line 478 didn't jump to line 479 because the condition on line 478 was never true
479 _scribe.warning(
480 "No per-project default coders found in configuration" )
481 return
482 filtered_configuration = dict( configuration )
483 filtered_configuration[ 'coders' ] = per_project_coders
484 prefix = __.absent if self.tag_prefix is None else self.tag_prefix
485 location = _cmdbase.retrieve_data_location( self.source, prefix )
486 _cmdbase.validate_data_source_structure(
487 location, ( 'per-project', ) )
488 items_attempted, items_copied, exclude_entries = (
489 _copy_distribution_items(
490 location,
491 filtered_configuration[ 'coders' ],
492 self.target,
493 configuration = filtered_configuration,
494 mode = 'per-project',
495 simulate = self.simulate ) )
496 if items_attempted > 0: 496 ↛ 503line 496 didn't jump to line 503 because the condition on line 496 was always true
497 if self.simulate: 497 ↛ 498line 497 didn't jump to line 498 because the condition on line 497 was never true
498 _scribe.info(
499 f"Would copy {items_attempted} items" )
500 else:
501 _scribe.info(
502 f"Copied {items_copied}/{items_attempted} items" )
503 _manage_project_auxiliaries(
504 filtered_configuration, location, self.target,
505 exclude_entries, self.simulate )
506 result = _results.ContentGenerationResult(
507 source_location = location,
508 target_location = self.target,
509 coders = tuple( configuration[ 'coders' ] ),
510 simulated = self.simulate,
511 items_generated = (
512 items_attempted if self.simulate else items_copied ),
513 )
514 await _core.render_and_print_result(
515 result, auxdata.display, auxdata.exits )
518class PopulateUserCommand( __.appcore_cli.Command ):
519 ''' Populates per-user global settings and executables. '''
521 source: SourceArgument = '.'
522 profile: __.typx.Annotated[
523 __.typx.Optional[ __.Path ],
524 __.tyro.conf.arg(
525 help = (
526 "Alternative Copier answers file (defaults to "
527 "auto-detected)" ),
528 prefix_name = False ),
529 ] = None
530 simulate: __.typx.Annotated[
531 bool,
532 __.tyro.conf.arg(
533 help = "Dry run mode - show what would be installed",
534 prefix_name = False ),
535 ] = False
536 tag_prefix: __.typx.Annotated[
537 __.typx.Optional[ str ],
538 __.tyro.conf.arg(
539 help = (
540 "Prefix for version tags (e.g., 'v', 'stable-', 'prod-'); "
541 "only tags with this prefix are considered and the prefix "
542 "is stripped before version parsing" ),
543 prefix_name = False ),
544 ] = None
546 @_cmdbase.intercept_errors( )
547 async def execute( self, auxdata: __.appcore.state.Globals ) -> None: # pyright: ignore[reportIncompatibleMethodOverride]
548 ''' Populates user-scoped settings and executables. '''
549 if not isinstance( auxdata, _core.Globals ): # pragma: no cover
550 raise _exceptions.ContextInvalidity
551 _scribe.info( f"Populating user configuration from {self.source}" )
552 configuration = await _cmdbase.retrieve_configuration(
553 __.Path.cwd( ), self.profile )
554 per_user_coders = _filter_coders_by_mode(
555 configuration[ 'coders' ], 'per-user' )
556 if not per_user_coders:
557 _scribe.warning(
558 "No per-user default coders found in configuration" )
559 return
560 prefix = __.absent if self.tag_prefix is None else self.tag_prefix
561 location = _cmdbase.retrieve_data_location( self.source, prefix )
562 _cmdbase.validate_data_source_structure(
563 location,
564 ( 'per-user', ) )
565 content_attempted, content_generated = _populate_per_user_content(
566 location,
567 per_user_coders,
568 configuration,
569 self.simulate,
570 )
571 if content_attempted > 0:
572 _scribe.info(
573 f"Generated {content_generated}/{content_attempted} items" )
574 globals_attempted, globals_updated = _userdata.populate_globals(
575 location,
576 per_user_coders,
577 configuration,
578 self.simulate,
579 )
580 _scribe.info(
581 f"Updated {globals_updated}/{globals_attempted} global files" )
582 wrappers_attempted, wrappers_installed = (
583 _userdata.populate_user_wrappers( location, self.simulate ) )
584 if wrappers_attempted > 0:
585 _scribe.info(
586 f"Installed {wrappers_installed}/{wrappers_attempted} "
587 "wrapper scripts" )
588 total_items = content_generated + globals_updated + wrappers_installed
589 result = _results.ContentGenerationResult(
590 source_location = location,
591 target_location = __.Path.home( ),
592 coders = per_user_coders,
593 simulated = self.simulate,
594 items_generated = total_items,
595 )
596 await _core.render_and_print_result(
597 result, auxdata.display, auxdata.exits )
600class PopulateCommand( __.appcore_cli.Command ):
601 ''' Populates agent content and configuration. '''
603 command: __.typx.Union[
604 __.typx.Annotated[
605 PopulateProjectCommand,
606 __.tyro.conf.subcommand( 'project', prefix_name = False ),
607 ],
608 __.typx.Annotated[
609 PopulateUserCommand,
610 __.tyro.conf.subcommand( 'user', prefix_name = False ),
611 ],
612 ] = __.dcls.field( default_factory = PopulateProjectCommand )
614 async def execute( self, auxdata: __.appcore.state.Globals ) -> None: # pyright: ignore[reportIncompatibleMethodOverride]
615 await self.command( auxdata )
618class GenerateCommand( __.appcore_cli.Command ):
619 ''' Generates pre-rendered artifacts from components/.
621 Two invocation shapes:
623 **Default**: ``agentsmgr generate`` reads 3-tier pipeline
624 source from ``components/`` and writes rendered commands and
625 agents to ``distribution/`` (or to ``--output PATH`` if given).
626 Skills are direct distribution artifacts and are not generated.
627 ``--check`` validates distribution/ is current without writing
628 files. ``--simulate`` shows what would be written.
630 **Answers-file**: ``agentsmgr generate --answers-file PATH
631 --output PATH`` uses the given Copier answers file as the
632 configuration and renders into the explicit ``--output``
633 target. The caller owns output allocation and cleanup;
634 production CLI does not manage temporary directories.
636 ``--check`` and ``--simulate`` are only valid in default
637 mode. ``--answers-file`` requires ``--output``.
638 '''
640 source: __.typx.Annotated[
641 str,
642 __.tyro.conf.arg(
643 help = "Components source path (defaults to 'components')",
644 prefix_name = False ),
645 ] = 'components'
646 output: __.typx.Annotated[
647 __.typx.Optional[ __.Path ],
648 __.tyro.conf.arg(
649 help = (
650 "Distribution output path. Default mode: defaults to "
651 "'distribution/'. Required when --answers-file is used." ),
652 prefix_name = False ),
653 ] = None
654 check: __.typx.Annotated[
655 bool,
656 __.tyro.conf.arg(
657 help = "Check mode - fail if distribution/ is stale",
658 prefix_name = False ),
659 ] = False
660 simulate: __.typx.Annotated[
661 bool,
662 __.tyro.conf.arg(
663 help = "Dry run mode - show what would be generated",
664 prefix_name = False ),
665 ] = False
666 answers_file: __.typx.Annotated[
667 __.typx.Optional[ __.Path ],
668 __.tyro.conf.arg(
669 help = (
670 "Answers-file mode: path to a Copier answers file to "
671 "use as the configuration source. Requires --output." ),
672 prefix_name = False ),
673 ] = None
675 @_cmdbase.intercept_errors( )
676 async def execute( self, auxdata: __.appcore.state.Globals ) -> None: # pyright: ignore[reportIncompatibleMethodOverride]
677 ''' Generates distribution artifacts from components. '''
678 if not isinstance( auxdata, _core.Globals ): # pragma: no cover
679 raise _exceptions.ContextInvalidity
680 if self.answers_file is not None and self.check:
681 raise _exceptions.ConfigurationInvalidity(
682 reason = '--check is not valid with --answers-file' )
683 if self.answers_file is not None and self.simulate:
684 raise _exceptions.ConfigurationInvalidity(
685 reason = '--simulate is not valid with --answers-file' )
686 if self.answers_file is not None and self.output is None:
687 raise _exceptions.ConfigurationInvalidity(
688 reason = '--answers-file requires --output' )
689 location = _cmdbase.retrieve_data_location( self.source )
690 _cmdbase.validate_data_source_structure(
691 location,
692 ( 'configurations', 'contents', 'templates' ) )
693 if self.answers_file is not None:
694 await self._execute_answers_file_mode(
695 auxdata, location )
696 return
697 await self._execute_default_mode( auxdata, location )
699 async def _execute_default_mode(
700 self,
701 auxdata: _core.Globals,
702 location: __.Path,
703 ) -> None:
704 ''' Default mode: renders distribution tree with universal config. '''
705 target = (
706 self.output if self.output is not None
707 else __.Path( 'distribution' ) )
708 _scribe.info(
709 f"Generating distribution from {self.source} to {target}" )
710 configuration = _produce_default_configuration( location )
711 generator = _generator.ContentGenerator(
712 location = location,
713 configuration = configuration,
714 application_configuration = auxdata.configuration,
715 mode = 'per-project',
716 )
717 if self.check: 717 ↛ 718line 717 didn't jump to line 718 because the condition on line 717 was never true
718 items_checked, diff_lines = (
719 _operations.check_distribution_staleness(
720 generator, target ) )
721 if diff_lines:
722 _scribe.error(
723 f"Distribution is stale ({items_checked} items checked):" )
724 for line in diff_lines:
725 print( line )
726 raise SystemExit( 1 )
727 _scribe.info(
728 f"Distribution is current ({items_checked} items checked)" )
729 return
730 items_attempted, items_generated = (
731 _operations.generate_distribution(
732 generator, target, self.simulate ) )
733 _scribe.info(
734 f"Generated {items_generated}/{items_attempted} artifacts" )
735 result = _results.ContentGenerationResult(
736 source_location = location,
737 target_location = target,
738 coders = tuple( configuration.get( 'coders', ( ) ) ),
739 simulated = self.simulate,
740 items_generated = items_generated,
741 )
742 await _core.render_and_print_result(
743 result, auxdata.display, auxdata.exits )
745 async def _execute_answers_file_mode(
746 self,
747 auxdata: _core.Globals,
748 location: __.Path,
749 ) -> None:
750 ''' Answers-file mode: renders against an explicit answers file
751 into the explicit --output target. Caller owns the target
752 directory and its cleanup.
753 '''
754 answers_file = __.typx.cast( __.Path, self.answers_file )
755 target = __.typx.cast( __.Path, self.output )
756 _scribe.info(
757 f"Generating distribution from {self.source} against "
758 f"{answers_file} to {target}" )
759 configuration = await _cmdbase.retrieve_configuration(
760 target = __.Path.cwd( ), profile = answers_file )
761 generator = _generator.ContentGenerator(
762 location = location, configuration = configuration )
763 items_attempted, items_generated = (
764 _operations.populate_directory(
765 generator, target, simulate = False ) )
766 _scribe.info(
767 f"Generated {items_generated}/{items_attempted} artifacts" )
768 result = _results.ContentGenerationResult(
769 source_location = location,
770 target_location = target,
771 coders = tuple( configuration.get( 'coders', ( ) ) ),
772 simulated = False,
773 items_generated = items_generated,
774 )
775 await _core.render_and_print_result(
776 result, auxdata.display, auxdata.exits )