Orca (full model shield)

Orca protects a whole model: it cleans the input before the model sees it, runs the model, and checks that the model's answer is not broken either.

Armatura protects one series. But if you have a model — any callable that takes a tensor and returns a tensor — the corruption can happen twice: once on the way in (garbage input) and once on the way out (the model itself producing a wrong response). Orca shields both sides in one call: it purifies the input, runs the model, and checks that the response itself isn't corrupted.

The model and the input

A small model, the identity times two, and a 4×200 batch of sine waves with one absurd value at (0, 50); ref is a clean reference of the same shape.

1. Protect the forward pass

import numpy as np
from dense_armor.utility.protect.orca import Orca

rng = np.random.default_rng(0)
ref = np.sin(np.linspace(0, 6, 200))[None, :] + 0.01 * rng.standard_normal((4, 200))
data = ref.copy()
data[0, 50] = 999.0
orca = Orca(min_free_ram_percentage=0.05)
out = orca.protect_and_forward(lambda z: 2.0 * z, data, x_reference=ref)
print(out.shape, round(float(out[0, 50]), 4), round(float(2 * ref[0, 50]), 4))

(4, 200) 2.0074 2.0031

out is the model's output on the cleaned input. The spike at (0, 50) never reaches the model: the output there is 2.0074, next to twice the clean value (2.0031), instead of 1998. min_free_ram_percentage lowers Orca's memory guard (default: 15 % free RAM required).

The four phases of protect_and_forward, in order:

1. Input shield — purify x_corrupted against x_reference (use_input_shield).
2. Model call — run the callable on the purified tensor (use_model_injection).
3. Output shield — check the model's response against its expected response to the clean reference (use_output_shield).
4. Report — return the protected output.

Every phase can be disabled individually: use_input_shield=False, use_model_injection=False, use_output_shield=False. All three default to True.

2. What each argument does

• ai_model_callable — any callable z -> z_out. In the example it is the identity scaled by two, chosen so the shield is the only thing acting on the tensor.
• x_corrupted — the tensor that would be passed to the model. It is cleaned before the model sees it.
• x_reference — optional but recommended. A clean tensor of the same shape as x_corrupted. When present, the input shield compares each sample against the corresponding reference sample instead of falling back to a robust blind estimate.
• use_arbiter=True — routes each point through the per-point classifier described in Arbiter: clean passes through whatever the standard shield produced, spike gets hard-rejected to the window median, regime passes through raw. Off by default.

3. Where the two stages come from

Orca's pipeline has two stages:

1. Stage 1 — AdaptiveSignalStabilizer: a causal recursive filter with a sigmoid damping curve. This is the adaptive engine. It does the actual purification of the input and the response.
2. Stage 2 — Collatz-based gate: a decision on how much to damp toward the clean reference. This is where Orca chooses between "reject the sample" and "trust it".

With use_arbiter=True, a third layer sits between Stage 1 and Stage 2 and decides, point by point, which of the two Stage-1 behaviours to use.

4. What the shield saw, after the fact

import numpy as np
from dense_armor.utility.protect.orca import Orca

rng = np.random.default_rng(0)
ref = np.sin(np.linspace(0, 6, 200))[None, :] + 0.01 * rng.standard_normal((4, 200))
data = ref.copy()
data[0, 50] = 999.0
orca = Orca(min_free_ram_percentage=0.05)
orca.protect_and_forward(lambda z: 2.0 * z, data, x_reference=ref, use_arbiter=True)
print(orca.tipi_corruzione_visti(data.shape[1:]))

Counter({'clean': 721, 'regime': 75, 'spike': 4})

