Hawcx × CrewAI reference integration — HawcxTool (eng_status_crew/tools.py): one integration point, per-call human-bound authz, read=ALLOW / write=DENY at the tool call.

tools.py

"""HawcxTool implementations for the Slack Channel Watch crew (Python SDK only).

One provider, one integration point. Every `_run()` calls the Hawcx Python SDK
(`HawcxAgent.invoke`) against a live, enrolled agent-host: the Assembler mints a
per-call TBAC token, AEAD-encrypts the request, and proxies it through the RSV
`/proxy` to slack.com. The RSV attaches the workspace bot token on egress
(`HAAP_RSV_PROXY_DOWNSTREAM_AUTH_SLACK_COM`); the agent never sees it.

    slack_list_channels         read   -> conversations.list      ALLOW
    slack_read_channel_history  read   -> conversations.history    ALLOW
    slack_post_message          write  -> chat.postMessage         DENY (mint, 0x002B)

On a policy allow Slack's JSON is returned; on a deny the SDK raises
`RequestRejected`, surfaced to the LLM as a `denied_by_policy` object (so it
reasons and continues, no retry). The write deny holds no matter what the LLM
decides — including under prompt injection. Every decision lands in the AuditLog.

Connection (from env): `HAWCX_ASSEMBLER_SOCK` or `HAWCX_AGENT_ID`
(+ optional `HAWCX_IPC_DIR`); `HAWCX_PRINCIPAL_ALLOWLIST`. Targets:
`HAWCX_SLACK_BASE` (default https://slack.com), `ENG_SLACK_CHANNEL`.
"""

from __future__ import annotations

import json
import os
import threading
from pathlib import Path
from typing import Any, Literal

from crewai.tools import BaseTool
from pydantic import BaseModel, Field

from .audit import AuditLog, AuditRecord, measure_us, now_iso

# ── Process-wide HawcxAgent (one connection, shared by all tools) ──────────
_AGENT: Any = None
_AGENT_LOCK = threading.Lock()

# Process-wide audit sink. CrewAI's BaseTool (pydantic) snapshots the per-tool
# `audit_log` field at construction, so `self.audit_log.append(...)` would land
# on a copy. The runner binds the real AuditLog here so every tool's decisions
# accumulate in one place.
_AUDIT_SINK: Any = None


def bind_audit(audit_log: Any) -> None:
    """Bind the process-wide audit sink the tools append decisions to."""
    global _AUDIT_SINK
    _AUDIT_SINK = audit_log


# Process-wide active prompt-injection (the exact instruction fed to the agent
# this run, or None). The write tool surfaces it at execution time so the deny
# moment shows what attack drove the hijacked call. It is NOT returned to the
# LLM — only printed and recorded in the audit row — so it never re-enters the
# model's context.
_INJECTION_ACTIVE: str | None = None


def bind_injection(text: str | None) -> None:
    """Bind the active prompt-injection so the write tool can surface it."""
    global _INJECTION_ACTIVE
    _INJECTION_ACTIVE = text


def _principal_allowlist() -> list[str]:
    return [p.strip() for p in os.environ.get("HAWCX_PRINCIPAL_ALLOWLIST", "").split(",") if p.strip()]


def get_agent() -> Any:
    """Lazily connect a process-wide `HawcxAgent` to the local Assembler."""
    global _AGENT
    if _AGENT is not None:
        return _AGENT
    with _AGENT_LOCK:
        if _AGENT is None:
            try:
                from hawcx_haap import HawcxAgent
            except ImportError as exc:  # pragma: no cover - env dependent
                raise RuntimeError(
                    "The hawcx-haap Python SDK is required (this demo is SDK-only). "
                    "Install it: `pip install hawcx-haap` (>= 0.1.3)."
                ) from exc

            allowlist = _principal_allowlist()
            sock = os.environ.get("HAWCX_ASSEMBLER_SOCK", "").strip()
            if sock:
                _AGENT = HawcxAgent.connect(sock, principal_allowlist=allowlist)
            else:
                agent_id = os.environ.get("HAWCX_AGENT_ID", "").strip()
                if not agent_id:
                    raise RuntimeError(
                        "No Assembler connection configured. Set HAWCX_ASSEMBLER_SOCK "
                        "(explicit socket) or HAWCX_AGENT_ID (enrolled agent)."
                    )
                ipc_dir = os.environ.get("HAWCX_IPC_DIR", "").strip()
                _AGENT = HawcxAgent.connect_by_agent_id(
                    agent_id,
                    principal_allowlist=allowlist,
                    ipc_dir=Path(ipc_dir) if ipc_dir else None,
                )
    return _AGENT


