Skip to content

State control

steerability.algorithms.state_control.base

State control base classes.

This module provides the abstract base classes for methods that steer through hooks into the model's forward pass (modifying intermediate representations during inference); state controls do not change model weights.

Three classes are provided:

  • StateControl: Abstract root the pipeline type-checks.
  • InterventionControl: A state control that is a tuple of declarative interventions.
  • HookControl: A state control that writes its own torch hooks.

State controls implement steering through runtime intervention in the model's forward pass, modifying internal states (activations, attention patterns) to produce generations following y ~ p_θᵃ(x), where "p_θᵃ" is the model with state controls.

Examples of state controls:

  • Activation steering (e.g., adding direction vectors)
  • Attention head manipulation and pruning
  • Layer-wise activation editing
  • Dynamic routing between components
  • Representation engineering techniques

Hooks travel only as HookEntry contributions on session items; the session that executes forwards owns registration. Controls never register hooks and hold no model reference.

See Also:

  • steerability.algorithms.state_control: Implementations of state control methods
  • steerability.core.steering_pipeline: Integration with steering pipeline

BackwardHook = Callable[[nn.Module, tuple, tuple], tuple] module-attribute

ForwardHook = Callable[[nn.Module, tuple, torch.Tensor], torch.Tensor] module-attribute

HookSpec = dict[str, str | PreHook | ForwardHook | BackwardHook] module-attribute

PreHook = Callable[[nn.Module, tuple], tuple | torch.Tensor] module-attribute

HookControl

Bases: StateControl

A state control that writes its own torch hooks.

Keeps the abstract get_hooks(input_ids, runtime_kwargs, **kwargs) and must fully re-derive its per-generation state inside get_hooks on every call. Controls whose behavior is a tuple of residual-stream interventions subclass InterventionControl instead; this class is for methods hooking other mechanisms (e.g. attention weights).

Keeps the conservative in-process generate requirement unless the subclass overrides requirements().

Source code in steerability/algorithms/state_control/base.py 
162
163
164
165
166
167
168
169
170
171
172
class HookControl(StateControl):
    """A state control that writes its own torch hooks.

    Keeps the abstract `get_hooks(input_ids, runtime_kwargs, **kwargs)` and must fully
    re-derive its per-generation state inside `get_hooks` on every call. Controls whose
    behavior is a tuple of residual-stream interventions subclass `InterventionControl`
    instead; this class is for methods hooking other mechanisms (e.g. attention weights).

    Keeps the conservative in-process generate requirement unless the subclass overrides
    `requirements()`.
    """

RUNTIME_KWARGS_SCHEMA = [] class-attribute instance-attribute

args = self.Args.validate(*args, **kwargs) instance-attribute

enabled = True class-attribute instance-attribute

supports_batching = False class-attribute instance-attribute

cleanup()

Release resources allocated during steer().

Override in subclasses that allocate GPU memory or other resources during steering. Default no-op.

Source code in steerability/algorithms/core/base_control.py 
195
196
197
198
199
200
201
def cleanup(self) -> None:
    """Release resources allocated during `steer()`.

    Override in subclasses that allocate GPU memory or other resources during steering. Default
    no-op.
    """
    pass

clone_for_call(seed=None)

A configuration-preserving shallow clone for one generation call.

The clone shares steer-time artifacts (memories, steering vectors, attached tokenizers) with the original but has its own attribute namespace, so per-call attribute mutation on the clone never races another call using the original. When seed is given and the control defines reseed(seed), the clone's client-side RNG is re-seeded.

Parameters:

Name Type Description Default
seed int | None

Optional seed forwarded to the clone's reseed().

 None

Returns:

Type Description
BaseControl

The clone.
Source code in steerability/algorithms/core/base_control.py 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
def clone_for_call(self, seed: int | None = None) -> "BaseControl":
    """A configuration-preserving shallow clone for one generation call.

    The clone shares steer-time artifacts (memories, steering vectors, attached tokenizers)
    with the original but has its own attribute namespace, so per-call attribute mutation on
    the clone never races another call using the original. When `seed` is given and the
    control defines `reseed(seed)`, the clone's client-side RNG is re-seeded.

    Args:
        seed: Optional seed forwarded to the clone's `reseed()`.

    Returns:
        The clone.
    """
    clone = copy.copy(self)
    if seed is not None:
        reseed = getattr(clone, "reseed", None)
        if callable(reseed):
            reseed(seed)
    return clone

export_intervention_spec(runtime_kwargs=None)

The control's InterventionSpec for intervention-capable backends, or None.

The spec is the second serialization of the tuple the control's hooks close over, emitted from the same transform, gate, and scope objects. Must be called after steer(). Returns None when the configuration has no wire form (the configuration is then hook-only) or when the control does not implement spec export at all.

Parameters:

Name Type Description Default
runtime_kwargs dict | None

Per-call parameters, mirroring get_hooks; per-item values (strengths, positions) serialize into the returned spec.

 None

Returns:

Type Description
InterventionSpec | None

The validated InterventionSpec with tensor payloads attached, or None.
Source code in steerability/algorithms/state_control/base.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def export_intervention_spec(self, runtime_kwargs: dict | None = None) -> "InterventionSpec | None":
    """The control's `InterventionSpec` for intervention-capable backends, or None.

    The spec is the second serialization of the tuple the control's hooks close over,
    emitted from the same transform, gate, and scope objects. Must be called after
    `steer()`. Returns None when the configuration has no wire form (the configuration is
    then hook-only) or when the control does not implement spec export at all.

    Args:
        runtime_kwargs: Per-call parameters, mirroring `get_hooks`; per-item values
            (strengths, positions) serialize into the returned spec.

    Returns:
        The validated `InterventionSpec` with tensor payloads attached, or None.
    """
    return None

export_state()

Steer-time state to persist when freezing, keyed by logical name.

Values are typed artifacts (SteeringVector, Probe, ProbeSet, a Memory implementation, CheckpointArtifact, LoRAArtifact, torch.Tensor, or a Path to an on-disk product). Must be called after steer(). The default returns an empty mapping (nothing to persist).

Returns:

Type Description
dict[str, Any]

Mapping from logical name to artifact value.
Source code in steerability/algorithms/core/base_control.py 
117
118
119
120
121
122
123
124
125
126
127
128
def export_state(self) -> dict[str, Any]:
    """Steer-time state to persist when freezing, keyed by logical name.

    Values are typed artifacts (`SteeringVector`, `Probe`, `ProbeSet`, a `Memory`
    implementation, `CheckpointArtifact`, `LoRAArtifact`, `torch.Tensor`, or a `Path` to an
    on-disk product). Must be called after `steer()`. The default returns an empty mapping
    (nothing to persist).

    Returns:
        Mapping from logical name to artifact value.
    """
    return {}

fit_identity()

The object whose canonical form defines this control's fit digest, or None.

The digest detects recipe edits that invalidate frozen artifacts, so the returned object should cover exactly the inputs a re-fit would consume (training data, fit specs) and exclude inert application parameters (strengths, multipliers). The default returns None (no fits).

Returns:

Type Description
Any | None

The fit-identity object, or None.
Source code in steerability/algorithms/core/base_control.py 
161
162
163
164
165
166
167
168
169
170
171
172
def fit_identity(self) -> Any | None:
    """The object whose canonical form defines this control's fit digest, or None.

    The digest detects recipe edits that invalidate frozen artifacts, so the returned
    object should cover exactly the inputs a re-fit would consume (training data, fit
    specs) and exclude inert application parameters (strengths, multipliers). The default
    returns None (no fits).

    Returns:
        The fit-identity object, or None.
    """
    return None

