ROS2 Integration

Why this bridge exists

Kyvern produces cryptographically signed Decision objects tied to a tamper-evident hash chain. This bridge publishes each Decision to a ROS2 topic so any ROS2 node — controller, logger, HMI — can consume decisions natively without coupling to the Kyvern Python API.

Quick start

Prerequisites: ROS2 Humble installed and sourced (source /opt/ros/humble/setup.bash).

Publish decisions:

from services.integrations.ros2_bridge import KyvernDecisionPublisher

pub = KyvernDecisionPublisher(topic="/kyvern/decisions")
pub.start()
pub.publish(signed_decision)   # signed_decision is a dict from sign_decision()
pub.stop()

Subscribe and verify:

from services.integrations.ros2_subscriber_example import KyvernDecisionVerifier

verifier = KyvernDecisionVerifier(public_key_path="/tmp/kyvern-demo/signing.pub")
verifier.start()   # blocks; Ctrl-C to stop

Or run the subscriber standalone:

python -m services.integrations.ros2_subscriber_example \
    --pubkey /tmp/kyvern-demo/signing.pub

Message format

Topic type: std_msgs/String

Payload: canonical JSON produced by decision_to_ros2_json(decision). Field order is alphabetical (sort_keys=True) for determinism.

Key fields:

Field	Type	Description
action	string	Decision action (allow / alert / halt / handoff / engage)
chain_index	int	Position in the hash chain
payload_hash	string	SHA-256 of canonical payload (hex)
prev_hash	string | null	Hash of preceding decision; null for chain head
policy_version_id	string	SHA-256 of the policy file at decision time
roe_reference	string	Rule ID that fired
signature	string	Base64 Ed25519 signature
timestamp_iso	string	ISO 8601 UTC timestamp

Why not a custom message type? A custom kyvern_msgs/Decision.msg would require every subscriber to build and source the kyvern_msgs package. std_msgs/String with JSON means a Python, C++, or Rust subscriber can parse the payload with a single json.loads call, with no build-time Kyvern dependency.

QoS recommendations

Scenario	Setting
Safety-critical decisions (halt, handoff, engage)	qos_reliability="reliable" (default)
High-frequency telemetry / allow decisions	qos_reliability="best_effort"

pub = KyvernDecisionPublisher(qos_reliability="best_effort")

End-to-end test (WSL + ROS2 Humble)

First generate demo files:

python scripts/generate_demo_chain.py

Terminal 1 — subscriber:

source /opt/ros/humble/setup.bash
python -m services.integrations.ros2_subscriber_example \
    --pubkey /tmp/kyvern-demo/signing.pub

Terminal 2 — publish one decision from the demo chain:

source /opt/ros/humble/setup.bash
python -c "
import json
from services.integrations.ros2_bridge import KyvernDecisionPublisher

chain = [json.loads(l) for l in open('/tmp/kyvern-demo/chain.jsonl')]
pub = KyvernDecisionPublisher()
pub.start()
pub.publish(chain[0])
pub.stop()
"

Terminal 1 should log: [VERIFIED] chain_index=0

Recording a robot's own decisions

The bridge publishes decisions Kyvern's engine made. To record the decisions your robot's own controller makes, call kyvern.record_decision() from your node; examples/ros2_safety_demo does this for a LaserScan-based safety controller.