🧹 [Code Health] Reduce cyclomatic complexity in main function

🎯 What: Refactored the `main` function in `src/main.py` by extracting command handling logic into individual helper functions and using a dispatch dictionary (`COMMAND_HANDLERS`).
💡 Why: The `main` function previously contained a long, complex chain of `if args.command == ...:` statements, resulting in high cyclomatic complexity. This refactoring greatly improves maintainability, readability, and testability by separating concerns and reducing the cognitive load required to understand the entry point. It also optimizes performance by lazily building `manifest` only for commands that need it.
✅ Verification: Ran `ruff format` and `ruff check` to ensure clean code. Executed `PYTHONPATH=. python3 -m unittest discover tests` to confirm no existing functionality was broken.
✨ Result: The `main` function is now clean and concise, and the code health issue regarding cyclomatic complexity is fully resolved.

Co-authored-by: google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com>

Author: badMade

.github/scripts/check_doc_source_of_truth.py

@@ -7,39 +7,39 @@
 
 ROOT = Path(__file__).resolve().parents[2]
 FILES = [
-    ROOT / 'README.md',
-    ROOT / 'USAGE.md',
-    ROOT / 'PARITY.md',
-    ROOT / 'PHILOSOPHY.md',
-    ROOT / 'ROADMAP.md',
-    ROOT / '.github' / 'FUNDING.yml',
+    ROOT / "README.md",
+    ROOT / "USAGE.md",
+    ROOT / "PARITY.md",
+    ROOT / "PHILOSOPHY.md",
+    ROOT / "ROADMAP.md",
+    ROOT / ".github" / "FUNDING.yml",
 ]
-FILES.extend(sorted((ROOT / 'docs').rglob('*.md')) if (ROOT / 'docs').exists() else [])
+FILES.extend(sorted((ROOT / "docs").rglob("*.md")) if (ROOT / "docs").exists() else [])
 
 FORBIDDEN = {
-    r'github\.com/Yeachan-Heo/claw-code(?!-parity)': 'replace old claw-code GitHub links with ultraworkers/claw-code',
-    r'github\.com/code-yeongyu/claw-code': 'replace stale alternate claw-code GitHub links with ultraworkers/claw-code',
-    r'discord\.gg/6ztZB9jvWq': 'replace the stale UltraWorkers Discord invite with the current invite',
-    r'api\.star-history\.com/svg\?repos=Yeachan-Heo/claw-code': 'update star-history embeds to ultraworkers/claw-code',
-    r'star-history\.com/#Yeachan-Heo/claw-code': 'update star-history links to ultraworkers/claw-code',
-    r'assets/clawd-hero\.jpeg': 'rename stale hero asset references to assets/claw-hero.jpeg',
-    r'assets/instructkr\.png': 'remove stale instructkr image references',
+    r"github\.com/Yeachan-Heo/claw-code(?!-parity)": "replace old claw-code GitHub links with ultraworkers/claw-code",
+    r"github\.com/code-yeongyu/claw-code": "replace stale alternate claw-code GitHub links with ultraworkers/claw-code",
+    r"discord\.gg/6ztZB9jvWq": "replace the stale UltraWorkers Discord invite with the current invite",
+    r"api\.star-history\.com/svg\?repos=Yeachan-Heo/claw-code": "update star-history embeds to ultraworkers/claw-code",
+    r"star-history\.com/#Yeachan-Heo/claw-code": "update star-history links to ultraworkers/claw-code",
+    r"assets/clawd-hero\.jpeg": "rename stale hero asset references to assets/claw-hero.jpeg",
+    r"assets/instructkr\.png": "remove stale instructkr image references",
 }
 
 errors: list[str] = []
 for path in FILES:
     if not path.exists():
         continue
-    text = path.read_text(encoding='utf-8')
+    text = path.read_text(encoding="utf-8")
     for pattern, message in FORBIDDEN.items():
         for match in re.finditer(pattern, text):
-            line = text.count('\n', 0, match.start()) + 1
-            errors.append(f'{path.relative_to(ROOT)}:{line}: {message}')
+            line = text.count("\n", 0, match.start()) + 1
+            errors.append(f"{path.relative_to(ROOT)}:{line}: {message}")
 
 if errors:
-    print('doc source-of-truth check failed:', file=sys.stderr)
+    print("doc source-of-truth check failed:", file=sys.stderr)
     for error in errors:
-        print(f'  - {error}', file=sys.stderr)
+        print(f"  - {error}", file=sys.stderr)
     sys.exit(1)
 
-print('doc source-of-truth check passed')
+print("doc source-of-truth check passed")

patch_pr.py

@@ -1,17 +1,19 @@
 import sys
 
+
 def replace_in_file(filepath, search_str, replace_str):
-    with open(filepath, 'r') as f:
+    with open(filepath, "r") as f:
         content = f.read()
 
     if search_str in content:
         content = content.replace(search_str, replace_str)
