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

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 symlink 

59 names (both newly created and pre-existing) for git exclude 

60 update. 

61 ''' 

62 all_symlink_names: list[ str ] = [ ] 

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

64 links_attempted, links_created, symlink_names = ( 

65 _memorylinks.create_memory_symlinks_for_coders( 

66 coders = configuration[ 'coders' ], 

67 target = target, 

68 renderers = _renderers.RENDERERS, 

69 simulate = simulate, 

70 ) ) 

71 all_symlink_names.extend( symlink_names ) 

72 if links_created > 0: 

73 _scribe.info( 

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

75 needs_coder_symlinks = ( 

76 mode == 'per-project' 

77 or ( mode == 'default' and any( 

78 _renderers.RENDERERS[ coder ].mode_default == 'per-project' 

79 for coder in configuration[ 'coders' ] ) ) ) 

80 if needs_coder_symlinks: 

81 ( coder_symlinks_attempted, 

82 coder_symlinks_created, 

83 coder_symlink_names ) = ( 

84 _create_coder_directory_symlinks( 

85 coders = configuration[ 'coders' ], 

86 target = target, 

87 renderers = _renderers.RENDERERS, 

88 simulate = simulate, 

89 ) ) 

90 all_symlink_names.extend( coder_symlink_names ) 

91 if coder_symlinks_created > 0: 

92 _scribe.info( 

93 f"Created {coder_symlinks_created}/" 

94 f"{coder_symlinks_attempted} coder directory symlinks" ) 

95 return tuple( all_symlink_names ) 

96 

97 

98def _create_coder_directory_symlinks( 

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

100 target: __.Path, 

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

102 simulate: bool = False, 

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

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

105 

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

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

108 while keeping actual files organized under 

109 .auxiliary/configuration/coders/. 

110 

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

112 symlink_names contains names of all symlinks (both newly created 

113 and pre-existing). 

114 ''' 

115 attempted = 0 

116 created = 0 

117 symlink_names: list[ str ] = [ ] 

118 for coder_name in coders: 

119 try: renderers[ coder_name ] 

120 except KeyError as exception: 

121 raise _exceptions.CoderAbsence( coder_name ) from exception 

122 

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

124 source = ( 

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

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

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

128 

129 attempted += 1 

130 was_created, symlink_name = _memorylinks.create_memory_symlink( 

131 source, link_path, simulate ) 

132 if was_created: created += 1 

133 symlink_names.append( symlink_name ) 

134 

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

136 if coder_name == 'claude': 

137 mcp_source = ( 

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

139 mcp_link = target / '.mcp.json' 

140 attempted += 1 

141 was_created, symlink_name = _memorylinks.create_memory_symlink( 

142 mcp_source, mcp_link, simulate ) 

143 if was_created: created += 1 

144 symlink_names.append( symlink_name ) 

145 

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

147 

148 

149class PopulateCommand( __.appcore_cli.Command ): 

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

151 

152 source: SourceArgument = '.' 

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

154 profile: __.typx.Annotated[ 

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

156 __.tyro.conf.arg( 

157 help = ( 

158 "Alternative Copier answers file (defaults to " 

159 "auto-detected)" ), 

160 prefix_name = False ), 

161 ] = None 

162 simulate: __.typx.Annotated[ 

163 bool, 

164 __.tyro.conf.arg( 

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

166 prefix_name = False ), 

167 ] = False 

168 mode: __.typx.Annotated[ 

169 _renderers.TargetMode, 

170 __.tyro.conf.arg( 

171 help = ( 

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

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

174 prefix_name = False ), 

175 ] = 'default' 

176 update_globals: __.typx.Annotated[ 

177 bool, 

178 __.tyro.conf.arg( 

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

180 prefix_name = False ), 

181 ] = False 

182 tag_prefix: __.typx.Annotated[ 

183 __.typx.Optional[ str ], 

184 __.tyro.conf.arg( 

185 help = ( 

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

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

188 "is stripped before version parsing" ), 

189 prefix_name = False ), 

190 ] = None 

191 

192 @_cmdbase.intercept_errors( ) 

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

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

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

196 raise _exceptions.ContextInvalidity 

197 _scribe.info( 

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

199 configuration = await _cmdbase.retrieve_configuration( 

200 self.target, self.profile ) 

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

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

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

204 prefix = ( 

205 __.absent if self.tag_prefix is None 

206 else self.tag_prefix ) 

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

208 generator = _generator.ContentGenerator( 

209 location = location, 

210 configuration = configuration, 

211 application_configuration = auxdata.configuration, 

212 mode = self.mode, 

213 ) 

214 items_attempted, items_generated = _operations.populate_directory( 

215 generator, self.target, self.simulate ) 

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

217 all_symlink_names = _create_all_symlinks( 

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

219 if self.update_globals: 

220 globals_attempted, globals_updated = ( 

221 _userdata.populate_globals( 

222 location, 

223 configuration[ 'coders' ], 

224 auxdata.configuration, 

225 self.simulate, 

226 ) ) 

227 _scribe.info( 

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

229 "global files" ) 

230 if all_symlink_names: 

231 excludes_added = _operations.update_git_exclude( 

232 self.target, all_symlink_names, self.simulate ) 

233 if excludes_added > 0: 

234 _scribe.info( 

235 f"Added {excludes_added} symlink names to " 

236 ".git/info/exclude" ) 

237 result = _results.ContentGenerationResult( 

238 source_location = location, 

239 target_location = self.target, 

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

241 simulated = self.simulate, 

242 items_generated = items_generated, 

243 ) 

244 await _core.render_and_print_result( 

245 result, auxdata.display, auxdata.exits )