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

146 statements  

« prev     ^ index     » next       coverage.py v7.15.0, created at 2026-07-08 06:57 +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 ] 

39 

40_TEMPLATE_PARTS_MINIMUM = 3 

41 

42 

43_scribe = __.provide_scribe( __name__ ) 

44 

45 

46class RenderedItem( __.immut.DataclassObject ): 

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

48 

49 content: str 

50 location: __.Path 

51 

52 

53class ItemRenderRequest( __.immut.DataclassObject ): 

54 item_type: str 

55 item_name: str 

56 template_name: str 

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

58 

59 

60class ContentGenerator( __.immut.DataclassObject ): 

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

62 

63 Provides template-based content generation with intelligent 

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

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

66 ''' 

67 

68 location: __.Path 

69 configuration: _cmdbase.CoderConfiguration 

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

71 __.dcls.field( 

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

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

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

75 

76 def __post_init__( self ) -> None: 

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

78 self._produce_jinja_environment( ) ) 

79 

80 

81 def _retrieve_fallback_mappings( self ) -> CoderFallbackMap: 

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

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

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

85 return __.immut.Dictionary( fallbacks ) 

86 

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

88 try: return _renderers.RENDERERS[ coder ] 

89 except KeyError as exception: 

90 raise _exceptions.CoderAbsence( coder ) from exception 

91 

92 def _resolve_actual_mode( 

93 self, 

94 renderer: _renderers.RendererBase, 

95 coder: str, 

96 ) -> _renderers.ExplicitTargetMode: 

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

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

99 actual_mode: _renderers.ExplicitTargetMode = self.mode 

100 renderer.validate_mode( actual_mode ) 

101 return actual_mode 

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

103 

104 def _render_content( 

105 self, 

106 template_name: str, 

107 body: str, 

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

109 ) -> str: 

110 template = self.jinja_environment.get_template( template_name ) 

111 normalized = _context.normalize_render_context( 

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

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

114 return template.render( **variables ) 

115 

116 def _produce_item_location( 

117 self, 

118 renderer: _renderers.RendererBase, 

119 actual_mode: _renderers.ExplicitTargetMode, 

120 target: __.Path, 

121 request: ItemRenderRequest, 

122 ) -> __.Path: 

123 base_directory = renderer.resolve_base_directory( 

124 mode = actual_mode, 

125 target = target, 

126 configuration = self.application_configuration, 

127 environment = __.os.environ, 

128 ) 

129 extension = self._parse_template_extension( request.template_name ) 

130 if request.item_type == 'skills': 130 ↛ 131line 130 didn't jump to line 131 because the condition on line 130 was never true

131 dirname = renderer.produce_output_structure( request.item_type ) 

132 return ( 

133 base_directory / dirname / request.item_name / 

134 f"SKILL.{extension}" 

135 ) 

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

137 if category is None: category = __.absent 

138 dirname = renderer.produce_output_structure( 

139 request.item_type, category ) 

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

141 

142 def render_single_item( 

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

144 ) -> RenderedItem: 

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

146 

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

148 final coder-specific file. Returns RenderedItem with content 

149 and location. 

150 

151 For skills, bypasses the 3-tier pipeline and copies content 

152 directly since skills are portable across coders. 

153 ''' 

154 renderer = self._resolve_renderer( coder ) 

155 actual_mode = self._resolve_actual_mode( renderer, coder ) 

156 if item_type == 'skills': 

157 body = self._retrieve_skill_content( item_name ) 

158 location = self._produce_skill_location( 

159 renderer, actual_mode, target, item_name ) 

160 return RenderedItem( content = body, location = location ) 

161 body = self._retrieve_content_with_fallback( 

162 item_type, item_name, coder ) 

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

164 template_name = self._select_template_for_coder( item_type, coder ) 

165 request = ItemRenderRequest( 

166 item_type = item_type, 

167 item_name = item_name, 

168 template_name = template_name, 

169 metadata = metadata, 

170 ) 

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

172 location = self._produce_item_location( 

173 renderer, 

174 actual_mode, 

175 target, 

176 request, 

177 ) 

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

179 

180 def _survey_available_templates( 

181 self, item_type: str, coder: str 

182 ) -> list[ str ]: 

183 directory = self.location / "templates" 

184 # Validate coder exists in registry 

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

186 raise _exceptions.CoderAbsence( coder ) 

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

188 source_dir = directory / item_type 

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

190 raise _exceptions.TemplateError.for_missing_template( 

191 coder, item_type 

192 ) 

193 templates = [ 

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

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

196 ] 

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

198 raise _exceptions.TemplateError.for_missing_template( 

199 coder, item_type 

200 ) 

201 return templates 

202 

