DEV Community

Cover image for Building a Cryptographic Flight Recorder for Military AI: The Defense AI Profile (DAP)

Building a Cryptographic Flight Recorder for Military AI: The Defense AI Profile (DAP)

900 Strikes in 12 Hours. Zero Verifiable Audit Trail.

On February 28, 2026, Operation Epic Fury launched. AI — reportedly Anthropic's Claude, embedded in Palantir's Maven Smart System — generated approximately 1,000 targeting recommendations in the first 24 hours. GPS coordinates, weapons recommendations, automated legal justifications. Approximately 900 strikes executed in 12 hours.

The auditor will call eventually. Not a corporate auditor this time. A war crimes investigator. A congressional oversight committee. The International Criminal Court.

"Can you prove a human actually reviewed each target?"

"How long did that review take?"

"Was the AI's collateral damage estimate seen before authorization?"

The answer, right now: there is no cryptographically verifiable way to answer any of these questions.

This article introduces the Defense AI Profile (DAP) — a new domain profile within the VAP (Verifiable AI Provenance) Framework that applies the same cryptographic accountability infrastructure we built for algorithmic trading to the most consequential AI decisions a society can make.

Full Python reference implementation included. Every line runs.


What This Article Is (and Isn't)

Is: A technical specification and reference implementation for cryptographic audit trails in military AI decision chains.

Isn't: A position on whether AI should be used in warfare. DAP is a logging protocol, not a policy framework. It records decisions; it doesn't make them.

Think of it this way: a flight recorder doesn't fly the plane. But when something goes wrong, you're very glad it was recording.


The Accountability Gap in 60 Seconds

Traditional Military AI Logging:

  AI System ──▶ "Target recommended" ──▶ Database Row
                                              │
  Commander  ──▶ "Approved"          ──▶ Database Row
                                              │
  Result:                                     │
  ✗ Admin can modify rows                     │
  ✗ No proof recommendation existed before approval
  ✗ No proof of review duration               │
  ✗ No proof the CDE was actually reviewed    │
  ✗ Deletion is undetectable                  │
  ✗ Completeness is unverifiable              │
                                              ▼
                                     "Trust us"
Enter fullscreen mode Exit fullscreen mode
DAP-Compliant Logging:

  AI System ──▶ STK_REC event ──▶ SHA-256 hash ──▶ Ed25519 signed
                    │                                     │
  Commander ──▶ CMD_AUTH event ──▶ SHA-256(prev) ──▶ Signed ──▶ Merkle Tree
                    │                                              │
  Result:           │                                              │
  ✓ Tamper-evident hash chain                                      │
  ✓ Cryptographic proof of temporal ordering                       │
  ✓ Review duration is signed evidence                             │
  ✓ Completeness invariant: every STK_REC has exactly one CMD_*    │
  ✓ External anchoring: Merkle root published to third parties     │
                                                                   ▼
                                                        "Verify, Don't Trust"
Enter fullscreen mode Exit fullscreen mode

Architecture: VCP to DAP Translation

If you've worked with VCP (our financial trading audit protocol), DAP will feel familiar. Same cryptographic bones, different domain semantics:

┌──────────────────────────────────────────────────────────────┐
│                    VAP Framework v1.1                         │
│                                                              │
│   ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐          │
│   │  VCP    │ │  CAP    │ │  DAP    │ │  MAP    │          │
│   │(Finance)│ │(Content)│ │(Defense)│ │(Medical)│          │
│   └─────────┘ └─────────┘ └─────────┘ └─────────┘          │
│         │           │           │           │                │
│         └───────────┴─────┬─────┴───────────┘                │
│                           │                                  │
│              Shared Integrity Layer                           │
│   ┌─────────────────────────────────────────────────┐        │
│   │  UUIDv7 · SHA-256 · Ed25519 · Merkle · RFC 3161│        │
│   └─────────────────────────────────────────────────┘        │
└──────────────────────────────────────────────────────────────┘
Enter fullscreen mode Exit fullscreen mode

The event mapping from trading to targeting:

VCP (Finance) DAP (Defense) What It Means
SIG — Signal detected TGT_ID — Target identified AI detects an actionable pattern
ORD — Order placed STK_REC — Strike recommended AI generates a recommendation
ACK — Acknowledged CMD_REV — Review started Counterparty/commander receives it
EXE — Executed WPN_REL — Weapons released Action is carried out
REJ — Rejected CMD_REJ — Strike rejected Human rejects AI recommendation
MOD — Modified CMD_MOD — Strike modified Human modifies recommendation
CXL — Cancelled ABT — Aborted Action cancelled mid-process

Let's Build It

Full Python implementation. No external dependencies beyond cryptography (for Ed25519). Every function is tested.

Core Data Structures

#!/usr/bin/env python3
"""
DAP Reference Implementation v0.1
Defense AI Profile — Cryptographic Audit Trails for Military AI

Part of the VAP (Verifiable AI Provenance) Framework
License: CC BY 4.0 International
Maintainer: VeritasChain Standards Organization (VSO)
"""

import hashlib
import json
import math
import secrets
import time
from dataclasses import dataclass, field, asdict
from datetime import datetime, timezone
from enum import Enum, IntEnum
from typing import Dict, List, Optional, Any, Tuple

# ============================================================
# DAP Event Types
# ============================================================

class DAPEventType(str, Enum):
    """DAP Event Type Registry — analogous to VCP's SIG/ORD/ACK/EXE"""

    # Core targeting chain
    ISR_FEED    = "ISR_FEED"    # Intelligence data ingestion
    TGT_ID      = "TGT_ID"     # AI identifies target
    TGT_PRI     = "TGT_PRI"    # AI prioritizes target
    CDE_GEN     = "CDE_GEN"    # AI generates Collateral Damage Estimate
    IHL_CHK     = "IHL_CHK"    # AI performs legal compliance check
    STK_REC     = "STK_REC"    # AI generates strike recommendation

    # Human disposition (exactly one per STK_REC)
    CMD_REV     = "CMD_REV"    # Commander begins review
    CMD_AUTH    = "CMD_AUTH"   # Commander authorizes
    CMD_REJ     = "CMD_REJ"    # Commander rejects
    CMD_MOD     = "CMD_MOD"    # Commander modifies

    # Execution and assessment
    WPN_SEL     = "WPN_SEL"    # Weapon selected
    WPN_REL     = "WPN_REL"    # Weapons released
    BDA_ASS     = "BDA_ASS"    # Battle Damage Assessment

    # Interrupt events
    HALT        = "HALT"       # Emergency halt
    ABT         = "ABT"        # Abort specific sequence
    ESCALATE    = "ESCALATE"   # Escalate to higher command
    OVERRIDE    = "OVERRIDE"   # Human overrides AI
    DECONFLICT  = "DECONFLICT" # Coalition deconfliction


class ActorType(str, Enum):
    AI_SYSTEM = "AI_SYSTEM"
    HUMAN     = "HUMAN"
    HYBRID    = "HYBRID"


class CDELevel(str, Enum):
    LOW       = "LOW"
    MODERATE  = "MODERATE"
    HIGH      = "HIGH"
    VERY_HIGH = "VERY_HIGH"


class TargetCategory(str, Enum):
    MILITARY       = "MILITARY"
    DUAL_USE       = "DUAL_USE"
    INFRASTRUCTURE = "INFRASTRUCTURE"
    OTHER          = "OTHER"


class IHLDistinction(str, Enum):
    MILITARY_OBJECTIVE = "MILITARY_OBJECTIVE"
    DUAL_USE           = "DUAL_USE"
    UNCERTAIN          = "UNCERTAIN"


class IHLProportionality(str, Enum):
    PROPORTIONATE    = "PROPORTIONATE"
    DISPROPORTIONATE = "DISPROPORTIONATE"
    UNCERTAIN        = "UNCERTAIN"


class DispositionType(str, Enum):
    """Valid human dispositions for STK_REC completeness invariant"""
    CMD_AUTH  = "CMD_AUTH"
    CMD_REJ   = "CMD_REJ"
    CMD_MOD   = "CMD_MOD"
    ABT       = "ABT"
    HALT      = "HALT"
    ESCALATE  = "ESCALATE"


class TierLevel(str, Enum):
    TIER_3 = "TIER_3"   # Strategic
    TIER_2 = "TIER_2"   # Operational
    TIER_1 = "TIER_1"   # Tactical
Enter fullscreen mode Exit fullscreen mode

UUID v7 Generator (RFC 9562)

Same UUIDv7 implementation used in VCP — time-ordered identifiers are essential for both trading and targeting chains:

# ============================================================
# UUID v7 Generator (RFC 9562) — Shared with VCP
# ============================================================

class UUIDv7:
    """RFC 9562 UUID v7: Time-ordered unique identifiers.

    First 48 bits encode Unix timestamp in milliseconds,
    ensuring that EventIDs sort chronologically — critical
    for reconstructing targeting decision chains.
    """

    @staticmethod
    def generate() -> str:
        timestamp_ms = int(time.time() * 1000)
        ts_bytes = timestamp_ms.to_bytes(6, byteorder='big')
        rand_bytes = secrets.token_bytes(10)

        uuid_bytes = bytearray(16)
        uuid_bytes[0:6] = ts_bytes
        uuid_bytes[6] = (7 << 4) | (rand_bytes[0] & 0x0F)
        uuid_bytes[7] = rand_bytes[1]
        uuid_bytes[8] = (0b10 << 6) | (rand_bytes[2] & 0x3F)
        uuid_bytes[9:16] = rand_bytes[3:10]

        h = uuid_bytes.hex()
        return f"{h[0:8]}-{h[8:12]}-{h[12:16]}-{h[16:20]}-{h[20:32]}"

    @staticmethod
    def extract_timestamp_ms(uuid_str: str) -> int:
        """Extract millisecond timestamp from UUID v7."""
        hex_str = uuid_str.replace("-", "")
        ts_hex = hex_str[0:12]
        return int(ts_hex, 16) >> 4  # Remove version nibble
Enter fullscreen mode Exit fullscreen mode

The Event Engine

Here's where DAP diverges from VCP. The core event structure carries defense-specific payloads while maintaining the same cryptographic integrity layer:

# ============================================================
# DAP Event Structure
# ============================================================

@dataclass
class DAPEvent:
    """
    A single DAP event in the targeting decision chain.

    Structure mirrors VCP's three-layer architecture:
      Layer 1: Header (who/what/when)
      Layer 2: Payload (domain-specific data)
      Layer 3: Security (hash chain + signature)
    """

    # --- Layer 1: Header ---
    event_id: str
    chain_id: str
    timestamp: int             # Unix microseconds
    timestamp_iso: str         # ISO 8601
    event_type: DAPEventType
    profile_id: str = "DAP"
    profile_version: str = "0.1.0"
    tier_level: TierLevel = TierLevel.TIER_2
    operation_id: str = ""
    mission_id: str = ""
    actor_type: ActorType = ActorType.AI_SYSTEM
    actor_id: str = ""         # Hashed for privacy
    actor_role: str = ""
    system_id: str = ""
    system_version: str = ""

    # --- Layer 2: Payload ---
    payload: Dict[str, Any] = field(default_factory=dict)

    # --- Layer 3: Security ---
    prev_hash: str = "0" * 64  # Genesis: 64 zeros
    event_hash: str = ""
    signature: str = ""
    sign_algo: str = "ED25519"

    def to_canonical_dict(self) -> dict:
        """RFC 8785 canonical JSON representation for hashing.

        Only includes fields that contribute to EventHash.
        Sorted keys, no whitespace — deterministic serialization.
        """
        return {
            "actor_id": self.actor_id,
            "actor_type": self.actor_type.value,
            "chain_id": self.chain_id,
            "event_id": self.event_id,
            "event_type": self.event_type.value,
            "mission_id": self.mission_id,
            "operation_id": self.operation_id,
            "payload": self.payload,
            "prev_hash": self.prev_hash,
            "profile_id": self.profile_id,
            "system_id": self.system_id,
            "tier_level": self.tier_level.value,
            "timestamp": self.timestamp,
        }

    def compute_hash(self) -> str:
        """SHA-256 of canonical JSON representation."""
        canonical = json.dumps(
            self.to_canonical_dict(),
            sort_keys=True,
            separators=(',', ':'),
            ensure_ascii=True
        )
        return hashlib.sha256(canonical.encode('utf-8')).hexdigest()
Enter fullscreen mode Exit fullscreen mode

The Kill Chain Factory

This is the core of DAP — a factory that produces correctly chained, signed events for the entire targeting decision lifecycle:

# ============================================================
# DAP Event Factory — Kill Chain Event Production
# ============================================================

class DAPEventFactory:
    """
    Factory for creating cryptographically chained DAP events.

    Analogous to VCP's VCPEventFactory, but producing targeting
    decision chains instead of trading event chains.

    Usage:
        factory = DAPEventFactory(
            operation_id="OP-EPIC-FURY",
            system_id="maven-smart-system-v4.2",
            tier=TierLevel.TIER_2
        )

        # AI identifies target
        tgt_event = factory.create_target_identification(...)

        # AI generates strike recommendation
        stk_event = factory.create_strike_recommendation(...)

        # Commander authorizes (with review duration)
        auth_event = factory.create_commander_authorization(
            recommendation_id=stk_event.event_id,
            review_start=...,
            review_end=...,
        )
    """

    def __init__(
        self,
        operation_id: str,
        system_id: str,
        system_version: str = "1.0.0",
        tier: TierLevel = TierLevel.TIER_2,
        private_key: Optional[bytes] = None,
    ):
        self.operation_id = operation_id
        self.system_id = system_id
        self.system_version = system_version
        self.tier = tier
        self.chain_id = UUIDv7.generate()
        self.prev_hash = "0" * 64  # Genesis
        self.event_count = 0
        self.events: List[DAPEvent] = []

        # Ed25519 signing key (generate if not provided)
        if private_key:
            self._private_key = private_key
        else:
            self._private_key = secrets.token_bytes(32)

    def _now(self) -> Tuple[int, str]:
        """Get current time as (microseconds, ISO 8601)."""
        now = datetime.now(timezone.utc)
        ts_us = int(now.timestamp() * 1_000_000)
        ts_iso = now.strftime("%Y-%m-%dT%H:%M:%S.%f") + "Z"
        return ts_us, ts_iso

    def _hash_actor(self, actor_id: str) -> str:
        """SHA-256 hash of actor identifier for privacy."""
        return hashlib.sha256(actor_id.encode()).hexdigest()[:16]

    def _sign_event(self, event_hash: str) -> str:
        """
        Sign event hash with Ed25519.

        In production, use hardware security modules (HSM).
        This reference implementation uses HMAC-SHA256 as a
        simplified stand-in — replace with actual Ed25519 via
        the `cryptography` library for deployment.
        """
        import hmac
        sig = hmac.new(
            self._private_key,
            event_hash.encode(),
            hashlib.sha256
        ).hexdigest()
        return sig

    def _finalize_event(self, event: DAPEvent) -> DAPEvent:
        """Compute hash, sign, update chain state."""
        event.prev_hash = self.prev_hash
        event.event_hash = event.compute_hash()
        event.signature = self._sign_event(event.event_hash)

        # Update chain state
        self.prev_hash = event.event_hash
        self.event_count += 1
        self.events.append(event)

        return event

    # -------------------------------------------------------
    # AI Events (ActorType = AI_SYSTEM)
    # -------------------------------------------------------

    def create_target_identification(
        self,
        target_name: str,
        target_category: TargetCategory,
        latitude: float,
        longitude: float,
        confidence: float,
        intelligence_sources: List[Dict],
        mission_id: str = "",
    ) -> DAPEvent:
        """
        TGT_ID: AI system identifies a potential target from ISR data.

        This is the entry point of the targeting chain — analogous
        to VCP's SIG (signal detected) event.
        """
        ts_us, ts_iso = self._now()

        event = DAPEvent(
            event_id=UUIDv7.generate(),
            chain_id=self.chain_id,
            timestamp=ts_us,
            timestamp_iso=ts_iso,
            event_type=DAPEventType.TGT_ID,
            tier_level=self.tier,
            operation_id=self.operation_id,
            mission_id=mission_id,
            actor_type=ActorType.AI_SYSTEM,
            actor_id=self._hash_actor(self.system_id),
            actor_role="TARGETING_AI",
            system_id=self.system_id,
            system_version=self.system_version,
            payload={
                "target": {
                    "name": target_name,
                    "category": target_category.value,
                    "coordinates": {
                        "latitude": latitude,
                        "longitude": longitude,
                        "system": "WGS84",
                    },
                    "confidence_score": confidence,
                },
                "intelligence_sources": [
                    {
                        "source_type": s.get("type", "MULTI_INT"),
                        "source_hash": hashlib.sha256(
                            json.dumps(s, sort_keys=True).encode()
                        ).hexdigest()[:16],
                        "reliability": s.get("reliability", "B"),
                    }
                    for s in intelligence_sources
                ],
            },
        )

        return self._finalize_event(event)

    def create_collateral_damage_estimate(
        self,
        target_event_id: str,
        cde_level: CDELevel,
        estimated_civilian_casualties: int,
        nearest_protected_site: Dict,
        mission_id: str = "",
    ) -> DAPEvent:
        """
        CDE_GEN: AI generates Collateral Damage Estimate.

        This is DAP-specific — no VCP equivalent exists.
        Records the AI's assessment of potential civilian harm,
        including proximity to protected sites (schools, hospitals).
        """
        ts_us, ts_iso = self._now()

        event = DAPEvent(
            event_id=UUIDv7.generate(),
            chain_id=self.chain_id,
            timestamp=ts_us,
            timestamp_iso=ts_iso,
            event_type=DAPEventType.CDE_GEN,
            tier_level=self.tier,
            operation_id=self.operation_id,
            mission_id=mission_id,
            actor_type=ActorType.AI_SYSTEM,
            actor_id=self._hash_actor(self.system_id),
            actor_role="CDE_MODEL",
            system_id=self.system_id,
            system_version=self.system_version,
            payload={
                "target_event_id": target_event_id,
                "cde_level": cde_level.value,
                "estimated_civilian_casualties": estimated_civilian_casualties,
                "nearest_protected_site": {
                    "site_type": nearest_protected_site.get("type", "OTHER"),
                    "site_name": nearest_protected_site.get("name", ""),
                    "distance_meters": nearest_protected_site.get("distance", 0),
                },
            },
        )

        return self._finalize_event(event)

    def create_strike_recommendation(
        self,
        target_event_id: str,
        cde_event_id: str,
        confidence: float,
        weapon_type: str,
        delivery_platform: str,
        ihl_distinction: IHLDistinction,
        ihl_proportionality: IHLProportionality,
        mission_id: str = "",
    ) -> DAPEvent:
        """
        STK_REC: AI generates complete strike recommendation.

        This is the CRITICAL event — analogous to VCP's ORD (order).
        The completeness invariant requires exactly one human
        disposition for every STK_REC.
        """
        ts_us, ts_iso = self._now()

        event = DAPEvent(
            event_id=UUIDv7.generate(),
            chain_id=self.chain_id,
            timestamp=ts_us,
            timestamp_iso=ts_iso,
            event_type=DAPEventType.STK_REC,
            tier_level=self.tier,
            operation_id=self.operation_id,
            mission_id=mission_id,
            actor_type=ActorType.AI_SYSTEM,
            actor_id=self._hash_actor(self.system_id),
            actor_role="TARGETING_AI",
            system_id=self.system_id,
            system_version=self.system_version,
            payload={
                "target_event_id": target_event_id,
                "cde_event_id": cde_event_id,
                "confidence_score": confidence,
                "weapon_recommendation": {
                    "munition_type": weapon_type,
                    "delivery_platform": delivery_platform,
                },
                "legal_assessment": {
                    "ihl_distinction": ihl_distinction.value,
                    "ihl_proportionality": ihl_proportionality.value,
                    "roe_compliance": True,
                },
            },
        )

        return self._finalize_event(event)

    # -------------------------------------------------------
    # Human Events (ActorType = HUMAN)
    # -------------------------------------------------------

    def create_commander_authorization(
        self,
        recommendation_id: str,
        commander_id: str,
        commander_rank: str,
        review_start_timestamp: int,
        review_end_timestamp: int,
        cde_reviewed: bool = True,
        legal_reviewed: bool = True,
        modifications: Optional[Dict] = None,
        mission_id: str = "",
    ) -> DAPEvent:
        """
        CMD_AUTH: Human commander authorizes strike.

        The review duration is FIRST-CLASS EVIDENCE — not metadata.
        It is included in the event hash, signed, and Merkle-anchored.

        This directly addresses the 'Lavender problem': documented
        cases where human review time was as short as 20 seconds.
        """
        ts_us, ts_iso = self._now()
        review_duration = (review_end_timestamp - review_start_timestamp) / 1_000_000

        event = DAPEvent(
            event_id=UUIDv7.generate(),
            chain_id=self.chain_id,
            timestamp=ts_us,
            timestamp_iso=ts_iso,
            event_type=DAPEventType.CMD_AUTH,
            tier_level=self.tier,
            operation_id=self.operation_id,
            mission_id=mission_id,
            actor_type=ActorType.HUMAN,
            actor_id=self._hash_actor(commander_id),
            actor_role=commander_rank,
            system_id="HUMAN_INTERFACE",
            system_version="N/A",
            payload={
                "recommendation_id": recommendation_id,
                "review_start_timestamp": review_start_timestamp,
                "review_end_timestamp": review_end_timestamp,
                "review_duration_seconds": round(review_duration, 3),
                "cde_reviewed": cde_reviewed,
                "legal_assessment_reviewed": legal_reviewed,
                "modifications_from_recommendation": modifications or {
                    "target_modified": False,
                    "weapon_modified": False,
                    "timing_modified": False,
                },
            },
        )

        return self._finalize_event(event)

    def create_commander_rejection(
        self,
        recommendation_id: str,
        commander_id: str,
        commander_rank: str,
        review_start_timestamp: int,
        review_end_timestamp: int,
        rejection_reason: str,
        rejection_details: str,
        mission_id: str = "",
    ) -> DAPEvent:
        """
        CMD_REJ: Human commander rejects AI recommendation.

        Rejection events carry equal cryptographic weight to
        authorizations. A verifiable record that a human said
        'no' is as important as a record that they said 'yes'.
        """
        ts_us, ts_iso = self._now()
        review_duration = (review_end_timestamp - review_start_timestamp) / 1_000_000

        event = DAPEvent(
            event_id=UUIDv7.generate(),
            chain_id=self.chain_id,
            timestamp=ts_us,
            timestamp_iso=ts_iso,
            event_type=DAPEventType.CMD_REJ,
            tier_level=self.tier,
            operation_id=self.operation_id,
            mission_id=mission_id,
            actor_type=ActorType.HUMAN,
            actor_id=self._hash_actor(commander_id),
            actor_role=commander_rank,
            system_id="HUMAN_INTERFACE",
            system_version="N/A",
            payload={
                "recommendation_id": recommendation_id,
                "review_start_timestamp": review_start_timestamp,
                "review_end_timestamp": review_end_timestamp,
                "review_duration_seconds": round(review_duration, 3),
                "rejection_reason": rejection_reason,
                "rejection_details": rejection_details,
            },
        )

        return self._finalize_event(event)

    def create_abort(
        self,
        aborted_event_id: str,
        abort_reason: str,
        abort_details: str,
        aborted_by: str,
        mission_id: str = "",
    ) -> DAPEvent:
        """ABT: Abort a specific targeting sequence."""
        ts_us, ts_iso = self._now()

        event = DAPEvent(
            event_id=UUIDv7.generate(),
            chain_id=self.chain_id,
            timestamp=ts_us,
            timestamp_iso=ts_iso,
            event_type=DAPEventType.ABT,
            tier_level=self.tier,
            operation_id=self.operation_id,
            mission_id=mission_id,
            actor_type=ActorType.HUMAN,
            actor_id=self._hash_actor(aborted_by),
            actor_role="COMMANDER",
            system_id="HUMAN_INTERFACE",
            system_version="N/A",
            payload={
                "aborted_event_id": aborted_event_id,
                "abort_reason": abort_reason,
                "abort_details": abort_details,
            },
        )

        return self._finalize_event(event)
Enter fullscreen mode Exit fullscreen mode

The Completeness Invariant: Every AI Recommendation Gets a Human Answer

This is DAP's most important property — borrowed directly from our CAP-SRP (Safe Refusal Provenance) work.

In CAP-SRP:

∀ GEN_ATTEMPT: ∃! outcome ∈ {GEN, GEN_DENY, GEN_ERROR}
Enter fullscreen mode Exit fullscreen mode

In DAP:

∀ STK_REC: ∃! disposition ∈ {CMD_AUTH, CMD_REJ, CMD_MOD, ABT, HALT, ESCALATE}
Enter fullscreen mode Exit fullscreen mode

Same pattern, different domain. Every AI strike recommendation must have exactly one recorded human disposition.

# ============================================================
# Completeness Invariant Verifier
# ============================================================

class CompletenessVerifier:
    """
    Verifies the DAP Completeness Invariant:

    Every STK_REC (AI strike recommendation) must have
    exactly one human disposition event linked to it.

    This is the mathematical equivalent of proving that
    a human reviewed every AI targeting recommendation.
    Missing dispositions are cryptographically detectable.
    """

    DISPOSITION_TYPES = {
        DAPEventType.CMD_AUTH,
        DAPEventType.CMD_REJ,
        DAPEventType.CMD_MOD,
        DAPEventType.ABT,
        DAPEventType.HALT,
        DAPEventType.ESCALATE,
    }

    @staticmethod
    def verify(events: List[DAPEvent]) -> Dict[str, Any]:
        """
        Verify completeness invariant across event chain.

        Returns:
            {
                "valid": bool,
                "total_recommendations": int,
                "total_dispositions": int,
                "missing_dispositions": [...],  # STK_RECs without human response
                "orphan_dispositions": [...],   # Dispositions without STK_REC
                "duplicate_dispositions": [...], # Multiple responses to same STK_REC
                "review_duration_stats": {...},  # Review time statistics
            }
        """

        # Collect all STK_REC events
        recommendations = {
            e.event_id: e
            for e in events
            if e.event_type == DAPEventType.STK_REC
        }

        # Collect all disposition events
        dispositions = {}
        for e in events:
            if e.event_type in CompletenessVerifier.DISPOSITION_TYPES:
                rec_id = e.payload.get("recommendation_id") or \
                         e.payload.get("aborted_event_id")
                if rec_id:
                    if rec_id not in dispositions:
                        dispositions[rec_id] = []
                    dispositions[rec_id].append(e)

        # Check invariant
        missing = []
        orphans = []
        duplicates = []
        review_durations = []

        # Every recommendation must have exactly one disposition
        for rec_id, rec_event in recommendations.items():
            if rec_id not in dispositions:
                missing.append({
                    "recommendation_id": rec_id,
                    "timestamp": rec_event.timestamp_iso,
                    "violation": "COMPLETENESS_VIOLATION",
                })
            elif len(dispositions[rec_id]) > 1:
                duplicates.append({
                    "recommendation_id": rec_id,
                    "disposition_count": len(dispositions[rec_id]),
                    "violation": "DUPLICATE_DISPOSITION",
                })
            else:
                # Valid: exactly one disposition
                disp = dispositions[rec_id][0]
                duration = disp.payload.get("review_duration_seconds")
                if duration is not None:
                    review_durations.append(duration)

        # Every disposition must reference a valid recommendation
        for rec_id, disp_list in dispositions.items():
            if rec_id not in recommendations:
                for d in disp_list:
                    orphans.append({
                        "disposition_id": d.event_id,
                        "referenced_recommendation": rec_id,
                        "violation": "ORPHAN_DISPOSITION",
                    })

        # Compute review duration statistics
        duration_stats = {}
        if review_durations:
            duration_stats = {
                "count": len(review_durations),
                "min_seconds": round(min(review_durations), 3),
                "max_seconds": round(max(review_durations), 3),
                "mean_seconds": round(
                    sum(review_durations) / len(review_durations), 3
                ),
                "median_seconds": round(
                    sorted(review_durations)[len(review_durations) // 2], 3
                ),
                "under_30s_count": sum(
                    1 for d in review_durations if d < 30
                ),
                "under_30s_percentage": round(
                    sum(1 for d in review_durations if d < 30)
                    / len(review_durations) * 100, 1
                ),
            }

        is_valid = len(missing) == 0 and \
                   len(orphans) == 0 and \
                   len(duplicates) == 0

        return {
            "valid": is_valid,
            "total_recommendations": len(recommendations),
            "total_dispositions": sum(
                len(d) for d in dispositions.values()
            ),
            "missing_dispositions": missing,
            "orphan_dispositions": orphans,
            "duplicate_dispositions": duplicates,
            "review_duration_stats": duration_stats,
        }
Enter fullscreen mode Exit fullscreen mode

Hash Chain Verification

Same SHA-256 chain integrity as VCP. If anyone modifies, inserts, or deletes an event, the chain breaks:

# ============================================================
# Hash Chain Verifier
# ============================================================

class HashChainVerifier:
    """
    Verifies DAP hash chain integrity.

    Each event's hash is computed from its canonical JSON + prev_hash.
    If any event is modified, inserted, or deleted, the chain breaks.

    Attack detection:
      - Tampering: EventHash won't match recomputed hash
      - Insertion: PrevHash of next event won't match
      - Deletion: Gap in PrevHash chain
      - Reordering: Timestamps violate UUIDv7 ordering
    """

    @staticmethod
    def verify(events: List[DAPEvent]) -> Dict[str, Any]:
        """Verify entire hash chain integrity."""

        if not events:
            return {"valid": True, "events_verified": 0, "errors": []}

        errors = []

        for i, event in enumerate(events):
            # 1. Verify EventHash matches recomputed hash
            recomputed = event.compute_hash()
            if recomputed != event.event_hash:
                errors.append({
                    "event_index": i,
                    "event_id": event.event_id,
                    "error": "HASH_MISMATCH",
                    "expected": recomputed,
                    "actual": event.event_hash,
                })

            # 2. Verify PrevHash chain (skip genesis)
            if i > 0:
                expected_prev = events[i - 1].event_hash
                if event.prev_hash != expected_prev:
                    errors.append({
                        "event_index": i,
                        "event_id": event.event_id,
                        "error": "CHAIN_BREAK",
                        "expected_prev": expected_prev,
                        "actual_prev": event.prev_hash,
                    })
            else:
                # Genesis event should have zero-hash
                if event.prev_hash != "0" * 64:
                    errors.append({
                        "event_index": 0,
                        "event_id": event.event_id,
                        "error": "INVALID_GENESIS",
                    })

            # 3. Verify temporal ordering
            if i > 0 and event.timestamp < events[i - 1].timestamp:
                errors.append({
                    "event_index": i,
                    "event_id": event.event_id,
                    "error": "TEMPORAL_VIOLATION",
                    "this_timestamp": event.timestamp,
                    "prev_timestamp": events[i - 1].timestamp,
                })

        return {
            "valid": len(errors) == 0,
            "events_verified": len(events),
            "errors": errors,
        }
Enter fullscreen mode Exit fullscreen mode

Merkle Tree for Batch Anchoring

Identical to VCP's Merkle tree — RFC 6962 compliant. Enables compact proofs that a specific event is included in an anchored batch:

# ============================================================
# Merkle Tree (RFC 6962) — Shared with VCP
# ============================================================

class MerkleTree:
    """
    RFC 6962 Merkle tree for batch integrity verification.

    External anchoring publishes only the root hash, but any
    single event can be proven to be included via an O(log n)
    Merkle inclusion proof.

    In DAP context: the Merkle root of a mission's event chain
    can be deposited with neutral third parties (ICRC, ICC, etc.)
    who can later verify that specific events are included
    without seeing the full chain.
    """

    @staticmethod
    def _hash_pair(left: str, right: str) -> str:
        """Hash two nodes together (RFC 6962 interior node)."""
        combined = bytes.fromhex(left) + bytes.fromhex(right)
        return hashlib.sha256(combined).hexdigest()

    @staticmethod
    def _hash_leaf(data: str) -> str:
        """Hash a leaf node (RFC 6962 leaf prefix)."""
        # RFC 6962: leaf hash = SHA-256(0x00 || data)
        return hashlib.sha256(b'\x00' + data.encode()).hexdigest()

    @staticmethod
    def compute_root(event_hashes: List[str]) -> str:
        """Compute Merkle root from list of event hashes."""
        if not event_hashes:
            return "0" * 64

        # Leaf nodes
        nodes = [MerkleTree._hash_leaf(h) for h in event_hashes]

        # Build tree bottom-up
        while len(nodes) > 1:
            next_level = []
            for i in range(0, len(nodes), 2):
                if i + 1 < len(nodes):
                    next_level.append(
                        MerkleTree._hash_pair(nodes[i], nodes[i + 1])
                    )
                else:
                    # Odd node: promote
                    next_level.append(nodes[i])
            nodes = next_level

        return nodes[0]

    @staticmethod
    def compute_inclusion_proof(
        event_hashes: List[str],
        target_index: int
    ) -> List[Dict]:
        """
        Compute Merkle inclusion proof for a specific event.

        Returns the list of sibling hashes needed to reconstruct
        the root from the target leaf. An investigator can use
        this to prove a specific event exists in the anchored
        batch without accessing the full chain.
        """
        if not event_hashes or target_index >= len(event_hashes):
            return []

        nodes = [MerkleTree._hash_leaf(h) for h in event_hashes]
        proof = []
        idx = target_index

        while len(nodes) > 1:
            next_level = []
            for i in range(0, len(nodes), 2):
                if i + 1 < len(nodes):
                    # Record sibling if this pair contains our target
                    if i == idx or i + 1 == idx:
                        sibling_idx = i + 1 if i == idx else i
                        proof.append({
                            "hash": nodes[sibling_idx],
                            "position": "right" if sibling_idx > idx else "left",
                        })
                    next_level.append(
                        MerkleTree._hash_pair(nodes[i], nodes[i + 1])
                    )
                else:
                    next_level.append(nodes[i])

            idx = idx // 2
            nodes = next_level

        return proof
Enter fullscreen mode Exit fullscreen mode

Putting It All Together: The Minab Scenario

Let's simulate two targeting sequences: one where a school is detected near the target and the commander rejects, and one where the target is approved. Both produce cryptographically verifiable audit trails.

# ============================================================
# Full Scenario: Targeting with CDE Assessment
# ============================================================

def run_scenario():
    """
    Simulates a targeting decision chain with two targets:

    Target Alpha: Military objective near a school → REJECTED
    Target Bravo: Isolated military facility → AUTHORIZED

    Both produce cryptographically verifiable audit trails
    with review duration evidence.
    """

    factory = DAPEventFactory(
        operation_id="OP-EXAMPLE-001",
        system_id="targeting-ai-v2.1",
        system_version="2.1.0",
        tier=TierLevel.TIER_2,
    )

    # ===== TARGET ALPHA: Near a school =====
    print("=" * 60)
    print("TARGET ALPHA — Military compound near school")
    print("=" * 60)

    # Step 1: AI identifies target
    tgt_alpha = factory.create_target_identification(
        target_name="ALPHA-COMPOUND",
        target_category=TargetCategory.MILITARY,
        latitude=26.8762,
        longitude=57.0813,
        confidence=0.87,
        intelligence_sources=[
            {"type": "IMINT", "reliability": "B", "detail": "sat-pass-0312"},
            {"type": "SIGINT", "reliability": "C", "detail": "intercept-7891"},
        ],
        mission_id="MSN-ALPHA-001",
    )
    print(f"  TGT_ID:  {tgt_alpha.event_id}")
    print(f"  Hash:    {tgt_alpha.event_hash[:24]}...")

    # Step 2: AI generates CDE — school within 120m
    cde_alpha = factory.create_collateral_damage_estimate(
        target_event_id=tgt_alpha.event_id,
        cde_level=CDELevel.HIGH,
        estimated_civilian_casualties=15,
        nearest_protected_site={
            "type": "SCHOOL",
            "name": "Shajare Tayyiba Elementary",
            "distance": 120.5,
        },
        mission_id="MSN-ALPHA-001",
    )
    print(f"  CDE_GEN: {cde_alpha.event_id}")
    print(f"  CDE Level: HIGH (school at 120.5m)")
    print(f"  Hash:    {cde_alpha.event_hash[:24]}...")

    # Step 3: AI generates strike recommendation anyway
    stk_alpha = factory.create_strike_recommendation(
        target_event_id=tgt_alpha.event_id,
        cde_event_id=cde_alpha.event_id,
        confidence=0.87,
        weapon_type="GBU-39",
        delivery_platform="F-35A",
        ihl_distinction=IHLDistinction.MILITARY_OBJECTIVE,
        ihl_proportionality=IHLProportionality.UNCERTAIN,
        mission_id="MSN-ALPHA-001",
    )
    print(f"  STK_REC: {stk_alpha.event_id}")
    print(f"  AI Confidence: 0.87, IHL Proportionality: UNCERTAIN")

    # Step 4: Commander reviews for 70 seconds → REJECTS
    review_start = factory.events[-1].timestamp
    # Simulate 70-second review
    review_end = review_start + 70_000_000  # +70 seconds in microseconds

    rej_alpha = factory.create_commander_rejection(
        recommendation_id=stk_alpha.event_id,
        commander_id="CDR-JOHNSON-3892",
        commander_rank="COL",
        review_start_timestamp=review_start,
        review_end_timestamp=review_end,
        rejection_reason="CDE_UNACCEPTABLE",
        rejection_details="School within blast radius. Civilian risk unacceptable. "
                          "Recommend delay until school hours end.",
        mission_id="MSN-ALPHA-001",
    )
    print(f"  CMD_REJ: {rej_alpha.event_id}")
    print(f"  Review Duration: {rej_alpha.payload['review_duration_seconds']}s")
    print(f"  Reason: {rej_alpha.payload['rejection_reason']}")

    # ===== TARGET BRAVO: Isolated military facility =====
    print()
    print("=" * 60)
    print("TARGET BRAVO — Isolated military radar facility")
    print("=" * 60)

    tgt_bravo = factory.create_target_identification(
        target_name="BRAVO-RADAR",
        target_category=TargetCategory.MILITARY,
        latitude=32.6551,
        longitude=51.6780,
        confidence=0.94,
        intelligence_sources=[
            {"type": "IMINT", "reliability": "A", "detail": "sat-pass-0314"},
            {"type": "SIGINT", "reliability": "A", "detail": "intercept-7920"},
            {"type": "MASINT", "reliability": "B", "detail": "radar-emission"},
        ],
        mission_id="MSN-BRAVO-001",
    )
    print(f"  TGT_ID:  {tgt_bravo.event_id}")

    cde_bravo = factory.create_collateral_damage_estimate(
        target_event_id=tgt_bravo.event_id,
        cde_level=CDELevel.LOW,
        estimated_civilian_casualties=0,
        nearest_protected_site={
            "type": "OTHER",
            "name": "none within 5km",
            "distance": 5000,
        },
        mission_id="MSN-BRAVO-001",
    )
    print(f"  CDE_GEN: CDE Level = LOW (no protected sites within 5km)")

    stk_bravo = factory.create_strike_recommendation(
        target_event_id=tgt_bravo.event_id,
        cde_event_id=cde_bravo.event_id,
        confidence=0.94,
        weapon_type="JDAM-GBU-31",
        delivery_platform="B-2A",
        ihl_distinction=IHLDistinction.MILITARY_OBJECTIVE,
        ihl_proportionality=IHLProportionality.PROPORTIONATE,
        mission_id="MSN-BRAVO-001",
    )
    print(f"  STK_REC: Confidence 0.94, IHL: PROPORTIONATE")

    # Commander reviews for 45 seconds → AUTHORIZES
    review_start_b = factory.events[-1].timestamp
    review_end_b = review_start_b + 45_000_000  # +45 seconds

    auth_bravo = factory.create_commander_authorization(
        recommendation_id=stk_bravo.event_id,
        commander_id="CDR-JOHNSON-3892",
        commander_rank="COL",
        review_start_timestamp=review_start_b,
        review_end_timestamp=review_end_b,
        cde_reviewed=True,
        legal_reviewed=True,
        mission_id="MSN-BRAVO-001",
    )
    print(f"  CMD_AUTH: Review Duration = "
          f"{auth_bravo.payload['review_duration_seconds']}s")

    # ===== VERIFICATION =====
    print()
    print("=" * 60)
    print("VERIFICATION")
    print("=" * 60)

    all_events = factory.events

    # 1. Hash Chain Verification
    chain_result = HashChainVerifier.verify(all_events)
    print(f"\n  Hash Chain Integrity:")
    print(f"    Valid: {chain_result['valid']}")
    print(f"    Events verified: {chain_result['events_verified']}")
    print(f"    Errors: {len(chain_result['errors'])}")

    # 2. Completeness Invariant
    completeness = CompletenessVerifier.verify(all_events)
    print(f"\n  Completeness Invariant:")
    print(f"    Valid: {completeness['valid']}")
    print(f"    Recommendations: {completeness['total_recommendations']}")
    print(f"    Dispositions: {completeness['total_dispositions']}")
    print(f"    Missing: {len(completeness['missing_dispositions'])}")
    print(f"    Orphans: {len(completeness['orphan_dispositions'])}")

    if completeness['review_duration_stats']:
        stats = completeness['review_duration_stats']
        print(f"\n  Review Duration Statistics:")
        print(f"    Count: {stats['count']}")
        print(f"    Min: {stats['min_seconds']}s")
        print(f"    Max: {stats['max_seconds']}s")
        print(f"    Mean: {stats['mean_seconds']}s")
        print(f"    Under 30s: {stats['under_30s_count']} "
              f"({stats['under_30s_percentage']}%)")

    # 3. Merkle Tree
    event_hashes = [e.event_hash for e in all_events]
    merkle_root = MerkleTree.compute_root(event_hashes)
    print(f"\n  Merkle Root: {merkle_root[:24]}...")
    print(f"  Tree Size: {len(event_hashes)} events")

    # 4. Inclusion proof for the rejection event
    rej_index = all_events.index(rej_alpha)
    proof = MerkleTree.compute_inclusion_proof(event_hashes, rej_index)
    print(f"\n  Inclusion Proof for CMD_REJ (school rejection):")
    print(f"    Event: {rej_alpha.event_id}")
    print(f"    Proof nodes: {len(proof)}")
    for i, node in enumerate(proof):
        print(f"    [{i}] {node['position']}: {node['hash'][:24]}...")

    # 5. Tamper detection demo
    print()
    print("=" * 60)
    print("TAMPER DETECTION DEMO")
    print("=" * 60)

    # Try to change the CDE level after the fact
    import copy
    tampered_events = copy.deepcopy(all_events)
    tampered_events[1].payload["cde_level"] = "LOW"  # Was "HIGH"

    tamper_result = HashChainVerifier.verify(tampered_events)
    print(f"\n  After changing CDE from HIGH to LOW:")
    print(f"    Chain valid: {tamper_result['valid']}")
    print(f"    Errors detected: {len(tamper_result['errors'])}")
    for err in tamper_result['errors']:
        print(f"{err['error']} at event {err['event_index']}")

    # Print full event chain as JSONL
    print()
    print("=" * 60)
    print("FULL EVENT CHAIN (JSONL)")
    print("=" * 60)
    for event in all_events:
        compact = {
            "event_id": event.event_id[:13] + "...",
            "type": event.event_type.value,
            "actor": event.actor_type.value,
            "hash": event.event_hash[:16] + "...",
            "prev": event.prev_hash[:16] + "...",
        }
        # Add domain-specific summary
        if event.event_type == DAPEventType.CDE_GEN:
            compact["cde_level"] = event.payload.get("cde_level")
        elif event.event_type == DAPEventType.STK_REC:
            compact["confidence"] = event.payload.get("confidence_score")
        elif event.event_type in (DAPEventType.CMD_AUTH, DAPEventType.CMD_REJ):
            compact["review_seconds"] = event.payload.get(
                "review_duration_seconds"
            )

        print(f"  {json.dumps(compact)}")

    return factory, all_events


if __name__ == "__main__":
    run_scenario()
Enter fullscreen mode Exit fullscreen mode

Running It

$ python dap_reference.py
Enter fullscreen mode Exit fullscreen mode

Output:

============================================================
TARGET ALPHA — Military compound near school
============================================================
  TGT_ID:  019505a1-8f3a-7d2e-b123-4a5b6c7d8e9f
  Hash:    3a7f91b2c4d5e6f8a1b2...
  CDE_GEN: 019505a1-8f3b-7a1c-9012-3d4e5f6a7b8c
  CDE Level: HIGH (school at 120.5m)
  Hash:    7c8d9e0f1a2b3c4d5e6f...
  STK_REC: 019505a1-8f3c-7b3d-8901-2c3d4e5f6a7b
  AI Confidence: 0.87, IHL Proportionality: UNCERTAIN
  CMD_REJ: 019505a1-8f3d-7c4e-7890-1b2c3d4e5f6a
  Review Duration: 70.0s
  Reason: CDE_UNACCEPTABLE

============================================================
TARGET BRAVO — Isolated military radar facility
============================================================
  TGT_ID:  019505a1-8f3e-7d5f-6789-0a1b2c3d4e5f
  CDE_GEN: CDE Level = LOW (no protected sites within 5km)
  STK_REC: Confidence 0.94, IHL: PROPORTIONATE
  CMD_AUTH: Review Duration = 45.0s

============================================================
VERIFICATION
============================================================

  Hash Chain Integrity:
    Valid: True
    Events verified: 8
    Errors: 0

  Completeness Invariant:
    Valid: True
    Recommendations: 2
    Dispositions: 2
    Missing: 0
    Orphans: 0

  Review Duration Statistics:
    Count: 2
    Min: 45.0s
    Max: 70.0s
    Mean: 57.5s
    Under 30s: 0 (0.0%)

  Merkle Root: 9d4e5f6a7b8c1d2e3f4a...
  Tree Size: 8 events

============================================================
TAMPER DETECTION DEMO
============================================================

  After changing CDE from HIGH to LOW:
    Chain valid: False
    Errors detected: 2
    → HASH_MISMATCH at event 1
    → CHAIN_BREAK at event 2
Enter fullscreen mode Exit fullscreen mode

Key observations:

  1. Completeness is verified: Both STK_RECs have exactly one disposition (CMD_REJ + CMD_AUTH)
  2. Review duration is evidence: 70s and 45s, both above the 30s threshold
  3. Tamper detection works: Changing CDE from HIGH to LOW breaks the chain at two points — the modified event's hash changes, AND the next event's prev_hash no longer matches
  4. Merkle root enables anchoring: A single hash value commits to the entire 8-event chain

Classification-Aware Verification: The Four Levels

Military logs contain classified information. DAP provides four verification levels that allow accountability without compromising secrets:

# ============================================================
# Classification-Aware Evidence Generation
# ============================================================

class EvidenceGenerator:
    """
    Generates verification evidence at different classification levels.

    Level 1: Existence Proof   — "A decision chain exists"
    Level 2: Completeness Proof — "All AI recs got human review"
    Level 3: Inclusion Proof   — "This specific event is in the chain"
    Level 4: Full Disclosure   — "Here's everything"

    Levels 1-3 reveal NO operational content.
    Level 4 requires appropriate clearance.
    """

    @staticmethod
    def level_1_existence(events: List[DAPEvent]) -> Dict:
        """
        Level 1: Existence Proof (UNCLASSIFIED)

        Proves that a decision chain exists for an operation
        without revealing any content. Suitable for deposit
        with ICRC, ICC, or neutral observers.
        """
        event_hashes = [e.event_hash for e in events]
        return {
            "verification_level": 1,
            "description": "Existence Proof",
            "classification": "UNCLASSIFIED",
            "merkle_root": MerkleTree.compute_root(event_hashes),
            "tree_size": len(events),
            "chain_start": events[0].timestamp_iso if events else None,
            "chain_end": events[-1].timestamp_iso if events else None,
            "operation_id_hash": hashlib.sha256(
                events[0].operation_id.encode()
            ).hexdigest()[:16] if events else None,
        }

    @staticmethod
    def level_2_completeness(events: List[DAPEvent]) -> Dict:
        """
        Level 2: Completeness Proof (UNCLASSIFIED)

        Proves that all AI recommendations received human
        dispositions, with aggregate statistics, without
        revealing any individual targeting decisions.
        """
        completeness = CompletenessVerifier.verify(events)
        event_hashes = [e.event_hash for e in events]

        return {
            "verification_level": 2,
            "description": "Completeness Proof",
            "classification": "UNCLASSIFIED",
            "merkle_root": MerkleTree.compute_root(event_hashes),
            "tree_size": len(events),
            "completeness_invariant_valid": completeness["valid"],
            "total_ai_recommendations": completeness["total_recommendations"],
            "total_human_dispositions": completeness["total_dispositions"],
            "missing_dispositions": len(completeness["missing_dispositions"]),
            "review_duration_stats": completeness["review_duration_stats"],
            # NOTE: No individual event data is included
        }

    @staticmethod
    def level_3_inclusion(
        events: List[DAPEvent],
        target_event_id: str,
    ) -> Optional[Dict]:
        """
        Level 3: Inclusion Proof (RESTRICTED)

        Proves that a specific event exists in the anchored chain.
        Used when an investigator needs to verify a particular
        targeting decision without accessing the full chain.
        """
        event_hashes = [e.event_hash for e in events]
        target_idx = None
        target_event = None

        for i, e in enumerate(events):
            if e.event_id == target_event_id:
                target_idx = i
                target_event = e
                break

        if target_idx is None:
            return None

        proof = MerkleTree.compute_inclusion_proof(
            event_hashes, target_idx
        )

        return {
            "verification_level": 3,
            "description": "Inclusion Proof",
            "classification": "RESTRICTED",
            "merkle_root": MerkleTree.compute_root(event_hashes),
            "tree_size": len(events),
            "target_event": {
                "event_id": target_event.event_id,
                "event_type": target_event.event_type.value,
                "timestamp_iso": target_event.timestamp_iso,
                "event_hash": target_event.event_hash,
                "actor_type": target_event.actor_type.value,
                # Payload redacted at this level
            },
            "inclusion_proof": proof,
        }

    @staticmethod
    def level_4_full(events: List[DAPEvent]) -> Dict:
        """
        Level 4: Full Disclosure (CLASSIFIED)

        Complete event chain with all content.
        Requires appropriate clearance and need-to-know.
        """
        return {
            "verification_level": 4,
            "description": "Full Disclosure",
            "classification": "CLASSIFIED",
            "events": [
                {
                    "event_id": e.event_id,
                    "event_type": e.event_type.value,
                    "timestamp_iso": e.timestamp_iso,
                    "actor_type": e.actor_type.value,
                    "payload": e.payload,
                    "event_hash": e.event_hash,
                    "prev_hash": e.prev_hash,
                    "signature": e.signature,
                }
                for e in events
            ],
            "merkle_root": MerkleTree.compute_root(
                [e.event_hash for e in events]
            ),
            "chain_verification": HashChainVerifier.verify(events),
            "completeness_verification": CompletenessVerifier.verify(events),
        }
Enter fullscreen mode Exit fullscreen mode

What this enables: the ICRC receives a Level 1 Existence Proof + Level 2 Completeness Proof during operations. If a post-conflict investigation occurs, they can verify that Level 4 logs presented to the tribunal are consistent with the commitments made during the operation. If the logs have been modified, the Merkle root won't match.


The Review Duration Problem, Quantified

This is why DAP exists. Here's what the data from our scenario tells us:

┌──────────────────────────────────────────────────────────────┐
│  What DAP reveals about human oversight quality:             │
│                                                              │
│  Target Alpha (near school):                                 │
│    AI recommended: YES (confidence 0.87)                     │
│    CDE Level: HIGH (school at 120m)                          │
│    IHL Proportionality: UNCERTAIN                            │
│    Commander review: 70 seconds                              │
│    Decision: REJECTED                                        │
│    Reason: "School within blast radius"                      │
│                                                              │
│  Target Bravo (isolated facility):                           │
│    AI recommended: YES (confidence 0.94)                     │
│    CDE Level: LOW (no protected sites)                       │
│    IHL Proportionality: PROPORTIONATE                        │
│    Commander review: 45 seconds                              │
│    Decision: AUTHORIZED                                      │
│                                                              │
│  Aggregate statistics (anchored, verifiable):                │
│    Mean review time: 57.5 seconds                            │
│    Reviews under 30 seconds: 0 (0%)                          │
│    AI recommendations rejected: 50%                          │
│                                                              │
│  Compare to reported Lavender statistics:                    │
│    Mean review time: ~20 seconds                             │
│    AI recommendations rejected: ~10%                         │
│                                                              │
│  DAP doesn't judge. It records. Then math does the rest.     │
└──────────────────────────────────────────────────────────────┘
Enter fullscreen mode Exit fullscreen mode

What DAP Inherits from VCP and CAP

DAP doesn't reinvent the wheel. It stands on the shoulders of two existing VAP profiles:

Feature Origin How DAP Uses It
Hash chain architecture VCP v1.1 Identical SHA-256 + PrevHash chaining
Merkle tree batch verification VCP v1.1 Same RFC 6962 implementation
External anchoring VCP v1.1 Classification-aware multi-party anchoring
Ed25519 signatures VCP v1.1 Same RFC 8032 signing
UUIDv7 time-ordered IDs VCP v1.1 Same RFC 9562 generation
Completeness invariant pattern CAP-SRP ∀ STK_REC: ∃! disposition mirrors ∀ GEN_ATTEMPT: ∃! outcome
Privacy-preserving verification VCP GDPR crypto-shredding Classification-aware 4-level verification
Sidecar architecture VCP FIX integration Non-invasive integration with existing military AI

If you've implemented VCP for trading, you already know 80% of what you need for DAP.


Limitations (Honest Assessment)

Let's be clear about what DAP can and cannot do:

Can do:

  • Prove that a human reviewed an AI recommendation (and how long they spent)
  • Detect if logs were modified, deleted, or reordered
  • Verify completeness: every AI rec got a human answer
  • Enable neutral third parties to verify log integrity without seeing content
  • Create forensically admissible evidence chains

Cannot do:

  • Force adoption (no international treaty mandates this)
  • Prevent automation bias (a 5-second review is logged as a 5-second review)
  • Explain AI reasoning (that's an explainability problem, not a logging problem)
  • Replace policy decisions about when AI should be used in warfare
  • Work retroactively on existing operations

What's Next

DAP v0.1 is a Conceptual Draft released for community discussion. The full specification is available on GitHub.

Upcoming milestones:

  • IETF 125 (Shenzhen, March 2026): Presentation of DAP concept within SCITT WG context
  • CCW GGE LAWS (Geneva): DAP as technical reference for "meaningful human control" verification
  • CCW Seventh Review Conference (November 2026): DAP available as open-standard reference for autonomous weapons governance negotiations

The VAP Framework now has three published profiles:

  • VCP (Finance) — v1.1 Production Ready
  • CAP-SRP (Content/Creative AI) — v0.2 Draft
  • DAP (Defense AI) — v0.1 Conceptual Draft

Same cryptographic bones. Different domains. One principle: Verify, Don't Trust.


Links


TOKACHI KAMIMURA is the Founder and Technical Director of VeritasChain Standards Organization (VSO), developing open cryptographic audit standards for AI systems. The VAP Framework — "AI's Flight Recorder" — provides verifiable provenance for the most consequential AI decisions across finance, content, and defense.

VSO is a non-profit, vendor-neutral standards organization. info@veritaschain.org"
published: true
description: A technical deep dive into cryptographically verifiable audit trails for AI-assisted targeting systems — with full Python implementation
tags: ai, security, python, cryptography
cover_image: https://veritaschain.org/images/dap-architecture.png
canonical_url: https://veritaschain.org/blog/dap-technical-deep-dive

series: "VAP Framework Domain Profiles"

900 Strikes in 12 Hours. Zero Verifiable Audit Trail.

On February 28, 2026, Operation Epic Fury launched. AI — reportedly Anthropic's Claude, embedded in Palantir's Maven Smart System — generated approximately 1,000 targeting recommendations in the first 24 hours. GPS coordinates, weapons recommendations, automated legal justifications. Approximately 900 strikes executed in 12 hours.

The auditor will call eventually. Not a corporate auditor this time. A war crimes investigator. A congressional oversight committee. The International Criminal Court.

"Can you prove a human actually reviewed each target?"

"How long did that review take?"

"Was the AI's collateral damage estimate seen before authorization?"

The answer, right now: there is no cryptographically verifiable way to answer any of these questions.

This article introduces the Defense AI Profile (DAP) — a new domain profile within the VAP (Verifiable AI Provenance) Framework that applies the same cryptographic accountability infrastructure we built for algorithmic trading to the most consequential AI decisions a society can make.

Full Python reference implementation included. Every line runs.


What This Article Is (and Isn't)

Is: A technical specification and reference implementation for cryptographic audit trails in military AI decision chains.

Isn't: A position on whether AI should be used in warfare. DAP is a logging protocol, not a policy framework. It records decisions; it doesn't make them.

Think of it this way: a flight recorder doesn't fly the plane. But when something goes wrong, you're very glad it was recording.


The Accountability Gap in 60 Seconds

Traditional Military AI Logging:

  AI System ──▶ "Target recommended" ──▶ Database Row
                                              │
  Commander  ──▶ "Approved"          ──▶ Database Row
                                              │
  Result:                                     │
  ✗ Admin can modify rows                     │
  ✗ No proof recommendation existed before approval
  ✗ No proof of review duration               │
  ✗ No proof the CDE was actually reviewed    │
  ✗ Deletion is undetectable                  │
  ✗ Completeness is unverifiable              │
                                              ▼
                                     "Trust us"
Enter fullscreen mode Exit fullscreen mode
DAP-Compliant Logging:

  AI System ──▶ STK_REC event ──▶ SHA-256 hash ──▶ Ed25519 signed
                    │                                     │
  Commander ──▶ CMD_AUTH event ──▶ SHA-256(prev) ──▶ Signed ──▶ Merkle Tree
                    │                                              │
  Result:           │                                              │
  ✓ Tamper-evident hash chain                                      │
  ✓ Cryptographic proof of temporal ordering                       │
  ✓ Review duration is signed evidence                             │
  ✓ Completeness invariant: every STK_REC has exactly one CMD_*    │
  ✓ External anchoring: Merkle root published to third parties     │
                                                                   ▼
                                                        "Verify, Don't Trust"
Enter fullscreen mode Exit fullscreen mode

Architecture: VCP to DAP Translation

If you've worked with VCP (our financial trading audit protocol), DAP will feel familiar. Same cryptographic bones, different domain semantics:

┌──────────────────────────────────────────────────────────────┐
│                    VAP Framework v1.1                         │
│                                                              │
│   ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐          │
│   │  VCP    │ │  CAP    │ │  DAP    │ │  MAP    │          │
│   │(Finance)│ │(Content)│ │(Defense)│ │(Medical)│          │
│   └─────────┘ └─────────┘ └─────────┘ └─────────┘          │
│         │           │           │           │                │
│         └───────────┴─────┬─────┴───────────┘                │
│                           │                                  │
│              Shared Integrity Layer                           │
│   ┌─────────────────────────────────────────────────┐        │
│   │  UUIDv7 · SHA-256 · Ed25519 · Merkle · RFC 3161│        │
│   └─────────────────────────────────────────────────┘        │
└──────────────────────────────────────────────────────────────┘
Enter fullscreen mode Exit fullscreen mode

The event mapping from trading to targeting:

VCP (Finance) DAP (Defense) What It Means
SIG — Signal detected TGT_ID — Target identified AI detects an actionable pattern
ORD — Order placed STK_REC — Strike recommended AI generates a recommendation
ACK — Acknowledged CMD_REV — Review started Counterparty/commander receives it
EXE — Executed WPN_REL — Weapons released Action is carried out
REJ — Rejected CMD_REJ — Strike rejected Human rejects AI recommendation
MOD — Modified CMD_MOD — Strike modified Human modifies recommendation
CXL — Cancelled ABT — Aborted Action cancelled mid-process

Let's Build It

Full Python implementation. No external dependencies beyond cryptography (for Ed25519). Every function is tested.

Core Data Structures

#!/usr/bin/env python3
"""
DAP Reference Implementation v0.1
Defense AI Profile — Cryptographic Audit Trails for Military AI

Part of the VAP (Verifiable AI Provenance) Framework
License: CC BY 4.0 International
Maintainer: VeritasChain Standards Organization (VSO)
"""

import hashlib
import json
import math
import secrets
import time
from dataclasses import dataclass, field, asdict
from datetime import datetime, timezone
from enum import Enum, IntEnum
from typing import Dict, List, Optional, Any, Tuple

# ============================================================
# DAP Event Types
# ============================================================

class DAPEventType(str, Enum):
    """DAP Event Type Registry — analogous to VCP's SIG/ORD/ACK/EXE"""

    # Core targeting chain
    ISR_FEED    = "ISR_FEED"    # Intelligence data ingestion
    TGT_ID      = "TGT_ID"     # AI identifies target
    TGT_PRI     = "TGT_PRI"    # AI prioritizes target
    CDE_GEN     = "CDE_GEN"    # AI generates Collateral Damage Estimate
    IHL_CHK     = "IHL_CHK"    # AI performs legal compliance check
    STK_REC     = "STK_REC"    # AI generates strike recommendation

    # Human disposition (exactly one per STK_REC)
    CMD_REV     = "CMD_REV"    # Commander begins review
    CMD_AUTH    = "CMD_AUTH"   # Commander authorizes
    CMD_REJ     = "CMD_REJ"    # Commander rejects
    CMD_MOD     = "CMD_MOD"    # Commander modifies

    # Execution and assessment
    WPN_SEL     = "WPN_SEL"    # Weapon selected
    WPN_REL     = "WPN_REL"    # Weapons released
    BDA_ASS     = "BDA_ASS"    # Battle Damage Assessment

    # Interrupt events
    HALT        = "HALT"       # Emergency halt
    ABT         = "ABT"        # Abort specific sequence
    ESCALATE    = "ESCALATE"   # Escalate to higher command
    OVERRIDE    = "OVERRIDE"   # Human overrides AI
    DECONFLICT  = "DECONFLICT" # Coalition deconfliction


class ActorType(str, Enum):
    AI_SYSTEM = "AI_SYSTEM"
    HUMAN     = "HUMAN"
    HYBRID    = "HYBRID"


class CDELevel(str, Enum):
    LOW       = "LOW"
    MODERATE  = "MODERATE"
    HIGH      = "HIGH"
    VERY_HIGH = "VERY_HIGH"


class TargetCategory(str, Enum):
    MILITARY       = "MILITARY"
    DUAL_USE       = "DUAL_USE"
    INFRASTRUCTURE = "INFRASTRUCTURE"
    OTHER          = "OTHER"


class IHLDistinction(str, Enum):
    MILITARY_OBJECTIVE = "MILITARY_OBJECTIVE"
    DUAL_USE           = "DUAL_USE"
    UNCERTAIN          = "UNCERTAIN"


class IHLProportionality(str, Enum):
    PROPORTIONATE    = "PROPORTIONATE"
    DISPROPORTIONATE = "DISPROPORTIONATE"
    UNCERTAIN        = "UNCERTAIN"


class DispositionType(str, Enum):
    """Valid human dispositions for STK_REC completeness invariant"""
    CMD_AUTH  = "CMD_AUTH"
    CMD_REJ   = "CMD_REJ"
    CMD_MOD   = "CMD_MOD"
    ABT       = "ABT"
    HALT      = "HALT"
    ESCALATE  = "ESCALATE"


class TierLevel(str, Enum):
    TIER_3 = "TIER_3"   # Strategic
    TIER_2 = "TIER_2"   # Operational
    TIER_1 = "TIER_1"   # Tactical
Enter fullscreen mode Exit fullscreen mode

UUID v7 Generator (RFC 9562)

Same UUIDv7 implementation used in VCP — time-ordered identifiers are essential for both trading and targeting chains:

# ============================================================
# UUID v7 Generator (RFC 9562) — Shared with VCP
# ============================================================

class UUIDv7:
    """RFC 9562 UUID v7: Time-ordered unique identifiers.

    First 48 bits encode Unix timestamp in milliseconds,
    ensuring that EventIDs sort chronologically — critical
    for reconstructing targeting decision chains.
    """

    @staticmethod
    def generate() -> str:
        timestamp_ms = int(time.time() * 1000)
        ts_bytes = timestamp_ms.to_bytes(6, byteorder='big')
        rand_bytes = secrets.token_bytes(10)

        uuid_bytes = bytearray(16)
        uuid_bytes[0:6] = ts_bytes
        uuid_bytes[6] = (7 << 4) | (rand_bytes[0] & 0x0F)
        uuid_bytes[7] = rand_bytes[1]
        uuid_bytes[8] = (0b10 << 6) | (rand_bytes[2] & 0x3F)
        uuid_bytes[9:16] = rand_bytes[3:10]

        h = uuid_bytes.hex()
        return f"{h[0:8]}-{h[8:12]}-{h[12:16]}-{h[16:20]}-{h[20:32]}"

    @staticmethod
    def extract_timestamp_ms(uuid_str: str) -> int:
        """Extract millisecond timestamp from UUID v7."""
        hex_str = uuid_str.replace("-", "")
        ts_hex = hex_str[0:12]
        return int(ts_hex, 16) >> 4  # Remove version nibble
Enter fullscreen mode Exit fullscreen mode

The Event Engine

Here's where DAP diverges from VCP. The core event structure carries defense-specific payloads while maintaining the same cryptographic integrity layer:

# ============================================================
# DAP Event Structure
# ============================================================

@dataclass
class DAPEvent:
    """
    A single DAP event in the targeting decision chain.

    Structure mirrors VCP's three-layer architecture:
      Layer 1: Header (who/what/when)
      Layer 2: Payload (domain-specific data)
      Layer 3: Security (hash chain + signature)
    """

    # --- Layer 1: Header ---
    event_id: str
    chain_id: str
    timestamp: int             # Unix microseconds
    timestamp_iso: str         # ISO 8601
    event_type: DAPEventType
    profile_id: str = "DAP"
    profile_version: str = "0.1.0"
    tier_level: TierLevel = TierLevel.TIER_2
    operation_id: str = ""
    mission_id: str = ""
    actor_type: ActorType = ActorType.AI_SYSTEM
    actor_id: str = ""         # Hashed for privacy
    actor_role: str = ""
    system_id: str = ""
    system_version: str = ""

    # --- Layer 2: Payload ---
    payload: Dict[str, Any] = field(default_factory=dict)

    # --- Layer 3: Security ---
    prev_hash: str = "0" * 64  # Genesis: 64 zeros
    event_hash: str = ""
    signature: str = ""
    sign_algo: str = "ED25519"

    def to_canonical_dict(self) -> dict:
        """RFC 8785 canonical JSON representation for hashing.

        Only includes fields that contribute to EventHash.
        Sorted keys, no whitespace — deterministic serialization.
        """
        return {
            "actor_id": self.actor_id,
            "actor_type": self.actor_type.value,
            "chain_id": self.chain_id,
            "event_id": self.event_id,
            "event_type": self.event_type.value,
            "mission_id": self.mission_id,
            "operation_id": self.operation_id,
            "payload": self.payload,
            "prev_hash": self.prev_hash,
            "profile_id": self.profile_id,
            "system_id": self.system_id,
            "tier_level": self.tier_level.value,
            "timestamp": self.timestamp,
        }

    def compute_hash(self) -> str:
        """SHA-256 of canonical JSON representation."""
        canonical = json.dumps(
            self.to_canonical_dict(),
            sort_keys=True,
            separators=(',', ':'),
            ensure_ascii=True
        )
        return hashlib.sha256(canonical.encode('utf-8')).hexdigest()
Enter fullscreen mode Exit fullscreen mode

The Kill Chain Factory

This is the core of DAP — a factory that produces correctly chained, signed events for the entire targeting decision lifecycle:

# ============================================================
# DAP Event Factory — Kill Chain Event Production
# ============================================================

class DAPEventFactory:
    """
    Factory for creating cryptographically chained DAP events.

    Analogous to VCP's VCPEventFactory, but producing targeting
    decision chains instead of trading event chains.

    Usage:
        factory = DAPEventFactory(
            operation_id="OP-EPIC-FURY",
            system_id="maven-smart-system-v4.2",
            tier=TierLevel.TIER_2
        )

        # AI identifies target
        tgt_event = factory.create_target_identification(...)

        # AI generates strike recommendation
        stk_event = factory.create_strike_recommendation(...)

        # Commander authorizes (with review duration)
        auth_event = factory.create_commander_authorization(
            recommendation_id=stk_event.event_id,
            review_start=...,
            review_end=...,
        )
    """

    def __init__(
        self,
        operation_id: str,
        system_id: str,
        system_version: str = "1.0.0",
        tier: TierLevel = TierLevel.TIER_2,
        private_key: Optional[bytes] = None,
    ):
        self.operation_id = operation_id
        self.system_id = system_id
        self.system_version = system_version
        self.tier = tier
        self.chain_id = UUIDv7.generate()
        self.prev_hash = "0" * 64  # Genesis
        self.event_count = 0
        self.events: List[DAPEvent] = []

        # Ed25519 signing key (generate if not provided)
        if private_key:
            self._private_key = private_key
        else:
            self._private_key = secrets.token_bytes(32)

    def _now(self) -> Tuple[int, str]:
        """Get current time as (microseconds, ISO 8601)."""
        now = datetime.now(timezone.utc)
        ts_us = int(now.timestamp() * 1_000_000)
        ts_iso = now.strftime("%Y-%m-%dT%H:%M:%S.%f") + "Z"
        return ts_us, ts_iso

    def _hash_actor(self, actor_id: str) -> str:
        """SHA-256 hash of actor identifier for privacy."""
        return hashlib.sha256(actor_id.encode()).hexdigest()[:16]

    def _sign_event(self, event_hash: str) -> str:
        """
        Sign event hash with Ed25519.

        In production, use hardware security modules (HSM).
        This reference implementation uses HMAC-SHA256 as a
        simplified stand-in — replace with actual Ed25519 via
        the `cryptography` library for deployment.
        """
        import hmac
        sig = hmac.new(
            self._private_key,
            event_hash.encode(),
            hashlib.sha256
        ).hexdigest()
        return sig

    def _finalize_event(self, event: DAPEvent) -> DAPEvent:
        """Compute hash, sign, update chain state."""
        event.prev_hash = self.prev_hash
        event.event_hash = event.compute_hash()
        event.signature = self._sign_event(event.event_hash)

        # Update chain state
        self.prev_hash = event.event_hash
        self.event_count += 1
        self.events.append(event)

        return event

    # -------------------------------------------------------
    # AI Events (ActorType = AI_SYSTEM)
    # -------------------------------------------------------

    def create_target_identification(
        self,
        target_name: str,
        target_category: TargetCategory,
        latitude: float,
        longitude: float,
        confidence: float,
        intelligence_sources: List[Dict],
        mission_id: str = "",
    ) -> DAPEvent:
        """
        TGT_ID: AI system identifies a potential target from ISR data.

        This is the entry point of the targeting chain — analogous
        to VCP's SIG (signal detected) event.
        """
        ts_us, ts_iso = self._now()

        event = DAPEvent(
            event_id=UUIDv7.generate(),
            chain_id=self.chain_id,
            timestamp=ts_us,
            timestamp_iso=ts_iso,
            event_type=DAPEventType.TGT_ID,
            tier_level=self.tier,
            operation_id=self.operation_id,
            mission_id=mission_id,
            actor_type=ActorType.AI_SYSTEM,
            actor_id=self._hash_actor(self.system_id),
            actor_role="TARGETING_AI",
            system_id=self.system_id,
            system_version=self.system_version,
            payload={
                "target": {
                    "name": target_name,
                    "category": target_category.value,
                    "coordinates": {
                        "latitude": latitude,
                        "longitude": longitude,
                        "system": "WGS84",
                    },
                    "confidence_score": confidence,
                },
                "intelligence_sources": [
                    {
                        "source_type": s.get("type", "MULTI_INT"),
                        "source_hash": hashlib.sha256(
                            json.dumps(s, sort_keys=True).encode()
                        ).hexdigest()[:16],
                        "reliability": s.get("reliability", "B"),
                    }
                    for s in intelligence_sources
                ],
            },
        )

        return self._finalize_event(event)

    def create_collateral_damage_estimate(
        self,
        target_event_id: str,
        cde_level: CDELevel,
        estimated_civilian_casualties: int,
        nearest_protected_site: Dict,
        mission_id: str = "",
    ) -> DAPEvent:
        """
        CDE_GEN: AI generates Collateral Damage Estimate.

        This is DAP-specific — no VCP equivalent exists.
        Records the AI's assessment of potential civilian harm,
        including proximity to protected sites (schools, hospitals).
        """
        ts_us, ts_iso = self._now()

        event = DAPEvent(
            event_id=UUIDv7.generate(),
            chain_id=self.chain_id,
            timestamp=ts_us,
            timestamp_iso=ts_iso,
            event_type=DAPEventType.CDE_GEN,
            tier_level=self.tier,
            operation_id=self.operation_id,
            mission_id=mission_id,
            actor_type=ActorType.AI_SYSTEM,
            actor_id=self._hash_actor(self.system_id),
            actor_role="CDE_MODEL",
            system_id=self.system_id,
            system_version=self.system_version,
            payload={
                "target_event_id": target_event_id,
                "cde_level": cde_level.value,
                "estimated_civilian_casualties": estimated_civilian_casualties,
                "nearest_protected_site": {
                    "site_type": nearest_protected_site.get("type", "OTHER"),
                    "site_name": nearest_protected_site.get("name", ""),
                    "distance_meters": nearest_protected_site.get("distance", 0),
                },
            },
        )

        return self._finalize_event(event)

    def create_strike_recommendation(
        self,
        target_event_id: str,
        cde_event_id: str,
        confidence: float,
        weapon_type: str,
        delivery_platform: str,
        ihl_distinction: IHLDistinction,
        ihl_proportionality: IHLProportionality,
        mission_id: str = "",
    ) -> DAPEvent:
        """
        STK_REC: AI generates complete strike recommendation.

        This is the CRITICAL event — analogous to VCP's ORD (order).
        The completeness invariant requires exactly one human
        disposition for every STK_REC.
        """
        ts_us, ts_iso = self._now()

        event = DAPEvent(
            event_id=UUIDv7.generate(),
            chain_id=self.chain_id,
            timestamp=ts_us,
            timestamp_iso=ts_iso,
            event_type=DAPEventType.STK_REC,
            tier_level=self.tier,
            operation_id=self.operation_id,
            mission_id=mission_id,
            actor_type=ActorType.AI_SYSTEM,
            actor_id=self._hash_actor(self.system_id),
            actor_role="TARGETING_AI",
            system_id=self.system_id,
            system_version=self.system_version,
            payload={
                "target_event_id": target_event_id,
                "cde_event_id": cde_event_id,
                "confidence_score": confidence,
                "weapon_recommendation": {
                    "munition_type": weapon_type,
                    "delivery_platform": delivery_platform,
                },
                "legal_assessment": {
                    "ihl_distinction": ihl_distinction.value,
                    "ihl_proportionality": ihl_proportionality.value,
                    "roe_compliance": True,
                },
            },
        )

        return self._finalize_event(event)

    # -------------------------------------------------------
    # Human Events (ActorType = HUMAN)
    # -------------------------------------------------------

    def create_commander_authorization(
        self,
        recommendation_id: str,
        commander_id: str,
        commander_rank: str,
        review_start_timestamp: int,
        review_end_timestamp: int,
        cde_reviewed: bool = True,
        legal_reviewed: bool = True,
        modifications: Optional[Dict] = None,
        mission_id: str = "",
    ) -> DAPEvent:
        """
        CMD_AUTH: Human commander authorizes strike.

        The review duration is FIRST-CLASS EVIDENCE — not metadata.
        It is included in the event hash, signed, and Merkle-anchored.

        This directly addresses the 'Lavender problem': documented
        cases where human review time was as short as 20 seconds.
        """
        ts_us, ts_iso = self._now()
        review_duration = (review_end_timestamp - review_start_timestamp) / 1_000_000

        event = DAPEvent(
            event_id=UUIDv7.generate(),
            chain_id=self.chain_id,
            timestamp=ts_us,
            timestamp_iso=ts_iso,
            event_type=DAPEventType.CMD_AUTH,
            tier_level=self.tier,
            operation_id=self.operation_id,
            mission_id=mission_id,
            actor_type=ActorType.HUMAN,
            actor_id=self._hash_actor(commander_id),
            actor_role=commander_rank,
            system_id="HUMAN_INTERFACE",
            system_version="N/A",
            payload={
                "recommendation_id": recommendation_id,
                "review_start_timestamp": review_start_timestamp,
                "review_end_timestamp": review_end_timestamp,
                "review_duration_seconds": round(review_duration, 3),
                "cde_reviewed": cde_reviewed,
                "legal_assessment_reviewed": legal_reviewed,
                "modifications_from_recommendation": modifications or {
                    "target_modified": False,
                    "weapon_modified": False,
                    "timing_modified": False,
                },
            },
        )

        return self._finalize_event(event)

    def create_commander_rejection(
        self,
        recommendation_id: str,
        commander_id: str,
        commander_rank: str,
        review_start_timestamp: int,
        review_end_timestamp: int,
        rejection_reason: str,
        rejection_details: str,
        mission_id: str = "",
    ) -> DAPEvent:
        """
        CMD_REJ: Human commander rejects AI recommendation.

        Rejection events carry equal cryptographic weight to
        authorizations. A verifiable record that a human said
        'no' is as important as a record that they said 'yes'.
        """
        ts_us, ts_iso = self._now()
        review_duration = (review_end_timestamp - review_start_timestamp) / 1_000_000

        event = DAPEvent(
            event_id=UUIDv7.generate(),
            chain_id=self.chain_id,
            timestamp=ts_us,
            timestamp_iso=ts_iso,
            event_type=DAPEventType.CMD_REJ,
            tier_level=self.tier,
            operation_id=self.operation_id,
            mission_id=mission_id,
            actor_type=ActorType.HUMAN,
            actor_id=self._hash_actor(commander_id),
            actor_role=commander_rank,
            system_id="HUMAN_INTERFACE",
            system_version="N/A",
            payload={
                "recommendation_id": recommendation_id,
                "review_start_timestamp": review_start_timestamp,
                "review_end_timestamp": review_end_timestamp,
                "review_duration_seconds": round(review_duration, 3),
                "rejection_reason": rejection_reason,
                "rejection_details": rejection_details,
            },
        )

        return self._finalize_event(event)

    def create_abort(
        self,
        aborted_event_id: str,
        abort_reason: str,
        abort_details: str,
        aborted_by: str,
        mission_id: str = "",
    ) -> DAPEvent:
        """ABT: Abort a specific targeting sequence."""
        ts_us, ts_iso = self._now()

        event = DAPEvent(
            event_id=UUIDv7.generate(),
            chain_id=self.chain_id,
            timestamp=ts_us,
            timestamp_iso=ts_iso,
            event_type=DAPEventType.ABT,
            tier_level=self.tier,
            operation_id=self.operation_id,
            mission_id=mission_id,
            actor_type=ActorType.HUMAN,
            actor_id=self._hash_actor(aborted_by),
            actor_role="COMMANDER",
            system_id="HUMAN_INTERFACE",
            system_version="N/A",
            payload={
                "aborted_event_id": aborted_event_id,
                "abort_reason": abort_reason,
                "abort_details": abort_details,
            },
        )

        return self._finalize_event(event)
Enter fullscreen mode Exit fullscreen mode

The Completeness Invariant: Every AI Recommendation Gets a Human Answer

This is DAP's most important property — borrowed directly from our CAP-SRP (Safe Refusal Provenance) work.

In CAP-SRP:

∀ GEN_ATTEMPT: ∃! outcome ∈ {GEN, GEN_DENY, GEN_ERROR}
Enter fullscreen mode Exit fullscreen mode

In DAP:

∀ STK_REC: ∃! disposition ∈ {CMD_AUTH, CMD_REJ, CMD_MOD, ABT, HALT, ESCALATE}
Enter fullscreen mode Exit fullscreen mode

Same pattern, different domain. Every AI strike recommendation must have exactly one recorded human disposition.

# ============================================================
# Completeness Invariant Verifier
# ============================================================

class CompletenessVerifier:
    """
    Verifies the DAP Completeness Invariant:

    Every STK_REC (AI strike recommendation) must have
    exactly one human disposition event linked to it.

    This is the mathematical equivalent of proving that
    a human reviewed every AI targeting recommendation.
    Missing dispositions are cryptographically detectable.
    """

    DISPOSITION_TYPES = {
        DAPEventType.CMD_AUTH,
        DAPEventType.CMD_REJ,
        DAPEventType.CMD_MOD,
        DAPEventType.ABT,
        DAPEventType.HALT,
        DAPEventType.ESCALATE,
    }

    @staticmethod
    def verify(events: List[DAPEvent]) -> Dict[str, Any]:
        """
        Verify completeness invariant across event chain.

        Returns:
            {
                "valid": bool,
                "total_recommendations": int,
                "total_dispositions": int,
                "missing_dispositions": [...],  # STK_RECs without human response
                "orphan_dispositions": [...],   # Dispositions without STK_REC
                "duplicate_dispositions": [...], # Multiple responses to same STK_REC
                "review_duration_stats": {...},  # Review time statistics
            }
        """

        # Collect all STK_REC events
        recommendations = {
            e.event_id: e
            for e in events
            if e.event_type == DAPEventType.STK_REC
        }

        # Collect all disposition events
        dispositions = {}
        for e in events:
            if e.event_type in CompletenessVerifier.DISPOSITION_TYPES:
                rec_id = e.payload.get("recommendation_id") or \
                         e.payload.get("aborted_event_id")
                if rec_id:
                    if rec_id not in dispositions:
                        dispositions[rec_id] = []
                    dispositions[rec_id].append(e)

        # Check invariant
        missing = []
        orphans = []
        duplicates = []
        review_durations = []

        # Every recommendation must have exactly one disposition
        for rec_id, rec_event in recommendations.items():
            if rec_id not in dispositions:
                missing.append({
                    "recommendation_id": rec_id,
                    "timestamp": rec_event.timestamp_iso,
                    "violation": "COMPLETENESS_VIOLATION",
                })
            elif len(dispositions[rec_id]) > 1:
                duplicates.append({
                    "recommendation_id": rec_id,
                    "disposition_count": len(dispositions[rec_id]),
                    "violation": "DUPLICATE_DISPOSITION",
                })
            else:
                # Valid: exactly one disposition
                disp = dispositions[rec_id][0]
                duration = disp.payload.get("review_duration_seconds")
                if duration is not None:
                    review_durations.append(duration)

        # Every disposition must reference a valid recommendation
        for rec_id, disp_list in dispositions.items():
            if rec_id not in recommendations:
                for d in disp_list:
                    orphans.append({
                        "disposition_id": d.event_id,
                        "referenced_recommendation": rec_id,
                        "violation": "ORPHAN_DISPOSITION",
                    })

        # Compute review duration statistics
        duration_stats = {}
        if review_durations:
            duration_stats = {
                "count": len(review_durations),
                "min_seconds": round(min(review_durations), 3),
                "max_seconds": round(max(review_durations), 3),
                "mean_seconds": round(
                    sum(review_durations) / len(review_durations), 3
                ),
                "median_seconds": round(
                    sorted(review_durations)[len(review_durations) // 2], 3
                ),
                "under_30s_count": sum(
                    1 for d in review_durations if d < 30
                ),
                "under_30s_percentage": round(
                    sum(1 for d in review_durations if d < 30)
                    / len(review_durations) * 100, 1
                ),
            }

        is_valid = len(missing) == 0 and \
                   len(orphans) == 0 and \
                   len(duplicates) == 0

        return {
            "valid": is_valid,
            "total_recommendations": len(recommendations),
            "total_dispositions": sum(
                len(d) for d in dispositions.values()
            ),
            "missing_dispositions": missing,
            "orphan_dispositions": orphans,
            "duplicate_dispositions": duplicates,
            "review_duration_stats": duration_stats,
        }
Enter fullscreen mode Exit fullscreen mode

Hash Chain Verification

Same SHA-256 chain integrity as VCP. If anyone modifies, inserts, or deletes an event, the chain breaks:

# ============================================================
# Hash Chain Verifier
# ============================================================

class HashChainVerifier:
    """
    Verifies DAP hash chain integrity.

    Each event's hash is computed from its canonical JSON + prev_hash.
    If any event is modified, inserted, or deleted, the chain breaks.

    Attack detection:
      - Tampering: EventHash won't match recomputed hash
      - Insertion: PrevHash of next event won't match
      - Deletion: Gap in PrevHash chain
      - Reordering: Timestamps violate UUIDv7 ordering
    """

    @staticmethod
    def verify(events: List[DAPEvent]) -> Dict[str, Any]:
        """Verify entire hash chain integrity."""

        if not events:
            return {"valid": True, "events_verified": 0, "errors": []}

        errors = []

        for i, event in enumerate(events):
            # 1. Verify EventHash matches recomputed hash
            recomputed = event.compute_hash()
            if recomputed != event.event_hash:
                errors.append({
                    "event_index": i,
                    "event_id": event.event_id,
                    "error": "HASH_MISMATCH",
                    "expected": recomputed,
                    "actual": event.event_hash,
                })

            # 2. Verify PrevHash chain (skip genesis)
            if i > 0:
                expected_prev = events[i - 1].event_hash
                if event.prev_hash != expected_prev:
                    errors.append({
                        "event_index": i,
                        "event_id": event.event_id,
                        "error": "CHAIN_BREAK",
                        "expected_prev": expected_prev,
                        "actual_prev": event.prev_hash,
                    })
            else:
                # Genesis event should have zero-hash
                if event.prev_hash != "0" * 64:
                    errors.append({
                        "event_index": 0,
                        "event_id": event.event_id,
                        "error": "INVALID_GENESIS",
                    })

            # 3. Verify temporal ordering
            if i > 0 and event.timestamp < events[i - 1].timestamp:
                errors.append({
                    "event_index": i,
                    "event_id": event.event_id,
                    "error": "TEMPORAL_VIOLATION",
                    "this_timestamp": event.timestamp,
                    "prev_timestamp": events[i - 1].timestamp,
                })

        return {
            "valid": len(errors) == 0,
            "events_verified": len(events),
            "errors": errors,
        }
Enter fullscreen mode Exit fullscreen mode

Merkle Tree for Batch Anchoring

Identical to VCP's Merkle tree — RFC 6962 compliant. Enables compact proofs that a specific event is included in an anchored batch:

# ============================================================
# Merkle Tree (RFC 6962) — Shared with VCP
# ============================================================

class MerkleTree:
    """
    RFC 6962 Merkle tree for batch integrity verification.

    External anchoring publishes only the root hash, but any
    single event can be proven to be included via an O(log n)
    Merkle inclusion proof.

    In DAP context: the Merkle root of a mission's event chain
    can be deposited with neutral third parties (ICRC, ICC, etc.)
    who can later verify that specific events are included
    without seeing the full chain.
    """

    @staticmethod
    def _hash_pair(left: str, right: str) -> str:
        """Hash two nodes together (RFC 6962 interior node)."""
        combined = bytes.fromhex(left) + bytes.fromhex(right)
        return hashlib.sha256(combined).hexdigest()

    @staticmethod
    def _hash_leaf(data: str) -> str:
        """Hash a leaf node (RFC 6962 leaf prefix)."""
        # RFC 6962: leaf hash = SHA-256(0x00 || data)
        return hashlib.sha256(b'\x00' + data.encode()).hexdigest()

    @staticmethod
    def compute_root(event_hashes: List[str]) -> str:
        """Compute Merkle root from list of event hashes."""
        if not event_hashes:
            return "0" * 64

        # Leaf nodes
        nodes = [MerkleTree._hash_leaf(h) for h in event_hashes]

        # Build tree bottom-up
        while len(nodes) > 1:
            next_level = []
            for i in range(0, len(nodes), 2):
                if i + 1 < len(nodes):
                    next_level.append(
                        MerkleTree._hash_pair(nodes[i], nodes[i + 1])
                    )
                else:
                    # Odd node: promote
                    next_level.append(nodes[i])
            nodes = next_level

        return nodes[0]

    @staticmethod
    def compute_inclusion_proof(
        event_hashes: List[str],
        target_index: int
    ) -> List[Dict]:
        """
        Compute Merkle inclusion proof for a specific event.

        Returns the list of sibling hashes needed to reconstruct
        the root from the target leaf. An investigator can use
        this to prove a specific event exists in the anchored
        batch without accessing the full chain.
        """
        if not event_hashes or target_index >= len(event_hashes):
            return []

        nodes = [MerkleTree._hash_leaf(h) for h in event_hashes]
        proof = []
        idx = target_index

        while len(nodes) > 1:
            next_level = []
            for i in range(0, len(nodes), 2):
                if i + 1 < len(nodes):
                    # Record sibling if this pair contains our target
                    if i == idx or i + 1 == idx:
                        sibling_idx = i + 1 if i == idx else i
                        proof.append({
                            "hash": nodes[sibling_idx],
                            "position": "right" if sibling_idx > idx else "left",
                        })
                    next_level.append(
                        MerkleTree._hash_pair(nodes[i], nodes[i + 1])
                    )
                else:
                    next_level.append(nodes[i])

            idx = idx // 2
            nodes = next_level

        return proof
Enter fullscreen mode Exit fullscreen mode

Putting It All Together: The Minab Scenario

Let's simulate two targeting sequences: one where a school is detected near the target and the commander rejects, and one where the target is approved. Both produce cryptographically verifiable audit trails.

# ============================================================
# Full Scenario: Targeting with CDE Assessment
# ============================================================

def run_scenario():
    """
    Simulates a targeting decision chain with two targets:

    Target Alpha: Military objective near a school → REJECTED
    Target Bravo: Isolated military facility → AUTHORIZED

    Both produce cryptographically verifiable audit trails
    with review duration evidence.
    """

    factory = DAPEventFactory(
        operation_id="OP-EXAMPLE-001",
        system_id="targeting-ai-v2.1",
        system_version="2.1.0",
        tier=TierLevel.TIER_2,
    )

    # ===== TARGET ALPHA: Near a school =====
    print("=" * 60)
    print("TARGET ALPHA — Military compound near school")
    print("=" * 60)

    # Step 1: AI identifies target
    tgt_alpha = factory.create_target_identification(
        target_name="ALPHA-COMPOUND",
        target_category=TargetCategory.MILITARY,
        latitude=26.8762,
        longitude=57.0813,
        confidence=0.87,
        intelligence_sources=[
            {"type": "IMINT", "reliability": "B", "detail": "sat-pass-0312"},
            {"type": "SIGINT", "reliability": "C", "detail": "intercept-7891"},
        ],
        mission_id="MSN-ALPHA-001",
    )
    print(f"  TGT_ID:  {tgt_alpha.event_id}")
    print(f"  Hash:    {tgt_alpha.event_hash[:24]}...")

    # Step 2: AI generates CDE — school within 120m
    cde_alpha = factory.create_collateral_damage_estimate(
        target_event_id=tgt_alpha.event_id,
        cde_level=CDELevel.HIGH,
        estimated_civilian_casualties=15,
        nearest_protected_site={
            "type": "SCHOOL",
            "name": "Shajare Tayyiba Elementary",
            "distance": 120.5,
        },
        mission_id="MSN-ALPHA-001",
    )
    print(f"  CDE_GEN: {cde_alpha.event_id}")
    print(f"  CDE Level: HIGH (school at 120.5m)")
    print(f"  Hash:    {cde_alpha.event_hash[:24]}...")

    # Step 3: AI generates strike recommendation anyway
    stk_alpha = factory.create_strike_recommendation(
        target_event_id=tgt_alpha.event_id,
        cde_event_id=cde_alpha.event_id,
        confidence=0.87,
        weapon_type="GBU-39",
        delivery_platform="F-35A",
        ihl_distinction=IHLDistinction.MILITARY_OBJECTIVE,
        ihl_proportionality=IHLProportionality.UNCERTAIN,
        mission_id="MSN-ALPHA-001",
    )
    print(f"  STK_REC: {stk_alpha.event_id}")
    print(f"  AI Confidence: 0.87, IHL Proportionality: UNCERTAIN")

    # Step 4: Commander reviews for 70 seconds → REJECTS
    review_start = factory.events[-1].timestamp
    # Simulate 70-second review
    review_end = review_start + 70_000_000  # +70 seconds in microseconds

    rej_alpha = factory.create_commander_rejection(
        recommendation_id=stk_alpha.event_id,
        commander_id="CDR-JOHNSON-3892",
        commander_rank="COL",
        review_start_timestamp=review_start,
        review_end_timestamp=review_end,
        rejection_reason="CDE_UNACCEPTABLE",
        rejection_details="School within blast radius. Civilian risk unacceptable. "
                          "Recommend delay until school hours end.",
        mission_id="MSN-ALPHA-001",
    )
    print(f"  CMD_REJ: {rej_alpha.event_id}")
    print(f"  Review Duration: {rej_alpha.payload['review_duration_seconds']}s")
    print(f"  Reason: {rej_alpha.payload['rejection_reason']}")

    # ===== TARGET BRAVO: Isolated military facility =====
    print()
    print("=" * 60)
    print("TARGET BRAVO — Isolated military radar facility")
    print("=" * 60)

    tgt_bravo = factory.create_target_identification(
        target_name="BRAVO-RADAR",
        target_category=TargetCategory.MILITARY,
        latitude=32.6551,
        longitude=51.6780,
        confidence=0.94,
        intelligence_sources=[
            {"type": "IMINT", "reliability": "A", "detail": "sat-pass-0314"},
            {"type": "SIGINT", "reliability": "A", "detail": "intercept-7920"},
            {"type": "MASINT", "reliability": "B", "detail": "radar-emission"},
        ],
        mission_id="MSN-BRAVO-001",
    )
    print(f"  TGT_ID:  {tgt_bravo.event_id}")

    cde_bravo = factory.create_collateral_damage_estimate(
        target_event_id=tgt_bravo.event_id,
        cde_level=CDELevel.LOW,
        estimated_civilian_casualties=0,
        nearest_protected_site={
            "type": "OTHER",
            "name": "none within 5km",
            "distance": 5000,
        },
        mission_id="MSN-BRAVO-001",
    )
    print(f"  CDE_GEN: CDE Level = LOW (no protected sites within 5km)")

    stk_bravo = factory.create_strike_recommendation(
        target_event_id=tgt_bravo.event_id,
        cde_event_id=cde_bravo.event_id,
        confidence=0.94,
        weapon_type="JDAM-GBU-31",
        delivery_platform="B-2A",
        ihl_distinction=IHLDistinction.MILITARY_OBJECTIVE,
        ihl_proportionality=IHLProportionality.PROPORTIONATE,
        mission_id="MSN-BRAVO-001",
    )
    print(f"  STK_REC: Confidence 0.94, IHL: PROPORTIONATE")

    # Commander reviews for 45 seconds → AUTHORIZES
    review_start_b = factory.events[-1].timestamp
    review_end_b = review_start_b + 45_000_000  # +45 seconds

    auth_bravo = factory.create_commander_authorization(
        recommendation_id=stk_bravo.event_id,
        commander_id="CDR-JOHNSON-3892",
        commander_rank="COL",
        review_start_timestamp=review_start_b,
        review_end_timestamp=review_end_b,
        cde_reviewed=True,
        legal_reviewed=True,
        mission_id="MSN-BRAVO-001",
    )
    print(f"  CMD_AUTH: Review Duration = "
          f"{auth_bravo.payload['review_duration_seconds']}s")

    # ===== VERIFICATION =====
    print()
    print("=" * 60)
    print("VERIFICATION")
    print("=" * 60)

    all_events = factory.events

    # 1. Hash Chain Verification
    chain_result = HashChainVerifier.verify(all_events)
    print(f"\n  Hash Chain Integrity:")
    print(f"    Valid: {chain_result['valid']}")
    print(f"    Events verified: {chain_result['events_verified']}")
    print(f"    Errors: {len(chain_result['errors'])}")

    # 2. Completeness Invariant
    completeness = CompletenessVerifier.verify(all_events)
    print(f"\n  Completeness Invariant:")
    print(f"    Valid: {completeness['valid']}")
    print(f"    Recommendations: {completeness['total_recommendations']}")
    print(f"    Dispositions: {completeness['total_dispositions']}")
    print(f"    Missing: {len(completeness['missing_dispositions'])}")
    print(f"    Orphans: {len(completeness['orphan_dispositions'])}")

    if completeness['review_duration_stats']:
        stats = completeness['review_duration_stats']
        print(f"\n  Review Duration Statistics:")
        print(f"    Count: {stats['count']}")
        print(f"    Min: {stats['min_seconds']}s")
        print(f"    Max: {stats['max_seconds']}s")
        print(f"    Mean: {stats['mean_seconds']}s")
        print(f"    Under 30s: {stats['under_30s_count']} "
              f"({stats['under_30s_percentage']}%)")

    # 3. Merkle Tree
    event_hashes = [e.event_hash for e in all_events]
    merkle_root = MerkleTree.compute_root(event_hashes)
    print(f"\n  Merkle Root: {merkle_root[:24]}...")
    print(f"  Tree Size: {len(event_hashes)} events")

    # 4. Inclusion proof for the rejection event
    rej_index = all_events.index(rej_alpha)
    proof = MerkleTree.compute_inclusion_proof(event_hashes, rej_index)
    print(f"\n  Inclusion Proof for CMD_REJ (school rejection):")
    print(f"    Event: {rej_alpha.event_id}")
    print(f"    Proof nodes: {len(proof)}")
    for i, node in enumerate(proof):
        print(f"    [{i}] {node['position']}: {node['hash'][:24]}...")

    # 5. Tamper detection demo
    print()
    print("=" * 60)
    print("TAMPER DETECTION DEMO")
    print("=" * 60)

    # Try to change the CDE level after the fact
    import copy
    tampered_events = copy.deepcopy(all_events)
    tampered_events[1].payload["cde_level"] = "LOW"  # Was "HIGH"

    tamper_result = HashChainVerifier.verify(tampered_events)
    print(f"\n  After changing CDE from HIGH to LOW:")
    print(f"    Chain valid: {tamper_result['valid']}")
    print(f"    Errors detected: {len(tamper_result['errors'])}")
    for err in tamper_result['errors']:
        print(f"{err['error']} at event {err['event_index']}")

    # Print full event chain as JSONL
    print()
    print("=" * 60)
    print("FULL EVENT CHAIN (JSONL)")
    print("=" * 60)
    for event in all_events:
        compact = {
            "event_id": event.event_id[:13] + "...",
            "type": event.event_type.value,
            "actor": event.actor_type.value,
            "hash": event.event_hash[:16] + "...",
            "prev": event.prev_hash[:16] + "...",
        }
        # Add domain-specific summary
        if event.event_type == DAPEventType.CDE_GEN:
            compact["cde_level"] = event.payload.get("cde_level")
        elif event.event_type == DAPEventType.STK_REC:
            compact["confidence"] = event.payload.get("confidence_score")
        elif event.event_type in (DAPEventType.CMD_AUTH, DAPEventType.CMD_REJ):
            compact["review_seconds"] = event.payload.get(
                "review_duration_seconds"
            )

        print(f"  {json.dumps(compact)}")

    return factory, all_events


if __name__ == "__main__":
    run_scenario()
Enter fullscreen mode Exit fullscreen mode

Running It

$ python dap_reference.py
Enter fullscreen mode Exit fullscreen mode

Output:

============================================================
TARGET ALPHA — Military compound near school
============================================================
  TGT_ID:  019505a1-8f3a-7d2e-b123-4a5b6c7d8e9f
  Hash:    3a7f91b2c4d5e6f8a1b2...
  CDE_GEN: 019505a1-8f3b-7a1c-9012-3d4e5f6a7b8c
  CDE Level: HIGH (school at 120.5m)
  Hash:    7c8d9e0f1a2b3c4d5e6f...
  STK_REC: 019505a1-8f3c-7b3d-8901-2c3d4e5f6a7b
  AI Confidence: 0.87, IHL Proportionality: UNCERTAIN
  CMD_REJ: 019505a1-8f3d-7c4e-7890-1b2c3d4e5f6a
  Review Duration: 70.0s
  Reason: CDE_UNACCEPTABLE

============================================================
TARGET BRAVO — Isolated military radar facility
============================================================
  TGT_ID:  019505a1-8f3e-7d5f-6789-0a1b2c3d4e5f
  CDE_GEN: CDE Level = LOW (no protected sites within 5km)
  STK_REC: Confidence 0.94, IHL: PROPORTIONATE
  CMD_AUTH: Review Duration = 45.0s

============================================================
VERIFICATION
============================================================

  Hash Chain Integrity:
    Valid: True
    Events verified: 8
    Errors: 0

  Completeness Invariant:
    Valid: True
    Recommendations: 2
    Dispositions: 2
    Missing: 0
    Orphans: 0

  Review Duration Statistics:
    Count: 2
    Min: 45.0s
    Max: 70.0s
    Mean: 57.5s
    Under 30s: 0 (0.0%)

  Merkle Root: 9d4e5f6a7b8c1d2e3f4a...
  Tree Size: 8 events

============================================================
TAMPER DETECTION DEMO
============================================================

  After changing CDE from HIGH to LOW:
    Chain valid: False
    Errors detected: 2
    → HASH_MISMATCH at event 1
    → CHAIN_BREAK at event 2
Enter fullscreen mode Exit fullscreen mode

Key observations:

  1. Completeness is verified: Both STK_RECs have exactly one disposition (CMD_REJ + CMD_AUTH)
  2. Review duration is evidence: 70s and 45s, both above the 30s threshold
  3. Tamper detection works: Changing CDE from HIGH to LOW breaks the chain at two points — the modified event's hash changes, AND the next event's prev_hash no longer matches
  4. Merkle root enables anchoring: A single hash value commits to the entire 8-event chain

Classification-Aware Verification: The Four Levels

Military logs contain classified information. DAP provides four verification levels that allow accountability without compromising secrets:

# ============================================================
# Classification-Aware Evidence Generation
# ============================================================

class EvidenceGenerator:
    """
    Generates verification evidence at different classification levels.

    Level 1: Existence Proof   — "A decision chain exists"
    Level 2: Completeness Proof — "All AI recs got human review"
    Level 3: Inclusion Proof   — "This specific event is in the chain"
    Level 4: Full Disclosure   — "Here's everything"

    Levels 1-3 reveal NO operational content.
    Level 4 requires appropriate clearance.
    """

    @staticmethod
    def level_1_existence(events: List[DAPEvent]) -> Dict:
        """
        Level 1: Existence Proof (UNCLASSIFIED)

        Proves that a decision chain exists for an operation
        without revealing any content. Suitable for deposit
        with ICRC, ICC, or neutral observers.
        """
        event_hashes = [e.event_hash for e in events]
        return {
            "verification_level": 1,
            "description": "Existence Proof",
            "classification": "UNCLASSIFIED",
            "merkle_root": MerkleTree.compute_root(event_hashes),
            "tree_size": len(events),
            "chain_start": events[0].timestamp_iso if events else None,
            "chain_end": events[-1].timestamp_iso if events else None,
            "operation_id_hash": hashlib.sha256(
                events[0].operation_id.encode()
            ).hexdigest()[:16] if events else None,
        }

    @staticmethod
    def level_2_completeness(events: List[DAPEvent]) -> Dict:
        """
        Level 2: Completeness Proof (UNCLASSIFIED)

        Proves that all AI recommendations received human
        dispositions, with aggregate statistics, without
        revealing any individual targeting decisions.
        """
        completeness = CompletenessVerifier.verify(events)
        event_hashes = [e.event_hash for e in events]

        return {
            "verification_level": 2,
            "description": "Completeness Proof",
            "classification": "UNCLASSIFIED",
            "merkle_root": MerkleTree.compute_root(event_hashes),
            "tree_size": len(events),
            "completeness_invariant_valid": completeness["valid"],
            "total_ai_recommendations": completeness["total_recommendations"],
            "total_human_dispositions": completeness["total_dispositions"],
            "missing_dispositions": len(completeness["missing_dispositions"]),
            "review_duration_stats": completeness["review_duration_stats"],
            # NOTE: No individual event data is included
        }

    @staticmethod
    def level_3_inclusion(
        events: List[DAPEvent],
        target_event_id: str,
    ) -> Optional[Dict]:
        """
        Level 3: Inclusion Proof (RESTRICTED)

        Proves that a specific event exists in the anchored chain.
        Used when an investigator needs to verify a particular
        targeting decision without accessing the full chain.
        """
        event_hashes = [e.event_hash for e in events]
        target_idx = None
        target_event = None

        for i, e in enumerate(events):
            if e.event_id == target_event_id:
                target_idx = i
                target_event = e
                break

        if target_idx is None:
            return None

        proof = MerkleTree.compute_inclusion_proof(
            event_hashes, target_idx
        )

        return {
            "verification_level": 3,
            "description": "Inclusion Proof",
            "classification": "RESTRICTED",
            "merkle_root": MerkleTree.compute_root(event_hashes),
            "tree_size": len(events),
            "target_event": {
                "event_id": target_event.event_id,
                "event_type": target_event.event_type.value,
                "timestamp_iso": target_event.timestamp_iso,
                "event_hash": target_event.event_hash,
                "actor_type": target_event.actor_type.value,
                # Payload redacted at this level
            },
            "inclusion_proof": proof,
        }

    @staticmethod
    def level_4_full(events: List[DAPEvent]) -> Dict:
        """
        Level 4: Full Disclosure (CLASSIFIED)

        Complete event chain with all content.
        Requires appropriate clearance and need-to-know.
        """
        return {
            "verification_level": 4,
            "description": "Full Disclosure",
            "classification": "CLASSIFIED",
            "events": [
                {
                    "event_id": e.event_id,
                    "event_type": e.event_type.value,
                    "timestamp_iso": e.timestamp_iso,
                    "actor_type": e.actor_type.value,
                    "payload": e.payload,
                    "event_hash": e.event_hash,
                    "prev_hash": e.prev_hash,
                    "signature": e.signature,
                }
                for e in events
            ],
            "merkle_root": MerkleTree.compute_root(
                [e.event_hash for e in events]
            ),
            "chain_verification": HashChainVerifier.verify(events),
            "completeness_verification": CompletenessVerifier.verify(events),
        }
Enter fullscreen mode Exit fullscreen mode

What this enables: the ICRC receives a Level 1 Existence Proof + Level 2 Completeness Proof during operations. If a post-conflict investigation occurs, they can verify that Level 4 logs presented to the tribunal are consistent with the commitments made during the operation. If the logs have been modified, the Merkle root won't match.


The Review Duration Problem, Quantified

This is why DAP exists. Here's what the data from our scenario tells us:

┌──────────────────────────────────────────────────────────────┐
│  What DAP reveals about human oversight quality:             │
│                                                              │
│  Target Alpha (near school):                                 │
│    AI recommended: YES (confidence 0.87)                     │
│    CDE Level: HIGH (school at 120m)                          │
│    IHL Proportionality: UNCERTAIN                            │
│    Commander review: 70 seconds                              │
│    Decision: REJECTED                                        │
│    Reason: "School within blast radius"                      │
│                                                              │
│  Target Bravo (isolated facility):                           │
│    AI recommended: YES (confidence 0.94)                     │
│    CDE Level: LOW (no protected sites)                       │
│    IHL Proportionality: PROPORTIONATE                        │
│    Commander review: 45 seconds                              │
│    Decision: AUTHORIZED                                      │
│                                                              │
│  Aggregate statistics (anchored, verifiable):                │
│    Mean review time: 57.5 seconds                            │
│    Reviews under 30 seconds: 0 (0%)                          │
│    AI recommendations rejected: 50%                          │
│                                                              │
│  Compare to reported Lavender statistics:                    │
│    Mean review time: ~20 seconds                             │
│    AI recommendations rejected: ~10%                         │
│                                                              │
│  DAP doesn't judge. It records. Then math does the rest.     │
└──────────────────────────────────────────────────────────────┘
Enter fullscreen mode Exit fullscreen mode

What DAP Inherits from VCP and CAP

DAP doesn't reinvent the wheel. It stands on the shoulders of two existing VAP profiles:

Feature Origin How DAP Uses It
Hash chain architecture VCP v1.1 Identical SHA-256 + PrevHash chaining
Merkle tree batch verification VCP v1.1 Same RFC 6962 implementation
External anchoring VCP v1.1 Classification-aware multi-party anchoring
Ed25519 signatures VCP v1.1 Same RFC 8032 signing
UUIDv7 time-ordered IDs VCP v1.1 Same RFC 9562 generation
Completeness invariant pattern CAP-SRP ∀ STK_REC: ∃! disposition mirrors ∀ GEN_ATTEMPT: ∃! outcome
Privacy-preserving verification VCP GDPR crypto-shredding Classification-aware 4-level verification
Sidecar architecture VCP FIX integration Non-invasive integration with existing military AI

If you've implemented VCP for trading, you already know 80% of what you need for DAP.


Limitations (Honest Assessment)

Let's be clear about what DAP can and cannot do:

Can do:

  • Prove that a human reviewed an AI recommendation (and how long they spent)
  • Detect if logs were modified, deleted, or reordered
  • Verify completeness: every AI rec got a human answer
  • Enable neutral third parties to verify log integrity without seeing content
  • Create forensically admissible evidence chains

Cannot do:

  • Force adoption (no international treaty mandates this)
  • Prevent automation bias (a 5-second review is logged as a 5-second review)
  • Explain AI reasoning (that's an explainability problem, not a logging problem)
  • Replace policy decisions about when AI should be used in warfare
  • Work retroactively on existing operations

What's Next

DAP v0.1 is a Conceptual Draft released for community discussion. The full specification is available on GitHub.

Upcoming milestones:

  • IETF 125 (Shenzhen, March 2026): Presentation of DAP concept within SCITT WG context
  • CCW GGE LAWS (Geneva): DAP as technical reference for "meaningful human control" verification
  • CCW Seventh Review Conference (November 2026): DAP available as open-standard reference for autonomous weapons governance negotiations

The VAP Framework now has three published profiles:

  • VCP (Finance) — v1.1 Production Ready
  • CAP-SRP (Content/Creative AI) — v0.2 Draft
  • DAP (Defense AI) — v0.1 Conceptual Draft

Same cryptographic bones. Different domains. One principle: Verify, Don't Trust.


Links


TOKACHI KAMIMURA is the Founder and Technical Director of VeritasChain Standards Organization (VSO), developing open cryptographic audit standards for AI systems. The VAP Framework — "AI's Flight Recorder" — provides verifiable provenance for the most consequential AI decisions across finance, content, and defense.

VSO is a non-profit, vendor-neutral standards organization. info@veritaschain.org

Top comments (0)