feat(ws0+ws2+ws4): plugin scaffold, Slack /new-task, auto-delegate hook

WS0 — sea-haven-claude-plugin/ scaffold: - CLAUDE.md: Sea Haven engineering context (pipeline overview, rules, /new-task + /delegate usage, available phases) - settings.template.json: UserPromptSubmit hook wiring template - hooks/user_prompt_submit.py: standalone script (WS4) WS2 — SlackListener /new-task intake: - Add NewTaskCallback type alias (Callable[[str, str], str]) - Add new_task_callback param to SlackListener.__init__ - _is_new_task_command() helper for slash_commands+/new-task detection - _handle_new_task_command() method: authorized-only, calls callback, exception-safe (listen loop stays alive on callback errors) - handle_event() routes /new-task BEFORE the answer path (post-AUTHZ-01) WS4 — UserPromptSubmit auto-delegate hook: - /delegate <text> and DELEGATE: <text> prefixes trigger delegation - Calls POST /tasks on the agent-team HTTP API (WS1) - Blocks the Claude Code prompt; shows thread_id + next-steps message - Graceful degradation: missing token, HTTP error, network error all produce a block with a human-readable reason - run() is a pure function for testability (no stdin/stdout in tests) Tests: 22 new tests in test_ws0_ws2_ws4_plugin_slack_hook.py. Full suite: 1066 passed. ruff clean. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01QYp761G9HojkmLqLZrASVi
2026-06-23 01:39:12 +00:00 · 2026-06-23 01:39:12 +00:00 · 2aa72d73f4
commit 2aa72d73f4
parent 71036cb3f4
5 changed files with 677 additions and 0 deletions
--- a/agent-team/agent_team/transport/slack_listener.py
+++ b/agent-team/agent_team/transport/slack_listener.py
@ -78,14 +78,25 @@ from collections.abc import Mapping
 from pathlib import Path
 from typing import Any

+from collections.abc import Callable
+
 from agent_team.db.schema import connect, find_open_question_by_channel_ref
 from agent_team.responder import AnswerOutcome, EnqueueResume, submit_answer
 from agent_team.transport.slack_adapter import SlackTransport

 __all__ = [
+    "NewTaskCallback",
    "SlackListener",
 ]

+# Injectable callback for /new-task slash commands: receives (task_text, transport)
+# and returns the minted thread_id. Injected at coordinator startup so the
+# listener is testable with no coordinator and no graph.
+NewTaskCallback = Callable[[str, str], str]
+
+# The Slack slash command that starts a new pipeline task.
+_NEW_TASK_COMMAND = "/new-task"
+
 _LOG = logging.getLogger(__name__)

 # Discriminating Slack event ``type`` values that can carry an answer for the
