feat: block work tools without active delegation (exit 2)

Scoped to 7 WORK_TOOLS only — Agent, Skill, TaskCreate etc. always
pass. Subagents and active delegations bypass.

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

Author: barkain

hooks/PreToolUse/require_delegation.py

@@ -3,22 +3,14 @@
 # requires-python = ">=3.12"
 # ///
 """
-PreToolUse Hook: Adaptive Delegation Nudge (cross-platform)
+PreToolUse Hook: Delegation Enforcement (cross-platform)
 
-Soft enforcement: never blocks. Tracks per-turn direct work-tool calls from
-the main agent and writes an escalating reminder to stderr. Cost (and signal)
-scales with violation count — silent for compliant sessions, louder when the
-main agent is repeatedly bypassing /workflow-orchestrator:delegate.
-
-Violation set is small and stable: only the 7 work-doing primitives that
-existed before the plugin and won't be renamed. New Claude Code tools never
-trigger nudges.
-
-Counter is reset by UserPromptSubmit (new turn) and zeroed when
-/workflow-orchestrator:delegate runs (handled by remind_skill_continuation.py).
+Blocks the 7 work-tool primitives unless a delegation is active.
+Subagents and active delegations bypass.
 
 EXIT CODES:
-- 0: always (this hook never blocks)
+- 0: tool allowed
+- 2: tool blocked (work tool without active delegation)
 """
 
 import io
@@ -28,7 +20,6 @@
 import sys
 from pathlib import Path
 
-# Force UTF-8 output on Windows (fixes emoji encoding errors)
 if sys.platform == "win32":
     sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
     sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8", errors="replace")
@@ -39,7 +30,6 @@
 _handler.setFormatter(logging.Formatter("%(message)s"))
 logger.addHandler(_handler)
 
-# Stable, work-doing primitives. New Claude Code tools never appear here.
 WORK_TOOLS = {
     "Bash",
     "Edit",
@@ -50,108 +40,47 @@
     "NotebookEdit",
 }
 
+BLOCK_MSG = "Run /workflow-orchestrator:delegate <task> first."
+
 
 def get_state_dir() -> Path:
-    """Get the state directory path."""
     project_dir = Path(os.environ.get("CLAUDE_PROJECT_DIR", str(Path.cwd()))).resolve()
     state_dir = project_dir / ".claude" / "state"
     state_dir.mkdir(parents=True, exist_ok=True)
     return state_dir
 
 
 def is_subagent() -> bool:
-    """True if running inside a subagent (delegation in progress)."""
     return bool(
         os.environ.get("CLAUDE_PARENT_SESSION_ID") or os.environ.get("CLAUDE_AGENT_ID")
     )
 
 
-def load_violations(path: Path) -> dict[str, int | str]:
-    """Load violations counter, or return a fresh one."""
-    if not path.exists():
-        return {"violations": 0, "delegations": 0, "turn_id": ""}
-    try:
-        return json.loads(path.read_text(encoding="utf-8"))
-    except (json.JSONDecodeError, OSError):
-        return {"violations": 0, "delegations": 0, "turn_id": ""}
-
-
-def save_violations(path: Path, data: dict[str, int | str]) -> None:
-    """Persist violations counter."""
-    try:
-        path.write_text(json.dumps(data), encoding="utf-8")
-    except OSError:
-        pass
-
-
-def message_for(count: int) -> str | None:
-    """Pick the escalation message for a given violation count.
-
-    Token cost scales with severity. Compliant sessions pay 0 tokens.
-    """
-    if count <= 0:
-        return None
-    if count == 1:
-        return (
-            "STOP. This tool call bypasses delegation. Abandon this step "
-            "and run: /workflow-orchestrator:delegate <your task>"
-        )
-    if count == 2:
-        return (
-            "STOP. 2nd direct tool call this turn. The main agent does not "
-            "execute work tools. Run: /workflow-orchestrator:delegate <your task>"
-        )
-    return (
-        f"STOP. {count} direct tool calls bypassing delegation. You are "
-        "losing planning, parallelization, and context isolation. Abandon "
-        "the current plan and run: /workflow-orchestrator:delegate <your task>"
-    )
-
-
 def main() -> int:
-    """Main entry point. Always returns 0."""
-    # Subagents are executing a delegation — never count their tool use.
     if is_subagent():
         return 0
 
     state_dir = get_state_dir()
 
-    # If a delegation is active in this session, the main agent is steering
-    # subagents — direct tool use is fine.
     if (state_dir / "delegation_active").exists():
         return 0
 
-    # Read tool name from stdin
     try:
         stdin_data = sys.stdin.read()
         data = json.loads(stdin_data) if stdin_data else {}
     except (OSError, json.JSONDecodeError):
         return 0
 
-    tool_name = str(data.get("tool_name", ""))
-    if tool_name not in WORK_TOOLS:
+    if str(data.get("tool_name", "")) not in WORK_TOOLS:
         return 0
 
-    # Count + nudge
-    counter_file = state_dir / "delegation_violations.json"
-    state = load_violations(counter_file)
-    count = int(state.get("violations", 0)) + 1
-    state["violations"] = count
-    save_violations(counter_file, state)
-
-    msg = message_for(count)
-    if msg:
-        logger.warning("%s", msg)
-
-    return 0
+    logger.warning("%s", BLOCK_MSG)
+    return 2
 
 
 if __name__ == "__main__":
     try:
         sys.exit(main())
     except Exception as e:  # noqa: BLE001
-        # Surface internal errors per hook exit-code policy (1 = hook error).
-        # Exit 1 is a non-blocking error in Claude Code — the tool call still
-        # proceeds, but the failure is visible in debug output.
         logger.error("require_delegation hook error: %s", e)
         sys.exit(1)

tests/test_require_delegation.py

@@ -1,4 +1,4 @@
-"""Tests for hooks/PreToolUse/require_delegation.py (soft adaptive nudges)."""
+"""Tests for hooks/PreToolUse/require_delegation.py (blocks work tools)."""
 
 import json
 import os
@@ -18,10 +18,8 @@ def _run(
     project_dir: Path,
     env_extra: dict[str, str] | None = None,
 ) -> tuple[str, str, int]:
-    """Run the hook in an isolated project dir, return (stdout, stderr, rc)."""
     env = os.environ.copy()
     env["CLAUDE_PROJECT_DIR"] = str(project_dir)
-    # Strip subagent markers so we test main-agent behavior unless overridden
     env.pop("CLAUDE_PARENT_SESSION_ID", None)
     env.pop("CLAUDE_AGENT_ID", None)
     if env_extra:
@@ -42,193 +40,61 @@ def _input(tool_name: str) -> str:
     return json.dumps({"tool_name": tool_name, "tool_input": {}})
 
 
-# ---------------------------------------------------------------------------
-# Always returns 0
-# ---------------------------------------------------------------------------
-
-
-class TestNeverBlocks:
-    def test_work_tool_returns_zero(self, tmp_path: Path) -> None:
-        _, _, rc = _run(_input("Bash"), tmp_path)
-        assert rc == 0  # noqa: S101
-
-    def test_unknown_tool_returns_zero(self, tmp_path: Path) -> None:
-        _, _, rc = _run(_input("FooBarBaz"), tmp_path)
-        assert rc == 0  # noqa: S101
-
-    def test_empty_stdin_returns_zero(self, tmp_path: Path) -> None:
-        _, _, rc = _run("", tmp_path)
-        assert rc == 0  # noqa: S101
-
-    def test_malformed_stdin_returns_zero(self, tmp_path: Path) -> None:
-        _, _, rc = _run("not json", tmp_path)
-        assert rc == 0  # noqa: S101
-
-
-# ---------------------------------------------------------------------------
-# Stable violation set: only WORK_TOOLS trigger nudges
-# ---------------------------------------------------------------------------
-
-
-class TestViolationSet:
+class TestBlocksWorkTools:
     @pytest.mark.parametrize(
         "tool",
         ["Bash", "Edit", "Write", "Glob", "Grep", "MultiEdit", "NotebookEdit"],
     )
