praval.eval

Provider-neutral evaluation contracts, datasets, and persistence.

class praval.eval.AgentJudge(*, agent, allowed_tools=(), tool_policy='evaluation_safe', max_tool_rounds=None, **kwargs)[source]

Bases: _StructuredJudge

Ordinary Praval agent judge with narrowed, evaluation-safe tools.

Parameters:

  • agent (Agent)

  • allowed_tools (tuple[str, ...])

  • tool_policy (Literal['evaluation_safe', 'read_only'])

  • max_tool_rounds (int | None)

  • kwargs (Any)

__init__(*, agent, allowed_tools=(), tool_policy='evaluation_safe', max_tool_rounds=None, **kwargs)[source]
Parameters:

  • agent (Agent)

  • allowed_tools (tuple[str, ...])

  • tool_policy (Literal['evaluation_safe', 'read_only'])

  • max_tool_rounds (int | None)

  • kwargs (Any)

Return type:

None

classmethod from_config(name, config, *, rubric, rubric_version, judge_version='1', registry=None)[source]

Resolve an ordinary named evaluator agent from validated config.

Return type:

AgentJudge

Parameters:

  • name (str)

  • config (Any)

  • rubric (str)

  • rubric_version (str)

  • judge_version (str)

  • registry (Any | None)

property model_name: str | None

Return the configured evaluator model identity when known.

class praval.eval.AgentEvaluationTarget(agent, *, isolate_conversation=True)[source]

Bases: object

Run one ordinary agent per case and capture its single observation.

Parameters:

  • agent (Agent)

  • isolate_conversation (bool)

__init__(agent, *, isolate_conversation=True)[source]
Parameters:

  • agent (Agent)

  • isolate_conversation (bool)

Return type:

None

async evaluate(case)[source]

Execute one case through the normal agent runtime.

Return type:

TargetResult

Parameters:

case (LoadedEvalCase)

class praval.eval.AttemptStatus(*values)[source]

Bases: str, Enum

Outcome of one evaluation job attempt.

RUNNING = 'running'
SUCCEEDED = 'succeeded'
FAILED = 'failed'
TIMEOUT = 'timeout'
CANCELLED = 'cancelled'
exception praval.eval.EvalDatasetError[source]

Bases: ValueError

A JSONL evaluation dataset violates its bounded public schema.

class praval.eval.EvalRunner(*, store, target, judges, metrics=None, concurrency=4, clock=None)[source]

Bases: object

Run a loaded suite with bounded target concurrency and persistence.

Parameters:
__init__(*, store, target, judges, metrics=None, concurrency=4, clock=None)[source]
Parameters:
async run(loaded_suite, *, evaluation_run_id=None)[source]

Execute, persist, and summarize one offline evaluation suite.

Return type:

EvaluationResult

Parameters:
class praval.eval.EvalCase(**data)[source]

Bases: _EvaluationModel

One versioned case whose content is represented by safe references.

Parameters:
schema_version: Literal[1]
case_id: str
name: str
input: ContentReference
expected_output: ContentReference | None
reference_contexts: tuple[ContentReference, ...]
expected_tool_calls: tuple[str, ...]
metadata: tuple[EvaluationMetadata, ...]
tags: tuple[str, ...]
classmethod validate_bounded_names(values)[source]

Reject blank, oversized, or duplicate names.

Return type:

tuple[str, ...]

Parameters:

values (tuple[str, ...])

classmethod validate_metadata_keys(values)[source]

Reject duplicate metadata keys.

Return type:

tuple[EvaluationMetadata, ...]

Parameters:

values (tuple[EvaluationMetadata, ...])

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class praval.eval.EvalSuite(**data)[source]

Bases: _EvaluationModel

Stable selection of cases, target, judges, metrics, and gates.

Parameters:

  • schema_version (Literal[1])

  • suite_id (str)

  • name (str)

  • target (str)

  • case_ids (tuple[str, ...])

  • judges (tuple[str, ...])

  • metrics (tuple[str, ...])

  • gates (tuple[Gate, ...])

  • tags (tuple[str, ...])

schema_version: Literal[1]
suite_id: str
name: str
target: str
case_ids: tuple[str, ...]
judges: tuple[str, ...]
metrics: tuple[str, ...]
gates: tuple[Gate, ...]
tags: tuple[str, ...]
classmethod validate_unique_names(values)[source]

Keep suite selectors stable and unambiguous.

Return type:

tuple[str, ...]

Parameters:

values (tuple[str, ...])

validate_gate_ids()[source]

Reject duplicate gate identities within a suite.

Return type:

EvalSuite

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class praval.eval.EvaluationAttempt(**data)[source]

Bases: _EvaluationModel

One bounded attempt to process an evaluation job.

Parameters:

  • schema_version (Literal[1])

  • attempt_id (str)

  • job_id (str)

  • attempt_number (int)

  • status (AttemptStatus)

  • started_at (datetime)

  • ended_at (datetime | None)

  • duration_ms (float | None)

  • usage (TokenUsageObservation | None)

  • cost_usd (float | None)

  • error_type (str | None)

schema_version: Literal[1]
attempt_id: str
job_id: str
attempt_number: int
status: AttemptStatus
started_at: datetime
ended_at: datetime | None
duration_ms: float | None
usage: TokenUsageObservation | None
cost_usd: float | None
error_type: str | None
classmethod create(**values)[source]

Create an attempt with a stable job-and-attempt-number identity.

Return type:

EvaluationAttempt

Parameters:

values (Any)

classmethod normalize_timestamp(value)[source]

Require timezone-aware attempt timestamps.

Return type:

datetime | None

Parameters:

value (datetime | None)

validate_lifecycle()[source]

Keep attempt status, timing, and errors consistent.

Return type:

EvaluationAttempt

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class praval.eval.EvaluationBaseline(**data)[source]

Bases: _EvaluationModel

Explicit promotion of one completed run as a suite baseline.

Parameters:

  • schema_version (Literal[1])

  • baseline_id (str)

  • suite_id (str)

  • source_evaluation_run_id (str)

  • promoted_at (datetime)

  • promoted_by (str)

  • active (bool)