@ -119,6 +130,17 @@ _ANSWER_BEARING_TYPES: frozenset[str] = frozenset(
 _EVENT_CALLBACK_TYPE = "event_callback"


+def _is_new_task_command(raw_payload: Mapping[str, Any]) -> bool:
+    """Return ``True`` iff this payload is a ``/new-task`` slash command.
+
+    Slash-command payloads carry ``{"type": "slash_commands", "command": "/new-task", ...}``.
+    """
+    return (
+        raw_payload.get("type") == "slash_commands"
+        and raw_payload.get("command") == _NEW_TASK_COMMAND
+    )
+
+
 class SlackListener:
    """Socket Mode inbound listener that drives answers into the responder.

@ -139,6 +161,11 @@ class SlackListener:
      may answer. If ``None``/empty the listener FAILS CLOSED and rejects every
      answer; :meth:`serve` sources it from ``AGENT_TEAM_SLACK_OWNER_IDS`` when
      not injected.
+    * ``new_task_callback`` — optional :data:`NewTaskCallback`; when set, the
+      listener handles ``/new-task <description>`` slash commands by calling it
+      with ``(task_text, "slack")`` and returning ``None`` (the task is started;
+      the owner will receive clarifying questions via the transport). When
+      ``None`` (the default), ``/new-task`` commands are ignored.

    The listener never resumes the graph; it only normalizes, submits, and
    enqueues. See the module SECURITY note for the trust boundary.
@ -153,6 +180,7 @@ class SlackListener:
        app_token: str | None = None,
        bot_token: str | None = None,
        owner_ids: set[str] | None = None,
+        new_task_callback: NewTaskCallback | None = None,
    ) -> None:
        self._transport = transport
        self._db_path = Path(db_path)
@ -162,6 +190,8 @@ class SlackListener:
        # The owner allowlist (AUTHZ-01). An empty set is the fail-closed default:
        # an unconfigured deploy rejects every answer.
        self._owner_ids: set[str] = set(owner_ids) if owner_ids else set()
+        # Injected /new-task callback (opt-in). None = ignore new-task commands.
+        self._new_task_callback = new_task_callback
        # The live Socket Mode handler, retained by :meth:`serve` so :meth:`close`
        # can stop it cleanly on daemon shutdown. ``None`` until ``serve`` opens
        # the socket.
@ -215,6 +245,13 @@ class SlackListener:
        if not self._is_authorized(raw_payload):
            return None

+        # /new-task intake path: handle BEFORE the answer path so a new-task
+        # slash command is never misrouted as an answer-to-a-question. Auth has
+        # already cleared above (AUTHZ-01), so only allowlisted owners can start
+        # tasks. The callback is opt-in; if not injected, /new-task is ignored.
+        if _is_new_task_command(raw_payload):
+            return self._handle_new_task_command(raw_payload)
+
        # ``submit_answer`` calls ``transport.parse_answer`` internally, which
        # raises ValueError when no question_id is recoverable. A real free-text
        # thread reply carries no callback_id / question_id / metadata, so its
@ -271,6 +308,44 @@ class SlackListener:
            )
        return outcome

+    def _handle_new_task_command(
+        self, raw_payload: Mapping[str, Any]
+    ) -> AnswerOutcome | None:
+        """Route a ``/new-task`` slash command to the injected start-task callback.
+
+        Extracts the task description from ``text`` (the words after the command
+        name). If no ``new_task_callback`` is configured, logs and returns
+        ``None`` (ignore). Otherwise calls ``new_task_callback(task_text, "slack")``
+        and returns ``None`` (the task is started; the owner receives clarifying
+        questions via the transport; there is no ``AnswerOutcome`` to return here).
+
+        Any exception raised by the callback is caught and logged; the listen
+        loop stays alive.
+        """
+        if self._new_task_callback is None:
+            _LOG.debug("/new-task received but no new_task_callback configured; ignoring")
+            return None
+
+        text = str(raw_payload.get("text") or "").strip()
+        if not text:
+            _LOG.info("/new-task received with empty description; ignoring")
+            return None
+
+        try:
+            thread_id = self._new_task_callback(text, "slack")
+            _LOG.info(
+                "new task started via /new-task: thread_id=%s task=%r",
+                thread_id,
+                text[:80],
+            )
+        except Exception:  # noqa: BLE001 - keep the listen loop alive
+            _LOG.warning(
+                "new_task_callback raised for /new-task (task=%r); ignored",
+                text[:80],
+                exc_info=True,
+            )
+        return None
+
    def _is_authorized(self, raw_payload: Mapping[str, Any]) -> bool:
        """Return ``True`` iff the payload's sender is an allowlisted owner.

--- a/agent-team/tests/test_ws0_ws2_ws4_plugin_slack_hook.py
+++ b/agent-team/tests/test_ws0_ws2_ws4_plugin_slack_hook.py
@ -0,0 +1,351 @@
+"""Tests for WS0 (plugin scaffold), WS2 (/new-task Slack handler), and WS4
+(UserPromptSubmit auto-delegate hook).
+
+WS0 — sea-haven-claude-plugin directory: structure + content checks (no
+network, no SDK).
+
+WS2 — SlackListener /new-task handler: the ``new_task_callback`` seam is
+injected so no coordinator is needed. Tests verify:
+  * /new-task from an authorized owner calls the callback with (text, "slack");
+  * /new-task from an unauthorized user is rejected (callback not called);
+  * /new-task with empty text is ignored (callback not called);
+  * non-/new-task slash commands are NOT intercepted by the new-task path;
+  * existing answer-handling is not disturbed (regression);
+  * callback exception does not crash the listen loop.
+
+WS4 — auto-delegate hook (sea-haven-claude-plugin/hooks/user_prompt_submit.py):
+the ``run()`` function is imported directly (no stdin) and tested with injected
+api_url / token so there is no network. Tests verify:
+  * non-delegate prompts are passed through (return {});
+  * /delegate <text> calls the API and blocks with thread_id in the reason;
+  * DELEGATE: prefix also triggers delegation;
+  * /delegate with no text returns usage hint (block);
+  * missing token → block with AGENT_TEAM_API_TOKEN hint;
+  * API HTTP error → block with error code;
+  * API network error → block with connectivity hint;
+  * unexpected exception → block with error message.
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import json
+import urllib.error
+from pathlib import Path
+from typing import Any
+from unittest.mock import patch
+
+import pytest
+
+from agent_team.transport.slack_listener import (
+    NewTaskCallback,
+    SlackListener,
+    _is_new_task_command,
+)
+
+# ---------------------------------------------------------------------------
+# Helpers / constants
+# ---------------------------------------------------------------------------
+
+_OWNER_ID = "U_OWNER"
+_PLUGIN_ROOT = Path(__file__).resolve().parents[2] / "sea-haven-claude-plugin"
+_HOOK_PATH = _PLUGIN_ROOT / "hooks" / "user_prompt_submit.py"
+
+
+def _load_hook_module():
+    """Import the hook script as a module without executing main()."""
+    spec = importlib.util.spec_from_file_location("user_prompt_submit", _HOOK_PATH)
+    assert spec is not None and spec.loader is not None
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)  # type: ignore[attr-defined]
+    return mod
+
+
+# ---------------------------------------------------------------------------
+# WS0 — plugin scaffold structure
+# ---------------------------------------------------------------------------
+
+
+def test_plugin_root_exists() -> None:
+    assert _PLUGIN_ROOT.is_dir(), f"sea-haven-claude-plugin/ not found at {_PLUGIN_ROOT}"
+
+
+def test_claude_md_exists_and_has_required_sections() -> None:
+    claude_md = _PLUGIN_ROOT / "CLAUDE.md"
+    assert claude_md.is_file(), "CLAUDE.md must exist in sea-haven-claude-plugin/"
+    text = claude_md.read_text(encoding="utf-8")
+    for keyword in ("Sea Haven", "/new-task", "/delegate", "agent-team", "pipeline"):
+        assert keyword in text, f"CLAUDE.md missing keyword {keyword!r}"
+
+
+def test_settings_template_exists_and_is_valid_json() -> None:
+    tmpl = _PLUGIN_ROOT / "settings.template.json"
+    assert tmpl.is_file(), "settings.template.json must exist in sea-haven-claude-plugin/"
+    data = json.loads(tmpl.read_text(encoding="utf-8"))
+    assert "hooks" in data, "settings.template.json must have a 'hooks' key"
+    assert "UserPromptSubmit" in data["hooks"], (
+        "settings.template.json must configure UserPromptSubmit hook"
+    )
+
+
+def test_hook_script_exists() -> None:
+    assert _HOOK_PATH.is_file(), f"hook script not found at {_HOOK_PATH}"
+
+
+# ---------------------------------------------------------------------------
+# WS2 — /new-task Slack handler
+# ---------------------------------------------------------------------------
+
+
+def _make_listener(
+    *,
+    new_task_callback: NewTaskCallback | None = None,
+    owner_ids: set[str] | None = None,
+    db_path: Path | None = None,
+) -> SlackListener:
+    from agent_team.transport.slack_adapter import SlackTransport
+
+    if db_path is None:
+        import tempfile
+        db_path = Path(tempfile.mkdtemp()) / "agent-team.db"
+        from agent_team.db.schema import init_db
+        init_db(db_path)
+
+    transport = SlackTransport(channel="C_FAKE")
+    return SlackListener(
+        transport,
+        db_path,
+        enqueue_resume=lambda _job: None,
+        owner_ids=owner_ids or {_OWNER_ID},
+        new_task_callback=new_task_callback,
+    )
+
+
+def _new_task_payload(text: str = "Add OAuth", user_id: str = _OWNER_ID) -> dict[str, Any]:
+    return {
+        "type": "slash_commands",
+        "command": "/new-task",
+        "text": text,
+        "user_id": user_id,
+    }
+
+
+def test_is_new_task_command_detects_slash_new_task() -> None:
+    assert _is_new_task_command(_new_task_payload())
+
+
+def test_is_new_task_command_ignores_other_commands() -> None:
+    payload = {"type": "slash_commands", "command": "/other", "text": "foo", "user_id": "U1"}
+    assert not _is_new_task_command(payload)
+
+
+def test_is_new_task_command_ignores_non_slash() -> None:
+    assert not _is_new_task_command({"type": "block_actions", "user": {"id": "U1"}})
+
+
+def test_new_task_calls_callback_with_text_and_transport() -> None:
+    calls: list[tuple[str, str]] = []
+
+    def cb(task: str, transport: str) -> str:
+        calls.append((task, transport))
+        return "thread-abc"
+
+    listener = _make_listener(new_task_callback=cb)
+    result = listener.handle_event(_new_task_payload("Add OAuth to admin portal"))
+
+    assert result is None  # no AnswerOutcome for new-task
+    assert calls == [("Add OAuth to admin portal", "slack")]
+
+
+def test_new_task_unauthorized_sender_rejected() -> None:
+    calls: list[Any] = []
+
+    def cb(task: str, transport: str) -> str:
+        calls.append(task)
+        return "thread-xyz"
+
+    listener = _make_listener(new_task_callback=cb)
+    result = listener.handle_event(_new_task_payload(user_id="U_ATTACKER"))
+
+    assert result is None
+    assert not calls, "callback must NOT be called for unauthorized sender"
+
+
+def test_new_task_empty_text_ignored() -> None:
+    calls: list[Any] = []
+
+    def cb(task: str, transport: str) -> str:
+        calls.append(task)
+        return "thread-123"
+
+    listener = _make_listener(new_task_callback=cb)
+    result = listener.handle_event(_new_task_payload(text=""))
+
+    assert result is None
+    assert not calls, "empty text must not trigger callback"
+
+
+def test_new_task_no_callback_configured_is_ignored() -> None:
+    listener = _make_listener(new_task_callback=None)
+    result = listener.handle_event(_new_task_payload("Some task"))
+    assert result is None
+
+
+def test_new_task_callback_exception_does_not_crash_listener() -> None:
+    def boom(task: str, transport: str) -> str:
+        raise RuntimeError("coordinator exploded")
+
+    listener = _make_listener(new_task_callback=boom)
+    result = listener.handle_event(_new_task_payload("Task that will fail"))
+    assert result is None  # exception swallowed; loop stays alive
+
+
+def test_other_slash_command_not_intercepted_by_new_task_path(
+    tmp_path: Path,
+) -> None:
+    from agent_team.db.schema import init_db
+
+    db_path = tmp_path / "agent-team.db"
+    init_db(db_path)
+    calls: list[Any] = []
+
+    def cb(task: str, transport: str) -> str:
+        calls.append(task)
+        return "thread-xyz"
+
+    listener = _make_listener(new_task_callback=cb, db_path=db_path)
+    other_slash = {
+        "type": "slash_commands",
+        "command": "/other",
+        "text": "some text",
+        "user_id": _OWNER_ID,
+    }
+    result = listener.handle_event(other_slash)
+    assert result is None
+    assert not calls, "non-/new-task slash commands must NOT call the new_task_callback"
+
+
+# ---------------------------------------------------------------------------
+# WS4 — auto-delegate hook
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture(scope="module")
+def hook():
+    """Import the hook module once per test session."""
+    return _load_hook_module()
+
+
+def test_passthrough_for_normal_prompt(hook: Any) -> None:
+    assert hook.run("Just a normal question") == {}
+
+
+def test_passthrough_for_empty_prompt(hook: Any) -> None:
+    assert hook.run("") == {}
+
+
+def test_delegate_prefix_triggers_delegation(hook: Any) -> None:
+    captured: list[dict[str, Any]] = []
+
+    def fake_call_api(task: str, *, api_url: str, token: str) -> dict[str, Any]:
+        captured.append({"task": task, "api_url": api_url, "token": token})
+        return {"thread_id": "thread-111"}
+
+    with patch.object(hook, "_call_api", fake_call_api):
+        result = hook.run(
+            "/delegate Add OAuth to the admin portal",
+            api_url="http://127.0.0.1:8765",
+            token="tok-abc",
+        )
+
+    assert result.get("action") == "block"
+    assert "thread-111" in result["reason"]
+    assert captured[0]["task"] == "Add OAuth to the admin portal"
+    assert captured[0]["token"] == "tok-abc"
+
+
+def test_delegate_colon_prefix_triggers_delegation(hook: Any) -> None:
+    captured: list[dict[str, Any]] = []
+
+    def fake_call_api(task: str, *, api_url: str, token: str) -> dict[str, Any]:
+        captured.append(task)
+        return {"thread_id": "thread-222"}
+
+    with patch.object(hook, "_call_api", fake_call_api):
+        result = hook.run(
+            "DELEGATE: Fix the memory leak in retriever.py",
+            api_url="http://127.0.0.1:8765",
+            token="tok-xyz",
+        )
+
+    assert result.get("action") == "block"
+    assert "thread-222" in result["reason"]
+    assert captured[0] == "Fix the memory leak in retriever.py"
+
+
+def test_delegate_empty_task_returns_usage_hint(hook: Any) -> None:
+    result = hook.run("/delegate", api_url="http://localhost:8765", token="tok")
+    assert result.get("action") == "block"
+    assert "Usage" in result["reason"] or "delegate" in result["reason"].lower()
+
+
+def test_missing_token_returns_block_with_hint(hook: Any) -> None:
+    import os
+    env_without_token = {k: v for k, v in os.environ.items() if k != "AGENT_TEAM_API_TOKEN"}
+    with patch.dict(os.environ, env_without_token, clear=True):
+        result = hook.run("/delegate Some task", api_url="http://localhost:8765", token="")
+    assert result.get("action") == "block"
+    assert "AGENT_TEAM_API_TOKEN" in result["reason"]
+
+
+def test_api_http_error_returns_block(hook: Any) -> None:
+    def raise_http(*_args: Any, **_kwargs: Any) -> Any:
+        raise urllib.error.HTTPError(
+            url="http://localhost:8765/tasks",
+            code=401,
+            msg="Unauthorized",
+            hdrs=None,  # type: ignore[arg-type]
+            fp=None,
+        )
+
+    with patch.object(hook, "_call_api", raise_http):
+        result = hook.run(
+            "/delegate Task that fails auth",
+            api_url="http://localhost:8765",
+            token="bad-token",
+        )
+
+    assert result.get("action") == "block"
+    assert "401" in result["reason"]
+
+
+def test_api_network_error_returns_block(hook: Any) -> None:
+    import urllib.error
+
+    def raise_url(*_args: Any, **_kwargs: Any) -> Any:
+        raise urllib.error.URLError("Connection refused")
+
+    with patch.object(hook, "_call_api", raise_url):
+        result = hook.run(
+            "/delegate Task when server is down",
+            api_url="http://localhost:8765",
+            token="tok",
+        )
+
+    assert result.get("action") == "block"
+    assert "run-team.py serve" in result["reason"] or "Connection refused" in result["reason"]
+
+
+def test_unexpected_exception_returns_block(hook: Any) -> None:
+    def explode(*_args: Any, **_kwargs: Any) -> Any:
+        raise ValueError("Totally unexpected")
+
+    with patch.object(hook, "_call_api", explode):
+        result = hook.run(
+            "/delegate Task",
+            api_url="http://localhost:8765",
+            token="tok",
+        )
+
+    assert result.get("action") == "block"
+    assert "Unexpected" in result["reason"] or "unexpected" in result["reason"].lower()
--- a/sea-haven-claude-plugin/CLAUDE.md
+++ b/sea-haven-claude-plugin/CLAUDE.md
@ -0,0 +1,76 @@
+# Sea Haven Industries — Claude Code Engineering Plugin
+
+This directory is the **Sea Haven Claude Code plugin**: context and hooks for
+Claude Code sessions run against the `sea-haven-industries/orchestrator` repo.
+Copy `settings.template.json` into `~/.claude/settings.json` (or merge it into
+an existing one) to activate the hooks.
+
+---
+
+## Project overview
+
+The orchestrator is the R720-hosted Plane-2 SDLC pipeline. It runs a
+multi-model multi-agent loop — **Claude Sonnet** (coordinator/clarifier/planner),
+**GPT-4.1** (cross-reviewer), **DeepSeek** (fast_coder/builder), **Gemini**
+(scanner) — gated by a CI trust boundary (`agent-team-apply-verify.yml`).
+
+Key directories:
+
+| Path | Purpose |
+|------|---------|
+| `agent-team/` | The pipeline package (`agent_team.*`) + tests |
+| `agent-team/run-team.py` | CLI: `serve`, `start`, `answer` sub-commands |
+| `agent-team/agent_team/coordinator.py` | Keystone: graph + transport wiring |
+| `agent-team/agent_team/graph.py` | LangGraph state machine (P1–P3+) |
+| `agent-team/agent_team/nodes/` | Pipeline node implementations |
+| `agent-team/agent_team/transport/` | Slack / GitHub / Claude Code adapters |
+| `.github/workflows/agent-team-apply-verify.yml` | CI trust boundary (NEVER edit directly; use a WS PR) |
+| `sea-haven-claude-plugin/` | This plugin |
+
+## Development rules
+
+1. **Green before merge**: `cd agent-team && python3 -m ruff check . && python3 -m pytest -q` must pass.
+2. **No secrets in code**: all tokens go in env vars or GitHub secrets.
+3. **No SSH/AWS/infra commands**: the R720 box is VPN-only; never attempt network access from a Claude Code session.
+4. **Draft PRs only**: all automated changes go out as `--draft`; humans merge.
+5. **CI YAML edits** require a security review (`/sh-security-review`) and a separate WS PR — never inline.
+
+## Delegating tasks to the pipeline
+
+### Via Slack
+
+Post `/new-task <description>` in any channel the Sea Haven bot is in. The
+listener routes it to the coordinator's `start_task`, which begins the
+CLARIFY → PLAN → REVIEW → BUILD → VERIFY pipeline. You'll receive a clarifier
+question back in Slack.
+
+### Via this Claude Code session (auto-delegate hook)
+
+Prefix your prompt with `/delegate ` (or `DELEGATE: `) to auto-send the task
+to the pipeline HTTP API instead of answering it locally:
+
+```
+/delegate Add OAuth2 to the admin portal — use PKCE, no client secret stored
+```
+
+The hook calls `POST http://127.0.0.1:8765/tasks` (requires
+`AGENT_TEAM_API_TOKEN` env var) and returns the `thread_id` so you can track
+the task in the coordinator.
+
+Set `AGENT_TEAM_API_URL` to override the default `http://127.0.0.1:8765`.
+Set `AGENT_TEAM_API_TOKEN` to the bearer token from `run-team.py serve`.
+
+## Available slash commands
+
+These are defined in `settings.template.json` (copy to `~/.claude/settings.json`):
+
+| Command | Effect |
+|---------|--------|
+| _(hooks auto-detect `/delegate` prefix)_ | Auto-delegate to pipeline |
+
+## Phases
+
+- **P1**: INTAKE → CLARIFY (human gate) → PLAN
+- **P2**: + adversarial REVIEW loop (GPT-4.1 cross-reviewer)
+- **P3**: + BUILD (DeepSeek) → VERIFY (CI gate)
+- **P3+**: + DISPATCH (push head branch + trigger `workflow_dispatch`)
--- a/sea-haven-claude-plugin/hooks/user_prompt_submit.py
+++ b/sea-haven-claude-plugin/hooks/user_prompt_submit.py
@ -0,0 +1,160 @@
+#!/usr/bin/env python3
+"""UserPromptSubmit hook: auto-delegate prefixed prompts to the agent-team API.
+
+Claude Code calls this script via the UserPromptSubmit hook whenever the user
+submits a prompt. If the prompt starts with ``/delegate`` (or ``DELEGATE:``),
+the hook forwards the task description to the agent-team HTTP API
+(``POST /tasks``) and blocks the prompt — the task is now running in the
+pipeline; no need to also answer it locally.
+
+Prompts that don't match the delegate prefix are passed through unchanged
+(exit 0 with no output).
+
+Configuration (env vars):
+  AGENT_TEAM_API_URL    Base URL of the agent-team HTTP API
+                        (default: http://127.0.0.1:8765)
+  AGENT_TEAM_API_TOKEN  Bearer token from ``run-team.py serve``
+                        (required when the API has token auth enabled)
+
+Claude Code hook contract:
+  * stdin:  JSON object with at least a ``prompt`` key.
+  * stdout: JSON object or empty.
+    - Empty / exit 0 → pass through (Claude answers normally).
+    - ``{"action": "block", "reason": "..."}`` → block the prompt; Claude
+      shows the reason to the user instead of answering.
+  * exit 0 = allowed (or delegated+blocked); non-zero = block with error.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+import urllib.error
+import urllib.request
+from typing import Any
+
+_DEFAULT_API_URL = "http://127.0.0.1:8765"
+_DELEGATE_PREFIXES = ("/delegate ", "DELEGATE: ")
+
+
+def _should_delegate(prompt: str) -> tuple[bool, str]:
+    """Return (True, task_text) when the prompt is a delegate command, else (False, '').
+
+    Matches ``/delegate <text>``, ``/delegate`` (bare — no text → usage hint),
+    and ``DELEGATE: <text>``.
+    """
+    stripped = prompt.strip()
+    for prefix in _DELEGATE_PREFIXES:
+        if stripped.startswith(prefix):
+            return True, stripped[len(prefix) :].strip()
+    # Bare /delegate with no trailing space or text.
+    if stripped == "/delegate":
+        return True, ""
+    return False, ""
+
+
+def _call_api(task: str, *, api_url: str, token: str) -> dict[str, Any]:
+    """POST /tasks and return the parsed response dict.
+
+    Raises urllib.error.URLError / urllib.error.HTTPError on network / HTTP
+    errors; the caller turns these into a block reason so Claude shows the error
+    to the user rather than silently passing through.
+    """
+    payload = json.dumps({"task": task, "transport": "claude_code"}).encode("utf-8")
+    headers: dict[str, str] = {"Content-Type": "application/json"}
+    if token:
+        headers["Authorization"] = f"Bearer {token}"
+    req = urllib.request.Request(
+        f"{api_url.rstrip('/')}/tasks",
+        data=payload,
+        headers=headers,
+        method="POST",
+    )
+    with urllib.request.urlopen(req, timeout=10) as resp:  # noqa: S310
+        return json.loads(resp.read().decode("utf-8"))
+
+
+def run(prompt: str, *, api_url: str = _DEFAULT_API_URL, token: str = "") -> dict[str, Any]:
+    """Core hook logic — pure function, fully testable without stdin/stdout.
+
+    Returns the hook output dict:
+    - ``{}`` to pass through (no delegation);
+    - ``{"action": "block", "reason": "..."}`` to block and show a message.
+    """
+    should, task_text = _should_delegate(prompt)
+    if not should:
+        return {}
+
+    if not task_text:
+        return {
+            "action": "block",
+            "reason": (
+                "Usage: /delegate <task description>\n"
+                "Example: /delegate Add OAuth2 to the admin portal"
+            ),
+        }
+
+    if not token:
+        env_token = os.environ.get("AGENT_TEAM_API_TOKEN", "")
+        if not env_token:
+            return {
+                "action": "block",
+                "reason": (
+                    "AGENT_TEAM_API_TOKEN is not set. "
+                    "Start the agent-team server with `run-team.py serve` "
+                    "and export the token it prints."
+                ),
+            }
+        token = env_token
+
+    try:
+        result = _call_api(task_text, api_url=api_url, token=token)
+    except urllib.error.HTTPError as exc:
+        return {
+            "action": "block",
+            "reason": f"agent-team API error {exc.code}: {exc.reason}. Is the server running?",
+        }
+    except urllib.error.URLError as exc:
+        return {
+            "action": "block",
+            "reason": (
+                f"Cannot reach agent-team API at {api_url}: {exc.reason}. "
+                "Is `run-team.py serve` running?"
+            ),
+        }
+    except Exception as exc:  # noqa: BLE001
+        return {
+            "action": "block",
+            "reason": f"Unexpected error delegating task: {exc}",
+        }
+
+    thread_id = result.get("thread_id", "(unknown)")
+    return {
+        "action": "block",
+        "reason": (
+            f"Task delegated to agent-team pipeline.\n"
+            f"thread_id: {thread_id}\n"
+            f"The coordinator will send a clarifying question via Slack."
+        ),
+    }
+
+
+def main() -> int:
+    try:
+        data = json.loads(sys.stdin.read())
+    except (json.JSONDecodeError, OSError):
+        return 0  # pass through on bad input
+
+    prompt = data.get("prompt", "") if isinstance(data, dict) else ""
+    api_url = os.environ.get("AGENT_TEAM_API_URL", _DEFAULT_API_URL)
+    token = os.environ.get("AGENT_TEAM_API_TOKEN", "")
+
+    output = run(prompt, api_url=api_url, token=token)
+    if output:
+        print(json.dumps(output))
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
--- a/sea-haven-claude-plugin/settings.template.json
+++ b/sea-haven-claude-plugin/settings.template.json
@ -0,0 +1,15 @@
+{
+  "hooks": {
+    "UserPromptSubmit": [
+      {
+        "matcher": "",
+        "hooks": [
+          {
+            "type": "command",
+            "command": "python3 ${SEA_HAVEN_PLUGIN_DIR}/hooks/user_prompt_submit.py"
+          }
+        ]
+      }
+    ]
+  }
+}