-        with open(filepath, 'w') as f:
+        with open(filepath, "w") as f:
             f.write(content)
         print("Replacement successful.")
     else:
         print("Search string not found.")
 
+
 search = """fn agent_detail(agent: &AgentSummary) -> String {
     let mut parts = vec![agent.name.as_str()];
     if let Some(description) = &agent.description {
@@ -39,4 +41,4 @@ def replace_in_file(filepath, search_str, replace_str):
     parts.join(" · ")
 }"""
 
-replace_in_file('rust/crates/commands/src/lib.rs', search, replace)
+replace_in_file("rust/crates/commands/src/lib.rs", search, replace)

rust/scripts/run_mock_parity_diff.py

@@ -66,9 +66,11 @@ def main() -> int:
 
     should_run = "--no-run" not in sys.argv[1:]
     report = run_harness(rust_root) if should_run else None
-    report_by_name = {
-        entry["name"]: entry for entry in report.get("scenarios", [])
-    } if report else {}
+    report_by_name = (
+        {entry["name"]: entry for entry in report.get("scenarios", [])}
+        if report
+        else {}
+    )
 
     print("Mock parity diff checklist")
     print(f"Repo root: {repo_root}")
@@ -79,7 +81,9 @@ def main() -> int:
     for entry in manifest:
         scenario_name = entry["name"]
         scenario_report = report_by_name.get(scenario_name)
-        status = "PASS" if scenario_report else ("MAPPED" if not should_run else "MISSING")
+        status = (
+            "PASS" if scenario_report else ("MAPPED" if not should_run else "MISSING")
+        )
         print(f"[{status}] {scenario_name} ({entry['category']})")
         print(f"  description: {entry['description']}")
         print(f"  parity refs: {' | '.join(entry['parity_refs'])}")

src/QueryEngine.py

@@ -7,13 +7,16 @@
 class QueryEngineRuntime(QueryEnginePort):
     def route(self, prompt: str, limit: int = 5) -> str:
         matches = PortRuntime().route_prompt(prompt, limit=limit)
-        lines = ['# Query Engine Route', '', f'Prompt: {prompt}', '']
+        lines = ["# Query Engine Route", "", f"Prompt: {prompt}", ""]
         if not matches:
-            lines.append('No mirrored command/tool matches found.')
-            return '\n'.join(lines)
-        lines.append('Matches:')
-        lines.extend(f'- [{match.kind}] {match.name} ({match.score}) — {match.source_hint}' for match in matches)
-        return '\n'.join(lines)
+            lines.append("No mirrored command/tool matches found.")
+            return "\n".join(lines)
+        lines.append("Matches:")
+        lines.extend(
+            f"- [{match.kind}] {match.name} ({match.score}) — {match.source_hint}"
+            for match in matches
+        )
+        return "\n".join(lines)
 
 
-__all__ = ['QueryEnginePort', 'QueryEngineRuntime']
+__all__ = ["QueryEnginePort", "QueryEngineRuntime"]

src/Tool.py

@@ -10,6 +10,6 @@ class ToolDefinition:
 
 
 DEFAULT_TOOLS = (
-    ToolDefinition('port_manifest', 'Summarize the active Python workspace'),
-    ToolDefinition('query_engine', 'Render a Python-first porting summary'),
+    ToolDefinition("port_manifest", "Summarize the active Python workspace"),
+    ToolDefinition("query_engine", "Render a Python-first porting summary"),
 )

src/__init__.py

@@ -10,20 +10,20 @@
 from .tools import PORTED_TOOLS, build_tool_backlog
 
 __all__ = [
-    'ParityAuditResult',
-    'PortManifest',
-    'PortRuntime',
-    'QueryEnginePort',
-    'RuntimeSession',
-    'StoredSession',
-    'TurnResult',
-    'PORTED_COMMANDS',
-    'PORTED_TOOLS',
-    'build_command_backlog',
-    'build_port_manifest',
-    'build_system_init_message',
-    'build_tool_backlog',
-    'load_session',
-    'run_parity_audit',
-    'save_session',
+    "ParityAuditResult",
+    "PortManifest",
+    "PortRuntime",
+    "QueryEnginePort",
+    "RuntimeSession",
+    "StoredSession",
+    "TurnResult",
+    "PORTED_COMMANDS",
+    "PORTED_TOOLS",
+    "build_command_backlog",
+    "build_port_manifest",
+    "build_system_init_message",
+    "build_tool_backlog",
+    "load_session",
+    "run_parity_audit",
+    "save_session",
 ]

src/bootstrap_graph.py

@@ -8,20 +8,20 @@ class BootstrapGraph:
     stages: tuple[str, ...]
 
     def as_markdown(self) -> str:
-        lines = ['# Bootstrap Graph', '']
-        lines.extend(f'- {stage}' for stage in self.stages)
-        return '\n'.join(lines)
+        lines = ["# Bootstrap Graph", ""]
+        lines.extend(f"- {stage}" for stage in self.stages)
+        return "\n".join(lines)
 
 
 def build_bootstrap_graph() -> BootstrapGraph:
     return BootstrapGraph(
         stages=(
-            'top-level prefetch side effects',
-            'warning handler and environment guards',
-            'CLI parser and pre-action trust gate',
-            'setup() + commands/agents parallel load',
-            'deferred init after trust',
-            'mode routing: local / remote / ssh / teleport / direct-connect / deep-link',
-            'query engine submit loop',
+            "top-level prefetch side effects",
+            "warning handler and environment guards",
+            "CLI parser and pre-action trust gate",
+            "setup() + commands/agents parallel load",
+            "deferred init after trust",
+            "mode routing: local / remote / ssh / teleport / direct-connect / deep-link",
+            "query engine submit loop",
         )
     )

src/command_graph.py

@@ -17,18 +17,29 @@ def flattened(self) -> tuple[PortingModule, ...]:
 
     def as_markdown(self) -> str:
         lines = [
-            '# Command Graph',
-            '',
-            f'Builtins: {len(self.builtins)}',
-            f'Plugin-like commands: {len(self.plugin_like)}',
-            f'Skill-like commands: {len(self.skill_like)}',
+            "# Command Graph",
+            "",
+            f"Builtins: {len(self.builtins)}",
+            f"Plugin-like commands: {len(self.plugin_like)}",
+            f"Skill-like commands: {len(self.skill_like)}",
         ]
-        return '\n'.join(lines)
+        return "\n".join(lines)
 
 
 def build_command_graph() -> CommandGraph:
     commands = get_commands()
-    builtins = tuple(module for module in commands if 'plugin' not in module.source_hint.lower() and 'skills' not in module.source_hint.lower())
-    plugin_like = tuple(module for module in commands if 'plugin' in module.source_hint.lower())
-    skill_like = tuple(module for module in commands if 'skills' in module.source_hint.lower())
-    return CommandGraph(builtins=builtins, plugin_like=plugin_like, skill_like=skill_like)
+    builtins = tuple(
+        module
+        for module in commands
+        if "plugin" not in module.source_hint.lower()
+        and "skills" not in module.source_hint.lower()
+    )
+    plugin_like = tuple(
+        module for module in commands if "plugin" in module.source_hint.lower()
+    )
+    skill_like = tuple(
+        module for module in commands if "skills" in module.source_hint.lower()
+    )
+    return CommandGraph(
+        builtins=builtins, plugin_like=plugin_like, skill_like=skill_like
+    )

src/context.py

@@ -18,30 +18,34 @@ class PortContext:
 
 def build_port_context(base: Path | None = None) -> PortContext:
     root = base or Path(__file__).resolve().parent.parent
-    source_root = root / 'src'
-    tests_root = root / 'tests'
-    assets_root = root / 'assets'
-    archive_root = root / 'archive' / 'claude_code_ts_snapshot' / 'src'
+    source_root = root / "src"
+    tests_root = root / "tests"
+    assets_root = root / "assets"
+    archive_root = root / "archive" / "claude_code_ts_snapshot" / "src"
     return PortContext(
         source_root=source_root,
         tests_root=tests_root,
         assets_root=assets_root,
         archive_root=archive_root,
-        python_file_count=sum(1 for path in source_root.rglob('*.py') if path.is_file()),
-        test_file_count=sum(1 for path in tests_root.rglob('*.py') if path.is_file()),
-        asset_file_count=sum(1 for path in assets_root.rglob('*') if path.is_file()),
+        python_file_count=sum(
+            1 for path in source_root.rglob("*.py") if path.is_file()
+        ),
+        test_file_count=sum(1 for path in tests_root.rglob("*.py") if path.is_file()),
+        asset_file_count=sum(1 for path in assets_root.rglob("*") if path.is_file()),
         archive_available=archive_root.exists(),
     )
 
 
 def render_context(context: PortContext) -> str:
-    return '\n'.join([
-        f'Source root: {context.source_root}',
-        f'Test root: {context.tests_root}',
-        f'Assets root: {context.assets_root}',
-        f'Archive root: {context.archive_root}',
-        f'Python files: {context.python_file_count}',
-        f'Test files: {context.test_file_count}',
-        f'Assets: {context.asset_file_count}',
-        f'Archive available: {context.archive_available}',
-    ])
+    return "\n".join(
+        [
+            f"Source root: {context.source_root}",
+            f"Test root: {context.tests_root}",
+            f"Assets root: {context.assets_root}",
+            f"Archive root: {context.archive_root}",
+            f"Python files: {context.python_file_count}",
+            f"Test files: {context.test_file_count}",
+            f"Assets: {context.asset_file_count}",
+            f"Archive available: {context.archive_available}",
+        ]
+    )

src/cost_tracker.py

@@ -10,4 +10,4 @@ class CostTracker:
 
     def record(self, label: str, units: int) -> None:
         self.total_units += units
-        self.events.append(f'{label}:{units}')
+        self.events.append(f"{label}:{units}")