-    def test_work_tools_count_as_violations(self, tmp_path: Path, tool: str) -> None:
-        _, stderr, _ = _run(_input(tool), tmp_path)
+    def test_work_tools_blocked(self, tmp_path: Path, tool: str) -> None:
+        _, stderr, rc = _run(_input(tool), tmp_path)
+        assert rc == 2  # noqa: S101
         assert "delegate" in stderr.lower()  # noqa: S101
 
-    def test_read_is_silent(self, tmp_path: Path) -> None:
-        """Read is exempt from delegation nudges (not in WORK_TOOLS)."""
+    def test_read_allowed(self, tmp_path: Path) -> None:
         _, stderr, rc = _run(_input("Read"), tmp_path)
         assert rc == 0  # noqa: S101
         assert stderr == ""  # noqa: S101
 
     @pytest.mark.parametrize(
         "tool",
-        [
-            "AskUserQuestion",
-            "TaskCreate",
-            "TaskUpdate",
-            "Agent",
-            "Task",
-            "Skill",
-            "SlashCommand",
-            "EnterPlanMode",
-            "ExitPlanMode",
-            "ToolSearch",
-            "TeamCreate",
-            "SendMessage",
-            "FooBar",  # hypothetical new tool
-            "CronCreate",
-        ],
+        ["Agent", "Skill", "SlashCommand", "TaskCreate", "TaskUpdate",
+         "EnterPlanMode", "ExitPlanMode", "ToolSearch", "TeamCreate",
+         "SendMessage", "AskUserQuestion", "FooBar", "CronCreate"],
     )
-    def test_non_work_tools_silent(self, tmp_path: Path, tool: str) -> None:
+    def test_non_work_tools_allowed(self, tmp_path: Path, tool: str) -> None:
         _, stderr, rc = _run(_input(tool), tmp_path)
         assert rc == 0  # noqa: S101
         assert stderr == ""  # noqa: S101
 
+    def test_empty_stdin_allowed(self, tmp_path: Path) -> None:
+        _, _, rc = _run("", tmp_path)
+        assert rc == 0  # noqa: S101
 
