Coverage for sources / agentsmgr / generator.py: 62%

146 statements  

« prev     ^ index     » next       coverage.py v7.13.3, created at 2026-02-04 21:55 +0000

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''' Content generation for coder-specific templates and items. 

22 

23 This module implements the ContentGenerator class which handles 

24 template-based content generation from structured data sources, 

25 including content fallback logic for compatible coders. 

26''' 

27 

28 

29import jinja2 as _jinja2 

30 

31from . import __ 

32from . import cmdbase as _cmdbase 

33from . import context as _context 

34from . import exceptions as _exceptions 

35from . import renderers as _renderers 

36 

37 

38CoderFallbackMap: __.typx.TypeAlias = __.immut.Dictionary[ str, str ] 

39PluralMappings: __.typx.TypeAlias = __.immut.Dictionary[ str, str ] 

40 

41_TEMPLATE_PARTS_MINIMUM = 3 

42 

43_PLURAL_TO_SINGULAR_MAP: PluralMappings = __.immut.Dictionary( { 

44 'commands': 'command', 

45 'agents': 'agent', 

46} ) 

47 

48 

49_scribe = __.provide_scribe( __name__ ) 

50 

51 

52class RenderedItem( __.immut.DataclassObject ): 

53 ''' Single rendered item with location and content. ''' 

54 

55 content: str 

56 location: __.Path 

57 

58 

59class ItemRenderRequest( __.immut.DataclassObject ): 

60 item_type: str 

61 item_name: str 

62 template_name: str 

63 metadata: dict[ str, __.typx.Any ] 

64 

65 

66class ContentGenerator( __.immut.DataclassObject ): 

67 ''' Generates coder-specific content from data sources. 

68 

69 Provides template-based content generation with intelligent 

70 fallback logic for compatible coders (Claude ↔ OpenCode). 

71 Supports configurable targeting modes (per-user or per-project). 

72 ''' 

73 

74 location: __.Path 

75 configuration: _cmdbase.CoderConfiguration 

76 application_configuration: __.cabc.Mapping[ str, __.typx.Any ] = ( 

77 __.dcls.field( 

78 default_factory = __.immut.Dictionary[ str, __.typx.Any ] ) ) 

79 mode: _renderers.TargetMode = 'per-project' 

80 jinja_environment: _jinja2.Environment = __.dcls.field( init = False ) 

81 

82 def __post_init__( self ) -> None: 

83 self.jinja_environment = ( # pyright: ignore[reportAttributeAccessIssue] 

84 self._produce_jinja_environment( ) ) 

85 

86 

87 def _retrieve_fallback_mappings( self ) -> CoderFallbackMap: 

88 ''' Retrieves coder fallback mappings from configuration. ''' 

89 content_config = self.application_configuration.get( 'content', { } ) 

90 fallbacks = content_config.get( 'fallbacks', { } ) 

91 return __.immut.Dictionary( fallbacks ) 

92 

93 def _resolve_renderer( self, coder: str ) -> _renderers.RendererBase: 

94 try: return _renderers.RENDERERS[ coder ] 

95 except KeyError as exception: 

96 raise _exceptions.CoderAbsence( coder ) from exception 

97 

98 def _resolve_actual_mode( 

99 self, 

100 renderer: _renderers.RendererBase, 

101 coder: str, 

102 ) -> _renderers.ExplicitTargetMode: 

103 if self.mode == 'default': return renderer.mode_default 103 ↛ exitline 103 didn't return from function '_resolve_actual_mode' because the return on line 103 wasn't executed

104 if self.mode in ( 'per-user', 'per-project' ): 104 ↛ 108line 104 didn't jump to line 108 because the condition on line 104 was always true

105 actual_mode: _renderers.ExplicitTargetMode = self.mode 

106 renderer.validate_mode( actual_mode ) 

107 return actual_mode 

108 raise _exceptions.TargetModeNoSupport( coder, self.mode ) 

109 

110 def _render_content( 

111 self, 

112 template_name: str, 

113 body: str, 

114 metadata: dict[ str, __.typx.Any ], 

115 ) -> str: 

116 template = self.jinja_environment.get_template( template_name ) 

117 normalized = _context.normalize_render_context( 

118 metadata[ 'context' ], metadata[ 'coder' ] ) 

119 variables: dict[ str, __.typx.Any ] = { 'content': body, **normalized } 

120 return template.render( **variables ) 

121 

122 def _produce_item_location( 

123 self, 

124 renderer: _renderers.RendererBase, 

125 actual_mode: _renderers.ExplicitTargetMode, 

126 target: __.Path, 

127 request: ItemRenderRequest, 

128 ) -> __.Path: 

129 base_directory = renderer.resolve_base_directory( 

130 mode = actual_mode, 

131 target = target, 

132 configuration = self.application_configuration, 

133 environment = __.os.environ, 

134 ) 

135 extension = self._parse_template_extension( request.template_name ) 