tipi_corruzione_visti(row_shape) returns how many times each Arbiter label was seen for rows of that shape (here (200,)), counted only when use_arbiter=True and x_reference is given. Over the 4 × 200 samples: 721 clean, 75 regime (the sine's own slow changes of level), 4 spikes.

API reference

orca

utility/orca.py Universal AI Orchestrator Shield (ORCA) -- orchestratore dinamico selettivo context-aware a 4 fasi.

Orca

Orca(
    static_threshold: float = 0.15,
    initial_damping: float = 0.85,
    alpha: float = 0.05,
    val_e: float = -4.0,
    chunk_threshold: Optional[int] = None,
    min_free_ram_percentage: float = 0.15,
    reference_memory_size: int = 32,
    reference_recall_min_score: float = 0.9,
)

val_e — esponente di scala target per la compressione log10 in ingresso; chunk_threshold — dimensione oltre la quale un batch viene processato a blocchi; None (default) lo auto-deriva da AIHardwareProfiler (RAM/backend dell'host corrente, vedi _CHUNK_THRESHOLD_BASELINE* sopra) invece del vecchio valore fisso uguale per qualunque macchina; reference_memory_size — quanti riferimenti puliti passati (per shape) tenere in memoria per il richiamo via risonanza in modalita' cieca (vedi _recall_reference); reference_recall_min_score — punteggio minimo di apply_fast_resonance sotto il quale un riferimento in memoria viene considerato non abbastanza simile e si ricade su _blind_reference (calibrato empiricamente: su vettori correlati lo score resta >0.95 anche con rumore moderato, su vettori scorrelati non supera ~0.75 — 0.90 lascia margine da entrambi i lati).

Source code in dense_armor/utility/protect/orca.py

def __init__(self, static_threshold: float = 0.15, initial_damping: float = 0.85,
             alpha: float = 0.05, val_e: float = -4.0,
             chunk_threshold: Optional[int] = None, min_free_ram_percentage: float = 0.15,
             reference_memory_size: int = 32, reference_recall_min_score: float = 0.90) -> None:
    """val_e — esponente di scala target per la compressione log10 in ingresso;
    chunk_threshold — dimensione oltre la quale un batch viene processato a blocchi;
    None (default) lo auto-deriva da AIHardwareProfiler (RAM/backend
    dell'host corrente, vedi _CHUNK_THRESHOLD_BASELINE* sopra) invece del
    vecchio valore fisso uguale per qualunque macchina;
    reference_memory_size — quanti riferimenti puliti passati (per shape) tenere in
    memoria per il richiamo via risonanza in modalita' cieca (vedi _recall_reference);
    reference_recall_min_score — punteggio minimo di apply_fast_resonance sotto il
    quale un riferimento in memoria viene considerato non abbastanza simile e si
    ricade su _blind_reference (calibrato empiricamente: su vettori correlati lo
    score resta >0.95 anche con rumore moderato, su vettori scorrelati non supera
    ~0.75 — 0.90 lascia margine da entrambi i lati)."""
    self.static_threshold = static_threshold
    self.initial_damping = initial_damping
    self.alpha = alpha
    if chunk_threshold is None:
        profiler = AIHardwareProfiler()
        scale = profiler.max_tensor_dim / self._CHUNK_THRESHOLD_BASELINE_TENSOR_DIM
        chunk_threshold = int(self._CHUNK_THRESHOLD_BASELINE * scale)
        logger.info(
            "chunk_threshold auto-derivato da AIHardwareProfiler: %d (%s)",
            chunk_threshold, profiler.get_profile_summary(),
        )
    self.chunk_threshold = int(chunk_threshold)
    self.val_e = float(val_e)
    self.min_free_ram = min_free_ram_percentage
    self.reference_recall_min_score = float(reference_recall_min_score)
    self.stabilizer = AdaptiveSignalStabilizer(static_threshold, initial_damping, alpha)
    self.shield = ABCollatz(epsilon_target=1.0)
    self.last_kappa = 0.0
    # Guardia di memoria condivisa (RAM+VRAM), non piu' un check psutil
    # ad-hoc duplicato qui dentro (vedi _gc_se_ram_bassa).
    self._memory_guard = UniversalMemoryGuard(min_free_ram_percentage=min_free_ram_percentage)
    # Banca di riferimenti puliti visti in passato, una deque per shape
    # (le shape diverse non sono comparabili tra loro): quando il chiamante
    # fornisce x_reference, le sue righe vengono ricordate qui; in modalita'
    # cieca (x_reference=None) una richiesta futura con un input corrotto
    # simile a uno gia' visto puo' riusare il riferimento vero invece di
    # ripartire da zero con la sola stima locale (vedi _recall_reference).
    self._reference_bank: dict = {}
    self._reference_memory_size = int(reference_memory_size)
    # Kernel JAX precompilati una sola volta (stesso principio del fix
    # allo scan in AdaptiveSignalStabilizer.filter_data_stream): senza
    # questo, la pipeline eager di _execute_4_phase_*_shield ridispaccia
    # e ricompila ogni singola operazione JAX ad ogni chiamata.
    self._compiled_input_shield_kernel = jax.jit(self._run_input_shield_kernel)
    self._compiled_output_shield_kernel = jax.jit(self._run_output_shield_kernel)
    # margine d'errore (come la covarianza di Kalman, ma definito semplicemente
    # come |valore_ricevuto - valore_corretto|: quanto piu' lo scudo ha dovuto
    # spostare un valore, tanto meno ci si deve fidare del risultato in quel punto)
    self.margine_ingresso = None       # array, stessa forma dell'input: incertezza in SPAZIO INPUT
    self.margine_ingresso_medio = 0.0
    self.margine_ingresso_max = 0.0
    self.margine_uscita = None         # array, stessa forma dell'output: incertezza in SPAZIO OUTPUT
    self.margine_uscita_medio = 0.0
    self.margine_uscita_max = 0.0

    # Etichette/incertezza dell'arbitro (solo se use_arbiter=True in
    # protect_and_forward) -- margine_ingresso dice GIA' quanto e' stato
    # spostato un valore; incertezza_arbitro dice quanto l'arbitro stesso
    # non e' sicuro di aver scelto la classificazione giusta (vicino al
    # confine deviante/pulito o spike/regime), non quanto grande e' stata
    # la correzione. I due segnali sono complementari, non ridondanti:
    # una correzione piccola con incertezza alta e' un caso ambiguo che
    # e' andato bene per caso, non uno davvero sicuro.
    self.etichette_arbitro = None      # array di stringhe 'clean'/'spike'/'regime', stessa forma dell'input
    self.incertezza_arbitro = None     # array 0..1, stessa forma dell'input
    self.incertezza_arbitro_media = 0.0
    # Memoria dei TIPI di corruzione gia' visti per ogni shape (non solo
    # i riferimenti puliti per forma, gia' in _reference_bank) -- un
    # Counter delle etichette dell'arbitro osservate quando x_reference
    # era noto, cosi' una chiamata futura in modalita' cieca sulla stessa
    # shape puo' sapere se quel sensore/flusso ha gia' mostrato spike
    # isolati, cambi di regime, o entrambi, invece di ripartire da zero.
    self._corruption_type_memory: dict = {}

tipi_corruzione_visti

tipi_corruzione_visti(slice_shape: tuple) -> Counter

Quali etichette dell'arbitro ('clean'/'spike'/'regime') sono gia' state osservate per questa shape, e quante volte -- Counter vuoto se questa shape non e' mai stata vista con use_arbiter=True e x_reference noto (l'unico modo in cui la memoria viene popolata, vedi _aggiorna_memoria_corruzione).

Source code in dense_armor/utility/protect/orca.py

def tipi_corruzione_visti(self, slice_shape: tuple) -> Counter:
    """Quali etichette dell'arbitro ('clean'/'spike'/'regime') sono gia'
    state osservate per questa shape, e quante volte -- Counter vuoto se
    questa shape non e' mai stata vista con use_arbiter=True e
    x_reference noto (l'unico modo in cui la memoria viene popolata,
    vedi _aggiorna_memoria_corruzione)."""
    return self._corruption_type_memory.get(slice_shape, Counter())

protect_and_forward

protect_and_forward(
    ai_model_callable: Optional[Callable],
    x_corrupted: ndarray,
    x_reference: Optional[ndarray] = None,
    use_input_shield: bool = True,
    use_model_injection: bool = True,
    use_output_shield: bool = True,
    use_arbiter: bool = False,
) -> np.ndarray

Esegue le 4 fasi (scudo entrata -> modello -> scudo uscita) e ritorna l'output protetto.

use_arbiter — se True, dopo lo scudo entrata standard instrada ogni punto verso il correttore giusto (utility/arbiter.py): impulso isolato -> rigetto duro, cambio di regime sostenuto -> passa grezzo, nessuna anomalia puntuale -> resta il risultato gia' prodotto dallo scudo standard (il suo smorzamento morbido, non un pass-through). Default False: comportamento e risultati identici a prima di questa opzione su tutti i test esistenti. Popola etichette_arbitro/incertezza_arbitro; se x_reference e' noto, aggiorna anche tipi_corruzione_visti(slice_shape).

Source code in dense_armor/utility/protect/orca.py

def protect_and_forward(
    self,
    ai_model_callable: Optional[Callable],
    x_corrupted: np.ndarray,
    x_reference: Optional[np.ndarray] = None,
    use_input_shield: bool = True,
    use_model_injection: bool = True,
    use_output_shield: bool = True,
    use_arbiter: bool = False,
) -> np.ndarray:
    """Esegue le 4 fasi (scudo entrata -> modello -> scudo uscita) e ritorna l'output protetto.

    use_arbiter — se True, dopo lo scudo entrata standard instrada ogni
    punto verso il correttore giusto (utility/arbiter.py): impulso
    isolato -> rigetto duro, cambio di regime sostenuto -> passa
    grezzo, nessuna anomalia puntuale -> resta il risultato gia'
    prodotto dallo scudo standard (il suo smorzamento morbido, non un
    pass-through). Default False: comportamento e risultati identici a
    prima di questa opzione su tutti i test esistenti. Popola
    etichette_arbitro/incertezza_arbitro; se x_reference e' noto,
    aggiorna anche tipi_corruzione_visti(slice_shape)."""
    is_simple_data_test = ai_model_callable is None or not use_model_injection
    if is_simple_data_test:
        logger.info("CONTRAZIONE LOGICA DETECTED: Riconosciuto Test di Protezione Dati Semplice (No IA Model).")
        use_model_injection = False

    # Un array 1D (es. una singola serie da sensore/pipeline) NON e' un
    # batch di N scalari indipendenti: e' UNA istanza con N campioni
    # correlati nel tempo. Senza questa promozione, B=N e ogni campione
    # veniva processato da solo (slice_shape=()), azzerando il contesto
    # su cui si basa il rilevamento outlier in modalita' cieca (senza
    # x_reference) -- il caso d'uso principale documentato nel README.
    was_1d = (x_corrupted.ndim == 1)
    if was_1d:
        x_corrupted = np.asarray(x_corrupted).reshape(1, -1)
        if x_reference is not None:
            x_reference = np.asarray(x_reference).reshape(1, -1)

    orig_shape = x_corrupted.shape
    B = orig_shape[0]
    slice_shape = orig_shape[1:]
    t_start = time.time()

    if use_input_shield:
        logger.info("Attivazione SCUDO ENTRATA (4 Fasi) su Ipervolume: %s", orig_shape)
        x_corrupted_np = np.array(x_corrupted)
        if x_reference is None:
            x_reference_np = np.zeros_like(x_corrupted_np)
            for b in range(B):
                row_flat = x_corrupted_np[b].flatten()
                recalled = self._recall_reference(row_flat, slice_shape)
                if recalled is not None:
                    x_reference_np[b] = recalled.reshape(slice_shape)
                else:
                    x_reference_np[b] = self._blind_reference(row_flat).reshape(slice_shape)
        else:
            x_reference_np = np.array(x_reference)
            self._remember_reference(x_reference_np, slice_shape)
        purified_batch = np.zeros(orig_shape, dtype=np.float64)
        margine_batch = np.zeros(orig_shape, dtype=np.float64)
        flat_cl_all = x_reference_np.reshape(B, -1)
        flat_co_all = x_corrupted_np.reshape(B, -1)
        total_elements = flat_cl_all.shape[1]
        if total_elements <= self.chunk_threshold:
            # Percorso vettorizzato: le B righe sono processate in UNA
            # chiamata (jax.vmap sotto, calibrazione indipendente per
            # riga) invece di B dispatch JAX separati -- vedi
            # _execute_4_phase_input_shield_batch e CHANGELOG per il
            # guadagno misurato. Ogni riga e' completamente indipendente
            # dalle altre a questo stadio (nessuno stato condiviso tra
            # righe qui: _reference_bank/_corruption_type_memory non
            # sono toccati in questo blocco).
            out_batch, marg_batch = self._execute_4_phase_input_shield_batch(flat_cl_all, flat_co_all)
            purified_batch = out_batch.reshape(orig_shape)
            margine_batch = marg_batch.reshape(orig_shape)
            self._gc_se_ram_bassa()
        else:
            # Fallback invariato: una singola riga eccede da sola
            # chunk_threshold (raro), va comunque spezzata in blocchi.
            for b in range(B):
                flat_cl, flat_co = flat_cl_all[b], flat_co_all[b]
                out_flat = np.zeros_like(flat_cl)
                margine_flat = np.zeros_like(flat_cl)
                c_size = self.chunk_threshold
                for start_idx in range(0, total_elements, c_size):
                    end_idx = min(start_idx + c_size, total_elements)
                    purified_chunk, margine_chunk = self._execute_4_phase_input_shield(flat_cl[start_idx:end_idx], flat_co[start_idx:end_idx])
                    out_flat[start_idx:end_idx] = purified_chunk
                    margine_flat[start_idx:end_idx] = margine_chunk
                    self._gc_se_ram_bassa()
                purified_batch[b] = out_flat.reshape(slice_shape)
                margine_batch[b] = margine_flat.reshape(slice_shape)

        if use_arbiter:
            etichette_batch = np.empty(orig_shape, dtype=object)
            incertezza_batch = np.zeros(orig_shape, dtype=np.float64)
            for b in range(B):
                co_flat = x_corrupted_np[b].flatten()
                pur_flat = purified_batch[b].flatten()
                corretto_flat, etichette_flat, incertezza_flat = self._applica_arbitro(co_flat, pur_flat)
                purified_batch[b] = corretto_flat.reshape(slice_shape)
                etichette_batch[b] = etichette_flat.reshape(slice_shape)
                incertezza_batch[b] = incertezza_flat.reshape(slice_shape)
                if x_reference is not None:
                    self._aggiorna_memoria_corruzione(etichette_flat, slice_shape)
            self.etichette_arbitro = etichette_batch
            self.incertezza_arbitro = incertezza_batch
            self.incertezza_arbitro_media = float(np.mean(incertezza_batch))

        x_for_model = jnp.array(purified_batch)
        self.margine_ingresso = margine_batch
        self.margine_ingresso_medio = float(np.mean(margine_batch))
        self.margine_ingresso_max = float(np.max(margine_batch))
        logger.info("Input purificato in %.3fs. Margine d'errore: medio=%.4g, max=%.4g",
                    time.time() - t_start, self.margine_ingresso_medio, self.margine_ingresso_max)
    else:
        logger.info("SCUDO ENTRATA disattivato. I dati transitano senza pre-filtri.")
        x_for_model = jnp.array(x_corrupted)
        self.margine_ingresso = None
        self.margine_ingresso_medio = self.margine_ingresso_max = 0.0

    if use_model_injection:
        t_ia = time.time()
        ai_output = ai_model_callable(x_for_model)
        jax.block_until_ready(ai_output)
        logger.info("Risposta IA ottenuta in %.3fs. Shape Output: %s", time.time() - t_ia, ai_output.shape)
    else:
        logger.info("INIEZIONE MODELLO bypassata. I dati purificati procedono verso la barriera spettrale.")
        ai_output = x_for_model

    if use_output_shield:
        t_out = time.time()
        logger.info("Attivazione SCUDO USCITA (4 Fasi) su Spettro Terminale...")
        if use_model_injection and x_reference is not None:
            # riferimento nello SPAZIO DI OUTPUT: risposta del modello al dato
            # pulito, non l'input purificato (spazio diverso se il modello e'
            # trasformativo, es. classificatori/embedding/reti non-lineari)
            output_reference = ai_model_callable(jnp.array(x_reference))
            jax.block_until_ready(output_reference)
        else:
            # nessun riferimento pulito noto: auto-consistenza cieca sull'output
            # stesso (stesso stabilizzatore della Fase 1, applicato qui all'uscita)
            output_reference = self.stabilizer.filter_batch_scenarios(
                ai_output.reshape(ai_output.shape[0], -1)).reshape(ai_output.shape)
        x_final, margine_out = self._execute_4_phase_output_shield(ai_output, output_reference)
        self.margine_uscita = np.array(margine_out)
        self.margine_uscita_medio = float(jnp.mean(margine_out))
        self.margine_uscita_max = float(jnp.max(margine_out))
        logger.info("Output rinormalizzato in %.3fs. Margine d'errore: medio=%.4g, max=%.4g",
                    time.time() - t_out, self.margine_uscita_medio, self.margine_uscita_max)
    else:
        logger.info("SCUDO USCITA disattivato. Emissione del flusso lineare.")
        x_final = ai_output
        self.margine_uscita = None
        self.margine_uscita_medio = self.margine_uscita_max = 0.0

    if was_1d:
        x_final = x_final.reshape(-1)
        if self.margine_ingresso is not None:
            self.margine_ingresso = self.margine_ingresso.reshape(-1)
        if self.margine_uscita is not None:
            self.margine_uscita = np.asarray(self.margine_uscita).reshape(-1)
        if self.etichette_arbitro is not None:
            self.etichette_arbitro = self.etichette_arbitro.reshape(-1)
        if self.incertezza_arbitro is not None:
            self.incertezza_arbitro = self.incertezza_arbitro.reshape(-1)

    logger.info("Transito concluso. Sistema sigillato in %.3f secondi totali.", time.time() - t_start)
    return x_final

---

See also: Adaptive engine — AdaptiveSignalStabilizer, Orca's Stage 1. Arbiter — the per-point classifier used by use_arbiter=True. Armatura — if you have a bare series rather than a model.