Coverage for sources/agentsmgr/generator.py: 37%
152 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-15 05:15 +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':
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': 156 ↛ 161line 156 didn't jump to line 161 because the condition on line 156 was always true
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:
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():
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:
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( ):
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 Per-project skills materialize under the canonical tree
273 ``.auxiliary/agents/skills/<item_name>/SKILL.md``. Per-user
274 mode keeps the coder base ``skills/<item_name>/SKILL.md`` path.
275 '''
276 if actual_mode == 'per-project': 276 ↛ 280line 276 didn't jump to line 280 because the condition on line 276 was always true
277 return (
278 target / '.auxiliary' / 'agents' / 'skills' /
279 item_name / 'SKILL.md' )
280 base_directory = renderer.resolve_base_directory(
281 mode = actual_mode,
282 target = target,
283 configuration = self.application_configuration,
284 environment = __.os.environ,
285 )
286 dirname = renderer.calculate_directory_location( 'skills' )
287 return base_directory / dirname / item_name / "SKILL.md"
289 def _parse_template_extension( self, template_name: str ) -> str:
290 ''' Extracts output extension from template filename.
292 Template names follow pattern: item.extension.jinja
293 This extracts the middle component as output extension.
294 '''
295 parts = template_name.split( '.' )
296 if len( parts ) >= _TEMPLATE_PARTS_MINIMUM and parts[ -1 ] == 'jinja':
297 return parts[ -2 ]
298 raise _exceptions.TemplateError.for_extension_parse( template_name )
300 def _load_item_metadata(
301 self, item_type: str, item_name: str, coder: str
302 ) -> dict[ str, __.typx.Any ]:
303 ''' Loads TOML metadata and extracts context and coder config.
305 Reads item configuration file and separates context fields
306 from coder-specific configuration.
307 '''
308 configuration_file = (
309 self.location / 'configurations' / item_type
310 / f"{item_name}.toml" )
311 if not configuration_file.exists( ):
312 raise _exceptions.ConfigurationAbsence( configuration_file )
313 try: toml_content = configuration_file.read_bytes( )
314 except ( OSError, IOError ) as exception:
315 raise _exceptions.ConfigurationAbsence( ) from exception
316 try: toml_data: dict[ str, __.typx.Any ] = __.tomli.loads(
317 toml_content.decode( 'utf-8' ) )
318 except __.tomli.TOMLDecodeError as exception:
319 raise _exceptions.ConfigurationInvalidity(
320 exception
321 ) from exception
322 context = toml_data.get( 'context', { } )
323 coders_list: list[ dict[ str, __.typx.Any ] ] = (
324 toml_data.get( 'coders', [ ] ) )
325 # Normalize coders table array to dict keyed by name
326 # TOML [[coders]] tables are optional; minimal config if absent
327 coders_dict: dict[ str, dict[ str, __.typx.Any ] ] = { }
328 for entry in coders_list:
329 if not isinstance( entry, __.cabc.Mapping ): continue
330 name_value = entry.get( 'name' )
331 if not isinstance( name_value, str ): continue
332 coders_dict[ name_value ] = entry
333 # Look up coder config from YAML, fallback to minimal config
334 coder_config = coders_dict.get( coder, { 'name': coder } )
335 return { 'context': context, 'coder': coder_config }
337 def _produce_jinja_environment( self ) -> _jinja2.Environment:
338 ''' Produces Jinja2 environment configured for templates directory.
340 Creates new Jinja2 environment instance with FileSystemLoader
341 pointing to data source templates directory.
342 '''
343 directory = self.location / "templates"
344 loader = _jinja2.FileSystemLoader( directory )
345 return _jinja2.Environment(
346 loader = loader,
347 autoescape = False, # noqa: S701 Markdown output, not HTML
348 )
351 def _select_template_for_coder( self, item_type: str, coder: str ) -> str:
352 try: renderer = _renderers.RENDERERS[ coder ]
353 except KeyError as exception:
354 raise _exceptions.CoderAbsence( coder ) from exception
355 flavor = renderer.get_template_flavor( item_type )
356 available = self._survey_available_templates( item_type, coder )
357 # Template paths always use plural item_type (commands, agents)
358 for extension in [ 'md', 'toml' ]:
359 organized_path = (
360 f"{item_type}/{flavor}.{extension}.jinja" )
361 if organized_path in available:
362 return organized_path
363 raise _exceptions.TemplateError.for_missing_template(
364 coder, item_type
365 )