Building Governance-First Physical AI

La majorité des travaux de sécurité de l'IA ciblent le modèle. L'IA physique exige davantage : des invariants de sécurité qui tiennent au niveau du protocole, appliqués avant que l'actionneur ne bouge. Ce tutoriel détaille la séquence de construction en six commits qui a produit un démonstrateur gouvernance-first fonctionnel sur l'Arduino VENTUNO Q, une carte à double cerveau associant un NPU Qualcomm Dragonwing IQ8 (40 TOPS, Ubuntu) à un coprocesseur temps réel STM32H5 (Zephyr OS + Arduino Core).

Le code source complet est disponible sur github.com/thierrysays/governed-edge-ai. Chaque étape ci-dessous correspond à un commit précis. Ce tutoriel est le complément technique de « When AI Controls Physical Systems: Governance Must Be a Hardware Invariant, Not a Policy Document » et de la couche 5 de l'ECM™ (Gouvernance IA), au sein de l'Enterprise Coherence Model.

＃L'architecture en une phrase

Le NPU exécute la perception et la gouvernance sous Linux. Le STM32H5 exécute l'actionnement temps réel. Entre les deux : un protocole IPC binaire vérifié sur UART, où le STM32H5 n'exécute aucune commande dépourvue d'une référence d'audit valide.

1Codec IPC, le langage commun entre les deux processeurs

Avant d'écrire la moindre ligne de code de perception ou de gouvernance, il faut définir le contrat de données entre les deux processeurs. Le codec IPC gère huit types de messages :

Chaque trame utilise un CRC-16/CCITT pour l'intégrité. La classe FrameParser gère la lecture incrémentale du flux, un point critique lorsqu'on lit depuis un UART où les trames peuvent arriver sur plusieurs appels read().

# Encode a governance-approved command
from ipc.codec import ActionType, Actor, CommandRequest, encode

frame = encode(CommandRequest(
    audit_ref=42,           # non-zero: log entry confirmed to exist
    timestamp_us=12345678,  # session-relative, wraps at uint32 max
    actor=Actor.AI,
    confidence=0.91,        # float32 on the wire
    action_type=ActionType.HALT,
    action_param=0,
))

2Simulateur STM32H5, testé sans matériel

Les tests avec matériel dans la boucle sont lents et fragiles. La classe MockSTM32H5 crée un pty Unix (pseudo-terminal) qui se comporte comme le vrai coprocesseur : elle décode les trames CommandRequest, applique son propre seuil de confiance, suit son état (ARMED/BUSY/HALTED/FAULT) et répond par CommandAck ou CommandReject.

from ipc.mock_peer import MockSTM32H5

with MockSTM32H5(watchdog_ms=10_000.0) as peer:
    ch = open(peer.device, "rb+", buffering=0)  # open the pty
    # ch is now a binary r/w channel identical to a real UART

Le simulateur applique le double seuil de confiance : son seuil est fixé à 0.70 en float32. Lorsque Linux envoie confidence=0.70 (float64), l'encodage sur le fil tronque la valeur légèrement en dessous de 0.70 en float32, et le simulateur la rejette. C'est un comportement voulu : la défense en profondeur s'applique jusqu'au niveau de la représentation des données.

# Trigger the hardware kill switch for testing
peer.trigger_kill_switch()
# All subsequent CommandRequests will receive CommandReject

3Journal d'audit, SQLite en mode WAL et ajout seul

Le journal d'audit est le registre de gouvernance. Il doit être :

