diff --git a/codewiki/src/be/caw_backend.py b/codewiki/src/be/caw_backend.py index 1db090a4..649aaa07 100644 --- a/codewiki/src/be/caw_backend.py +++ b/codewiki/src/be/caw_backend.py @@ -23,7 +23,7 @@ import logging import os import shutil -from typing import Any, Dict, List +from typing import Any from caw import Agent as CawAgent from caw import ToolGroup @@ -170,7 +170,7 @@ def _with_allowed_tools(cmd): allowed = ",".join(f"mcp__{s}" for s in servers) logger.info("Injected --allowedTools for MCP servers: %s", servers) return list(cmd) + ["--allowedTools", allowed] - except Exception as e: # never break the spawn on a patch hiccup + except Exception as e: # noqa: BLE001 — never break the spawn on a patch hiccup logger.warning("claude allowedTools patch skipped: %s", e) return cmd @@ -266,11 +266,11 @@ def complete( async def run_module_agent( self, module_name: str, - components: Dict[str, Node], - core_component_ids: List[str], - module_path: List[str], + components: dict[str, Node], + core_component_ids: list[str], + module_path: list[str], working_dir: str, - ) -> Dict[str, Any]: + ) -> dict[str, Any]: # caw.completion shells out to a subprocess and blocks the calling # thread. Push it off the event loop so the rest of the async # pipeline keeps moving. @@ -293,13 +293,13 @@ async def run_module_agent( def _run_module_agent_sync( self, module_name: str, - components: Dict[str, Node], - core_component_ids: List[str], - module_path: List[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, - ) -> Dict[str, Any]: + module_tree: dict[str, Any] | None = None, + ) -> dict[str, Any]: # ``start_depth`` lets the recursion preserve max_depth across nested # _run_module_agent_sync calls — each fresh deps object would otherwise # reset current_depth to 1 and silently bypass max_depth guards. @@ -314,10 +314,15 @@ def _run_module_agent_sync( if module_tree is None: module_tree = file_manager.load_json(module_tree_path) - overview_docs_path = os.path.join(working_dir, OVERVIEW_FILENAME) - if os.path.exists(overview_docs_path): - logger.info("✓ Overview docs already exists at %s", overview_docs_path) - return module_tree + # overview.md is the root module's own doc (renamed from + # {repo_name}.md), so its presence only proves the root is done — + # nested modules must still be checked against their own doc file, + # or a resume after a partial run silently skips every missing one. + if not module_path: + overview_docs_path = os.path.join(working_dir, OVERVIEW_FILENAME) + if os.path.exists(overview_docs_path): + logger.info("✓ Overview docs already exists at %s", overview_docs_path) + return module_tree docs_path = os.path.join(working_dir, f"{module_name}.md") if os.path.exists(docs_path): logger.info("✓ Module docs already exists at %s", docs_path) @@ -333,16 +338,16 @@ def _run_module_agent_sync( # agent call per sub-spec even when a single leaf write would suffice. # See generate_sub_module_documentation_tool for the pydantic-ai # equivalent. - _, components_with_code = format_potential_core_components( - core_component_ids, components - ) + _, components_with_code = format_potential_core_components(core_component_ids, components) num_tokens = count_tokens(components_with_code) can_delegate = ( is_complex_module(components, core_component_ids) and start_depth < config.max_depth and num_tokens >= config.max_token_per_leaf_module ) - 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}") + 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}" + ) if can_delegate: system_prompt = format_system_prompt(module_name, custom_instructions) diff --git a/codewiki/src/be/documentation_generator.py b/codewiki/src/be/documentation_generator.py index 654f0420..74ce15b7 100644 --- a/codewiki/src/be/documentation_generator.py +++ b/codewiki/src/be/documentation_generator.py @@ -297,13 +297,10 @@ async def generate_parent_module_docs( module_tree_path = os.path.join(working_dir, MODULE_TREE_FILENAME) module_tree = file_manager.load_json(module_tree_path) - # check if overview docs already exists - overview_docs_path = os.path.join(working_dir, OVERVIEW_FILENAME) - if os.path.exists(overview_docs_path): - logger.info(f"✓ Overview docs already exists at {overview_docs_path}") - return module_tree - - # check if parent docs already exists + # check if parent docs already exists (for the root module the doc + # path below resolves to overview.md, so a blanket "overview exists → + # skip" check is not needed here and would mask missing parent docs + # on resume) parent_docs_path = os.path.join( working_dir, f"{module_name if len(module_path) >= 1 else OVERVIEW_FILENAME.replace('.md', '')}.md", diff --git a/codewiki/src/be/prompt_template.py b/codewiki/src/be/prompt_template.py index 49783052..95e0b593 100644 --- a/codewiki/src/be/prompt_template.py +++ b/codewiki/src/be/prompt_template.py @@ -257,10 +257,32 @@ Reasoning at first, then return the list of relative paths in JSON format. """ +import logging +from collections import defaultdict from typing import Any from codewiki.src.utils import file_manager +logger = logging.getLogger(__name__) + +# codex rejects any turn whose total input exceeds 1,048,576 characters +# (input_too_large, code -32602 — server-side, not configurable). Cap the +# user prompt below that, leaving headroom for the system prompt, tool +# schemas and protocol overhead. +MAX_USER_PROMPT_CHARS = 900_000 + +MODULE_TREE_TRIMMED_NOTE = ( + "NOTE: per-module component listings were omitted because the full module " + "tree exceeds the model input limit. Module names and hierarchy are " + "complete; read the referenced modules' documentation files or use your " + "code-reading tools when you need component-level detail." +) + +CODE_TRUNCATED_NOTE = ( + "\n... [file contents truncated to fit the model input limit — use your " + "file-reading tools to read the full files]" +) + EXTENSION_TO_LANGUAGE = { ".py": "python", ".md": "markdown", @@ -289,6 +311,54 @@ } +def _format_module_tree_str( + module_tree: dict[str, Any], + current_module_name: str | None = None, + include_components: bool = True, +) -> str: + """ + Render a module tree as an indented text outline. + + With include_components=False only module names and hierarchy are + emitted, which keeps the outline small enough for huge trees that would + otherwise blow past MAX_USER_PROMPT_CHARS. + """ + lines: list[str] = [] + + def _walk(tree: dict[str, Any], indent: int = 0) -> None: + for key, value in tree.items(): + if key == current_module_name: + lines.append(f"{' ' * indent}{key} (current module)") + else: + lines.append(f"{' ' * indent}{key}") + + if include_components: + # Group components by file + by_file = defaultdict(list) + for c in value["components"]: + if "::" in c: + fpath, name = c.split("::", 1) + by_file[fpath].append(name) + else: + by_file[""].append(c) + for fpath, names in by_file.items(): + if fpath: + lines.append(f"{' ' * (indent + 1)} {fpath}: {', '.join(names)}") + else: + lines.append(f"{' ' * (indent + 1)} {', '.join(names)}") + + if ( + ("children" in value) + and isinstance(value["children"], dict) + and len(value["children"]) > 0 + ): + lines.append(f"{' ' * (indent + 1)} Children:") + _walk(value["children"], indent + 2) + + _walk(module_tree, 0) + return "\n".join(lines) + + def format_user_prompt( module_name: str, core_component_ids: list[str], @@ -306,41 +376,7 @@ def format_user_prompt( Returns: Formatted user prompt string """ - - # format module tree - lines = [] - - def _format_module_tree(module_tree: dict[str, any], indent: int = 0): - for key, value in module_tree.items(): - if key == module_name: - lines.append(f"{' ' * indent}{key} (current module)") - else: - lines.append(f"{' ' * indent}{key}") - - # Group components by file - from collections import defaultdict - - by_file = defaultdict(list) - for c in value["components"]: - if "::" in c: - fpath, name = c.split("::", 1) - by_file[fpath].append(name) - else: - by_file[""].append(c) - for fpath, names in by_file.items(): - if fpath: - lines.append(f"{' ' * (indent + 1)} {fpath}: {', '.join(names)}") - else: - lines.append(f"{' ' * (indent + 1)} {', '.join(names)}") - - if isinstance(value["children"], dict) and len(value["children"]) > 0: - lines.append(f"{' ' * (indent + 1)} Children:") - _format_module_tree(value["children"], indent + 2) - - _format_module_tree(module_tree, 0) - formatted_module_tree = "\n".join(lines) - - # print(f"Formatted module tree:\n{formatted_module_tree}") + formatted_module_tree = _format_module_tree_str(module_tree, module_name) # Group core component IDs by their file path grouped_components: dict[str, list[str]] = {} @@ -375,12 +411,56 @@ def _format_module_tree(module_tree: dict[str, any], indent: int = 0): core_component_codes += "```\n\n" - return USER_PROMPT.format( + prompt = USER_PROMPT.format( module_name=module_name, formatted_core_component_codes=core_component_codes, module_tree=formatted_module_tree, ) + if len(prompt) > MAX_USER_PROMPT_CHARS: + full_len = len(prompt) + formatted_module_tree = ( + MODULE_TREE_TRIMMED_NOTE + + "\n\n" + + _format_module_tree_str(module_tree, module_name, include_components=False) + ) + prompt = USER_PROMPT.format( + module_name=module_name, + formatted_core_component_codes=core_component_codes, + module_tree=formatted_module_tree, + ) + logger.warning( + "Module %s: user prompt (%d chars) exceeds %d; " + "module tree trimmed to names only (%d chars)", + module_name, + full_len, + MAX_USER_PROMPT_CHARS, + len(prompt), + ) + + if len(prompt) > MAX_USER_PROMPT_CHARS: + # Even the slim tree was not enough — the inlined file contents + # dominate. Truncation is recoverable: the agent has file-reading + # tools (read_code_components / str_replace_editor). + excess = len(prompt) - MAX_USER_PROMPT_CHARS + len(CODE_TRUNCATED_NOTE) + core_component_codes = ( + core_component_codes[: max(0, len(core_component_codes) - excess)] + CODE_TRUNCATED_NOTE + ) + prompt = USER_PROMPT.format( + module_name=module_name, + formatted_core_component_codes=core_component_codes, + module_tree=formatted_module_tree, + ) + logger.warning( + "Module %s: user prompt still over %d chars after tree trim; " + "truncated inlined file contents (now %d chars)", + module_name, + MAX_USER_PROMPT_CHARS, + len(prompt), + ) + + return prompt + def format_cluster_prompt( potential_core_components: str, @@ -393,53 +473,38 @@ def format_cluster_prompt( if module_tree is None: module_tree = {} - # format module tree - lines = [] - - # print(f"Module tree:\n{json.dumps(module_tree, indent=2)}") - - def _format_module_tree(module_tree: dict[str, any], indent: int = 0): - for key, value in module_tree.items(): - if key == module_name: - lines.append(f"{' ' * indent}{key} (current module)") - else: - lines.append(f"{' ' * indent}{key}") - - # Group components by file - from collections import defaultdict - - by_file = defaultdict(list) - for c in value["components"]: - if "::" in c: - fpath, name = c.split("::", 1) - by_file[fpath].append(name) - else: - by_file[""].append(c) - for fpath, names in by_file.items(): - if fpath: - lines.append(f"{' ' * (indent + 1)} {fpath}: {', '.join(names)}") - else: - lines.append(f"{' ' * (indent + 1)} {', '.join(names)}") - - if ( - ("children" in value) - and isinstance(value["children"], dict) - and len(value["children"]) > 0 - ): - lines.append(f"{' ' * (indent + 1)} Children:") - _format_module_tree(value["children"], indent + 2) - - _format_module_tree(module_tree, 0) - formatted_module_tree = "\n".join(lines) - if module_tree == {}: return CLUSTER_REPO_PROMPT.format(potential_core_components=potential_core_components) - else: - return CLUSTER_MODULE_PROMPT.format( + + formatted_module_tree = _format_module_tree_str(module_tree, module_name) + prompt = CLUSTER_MODULE_PROMPT.format( + potential_core_components=potential_core_components, + module_tree=formatted_module_tree, + module_name=module_name, + ) + + if len(prompt) > MAX_USER_PROMPT_CHARS: + full_len = len(prompt) + formatted_module_tree = ( + MODULE_TREE_TRIMMED_NOTE + + "\n\n" + + _format_module_tree_str(module_tree, module_name, include_components=False) + ) + prompt = CLUSTER_MODULE_PROMPT.format( potential_core_components=potential_core_components, module_tree=formatted_module_tree, module_name=module_name, ) + logger.warning( + "Module %s: cluster prompt (%d chars) exceeds %d; " + "module tree trimmed to names only (%d chars)", + module_name, + full_len, + MAX_USER_PROMPT_CHARS, + len(prompt), + ) + + return prompt def format_super_group_prompt(module_tree: dict[str, Any]) -> str: diff --git a/codewiki/src/be/pydantic_ai_backend.py b/codewiki/src/be/pydantic_ai_backend.py index 38c45268..3b3ddea9 100644 --- a/codewiki/src/be/pydantic_ai_backend.py +++ b/codewiki/src/be/pydantic_ai_backend.py @@ -11,7 +11,7 @@ import logging import os import traceback -from typing import Any, Dict, List +from typing import Any from pydantic_ai import Agent @@ -55,19 +55,24 @@ def complete( async def run_module_agent( self, module_name: str, - components: Dict[str, Node], - core_component_ids: List[str], - module_path: List[str], + components: dict[str, Node], + core_component_ids: list[str], + module_path: list[str], working_dir: str, - ) -> Dict[str, Any]: + ) -> dict[str, Any]: config = self._config module_tree_path = os.path.join(working_dir, MODULE_TREE_FILENAME) module_tree = file_manager.load_json(module_tree_path) - overview_docs_path = os.path.join(working_dir, OVERVIEW_FILENAME) - if os.path.exists(overview_docs_path): - logger.info("✓ Overview docs already exists at %s", overview_docs_path) - return module_tree + # overview.md is the root module's own doc (renamed from + # {repo_name}.md), so its presence only proves the root is done — + # nested modules must still be checked against their own doc file, + # or a resume after a partial run silently skips every missing one. + if not module_path: + overview_docs_path = os.path.join(working_dir, OVERVIEW_FILENAME) + if os.path.exists(overview_docs_path): + logger.info("✓ Overview docs already exists at %s", overview_docs_path) + return module_tree docs_path = os.path.join(working_dir, f"{module_name}.md") if os.path.exists(docs_path): logger.info("✓ Module docs already exists at %s", docs_path)