Coverage for sources/agentsmgr/commands/population.py: 22%

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 base as _base 

26from . import generator as _generator 

27from . import memorylinks as _memorylinks 

28from . import operations as _operations 

29from . import userdata as _userdata 

30 

31 

32_scribe = __.provide_scribe( __name__ ) 

33 

34 

35def _create_coder_directory_symlinks( 

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

37 target: __.Path, 

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

39 simulate: bool = False, 

40) -> tuple[ int, int ]: 

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

42 

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

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

45 while keeping actual files organized under 

46 .auxiliary/configuration/coders/. 

47 

48 Returns tuple of (attempted, created) counts. 

49 ''' 

50 attempted = 0 

51 created = 0 

52 for coder_name in coders: 

53 try: renderers[ coder_name ] 

54 except KeyError as exception: 

55 raise __.CoderAbsence( coder_name ) from exception 

56 

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

58 source = ( 

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

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

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

62 

63 attempted += 1 

64 if _memorylinks.create_memory_symlink( source, link_path, simulate ): 

65 created += 1 

66 

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

68 if coder_name == 'claude': 

69 mcp_source = ( 

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

71 mcp_link = target / '.mcp.json' 

72 attempted += 1 

73 if _memorylinks.create_memory_symlink( 

74 mcp_source, mcp_link, simulate ): 

75 created += 1 

76 

77 return ( attempted, created ) 

78 

79 

80class PopulateCommand( __.appcore_cli.Command ): 

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

82 

83 source: __.typx.Annotated[ 

84 str, 

85 __.tyro.conf.arg( 

86 help = "Data source (local path or git URL)", 

87 prefix_name = False ), 

88 ] = '.' 

89 target: __.typx.Annotated[ 

90 __.Path, 

91 __.tyro.conf.arg( 

92 help = "Target directory for content generation", 

93 prefix_name = False ), 

94 ] = __.dcls.field( default_factory = __.Path.cwd ) 

95 simulate: __.typx.Annotated[ 

96 bool, 

97 __.tyro.conf.arg( 

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

99 prefix_name = False ), 

100 ] = False 

101 mode: __.typx.Annotated[ 

102 __.TargetMode, 

103 __.tyro.conf.arg( 

104 help = ( 

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

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

107 prefix_name = False ), 

108 ] = 'default' 

109 update_globals: __.typx.Annotated[ 

110 bool, 

111 __.tyro.conf.arg( 

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

113 prefix_name = False ), 

114 ] = False 

115 

116 @_base.intercept_errors( ) 

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

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

119 if not isinstance( auxdata, __.Globals ): # pragma: no cover 

120 raise __.ContextInvalidity 

121 _scribe.info( 

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

123 configuration = await _base.retrieve_configuration( self.target ) 

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

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

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

127 location = _base.retrieve_data_location( self.source ) 

128 generator = _generator.ContentGenerator( 

129 location = location, 

130 configuration = configuration, 

131 application_configuration = auxdata.configuration, 

132 mode = self.mode, 

133 ) 

134 items_attempted, items_generated = _operations.populate_directory( 

135 generator, self.target, self.simulate ) 

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

137 if self.mode != 'nowhere': 

138 links_attempted, links_created = ( 

139 _memorylinks.create_memory_symlinks_for_coders( 

140 coders = configuration[ 'coders' ], 

141 target = self.target, 

142 renderers = __.RENDERERS, 

143 simulate = self.simulate, 

144 ) ) 

145 if links_created > 0: 

146 _scribe.info( 

147 f"Created {links_created}/{links_attempted} " 

148 "memory symlinks" ) 

149 # Create coder directory symlinks for per-project mode 

150 if self.mode == 'per-project': 

151 coder_symlinks_attempted, coder_symlinks_created = ( 

152 _create_coder_directory_symlinks( 

153 coders = configuration[ 'coders' ], 

154 target = self.target, 

155 renderers = __.RENDERERS, 

156 simulate = self.simulate, 

157 ) ) 

158 if coder_symlinks_created > 0: 

159 _scribe.info( 

160 f"Created {coder_symlinks_created}/" 

161 f"{coder_symlinks_attempted} coder directory symlinks") 

162 if self.update_globals: 

163 globals_attempted, globals_updated = ( 

164 _userdata.populate_globals( 

165 location, 

166 configuration[ 'coders' ], 

167 auxdata.configuration, 

168 self.simulate, 

169 ) ) 

170 _scribe.info( 

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

172 "global files" ) 

173 result = __.ContentGenerationResult( 

174 source_location = location, 

175 target_location = self.target, 

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

177 simulated = self.simulate, 

178 items_generated = items_generated, 

179 ) 

180 await __.render_and_print_result( 

181 result, auxdata.display, auxdata.exits )