Skip to content

Run a fail-closed remote evaluation worker

The rc37 private runtime adds an evaluation-only contract for a simulated-user evaluator and a deployed Assistant. It is published in the private registry; evaluation mode remains disabled by default, and normal worker behaviour is unchanged. The publication record does not certify hosted credentialed LiveKit metadata propagation, which has not run.

vaani-evals 0.1.1 does not consume this contract. Its companion receiver and cross-package test are owned by vaani-evals issue #19 or later. Vaani freezes the public schema and vectors now so that receiver can be implemented without guessing; see contracts/public/v1/schemas/evaluation-worker-evidence.schema.json and contracts/public/v1/vectors/evaluation-worker-contract.v1.json.

Evaluation safety is an invocation boundary

Starting a Birla worker without Salesforce can be useful during development:

./scripts/dev --without-salesforce

That switch is not an evaluation safety control. It does not prove that SMS, WhatsApp, transfers, payments, customer mutations, provider lookups, or recording cannot reach a real service.

An evaluation worker needs all four deployment-owned controls:

  1. A complete safety manifest, covering every EvaluationCapability.
  2. A distinct EvaluationSafeOperation object for every fake or sandbox route. The object, not its public binding_id, is invoked in evaluation mode. Forbidden routes have no operation.
  3. A shared replay store with an atomic consume operation. An in-memory set is not sufficient for multiple worker processes or replicas.
  4. A distinct EvaluationSafePreparation for every registered Assistant.prepare hook, and an EvaluationSafePromptUpdateObserver for every prompt-update observer. Evaluation never invokes the production hooks; missing safe bindings reject startup.

Every consumer-owned Assistant tool must also be registered with an EvaluationToolBinding. Evaluation startup rejects a tool without one. In an evaluation call VAANI invokes the matching safe operation, never the tool's production callback. Standard LiveKit transfer follows the same rule.

The current contract is text-tier only: recording must be declared forbidden, and any configured recording adapter rejects evaluation startup before LiveKit worker registration. A future recording-capable profile needs a separate public contract and an explicit fake or sandbox recording operation.

import os

from vaani.evaluation import (
    EvaluationAttemptAuthority,
    EvaluationAttemptReplayStore,
    EvaluationCapability,
    EvaluationCapabilityBinding,
    EvaluationCapabilityRoute,
    EvaluationPreparationBinding,
    EvaluationPromptUpdateObserverBinding,
    EvaluationSafeOperation,
    EvaluationSafePreparation,
    EvaluationSafePromptUpdateObserver,
    EvaluationSafetyManifest,
    EvaluationToolBinding,
    EvaluationWorkerContract,
)
from vaani.assistant.worker import run_livekit_worker


class RedisAttemptStore(EvaluationAttemptReplayStore):
    def consume(
        self,
        *,
        attempt_id: str,
        room_id: str,
        dispatch_id: str,
        expires_at_unix_ns: int,
    ) -> bool:
        # Use one Redis/Lua or database conditional-write transaction keyed by
        # attempt_id, room_id, and dispatch_id, with expiry enforcement.
        raise NotImplementedError


def record_fake_notification(*args: object, **kwargs: object) -> object:
    # A separately reviewed fake implementation, not a production notification
    # client or a wrapper around one.
    return {"route": "fake"}


def lookup_sandbox_fixture(*args: object, **kwargs: object) -> object:
    # Resolve only synthetic fixture state in a sandbox service.
    return {"route": "sandbox"}


def arrange_safe_fixture(_call: object) -> None:
    # Read only call.evaluation.fixture and record synthetic facts.
    return None


def record_safe_observation(_call: object, **_values: object) -> None:
    # A reviewed no-network replacement for production analytics.
    return None


evaluation_contract = EvaluationWorkerContract(
    profile_id="birla-uat@1",
    worker_revision=os.environ["BIRLA_WORKER_REVISION"],
    vaani_version_specifier=os.environ["BIRLA_VAANI_VERSION_SPECIFIER"],
    authority=EvaluationAttemptAuthority(
        os.environ["BIRLA_EVALUATION_SHARED_SECRET"]
    ),
    safety_manifest=EvaluationSafetyManifest(
        {
            EvaluationCapability.TRANSFER: EvaluationCapabilityBinding(
                EvaluationCapabilityRoute.FORBIDDEN
            ),
            EvaluationCapability.RECORDING: EvaluationCapabilityBinding(
                EvaluationCapabilityRoute.FORBIDDEN
            ),
            EvaluationCapability.NOTIFICATION: EvaluationCapabilityBinding(
                EvaluationCapabilityRoute.FAKE,
                "birla-eval-notification-sink",
            ),
            EvaluationCapability.PAYMENT: EvaluationCapabilityBinding(
                EvaluationCapabilityRoute.FORBIDDEN
            ),
            EvaluationCapability.CUSTOMER_MUTATION: EvaluationCapabilityBinding(
                EvaluationCapabilityRoute.FORBIDDEN
            ),
            EvaluationCapability.PROVIDER_LOOKUP: EvaluationCapabilityBinding(
                EvaluationCapabilityRoute.SANDBOX,
                "birla-sandbox-lookup",
            ),
        }
    ),
    safe_operations=(
        EvaluationSafeOperation(
            EvaluationCapability.NOTIFICATION,
            "birla-eval-notification-sink",
            record_fake_notification,
        ),
        EvaluationSafeOperation(
            EvaluationCapability.PROVIDER_LOOKUP,
            "birla-sandbox-lookup",
            lookup_sandbox_fixture,
        ),
    ),
    safe_preparations=(
        EvaluationSafePreparation("birla-eval-arrangement", arrange_safe_fixture),
    ),
    safe_prompt_update_observers=(
        EvaluationSafePromptUpdateObserver("birla-eval-observer", record_safe_observation),
    ),
    protected_literals=(os.environ["BIRLA_EVALUATION_SHARED_SECRET"],),
)


