diff --git a/src/deepwork/core/command_executor.py b/src/deepwork/core/command_executor.py
index 629b4f50..74288a13 100644
--- a/src/deepwork/core/command_executor.py
+++ b/src/deepwork/core/command_executor.py
@@ -159,15 +159,32 @@ def all_commands_succeeded(results: list[CommandResult]) -> bool:
return all(r.success for r in results)
-def format_command_errors(results: list[CommandResult]) -> str:
- """Format error messages from failed commands."""
+def format_command_errors(
+ results: list[CommandResult],
+ rule_name: str | None = None,
+) -> str:
+ """Format detailed error messages from failed commands.
+
+ Args:
+ results: List of command execution results
+ rule_name: Optional rule name to include in error message
+
+ Returns:
+ Formatted error message with command, exit code, stdout, and stderr
+ """
errors: list[str] = []
for result in results:
if not result.success:
- msg = f"Command failed: {result.command}\n"
- if result.stderr:
- msg += f"Error: {result.stderr}\n"
- if result.exit_code != 0:
- msg += f"Exit code: {result.exit_code}\n"
- errors.append(msg)
- return "\n".join(errors)
+ parts: list[str] = []
+ if rule_name:
+ parts.append(f"Rule: {rule_name}")
+ parts.append(f"Command: {result.command}")
+ parts.append(f"Exit code: {result.exit_code}")
+ if result.stdout and result.stdout.strip():
+ parts.append(f"Stdout:\n{result.stdout.strip()}")
+ if result.stderr and result.stderr.strip():
+ parts.append(f"Stderr:\n{result.stderr.strip()}")
+ if not result.stdout.strip() and not result.stderr.strip():
+ parts.append("(no output)")
+ errors.append("\n".join(parts))
+ return "\n\n".join(errors)
diff --git a/src/deepwork/hooks/rules_check.py b/src/deepwork/hooks/rules_check.py
index 024b94ad..3d71dcce 100644
--- a/src/deepwork/hooks/rules_check.py
+++ b/src/deepwork/hooks/rules_check.py
@@ -654,10 +654,10 @@ def rules_check_hook(hook_input: HookInput) -> HookOutput:
),
)
else:
- # Command failed
- error_msg = format_command_errors(cmd_results)
- skip_hint = f"To skip, include `✓ {rule.name}` in your response.\n"
- command_errors.append(f"## {rule.name}\n{error_msg}{skip_hint}")
+ # Command failed - format detailed error message
+ error_msg = format_command_errors(cmd_results, rule_name=rule.name)
+ skip_hint = f"\nTo skip, include `✓ {rule.name}` in your response."
+ command_errors.append(f"{error_msg}{skip_hint}")
queue.update_status(
trigger_hash,
QueueEntryStatus.FAILED,
@@ -694,17 +694,33 @@ def rules_check_hook(hook_input: HookInput) -> HookOutput:
def main() -> None:
"""Entry point for the rules check hook."""
- # Determine platform from environment
platform_str = os.environ.get("DEEPWORK_HOOK_PLATFORM", "claude")
try:
platform = Platform(platform_str)
except ValueError:
platform = Platform.CLAUDE
- # Run the hook with the wrapper
exit_code = run_hook(rules_check_hook, platform)
sys.exit(exit_code)
if __name__ == "__main__":
- main()
+ # Wrap entry point to catch early failures (e.g., import errors in wrapper.py)
+ try:
+ main()
+ except Exception as e:
+ # Last resort error handling - output JSON manually since wrapper may be broken
+ import json
+ import traceback
+
+ error_output = {
+ "decision": "block",
+ "reason": (
+ "## Hook Script Error\n\n"
+ f"Error type: {type(e).__name__}\n"
+ f"Error: {e}\n\n"
+ f"Traceback:\n```\n{traceback.format_exc()}\n```"
+ ),
+ }
+ print(json.dumps(error_output))
+ sys.exit(0)
diff --git a/src/deepwork/hooks/wrapper.py b/src/deepwork/hooks/wrapper.py
index ef20899c..96dba3a9 100644
--- a/src/deepwork/hooks/wrapper.py
+++ b/src/deepwork/hooks/wrapper.py
@@ -321,6 +321,55 @@ def write_stdout(data: str) -> None:
print(data)
+def format_hook_error(
+ error: Exception,
+ context: str = "",
+) -> dict[str, Any]:
+ """
+ Format an error into a blocking JSON response with detailed information.
+
+ This is used when the hook script itself fails, to provide useful
+ error information to the user instead of a generic "non-blocking status code" message.
+
+ Args:
+ error: The exception that occurred
+ context: Additional context about where the error occurred
+
+ Returns:
+ Dict with decision="block" and detailed error message
+ """
+ import traceback
+
+ error_type = type(error).__name__
+ error_msg = str(error)
+ tb = traceback.format_exc()
+
+ parts = ["## Hook Script Error", ""]
+ if context:
+ parts.append(f"Context: {context}")
+ parts.append(f"Error type: {error_type}")
+ parts.append(f"Error: {error_msg}")
+ parts.append("")
+ parts.append("Traceback:")
+ parts.append(f"```\n{tb}\n```")
+
+ return {
+ "decision": "block",
+ "reason": "\n".join(parts),
+ }
+
+
+def output_hook_error(error: Exception, context: str = "") -> None:
+ """
+ Output a hook error as JSON to stdout.
+
+ Use this in exception handlers to ensure the hook always outputs
+ valid JSON even when crashing.
+ """
+ error_dict = format_hook_error(error, context)
+ print(json.dumps(error_dict))
+
+
def run_hook(
hook_fn: Callable[[HookInput], HookOutput],
platform: Platform,
@@ -340,24 +389,25 @@ def run_hook(
platform: The platform calling this hook
Returns:
- Exit code (0 for success, 2 for blocking)
+ Exit code (0 for success)
"""
- # Read and normalize input
- raw_input = read_stdin()
- hook_input = normalize_input(raw_input, platform)
-
- # Call the hook
try:
+ # Read and normalize input
+ raw_input = read_stdin()
+ hook_input = normalize_input(raw_input, platform)
+
+ # Call the hook
hook_output = hook_fn(hook_input)
- except Exception as e:
- # On error, allow the action but log
- print(f"Hook error: {e}", file=sys.stderr)
- hook_output = HookOutput()
- # Denormalize and write output
- output_json = denormalize_output(hook_output, platform, hook_input.event)
- write_stdout(output_json)
+ # Denormalize and write output
+ output_json = denormalize_output(hook_output, platform, hook_input.event)
+ write_stdout(output_json)
- # Always return 0 when using JSON output format
- # The decision field in the JSON controls blocking behavior
- return 0
+ # Always return 0 when using JSON output format
+ # The decision field in the JSON controls blocking behavior
+ return 0
+
+ except Exception as e:
+ # On any error, output a proper JSON error response
+ output_hook_error(e, context=f"Running hook {hook_fn.__name__}")
+ return 0 # Return 0 so Claude Code processes our JSON output
diff --git a/tests/unit/test_command_executor.py b/tests/unit/test_command_executor.py
index 77d7b320..12472729 100644
--- a/tests/unit/test_command_executor.py
+++ b/tests/unit/test_command_executor.py
@@ -170,7 +170,7 @@ def test_single_error(self) -> None:
),
]
output = format_command_errors(results)
- assert "failing_cmd" in output
+ assert "Command: failing_cmd" in output
assert "Something went wrong" in output
assert "Exit code: 1" in output
@@ -195,3 +195,70 @@ def test_ignores_success(self) -> None:
output = format_command_errors(results)
assert "good_cmd" not in output
assert "bad_cmd" in output
+
+ def test_includes_rule_name(self) -> None:
+ """Include rule name when provided."""
+ results = [
+ CommandResult(
+ success=False,
+ exit_code=1,
+ stdout="",
+ stderr="Error output",
+ command="test_cmd",
+ ),
+ ]
+ output = format_command_errors(results, rule_name="My Test Rule")
+ assert "Rule: My Test Rule" in output
+ assert "Command: test_cmd" in output
+ assert "Exit code: 1" in output
+ assert "Stderr:\nError output" in output
+
+ def test_includes_stdout(self) -> None:
+ """Include stdout when present."""
+ results = [
+ CommandResult(
+ success=False,
+ exit_code=1,
+ stdout="Standard output here",
+ stderr="Standard error here",
+ command="test_cmd",
+ ),
+ ]
+ output = format_command_errors(results)
+ assert "Stdout:\nStandard output here" in output
+ assert "Stderr:\nStandard error here" in output
+
+ def test_shows_no_output_message(self) -> None:
+ """Show '(no output)' when no stdout or stderr."""
+ results = [
+ CommandResult(
+ success=False,
+ exit_code=42,
+ stdout="",
+ stderr="",
+ command="silent_cmd",
+ ),
+ ]
+ output = format_command_errors(results)
+ assert "Command: silent_cmd" in output
+ assert "Exit code: 42" in output
+ assert "(no output)" in output
+
+ def test_full_error_format(self) -> None:
+ """Test complete error format with all fields."""
+ results = [
+ CommandResult(
+ success=False,
+ exit_code=42,
+ stdout="stdout output",
+ stderr="stderr output",
+ command="echo test && exit 42",
+ ),
+ ]
+ output = format_command_errors(results, rule_name="Command Failure Rule")
+ # Verify all parts are present in the correct format
+ assert "Rule: Command Failure Rule" in output
+ assert "Command: echo test && exit 42" in output
+ assert "Exit code: 42" in output
+ assert "Stdout:\nstdout output" in output
+ assert "Stderr:\nstderr output" in output
diff --git a/tests/unit/test_hook_wrapper.py b/tests/unit/test_hook_wrapper.py
index fd1a51d9..a0c84df6 100644
--- a/tests/unit/test_hook_wrapper.py
+++ b/tests/unit/test_hook_wrapper.py
@@ -32,7 +32,10 @@
NormalizedEvent,
Platform,
denormalize_output,
+ format_hook_error,
normalize_input,
+ output_hook_error,
+ run_hook,
)
@@ -555,3 +558,76 @@ def sample_hook(hook_input: HookInput) -> HookOutput:
assert claude_result["decision"] == "block"
assert gemini_result["decision"] == "deny"
assert claude_result["reason"] == gemini_result["reason"]
+
+
+class TestHookErrorHandling:
+ """Tests for hook script error handling.
+
+ When a hook script crashes, it should output valid JSON with detailed
+ error information rather than causing a generic "non-blocking status code" error.
+ """
+
+ def test_format_hook_error_basic(self) -> None:
+ """Test that format_hook_error produces blocking JSON with error details."""
+
+ error = ValueError("Something went wrong")
+ result = format_hook_error(error)
+
+ assert result["decision"] == "block"
+ assert "ValueError" in result["reason"]
+ assert "Something went wrong" in result["reason"]
+ assert "Traceback" in result["reason"]
+
+ def test_format_hook_error_with_context(self) -> None:
+ """Test that format_hook_error includes context when provided."""
+
+ error = RuntimeError("Test error")
+ result = format_hook_error(error, context="Loading rules")
+
+ assert result["decision"] == "block"
+ assert "Loading rules" in result["reason"]
+ assert "RuntimeError" in result["reason"]
+ assert "Test error" in result["reason"]
+
+ def test_format_hook_error_includes_traceback(self) -> None:
+ """Test that format_hook_error includes the full traceback."""
+
+ try:
+ raise KeyError("missing_key")
+ except KeyError as e:
+ result = format_hook_error(e)
+
+ assert "Traceback" in result["reason"]
+ assert "KeyError" in result["reason"]
+ assert "missing_key" in result["reason"]
+
+ def test_output_hook_error_produces_json(self, capsys: object) -> None:
+ """Test that output_hook_error writes valid JSON to stdout."""
+
+ error = TypeError("Type mismatch")
+ output_hook_error(error, context="test context")
+
+ captured = capsys.readouterr() # type: ignore[attr-defined]
+ result = json.loads(captured.out.strip())
+
+ assert result["decision"] == "block"
+ assert "TypeError" in result["reason"]
+ assert "test context" in result["reason"]
+
+ def test_run_hook_catches_hook_function_errors(self, capsys: object) -> None:
+ """Test that run_hook outputs JSON when hook function raises an exception."""
+ from deepwork.hooks.wrapper import HookInput, HookOutput, Platform
+
+ def failing_hook(hook_input: HookInput) -> HookOutput:
+ raise RuntimeError("Hook crashed!")
+
+ exit_code = run_hook(failing_hook, Platform.CLAUDE)
+
+ captured = capsys.readouterr() # type: ignore[attr-defined]
+ result = json.loads(captured.out.strip())
+
+ assert exit_code == 0 # Should return 0 so JSON is processed
+ assert result["decision"] == "block"
+ assert "RuntimeError" in result["reason"]
+ assert "Hook crashed!" in result["reason"]
+ assert "failing_hook" in result["reason"] # Context includes function name