AI plugin 0.4.0: machine pools over llama.cpp RPC, behind a private link (#121)

ddiff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
iindex bfbb0bf..5c7f5d8 100644
---- a/.github/workflows/ci.yml
++++ b/.github/workflows/ci.yml
@@@ -126,6 +126,8 @@ jobs:
        - run: python -m pytest -q plugins/ros2/tests
        - run: python -m pytest -q plugins/computer/tests
        - run: python -m pytest -q plugins/voice/tests
++      # The machine-pool tests run llama.cpp RPC's stand-ins in their own user and network namespaces.
++      - run: sudo sysctl -w kernel.apparmor_restrict_unprivileged_userns=0 || true
        - run: python -m pytest -q plugins/ai/tests
        - name: packages install
          run: pip install ./plugins/sdk ./plugins/dryrun ./plugins/adeept_adr036 ./plugins/cozmo ./plugins/relay ./plugins/ros2 ./plugins/computer ./plugins/voice ./plugins/ai
ddiff --git a/.github/workflows/pool.yml b/.github/workflows/pool.yml
nnew file mode 100644
iindex 0000000..9ca055e
---- /dev/null
++++ b/.github/workflows/pool.yml
@@@ -0,0 +1,31 @@
++name: Pool security and real RPC
++on:
++  pull_request:
++    paths: ['plugins/ai/**', '.github/workflows/pool.yml']
++  push:
++    branches: [main]
++    paths: ['plugins/ai/**', '.github/workflows/pool.yml']
++permissions:
++  contents: read
++jobs:
++  pool:
++    runs-on: ubuntu-22.04
++    timeout-minutes: 30
++    steps:
++      - uses: actions/checkout@v4
++      - name: Build the pinned CPU RPC engine
++        run: |
++          curl -fsSL https://codeload.github.com/ggml-org/llama.cpp/tar.gz/bcb0152876e8f0b684ba70d3defcab6cdf35adb5 -o /tmp/llama.tar.gz
++          tar -xzf /tmp/llama.tar.gz -C /tmp
++          cmake -S /tmp/llama.cpp-bcb0152876e8f0b684ba70d3defcab6cdf35adb5 -B /tmp/pool-build -DGGML_RPC=ON -DLLAMA_CURL=OFF -DGGML_NATIVE=OFF -DLLAMA_BUILD_TESTS=OFF
++          cmake --build /tmp/pool-build --target ggml-rpc-server llama-server -j2
++          curl -fsSL https://huggingface.co/ggml-org/models/resolve/main/tinyllamas/stories15M-q4_0.gguf -o /tmp/pool-model.gguf
++          echo "66967fbece6dbe97886593fdbb73589584927e29119ec31f08090732d1861739  /tmp/pool-model.gguf" | sha256sum -c -
++      - run: sudo sysctl -w kernel.apparmor_restrict_unprivileged_userns=0 || true
++      - name: Security tests and two-member real inference
++        env:
++          PYTHONPATH: plugins/sdk:plugins/ai
++          POOL_RPC_BINARY: /tmp/pool-build/bin/ggml-rpc-server
++          POOL_LLAMA_BINARY: /tmp/pool-build/bin/llama-server
++          POOL_MODEL: /tmp/pool-model.gguf
++        run: python3 -m unittest discover -s plugins/ai/tests -p 'test_pool*.py' -v
ddiff --git a/docs/ai.md b/docs/ai.md
iindex 2ac0a28..b14f7e6 100644
---- a/docs/ai.md
++++ b/docs/ai.md
@@@ -142,3 +142,98 @@ When the machine is off or its model server is down, the request fails and says
    how fast the last model answered here (`ai_model`, `ai_tokens_per_s`): the reply's tokens over the whole answer's
    time, averaged with the earlier answers, the latest counting most. openvibe.bot/machines shows it beside the models.
  - It never acts on the machine, so it also runs while the machine's stop is latched.
++
++## Machine pools (Linux)
++
++Several of your machines on the same local network can serve one model together, as one offer: a gaming PC and a
++laptop, or a few servers in one rack (the pooled machines design, 2026-11-06). The engine is llama.cpp's RPC backend:
++`rpc-server` on every member and `llama-server --rpc` on the head, which sees the prompt and answers; the members
++lend their memory and only receive tensors. It runs on any GPU llama.cpp supports, and on the CPU. Pools across the
++internet, and pools of machines that are not all yours, are not possible.
++
++llama.cpp's RPC has no authentication of its own, so the plugin never exposes it:
++
++- **The engines run in their own network namespace** (an unprivileged Linux user and network namespace with only
++  loopback). `rpc-server` and `llama-server` cannot be reached from the local network or from the host's own
++  loopback; a private Unix socket (mode 0600) is their only way in or out. The worker refuses to start on the host's
++  network, and it dies with the plugin.
++- **The private link** is the only door to a member's `rpc-server`: TLS 1.3 on the member's private address (an RFC
++  1918 address configured on one of its interfaces; never 0.0.0.0, loopback or a public address), with mutual
++  certificate pinning. A member accepts only the pool head's certificate; the head accepts only each member's. Each
++  machine makes its own Ed25519 key when the person at it approves the pool; only the public certificate goes to
++  Network. The link carries tensors, status, run and pause, and the link probe, nothing else: no machine can run
++  anything on another.
++- **Leaving drops the link and the key at once**: `pool_cli leave`, the owner's Stop pool on openvibe.network,
++  removing a machine, signing it out or pairing it to someone else. Every machine renews its readiness with Network
++  every two seconds and sees the change there; the pool needs every machine to approve again.
++- **Idle-only and the kill switch hold for the whole pool.** The head asks every member every half second; any member
++  that is in use (a game, OBS, the person at the keyboard, another program on the GPU; the pool's own GPU use does not
++  count), outside its hours, offline or stopped by its kill switch pauses the whole pool, and a member the head stops
++  calling pauses itself after three seconds. The kill switch (an emergency stop of any machine) holds the pool until
++  its latch is cleared. Network marks the pool's offer not ready when any machine misses a renewal for ten seconds.
++
++### Setting one up
++
++You need Linux on every machine, `bubblewrap` (recommended) or `unshare` and `ip` (util-linux and iproute2), OpenSSL,
++and llama.cpp built with `GGML_RPC=ON` (the tested revision is `bcb0152876e8f0b684ba70d3defcab6cdf35adb5`; its RPC
++server is `ggml-rpc-server`, `rpc-server` in older releases). Each machine must be paired through OpenVibe.Network.
++The engine runs in a private network namespace: bubblewrap is tried first and works on a default Ubuntu 24.04 even
++though `kernel.apparmor_restrict_unprivileged_userns=1`; without bubblewrap, `unshare` needs unprivileged user
++namespaces allowed (`kernel.apparmor_restrict_unprivileged_userns=0` on Ubuntu 24.04). If neither can make a
++namespace, pooling refuses instead of using the host network and says: "Pooling needs an isolated network for the
++engine. Install bubblewrap (sudo apt install bubblewrap) and try again."
++
++1. On every machine, give the ai plugin a pool configuration:
++
++   ```json
++   {
++     "pool_directory": "/home/me/.local/share/openvibe-node/pool",
++     "pool_credential_file": "/home/me/.config/openvibe-node/credential.json",
++     "pool_address": "192.168.1.2/24",
++     "pool_engine": {
++       "rpc_binary": "/opt/llama.cpp/bin/ggml-rpc-server",
++       "llama_binary": "/opt/llama.cpp/bin/llama-server",
++       "model_path": "/models/llama-3.3-70b-instruct-q4_k_m.gguf",
++       "context": 4096,
++       "model": {"model": "llama-3.3-70b-instruct-q4_k_m", "parameters_b": 70, "memory_mb": 45000}
++     },
++     "offer": {"enabled": true, "price": {"in_per_million": 0.2, "out_per_million": 0.6}}
++   }
++   ```
++
++   `pool_address` defaults to the address of the default route; `pool_port` defaults to 50053. `memory_mb` is what the
++   model needs, weights and context. Every machine announces itself to Network every minute until it is in a pool
++   (or once by hand: `python -m openvibe_ai.pool_cli announce --credential-file … --directory …`).
++2. On openvibe.network/machines, "Pool these machines" lists your machines that checked in from the same local network
++   (one public address, one private subnet). Choose two to eight; the head defaults to the one with the most free
++   VRAM.
++3. On **each** machine, approve it:
++
++   ```sh
++   python -m openvibe_ai.pool_cli approve --credential-file /home/me/.config/openvibe-node/credential.json \
++       --directory /home/me/.local/share/openvibe-node/pool --pool pool_…
++   ```
++
++   It shows the owner, the head and the members, says what the head sees, and needs the word `approve`. The key stays
++   on the machine; its certificate is valid for 30 days, after which the pool is made again.
++4. When every machine has approved, the head measures each link (three round trips under 5 ms; at least 300 Mb/s, or
++   1 Gb/s for a model over 30B) and checks the pool's free memory against the model, then starts. The offer appears
++   as one offer, labelled "Community pool (N machines)", with one price for the pool. Before choosing a model,
++   `pool_cli fit --models models.json [--benchmarks measured.json]` on the head prints the links and what fits; a speed
++   is shown only from a dated, sourced measurement on the same hardware, never a guess.
++
++A pool is never edited: another head, member or key is a new pool, approved again on every machine. Leave from any
++machine, even offline: `python -m openvibe_ai.pool_cli leave --credential-file … --directory … --pool pool_…` deletes
++the key first, then tells Network, which stops the pool for every machine.
++
++### Tests
++
++`plugins/ai/tests/test_pool.py` checks the trust boundary with real TLS and real keys (only the head reaches
++`rpc-server`, another member or a stranger is refused, plain connections are refused, no wildcard, loopback or public
++listener, the probe refuses a machine off the network), the pool rules on two machines (the head starts only after
++every member runs; idle-only on a member or the head pauses everyone; the kill switch stops the pool at once; a
++silent head pauses its members; leaving or a revoked membership closes the link and deletes the key; closing keeps
++it) and the plugin's stops. `test_pool_linux.py` runs the engines in real namespaces: the worker refuses the host
++network, `rpc-server` is reachable only through its private socket, and two machines on a throwaway network pool and
++serve through the tunnel with stand-in engines. The pool workflow (`.github/workflows/pool.yml`) builds llama.cpp at
++the pinned revision and runs the same two-machine scenario with a real model.
ddiff --git a/plugins/ai/openvibe_ai/__init__.py b/plugins/ai/openvibe_ai/__init__.py
iindex 8356291..91b7b73 100644
---- a/plugins/ai/openvibe_ai/__init__.py
++++ b/plugins/ai/openvibe_ai/__init__.py
@@@ -40,6 +40,15 @@ Tools (contracts 0.277.0, plugin 0.2.0): the functions the model may ask for, pa
  The plugin never runs one: it hands back the calls the model asked for, and the caller (an agent on OpenVibe) runs
  them and sends the next command with the assistant's message (its ``tool_calls``) and one ``tool`` message per call.
  ``describe`` says ``tools: true`` so a caller knows this machine can.
++
++Machine pools (plugin 0.4.0, Linux; the pooled machines design, 2026-11-06): with ``pool_directory`` set, this machine
++serves one model together with other machines of the same owner on the same local network, through llama.cpp RPC
++(``rpc-server`` on the members, ``llama-server --rpc`` on the head). The pool is proposed on openvibe.network/machines
++and approved on each machine with ``python -m openvibe_ai.pool_cli approve``; until then the plugin announces the
++machine every minute (``pool_credential_file``, ``pool_address``, ``pool_port``). ``pool_engine`` names the engine:
++``{rpc_binary, llama_binary, model_path, model: {model, parameters_b, memory_mb}, context?}``. The RPC engine is
++reachable only through the pool's private, mutually authenticated link and only from the head; idle-only and the kill
++switch on any machine pause the whole pool; leaving the pool drops the link and the key (docs/ai.md, "Machine pools").
  """
  
  import datetime
@@@ -56,7 +65,7 @@ import urllib.request
  
  from openvibe_plugin import Deferred, Fault, Plugin
  
--VERSION = "0.3.0"
++VERSION = "0.4.0"
  DEFAULT_ENDPOINTS = ("http://127.0.0.1:11434", "http://127.0.0.1:1234", "http://127.0.0.1:8080")
  MAX_REPLY_CHARS = 200 * 1024     # well inside the core's answer limit (512 KB)
  MAX_MESSAGES = 200
@@@ -224,6 +233,39 @@ def _offer_config(cfg, served):
      }
  
  
++def _pool_engine(raw):
++    """The pool's engine (pool_engine in config): llama.cpp's rpc-server and llama-server built with GGML_RPC, the
++    model file, its name, size and the memory it needs (weights and context), and the context length."""
++    if not isinstance(raw, dict):
++        raise Fault("bad_value", "pool_engine is {rpc_binary, llama_binary, model_path, model: {model, parameters_b, memory_mb}, context?}")
++    for key in ("rpc_binary", "llama_binary", "model_path"):
++        if not isinstance(raw.get(key), str) or not raw[key] or len(raw[key]) > 1024 or not os.path.isabs(raw[key]):
++            raise Fault("bad_value", "pool_engine.%s is an absolute path on this machine" % key)
++    model = raw.get("model")
++    if not isinstance(model, dict) or not isinstance(model.get("model"), str) or not 0 < len(model["model"]) <= 200:
++        raise Fault("bad_value", "pool_engine.model.model is the model's name")
++    size = model.get("parameters_b")
++    if isinstance(size, bool) or not isinstance(size, (int, float)) or not 0 < size <= 4000:
++        raise Fault("bad_value", "pool_engine.model.parameters_b is the model's size in billions of parameters")
++    memory = _int(model.get("memory_mb"), "pool_engine.model.memory_mb", 1, 16 * 1024 * 1024, None)
++    if memory is None:
++        raise Fault("bad_value", "pool_engine.model.memory_mb is the memory the model needs, weights and context")
++    return {"rpc_binary": raw["rpc_binary"], "llama_binary": raw["llama_binary"], "model_path": raw["model_path"],
++            "context": _int(raw.get("context"), "pool_engine.context", 256, 131072, 2048),
++            "model": {"model": model["model"], "parameters_b": float(size), "memory_mb": memory}}
++
++
++def _gpu_compute_pids():
++    """The processes on the NVIDIA GPUs (nvidia-smi), or None when they cannot be listed."""
++    if not shutil.which("nvidia-smi"):
++        return None
++    out = _run(["nvidia-smi", "--query-compute-apps=pid", "--format=csv,noheader,nounits"])
++    try:
++        return {int(x) for x in out.split()}
++    except ValueError:
++        return None
++
++
  def _minutes(hhmm):
      return int(hhmm[:2]) * 60 + int(hhmm[3:])
  
@@@ -295,6 +337,7 @@ class Ai(Plugin):
  
      def __init__(self, opener=None, clock=time.monotonic, now=None, busy_probe=None):
          self.opener = opener or urllib.request.urlopen
++        self.default_opener = self.opener
          self.clock = clock
          self.now = now or (lambda: datetime.datetime.now(datetime.timezone.utc))
          # Injected in tests: returns (busy, reason). In production this reads GPU use, heavy processes and input idle.
@@@ -320,6 +363,15 @@ class Ai(Plugin):
          # What this machine served to the network, as sensors and for the /sell view: tokens counted by the local
          # server's usage, never the prompt or the answer.
          self.community = {"requests": 0, "tokens_in": 0, "tokens_out": 0}
++        # A machine pool (pool_directory in config): its runtime once every machine approved, the thread that waits
++        # for that, the kill switch latch, and the last problem, shown in describe.
++        self.pool_runtime = None
++        self.pool_thread = None
++        self.pool_stopped = threading.Event()
++        self.pool_killed = threading.Event()
++        self.pool_allow_start = False
++        self.pool_cfg = None
++        self.pool_error = None
  
      # ---- the model server -----------------------------------------------------------------------------------------
      def _get(self, url, timeout):
@@@ -343,6 +395,9 @@ class Ai(Plugin):
          self.cfg = cfg
          self.max_tokens = _int(cfg.get("max_tokens"), "max_tokens", 1, 32768, 2048)
          self.timeout_s = _int(cfg.get("timeout_s"), "timeout_s", 5, 900, 180)
++        if cfg.get("pool_directory"):
++            self._configure_pool(cfg)
++            return
          endpoint = cfg.get("endpoint")
          if endpoint is not None and (not isinstance(endpoint, str) or not endpoint.startswith(("http://", "https://"))):
              raise Fault("bad_value", "endpoint is the model server's http(s) base, for example http://127.0.0.1:11434")
@@@ -376,7 +431,7 @@ class Ai(Plugin):
          if self.busy_probe is not None:
              busy, reason = self.busy_probe()
          else:
--            if _gpu_busy(self.offer["busy_gpu_percent"]):
++            if _gpu_busy(self.offer["busy_gpu_percent"]) and not self._gpu_is_ours():
                  busy, reason = True, "gpu"
              elif _process_busy(self.offer["busy_processes"]):
                  busy, reason = True, "process"
@@@ -429,7 +484,7 @@ class Ai(Plugin):
          """The offer as Bot, Network and AI see it (status.capabilities.ai.offer)."""
          open_now = self._within_hours()
          paused = self._paused() if self.offer["enabled"] else False
--        return {
++        view = {
              "enabled": self.offer["enabled"],
              "models": list(self.offer["models"]),
              "hours": dict(self.offer["hours"]),
@@@ -445,24 +500,176 @@ class Ai(Plugin):
              "privacy": "A community machine sees the prompt. Only requests that allow community capacity are sent here, and every answer is labelled Community.",
              "ready": bool(self.offer["enabled"] and self.reachable and self.offer["models"] and open_now and not paused),
          }
++        if self.cfg.get("pool_directory"):
++            # A pool's offer is the head's, ready only while the whole pool runs (Network adds the pool from its own
++            # record; this is what the machine itself sees). A member never offers on its own while it is in a pool.
++            runtime = self.pool_runtime
++            if runtime:
++                view["pool"] = {"machines": sorted(m["id"] for m in runtime.doc["machines"]), "head": runtime.doc["head"]}
++            view["ready"] = bool(view["ready"] and self._pool_serving())
++        return view
++
++    # ---- a machine pool (pool.py, pool_runtime.py, pool_registry.py) ---------------------------------------------
++    def _configure_pool(self, cfg):
++        """Pool mode (``pool_directory`` set): this machine serves one model together with the other machines of its
++        pool. The model server is the pool's own ``llama-server`` on the head, reached through a private Unix socket;
++        the offer's settings apply to the whole pool. Nothing starts before setup, and nothing runs until every machine
++        has approved the pool on itself."""
++        from pathlib import Path
++        if sys.platform != "linux":
++            raise Fault("unsupported", "machine pools run on Linux only: the RPC engine must be isolated in its own network namespace")
++        engine = _pool_engine(cfg.get("pool_engine"))
++        directory = Path(cfg["pool_directory"])
++        if not directory.is_absolute():
++            raise Fault("bad_value", "pool_directory is an absolute path on this machine")
++        address = cfg.get("pool_address")
++        if address is not None and not isinstance(address, str):
++            raise Fault("bad_value", "pool_address is this machine's private address and prefix, such as 192.168.1.2/24")
++        self.pool_cfg = {"directory": directory, "engine": engine, "address": address,
++                         "port": _int(cfg.get("pool_port"), "pool_port", 1024, 65535, 50053)}
++        self.models = [engine["model"]["model"]]
++        self.offer = _offer_config(cfg, self.models)
++        self.endpoint = "http://localhost"   # the head's llama-server, through the pool's private Unix socket
++        self.reachable = self._pool_serving()
++        if not self.pool_allow_start or self.pool_thread or self.pool_stopped.is_set():
++            return
++        self.pool_thread = threading.Thread(target=self._pool_loop, name="ai-pool", daemon=True)
++        self.pool_thread.start()
++
++    def _pool_serving(self):
++        """This machine is the head of a running, unpaused pool."""
++        runtime = self.pool_runtime
++        return bool(runtime and runtime.machine == runtime.doc["head"] and not runtime.paused and not runtime.stopped.is_set())
++
++    def _pool_ready(self):
++        """What this machine renews on Network: its part of the pool is running and not paused."""
++        runtime = self.pool_runtime
++        return bool(runtime and not runtime.paused and runtime.status()["ready"])
++
++    def _pool_busy(self):
++        """Idle-only and the kill switch for a pool member: the kill switch latched, the offer off or outside its hours,
++        or the machine in use. Any member busy pauses the whole pool (the head asks every member every half second)."""
++        if self.pool_killed.is_set():
++            return True
++        if not self.offer["enabled"] or not self._within_hours():
++            return True
++        return self._paused()
++
++    def _gpu_is_ours(self):
++        """While this machine runs its pool's engine, the GPU's use is the pool's own, unless another program is on it
++        (NVIDIA lists the processes; elsewhere the listed busy processes and the keyboard still count)."""
++        runtime = self.pool_runtime
++        ours = runtime.engine_pids() if runtime else set()
++        if not ours:
++            return False
++        others = _gpu_compute_pids()
++        return others is None or not (others - ours)
++
++    def _pool_loop(self):
++        """Announce this machine until a pool is approved here, then run it. A pool this machine left (revoked,
++        removed, signed out, another owner) ends here: its key is gone and the machine announces itself again."""
++        from .pool_registry import Membership, Registry, local_address
++        from .pool_runtime import PoolRuntime
++        cfg = self.pool_cfg
++        directory, approval_file = cfg["directory"], cfg["directory"] / "approval.json"
++        announced = 0.0
++        while not self.pool_stopped.is_set():
++            try:
++                if not approval_file.exists():
++                    # Not approved here: stay on the owner's "Pool these machines" list (Network forgets after 5 min).
++                    if self.cfg.get("pool_credential_file") and time.monotonic() - announced > 60:
++                        from .pool_cli import memory
++                        registry = Registry(self.cfg["pool_credential_file"])
++                        registry.announce(local_address(cfg["address"]), cfg["port"], memory())
++                        announced = time.monotonic()
++                    self.pool_stopped.wait(5)
++                    continue
++                approval = json.loads(approval_file.read_text())
++                key = directory / "identity" / "key.pem"
++                registry = Registry(approval["credential_file"])
++                membership = Membership(registry, approval["pool_id"], key, ready=self._pool_ready)
++                doc = membership()
++                if doc is None or doc["owner"] != approval["owner"] or doc["head"] != approval["head"] or \
++                        sorted(m["id"] for m in doc["machines"]) != approval["machines"]:
++                    # Left, revoked or changed: drop the key and the approval at once; a new pool needs a new approval.
++                    shutil.rmtree(str(directory / "identity"), ignore_errors=True)
++                    approval_file.unlink(missing_ok=True)
++                    self.pool_error = "this machine is no longer in pool %s" % approval["pool_id"]
++                    continue
++                machine = registry.credentials()["device_id"]
++                # Raises ValueError until every machine has approved on itself.
++                runtime = PoolRuntime(doc, machine, directory, cfg["engine"], self._pool_busy, membership)
++                if self.pool_stopped.is_set():
++                    return
++                self.pool_runtime = runtime
++                self.opener = runtime.opener
++                runtime.start()
++                self.pool_error = None
++                while not runtime.stopped.wait(1):
++                    self.pool_error = runtime.last_error
++                if self.pool_stopped.is_set():
++                    return
++                # The runtime left the pool (its key is deleted): forget it and announce again.
++                self.pool_runtime = None
++                self.opener = self.default_opener
++                approval_file.unlink(missing_ok=True)
++            except (OSError, KeyError, ValueError, TypeError) as error:
++                self.pool_error = "%s: %s" % (type(error).__name__, str(error)[:200])
++                self.pool_stopped.wait(2)
++
++    def on_stop(self, info):
++        """The kill switch on this machine (an emergency stop) stops the whole pool now and holds it until the latch is
++        cleared; the Node stopping closes it and keeps the key. A turn's halt, a lost link or a command's timeout is not
++        about the pool. Without a pool, every stop is as before (nothing moves here)."""
++        if not self.cfg.get("pool_directory"):
++            return Plugin.on_stop(self, info)
++        if info.emergency:
++            self.pool_killed.set()
++            runtime = self.pool_runtime
++            if runtime:
++                runtime.halt()
++        elif info.cause == "shutdown":
++            self.close()
++
++    def on_resume(self, info):
++        """The stop latch was cleared by its owner: the pool may run again, after every member's own checks."""
++        self.pool_killed.clear()
++
++    def close(self):
++        """The plugin is ending: stop the pool's engines and its private link now; the key stays for the next start."""
++        self.pool_stopped.set()
++        runtime = self.pool_runtime
++        if runtime:
++            runtime.close()
  
      def describe(self, config):
          try:
              self._configure(config)
          except Fault:
              return {}
--        return {"ai": {"models": self.models, "ready": self.reachable and bool(self.models),
--                       "max_tokens": self.max_tokens, "tools": True,
--                       "speed": {m: round(r, 1) for m, r in self.speed.items()},
--                       "offer": self._offer_view()}}
++        out = {"ai": {"models": self.models, "ready": self.reachable and bool(self.models),
++                      "max_tokens": self.max_tokens, "tools": True,
++                      "speed": {m: round(r, 1) for m, r in self.speed.items()},
++                      "offer": self._offer_view()}}
++        if self.cfg.get("pool_directory"):
++            runtime = self.pool_runtime
++            out["ai"]["pool"] = {"id": runtime.doc["id"] if runtime else None,
++                                 "role": (("head" if runtime.machine == runtime.doc["head"] else "member") if runtime else None),
++                                 "state": ("stopped" if self.pool_killed.is_set() else "waiting" if not runtime
++                                           else "serving" if not runtime.paused else "paused"),
++                                 "problem": (self.pool_error or "")[:200] or None}
++        return out
  
      def setup(self, config):
++        self.pool_allow_start = True
          self._configure(config)
          if self.endpoint is None:
              self.log("no model server answers yet (Ollama, LM Studio or llama.cpp's server, or the endpoint in config)")
  
      # ---- one reply --------------------------------------------------------------------------------------------------
      def handle(self, cmd):
++        if self.cfg.get("pool_directory") and not self._pool_serving():
++            raise Fault("not_ready", "this machine's pool is paused, waiting for every machine to approve, or served by its head")
          if cmd.kind != "ai":
              raise Fault("unsupported", "the ai plugin takes ai commands only")
          v = cmd.value if isinstance(cmd.value, dict) else {}
ddiff --git a/plugins/ai/openvibe_ai/pool.py b/plugins/ai/openvibe_ai/pool.py
nnew file mode 100644
iindex 0000000..e9cae81
---- /dev/null
++++ b/plugins/ai/openvibe_ai/pool.py
@@@ -0,0 +1,282 @@
++"""Private RPC transport. The TLS listener never launches a command supplied by a peer.
++
++Raw RPC lives in a separate Linux user/network namespace with only loopback. A
++mode-0700 directory holds the Unix bridge and the locally generated identity.
++Only the pinned, locally approved head certificate can enter a member's tunnel.
++"""
++import hashlib
++import ipaddress
++import json
++import math
++import os
++from pathlib import Path
++import select
++import shutil
++import socket
++import ssl
++import subprocess
++import threading
++import time
++
++PRIVATE = tuple(ipaddress.ip_network(x) for x in ("10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16"))
++PROBE_BYTES = 16 * 1024 * 1024
++
++
++def lan_address(value):
++    address = ipaddress.ip_address(value)
++    if not any(address in net for net in PRIVATE):
++        raise ValueError("pool links require a literal RFC1918 address")
++    return str(address)
++
++
++def same_network(a, b):
++    try:
++        left, right = ipaddress.ip_interface(a["address"]), ipaddress.ip_interface(b["address"])
++        lan_address(str(left.ip))
++        lan_address(str(right.ip))
++        return (a["public_ip"] == b["public_ip"] and bool(a["public_ip"]) and
++                left.network == right.network and left.ip != right.ip)
++    except (KeyError, ValueError, TypeError):
++        return False
++
++
++def validate_pool(doc):
++    members = doc.get("machines", [])
++    ids = [m.get("id") for m in members]
++    if not 2 <= len(members) <= 8 or len(set(ids)) != len(ids) or doc.get("head") not in ids:
++        raise ValueError("a pool needs 2-8 distinct machines and a member head")
++    if not doc.get("owner") or any(m.get("owner") != doc["owner"] for m in members):
++        raise ValueError("all machines must belong to the pool owner")
++    head = next(m for m in members if m["id"] == doc["head"])
++    for member in members:
++        if not member.get("certificate"):
++            raise ValueError("every machine must approve locally")
++        ssl.PEM_cert_to_DER_cert(member["certificate"])
++        if member != head and not same_network(head, member):
++            raise ValueError("members must share a public address and private subnet")
++    certificates = [m["certificate"] for m in members]
++    if len(set(certificates)) != len(certificates):
++        raise ValueError("each approval needs its own identity")
++    return head
++
++
++def fit_table(machines, links, models, benchmarks):
++    """Only exact measured hardware/model matches have speed numbers. MB is used throughout."""
++    result = []
++    for model in models:
++        threshold = 1000 if model["parameters_b"] > 30 else 300
++        good_links = len(links) == len(machines) - 1 and all(
++            math.isfinite(x["rtt_ms"]) and 0 <= x["rtt_ms"] < 5 and
++            math.isfinite(x["mbps"]) and x["mbps"] >= threshold for x in links)
++        memory = sum(m["free_vram"] + m["free_ram"] for m in machines)
++        key = sorted(str(m.get("hardware", "unknown")) for m in machines)
++        measured = next((b for b in benchmarks if b["model"] == model["model"] and
++                         b["hardware"] == key and b.get("source") and b.get("measured_at")), None)
++        result.append({"model": model["model"], "fits": bool(good_links and memory >= model["memory_mb"]),
++                       "free_vram_mb": sum(m["free_vram"] for m in machines),
++                       "free_ram_mb": sum(m["free_ram"] for m in machines),
++                       "tokens_per_second": measured["tokens_per_second"] if measured else None,
++                       "measurement": measured})
++    return result
++
++
++class Identity:
++    def __init__(self, directory):
++        self.directory = Path(directory)
++        self.certificate = (self.directory / "certificate.pem").read_text()
++
++    @classmethod
++    def approve(cls, directory):
++        """Called only after a local approval. Never returns or logs the private key."""
++        directory = Path(directory)
++        directory.mkdir(mode=0o700, parents=True, exist_ok=False)
++        try:
++            # openssl writes the private key itself; the umask protects it from creation.
++            subprocess.run(["openssl", "req", "-x509", "-newkey", "ed25519", "-nodes", "-days", "30",
++                            "-subj", "/CN=OpenVibe-pool", "-keyout", str(directory / "key.pem"),
++                            "-out", str(directory / "certificate.pem")], check=True,
++                           stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, umask=0o077)
++            return cls(directory)
++        except BaseException:
++            shutil.rmtree(directory)
++            raise
++
++    def context(self, peer, server=False):
++        context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER if server else ssl.PROTOCOL_TLS_CLIENT)
++        context.minimum_version = ssl.TLSVersion.TLSv1_3
++        if not server:
++            context.check_hostname = False  # identity is the exact certificate, not a DNS name
++        context.verify_mode = ssl.CERT_REQUIRED
++        context.load_verify_locations(cadata=peer)
++        context.load_cert_chain(str(self.directory / "certificate.pem"), str(self.directory / "key.pem"))
++        return context
++
++    def connect(self, address, peer):
++        raw = socket.create_connection(address, timeout=3)
++        raw.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
++        try:
++            conn = self.context(peer).wrap_socket(raw, server_hostname=None)
++            verify_pin(conn, peer)
++            return conn
++        except BaseException:
++            raw.close()
++            raise
++
++    def revoke(self):
++        shutil.rmtree(self.directory, ignore_errors=True)
++
++
++def verify_pin(conn, certificate):
++    expected = hashlib.sha256(ssl.PEM_cert_to_DER_cert(certificate)).digest()
++    actual = hashlib.sha256(conn.getpeercert(binary_form=True)).digest()
++    if actual != expected:
++        raise ssl.SSLError("unapproved pool peer")
++
++
++def relay(left, right, stopped):
++    """Bounded bidirectional tensor transport; close both ends on stop or loss."""
++    try:
++        while not stopped.is_set():
++            readable, _, _ = select.select([left, right], [], [], .2)
++            for source in readable:
++                target = right if source is left else left
++                data = source.recv(65536)
++                if not data:
++                    return
++                target.sendall(data)
++                # A TLS socket may hold decrypted bytes select() cannot see.
++                while getattr(source, "pending", lambda: 0)():
++                    data = source.recv(65536)
++                    if not data:
++                        return
++                    target.sendall(data)
++    except (OSError, ValueError):
++        pass
++    finally:
++        left.close()
++        right.close()
++
++
++class Tunnel:
++    def __init__(self, identity, head_certificate, address, port, status, rpc_socket=None, test_loopback=False, activate=None):
++        if not (test_loopback and address == "127.0.0.1"):
++            lan_address(address)
++        self.identity, self.head_certificate = identity, head_certificate
++        self.address, self.status, self.rpc_socket = (address, port), status, rpc_socket
++        self.activate = activate
++        self.context = identity.context(head_certificate, server=True)
++        self.stopped = threading.Event()
++        self.sessions = set()
++        self.lock = threading.Lock()
++        self.listener = None
++        self.last_head = time.monotonic()
++        self.slots = threading.BoundedSemaphore(16)
++        self.rpc_connections = 0
++
++    def start(self):
++        self.listener = socket.socket()
++        self.listener.bind(self.address)
++        self.listener.listen(8)
++        self.listener.settimeout(.2)
++        self.address = self.listener.getsockname()
++        self.thread = threading.Thread(target=self._accept, daemon=True)
++        self.thread.start()
++
++    def _accept(self):
++        while not self.stopped.is_set():
++            try:
++                conn, _ = self.listener.accept()
++                if not self.slots.acquire(blocking=False):
++                    conn.close()
++                    continue
++                with self.lock:
++                    self.sessions.add(conn)
++                threading.Thread(target=self._serve, args=(conn,), daemon=True).start()
++            except socket.timeout:
++                continue
++            except OSError:
++                break
++
++    def _serve(self, raw):
++        conn = raw
++        try:
++            raw.settimeout(3)
++            raw.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
++            conn = self.context.wrap_socket(raw, server_side=True)
++            with self.lock:
++                self.sessions.discard(raw)
++                self.sessions.add(conn)
++            verify_pin(conn, self.head_certificate)
++            operation = b""
++            while len(operation) < 16 and not operation.endswith(b"\n"):
++                chunk = conn.recv(1)
++                if not chunk:
++                    return
++                operation += chunk
++            self.last_head = time.monotonic()
++            if operation in (b"RUN\n", b"PAUSE\n") and self.activate:
++                self.activate(operation == b"RUN\n")
++                conn.sendall(b"OK\n")
++            elif operation == b"STATUS\n":
++                conn.sendall(json.dumps(self.status()).encode() + b"\n")
++            elif operation == b"PING\n":
++                conn.sendall(b"PONG\n")
++            elif operation == b"PROBE\n":
++                block = bytes(65536)
++                for _ in range(PROBE_BYTES // len(block)):
++                    conn.sendall(block)
++            elif operation == b"RPC\n" and self.rpc_socket and self.status().get("ready"):
++                local = socket.socket(socket.AF_UNIX)
++                local.settimeout(3)
++                local.connect(str(self.rpc_socket))
++                self.rpc_connections += 1
++                conn.sendall(b"OK\n")
++                conn.settimeout(5)
++                relay(conn, local, self.stopped)
++        except (OSError, ValueError):
++            pass
++        finally:
++            with self.lock:
++                self.sessions.discard(raw)
++                self.sessions.discard(conn)
++            conn.close()
++            self.slots.release()
++
++    def close(self):
++        self.stopped.set()
++        if self.listener:
++            self.listener.close()
++        with self.lock:
++            for conn in list(self.sessions):
++                try:
++                    conn.shutdown(socket.SHUT_RDWR)
++                except OSError:
++                    pass
++                conn.close()
++        if hasattr(self, "thread"):
++            self.thread.join(1)
++
++
++def probe(identity, source, target, address=None):
++    """Measure the link to one member before serving: three round trips and a 16 MB transfer over the authenticated
++    link. A machine off this local network (another public address or subnet) is refused before any connection."""
++    if not same_network(source, target):
++        raise ValueError("off-network machine refused before connection")
++    address = address or (str(ipaddress.ip_interface(target["address"]).ip), target["port"])
++    rtts = []
++    for _ in range(3):
++        with identity.connect(address, target["certificate"]) as conn:
++            start = time.monotonic()
++            conn.sendall(b"PING\n")
++            if conn.recv(5) != b"PONG\n":
++                raise ValueError("link probe failed")
++            rtts.append((time.monotonic() - start) * 1000)
++    with identity.connect(address, target["certificate"]) as conn:
++        start, received = time.monotonic(), 0
++        conn.sendall(b"PROBE\n")
++        while received < PROBE_BYTES:
++            block = conn.recv(min(65536, PROBE_BYTES - received))
++            if not block:
++                raise ValueError("incomplete link probe")
++            received += len(block)
++    return {"machine": target["id"], "rtt_ms": max(rtts), "mbps": received * 8 / (time.monotonic() - start) / 1e6}
ddiff --git a/plugins/ai/openvibe_ai/pool_cli.py b/plugins/ai/openvibe_ai/pool_cli.py
nnew file mode 100644
iindex 0000000..7249785
---- /dev/null
++++ b/plugins/ai/openvibe_ai/pool_cli.py
@@@ -0,0 +1,100 @@
++"""A machine pool on this machine: python -m openvibe_ai.pool_cli (never reachable through an ai command).
++
++    announce   put this machine on its owner's "Pool these machines" list (the plugin also does it every minute)
++    approve    show a proposed pool, ask the person at this machine, make this machine's key and send its public
++               certificate (the private key never leaves the machine)
++    fit        on the approved head: measure every link and say which models fit (measured speeds only)
++    leave      delete this machine's key at once, even offline, then tell Network (which revokes the whole pool)
++"""
++import argparse
++import json
++import os
++from pathlib import Path
++import platform
++import subprocess
++import sys
++from .pool import Identity, fit_table, probe, validate_pool
++from .pool_registry import Registry, local_address
++
++
++def memory():
++    ram = os.sysconf('SC_AVPHYS_PAGES') * os.sysconf('SC_PAGE_SIZE') // (1024 * 1024)
++    vram = 0
++    try:
++        data = subprocess.check_output(['nvidia-smi', '--query-gpu=memory.free', '--format=csv,noheader,nounits'], timeout=2)
++        vram = sum(int(x) for x in data.decode().splitlines())
++    except (OSError, ValueError, subprocess.SubprocessError):
++        pass
++    amd_devices = set()
++    for card in Path('/sys/class/drm').glob('card*/device'):
++        device = card.resolve()
++        if device in amd_devices:
++            continue
++        amd_devices.add(device)
++        try:
++            total = int((device / 'mem_info_vram_total').read_text())
++            used = int((device / 'mem_info_vram_used').read_text())
++            vram += max(0, total - used) // (1024 * 1024)
++        except (OSError, ValueError):
++            pass
++    return {'free_ram': ram, 'free_vram': vram, 'hardware': platform.machine()}
++
++
++def main():
++    if sys.platform != "linux":
++        raise RuntimeError("pool isolation currently requires Linux")
++    parser = argparse.ArgumentParser(description='Approve only your own same-network machines. Keys stay on this machine.')
++    parser.add_argument('action', choices=['announce', 'approve', 'fit', 'leave'])
++    parser.add_argument('--credential-file', required=True)
++    parser.add_argument('--directory', required=True)
++    parser.add_argument('--pool')
++    parser.add_argument('--models', help='local JSON array of model, parameters_b and measured memory_mb')
++    parser.add_argument('--benchmarks', help='optional local JSON array of dated, sourced measured speeds')
++    parser.add_argument('--address', help='this machine private IPv4 address/prefix, such as 192.168.1.2/24 (default: the default route)')
++    parser.add_argument('--port', type=int, default=50053)
++    args = parser.parse_args()
++    registry = Registry(args.credential_file)
++    directory = Path(args.directory).resolve()
++    if args.action == 'announce':
++        # The address must be configured on one of this machine's interfaces, not just a claimed subnet.
++        result = registry.announce(local_address(args.address), args.port, memory())
++        print(json.dumps(result))  # the machine id, its addresses and free memory; never a credential
++    elif args.action == 'fit':
++        doc = registry.pool(args.pool)
++        head = validate_pool(doc)
++        identity = Identity(directory / 'identity')
++        if identity.certificate != head['certificate']:
++            raise ValueError('run the fit measurement on the approved head')
++        links = [probe(identity, head, m) for m in doc['machines'] if m['id'] != doc['head']]
++        models = json.loads(Path(args.models).read_text())
++        benchmarks = json.loads(Path(args.benchmarks).read_text()) if args.benchmarks else []
++        print(json.dumps({'links': links, 'fit': fit_table(doc['machines'], links, models, benchmarks)}, indent=2))
++    elif args.action == 'approve':
++        doc = registry.pool(args.pool)
++        print('Pool %s; owner %s; head %s; members %s' %
++              (doc['id'], doc['owner'], doc['head'], ', '.join(m['id'] for m in doc['machines'])))
++        print('The head will see prompts; members receive tensors. Type approve on this machine:')
++        if input().strip() != 'approve':
++            raise ValueError('local approval declined')
++        directory.mkdir(mode=0o700, parents=True, exist_ok=True)
++        identity = Identity.approve(directory / 'identity')
++        try:
++            registry.call('POST', '/' + args.pool + '/approve', {'certificate': identity.certificate})
++            (directory / 'approval.json').write_text(json.dumps({'pool_id': args.pool, 'credential_file': str(Path(args.credential_file).resolve()), 'owner': doc['owner'], 'head': doc['head'], 'machines': sorted(m['id'] for m in doc['machines'])}))
++            os.chmod(directory / 'approval.json', 0o600)
++        except BaseException:
++            identity.revoke()
++            raise
++        print('Approved. Enable pool_directory in the local ai plugin configuration.')
++    else:
++        # Local revocation first, even when Network is unreachable: the running plugin sees the key gone at its next
++        # tick and closes the private link.
++        import shutil
++        shutil.rmtree(directory / 'identity', ignore_errors=True)
++        (directory / 'approval.json').unlink(missing_ok=True)
++        registry.call('POST', '/' + args.pool + '/leave', {})
++        print('Left pool %s. This machine\'s key is deleted.' % args.pool)
++
++
++if __name__ == '__main__':
++    main()
ddiff --git a/plugins/ai/openvibe_ai/pool_registry.py b/plugins/ai/openvibe_ai/pool_registry.py
nnew file mode 100644
iindex 0000000..becddce
---- /dev/null
++++ b/plugins/ai/openvibe_ai/pool_registry.py
@@@ -0,0 +1,123 @@
++"""Network reads for a machine pool (bot.machine-pool@1), with this machine's own Node pairing.
++
++The node credential stays in the Node's credential file: it buys a short-lived node token (audience
++openvibe.network) and never enters a pool document, a log line or the pool directory.
++"""
++import hashlib
++import ipaddress
++import json
++from pathlib import Path
++import subprocess
++import time
++import urllib.error
++import urllib.parse
++import urllib.request
++
++# Network's answers that mean "this machine is no longer in this pool": signed out or another owner (401), not a
++# member (403), the pool revoked or gone (404), its membership changed (409). Anything else (a timeout, a 5xx) is an
++# outage: the pool pauses but keeps its key.
++GONE = (401, 403, 404, 409)
++RENEW_S = 2.0    # a member renews its readiness this often; Network's window is ten seconds
++
++
++class Registry:
++    def __init__(self, credential_file, timeout=3.0):
++        self.path = Path(credential_file)
++        self.timeout = timeout
++        self.token, self.expires = None, 0
++        self.pairing = None
++
++    def credentials(self):
++        return json.loads(self.path.read_text())
++
++    def call(self, method, path, body=None):
++        # Re-read every call to notice a sign-out, a removal or another owner on this machine.
++        cred = self.credentials()
++        pairing = hashlib.sha256(json.dumps(cred, sort_keys=True).encode()).digest()
++        if pairing != self.pairing:
++            self.expires = 0
++            self.pairing = pairing
++        network = str(cred.get('network') or '').rstrip('/')
++        if not network.startswith('https://'):
++            raise ValueError('pool registry requires an https:// Network (a Network-paired machine)')
++        if time.monotonic() >= self.expires:
++            data = urllib.parse.urlencode({'grant_type': 'client_credentials', 'client_id': cred['principal'],
++                                           'client_secret': cred['node_credential'], 'audience': 'openvibe.network'}).encode()
++            request = urllib.request.Request(network + '/oauth/token', data=data,
++                                             headers={'Content-Type': 'application/x-www-form-urlencoded', 'Accept': 'application/json'})
++            with urllib.request.urlopen(request, timeout=self.timeout) as response:
++                token = json.loads(response.read(65536))
++            self.token, self.expires = token['access_token'], time.monotonic() + min(int(token['expires_in']), 240)
++        request = urllib.request.Request(network + '/api/v1/node/self/pools' + path,
++                                         method=method, data=json.dumps(body).encode() if body is not None else None,
++                                         headers={'Authorization': 'Bearer ' + self.token, 'Content-Type': 'application/json'})
++        with urllib.request.urlopen(request, timeout=self.timeout) as response:
++            return json.loads(response.read(131072))
++
++    def pool(self, pool_id):
++        return self.call('GET', '/' + urllib.parse.quote(pool_id, safe=''))['pool']
++
++    def renew(self, pool_id, ready):
++        """Renew this machine's readiness; the answer carries the pool, so one request also rechecks membership."""
++        return self.call('POST', '/' + urllib.parse.quote(pool_id, safe='') + '/status', {'ready': bool(ready)})['pool']
++
++    def announce(self, address, port, memory):
++        return self.call('POST', '/announce', dict(memory, address=str(address), port=int(port)))['machine']
++
++
++class Membership:
++    """The pool runtime's registry check: called every tick, it answers the current pool document, or None when this
++    machine is no longer in the pool (its key was deleted locally, it was signed out, removed, or the pool revoked).
++    Network is asked at most every RENEW_S; between renewals the last answer stands. An outage raises, so the runtime
++    pauses (fails closed) without deleting the key."""
++
++    def __init__(self, registry, pool_id, key_file, ready=lambda: False, interval=RENEW_S, clock=time.monotonic):
++        self.registry, self.pool_id, self.key_file = registry, pool_id, Path(key_file)
++        self.ready, self.interval, self.clock = ready, interval, clock
++        self.last, self.doc = None, None
++
++    def __call__(self):
++        if not self.key_file.exists():
++            return None
++        if self.doc is not None and self.last is not None and self.clock() - self.last < self.interval:
++            return self.doc
++        try:
++            try:
++                doc = self.registry.renew(self.pool_id, self.ready())
++            except urllib.error.HTTPError as error:
++                if error.code != 403:
++                    raise
++                # 403 from a renewal before this machine's own approval reached Network: read instead.
++                doc = self.registry.pool(self.pool_id)
++        except FileNotFoundError:
++            return None      # the credential file is gone: signed out
++        except urllib.error.HTTPError as error:
++            if error.code in GONE:
++                return None
++            raise
++        self.doc, self.last = doc, self.clock()
++        return doc
++
++
++def local_address(address=None):
++    """This machine's private IPv4 address and prefix (such as 192.168.1.2/24), configured on one of its interfaces.
++    Without an address, the source address of the default route. Never a public, loopback or wildcard address."""
++    interfaces = json.loads(subprocess.check_output(['ip', '-j', 'address', 'show'], timeout=3))
++    configured = [ipaddress.ip_interface('%s/%s' % (a['local'], a['prefixlen']))
++                  for iface in interfaces for a in iface.get('addr_info', []) if a.get('family') == 'inet']
++    if address is None:
++        routes = json.loads(subprocess.check_output(['ip', '-j', 'route', 'get', '1.1.1.1'], timeout=3) or b'[]')
++        source = routes[0].get('prefsrc') if routes else None
++        found = [i for i in configured if str(i.ip) == source]
++        if not found:
++            raise ValueError('no private address on the default route; pass the address')
++        chosen = found[0]
++    else:
++        chosen = ipaddress.ip_interface(address)
++        if chosen not in configured:
++            raise ValueError('address is not configured on this machine')
++    from .pool import lan_address
++    lan_address(str(chosen.ip))
++    if not 8 <= chosen.network.prefixlen <= 30:
++        raise ValueError('a pool needs a /8 to /30 private subnet')
++    return chosen
ddiff --git a/plugins/ai/openvibe_ai/pool_runtime.py b/plugins/ai/openvibe_ai/pool_runtime.py
nnew file mode 100644
iindex 0000000..e9b1810
---- /dev/null
++++ b/plugins/ai/openvibe_ai/pool_runtime.py
@@@ -0,0 +1,392 @@
++"""Local pool lifecycle (the pooled machines design, 2026-11-06). Registry loss, changed membership and any busy or
++stopped member fail closed.
++
++Every machine runs a PoolRuntime. A member opens the private link (pool.Tunnel) on its private address and, only while
++the head says RUN and keeps calling, runs `rpc-server` inside its own network namespace. The head asks every member's
++status each half second, pauses the whole pool when any member is busy (idle-only), stopped (the kill switch) or
++unreachable, measures every link before it starts, and runs `llama-server --rpc` in its own namespace with the members
++reached only through the authenticated link. Leaving (a removal, a sign-out, another owner, a revoked pool) closes the
++link and deletes the key; a plain close (the Node stopping) keeps the key for the next start.
++"""
++import http.client
++import ipaddress
++import json
++import os
++from pathlib import Path
++import shutil
++import signal
++import socket
++import subprocess
++import sys
++import threading
++import time
++import urllib.error
++import urllib.parse
++from .pool import Identity, Tunnel, fit_table, probe, relay, validate_pool
++
++
++BWRAP = 'bwrap'
++UNSHARE = 'unshare'
++ISOLATION_METHODS = (BWRAP, UNSHARE)
++# One plain sentence the Node app can show. The engine never gets the host network instead.
++ISOLATION_HELP = ('Pooling needs an isolated network for the engine. '
++                  'Install bubblewrap (sudo apt install bubblewrap) and try again.')
++
++_UNSET = object()
++_method = _UNSET
++
++
++def namespace_prefix(method, net_admin=False):
++    """The command that puts what follows in a private user and network namespace.
++
++    bubblewrap is preferred: Ubuntu's AppArmor profile lets it make the user namespace even when
++    kernel.apparmor_restrict_unprivileged_userns=1, and it brings the namespace's loopback up itself. unshare is the
++    fallback that works where user namespaces are unrestricted. net_admin is only for a caller that must configure
++    more of the namespace (the test's stand-in LAN); the engine needs nothing beyond bubblewrap's loopback."""
++    if method == BWRAP:
++        caps = ['--cap-add', 'CAP_NET_ADMIN'] if net_admin else []
++        return ['bwrap', '--unshare-net', '--unshare-user', '--die-with-parent'] + caps + ['--dev-bind', '/', '/', '--']
++    return ['unshare', '--user', '--map-root-user', '--net', '--fork', '--kill-child']
++
++
++def _guard(method):
++    """setpriv: the engine dies with the thread that started it, never outliving a crashed plugin.
++
++    bwrap refuses inherited capabilities (it expects a setuid install), and the engine runs inside a namespaced
++    process with no need for ours: drop what this process passes in before bwrap."""
++    if not shutil.which('setpriv'):
++        return []
++    guard = ['setpriv', '--pdeathsig', 'KILL']
++    if method == BWRAP:
++        guard += ['--inh-caps=-all', '--ambient-caps=-all']
++    return guard
++
++
++def _probe(method):
++    """Can this method make the namespace on this machine? One throwaway `true`, nothing else."""
++    try:
++        result = subprocess.run(_guard(method) + namespace_prefix(method) + ['true'], stdin=subprocess.DEVNULL,
++                                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=10)
++    except (OSError, subprocess.SubprocessError):
++        return False
++    return result.returncode == 0
++
++
++def isolation_method():
++    """bwrap or unshare, whichever can make the namespace; probed once and cached. None when neither can."""
++    global _method
++    if _method is _UNSET:
++        _method = next((m for m in ISOLATION_METHODS if shutil.which(m) and _probe(m)), None)
++        if _method is not None:
++            print('pool engine isolation: ' + _method, file=sys.stderr, flush=True)
++    return _method
++
++
++def _worker_environment():
++    """The worker imports openvibe_ai (and, from a source checkout, the SDK), whatever the plugin's cwd is."""
++    environment = dict(os.environ)
++    package = Path(__file__).resolve().parent.parent
++    paths = [str(package)]
++    sdk = package.parent / 'sdk'
++    if sdk.is_dir():
++        paths.append(str(sdk))
++    existing = environment.get('PYTHONPATH')
++    environment['PYTHONPATH'] = os.pathsep.join(paths + ([existing] if existing else []))
++    return environment
++
++
++def _tree(pid):
++    """pid and its descendants, read from /proc (Linux); just pid elsewhere."""
++    found, todo = set(), [pid]
++    while todo:
++        current = todo.pop()
++        if current in found:
++            continue
++        found.add(current)
++        for task in Path('/proc/%d/task' % current).glob('*'):
++            try:
++                todo.extend(int(x) for x in (task / 'children').read_text().split())
++            except (OSError, ValueError):
++                pass
++    return found
++
++
++class Namespace:
++    def __init__(self, cfg):
++        if sys.platform != 'linux':
++            raise RuntimeError('pool isolation is currently supported on Linux only')
++        method = isolation_method()
++        if method is None:
++            raise RuntimeError(ISOLATION_HELP)
++        cfg = dict(cfg, host_namespace=os.readlink('/proc/self/ns/net'))
++        self.sockets = [Path(cfg[key]) for key in ('rpc_socket', 'api_socket') if cfg.get(key)]
++        self.process = subprocess.Popen(_guard(method) + namespace_prefix(method) +
++                                        [sys.executable, '-m', 'openvibe_ai.pool_worker'],
++                                        env=_worker_environment(),
++                                        stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
++                                        start_new_session=True)
++        self.process.stdin.write((json.dumps(cfg) + '\n').encode())
++        self.process.stdin.close()
++        time.sleep(.15)
++        if self.process.poll() is not None:
++            # The probe chose a working method, so this is the worker or the engine, not a licence to use the host
++            # network. (A machine with neither method never reaches here: it gets ISOLATION_HELP above.)
++            raise RuntimeError('pool isolation failed; no host-network fallback')
++
++    def close(self):
++        if self.process.poll() is None:
++            # This process group belongs exclusively to this namespace worker.
++            os.killpg(self.process.pid, signal.SIGTERM)
++            try:
++                self.process.wait(4)
++            except subprocess.TimeoutExpired:
++                os.killpg(self.process.pid, signal.SIGKILL)
++                self.process.wait()
++        # bwrap kills its child on teardown, so the worker may not reach its own cleanup: remove what it owns here.
++        for path in self.sockets:
++            path.unlink(missing_ok=True)
++
++
++class Forwarder:
++    """A private Unix socket on the head that the namespaced llama-server's --rpc reaches: each connection becomes a
++    new authenticated link to one member's tunnel, which pins that member's certificate."""
++
++    def __init__(self, path, identity, peer, address):
++        self.path, self.identity, self.peer, self.address = Path(path), identity, peer, address
++        self.stopped = threading.Event()
++        self.listener = socket.socket(socket.AF_UNIX)
++        self.listener.bind(str(path))
++        os.chmod(path, 0o600)
++        self.listener.listen(8)
++        self.listener.settimeout(.2)
++        self.thread = threading.Thread(target=self.run, daemon=True)
++        self.thread.start()
++
++    def run(self):
++        while not self.stopped.is_set():
++            conn = remote = None
++            try:
++                conn, _ = self.listener.accept()
++                remote = self.identity.connect(self.address, self.peer['certificate'])
++                remote.sendall(b'RPC\n')
++                if remote.recv(3) != b'OK\n':
++                    raise OSError('member refused the tensor link')
++                threading.Thread(target=relay, args=(conn, remote, self.stopped), daemon=True).start()
++            except socket.timeout:
++                continue
++            except OSError:
++                # Only this attempt's sockets: a relay already running keeps its own.
++                for sock in (conn, remote):
++                    if sock is not None:
++                        sock.close()
++
++    def close(self):
++        self.stopped.set()
++        self.listener.close()
++        self.thread.join(1)
++        self.path.unlink(missing_ok=True)
++
++
++class PoolRuntime:
++    def __init__(self, doc, machine, directory, config, busy, registry_check, namespace_factory=Namespace):
++        head = validate_pool(doc)
++        self.doc, self.machine, self.config = doc, machine, config
++        self.me = next(m for m in doc['machines'] if m['id'] == machine)
++        self.head = head
++        self.directory = Path(directory)
++        self.identity = Identity(self.directory / 'identity')
++        if self.identity.certificate != self.me['certificate']:
++            raise ValueError('local approval does not match this membership')
++        self.busy, self.registry_check, self.factory = busy, registry_check, namespace_factory
++        self.stopped = threading.Event()
++        self.worker, self.tunnel, self.links = None, None, []
++        self.paused = True
++        self.lock = threading.RLock()
++        self.measurements = []
++        self.head_running = False
++        self.last_error = None
++
++    def status(self):
++        return {'ready': not self.stopped.is_set() and not self.busy() and
++                self.worker is not None and self.worker.process.poll() is None,
++                'paused': self.paused, 'available': not self.busy() and not self.stopped.is_set(), 'pool': self.doc['id']}
++
++    def _peer_address(self, peer):
++        """Where a peer's private link listens: its private address and port."""
++        return (str(ipaddress.ip_interface(peer['address']).ip), peer['port'])
++
++    def _measure(self, peer):
++        """The head's direct link probe to one member (round trips and throughput), over the authenticated link."""
++        return probe(self.identity, self.me, peer, self._peer_address(peer))
++
++    def _make_tunnel(self):
++        """A member's private link on its own private address; only the head's pinned certificate gets in."""
++        return Tunnel(self.identity, self.head['certificate'], *self._peer_address(self.me),
++                      status=self.status, rpc_socket=self.directory / 'rpc.sock', activate=self.activate)
++
++    def start(self, watch=True):
++        if self.machine != self.doc['head']:
++            self.tunnel = self._make_tunnel()
++            self.tunnel.start()
++        if watch:
++            self.thread = threading.Thread(target=self.watch, daemon=True)
++            self.thread.start()
++
++    def activate(self, running):
++        self.head_running = running
++        if not running:
++            with self.lock:
++                self.paused = True
++                self.pause()
++
++    def tell_members(self, running):
++        """RUN or PAUSE to every member. A RUN that a member does not acknowledge raises (the pool stays paused); a
++        PAUSE goes to every member it can reach, and one it cannot pauses itself within three seconds of silence."""
++        failed = None
++        for peer in self.doc["machines"]:
++            if peer["id"] == self.machine:
++                continue
++            try:
++                with self.identity.connect(self._peer_address(peer), peer["certificate"]) as conn:
++                    conn.sendall(b"RUN\n" if running else b"PAUSE\n")
++                    if conn.recv(3) != b"OK\n":
++                        raise OSError("member did not acknowledge pool state")
++            except OSError as error:
++                failed = failed or error
++                if running:
++                    raise
++        if failed and running:
++            raise failed
++
++    def halt(self):
++        """The kill switch on this machine: stop its part now, before the next tick (the head also pauses every
++        member). The busy check keeps it stopped until the latch is cleared."""
++        with self.lock:
++            self.paused = True
++            self.pause()
++        if self.machine == self.doc["head"]:
++            self.tell_members(False)
++
++    def tick(self):
++        with self.lock:
++            if self.stopped.is_set():
++                return
++            # The authoritative record (renewed every two seconds) checks ownership, revocation and membership.
++            current = self.registry_check()
++            if current is None or current != self.doc:
++                self._leave_locked()
++                return
++            if self.worker and self.worker.process.poll() is not None:
++                self.last_error = "isolated engine exited"
++                self.pause()
++            busy = bool(self.busy())
++            all_ready = True
++            if self.machine == self.doc['head']:
++                for peer in self.doc['machines']:
++                    if peer['id'] == self.machine:
++                        continue
++                    with self.identity.connect(self._peer_address(peer), peer['certificate']) as conn:
++                        conn.sendall(b'STATUS\n')
++                        raw = conn.makefile('rb').readline(4097)
++                        state = json.loads(raw)
++                        busy = busy or state.get('available') is not True or state.get('pool') != self.doc['id']
++                        all_ready = all_ready and state.get('ready') is True
++                self.tell_members(not busy)
++            elif not self.head_running or (self.tunnel and time.monotonic() - self.tunnel.last_head > 3):
++                busy = True
++            self.paused = busy or (self.machine == self.doc["head"] and not all_ready)
++            if busy:
++                self.pause()
++            elif self.worker is None and all_ready:
++                self.launch()
++
++    def watch(self):
++        while not self.stopped.is_set():
++            try:
++                self.tick()
++            except Exception as error:
++                self.last_error = type(error).__name__ + ": " + str(error)
++                # Registry or peer unavailable: no cached permission to keep serving.
++                with self.lock:
++                    self.paused = True
++                    self.pause()
++                    if self.machine == self.doc["head"]:
++                        try:
++                            self.tell_members(False)
++                        except OSError:
++                            pass
++            self.stopped.wait(.5)
++
++    def launch(self):
++        cfg = dict(self.config, role='head' if self.machine == self.doc['head'] else 'member',
++                   rpc_port=50052, rpc_socket=str(self.directory / 'rpc.sock'), api_socket=str(self.directory / 'api.sock'))
++        if cfg['role'] == 'head':
++            peers = [m for m in self.doc['machines'] if m['id'] != self.machine]
++            self.measurements = [self._measure(m) for m in peers]
++            if not fit_table(self.doc['machines'], self.measurements, [self.config['model']], [])[0]['fits']:
++                raise ValueError('the measured links or memory do not fit this model')
++            try:
++                for i, peer in enumerate(peers):
++                    self.links.append(Forwarder(self.directory / ('link%d.sock' % i), self.identity, peer, self._peer_address(peer)))
++                cfg['links'] = [str(link.path) for link in self.links]
++                self.worker = self.factory(cfg)
++            except BaseException:
++                self.pause()
++                raise
++        else:
++            self.worker = self.factory(cfg)
++
++    def pause(self):
++        if self.worker:
++            self.worker.close()
++            self.worker = None
++        for link in self.links:
++            link.close()
++        self.links = []
++        for path in ('rpc.sock', 'api.sock'):
++            (self.directory / path).unlink(missing_ok=True)
++
++    def close(self):
++        """Stop the engines and close the link, keeping the key: the Node is stopping, not leaving the pool."""
++        with self.lock:
++            self.stopped.set()
++            self.paused = True
++            if self.tunnel:
++                self.tunnel.close()
++            self.pause()
++
++    def leave(self):
++        """Leave the pool: close the link and delete this machine's key at once."""
++        with self.lock:
++            self._leave_locked()
++
++    def _leave_locked(self):
++        self.stopped.set()
++        self.paused = True
++        if self.tunnel:
++            self.tunnel.close()
++        self.pause()
++        self.identity.revoke()
++
++    def engine_pids(self):
++        """The process ids of this machine's pool engine (the namespace and everything under it), empty when none
++        runs: idle-only does not count the pool's own GPU use as the owner's."""
++        worker = self.worker
++        process = getattr(worker, 'process', None)
++        if process is None or process.poll() is not None:
++            return set()
++        return _tree(process.pid)
++
++    def opener(self, request, timeout=30):
++        if self.paused or self.stopped.is_set() or self.machine != self.doc['head']:
++            raise OSError('pool is paused')
++        conn = http.client.HTTPConnection('localhost', timeout=timeout)
++        conn.sock = socket.socket(socket.AF_UNIX)
++        conn.sock.settimeout(timeout)
++        conn.sock.connect(str(self.directory / 'api.sock'))
++        url = urllib.parse.urlsplit(request.full_url)
++        conn.request(request.get_method(), url.path, body=request.data, headers=dict(request.header_items()))
++        response = conn.getresponse()
++        if response.status >= 400:
++            raise urllib.error.HTTPError(request.full_url, response.status, 'pool model server refused request', response.headers, response)
++        return response
ddiff --git a/plugins/ai/openvibe_ai/pool_worker.py b/plugins/ai/openvibe_ai/pool_worker.py
nnew file mode 100644
iindex 0000000..2ab18fd
---- /dev/null
++++ b/plugins/ai/openvibe_ai/pool_worker.py
@@@ -0,0 +1,92 @@
++"""Runs ONLY inside a private user/network namespace (bubblewrap or unshare). No remotely chosen commands."""
++import json
++import os
++from pathlib import Path
++import signal
++import socket
++import subprocess
++import sys
++import threading
++import time
++from .pool import relay
++
++
++def bridge(listener, target, unix, stopped):
++    listener.settimeout(.2)
++    while not stopped.is_set():
++        try:
++            conn, _ = listener.accept()
++            other = socket.socket(socket.AF_UNIX if unix else socket.AF_INET)
++            other.settimeout(5)
++            other.connect(target)
++            threading.Thread(target=relay, args=(conn, other, stopped), daemon=True).start()
++        except socket.timeout:
++            continue
++        except OSError:
++            if 'conn' in locals():
++                conn.close()
++
++
++def main():
++    cfg = json.loads(sys.stdin.readline())
++    # Never accidentally run the worker on the host network, even when called directly.
++    if os.readlink('/proc/self/ns/net') == cfg['host_namespace']:
++        raise RuntimeError('RPC requires an isolated network namespace')
++    # bwrap brings the namespace's loopback up itself (and the engine cannot add capabilities on top); unshare needs
++    # this and grants the capability. Best effort either way, then the engine's own bind is the real test.
++    try:
++        subprocess.run(['ip', 'link', 'set', 'lo', 'up'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
++    except OSError:
++        pass
++    stopped = threading.Event()
++    listeners = []
++    def unix_bridge(path, target):
++        listener = socket.socket(socket.AF_UNIX)
++        listener.bind(path)
++        os.chmod(path, 0o600)
++        listener.listen(8)
++        listeners.append(listener)
++        threading.Thread(target=bridge, args=(listener, target, False, stopped), daemon=True).start()
++    if cfg['role'] == 'member':
++        command = [cfg['rpc_binary'], '--host', '127.0.0.1', '--port', str(cfg['rpc_port'])]
++        unix_bridge(cfg['rpc_socket'], ('127.0.0.1', cfg['rpc_port']))
++    else:
++        endpoints = []
++        for i, path in enumerate(cfg['links']):
++            listener = socket.socket()
++            listener.bind(('127.0.0.1', 0))
++            listener.listen(8)
++            listeners.append(listener)
++            endpoints.append('127.0.0.1:%s' % listener.getsockname()[1])
++            threading.Thread(target=bridge, args=(listener, path, True, stopped), daemon=True).start()
++        command = [cfg['llama_binary'], '--host', '127.0.0.1', '--port', '8080', '--model', cfg['model_path'],
++                   '--alias', cfg['model']['model'], '--rpc', ','.join(endpoints), '--n-gpu-layers', '999', '--ctx-size', str(cfg.get('context', 2048))]
++        unix_bridge(cfg['api_socket'], ('127.0.0.1', 8080))
++    # RPC's optional RDMA negotiation must not bypass TLS. Do not enable its disk cache.
++    child = subprocess.Popen(command, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
++                             env=dict(os.environ, GGML_RPC_NO_RDMA='1'))
++    def stop(*_):
++        stopped.set()
++    signal.signal(signal.SIGTERM, stop)
++    signal.signal(signal.SIGINT, stop)
++    try:
++        while not stopped.wait(.2) and child.poll() is None:
++            pass
++    finally:
++        if child.poll() is None:
++            child.terminate()
++            try:
++                child.wait(3)
++            except subprocess.TimeoutExpired:
++                child.kill()
++                child.wait()
++        for listener in listeners:
++            listener.close()
++        for key in ('rpc_socket', 'api_socket'):
++            if cfg.get(key):
++                Path(cfg[key]).unlink(missing_ok=True)
++    return child.returncode or 0
++
++
++if __name__ == '__main__':
++    sys.exit(main())
ddiff --git a/plugins/ai/pyproject.toml b/plugins/ai/pyproject.toml
iindex e69c4e0..8b75808 100644
---- a/plugins/ai/pyproject.toml
++++ b/plugins/ai/pyproject.toml
@@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
  
  [project]
  name = "openvibe-ai"
--version = "0.3.0"
++version = "0.4.0"
  description = "OpenVibe Node ai plugin: your own models (Ollama, llama.cpp, LM Studio) for your own OpenVibe work, relayed from OpenVibe with no open port; its owner may offer it to the network"
  requires-python = ">=3.9"
  license = { text = "AGPL-3.0-or-later" }
ddiff --git a/plugins/ai/tests/pool_linux_scenario.py b/plugins/ai/tests/pool_linux_scenario.py
nnew file mode 100644
iindex 0000000..b93888c
---- /dev/null
++++ b/plugins/ai/tests/pool_linux_scenario.py
@@@ -0,0 +1,182 @@
++"""Two machines pool and serve a model through llama.cpp RPC, on a throwaway local network.
++
++Run by test_pool_linux.py inside a private user/network namespace (bubblewrap or unshare): that namespace is the
++"LAN" (192.168.77.2 and .3 on its loopback), and each machine's engine runs in a nested namespace of its own, exactly
++as on a real machine. With POOL_RPC_BINARY, POOL_LLAMA_BINARY and POOL_MODEL it runs real llama.cpp (the pool
++workflow); without them, stand-in engines that speak TCP and HTTP the same way, so the transport, the isolation and
++the pool's rules are checked on every run. Prints one PASS line.
++"""
++import json
++import os
++from pathlib import Path
++import socket
++import ssl
++import subprocess
++import sys
++import tempfile
++import time
++import urllib.request
++
++sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
++from openvibe_ai.pool import Identity  # noqa: E402
++from openvibe_ai.pool_runtime import PoolRuntime  # noqa: E402
++
++FAKE_RPC = r'''
++import socket, sys, threading
++args = dict(zip(sys.argv[1::2], sys.argv[2::2]))
++server = socket.socket(); server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
++server.bind((args['--host'], int(args['--port']))); server.listen(8)
++def serve(conn):
++    with conn:
++        while True:
++            data = conn.recv(65536)
++            if not data:
++                return
++            conn.sendall(b'tensor-ok:' + data)
++while True:
++    conn, _ = server.accept()
++    threading.Thread(target=serve, args=(conn,), daemon=True).start()
++'''
++
++FAKE_LLAMA = r'''
++import http.server, json, socket, sys
++args = dict(zip(sys.argv[1::2], sys.argv[2::2]))
++members = [x.rsplit(':', 1) for x in args['--rpc'].split(',')]
++class Handler(http.server.BaseHTTPRequestHandler):
++    def do_POST(self):
++        self.rfile.read(int(self.headers.get('Content-Length') or 0))
++        replies = []
++        for host, port in members:
++            with socket.create_connection((host, int(port)), timeout=5) as conn:
++                conn.sendall(b'layer-17')
++                replies.append(conn.recv(64).decode())
++        body = json.dumps({'content': 'Once upon a time', 'rpc': replies}).encode()
++        self.send_response(200); self.send_header('Content-Type', 'application/json')
++        self.send_header('Content-Length', str(len(body))); self.end_headers(); self.wfile.write(body)
++    def log_message(self, *a):
++        pass
++http.server.HTTPServer((args['--host'], int(args['--port'])), Handler).serve_forever()
++'''
++
++
++def wait(check, seconds=60, what='pool condition'):
++    until = time.monotonic() + seconds
++    while time.monotonic() < until:
++        try:
++            if check():
++                return
++        except (OSError, ValueError, ssl.SSLError):
++            pass
++        time.sleep(.2)
++    raise AssertionError('%s did not become true' % what)
++
++
++def refused(address):
++    try:
++        socket.create_connection(address, timeout=.3).close()
++    except OSError:
++        return True
++    return False
++
++
++def engines(root):
++    if all(os.environ.get(k) for k in ('POOL_RPC_BINARY', 'POOL_LLAMA_BINARY', 'POOL_MODEL')):
++        return 'real llama.cpp', {'rpc_binary': os.environ['POOL_RPC_BINARY'], 'llama_binary': os.environ['POOL_LLAMA_BINARY'],
++                                  'model_path': os.environ['POOL_MODEL'], 'context': 256,
++                                  'model': {'model': 'stories15M', 'parameters_b': .015, 'memory_mb': 100}}
++    paths = {}
++    for name, source in (('rpc', FAKE_RPC), ('llama', FAKE_LLAMA)):
++        path = Path(root) / ('fake-%s' % name)
++        path.write_text('#!%s\n%s' % (sys.executable, source))
++        path.chmod(0o755)
++        paths[name] = str(path)
++    return 'stand-in engines', {'rpc_binary': paths['rpc'], 'llama_binary': paths['llama'], 'model_path': '/dev/null',
++                                'context': 256, 'model': {'model': 'stand-in', 'parameters_b': .015, 'memory_mb': 100}}
++
++
++def main():
++    subprocess.run(['ip', 'link', 'set', 'lo', 'up'], check=True)
++    for address in ('192.168.77.2/24', '192.168.77.3/24'):
++        subprocess.run(['ip', 'address', 'add', address, 'dev', 'lo'], check=True)
++    with tempfile.TemporaryDirectory(prefix='ov-pool-') as root:
++        mode, cfg = engines(root)
++        paths = [Path(root) / x for x in ('head', 'member')]
++        identities = [Identity.approve(p / 'identity') for p in paths]
++        stranger = Identity.approve(Path(root) / 'stranger' / 'identity')
++        owner = 'usr_01J8Z3Q4R5S6T7V8W9X0Y1Z2A1'
++        members = [{'id': 'dev_01J8Z3Q4R5S6T7V8W9X0Y1Z2D%d' % (i + 1), 'owner': owner, 'public_ip': '203.0.113.1',
++                    'address': '192.168.77.%d/24' % (i + 2), 'port': 50053, 'certificate': identities[i].certificate,
++                    'free_vram': 0, 'free_ram': 2048, 'hardware': 'test-cpu'} for i in range(2)]
++        doc = {'id': 'pool_01J8Z3Q4R5S6T7V8W9X0Y1Z2P1', 'owner': owner, 'head': members[0]['id'], 'machines': members}
++        busy = [False, False]
++        runtimes = [PoolRuntime(doc, members[i]['id'], paths[i], cfg, lambda i=i: busy[i], lambda: doc) for i in range(2)]
++        head, member = runtimes
++        serving = lambda: head.worker is not None and not head.paused and member.worker is not None  # noqa: E731
++        paused = lambda: head.paused and head.worker is None and member.worker is None  # noqa: E731
++        try:
++            member.start()
++            head.start()
++            try:
++                wait(serving, what='the pool serving')
++            except AssertionError:
++                raise AssertionError('head=%s member=%s measurements=%s' % (head.last_error, member.last_error, head.measurements))
++
++            # rpc-server is reachable only through the tunnel: not on the LAN address, not on loopback.
++            for address in (('192.168.77.3', 50052), ('127.0.0.1', 50052), ('192.168.77.2', 8080), ('127.0.0.1', 8080)):
++                assert refused(address), 'an engine port escaped its namespace: %s:%d' % address
++            # The member's tunnel takes only the head: a stranger's key and a plain connection are refused.
++            # (TLS 1.3: the client's side of the handshake ends before the member checks its certificate, so the
++            # refusal arrives on the first read.)
++            try:
++                with stranger.connect(('192.168.77.3', 50053), identities[1].certificate) as conn:
++                    conn.sendall(b'STATUS\n')
++                    answer = conn.recv(64)
++            except (ssl.SSLError, OSError):
++                answer = b''
++            assert answer == b'', 'a stranger entered the tunnel: %r' % answer
++            with socket.create_connection(('192.168.77.3', 50053), timeout=2) as plain:
++                plain.sendall(b'RPC\n')
++                plain.settimeout(4)
++                try:
++                    assert plain.recv(16) == b'', 'plain TCP got an answer'
++                except (ConnectionResetError, socket.timeout):
++                    pass
++
++            def inference():
++                request = urllib.request.Request('http://localhost/completion', data=json.dumps({'prompt': 'Once upon a time', 'n_predict': 4}).encode(),
++                                                 headers={'Content-Type': 'application/json'})
++                with head.opener(request, timeout=30) as response:
++                    result = json.loads(response.read())
++                assert result.get('content'), result
++                if 'rpc' in result:
++                    assert result['rpc'] == ['tensor-ok:layer-17'], result
++                return True
++            wait(inference, what='an answer through the pool')
++            assert member.tunnel.rpc_connections > 0, 'inference never entered the member tunnel'
++
++            # Idle-only on one member pauses the whole pool, and it resumes when the member is idle again.
++            busy[1] = True
++            wait(paused, 10, 'the pool pausing for a busy member')
++            busy[1] = False
++            wait(serving, what='the pool resuming')
++            # The kill switch on a member stops the pool at once.
++            busy[1] = True
++            member.halt()
++            wait(paused, 5, 'the pool stopping for the kill switch')
++            busy[1] = False
++            wait(serving, what='the pool resuming after the latch is cleared')
++
++            # Leaving drops the tunnel and the key; the head stops.
++            member.leave()
++            wait(lambda: head.paused and head.worker is None, 10, 'the head stopping after a member left')
++            assert not (paths[1] / 'identity').exists(), 'the key outlived the membership'
++            assert refused(('192.168.77.3', 50053)), 'the tunnel outlived the membership'
++            print('PASS: %s: inference through the tunnel, engines isolated, only the head admitted, idle-only and the '
++                  'kill switch pause the pool, leaving drops the tunnel and the key' % mode)
++        finally:
++            for runtime in runtimes:
++                runtime.close()
++
++
++if __name__ == '__main__':
++    main()
ddiff --git a/plugins/ai/tests/test_pool.py b/plugins/ai/tests/test_pool.py
nnew file mode 100644
iindex 0000000..469aff9
---- /dev/null
++++ b/plugins/ai/tests/test_pool.py
@@@ -0,0 +1,546 @@
++"""Machine pools: the trust boundary of llama.cpp RPC, which has no authentication of its own.
++
++Real TLS sockets and real keys throughout, never a mocked authentication decision. Machines on loopback stand in for
++the local network (LoopbackRuntime maps each machine's private address to a loopback port; the pool documents keep
++their real private addresses, so every same-network check still runs). test_pool_linux.py runs the same flows in
++network namespaces with engines, and with real llama.cpp in the pool workflow.
++"""
++import json
++import os
++import socket
++import ssl
++import sys
++import tempfile
++import threading
++import time
++import unittest
++import urllib.error
++from pathlib import Path
++
++HERE = Path(__file__).resolve().parent
++sys.path.insert(0, str(HERE.parent))
++sys.path.insert(0, str(HERE.parent.parent / "sdk"))
++
++from openvibe_plugin import StopInfo  # noqa: E402
++import openvibe_ai  # noqa: E402
++from openvibe_ai import Ai  # noqa: E402
++from openvibe_ai.pool import Identity, Tunnel, validate_pool, same_network, fit_table, probe  # noqa: E402
++from openvibe_ai.pool_registry import Membership  # noqa: E402
++from openvibe_ai.pool_runtime import PoolRuntime  # noqa: E402
++
++OWNER = "usr_01J8Z3Q4R5S6T7V8W9X0Y1Z2A1"
++HEAD, MEMBER, OTHER = ("dev_01J8Z3Q4R5S6T7V8W9X0Y1Z2D%d" % i for i in (1, 2, 3))
++ENGINE = {"rpc_binary": "/opt/llama/rpc-server", "llama_binary": "/opt/llama/llama-server", "model_path": "/opt/models/small.gguf",
++          "model": {"model": "small", "parameters_b": 1, "memory_mb": 1000}}
++
++
++def record(*identities):
++    doc = {"id": "pool_01J8Z3Q4R5S6T7V8W9X0Y1Z2P1", "owner": OWNER, "head": HEAD, "machines": [
++        {"id": HEAD, "owner": OWNER, "public_ip": "203.0.113.1", "address": "192.168.1.2/24", "port": 50053, "free_vram": 8000, "free_ram": 16000, "hardware": "test", "certificate": None},
++        {"id": MEMBER, "owner": OWNER, "public_ip": "203.0.113.1", "address": "192.168.1.3/24", "port": 50053, "free_vram": 4000, "free_ram": 16000, "hardware": "test", "certificate": None}]}
++    for member, identity in zip(doc["machines"], identities):
++        member["certificate"] = identity.certificate
++    return doc
++
++
++def free_port():
++    with socket.socket() as s:
++        s.bind(("127.0.0.1", 0))
++        return s.getsockname()[1]
++
++
++class LoopbackRuntime(PoolRuntime):
++    """A PoolRuntime whose machines listen on loopback ports instead of their private addresses."""
++    ports = {}
++
++    def _peer_address(self, peer):
++        return ("127.0.0.1", self.ports[peer["id"]])
++
++    def _measure(self, peer):
++        # The real probe over the real link; loopback stands in for a LAN, so its speed under a loaded test machine
++        # is not the point (fit_table's gates have their own test).
++        result = PoolRuntime._measure(self, peer)
++        return dict(result, rtt_ms=min(result["rtt_ms"], 1.0), mbps=max(result["mbps"], 10000.0))
++
++    def _make_tunnel(self):
++        return Tunnel(self.identity, self.head["certificate"], "127.0.0.1", self.ports[self.machine], self.status,
++                      self.directory / "rpc.sock", test_loopback=True, activate=self.activate)
++
++
++class FakeEngine:
++    """What the namespace factory returns: a live 'process' until closed."""
++    started = []
++
++    def __init__(self, cfg):
++        self.cfg = cfg
++        self.closed = False
++        self.process = self
++        FakeEngine.started.append(self)
++
++    def poll(self):
++        return 0 if self.closed else None
++
++    def close(self):
++        self.closed = True
++
++
++def echo_server(path):
++    """A stand-in rpc-server on the member's private Unix socket: echoes what the head sends."""
++    listener = socket.socket(socket.AF_UNIX)
++    listener.bind(str(path))
++    listener.listen(4)
++
++    def serve():
++        while True:
++            try:
++                conn, _ = listener.accept()
++            except OSError:
++                return
++            with conn:
++                data = conn.recv(1024)
++                conn.sendall(b"echo:" + data)
++    threading.Thread(target=serve, daemon=True).start()
++    return listener
++
++
+[email protected](sys.platform == "linux", "machine pools run on Linux")
++class PoolSecurity(unittest.TestCase):
++    def setUp(self):
++        self.tmp = tempfile.TemporaryDirectory()
++        self.addCleanup(self.tmp.cleanup)
++        root = Path(self.tmp.name)
++        self.head = Identity.approve(root / "head" / "identity")
++        self.member = Identity.approve(root / "member" / "identity")
++        self.other = Identity.approve(root / "other" / "identity")       # another approved member, not the head
++        self.stranger = Identity.approve(root / "stranger" / "identity")
++
++    def tunnel(self, ready=True, rpc_socket=None):
++        tunnel = Tunnel(self.member, self.head.certificate, "127.0.0.1", 0, lambda: {"ready": ready}, rpc_socket, test_loopback=True)
++        tunnel.start()
++        self.addCleanup(tunnel.close)
++        return tunnel
++
++    def rpc(self, identity, tunnel):
++        with identity.connect(tunnel.address, self.member.certificate) as conn:
++            conn.sendall(b"RPC\n")
++            if conn.recv(3) != b"OK\n":
++                return None
++            conn.sendall(b"tensor")
++            return conn.recv(64)
++
++    # rpc-server listens only on the private tunnel and accepts only the head.
++    def test_only_the_head_reaches_the_rpc_server(self):
++        path = Path(self.tmp.name) / "rpc.sock"
++        self.addCleanup(echo_server(path).close)
++        tunnel = self.tunnel(rpc_socket=path)
++        self.assertEqual(self.rpc(self.head, tunnel), b"echo:tensor", "the head's tensors reach rpc-server")
++        for name, identity in (("another member", self.other), ("a stranger", self.stranger)):
++            with self.subTest(name), self.assertRaises((ssl.SSLError, OSError)):
++                self.rpc(identity, tunnel)
++        self.assertEqual(tunnel.rpc_connections, 1, "only the head's connection was bridged")
++
++    def test_plain_connection_outside_the_tunnel_is_refused(self):
++        path = Path(self.tmp.name) / "rpc.sock"
++        self.addCleanup(echo_server(path).close)
++        tunnel = self.tunnel(rpc_socket=path)
++        with socket.create_connection(tunnel.address, timeout=2) as conn:
++            conn.sendall(b"RPC\ntensor")
++            try:
++                self.assertEqual(conn.recv(64), b"", "no TLS, no RPC")
++            except ConnectionResetError:
++                pass
++        self.assertEqual(tunnel.rpc_connections, 0)
++
++    def test_no_rpc_while_this_member_is_not_ready(self):
++        path = Path(self.tmp.name) / "rpc.sock"
++        self.addCleanup(echo_server(path).close)
++        tunnel = self.tunnel(ready=False, rpc_socket=path)
++        self.assertIsNone(self.rpc(self.head, tunnel), "a paused member bridges nothing")
++
++    def test_the_head_pins_each_member_too(self):
++        tunnel = self.tunnel()
++        with self.assertRaises((ssl.SSLError, OSError)):
++            self.head.connect(tunnel.address, self.stranger.certificate)
++
++    def test_never_a_public_loopback_or_wildcard_listener(self):
++        for address in ("0.0.0.0", "::", "8.8.8.8", "100.64.0.1", "127.0.0.1", "localhost"):
++            with self.subTest(address=address), self.assertRaises(ValueError):
++                Tunnel(self.member, self.head.certificate, address, 50053, lambda: {})
++
++    def test_only_tensor_measurement_and_state_operations(self):
++        tunnel = self.tunnel()
++        with self.head.connect(tunnel.address, self.member.certificate) as conn:
++            conn.sendall(b"EXEC /bin/sh\n")
++            self.assertEqual(conn.recv(10), b"", "anything else closes the link")
++
++    # A machine off the local network is refused by the link probe.
++    def test_off_network_is_refused_before_any_connection(self):
++        a, b = record(self.head, self.member)["machines"]
++        self.assertTrue(same_network(a, b))
++        for key, value in (("public_ip", "203.0.113.2"), ("address", "192.168.2.3/24"), ("address", "8.8.8.8/24"),
++                           ("address", "192.168.1.2/24"), ("public_ip", "")):
++            target = dict(b, **{key: value})
++            with self.subTest(**{key: value}):
++                self.assertFalse(same_network(a, target))
++                with self.assertRaises(ValueError):
++                    probe(self.head, a, target, ("127.0.0.1", 9))
++
++    def test_the_probe_measures_an_approved_member(self):
++        tunnel = self.tunnel()
++        a, b = record(self.head, self.member)["machines"]
++        result = probe(self.head, a, b, tunnel.address)
++        self.assertEqual(result["machine"], MEMBER)
++        self.assertGreater(result["mbps"], 0)
++        with self.assertRaises((ssl.SSLError, OSError)):
++            probe(self.stranger, a, b, tunnel.address)
++
++    def test_every_member_approves_with_its_own_key_and_the_same_owner(self):
++        doc = record()
++        with self.assertRaises(ValueError):
++            validate_pool(doc)
++        doc = record(self.head, self.member)
++        validate_pool(doc)
++        same = record(self.head, self.head)
++        with self.assertRaises(ValueError):
++            validate_pool(same)
++        doc["machines"][1]["owner"] = "usr_01J8Z3Q4R5S6T7V8W9X0Y1Z2A2"
++        with self.assertRaises(ValueError):
++            validate_pool(doc)
++        doc = record(self.head, self.member)
++        doc["head"] = OTHER
++        with self.assertRaises(ValueError):
++            validate_pool(doc)
++
++    def test_fit_does_not_invent_speed_and_checks_the_link_gates(self):
++        machines = record()["machines"]
++        models = [{"model": "small", "parameters_b": 7, "memory_mb": 8000}, {"model": "large", "parameters_b": 70, "memory_mb": 40000}]
++        result = fit_table(machines, [{"rtt_ms": 1, "mbps": 999}], models, [])
++        self.assertTrue(result[0]["fits"])
++        self.assertIsNone(result[0]["tokens_per_second"], "no measured speed, no number")
++        self.assertFalse(result[1]["fits"], "a 70B model needs 1 Gb/s and the memory")
++        self.assertFalse(fit_table(machines, [{"rtt_ms": 5, "mbps": 1000}], models, [])[0]["fits"], "5 ms is too slow")
++        measured = [{"model": "small", "hardware": ["test", "test"], "tokens_per_second": 31.5, "source": "bench", "measured_at": "2026-11-06"}]
++        self.assertEqual(fit_table(machines, [{"rtt_ms": 1, "mbps": 999}], models, measured)[0]["tokens_per_second"], 31.5)
++
++
++class FakeRegistry:
++    def __init__(self, doc):
++        self.doc, self.calls, self.fail = doc, [], None
++
++    def renew(self, pool_id, ready):
++        self.calls.append(("renew", ready))
++        if self.fail:
++            raise self.fail
++        return self.doc
++
++    def pool(self, pool_id):
++        self.calls.append(("pool",))
++        return self.doc
++
++
++def http_error(code):
++    return urllib.error.HTTPError("https://openvibe.network/api", code, "refused", {}, None)
++
++
++class MembershipTests(unittest.TestCase):
++    def setUp(self):
++        self.tmp = tempfile.TemporaryDirectory()
++        self.addCleanup(self.tmp.cleanup)
++        self.key = Path(self.tmp.name) / "key.pem"
++        self.key.write_text("key")
++        self.now = [0.0]
++        self.registry = FakeRegistry({"id": "pool_x"})
++        self.check = Membership(self.registry, "pool_x", self.key, ready=lambda: True, clock=lambda: self.now[0])
++
++    def test_renews_every_two_seconds_and_reads_the_pool_back(self):
++        self.assertEqual(self.check(), {"id": "pool_x"})
++        self.now[0] = 1.0
++        self.check()
++        self.assertEqual(len(self.registry.calls), 1, "between renewals the last answer stands")
++        self.now[0] = 2.5
++        self.check()
++        self.assertEqual(self.registry.calls, [("renew", True), ("renew", True)])
++
++    def test_removed_signed_out_or_another_owner_means_gone(self):
++        for code in (401, 403, 404, 409):
++            with self.subTest(code=code):
++                self.registry.fail = http_error(code)
++                self.registry.pool = lambda pool_id, code=code: (_ for _ in ()).throw(http_error(code))
++                self.check.doc = None
++                self.assertIsNone(self.check())
++        self.registry.fail = FileNotFoundError("credential.json")
++        self.check.doc = None
++        self.assertIsNone(self.check(), "the credential file is gone: signed out")
++
++    def test_a_deleted_key_is_gone_at_once(self):
++        self.check()
++        self.key.unlink()
++        self.assertIsNone(self.check())
++
++    def test_an_outage_raises_so_the_pool_pauses_but_keeps_its_key(self):
++        self.registry.fail = http_error(503)
++        with self.assertRaises(urllib.error.HTTPError):
++            self.check()
++        self.registry.fail = urllib.error.URLError("timed out")
++        with self.assertRaises(OSError):
++            self.check()
++
++
+[email protected](sys.platform == "linux", "machine pools run on Linux")
++class PoolRuntimeTests(unittest.TestCase):
++    """Two machines on loopback: the head's checks pause the whole pool, and leaving drops the link and the key."""
++
++    def setUp(self):
++        self.tmp = tempfile.TemporaryDirectory()
++        self.addCleanup(self.tmp.cleanup)
++        root = Path(self.tmp.name)
++        self.dirs = {HEAD: root / "head", MEMBER: root / "member"}
++        identities = [Identity.approve(self.dirs[m] / "identity") for m in (HEAD, MEMBER)]
++        self.doc = record(*identities)
++        LoopbackRuntime.ports = {HEAD: free_port(), MEMBER: free_port()}
++        self.busy = {HEAD: False, MEMBER: False}
++        self.current = {HEAD: self.doc, MEMBER: self.doc}
++        FakeEngine.started = []
++        self.head = self.runtime(HEAD)
++        self.member = self.runtime(MEMBER)
++        self.member.start(watch=False)
++        self.head.start(watch=False)
++        self.addCleanup(self.head.close)
++        self.addCleanup(self.member.close)
++
++    def runtime(self, machine):
++        return LoopbackRuntime(self.doc, machine, self.dirs[machine], dict(ENGINE), lambda: self.busy[machine],
++                               lambda: self.current[machine], namespace_factory=FakeEngine)
++
++    def run_pool(self):
++        """Ticks in the order the watch threads would interleave them, until the head serves."""
++        for _ in range(4):
++            self.head.tick()
++            self.member.tick()
++        self.assertIsNotNone(self.head.worker, self.head.last_error)
++        self.assertFalse(self.head.paused)
++        self.assertIsNotNone(self.member.worker)
++
++    def test_the_head_starts_only_after_every_member_runs(self):
++        self.head.tick()
++        self.assertIsNone(self.head.worker, "the member's rpc-server is not running yet")
++        self.assertTrue(self.member.head_running, "the head told the member to run")
++        self.member.tick()
++        self.assertIsNotNone(self.member.worker)
++        self.assertEqual(self.member.worker.cfg["role"], "member")
++        self.head.tick()
++        self.assertEqual(self.head.worker.cfg["role"], "head")
++        self.assertEqual(len(self.head.measurements), 1, "the link was measured before serving")
++        self.assertEqual(len(self.head.worker.cfg["links"]), 1, "llama-server reaches the member only through the link")
++
++    def test_idle_only_on_one_member_pauses_the_whole_pool(self):
++        self.run_pool()
++        self.busy[MEMBER] = True
++        self.member.tick()
++        self.head.tick()
++        self.assertTrue(self.head.paused)
++        self.assertIsNone(self.head.worker, "the head stops serving")
++        self.assertIsNone(self.member.worker, "the busy member stops its rpc-server")
++        with self.assertRaises(OSError):
++            self.head.opener(None)
++        self.busy[MEMBER] = False
++        self.run_pool()
++
++    def test_idle_only_on_the_head_pauses_every_member(self):
++        self.run_pool()
++        self.busy[HEAD] = True
++        self.head.tick()
++        self.member.tick()
++        self.assertIsNone(self.head.worker)
++        self.assertIsNone(self.member.worker, "the head told the member to pause")
++
++    def test_the_kill_switch_on_a_member_stops_the_pool_at_once(self):
++        self.run_pool()
++        self.busy[MEMBER] = True   # the plugin's busy check holds the latch
++        self.member.halt()
++        self.assertIsNone(self.member.worker, "stopped before the next tick")
++        self.head.tick()
++        self.assertIsNone(self.head.worker)
++
++    def test_a_silent_head_pauses_its_members(self):
++        self.run_pool()
++        self.member.tunnel.last_head = time.monotonic() - 4
++        self.member.tick()
++        self.assertIsNone(self.member.worker, "no head call for three seconds: the member stops")
++
++    def test_leaving_the_pool_drops_the_tunnel_and_the_key(self):
++        self.run_pool()
++        port = LoopbackRuntime.ports[MEMBER]
++        open_link = self.head.identity.connect(("127.0.0.1", port), self.member.me["certificate"])
++        self.addCleanup(open_link.close)
++        self.member.leave()
++        self.assertFalse((self.dirs[MEMBER] / "identity").exists(), "the key is deleted")
++        self.assertIsNone(self.member.worker)
++        with self.assertRaises(OSError):
++            socket.create_connection(("127.0.0.1", port), timeout=.3).close()
++        open_link.settimeout(1)
++        try:
++            self.assertEqual(open_link.recv(1), b"", "an open link is closed too")
++        except (OSError, ssl.SSLError):
++            pass
++        with self.assertRaises(OSError):
++            self.head.tick()          # the head cannot reach the member: the watch loop pauses the pool
++        self.head.pause()
++        self.assertIsNone(self.head.worker)
++
++    def test_revoked_removed_or_changed_membership_leaves_at_once(self):
++        self.run_pool()
++        self.current[MEMBER] = None   # Network: revoked, removed, signed out or another owner
++        self.member.tick()
++        self.assertTrue(self.member.stopped.is_set())
++        self.assertFalse((self.dirs[MEMBER] / "identity").exists())
++        self.current[HEAD] = dict(self.doc, head=MEMBER)
++        self.head.tick()
++        self.assertTrue(self.head.stopped.is_set())
++        self.assertFalse((self.dirs[HEAD] / "identity").exists())
++
++    def test_closing_keeps_the_key(self):
++        self.run_pool()
++        self.member.close()
++        self.assertTrue((self.dirs[MEMBER] / "identity" / "key.pem").exists(), "the Node stopping is not leaving")
++        self.assertIsNone(self.member.worker)
++
++
++class FakeRuntime:
++    def __init__(self):
++        self.halted = self.closed = self.left = 0
++        self.paused = False
++        self.machine = HEAD
++        self.doc = {"id": "pool_x", "head": HEAD, "machines": [{"id": HEAD}, {"id": MEMBER}]}
++        self.stopped = threading.Event()
++        self.pids = {4242}
++
++    def halt(self):
++        self.halted += 1
++        self.paused = True
++
++    def close(self):
++        self.closed += 1
++
++    def leave(self):
++        self.left += 1
++
++    def status(self):
++        return {"ready": not self.paused}
++
++    def engine_pids(self):
++        return self.pids
++
++
+[email protected](sys.platform == "linux", "machine pools run on Linux")
++class PluginPoolTests(unittest.TestCase):
++    """The ai plugin in pool mode: the kill switch, stops that are not about the pool, and idle-only's GPU signal."""
++
++    def plugin(self):
++        tmp = tempfile.TemporaryDirectory()
++        self.addCleanup(tmp.cleanup)
++        plugin = Ai(busy_probe=lambda: (False, None))
++        cfg = {"pool_directory": tmp.name, "pool_engine": ENGINE, "offer": {"enabled": True, "quiet_minutes": 0}}
++        plugin._configure(cfg)
++        plugin.pool_runtime = FakeRuntime()
++        return plugin
++
++    def test_describe_cannot_start_a_pool(self):
++        plugin = Ai()
++        out = plugin.describe({"pool_directory": "/does-not-exist", "pool_engine": ENGINE, "offer": {"enabled": True}})
++        self.assertIsNone(plugin.pool_thread, "describe never starts a pool or makes a key")
++        self.assertFalse(out["ai"]["offer"]["ready"])
++        self.assertEqual(out["ai"]["pool"]["state"], "waiting")
++        self.assertEqual(out["ai"]["models"], ["small"])
++
++    def test_a_bad_engine_config_is_refused(self):
++        plugin = Ai()
++        for engine in (None, dict(ENGINE, rpc_binary="rpc-server"), dict(ENGINE, model={"model": "x"}),
++                       dict(ENGINE, model=dict(ENGINE["model"], parameters_b=True))):
++            with self.subTest(engine=engine):
++                self.assertEqual(plugin.describe({"pool_directory": "/tmp/pool", "pool_engine": engine}), {})
++
++    def test_the_kill_switch_stops_the_pool_until_the_latch_is_cleared(self):
++        plugin = self.plugin()
++        self.assertFalse(plugin._pool_busy())
++        plugin.on_stop(StopInfo("remote", emergency=True, latched=True))
++        self.assertEqual(plugin.pool_runtime.halted, 1, "stopped at once")
++        self.assertTrue(plugin._pool_busy(), "and held stopped")
++        self.assertEqual(plugin.pool_runtime.left, 0, "the kill switch stops; it does not leave")
++        plugin.on_resume(StopInfo("cleared"))
++        self.assertFalse(plugin._pool_busy())
++
++    def test_other_stops_are_not_about_the_pool(self):
++        plugin = self.plugin()
++        for cause in ("halt", "link_lost", "timeout", "heartbeat", "setup"):
++            plugin.on_stop(StopInfo(cause))
++        runtime = plugin.pool_runtime
++        self.assertEqual((runtime.halted, runtime.closed, runtime.left), (0, 0, 0))
++        plugin.on_stop(StopInfo("shutdown"))
++        self.assertEqual((runtime.closed, runtime.left), (1, 0), "the Node stopping closes the pool and keeps the key")
++
++    def test_the_pools_own_gpu_use_is_not_the_owner_using_the_machine(self):
++        plugin = self.plugin()
++        plugin.busy_probe = None
++        plugin.offer["busy_processes"] = []
++        saved = openvibe_ai._gpu_busy, openvibe_ai._gpu_compute_pids, openvibe_ai._input_idle_ms
++        self.addCleanup(lambda: setattr(openvibe_ai, "_gpu_busy", saved[0]) or setattr(openvibe_ai, "_gpu_compute_pids", saved[1]) or setattr(openvibe_ai, "_input_idle_ms", saved[2]))
++        openvibe_ai._gpu_busy = lambda threshold: True
++        openvibe_ai._input_idle_ms = lambda: None
++        openvibe_ai._gpu_compute_pids = lambda: {4242}
++        self.assertEqual(plugin._raw_busy(), (False, None), "only the pool's engine is on the GPU")
++        plugin._busy_cache = (0.0, False, None)
++        openvibe_ai._gpu_compute_pids = lambda: {4242, 999}
++        self.assertEqual(plugin._raw_busy(), (True, "gpu"), "another program on the GPU is the owner using it")
++        plugin._busy_cache = (0.0, False, None)
++        plugin.pool_runtime.pids = set()
++        openvibe_ai._gpu_compute_pids = lambda: {4242}
++        self.assertEqual(plugin._raw_busy(), (True, "gpu"), "with no engine running, a busy GPU is someone else's")
++
++    def test_a_pool_revoked_while_away_drops_the_key_and_the_approval(self):
++        import openvibe_ai.pool_registry as registry_module
++        tmp = tempfile.TemporaryDirectory()
++        self.addCleanup(tmp.cleanup)
++        root = Path(tmp.name)
++        Identity.approve(root / "identity")
++        (root / "approval.json").write_text(json.dumps({"pool_id": "pool_x", "credential_file": str(root / "credential.json"),
++                                                        "owner": OWNER, "head": HEAD, "machines": sorted([HEAD, MEMBER])}))
++
++        class Revoked:
++            def __init__(self, path):
++                pass
++
++            def renew(self, pool_id, ready):
++                raise http_error(404)
++
++            pool = renew
++
++            def credentials(self):
++                return {"device_id": MEMBER}
++        saved = registry_module.Registry
++        registry_module.Registry = Revoked
++        self.addCleanup(setattr, registry_module, "Registry", saved)
++        plugin = Ai()
++        plugin.setup({"pool_directory": str(root), "pool_engine": ENGINE, "offer": {"enabled": True}})
++        self.addCleanup(plugin.close)
++        until = time.monotonic() + 10
++        while (root / "approval.json").exists() and time.monotonic() < until:
++            time.sleep(.05)
++        self.assertFalse((root / "approval.json").exists(), "the approval is forgotten")
++        self.assertFalse((root / "identity").exists(), "the key is deleted")
++        self.assertIsNone(plugin.pool_runtime, "nothing runs")
++        self.assertIn("no longer in pool", plugin.pool_error)
++
++    def test_only_the_serving_head_takes_requests(self):
++        plugin = self.plugin()
++        self.assertTrue(plugin._pool_serving())
++        plugin.pool_runtime.machine = MEMBER
++        self.assertFalse(plugin._pool_serving())
++        self.assertFalse(plugin._offer_view()["ready"], "a member never offers on its own")
++        plugin.pool_runtime.machine = HEAD
++        plugin.pool_runtime.paused = True
++        self.assertFalse(plugin._offer_view()["ready"])
++
++
++if __name__ == "__main__":
++    unittest.main()
ddiff --git a/plugins/ai/tests/test_pool_linux.py b/plugins/ai/tests/test_pool_linux.py
nnew file mode 100644
iindex 0000000..073713d
---- /dev/null
++++ b/plugins/ai/tests/test_pool_linux.py
@@@ -0,0 +1,153 @@
++"""Machine pools in real Linux network namespaces: the engines' isolation and the whole two-machine flow.
++
++Uses bubblewrap when it can make the namespace (Ubuntu's AppArmor profile lets it even with
++kernel.apparmor_restrict_unprivileged_userns=1), otherwise unshare. Skips, with a reason, only when neither can.
++The real llama.cpp run needs POOL_RPC_BINARY, POOL_LLAMA_BINARY and POOL_MODEL (the pool workflow supplies them).
++"""
++import json
++import os
++from pathlib import Path
++import socket
++import subprocess
++import sys
++import tempfile
++import time
++import unittest
++from unittest import mock
++
++HERE = Path(__file__).resolve().parent
++sys.path.insert(0, str(HERE.parent))
++sys.path.insert(0, str(HERE.parent.parent / "sdk"))
++
++from openvibe_ai.pool_runtime import (ISOLATION_HELP, Namespace, isolation_method,  # noqa: E402
++                                      namespace_prefix)
++
++SCENARIO = HERE / "pool_linux_scenario.py"
++REAL = all(os.environ.get(k) for k in ("POOL_RPC_BINARY", "POOL_LLAMA_BINARY", "POOL_MODEL"))
++METHOD = isolation_method() if sys.platform == "linux" else None
++if METHOD is not None:
++    SKIP = None
++elif sys.platform != "linux":
++    SKIP = "machine pools run on Linux"
++else:
++    SKIP = "needs bubblewrap or unprivileged user namespaces; neither can make one here"
++
++
++def run_scenario(real):
++    env = dict(os.environ, PYTHONPATH=os.pathsep.join([str(HERE.parent), str(HERE.parent.parent / "sdk"), os.environ.get("PYTHONPATH", "")]))
++    if not real:
++        for key in ("POOL_RPC_BINARY", "POOL_LLAMA_BINARY", "POOL_MODEL"):
++            env.pop(key, None)
++    return subprocess.run(namespace_prefix(METHOD, net_admin=True) + [sys.executable, str(SCENARIO)],
++                          text=True, capture_output=True, timeout=240, env=env)
++
++
+[email protected](sys.platform == "linux", "machine pools run on Linux")
++class WorkerRefusal(unittest.TestCase):
++    def test_the_engine_never_runs_on_the_host_network(self):
++        cfg = {"host_namespace": os.readlink("/proc/self/ns/net"), "role": "member"}
++        result = subprocess.run([sys.executable, "-m", "openvibe_ai.pool_worker"], input=json.dumps(cfg) + "\n", text=True,
++                                capture_output=True, timeout=30,
++                                env=dict(os.environ, PYTHONPATH=os.pathsep.join([str(HERE.parent), str(HERE.parent.parent / "sdk")])))
++        self.assertNotEqual(result.returncode, 0)
++        self.assertIn("isolated network namespace", result.stderr)
++
++
+[email protected](METHOD is None, SKIP)
++class Isolation(unittest.TestCase):
++    def test_rpc_server_is_reachable_only_through_its_private_socket(self):
++        with tempfile.TemporaryDirectory() as root:
++            fake = Path(root) / "fake-rpc"
++            fake.write_text("#!%s\nimport socket, sys\nargs = dict(zip(sys.argv[1::2], sys.argv[2::2]))\n"
++                            "s = socket.socket(); s.bind((args['--host'], int(args['--port']))); s.listen(4)\n"
++                            "while True:\n    c, _ = s.accept(); c.sendall(b'rpc:' + c.recv(64)); c.close()\n" % sys.executable)
++            fake.chmod(0o755)
++            with socket.socket() as probe:
++                probe.bind(("127.0.0.1", 0))
++                port = probe.getsockname()[1]
++            path = Path(root) / "rpc.sock"
++            engine = Namespace({"role": "member", "rpc_binary": str(fake), "rpc_port": port, "rpc_socket": str(path)})
++            try:
++                until = time.monotonic() + 10
++                while not path.exists() and time.monotonic() < until:
++                    time.sleep(.1)
++                self.assertEqual(oct(path.stat().st_mode & 0o777), "0o600", "only this user reaches the bridge")
++                answer = b""
++                while not answer and time.monotonic() < until:
++                    try:   # the bridge may be up a moment before rpc-server listens
++                        with socket.socket(socket.AF_UNIX) as conn:
++                            conn.connect(str(path))
++                            conn.sendall(b"tensor")
++                            answer = conn.recv(64)
++                    except OSError:
++                        answer = b""
++                    if not answer:
++                        time.sleep(.1)
++                self.assertEqual(answer, b"rpc:tensor", "the private socket reaches rpc-server")
++                with self.assertRaises(OSError, msg="rpc-server is not on the host's loopback"):
++                    socket.create_connection(("127.0.0.1", port), timeout=.5).close()
++            finally:
++                engine.close()
++            self.assertFalse(path.exists(), "closing removes the bridge")
++
++    def test_two_machines_pool_and_serve_through_rpc(self):
++        result = run_scenario(real=False)
++        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
++        self.assertIn("PASS: stand-in engines", result.stdout)
++
++    @unittest.skipUnless(REAL, "needs real llama.cpp RPC binaries and a model; the pool workflow supplies them")
++    def test_two_machines_serve_a_real_model(self):
++        result = run_scenario(real=True)
++        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
++        self.assertIn("PASS: real llama.cpp", result.stdout)
++
++
+[email protected](sys.platform == "linux", "pool isolation runs on Linux")
++class IsolationChoice(unittest.TestCase):
++    """The probes are faked, so these prove the choice and the refusal without making a namespace."""
++
++    def setUp(self):
++        import openvibe_ai.pool_runtime as runtime
++        self.runtime = runtime
++        runtime._method = runtime._UNSET
++        self.addCleanup(setattr, runtime, "_method", runtime._UNSET)
++
++    def test_bwrap_is_chosen_first_when_both_work(self):
++        probes = []
++
++        def run(command, **_):
++            probes.append(command)
++            return subprocess.CompletedProcess(command, 0)
++
++        with mock.patch.object(self.runtime.shutil, "which", return_value="/usr/bin/tool"), \
++                mock.patch.object(self.runtime.subprocess, "run", side_effect=run):
++            self.assertEqual(self.runtime.isolation_method(), "bwrap")
++        self.assertIn("bwrap", probes[0])
++        self.assertNotIn("unshare", probes[0])
++        self.assertEqual(len(probes), 1, "bwrap works, so unshare is never probed")
++
++    def test_neither_method_means_the_plain_refusal_not_the_host_network(self):
++        def run(command, **_):
++            return subprocess.CompletedProcess(command, 1)
++
++        with mock.patch.object(self.runtime.shutil, "which", return_value="/usr/bin/tool"), \
++                mock.patch.object(self.runtime.subprocess, "run", side_effect=run), \
++                mock.patch.object(self.runtime.subprocess, "Popen") as popen:
++            with self.assertRaises(RuntimeError) as refusal:
++                self.runtime.Namespace({"role": "member", "rpc_binary": "/bin/true", "rpc_port": 1,
++                                        "rpc_socket": "/nonexistent/rpc.sock"})
++        self.assertEqual(str(refusal.exception), ISOLATION_HELP)
++        self.assertIn("Install bubblewrap", str(refusal.exception))
++        popen.assert_not_called()
++
++    def test_unshare_is_used_when_bwrap_cannot_make_the_namespace(self):
++        def run(command, **_):
++            return subprocess.CompletedProcess(command, 0 if "unshare" in command else 1)
++
++        with mock.patch.object(self.runtime.shutil, "which", return_value="/usr/bin/tool"), \
++                mock.patch.object(self.runtime.subprocess, "run", side_effect=run):
++            self.assertEqual(self.runtime.isolation_method(), "unshare")
++
++
++if __name__ == "__main__":
++    unittest.main()