Shared Contracts

The raes_contracts package owns neutral DTOs and enums shared across processor, runtime, backend protocols, conformance, and authoring surfaces. These contracts are the typed boundary for backend-facing runtime calls; they must not live in runtime or processor implementation modules.

Diagnostics

Shared diagnostic contracts.

class raes_contracts.diagnostics.Diagnostic

Bases: object

Structured planner/runtime message.

code: str
domain: str
address: str
message: str
severity: Severity = 'error'
property is_error: bool
__init__(code, domain, address, message, severity=Severity.ERROR)
Parameters:

  • code (str)

  • domain (str)

  • address (str)

  • message (str)

  • severity (Severity)

Return type:

None

class raes_contracts.diagnostics.DiagnosticModel

Bases: BaseModel

Closed portable diagnostic shape for published contracts.

model_config = {'extra': 'forbid', 'frozen': True}

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

code: str
domain: str
address: str
message: str
severity: Severity
class raes_contracts.diagnostics.Severity

Bases: str, Enum

Diagnostic severity level.

ERROR = 'error'
WARNING = 'warning'
INFO = 'info'
__new__(value)
raes_contracts.diagnostics.diagnostic_model(diagnostic)

Convert the internal immutable diagnostic to its closed contract model.

Parameters:

diagnostic (Diagnostic)

Return type:

DiagnosticModel

raes_contracts.diagnostics.diagnostic_payload(diagnostic)

Render a diagnostic as a JSON-ready mapping (severity as its string value).

Parameters:

diagnostic (Diagnostic)

Return type:

dict[str, Any]

Planning

Shared runtime planning contracts.

class raes_contracts.planning.ChangeAction

Bases: str, Enum

Planner reconciliation result for a resource.

CREATE = 'create'
UPDATE = 'update'
DELETE = 'delete'
UNCHANGED = 'unchanged'
__new__(value)
class raes_contracts.planning.EvaluationOp

Bases: PlanOperation

Evaluation reconciliation operation.

class raes_contracts.planning.EvaluationPlan

Bases: object

Resolved evaluation graph and reconciliation actions.

resources: dict[str, PlannedResource]
operations: list[EvaluationOp]
startup_order: list[str]
diagnostics: list[Diagnostic]
property actionable_operations: list[EvaluationOp]
__init__(resources=<factory>, operations=<factory>, startup_order=<factory>, diagnostics=<factory>)
Parameters:
Return type:

None

class raes_contracts.planning.OrchestrationOp

Bases: PlanOperation

Orchestration reconciliation operation.

class raes_contracts.planning.OrchestrationPlan

Bases: object

Resolved orchestration graph and reconciliation actions.

resources: dict[str, PlannedResource]
operations: list[OrchestrationOp]
startup_order: list[str]
diagnostics: list[Diagnostic]
property actionable_operations: list[OrchestrationOp]
__init__(resources=<factory>, operations=<factory>, startup_order=<factory>, diagnostics=<factory>)
Parameters:
Return type:

None

class raes_contracts.planning.PlanOperation

Bases: object

A reconciliation operation for a planned resource.

action: ChangeAction
address: str
resource_type: str
payload: dict[str, Any]
ordering_dependencies: tuple[str, ...] = ()
refresh_dependencies: tuple[str, ...] = ()
__init__(action, address, resource_type, payload, ordering_dependencies=(), refresh_dependencies=())
Parameters:

  • action (ChangeAction)

  • address (str)

  • resource_type (str)

  • payload (dict[str, Any])

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

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

Return type:

None

class raes_contracts.planning.PlannedResource

Bases: object

Normalized resource used by the planner and snapshot.

address: str
domain: RuntimeDomain
resource_type: str
payload: dict[str, Any]
ordering_dependencies: tuple[str, ...] = ()
refresh_dependencies: tuple[str, ...] = ()
__init__(address, domain, resource_type, payload, ordering_dependencies=(), refresh_dependencies=())
Parameters:

  • address (str)

  • domain (RuntimeDomain)

  • resource_type (str)

  • payload (dict[str, Any])

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

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

Return type:

None

class raes_contracts.planning.ProvisionOp

Bases: PlanOperation

Provisioning reconciliation operation.

class raes_contracts.planning.ProvisioningPlan

Bases: object

Provisioning plan over canonical deployment resources.

