Coverage for sources/agentsmgr/population.py: 25%

1# vim: set filetype=python fileencoding=utf-8: 

2# -*- coding: utf-8 -*- 

3 

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#============================================================================# 

19 

20 

21''' Command for populating agent content from data sources. ''' 

22 

23 

24from . import __ 

25from . import cmdbase as _cmdbase 

26from . import core as _core 

27from . import exceptions as _exceptions 

28from . import generator as _generator 

29from . import memorylinks as _memorylinks 

30from . import operations as _operations 

31from . import renderers as _renderers 

32from . import results as _results 

33from . import userdata as _userdata 

34 

35 

36_scribe = __.provide_scribe( __name__ ) 

37 

38 

39SourceArgument: __.typx.TypeAlias = __.typx.Annotated[ 

40 __.tyro.conf.Positional[ str ], 

41 __.tyro.conf.arg( help = "Data source (local path or git URL)" ), 

42] 

43TargetArgument: __.typx.TypeAlias = __.typx.Annotated[ 

44 __.tyro.conf.Positional[ __.Path ], 

45 __.tyro.conf.arg( help = "Target directory for content generation" ), 

46] 

47 

48 

49def _create_all_symlinks( 

50 configuration: __.cabc.Mapping[ str, __.typx.Any ], 

51 target: __.Path, 

52 mode: str, 

53 simulate: bool, 

54) -> tuple[ str, ... ]: 

55 ''' Creates all symlinks and returns their names for git exclude. 

56 

57 Creates memory symlinks for all coders and coder directory 

58 symlinks for per-project mode. Returns tuple of all created 

59 symlink names. 

60 ''' 

61 all_symlink_names: list[ str ] = [ ] 

62 if mode == 'nowhere': return tuple( all_symlink_names ) 

63 links_attempted, links_created, symlink_names = ( 

64 _memorylinks.create_memory_symlinks_for_coders( 

65 coders = configuration[ 'coders' ], 

66 target = target, 

67 renderers = _renderers.RENDERERS, 

68 simulate = simulate, 

69 ) ) 

70 all_symlink_names.extend( symlink_names ) 

71 if links_created > 0: 

72 _scribe.info( 

73 f"Created {links_created}/{links_attempted} memory symlinks" ) 

74 if mode == 'per-project': 

75 ( coder_symlinks_attempted, 

76 coder_symlinks_created, 

77 coder_symlink_names ) = ( 

78 _create_coder_directory_symlinks( 

79 coders = configuration[ 'coders' ], 

80 target = target, 

81 renderers = _renderers.RENDERERS, 

82 simulate = simulate, 

83 ) ) 

84 all_symlink_names.extend( coder_symlink_names ) 

85 if coder_symlinks_created > 0: 

86 _scribe.info( 

87 f"Created {coder_symlinks_created}/" 

88 f"{coder_symlinks_attempted} coder directory symlinks" ) 

89 return tuple( all_symlink_names ) 

90 

91 

92def _create_coder_directory_symlinks( 

93 coders: __.cabc.Sequence[ str ], 

94 target: __.Path, 

95 renderers: __.cabc.Mapping[ str, __.typx.Any ], 

96 simulate: bool = False, 

97) -> tuple[ int, int, tuple[ str, ... ] ]: 

98 ''' Creates symlinks from .{coder} to .auxiliary/configuration/coders/. 

99 

100 For per-project mode, creates symlinks that make coder directories 

101 accessible at their expected locations (.claude, .opencode, etc.) 

102 while keeping actual files organized under 

103 .auxiliary/configuration/coders/. 

104 

105 Returns tuple of (attempted, created, symlink_names) where 

106 symlink_names contains names of all created symlinks. 

107 ''' 

108 attempted = 0 

109 created = 0 

110 symlink_names: list[ str ] = [ ] 

111 for coder_name in coders: 

112 try: renderers[ coder_name ] 

113 except KeyError as exception: 

114 raise _exceptions.CoderAbsence( coder_name ) from exception 

115 

116 # Source: actual location under .auxiliary/configuration/coders/ 

117 source = ( 

118 target / '.auxiliary' / 'configuration' / 'coders' / coder_name ) 

119 # Link: expected location for coder (.claude, .opencode, etc.) 

120 link_path = target / f'.{coder_name}' 

121 

122 attempted += 1 

123 was_created, symlink_name = _memorylinks.create_memory_symlink( 

124 source, link_path, simulate ) 

125 if was_created: 

126 created += 1 

127 symlink_names.append( symlink_name ) 

128 

129 # Create .mcp.json symlink for Claude coder specifically 

130 if coder_name == 'claude': 

