mud_server.db.policy_repo

SQLite repository functions for canonical policy authoring state.

This module is intentionally limited to storage concerns: - inserting/updating policy identity rows - inserting/updating policy variants - recording validation, activation, and publish history - reading normalized rows for API/service callers

Business rules (for example policy-type validation and world existence checks) belong in mud_server.services.policy_service. Keeping that split makes both layers easier to test and reason about.

Functions

upsert_policy_item(*, policy_id, policy_type, ...)	Insert one policy identity row when it does not already exist.
upsert_policy_variant(*, policy_id, variant, ...)	Insert or update one policy variant and return the canonical row.
list_policies(*[, policy_type, namespace, status])	List policy variants with optional filters.
get_policy(*, policy_id[, variant])	Get one policy variant by id and optional variant key.
insert_validation_run(*, policy_id, variant, is_valid, ...)	Insert one validation-run row and return row id.
get_activation_event(event_id)	Return one activation audit event row by id.
list_activation_events(*, world_id, client_profile[, ...])	List activation audit events for one scope ordered by write sequence.
set_policy_activation(*, world_id, client_profile, ...)	Atomically upsert the active pointer and emit an activation audit row.
list_policy_activations(*, world_id, client_profile)	List active policy pointers for one activation scope.
insert_publish_run(*, world_id, client_profile, actor, ...)	Insert one publish run plus audit event and return run id.
get_publish_run(*, publish_run_id)	Return one publish-run row with parsed manifest payload.

Module Contents

mud_server.db.policy_repo.upsert_policy_item(*, policy_id, policy_type, namespace, policy_key)

Insert one policy identity row when it does not already exist.

policy_item is immutable identity metadata (type, namespace, key). Variant content and status live in policy_variant.

mud_server.db.policy_repo.upsert_policy_variant(*, policy_id, variant, schema_version, policy_version, status, content, content_hash, updated_at, updated_by)

Insert or update one policy variant and return the canonical row.

The method intentionally performs a read-after-write to ensure callers receive the database-normalized representation (including JSON decoding and integer coercions) from a single code path.

mud_server.db.policy_repo.list_policies(*, policy_type=None, namespace=None, status=None)

List policy variants with optional filters.

Parameters:

• policy_type (str | None) – Optional exact filter for policy_item.policy_type.

• namespace (str | None) – Optional exact filter for policy_item.namespace.

• status (str | None) – Optional exact filter for policy_variant.status.

Returns:

A list of normalized policy-object dictionaries sorted by stable identity keys and descending version within each policy.

Return type:

list[dict[str, Any]]

mud_server.db.policy_repo.get_policy(*, policy_id, variant=None)

Get one policy variant by id and optional variant key.

If variant is omitted, returns the highest policy_version variant.

mud_server.db.policy_repo.insert_validation_run(*, policy_id, variant, is_valid, errors, validated_at, validated_by)

Insert one validation-run row and return row id.

Validation history is append-only. It allows audit replay even when a variant is later updated.

mud_server.db.policy_repo.get_activation_event(event_id)

Return one activation audit event row by id.

mud_server.db.policy_repo.list_activation_events(*, world_id, client_profile, policy_id=None)

List activation audit events for one scope ordered by write sequence.

Replay logic in the service layer depends on deterministic ordering by monotonically increasing event id. Optional policy_id filtering allows focused checks without scanning unrelated policy streams.

mud_server.db.policy_repo.set_policy_activation(*, world_id, client_profile, policy_id, variant, activated_by, activated_at, rollback_of_activation_id)

Atomically upsert the active pointer and emit an activation audit row.

This method runs inside one write transaction so pointer state and audit history move together.

mud_server.db.policy_repo.list_policy_activations(*, world_id, client_profile)

List active policy pointers for one activation scope.

mud_server.db.policy_repo.insert_publish_run(*, world_id, client_profile, actor, manifest, created_at)

Insert one publish run plus audit event and return run id.

The publish manifest is stored verbatim for later inspection and replay.

mud_server.db.policy_repo.get_publish_run(*, publish_run_id)

Return one publish-run row with parsed manifest payload.

Parameters:

publish_run_id (int) – Integer primary key from policy_publish_run.

Returns:

Normalized dictionary when found, otherwise None.

Return type:

dict[str, Any] | None