Refactor code for consistency and add Pylint backlog

pyproject.toml

@@ -52,8 +52,18 @@ disable = [
     "all",
 ]
 enable = [
+    "C0303",
+    "C0305",
+    "C0325",
+    "C1804",
+    "C1805",
     "E",
     "F",
+    "R1716",
+    "R1705",
+    "R1721",
+    "R1724",
+    "R1732",
     "W0101",
     "W0102",
     "W0104",
@@ -183,3 +193,28 @@ enable = [
     "W4905",
     "W4906",
 ]
+
+# Pylint rollout backlog, ordered by priority: useful/simple first,
+# noisy/architectural last. Keep commented until the rule is clean and enabled.
+# 01. C0414 - useless import alias (1 finding, low risk)
+# 02. C0411 - wrong import order (5 findings, medium risk)
+# 03. C0413 - import position (5 findings, medium risk)
+# 04. R0916 - too many boolean expressions (3 findings, medium risk)
+# 05. C0103 - invalid naming style (13 findings, medium risk)
+# 06. R0903 - too few public methods (2 findings, medium risk)
+# 07. C0114 - missing module docstring (7 findings, low value)
+# 08. C0115 - missing class docstring (25 findings, boilerplate risk)
+# 09. C0116 - missing function docstring (73 findings, boilerplate risk)
+# 10. R1702 - too many nested blocks (26 findings, refactor risk)
+# 11. R0911 - too many return statements (47 findings, refactor risk)
+# 12. R0914 - too many local variables (123 findings, extraction risk)
+# 13. R0801 - duplicate code (39 findings, ownership/refactor risk)
+# 14. R0912 - too many branches (84 findings, high refactor risk)
+# 15. R0915 - too many statements (52 findings, high refactor risk)
+# 16. R0913 - too many arguments (9 findings, API churn risk)
+# 17. R0917 - too many positional arguments (5 findings, API churn risk)
+# 18. R0902 - too many instance attributes (18 findings, data model risk)
+# 19. C0302 - too many lines in module (13 findings, module split risk)
+# 20. C0415 - import outside top-level (240 findings, lazy import risk)
+# 21. R0401 - cyclic imports (16 findings, architecture risk)
+# 22. C0301 - line too long (221 findings, noisy formatting pass)

skills/studio/scripts/studio/cli.py

@@ -261,7 +261,7 @@ def main(argv: Optional[List[str]] = None) -> int:
             print(json.dumps({
                 "usage": "cfs <command> [options]",
                 "commands": _cmd_descriptions,
-                "sections": {name: cmds for name, cmds in _sections},
+                "sections": dict(_sections),
             }, indent=2, ensure_ascii=False))
         else:
             ui.header("Constructor Studio CLI")
@@ -317,77 +317,76 @@ def main(argv: Optional[List[str]] = None) -> int:
     # Dispatch to appropriate command handler
     if cmd == "validate":
         return _cmd_validate(rest)
-    elif cmd == "validate-code":
+    if cmd == "validate-code":
         # Legacy alias: keep for compatibility.
         return _cmd_validate(rest)
-    elif cmd in ("validate-kits", "validate-rules", "self-check"):
+    if cmd in ("validate-kits", "validate-rules", "self-check"):
         return _cmd_validate_kits(rest)
-    elif cmd == "init":
+    if cmd == "init":
         return _cmd_init(rest)
-    elif cmd == "update":
+    if cmd == "update":
         return _cmd_update(rest)
-    elif cmd == "list-ids":
+    if cmd == "list-ids":
         return _cmd_list_ids(rest)
-    elif cmd == "list-id-kinds":
+    if cmd == "list-id-kinds":
         return _cmd_list_id_kinds(rest)
-    elif cmd == "get-content":
+    if cmd == "get-content":
         return _cmd_get_content(rest)
-    elif cmd == "where-defined":
+    if cmd == "where-defined":
         return _cmd_where_defined(rest)
-    elif cmd == "where-used":
+    if cmd == "where-used":
         return _cmd_where_used(rest)
-    elif cmd == "info":
+    if cmd == "info":
         return _cmd_studio_info(rest)
-    elif cmd == "resolve-vars":
+    if cmd == "resolve-vars":
         return _cmd_resolve_vars(rest)
-    elif cmd == "agents":
+    if cmd == "agents":
         return _cmd_agents(rest)
-    elif cmd == "generate-agents":
+    if cmd == "generate-agents":
         return _cmd_generate_agents(rest)
-    elif cmd == "kit":
+    if cmd == "kit":
         return _cmd_kit(rest)
-    elif cmd == "generate-resources":
+    if cmd == "generate-resources":
         return _cmd_generate_resources(rest)
-    elif cmd == "toc":
+    if cmd == "toc":
         return _cmd_toc(rest)
-    elif cmd == "validate-toc":
+    if cmd == "validate-toc":
         return _cmd_validate_toc(rest)
-    elif cmd == "spec-coverage":
+    if cmd == "spec-coverage":
         return _cmd_spec_coverage(rest)
-    elif cmd == "chunk-input":
+    if cmd == "chunk-input":
         return _cmd_chunk_input(rest)
-    elif cmd == "workspace-init":
+    if cmd == "workspace-init":
         return _cmd_workspace_init(rest)
-    elif cmd == "workspace-add":
+    if cmd == "workspace-add":
         return _cmd_workspace_add(rest)
-    elif cmd == "workspace-info":
+    if cmd == "workspace-info":
         return _cmd_workspace_info(rest)
-    elif cmd == "workspace-sync":
+    if cmd == "workspace-sync":
         return _cmd_workspace_sync(rest)
-    elif cmd == "delegate":
+    if cmd == "delegate":
         return _cmd_delegate(rest)
-    elif cmd == "doctor":
+    if cmd == "doctor":
         return _cmd_doctor(rest)
-    elif cmd == "check-language":
+    if cmd == "check-language":
         return _cmd_check_language(rest)
-    elif cmd == "pdsl":
+    if cmd == "pdsl":
         return _cmd_pdsl(rest)
-    elif cmd == "map":
+    if cmd == "map":
         return _cmd_map(rest)
-    else:
-        # @cpt-begin:cpt-studio-algo-core-infra-route-command:p1:inst-if-no-handler
-        # @cpt-begin:cpt-studio-algo-core-infra-route-command:p1:inst-return-unknown
-        from .utils.ui import ui
-        ui.result(
-            {"status": "ERROR", "message": f"Unknown command: {cmd}", "available": all_commands},
-            human_fn=lambda d: (
-                ui.error(f"Unknown command: {cmd}"),
-                ui.hint(f"Available commands: {', '.join(all_commands)}"),
-                ui.hint("Run 'cfs --help' for usage."),
-            ),
-        )
-        return 1
-        # @cpt-end:cpt-studio-algo-core-infra-route-command:p1:inst-return-unknown
+    # @cpt-begin:cpt-studio-algo-core-infra-route-command:p1:inst-if-no-handler
+    # @cpt-begin:cpt-studio-algo-core-infra-route-command:p1:inst-return-unknown
+    from .utils.ui import ui
+    ui.result(
+        {"status": "ERROR", "message": f"Unknown command: {cmd}", "available": all_commands},
+        human_fn=lambda d: (
+            ui.error(f"Unknown command: {cmd}"),
+            ui.hint(f"Available commands: {', '.join(all_commands)}"),
+            ui.hint("Run 'cfs --help' for usage."),
+        ),
+    )
+    return 1
+    # @cpt-end:cpt-studio-algo-core-infra-route-command:p1:inst-return-unknown
         # @cpt-end:cpt-studio-algo-core-infra-route-command:p1:inst-if-no-handler
     # @cpt-end:cpt-studio-algo-core-infra-route-command:p1:inst-return-code
     # @cpt-end:cpt-studio-algo-core-infra-route-command:p1:inst-serialize-json

skills/studio/scripts/studio/commands/agents.py

@@ -4249,7 +4249,7 @@ def _confirm_v2_generation(
     """
     if args.dry_run:
         return False
-    if preview_create == 0 and preview_update == 0 and preview_delete == 0:
+    if not preview_create and not preview_update and not preview_delete:
         ui.info("No changes needed — agent files are up to date.")
         return False
     from ..utils.ui import is_json_mode
@@ -4600,7 +4600,7 @@ def cmd_generate_agents(argv: List[str]) -> int:
         )
         return 1
 
-    if total_create == 0 and total_update == 0 and total_delete == 0:
+    if not total_create and not total_update and not total_delete:
         from ..utils.ui import is_json_mode
         if is_json_mode():
             agents_result = _build_result(
@@ -4622,28 +4622,27 @@ def cmd_generate_agents(argv: List[str]) -> int:
             ui.info("No changes needed — agent files are up to date.")
         return 0
     # @cpt-end:cpt-studio-flow-agent-integration-generate:p1:inst-return-report
-    else:
-        from ..utils.ui import is_json_mode
-        if not is_json_mode():
-            auto_approve = getattr(args, "yes", False)
-            if not auto_approve:
-                _human_generate_agents_preview(agents_to_process, preview_results, project_root)
-            if not auto_approve and sys.stdin.isatty():
-                try:
-                    answer = input(
-                        "  Reply with `y` to write these generated files or `n` to abort.\n"
-                        "  Suggested: `y` when the previewed create/update set matches your intent.\n"
-                        "  `y` = continue with file generation. `n` = stop without writing.\n"
-                        "  Proceed? [Y/n] "
-                    ).strip().lower()
-                except (EOFError, KeyboardInterrupt):
-                    answer = "n"
-                if answer and answer not in ("y", "yes"):
-                    ui.result(
-                        {"status": "ABORTED", "message": "Cancelled by user"},
-                        human_fn=lambda d: (ui.warn("Aborted."), ui.blank()),
-                    )
-                    return 1
+    from ..utils.ui import is_json_mode
+    if not is_json_mode():
+        auto_approve = getattr(args, "yes", False)
+        if not auto_approve:
+            _human_generate_agents_preview(agents_to_process, preview_results, project_root)
+        if not auto_approve and sys.stdin.isatty():
+            try:
+                answer = input(
+                    "  Reply with `y` to write these generated files or `n` to abort.\n"
+                    "  Suggested: `y` when the previewed create/update set matches your intent.\n"
+                    "  `y` = continue with file generation. `n` = stop without writing.\n"
+                    "  Proceed? [Y/n] "
+                ).strip().lower()
+            except (EOFError, KeyboardInterrupt):
+                answer = "n"
+            if answer and answer not in ("y", "yes"):
+                ui.result(
+                    {"status": "ABORTED", "message": "Cancelled by user"},
+                    human_fn=lambda d: (ui.warn("Aborted."), ui.blank()),
+                )
+                return 1
 
     # Step 3: Execute the actual write
     # @cpt-begin:cpt-studio-flow-agent-integration-generate:p1:inst-for-each-agent

skills/studio/scripts/studio/commands/doctor.py

@@ -137,5 +137,3 @@ def _check_ralphex(project_root: Path) -> dict:
         "message": result["message"],
     }
 # @cpt-end:cpt-studio-algo-developer-experience-doctor:p2:inst-check-ralphex
-
-

skills/studio/scripts/studio/commands/kit.py

@@ -318,7 +318,7 @@ def _validate_tar_archive_before_extract(
                 )
 
     archive_size = tar_path.stat().st_size
-    if total_size > 0 and archive_size <= 0:
+    if archive_size <= 0 < total_size:
         raise RuntimeError(
             "Archive extraction blocked: invalid compressed archive size "
             f"({archive_size} bytes)"
@@ -893,7 +893,7 @@ def _collect_registered_kit_metadata(
             continue
         if kind == "skill":
             continue
-        elif kind == "rule":
+        if kind == "rule":
             try:
                 agents_parts.append(binding_abs.read_text(encoding="utf-8"))
             except OSError:
@@ -3195,7 +3195,7 @@ def cmd_kit_update(argv: List[str]) -> int:
                     "message": "All kits are up to date",
                 }, human_fn=_human_kit_update)
                 return 0
-            elif source_failures:
+            if source_failures:
                 ui.result({
                     "status": "FAIL",
                     "message": "All kits failed source resolution",
@@ -3310,7 +3310,7 @@ def cmd_kit_update(argv: List[str]) -> int:
     }
     if errors:
         output["errors"] = errors
-    if n_updated == 0 and not errors:
+    if not n_updated and not errors:
         output["message"] = "All kits are up to date"
 
     ui.result(output, human_fn=_human_kit_update)
@@ -4503,7 +4503,7 @@ def update_kit(
         # resource bindings). Bumping after `partial` (everything declined)
         # would mark the kit "current" against a remote it does not match,
         # silently hiding the pending update on the next `cfs kit update`.
-        bumped_safe_to_record = (ver_status != "partial")
+        bumped_safe_to_record = ver_status != "partial"
         if (source_version and bumped_safe_to_record) or _merged_resources:
             _kit_root_rel = registered_kit_path if _manifest is not None else ""
             # When all changes were declined, preserve the previously
@@ -4614,20 +4614,19 @@ def human_fn(_d: dict) -> tuple:
     # @cpt-begin:cpt-studio-flow-kit-dispatch:p1:inst-route
     if subcmd == "install":
         return cmd_kit_install(rest)
-    elif subcmd == "update":
+    if subcmd == "update":
         return cmd_kit_update(rest)
-    elif subcmd == "check-updates":
+    if subcmd == "check-updates":
         return cmd_kit_check_updates(rest)
-    elif subcmd == "validate":
+    if subcmd == "validate":
         from .validate_kits import cmd_validate_kits
         return cmd_validate_kits(rest)
-    elif subcmd == "normalize":
+    if subcmd == "normalize":
         return cmd_kit_normalize(rest)
-    elif subcmd == "migrate":
+    if subcmd == "migrate":
         return cmd_kit_migrate(rest)
-    else:
-        ui.result({"status": "ERROR", "message": f"Unknown kit subcommand: {subcmd}", "subcommands": subcommands})
-        return 1
+    ui.result({"status": "ERROR", "message": f"Unknown kit subcommand: {subcmd}", "subcommands": subcommands})
+    return 1
     # @cpt-end:cpt-studio-flow-kit-dispatch:p1:inst-route
 
 # ---------------------------------------------------------------------------
@@ -4910,7 +4909,7 @@ def _register_kit_in_core_toml(
         existing["install_mode"] = install_mode
     if source_provenance:
         existing["source_provenance"] = {
-            key: value for key, value in source_provenance.items() if value != ""
+            key: value for key, value in source_provenance.items() if value
         }
     if authority_metadata:
         # @cpt-begin:cpt-studio-algo-kit-github-version-authority:p1:inst-persist-authority-metadata
@@ -4937,7 +4936,7 @@ def _register_kit_in_core_toml(
             "freshness": authority_metadata.get("freshness", "unknown"),
         }
         existing["source_provenance"] = {
-            key: value for key, value in source_provenance.items() if value != ""
+            key: value for key, value in source_provenance.items() if value
         }
         authority_content_identity = {
             "vcs": authority_metadata.get("content_identity", {}).get("vcs", "") if isinstance(authority_metadata.get("content_identity"), dict) else "",
@@ -4947,7 +4946,7 @@ def _register_kit_in_core_toml(
             "identity": authority_metadata.get("identity", ""),
         }
         existing["content_identity"] = {
-            key: value for key, value in authority_content_identity.items() if value != ""
+            key: value for key, value in authority_content_identity.items() if value
         }
         authority_version = (
             authority_metadata.get("installed_version")

skills/studio/scripts/studio/commands/map/cli.py

@@ -270,7 +270,7 @@ def _load_template_vars(primary_root: Path) -> Dict[str, str]:
             )
         except Exception:  # pylint: disable=broad-exception-caught
             continue
-        if out.returncode != 0 or not out.stdout.strip():
+        if out.returncode or not out.stdout.strip():
             continue
         try:
             data = json.loads(out.stdout)

skills/studio/scripts/studio/commands/map/cpt_edges.py

@@ -82,7 +82,7 @@ def build_cpt_edges(nodes: Sequence[Node]) -> Tuple[List[Edge], List[Node]]:
                     continue  # self
                 to_id = target.id
                 dangling = False
-                cross_repo = (src.source != target.source)
+                cross_repo = src.source != target.source
 
             edge_type = "cpt-doc" if src.kind == "markdown" else "cpt-impl"
             key = (src.id, to_id, use.cpt_id)

skills/studio/scripts/studio/commands/map/scan.py

@@ -68,7 +68,7 @@ def _section_around(lines_raw, line_no: int) -> str:
     """
     # @cpt-begin:cpt-studio-algo-map-scan:p1:inst-section-around
     n = len(lines_raw)
-    if not (1 <= line_no <= n):
+    if not 1 <= line_no <= n:
         return ""
     target_idx = line_no - 1