diff --git a/openspace/skill_engine/evolution/orchestrator.py b/openspace/skill_engine/evolution/orchestrator.py
new file mode 100644
index 00000000..c5ef9e68
--- /dev/null
+++ b/openspace/skill_engine/evolution/orchestrator.py
@@ -0,0 +1,148 @@
+"""Evolution orchestration — dispatch, throttle, and background scheduling.
+
+Functions extracted from ``SkillEvolver`` (Epic 5.2):
+- ``dispatch_evolution``   — route a single context to the right engine
+- ``execute_contexts``     — run a batch of contexts in parallel (throttled)
+- ``schedule_background``  — fire-and-forget background evolution task
+- ``log_background_result`` — callback that logs task outcome
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import TYPE_CHECKING, List, Optional
+
+from openspace.utils.logging import Logger
+
+from .models import EvolutionContext
+
+from openspace.skill_engine.types import EvolutionType
+
+if TYPE_CHECKING:
+    from openspace.skill_engine.types import SkillRecord
+
+logger = Logger.get_logger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# Single-context dispatch
+# ---------------------------------------------------------------------------
+
+async def dispatch_evolution(
+    evolver,
+    ctx: EvolutionContext,
+) -> Optional["SkillRecord"]:
+    """Execute one evolution action.  Returns new ``SkillRecord`` or *None*.
+
+    Delegates to the appropriate engine method on *evolver*:
+    ``_evolve_fix``, ``_evolve_derived``, or ``_evolve_captured``.
+
+    The global semaphore is **not** acquired here — it is managed by
+    ``execute_contexts`` so the concurrency limit covers whole batches.
+    """
+    evo_type: EvolutionType = ctx.suggestion.evolution_type
+    try:
+        if evo_type == EvolutionType.FIX:
+            return await evolver._evolve_fix(ctx)
+        elif evo_type == EvolutionType.DERIVED:
+            return await evolver._evolve_derived(ctx)
+        elif evo_type == EvolutionType.CAPTURED:
+            return await evolver._evolve_captured(ctx)
+        else:
+            logger.warning("Unknown evolution type: %s", evo_type)
+            return None
+    except Exception as e:
+        targets = "+".join(ctx.suggestion.target_skill_ids) or "(new)"
+        logger.error(
+            "Evolution failed [%s] target=%s: %s",
+            evo_type.value,
+            targets,
+            e,
+        )
+        return None
+
+
+# ---------------------------------------------------------------------------
+# Batch execution with throttling
+# ---------------------------------------------------------------------------
+
+async def execute_contexts(
+    evolver,
+    contexts: List[EvolutionContext],
+    trigger_label: str,
+) -> List["SkillRecord"]:
+    """Execute a list of evolution contexts in parallel (throttled).
+
+    Uses *evolver._semaphore* for concurrency control and routes each
+    context through ``dispatch_evolution``.
+    """
+
+    async def _throttled(c: EvolutionContext) -> Optional["SkillRecord"]:
+        async with evolver._semaphore:
+            return await evolver.evolve(c)
+
+    raw = await asyncio.gather(
+        *[_throttled(c) for c in contexts],
+        return_exceptions=True,
+    )
+    results: List["SkillRecord"] = []
+    for r in raw:
+        if isinstance(r, BaseException):
+            logger.error("[Trigger:%s] Evolution task raised: %s", trigger_label, r)
+        elif r is not None:
+            results.append(r)
+
+    if results:
+        names = [r.name for r in results]
+        logger.info(
+            "[Trigger:%s] Evolved %d skill(s): %s",
+            trigger_label,
+            len(results),
+            names,
+        )
+    return results
+
+
+# ---------------------------------------------------------------------------
+# Background task scheduling
+# ---------------------------------------------------------------------------
+
+def schedule_background(
+    background_tasks: set,
+    coro,
+    *,
+    label: str = "background_evolution",
+) -> Optional[asyncio.Task]:
+    """Launch *coro* as a background ``asyncio.Task``.
+
+    *background_tasks* is the ``set`` that tracks outstanding tasks
+    (typically ``SkillEvolver._background_tasks``).  The task is removed
+    from the set on completion via ``discard`` callback.
+    """
+    try:
+        loop = asyncio.get_running_loop()
+    except RuntimeError:
+        logger.warning("No running event loop — cannot schedule %s", label)
+        return None
+
+    task = loop.create_task(coro, name=label)
+    background_tasks.add(task)
+    task.add_done_callback(background_tasks.discard)
+    task.add_done_callback(log_background_result)
+    return task
+
+
+def log_background_result(task: asyncio.Task) -> None:
+    """Log the outcome of a background evolution task."""
+    if task.cancelled():
+        logger.debug("Background task '%s' was cancelled", task.get_name())
+        return
+    exc = task.exception()
+    if exc:
+        logger.error(
+            "Background task '%s' failed: %s",
+            task.get_name(),
+            exc,
+            exc_info=exc,
+        )
diff --git a/openspace/skill_engine/evolver.py b/openspace/skill_engine/evolver.py
index 79c72524..a260122e 100644
--- a/openspace/skill_engine/evolver.py
+++ b/openspace/skill_engine/evolver.py
@@ -33,6 +33,12 @@
     EvolutionTrigger,
     _sanitize_skill_name,
 )
+from .evolution.orchestrator import (
+    dispatch_evolution as _dispatch_evolution,
+    execute_contexts as _execute_contexts_impl,
+    log_background_result as _log_background_result_impl,
+    schedule_background as _schedule_background_impl,
+)
 from .patch import (
     SKILL_FILENAME,
     PatchType,
@@ -181,21 +187,7 @@ async def evolve(self, ctx: EvolutionContext) -> Optional[SkillRecord]:
         The global semaphore is NOT acquired here — it is managed at the
         trigger-method level so the concurrency limit covers the whole batch.
         """
