Coverage for sources/agentsmgr/generator.py: 82%
150 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-15 21:08 +0000
1# vim: set filetype=python fileencoding=utf-8:
2# -*- coding: utf-8 -*-
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#============================================================================#
21''' Content generation for coder-specific templates and items.
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'''
29import jinja2 as _jinja2
31from . import __
32from . import cmdbase as _cmdbase
33from . import context as _context
34from . import exceptions as _exceptions
35from . import renderers as _renderers
38CoderFallbackMap: __.typx.TypeAlias = __.immut.Dictionary[ str, str ]
40_TEMPLATE_PARTS_MINIMUM = 3
43_scribe = __.provide_scribe( __name__ )
46class RenderedItem( __.immut.DataclassObject ):
47 ''' Single rendered item with location and content. '''
49 content: str
50 location: __.Path
53class ItemRenderRequest( __.immut.DataclassObject ):
54 item_type: str
55 item_name: str
56 template_name: str
57 metadata: dict[ str, __.typx.Any ]
60class ContentGenerator( __.immut.DataclassObject ):
61 ''' Generates coder-specific content from data sources.
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 '''
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 )
76 def __post_init__( self ) -> None:
77 self.jinja_environment = ( # pyright: ignore[reportAttributeAccessIssue]
78 self._produce_jinja_environment( ) )
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 )
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
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 )
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 )
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}"
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.
147 Combines TOML metadata, content body, and template to produce
148 final coder-specific file. Returns RenderedItem with content
149 and location.
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 )
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
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.
208 Returns tuple of (primary_path, fallback_path) where fallback_path
209 is None if no fallback coder is configured.
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 )
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.
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 )
244 def _retrieve_skill_content( self, item_name: str ) -> str:
245 ''' Retrieves SKILL.md body from distribution skills layout.
247 Prefers directory packages
248 (``per-project/general/skills/<name>/SKILL.md``) over legacy
249 flat files (``per-project/general/skills/<name>.md``). Skills
250 are portable across coders; supporting files are copied by
251 populate, not returned here.
252 '''
253 skills_root = (
254 self.location / "per-project" / "general" / "skills" )
255 directory_skill = skills_root / item_name / "SKILL.md"
256 if directory_skill.is_file( ):
257 return directory_skill.read_text( encoding = 'utf-8' )
258 flat_skill = skills_root / f"{item_name}.md"
259 if flat_skill.is_file( ): 259 ↛ 261line 259 didn't jump to line 261 because the condition on line 259 was always true
260 return flat_skill.read_text( encoding = 'utf-8' )
261 raise _exceptions.ContentAbsence( 'skills', item_name, 'common' )
263 def _produce_skill_location(
264 self,
265 renderer: _renderers.RendererBase,
266 actual_mode: _renderers.ExplicitTargetMode,
267 target: __.Path,
268 item_name: str,
269 ) -> __.Path:
270 ''' Produces output location for a skill.
272 Skills always use the pattern:
273 <base>/skills/<item_name>/SKILL.md
274 '''
275 base_directory = renderer.resolve_base_directory(
276 mode = actual_mode,
277 target = target,
278 configuration = self.application_configuration,
279 environment = __.os.environ,
280 )
281 dirname = renderer.calculate_directory_location( 'skills' )
282 return base_directory / dirname / item_name / "SKILL.md"
284 def _parse_template_extension( self, template_name: str ) -> str:
285 ''' Extracts output extension from template filename.
287 Template names follow pattern: item.extension.jinja
288 This extracts the middle component as output extension.
289 '''
290 parts = template_name.split( '.' )
291 if len( parts ) >= _TEMPLATE_PARTS_MINIMUM and parts[ -1 ] == 'jinja': 291 ↛ 293line 291 didn't jump to line 293 because the condition on line 291 was always true
292 return parts[ -2 ]
293 raise _exceptions.TemplateError.for_extension_parse( template_name )
295 def _load_item_metadata(
296 self, item_type: str, item_name: str, coder: str
297 ) -> dict[ str, __.typx.Any ]:
298 ''' Loads TOML metadata and extracts context and coder config.
300 Reads item configuration file and separates context fields
301 from coder-specific configuration.
302 '''
303 configuration_file = (
304 self.location / 'configurations' / item_type
305 / f"{item_name}.toml" )
306 if not configuration_file.exists( ): 306 ↛ 307line 306 didn't jump to line 307 because the condition on line 306 was never true
307 raise _exceptions.ConfigurationAbsence( configuration_file )
308 try: toml_content = configuration_file.read_bytes( )
309 except ( OSError, IOError ) as exception:
310 raise _exceptions.ConfigurationAbsence( ) from exception
311 try: toml_data: dict[ str, __.typx.Any ] = __.tomli.loads(
312 toml_content.decode( 'utf-8' ) )
313 except __.tomli.TOMLDecodeError as exception:
314 raise _exceptions.ConfigurationInvalidity(
315 exception
316 ) from exception
317 context = toml_data.get( 'context', { } )
318 coders_list: list[ dict[ str, __.typx.Any ] ] = (
319 toml_data.get( 'coders', [ ] ) )
320 # Normalize coders table array to dict keyed by name
321 # TOML [[coders]] tables are optional; minimal config if absent
322 coders_dict: dict[ str, dict[ str, __.typx.Any ] ] = { }
323 for entry in coders_list:
324 if not isinstance( entry, __.cabc.Mapping ): continue 324 ↛ 323line 324 didn't jump to line 323 because the continue on line 324 wasn't executed
325 name_value = entry.get( 'name' )
326 if not isinstance( name_value, str ): continue 326 ↛ 323line 326 didn't jump to line 323 because the continue on line 326 wasn't executed
327 coders_dict[ name_value ] = entry
328 # Look up coder config from YAML, fallback to minimal config
329 coder_config = coders_dict.get( coder, { 'name': coder } )
330 return { 'context': context, 'coder': coder_config }
332 def _produce_jinja_environment( self ) -> _jinja2.Environment:
333 ''' Produces Jinja2 environment configured for templates directory.
335 Creates new Jinja2 environment instance with FileSystemLoader
336 pointing to data source templates directory.
337 '''
338 directory = self.location / "templates"
339 loader = _jinja2.FileSystemLoader( directory )
340 return _jinja2.Environment(
341 loader = loader,
342 autoescape = False, # noqa: S701 Markdown output, not HTML
343 )
346 def _select_template_for_coder( self, item_type: str, coder: str ) -> str:
347 try: renderer = _renderers.RENDERERS[ coder ]
348 except KeyError as exception:
349 raise _exceptions.CoderAbsence( coder ) from exception
350 flavor = renderer.get_template_flavor( item_type )
351 available = self._survey_available_templates( item_type, coder )
352 # Template paths always use plural item_type (commands, agents)
353 for extension in [ 'md', 'toml' ]: 353 ↛ 358line 353 didn't jump to line 358 because the loop on line 353 didn't complete
354 organized_path = (
355 f"{item_type}/{flavor}.{extension}.jinja" )
356 if organized_path in available: 356 ↛ 353line 356 didn't jump to line 353 because the condition on line 356 was always true
357 return organized_path
358 raise _exceptions.TemplateError.for_missing_template(
359 coder, item_type
360 )