Federated Safety Signal Sharing¶

FederatedSafetySignalAggregator turns tenant-safe DirectorSafetySignal envelopes into anonymous, differentially private aggregate guard telemetry for cross-organisation sharing.

It enforces four boundaries before release:

input must validate as a DirectorSafetySignal;
each signal must carry a tenant id for contribution bounding;
each tenant can contribute at most one count per category per release window;
releases require at least min_tenants distinct tenants and omit raw counts by default.

from director_ai.core import (
    FederatedSafetySignalAggregator,
    PrivacyAccountant,
    director_safety_signal_from_event,
)
from director_ai.core.safety_event import SafetyEvent

accountant = PrivacyAccountant(max_epsilon=5.0)
aggregator = FederatedSafetySignalAggregator(
    epsilon=0.9,
    accountant=accountant,
    min_tenants=2,
)

event = SafetyEvent.from_policy_decision(
    hook_id="stream",
    hook_scope="streaming",
    policy_decision="halt",
    halt_reason="coherence",
    tenant_safe_explanation="Tenant-safe halt summary.",
    tenant_id="tenant-a",
)
signal = director_safety_signal_from_event(
    event,
    producer_id="runtime-a",
    framework="generic",
)

aggregator.submit_signal(signal)

release() returns noised counts and privacy metadata:

release = aggregator.release()
payload = release.to_dict()

payload contains noisy_counts, epsilon_spent, category names, cohort counts, and a privacy block with payload_classification set to anonymous_dp_aggregate. It does not include tenant ids, raw prompts, raw completions, credentials, media, sensor payloads, or raw aggregate counts.

Raw aggregate counts can be included only for local audit/debug use:

local_audit_payload = release.to_dict(include_raw=True)

Do not publish include_raw=True payloads across tenant or organisation boundaries.

director_ai.core.federated_privacy.signal_sharing.FederatedSafetySignalAggregator ¶

FederatedSafetySignalAggregator(*, epsilon: float, categories: Sequence[str] = DEFAULT_SIGNAL_CATEGORIES, accountant: PrivacyAccountant | None = None, min_tenants: int = 2, seed: int | None = None, allow_insecure_seed: bool = False)

Aggregate tenant-safe guard signals for federated sharing.

The aggregator accepts only validated :class:DirectorSafetySignal envelopes. Each tenant can contribute at most one count to each category within a release window, bounding per-category tenant sensitivity at one. Releases are Laplace-noised through :class:FederatedHistogram and blocked until a minimum distinct-tenant cohort is present.

submit_transport ¶

submit_transport(payload: Mapping[str, Any]) -> tuple[str, ...]

Validate and submit a transport payload.

submit_signal ¶

submit_signal(signal: DirectorSafetySignal) -> tuple[str, ...]

Submit one tenant-safe signal and return accepted categories.

release ¶

release() -> FederatedSafetySignalRelease

Release a DP-noised aggregate and reset the contribution window.

reset ¶

reset() -> None

Drop pending contributions without releasing or charging budget.

director_ai.core.federated_privacy.signal_sharing.FederatedSafetySignalRelease `dataclass` ¶

FederatedSafetySignalRelease(noisy_counts: Mapping[str, float], epsilon_spent: float, categories: tuple[str, ...], signal_count: int, distinct_tenants: int, privacy_unit: str = 'tenant', mechanism: str = 'laplace', raw_counts: Mapping[str, int] = dict())

One anonymous differentially private safety-signal aggregate.

to_dict ¶

to_dict(*, include_raw: bool = False) -> dict[str, Any]

Return a transport-safe release payload.

Raw aggregate counts are omitted by default. They are useful for local audits and tests, but cross-organisation sharing should use the noised values plus the attached privacy metadata.