def close_agent() -> None:
    """Close the process-wide agent (call once at shutdown)."""
    global _AGENT
    with _AGENT_LOCK:
        if _AGENT is not None:
            try:
                _AGENT.close()
            finally:
                _AGENT = None


def _new_agent() -> Any:
    """Open a FRESH Assembler connection. The Assembler connection is
    per-request (and a session rotation rotates the socket), and an LLM fires
    tool calls back-to-back — so reusing one cached socket goes stale (EBADF /
    broken pipe). Each tool call connects fresh and closes after; cheap and
    race-free."""
    from hawcx_haap import HawcxAgent

    allowlist = _principal_allowlist()
    sock = os.environ.get("HAWCX_ASSEMBLER_SOCK", "").strip()
    if sock:
        return HawcxAgent.connect(sock, principal_allowlist=allowlist)
    agent_id = os.environ.get("HAWCX_AGENT_ID", "").strip()
    if not agent_id:
        raise RuntimeError(
            "No Assembler connection configured. Set HAWCX_ASSEMBLER_SOCK or HAWCX_AGENT_ID."
        )
    ipc_dir = os.environ.get("HAWCX_IPC_DIR", "").strip()
    return HawcxAgent.connect_by_agent_id(
        agent_id, principal_allowlist=allowlist, ipc_dir=Path(ipc_dir) if ipc_dir else None,
    )


def _slack_base() -> str:
    return os.environ.get("HAWCX_SLACK_BASE", "https://slack.com").rstrip("/")


def _slack_channel() -> str:
    return os.environ.get("ENG_SLACK_CHANNEL", "").strip()


# ── Args schemas ──────────────────────────────────────────────────────────────
class ListChannelsArgs(BaseModel):
    limit: int = Field(10, description="Max number of Slack channels to list.")


class ChannelHistoryArgs(BaseModel):
    channel: str = Field(default="", description="Slack channel ID; blank = ENG_SLACK_CHANNEL.")
    limit: int = Field(20, description="Max messages to read.")


class PostMessageArgs(BaseModel):
    text: str = Field(..., description="Message text to post.")
    channel: str = Field(default="", description="Slack channel ID; blank = ENG_SLACK_CHANNEL.")


