feat(triggers): fire env triggers on requests to the env's websites

src/agent_env/env/gateway/constants.py

@@ -114,8 +114,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",

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
@@ -177,14 +178,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:
@@ -508,6 +509,26 @@ async def _proxy_rest_request(self, request: Request, url_map: dict[str, str] |
         return await forward(request, base_url, role=self._request_role(request), timeout=self.REST_PROXY_TIMEOUT,
                              mcp_server=url_map is self.rest_proxy_urls and service_name.startswith("mcp-"))
 
+    async def _proxy_website_request(self, request: Request) -> Response:
+        """Proxy a website request; with a website trigger registered, feed it to the engine and hold it for barriers."""
+        response = await self._proxy_rest_request(request, url_map=self.website_urls)
+        app = request.path_params["service_name"]
+        if app in self.website_urls and self._trigger_engine.has_website_triggers:
+            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:
+            args = _request_args(request, await request.body())
+            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."""
@@ -1193,6 +1214,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),