diff --git a/src/agent_env/env/gateway/constants.py b/src/agent_env/env/gateway/constants.py index cdc8420c..3b651d78 100644 --- a/src/agent_env/env/gateway/constants.py +++ b/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", diff --git a/src/agent_env/env/gateway/gateway.py b/src/agent_env/env/gateway/gateway.py index 4a4a4d4d..7bd0d891 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 @@ -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,30 @@ 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 request to one of the env's websites, then, once the website has answered, feed it to the trigger + engine and hold the response for the barriers it provokes: an agent browsing with its own browser or HTTP + client provokes triggers as a tool call does.""" + response = await self._proxy_rest_request(request, url_map=self.website_urls) + app = request.path_params["service_name"] + if app in self.website_urls: + barrier_tasks, barrier_timeout_s = self._fire_website_triggers(request, app, response.status_code, + await request.body()) + await self._await_barriers(barrier_tasks, barrier_timeout_s) + return response + + def _fire_website_triggers(self, request: Request, app: str, status: int, + body: bytes) -> tuple[list[asyncio.Task], float | None]: + """Feed an answered website request to the trigger engine (fail-open), with its path beneath the website's root + and its query and body fields as its args.""" + try: + path = "/" + request.path_params.get("path", "") + return self._trigger_engine.on_website_request(self._request_role(request), app, request.method, path, + status, _request_args(request, body)) + 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 +1218,30 @@ def model_dump_one_level(self) -> dict[str, Any]: return kwargs +def _request_args(request: Request, body: bytes) -> dict: + """A website request's args, as a trigger reads them: its query parameters, then the fields of a urlencoded form or + JSON-object body, which win a clash. A parameter given more than once is the list of its values.""" + 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/triggers.py b/src/agent_env/env/gateway/triggers.py index 41da5e1d..af3a737d 100644 --- a/src/agent_env/env/gateway/triggers.py +++ b/src/agent_env/env/gateway/triggers.py @@ -1,5 +1,6 @@ """Typed Action/State trigger engine (urn:agentenv:triggers/v1) hosted on the gateway: -post-call detection on watched-role traffic, async fail-open firing, additive fail-loud registration.""" +post-call detection on watched-role traffic (tool calls and requests to the env's websites), async fail-open +firing, additive fail-loud registration.""" from __future__ import annotations import asyncio @@ -26,6 +27,12 @@ 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 type absent here has no provoking call. A +# website response streams through the proxy unread, so a website request carries no result. +_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 +41,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 +330,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 where the trigger's provoking call carries that + root (`roots`, from _CTX_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 +344,10 @@ 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 call; only action and website triggers carry one") + 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 +640,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 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 +651,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 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 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 +696,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, state 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 +706,32 @@ 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 when.type=='website' spec: one of the gateway's websites, the method (normalized to upper case), + the `path` and `status` predicates, then `where` over the request's args and `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 +764,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 +786,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.""" @@ -837,20 +863,17 @@ def on_tool_call(self, role: str, name: str, arguments: dict, """Synchronous post-call detection hook (invoked from the gateway's role-filter wrapper). Returns the fire tasks of triggers whose barrier waits at `provoking_call`, and how long to - 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. + hold the call for (see _detect). """ - 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 +890,43 @@ 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]: + """Synchronous detection hook for a request the gateway proxied to its website ``app``, invoked once the website + has answered with ``status`` and before the response is passed on; returns what on_tool_call does. ``path`` is + the request's path beneath the website's root, ``args`` its query and body fields. A request that asks for a + change (an unsafe method) and did not fail prompts the state triggers as a non-readonly 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 each trigger the provoking call matched whose `where` holds and that can take it (armed, or a repeat + one mid-fire). Returns the fires whose barrier waits at `provoking_call`, and how long to 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 + 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 +938,14 @@ 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 _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 +1240,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 +1484,35 @@ 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` predicates over the roots their provoking call carries, + and `repeat`. A bare `where` key is a top-level argument whatever its name.""" + 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 website request matches a website trigger's `when`, `where` aside. Without a `status` predicate only + a response below 400 matches, as only a tool call that did not fail provokes an action trigger.""" + 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..d65fbbbb 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,61 @@ function main(): void { ); } + // A website request is named by method, site, path and status, whether it + // matched a website trigger or prompted a state trigger's re-evaluation. + { + 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 { 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 | 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 { const provoking = asRecord(raw.provoking); switch (kind) { @@ -136,9 +146,9 @@ function envSummary(kind: string, raw: Record): 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 17761124..ccd83747 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 @@ -42,6 +42,8 @@ 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 matches the requests the gateway proxies to one of the env's websites, so an agent that browses + with its own browser or HTTP client provokes triggers as one calling tools does. `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.""" diff --git a/tst/integration/env/gateway/gateway_test.py b/tst/integration/env/gateway/gateway_test.py index 601eb5d5..c3c07e06 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, @@ -2717,11 +2718,11 @@ async def test_gateway_trigger_registration_is_fail_loud(sandbox_provider, mcp_s "barrier": {"at": "provoking_call", "timeout_seconds": 0}, "actions": []}, "positive, finite"), ("repeat on a state trigger", {"id": "b", "when": dict(state_when, repeat=True), "actions": []}, - "only valid on action triggers"), + "only valid on action and website triggers"), ("ctx placeholder on a state trigger", {"id": "b", "when": state_when, "actions": [{"type": "tool", "tool": "move_email", "args": {"email_id": "${result.email_id}"}}]}, - "only action triggers carry"), + "only action and website triggers carry"), ] try: result = await email_env.deploy() @@ -2749,6 +2750,44 @@ 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 to the env's website through the gateway's /website proxy, as an agent's own browser makes it, + provokes a website trigger: the barrier holds the page until the action has landed, and the firing log attributes + the request. Only the request the trigger names fires it.""" + 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/triggers_ctx_test.py b/tst/unit/env/gateway/triggers_ctx_test.py index b7d3a328..312829a3 100644 --- a/tst/unit/env/gateway/triggers_ctx_test.py +++ b/tst/unit/env/gateway/triggers_ctx_test.py @@ -131,7 +131,7 @@ def test_template_unresolved_ctx_raises_and_bare_names_stay_literal(): @pytest.mark.parametrize("bad,fragment", [ ({"id": "x", "when": {"type": "time", "at": "PT1H"}, - "actions": [{"type": "tool", "tool": "t", "args": {"a": "${args.b}"}}]}, "only action triggers carry"), + "actions": [{"type": "tool", "tool": "t", "args": {"a": "${args.b}"}}]}, "only action and website triggers carry"), ({"id": "x", "when": {"type": "action", "tool": "t", "repeat": "yes"}, "actions": []}, "repeat must be a boolean"), ({"id": "x", "when": {"type": "action", "tool": "t"}, "barrier": True, "actions": []}, "barrier must be an object"), @@ -146,15 +146,15 @@ def test_template_unresolved_ctx_raises_and_bare_names_stay_literal(): "barrier": {"at": "provoking_call", "timeout_seconds": 0}, "actions": []}, "timeout_seconds must be a positive, finite number"), ({"id": "x", "when": {"type": "time", "at": "PT1H"}, "barrier": {"at": "provoking_call"}, "actions": []}, - "only valid on action triggers"), + "only valid on action and website triggers"), ({"id": "x", "when": {"type": "action", "tool": "t"}, "actions": [{"type": "tool", "tool": "t2", "args": {"a": "${bogus.path}"}}]}, "must be a check binding name"), ({"id": "x", "when": {"type": "action", "tool": "t", "where": {"result.a b": {"exists": True}}}, "actions": []}, "invalid path segment"), ({"id": "x", "when": {"type": "state", "check": {"tool": "t", "predicate": {"exists": True}}, "repeat": True}, - "actions": []}, "only valid on action triggers"), + "actions": []}, "only valid on action and website triggers"), ({"id": "x", "when": {"type": "state", "check": {"tool": "t", "args": {"q": "${args.x}"}, "predicate": {"exists": True}}}, - "actions": []}, "only action triggers carry"), + "actions": []}, "only action and website triggers carry"), ({"id": "x", "when": {"type": "action", "tool": "t"}, "actions": [{"type": "tool", "tool": "t2", "args": {"a": "${result.}"}}]}, "has an empty path"), ]) @@ -478,7 +478,7 @@ def test_optional_placeholder_validates_like_a_required_one(monkeypatch): engine = _make_engine(monkeypatch, {}) engine.register({"triggers": [{"id": "ok", "when": {"type": "action", "tool": "t"}, "actions": [{"type": "tool", "tool": "t2", "args": {"a": "${args.x?}", "b": "${result.y[0]?}"}}]}]}) - with pytest.raises(TriggerError, match="only action triggers carry"): + with pytest.raises(TriggerError, match="only action and website triggers carry"): engine.register({"triggers": [{"id": "bad", "when": {"type": "time", "at": "PT1H"}, "actions": [{"type": "tool", "tool": "t2", "args": {"a": "${args.x?}"}}]}]}) diff --git a/tst/unit/env/gateway/triggers_test.py b/tst/unit/env/gateway/triggers_test.py index 5336a48f..f559d966 100644 --- a/tst/unit/env/gateway/triggers_test.py +++ b/tst/unit/env/gateway/triggers_test.py @@ -207,7 +207,7 @@ def _registration(**overrides): *[({"triggers": [{"id": "x", "when": {"type": "action", "tool": "t", "roles": roles}, "actions": []}]}, "when.roles must be a non-empty list") for roles in ("denise", [], [""], [5])], ({"triggers": [{"id": "x", "when": {"type": "time", "after": "y", "offset": "PT1S", "roles": ["denise"]}, - "actions": []}]}, "only valid on action and state triggers"), + "actions": []}]}, "only valid on action, state and website triggers"), ({"executors": {"denise": {"a2a_url": "http://d", "role": "denise@example.com"}}, "triggers": [{"id": "x", "when": {"type": "action", "tool": "t", "roles": ["denise@example.com"]}, "actions": [{"type": "nl", "instruction": "hi", "executor": "denise"}]}]}, 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..e26d2483 --- /dev/null +++ b/tst/unit/env/gateway/triggers_website_test.py @@ -0,0 +1,358 @@ +"""Website triggers (urn:agentenv:triggers/v1, `when.type: website`): a request the gateway proxies to one of its +websites provokes triggers as a tool call does, for an agent that browses with its own browser or HTTP client.""" +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 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]}) # identical re-add: no-op, though the stored method was upper-cased + [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), # no status predicate: a failed request is no provoking call + ("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, as uvicorn would build its scope; returns the status, and appends + "response" to `order` when the status line goes out.""" + 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() + 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_a_failing_trigger_evaluation_still_answers_the_request(monkeypatch): + _Website(status=200).install(monkeypatch) + gw = _gateway() + + def explode(*_): + raise RuntimeError("boom") + + gw._trigger_engine.on_website_request = explode + assert await _request(gw, "GET", "/website/mail/inbox") == 200