AuthorityCommandHandler

Source:

src/ometeotl_core/generic/authority.py

Local role: Single server-side mutation path that validates and applies allowlisted commands.

Big-picture role: Authority boundary that locks direct mutation on World, enforces sequencing, handles deduplication, and emits AuditEntry records.

Inheritance:

standard class

Constructor parameters:

world: World
allowed_command_types: Optional[Sequence[str]]
custom_command_handlers: Optional[Mapping[str, CommandHandler]]
object_factories: Optional[Mapping[str, ObjectFactory]]
audit_log_maxlen: int
processed_ids_maxlen: int
sequence_tracker_max_actors: Optional[int]
sequence_history_max_actors: Optional[int]
validation_soft_gate: bool
validation_policy_profile: str (observe_only, enforce_structure, enforce_domain)
validation_stage_mode_overrides: Optional[Mapping[str, str]]
validation_block_on_error: bool
validation_completeness_level: str

Methods:

submit(command: CommandEnvelope) -> CommandResult
close() -> None
audit_log -> list[AuditEntry]
internal validation and handler methods for add_space, add_space_relation, place_object, register_object, unregister_object

Validation behavior:

Runs a staged validation pipeline before command application when validation_soft_gate is enabled.
Includes syntactic, structural, completeness, temporal, spatial, admissibility, and epistemic validators.
Exposes structured validation summaries in CommandResult and AuditEntry.
Can reject commands based on validation errors when validation_block_on_error=True.

Example:

from ometeotl_core.generic.authority import AuthorityCommandHandler, CommandEnvelope
import datetime

handler = AuthorityCommandHandler(
    world=world,
    allowed_command_types=["add_space", "place_object"],
    validation_policy_profile="enforce_structure",
    validation_soft_gate=True,
)
envelope = CommandEnvelope(
    command_id="cmd-001",
    actor_id="actor-1",
    command_type="add_space",
    payload={"space": space.to_dict()},
    sequence=1,
    issued_at=datetime.datetime.utcnow().isoformat(),
)
result = handler.submit(envelope)
print(result.accepted, result.reason)

for entry in handler.audit_log:
    print(entry.command_id, entry.accepted)

handler.close()

Ometeotl