CAST

steerability.algorithms.state_control.cast

args

CAST argument validation.

control

CAST control: conditional activation steering, composed from common components.

logger = logging.getLogger(__name__) module-attribute

CAST

Bases: InterventionControl

Conditional Activation Steering (CAST).

CAST enables selective control of LLM behavior by conditionally applying activation steering based on input context. It operates in two phases:

1. Condition Detection: Scores hidden-state activation patterns at the condition layer(s) against a learned condition direction to detect whether the prompt matches the target condition.

2. Conditional Behavior Modification: When the condition is met, applies a behavior transform to hidden states at the behavior layers.

The control is declarative: _configure maps the validated args onto one Intervention at the layer-input boundary whose gate comes from a ConditionPointSearch source (fitting the condition vector and grid-searching the gate point at bind time), and whose transform comes from the default additive build or the behavior_transform slot. The runtime pieces it resolves to are the common component families:

• ContrastiveDirectionEstimator / MeanDifferenceEstimator: learn per-layer direction vectors from contrastive text pairs.
• ConditionPointSelector: grid-searches the (layer, threshold, comparator) that best separates positive from negative calibration examples.
• Gate(Evidence(..., ProjectedCosineReadout(...)), PerKeyThreshold(...)): row-vectorized gating. Evidence pooling is pad-aware ("mean" or "last") and each pooled state is scored per row via projected cosine similarity. Each prompt in a batch is gated independently; beam-expanded rows of one prompt share that prompt's decision; the decision freezes after the prefill pass (the runtime stops condition scoring once the gate reports ready).
• The behavior transform: AdditiveTransform (scaled direction addition, optionally wrapped in NormPreservingTransform) by default, or any BaseTransform supplied via behavior_transform (e.g. ProjectionTransform for conditional ablation).

The intervention applies at the layer-input boundary. Behavior directions are estimated at the output of layer l (hidden_states[l+1]) and applied at the input of layer l (the output of layer l-1), a one-layer skew. Condition directions are estimated by default at the input of layer l (VectorTrainSpec(location="layer_input") in CASTArgs.condition_fit), the boundary the ConditionPointSelector calibrates on and the runtime condition pre-hook scores, so condition fit, calibration, and runtime are aligned.

Within the prefill pass, hooks fire in layer order. A behavior layer below the condition layer sees a still-closed gate with no evidence yet, while a behavior layer above it sees the decided gate. When the calibrated condition layer sits above the behavior layers, prompt tokens pass the behavior layers unsteered and only decode steps are steered. Token scope composes with this: "all" permits prompt-token steering wherever the gate is already decided during prefill; "after_prompt" restricts steering to generated tokens regardless of layer order.

Batching is supported (supports_batching = True). Row-vectorized gates let one batched generate call gate and steer each prompt independently.

Reference:

• "Programming Refusal with Conditional Activation Steering" Bruce W. Lee, Inkit Padhi, Karthikeyan Natesan Ramamurthy, Erik Miehling, Pierre Dognin, Manish Nagireddy, Amit Dhurandhar https://arxiv.org/abs/2409.05907

Source code in steerability/algorithms/state_control/cast/control.py

