diff --git a/src/agent_env/env/gateway/constants.py b/src/agent_env/env/gateway/constants.py
index 4acb7b7e..a29c5e9e 100644
--- a/src/agent_env/env/gateway/constants.py
+++ b/src/agent_env/env/gateway/constants.py
@@ -119,8 +119,9 @@ def sources(names: list[str]) -> str:
 TRIGGER_ACTION_DELAY = "action_delay"
 TRIGGER_EXECUTOR_HOURS = "executor_hours"
 TRIGGER_UNLESS = "unless"
+TRIGGER_WEBSITE = "website"
 TRIGGER_FEATURES = (TRIGGER_NAMED_EXECUTORS, TRIGGER_PROVOKING_ROLES, TRIGGER_ACTION_DELAY, TRIGGER_EXECUTOR_HOURS,
-                    TRIGGER_UNLESS)
+                    TRIGGER_UNLESS, TRIGGER_WEBSITE)
 
 GATEWAY_TRAJECTORY_FILE = os.environ.get(
     "GATEWAY_TRAJECTORY_FILE",
diff --git a/src/agent_env/env/gateway/gateway.py b/src/agent_env/env/gateway/gateway.py
index 9c732539..a0d7af36 100644
--- a/src/agent_env/env/gateway/gateway.py
+++ b/src/agent_env/env/gateway/gateway.py
@@ -13,6 +13,7 @@
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
+from urllib.parse import parse_qsl
 from uuid import uuid4
 
 import anyio
@@ -69,7 +70,7 @@
 from .child_transport import child_http_client
 from .clock import Clock, ClockError, _iso
 from .get_time import GET_TIME_TOOL_NAME, build_get_time_tool
-from .rest_proxy import ROLE_HEADER_BYTES, fetch_child_card, forward, rewrite_child_card
+from .rest_proxy import ROLE_HEADER_BYTES, UpstreamResponse, fetch_child_card, forward, rewrite_child_card
 from .triggers import TriggerEngine, TriggerError
 
 logger = logging.getLogger(__name__)
@@ -93,6 +94,7 @@ class Gateway:
     TOOL_CALL_TIMEOUT_S = 600.0
     STARTUP_DISCOVERY_TIMEOUT_S = 60.0  # startup discovery past it fails open, like a failed one; uvicorn must come up
     CHILD_CLOSE_TIMEOUT_S = 3.0  # wait for the owner task to exit a retired child-session generation; past it, cancel it
+    BARRIER_READ_AHEAD_MAX_BYTES = 8 * 1024 * 1024  # a larger sized body held by a barrier streams, its window to the last byte
     TRIGGER_BARRIER_TIMEOUT_S = 30.0  # a barrier may ask for less; past it the call is answered anyway
     # Must be >= the largest timeout any client can ask for, or the proxy severs a load the
     # client was still legitimately waiting on -- the ordering this module's constants file
@@ -186,14 +188,14 @@ async def proxy_to_service(request: Request) -> Response:
                 methods=["GET", "POST", "PUT", "DELETE", "PATCH"],
             )
             async def proxy_to_frontend(request: Request) -> Response:
-                return await self._proxy_rest_request(request, url_map=self.website_urls)
+                return await self._proxy_website_request(request)
 
             @self._mcp.custom_route(
                 "/website/{service_name}",
                 methods=["GET", "POST", "PUT", "DELETE", "PATCH"],
             )
             async def proxy_to_frontend_root(request: Request) -> Response:
-                return await self._proxy_rest_request(request, url_map=self.website_urls)
+                return await self._proxy_website_request(request)
 
         @self._mcp.custom_route(f"/tools/{TOOL_DISABLE_ACTION}", methods=["POST"])
         async def tools_disable(request: Request) -> Response:
@@ -546,6 +548,34 @@ async def log_write(status: int) -> None:
         return await forward(request, base_url, role=role, timeout=self.REST_PROXY_TIMEOUT, mcp_server=mcp_server,
                              on_done=log_write)
 
+    async def _proxy_website_request(self, request: Request) -> Response:
+        """Proxy a website request; with a website or state trigger registered, feed it to the engine and hold barriers."""
+        response = await self._proxy_rest_request(request, url_map=self.website_urls)
+        app = request.path_params["service_name"]
+        engine = self._trigger_engine
+        if app in self.website_urls and (engine.has_website_triggers or engine.has_state_triggers):
+            if isinstance(response, UpstreamResponse):
+                if not engine.may_hold_website_request(self._request_role(request), app, request.method,
+                                                       "/" + request.path_params.get("path", ""), response.status_code):
+                    response.after_body(lambda status: self._fire_website_triggers(request, app, status))  # answered at its last byte
+                    return response
+                await response.read_ahead_if_sized(self.BARRIER_READ_AHEAD_MAX_BYTES)  # its window ends before its triggers run
+            barrier_tasks, barrier_timeout_s = await self._fire_website_triggers(request, app, response.status_code)
+            await self._await_barriers(barrier_tasks, barrier_timeout_s)
+        return response
+
+    async def _fire_website_triggers(self, request: Request, app: str,
+                                     status: int) -> tuple[list[asyncio.Task], float | None]:
+        """Feed an answered website request to the trigger engine (fail-open)."""
+        try:
+            # Only a website trigger reads the request's args; a state re-check needs just the method and status.
+            args = _request_args(request, await request.body()) if self._trigger_engine.has_website_triggers else {}
+            return self._trigger_engine.on_website_request(self._request_role(request), app, request.method,
+                                                           "/" + request.path_params.get("path", ""), status, args)
+        except Exception:
+            logger.exception("Trigger evaluation failed (fail-open)")
+            return [], None
+
     @staticmethod
     def _request_role(request: Request) -> str:
         """The role the gateway attributes a REST request to: the first AgentEnv-Role header, else the default role."""
@@ -1289,6 +1319,29 @@ def model_dump_one_level(self) -> dict[str, Any]:
         return kwargs
 
 
+def _request_args(request: Request, body: bytes) -> dict:
+    """A request's query parameters and form or JSON-object body fields (the body wins a clash; repeats become lists)."""
+    args = _fields(parse_qsl(request.url.query, keep_blank_values=True))
+    media_type = request.headers.get("content-type", "").partition(";")[0].strip().lower()
+    if media_type == "application/x-www-form-urlencoded":
+        args.update(_fields(parse_qsl(body.decode("utf-8", "replace"), keep_blank_values=True)))
+    elif media_type == "application/json" or media_type.endswith("+json"):
+        try:
+            parsed = json.loads(body)
+        except ValueError:
+            parsed = None
+        if isinstance(parsed, dict):
+            args.update(parsed)
+    return args
+
+
+def _fields(pairs: list[tuple[str, str]]) -> dict:
+    grouped: dict[str, list[str]] = {}
+    for key, value in pairs:
+        grouped.setdefault(key, []).append(value)
+    return {key: values[0] if len(values) == 1 else values for key, values in grouped.items()}
+
+
 def _is_dead_session(exc: BaseException) -> bool:
     """Whether a child call failed because its session is gone, so the generation is retired: the stream closed under
     it, the child answered 404 for the session id (idle expiry or a restart, which the client reports as code 32600),
diff --git a/src/agent_env/env/gateway/rest_proxy.py b/src/agent_env/env/gateway/rest_proxy.py
index 0114ae04..0812440e 100644
--- a/src/agent_env/env/gateway/rest_proxy.py
+++ b/src/agent_env/env/gateway/rest_proxy.py
@@ -24,6 +24,7 @@
 import logging
 import posixpath
 from collections.abc import AsyncIterator, Awaitable, Callable
+from typing import Any
 from urllib.parse import unquote
 
 import httpx
@@ -190,6 +191,36 @@ async def _body(upstream: httpx.Response, target_url: str) -> AsyncIterator[byte
             logger.error(f"Proxy to {target_url} failed after {sent} bytes: {e!r}")
             raise
 
+    def after_body(self, then: Callable[[int], Awaitable[Any]]) -> None:
+        """Await `then` with the status once the response has ended, after `on_done`."""
+        on_done = self._on_done
+
+        async def both(status: int) -> None:
+            if on_done is not None:
+                await on_done(status)
+            await then(status)
+        self._on_done = both
+
+    async def read_ahead_if_sized(self, max_bytes: int) -> None:
+        """Read a body sized at most `max_bytes` to its end now and run `on_done` there, so the response then replays it."""
+        size = self._upstream.headers.get("content-length", "")
+        if self._on_done is None or not size.isdigit() or int(size) > max_bytes:
+            return
+        chunks: list[bytes] = []
+        failure: Exception | None = None
+        try:
+            async for chunk in self.body_iterator:
+                chunks.append(chunk)
+        except Exception as e:  # already logged by `_body`; the client still gets the cut body when it is sent
+            failure = e
+        except BaseException:  # cancelled before `__call__`, which would otherwise close them, ever runs
+            await self._upstream.aclose()
+            await self._client.aclose()
+            raise
+        self.body_iterator = _replay(chunks, failure)
+        on_done, self._on_done = self._on_done, None
+        await on_done(self.status_code)
+
     async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
         try:
             await super().__call__(scope, receive, send)
@@ -200,6 +231,13 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
                 await self._on_done(self.status_code)
 
 
+async def _replay(chunks: list[bytes], failure: Exception | None) -> AsyncIterator[bytes]:
+    for chunk in chunks:
+        yield chunk
+    if failure is not None:
+        raise failure
+
+
 async def fetch_child_card(client: httpx.AsyncClient, key: str, base_url: str) -> dict | None:
     """One backing server's card as it serves it; None (omit) on 404 / error / timeout / malformed JSON."""
     url = f"{base_url.rstrip('/')}{WELL_KNOWN_PATH}"
