refactor(prek): enhance type handling and improve workspace resolution

- Updated `run_main` function to utilize a generic type parameter for better type inference.
- Refactored workspace resolution functions to improve clarity and maintainability, including renaming `require_absolute_workspace_dir` to `resolve_absolute_workspace_path`.
- Enhanced CLI invocation handling by integrating new utility functions for workspace detection and command processing.
- Cleaned up various functions to streamline argument handling and improve overall code readability.

This update improves type safety and enhances the clarity of workspace management within the CLI, contributing to a more robust and maintainable codebase.

Author: qdrddr

src/cyt/cli/exit.py

@@ -5,13 +5,10 @@
 import sys
 from collections.abc import Callable
 from pathlib import Path
-from typing import TypeVar
 
 INTERRUPTED_EXIT_CODE = 130
 QUIET_STOP_EXIT_CODE = 1
 
-_T = TypeVar("_T")
-
 
 def parse_quiet_cli_flags(argv: list[str]) -> tuple[bool, Path | None]:
     """Parse ``--verbose`` and ``--config PATH`` from trailing stop flags."""
@@ -32,7 +29,7 @@ def parse_quiet_cli_flags(argv: list[str]) -> tuple[bool, Path | None]:
     return verbose, config_path
 
 
-def run_main(main_fn: Callable[..., _T], /, *args: object, **kwargs: object) -> _T:
+def run_main[T](main_fn: Callable[..., T], /, *args: object, **kwargs: object) -> T:
     """Run *main_fn* and exit silently on ``KeyboardInterrupt``."""
     try:
         return main_fn(*args, **kwargs)

src/cyt/cloudflare/runtime.py

@@ -91,23 +91,35 @@ def tools_hook_cloudflare_cache_settings(config: dict[str, Any] | None = None) -
 
 def tools_hook_cloudflare_url(config: dict[str, Any] | None = None) -> str:
     if _tools_hook_cloudflare_url is None:
-        return ""
+        from cyt.config import tools_hook_cloudflare_url as config_tools_hook_cloudflare_url
+
+        return config_tools_hook_cloudflare_url(config)
     return _tools_hook_cloudflare_url(config)
 
 
 def tools_hook_cloudflare_access_client_id_var(config: dict[str, Any] | None = None) -> str:
     if _tools_hook_cloudflare_access_client_id_var is None:
-        return "CF_ACCESS_CLIENT_ID"
+        from cyt.config import (
+            tools_hook_cloudflare_access_client_id_var as config_tools_hook_cloudflare_access_client_id_var,
+        )
+
+        return config_tools_hook_cloudflare_access_client_id_var(config)
     return _tools_hook_cloudflare_access_client_id_var(config)
 
 
 def tools_hook_cloudflare_access_client_secret_var(config: dict[str, Any] | None = None) -> str:
     if _tools_hook_cloudflare_access_client_secret_var is None:
-        return "CF_ACCESS_CLIENT_SECRET"
+        from cyt.config import (
+            tools_hook_cloudflare_access_client_secret_var as config_tools_hook_cloudflare_access_client_secret_var,
+        )
+
+        return config_tools_hook_cloudflare_access_client_secret_var(config)
     return _tools_hook_cloudflare_access_client_secret_var(config)
 
 
 def uses_cloudflare_tool_catalog(config: dict[str, Any] | None = None) -> bool:
     if _uses_cloudflare_tool_catalog is None:
-        return False
+        from cyt.config import uses_cloudflare_tool_catalog as config_uses_cloudflare_tool_catalog
+
+        return config_uses_cloudflare_tool_catalog(config)
     return _uses_cloudflare_tool_catalog(config)

src/cyt/hook/cli_invocation.py

@@ -53,10 +53,10 @@
     "INSTALLED_CYT_DAEMON_START_COMMAND_BASE",
     "INSTALLED_CYT_MCP_COMMAND",
     "HookCliInvocation",
+    "agent_hook_command_env",
     "build_installed_cyt_client_command",
     "build_installed_cyt_daemon_restart_command",
     "build_installed_cyt_daemon_start_command",
-    "agent_hook_command_env",
     "build_uv_run_dev_command",
     "cursor_hooks_dir",
     "cyt_cli_script_path",
@@ -69,9 +69,9 @@
     "cyt_mcp_mcp_server_entry",
     "detect_cyt_mcp_cli_invocation",
     "detect_hook_cli_invocation",
-    "prefix_agent_hook_command",
     "invoked_via_cyt_cli_script",
     "is_uv_run_dev_hook_command",
+    "prefix_agent_hook_command",
     "proxy_cli_script_path",
     "repo_root_from_cyt_cli_script",
     "repo_root_from_proxy_cli_script",
@@ -237,14 +237,14 @@ def prefix_command_env(env: dict[str, str], command: str) -> str:
             return command
         parts: list[str] = []
         for key, value in env.items():
-            if any(char in value for char in (' ', '"', "&", "|", "<", ">", "^")):
+            if any(char in value for char in (" ", '"', "&", "|", "<", ">", "^")):
                 escaped = value.replace('"', '""')
                 parts.append(f'set "{key}={escaped}"')
             else:
                 parts.append(f"set {key}={value}")
         if command.lower().endswith(".cmd"):
             tail = f'call "{command}"'
-        elif any(char in command for char in (' ', "&", "|", "<", ">")):
+        elif any(char in command for char in (" ", "&", "|", "<", ">")):
             tail = f'"{command}"'
         else:
             tail = command
