feat: add tool requirements policy enforcement system

Introduces a PreToolUse hook-based policy system that evaluates tool calls
against RFC 2119-style requirements defined in .deepwork/tool_requirements/*.yml.
Policies are checked via an HTTP sidecar server (spawned alongside the MCP server)
using Haiku for semantic evaluation. Failed checks can be appealed via a new
appeal_tool_requirement MCP tool. Approvals are cached with a 1-hour TTL.

Key features:
- Policy files with tools, match (param regex), requirements, extends (inheritance)
- no_exception rules that cannot be appealed
- Fail-closed: hook denies if MCP sidecar is unreachable
- Loop prevention: appeal tool calls skip the hook
- Multi-instance support via PID-keyed + session-keyed port files
- Evaluator encapsulated behind ABC for future swap to direct API calls

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: nhorton

plugins/claude/hooks/hooks.json

@@ -1,5 +1,16 @@
 {
   "hooks": {
+    "PreToolUse": [
+      {
+        "matcher": "",
+        "hooks": [
+          {
+            "type": "command",
+            "command": "${CLAUDE_PLUGIN_ROOT}/hooks/tool_requirements.sh"
+          }
+        ]
+      }
+    ],
     "SessionStart": [
       {
         "matcher": "",

plugins/claude/hooks/tool_requirements.sh

@@ -0,0 +1,15 @@
+#!/usr/bin/env bash
+# tool_requirements.sh - PreToolUse hook for tool requirements enforcement
+#
+# Fires before every tool call. Delegates to the Python hook which contacts
+# the MCP sidecar to check policies.
+#
+# Input (stdin):  JSON from Claude Code PreToolUse hook
+# Output (stdout): JSON with hookSpecificOutput.permissionDecision
+# Exit codes:
+#   0 - Always (decision encoded in JSON output)
+
+INPUT=$(cat)
+export DEEPWORK_HOOK_PLATFORM="claude"
+echo "${INPUT}" | deepwork hook tool_requirements
+exit $?

src/deepwork/cli/serve.py

@@ -122,6 +122,9 @@ def _serve_mcp(
             "# Ignore everything in this directory\n*\n# But keep this .gitignore\n!.gitignore\n"
         )
 
+    # Start tool requirements sidecar (if policies exist)
+    _start_tool_requirements_sidecar(project_path)
+
     # Create and run server
     from deepwork.jobs.mcp.server import create_server
 
@@ -135,3 +138,23 @@ def _serve_mcp(
         server.run(transport="stdio")
     else:
         server.run(transport="sse", port=port)
+
+
+def _start_tool_requirements_sidecar(project_path: Path) -> None:
+    """Start the tool requirements sidecar if policy files exist."""
+    policy_dir = project_path / ".deepwork" / "tool_requirements"
+    if not policy_dir.is_dir():
+        return
+    if not any(policy_dir.glob("*.yml")):
+        return
+
+    try:
+        from deepwork.tool_requirements.sidecar import start_sidecar
+
+        start_sidecar(project_path)
+    except Exception:
+        import logging
+
+        logging.getLogger("deepwork.tool_requirements").warning(
+            "Failed to start tool requirements sidecar", exc_info=True
+        )

src/deepwork/hooks/tool_requirements.py

@@ -0,0 +1,126 @@
+"""PreToolUse hook for tool requirements policy enforcement.
+
+Fires before every tool call. Contacts the MCP sidecar server to check
+whether the call complies with policies defined in
+.deepwork/tool_requirements/*.yml.
+
+Fail-closed: if the sidecar is unreachable, the hook denies the call
+with a message to restart the MCP server.
+"""
+
+from __future__ import annotations
+
+import http.client
+import json
+import os
+import sys
+from pathlib import Path
+from typing import Any
+
+from deepwork.hooks.wrapper import (
+    HookInput,
+    HookOutput,
+    NormalizedEvent,
+    Platform,
+    output_hook_error,
+    run_hook,
+)
+from deepwork.tool_requirements.sidecar import discover_sidecar
+
+# Tool name substrings to skip (loop prevention)
+_SKIP_TOOLS = ("appeal_tool_requirement",)
+
+
+def tool_requirements_hook(hook_input: HookInput) -> HookOutput:
+    """Pre-tool hook: check tool call against requirement policies."""
+    if hook_input.event != NormalizedEvent.BEFORE_TOOL:
+        return HookOutput()
+
+    # Loop prevention: skip the appeal MCP tool itself
+    raw_tool = hook_input.raw_input.get("tool_name", "")
+    for skip in _SKIP_TOOLS:
+        if skip in raw_tool:
+            return HookOutput()
+
+    cwd = hook_input.cwd or os.getcwd()
+    session_id = hook_input.session_id or ""
+
+    # Discover sidecar
+    sidecar = discover_sidecar(Path(cwd), session_id)
+    if sidecar is None:
+        return _deny(
+            "DeepWork Tool Requirements: MCP server is not running. "
+            "The tool_requirements system requires the MCP server to be active. "
+            "Please restart the MCP server."
+        )
+
+    # Send check request to sidecar
+    try:
+        response = _http_post(sidecar["port"], "/check", {
+            "tool_name": hook_input.tool_name,
+            "tool_input": hook_input.tool_input,
+            "raw_tool_name": raw_tool,
+            "session_id": session_id,
+        })
+    except Exception as e:
+        return _deny(
+            f"DeepWork Tool Requirements: Failed to reach MCP server sidecar: {e}. "
+            "Please restart the MCP server."
+        )
+
+    if response.get("decision") == "allow":
+        return HookOutput()
+
+    if response.get("decision") == "deny":
+        reason = response.get("reason", "Tool call blocked by policy")
+        return _deny(reason)
+
+    # Unexpected response — allow (fail-open only for malformed responses
+    # from an actually-running sidecar, not for missing sidecars)
+    return HookOutput()
+
+
+def _deny(reason: str) -> HookOutput:
+    """Create a deny output for PreToolUse with proper Claude Code format."""
+    return HookOutput(
+        raw_output={
+            "hookSpecificOutput": {
+                "hookEventName": "PreToolUse",
+                "permissionDecision": "deny",
+                "permissionDecisionReason": reason,
+            }
+        }
+    )
+
+
+def _http_post(port: int, path: str, body: dict[str, Any]) -> dict[str, Any]:
+    """Send an HTTP POST to the sidecar on localhost."""
+    conn = http.client.HTTPConnection("127.0.0.1", port, timeout=30)
+    try:
+        payload = json.dumps(body).encode("utf-8")
+        conn.request(
+            "POST",
+            path,
+            body=payload,
+            headers={"Content-Type": "application/json"},
+        )
+        response = conn.getresponse()
+        data = response.read()
+        result: dict[str, Any] = json.loads(data)
+        return result
+    finally:
+        conn.close()
+
+
+def main() -> int:
+    """Entry point for the hook CLI."""
+    platform = Platform(os.environ.get("DEEPWORK_HOOK_PLATFORM", "claude"))
+    return run_hook(tool_requirements_hook, platform)
+
+
+if __name__ == "__main__":
+    try:
+        sys.exit(main())
+    except Exception as e:
+        output_hook_error(e, context="tool_requirements hook")
+        sys.exit(0)

src/deepwork/jobs/mcp/server.py

@@ -149,6 +149,21 @@ def _log_tool_call(
             log_data["params"] = params
         logger.info("MCP tool call: %s", log_data)
 
+    # Track whether session has been registered for tool requirements sidecar
+    _registered_sessions: set[str] = set()
+
+    def _maybe_register_session(session_id: str | None) -> None:
+        """Register session with the tool requirements sidecar on first tool call."""
+        if not session_id or session_id in _registered_sessions:
+            return
+        _registered_sessions.add(session_id)
+        try:
+            from deepwork.tool_requirements.sidecar import register_session
+
+            register_session(project_path, session_id)
+        except Exception:
+            pass  # Best-effort — sidecar may not be running
+
     @mcp.tool(
         description=(
             "List all available DeepWork workflows. "
@@ -186,6 +201,7 @@ async def start_workflow(
         agent_id: str | None = None,
     ) -> dict[str, Any]:
         """Start a workflow and get first step instructions."""
+        _maybe_register_session(session_id)
         _log_tool_call(
             "start_workflow",
             {
@@ -505,6 +521,84 @@ async def mark_review_as_passed(review_id: str, ctx: Context) -> str:
         except ValueError as e:
             return f"Validation error: {e}"
 
+    # ---- Tool Requirements: appeal tool ----
+
+    @mcp.tool(
+        description=(
+            "Appeal a tool requirement policy denial. When a tool call is blocked "
+            "by a tool requirement policy, call this to appeal specific failed "
+            "checks by providing justifications. "
+            "Required: tool_name (the normalized tool name that was blocked), "
+            "tool_input (the exact tool_input that was blocked), "
+            "policy_justification (dict mapping each failed check name to a "
+            "justification string explaining why the check should pass). "
+            "Optional: session_id (CLAUDE_CODE_SESSION_ID). "
+            "Some checks are marked no_exception and cannot be appealed. "
+            "If the appeal succeeds, the tool call is cached as approved and "
+            "you can retry the original tool call."
+        )
+    )
+    async def appeal_tool_requirement(
+        tool_name: str,
+        tool_input: dict[str, Any],
+        policy_justification: dict[str, str],
+        ctx: Context,
+        session_id: str | None = None,
+    ) -> dict[str, Any]:
+        """Appeal a tool requirement denial with justifications."""
+        _log_tool_call(
+            "appeal_tool_requirement",
+            {
+                "tool_name": tool_name,
+                "justification_keys": list(policy_justification.keys()),
+            },
+            session_id=session_id,
+        )
+        _maybe_register_session(session_id)
+
+        root = await root_resolver.get_root(ctx)
+
+        # Delegate to sidecar (same process) or engine directly
+        try:
+            from deepwork.tool_requirements.sidecar import discover_sidecar
+
+            sidecar = discover_sidecar(root, session_id or "")
+            if sidecar is None:
+                return {
+                    "passed": False,
+                    "reason": "Tool requirements sidecar is not running. "
+                    "Please restart the MCP server.",
+                }
+
+            import http.client
+            import json as json_mod
+
+            conn = http.client.HTTPConnection(
+                "127.0.0.1", sidecar["port"], timeout=60
+            )
+            try:
+                payload = json_mod.dumps({
+                    "tool_name": tool_name,
+                    "tool_input": tool_input,
+                    "policy_justification": policy_justification,
+                }).encode("utf-8")
+                conn.request(
+                    "POST",
+                    "/appeal",
+                    body=payload,
+                    headers={"Content-Type": "application/json"},
+                )
+                response = conn.getresponse()
+                return json_mod.loads(response.read())
+            finally:
+                conn.close()
+        except Exception as e:
+            logger.exception("Error in appeal_tool_requirement")
+            return {
+                "passed": False,
+                "reason": f"Appeal failed: {e}",
+            }
+
     return mcp
 
 

src/deepwork/schemas/tool_requirements_schema.json

@@ -0,0 +1,62 @@
+{
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "title": "Tool Requirements Policy",
+    "description": "Schema for .deepwork/tool_requirements/*.yml policy files that define RFC 2119-style rules for AI agent tool calls.",
+    "type": "object",
+    "required": ["tools", "requirements"],
+    "additionalProperties": false,
+    "properties": {
+        "summary": {
+            "type": "string",
+            "description": "Human-readable summary of what this policy enforces."
+        },
+        "tools": {
+            "type": "array",
+            "description": "Normalized tool names (shell, write_file, edit_file, etc.) or MCP tool names (mcp__server__tool) this policy applies to.",
+            "items": {
+                "type": "string"
+            },
+            "minItems": 1
+        },
+        "match": {
+            "type": "object",
+            "description": "Optional parameter-level filtering. Keys are tool_input parameter names, values are regex patterns. Policy only applies when at least one pattern matches.",
+            "patternProperties": {
+                "^[a-zA-Z0-9_-]+$": {
+                    "type": "string"
+                }
+            },
+            "additionalProperties": false
+        },
+        "extends": {
+            "type": "array",
+            "description": "List of policy file stems to inherit requirements from.",
+            "items": {
+                "type": "string"
+            }
+        },
+        "requirements": {
+            "type": "object",
+            "description": "RFC 2119 keyed requirements. Keys are requirement identifiers, values define the rule and exception policy.",
+            "patternProperties": {
+                "^[a-zA-Z0-9_-]+$": {
+                    "type": "object",
+                    "required": ["rule"],
+                    "additionalProperties": false,
+                    "properties": {
+                        "rule": {
+                            "type": "string",
+                            "description": "RFC 2119 statement (using MUST, SHOULD, MAY, etc.)."
+                        },
+                        "no_exception": {
+                            "type": "boolean",
+                            "description": "If true, this requirement cannot be appealed. Defaults to false.",
+                            "default": false
+                        }
+                    }
+                }
+            },
+            "additionalProperties": false
+        }
+    }
+}

src/deepwork/tool_requirements/__init__.py

@@ -0,0 +1 @@
+"""Tool Requirements — policy enforcement for AI agent tool calls."""