diff --git a/src/agent_env/env/gateway/triggers.py b/src/agent_env/env/gateway/triggers.py
index 41da5e1d..43943a13 100644
--- a/src/agent_env/env/gateway/triggers.py
+++ b/src/agent_env/env/gateway/triggers.py
@@ -26,6 +26,11 @@
 logger = logging.getLogger(__name__)
 
 _PREDICATE_KEYS = ("regex", "equals", "exists")
+_WHEN_TYPES = ("action", "state", "time", "website")
+# The provoking-call roots each when.type's templates and `where` read; a website response streams through unread.
+_CTX_ROOTS = {"action": ("args", "result"), "website": ("args",)}
+# RFC 9110's safe methods: a request using one asks for no change, so it prompts no state re-evaluation.
+_SAFE_METHODS = frozenset({"GET", "HEAD", "OPTIONS", "TRACE"})
 _ACTION_TYPES = ("nl", "tool", "permission")
 _TERMINAL_A2A_STATES = ("completed", "failed", "canceled", "rejected")
 # Bounds the per-evaluation catch-up burst; the backlog drains on later evaluations.
@@ -34,7 +39,8 @@
 _EVENTS_HEAD = 1000
 _EVENTS_TAIL = 9000
 # Schedule keys state() echoes; excludes the unbounded `check`/`where`.
-_WHEN_SUMMARY_KEYS = ("type", "tool", "roles", "repeat", "at", "after", "offset", "every", "count", "until", "seed")
+_WHEN_SUMMARY_KEYS = ("type", "tool", "app", "method", "path", "status", "roles", "repeat", "at", "after", "offset",
+                      "every", "count", "until", "seed")
 # Placeholder roots reading the provoking call; a bare name is a check-pipeline binding.
 _CTX_PREFIXES = ("args", "result")
 # Where a barrier waits; a further wait point is a new value here, not a new top-level field.
@@ -322,9 +328,9 @@ def _validate_path(path: str, where: str) -> None:
             raise TriggerError(f"{where}: invalid path segment {segment!r} in {path!r}")
 
 
-def _validate_placeholders(value: Any, where: str, allow_ctx: bool) -> None:
-    """Fail-loud at registration: `${args.*}` / `${result.*}` only on action triggers (they alone carry a
-    provoking call), any other placeholder must be a bare check-binding name. A shape the templater
+def _validate_placeholders(value: Any, where: str, roots: tuple[str, ...]) -> None:
+    """Fail-loud at registration: `${args.*}` / `${result.*}` only on triggers that carry a provoking call
+    (`roots`), any other placeholder must be a bare check-binding name. A shape the templater
     cannot parse is rejected first, or it would reach the tool as literal text."""
     inert = _inert_placeholders(value)
     if inert:
@@ -336,8 +342,11 @@ def _validate_placeholders(value: Any, where: str, allow_ctx: bool) -> None:
         if root_name is None:
             if re.fullmatch(r"\w+", name) is None:
                 raise TriggerError(f"{where}: placeholder ${{{name}}} must be a check binding name or an args./result. path")
-        elif not allow_ctx:
-            raise TriggerError(f"{where}: ${{{name}}} needs a provoking tool call; only action triggers carry args/result")
+        elif not roots:
+            raise TriggerError(f"{where}: ${{{name}}} needs a provoking tool call; only action triggers carry args/result "
+                               "(website triggers carry args)")
+        elif root_name not in roots:
+            raise TriggerError(f"{where}: ${{{name}}} reads a result; a website request carries args, not a result")
         elif path:
             _validate_path(path, where)
         elif "." in name:
@@ -630,8 +639,8 @@ def _normalize_barrier(spec: dict, tid: str, has_ctx: bool) -> None:
         if at not in _BARRIER_AT:
             raise TriggerError(f"{w}.at must be one of {list(_BARRIER_AT)} (got {at!r})")
         if at == "provoking_call" and not has_ctx:
-            raise TriggerError(f"{w}.at 'provoking_call' is only valid on action triggers — they alone have a "
-                               "provoking call to hold")
+            raise TriggerError(f"{w}.at 'provoking_call' is only valid on action triggers and website triggers — they "
+                               "alone have a provoking call to hold")
         timeout = barrier.get("timeout_seconds")
         if timeout is not None and (isinstance(timeout, bool) or not isinstance(timeout, (int, float))
                                     or not math.isfinite(timeout) or timeout <= 0):
@@ -641,32 +650,35 @@ def _normalize_trigger(self, spec: dict, executors: dict[str | None, dict]) -> d
         """Validate + normalize a trigger spec in place (fail-loud); returns it, so pass a copy."""
         tid = spec["id"]
         when = spec.get("when")
-        if not isinstance(when, dict) or when.get("type") not in ("action", "state", "time"):
-            raise TriggerError(f"trigger '{tid}': when.type must be 'action', 'state', or 'time'")
+        if not isinstance(when, dict) or when.get("type") not in _WHEN_TYPES:
+            raise TriggerError(f"trigger '{tid}': when.type must be 'action', 'state', 'time', or 'website'")
         self._normalize_roles(when, tid)
         if when.get("unless") is None:
             when.pop("unless", None)
         if "unless" in when and when["type"] != "time":
             raise TriggerError(f"trigger '{tid}': when.unless is only valid on time triggers")
+        roots = _CTX_ROOTS.get(when["type"], ())
         if when["type"] == "action":
             self._normalize_action_when(when, tid)
+        elif when["type"] == "website":
+            self._normalize_website_when(when, tid)
         elif when["type"] == "state":
             if "repeat" in when:
-                raise TriggerError(f"trigger '{tid}': when.repeat is only valid on action triggers")
+                raise TriggerError(f"trigger '{tid}': when.repeat is only valid on action triggers and website triggers")
             when["check"] = _validate_check(when.get("check"), f"trigger '{tid}': when.check")
             for j, step in enumerate(when["check"]["steps"]):
-                _validate_placeholders(step.get("args", {}), f"trigger '{tid}': when.check.steps[{j}].args", allow_ctx=False)
+                _validate_placeholders(step.get("args", {}), f"trigger '{tid}': when.check.steps[{j}].args", roots)
         else:
             if "repeat" in when:
-                raise TriggerError(f"trigger '{tid}': when.repeat is only valid on action triggers (time triggers recur via 'every')")
+                raise TriggerError(f"trigger '{tid}': when.repeat is only valid on action triggers and website triggers "
+                                   "(time triggers recur via 'every')")
             self._normalize_time_when(when, tid)
-        has_ctx = when["type"] == "action"
-        self._normalize_barrier(spec, tid, has_ctx)
+        self._normalize_barrier(spec, tid, bool(roots))
         actions = spec.get("actions")
         if not isinstance(actions, list):
             raise TriggerError(f"trigger '{tid}': actions must be a list")
         for i, action in enumerate(actions):
-            self._normalize_action(action, f"trigger '{tid}': actions[{i}]", has_ctx, executors)
+            self._normalize_action(action, f"trigger '{tid}': actions[{i}]", roots, executors)
             target = executors.get(action.get("executor"))
             if action["type"] == "nl" and target is not None and target["role"] in when.get("roles", ()):
                 raise TriggerError(f"trigger '{tid}': when.roles includes {target['role']!r}, the role of the executor "
@@ -683,7 +695,7 @@ def _normalize_roles(when: dict, tid: str) -> None:
             when.pop("roles", None)
             return
         if when["type"] == "time":
-            raise TriggerError(f"trigger '{tid}': when.roles is only valid on action and state triggers")
+            raise TriggerError(f"trigger '{tid}': when.roles is only valid on action and state triggers, and website triggers")
         roles = when["roles"]
         if not isinstance(roles, list) or not roles or not all(isinstance(r, str) and r for r in roles):
             raise TriggerError(f"trigger '{tid}': when.roles must be a non-empty list of non-empty AgentEnv-Roles")
@@ -693,19 +705,31 @@ def _normalize_action_when(when: dict, tid: str) -> None:
         """Validate a when.type=='action' spec: the provoking tool, its `where` predicates, `repeat`."""
         if not isinstance(when.get("tool"), str) or not when["tool"]:
             raise TriggerError(f"trigger '{tid}': when.tool must be a non-empty string")
-        where = when.get("where", {})
-        if not isinstance(where, dict):
-            raise TriggerError(f"trigger '{tid}': when.where must be an object")
-        for field, pred in where.items():
-            _validate_predicate(pred, f"trigger '{tid}': where.{field}")
-            _, path = _ctx_path(field)
-            if path:
-                _validate_path(path, f"trigger '{tid}': where.{field}")
-        if "repeat" in when and not isinstance(when["repeat"], bool):
-            raise TriggerError(f"trigger '{tid}': when.repeat must be a boolean")
+        _normalize_provoked_when(when, tid)
+
+    def _normalize_website_when(self, when: dict, tid: str) -> None:
+        """Validate a website `when`: a known app, upper-cased method, `path`/`status` predicates, `where`, `repeat`."""
+        w = f"trigger '{tid}'"
+        app = when.get("app")
+        if not isinstance(app, str) or not app:
+            raise TriggerError(f"{w}: when.app must be a non-empty string naming one of the env's websites")
+        if app not in self._gw.website_urls:
+            raise TriggerError(f"{w}: when.app names unknown website {app!r} (websites: {sorted(self._gw.website_urls)})")
+        if when.get("method") is None:
+            when.pop("method", None)
+        elif not isinstance(when["method"], str) or not when["method"].strip():
+            raise TriggerError(f"{w}: when.method must be a non-empty HTTP method, e.g. 'POST'")
+        else:
+            when["method"] = when["method"].strip().upper()
+        for key in ("path", "status"):
+            if when.get(key) is None:
+                when.pop(key, None)
+            else:
+                _validate_predicate(when[key], f"{w}: when.{key}")
+        _normalize_provoked_when(when, tid)
 
     @staticmethod
-    def _normalize_action(action: Any, w: str, has_ctx: bool, executors: dict[str | None, dict]) -> None:
+    def _normalize_action(action: Any, w: str, roots: tuple[str, ...], executors: dict[str | None, dict]) -> None:
         """Validate one fire-action in place, plus its optional `verify` check, `as` role and nl `executor` name."""
         atype = action.get("type") if isinstance(action, dict) else None
         if atype not in _ACTION_TYPES:
@@ -738,7 +762,7 @@ def _normalize_action(action: Any, w: str, has_ctx: bool, executors: dict[str |
                 raise TriggerError(f"{w}: tool must be a non-empty string")
             if not isinstance(action.get("args", {}), dict):
                 raise TriggerError(f"{w}: args must be an object")
-            _validate_placeholders(action.get("args", {}), f"{w}.args", allow_ctx=has_ctx)
+            _validate_placeholders(action.get("args", {}), f"{w}.args", roots)
             if action.get("as") is not None:
                 if not isinstance(action["as"], str):
                     raise TriggerError(f"{w}: 'as' must be a non-empty AgentEnv-Role other than {DEFAULT_ROLE!r}")
@@ -760,7 +784,7 @@ def _normalize_action(action: Any, w: str, has_ctx: bool, executors: dict[str |
         if "verify" in action:
             action["verify"] = _validate_check(action["verify"], f"{w}.verify")
             for j, step in enumerate(action["verify"]["steps"]):
-                _validate_placeholders(step.get("args", {}), f"{w}.verify.steps[{j}].args", allow_ctx=has_ctx)
+                _validate_placeholders(step.get("args", {}), f"{w}.verify.steps[{j}].args", roots)
 
     def _normalize_time_when(self, when: dict, tid: str) -> None:
         """Validate a when.type=='time' spec fail-loud, leaving it JSON-clean."""
@@ -840,17 +864,15 @@ def on_tool_call(self, role: str, name: str, arguments: dict,
         hold the call for: the widest of their timeouts, counting an omitted one as the gateway
         default so a neighbour's tighter ask cannot cut it short.
         """