class CAST(InterventionControl):
    """Conditional Activation Steering (CAST).

    CAST enables selective control of LLM behavior by conditionally applying activation steering
    based on input context. It operates in two phases:

    1. **Condition Detection**: Scores hidden-state activation patterns at the condition layer(s)
       against a learned condition direction to detect whether the prompt matches the target
       condition.

    2. **Conditional Behavior Modification**: When the condition is met, applies a behavior
       transform to hidden states at the behavior layers.

    The control is declarative: `_configure` maps the validated args onto one `Intervention`
    at the layer-input boundary whose gate comes from a `ConditionPointSearch` source (fitting
    the condition vector and grid-searching the gate point at bind time), and whose transform
    comes from the default additive build or the `behavior_transform` slot. The runtime pieces
    it resolves to are the `common` component families:

    - `ContrastiveDirectionEstimator` / `MeanDifferenceEstimator`: learn per-layer direction
      vectors from contrastive text pairs.
    - `ConditionPointSelector`: grid-searches the (layer, threshold, comparator) that best
      separates positive from negative calibration examples.
    - `Gate(Evidence(..., ProjectedCosineReadout(...)), PerKeyThreshold(...))`: row-vectorized
      gating. Evidence pooling is pad-aware ("mean" or "last") and each pooled state is scored
      per row via projected cosine similarity. Each prompt in a batch is gated independently;
      beam-expanded rows of one prompt share that prompt's decision; the decision freezes after
      the prefill pass (the runtime stops condition scoring once the gate reports ready).
    - The behavior transform: `AdditiveTransform` (scaled direction addition, optionally wrapped
      in `NormPreservingTransform`) by default, or any `BaseTransform` supplied via
      `behavior_transform` (e.g. `ProjectionTransform` for conditional ablation).

    The intervention applies at the layer-input boundary. Behavior directions are estimated at
    the output of layer l (`hidden_states[l+1]`) and applied at the input of layer l (the
    output of layer l-1), a one-layer skew. Condition directions are estimated by default at
    the input of layer l (`VectorTrainSpec(location="layer_input")` in `CASTArgs.condition_fit`),
    the boundary the `ConditionPointSelector` calibrates on and the runtime condition pre-hook
    scores, so condition fit, calibration, and runtime are aligned.

    Within the prefill pass, hooks fire in layer order. A behavior layer below the condition layer
    sees a still-closed gate with no evidence yet, while a behavior layer above it sees the decided
    gate. When the calibrated condition layer sits above the behavior layers, prompt tokens pass
    the behavior layers unsteered and only decode steps are steered. Token scope composes with
    this: `"all"` permits prompt-token steering wherever the gate is already decided during
    prefill; `"after_prompt"` restricts steering to generated tokens regardless of layer order.

    Batching is supported (`supports_batching = True`). Row-vectorized gates let one batched
    `generate` call gate and steer each prompt independently.

    Reference:

    - "Programming Refusal with Conditional Activation Steering"
    Bruce W. Lee, Inkit Padhi, Karthikeyan Natesan Ramamurthy, Erik Miehling, Pierre Dognin,
    Manish Nagireddy, Amit Dhurandhar
    [https://arxiv.org/abs/2409.05907](https://arxiv.org/abs/2409.05907)
    """

    Args = CASTArgs
    supports_batching = True

    def _configure(self):
        if self.behavior_transform is not None:
            transform = self.behavior_transform
            require_coverage = True
        else:
            if self.behavior_vector is not None:
                source = _Precomputed(self.behavior_vector.clone())
            else:
                source = _BehaviorFit(self.behavior_data, self.behavior_fit)
            transform = _BehaviorBuild(
                source,
                strength=self.behavior_vector_strength,
                use_explained_variance=self.use_explained_variance,
                norm_preserving=self.use_ooi_preventive_normalization,
            )
            require_coverage = False

        self._condition_source = ConditionPointSearch(
            condition_vector=self.condition_vector.clone() if self.condition_vector is not None else None,
            condition_data=self.condition_data,
            condition_fit=self.condition_fit,
            search=self.search,
            layer_ids=self.condition_layer_ids,
            threshold=self.condition_vector_threshold,
            comparator=self.condition_comparator_threshold_is,
            comparison_mode=self.condition_threshold_comparison_mode,
        )

        self._template = (Intervention(
            layers=tuple(sorted(set(int(lid) for lid in self.behavior_layer_ids)))
                   if self.behavior_layer_ids is not None else LateThirdSelector(),
            transform=transform,
            scope=TokenScope(self.token_scope, last_k=self.last_k, from_position=self.from_position),
            gate=self._condition_source,
            boundary="layer_input",
            require_coverage=require_coverage,
        ),)

    @property
    def _behavior_layer_ids(self) -> list[int]:
        """The resolved behavior layers (empty before `steer()`)."""
        return list(self.interventions[0].layers) if self.interventions else []

    @property
    def _threshold_rule(self) -> PerKeyThreshold | None:
        """The gate's threshold rule, for diagnostics; None when unconditional or unbound."""
        gate = self._gate
        if isinstance(gate, Gate) and isinstance(gate.rule, PerKeyThreshold):
            return gate.rule
        return None

    @property
    def _cond_config(self) -> ConditionPointConfig | None:
        """The resolved condition point, or None before `steer()`."""
        if not self.interventions:
            return None
        point = self._condition_source.resolved_point
        if point is None:
            return ConditionPointConfig(
                layer_ids=frozenset(),
                threshold=None,
                comparator=self.condition_comparator_threshold_is,
                comparison_mode=self.condition_threshold_comparison_mode,
                enabled=False,
            )
        return ConditionPointConfig(
            layer_ids=frozenset(point["layer_ids"]),
            threshold=point["threshold"],
            comparator=point["comparator"],
            comparison_mode=point["comparison_mode"],
            enabled=True,
        )

    @property
    def latest_decision(self) -> CASTDecision | None:
        """The most recent condition decision, or None before the condition has been evaluated.

        Assembled on demand from the gate's retained evidence; cleared when the next
        generation's hooks are built.
        """
        rule = self._threshold_rule
        gate = self._gate
        if rule is None or gate is None or not gate.is_ready():
            return None
        evidence = gate.evidence_values()
        if not evidence:
            return None
        open_rows = gate.open_rows()
        return CASTDecision(
            scores={lid: float(rows[0]) for lid, rows in evidence.items()},
            scores_per_row={lid: tuple(float(x) for x in rows) for lid, rows in evidence.items()},
            threshold=rule.threshold,
            comparator=rule.comparator,
            open_per_row=tuple(bool(x) for x in open_rows.tolist()),
        )

    @property
    def condition_point(self) -> dict | None:
        """The resolved condition configuration, or None when the control is unconditional.

        Populated by `steer()` from either the auto-search or the caller-supplied
        `condition_layer_ids` / `condition_vector_threshold` / `condition_comparator_threshold_is`.
        """
        cfg = self._cond_config
        if cfg is None or not cfg.enabled:
            return None
        return {
            "layer_ids": sorted(cfg.layer_ids),
            "threshold": cfg.threshold,
            "comparator": cfg.comparator,
            "comparison_mode": cfg.comparison_mode,
        }

    def cleanup(self) -> None:
        """Drop references to fitted artifacts and runtime state."""
        self.interventions = ()