def production_notification(message: str) -> object:
    # This production function keeps its normal production behaviour.
    return message


def production_arrangement(_call: object) -> None:
    # Normal production preparation remains available outside evaluation.
    return None


def production_prompt_analytics(_call: object, **_values: object) -> None:
    # Normal production analytics remains available outside evaluation.
    return None


assistant.prepare(
    production_arrangement,
    evaluation_binding=EvaluationPreparationBinding("birla-eval-arrangement"),
)


assistant.tool(
    production_notification,
    evaluation_binding=EvaluationToolBinding(
        EvaluationCapability.NOTIFICATION,
        "birla-eval-notification-sink",
    ),
    on_prompt_update_committed=production_prompt_analytics,
    evaluation_observer_binding=EvaluationPromptUpdateObserverBinding(
        "birla-eval-observer"
    ),
)

run_livekit_worker(
    assistant,
    agent_name="birla-evaluation",
    evaluation_contract=evaluation_contract,
    evaluation_replay_store=RedisAttemptStore(),
)

The contract is accepted only in the development environment. It cannot be combined with the legacy evaluation_evidence=True compatibility mode. A missing safe operation, unsafe binding, worker tool binding, room identity, dispatch identity, or replay store rejects evaluation startup or admission.

Authorize one room-bound attempt

The evaluator resets synthetic state and then creates generated opaque identifiers. They are not emails, phone numbers, Aadhaar/PAN, account numbers, credentials, or customer record identifiers. The LiveKit room name and the agent dispatch id must exactly equal the signed values.

import secrets
import time
import json

from livekit import api
from vaani.evaluation import EvaluationFixtureReference


def opaque(prefix: str) -> str:
    return prefix + secrets.token_hex(16)


preparation = evaluation_contract.prepare_dispatch(
    attempt_id=opaque("vea_"),
    room_id=opaque("ver_"),
    consumer_identity=opaque("vep_"),
    fixture=EvaluationFixtureReference(opaque("vef_")),
    expires_at_unix_ns=time.time_ns() + 120_000_000_000,
)
dispatch = await livekit_api.agent_dispatch.create_dispatch(
    api.CreateAgentDispatchRequest(
        agent_name="birla-evaluation",
        room=preparation.room_id,
        metadata=json.dumps(preparation.job_metadata()),
    )
)
attempt = preparation.bind_livekit_dispatch(dispatch.id)
await livekit_api.room.update_room_metadata(
    api.UpdateRoomMetadataRequest(
        room=attempt.room_id,
        metadata=json.dumps(evaluation_contract.room_metadata(attempt)),
    )
)

CreateAgentDispatch returns the authoritative server-owned AD_… identity; the evaluator never invents one. LiveKit exposes no dispatch-update API, so the signed authorization is written to room metadata after that ID is returned. The worker waits fail-closed for job.job.dispatch_id, verifies it through agent_dispatch.get_dispatch, then reads the exact room metadata through room.list_rooms before consuming the attempt. The LiveKit job id (AJ_…) and dispatch id (AD_…) are distinct and never substituted for one another. Two workers sharing a correctly atomic store can accept the same attempt only once; process-local memory is never represented as distributed replay protection.

The worker binds the actual raw call/participant identity internally and emits only HMAC-derived vec_... and vep_... pseudonyms. Evidence creation rejects known credential and PII shapes, along with configured exact protected literals.

Verify ready and final evidence

Before sending its first simulated turn, a receiver must verify a reliable, server-authored packet on vaani.evaluation.contract.v1 with an exact EvaluationEvidenceExpectation. Verification checks the exact keys and byte bound, evidence-domain signature, stage, expiry, room/dispatch/attempt, pseudonyms, fixture, worker, safety manifest, and final result digest. A dispatch-domain signature cannot verify as evidence.

from vaani.evaluation import (
    EvaluationEvidenceExpectation,
    EvaluationEvidenceVerifier,
    EvaluationRuntimeExpectation,
)


ready_expectation = EvaluationEvidenceExpectation(
    stage="ready",
    attempt_id=attempt.attempt_id,
    room_id=attempt.room_id,
    dispatch_id=attempt.dispatch_id,
    call_pseudonym="vec_" + "0" * 64,
    participant_pseudonym="vep_" + "0" * 64,
    fixture_id=attempt.fixture.fixture_id,
    expires_at_unix_ns=attempt.expires_at_unix_ns,
    runtime=EvaluationRuntimeExpectation(
        profile_id="birla-uat@1",
        worker_revision=os.environ["BIRLA_WORKER_REVISION"],
        vaani_version_specifier=os.environ["BIRLA_VAANI_VERSION_SPECIFIER"],
    ),
)
receiver = EvaluationEvidenceVerifier(
    EvaluationAttemptAuthority(os.environ["BIRLA_EVALUATION_SHARED_SECRET"]),
    receiver_safety_manifest,  # independently configured public safety summary
    protected_literals=(os.environ["BIRLA_EVALUATION_SHARED_SECRET"],),
)
assert receiver.verify_evidence(
    ready_packet,
    expected=ready_expectation,
)

After closure the worker emits a final packet with deterministic bounded facts and the SHA-256 digest of the canonical CallResult. Accept a product verdict only when the final packet verifies against a final expectation containing that same digest. Missing, malformed, unrelated, expired, or unsigned evidence is an infrastructure failure, never a passing result.