-        evo_type = ctx.suggestion.evolution_type
-        try:
-            if evo_type == EvolutionType.FIX:
-                return await self._evolve_fix(ctx)
-            elif evo_type == EvolutionType.DERIVED:
-                return await self._evolve_derived(ctx)
-            elif evo_type == EvolutionType.CAPTURED:
-                return await self._evolve_captured(ctx)
-            else:
-                logger.warning(f"Unknown evolution type: {evo_type}")
-                return None
-        except Exception as e:
-            targets = "+".join(ctx.suggestion.target_skill_ids) or "(new)"
-            logger.error(f"Evolution failed [{evo_type.value}] target={targets}: {e}")
-            return None
+        return await _dispatch_evolution(self, ctx)
 
     # Trigger 1: post-analysis
     async def process_analysis(
@@ -441,26 +433,7 @@ async def _execute_contexts(
 
         Used by all three triggers after building/confirming contexts.
         """
-
-        async def _throttled(c: EvolutionContext) -> Optional[SkillRecord]:
-            async with self._semaphore:
-                return await self.evolve(c)
-
-        raw = await asyncio.gather(
-            *[_throttled(c) for c in contexts],
-            return_exceptions=True,
-        )
-        results: List[SkillRecord] = []
-        for r in raw:
-            if isinstance(r, BaseException):
-                logger.error(f"[Trigger:{trigger_label}] Evolution task raised: {r}")
-            elif r is not None:
-                results.append(r)
-
-        if results:
-            names = [r.name for r in results]
-            logger.info(f"[Trigger:{trigger_label}] Evolved {len(results)} skill(s): {names}")
-        return results
+        return await _execute_contexts_impl(self, contexts, trigger_label)
 
     def schedule_background(
         self,
@@ -474,30 +447,11 @@ def schedule_background(
         ``background_triggers`` is True.  The task is tracked so it can
         be awaited on shutdown via ``wait_background()``.
         """
-        try:
-            loop = asyncio.get_running_loop()
-        except RuntimeError:
-            logger.warning(f"No running event loop — cannot schedule {label}")
-            return None
-
-        task = loop.create_task(coro, name=label)
-        self._background_tasks.add(task)
-        task.add_done_callback(self._background_tasks.discard)
-        task.add_done_callback(self._log_background_result)
-        return task
+        return _schedule_background_impl(
+            self._background_tasks, coro, label=label,
+        )
 
-    @staticmethod
-    def _log_background_result(task: asyncio.Task) -> None:
-        """Log the outcome of a background evolution task."""
-        if task.cancelled():
-            logger.debug(f"Background task '{task.get_name()}' was cancelled")
-            return
-        exc = task.exception()
-        if exc:
-            logger.error(
-                f"Background task '{task.get_name()}' failed: {exc}",
-                exc_info=exc,
-            )
+    _log_background_result = staticmethod(_log_background_result_impl)
 
     # LLM confirmation for Trigger 2/3
     async def _llm_confirm_evolution(
diff --git a/tests/test_evolution_orchestrator.py b/tests/test_evolution_orchestrator.py
new file mode 100644
index 00000000..8c5d0680
--- /dev/null
+++ b/tests/test_evolution_orchestrator.py
@@ -0,0 +1,448 @@
+"""Tests for openspace.skill_engine.evolution.orchestrator (Epic 5.2).
+
+Verifies:
+  1. dispatch_evolution routes FIX/DERIVED/CAPTURED correctly
+  2. dispatch_evolution handles unknown types and exceptions gracefully
+  3. execute_contexts throttles via semaphore and collects results
+  4. execute_contexts tolerates exceptions in individual contexts
+  5. schedule_background creates and tracks tasks
+  6. schedule_background handles no-event-loop case
+  7. log_background_result logs errors/cancellations
+  8. Backward compat: SkillEvolver.evolve delegates to orchestrator
+  9. Backward compat: SkillEvolver.schedule_background delegates
+  10. Identity: orchestrator functions are the canonical implementations
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import List, Optional, Set
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+# ---------------------------------------------------------------------------
+# Minimal stubs so we don't import the whole skill engine
+# ---------------------------------------------------------------------------
+
+class _FakeEvolutionType(Enum):
+    FIX = "fix"
+    DERIVED = "derived"
+    CAPTURED = "captured"
+    UNKNOWN = "unknown"
+
+
+@dataclass
+class _FakeSuggestion:
+    evolution_type: _FakeEvolutionType = _FakeEvolutionType.FIX
+    target_skill_ids: List[str] = field(default_factory=lambda: ["skill-1"])
+
+
+@dataclass
+class _FakeContext:
+    suggestion: _FakeSuggestion = field(default_factory=_FakeSuggestion)
+
+
+@dataclass
+class _FakeSkillRecord:
+    name: str = "test-skill"
+
+
+class _FakeEvolver:
+    """Minimal duck-typed evolver for testing orchestrator functions."""
+
+    def __init__(self, max_concurrent: int = 3):
+        self._semaphore = asyncio.Semaphore(max_concurrent)
+        self._background_tasks: Set[asyncio.Task] = set()
+        self._evolve_fix = AsyncMock(return_value=_FakeSkillRecord(name="fixed"))
+        self._evolve_derived = AsyncMock(return_value=_FakeSkillRecord(name="derived"))
+        self._evolve_captured = AsyncMock(return_value=_FakeSkillRecord(name="captured"))
+
+    async def evolve(self, ctx):
+        """Delegate to dispatch_evolution, mirroring real SkillEvolver."""
+        return await dispatch_evolution(self, ctx)
+
+
+# ---------------------------------------------------------------------------
+# Import orchestrator — patch EvolutionType so dispatch_evolution resolves
+# ---------------------------------------------------------------------------
+
+from openspace.skill_engine.evolution.orchestrator import (
+    dispatch_evolution,
+    execute_contexts,
+    log_background_result,
+    schedule_background,
+)
+
+
+# ---------------------------------------------------------------------------
+# dispatch_evolution tests
+# ---------------------------------------------------------------------------
+
+class TestDispatchEvolution:
+    @pytest.mark.asyncio
+    async def test_routes_fix(self):
+        evolver = _FakeEvolver()
+        ctx = _FakeContext(_FakeSuggestion(evolution_type=_FakeEvolutionType.FIX))
+
+        with patch("openspace.skill_engine.evolution.orchestrator.EvolutionType", _FakeEvolutionType):
+            result = await dispatch_evolution(evolver, ctx)
+
+        assert result is not None
+        assert result.name == "fixed"
+        evolver._evolve_fix.assert_awaited_once_with(ctx)
+
+    @pytest.mark.asyncio
+    async def test_routes_derived(self):
+        evolver = _FakeEvolver()
+        ctx = _FakeContext(_FakeSuggestion(evolution_type=_FakeEvolutionType.DERIVED))
+
+        with patch("openspace.skill_engine.evolution.orchestrator.EvolutionType", _FakeEvolutionType):
+            result = await dispatch_evolution(evolver, ctx)
+
+        assert result is not None
+        assert result.name == "derived"
+        evolver._evolve_derived.assert_awaited_once_with(ctx)
+
+    @pytest.mark.asyncio
+    async def test_routes_captured(self):
+        evolver = _FakeEvolver()
+        ctx = _FakeContext(_FakeSuggestion(evolution_type=_FakeEvolutionType.CAPTURED))
+
+        with patch("openspace.skill_engine.evolution.orchestrator.EvolutionType", _FakeEvolutionType):
+            result = await dispatch_evolution(evolver, ctx)
+
+        assert result is not None
+        assert result.name == "captured"
+        evolver._evolve_captured.assert_awaited_once_with(ctx)
+
+    @pytest.mark.asyncio
+    async def test_unknown_type_returns_none(self):
+        evolver = _FakeEvolver()
+        ctx = _FakeContext(_FakeSuggestion(evolution_type=_FakeEvolutionType.UNKNOWN))
+
+        with patch("openspace.skill_engine.evolution.orchestrator.EvolutionType", _FakeEvolutionType):
+            result = await dispatch_evolution(evolver, ctx)
+
+        assert result is None
+
+    @pytest.mark.asyncio
+    async def test_exception_returns_none_and_logs(self, caplog):
+        evolver = _FakeEvolver()
+        evolver._evolve_fix = AsyncMock(side_effect=RuntimeError("boom"))
+        ctx = _FakeContext(_FakeSuggestion(evolution_type=_FakeEvolutionType.FIX))
+
+        with patch("openspace.skill_engine.evolution.orchestrator.EvolutionType", _FakeEvolutionType):
+            with caplog.at_level(logging.ERROR):
+                result = await dispatch_evolution(evolver, ctx)
+
+        assert result is None
+        assert "boom" in caplog.text
+
+
+# ---------------------------------------------------------------------------
+# execute_contexts tests
+# ---------------------------------------------------------------------------
+
+class TestExecuteContexts:
+    @pytest.mark.asyncio
+    async def test_runs_all_contexts_and_collects_results(self):
+        evolver = _FakeEvolver()
+        ctxs = [
+            _FakeContext(_FakeSuggestion(evolution_type=_FakeEvolutionType.FIX)),
+            _FakeContext(_FakeSuggestion(evolution_type=_FakeEvolutionType.DERIVED)),
+        ]
+
+        with patch("openspace.skill_engine.evolution.orchestrator.EvolutionType", _FakeEvolutionType):
+            results = await execute_contexts(evolver, ctxs, "test-trigger")
+
+        assert len(results) == 2
+        names = {r.name for r in results}
+        assert "fixed" in names
+        assert "derived" in names
+
+    @pytest.mark.asyncio
+    async def test_empty_contexts_returns_empty(self):
+        evolver = _FakeEvolver()
+
+        with patch("openspace.skill_engine.evolution.orchestrator.EvolutionType", _FakeEvolutionType):
+            results = await execute_contexts(evolver, [], "empty")
+
+        assert results == []
+
+    @pytest.mark.asyncio
+    async def test_exception_in_one_context_does_not_block_others(self, caplog):
+        evolver = _FakeEvolver()
+        evolver._evolve_fix = AsyncMock(side_effect=RuntimeError("fail-fix"))
+        ctxs = [
+            _FakeContext(_FakeSuggestion(evolution_type=_FakeEvolutionType.FIX)),
+            _FakeContext(_FakeSuggestion(evolution_type=_FakeEvolutionType.DERIVED)),
+        ]
+
+        with patch("openspace.skill_engine.evolution.orchestrator.EvolutionType", _FakeEvolutionType):
+            with caplog.at_level(logging.ERROR):
+                results = await execute_contexts(evolver, ctxs, "mixed")
+
+        # Derived should still succeed
+        assert len(results) == 1
+        assert results[0].name == "derived"
+
+    @pytest.mark.asyncio
+    async def test_semaphore_limits_concurrency(self):
+        """Verify semaphore actually gates concurrent executions."""
+        evolver = _FakeEvolver(max_concurrent=1)
+        order = []
+
+        async def slow_fix(ctx):
+            order.append("fix-start")
+            await asyncio.sleep(0.05)
+            order.append("fix-end")
+            return _FakeSkillRecord(name="fixed")
+
+        async def slow_derived(ctx):
+            order.append("derived-start")
+            await asyncio.sleep(0.05)
+            order.append("derived-end")
+            return _FakeSkillRecord(name="derived")
+
+        evolver._evolve_fix = slow_fix
+        evolver._evolve_derived = slow_derived
+
+        ctxs = [
+            _FakeContext(_FakeSuggestion(evolution_type=_FakeEvolutionType.FIX)),
+            _FakeContext(_FakeSuggestion(evolution_type=_FakeEvolutionType.DERIVED)),
+        ]
+
+        with patch("openspace.skill_engine.evolution.orchestrator.EvolutionType", _FakeEvolutionType):
+            results = await execute_contexts(evolver, ctxs, "serial")
+
+        assert len(results) == 2
+        # With semaphore=1, tasks run serially: first ends before second starts
+        fix_end = order.index("fix-end")
+        derived_start = order.index("derived-start")
+        assert fix_end < derived_start, f"Expected serial execution, got {order}"
+
+
+# ---------------------------------------------------------------------------
+# schedule_background tests
+# ---------------------------------------------------------------------------
+
+class TestScheduleBackground:
+    @pytest.mark.asyncio
+    async def test_creates_and_tracks_task(self):
+        tasks: Set[asyncio.Task] = set()
+
+        async def noop():
+            return 42
+
+        task = schedule_background(tasks, noop(), label="test-bg")
+        assert task is not None
+        assert task in tasks
+        assert task.get_name() == "test-bg"
+
+        result = await task
+        assert result == 42
+        # done callback should have removed it
+        assert task not in tasks
+
+    @pytest.mark.asyncio
+    async def test_failed_task_removed_and_logged(self, caplog):
+        tasks: Set[asyncio.Task] = set()
+
+        async def fail():
+            raise ValueError("bg-boom")
+
+        task = schedule_background(tasks, fail(), label="fail-bg")
+        assert task is not None
+
+        with caplog.at_level(logging.ERROR):
+            # Wait for task to complete (it will raise)
+            await asyncio.sleep(0.1)
+
+        assert task not in tasks
+        assert "bg-boom" in caplog.text
+
+    def test_no_event_loop_returns_none(self, caplog):
+        """When called outside an event loop, returns None gracefully."""
+        tasks: set = set()
+
+        async def noop():
+            pass
+
+        coro = noop()
+        # We need to be outside a running loop for this test
+        # Since pytest-asyncio provides a loop, we test via a sync function
+        # that patches get_running_loop to raise
+        with patch("openspace.skill_engine.evolution.orchestrator.asyncio") as mock_asyncio:
+            mock_asyncio.get_running_loop.side_effect = RuntimeError("no loop")
+            with caplog.at_level(logging.WARNING):
+                result = schedule_background(tasks, coro, label="no-loop")
+
+        assert result is None
+        assert len(tasks) == 0
+        # Clean up the coroutine
+        coro.close()
+
+
+# ---------------------------------------------------------------------------
+# log_background_result tests
+# ---------------------------------------------------------------------------
+
+class TestLogBackgroundResult:
+    def test_cancelled_task_logs_debug(self, caplog):
+        task = MagicMock()
+        task.cancelled.return_value = True
+        task.get_name.return_value = "cancelled-task"
+
+        with caplog.at_level(logging.DEBUG):
+            log_background_result(task)
+
+        assert "cancelled" in caplog.text
+
+    def test_failed_task_logs_error(self, caplog):
+        task = MagicMock()
+        task.cancelled.return_value = False
+        task.exception.return_value = RuntimeError("task-error")
+        task.get_name.return_value = "error-task"
+
+        with caplog.at_level(logging.ERROR):
+            log_background_result(task)
+
+        assert "task-error" in caplog.text
+
+    def test_successful_task_logs_nothing(self, caplog):
+        task = MagicMock()
+        task.cancelled.return_value = False
+        task.exception.return_value = None
+        task.get_name.return_value = "ok-task"
+
+        with caplog.at_level(logging.DEBUG):
+            log_background_result(task)
+
+        assert "ok-task" not in caplog.text
+
+
+# ---------------------------------------------------------------------------
+# Backward compatibility — SkillEvolver delegates to orchestrator
+# ---------------------------------------------------------------------------
+
+try:
+    from openspace.skill_engine.evolution.orchestrator import (
+        dispatch_evolution as _canonical_dispatch,
+        execute_contexts as _canonical_execute,
+        log_background_result as _canonical_log,
+        schedule_background as _canonical_schedule,
+    )
+    _HAS_COMPAT = True
+except ImportError:
+    _HAS_COMPAT = False
+
+
+@pytest.mark.skipif(not _HAS_COMPAT, reason="orchestrator not importable")
+class TestBackwardCompat:
+    def test_orchestrator_importable_from_evolution_package(self):
+        """Canonical import path works."""
+        from openspace.skill_engine.evolution import orchestrator
+        assert hasattr(orchestrator, "dispatch_evolution")
+        assert hasattr(orchestrator, "execute_contexts")
+        assert hasattr(orchestrator, "schedule_background")
+        assert hasattr(orchestrator, "log_background_result")
+
+    def test_functions_are_canonical(self):
+        """The orchestrator module functions are the real implementations."""
+        import types
+        assert isinstance(_canonical_dispatch, types.FunctionType)
+        assert isinstance(_canonical_execute, types.FunctionType)
+        assert isinstance(_canonical_schedule, types.FunctionType)
+        assert isinstance(_canonical_log, types.FunctionType)
+
+
+# ---------------------------------------------------------------------------
+# Delegation integration — SkillEvolver → orchestrator seam
+# ---------------------------------------------------------------------------
+
+try:
+    from openspace.skill_engine.evolver import SkillEvolver
+    _HAS_EVOLVER = True
+except ImportError:
+    _HAS_EVOLVER = False
+
+
+@pytest.mark.skipif(not _HAS_EVOLVER, reason="SkillEvolver not importable (litellm)")
+class TestDelegationSeam:
+    """Verify SkillEvolver methods actually delegate to orchestrator functions."""
+
+    @pytest.mark.asyncio
+    async def test_evolve_reaches_dispatch(self):
+        """SkillEvolver.evolve() → dispatch_evolution → _evolve_fix."""
+        evolver = object.__new__(SkillEvolver)  # skip __init__
+        called_with = []
+
+        async def fake_fix(ctx):
+            called_with.append(ctx)
+            return _FakeSkillRecord(name="via-evolve")
+
+        evolver._evolve_fix = fake_fix
+
+        ctx = _FakeContext(_FakeSuggestion(evolution_type=_FakeEvolutionType.FIX))
+        with patch("openspace.skill_engine.evolution.orchestrator.EvolutionType", _FakeEvolutionType):
+            result = await evolver.evolve(ctx)
+
+        assert result is not None
+        assert result.name == "via-evolve"
+        assert called_with == [ctx], "evolve() should delegate through to _evolve_fix"
+
+    @pytest.mark.asyncio
+    async def test_execute_contexts_goes_through_evolve(self):
+        """_execute_contexts calls self.evolve(), not dispatch_evolution directly."""
+        evolver = object.__new__(SkillEvolver)
+        evolver._semaphore = asyncio.Semaphore(2)
+        evolve_calls = []
+
+        async def tracking_evolve(ctx):
+            evolve_calls.append(ctx)
+            return _FakeSkillRecord(name="tracked")
+
+        evolver.evolve = tracking_evolve
+
+        ctxs = [
+            _FakeContext(_FakeSuggestion(evolution_type=_FakeEvolutionType.FIX)),
+            _FakeContext(_FakeSuggestion(evolution_type=_FakeEvolutionType.DERIVED)),
+        ]
+        results = await evolver._execute_contexts(ctxs, "seam-test")
+
+        assert len(results) == 2
+        assert len(evolve_calls) == 2, "Batch path must go through evolve(), not bypass it"
+
+    @pytest.mark.asyncio
+    async def test_schedule_background_delegates(self):
+        """SkillEvolver.schedule_background() tracks task in _background_tasks."""
+        evolver = object.__new__(SkillEvolver)
+        evolver._background_tasks = set()
+
+        async def noop():
+            return "bg-done"
+
+        task = evolver.schedule_background(noop(), label="seam-bg")
+        assert task is not None
+        assert task in evolver._background_tasks
+
+        result = await task
+        assert result == "bg-done"
+
+
+# ---------------------------------------------------------------------------
+# Size guard
+# ---------------------------------------------------------------------------
+
+class TestSizeGuard:
+    def test_orchestrator_module_size(self):
+        """orchestrator.py should stay focused (< 160 lines)."""
+        from pathlib import Path
+        mod_path = Path(__file__).resolve().parent.parent / "openspace" / "skill_engine" / "evolution" / "orchestrator.py"
+        assert mod_path.exists(), f"orchestrator.py not found at {mod_path}"
+        lines = mod_path.read_text(encoding="utf-8").splitlines()
+        assert len(lines) < 160, f"orchestrator.py has {len(lines)} lines (limit 160)"