RUNTIME_KWARGS_SCHEMA = [] class-attribute instance-attribute

_behavior_layer_ids property

The resolved behavior layers (empty before steer()).

_cond_config property

The resolved condition point, or None before steer().

_gate property

The first intervention's gate (None before steer() or when unconditional).

_template = () class-attribute instance-attribute

_threshold_rule property

The gate's threshold rule, for diagnostics; None when unconditional or unbound.

_transform property writable

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.

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

condition_point property

The resolved condition configuration, or None when the control is unconditional.

Populated by steer() from either the auto-search or the caller-supplied condition_layer_ids / condition_vector_threshold / condition_comparator_threshold_is.

enabled = True class-attribute instance-attribute

hook_only_hint = None class-attribute instance-attribute

interventions = () class-attribute instance-attribute

latest_decision property

The most recent condition decision, or None before the condition has been evaluated.

Assembled on demand from the gate's retained evidence; cleared when the next generation's hooks are built.

supports_batching = True class-attribute instance-attribute

tokenizer = None class-attribute instance-attribute

__init__(*args, **kwargs)

Source code in steerability/algorithms/core/base_control.py

def __init__(self, *args, **kwargs) -> None:
    if self.Args is None:  # null / arg-free control
        if args or kwargs:
            raise TypeError(f"{type(self).__name__} accepts no constructor arguments.")
        self._configure()
        return

    self.args: BaseArgs = self.Args.validate(*args, **kwargs)

    # move fields to attributes, skipping any name the control exposes as a property
    # (e.g. CAST.condition_point); the raw value stays reachable via self.args.<name>
    for field in fields(self.args):
        if isinstance(getattr(type(self), field.name, None), property):
            continue
        setattr(self, field.name, getattr(self.args, field.name))

    self._configure()

_configure()

Post-construction hook, called after Args fields are mirrored onto the instance.

Driver presets override this to map their mirrored args onto the fields their generic base reads, so subclasses never bypass a parent __init__. Default no-op.

Source code in steerability/algorithms/state_control/cast/control.py

def _configure(self):
    if self.behavior_transform is not None:
        transform = self.behavior_transform
        require_coverage = True
    else:
        if self.behavior_vector is not None:
            source = _Precomputed(self.behavior_vector.clone())
        else:
            source = _BehaviorFit(self.behavior_data, self.behavior_fit)
        transform = _BehaviorBuild(
            source,
            strength=self.behavior_vector_strength,
            use_explained_variance=self.use_explained_variance,
            norm_preserving=self.use_ooi_preventive_normalization,
        )
        require_coverage = False

    self._condition_source = ConditionPointSearch(
        condition_vector=self.condition_vector.clone() if self.condition_vector is not None else None,
        condition_data=self.condition_data,
        condition_fit=self.condition_fit,
        search=self.search,
        layer_ids=self.condition_layer_ids,
        threshold=self.condition_vector_threshold,
        comparator=self.condition_comparator_threshold_is,
        comparison_mode=self.condition_threshold_comparison_mode,
    )

    self._template = (Intervention(
        layers=tuple(sorted(set(int(lid) for lid in self.behavior_layer_ids)))
               if self.behavior_layer_ids is not None else LateThirdSelector(),
        transform=transform,
        scope=TokenScope(self.token_scope, last_k=self.last_k, from_position=self.from_position),
        gate=self._condition_source,
        boundary="layer_input",
        require_coverage=require_coverage,
    ),)