136 if request.item_type == 'skills': 136 ↛ 142line 136 didn't jump to line 142 because the condition on line 136 was always true

137 dirname = renderer.produce_output_structure( request.item_type ) 

138 return ( 

139 base_directory / dirname / request.item_name / 

140 f"SKILL.{extension}" 

141 ) 

142 category = request.metadata[ 'context' ].get( 'category' ) 

143 if category is None: category = __.absent 

144 dirname = renderer.produce_output_structure( 

145 request.item_type, category ) 

146 return base_directory / dirname / f"{request.item_name}.{extension}" 

147 

148 def render_single_item( 

149 self, item_type: str, item_name: str, coder: str, target: __.Path 

150 ) -> RenderedItem: 

151 ''' Renders a single item for a coder. 

152 

153 Combines TOML metadata, content body, and template to produce 

154 final coder-specific file. Returns RenderedItem with content 

155 and location. 

156 ''' 

157 renderer = self._resolve_renderer( coder ) 

158 actual_mode = self._resolve_actual_mode( renderer, coder ) 

159 body = self._retrieve_content_with_fallback( 

160 item_type, item_name, coder ) 

161 metadata = self._load_item_metadata( item_type, item_name, coder ) 

162 template_name = self._select_template_for_coder( item_type, coder ) 

163 request = ItemRenderRequest( 

164 item_type = item_type, 

165 item_name = item_name, 

166 template_name = template_name, 

167 metadata = metadata, 

168 ) 

169 content = self._render_content( template_name, body, metadata ) 

170 location = self._produce_item_location( 

171 renderer, 

172 actual_mode, 

173 target, 

174 request, 

175 ) 

176 return RenderedItem( content = content, location = location ) 

177 

178 def _survey_available_templates( 

179 self, item_type: str, coder: str 

180 ) -> list[ str ]: 

181 directory = self.location / "templates" 

182 # Validate coder exists in registry 

183 if coder not in _renderers.RENDERERS: 183 ↛ 184line 183 didn't jump to line 184 because the condition on line 183 was never true

184 raise _exceptions.CoderAbsence( coder ) 

185 # Template directories always use plural form (commands, agents) 

186 source_dir = directory / item_type 

187 if not source_dir.exists(): 187 ↛ 188line 187 didn't jump to line 188 because the condition on line 187 was never true

188 raise _exceptions.TemplateError.for_missing_template( 

189 coder, item_type 

190 ) 

191 templates = [ 

192 f"{item_type}/{p.name}" 

193 for p in source_dir.glob( "*.jinja" ) 

194 ] 

195 if not templates: 195 ↛ 196line 195 didn't jump to line 196 because the condition on line 195 was never true

196 raise _exceptions.TemplateError.for_missing_template( 

197 coder, item_type 

198 ) 

199 return templates 

200 

201 def resolve_content_paths( 

202 self, item_type: str, item_name: str, coder: str 

203 ) -> tuple[ __.Path, __.typx.Optional[ __.Path ] ]: 

204 ''' Resolves primary and fallback content paths. 

205 

206 Returns tuple of (primary_path, fallback_path) where fallback_path 

207 is None if no fallback coder is configured. 

208 

209 This method is public to allow operations module to pre-check 

210 content availability without loading files. 

211 ''' 

212 if item_type == 'skills': 212 ↛ 220line 212 didn't jump to line 220 because the condition on line 212 was always true

213 primary_path = ( 

214 self.location / "contents" / item_type / coder / 

215 f"{item_name}.md" ) 

216 fallback_path = ( 

217 self.location / "contents" / item_type / 'common' / 

218 f"{item_name}.md" ) 

219 return ( primary_path, fallback_path ) 

220 primary_path = ( 

221 self.location / "contents" / item_type / coder / 

222 f"{item_name}.md" ) 

223 fallback_path = None 

224 fallback_mappings = self._retrieve_fallback_mappings( ) 

225 fallback_coder = fallback_mappings.get( coder ) 

226 if fallback_coder: 

227 fallback_path = ( 

228 self.location / "contents" / item_type / 

229 fallback_coder / f"{item_name}.md" ) 

230 return ( primary_path, fallback_path ) 

231 

232 def _retrieve_content_with_fallback( 

233 self, item_type: str, item_name: str, coder: str 

234 ) -> str: 

235 ''' Retrieves content with fallback logic for compatible coders. 

236 

237 Attempts to read content from coder-specific location first, 

238 then falls back to compatible coder if content is missing. 

239 ''' 

240 primary_path, fallback_path = self.resolve_content_paths( 

241 item_type, item_name, coder ) 

242 if primary_path.exists( ): 242 ↛ 243line 242 didn't jump to line 243 because the condition on line 242 was never true

243 return primary_path.read_text( encoding = 'utf-8' ) 

244 if fallback_path and fallback_path.exists( ): 244 ↛ 250line 244 didn't jump to line 250 because the condition on line 244 was always true