frozen_form(state)

The (registry method key, constructor kwargs) of this control's frozen form.

state is the output of export_state(); artifact values inside the returned kwargs are replaced by store references at encode time. When steer_fits() is empty and state is empty, the recipe is the frozen form and the control's own key and recipe args are returned unchanged. A control whose freeze expands into several entries may return a list of (method key, kwargs) pairs instead.

Parameters:

Name Type Description Default
state dict[str, Any]

The output of export_state().

 required

Returns:

Type Description
tuple[str, dict[str, Any]]

The frozen method key and its constructor kwargs.

Raises:

Type Description
NotFreezableError

If the control produces fits or state but declares no frozen form.
Source code in steerability/algorithms/core/base_control.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
def frozen_form(self, state: dict[str, Any]) -> tuple[str, dict[str, Any]]:
    """The `(registry method key, constructor kwargs)` of this control's frozen form.

    `state` is the output of `export_state()`; artifact values inside the returned kwargs
    are replaced by store references at encode time. When `steer_fits()` is empty and
    `state` is empty, the recipe is the frozen form and the control's own key and recipe
    args are returned unchanged. A control whose freeze expands into several entries may
    return a list of `(method key, kwargs)` pairs instead.

    Args:
        state: The output of `export_state()`.

    Returns:
        The frozen method key and its constructor kwargs.

    Raises:
        NotFreezableError: If the control produces fits or state but declares no frozen
            form.
    """
    if not state and not self.steer_fits():
        from steerability.algorithms.core.registry import method_key_for

        args = getattr(self, "args", None)
        kwargs = {field.name: getattr(args, field.name) for field in fields(args) if field.init} \
            if args is not None else {}
        return method_key_for(type(self)), kwargs
    raise NotFreezableError(
        f"{type(self).__name__} produces steer-time state but declares no frozen form; "
        "override frozen_form() (and export_state()) to support freezing."
    )

get_hooks(input_ids, runtime_kwargs, **kwargs) abstractmethod

Create hook specifications for the current generation.

The pipeline forwards attention_mask (the prompt attention mask matching input_ids, or None) through **kwargs; controls that score prompt tokens may consume it to align with the real (non-pad) positions instead of re-deriving a mask by token identity.

Parameters:

Name Type Description Default
input_ids Tensor

Prompt token ids of shape [batch, seq_len].

 required
runtime_kwargs dict | None

Per-call parameters for the control.

 required
Source code in steerability/algorithms/state_control/base.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
@abstractmethod
def get_hooks(
    self,
    input_ids: torch.Tensor,
    runtime_kwargs: dict | None,
    **kwargs,
) -> dict[str, list[HookSpec]]:
    """Create hook specifications for the current generation.

    The pipeline forwards `attention_mask` (the prompt attention mask matching `input_ids`,
    or None) through `**kwargs`; controls that score prompt tokens may consume it to align
    with the real (non-pad) positions instead of re-deriving a mask by token identity.

    Args:
        input_ids: Prompt token ids of shape [batch, seq_len].
        runtime_kwargs: Per-call parameters for the control.
    """
    pass

requirements()

Backend requirements computed from this instance's configuration, per phase.

The default requires Capability.IN_PROCESS_TORCH at generate and nothing at score, which only the Hugging Face backend satisfies. A control with portable mechanisms overrides this to state weaker or alternative requirements. Configuration determines the result, so two configurations of one class may differ. Only enabled controls are consulted during support evaluation.

Returns:

Type Description
Requirements

The control's phase-keyed requirements.
Source code in steerability/algorithms/core/base_control.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
def requirements(self) -> Requirements:
    """Backend requirements computed from this instance's configuration, per phase.

    The default requires `Capability.IN_PROCESS_TORCH` at generate and nothing at score,
    which only the Hugging Face backend satisfies. A control with portable mechanisms
    overrides this to state weaker or alternative requirements. Configuration determines the
    result, so two configurations of one class may differ. Only enabled controls are
    consulted during support evaluation.

    Returns:
        The control's phase-keyed requirements.
    """
    return Requirements(generate=needs(Capability.IN_PROCESS_TORCH))

steer(model, tokenizer=None, session=None, **kwargs)

Optional steering/preparation.

session is a SteeringSession on the steering backend, provided by the pipeline.

Source code in steerability/algorithms/state_control/base.py 
127
128
129
130
131
132
133
134
135
136
def steer(self,
          model: PreTrainedModel,
          tokenizer: PreTrainedTokenizerBase = None,
          session=None,
          **kwargs) -> None:
    """Optional steering/preparation.

    `session` is a `SteeringSession` on the steering backend, provided by the pipeline.
    """
    pass

steer_access()

The model access this instance's steer step requires, on the ModelAccess ladder.

The default is ModelAccess.FACTS (layout and tokenizer only). A control whose steer step generates or scores through the session declares ROLLOUTS, one that captures hidden states declares CAPTURE, and one that touches the model as a live torch.nn.Module declares MODULE. Configuration determines the result. The pipeline hands steer() a session scoped to the declared rung, and the live model only at MODULE. A control may retain the pipeline model beyond steer() only if its generate phase requires Capability.IN_PROCESS_TORCH.

Returns:

Type Description
ModelAccess

The declared access rung.
Source code in steerability/algorithms/core/base_control.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def steer_access(self) -> ModelAccess:
    """The model access this instance's steer step requires, on the `ModelAccess` ladder.

    The default is `ModelAccess.FACTS` (layout and tokenizer only). A control whose steer
    step generates or scores through the session declares `ROLLOUTS`, one that captures
    hidden states declares `CAPTURE`, and one that touches the model as a live
    `torch.nn.Module` declares `MODULE`. Configuration determines the result. The pipeline
    hands `steer()` a session scoped to the declared rung, and the live model only at
    `MODULE`. A control may retain the pipeline model beyond `steer()` only if its
    generate phase requires `Capability.IN_PROCESS_TORCH`.

    Returns:
        The declared access rung.
    """
    return ModelAccess.FACTS

steer_fits()

The fit artifacts this instance's steer step will produce, for the steer plan.

Each entry is (artifact, artifact_class), where artifact is the fit source or recipe class name and artifact_class is "direction" or "calibrated". The default is an empty tuple (no fits).

Returns:

Type Description
tuple[tuple[str, str], ...]

The declared fit artifacts, in declaration order.
Source code in steerability/algorithms/core/base_control.py 
105
106
107
108
109
110
111
112
113
114
115
def steer_fits(self) -> tuple[tuple[str, str], ...]:
    """The fit artifacts this instance's steer step will produce, for the steer plan.

    Each entry is `(artifact, artifact_class)`, where `artifact` is the fit source or
    recipe class name and `artifact_class` is `"direction"` or `"calibrated"`. The default
    is an empty tuple (no fits).

    Returns:
        The declared fit artifacts, in declaration order.
    """
    return ()

InterventionControl

Bases: StateControl

A state control that is a tuple of interventions.

Subclasses declare an unbound intervention template, usually in _configure(); the base steer() binds it. There is no per-generation protocol on the control: hook construction, gate sizing, and position state are owned by build_hooks, and lowering to InterventionSpec is owned by lower_interventions.

Class attributes

hook_only_hint: Fix text used in unsupported-generate verdicts when the template has no wire form.

