fix(billing): bind RPC profile scope and honor scoped Portal routing

hermes_cli/nous_account.py

@@ -717,7 +717,9 @@ def resolve_nous_portal_base_url() -> str:
 
 
 def _portal_base_url(state: dict[str, Any]) -> Optional[str]:
-    value = state.get("portal_base_url")
+    from hermes_cli.auth_nous import _nous_portal_env_override
+
+    value = _nous_portal_env_override() or state.get("portal_base_url")
     return value.strip().rstrip("/") if _nonblank(value) else None
 
 

hermes_cli/nous_billing.py

@@ -99,7 +99,9 @@ class BillingUpgradeCapExceeded(BillingTransient):
 
 def resolve_portal_base_url(state: Optional[dict[str, Any]] = None) -> str:
     """Resolve the portal base URL with login-time precedence: env, stored state, default."""
-    env = os.getenv("HERMES_PORTAL_BASE_URL") or os.getenv("NOUS_PORTAL_BASE_URL")
+    from hermes_cli.auth_nous import _nous_portal_env_override
+
+    env = _nous_portal_env_override()
     for candidate in (env, state.get("portal_base_url") if state else None):
         if isinstance(candidate, str) and candidate.strip():
             return candidate.strip().rstrip("/")

tests/tui_gateway/test_billing_profile_scope.py