_fit_sources()

Yield the template's fit sources: unbound sources carrying an artifact_class whose resolution does model-side work (access above ModelAccess.FACTS).

Source code in steerability/algorithms/state_control/base.py

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

_resolve_module_layout(model=None)

The module-path layout, resolved from the module tree on first use.

Source code in steerability/algorithms/state_control/base.py

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

_unbound_sources()

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.

Source code in steerability/algorithms/state_control/base.py

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

cleanup()

Drop references to fitted artifacts and runtime state.

Source code in steerability/algorithms/state_control/cast/control.py

def cleanup(self) -> None:
    """Drop references to fitted artifacts and runtime state."""
    self.interventions = ()

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

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

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

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

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

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

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

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

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

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

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

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)

CASTDecision dataclass

Diagnostics snapshot of the most recent condition decision.

Attributes:

Name	Type	Description
scores	dict[int, float]	Per-condition-layer score for the first logical row (single-prompt convenience).
scores_per_row	dict[int, tuple[float, ...]]	Per-condition-layer scores for every logical row.
threshold	float | None	The gate threshold in effect.
comparator	Comparator | None	The gate comparator in effect.
open_per_row	tuple[bool, ...]	Whether the gate opened, per logical row.

Source code in steerability/algorithms/state_control/cast/control.py

@dataclass(frozen=True)
class CASTDecision:
    """Diagnostics snapshot of the most recent condition decision.

    Attributes:
        scores: Per-condition-layer score for the first logical row (single-prompt convenience).
        scores_per_row: Per-condition-layer scores for every logical row.
        threshold: The gate threshold in effect.
        comparator: The gate comparator in effect.
        open_per_row: Whether the gate opened, per logical row.
    """

    scores: dict[int, float]
    scores_per_row: dict[int, tuple[float, ...]]
    threshold: float | None
    comparator: Comparator | None
    open_per_row: tuple[bool, ...]

    @property
    def is_open(self) -> bool:
        """True if the gate opened for the first logical row (single-prompt convenience)."""
        return bool(self.open_per_row and self.open_per_row[0])

comparator instance-attribute

is_open property

True if the gate opened for the first logical row (single-prompt convenience).

open_per_row instance-attribute

scores instance-attribute

scores_per_row instance-attribute

threshold instance-attribute

__init__(scores, scores_per_row, threshold, comparator, open_per_row)

ConditionPointConfig dataclass

Fully-resolved condition point produced by binding the condition source.

Attributes:

Name	Type	Description
layer_ids	frozenset[int]	Condition layer ids (0-based), empty when unconditional.
threshold	float | None	Gate threshold, or None when unconditional.
comparator	Comparator	Gate comparator.
comparison_mode	CompMode	Runtime token aggregation for condition scoring ("mean" or "last").
enabled	bool	Whether conditional gating is active. When False, behavior steering is always on.

Source code in steerability/algorithms/state_control/cast/control.py

@dataclass(frozen=True)
class ConditionPointConfig:
    """Fully-resolved condition point produced by binding the condition source.

    Attributes:
        layer_ids: Condition layer ids (0-based), empty when unconditional.
        threshold: Gate threshold, or None when unconditional.
        comparator: Gate comparator.
        comparison_mode: Runtime token aggregation for condition scoring ("mean" or "last").
        enabled: Whether conditional gating is active. When False, behavior steering is always on.
    """

    layer_ids: frozenset[int]
    threshold: float | None
    comparator: Comparator
    comparison_mode: CompMode
    enabled: bool

comparator instance-attribute

comparison_mode instance-attribute

enabled instance-attribute

layer_ids instance-attribute

threshold instance-attribute

__init__(layer_ids, threshold, comparator, comparison_mode, enabled)

_BehaviorBuild