Attributes:

Name Type Description
interventions tuple

The bound interventions, populated by steer().
Source code in steerability/algorithms/state_control/base.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
class InterventionControl(StateControl):
    """A state control that is a tuple of interventions.

    Subclasses declare an unbound intervention template, usually in `_configure()`; the base
    `steer()` binds it. There is no per-generation protocol on the control: hook construction,
    gate sizing, and position state are owned by `build_hooks`, and lowering to
    `InterventionSpec` is owned by `lower_interventions`.

    Class attributes:
        hook_only_hint: Fix text used in unsupported-generate verdicts when the template has
            no wire form.

    Attributes:
        interventions: The bound interventions, populated by `steer()`.
    """

    supports_batching = True
    hook_only_hint: str | None = None

    tokenizer = None
    interventions: tuple = ()
    _template: tuple = ()

    def steer(
        self,
        model: PreTrainedModel | None = None,
        tokenizer: PreTrainedTokenizerBase | None = None,
        session: "SteeringSession | None" = None,
        **kwargs,
    ) -> PreTrainedModel | None:
        """Bind the intervention template against the model (or the session's layout).

        Structural facts come from the steering session's layout when a session is given, so a
        fully concrete template (precomputed vectors, manual thresholds) binds with
        `model=None`. Templates carrying sources (fits, searches) resolve them here.

        Args:
            model: The base language model, or None for concrete templates bound against a
                session layout.
            tokenizer: Tokenizer used when fitting sources.
            session: `SteeringSession` on the steering backend, provided by the pipeline.

        Returns:
            The input model, unchanged.
        """
        from steerability.algorithms.core.internals.model_layout import resolve_model_layout
        from steerability.algorithms.state_control.common.layout_facts import resolve_layout

        layout = resolve_layout(model, session)
        self._num_layers = layout.num_layers
        self._module_layout = resolve_model_layout(model) if model is not None else None
        if tokenizer is not None:
            self.tokenizer = tokenizer
        self.interventions = tuple(
            intervention.bind(model, tokenizer, layout=layout, session=session)
            for intervention in self._template
        )
        return model

    @property
    def _transform(self):
        """The first intervention's transform (None before `steer()`).

        Assignment replaces the transform on the first intervention, so a wrapped or
        instrumented transform takes effect in subsequently built hooks.
        """
        return self.interventions[0].transform if self.interventions else None

    @_transform.setter
    def _transform(self, value) -> None:
        import dataclasses

        if not self.interventions:
            raise AttributeError("No bound interventions; call steer() first.")
        first, *rest = self.interventions
        self.interventions = (dataclasses.replace(first, transform=value), *rest)

    @property
    def _gate(self):
        """The first intervention's gate (None before `steer()` or when unconditional)."""
        return self.interventions[0].gate if self.interventions else None

    def _resolve_module_layout(self, model=None):
        """The module-path layout, resolved from the module tree on first use."""
        layout = getattr(self, "_module_layout", None)
        if layout is None:
            from steerability.algorithms.core.internals.model_layout import resolve_model_layout

            if model is None:
                raise RuntimeError(
                    f"{type(self).__name__} was steered without a live model, so hook module "
                    "names are unresolved; provide the model (the pipeline does) or steer with "
                    "a model."
                )
            layout = resolve_model_layout(model)
            self._module_layout = layout
        return layout

    def get_hooks(
        self,
        input_ids: torch.Tensor | dict,
        runtime_kwargs: dict | None = None,
        attention_mask: torch.Tensor | None = None,
        **kwargs,
    ) -> dict[str, list[HookSpec]]:
        """Compile the bound interventions to hooks for the current generation.

        Delegates to `build_hooks`: a fresh hook runtime is created, gates reset to the
        logical batch, and one behavior hook is emitted per (intervention, layer).

        Generation-time context arrives through `**kwargs`; `model` is consulted to resolve
        hook module names when steering ran without a live model.

        Args:
            input_ids: Prompt token ids of shape `[B, T]` or `[T]`.
            runtime_kwargs: Unused.
            attention_mask: The prompt attention mask matching `input_ids`, forwarded to
                gate evidence pooling on the prefill pass. When None and the tokenizer defines a
                pad token, a mask is inferred from leading and trailing pad runs.

        Returns:
            Hook specifications with `"pre"`, `"forward"`, `"backward"` keys.
        """
        from steerability.algorithms.state_control.common.runtime import build_hooks
        from steerability.algorithms.state_control.common.token_scope import compute_prompt_lens
        from steerability.utils.tokenization import infer_attention_mask_from_ids

        ids = input_ids if isinstance(input_ids, torch.Tensor) else input_ids["input_ids"]
        if ids.ndim == 1:
            ids = ids.unsqueeze(0)

        layout = self._resolve_module_layout(kwargs.get("model"))
        pad_token_id = getattr(self.tokenizer, "pad_token_id", None) if self.tokenizer is not None else None
        prompt_lens = compute_prompt_lens(ids, pad_token_id)

        if attention_mask is not None:
            mask = attention_mask if isinstance(attention_mask, torch.Tensor) else torch.as_tensor(attention_mask)
            prompt_mask = (mask.unsqueeze(0) if mask.ndim == 1 else mask).to(torch.bool)
        elif pad_token_id is not None:
            prompt_mask = infer_attention_mask_from_ids(ids, pad_token_id).to(torch.bool)
        else:
            prompt_mask = None

        return build_hooks(self.interventions, layout, prompt_lens, prompt_mask, model=kwargs.get("model"))

    def export_intervention_spec(self, runtime_kwargs: dict | None = None):
        """The control's `InterventionSpec`, lowered from the bound interventions, or None.

        Must be called after `steer()`. Returns None when the configuration has no wire form.
        """
        from steerability.algorithms.state_control.common.lowering import lower_interventions

        if not self.interventions or getattr(self, "_num_layers", None) is None:
            return None
        kinds = self.wire_kinds()
        if kinds is None:
            return None
        return lower_interventions(self.interventions, num_layers=self._num_layers)

    def wire_kinds(self):
        """The combined wire kinds of the bound interventions (or the template before
        `steer()`), or None when any intervention is hook-only."""
        from steerability.algorithms.state_control.common.specs import combine_kinds

        source = self.interventions or self._template
        return combine_kinds(intervention.wire_kinds() for intervention in source)

    def _unbound_sources(self):
        """Yield each unbound template element's source (or undeclared factory slot).

        Yields the transform sources of unbound transform elements, factory transform slots
        themselves (which declare their own `access` or default to the live model), and
        unresolved gate sources, in template order.
        """
        from steerability.algorithms.state_control.common.transforms.base import BaseTransform, unwrap_modifiers

        for intervention in self._template:
            transform = intervention.transform
            if isinstance(transform, BaseTransform):
                core, wrappers = unwrap_modifiers(transform)
                for element in (core, *wrappers):
                    if not element.is_bound and element.source is not None:
                        yield element.source
            else:
                yield transform
            gate = intervention.gate
            if gate is not None and not _is_concrete_gate(gate):
                yield gate

    def steer_access(self) -> ModelAccess:
        """The model access the template's steer step requires, folded over its sources.

        A fully bound template requires `ModelAccess.FACTS`, since pure layer selectors
        resolve from structural facts available on any session. Otherwise the strongest
        declared source access wins; a source or factory slot without an `access` declaration
        builds against the live model, so it counts as `ModelAccess.MODULE`.
        """
        access = ModelAccess.FACTS
        for source in self._unbound_sources():
            access = max(access, getattr(source, "access", ModelAccess.MODULE))
        return access

    def _fit_sources(self):
        """Yield the template's fit sources: unbound sources carrying an `artifact_class`
        whose resolution does model-side work (`access` above `ModelAccess.FACTS`)."""
        for source in self._unbound_sources():
            if getattr(source, "artifact_class", None) is None:
                continue
            if getattr(source, "access", ModelAccess.MODULE) == ModelAccess.FACTS:
                continue
            yield source

    def steer_fits(self) -> tuple[tuple[str, str], ...]:
        """The template's fit artifacts, i.e. every unbound source carrying an
        `artifact_class`, as `(artifact, artifact_class)` pairs in template order."""
        return tuple(
            (type(source).__name__, source.artifact_class) for source in self._fit_sources()
        )

    def fit_identity(self) -> Any | None:
        """The template's fit sources in template order, each in its encodable form, or None
        when the template declares no fits."""
        sources = list(self._fit_sources())
        if not sources:
            return None
        return tuple(_fit_ingredients(source) for source in sources)

    def export_state(self) -> dict[str, Any]:
        """Bound steering artifacts, keyed by intervention position.

        For intervention `i`, the core transform's artifact exports as
        `"intervention_{i}/transform"` (a `SteeringVector` view), each wrapper's own artifact
        as `"intervention_{i}/modifier_{j}"`, and the resolved gate as
        `"intervention_{i}/gate"` when the template's gate slot held a source. Must be called
        after `steer()`.

        Returns:
            Mapping from logical name to artifact value.
        """
        from steerability.algorithms.state_control.common.gating import Gate
        from steerability.algorithms.state_control.common.transforms.base import BaseTransform, unwrap_modifiers

        state: dict[str, Any] = {}
        for i, intervention in enumerate(self.interventions):
            if isinstance(intervention.transform, BaseTransform):
                core, wrappers = unwrap_modifiers(intervention.transform)
                view = _core_artifact_view(core)
                if view is not None:
                    state[f"intervention_{i}/transform"] = view
                for j, wrapper in enumerate(wrappers):
                    own = getattr(wrapper, "steering_vector", None)
                    if own is not None:
                        state[f"intervention_{i}/modifier_{j}"] = own
            template_gate = self._template[i].gate if i < len(self._template) else None
            if isinstance(intervention.gate, Gate) and template_gate is not None \
                    and not _is_concrete_gate(template_gate):
                state[f"intervention_{i}/gate"] = intervention.gate
        return state

    def frozen_form(self, state: dict[str, Any]) -> list[tuple[str, dict[str, Any]]]:
        """One `activation_adapter` entry per bound intervention, in intervention order.

        Each entry's args mirror the intervention: the bound transform, the resolved behavior
        layers, the hook point, the resolved gate, and the token scope. Must be called after
        `steer()`.

        Returns:
            List of `("state_control/activation_adapter", kwargs)` pairs.

        Raises:
            NotFreezableError: If the control is unsteered, an intervention hooks the
                `norm_input` site (no `activation_adapter` form), or an intervention follows
                an externally driven shared gate (an in-memory relationship that does not
                serialize).
        """
        if not self.interventions:
            raise NotFreezableError(
                f"{type(self).__name__} has no bound interventions; call steer() before freezing."
            )
        entries: list[tuple[str, dict[str, Any]]] = []
        for intervention in self.interventions:
            if intervention.resolved_site() == "norm_input":
                raise NotFreezableError(
                    f"{type(self).__name__} hooks the norm_input site, which has no "
                    "activation_adapter form; configure the layer_output intervention point "
                    "to freeze this control."
                )
            if intervention.gate_driven_externally:
                raise NotFreezableError(
                    f"{type(self).__name__} follows an externally driven shared gate, an "
                    "in-memory relationship that does not serialize; freeze the driving "
                    "control's pipeline without the follower, or gate this control directly."
                )
            entries.append(("state_control/activation_adapter", {
                "transform": intervention.transform,
                "layer_ids": [int(lid) for lid in intervention.layers],
                "hook_point": intervention.boundary,
                "gate": intervention.gate,
                "token_scope": intervention.scope.kind,
                "last_k": intervention.scope.last_k,
                "from_position": intervention.scope.from_position,
            }))
        return entries

    def requirements(self) -> Requirements:
        """Backend requirements derived from the declared interventions, per phase.

        Generate offers the intervention-spec alternative whenever every component of every
        intervention has a wire form; hook-only configurations require the in-process backend.
        Score is in-process: remote prompt-logprob scoring anchors token scopes at the
        request's prompt end (the end of the prompt-plus-reference concatenation), which would
        silently unanchor prompt-relative interventions.
        """
        from steerability.algorithms.core.execution.contracts import Capability, Requirements, any_of, needs

        kinds = self.wire_kinds()
        in_process = needs(Capability.IN_PROCESS_TORCH)
        score = needs(
            Capability.IN_PROCESS_TORCH,
            hint=(
                "remote prompt-logprob scoring anchors token scopes at the request's prompt "
                "end, so scoped interventions would not cover the reference; score on the "
                "huggingface backend"
            ),
        )
        if kinds is None:
            return Requirements(
                generate=needs(Capability.IN_PROCESS_TORCH, hint=self.hook_only_hint),
                score=score,
            )
        return Requirements(
            generate=any_of(
                in_process,
                needs(
                    Capability.INTERVENTION_SPECS,
                    kinds=kinds,
                    hint="serve this intervention through the vLLM-Hook plugin",
                ),
            ),
            score=score,
        )

    def clone_for_call(self, seed: int | None = None):
        """A per-call clone whose interventions carry independent gate state.

        Gates are deep-copied with one shared memo across the control's interventions, so a
        gate instance shared by several interventions stays shared inside the clone while
        being isolated from the original and from sibling clones. Transforms, scorers, and
        steer-time artifacts stay shared.
        """
        import dataclasses

        clone = super().clone_for_call(seed)
        if self.interventions:
            memo: dict = {}
            clone.interventions = tuple(
                dataclasses.replace(intervention, gate=copy.deepcopy(intervention.gate, memo))
                for intervention in self.interventions
            )
        return clone