@@ -0,0 +1,212 @@
+"""Regression for #132728: registered Portal RPCs bind the selected profile.
+
+Exercise real RPC registration, worker dispatch, OAuth refresh, and endpoint
+resolution against two synthetic homes. Only external HTTP is substituted.
+"""
+
+import base64
+import io
+import json
+import logging
+import os
+import threading
+import time
+from pathlib import Path
+from types import SimpleNamespace
+
+import httpx
+import pytest
+
+
+READS = ("billing.state", "subscription.state", "usage.bars")
+
+
+def _jwt(label):
+    def part(value):
+        return base64.urlsafe_b64encode(json.dumps(value).encode()).decode().rstrip("=")
+    return ".".join((part({"alg": "none"}), part({
+        "sub": label, "scope": "inference:invoke", "exp": int(time.time()) + 3600,
+    }), "fixture"))
+
+
+def _home(home, label, *, expired):
+    home.mkdir(parents=True)
+    (home / "config.yaml").write_text("model:\n  provider: nous\n  default: fixture-model\n")
+    (home / ".env").write_text(
+        f"HERMES_PORTAL_BASE_URL=https://{label}-portal.invalid\n"
+        f"NOUS_INFERENCE_BASE_URL=https://{label}-inference.invalid/v1\n"
+    )
+    (home / "auth.json").write_text(json.dumps({"version": 1, "providers": {"nous": {
+        "access_token": _jwt(label), "refresh_token": f"{label}-refresh",
+        "expires_at": "2000-01-01T00:00:00+00:00" if expired else "2999-01-01T00:00:00+00:00",
+        "scope": "inference:invoke", "client_id": "fixture-client",
+        "portal_base_url": "https://portal.nousresearch.com",
+        "inference_base_url": "https://inference-api.nousresearch.com/v1",
+    }}}))
+
+
+@pytest.fixture(params=[False, True], ids=["valid-token", "expired-token"])
+def runtime(request, tmp_path, monkeypatch):
+    launch = tmp_path / "launch-home"
+    secondary = launch / "profiles" / "secondary"
+    _home(launch, "launch", expired=request.param)
+    _home(secondary, "secondary", expired=request.param)
+    monkeypatch.setenv("HERMES_HOME", str(launch))
+    monkeypatch.setenv("HERMES_SHARED_AUTH_DIR", str(tmp_path / "shared-auth"))
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+
+    from agent import billing_usage, billing_view, secret_scope, subscription_view
+    from hermes_cli import auth, nous_account, nous_billing
+    from tui_gateway import launch_profile_policy, server
+
+    for key in ("HERMES_PORTAL_BASE_URL", "NOUS_PORTAL_BASE_URL", "NOUS_INFERENCE_BASE_URL",
+                "HERMES_DEV_BILLING_FIXTURE", "HERMES_DEV_SUBSCRIPTION_FIXTURE", "HERMES_DEV_CREDITS_FIXTURE"):
+        monkeypatch.delenv(key, raising=False)
+    monkeypatch.setattr(launch_profile_policy, "_snapshot", None)
+    monkeypatch.setattr(server, "_hermes_home", str(launch))
+    monkeypatch.setattr(server, "_served_profile_homes", set())
+    monkeypatch.setattr(auth, "_RESOLVE_TOKEN_CACHE", {})
+    monkeypatch.setattr(nous_billing, "_token_cache", {})
+    nous_account.reset_nous_portal_account_info_cache()
+    previous = secret_scope.is_multiplex_active()
+    launch_profile_policy.activate_multi_profile_hosting()
+    calls = []
+
+    def http_edge(request):
+        calls.append((request.method, str(request.url), request.headers.get("authorization")))
+        label = "secondary" if request.url.host == "secondary-portal.invalid" else "launch"
+        if request.url.path.endswith("/token"):
+            return httpx.Response(200, json={
+                # Nous deliberately mirrors refreshed OAuth tokens across profiles.
+                "access_token": _jwt("shared-account"), "refresh_token": "shared-rotated",
+                "expires_in": 3600, "scope": "inference:invoke",
+                "inference_base_url": f"https://{label}-inference.invalid/v1",
+            })
+        return httpx.Response(200, json={"paid_service_access": True})
+
+    real_client = httpx.Client
+    def client(*args, **kwargs):
+        kwargs["transport"] = httpx.MockTransport(http_edge)
+        return real_client(*args, **kwargs)
+    monkeypatch.setattr(httpx, "Client", client)
+
+    def urlopen(request, *args, **kwargs):
+        calls.append((request.get_method(), request.full_url, request.get_header("Authorization")))
+        response = io.BytesIO(json.dumps({
+            "ok": True, "balanceUsd": "0", "role": "MEMBER", "effect": "no_op",
+            "chargeId": "fixture-charge", "status": "settled", "amountUsd": "1",
+        }).encode())
+        response.status = 200
+        return response
+    monkeypatch.setattr("urllib.request.urlopen", urlopen)
+    try:
+        yield SimpleNamespace(server=server, homes={"launch": launch, "secondary": secondary},
+                              calls=calls, expired=request.param)
+    finally:
+        secret_scope.set_multiplex_active(previous)
+
+
+class Wire:
+    def __init__(self):
+        self.event = threading.Event()
+        self.response = None
+
+    def write(self, response):
+        self.response = response
+        self.event.set()
+        return True
+
+
+def _call(server, method, params):
+    wire = Wire()
+    response = server.dispatch({"jsonrpc": "2.0", "id": "probe", "method": method, "params": params}, wire)
+    if response is None:
+        assert wire.event.wait(10), "registered RPC did not return"
+        response = wire.response
+    assert "result" in response, response
+    return response["result"]
+
+
+@pytest.mark.parametrize("method", READS)
+@pytest.mark.parametrize("profile", ["launch", "secondary"])
+def test_reads_keep_scope_and_route_every_request(runtime, caplog, monkeypatch, method, profile):
+    params = {} if profile == "launch" else {"profile": "secondary"}
+    # A sibling's later ambient edit must not override the frozen launch scope or the selected profile.
+    monkeypatch.setenv("HERMES_PORTAL_BASE_URL", "https://ambient-other.invalid")
+    monkeypatch.setenv("NOUS_INFERENCE_BASE_URL", "https://ambient-other-inference.invalid/v1")
+    ambient = dict(os.environ)
+    with caplog.at_level(logging.WARNING):
+        result = _call(runtime.server, method, params)
+    assert result.get("logged_in", result.get("available")) is True, result
+    warnings = [r.getMessage() for r in caplog.records if "no profile secret scope" in r.getMessage()]
+    assert not warnings, warnings
+    assert runtime.calls
+    assert all(url.startswith(f"https://{profile}-portal.invalid/") for _, url, _ in runtime.calls), runtime.calls
+    assert any(verb == "POST" for verb, _, _ in runtime.calls) is runtime.expired, runtime.calls
+    assert dict(os.environ) == ambient
+
+
+@pytest.mark.parametrize("method", READS)
+def test_reads_keep_profile_routing_when_switching_homes(runtime, caplog, method):
+    with caplog.at_level(logging.WARNING):
+        for profile in ("launch", "secondary", "launch"):
+            runtime.calls.clear()
+            result = _call(runtime.server, method, {} if profile == "launch" else {"profile": profile})
+            assert result.get("logged_in", result.get("available")) is True, result
+            # The third call can be answered from the same-profile account cache.
+            for verb, url, bearer in runtime.calls:
+                assert url.startswith(f"https://{profile}-portal.invalid/"), runtime.calls
+                if verb == "GET":
+                    assert bearer and bearer.startswith("Bearer "), runtime.calls
+                    payload = bearer.removeprefix("Bearer ").split(".")[1]
+                    claims = json.loads(base64.urlsafe_b64decode(payload + "=" * (-len(payload) % 4)))
+                    assert claims["sub"] == ("shared-account" if runtime.expired else profile), runtime.calls
+    assert not [r for r in caplog.records if "no profile secret scope" in r.getMessage()]
+
+
+@pytest.mark.parametrize("method,params", [
+    ("subscription.preview", {"subscription_type_id": "fixture-tier"}),
+    ("subscription.change", {"cancel": True}),
+    ("subscription.resume", {}),
+    ("subscription.upgrade", {"subscription_type_id": "fixture-tier"}),
+    ("billing.charge", {"amount_usd": "1"}),
+    ("billing.charge_status", {"charge_id": "fixture-charge"}),
+    ("billing.auto_reload", {"threshold": "1", "top_up_amount": "1"}),
+])
+@pytest.mark.parametrize("profile", ["launch", "secondary"])
+def test_sibling_routes_use_the_same_profile_scope(runtime, caplog, method, params, profile):
+    # These are HTTP-intercepted synthetic calls, including the money routes.
+    with caplog.at_level(logging.WARNING):
+        result = _call(runtime.server, method, params | ({} if profile == "launch" else {"profile": profile}))
+    assert result["ok"] is True, result
+    assert runtime.calls
+    assert all(url.startswith(f"https://{profile}-portal.invalid/") for _, url, _ in runtime.calls), runtime.calls
+    assert not [r for r in caplog.records if "no profile secret scope" in r.getMessage()]
+
+
+@pytest.mark.parametrize("profile", ["launch", "secondary"])
+def test_step_up_binds_the_profile_before_starting_the_external_flow(runtime, monkeypatch, profile):
+    from agent.secret_scope import get_secret
+    from hermes_cli import auth
+    from hermes_constants import get_hermes_home
+
+    seen = []
+    def external_device_flow(**kwargs):
+        seen.append((get_hermes_home(), get_secret("HERMES_PORTAL_BASE_URL")))
+        return True
+    monkeypatch.setattr(auth, "step_up_nous_billing_scope", external_device_flow)
+    result = _call(runtime.server, "billing.step_up", {} if profile == "launch" else {"profile": profile})
+    assert result == {"ok": True, "granted": True}, result
+    assert seen == [(runtime.homes[profile], f"https://{profile}-portal.invalid")]
+
+
+@pytest.mark.parametrize("method", READS)
+def test_missing_profile_fails_before_any_portal_request(runtime, method):
+    wire = Wire()
+    response = runtime.server.dispatch({"jsonrpc": "2.0", "id": "probe", "method": method,
+                                       "params": {"profile": "missing"}}, wire)
+    if response is None:
+        assert wire.event.wait(10)
+        response = wire.response
+    assert response["error"]["code"] == 4064, response
+    assert runtime.calls == []

