feat(hooks): Support file-based hook scripts in .openhands directory

openhands-sdk/openhands/sdk/hooks/config.py

@@ -1,5 +1,7 @@
 """Hook configuration loading and management."""
 
+from __future__ import annotations
+
 import json
 import logging
 import re
@@ -212,41 +214,94 @@ def _normalize_hooks_input(cls, data: Any) -> Any:
 
     @classmethod
     def load(
-        cls, path: str | Path | None = None, working_dir: str | Path | None = None
-    ) -> "HookConfig":
+        cls,
+        path: str | Path | None = None,
+        working_dir: str | Path | list[str | Path] | None = None,
+    ) -> HookConfig:
         """Load config from path or search .openhands/hooks.json locations.
 
         Args:
-            path: Explicit path to hooks.json file. If provided, working_dir is ignored.
-            working_dir: Project directory for discovering .openhands/hooks.json.
-                Falls back to cwd if not provided.
+            path: Explicit path to hooks.json file. If provided, only this file
+                is loaded (working_dir is still used for additional configs).
+            working_dir: Project directory (or list of directories) for discovering
+                .openhands/hooks.json. When a list is provided, hooks from all
+                directories are merged. Falls back to cwd if not provided.
+
+        The method loads hooks from:
+        1. The explicit path if provided
+        2. Each working directory's .openhands/hooks.json
+        3. ~/.openhands/hooks.json (global user config)
+
+        All discovered hooks are merged together.
         """
+        configs_to_merge: list[HookConfig] = []
+
+        # Handle explicit path if provided
+        if path is not None:
+            path = Path(path)
+            if path.exists():
+                try:
+                    with open(path) as f:
+                        data = json.load(f)
+                    json_config = cls.model_validate(data)
+                    if not json_config.is_empty():
+                        configs_to_merge.append(json_config)
+                except (json.JSONDecodeError, OSError) as e:
+                    logger.warning(f"Failed to load hooks from {path}: {e}")
+
+        # Normalize working_dir to a list of paths
+        if working_dir is None:
+            working_dirs = [Path.cwd()]
+        elif isinstance(working_dir, list):
+            working_dirs = [Path(d) for d in working_dir]
+        else:
+            working_dirs = [Path(working_dir)]
+
+        # Load hooks.json from each working directory
+        for base_dir in working_dirs:
+            hooks_json_path = base_dir / ".openhands" / "hooks.json"
+            if hooks_json_path.exists():
+                try:
+                    with open(hooks_json_path) as f:
+                        data = json.load(f)
+                    dir_config = cls.model_validate(data)
+                    if not dir_config.is_empty():
+                        configs_to_merge.append(dir_config)
+                except (json.JSONDecodeError, OSError) as e:
+                    logger.warning(f"Failed to load hooks from {hooks_json_path}: {e}")
+
+        # Load global user config from ~/.openhands/hooks.json
+        # Only load if no explicit path was provided
         if path is None:
-            # Search for hooks.json in standard locations
-            base_dir = Path(working_dir) if working_dir else Path.cwd()
-            search_paths = [
-                base_dir / ".openhands" / "hooks.json",
-                Path.home() / ".openhands" / "hooks.json",
-            ]
-            for search_path in search_paths:
-                if search_path.exists():
-                    path = search_path
-                    break
-
-        if path is None:
+            home_hooks_path = Path.home() / ".openhands" / "hooks.json"
+            # Check if we already loaded from home directory via working_dirs
+            home_already_loaded = any(
+                (wd / ".openhands" / "hooks.json").resolve()
+                == home_hooks_path.resolve()
+                for wd in working_dirs
+                if (wd / ".openhands" / "hooks.json").exists()
+            )
+            if not home_already_loaded and home_hooks_path.exists():
+                try:
+                    with open(home_hooks_path) as f:
+                        data = json.load(f)
+                    home_config = cls.model_validate(data)
+                    if not home_config.is_empty():
+                        configs_to_merge.append(home_config)
+                except (json.JSONDecodeError, OSError) as e:
+                    logger.warning(f"Failed to load hooks from {home_hooks_path}: {e}")
+
+        if not configs_to_merge:
             return cls()
 
-        path = Path(path)
-        if not path.exists():
-            return cls()
+        if len(configs_to_merge) == 1:
+            return configs_to_merge[0]
 
-        with open(path) as f:
-            data = json.load(f)
-        # Use model_validate which triggers the model_validator
-        return cls.model_validate(data)
+        merged = cls.merge(configs_to_merge)
+        return merged if merged is not None else cls()
 
     @classmethod