schema_version: Literal[1]
baseline_id: str
suite_id: str
source_evaluation_run_id: str
promoted_at: datetime
promoted_by: str
active: bool
classmethod create(**values)[source]

Create an idempotent promotion identity for a suite and run.

Return type:

EvaluationBaseline

Parameters:

values (Any)

classmethod normalize_promoted_at(value)[source]

Require a timezone-aware promotion time.

Return type:

datetime

Parameters:

value (datetime)

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

exception praval.eval.EvaluationConflictError[source]

Bases: EvaluationStoreError

A natural idempotency key was reused with different data.

exception praval.eval.EvaluationExecutionError[source]

Bases: RuntimeError

An evaluation cannot produce a valid persisted result.

class praval.eval.EvaluationJob(**data)[source]

Bases: _EvaluationModel

Durable unit of deferred evaluation work.

Parameters:

  • schema_version (Literal[1])

  • job_id (str)

  • evaluation_run_id (str)

  • suite_id (str)

  • case_id (str)

  • subject_id (str)

  • status (JobStatus)

  • available_at (datetime)

  • lease_owner (str | None)

  • lease_expires_at (datetime | None)

  • attempt_count (int)

  • max_attempts (int)

  • error_type (str | None)

  • created_at (datetime)

  • updated_at (datetime)

schema_version: Literal[1]
job_id: str
evaluation_run_id: str
suite_id: str
case_id: str
subject_id: str
status: JobStatus
available_at: datetime
lease_owner: str | None
lease_expires_at: datetime | None
attempt_count: int
max_attempts: int
error_type: str | None
created_at: datetime
updated_at: datetime
classmethod create(**values)[source]

Create a pending job with a stable at-least-once delivery identity.

Return type:

EvaluationJob

Parameters:

values (Any)

classmethod normalize_timestamp(value)[source]

Require timezone-aware job timestamps.

Return type:

datetime | None

Parameters:

value (datetime | None)

validate_lifecycle()[source]

Keep leasing, attempts, timing, and terminal errors consistent.

Return type:

EvaluationJob

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class praval.eval.EvaluationMetadata(**data)[source]

Bases: _EvaluationModel

One bounded, query-safe scalar attached to an evaluation case.

Parameters:

  • key (str)

  • value (str | int | float | bool | None)

key: str
value: str | int | float | bool | None
classmethod require_finite_value(value)[source]

Reject non-finite floats that cannot round-trip through strict JSON.

Return type:

str | int | float | bool | None

Parameters:

value (str | int | float | bool | None)

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class praval.eval.EvaluationResult(**data)[source]

Bases: _EvaluationModel

Immutable summary of a completed evaluation run.

Parameters:

  • schema_version (Literal[1])

  • evaluation_run_id (str)

  • status (Literal[EvaluationRunStatus.COMPLETED, EvaluationRunStatus.FAILED, EvaluationRunStatus.CANCELLED])

  • total_cases (int)

  • passed_cases (int)

  • failed_cases (int)

  • errored_cases (int)

  • skipped_cases (int)

  • metric_result_ids (tuple[str, ...])

  • judge_result_ids (tuple[str, ...])

  • gate_result_ids (tuple[str, ...])

  • completed_at (datetime)

schema_version: Literal[1]
evaluation_run_id: str
status: Literal[EvaluationRunStatus.COMPLETED, EvaluationRunStatus.FAILED, EvaluationRunStatus.CANCELLED]
total_cases: int
passed_cases: int
failed_cases: int
errored_cases: int
skipped_cases: int
metric_result_ids: tuple[str, ...]
judge_result_ids: tuple[str, ...]
gate_result_ids: tuple[str, ...]
completed_at: datetime
classmethod normalize_completed_at(value)[source]

Require a timezone-aware completion time.

Return type:

datetime

Parameters:

value (datetime)

validate_case_totals()[source]

Require the terminal case counts to add up.

Return type:

EvaluationResult

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class praval.eval.EvaluationRun(**data)[source]

Bases: _EvaluationModel

Lifecycle record for one execution of an evaluation suite.

Parameters:

  • schema_version (Literal[1])

  • evaluation_run_id (str)

  • suite_id (str)

  • target (str)

  • status (EvaluationRunStatus)

  • started_at (datetime)

  • completed_at (datetime | None)

  • baseline_id (str | None)

  • error_type (str | None)

schema_version: Literal[1]
evaluation_run_id: str
suite_id: str
target: str
status: EvaluationRunStatus
started_at: datetime
completed_at: datetime | None
baseline_id: str | None
error_type: str | None
classmethod normalize_timestamp(value)[source]

Require timezone-aware timestamps and normalize them to UTC.

Return type:

datetime | None

Parameters:

value (datetime | None)

validate_lifecycle()[source]

Keep terminal state, timing, and error fields consistent.

Return type:

EvaluationRun

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class praval.eval.EvaluationRunStatus(*values)[source]

Bases: str, Enum

Lifecycle of an evaluation suite run.

PENDING = 'pending'
RUNNING = 'running'
COMPLETED = 'completed'
FAILED = 'failed'
CANCELLED = 'cancelled'
class praval.eval.EvaluationStore(*args, **kwargs)[source]

Bases: Protocol

Common asynchronous persistence and query contract for evaluation.

async migrate()[source]

Apply all pending schema migrations idempotently.

Return type:

None

async schema_version()[source]

Return the latest applied schema version.

Return type:

int

async close()[source]

Release store-owned resources.

Return type:

None

async put_case(case)[source]

Persist an immutable case idempotently.

Return type:

EvalCase

Parameters:

case (EvalCase)

async get_case(case_id)[source]

Load one case by identity.

Return type:

EvalCase | None

Parameters:

case_id (str)

async list_cases(*, limit=100)[source]

List cases in stable identity order.

Return type:

list[EvalCase]

Parameters:

limit (int)

async put_suite(suite)[source]

Persist an immutable suite idempotently.

Return type:

EvalSuite

Parameters:

suite (EvalSuite)