Transform factory for CAST's default additive path.

Resolves the behavior artifact, squeezes each covered layer's direction, applies the explained-variance scaling when enabled, and builds the additive transform (optionally norm-preserving). Behavior layers without a fitted direction are skipped, so their hooks pass through unchanged. The factory declares its wire plan (wire_plan, wire_modifiers), so kind identity is readable before binding.

Source code in steerability/algorithms/state_control/cast/control.py

class _BehaviorBuild:
    """Transform factory for CAST's default additive path.

    Resolves the behavior artifact, squeezes each covered layer's direction, applies the
    explained-variance scaling when enabled, and builds the additive transform (optionally
    norm-preserving). Behavior layers without a fitted direction are skipped, so their hooks
    pass through unchanged. The factory declares its wire plan (`wire_plan`,
    `wire_modifiers`), so kind identity is readable before binding.
    """

    def __init__(self, source, strength: float, use_explained_variance: bool, norm_preserving: bool):
        self._source = source
        self._strength = strength
        self._use_explained_variance = use_explained_variance
        self._norm_preserving = norm_preserving

    @property
    def access(self) -> ModelAccess:
        return getattr(self._source, "access", ModelAccess.MODULE)

    @property
    def artifact_class(self) -> str | None:
        return getattr(self._source, "artifact_class", None)

    def fit_ingredients(self) -> dict:
        """The encodable fit-identity form: the source's ingredients plus the fit-relevant
        scaling switch."""
        from steerability.algorithms.state_control.base import _fit_ingredients

        return {
            "kind": "behavior_build",
            "source": _fit_ingredients(self._source),
            "use_explained_variance": self._use_explained_variance,
        }

    def wire_plan(self) -> str | None:
        """`"additive"` for broadcast behavior directions; None when the source is positional."""
        if getattr(self._source, "produces_positional", False):
            return None
        return "additive"

    def wire_modifiers(self) -> tuple[str, ...]:
        """The planned wrapper kinds: `norm_preserving` when OOI normalization is enabled."""
        return ("norm_preserving",) if self._norm_preserving else ()

    def __call__(self, ctx) -> BaseTransform:
        behavior_vec = ctx.resolve(self._source)
        directions: dict[int, torch.Tensor] = {}
        for layer_id in ctx.layer_ids:
            direction = behavior_vec.directions.get(layer_id)
            if direction is None:
                continue
            direction = _squeeze_direction(direction)
            if self._use_explained_variance and behavior_vec.explained_variances:
                direction = direction * float(behavior_vec.explained_variances.get(layer_id, 1.0))
            directions[layer_id] = direction

        base_transform: BaseTransform = AdditiveTransform(directions, strength=self._strength)
        if self._norm_preserving:
            return NormPreservingTransform(base_transform)
        return base_transform

_norm_preserving = norm_preserving instance-attribute

_source = source instance-attribute

_strength = strength instance-attribute

_use_explained_variance = use_explained_variance instance-attribute

access property

artifact_class property

__call__(ctx)

Source code in steerability/algorithms/state_control/cast/control.py

def __call__(self, ctx) -> BaseTransform:
    behavior_vec = ctx.resolve(self._source)
    directions: dict[int, torch.Tensor] = {}
    for layer_id in ctx.layer_ids:
        direction = behavior_vec.directions.get(layer_id)
        if direction is None:
            continue
        direction = _squeeze_direction(direction)
        if self._use_explained_variance and behavior_vec.explained_variances:
            direction = direction * float(behavior_vec.explained_variances.get(layer_id, 1.0))
        directions[layer_id] = direction

    base_transform: BaseTransform = AdditiveTransform(directions, strength=self._strength)
    if self._norm_preserving:
        return NormPreservingTransform(base_transform)
    return base_transform

__init__(source, strength, use_explained_variance, norm_preserving)

Source code in steerability/algorithms/state_control/cast/control.py

def __init__(self, source, strength: float, use_explained_variance: bool, norm_preserving: bool):
    self._source = source
    self._strength = strength
    self._use_explained_variance = use_explained_variance
    self._norm_preserving = norm_preserving

fit_ingredients()

The encodable fit-identity form: the source's ingredients plus the fit-relevant scaling switch.

Source code in steerability/algorithms/state_control/cast/control.py