resources: dict[str, PlannedResource]
operations: list[ProvisionOp]
diagnostics: list[Diagnostic]
realization_envelope: RealizationEnvelopeIdentityModel | None = None
property actionable_operations: list[ProvisionOp]
__init__(resources=<factory>, operations=<factory>, diagnostics=<factory>, realization_envelope=None)
Parameters:
Return type:

None

class raes_contracts.planning.RuntimeDomain

Bases: str, Enum

Top-level runtime concern.

PROVISIONING = 'provisioning'
ORCHESTRATION = 'orchestration'
EVALUATION = 'evaluation'
PARTICIPANT = 'participant'
__new__(value)
raes_contracts.planning.require_plan_operation_identity(domain, address, resource_type)

Reject operations outside a plan endpoint’s closed identity domain.

Parameters:
Return type:

None

Runtime State

Shared runtime state and control-plane result contracts.

class raes_contracts.runtime_state.ApplyResult

Bases: object

Result of applying or starting a runtime plan.

success: bool
snapshot: RuntimeSnapshot
diagnostics: list[Diagnostic]
changed_addresses: list[str]
details: dict[str, Any]
__init__(success, snapshot, diagnostics=<factory>, changed_addresses=<factory>, details=<factory>)
Parameters:

  • success (bool)

  • snapshot (RuntimeSnapshot)

  • diagnostics (list[Diagnostic])

  • changed_addresses (list[str])

  • details (dict[str, Any])

Return type:

None

class raes_contracts.runtime_state.ExplicitnessClass

Bases: str, Enum

SEM-218 author-intent class for a declaration.

EXACT = 'exact'
CONSTRAINED = 'constrained'
OPEN = 'open'
__new__(value)
class raes_contracts.runtime_state.ExplicitnessProvenance

Bases: str, Enum

Where the classified value came from in the SDL lifecycle.

AUTHOR_DECLARED = 'author-declared'
PROCESSOR_DERIVED = 'processor-derived'
BACKEND_REALIZED = 'backend-realized'
__new__(value)
class raes_contracts.runtime_state.OperationReceipt

Bases: object

Portable acknowledgment for an accepted control-plane operation.

schema_version: str = 'runtime-operation/v1'
operation_id: str = ''
domain: RuntimeDomain = 'provisioning'
submitted_at: str = ''
accepted: bool = True
diagnostics: list[Diagnostic]
__init__(schema_version='runtime-operation/v1', operation_id='', domain=RuntimeDomain.PROVISIONING, submitted_at='', accepted=True, diagnostics=<factory>)
Parameters:

  • schema_version (str)

  • operation_id (str)

  • domain (RuntimeDomain)

  • submitted_at (str)

  • accepted (bool)

  • diagnostics (list[Diagnostic])

Return type:

None

class raes_contracts.runtime_state.OperationState

Bases: str, Enum

Lifecycle for async control-plane operations.

ACCEPTED = 'accepted'
RUNNING = 'running'
SUCCEEDED = 'succeeded'
FAILED = 'failed'
CANCELLED = 'cancelled'
__new__(value)
class raes_contracts.runtime_state.OperationStatus

Bases: object

Portable status for a submitted control-plane operation.

schema_version: str = 'runtime-operation/v1'
operation_id: str = ''
domain: RuntimeDomain = 'provisioning'
state: OperationState = 'accepted'
submitted_at: str = ''
updated_at: str = ''
diagnostics: list[Diagnostic]
changed_addresses: list[str]
__init__(schema_version='runtime-operation/v1', operation_id='', domain=RuntimeDomain.PROVISIONING, state=OperationState.ACCEPTED, submitted_at='', updated_at='', diagnostics=<factory>, changed_addresses=<factory>)
Parameters:
Return type:

None

class raes_contracts.runtime_state.RealizationProvenanceEntry

Bases: object

SEM-218 provenance for one realized realization concern (invariant I5).

Records, for a single realization concern that entered a runtime snapshot / result / history surface, its SEM-218 explicitness class and the origin of the realized value: author-declared (honoured exactly as the author wrote it), processor-derived (produced by deterministic processor activity), or backend-realized (picked by the backend from an open or constrained surface admitted by I3). It carries field-path and kind references only — never the realized value itself, which may carry sensitive material (SEM-218 host-exposure gate).