-# ---------------------------------------------------------------------------
-# Escalation ladder
-# ---------------------------------------------------------------------------
-
-
-class TestEscalationLadder:
-    def test_first_violation_is_imperative(self, tmp_path: Path) -> None:
-        _, stderr, _ = _run(_input("Bash"), tmp_path)
-        # New imperative message must include STOP + the delegate command
-        assert "STOP" in stderr  # noqa: S101
-        assert "/workflow-orchestrator:delegate" in stderr  # noqa: S101
-
-    def test_second_violation_is_medium(self, tmp_path: Path) -> None:
-        _run(_input("Bash"), tmp_path)
-        _, stderr, _ = _run(_input("Edit"), tmp_path)
-        assert "STOP" in stderr  # noqa: S101
-        assert "/workflow-orchestrator:delegate" in stderr  # noqa: S101
-        # Distinct 2nd-call phrasing
-        assert "2nd direct tool call" in stderr  # noqa: S101
-
-    def test_third_violation_is_warning(self, tmp_path: Path) -> None:
-        for tool in ("Bash", "Edit", "Write"):
-            _run(_input(tool), tmp_path)
-        _, stderr, _ = _run(_input("Glob"), tmp_path)
-        # 4th call -> "STOP. 4 direct tool calls bypassing delegation..."
-        assert "STOP" in stderr  # noqa: S101
-        assert "4" in stderr  # noqa: S101
-        assert "/workflow-orchestrator:delegate" in stderr  # noqa: S101
-
-    def test_fifth_plus_is_strong(self, tmp_path: Path) -> None:
-        for _ in range(5):
-            _run(_input("Bash"), tmp_path)
-        _, stderr, _ = _run(_input("Bash"), tmp_path)  # 6th call
-        assert "STOP" in stderr  # noqa: S101
-        # The ≥3 message explains what's being lost
-        assert "losing planning, parallelization, and context isolation" in stderr  # noqa: S101
-        # Long message — over ~100 chars
-        assert len(stderr.strip()) > 100  # noqa: S101
-
-    def test_message_length_grows(self, tmp_path: Path) -> None:
-        """Verify escalation adds context (proxied by message length/content)."""
-        lengths = []
-        messages = []
-        for _ in range(6):
-            _, stderr, _ = _run(_input("Bash"), tmp_path)
-            lengths.append(len(stderr.strip()))
-            messages.append(stderr.strip())
-        # Each level should be at least as long as the previous (monotonic)
-        for i in range(1, len(lengths)):
-            assert lengths[i] >= lengths[i - 1]  # noqa: S101
-        # The final message should be strictly longer than the first
-        assert lengths[-1] > lengths[0]  # noqa: S101
-        # And must include the "what's being lost" context that proves escalation
-        assert "losing planning, parallelization, and context isolation" in messages[-1]  # noqa: S101
-
-
-# ---------------------------------------------------------------------------
-# Counter persistence
-# ---------------------------------------------------------------------------
-
-
-class TestCounterPersistence:
-    def test_counter_persists_across_calls(self, tmp_path: Path) -> None:
-        for _ in range(3):
-            _run(_input("Bash"), tmp_path)
-        counter = json.loads(
-            (tmp_path / ".claude" / "state" / "delegation_violations.json").read_text()
-        )
-        assert counter["violations"] == 3  # noqa: S101
-
-    def test_fresh_project_starts_at_zero(self, tmp_path: Path) -> None:
-        _run(_input("Bash"), tmp_path)
-        counter = json.loads(
-            (tmp_path / ".claude" / "state" / "delegation_violations.json").read_text()
-        )
-        assert counter["violations"] == 1  # noqa: S101
-
-
-# ---------------------------------------------------------------------------
-# Subagent immunity
-# ---------------------------------------------------------------------------
+    def test_malformed_stdin_allowed(self, tmp_path: Path) -> None:
+        _, _, rc = _run("not json", tmp_path)
+        assert rc == 0  # noqa: S101
 
 
 class TestSubagentImmunity:
-    def test_parent_session_id_skips(self, tmp_path: Path) -> None:
+    def test_parent_session_id_allows(self, tmp_path: Path) -> None:
         _, stderr, rc = _run(
-            _input("Bash"),
-            tmp_path,
+            _input("Bash"), tmp_path,
             env_extra={"CLAUDE_PARENT_SESSION_ID": "abc123"},
         )
         assert rc == 0  # noqa: S101
         assert stderr == ""  # noqa: S101
 
-    def test_agent_id_skips(self, tmp_path: Path) -> None:
+    def test_agent_id_allows(self, tmp_path: Path) -> None:
         _, stderr, rc = _run(
-            _input("Bash"),
-            tmp_path,
+            _input("Bash"), tmp_path,
             env_extra={"CLAUDE_AGENT_ID": "xyz"},
         )
         assert rc == 0  # noqa: S101
         assert stderr == ""  # noqa: S101
 
-    def test_subagent_does_not_increment_counter(self, tmp_path: Path) -> None:
-        _run(
-            _input("Bash"),
-            tmp_path,
-            env_extra={"CLAUDE_PARENT_SESSION_ID": "x"},
-        )
-        counter_file = tmp_path / ".claude" / "state" / "delegation_violations.json"
-        assert not counter_file.exists()  # noqa: S101
-
-
-# ---------------------------------------------------------------------------
-# delegation_active flag suppresses nudges
-# ---------------------------------------------------------------------------
-
 
 class TestDelegationActiveFlag:
-    def test_active_flag_suppresses_nudge(self, tmp_path: Path) -> None:
+    def test_active_flag_allows(self, tmp_path: Path) -> None:
         state = tmp_path / ".claude" / "state"
         state.mkdir(parents=True)
         (state / "delegation_active").touch()