Coverage for sources/agentsmgr/operations.py: 60%
242 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 -*-
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''' Core operations for content generation and directory population.
23 This module provides functions for orchestrating content generation,
24 including directory population and file writing operations with
25 simulation support. Also provides generation from components/ to
26 distribution/ with staleness checking.
27'''
30import difflib as _difflib
32from . import __
33from . import exceptions as _exceptions
34from . import generator as _generator
35from . import renderers as _renderers
38_MANAGED_BLOCK_BEGIN = '# BEGIN: Managed by agentsmgr (emcd-agents)'
39_MANAGED_BLOCK_WARNING = '# Do not manually edit entries in this block.'
40_MANAGED_BLOCK_END = '# END: Managed by agentsmgr (emcd-agents)'
41_EXTENSION_PARTS_MINIMUM = 2
44def populate_directory(
45 generator: _generator.ContentGenerator,
46 target: __.Path,
47 simulate: bool = False
48) -> tuple[ int, int ]:
49 ''' Generates all content items to target directory.
51 Orchestrates content generation for all coders and item types
52 configured in generator. Returns tuple of (items_attempted,
53 items_written).
54 '''
55 items_attempted = 0
56 items_written = 0
57 _ensure_output_directories( generator, target, simulate )
58 for coder_name in generator.configuration[ 'coders' ]:
59 try: renderer = _renderers.RENDERERS[ coder_name ]
60 except KeyError: continue
61 for item_type in renderer.item_types_available:
62 attempted, written = generate_coder_item_type(
63 generator, coder_name, item_type, target, simulate )
64 items_attempted += attempted
65 items_written += written
66 return ( items_attempted, items_written )
68def _ensure_output_directories(
69 generator: _generator.ContentGenerator,
70 target: __.Path,
71 simulate: bool,
72) -> None:
73 if simulate or generator.mode == 'nowhere': return
74 for coder_name in generator.configuration[ 'coders' ]:
75 try: renderer = _renderers.RENDERERS[ coder_name ]
76 except KeyError: continue
77 if generator.mode == 'default': actual_mode = renderer.mode_default
78 else: actual_mode = generator.mode
79 if actual_mode not in ( 'per-user', 'per-project' ): continue
80 if actual_mode not in renderer.modes_available: continue
81 base_directory = renderer.resolve_base_directory(
82 mode = actual_mode,
83 target = target,
84 configuration = generator.application_configuration,
85 environment = __.os.environ,
86 )
87 for item_type in renderer.item_types_available:
88 dirname = renderer.produce_output_structure( item_type )
89 ( base_directory / dirname ).mkdir(
90 parents = True, exist_ok = True )
93def _content_exists(
94 generator: _generator.ContentGenerator,
95 item_type: str,
96 item_name: str,
97 coder: str
98) -> bool:
99 ''' Checks if content file exists without loading it.
101 Uses path resolution from ContentGenerator to check both primary
102 and fallback locations. Returns True if content is available.
103 '''
104 primary_path, fallback_path = generator.resolve_content_paths(
105 item_type, item_name, coder )
106 if primary_path.exists( ):
107 return True
108 return bool( fallback_path and fallback_path.exists( ) )
111def generate_coder_item_type(
112 generator: _generator.ContentGenerator,
113 coder: str,
114 item_type: str,
115 target: __.Path,
116 simulate: bool
117) -> tuple[ int, int ]:
118 ''' Generates items of specific type for a coder.
120 Generates all items (commands or agents) for specified coder by
121 iterating through configuration files. Skills are direct
122 distribution artifacts and are not generated from components.
123 Pre-checks content availability and skips items with missing
124 content. Returns tuple of (items_attempted, items_written).
125 '''
126 items_attempted = 0
127 items_written = 0
128 if generator.mode == 'nowhere':
129 return ( items_attempted, items_written )
130 configuration_directory = (
131 generator.location / 'configurations' / item_type )
132 if not configuration_directory.exists( ):
133 return ( items_attempted, items_written )
134 for configuration_file in configuration_directory.glob( '*.toml' ):
135 item_name = configuration_file.stem
136 if not _content_exists( generator, item_type, item_name, coder ):
137 __.provide_scribe( __name__ ).warning(
138 f"Skipping {item_type}/{item_name} for {coder}: "
139 "content not found" )
140 continue
141 items_attempted += 1
142 result = generator.render_single_item(
143 item_type, item_name, coder, target )
144 if save_content( result.content, result.location, simulate ):
145 items_written += 1
146 return ( items_attempted, items_written )
149def save_content(
150 content: str, location: __.Path, simulate: bool = False
151) -> bool:
152 ''' Saves content to location, creating parent directories as needed.
154 Writes content to specified location, creating parent directories
155 if necessary. In simulation mode, no actual writing occurs.
156 Returns True if file was written, False if simulated.
157 '''
158 if simulate: return False 158 ↛ exitline 158 didn't return from function 'save_content' because the return on line 158 wasn't executed
159 try: location.parent.mkdir( parents = True, exist_ok = True )
160 except ( OSError, IOError ) as exception:
161 raise _exceptions.FileOperationFailure(
162 location.parent, "create directory" ) from exception
163 try: location.write_text( content, encoding = 'utf-8' )
164 except ( OSError, IOError ) as exception:
165 raise _exceptions.FileOperationFailure(
166 location, "save content" ) from exception
167 return True
170def update_git_exclude(
171 target: __.Path,
172 entries: __.cabc.Collection[ str ],
173 simulate: bool = False
174) -> int:
175 ''' Updates .git/info/exclude with managed block of agentsmgr entries.
177 Maintains a clearly-marked block of entries managed by agentsmgr,
178 with complete replacement on each update. Entries are sorted
179 lexicographically within the block. User entries outside the
180 managed block are preserved.
182 Uses repository discovery from the explicit target and uses the
183 common git directory for shared resources in worktrees.
185 Returns count of entries in managed block.
186 '''
187 if simulate: return 0 187 ↛ exitline 187 didn't return from function 'update_git_exclude' because the return on line 187 wasn't executed
188 git_dir = _resolve_git_directory( target )
189 if not git_dir: return 0
190 exclude_file = git_dir / 'info' / 'exclude'
191 if not exclude_file.exists( ): return 0 191 ↛ exitline 191 didn't return from function 'update_git_exclude' because the return on line 191 wasn't executed
192 try: content = exclude_file.read_text( encoding = 'utf-8' )
193 except ( OSError, IOError ) as exception:
194 raise _exceptions.FileOperationFailure(
195 exclude_file, "read git exclude file" ) from exception
196 normalized_entries = sorted( {
197 _normalize_git_exclude_entry( entry )
198 for entry in entries if entry.strip( )
199 } )
200 if not normalized_entries: 200 ↛ 201line 200 didn't jump to line 201 because the condition on line 200 was never true
201 new_content = _remove_managed_block( content )
202 if new_content == content: return 0
203 else:
204 new_content = _update_managed_block( content, normalized_entries )
205 try: exclude_file.write_text( new_content, encoding = 'utf-8' )
206 except ( OSError, IOError ) as exception:
207 raise _exceptions.FileOperationFailure(
208 exclude_file, "update git exclude file" ) from exception
209 return len( normalized_entries )
212def _normalize_git_exclude_entry( entry: str ) -> str:
213 ''' Normalizes a managed git exclude entry. '''
214 normalized = entry.strip( ).replace( '\\', '/' )
215 if not normalized.startswith( '/' ):
216 normalized = f"/{normalized}"
217 return normalized
220def _update_managed_block(
221 content: str, entries: __.cabc.Sequence[ str ]
222) -> str:
223 ''' Updates content with new managed block containing sorted entries.
225 Locates existing managed block (if present) and replaces it with
226 new block. If no block exists, appends to end of file. Preserves
227 user content outside the managed block.
228 '''
229 lines = content.splitlines( )
230 before_block, after_block = _partition_around_managed_block( lines )
231 block_lines = [ _MANAGED_BLOCK_BEGIN, _MANAGED_BLOCK_WARNING ]
232 block_lines.extend( entries )
233 block_lines.append( _MANAGED_BLOCK_END )
234 if before_block and before_block[ -1 ].strip( ): 234 ↛ 236line 234 didn't jump to line 236 because the condition on line 234 was always true
235 before_block.append( '' )
236 result_lines = before_block + block_lines
237 if after_block: 237 ↛ 238line 237 didn't jump to line 238 because the condition on line 237 was never true
238 result_lines.append( '' )
239 result_lines.extend( after_block )
240 return '\n'.join( result_lines ) + '\n'
243def _remove_managed_block( content: str ) -> str:
244 ''' Removes managed block from content, preserving user entries.
246 Locates and removes managed block if present. Returns content
247 unchanged if no block found.
248 '''
249 lines = content.splitlines( )
250 before_block, after_block = _partition_around_managed_block( lines )
251 if not before_block and not after_block:
252 return content
253 result_lines = before_block
254 if result_lines and after_block:
255 if result_lines[ -1 ].strip( ):
256 result_lines.append( '' )
257 result_lines.extend( after_block )
258 elif after_block:
259 result_lines = after_block
260 if not result_lines: return ''
261 return '\n'.join( result_lines ) + '\n'
264def _partition_around_managed_block(
265 lines: __.cabc.Sequence[ str ]
266) -> tuple[ list[ str ], list[ str ] ]:
267 ''' Partitions lines into content before and after managed block.
269 Locates managed block markers and returns (before, after) tuple.
270 If block is malformed or not found, returns (all_lines, []).
271 Malformed blocks are treated as non-existent.
272 '''
273 try: begin_index = lines.index( _MANAGED_BLOCK_BEGIN )
274 except ValueError: return ( list( lines ), [ ] )
275 try: end_index = lines.index( _MANAGED_BLOCK_END, begin_index )
276 except ValueError:
277 __.provide_scribe( __name__ ).warning(
278 "Malformed agentsmgr block in .git/info/exclude; rebuilding." )
279 return ( list( lines ), [ ] )
280 if end_index < begin_index: 280 ↛ 281line 280 didn't jump to line 281 because the condition on line 280 was never true
281 __.provide_scribe( __name__ ).warning(
282 "Malformed agentsmgr block in .git/info/exclude; rebuilding." )
283 return ( list( lines ), [ ] )
284 before_block = list( lines[ :begin_index ] )
285 while before_block and not before_block[ -1 ].strip( ):
286 before_block.pop( )
287 after_block = list( lines[ end_index + 1: ] )
288 while after_block and not after_block[ 0 ].strip( ): 288 ↛ 289line 288 didn't jump to line 289 because the condition on line 288 was never true
289 after_block.pop( 0 )
290 return ( before_block, after_block )
292def _resolve_git_directory(
293 start_path: __.Path
294) -> __.typx.Optional[ __.Path ]:
295 ''' Resolves git directory location, handling worktrees.
297 Uses Dulwich to discover the repository from the explicit target.
298 Returns common git directory (shared across worktrees) for access
299 to shared resources like info/exclude.
301 Returns None if not in a git repository or on error.
302 '''
303 from dulwich.repo import Repo
304 try: repo = Repo.discover( str( start_path ) )
305 except Exception: return None
306 git_dir_path = __.Path( repo.controldir( ) )
307 return _discover_common_git_directory( git_dir_path )
309def _discover_common_git_directory( git_dir: __.Path ) -> __.Path:
310 ''' Discovers common git directory, handling worktree commondir.
312 For worktrees, reads commondir file to find shared resources.
313 For standard repos, returns git_dir unchanged.
314 '''
315 commondir_file = git_dir / 'commondir'
316 if not commondir_file.exists( ): 316 ↛ 318line 316 didn't jump to line 318 because the condition on line 316 was always true
317 return git_dir
318 try: common_path = commondir_file.read_text( encoding = 'utf-8' ).strip( )
319 except ( OSError, IOError ): return git_dir
320 return ( git_dir / common_path ).resolve( )
323def generate_distribution(
324 generator: _generator.ContentGenerator,
325 distribution: __.Path,
326 simulate: bool = False,
327) -> tuple[ int, int ]:
328 ''' Generates pre-rendered artifacts from components/ to distribution/.
330 Reads from the 3-tier pipeline source (configurations, templates,
331 per-coder contents) and writes rendered commands and agents to
332 distribution/. Skills are not generated; they are direct
333 distribution artifacts.
335 Returns tuple of (items_attempted, items_written).
336 '''
337 items_attempted = 0
338 items_written = 0
339 for coder_name in generator.configuration[ 'coders' ]:
340 try: renderer = _renderers.RENDERERS[ coder_name ]
341 except KeyError: continue
342 for item_type in renderer.item_types_available:
343 if item_type == 'skills':
344 continue # Skills are direct distribution artifacts.
345 attempted, written = _generate_for_distribution(
346 generator, coder_name, item_type, distribution, simulate )
347 items_attempted += attempted
348 items_written += written
349 return ( items_attempted, items_written )
352def _generate_for_distribution(
353 generator: _generator.ContentGenerator,
354 coder: str,
355 item_type: str,
356 distribution: __.Path,
357 simulate: bool,
358) -> tuple[ int, int ]:
359 ''' Generates items of a type for a coder into distribution/.
361 Reads configuration and content from components/, renders through
362 the 3-tier pipeline, and writes to
363 distribution/per-project/coders/<coder>/<item_type>/.
364 Returns tuple of (items_attempted, items_written).
365 '''
366 items_attempted = 0
367 items_written = 0
368 configuration_directory = (
369 generator.location / 'configurations' / item_type )
370 if not configuration_directory.exists( ): 370 ↛ 371line 370 didn't jump to line 371 because the condition on line 370 was never true
371 return ( items_attempted, items_written )
372 for configuration_file in configuration_directory.glob( '*.toml' ):
373 item_name = configuration_file.stem
374 if not _content_exists( generator, item_type, item_name, coder ): 374 ↛ 375line 374 didn't jump to line 375 because the condition on line 374 was never true
375 __.provide_scribe( __name__ ).warning(
376 f"Skipping {item_type}/{item_name} for {coder}: "
377 "content not found" )
378 continue
379 items_attempted += 1
380 result = generator.render_single_item(
381 item_type, item_name, coder, distribution )
382 renderer = _renderers.RENDERERS[ coder ]
383 dirname = renderer.produce_output_structure( item_type )
384 output_path = (
385 distribution / 'per-project' / 'coders' / coder / dirname /
386 f"{item_name}.{_parse_output_extension( result.location )}" )
387 if save_content( result.content, output_path, simulate ): 387 ↛ 372line 387 didn't jump to line 372 because the condition on line 387 was always true
388 items_written += 1
389 return ( items_attempted, items_written )
392def _parse_output_extension( location: __.Path ) -> str:
393 ''' Extracts output extension from rendered location path.
395 Skips the last suffix (which is the item name suffix) and returns
396 the meaningful extension. For SKILL.md returns "md".
397 '''
398 name = location.name
399 if name == 'SKILL.md': return 'md' 399 ↛ exitline 399 didn't return from function '_parse_output_extension' because the return on line 399 wasn't executed
400 parts = name.split( '.' )
401 if len( parts ) >= _EXTENSION_PARTS_MINIMUM: return parts[ -1 ] 401 ↛ 402line 401 didn't jump to line 402 because the condition on line 401 was always true
402 return 'md'
405def check_distribution_staleness(
406 generator: _generator.ContentGenerator,
407 distribution: __.Path,
408) -> tuple[ int, list[ str ] ]:
409 ''' Checks for staleness between components/ and distribution/.
411 Regenerates from components/ and compares against existing
412 distribution/ files. Also detects orphaned artifacts that exist
413 in distribution/ but are no longer generated from components/.
414 Returns tuple of (items_checked, diff_lines).
415 Empty diff_lines means distribution is current.
416 '''
417 items_checked = 0
418 all_diffs: list[ str ] = [ ]
419 expected_paths: set[ __.Path ] = set( )
420 for coder_name in generator.configuration[ 'coders' ]:
421 try: renderer = _renderers.RENDERERS[ coder_name ]
422 except KeyError: continue
423 for item_type in renderer.item_types_available:
424 if item_type == 'skills':
425 continue
426 checked, diffs, paths = _check_staleness_for_type(
427 generator, coder_name, item_type, distribution )
428 items_checked += checked
429 all_diffs.extend( diffs )
430 expected_paths.update( paths )
431 # Detect orphaned artifacts in generated directories
432 orphans = _detect_orphaned_artifacts(
433 distribution, generator.configuration[ 'coders' ], expected_paths )
434 all_diffs.extend( orphans )
435 return ( items_checked, all_diffs )
438def _check_staleness_for_type(
439 generator: _generator.ContentGenerator,
440 coder: str,
441 item_type: str,
442 distribution: __.Path,
443) -> tuple[ int, list[ str ], set[ __.Path ] ]:
444 ''' Checks staleness for items of a specific type.
446 Renders from components/ and compares against distribution/.
447 Returns tuple of (items_checked, diff_lines, expected_paths).
448 '''
449 items_checked = 0
450 diffs: list[ str ] = [ ]
451 expected_paths: set[ __.Path ] = set( )
452 configuration_directory = (
453 generator.location / 'configurations' / item_type )
454 if not configuration_directory.exists( ): 454 ↛ 455line 454 didn't jump to line 455 because the condition on line 454 was never true
455 return ( items_checked, diffs, expected_paths )
456 for configuration_file in configuration_directory.glob( '*.toml' ):
457 item_name = configuration_file.stem
458 if not _content_exists( generator, item_type, item_name, coder ): 458 ↛ 459line 458 didn't jump to line 459 because the condition on line 458 was never true
459 continue
460 items_checked += 1
461 result = generator.render_single_item(
462 item_type, item_name, coder, distribution )
463 renderer = _renderers.RENDERERS[ coder ]
464 dirname = renderer.produce_output_structure( item_type )
465 output_path = (
466 distribution / 'per-project' / 'coders' / coder / dirname /
467 f"{item_name}.{_parse_output_extension( result.location )}" )
468 expected_paths.add( output_path )
469 if not output_path.exists( ):
470 diffs.append(
471 f"+ {item_type}/{item_name}: "
472 f"missing from distribution" )
473 continue
474 existing_content = output_path.read_text( encoding = 'utf-8' )
475 if result.content != existing_content: 475 ↛ 476line 475 didn't jump to line 476 because the condition on line 475 was never true
476 diff_lines = list( _difflib.unified_diff(
477 existing_content.splitlines( ),
478 result.content.splitlines( ),
479 fromfile = f"distribution/{item_type}/{output_path.name}",
480 tofile = f"components/{item_type}/{item_name}",
481 lineterm = '' ) )
482 diffs.extend( diff_lines )
483 return ( items_checked, diffs, expected_paths )
486def _detect_orphaned_artifacts(
487 distribution: __.Path,
488 coders: __.cabc.Sequence[ str ],
489 expected_paths: set[ __.Path ],
490) -> list[ str ]:
491 ''' Detects orphaned artifacts in distribution/.
493 Scans distribution/per-project/coders/<coder>/ for generated
494 item directories (commands, agents) and reports any files not
495 in the expected_paths set. Also scans legacy singular directory
496 names (command, agent) to detect stale artifacts from before the
497 plural cutover.
498 '''
499 orphans: list[ str ] = [ ]
500 generated_dirs = ( 'commands', 'agents' )
501 legacy_dirs = ( 'command', 'agent' )
502 for coder in coders:
503 coder_dir = distribution / 'per-project' / 'coders' / coder
504 if not coder_dir.exists( ): continue
505 for dirname in generated_dirs:
506 item_dir = coder_dir / dirname
507 if not item_dir.exists( ): continue 507 ↛ 505line 507 didn't jump to line 505 because the continue on line 507 wasn't executed
508 orphans.extend(
509 f"- {coder}/{dirname}/{item_file.name}: orphaned artifact"
510 for item_file in item_dir.glob( '*.md' )
511 if item_file not in expected_paths
512 )
513 for dirname in legacy_dirs:
514 item_dir = coder_dir / dirname
515 if not item_dir.exists( ): continue
516 orphans.extend(
517 f"- {coder}/{dirname}/{item_file.name}: "
518 f"stale legacy artifact (use plural '{dirname}s/' instead)"
519 for item_file in item_dir.glob( '*.md' )
520 )
521 return orphans