address: str
field_path: str
domain: str
requirement_kind: str
explicitness: ExplicitnessClass
provenance: ExplicitnessProvenance
governing_scope: str | None = None
artifact_satisfaction: ArtifactSatisfactionDisclosureModel | None = None
__init__(address, field_path, domain, requirement_kind, explicitness, provenance, governing_scope=None, artifact_satisfaction=None)
Parameters:

  • address (str)

  • field_path (str)

  • domain (str)

  • requirement_kind (str)

  • explicitness (ExplicitnessClass)

  • provenance (ExplicitnessProvenance)

  • governing_scope (str | None)

  • artifact_satisfaction (ArtifactSatisfactionDisclosureModel | None)

Return type:

None

class raes_contracts.runtime_state.RuntimeSnapshot

Bases: object

Current runtime snapshot.

entries: dict[str, SnapshotEntry]
orchestration_results: dict[str, dict[str, Any]]
orchestration_history: dict[str, list[dict[str, Any]]]
evaluation_results: dict[str, dict[str, Any]]
evaluation_history: dict[str, list[dict[str, Any]]]
proposition_truth_results: dict[str, dict[str, Any]]
participant_episode_results: dict[str, dict[str, Any]]
participant_episode_history: dict[str, list[dict[str, Any]]]
participant_behavior_history: dict[str, list[dict[str, Any]]]
participant_control_history: dict[str, list[dict[str, Any]]]
participant_crossing_history: dict[str, list[dict[str, Any]]]
participant_autonomous_execution_states: dict[str, dict[str, Any]]
participant_execution_services: dict[str, dict[str, Any]]
participant_resource_budget_states: dict[str, dict[str, Any]]
participant_resource_pool_states: dict[str, dict[str, Any]]
participant_resource_budget_events: dict[str, dict[str, Any]]
shared_state_records: dict[str, dict[str, Any]]
shared_state_history: dict[str, list[dict[str, Any]]]
joint_action_records: dict[str, dict[str, Any]]
time_management_contexts: dict[str, dict[str, Any]]
time_model_state: TimeRuntimeStateModel | None = None
realization_provenance: tuple[RealizationProvenanceEntry, ...] = ()
realization_envelope: RealizationEnvelopeIdentityModel | None = None
metadata: dict[str, Any]
get(address)
Parameters:

address (str)

Return type:

SnapshotEntry | None

for_domain(domain)
Parameters:

domain (RuntimeDomain)

Return type:

dict[str, SnapshotEntry]

with_entries(entries, **updates)
Parameters:
Return type:

RuntimeSnapshot

__init__(entries=<factory>, orchestration_results=<factory>, orchestration_history=<factory>, evaluation_results=<factory>, evaluation_history=<factory>, proposition_truth_results=<factory>, participant_episode_results=<factory>, participant_episode_history=<factory>, participant_behavior_history=<factory>, participant_control_history=<factory>, participant_crossing_history=<factory>, participant_autonomous_execution_states=<factory>, participant_execution_services=<factory>, participant_resource_budget_states=<factory>, participant_resource_pool_states=<factory>, participant_resource_budget_events=<factory>, shared_state_records=<factory>, shared_state_history=<factory>, joint_action_records=<factory>, time_management_contexts=<factory>, time_model_state=None, realization_provenance=(), realization_envelope=None, metadata=<factory>)
Parameters:

  • entries (dict[str, SnapshotEntry])

  • orchestration_results (dict[str, dict[str, Any]])

  • orchestration_history (dict[str, list[dict[str, Any]]])

  • evaluation_results (dict[str, dict[str, Any]])

  • evaluation_history (dict[str, list[dict[str, Any]]])

  • proposition_truth_results (dict[str, dict[str, Any]])

  • participant_episode_results (dict[str, dict[str, Any]])

  • participant_episode_history (dict[str, list[dict[str, Any]]])

  • participant_behavior_history (dict[str, list[dict[str, Any]]])

  • participant_control_history (dict[str, list[dict[str, Any]]])

  • participant_crossing_history (dict[str, list[dict[str, Any]]])

  • participant_autonomous_execution_states (dict[str, dict[str, Any]])

  • participant_execution_services (dict[str, dict[str, Any]])

  • participant_resource_budget_states (dict[str, dict[str, Any]])

  • participant_resource_pool_states (dict[str, dict[str, Any]])

  • participant_resource_budget_events (dict[str, dict[str, Any]])

  • shared_state_records (dict[str, dict[str, Any]])

  • shared_state_history (dict[str, list[dict[str, Any]]])

  • joint_action_records (dict[str, dict[str, Any]])

  • time_management_contexts (dict[str, dict[str, Any]])

  • time_model_state (TimeRuntimeStateModel | None)

  • realization_provenance (tuple[RealizationProvenanceEntry, ...])

  • realization_envelope (RealizationEnvelopeIdentityModel | None)

  • metadata (dict[str, Any])

Return type:

None

class raes_contracts.runtime_state.RuntimeSnapshotEnvelope

Bases: object

Portable envelope around the current runtime snapshot.

schema_version: str = 'runtime-snapshot/v1'
snapshot: RuntimeSnapshot
__init__(schema_version='runtime-snapshot/v1', snapshot=<factory>)
Parameters:
Return type:

None

class raes_contracts.runtime_state.SnapshotEntry

Bases: object

Recorded runtime state for a single canonical resource.

address: str
domain: RuntimeDomain
resource_type: str
payload: dict[str, Any]
ordering_dependencies: tuple[str, ...] = ()
refresh_dependencies: tuple[str, ...] = ()
status: str = 'ready'
__init__(address, domain, resource_type, payload, ordering_dependencies=(), refresh_dependencies=(), status='ready')
Parameters:

  • address (str)

  • domain (RuntimeDomain)

  • resource_type (str)

  • payload (dict[str, Any])

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

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

  • status (str)

Return type:

None

Workflow Results

Shared workflow runtime result contracts.

class raes_contracts.workflow.WorkflowCancellationRequest

Bases: object

Portable request for cancelling one workflow run.

workflow_address: str
run_id: str | None = None
reason: str = 'cancelled by operator'
__init__(workflow_address, run_id=None, reason='cancelled by operator')
Parameters:

  • workflow_address (str)

  • run_id (str | None)

  • reason (str)

Return type:

None

class raes_contracts.workflow.WorkflowCompensationStatus

Bases: str, Enum

Portable workflow compensation status.

NOT_REQUIRED = 'not_required'
PENDING = 'pending'
RUNNING = 'running'
SUCCEEDED = 'succeeded'
FAILED = 'failed'
__new__(value)
class raes_contracts.workflow.WorkflowExecutionContract

Bases: object

Compiled contract for validating workflow-level execution state/history.

state_schema_version: str = 'workflow-step-state/v1'
start_step: str = ''
timeout_seconds: int | None = None
steps: dict[str, WorkflowStepSemanticContract]
step_types: dict[str, str]
control_edges: dict[str, tuple[str, ...]]
join_owners: dict[str, str]
call_steps: dict[str, str]
compensation_mode: str = 'disabled'
compensation_triggers: tuple[str, ...] = ()
compensation_targets: dict[str, str]
compensation_ordering: str = 'reverse_completion'
compensation_failure_policy: str = 'fail_workflow'
observable_steps: tuple[str, ...] = ()
classmethod from_mapping(payload)
Parameters:

payload (Mapping[str, Any])

Return type:

WorkflowExecutionContract

__init__(state_schema_version='workflow-step-state/v1', start_step='', timeout_seconds=None, steps=<factory>, step_types=<factory>, control_edges=<factory>, join_owners=<factory>, call_steps=<factory>, compensation_mode='disabled', compensation_triggers=(), compensation_targets=<factory>, compensation_ordering='reverse_completion', compensation_failure_policy='fail_workflow', observable_steps=())
Parameters:

  • state_schema_version (str)

  • start_step (str)

  • timeout_seconds (int | None)

  • steps (dict[str, WorkflowStepSemanticContract])

  • step_types (dict[str, str])

  • control_edges (dict[str, tuple[str, ...]])

  • join_owners (dict[str, str])

  • call_steps (dict[str, str])

  • compensation_mode (str)

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

  • compensation_targets (dict[str, str])

  • compensation_ordering (str)

  • compensation_failure_policy (str)

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

Return type:

None

class raes_contracts.workflow.WorkflowExecutionState

Bases: object

Internal normalized workflow result envelope.

state_schema_version: str = 'workflow-step-state/v1'
workflow_status: WorkflowStatus = 'pending'
run_id: str = ''
started_at: str = ''
updated_at: str = ''
terminal_reason: str | None = None
compensation_status: WorkflowCompensationStatus = 'not_required'
compensation_started_at: str | None = None
compensation_updated_at: str | None = None
compensation_failures: list[dict[str, Any]]
steps: dict[str, WorkflowStepExecutionState]
classmethod from_payload(payload)
Parameters:

payload (Mapping[str, Any])

Return type:

WorkflowExecutionState

to_payload()
Return type:

dict[str, Any]

__init__(state_schema_version='workflow-step-state/v1', workflow_status=WorkflowStatus.PENDING, run_id='', started_at='', updated_at='', terminal_reason=None, compensation_status=WorkflowCompensationStatus.NOT_REQUIRED, compensation_started_at=None, compensation_updated_at=None, compensation_failures=<factory>, steps=<factory>)
Parameters:

  • state_schema_version (str)

  • workflow_status (WorkflowStatus)

  • run_id (str)

  • started_at (str)

  • updated_at (str)

  • terminal_reason (str | None)

  • compensation_status (WorkflowCompensationStatus)

  • compensation_started_at (str | None)

  • compensation_updated_at (str | None)

  • compensation_failures (list[dict[str, Any]])

  • steps (dict[str, WorkflowStepExecutionState])

Return type:

None

class raes_contracts.workflow.WorkflowHistoryEvent

Bases: object

Internal normalized workflow history event.

event_type: WorkflowHistoryEventType
timestamp: str
step_name: str | None = None
branch_name: str | None = None
join_step: str | None = None
outcome: WorkflowStepOutcome | None = None
details: dict[str, Any]
classmethod from_payload(payload)
Parameters:

payload (Mapping[str, Any])

Return type:

WorkflowHistoryEvent

to_payload()
Return type:

dict[str, Any]

__init__(event_type, timestamp, step_name=None, branch_name=None, join_step=None, outcome=None, details=<factory>)
Parameters:
Return type:

None

class raes_contracts.workflow.WorkflowHistoryEventType

Bases: str, Enum

Portable workflow history event kinds.

WORKFLOW_STARTED = 'workflow_started'
STEP_STARTED = 'step_started'
STEP_COMPLETED = 'step_completed'
SWITCH_CASE_SELECTED = 'switch_case_selected'
CALL_STARTED = 'call_started'
CALL_COMPLETED = 'call_completed'
BRANCH_ENTERED = 'branch_entered'
BRANCH_CONVERGED = 'branch_converged'
WORKFLOW_COMPLETED = 'workflow_completed'
WORKFLOW_FAILED = 'workflow_failed'
WORKFLOW_CANCELLED = 'workflow_cancelled'
WORKFLOW_TIMED_OUT = 'workflow_timed_out'
COMPENSATION_REGISTERED = 'compensation_registered'
COMPENSATION_STARTED = 'compensation_started'
COMPENSATION_WORKFLOW_STARTED = 'compensation_workflow_started'
COMPENSATION_WORKFLOW_COMPLETED = 'compensation_workflow_completed'
COMPENSATION_WORKFLOW_FAILED = 'compensation_workflow_failed'
COMPENSATION_COMPLETED = 'compensation_completed'
COMPENSATION_FAILED = 'compensation_failed'
__new__(value)
class raes_contracts.workflow.WorkflowResultContract

Bases: object

Compiled contract for validating portable workflow result envelopes.

state_schema_version: str = 'workflow-step-state/v1'
observable_steps: dict[str, WorkflowStepSemanticContract]
classmethod from_mapping(payload)
Parameters:

payload (Mapping[str, Any])

Return type:

WorkflowResultContract

__init__(state_schema_version='workflow-step-state/v1', observable_steps=<factory>)
Parameters:
Return type:

None

class raes_contracts.workflow.WorkflowStatus

Bases: str, Enum

Portable workflow-level execution status.

PENDING = 'pending'
RUNNING = 'running'
SUCCEEDED = 'succeeded'
FAILED = 'failed'
CANCELLED = 'cancelled'
TIMED_OUT = 'timed_out'
__new__(value)
class raes_contracts.workflow.WorkflowStepExecutionState

Bases: object

