Extending¶

Five extension points, all the same shape: implement a small protocol, register it under a name, select it by that name in a config. None of them requires changing skyfall-crl.

You want to add	You implement	You register with
A reward signal	`RewardComponent`	`rewards.register`
An algorithm	`Algorithm` (a Protocol — inherit nothing)	`train.register`
A way to decide the configuration	`RegimeProvider`	`regime.register_provider`
A different deployment	`MorpheusBackend`	passed directly
A different simulator	`gymnasium.Env`	named by dotted path in a config

Registries resolve on first use, so naming something costs no import and a config that selects a GPU backend still validates on a machine with no ML stack.

Register before the config is read

A backend name is checked against the registry when a configuration loads, so a typo fails immediately rather than an hour into a run. Import your plugin module before reading a config that names it — and from the command line pass --plugin my_package.module (or a path to a .py file, repeatable) to run, eval or aggregate, which imports it before anything is read.

A reward component¶

skyfall-crl's reward is a composition of independent reward components. Beyond the built-ins (failure, ledger, throughput, verification progress, chaos tickets, financial profit, step efficiency), you can add your own and select it by name in any reward spec — without changing skyfall-crl.

Three steps¶

Subclass RewardComponent. Declare its identity and shape, and implement compute:

from skyfall_crl.rewards import Cadence, RewardComponent, RewardSignal, StepContext

class OnTimeDeliveryComponent(RewardComponent):
    name = "on_time_delivery"       # the key used in a spec
    cadence = Cadence.STEP          # STEP (every step) or EPISODE_END (final step only)
    clip = (0.0, 1.0)               # the value's range
    default_weight = 1.0            # overridable per spec

    def compute(self, ctx: StepContext) -> RewardSignal:
        otif = float((ctx.curr_state or {}).get("otif_rate", 0.0))
        return self._signal(otif)

Register it by name:

from skyfall_crl.rewards import register
register(OnTimeDeliveryComponent.name, OnTimeDeliveryComponent)

The common pattern is to call register(...) at the bottom of your plugin module, so importing the module makes the component available.

Use it in a spec — by the registered name, mixed with any built-ins:

name: logistics
components:
  - type: failure
    weight: 0.5
  - type: on_time_delivery   # your component
    weight: 2.0

The component contract¶

compute(ctx: StepContext) -> RewardSignal — read whatever StepContext fields you need and return self._signal(value). StepContext carries the step's op-log slice.

The substrate-neutral fields are step, action / action_valid / action_result, episode, prev_state / curr_state, observed_at, is_last_step, regime_id, regime_origin (present for logging only — never condition reward on it), and extras.

failure_types, tickets_before / tickets_after, verifications, ledger and throughput describe a MORPHEUS world specifically, and are filled in by its environment.

Writing a component for a different environment? Put whatever it needs in extras and read it back from there. Nothing about the reward system assumes incident tickets, so a component that reads only extras composes alongside the built-in ones exactly the same way.

Declare what you read, too — reads = ("extras",) on the class. It costs one line and it is what lets the engine tell a user why a reward is zero rather than leaving them to guess:

engine = CompositionEngine(spec)
engine.declared_reads()      # what this spec's components consume
engine.unmet_reads(ctx)      # what this context does not provide

A spec whose components can read nothing from the environment they are scoring warns once. That is the failure mode worth designing against: a reward that is structurally zero looks exactly like an agent that has not learned anything yet.

The shipped specs are not a starting point for you. paper, experiment, eval8 and verification_progress all read MORPHEUS's operational fields, which is why RewardConfig has no default — an environment that reports something else has to name its own. - cadence — STEP fires every step; EPISODE_END fires once, on the final step (for episode-level snapshots such as ledger totals). EPISODE_END components must be stateless. - clip — the (low, high) range of your value; the engine clips to it before weighting. - default_weight — used when a spec doesn't override weight. - reset() — override it if your component keeps per-episode state; it is called at each episode start. The default is a no-op.

Registry API¶

register(name, component_cls, *, overwrite=False) — add a component.
get_component(name) — the class registered under a name.
available() — the sorted list of registered component names.

A complete runnable example is in examples/custom_reward_plugin.py.

Check it. skyfall-crl conformance --reward mypkg.rewards:make runs the contract above against your implementation and reports what it found — see Conformance.

An algorithm backend¶

Algorithm is a Protocol, so you inherit nothing — an object with these four members satisfies it:

from skyfall_crl.train import RolloutWindow, register

class MyBackend:
    def __init__(self, learning_rate: float = 1e-4) -> None:
        self._policy = MyPolicy()

    @property
    def policy(self):
        return self._policy                          # anything with .act(observation)

    def update(self, window: RolloutWindow) -> dict[str, float]:
        return {"reward/window": window.total_reward}

    def on_regime_shift(self, regime: dict) -> None:
        ...                                          # consolidate, snapshot, or ignore

    def save_checkpoint(self, output_dir: str) -> None:
        ...

register("my_backend", MyBackend)

Select it, and pass constructor arguments through params:

algorithm:
  name: my_backend
  params: {learning_rate: 3.0e-5}

A window holds plain Python and NumPy — observations and actions exactly as your environment's spaces produce them, never tokens — so the same window describes a CartPole step and a tool call. Converting to tensors is the backend's business.

A complete runnable example is in examples/hill_climbing_backend.py.