RUNTIME_KWARGS_SCHEMA = [] class-attribute instance-attribute

args = self.Args.validate(*args, **kwargs) instance-attribute

enabled = True class-attribute instance-attribute

hook_only_hint = None class-attribute instance-attribute

interventions = () class-attribute instance-attribute

supports_batching = True class-attribute instance-attribute

tokenizer = None class-attribute instance-attribute

cleanup()

Release resources allocated during steer().

Override in subclasses that allocate GPU memory or other resources during steering. Default no-op.

Source code in steerability/algorithms/core/base_control.py 
195
196
197
198
199
200
201
def cleanup(self) -> None:
    """Release resources allocated during `steer()`.

    Override in subclasses that allocate GPU memory or other resources during steering. Default
    no-op.
    """
    pass

clone_for_call(seed=None)

A per-call clone whose interventions carry independent gate state.

Gates are deep-copied with one shared memo across the control's interventions, so a gate instance shared by several interventions stays shared inside the clone while being isolated from the original and from sibling clones. Transforms, scorers, and steer-time artifacts stay shared.

Source code in steerability/algorithms/state_control/base.py 
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
def clone_for_call(self, seed: int | None = None):
    """A per-call clone whose interventions carry independent gate state.

    Gates are deep-copied with one shared memo across the control's interventions, so a
    gate instance shared by several interventions stays shared inside the clone while
    being isolated from the original and from sibling clones. Transforms, scorers, and
    steer-time artifacts stay shared.
    """
    import dataclasses

    clone = super().clone_for_call(seed)
    if self.interventions:
        memo: dict = {}
        clone.interventions = tuple(
            dataclasses.replace(intervention, gate=copy.deepcopy(intervention.gate, memo))
            for intervention in self.interventions
        )
    return clone