async get_suite(suite_id)[source]

Load one suite by identity.

Return type:

EvalSuite | None

Parameters:

suite_id (str)

async put_run(run)[source]

Create or update a run lifecycle record.

Return type:

EvaluationRun

Parameters:

run (EvaluationRun)

async get_run(evaluation_run_id)[source]

Load one evaluation run.

Return type:

EvaluationRun | None

Parameters:

evaluation_run_id (str)

async list_runs(*, suite_id=None, limit=100)[source]

List recent runs, optionally restricted to a suite.

Return type:

list[EvaluationRun]

Parameters:

  • suite_id (str | None)

  • limit (int)

async put_subject(subject)[source]

Persist one immutable agent or workflow subject.

Return type:

EvaluationSubject

Parameters:

subject (EvaluationSubject)

async get_subject(subject_id)[source]

Load one subject.

Return type:

EvaluationSubject | None

Parameters:

subject_id (str)

async list_subjects(*, evaluation_run_id, limit=100)[source]

List subjects belonging to a run.

Return type:

list[EvaluationSubject]

Parameters:

  • evaluation_run_id (str)

  • limit (int)

async put_metric_result(result)[source]

Persist one immutable, idempotent metric result.

Return type:

MetricResult

Parameters:

result (MetricResult)

async list_metric_results(*, evaluation_run_id, metric=None, limit=1000)[source]

List metric results for a run.

Return type:

list[MetricResult]

Parameters:

  • evaluation_run_id (str)

  • metric (str | None)

  • limit (int)

async put_judge_result(result)[source]

Persist one immutable, idempotent judge result.

Return type:

JudgeResult

Parameters:

result (JudgeResult)

async list_judge_results(*, evaluation_run_id, limit=1000)[source]

List judge results for a run.

Return type:

list[JudgeResult]

Parameters:

  • evaluation_run_id (str)

  • limit (int)

async put_gate_result(result)[source]

Persist one immutable, idempotent gate result.

Return type:

GateResult

Parameters:

result (GateResult)

async list_gate_results(*, evaluation_run_id, limit=1000)[source]

List gate decisions for a run.

Return type:

list[GateResult]

Parameters:

  • evaluation_run_id (str)

  • limit (int)

async put_evaluation_result(result)[source]

Persist one immutable terminal run summary.

Return type:

EvaluationResult

Parameters:

result (EvaluationResult)

async get_evaluation_result(evaluation_run_id)[source]

Load a terminal run summary.

Return type:

EvaluationResult | None

Parameters:

evaluation_run_id (str)

async promote_baseline(baseline)[source]

Atomically make an explicit baseline active for its suite.

Return type:

EvaluationBaseline

Parameters:

baseline (EvaluationBaseline)

async get_active_baseline(suite_id)[source]

Load the active baseline for a suite.

Return type:

EvaluationBaseline | None

Parameters:

suite_id (str)

async list_baselines(*, suite_id, limit=100)[source]

List baseline promotion history for a suite.

Return type:

list[EvaluationBaseline]

Parameters:

  • suite_id (str)

  • limit (int)

async put_job(job)[source]

Create or update a durable evaluation job.

Return type:

EvaluationJob

Parameters:

job (EvaluationJob)

async get_job(job_id)[source]

Load one job.

Return type:

EvaluationJob | None

Parameters:

job_id (str)

async list_jobs(*, status=None, limit=100)[source]

List jobs, optionally filtered by status.

Return type:

list[EvaluationJob]

Parameters:
async lease_job(*, worker_id, now, lease_seconds)[source]

Atomically lease one ready or expired job and count its attempt.

Return type:

EvaluationJob | None

Parameters:

  • worker_id (str)

  • now (datetime)

  • lease_seconds (float)

async complete_job(*, job_id, worker_id, now)[source]

Atomically complete a job held by worker_id.

Return type:

EvaluationJob

Parameters:

  • job_id (str)

  • worker_id (str)

  • now (datetime)

async retry_job(*, job_id, worker_id, now, error_type, retry_delay_seconds)[source]

Release a failed lease for retry or move it to dead letter.

Return type:

EvaluationJob

Parameters:

  • job_id (str)

  • worker_id (str)

  • now (datetime)

  • error_type (str)

  • retry_delay_seconds (float)

async put_attempt(attempt)[source]

Persist one immutable job attempt.

Return type:

EvaluationAttempt

Parameters:

attempt (EvaluationAttempt)

async list_attempts(*, job_id, limit=100)[source]

List attempts for one job in attempt order.

Return type:

list[EvaluationAttempt]

Parameters:

  • job_id (str)

  • limit (int)

__init__(*args, **kwargs)
exception praval.eval.EvaluationStoreError[source]

Bases: RuntimeError

Base error raised by an evaluation store.

class praval.eval.EvaluationSubject(**data)[source]

Bases: _EvaluationModel

Exactly one agent or workflow observation selected for evaluation.

Parameters:

  • schema_version (Literal[1])

  • subject_id (str)

  • evaluation_run_id (str)

  • case_id (str)

  • observation_id (str)

  • execution_run_id (str)

  • kind (ObservationKind)

  • response_id (str | None)

  • observation (ExecutionObservation)

schema_version: Literal[1]
subject_id: str
evaluation_run_id: str
case_id: str
observation_id: str
execution_run_id: str
kind: ObservationKind
response_id: str | None
observation: ExecutionObservation
classmethod from_observation(*, evaluation_run_id, case_id, observation)[source]

Map one frozen runtime observation into one evaluation subject.

Return type:

EvaluationSubject

Parameters:
validate_observation_identity()[source]

Prevent duplicated query fields from drifting from the payload.

Return type:

EvaluationSubject

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class praval.eval.EvaluationTarget(*args, **kwargs)[source]

Bases: Protocol

Async target capable of evaluating one loaded case.

async evaluate(case)[source]

Execute one case and return its observation and ephemeral output.

Return type:

TargetResult

Parameters:

case (LoadedEvalCase)

__init__(*args, **kwargs)
class praval.eval.ExactMatchMetric(*, clock=None)[source]