Internal normalized execution state for one workflow-visible step.

lifecycle: WorkflowStepLifecycle = 'pending'
outcome: WorkflowStepOutcome | None = None
attempts: int = 0
attempt_provenance: tuple[WorkflowStepAttemptProvenance, ...] = ()
classmethod from_payload(payload)
Parameters:

payload (Mapping[str, Any])

Return type:

WorkflowStepExecutionState

to_payload()
Return type:

dict[str, Any]

__init__(lifecycle=WorkflowStepLifecycle.PENDING, outcome=None, attempts=0, attempt_provenance=())
Parameters:
Return type:

None

class raes_contracts.workflow.WorkflowStepAttemptProvenance

Bases: object

Portable provenance for one governed workflow-step realization attempt.

step_name: str
execution_mode: str
attempt_id: str
objective_address: str = ''
procedure_ref: str = ''
exposed_scaffold_refs: tuple[str, ...] = ()
allowed_action_families: tuple[str, ...] = ()
selected_action_family: str = ''
selected_tool_ref: str = ''
selected_affordance_ref: str = ''
fact_versions: tuple[str, ...] = ()
outcome: str = ''
evidence_refs: tuple[str, ...] = ()
assertion_truth_refs: tuple[str, ...] = ()
participant_report: str = ''
classmethod from_payload(payload)
Parameters:

payload (Mapping[str, Any])

Return type:

WorkflowStepAttemptProvenance

to_payload()
Return type:

dict[str, Any]

__init__(step_name, execution_mode, attempt_id, objective_address='', procedure_ref='', exposed_scaffold_refs=(), allowed_action_families=(), selected_action_family='', selected_tool_ref='', selected_affordance_ref='', fact_versions=(), outcome='', evidence_refs=(), assertion_truth_refs=(), participant_report='')
Parameters:

  • step_name (str)

  • execution_mode (str)

  • attempt_id (str)

  • objective_address (str)

  • procedure_ref (str)

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

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

  • selected_action_family (str)

  • selected_tool_ref (str)

  • selected_affordance_ref (str)

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

  • outcome (str)

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

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

  • participant_report (str)

Return type:

None

class raes_contracts.workflow.WorkflowStepLifecycle

Bases: str, Enum

Portable execution lifecycle for workflow-visible step state.

PENDING = 'pending'
RUNNING = 'running'
COMPLETED = 'completed'
__new__(value)
class raes_contracts.workflow.WorkflowStepOutcome

Bases: str, Enum

Portable execution outcomes for workflow-visible step state.

SUCCEEDED = 'succeeded'
FAILED = 'failed'
EXHAUSTED = 'exhausted'
__new__(value)
raes_contracts.workflow.validate_workflow_step_result_contract(contract, *, lifecycle, outcome, attempts)

Validate a backend-reported workflow step result against a compiled contract.

Parameters:
Return type:

tuple[str, …]

Evaluation Results

Shared evaluation runtime result contracts.

class raes_contracts.evaluation.EvaluationExecutionContract

Bases: object

Compiled contract for validating evaluator history/state transitions.

state_schema_version: str = 'evaluation-result-state/v1'
resource_type: str = ''
allowed_statuses: tuple[str, ...] = ('pending', 'running', 'ready', 'failed')
history_event_types: tuple[str, ...] = ('evaluation_started', 'evaluation_updated', 'evaluation_ready', 'evaluation_failed')
requires_start_event: bool = True
classmethod from_mapping(payload)
Parameters:

payload (Mapping[str, Any])

Return type:

EvaluationExecutionContract

__init__(state_schema_version='evaluation-result-state/v1', resource_type='', allowed_statuses=('pending', 'running', 'ready', 'failed'), history_event_types=('evaluation_started', 'evaluation_updated', 'evaluation_ready', 'evaluation_failed'), requires_start_event=True)
Parameters:

  • state_schema_version (str)

  • resource_type (str)

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

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

  • requires_start_event (bool)

Return type:

None

class raes_contracts.evaluation.EvaluationExecutionState

Bases: object

Internal normalized execution state for one evaluator-observable resource.

state_schema_version: str = 'evaluation-result-state/v1'
resource_type: str = ''
run_id: str = ''
status: EvaluationResultStatus = 'pending'
observed_at: str = ''
updated_at: str = ''
passed: bool | None = None
score: float | int | None = None
max_score: int | None = None
detail: str | None = None
evidence_refs: tuple[str, ...] = ()
classmethod from_payload(payload)
Parameters:

payload (Mapping[str, Any])

Return type:

EvaluationExecutionState

to_payload()
Return type:

dict[str, Any]

__init__(state_schema_version='evaluation-result-state/v1', resource_type='', run_id='', status=EvaluationResultStatus.PENDING, observed_at='', updated_at='', passed=None, score=None, max_score=None, detail=None, evidence_refs=())
Parameters:

  • state_schema_version (str)

  • resource_type (str)

  • run_id (str)

  • status (EvaluationResultStatus)

  • observed_at (str)

  • updated_at (str)

  • passed (bool | None)

  • score (float | int | None)

  • max_score (int | None)

  • detail (str | None)

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

Return type:

None

class raes_contracts.evaluation.EvaluationHistoryEvent

Bases: object

Internal normalized evaluator history event.

event_type: EvaluationHistoryEventType
timestamp: str
status: EvaluationResultStatus
passed: bool | None = None
score: float | int | None = None
max_score: int | None = None
detail: str | None = None
evidence_refs: tuple[str, ...] = ()
details: dict[str, Any]
classmethod from_payload(payload)
Parameters:

payload (Mapping[str, Any])

Return type:

EvaluationHistoryEvent

to_payload()
Return type:

dict[str, Any]

__init__(event_type, timestamp, status, passed=None, score=None, max_score=None, detail=None, evidence_refs=(), details=<factory>)
Parameters:
Return type:

None

class raes_contracts.evaluation.EvaluationHistoryEventType

Bases: str, Enum

Portable evaluator history event kinds.

EVALUATION_STARTED = 'evaluation_started'
EVALUATION_UPDATED = 'evaluation_updated'
EVALUATION_READY = 'evaluation_ready'
EVALUATION_FAILED = 'evaluation_failed'
__new__(value)
class raes_contracts.evaluation.EvaluationResultContract

Bases: object

Compiled contract for validating evaluator result envelopes.

state_schema_version: str = 'evaluation-result-state/v1'
resource_type: str = ''
supports_passed: bool = False
supports_score: bool = False
fixed_max_score: int | None = None
classmethod from_mapping(payload)
Parameters:

payload (Mapping[str, Any])

Return type:

EvaluationResultContract

__init__(state_schema_version='evaluation-result-state/v1', resource_type='', supports_passed=False, supports_score=False, fixed_max_score=None)
Parameters:

  • state_schema_version (str)

  • resource_type (str)

  • supports_passed (bool)

  • supports_score (bool)

  • fixed_max_score (int | None)

Return type:

None

class raes_contracts.evaluation.EvaluationResultStatus

Bases: str, Enum

Portable lifecycle for evaluator-observable results.

PENDING = 'pending'
RUNNING = 'running'
READY = 'ready'
FAILED = 'failed'
__new__(value)
raes_contracts.evaluation.validate_evaluation_result(contract, state)

Return contract violations for one evaluator result envelope.

Parameters:
Return type:

list[str]

Participant Episodes

Shared participant episode runtime contracts.

class raes_contracts.participant_episode.ParticipantEpisodeControlAction

Bases: str, Enum

INITIALIZE = 'initialize'
RESET = 'reset'
RESTART = 'restart'
__new__(value)
class raes_contracts.participant_episode.ParticipantEpisodeExecutionState

Bases: object

Internal normalized participant-episode execution state envelope.

state_schema_version: str = 'participant-episode-state/v1'
participant_address: str = ''
episode_id: str = ''
sequence_number: int = 0
status: ParticipantEpisodeStatus = 'initializing'
terminal_reason: ParticipantEpisodeTerminalReason | None = None
initialized_at: str = ''
updated_at: str = ''
terminated_at: str | None = None
last_control_action: ParticipantEpisodeControlAction = 'initialize'
previous_episode_id: str | None = None
classmethod from_payload(payload)
Parameters:

payload (Mapping[str, Any])

Return type:

ParticipantEpisodeExecutionState

to_payload()
Return type:

dict[str, Any]