245 if item_type == 'skills': 245 ↛ 247line 245 didn't jump to line 247 because the condition on line 245 was always true

246 return fallback_path.read_text( encoding = 'utf-8' ) 

247 fallback_coder = self._retrieve_fallback_mappings( ).get( coder ) 

248 _scribe.debug( f"Using {fallback_coder} content for {coder}" ) 

249 return fallback_path.read_text( encoding = 'utf-8' ) 

250 raise _exceptions.ContentAbsence( item_type, item_name, coder ) 

251 

252 def _parse_template_extension( self, template_name: str ) -> str: 

253 ''' Extracts output extension from template filename. 

254 

255 Template names follow pattern: item.extension.jinja 

256 This extracts the middle component as output extension. 

257 ''' 

258 parts = template_name.split( '.' ) 

259 if len( parts ) >= _TEMPLATE_PARTS_MINIMUM and parts[ -1 ] == 'jinja': 259 ↛ 261line 259 didn't jump to line 261 because the condition on line 259 was always true

260 return parts[ -2 ] 

261 raise _exceptions.TemplateError.for_extension_parse( template_name ) 

262 

263 def _load_item_metadata( 

264 self, item_type: str, item_name: str, coder: str 

265 ) -> dict[ str, __.typx.Any ]: 

266 ''' Loads TOML metadata and extracts context and coder config. 

267 

268 Reads item configuration file and separates context fields 

269 from coder-specific configuration. 

270 ''' 

271 configuration_file = ( 

272 self.location / 'configurations' / item_type 

273 / f"{item_name}.toml" ) 

274 if not configuration_file.exists( ): 274 ↛ 275line 274 didn't jump to line 275 because the condition on line 274 was never true

275 raise _exceptions.ConfigurationAbsence( configuration_file ) 

276 try: toml_content = configuration_file.read_bytes( ) 

277 except ( OSError, IOError ) as exception: 

278 raise _exceptions.ConfigurationAbsence( ) from exception 

279 try: toml_data: dict[ str, __.typx.Any ] = __.tomli.loads( 

280 toml_content.decode( 'utf-8' ) ) 

281 except __.tomli.TOMLDecodeError as exception: 

282 raise _exceptions.ConfigurationInvalidity( 

283 exception 

284 ) from exception 

285 context = toml_data.get( 'context', { } ) 

286 coders_list: list[ dict[ str, __.typx.Any ] ] = ( 

287 toml_data.get( 'coders', [ ] ) ) 

288 # Normalize coders table array to dict keyed by name 

289 # TOML [[coders]] tables are optional; minimal config if absent 

290 coders_dict: dict[ str, dict[ str, __.typx.Any ] ] = { } 

291 for entry in coders_list: 291 ↛ 292line 291 didn't jump to line 292 because the loop on line 291 never started

292 if not isinstance( entry, __.cabc.Mapping ): continue 

293 name_value = entry.get( 'name' ) 

294 if not isinstance( name_value, str ): continue 

295 coders_dict[ name_value ] = entry 

296 # Look up coder config from YAML, fallback to minimal config 

297 coder_config = coders_dict.get( coder, { 'name': coder } ) 

298 return { 'context': context, 'coder': coder_config } 

299 

300 def _produce_jinja_environment( self ) -> _jinja2.Environment: 

301 ''' Produces Jinja2 environment configured for templates directory. 

302 

303 Creates new Jinja2 environment instance with FileSystemLoader 

304 pointing to data source templates directory. 

305 ''' 

306 directory = self.location / "templates" 

307 loader = _jinja2.FileSystemLoader( directory ) 

308 return _jinja2.Environment( 

309 loader = loader, 

310 autoescape = False, # noqa: S701 Markdown output, not HTML 

311 ) 

312 

313 

314 def _select_template_for_coder( self, item_type: str, coder: str ) -> str: 

315 try: renderer = _renderers.RENDERERS[ coder ] 

316 except KeyError as exception: 

317 raise _exceptions.CoderAbsence( coder ) from exception 

318 if item_type == 'skills': 318 ↛ 325line 318 didn't jump to line 325 because the condition on line 318 was always true

319 candidate = f"{item_type}/common.md.jinja" 

320 available = self._survey_available_templates( item_type, coder ) 

321 if candidate in available: return candidate 321 ↛ 322line 321 didn't jump to line 322 because the condition on line 321 was always true

322 raise _exceptions.TemplateError.for_missing_template( 

323 coder, item_type 

324 ) 

325 flavor = renderer.get_template_flavor( item_type ) 

326 available = self._survey_available_templates( item_type, coder ) 

327 # Template paths always use plural item_type (commands, agents) 

328 for extension in [ 'md', 'toml' ]: 

329 organized_path = ( 

330 f"{item_type}/{flavor}.{extension}.jinja" ) 

331 if organized_path in available: 

332 return organized_path 

333 raise _exceptions.TemplateError.for_missing_template( 

334 coder, item_type 

335 )