Coverage for sources/agentsmgr/population.py: 23%
95 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-10-24 01:49 +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 _create_all_symlinks(
51 configuration: __.cabc.Mapping[ str, __.typx.Any ],
52 target: __.Path,
53 mode: str,
54 simulate: bool,
55) -> tuple[ str, ... ]:
56 ''' Creates all symlinks and returns their names for git exclude.
58 Creates memory symlinks for all coders and coder directory
59 symlinks for per-project mode. Returns tuple of all symlink
60 names (both newly created and pre-existing) for git exclude
61 update.
62 '''
63 all_symlink_names: list[ str ] = [ ]
64 if mode == 'nowhere': return tuple( all_symlink_names )
65 links_attempted, links_created, symlink_names = (
66 _memorylinks.create_memory_symlinks_for_coders(
67 coders = configuration[ 'coders' ],
68 target = target,
69 renderers = _renderers.RENDERERS,
70 simulate = simulate,
71 ) )
72 all_symlink_names.extend( symlink_names )
73 if links_created > 0:
74 _scribe.info(
75 f"Created {links_created}/{links_attempted} memory symlinks" )
76 needs_coder_symlinks = (
77 mode == 'per-project'
78 or ( mode == 'default' and any(
79 _renderers.RENDERERS[ coder ].mode_default == 'per-project'
80 for coder in configuration[ 'coders' ] ) ) )
81 if needs_coder_symlinks:
82 ( coder_symlinks_attempted,
83 coder_symlinks_created,
84 coder_symlink_names ) = (
85 _create_coder_directory_symlinks(
86 coders = configuration[ 'coders' ],
87 target = target,
88 renderers = _renderers.RENDERERS,
89 simulate = simulate,
90 ) )
91 all_symlink_names.extend( coder_symlink_names )
92 if coder_symlinks_created > 0:
93 _scribe.info(
94 f"Created {coder_symlinks_created}/"
95 f"{coder_symlinks_attempted} coder directory symlinks" )
96 return tuple( all_symlink_names )
99def _populate_instructions_if_configured(
100 configuration: __.cabc.Mapping[ str, __.typx.Any ],
101 target: __.Path,
102 tag_prefix: __.Absential[ str ],
103 simulate: bool,
104) -> tuple[ bool, str ]:
105 ''' Populates instructions if configured and returns status.
107 Returns tuple of (sources_present, instructions_target_path).
108 sources_present indicates whether instruction sources were
109 configured and processed.
110 '''
111 if not configuration.get( 'provide_instructions', False ):
112 return ( False, '' )
113 instructions_sources = configuration.get( 'instructions_sources', [ ] )
114 instructions_target = configuration.get(
115 'instructions_target', '.auxiliary/instructions' )
116 if not instructions_sources:
117 return ( False, instructions_target )
118 instructions_attempted, instructions_updated = (
119 _instructions.populate_instructions(
120 instructions_sources,
121 target / instructions_target,
122 tag_prefix,
123 simulate,
124 ) )
125 _scribe.info(
126 f"Updated {instructions_updated}/"
127 f"{instructions_attempted} instruction files" )
128 return ( True, instructions_target )
131def _create_coder_directory_symlinks(
132 coders: __.cabc.Sequence[ str ],
133 target: __.Path,
134 renderers: __.cabc.Mapping[ str, __.typx.Any ],
135 simulate: bool = False,
136) -> tuple[ int, int, tuple[ str, ... ] ]:
137 ''' Creates symlinks from .{coder} to .auxiliary/configuration/coders/.
139 For per-project mode, creates symlinks that make coder directories
140 accessible at their expected locations (.claude, .opencode, etc.)
141 while keeping actual files organized under
142 .auxiliary/configuration/coders/.
144 Returns tuple of (attempted, created, symlink_names) where
145 symlink_names contains names of all symlinks (both newly created
146 and pre-existing).
147 '''
148 attempted = 0
149 created = 0
150 symlink_names: list[ str ] = [ ]
151 for coder_name in coders:
152 try: renderers[ coder_name ]
153 except KeyError as exception:
154 raise _exceptions.CoderAbsence( coder_name ) from exception
156 # Source: actual location under .auxiliary/configuration/coders/
157 source = (
158 target / '.auxiliary' / 'configuration' / 'coders' / coder_name )
159 # Link: expected location for coder (.claude, .opencode, etc.)
160 link_path = target / f'.{coder_name}'
162 attempted += 1
163 was_created, symlink_name = _memorylinks.create_memory_symlink(
164 source, link_path, simulate )
165 if was_created: created += 1
166 symlink_names.append( symlink_name )
168 # Create .mcp.json symlink for Claude coder specifically
169 if coder_name == 'claude':
170 mcp_source = (
171 target / '.auxiliary' / 'configuration' / 'mcp-servers.json' )
172 mcp_link = target / '.mcp.json'
173 attempted += 1
174 was_created, symlink_name = _memorylinks.create_memory_symlink(
175 mcp_source, mcp_link, simulate )
176 if was_created: created += 1
177 symlink_names.append( symlink_name )
179 return ( attempted, created, tuple( symlink_names ) )
182class PopulateCommand( __.appcore_cli.Command ):
183 ''' Generates dynamic agent content from data sources. '''
185 source: SourceArgument = '.'
186 target: TargetArgument = __.dcls.field( default_factory = __.Path.cwd )
187 profile: __.typx.Annotated[
188 __.typx.Optional[ __.Path ],
189 __.tyro.conf.arg(
190 help = (
191 "Alternative Copier answers file (defaults to "
192 "auto-detected)" ),
193 prefix_name = False ),
194 ] = None
195 simulate: __.typx.Annotated[
196 bool,
197 __.tyro.conf.arg(
198 help = "Dry run mode - show generated content",
199 prefix_name = False ),
200 ] = False
201 mode: __.typx.Annotated[
202 _renderers.TargetMode,
203 __.tyro.conf.arg(
204 help = (
205 "Targeting mode: default (use coder defaults), per-user, "
206 "per-project, or nowhere (skip generation)" ),
207 prefix_name = False ),
208 ] = 'default'
209 update_globals: __.typx.Annotated[
210 bool,
211 __.tyro.conf.arg(
212 help = "Update per-user global files (orthogonal to mode)",
213 prefix_name = False ),
214 ] = False
215 tag_prefix: __.typx.Annotated[
216 __.typx.Optional[ str ],
217 __.tyro.conf.arg(
218 help = (
219 "Prefix for version tags (e.g., 'v', 'stable-', 'prod-'); "
220 "only tags with this prefix are considered and the prefix "
221 "is stripped before version parsing" ),
222 prefix_name = False ),
223 ] = None
225 @_cmdbase.intercept_errors( )
226 async def execute( self, auxdata: __.appcore.state.Globals ) -> None: # pyright: ignore[reportIncompatibleMethodOverride]
227 ''' Generates content from data sources and displays result. '''
228 if not isinstance( auxdata, _core.Globals ): # pragma: no cover
229 raise _exceptions.ContextInvalidity
230 _scribe.info(
231 f"Populating agent content from {self.source} to {self.target}" )
232 configuration = await _cmdbase.retrieve_configuration(
233 self.target, self.profile )
234 coder_count = len( configuration[ 'coders' ] )
235 _scribe.debug( f"Detected configuration with {coder_count} coders" )
236 _scribe.debug( f"Using {self.mode} targeting mode" )
237 prefix = (
238 __.absent if self.tag_prefix is None
239 else self.tag_prefix )
240 location = _cmdbase.retrieve_data_location( self.source, prefix )
241 generator = _generator.ContentGenerator(
242 location = location,
243 configuration = configuration,
244 application_configuration = auxdata.configuration,
245 mode = self.mode,
246 )
247 items_attempted, items_generated = _operations.populate_directory(
248 generator, self.target, self.simulate )
249 _scribe.info( f"Generated {items_generated}/{items_attempted} items" )
250 instructions_populated, instructions_target = (
251 _populate_instructions_if_configured(
252 configuration, self.target, prefix, self.simulate ) )
253 all_symlink_names = _create_all_symlinks(
254 configuration, self.target, self.mode, self.simulate )
255 git_exclude_entries: list[ str ] = [ ]
256 if instructions_populated:
257 git_exclude_entries.append( instructions_target )
258 git_exclude_entries.extend( all_symlink_names )
259 if self.update_globals:
260 globals_attempted, globals_updated = (
261 _userdata.populate_globals(
262 location,
263 configuration[ 'coders' ],
264 auxdata.configuration,
265 self.simulate,
266 ) )
267 _scribe.info(
268 f"Updated {globals_updated}/{globals_attempted} "
269 "global files" )
270 if git_exclude_entries:
271 excludes_added = _operations.update_git_exclude(
272 self.target, git_exclude_entries, self.simulate )
273 if excludes_added > 0:
274 _scribe.info(
275 f"Added {excludes_added} entries to .git/info/exclude" )
276 result = _results.ContentGenerationResult(
277 source_location = location,
278 target_location = self.target,
279 coders = tuple( configuration[ 'coders' ] ),
280 simulated = self.simulate,
281 items_generated = items_generated,
282 )
283 await _core.render_and_print_result(
284 result, auxdata.display, auxdata.exits )