Bases: _DeterministicMetric

Compare canonical JSON values without model calls.

Parameters:

clock (Callable[[], datetime] | None)

name: str = 'exact_match'
async evaluate(context)[source]

Evaluate one completed target subject.

Return type:

MetricResult

Parameters:

context (JudgeContext)

class praval.eval.Gate(**data)[source]

Bases: _EvaluationModel

Declarative threshold over one aggregated metric.

Parameters:

  • schema_version (Literal[1])

  • gate_id (str)

  • metric (str)

  • aggregation (GateAggregation)

  • operator (GateOperator)

  • threshold (float)

  • required (bool)

  • percentile (float | None)

  • baseline_max_regression (float | None)

schema_version: Literal[1]
gate_id: str
metric: str
aggregation: GateAggregation
operator: GateOperator
threshold: float
required: bool
percentile: float | None
baseline_max_regression: float | None
classmethod validate_finite_gate_value(value)[source]

Reject non-finite thresholds and regression bounds.

Return type:

float | None

Parameters:

value (float | None)

validate_percentile()[source]

Require a percentile only for percentile aggregation.

Return type:

Gate

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class praval.eval.GateAggregation(*values)[source]

Bases: str, Enum

Supported aggregation for a quality gate.

MEAN = 'mean'
MINIMUM = 'minimum'
MAXIMUM = 'maximum'
PERCENTILE = 'percentile'
COUNT = 'count'
PASS_RATE = 'pass_rate'
exception praval.eval.GateEvaluationError[source]

Bases: ValueError

Gate input is ambiguous or violates deterministic evaluation policy.

class praval.eval.GateOperator(*values)[source]

Bases: str, Enum

Comparison applied to an aggregate gate value.

GREATER_THAN_OR_EQUAL = '>='
GREATER_THAN = '>'
LESS_THAN_OR_EQUAL = '<='
LESS_THAN = '<'
EQUAL = '=='
class praval.eval.GateResult(**data)[source]

Bases: _EvaluationModel

Persisted decision from applying a gate to an aggregate.

Parameters:

  • schema_version (Literal[1])

  • gate_result_id (str)

  • evaluation_run_id (str)

  • gate_id (str)

  • metric (str)

  • status (GateStatus)

  • observed_value (float | None)

  • threshold (float)

  • baseline_value (float | None)

  • regression_delta (float | None)

  • error_type (str | None)

  • created_at (datetime)

schema_version: Literal[1]
gate_result_id: str
evaluation_run_id: str
gate_id: str
metric: str
status: GateStatus
observed_value: float | None
threshold: float
baseline_value: float | None
regression_delta: float | None
error_type: str | None
created_at: datetime
classmethod validate_finite_value(value)[source]

Keep persisted gate values portable across JSON and databases.

Return type:

float | None

Parameters:

value (float | None)

classmethod create(**values)[source]

Create one idempotent gate decision per run and gate.

Return type:

GateResult

Parameters:

values (Any)

classmethod normalize_created_at(value)[source]

Require a timezone-aware creation time.

Return type:

datetime

Parameters:

value (datetime)

validate_result()[source]

Keep gate errors separate from measured decisions.

Return type:

GateResult

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class praval.eval.GateStatus(*values)[source]

Bases: str, Enum

Outcome of applying one quality gate.

PASSED = 'passed'
FAILED = 'failed'
ERROR = 'error'
class praval.eval.JobStatus(*values)[source]

Bases: str, Enum

Durable evaluation job lifecycle.

PENDING = 'pending'
LEASED = 'leased'
AWAITING_INTERVENTION = 'awaiting_intervention'
COMPLETED = 'completed'
FAILED = 'failed'
DEAD_LETTER = 'dead_letter'
class praval.eval.Judge(*args, **kwargs)[source]

Bases: Protocol

Configured evaluator that returns a validated judge result.

name: str
async evaluate(context)[source]

Evaluate one completed target subject.

Return type:

JudgeResult

Parameters:

context (JudgeContext)

__init__(*args, **kwargs)
exception praval.eval.JudgeConfigurationError[source]

Bases: ValueError

A judge configuration violates identity, safety, or resource policy.

class praval.eval.JudgeContext(evaluation_run_id, case, subject, target_result)[source]

Bases: object

Bounded input supplied to one configured judge.

Parameters:
evaluation_run_id: str
case: LoadedEvalCase
subject: EvaluationSubject
target_result: TargetResult
__init__(evaluation_run_id, case, subject, target_result)
Parameters:
Return type:

None

class praval.eval.JudgeResult(**data)[source]

Bases: _EvaluationModel

Validated result from one versioned evaluator agent or model.

Parameters:

  • schema_version (Literal[1])

  • judge_result_id (str)

  • evaluation_run_id (str)

  • case_id (str)

  • subject_id (str)

  • judge (str)

  • judge_version (str)

  • prompt_sha256 (str)

  • rubric_version (str)

  • status (ResultStatus)

  • score (float | None)

  • label (str | None)

  • explanation (str | None)

  • evidence (tuple[ContentReference, ...])

  • privacy (ObservationPrivacy)

  • model (str | None)

  • usage (TokenUsageObservation | None)

  • cost_usd (float | None)

  • duration_ms (float | None)

  • attempt_count (int)

  • error_type (str | None)

  • created_at (datetime)

schema_version: Literal[1]
judge_result_id: str
evaluation_run_id: str
case_id: str
subject_id: str
judge: str
judge_version: str
prompt_sha256: str
rubric_version: str
status: ResultStatus
score: float | None
label: str | None
explanation: str | None
evidence: tuple[ContentReference, ...]
privacy: ObservationPrivacy
model: str | None
usage: TokenUsageObservation | None
cost_usd: float | None
duration_ms: float | None
attempt_count: int
error_type: str | None
created_at: datetime
classmethod create(**values)[source]

Create a judge result with its natural idempotency identity.

Return type:

JudgeResult

Parameters:

values (Any)

classmethod normalize_created_at(value)[source]

Require a timezone-aware creation time.

Return type:

datetime

