fix(runtime): reload fiable, fuites au unload, budget CPU sandbox (2.6.3→2.6.8)

CLAUDE.md

@@ -6,7 +6,7 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co
 
 ## Présentation
 
-**xcore v2.5.3** — framework d'orchestration plugin-first construit sur FastAPI.
+**xcore v2.6.8** — framework d'orchestration plugin-first construit sur FastAPI.
 Charge, isole et gère des plugins modulaires dans un environnement sandboxé.
 
 - **Language** : Python 3.12+
@@ -240,3 +240,10 @@ Le cache `.venv` est clé sur `poetry.lock` — un changement de dépendances in
 - **Branche principale** : `main`. Ne pas confondre avec les branches de feature (`add-ephemeral`, etc.).
 - **Seuil de coverage** : `fail_under = 80` dans `pyproject.toml` (`branch = true`). Le CI échoue en dessous. Patcher les imports locaux dans `boot()` au niveau du module source (`xcore.services.container.ServiceContainer`) et non au niveau `xcore`.
 - **`asyncio_mode = auto`** dans `pyproject.toml` — pas besoin de `@pytest.mark.asyncio` sur chaque test async.
+- **Tâches de fond d'un plugin** — `ctx.spawn_task()`, jamais `asyncio.create_task()` brut : seules les tâches suivies par le kernel sont annulées *et attendues* au unload/reload. Une tâche brute épingle l'ancienne génération du plugin (visible dans les logs : `plugin instance still referenced after unload`).
+- **`self._services` d'un plugin est le dict partagé du `ServiceContainer`** (`LifecycleManager._services`) : n'y écrire que des services *exportés* par le plugin. `propagate_services()` ignore ce qui est identique à l'objet injecté (`_injected_services`) — ne jamais réintroduire un `update()` brut des services injectés (proxy de ramasse-miette, wrappers tenant) dans ce dict : ça casse le reload/load de tous les plugins.
+- **`get_service()` des services noyau** — `PluginContext.get_service()` sert les services noyau (`PluginRegistry.is_core_service`) depuis `ctx.services` du plugin, pas depuis le registre (qui contient les objets bruts après le boot : le suivi des jobs et l'isolation tenant seraient contournés).
+- **Plugins Ephemeral** — chaque instance poolée est un `LifecycleManager(..., pooled=True)` : espace de noms `sys.modules` propre, pas de désinscription du registre à l'unload.
+- **Routes de plugin** — monter/retirer via `Xcore._mount_plugin_router` / `_unmount_plugin_router` (suivi des objets route) ; `app.routes` n'a pas de setter et les `_IncludedRouter` de FastAPI ≥ 0.14x n'ont pas de `.path`.
+- **Budget CPU du sandbox** — `RLIMIT_CPU` est cumulatif : limite souple réarmée par requête dans `worker.py` (`_arm_cpu_budget`), limite dure = plafond de vie du worker (`max_cpu_lifetime_seconds`), recyclage propre via `RECYCLE_EXIT_CODE`. Ne pas reposer une limite unique au démarrage.
+- **Tests d'intégration du reload** — utiliser un vrai `PluginRegistry` + `ServiceContainer` (+ `register_core_service` pour simuler « après boot »), pas des `MagicMock` : les mocks avaient masqué ces bugs.

doc/plugins/trusted-plugins.md

@@ -134,6 +134,7 @@ class Plugin(TrustedBase):
 | `get_service(name)` | Returns a service from the container. Supports literal overloads for IDE typing. |
 | `get_service_as(name, type)` | Returns a service cast to a specific type (e.g., `AsyncSQLAdapter`). |
 | `call_plugin(name, action, payload)` | IPC helper to call another plugin from within the Trusted environment. |
+| `ctx.spawn_task(coro, name=None)` | Creates a background task tracked by the kernel: it is cancelled **and awaited** when the plugin is unloaded or reloaded. Use it instead of a bare `asyncio.create_task()`. |
 
 ---
 
@@ -156,8 +157,30 @@ entry_point: "src/main.py"
     **Fix**: Prefix your service names (e.g., `myplugin_db`) or use the `PluginRegistry` to set them as private.
 
 !!! warning "Synchronous Blocking"
-    Trusted plugins run in the main event loop. If you perform blocking I/O (like `time.sleep()` or synchronous requests) inside a hook or `handle`, you will freeze the entire application.
-    **Fix**: Always use `async`/`await` or `run_in_executor`.
+    Trusted plugins run in the main event loop, in a single thread. Anything your code does *between two `await`s* — CPU work, `time.sleep()`, a synchronous HTTP or database call — freezes the **entire application**, and the Python GIL means threads do not change that for CPU-bound work.
+    **Fix**: use `async`/`await`; for blocking I/O use `asyncio.to_thread()`; for heavy CPU work use `execution_mode: sandboxed` (one process per plugin, outside the main loop and its GIL).
+
+!!! warning "`timeout_seconds` cannot interrupt synchronous code"
+    The `resources.timeout_seconds` limit is enforced with `asyncio.wait_for`, which can only act at an `await`. A `handle()` that burns 1 s of CPU without awaiting and has `timeout_seconds: 0.2` still returns normally after 1 s.
+    Xcore now **reports** it instead: a synchronous step longer than `plugins.loop_block_warn_ms` (default `250`, `0` disables) logs `plugin blocked the event loop` with the plugin and action, rate-limited to one warning per plugin every 10 s, and `status()` exposes `max_loop_block_ms`. The same warning exists for scheduler jobs (`scheduler job blocked the event loop`) and for synchronous hooks that exceed their timeout (their worker thread cannot be interrupted and keeps running).
+
+---
+
+### Reload, Unload and Memory
+
+When a plugin is unloaded or reloaded, the kernel releases what it can track: scheduler jobs, health checks, event/hook subscriptions, services you exported, your router and middlewares, and tasks created with `ctx.spawn_task()`. A plugin instance lives in reference cycles, so Xcore also schedules a garbage collection shortly afterwards (`plugins.gc_after_unload`, default `true`) and checks that the old instance is really gone.
+
+If you see `plugin instance still referenced after unload` in the logs, something outside the plugin context still holds it — typically:
+
+- a task started with a bare `asyncio.create_task()` (use `ctx.spawn_task()`),
+- a callback or bound method registered on a global object you imported yourself,
+- a service object stored somewhere that outlives the plugin.
+
+!!! note "Scheduler job ids are namespaced"
+    Jobs you register are stored as `<plugin>:<job_id>` so two plugins can both have a `cleanup` job. Inside your plugin you keep using your own id (`remove_job("cleanup")`).
+
+!!! warning "Reload only affects one worker process"
+    With several server workers (`uvicorn --workers N`, `xcli manager start --workers N`) each worker process loads its own copy of every plugin. A `POST /plugins/<name>/reload` is handled by **one** worker only; the others keep running the old code until restarted. There is no cross-worker reload broadcast yet: restart the workers (or reload through each of them) to roll out a new plugin version.
 
 ---
 

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "XCoreRuntime"
-version = "2.6.2"
+version = "2.6.8"
 description = "Plugin-first orchestration framework built on FastAPI"
 authors = [
     { name = "Eliezer Traore", email = "68350805+traoreera@users.noreply.github.com" },

tests/unit/kernel/test_blocking_sources.py

@@ -0,0 +1,103 @@
+"""
+Les autres sources de code synchrone dans le event loop sont signalées elles
+aussi : jobs du scheduler et hooks synchrones qui dépassent leur timeout.
+"""
+
+import asyncio
+import logging
+import threading
+import time
+
+from xcore.kernel.events.hooks import HookManager
+from xcore.services.scheduler import service as scheduler_module
+
+
+def _spin(seconds: float) -> None:
+    end = time.perf_counter() + seconds
+    while time.perf_counter() < end:
+        pass
+
+
+class TestSchedulerJobs:
+    async def test_sync_job_blocking_the_loop_is_reported(self, caplog):
+        scheduler_module._JOB_REGISTRY["sync_job"] = lambda: _spin(0.06)
+        try:
+            scheduler_module._BLOCK_WARN_MS, old = 30, scheduler_module._BLOCK_WARN_MS
+            with caplog.at_level(logging.WARNING):
+                await scheduler_module._dispatch_job("sync_job")
+        finally:
+            scheduler_module._BLOCK_WARN_MS = old
+            scheduler_module._JOB_REGISTRY.pop("sync_job", None)
+
+        assert any(
+            "scheduler job blocked the event loop" in r.getMessage()
+            for r in caplog.records
+        )
+
+    async def test_blocking_step_inside_an_async_job_is_reported(self, caplog):
+        async def job():
+            await asyncio.sleep(0)
+            _spin(0.06)
+
+        scheduler_module._JOB_REGISTRY["async_job"] = job
+        try:
+            scheduler_module._BLOCK_WARN_MS, old = 30, scheduler_module._BLOCK_WARN_MS
+            with caplog.at_level(logging.WARNING):
+                await scheduler_module._dispatch_job("async_job")
+        finally:
+            scheduler_module._BLOCK_WARN_MS = old
+            scheduler_module._JOB_REGISTRY.pop("async_job", None)
+
+        assert any("blocked the event loop" in r.getMessage() for r in caplog.records)
+
+    async def test_cooperative_async_job_is_silent_and_runs(self, caplog):
+        ran = []
+
+        async def job():
+            await asyncio.sleep(0.001)
+            ran.append(1)
+
+        scheduler_module._JOB_REGISTRY["quiet_job"] = job
+        try:
+            with caplog.at_level(logging.WARNING):
+                await scheduler_module._dispatch_job("quiet_job")
+        finally:
+            scheduler_module._JOB_REGISTRY.pop("quiet_job", None)
+
+        assert ran == [1]
+        assert not [r for r in caplog.records if "blocked" in r.getMessage()]
+
+
+class TestSyncHookTimeout:
+    async def test_sync_hook_timeout_warns_that_the_thread_keeps_running(self, caplog):
+        release = threading.Event()
+
+        def slow_hook(event):
+            release.wait(2)
+
+        hooks = HookManager()
+        hooks.register("evt", slow_hook, timeout=0.05)
+
+        try:
+            with caplog.at_level(logging.WARNING):
+                results = await hooks.emit("evt")
+        finally:
+            release.set()
+
+        assert results and results[0].error is not None
+        assert any(
+            "worker thread keeps running" in r.getMessage() for r in caplog.records
+        )
+
+    async def test_async_hook_timeout_does_not_emit_the_thread_warning(self, caplog):
+        async def slow_hook(event):
+            await asyncio.sleep(1)
+
+        hooks = HookManager()
+        hooks.register("evt", slow_hook, timeout=0.05)
+
+        with caplog.at_level(logging.WARNING):
+            results = await hooks.emit("evt")
+
+        assert results[0].error is not None
+        assert not [r for r in caplog.records if "worker thread" in r.getMessage()]

tests/unit/kernel/test_context.py

@@ -63,3 +63,24 @@ def test_env_default_empty(self):
     def test_config_default_empty(self):
         ctx = PluginContext(name="auth")
         assert ctx.config == {}
+
+
+    def test_get_service_core_service_served_from_plugin_context(self):
+        """Un service noyau vient du contexte du plugin (proxy/tenant), pas du registre brut."""
+        registry = MagicMock()
+        registry.is_core_service.return_value = True
+        registry.get_service.return_value = "raw_scheduler"
+        ctx = PluginContext(
+            name="auth", services={"scheduler": "scoped_scheduler"}, registry=registry
+        )
+
+        assert ctx.get_service("scheduler") == "scoped_scheduler"
+        registry.get_service.assert_not_called()
+
+    def test_get_service_plugin_export_still_resolved_by_registry(self):
+        registry = MagicMock()
+        registry.is_core_service.return_value = False
+        registry.get_service.return_value = "exported"
+        ctx = PluginContext(name="auth", services={"cart": "stale"}, registry=registry)
+
+        assert ctx.get_service("cart") == "exported"

tests/unit/kernel/test_ephemeral_isolation.py

@@ -0,0 +1,150 @@
+"""
+Régression : plusieurs instances d'un même plugin Ephemeral coexistent (warm
+pool, appels concurrents). Avant le correctif elles partageaient le même nom de
+module dans `sys.modules` et chaque unload faisait `registry.unregister(plugin)` :
+décharger UNE instance cassait les imports paresseux des instances sœurs
+(`ModuleNotFoundError: No module named 'xcore_plugin_<nom>'`) et retirait du
+registre un plugin pourtant toujours actif.
+"""
+
+import sys
+from types import SimpleNamespace
+
+from xcore.configurations.sections import EphemeralConfig, ServicesConfig
+from xcore.kernel.context import KernelContext
+from xcore.kernel.events.bus import EventBus
+from xcore.kernel.events.hooks import HookManager
+from xcore.kernel.observability.health import HealthChecker
+from xcore.kernel.runtime.ephemeral_handler import EphemeralHandler
+from xcore.kernel.runtime.lifecycle import LifecycleManager
+from xcore.kernel.runtime.warm_pool import WarmPool
+from xcore.registry.index import PluginRegistry
+from xcore.services.container import ServiceContainer
+
+PLUGIN_SRC = """
+from xcore.kernel.api.contract import TrustedBase
+
+class Plugin(TrustedBase):
+    async def handle(self, action, payload):
+        from . import helper          # import paresseux relatif
+        return {"status": "ok", "v": helper.VALUE}
+"""
+
+
+def _setup(tmp_path):
+    plugin_dir = tmp_path / "eph"
+    (plugin_dir / "src").mkdir(parents=True)
+    (plugin_dir / "src" / "main.py").write_text(PLUGIN_SRC)
+    (plugin_dir / "src" / "helper.py").write_text("VALUE = 42\n")
+    manifest = SimpleNamespace(
+        name="eph",
+        plugin_dir=plugin_dir,
+        entry_point="src/main.py",
+        resources=SimpleNamespace(timeout_seconds=10),
+        env={},
+        requires=[],
+        extra={},
+    )
+    registry = PluginRegistry()
+    ctx = KernelContext(
+        config=SimpleNamespace(tenancy=None),
+        services=ServiceContainer(ServicesConfig()),
+        registry=registry,
+        events=EventBus(),
+        hooks=HookManager(),
+        health=HealthChecker(),
+    )
+    return manifest, ctx, registry
+
+
+class TestPooledLifecycleManagers:
+    async def test_unloading_one_instance_does_not_break_its_sibling(self, tmp_path):
+        manifest, ctx, _ = _setup(tmp_path)
+        a = LifecycleManager(manifest, ctx, pooled=True)
+        b = LifecycleManager(manifest, ctx, pooled=True)
+        await a.load()
+        await b.load()
+
+        await a.unload()
+
+        assert await b.call("x", {}) == {"status": "ok", "v": 42}
+        await b.unload()
+
+    async def test_unloading_a_pooled_instance_keeps_the_registry_entry(self, tmp_path):
+        manifest, ctx, registry = _setup(tmp_path)
+        registry.register("eph", object())
+        lm = LifecycleManager(manifest, ctx, pooled=True)
+        await lm.load()
+
+        await lm.unload()
+
+        assert registry.has("eph")
+
+    async def test_pooled_instances_get_distinct_module_namespaces(self, tmp_path):
+        manifest, ctx, _ = _setup(tmp_path)
+        a = LifecycleManager(manifest, ctx, pooled=True)
+        b = LifecycleManager(manifest, ctx, pooled=True)
+        persistent = LifecycleManager(manifest, ctx)
+
+        assert len({a._module_name, b._module_name, persistent._module_name}) == 3
+        assert persistent._module_name == "xcore_plugin_eph"
+
+    async def test_unload_only_purges_its_own_modules(self, tmp_path):
+        manifest, ctx, _ = _setup(tmp_path)
+        a = LifecycleManager(manifest, ctx, pooled=True)
+        b = LifecycleManager(manifest, ctx, pooled=True)
+        await a.load()
+        await b.load()
+
+        await a.unload()
+
+        assert not [m for m in sys.modules if m.startswith(a._module_name)]
+        assert f"{b._module_name}.main" in sys.modules
+        await b.unload()
+
+    async def test_non_pooled_unload_still_unregisters_the_plugin(self, tmp_path):
+        manifest, ctx, registry = _setup(tmp_path)
+        registry.register("eph", object())
+        lm = LifecycleManager(manifest, ctx)
+        await lm.load()
+
+        await lm.unload()
+
+        assert not registry.has("eph")
+
+
+class TestWarmPoolInstances:
+    async def test_discarding_a_pool_instance_keeps_the_other_ones_working(
+        self, tmp_path
+    ):
+        manifest, ctx, _ = _setup(tmp_path)
+        pool = WarmPool(manifest=manifest, ctx=ctx, pool_size=2)
+        await pool.start()
+        try:
+            first = await pool.acquire()
+            second = await pool.acquire()
+            assert first._module_name != second._module_name
+
+            await pool.discard(first)
+
+            assert await second.call("x", {}) == {"status": "ok", "v": 42}
+            await pool.release(second)
+        finally:
+            await pool.shutdown()
+
+
+class TestEphemeralHandlerRegistry:
+    async def test_stop_unregisters_the_plugin_from_the_registry(self, tmp_path):
+        manifest, ctx, registry = _setup(tmp_path)
+        registry.register("eph", object())
+        handler = EphemeralHandler(
+            manifest=manifest, ctx=ctx, config=EphemeralConfig(pool_size=1)
+        )
+        await handler.start()
+        assert registry.has("eph")  # démarrer / servir ne la désinscrit pas
+        assert (await handler.call("x", {}))["v"] == 42
+        assert registry.has("eph")
+
+        await handler.stop()
+
+        assert not registry.has("eph")

tests/unit/kernel/test_lifecycle.py

@@ -420,7 +420,7 @@ async def _inject_context(self, ctx):
         await manager.load()
         real_scheduler.add_job.assert_called_once()
         await manager.unload()
-        real_scheduler.remove_job.assert_called_once_with("nightly")
+        real_scheduler.remove_job.assert_called_once_with("test_plugin:nightly")
 
     @pytest.mark.asyncio
     async def test_unload_removes_health_check(self, mock_manifest):