__init__(state_schema_version='participant-episode-state/v1', participant_address='', episode_id='', sequence_number=0, status=ParticipantEpisodeStatus.INITIALIZING, terminal_reason=None, initialized_at='', updated_at='', terminated_at=None, last_control_action=ParticipantEpisodeControlAction.INITIALIZE, previous_episode_id=None)
Parameters:
Return type:

None

class raes_contracts.participant_episode.ParticipantEpisodeHistoryEvent

Bases: object

Internal normalized participant-episode history event.

event_type: ParticipantEpisodeHistoryEventType
timestamp: str
participant_address: str
episode_id: str
sequence_number: int
terminal_reason: ParticipantEpisodeTerminalReason | None = None
control_action: ParticipantEpisodeControlAction | None = None
details: dict[str, Any]
classmethod from_payload(payload)
Parameters:

payload (Mapping[str, Any])

Return type:

ParticipantEpisodeHistoryEvent

to_payload()
Return type:

dict[str, Any]

__init__(event_type, timestamp, participant_address, episode_id, sequence_number, terminal_reason=None, control_action=None, details=<factory>)
Parameters:
Return type:

None

class raes_contracts.participant_episode.ParticipantEpisodeHistoryEventType

Bases: str, Enum

EPISODE_INITIALIZED = 'episode_initialized'
EPISODE_RUNNING = 'episode_running'
EPISODE_COMPLETED = 'episode_completed'
EPISODE_TIMED_OUT = 'episode_timed_out'
EPISODE_TRUNCATED = 'episode_truncated'
EPISODE_INTERRUPTED = 'episode_interrupted'
EPISODE_RESET = 'episode_reset'
EPISODE_RESTARTED = 'episode_restarted'
__new__(value)
class raes_contracts.participant_episode.ParticipantEpisodeInitializeRequest

Bases: object

Portable request for initializing the first episode of a participant.

participant_address: str
episode_id: str | None = None
__init__(participant_address, episode_id=None)
Parameters:

  • participant_address (str)

  • episode_id (str | None)

Return type:

None

class raes_contracts.participant_episode.ParticipantEpisodeResetRequest

Bases: object

Portable request for resetting a non-terminal participant episode.

participant_address: str
episode_id: str | None = None
reason: str = 'reset by operator'
__init__(participant_address, episode_id=None, reason='reset by operator')
Parameters:

  • participant_address (str)

  • episode_id (str | None)

  • reason (str)

Return type:

None

class raes_contracts.participant_episode.ParticipantEpisodeRestartRequest

Bases: object

Portable request for restarting a terminated participant episode.

participant_address: str
episode_id: str | None = None
reason: str = 'restarted by operator'
__init__(participant_address, episode_id=None, reason='restarted by operator')
Parameters:

  • participant_address (str)

  • episode_id (str | None)

  • reason (str)

Return type:

None

class raes_contracts.participant_episode.ParticipantEpisodeStatus

Bases: str, Enum

INITIALIZING = 'initializing'
RUNNING = 'running'
TERMINATED = 'terminated'
__new__(value)
class raes_contracts.participant_episode.ParticipantEpisodeTerminalReason

Bases: str, Enum

COMPLETED = 'completed'
TIMED_OUT = 'timed_out'
TRUNCATED = 'truncated'
INTERRUPTED = 'interrupted'
__new__(value)
class raes_contracts.participant_episode.ParticipantEpisodeTerminateRequest

Bases: object

Portable request for driving the current episode to TERMINATED.

participant_address: str
terminal_reason: ParticipantEpisodeTerminalReason = 'interrupted'
detail: str = 'terminated by operator'
__init__(participant_address, terminal_reason=ParticipantEpisodeTerminalReason.INTERRUPTED, detail='terminated by operator')
Parameters:
Return type:

None

raes_contracts.participant_episode.iter_participant_episode_snapshot_violations(participant_episode_results, participant_episode_history)
Parameters:

  • participant_episode_results (object)

  • participant_episode_history (object)

Return type:

Iterator[tuple[str, str]]

Participant Implementation Manifests

ParticipantImplementationManifestModel and ParticipantImplementationProvenanceModel live in raes_contracts.contracts with the other published contract models. The manifest declares participant implementation identity, capabilities, compatibility, decision-surface modes, tool-affordance expectations, and constraints. The provenance record preserves the participant implementation, selected manifest, selected configuration reference, participant contract versions, and decision-surface exposure policy used in a run.