Parameters:

value (datetime)

validate_result()[source]

Enforce result status and metadata-only privacy defaults.

Return type:

JudgeResult

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

exception praval.eval.JudgeResponseError[source]

Bases: ValueError

A judge returned content that does not match the strict result schema.

class praval.eval.LoadedEvalCase(case, input, expected_output, reference_contexts)[source]

Bases: object

Persistable case metadata paired with ephemeral execution material.

Parameters:

  • case (EvalCase)

  • input (Any)

  • expected_output (Any | None)

  • reference_contexts (tuple[Any, ...])

case: EvalCase
input: Any
expected_output: Any | None
reference_contexts: tuple[Any, ...]
__init__(case, input, expected_output, reference_contexts)
Parameters:

  • case (EvalCase)

  • input (Any)

  • expected_output (Any | None)

  • reference_contexts (tuple[Any, ...])

Return type:

None

class praval.eval.LoadedEvalSuite(suite, cases)[source]

Bases: object

One selected suite and its ephemeral case material.

Parameters:
suite: EvalSuite
cases: tuple[LoadedEvalCase, ...]
__init__(suite, cases)
Parameters:
Return type:

None

class praval.eval.MetricResult(**data)[source]

Bases: _EvaluationModel

Normalized result from one deterministic or plugin metric.

Parameters:

  • schema_version (Literal[1])

  • metric_result_id (str)

  • evaluation_run_id (str)

  • case_id (str)

  • subject_id (str)

  • metric (str)

  • metric_version (str)

  • status (ResultStatus)

  • score (float | None)

  • label (str | None)

  • error_type (str | None)

  • created_at (datetime)

schema_version: Literal[1]
metric_result_id: str
evaluation_run_id: str
case_id: str
subject_id: str
metric: str
metric_version: str
status: ResultStatus
score: float | None
label: str | None
error_type: str | None
created_at: datetime
classmethod create(**values)[source]

Create a result with its natural idempotency identity.

Return type:

MetricResult

Parameters:

values (Any)

classmethod normalize_created_at(value)[source]

Require a timezone-aware creation time.

Return type:

datetime

Parameters:

value (datetime)

validate_result()[source]

Keep scores and errors consistent with the result status.

Return type:

MetricResult

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class praval.eval.MetricComparison(kind, name, version, current_value, baseline_value, delta, regressed)[source]

Bases: object

One metric or judge score comparison between two completed runs.

Parameters:

  • kind (Literal['metric', 'judge'])

  • name (str)

  • version (str)

  • current_value (float | None)

  • baseline_value (float | None)

  • delta (float | None)

  • regressed (bool)

kind: Literal['metric', 'judge']
name: str
version: str
current_value: float | None
baseline_value: float | None
delta: float | None
regressed: bool
__init__(kind, name, version, current_value, baseline_value, delta, regressed)
Parameters:

  • kind (Literal['metric', 'judge'])

  • name (str)

  • version (str)

  • current_value (float | None)

  • baseline_value (float | None)

  • delta (float | None)

  • regressed (bool)

Return type:

None

class praval.eval.Metric(*args, **kwargs)[source]

Bases: Protocol

Public contract implemented by deterministic and plugin metrics.

name: str
version: str
async evaluate(context)[source]

Evaluate one completed immutable subject.

Return type:

MetricResult

Parameters:

context (JudgeContext)

__init__(*args, **kwargs)
exception praval.eval.MetricPluginError[source]

Bases: ValueError

A discovered metric plugin violates the public plugin contract.

class praval.eval.ModelJudge(*, runtime, **kwargs)[source]

Bases: _StructuredJudge

Direct foundation-model judge using a provider-neutral ModelRuntime.

Parameters:
__init__(*, runtime, **kwargs)[source]
Parameters:
Return type:

None

classmethod from_config(name, config, *, rubric, rubric_version, judge_version='1', provider_factory=None)[source]

Resolve a direct-model judge from a validated PravalConfig.

Return type:

ModelJudge

Parameters:

  • name (str)

  • config (Any)

  • rubric (str)

  • rubric_version (str)

  • judge_version (str)

  • provider_factory (Callable[[str, Any], Any] | None)

property model_name: str | None

Return the configured evaluator model identity when known.

class praval.eval.OnlineContextLoader(*args, **kwargs)[source]

Bases: Protocol

Resolve ephemeral candidate content away from the request path.

async __call__(job, subject)[source]

Build the bounded judge context for one durable subject.

Return type:

JudgeContext

Parameters:
__init__(*args, **kwargs)
class praval.eval.OnlineEvaluationProcessor(*args, **kwargs)[source]

Bases: Protocol

Application evaluator invoked only by a durable worker.

async __call__(job, subject)[source]

Persist idempotent metric or judge results for one subject.

Return type:

None

Parameters:
__init__(*args, **kwargs)
class praval.eval.OnlineEvaluationService(*, store, suite, processor, config=None, clock=None)[source]

Bases: object

Bounded recorder, durable scheduler, and post-hoc worker lifecycle.

Parameters:
__init__(*, store, suite, processor, config=None, clock=None)[source]
Parameters:
Return type:

None

property queue_depth: int
property active: bool
stats()[source]

Return an atomic metadata-only health snapshot.

Return type:

OnlineEvaluationStats

async start()[source]

Migrate storage and start explicitly owned async tasks.

Return type:

None

record(observation)[source]

Schedule one completed observation without storage or judge calls.

Return type:

None

Parameters:

observation (ExecutionObservation)

async shutdown(timeout_seconds=None)[source]

Stop acceptance and make one bounded drain/cancellation attempt.

Return type:

bool

Parameters:

timeout_seconds (float | None)

class praval.eval.OnlineEvaluationStats(sampled, skipped, enqueued, persisted, processed, retries, dropped, store_failures, processor_failures, dead_lettered, queue_depth, queue_capacity)[source]

Bases: object

Bounded service-health snapshot with no candidate content.