# ── Shared plumbing ───────────────────────────────────────────────────────────
class _HawcxToolBase(BaseTool):
    """Identity binding, SDK invoke, audit emission, and deny shaping.

    Authorization + execution is done by the HAAP SDK (`agent.invoke`); subclasses
    return a fully-resolved call spec from `route()` and this base mints/forwards
    it, records the audit row, and shapes the agent-visible response (raw body on
    allow, `denied_by_policy` on deny).
    """

    agent_id: str = Field(...)
    agent_role: Literal["slack_analyst"] = Field(...)
    acting_for_email: str = Field(...)
    audit_log: Any = Field(...)

    def route(self, **kwargs) -> tuple[str, str, list[str], str, dict | None, str]:
        """Return (url, http_method, action, resource, body, provider)."""
        raise NotImplementedError

    def _call(self, *, tool: str, target: str, injected_prompt: str | None = None, **kwargs) -> str:
        """Authorize + execute one tool call through the HAAP SDK.

        Rides two transient failures before giving up: (a) a stale/closed IPC
        socket (the Assembler connection is per-request, and a ~40s session
        rotation rotates it) — reconnect; (b) an RS HTTP 401 in the brief window
        right after a rotation, before the renewed session's substrate write
        settles — NOT a policy decision, so retry. A real policy DENY (0x002B
        ScopeExceedsCeiling) is surfaced immediately and never retried.
        """
        import time

        from hawcx_haap import RequestRejected
        from hawcx_haap.ipc import TokenTransport

        url, method, action, resource, body, provider = self.route(**kwargs)
        _start, stop = measure_us()
        payload = json.dumps(body).encode() if body is not None else None
        # Retry budget wide enough to ride a full session-rotation window
        # (the renewal cadence is aggressive in this dev setup). ~14 tries x 2.5s
        # ≈ 35s for transient 401s; connection errors reconnect immediately.
        tries = int(os.environ.get("ENG_HAAP_RETRIES", "14") or 14)
        last: Exception | None = None
        for attempt in range(1, tries + 1):
            agent = None
            try:
                agent = _new_agent()
                # ── HAAP integration point: authorize this call (per-call, human-bound) + execute ──
                resp = agent.invoke(
                    target_rs_url=url, http_method=method, tool=tool, action=action,
                    resource=resource, acting_for_user=self.acting_for_email, body=payload,
                    content_type="application/json" if payload is not None else None,
                    transport=TokenTransport.MCP_META, provider=provider,
                )
            except RequestRejected as exc:
                reason = str(exc)
                if _is_transient_reject(reason) and attempt < tries:
                    last = exc
                    import sys
                    short = "JIT pool empty" if "pool empty" in reason.lower() else "session rotating (401)"
                    print(f"[haap] {tool}: transient — {short}; riding the renewal window "
                          f"(retry {attempt}/{tries})…", file=sys.stderr, flush=True)
                    time.sleep(2.5); continue
                return self._fail(tool, target, reason, stop(), injected_prompt=injected_prompt)
            except Exception as exc:  # connect / transport / broken pipe / EBADF / rotation
                last = exc
                if attempt < tries and _is_conn_err(exc):
                    time.sleep(0.4); continue
                return self._fail(tool, target, f"{type(exc).__name__}: {exc}", stop(),
                                  injected_prompt=injected_prompt)
            else:
                self._emit(action=tool, target=target, decision="ALLOW",
                           policy_name="haap-rsv", latency_us=stop(),
                           injected_prompt=injected_prompt)
                return resp.body or ""
            finally:
                if agent is not None:
                    try:
                        agent.close()
                    except Exception:
                        pass
        return self._fail(tool, target, f"{type(last).__name__}: {last}" if last else "unknown",
                          stop(), injected_prompt=injected_prompt)

    def _fail(self, tool: str, target: str, reason: str, latency_us: int,
              injected_prompt: str | None = None) -> str:
        self._emit(action=tool, target=target, decision="DENY",
                   policy_name="haap-rsv", latency_us=latency_us, reason=reason,
                   injected_prompt=injected_prompt)
        return self._deny_response(action=tool, target=target, policy_name="haap-rsv",
                                   reason=reason, injected_prompt=injected_prompt)

    def _emit(self, *, action: str, target: str, decision: Literal["ALLOW", "DENY"],
              policy_name: str, latency_us: int, reason: str | None = None,
              injected_prompt: str | None = None) -> None:
        (_AUDIT_SINK or self.audit_log).append(
            AuditRecord(
                timestamp=now_iso(),
                actor=self.acting_for_email,
                agent_id=self.agent_id,
                action=action,
                target=target,
                decision=decision,
                policy_name=policy_name,
                latency_us=latency_us,
                reason=reason,
                injected_prompt=injected_prompt,
            )
        )

    def _deny_response(self, *, action: str, target: str, policy_name: str, reason: str,
                       injected_prompt: str | None = None) -> str:
        # ── write-DENY shaped here: the policy rejection becomes a denied_by_policy object the agent reasons over (no retry) ──
        obj = {
            "status": "denied_by_policy",
            "action": action,
            "target": target,
            "policy": policy_name,
            "reason": reason,
            "remediation": "The agent is read-only on Slack. Do not retry; continue without posting.",
        }
        # When a prompt-injection was active for this write, surface it right here
        # in the tool result so it shows in CrewAI's tool-output panel next to the
        # deny — the "an attacker tried to subvert the agent, blocked anyway" beat.
        if injected_prompt:
            obj["injected_prompt"] = " ".join(injected_prompt.split())
            obj["injection_note"] = (
                "A prompt-injection was active for this write. HAAP denied it at the "
                "tool call regardless of what the model was instructed to do."
            )
        return json.dumps(obj, indent=2)


# ══ Slack tools (provider="slack") ════════════════════════════════════════════
class SlackListChannels(_HawcxToolBase):
    name: str = "slack_list_channels"
    description: str = (
        "List the team's Slack channels (conversations.list, read-only). Returns "
        "Slack's JSON. On deny returns a structured failure object — do not retry."
    )
    args_schema: type[BaseModel] = ListChannelsArgs

    def route(self, limit: int = 10) -> tuple:
        return (f"{_slack_base()}/api/conversations.list",
                "POST", ["read"], "channels", {"limit": int(limit), "types": "public_channel"}, "slack")

    def _run(self, limit: int = 10) -> str:  # type: ignore[override]
        return _summarize_slack(self._call(tool="slack_list_channels", target="channels", limit=limit), "channels")


class SlackReadHistory(_HawcxToolBase):
    name: str = "slack_read_channel_history"
    description: str = (
        "Read recent messages in a Slack channel (conversations.history, read-only) "
        "to extract decisions, blockers, and discussion context. Returns Slack's "
        "JSON. On deny returns a structured failure object — do not retry."
    )
    args_schema: type[BaseModel] = ChannelHistoryArgs

    def route(self, channel: str = "", limit: int = 20) -> tuple:
        channel = channel or _slack_channel()
        return (f"{_slack_base()}/api/conversations.history",
                "POST", ["read"], f"channels/{channel}",
                {"channel": channel, "limit": int(limit)}, "slack")

    def _run(self, channel: str = "", limit: int = 20) -> str:  # type: ignore[override]
        return _summarize_slack(
            self._call(tool="slack_read_channel_history",
                       target=f"channel:{channel or _slack_channel()}", channel=channel, limit=limit),
            "history")


