A Technical Deep Dive into VCP v1.1 Applied to Automotive Safety
Abstract
When an autonomous vehicle is involved in an accident, investigators face a critical question: "What did the AI see, what did it decide, and why?" Traditional Event Data Recorders (EDRs) capture physical state—speed, braking, steering—but not the AI's decision-making process. The AI remains a black box.
DVP (Driving Vehicle Protocol) fills this gap. Building on the production-ready VeritasChain Protocol v1.1 architecture, DVP provides a cryptographically verifiable audit trail for autonomous driving decisions—from sensor input to control output.
This article presents a technical deep dive into DVP's architecture, showing how VCP v1.1's three-layer integrity model, Merkle tree anchoring, and sidecar pattern translate to the automotive domain.
Keywords: Autonomous Vehicles, AI Safety, Event Data Recorder, Cryptographic Audit Trail, UNECE WP.29, EU AI Act, Merkle Tree, VCP v1.1
Table of Contents
- The Black Box Problem in Autonomous Driving
- From VCP v1.1 to DVP: Architecture Translation
- DVP Three-Layer Architecture
- Event Schema Design
- Sensor-to-Decision Pipeline Logging
- Merkle Tree and External Anchoring
- DVP-XREF: Multi-System Cross-Reference
- EDR Integration Architecture
- Reference Implementation
- Regulatory Compliance Mapping
- Conclusion
1. The Black Box Problem
1.1 Real-World Investigation Challenges
Consider these actual scenarios:
2018, Tempe, Arizona: An Uber autonomous vehicle struck and killed a pedestrian. Investigators had to reconstruct what the AI "saw" from fragmentary logs. The critical question—whether the perception system classified the victim correctly—took months to answer.
2016, Florida: A Tesla operating in Autopilot mode collided with a truck. The AI's decision to not engage emergency braking became the subject of intense scrutiny, but the complete decision chain was unavailable.
In both cases, investigators faced the same fundamental problem:
┌─────────────────────────────────────────────────────────────┐
│ THE INVESTIGATION GAP │
├─────────────────────────────────────────────────────────────┤
│ │
│ WHAT EDR CAPTURES: WHAT'S MISSING: │
│ ───────────────── ────────────── │
│ ✓ Speed at impact ✗ Object detection results │
│ ✓ Brake application time ✗ Classification confidence│
│ ✓ Steering angle ✗ Decision rationale │
│ ✓ Seatbelt status ✗ Path planning candidates │
│ ✓ Airbag deployment ✗ Risk assessment scores │
│ ✗ Why action X vs action Y │
│ │
│ Physical State AI Cognition │
│ (Recorded) (Black Box) │
│ │
└─────────────────────────────────────────────────────────────┘
1.2 The Aviation Analogy
Aviation solved this problem decades ago. After catastrophic accidents where investigators couldn't determine cause, regulators mandated Flight Data Recorders (FDRs)—tamper-resistant devices that capture every flight parameter.
The result: Aviation became the safest form of transportation, with continuous safety improvements driven by incident analysis.
Autonomous vehicles need the equivalent—not just for physical state, but for AI decision-making.
DVP is that equivalent.
2. From VCP v1.1 to DVP: Architecture Translation
2.1 VCP v1.1 Core Innovations
VCP (VeritasChain Protocol) v1.1 introduced several critical improvements for audit trail integrity:
| VCP v1.1 Feature | Purpose | DVP Application |
|---|---|---|
| Three-Layer Architecture | Separates event generation, integrity, and anchoring | Maps to Sensor/Perception/Planning/Control layers |
| External Anchor REQUIRED | Third-party verifiability for all tiers | Independent timestamp for accident reconstruction |
| Policy Identification | Declares conformance tier and verification depth | Vehicle-specific safety policy declaration |
| VCP-XREF Dual Logging | Cross-reference between parties | Multi-ECU and V2X cross-verification |
| Completeness Guarantees | Proves no events were omitted | Critical for proving no log tampering |
2.2 Domain Translation Matrix
| VCP v1.1 Concept | Financial Domain | Automotive Domain (DVP) |
|---|---|---|
| Event Producer | Trading Algorithm | Autonomous Driving Stack |
| Event Types | SIG, ORD, ACK, EXE | PERCEPTION, PLANNING, CONTROL |
| Timing Requirements | Microsecond (HFT) | Millisecond (real-time control) |
| External Anchor Target | Blockchain, TSA | TSA + Dedicated Safety Authority |
| Regulatory Framework | MiFID II, SEC | UNECE WP.29, EU AI Act |
| Counterparty | Broker | V2X Infrastructure, Other Vehicles |
2.3 Compliance Tier Mapping
VCP v1.1's three tiers map to automotive contexts:
| VCP Tier | Automotive Equivalent | Use Case |
|---|---|---|
| Platinum | Safety-Critical AV | SAE Level 4-5 autonomous vehicles |
| Gold | ADAS / Level 2-3 | Advanced driver assistance systems |
| Silver | Development/Testing | Simulation, closed-course testing |
3. DVP Three-Layer Architecture
3.1 Architecture Overview
DVP adopts VCP v1.1's three-layer model, adapted for automotive requirements:
┌─────────────────────────────────────────────────────────────────────────┐
│ DVP THREE-LAYER ARCHITECTURE │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ LAYER 3: EXTERNAL ANCHORING │
│ ───────────────────────────── │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ RFC 3161 TSA │ Transparency Log │ Safety Authority Portal │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ ▲ │
│ │ Merkle Root (every 10 seconds) │
│ │
│ LAYER 2: LOCAL INTEGRITY │
│ ──────────────────────── │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ MERKLE TREE BUILDER │ │
│ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │
│ │ │Event 1 │ │Event 2 │ │Event 3 │ │Event 4 │ │Event 5 │ │ │
│ │ │Percept. │ │Planning │ │Control │ │Sensor │ │Percept. │ │ │
│ │ └────┬────┘ └────┬────┘ └────┬────┘ └────┬────┘ └────┬────┘ │ │
│ │ └─────┬─────┘ └─────┬─────┘ │ │ │
│ │ └───────────┬───────────┘ │ │ │
│ │ └───────────────┬─────────────┘ │ │
│ │ ▼ │ │
│ │ [MERKLE ROOT] │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ ▲ │
│ │ │
│ LAYER 1: EVENT GENERATION │
│ ──────────────────────── │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ DVP SIDECAR │ │
│ │ │ │
│ │ [Event Capture] → [Schema Validate] → [Hash] → [Sign] → [Queue]│ │
│ │ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ ▲ │
│ │ │
│ ┌──────────────────────────────────────────────────────────────────┐ │
│ │ AUTONOMOUS DRIVING STACK │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │
│ │ │ Sensors │→ │Perception│→ │ Planning │→ │ Control │ │ │
│ │ │LiDAR/Cam │ │Detection │ │Path/Speed│ │Actuators │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │ │
│ └──────────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────┘
3.2 Layer Responsibilities
| Layer | Function | DVP-Specific Requirements |
|---|---|---|
| L1: Event Generation | Capture AD stack decisions, hash, sign | <10ms latency, 1000+ events/sec |
| L2: Local Integrity | Merkle tree construction, batching | In-vehicle tamper-evident storage |
| L3: External Anchor | Third-party timestamp, completeness proof | Crash-survivable, post-incident retrieval |
3.3 Why Three Layers Matter for Automotive
Layer 1 alone is insufficient:
- OEM-controlled signing keys could be compromised
- Logs could be selectively deleted before anchoring
Layer 2 alone is insufficient:
- Merkle roots without external anchoring prove nothing to third parties
- OEM could regenerate entire tree with modified events
Layer 3 completes the guarantee:
- External timestamps prove log existence at specific time
- Independent authority prevents OEM manipulation
- Completeness proofs show no events were omitted
4. Event Schema Design
4.1 DVP Event Type Registry
DVP defines event types across the autonomous driving pipeline:
enum DVPEventType {
// Sensor Layer
SENSOR_FRAME_CAPTURED = 'SENSOR_FRAME', // Raw sensor data hash
SENSOR_CALIBRATION = 'SENSOR_CAL', // Calibration status change
SENSOR_FAULT = 'SENSOR_FAULT', // Sensor malfunction
// Perception Layer
PERCEPTION_DETECTION = 'PERCEPT_DETECT', // Object detection result
PERCEPTION_CLASSIFICATION = 'PERCEPT_CLASS', // Object classification
PERCEPTION_TRACKING = 'PERCEPT_TRACK', // Object tracking update
PERCEPTION_FUSION = 'PERCEPT_FUSION', // Multi-sensor fusion result
// Planning Layer
PLANNING_PATH_GENERATED = 'PLAN_PATH', // Path planning decision
PLANNING_SPEED_PROFILE = 'PLAN_SPEED', // Speed planning
PLANNING_MANEUVER = 'PLAN_MANEUVER', // Maneuver decision (lane change, etc.)
PLANNING_RISK_ASSESSMENT = 'PLAN_RISK', // Risk score calculation
// Control Layer
CONTROL_COMMAND = 'CTRL_CMD', // Actuator command issued
CONTROL_EXECUTED = 'CTRL_EXEC', // Actuator feedback
CONTROL_OVERRIDE = 'CTRL_OVERRIDE', // Human takeover
CONTROL_EMERGENCY = 'CTRL_EMERGENCY', // Emergency intervention
// System Events
SYSTEM_MODE_CHANGE = 'SYS_MODE', // Autonomy level change
SYSTEM_FAULT = 'SYS_FAULT', // System-level fault
SYSTEM_HEARTBEAT = 'SYS_HEARTBEAT', // Periodic health check
// V2X Events
V2X_MESSAGE_RECEIVED = 'V2X_RX', // V2X message received
V2X_MESSAGE_SENT = 'V2X_TX' // V2X message transmitted
}
4.2 Core Event Schema
Following VCP v1.1's schema design:
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://veritaschain.org/schemas/dvp/v0.1/event.json",
"title": "DVP Event",
"type": "object",
"required": ["Header", "Provenance", "Integrity"],
"properties": {
"Header": {
"type": "object",
"required": ["EventID", "EventType", "Timestamp", "VehicleID"],
"properties": {
"EventID": {
"type": "string",
"format": "uuid",
"description": "UUID v7 (time-ordered)"
},
"EventType": {
"type": "string",
"enum": ["SENSOR_FRAME", "PERCEPT_DETECT", "PLAN_PATH", "CTRL_CMD", "..."]
},
"Timestamp": {
"type": "object",
"properties": {
"UnixNs": { "type": "integer" },
"ISO8601": { "type": "string", "format": "date-time" },
"Precision": { "type": "string", "enum": ["NANOSECOND", "MICROSECOND", "MILLISECOND"] },
"SyncStatus": { "type": "string", "enum": ["PTP_LOCKED", "GNSS_SYNCED", "NTP_SYNCED", "UNSYNCHRONIZED"] }
}
},
"VehicleID": {
"type": "string",
"description": "Vehicle Identification Number (VIN)"
},
"SequenceNumber": {
"type": "integer",
"description": "Monotonically increasing per-vehicle sequence"
}
}
},
"PolicyIdentification": {
"type": "object",
"description": "VCP v1.1 Policy Identification (REQUIRED)",
"properties": {
"Version": { "type": "string", "const": "1.1" },
"PolicyID": { "type": "string" },
"ConformanceTier": { "type": "string", "enum": ["SILVER", "GOLD", "PLATINUM"] },
"RegistrationPolicy": {
"type": "object",
"properties": {
"Issuer": { "type": "string" },
"PolicyURI": { "type": "string", "format": "uri" }
}
}
}
},
"Provenance": {
"type": "object",
"properties": {
"Actor": {
"type": "object",
"properties": {
"Type": { "type": "string", "enum": ["AI_MODEL", "ECU", "SENSOR", "HUMAN", "V2X"] },
"Identifier": { "type": "string" },
"ModelVersion": { "type": "string" },
"ModelHash": { "type": "string" }
}
},
"Input": {
"type": "object",
"properties": {
"SensorFrameIDs": { "type": "array", "items": { "type": "string" } },
"InputDataHash": { "type": "string" },
"DependentEventIDs": { "type": "array", "items": { "type": "string" } }
}
},
"Context": {
"type": "object",
"properties": {
"AutonomyLevel": { "type": "string", "enum": ["L0", "L1", "L2", "L3", "L4", "L5"] },
"OperationalDomain": { "type": "string" },
"SpeedKmh": { "type": "number" },
"Weather": { "type": "string" },
"Visibility": { "type": "number" }
}
},
"Action": {
"type": "object",
"description": "Domain-specific action details"
}
}
},
"Integrity": {
"type": "object",
"required": ["EventHash", "Signature"],
"properties": {
"PrevHash": { "type": "string", "description": "OPTIONAL in v1.1" },
"EventHash": { "type": "string" },
"Signature": { "type": "string" },
"SignerPublicKey": { "type": "string" },
"SignAlgo": { "type": "string", "enum": ["ED25519", "DILITHIUM2"] }
}
}
}
}
4.3 Domain-Specific Action Schemas
Perception Detection Event
{
"Action": {
"DetectedObjects": [
{
"ObjectID": "obj_001",
"Class": "PEDESTRIAN",
"Confidence": 0.94,
"BoundingBox": {
"X": 120.5, "Y": 45.2, "Z": 0.0,
"Width": 0.6, "Height": 1.8, "Depth": 0.4
},
"Velocity": { "Vx": -1.2, "Vy": 0.3, "Vz": 0.0 },
"TrackingID": "track_789",
"TimeToCollision": 2.3,
"ThreatLevel": "HIGH"
}
],
"ProcessingTimeMs": 12.5,
"ModelInferenceDetails": {
"ModelName": "perception_v3.2.1",
"InputResolution": "1920x1080",
"QuantizationLevel": "INT8"
}
}
}
Planning Decision Event
{
"Action": {
"Decision": "EMERGENCY_BRAKE",
"DecisionConfidence": 0.98,
"Trigger": {
"TriggerType": "COLLISION_IMMINENT",
"TriggerObjectID": "obj_001",
"TimeToCollision": 1.2
},
"PathCandidates": [
{
"PathID": "path_001",
"Maneuver": "BRAKE_ONLY",
"RiskScore": 0.15,
"Feasibility": 0.99,
"Selected": true
},
{
"PathID": "path_002",
"Maneuver": "SWERVE_LEFT",
"RiskScore": 0.45,
"Feasibility": 0.72,
"Selected": false,
"RejectionReason": "ONCOMING_TRAFFIC"
}
],
"ExpectedOutcome": {
"StoppingDistanceM": 8.5,
"Deceleration": -8.2,
"CollisionProbability": 0.05
}
}
}
Control Command Event
{
"Action": {
"Commands": [
{
"Actuator": "BRAKE",
"CommandValue": 0.95,
"CommandUnit": "NORMALIZED",
"Priority": "EMERGENCY"
},
{
"Actuator": "THROTTLE",
"CommandValue": 0.0,
"CommandUnit": "NORMALIZED",
"Priority": "EMERGENCY"
}
],
"SafetySystemActive": true,
"AEBEngaged": true,
"HumanOverrideAvailable": true,
"CommandLatencyUs": 450
}
}
5. Sensor-to-Decision Pipeline Logging
5.1 Complete Decision Chain
For post-incident analysis, investigators need the complete causal chain:
┌─────────────────────────────────────────────────────────────────────────┐
│ COMPLETE DECISION CHAIN LOGGING │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ T+0ms: SENSOR_FRAME │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ LiDAR frame #12345 │ │
│ │ Camera frame #12345 (front, left, right) │ │
│ │ Radar returns #12345 │ │
│ │ → InputDataHash: sha256:abc123... │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ T+15ms: PERCEPT_DETECT │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ DependentEventIDs: [SENSOR_FRAME #12345] │ │
│ │ Detected: PEDESTRIAN at (12.5m, 3.2m), confidence 0.94 │ │
│ │ Classification: ADULT_WALKING, confidence 0.89 │ │
│ │ Velocity: 1.2 m/s crossing trajectory │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ T+25ms: PLAN_RISK │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ DependentEventIDs: [PERCEPT_DETECT #xyz] │ │
│ │ Time to Collision: 2.3 seconds │ │
│ │ Risk Score: 0.87 (HIGH) │ │
│ │ Recommended Action: EMERGENCY_BRAKE │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ T+30ms: PLAN_PATH │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ DependentEventIDs: [PLAN_RISK #xyz] │ │
│ │ Selected Path: EMERGENCY_BRAKE │ │
│ │ Rejected Alternatives: │ │
│ │ - SWERVE_LEFT: Oncoming traffic (risk 0.92) │ │
│ │ - SWERVE_RIGHT: Obstacle (risk 0.88) │ │
│ │ - CONTINUE: Collision certain (risk 1.0) │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ T+32ms: CTRL_CMD │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ DependentEventIDs: [PLAN_PATH #xyz] │ │
│ │ BRAKE: 95% application │ │
│ │ THROTTLE: 0% │ │
│ │ AEB System: ENGAGED │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ T+35ms: CTRL_EXEC │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ DependentEventIDs: [CTRL_CMD #xyz] │ │
│ │ Brake Pressure: 18.5 MPa │ │
│ │ Deceleration: -8.2 m/s² │ │
│ │ Latency: 3ms (command to actuation) │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────┘
5.2 Causal Linkage Implementation
from dataclasses import dataclass
from typing import List, Optional
import hashlib
import time
import uuid
@dataclass
class DVPEvent:
"""DVP Event with causal linkage."""
event_id: str
event_type: str
timestamp_ns: int
vehicle_id: str
dependent_event_ids: List[str] # Causal chain
input_data_hash: str
action: dict
prev_hash: Optional[str] = None
event_hash: Optional[str] = None
signature: Optional[str] = None
class DVPCausalChainLogger:
"""
Logger that maintains causal relationships between events.
Key insight: DependentEventIDs create a DAG (Directed Acyclic Graph)
that represents the decision pipeline. This enables reconstruction
of "why this decision was made" during incident analysis.
"""
def __init__(self, vehicle_id: str, signer):
self.vehicle_id = vehicle_id
self.signer = signer
self.event_cache = {} # event_id -> DVPEvent
self.pending_merkle = []
def log_sensor_frame(
self,
frame_id: str,
sensor_data_hash: str,
sensors: List[str]
) -> DVPEvent:
"""Log raw sensor frame capture."""
event = DVPEvent(
event_id=self._generate_uuid7(),
event_type="SENSOR_FRAME",
timestamp_ns=time.time_ns(),
vehicle_id=self.vehicle_id,
dependent_event_ids=[], # Sensor frames have no dependencies
input_data_hash=sensor_data_hash,
action={
"frame_id": frame_id,
"sensors": sensors,
"capture_timestamp_ns": time.time_ns()
}
)
return self._finalize_event(event)
def log_perception(
self,
sensor_event_ids: List[str],
detections: List[dict],
model_version: str
) -> DVPEvent:
"""
Log perception result with causal link to sensor events.
The DependentEventIDs field creates the audit trail:
"This detection was produced from THESE specific sensor frames"
"""
# Compute input hash from dependent events
dependent_hashes = [
self.event_cache[eid].event_hash
for eid in sensor_event_ids
]
input_hash = hashlib.sha256(
''.join(dependent_hashes).encode()
).hexdigest()
event = DVPEvent(
event_id=self._generate_uuid7(),
event_type="PERCEPT_DETECT",
timestamp_ns=time.time_ns(),
vehicle_id=self.vehicle_id,
dependent_event_ids=sensor_event_ids, # Causal link
input_data_hash=input_hash,
action={
"detections": detections,
"model_version": model_version,
"processing_time_ms": 12.5
}
)
return self._finalize_event(event)
def log_planning_decision(
self,
perception_event_ids: List[str],
decision: str,
path_candidates: List[dict],
risk_score: float
) -> DVPEvent:
"""
Log planning decision with causal link to perception.
Critical for incident analysis: "Why did the AI choose this action?"
The path_candidates field shows what alternatives were considered
and why they were rejected.
"""
event = DVPEvent(
event_id=self._generate_uuid7(),
event_type="PLAN_PATH",
timestamp_ns=time.time_ns(),
vehicle_id=self.vehicle_id,
dependent_event_ids=perception_event_ids,
input_data_hash=self._compute_input_hash(perception_event_ids),
action={
"decision": decision,
"path_candidates": path_candidates,
"risk_score": risk_score,
"decision_rationale": self._generate_rationale(
decision, path_candidates
)
}
)
return self._finalize_event(event)
def log_control_command(
self,
planning_event_id: str,
commands: List[dict],
safety_system_active: bool
) -> DVPEvent:
"""Log control command with causal link to planning."""
event = DVPEvent(
event_id=self._generate_uuid7(),
event_type="CTRL_CMD",
timestamp_ns=time.time_ns(),
vehicle_id=self.vehicle_id,
dependent_event_ids=[planning_event_id],
input_data_hash=self._compute_input_hash([planning_event_id]),
action={
"commands": commands,
"safety_system_active": safety_system_active,
"aeb_engaged": any(
c["priority"] == "EMERGENCY" for c in commands
)
}
)
return self._finalize_event(event)
def reconstruct_decision_chain(
self,
event_id: str
) -> List[DVPEvent]:
"""
Reconstruct the complete decision chain leading to an event.
This is the key capability for incident investigation:
Given a control command, trace back through planning,
perception, to the original sensor data.
"""
chain = []
visited = set()
def traverse(eid: str):
if eid in visited:
return
visited.add(eid)
event = self.event_cache.get(eid)
if event:
# First traverse dependencies (depth-first)
for dep_id in event.dependent_event_ids:
traverse(dep_id)
chain.append(event)
traverse(event_id)
return chain
def _finalize_event(self, event: DVPEvent) -> DVPEvent:
"""Compute hash and signature."""
# Compute event hash
event.event_hash = self._compute_event_hash(event)
# Sign
event.signature = self.signer.sign(event.event_hash)
# Cache for causal chain reconstruction
self.event_cache[event.event_id] = event
# Add to Merkle batch
self.pending_merkle.append(event)
return event
def _compute_event_hash(self, event: DVPEvent) -> str:
"""SHA-256 of canonicalized event."""
import json
canonical = json.dumps({
"event_id": event.event_id,
"event_type": event.event_type,
"timestamp_ns": event.timestamp_ns,
"vehicle_id": event.vehicle_id,
"dependent_event_ids": event.dependent_event_ids,
"input_data_hash": event.input_data_hash,
"action": event.action
}, sort_keys=True, separators=(',', ':'))
return hashlib.sha256(canonical.encode()).hexdigest()
def _generate_uuid7(self) -> str:
"""Generate time-ordered UUID v7."""
timestamp_ms = int(time.time() * 1000)
uuid_bytes = timestamp_ms.to_bytes(6, 'big') + \
uuid.uuid4().bytes[6:]
return str(uuid.UUID(bytes=bytes(uuid_bytes)))
def _compute_input_hash(self, event_ids: List[str]) -> str:
hashes = [self.event_cache[eid].event_hash for eid in event_ids]
return hashlib.sha256(''.join(hashes).encode()).hexdigest()
def _generate_rationale(
self,
decision: str,
candidates: List[dict]
) -> str:
"""Generate human-readable decision rationale."""
selected = next((c for c in candidates if c.get("selected")), None)
rejected = [c for c in candidates if not c.get("selected")]
rationale = f"Selected {decision} (risk: {selected['risk_score']:.2f}). "
rationale += "Rejected: "
rationale += ", ".join([
f"{c['maneuver']} ({c.get('rejection_reason', 'higher risk')})"
for c in rejected
])
return rationale
6. Merkle Tree and External Anchoring
6.1 Automotive-Specific Anchoring Requirements
Following VCP v1.1, DVP requires external anchoring for all tiers:
| DVP Tier | Anchor Frequency | Target | Rationale |
|---|---|---|---|
| Platinum (L4-5 AV) | 10 seconds | TSA + Safety Authority | Real-time completeness for incident |
| Gold (ADAS) | 1 minute | TSA | Balance of assurance and bandwidth |
| Silver (Testing) | Trip end | Database | Development efficiency |
6.2 RFC 6962-Compliant Merkle Tree
import hashlib
from typing import List, Tuple
class DVPMerkleTree:
"""
RFC 6962-compliant Merkle tree for DVP events.
Domain separation prevents second-preimage attacks:
- Leaf: H(0x00 || event_hash)
- Node: H(0x01 || left || right)
"""
LEAF_PREFIX = b'\x00'
NODE_PREFIX = b'\x01'
def __init__(self):
self.leaves: List[bytes] = []
self.event_ids: List[str] = []
def add_event(self, event: DVPEvent):
"""Add event to tree."""
leaf_hash = self._hash_leaf(bytes.fromhex(event.event_hash))
self.leaves.append(leaf_hash)
self.event_ids.append(event.event_id)
def get_root(self) -> str:
"""Compute Merkle root."""
if not self.leaves:
return hashlib.sha256(b'empty').hexdigest()
nodes = self.leaves.copy()
while len(nodes) > 1:
next_level = []
for i in range(0, len(nodes), 2):
left = nodes[i]
right = nodes[i + 1] if i + 1 < len(nodes) else left
parent = self._hash_node(left, right)
next_level.append(parent)
nodes = next_level
return nodes[0].hex()
def get_inclusion_proof(self, event_id: str) -> List[Tuple[str, str]]:
"""
Generate Merkle inclusion proof for an event.
Returns list of (sibling_hash, position) tuples.
"""
try:
idx = self.event_ids.index(event_id)
except ValueError:
raise ValueError(f"Event {event_id} not in tree")
proof = []
nodes = self.leaves.copy()
while len(nodes) > 1:
sibling_idx = idx ^ 1 # XOR to get sibling
if sibling_idx < len(nodes):
position = "right" if idx % 2 == 0 else "left"
proof.append((nodes[sibling_idx].hex(), position))
# Move to parent level
next_level = []
for i in range(0, len(nodes), 2):
left = nodes[i]
right = nodes[i + 1] if i + 1 < len(nodes) else left
next_level.append(self._hash_node(left, right))
nodes = next_level
idx //= 2
return proof
@staticmethod
def verify_inclusion(
event_hash: str,
proof: List[Tuple[str, str]],
root: str
) -> bool:
"""
Verify event inclusion using Merkle proof.
Can be performed by any third party with:
- The event hash
- The inclusion proof
- The anchored Merkle root
"""
current = DVPMerkleTree._hash_leaf_static(bytes.fromhex(event_hash))
for sibling_hash, position in proof:
sibling = bytes.fromhex(sibling_hash)
if position == "right":
current = DVPMerkleTree._hash_node_static(current, sibling)
else:
current = DVPMerkleTree._hash_node_static(sibling, current)
return current.hex() == root
def _hash_leaf(self, data: bytes) -> bytes:
return hashlib.sha256(self.LEAF_PREFIX + data).digest()
def _hash_node(self, left: bytes, right: bytes) -> bytes:
return hashlib.sha256(self.NODE_PREFIX + left + right).digest()
@staticmethod
def _hash_leaf_static(data: bytes) -> bytes:
return hashlib.sha256(DVPMerkleTree.LEAF_PREFIX + data).digest()
@staticmethod
def _hash_node_static(left: bytes, right: bytes) -> bytes:
return hashlib.sha256(DVPMerkleTree.NODE_PREFIX + left + right).digest()
6.3 External Anchor Service
import requests
from dataclasses import dataclass
@dataclass
class AnchorReceipt:
"""Proof of external anchoring."""
merkle_root: str
timestamp: int
anchor_target: str # e.g., "rfc3161:freetsa.org"
anchor_proof: bytes # TSA response or blockchain tx
event_count: int
class DVPAnchorService:
"""
External anchoring service for DVP.
Supports multiple anchor targets for redundancy:
- RFC 3161 TSA (primary)
- Transparency Log (secondary)
- Safety Authority Portal (regulatory)
"""
def __init__(self, tsa_url: str, safety_authority_url: str = None):
self.tsa_url = tsa_url
self.safety_authority_url = safety_authority_url
def anchor_batch(
self,
merkle_tree: DVPMerkleTree,
vehicle_id: str
) -> AnchorReceipt:
"""
Anchor Merkle root to external timestamping authority.
For Platinum tier, also submit to Safety Authority.
"""
root = merkle_tree.get_root()
# RFC 3161 Timestamp Request
tsa_proof = self._request_timestamp(root)
# Optionally submit to Safety Authority
if self.safety_authority_url:
self._submit_to_authority(
root,
vehicle_id,
merkle_tree.event_ids
)
return AnchorReceipt(
merkle_root=root,
timestamp=int(time.time() * 1_000_000_000),
anchor_target=f"rfc3161:{self.tsa_url}",
anchor_proof=tsa_proof,
event_count=len(merkle_tree.leaves)
)
def _request_timestamp(self, digest_hex: str) -> bytes:
"""Request RFC 3161 timestamp."""
from asn1crypto import tsp, core
digest = bytes.fromhex(digest_hex)
ts_req = tsp.TimeStampReq({
'version': 1,
'message_imprint': {
'hash_algorithm': {'algorithm': 'sha256'},
'hashed_message': digest
},
'cert_req': True
})
response = requests.post(
self.tsa_url,
data=ts_req.dump(),
headers={'Content-Type': 'application/timestamp-query'}
)
return response.content
def _submit_to_authority(
self,
root: str,
vehicle_id: str,
event_ids: List[str]
):
"""Submit to Safety Authority (UNECE DSSAD compliance)."""
payload = {
"vehicle_id": vehicle_id,
"merkle_root": root,
"event_count": len(event_ids),
"anchor_timestamp": time.time_ns(),
"protocol_version": "DVP/0.1"
}
requests.post(
f"{self.safety_authority_url}/anchor",
json=payload
)
7. DVP-XREF: Multi-System Cross-Reference
7.1 Automotive Cross-Reference Scenarios
VCP v1.1's VCP-XREF enables dual logging between parties. In automotive, this applies to:
| Scenario | Party A | Party B | Benefit |
|---|---|---|---|
| V2X Communication | Vehicle | Infrastructure | Prove message was sent/received |
| Multi-ECU Verification | Perception ECU | Planning ECU | Cross-verify internal decisions |
| OTA Update Logging | Vehicle | OEM Server | Prove software state at incident |
| Fleet Management | Vehicle | Fleet Operator | Independent audit trails |
7.2 V2X Cross-Reference Implementation
@dataclass
class DVPXRefEvent:
"""DVP event with cross-reference capability."""
base_event: DVPEvent
xref: dict
class DVPXRefLogger:
"""
Cross-reference logger for V2X and multi-system scenarios.
Key guarantee: Unless both parties collude, manipulation
by one party is detectable by the other.
"""
def log_v2x_sent(
self,
message_type: str,
recipient_id: str,
message_hash: str
) -> DVPXRefEvent:
"""Log outgoing V2X message with cross-reference."""
xref_id = str(uuid.uuid4())
base = self.logger.log_event(
event_type="V2X_TX",
action={
"message_type": message_type,
"recipient": recipient_id,
"message_hash": message_hash
}
)
xref = {
"CrossReferenceID": xref_id,
"PartyRole": "INITIATOR",
"CounterpartyID": recipient_id,
"SharedEventKey": {
"MessageHash": message_hash,
"Timestamp": base.timestamp_ns,
"ToleranceMs": 100
},
"ReconciliationStatus": "PENDING"
}
return DVPXRefEvent(base_event=base, xref=xref)
def log_v2x_received(
self,
xref_id: str,
sender_id: str,
message_type: str,
message_hash: str,
sender_event_hash: str
) -> DVPXRefEvent:
"""Log incoming V2X message with cross-reference to sender."""
base = self.logger.log_event(
event_type="V2X_RX",
action={
"message_type": message_type,
"sender": sender_id,
"message_hash": message_hash
}
)
xref = {
"CrossReferenceID": xref_id,
"PartyRole": "COUNTERPARTY",
"CounterpartyID": sender_id,
"SharedEventKey": {
"MessageHash": message_hash,
"Timestamp": base.timestamp_ns,
"ToleranceMs": 100
},
"ExpectedCounterpartyHash": sender_event_hash,
"ReconciliationStatus": "MATCHED"
}
return DVPXRefEvent(base_event=base, xref=xref)
def verify_v2x_exchange(
self,
sender_event: DVPXRefEvent,
receiver_event: DVPXRefEvent
) -> dict:
"""
Verify V2X message exchange consistency.
This can detect:
- Sender claiming message was sent when it wasn't
- Receiver claiming message wasn't received when it was
- Message content tampering
"""
results = {
"xref_id_match": False,
"message_hash_match": False,
"timestamp_within_tolerance": False,
"overall": "DISCREPANCY"
}
# Check CrossReferenceID
if sender_event.xref["CrossReferenceID"] == \
receiver_event.xref["CrossReferenceID"]:
results["xref_id_match"] = True
# Check message hash
if sender_event.xref["SharedEventKey"]["MessageHash"] == \
receiver_event.xref["SharedEventKey"]["MessageHash"]:
results["message_hash_match"] = True
# Check timestamp tolerance
time_diff = abs(
sender_event.base_event.timestamp_ns -
receiver_event.base_event.timestamp_ns
)
tolerance_ns = sender_event.xref["SharedEventKey"]["ToleranceMs"] * 1_000_000
if time_diff <= tolerance_ns:
results["timestamp_within_tolerance"] = True
# Overall result
if all([
results["xref_id_match"],
results["message_hash_match"],
results["timestamp_within_tolerance"]
]):
results["overall"] = "MATCHED"
return results
8. EDR Integration Architecture
8.1 Complementary Roles
DVP does not replace traditional EDRs—it complements them:
┌─────────────────────────────────────────────────────────────────────────┐
│ INTEGRATED FLIGHT RECORDER │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌───────────────────────────────┐ ┌───────────────────────────────┐ │
│ │ TRADITIONAL EDR │ │ DVP LAYER │ │
│ │ (Physical State) │ │ (AI Decisions) │ │
│ ├───────────────────────────────┤ ├───────────────────────────────┤ │
│ │ • Vehicle speed │ │ • Sensor input hashes │ │
│ │ • Acceleration/deceleration │ │ • Object detection results │ │
│ │ • Brake application │ │ • Classification confidence │ │
│ │ • Steering angle │ │ • Path planning candidates │ │
│ │ • Throttle position │ │ • Decision rationale │ │
│ │ • Seatbelt status │ │ • Risk assessment scores │ │
│ │ • Airbag deployment │ │ • Control command reasoning │ │
│ │ • Engine RPM │ │ • Human override events │ │
│ │ • GPS position │ │ • V2X message logs │ │
│ ├───────────────────────────────┤ ├───────────────────────────────┤ │
│ │ Storage: Crash-survivable │ │ Storage: Crash-survivable │ │
│ │ Retention: 5 seconds pre- │ │ Retention: 5+ seconds pre- │ │
│ │ crash minimum │ │ crash + full trip │ │
│ │ Format: Proprietary │ │ Format: Open standard (JSON) │ │
│ │ Integrity: None │ │ Integrity: Merkle + Anchor │ │
│ └───────────────────────────────┘ └───────────────────────────────┘ │
│ │
│ ┌─────────────┐ │
│ │ COMBINED │ │
│ │ ANALYSIS │ │
│ └─────────────┘ │
│ │ │
│ ┌────────────────┼────────────────┐ │
│ ▼ ▼ ▼ │
│ "What happened" "Why it happened" "Was it tampered" │
│ (EDR data) (DVP decisions) (Merkle proofs) │
│ │
└─────────────────────────────────────────────────────────────────────────┘
8.2 Physical Integration Options
| Option | Description | Pros | Cons |
|---|---|---|---|
| Separate Module | DVP as standalone crash-survivable unit | No EDR modification | Additional hardware |
| EDR Extension | DVP data in extended EDR storage | Single unit | Requires EDR vendor cooperation |
| Cloud Backup | Real-time upload + local buffer | Redundancy | Connectivity dependency |
| Hybrid | Local EDR + periodic cloud sync | Best of both | Complexity |
9. Reference Implementation
9.1 Complete DVP Sidecar
"""
DVP Sidecar Reference Implementation
This sidecar integrates with an autonomous driving stack,
capturing events and producing VCP v1.1-compliant audit trails.
"""
import asyncio
from dataclasses import dataclass
from typing import Callable, List
import json
import time
class DVPSidecar:
"""
DVP Sidecar for autonomous driving systems.
Integration pattern: Event stream tap
Non-invasive: Does not modify AD stack behavior
Fail-safe: Sidecar failure does not impact driving
"""
def __init__(
self,
vehicle_id: str,
signing_key: bytes,
anchor_service: DVPAnchorService,
tier: str = "PLATINUM"
):
self.vehicle_id = vehicle_id
self.signer = Ed25519Signer(signing_key)
self.anchor_service = anchor_service
self.tier = tier
self.chain_logger = DVPCausalChainLogger(vehicle_id, self.signer)
self.merkle_tree = DVPMerkleTree()
# Anchor frequency based on tier
self.anchor_intervals = {
"PLATINUM": 10, # 10 seconds
"GOLD": 60, # 1 minute
"SILVER": 3600 # 1 hour
}
self.running = False
self.event_queue = asyncio.Queue()
async def start(self):
"""Start the sidecar processing loop."""
self.running = True
# Start background tasks
asyncio.create_task(self._process_events())
asyncio.create_task(self._periodic_anchor())
print(f"DVP Sidecar started for vehicle {self.vehicle_id}")
async def stop(self):
"""Stop sidecar and perform final anchor."""
self.running = False
# Final anchor before shutdown
await self._anchor_current_batch()
print("DVP Sidecar stopped")
# === Event Ingestion API ===
async def on_sensor_frame(
self,
frame_id: str,
sensor_data_hash: str,
sensors: List[str]
):
"""Called when sensor frame is captured."""
await self.event_queue.put({
"type": "sensor",
"frame_id": frame_id,
"data_hash": sensor_data_hash,
"sensors": sensors
})
async def on_perception_result(
self,
sensor_frame_ids: List[str],
detections: List[dict],
model_version: str
):
"""Called when perception produces results."""
await self.event_queue.put({
"type": "perception",
"sensor_ids": sensor_frame_ids,
"detections": detections,
"model": model_version
})
async def on_planning_decision(
self,
perception_event_ids: List[str],
decision: str,
candidates: List[dict],
risk_score: float
):
"""Called when planning makes a decision."""
await self.event_queue.put({
"type": "planning",
"perception_ids": perception_event_ids,
"decision": decision,
"candidates": candidates,
"risk": risk_score
})
async def on_control_command(
self,
planning_event_id: str,
commands: List[dict],
safety_active: bool
):
"""Called when control command is issued."""
await self.event_queue.put({
"type": "control",
"planning_id": planning_event_id,
"commands": commands,
"safety": safety_active
})
# === Internal Processing ===
async def _process_events(self):
"""Process events from queue."""
while self.running:
try:
event_data = await asyncio.wait_for(
self.event_queue.get(),
timeout=0.1
)
event = self._create_event(event_data)
self.merkle_tree.add_event(event)
except asyncio.TimeoutError:
continue
except Exception as e:
# Log error but don't crash
print(f"Event processing error: {e}")
def _create_event(self, data: dict) -> DVPEvent:
"""Create DVP event from raw data."""
event_type = data["type"]
if event_type == "sensor":
return self.chain_logger.log_sensor_frame(
data["frame_id"],
data["data_hash"],
data["sensors"]
)
elif event_type == "perception":
return self.chain_logger.log_perception(
data["sensor_ids"],
data["detections"],
data["model"]
)
elif event_type == "planning":
return self.chain_logger.log_planning_decision(
data["perception_ids"],
data["decision"],
data["candidates"],
data["risk"]
)
elif event_type == "control":
return self.chain_logger.log_control_command(
data["planning_id"],
data["commands"],
data["safety"]
)
async def _periodic_anchor(self):
"""Periodically anchor to external service."""
interval = self.anchor_intervals[self.tier]
while self.running:
await asyncio.sleep(interval)
await self._anchor_current_batch()
async def _anchor_current_batch(self):
"""Anchor current Merkle tree batch."""
if len(self.merkle_tree.leaves) == 0:
return
try:
receipt = self.anchor_service.anchor_batch(
self.merkle_tree,
self.vehicle_id
)
print(f"Anchored {receipt.event_count} events, "
f"root: {receipt.merkle_root[:16]}...")
# Start new tree
self.merkle_tree = DVPMerkleTree()
except Exception as e:
print(f"Anchor failed: {e}")
# Events remain in tree for next attempt
# === Usage Example ===
async def main():
"""Example usage of DVP Sidecar."""
# Initialize
sidecar = DVPSidecar(
vehicle_id="VIN_XXXXXXXXXXXX",
signing_key=generate_ed25519_key(),
anchor_service=DVPAnchorService("https://freetsa.org/tsr"),
tier="PLATINUM"
)
await sidecar.start()
# Simulate AD stack events
sensor_event = await sidecar.on_sensor_frame(
"frame_001",
"sha256:abc123...",
["lidar_front", "camera_front", "radar_front"]
)
perception_event = await sidecar.on_perception_result(
["frame_001"],
[{"class": "PEDESTRIAN", "confidence": 0.94}],
"perception_v3.2.1"
)
planning_event = await sidecar.on_planning_decision(
[perception_event.event_id],
"EMERGENCY_BRAKE",
[{"maneuver": "BRAKE", "risk": 0.15, "selected": True}],
0.87
)
await sidecar.on_control_command(
planning_event.event_id,
[{"actuator": "BRAKE", "value": 0.95}],
safety_active=True
)
# Wait for anchor
await asyncio.sleep(15)
await sidecar.stop()
if __name__ == "__main__":
asyncio.run(main())
10. Regulatory Compliance Mapping
10.1 UNECE WP.29 R157 (ALKS)
UNECE Regulation 157 defines requirements for Automated Lane Keeping Systems (ALKS), including the Data Storage System for Automated Driving (DSSAD):
| R157 Requirement | DVP Implementation |
|---|---|
| 5-second pre-accident data retention | ✅ Continuous logging with Merkle anchoring |
| Timestamp accuracy | ✅ PTP/GNSS synchronized, UUID v7 ordering |
| Data integrity guarantee | ✅ SHA-256 hash chain + Merkle tree |
| Authority access | ✅ Standard JSON format, inclusion proofs |
| Crash survivability | ✅ Requires crash-survivable storage (EDR integration) |
10.2 EU AI Act Compliance
EU AI Act classifies autonomous driving AI as high-risk (Annex III):
| EU AI Act Article | DVP Alignment |
|---|---|
| Article 12: Record-keeping | ✅ Complete decision audit trail |
| Article 14: Human oversight | ✅ CTRL_OVERRIDE events logged |
| Article 13: Transparency | ✅ Open schema, decision rationale |
| Article 15: Accuracy/robustness | ✅ Model version tracking, confidence scores |
10.3 ISO 26262 / SOTIF
| Standard | DVP Support |
|---|---|
| ISO 26262 (Functional Safety) | Traceability of safety-critical decisions |
| ISO/PAS 21448 (SOTIF) | Evidence for unknown/unknowable hazard analysis |
11. Conclusion
11.1 Key Takeaways
The Problem is Real: Current EDRs capture physical state but not AI cognition. Incident investigation requires both.
VCP v1.1 Provides the Foundation: The three-layer architecture, mandatory external anchoring, and completeness guarantees translate directly to automotive needs.
DVP Extends, Not Replaces: DVP complements traditional EDRs, adding the AI decision layer that regulations increasingly require.
Standards Alignment is Critical: UNECE WP.29 R157, EU AI Act, and ISO standards all point toward cryptographically verifiable AI audit trails.
Implementation is Feasible: The sidecar pattern enables integration without modifying existing AD stacks.
11.2 Roadmap
| Milestone | Target |
|---|---|
| DVP Draft v0.1 | 2025 Q4 |
| OEM Technical Validation | 2026 Q1 |
| UNECE GRVA Proposal | 2026 Q2 |
| DVP v1.0 Official Release | 2026 Q3 |
| ISO/SAE Collaboration | 2027 |
11.3 Get Involved
DVP is under active development. We welcome participation from:
- OEMs and Tier 1 suppliers
- Regulatory authorities
- Safety researchers
- Open-source contributors
📧 Contact: standards@veritaschain.org
🔗 GitHub: https://github.com/veritaschain
📄 Specification: https://veritaschain.org/vap/dvp/
"Aircraft have physical black boxes. Autonomous vehicles need AI black boxes too."
— VeritasChain Standards Organization
"The question is not whether autonomous vehicles will have accidents.
The question is whether we can prove what happened when they do."
References:
- UNECE WP.29 R157: Automated Lane Keeping Systems
- EU AI Act (Regulation 2024/1689)
- ISO 26262: Functional Safety
- ISO/PAS 21448: Safety of the Intended Functionality
- VCP Specification v1.1: https://veritaschain.org/vcp/
- RFC 6962: Certificate Transparency
- RFC 3161: Time-Stamp Protocol
- IEEE 1588-2019: Precision Time Protocol
License: CC BY 4.0 International
Tags: #autonomousvehicles #ai #safety #cryptography #automotive #audittrail
© 2025 VeritasChain Standards Organization. All rights reserved.
Top comments (0)