(
self,
module_name: str,
components: Dict[str, Node],
core_component_ids: List[str],
module_path: List[str],
working_dir: str,
start_depth: int = 1,
module_tree: Dict[str, Any] | None = None,
)
| 204 | ) |
| 205 | |
| 206 | def _run_module_agent_sync( |
| 207 | self, |
| 208 | module_name: str, |
| 209 | components: Dict[str, Node], |
| 210 | core_component_ids: List[str], |
| 211 | module_path: List[str], |
| 212 | working_dir: str, |
| 213 | start_depth: int = 1, |
| 214 | module_tree: Dict[str, Any] | None = None, |
| 215 | ) -> Dict[str, Any]: |
| 216 | # ``start_depth`` lets the recursion preserve max_depth across nested |
| 217 | # _run_module_agent_sync calls — each fresh deps object would otherwise |
| 218 | # reset current_depth to 1 and silently bypass max_depth guards. |
| 219 | # ``module_tree`` carries the parent's in-memory tree across the |
| 220 | # recursion. Reloading from disk only works at the top level — by the |
| 221 | # time a sub-agent runs, the parent has staged new branches in memory |
| 222 | # but has not yet saved (save happens after agent.completion returns). |
| 223 | from codewiki.src.be.caw_toolkit import CawToolKit # local import to avoid cycles |
| 224 | |
| 225 | config = self._config |
| 226 | module_tree_path = os.path.join(working_dir, MODULE_TREE_FILENAME) |
| 227 | if module_tree is None: |
| 228 | module_tree = file_manager.load_json(module_tree_path) |
| 229 | |
| 230 | overview_docs_path = os.path.join(working_dir, OVERVIEW_FILENAME) |
| 231 | if os.path.exists(overview_docs_path): |
| 232 | logger.info("✓ Overview docs already exists at %s", overview_docs_path) |
| 233 | return module_tree |
| 234 | docs_path = os.path.join(working_dir, f"{module_name}.md") |
| 235 | if os.path.exists(docs_path): |
| 236 | logger.info("✓ Module docs already exists at %s", docs_path) |
| 237 | return module_tree |
| 238 | |
| 239 | custom_instructions = config.get_prompt_addition() |
| 240 | |
| 241 | # Mirror PydanticAIBackend's early-cut: a module is only worth |
| 242 | # delegating to sub-agents when it spans multiple files AND has enough |
| 243 | # content to justify the cost AND we still have recursion budget. |
| 244 | # Without this gate the caw path would give every multi-file sub-module |
| 245 | # the recursive SYSTEM_PROMPT + delegation tool and fan out one extra |
| 246 | # agent call per sub-spec even when a single leaf write would suffice. |
| 247 | # See generate_sub_module_documentation_tool for the pydantic-ai |
| 248 | # equivalent. |
| 249 | _, components_with_code = format_potential_core_components( |
| 250 | core_component_ids, components |
| 251 | ) |
| 252 | num_tokens = count_tokens(components_with_code) |
| 253 | can_delegate = ( |
| 254 | is_complex_module(components, core_component_ids) |
| 255 | and start_depth < config.max_depth |
| 256 | and num_tokens >= config.max_token_per_leaf_module |
| 257 | ) |
| 258 | logger.info(f"Module {module_name} can delegate: {can_delegate} - is_complex_module: {is_complex_module(components, core_component_ids)} - start_depth: {start_depth} - num_tokens: {num_tokens} - max_depth: {config.max_depth} - max_token_per_leaf_module: {config.max_token_per_leaf_module}") |
| 259 | |
| 260 | if can_delegate: |
| 261 | system_prompt = format_system_prompt(module_name, custom_instructions) |
| 262 | else: |
| 263 | system_prompt = format_leaf_system_prompt(module_name, custom_instructions) |
no test coverage detected