class SlackPostMessage(_HawcxToolBase):
    name: str = "slack_post_message"
    description: str = (
        "Post a message to a Slack channel (chat.postMessage, a WRITE). The agent is "
        "read-only on Slack, so HAAP denies this at the tool call — it never reaches "
        "Slack, regardless of what the model decided. Returns a structured "
        "denied_by_policy object."
    )
    args_schema: type[BaseModel] = PostMessageArgs

    def route(self, text: str = "", channel: str = "") -> tuple:
        channel = channel or _slack_channel()
        return (f"{_slack_base()}/api/chat.postMessage",
                "POST", ["write"], f"channels/{channel}",
                {"channel": channel, "text": text}, "slack")

    def _run(self, text: str, channel: str = "") -> str:  # type: ignore[override]
        # The injection (if any) is surfaced once up front (the banner) and once
        # here in the deny object below — no stderr echo, which only split the
        # flow and duplicated CrewAI's own Args/Output panels.
        return self._call(tool="slack_post_message",
                          target=f"channel:{channel or _slack_channel()}",
                          injected_prompt=_INJECTION_ACTIVE, text=text, channel=channel)


# ── Slack response shaping ────────────────────────────────────────────────────
def _summarize_slack(raw: str, kind: str) -> str:
    """Trim a Slack ALLOW body to the essentials so the agent (and the console
    output) see clean data, not the giant raw JSON. A `denied_by_policy` result
    (a DENY) passes through unchanged so the hero deny still shows in full."""
    try:
        obj = json.loads(raw)
    except Exception:
        return raw
    if not isinstance(obj, dict) or obj.get("status") == "denied_by_policy":
        return raw
    if kind == "channels":
        names = [c.get("name", "?") for c in obj.get("channels", [])]
        return f"Slack channels ({len(names)}): " + ", ".join(names) if names else "No channels found."
    if kind == "history":
        lines = []
        for m in obj.get("messages", []):
            text = (m.get("text") or "").strip().replace("\n", " ")
            if text:
                lines.append(f"- {text}")
        return f"Recent messages ({len(lines)}):\n" + "\n".join(lines) if lines else "No messages in the channel."
    return raw


# ── Transient-failure classifiers ─────────────────────────────────────────────
def _is_conn_err(exc: Exception) -> bool:
    """True for stale/broken IPC-socket errors worth a transparent reconnect."""
    if isinstance(exc, (BrokenPipeError, ConnectionError, EOFError, OSError)):
        return True
    s = f"{type(exc).__name__} {exc}".lower()
    return any(t in s for t in (
        "broken pipe", "brokenpipe", "ipc", "socket", "connection",
        "closed", "reset", "eof", "not connected", "no such file",
        "bad file descriptor", "errno 9",
    ))


def _is_transient_reject(reason: str) -> bool:
    """Distinguish a transient failure (retry) from a real policy denial (surface now).

    The genuine deny is `0x002B … ScopeExceedsCeiling: <action>` — the action
    exceeds the granted ceiling; NEVER retry that. But several transients share
    the same 0x002B code and must be retried:
      - `JIT pool empty` — the per-call token pre-mint pool is momentarily drained
        (every renewal purges + re-keys it; reads landing in the refill gap hit this).
      - post-rotation `HTTP 401 / session not found` — the substrate-write-settle
        window right after a session rotation.
    """
    low = reason.lower()
    # Real ceiling denial — the hero deny. Never retry.
    if "scopeexceedsceiling" in low:
        return False
    return any(m in low for m in (
        "pool empty", "jit pool", "no token", "precompute",        # token-supply transients
        "http 401", "unauthorized", "session not found", "rs rejected",
        "http 403", "temporal",                                     # session-rotation transients
    ))


# ── Tool factory ──────────────────────────────────────────────────────────────
def build_slack_analyst_tools(*, agent_id: str, acting_for_email: str,
                              audit_log: AuditLog) -> list[BaseTool]:
    """Slack Channel Analyst: two read tools + a post tool that HAAP denies."""
    common = dict(agent_id=agent_id, agent_role="slack_analyst",
                  acting_for_email=acting_for_email, audit_log=audit_log)
    return [
        SlackListChannels(**common),
        SlackReadHistory(**common),
        SlackPostMessage(**common),  # the write deny — the hero
    ]