203 def resolve_content_paths( 

204 self, item_type: str, item_name: str, coder: str 

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

206 ''' Resolves primary and fallback content paths. 

207 

208 Returns tuple of (primary_path, fallback_path) where fallback_path 

209 is None if no fallback coder is configured. 

210 

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

212 content availability without loading files. 

213 ''' 

214 primary_path = ( 

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

216 f"{item_name}.md" ) 

217 fallback_path = None 

218 fallback_mappings = self._retrieve_fallback_mappings( ) 

219 fallback_coder = fallback_mappings.get( coder ) 

220 if fallback_coder: 

221 fallback_path = ( 

222 self.location / "contents" / item_type / 

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

224 return ( primary_path, fallback_path ) 

225 

226 def _retrieve_content_with_fallback( 

227 self, item_type: str, item_name: str, coder: str 

228 ) -> str: 

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

230 

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

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

233 ''' 

234 primary_path, fallback_path = self.resolve_content_paths( 

235 item_type, item_name, coder ) 

236 if primary_path.exists( ): 

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

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

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

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

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

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

243 

244 def _retrieve_skill_content( self, item_name: str ) -> str: 

245 ''' Retrieves skill content directly from 

246 distribution/per-project/general/skills/. 

247 

248 Skills are portable across coders, so no coder-specific 

249 lookup or fallback logic is needed. 

250 ''' 

251 path = ( 

252 self.location / "per-project" / "general" / "skills" / 

253 f"{item_name}.md" ) 

254 if path.exists( ): 254 ↛ 256line 254 didn't jump to line 256 because the condition on line 254 was always true

255 return path.read_text( encoding = 'utf-8' ) 

256 raise _exceptions.ContentAbsence( 'skills', item_name, 'common' ) 

257 

258 def _produce_skill_location( 

259 self, 

260 renderer: _renderers.RendererBase, 

261 actual_mode: _renderers.ExplicitTargetMode, 

262 target: __.Path, 

263 item_name: str, 

264 ) -> __.Path: 

265 ''' Produces output location for a skill. 

266 

267 Skills always use the pattern: 

268 <base>/skills/<item_name>/SKILL.md 

269 ''' 

270 base_directory = renderer.resolve_base_directory( 

271 mode = actual_mode, 

272 target = target, 

273 configuration = self.application_configuration, 

274 environment = __.os.environ, 

275 ) 

276 dirname = renderer.calculate_directory_location( 'skills' ) 

277 return base_directory / dirname / item_name / "SKILL.md" 

278 

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

280 ''' Extracts output extension from template filename. 

281 

282 Template names follow pattern: item.extension.jinja 

283 This extracts the middle component as output extension. 

284 ''' 

285 parts = template_name.split( '.' ) 

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

287 return parts[ -2 ] 

288 raise _exceptions.TemplateError.for_extension_parse( template_name ) 

289 

290 def _load_item_metadata( 

291 self, item_type: str, item_name: str, coder: str 

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

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

294 

295 Reads item configuration file and separates context fields 

296 from coder-specific configuration. 

297 ''' 

298 configuration_file = ( 

299 self.location / 'configurations' / item_type 

300 / f"{item_name}.toml" ) 

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

302 raise _exceptions.ConfigurationAbsence( configuration_file ) 

303 try: toml_content = configuration_file.read_bytes( ) 

304 except ( OSError, IOError ) as exception: 

305 raise _exceptions.ConfigurationAbsence( ) from exception 

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

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

308 except __.tomli.TOMLDecodeError as exception: 

309 raise _exceptions.ConfigurationInvalidity( 

310 exception 

311 ) from exception 

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

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

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

315 # Normalize coders table array to dict keyed by name 

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

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

318 for entry in coders_list: 

319 if not isinstance( entry, __.cabc.Mapping ): continue 319 ↛ 318line 319 didn't jump to line 318 because the continue on line 319 wasn't executed

320 name_value = entry.get( 'name' ) 

321 if not isinstance( name_value, str ): continue 321 ↛ 318line 321 didn't jump to line 318 because the continue on line 321 wasn't executed

322 coders_dict[ name_value ] = entry 

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

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

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

326 

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

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

329 

330 Creates new Jinja2 environment instance with FileSystemLoader 

331 pointing to data source templates directory. 

332 ''' 

333 directory = self.location / "templates" 

334 loader = _jinja2.FileSystemLoader( directory ) 

335 return _jinja2.Environment( 

336 loader = loader, 

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

338 ) 

339 

340 

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

342 try: renderer = _renderers.RENDERERS[ coder ] 

343 except KeyError as exception: 

344 raise _exceptions.CoderAbsence( coder ) from exception 

345 flavor = renderer.get_template_flavor( item_type ) 

346 available = self._survey_available_templates( item_type, coder ) 

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

348 for extension in [ 'md', 'toml' ]: 348 ↛ 353line 348 didn't jump to line 353 because the loop on line 348 didn't complete

349 organized_path = ( 

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

351 if organized_path in available: 351 ↛ 348line 351 didn't jump to line 348 because the condition on line 351 was always true

352 return organized_path 

353 raise _exceptions.TemplateError.for_missing_template( 

354 coder, item_type 

355 )