Check it. skyfall-crl conformance --algorithm mypkg.algos:make runs the contract above against your implementation and reports what it found — see Conformance.

A configuration provider¶

One method. Consumers derive boundaries by watching the returned id change, which is why a schedule and a label-free change-point detector satisfy the same interface — a schedule can answer is this a boundary in advance and a detector cannot, so it is not on the protocol.

from skyfall_crl.regime import register_provider

class DriftDetector:
    def regime_id(self, step: int) -> str:
        return "post_shift" if self._surprise_exceeded(step) else "baseline"

register_provider("drift", DriftDetector)

regime:
  provider: drift
  params: {threshold: 2.5}

Check it. skyfall-crl conformance --regime mypkg.regimes:make runs the contract above against your implementation and reports what it found — see Conformance.

An upper-bound derivation¶

Three of the six metrics are measured against the reward a perfect policy could reach, and it is normally derived from the reward specification itself — every term at its maximum, split by which terms can fire on an ordinary step and which only at the end of an episode.

That works whenever a component declares a limit. When one does not, there is no closed form, and the derivation raises naming the component rather than substituting a plausible number. Two ways out. Estimate it from contexts you believe represent perfect play:

from skyfall_crl.eval import numeric_upper_bound

bound = numeric_upper_bound(spec, perfect_play_contexts)

Or, when your reward's ceiling depends on something the specification does not state — how many incidents a configuration generates, say — register a closed form of your own:

from skyfall_crl.eval import UpperBound, Variant, register_derivation

def arrivals_bound(spec, *, arrival_rate: float, **_):
    """A ceiling that scales with how much work a configuration produces."""
    per_step = arrival_rate * 2.0
    return UpperBound(per_step, per_step, "arrivals", Variant.ANALYTIC)

register_derivation("arrivals", arrivals_bound)

Nothing environment-specific ships in this registry. The extension point does, because a ceiling that depends on the world is knowledge the library does not have.

A policy that can be exported and served¶

A policy needs nothing special to be trained. To come back out of an export bundle it needs one of two things, and the bundle records which it found:

class MyPolicy:
    def act(self, observation): ...

    def load_weights(self, path: str) -> None:
        """Fill this policy from a checkpoint directory written by the backend."""
        self.weights = np.load(Path(path) / "weights.npy")

That is the weights protocol: the policy is built from its recorded id and kwargs, then filled. Validate the shape here rather than trusting the file — two policies can write the same filename with incompatible contents, and the one that knows which is which is the policy.

The alternative is the checkpoint protocol: a from_checkpoint(path, **kwargs) classmethod, for a policy that is rebuilt from its checkpoint rather than filled — a language model, where building a base model only to discard it is not a detail. The recorded factory is then called with the checkpoint directory as its first argument.

A policy with neither still exports; its bundle records the schemas and the weights and says it cannot reconstitute the policy. For a bundle to name your policy at all, the experiment's algorithm.policy section must name it — so the factory has to take plain values a configuration document can express.

A different deployment¶

MorpheusBackend is the whole surface the environment needs from a world: lifecycle, acting, observing, the incident stream, verification, and chaos configuration. Implement it and the environment does not care what is on the other side — which is also how the test suite runs the whole environment layer with no deployment at all.

from skyfall_crl.env.morpheus import MorpheusEnv

env = MorpheusEnv(backend=MyBackend(), layout="my-layout")

For an LLM policy that should see a curated action schema rather than raw tools, implement TaskView: it defines the action space, validates what the policy emits, and maps it down to a tool call — returning None for an action with no direct mapping, which is a first-class case rather than an error.

Check it. skyfall-crl conformance --env mypkg.envs:make --tier 2 runs the environment contract, and naming an algorithm alongside it drives a short real run — which is what catches an environment that satisfies every method and still cannot be trained on. See Conformance.

A different simulator¶

Everything above extends this package within MORPHEUS. Adding an environment that is not a MORPHEUS world at all needs no registry entry: name a factory by dotted path and the harness builds it.

env:
  id: my_package.envs:make_env
  kwargs: {difficulty: hard}

There are two tiers, and the difference is what the environment says about itself.

Tier 1 is the whole contract: be a gymnasium.Env. Return your own scalar reward and put nothing in info. In exchange you get configuration scheduling, rollout collection, tracing and the six-metric benchmark. Give the run a regime: section — otherwise every step carries the same configuration label, and adaptation speed, recovery time and forgetting all have nothing to measure. Set rollout.auto_reset deliberately: true for an environment with a real terminal state, false for a persistent one where a reset destroys the world.

Tier 2 adds three things, and in exchange the composable reward system scores the run instead of you:

describe() returning any mapping — recorded verbatim as run provenance, interpreted by nothing.
info["regime"] — {"id": ..., "boundary": ..., "origin": ...}. origin names the configuration a revisit repeats, which is what pairs two visits for the forgetting metric.
info["trace"] — an EpisodeStep, the class rather than a dict. Build it with step_from_result(ctx, result) from a StepContext you filled and a RewardResult the composition engine returned.

Put your own per-step payload in StepContext.extras and have your components declare reads = ("extras",). Six of that class's eighteen fields describe MORPHEUS's operational model, so a simulator that reports none of them uses extras instead — and for the same reason, do not reuse a shipped reward spec, which would read those six fields and score a structural zero forever.

A worked example, with what each tier cost to write, is in Adoption cost.