-        pending: list[asyncio.Task] = []
-        timeout_s: float | None = None
         provokable = [(tid, trig) for tid, trig in self._triggers.items() if self._provoked_by(trig, role)]
         if not provokable:
-            return pending, timeout_s
+            return [], None
         # Before the isError guard: an errored call is still a clock tick.
         if self._has_time_triggers:
             self._eval_time_triggers()
         # Handled refusals arrive flagged too: BaseService.flag_handled_errors sets isError at the wire.
         if getattr(result, "isError", False):
-            return pending, timeout_s
+            return [], None
         data = _result_data(result)
         ctx = {"args": arguments, "result": data}
         provoking = {"tool": name, "role": role}
@@ -867,10 +889,37 @@ def on_tool_call(self, role: str, name: str, arguments: dict,
             if all(_eval_predicate(pred, _where_value(ctx, field)) for field, pred in unless.get("where", {}).items()):
                 self._cancel(tid, trig, provoking=provoking)
                 self._recompute_time_flag()
-        for tid, trig in provokable:
+        matched = [(tid, trig) for tid, trig in provokable
+                   if trig["spec"]["when"]["type"] == "action" and trig["spec"]["when"]["tool"] == name]
+        pending, timeout_s = self._detect(matched, provoking, ctx)
+        if not self._is_readonly(name):
+            self._reevaluate_states(provokable, provoking)
+        return pending, timeout_s
+
+    def on_website_request(self, role: str, app: str, method: str, path: str, status: int,
+                           args: dict) -> tuple[list[asyncio.Task], float | None]:
+        """Detection hook for an answered website request, before its response goes out; returns what on_tool_call does."""
+        provokable = [(tid, trig) for tid, trig in self._triggers.items() if self._provoked_by(trig, role)]
+        if not provokable:
+            return [], None
+        if self._has_time_triggers:
+            self._eval_time_triggers()
+        request = {"app": app, "method": method, "path": path, "status": status}
+        provoking = {"source": "website", **request, "role": role}
+        matched = [(tid, trig) for tid, trig in provokable
+                   if _website_matches(trig["spec"]["when"], app, method, path, status)]
+        pending, timeout_s = self._detect(matched, provoking, {"args": args, "request": request})
+        if method not in _SAFE_METHODS and status < 400:
+            self._reevaluate_states(provokable, provoking)
+        return pending, timeout_s
+
+    def _detect(self, matched: list[tuple[str, dict]], provoking: dict,
+                ctx: dict) -> tuple[list[asyncio.Task], float | None]:
+        """Detect the matched triggers whose `where` holds and that can take the call; returns what on_tool_call does."""
+        pending: list[asyncio.Task] = []
+        timeout_s: float | None = None
+        for tid, trig in matched:
             when = trig["spec"]["when"]
-            if when["type"] != "action" or when["tool"] != name:
-                continue
             if trig["status"] != "armed" and not (when.get("repeat") and trig["status"] in TRIGGER_IN_FLIGHT_STATUSES):
                 continue
             if not all(_eval_predicate(pred, _where_value(ctx, field))
@@ -882,13 +931,27 @@ def on_tool_call(self, role: str, name: str, arguments: dict,
                 pending.append(task)
                 asked = barrier.get("timeout_seconds") or self._gw.TRIGGER_BARRIER_TIMEOUT_S
                 timeout_s = asked if timeout_s is None else max(timeout_s, asked)
-        if not self._is_readonly(name):
-            for tid, trig in provokable:
-                if trig["status"] == "armed" and trig["spec"]["when"]["type"] == "state" and not trig["evaluating"]:
-                    trig["evaluating"] = True
-                    self._schedule(self._eval_state(tid, trig, provoking))
         return pending, timeout_s
 
+    def _reevaluate_states(self, provokable: list[tuple[str, dict]], provoking: dict) -> None:
+        for tid, trig in provokable:
+            if trig["status"] == "armed" and trig["spec"]["when"]["type"] == "state" and not trig["evaluating"]:
+                trig["evaluating"] = True
+                self._schedule(self._eval_state(tid, trig, provoking))
+
+    def may_hold_website_request(self, role: str, app: str, method: str, path: str, status: int) -> bool:
+        """Whether a `provoking_call` barrier could hold this website request: a matcher check, `where` and status aside."""
+        return any((trig["spec"].get("barrier") or {}).get("at") == "provoking_call" and self._provoked_by(trig, role)
+                   and _website_matches(trig["spec"]["when"], app, method, path, status) for trig in self._triggers.values())
+
+    @property
+    def has_website_triggers(self) -> bool:
+        return any(t["spec"]["when"]["type"] == "website" for t in self._triggers.values())
+
+    @property
+    def has_state_triggers(self) -> bool:
+        return any(t["spec"]["when"]["type"] == "state" for t in self._triggers.values())
+
     def _provoked_by(self, trig: dict, role: str) -> bool:
         """Whether a call by ``role`` can provoke ``trig``: the roles its `when` (or a deadline's `unless`) lists, else the
         watched ones. Unlisted, an executor's own calls never provoke anything, so it cannot set off a cascade unasked."""
@@ -1183,7 +1246,8 @@ async def _fire(self, tid: str, trig: dict, ctx: dict, since: float) -> None:
                 trig["attempts"] += 1
                 try:
                     await self._gw._log_event({"event_type": "trigger_fired", "source": "trigger_engine", "trigger_id": tid,
-                                               "provoking_tool": (ctx.get("args") is not None and spec["when"].get("tool")) or None})
+                                               "provoking_tool": (ctx.get("args") is not None and spec["when"].get("tool")) or None,
+                                               **({"provoking_request": ctx["request"]} if "request" in ctx else {})})
                     for index, action in enumerate(spec["actions"]):
                         if not await self._run_action(tid, trig, index, action, ctx, since):
                             ok = False
@@ -1426,6 +1490,33 @@ def _delay_rngs(spec: dict) -> dict[int, random.Random]:
             if isinstance(action.get("delay"), dict)}
 
 
