Coverage for sources/agentsmgr/operations.py: 76%
243 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''' 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 73 ↛ exitline 73 didn't return from function '_ensure_output_directories' because the return on line 73 wasn't executed
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 77 ↛ 79line 77 didn't jump to line 79 because
78 else: actual_mode = generator.mode
79 if actual_mode not in ( 'per-user', 'per-project' ): continue 79 ↛ 74line 79 didn't jump to line 74 because the continue on line 79 wasn't executed
80 if actual_mode not in renderer.modes_available: continue 80 ↛ 74line 80 didn't jump to line 74 because the continue on line 80 wasn't executed
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': 128 ↛ 129line 128 didn't jump to line 129 because the condition on line 128 was never true
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_text( result.content, result.location, simulate ): 144 ↛ 134line 144 didn't jump to line 134 because the condition on line 144 was always true
145 items_written += 1
146 return ( items_attempted, items_written )
149def save_content_text(
150 content: str, location: __.Path, simulate: bool = False
151) -> bool:
152 ''' Saves text 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 return save_content_bytes(
159 content.encode( 'utf-8' ), location, simulate = simulate )
162def save_content_bytes(
163 content: bytes, location: __.Path, simulate: bool = False
164) -> bool:
165 ''' Saves bytes to location, creating parent directories as needed.
167 Binary-safe write for skill assets and other non-text artifacts.
168 In simulation mode, no actual writing occurs. Returns True if file
169 was written, False if simulated.
170 '''
171 if simulate: return False 171 ↛ exitline 171 didn't return from function 'save_content_bytes' because the return on line 171 wasn't executed
172 try: location.parent.mkdir( parents = True, exist_ok = True )
173 except ( OSError, IOError ) as exception:
174 raise _exceptions.FileOperationFailure(
175 location.parent, "create directory" ) from exception
176 try: location.write_bytes( content )
177 except ( OSError, IOError ) as exception:
178 raise _exceptions.FileOperationFailure(
179 location, "save content" ) from exception
180 return True
183def update_git_exclude(
184 target: __.Path,
185 entries: __.cabc.Collection[ str ],
186 simulate: bool = False
187) -> int:
188 ''' Updates .git/info/exclude with managed block of agentsmgr entries.
190 Maintains a clearly-marked block of entries managed by agentsmgr,
191 with complete replacement on each update. Entries are sorted
192 lexicographically within the block. User entries outside the
193 managed block are preserved.
195 Uses repository discovery from the explicit target and uses the
196 common git directory for shared resources in worktrees.
198 Returns count of entries in managed block.
199 '''
200 if simulate: return 0 200 ↛ exitline 200 didn't return from function 'update_git_exclude' because the return on line 200 wasn't executed
201 git_dir = _resolve_git_directory( target )
202 if not git_dir: return 0
203 exclude_file = git_dir / 'info' / 'exclude'
204 if not exclude_file.exists( ): return 0 204 ↛ exitline 204 didn't return from function 'update_git_exclude' because the return on line 204 wasn't executed
205 try: content = exclude_file.read_text( encoding = 'utf-8' )
206 except ( OSError, IOError ) as exception:
207 raise _exceptions.FileOperationFailure(
208 exclude_file, "read git exclude file" ) from exception
209 normalized_entries = sorted( {
210 _normalize_git_exclude_entry( entry )
211 for entry in entries if entry.strip( )
212 } )
213 if not normalized_entries: 213 ↛ 214line 213 didn't jump to line 214 because the condition on line 213 was never true
214 new_content = _remove_managed_block( content )
215 if new_content == content: return 0
216 else:
217 new_content = _update_managed_block( content, normalized_entries )
218 try: exclude_file.write_text( new_content, encoding = 'utf-8' )
219 except ( OSError, IOError ) as exception:
220 raise _exceptions.FileOperationFailure(
221 exclude_file, "update git exclude file" ) from exception
222 return len( normalized_entries )
225def _normalize_git_exclude_entry( entry: str ) -> str:
226 ''' Normalizes a managed git exclude entry. '''
227 normalized = entry.strip( ).replace( '\\', '/' )
228 if not normalized.startswith( '/' ):
229 normalized = f"/{normalized}"
230 return normalized
233def _update_managed_block(
234 content: str, entries: __.cabc.Sequence[ str ]
235) -> str:
236 ''' Updates content with new managed block containing sorted entries.
238 Locates existing managed block (if present) and replaces it with
239 new block. If no block exists, appends to end of file. Preserves
240 user content outside the managed block.
241 '''
242 lines = content.splitlines( )
243 before_block, after_block = _partition_around_managed_block( lines )
244 block_lines = [ _MANAGED_BLOCK_BEGIN, _MANAGED_BLOCK_WARNING ]
245 block_lines.extend( entries )
246 block_lines.append( _MANAGED_BLOCK_END )
247 if before_block and before_block[ -1 ].strip( ): 247 ↛ 249line 247 didn't jump to line 249 because the condition on line 247 was always true
248 before_block.append( '' )
249 result_lines = before_block + block_lines
250 if after_block: 250 ↛ 251line 250 didn't jump to line 251 because the condition on line 250 was never true
251 result_lines.append( '' )
252 result_lines.extend( after_block )
253 return '\n'.join( result_lines ) + '\n'
256def _remove_managed_block( content: str ) -> str:
257 ''' Removes managed block from content, preserving user entries.
259 Locates and removes managed block if present. Returns content
260 unchanged if no block found.
261 '''
262 lines = content.splitlines( )
263 before_block, after_block = _partition_around_managed_block( lines )
264 if not before_block and not after_block:
265 return content
266 result_lines = before_block
267 if result_lines and after_block:
268 if result_lines[ -1 ].strip( ):
269 result_lines.append( '' )
270 result_lines.extend( after_block )
271 elif after_block:
272 result_lines = after_block
273 if not result_lines: return ''
274 return '\n'.join( result_lines ) + '\n'
277def _partition_around_managed_block(
278 lines: __.cabc.Sequence[ str ]
279) -> tuple[ list[ str ], list[ str ] ]:
280 ''' Partitions lines into content before and after managed block.
282 Locates managed block markers and returns (before, after) tuple.
283 If block is malformed or not found, returns (all_lines, []).
284 Malformed blocks are treated as non-existent.
285 '''
286 try: begin_index = lines.index( _MANAGED_BLOCK_BEGIN )
287 except ValueError: return ( list( lines ), [ ] )
288 try: end_index = lines.index( _MANAGED_BLOCK_END, begin_index )
289 except ValueError:
290 __.provide_scribe( __name__ ).warning(
291 "Malformed agentsmgr block in .git/info/exclude; rebuilding." )
292 return ( list( lines ), [ ] )
293 if end_index < begin_index: 293 ↛ 294line 293 didn't jump to line 294 because the condition on line 293 was never true
294 __.provide_scribe( __name__ ).warning(
295 "Malformed agentsmgr block in .git/info/exclude; rebuilding." )
296 return ( list( lines ), [ ] )
297 before_block = list( lines[ :begin_index ] )
298 while before_block and not before_block[ -1 ].strip( ):
299 before_block.pop( )
300 after_block = list( lines[ end_index + 1: ] )
301 while after_block and not after_block[ 0 ].strip( ): 301 ↛ 302line 301 didn't jump to line 302 because the condition on line 301 was never true
302 after_block.pop( 0 )
303 return ( before_block, after_block )
305def _resolve_git_directory(
306 start_path: __.Path
307) -> __.typx.Optional[ __.Path ]:
308 ''' Resolves git directory location, handling worktrees.
310 Uses Dulwich to discover the repository from the explicit target.
311 Returns common git directory (shared across worktrees) for access
312 to shared resources like info/exclude.
314 Returns None if not in a git repository or on error.
315 '''
316 from dulwich.repo import Repo
317 try: repo = Repo.discover( str( start_path ) )
318 except Exception: return None
319 git_dir_path = __.Path( repo.controldir( ) )
320 return _discover_common_git_directory( git_dir_path )
322def _discover_common_git_directory( git_dir: __.Path ) -> __.Path:
323 ''' Discovers common git directory, handling worktree commondir.
325 For worktrees, reads commondir file to find shared resources.
326 For standard repos, returns git_dir unchanged.
327 '''
328 commondir_file = git_dir / 'commondir'
329 if not commondir_file.exists( ): 329 ↛ 331line 329 didn't jump to line 331 because the condition on line 329 was always true
330 return git_dir
331 try: common_path = commondir_file.read_text( encoding = 'utf-8' ).strip( )
332 except ( OSError, IOError ): return git_dir
333 return ( git_dir / common_path ).resolve( )
336def generate_distribution(
337 generator: _generator.ContentGenerator,
338 distribution: __.Path,
339 simulate: bool = False,
340) -> tuple[ int, int ]:
341 ''' Generates pre-rendered artifacts from components/ to distribution/.
343 Reads from the 3-tier pipeline source (configurations, templates,
344 per-coder contents) and writes rendered commands and agents to
345 distribution/. Skills are not generated; they are direct
346 distribution artifacts.
348 Returns tuple of (items_attempted, items_written).
349 '''
350 items_attempted = 0
351 items_written = 0
352 for coder_name in generator.configuration[ 'coders' ]:
353 try: renderer = _renderers.RENDERERS[ coder_name ]
354 except KeyError: continue
355 for item_type in renderer.item_types_available:
356 if item_type == 'skills':
357 continue # Skills are direct distribution artifacts.
358 attempted, written = _generate_for_distribution(
359 generator, coder_name, item_type, distribution, simulate )
360 items_attempted += attempted
361 items_written += written
362 return ( items_attempted, items_written )
365def _generate_for_distribution(
366 generator: _generator.ContentGenerator,
367 coder: str,
368 item_type: str,
369 distribution: __.Path,
370 simulate: bool,
371) -> tuple[ int, int ]:
372 ''' Generates items of a type for a coder into distribution/.
374 Reads configuration and content from components/, renders through
375 the 3-tier pipeline, and writes to
376 distribution/per-project/coders/<coder>/<item_type>/.
377 Returns tuple of (items_attempted, items_written).
378 '''
379 items_attempted = 0
380 items_written = 0
381 configuration_directory = (
382 generator.location / 'configurations' / item_type )
383 if not configuration_directory.exists( ): 383 ↛ 384line 383 didn't jump to line 384 because the condition on line 383 was never true
384 return ( items_attempted, items_written )
385 for configuration_file in configuration_directory.glob( '*.toml' ):
386 item_name = configuration_file.stem
387 if not _content_exists( generator, item_type, item_name, coder ): 387 ↛ 388line 387 didn't jump to line 388 because the condition on line 387 was never true
388 __.provide_scribe( __name__ ).warning(
389 f"Skipping {item_type}/{item_name} for {coder}: "
390 "content not found" )
391 continue
392 items_attempted += 1
393 result = generator.render_single_item(
394 item_type, item_name, coder, distribution )
395 renderer = _renderers.RENDERERS[ coder ]
396 dirname = renderer.produce_output_structure( item_type )
397 output_path = (
398 distribution / 'per-project' / 'coders' / coder / dirname /
399 f"{item_name}.{_parse_output_extension( result.location )}" )
400 if save_content_text( result.content, output_path, simulate ): 400 ↛ 385line 400 didn't jump to line 385 because the condition on line 400 was always true
401 items_written += 1
402 return ( items_attempted, items_written )
405def _parse_output_extension( location: __.Path ) -> str:
406 ''' Extracts output extension from rendered location path.
408 Skips the last suffix (which is the item name suffix) and returns
409 the meaningful extension. For SKILL.md returns "md".
410 '''
411 name = location.name
412 if name == 'SKILL.md': return 'md' 412 ↛ exitline 412 didn't return from function '_parse_output_extension' because the return on line 412 wasn't executed
413 parts = name.split( '.' )
414 if len( parts ) >= _EXTENSION_PARTS_MINIMUM: return parts[ -1 ] 414 ↛ 415line 414 didn't jump to line 415 because the condition on line 414 was always true
415 return 'md'
418def check_distribution_staleness(
419 generator: _generator.ContentGenerator,
420 distribution: __.Path,
421) -> tuple[ int, list[ str ] ]:
422 ''' Checks for staleness between components/ and distribution/.
424 Regenerates from components/ and compares against existing
425 distribution/ files. Also detects orphaned artifacts that exist
426 in distribution/ but are no longer generated from components/.
427 Returns tuple of (items_checked, diff_lines).
428 Empty diff_lines means distribution is current.
429 '''
430 items_checked = 0
431 all_diffs: list[ str ] = [ ]
432 expected_paths: set[ __.Path ] = set( )
433 for coder_name in generator.configuration[ 'coders' ]:
434 try: renderer = _renderers.RENDERERS[ coder_name ]
435 except KeyError: continue
436 for item_type in renderer.item_types_available:
437 if item_type == 'skills':
438 continue
439 checked, diffs, paths = _check_staleness_for_type(
440 generator, coder_name, item_type, distribution )
441 items_checked += checked
442 all_diffs.extend( diffs )
443 expected_paths.update( paths )
444 # Detect orphaned artifacts in generated directories
445 orphans = _detect_orphaned_artifacts(
446 distribution, generator.configuration[ 'coders' ], expected_paths )
447 all_diffs.extend( orphans )
448 return ( items_checked, all_diffs )
451def _check_staleness_for_type(
452 generator: _generator.ContentGenerator,
453 coder: str,
454 item_type: str,
455 distribution: __.Path,
456) -> tuple[ int, list[ str ], set[ __.Path ] ]:
457 ''' Checks staleness for items of a specific type.
459 Renders from components/ and compares against distribution/.
460 Returns tuple of (items_checked, diff_lines, expected_paths).
461 '''
462 items_checked = 0
463 diffs: list[ str ] = [ ]
464 expected_paths: set[ __.Path ] = set( )
465 configuration_directory = (
466 generator.location / 'configurations' / item_type )
467 if not configuration_directory.exists( ): 467 ↛ 468line 467 didn't jump to line 468 because the condition on line 467 was never true
468 return ( items_checked, diffs, expected_paths )
469 for configuration_file in configuration_directory.glob( '*.toml' ):
470 item_name = configuration_file.stem
471 if not _content_exists( generator, item_type, item_name, coder ): 471 ↛ 472line 471 didn't jump to line 472 because the condition on line 471 was never true
472 continue
473 items_checked += 1
474 result = generator.render_single_item(
475 item_type, item_name, coder, distribution )
476 renderer = _renderers.RENDERERS[ coder ]
477 dirname = renderer.produce_output_structure( item_type )
478 output_path = (
479 distribution / 'per-project' / 'coders' / coder / dirname /
480 f"{item_name}.{_parse_output_extension( result.location )}" )
481 expected_paths.add( output_path )
482 if not output_path.exists( ):
483 diffs.append(
484 f"+ {item_type}/{item_name}: "
485 f"missing from distribution" )
486 continue
487 existing_content = output_path.read_text( encoding = 'utf-8' )
488 if result.content != existing_content: 488 ↛ 489line 488 didn't jump to line 489 because the condition on line 488 was never true
489 diff_lines = list( _difflib.unified_diff(
490 existing_content.splitlines( ),
491 result.content.splitlines( ),
492 fromfile = f"distribution/{item_type}/{output_path.name}",
493 tofile = f"components/{item_type}/{item_name}",
494 lineterm = '' ) )
495 diffs.extend( diff_lines )
496 return ( items_checked, diffs, expected_paths )
499def _detect_orphaned_artifacts(
500 distribution: __.Path,
501 coders: __.cabc.Sequence[ str ],
502 expected_paths: set[ __.Path ],
503) -> list[ str ]:
504 ''' Detects orphaned artifacts in distribution/.
506 For each coder, asks its renderer which item directory names
507 it owns (via ``calculate_directory_location``) and which file
508 glob pattern the renderer uses for artifacts of that type (via
509 ``calculate_artifact_pattern``). Reports any files matching the
510 pattern that are not in expected_paths.
511 '''
512 orphans: list[ str ] = [ ]
513 for coder in coders:
514 try: renderer = _renderers.RENDERERS[ coder ]
515 except KeyError: continue
516 coder_dir = distribution / 'per-project' / 'coders' / coder
517 if not coder_dir.exists( ): continue
518 for item_type in renderer.item_types_available:
519 if item_type == 'skills': continue
520 dirname = renderer.calculate_directory_location( item_type )
521 item_dir = coder_dir / dirname
522 if not item_dir.exists( ): continue
523 pattern = renderer.calculate_artifact_pattern( item_type )
524 orphans.extend(
525 f"- {coder}/{dirname}/{item_file.name}: orphaned artifact"
526 for item_file in item_dir.glob( pattern )
527 if item_file not in expected_paths
528 )
529 return orphans