@@ -287,8 +287,7 @@ def _windows_wrapper_env_lines(env: dict[str, str]) -> list[str]:
         if value == _WORKSPACE_FOLDER_TEMPLATE and key == CYT_WORKSPACE_ENV:
             for agent_var in _AGENT_WORKSPACE_ENV_VARS:
                 lines.append(
-                    f'if not defined {key} if defined {agent_var} '
-                    f'set "{key}=!{agent_var}!"',
+                    f'if not defined {key} if defined {agent_var} set "{key}=!{agent_var}!"',
                 )
             continue
         escaped = value.replace("%", "%%")

src/cyt/hook/setup_wizard.py

@@ -4,7 +4,6 @@
 
 import copy
 import json
-import os
 import sys
 from collections.abc import Iterator
 from pathlib import Path
@@ -47,6 +46,7 @@
     prefix_command_env,
     remove_windows_hook_wrappers,
 )
+from cyt.hook.install_scope import CytInstallScope
 from cyt.launch.inject_via_prompt import ensure_hook_inject_via
 from cyt.mcpc.readiness import report_mcpc_hook_readiness
 from cyt.platform.compat import is_windows
@@ -873,13 +873,21 @@ def _is_legacy_cyt_hook_stdin_command(command: str) -> bool:
     )
 
 
+def _command_invokes_cyt_client(command: str) -> bool:
+    normalized = command.strip()
+    if normalized == CYT_CLIENT_COMMAND or normalized.endswith(f" {CYT_CLIENT_COMMAND}"):
+        return True
+    parts = normalized.split()
+    return bool(parts) and Path(parts[-1]).name == CYT_CLIENT_COMMAND
+
+
 def _is_cyt_hook_command(command: object) -> bool:
     if not isinstance(command, str):
         return False
     normalized = command.strip()
     if is_windows_hook_wrapper_command(normalized):
         return True
-    if normalized == CYT_CLIENT_COMMAND or normalized.endswith(f" {CYT_CLIENT_COMMAND}"):
+    if _command_invokes_cyt_client(normalized):
         return True
     if is_dev_cyt_hook_command(normalized):
         return True
@@ -2526,27 +2534,12 @@ def _finish_hook_setup(
     _print_hook_uninstall_instructions()
 
 
-def run_hook_setup(
+def _prepare_hook_setup_workspace(
     *,
-    config_path: Path | None = None,
-    agents: list[HookAgentName] | None = None,
-    prevent_hallucinations: bool = False,
-    workspace: Path | None = None,
-) -> None:
-    """Install CYT agent hooks and ensure runtime credentials."""
-    selected_agents = _resolve_hook_setup_agents(agents)
-    if len(selected_agents) == 1:
-        print(f"CYT hook setup ({selected_agents[0]})\n")
-    else:
-        print("CYT hook setup\n")
-
-    resolved_config_path = resolve_setup_config_path(config_path)
-    from cyt.migrations.migrate import (
-        ensure_config_file_current,
-        ensure_workspace_config_current,
-    )
-
-    ensure_config_file_current(resolved_config_path, scope="user")
+    workspace: Path | None,
+    hook_agent: str,
+    selected_agents: list[HookAgentName],
+) -> tuple[Any, Path | None]:
     from cyt.hook.workspace_resolution import (
         WorkspaceResolutionConflictError,
         ensure_cyt_mcp_dev_wrapper,
@@ -2556,8 +2549,8 @@ def run_hook_setup(
         print_hook_setup_workspace_report,
         resolve_hook_setup_workspace,
     )
+    from cyt.migrations.migrate import ensure_workspace_config_current
 
-    hook_agent = selected_agents[0] if len(selected_agents) == 1 else "cursor"
     for agent in selected_agents:
         ensure_cyt_uv_wrapper_scripts(agent)
         ensure_cyt_mcp_dev_wrapper(agent)
@@ -2583,6 +2576,34 @@ def run_hook_setup(
                 f"Updated {settings_path} with terminal.integrated.env "
                 f"CYT_WORKSPACE (Windows/Linux/macOS)",
             )
+    return hook_workspace_resolution, consumer_root
+
+
+def run_hook_setup(
+    *,
+    config_path: Path | None = None,
+    agents: list[HookAgentName] | None = None,
+    prevent_hallucinations: bool = False,
+    workspace: Path | None = None,
+) -> None:
+    """Install CYT agent hooks and ensure runtime credentials."""
+    selected_agents = _resolve_hook_setup_agents(agents)
+    if len(selected_agents) == 1:
+        print(f"CYT hook setup ({selected_agents[0]})\n")
+    else:
+        print("CYT hook setup\n")
+
+    resolved_config_path = resolve_setup_config_path(config_path)
+    from cyt.migrations.migrate import ensure_config_file_current
+
+    ensure_config_file_current(resolved_config_path, scope="user")
+
+    hook_agent = selected_agents[0] if len(selected_agents) == 1 else "cursor"
+    hook_workspace_resolution, consumer_root = _prepare_hook_setup_workspace(
+        workspace=workspace,
+        hook_agent=hook_agent,
+        selected_agents=selected_agents,
+    )
     from cyt.hook.install_scope import CytInstallScope
     from cyt.hook.workspace_resolution import hook_setup_mcp_workspace_root
 

src/cyt/hook/workspace_config.py

@@ -73,10 +73,10 @@ def set_hook_workspace_in_config(
     if workspace is None:
         merged.pop(HOOK_WORKSPACE_CONFIG_KEY, None)
         return merged
-    from cyt.hook.workspace_resolution import require_absolute_workspace_dir
+    from cyt.hook.workspace_resolution import resolve_absolute_workspace_path
 
     merged[HOOK_WORKSPACE_CONFIG_KEY] = str(
-        require_absolute_workspace_dir(workspace, label="hook workspace"),
+        resolve_absolute_workspace_path(workspace, label="hook workspace"),
     )
     return merged