Parameters:

  • sampled (int)

  • skipped (int)

  • enqueued (int)

  • persisted (int)

  • processed (int)

  • retries (int)

  • dropped (int)

  • store_failures (int)

  • processor_failures (int)

  • dead_lettered (int)

  • queue_depth (int)

  • queue_capacity (int)

sampled: int
skipped: int
enqueued: int
persisted: int
processed: int
retries: int
dropped: int
store_failures: int
processor_failures: int
dead_lettered: int
queue_depth: int
queue_capacity: int
__init__(sampled, skipped, enqueued, persisted, processed, retries, dropped, store_failures, processor_failures, dead_lettered, queue_depth, queue_capacity)
Parameters:

  • sampled (int)

  • skipped (int)

  • enqueued (int)

  • persisted (int)

  • processed (int)

  • retries (int)

  • dropped (int)

  • store_failures (int)

  • processor_failures (int)

  • dead_lettered (int)

  • queue_depth (int)

  • queue_capacity (int)

Return type:

None

class praval.eval.OnlineSubjectEvaluator(*, store, suite, context_loader, judges, metrics, clock=None)[source]

Bases: object

Run configured judges and metrics for a leased online subject.

Parameters:
__init__(*, store, suite, context_loader, judges, metrics, clock=None)[source]
Parameters:
Return type:

None

class praval.eval.PostgresEvaluationStore(dsn, *, min_pool_size=1, max_pool_size=10, command_timeout=30.0)[source]

Bases: object

Pooled async PostgreSQL implementation of EvaluationStore.

Parameters:

  • dsn (str)

  • min_pool_size (int)

  • max_pool_size (int)

  • command_timeout (float)

__init__(dsn, *, min_pool_size=1, max_pool_size=10, command_timeout=30.0)[source]
Parameters:

  • dsn (str)

  • min_pool_size (int)

  • max_pool_size (int)

  • command_timeout (float)

async migrate()[source]

Apply the initial schema transactionally and idempotently.

Return type:

None

async schema_version()[source]

Return the latest applied migration version.

Return type:

int

async close()[source]

Close the owned connection pool.

Return type:

None

async put_case(case)[source]

Persist an immutable case idempotently.

Return type:

EvalCase

Parameters:

case (EvalCase)

async get_case(case_id)[source]

Load one case by identity.

Return type:

EvalCase | None

Parameters:

case_id (str)

async list_cases(*, limit=100)[source]

List cases in stable identity order.

Return type:

list[EvalCase]

Parameters:

limit (int)

async put_suite(suite)[source]

Persist an immutable suite idempotently.

Return type:

EvalSuite

Parameters:

suite (EvalSuite)

async get_suite(suite_id)[source]

Load one suite by identity.

Return type:

EvalSuite | None

Parameters:

suite_id (str)

async put_run(run)[source]

Create or update a run lifecycle record.

Return type:

EvaluationRun

Parameters:

run (EvaluationRun)

async get_run(evaluation_run_id)[source]

Load one evaluation run.

Return type:

EvaluationRun | None

Parameters:

evaluation_run_id (str)

async list_runs(*, suite_id=None, limit=100)[source]

List recent runs, optionally restricted to a suite.

Return type:

list[EvaluationRun]

Parameters:

  • suite_id (str | None)

  • limit (int)

async put_subject(subject)[source]

Persist one immutable agent or workflow subject.

Return type:

EvaluationSubject

Parameters:

subject (EvaluationSubject)

async get_subject(subject_id)[source]

Load one subject.

Return type:

EvaluationSubject | None

Parameters:

subject_id (str)

async list_subjects(*, evaluation_run_id, limit=100)[source]

List subjects belonging to a run.

Return type:

list[EvaluationSubject]

Parameters:

  • evaluation_run_id (str)

  • limit (int)

async put_metric_result(result)[source]

Persist one immutable, idempotent metric result.

Return type:

MetricResult

Parameters:

result (MetricResult)

async list_metric_results(*, evaluation_run_id, metric=None, limit=1000)[source]

List metric results for a run.

Return type:

list[MetricResult]

Parameters:

  • evaluation_run_id (str)

  • metric (str | None)

  • limit (int)

async put_judge_result(result)[source]

Persist one immutable, idempotent judge result.

Return type:

JudgeResult

Parameters:

result (JudgeResult)

async list_judge_results(*, evaluation_run_id, limit=1000)[source]

List judge results for a run.

Return type:

list[JudgeResult]

Parameters:

  • evaluation_run_id (str)

  • limit (int)

async put_gate_result(result)[source]

Persist one immutable, idempotent gate result.

Return type:

GateResult

Parameters:

result (GateResult)

async list_gate_results(*, evaluation_run_id, limit=1000)[source]

List gate decisions for a run.

Return type:

list[GateResult]

Parameters:

  • evaluation_run_id (str)

  • limit (int)

async put_evaluation_result(result)[source]

Persist one immutable terminal run summary.

Return type:

EvaluationResult

Parameters:

result (EvaluationResult)

async get_evaluation_result(evaluation_run_id)[source]

Load a terminal run summary.

Return type:

EvaluationResult | None

Parameters:

evaluation_run_id (str)

async promote_baseline(baseline)[source]

Atomically make an explicit baseline active for its suite.

Return type:

EvaluationBaseline

Parameters:

baseline (EvaluationBaseline)

async get_active_baseline(suite_id)[source]

Load the active baseline for a suite.

Return type:

EvaluationBaseline | None

Parameters:

suite_id (str)

async list_baselines(*, suite_id, limit=100)[source]

List baseline promotion history for a suite.

Return type:

list[EvaluationBaseline]

Parameters:

  • suite_id (str)

  • limit (int)

async put_job(job)[source]

Create a durable job idempotently without reverting its lifecycle.

Return type:

EvaluationJob

Parameters:

job (EvaluationJob)

async get_job(job_id)[source]

Load one job.

Return type:

EvaluationJob | None

Parameters:

job_id (str)

async list_jobs(*, status=None, limit=100)[source]

List jobs, optionally filtered by status.

Return type:

list[EvaluationJob]

Parameters:
async lease_job(*, worker_id, now, lease_seconds)[source]

