Phase 16.4 — `SelfOptimiser`: action-enacting layer for the self-improvement pipeline

[web3guru888]

Phase 16.4 — SelfOptimiser: action-enacting layer for the self-improvement pipeline

Phase: 16 — Cognitive Reflection & Self-Improvement
Component: SelfOptimiser
Depends on: Phase 16.3 ImprovementPlanner (#423), Phase 15.5 LiveModuleOrchestrator (#413), Phase 10.5 ReplanningEngine (#333)
Feeds into: Phase 16.5 ReflectionCycle

---

Motivation

ImprovementPlanner produces a priority-ranked list of ImprovementAction items. SelfOptimiser is the executor layer — it inspects each action, routes it to the correct subsystem (orchestrator, replanner, budget manager, or safety escalation), tracks outcomes, and surfaces enactment telemetry to ReflectionCycle for retrospective analysis.

Without SelfOptimiser, the planning pipeline is read-only: it diagnoses and proposes but never acts. SelfOptimiser closes that loop, making the system genuinely self-improving.

---

Data Model

from __future__ import annotations
from dataclasses import dataclass, field
from enum import Enum, auto
from typing import Sequence

class EnactmentStatus(Enum):
    SUCCESS      = auto()   # action fully applied
    PARTIAL      = auto()   # action applied but with reduced effect
    SKIPPED      = auto()   # preconditions not met; action deferred
    FAILED       = auto()   # subsystem call raised or returned error
    RATE_LIMITED = auto()   # cool-down window active; action deferred

@dataclass(frozen=True)
class EnactmentRecord:
    """Outcome of a single ImprovementAction enactment attempt."""
    action:    ImprovementAction
    status:    EnactmentStatus
    detail:    str
    duration_s: float           # wall-clock seconds for enactment
    attempted_at: float         # time.monotonic() timestamp

@dataclass(frozen=True)
class OptimiserConfig:
    max_enactments_per_cycle: int   = 5      # cap applied actions per reflection cycle
    cool_down_s:              float = 60.0   # minimum gap between two same-kind actions on same module
    dry_run:                  bool  = False  # log-only mode; no subsystem calls made
    enable_rate_limit:        bool  = True
    max_hot_swap_per_cycle:   int   = 2      # HOT_SWAP_MODULE actions are expensive; separate cap

---

Protocol

from typing import Protocol, runtime_checkable

@runtime_checkable
class SelfOptimiser(Protocol):
    async def enact(
        self,
        actions: Sequence[ImprovementAction],
        *,
        config: OptimiserConfig | None = None,
    ) -> list[EnactmentRecord]: ...

    async def stats(self) -> dict[str, int]: ...
    async def reset(self) -> None: ...

---

Dispatcher: ActionKind → Subsystem

ActionKind	Subsystem Called	Notes
TUNE_THRESHOLD	DetectorConfig hot-update via _apply_tune	Adjusts WeaknessDetector thresholds in-place
INCREASE_BUDGET	CognitiveCycle._budget_map	Raises allocated wall-time for target module
REDUCE_LOAD	CognitiveCycle._load_shedder	Signals load-shedder to de-prioritise module tasks
HOT_SWAP_MODULE	LiveModuleOrchestrator.orchestrate_swap()	Full Phase 15 swap pipeline
FLAG_FOR_REVIEW	asyncio.Queue → operator alert channel	Enqueued for human review; always SUCCESS
NO_OP	—	Logged and counted; always SKIPPED

---

AsyncSelfOptimiser — Reference Implementation

import asyncio, logging, time
from collections import defaultdict
from typing import Sequence

log = logging.getLogger("asi_build.self_optimiser")

class AsyncSelfOptimiser:
    """
    Routes ImprovementAction items to the appropriate subsystem.
    Rate-limits repeated same-kind actions on the same module.
    """

    def __init__(
        self,
        orchestrator=None,     # LiveModuleOrchestrator | None
        cognitive_cycle=None,  # CognitiveCycle | None (duck-typed)
        alert_queue: asyncio.Queue | None = None,
        config: OptimiserConfig | None = None,
    ) -> None:
        self._orch   = orchestrator
        self._cycle  = cognitive_cycle
        self._alerts = alert_queue or asyncio.Queue()
        self._cfg    = config or OptimiserConfig()
        self._last_enacted: dict[tuple[str, str], float] = {}  # (module, kind) → monotonic
        self._lock   = asyncio.Lock()
        self._stats: dict[str, int] = defaultdict(int)

    # ------------------------------------------------------------------ #
    #  Public interface                                                    #
    # ------------------------------------------------------------------ #

    async def enact(
        self,
        actions: Sequence[ImprovementAction],
        *,
        config: OptimiserConfig | None = None,
    ) -> list[EnactmentRecord]:
        cfg = config or self._cfg
        records: list[EnactmentRecord] = []
        hot_swap_count = 0

        async with self._lock:
            for action in actions:
                if len(records) >= cfg.max_enactments_per_cycle:
                    break

                # Rate-limit gate
                key = (action.module_name, action.action_kind.name)
                if cfg.enable_rate_limit:
                    last = self._last_enacted.get(key, 0.0)
                    if time.monotonic() - last < cfg.cool_down_s:
                        records.append(EnactmentRecord(
                            action=action,
                            status=EnactmentStatus.RATE_LIMITED,
                            detail=f"cool-down active ({cfg.cool_down_s}s)",
                            duration_s=0.0,
                            attempted_at=time.monotonic(),
                        ))
                        self._stats["rate_limited"] += 1
                        continue

                # HOT_SWAP cap
                if action.action_kind.name == "HOT_SWAP_MODULE":
                    if hot_swap_count >= cfg.max_hot_swap_per_cycle:
                        records.append(EnactmentRecord(
                            action=action,
                            status=EnactmentStatus.SKIPPED,
                            detail="hot-swap cap reached for this cycle",
                            duration_s=0.0,
                            attempted_at=time.monotonic(),
                        ))
                        self._stats["skipped"] += 1
                        continue
                    hot_swap_count += 1

                record = await self._dispatch(action, cfg)
                records.append(record)
                if record.status == EnactmentStatus.SUCCESS:
                    self._last_enacted[key] = time.monotonic()
                    self._stats["enacted"] += 1
                elif record.status == EnactmentStatus.FAILED:
                    self._stats["failed"] += 1
                else:
                    self._stats["skipped"] += 1

            self._stats["enact_calls"] += 1
        return records

    async def stats(self) -> dict[str, int]:
        return dict(self._stats)

    async def reset(self) -> None:
        async with self._lock:
            self._stats.clear()
            self._last_enacted.clear()

    # ------------------------------------------------------------------ #
    #  Dispatcher                                                          #
    # ------------------------------------------------------------------ #

    async def _dispatch(
        self, action: ImprovementAction, cfg: OptimiserConfig
    ) -> EnactmentRecord:
        t0 = time.monotonic()
        params = dict(action.parameters)

        if cfg.dry_run:
            return self._record(action, EnactmentStatus.SKIPPED,
                                f"dry-run: would enact {action.action_kind.name}", t0)

        match action.action_kind.name:
            case "TUNE_THRESHOLD":
                return await self._apply_tune(action, params, t0)
            case "INCREASE_BUDGET":
                return await self._apply_budget(action, params, t0)
            case "REDUCE_LOAD":
                return await self._apply_load_shed(action, params, t0)
            case "HOT_SWAP_MODULE":
                return await self._apply_hot_swap(action, params, t0)
            case "FLAG_FOR_REVIEW":
                await self._alerts.put(action)
                return self._record(action, EnactmentStatus.SUCCESS, "queued for review", t0)
            case "NO_OP" | _:
                return self._record(action, EnactmentStatus.SKIPPED, "no-op", t0)

    async def _apply_tune(self, action, params, t0) -> EnactmentRecord:
        """Adjust a WeaknessDetector threshold for the target module."""
        try:
            delta = float(params.get("suggested_delta", "0.1"))
            if self._cycle and hasattr(self._cycle, "_detector_config"):
                cfg = self._cycle._detector_config
                # Clamp new threshold to reasonable range
                new_p95 = max(50.0, cfg.latency_p95_threshold_ms * (1 + delta))
                self._cycle._detector_config = DetectorConfig(
                    latency_p95_threshold_ms=new_p95,
                    latency_p99_threshold_ms=cfg.latency_p99_threshold_ms,
                    error_rate_threshold=cfg.error_rate_threshold,
                )
                return self._record(action, EnactmentStatus.SUCCESS,
                                    f"p95_threshold adjusted to {new_p95:.1f}ms", t0)
            return self._record(action, EnactmentStatus.SKIPPED,
                                "no detector_config available on cycle", t0)
        except Exception as exc:
            return self._record(action, EnactmentStatus.FAILED, str(exc), t0)

    async def _apply_budget(self, action, params, t0) -> EnactmentRecord:
        """Increase the time budget allocated to the target module."""
        try:
            factor = float(params.get("budget_factor", "1.5"))
            if self._cycle and hasattr(self._cycle, "_budget_map"):
                old = self._cycle._budget_map.get(action.module_name, 1.0)
                self._cycle._budget_map[action.module_name] = old * factor
                return self._record(action, EnactmentStatus.SUCCESS,
                                    f"budget {old:.2f}→{old*factor:.2f}", t0)
            return self._record(action, EnactmentStatus.SKIPPED, "no budget_map on cycle", t0)
        except Exception as exc:
            return self._record(action, EnactmentStatus.FAILED, str(exc), t0)

    async def _apply_load_shed(self, action, params, t0) -> EnactmentRecord:
        """Signal the load-shedder to reduce pressure on the target module."""
        try:
            if self._cycle and hasattr(self._cycle, "_load_shedder"):
                await self._cycle._load_shedder.shed(action.module_name)
                return self._record(action, EnactmentStatus.SUCCESS,
                                    "load-shed signal sent", t0)
            return self._record(action, EnactmentStatus.SKIPPED, "no load_shedder on cycle", t0)
        except Exception as exc:
            return self._record(action, EnactmentStatus.FAILED, str(exc), t0)

    async def _apply_hot_swap(self, action, params, t0) -> EnactmentRecord:
        """Invoke Phase 15 LiveModuleOrchestrator to swap the module."""
        if self._orch is None:
            return self._record(action, EnactmentStatus.SKIPPED, "no orchestrator injected", t0)
        try:
            import hashlib, os
            req = OrchestratorRequest(
                module_id   = action.module_name,
                new_code    = b"",              # placeholder; real impl fetches from CodeSynthesiser
                new_version = params.get("target_version", "auto"),
                requester   = "self_optimiser",
            )
            result = await self._orch.orchestrate_swap(req)
            status = (EnactmentStatus.SUCCESS if result.outcome.name == "SUCCESS"
                      else EnactmentStatus.FAILED)
            return self._record(action, status,
                                f"orchestrator: {result.outcome.name} — {result.detail}", t0)
        except Exception as exc:
            return self._record(action, EnactmentStatus.FAILED, str(exc), t0)

    @staticmethod
    def _record(action, status, detail, t0) -> EnactmentRecord:
        return EnactmentRecord(
            action=action, status=status, detail=detail,
            duration_s=time.monotonic() - t0,
            attempted_at=time.monotonic(),
        )


class NullOptimiser:
    """No-op optimiser — for testing or disabled-enactment mode."""
    async def enact(self, actions, *, config=None) -> list[EnactmentRecord]:
        return []
    async def stats(self) -> dict[str, int]:
        return {}
    async def reset(self) -> None:
        pass

---

CognitiveCycle Integration

# In CognitiveCycle._reflection_step():
profiles  = await self._profiler.stats()
reports   = await self._weakness_detector.analyse(list(profiles.values()))
actions   = await self._improvement_planner.plan(reports)
records   = await self._self_optimiser.enact(actions)
# records are forwarded to ReflectionCycle (16.5) for retrospective logging
await self._reflection_cycle.record_enactments(records)

---

Prometheus Metrics

Metric	Type	Labels
asi_optimiser_enact_calls_total	Counter	—
asi_optimiser_actions_enacted_total	Counter	action_kind, status
asi_optimiser_actions_rate_limited_total	Counter	module, action_kind
asi_optimiser_hot_swaps_total	Counter	module, outcome
asi_optimiser_enact_duration_seconds	Histogram	action_kind

# Enactment failure rate
rate(asi_optimiser_actions_enacted_total{status="FAILED"}[5m])
  / rate(asi_optimiser_actions_enacted_total[5m]) > 0.3

# Alert: hot-swap success rate below 80%
rate(asi_optimiser_hot_swaps_total{outcome="SUCCESS"}[10m])
  / rate(asi_optimiser_hot_swaps_total[10m]) < 0.8

---

Test Targets (12)

1. test_flag_for_review_always_success — FLAG_FOR_REVIEW always enqueued; status=SUCCESS
2. test_no_op_returns_skipped — NO_OP action → SKIPPED record
3. test_dry_run_skips_all — dry_run=True → all records SKIPPED, no subsystem calls
4. test_rate_limit_gate — same module+kind twice within cool_down_s → second RATE_LIMITED
5. test_rate_limit_expires — after cool_down_s elapses, second enactment proceeds
6. test_hot_swap_cap — max_hot_swap_per_cycle=1, two HOT_SWAP actions → second SKIPPED
7. test_max_enactments_cap — max_enactments_per_cycle=2, 5 actions → 2 records returned
8. test_orchestrator_success — mock orchestrator → SUCCESS EnactmentRecord
9. test_orchestrator_failure — mock orchestrator raises → FAILED record, no exception propagation
10. test_tune_threshold_adjusts_config — CognitiveCycle mock with _detector_config attribute
11. test_concurrent_enact_calls — asyncio.gather 3× enact → Lock prevents race
12. test_reset_clears_state — reset() → stats() empty, last_enacted cleared

---

Implementation Order (14 steps)

1. EnactmentStatus enum
2. EnactmentRecord frozen dataclass
3. OptimiserConfig frozen dataclass
4. SelfOptimiser Protocol (@runtime_checkable)
5. NullOptimiser no-op
6. AsyncSelfOptimiser.__init__ (inject orchestrator + cycle + alert_queue)
7. _apply_tune() dispatcher method
8. _apply_budget() dispatcher method
9. _apply_load_shed() dispatcher method
10. _apply_hot_swap() dispatcher method (delegates to LiveModuleOrchestrator)
11. _dispatch() match statement + FLAG_FOR_REVIEW + NO_OP branches
12. enact() — rate-limit gate + hot-swap cap + loop + stats
13. Prometheus metrics (instrument enact() with histogram + counters)
14. All 12 tests green; mypy --strict clean

Show & Tell: #426 | Q&A: #427 | Wiki: Phase 16 — SelfOptimiser

---

Phase 16 Sub-Phase Tracker

Sub-phase	Component	Issue	Status
16.1	PerformanceProfiler	#417	🟡 Spec filed
16.2	WeaknessDetector	#420	🟡 Spec filed
16.3	ImprovementPlanner	#423	🟡 Spec filed
16.4	SelfOptimiser	this	🟡 Spec filed
16.5	ReflectionCycle	—	⏳ Upcoming

[web3guru888]

Implementation Notes

_last_enacted dict key design

Using (module_name, action_kind.name) as the rate-limit key means:

• Same module + same ActionKind is throttled after success
• Different ActionKinds on the same module are not cross-throttled
• After reset(), all rate limits are cleared (e.g. for test isolation)

key = (action.module_name, action.action_kind.name)   # str × str — hashable

Avoid using action_kind (the enum value) directly as a dict key if you plan to serialise state — .name is stable across restarts.

---

Lock scope: why hold _lock for the full enact() call

The _lock is held across the entire action loop (not per-action) to prevent two concurrent enact() calls from both passing the rate-limit check for the same (module, kind) key simultaneously. If you need parallel enactment later, consider per-key locks instead.

# Current: coarse lock — serialises all enactments
async with self._lock:
    for action in actions: ...

# Future option: fine-grained — parallel enactments for different keys
# async with self._key_locks[key]: ...

---

HOT_SWAP_MODULE new_code=b"" placeholder

In the reference implementation, _apply_hot_swap sends new_code=b"" to the orchestrator. In production, SelfOptimiser should receive a CodeSynthesiser handle (Phase 14.1) and fetch the compiled patch bytes:

new_code = await self._synthesiser.get_artifact(action.module_name)

For spec purposes, the placeholder keeps SelfOptimiser decoupled from CodeSynthesiser — inject as optional dependency.

---

dry_run mode test pattern

cfg = OptimiserConfig(dry_run=True)
opt = AsyncSelfOptimiser(orchestrator=mock_orch, config=cfg)
records = await opt.enact([hot_swap_action])
assert all(r.status == EnactmentStatus.SKIPPED for r in records)
mock_orch.orchestrate_swap.assert_not_called()

---

mypy table

Symbol	Notes
EnactmentRecord	frozen=True; attempted_at: float avoids time.monotonic return-type issues
SelfOptimiser	@runtime_checkable Protocol
_last_enacted	dict[tuple[str, str], float] — explicit type annotation required
_dispatch return	always EnactmentRecord, no implicit None path
match action.action_kind.name	match on str, not enum — avoids incomplete-match warning

---

test_concurrent_enact_calls skeleton

async def test_concurrent_enact_calls():
    opt = AsyncSelfOptimiser()
    actions = [make_flag_action("modA"), make_flag_action("modB"), make_noop_action("modC")]
    results = await asyncio.gather(
        opt.enact([actions[0]]),
        opt.enact([actions[1]]),
        opt.enact([actions[2]]),
    )
    assert all(len(r) == 1 for r in results)

async def test_rate_limit_expires():
    cfg = OptimiserConfig(cool_down_s=0.01, enable_rate_limit=True)
    opt = AsyncSelfOptimiser(config=cfg)
    act = make_flag_action("modX")
    r1 = await opt.enact([act])
    await asyncio.sleep(0.02)        # let cool-down expire
    r2 = await opt.enact([act])
    assert r1[0].status == EnactmentStatus.SUCCESS
    assert r2[0].status == EnactmentStatus.SUCCESS   # not RATE_LIMITED

[web3guru888]

Implementation Notes — AsyncSelfOptimiser

Rate-limiter token-bucket pattern

The rate_limit_per_minute cap is enforced with a minimal token-bucket that refills on a 60-second rolling window:

import asyncio, time
from collections import deque

class _RateLimiter:
    def __init__(self, max_per_minute: int) -> None:
        self._max   = max_per_minute
        self._calls: deque[float] = deque()

    async def acquire(self) -> None:
        while True:
            now = time.monotonic()
            # evict calls older than 60 s
            while self._calls and now - self._calls[0] > 60.0:
                self._calls.popleft()
            if len(self._calls) < self._max:
                self._calls.append(now)
                return
            wait = 60.0 - (now - self._calls[0]) + 0.01
            await asyncio.sleep(wait)

Call await self._limiter.acquire() before each _execute_single() invocation.

---

Semaphore controls concurrent actions (not per-module)

self._sem = asyncio.Semaphore(config.max_concurrent_actions)

async def _execute_single(self, action) -> OptimiserResult:
    async with self._sem:
        await self._limiter.acquire()
        ...dispatch...

This is different from Phase 15.5 (LiveModuleOrchestrator), which uses a per-module lock inside HotSwapper. Here the Semaphore bounds total in-flight optimisation actions regardless of which modules they touch.

---

HOT_SWAP wiring to Phase 15 LiveModuleOrchestrator

async def _do_hot_swap(self, action: OptimisationAction) -> OptimiserResult:
    if self._cfg.dry_run:
        return OptimiserResult(
            action_type=ActionType.HOT_SWAP, module_name=action.module_name,
            success=True, error_msg="dry-run", duration_ms=0.0, metadata={"dry_run": True}
        )
    req = OrchestratorRequest(
        module_name  = action.module_name,
        new_artifact = action.artifact,      # bytes | Path
        requested_by = "SelfOptimiser",
        metadata     = action.metadata,
    )
    result = await self._orchestrator.orchestrate(req)   # Phase 15.5
    return OptimiserResult(
        action_type = ActionType.HOT_SWAP,
        module_name = action.module_name,
        success     = result.outcome == SwapOutcome.SUCCESS,
        error_msg   = result.error_msg or "",
        duration_ms = result.duration_ms,
        metadata    = {"swap_outcome": result.outcome.value},
    )

---

CONFIG_TUNE uses dataclass.replace()

async def _do_config_tune(self, action: OptimisationAction) -> OptimiserResult:
    current = self._config_registry.get(action.module_name)
    updated = dataclasses.replace(current, **action.param_overrides)
    self._config_registry.set(action.module_name, updated)
    return OptimiserResult(..., success=True, metadata={"params": action.param_overrides})

Keep the config registry (dict[str, dataclass]) as an injected dependency so tests can supply a mock.

---

SANDBOX_TEST wiring to Phase 14 SandboxRunner

async def _do_sandbox_test(self, action: OptimisationAction) -> OptimiserResult:
    run = SandboxRun(
        code       = action.code_patch,
        test_suite = action.test_suite,
        timeout_s  = self._cfg.timeout_seconds,
    )
    result = await self._sandbox.run(run)    # Phase 14.2
    return OptimiserResult(
        action_type = ActionType.SANDBOX_TEST,
        success     = result.verdict == Verdict.PASS,
        ...
    )

---

mypy narrowing

Pattern	Tip
result: OptimiserResult | None	Use assert result is not None or if result is None: return before accessing fields
action.action_type in match	match action.action_type: — mypy narrows within each case arm
frozenset membership	if ActionType.HOT_SWAP not in self._cfg.enabled_action_types:

---

Test skeletons

@pytest.mark.asyncio
async def test_rate_limit_throttles():
    cfg = OptimiserConfig(rate_limit_per_minute=2, max_concurrent_actions=4)
    opt = AsyncSelfOptimiser(cfg, orchestrator=mock_orch, sandbox=mock_sb, config_registry={})
    actions = [make_noop_action() for _ in range(5)]
    t0 = time.monotonic()
    await opt.execute_batch(actions)
    elapsed = time.monotonic() - t0
    # 3rd action should be throttled — at least 60 s would pass in production;
    # in test use rate_limit_per_minute=120 so 60/120 = 0.5 s minimum between calls.
    # Adjust as needed for CI speed.
    assert elapsed >= 0.0  # placeholder — tighten with mock clock

@pytest.mark.asyncio
async def test_dry_run_noop():
    cfg = OptimiserConfig(dry_run=True, rate_limit_per_minute=60)
    opt = AsyncSelfOptimiser(cfg, orchestrator=mock_orch, sandbox=mock_sb, config_registry={})
    action = make_hot_swap_action("module_x", artifact=b"fake")
    result = await opt.execute(action)
    assert result.success is True
    assert result.metadata.get("dry_run") is True
    mock_orch.orchestrate.assert_not_called()

---

Phase 16 sub-phase tracker: 16.1 PerformanceProfiler #417 🟡 · 16.2 WeaknessDetector #420 🟡 · 16.3 ImprovementPlanner #423 🟡 · 16.4 SelfOptimiser #426 🟡 · 16.5 ReflectionCycle ⏳

[web3guru888]

Implementation Notes — SelfOptimiser (Phase 16.4)

1. Token-Bucket Rate Limiter

import asyncio, time

class _TokenBucket:
    """Thread-safe token-bucket for rate-limiting optimise actions."""

    def __init__(self, rate: float, capacity: float) -> None:
        self._rate     = rate       # tokens per second
        self._capacity = capacity   # max burst
        self._tokens   = capacity
        self._last     = time.monotonic()
        self._lock     = asyncio.Lock()

    async def acquire(self) -> None:
        async with self._lock:
            now = time.monotonic()
            self._tokens = min(
                self._capacity,
                self._tokens + (now - self._last) * self._rate,
            )
            self._last = now
            if self._tokens < 1.0:
                wait = (1.0 - self._tokens) / self._rate
                await asyncio.sleep(wait)
                self._tokens = 0.0
            else:
                self._tokens -= 1.0

SelfOptimiserConfig.max_actions_per_minute maps to rate = max_actions_per_minute / 60.0.

---

2. Semaphore vs Per-Action Lock

• Use asyncio.Semaphore(max_concurrent_actions) as a concurrency cap across all running action coroutines.
• Use per-module asyncio.Lock (keyed by action.target_module) to prevent two concurrent actions on the same module.

async def _execute_one(self, action: ImprovementAction) -> OptimiseResult:
    await self._bucket.acquire()                        # rate limit
    async with self._semaphore:                         # concurrency cap
        async with self._module_locks[action.target_module]:  # per-module
            return await self._dispatch(action)

---

3. HOT_SWAP wiring to Phase 15 LiveModuleOrchestrator

async def _handle_hot_swap(self, action: ImprovementAction) -> OptimiseResult:
    request = OrchestratorRequest(
        module_name=action.target_module,
        new_version=action.payload["new_version"],
        requester="SelfOptimiser",
    )
    result = await self._orchestrator.orchestrate(request)
    applied = result.outcome == SwapOutcome.SUCCESS
    return OptimiseResult(action=action, applied=applied, details=result.outcome.name)

---

4. CONFIG_TUNE pattern (dataclass.replace)

async def _handle_config_tune(self, action: ImprovementAction) -> OptimiseResult:
    module = self._module_registry.get(action.target_module)
    old_config = module.config
    new_config = dataclasses.replace(old_config, **action.payload["overrides"])
    await module.apply_config(new_config)
    return OptimiseResult(action=action, applied=True,
                          details=f"config tuned: {action.payload[overrides]}")

Key: dataclasses.replace() returns a new frozen dataclass — safe for concurrent readers.

---

5. SANDBOX_TEST wiring to Phase 14 SandboxRunner

async def _handle_sandbox_test(self, action: ImprovementAction) -> OptimiseResult:
    job = SandboxJob(
        code=action.payload["patch"],
        timeout_seconds=action.payload.get("timeout", 30),
    )
    run_result = await self._sandbox.run(job)
    if not run_result.passed:
        return OptimiseResult(action=action, applied=False,
                              details=f"sandbox failed: {run_result.stderr[:200]}")
    # patch passed — hand off to HOT_SWAP
    return await self._handle_hot_swap(action)

---

6. dry_run passthrough

async def _dispatch(self, action: ImprovementAction) -> OptimiseResult:
    if self._config.dry_run:
        return OptimiseResult(action=action, applied=False, details="dry_run=True")
    match action.action_type:
        case ActionType.HOT_SWAP:      return await self._handle_hot_swap(action)
        case ActionType.CONFIG_TUNE:   return await self._handle_config_tune(action)
        case ActionType.SANDBOX_TEST:  return await self._handle_sandbox_test(action)
        case ActionType.RETRAIN:       return await self._handle_retrain(action)
        case _:                        return OptimiseResult(action=action, applied=False,
                                                             details="unknown action type")

---

7. mypy Narrowing Tips

Situation	Fix
action.payload is dict[str, Any]	Cast: cast(str, action.payload["new_version"])
match action.action_type exhaustive	Add case _: assert_never(action.action_type)
results list from gather	Annotate: results: list[OptimiseResult] = list(await asyncio.gather(*coros))
module_locks defaultdict	defaultdict[str, asyncio.Lock] with asyncio.Lock factory

---

8. Test Skeletons

@pytest.mark.asyncio
async def test_rate_limit_throttles():
    """Verify token bucket slows execution when rate exceeded."""
    config = SelfOptimiserConfig(max_actions_per_minute=6)  # 0.1/s
    optimiser = AsyncSelfOptimiser(config=config, ...)
    actions = [make_action(ActionType.CONFIG_TUNE) for _ in range(3)]
    t0 = time.monotonic()
    await optimiser.execute_batch(actions)
    elapsed = time.monotonic() - t0
    assert elapsed >= 20.0  # 3 actions at 0.1/s = 20s minimum

@pytest.mark.asyncio
async def test_dry_run_noop():
    """dry_run=True must return applied=False for all actions without side effects."""
    mock_orchestrator = AsyncMock()
    config = SelfOptimiserConfig(dry_run=True)
    optimiser = AsyncSelfOptimiser(config=config, orchestrator=mock_orchestrator, ...)
    action = make_action(ActionType.HOT_SWAP, payload={"new_version": "2.0.0"})
    results = await optimiser.execute_batch([action])
    assert all(not r.applied for r in results)
    mock_orchestrator.orchestrate.assert_not_called()

---

Phase 16 Sub-Phase Tracker

Sub-phase	Component	Issue
16.1	PerformanceProfiler	#417 🟡
16.2	WeaknessDetector	#420 🟡
16.3	ImprovementPlanner	#423 🟡
16.4	SelfOptimiser	#426 🟡
16.5	ReflectionCycle	#430 🟡

Phase 16 COMPLETE — ReflectionCycle (#430) filed!