export_intervention_spec(runtime_kwargs=None)

The control's InterventionSpec, lowered from the bound interventions, or None.

Must be called after steer(). Returns None when the configuration has no wire form.

Source code in steerability/algorithms/state_control/base.py 
320
321
322
323
324
325
326
327
328
329
330
331
332
def export_intervention_spec(self, runtime_kwargs: dict | None = None):
    """The control's `InterventionSpec`, lowered from the bound interventions, or None.

    Must be called after `steer()`. Returns None when the configuration has no wire form.
    """
    from steerability.algorithms.state_control.common.lowering import lower_interventions

    if not self.interventions or getattr(self, "_num_layers", None) is None:
        return None
    kinds = self.wire_kinds()
    if kinds is None:
        return None
    return lower_interventions(self.interventions, num_layers=self._num_layers)

export_state()

Bound steering artifacts, keyed by intervention position.

For intervention i, the core transform's artifact exports as "intervention_{i}/transform" (a SteeringVector view), each wrapper's own artifact as "intervention_{i}/modifier_{j}", and the resolved gate as "intervention_{i}/gate" when the template's gate slot held a source. Must be called after steer().

Returns:

Type Description
dict[str, Any]

Mapping from logical name to artifact value.
Source code in steerability/algorithms/state_control/base.py 
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
def export_state(self) -> dict[str, Any]:
    """Bound steering artifacts, keyed by intervention position.

    For intervention `i`, the core transform's artifact exports as
    `"intervention_{i}/transform"` (a `SteeringVector` view), each wrapper's own artifact
    as `"intervention_{i}/modifier_{j}"`, and the resolved gate as
    `"intervention_{i}/gate"` when the template's gate slot held a source. Must be called
    after `steer()`.

    Returns:
        Mapping from logical name to artifact value.
    """
    from steerability.algorithms.state_control.common.gating import Gate
    from steerability.algorithms.state_control.common.transforms.base import BaseTransform, unwrap_modifiers

    state: dict[str, Any] = {}
    for i, intervention in enumerate(self.interventions):
        if isinstance(intervention.transform, BaseTransform):
            core, wrappers = unwrap_modifiers(intervention.transform)
            view = _core_artifact_view(core)
            if view is not None:
                state[f"intervention_{i}/transform"] = view
            for j, wrapper in enumerate(wrappers):
                own = getattr(wrapper, "steering_vector", None)
                if own is not None:
                    state[f"intervention_{i}/modifier_{j}"] = own
        template_gate = self._template[i].gate if i < len(self._template) else None
        if isinstance(intervention.gate, Gate) and template_gate is not None \
                and not _is_concrete_gate(template_gate):
            state[f"intervention_{i}/gate"] = intervention.gate
    return state

fit_identity()

The template's fit sources in template order, each in its encodable form, or None when the template declares no fits.

Source code in steerability/algorithms/state_control/base.py 
394
395
396
397
398
399
400
def fit_identity(self) -> Any | None:
    """The template's fit sources in template order, each in its encodable form, or None
    when the template declares no fits."""
    sources = list(self._fit_sources())
    if not sources:
        return None
    return tuple(_fit_ingredients(source) for source in sources)

frozen_form(state)

One activation_adapter entry per bound intervention, in intervention order.

Each entry's args mirror the intervention: the bound transform, the resolved behavior layers, the hook point, the resolved gate, and the token scope. Must be called after steer().

Returns:

Type Description
list[tuple[str, dict[str, Any]]]

List of ("state_control/activation_adapter", kwargs) pairs.

Raises:

Type Description
NotFreezableError

If the control is unsteered, an intervention hooks the norm_input site (no activation_adapter form), or an intervention follows an externally driven shared gate (an in-memory relationship that does not serialize).
Source code in steerability/algorithms/state_control/base.py 
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
def frozen_form(self, state: dict[str, Any]) -> list[tuple[str, dict[str, Any]]]:
    """One `activation_adapter` entry per bound intervention, in intervention order.

    Each entry's args mirror the intervention: the bound transform, the resolved behavior
    layers, the hook point, the resolved gate, and the token scope. Must be called after
    `steer()`.

    Returns:
        List of `("state_control/activation_adapter", kwargs)` pairs.

    Raises:
        NotFreezableError: If the control is unsteered, an intervention hooks the
            `norm_input` site (no `activation_adapter` form), or an intervention follows
            an externally driven shared gate (an in-memory relationship that does not
            serialize).
    """
    if not self.interventions:
        raise NotFreezableError(
            f"{type(self).__name__} has no bound interventions; call steer() before freezing."
        )
    entries: list[tuple[str, dict[str, Any]]] = []
    for intervention in self.interventions:
        if intervention.resolved_site() == "norm_input":
            raise NotFreezableError(
                f"{type(self).__name__} hooks the norm_input site, which has no "
                "activation_adapter form; configure the layer_output intervention point "
                "to freeze this control."
            )
        if intervention.gate_driven_externally:
            raise NotFreezableError(
                f"{type(self).__name__} follows an externally driven shared gate, an "
                "in-memory relationship that does not serialize; freeze the driving "
                "control's pipeline without the follower, or gate this control directly."
            )
        entries.append(("state_control/activation_adapter", {
            "transform": intervention.transform,
            "layer_ids": [int(lid) for lid in intervention.layers],
            "hook_point": intervention.boundary,
            "gate": intervention.gate,
            "token_scope": intervention.scope.kind,
            "last_k": intervention.scope.last_k,
            "from_position": intervention.scope.from_position,
        }))
    return entries

get_hooks(input_ids, runtime_kwargs=None, attention_mask=None, **kwargs)

Compile the bound interventions to hooks for the current generation.

Delegates to build_hooks: a fresh hook runtime is created, gates reset to the logical batch, and one behavior hook is emitted per (intervention, layer).

Generation-time context arrives through **kwargs; model is consulted to resolve hook module names when steering ran without a live model.

Parameters:

Name Type Description Default
input_ids Tensor | dict

Prompt token ids of shape [B, T] or [T].

 required
runtime_kwargs dict | None

Unused.

 None
attention_mask Tensor | None

The prompt attention mask matching input_ids, forwarded to gate evidence pooling on the prefill pass. When None and the tokenizer defines a pad token, a mask is inferred from leading and trailing pad runs.

 None

Returns:

Type Description
dict[str, list[HookSpec]]

