Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 26 additions & 9 deletions src/deepwork/core/command_executor.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
30 changes: 23 additions & 7 deletions src/deepwork/hooks/rules_check.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 `<promise>✓ {rule.name}</promise>` 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 `<promise>✓ {rule.name}</promise>` in your response."
command_errors.append(f"{error_msg}{skip_hint}")
queue.update_status(
trigger_hash,
QueueEntryStatus.FAILED,
Expand Down Expand Up @@ -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)
82 changes: 66 additions & 16 deletions src/deepwork/hooks/wrapper.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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
69 changes: 68 additions & 1 deletion tests/unit/test_command_executor.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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
76 changes: 76 additions & 0 deletions tests/unit/test_hook_wrapper.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,10 @@
NormalizedEvent,
Platform,
denormalize_output,
format_hook_error,
normalize_input,
output_hook_error,
run_hook,
)


Expand Down Expand Up @@ -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