diff --git a/openhands-sdk/openhands/sdk/hooks/config.py b/openhands-sdk/openhands/sdk/hooks/config.py index 1d5a8bdd92..18d26cfbec 100644 --- a/openhands-sdk/openhands/sdk/hooks/config.py +++ b/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, diff --git a/tests/sdk/hooks/test_config.py b/tests/sdk/hooks/test_config.py index dc97cf65b6..f744fb7f21 100644 --- a/tests/sdk/hooks/test_config.py +++ b/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