Hook specifications with "pre", "forward", "backward" keys.
Source code in steerability/algorithms/state_control/base.py 
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
def get_hooks(
    self,
    input_ids: torch.Tensor | dict,
    runtime_kwargs: dict | None = None,
    attention_mask: torch.Tensor | None = None,
    **kwargs,
) -> dict[str, list[HookSpec]]:
    """Compile the bound interventions to hooks for the current generation.

    Delegates to `build_hooks`: a fresh hook runtime is created, gates reset to the
    logical batch, and one behavior hook is emitted per (intervention, layer).

    Generation-time context arrives through `**kwargs`; `model` is consulted to resolve
    hook module names when steering ran without a live model.

    Args:
        input_ids: Prompt token ids of shape `[B, T]` or `[T]`.
        runtime_kwargs: Unused.
        attention_mask: The prompt attention mask matching `input_ids`, forwarded to
            gate evidence pooling on the prefill pass. When None and the tokenizer defines a
            pad token, a mask is inferred from leading and trailing pad runs.

    Returns:
        Hook specifications with `"pre"`, `"forward"`, `"backward"` keys.
    """
    from steerability.algorithms.state_control.common.runtime import build_hooks
    from steerability.algorithms.state_control.common.token_scope import compute_prompt_lens
    from steerability.utils.tokenization import infer_attention_mask_from_ids

    ids = input_ids if isinstance(input_ids, torch.Tensor) else input_ids["input_ids"]
    if ids.ndim == 1:
        ids = ids.unsqueeze(0)

    layout = self._resolve_module_layout(kwargs.get("model"))
    pad_token_id = getattr(self.tokenizer, "pad_token_id", None) if self.tokenizer is not None else None
    prompt_lens = compute_prompt_lens(ids, pad_token_id)

    if attention_mask is not None:
        mask = attention_mask if isinstance(attention_mask, torch.Tensor) else torch.as_tensor(attention_mask)
        prompt_mask = (mask.unsqueeze(0) if mask.ndim == 1 else mask).to(torch.bool)
    elif pad_token_id is not None:
        prompt_mask = infer_attention_mask_from_ids(ids, pad_token_id).to(torch.bool)
    else:
        prompt_mask = None

    return build_hooks(self.interventions, layout, prompt_lens, prompt_mask, model=kwargs.get("model"))

requirements()

Backend requirements derived from the declared interventions, per phase.

Generate offers the intervention-spec alternative whenever every component of every intervention has a wire form; hook-only configurations require the in-process backend. Score is in-process: remote prompt-logprob scoring anchors token scopes at the request's prompt end (the end of the prompt-plus-reference concatenation), which would silently unanchor prompt-relative interventions.

Source code in steerability/algorithms/state_control/base.py 
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
def requirements(self) -> Requirements:
    """Backend requirements derived from the declared interventions, per phase.

    Generate offers the intervention-spec alternative whenever every component of every
    intervention has a wire form; hook-only configurations require the in-process backend.
    Score is in-process: remote prompt-logprob scoring anchors token scopes at the
    request's prompt end (the end of the prompt-plus-reference concatenation), which would
    silently unanchor prompt-relative interventions.
    """
    from steerability.algorithms.core.execution.contracts import Capability, Requirements, any_of, needs

    kinds = self.wire_kinds()
    in_process = needs(Capability.IN_PROCESS_TORCH)
    score = needs(
        Capability.IN_PROCESS_TORCH,
        hint=(
            "remote prompt-logprob scoring anchors token scopes at the request's prompt "
            "end, so scoped interventions would not cover the reference; score on the "
            "huggingface backend"
        ),
    )
    if kinds is None:
        return Requirements(
            generate=needs(Capability.IN_PROCESS_TORCH, hint=self.hook_only_hint),
            score=score,
        )
    return Requirements(
        generate=any_of(
            in_process,
            needs(
                Capability.INTERVENTION_SPECS,
                kinds=kinds,
                hint="serve this intervention through the vLLM-Hook plugin",
            ),
        ),
        score=score,
    )

steer(model=None, tokenizer=None, session=None, **kwargs)

