mud_server.services.condition_axis_service

Canonical condition-axis generation service.

This module centralizes all condition-axis generation behavior so API routes and internal callsites can rely on one stable adapter contract.

Primary responsibilities: - Validate seed and runtime input contracts. - Resolve policy bundle metadata and reproducibility hashes. - Call the configured upstream entity generator adapter. - Normalize returned axis payloads into canonical axis -> score shape. - Map all known failures into structured, stable service errors.

Attributes

`SEED_MIN`
`SEED_MAX`
`logger`

Exceptions

ConditionAxisServiceError

Typed service error carrying stable HTTP and contract metadata.

Classes

`ConditionAxisPolicyContext`	Resolved policy metadata for one generation request.
`ConditionAxisProvenance`	Canonical provenance metadata returned to clients.
`ConditionAxisGenerationResult`	Service result payload for canonical condition-axis generation.

Functions

`generate_condition_axis`(*, world_id, world_root[, ...])	Generate canonical condition-axis values for one world.
`get_condition_axis_upstream_metrics`()	Return an in-memory snapshot of upstream adapter metrics counters.

Module Contents

mud_server.services.condition_axis_service.SEED_MIN = 1

mud_server.services.condition_axis_service.SEED_MAX = 2147483647

mud_server.services.condition_axis_service.logger

exception mud_server.services.condition_axis_service.ConditionAxisServiceError(*, status_code, code, detail, stage=_SERVICE_STAGE)[source]

Bases: RuntimeError

Typed service error carrying stable HTTP and contract metadata.

status_code: HTTP status to be surfaced by API route handlers.

code: Stable machine-readable error code for UI/client mapping.

detail: Human-readable summary of the failure.

stage: Stable processing stage key for pipeline diagnostics.

Initialize self. See help(type(self)) for accurate signature.

status_code

code

detail

stage = 'axis_input'

to_response_payload()[source]

Return canonical error payload for API responses.

class mud_server.services.condition_axis_service.ConditionAxisPolicyContext[source]

Resolved policy metadata for one generation request.

bundle_id: Effective bundle id used for the generation request.

bundle_version: Effective bundle version string.

policy_hash: Deterministic hash of resolved policy inputs.

required_runtime_inputs: Runtime keys required for strict validation.

required_axes: Canonical axis key set required by active policy bundle.

axis_label_scores: Deterministic axis -> label -> score lookup derived from policy thresholds/orderings for label-only upstream normalization.

bundle_id: str

bundle_version: str

policy_hash: str | None

required_runtime_inputs: set[str]

required_axes: set[str]

axis_label_scores: dict[str, dict[str, float]]

class mud_server.services.condition_axis_service.ConditionAxisProvenance[source]

Canonical provenance metadata returned to clients.

source: Canonical owner identifier for the response contract.

served_via: API route path that served the response.

generator: Upstream generator system identifier.

generator_version: Upstream generator version/capability value.

generator_capabilities: Ordered upstream capability tokens.

generated_at: ISO-8601 timestamp for generation completion.

source: str

served_via: str

generator: str

generator_version: str

generator_capabilities: tuple[str, Ellipsis]

generated_at: str

class mud_server.services.condition_axis_service.ConditionAxisGenerationResult[source]

Service result payload for canonical condition-axis generation.

world_id: Target world id used for policy resolution.

bundle_id: Effective bundle id used for generation.

bundle_version: Effective bundle version string.

policy_hash: Deterministic policy hash for reproducibility.

seed: Deterministic generation seed.

axes: Canonical axis map of axis_name -> score.

provenance: Canonical provenance block for diagnostics/auditability.

entity_state: Raw upstream entity payload for internal reuse/callers.

world_id: str

bundle_id: str

bundle_version: str

policy_hash: str | None

seed: int

axes: dict[str, float]

provenance: ConditionAxisProvenance

entity_state: dict[str, Any]

mud_server.services.condition_axis_service.generate_condition_axis(*, world_id, world_root, seed=None, bundle_id=None, inputs=None, strict_inputs=False)[source]

Generate canonical condition-axis values for one world.

Parameters:

world_id (str) – Canonical world identifier.
world_root (pathlib.Path) – Resolved world package root.
seed (int | None) – Optional deterministic generation seed.
bundle_id (str | None) – Optional bundle override requested by caller.
inputs (dict[str, Any] | None) – Optional runtime input payload.
strict_inputs (bool) – Enforce strict runtime input validation when True.

Returns:

Canonical service result with normalized axes and provenance.

Raises:

ConditionAxisServiceError – If input validation, policy resolution, or upstream generation fails.

Return type:

ConditionAxisGenerationResult

mud_server.services.condition_axis_service.get_condition_axis_upstream_metrics()[source]

Return an in-memory snapshot of upstream adapter metrics counters.

Returns:: Mapping of metric name to cumulative count since process start.
Return type:: dict[str, int]