diff --git a/evals/desktop_mcp_oauth/backend_http_fixture.py b/evals/desktop_mcp_oauth/backend_http_fixture.py
new file mode 100644
index 0000000000..a492efb85b
--- /dev/null
+++ b/evals/desktop_mcp_oauth/backend_http_fixture.py
@@ -0,0 +1,345 @@
+#!/usr/bin/env python3
+"""Credential-free live OAuth/MCP persistence probe (no production mocks).
+
+Run with the repository's Python environment:
+  python evals/desktop_mcp_oauth/backend_http_fixture.py --repo . --output receipt.json
+
+The parent creates an ephemeral HOME and re-execs with an environment allowlist.
+Only the redacted receipt survives; token files are verified then deleted with HOME.
+This exercises production session handlers, not Electron or gateway transport.
+"""
+from __future__ import annotations
+
+import argparse
+import asyncio
+import base64
+import contextlib
+import hashlib
+import json
+import logging
+import os
+from pathlib import Path
+import secrets
+import subprocess
+import sys
+import tempfile
+import threading
+import time
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from urllib.parse import parse_qs, urlencode, urlsplit
+
+
+class Provider(ThreadingHTTPServer):
+    def __init__(self):
+        super().__init__(("127.0.0.1", 0), ProviderHandler)
+        self.origin = f"http://127.0.0.1:{self.server_port}"
+        self.events = []
+        self.clients = {}
+        self.codes = {}
+        self.tokens = set()
+        self.pkce_verified = 0
+
+
+class ProviderHandler(BaseHTTPRequestHandler):
+    server: Provider
+
+    def log_message(self, format, *args):
+        pass  # Never log callback queries, codes or bearer headers.
+
+    def reply(self, status, payload=None, headers=None):
+        body = json.dumps(payload).encode() if payload is not None else b""
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(body)))
+        for key, value in (headers or {}).items():
+            self.send_header(key, value)
+        self.end_headers()
+        self.wfile.write(body)
+
+    def do_GET(self):  # noqa: N802
+        path = urlsplit(self.path).path
+        self.server.events.append(["GET", path])
+        if path.startswith("/.well-known/oauth-protected-resource"):
+            return self.reply(200, {"resource": self.server.origin + "/mcp",
+                                    "authorization_servers": [self.server.origin],
+                                    "scopes_supported": ["tools:read"]})
+        if path.startswith("/.well-known/oauth-authorization-server"):
+            return self.reply(200, {
+                "issuer": self.server.origin,
+                "authorization_endpoint": self.server.origin + "/authorize",
+                "token_endpoint": self.server.origin + "/token",
+                "registration_endpoint": self.server.origin + "/register",
+                "response_types_supported": ["code"],
+                "grant_types_supported": ["authorization_code", "refresh_token"],
+                "token_endpoint_auth_methods_supported": ["none"],
+                "code_challenge_methods_supported": ["S256"],
+                "scopes_supported": ["tools:read"],
+            })
+        if path == "/authorize":
+            query = {k: v[0] for k, v in parse_qs(urlsplit(self.path).query).items()}
+            client = self.server.clients.get(query.get("client_id"))
+            if (not client or query.get("redirect_uri") not in client["redirect_uris"]
+                    or query.get("code_challenge_method") != "S256"):
+                return self.reply(400, {"error": "invalid_request"})
+            code = secrets.token_urlsafe(24)
+            self.server.codes[code] = query
+            target = query["redirect_uri"] + "?" + urlencode({"code": code, "state": query["state"]})
+            return self.reply(302, headers={"Location": target})
+        self.reply(405 if path == "/mcp" else 404)
+
+    def do_POST(self):  # noqa: N802
+        path = urlsplit(self.path).path
+        self.server.events.append(["POST", path])
+        body = self.rfile.read(int(self.headers.get("Content-Length", 0)))
+        if path == "/register":
+            client = json.loads(body)
+            client["client_id"] = secrets.token_urlsafe(16)
+            client["token_endpoint_auth_method"] = "none"
+            self.server.clients[client["client_id"]] = client
+            return self.reply(201, client)
+        if path == "/token":
+            form = {k: v[0] for k, v in parse_qs(body.decode()).items()}
+            auth = self.server.codes.pop(form.get("code"), None)
+            challenge = base64.urlsafe_b64encode(hashlib.sha256(
+                form.get("code_verifier", "").encode()).digest()).rstrip(b"=").decode()
+            if (not auth or challenge != auth["code_challenge"]
+                    or form.get("redirect_uri") != auth["redirect_uri"]
+                    or form.get("client_id") != auth["client_id"]):
+                return self.reply(400, {"error": "invalid_grant"})
+            self.server.pkce_verified += 1
+            token = secrets.token_urlsafe(32)
+            self.server.tokens.add(token)
+            return self.reply(200, {"access_token": token, "token_type": "Bearer",
+                                    "expires_in": 3600, "scope": "tools:read",
+                                    "refresh_token": secrets.token_urlsafe(32)})
+        if path != "/mcp":
+            return self.reply(404)
+        if self.headers.get("Authorization", "").removeprefix("Bearer ") not in self.server.tokens:
+            return self.reply(401, {"error": "unauthorized"}, {"WWW-Authenticate":
+                f'Bearer resource_metadata="{self.server.origin}/.well-known/oauth-protected-resource"'})
+        request = json.loads(body)
+        method = request["method"]
+        self.server.events.append(["MCP", method])
+        if "id" not in request:
+            return self.reply(202)
+        results = {
+            "initialize": {"protocolVersion": request.get("params", {}).get("protocolVersion", "2025-03-26"),
+                           "capabilities": {"tools": {}},
+                           "serverInfo": {"name": "local-oauth-fixture", "version": "1"}},
+            "tools/list": {"tools": [{"name": "fixture_ping", "description": "Local fixture ping",
+                                       "inputSchema": {"type": "object", "properties": {}}}]},
+            "ping": {},
+        }
+        if method not in results:
+            return self.reply(200, {"jsonrpc": "2.0", "id": request["id"],
+                                    "error": {"code": -32601, "message": "Method not found"}})
+        self.reply(200, {"jsonrpc": "2.0", "id": request["id"], "result": results[method]})
+
+    def do_DELETE(self):  # noqa: N802
+        self.reply(200)
+
+
+class CallbackServer(ThreadingHTTPServer):
+    callback: dict[str, str]
+
+
+class CallbackHandler(BaseHTTPRequestHandler):
+    server: CallbackServer
+
+    def log_message(self, format, *args):
+        pass
+
+    def do_GET(self):  # noqa: N802
+        self.server.callback = {k: v[0] for k, v in parse_qs(urlsplit(self.path).query).items()}
+        self.send_response(200)
+        self.end_headers()
+        self.wfile.write(b"Local fixture callback captured")
+
+
+@contextlib.contextmanager
+def serving(server):
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    try:
+        yield server
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(5)
+
+
+def eventually(predicate, label, timeout=25):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        result = predicate()
+        if result:
+            return result
+        time.sleep(0.05)
+    raise AssertionError(label)
+
+
+def run_probe(repo, receipt):
+    sys.path.insert(0, str(repo))
+    logging.disable(logging.CRITICAL)
+    import httpx
+    from hermes_constants import set_hermes_home_override, reset_hermes_home_override
+    from tui_gateway import mcp_oauth_sessions as sessions
+    from tools.mcp_oauth import HermesTokenStorage
+
+    owner = Path(os.environ["HERMES_HOME"])
+    other = owner.parent / "other-profile"
+    other.mkdir()
+    checks = receipt["checks"]
+
+    def check(name, result):
+        checks[name] = bool(result)
+        if not result:
+            raise AssertionError(name)
+
+    with serving(Provider()) as provider, serving(CallbackServer(
+            ("127.0.0.1", 0), CallbackHandler)) as callback, httpx.Client(trust_env=False) as browser:
+        cfg = {"url": provider.origin + "/mcp", "auth": "oauth",
+               "oauth": {"cimd": False, "scope": "tools:read", "timeout": 30}}
+        redirect = f"http://127.0.0.1:{callback.server_port}/callback"
+
+        def begin(name, remote=True):
+            return sessions.start_flow(str(owner), name, cfg, url_timeout=25,
+                                       client_redirect_uri=redirect if remote else None)
+
+        def done(flow, name):
+            sid = flow["session_id"]
+            eventually(lambda: sessions._sessions[sid]["flow"].worker_done, "worker completed")
+            return sessions.poll_flow(sid, name)
+
+        flow = begin("positive")
+        sid = flow["session_id"]
+        check("remote_does_not_bind_backend_listener", sessions._sessions[sid]["httpd"] is None)
+        check("initial_poll_pending", sessions.poll_flow(sid, "positive")["status"] == "pending")
+        check("wrong_state_rejected", not sessions.deliver_callback_flow(
+            sid, "positive", code="invalid", state="invalid")["ok"])
+        check("wrong_state_keeps_pending", sessions.poll_flow(sid, "positive")["status"] == "pending")
+        check("wrong_server_poll_rejected", sessions.poll_flow(sid, "wrong")["status"] == "error")
+        check("wrong_server_callback_rejected", not sessions.deliver_callback_flow(
+            sid, "wrong", code="invalid", state="invalid")["ok"])
+        check("wrong_owner_cancel_rejected", not sessions.cancel_flow(sid, "positive", str(other))["ok"])
+        check("wrong_owner_did_not_cancel", sessions.poll_flow(sid, "positive")["status"] == "pending")
+        response = browser.get(flow["auth_url"], follow_redirects=True)
+        check("real_http_callback_received", response.status_code == 200 and bool(callback.callback.get("code")))
+        captured = callback.callback
+        check("correct_callback_accepted", sessions.deliver_callback_flow(sid, "positive", **captured)["ok"])
+        result = done(flow, "positive")
+        receipt["positive_status"] = result["status"]
+        check("positive_approved", result["status"] == "approved")
+        check("real_mcp_tools_discovered", any(t["name"] == "fixture_ping" for t in result["tools"]))
+        check("callback_replay_rejected", not sessions.deliver_callback_flow(sid, "positive", **captured)["ok"])
+        token_path = owner / "mcp-tokens" / "positive.json"
+        stored = json.loads(token_path.read_text(encoding="utf-8"))
+        check("disk_token_matches_provider", stored["access_token"] in provider.tokens)
+        check("disk_refresh_token_and_absolute_expiry", bool(stored.get("refresh_token")) and stored["expires_at"] > time.time())
+        check("token_file_private", token_path.stat().st_mode & 0o777 == 0o600)
+        reloaded = asyncio.run(HermesTokenStorage("positive", hermes_home=owner).get_tokens())
+        check("fresh_storage_reload", reloaded.access_token in provider.tokens)
+        check("wrong_profile_has_no_token", not (other / "mcp-tokens" / "positive.json").exists())
+        check("config_saved_to_owner", "positive" in (owner / "config.yaml").read_text(encoding="utf-8"))
+        exchanges = provider.pkce_verified
+        cold = subprocess.run([sys.executable, str(Path(__file__).resolve()),
+            "--repo", str(repo), "--output", os.devnull, "--cold-probe"],
+            stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+            timeout=40, check=False)
+        check("fresh_process_authenticated_mcp_from_disk", cold.returncode == 0)
+        check("fresh_process_did_not_reauthorize", provider.pkce_verified == exchanges)
+
+        flow = begin("cancelled")
+        sid = flow["session_id"]
+        exchanges = provider.pkce_verified
+        check("owner_cancel_accepted", sessions.cancel_flow(sid, "cancelled", str(owner))["ok"])
+        check("cancel_worker_terminated", done(flow, "cancelled")["status"] == "error")
+        check("cancel_did_not_exchange_or_persist", provider.pkce_verified == exchanges and
+              not (owner / "mcp-tokens" / "cancelled.json").exists())
+        check("callback_after_cancel_rejected", not sessions.deliver_callback_flow(
+            sid, "cancelled", code="invalid", state="invalid")["ok"])
+
+        flow = begin("loopback", remote=False)
+        check("backend_listener_http_success", browser.get(flow["auth_url"], follow_redirects=True).status_code == 200)
+        check("backend_listener_approved", done(flow, "loopback")["status"] == "approved")
+        check("backend_listener_closed", eventually(
+            lambda: sessions._sessions[flow["session_id"]]["httpd"] is None,
+            "backend listener shutdown completed"))
+
+        # Even possession of valid session/state cannot cross the resolved home.
+        flow = begin("owner_boundary")
+        browser.get(flow["auth_url"], follow_redirects=True)
+        override = set_hermes_home_override(other)
+        try:
+            foreign_poll = sessions.poll_flow(flow["session_id"], "owner_boundary")
+            foreign_callback = sessions.deliver_callback_flow(flow["session_id"], "owner_boundary", **callback.callback)
+        finally:
+            reset_hermes_home_override(override)
+        receipt["ownership_boundary"] = {
+            "cancel_owner_enforced": checks["wrong_owner_cancel_rejected"],
+            "cross_profile_poll_rejected": foreign_poll["status"] == "error",
+            "cross_profile_callback_rejected_with_session_and_state": not foreign_callback["ok"],
+        }
+        check("wrong_owner_poll_rejected", foreign_poll["status"] == "error" and "auth_url" not in foreign_poll)
+        check("wrong_owner_callback_rejected", not foreign_callback["ok"])
+        check("owner_callback_after_foreign_attempt_accepted", sessions.deliver_callback_flow(
+            flow["session_id"], "owner_boundary", **callback.callback)["ok"])
+        check("boundary_probe_worker_completed", done(flow, "owner_boundary")["status"] == "approved")
+        check("boundary_exchange_still_persists_only_to_owner", (owner / "mcp-tokens" / "owner_boundary.json").exists()
+              and not (other / "mcp-tokens" / "owner_boundary.json").exists())
+        receipt["http_events"] = provider.events
+        receipt["pkce_exchanges_verified"] = provider.pkce_verified
+        check("real_discovery_dcr_token_http", all(any(path == target for _, path in provider.events)
+              for target in ["/.well-known/oauth-protected-resource", "/register", "/token"]))
+        receipt["persisted_files_verified_then_removed"] = sorted(p.name for p in (owner / "mcp-tokens").iterdir())
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--repo", type=Path, required=True)
+    parser.add_argument("--output", type=Path, required=True)
+    parser.add_argument("--isolated-worker", action="store_true", help=argparse.SUPPRESS)
+    parser.add_argument("--cold-probe", action="store_true", help=argparse.SUPPRESS)
+    args = parser.parse_args()
+    repo, output = args.repo.resolve(), args.output.resolve()
+    if args.cold_probe:
+        sys.path.insert(0, str(repo))
+        logging.disable(logging.CRITICAL)
+        from hermes_cli.mcp_config import _get_mcp_servers, _probe_single_server
+        from tools.mcp_oauth import suppress_interactive_oauth
+        with suppress_interactive_oauth():
+            tools = _probe_single_server("positive", _get_mcp_servers()["positive"])
+        return 0 if any(name == "fixture_ping" for name, _ in tools) else 1
+    output.parent.mkdir(parents=True, exist_ok=True)
+    if not args.isolated_worker:
+        with tempfile.TemporaryDirectory(prefix="hermes-oauth-http-") as temp:
+            home = Path(temp)
+            (home / ".hermes").mkdir()
+            env = {"PATH": os.environ.get("PATH", ""), "HOME": temp,
+                   "HERMES_HOME": str(home / ".hermes"), "LANG": "C.UTF-8", "TZ": "UTC",
+                   "PYTHONNOUSERSITE": "1"}
+            completed = subprocess.run([sys.executable, str(Path(__file__).resolve()),
+                "--repo", str(repo), "--output", str(output), "--isolated-worker"],
+                env=env, cwd=temp, stdin=subprocess.DEVNULL, timeout=180, check=False)
+        print(json.dumps({"receipt": str(output), "exit_code": completed.returncode, "isolated_home_removed": True}))
+        return completed.returncode
+    receipt = {"checks": {}, "repo": str(repo), "fidelity":
+               "Real production OAuth session functions + real HTTP discovery/DCR/PKCE/token/MCP/callback + real disk persistence; no production mocks; not gateway RPC transport or Electron",
+               "isolated_home": True, "credentials": "ephemeral local fixture only; no token material in receipt"}
+    try:
+        with open(os.devnull, "w", encoding="utf-8") as sink, contextlib.redirect_stdout(sink), contextlib.redirect_stderr(sink):
+            run_probe(repo, receipt)
+        receipt["status"] = "passed"
+    except Exception as exc:
+        receipt["status"] = "failed"
+        receipt["error_type"] = type(exc).__name__
+        receipt["failed_check"] = next((k for k, v in receipt["checks"].items() if not v), None)
+        # Assertions are our fixed labels, not provider bodies or secret-bearing URLs.
+        if isinstance(exc, AssertionError):
+            receipt["assertion"] = str(exc)
+    output.write_text(json.dumps(receipt, indent=2) + "\n", encoding="utf-8")
+    return 0 if receipt["status"] == "passed" else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/evals/desktop_mcp_oauth/renderer_lifecycle.mjs b/evals/desktop_mcp_oauth/renderer_lifecycle.mjs
new file mode 100644
index 0000000000..ec35c326c7
--- /dev/null
+++ b/evals/desktop_mcp_oauth/renderer_lifecycle.mjs
@@ -0,0 +1,196 @@
+import { createRequire } from 'node:module'
+import fs from 'node:fs/promises'
+import assert from 'node:assert/strict'
+import http from 'node:http'
+import path from 'node:path'
+const repo = process.argv[2],
+  out = process.argv[3],
+  scenario = process.argv[4] || 'approved'
+const require = createRequire(path.join(repo, 'package.json'))
+const { build } = require('esbuild')
+const { chromium } = require('playwright')
+const requests = []
+const rpcCalls = []
+const nativeCalls = []
+const handlers = new Map()
+const nativeBuild = await build({
+  entryPoints: [repo + '/apps/desktop/electron/mcp-oauth-callback-ipc.ts'],
+  bundle: true,
+  write: false,
+  platform: 'node',
+  format: 'cjs',
+  external: ['electron']
+})
+const nativeModule = { exports: {} }
+new Function('require', 'module', 'exports', nativeBuild.outputFiles[0].text)(
+  id => (id === 'electron' ? { ipcMain: { handle: (name, fn) => handlers.set(name, fn) } } : require(id)),
+  nativeModule,
+  nativeModule.exports
+)
+nativeModule.exports.registerMcpOauthCallbackIpc()
+const { WebSocketServer } = require('ws')
+const wss = new WebSocketServer({ port: 0, host: '127.0.0.1' })
+await new Promise(r => wss.once('listening', r))
+let approved = false
+let redirectUri
+wss.on('connection', ws =>
+  ws.on('message', raw => {
+    const m = JSON.parse(raw)
+    rpcCalls.push(m)
+    let result = { ok: true }
+    if (m.method === 'mcp.servers.oauth.start') {
+      redirectUri = m.params.client_redirect_uri
+      result = {
+        ok: true,
+        session_id: 'fixture-session',
+        auth_url:
+          'http://127.0.0.1/authorize?state=fixture-state&redirect_uri=' +
+          encodeURIComponent(scenario === 'legacy' ? 'http://remote.invalid/callback' : redirectUri)
+      }
+    }
+    if (m.method === 'mcp.servers.oauth.callback') {
+      approved = m.params.state === 'fixture-state' && m.params.code === 'fixture-code'
+      result = { ok: approved }
+    }
+    if (m.method === 'mcp.servers.oauth.poll')
+      result = { ok: true, status: approved ? 'approved' : 'pending', tools: [] }
+    ws.send(JSON.stringify({ jsonrpc: '2.0', id: m.id, result }))
+  })
+)
+const entry = `import React from 'react'; import {createRoot} from 'react-dom/client';
+import {QueryClient, QueryClientProvider} from '@tanstack/react-query';
+import {McpTab} from '${repo}/apps/desktop/src/app/skills/mcp-tab.tsx';
+import {setApiRequestProfile,setApiRequestConnection} from '${repo}/apps/desktop/src/api/client.ts';
+setApiRequestProfile('profile-b');setApiRequestConnection('fixture-remote');
+const queryClient=new QueryClient({defaultOptions:{queries:{retry:false}}});function Probe(){const [profile,setProfile]=React.useState('profile-b');const [shown,setShown]=React.useState(true);window.cancelFlow=()=>setProfile('profile-c');window.unmountFlow=()=>setShown(false);return <QueryClientProvider client={queryClient}>{shown && <McpTab gateway={null} profile={{connectionId:'fixture-remote',profile}} />}</QueryClientProvider>};createRoot(document.getElementById('root')).render(<Probe/>);`
+const bundle = await build({
+  stdin: { contents: entry, resolveDir: repo, loader: 'tsx' },
+  bundle: true,
+  write: false,
+  format: 'esm',
+  platform: 'browser',
+  jsx: 'automatic',
+  alias: { '@': repo + '/apps/desktop/src', '@hermes/shared': repo + '/apps/shared/src' },
+  define: {
+    'process.env.NODE_ENV': '"production"',
+    'import.meta.env': '{}',
+    'import.meta.env.DEV': 'false',
+    'import.meta.env.PROD': 'true'
+  },
+  loader: { '.woff2': 'dataurl', '.woff': 'dataurl', '.svg': 'dataurl' },
+  logLevel: 'warning'
+})
+const server = http.createServer(async (req, res) => {
+  res.setHeader('Content-Type', req.url === '/app.js' ? 'text/javascript' : 'text/html')
+  res.end(
+    req.url === '/app.js'
+      ? bundle.outputFiles[0].text
+      : '<div id="root"></div><script type="module" src="/app.js"></script>'
+  )
+})
+await new Promise(r => server.listen(0, '127.0.0.1', r))
+const browser = await chromium.launch({
+  headless: true,
+  args: ['--no-sandbox'],
+  ...(process.env.CHROMIUM_EXECUTABLE ? { executablePath: process.env.CHROMIUM_EXECUTABLE } : {})
+})
+try {
+  const page = await browser.newPage()
+  const errors = []
+  page.on('pageerror', e => {
+    errors.push(e.message)
+    console.error(e.stack)
+  })
+  page.on('console', m => console.log(m.type(), m.text()))
+  await page.exposeFunction('recordRequest', r => {
+    requests.push(r)
+    console.log('request', JSON.stringify(r))
+    if (r.path === '/api/config')
+      return { mcp_servers: { reports: { url: 'https://fixture.invalid/mcp', auth: 'oauth' } } }
+    if (r.path.endsWith('/test')) return { ok: false, tools: [], auth_required: true, error: 'OAuth required' }
+    if (r.path.includes('/logs')) return { lines: [] }
+    if (r.path.includes('/catalog')) return { entries: [] }
+    if (r.path.endsWith('/auth'))
+      return { flow_id: 'f', status: 'error', error: 'fixture remote HTTP redirect rejected' }
+    return { ok: true, providers: [] }
+  })
+  await page.exposeFunction('nativeOAuth', async (action, id) => {
+    nativeCalls.push({ action, id })
+    return handlers.get('hermes:mcp-oauth:' + action)({}, id)
+  })
+  await page.exposeFunction('fixtureConnection', async scope => {
+    nativeCalls.push({ action: 'connection', scope })
+    return { wsUrl: 'ws://127.0.0.1:' + wss.address().port, connectionId: scope.connectionId, authMode: 'token' }
+  })
+  await page.exposeFunction('openAuthorization', async url => {
+    nativeCalls.push({ action: 'open', url })
+    if (scenario === 'cancel' || scenario === 'unmount') {
+      await page.evaluate(s => (s === 'unmount' ? window.unmountFlow() : window.cancelFlow()), scenario)
+      return
+    }
+    const u = new URL(url)
+    await fetch(u.searchParams.get('redirect_uri') + '?code=fixture-code&state=' + u.searchParams.get('state'))
+  })
+  await page.addInitScript(() => {
+    window.hermesDesktop = {
+      api: r => window.recordRequest(r),
+      getConnectionFor: r => window.fixtureConnection(r),
+      mcpOauth: {
+        listen: () => {
+          window.nativeListen = true
+          return window.nativeOAuth('listen')
+        },
+        wait: id => window.nativeOAuth('wait', id),
+        cancel: id => window.nativeOAuth('cancel', id)
+      },
+      openExternal: url => window.openAuthorization(url)
+    }
+  })
+  await page.goto('http://127.0.0.1:' + server.address().port)
+  await page.waitForTimeout(1500)
+  console.log('BODY', await page.locator('body').innerText())
+  await page.getByText('Reports', { exact: true }).first().click()
+  await page.waitForTimeout(300)
+  console.log('SELECTED', await page.locator('body').innerText())
+  await page
+    .getByRole('button', { name: /Authenticate|Sign in/ })
+    .first()
+    .click()
+  await page.waitForTimeout(2500)
+  const hasScope = await page.evaluate(() => !!window.nativeListen)
+  const result = {
+    repo,
+    scenario,
+    requests,
+    rpcCalls,
+    nativeCalls,
+    approved,
+    hasScope,
+    errors,
+    text: await page.locator('body').innerText(),
+    fidelity: 'production McpTab + all real renderer imports in headless Chromium; fixture native API transport'
+  }
+  await fs.writeFile(out, JSON.stringify(result, null, 2))
+  console.log(JSON.stringify(result))
+  if (scenario === 'cancel' || scenario === 'unmount') {
+    assert(
+      rpcCalls.some(m => m.method === 'mcp.servers.oauth.cancel'),
+      'abandoned component must cancel backend'
+    )
+    assert(
+      nativeCalls.some(m => m.action === 'cancel'),
+      'abandoned component must close listener'
+    )
+    assert(
+      nativeCalls.filter(m => m.action === 'connection').every(m => m.scope.profile === 'profile-b'),
+      'cleanup must retain original owner'
+    )
+    assert(!rpcCalls.some(m => m.method === 'mcp.servers.oauth.callback'), 'no callback after abandonment')
+  }
+  assert.deepEqual(errors, [])
+} finally {
+  await browser.close()
+  server.close()
+  for (const c of wss.clients) c.terminate()
+  wss.close()
+}
diff --git a/tests/tui_gateway/test_mcp_oauth_cancel.py b/tests/tui_gateway/test_mcp_oauth_cancel.py
index 26238f4a38..0fd7918f0e 100644
--- a/tests/tui_gateway/test_mcp_oauth_cancel.py
+++ b/tests/tui_gateway/test_mcp_oauth_cancel.py
@@ -34,6 +34,7 @@ def test_cancel_is_scoped_idempotent_and_releases_worker(
 
     monkeypatch.setattr(sessions, "_worker", worker)
     home = str(tmp_path / "origin")
+    monkeypatch.setenv("HERMES_HOME", home)
     result = sessions.start_flow(
         home,
         "reports",
@@ -79,6 +80,43 @@ def test_cancel_is_scoped_idempotent_and_releases_worker(
         finished.wait(5)
 
 
+@pytest.mark.parametrize("operation", ["poll", "callback", "cancel"])
+def test_session_operations_require_resolved_owner(tmp_path, monkeypatch, operation):
+    from hermes_constants import reset_hermes_home_override, set_hermes_home_override
+
+    home = str(tmp_path / "owner")
+    flow = DashboardOAuthFlow(
+        "owned", "reports", None, home, "http://127.0.0.1:49152/callback"
+    )
+    asyncio.run(flow.publish_authorization_url("https://idp.example/authorize?state=test"))
+    monkeypatch.setattr(sessions, "_sessions", {
+        "owned": {"flow": flow, "server_name": "reports", "hermes_home": home, "httpd": None}
+    })
+
+    def invoke():
+        if operation == "poll":
+            return sessions.poll_flow("owned", "reports")
+        if operation == "callback":
+            return sessions.deliver_callback_flow("owned", "reports", code="valid", state="test")
+        from hermes_constants import get_hermes_home
+        return sessions.cancel_flow("owned", "reports", str(get_hermes_home()))
+
+    token = set_hermes_home_override(tmp_path / "other")
+    try:
+        rejected = invoke()
+    finally:
+        reset_hermes_home_override(token)
+    assert "profile mismatch" in (rejected.get("error_message") or "")
+    assert "auth_url" not in rejected
+    assert flow.snapshot()["status"] == "authorization_required"
+    token = set_hermes_home_override(home)
+    try:
+        accepted = invoke()
+    finally:
+        reset_hermes_home_override(token)
+    assert accepted.get("ok", accepted.get("status") == "pending") is True
+
+
 def test_cancel_does_not_revoke_an_approved_flow(tmp_path, monkeypatch):
     home = str(tmp_path)
     flow = DashboardOAuthFlow(
diff --git a/tests/tui_gateway/test_mcp_oauth_client_callback.py b/tests/tui_gateway/test_mcp_oauth_client_callback.py
index 1ef9b99a46..e89a2eb245 100644
--- a/tests/tui_gateway/test_mcp_oauth_client_callback.py
+++ b/tests/tui_gateway/test_mcp_oauth_client_callback.py
@@ -16,6 +16,7 @@ import threading
 
 import pytest
 
+from hermes_constants import get_hermes_home
 from tools.mcp_dashboard_oauth import DashboardOAuthFlow
 from tui_gateway import mcp_oauth_sessions
 from tui_gateway.mcp_oauth_sessions import (
@@ -106,7 +107,7 @@ def test_start_flow_client_redirect_skips_gateway_listener(monkeypatch):
     )
 
     result = mcp_oauth_sessions.start_flow(
-        "/tmp/hermes-test-home",
+        str(get_hermes_home()),
         "clicky",
         {"url": "https://mcp.example.com/mcp", "auth": "oauth"},
         client_redirect_uri="http://127.0.0.1:8412/callback",
@@ -131,7 +132,7 @@ def test_start_flow_rejects_bad_client_redirect(monkeypatch):
     _fake_worker_publishes_url(monkeypatch)
     with pytest.raises(ValueError):
         mcp_oauth_sessions.start_flow(
-            "/tmp/hermes-test-home",
+            str(get_hermes_home()),
             "clicky2",
             {"url": "https://mcp.example.com/mcp", "auth": "oauth"},
             client_redirect_uri="https://evil.example.com/callback",
@@ -152,7 +153,7 @@ def _make_session(session_id="sess-relay-1", server="hosp", state="s3cr3tstate")
         flow_id=session_id,
         server_name=server,
         profile=None,
-        hermes_home="/tmp/hermes-test-home",
+        hermes_home=str(get_hermes_home()),
         redirect_uri="http://127.0.0.1:9000/callback",
     )
     # Pin the expected state the way publish_authorization_url does.
@@ -166,7 +167,7 @@ def _make_session(session_id="sess-relay-1", server="hosp", state="s3cr3tstate")
     rec = {
         "session_id": session_id,
         "server_name": server,
-        "hermes_home": "/tmp/hermes-test-home",
+        "hermes_home": str(get_hermes_home()),
         "flow": flow,
         "httpd": None,
         "created_at": __import__("time").time(),
diff --git a/tui_gateway/mcp_oauth_sessions.py b/tui_gateway/mcp_oauth_sessions.py
index 52ba8f06cd..fc74662fab 100644
--- a/tui_gateway/mcp_oauth_sessions.py
+++ b/tui_gateway/mcp_oauth_sessions.py
@@ -207,14 +207,19 @@ def start_flow(
     return {"session_id": session_id, "auth_url": auth_url, "flow": "pkce"}
 
 
-def _lookup(session_id: str, server_name: str) -> "tuple[Dict[str, Any] | None, str | None]":
-    """Find a session record; returns ``(rec, None)`` or ``(None, error_message)``."""
+def _lookup(
+    session_id: str, server_name: str, hermes_home: Optional[str] = None,
+) -> "tuple[Dict[str, Any] | None, str | None]":
+    """Find a session belonging to the caller's resolved profile."""
+    from hermes_constants import hermes_home_key
     with _sessions_lock:
         rec = _sessions.get(session_id)
     if rec is None:
         return None, "OAuth session not found or expired"
     if rec["server_name"] != server_name:
         return None, "server name mismatch for session"
+    if hermes_home_key(rec["hermes_home"]) != hermes_home_key(hermes_home):
+        return None, "profile mismatch for session"
     return rec, None
 
 
@@ -239,11 +244,9 @@ def poll_flow(session_id: str, server_name: str) -> Dict[str, Any]:
 
 def cancel_flow(session_id: str, server_name: str, hermes_home: str) -> Dict[str, Any]:
     """Cancel only the owning profile's flow and release its callback waiter."""
-    rec, err = _lookup(session_id, server_name)
+    rec, err = _lookup(session_id, server_name, hermes_home)
     if rec is None:
         return {"ok": False, "error_message": err}
-    if rec["hermes_home"] != hermes_home:
-        return {"ok": False, "error_message": "profile mismatch for session"}
     flow = rec["flow"]
     flow.mark_error("OAuth cancelled by user")
     _shutdown_listener(rec)
diff --git a/website/docs/guides/oauth-over-ssh.md b/website/docs/guides/oauth-over-ssh.md
index c58aeb401d..258f1d1324 100644
--- a/website/docs/guides/oauth-over-ssh.md
+++ b/website/docs/guides/oauth-over-ssh.md
@@ -44,6 +44,14 @@ If your provider isn't in the table, you don't need a tunnel.
 
 ## MCP Servers
 
+**Desktop Skills → MCP:** the native app receives the callback on your computer
+and relays it to the selected connection and profile, so this flow does not need
+an SSH callback tunnel or `dashboard.public_url`. Tokens stay on the owning
+backend profile. Leaving the MCP tab or changing its scope cancels pending
+sign-in. If Desktop asks you to update the backend, update it before retrying;
+it does not fall back to a remote HTTP callback. The terminal workflows below
+are unchanged.
+
 Remote MCP servers (Linear, Sentry, Atlassian, Asana, Figma, etc.) use the same loopback redirect flow. Hermes auto-picks a free port per server and prints the authorize URL when the OAuth flow kicks off — either at startup (when a new server appears in `mcp_servers:`) or when you run `hermes mcp login <server>`.
 
 You have two ways to complete it from a remote host: