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.
Two helpers keep records opaque and time-consistent:
new_safety_event_id() returns the opaque event ID and
utc_timestamp() the RFC-3339 UTC timestamp stamped into every record.
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.
__post_init__
¶
Validate event identity, scores, scope, decision, and safe metadata.
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
¶
Serialise to a JSON-safe dict without raw prompt or fact text.
director_ai.core.safety_event.validate_safety_event_payload
¶
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.