diff --git a/verifiers/v1/agent.py b/verifiers/v1/agent.py index f99158f3a..d02b69928 100644 --- a/verifiers/v1/agent.py +++ b/verifiers/v1/agent.py @@ -334,6 +334,10 @@ def _interception_for( return self.interception if self._server is None: return None + if any(tool.state_secret for tool in shared_tools.values()): + # Shared state credentials are attached per run, after this owned + # server was created; let the rollout size a scoped server instead. + return None if self._server.tunnel is not None or ( run_is_local and not shared_tools and not type(task).tools ): diff --git a/verifiers/v1/env.py b/verifiers/v1/env.py index 0fcee6d17..c6bc3a6d3 100644 --- a/verifiers/v1/env.py +++ b/verifiers/v1/env.py @@ -342,7 +342,13 @@ async def serving(self): run slots inside. Torn down on exit (`teardown()`, then the framework's).""" async with self.shared_tools() as shared: interception = make_interception( - self.config.interception, requires_tunnel=self._requires_tunnel(shared) + self.config.interception, + requires_tunnel=self._requires_tunnel(shared), + state_service_secrets=tuple( + server.state_secret + for server in shared.values() + if server.state_secret + ), ) async with interception: self._shared_tools = shared diff --git a/verifiers/v1/interception/__init__.py b/verifiers/v1/interception/__init__.py index 9d5646f87..ecb999e20 100644 --- a/verifiers/v1/interception/__init__.py +++ b/verifiers/v1/interception/__init__.py @@ -54,16 +54,19 @@ def requires_tunnel( def make_interception( - config: InterceptionConfig, *, requires_tunnel: bool + config: InterceptionConfig, + *, + requires_tunnel: bool, + state_service_secrets: tuple[str, ...] = (), ) -> Interception: """The interception for a config, picked by type (the host-side counterpart to `make_runtime`). With `requires_tunnel`, each server is exposed through its configured tunnel; otherwise it remains on host loopback. The caller computes this requirement.""" if isinstance(config, InterceptionServerConfig): - return InterceptionServer(config, requires_tunnel) + return InterceptionServer(config, requires_tunnel, state_service_secrets) if isinstance(config, StaticInterceptionPoolConfig): - return StaticInterceptionPool(config, requires_tunnel) - return ElasticInterceptionPool(config, requires_tunnel) + return StaticInterceptionPool(config, requires_tunnel, state_service_secrets) + return ElasticInterceptionPool(config, requires_tunnel, state_service_secrets) __all__ = [ diff --git a/verifiers/v1/interception/base.py b/verifiers/v1/interception/base.py index 2a073ec96..95d771bd4 100644 --- a/verifiers/v1/interception/base.py +++ b/verifiers/v1/interception/base.py @@ -26,12 +26,10 @@ class BaseInterceptionConfig(BaseConfig): `multiplex`).""" -# (base_url, secret): the interception server's reachable base URL for this rollout, and the -# bearer the harness/tool servers authenticate with. The harness reaches the model at -# `{base_url}/v1`; tool servers reach this rollout's shared state at `{base_url}/state` -# + `/task`. `base_url` is universally reachable — the interception is exposed (tunnel) +# (base_url, model_secret, state_secret): model inference and task state deliberately use +# separate capabilities. `base_url` is universally reachable — the interception is exposed # whenever any consumer is remote. -Slot = tuple[str, str] +Slot = tuple[str, str, str] class Interception(ABC): diff --git a/verifiers/v1/interception/pool.py b/verifiers/v1/interception/pool.py index f004a59f8..7ce6f058c 100644 --- a/verifiers/v1/interception/pool.py +++ b/verifiers/v1/interception/pool.py @@ -48,12 +48,16 @@ class StaticInterceptionPool(Interception): operator's call (it's the shape for pre-provisioned/bring-your-own endpoints).""" def __init__( - self, config: StaticInterceptionPoolConfig, requires_tunnel: bool = False + self, + config: StaticInterceptionPoolConfig, + requires_tunnel: bool = False, + state_service_secrets: tuple[str, ...] = (), ) -> None: super().__init__() self.config = config self.servers = [ - InterceptionServer(server, requires_tunnel) for server in config.servers + InterceptionServer(server, requires_tunnel, state_service_secrets) + for server in config.servers ] async def start(self) -> None: @@ -89,10 +93,12 @@ def __init__( self, config: ElasticInterceptionPoolConfig | None = None, requires_tunnel: bool = False, + state_service_secrets: tuple[str, ...] = (), ) -> None: super().__init__() self.config = config or ElasticInterceptionPoolConfig() self.requires_tunnel = requires_tunnel + self.state_service_secrets = state_service_secrets self.servers: list[InterceptionServer] = [] self._lock = asyncio.Lock() self._warm_task: asyncio.Task[InterceptionServer] | None = None @@ -116,7 +122,9 @@ async def _server(self) -> InterceptionServer: return server # Pin prime explicitly — the only tunnel kind that can be minted on demand. server = InterceptionServer( - InterceptionServerConfig(tunnel=PrimeTunnelConfig()), self.requires_tunnel + InterceptionServerConfig(tunnel=PrimeTunnelConfig()), + self.requires_tunnel, + self.state_service_secrets, ) await self.stack.enter_async_context(server) self.servers.append(server) @@ -136,8 +144,8 @@ async def acquire(self, session: RolloutSession) -> AsyncIterator[Slot]: # Register under the lock so concurrent acquires see each other's load. async with self._lock: server = await self._server() - secret = server.register(session) + model_secret, state_secret = server.register(session) try: - yield server.base_url, secret + yield server.base_url, model_secret, state_secret finally: - server.unregister(secret) + server.unregister(model_secret, state_secret) diff --git a/verifiers/v1/interception/server.py b/verifiers/v1/interception/server.py index b9e5574d1..32870067a 100644 --- a/verifiers/v1/interception/server.py +++ b/verifiers/v1/interception/server.py @@ -6,10 +6,10 @@ `OPENAI_BASE_URL`/`OPENAI_API_KEY` so the program's SDK talks to us. Both non-streaming and SSE requests are supported. -One server multiplexes many rollouts: each rollout registers a `RolloutSession` under its -own secret (the bearer token the harness already sends), and the server routes by that -secret to the right session. So N rollouts need one server (and, behind a remote runtime, -one tunnel) per pool member rather than one each — see `interception.pool`. +One server multiplexes many rollouts: each rollout registers separate model and state +capabilities, and the server routes each to the right session. So N rollouts need one +server (and, behind a remote runtime, one tunnel) per pool member rather than one each — +see `interception.pool`. The server is a pure model boundary: one request, one turn — refusal checks (limits, `@stop`s), the model call, the graph commit, retry atomicity. A run's user exchange @@ -25,7 +25,7 @@ import secrets import time import traceback -from collections.abc import AsyncIterator +from collections.abc import AsyncIterator, Collection from contextlib import asynccontextmanager from typing import Literal @@ -126,9 +126,13 @@ def __init__( self, config: InterceptionServerConfig | None = None, requires_tunnel: bool = False, + state_service_secrets: Collection[str] = (), ) -> None: super().__init__() self.sessions: dict[str, RolloutSession] = {} + self.state_sessions: dict[str, RolloutSession] = {} + self.state_routes: dict[str, RolloutSession] = {} + self.state_service_secrets = frozenset(state_service_secrets) self.config = config or InterceptionServerConfig() self.tunnel: Tunnel | None = ( make_tunnel(self.config.tunnel) if requires_tunnel else None @@ -143,16 +147,20 @@ def load(self) -> int: """Rollouts currently registered — what the pools balance on.""" return len(self.sessions) - def register(self, session: RolloutSession) -> str: - """Add a session under a fresh secret (the bearer token the harness must send) and - return it.""" - secret = secrets.token_urlsafe(16) - self.sessions[secret] = session - return secret - - def unregister(self, secret: str) -> None: - session = self.sessions.pop(secret, None) + def register(self, session: RolloutSession) -> tuple[str, str]: + """Register separate capabilities for model inference and private task state.""" + model_secret = secrets.token_urlsafe(16) + state_secret = secrets.token_urlsafe(16) + self.sessions[model_secret] = session + self.state_sessions[state_secret] = session + self.state_routes[session.trace.id] = session + return model_secret, state_secret + + def unregister(self, model_secret: str, state_secret: str) -> None: + session = self.sessions.pop(model_secret, None) + self.state_sessions.pop(state_secret, None) if session is not None: + self.state_routes.pop(session.trace.id, None) # The rollout concluded; its trace is sealed. Cancel straggler handlers # (aiohttp keeps them alive past client death) so a slow upstream call # can't commit a late turn onto the concluded trace. @@ -160,11 +168,11 @@ def unregister(self, secret: str) -> None: @asynccontextmanager async def acquire(self, session: RolloutSession) -> AsyncIterator[Slot]: - secret = self.register(session) + model_secret, state_secret = self.register(session) try: - yield self.base_url, secret + yield self.base_url, model_secret, state_secret finally: - self.unregister(secret) + self.unregister(model_secret, state_secret) def _handler_for(self, dialect: Dialect): """Bind a route's dialect to the request handler — the route the SDK posts to is what @@ -188,12 +196,11 @@ async def start(self) -> None: app.router.add_post(route, self._handler_for(dialect)) for aux in dialect.aux_routes: app.router.add_post(aux, self._aux_handler_for(dialect, aux)) - # The shared-state back-channel (see `verifiers.v1.state`): a rollout's tool servers - # GET/PUT their `self.state` here, keyed by the same bearer secret as the model routes. + # Tool servers use a state-only capability; the model bearer cannot reach these. app.router.add_get("/state", self.handle_state_get) app.router.add_put("/state", self.handle_state_put) # A launched tool server fetches its rollout's task here to run `setup_task` — the task - # is never passed via env, only over this channel, keyed by the same bearer secret. + # is never passed via env, only over this channel, keyed by the state bearer. app.router.add_get("/task", self.handle_task_get) self.runner = web.AppRunner(app) await self.runner.setup() @@ -705,12 +712,17 @@ async def handle_aux( return web.json_response(dialect.error_body(str(e)), status=502) return web.json_response(result) - def _session_for(self, request: web.Request) -> RolloutSession | None: - """The session a state request belongs to, by its `Authorization: Bearer ` — the - same per-rollout secret the model routes use (dialect-independent, so parsed directly).""" + def _session_for( + self, request: web.Request, *, allow_service: bool = False + ) -> RolloutSession | None: + """Resolve a private state bearer, or a trusted shared server plus route id.""" auth = request.headers.get("Authorization", "") secret = auth[len("Bearer ") :] if auth.startswith("Bearer ") else "" - session = self.sessions.get(secret) + session = self.state_sessions.get(secret) + if session is None and allow_service and secret in self.state_service_secrets: + session = self.state_routes.get( + request.headers.get("X-Verifiers-State-Route", "") + ) if session is not None: # state writes must not land on a sealed trace either session.adopt(asyncio.current_task()) return session @@ -718,7 +730,7 @@ def _session_for(self, request: web.Request) -> RolloutSession | None: async def handle_state_get(self, request: web.Request) -> web.Response: """Hand a rollout's tool server the current shared `trace.state` (it pulls before each `@vf.tool` call, so it sees writes from the other servers).""" - session = self._session_for(request) + session = self._session_for(request, allow_service=True) if session is None: return web.json_response({"error": "unauthorized"}, status=401) logger.debug("intercept GET /state: id=%s", session.trace.id) @@ -732,7 +744,7 @@ async def handle_state_get(self, request: web.Request) -> web.Response: async def handle_task_get(self, request: web.Request) -> web.Response: """Hand a launched tool server the rollout's task (class ref + JSON) so it can run - `setup_task` for this rollout — keyed by the same bearer secret as the state channel.""" + `setup_task` for this rollout — keyed by its private state bearer.""" session = self._session_for(request) if session is None: return web.json_response({"error": "unauthorized"}, status=401) @@ -749,7 +761,7 @@ async def handle_state_put(self, request: web.Request) -> web.Response: """Replace a rollout's shared `trace.state` with a server's pushed copy (validated into the trace's `State` type). Last write wins per call. A task ends the trajectory from state via its own `@stop` (run in `RolloutSession.refused` before each model call).""" - session = self._session_for(request) + session = self._session_for(request, allow_service=True) if session is None: return web.json_response({"error": "unauthorized"}, status=401) logger.debug("intercept PUT /state: id=%s", session.trace.id) diff --git a/verifiers/v1/mcp/launch.py b/verifiers/v1/mcp/launch.py index 7d92a31f2..e4e72dbba 100644 --- a/verifiers/v1/mcp/launch.py +++ b/verifiers/v1/mcp/launch.py @@ -5,12 +5,13 @@ import importlib.metadata import io import logging +import secrets import shlex import sys import tarfile import uuid from collections.abc import AsyncIterator -from dataclasses import dataclass +from dataclasses import dataclass, field from functools import cache from pathlib import Path from typing import TYPE_CHECKING @@ -18,7 +19,13 @@ from verifiers.v1.errors import ToolsetError from verifiers.v1.interception.tunnel import PrimeTunnel -from verifiers.v1.mcp.server import STATE_SECRET_PARAM, STATE_URL_PARAM, ServerBase +from verifiers.v1.mcp.server import ( + STATE_ROUTE_PARAM, + STATE_SIGNATURE_PARAM, + STATE_URL_PARAM, + ServerBase, + state_signature, +) from verifiers.v1.runtimes import ( NetworkPolicyConfig, Runtime, @@ -26,6 +33,7 @@ runtime_is_local, ) from verifiers.v1.runtimes.base import _ENSURE_UV +from verifiers.v1.state import State if TYPE_CHECKING: from verifiers.v1.mcp.toolset import Toolset @@ -170,14 +178,17 @@ async def serve_in_runtime( the OS choose and report the result through a file. With a state channel, the server fetches the current rollout task from the adjacent `/task` endpoint rather than a launch argument. """ - env = {"VF_CONFIG": server.config.model_dump_json()} + # A shared server has a private service secret but no fixed state URL. Set + # both controls explicitly so a subprocess cannot inherit stale host values. + env = { + "VF_CONFIG": server.config.model_dump_json(), + "VF_STATE_URL": state_url or "", + "VF_STATE_SECRET": state_secret, + } if runtime.type == "subprocess": # Keep provider temp files in the runtime workdir so cleanup removes them. assert runtime.info.id is not None env["TMPDIR"] = runtime.info.id - if state_url: - env["VF_STATE_URL"] = state_url - env["VF_STATE_SECRET"] = state_secret if runtime.published_port is not None: env["MCP_HOST"] = "0.0.0.0" fixed = runtime.published_port if exposed else None @@ -240,8 +251,14 @@ async def reachable_url( yield url +@dataclass(frozen=True) +class _ServedServer: + url: str + runtime: Runtime + + @contextlib.asynccontextmanager -async def serve( +async def _serve( server: ServerBase, harness_runtime: Runtime | None = None, harness_is_local: bool = True, @@ -307,7 +324,27 @@ async def serve( base = base.replace("127.0.0.1", "localhost", 1) elif not colocated and harness_runtime is not None: base = harness_runtime.host_url(base) - yield f"{base.rstrip('/')}/mcp" + yield _ServedServer(f"{base.rstrip('/')}/mcp", runtime) + + +@contextlib.asynccontextmanager +async def serve( + server: ServerBase, + harness_runtime: Runtime | None = None, + harness_is_local: bool = True, + *, + state_secret: str = "", + state_base: str | None = None, +): + """Serve one MCP server and yield the URL visible to its consumer.""" + async with _serve( + server, + harness_runtime, + harness_is_local, + state_secret=state_secret, + state_base=state_base, + ) as served: + yield served.url @dataclass(frozen=True) @@ -323,6 +360,8 @@ class SharedToolServer: url: str local: bool external: bool = False + runtime: Runtime | None = field(default=None, repr=False) + state_secret: str = field(default="", repr=False) @contextlib.asynccontextmanager @@ -359,24 +398,45 @@ async def serve_shared(toolsets: list[Toolset], harness_is_local: bool = True): url=cfg.url, local=False, external=True ) else: - url = await stack.enter_async_context( - serve(toolset, harness_is_local=harness_is_local) + state_secret = ( + secrets.token_urlsafe(24) if toolset._state_cls is not State else "" + ) + served = await stack.enter_async_context( + _serve( + toolset, + harness_is_local=harness_is_local, + state_secret=state_secret, + ) ) servers[name] = SharedToolServer( - url=url, local=runtime_is_local(cfg.runtime) + url=served.url, + local=runtime_is_local(cfg.runtime), + runtime=served.runtime, + state_secret=state_secret, ) logger.info("shared tool server '%s': %s", name, servers[name].url) yield servers -def _shared_url_for_rollout(url: str, state_base: str | None, state_secret: str) -> str: - """Attach one rollout's state bridge to a shared server URL.""" - if not state_base: - return url - parts = urlsplit(url) +def _shared_url_for_rollout( + server: SharedToolServer, + visible_url: str, + state_base: str | None, + state_route: str, +) -> str: + """Attach signed state coordinates; the shared server keeps its bearer private.""" + if not state_base or not server.state_secret: + return visible_url + state_url = f"{state_base.rstrip('/')}/state" + if server.runtime is not None: + state_url = server.runtime.host_url(state_url) + parts = urlsplit(visible_url) query = dict(parse_qsl(parts.query)) - query[STATE_URL_PARAM] = f"{state_base.rstrip('/')}/state" - query[STATE_SECRET_PARAM] = state_secret + query[STATE_URL_PARAM] = state_url + query[STATE_ROUTE_PARAM] = state_route + query[STATE_SIGNATURE_PARAM] = state_signature( + server.state_secret, state_url, state_route + ) return urlunsplit(parts._replace(query=urlencode(query))) @@ -387,6 +447,7 @@ async def serve_tools( shared: dict[str, SharedToolServer] | None = None, *, state_secret: str = "", + state_route: str = "", state_base: str | None = None, ): """Bring up a rollout's tool servers and yield `{name: url}` the harness reaches: the @@ -394,9 +455,9 @@ async def serve_tools( server fetches its task over the interception `/task` channel), and the taskset-scoped `shared` servers — already running eval-level (see `serve_shared`) — join under their per-rollout state tag. - `state_base`/`state_secret` wire each server to the interception server's shared-state - channel — `state_base` is universally reachable, so every server (per-rollout or - `shared`, any runtime) uses it directly.""" + `state_secret` is private to task-scoped servers; shared servers keep an + eval-level service secret and receive only signed `state_route` coordinates. + `state_base` is universally reachable from either placement.""" urls: dict[str, str] = {} async with contextlib.AsyncExitStack() as stack: for name, server in (shared or {}).items(): @@ -408,8 +469,7 @@ async def serve_tools( logger.info("tool server '%s' (shared, external): %s", name, server.url) continue url = harness_runtime.host_url(server.url) if server.local else server.url - urls[name] = _shared_url_for_rollout(url, state_base, state_secret) - # The tagged URL contains the bearer secret; log only the untagged base URL. + urls[name] = _shared_url_for_rollout(server, url, state_base, state_route) logger.info("tool server '%s' (shared): %s", name, server.url) for toolset in toolsets: name = toolset.server_name diff --git a/verifiers/v1/mcp/server.py b/verifiers/v1/mcp/server.py index 5c35c6fce..958d95ee5 100644 --- a/verifiers/v1/mcp/server.py +++ b/verifiers/v1/mcp/server.py @@ -3,6 +3,8 @@ import contextlib import contextvars import functools +import hashlib +import hmac import inspect import logging import os @@ -32,6 +34,7 @@ async def _channel_request( url: str, secret: str, *, + route: str | None = None, content: bytes | None = None, client: AsyncClient | None = None, ) -> Response: @@ -57,6 +60,8 @@ async def request() -> Response: ) async with manager as request_client: headers = {"Authorization": f"Bearer {secret}"} + if route: + headers["X-Verifiers-State-Route"] = route if content is not None: headers["Content-Type"] = "application/json" resp = await request_client.request( @@ -76,7 +81,15 @@ async def request() -> Response: # Shared servers receive the calling rollout's state channel in URL parameters because one # process cannot carry a single rollout's channel in its environment. STATE_URL_PARAM = "vf_state_url" -STATE_SECRET_PARAM = "vf_state_secret" +STATE_ROUTE_PARAM = "vf_state_route" +STATE_SIGNATURE_PARAM = "vf_state_signature" + + +def state_signature(secret: str, url: str, route: str) -> str: + """Authenticate one exact shared-server state destination.""" + message = f"vf-state-v1\0{url}\0{route}".encode() + return hmac.new(secret.encode(), message, hashlib.sha256).hexdigest() + # A context variable isolates the state seen by concurrent calls on one shared server. _call_state: contextvars.ContextVar[State | None] = contextvars.ContextVar( @@ -91,7 +104,12 @@ def _request_query(name: str) -> str | None: request = request_ctx.get().request except LookupError: return None - return request.query_params.get(name) if request is not None else None + if request is None: + return None + values = request.query_params.getlist(name) + if len(values) > 1: + raise ValueError(f"duplicate {name!r} state coordinate") + return values[0] if values else None def _die_with_parent() -> None: @@ -135,19 +153,32 @@ def state(self) -> StateT: current = _call_state.get() return current if current is not None else self._inert_state # type: ignore[return-value] - def _state_channel(self) -> tuple[str | None, str]: - """Prefer per-call coordinates for shared servers, then the process environment.""" - url = _request_query(STATE_URL_PARAM) or os.environ.get("VF_STATE_URL") - secret = _request_query(STATE_SECRET_PARAM) or os.environ.get( - "VF_STATE_SECRET", "" - ) - return url, secret + def _state_channel(self) -> tuple[str | None, str, str | None]: + """Use an env channel, or authenticate the exact coordinates of a shared call.""" + url = os.environ.get("VF_STATE_URL") + secret = os.environ.get("VF_STATE_SECRET", "") + if url: + return url, secret, None + shared_url = _request_query(STATE_URL_PARAM) + route = _request_query(STATE_ROUTE_PARAM) + signature = _request_query(STATE_SIGNATURE_PARAM) + if not any((shared_url, route, signature)): + return None, "", None + if not secret or not shared_url or not route or not signature: + raise ValueError("invalid shared state coordinates") + if not hmac.compare_digest( + signature, state_signature(secret, shared_url, route) + ): + raise ValueError("invalid shared state coordinates") + return shared_url, secret, route async def _pull_state(self) -> State: - url, secret = self._state_channel() + url, secret, route = self._state_channel() if not url: return self._state_cls() - response = await _channel_request("GET", url, secret, client=self._state_client) + response = await _channel_request( + "GET", url, secret, route=route, client=self._state_client + ) try: return self._state_adapter.validate_json(response.content) except ValidationError as e: @@ -157,7 +188,7 @@ async def _pull_state(self) -> State: raise async def _push_state(self, before: bytes) -> None: - url, secret = self._state_channel() + url, secret, route = self._state_channel() if not url: return state = _call_state.get() @@ -166,7 +197,12 @@ async def _push_state(self, before: bytes) -> None: if after == before: return await _channel_request( - "PUT", url, secret, content=after, client=self._state_client + "PUT", + url, + secret, + route=route, + content=after, + client=self._state_client, ) async def _fetch_task(self, state_url: str | None, secret: str): @@ -263,7 +299,8 @@ async def serve() -> None: ): self._state_client = client await self.setup() - await self._setup_task_from_channel(*self._state_channel()) + state_url, secret, _ = self._state_channel() + await self._setup_task_from_channel(state_url, secret) # These servers are reached through localhost or a tunnel, never a browser. security = TransportSecuritySettings( enable_dns_rebinding_protection=False diff --git a/verifiers/v1/rollout.py b/verifiers/v1/rollout.py index 8db36ccda..19d0ac953 100644 --- a/verifiers/v1/rollout.py +++ b/verifiers/v1/rollout.py @@ -85,7 +85,12 @@ async def _serve_interception( [server.config for server in servers], shared_tools.values(), ) - server = InterceptionServer(requires_tunnel=tunneled) + server = InterceptionServer( + requires_tunnel=tunneled, + state_service_secrets=tuple( + tool.state_secret for tool in shared_tools.values() if tool.state_secret + ), + ) async with server, server.acquire(session) as slot: yield slot @@ -265,7 +270,11 @@ async def open(self) -> bool: # The harness reaches the model at `{base_url}/v1`; tool servers reach this # rollout's `/state` + `/task` at `base_url` — it's universally reachable # (the interception is exposed whenever any consumer is remote). - base_url, secret = await self._stack.enter_async_context( + ( + base_url, + model_secret, + state_secret, + ) = await self._stack.enter_async_context( _serve_interception( self._interception, runtime, @@ -275,13 +284,14 @@ async def open(self) -> bool: ) ) self._endpoint = f"{runtime.host_url(base_url)}/v1" - self._secret = secret + self._secret = model_secret self._urls = await self._stack.enter_async_context( serve_tools( tool_servers, runtime, shared=self._shared_tools, - state_secret=secret, + state_secret=state_secret, + state_route=self.trace.id, state_base=base_url, ) )