-    def from_dict(cls, data: dict[str, Any]) -> "HookConfig":
+    def from_dict(cls, data: dict[str, Any]) -> HookConfig:
         """Create HookConfig from a dictionary.
 
         Supports both legacy format with "hooks" wrapper and direct format:
@@ -290,7 +345,7 @@ def save(self, path: str | Path) -> None:
             json.dump(self.model_dump(mode="json", exclude_defaults=True), f, indent=2)
 
     @classmethod
-    def merge(cls, configs: list["HookConfig"]) -> "HookConfig | None":
+    def merge(cls, configs: list[HookConfig]) -> HookConfig | None:
         """Merge multiple hook configs by concatenating handlers per event type.
 
         Each hook config may have multiple event types (pre_tool_use,

tests/sdk/hooks/test_config.py

@@ -1,6 +1,7 @@
 """Tests for hook configuration loading and management."""
 
 import json
+import os
 import tempfile
 
 from openhands.sdk.hooks.config import HookConfig, HookDefinition, HookMatcher
@@ -229,3 +230,174 @@ def test_unknown_event_type_raises_error(self):
             HookConfig.from_dict(
                 {"PreToolExecute": [{"hooks": [{"command": "test.sh"}]}]}
             )
+
+
+class TestMultipleWorkingDirs:
+    """Tests for loading hooks from multiple working directories."""
+
+    def test_load_with_list_of_working_dirs(self):
+        """Test that load() accepts a list of working directories."""
+        with (
+            tempfile.TemporaryDirectory() as tmpdir1,
+            tempfile.TemporaryDirectory() as tmpdir2,
+        ):
+            # Create hooks.json in first directory
+            openhands_dir1 = os.path.join(tmpdir1, ".openhands")
+            os.makedirs(openhands_dir1)
+            with open(os.path.join(openhands_dir1, "hooks.json"), "w") as f:
+                json.dump({"PreToolUse": [{"hooks": [{"command": "hook1.sh"}]}]}, f)
+
+            # Create hooks.json in second directory
+            openhands_dir2 = os.path.join(tmpdir2, ".openhands")
+            os.makedirs(openhands_dir2)
+            with open(os.path.join(openhands_dir2, "hooks.json"), "w") as f:
+                json.dump({"PostToolUse": [{"hooks": [{"command": "hook2.sh"}]}]}, f)
+
+            # Load from both directories
+            config = HookConfig.load(working_dir=[tmpdir1, tmpdir2])
+
+            # Should have hooks from both directories
+            assert config.has_hooks_for_event(HookEventType.PRE_TOOL_USE)
+            assert config.has_hooks_for_event(HookEventType.POST_TOOL_USE)
+
+            pre_hooks = config.get_hooks_for_event(
+                HookEventType.PRE_TOOL_USE, "AnyTool"
+            )
+            assert len(pre_hooks) == 1
+            assert pre_hooks[0].command == "hook1.sh"
+
+            post_hooks = config.get_hooks_for_event(
+                HookEventType.POST_TOOL_USE, "AnyTool"
+            )
+            assert len(post_hooks) == 1
+            assert post_hooks[0].command == "hook2.sh"
+
+    def test_load_merges_hooks_from_multiple_dirs(self):
+        """Test that hooks from multiple directories are merged."""
+        with (
+            tempfile.TemporaryDirectory() as tmpdir1,
+            tempfile.TemporaryDirectory() as tmpdir2,
+        ):
+            # Create PreToolUse hook in first directory
+            openhands_dir1 = os.path.join(tmpdir1, ".openhands")
+            os.makedirs(openhands_dir1)
+            with open(os.path.join(openhands_dir1, "hooks.json"), "w") as f:
+                json.dump({"PreToolUse": [{"hooks": [{"command": "hook1.sh"}]}]}, f)
+
+            # Create another PreToolUse hook in second directory
+            openhands_dir2 = os.path.join(tmpdir2, ".openhands")
+            os.makedirs(openhands_dir2)
+            with open(os.path.join(openhands_dir2, "hooks.json"), "w") as f:
+                json.dump({"PreToolUse": [{"hooks": [{"command": "hook2.sh"}]}]}, f)
+
+            # Load from both directories
+            config = HookConfig.load(working_dir=[tmpdir1, tmpdir2])
+
+            # Should have both hooks merged
+            hooks = config.get_hooks_for_event(HookEventType.PRE_TOOL_USE, "AnyTool")
+            assert len(hooks) == 2
+            commands = [h.command for h in hooks]
+            assert "hook1.sh" in commands
+            assert "hook2.sh" in commands
+
+    def test_load_with_empty_list_returns_empty_config(self):
+        """Test that load() with empty list returns empty config."""
+        config = HookConfig.load(working_dir=[])
+        assert config.is_empty()
+
+    def test_load_with_single_item_list(self):
+        """Test that load() works with a single-item list."""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            openhands_dir = os.path.join(tmpdir, ".openhands")
+            os.makedirs(openhands_dir)
+            with open(os.path.join(openhands_dir, "hooks.json"), "w") as f:
+                json.dump({"PreToolUse": [{"hooks": [{"command": "hook.sh"}]}]}, f)
+
+            # Load with single-item list
+            config = HookConfig.load(working_dir=[tmpdir])
+
+            assert config.has_hooks_for_event(HookEventType.PRE_TOOL_USE)
+
+    def test_load_with_nonexistent_dirs_in_list(self):
+        """Test that load() handles nonexistent directories in the list gracefully."""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            openhands_dir = os.path.join(tmpdir, ".openhands")
+            os.makedirs(openhands_dir)
+            with open(os.path.join(openhands_dir, "hooks.json"), "w") as f:
+                json.dump({"PreToolUse": [{"hooks": [{"command": "hook.sh"}]}]}, f)
+
+            # Load with mix of existing and nonexistent directories
+            config = HookConfig.load(working_dir=["/nonexistent/path", tmpdir])
+
+            # Should still load hooks from the existing directory
+            assert config.has_hooks_for_event(HookEventType.PRE_TOOL_USE)
+
+
+class TestHomeDirDeduplication:
+    """Tests for preventing duplicate loading of ~/.openhands/hooks.json."""
+
+    def test_home_dir_not_loaded_twice_with_multiple_working_dirs(
+        self, tmp_path, monkeypatch
+    ):
+        """Test ~/.openhands/hooks.json is only loaded once with multiple dirs."""
+        # Create a fake home directory with hooks.json
+        fake_home = tmp_path / "fake_home"
+        fake_home.mkdir()
+        fake_openhands = fake_home / ".openhands"
+        fake_openhands.mkdir()
+        hooks_file = fake_openhands / "hooks.json"
+        hooks_file.write_text(
+            json.dumps({"PreToolUse": [{"hooks": [{"command": "home-hook.sh"}]}]})
+        )
+
+        # Monkeypatch Path.home() to return our fake home
+        monkeypatch.setattr("pathlib.Path.home", lambda: fake_home)
+
+        # Create two working directories without hooks
+        work_dir1 = tmp_path / "work1"
+        work_dir1.mkdir()
+        work_dir2 = tmp_path / "work2"
+        work_dir2.mkdir()
+
+        # Load from both working directories
+        config = HookConfig.load(working_dir=[str(work_dir1), str(work_dir2)])
+
+        # Should have exactly one hook from home directory (not duplicated)
+        hooks = config.get_hooks_for_event(HookEventType.PRE_TOOL_USE, "AnyTool")
+        assert len(hooks) == 1
+        assert hooks[0].command == "home-hook.sh"
+
+    def test_home_dir_merged_with_working_dir_hooks(self, tmp_path, monkeypatch):
+        """Test that home dir hooks are merged with working dir hooks."""
+        # Create a fake home directory with hooks.json
+        fake_home = tmp_path / "fake_home"
+        fake_home.mkdir()
+        fake_openhands = fake_home / ".openhands"
+        fake_openhands.mkdir()
+        home_hooks_file = fake_openhands / "hooks.json"
+        home_hooks_file.write_text(
+            json.dumps({"PreToolUse": [{"hooks": [{"command": "home-hook.sh"}]}]})
+        )
+
+        # Monkeypatch Path.home() to return our fake home
+        monkeypatch.setattr("pathlib.Path.home", lambda: fake_home)
+
+        # Create a working directory with its own hooks
+        work_dir = tmp_path / "work"
+        work_dir.mkdir()
+        work_openhands = work_dir / ".openhands"
+        work_openhands.mkdir()
+        work_hooks_file = work_openhands / "hooks.json"
+        work_hooks_file.write_text(
+            json.dumps({"PreToolUse": [{"hooks": [{"command": "work-hook.sh"}]}]})
+        )
+
+        # Load from working directory
+        config = HookConfig.load(working_dir=str(work_dir))
+
+        # Should have both hooks merged
+        hooks = config.get_hooks_for_event(HookEventType.PRE_TOOL_USE, "AnyTool")
+        assert len(hooks) == 2
+        commands = [h.command for h in hooks]
+        assert "work-hook.sh" in commands
+        assert "home-hook.sh" in commands