Bind the intervention template against the model (or the session's layout).

Structural facts come from the steering session's layout when a session is given, so a fully concrete template (precomputed vectors, manual thresholds) binds with model=None. Templates carrying sources (fits, searches) resolve them here.

Parameters:

Name Type Description Default
model PreTrainedModel | None

The base language model, or None for concrete templates bound against a session layout.

 None
tokenizer PreTrainedTokenizerBase | None

Tokenizer used when fitting sources.

 None
session SteeringSession | None

SteeringSession on the steering backend, provided by the pipeline.

 None

Returns:

Type Description
PreTrainedModel | None

The input model, unchanged.
Source code in steerability/algorithms/state_control/base.py 
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
def steer(
    self,
    model: PreTrainedModel | None = None,
    tokenizer: PreTrainedTokenizerBase | None = None,
    session: "SteeringSession | None" = None,
    **kwargs,
) -> PreTrainedModel | None:
    """Bind the intervention template against the model (or the session's layout).

    Structural facts come from the steering session's layout when a session is given, so a
    fully concrete template (precomputed vectors, manual thresholds) binds with
    `model=None`. Templates carrying sources (fits, searches) resolve them here.

    Args:
        model: The base language model, or None for concrete templates bound against a
            session layout.
        tokenizer: Tokenizer used when fitting sources.
        session: `SteeringSession` on the steering backend, provided by the pipeline.

    Returns:
        The input model, unchanged.
    """
    from steerability.algorithms.core.internals.model_layout import resolve_model_layout
    from steerability.algorithms.state_control.common.layout_facts import resolve_layout

    layout = resolve_layout(model, session)
    self._num_layers = layout.num_layers
    self._module_layout = resolve_model_layout(model) if model is not None else None
    if tokenizer is not None:
        self.tokenizer = tokenizer
    self.interventions = tuple(
        intervention.bind(model, tokenizer, layout=layout, session=session)
        for intervention in self._template
    )
    return model

steer_access()

The model access the template's steer step requires, folded over its sources.

A fully bound template requires ModelAccess.FACTS, since pure layer selectors resolve from structural facts available on any session. Otherwise the strongest declared source access wins; a source or factory slot without an access declaration builds against the live model, so it counts as ModelAccess.MODULE.

Source code in steerability/algorithms/state_control/base.py 
364
365
366
367
368
369
370
371
372
373
374
375
def steer_access(self) -> ModelAccess:
    """The model access the template's steer step requires, folded over its sources.

    A fully bound template requires `ModelAccess.FACTS`, since pure layer selectors
    resolve from structural facts available on any session. Otherwise the strongest
    declared source access wins; a source or factory slot without an `access` declaration
    builds against the live model, so it counts as `ModelAccess.MODULE`.
    """
    access = ModelAccess.FACTS
    for source in self._unbound_sources():
        access = max(access, getattr(source, "access", ModelAccess.MODULE))
    return access

steer_fits()

The template's fit artifacts, i.e. every unbound source carrying an artifact_class, as (artifact, artifact_class) pairs in template order.

Source code in steerability/algorithms/state_control/base.py 
387
388
389
390
391
392
def steer_fits(self) -> tuple[tuple[str, str], ...]:
    """The template's fit artifacts, i.e. every unbound source carrying an
    `artifact_class`, as `(artifact, artifact_class)` pairs in template order."""
    return tuple(
        (type(source).__name__, source.artifact_class) for source in self._fit_sources()
    )

wire_kinds()

The combined wire kinds of the bound interventions (or the template before steer()), or None when any intervention is hook-only.

Source code in steerability/algorithms/state_control/base.py 
334
335
336
337
338
339
340
def wire_kinds(self):
    """The combined wire kinds of the bound interventions (or the template before
    `steer()`), or None when any intervention is hook-only."""
    from steerability.algorithms.state_control.common.specs import combine_kinds

    source = self.interventions or self._template
    return combine_kinds(intervention.wire_kinds() for intervention in source)

StateControl

Bases: BaseControl

Abstract root for state control steering methods; the class the pipeline type-checks.

Concrete state controls are either InterventionControl (a declarative intervention tuple, compiled to hooks by build_hooks and to InterventionSpec payloads by lower_interventions) or HookControl (raw torch hooks). Hooks are per-generation products the pipeline collects as HookEntry contributions; controls never register hooks and hold no model reference.

Methods:

Name Description
get_hooks

Create hook specs (required)
steer

One-time preparation (optional)
Source code in steerability/algorithms/state_control/base.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
class StateControl(BaseControl):
    """Abstract root for state control steering methods; the class the pipeline type-checks.

    Concrete state controls are either `InterventionControl` (a declarative intervention
    tuple, compiled to hooks by `build_hooks` and to `InterventionSpec` payloads by
    `lower_interventions`) or `HookControl` (raw torch hooks). Hooks are per-generation
    products the pipeline collects as `HookEntry` contributions; controls never register
    hooks and hold no model reference.

    Methods:
        get_hooks(input_ids, runtime_kwargs, **kwargs) -> dict: Create hook specs (required)
        steer(model, tokenizer, **kwargs) -> None: One-time preparation (optional)
    """

    Args: type[BaseArgs] | None = None
    RUNTIME_KWARGS_SCHEMA: list[dict] = []

    enabled: bool = True
    supports_batching: bool = False

    @abstractmethod
    def get_hooks(
        self,
        input_ids: torch.Tensor,
        runtime_kwargs: dict | None,
        **kwargs,
    ) -> dict[str, list[HookSpec]]:
        """Create hook specifications for the current generation.

        The pipeline forwards `attention_mask` (the prompt attention mask matching `input_ids`,
        or None) through `**kwargs`; controls that score prompt tokens may consume it to align
        with the real (non-pad) positions instead of re-deriving a mask by token identity.

        Args:
            input_ids: Prompt token ids of shape [batch, seq_len].
            runtime_kwargs: Per-call parameters for the control.
        """
        pass

    def steer(self,
              model: PreTrainedModel,
              tokenizer: PreTrainedTokenizerBase = None,
              session=None,
              **kwargs) -> None:
        """Optional steering/preparation.

        `session` is a `SteeringSession` on the steering backend, provided by the pipeline.
        """
        pass

    def export_intervention_spec(self, runtime_kwargs: dict | None = None) -> "InterventionSpec | None":
        """The control's `InterventionSpec` for intervention-capable backends, or None.

        The spec is the second serialization of the tuple the control's hooks close over,
        emitted from the same transform, gate, and scope objects. Must be called after
        `steer()`. Returns None when the configuration has no wire form (the configuration is
        then hook-only) or when the control does not implement spec export at all.

        Args:
            runtime_kwargs: Per-call parameters, mirroring `get_hooks`; per-item values
                (strengths, positions) serialize into the returned spec.

        Returns:
            The validated `InterventionSpec` with tensor payloads attached, or None.
        """
        return None

RUNTIME_KWARGS_SCHEMA = [] class-attribute instance-attribute

args = self.Args.validate(*args, **kwargs) instance-attribute

enabled = True class-attribute instance-attribute

supports_batching = False class-attribute instance-attribute

cleanup()

Release resources allocated during steer().

Override in subclasses that allocate GPU memory or other resources during steering. Default no-op.

Source code in steerability/algorithms/core/base_control.py 
195
196
197
198
199
200
201
def cleanup(self) -> None:
    """Release resources allocated during `steer()`.

    Override in subclasses that allocate GPU memory or other resources during steering. Default
    no-op.
    """
    pass

clone_for_call(seed=None)

A configuration-preserving shallow clone for one generation call.

The clone shares steer-time artifacts (memories, steering vectors, attached tokenizers) with the original but has its own attribute namespace, so per-call attribute mutation on the clone never races another call using the original. When seed is given and the control defines reseed(seed), the clone's client-side RNG is re-seeded.

Parameters:

Name Type Description Default
seed int | None

Optional seed forwarded to the clone's reseed().

 None

Returns:

Type Description
BaseControl

The clone.
Source code in steerability/algorithms/core/base_control.py 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
def clone_for_call(self, seed: int | None = None) -> "BaseControl":
    """A configuration-preserving shallow clone for one generation call.

    The clone shares steer-time artifacts (memories, steering vectors, attached tokenizers)
    with the original but has its own attribute namespace, so per-call attribute mutation on
    the clone never races another call using the original. When `seed` is given and the
    control defines `reseed(seed)`, the clone's client-side RNG is re-seeded.

    Args:
        seed: Optional seed forwarded to the clone's `reseed()`.

    Returns:
        The clone.
    """
    clone = copy.copy(self)
    if seed is not None:
        reseed = getattr(clone, "reseed", None)
        if callable(reseed):
            reseed(seed)
    return clone

export_intervention_spec(runtime_kwargs=None)

The control's InterventionSpec for intervention-capable backends, or None.

The spec is the second serialization of the tuple the control's hooks close over, emitted from the same transform, gate, and scope objects. Must be called after steer(). Returns None when the configuration has no wire form (the configuration is then hook-only) or when the control does not implement spec export at all.

Parameters:

Name Type Description Default
runtime_kwargs dict | None

Per-call parameters, mirroring get_hooks; per-item values (strengths, positions) serialize into the returned spec.

 None

Returns:

Type Description
InterventionSpec | None

The validated InterventionSpec with tensor payloads attached, or None.
Source code in steerability/algorithms/state_control/base.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def export_intervention_spec(self, runtime_kwargs: dict | None = None) -> "InterventionSpec | None":
    """The control's `InterventionSpec` for intervention-capable backends, or None.

    The spec is the second serialization of the tuple the control's hooks close over,
    emitted from the same transform, gate, and scope objects. Must be called after
    `steer()`. Returns None when the configuration has no wire form (the configuration is
    then hook-only) or when the control does not implement spec export at all.

    Args:
        runtime_kwargs: Per-call parameters, mirroring `get_hooks`; per-item values
            (strengths, positions) serialize into the returned spec.

    Returns:
        The validated `InterventionSpec` with tensor payloads attached, or None.
    """
    return None

export_state()

Steer-time state to persist when freezing, keyed by logical name.

Values are typed artifacts (SteeringVector, Probe, ProbeSet, a Memory implementation, CheckpointArtifact, LoRAArtifact, torch.Tensor, or a Path to an on-disk product). Must be called after steer(). The default returns an empty mapping (nothing to persist).

Returns:

Type Description
dict[str, Any]

Mapping from logical name to artifact value.
Source code in steerability/algorithms/core/base_control.py 
117
118
119
120
121
122
123
124
125
126
127
128
def export_state(self) -> dict[str, Any]:
    """Steer-time state to persist when freezing, keyed by logical name.

    Values are typed artifacts (`SteeringVector`, `Probe`, `ProbeSet`, a `Memory`
    implementation, `CheckpointArtifact`, `LoRAArtifact`, `torch.Tensor`, or a `Path` to an
    on-disk product). Must be called after `steer()`. The default returns an empty mapping
    (nothing to persist).

    Returns:
        Mapping from logical name to artifact value.
    """
    return {}

fit_identity()

The object whose canonical form defines this control's fit digest, or None.

The digest detects recipe edits that invalidate frozen artifacts, so the returned object should cover exactly the inputs a re-fit would consume (training data, fit specs) and exclude inert application parameters (strengths, multipliers). The default returns None (no fits).

Returns:

Type Description
Any | None

The fit-identity object, or None.
Source code in steerability/algorithms/core/base_control.py 
161
162
163
164
165
166
167
168
169
170
171
172
def fit_identity(self) -> Any | None:
    """The object whose canonical form defines this control's fit digest, or None.

    The digest detects recipe edits that invalidate frozen artifacts, so the returned
    object should cover exactly the inputs a re-fit would consume (training data, fit
    specs) and exclude inert application parameters (strengths, multipliers). The default
    returns None (no fits).

    Returns:
        The fit-identity object, or None.
    """
    return None

frozen_form(state)

The (registry method key, constructor kwargs) of this control's frozen form.

state is the output of export_state(); artifact values inside the returned kwargs are replaced by store references at encode time. When steer_fits() is empty and state is empty, the recipe is the frozen form and the control's own key and recipe args are returned unchanged. A control whose freeze expands into several entries may return a list of (method key, kwargs) pairs instead.

Parameters:

Name Type Description Default
state dict[str, Any]

The output of export_state().

 required

Returns:

Type Description
tuple[str, dict[str, Any]]

The frozen method key and its constructor kwargs.

Raises:

Type Description
NotFreezableError

If the control produces fits or state but declares no frozen form.
Source code in steerability/algorithms/core/base_control.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
def frozen_form(self, state: dict[str, Any]) -> tuple[str, dict[str, Any]]:
    """The `(registry method key, constructor kwargs)` of this control's frozen form.

    `state` is the output of `export_state()`; artifact values inside the returned kwargs
    are replaced by store references at encode time. When `steer_fits()` is empty and
    `state` is empty, the recipe is the frozen form and the control's own key and recipe
    args are returned unchanged. A control whose freeze expands into several entries may
    return a list of `(method key, kwargs)` pairs instead.

    Args:
        state: The output of `export_state()`.

    Returns:
        The frozen method key and its constructor kwargs.

    Raises:
        NotFreezableError: If the control produces fits or state but declares no frozen
            form.
    """
    if not state and not self.steer_fits():
        from steerability.algorithms.core.registry import method_key_for

        args = getattr(self, "args", None)
        kwargs = {field.name: getattr(args, field.name) for field in fields(args) if field.init} \
            if args is not None else {}
        return method_key_for(type(self)), kwargs
    raise NotFreezableError(
        f"{type(self).__name__} produces steer-time state but declares no frozen form; "
        "override frozen_form() (and export_state()) to support freezing."
    )

get_hooks(input_ids, runtime_kwargs, **kwargs) abstractmethod

Create hook specifications for the current generation.

The pipeline forwards attention_mask (the prompt attention mask matching input_ids, or None) through **kwargs; controls that score prompt tokens may consume it to align with the real (non-pad) positions instead of re-deriving a mask by token identity.

Parameters:

Name Type Description Default
input_ids Tensor

Prompt token ids of shape [batch, seq_len].

 required
runtime_kwargs dict | None

Per-call parameters for the control.

 required
Source code in steerability/algorithms/state_control/base.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
@abstractmethod
def get_hooks(
    self,
    input_ids: torch.Tensor,
    runtime_kwargs: dict | None,
    **kwargs,
) -> dict[str, list[HookSpec]]:
    """Create hook specifications for the current generation.

    The pipeline forwards `attention_mask` (the prompt attention mask matching `input_ids`,
    or None) through `**kwargs`; controls that score prompt tokens may consume it to align
    with the real (non-pad) positions instead of re-deriving a mask by token identity.

    Args:
        input_ids: Prompt token ids of shape [batch, seq_len].
        runtime_kwargs: Per-call parameters for the control.
    """
    pass

requirements()

Backend requirements computed from this instance's configuration, per phase.

The default requires Capability.IN_PROCESS_TORCH at generate and nothing at score, which only the Hugging Face backend satisfies. A control with portable mechanisms overrides this to state weaker or alternative requirements. Configuration determines the result, so two configurations of one class may differ. Only enabled controls are consulted during support evaluation.

Returns:

Type Description
Requirements

The control's phase-keyed requirements.
Source code in steerability/algorithms/core/base_control.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
def requirements(self) -> Requirements:
    """Backend requirements computed from this instance's configuration, per phase.

    The default requires `Capability.IN_PROCESS_TORCH` at generate and nothing at score,
    which only the Hugging Face backend satisfies. A control with portable mechanisms
    overrides this to state weaker or alternative requirements. Configuration determines the
    result, so two configurations of one class may differ. Only enabled controls are
    consulted during support evaluation.

    Returns:
        The control's phase-keyed requirements.
    """
    return Requirements(generate=needs(Capability.IN_PROCESS_TORCH))

steer(model, tokenizer=None, session=None, **kwargs)

Optional steering/preparation.

session is a SteeringSession on the steering backend, provided by the pipeline.

Source code in steerability/algorithms/state_control/base.py 
127
128
129
130
131
132
133
134
135
136
def steer(self,
          model: PreTrainedModel,
          tokenizer: PreTrainedTokenizerBase = None,
          session=None,
          **kwargs) -> None:
    """Optional steering/preparation.

    `session` is a `SteeringSession` on the steering backend, provided by the pipeline.
    """
    pass

steer_access()

The model access this instance's steer step requires, on the ModelAccess ladder.

The default is ModelAccess.FACTS (layout and tokenizer only). A control whose steer step generates or scores through the session declares ROLLOUTS, one that captures hidden states declares CAPTURE, and one that touches the model as a live torch.nn.Module declares MODULE. Configuration determines the result. The pipeline hands steer() a session scoped to the declared rung, and the live model only at MODULE. A control may retain the pipeline model beyond steer() only if its generate phase requires Capability.IN_PROCESS_TORCH.

Returns:

Type Description
ModelAccess

The declared access rung.
Source code in steerability/algorithms/core/base_control.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def steer_access(self) -> ModelAccess:
    """The model access this instance's steer step requires, on the `ModelAccess` ladder.

    The default is `ModelAccess.FACTS` (layout and tokenizer only). A control whose steer
    step generates or scores through the session declares `ROLLOUTS`, one that captures
    hidden states declares `CAPTURE`, and one that touches the model as a live
    `torch.nn.Module` declares `MODULE`. Configuration determines the result. The pipeline
    hands `steer()` a session scoped to the declared rung, and the live model only at
    `MODULE`. A control may retain the pipeline model beyond `steer()` only if its
    generate phase requires `Capability.IN_PROCESS_TORCH`.

    Returns:
        The declared access rung.
    """
    return ModelAccess.FACTS

steer_fits()

The fit artifacts this instance's steer step will produce, for the steer plan.

Each entry is (artifact, artifact_class), where artifact is the fit source or recipe class name and artifact_class is "direction" or "calibrated". The default is an empty tuple (no fits).

Returns:

Type Description
tuple[tuple[str, str], ...]

The declared fit artifacts, in declaration order.
Source code in steerability/algorithms/core/base_control.py 
105
106
107
108
109
110
111
112
113
114
115
def steer_fits(self) -> tuple[tuple[str, str], ...]:
    """The fit artifacts this instance's steer step will produce, for the steer plan.

    Each entry is `(artifact, artifact_class)`, where `artifact` is the fit source or
    recipe class name and `artifact_class` is `"direction"` or `"calibrated"`. The default
    is an empty tuple (no fits).

    Returns:
        The declared fit artifacts, in declaration order.
    """
    return ()