Harden Gateway runtime identity boundaries

ax_cli/commands/gateway.py

@@ -55,10 +55,12 @@
     evaluate_runtime_attestation,
     find_agent_entry,
     gateway_dir,
+    gateway_environment,
     get_gateway_approval,
     hermes_setup_status,
     infer_asset_descriptor,
     list_gateway_approvals,
+    load_gateway_managed_agent_token,
     load_gateway_registry,
     load_gateway_session,
     load_recent_gateway_activity,
@@ -172,13 +174,10 @@ def _load_managed_agent_or_exit(name: str) -> dict:
 
 
 def _load_managed_agent_client(entry: dict) -> AxClient:
-    token_file = Path(str(entry.get("token_file") or "")).expanduser()
-    if not token_file.exists():
-        err_console.print(f"[red]Managed agent token is missing:[/red] {token_file}")
-        raise typer.Exit(1)
-    token = token_file.read_text().strip()
-    if not token:
-        err_console.print(f"[red]Managed agent token file is empty:[/red] {token_file}")
+    try:
+        token = load_gateway_managed_agent_token(entry)
+    except ValueError as exc:
+        err_console.print(f"[red]{exc}[/red]")
         raise typer.Exit(1)
     return AxClient(
         base_url=str(entry.get("base_url") or ""),
@@ -650,6 +649,7 @@ def _status_payload(*, activity_limit: int = 10) -> dict:
         gateway["pid"] = None
     payload = {
         "gateway_dir": str(gateway_dir()),
+        "gateway_environment": gateway_environment(),
         "connected": bool(session),
         "base_url": session.get("base_url") if session else None,
         "space_id": session.get("space_id") if session else None,
@@ -1401,7 +1401,8 @@ def _render_gateway_overview(payload: dict) -> Panel:
     )
     grid.add_row("User", str(payload.get("user") or "-"), "Base URL", str(payload.get("base_url") or "-"))
     space_label = str(payload.get("space_name") or payload.get("space_id") or "-")
-    grid.add_row("Space", space_label, "PID", str(payload["daemon"].get("pid") or "-"))
+    grid.add_row("Space", space_label, "Environment", str(payload.get("gateway_environment") or "default"))
+    grid.add_row("PID", str(payload["daemon"].get("pid") or "-"), "State Dir", str(payload.get("gateway_dir") or "-"))
     grid.add_row("UI", str(ui.get("url") or "-"), "UI PID", str(ui.get("pid") or "-"))
     grid.add_row(
         "Session",

ax_cli/gateway.py

@@ -1389,12 +1389,12 @@ def _normalize_allowed_spaces_payload(payload: object) -> list[dict[str, Any]]:
 
 
 def _fetch_allowed_spaces_for_entry(entry: dict[str, Any]) -> list[dict[str, Any]] | None:
-    token_file = Path(str(entry.get("token_file") or "")).expanduser()
     base_url = _normalized_base_url(entry.get("base_url"))
-    if not token_file.exists() or not base_url:
+    if not base_url:
         return None
-    token = token_file.read_text().strip()
-    if not token:
+    try:
+        token = load_gateway_managed_agent_token(entry)
+    except ValueError:
         return None
     client = AxClient(
         base_url=base_url,
@@ -2189,12 +2189,32 @@ def annotate_runtime_health(
 
 
 def gateway_dir() -> Path:
-    path = _global_config_dir() / "gateway"
+    explicit = str(os.environ.get("AX_GATEWAY_DIR") or "").strip()
+    if explicit:
+        path = Path(explicit).expanduser()
+    else:
+        root = _global_config_dir() / "gateway"
+        env_name = gateway_environment()
+        path = root if env_name is None else root / "envs" / env_name
     path.mkdir(parents=True, exist_ok=True)
     path.chmod(0o700)
     return path
 
 
+def gateway_environment() -> str | None:
+    raw = (
+        str(os.environ.get("AX_GATEWAY_ENV") or "").strip()
+        or str(os.environ.get("AX_USER_ENV") or "").strip()
+        or str(os.environ.get("AX_ENV") or "").strip()
+    )
+    if not raw:
+        return None
+    normalized = re.sub(r"[^a-z0-9_.-]+", "-", raw.lower()).strip(".-")
+    if not normalized or normalized in {"default", "user"}:
+        return None
+    return normalized
+
+
 def gateway_agents_dir() -> Path:
     path = gateway_dir() / "agents"
     path.mkdir(parents=True, exist_ok=True)
@@ -2241,6 +2261,24 @@ def agent_token_path(name: str) -> Path:
     return agent_dir(name) / "token"
 
 
+def load_gateway_managed_agent_token(entry: dict[str, Any]) -> str:
+    """Read a Gateway-managed runtime token and reject bootstrap credentials."""
+    token_file = Path(str(entry.get("token_file") or "")).expanduser()
+    if not token_file.exists():
+        raise ValueError(f"Gateway-managed token file is missing: {token_file}")
+    token = token_file.read_text().strip()
+    if not token:
+        raise ValueError(f"Gateway-managed token file is empty: {token_file}")
+    if token.startswith("axp_u_"):
+        raise ValueError(
+            "Gateway-managed agents require an agent-bound token. "
+            f"Refusing to use a user bootstrap PAT from {token_file}."
+        )
+    if not str(entry.get("agent_id") or "").strip():
+        raise ValueError("Gateway-managed agents require a bound agent_id before runtime use.")
+    return token
+
+
 def agent_pending_queue_path(name: str) -> Path:
     return agent_dir(name) / "pending.json"
 
@@ -2394,6 +2432,8 @@ def daemon_status() -> dict[str, Any]:
     return {
         "pid": pid,
         "running": running,
+        "gateway_dir": str(gateway_dir()),
+        "gateway_environment": gateway_environment(),
         "registry_path": str(registry_path()),
         "session_path": str(session_path()),
         "registry": registry,
@@ -2896,8 +2936,7 @@ def _build_hermes_sentinel_cmd(entry: dict[str, Any]) -> list[str]:
 
 def _build_hermes_sentinel_env(entry: dict[str, Any]) -> dict[str, str]:
     env = {k: v for k, v in os.environ.items() if k not in ENV_DENYLIST}
-    token_file = Path(str(entry.get("token_file") or "")).expanduser()
-    token = token_file.read_text().strip() if token_file.exists() else ""
+    token = load_gateway_managed_agent_token(entry)
     workdir = _hermes_sentinel_workdir(entry)
     agents_dir = _agents_dir_for_entry(entry)
     hermes_repo = str(entry.get("hermes_repo_path") or "").strip() or "/home/ax-agent/shared/repos/hermes-agent"
@@ -3149,7 +3188,7 @@ def _log(self, message: str) -> None:
         self.logger(f"{self.name}: {message}")
 
     def _token(self) -> str:
-        return self.token_file.read_text().strip()
+        return load_gateway_managed_agent_token(self.entry)
 
     def _new_client(self):
         return self.client_factory(
@@ -3313,16 +3352,17 @@ def _hermes_sentinel_log_path(self) -> Path:
     def _start_hermes_sentinel_process(self, *, runtime_instance_id: str) -> None:
         workdir = _hermes_sentinel_workdir(self.entry)
         script = _hermes_sentinel_script(self.entry)
-        token_file = Path(str(self.entry.get("token_file") or "")).expanduser()
         if not script.exists():
             error = f"Hermes sentinel script not found: {script}"
             self._update_state(
                 effective_state="error", current_status="error", current_activity=error, last_error=error
             )
             record_gateway_activity("runtime_error", entry=self.entry, error=error)
             return
-        if not token_file.exists() or not token_file.read_text().strip():
-            error = f"Gateway-managed token file is missing or empty: {token_file}"
+        try:
+            load_gateway_managed_agent_token(self.entry)
+        except ValueError as exc:
+            error = str(exc)
             self._update_state(
                 effective_state="error", current_status="error", current_activity=error, last_error=error
             )

docs/gateway-agent-runtimes.md

@@ -22,6 +22,12 @@ Gateway keeps those pieces, but moves operator management into one place:
 - Show liveness, queue state, activity, and tool signals.
 - Provide a single CLI/UI for dev, staging, and production operators.
 
+Use separate Gateway state per environment. `AX_GATEWAY_ENV=dev/staging` stores
+state under `~/.ax/gateway/envs/dev-staging`, while `AX_GATEWAY_ENV=prod`
+stores a separate registry, session, PID file, UI state, queues, and agent token
+files. `AX_GATEWAY_DIR=/path/to/gateway-state` is available when a deployment
+needs an explicit state root.
+
 ## Current PR88 State
 
 PR88 has enough Gateway plumbing to register agents, mint tokens, show status,
@@ -80,6 +86,10 @@ owns the Hermes session, runtime plugin, message queue, and tool callbacks. The
 Gateway owns the credentials, process lifecycle, binding verification, and
 operator status.
 
+Runtime token files must contain an agent-bound credential for the managed
+agent. Gateway rejects user bootstrap PATs before sends or runtime launch so a
+copied user token cannot become an agent runtime identity.
+
 Do not treat the one-shot `examples/hermes_sentinel/hermes_bridge.py` demo as
 the production sentinel pattern. It is useful for proving that a Gateway command
 bridge can call Hermes, but it creates a fresh agent per message and does not