orchestrata_ultimate.tools.remote_trigger_tool

Reference › orchestrata_ultimate › tools › remote_trigger_tool

RemoteTriggerTool (SPEC-067) sends an outbound HTTP request, such as a webhook or REST call, with pluggable authentication, bounded retry with exponential backoff, timeout handling and response validation. It uses httpx. It is a BaseTool, registered by agents.tool_registry and register_remote_trigger_tool, and exported from the package __init__. tests/tools/test_remote_trigger_tool.py covers it.

The pieces are small classes composed by the tool:

• Configuration dataclasses: AuthConfig, RetryConfig, ValidationConfig, TriggerRequest and TriggerResponse.
• SSRFGuard blocks loopback, private, link-local and reserved targets and known metadata endpoints, by resolving the host once. It does not defend against DNS rebinding and has no allowlist. Per-request allow_internal or the environment variable REMOTE_TRIGGER_ALLOW_PRIVATE lifts the block.
• AuthManager applies API key (header or query), bearer, basic, OAuth2 client credentials (one token fetch per request, no caching) and HMAC signing. Secret values can be given as literals or as environment variable names resolved by _resolve_secret.
• RetryExecutor retries on configured status codes and optionally on timeouts, honours Retry-After, and returns the response and the retry count; it returns no response when every attempt timed out.
• ResponseValidator checks a JSON Schema subset: type, required, properties and items.

One httpx client is built per call and reused across that call's attempts. Failures are reported as RemoteTriggerError with a RemoteTriggerErrorCode, and become failed ToolResult values.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

RemoteTriggerErrorCode

Decorators: traces(spec='067', story='US-067-001', reqs=['REQ-067-023']) · Line 110

String error codes for the RemoteTriggerTool pipeline (spec.md ERR-00N).

RemoteTriggerError

Bases: Exception · Decorators: traces(spec='067', story='US-067-001', reqs=['REQ-067-023']) · Line 134

Raised for any spec.md ERR-00N condition; renders as a failed ToolResult.

AuthConfig (dataclass)

Decorators: traces(spec='067', story='US-067-002', reqs=['REQ-067-002', 'REQ-067-003']), dataclass(slots=True, frozen=True) · Line 157

Resolved authentication configuration (spec.md AuthConfig, all variants).

RetryConfig (dataclass)

Decorators: traces(spec='067', story='US-067-003', reqs=['REQ-067-008', 'REQ-067-009']), dataclass(slots=True, frozen=True) · Line 221

Retry behaviour (spec.md RetryConfig).

ValidationConfig (dataclass)

Decorators: traces(spec='067', story='US-067-004', reqs=['REQ-067-011', 'REQ-067-012', 'REQ-067-013', 'REQ-067-014']), dataclass(slots=True, frozen=True) · Line 240

Response validation rules (spec.md ValidationConfig).

TriggerRequest (dataclass)

Decorators: traces(spec='067', story='US-067-001', reqs=['REQ-067-001', 'REQ-067-007']), dataclass(slots=True, frozen=True) · Line 258

A validated remote-trigger request (spec.md TriggerRequest).

TriggerResponse (dataclass)

Decorators: traces(spec='067', story='US-067-001', reqs=['REQ-067-001', 'REQ-067-018']), dataclass(slots=True) · Line 386

Result of a trigger execution (spec.md TriggerResponse).

SSRFGuard

Decorators: traces(spec='067', story='US-067-005', reqs=['REQ-067-020']) · Line 658

Blocks outbound requests to loopback/private/link-local/metadata targets.

AuthManager

Decorators: traces(spec='067', story='US-067-002', reqs=['REQ-067-002', 'REQ-067-003', 'REQ-067-004', 'REQ-067-005', 'REQ-067-006']) · Line 703

Applies the configured :class:AuthConfig as request headers/params.

RetryExecutor

Decorators: traces(spec='067', story='US-067-003', reqs=['REQ-067-008', 'REQ-067-009', 'REQ-067-010']) · Line 823

Bounded exponential-backoff retry over an async request callable.

ResponseValidator

Decorators: traces(spec='067', story='US-067-004', reqs=['REQ-067-011', 'REQ-067-012', 'REQ-067-013', 'REQ-067-014']) · Line 874

Validates a response against a :class:ValidationConfig (FR-005).

RemoteTriggerTool

