diff --git a/tests/e2e/core/tenancy/_helpers.py b/tests/e2e/core/tenancy/_helpers.py
new file mode 100644
index 0000000000..2808c7ded3
--- /dev/null
+++ b/tests/e2e/core/tenancy/_helpers.py
@@ -0,0 +1,527 @@
+"""Lane-private harness for the C7 two-tenant canary suites.
+
+Every tenant (profile) owns a set of random canaries: its provider key (same env var NAME in every
+profile, different value), its API-server key, a non-secret ``.env`` marker, its model id, memory
+text, SOUL text, terminal cwd and a cron prompt. Each tenant also owns its own loopback provider
+(``FakeLLMServer``) that accepts only its own key, so every request the provider records is proof of
+WHO sent it. The invariant shared by every scenario is ``leaks(...) == []``: no tenant's canary may
+appear in another tenant's provider request, tool-subprocess env snapshot, or on-disk file.
+
+Hermes runs for real in child processes with HOME=<tmp>/home and HERMES_HOME=<tmp>/home/.hermes
+(profiles resolve under $HOME, never the real install), every credential env var stripped.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import secrets
+import signal
+import socket
+import subprocess
+import sys
+import time
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Callable, Iterable
+
+import yaml
+
+from tests.fakes.fake_llm_provider import FakeLLMServer, Response, Text, ToolCall
+
+REPO_ROOT = Path(__file__).resolve().parents[4]
+ENV_PROBE = "echo SNAPSHOT-BEGIN; env | sort; echo CWD=$(pwd); echo SNAPSHOT-END"
+PROVIDER_KEY_ENV = "TENANT_PROVIDER_KEY"  # same NAME in every profile's .env, distinct VALUE
+
+_STRIP_SUFFIXES = ("_API_KEY", "_TOKEN", "_BASE_URL", "_SECRET", "_ACCESS_KEY", "_KEY_ID", "_KEY")
+_STRIP_PREFIXES = ("HERMES_", "OPENAI", "ANTHROPIC", "OPENROUTER", "AWS_", "AZURE_", "GOOGLE_", "GEMINI",
+                   "PYTEST_", "NOUS_", "XAI_", "LLM_", "CUSTOM_", "TERMINAL_", "TENANT_", "API_SERVER_")
+
+
+def real_user_home() -> Path:
+    import pwd
+    return Path(pwd.getpwuid(os.getuid()).pw_dir).resolve()
+
+
+def hermetic_env(home: Path, extra: dict[str, str] | None = None) -> dict[str, str]:
+    """Child env: fake HOME (profile root anchor) + HERMES_HOME under it, nothing credential-shaped."""
+    home = home.resolve()
+    assert home != real_user_home() and home / ".hermes" != real_user_home() / ".hermes", home
+    env = {k: v for k, v in os.environ.items()
+           if not (k.endswith(_STRIP_SUFFIXES) or k.startswith(_STRIP_PREFIXES))}
+    for var in ("HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "http_proxy", "https_proxy", "all_proxy", "XDG_STATE_HOME",
+                "XDG_CONFIG_HOME", "XDG_DATA_HOME", "XDG_CACHE_HOME",
+                # no route to the developer's systemd --user bus: a child can never see or touch the
+                # live hermes-gateway unit, whatever it decides about service management.
+                "DBUS_SESSION_BUS_ADDRESS", "XDG_RUNTIME_DIR"):
+        env.pop(var, None)
+    env.update(
+        HOME=str(home),
+        HERMES_HOME=str(home / ".hermes"),
+        XDG_STATE_HOME=str(home / ".local" / "state"),
+        PYTHONPATH=str(REPO_ROOT),
+        NO_COLOR="1",
+        TERM="dumb",
+        NO_PROXY="127.0.0.1,localhost",
+        no_proxy="127.0.0.1,localhost",
+        # The live-DB guard treats $HOME/.hermes/state.db of a pytest descendant as production;
+        # this HOME is the test's own tmp dir (asserted above).
+        HERMES_STATE_DB_GUARD_BYPASS="1",
+        HERMES_ACCEPT_HOOKS="1",
+    )
+    env.update(extra or {})
+    return env
+
+
+def free_port() -> int:
+    with socket.socket() as s:
+        s.bind(("127.0.0.1", 0))
+        return s.getsockname()[1]
+
+
+def poll(pred: Callable[[], Any], timeout: float, what: str, interval: float = 0.1) -> Any:
+    deadline = time.monotonic() + timeout
+    while True:
+        got = pred()
+        if got:
+            return got
+        if time.monotonic() > deadline:
+            raise AssertionError(f"timed out after {timeout}s waiting for {what}")
+        time.sleep(interval)
+
+
+# Tenants -------------------------------------------------------------------------------------------
+
+
+@dataclass
+class Tenant:
+    name: str
+    home: Path  # the profile's HERMES_HOME
+    workdir: Path
+    tag: str = field(default_factory=lambda: secrets.token_hex(5))
+    srv: FakeLLMServer | None = None
+    extra: dict[str, str] = field(default_factory=dict)  # canaries minted mid-scenario (RPC writes)
+
+    # canaries (each embeds the random tag, so a hit can never be a coincidence)
+    @property
+    def provider_key(self) -> str:
+        return f"sk-prov-{self.name}-{self.tag}"
+
+    @property
+    def api_server_key(self) -> str:
+        return f"apisrv-{self.name}-{self.tag}"
+
+    @property
+    def env_marker(self) -> str:
+        return f"envmark-{self.name}-{self.tag}"
+
+    @property
+    def model(self) -> str:
+        return f"model-{self.name}-{self.tag}"
+
+    @property
+    def memory(self) -> str:
+        return f"memory-canary-{self.name}-{self.tag}"
+
+    @property
+    def soul(self) -> str:
+        return f"soul-canary-{self.name}-{self.tag}"
+
+    @property
+    def shell_state(self) -> str:
+        """Exported into the tool shell AFTER each env dump: a shell reused by another tenant shows it."""
+        return f"shellstate-{self.name}-{self.tag}"
+
+    @property
+    def cron_prompt(self) -> str:
+        return f"cron-canary-{self.name}-{self.tag}: run the environment probe"
+
+    def canaries(self) -> dict[str, str]:
+        return {
+            "provider_key": self.provider_key, "api_server_key": self.api_server_key,
+            "env_marker": self.env_marker, "model": self.model, "memory": self.memory, "soul": self.soul,
+            "workdir": str(self.workdir), "cron_prompt": self.cron_prompt.split(":")[0],
+            "shell_state": self.shell_state, **self.extra,
+        }
+
+    def secrets(self) -> dict[str, str]:
+        return {"provider_key": self.provider_key, "api_server_key": self.api_server_key}
+
+
+def _responder(t: Tenant) -> Callable[[dict[str, Any]], Response]:
+    """Every user turn runs the env probe in the terminal tool; the follow-up answers with text."""
+    def respond(record: dict[str, Any]) -> Response:
+        msgs = record["body"].get("messages") or []
+        last = msgs[-1] if msgs else {}
+        if last.get("role") == "tool":
+            return Text(f"probe done for {t.name}")
+        return ToolCall("terminal", {"command": f"{ENV_PROBE}; export TENANT_SHELL_STATE={t.shell_state}"})
+    return respond
+
+
+def make_tenants(root: Path, names: Iterable[str], launch: str = "default") -> dict[str, Tenant]:
+    """Build tenant homes on disk: launch profile at HOME/.hermes, the rest under profiles/<name>."""
+    hermes_home = root / "home" / ".hermes"
+    tenants: dict[str, Tenant] = {}
+    for name in names:
+        home = hermes_home if name == launch else hermes_home / "profiles" / name
+        t = Tenant(name=name, home=home, workdir=root / "work")
+        t.workdir = root / f"work-{name}-{t.tag}"
+        t.workdir.mkdir(parents=True)
+        t.srv = FakeLLMServer(_responder(t), api_key=t.provider_key)
+        tenants[name] = t
+    return tenants
+
+
+def write_tenant_home(t: Tenant, extra_config: dict[str, Any] | None = None,
+                      extra_env: dict[str, str] | None = None) -> None:
+    assert t.srv is not None
+    t.home.mkdir(parents=True, exist_ok=True)
+    cfg: dict[str, Any] = {
+        "model": {"provider": "custom", "base_url": t.srv.base_url, "default": t.model,
+                  "key_env": PROVIDER_KEY_ENV, "context_length": 128000},
+        "agent": {"api_max_retries": 1},
+        "terminal": {"backend": "local", "cwd": str(t.workdir)},
+        "memory": {"memory_enabled": True},
+        "compression": {"enabled": False},
+    }
+    for k, v in (extra_config or {}).items():
+        cfg[k] = {**cfg.get(k, {}), **v} if isinstance(v, dict) and isinstance(cfg.get(k), dict) else v
+    (t.home / "config.yaml").write_text(yaml.safe_dump(cfg, sort_keys=False), encoding="utf-8")
+    env = {PROVIDER_KEY_ENV: t.provider_key, "API_SERVER_KEY": t.api_server_key, "TENANT_MARKER": t.env_marker,
+           **(extra_env or {})}
+    (t.home / ".env").write_text("".join(f"{k}={v}\n" for k, v in env.items()), encoding="utf-8")
+    (t.home / "memories").mkdir(exist_ok=True)
+    (t.home / "memories" / "MEMORY.md").write_text(f"{t.memory}\n", encoding="utf-8")
+    (t.home / "SOUL.md").write_text(f"You are {t.soul}.\n", encoding="utf-8")
+
+
+def assert_profiles_root_under(root: Path, home: Path) -> None:
+    """The profile root is HOME-anchored: prove it resolves inside ``root`` before any write."""
+    probe = subprocess.run(
+        [sys.executable, "-c", "from hermes_cli.profiles import _get_profiles_root as r; print(r())"],
+        env=hermetic_env(home), cwd=str(home), capture_output=True, text=True, timeout=120,
+        stdin=subprocess.DEVNULL,
+    )
+    assert probe.returncode == 0, probe.stderr[-2000:]
+    got = Path(probe.stdout.strip().splitlines()[-1]).resolve()
+    assert str(got).startswith(str(root.resolve())), f"profiles root escaped the sandbox: {got}"
+
+
+# Invariants ----------------------------------------------------------------------------------------
+
+
+def _bearer(record: dict[str, Any]) -> str:
+    auth = record.get("auth") or ""
+    return auth[7:] if auth.lower().startswith("bearer ") else auth
+
+
+def request_leaks(tenants: dict[str, Tenant]) -> list[str]:
+    """Every provider request carries its own tenant's key/model and no other tenant's canary."""
+    problems: list[str] = []
+    for t in tenants.values():
+        assert t.srv is not None
+        for i, r in enumerate(list(t.srv.requests)):
+            where = f"provider[{t.name}] request #{i} ({r['kind']} {r['path']})"
+            if _bearer(r) != t.provider_key:
+                owner = next((u.name for u in tenants.values() if u.provider_key == _bearer(r)), "<none>")
+                problems.append(f"{where}: Authorization is {owner!r}'s key, not {t.name!r}'s")
+            if r["kind"] == "main" and r["body"].get("model") != t.model:
+                problems.append(f"{where}: model {r['body'].get('model')!r} != {t.model!r}")
+            blob = json.dumps(r["body"]) + json.dumps(r.get("headers") or {})
+            problems += [f"{where}: carries {u.name}'s {kind} ({value})"
+                         for u in tenants.values() if u is not t
+                         for kind, value in u.canaries().items() if value in blob]
+    return problems
+
+
+def env_snapshots(t: Tenant) -> list[tuple[str, str]]:
+    """``(origin, output)`` of every ENV_PROBE terminal call as sent back to the tenant's provider;
+    origin is ``cron job`` when the conversation is the tenant's cron prompt, else ``turn``."""
+    assert t.srv is not None
+    seen: dict[str, tuple[str, str]] = {}  # tool_call_id -> (origin, output); results are re-sent later
+    cron_mark = t.cron_prompt.split(":")[0]
+    for r in list(t.srv.requests):
+        origin = "cron job" if cron_mark in json.dumps(r["body"].get("messages") or []) else "turn"
+        for m in r["body"].get("messages") or []:
+            content = m.get("content")
+            text = content if isinstance(content, str) else json.dumps(content)
+            if m.get("role") == "tool" and "SNAPSHOT-BEGIN" in text:
+                seen.setdefault(str(m.get("tool_call_id") or text), (origin, text))
+    return list(seen.values())
+
+
+def snapshot_problems(tenants: dict[str, Tenant], min_per_tenant: int = 1) -> list[str]:
+    """Inside a tool subprocess: cwd is the tenant's own, and no other tenant's canary is visible."""
+    problems: list[str] = []
+    for t in tenants.values():
+        snaps = env_snapshots(t)
+        if len(snaps) < min_per_tenant:
+            problems.append(f"{t.name}: {len(snaps)} env snapshot(s) reached its provider, expected >= {min_per_tenant}")
+        for origin, s in snaps:
+            text = s.encode().decode("unicode_escape", errors="ignore") if "\\n" in s else s
+            own_dirs = [str(t.workdir), *(v for k, v in t.extra.items() if k.startswith("workdir"))]
+            if not any(f"CWD={d}\n" in text + "\n" for d in own_dirs):
+                cwd = next((ln for ln in text.splitlines() if ln.startswith("CWD=")), "<no CWD line>")
+                problems.append(f"{t.name}: terminal ({origin}) ran in {cwd.strip()!r}, not its own {own_dirs}")
+            problems += [f"{t.name}: tool subprocess ({origin}) carries {u.name}'s {kind}: "
+                         f"{next((ln.strip() for ln in text.splitlines() if value in ln), value)[:300]!r}"
+                         for u in tenants.values() if u is not t
+                         for kind, value in u.canaries().items() if value in text]
+    return problems
+
+
+def file_leaks(tenants: dict[str, Tenant], launch: str = "default") -> list[str]:
+    """Byte-scan every file of every tenant home (state.db + WAL, logs, sessions, config, .env,
+    memories, cron output) for any other tenant's canary. The launch home's ``profiles/`` subtree
+    belongs to the other tenants and is scanned as theirs."""
+    problems: list[str] = []
+    for t in tenants.values():
+        foreign = [(u.name, kind, value.encode()) for u in tenants.values() if u is not t
+                   for kind, value in u.canaries().items()]
+        for path in t.home.rglob("*"):
+            if not path.is_file() or (t.name == launch and "profiles" in path.relative_to(t.home).parts[:1]):
+                continue
+            try:
+                data = path.read_bytes()
+            except OSError:
+                continue
+            problems += [f"{t.name}: {path.relative_to(t.home)} contains {owner}'s {kind}"
+                         for owner, kind, value in foreign if value in data]
+    return problems
+
+
+def check_isolation(tenants: dict[str, Tenant], *, launch: str = "default", min_snapshots: int = 1,
+                    extra: Iterable[str] = ()) -> None:
+    problems = [*request_leaks(tenants), *snapshot_problems(tenants, min_snapshots), *file_leaks(tenants, launch),
+                *extra]
+    assert not problems, "cross-tenant leak(s):\n  " + "\n  ".join(dict.fromkeys(problems))
+
+
+def text_leaks(label: str, text: str, tenants: dict[str, Tenant], owner: str | None = None) -> list[str]:
+    """Canaries of any tenant other than ``owner`` in a log/stream (``owner=None``: secrets only)."""
+    out: list[str] = []
+    for u in tenants.values():
+        if u.name == owner:
+            continue
+        values = u.canaries() if owner is not None else u.secrets()
+        out += [f"{label}: carries {u.name}'s {kind}" for kind, value in values.items() if value in text]
+    return out
+
+
+_SEED_CRON = ("import sys; from cron.jobs import create_job, trigger_job, list_jobs; "
+              "jobs = [j for j in list_jobs(include_disabled=True) if j.get('name') == 'tenancy-canary']; "
+              "j = jobs[0] if jobs else create_job(prompt=sys.argv[1], schedule='every 1h', name='tenancy-canary', "
+              "deliver='local'); trigger_job(j['id']); print(j['id'])")
+
+
+def seed_due_cron_job(t: Tenant, home: Path) -> str:
+    """Create (once) an hourly job in the tenant's own cron store and mark it due now (the state the
+    "run on next tick" action produces), so the multiplexed ticker's first tick fires it."""
+    r = subprocess.run([sys.executable, "-c", _SEED_CRON, t.cron_prompt], cwd=str(home),
+                       env=hermetic_env(home, {"HERMES_HOME": str(t.home)}), capture_output=True, text=True,
+                       timeout=120, stdin=subprocess.DEVNULL)
+    assert r.returncode == 0, r.stderr[-2000:]
+    return r.stdout.strip().splitlines()[-1]
+
+
+def cron_requests(t: Tenant) -> int:
+    """Provider requests driven by the tenant's cron job (its prompt canary is in the user turn)."""
+    assert t.srv is not None
+    return sum(t.cron_prompt.split(":")[0] in json.dumps(r["body"]) for r in list(t.srv.requests))
+
+
+# Processes -----------------------------------------------------------------------------------------
+
+
+def kill_group(proc: subprocess.Popen, sig: int = signal.SIGKILL) -> None:
+    try:
+        os.killpg(proc.pid, sig)
+    except (ProcessLookupError, PermissionError):
+        pass
+
+
+def run_hermes(argv: list[str], home: Path, timeout: float = 120.0,
+               extra_env: dict[str, str] | None = None) -> subprocess.CompletedProcess:
+    proc = subprocess.Popen(
+        [sys.executable, "-m", "hermes_cli.main", *argv], cwd=str(home), env=hermetic_env(home, extra_env),
+        stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
+        start_new_session=True,
+    )
+    try:
+        out, err = proc.communicate(timeout=timeout)
+    except subprocess.TimeoutExpired:
+        kill_group(proc)
+        out, err = proc.communicate()
+        err += f"\n[harness] killed after {timeout}s"
+    kill_group(proc)
+    return subprocess.CompletedProcess(argv, proc.returncode, out, err)
+
+
+class TuiBackend:
+    """The real stdio JSON-RPC backend the TUI/Desktop drive (``python -m tui_gateway.entry``)."""
+
+    def __init__(self, home: Path, log_path: Path, extra_env: dict[str, str] | None = None) -> None:
+        import queue
+        import threading
+
+        self._log = open(log_path, "a", encoding="utf-8")  # noqa: SIM115 - closed in close()
+        self.proc = subprocess.Popen(
+            [sys.executable, "-m", "tui_gateway.entry"], cwd=str(home), env=hermetic_env(home, extra_env),
+            stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=self._log, text=True, bufsize=1,
+            start_new_session=True,
+        )
+        self._q: queue.Queue[dict[str, Any]] = queue.Queue()
+        self._rid = 0
+        self.seen: list[dict[str, Any]] = []
+        threading.Thread(target=self._read, daemon=True, name="tenancy-tui-reader").start()
+
+    def _read(self) -> None:
+        assert self.proc.stdout is not None
+        for line in self.proc.stdout:
+            try:
+                self._q.put(json.loads(line))
+            except json.JSONDecodeError:
+                continue
+
+    def wait(self, pred: Callable[[dict[str, Any]], bool], timeout: float = 90.0) -> dict[str, Any]:
+        import queue
+
+        deadline = time.monotonic() + timeout
+        while time.monotonic() < deadline:
+            if self.proc.poll() is not None and self._q.empty():
+                break
+            try:
+                msg = self._q.get(timeout=0.25)
+            except queue.Empty:
+                continue
+            self.seen.append(msg)
+            if pred(msg):
+                return msg
+        tail = [(m.get("method"), (m.get("params") or {}).get("type"), m.get("id")) for m in self.seen[-20:]]
+        raise AssertionError(f"tui_gateway: condition not met within {timeout}s (rc={self.proc.poll()}); last {tail}")
+
+    def call(self, method: str, params: dict[str, Any], timeout: float = 90.0) -> dict[str, Any]:
+        self._rid += 1
+        rid = self._rid
+        assert self.proc.stdin is not None
+        self.proc.stdin.write(json.dumps({"jsonrpc": "2.0", "id": rid, "method": method, "params": params}) + "\n")
+        self.proc.stdin.flush()
+        return self.wait(lambda m: m.get("id") == rid and "method" not in m, timeout)
+
+    def ok(self, method: str, params: dict[str, Any], timeout: float = 90.0) -> dict[str, Any]:
+        reply = self.call(method, params, timeout)
+        assert "result" in reply, f"{method} failed: {reply.get('error')}"
+        return reply["result"]
+
+    def turn(self, sid: str, text: str, timeout: float = 90.0) -> dict[str, Any]:
+        """Submit one prompt and wait for that session's terminal turn event."""
+        self.ok("prompt.submit", {"session_id": sid, "text": text})
+
+        def done(m: dict[str, Any]) -> bool:
+            p = m.get("params") or {}
+            return (m.get("method") == "event" and p.get("session_id") == sid
+                    and p.get("type") in {"message.complete", "error"})
+        ev = self.wait(done, timeout)
+        assert ev["params"]["type"] == "message.complete", f"turn failed: {ev}"
+        return ev
+
+    def close(self) -> None:
+        try:
+            if self.proc.stdin:
+                self.proc.stdin.close()
+            self.proc.wait(timeout=60)
+        except (subprocess.TimeoutExpired, OSError, ValueError):
+            pass
+        kill_group(self.proc)
+        try:
+            self.proc.wait(timeout=10)
+        except subprocess.TimeoutExpired:
+            pass
+        self._log.close()
+
+
+class ServeBackend(TuiBackend):
+    """The real Desktop backend: ``hermes serve --port 0`` with the JSON-RPC surface on ``/api/ws``.
+
+    ``HERMES_DESKTOP=1`` marks it app-spawned, which is also what runs the cron ticker in-process.
+    """
+
+    def __init__(self, home: Path, log_path: Path, extra_env: dict[str, str] | None = None) -> None:
+        import queue
+        import re
+        import threading
+
+        from websockets.sync.client import connect  # ``websockets`` is a core dependency
+
+        self.token = secrets.token_urlsafe(24)
+        self._log = open(log_path, "a", encoding="utf-8")  # noqa: SIM115 - closed in close()
+        env = {"HERMES_DASHBOARD_SESSION_TOKEN": self.token, "HERMES_DESKTOP": "1", **(extra_env or {})}
+        self.proc = subprocess.Popen(
+            [sys.executable, "-m", "hermes_cli.main", "serve", "--host", "127.0.0.1", "--port", "0"],
+            cwd=str(home), env=hermetic_env(home, env), stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT, text=True, bufsize=1, start_new_session=True,
+        )
+        port_box: list[int] = []
+
+        def pump() -> None:
+            assert self.proc.stdout is not None
+            for line in self.proc.stdout:
+                self._log.write(line)
+                self._log.flush()
+                m = re.search(r"backend listening on [\d.]+:(\d+)", line)
+                if m and not port_box:
+                    port_box.append(int(m.group(1)))
+        threading.Thread(target=pump, daemon=True, name="tenancy-serve-stdout").start()
+        self.port = poll(lambda: port_box[0] if port_box else (self.proc.poll() is not None and -1), 120,
+                         "hermes serve to report its port")
+        assert self.port > 0, f"hermes serve exited rc={self.proc.returncode}"
+        self.ws = connect(f"ws://127.0.0.1:{self.port}/api/ws?token={self.token}", open_timeout=90, max_size=None)
+        self._q: queue.Queue[dict[str, Any]] = queue.Queue()
+        self._rid = 0
+        self.seen: list[dict[str, Any]] = []
+        threading.Thread(target=self._read, daemon=True, name="tenancy-ws-reader").start()
+
+    def _read(self) -> None:
+        while True:
+            try:
+                raw = self.ws.recv()
+            except Exception:
+                return
+            if not raw:
+                return
+            try:
+                self._q.put(json.loads(raw))
+            except json.JSONDecodeError:
+                continue
+
+    def call(self, method: str, params: dict[str, Any], timeout: float = 90.0) -> dict[str, Any]:
+        self._rid += 1
+        rid = self._rid
+        self.ws.send(json.dumps({"jsonrpc": "2.0", "id": rid, "method": method, "params": params}))
+        return self.wait(lambda m: m.get("id") == rid and "method" not in m, timeout)
+
+    def close(self) -> None:
+        try:
+            self.ws.close()
+        except Exception:
+            pass
+        kill_group(self.proc, signal.SIGTERM)
+        try:
+            self.proc.wait(timeout=60)
+        except subprocess.TimeoutExpired:
+            kill_group(self.proc)
+            self.proc.wait(timeout=10)
+        kill_group(self.proc)
+        self._log.close()
+
+
+def pid_alive(pid: int) -> bool:
+    try:
+        os.kill(pid, 0)
+    except ProcessLookupError:
+        return False
+    except PermissionError:
+        return True
+    stat = Path(f"/proc/{pid}/stat")
+    return not (stat.exists() and stat.read_text().split(") ", 1)[-1].startswith("Z"))
diff --git a/tests/e2e/core/tenancy/test_two_tenant_gateway.py b/tests/e2e/core/tenancy/test_two_tenant_gateway.py
new file mode 100644
index 0000000000..5d5de20d8a
--- /dev/null
+++ b/tests/e2e/core/tenancy/test_two_tenant_gateway.py
@@ -0,0 +1,201 @@
+"""C7 two-tenant canary: ONE multiplexed messaging gateway serving three profiles.
+
+Class: multiplex / tenancy isolation (issue_classes.md C7). Users hit it as "profile B's cron job ran
+with profile A's terminal settings", "my secondary profile answered with the launch profile's key",
+"deleting a profile killed the shared gateway", "a restart re-pinned one profile's env for all"
+(#89315, #101719, #105396, #111151, #102769, #107692).
+
+Harness: a real ``hermes gateway run`` child process with ``gateway.multiplex_profiles: true`` serving
+the launch profile ``default`` plus ``alpha`` and ``beta``. Each profile owns its own loopback provider
+(accepting only its own key), and a distinct value for EVERY per-profile knob: provider key (same env
+var NAME, different value), API-server key, a non-secret .env marker, model id, MEMORY.md, SOUL.md,
+terminal cwd, cron prompt, and a shell variable its tool call exports. Turns are driven through the
+real api_server platform (``/v1`` for the launch profile, ``/p/<name>/v1`` for the others); every
+turn runs ``env | sort; pwd`` in the terminal tool, so the provider receives an env snapshot taken
+INSIDE the tool subprocess.
+
+Phases (same invariant after each): interleaved turns + a cron tick firing one job per profile ->
+cross-profile API keys -> create/attach/delete a fourth profile (host PID must not change) ->
+SIGTERM restart + a second cron tick. Invariant (``check_isolation``): every request a provider
+recorded carries only its own tenant's key/model and no other tenant's canary; every env snapshot ran
+in its own cwd and shows no other tenant's canary; no file in any profile home (state.db + WAL, logs,
+sessions, cron output, config, .env) contains another tenant's canary; the gateway's own output
+never contains any tenant's secret.
+"""
+
+from __future__ import annotations
+
+import json
+import signal
+import subprocess
+import sys
+import urllib.error
+import urllib.request
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from . import _helpers as H
+
+# The gateway child is spawned and reaped by this module (fake HOME, no systemd bus in its env).
+pytestmark = pytest.mark.spawns_gateway_lookalike
+
+NAMES = ("default", "alpha", "beta")
+
+
+class MultiplexGateway:
+    def __init__(self, root: Path, port: int) -> None:
+        self.root, self.home, self.port = root, root / "home", port
+        self.log_path = root / "gateway.log"
+        self.proc: subprocess.Popen | None = None
+        self.pids: list[int] = []
+
+    def start(self) -> None:
+        log = open(self.log_path, "a", encoding="utf-8")  # noqa: SIM115 - handed to the child
+        self.proc = subprocess.Popen(
+            [sys.executable, "-m", "hermes_cli.main", "gateway", "run"], cwd=str(self.home),
+            env=H.hermetic_env(self.home), stdin=subprocess.DEVNULL, stdout=log, stderr=subprocess.STDOUT,
+            start_new_session=True,
+        )
+        log.close()
+        self.pids.append(self.proc.pid)
+        H.poll(lambda: self.healthy() or self._died(), 120, "multiplexed gateway /health")
+        assert self.proc.poll() is None, f"gateway exited rc={self.proc.returncode}: {self.tail()}"
+
+    def _died(self) -> bool:
+        return self.proc is not None and self.proc.poll() is not None
+
+    def healthy(self) -> bool:
+        try:
+            with urllib.request.urlopen(f"http://127.0.0.1:{self.port}/health", timeout=3) as r:
+                return r.status == 200
+        except (OSError, urllib.error.URLError):
+            return False
+
+    def chat(self, prefix: str, api_key: str, text: str) -> tuple[int, Any]:
+        req = urllib.request.Request(
+            f"http://127.0.0.1:{self.port}{prefix}/v1/chat/completions",
+            data=json.dumps({"model": "hermes", "messages": [{"role": "user", "content": text}]}).encode(),
+            headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
+        )
+        try:
+            with urllib.request.urlopen(req, timeout=90) as r:
+                return r.status, json.loads(r.read())
+        except urllib.error.HTTPError as e:
+            return e.code, e.read()[:500]
+
+    def stop(self) -> None:
+        if self.proc is None:
+            return
+        H.kill_group(self.proc, signal.SIGTERM)
+        try:
+            self.proc.wait(timeout=90)
+        except subprocess.TimeoutExpired:
+            H.kill_group(self.proc)
+            self.proc.wait(timeout=10)
+        H.kill_group(self.proc)
+
+    def tail(self, n: int = 3000) -> str:
+        return self.log_path.read_text(encoding="utf-8", errors="replace")[-n:] if self.log_path.exists() else ""
+
+
+def prefix(name: str) -> str:
+    return "" if name == "default" else f"/p/{name}"
+
+
+@pytest.fixture(scope="module")
+def fleet(tmp_path_factory: pytest.TempPathFactory):
+    root = tmp_path_factory.mktemp("c7-gateway")
+    home = root / "home"
+    tenants = H.make_tenants(root, NAMES)
+    port = H.free_port()
+    for t in tenants.values():
+        assert t.srv is not None
+        t.srv.start()
+    # The profile root is HOME-anchored: prove it is inside the sandbox before writing any profile.
+    home.mkdir(parents=True)
+    H.assert_profiles_root_under(root, home)
+    for t in tenants.values():
+        extra = {} if t.name != "default" else {
+            "gateway": {"multiplex_profiles": True},
+            "platforms": {"api_server": {"enabled": True, "extra": {"host": "127.0.0.1", "port": port}}},
+        }
+        H.write_tenant_home(t, extra)
+    gw = MultiplexGateway(root, port)
+    try:
+        yield root, tenants, gw
+    finally:
+        gw.stop()
+        for pid in gw.pids:
+            assert not H.pid_alive(pid), f"gateway pid {pid} survived teardown"
+        for t in tenants.values():
+            assert t.srv is not None
+            t.srv.stop()
+
+
+def _turns(gw: MultiplexGateway, tenants: dict[str, H.Tenant], order: list[str]) -> None:
+    for name in order:
+        t = tenants[name]
+        before = len(t.srv.requests)  # type: ignore[union-attr]
+        status, body = gw.chat(prefix(name), t.api_server_key, f"turn for {name}")
+        assert status == 200, f"{name} turn via api_server failed: {status} {body!r}\n{gw.tail()}"
+        answer = body["choices"][0]["message"]["content"]
+        assert f"probe done for {name}" in answer, f"{name} was answered by someone else: {answer!r}"
+        assert len(t.srv.requests) >= before + 2, f"{name}'s turn never reached its own provider"  # type: ignore[union-attr]
+
+
+def _await_cron(tenants: dict[str, H.Tenant], at_least: int) -> None:
+    # tool call + follow-up per fire; the ticker's first tick runs right after boot.
+    H.poll(lambda: all(H.cron_requests(t) >= 2 * at_least for t in tenants.values()), 150,
+           f"cron fire #{at_least} in every profile")
+
+
+def test_multiplexed_gateway_never_crosses_tenants(fleet) -> None:
+    root, tenants, gw = fleet
+    home = root / "home"
+
+    # Phase 1: boot with one due cron job per profile; interleave turns across all three.
+    for t in tenants.values():
+        H.seed_due_cron_job(t, home)
+    gw.start()
+    host_pid = gw.proc.pid  # type: ignore[union-attr]
+    _turns(gw, tenants, ["alpha", "default", "beta", "alpha", "beta", "default"])
+    _await_cron(tenants, 1)
+    H.check_isolation(tenants, min_snapshots=3)
+
+    # Phase 2: each profile's API-server key is honoured only on its own route.
+    for owner in tenants.values():
+        for route in tenants.values():
+            if route is owner:
+                continue
+            seen = len(route.srv.requests)  # type: ignore[union-attr]
+            status, _ = gw.chat(prefix(route.name), owner.api_server_key, "cross-tenant key")
+            assert status in (401, 403), f"{owner.name}'s API key opened {route.name}'s route (HTTP {status})"
+            assert len(route.srv.requests) == seen, f"{owner.name}'s key drove {route.name}'s provider"  # type: ignore[union-attr]
+
+    # Phase 3: profile churn next to the live host never restarts or kills it.
+    created = H.run_hermes(["profile", "create", "gamma"], home)
+    assert created.returncode == 0, created.stderr[-2000:]
+    attach = H.run_hermes(["-p", "gamma", "gateway", "run"], home, timeout=150)
+    assert "[harness] killed" not in attach.stderr, "`-p gamma gateway run` started a second gateway"
+    assert gw.proc.pid == host_pid and gw.proc.poll() is None, "profile create/attach replaced the host"  # type: ignore[union-attr]
+    deleted = H.run_hermes(["profile", "delete", "gamma", "--yes"], home)
+    assert deleted.returncode == 0, deleted.stderr[-2000:]
+    assert gw.proc.pid == host_pid and gw.proc.poll() is None, f"profile delete killed the host\n{gw.tail()}"  # type: ignore[union-attr]
+    _turns(gw, tenants, ["alpha", "beta"])
+    H.check_isolation(tenants, min_snapshots=3)
+
+    # Phase 4: SIGTERM restart; the new host re-serves all three and ticks each profile's cron again.
+    gw.stop()
+    assert not H.pid_alive(host_pid), "old gateway still alive after SIGTERM"
+    for t in tenants.values():
+        H.seed_due_cron_job(t, home)
+    gw.start()
+    _turns(gw, tenants, ["beta", "default", "alpha"])
+    _await_cron(tenants, 2)
+    H.check_isolation(tenants, min_snapshots=5)
+
+    gw.stop()
+    leaks = H.text_leaks("gateway output", gw.tail(10**9), tenants)
+    assert not leaks, "\n".join(leaks)