Sustainability Scoring

Sustainability scoring converts token demand, estimated energy, estimated carbon, and tenant quota state into the shared GuardDecision and SafetyEvent contracts.

Policy Semantics

SustainabilityPolicyAdapter is a policy adapter, not a factual verifier. It emits a VerifierSignal with modality sustainability so downstream orchestrators can keep cost, quota, carbon, and forecast pressure separate from truthfulness evidence.

The adapter enforces these defaults:

• estimates disclose whether hardware values are measured, configured, or projected
• quota exhaustion maps to halt for ordinary reversible actions
• forecast headroom exhaustion maps to warn
• high-carbon requests map to warn with recommended_action="defer"
• high-risk safety actions are not halted solely for sustainability reasons
• events and telemetry omit raw prompts, completions, media, credentials, and other tenant payloads

from director_ai.core.guard_control import RiskEnvelope
from director_ai.core.sustainability import (
    HardwareProfile,
    HardwareProfileRegistry,
    SustainabilityPolicyAdapter,
    TokenEnergyCostEstimator,
)

profile = HardwareProfile(
    profile_id="edge-gpu-a",
    energy_kwh_per_1k_tokens=0.0012,
    carbon_kg_per_kwh=0.18,
    provenance="measured",
)
registry = HardwareProfileRegistry((profile,))
estimator = TokenEnergyCostEstimator(
    hardware_profile=registry.get("edge-gpu-a"),
    cost_per_1k_tokens=0.004,
)
adapter = SustainabilityPolicyAdapter(
    estimator=estimator,
    policy_id="policy.sustainability.production",
    carbon_defer_kg=0.25,
    forecast_headroom_ratio=0.2,
)

decision = adapter.evaluate(
    tenant_id="tenant-a",
    input_tokens=800,
    output_tokens=300,
    quota_remaining_tokens=1200,
    forecast_next_tokens=100,
    risk_envelope=RiskEnvelope(
        action_category="text",
        reversibility="reversible",
        domain="general",
        calibrated_threshold=0.5,
        no_go_threshold=0.85,
    ),
)

For physical, medical, legal, security, irreversible, tool, code, or training actions, sustainability pressure becomes a warning unless another guard-control policy independently blocks the action.

Tenant Telemetry

SustainabilityTelemetry aggregates estimates per tenant and returns threshold alerts without serialising raw request content.

from director_ai.core.sustainability import SustainabilityTelemetry

telemetry = SustainabilityTelemetry(
    token_alert_threshold=100_000,
    cost_alert_threshold=10.0,
    carbon_alert_threshold=1.0,
)
estimate = estimator.estimate(input_tokens=800, output_tokens=300)
telemetry.record("tenant-a", estimate)

The telemetry record() method accepts SustainabilityEstimate values. Store only aggregate summaries in monitoring backends unless the deployment has a separate tenant-approved data-retention policy.

Full API

director_ai.core.sustainability.policy_adapter.HardwareProfile dataclass

HardwareProfile(profile_id: str, energy_kwh_per_1k_tokens: float, carbon_kg_per_kwh: float, provenance: EstimateProvenance)

Deployment hardware profile with explicit measurement provenance.

__post_init__

__post_init__() -> None

Validate profile identity, non-negative factors, and provenance.

to_dict

to_dict() -> dict[str, str | float]

Return tenant-safe profile metadata.

director_ai.core.sustainability.policy_adapter.HardwareProfileRegistry

HardwareProfileRegistry(profiles: tuple[HardwareProfile, ...] = ())

Thread-safe registry for deployment hardware profiles.

Create a registry and register the supplied hardware profiles.

register

register(profile: HardwareProfile, *, replace: bool = False) -> None

Register a hardware profile by id.

get

get(profile_id: str) -> HardwareProfile

Return a profile by id.

snapshot

snapshot() -> dict[str, dict[str, str | float]]

Return tenant-safe profile metadata for all registered profiles.

director_ai.core.sustainability.policy_adapter.SustainabilityEstimate dataclass

SustainabilityEstimate(input_tokens: int, output_tokens: int, total_tokens: int, energy_kwh: float, carbon_kg: float, cost: float, provenance: EstimateProvenance, hardware_profile_id: str)

Tenant-safe token/cost/energy/carbon estimate for one request.

__post_init__

__post_init__() -> None

Validate token accounting, estimate magnitudes, and provenance.

to_dict

to_dict() -> dict[str, str | int | float]

Return a serialisable estimate without prompt or completion payloads.

director_ai.core.sustainability.policy_adapter.TokenEnergyCostEstimator

TokenEnergyCostEstimator(*, hardware_profile: HardwareProfile, cost_per_1k_tokens: float)

Deterministic token-cost-energy estimator for a hardware profile.

Bind estimator policy to one hardware profile and token price.

hardware_profile property

hardware_profile: HardwareProfile

Return the hardware profile used for estimates.

cost_per_1k_tokens property

cost_per_1k_tokens: float

Return the configured cost per thousand tokens.

estimate

estimate(*, input_tokens: int, output_tokens: int) -> SustainabilityEstimate

Estimate request resource usage from token counts.

director_ai.core.sustainability.policy_adapter.SustainabilityPolicyAdapter

SustainabilityPolicyAdapter(*, estimator: TokenEnergyCostEstimator, policy_id: str, carbon_defer_kg: float, forecast_headroom_ratio: float = 0.1)

Convert sustainability estimates into guard-control decisions.

Create a sustainability guard adapter with quota and carbon policy.

evaluate

evaluate(*, tenant_id: str, input_tokens: int, output_tokens: int, quota_remaining_tokens: int, forecast_next_tokens: int, risk_envelope: RiskEnvelope, evidence_ref: str = 'sustainability://estimate') -> GuardDecision

Return a guard decision for the sustainability policy state.

director_ai.core.sustainability.policy_adapter.SustainabilityTelemetry

SustainabilityTelemetry(*, token_alert_threshold: int, cost_alert_threshold: float, carbon_alert_threshold: float)

Thread-safe in-memory per-tenant sustainability telemetry.

Create an in-memory telemetry collector with alert thresholds.

record

record(tenant_id: str, estimate: SustainabilityEstimate) -> None

Append one estimate for a tenant.

summary

summary(tenant_id: str) -> SustainabilityTelemetrySummary

Return aggregate telemetry and threshold alerts for one tenant.

director_ai.core.sustainability.policy_adapter.SustainabilityTelemetrySummary dataclass

SustainabilityTelemetrySummary(tenant_id: str, request_count: int, total_tokens: int, energy_kwh: float, carbon_kg: float, cost: float, alerts: tuple[str, ...])

Per-tenant aggregate telemetry with threshold alert names.

__post_init__

__post_init__() -> None

Validate aggregate telemetry counters and tenant identity.

to_dict

to_dict() -> dict[str, str | int | float | list[str]]

Return a tenant-safe aggregate summary.