Skip to content

Safety Event Schema

SafetyEvent is the canonical tenant-safe telemetry record for runtime halt and policy decisions. The same schema is used by streaming guards, inference server hooks, containment, attestation, ontology, trajectory, cyber-physical, swarm, and agent surfaces.

The independent interoperability specification is published at Director Safety Telemetry v1. Its machine-readable JSON Schema lives at schemas/safety-event.schema.json and is tested against the runtime schema constant.

The schema is intentionally payload-safe:

  • evidence is represented by references only;
  • raw prompts, completions, media, credentials, private sensor payloads, and token-bearing secrets are not allowed;
  • scores and thresholds are bounded to [0, 1];
  • hook scopes and policy decisions are closed enums;
  • hook-specific telemetry goes under string-only attributes.
from director_ai.core import (
    SAFETY_EVENT_JSON_SCHEMA,
    SafetyEvent,
    validate_safety_event_payload,
)

event = SafetyEvent.from_policy_decision(
    hook_id="inference_server.vllm.prehalt",
    hook_scope="inference_server",
    policy_decision="halt",
    halt_reason="coherence_below_threshold",
    tenant_safe_explanation="Pre-sampling halt fired.",
    threshold=0.5,
    observed_score=0.31,
    evidence_refs=("trajectory:7",),
    attributes={"server": "vllm", "token_id": "2"},
)

payload = event.to_dict()
validated = validate_safety_event_payload(payload)
assert validated == event

SAFETY_EVENT_JSON_SCHEMA is JSON-serialisable and can be handed to downstream schema registries, inference-server adapters, dashboards, or audit pipelines. The stdlib validator validate_safety_event_payload() enforces the same critical constraints without requiring a runtime jsonschema dependency.

For cross-runtime sharing, wrap validated events in DirectorSafetySignal. The signal validator uses the same event validator internally, so direct event ingestion and protocol transport stay aligned.

director_ai.core.safety_event.SafetyEvent dataclass 

SafetyEvent(event_id: str, timestamp: str, hook_id: str, hook_scope: str, policy_decision: str, halt_reason: str, tenant_safe_explanation: str, schema_version: str = SAFETY_EVENT_SCHEMA_VERSION, request_id: str = '', tenant_id: str = '', threshold: float | None = None, observed_score: float | None = None, latency_ms: float | None = None, evidence_refs: tuple[str, ...] = tuple(), trace_attribution: HaltTraceAttribution | None = None, attributes: dict[str, str] = dict())

Single halt or policy decision emitted by a runtime hook.

from_halt_evidence classmethod 

from_halt_evidence(evidence: HaltEvidence, *, hook_id: str, hook_scope: str = 'streaming', policy_decision: str = 'halt', event_id: str | None = None, timestamp: str | None = None, request_id: str = '', tenant_id: str = '', latency_ms: float | None = None, attributes: dict[str, str] | None = None) -> SafetyEvent

Build a tenant-safe event from structured halt evidence.

from_policy_decision classmethod 

from_policy_decision(*, hook_id: str, hook_scope: str, policy_decision: str, halt_reason: str, tenant_safe_explanation: str, event_id: str | None = None, timestamp: str | None = None, request_id: str = '', tenant_id: str = '', threshold: float | None = None, observed_score: float | None = None, latency_ms: float | None = None, evidence_refs: Sequence[str] = (), attributes: dict[str, str] | None = None) -> SafetyEvent

Build a tenant-safe event from a hook policy decision.

to_dict 

to_dict() -> dict[str, Any]

Serialise to a JSON-safe dict without raw prompt or fact text.

director_ai.core.safety_event.validate_safety_event_payload 

validate_safety_event_payload(payload: Mapping[str, Any]) -> SafetyEvent

Validate and reconstruct a tenant-safe SafetyEvent payload.

This is a lightweight stdlib validator for the exported JSON schema shape. It intentionally checks the deployment-critical constraints locally instead of requiring the optional jsonschema package at runtime.