tui_gateway/billing_view.py

@@ -20,6 +20,11 @@ def _wire_str(value):
     return None if value is None else str(value)
 
 
+def _billing_pick(result: dict, **fields) -> dict:
+    """``{"ok": True, <snake>: result[<camel>], ...}`` in ``fields`` order."""
+    return {"ok": True, **{key: result.get(src) for key, src in fields.items()}}
+
+
 def _serialize_billing_error(exc) -> dict:
     """Map a BillingError into the result.error envelope the TUI branches on."""
     from hermes_cli.nous_billing import (

tui_gateway/methods_session.py

@@ -222,11 +222,6 @@ def _billing_invalid(rid, message: str, error: str = "invalid_request") -> dict:
     return _ok(rid, {"ok": False, "error": error, "message": message})
 
 
-def _billing_pick(result: dict, **fields) -> dict:
-    """``{"ok": True, <snake>: result[<camel>], ...}`` in ``fields`` order."""
-    return {"ok": True, **{key: result.get(src) for key, src in fields.items()}}
-
-
 def _billing_pending_change(result: dict) -> dict:
     return {"ok": True, "message": result.get("message"), "payload": result}
 
@@ -1815,9 +1810,10 @@ def _(rid, params: dict) -> dict:
 # ``error`` (not a JSON-RPC error) so the TUI maps it to copy. ``billing:manage`` routes
 # return error=insufficient_scope on 403, which drives the ``billing.step_up`` device flow.
 def _billing_view(name: str, module: str, builder: str, serializer: str, fallback: dict) -> None:
-    """Read-only view RPC (no scope required): ``serializer(module.builder())``, ``fallback`` on any error.
+    """Profile-bound read-only view: ``serializer(module.builder())``, ``fallback`` on any error.
     The view module stays a lazy import (startup budget); the serializer is a server global."""
     @method(name)
+    @_profile_scoped
     def _(rid, params: dict) -> dict:
         try:
             from importlib import import_module
@@ -1827,8 +1823,9 @@ def _billing_view(name: str, module: str, builder: str, serializer: str, fallbac
 
 
 @method("billing.state")
+@_profile_scoped
 def _(rid, params: dict) -> dict:
-    """Read-only billing view (no scope required); fail-open. The Nous free tier has no account to
+    """Profile-bound read-only billing view; fail-open. The Nous free tier has no account to
     bill, so its state is answered locally (``free_tier_account`` set, ``logged_in`` false) without a portal
     round-trip that could only fail."""
     try:
@@ -1849,6 +1846,7 @@ _billing_view("subscription.state", "agent.subscription_view", "build_subscripti
 
 
 @method("subscription.preview")
+@_profile_scoped
 def _(rid, params: dict) -> dict:
     """POST /api/billing/subscription/preview → chargeless effect quote. billing:manage."""
     from agent.subscription_view import subscription_change_preview_from_payload
@@ -1865,6 +1863,7 @@ def _billing_route(name: str, call, *, invalid=None, message: str = "", error: s
     → ``_billing_invalid(message, error)``; ``call(nb, params, key)`` performs the request. ``idempotent``
     mints ``idempotency_key`` if absent and echoes it (also on error) so the TUI retries the SAME operation."""
     @method(name)
+    @_profile_scoped
     def _(rid, params: dict) -> dict:
         import hermes_cli.nous_billing as nb
         if invalid is not None and invalid(params):
@@ -1916,6 +1915,7 @@ _billing_route("billing.keep_going_on_credits", lambda nb, p, _k: _billing_pick(
 
 
 @method("billing.step_up")
+@_profile_scoped
 def _(rid, params: dict) -> dict:
     """billing:manage step-up device flow → {ok, granted} (false when the server downscopes). Pooled (blocks
     for minutes); URL/code reach the TUI via ``billing.step_up.verification`` (stdout is the RPC pipe) and the