def fit_ingredients(self) -> dict:
    """The encodable fit-identity form: the source's ingredients plus the fit-relevant
    scaling switch."""
    from steerability.algorithms.state_control.base import _fit_ingredients

    return {
        "kind": "behavior_build",
        "source": _fit_ingredients(self._source),
        "use_explained_variance": self._use_explained_variance,
    }

wire_modifiers()

The planned wrapper kinds: norm_preserving when OOI normalization is enabled.

Source code in steerability/algorithms/state_control/cast/control.py

def wire_modifiers(self) -> tuple[str, ...]:
    """The planned wrapper kinds: `norm_preserving` when OOI normalization is enabled."""
    return ("norm_preserving",) if self._norm_preserving else ()

wire_plan()

"additive" for broadcast behavior directions; None when the source is positional.

Source code in steerability/algorithms/state_control/cast/control.py

def wire_plan(self) -> str | None:
    """`"additive"` for broadcast behavior directions; None when the source is positional."""
    if getattr(self._source, "produces_positional", False):
        return None
    return "additive"

_BehaviorFit

A fit recipe for CAST's behavior vector, dispatched through _make_estimator.

Source code in steerability/algorithms/state_control/cast/control.py

class _BehaviorFit:
    """A fit recipe for CAST's behavior vector, dispatched through `_make_estimator`."""

    access = ModelAccess.MODULE
    artifact_class = "direction"

    def __init__(self, data, fit_spec: VectorTrainSpec):
        self._data = data
        self._fit_spec = fit_spec

    def fit_ingredients(self) -> dict:
        """The encodable fit-identity form: the contrastive data and the fit spec."""
        return {"kind": "behavior_fit", "data": self._data, "fit_spec": self._fit_spec}

    def resolve(self, model, tokenizer, *, session=None):
        estimator = _make_estimator(self._fit_spec)
        return estimator.fit(model, tokenizer, data=self._data, spec=self._fit_spec, session=session)

_data = data instance-attribute

_fit_spec = fit_spec instance-attribute

access = ModelAccess.MODULE class-attribute instance-attribute

artifact_class = 'direction' class-attribute instance-attribute

__init__(data, fit_spec)

Source code in steerability/algorithms/state_control/cast/control.py

def __init__(self, data, fit_spec: VectorTrainSpec):
    self._data = data
    self._fit_spec = fit_spec

fit_ingredients()

The encodable fit-identity form: the contrastive data and the fit spec.

Source code in steerability/algorithms/state_control/cast/control.py

def fit_ingredients(self) -> dict:
    """The encodable fit-identity form: the contrastive data and the fit spec."""
    return {"kind": "behavior_fit", "data": self._data, "fit_spec": self._fit_spec}

resolve(model, tokenizer, *, session=None)

Source code in steerability/algorithms/state_control/cast/control.py

def resolve(self, model, tokenizer, *, session=None):
    estimator = _make_estimator(self._fit_spec)
    return estimator.fit(model, tokenizer, data=self._data, spec=self._fit_spec, session=session)

_make_estimator(spec)

Dispatch a fit spec to its estimator.

Raises:

Type	Description
ValueError	If method == "mean_diff" is combined with accumulate == "suffix-only", which the mean-difference estimator does not support.

Source code in steerability/algorithms/state_control/cast/control.py

def _make_estimator(spec: VectorTrainSpec):
    """Dispatch a fit spec to its estimator.

    Raises:
        ValueError: If `method == "mean_diff"` is combined with `accumulate == "suffix-only"`,
            which the mean-difference estimator does not support.
    """
    if spec.method == "mean_diff":
        if spec.accumulate == "suffix-only":
            raise ValueError(
                "method='mean_diff' does not support accumulate='suffix-only'; "
                "use accumulate='all' or 'last_token', or method='pca_pairwise'/'pca_center'."
            )
        return MeanDifferenceEstimator()
    return ContrastiveDirectionEstimator()

_squeeze_direction(d)

Squeeze a [1, H] direction to [H] for scalar operations.

Handles both 1D [H] and 2D [K, D] tensors. For K=1, squeezes to [D]. For K>1, returns as-is (caller must handle).

Source code in steerability/algorithms/state_control/cast/control.py

def _squeeze_direction(d: torch.Tensor) -> torch.Tensor:
    """Squeeze a [1, H] direction to [H] for scalar operations.

    Handles both 1D [H] and 2D [K, D] tensors. For K=1, squeezes to [D].
    For K>1, returns as-is (caller must handle).
    """
    if d.ndim == 2 and d.shape[0] == 1:
        return d.squeeze(0)
    return d