Bases: BaseTool · Decorators: traces(spec='067', story='US-067-001', reqs=['REQ-067-001', 'REQ-067-007', 'REQ-067-015', 'REQ-067-016', 'REQ-067-017',… · Line 1037

The registrable BaseTool (Command) callers/LLMs invoke.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/tools/remote_trigger_tool.py

"""RemoteTriggerTool — outbound webhook/API trigger with auth + retry/backoff.

Spec: SPEC-067 (RemoteTriggerTool). A :class:`~orchestrata_ultimate.tools.base.BaseTool`
that sends an HTTP request (webhook or REST call) with pluggable authentication,
bounded exponential-backoff retry, timeout handling, and response validation —
built on ``httpx`` (already declared in ``pyproject.toml``; no new dependency).

**MVP scope / forks from spec.md** (reported to the orchestrator, not silently
guessed):

* **No pydantic models.** spec.md's API Contracts sketch ``TriggerRequest`` /
  ``RemoteTriggerConfig`` as pydantic ``BaseModel``\\s; this module uses plain
  ``@dataclass``\\es instead, matching the prevailing style of the other
  SPEC-06x tools (``tools/search.py``, ``tools/worktree_tools.py``) rather than
  introducing a second modelling convention.
* **OAuth2** implements a single client-credentials token fetch per request
  (no token caching/refresh-lifecycle — spec.md's "Manages OAuth2 token
  refresh lifecycle" is a multi-call optimization, out of MVP scope).
* **JSON Schema validation** (FR-005) is a lightweight subset: ``type``,
  ``required``, ``properties``, ``items`` — not full JSON Schema draft-07
  (no ``format``/``pattern``/``enum``/``$ref``; no external ``jsonschema``
  dependency is declared in ``pyproject.toml``, and adding one is out of
  scope for this brief).
* **SSRF protection** (Security Requirements) resolves the target hostname
  once via :func:`socket.getaddrinfo` and blocks loopback/private/link-local/
  reserved addresses and known metadata endpoints — mirroring the existing
  ``agents.tool_registry._is_blocked_address``/``_web_tools_allow_private``
  pattern for ``web_fetch``. It does **not** implement DNS-rebinding
  protection (re-resolving between check and connect) or an allowlist;
  ``allow_internal=True`` (per-request) or the ``REMOTE_TRIGGER_ALLOW_PRIVATE``
  env var (read at call time, never hardcoded — REQ project rule) are the
  sanctioned escape hatches, used by this module's own tests to target a real
  local HTTP server.
* **Connection pooling** (FR-007) is scoped to the lifetime of a single
  ``execute``/``execute_simple`` call: one ``httpx.AsyncClient`` (with
  ``httpx.Limits``) is built per call and its pool is reused across that
  call's own retry attempts. A cross-call persistent client is deferred —
  the stateless ``BaseTool.execute(arguments) -> ToolResult`` contract has no
  lifecycle hook (no ``close()``/``aclose()`` on the Protocol) to tear one
  down safely.
* **Query parameters** (``params``) and **field extraction**
  (``extract_fields``) are MVP additions beyond spec.md's literal
  ``TriggerRequest`` JSON Schema (needed to support ``api_key`` auth placed
  "in": "query", and FR-009's "extract specific fields from responses").
* ``execute_simple``'s ``auth`` parameter is typed ``dict[str, Any] | None``
  (the same raw shape ``TriggerRequest.from_arguments`` parses), not spec.md's
  ``AuthConfig`` object — :class:`AuthConfig` is this module's internal,
  already-resolved representation, not a caller-facing constructor.

Architecture: :class:`RemoteTriggerTool` (the registrable ``BaseTool``)
composes :class:`SSRFGuard`, :class:`AuthManager`, :class:`RetryExecutor`, and
:class:`ResponseValidator` — each a focused, independently testable unit.
"""

from __future__ import annotations

import base64
import hashlib
import hmac
import ipaddress
import itertools
import json
import logging
import os
import random
import socket
import time
import urllib.parse
from collections.abc import Awaitable, Callable, Mapping
from dataclasses import dataclass, field
from datetime import UTC, datetime
from typing import Any

import httpx

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.tools.base import (
    BaseTool,
    ToolCategory,
    ToolMetadata,
    ToolPriority,
    ToolResult,
)
from orchestrata_ultimate.tools.registry import ToolRegistry

logger = logging.getLogger(__name__)

# ── constants ────────────────────────────────────────────────────────────

_ALLOWED_METHODS = frozenset(
    {"GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"}
)
_ALLOWED_AUTH_KINDS = frozenset({"api_key", "bearer", "basic", "oauth2", "hmac"})
_ALLOWED_HMAC_ALGORITHMS = frozenset({"sha256", "sha512", "sha1"})
_DEFAULT_RETRY_STATUSES: tuple[int, ...] = (429, 500, 502, 503, 504)
_DEFAULT_SENSITIVE_HEADERS = frozenset({"authorization", "x-api-key"})
_BLOCKED_HOSTNAMES = frozenset(
    {"metadata.google.internal", "metadata", "metadata.internal"}
)
# The fixed IANA link-local address every cloud provider (AWS/GCP/Azure/
# DigitalOcean) uses for its metadata service — an SSRF blocklist constant,
# not deployment infra (link-local is skip-listed by the HARDCODED_IP gate).
_METADATA_IPS = frozenset({"169.254.169.254"})


# ── error codes / exception ─────────────────────────────────────────────


@traces(spec="067", story="US-067-001", reqs=["REQ-067-023"])
class RemoteTriggerErrorCode:
    """String error codes for the RemoteTriggerTool pipeline (spec.md ERR-00N)."""

    INVALID_URL = "invalid_url"
    AUTH_FAILED = "auth_failed"
    FORBIDDEN = "forbidden"
    NOT_FOUND = "not_found"
    RATE_LIMITED = "rate_limited"
    SERVER_ERROR = "server_error"
    TIMEOUT = "timeout"
    CONNECTION_ERROR = "connection_error"
    SSL_ERROR = "ssl_error"
    VALIDATION_FAILED = "validation_failed"
    INVALID_SIGNATURE = "invalid_signature"
    RETRY_EXHAUSTED = "retry_exhausted"
    INVALID_CONFIG = "invalid_config"
    BODY_TOO_LARGE = "body_too_large"
    DNS_FAILURE = "dns_failure"
    INVALID_JSON = "invalid_json"
    CONTENT_TYPE_MISMATCH = "content_type_mismatch"
    SSRF_BLOCKED = "ssrf_blocked"


@traces(spec="067", story="US-067-001", reqs=["REQ-067-023"])
class RemoteTriggerError(Exception):
    """Raised for any spec.md ERR-00N condition; renders as a failed ToolResult."""

    def __init__(self, code: str, message: str) -> None:
        super().__init__(message)
        self.code = code
        self.message = message

    @traces(spec="067", reqs=["REQ-067-023"])
    def to_result(self, tool_name: str) -> ToolResult:
        return ToolResult(
            tool_name=tool_name,
            success=False,
            output="",
            error=f"{self.code}: {self.message}",
        )


# ── auth / retry / validation config ────────────────────────────────────


@traces(spec="067", story="US-067-002", reqs=["REQ-067-002", "REQ-067-003"])
@dataclass(slots=True, frozen=True)
class AuthConfig:
    """Resolved authentication configuration (spec.md AuthConfig, all variants).

    A single flat dataclass discriminated by ``kind`` rather than 5 separate
    types (spec.md's ``oneOf`` union) — simpler to build from a raw tool-call
    ``arguments["auth"]`` dict and to pattern-match on in :class:`AuthManager`.
    """

    kind: str
    key: str = ""
    value: str = ""
    value_env: str = ""
    location: str = "header"
    token: str = ""
    token_env: str = ""
    username: str = ""
    password: str = ""
    password_env: str = ""
    client_id: str = ""
    client_secret: str = ""
    client_secret_env: str = ""
    token_url: str = ""
    scopes: tuple[str, ...] = ()
    grant_type: str = "client_credentials"
    secret: str = ""
    secret_env: str = ""
    algorithm: str = "sha256"
    header_name: str = "X-Signature"

    @classmethod
    @traces(spec="067", story="US-067-002", reqs=["REQ-067-002", "REQ-067-003"])
    def from_dict(cls, raw: dict[str, Any]) -> AuthConfig:
        kind = str(raw.get("type", "")).lower()
        if kind not in _ALLOWED_AUTH_KINDS:
            raise RemoteTriggerError(
                RemoteTriggerErrorCode.INVALID_CONFIG,
                f"Unknown auth type: {raw.get('type')!r}",
            )
        return cls(
            kind=kind,
            key=str(raw.get("key", "")),
            value=str(raw.get("value", "")),
            value_env=str(raw.get("value_env", "")),
            location=str(raw.get("in", "header")),
            token=str(raw.get("token", "")),
            token_env=str(raw.get("token_env", "")),
            username=str(raw.get("username", "")),
            password=str(raw.get("password", "")),
            password_env=str(raw.get("password_env", "")),
            client_id=str(raw.get("client_id", "")),
            client_secret=str(raw.get("client_secret", "")),
            client_secret_env=str(raw.get("client_secret_env", "")),
            token_url=str(raw.get("token_url", "")),
            scopes=tuple(raw.get("scopes", ()) or ()),
            grant_type=str(raw.get("grant_type", "client_credentials")),
            secret=str(raw.get("secret", "")),
            secret_env=str(raw.get("secret_env", "")),
            algorithm=str(raw.get("algorithm", "sha256")),
            header_name=str(raw.get("header_name", "X-Signature")),
        )


@traces(spec="067", story="US-067-003", reqs=["REQ-067-008", "REQ-067-009"])
@dataclass(slots=True, frozen=True)
class RetryConfig:
    """Retry behaviour (spec.md RetryConfig)."""

    enabled: bool = True
    max_attempts: int = 3
    initial_delay: float = 1.0
    max_delay: float = 60.0
    backoff_factor: float = 2.0
    retry_on_status: tuple[int, ...] = _DEFAULT_RETRY_STATUSES
    retry_on_timeout: bool = True
    jitter: bool = False


@traces(
    spec="067",
    story="US-067-004",
    reqs=["REQ-067-011", "REQ-067-012", "REQ-067-013", "REQ-067-014"],
)
@dataclass(slots=True, frozen=True)
class ValidationConfig:
    """Response validation rules (spec.md ValidationConfig)."""

    expected_status: tuple[int, ...] = ()
    min_response_time: float | None = None
    max_response_time: float | None = None
    schema: dict[str, Any] | None = None
    signature_secret: str = ""
    signature_secret_env: str = ""
    signature_header: str = "X-Signature"
    signature_algorithm: str = "sha256"


# ── request / response ──────────────────────────────────────────────────


@traces(spec="067", story="US-067-001", reqs=["REQ-067-001", "REQ-067-007"])
@dataclass(slots=True, frozen=True)
class TriggerRequest:
    """A validated remote-trigger request (spec.md TriggerRequest)."""

    url: str
    method: str = "POST"
    headers: dict[str, str] = field(default_factory=dict)
    body: Any = None
    params: dict[str, str] = field(default_factory=dict)
    auth: AuthConfig | None = None
    # int per the public tool-call JSON schema (spec.md: integer, 1-300s);
    # widened to accept float here too so callers constructing a
    # TriggerRequest directly (bypassing from_arguments's int() cast) can use
    # sub-second timeouts — httpx.Timeout accepts a float natively.
    timeout: int | float = 30
    retry: RetryConfig = field(default_factory=RetryConfig)
    validation: ValidationConfig = field(default_factory=ValidationConfig)
    verify_ssl: bool = True
    ca_bundle_path: str | None = None
    connection_pool_size: int = 10
    allow_internal: bool = False
    extract_fields: tuple[str, ...] = ()
    env: dict[str, str] | None = None

    @classmethod
    @traces(spec="067", story="FR-003", reqs=["REQ-067-007", "REQ-067-023"])
    def from_arguments(
        cls, arguments: dict[str, Any], env: Mapping[str, str] | None = None
    ) -> TriggerRequest:
        """Build a :class:`TriggerRequest` from raw tool-call ``arguments``.

        Raises:
            RemoteTriggerError: ``INVALID_URL`` (ERR-001) for a missing/
                malformed URL, ``INVALID_CONFIG`` (ERR-013) for any other
                malformed field.
        """
        url = str(arguments.get("url") or "").strip()
        parsed = urllib.parse.urlsplit(url)
        if (
            not url
            or parsed.scheme.lower() not in ("http", "https")
            or not parsed.hostname
        ):
            raise RemoteTriggerError(
                RemoteTriggerErrorCode.INVALID_URL,
                f"Invalid or malformed URL: {url}",
            )
        method = str(arguments.get("method", "POST")).upper()
        if method not in _ALLOWED_METHODS:
            raise RemoteTriggerError(
                RemoteTriggerErrorCode.INVALID_CONFIG,
                f"Unsupported HTTP method: {method}",
            )
        auth_raw = arguments.get("auth")
        auth = AuthConfig.from_dict(auth_raw) if auth_raw else None

        retry_raw = arguments.get("retry") or {}
        validation_raw = arguments.get("validation") or {}
        try:
            retry = RetryConfig(
                enabled=bool(retry_raw.get("enabled", True)),
                max_attempts=int(retry_raw.get("max_attempts", 3)),
                initial_delay=float(retry_raw.get("initial_delay", 1.0)),
                max_delay=float(retry_raw.get("max_delay", 60.0)),
                backoff_factor=float(retry_raw.get("backoff_factor", 2.0)),
                retry_on_status=tuple(
                    int(s)
                    for s in retry_raw.get("retry_on_status", _DEFAULT_RETRY_STATUSES)
                ),
                retry_on_timeout=bool(retry_raw.get("retry_on_timeout", True)),
                jitter=bool(retry_raw.get("jitter", False)),
            )
            validation = ValidationConfig(
                expected_status=tuple(
                    int(s) for s in validation_raw.get("expected_status", ())
                ),
                min_response_time=(
                    float(validation_raw["min_response_time"])
                    if validation_raw.get("min_response_time") is not None
                    else None
                ),
                max_response_time=(
                    float(validation_raw["max_response_time"])
                    if validation_raw.get("max_response_time") is not None
                    else None
                ),
                schema=validation_raw.get("schema"),
                signature_secret=str(validation_raw.get("signature_secret", "")),
                signature_secret_env=str(
                    validation_raw.get("signature_secret_env", "")
                ),
                signature_header=str(
                    validation_raw.get("signature_header", "X-Signature")
                ),
                signature_algorithm=str(
                    validation_raw.get("signature_algorithm", "sha256")
                ),
            )
            timeout = int(arguments.get("timeout", 30))
            pool_size = int(arguments.get("connection_pool_size", 10))
        except (TypeError, ValueError) as exc:
            raise RemoteTriggerError(
                RemoteTriggerErrorCode.INVALID_CONFIG,
                f"Configuration error: {exc}",
            ) from exc

        return cls(
            url=url,
            method=method,
            headers={
                str(k): str(v) for k, v in (arguments.get("headers") or {}).items()
            },
            body=arguments.get("body"),
            params={str(k): str(v) for k, v in (arguments.get("params") or {}).items()},
            auth=auth,
            timeout=timeout,
            retry=retry,
            validation=validation,
            verify_ssl=bool(arguments.get("verify_ssl", True)),
            ca_bundle_path=arguments.get("ca_bundle_path"),
            connection_pool_size=pool_size,
            allow_internal=bool(arguments.get("allow_internal", False)),
            extract_fields=tuple(arguments.get("extract_fields", ()) or ()),
            env=dict(env) if env is not None else None,
        )


@traces(spec="067", story="US-067-001", reqs=["REQ-067-001", "REQ-067-018"])
@dataclass(slots=True)
class TriggerResponse:
    """Result of a trigger execution (spec.md TriggerResponse)."""

    success: bool
    status_code: int
    status_text: str = ""
    timestamp: str = ""
    elapsed_ms: int = 0
    headers: dict[str, str] = field(default_factory=dict)
    body: Any = None
    extracted: dict[str, Any] = field(default_factory=dict)
    validation_passed: bool = True
    validation_errors: list[str] = field(default_factory=list)
    retry_count: int = 0
    error: str | None = None

    @traces(spec="067", story="US-067-001", reqs=["REQ-067-001", "REQ-067-018"])
    def to_json(self) -> str:
        return json.dumps(
            {
                "success": self.success,
                "status_code": self.status_code,
                "status_text": self.status_text,
                "timestamp": self.timestamp,
                "elapsed_ms": self.elapsed_ms,
                "headers": self.headers,
                "body": self.body,
                "extracted": self.extracted,
                "validation_passed": self.validation_passed,
                "validation_errors": self.validation_errors,
                "retry_count": self.retry_count,
                "error": self.error,
            },
            default=str,
        )


# ── helpers: env / body / logging / status messages ────────────────────


@traces(spec="067", reqs=["REQ-067-003", "REQ-067-004", "REQ-067-006"])
def _resolve_secret(
    literal: str, env_name: str, env: Mapping[str, str] | None, field_label: str
) -> str:
    """Resolve a secret from a literal value or an ``*_env`` variable name.

    Reads the environment at call time via ``os.environ.get`` (never a bare
    subscript) — an optional ``env`` mapping lets callers (tests) override it
    without mutating ``os.environ``.
    """
    if env_name:
        source = os.environ if env is None else env
        value = source.get(env_name, "")
        if not value:
            raise RemoteTriggerError(
                RemoteTriggerErrorCode.AUTH_FAILED,
                f"Environment variable {env_name!r} for {field_label} is not set",
            )
        return value
    if literal:
        return literal
    raise RemoteTriggerError(
        RemoteTriggerErrorCode.INVALID_CONFIG,
        f"Missing {field_label}: provide a literal value or an *_env variable name",
    )


@traces(spec="067", story="FR-002", reqs=["REQ-067-006"])
def _encode_body(body: Any) -> bytes:
    """Render a request body to bytes (for HMAC signing)."""
    if body is None:
        return b""
    if isinstance(body, (dict, list)):
        return json.dumps(body, sort_keys=True).encode()
    if isinstance(body, str):
        return body.encode()
    return str(body).encode()


@traces(spec="067", story="FR-002", reqs=["REQ-067-006"])
def _hmac_digest(secret: str, payload: bytes, algorithm: str) -> str:
    algorithm = algorithm.lower()
    if algorithm not in _ALLOWED_HMAC_ALGORITHMS:
        raise RemoteTriggerError(
            RemoteTriggerErrorCode.INVALID_CONFIG,
            f"Unsupported HMAC algorithm: {algorithm}",
        )
    digestmod = getattr(hashlib, algorithm)
    return hmac.new(secret.encode(), payload, digestmod).hexdigest()


@traces(spec="067", story="FR-005", reqs=["REQ-067-014", "REQ-067-022"])
def _verify_hmac(payload: bytes, signature: str, secret: str, algorithm: str) -> bool:
    """Constant-time HMAC comparison; tolerates a ``"sha256=<hex>"`` prefix."""
    try:
        expected = _hmac_digest(secret, payload, algorithm)
    except RemoteTriggerError:
        return False
    sig = (signature or "").strip()
    if "=" in sig:
        _, _, sig = sig.partition("=")
    try:
        return hmac.compare_digest(expected, sig)
    except TypeError:
        return False


@traces(spec="067", story="FR-010", reqs=["REQ-067-019"])
def _sanitize_url(url: str) -> str:
    """Strip query string (may carry an API key) before logging a URL."""
    parsed = urllib.parse.urlsplit(url)
    return urllib.parse.urlunsplit((parsed.scheme, parsed.netloc, parsed.path, "", ""))


@traces(spec="067", story="FR-010", reqs=["REQ-067-019"])
def _sanitize_headers(
    headers: dict[str, str], sensitive: frozenset[str] = _DEFAULT_SENSITIVE_HEADERS
) -> dict[str, str]:
    return {
        k: ("***REDACTED***" if k.lower() in sensitive else v)
        for k, v in headers.items()
    }


def _now_iso() -> str:
    return datetime.now(UTC).isoformat()


@traces(spec="067", reqs=["REQ-067-023"])
def _status_error_message(
    status: int, url: str, retry_count: int, retry_cfg: RetryConfig
) -> str:
    """Map a non-2xx status to a spec.md ERR-00N message."""
    if (
        retry_cfg.enabled
        and retry_cfg.max_attempts > 1
        and status in retry_cfg.retry_on_status
        and retry_count >= retry_cfg.max_attempts - 1
    ):
        return f"Max retry attempts ({retry_cfg.max_attempts}) exceeded"
    if status == 401:
        return "Authentication failed: invalid or missing credentials"
    if status == 403:
        return f"Access forbidden to {url}"
    if status == 404:
        return f"Endpoint not found: {url}"
    if status == 413:
        return "Request body exceeds maximum allowed size"
    if status == 429:
        return "Rate limit exceeded"
    if 500 <= status < 600:
        return f"Server error: {status}"
    return f"Unexpected status code: {status}"


@traces(spec="067", story="FR-009", reqs=["REQ-067-018"])
def _parse_body(response: httpx.Response) -> Any:
    """Parse JSON responses; fall back to raw text (FR-009)."""
    content_type = response.headers.get("content-type", "")
    if "json" in content_type.lower():
        try:
            return response.json()
        except ValueError:
            return response.text
    return response.text


@traces(spec="067", story="FR-009", reqs=["REQ-067-018"])
def _extract_fields(body: Any, fields: tuple[str, ...]) -> dict[str, Any]:
    """Pull dotted-path ``fields`` out of a JSON object ``body`` (FR-009)."""
    if not fields or not isinstance(body, dict):
        return {}
    result: dict[str, Any] = {}
    for path in fields:
        node: Any = body
        for part in path.split("."):
            if isinstance(node, dict) and part in node:
                node = node[part]
            else:
                node = None
                break
        result[path] = node
    return result


# ── lightweight JSON-schema subset validator (FR-005) ───────────────────

_SCHEMA_TYPE_MAP: dict[str, Any] = {
    "object": dict,
    "array": list,
    "string": str,
    "integer": int,
    "number": (int, float),
    "boolean": bool,
    "null": type(None),
}


@traces(spec="067", story="FR-005", reqs=["REQ-067-013"])
def _type_matches(instance: Any, expected_type: Any) -> bool:
    types = expected_type if isinstance(expected_type, list) else [expected_type]
    for t in types:
        py_type = _SCHEMA_TYPE_MAP.get(t)
        if py_type is None:
            return True  # unrecognized type keyword: don't fail on it
        if isinstance(instance, bool):
            if t == "boolean":
                return True
            continue  # bool is an int subclass; don't let it satisfy number/integer
        if isinstance(instance, py_type):
            return True
    return False


@traces(spec="067", story="FR-005", reqs=["REQ-067-013"])
def _validate_schema(
    instance: Any, schema: dict[str, Any], path: str = "$"
) -> list[str]:
    """Validate ``instance`` against a JSON-Schema *subset*: type/required/properties/items."""
    errors: list[str] = []
    expected_type = schema.get("type")
    if expected_type is not None and not _type_matches(instance, expected_type):
        errors.append(
            f"{path}: expected type {expected_type!r}, got {type(instance).__name__}"
        )
        return errors
    if isinstance(instance, dict):
        for required_key in schema.get("required", ()):
            if required_key not in instance:
                errors.append(f"{path}: missing required property {required_key!r}")
        for key, sub_schema in schema.get("properties", {}).items():
            if key in instance:
                errors.extend(
                    _validate_schema(instance[key], sub_schema, f"{path}.{key}")
                )
    elif isinstance(instance, list):
        items_schema = schema.get("items")
        if items_schema:
            errors.extend(
                itertools.chain.from_iterable(
                    _validate_schema(item, items_schema, f"{path}[{i}]")
                    for i, item in enumerate(instance)
                )
            )
    return errors


# ── SSRF guard ───────────────────────────────────────────────────────────


@traces(spec="067", reqs=["REQ-067-020"])
def _is_blocked_address(ip_str: str) -> bool:
    """True if ``ip_str`` is loopback/private/link-local/reserved (or unparsable)."""
    try:
        addr = ipaddress.ip_address(ip_str)
    except ValueError:
        return True
    return addr.is_loopback or addr.is_private or addr.is_link_local or addr.is_reserved


@traces(spec="067", reqs=["REQ-067-020"])
def _allow_private_from_env(env: Mapping[str, str] | None) -> bool:
    source = os.environ if env is None else env
    return source.get("REMOTE_TRIGGER_ALLOW_PRIVATE", "").strip().lower() in (
        "1",
        "true",
        "yes",
        "on",
    )


@traces(spec="067", story="US-067-005", reqs=["REQ-067-020"])
class SSRFGuard:
    """Blocks outbound requests to loopback/private/link-local/metadata targets.

    See the module docstring's "SSRF protection" fork note for MVP scope
    (single-resolution check, no DNS-rebinding protection/allowlist).
    """

    @traces(spec="067", reqs=["REQ-067-020"])
    def check(
        self, url: str, allow_internal: bool, env: Mapping[str, str] | None = None
    ) -> None:
        if allow_internal or _allow_private_from_env(env):
            return
        parsed = urllib.parse.urlsplit(url)
        hostname = (parsed.hostname or "").lower()
        if hostname in _BLOCKED_HOSTNAMES:
            raise RemoteTriggerError(
                RemoteTriggerErrorCode.SSRF_BLOCKED,
                f"Refused: {hostname} is a blocked hostname",
            )
        port = parsed.port or (443 if parsed.scheme == "https" else 80)
        try:
            infos = socket.getaddrinfo(hostname, port)
        except OSError as exc:
            raise RemoteTriggerError(
                RemoteTriggerErrorCode.DNS_FAILURE,
                f"Failed to resolve DNS for {hostname}",
            ) from exc
        for info in infos:
            ip_str = str(info[4][0])
            if ip_str in _METADATA_IPS or _is_blocked_address(ip_str):
                raise RemoteTriggerError(
                    RemoteTriggerErrorCode.SSRF_BLOCKED,
                    f"Refused: {hostname} resolves to a disallowed address ({ip_str})",
                )


# ── authentication manager ──────────────────────────────────────────────


@traces(
    spec="067",
    story="US-067-002",
    reqs=["REQ-067-002", "REQ-067-003", "REQ-067-004", "REQ-067-005", "REQ-067-006"],
)
class AuthManager:
    """Applies the configured :class:`AuthConfig` as request headers/params."""

    @traces(
        spec="067",
        story="FR-002",
        reqs=[
            "REQ-067-002",
            "REQ-067-003",
            "REQ-067-004",
            "REQ-067-005",
            "REQ-067-006",
        ],
    )
    async def apply(
        self, request: TriggerRequest, client: httpx.AsyncClient, body_bytes: bytes
    ) -> tuple[dict[str, str], dict[str, str]]:
        """Return ``(extra_headers, extra_params)`` for ``request``'s auth."""
        auth = request.auth
        if auth is None:
            return {}, {}
        env = request.env
        headers: dict[str, str] = {}
        params: dict[str, str] = {}
        if auth.kind == "api_key":
            value = _resolve_secret(auth.value, auth.value_env, env, "api_key value")
            if auth.location == "query":
                params[auth.key] = value
            else:
                headers[auth.key] = value
        elif auth.kind == "bearer":
            token = _resolve_secret(auth.token, auth.token_env, env, "bearer token")
            headers["Authorization"] = f"Bearer {token}"
        elif auth.kind == "basic":
            password = (
                _resolve_secret(
                    auth.password, auth.password_env, env, "basic auth password"
                )
                if (auth.password or auth.password_env)
                else ""
            )
            raw = f"{auth.username}:{password}".encode()
            headers["Authorization"] = f"Basic {base64.b64encode(raw).decode()}"
        elif auth.kind == "oauth2":
            token = await self._fetch_oauth2_token(auth, client, env)
            headers["Authorization"] = f"Bearer {token}"
        elif auth.kind == "hmac":
            secret = _resolve_secret(auth.secret, auth.secret_env, env, "hmac secret")
            headers[auth.header_name] = _hmac_digest(secret, body_bytes, auth.algorithm)
        return headers, params

    @traces(spec="067", story="FR-002", reqs=["REQ-067-005"])
    async def _fetch_oauth2_token(
        self, auth: AuthConfig, client: httpx.AsyncClient, env: Mapping[str, str] | None
    ) -> str:
        secret = (
            _resolve_secret(
                auth.client_secret, auth.client_secret_env, env, "oauth2 client secret"
            )
            if (auth.client_secret or auth.client_secret_env)
            else ""
        )
        try:
            resp = await client.post(
                auth.token_url,
                data={
                    "grant_type": auth.grant_type or "client_credentials",
                    "client_id": auth.client_id,
                    "client_secret": secret,
                    "scope": " ".join(auth.scopes),
                },
            )
        except httpx.HTTPError as exc:
            raise RemoteTriggerError(
                RemoteTriggerErrorCode.AUTH_FAILED,
                f"OAuth2 token request failed: {exc}",
            ) from exc
        if resp.status_code >= 400:
            raise RemoteTriggerError(
                RemoteTriggerErrorCode.AUTH_FAILED,
                f"OAuth2 token endpoint returned {resp.status_code}",
            )
        try:
            token = resp.json()["access_token"]
        except (ValueError, KeyError) as exc:
            raise RemoteTriggerError(
                RemoteTriggerErrorCode.AUTH_FAILED,
                "OAuth2 token response missing 'access_token'",
            ) from exc
        return str(token)


# ── retry executor ───────────────────────────────────────────────────────


@traces(spec="067", story="FR-004", reqs=["REQ-067-010"])
def _parse_retry_after(value: str | None) -> float | None:
    if value is None:
        return None
    try:
        return float(value)
    except ValueError:
        return None  # HTTP-date form not supported (MVP scope)


@traces(spec="067", story="FR-004", reqs=["REQ-067-008", "REQ-067-010"])
def _compute_delay(
    attempt: int, retry: RetryConfig, retry_after: float | None
) -> float:
    if retry_after is not None:
        return max(0.0, min(retry_after, retry.max_delay))
    delay = min(retry.initial_delay * (retry.backoff_factor**attempt), retry.max_delay)
    if retry.jitter:
        delay *= 0.5 + random.random() * 0.5
    return delay


@traces(
    spec="067", story="US-067-003", reqs=["REQ-067-008", "REQ-067-009", "REQ-067-010"]
)
class RetryExecutor:
    """Bounded exponential-backoff retry over an async request callable."""

    @traces(
        spec="067", story="FR-004", reqs=["REQ-067-008", "REQ-067-009", "REQ-067-010"]
    )
    async def run(
        self,
        do_request: Callable[[], Awaitable[httpx.Response]],
        retry: RetryConfig,
        sleep: Callable[[float], Awaitable[None]],
    ) -> tuple[httpx.Response | None, int]:
        """Returns ``(response_or_None, retry_count)``.

        ``response`` is ``None`` only when every attempt timed out (the
        caller renders the ERR-007 timeout message).
        """
        attempt = 0
        while True:
            try:
                response = await do_request()
            except httpx.TimeoutException:
                if (
                    retry.enabled
                    and retry.retry_on_timeout
                    and attempt < retry.max_attempts - 1
                ):
                    await sleep(_compute_delay(attempt, retry, None))
                    attempt += 1
                    continue
                return None, attempt
            if (
                retry.enabled
                and response.status_code in retry.retry_on_status
                and attempt < retry.max_attempts - 1
            ):
                retry_after = _parse_retry_after(response.headers.get("Retry-After"))
                await sleep(_compute_delay(attempt, retry, retry_after))
                attempt += 1
                continue
            return response, attempt


# ── response validator ──────────────────────────────────────────────────


@traces(
    spec="067",
    story="US-067-004",
    reqs=["REQ-067-011", "REQ-067-012", "REQ-067-013", "REQ-067-014"],
)
class ResponseValidator:
    """Validates a response against a :class:`ValidationConfig` (FR-005)."""

    @traces(
        spec="067",
        story="FR-005",
        reqs=["REQ-067-011", "REQ-067-012", "REQ-067-013", "REQ-067-014"],
    )
    def validate(
        self,
        response: httpx.Response,
        elapsed_ms: int,
        cfg: ValidationConfig,
        env: Mapping[str, str] | None,
    ) -> tuple[bool, list[str]]:
        errors: list[str] = []
        if cfg.expected_status and response.status_code not in cfg.expected_status:
            errors.append(
                f"Unexpected status code {response.status_code}; "
                f"expected one of {list(cfg.expected_status)}"
            )
        # No implicit "must be 2xx" check here when expected_status is unset:
        # RemoteTriggerTool._run already derives success from
        # response.status_code < 400 and, on failure, renders the specific
        # ERR-00N wording (401/403/404/429/5xx/retry-exhausted) via
        # _status_error_message — duplicating a generic "non-success status"
        # message here would shadow that more useful text.

        elapsed_s = elapsed_ms / 1000.0
        if cfg.min_response_time is not None and elapsed_s < cfg.min_response_time:
            errors.append(
                f"Response time {elapsed_s:.3f}s below minimum {cfg.min_response_time}s"
            )
        if cfg.max_response_time is not None and elapsed_s > cfg.max_response_time:
            errors.append(
                f"Response time {elapsed_s:.3f}s exceeds maximum {cfg.max_response_time}s"
            )

        if cfg.schema is not None:
            errors.extend(_validate_schema(_parse_body(response), cfg.schema))

        if cfg.signature_secret or cfg.signature_secret_env:
            try:
                secret = _resolve_secret(
                    cfg.signature_secret,
                    cfg.signature_secret_env,
                    env,
                    "validation signature secret",
                )
            except RemoteTriggerError as exc:
                errors.append(str(exc))
                secret = ""
            provided = response.headers.get(cfg.signature_header, "")
            if (
                not secret
                or not provided
                or not _verify_hmac(
                    response.content, provided, secret, cfg.signature_algorithm
                )
            ):
                errors.append("Webhook signature verification failed")

        return (len(errors) == 0, errors)


# ── the tool itself ──────────────────────────────────────────────────────

_PARAMETERS: dict[str, Any] = {
    "type": "object",
    "properties": {
        "url": {
            "type": "string",
            "description": "Target webhook or API endpoint URL.",
        },
        "method": {
            "type": "string",
            "enum": sorted(_ALLOWED_METHODS),
            "default": "POST",
        },
        "headers": {"type": "object", "additionalProperties": {"type": "string"}},
        "body": {"description": "Request body (JSON object/array, or a raw string)."},
        "params": {
            "type": "object",
            "additionalProperties": {"type": "string"},
            "description": "Query parameters (also where 'api_key' auth 'in':'query' lands).",
        },
        "auth": {
            "type": "object",
            "description": (
                "Authentication config: "
                "{type: api_key|bearer|basic|oauth2|hmac, ...type-specific fields}."
            ),
        },
        "timeout": {"type": "integer", "minimum": 1, "maximum": 300, "default": 30},
        "retry": {
            "type": "object",
            "description": (
                "enabled/max_attempts/initial_delay/max_delay/backoff_factor/"
                "retry_on_status/retry_on_timeout/jitter."
            ),
        },
        "validation": {
            "type": "object",
            "description": (
                "expected_status/min_response_time/max_response_time/schema/"
                "signature_secret/signature_secret_env/signature_header/"
                "signature_algorithm."
            ),
        },
        "verify_ssl": {"type": "boolean", "default": True},
        "ca_bundle_path": {"type": "string"},
        "connection_pool_size": {
            "type": "integer",
            "minimum": 1,
            "maximum": 100,
            "default": 10,
        },
        "allow_internal": {
            "type": "boolean",
            "default": False,
            "description": "Bypass the SSRF guard for a trusted internal/loopback target.",
        },
        "extract_fields": {
            "type": "array",
            "items": {"type": "string"},
            "description": "Dotted paths to extract from a JSON response body.",
        },
    },
    "required": ["url"],
}


def _build_metadata() -> ToolMetadata:
    return ToolMetadata(
        name="remote_trigger",
        description=(
            "Trigger remote webhooks or API calls with authentication, "
            "retry/backoff, timeout handling, and response validation."
        ),
        category=ToolCategory.WEB,
        parameters=_PARAMETERS,
        priority=ToolPriority.P1,
        requires_approval=True,
        governance_article="tool:execute",
        tags=("webhook", "http", "api", "trigger"),
    )


@traces(
    spec="067",
    story="US-067-001",
    reqs=[
        "REQ-067-001",
        "REQ-067-007",
        "REQ-067-015",
        "REQ-067-016",
        "REQ-067-017",
        "REQ-067-018",
        "REQ-067-019",
        "REQ-067-021",
        "REQ-067-022",
    ],
)
class RemoteTriggerTool(BaseTool):
    """The registrable ``BaseTool`` (Command) callers/LLMs invoke.

    Composes :class:`SSRFGuard`, :class:`AuthManager`, :class:`RetryExecutor`,
    and :class:`ResponseValidator` — one focused unit per FR area.
    """

    def __init__(self, metadata: ToolMetadata | None = None) -> None:
        self._metadata = metadata or _build_metadata()
        self._ssrf = SSRFGuard()
        self._auth = AuthManager()
        self._retry = RetryExecutor()
        self._validator = ResponseValidator()

    @property
    def metadata(self) -> ToolMetadata:
        return self._metadata

    @traces(spec="067", story="US-067-001", reqs=["REQ-067-001"])
    async def execute(self, arguments: dict[str, Any]) -> ToolResult:
        """Run a trigger from raw tool-call ``arguments`` (FR-001). Never raises."""
        try:
            request = TriggerRequest.from_arguments(arguments)
        except RemoteTriggerError as exc:
            return exc.to_result(self.metadata.name)
        return await self._execute_request(request)

    @traces(spec="067", story="US-067-001", reqs=["REQ-067-021"])
    async def execute_simple(
        self,
        url: str,
        method: str = "POST",
        headers: dict[str, str] | None = None,
        body: Any = None,
        auth: dict[str, Any] | None = None,
        timeout: int = 30,
        env: Mapping[str, str] | None = None,
    ) -> ToolResult:
        """Simplified interface for basic triggers (spec.md API Contract)."""
        arguments: dict[str, Any] = {
            "url": url,
            "method": method,
            "headers": headers or {},
            "body": body,
            "timeout": timeout,
        }
        if auth is not None:
            arguments["auth"] = auth
        try:
            request = TriggerRequest.from_arguments(arguments, env=env)
        except RemoteTriggerError as exc:
            return exc.to_result(self.metadata.name)
        return await self._execute_request(request)

    @traces(spec="067", story="FR-001", reqs=["REQ-067-001"])
    async def _execute_request(self, request: TriggerRequest) -> ToolResult:
        try:
            response = await self._run(request)
        except RemoteTriggerError as exc:
            return exc.to_result(self.metadata.name)
        error = None
        if not response.success:
            error = (
                response.error
                or "; ".join(response.validation_errors)
                or "remote trigger failed"
            )
        return ToolResult(
            tool_name=self.metadata.name,
            success=response.success,
            output=response.to_json(),
            error=error,
        )

    @traces(spec="067", story="US-067-004", reqs=["REQ-067-022"])
    async def verify_signature(
        self, payload: bytes, signature: str, secret: str, algorithm: str = "sha256"
    ) -> bool:
        """Verify a webhook HMAC signature (spec.md API Contract, FR-011)."""
        return _verify_hmac(payload, signature, secret, algorithm)

    @traces(spec="067", story="US-067-004", reqs=["REQ-067-018", "REQ-067-011"])
    async def parse_response(
        self, response: httpx.Response, validation: ValidationConfig | None = None
    ) -> dict[str, Any]:
        """Parse (and optionally validate) an ``httpx.Response`` (FR-009)."""
        body = _parse_body(response)
        result: dict[str, Any] = {"status_code": response.status_code, "body": body}
        if validation is not None:
            passed, errors = self._validator.validate(response, 0, validation, None)
            result["validation_passed"] = passed
            result["validation_errors"] = errors
        return result

    @traces(
        spec="067",
        story="US-067-001",
        reqs=[
            "REQ-067-001",
            "REQ-067-015",
            "REQ-067-016",
            "REQ-067-017",
            "REQ-067-020",
        ],
    )
    async def _run(self, request: TriggerRequest) -> TriggerResponse:
        """Execute ``request`` end-to-end: SSRF guard -> auth -> retry -> validate."""
        self._ssrf.check(request.url, request.allow_internal, request.env)

        body_bytes = _encode_body(request.body)
        limits = httpx.Limits(
            max_connections=request.connection_pool_size,
            max_keepalive_connections=request.connection_pool_size,
        )
        verify: bool | str = request.ca_bundle_path or request.verify_ssl
        if not request.verify_ssl:
            logger.warning(
                "remote_trigger: SSL verification disabled for %s",
                _sanitize_url(request.url),
            )
        timeout = httpx.Timeout(float(request.timeout))

        start = time.monotonic()
        async with httpx.AsyncClient(
            limits=limits, verify=verify, timeout=timeout
        ) as client:
            extra_headers, extra_params = await self._auth.apply(
                request, client, body_bytes
            )
            merged_headers = {**request.headers, **extra_headers}
            merged_params = {**request.params, **extra_params}
            logger.debug(
                "remote_trigger: %s %s headers=%s",
                request.method,
                _sanitize_url(request.url),
                _sanitize_headers(merged_headers),
            )

            async def do_request() -> httpx.Response:
                kwargs: dict[str, Any] = {
                    "headers": merged_headers,
                    "params": merged_params,
                }
                if isinstance(request.body, (dict, list)):
                    kwargs["json"] = request.body
                elif isinstance(request.body, str):
                    kwargs["content"] = request.body.encode()
                elif isinstance(request.body, bool | int | float):
                    # spec.md TriggerRequest.body allows a bare scalar; JSON-encode
                    # it so e.g. body=true / body=3 still reaches the server.
                    kwargs["content"] = json.dumps(request.body).encode()
                    kwargs["headers"] = {
                        **merged_headers,
                        "Content-Type": "application/json",
                    }
                return await client.request(request.method, request.url, **kwargs)

            try:
                import asyncio  # noqa: PLC0415 - only needed for the sleep seam

                response, retry_count = await self._retry.run(
                    do_request, request.retry, asyncio.sleep
                )
            except httpx.ConnectError as exc:
                message = str(exc)
                code = (
                    RemoteTriggerErrorCode.SSL_ERROR
                    if "certificate" in message.lower()
                    else RemoteTriggerErrorCode.CONNECTION_ERROR
                )
                logger.warning(
                    "remote_trigger: %s %s failed: %s",
                    request.method,
                    _sanitize_url(request.url),
                    message,
                )
                raise RemoteTriggerError(
                    code,
                    f"Failed to connect to {urllib.parse.urlsplit(request.url).hostname}: {message}",
                ) from exc

        elapsed_ms = int((time.monotonic() - start) * 1000)

        if response is None:
            logger.warning(
                "remote_trigger: timeout after %ss for %s",
                request.timeout,
                _sanitize_url(request.url),
            )
            return TriggerResponse(
                success=False,
                status_code=0,
                timestamp=_now_iso(),
                elapsed_ms=elapsed_ms,
                retry_count=retry_count,
                error=f"Request timeout after {request.timeout} seconds",
            )

        resp_body = _parse_body(response)
        extracted = _extract_fields(resp_body, request.extract_fields)
        passed, errors = self._validator.validate(
            response, elapsed_ms, request.validation, request.env
        )

        error_message = None
        if not passed:
            error_message = "; ".join(errors)
        elif response.status_code >= 400:
            error_message = _status_error_message(
                response.status_code, request.url, retry_count, request.retry
            )

        logger.info(
            "remote_trigger: %s %s -> %s (%sms, retries=%s)",
            request.method,
            _sanitize_url(request.url),
            response.status_code,
            elapsed_ms,
            retry_count,
        )

        return TriggerResponse(
            success=passed and response.status_code < 400,
            status_code=response.status_code,
            status_text=response.reason_phrase or "",
            timestamp=_now_iso(),
            elapsed_ms=elapsed_ms,
            headers=dict(response.headers),
            body=resp_body,
            extracted=extracted,
            validation_passed=passed,
            validation_errors=errors,
            retry_count=retry_count,
            error=error_message,
        )


@traces(spec="067", story="US-067-001", reqs=["REQ-067-001"])
def register_remote_trigger_tool(registry: ToolRegistry) -> RemoteTriggerTool:
    """Build a :class:`RemoteTriggerTool` and register it into ``registry``.

    Not wired into ``registry.py``'s default registration set per the build
    brief (new-files-only) — a caller (e.g. the volk tool-bootstrap path)
    invokes this explicitly to add ``remote_trigger`` to a live registry.
    """
    tool = RemoteTriggerTool()
    registry.register(tool, replace=True)
    return tool