+def _normalize_provoked_when(when: dict, tid: str) -> None:
+    """Validate what action and website triggers share: `where` over their provoking call's roots, and `repeat`."""
+    roots = _CTX_ROOTS[when["type"]]
+    where = when.get("where", {})
+    if not isinstance(where, dict):
+        raise TriggerError(f"trigger '{tid}': when.where must be an object")
+    for field, pred in where.items():
+        _validate_predicate(pred, f"trigger '{tid}': where.{field}")
+        root_name, path = _ctx_path(field)
+        if path and root_name not in roots:
+            raise TriggerError(f"trigger '{tid}': where.{field} reads a result; a website request carries args, "
+                               "not a result")
+        if path:
+            _validate_path(path, f"trigger '{tid}': where.{field}")
+    if "repeat" in when and not isinstance(when["repeat"], bool):
+        raise TriggerError(f"trigger '{tid}': when.repeat must be a boolean")
+
+
+def _website_matches(when: dict, app: str, method: str, path: str, status: int) -> bool:
+    """Whether a request matches a website `when` (`where` aside); without `status`, only a response below 400 does."""
+    if when["type"] != "website" or when["app"] != app or when.get("method", method) != method:
+        return False
+    if "path" in when and not _eval_predicate(when["path"], path):
+        return False
+    return _eval_predicate(when["status"], status) if "status" in when else status < 400
+
+
 def _normalize_unless(unless: Any, w: str) -> None:
     """Validate a deadline's `unless`: the call that cancels it, matched like an action trigger's."""
     if not isinstance(unless, dict) or not isinstance(unless.get("tool"), str) or not unless["tool"]:
diff --git a/src/agent_env/explorer/ui/src/lib/parse-trigger-events.smoke.ts b/src/agent_env/explorer/ui/src/lib/parse-trigger-events.smoke.ts
index 8d9fd0a3..603b28bf 100644
--- a/src/agent_env/explorer/ui/src/lib/parse-trigger-events.smoke.ts
+++ b/src/agent_env/explorer/ui/src/lib/parse-trigger-events.smoke.ts
@@ -997,6 +997,60 @@ function main(): void {
     );
   }
 
+  // A website request names the cause of website and state-trigger detections alike.
+  {
+    const request = {
+      source: 'website',
+      app: 'mail',
+      method: 'POST',
+      path: '/api/messages',
+      status: 201,
+      role: 'default',
+    };
+    const t = parseTriggerTimeline({
+      state: {
+        'env-x': {
+          triggers: [
+            { id: 'w', type: 'website' },
+            { id: 's', type: 'state' },
+          ],
+          events: ['w', 's'].flatMap((tid, i) => [
+            {
+              seq: 2 * i + 1,
+              ts: `2026-08-06T10:00:0${2 * i}+00:00`,
+              kind: 'detected',
+              trigger_id: tid,
+              provoking: request,
+            },
+            {
+              seq: 2 * i + 2,
+              ts: `2026-08-06T10:00:0${2 * i + 1}+00:00`,
+              kind: 'fired',
+              trigger_id: tid,
+              fire_count: 1,
+            },
+          ]),
+        },
+      },
+    });
+    const row = (tid: string, kind: string) =>
+      t?.rows.find(r => r.kind === kind && r.triggerId === tid);
+    const call = 'POST mail /api/messages → 201 (default)';
+    assert(
+      row('w', 'detected')?.summary === `condition matched on ${call}`,
+      `a website detection names the request (got ${row('w', 'detected')?.summary})`,
+    );
+    assert(
+      row('w', 'fired')?.reason === `the agent requested ${call}`,
+      `a website firing cites the request (got ${row('w', 'fired')?.reason})`,
+    );
+    assert(
+      row('s', 'fired')?.reason ===
+        `its check matched, re-evaluated after ${call}`,
+      `a state firing after a request stays a re-evaluation (got ${row('s', 'fired')?.reason})`,
+    );
+  }
+
   // An unknown trigger type must not fall through to the causal wording. The
   // spec comes from a different part of the payload than the events, so absence
   // is reachable — and "the agent called X" would be wrong for a state trigger.
diff --git a/src/agent_env/explorer/ui/src/lib/parse-trigger-events.ts b/src/agent_env/explorer/ui/src/lib/parse-trigger-events.ts
index 7190a364..f69d3c0d 100644
--- a/src/agent_env/explorer/ui/src/lib/parse-trigger-events.ts
+++ b/src/agent_env/explorer/ui/src/lib/parse-trigger-events.ts
@@ -115,6 +115,16 @@ function actionSummary(raw: Record<string, unknown>): string {
   return 'action ran';
 }
 