• Écrit avant toute transmission de commande (journaliser avant d'agir)
• En ajout seul en fonctionnement normal
• Interrogeable à des fins d'analyse forensique

from logger import AuditEvent, AuditLogger

with AuditLogger("audit.db") as logger:
    session_id = logger.open_session(board_serial="VENTUNO-001")

    # Returns the SQLite rowid (≥ 1); this becomes the audit_ref
    audit_ref = logger.log_event(AuditEvent(
        session_id=session_id,
        actor="ai",
        detection_type="object",
        detection_label="person",
        confidence=0.91,
        command="HALT",
        command_sent=True,
        stm32_ack=None,         # filled in after MCU responds
    ))

    # After the MCU responds:
    logger.update_stm32_ack(audit_ref, ack=True)

Le schéma enregistre : horodatage, session, acteur, type/étiquette de détection, confiance, commande, indicateur d'envoi, et l'acquittement du MCU. La colonne stm32_ack reste NULL tant que le MCU n'a pas répondu. Si le système plante entre log_event et update_stm32_ack, ce NULL a une signification forensique précise : une commande a été émise sans confirmation de réponse.

4Chaîne de perception, détections typées et briques interchangeables

La couche de perception est volontairement minimale. Son rôle est de produire des objets DetectionResult typés que le filtre de gouvernance peut évaluer sans savoir quel backend les a générés.

from perception.base import DetectionResult

# Frozen dataclass, immutable once created
result = DetectionResult(
    detection_type="object",    # object | gesture | pose
    label="person",
    confidence=0.91,            # clamped to [0.0, 1.0]
)

# The governance filter calls this:
result.passes_threshold(0.70)  # True

Trois backends factices sont livrés avec le démonstrateur :

from perception.backends import StubObjectDetector, StubGestureRecognizer, StubPoseEstimator

detections = (
    StubObjectDetector(confidence=0.91).run(None)   # → person @ 0.91
    + StubGestureRecognizer(confidence=0.88).run(None)  # → thumbs_up @ 0.88
    + StubPoseEstimator(confidence=0.76).run(None)  # → proximity_breach @ 0.76
)

N'importe quel backend factice peut être remplacé par un backend réel (YOLO, MediaPipe, un modèle NPU sur mesure) sans toucher à la couche de gouvernance. La classe abstraite impose le contrat run(frame) -> list[DetectionResult].

5Tableau de bord, visibilité de l'audit en temps réel

Un tableau de bord Flask/SQLAlchemy interroge le journal d'audit SQLite en temps réel. Vues principales :

• Liste des sessions : toutes les sessions avec numéro de série de la carte, heure de démarrage, nombre d'événements
• Flux d'événements : journal chronologique avec étiquette de détection, confiance, commande, envoyé/supprimé, ACK/REJECT
• Taux de suppression : détections sous le seuil rapportées au total, par session

Le tableau de bord est en lecture seule par conception. Il ne peut pas modifier les entrées d'audit. La séparation des chemins de lecture et d'écriture est délibérée : le filtre de gouvernance écrit via AuditLogger ; tout le reste se contente de lire.

6Filtre de gouvernance, le verrou de sécurité

C'est le composant qui referme la boucle. Le GovernanceFilter se situe entre la chaîne de perception et le canal IPC. Son contrat est simple et non négociable :

Contrat de gouvernance, six invariants

1. Journaliser avant d'agir : audit_ref est obtenu avant toute transmission de trame
2. Pas de journal, pas de commande : si log_event() lève une exception, celle-ci se propage et aucune trame n'est envoyée
3. Seuil de confiance (côté Linux) : les détections sous le seuil sont journalisées avec command_sent=False
4. Une commande par trame : la détection de plus haute confiance franchissant le seuil est retenue ; toutes les autres sont supprimées
5. Double verrou : le verrou Linux et le verrou STM32H5 fonctionnent indépendamment
6. Suivi ACK/REJECT : update_stm32_ack() est appelé exactement une fois par commande transmise ; un délai dépassé laisse stm32_ack à NULL

from governance.filter import GovernanceFilter

gov = GovernanceFilter(
    logger=audit_logger,
    session_id=session_id,
    channel=uart_channel,          # real serial port or MockSTM32H5 pty
    confidence_threshold=0.70,
    response_timeout_s=0.5,
)

# Each frame from the perception pipeline:
gov.process_frame(detections)

L'implémentation de process_frame trie les détections par confiance décroissante, journalise chaque détection, et n'envoie une CommandRequest que pour la détection de plus haute confiance qui franchit le seuil :

def process_frame(self, detections: list[DetectionResult]) -> None:
    if not detections:
        return

    by_confidence = sorted(detections, key=lambda d: d.confidence, reverse=True)
    command_sent_this_frame = False

    for detection in by_confidence:
        action_type, action_param = self._command_map.get(detection.label, _DEFAULT_ACTION)
        should_send = (
            not command_sent_this_frame
            and detection.passes_threshold(self._threshold)
        )

        # Log-before-act: audit_ref is non-zero or log_event() raises
        audit_ref = self._logger.log_event(AuditEvent(
            session_id=self._session_id,
            actor="ai",
            detection_type=detection.detection_type,
            detection_label=detection.label,
            confidence=detection.confidence,
            command=action_type.name,
            command_sent=should_send,
            stm32_ack=None,
        ))

        if should_send:
            command_sent_this_frame = True
            ack = self._send_command(detection, action_type, action_param, audit_ref)
            if ack is not None:
                self._logger.update_stm32_ack(audit_ref, ack)

Table de commandes par défaut · conservateur par construction : toute étiquette inconnue déclenche HALT par défaut

✓Couverture de tests

La suite de tests complète couvre 184 tests répartis sur les deux modules, avec 97 % de couverture de lignes :

linux-stack/tests/
  conftest.py                  # sys.path wiring for audit-service
  test_smoke_governance.py     # 7 smoke tests (fast, run first in CI)
  test_governance.py           # 36 unit tests across 7 classes

audit-service/tests/
  test_logger.py               # AuditLogger unit tests
  test_dashboard.py            # Dashboard integration tests

Quelques schémas de test à retenir :

Tester le chemin de délai dépassé sans matériel : on utilise un tube (pipe) du système. L'extrémité en écriture permet au filtre de gouvernance d'envoyer des octets ; l'extrémité en lecture n'est jamais ouverte, si bien que _read_response expire et renvoie None :

import os
rfd, wfd = os.pipe()
wfile = open(wfd, "wb", buffering=0)

gov = GovernanceFilter(
    logger=audit_logger, session_id=session_id,
    channel=wfile, response_timeout_s=0.05,  # short for test speed
)
gov.process_frame([det(confidence=0.91)])
# stm32_ack remains NULL (timeout)
os.close(rfd); wfile.close()

Tester le chemin d'arrêt d'urgence :

peer.trigger_kill_switch()
time.sleep(0.05)  # wait for mock peer state change
gov.process_frame([det(confidence=0.91)])
# command_sent=True (Linux passed it), stm32_ack=False (MCU rejected)

!Ce que cette architecture ne résout pas (encore)

Ce démonstrateur prouve une posture de gouvernance, ce n'est pas un déploiement en production. Trois lacunes subsistent :

▸L'exécuter soi-même

# Clone and install
git clone https://github.com/thierrysays/governed-edge-ai
cd governed-edge-ai

# Run the full test suite
make test

# Run smoke tests only (fast, hardware-free)
make smoke

# Run lint + typecheck + security scan + tests
make qa

Tous les tests s'exécutent sans matériel. MockSTM32H5 fournit une simulation du coprocesseur basée sur un pty, incluant le seuil de confiance, l'arrêt d'urgence et le watchdog.