131 mcp_source = ( 

132 target / '.auxiliary' / 'configuration' / 'mcp-servers.json' ) 

133 mcp_link = target / '.mcp.json' 

134 attempted += 1 

135 was_created, symlink_name = _memorylinks.create_memory_symlink( 

136 mcp_source, mcp_link, simulate ) 

137 if was_created: 

138 created += 1 

139 symlink_names.append( symlink_name ) 

140 

141 return ( attempted, created, tuple( symlink_names ) ) 

142 

143 

144class PopulateCommand( __.appcore_cli.Command ): 

145 ''' Generates dynamic agent content from data sources. ''' 

146 

147 source: SourceArgument = '.' 

148 target: TargetArgument = __.dcls.field( default_factory = __.Path.cwd ) 

149 profile: __.typx.Annotated[ 

150 __.typx.Optional[ __.Path ], 

151 __.tyro.conf.arg( 

152 help = ( 

153 "Alternative Copier answers file (defaults to " 

154 "auto-detected)" ), 

155 prefix_name = False ), 

156 ] = None 

157 simulate: __.typx.Annotated[ 

158 bool, 

159 __.tyro.conf.arg( 

160 help = "Dry run mode - show generated content", 

161 prefix_name = False ), 

162 ] = False 

163 mode: __.typx.Annotated[ 

164 _renderers.TargetMode, 

165 __.tyro.conf.arg( 

166 help = ( 

167 "Targeting mode: default (use coder defaults), per-user, " 

168 "per-project, or nowhere (skip generation)" ), 

169 prefix_name = False ), 

170 ] = 'default' 

171 update_globals: __.typx.Annotated[ 

172 bool, 

173 __.tyro.conf.arg( 

174 help = "Update per-user global files (orthogonal to mode)", 

175 prefix_name = False ), 

176 ] = False 

177 tag_prefix: __.typx.Annotated[ 

178 __.typx.Optional[ str ], 

179 __.tyro.conf.arg( 

180 help = ( 

181 "Prefix for version tags (e.g., 'v', 'stable-', 'prod-'); " 

182 "only tags with this prefix are considered and the prefix " 

183 "is stripped before version parsing" ), 

184 prefix_name = False ), 

185 ] = None 

186 

187 @_cmdbase.intercept_errors( ) 

188 async def execute( self, auxdata: __.appcore.state.Globals ) -> None: # pyright: ignore[reportIncompatibleMethodOverride] 

189 ''' Generates content from data sources and displays result. ''' 

190 if not isinstance( auxdata, _core.Globals ): # pragma: no cover 

191 raise _exceptions.ContextInvalidity 

192 _scribe.info( 

193 f"Populating agent content from {self.source} to {self.target}" ) 

194 configuration = await _cmdbase.retrieve_configuration( 

195 self.target, self.profile ) 

196 coder_count = len( configuration[ 'coders' ] ) 

197 _scribe.debug( f"Detected configuration with {coder_count} coders" ) 

198 _scribe.debug( f"Using {self.mode} targeting mode" ) 

199 prefix = ( 

200 __.absent if self.tag_prefix is None 

201 else self.tag_prefix ) 

202 location = _cmdbase.retrieve_data_location( self.source, prefix ) 

203 generator = _generator.ContentGenerator( 

204 location = location, 

205 configuration = configuration, 

206 application_configuration = auxdata.configuration, 

207 mode = self.mode, 

208 ) 

209 items_attempted, items_generated = _operations.populate_directory( 

210 generator, self.target, self.simulate ) 

211 _scribe.info( f"Generated {items_generated}/{items_attempted} items" ) 

212 all_symlink_names = _create_all_symlinks( 

213 configuration, self.target, self.mode, self.simulate ) 

214 if self.update_globals: 

215 globals_attempted, globals_updated = ( 

216 _userdata.populate_globals( 

217 location, 

218 configuration[ 'coders' ], 

219 auxdata.configuration, 

220 self.simulate, 

221 ) ) 

222 _scribe.info( 

223 f"Updated {globals_updated}/{globals_attempted} " 

224 "global files" ) 

225 if all_symlink_names: 

226 excludes_added = _operations.update_git_exclude( 

227 self.target, all_symlink_names, self.simulate ) 

228 if excludes_added > 0: 

229 _scribe.info( 

230 f"Added {excludes_added} symlink names to " 

231 ".git/info/exclude" ) 

232 result = _results.ContentGenerationResult( 

233 source_location = location, 

234 target_location = self.target, 

235 coders = tuple( configuration[ 'coders' ] ), 

236 simulated = self.simulate, 

237 items_generated = items_generated, 

238 ) 

239 await _core.render_and_print_result( 

240 result, auxdata.display, auxdata.exits )