Atomically lease one ready job with FOR UPDATE SKIP LOCKED.

Return type:

EvaluationJob | None

Parameters:

  • worker_id (str)

  • now (datetime)

  • lease_seconds (float)

async complete_job(*, job_id, worker_id, now)[source]

Complete an actively leased job atomically.

Return type:

EvaluationJob

Parameters:

  • job_id (str)

  • worker_id (str)

  • now (datetime)

async retry_job(*, job_id, worker_id, now, error_type, retry_delay_seconds)[source]

Release a failed lease or dead-letter an exhausted job.

Return type:

EvaluationJob

Parameters:

  • job_id (str)

  • worker_id (str)

  • now (datetime)

  • error_type (str)

  • retry_delay_seconds (float)

async put_attempt(attempt)[source]

Persist one immutable job attempt and enforce its natural key.

Return type:

EvaluationAttempt

Parameters:

attempt (EvaluationAttempt)

async list_attempts(*, job_id, limit=100)[source]

List attempts for one job in attempt order.

Return type:

list[EvaluationAttempt]

Parameters:

  • job_id (str)

  • limit (int)

class praval.eval.ResultStatus(*values)[source]

Bases: str, Enum

Outcome shared by metric and judge results.

PASSED = 'passed'
FAILED = 'failed'
ERROR = 'error'
SKIPPED = 'skipped'
class praval.eval.RunComparison(current_run_id, baseline_run_id, max_regression, direction, metrics)[source]

Bases: object

Deterministic score comparison that never promotes a baseline.

Parameters:

  • current_run_id (str)

  • baseline_run_id (str)

  • max_regression (float)

  • direction (Literal['higher', 'lower'])

  • metrics (tuple[MetricComparison, ...])

current_run_id: str
baseline_run_id: str
max_regression: float
direction: Literal['higher', 'lower']
metrics: tuple[MetricComparison, ...]
property regressed: bool

Return whether any comparable score exceeded the regression bound.

__init__(current_run_id, baseline_run_id, max_regression, direction, metrics)
Parameters:

  • current_run_id (str)

  • baseline_run_id (str)

  • max_regression (float)

  • direction (Literal['higher', 'lower'])

  • metrics (tuple[MetricComparison, ...])

Return type:

None

class praval.eval.SQLiteEvaluationStore(db_path, *, busy_timeout_ms=5000)[source]

Bases: object

Async SQLite implementation of the evaluation persistence contract.

Blocking SQLite calls run in worker threads. A per-store async lock keeps transactions ordered, while WAL and a bounded busy timeout make separate store instances safe for local concurrent writers.

Parameters:

  • db_path (str | os.PathLike[str])

  • busy_timeout_ms (int)

__init__(db_path, *, busy_timeout_ms=5000)[source]
Parameters:

  • db_path (str | PathLike[str])

  • busy_timeout_ms (int)

async migrate()[source]

Apply the initial schema transactionally and idempotently.

Return type:

None

async schema_version()[source]

Return the latest applied migration version.

Return type:

int

async close()[source]

Close the store; connections are scoped per operation.

Return type:

None

async put_case(case)[source]

Persist an immutable case idempotently.

Return type:

EvalCase

Parameters:

case (EvalCase)

async get_case(case_id)[source]

Load one case by identity.

Return type:

EvalCase | None

Parameters:

case_id (str)

async list_cases(*, limit=100)[source]

List cases in stable identity order.

Return type:

list[EvalCase]

Parameters:

limit (int)

async put_suite(suite)[source]

Persist an immutable suite idempotently.

Return type:

EvalSuite

Parameters:

suite (EvalSuite)

async get_suite(suite_id)[source]

Load one suite by identity.

Return type:

EvalSuite | None

Parameters:

suite_id (str)

async put_run(run)[source]

Create or update a run lifecycle record.

Return type:

EvaluationRun

Parameters:

run (EvaluationRun)

async get_run(evaluation_run_id)[source]

Load one evaluation run.

Return type:

EvaluationRun | None

Parameters:

evaluation_run_id (str)

async list_runs(*, suite_id=None, limit=100)[source]

List recent runs, optionally restricted to a suite.

Return type:

list[EvaluationRun]

Parameters:

  • suite_id (str | None)

  • limit (int)

async put_subject(subject)[source]

Persist one immutable agent or workflow subject.

Return type:

EvaluationSubject

Parameters:

subject (EvaluationSubject)

async get_subject(subject_id)[source]

Load one subject.

Return type:

EvaluationSubject | None

Parameters:

subject_id (str)

async list_subjects(*, evaluation_run_id, limit=100)[source]

List subjects belonging to a run.

Return type:

list[EvaluationSubject]

Parameters:

  • evaluation_run_id (str)

  • limit (int)

async put_metric_result(result)[source]

Persist one immutable, idempotent metric result.

Return type:

MetricResult

Parameters:

result (MetricResult)

async list_metric_results(*, evaluation_run_id, metric=None, limit=1000)[source]

List metric results for a run.

Return type:

list[MetricResult]

Parameters:

  • evaluation_run_id (str)

  • metric (str | None)

  • limit (int)

async put_judge_result(result)[source]

Persist one immutable, idempotent judge result.

Return type:

JudgeResult

Parameters:

result (JudgeResult)

async list_judge_results(*, evaluation_run_id, limit=1000)[source]

List judge results for a run.

Return type:

list[JudgeResult]

Parameters:

  • evaluation_run_id (str)

  • limit (int)

async put_gate_result(result)[source]

Persist one immutable, idempotent gate result.

Return type:

GateResult

Parameters:

result (GateResult)

async list_gate_results(*, evaluation_run_id, limit=1000)[source]

List gate decisions for a run.

Return type:

list[GateResult]

Parameters:

  • evaluation_run_id (str)

  • limit (int)

async put_evaluation_result(result)[source]

Persist one immutable terminal run summary.

Return type:

EvaluationResult

Parameters:

result (EvaluationResult)

async get_evaluation_result(evaluation_run_id)[source]

Load a terminal run summary.

Return type:

EvaluationResult | None

Parameters:

evaluation_run_id (str)

async promote_baseline(baseline)[source]

Atomically make an explicit baseline active for its suite.

Return type:

EvaluationBaseline

Parameters:

baseline (EvaluationBaseline)

async get_active_baseline(suite_id)[source]

Load the active baseline for a suite.

Return type:

EvaluationBaseline | None

Parameters:

suite_id (str)

async list_baselines(*, suite_id, limit=100)[source]

List baseline promotion history for a suite.

Return type:

list[EvaluationBaseline]

Parameters:

  • suite_id (str)

  • limit (int)

async put_job(job)[source]

Create a durable job idempotently without reverting its lifecycle.

Return type:

EvaluationJob

Parameters:

job (EvaluationJob)

async get_job(job_id)[source]

Load one job.

Return type:

EvaluationJob | None

Parameters:

job_id (str)

async list_jobs(*, status=None, limit=100)[source]

List jobs, optionally filtered by status.

Return type:

list[EvaluationJob]

Parameters:
async lease_job(*, worker_id, now, lease_seconds)[source]

Atomically lease one ready job, including an expired prior lease.

Return type:

EvaluationJob | None

Parameters:

  • worker_id (str)

  • now (datetime)

  • lease_seconds (float)

async complete_job(*, job_id, worker_id, now)[source]

Complete an actively leased job atomically.

Return type:

EvaluationJob

Parameters:

  • job_id (str)

  • worker_id (str)

  • now (datetime)

async retry_job(*, job_id, worker_id, now, error_type, retry_delay_seconds)[source]

Release a failed lease or dead-letter an exhausted job.

Return type:

EvaluationJob

Parameters:

  • job_id (str)

  • worker_id (str)

  • now (datetime)

  • error_type (str)

  • retry_delay_seconds (float)

async put_attempt(attempt)[source]

Persist one immutable job attempt and enforce its natural key.

Return type:

EvaluationAttempt

Parameters:

attempt (EvaluationAttempt)

async list_attempts(*, job_id, limit=100)[source]

List attempts for one job in attempt order.

Return type:

list[EvaluationAttempt]

Parameters:

  • job_id (str)

  • limit (int)

class praval.eval.TargetResult(observation, output)[source]

Bases: object

One frozen observation plus output retained only for active evaluation.

Parameters:
observation: ExecutionObservation
output: Any
__init__(observation, output)
Parameters:
Return type:

None

class praval.eval.TerminalSuccessMetric(*, clock=None)[source]

Bases: _DeterministicMetric

Score the immutable target observation’s terminal status.

Parameters:

clock (Callable[[], datetime] | None)

name: str = 'terminal_success'
async evaluate(context)[source]

Evaluate one completed target subject.

Return type:

MetricResult

Parameters:

context (JudgeContext)

class praval.eval.ToolCallMatchMetric(*, clock=None)[source]

Bases: _DeterministicMetric

Require the observed ordered tool names to match the case exactly.

Parameters:

clock (Callable[[], datetime] | None)

name: str = 'tool_call_match'
async evaluate(context)[source]

Evaluate one completed target subject.

Return type:

MetricResult

Parameters:

context (JudgeContext)

praval.eval.available_metrics(*, entry_points=None)[source]

Combine built-ins and installed plugins without allowing shadowing.

Return type:

dict[str, Metric]

Parameters:

entry_points (Iterable[Any] | None)

praval.eval.builtin_metrics()[source]

Return fresh stateless built-ins for runner or CLI composition.

Return type:

dict[str, Metric]

praval.eval.evaluation_call_scope()[source]

Mark model, tool, and agent calls as evaluation work for this context.

Return type:

Iterator[None]

async praval.eval.compare_evaluation_runs(store, *, current_run_id, baseline_run_id, max_regression=0.0, direction='higher')[source]

Compare mean scores without mutating the active baseline.

Return type:

RunComparison

Parameters:

  • store (EvaluationStore)

  • current_run_id (str)

  • baseline_run_id (str)

  • max_regression (float)

  • direction (Literal['higher', 'lower'])

praval.eval.discover_metric_plugins(entry_points=None)[source]

Load installed praval.eval.metrics entry points deterministically.

Return type:

dict[str, Metric]

Parameters:

entry_points (Iterable[Any] | None)

praval.eval.evaluate_gate(evaluation_run_id, gate, results, *, baseline_results=(), created_at=None)[source]

Apply one absolute and optional baseline-relative gate.

Return type:

GateResult | None

Parameters:
praval.eval.is_evaluation_call()[source]

Return whether the current sync or async context is evaluation work.

Return type:

bool

praval.eval.load_jsonl_suite(path, *, suite_id, name, target, judges=(), metrics=(), gates=(), tags=(), case_ids=None, include_tags=(), limit=None, seed=0, max_cases=100000, max_line_bytes=1048576, max_content_bytes=262144)[source]

Load, validate, and deterministically select a JSONL evaluation suite.

Return type:

LoadedEvalSuite

Parameters:

  • path (str | Path)

  • suite_id (str)

  • name (str)

  • target (str)

  • judges (tuple[str, ...])

  • metrics (tuple[str, ...])

  • gates (tuple[Gate, ...])

  • tags (tuple[str, ...])

  • case_ids (tuple[str, ...] | None)

  • include_tags (tuple[str, ...])

  • limit (int | None)

  • seed (int)

  • max_cases (int)

  • max_line_bytes (int)

  • max_content_bytes (int)

async praval.eval.promote_evaluation_baseline(store, *, suite_id, evaluation_run_id, promoted_by, promoted_at=None)[source]

Explicitly promote one completed run after validating its suite.

Return type:

EvaluationBaseline

Parameters:

  • store (EvaluationStore)

  • suite_id (str)

  • evaluation_run_id (str)

  • promoted_by (str)

  • promoted_at (datetime | None)

praval.eval.trace_sampled(trace_id, sample_ratio)[source]

Make a stable all-or-nothing decision from a 128-bit trace identity.

Return type:

bool

Parameters:

  • trace_id (str | None)

  • sample_ratio (float)