+/** The website request a detection names as its cause, as `POST mail /api/messages → 201`; undefined for any other cause. */
+function websiteRequest(
+  provoking: Record<string, unknown>,
+): string | undefined {
+  if (provoking.source !== 'website') return undefined;
+  return `${str(provoking.method) ?? '?'} ${str(provoking.app) ?? '?'} ${
+    str(provoking.path) ?? '?'
+  } → ${num(provoking.status) ?? '?'}`;
+}
+
 function envSummary(kind: string, raw: Record<string, unknown>): string {
   const provoking = asRecord(raw.provoking);
   switch (kind) {
@@ -136,9 +146,9 @@ function envSummary(kind: string, raw: Record<string, unknown>): string {
           due === 1 ? '' : 's'
         } due${window}${capped}`;
       }
-      return `condition matched on ${str(provoking?.tool) ?? '?'} (${
-        str(provoking?.role) ?? '?'
-      })`;
+      return `condition matched on ${
+        (provoking && websiteRequest(provoking)) ?? str(provoking?.tool) ?? '?'
+      } (${str(provoking?.role) ?? '?'})`;
     case 'anchored':
       return `anchored to ${str(raw.anchor) ?? '?'} — due ${shortStamp(
         str(raw.mark),
@@ -464,7 +474,7 @@ function envFiringReason(
     return `the virtual clock reached ${shortStamp(mark)}${capped}`;
   }
   const provoking = detected && asRecord(detected.raw.provoking);
-  const tool = str(provoking?.tool);
+  const tool = (provoking && websiteRequest(provoking)) ?? str(provoking?.tool);
   if (!tool) return undefined;
   const role = str(provoking?.role);
   const call = role ? `${tool} (${role})` : tool;
@@ -474,6 +484,7 @@ function envFiringReason(
   if (spec?.type === 'state')
     return `its check matched, re-evaluated after ${call}`;
   if (spec?.type === 'action') return `the agent called ${call}`;
+  if (spec?.type === 'website') return `the agent requested ${call}`;
   return `it was detected after ${call}`;
 }
 
diff --git a/src/agent_env/explorer/ui/src/lib/parse-triggers.smoke.ts b/src/agent_env/explorer/ui/src/lib/parse-triggers.smoke.ts
index cefa403d..fa09967c 100644
--- a/src/agent_env/explorer/ui/src/lib/parse-triggers.smoke.ts
+++ b/src/agent_env/explorer/ui/src/lib/parse-triggers.smoke.ts
@@ -406,6 +406,27 @@ function main(): void {
     );
   }
 
+  {
+    assert(
+      summarizeWhen({
+        type: 'website',
+        app: 'mail',
+        method: 'POST',
+        path: { regex: '^/api/messages' },
+      }).label === 'POST mail /^/api/messages/',
+      'website: method, site and path regex',
+    );
+    assert(
+      summarizeWhen({ type: 'website', app: 'mail', path: { equals: '/' } })
+        .label === 'any mail /',
+      'website: no method is any, an exact path as is',
+    );
+    assert(
+      summarizeWhen({ type: 'website', app: 'mail' }).label === 'any mail',
+      'website: no path predicate, just the site',
+    );
+  }
+
   {
     // Dangling env_trigger reference: node kept, edge dropped.
     const graph = parseTriggerGraph([
diff --git a/src/agent_env/explorer/ui/src/lib/parse-triggers.ts b/src/agent_env/explorer/ui/src/lib/parse-triggers.ts
index adf9dbff..9ac90a2c 100644
--- a/src/agent_env/explorer/ui/src/lib/parse-triggers.ts
+++ b/src/agent_env/explorer/ui/src/lib/parse-triggers.ts
@@ -181,6 +181,15 @@ export function summarizeWhen(when: unknown): WhenSummary {
     case 'action':
       label = String(w.tool ?? '?');
       break;
+    case 'website': {
+      // The path predicate's own text: an exact path, or a regex between slashes.
+      const path = asRecord(w.path);
+      let shown = '';
+      if (typeof path?.equals === 'string') shown = ` ${path.equals}`;
+      else if (typeof path?.regex === 'string') shown = ` /${path.regex}/`;
+      label = `${String(w.method ?? 'any')} ${String(w.app ?? '?')}${shown}`;
+      break;
+    }
     case 'state': {
       const steps = asRecord(w.check)?.steps;
       label = Array.isArray(steps) ? `${steps.length}-step check` : 'check';
diff --git a/src/agent_env/task_step/task_steps/register_env_triggers.py b/src/agent_env/task_step/task_steps/register_env_triggers.py
index d92497f0..327e463a 100644
--- a/src/agent_env/task_step/task_steps/register_env_triggers.py
+++ b/src/agent_env/task_step/task_steps/register_env_triggers.py
@@ -18,6 +18,7 @@
     TRIGGER_NAMED_EXECUTORS,
     TRIGGER_PROVOKING_ROLES,
     TRIGGER_UNLESS,
+    TRIGGER_WEBSITE,
 )
 from agent_env.task_step.context import TaskStepContext
 from agent_env.entity.refs import EntityRef
@@ -42,6 +43,10 @@ class RegisterEnvTriggersStep(TaskStep):
     each other. Such a chain ends when a persona's turn makes no matching write, so give conversational nl actions no
     `verify` (its retry forces a write), give persona prompts a stop condition, and prefer a `where` off-ramp. An
     action's `delay` and `executor_hours` pace replies in virtual time, and a time trigger's `unless` makes it a deadline.
+    A `website` trigger fires on requests the gateway proxies to the env's websites, for agents with their own browser.
+    A browser sends no AgentEnv-Role header, so its requests count as the `default` role: when `watch_roles` names the
+    agent's role, give a website trigger `when.roles: ["default"]`, or have the agent's client send the header.
+    A front door that proxies the browser's site requests to the gateway can also set the header, dropping the client's.
     `await_fired` holds the step until the named one-shot triggers have fired (a recurrence re-arms after each fire), so
     a dependent step starts only once start-of-run time triggers have landed in the world."""
 
@@ -142,7 +147,9 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
         # would fire even after its `unless` call.
         used = {TRIGGER_NAMED_EXECUTORS: bool(named), TRIGGER_PROVOKING_ROLES: _when_has(self.triggers, "roles"),
                 TRIGGER_ACTION_DELAY: _delays(self.triggers), TRIGGER_EXECUTOR_HOURS: bool(self.executor_hours),
-                TRIGGER_UNLESS: _when_has(self.triggers, "unless")}
+                TRIGGER_UNLESS: _when_has(self.triggers, "unless"),
+                TRIGGER_WEBSITE: any(isinstance(t.get("when"), dict) and t["when"].get("type") == "website"
+                                     for t in self.triggers)}
         missing = [feature for feature, needed in used.items() if needed and feature not in _trigger_features(deployed)]
         if missing:
             raise RuntimeError(
diff --git a/tst/integration/env/gateway/gateway_test.py b/tst/integration/env/gateway/gateway_test.py
index a7b58644..abc6a7c4 100644
--- a/tst/integration/env/gateway/gateway_test.py
+++ b/tst/integration/env/gateway/gateway_test.py
@@ -25,6 +25,7 @@
 from agentenv_protocol import DataPart, FilePart, ToolSession, client as protocol_v1
 from agent_env.env.envs import WebsiteEnv
 from agent_env.env.gateway import AGENT_ENV_ROLE_HEADER, GatewayMode, TOOL_DISABLE_ACTION, TOOL_ENABLE_ACTION
+from agent_env.env.gateway.constants import EXT_TRIGGERS_URI, TRIGGER_WEBSITE
 from agent_env.env.gateway.get_time import (
     GET_TIME_TOOL_DESCRIPTION as GET_TIME_DESC,
     GET_TIME_TOOL_NAME as GET_TIME,
@@ -2822,6 +2823,42 @@ async def test_gateway_trigger_registration_is_fail_loud(sandbox_provider, mcp_s
         await email_env.close()
 
 
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_gateway_website_trigger_fires_on_a_request_through_the_website_proxy(slack_website_env):
+    """A request through the /website proxy fires the website trigger that names it, held by its barrier and attributed."""
+    visit = {"id": "visit", "when": {"type": "website", "app": "slack", "method": "GET", "path": {"equals": "/"}},
+             "barrier": {"at": "provoking_call"},
+             "actions": [{"type": "permission", "action": "disable", "role": "default", "tools": ["list_website_urls"]}]}
+    try:
+        result = await slack_website_env.deploy()
+        gw = result.gateway_url
+        card = await protocol_v1.get_card(gw)
+        assert TRIGGER_WEBSITE in protocol_v1.extension_params(card, EXT_TRIGGERS_URI)["features"], card
+
+        async with httpx.AsyncClient() as client:
+            unknown = await client.post(f"{gw}/triggers/register", timeout=30, json={
+                "triggers": [{**visit, "when": {**visit["when"], "app": "calendar"}}]})
+            assert unknown.status_code == 400 and "unknown website 'calendar'" in unknown.text, unknown.text
+            registered = await client.post(f"{gw}/triggers/register", json={"triggers": [visit]}, timeout=30)
+            assert registered.status_code == 200, registered.text
+
+            missing = await client.get(f"{gw}/website/slack/missing.css", timeout=30)
+            assert missing.status_code == 404
+            page = await client.get(f"{gw}/website/slack/", timeout=30)
+            assert page.status_code == 200 and "Slack Workspace" in page.text
+            roles = (await client.get(f"{gw}/state", timeout=30)).json()["roles"]
+            assert roles.get("default", {}).get("disabled") == ["list_website_urls"], roles
+
+            state = await _triggers_state(client, gw)
+            [row] = state["triggers"]
+            assert (row["type"], row["status"], row["fire_count"]) == ("website", "fired", 1), row
+            assert [e["provoking"] for e in state["events"] if e["kind"] == "detected"] == [
+                {"source": "website", "app": "slack", "method": "GET", "path": "/", "status": 200, "role": "default"}]
+    finally:
+        await slack_website_env.close()
+
+
 # One internal MCP round trip each, measured at ~8ms. A `queued` trigger exists only while a fire
 # is still running, so the chain is what holds that window open long enough to sample from outside
 # the deployment -- there is no slow tool to lean on, and a fire cannot be paused from the client.
diff --git a/tst/unit/env/gateway/changes_test.py b/tst/unit/env/gateway/changes_test.py
index 68180016..c7d068a5 100644
--- a/tst/unit/env/gateway/changes_test.py
+++ b/tst/unit/env/gateway/changes_test.py
@@ -414,6 +414,132 @@ async def body():
     assert [e for e in _trajectory()[before:] if e["event_type"] == "http_request"] == []
 
 
+MIRROR = {"id": "mirror", "barrier": {"at": "provoking_call"}, "actions": [{"type": "tool", "tool": TOOL.name, "args": {}}],
+          "when": {"type": "website", "app": "mail", "method": "POST", "path": {"equals": "/api/drafts"}}}
+
+
+@pytest.mark.asyncio
+async def test_a_website_write_held_by_its_trigger_s_barrier_does_not_claim_the_trigger_s_write(monkeypatch):
+    changelog = _Changelog()
+
+    async def body():
+        yield b'{"ok": '
+        changelog.write("drafts", "d2")  # sized, yet still writing as its body goes out
+        yield b"true}"
+
+    def website(request: httpx.Request) -> httpx.Response:
+        changelog.write("drafts", "d1")
+        return httpx.Response(201, headers={"content-length": "12"}, content=body())
+
+    transport = httpx.MockTransport(website)
+    monkeypatch.setattr(rest_proxy_module.httpx, "AsyncClient", lambda **kw: _RealAsyncClient(transport=transport, **kw))
+    gw = _gateway(changelog, website_urls={"mail": "http://mail-frontend"})
+    gw._trigger_engine.register({"triggers": [MIRROR]})
+    try:
+        status, body_bytes = await _http(gw, "POST", "/website/mail/api/drafts")
+    finally:
+        await gw._close_child_sessions()
+
+    assert status == 201 and json.loads(body_bytes) == {"ok": True}
+    _, changes = await _http(gw, "GET", "/changes")
+    assert [(c["row_id"], [(a["kind"], a["role"]) for a in c["actors"]]) for c in json.loads(changes)["changes"]] == [
+        ("d1", [("request", "default")]), ("d2", [("request", "default")]), ("m3", [("trigger_action", None)])]
+
+
+@pytest.mark.asyncio
+async def test_a_sized_website_write_no_barrier_holds_still_streams(monkeypatch):
+    changelog = _Changelog()
+    gate = asyncio.Event()
+
+    async def body():
+        yield b"a" * 1024
+        await gate.wait()
+        changelog.write("drafts", "d1")
+        yield b"b" * 1024
+
+    transport = httpx.MockTransport(lambda request: httpx.Response(201, headers={"content-length": "2048"}, content=body()))
+    monkeypatch.setattr(rest_proxy_module.httpx, "AsyncClient", lambda **kw: _RealAsyncClient(transport=transport, **kw))
+    gw = _gateway(changelog, website_urls={"mail": "http://mail-frontend"})
+    gw._trigger_engine.register({"triggers": [MIRROR]})
+    first_chunk = asyncio.Event()
+    received = False
+
+    async def receive():
+        nonlocal received
+        if not received:
+            received = True
+            return {"type": "http.request", "body": b"", "more_body": False}
+        await asyncio.sleep(3600)
+
+    async def send(message):
+        if message["type"] == "http.response.body" and message.get("body"):
+            first_chunk.set()
+
+    scope = {"type": "http", "asgi": {"version": "3.0", "spec_version": "2.3"}, "http_version": "1.1", "method": "POST",
+             "scheme": "http", "path": "/website/mail/api/other", "raw_path": b"/website/mail/api/other",
+             "query_string": b"", "root_path": "", "server": ("g", 80), "client": ("c", 1), "headers": []}
+    task = asyncio.create_task(gw._mcp.streamable_http_app()(scope, receive, send))
+    try:
+        await asyncio.wait_for(first_chunk.wait(), 5)
+    finally:
+        gate.set()
+        await asyncio.wait_for(task, 5)
+        await gw._close_child_sessions()
+    _, changes = await _http(gw, "GET", "/changes")
+    assert [[a["kind"] for a in c["actors"]] for c in json.loads(changes)["changes"]] == [["request"]]
+
+
+@pytest.mark.asyncio
+async def test_a_website_write_no_barrier_holds_provokes_its_triggers_once_its_body_has_ended(monkeypatch):
+    changelog = _Changelog()
+
+    async def body():
+        yield b"a"
+        changelog.write("drafts", "d1")
+        yield b"b"
+
+    transport = httpx.MockTransport(lambda request: httpx.Response(201, content=body()))
+    monkeypatch.setattr(rest_proxy_module.httpx, "AsyncClient", lambda **kw: _RealAsyncClient(transport=transport, **kw))
+    gw = _gateway(changelog, website_urls={"mail": "http://mail-frontend"})
+    gw._trigger_engine.register({"triggers": [{**MIRROR, "barrier": None}]})
+    seen: list = []
+    real = gw._trigger_engine.on_website_request
+    gw._trigger_engine.on_website_request = lambda *a: seen.append(changelog.max_id()) or real(*a)
+    try:
+        status, body_bytes = await _http(gw, "POST", "/website/mail/api/drafts")
+        await asyncio.sleep(0.1)
+    finally:
+        await gw._close_child_sessions()
+
+    assert (status, body_bytes, seen) == (201, b"ab", [1])
+    _, changes = await _http(gw, "GET", "/changes")
+    assert [[a["kind"] for a in c["actors"]] for c in json.loads(changes)["changes"]] == [["request"], ["trigger_action"]]
+
+
+@pytest.mark.asyncio
+async def test_a_website_write_still_streaming_under_a_barrier_keeps_its_window_to_the_last_byte(monkeypatch):
+    changelog = _Changelog()
+
+    async def body():
+        yield b"a"
+        changelog.write("drafts", "d1")
+        yield b"b"
+
+    transport = httpx.MockTransport(lambda request: httpx.Response(201, content=body()))
+    monkeypatch.setattr(rest_proxy_module.httpx, "AsyncClient", lambda **kw: _RealAsyncClient(transport=transport, **kw))
+    gw = _gateway(changelog, website_urls={"mail": "http://mail-frontend"})
+    gw._trigger_engine.register({"triggers": [MIRROR]})
+    try:
+        status, body_bytes = await _http(gw, "POST", "/website/mail/api/drafts")
+    finally:
+        await gw._close_child_sessions()
+
+    assert (status, body_bytes) == (201, b"ab")
+    _, changes = await _http(gw, "GET", "/changes")
+    by_row = {c["row_id"]: [a["kind"] for a in c["actors"]] for c in json.loads(changes)["changes"]}
+    assert by_row["d1"] == ["request"] and "trigger_action" in by_row["m1"]
+
+
 @pytest.mark.asyncio
 async def test_changes_pages_by_after_id_and_limit():
     changelog = _Changelog()
diff --git a/tst/unit/env/gateway/triggers_website_test.py b/tst/unit/env/gateway/triggers_website_test.py
new file mode 100644
index 00000000..59333dda
--- /dev/null
+++ b/tst/unit/env/gateway/triggers_website_test.py
@@ -0,0 +1,421 @@
+"""Website triggers (`when.type: website`): a request the gateway proxies to a website provokes triggers like a tool call."""
+from __future__ import annotations
+
+import asyncio
+import json
+from urllib.parse import unquote, urlsplit
+
+import httpx
+import pytest
+from mcp.types import CallToolResult, TextContent
+
+from agent_env.env.gateway import gateway as gateway_module
+from agent_env.env.gateway import rest_proxy as rest_proxy_module
+from agent_env.env.gateway.clock import Clock
+from agent_env.env.gateway.constants import EXT_TRIGGERS_URI, GATEWAY_EXTENSIONS, TRIGGER_WEBSITE
+from agent_env.env.gateway.gateway import GATEWAY_TRAJECTORY_FILE, Gateway
+from agent_env.env.gateway.triggers import TriggerEngine, TriggerError
+
+_RealAsyncClient = httpx.AsyncClient  # captured before any test patches the module attribute
+MESSAGE = {"to": "dana@example.com", "subject": "Lunch?"}
+
+
+class _FakeGateway:
+    def __init__(self, responses: dict) -> None:
+        self.responses = responses
+        self.calls: list[tuple[str, dict]] = []
+        self.events: list[dict] = []
+        self.rules: list[tuple[str, str, bool]] = []
+        self.website_urls = {"mail": "http://mail-frontend", "shop": "http://shop-frontend"}
+        self._role_rules_lock = asyncio.Lock()
+        self._server_tools: dict = {}
+        self._tool_server_urls: dict = {}
+        self._clock = Clock()
+        self.TRIGGER_BARRIER_TIMEOUT_S = 30.0
+
+    def _apply_rule(self, role, tool, value):
+        self.rules.append((role, tool, value))
+
+    def _query_changelog_id(self):
+        return None
+
+    async def _log_event(self, event):
+        self.events.append(dict(event))
+        return f"event_{len(self.events)}"
+
+
+def _result(payload) -> CallToolResult:
+    return CallToolResult(content=[TextContent(type="text", text=json.dumps(payload))], isError=False)
+
+
+def _make_engine(monkeypatch, responses: dict | None = None) -> TriggerEngine:
+    gw = _FakeGateway(responses or {"mail_deliver_message": lambda a: {"delivered": True}})
+
+    async def fake_internal_call(self, tool_name, arguments, role=None):
+        gw.calls.append((tool_name, dict(arguments)))
+        return _result(gw.responses[tool_name](arguments))
+
+    monkeypatch.setattr(TriggerEngine, "_internal_call", fake_internal_call)
+    engine = TriggerEngine(gw)
+    engine._test_gw = gw
+    return engine
+
+
+async def _settle(engine, timeout=2.0):
+    for _ in range(int(timeout / 0.02)):
+        await asyncio.sleep(0.02)
+        if all(t["status"] not in ("firing", "queued") and not t["evaluating"] for t in engine._triggers.values()):
+            return
+    raise AssertionError(f"engine did not settle: {[(k, v['status']) for k, v in engine._triggers.items()]}")
+
+
+def _detected(engine) -> list[tuple[str, dict]]:
+    return [(e["trigger_id"], e["provoking"]) for e in engine._events if e["kind"] == "detected"]
+
+
+REPLY = {"id": "reply", "when": {"type": "website", "app": "mail", "method": "post",
+                                 "path": {"regex": "^/api/messages/?$"}, "repeat": True},
+         "actions": [{"type": "tool", "tool": "mail_deliver_message",
+                      "args": {"from": "${args.to}", "subject": "Re: ${args.subject}"}}]}
+
+
+# ---------------------------------------------------------------- registration
+
+@pytest.mark.parametrize("when,actions,fragment", [
+    ({"type": "website"}, [], "when.app must be a non-empty string"),
+    ({"type": "website", "app": "calendar"}, [], "unknown website 'calendar' (websites: ['mail', 'shop'])"),
+    ({"type": "website", "app": "mail", "method": 5}, [], "when.method must be a non-empty HTTP method"),
+    ({"type": "website", "app": "mail", "method": " "}, [], "when.method must be a non-empty HTTP method"),
+    ({"type": "website", "app": "mail", "path": "/api"}, [], "when.path: predicate must have exactly one of"),
+    ({"type": "website", "app": "mail", "status": {"regex": "("}}, [], "when.status: invalid regex"),
+    ({"type": "website", "app": "mail", "repeat": "yes"}, [], "when.repeat must be a boolean"),
+    ({"type": "website", "app": "mail", "where": {"result.id": {"exists": True}}}, [],
+     "a website request carries args, not a result"),
+    ({"type": "website", "app": "mail"}, [{"type": "tool", "tool": "t", "args": {"id": "${result.id}"}}],
+     "a website request carries args, not a result"),
+    ({"type": "website", "app": "mail"},
+     [{"type": "tool", "tool": "t", "verify": {"tool": "v", "args": {"id": "${result.id?}"}, "predicate": {"exists": True}}}],
+     "a website request carries args, not a result"),
+    ({"type": "website", "app": "mail", "unless": {"tool": "t"}}, [], "when.unless is only valid on time triggers"),
+    ({"type": "website", "app": "mail", "roles": []}, [], "when.roles must be a non-empty list"),
+])
+def test_registration_fails_loud(monkeypatch, when, actions, fragment):
+    engine = _make_engine(monkeypatch)
+    with pytest.raises(TriggerError) as e:
+        engine.register({"triggers": [{"id": "x", "when": when, "actions": actions}]})
+    assert fragment in str(e.value)
+
+
+def test_registration_normalizes_the_method_and_state_echoes_the_request_matcher(monkeypatch):
+    engine = _make_engine(monkeypatch)
+    spec = {**REPLY, "when": {**REPLY["when"], "status": {"regex": "^2"}, "roles": ["default"],
+                              "where": {"to": {"exists": True}}},
+            "barrier": {"at": "provoking_call", "timeout_seconds": 5}}
+    engine.register({"triggers": [spec]})
+    engine.register({"triggers": [spec]})
+    [state] = engine.state()["triggers"]
+    assert state["type"] == "website" and state["barrier"] == {"at": "provoking_call", "timeout_seconds": 5}
+    assert state["when"] == {"type": "website", "app": "mail", "method": "POST", "path": {"regex": "^/api/messages/?$"},
+                             "status": {"regex": "^2"}, "roles": ["default"], "repeat": True}
+
+
+def test_the_card_declares_website_triggers():
+    triggers = next(e for e in GATEWAY_EXTENSIONS if e["uri"] == EXT_TRIGGERS_URI)
+    assert TRIGGER_WEBSITE in triggers["params"]["features"]
+
+
+# ---------------------------------------------------------------- detection
+
+@pytest.mark.asyncio
+async def test_a_matching_request_fires_with_its_args_and_is_attributed_in_the_log(monkeypatch):
+    engine = _make_engine(monkeypatch)
+    engine.register({"triggers": [REPLY]})
+    assert engine.on_website_request("default", "mail", "POST", "/api/messages", 201, dict(MESSAGE)) == ([], None)
+    await _settle(engine)
+
+    assert engine._test_gw.calls == [("mail_deliver_message", {"from": "dana@example.com", "subject": "Re: Lunch?"})]
+    request = {"app": "mail", "method": "POST", "path": "/api/messages", "status": 201}
+    assert _detected(engine) == [("reply", {"source": "website", **request, "role": "default"})]
+    [fired] = [e for e in engine._test_gw.events if e["event_type"] == "trigger_fired"]
+    assert fired["provoking_request"] == request and fired["provoking_tool"] is None
+
+
+@pytest.mark.asyncio
+async def test_only_a_request_matching_app_method_path_status_and_where_provokes(monkeypatch):
+    engine = _make_engine(monkeypatch)
+    engine.register({"triggers": [{**REPLY, "when": {**REPLY["when"], "where": {"to": {"regex": r"@example\.com$"}}}}]})
+    for app, method, path, status, args in [
+        ("shop", "POST", "/api/messages", 201, MESSAGE),
+        ("mail", "GET", "/api/messages", 200, MESSAGE),
+        ("mail", "POST", "/api/messages/7/archive", 200, MESSAGE),
+        ("mail", "POST", "/api/messages", 500, MESSAGE),
+        ("mail", "POST", "/api/messages", 201, {**MESSAGE, "to": "lee@elsewhere.test"}),
+        ("mail", "POST", "/api/messages/", 303, MESSAGE),
+    ]:
+        engine.on_website_request("default", app, method, path, status, dict(args))
+        await _settle(engine)
+    assert [p["path"] for _, p in _detected(engine)] == ["/api/messages/"]
+    assert len(engine._test_gw.calls) == 1
+
+
+@pytest.mark.asyncio
+async def test_a_status_predicate_can_match_a_failed_request(monkeypatch):
+    engine = _make_engine(monkeypatch)
+    engine.register({"triggers": [{"id": "lockout", "when": {"type": "website", "app": "mail", "method": "POST",
+                                                             "path": {"equals": "/login"}, "status": {"equals": 401},
+                                                             "repeat": True},
+                                   "actions": [{"type": "permission", "action": "disable", "role": "default",
+                                                "tools": "*"}]}]})
+    engine.on_website_request("default", "mail", "POST", "/login", 200, {})
+    engine.on_website_request("default", "mail", "POST", "/login", 401, {})
+    await _settle(engine)
+    assert [p["status"] for _, p in _detected(engine)] == [401]
+    assert engine._test_gw.rules == [("default", "*", True)]
+
+
+@pytest.mark.asyncio
+async def test_a_one_shot_fires_once_and_a_repeat_once_per_request(monkeypatch):
+    engine = _make_engine(monkeypatch)
+    once = {**REPLY, "id": "once", "when": {k: v for k, v in REPLY["when"].items() if k != "repeat"}}
+    engine.register({"triggers": [REPLY, once]})
+    for _ in range(3):
+        engine.on_website_request("default", "mail", "POST", "/api/messages", 201, dict(MESSAGE))
+        await _settle(engine)
+    assert {tid: t["fire_count"] for tid, t in engine._triggers.items()} == {"reply": 3, "once": 1}
+
+
+@pytest.mark.asyncio
+async def test_requests_by_an_unwatched_role_provoke_only_the_triggers_that_list_it(monkeypatch):
+    engine = _make_engine(monkeypatch)
+    listed = {**REPLY, "id": "listed", "when": {**REPLY["when"], "roles": ["carol@example.com"]}}
+    engine.register({"watch_roles": ["default"], "triggers": [REPLY, listed]})
+    engine.on_website_request("carol@example.com", "mail", "POST", "/api/messages", 201, dict(MESSAGE))
+    await _settle(engine)
+    assert [(tid, p["role"]) for tid, p in _detected(engine)] == [("listed", "carol@example.com")]
+
+
+@pytest.mark.asyncio
+async def test_a_barrier_hands_back_the_fire_to_hold_the_response_for(monkeypatch):
+    engine = _make_engine(monkeypatch)
+    engine.register({"triggers": [{**REPLY, "barrier": {"at": "provoking_call", "timeout_seconds": 5}}]})
+    pending, timeout_s = engine.on_website_request("default", "mail", "POST", "/api/messages", 201, dict(MESSAGE))
+    assert len(pending) == 1 and timeout_s == 5
+    await asyncio.wait(pending, timeout=2)
+    assert engine._triggers["reply"]["fire_count"] == 1
+
+
+@pytest.mark.asyncio
+async def test_only_an_unsafe_request_that_succeeded_prompts_the_state_triggers(monkeypatch):
+    inbox: list[str] = []
+    engine = _make_engine(monkeypatch, {"mail_list_messages": lambda a: {"subjects": list(inbox)}})
+    engine.register({"triggers": [{"id": "sent", "when": {"type": "state", "check": {
+        "tool": "mail_list_messages", "extract": "subjects", "predicate": {"regex": "Lunch"}}}, "actions": []}]})
+    inbox.append("Lunch?")
+    engine.on_website_request("default", "mail", "GET", "/api/messages", 200, {})
+    engine.on_website_request("default", "mail", "POST", "/api/messages", 500, dict(MESSAGE))
+    await _settle(engine)
+    assert engine._test_gw.calls == [] and engine._triggers["sent"]["status"] == "armed"
+
+    engine.on_website_request("default", "mail", "POST", "/api/messages", 201, dict(MESSAGE))
+    await _settle(engine)
+    assert engine._test_gw.calls == [("mail_list_messages", {})] and engine._triggers["sent"]["status"] == "fired"
+    assert _detected(engine)[0][1]["source"] == "website"
+
+
+# ---------------------------------------------------------------- through the gateway's /website proxy
+
+class _Website:
+    """A MockTransport upstream for the gateway's websites, answering every request with `status`."""
+
+    def __init__(self, status: int = 201) -> None:
+        self.status = status
+        self.requests: list[httpx.Request] = []
+
+    def install(self, monkeypatch) -> None:
+        async def body():
+            yield b'{"ok": true}'
+
+        def handler(request: httpx.Request) -> httpx.Response:
+            self.requests.append(request)
+            return httpx.Response(self.status, headers={"content-type": "application/json"}, content=body())
+
+        transport = httpx.MockTransport(handler)
+        monkeypatch.setattr(rest_proxy_module.httpx, "AsyncClient", lambda **kw: _RealAsyncClient(transport=transport, **kw))
+
+
+def _gateway() -> Gateway:
+    return Gateway(host="127.0.0.1", port=0, server_name="t", internal_mcp_servers=[],
+                   website_urls={"mail": "http://mail-frontend"})
+
+
+async def _request(gw: Gateway, method: str, target: str, *, headers: dict | None = None, body: bytes = b"",
+                   order: list | None = None) -> int:
+    """Run one request through the gateway's ASGI app; return its status, appending "response" to `order` at the status line."""
+    url = urlsplit(target)
+    scope = {
+        "type": "http", "asgi": {"version": "3.0", "spec_version": "2.3"}, "http_version": "1.1",
+        "method": method, "scheme": "http", "path": unquote(url.path), "raw_path": url.path.encode(),
+        "query_string": url.query.encode(), "root_path": "", "server": ("gateway", 80), "client": ("c", 1),
+        "headers": [(k.lower().encode(), v.encode()) for k, v in (headers or {}).items()],
+    }
+    done = asyncio.Event()
+    received = False
+    statuses: list[int] = []
+
+    async def receive():
+        nonlocal received
+        if not received:
+            received = True
+            return {"type": "http.request", "body": body, "more_body": False}
+        await done.wait()
+        return {"type": "http.disconnect"}
+
+    async def send(message):
+        if message["type"] == "http.response.start":
+            statuses.append(message["status"])
+            if order is not None:
+                order.append("response")
+
+    try:
+        await gw._mcp.streamable_http_app()(scope, receive, send)
+    finally:
+        done.set()
+    return statuses[0]
+
+
+def _trajectory() -> list[dict]:
+    with open(GATEWAY_TRAJECTORY_FILE) as f:
+        return [json.loads(line) for line in f]
+
+
+@pytest.mark.asyncio
+async def test_a_barrier_holds_the_website_s_response_until_the_mirror_has_landed(monkeypatch):
+    _Website(status=201).install(monkeypatch)
+    order: list[str] = []
+    calls: list[tuple[str, dict, str | None]] = []
+
+    async def slow_internal_call(self, tool_name, arguments, role=None):
+        await asyncio.sleep(0.05)
+        calls.append((tool_name, dict(arguments), role))
+        order.append("mirror")
+        return _result({"delivered": True})
+
+    monkeypatch.setattr(TriggerEngine, "_internal_call", slow_internal_call)
+    gw = _gateway()
+    gw._trigger_engine.register({"triggers": [{**REPLY, "barrier": {"at": "provoking_call"}}]})
+    before = len(_trajectory())
+
+    status = await _request(gw, "POST", "/website/mail/api/messages", headers={"content-type": "application/json"},
+                            body=json.dumps(MESSAGE).encode(), order=order)
+
+    assert status == 201 and order == ["mirror", "response"]
+    assert calls == [("mail_deliver_message", {"from": "dana@example.com", "subject": "Re: Lunch?"}, None)]
+    [fired] = [e for e in _trajectory()[before:] if e["event_type"] == "trigger_fired"]
+    assert fired["provoking_request"] == {"app": "mail", "method": "POST", "path": "/api/messages", "status": 201}
+
+
+@pytest.mark.asyncio
+async def test_the_proxy_hands_the_engine_the_role_path_status_and_args_of_each_request(monkeypatch):
+    website = _Website(status=200)
+    website.install(monkeypatch)
+    gw = _gateway()
+    gw._trigger_engine.register({"triggers": [REPLY]})
+    seen: list[tuple] = []
+    gw._trigger_engine.on_website_request = lambda *a: seen.append(a) or ([], None)
+    form = {"content-type": "application/x-www-form-urlencoded; charset=utf-8"}
+    json_body = {"content-type": "application/json"}
+
+    await _request(gw, "GET", "/website/mail")
+    await _request(gw, "GET", "/website/mail/inbox?folder=work&tag=a&tag=b", headers={"AgentEnv-Role": "carol@example.com"})
+    await _request(gw, "POST", "/website/mail/api/messages?draft=1", headers=json_body,
+                   body=json.dumps({"to": "dana@example.com", "draft": "0"}).encode())
+    await _request(gw, "POST", "/website/mail/compose", headers=form, body=b"subject=Hi+there&to=dana%40example.com")
+    await _request(gw, "POST", "/website/mail/api/batch?op=send", headers=json_body, body=b"[1, 2]")
+    await _request(gw, "PUT", "/website/mail/api/drafts/a%2Fb", headers=json_body, body=b"not json")
+    assert await _request(gw, "GET", "/website/calendar/") == 404
+
+    assert seen == [
+        ("default", "mail", "GET", "/", 200, {}),
+        ("carol@example.com", "mail", "GET", "/inbox", 200, {"folder": "work", "tag": ["a", "b"]}),
+        ("default", "mail", "POST", "/api/messages", 200, {"draft": "0", "to": "dana@example.com"}),
+        ("default", "mail", "POST", "/compose", 200, {"subject": "Hi there", "to": "dana@example.com"}),
+        ("default", "mail", "POST", "/api/batch", 200, {"op": "send"}),
+        ("default", "mail", "PUT", "/api/drafts/a/b", 200, {}),
+    ]
+    assert [r.content for r in website.requests][2:4] == [b'{"to": "dana@example.com", "draft": "0"}',
+                                                          b"subject=Hi+there&to=dana%40example.com"]
+
+
+@pytest.mark.asyncio
+async def test_without_a_website_trigger_a_mutating_website_request_still_rechecks_the_state_triggers(monkeypatch):
+    _Website(status=201).install(monkeypatch)
+    calls: list[str] = []
+
+    async def internal_call(self, tool_name, arguments, role=None):
+        calls.append(tool_name)
+        return _result({"subjects": ["Lunch?"]})
+
+    monkeypatch.setattr(TriggerEngine, "_internal_call", internal_call)
+    gw = _gateway()
+    gw._trigger_engine.register({"triggers": [{"id": "sent", "actions": [], "when": {"type": "state", "check": {
+        "tool": "mail_list_messages", "extract": "subjects", "predicate": {"regex": "Lunch"}}}}]})
+    parsed: list[bytes] = []
+    real_args = gateway_module._request_args
+    monkeypatch.setattr(gateway_module, "_request_args", lambda request, body: parsed.append(body) or real_args(request, body))
+
+    assert await _request(gw, "POST", "/website/mail/api/messages", headers={"content-type": "application/json"},
+                          body=json.dumps(MESSAGE).encode()) == 201
+    await asyncio.sleep(0.1)
+
+    assert parsed == [] and calls == ["mail_list_messages"] and gw._trigger_engine._triggers["sent"]["status"] == "fired"
+
+
+@pytest.mark.asyncio
+async def test_with_neither_a_website_nor_a_state_trigger_the_proxy_never_reaches_the_engine(monkeypatch):
+    _Website(status=201).install(monkeypatch)
+    gw = _gateway()
+    gw._trigger_engine.register({"triggers": [{"id": "later", "actions": [], "when": {"type": "time", "at": "PT1H"}}]})
+    seen: list[tuple] = []
+    real = gw._trigger_engine.on_website_request
+    gw._trigger_engine.on_website_request = lambda *a: seen.append(a) or real(*a)
+
+    assert await _request(gw, "POST", "/website/mail/api/messages", headers={"content-type": "application/json"},
+                          body=json.dumps(MESSAGE).encode()) == 201
+
+    assert seen == []
+
+
+@pytest.mark.asyncio
+async def test_with_a_website_trigger_a_mutating_website_request_rechecks_the_state_triggers(monkeypatch):
+    _Website(status=201).install(monkeypatch)
+    calls: list[str] = []
+
+    async def internal_call(self, tool_name, arguments, role=None):
+        calls.append(tool_name)
+        return _result({"subjects": ["Lunch?"], "delivered": True})
+
+    monkeypatch.setattr(TriggerEngine, "_internal_call", internal_call)
+    gw = _gateway()
+    gw._trigger_engine.register({"triggers": [
+        {"id": "sent", "actions": [], "when": {"type": "state", "check": {
+            "tool": "mail_list_messages", "extract": "subjects", "predicate": {"regex": "Lunch"}}}},
+        {**REPLY, "when": {**REPLY["when"], "path": {"equals": "/never"}}}]})
+
+    await _request(gw, "POST", "/website/mail/api/messages", headers={"content-type": "application/json"},
+                   body=json.dumps(MESSAGE).encode())
+    await asyncio.sleep(0.1)
+
+    assert calls == ["mail_list_messages"] and gw._trigger_engine._triggers["sent"]["status"] == "fired"
+
+
+@pytest.mark.asyncio
+async def test_a_failing_trigger_evaluation_still_answers_the_request(monkeypatch):
+    _Website(status=200).install(monkeypatch)
+    gw = _gateway()
+    gw._trigger_engine.register({"triggers": [REPLY]})
+
+    def explode(*_):
+        raise RuntimeError("boom")
+
+    gw._trigger_engine.on_website_request = explode
+    assert await _request(gw, "GET", "/website/mail/inbox") == 200
diff --git a/tst/unit/task_step/test_register_env_triggers.py b/tst/unit/task_step/test_register_env_triggers.py
index 9e11b4bf..6b99efa1 100644
--- a/tst/unit/task_step/test_register_env_triggers.py
+++ b/tst/unit/task_step/test_register_env_triggers.py
@@ -228,6 +228,14 @@ async def test_a_gateway_without_provoking_roles_refuses_a_trigger_that_lists_ro
     assert sent == []
 
 
+@pytest.mark.asyncio
+async def test_a_gateway_that_declares_website_triggers_takes_them(monkeypatch):
+    sent = _mock_gateway(monkeypatch, {"ok": True, "added": ["visit"], "all": ["visit"]})
+    triggers = [{"id": "visit", "when": {"type": "website", "app": "mail", "method": "GET"}, "actions": []}]
+    await _step(triggers=triggers).execute(_context())
+    assert json.loads(sent[0].content)["triggers"] == triggers
+
+
 _HOURS = {"start": "09:00", "end": "17:00", "utc_offset": "-07:00"}
 
 
@@ -264,7 +272,8 @@ async def test_executor_hours_for_an_agent_no_trigger_uses_raise(monkeypatch):
      None, "unless"),
     ([{"id": "x", "when": {"type": "action", "tool": "t"}, "actions": [{"type": "nl", "instruction": "x"}]}],
      {"exec-1": _HOURS}, "executor_hours"),
-], ids=["delay", "unless", "hours"])
+    ([{"id": "x", "when": {"type": "website", "app": "mail", "method": "POST"}, "actions": []}], None, "website"),
+], ids=["delay", "unless", "hours", "website"])
 async def test_a_gateway_without_a_feature_refuses_the_triggers_that_use_it(monkeypatch, triggers, hours, missing):
     sent = _mock_gateway(monkeypatch, {})
     context = _context(with_executor=True)